fix(identity): align AdCP 3.3 agent identity with the final spec and the Go and TS SDKs

pyproject.toml

@@ -91,7 +91,15 @@ dependencies = [
     # streamable-HTTP session manager's explicit configuration arguments.
     "mcp>=2.0.0,<3.0",
     "email-validator>=2.0.0",
-    "cryptography>=41.0.0",
+    # Floor at 45.0.1 because joserfc (below) requires it.
+    "cryptography>=45.0.1",
+    # JOSE (RFC 7515 JWS, RFC 7517 JWK, RFC 7638 thumbprints) for
+    # ``adcp.signing.jws`` and ``adcp.signing.keygen``: the SDK keeps only the
+    # AdCP profile rules (typ, alg allow-list, strict JSON) on top. BSD-3.
+    "joserfc>=1.7.5,<2",
+    # Standard Webhooks v1, the specification's own reference package, for
+    # ``adcp.signing.standard_webhooks``. MIT, no dependencies.
+    "standardwebhooks>=1.1.0,<2",
     # RFC 8785 JSON Canonicalization Scheme — used by the server-side
     # idempotency middleware to compute the spec-mandated payload hash for
     # replay detection. Tiny pure-Python dep, no transitive weight.

scripts/export_resolution.py

@@ -39,6 +39,8 @@ def snapshot_modules() -> tuple[tuple[str, str], ...]:
         ("adcp", "adcp"),
         ("adcp.types", "adcp.types"),
         ("adcp.types._generated", "adcp.types._generated"),
+        ("adcp.signing", "adcp.signing"),
+        ("adcp.types.experimental_identity", "adcp.types.experimental_identity"),
         *(
             (f"adcp.types.domains.{domain}", f"adcp.types.domains.{domain}")
             for domain in adcp.types.domains.DOMAINS

scripts/generate_identity_preview.py

@@ -1,7 +1,9 @@
 """Generate experimental identity types from pinned draft schemas, offline.
 
 The stable AdCP wire pin and generated domain tree are intentionally unchanged.
-The two onboarding slots are an explicit composition of independent P5/P6 drafts.
+Each source schema is copied verbatim from the open spec branch recorded in
+_identity_schemas/UPSTREAM.json; trust/v1/trust.json is composed there (spec main
+plus the onboarding and grants branch diffs), so this script composes nothing.
 """
 
 from __future__ import annotations
@@ -38,18 +40,6 @@ def load(path):
         if not source.exists():
             source = STABLE / path
         schema = json.loads(source.read_text())
-        # Compose the independently proposed buyer posture slots without editing
-        # the immutable source schemas or claiming the composition is a release.
-        if path == "trust/v1/trust.json":
-            for name in ("agent", "record_profiles"):
-                profiles = (
-                    schema["definitions"][name]["properties"]["profiles"]
-                    if name == "agent"
-                    else schema["definitions"][name]
-                )
-                profiles["allOf"][1]["properties"]["adcp"]["properties"]["onboarding"] = {
-                    "$ref": "/schemas/core/buyer-onboarding-posture.json"
-                }
 
         def walk(value):
             if isinstance(value, dict):

src/adcp/decisioning/handler.py

@@ -1714,13 +1714,13 @@ async def _resolve_account(
                 ctx.metadata.get(_BUYER_AGENT_METADATA_KEY),
                 signing_profile=self._grant_signing_profile(ctx),
                 operation=tool_name,
-                bound_wba_identity=self._has_bound_wba_identity(ctx),
                 operation_params=(
                     request_params.model_dump(mode="json", exclude_unset=True)
                     if hasattr(request_params, "model_dump")
                     else request_params
                 ),
                 spend_committing=self._wba_spend_decision(ctx),
+                wba_revocation=self._wba_revocation(ctx),
             )
         elif self._brand_authorization_gate is not None:
             buyer_agent_for_gate = cast(
@@ -1741,15 +1741,13 @@ def _wba_spend_decision(cls, ctx: ToolContext) -> bool | None:
         return value if isinstance(value, bool) else None
 
     @classmethod
-    def _has_bound_wba_identity(cls, ctx: ToolContext) -> bool:
-        """Only trusted transport state can preserve an established binding."""
+    def _wba_revocation(cls, ctx: ToolContext) -> str | None:
+        """The verified signer's revocation state, from trusted transport metadata."""
         auth_info = cls._extract_auth_info(ctx)
-        return bool(
-            auth_info is not None
-            and auth_info.extra.get("signing_profile") == "web-bot-auth"
-            and isinstance(auth_info.extra.get("principal_id"), str)
-            and auth_info.extra["principal_id"]
-        )
+        if auth_info is None or auth_info.extra.get("signing_profile") != "web-bot-auth":
+            return None
+        value = auth_info.extra.get("wba_revocation")
+        return value if isinstance(value, str) else None
 
     @classmethod
     def _grant_signing_profile(
@@ -3177,14 +3175,20 @@ async def sync_accounts(  # type: ignore[override]
                     if hasattr(declaration, "model_dump")
                     else declaration
                 )
-                if isinstance(row, dict) and row.get("account_id") and "brand" not in row:
-                    await self._resolve_account(row, tool_ctx, tool_name="sync_accounts")
-                else:
-                    await self._trust_grant_gate.require_declaration(
-                        row,
-                        tool_ctx.metadata.get(_BUYER_AGENT_METADATA_KEY),
-                        signing_profile=self._grant_signing_profile(tool_ctx),
+                if isinstance(row, dict) and row.get("account_id"):
+                    # An existing account's authority comes from seller-owned
+                    # state, never from brand/operator fields beside its id.
+                    await self._resolve_account(
+                        {"account_id": row["account_id"]}, tool_ctx, tool_name="sync_accounts"
                     )
+                    if "brand" not in row and "operator" not in row:
+                        continue
+                await self._trust_grant_gate.require_declaration(
+                    row if isinstance(row, Mapping) else {},
+                    tool_ctx.metadata.get(_BUYER_AGENT_METADATA_KEY),
+                    signing_profile=self._grant_signing_profile(tool_ctx),
+                    wba_revocation=self._wba_revocation(tool_ctx),
+                )
         resolve_ctx = self._make_resolve_context(tool_ctx, "sync_accounts")
         if callable(upsert_request):
             result = _call_with_optional_ctx(upsert_request, params, ctx=resolve_ctx)