fix(decisioning): supervise timed and canceled work

[bokelley]

Summary

• supervise synchronous work that outlives request cancellation or time budgets
• bound timed synchronous admission, including router-backed platforms
• preserve idempotency and proposal lifecycle ownership until underlying work completes

Why

Timed-out or cancelled requests could release reservations while worker threads were still mutating state. Under load this allowed duplicate work and unbounded executor queues.

Validation

• focused decisioning, time-budget, proposal, and router saturation tests
• independent concurrency review against current origin/main

Compatibility

Admission and executor context now propagate through eager, lazy, proposal-manager, and registry routers. Existing direct asynchronous platform behavior is unchanged.

[aao-ipr-bot]

Approving. Right shape: a cancelled or timed-out request can no longer release a durable reservation while the worker thread is still mutating state, and the fix bounds the pre-existing thread-pool slot leak instead of papering over it.

The load-bearing move is asyncio.shield over the wrapped worker future plus a supervised settle-task (_settle_cancelled_sync_lifecycle) that fires on_complete/on_failure from the worker's real terminal outcome — so the CONSUMING→CONSUMED transition happens even after the buyer disconnects. Verified in test_sync_create_media_buy_cancellation_waits_for_worker_success.

Things I checked

• Permit accounting is leak-free. SyncExecutorAdmission acquires before executor.submit, releases exactly once via the future's done-callback (fires on success and worker exception), and releases in the except on submit failure without adding the callback. Release is marshaled back with loop.call_soon_threadsafe — correct, since asyncio.Semaphore is not thread-safe (time_budget.py:95, dispatch.py:1445-1470).
• No double settlement. Direct-sync cancel sets sync_lifecycle_continues=True in the inner except asyncio.CancelledError; the re-raised CancelledError then re-enters the outer except BaseException, which sees the flag and skips _safe_on_failure_call. Router-path cancel carries the _adcp_sync_worker_future marker and is settled once in the outer handler. security-reviewer and code-reviewer both traced this — single settlement each path.
• _project_invocation_result extraction is faithful. params→request_params rename threads through; the sync/handoff/workflow arms are unchanged; happy path is unchanged.
• get_adcp_capabilities INTERNAL_ERROR alignment closes a wire leak. It was the last outlier hand-rolling caused_by={type, message}; it now routes through _internal_error_message/_internal_error_details, which emit {type} only. str(exc) is deliberately dropped so a platform that raises on secret material can't leak an OAuth secret on the wire. Confirmed by the new assertions at test_decisioning_capabilities_projection.py:640-642.
• Not a wire-contract break. ad-tech-protocol-expert: sound — per schemas/cache/3.1/core/error.json, details is additionalProperties: true and caused_by is not a defined property; the normative recovery signal is error.recovery (still terminal). incomplete[]-when-saturated is invisible to the buyer and matches the existing timeout contract.
• No tenant/auth regression. _run_sync_delegate carries only admission + executor through two ContextVars; the tenant-scoped ctx is still a positional arg, and _bind_routed_sync_execution resets both tokens in finally so nothing bleeds across sibling delegates.
• proposal_dispatch.py except Exception→except BaseException. Awaiting release_consumption inside a CancelledError handler is safe — the exception is already caught, the bare raise re-raises it, and the inner except BaseException absorbs a second cancel during release. This is the point: cancellation must not strand a CONSUMING reservation.

Follow-ups (non-blocking — file as issues)

• Routed, no-deadline get_products cancel drops its on_complete. In _run_sync_delegate the sync_admission is None branch (platform_router.py:135-139) sets the marker and re-raises, but dispatch's outer handler only builds a settle-task when on_failure is not None. get_products wires on_complete=_persist_draft_hook with no on_failure, so on client-disconnect the worker is orphaned (stray "Task exception was never retrieved" on failure; draft-persist silently skipped on success). No reservation is held on this path, so it's not corruption — but it's the one gap in "supervise cancelled work." Supervise whenever on_complete is not None or on_failure is not None.
• asyncio.BoundedSemaphore over Semaphore. Accounting is correct today, but a future stray release() would silently raise the ceiling above timed_sync_get_products_limit rather than fail-fast (time_budget.py:95).
• Admission is per-handler, not per-tenant. One tenant bursting slow/cancelled timed get_products can hold all workers//2 permits and stall other tenants' timed calls. Strictly better than the pre-PR shared-pool exhaustion; key on ctx.account if per-tenant fairness matters.

