Module adcp.decisioning.dispatch
Dispatch layer for the v6.0 DecisioningPlatform framework.
The dispatch layer ties everything together at the seam between the
existing adcp.server transport machinery and the new
DecisioningPlatform Protocol-driven adopter shape:
- :func:
validate_platform()— server-boot fail-fast: confirms every claimed specialism has its required methods, governance opt-in is honored, andaccountsis a realAccountStore. - :func:
compose_caller_identity()— composite cache scope keyf"{store_qualname}:{account.id}"(round-3 D9 — structural cross-store isolation). - :func:
_build_request_context— the hydration helper that turns aToolContext+ resolvedAccountinto a typedRequestContextper D2 / D9 / D15. - :func:
_invoke_platform_method— the method-call seam. Detects async-vs-sync, runs sync on a thread-pool executor withcontextvarssnapshot, projectsTaskHandoffreturns, wraps non-AdcpErrorexceptions toINTERNAL_ERROR(wire never leaks a stack trace). - :func:
_project_handoff— TaskHandoff lifecycle: allocatestask_id, projects the wireSubmittedenvelope, kicks off the adopter's handoff fn in the background, persists terminal artifact via the task registry.
Codegen-emitted handler.py (Stage 3 next file) calls
_invoke_platform_method from each typed shim; serve.py
(Stage 3 last) wires the executor + registry + middleware.
This module is framework-internal — adopters import nothing from
here. The Protocol contracts adopters write against live in
:mod:adcp.decisioning.specialisms.*.
Global variables
var REQUIRED_METHODS_PER_SPECIALISM : dict[str, frozenset[str]]-
Drift policy: when a specialism Protocol gains a required method, bump this map AND add a v6.x migration note. The v6.0 enforced subset is intentionally narrow — adding a method here without a Protocol behind it would break adopters mid-version.
var SPEC_SPECIALISM_ENUM : frozenset[str]-
Drift policy: when the spec adds a specialism, bump this constant. A unit test (
test_spec_specialism_enum_matches_schema_cache) reads the on-disk enum and asserts equality, so out-of-band drift surfaces in CI.
Functions
def compose_caller_identity(account: Account[Any], store: AccountStore[Any]) ‑> str-
Expand source code
def compose_caller_identity( account: Account[Any], store: AccountStore[Any], ) -> str: """Compose the cache scope key from ``module + qualname + account.id``. Round-3 D9 + Round-4 review: the framework's idempotency middleware reads ``ctx.caller_identity`` for cache scoping. Using ``account.id`` alone leaks across stores when two adopters use different ``AccountStore`` impls but happen to mint colliding ids. The composite ``f"{store_module}.{store_qualname}:{account.id}"`` gives structural cross-store isolation at zero coordination cost. Includes ``__module__`` because ``__qualname__`` is the dotted path *within* a module — two ``MyStore`` classes in different packages share the same qualname. Without the module prefix the isolation promise breaks across cross-package re-implementations. Empty / whitespace ``account.id`` raises ``AdcpError`` — ``Account(id="")`` would silently collapse every tenant whose AccountStore returns the empty default into a single cache scope. The dataclass default ``Account(id="<unset>")`` is also rejected so a misconfigured store that forgets to populate ``id`` fails fast rather than leaking buy-side data. Within-store collisions (one impl, identical ``account.id`` for two distinct accounts) remain an adopter bug at ``AccountStore.resolve``; the framework can't structurally prevent that without a runtime registry costing more than it buys. """ if not account.id or not account.id.strip() or account.id == "<unset>": raise AdcpError( "INVALID_REQUEST", message=( f"AccountStore returned an account with empty/unset id " f"({account.id!r}). The framework refuses to scope the " "idempotency cache by an empty key — every empty-id " "tenant would share state. Fix: ensure your " "AccountStore.resolve always returns Account(id=<non-empty>) " "and never leaves the dataclass default." ), recovery="terminal", ) cls = type(store) return f"{cls.__module__}.{cls.__qualname__}:{account.id}"Compose the cache scope key from
module + qualname + account.id.Round-3 D9 + Round-4 review: the framework's idempotency middleware reads
ctx.caller_identityfor cache scoping. Usingaccount.idalone leaks across stores when two adopters use differentAccountStoreimpls but happen to mint colliding ids. The compositef"{store_module}.{store_qualname}:{account.id}"gives structural cross-store isolation at zero coordination cost.Includes
__module__because__qualname__is the dotted path within a module — twoMyStoreclasses in different packages share the same qualname. Without the module prefix the isolation promise breaks across cross-package re-implementations.Empty / whitespace
account.idraisesAdcpError—Account(id="")would silently collapse every tenant whose AccountStore returns the empty default into a single cache scope. The dataclass defaultAccount(id="<unset>")is also rejected so a misconfigured store that forgets to populateidfails fast rather than leaking buy-side data.Within-store collisions (one impl, identical
account.idfor two distinct accounts) remain an adopter bug atAccountStore.resolve; the framework can't structurally prevent that without a runtime registry costing more than it buys. 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.