From 0f1d44d85d556d0e781d912d893a2f5b8e6c4cb5 Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Mon, 22 Jun 2026 13:09:12 -0700 Subject: [PATCH 1/3] docs: add registry anchoring and audit chain verification tutorials MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two new tutorials filling gaps identified in the tutorial coverage audit: - anchoring-to-the-registry.md: how to set the transparency field via direct SCITT HTTP submission (no SDK registry client exists yet); explains the pending-placeholder pattern for dev, re-sign after anchoring, and what verification step 6 requires of a verifier - verifying-the-audit-chain.md: how to verify tool_transcript.hash against the external transcript bytes, validate call count, and understand external execution receipts per spec §3.3.1 mkdocs.yml: adds both tutorials to the Tutorials nav Co-Authored-By: Claude Sonnet 4.6 --- docs/tutorials/anchoring-to-the-registry.md | 172 ++++++++++++++ docs/tutorials/verifying-the-audit-chain.md | 251 ++++++++++++++++++++ mkdocs.yml | 2 + 3 files changed, 425 insertions(+) create mode 100644 docs/tutorials/anchoring-to-the-registry.md create mode 100644 docs/tutorials/verifying-the-audit-chain.md diff --git a/docs/tutorials/anchoring-to-the-registry.md b/docs/tutorials/anchoring-to-the-registry.md new file mode 100644 index 0000000..ead0f9e --- /dev/null +++ b/docs/tutorials/anchoring-to-the-registry.md @@ -0,0 +1,172 @@ +# Anchoring a Trust Record to the TRACE registry + +After signing a Trust Record, you should anchor it to the TRACE transparency registry. The anchor receipt proves that the record existed at a specific time and has not been altered since — tamper evidence that holds even if the operator who produced the record is later compromised. + +**What you need:** A signed Trust Record (from [Signing your first trust record](signing-your-first-trust-record.md)). + +**What you'll do:** Submit the record to the registry, receive a SCITT receipt, and set the `transparency` field to the canonical receipt URI. + +--- + +## Why transparency anchoring matters + +A Trust Record carries a signature from the issuer's key. A verifier holding that key can confirm the record has not been modified — but only if the key is trustworthy. If the issuer is later compromised, an attacker could forge records backdated to before the compromise. + +The `transparency` field solves this with a different trust root: an append-only log operated by an independent party. Once a record is registered, its content is fixed in the log at that timestamp. A verifier checks that the record's digest matches the log entry — no trust in the operator required. + +TRACE uses SCITT ([draft-ietf-scitt-architecture](https://datatracker.ietf.org/doc/draft-ietf-scitt-architecture/)) as its transparency log substrate. + +--- + +## The `transparency` field + +In the `TrustRecord` schema, `transparency` is a required string: + +```python +transparency: Annotated[str, Field(min_length=1)] +``` + +It holds the canonical URI of the registry entry — the URL at which any verifier can independently retrieve the SCITT receipt and confirm the record's inclusion. + +Example value from the spec: + +``` +https://registry.agentrust.io/claim/trace-2026-06-23T09:15:42Z-f2a8d1 +``` + +--- + +## Step 1 — Build and sign the record with a placeholder + +During development, use a placeholder for `transparency` so you can construct and sign a valid record before anchoring: + +```python +import time +from agentrust_trace.sign import generate_key, sign_record + +key = generate_key() + +record = { + "eat_profile": "tag:agentrust.io,2026:trace-v0.1", + "iat": int(time.time()), + "subject": "spiffe://example.org/agent/my-agent", + "model": { + "provider": "anthropic", + "model_id": "claude-sonnet-4-6", + }, + "runtime": { + "platform": "software-only", + "measurement": "sha256:" + "0" * 64, + }, + "policy": { + "bundle_hash": "sha256:" + "a" * 64, + "enforcement_mode": "enforce", + }, + "data_class": "internal", + "build_provenance": { + "slsa_level": 0, + "digest": "sha256:" + "b" * 64, + }, + "appraisal": { + "status": "none", + "verifier": "self", + }, + # Placeholder — replace after anchoring + "transparency": "pending", +} + +signed = sign_record(record, key) +``` + +!!! note + `transparency: "pending"` is valid for local development. Production records MUST carry a real receipt URI before being handed to a verifier that enforces §3.3 step 6. + +--- + +## Step 2 — Submit to the registry + +!!! info "No SDK client yet" + The `agentrust_trace` SDK does not yet include a registry client. Submit directly via HTTP using the SCITT Reference API ([draft-ietf-scitt-scrapi](https://datatracker.ietf.org/doc/draft-ietf-scitt-scrapi/)). A Python client will be added to the SDK in a future release. + +Submit the signed record as a SCITT Signed Statement: + +```python +import json +import requests # pip install requests + +REGISTRY_URL = "https://registry.agentrust.io" + +response = requests.post( + f"{REGISTRY_URL}/entries", + headers={"Content-Type": "application/json"}, + data=json.dumps(signed), + timeout=30, +) +response.raise_for_status() + +entry = response.json() +# entry contains the registry-assigned receipt URI +receipt_uri = entry["receipt_uri"] +print(f"Anchored: {receipt_uri}") +``` + +The registry returns a JSON object with at minimum: + +| Field | Description | +|---|---| +| `receipt_uri` | Canonical URL of the SCITT receipt — use this as `transparency` | +| `entry_id` | Registry-internal identifier for the log entry | +| `registered_at` | ISO 8601 timestamp of registration | + +--- + +## Step 3 — Set `transparency` and re-sign + +Replace the placeholder and re-sign the record with the real receipt URI: + +```python +record["transparency"] = receipt_uri +signed_final = sign_record(record, key) +``` + +The signature now covers the real `transparency` value. A verifier who later retrieves the receipt from `receipt_uri` can confirm the digest matches — without contacting the original operator. + +--- + +## Step 4 — Verify the receipt (optional, recommended) + +To confirm the registry accepted the record correctly, retrieve and inspect the receipt: + +```python +receipt_resp = requests.get(receipt_uri, timeout=30) +receipt_resp.raise_for_status() +receipt = receipt_resp.json() + +# The receipt contains the log entry digest and a Merkle inclusion proof. +# Check that it references your record's content. +assert receipt["subject"] == signed_final["subject"] +assert receipt["iat"] == signed_final["iat"] +``` + +A full cryptographic Merkle proof verification is not yet in the SDK. The registry exposes the raw proof fields for implementers who want to verify inclusion independently. + +--- + +## Verification step 6 + +When a verifier calls `verify_record()`, it checks the signature. Step 6 of the TRACE verification procedure (spec §3.3) additionally requires the transparency receipt to resolve: + +> SCITT receipt resolves on the named transparency log. + +A verifier configured with `require_transparency: true` will retrieve `transparency` and check the digest against the log. Verifiers that skip this step accept a weaker guarantee — signature-only, not log-anchored. + +--- + +## Summary + +| Step | What happens | +|---|---| +| Build record with `"pending"` | Valid locally, not for production hand-off | +| POST to registry | SCITT log records the digest at this timestamp | +| Replace `transparency`, re-sign | Signature now covers the real receipt URI | +| Verifier retrieves receipt URI | Tamper evidence independent of operator trust | diff --git a/docs/tutorials/verifying-the-audit-chain.md b/docs/tutorials/verifying-the-audit-chain.md new file mode 100644 index 0000000..e18ab04 --- /dev/null +++ b/docs/tutorials/verifying-the-audit-chain.md @@ -0,0 +1,251 @@ +# Verifying the tool call transcript + +A TRACE Trust Record commits the evidence of every tool call by hash. This tutorial explains what the `tool_transcript` field contains, how to verify that a received record's transcript hash is consistent with the actual tool calls, and how external execution receipts extend the chain. + +**What you need:** A Trust Record with a `tool_transcript` field, the matching transcript file, and the issuer's public key. + +--- + +## What `tool_transcript` captures + +The `tool_transcript` field in `TrustRecord` has three fields: + +```python +class ToolTranscript(BaseModel): + hash: DigestStr # sha256 or sha384 digest of all tool call content + call_count: int | None # number of calls in this session (optional) + transcript_uri: str | None # where the full transcript can be retrieved +``` + +`hash` is the binding between the Trust Record (which is signed) and the full transcript (which is stored externally). When the Trust Record signature verifies, the `hash` inside it is signed. When the hash matches the transcript you retrieve from `transcript_uri`, you know the transcript has not been altered since the record was signed. + +The full transcript is NOT embedded in the Trust Record — it lives at `transcript_uri`. This keeps records small enough to sign and transmit while still committing all call-level evidence. + +--- + +## Step 1 — Retrieve and verify the record signature + +Start by checking the Trust Record signature with the issuer's public key: + +```python +from agentrust_trace.sign import verify_record, load_key + +with open("issuer_pub.pem", "rb") as f: + public_key = load_key(f.read()) + +with open("trust_record.json") as f: + import json + record = json.load(f) + +result = verify_record(record, public_key_or_jwk=public_key) +# raises agentrust_trace.exceptions.VerificationError on failure +# returns True on success +``` + +`verify_record` confirms that the signed content of the record has not been altered. This includes the `tool_transcript.hash` field — if the signature is valid, you have a trusted copy of the hash. + +--- + +## Step 2 — Retrieve the transcript + +The full transcript lives at `tool_transcript.transcript_uri`. Retrieve it and hold the raw bytes for hashing: + +```python +import requests + +transcript_uri = record["tool_transcript"]["transcript_uri"] +response = requests.get(transcript_uri, timeout=30) +response.raise_for_status() + +# Hold raw bytes — hash must be computed over the exact bytes served +transcript_bytes = response.content +``` + +!!! warning "Hash bytes, not parsed content" + The `tool_transcript.hash` is computed over the raw bytes of the transcript as stored. Do not decode, re-encode, or reformat before hashing — JSON parsing and re-serialization changes whitespace and key order, which changes the hash. + +--- + +## Step 3 — Verify the transcript hash + +Parse the `hash` field to determine the algorithm, then compute and compare: + +```python +import hashlib + +expected = record["tool_transcript"]["hash"] +# expected is a DigestStr: "sha256:" or "sha384:" + +algorithm, expected_hex = expected.split(":", 1) + +if algorithm == "sha256": + computed = hashlib.sha256(transcript_bytes).hexdigest() +elif algorithm == "sha384": + computed = hashlib.sha384(transcript_bytes).hexdigest() +else: + raise ValueError(f"Unsupported digest algorithm: {algorithm}") + +if computed != expected_hex: + raise RuntimeError( + f"Transcript hash mismatch.\n" + f" Expected: {expected}\n" + f" Computed: {algorithm}:{computed}" + ) + +print(f"Transcript verified: {len(transcript_bytes)} bytes, {algorithm}:{computed[:16]}...") +``` + +If this check passes, the transcript at `transcript_uri` is byte-for-byte what was hashed when the Trust Record was signed. Combined with the signature check from Step 1, this gives you end-to-end integrity: record → hash → transcript. + +--- + +## Step 4 — Inspect individual call records + +The transcript is a JSON array of tool call records. Each entry captures one call: + +```json +[ + { + "call_index": 0, + "tool_name": "read_file", + "input_hash": "sha256:...", + "output_hash": "sha256:...", + "started_at": "2026-06-23T09:14:58Z", + "duration_ms": 142 + } +] +``` + +The inputs and outputs are themselves hashed — the raw argument and response values are not in the transcript by default. This protects sensitive tool arguments while still committing the content: + +```python +import json + +calls = json.loads(transcript_bytes) + +print(f"Total calls: {len(calls)}") +for call in calls: + print(f" [{call['call_index']}] {call['tool_name']}") + print(f" input: {call.get('input_hash', 'not committed')}") + print(f" output: {call.get('output_hash', 'not committed')}") +``` + +Cross-check against `call_count` if it was set in the Trust Record: + +```python +call_count = record["tool_transcript"].get("call_count") +if call_count is not None and len(calls) != call_count: + print(f"Warning: record says {call_count} calls but transcript has {len(calls)}") +``` + +--- + +## External execution receipts + +For high-assurance scenarios, individual calls may carry external execution receipts — signed by a third-party (the caller, an orchestrator, or a notary) rather than the agent that produced the Trust Record. + +The spec (§3.3.1) defines the receipt structure: + +| Field | Description | +|---|---| +| `issuer` | URI identifying the signing party | +| `issuer_key_id` | Key identifier within that party's key set | +| `signature` | Signature over `evidence_hash` | +| `evidence_hash` | Digest of the specific call being attested | +| `evidence_type` | Content type of the evidence (e.g., `application/json`) | +| `linked_call_id` | The call index this receipt binds to | + +To verify a receipt against a specific call: + +```python +def verify_external_receipt(call, receipt, issuer_public_key): + expected_hash = call["input_hash"] # or output_hash depending on what was attested + algorithm, expected_hex = expected_hash.split(":", 1) + + receipt_evidence_hash = receipt["evidence_hash"] + receipt_alg, receipt_hex = receipt_evidence_hash.split(":", 1) + + # The receipt's evidence_hash must match the call's committed hash + if receipt_hex != expected_hex or receipt_alg != algorithm: + raise RuntimeError( + f"Receipt evidence_hash does not match call {call['call_index']}" + ) + + # The signature covers the evidence_hash bytes (algorithm-specific) + # Verify using the issuer's public key from their published key set + # (Key retrieval from issuer URI is application-specific) + # ... + return True +``` + +!!! info "No SDK helper for receipt verification" + The `agentrust_trace` SDK does not include an issuer key resolver or receipt chain verifier. Resolution of `issuer` URIs to public keys is application-specific — typically a DID document or a published JWK Set at a well-known endpoint. + +--- + +## Putting it together + +A complete audit verification run: + +```python +from agentrust_trace.sign import verify_record, load_key +import hashlib +import json +import requests + +def verify_audit_chain(record_path, public_key_path): + with open(public_key_path, "rb") as f: + public_key = load_key(f.read()) + + with open(record_path) as f: + record = json.load(f) + + # Step 1: Verify record signature + verify_record(record, public_key_or_jwk=public_key) + print("Signature: OK") + + tt = record.get("tool_transcript") + if not tt: + print("No tool_transcript — nothing further to verify") + return + + # Step 2: Retrieve transcript + uri = tt.get("transcript_uri") + if not uri: + print("No transcript_uri — cannot retrieve transcript") + return + + transcript_bytes = requests.get(uri, timeout=30).content + + # Step 3: Hash check + algorithm, expected_hex = tt["hash"].split(":", 1) + hashfn = hashlib.sha256 if algorithm == "sha256" else hashlib.sha384 + computed = hashfn(transcript_bytes).hexdigest() + + if computed != expected_hex: + raise RuntimeError(f"Transcript hash mismatch: got {algorithm}:{computed}") + + print(f"Transcript hash: OK ({algorithm}:{computed[:16]}...)") + + # Step 4: Call count + calls = json.loads(transcript_bytes) + call_count = tt.get("call_count") + if call_count is not None: + match = "OK" if len(calls) == call_count else "MISMATCH" + print(f"Call count: {len(calls)}/{call_count} [{match}]") + else: + print(f"Calls in transcript: {len(calls)}") +``` + +--- + +## Summary + +| Step | What it proves | +|---|---| +| `verify_record()` | Record was not altered after signing; `tool_transcript.hash` is trusted | +| Transcript hash check | Transcript bytes are exactly what was hashed at signing time | +| Call count check | Transcript was not truncated | +| External receipt check | Third-party confirms specific call inputs/outputs (optional) | + +The chain of custody runs: hardware/software measurement → signed Trust Record → committed transcript hash → per-call hashes → optional external receipts. Each link is independently verifiable without contacting the operator who produced the record. diff --git a/mkdocs.yml b/mkdocs.yml index 3f6a1ac..907aa9b 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -157,6 +157,8 @@ nav: - Verify a trust record: docs/tutorials/verifying-a-trust-record.md - Hardware attestation platforms: docs/tutorials/hardware-attestation-platforms.md - Integration with cMCP: docs/tutorials/integrating-with-cmcp.md + - Anchor to the registry: docs/tutorials/anchoring-to-the-registry.md + - Verify the tool transcript: docs/tutorials/verifying-the-audit-chain.md - Specification: spec/trace-v0.1.md - Integration: - AGT: docs/integration/agt.md From 8870c540f4baa56d51556da523ec9b681e686322 Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Thu, 25 Jun 2026 13:47:41 -0700 Subject: [PATCH 2/3] feat(adapters): add TraceAGTAdapter to eliminate AGT integration boilerplate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes agentrust-io/trace-spec#64. Adds `src/agentrust_trace/adapters/agt.py` with `TraceAGTAdapter` and `AGTSessionResult`. Replaces ~50 lines of manual policy-hash + audit-transcript + measurement wiring with a single `build_trust_record()` call. Exported from `agentrust_trace.adapters` and re-exported from the top-level package. - 16 unit tests covering all field mappings, edge cases, sign/verify round-trip, and structural TrustRecord validation (56/56 suite passes) - docs/integration/agt.md updated to show adapter as the recommended path - docs/tutorials/agt-adapter.md: full walkthrough including field mapping table, how to collect AGT session inputs, and Level 0 → Level 2 upgrade path via cMCP Co-Authored-By: Claude Sonnet 4.6 --- docs/integration/agt.md | 39 ++- docs/tutorials/agt-adapter.md | 183 ++++++++++++++ src/agentrust_trace/__init__.py | 3 + src/agentrust_trace/adapters/__init__.py | 5 + .../__pycache__/__init__.cpython-312.pyc | Bin 0 -> 406 bytes .../adapters/__pycache__/agt.cpython-312.pyc | Bin 0 -> 7970 bytes src/agentrust_trace/adapters/agt.py | 192 +++++++++++++++ tests/test_agt_adapter.py | 225 ++++++++++++++++++ 8 files changed, 646 insertions(+), 1 deletion(-) create mode 100644 docs/tutorials/agt-adapter.md create mode 100644 src/agentrust_trace/adapters/__init__.py create mode 100644 src/agentrust_trace/adapters/__pycache__/__init__.cpython-312.pyc create mode 100644 src/agentrust_trace/adapters/__pycache__/agt.cpython-312.pyc create mode 100644 src/agentrust_trace/adapters/agt.py create mode 100644 tests/test_agt_adapter.py diff --git a/docs/integration/agt.md b/docs/integration/agt.md index ca253c6..fa9429f 100644 --- a/docs/integration/agt.md +++ b/docs/integration/agt.md @@ -26,7 +26,44 @@ AGT emits **Level 0 (software-only)** TRACE records. The record is signed with a pip install agentmesh agentrust-trace ``` -## Basic usage +## Quick start: TraceAGTAdapter + +`TraceAGTAdapter` eliminates the ~50-line field-mapping boilerplate. Install: + +```bash +pip install agentrust-trace +``` + +One-liner upgrade path for any AGT-governed session: + +```python +from agentrust_trace.adapters import TraceAGTAdapter, AGTSessionResult +from agentrust_trace import sign_record, generate_key + +adapter = TraceAGTAdapter( + model_provider="anthropic", + model_id="claude-sonnet-4-6", + model_version="20251001", + build_provenance_digest="sha256:e5f6a7b8...", + transparency="https://registry.agentrust.io/claim/...", +) + +session = AGTSessionResult( + agent_did="spiffe://trust.example.org/agent/my-agent", + policy_bundle_bytes=Path("policy.cedar").read_bytes(), + audit_entries=govern_fn.get_audit_entries(), + merkle_chain_tip=govern_fn.chain_tip, +) + +record = adapter.build_trust_record(session) +signed = sign_record(record, generate_key()) # or load_signing_key() for production +``` + +→ Full walkthrough: [TraceAGTAdapter tutorial](../tutorials/agt-adapter.md) + +## Manual wiring (legacy) + +The following is the raw field-mapping approach — kept for reference. Prefer `TraceAGTAdapter` for new integrations. ```python from agentmesh.governance import govern, GovernanceConfig diff --git a/docs/tutorials/agt-adapter.md b/docs/tutorials/agt-adapter.md new file mode 100644 index 0000000..7e26054 --- /dev/null +++ b/docs/tutorials/agt-adapter.md @@ -0,0 +1,183 @@ +# TraceAGTAdapter: One-line AGT → TRACE upgrade + +Replace ~50 lines of manual field wiring with a single `build_trust_record()` call. + +## What you'll learn + +- How `TraceAGTAdapter` maps AGT session data to TRACE Trust Record fields +- How to collect the three inputs AGT exposes (`policy_bundle_bytes`, `audit_entries`, `merkle_chain_tip`) +- How to sign and validate the resulting record +- How to upgrade from Level 0 (software-only) to Level 2 (hardware-rooted) inside cMCP + +## Prerequisites + +```bash +pip install agentrust-trace +``` + +--- + +## The problem: 50 lines of boilerplate per project + +Every project that integrates AGT with TRACE has to wire the same field mappings by hand: + +```python +import hashlib, json, time +from agentrust_trace import ( + TrustRecord, ModelInfo, RuntimeInfo, PolicyInfo, + ToolTranscript, BuildProvenance, Appraisal, ConfirmationKey, JWK, +) + +# Hash the Cedar bundle +bundle_bytes = Path("policy.cedar").read_bytes() +bundle_hash = "sha256:" + hashlib.sha256(bundle_bytes).hexdigest() + +# Hash the audit entries +entries_json = json.dumps(audit_entries, sort_keys=True, separators=(",", ":")) +transcript_hash = "sha256:" + hashlib.sha256(entries_json.encode()).hexdigest() + +# Hash the Merkle chain tip +measurement = "sha256:" + hashlib.sha256(chain_tip.encode()).hexdigest() + +# Build the record manually +record = TrustRecord( + eat_profile="tag:agentrust.io,2026:trace-v0.1", + iat=int(time.time()), + subject=agent_did, + model=ModelInfo(provider="anthropic", model_id="claude-sonnet-4-6", version="20251001"), + runtime=RuntimeInfo(platform="software-only", measurement=measurement), + policy=PolicyInfo(bundle_hash=bundle_hash, enforcement_mode="enforce"), + data_class="confidential", + tool_transcript=ToolTranscript(hash=transcript_hash, call_count=len(audit_entries)), + build_provenance=BuildProvenance(slsa_level=2, digest="sha256:e5f6..."), + appraisal=Appraisal(status="affirming", verifier="https://agentrust.io/verify"), + transparency="https://registry.agentrust.io/claim/...", + cnf=ConfirmationKey(jwk=JWK(kty="OKP", crv="Ed25519", x="...")), +) +``` + +`TraceAGTAdapter` encapsulates all of this. + +--- + +## The solution: TraceAGTAdapter + +```python +from pathlib import Path +from agentrust_trace.adapters import TraceAGTAdapter, AGTSessionResult +from agentrust_trace import sign_record, load_signing_key, TrustRecord + +# 1. Configure once per deployment +adapter = TraceAGTAdapter( + model_provider="anthropic", + model_id="claude-sonnet-4-6", + model_version="20251001", + build_provenance_digest="sha256:e5f6a7b8...", + transparency="https://registry.agentrust.io/claim/...", +) + +# 2. Collect AGT session data after govern_fn.close_session() +session = AGTSessionResult( + agent_did="spiffe://trust.example.org/agent/my-agent", + policy_bundle_bytes=Path("policy.cedar").read_bytes(), + audit_entries=govern_fn.get_audit_entries(), # list[dict] + merkle_chain_tip=govern_fn.chain_tip, # hex string +) + +# 3. Build and sign +record = adapter.build_trust_record(session) +key = load_signing_key() # reads TRACE_PRIVATE_KEY_PEM env var +signed = sign_record(record, key) + +# 4. Validate structure before writing +TrustRecord.model_validate(signed) + +import json +Path("session.trace.json").write_text(json.dumps(signed, indent=2)) +``` + +--- + +## Field mapping reference + +| TRACE field | Source | +|---|---| +| `subject` | `AGTSessionResult.agent_did` | +| `policy.bundle_hash` | `sha256(policy_bundle_bytes)` | +| `policy.enforcement_mode` | `TraceAGTAdapter(enforcement_mode=...)` (default: `enforce`) | +| `tool_transcript.hash` | `sha256(canonical_json(audit_entries))` | +| `tool_transcript.call_count` | `len(audit_entries)` or `AGTSessionResult.call_count` override | +| `runtime.platform` | Always `software-only` (Level 0) | +| `runtime.measurement` | `sha256(merkle_chain_tip)` | +| `appraisal.status` | Always `affirming` (Phase 1) | +| `model`, `data_class`, `build_provenance` | `TraceAGTAdapter(...)` constructor params | +| `iat`, `appraisal.timestamp` | `AGTSessionResult.iat` (default: current time) | + +--- + +## Collecting the three inputs from AGT + +### `policy_bundle_bytes` + +Read the Cedar bundle from disk immediately after calling `govern()`. The hash must match what the session evaluated against. + +```python +from pathlib import Path + +policy_bundle_bytes = Path(config.policy_path).read_bytes() +``` + +### `audit_entries` + +AGT's `govern()` returns a wrapped callable with `.get_audit_entries()`. Call it after `.close_session()`: + +```python +governed_fn = govern(my_tool, agent_did=agent_did, config=config) +result = governed_fn(input_data) +governed_fn.close_session() + +audit_entries = governed_fn.get_audit_entries() # list of Merkle AuditEntry dicts +``` + +### `merkle_chain_tip` + +The Merkle chain tip is the hash of the last `AuditEntry` in the chain: + +```python +chain_tip = governed_fn.chain_tip # hex string, e.g. "deadbeef..." +``` + +--- + +## Adapting to different enforcement modes + +```python +adapter = TraceAGTAdapter( + ... + enforcement_mode="advisory", # "enforce" | "advisory" | "silent" +) +``` + +`enforce` (default) means policy decisions are binding — tool calls blocked by a `forbid` rule do not execute. `advisory` means decisions are logged but not enforced. The mode appears in `policy.enforcement_mode` in the TRACE record so verifiers know what the policy actually did. + +--- + +## Upgrading to Level 2 (hardware-rooted) + +`TraceAGTAdapter` produces Level 0 records — `runtime.platform` is `software-only` and the signing key is not TEE-bound. For Level 2: + +1. Deploy your AGT-governed agent inside cMCP on an Azure DCasv5 (SEV-SNP) or DCesv6 (TDX) VM, or GCP N2D (SEV-SNP) or C3 (TDX) +2. cMCP measures the Cedar policy bundle into the TEE hardware at startup +3. The cMCP runtime generates a TEE-bound key and emits a Level 2 TRACE record that supersedes the Level 0 record for the same session +4. Both records share `subject` and `tool_transcript.hash` and are mutually verifiable + +The Level 0 record from `TraceAGTAdapter` remains valid — it is evidence of policy enforcement at the software layer. The Level 2 record from cMCP adds hardware attestation on top. + +→ [Deploy on Azure](deploy-azure.md) — `Standard_DC2as_v5` (SEV-SNP) or `Standard_DC2es_v6` (TDX) +→ [Deploy on GCP](deploy-gcp.md) — `n2d-standard-4` (SEV-SNP) or `c3-standard-4` (TDX) + +--- + +## Summary + +`TraceAGTAdapter` turns 50 lines of manual field wiring into three calls: configure the adapter once, collect the three AGT session values (`policy_bundle_bytes`, `audit_entries`, `merkle_chain_tip`) after each session, call `build_trust_record()`. The record is structurally valid and ready for `sign_record()` without any additional construction. diff --git a/src/agentrust_trace/__init__.py b/src/agentrust_trace/__init__.py index bdfc336..4340845 100644 --- a/src/agentrust_trace/__init__.py +++ b/src/agentrust_trace/__init__.py @@ -1,5 +1,6 @@ """agentrust-trace — TRACE Trust Record models, validation, and signing.""" +from agentrust_trace.adapters import AGTSessionResult, TraceAGTAdapter from agentrust_trace.models import ( Appraisal, BuildProvenance, @@ -29,6 +30,8 @@ __all__ = [ "__version__", + "AGTSessionResult", + "TraceAGTAdapter", "Appraisal", "BuildProvenance", "ConfirmationKey", diff --git a/src/agentrust_trace/adapters/__init__.py b/src/agentrust_trace/adapters/__init__.py new file mode 100644 index 0000000..e300d6e --- /dev/null +++ b/src/agentrust_trace/adapters/__init__.py @@ -0,0 +1,5 @@ +"""First-party adapters that map governance framework outputs to TRACE Trust Records.""" + +from agentrust_trace.adapters.agt import AGTSessionResult, TraceAGTAdapter + +__all__ = ["AGTSessionResult", "TraceAGTAdapter"] diff --git a/src/agentrust_trace/adapters/__pycache__/__init__.cpython-312.pyc b/src/agentrust_trace/adapters/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..a177fd2d85a9fe19fc870522f3d8053c90a2fbc8 GIT binary patch literal 406 zcmZ9GF-ycS6vvbH&eJ;%ot#e0dczf6MTO%~(ZQiskP--?iM84`;JVa+Iz1p^qp#Dco3XOtrL~|QkSWzmIL6CwBN&@Mc~>= zXxMbzZ&oeFgo+redOVi7shP~w0&niiKSj6I-Pz4G&Lv_(&~^-q%wcuut{ zA(Hf)G0=sKsD_CCgVQ;jGM40tVazKg)#7%hM48H6=vXqgN3YuS;bM%R(b*e1dg-i9 O*Ego^&D$0%+13}Ik$ao~ literal 0 HcmV?d00001 diff --git a/src/agentrust_trace/adapters/__pycache__/agt.cpython-312.pyc b/src/agentrust_trace/adapters/__pycache__/agt.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..201e39ee038f1f1e502499a444184fedca4570bf GIT binary patch literal 7970 zcmb7JU2GfKb)F%I|CA_O|Dr5k$?-HtpBhb&sa31A?3D2fg2hZafE^rg=M`XZJtNK9SSZSvALI^Mv*KJ}bC zGbBa3>&>-w=iL8$&pr2??;QSXPfuKcwDQ+0dmoGo!f&wPr;tmCPyY^xyMij@1XWZ6 zx>yY40wU5uJ(vsfuauMcHsV zx0uSMiha4h;<4PZ;_=+^Vt=l`IFK6<1sVwmmjpF(TTr9-1MXh#dwa_b^0pYX#r?M2 zP$ig2l$+3HlPP(+c4>1>RZ2Exg?2OlOEiElV>EVwUYv*(SD0 z6ta1B?ZQQ}$;y^ZuF|~8)NDL{m6mjv&?4VoIzx2L0Lm=DghpA>NkOB!NKMWVuH%nH#AOH(PyzjNVAHvYfJmW=2m6vHrWzNod42{Pm@43d}5GjVFXKWua{3QAmhVKuO)&%Il^@ zWpDNA%+}WOay*U`V)8e&f?GVY|LImA?5FHd!U9;@>RV{D752NM?$o>Z( zI6X%;<@9xLV`D!K&zBsK4bz}b9ABuMH$lvgh_D2*1Y#^cMWMPY*q*2fg6#>nAnbW+ z%N0U5_f^<^um3f8xcF9su%sj5OZvI+6uW-@#)akU?^%>t*R>)%sq2O2b?>};vxL|@aD;_| z=7`YO{}&q#=}#VyPSm7{hVv~Uu^~g&>9tn$7<53UznJ>wf>`cO`2>-x&KZL zR8OxwOs(F!aHqc}tu~~ryCAN0R0ZDv%5AwE-ay671R3$Y4|3dH=RwMVKG0;|j z*gv&LjouD}1Pfc{xM4{#w=(cvBSS64-?xea*p7NI< zgj}zGfu{2G5o){t$0&mFLv?c5uxtgiykb+ZNO?+1psHL$RD@?w@@Ox1%C5Pa<0cI> z8CR7cl_;Bx+fr0#e8Scw9h$aN&v`;jR=^Y=G37_q97r403PloHd785k> z4z=u+DQj0*TspT*mkQ^U^V?s~X0u%*;f)$r3GBO(udGb%+IGoWURYps2lN!HxH=jc zmetG!*sE4tI66nBZ3|qM0*gFmWnaK=&t7<=Ho@>PYL}S-h6j}*(JZJ?)Cxx!TU(a4 zV{}{M-zGm(4d6T?3;@c|SxhaI$h9Rt3ZA zwE=HE$=WnYxdb=~pbQvS*QhR%>0O1X90f7cw5ghz^I=Z~bRaj+T2X(NAeX_fXwY#;I&rRS16r0A`;m}b?N6M(b6qN))7rQBhNMR3d)>;=UU24bB-%MB$W9k zH0;@~^Xx051*gy396o+j`84kvza$(i41e4({+R2{wqZOF4+00l2jYFtKpqIPKTbH1 zpwu##b_9mBbg`Hyn zvH(1CedQtT9#>#Ih2ZqdZFaY=V6daI;|`7i;NlX}x`a%RBUw~0IKA#!{IjuhusWN= zY#y^LZf3Z9FP;u#3)tlhc1dCg44!HXo@fkAeio62<9DRLN;Kob_$zm$htZKolV=*ki;e!7#=u-7 zz3^EqGzC2#Mkks{VPXcUvBs$-{F27$Z{T;TF?sfm^zYFLZq)NFiGua|;E&6t1o>Uz zK-`0h?)2x$7uOpDZCD3AS`yeEf8Uekn*mjX9xXX+_kh}#1a{O4d)x zApux1QihJ9pgHI6s|LuY(@plwIckF|5N_a|z+QsSfOh>TvT#l79Qlw| z5K2J1qM}QUYb$^g=o|+fabLCY3BCYlxbxvuFGCwR;~LwsmRoaUjR%sJJI_P7+n;xR zcwmU{owuFa=lt{z7PuVkT4s$v6iGA4`n7Lw%!6A8H~_*g+YkxjPz>>b$%+_7f&(Ly z-AlM}CVQ4x`CVF6Tr7cG#)MD@!UaYlyZ6zDaCjc0+BJ1R!hpaT`r^bek;@ROg4hDA z2N!H>de?RMnB1L}kdj#{>oC9~n%xdq;N{@nV7m4;_KB_REO)|}IY7+kmcejHVIg5v~B(U|0>0?j(v-vEcsEqJc-8@M01Oq`1+6I*Aw1X@n3! z0fSVqDX`RGa8T^BflW6*laT*#QK}P!uV>Vf9&q7pS;=_o~RE`SBIw?gQIuf|K$Dp;B0kp_E!UkgR`~4 z)kniqjnV1GIH`}%SI6f+-Kl5atY+V=jbHjK6ddV&78X*YKivD7dS|cJJNsye9O-$v zn!Q{bUvEsFu1}t?PM(J$!(BrfV-xkUnd;cgqm%Qn0e739p0CfoUY&itHn`SEpN3JS zdldOfot&$Zb9J&-C2PO756N1MY(5&DYT$lP)JM-&N6$9WC+q2@YI>56S?`1p#l5 z!=?=&;Aa4_4d`DK-#TbPjJ*~x$7$T;hc@=1cun|uVB_(3t=tuQ~d8~b_Lk>NKFKz)GDkXev!!fzmU>-aCS>0s%-_>w!q&3*Rx z=xGmKimv4MaaouNZES*G6z3dq8CV#zOlHfVq%Ck>DEJi$#Oo|j!3HW;S+VjO=uDU~ z6Luu@>75XF>X`v`%!JGdLu>+EI);(?On~bcCze-0d20{}wGdHzPnqoiz26<8$#!dv ze}-NbYL#2UZwE&1u79%rHwJ2yy(E5zH?V$QH+29wSWZrK22YI#h$gYjKQKpu%OB$n|Td zlu~GWfs}TrBj8bc?gp--(AW}mRs%{P=H6olK%~7Prur-a;xR9P`@^vSK zc< None: + self._model = ModelInfo( + provider=model_provider, + model_id=model_id, + version=model_version, + ) + self._data_class = data_class + self._build_provenance = BuildProvenance( + slsa_level=build_provenance_slsa_level, + digest=build_provenance_digest, + builder=build_provenance_builder, + provenance_uri=build_provenance_uri, + ) + self._transparency = transparency + self._appraisal_verifier = appraisal_verifier + self._appraisal_policy_ref = appraisal_policy_ref + self._enforcement_mode = enforcement_mode + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]: + """Return an unsigned TRACE Trust Record dict for the given AGT session. + + Pass the result to ``sign_record(record, key)`` to add a signature, then + to ``TrustRecord.model_validate(record)`` for structural validation. + + Args: + session: AGT session data collected after ``govern_fn.close_session()``. + + Returns: + A plain JSON-serialisable dict conforming to the TRACE v0.1 schema. + The ``cnf.jwk`` placeholder carries no key material until ``sign_record()`` + populates it from the signing key. + """ + call_count = ( + session.call_count + if session.call_count is not None + else len(session.audit_entries) + ) + + record: dict[str, Any] = { + "eat_profile": "tag:agentrust.io,2026:trace-v0.1", + "iat": session.iat, + "subject": session.agent_did, + "model": self._model.model_dump(exclude_none=True), + "runtime": RuntimeInfo( + platform="software-only", + measurement=self._measurement(session.merkle_chain_tip), + ).model_dump(exclude_none=True), + "policy": PolicyInfo( + bundle_hash=self._bundle_hash(session.policy_bundle_bytes), + enforcement_mode=self._enforcement_mode, # type: ignore[arg-type] + ).model_dump(exclude_none=True), + "data_class": self._data_class, + "tool_transcript": ToolTranscript( + hash=self._transcript_hash(session.audit_entries), + call_count=call_count, + ).model_dump(exclude_none=True), + "build_provenance": self._build_provenance.model_dump(exclude_none=True), + "appraisal": Appraisal( + status="affirming", + verifier=self._appraisal_verifier, + policy_ref=self._appraisal_policy_ref, + timestamp=session.iat, + ).model_dump(exclude_none=True), + "transparency": self._transparency, + # cnf is populated by sign_record(); placeholder keeps schema valid + "cnf": {"jwk": {"kty": "OKP", "crv": "Ed25519", "x": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"}}, + } + return record + + # ------------------------------------------------------------------ + # Hash helpers + # ------------------------------------------------------------------ + + @staticmethod + def _bundle_hash(policy_bundle_bytes: bytes) -> str: + return "sha256:" + hashlib.sha256(policy_bundle_bytes).hexdigest() + + @staticmethod + def _transcript_hash(audit_entries: list[dict[str, Any]]) -> str: + canonical = json.dumps(audit_entries, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + return "sha256:" + hashlib.sha256(canonical.encode()).hexdigest() + + @staticmethod + def _measurement(merkle_chain_tip: str) -> str: + return "sha256:" + hashlib.sha256(merkle_chain_tip.encode()).hexdigest() diff --git a/tests/test_agt_adapter.py b/tests/test_agt_adapter.py new file mode 100644 index 0000000..e5ef937 --- /dev/null +++ b/tests/test_agt_adapter.py @@ -0,0 +1,225 @@ +"""Unit tests for TraceAGTAdapter and AGTSessionResult.""" + +from __future__ import annotations + +import hashlib +import json +import time + +import pytest +from pydantic import ValidationError + +from agentrust_trace import TrustRecord, sign_record, generate_key, verify_record +from agentrust_trace.adapters import AGTSessionResult, TraceAGTAdapter + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +BUNDLE_BYTES = b'permit(principal, action, resource);' +CHAIN_TIP = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef" +AUDIT_ENTRIES: list[dict] = [ + {"entry_id": 1, "tool": "crm.get_customer", "decision": "permit"}, + {"entry_id": 2, "tool": "support.create_ticket", "decision": "permit"}, +] +AGENT_DID = "spiffe://trust.example.org/agent/test-agent/prod" +TRANSPARENCY = "https://registry.agentrust.io/claim/test-abc123" + + +def _make_adapter(**overrides) -> TraceAGTAdapter: + defaults = dict( + model_provider="anthropic", + model_id="claude-sonnet-4-6", + model_version="20251001", + data_class="confidential", + build_provenance_slsa_level=2, + build_provenance_digest="sha256:" + "a" * 64, + transparency=TRANSPARENCY, + ) + defaults.update(overrides) + return TraceAGTAdapter(**defaults) + + +def _make_session(**overrides) -> AGTSessionResult: + defaults = dict( + agent_did=AGENT_DID, + policy_bundle_bytes=BUNDLE_BYTES, + audit_entries=AUDIT_ENTRIES, + merkle_chain_tip=CHAIN_TIP, + ) + defaults.update(overrides) + return AGTSessionResult(**defaults) + + +# --------------------------------------------------------------------------- +# 1. build_trust_record produces a structurally valid TrustRecord +# --------------------------------------------------------------------------- + +def test_build_produces_valid_trust_record() -> None: + adapter = _make_adapter() + session = _make_session() + record = adapter.build_trust_record(session) + # Must parse without ValidationError + tr = TrustRecord.model_validate(record) + assert tr.eat_profile == "tag:agentrust.io,2026:trace-v0.1" + + +# --------------------------------------------------------------------------- +# 2. runtime.platform is always software-only +# --------------------------------------------------------------------------- + +def test_runtime_platform_is_software_only() -> None: + record = _make_adapter().build_trust_record(_make_session()) + assert record["runtime"]["platform"] == "software-only" + + +# --------------------------------------------------------------------------- +# 3. policy.bundle_hash is SHA-256 of policy_bundle_bytes +# --------------------------------------------------------------------------- + +def test_policy_bundle_hash_correct() -> None: + expected = "sha256:" + hashlib.sha256(BUNDLE_BYTES).hexdigest() + record = _make_adapter().build_trust_record(_make_session()) + assert record["policy"]["bundle_hash"] == expected + + +# --------------------------------------------------------------------------- +# 4. tool_transcript.hash is SHA-256 of canonical JSON of audit_entries +# --------------------------------------------------------------------------- + +def test_transcript_hash_correct() -> None: + canonical = json.dumps(AUDIT_ENTRIES, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + expected = "sha256:" + hashlib.sha256(canonical.encode()).hexdigest() + record = _make_adapter().build_trust_record(_make_session()) + assert record["tool_transcript"]["hash"] == expected + + +# --------------------------------------------------------------------------- +# 5. runtime.measurement is SHA-256 of merkle_chain_tip string +# --------------------------------------------------------------------------- + +def test_measurement_is_sha256_of_chain_tip() -> None: + expected = "sha256:" + hashlib.sha256(CHAIN_TIP.encode()).hexdigest() + record = _make_adapter().build_trust_record(_make_session()) + assert record["runtime"]["measurement"] == expected + + +# --------------------------------------------------------------------------- +# 6. call_count defaults to len(audit_entries); explicit override is respected +# --------------------------------------------------------------------------- + +def test_call_count_defaults_to_entries_length() -> None: + record = _make_adapter().build_trust_record(_make_session()) + assert record["tool_transcript"]["call_count"] == len(AUDIT_ENTRIES) + + +def test_call_count_override_respected() -> None: + session = _make_session(call_count=99) + record = _make_adapter().build_trust_record(session) + assert record["tool_transcript"]["call_count"] == 99 + + +# --------------------------------------------------------------------------- +# 7. subject matches agent_did verbatim +# --------------------------------------------------------------------------- + +def test_subject_matches_agent_did() -> None: + record = _make_adapter().build_trust_record(_make_session()) + assert record["subject"] == AGENT_DID + + +def test_did_web_subject_accepted() -> None: + session = _make_session(agent_did="did:web:example.org:agents:my-agent") + record = _make_adapter().build_trust_record(session) + tr = TrustRecord.model_validate(record) + assert tr.subject.startswith("did:") + + +# --------------------------------------------------------------------------- +# 8. build_trust_record output survives sign_record + verify_record round-trip +# --------------------------------------------------------------------------- + +def test_sign_and_verify_round_trip() -> None: + adapter = _make_adapter() + session = _make_session() + record = adapter.build_trust_record(session) + key = generate_key() + signed = sign_record(record, key) + # Must not raise + verify_record(signed) + # Structural validation of signed record + TrustRecord.model_validate(signed) + + +# --------------------------------------------------------------------------- +# 9. enforcement_mode propagates from adapter config +# --------------------------------------------------------------------------- + +def test_enforcement_mode_propagates() -> None: + adapter = _make_adapter(enforcement_mode="advisory") + record = adapter.build_trust_record(_make_session()) + assert record["policy"]["enforcement_mode"] == "advisory" + + +# --------------------------------------------------------------------------- +# 10. empty audit_entries produces valid record with call_count=0 +# --------------------------------------------------------------------------- + +def test_empty_audit_entries() -> None: + session = _make_session(audit_entries=[]) + record = _make_adapter().build_trust_record(session) + TrustRecord.model_validate(record) + assert record["tool_transcript"]["call_count"] == 0 + + +# --------------------------------------------------------------------------- +# 11. iat propagates from session +# --------------------------------------------------------------------------- + +def test_iat_propagates_from_session() -> None: + fixed_ts = 1750000000 + session = AGTSessionResult( + agent_did=AGENT_DID, + policy_bundle_bytes=BUNDLE_BYTES, + audit_entries=AUDIT_ENTRIES, + merkle_chain_tip=CHAIN_TIP, + iat=fixed_ts, + ) + record = _make_adapter().build_trust_record(session) + assert record["iat"] == fixed_ts + assert record["appraisal"]["timestamp"] == fixed_ts + + +# --------------------------------------------------------------------------- +# 12. invalid build_provenance_digest raises at adapter construction +# --------------------------------------------------------------------------- + +def test_invalid_build_provenance_digest_raises() -> None: + with pytest.raises(ValidationError): + _make_adapter(build_provenance_digest="not-a-digest") + + +# --------------------------------------------------------------------------- +# 13. different policy bundles produce different bundle hashes +# --------------------------------------------------------------------------- + +def test_different_bundles_produce_different_hashes() -> None: + s1 = _make_session(policy_bundle_bytes=b"bundle-alpha") + s2 = _make_session(policy_bundle_bytes=b"bundle-beta") + adapter = _make_adapter() + h1 = adapter.build_trust_record(s1)["policy"]["bundle_hash"] + h2 = adapter.build_trust_record(s2)["policy"]["bundle_hash"] + assert h1 != h2 + + +# --------------------------------------------------------------------------- +# 14. different chain tips produce different measurements +# --------------------------------------------------------------------------- + +def test_different_chain_tips_produce_different_measurements() -> None: + s1 = _make_session(merkle_chain_tip="aaa" + "0" * 61) + s2 = _make_session(merkle_chain_tip="bbb" + "0" * 61) + adapter = _make_adapter() + m1 = adapter.build_trust_record(s1)["runtime"]["measurement"] + m2 = adapter.build_trust_record(s2)["runtime"]["measurement"] + assert m1 != m2 From 38204595dc86f4f4c70e3aff3a9013dfd6adc513 Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Thu, 25 Jun 2026 13:54:45 -0700 Subject: [PATCH 3/3] fix(lint): remove unused imports, fix line length, rewrite dict literals - Remove ConfirmationKey, JWK, TrustRecord from agt.py imports (unused) - Break cnf placeholder and json.dumps calls to fit 100-char limit - Remove unused time import from test_agt_adapter.py - Rewrite dict() calls as dict literals (C408) Co-Authored-By: Claude Sonnet 4.6 --- src/agentrust_trace/adapters/agt.py | 15 +++++++++----- tests/test_agt_adapter.py | 31 ++++++++++++++--------------- 2 files changed, 25 insertions(+), 21 deletions(-) diff --git a/src/agentrust_trace/adapters/agt.py b/src/agentrust_trace/adapters/agt.py index 22d1790..2f9407a 100644 --- a/src/agentrust_trace/adapters/agt.py +++ b/src/agentrust_trace/adapters/agt.py @@ -15,13 +15,10 @@ from agentrust_trace.models import ( Appraisal, BuildProvenance, - ConfirmationKey, - JWK, ModelInfo, PolicyInfo, RuntimeInfo, ToolTranscript, - TrustRecord, ) @@ -170,7 +167,13 @@ def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]: ).model_dump(exclude_none=True), "transparency": self._transparency, # cnf is populated by sign_record(); placeholder keeps schema valid - "cnf": {"jwk": {"kty": "OKP", "crv": "Ed25519", "x": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"}}, + "cnf": { + "jwk": { + "kty": "OKP", + "crv": "Ed25519", + "x": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + }, + }, } return record @@ -184,7 +187,9 @@ def _bundle_hash(policy_bundle_bytes: bytes) -> str: @staticmethod def _transcript_hash(audit_entries: list[dict[str, Any]]) -> str: - canonical = json.dumps(audit_entries, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + canonical = json.dumps( + audit_entries, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ) return "sha256:" + hashlib.sha256(canonical.encode()).hexdigest() @staticmethod diff --git a/tests/test_agt_adapter.py b/tests/test_agt_adapter.py index e5ef937..5b154d4 100644 --- a/tests/test_agt_adapter.py +++ b/tests/test_agt_adapter.py @@ -4,7 +4,6 @@ import hashlib import json -import time import pytest from pydantic import ValidationError @@ -27,26 +26,26 @@ def _make_adapter(**overrides) -> TraceAGTAdapter: - defaults = dict( - model_provider="anthropic", - model_id="claude-sonnet-4-6", - model_version="20251001", - data_class="confidential", - build_provenance_slsa_level=2, - build_provenance_digest="sha256:" + "a" * 64, - transparency=TRANSPARENCY, - ) + defaults = { + "model_provider": "anthropic", + "model_id": "claude-sonnet-4-6", + "model_version": "20251001", + "data_class": "confidential", + "build_provenance_slsa_level": 2, + "build_provenance_digest": "sha256:" + "a" * 64, + "transparency": TRANSPARENCY, + } defaults.update(overrides) return TraceAGTAdapter(**defaults) def _make_session(**overrides) -> AGTSessionResult: - defaults = dict( - agent_did=AGENT_DID, - policy_bundle_bytes=BUNDLE_BYTES, - audit_entries=AUDIT_ENTRIES, - merkle_chain_tip=CHAIN_TIP, - ) + defaults = { + "agent_did": AGENT_DID, + "policy_bundle_bytes": BUNDLE_BYTES, + "audit_entries": AUDIT_ENTRIES, + "merkle_chain_tip": CHAIN_TIP, + } defaults.update(overrides) return AGTSessionResult(**defaults)