Minor nits (non-blocking)

1. Stale comment now contradicts the invariant. dispatch.py:1529-1532 still reads "We expose only the exception class name + str (not the traceback)." _internal_error_details no longer exposes str — that omission is the whole point of the sanitization this PR leans on. Pre-existing, but adjacent enough that a maintainer could "restore" the str to match the comment and reintroduce the leak. Drop "+ str."

Independent concurrency review claimed in the PR body; the test matrix (direct-sync, eager/lazy router, campaign-unit bypass, saturation-without-queue-growth) covers the new paths well. Safe to merge once CI is green.

[KonstantinMirin]

Review — PR #1000

Overview — The direct-sync path now does what the title says. asyncio.shield over the wrapped worker plus _settle_cancelled_sync_lifecycle fires the lifecycle hooks from the worker's real terminal outcome, and test_sync_create_media_buy_cancellation_waits_for_worker_success proves the CONSUMING→CONSUMED transition survives a buyer disconnect. Permit accounting is single-release on both submit sites. Two things need work before this lands: the new except BaseException arm in _run() releases a proposal reservation when a background handoff task is cancelled, which lets a retry double-book a media buy; and both supervision gates key on on_failure, which get_products never wires, so the timed path the PR is named for is the one path that stays unsupervised.

Should fix

Findings 2, 4, 5 and 10 are one root, already open as #1004: a policy re-detected by inspecting objects (on_failure is not None, executor._max_workers) or hand-copied to a subset of its sites, instead of owned at one boundary. Evidence from this PR is added there; the sites below stand on their own.

1. Cancelling a background handoff task releases the reservation while the adopter's work is still outstanding

src/adcp/decisioning/dispatch.py:1972-1977:

except BaseException as exc:
    if on_failure is not None:
        await _safe_on_failure_call(on_failure, exc, method_name)
    raise

For create_media_buy that hook is _release_reservation_hook → release_proposal_reservation → CONSUMING → COMMITTED. Background handoff tasks are detached create_tasks (dispatch.py:2069); ordinary loop/SIGTERM shutdown cancels them. Cancel one after the handoff fn has issued its upstream create:

HEAD   state after bg cancel: ProposalState.COMMITTED | registry row: submitted
       RETRY ACCEPTED -> {'media_buy_id': 'mb_duplicate_2'}
BASE   state after bg cancel: ProposalState.CONSUMING
       RETRY REJECTED -> AdcpError[PROPOSAL_NOT_COMMITTED / correctable]

One committed proposal, two media buys upstream. The arm also skips _fail(), so the registry row stays submitted with no terminal webhook — the buyer is left holding a task that never terminates, which is exactly what provokes the retry.

Root cause: this arm infers "our asyncio task was cancelled" ⇒ "the adopter's work did not happen". That is the inference the rest of this PR exists to refute on the request path. The handoff fn is not shielded here, and for a sync fn it is a run_in_executor thread that cannot be stopped at all.

Either do not fire on_failure from this arm (leave CONSUMING, which fails closed and is what eviction exists for), or mirror _settle_cancelled_sync_lifecycle — shield the fn and settle from its real terminal outcome. Either way a release must be paired with a terminal registry state so the buyer is not invited to double-book. Lines 1972-1977 never execute under the suite; add a test that cancels a background handoff task with a reserved proposal.

Separately, "then remain cancellation to the task scheduler" in that comment does not parse.

2. Supervision keys on on_failure, so no timed get_products is supervised

Raised in the aao-ipr-bot review of 2026-07-29 ("Routed, no-deadline get_products cancel drops its on_complete"), still open on this push, and wider than the routed no-deadline case — it covers the whole timed path.

Both settle sites gate on the same field:

• src/adcp/decisioning/dispatch.py:1469 — except asyncio.CancelledError: if on_failure is not None:
• src/adcp/decisioning/dispatch.py:1605 — if isinstance(nested_sync_future, asyncio.Future) and on_failure is not None:

