feat: schedule status checks from expected_confirmation_at

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. |

docs/adcp-3.3-sdk-inventory.md

@@ -39,15 +39,15 @@ 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
 
 ### Media buy lifecycle
 
 | 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) |

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` |
 

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,

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(

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:

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: