From 241b70533738837f918f6e9c800986ea2ef28aa6 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 24 Aug 2026 14:03:13 +0200 Subject: [PATCH] feat(client): add per-call task deadlines --- README.md | 29 ++ src/adcp/__init__.py | 7 + src/adcp/client.py | 541 ++++++++++++++++++++---- src/adcp/exceptions.py | 20 +- src/adcp/protocols/a2a.py | 7 + src/adcp/protocols/base.py | 2 + src/adcp/protocols/mcp.py | 13 + src/adcp/task_options.py | 133 ++++++ tests/fixtures/public_api_snapshot.json | 2 + tests/test_task_options.py | 442 +++++++++++++++++++ tests/type_checks/task_options.py | 10 + 11 files changed, 1119 insertions(+), 87 deletions(-) create mode 100644 src/adcp/task_options.py create mode 100644 tests/test_task_options.py create mode 100644 tests/type_checks/task_options.py diff --git a/README.md b/README.md index 45c53400e..17ce9dfa2 100644 --- a/README.md +++ b/README.md @@ -799,6 +799,35 @@ finally: In most cases, prefer the context manager pattern. +### Per-call task deadlines + +Use `TaskOptions` when a complete SDK call must fit one wall-clock budget: + +```python +from adcp import ADCPTimeoutError, TaskOptions + +try: + result = await client.create_media_buy( + request, + options=TaskOptions(timeout=15.0), + ) +except ADCPTimeoutError as error: + if error.recovery is not None: + # Dispatch began, so the seller may have committed the mutation. + # Retry the exact request with this same key; never mint a new one. + retry_key = error.recovery.idempotency_key +``` + +The deadline includes discovery, capability/version and signing preflight, +protocol dispatch, response validation, and postflight projection. It never +resets when one phase finishes. `AgentConfig.timeout` remains a separate +transport timeout for connection/read-idle behavior. + +Every single-agent task method accepts the keyword-only `options` argument. +Multi-agent fan-out intentionally does not yet accept it because one timeout +exception cannot safely represent several sellers' independent mutation +outcomes and idempotency keys. + ### Error Handling The library provides a comprehensive exception hierarchy with helpful error messages: diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py index 7b88295bc..3ffb272a0 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -140,6 +140,10 @@ def _resolve_version() -> str: "ADCPMultiAgentClient", "Checkpoint", ), + "adcp.task_options": ( + "TaskOptions", + "TaskRecoveryMetadata", + ), "adcp.exceptions": ( "AdagentsAccessBlockedError", "AdagentsNotFoundError", @@ -843,6 +847,8 @@ def get_adcp_version() -> str: "ADCPClient", "ADCPMultiAgentClient", "Checkpoint", + "TaskOptions", + "TaskRecoveryMetadata", "RegistryClient", "PropertyRegistry", "RegistrySync", @@ -1564,6 +1570,7 @@ def get_adcp_version() -> str: encode_unreserved, translate_universal_macros, ) + from adcp.task_options import TaskOptions, TaskRecoveryMetadata from adcp.testing import ( CREATIVE_AGENT_CONFIG, TEST_AGENT_A2A_CONFIG, diff --git a/src/adcp/client.py b/src/adcp/client.py index 118216c27..f29a0ef55 100644 --- a/src/adcp/client.py +++ b/src/adcp/client.py @@ -11,9 +11,10 @@ import os import time import warnings -from collections.abc import Callable, Iterator, Mapping +from collections.abc import Awaitable, Callable, Coroutine, Iterator, Mapping from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, TypedDict, cast +from functools import wraps +from typing import TYPE_CHECKING, Any, ParamSpec, TypedDict, TypeVar, cast from uuid import uuid4 from a2a.types import Task, TaskStatusUpdateEvent @@ -39,7 +40,7 @@ ) from adcp.capabilities import TASK_FEATURE_MAP, FeatureResolver, looks_like_v3_capabilities from adcp.compat.legacy import LEGACY_ADAPTER_VERSIONS -from adcp.exceptions import ADCPError, ADCPWebhookSignatureError +from adcp.exceptions import ADCPError, ADCPTimeoutError, ADCPWebhookSignatureError from adcp.negotiation import ( WIRE_RESPONSE_METADATA_KEY, VerifiedRefinementResult, @@ -60,6 +61,15 @@ current_operation as _signing_current_operation, ) from adcp.signing.signer import sign_request +from adcp.task_options import ( + TaskOptions, + _get_task_execution, + _reset_task_execution, + _set_task_execution, + _TaskDeadlineExpiredError, + _TaskExecutionState, + run_with_timeout, +) from adcp.types import ( AcceptProposalRequest, AcceptProposalResponse, @@ -385,6 +395,38 @@ logger = logging.getLogger(__name__) +P = ParamSpec("P") +R = TypeVar("R") + + +def _task_options_method( + method: Callable[P, Coroutine[Any, Any, R]], +) -> Callable[P, Coroutine[Any, Any, R]]: + """Apply an explicit method's ``options`` under one outer deadline.""" + + @wraps(method) + async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R: + call_kwargs = cast(dict[str, Any], kwargs) + options = call_kwargs.pop("options", None) + client = cast("ADCPClient", args[0]) + if options is not None and not isinstance(options, TaskOptions): + raise TypeError("options must be a TaskOptions instance") + if options is None: + return await method(*args, **kwargs) + task_name = method.__name__ + if task_name in {"execute_task", "execute_task_legacy"}: + requested_name = args[1] if len(args) > 1 else call_kwargs.get("task_name") + if isinstance(requested_name, str): + task_name = requested_name + return await client._run_with_task_options( + task_name, + lambda: method(*args, **kwargs), + options, + ) + + return wrapped + + _LEGACY_CREATIVE_TASKS = frozenset( { "build_creative", @@ -678,6 +720,7 @@ def __init__( from uuid import uuid4 as _uuid4 self._idempotency_client_token: str = _uuid4().hex + self._task_options_token = object() if force_a2a_version is not None and agent_config.protocol != Protocol.A2A: raise TypeError( @@ -695,6 +738,7 @@ def __init__( ) else: raise ValueError(f"Unsupported protocol: {agent_config.protocol}") + self.adapter.task_options_client_token = self._task_options_token self.adapter.idempotency_client_token = self._idempotency_client_token if strict_idempotency: @@ -1086,7 +1130,7 @@ async def _ensure_idempotency_capability(self) -> None: "replay_ttl_seconds" ), ) - except Exception: + except BaseException: self._idempotency_capability_verified = False raise @@ -1227,6 +1271,62 @@ def _emit_activity(self, activity: Activity) -> None: if self.on_activity: self.on_activity(activity) + async def _run_with_task_options( + self, + task_name: str, + awaitable_factory: Callable[[], Awaitable[R]], + options: TaskOptions, + ) -> R: + """Run one outermost task call under a single non-resetting deadline.""" + + # Generic execute_task() and convenience helpers call other public + # client methods. They inherit the active outer deadline instead of + # starting a fresh budget at each layer. + if _get_task_execution(self._task_options_token) is not None: + return await awaitable_factory() + + operation_id = self._task_operation_id() + loop = asyncio.get_running_loop() + deadline = loop.time() + options.timeout if options.timeout is not None else None + state = _TaskExecutionState( + client_token=self._task_options_token, + operation_id=operation_id, + timeout=options.timeout, + deadline=deadline, + task_name=task_name, + ) + token = _set_task_execution(state) + try: + if options.timeout is None: + return await awaitable_factory() + try: + assert deadline is not None + result = await run_with_timeout( + awaitable_factory, + max(0.0, deadline - loop.time()), + ) + if loop.time() >= deadline: + raise _TaskDeadlineExpiredError + return result + except _TaskDeadlineExpiredError as exc: + raise ADCPTimeoutError( + f"Task {task_name!r} timed out after {options.timeout}s", + agent_id=self.agent_config.id, + agent_uri=self.agent_config.agent_uri, + timeout=options.timeout, + task_name=state.task_name, + operation_id=operation_id, + recovery=state.mutation_recovery, + ) from exc + finally: + _reset_task_execution(token) + + def _task_operation_id(self) -> str: + """Reuse the outer task ID for activities and timeout recovery.""" + + state = _get_task_execution(self._task_options_token) + return state.operation_id if state is not None else create_operation_id() + @contextlib.contextmanager def use_idempotency_key(self, key: str) -> Iterator[str]: """Pin an ``idempotency_key`` for the next mutating call on THIS client. @@ -1831,7 +1931,7 @@ async def _execute_typed_task( response_type: type[BaseModel] | Any, ) -> TaskResult[Any]: """Execute and parse one typed AdCP task with activity events.""" - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( Activity( @@ -1861,15 +1961,22 @@ async def _execute_typed_task( parsed_result = parsed_result.model_copy(update={"metadata": metadata}) return parsed_result - async def list_products(self, request: ListProductsRequest) -> TaskResult[ListProductsResponse]: + @_task_options_method + async def list_products( + self, request: ListProductsRequest, *, options: TaskOptions | None = None + ) -> TaskResult[ListProductsResponse]: """List products using the AdCP 3.2 compact discovery lifecycle.""" return cast( TaskResult[ListProductsResponse], await self._execute_typed_task("list_products", request, ListProductsResponse), ) + @_task_options_method async def request_proposals( - self, request: RequestProposalsRequest + self, + request: RequestProposalsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[RequestProposalsResponse]: """Request seller proposals for selected products.""" return cast( @@ -1877,8 +1984,12 @@ async def request_proposals( await self._execute_typed_task("request_proposals", request, RequestProposalsResponse), ) + @_task_options_method async def refine_proposals( - self, request: RefineProposalsRequest + self, + request: RefineProposalsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[RefineProposalsResponse]: """Refine one or more seller proposals.""" return cast( @@ -1886,6 +1997,7 @@ async def refine_proposals( await self._execute_typed_task("refine_proposals", request, RefineProposalsResponse), ) + @_task_options_method async def refine_proposals_verified( self, request: RefineProposalsRequest, @@ -1893,6 +2005,7 @@ async def refine_proposals_verified( *, source_proposals: Mapping[str, BaseModel | Mapping[str, Any]] | None = None, now: datetime | None = None, + options: TaskOptions | None = None, ) -> VerifiedRefinementResult: """Preflight, execute, and verify a proposal-refinement request. @@ -1910,6 +2023,7 @@ async def refine_proposals_verified( now=now, ) + @_task_options_method async def wait_for_refinement_verified( self, request: RefineProposalsRequest, @@ -1920,6 +2034,7 @@ async def wait_for_refinement_verified( timeout: float = 300.0, poll_interval: float = 1.0, now: datetime | None = None, + options: TaskOptions | None = None, ) -> VerifiedRefinementResult: """Poll a submitted refinement and verify its exact terminal payload.""" @@ -1976,8 +2091,12 @@ async def wait_for_refinement_verified( raise TimeoutError(f"refinement task {task_id!r} did not complete in time") await asyncio.sleep(min(poll_interval, remaining)) + @_task_options_method async def decline_proposals( - self, request: DeclineProposalsRequest + self, + request: DeclineProposalsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[DeclineProposalsResponse]: """Decline one or more seller proposals.""" return cast( @@ -1985,15 +2104,22 @@ async def decline_proposals( await self._execute_typed_task("decline_proposals", request, DeclineProposalsResponse), ) - async def buy_products(self, request: BuyProductsRequest) -> TaskResult[BuyProductsResponse]: + @_task_options_method + async def buy_products( + self, request: BuyProductsRequest, *, options: TaskOptions | None = None + ) -> TaskResult[BuyProductsResponse]: """Commit a direct product purchase.""" return cast( TaskResult[BuyProductsResponse], await self._execute_typed_task("buy_products", request, BuyProductsResponse), ) + @_task_options_method async def accept_proposal( - self, request: AcceptProposalRequest + self, + request: AcceptProposalRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[AcceptProposalResponse]: """Accept a seller proposal and create its media buy.""" return cast( @@ -2001,8 +2127,12 @@ async def accept_proposal( await self._execute_typed_task("accept_proposal", request, AcceptProposalResponse), ) + @_task_options_method async def control_media_buy( - self, request: ControlMediaBuyRequest + self, + request: ControlMediaBuyRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ControlMediaBuyResponse]: """Apply lifecycle controls to an existing media buy.""" return cast( @@ -2010,12 +2140,15 @@ async def control_media_buy( await self._execute_typed_task("control_media_buy", request, ControlMediaBuyResponse), ) + @_task_options_method async def get_products( self, request: GetProductsRequest, fetch_previews: bool = False, preview_output_format: str = "url", creative_agent_client: ADCPClient | None = None, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetProductsResponse]: """ Get advertising products. @@ -2039,7 +2172,7 @@ async def get_products( raise ValueError("creative_agent_client is required when fetch_previews=True") self._creative_dialect(request, legacy_projection_available=True) - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2087,9 +2220,12 @@ async def get_products( return result + @_task_options_method async def get_products_legacy( self, request: LegacyGetProductsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacyGetProductsResponse]: """Return the raw AdCP 3.x product wire shape for migration tooling.""" @@ -2105,11 +2241,14 @@ def _warn_legacy_creative_api(method: str) -> None: stacklevel=3, ) + @_task_options_method async def list_creative_formats_legacy( self, request: ListCreativeFormatsRequest, fetch_previews: bool = False, preview_output_format: str = "url", + *, + options: TaskOptions | None = None, ) -> TaskResult[ListCreativeFormatsResponse]: """ List supported creative formats. @@ -2125,7 +2264,7 @@ async def list_creative_formats_legacy( TaskResult containing ListCreativeFormatsResponse with optional preview URLs in metadata """ self._warn_legacy_creative_api("list_creative_formats_legacy") - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2169,8 +2308,12 @@ async def list_creative_formats_legacy( return result + @_task_options_method async def create_media_buy_legacy( - self, request: LegacyCreateMediaBuyRequest + self, + request: LegacyCreateMediaBuyRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacyCreateMediaBuyResponse]: """Execute create_media_buy without the canonical application boundary.""" @@ -2180,8 +2323,12 @@ async def create_media_buy_legacy( ) return self.adapter._parse_response(raw, LegacyCreateMediaBuyResponse) + @_task_options_method async def update_media_buy_legacy( - self, request: LegacyUpdateMediaBuyRequest + self, + request: LegacyUpdateMediaBuyRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacyUpdateMediaBuyResponse]: """Execute update_media_buy without the canonical application boundary.""" @@ -2191,8 +2338,12 @@ async def update_media_buy_legacy( ) return self.adapter._parse_response(raw, LegacyUpdateMediaBuyResponse) + @_task_options_method async def sync_creatives_legacy( - self, request: LegacySyncCreativesRequest + self, + request: LegacySyncCreativesRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacySyncCreativesResponse]: """Execute sync_creatives without the canonical application boundary.""" @@ -2200,8 +2351,12 @@ async def sync_creatives_legacy( raw = await self.adapter.sync_creatives(request.model_dump(mode="json", exclude_none=True)) return self.adapter._parse_response(raw, LegacySyncCreativesResponse) + @_task_options_method async def list_creatives_legacy( - self, request: LegacyListCreativesRequest + self, + request: LegacyListCreativesRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacyListCreativesResponse]: """Return raw creative rows carrying legacy format identity.""" @@ -2209,8 +2364,12 @@ async def list_creatives_legacy( raw = await self.adapter.list_creatives(request.model_dump(mode="json", exclude_none=True)) return self.adapter._parse_response(raw, LegacyListCreativesResponse) + @_task_options_method async def get_media_buys_legacy( - self, request: GetMediaBuysRequest + self, + request: GetMediaBuysRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacyGetMediaBuysResponse]: """Return raw media-buy rows carrying legacy format identity.""" @@ -2218,8 +2377,12 @@ async def get_media_buys_legacy( raw = await self.adapter.get_media_buys(request.model_dump(mode="json", exclude_none=True)) return self.adapter._parse_response(raw, LegacyGetMediaBuysResponse) + @_task_options_method async def get_media_buy_delivery_legacy( - self, request: GetMediaBuyDeliveryRequest + self, + request: GetMediaBuyDeliveryRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacyGetMediaBuyDeliveryResponse]: """Return raw media-buy delivery carrying legacy format identity.""" @@ -2229,8 +2392,12 @@ async def get_media_buy_delivery_legacy( ) return self.adapter._parse_response(raw, LegacyGetMediaBuyDeliveryResponse) + @_task_options_method async def get_creative_delivery_legacy( - self, request: GetCreativeDeliveryRequest + self, + request: GetCreativeDeliveryRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacyGetCreativeDeliveryResponse]: """Return raw creative delivery carrying legacy format identity.""" @@ -2240,9 +2407,12 @@ async def get_creative_delivery_legacy( ) return self.adapter._parse_response(raw, LegacyGetCreativeDeliveryResponse) + @_task_options_method async def preview_creative_legacy( self, request: LegacyPreviewCreativeRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacyPreviewCreativeResponse]: """ Generate preview of a creative manifest. @@ -2253,7 +2423,7 @@ async def preview_creative_legacy( Returns: TaskResult containing PreviewCreativeResponse with preview URLs """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2282,9 +2452,12 @@ async def preview_creative_legacy( self._warn_legacy_creative_api("preview_creative_legacy") return self.adapter._parse_response(raw_result, LegacyPreviewCreativeResponse) + @_task_options_method async def sync_creatives( self, request: SyncCreativesRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SyncCreativesResponse]: """ Sync Creatives. @@ -2296,7 +2469,7 @@ async def sync_creatives( TaskResult containing SyncCreativesResponse """ dialect = self._creative_dialect(request, legacy_projection_available=True) - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = self._prepare_creative_params(request) self._emit_activity( @@ -2333,9 +2506,12 @@ async def sync_creatives( ) return self.adapter._parse_response(raw_result, SyncCreativesResponse) + @_task_options_method async def list_creatives( self, request: ListCreativesRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ListCreativesResponse]: """ List Creatives. @@ -2347,7 +2523,7 @@ async def list_creatives( TaskResult containing ListCreativesResponse """ dialect = self._creative_dialect(request, legacy_projection_available=True) - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2386,9 +2562,12 @@ async def list_creatives( ), ) + @_task_options_method async def get_media_buy_delivery( self, request: GetMediaBuyDeliveryRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetMediaBuyDeliveryResponse]: """ Get Media Buy Delivery. @@ -2400,7 +2579,7 @@ async def get_media_buy_delivery( TaskResult containing GetMediaBuyDeliveryResponse """ dialect = self._creative_dialect(request, legacy_projection_available=True) - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2437,9 +2616,12 @@ async def get_media_buy_delivery( ) return self.adapter._parse_response(raw_result, GetMediaBuyDeliveryResponse) + @_task_options_method async def get_media_buys( self, request: GetMediaBuysRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetMediaBuysResponse]: """ Get Media Buys. @@ -2451,7 +2633,7 @@ async def get_media_buys( TaskResult containing GetMediaBuysResponse """ dialect = self._creative_dialect(request, legacy_projection_available=True) - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) if params.get("include_webhook_activity") is False: params.pop("include_webhook_activity") @@ -2492,9 +2674,12 @@ async def get_media_buys( ) return self.adapter._parse_response(raw_result, GetMediaBuysResponse) + @_task_options_method async def get_signals( self, request: GetSignalsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetSignalsResponse]: """ Get Signals. @@ -2505,7 +2690,7 @@ async def get_signals( Returns: TaskResult containing GetSignalsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2533,9 +2718,12 @@ async def get_signals( return self.adapter._parse_response(raw_result, GetSignalsResponse) + @_task_options_method async def activate_signal( self, request: ActivateSignalRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ActivateSignalResponse]: """ Activate Signal. @@ -2546,7 +2734,7 @@ async def activate_signal( Returns: TaskResult containing ActivateSignalResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2574,9 +2762,12 @@ async def activate_signal( return self.adapter._parse_response(raw_result, ActivateSignalResponse) + @_task_options_method async def provide_performance_feedback( self, request: ProvidePerformanceFeedbackRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ProvidePerformanceFeedbackResponse]: """ Provide Performance Feedback. @@ -2587,7 +2778,7 @@ async def provide_performance_feedback( Returns: TaskResult containing ProvidePerformanceFeedbackResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2615,9 +2806,12 @@ async def provide_performance_feedback( return self.adapter._parse_response(raw_result, ProvidePerformanceFeedbackResponse) + @_task_options_method async def create_media_buy( self, request: CreateMediaBuyRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[CreateMediaBuyResponse]: """ Create a new media buy reservation. @@ -2653,7 +2847,7 @@ async def create_media_buy( ... media_buy_id = result.data.media_buy_id """ dialect = self._creative_dialect(request, legacy_projection_available=True) - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = self._prepare_creative_params(request) self._emit_activity( @@ -2690,9 +2884,12 @@ async def create_media_buy( ) return self.adapter._parse_response(raw_result, CreateMediaBuyResponse) + @_task_options_method async def update_media_buy( self, request: UpdateMediaBuyRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[UpdateMediaBuyResponse]: """ Update an existing media buy reservation. @@ -2727,7 +2924,7 @@ async def update_media_buy( ... updated_packages = result.data.packages """ dialect = self._creative_dialect(request, legacy_projection_available=True) - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = self._prepare_creative_params(request) self._emit_activity( @@ -2764,9 +2961,12 @@ async def update_media_buy( ) return self.adapter._parse_response(raw_result, UpdateMediaBuyResponse) + @_task_options_method async def build_creative_legacy( self, request: LegacyBuildCreativeRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LegacyBuildCreativeResponse]: """ Generate production-ready creative assets. @@ -2801,7 +3001,7 @@ async def build_creative_legacy( >>> if result.success: ... vast_url = result.data.assets[0].url """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2830,9 +3030,12 @@ async def build_creative_legacy( self._warn_legacy_creative_api("build_creative_legacy") return self.adapter._parse_response(raw_result, LegacyBuildCreativeResponse) + @_task_options_method async def list_accounts( self, request: ListAccountsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ListAccountsResponse]: """ List Accounts. @@ -2843,7 +3046,7 @@ async def list_accounts( Returns: TaskResult containing ListAccountsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2871,9 +3074,12 @@ async def list_accounts( return self.adapter._parse_response(raw_result, ListAccountsResponse) + @_task_options_method async def sync_accounts( self, request: SyncAccountsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SyncAccountsResponse]: """ Sync Accounts. @@ -2884,7 +3090,7 @@ async def sync_accounts( Returns: TaskResult containing SyncAccountsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2912,9 +3118,12 @@ async def sync_accounts( return self.adapter._parse_response(raw_result, SyncAccountsResponse) + @_task_options_method async def get_account_financials( self, request: GetAccountFinancialsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetAccountFinancialsResponse]: """ Get Account Financials. @@ -2925,7 +3134,7 @@ async def get_account_financials( Returns: TaskResult containing GetAccountFinancialsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2953,9 +3162,12 @@ async def get_account_financials( return self.adapter._parse_response(raw_result, GetAccountFinancialsResponse) + @_task_options_method async def report_usage( self, request: ReportUsageRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ReportUsageResponse]: """ Report Usage. @@ -2966,7 +3178,7 @@ async def report_usage( Returns: TaskResult containing ReportUsageResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -2994,9 +3206,12 @@ async def report_usage( return self.adapter._parse_response(raw_result, ReportUsageResponse) + @_task_options_method async def log_event( self, request: LogEventRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[LogEventResponse]: """ Log Event. @@ -3008,7 +3223,7 @@ async def log_event( TaskResult containing LogEventResponse """ self._validate_task_features("log_event") - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3036,9 +3251,12 @@ async def log_event( return self.adapter._parse_response(raw_result, LogEventResponse) + @_task_options_method async def sync_event_sources( self, request: SyncEventSourcesRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SyncEventSourcesResponse]: """ Sync Event Sources. @@ -3050,7 +3268,7 @@ async def sync_event_sources( TaskResult containing SyncEventSourcesResponse """ self._validate_task_features("sync_event_sources") - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3078,9 +3296,12 @@ async def sync_event_sources( return self.adapter._parse_response(raw_result, SyncEventSourcesResponse) + @_task_options_method async def sync_audiences( self, request: SyncAudiencesRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SyncAudiencesResponse]: """ Sync Audiences. @@ -3091,7 +3312,7 @@ async def sync_audiences( Returns: TaskResult containing SyncAudiencesResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3119,9 +3340,12 @@ async def sync_audiences( return self.adapter._parse_response(raw_result, SyncAudiencesResponse) + @_task_options_method async def sync_catalogs( self, request: SyncCatalogsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SyncCatalogsResponse]: """ Sync Catalogs. @@ -3132,7 +3356,7 @@ async def sync_catalogs( Returns: TaskResult containing SyncCatalogsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3160,9 +3384,12 @@ async def sync_catalogs( return self.adapter._parse_response(raw_result, SyncCatalogsResponse) + @_task_options_method async def get_creative_delivery( self, request: GetCreativeDeliveryRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetCreativeDeliveryResponse]: """ Get Creative Delivery. @@ -3174,7 +3401,7 @@ async def get_creative_delivery( TaskResult containing GetCreativeDeliveryResponse """ self._creative_dialect(request, legacy_projection_available=True) - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3216,9 +3443,12 @@ async def get_creative_delivery( ), ) + @_task_options_method async def list_transformers( self, request: ListTransformersRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ListTransformersResponse]: """ List Creative Transformers. @@ -3229,7 +3459,7 @@ async def list_transformers( Returns: TaskResult containing ListTransformersResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3261,9 +3491,12 @@ async def list_transformers( # V3 Protocol Methods - Protocol Discovery # ======================================================================== + @_task_options_method async def get_adcp_capabilities( self, request: GetAdcpCapabilitiesRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetAdcpCapabilitiesResponse]: """ Get AdCP capabilities from the agent. @@ -3282,7 +3515,7 @@ async def get_adcp_capabilities( - sponsored_intelligence: SI capabilities (if supported) - signals: Signals capabilities (if supported) """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3310,9 +3543,12 @@ async def get_adcp_capabilities( return self.adapter._parse_response(raw_result, GetAdcpCapabilitiesResponse) + @_task_options_method async def sync_agent_notification_configs( self, request: SyncAgentNotificationConfigsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SyncAgentNotificationConfigsResponse]: """Replace the caller-scoped agent notification subscriber set.""" return cast( @@ -3324,9 +3560,12 @@ async def sync_agent_notification_configs( ), ) + @_task_options_method async def get_task_status( self, request: GetTaskStatusRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetTaskStatusResponse]: """ Get Task Status. @@ -3337,7 +3576,7 @@ async def get_task_status( Returns: TaskResult containing GetTaskStatusResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3365,9 +3604,12 @@ async def get_task_status( return self.adapter._parse_response(raw_result, GetTaskStatusResponse) + @_task_options_method async def list_tasks( self, request: ListTasksRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ListTasksResponse]: """ List Tasks. @@ -3378,7 +3620,7 @@ async def list_tasks( Returns: TaskResult containing ListTasksResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3410,9 +3652,12 @@ async def list_tasks( # V3 Protocol Methods - Content Standards # ======================================================================== + @_task_options_method async def create_content_standards( self, request: CreateContentStandardsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[CreateContentStandardsResponse]: """ Create a new content standards configuration. @@ -3426,7 +3671,7 @@ async def create_content_standards( Returns: TaskResult containing CreateContentStandardsResponse with standards_id """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3454,9 +3699,12 @@ async def create_content_standards( return self.adapter._parse_response(raw_result, CreateContentStandardsResponse) + @_task_options_method async def get_content_standards( self, request: GetContentStandardsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetContentStandardsResponse]: """ Get a content standards configuration by ID. @@ -3467,7 +3715,7 @@ async def get_content_standards( Returns: TaskResult containing GetContentStandardsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3495,9 +3743,12 @@ async def get_content_standards( return self.adapter._parse_response(raw_result, GetContentStandardsResponse) + @_task_options_method async def list_content_standards( self, request: ListContentStandardsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ListContentStandardsResponse]: """ List content standards configurations. @@ -3508,7 +3759,7 @@ async def list_content_standards( Returns: TaskResult containing ListContentStandardsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3536,9 +3787,12 @@ async def list_content_standards( return self.adapter._parse_response(raw_result, ListContentStandardsResponse) + @_task_options_method async def update_content_standards( self, request: UpdateContentStandardsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[UpdateContentStandardsResponse]: """ Update a content standards configuration. @@ -3549,7 +3803,7 @@ async def update_content_standards( Returns: TaskResult containing UpdateContentStandardsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3577,9 +3831,12 @@ async def update_content_standards( return self.adapter._parse_response(raw_result, UpdateContentStandardsResponse) + @_task_options_method async def calibrate_content( self, request: CalibrateContentRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[CalibrateContentResponse]: """ Calibrate content against standards. @@ -3593,7 +3850,7 @@ async def calibrate_content( Returns: TaskResult containing CalibrateContentResponse with verdict """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3621,9 +3878,12 @@ async def calibrate_content( return self.adapter._parse_response(raw_result, CalibrateContentResponse) + @_task_options_method async def validate_content_delivery( self, request: ValidateContentDeliveryRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ValidateContentDeliveryResponse]: """ Validate content delivery against standards. @@ -3636,7 +3896,7 @@ async def validate_content_delivery( Returns: TaskResult containing ValidateContentDeliveryResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3664,9 +3924,12 @@ async def validate_content_delivery( return self.adapter._parse_response(raw_result, ValidateContentDeliveryResponse) + @_task_options_method async def get_media_buy_artifacts( self, request: GetMediaBuyArtifactsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetMediaBuyArtifactsResponse]: """ Get artifacts associated with a media buy. @@ -3679,7 +3942,7 @@ async def get_media_buy_artifacts( Returns: TaskResult containing GetMediaBuyArtifactsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3711,9 +3974,12 @@ async def get_media_buy_artifacts( # V3 Protocol Methods - Sponsored Intelligence # ======================================================================== + @_task_options_method async def si_get_offering( self, request: SiGetOfferingRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SiGetOfferingResponse]: """ Get sponsored intelligence offering. @@ -3727,7 +3993,7 @@ async def si_get_offering( Returns: TaskResult containing SiGetOfferingResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3755,9 +4021,12 @@ async def si_get_offering( return self.adapter._parse_response(raw_result, SiGetOfferingResponse) + @_task_options_method async def si_initiate_session( self, request: SiInitiateSessionRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SiInitiateSessionResponse]: """ Initiate a sponsored intelligence session. @@ -3770,7 +4039,7 @@ async def si_initiate_session( Returns: TaskResult containing SiInitiateSessionResponse with session_id """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3798,9 +4067,12 @@ async def si_initiate_session( return self.adapter._parse_response(raw_result, SiInitiateSessionResponse) + @_task_options_method async def si_send_message( self, request: SiSendMessageRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SiSendMessageResponse]: """ Send a message in a sponsored intelligence session. @@ -3813,7 +4085,7 @@ async def si_send_message( Returns: TaskResult containing SiSendMessageResponse with brand response """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3841,9 +4113,12 @@ async def si_send_message( return self.adapter._parse_response(raw_result, SiSendMessageResponse) + @_task_options_method async def si_terminate_session( self, request: SiTerminateSessionRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SiTerminateSessionResponse]: """ Terminate a sponsored intelligence session. @@ -3856,7 +4131,7 @@ async def si_terminate_session( Returns: TaskResult containing SiTerminateSessionResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3888,12 +4163,15 @@ async def si_terminate_session( # V3 Governance Methods # ======================================================================== + @_task_options_method async def get_creative_features( self, request: GetCreativeFeaturesRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetCreativeFeaturesResponse]: """Evaluate governance features for a creative manifest.""" - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3921,12 +4199,15 @@ async def get_creative_features( return self.adapter._parse_response(raw_result, GetCreativeFeaturesResponse) + @_task_options_method async def sync_plans( self, request: SyncPlansRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SyncPlansResponse]: """Sync campaign governance plans to the governance agent.""" - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3954,12 +4235,15 @@ async def sync_plans( return self.adapter._parse_response(raw_result, SyncPlansResponse) + @_task_options_method async def check_governance( self, request: CheckGovernanceRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[CheckGovernanceResponse]: """Check a proposed or committed action against campaign governance.""" - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -3987,12 +4271,15 @@ async def check_governance( return self.adapter._parse_response(raw_result, CheckGovernanceResponse) + @_task_options_method async def report_plan_outcome( self, request: ReportPlanOutcomeRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ReportPlanOutcomeResponse]: """Report the outcome of a governed action to the governance agent.""" - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4020,9 +4307,12 @@ async def report_plan_outcome( return self.adapter._parse_response(raw_result, ReportPlanOutcomeResponse) + @_task_options_method async def report_plan_adjustment( self, request: ReportPlanAdjustmentRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ReportPlanAdjustmentResponse]: """Report or review an adjustment to a governed plan outcome.""" return cast( @@ -4032,12 +4322,15 @@ async def report_plan_adjustment( ), ) + @_task_options_method async def get_plan_audit_logs( self, request: GetPlanAuditLogsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetPlanAuditLogsResponse]: """Retrieve governance state and audit logs for one or more plans.""" - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4065,9 +4358,12 @@ async def get_plan_audit_logs( return self.adapter._parse_response(raw_result, GetPlanAuditLogsResponse) + @_task_options_method async def create_property_list( self, request: CreatePropertyListRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[CreatePropertyListResponse]: """ Create a property list for governance filtering. @@ -4081,7 +4377,7 @@ async def create_property_list( Returns: TaskResult containing CreatePropertyListResponse with list_id """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4109,9 +4405,12 @@ async def create_property_list( return self.adapter._parse_response(raw_result, CreatePropertyListResponse) + @_task_options_method async def get_property_list( self, request: GetPropertyListRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetPropertyListResponse]: """ Get a property list with optional resolution. @@ -4125,7 +4424,7 @@ async def get_property_list( Returns: TaskResult containing GetPropertyListResponse with identifiers """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4153,9 +4452,12 @@ async def get_property_list( return self.adapter._parse_response(raw_result, GetPropertyListResponse) + @_task_options_method async def list_property_lists( self, request: ListPropertyListsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ListPropertyListsResponse]: """ List property lists owned by a principal. @@ -4169,7 +4471,7 @@ async def list_property_lists( Returns: TaskResult containing ListPropertyListsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4197,9 +4499,12 @@ async def list_property_lists( return self.adapter._parse_response(raw_result, ListPropertyListsResponse) + @_task_options_method async def update_property_list( self, request: UpdatePropertyListRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[UpdatePropertyListResponse]: """ Update a property list. @@ -4213,7 +4518,7 @@ async def update_property_list( Returns: TaskResult containing UpdatePropertyListResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4241,9 +4546,12 @@ async def update_property_list( return self.adapter._parse_response(raw_result, UpdatePropertyListResponse) + @_task_options_method async def delete_property_list( self, request: DeletePropertyListRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[DeletePropertyListResponse]: """ Delete a property list. @@ -4257,7 +4565,7 @@ async def delete_property_list( Returns: TaskResult containing DeletePropertyListResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4289,9 +4597,12 @@ async def delete_property_list( # V3 Protocol Methods - Governance (Collection Lists) # ======================================================================== + @_task_options_method async def create_collection_list( self, request: CreateCollectionListRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[CreateCollectionListResponse]: """Create a collection list for governance filtering. @@ -4304,7 +4615,7 @@ async def create_collection_list( Returns: TaskResult containing CreateCollectionListResponse with list_id """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4332,9 +4643,12 @@ async def create_collection_list( return self.adapter._parse_response(raw_result, CreateCollectionListResponse) + @_task_options_method async def get_collection_list( self, request: GetCollectionListRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetCollectionListResponse]: """Get a collection list with optional resolution. @@ -4346,7 +4660,7 @@ async def get_collection_list( Returns: TaskResult containing GetCollectionListResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4374,9 +4688,12 @@ async def get_collection_list( return self.adapter._parse_response(raw_result, GetCollectionListResponse) + @_task_options_method async def list_collection_lists( self, request: ListCollectionListsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ListCollectionListsResponse]: """List collection lists owned by a principal. @@ -4386,7 +4703,7 @@ async def list_collection_lists( Returns: TaskResult containing ListCollectionListsResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4414,9 +4731,12 @@ async def list_collection_lists( return self.adapter._parse_response(raw_result, ListCollectionListsResponse) + @_task_options_method async def update_collection_list( self, request: UpdateCollectionListRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[UpdateCollectionListResponse]: """Update a collection list. @@ -4426,7 +4746,7 @@ async def update_collection_list( Returns: TaskResult containing UpdateCollectionListResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4454,9 +4774,12 @@ async def update_collection_list( return self.adapter._parse_response(raw_result, UpdateCollectionListResponse) + @_task_options_method async def delete_collection_list( self, request: DeleteCollectionListRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[DeleteCollectionListResponse]: """Delete a collection list. @@ -4466,7 +4789,7 @@ async def delete_collection_list( Returns: TaskResult containing DeleteCollectionListResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4498,9 +4821,12 @@ async def delete_collection_list( # V3 Protocol Methods - Governance (Sync Governance) # ======================================================================== + @_task_options_method async def sync_governance( self, request: SyncGovernanceRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[SyncGovernanceResponse]: """Sync governance agents attached to an account. @@ -4513,7 +4839,7 @@ async def sync_governance( Returns: TaskResult containing SyncGovernanceResponse """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4545,9 +4871,12 @@ async def sync_governance( # V3 Protocol Methods - Temporal Matching Protocol (TMP) # ======================================================================== + @_task_options_method async def context_match( self, request: ContextMatchRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ContextMatchResponse]: """Match ad context to buyer packages. @@ -4561,7 +4890,7 @@ async def context_match( Returns: TaskResult containing ContextMatchResponse with offers. """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True, by_alias=True) self._emit_activity( @@ -4589,9 +4918,12 @@ async def context_match( return self.adapter._parse_response(raw_result, ContextMatchResponse) + @_task_options_method async def identity_match( self, request: IdentityMatchRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[IdentityMatchResponse]: """Match user identity for package eligibility. @@ -4605,7 +4937,7 @@ async def identity_match( Returns: TaskResult containing IdentityMatchResponse with eligible_package_ids. """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True, by_alias=True) self._emit_activity( @@ -4637,9 +4969,12 @@ async def identity_match( # V3 Protocol Methods - Brand Rights # ======================================================================== + @_task_options_method async def get_brand_identity( self, request: GetBrandIdentityRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetBrandIdentityResponse]: """Get brand identity information. @@ -4652,7 +4987,7 @@ async def get_brand_identity( Returns: TaskResult containing GetBrandIdentityResponse. """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4680,9 +5015,12 @@ async def get_brand_identity( return self.adapter._parse_response(raw_result, GetBrandIdentityResponse) + @_task_options_method async def get_rights( self, request: GetRightsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[GetRightsResponse]: """Get available rights for licensing. @@ -4695,7 +5033,7 @@ async def get_rights( Returns: TaskResult containing GetRightsResponse with matched rights. """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4723,9 +5061,12 @@ async def get_rights( return self.adapter._parse_response(raw_result, GetRightsResponse) + @_task_options_method async def acquire_rights( self, request: AcquireRightsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[AcquireRightsResponse]: """Acquire rights for brand content usage. @@ -4740,7 +5081,7 @@ async def acquire_rights( TaskResult containing AcquireRightsResponse (acquired, pending_approval, rejected, or error). """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4768,9 +5109,12 @@ async def acquire_rights( return self.adapter._parse_response(raw_result, AcquireRightsResponse) + @_task_options_method async def update_rights( self, request: UpdateRightsRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[UpdateRightsResponse]: """Update terms of an existing rights acquisition. @@ -4797,7 +5141,7 @@ async def update_rights( Returns: TaskResult containing UpdateRightsResponse (updated or error). """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4825,7 +5169,10 @@ async def update_rights( return self.adapter._parse_response(raw_result, UpdateRightsResponse) - async def validate_input(self, request: Any) -> TaskResult[Any]: + @_task_options_method + async def validate_input( + self, request: Any, *, options: TaskOptions | None = None + ) -> TaskResult[Any]: """Validate creative input against a format declaration.""" from adcp.types import _generated as gen @@ -4833,7 +5180,10 @@ async def validate_input(self, request: Any) -> TaskResult[Any]: raw_result = await self.adapter.validate_input(params) return self.adapter._parse_response(raw_result, gen.ValidateInputResponse) - async def verify_brand_claim(self, request: Any) -> TaskResult[Any]: + @_task_options_method + async def verify_brand_claim( + self, request: Any, *, options: TaskOptions | None = None + ) -> TaskResult[Any]: """Verify a single brand claim.""" from adcp.types import _generated as gen @@ -4841,7 +5191,10 @@ async def verify_brand_claim(self, request: Any) -> TaskResult[Any]: raw_result = await self.adapter.verify_brand_claim(params) return self.adapter._parse_response(raw_result, gen.VerifyBrandClaimResponse) - async def verify_brand_claims(self, request: Any) -> TaskResult[Any]: + @_task_options_method + async def verify_brand_claims( + self, request: Any, *, options: TaskOptions | None = None + ) -> TaskResult[Any]: """Verify multiple brand claims.""" from adcp.types import _generated as gen @@ -4853,9 +5206,12 @@ async def verify_brand_claims(self, request: Any) -> TaskResult[Any]: # V3 Protocol Methods - Compliance # ======================================================================== + @_task_options_method async def comply_test_controller( self, request: ComplyTestControllerRequest, + *, + options: TaskOptions | None = None, ) -> TaskResult[ComplyTestControllerResponse]: """Compliance test controller for sandbox testing. @@ -4868,7 +5224,7 @@ async def comply_test_controller( Returns: TaskResult containing ComplyTestControllerResponse. """ - operation_id = create_operation_id() + operation_id = self._task_operation_id() params = request.model_dump(mode="json", exclude_none=True) self._emit_activity( @@ -4896,7 +5252,14 @@ async def comply_test_controller( return self.adapter._parse_response(raw_result, ComplyTestControllerResponse) - async def execute_task(self, task_name: str, request: BaseModel) -> TaskResult[Any]: + @_task_options_method + async def execute_task( + self, + task_name: str, + request: BaseModel, + *, + options: TaskOptions | None = None, + ) -> TaskResult[Any]: """Execute a standard task through the canonical primary API map.""" legacy_only_tasks = { @@ -4920,7 +5283,14 @@ async def execute_task(self, task_name: str, request: BaseModel) -> TaskResult[A raise ValueError(f"Unknown canonical AdCP task: {task_name}") return cast(TaskResult[Any], await method(request)) - async def execute_task_legacy(self, task_name: str, request: BaseModel) -> TaskResult[Any]: + @_task_options_method + async def execute_task_legacy( + self, + task_name: str, + request: BaseModel, + *, + options: TaskOptions | None = None, + ) -> TaskResult[Any]: """Execute an explicitly raw creative task for migration tooling.""" methods: dict[str, Callable[[Any], Any]] = { @@ -5432,8 +5802,7 @@ async def _handle_mcp_webhook( if preserve_legacy_identity: if authenticated_task_type not in _LEGACY_CREATIVE_TASKS: raise ValueError( - f"{authenticated_task_type} is not a legacy-only callback; use " - "handle_webhook()" + f"{authenticated_task_type} is not a legacy-only callback; use handle_webhook()" ) # Emit activity for monitoring diff --git a/src/adcp/exceptions.py b/src/adcp/exceptions.py index a10a41734..a410bfc7b 100644 --- a/src/adcp/exceptions.py +++ b/src/adcp/exceptions.py @@ -4,6 +4,8 @@ from typing import Any, TypedDict +from adcp.task_options import TaskRecoveryMetadata + class ADCPError(Exception): """Base exception for all AdCP client errors.""" @@ -86,12 +88,28 @@ def __init__( agent_id: str | None = None, agent_uri: str | None = None, timeout: float | None = None, + *, + task_name: str | None = None, + operation_id: str | None = None, + recovery: TaskRecoveryMetadata | None = None, ): """Initialize timeout error.""" + self.timeout = timeout + self.task_name = task_name + self.operation_id = operation_id + self.recovery = recovery suggestion = ( f"The request took longer than {timeout}s." if timeout else "The request timed out." ) - suggestion += "\n Try increasing the timeout value or check if the agent is overloaded." + if recovery is not None: + suggestion += ( + "\n The mutation may have succeeded. Retry the exact request with " + "the recovery idempotency_key; do not mint a new key." + ) + else: + suggestion += ( + "\n Try increasing the timeout value or check if the agent is overloaded." + ) super().__init__(message, agent_id, agent_uri, suggestion) @property diff --git a/src/adcp/protocols/a2a.py b/src/adcp/protocols/a2a.py index 6ddb8c527..ee90b45b9 100644 --- a/src/adcp/protocols/a2a.py +++ b/src/adcp/protocols/a2a.py @@ -24,6 +24,7 @@ from adcp.protocols._adcp_errors import validate_adcp_error from adcp.protocols.base import ProtocolAdapter from adcp.signing.autosign import current_operation as _signing_operation +from adcp.task_options import mark_task_dispatched from adcp.types.core import AgentConfig, DebugInfo, TaskResult, TaskStatus from adcp.validation.client_hooks import ( validate_incoming_response, @@ -467,6 +468,12 @@ async def _call_a2a_tool( signing_token = _signing_operation.set(tool_name) try: # Non-streaming send returns a single StreamResponse envelope. + mark_task_dispatched( + self.task_options_client_token, + tool_name, + mutating=_idempotency.is_mutating(tool_name), + idempotency_key=idempotency_key, + ) stream_event = await self._send_and_aggregate(a2a_client, request) payload_kind = stream_event.WhichOneof("payload") diff --git a/src/adcp/protocols/base.py b/src/adcp/protocols/base.py index a64ada865..9aabcd1c7 100644 --- a/src/adcp/protocols/base.py +++ b/src/adcp/protocols/base.py @@ -38,6 +38,8 @@ def __init__(self, agent_config: AgentConfig): # ``use_idempotency_key`` so a key pinned on one client does not bleed # to sibling clients (cross-seller correlation risk per AdCP #2315). self.idempotency_client_token: str | None = None + # Owning client identity for task-local deadline/recovery attribution. + self.task_options_client_token: object | None = None # Optional httpx request event hook. ADCPClient installs one when a # SigningConfig is present; the hook attaches RFC 9421 Signature-Input # / Signature / Content-Digest headers to outgoing requests that the diff --git a/src/adcp/protocols/mcp.py b/src/adcp/protocols/mcp.py index ea676d9c7..a6eb2720d 100644 --- a/src/adcp/protocols/mcp.py +++ b/src/adcp/protocols/mcp.py @@ -74,6 +74,7 @@ from adcp.protocols._adcp_errors import validate_adcp_error as _validate_adcp_error from adcp.protocols.base import ProtocolAdapter from adcp.signing.autosign import current_operation as _signing_operation +from adcp.task_options import mark_task_dispatched from adcp.types.core import DebugInfo, TaskResult, TaskStatus from adcp.validation.client_hooks import ( validate_incoming_response, @@ -642,6 +643,12 @@ async def _get_session(self) -> ClientSession: # Clean up the exit stack on failure to avoid resource leaks await self._cleanup_failed_connection("during connection attempt") + # A task-level deadline and caller cancellation both use + # CancelledError to abort discovery. Never reinterpret it + # as a failed URL probe or continue to a fallback URL. + if isinstance(e, asyncio.CancelledError): + raise + # If this isn't the last URL to try, create a new exit stack and continue if url != urls_to_try[-1]: logger.debug(f"Retrying with next URL after error: {last_error}") @@ -771,6 +778,12 @@ async def _call_mcp_tool(self, tool_name: str, params: dict[str, Any]) -> TaskRe signing_token = _signing_operation.set(tool_name) try: # Call the tool using MCP client session + mark_task_dispatched( + self.task_options_client_token, + tool_name, + mutating=_idempotency.is_mutating(tool_name), + idempotency_key=idempotency_key, + ) result = await session.call_tool(tool_name, params) finally: _signing_operation.reset(signing_token) diff --git a/src/adcp/task_options.py b/src/adcp/task_options.py new file mode 100644 index 000000000..a3ec9f79c --- /dev/null +++ b/src/adcp/task_options.py @@ -0,0 +1,133 @@ +"""Per-call execution options and deadline recovery metadata.""" + +from __future__ import annotations + +import asyncio +import math +from collections.abc import Awaitable, Callable +from contextvars import ContextVar, Token +from dataclasses import dataclass, field +from typing import TypeVar + +T = TypeVar("T") + + +@dataclass(frozen=True, slots=True) +class TaskOptions: + """Options for one complete SDK task call. + + ``timeout`` is a non-resetting wall-clock budget in seconds. It covers the + full client lifecycle: discovery, capability/version and signing preflight, + protocol dispatch, response validation, and postflight projection. It does + not replace the transport timeout configured on :class:`~adcp.AgentConfig`. + + A timeout of ``None`` disables the task-level deadline. + """ + + timeout: float | None = None + + def __post_init__(self) -> None: + timeout = self.timeout + if timeout is None: + return + if isinstance(timeout, bool) or not math.isfinite(timeout) or timeout <= 0: + raise ValueError("timeout must be a finite positive number of seconds") + + +@dataclass(frozen=True, slots=True) +class TaskRecoveryMetadata: + """Safe retry identity for a timed-out mutating task. + + ``outcome_unknown`` is always true: once dispatch begins, a deadline cannot + prove whether the seller committed the operation. Retry the exact original + request with ``idempotency_key``; do not mint a new key. + """ + + task_name: str + operation_id: str + idempotency_key: str = field(repr=False) + outcome_unknown: bool = True + + +@dataclass(slots=True) +class _TaskExecutionState: + client_token: object + operation_id: str + timeout: float | None + deadline: float | None + task_name: str + mutation_recovery: TaskRecoveryMetadata | None = None + + +_current_task_execution: ContextVar[_TaskExecutionState | None] = ContextVar( + "adcp_current_task_execution", default=None +) + + +def _set_task_execution(state: _TaskExecutionState) -> Token[_TaskExecutionState | None]: + return _current_task_execution.set(state) + + +def _reset_task_execution(token: Token[_TaskExecutionState | None]) -> None: + _current_task_execution.reset(token) + + +def _get_task_execution(client_token: object) -> _TaskExecutionState | None: + state = _current_task_execution.get() + if state is None or state.client_token is not client_token: + return None + return state + + +def mark_task_dispatched( + client_token: object | None, + task_name: str, + *, + mutating: bool, + idempotency_key: str | None, +) -> None: + """Record that a protocol call may now have reached the seller.""" + + state = _current_task_execution.get() + if state is None or state.client_token is not client_token: + return + if state.deadline is not None and asyncio.get_running_loop().time() >= state.deadline: + raise _TaskDeadlineExpiredError + if mutating and idempotency_key is not None and state.mutation_recovery is None: + state.mutation_recovery = TaskRecoveryMetadata( + task_name=task_name, + operation_id=state.operation_id, + idempotency_key=idempotency_key, + ) + + +_TIMEOUT_ERRORS: tuple[type[BaseException], ...] = (TimeoutError, asyncio.TimeoutError) + + +class _BodyTimeoutError(Exception): + """Distinguish an inner transport timeout from the task deadline.""" + + def __init__(self, cause: BaseException): + self.cause = cause + super().__init__(str(cause)) + + +class _TaskDeadlineExpiredError(BaseException): + """Raised only when the SDK-owned task deadline expires.""" + + +async def run_with_timeout(awaitable_factory: Callable[[], Awaitable[T]], timeout: float) -> T: + """Run under ``asyncio.wait_for`` without relabeling inner timeouts.""" + + async def run() -> T: + try: + return await awaitable_factory() + except _TIMEOUT_ERRORS as exc: + raise _BodyTimeoutError(exc) from exc + + try: + return await asyncio.wait_for(run(), timeout=timeout) + except _BodyTimeoutError as exc: + raise exc.cause from exc.cause.__cause__ + except _TIMEOUT_ERRORS as exc: + raise _TaskDeadlineExpiredError from exc diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 2e66c65a3..0b0c83102 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -460,6 +460,8 @@ "TEST_AGENT_MCP_NO_AUTH_CONFIG", "TEST_AGENT_TOKEN", "TargetingOverlay", + "TaskOptions", + "TaskRecoveryMetadata", "TaskResult", "TaskStatus", "TextContent", diff --git a/tests/test_task_options.py b/tests/test_task_options.py new file mode 100644 index 000000000..415ac0533 --- /dev/null +++ b/tests/test_task_options.py @@ -0,0 +1,442 @@ +"""Per-call TaskOptions deadline and mutation-recovery contracts.""" + +from __future__ import annotations + +import asyncio +import inspect +import time +import warnings +from typing import Any, get_type_hints +from unittest.mock import patch + +import pytest +from pydantic import BaseModel, TypeAdapter + +from adcp import ADCPClient, TaskOptions, TaskRecoveryMetadata +from adcp.exceptions import ADCPTimeoutError +from adcp.protocols.base import ProtocolAdapter +from adcp.protocols.mcp import MCPAdapter +from adcp.task_options import mark_task_dispatched +from adcp.types import GetProductsRequest +from adcp.types.core import ActivityType, AgentConfig, Protocol, TaskResult, TaskStatus + + +class _Request(BaseModel): + idempotency_key: str = "0123456789abcdef-task-options" + + +class _NoKeyRequest(BaseModel): + value: str = "test" + + +def _client(**kwargs: Any) -> ADCPClient: + return ADCPClient( + AgentConfig( + id=kwargs.pop("agent_id", "seller"), + agent_uri="https://seller.example/mcp", + protocol=Protocol.MCP, + timeout=17.0, + ), + **kwargs, + ) + + +def _get_products_request() -> GetProductsRequest: + return TypeAdapter(GetProductsRequest).validate_python( + {"buying_mode": "brief", "brief": "deadline test"} + ) + + +def _completed(data: dict[str, Any] | None = None) -> TaskResult[Any]: + return TaskResult[Any]( + status=TaskStatus.COMPLETED, + success=True, + data=data or {"products": []}, + ) + + +@pytest.mark.parametrize("timeout", [0, -1, float("inf"), float("nan"), True]) +def test_task_options_requires_finite_positive_timeout(timeout: Any) -> None: + with pytest.raises(ValueError, match="finite positive"): + TaskOptions(timeout=timeout) + + +def test_task_options_public_exports_are_immutable() -> None: + options = TaskOptions(timeout=1.0) + with pytest.raises((AttributeError, TypeError)): + options.timeout = 2.0 # type: ignore[misc] + assert TaskRecoveryMetadata.__module__ == "adcp.task_options" + assert get_type_hints(ADCPTimeoutError.__init__)["recovery"] == (TaskRecoveryMetadata | None) + + +@pytest.mark.asyncio +async def test_effectively_expired_deadline_does_not_leak_coroutine() -> None: + client = _client() + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + with pytest.raises(ADCPTimeoutError): + await client.get_products( + _get_products_request(), + options=TaskOptions(timeout=1e-15), + ) + + +@pytest.mark.asyncio +async def test_read_deadline_preserves_activity_operation_id() -> None: + activities = [] + client = _client(on_activity=activities.append) + + async def delayed(_params: dict[str, Any]) -> TaskResult[Any]: + await asyncio.sleep(1) + return _completed() + + with patch.object(client.adapter, "get_products", new=delayed): + with pytest.raises(ADCPTimeoutError) as caught: + await client.get_products(_get_products_request(), options=TaskOptions(timeout=0.01)) + + error = caught.value + request_activity = next(a for a in activities if a.type == ActivityType.PROTOCOL_REQUEST) + assert error.operation_id == request_activity.operation_id + assert error.task_name == "get_products" + assert error.recovery is None + assert client.agent_config.timeout == 17.0 + + +@pytest.mark.asyncio +async def test_inner_transport_timeout_is_not_relabelled() -> None: + client = _client() + inner = TimeoutError("transport read timed out") + + async def fail(_params: dict[str, Any]) -> TaskResult[Any]: + raise inner + + with patch.object(client.adapter, "get_products", new=fail): + with pytest.raises(TimeoutError) as caught: + await client.get_products(_get_products_request(), options=TaskOptions(timeout=1.0)) + assert caught.value is inner + + +@pytest.mark.asyncio +async def test_external_cancellation_is_not_relabelled() -> None: + client = _client() + entered = asyncio.Event() + + async def delayed(_params: dict[str, Any]) -> TaskResult[Any]: + entered.set() + await asyncio.sleep(10) + return _completed() + + with patch.object(client.adapter, "get_products", new=delayed): + task = asyncio.create_task( + client.get_products(_get_products_request(), options=TaskOptions(timeout=5.0)) + ) + await entered.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +@pytest.mark.asyncio +async def test_mutation_timeout_carries_secret_safe_recovery() -> None: + activities = [] + client = _client(on_activity=activities.append) + key = "0123456789abcdef-recovery-key" + + async def dispatched(_params: dict[str, Any]) -> TaskResult[Any]: + mark_task_dispatched( + client.adapter.task_options_client_token, + "build_creative", + mutating=True, + idempotency_key=key, + ) + await asyncio.sleep(1) + return _completed() + + with ( + warnings.catch_warnings(), + patch.object(client.adapter, "build_creative", new=dispatched), + ): + warnings.simplefilter("ignore", DeprecationWarning) + with pytest.raises(ADCPTimeoutError) as caught: + await client.build_creative_legacy( + _Request(idempotency_key=key), + options=TaskOptions(timeout=0.01), # type: ignore[arg-type] + ) + + error = caught.value + assert error.recovery is not None + assert error.recovery.idempotency_key == key + assert error.recovery.task_name == "build_creative" + request_activity = next(a for a in activities if a.type == ActivityType.PROTOCOL_REQUEST) + assert error.recovery.operation_id == request_activity.operation_id == error.operation_id + assert key not in str(error) + assert key not in repr(error) + assert key not in repr(error.recovery) + + +@pytest.mark.asyncio +async def test_real_mcp_dispatch_reports_the_exact_pinned_wire_key() -> None: + class BlockingSession: + sent: dict[str, Any] | None = None + + async def call_tool(self, _name: str, params: dict[str, Any]) -> Any: + self.sent = params + await asyncio.sleep(1) + + session = BlockingSession() + client = ADCPClient.from_mcp_client(session, agent_id="in-process") # type: ignore[arg-type] + client.adapter.request_validation_mode = "off" + key = "0123456789abcdef-pinned-wire-key" + + with warnings.catch_warnings(), client.use_idempotency_key(key): + warnings.simplefilter("ignore", DeprecationWarning) + with pytest.raises(ADCPTimeoutError) as caught: + await client.build_creative_legacy( # type: ignore[arg-type] + _NoKeyRequest(), options=TaskOptions(timeout=0.05) + ) + + assert session.sent is not None + assert session.sent["idempotency_key"] == key + assert caught.value.recovery is not None + assert caught.value.recovery.idempotency_key == key + + +@pytest.mark.asyncio +async def test_expired_synchronous_preflight_never_marks_dispatch() -> None: + client = _client() + dispatched = False + + async def blocked(_params: dict[str, Any]) -> TaskResult[Any]: + nonlocal dispatched + time.sleep(0.02) + mark_task_dispatched( + client.adapter.task_options_client_token, + "build_creative", + mutating=True, + idempotency_key="0123456789abcdef-late", + ) + dispatched = True + return _completed() + + with ( + warnings.catch_warnings(), + patch.object(client.adapter, "build_creative", new=blocked), + ): + warnings.simplefilter("ignore", DeprecationWarning) + with pytest.raises(ADCPTimeoutError) as caught: + await client.build_creative_legacy( # type: ignore[arg-type] + _Request(), options=TaskOptions(timeout=0.005) + ) + + assert dispatched is False + assert caught.value.recovery is None + + +@pytest.mark.asyncio +async def test_synchronous_postflight_overrun_is_rejected() -> None: + client = _client() + raw = _completed() + + async def immediate(_params: dict[str, Any]) -> TaskResult[Any]: + return raw + + def slow_postflight(_raw: TaskResult[Any]) -> TaskResult[Any]: + time.sleep(0.02) + return raw + + with ( + patch.object(client.adapter, "get_products", new=immediate), + patch.object(client, "_canonicalize_get_products_result", new=slow_postflight), + ): + with pytest.raises(ADCPTimeoutError): + await client.get_products(_get_products_request(), options=TaskOptions(timeout=0.005)) + + +@pytest.mark.asyncio +async def test_strict_preflight_cancellation_resets_verification_guard() -> None: + client = _client(strict_idempotency=True) + + async def delayed_capabilities() -> Any: + await asyncio.sleep(1) + + with ( + warnings.catch_warnings(), + patch.object(client, "fetch_capabilities", new=delayed_capabilities), + ): + warnings.simplefilter("ignore", DeprecationWarning) + with pytest.raises(ADCPTimeoutError): + await client.build_creative_legacy( # type: ignore[arg-type] + _Request(), options=TaskOptions(timeout=0.01) + ) + assert client._idempotency_capability_verified is False + + +@pytest.mark.asyncio +async def test_mcp_connection_cancellation_does_not_try_fallback_urls() -> None: + class CancelledStack: + enter_calls = 0 + closed = False + + async def enter_async_context(self, _context: Any) -> Any: + self.enter_calls += 1 + raise asyncio.CancelledError + + async def aclose(self) -> None: + self.closed = True + + client = _client() + assert isinstance(client.adapter, MCPAdapter) + stack = CancelledStack() + with ( + patch("adcp.protocols.mcp.AsyncExitStack", return_value=stack), + patch("adcp.protocols.mcp.streamablehttp_client", return_value=object()), + patch.object(client.adapter, "_streamable_http_client_factory", return_value=object()), + ): + with pytest.raises(asyncio.CancelledError): + await client.adapter._get_session() + + assert stack.enter_calls == 1 + assert stack.closed is True + + +@pytest.mark.asyncio +async def test_nested_generic_call_uses_one_deadline_and_root_name() -> None: + client = _client() + + async def delayed(_params: dict[str, Any]) -> TaskResult[Any]: + await asyncio.sleep(1) + return _completed() + + with ( + warnings.catch_warnings(), + patch.object(client.adapter, "build_creative", new=delayed), + ): + warnings.simplefilter("ignore", DeprecationWarning) + with pytest.raises(ADCPTimeoutError) as caught: + await client.execute_task_legacy( + "build_creative", + _Request(), + options=TaskOptions(timeout=0.01), + ) + assert caught.value.task_name == "build_creative" + + +@pytest.mark.asyncio +async def test_concurrent_deadlines_keep_recovery_keys_isolated() -> None: + client = _client() + + async def dispatched(params: dict[str, Any]) -> TaskResult[Any]: + mark_task_dispatched( + client.adapter.task_options_client_token, + "build_creative", + mutating=True, + idempotency_key=params["idempotency_key"], + ) + await asyncio.sleep(1) + return _completed() + + keys = ("0123456789abcdef-concurrent-a", "0123456789abcdef-concurrent-b") + with ( + warnings.catch_warnings(), + patch.object(client.adapter, "build_creative", new=dispatched), + ): + warnings.simplefilter("ignore", DeprecationWarning) + outcomes = await asyncio.gather( + *( + client.build_creative_legacy( # type: ignore[arg-type] + _Request(idempotency_key=key), + options=TaskOptions(timeout=0.01), + ) + for key in keys + ), + return_exceptions=True, + ) + + assert all(isinstance(outcome, ADCPTimeoutError) for outcome in outcomes) + recovered = { + outcome.recovery.idempotency_key + for outcome in outcomes + if isinstance(outcome, ADCPTimeoutError) and outcome.recovery is not None + } + assert recovered == set(keys) + + +@pytest.mark.asyncio +async def test_cross_client_dispatch_cannot_stamp_outer_recovery() -> None: + client_a = _client(agent_id="seller-a") + client_b = _client(agent_id="seller-b") + + async def b_dispatched(_params: dict[str, Any]) -> TaskResult[Any]: + mark_task_dispatched( + client_b.adapter.task_options_client_token, + "build_creative", + mutating=True, + idempotency_key="0123456789abcdef-seller-b", + ) + await asyncio.sleep(1) + return _completed() + + async def a_read(_params: dict[str, Any]) -> TaskResult[Any]: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + return await client_b.build_creative_legacy(_Request()) # type: ignore[arg-type] + + with ( + patch.object(client_a.adapter, "get_products", new=a_read), + patch.object(client_b.adapter, "build_creative", new=b_dispatched), + ): + with pytest.raises(ADCPTimeoutError) as caught: + await client_a.get_products(_get_products_request(), options=TaskOptions(timeout=0.01)) + assert caught.value.agent_id == "seller-a" + assert caught.value.recovery is None + + +def test_all_single_agent_protocol_tasks_expose_keyword_only_options() -> None: + exclusions = { + "close", + "get_agent_info", + "list_tools", + } + adapter_tasks = { + name + for name, member in inspect.getmembers(ProtocolAdapter, inspect.iscoroutinefunction) + if not name.startswith("_") and name not in exclusions + } + workflow_tasks = { + "execute_task", + "execute_task_legacy", + "refine_proposals_verified", + "wait_for_refinement_verified", + } + legacy_only = {"build_creative", "list_creative_formats", "preview_creative"} + methods = workflow_tasks | { + f"{name}_legacy" if name in legacy_only else name for name in adapter_tasks + } + # Canonical creative tasks retain explicit legacy escape hatches too. + methods |= { + "create_media_buy_legacy", + "get_creative_delivery_legacy", + "get_media_buy_delivery_legacy", + "get_media_buys_legacy", + "get_products_legacy", + "list_creatives_legacy", + "sync_creatives_legacy", + "update_media_buy_legacy", + } + + missing: list[str] = [] + for name in sorted(methods): + method = getattr(ADCPClient, name, None) + if method is None: + missing.append(name) + continue + parameter = inspect.signature(method).parameters.get("options") + if parameter is None or parameter.kind is not inspect.Parameter.KEYWORD_ONLY: + missing.append(name) + assert missing == [] + + +def test_multi_agent_client_does_not_claim_task_options_support() -> None: + from adcp import ADCPMultiAgentClient + + assert "options" not in inspect.signature(ADCPMultiAgentClient.get_products).parameters diff --git a/tests/type_checks/task_options.py b/tests/type_checks/task_options.py new file mode 100644 index 000000000..e5419e1c8 --- /dev/null +++ b/tests/type_checks/task_options.py @@ -0,0 +1,10 @@ +"""Adopter-facing type checks for per-call task options.""" + +from adcp import ADCPClient, TaskOptions +from adcp.types import GetProductsRequest + + +async def call_with_deadline(client: ADCPClient, request: GetProductsRequest) -> None: + options = TaskOptions(timeout=10.0) + await client.get_products(request, options=options) + await client.execute_task("get_products", request, options=options)