get_products is the only method that carries a deadline and the only caller that passes sync_admission, and it wires on_complete=_persist_draft_hook with no on_failure (src/adcp/decisioning/handler.py:1974-1979). Cancel a sync get_products mid-worker and let the worker finish:

on_complete only     -> ON_COMPLETE CALLS: []
add on_failure=noop  -> ON_COMPLETE CALLS: [{'products': []}]

The gate is the whole difference. Three consequences follow from it:

• The persist-draft hook is silently dropped on every timed-out sync get_products.
• The shielded future has no owner in the else-branch, so a worker that raises after the deadline lands in asyncio's default handler. New at HEAD, absent at b7ef1dfc:

  ERROR:asyncio:Future exception was never retrieved
  future: <Future finished exception=RuntimeError('upstream token sk-SECRET-LEAK rejected')>
  Traceback (most recent call last): ... (full adopter traceback)
  

  That is the adopter's raw exception text at ERROR, outside the framework's error funnel — while the same PR strips str(exc) from details.caused_by so a platform raising on secret material cannot leak it. The routed branch has the same shape via asyncio.create_task (platform_router.py:137-139) and produces "Task exception was never retrieved".
• _adcp_sync_worker_future, set at platform_router.py:166, has one consumer (dispatch.py:1604) behind the same gate, so on the get_products path the marker is written and never read.

The same root shows up one function over: the new pre_handoff_reject → on_failure branch at dispatch.py:1668-1669 is dead by construction. pre_handoff_reject= is passed by exactly two callers (handler.py:1976 get_products, handler.py:2676 get_signals); on_failure= by exactly one (handler.py:2192 create_media_buy). No caller passes both, and line 1669 never executes.

Root cause: the lifecycle contract is the pair (on_complete, on_failure) and the supervision predicate reads one half of it. It should ask whether a lifecycle exists at all. The spawn block — create_task(_settle_cancelled_sync_lifecycle(...)) + _SUPERVISED_SYNC_LIFECYCLES.add + discard — is also written out twice verbatim at 1471-1487 and 1607-1623, which is what let both copies share the wrong predicate.

Gate both sites on on_complete is not None or on_failure is not None, extract the spawn into one _supervise_sync_lifecycle(...), and attach a consumer to the shielded future in every branch so a post-cancellation worker failure is logged by the framework rather than by asyncio's GC hook. Add a cancellation test for an on_complete-only timed sync get_products, covering both the late-success and late-raise legs.

3. except BaseException around awaited cleanup swallows the cancellation

Three handlers widened from except Exception to except BaseException around an await, then log and continue:

• src/adcp/decisioning/dispatch.py:1774 (_safe_on_failure_call)
• src/adcp/decisioning/proposal_dispatch.py:755
• src/adcp/decisioning/proposal_dispatch.py:888

on_failure hooks do durable-store I/O, so they contain await points where a cancellation lands. When it does, the hook's CancelledError is logged and discarded and the caller re-raises the original exception. Adopter raises AdcpError, request task is cancelled while on_failure is mid store round-trip:

HEAD   RESULT: cancellation SWALLOWED, task raised AdcpError[INTERNAL_ERROR / terminal]
       cancelled() -> False | reservation released -> []
BASE   RESULT: task honoured cancellation
       cancelled() -> True  | reservation released -> []

Worst of both: a graceful-shutdown drain doing task.cancel(); await task gets a completed task back, and the reservation release is left half-done anyway.

Root cause: log-and-continue is the right policy for Exception and the wrong one for a BaseException that is not an Exception. Catch BaseException so the cleanup runs, then re-raise when the caught exception is not an Exception — the hook's cancellation is not the framework's to absorb.

One more at proposal_dispatch.py:747: the widened handler guards the try: at lines 698-746, and that block contains no await — _derive_packages, the params.packages assignment and validate_capability_overlap are all synchronous. CancelledError cannot be delivered there, so the new comment ("Cancellation and shutdown must not strand a durable reservation") describes a state the code cannot reach, and no test can be written for it. Narrow it back, or state the reachable trigger (a synchronous SystemExit/KeyboardInterrupt inside derivation) and test that. Lines 755, 756 and 888 never execute under the suite.

