Module adcp.decisioning.validate_capabilities
Boot-time validation of the projected get_adcp_capabilities response.
The framework auto-projects :class:DecisioningCapabilities into a
spec-shaped get_adcp_capabilities response (see
:meth:PlatformHandler.get_adcp_capabilities()).
Adopters may also override capabilities per request via
DecisioningPlatform.get_adcp_capabilities_for_request or override the
projection on a handler subclass. Either way, the response that ships on
the wire must satisfy the
protocol/get-adcp-capabilities-response.json schema and the
spec invariants the schema cannot fully express on its own (e.g.
"account.supported_billing must exist and be non-empty whenever the
seller claims media_buy").
This module exercises the projection at boot — invokes
handler.get_adcp_capabilities() with a synthetic request and
validates the returned dict — so misconfiguration surfaces as a
structured :class:AdcpError before the server starts accepting
traffic. The historical motivator is the v3 reference seller, which
shipped a non-conformant capabilities response until #402 added
supported_billing manually; the validator below would have caught
that at boot.
Functions
def validate_capabilities_response_shape(handler: PlatformHandler) ‑> None-
Expand source code
def validate_capabilities_response_shape(handler: PlatformHandler) -> None: """Boot-time validator for the projected capabilities response. Calls ``handler.get_adcp_capabilities()`` with a synthetic request, then enforces: 1. The response validates against the bundled ``protocol/get-adcp-capabilities-response.json`` schema (via :func:`adcp.validation.schema_validator.validate_response`). 2. ``supported_protocols`` is present and non-empty (spec ``minItems: 1``; doubled-up here so the diagnostic names the invariant directly). 3. When the seller claims ``media_buy``, ``account.supported_billing`` is present and non-empty (the invariant the v3 ref seller violated pre-#402; spec ``protocol/get-adcp-capabilities-response.json`` requires ``account.required: ["supported_billing"]`` with ``minItems: 1``). Synchronous entry point — drives the async handler via :func:`asyncio.run`, which means **this function cannot be called from inside a running event loop**. Async callers (test fixtures, Starlette ``lifespan`` handlers, anything inside ``asyncio.run``) should use :func:`validate_capabilities_response_shape_async` instead and pair it with ``create_adcp_server_from_platform(..., validate_at_init=False)``. :raises AdcpError: ``INVALID_REQUEST`` with ``recovery="terminal"`` on any violation; ``details`` carry the offending response and a structured issue list so operators can index the failure programmatically. :raises RuntimeError: when called from inside a running event loop (the ``asyncio.run`` machinery raises this directly). """ _validate_response_dict(_invoke_capabilities(handler))Boot-time validator for the projected capabilities response.
Calls
handler.get_adcp_capabilities()with a synthetic request, then enforces:- The response validates against the bundled
protocol/get-adcp-capabilities-response.jsonschema (via :func:validate_response()). supported_protocolsis present and non-empty (specminItems: 1; doubled-up here so the diagnostic names the invariant directly).- When the seller claims
media_buy,account.supported_billingis present and non-empty (the invariant the v3 ref seller violated pre-#402; specprotocol/get-adcp-capabilities-response.jsonrequiresaccount.required: ["supported_billing"]withminItems: 1).
Synchronous entry point — drives the async handler via :func:
asyncio.run, which means this function cannot be called from inside a running event loop. Async callers (test fixtures, Starlettelifespanhandlers, anything insideasyncio.run) should use :func:validate_capabilities_response_shape_async()instead and pair it withcreate_adcp_server_from_platform(..., validate_at_init=False).:raises AdcpError:
INVALID_REQUESTwithrecovery="terminal"on any violation;detailscarry the offending response and a structured issue list so operators can index the failure programmatically. :raises RuntimeError: when called from inside a running event loop (theasyncio.runmachinery raises this directly). - The response validates against the bundled
async def validate_capabilities_response_shape_async(handler: PlatformHandler) ‑> None-
Expand source code
async def validate_capabilities_response_shape_async(handler: PlatformHandler) -> None: """Async sibling of :func:`validate_capabilities_response_shape`. Identical diagnostic surface; awaits ``handler.get_adcp_capabilities()`` directly instead of driving it through :func:`asyncio.run`. Use this from async contexts (test fixtures, Starlette ``lifespan``, in-process A2A test clients) so the SDK doesn't try to spin up a second event loop and crash with ``RuntimeError: asyncio.run() cannot be called from a running event loop``. Typical pairing — async caller bypasses the init-time sync validation and runs the async validator themselves:: handler, executor, registry = create_adcp_server_from_platform( platform, validate_at_init=False, ) await validate_capabilities_response_shape_async(handler) """ _validate_response_dict(await handler.get_adcp_capabilities())Async sibling of :func:
validate_capabilities_response_shape().Identical diagnostic surface; awaits
handler.get_adcp_capabilities()directly instead of driving it through :func:asyncio.run. Use this from async contexts (test fixtures, Starlettelifespan, in-process A2A test clients) so the SDK doesn't try to spin up a second event loop and crash withRuntimeError: asyncio.run() cannot be called from a running event loop.Typical pairing — async caller bypasses the init-time sync validation and runs the async validator themselves::
handler, executor, registry = create_adcp_server_from_platform( platform, validate_at_init=False, ) await validate_capabilities_response_shape_async(handler) def validate_platform(platform: DecisioningPlatform) ‑> None-
Expand source code
def validate_platform(platform: DecisioningPlatform) -> None: """Server-boot validator — fail-fast before the first request. Checks (in order): 1. ``platform.capabilities`` is a populated :class:`DecisioningCapabilities` (not the base default). 2. ``platform.accounts`` is a real :class:`AccountStore` (anything truthy with a ``resolve`` method) — None catches subclasses that forgot to attach a store. 3. Each claimed specialism's required methods are implemented on the platform subclass. Unknown specialisms emit ``UserWarning`` (forward-compat with v6.x+ specs); known specialisms missing methods raise an INVALID_REQUEST error. 4. Each claimed specialism's *recommended* methods (the v6.0 rc.1 staging set in :data:`RECOMMENDED_METHODS_PER_SPECIALISM` — sales-* surface broadening per DX-423) are implemented on the platform subclass. Misses emit one ``UserWarning`` per method (deduped across overlapping specialisms). Setting ``ADCP_DECISIONING_STRICT_VALIDATE_PLATFORM=1`` flips the soft warning into a hard INVALID_REQUEST error. 5. **Governance opt-in fail-fast (D15 round-4):** if any claimed specialism is in :data:`GOVERNANCE_SPECIALISMS` AND ``capabilities.governance_aware`` is False AND the platform hasn't wired a custom :class:`StateReader` (i.e., the dispatch hydration helper would supply ``_NotYetWiredStateReader``), raise. Silent governance-gate skipping is a security regression the framework refuses to ship. Catches per-validator exceptions and re-projects to ``AdcpError("INVALID_REQUEST")`` so server boot never crashes with a raw stack trace — the operator sees one structured diagnostic per problem (Round-4 Emma #16). :raises AdcpError: on any blocking validation failure. The error ``details`` carry per-issue diagnostics for operator triage. """ if not isinstance(platform.capabilities, DecisioningCapabilities): raise AdcpError( "INVALID_REQUEST", message=( "DecisioningPlatform.capabilities must be a " "DecisioningCapabilities instance — found " f"{type(platform.capabilities).__name__!r}. Subclasses MUST " "set ``capabilities = DecisioningCapabilities(...)`` on the " "class body." ), recovery="terminal", ) accounts = getattr(platform, "accounts", None) if accounts is None: raise AdcpError( "INVALID_REQUEST", message=( "DecisioningPlatform.accounts is None — subclasses MUST set " "an AccountStore (SingletonAccounts, ExplicitAccounts, " "FromAuthAccounts, or a custom AccountStore impl) on the " "class body." ), recovery="terminal", ) # Specialism-method coverage. # ``capabilities.specialisms`` is ``list[Specialism | str]`` — # spec-known entries are coerced to enum at construction; novel / # pre-spec slugs pass through as strings (so this validator can # surface them with typo-vs-novel diagnostics). Lookup tables are # keyed by AdCP slug strings, so extract a slug regardless of form. missing: list[tuple[str, str]] = [] unknown: list[str] = [] governance_specialisms_claimed: list[str] = [] for entry in platform.capabilities.specialisms: specialism = entry.value if hasattr(entry, "value") else entry if specialism in GOVERNANCE_SPECIALISMS: governance_specialisms_claimed.append(specialism) try: required = REQUIRED_METHODS_PER_SPECIALISM.get(specialism) except Exception as exc: # Defensive: a custom REQUIRED_METHODS_PER_SPECIALISM impl # (test-monkeypatch, etc.) that raises must not crash boot. # Round-4 Emma #16 — wrap validator throws. logger.warning( "REQUIRED_METHODS_PER_SPECIALISM lookup raised for %r: %r", specialism, exc, ) required = None if required is None: unknown.append(specialism) continue for method_name in required: if not _has_overridden_method(platform, method_name): missing.append((specialism, method_name)) if unknown: # Three buckets: # - typo: close-match to any spec slug → hard fail with hint # - unenforced: spec-recognized but no method-coverage rules in # this framework version → soft UserWarning (Protocol lands # in v6.1+) # - novel: not in spec at all → forward-compat UserWarning # The typo detector compares against the full spec enum (not just # REQUIRED_METHODS keys) so misspelling a spec slug we don't yet # enforce still surfaces as a typo. spec_known = sorted(SPEC_SPECIALISM_ENUM) typo_suggestions: list[tuple[str, str]] = [] unenforced: list[str] = [] novel: list[str] = [] for slug in unknown: if slug in SPEC_SPECIALISM_ENUM: # Spec-recognized but not in REQUIRED_METHODS — adopter # claimed a real spec slug whose Protocol hasn't shipped # method-coverage rules yet. unenforced.append(slug) continue close = difflib.get_close_matches(slug, spec_known, n=1, cutoff=0.7) if close: typo_suggestions.append((slug, close[0])) else: novel.append(slug) if typo_suggestions: hints = "; ".join( f"{slug!r} → did you mean {match!r}?" for slug, match in sorted(typo_suggestions) ) raise AdcpError( "INVALID_REQUEST", message=( f"DecisioningPlatform claims unknown specialism(s) " f"that look like typos: {hints}. " "Forward-compat tolerance applies only to genuinely " "novel specialism slugs (not close spelling matches). " f"Known spec specialisms: {spec_known}" ), recovery="terminal", details={ "typo_suggestions": [ {"claimed": slug, "did_you_mean": match} for slug, match in typo_suggestions ], "spec_specialisms": spec_known, }, ) if unenforced: warnings.warn( ( f"DecisioningPlatform claims spec-recognized specialism(s) " f"{sorted(unenforced)!r} that this framework version " f"doesn't yet enforce method coverage for. The claim is " f"valid; required-method validation is skipped until the " f"per-Protocol coverage lands. Implement the spec methods " f"on your platform subclass so buyers don't 404." ), UserWarning, stacklevel=2, ) if novel: warnings.warn( ( f"DecisioningPlatform claims novel specialism(s) " f"{sorted(novel)!r} that aren't in the spec enum at " f"schemas/cache/enums/specialism.json. Your framework " f"version predates the spec, OR you're piloting a future " f"specialism. Required-method validation skipped. " f"Known spec specialisms: {spec_known}" ), UserWarning, stacklevel=2, ) if missing: raise AdcpError( "INVALID_REQUEST", message=( "DecisioningPlatform claims specialisms but is missing " f"required methods: {missing}. Implement each on your " "subclass or remove the specialism from " "capabilities.specialisms." ), recovery="terminal", details={"missing": [{"specialism": s, "method": m} for s, m in missing]}, ) # Recommended (v6.0 rc.1 staging) coverage — soft-warn by default, # hard-fail under ``ADCP_DECISIONING_STRICT_VALIDATE_PLATFORM=1``. # Dedup by method name: a platform claiming both ``sales-guaranteed`` # and ``sales-non-guaranteed`` shares the same recommended set, so # ``get_media_buys`` should warn once, not twice. We walk specialisms # in declared order and remember the first specialism that surfaced # each missing method — that becomes the "blame" specialism in the # diagnostic. recommended_missing: list[tuple[str, str]] = [] seen_methods: set[str] = set() for entry in platform.capabilities.specialisms: specialism = entry.value if hasattr(entry, "value") else entry recommended = RECOMMENDED_METHODS_PER_SPECIALISM.get(specialism) if recommended is None: continue for method_name in sorted(recommended): if method_name in seen_methods: continue if not _has_overridden_method(platform, method_name): recommended_missing.append((specialism, method_name)) seen_methods.add(method_name) if recommended_missing: if _strict_validate_platform(): raise AdcpError( "INVALID_REQUEST", message=( "DecisioningPlatform claims sales-* specialism(s) but is " f"missing v6.0 rc.1 required methods: {recommended_missing}. " "Strict mode is enabled " f"({_STRICT_VALIDATE_ENV}=1); implement each on your " "subclass. See the SalesPlatform Protocol docstring at " "src/adcp/decisioning/specialisms/sales.py:184-227 for the " "canonical method list." ), recovery="terminal", details={ "missing_recommended": [ {"specialism": s, "method": m} for s, m in recommended_missing ], "strict_env_var": _STRICT_VALIDATE_ENV, }, ) # ``stacklevel=3`` so the warning points at the adopter's # ``serve(platform)`` call site, not the SDK internals # (validate_platform is invoked from serve, which is invoked by # the adopter — three frames up lands on adopter code). for specialism, method_name in recommended_missing: warnings.warn( ( f"DecisioningPlatform claims {specialism!r} but is missing " f"{method_name!r} — required by the SalesPlatform Protocol " "for any sales-* specialism in v6.0 rc.1+. See the Protocol " "docstring at src/adcp/decisioning/specialisms/sales.py:" "184-227 for the full required method list. The framework " "currently soft-warns to ease v6.0 rc.1 migration; set " f"{_STRICT_VALIDATE_ENV}=1 to fail-fast at boot instead." ), UserWarning, stacklevel=3, ) # Governance opt-in fail-fast (D15 round-4). if governance_specialisms_claimed and not platform.capabilities.governance_aware: raise AdcpError( "INVALID_REQUEST", message=( f"Platform claims governance-* specialism(s) " f"{governance_specialisms_claimed!r} but " "capabilities.governance_aware is False. Set " "governance_aware=True AND wire a custom StateReader that " "returns real GovernanceContextJWS values, OR drop the " "governance-* specialism claim. Silent governance-gate " "skipping is a security boundary; the framework refuses " "to ship that. See " "docs/proposals/decisioning-platform-dispatch-design.md#d15" ), recovery="terminal", details={ "governance_specialisms": sorted(governance_specialisms_claimed), "governance_aware": False, }, )Server-boot validator — fail-fast before the first request.
Checks (in order):
platform.capabilitiesis a populated :class:DecisioningCapabilities(not the base default).platform.accountsis a real :class:AccountStore(anything truthy with aresolvemethod) — None catches subclasses that forgot to attach a store.- Each claimed specialism's required methods are implemented
on the platform subclass. Unknown specialisms emit
UserWarning(forward-compat with v6.x+ specs); known specialisms missing methods raise an INVALID_REQUEST error. - Each claimed specialism's recommended methods (the v6.0 rc.1
staging set in :data:
RECOMMENDED_METHODS_PER_SPECIALISM— sales-* surface broadening per DX-423) are implemented on the platform subclass. Misses emit oneUserWarningper method (deduped across overlapping specialisms). SettingADCP_DECISIONING_STRICT_VALIDATE_PLATFORM=1flips the soft warning into a hard INVALID_REQUEST error. - Governance opt-in fail-fast (D15 round-4): if any claimed
specialism is in :data:
GOVERNANCE_SPECIALISMSANDcapabilities.governance_awareis False AND the platform hasn't wired a custom :class:StateReader(i.e., the dispatch hydration helper would supply_NotYetWiredStateReader), raise. Silent governance-gate skipping is a security regression the framework refuses to ship.
Catches per-validator exceptions and re-projects to
AdcpError("INVALID_REQUEST")so server boot never crashes with a raw stack trace — the operator sees one structured diagnostic per problem (Round-4 Emma #16).:raises AdcpError: on any blocking validation failure. The error
detailscarry per-issue diagnostics for operator triage.