diff --git a/CHANGELOG.md b/CHANGELOG.md index 860d33d7..e5c8f6e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ ### Bug Fixes +* **decisioning:** supervise timed/cancelled synchronous work and sanitize INTERNAL_ERROR cause details to exception type only; `details.caused_by.message` is removed (non-normative under AdCP 3.1.8, with `recovery` unchanged) * **security:** harden SDK auth transports ([a2610a5](https://github.com/adcontextprotocol/adcp-client-python/commit/a2610a5b4f8e0d0d1d500fc3908be1d6862b0764)) * **server:** unify divergent host normalizers behind one helper ([#997](https://github.com/adcontextprotocol/adcp-client-python/issues/997)) ([6fb1b72](https://github.com/adcontextprotocol/adcp-client-python/commit/6fb1b72d45ec8adfc2be93c5232464d6ff1c69e3)) * **signing:** block CGNAT and 6to4 relay ranges in SSRF validation ([#974](https://github.com/adcontextprotocol/adcp-client-python/issues/974)) ([0207429](https://github.com/adcontextprotocol/adcp-client-python/commit/020742979cabb5e721ab33c41caddb6b9849074a)) diff --git a/src/adcp/decisioning/dispatch.py b/src/adcp/decisioning/dispatch.py index e7672d3e..d77e468a 100644 --- a/src/adcp/decisioning/dispatch.py +++ b/src/adcp/decisioning/dispatch.py @@ -59,6 +59,12 @@ TaskHandoffContext, TaskRegistry, ) +from adcp.decisioning.time_budget import ( + RoutedSyncExecution, + SyncExecutorAdmission, + _bind_routed_sync_execution, + submit_supervised, +) from adcp.decisioning.types import ( AdcpError, TaskHandoff, @@ -85,6 +91,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 # --------------------------------------------------------------------------- @@ -583,6 +594,11 @@ def _internal_error_message(method_name: str, exc: BaseException) -> str: return f"Platform method {method_name!r} raised {cls_name}; see details for cause" +def _exception_cause_details(exc: BaseException) -> dict[str, Any]: + """Return the shared sanitized exception-type breadcrumb.""" + return {"caused_by": {"type": type(exc).__name__}} + + def _internal_error_details(exc: BaseException) -> dict[str, Any]: """Build the wire-side ``details`` payload for an INTERNAL_ERROR wrap. @@ -624,11 +640,7 @@ def _internal_error_details(exc: BaseException) -> dict[str, Any]: where a structured field list is meaningful, so we don't generalize this to other exception types. """ - details: dict[str, Any] = { - "caused_by": { - "type": type(exc).__name__, - } - } + details = _exception_cause_details(exc) # Try to import lazily so a future refactor that splits the # validation tooling can't ripple through the dispatch layer. try: @@ -647,8 +659,7 @@ def _internal_error_details(exc: BaseException) -> dict[str, Any]: details["validation_errors"] = list(narrow_union_errors(errors_list)) except Exception: # Defensive — never let a narrowing bug 500 the wire. - # The caused_by.message already carries the truncated - # repr; adopters can still triage via server logs. + # The exception type still lets adopters triage via server logs. pass return details @@ -1313,6 +1324,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. @@ -1383,12 +1395,17 @@ 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 + routed_sync_execution: RoutedSyncExecution | None = None # 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. @@ -1407,31 +1424,50 @@ 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) as routed_sync_execution: + 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(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(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(method, params, ctx) + + worker_async_future = await submit_supervised( + executor, + sync_admission, + worker_call, + ) + try: + result = await asyncio.shield(worker_async_future) + except asyncio.CancelledError: + if on_complete is not None or on_failure is not None: + sync_lifecycle_continues = True + _supervise_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, + ) + raise except AdcpError as exc: # Adopter raised structured error — propagate verbatim. The # outer middleware projects to the wire envelope. Fire @@ -1526,8 +1562,8 @@ async def _invoke_platform_method( # The ``details.caused_by`` shape (Emma AudioStack P2) gives # adopters a breadcrumb on the wire — without it, "An internal # error occurred" is a dead end and adopters have to grep - # server logs. We expose only the exception class name + str - # (not the traceback) so a misconfigured platform that throws + # server logs. We expose only the exception class name (not the + # message or traceback) so a misconfigured platform that throws # on secret material doesn't leak the secret value through # the wire response. logger.exception( @@ -1543,7 +1579,64 @@ 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 = ( + routed_sync_execution.worker if routed_sync_execution is not None else None + ) + if isinstance(nested_sync_future, asyncio.Future) and ( + on_complete is not None or on_failure is not None + ): + sync_lifecycle_continues = True + _supervise_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, + ) + 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 @@ -1560,7 +1653,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, ) @@ -1571,7 +1664,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 @@ -1596,6 +1689,99 @@ 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 asyncio.shield(worker_future) + except asyncio.CancelledError: + # Cancelling this supervisor must not cancel or roll back the + # non-cancellable thread it observes. Its reservation remains held. + raise + except Exception as exc: + if on_failure is not None: + await _safe_on_failure_call(on_failure, exc, method_name) + return + if is_task_handoff(result) or is_workflow_handoff(result): + # The cancelled caller never received a task id. Do not promote an + # unreachable handoff; returning a handoff has not executed its work. + if on_failure is not None: + await _safe_on_failure_call(on_failure, asyncio.CancelledError(), method_name) + logger.warning( + "Discarded %s handoff returned after request cancellation; no task id was issued", + 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 Exception: + # 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, + ) + + +def _supervise_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: + """Own a cancelled request's worker until its lifecycle settles.""" + lifecycle = asyncio.create_task( + _settle_cancelled_sync_lifecycle( + worker_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=request_params, + webhook_target=webhook_target, + webhook_auto_emit=webhook_auto_emit, + ) + ) + _SUPERVISED_SYNC_LIFECYCLES.add(lifecycle) + lifecycle.add_done_callback(_SUPERVISED_SYNC_LIFECYCLES.discard) + + async def _safe_on_failure_call( on_failure: Callable[[BaseException], Awaitable[None]], exc: BaseException, @@ -1808,6 +1994,11 @@ async def _run() -> None: ) await _fail(wrapped) return + except BaseException: + # Cancellation does not prove adopter work stopped. Leave any + # reservation fail-closed for expiry/reconciliation rather than + # release it while side effects may still be outstanding. + raise # Framework completion hook (e.g., proposal_store.commit for # finalize, mark_proposal_consumed for create_media_buy). Runs diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index d6428762..f84c7561 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -55,6 +55,9 @@ ) from adcp.decisioning.dispatch import ( _build_request_context, + _exception_cause_details, + _internal_error_details, + _internal_error_message, _invoke_platform_method, ) from adcp.decisioning.implementation_config import ProductConfigStore @@ -79,7 +82,11 @@ has_refine_support, project_refine_response, ) -from adcp.decisioning.time_budget import project_incomplete_response, resolve_time_budget +from adcp.decisioning.time_budget import ( + SyncExecutorAdmission, + project_incomplete_response, + resolve_time_budget, +) from adcp.decisioning.types import ( Account as _DecisioningAccount, ) @@ -1296,6 +1303,7 @@ def __init__( property_list_fetcher: PropertyListFetcher | None = None, media_buy_store: MediaBuyStore | None = None, advertise_all: bool = False, + timed_sync_get_products_limit: int | None = None, ) -> None: super().__init__() self._platform = platform @@ -1321,6 +1329,13 @@ def __init__( self.canonical_format_legacy_resolver = getattr( platform, "canonical_format_legacy_resolver", None ) + # Direct PlatformHandler construction has no public way to inspect a + # BYO executor's capacity. The adopter-facing composition root passes + # an explicit resolved value; direct construction defaults to one. + admission_limit = ( + timed_sync_get_products_limit if timed_sync_get_products_limit is not None else 1 + ) + self._timed_sync_get_products_admission = SyncExecutorAdmission(admission_limit) # Cache whether the platform's create_media_buy accepts 'configs' # so we only pay the inspect.signature cost at construction time. @@ -1653,17 +1668,9 @@ async def get_adcp_capabilities( ) raise AdcpError( "INTERNAL_ERROR", - message=( - "Unhandled exception in platform.get_adcp_capabilities_for_request: " - f"{type(exc).__name__}: {exc}" - ), + message=_internal_error_message("get_adcp_capabilities_for_request", exc), recovery="terminal", - details={ - "caused_by": { - "type": type(exc).__name__, - "message": str(exc), - } - }, + details=_internal_error_details(exc), ) from exc has_scoped_caps = scoped_caps is not None if scoped_caps is not None: @@ -1999,6 +2006,9 @@ async def _persist_draft_hook(get_products_result: Any) -> None: registry=self._registry, on_complete=_persist_draft_hook, pre_handoff_reject=pre_handoff_reject, + sync_admission=( + self._timed_sync_get_products_admission if deadline is not None else None + ), **self._handoff_webhook_kwargs(), ) try: @@ -2007,9 +2017,9 @@ async def _persist_draft_hook(get_products_result: Any) -> None: ) except asyncio.TimeoutError: # Deadline expired. The platform coroutine is cancelled; for - # sync adopters the underlying thread runs to completion but the - # asyncio side has moved on (thread-pool slot leak documented in - # adcp.decisioning.time_budget module header). + # sync adopters an admitted underlying thread runs to completion, + # retaining its bounded admission permit. Saturated calls that + # never acquired a permit were not submitted to the executor. tb = params.time_budget interval = tb.interval if tb is not None else 0 unit_raw = tb.unit if tb is not None else None @@ -2022,7 +2032,9 @@ async def _persist_draft_hook(get_products_result: Any) -> None: "[adcp.decisioning] get_products timed out after %ds " "(time_budget=%d %s); returning incomplete response. " "To avoid timeout cancellations, optimise get_products " - "latency or reduce the platform's search scope.", + "latency or reduce the platform's search scope. Saturated " + "sync calls wait behind timed_sync_get_products_limit and " + "are not submitted after their budget expires.", deadline, interval, unit, @@ -2124,7 +2136,7 @@ async def create_media_buy( # type: ignore[override] "if the problem persists contact the seller." ), recovery="transient", - details={"caused_by": {"type": type(exc).__name__}}, + details=_exception_cause_details(exc), ) from exc # v1.5: when params.proposal_id is set AND a tenant store is diff --git a/src/adcp/decisioning/platform_router.py b/src/adcp/decisioning/platform_router.py index 5619e68b..6d60bdad 100644 --- a/src/adcp/decisioning/platform_router.py +++ b/src/adcp/decisioning/platform_router.py @@ -119,6 +119,7 @@ SalesPlatform, SignalsPlatform, ) +from adcp.decisioning.time_budget import _routed_sync_execution, submit_supervised from adcp.decisioning.types import AdcpError if TYPE_CHECKING: @@ -128,6 +129,26 @@ from adcp.decisioning.proposal_store import ProposalStore +async def _run_sync_delegate(method: Any, *args: Any, **kwargs: Any) -> Any: + """Run a sync child and expose its live future across request cancellation.""" + execution = _routed_sync_execution() + if execution is None: + worker: asyncio.Future[Any] = asyncio.create_task( + asyncio.to_thread(method, *args, **kwargs) + ) + else: + worker = await submit_supervised( + execution.executor, + execution.admission, + lambda: method(*args, **kwargs), + ) + execution.worker = worker + try: + return await asyncio.shield(worker) + except asyncio.CancelledError: + raise + + # Every specialism Protocol the framework knows about. New Protocol # classes added to ``adcp.decisioning.specialisms`` get picked up by # adding them here. Walking ``__protocol_attrs__`` (set by the runtime @@ -503,7 +524,7 @@ async def get_products(self, *args: Any, **kwargs: Any) -> Any: method = getattr(manager, method_name) if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) # No proposal_manager for this tenant — fall through to the # platform. Reuses the same lookup helper as the synthesized @@ -512,7 +533,7 @@ async def get_products(self, *args: Any, **kwargs: Any) -> Any: method = getattr(platform, "get_products") if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) def _make_delegate(self, method_name: str) -> Any: """Create a delegating callable for ``method_name``. @@ -549,7 +570,7 @@ async def _delegate(*args: Any, **kwargs: Any) -> Any: # contextvars snapshot; ``asyncio.to_thread`` does the same # using the running loop's default executor with copied # context. - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) _delegate.__name__ = method_name _delegate.__qualname__ = f"PlatformRouter.{method_name}" @@ -1023,7 +1044,7 @@ async def _delegate(*args: Any, **kwargs: Any) -> Any: method = getattr(platform, method_name) if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) _delegate.__name__ = method_name _delegate.__qualname__ = f"LazyPlatformRouter.{method_name}" @@ -1057,13 +1078,13 @@ async def get_products(self, *args: Any, **kwargs: Any) -> Any: method = getattr(manager, method_name) if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) platform = await self._platform_for_method(ctx, "get_products") method = getattr(platform, "get_products") if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) def proposal_manager_for_tenant(self, tenant_id: str) -> ProposalManager | None: """Return the :class:`ProposalManager` for ``tenant_id``, or ``None``.""" @@ -1245,7 +1266,7 @@ async def _delegate(*args: Any, **kwargs: Any) -> Any: method = getattr(platform, method_name) if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) _delegate.__name__ = method_name _delegate.__qualname__ = f"_RegistryPlatformAdapter.{method_name}" @@ -1272,7 +1293,7 @@ async def get_products(self, *args: Any, **kwargs: Any) -> Any: method = getattr(platform, "get_products") if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) def _make_registry_platform_adapter( diff --git a/src/adcp/decisioning/proposal_dispatch.py b/src/adcp/decisioning/proposal_dispatch.py index e989cc7d..b7e34a5a 100644 --- a/src/adcp/decisioning/proposal_dispatch.py +++ b/src/adcp/decisioning/proposal_dispatch.py @@ -48,7 +48,6 @@ from __future__ import annotations import asyncio -import contextvars import functools import logging from datetime import datetime, timedelta, timezone @@ -73,6 +72,7 @@ _await_maybe, ) from adcp.decisioning.recipe import Recipe +from adcp.decisioning.time_budget import submit_supervised from adcp.decisioning.types import AdcpError, is_task_handoff if TYPE_CHECKING: @@ -84,6 +84,57 @@ from adcp.decisioning.task_registry import TaskRegistry logger = logging.getLogger("adcp.decisioning.proposal_dispatch") +_SUPERVISED_FINALIZATIONS: set[asyncio.Task[None]] = set() + + +async def _settle_cancelled_finalize( + worker: asyncio.Future[Any], + *, + store: Any, + proposal_id: str, + account_id: str, +) -> None: + """Commit a sync finalize result after its request task is cancelled.""" + try: + result = await asyncio.shield(worker) + except asyncio.CancelledError: + # Process shutdown may cancel this observer; never cancel the thread. + raise + except Exception: + logger.exception( + "Cancelled finalize_proposal worker failed for proposal %s; draft retained", + proposal_id, + ) + return + + if not isinstance(result, FinalizeProposalSuccess): + logger.error( + "Cancelled finalize_proposal returned %s for proposal %s; draft retained", + type(result).__name__, + proposal_id, + ) + return + try: + await _await_maybe( + store.commit( + proposal_id, + expires_at=result.expires_at, + proposal_payload=dict(result.proposal), + expected_account_id=account_id, + ) + ) + except Exception: + logger.exception( + "Cancelled finalize_proposal succeeded but proposal %s commit failed", + proposal_id, + ) + return + finalize_succeeded_log( + proposal_id=proposal_id, + account_id=account_id, + expires_at=result.expires_at, + path="inline-after-cancellation", + ) # --------------------------------------------------------------------------- @@ -239,12 +290,25 @@ async def maybe_intercept_finalize( if asyncio.iscoroutinefunction(method): result = await method(finalize_req, ctx) else: - ctx_snapshot = contextvars.copy_context() - loop = asyncio.get_running_loop() - result = await loop.run_in_executor( + worker = await submit_supervised( executor, - functools.partial(ctx_snapshot.run, method, finalize_req, ctx), + None, + functools.partial(method, finalize_req, ctx), ) + try: + result = await asyncio.shield(worker) + except asyncio.CancelledError: + supervisor = asyncio.create_task( + _settle_cancelled_finalize( + worker, + store=store, + proposal_id=proposal_id, + account_id=account_id, + ) + ) + _SUPERVISED_FINALIZATIONS.add(supervisor) + supervisor.add_done_callback(_SUPERVISED_FINALIZATIONS.discard) + raise if is_task_handoff(result): # HITL slow path. Per § D2 + § D3: framework projects Submitted @@ -745,12 +809,8 @@ async def maybe_hydrate_recipes_for_create_media_buy( field_path_prefix="packages", ) except Exception: - # Narrow to Exception (not BaseException): CancelledError / - # SystemExit / KeyboardInterrupt skip the release path — under - # cancellation the next worker will read the stale reservation - # and eviction handles it; under shutdown we want fast exit. - # release_consumption is idempotent on already-COMMITTED so a - # release that races with a concurrent worker is harmless. + # Ordinary derivation/validation failures release the reservation. + # Process-control BaseExceptions propagate without being intercepted. try: await _await_maybe( store.release_consumption(proposal_id, expected_account_id=ctx.account.id) diff --git a/src/adcp/decisioning/serve.py b/src/adcp/decisioning/serve.py index 05e4ac2c..212c1723 100644 --- a/src/adcp/decisioning/serve.py +++ b/src/adcp/decisioning/serve.py @@ -81,6 +81,7 @@ def create_adcp_server_from_platform( *, executor: ThreadPoolExecutor | None = None, thread_pool_size: int | None = None, + timed_sync_get_products_limit: int | None = None, registry: TaskRegistry | None = None, state_reader: StateReader | None = None, resource_resolver: ResourceResolver | None = None, @@ -121,10 +122,19 @@ def create_adcp_server_from_platform( for operators with audit-instrumented thread pools or wrappers around stdlib's executor. Mutually exclusive with ``thread_pool_size``. Operator owns lifecycle (caller's - ``shutdown(wait=True)`` responsibility). + ``shutdown(wait=True)`` responsibility). Requires an explicit + ``timed_sync_get_products_limit`` because executor wrappers expose no + public capacity contract. :param thread_pool_size: Size the default framework-allocated executor. Mutually exclusive with ``executor``. Default is :func:`_default_thread_pool_size`. + :param timed_sync_get_products_limit: Maximum synchronous + ``get_products`` calls with SDK-managed deadlines admitted to the + executor at once. Saturated calls wait within their own time budget + and return ``incomplete`` without being submitted if it expires. + For framework-allocated pools, defaults to half the configured worker + count (minimum one), reserving capacity for other tools. Required with + ``executor=``. :param registry: Bring-your-own :class:`TaskRegistry` — typically a v6.1 durable backing store. Default is :class:`InMemoryTaskRegistry`, which the production-mode @@ -256,13 +266,28 @@ def create_adcp_server_from_platform( "vetted threadpool." ) - # Allocate executor. + # Allocate executor and resolve admission sizing while the public worker + # count is still available. Executor wrappers expose no stable capacity + # attribute, so BYO pools must provide the explicit admission limit. if executor is None: size = thread_pool_size if thread_pool_size is not None else _default_thread_pool_size() executor = ThreadPoolExecutor( max_workers=size, thread_name_prefix="adcp-decisioning-", ) + resolved_timed_sync_limit = ( + timed_sync_get_products_limit + if timed_sync_get_products_limit is not None + else max(1, size // 2) + ) + else: + if timed_sync_get_products_limit is None: + raise ValueError( + "executor= requires timed_sync_get_products_limit= because executor " + "wrappers expose no public worker-count contract. Pass an explicit " + "positive admission limit or use thread_pool_size=." + ) + resolved_timed_sync_limit = timed_sync_get_products_limit # Allocate registry, with production-mode gate (Emma #8). # Gate reads the registry's is_durable class-level marker rather @@ -371,6 +396,7 @@ def create_adcp_server_from_platform( property_list_fetcher=property_list_fetcher, media_buy_store=media_buy_store, advertise_all=advertise_all, + timed_sync_get_products_limit=resolved_timed_sync_limit, ) # Boot-time fail-fast: property_list_filtering declared but no fetcher wired. @@ -455,6 +481,7 @@ def serve( name: str | None = None, executor: ThreadPoolExecutor | None = None, thread_pool_size: int | None = None, + timed_sync_get_products_limit: int | None = None, registry: TaskRegistry | None = None, state_reader: StateReader | None = None, resource_resolver: ResourceResolver | None = None, @@ -484,8 +511,12 @@ def serve( :param name: Server name advertised on AdCP capabilities. Defaults to the platform class's ``__name__``. :param executor: BYO :class:`ThreadPoolExecutor` per - :func:`create_adcp_server_from_platform` D5 contract. + :func:`create_adcp_server_from_platform` D5 contract. Requires + ``timed_sync_get_products_limit``. :param thread_pool_size: Default-executor size override. + :param timed_sync_get_products_limit: Bounded admission limit for + deadline-managed synchronous ``get_products`` calls. See + :func:`create_adcp_server_from_platform`. :param registry: BYO :class:`TaskRegistry`. Default is :class:`InMemoryTaskRegistry` (gated for production). :param state_reader: Custom :class:`StateReader` impl (D15). @@ -573,6 +604,7 @@ def serve( platform, executor=executor, thread_pool_size=thread_pool_size, + timed_sync_get_products_limit=timed_sync_get_products_limit, registry=registry, state_reader=state_reader, resource_resolver=resource_resolver, diff --git a/src/adcp/decisioning/time_budget.py b/src/adcp/decisioning/time_budget.py index 65fd40d4..9fe4a1d0 100644 --- a/src/adcp/decisioning/time_budget.py +++ b/src/adcp/decisioning/time_budget.py @@ -17,15 +17,17 @@ past the ``except Exception`` in ``_invoke_platform_method`` cleanly. This invariant MUST be preserved if ``get_products`` ever gains registry work. -* **Thread-pool warning for sync adopters.** When a sync adopter runs via +* **Bounded sync-adopter admission.** When a sync adopter runs via ``loop.run_in_executor`` and ``asyncio.wait_for`` fires, the asyncio side moves on but the underlying thread continues until its blocking call - returns. No Python mechanism can interrupt a running thread. The pool - slot is occupied for the full duration; on a short-budget burst against a - slow sync adopter this can exhaust the pool. Async adopters are - unaffected. Adopters who need to co-operate with deadline cancellation - should implement the ``IncrementalGetProducts`` protocol or migrate to an - async ``get_products``. + returns. No Python mechanism can interrupt a running thread. The framework + therefore admits only a bounded number of deadline-managed synchronous + calls and holds each permit until the worker really exits, even after the + response timed out. Saturated calls spend their budget waiting for a permit + and return ``incomplete[]`` without entering the executor. By default the + limit is half the executor workers (minimum one), preserving capacity for + other tools; operators can tune it at server construction. Async adopters + are unaffected. * **``campaign`` unit → no SDK-managed deadline.** ``unit='campaign'`` means "the seller has the full campaign flight to respond" — this is a @@ -60,16 +62,117 @@ from __future__ import annotations +import asyncio +import contextvars import logging -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Callable +from concurrent.futures import Executor +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable if TYPE_CHECKING: + from collections.abc import Iterator + from adcp.decisioning.context import RequestContext from adcp.types import GetProductsRequest logger = logging.getLogger(__name__) + +class SyncExecutorAdmission: + """Bound outstanding deadline-managed synchronous executor work. + + A permit represents a worker submission, not a waiting HTTP request. It + is released only by the underlying ``concurrent.futures.Future`` done + callback, because cancelling its asyncio wrapper cannot stop a running + Python thread. + """ + + def __init__(self, limit: int) -> None: + if not isinstance(limit, int) or isinstance(limit, bool) or limit < 1: + raise ValueError("sync executor admission limit must be a positive integer") + self.limit = limit + self._semaphore = asyncio.BoundedSemaphore(limit) + + async def acquire(self) -> None: + """Wait until a bounded worker slot is available.""" + await self._semaphore.acquire() + + def release(self) -> None: + """Return a worker slot after its real thread future completes.""" + self._semaphore.release() + + +async def submit_supervised( + executor: Executor, + admission: SyncExecutorAdmission | None, + call: Callable[[], Any], +) -> asyncio.Future[Any]: + """Submit sync work and bind admission to the real worker lifetime. + + The returned asyncio future may be shielded or observed by another task; + its cancellation cannot stop the underlying thread. A permit is released + exactly once by the concurrent future's completion callback. + """ + if admission is not None: + await admission.acquire() + loop = asyncio.get_running_loop() + try: + snapshot = contextvars.copy_context() + concurrent_worker = executor.submit(snapshot.run, call) + except Exception: + if admission is not None: + admission.release() + raise + + if admission is not None: + + def _release_admission(_future: object) -> None: + try: + loop.call_soon_threadsafe(admission.release) + except RuntimeError: + # Event loop already closed during process teardown. + pass + + concurrent_worker.add_done_callback(_release_admission) + return asyncio.wrap_future(concurrent_worker, loop=loop) + + +@dataclass +class RoutedSyncExecution: + """Typed request scope shared by dispatch and a routed sync delegate.""" + + admission: SyncExecutorAdmission | None + executor: Executor + worker: asyncio.Future[Any] | None = None + + +_ROUTED_SYNC_EXECUTION: ContextVar[RoutedSyncExecution | None] = ContextVar( + "adcp_routed_sync_execution", default=None +) + + +@contextmanager +def _bind_routed_sync_execution( + admission: SyncExecutorAdmission | None, + executor: Executor, +) -> Iterator[RoutedSyncExecution]: + """Expose deadline admission to an async router's eventual sync child.""" + execution = RoutedSyncExecution(admission=admission, executor=executor) + token = _ROUTED_SYNC_EXECUTION.set(execution) + try: + yield execution + finally: + _ROUTED_SYNC_EXECUTION.reset(token) + + +def _routed_sync_execution() -> RoutedSyncExecution | None: + """Return the admission/executor inherited by a router delegate.""" + return _ROUTED_SYNC_EXECUTION.get() + + # ---- Unit conversion ---- _UNIT_TO_SECONDS: dict[str, float] = { @@ -258,6 +361,7 @@ async def get_products_incremental( __all__ = [ "IncrementalGetProducts", "ProductsCheckpoint", + "SyncExecutorAdmission", "project_incomplete_response", "resolve_time_budget", ] diff --git a/tests/test_capabilities_response_shape_validation.py b/tests/test_capabilities_response_shape_validation.py index a971ad7b..cefb5bba 100644 --- a/tests/test_capabilities_response_shape_validation.py +++ b/tests/test_capabilities_response_shape_validation.py @@ -396,6 +396,7 @@ async def test_create_adcp_server_validate_at_init_false_works_in_async_context( handler, _executor, _registry = create_adcp_server_from_platform( _ConformantSalesPlatform(), executor=pool, + timed_sync_get_products_limit=1, registry=InMemoryTaskRegistry(), auto_emit_completion_webhooks=False, validate_at_init=False, @@ -427,6 +428,7 @@ async def test_create_adcp_server_default_init_blows_up_in_async_context() -> No create_adcp_server_from_platform( _ConformantSalesPlatform(), executor=pool, + timed_sync_get_products_limit=1, registry=InMemoryTaskRegistry(), auto_emit_completion_webhooks=False, # default validate_at_init=True — the boom case. @@ -441,6 +443,7 @@ def test_create_adcp_server_validate_at_init_true_still_validates_conformant() - handler, _executor, _registry = create_adcp_server_from_platform( _ConformantSalesPlatform(), executor=pool, + timed_sync_get_products_limit=1, registry=InMemoryTaskRegistry(), auto_emit_completion_webhooks=False, # validate_at_init=True is the default @@ -458,6 +461,7 @@ def test_create_adcp_server_validate_at_init_true_rejects_bad_platform() -> None create_adcp_server_from_platform( _MediaBuyMissingBillingPlatform(), executor=pool, + timed_sync_get_products_limit=1, registry=InMemoryTaskRegistry(), auto_emit_completion_webhooks=False, ) diff --git a/tests/test_decisioning_capabilities_projection.py b/tests/test_decisioning_capabilities_projection.py index 391dc5e6..f2fecd16 100644 --- a/tests/test_decisioning_capabilities_projection.py +++ b/tests/test_decisioning_capabilities_projection.py @@ -657,6 +657,9 @@ def get_adcp_capabilities_for_request(self, params=None, context=None): assert exc_info.value.code == "INTERNAL_ERROR" assert exc_info.value.details["caused_by"]["type"] == "RuntimeError" + assert exc_info.value.details["caused_by"] == {"type": "RuntimeError"} + assert "tenant lookup failed" not in str(exc_info.value) + assert "tenant lookup failed" not in str(exc_info.value.details) def test_request_scoped_capabilities_hook_may_be_async( diff --git a/tests/test_decisioning_dispatch.py b/tests/test_decisioning_dispatch.py index 55841987..d71bc4d3 100644 --- a/tests/test_decisioning_dispatch.py +++ b/tests/test_decisioning_dispatch.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio +import threading import warnings from concurrent.futures import ThreadPoolExecutor from contextvars import ContextVar @@ -32,6 +33,8 @@ _coerce_params_to_platform_type, _invoke_platform_method, _project_handoff, + _safe_on_failure_call, + _settle_cancelled_sync_lifecycle, compose_caller_identity, validate_platform, ) @@ -1642,6 +1645,183 @@ async def get_products(self, req: _StrictSubRequest, ctx): assert on_failure_calls[0] is exc_info.value +@pytest.mark.asyncio +async def test_cancellation_fires_on_failure_and_propagates_unchanged( + executor: ThreadPoolExecutor, +) -> None: + entered = asyncio.Event() + on_failure_calls: list[BaseException] = [] + + class _WaitingPlatform(DecisioningPlatform): + capabilities = DecisioningCapabilities() + accounts = SingletonAccounts(account_id="x") + + async def get_products(self, req: _BaseRequest, ctx): + entered.set() + await asyncio.Event().wait() + + async def _on_failure(exc: BaseException) -> None: + on_failure_calls.append(exc) + + ctx = _build_request_context(ToolContext(), Account(id="x"), None) + task = asyncio.create_task( + _invoke_platform_method( + _WaitingPlatform(), + "get_products", + _BaseRequest(known_field="wait"), + ctx, + executor=executor, + registry=InMemoryTaskRegistry(), + on_failure=_on_failure, + ) + ) + await entered.wait() + task.cancel("client disconnected") + + # Python 3.10 does not preserve Task.cancel(msg) text through shield(). + with pytest.raises(asyncio.CancelledError): + await task + assert len(on_failure_calls) == 1 + assert isinstance(on_failure_calls[0], asyncio.CancelledError) + + +@pytest.mark.asyncio +async def test_sync_cancellation_settles_success_before_on_complete( + executor: ThreadPoolExecutor, +) -> None: + entered = threading.Event() + release = threading.Event() + settled = asyncio.Event() + completed: list[Any] = [] + + class _SyncPlatform(DecisioningPlatform): + capabilities = DecisioningCapabilities() + accounts = SingletonAccounts(account_id="x") + + def get_products(self, req: _BaseRequest, ctx): + entered.set() + release.wait(timeout=2) + return {"products": []} + + async def _on_complete(result: Any) -> None: + completed.append(result) + settled.set() + + ctx = _build_request_context(ToolContext(), Account(id="x"), None) + task = asyncio.create_task( + _invoke_platform_method( + _SyncPlatform(), + "get_products", + _BaseRequest(known_field="wait"), + ctx, + executor=executor, + registry=InMemoryTaskRegistry(), + on_complete=_on_complete, + ) + ) + assert await asyncio.to_thread(entered.wait, 1) + task.cancel("client disconnected") + with pytest.raises(asyncio.CancelledError): + _ = await task + assert completed == [] + + release.set() + await asyncio.wait_for(settled.wait(), 1) + assert completed == [{"products": []}] + + +@pytest.mark.asyncio +async def test_sync_cancellation_settles_real_failure_before_on_failure( + executor: ThreadPoolExecutor, +) -> None: + entered = threading.Event() + release = threading.Event() + settled = asyncio.Event() + failures: list[BaseException] = [] + + class _FailingSyncPlatform(DecisioningPlatform): + capabilities = DecisioningCapabilities() + accounts = SingletonAccounts(account_id="x") + + def get_products(self, req: _BaseRequest, ctx): + entered.set() + release.wait(timeout=2) + raise RuntimeError("worker failed") + + async def _on_failure(exc: BaseException) -> None: + failures.append(exc) + settled.set() + + ctx = _build_request_context(ToolContext(), Account(id="x"), None) + task = asyncio.create_task( + _invoke_platform_method( + _FailingSyncPlatform(), + "get_products", + _BaseRequest(known_field="wait"), + ctx, + executor=executor, + registry=InMemoryTaskRegistry(), + on_failure=_on_failure, + ) + ) + assert await asyncio.to_thread(entered.wait, 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert failures == [] + + release.set() + await asyncio.wait_for(settled.wait(), 1) + assert len(failures) == 1 + assert isinstance(failures[0], RuntimeError) + assert str(failures[0]) == "worker failed" + + +@pytest.mark.asyncio +async def test_cancelling_sync_supervisor_does_not_cancel_worker_or_release( + executor: ThreadPoolExecutor, +) -> None: + worker: asyncio.Future[Any] = asyncio.get_running_loop().create_future() + failures: list[BaseException] = [] + + async def _on_failure(exc: BaseException) -> None: + failures.append(exc) + + ctx = _build_request_context(ToolContext(), Account(id="x"), None) + supervisor = asyncio.create_task( + _settle_cancelled_sync_lifecycle( + worker, + ctx=ctx, + method_name="create_media_buy", + registry=InMemoryTaskRegistry(), + executor=executor, + on_complete=None, + on_failure=_on_failure, + pre_handoff_reject=None, + request_params=_BaseRequest(known_field="wait"), + webhook_target=None, + webhook_auto_emit=False, + ) + ) + await asyncio.sleep(0) + supervisor.cancel() + with pytest.raises(asyncio.CancelledError): + _ = await supervisor + + assert worker.cancelled() is False + assert failures == [] + worker.cancel() + + +@pytest.mark.asyncio +async def test_on_failure_hook_cancellation_propagates() -> None: + async def _cancelled_hook(_exc: BaseException) -> None: + raise asyncio.CancelledError + + with pytest.raises(asyncio.CancelledError): + await _safe_on_failure_call(_cancelled_hook, RuntimeError("original"), "get_products") + + def test_coerce_varargs_annotation_is_noop() -> None: """Annotated *args should not trigger coercion — VAR_POSITIONAL guard fires.""" diff --git a/tests/test_decisioning_serve.py b/tests/test_decisioning_serve.py index 8897795e..00cfebfc 100644 --- a/tests/test_decisioning_serve.py +++ b/tests/test_decisioning_serve.py @@ -17,9 +17,10 @@ from __future__ import annotations +import importlib import os from concurrent.futures import ThreadPoolExecutor -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest @@ -35,6 +36,9 @@ _is_production_env, create_adcp_server_from_platform, ) +from adcp.decisioning.serve import ( + serve as serve_platform, +) class _BarePlatform(DecisioningPlatform): @@ -170,12 +174,71 @@ def test_create_uses_byo_executor_unchanged() -> None: platform = _BarePlatform() custom = ThreadPoolExecutor(max_workers=2, thread_name_prefix="byo-") try: - _, executor, _ = create_adcp_server_from_platform(platform, executor=custom) + _, executor, _ = create_adcp_server_from_platform( + platform, + executor=custom, + timed_sync_get_products_limit=1, + ) assert executor is custom finally: custom.shutdown(wait=True) +def test_create_requires_admission_limit_for_byo_executor() -> None: + custom = ThreadPoolExecutor(max_workers=64) + try: + with pytest.raises(ValueError, match="timed_sync_get_products_limit"): + create_adcp_server_from_platform(_BarePlatform(), executor=custom) + finally: + custom.shutdown(wait=True) + + +def test_create_projects_resolved_admission_limit_to_handler() -> None: + handler, executor, _ = create_adcp_server_from_platform( + _BarePlatform(), + thread_pool_size=4, + ) + try: + assert handler._timed_sync_get_products_admission.limit == 2 # noqa: SLF001 + finally: + executor.shutdown(wait=True) + + +def test_create_projects_explicit_admission_limit_to_handler() -> None: + handler, executor, _ = create_adcp_server_from_platform( + _BarePlatform(), + timed_sync_get_products_limit=1, + ) + try: + assert handler._timed_sync_get_products_admission.limit == 1 # noqa: SLF001 + finally: + executor.shutdown(wait=True) + + +def test_serve_forwards_timed_sync_admission_limit() -> None: + handler = MagicMock() + executor = MagicMock() + registry = MagicMock() + decisioning_serve_module = importlib.import_module("adcp.decisioning.serve") + server_serve_module = importlib.import_module("adcp.server.serve") + with ( + patch.object( + decisioning_serve_module, + "create_adcp_server_from_platform", + return_value=(handler, executor, registry), + ) as create, + patch.object(server_serve_module, "serve") as server_serve, + ): + serve_platform( + _BarePlatform(), + timed_sync_get_products_limit=3, + validate_at_init=False, + ) + + assert create.call_args.kwargs["timed_sync_get_products_limit"] == 3 + server_serve.assert_called_once() + + def test_create_thread_pool_size_overrides_default() -> None: """``thread_pool_size=`` sizes the framework-allocated default executor.""" diff --git a/tests/test_proposal_lifecycle_e2e.py b/tests/test_proposal_lifecycle_e2e.py index e41a1677..37ea847d 100644 --- a/tests/test_proposal_lifecycle_e2e.py +++ b/tests/test_proposal_lifecycle_e2e.py @@ -31,6 +31,7 @@ import asyncio import sys +import threading from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from pathlib import Path @@ -291,6 +292,69 @@ async def test_finalize_commits_proposal( assert record.expires_at is not None +@pytest.mark.asyncio +async def test_cancelled_sync_finalize_commits_after_worker_finishes( + router: Any, + store: InMemoryProposalStore, + executor: ThreadPoolExecutor, +) -> None: + """Cancellation cannot leave a completed sync finalize as a draft.""" + from adcp.types import GetProductsRequest + + handler = PlatformHandler( + router, + executor=executor, + registry=InMemoryTaskRegistry(), + ) + await handler.get_products( + GetProductsRequest(buying_mode="brief", brief="initial"), + ToolContext(), + ) + manager = router.proposal_manager_for_tenant("default") + original = manager.finalize_proposal + entered = threading.Event() + release = threading.Event() + + def _blocking_finalize(req: Any, ctx: Any) -> Any: + entered.set() + release.wait(timeout=2) + return asyncio.run(original(req, ctx)) + + manager.finalize_proposal = _blocking_finalize # type: ignore[method-assign] + finalize_req = GetProductsRequest.model_validate( + { + "buying_mode": "refine", + "refine": [ + { + "scope": "proposal", + "proposal_id": PROPOSAL_ID, + "action": "finalize", + } + ], + } + ) + try: + task = asyncio.create_task(handler.get_products(finalize_req, ToolContext())) + assert await asyncio.to_thread(entered.wait, 1) + task.cancel("buyer disconnected") + with pytest.raises(asyncio.CancelledError): + _ = await task + + draft = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + assert draft is not None and draft.state == ProposalState.DRAFT + release.set() + for _ in range(100): + committed = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + if committed is not None and committed.state == ProposalState.COMMITTED: + break + await asyncio.sleep(0.01) + else: + pytest.fail("cancelled sync finalize did not commit after worker completion") + finally: + release.set() + manager.finalize_proposal = original # type: ignore[method-assign] + + @pytest.mark.asyncio async def test_finalize_unknown_proposal_is_correctable( handler: PlatformHandler, @@ -1398,6 +1462,96 @@ async def _seed_committed_proposal(handler: PlatformHandler) -> None: ) +@pytest.mark.asyncio +async def test_create_media_buy_cancellation_releases_reservation( + executor: ThreadPoolExecutor, + registry: InMemoryTaskRegistry, +) -> None: + from examples.sales_proposal_mode_seller.src.app import build_router + + router = build_router() + store = router.proposal_store_for_tenant("default") + handler = _build_handler(router, executor, registry) + await _seed_committed_proposal(handler) + + entered = asyncio.Event() + target_platform = router._platforms["default"] # noqa: SLF001 + original = target_platform.create_media_buy + + async def _waiting_create(req: Any, ctx: Any) -> Any: + del req, ctx + entered.set() + await asyncio.Event().wait() + + target_platform.create_media_buy = _waiting_create # type: ignore[method-assign] + try: + task = asyncio.create_task( + handler.create_media_buy(_build_create_media_buy_request("cancel"), ToolContext()) + ) + await entered.wait() + + reserved = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + assert reserved is not None and reserved.state == ProposalState.CONSUMING + + task.cancel("buyer disconnected") + with pytest.raises(asyncio.CancelledError): + await task + + released = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + assert released is not None and released.state == ProposalState.COMMITTED + finally: + target_platform.create_media_buy = original # type: ignore[method-assign] + + +@pytest.mark.asyncio +async def test_sync_create_media_buy_cancellation_waits_for_worker_success( + executor: ThreadPoolExecutor, + registry: InMemoryTaskRegistry, +) -> None: + """A cancelled request cannot release a reservation while its thread runs.""" + router = build_router() + store = router.proposal_store_for_tenant("default") + handler = _build_handler(router, executor, registry) + await _seed_committed_proposal(handler) + + entered = threading.Event() + release = threading.Event() + target_platform = router._platforms["default"] # noqa: SLF001 + original = target_platform.create_media_buy + + def _blocking_create(req: Any, ctx: Any) -> Any: + del req, ctx + entered.set() + release.wait(timeout=2) + return {"media_buy_id": "mb_sync_cancel", "status": "active"} + + target_platform.create_media_buy = _blocking_create # type: ignore[method-assign] + try: + task = asyncio.create_task( + handler.create_media_buy(_build_create_media_buy_request("sync-cancel"), ToolContext()) + ) + assert await asyncio.to_thread(entered.wait, 1) + task.cancel("buyer disconnected") + with pytest.raises(asyncio.CancelledError): + await task + + reserved = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + assert reserved is not None and reserved.state == ProposalState.CONSUMING + + release.set() + for _ in range(100): + consumed = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + if consumed is not None and consumed.state == ProposalState.CONSUMED: + break + await asyncio.sleep(0.01) + else: + pytest.fail("sync worker success did not finalize proposal reservation") + assert consumed.media_buy_id == "mb_sync_cancel" + finally: + release.set() + target_platform.create_media_buy = original # type: ignore[method-assign] + + @pytest.mark.asyncio async def test_create_media_buy_handoff_finalizes_consumption_on_completion( executor: ThreadPoolExecutor, diff --git a/tests/test_time_budget.py b/tests/test_time_budget.py index d8b08ebd..7c695eb3 100644 --- a/tests/test_time_budget.py +++ b/tests/test_time_budget.py @@ -14,8 +14,8 @@ from __future__ import annotations import asyncio +import threading from concurrent.futures import ThreadPoolExecutor -from unittest.mock import AsyncMock import pytest @@ -24,18 +24,20 @@ DecisioningPlatform, IncrementalGetProducts, InMemoryTaskRegistry, + LazyPlatformRouter, + PlatformRouter, ProductsCheckpoint, SingletonAccounts, ) from adcp.decisioning.handler import PlatformHandler from adcp.decisioning.time_budget import ( + SyncExecutorAdmission, project_incomplete_response, resolve_time_budget, ) from adcp.server.base import ToolContext from adcp.types import GetProductsRequest - # --------------------------------------------------------------------------- # resolve_time_budget # --------------------------------------------------------------------------- @@ -113,6 +115,18 @@ def test_project_incomplete_response_contains_budget_info(): assert "minutes" in description +@pytest.mark.parametrize("limit", [0, -1, True]) +def test_sync_executor_admission_rejects_invalid_limits(limit) -> None: + with pytest.raises(ValueError, match="positive integer"): + SyncExecutorAdmission(limit) + + +def test_sync_executor_admission_release_is_bounded() -> None: + admission = SyncExecutorAdmission(1) + with pytest.raises(ValueError, match="released too many times"): + admission.release() + + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -204,12 +218,20 @@ async def get_products(self, req, ctx): ) result = await handler.get_products(req, context=ToolContext()) - products = result.get("products") if isinstance(result, dict) else list(getattr(result, "products", [])) # type: ignore[union-attr] + products = ( + result.get("products") + if isinstance(result, dict) + else list(getattr(result, "products", [])) + ) # type: ignore[union-attr] assert len(products) == 1 pid = products[0].get("product_id") if isinstance(products[0], dict) else products[0].product_id # type: ignore[union-attr] assert pid == "p1" # No incomplete key / field when fully resolved - incomplete = result.get("incomplete") if isinstance(result, dict) else getattr(result, "incomplete", None) + incomplete = ( + result.get("incomplete") + if isinstance(result, dict) + else getattr(result, "incomplete", None) + ) assert not incomplete @@ -232,7 +254,11 @@ async def get_products(self, req, ctx): ) req = GetProductsRequest.model_construct(account=None, time_budget=None) result = await handler.get_products(req, context=ToolContext()) - products = result.get("products") if isinstance(result, dict) else list(getattr(result, "products", [])) # type: ignore[union-attr] + products = ( + result.get("products") + if isinstance(result, dict) + else list(getattr(result, "products", [])) + ) # type: ignore[union-attr] assert len(products) == 1 @@ -258,10 +284,237 @@ async def get_products(self, req, ctx): time_budget=_make_time_budget(interval=1, unit="campaign"), ) result = await handler.get_products(req, context=ToolContext()) - products = result.get("products") if isinstance(result, dict) else list(getattr(result, "products", [])) # type: ignore[union-attr] + products = ( + result.get("products") + if isinstance(result, dict) + else list(getattr(result, "products", [])) + ) # type: ignore[union-attr] assert len(products) == 1 +@pytest.mark.asyncio +async def test_sync_timeout_admission_saturates_without_executor_queue_growth( + executor: ThreadPoolExecutor, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Timed-out threads retain permits; later short-budget calls are not submitted.""" + release = threading.Event() + two_started = threading.Event() + calls = 0 + calls_lock = threading.Lock() + deadline = [0.05] + + class _BlockingSyncSeller(DecisioningPlatform): + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + accounts = SingletonAccounts(account_id="test") + + def get_products(self, req, ctx): + nonlocal calls + with calls_lock: + calls += 1 + if calls == 2: + two_started.set() + release.wait(timeout=2.0) + return {"products": [{"product_id": f"p{calls}", "name": "Recovered"}]} + + monkeypatch.setattr("adcp.decisioning.handler.resolve_time_budget", lambda _value: deadline[0]) + handler = PlatformHandler( + _BlockingSyncSeller(), + executor=executor, + registry=InMemoryTaskRegistry(), + timed_sync_get_products_limit=2, + ) + req = GetProductsRequest.model_construct( + account=None, + time_budget=_make_time_budget(interval=1, unit="seconds"), + ) + + first_two = [ + asyncio.create_task(handler.get_products(req, context=ToolContext())) for _ in range(2) + ] + assert await asyncio.to_thread(two_started.wait, 1.0) + timed_out = await asyncio.gather(*first_two) + assert all(getattr(result, "incomplete", None) for result in timed_out) + + # Both permits remain attached to the still-running worker threads. This + # request exhausts its budget waiting and never reaches executor.submit. + saturated = await handler.get_products(req, context=ToolContext()) + assert getattr(saturated, "incomplete", None) + assert calls == 2 + + # Once real worker completion callbacks return the permits, admission + # recovers and a later call executes normally. + release.set() + deadline[0] = 0.5 + recovered = await handler.get_products(req, context=ToolContext()) + products = ( + recovered.get("products", []) + if isinstance(recovered, dict) + else list(getattr(recovered, "products", [])) + ) + assert len(products) == 1 + assert calls == 3 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("router_kind", ["eager", "lazy"]) +async def test_router_sync_timeout_uses_bounded_admission( + router_kind: str, + executor: ThreadPoolExecutor, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Eager and lazy async routers must not bypass sync-child admission.""" + release = threading.Event() + started = threading.Event() + calls = 0 + deadline = [0.05] + + class _BlockingChild(DecisioningPlatform): + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + accounts = SingletonAccounts(account_id="child") + + def get_products(self, req, ctx): + nonlocal calls + calls += 1 + started.set() + release.wait(timeout=2.0) + return {"products": [{"product_id": f"p{calls}", "name": "Recovered"}]} + + accounts = SingletonAccounts( + account_id="router", + metadata_factory=lambda: {"tenant_id": "tenant-a"}, + ) + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + if router_kind == "eager": + platform: DecisioningPlatform = PlatformRouter( + accounts=accounts, + platforms={"tenant-a": _BlockingChild()}, + capabilities=capabilities, + ) + else: + platform = LazyPlatformRouter( + accounts=accounts, + factory=lambda _tenant_id: _BlockingChild(), + capabilities=capabilities, + ) + + monkeypatch.setattr("adcp.decisioning.handler.resolve_time_budget", lambda _value: deadline[0]) + handler = PlatformHandler( + platform, + executor=executor, + registry=InMemoryTaskRegistry(), + timed_sync_get_products_limit=1, + ) + req = GetProductsRequest.model_construct( + account=None, + time_budget=_make_time_budget(interval=1, unit="seconds"), + ) + + first = asyncio.create_task(handler.get_products(req, context=ToolContext())) + assert await asyncio.to_thread(started.wait, 1.0) + assert getattr(await first, "incomplete", None) + + # The first timed-out child still owns the sole permit, so this request + # times out waiting for admission and is never submitted. + assert getattr(await handler.get_products(req, context=ToolContext()), "incomplete", None) + assert calls == 1 + + release.set() + deadline[0] = 0.5 + recovered = await handler.get_products(req, context=ToolContext()) + products = ( + recovered.get("products", []) + if isinstance(recovered, dict) + else list(getattr(recovered, "products", [])) + ) + assert len(products) == 1 + assert calls == 2 + + +@pytest.mark.asyncio +async def test_router_sync_without_deadline_uses_configured_executor() -> None: + observed_threads: list[str] = [] + + class _SyncChild(DecisioningPlatform): + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + accounts = SingletonAccounts(account_id="child") + + def get_products(self, req, ctx): + observed_threads.append(threading.current_thread().name) + return {"products": []} + + router = PlatformRouter( + accounts=SingletonAccounts( + account_id="router", + metadata_factory=lambda: {"tenant_id": "tenant-a"}, + ), + platforms={"tenant-a": _SyncChild()}, + capabilities=DecisioningCapabilities(specialisms=["sales-non-guaranteed"]), + ) + with ThreadPoolExecutor(max_workers=2, thread_name_prefix="framework-router-") as pool: + handler = PlatformHandler( + router, + executor=pool, + registry=InMemoryTaskRegistry(), + ) + req = GetProductsRequest.model_construct(account=None, time_budget=None) + await handler.get_products(req, context=ToolContext()) + + assert len(observed_threads) == 1 + assert observed_threads[0].startswith("framework-router-") + + +@pytest.mark.asyncio +async def test_sync_campaign_requests_bypass_deadline_admission() -> None: + """Campaign-unit semantics remain unlimited by the deadline-only gate.""" + release = threading.Event() + two_started = threading.Event() + calls = 0 + calls_lock = threading.Lock() + + class _CampaignSyncSeller(DecisioningPlatform): + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + accounts = SingletonAccounts(account_id="test") + + def get_products(self, req, ctx): + nonlocal calls + with calls_lock: + calls += 1 + if calls == 2: + two_started.set() + release.wait(timeout=2.0) + return {"products": [{"product_id": "campaign", "name": "Campaign"}]} + + with ThreadPoolExecutor(max_workers=2) as campaign_executor: + handler = PlatformHandler( + _CampaignSyncSeller(), + executor=campaign_executor, + registry=InMemoryTaskRegistry(), + timed_sync_get_products_limit=1, + ) + req = GetProductsRequest.model_construct( + account=None, + time_budget=_make_time_budget(interval=1, unit="campaign"), + ) + tasks = [ + asyncio.create_task(handler.get_products(req, context=ToolContext())) for _ in range(2) + ] + assert await asyncio.to_thread(two_started.wait, 1.0) + release.set() + results = await asyncio.gather(*tasks) + + assert calls == 2 + assert all( + len( + result.get("products", []) + if isinstance(result, dict) + else list(getattr(result, "products", [])) + ) + == 1 + for result in results + ) + + @pytest.mark.asyncio async def test_get_products_timeout_logs_warning(executor, caplog): """A timeout emits a WARNING with budget info.""" @@ -287,15 +540,15 @@ async def test_get_products_timeout_logs_warning(executor, caplog): def test_incremental_get_products_importable_from_decisioning(): - from adcp.decisioning import IncrementalGetProducts as IGP # noqa: F401 + from adcp.decisioning import IncrementalGetProducts as ImportedIncrementalGetProducts - assert IGP is IncrementalGetProducts + assert ImportedIncrementalGetProducts is IncrementalGetProducts def test_products_checkpoint_importable_from_decisioning(): - from adcp.decisioning import ProductsCheckpoint as PC # noqa: F401 + from adcp.decisioning import ProductsCheckpoint as ImportedProductsCheckpoint - assert PC is ProductsCheckpoint + assert ImportedProductsCheckpoint is ProductsCheckpoint def test_products_checkpoint_accumulates_batches():