test(adcp): spec-conformance test for AdcpError code names against error-code.json

src/adcp/decisioning/dispatch.py

@@ -16,7 +16,7 @@
 * :func:`_invoke_platform_method` — the method-call seam. Detects
   async-vs-sync, runs sync on a thread-pool executor with
   ``contextvars`` snapshot, projects ``TaskHandoff`` returns, wraps
-  non-``AdcpError`` exceptions to ``INTERNAL_ERROR`` (wire never
+  non-``AdcpError`` exceptions to ``X_INTERNAL_ERROR`` (wire never
   leaks a stack trace).
 * :func:`_project_handoff` — TaskHandoff lifecycle: allocates
   ``task_id``, projects the wire ``Submitted`` envelope, kicks off
@@ -344,7 +344,7 @@
 
 
 def _internal_error_message(method_name: str, exc: BaseException) -> str:
-    """Build the wire-side ``message`` for an INTERNAL_ERROR wrap.
+    """Build the wire-side ``message`` for an X_INTERNAL_ERROR wrap.
 
     Adopters debugging "An internal error occurred" with no breadcrumb
     have to grep server logs to even see which exception fired (Emma
@@ -356,7 +356,7 @@ def _internal_error_message(method_name: str, exc: BaseException) -> str:
 
 
 def _internal_error_details(exc: BaseException) -> dict[str, Any]:
-    """Build the wire-side ``details`` payload for an INTERNAL_ERROR
+    """Build the wire-side ``details`` payload for an X_INTERNAL_ERROR
     wrap.
 
     ``details.caused_by`` carries the exception class name + truncated
@@ -814,7 +814,7 @@ async def _invoke_platform_method(
     Submitted envelope.
 
     Wraps any non-:class:`AdcpError` exception to
-    ``AdcpError("INTERNAL_ERROR", recovery="terminal")`` so the wire
+    ``AdcpError("X_INTERNAL_ERROR", recovery="terminal")`` so the wire
     response never leaks a stack trace. Adopters get the original
     exception logged via the framework's observability hooks (the
     raise re-raises the wrapped error; the original is the
@@ -855,7 +855,7 @@ async def _invoke_platform_method(
     except TypeError as exc:
         # Most likely an arg_projector signature-drift bug — adopter
         # renamed update_media_buy's `patch` kwarg → `update`, etc.
-        # Bare INTERNAL_ERROR would hide the cause; project to
+        # Bare X_INTERNAL_ERROR would hide the cause; project to
         # INVALID_REQUEST with a hint pointing at the adopter's
         # method signature so they fix it without a server-log dive.
         # Note: server logs see the full traceback; wire response
@@ -881,11 +881,11 @@ async def _invoke_platform_method(
             ) from exc
         # Non-projected TypeError — fall through to generic wrap.
         logger.exception(
-            "Unhandled exception in platform.%s — wrapping to INTERNAL_ERROR",
+            "Unhandled exception in platform.%s — wrapping to X_INTERNAL_ERROR",
             method_name,
         )
         raise AdcpError(
-            "INTERNAL_ERROR",
+            "X_INTERNAL_ERROR",
             message=_internal_error_message(method_name, exc),
             recovery="terminal",
             details=_internal_error_details(exc),
@@ -903,11 +903,11 @@ async def _invoke_platform_method(
         # on secret material doesn't leak the secret value through
         # the wire response.
         logger.exception(
-            "Unhandled exception in platform.%s — wrapping to INTERNAL_ERROR",
+            "Unhandled exception in platform.%s — wrapping to X_INTERNAL_ERROR",
             method_name,
         )
         raise AdcpError(
-            "INTERNAL_ERROR",
+            "X_INTERNAL_ERROR",
             message=_internal_error_message(method_name, exc),
             recovery="terminal",
             details=_internal_error_details(exc),
@@ -957,7 +957,7 @@ async def _project_handoff(
        calls ``registry.complete(task_id, result.model_dump() if
        Pydantic else result)``; on :class:`AdcpError` calls
        ``registry.fail(task_id, error.to_wire())``; on any other
-       exception, wraps to ``INTERNAL_ERROR`` and calls
+       exception, wraps to ``X_INTERNAL_ERROR`` and calls
        ``registry.fail``.
     4. Returns the wire ``Submitted`` envelope dict to the synchronous
        caller (the platform method's typed shim), which projects it
@@ -1004,7 +1004,7 @@ async def _run() -> None:
                 task_id,
             )
             wrapped = AdcpError(
-                "INTERNAL_ERROR",
+                "X_INTERNAL_ERROR",
                 message=(
                     f"Background task for {method_name!r} raised "
                     f"{type(exc).__name__}; see details for cause"

src/adcp/decisioning/handler.py

@@ -432,11 +432,11 @@ async def _resolve_buyer_agent(
         else:
             # Defensive: a future Credential variant lands and the
             # dispatch path doesn't know how to route it. Fail closed
-            # with INTERNAL_ERROR rather than silently passing the
+            # with X_INTERNAL_ERROR rather than silently passing the
             # request through (which would skip the registry gate
             # entirely and leak the upgrade footgun into production).
             raise AdcpError(
-                "INTERNAL_ERROR",
+                "X_INTERNAL_ERROR",
                 message=(
                     f"BuyerAgentRegistry dispatch received an unknown "
                     f"Credential variant {type(credential).__name__!r}. "
@@ -1130,7 +1130,7 @@ def _require_platform_method(self, method_name: str) -> None:
         at server boot by ``validate_platform``; optional methods can
         legitimately be absent and need a runtime gate. Without this,
         a buyer calling an optional method on a platform that doesn't
-        implement it would see ``INTERNAL_ERROR`` from the
+        implement it would see ``X_INTERNAL_ERROR`` from the
         AttributeError wrapper in ``_invoke_platform_method`` —
         adopter contract violation, not buyer-fixable.
         """

src/adcp/decisioning/registry.py

@@ -19,7 +19,7 @@
 drift is invisible to buyers. With :class:`BuyerAgent.billing_capabilities`
 the framework rejects mismatched ``billing`` values with a structured
 :class:`adcp.decisioning.AdcpError`
-``code="BILLING_NOT_PERMITTED_FOR_AGENT"``.
+``code="X_BILLING_NOT_PERMITTED_FOR_AGENT"``.
 
 Per :issue:`adcp-client#1269` and the v3-identity-bundle RFC, the
 registry is *durable* infrastructure — it stays useful even after
@@ -338,7 +338,7 @@ def validate_billing_for_agent(
     agent: BuyerAgent,
 ) -> None:
     """Raise :class:`adcp.decisioning.AdcpError`
-    ``BILLING_NOT_PERMITTED_FOR_AGENT`` when ``requested_billing`` is
+    ``X_BILLING_NOT_PERMITTED_FOR_AGENT`` when ``requested_billing`` is
     not in ``agent.billing_capabilities``.
 
     Called by the framework's ``sync_accounts`` shim before invoking
@@ -371,7 +371,7 @@ def validate_billing_for_agent(
         details["suggested_billing"] = suggested
 
     raise AdcpError(
-        "BILLING_NOT_PERMITTED_FOR_AGENT",
+        "X_BILLING_NOT_PERMITTED_FOR_AGENT",
         message=(
             f"Buyer agent {agent.agent_url!r} is not authorized for "
             f"billing={requested_billing!r}. Common cause: this agent "

tests/test_buyer_agent_registry.py

@@ -6,7 +6,7 @@
 * Three implementer postures (signing-only / bearer-only / mixed)
   reject the off-posture credential type by returning ``None``.
 * :func:`validate_billing_for_agent` accepts permitted modes and
-  raises ``BILLING_NOT_PERMITTED_FOR_AGENT`` on others, with details
+  raises ``X_BILLING_NOT_PERMITTED_FOR_AGENT`` on others, with details
   scoped to ``rejected_billing`` (and optional ``suggested_billing``)
   — the full ``permitted_billing`` subset is never leaked.
 * ``BuyerAgent`` defaults match the pre-trust beta passthrough-only
@@ -294,7 +294,7 @@ def test_validate_billing_rejects_passthrough_only_with_agent_billing() -> None:
     )
     with pytest.raises(AdcpError) as exc:
         validate_billing_for_agent(requested_billing="agent", agent=agent)
-    assert exc.value.code == "BILLING_NOT_PERMITTED_FOR_AGENT"
+    assert exc.value.code == "X_BILLING_NOT_PERMITTED_FOR_AGENT"
     assert exc.value.field == "billing"
     assert exc.value.recovery == "correctable"
     details = exc.value.details
@@ -319,7 +319,7 @@ def test_validate_billing_rejects_advertiser_when_not_in_capabilities() -> None:
     )
     with pytest.raises(AdcpError) as exc:
         validate_billing_for_agent(requested_billing="advertiser", agent=agent)
-    assert exc.value.code == "BILLING_NOT_PERMITTED_FOR_AGENT"
+    assert exc.value.code == "X_BILLING_NOT_PERMITTED_FOR_AGENT"
     assert "advertiser" in str(exc.value)
     # Sanity: with a non-empty permitted set, suggested_billing is set.
     assert exc.value.details["suggested_billing"] in {"agent", "operator"}

tests/test_decisioning_dispatch.py

@@ -639,7 +639,7 @@ async def get_products(self, req, ctx):
             executor=executor,
             registry=InMemoryTaskRegistry(),
         )
-    assert exc_info.value.code == "INTERNAL_ERROR"
+    assert exc_info.value.code == "X_INTERNAL_ERROR"
     assert exc_info.value.recovery == "terminal"
     # Original exception preserved as __cause__ for server-side
     # debugging.
@@ -726,7 +726,7 @@ async def get_products(self, req, ctx):
             executor=executor,
             registry=InMemoryTaskRegistry(),
         )
-    assert exc_info.value.code == "INTERNAL_ERROR"
+    assert exc_info.value.code == "X_INTERNAL_ERROR"
     # ``caused_by`` still surfaces the exception class for triage.
     assert exc_info.value.details["caused_by"]["type"] == "ValidationError"
     # NEW: ``validation_errors`` is populated with structured field
@@ -944,7 +944,7 @@ async def _handoff_fn(task_ctx):
     rec = await registry.get(envelope["task_id"], expected_account_id="acct_a")
     assert rec is not None
     assert rec["state"] == "failed"
-    assert rec["error"]["code"] == "INTERNAL_ERROR"
+    assert rec["error"]["code"] == "X_INTERNAL_ERROR"
     # Original exception text NOT exposed.
     assert "internal bug" not in rec["error"].get("message", "")
 

tests/test_decisioning_handler.py

@@ -167,7 +167,7 @@ async def get_products(self, req, ctx):
         await handler.get_products(
             GetProductsRequest(buying_mode="brief", brief="any"), ToolContext()
         )
-    assert exc_info.value.code == "INTERNAL_ERROR"
+    assert exc_info.value.code == "X_INTERNAL_ERROR"
     # Original exception preserved as __cause__; not exposed in message.
     assert isinstance(exc_info.value.__cause__, KeyError)
 

tests/test_error_code_conformance.py

@@ -0,0 +1,154 @@
+"""Spec-conformance test: AdcpError code names vs. the error-code.json enum.
+
+Walks every AdcpError(...) constructor call in src/adcp/ using the AST and
+asserts each literal code string is either:
+  - in the canonical schemas/cache/enums/error-code.json enum, OR
+  - prefixed with X_ (vendor-extension mechanism per the spec).
+
+If this test fails on your PR:
+  - If your code is a custom/platform-specific extension: prefix with X_
+    (e.g. X_CHECKOUT_BLOCKED).
+  - If your code belongs in the spec: open an issue on adcontextprotocol/adcp
+    first; add it to the allowlist below once the spec PR merges.
+
+Born from issue #375 / PR #393, where AGENT_SUSPENDED, AGENT_BLOCKED,
+REQUEST_AUTH_UNRECOGNIZED_AGENT, and INVALID_BILLING_MODEL shipped as non-spec
+codes for months before being caught in manual review.
+
+Note: examples/ is excluded from this scan. The reference seller in
+examples/v3_reference_seller/ may use non-spec codes that pre-date this
+conformance gate; fixing them is tracked separately.
+"""
+
+from __future__ import annotations
+
+import ast
+import json
+from pathlib import Path
+
+import pytest
+
+_SCHEMA_PATH = (
+    Path(__file__).parent.parent / "schemas" / "cache" / "enums" / "error-code.json"
+)
+_SRC_PATH = Path(__file__).parent.parent / "src" / "adcp"
+
+# Codes not yet in the spec enum but named in the spec prose as forthcoming.
+# Each entry MUST have a comment citing the spec reference so it can be removed
+# once the spec formalizes the split.
+#
+# AUTH_INVALID: the AUTH_REQUIRED enumDescription explicitly states "A future
+# minor release splits this code into AUTH_MISSING (correctable) and AUTH_INVALID
+# (terminal)". Keep this entry until that split lands in the schema.
+_PROVISIONAL_FUTURE_SPEC_CODES: frozenset[str] = frozenset({"AUTH_INVALID"})
+
+
+def _collect_adcp_error_codes(
+    src_path: Path,
+) -> tuple[list[tuple[str, str, int]], int]:
+    """AST-walk src_path for AdcpError() calls; return (literals, dynamic_count).
+
+    literals: list of (code_value, repo-relative path, lineno) for calls where
+              the code argument is a string constant.
+    dynamic_count: number of AdcpError() calls where the code is not a literal
+                   (e.g. a variable or f-string). Should stay zero — any
+                   non-zero value means new dynamic code construction was added,
+                   which bypasses this conformance gate.
+    """
+    literals: list[tuple[str, str, int]] = []
+    dynamic_count = 0
+    repo_root = src_path.parent.parent
+
+    for py_file in src_path.rglob("*.py"):
+        # Skip auto-generated code — do not edit generated_poc/ files.
+        if "generated_poc" in py_file.parts:
+            continue
+        try:
+            source = py_file.read_text(encoding="utf-8")
+            tree = ast.parse(source, filename=str(py_file))
+        except SyntaxError:
+            continue
+
+        for node in ast.walk(tree):
+            if not isinstance(node, ast.Call):
+                continue
+            func = node.func
+            if not (isinstance(func, ast.Name) and func.id == "AdcpError"):
+                continue
+
+            # Locate the code argument: first positional or keyword 'code'.
+            code_node: ast.expr | None = None
+            if node.args:
+                code_node = node.args[0]
+            else:
+                for kw in node.keywords:
+                    if kw.arg == "code":
+                        code_node = kw.value
+                        break
+
+            if code_node is None:
+                continue
+
+            rel = str(py_file.relative_to(repo_root))
+            if isinstance(code_node, ast.Constant) and isinstance(
+                code_node.value, str
+            ):
+                literals.append((code_node.value, rel, node.lineno))
+            else:
+                dynamic_count += 1
+
+    return literals, dynamic_count
+
+
+@pytest.fixture(scope="module")
+def spec_codes() -> frozenset[str]:
+    if not _SCHEMA_PATH.exists():
+        pytest.skip(
+            "schemas/cache/enums/error-code.json not found — run from repo checkout"
+        )
+    data = json.loads(_SCHEMA_PATH.read_text(encoding="utf-8"))
+    codes = frozenset(data.get("enum", []))
+    if not codes:
+        pytest.fail(
+            f"{_SCHEMA_PATH} exists but contains no 'enum' list — "
+            "schema structure may have changed upstream"
+        )
+    return codes
+
+
+def test_no_dynamic_adcp_error_codes() -> None:
+    """AdcpError code args must always be string literals so the AST scan is complete."""
+    _, dynamic_count = _collect_adcp_error_codes(_SRC_PATH)
+    assert dynamic_count == 0, (
+        f"Found {dynamic_count} AdcpError() call(s) where the code argument is "
+        "not a string literal (variable, f-string, or expression). "
+        "Dynamic codes bypass this conformance gate — use a literal string instead."
+    )
+
+
+def test_all_adcp_error_codes_are_spec_or_vendor_prefixed(
+    spec_codes: frozenset[str],
+) -> None:
+    """Every AdcpError code in src/adcp/ must be in the spec enum or use X_ prefix."""
+    literals, _ = _collect_adcp_error_codes(_SRC_PATH)
+
+    violations = [
+        (code, path, lineno)
+        for code, path, lineno in literals
+        if not (
+            code in spec_codes
+            or code.startswith("X_")
+            or code in _PROVISIONAL_FUTURE_SPEC_CODES
+        )
+    ]
+
+    assert not violations, (
+        "Non-spec AdcpError codes found. Each must be in "
+        "schemas/cache/enums/error-code.json, use the X_ vendor prefix, "
+        "or be added to _PROVISIONAL_FUTURE_SPEC_CODES with a spec citation.\n"
+        + "\n".join(
+            f"  {code!r} at {path}:{lineno}"
+            f" — add X_ prefix or open a spec issue"
+            for code, path, lineno in violations
+        )
+    )

tests/test_tier2_spec_conformance.py

@@ -10,7 +10,7 @@
 
 * All four denial paths surface a code from the spec vocabulary
   (``PERMISSION_DENIED`` for the three commercial-identity paths;
-  ``BILLING_NOT_PERMITTED_FOR_AGENT`` for the billing-capability path —
+  ``X_BILLING_NOT_PERMITTED_FOR_AGENT`` for the billing-capability path —
   see PR notes for the spec status of the billing code).
 * Recognized-but-denied paths (suspended / blocked) carry
   ``details.scope="agent"`` + ``details.status``.
@@ -323,7 +323,7 @@ def test_billing_validation_carries_rejected_billing() -> None:
     )
     with pytest.raises(AdcpError) as exc:
         validate_billing_for_agent(requested_billing="agent", agent=agent)
-    assert exc.value.code == "BILLING_NOT_PERMITTED_FOR_AGENT"
+    assert exc.value.code == "X_BILLING_NOT_PERMITTED_FOR_AGENT"
     assert exc.value.recovery == "correctable"
     assert exc.value.details["rejected_billing"] == "agent"