adapters: make AGT transparency optional and validate session fields (#534)

docs/tutorials/agt-adapter.md

@@ -24,13 +24,11 @@ session = AGTSessionResult(
 adapter = TraceAGTAdapter(
     model_provider="example", model_id="synthetic-demo",
     build_provenance_digest="sha256:" + "e" * 64,
-    transparency="https://example.test/unused",
     enforcement_mode="declared",  # synthetic input: no policy engine ran
 )
 record = adapter.build_trust_record(session)
 assert record["policy"]["enforcement_mode"] == "declared"
-# The adapter currently requires a URI argument but performs no registration.
-record.pop("transparency")
+assert "transparency" not in record
 # `appraisal.status` defaults to "none", which is correct here: synthetic input has not
 # been appraised. Pass appraisal_status only when an appraisal actually happened.
 key = generate_key()

src/agentrust_trace/adapters/agt.py

@@ -11,6 +11,7 @@
 from dataclasses import dataclass, field
 from typing import Any, Literal
 
+from agentrust_trace._patterns import _PYTHON_PATTERNS
 from agentrust_trace.models import (
     JCS_SAFE_INTEGER,
     Appraisal,
@@ -27,6 +28,67 @@
 # failure this check exists to stop -- so the floor is the contract's, not zero.
 TRACE_MIN_IAT = 1700000000
 
+_SUBJECT_RE = _PYTHON_PATTERNS[r"^(spiffe://[^/]+/.+|did:[a-z0-9]+:.+)$"]
+
+
+def _check_agent_did(value: object) -> None:
+    if not isinstance(value, str):
+        raise ValueError(f"agent_did must be a string, got {type(value).__name__}")
+    if not _SUBJECT_RE.match(value):
+        raise ValueError(
+            f"agent_did {value!r} must be a SPIFFE URI "
+            "('spiffe://<trust-domain>/<path>') or a DID ('did:<method>:<id>'). "
+            "It becomes the record subject, which is what a verifier keys on."
+        )
+
+
+def _check_policy_bundle_bytes(value: object) -> None:
+    if not isinstance(value, bytes):
+        raise ValueError(
+            f"policy_bundle_bytes must be bytes, got {type(value).__name__}"
+        )
+
+
+def _check_audit_entries(value: object) -> None:
+    if not isinstance(value, list) or any(not isinstance(entry, dict) for entry in value):
+        raise ValueError(
+            "audit_entries must be a list of dict objects, got "
+            f"{type(value).__name__}"
+        )
+
+
+def _check_merkle_chain_tip(value: object) -> None:
+    if not isinstance(value, str) or not value:
+        raise ValueError(
+            f"merkle_chain_tip must be a non-empty string, got {value!r}"
+        )
+    try:
+        value.encode("utf-8")
+    except UnicodeEncodeError as exc:
+        raise ValueError(
+            "merkle_chain_tip must be valid UTF-8 without lone surrogates"
+        ) from exc
+
+
+def _check_call_count(value: object) -> None:
+    if value is None:
+        return
+    if (
+        not isinstance(value, int)
+        or isinstance(value, bool)
+        or value < 0
+        or value > JCS_SAFE_INTEGER
+    ):
+        raise ValueError(
+            f"call_count must be an integer within [0, {JCS_SAFE_INTEGER}] or None, "
+            f"got {value!r}"
+        )
+
+
+def _check_transparency(value: object) -> None:
+    if value is not None and (not isinstance(value, str) or not value):
+        raise ValueError("transparency must be a non-empty URI string or None")
+
 
 def _check_iat(value: object) -> None:
     """The bound ``TrustRecord.iat`` carries, applied to whatever is about to be signed.
@@ -86,6 +148,11 @@ class AGTSessionResult:
     """Issuance timestamp. Defaults to now."""
 
     def __post_init__(self) -> None:
+        _check_agent_did(self.agent_did)
+        _check_policy_bundle_bytes(self.policy_bundle_bytes)
+        _check_audit_entries(self.audit_entries)
+        _check_merkle_chain_tip(self.merkle_chain_tip)
+        _check_call_count(self.call_count)
         _check_iat(self.iat)
 
 
@@ -125,7 +192,7 @@ def __init__(
         build_provenance_digest: str,
         build_provenance_builder: str | None = None,
         build_provenance_uri: str | None = None,
-        transparency: str,
+        transparency: str | None = None,
         appraisal_status: Literal["affirming", "warning", "contraindicated", "none"] = "none",
         appraisal_verifier: str = "https://agentrust-io.com/verify",
         appraisal_policy_ref: str | None = None,
@@ -144,7 +211,10 @@ def __init__(
                 being built, and stamping ``affirming`` on an unappraised record puts a
                 verdict in the field a consumer reads to find out whether anybody
                 checked. Set this only when an appraisal actually happened.
+            transparency: SCITT receipt URI. ``None`` means unanchored, which is correct
+                below Level 2 and leaves the key out of the record.
         """
+        _check_transparency(transparency)
         self._model = ModelInfo(
             provider=model_provider,
             model_id=model_id,
@@ -181,14 +251,24 @@ def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]:
             The ``cnf.jwk`` placeholder carries no key material until ``sign_record()``
             populates it from the signing key.
         """
+        if not isinstance(session, AGTSessionResult):
+            raise ValueError(
+                f"session must be an AGTSessionResult, got {type(session).__name__}"
+            )
+        _check_agent_did(session.agent_did)
+        _check_policy_bundle_bytes(session.policy_bundle_bytes)
+        _check_audit_entries(session.audit_entries)
+        _check_merkle_chain_tip(session.merkle_chain_tip)
+        _check_call_count(session.call_count)
+        _check_iat(session.iat)
+        _check_transparency(self._transparency)
+
         call_count = (
             session.call_count
             if session.call_count is not None
             else len(session.audit_entries)
         )
 
-        _check_iat(session.iat)
-
         record: dict[str, Any] = {
             "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2",
             "iat": session.iat,
@@ -214,7 +294,6 @@ def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]:
                 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": {
@@ -224,6 +303,8 @@ def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]:
                 },
             },
         }
+        if self._transparency is not None:
+            record["transparency"] = self._transparency
         return record
 
     # ------------------------------------------------------------------
@@ -232,14 +313,17 @@ def build_trust_record(self, session: AGTSessionResult) -> dict[str, Any]:
 
     @staticmethod
     def _bundle_hash(policy_bundle_bytes: bytes) -> str:
+        _check_policy_bundle_bytes(policy_bundle_bytes)
         return "sha256:" + hashlib.sha256(policy_bundle_bytes).hexdigest()
 
     @staticmethod
     def _transcript_hash(audit_entries: list[dict[str, Any]]) -> str:
         # RFC 8785 (JCS): same canonicalization TraceSandboxAdapter already uses
         # for the same field, and what docs/schema.md calls "canonical JSON".
+        _check_audit_entries(audit_entries)
         return "sha256:" + hashlib.sha256(_canonical_bytes(audit_entries)).hexdigest()
 
     @staticmethod
     def _measurement(merkle_chain_tip: str) -> str:
+        _check_merkle_chain_tip(merkle_chain_tip)
         return "sha256:" + hashlib.sha256(merkle_chain_tip.encode()).hexdigest()

src/agentrust_trace/adapters/sandbox.py

@@ -130,6 +130,71 @@ def _check_iat(value: object) -> None:
 _DIGEST_RE = _PYTHON_PATTERNS[r"^sha(256:[0-9a-f]{64}|384:[0-9a-f]{96})$"]
 
 
+def _check_sandbox_id(value: object) -> None:
+    if not isinstance(value, str):
+        raise ValueError(f"sandbox_id must be a string, got {type(value).__name__}")
+    if not _SUBJECT_RE.match(value):
+        raise ValueError(
+            f"sandbox_id {value!r} must be a SPIFFE URI "
+            "('spiffe://<trust-domain>/<path>') or a DID ('did:<method>:<id>'). "
+            "It becomes the record subject, which is what a verifier keys on."
+        )
+
+
+def _check_image_digest(value: object) -> None:
+    if not isinstance(value, str):
+        raise ValueError(
+            f"image_digest must be a string, got {type(value).__name__}"
+        )
+    if not _DIGEST_RE.match(value):
+        raise ValueError(
+            f"image_digest {value!r} must be a sha256: or sha384: digest."
+        )
+
+
+def _check_policy_bundle_bytes(value: object) -> None:
+    if not isinstance(value, bytes):
+        raise ValueError(
+            f"policy_bundle_bytes must be bytes, got {type(value).__name__}"
+        )
+
+
+def _check_decisions(value: object) -> None:
+    if not isinstance(value, list) or any(not isinstance(d, dict) for d in value):
+        raise ValueError(
+            f"decisions must be a list of dict objects, got {type(value).__name__}"
+        )
+
+
+def _check_call_count(value: object) -> None:
+    if value is None:
+        return
+    if (
+        not isinstance(value, int)
+        or isinstance(value, bool)
+        or value < 0
+        or value > JCS_SAFE_INTEGER
+    ):
+        raise ValueError(
+            f"call_count must be an integer within [0, {JCS_SAFE_INTEGER}] or None, "
+            f"got {value!r}"
+        )
+
+
+def _check_transparency(value: object) -> None:
+    if value is not None and (not isinstance(value, str) or not value):
+        raise ValueError("transparency must be a non-empty URI string or None")
+
+
+def _check_bundle_hash(value: object) -> None:
+    if not isinstance(value, str):
+        raise ValueError(f"bundle_hash must be a string, got {type(value).__name__}")
+    if not _DIGEST_RE.match(value):
+        raise ValueError(
+            f"bundle_hash {value!r} must be a sha256: or sha384: digest."
+        )
+
+
 @dataclass(frozen=True)
 class SandboxAttestation:
     """Hardware evidence for the machine a sandbox ran on.
@@ -203,6 +268,16 @@ def __post_init__(self) -> None:
             )
 
 
+def _check_attestation(value: object) -> None:
+    if value is None:
+        return
+    if not isinstance(value, SandboxAttestation):
+        raise ValueError(
+            f"attestation must be a SandboxAttestation or None, got {type(value).__name__}"
+        )
+    value.__post_init__()
+
+
 @dataclass(frozen=True)
 class SandboxSessionResult:
     """One sandbox session, as the runtime has it at close.
@@ -238,22 +313,12 @@ class SandboxSessionResult:
     """Issuance timestamp. Defaults to now."""
 
     def __post_init__(self) -> None:
-        if not isinstance(self.sandbox_id, str):
-            raise ValueError(f"sandbox_id must be a string, got {type(self.sandbox_id).__name__}")
-        if not _SUBJECT_RE.match(self.sandbox_id):
-            raise ValueError(
-                f"sandbox_id {self.sandbox_id!r} must be a SPIFFE URI "
-                "('spiffe://<trust-domain>/<path>') or a DID ('did:<method>:<id>'). "
-                "It becomes the record subject, which is what a verifier keys on."
-            )
-        if not isinstance(self.image_digest, str):
-            raise ValueError(
-                f"image_digest must be a string, got {type(self.image_digest).__name__}"
-            )
-        if not _DIGEST_RE.match(self.image_digest):
-            raise ValueError(
-                f"image_digest {self.image_digest!r} must be a sha256: or sha384: digest."
-            )
+        _check_sandbox_id(self.sandbox_id)
+        _check_image_digest(self.image_digest)
+        _check_policy_bundle_bytes(self.policy_bundle_bytes)
+        _check_decisions(self.decisions)
+        _check_attestation(self.attestation)
+        _check_call_count(self.call_count)
         _check_iat(self.iat)
 
 
@@ -311,6 +376,7 @@ def __init__(
             build_provenance_slsa_level: Defaults to 0, meaning no provenance claim. Raise
                 it only when a builder genuinely produced the attested level.
         """
+        _check_transparency(transparency)
         self._model = ModelInfo(
             provider=model_provider, model_id=model_id, version=model_version
         )
@@ -346,11 +412,22 @@ def build_trust_record(self, session: SandboxSessionResult) -> dict[str, Any]:
         passed here. See the module docstring and :class:`SandboxAttestation` for this
         boundary.
         """
+        if not isinstance(session, SandboxSessionResult):
+            raise ValueError(
+                f"session must be a SandboxSessionResult, got {type(session).__name__}"
+            )
+        _check_sandbox_id(session.sandbox_id)
+        _check_image_digest(session.image_digest)
+        _check_policy_bundle_bytes(session.policy_bundle_bytes)
+        _check_decisions(session.decisions)
+        _check_attestation(session.attestation)
+        _check_call_count(session.call_count)
+        _check_iat(session.iat)
+        _check_transparency(self._transparency)
+
         bundle_hash = self.bundle_hash(session.policy_bundle_bytes)
         runtime = self._runtime(session, bundle_hash)
 
-        _check_iat(session.iat)
-
         record: dict[str, Any] = {
             "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2",
             "iat": session.iat,
@@ -404,6 +481,7 @@ def build_trust_record(self, session: SandboxSessionResult) -> dict[str, Any]:
     @staticmethod
     def bundle_hash(policy_bundle_bytes: bytes) -> str:
         """SHA-256 of the policy bundle bytes, exactly as supplied."""
+        _check_policy_bundle_bytes(policy_bundle_bytes)
         return "sha256:" + hashlib.sha256(policy_bundle_bytes).hexdigest()
 
     @staticmethod
@@ -416,6 +494,7 @@ def transcript_hash(decisions: list[dict[str, Any]]) -> str:
         pre-image (see :func:`~agentrust_trace.sign.sign_record`) keeps a record
         reproducible by an implementation in another language.
         """
+        _check_decisions(decisions)
         return "sha256:" + hashlib.sha256(_canonical_bytes(decisions)).hexdigest()
 
     @staticmethod
@@ -431,6 +510,8 @@ def software_measurement(image_digest: str, bundle_hash: str) -> str:
         ``platform: "software-only"``, which the model documents as never to be mistaken
         for hardware-backed evidence.
         """
+        _check_image_digest(image_digest)
+        _check_bundle_hash(bundle_hash)
         return (
             "sha256:"
             + hashlib.sha256(f"{image_digest}\n{bundle_hash}".encode()).hexdigest()