4. The admission limit is read off executor._max_workers

src/adcp/decisioning/handler.py:1311-1317:

worker_count = int(getattr(executor, "_max_workers", 1))
admission_limit = ... else max(1, worker_count // 2)

serve.py:120-124 documents BYO executors as "for operators with audit-instrumented thread pools or wrappers around stdlib's executor". A wrapper is exactly the shape that has no _max_workers:

wrapper around a 64-worker pool -> admission limit 1
plain 64-worker pool            -> admission limit 32

Every deadline-managed sync get_products in that deployment then serialises behind one permit held for the full worker duration, with no warning and no public way to observe the value. The parameter is annotated executor: ThreadPoolExecutor (handler.py:1275-1278), so by the declared type the getattr default is dead code; for the shape the docs advertise, it drops the limit to 1 with no warning. int() on an executor whose _max_workers is not int-able raises at handler construction.

Root cause: the composition root owns pool sizing — serve.py already resolved thread_pool_size / _default_thread_pool_size() at serve.py:70,268 — and the handler re-derives it downstream from a CPython-internal field that is not part of any stability contract. Thread the resolved worker count (or the resolved admission limit) from create_adcp_server_from_platform into PlatformHandler and keep timed_sync_get_products_limit as the explicit override. If a BYO executor cannot supply a count, require the override rather than silently choosing 1.

5. Supervised submit has no owned home: written twice, and a third site was left behind

src/adcp/decisioning/dispatch.py:1445-1467 and src/adcp/decisioning/platform_router.py:141-162 run the same sequence in the same order — await admission.acquire() → executor.submit(call) → except BaseException: admission.release(); raise → a _release_admission done-callback wrapping loop.call_soon_threadsafe(admission.release) in try/except RuntimeError → asyncio.wrap_future(..., loop=loop) → await asyncio.shield(worker). They differ only in identifiers and indentation, and already diverge on the fallback branch and on what happens at cancellation. Every step is load-bearing for permit accounting.

The third framework sync-submit site was not migrated. src/adcp/decisioning/proposal_dispatch.py:244 is still the pre-PR shape:

result = await loop.run_in_executor(
    executor, functools.partial(ctx_snapshot.run, method, finalize_req, ctx)
)

No shield, no admission, no settlement. Cancel a sync ProposalManager.finalize_proposal and the asyncio side unwinds immediately while the thread runs to completion and applies the adopter's finalize side effects; the framework's store.commit below never runs:

worker completed: True
proposal state after cancelled finalize: ProposalState.DRAFT

That is the same "worker mutated state, ledger disagrees" failure the PR body claims to close, on the file this PR opened to close it.

Root cause: "submit sync adopter work under supervision" is inline in dispatch and re-implemented in platform_router, so the third call site has nothing to call. SyncExecutorAdmission (time_budget.py:82) exposes only acquire/release and leaves the ordering contract — release exactly once, from the concurrent future's callback, marshalled back to the loop — to be re-derived at each site. Extract one submit_supervised(executor, admission, call) -> asyncio.Future next to the semaphore and route all three callers through it, with an e2e test for the cancelled finalize alongside test_sync_create_media_buy_cancellation_waits_for_worker_success.

6. The settle task reruns the full projection, so a cancelled request can mint a task the buyer never learns about

_settle_cancelled_sync_lifecycle (src/adcp/decisioning/dispatch.py:1715) calls _project_invocation_result at dispatch.py:1737, which reruns the full result projection rather than only the lifecycle hooks. When the cancelled sync adopter returns a TaskHandoff, the settle path runs _project_handoff: it issues a registry task_id, launches the background handoff, persists the terminal artifact, and emits the completion webhook if a push config was supplied. The Submitted envelope it builds is then discarded — there is no waiter. Cancel a sync create_media_buy that returns a handoff, then release the worker:

REGISTRY RECORDS: {'task_dc9d25f8c16e4add': TaskRecord(state='completed',
  task_type='create_media_buy', result={'media_buy_id': 'mb_orphan', ...})}

Absent a push config the buyer holds no task_id, so the task is unreachable via tasks/get and their only recourse is a retry that re-executes the work. This arm could not run before the PR (await run_in_executor raised CancelledError before the handoff branch), so the diff introduces it, and neither of the two new dispatch tests exercises it — they cover a plain dict return and a RuntimeError.

Decide and state the contract: either restrict the settle path to the lifecycle hooks it exists for and refuse to promote a handoff once the caller is gone, or keep the promotion and make the task recoverable — log the issued task_id with the request correlation id at WARNING, and name the push-notification surface (schemas/cache/3.1/core/protocol-envelope.json @ AdCP 3.1.8) as the delivery channel that makes it legal. Either way pin the chosen behavior with a test for the TaskHandoff and WorkflowHandoff arms.

7. The bound executor is ignored without an admission controller, and the worker future travels on the exception

Two defects, both in _run_sync_delegate.

src/adcp/decisioning/platform_router.py:137 reads if admission is None or executor is None: → asyncio.to_thread(...). _bind_routed_sync_execution(sync_admission, executor) always binds a non-None executor (dispatch.py:1425), but the delegate only uses it when an admission controller is also present, i.e. only for deadline-managed get_products. Every other routed sync child — create_media_buy, update_media_buy, refine_get_products, all synthesized delegates — runs on the loop's default executor, outside the adopter's BYO pool, with the right executor sitting in the ContextVar one line above. With a BYO pool named FRAMEWORK, a routed sync child on a no-deadline get_products ran on thread asyncio_0. That undercuts the D5 BYO-executor contract documented in serve.py.

src/adcp/decisioning/platform_router.py:166 signals dispatch by setattr(exc, "_adcp_sync_worker_future", worker), read back at dispatch.py:1604 as Any and narrowed only by isinstance(..., asyncio.Future). platform_router is a DecisioningPlatform implementation below dispatch, and it now encodes dispatch's settlement protocol on an exception instance. Nothing fails at type-check time if the attribute name, producer or consumer drifts, and the channel is lossy: any frame between the router and dispatch that catches CancelledError and re-raises a fresh instance (adopter middleware, a wrapping platform) drops the marker with no log at either end.

Root cause for both: the executor choice is execution plumbing, not time-budget policy, and putting the ThreadPoolExecutor ContextVar in time_budget.py next to the deadline machinery is what made "no deadline" read as "no configured executor" and left the live worker future with nowhere typed to live. Split the condition (executor is not None → submit to it; admission is not None → additionally gate on a permit), carry the worker future on the same scope object the two modules already share (or on a SyncWorkerCancelled(CancelledError) subclass with a typed worker: asyncio.Future[Any]), and log when dispatch unwinds a cancelled routed sync call with no marker. Add a test asserting a routed sync child runs on the handler's executor (thread_name_prefix) with and without a time budget.

8. New surfaces shipped without a test that fails without them

Four groups, each verified by mutation or by coverage of the added lines:

• The public knob. timed_sync_get_products_limit was added to create_adcp_server_from_platform (serve.py:84,381) and serve (serve.py:466,588), and appears in tests only as a direct PlatformHandler(...) kwarg (tests/test_time_budget.py:312,393,447). Deleting both plumbing lines leaves the suite green — the parameter becomes a documented no-op and nothing notices. The ValueError guard at time_budget.py:93 never executes, so timed_sync_get_products_limit=0 producing a permanently saturated server at boot is unasserted, as is the documented max(1, worker_count // 2) default.
• Permit accounting on return and on submit failure. Making SyncExecutorAdmission.release() release twice — an unbounded ceiling after every completed worker — leaves tests/test_time_budget.py and tests/test_decisioning_dispatch.py green (84 passed). The submit-failure release paths at dispatch.py:1449-1452 and platform_router.py:148-150 never execute. Add: an executor stub whose submit raises RuntimeError (post-shutdown behavior) → the next timed call is still admitted; and after N timed calls fully complete, a burst of N+1 blocks exactly one.
• The lazy proposal-manager arm. platform_router.py:1099 is the only one of the eight migrated asyncio.to_thread → _run_sync_delegate sites with no execution. test_router_sync_timeout_uses_bounded_admission[lazy] covers the platform arm, not proposal_manager_for_tenant, so that path silently changed thread-dispatch semantics with no test.
• The settle task's failure arms. dispatch.py:1750,1754 (except BaseException + logger.exception) and dispatch.py:1774 never execute; reverting _safe_on_failure_call to except Exception leaves the suite green.

9. Admission saturation emits the timeout incomplete[] verbatim

The bounded-admission path introduces a genuinely new server-side condition — the seller never searched, because no permit came free before the budget expired — and reuses the pre-existing timeout payload word for word (src/adcp/decisioning/time_budget.py:219-233): "time_budget exhausted (N unit); return the best results achievable within the budget. Retry with a larger time_budget…". Per schemas/cache/3.1/media-buy/get-products-response.json @ AdCP 3.1.8, incomplete[] "Declares what the seller could not finish within the buyer's time_budget or due to internal limits", and the per-entry description is the "Human-readable explanation of what is missing and why". The spec separates exactly these two causes; today a buyer cannot tell "searched and ran out of time" from "declined admission and searched nothing". scope: "products" is correct on both.

Grading is thin on the same path: both new saturation tests assert only getattr(result, "incomplete", None) truthiness, so a regression emitting an off-enum scope or an empty incomplete[] (minItems: 1) would still pass. Give the saturation path its own description naming the internal admission limit, and assert the full projected incomplete[0] (scope, description, products == []).

10. The sanitized caused_by shape was made uniform, then stopped one site short

The diff imports _internal_error_details into handler.py and routes the get_adcp_capabilities INTERNAL_ERROR through it (handler.py:1650-1653), dropping str(exc) from details.caused_by. That is the right call and the new assertions at tests/test_decisioning_capabilities_projection.py:640-642 pin it. Three follow-throughs are missing:

• handler.py:2107 still hand-rolls details={"caused_by": {"type": type(exc).__name__}} for the ProductConfigStore.lookup_implementation_configs SERVICE_UNAVAILABLE wrap — the same wire shape under a different name, so the next change to the sanitized details contract has to be made twice. Route it through _internal_error_details(exc), deciding explicitly whether the helper's details.validation_errors addition is wanted there, or factor the caused_by-only core into a helper both call.
• dispatch.py:1583-1586 still reads "We expose only the exception class name + str (not the traceback)" directly above the _internal_error_details(exc) call, which emits {"type": ...} only. This PR makes that sanitization the single shape, so the comment is now the one place telling a future maintainer to put str(exc) back on the wire. Drop "+ str" and state the invariant positively: class name only, message in the server log via logger.exception. Raised in the 2026-07-29 review, still open.
• The change is buyer-visible on two values — the message string and the loss of details.caused_by.message — and reaches MCP and A2A buyers. The PR's Compatibility section says only "Existing direct asynchronous platform behavior is unchanged". Note the details.caused_by.message removal there and in the changelog, and cite schemas/cache/3.1/core/error.json @ AdCP 3.1.8 (details open with additionalProperties: true, caused_by non-normative, recovery unchanged at terminal) as the grounding.

Notes

• Permit accounting under a cancelled acquire() was the shape most likely to bite — short time_budget while permits are held, each wait_for cancelling a pending acquire, capacity ratcheting to zero. 200 cancelled acquires per trial, 5 trials, on 3.12 and on 3.11's asyncio.Semaphore: effective capacity returned to exactly the limit every time. Not exercised on 3.10, which ships the older Semaphore.acquire — worth one CI run.
• Per-tenant fairness of SyncExecutorAdmission is out of scope: one tenant can hold every permit, but the diff replaces no bound with a bound, and keying the semaphore on ctx.account is a capacity-policy decision rather than a defect this PR introduces.
• _SUPERVISED_SYNC_LIFECYCLES is a module-global task set with no shutdown drain. Out of scope: it follows the module's existing convention (_BACKGROUND_HANDOFF_TASKS at dispatch.py:2091, _BACKGROUND_WEBHOOK_TASKS at webhook_emit.py:107), so changing it is a repo-wide decision.
• tests/test_pg_idempotency_backend.py::test_delete_expired_defaults_to_wall_clock failed once across 16 full-suite runs, while two suites ran concurrently, and passed in the other 15. Pre-existing wall-clock sensitivity, unrelated to this diff — recorded so the single red line in the logs is not read as a finding.

[KonstantinMirin]

This releases the proposal reservation while the adopter's handoff work is still outstanding. For create_media_buy, on_failure is _release_reservation_hook -> release_proposal_reservation -> CONSUMING -> COMMITTED. Background handoff tasks are detached create_tasks (dispatch.py:2069) that ordinary loop/SIGTERM shutdown cancels. Measured, cancelling after the handoff fn issued its upstream create:

HEAD  state after bg cancel: ProposalState.COMMITTED | registry row: submitted
      RETRY ACCEPTED -> {'media_buy_id': 'mb_duplicate_2'}
BASE  state after bg cancel: ProposalState.CONSUMING
      RETRY REJECTED -> AdcpError[PROPOSAL_NOT_COMMITTED / correctable]

One committed proposal, two media buys. The arm also skips _fail(), so the registry row stays submitted with no terminal webhook — the buyer holds a task that never terminates, which is what provokes the retry.

The arm infers "our asyncio task was cancelled" => "the adopter's work did not happen", the inference the rest of this PR refutes on the request path. The handoff fn is not shielded here, and for a sync fn it is a run_in_executor thread that cannot be stopped. Either don't fire on_failure here (leave CONSUMING, which fails closed and is what eviction is for), or mirror _settle_cancelled_sync_lifecycle and settle from the fn's real terminal outcome. A release must be paired with a terminal registry state either way. Lines 1972-1977 never execute under the suite.

Also: "then remain cancellation to the task scheduler" does not parse.

[KonstantinMirin]

This predicate reads one half of the (on_complete, on_failure) lifecycle contract. get_products is the only method with a deadline and the only caller passing sync_admission, and it wires on_complete=_persist_draft_hook with no on_failure (handler.py:1974-1979), so no timed get_products is ever supervised:

on_complete only     -> ON_COMPLETE CALLS: []
add on_failure=noop  -> ON_COMPLETE CALLS: [{'products': []}]

The else-branch also leaves the shielded future with no owner, so a worker raising after the deadline reaches asyncio's default handler. New at HEAD, absent at b7ef1dfc:

ERROR:asyncio:Future exception was never retrieved
future: <Future finished exception=RuntimeError('upstream token sk-SECRET-LEAK rejected')>

That is the adopter's raw exception text at ERROR, outside the framework's error funnel, while this same PR strips str(exc) from details.caused_by for exactly that reason.

Gate on on_complete is not None or on_failure is not None, and attach a consumer to worker_async_future in every branch. Raised in the 2026-07-29 review as a follow-up; still open.

[KonstantinMirin]

Same predicate as dispatch.py:1469, and the same consequence on the routed path: platform_router.py:166 sets _adcp_sync_worker_future on every cancelled routed sync call, but this is its only consumer, so on the get_products path the marker is written and never read.

This block is also byte-identical to 1471-1487 bar the future it settles. Extract one _supervise_sync_lifecycle(future, ...) so the predicate lives in a single place — two verbatim copies is what let both share the wrong condition.

[KonstantinMirin]

This acquire/submit/release/wrap/shield sequence is duplicated verbatim in platform_router.py:141-162 — same steps, same order, differing only in identifiers and indentation, and already diverging on the fallback branch and on cancellation. Every step is load-bearing for permit accounting, so a fix applied to one copy silently misses the other.

SyncExecutorAdmission (time_budget.py:82) exposes only acquire/release and leaves the ordering contract — release exactly once, from the concurrent future's callback, marshalled back to the loop — to be re-derived at each call site. Extract submit_supervised(executor, admission, call) -> asyncio.Future next to the semaphore. The third framework sync-submit site, proposal_dispatch.py:244, is still the pre-PR bare run_in_executor and would be its third caller.

[KonstantinMirin]

This branch is dead by construction. pre_handoff_reject= is passed by exactly two callers (handler.py:1976 get_products, handler.py:2676 get_signals); on_failure= by exactly one (handler.py:2192 create_media_buy). No caller passes both, and coverage confirms line 1669 never executes.

Same root as the supervision gate: the framework's two discovery tools carry a completion hook and no failure hook. Give them one, or hoist supervision off the on_failure predicate — then this branch becomes reachable and testable.

[KonstantinMirin]

_bind_routed_sync_execution(sync_admission, executor) always binds a non-None executor (dispatch.py:1425), but this condition only honours it when an admission controller is also present — i.e. only for deadline-managed get_products. Every other routed sync child (create_media_buy, update_media_buy, refine_get_products, all synthesized delegates) runs on the loop's default executor with the right executor sitting in the ContextVar one line above. With a BYO pool named FRAMEWORK, a routed sync child on a no-deadline get_products ran on thread asyncio_0, which undercuts the D5 BYO-executor contract documented in serve.py.

Two orthogonal decisions are collapsed into one condition. Split them: executor is not None -> submit to it; admission is not None -> additionally gate on a permit. Add a test asserting a routed sync child runs on the handler's executor (thread_name_prefix) with and without a time budget.

[KonstantinMirin]

This encodes dispatch's settlement protocol as a private attribute on an exception instance, read back at dispatch.py:1604 as Any and narrowed only by isinstance(..., asyncio.Future). platform_router is a DecisioningPlatform implementation below dispatch, so the dependency is inverted, and there is no type contract — nothing fails at type-check time if the attribute name, producer or consumer drifts.

The channel is also lossy: any frame between the router and dispatch that catches CancelledError and re-raises a fresh instance (adopter middleware, a wrapping platform) drops the marker with no log at either end, and the worker is orphaned.

The two modules already share a typed, scoped channel — the _bind_routed_sync_execution / _routed_sync_execution pair. Put the live worker future on that scope object, or on a SyncWorkerCancelled(CancelledError) subclass with a typed worker: asyncio.Future[Any]. Either way, log when dispatch unwinds a cancelled routed sync call with no marker.

[KonstantinMirin]

Deleting this line and its twin at serve.py:588 leaves the whole suite green — the new public parameter becomes a documented no-op and nothing notices. timed_sync_get_products_limit appears in tests only as a direct PlatformHandler(...) kwarg (tests/test_time_budget.py:312,393,447); neither adopter-facing entry point is exercised.

Add create_adcp_server_from_platform(platform, timed_sync_get_products_limit=1) and assert the resulting handler's admission limit, plus a handler built on ThreadPoolExecutor(max_workers=4) with no override that admits 2 and blocks the 3rd.

[KonstantinMirin]

This guard never executes under the suite, so timed_sync_get_products_limit=0 producing a permanently saturated server at boot is unasserted. Add SyncExecutorAdmission(0) / (-1) / (True) raising ValueError.

While here: release() is unbounded — making it release twice (an ever-growing ceiling after every completed worker) leaves tests/test_time_budget.py and tests/test_decisioning_dispatch.py green at 84 passed. The assertion that pins release-exactly-once is: after N timed calls have fully completed, a burst of N+1 must still block exactly one.

[KonstantinMirin]

Routing this through _internal_error_message / _internal_error_details is the right call — it drops str(exc) so a platform raising on secret material cannot leak it on the wire, and the new assertions at tests/test_decisioning_capabilities_projection.py:640-642 pin that. Three follow-throughs are missing:

1. handler.py:2107 still hand-rolls details={"caused_by": {"type": type(exc).__name__}} for the ProductConfigStore.lookup_implementation_configs SERVICE_UNAVAILABLE wrap — same wire shape under a different name, so the next change to the sanitized details contract has to be made twice. Decide explicitly whether the helper's details.validation_errors addition is wanted there, or factor the caused_by-only core into a helper both call.
2. dispatch.py:1583-1586 still reads "We expose only the exception class name + str (not the traceback)" directly above the _internal_error_details(exc) call, which emits {"type": ...} only. Drop "+ str" — as written it is the one place telling a future maintainer to put the leak back. Raised in the 2026-07-29 review, still open.
3. This is buyer-visible on two values (the message string, and the loss of details.caused_by.message) and reaches MCP and A2A buyers. The Compatibility section says only "Existing direct asynchronous platform behavior is unchanged". Note the details.caused_by.message removal there and in the changelog, citing schemas/cache/3.1/core/error.json @ AdCP 3.1.8 (details open with additionalProperties: true, caused_by non-normative, recovery unchanged at terminal).