Skip to content

fix(decisioning): supervise timed and canceled work #1000

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bokelley wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(decisioning): supervise timed and canceled work #1000

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
212 changes: 189 additions & 23 deletions src/adcp/decisioning/dispatch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@
TaskHandoffContext,
TaskRegistry,
)
from adcp.decisioning.time_budget import SyncExecutorAdmission, _bind_routed_sync_execution
from adcp.decisioning.types import (
AdcpError,
TaskHandoff,
Expand All @@ -85,6 +86,11 @@

logger = logging.getLogger(__name__)

# Strong references for synchronous adopter lifecycles that outlive a
# cancelled request. A Python thread cannot be cancelled; its completion hooks
# must still settle durable proposal/idempotency state.
_SUPERVISED_SYNC_LIFECYCLES: set[asyncio.Task[Any]] = set()

# ---------------------------------------------------------------------------
# Specialism enum — spec slugs known to the framework
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -1313,6 +1319,7 @@ async def _invoke_platform_method(
webhook_target: WebhookDeliveryTarget | None = None,
webhook_auto_emit: bool = True,
pre_handoff_reject: Callable[[], None] | None = None,
sync_admission: SyncExecutorAdmission | None = None,
) -> Any:
"""Invoke a platform method, projecting hybrid returns.

Expand Down Expand Up @@ -1383,12 +1390,16 @@ async def _invoke_platform_method(
off on a ``wholesale`` request is rejected cleanly instead of
leaking a task the buyer was told was rejected. Runs only on the
``TaskHandoff`` arm; sync / workflow-handoff returns ignore it.
:param sync_admission: Optional bounded admission controller for a sync
method. Its permit remains held until the underlying thread future
actually completes, including after caller cancellation.
"""
# pydantic is a required dep; import here (not at module level) to mirror
# the lazy-import discipline used throughout this module.
from pydantic import ValidationError as _ValidationError # noqa: PLC0415

method = getattr(platform, method_name)
sync_lifecycle_continues = False
# Re-validate through the platform method's own annotation when it's a
# stricter subclass of the shim's already-deserialized type. Skipped
# when arg_projector is set — that path replaces positional args entirely.
Expand All @@ -1407,31 +1418,74 @@ async def _invoke_platform_method(

try:
if asyncio.iscoroutinefunction(method):
if arg_projector is not None:
result = await method(**arg_projector, ctx=ctx)
elif extra_kwargs:
result = await method(params, ctx, **extra_kwargs)
else:
result = await method(params, ctx)
# Async router delegates may resolve to synchronous tenant
# children only after account routing. Propagate the same bounded
# admission controller and configured executor through ContextVars
# so that path cannot bypass the timed-sync limit.
with _bind_routed_sync_execution(sync_admission, executor):
if arg_projector is not None:
result = await method(**arg_projector, ctx=ctx)
elif extra_kwargs:
result = await method(params, ctx, **extra_kwargs)
else:
result = await method(params, ctx)
else:
ctx_snapshot = contextvars.copy_context()
loop = asyncio.get_running_loop()
if arg_projector is not None:
projected_kwargs = {**arg_projector, "ctx": ctx}
result = await loop.run_in_executor(
executor,
functools.partial(ctx_snapshot.run, method, **projected_kwargs),
)
worker_call = functools.partial(ctx_snapshot.run, method, **projected_kwargs)
elif extra_kwargs:
result = await loop.run_in_executor(
executor,
functools.partial(ctx_snapshot.run, method, params, ctx, **extra_kwargs),
worker_call = functools.partial(
ctx_snapshot.run, method, params, ctx, **extra_kwargs
)
else:
result = await loop.run_in_executor(
executor,
functools.partial(ctx_snapshot.run, method, params, ctx),
)
worker_call = functools.partial(ctx_snapshot.run, method, params, ctx)

if sync_admission is not None:

@KonstantinMirin KonstantinMirin Jul 31, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

await sync_admission.acquire()
try:
worker_future = executor.submit(worker_call)
except BaseException:
if sync_admission is not None:
sync_admission.release()
raise

if sync_admission is not None:

def _release_admission(_future: object) -> None:
try:
loop.call_soon_threadsafe(sync_admission.release)
except RuntimeError:
# Event loop already closed during process teardown.
pass

worker_future.add_done_callback(_release_admission)

worker_async_future = asyncio.wrap_future(worker_future, loop=loop)
try:
result = await asyncio.shield(worker_async_future)
except asyncio.CancelledError:
if on_failure is not None:

@KonstantinMirin KonstantinMirin Jul 31, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

sync_lifecycle_continues = True
lifecycle = asyncio.create_task(
_settle_cancelled_sync_lifecycle(
worker_async_future,
ctx=ctx,
method_name=method_name,
registry=registry,
executor=executor,
on_complete=on_complete,
on_failure=on_failure,
pre_handoff_reject=pre_handoff_reject,
request_params=params,
webhook_target=webhook_target,
webhook_auto_emit=webhook_auto_emit,
)
)
_SUPERVISED_SYNC_LIFECYCLES.add(lifecycle)
lifecycle.add_done_callback(_SUPERVISED_SYNC_LIFECYCLES.discard)
raise
except AdcpError as exc:
# Adopter raised structured error — propagate verbatim. The
# outer middleware projects to the wire envelope. Fire
Expand Down Expand Up @@ -1543,15 +1597,77 @@ async def _invoke_platform_method(
if on_failure is not None:
await _safe_on_failure_call(on_failure, wrapped, method_name)
raise wrapped from exc
except BaseException as exc:
# ``asyncio.CancelledError`` (and shutdown BaseExceptions) bypass the
# wire-error wrapping above, but must still release framework state
# reserved before adapter dispatch. Preserve the exact exception.
nested_sync_future = getattr(exc, "_adcp_sync_worker_future", None)
if isinstance(nested_sync_future, asyncio.Future) and on_failure is not None:

@KonstantinMirin KonstantinMirin Jul 31, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

sync_lifecycle_continues = True
lifecycle = asyncio.create_task(
_settle_cancelled_sync_lifecycle(
nested_sync_future,
ctx=ctx,
method_name=method_name,
registry=registry,
executor=executor,
on_complete=on_complete,
on_failure=on_failure,
pre_handoff_reject=pre_handoff_reject,
request_params=params,
webhook_target=webhook_target,
webhook_auto_emit=webhook_auto_emit,
)
)
_SUPERVISED_SYNC_LIFECYCLES.add(lifecycle)
lifecycle.add_done_callback(_SUPERVISED_SYNC_LIFECYCLES.discard)
if on_failure is not None and not sync_lifecycle_continues:
await _safe_on_failure_call(on_failure, exc, method_name)
raise

return await _project_invocation_result(
result,
ctx=ctx,
method_name=method_name,
registry=registry,
executor=executor,
on_complete=on_complete,
on_failure=on_failure,
pre_handoff_reject=pre_handoff_reject,
request_params=params,
webhook_target=webhook_target,
webhook_auto_emit=webhook_auto_emit,
)


async def _project_invocation_result(
result: Any,
*,
ctx: RequestContext[Any],
method_name: str,
registry: TaskRegistry,
executor: ThreadPoolExecutor,
on_complete: Callable[[Any], Awaitable[None]] | None,
on_failure: Callable[[BaseException], Awaitable[None]] | None,
pre_handoff_reject: Callable[[], None] | None,
request_params: BaseModel,
webhook_target: WebhookDeliveryTarget | None,
webhook_auto_emit: bool,
) -> Any:
"""Project a raw adopter result and settle its framework lifecycle hooks."""
if is_task_handoff(result):
# Reject before any side effect (registry row, background task,
# completion webhook) is created. The wholesale discovery guard
# uses this so an adopter handing off on a synchronous-only
# wholesale request never leaks a task the buyer is told is
# rejected.
if pre_handoff_reject is not None:
pre_handoff_reject()
try:
pre_handoff_reject()
except BaseException as exc:
if on_failure is not None:

@KonstantinMirin KonstantinMirin Jul 31, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

await _safe_on_failure_call(on_failure, exc, method_name)
raise
return await _project_handoff(
result,
ctx,
Expand All @@ -1560,7 +1676,7 @@ async def _invoke_platform_method(
executor=executor,
on_complete=on_complete,
on_failure=on_failure,
request_params=params,
request_params=request_params,
webhook_target=webhook_target,
webhook_auto_emit=webhook_auto_emit,
)
Expand All @@ -1571,7 +1687,7 @@ async def _invoke_platform_method(
method_name=method_name,
registry=registry,
executor=executor,
request_params=params,
request_params=request_params,
)

# Sync return path. Fire on_complete with the typed result before
Expand All @@ -1596,6 +1712,51 @@ async def _invoke_platform_method(
return strip_credentials_from_wire_result(method_name, result)


async def _settle_cancelled_sync_lifecycle(
worker_future: asyncio.Future[Any],
*,
ctx: RequestContext[Any],
method_name: str,
registry: TaskRegistry,
executor: ThreadPoolExecutor,
on_complete: Callable[[Any], Awaitable[None]] | None,
on_failure: Callable[[BaseException], Awaitable[None]] | None,
pre_handoff_reject: Callable[[], None] | None,
request_params: BaseModel,
webhook_target: WebhookDeliveryTarget | None,
webhook_auto_emit: bool,
) -> None:
"""Settle a sync worker after its request task has been cancelled."""
try:
result = await worker_future
except BaseException as exc:
if on_failure is not None:
await _safe_on_failure_call(on_failure, exc, method_name)
return
try:
await _project_invocation_result(
result,
ctx=ctx,
method_name=method_name,
registry=registry,
executor=executor,
on_complete=on_complete,
on_failure=on_failure,
pre_handoff_reject=pre_handoff_reject,
request_params=request_params,
webhook_target=webhook_target,
webhook_auto_emit=webhook_auto_emit,
)
except BaseException:
# Lifecycle hooks already apply their own rollback semantics. There is
# no request waiter left to receive this exception, so retain it in
# server logs rather than producing an unhandled-task warning.
logger.exception(
"Cancelled request's synchronous %s lifecycle failed while settling",
method_name,
)


async def _safe_on_failure_call(
on_failure: Callable[[BaseException], Awaitable[None]],
exc: BaseException,
Expand All @@ -1610,10 +1771,9 @@ async def _safe_on_failure_call(
"""
try:
await on_failure(exc)
except Exception:
except BaseException:

@KonstantinMirin KonstantinMirin Jul 31, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Widening to BaseException here swallows a CancelledError delivered while on_failure is awaited. on_failure hooks do durable-store I/O, so they contain await points where 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 cancelled 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 -> []

A graceful-shutdown drain doing task.cancel(); await task gets a completed task back, and the release is left half-done anyway. Log-and-continue is right for Exception; for a BaseException that is not an Exception, log and raise. Line 1774 never executes under the suite.

logger.exception(
"on_failure hook raised while handling %s for %s — original "
"exception still propagates",
"on_failure hook raised while handling %s for %s — original exception still propagates",
type(exc).__name__,
method_name,
)
Expand Down Expand Up @@ -1809,6 +1969,12 @@ async def _run() -> None:
)
await _fail(wrapped)
return
except BaseException as exc:

@KonstantinMirin KonstantinMirin Jul 31, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

# Background task cancellation must release any proposal
# reservation, then remain cancellation to the task scheduler.
if on_failure is not None:
await _safe_on_failure_call(on_failure, exc, method_name)
raise

# Framework completion hook (e.g., proposal_store.commit for
# finalize, mark_proposal_consumed for create_media_buy). Runs
Expand Down
Loading
Loading