diff --git a/MIGRATION_v9_to_v10.md b/MIGRATION_v9_to_v10.md index 00f3ca8dd..ffe977751 100644 --- a/MIGRATION_v9_to_v10.md +++ b/MIGRATION_v9_to_v10.md @@ -18,6 +18,7 @@ the SDK-relevant inventory is [docs/adcp-3.3-sdk-inventory.md](docs/adcp-3.3-sdk | `list_proposals` and `list_collections` are served from declarations | The experimental 3.3 reads were already on `ADCPClient`, `ADCPHandler`, the adapters and the CLI. `PlatformHandler` now serves them: implement `list_proposals` (`OpenOpportunitiesPlatform`; declare it in `media_buy.lifecycle_tools` or `media_buy.open_opportunities.list_proposals`) or `list_collections` (`CollectionCatalogPlatform`; declare `media_buy.collection_catalog` and `media_buy.collection_cards` in `experimental_features`). A declaration with no method behind it now fails `validate_platform` at boot, and a method with no declaration is not advertised. `adcp.decisioning.capabilities` exports `OpenOpportunities` and `LaterProposals`. | | A task the caller's release does not define fails with `UNSUPPORTED_FEATURE` | A server asked for a task its caller's negotiated release lacks (`list_proposals` or `list_collections` from a 3.2 buyer, `list_products` from a 3.0 buyer) used to answer `VERSION_UNSUPPORTED`; it now answers `UNSUPPORTED_FEATURE` with `details: {"task", "adcp_version"}`, as the spec requires for `list_collections`. The caller's version is supported, so re-pinning is not the fix. Buyers matching on `VERSION_UNSUPPORTED` for this case should match `UNSUPPORTED_FEATURE`. The SDK client already refused such a call before transport with `UNSUPPORTED_FEATURE`. | | Agent-level notification configs classify the 3.3 account-anchored types | `sync_agent_notification_configs` / `sync_principal` now require `all_authorized_accounts: true` for `proposal.created`, `proposal.offered`, `collection_card.updated`, `collection_card.removed`, `collection_catalog.bulk_change` and `creative.staged_update_changed`, like the other account-anchored types; a registration without it is rejected. `WebhookSender.prepare_proposal_created` / `send_proposal_created` build the schema-validated `proposal.created` payload. | +| `expected_confirmation_at` follows the 3.3 send rules | Sellers set the AdCP 3.3 confirmation estimate with `adcp.server.submitted_response(task_id, expected_confirmation_at=...)` or `ctx.handoff_to_task(fn, expected_confirmation_at=...)` / `handoff_to_workflow(...)`; a naive datetime is refused. Version conversion now drops it for a caller below 3.3 on `buy_products`, `accept_proposal` and `refine_proposals` (earlier SDK 10 pre-releases passed it through on the first two's submitted envelope) and keeps it on `create_media_buy` and `get_media_buys` for 3.0 to 3.2 callers (it used to be dropped). Buyers read `TaskResult.expected_confirmation_at` and schedule the next check with `adcp.next_status_check_delay`; `wait_for_refinement_verified` does so itself, capped by the new `max_poll_interval` (default 60 seconds). | | The reporting-subscription surface is served from 3.3.0-beta.3 on, unprojected | The beta.1 / beta.2 field projections are gone, with the preview module that patched the fields onto the models: `project_offering`, `project_reporting_delivery`, `project_subscription_state`, `project_subscription_coverage` and `serves_settled_subscription_fields` are removed from `adcp.reporting.subscriptions`, and `ReportingSubscriptionEngine.subscription_coverage()` no longer takes `adcp_version`. One predicate, `adcp.reporting.subscriptions.serves_subscriptions(adcp_version)` (from `SUBSCRIPTIONS_MIN_ADCP_VERSION`, `3.3.0-beta.3`), gates the whole surface: capabilities, `get_reporting_status` coverage (including `status: "inactive"` with `deactivated_at`) and the principal readback. A caller served an earlier release gets none of it. The extensible `reporting-dimension` vocabulary accepts unknown values on selectors, offerings and the capability block. | | Reporting-subscription registration follows the 3.3.0-beta.3 rules | A caller may hold several pacing or analytics subscriptions per account and purpose (the lowest `subscription_id` owes a shared offering; the others record a package-level `overridden` exception), while billing stays one per account at a scope level and must use `match: exact` with `dimensions`. Membership edits (`scope.account_ids`, `exclude_account_ids`) no longer need a new `subscription_version`, and `client.reporting.subscribe` no longer bumps the version for them. `destination_overrides` must name existing destinations. The readback lists 16 prior versions and 1,000 accounts inline with `*_count` totals; page the rest with `get_principal` `reporting_subscription_page`. A partial grant refuses an unlisted reporting task with `SCOPE_INSUFFICIENT`. `AccountReady` gains `resolution`, `billing`, `billing_entity`, `payment_terms`, `effective_at` and `not_carried_over`. | | A server that restricts `supported_versions` runs on its newest listed release | Any server (`create_mcp_server`, `create_a2a_server`, `serve`, `MCPToolSet`, `ADCPAgentExecutor`, `create_tool_caller`, `PlatformHandler`, `create_adcp_server_from_platform`) with `supported_versions=["3.2"]` and no `adcp_version` runs its handlers on 3.2, instead of raising because the SDK default is outside the list. Pass `adcp_version` explicitly to choose another listed release. | diff --git a/docs/adcp-3.3-sdk-inventory.md b/docs/adcp-3.3-sdk-inventory.md index bdee49f4d..97920e74c 100644 --- a/docs/adcp-3.3-sdk-inventory.md +++ b/docs/adcp-3.3-sdk-inventory.md @@ -39,7 +39,7 @@ it adopted the bundle, and what is left to the codegen/wire-fidelity (2) and | `list_proposals` | `media-buy/list-proposals-request.json`, `-response.json` | Experimental `media_buy.open_opportunities`: snapshots beside `disposition`, `origin`, `later_reason` / `offer_reason`, plus `opportunities[]`; filters `opportunity_ids`, `media_buy_ids`, `proposal_ids`, `dispositions`. Added to `LifecycleTool` and to `account-change.json` `Task`. | done (#1645): client surface as above; `OpenOpportunitiesPlatform` hook, advertised and boot-validated from `lifecycle_tools` or `open_opportunities.list_proposals`. Serving it from a `ProposalStore` is deferred: the store keeps legacy `get_products` proposals, not canonical snapshots with dispositions, and has no account listing | | `get_property_list` over MCP | existing schema | Property list references resolve through AdCP tasks at `agent_url`, `auth_token` as Bearer, every page fetched and validated before a cached snapshot is replaced. | done: `property_lists` resolves over MCP with the list-scoped Bearer token and fetches every page before replacing a snapshot (verified by 3) | | `proposal.created` webhook | `core/proposal-created-webhook.json`; `NotificationType` / `EventType` `proposal_created` | Seller-initiated proposals under `open_opportunities`. | done (#1645): `WebhookSender.prepare_proposal_created` / `send_proposal_created` (schema-validated, `notification_id` bound to `proposal_id`); agent-level registrations treat it, `proposal.offered`, `collection_card.*`, `collection_catalog.bulk_change` and `creative.staged_update_changed` as account-anchored | -| `refine_proposals` submitted response | `core/async_response_refs/media_buy/refine_proposals_async_response_submitted.json` | New async `submitted` envelope for refinement with `expected_confirmation_at`. | types generated; async response dispatch table: (3), #1640 | +| `refine_proposals` submitted response | `core/async_response_refs/media_buy/refine_proposals_async_response_submitted.json` | New async `submitted` envelope for refinement with `expected_confirmation_at`. | done (#1640): parsed into `RefineProposalsResponse`; buyer and seller handling as in the `expected_confirmation_at` row | ## 3. Changed tasks and request/response fields @@ -47,7 +47,7 @@ it adopted the bundle, and what is left to the codegen/wire-fidelity (2) and | Field | Where | Rule | SDK status | | --- | --- | --- | --- | -| `expected_confirmation_at` | `submitted` envelopes of `create_media_buy`, `buy_products`, `accept_proposal`, `refine_proposals`; `MediaBuy` on `get_media_buys` | Estimate. Send on `accept_proposal`, `buy_products`, `refine_proposals` only to 3.3 peers (safe on `create_media_buy`). | generated; buyer status scheduling and downgrade drop rule: (3), #1640 | +| `expected_confirmation_at` | `submitted` envelopes of `create_media_buy`, `buy_products`, `accept_proposal`, `refine_proposals`; `MediaBuy` on `get_media_buys` | Estimate. Send on `accept_proposal`, `buy_products`, `refine_proposals` only to 3.3 peers (safe on `create_media_buy`). | done (#1640): `TaskResult.expected_confirmation_at`, `adcp.next_status_check_delay` (floor `poll_interval`, ceiling `max_interval`), `wait_for_refinement_verified(max_poll_interval=...)`, `CompatibilityPurchaseOperation.next_refresh_delay`; sellers use `submitted_response(..., expected_confirmation_at=...)` or `ctx.handoff_to_task` / `handoff_to_workflow(..., expected_confirmation_at=...)`. Conversion drops it below 3.3 on the three compact tasks (recorded loss) and keeps it on `create_media_buy` / `get_media_buys` | | `predecessor_media_buy_id`, `lineage_reason` | `CreateMediaBuyRequest`, `BuyProductsRequest`, `AcceptProposalRequest`, `MediaBuy`, `GetMediaBuysRequest` filter; enum `media-buy-lineage-reason` | Experimental lineage. | generated | | ~~`commercial_terms.definition_pins[]`~~ (withdrawn in 3.3.0-beta.2, section 12) | `media-buy/commercial-terms.json` via `core/definition-pin.json`; enum `definition-pin-ref-kind` | Required for callers negotiating 3.3. Drift returns `unable` with `definition_changed` and `drifted_pins[]` (`refine_proposals` `Results*`), or `REFERENCE_DEFINITION_CHANGED` before `accept_proposal` mutates. Never sent to callers below 3.3. | generated; pin minting, drift detection, downgrade stripping: not built, #1641 closed as withdrawn (moved to 4.0, adcontextprotocol/adcp#8537) | | `criteria.outcome_target.goal` `kind: "vendor_metric"` | `media-buy/outcome-target.json` `Goal2`; `vendor_metric_values[]`; capability `VendorMetricOptimization.supported_metrics` | Buyer sends only to 3.3 peers. | generated; gate: (3) | diff --git a/docs/cross-version-conversion.md b/docs/cross-version-conversion.md index 8071db16f..82a2332bc 100644 --- a/docs/cross-version-conversion.md +++ b/docs/cross-version-conversion.md @@ -235,6 +235,8 @@ Some 3.3 additions need more than dropping the field. These rules run on the | `age_determination_bases` | Moved to `ext.adcp.age_determination_bases`, replacing any interim value there | `moved_to_ext_adcp` | | `price_breakdown` with an `operator_fee` adjustment | Response: breakdown omitted, since the rest no longer sums to `fixed_price`; request: error | `enum_unavailable` on the kind, `unrepresentable_in_target` on the breakdown | | `pixel_tracker` `event` `start`, `first_quartile`, `midpoint`, `third_quartile`, `complete` | Rewritten to `event: custom` with `custom_event_name` set to the 3.3 value, the 3.2 interim form; never an impression tracker. Upgrading keeps the `custom` form | `playback_event_as_custom` | +| `expected_confirmation_at` on `buy_products`, `accept_proposal`, `refine_proposals` | Dropped from every arm, submitted included, even where the 3.2 async schema is open: the spec sends it only to 3.3 callers | `field_unavailable` | +| `expected_confirmation_at` on `create_media_buy`'s submitted envelope and on `get_media_buys` buys | Kept for 3.2, 3.1 and 3.0, whose objects are open: the spec makes it safe for any release to ignore | none | | New `error.code` values (`RIGHTS_CHANNEL_VIOLATION`, `ACCOUNT_CHANGE_REQUIRES_NEW_ACCOUNT`, ...) | Kept: 3.2 `error.code` is open, and receivers classify it by `recovery` | none | | Other new enum values | Dropped when optional, refused when required (see above) | `enum_unavailable` | diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py index 9c78bd644..8142eede1 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -100,6 +100,10 @@ def _resolve_version() -> str: "validate_identifier", "validate_publisher_domain", ), + "adcp.status_checks": ( + "expected_confirmation_at", + "next_status_check_delay", + ), "adcp.property_lists": ( "MCPPropertyListFetcher", "ResolvedPropertyList", @@ -1043,6 +1047,8 @@ def get_adcp_version() -> str: "normalize_identifier", "validate_identifier", "validate_publisher_domain", + "expected_confirmation_at", + "next_status_check_delay", "MCPPropertyListFetcher", "ProductPropertyAssessment", "ProductPropertyPolicy", @@ -2062,6 +2068,7 @@ def get_adcp_version() -> str: ReportingResourceReader, ReportingTrustedOriginPolicy, ) + from adcp.status_checks import expected_confirmation_at, next_status_check_delay from adcp.substitution import ( MacroMapping, MacroMappingEntry, diff --git a/src/adcp/client.py b/src/adcp/client.py index ec0c9a546..5915037ae 100644 --- a/src/adcp/client.py +++ b/src/adcp/client.py @@ -111,6 +111,7 @@ ) from adcp.signing.negotiated_profiles import NegotiatedSigningProfiles from adcp.signing.signer import sign_request +from adcp.status_checks import expected_confirmation_at, next_status_check_delay from adcp.task_options import ( TaskOptions, _get_task_execution, @@ -2752,16 +2753,26 @@ async def wait_for_refinement_verified( source_proposals: Mapping[str, BaseModel | Mapping[str, Any]] | None = None, timeout: float = 300.0, poll_interval: float = 1.0, + max_poll_interval: float = 60.0, now: datetime | None = None, options: TaskOptions | None = None, ) -> VerifiedRefinementResult: - """Poll a submitted refinement and verify its exact terminal payload.""" + """Poll a submitted refinement and verify its exact terminal payload. + + When the submitted response carries ``expected_confirmation_at`` + (AdCP 3.3), the next check is scheduled for that time, bounded by + ``poll_interval`` (floor) and ``max_poll_interval`` (ceiling); after it + passes, or without one, checks run every ``poll_interval`` seconds. + """ if initial.valid or not initial.pending: return initial - if timeout <= 0 or poll_interval <= 0: - raise ValueError("timeout and poll_interval must be positive") + if timeout <= 0 or poll_interval <= 0 or max_poll_interval <= 0: + raise ValueError("timeout, poll_interval and max_poll_interval must be positive") raw = (initial.task_result.metadata or {}).get(WIRE_RESPONSE_METADATA_KEY) + # The seller's estimate rides on the submitted response; task status + # polls do not repeat it, so it schedules every check until it passes. + estimate = initial.task_result.expected_confirmation_at or expected_confirmation_at(raw) task_id = raw.get("task_id") if isinstance(raw, Mapping) else None if not isinstance(task_id, str): data = initial.task_result.data @@ -2808,7 +2819,12 @@ async def wait_for_refinement_verified( remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError(f"refinement task {task_id!r} did not complete in time") - await asyncio.sleep(min(poll_interval, remaining)) + delay = next_status_check_delay( + estimate, + poll_interval=poll_interval, + max_interval=max(max_poll_interval, poll_interval), + ) + await asyncio.sleep(min(delay, remaining)) @_task_options_method async def decline_proposals( diff --git a/src/adcp/compat/purchase_continuation.py b/src/adcp/compat/purchase_continuation.py index 79642181d..74dd7d2f0 100644 --- a/src/adcp/compat/purchase_continuation.py +++ b/src/adcp/compat/purchase_continuation.py @@ -250,6 +250,38 @@ class CompatibilityPurchaseOperation: reserved_result_bytes: int = 0 result: JsonObject | None = None + @property + def expected_confirmation_at(self) -> datetime | None: + """The seller's confirmation estimate on a pending result (AdCP 3.3), if any.""" + if self.state != CompatibilityOperationState.PENDING or self.result is None: + return None + from adcp.status_checks import expected_confirmation_at + + return expected_confirmation_at(self.result) + + def next_refresh_delay( + self, + *, + poll_interval: float, + max_interval: float, + now: datetime | None = None, + ) -> float: + """Seconds before calling ``refresh_pending_legacy_purchase`` again. + + Scheduled from :attr:`expected_confirmation_at` when the seller sent + one, bounded by ``poll_interval`` (floor) and ``max_interval`` + (ceiling); otherwise ``poll_interval``. The estimate is never a + failure deadline. + """ + from adcp.status_checks import next_status_check_delay + + return next_status_check_delay( + self.expected_confirmation_at, + poll_interval=poll_interval, + max_interval=max_interval, + now=now, + ) + @dataclass(frozen=True) class LegacyPurchaseExecution: diff --git a/src/adcp/compat/versions.py b/src/adcp/compat/versions.py index 7f9f95c7d..d63425f92 100644 --- a/src/adcp/compat/versions.py +++ b/src/adcp/compat/versions.py @@ -70,6 +70,19 @@ _PLAYBACK_PIXEL_EVENTS = frozenset( {"start", "first_quartile", "midpoint", "third_quartile", "complete"} ) +#: Newer response fields the spec lets a seller send to older peers whose +#: object is open: ``expected_confirmation_at`` on ``create_media_buy``'s +#: submitted envelope and on ``get_media_buys`` buys is an estimate any +#: release can ignore. Where the older object is closed it is dropped as usual. +_OLDER_PEER_SAFE_FIELDS = frozenset({"expected_confirmation_at"}) +#: Top-level response fields a task sends only to peers on the 3.3 line +#: (AdCP 3.3: ``expected_confirmation_at`` on ``buy_products``, +#: ``accept_proposal`` and ``refine_proposals``). Each async schema may not +#: declare the field, so the drop is explicit rather than left to projection. +_LATEST_LINE_ONLY_FIELDS: dict[str, tuple[str, ...]] = { + tool: ("expected_confirmation_at",) + for tool in ("buy_products", "accept_proposal", "refine_proposals") +} @dataclass(frozen=True) @@ -610,6 +623,9 @@ def _project( target_version=target_version, reason="the older contract cannot express an explicit null clear", ) + if key in _OLDER_PEER_SAFE_FIELDS and response and _is_open(target): + projected[key] = child + continue # Schema maps (e.g. asset IDs) are application-owned keys. additional = target.get("additionalProperties") if isinstance(additional, dict) or target.get("patternProperties"): @@ -1248,6 +1264,11 @@ def _convert_payload( result.setdefault("cache_scope", scope) if direction != "request": _convert_compact_status(tool, result, source, target, step, losses) + if step < 0 and minor == 3: + for name in _LATEST_LINE_ONLY_FIELDS.get(tool, ()): + if name in result: + del result[name] + losses.append(ConversionLoss(_pointer("", name), source, target)) if step > 0 and _carries_interim_ext(result): _hoist_interim_ext(result, target_schema, target_schema) if step < 0: diff --git a/src/adcp/decisioning/context.py b/src/adcp/decisioning/context.py index c1eadfb79..082b852e4 100644 --- a/src/adcp/decisioning/context.py +++ b/src/adcp/decisioning/context.py @@ -609,6 +609,8 @@ class RequestContext(ToolContext, Generic[TMeta]): def handoff_to_task( self, fn: Callable[[Any], Awaitable[T] | T], + *, + expected_confirmation_at: datetime | str | None = None, ) -> TaskHandoff[T]: """Promote this call to a background task. @@ -635,12 +637,23 @@ def handoff_to_task( — use :meth:`handoff_to_workflow` instead. The split is purely about where the work runs (in-process / framework-managed vs. adopter-owned external system). + + ``expected_confirmation_at`` (AdCP 3.3) is the seller's estimate of + when it will confirm, for ``create_media_buy``, ``buy_products``, + ``accept_proposal`` and finalize-only or extend_hold-only + ``refine_proposals``. It rides on the Submitted envelope; see + :func:`adcp.server.responses.submitted_response` for the version + rules. """ - return TaskHandoff(fn) + return TaskHandoff( + fn, expected_confirmation_at=_confirmation_estimate(expected_confirmation_at) + ) def handoff_to_workflow( self, fn: Callable[[Any], Awaitable[None] | None], + *, + expected_confirmation_at: datetime | str | None = None, ) -> WorkflowHandoff: """Promote this call to an externally-completed task. @@ -708,6 +721,18 @@ async def on_decision(self, task_id, decision): ) See :class:`adcp.decisioning.WorkflowHandoff` for the full - semantics. + semantics. ``expected_confirmation_at`` is as on + :meth:`handoff_to_task`. """ - return WorkflowHandoff(fn) + return WorkflowHandoff( + fn, expected_confirmation_at=_confirmation_estimate(expected_confirmation_at) + ) + + +def _confirmation_estimate(value: datetime | str | None) -> str | None: + """Validate an ``expected_confirmation_at`` estimate at the handoff site.""" + if value is None: + return None + from adcp.server.responses import format_expected_confirmation_at + + return format_expected_confirmation_at(value) diff --git a/src/adcp/decisioning/dispatch.py b/src/adcp/decisioning/dispatch.py index 43ebbcffa..905a471f2 100644 --- a/src/adcp/decisioning/dispatch.py +++ b/src/adcp/decisioning/dispatch.py @@ -2483,10 +2483,7 @@ async def _run() -> None: # deliberately NOT on the wire — it lives on TaskRecord for # ``tasks/get`` reads only, since the Python method name leaking to # buyers would couple the wire to handler-internal naming. - return _FrameworkSubmittedProjection( - task_id=task_id, - status="submitted", - ) + return _submitted_projection(task_id, method_name, handoff._expected_confirmation_at) #: Strong-ref the in-flight handoff tasks so the asyncio loop's @@ -2606,10 +2603,33 @@ async def _project_workflow_handoff( # intentional. ``task_type`` lives on the registry row (for # ``tasks/get``), not on the wire envelope, per the same posture # as :func:`_project_handoff`. - return _FrameworkSubmittedProjection( - task_id=task_id, - status="submitted", - ) + return _submitted_projection(task_id, method_name, handoff._expected_confirmation_at) + + +def _submitted_projection( + task_id: str, method_name: str, expected_confirmation_at: str | None +) -> _FrameworkSubmittedProjection: + """The wire Submitted envelope, with the AdCP 3.3 confirmation estimate. + + The estimate is defined only on the tasks in + :data:`adcp.server.responses.EXPECTED_CONFIRMATION_TASKS`; on any other + task it is dropped with a warning rather than sent off-schema. The + dispatcher's version conversion drops it for older callers where the + spec requires. + """ + from adcp.server.responses import EXPECTED_CONFIRMATION_TASKS + + projection = _FrameworkSubmittedProjection(task_id=task_id, status="submitted") + if expected_confirmation_at is not None: + if method_name in EXPECTED_CONFIRMATION_TASKS: + projection["expected_confirmation_at"] = expected_confirmation_at + else: + logger.warning( + "expected_confirmation_at is not defined for %s; omitted from the " + "submitted envelope", + method_name, + ) + return projection __all__ = [ diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index 6ca71b65b..a1b86a942 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -64,6 +64,7 @@ _internal_error_details, _internal_error_message, _invoke_platform_method, + _is_framework_submitted_projection, ) from adcp.decisioning.domain_lists import bound_capability_domain_lists from adcp.decisioning.experimental_reads import ( @@ -1994,7 +1995,7 @@ def _maybe_auto_emit_sync_completion( "submitted"`` predicate so an adopter who legitimately returns a sync ``{"status": "submitted", ...}`` remains an inline result. """ - if ( + if _is_framework_submitted_projection(result) or ( isinstance(result, dict) and set(result.keys()) == {"task_id", "status"} and result.get("status") == "submitted" diff --git a/src/adcp/decisioning/types.py b/src/adcp/decisioning/types.py index 0b164732c..dfc864a46 100644 --- a/src/adcp/decisioning/types.py +++ b/src/adcp/decisioning/types.py @@ -325,15 +325,22 @@ def create_media_buy(self, req, ctx): where the work runs (in-process / framework-managed / adopter-owned). """ - __slots__ = ("_fn",) + __slots__ = ("_fn", "_expected_confirmation_at") - def __init__(self, fn: Callable[[Any], Awaitable[T] | T]) -> None: + def __init__( + self, + fn: Callable[[Any], Awaitable[T] | T], + *, + expected_confirmation_at: str | None = None, + ) -> None: # ``fn`` is ``Callable[[TaskHandoffContext], Awaitable[T] | T]`` # but TaskHandoffContext lives in dispatch.py to avoid a cycle. # The framework calls ``handoff._fn(task_ctx)`` at dispatch time; # adopters pass either a coroutine function or a sync callable # and the dispatcher detects via ``inspect.iscoroutine``. self._fn = fn + # RFC 3339 estimate projected onto the Submitted envelope (AdCP 3.3). + self._expected_confirmation_at = expected_confirmation_at def __repr__(self) -> str: return "TaskHandoff()" @@ -421,9 +428,14 @@ def _enqueue(self, task_ctx, req): registry-side only. """ - __slots__ = ("_fn",) + __slots__ = ("_fn", "_expected_confirmation_at") - def __init__(self, fn: Callable[[Any], Awaitable[None] | None]) -> None: + def __init__( + self, + fn: Callable[[Any], Awaitable[None] | None], + *, + expected_confirmation_at: str | None = None, + ) -> None: # ``fn`` is ``Callable[[TaskHandoffContext], Awaitable[None] | # None]`` — the framework calls it once synchronously (or # awaits it if a coroutine) at handoff time. Return value @@ -431,6 +443,8 @@ def __init__(self, fn: Callable[[Any], Awaitable[None] | None]) -> None: # ``registry.complete()`` later, NOT via fn return. # TaskHandoffContext lives in task_registry.py to avoid a cycle. self._fn = fn + # RFC 3339 estimate projected onto the Submitted envelope (AdCP 3.3). + self._expected_confirmation_at = expected_confirmation_at def __repr__(self) -> str: return "WorkflowHandoff()" diff --git a/src/adcp/server/__init__.py b/src/adcp/server/__init__.py index ed0508e3f..6cfa1ddb0 100644 --- a/src/adcp/server/__init__.py +++ b/src/adcp/server/__init__.py @@ -176,6 +176,7 @@ async def get_products(params, context=None): preview_creative_response, products_response, signals_response, + submitted_response, sync_accounts_response, sync_catalogs_response, sync_creatives_response, @@ -378,6 +379,7 @@ async def get_products(params, context=None): "preview_creative_response", "products_response", "signals_response", + "submitted_response", "sync_accounts_response", "sync_catalogs_response", "sync_creatives_response", diff --git a/src/adcp/server/responses.py b/src/adcp/server/responses.py index 5c689bd81..827a3ac8c 100644 --- a/src/adcp/server/responses.py +++ b/src/adcp/server/responses.py @@ -913,6 +913,77 @@ def sync_catalogs_response( } +# ============================================================================ +# Submitted tasks +# ============================================================================ + +#: Tasks whose AdCP 3.3 ``submitted`` envelope carries ``expected_confirmation_at``. +#: ``refine_proposals`` carries it only for a finalize-only or extend_hold-only +#: request; sellers SHOULD NOT send it on revise requests. +EXPECTED_CONFIRMATION_TASKS: frozenset[str] = frozenset( + {"create_media_buy", "buy_products", "accept_proposal", "refine_proposals"} +) + + +def format_expected_confirmation_at(value: datetime | str) -> str: + """Normalize an ``expected_confirmation_at`` estimate to an RFC 3339 UTC string. + + Accepts a timezone-aware :class:`~datetime.datetime` or an RFC 3339 + string with an offset. A naive datetime or an offset-less string is + refused: the estimate must name an instant. A time in the past is + allowed (the spec says sellers SHOULD NOT send one) and logged. + """ + if isinstance(value, str): + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError("expected_confirmation_at must be an RFC 3339 date-time") from exc + elif isinstance(value, datetime): + parsed = value + else: + raise TypeError("expected_confirmation_at must be a datetime or an RFC 3339 string") + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise ValueError("expected_confirmation_at must carry a timezone") + parsed = parsed.astimezone(timezone.utc) + if parsed < datetime.now(timezone.utc): + _logger.warning("expected_confirmation_at %s is in the past", parsed.isoformat()) + return parsed.isoformat().replace("+00:00", "Z") + + +def submitted_response( + task_id: str, + *, + expected_confirmation_at: datetime | str | None = None, + message: str | None = None, +) -> dict[str, Any]: + """Build a ``submitted`` task envelope. + + ``expected_confirmation_at`` (AdCP 3.3) is the seller's estimate of when + it will confirm, for confirmation that waits on a scheduled event such as + a booking round or a review window. It is valid on ``create_media_buy``, + ``buy_products``, ``accept_proposal`` and finalize-only or + extend_hold-only ``refine_proposals``. Sellers SHOULD set it when they + know that event. The dispatcher drops it for a ``buy_products``, + ``accept_proposal`` or ``refine_proposals`` caller below 3.3, recording a + conversion loss; ``create_media_buy`` keeps it for every release, whose + submitted envelope is open. + + Decisioning platforms pass the same value to + ``ctx.handoff_to_task(fn, expected_confirmation_at=...)`` or + ``ctx.handoff_to_workflow(...)`` instead. + """ + if not task_id: + raise ValueError("task_id must be non-empty") + response: dict[str, Any] = {"status": "submitted", "task_id": task_id} + if expected_confirmation_at is not None: + response["expected_confirmation_at"] = format_expected_confirmation_at( + expected_confirmation_at + ) + if message is not None: + response["message"] = message + return response + + # ============================================================================ # Generic Helpers # ============================================================================ diff --git a/src/adcp/status_checks.py b/src/adcp/status_checks.py new file mode 100644 index 000000000..db252e98a --- /dev/null +++ b/src/adcp/status_checks.py @@ -0,0 +1,87 @@ +"""Schedule status checks from a seller's ``expected_confirmation_at`` (AdCP 3.3). + +A seller whose confirmation waits on a scheduled event (a booking round, a +review window) SHOULD set ``expected_confirmation_at`` on the ``submitted`` +envelope of ``create_media_buy``, ``buy_products``, ``accept_proposal`` and +finalize-only ``refine_proposals``, and on a ``get_media_buys`` buy whose +``confirmed_at`` is still null. Buyers SHOULD check status at that time rather +than polling tightly before it. + +It is an estimate, not a deadline: the seller may confirm earlier or later, +the task status stays authoritative, and a passed timestamp is never a +failure. :func:`next_status_check_delay` therefore bounds the wait between a +floor (the normal polling interval, also used once the estimate has passed) +and a ceiling, so an early confirmation is still noticed. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime, timezone +from typing import Any + +__all__ = ["expected_confirmation_at", "next_status_check_delay"] + + +def _parse(value: Any) -> datetime | None: + if isinstance(value, datetime): + parsed = value + elif isinstance(value, str): + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + else: + return None + if parsed.tzinfo is None or parsed.utcoffset() is None: + return None + return parsed + + +def expected_confirmation_at(source: Any) -> datetime | None: + """Return the seller's confirmation estimate carried by ``source``, if any. + + ``source`` is a :class:`~adcp.types.core.TaskResult`, a response model, a + ``get_media_buys`` buy, or a wire mapping. Returns a timezone-aware + datetime, or ``None`` when the field is absent or not an RFC 3339 instant. + """ + from adcp.types.core import TaskResult + + if isinstance(source, TaskResult): + found = expected_confirmation_at(source.data) + if found is None and source.metadata: + found = _parse(source.metadata.get("expected_confirmation_at")) + return found + if isinstance(source, Mapping): + return _parse(source.get("expected_confirmation_at")) + root = getattr(source, "root", None) + if root is not None and root is not source: + return expected_confirmation_at(root) + return _parse(getattr(source, "expected_confirmation_at", None)) + + +def next_status_check_delay( + estimate: datetime | str | None, + *, + poll_interval: float, + max_interval: float, + now: datetime | None = None, +) -> float: + """Seconds to wait before the next status check. + + Without an estimate, or once it has passed, this is ``poll_interval``. With + an estimate in the future it is the time until the estimate, never less + than ``poll_interval`` (the floor) and never more than ``max_interval`` + (the ceiling), so a seller that confirms early is still seen within + ``max_interval``. + """ + if poll_interval <= 0 or max_interval <= 0: + raise ValueError("poll_interval and max_interval must be positive") + if max_interval < poll_interval: + raise ValueError("max_interval must be at least poll_interval") + when = _parse(estimate) + if when is None: + return poll_interval + current = now if now is not None else datetime.now(timezone.utc) + until = (when - current).total_seconds() + return min(max(until, poll_interval), max_interval) diff --git a/src/adcp/types/core.py b/src/adcp/types/core.py index a33d257f2..1f3ea7019 100644 --- a/src/adcp/types/core.py +++ b/src/adcp/types/core.py @@ -3,6 +3,7 @@ from __future__ import annotations import ipaddress +from datetime import datetime from enum import Enum from typing import Any, Generic, Literal, TypeVar from urllib.parse import urlparse @@ -271,6 +272,21 @@ class TaskResult(BaseModel, Generic[T]): # ``adcp.exceptions.is_pre_dispatch_error``. dispatched: Literal[False] | None = None + @property + def expected_confirmation_at(self) -> datetime | None: + """The seller's estimate of when a submitted task confirms (AdCP 3.3). + + Read from the response (``create_media_buy``, ``buy_products``, + ``accept_proposal`` and finalize-only ``refine_proposals`` submitted + envelopes). ``None`` when the seller sent none. An estimate, not a + deadline: schedule the next status check with + :func:`adcp.next_status_check_delay`, never treat a passed time as + failure. + """ + from adcp.status_checks import expected_confirmation_at + + return expected_confirmation_at(self) + class ActivityType(str, Enum): """Types of activity events.""" diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 374fb0c59..4996307ee 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -700,6 +700,7 @@ "domain_matches": "adcp.adagents", "encode_unreserved": "adcp.substitution", "evaluate_action_constraints": "adcp.media_buy_actions", + "expected_confirmation_at": "adcp.status_checks", "extract_adcp_error_info": "adcp.exceptions", "extract_webhook_result_data": "adcp.webhooks", "fetch_adagents": "adcp.adagents", @@ -731,6 +732,7 @@ "inject_trace_headers": "adcp.observability", "is_tracing_available": "adcp.observability", "materialize_change_terms": "adcp.media_buy_actions", + "next_status_check_delay": "adcp.status_checks", "next_step": "adcp.account_outcomes", "normalize_assets_required": "adcp.utils.format_assets", "normalize_domain": "adcp.identifiers", diff --git a/tests/test_expected_confirmation_at.py b/tests/test_expected_confirmation_at.py new file mode 100644 index 000000000..213e4c161 --- /dev/null +++ b/tests/test_expected_confirmation_at.py @@ -0,0 +1,345 @@ +"""AdCP 3.3 ``expected_confirmation_at``: buyer scheduling, seller helpers, downgrade rules. + +The seller's estimate of when a submitted task confirms rides on the submitted +envelope of ``create_media_buy``, ``buy_products``, ``accept_proposal`` and +finalize-only ``refine_proposals``, and on ``get_media_buys`` buys. On the three +compact tasks it is sent only to callers on 3.3; ``create_media_buy`` and +``get_media_buys`` keep it for older callers, whose objects are open. +""" + +from __future__ import annotations + +import logging +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timedelta, timezone +from typing import Any + +import pytest + +import adcp.client as client_module +from adcp import ADCPClient, expected_confirmation_at, next_status_check_delay +from adcp.compat.purchase_continuation import ( + CompatibilityOperationState, + CompatibilityPurchaseOperation, +) +from adcp.compat.versions import CONVERSION_METADATA_KEY, convert_payload +from adcp.decisioning import ( + DecisioningCapabilities, + DecisioningPlatform, + InMemoryTaskRegistry, + SingletonAccounts, +) +from adcp.decisioning.capabilities import LifecycleTool, MediaBuy +from adcp.decisioning.handler import PlatformHandler +from adcp.negotiation import WIRE_RESPONSE_METADATA_KEY, VerifiedRefinementResult +from adcp.server import ADCPHandler, ToolContext, submitted_response +from adcp.server.mcp_tools import create_tool_caller +from adcp.server.responses import format_expected_confirmation_at +from adcp.types import BuyProductsRequest, CreateMediaBuySubmittedResponse, RefineProposalsRequest +from adcp.types.core import AgentConfig, Protocol, TaskResult, TaskStatus +from adcp.validation import validate_response + +PIN = "3.3-beta.3" +ESTIMATE = "2030-01-15T09:00:00Z" +SUBMITTED = {"status": "submitted", "task_id": "task_1", "expected_confirmation_at": ESTIMATE} +GATED = ["buy_products", "accept_proposal", "refine_proposals"] + + +# --- downgrade rules ------------------------------------------------------- + + +@pytest.mark.parametrize("tool", GATED) +@pytest.mark.parametrize("direction", ["submitted", "sync"]) +def test_gated_tasks_drop_the_estimate_below_3_3_with_a_loss(tool: str, direction: str) -> None: + converted = convert_payload( + tool, SUBMITTED, direction=direction, source_version=PIN, target_version="3.2" + ) + assert "expected_confirmation_at" not in converted.payload + assert [(loss.pointer, loss.reason) for loss in converted.losses] == [ + ("/expected_confirmation_at", "field_unavailable") + ] + assert validate_response(tool, converted.payload, version="3.2").valid + + +@pytest.mark.parametrize("tool", GATED) +def test_gated_tasks_keep_the_estimate_for_3_3(tool: str) -> None: + converted = convert_payload( + tool, SUBMITTED, direction="submitted", source_version=PIN, target_version=PIN + ) + assert converted.payload["expected_confirmation_at"] == ESTIMATE + + +@pytest.mark.parametrize("target", ["3.2", "3.1", "3.0"]) +@pytest.mark.parametrize("direction", ["submitted", "sync"]) +def test_create_media_buy_keeps_the_estimate_for_every_release(target: str, direction: str) -> None: + converted = convert_payload( + "create_media_buy", + SUBMITTED, + direction=direction, + source_version=PIN, + target_version=target, + ) + assert converted.payload["expected_confirmation_at"] == ESTIMATE + assert converted.losses == () + assert validate_response("create_media_buy", converted.payload, version=target).valid + + +@pytest.mark.parametrize("target", ["3.2", "3.1"]) +def test_get_media_buys_keeps_the_estimate_on_a_provisional_buy(target: str) -> None: + payload = { + "media_buys": [ + { + "media_buy_id": "mb_1", + "status": "pending_creatives", + "confirmed_at": None, + "expected_confirmation_at": ESTIMATE, + "currency": "USD", + "total_budget": 100, + "packages": [], + } + ] + } + converted = convert_payload( + "get_media_buys", payload, direction="sync", source_version=PIN, target_version=target + ) + assert converted.payload["media_buys"][0]["expected_confirmation_at"] == ESTIMATE + assert not any(loss.pointer.endswith("expected_confirmation_at") for loss in converted.losses) + + +# --- seller helpers -------------------------------------------------------- + + +def test_submitted_response_normalizes_the_estimate_to_utc() -> None: + aware = datetime(2030, 1, 15, 10, 0, tzinfo=timezone(timedelta(hours=1))) + assert submitted_response("task_1", expected_confirmation_at=aware) == SUBMITTED + assert submitted_response("task_1", expected_confirmation_at="2030-01-15T09:00:00+00:00") == ( + SUBMITTED + ) + assert submitted_response("task_1") == {"status": "submitted", "task_id": "task_1"} + + +@pytest.mark.parametrize("value", [datetime(2030, 1, 15, 9, 0), "2030-01-15T09:00:00", "soon"]) +def test_estimate_must_name_an_instant(value: Any) -> None: + with pytest.raises(ValueError): + format_expected_confirmation_at(value) + + +def test_a_past_estimate_is_allowed_and_logged(caplog: pytest.LogCaptureFixture) -> None: + with caplog.at_level(logging.WARNING, logger="adcp.server"): + assert format_expected_confirmation_at("2001-01-01T00:00:00Z") == "2001-01-01T00:00:00Z" + assert "in the past" in caplog.text + + +class _Seller(ADCPHandler[Any]): + async def buy_products(self, params, context=None): + return submitted_response("task_1", expected_confirmation_at=ESTIMATE) + + async def create_media_buy(self, params, context=None): + return submitted_response("task_2", expected_confirmation_at=ESTIMATE) + + +_BUY = { + "idempotency_key": "expected-confirmation-1640", + "account": {"account_id": "acct_1"}, + "feed_version": "feed_1", + "purchases": [{"product_id": "prod_1", "pricing_option_id": "po_1", "budget": 1000}], + "start_time": "2030-01-20T00:00:00Z", + "end_time": "2030-02-01T00:00:00Z", +} + + +@pytest.mark.asyncio +async def test_server_drops_the_estimate_for_a_3_2_buy_products_caller() -> None: + caller = create_tool_caller(_Seller(), "buy_products") + context = ToolContext() + older = await caller({**_BUY, "adcp_version": "3.2"}, context) + assert "expected_confirmation_at" not in older + losses = context.metadata[CONVERSION_METADATA_KEY]["losses"] + assert any(loss["pointer"] == "/expected_confirmation_at" for loss in losses) + + current = await caller({**_BUY, "adcp_version": PIN}) + assert current["expected_confirmation_at"] == ESTIMATE + + +class _Platform(DecisioningPlatform): + capabilities = DecisioningCapabilities( + specialisms=["sales-non-guaranteed"], + media_buy=MediaBuy(lifecycle_tools=[LifecycleTool.buy_products]), + ) + accounts = SingletonAccounts(account_id="acct_1") + + def get_products(self, req, ctx): + return {"products": []} + + def create_media_buy(self, req, ctx): + return {"media_buy_id": "mb_1"} + + def update_media_buy(self, media_buy_id, patch, ctx): + return {"media_buy_id": media_buy_id} + + def sync_creatives(self, req, ctx): + return {"creatives": []} + + def get_media_buy_delivery(self, req, ctx): + return {"media_buy_deliveries": []} + + def buy_products(self, req, ctx): + return ctx.handoff_to_workflow( + lambda task_ctx: None, + expected_confirmation_at=datetime(2030, 1, 15, 9, 0, tzinfo=timezone.utc), + ) + + +@pytest.mark.asyncio +async def test_decisioning_handoff_carries_the_estimate_on_the_submitted_envelope() -> None: + with ThreadPoolExecutor(max_workers=1) as executor: + handler = PlatformHandler(_Platform(), executor=executor, registry=InMemoryTaskRegistry()) + response = await handler.buy_products( + BuyProductsRequest.model_validate(_BUY), ToolContext() + ) + assert response["status"] == "submitted" + assert response["expected_confirmation_at"] == ESTIMATE + assert validate_response("buy_products", dict(response), version=PIN).valid + + +def test_handoff_refuses_a_naive_estimate() -> None: + from adcp.decisioning.context import RequestContext + + with pytest.raises(ValueError, match="timezone"): + RequestContext().handoff_to_task( + lambda task_ctx: None, expected_confirmation_at=datetime(2030, 1, 15) + ) + + +def test_estimate_on_a_task_that_does_not_define_it_is_omitted( + caplog: pytest.LogCaptureFixture, +) -> None: + from adcp.decisioning.dispatch import _submitted_projection + + with caplog.at_level(logging.WARNING): + projection = _submitted_projection("task_1", "sync_creatives", ESTIMATE) + assert projection == {"task_id": "task_1", "status": "submitted"} + assert "not defined for sync_creatives" in caplog.text + assert _submitted_projection("task_1", "refine_proposals", ESTIMATE) == { + "task_id": "task_1", + "status": "submitted", + "expected_confirmation_at": ESTIMATE, + } + + +# --- buyer scheduling ------------------------------------------------------ + + +def test_task_result_surfaces_the_estimate_from_typed_and_wire_data() -> None: + typed = CreateMediaBuySubmittedResponse.model_validate(SUBMITTED) + result = TaskResult[Any](status=TaskStatus.SUBMITTED, data=typed) + assert result.expected_confirmation_at == datetime(2030, 1, 15, 9, 0, tzinfo=timezone.utc) + wire = TaskResult[Any](status=TaskStatus.SUBMITTED, data=dict(SUBMITTED)) + assert wire.expected_confirmation_at == result.expected_confirmation_at + assert TaskResult[Any](status=TaskStatus.SUBMITTED, data={}).expected_confirmation_at is None + assert expected_confirmation_at({"expected_confirmation_at": "not a time"}) is None + + +NOW = datetime(2030, 1, 15, 8, 0, tzinfo=timezone.utc) + + +@pytest.mark.parametrize( + ("estimate", "expected"), + [ + (None, 5.0), # no estimate: normal cadence + ("2030-01-15T07:00:00Z", 5.0), # passed: normal cadence, never a failure + ("2030-01-15T08:00:02Z", 5.0), # sooner than the floor + ("2030-01-15T08:01:00Z", 60.0), # scheduled for the estimate + ("2030-01-15T09:00:00Z", 300.0), # capped at the ceiling + ], +) +def test_next_check_is_bounded_by_floor_and_ceiling(estimate: str | None, expected: float) -> None: + assert next_status_check_delay(estimate, poll_interval=5, max_interval=300, now=NOW) == expected + + +def test_delay_bounds_are_validated() -> None: + with pytest.raises(ValueError): + next_status_check_delay(None, poll_interval=10, max_interval=5) + with pytest.raises(ValueError): + next_status_check_delay(None, poll_interval=0, max_interval=5) + + +class _StopError(Exception): + pass + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("offset", "expected"), + [(timedelta(hours=2), 30.0), (timedelta(hours=-1), 0.5), (None, 0.5)], +) +async def test_refinement_waiter_schedules_checks_from_the_estimate( + monkeypatch: pytest.MonkeyPatch, offset: timedelta | None, expected: float +) -> None: + client = ADCPClient( + AgentConfig(id="seller", agent_uri="https://seller.example", protocol=Protocol.A2A) + ) + wire: dict[str, Any] = {"status": "submitted", "task_id": "task-1"} + if offset is not None: + wire["expected_confirmation_at"] = (datetime.now(timezone.utc) + offset).isoformat() + initial = VerifiedRefinementResult( + TaskResult( + status=TaskStatus.SUBMITTED, + data={"task_id": "task-1"}, + metadata={WIRE_RESPONSE_METADATA_KEY: wire}, + ) + ) + + class _Working: + status = "working" + + async def get_task_status(_request): + return TaskResult(status=TaskStatus.WORKING, data=_Working()) + + delays: list[float] = [] + + async def sleep(delay: float) -> None: + delays.append(delay) + raise _StopError + + monkeypatch.setattr(client, "get_task_status", get_task_status) + monkeypatch.setattr(client_module.asyncio, "sleep", sleep) + with pytest.raises(_StopError): + await client.wait_for_refinement_verified( + RefineProposalsRequest.model_validate( + { + "idempotency_key": "refine-expected-1640", + "account": {"account_id": "acct_1"}, + "refinements": [{"proposal_id": "prop_1", "action": "finalize"}], + } + ), + initial, + poll_interval=0.5, + max_poll_interval=30.0, + ) + assert delays == [pytest.approx(expected)] + + +@pytest.mark.parametrize( + ("state", "result", "expected"), + [ + (CompatibilityOperationState.PENDING, dict(SUBMITTED), 300.0), + (CompatibilityOperationState.PENDING, {"status": "submitted", "task_id": "t"}, 5.0), + (CompatibilityOperationState.SUCCEEDED, dict(SUBMITTED), 5.0), + ], +) +def test_purchase_continuation_schedules_refresh_from_the_estimate( + state: CompatibilityOperationState, result: dict[str, Any], expected: float +) -> None: + operation = CompatibilityPurchaseOperation( + operation_id="op_1", + principal_id="buyer", + idempotency_key="key", + token_hash="hash", + payload_hash="hash", + state=state, + revision=1, + execution_input={}, + result=result, + ) + assert operation.next_refresh_delay(poll_interval=5, max_interval=300, now=NOW) == expected