From 1cbec5845ef8e23b38d889179908cd4748e4c868 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 21 Aug 2026 08:55:07 +0200 Subject: [PATCH] fix(compat): harden legacy purchase continuation recovery --- docs/legacy-purchase-continuations.md | 80 ++- src/adcp/compat/purchase_continuation.py | 605 ++++++++++++++--- src/adcp/compat/sqlite_continuation_store.py | 388 +++++++++-- tests/test_purchase_continuation.py | 666 ++++++++++++++++++- 4 files changed, 1578 insertions(+), 161 deletions(-) diff --git a/docs/legacy-purchase-continuations.md b/docs/legacy-purchase-continuations.md index 47cc5c39..f43b5bbc 100644 --- a/docs/legacy-purchase-continuations.md +++ b/docs/legacy-purchase-continuations.md @@ -58,8 +58,11 @@ token = await coordinator.issue_legacy_create_continuation( observed_request=complete_get_products_request, observed_response=complete_get_products_response, product_ids=[product["product_id"] for product in products], + buyer_visible_products=projected_products_shown_to_buyer, losses=["feed_version_not_atomic", "pricing_version_not_atomic"], target_binding=stable_seller_session_id, + # Set true only when the actual 3.0/3.1 peer guarantees mutation replay. + mutation_idempotency_guaranteed=True, ) ``` @@ -80,12 +83,29 @@ processes sharing one ordinary local filesystem. Distributed deployments should implement `CompatibilityContinuationStore` on their transactional database and preserve the same atomic state transitions. -The SQLite ledger is created with mode `0600` and an existing file with group -or other access is rejected. This is access control, not encryption; use an -encrypted volume or an application-owned encrypted store when payloads require -encryption at rest. `purge_resolved_before(cutoff)` removes only old succeeded -and never-claimed continuations. It deliberately retains claimed, `in_flight`, -and `ambiguous` operations regardless of age. +The SQLite ledger is created with mode `0600`; its direct parent must be owned +by the current user and cannot be group/world writable. Existing database and +sidecar files with group or other access are rejected before every pathname +open. This is access control, not encryption; use an encrypted volume or an +application-owned encrypted store when payloads require encryption at rest. +`purge_resolved_before(cutoff)` removes only old succeeded/failed and +never-claimed continuations. It deliberately retains claimed, `in_flight`, +`pending`, and `ambiguous` operations regardless of age. + +Ledgers created by the initial pre-release coordinator are migrated in place. +The old ledger did not retain the buyer-visible pricing subset, so unresolved +old rows are explicitly non-executable instead of exposing seller-only options. +The first exact retry may atomically adopt the sanitized execution snapshot +while incrementing the operation revision. A pre-release 3.0/3.1 row also did +not bind a verified replay guarantee, which the SDK cannot infer after the +fact. Already-terminal results still replay, and authoritative reconciliation +may still recover an applied result. Never start a fresh purchase until an +unresolved old operation has been reconciled. + +For a migrated unresolved operation, call `continue_legacy_purchase` once with +the exact original input before using the recovery API. That retry adopts the +missing execution snapshot but still fails closed without executing; then look +up the new operation revision and perform fenced recovery. ## What the application owns @@ -114,8 +134,8 @@ natural key. The durable operation ledger moves through: ```text -claimed -> in_flight -> succeeded - \-> ambiguous +claimed -> in_flight -> succeeded | failed | pending + \-> ambiguous -> claimed (only after authoritative absence) ``` The token is consumed when the first seller mutation is reserved. Exact @@ -126,11 +146,29 @@ cannot claim the token. An exception, timeout, or cancellation observed by the coordinator after `in_flight` is marked `ambiguous`. A hard process loss leaves the durable row `in_flight`; recovery remains closed until the application fences the old -executor and explicitly transitions that row to `ambiguous`. The SDK never -reopens the token by elapsed time and never blindly resends the legacy request. -A reconciler may then prove that the mutation was applied and supply its -result, or prove it was not applied and allow the same durable operation to -resume. An inconclusive or absent reconciler raises +executor. Look up a revision-bearing snapshot and use the fenced recovery API: + +```python +operation = await coordinator.get_legacy_purchase_operation( + operation_id, + principal_id=authenticated_principal, +) + +# First revoke/fence the old worker's ability to reach the seller. The CAS +# revision fences stale ledger completion, but cannot revoke network access. +result = await coordinator.recover_legacy_purchase( + operation, + principal_id=authenticated_principal, + target_binding=stable_seller_session_id, +) +``` + +Recovery atomically fences `in_flight` to `ambiguous` with a revision CAS. A +stale snapshot cannot recover or complete the operation. The SDK never reopens +the token by elapsed time and never blindly resends the legacy request. A +reconciler may then prove that the mutation was applied and supply its exact +source-version response, or prove it was not applied and allow the same durable +operation to resume. An inconclusive or absent reconciler raises `CompatibilityContinuationError` with code `ambiguous_legacy_mutation` and an `operation_id` for operators. @@ -140,8 +178,20 @@ Before the atomic claim, the coordinator rejects expiry, principal/account or target rebinding, product substitution, package-set drift, duplicate selected IDs, stale/partial/excess loss consent, and source-schema violations. The nested request is validated against the exact source-version -`create-media-buy-request` schema and must use explicit packages. AdCP 2.5 -continuations must also declare `mutation_idempotency_not_guaranteed`. +`create-media-buy-request` schema and must use explicit packages. Pricing +selection and all projected pricing terms are checked against the token-bound +buyer-visible product projection and complete observed option, not merely the +seller's larger set of option IDs. AdCP 2.5 continuations, and +3.0/3.1 peers without a verified mutation replay guarantee, must declare +`mutation_idempotency_not_guaranteed`. + +Executor and reconciliation results are validated against the exact legacy +`create_media_buy` response schema before persistence. Synchronous success, +terminal errors, and submitted task envelopes are stored and replayed in their +distinct durable states; arbitrary mappings are never promoted to success. +SDK `TaskResult` wrappers are unwrapped or projected only when they can produce +a valid source-version envelope. Synchronous executor/reconciler callbacks run +in a worker thread so they do not block the event loop. `listed_purchase` is different: it is executable only with a real account-scoped seller feed and unchanged seller-issued `feed_version` and diff --git a/src/adcp/compat/purchase_continuation.py b/src/adcp/compat/purchase_continuation.py index 71add6d1..57b7ce31 100644 --- a/src/adcp/compat/purchase_continuation.py +++ b/src/adcp/compat/purchase_continuation.py @@ -29,6 +29,7 @@ from pydantic import BaseModel, ValidationError from adcp.types import AccountReference, CompatibilityPurchaseCoordinatorInput +from adcp.types.core import TaskResult, TaskStatus from adcp.validation import ( get_bundle_adcp_version, validate_request, @@ -36,7 +37,7 @@ ) JsonObject: TypeAlias = dict[str, Any] -LegacyPurchaseResult: TypeAlias = Mapping[str, Any] | BaseModel +LegacyPurchaseResult: TypeAlias = Mapping[str, Any] | BaseModel | TaskResult[Any] LegacyPurchaseExecutor: TypeAlias = Callable[ ["LegacyPurchaseExecution"], LegacyPurchaseResult | Awaitable[LegacyPurchaseResult] ] @@ -66,6 +67,7 @@ class CompatibilityContinuationErrorCode(str, Enum): ALREADY_CLAIMED = "continuation_already_claimed" IDEMPOTENCY_CONFLICT = "continuation_idempotency_conflict" AMBIGUOUS_MUTATION = "ambiguous_legacy_mutation" + INVALID_LEGACY_RESPONSE = "invalid_legacy_create_response" STORE_CONFLICT = "continuation_store_conflict" @@ -95,7 +97,9 @@ class CompatibilityOperationState(str, Enum): CLAIMED = "claimed" IN_FLIGHT = "in_flight" + PENDING = "pending" SUCCEEDED = "succeeded" + FAILED = "failed" AMBIGUOUS = "ambiguous" @@ -118,7 +122,9 @@ class LegacyPurchaseContinuation: observed_response: JsonObject observed_payload_hash: str product_ids: tuple[str, ...] + projected_products: tuple[JsonObject, ...] | None losses: frozenset[str] + mutation_idempotency_guaranteed: bool target_binding: str listed_purchase_context: JsonObject | None = None @@ -133,6 +139,8 @@ class CompatibilityPurchaseOperation: token_hash: str payload_hash: str state: CompatibilityOperationState + revision: int + execution_input: JsonObject result: JsonObject | None = None @@ -164,11 +172,11 @@ class ReconciliationResult: """Authoritative result of reconciling an interrupted seller mutation.""" status: ReconciliationStatus - result: JsonObject | None = None + result: LegacyPurchaseResult | None = None @classmethod def applied(cls, result: LegacyPurchaseResult) -> ReconciliationResult: - return cls(ReconciliationStatus.APPLIED, _result_payload(result)) + return cls(ReconciliationStatus.APPLIED, copy.deepcopy(result)) @classmethod def not_applied(cls) -> ReconciliationResult: @@ -206,9 +214,14 @@ async def claim( principal_id: str, idempotency_key: str, payload_hash: str, + execution_input: Mapping[str, Any], now: datetime, ) -> CompatibilityPurchaseOperation: ... + async def get_operation( + self, operation_id: str, *, principal_id: str + ) -> CompatibilityPurchaseOperation | None: ... + async def mark_in_flight( self, operation: CompatibilityPurchaseOperation ) -> CompatibilityPurchaseOperation: ... @@ -221,8 +234,16 @@ async def complete( self, operation: CompatibilityPurchaseOperation, result: Mapping[str, Any], + *, + state: CompatibilityOperationState = CompatibilityOperationState.SUCCEEDED, ) -> CompatibilityPurchaseOperation: ... + async def fence_in_flight( + self, operation: CompatibilityPurchaseOperation + ) -> CompatibilityPurchaseOperation: + """CAS ``IN_FLIGHT`` to ``AMBIGUOUS`` using the operation revision.""" + ... + async def resume_after_not_applied( self, operation: CompatibilityPurchaseOperation ) -> CompatibilityPurchaseOperation: @@ -268,6 +289,7 @@ async def claim( principal_id: str, idempotency_key: str, payload_hash: str, + execution_input: Mapping[str, Any], now: datetime, ) -> CompatibilityPurchaseOperation: async with self._lock: @@ -281,6 +303,8 @@ async def claim( "idempotency key was already used with a different logical payload", "Use the original payload or start a new projected purchase.", ) + if existing.execution_input != _json_copy(execution_input): + raise _store_state_error("stored execution input changed for idempotent claim") return _copy_operation(existing) record = self._continuations.get(token_hash) @@ -306,11 +330,25 @@ async def claim( token_hash=token_hash, payload_hash=payload_hash, state=CompatibilityOperationState.CLAIMED, + revision=1, + execution_input=_json_copy(execution_input), ) self._operations[key] = operation self._claimed_by[token_hash] = operation.operation_id return _copy_operation(operation) + async def get_operation( + self, operation_id: str, *, principal_id: str + ) -> CompatibilityPurchaseOperation | None: + async with self._lock: + for operation in self._operations.values(): + if ( + operation.operation_id == operation_id + and operation.principal_id == principal_id + ): + return _copy_operation(operation) + return None + async def mark_in_flight( self, operation: CompatibilityPurchaseOperation ) -> CompatibilityPurchaseOperation: @@ -333,18 +371,36 @@ async def complete( self, operation: CompatibilityPurchaseOperation, result: Mapping[str, Any], + *, + state: CompatibilityOperationState = CompatibilityOperationState.SUCCEEDED, ) -> CompatibilityPurchaseOperation: + if state not in { + CompatibilityOperationState.PENDING, + CompatibilityOperationState.SUCCEEDED, + CompatibilityOperationState.FAILED, + }: + raise ValueError("completed result requires pending, succeeded, or failed state") copied = _json_copy(result) return await self._transition( operation, allowed={ CompatibilityOperationState.IN_FLIGHT, CompatibilityOperationState.AMBIGUOUS, + CompatibilityOperationState.PENDING, }, - target=CompatibilityOperationState.SUCCEEDED, + target=state, result=copied, ) + async def fence_in_flight( + self, operation: CompatibilityPurchaseOperation + ) -> CompatibilityPurchaseOperation: + return await self._transition( + operation, + allowed={CompatibilityOperationState.IN_FLIGHT}, + target=CompatibilityOperationState.AMBIGUOUS, + ) + async def resume_after_not_applied( self, operation: CompatibilityPurchaseOperation ) -> CompatibilityPurchaseOperation: @@ -367,11 +423,18 @@ async def _transition( current = self._operations.get(key) if current is None or current.operation_id != operation.operation_id: raise _store_state_error("operation is missing from continuation store") + if current.revision != operation.revision: + raise _store_state_error("operation revision changed concurrently") if current.state not in allowed: raise _store_state_error( f"cannot transition operation from {current.state.value} to {target.value}" ) - updated = replace(current, state=target, result=copy.deepcopy(result)) + updated = replace( + current, + state=target, + revision=current.revision + 1, + result=copy.deepcopy(result), + ) self._operations[key] = updated return _copy_operation(updated) @@ -410,14 +473,18 @@ async def issue_legacy_create_continuation( observed_request: Mapping[str, Any], observed_response: Mapping[str, Any], product_ids: list[str] | tuple[str, ...], + buyer_visible_products: list[Mapping[str, Any]] | tuple[Mapping[str, Any], ...], losses: list[str] | tuple[str, ...] | frozenset[str], target_binding: str, + mutation_idempotency_guaranteed: bool = False, listed_purchase_context: Mapping[str, Any] | None = None, ) -> str: """Persist all projection bindings and return the opaque bearer token.""" _require_text(principal_id, "principal_id") _require_text(target_binding, "target_binding") + if type(mutation_idempotency_guaranteed) is not bool: + raise _invalid("mutation_idempotency_guaranteed must be a boolean") _validate_source_version(source_adcp_version) expires_at = _aware_utc(expires_at, field="expires_at") now = _aware_utc(self._clock(), field="clock result") @@ -427,7 +494,11 @@ async def issue_legacy_create_continuation( account_payload = _account_payload(account) account_identity = canonical_account_identity(account_payload) ids = _unique_nonempty_strings(product_ids, field="product_ids") - loss_set = _validate_loss_set(losses, source_adcp_version=source_adcp_version) + loss_set = _validate_loss_set( + losses, + source_adcp_version=source_adcp_version, + mutation_idempotency_guaranteed=mutation_idempotency_guaranteed, + ) observed_req = _json_copy(observed_request) observed_resp = _json_copy(observed_response) _validate_source_discovery( @@ -437,6 +508,12 @@ async def issue_legacy_create_continuation( account_identity=account_identity, ) _validate_observed_product_ids(observed_resp, ids) + projected = _validate_projected_products( + buyer_visible_products, + observed_resp, + ids, + source_adcp_version=source_adcp_version, + ) listed = ( _json_copy(listed_purchase_context) if listed_purchase_context is not None else None ) @@ -454,7 +531,9 @@ async def issue_legacy_create_continuation( observed_response=observed_resp, observed_payload_hash=observed_payload_hash, product_ids=ids, + projected_products=projected, losses=loss_set, + mutation_idempotency_guaranteed=mutation_idempotency_guaranteed, target_binding=target_binding, listed_purchase_context=listed, ) @@ -486,78 +565,148 @@ async def continue_legacy_purchase( principal_id=principal_id, idempotency_key=payload["idempotency_key"], payload_hash=payload_hash, + execution_input=_execution_input(payload), now=now, ) - execution = _execution_from(operation, record, payload, target_binding) + return await self._drive(operation, record, target_binding=target_binding) + + async def get_legacy_purchase_operation( + self, operation_id: str, *, principal_id: str + ) -> CompatibilityPurchaseOperation: + """Return a principal-scoped operation snapshot carrying its CAS revision.""" + + _require_text(operation_id, "operation_id") + _require_text(principal_id, "principal_id") + operation = await self.store.get_operation(operation_id, principal_id=principal_id) + if operation is None: + raise _not_found() + return _copy_operation(operation) + + async def recover_legacy_purchase( + self, + operation: CompatibilityPurchaseOperation, + *, + principal_id: str, + target_binding: str, + ) -> JsonObject: + """Fence an abandoned executor, reconcile, and resume using a CAS snapshot. + + Callers must first ensure the old executor cannot still reach the seller. + The operation revision fences stale durable completions; it cannot revoke + external network credentials or an already-running seller request. + """ + + _require_text(principal_id, "principal_id") + _require_text(target_binding, "target_binding") + if operation.principal_id != principal_id: + raise _not_found() + current = await self.store.get_operation(operation.operation_id, principal_id=principal_id) + if current is None: + raise _not_found() + if current.revision != operation.revision: + raise _store_state_error("operation revision changed before recovery fence") + record = await self.store.get_continuation(current.token_hash, principal_id=principal_id) + if record is None: + raise _not_found() + if not current.execution_input: + raise _error( + CompatibilityContinuationErrorCode.STORE_CONFLICT, + "migrated operation has no recoverable execution snapshot", + "Submit one exact retry through continue_legacy_purchase first, then look up " + "a fresh revision-bearing operation snapshot.", + details={"operation_id": current.operation_id}, + ) + payload = _json_copy(current.execution_input) + self._validate_bindings(payload, record, target_binding=target_binding) + if current.state == CompatibilityOperationState.IN_FLIGHT: + current = await self.store.fence_in_flight(current) + return await self._drive(current, record, target_binding=target_binding, recovery=True) - if operation.state == CompatibilityOperationState.SUCCEEDED: + async def _drive( + self, + operation: CompatibilityPurchaseOperation, + record: LegacyPurchaseContinuation, + *, + target_binding: str, + recovery: bool = False, + ) -> JsonObject: + execution = _execution_from(operation, record, operation.execution_input, target_binding) + if operation.state in { + CompatibilityOperationState.PENDING, + CompatibilityOperationState.SUCCEEDED, + CompatibilityOperationState.FAILED, + }: if operation.result is None: - raise _store_state_error("succeeded operation has no stored result") + raise _store_state_error("terminal operation has no stored result") return copy.deepcopy(operation.result) if operation.state in { CompatibilityOperationState.IN_FLIGHT, CompatibilityOperationState.AMBIGUOUS, }: - operation = await self._reconcile(execution, operation) - if operation.state == CompatibilityOperationState.SUCCEEDED: + operation = await self._reconcile(execution, operation, allow_not_applied=recovery) + if operation.state in { + CompatibilityOperationState.PENDING, + CompatibilityOperationState.SUCCEEDED, + CompatibilityOperationState.FAILED, + }: assert operation.result is not None return copy.deepcopy(operation.result) + # A migrated row may replay a terminal result, or reconcile an already + # applied mutation, without relying on a seller replay guarantee. The + # guarantee becomes mandatory only before this coordinator can issue + # another mutation call. + if record.projected_products is None: + raise _invalid( + "legacy continuation predates buyer-visible pricing binding and cannot execute" + ) + _validate_loss_set( + record.losses, + source_adcp_version=record.source_adcp_version, + mutation_idempotency_guaranteed=record.mutation_idempotency_guaranteed, + ) + try: - operation = await self.store.mark_in_flight(operation) + operation = await self._reserve_execution(operation) except CompatibilityContinuationError as exc: if exc.code != CompatibilityContinuationErrorCode.STORE_CONFLICT: raise - # An exact concurrent retry may have won the CLAIMED -> IN_FLIGHT - # CAS after our claim read. Reload it through claim; never execute - # in both callers. - operation = await self.store.claim( - token_hash, - principal_id=principal_id, - idempotency_key=payload["idempotency_key"], - payload_hash=payload_hash, - now=now, + latest = await self.store.get_operation( + operation.operation_id, principal_id=operation.principal_id ) - if operation.state == CompatibilityOperationState.SUCCEEDED: - if operation.result is None: - raise _store_state_error("succeeded operation has no stored result") - return copy.deepcopy(operation.result) - if operation.state in { - CompatibilityOperationState.IN_FLIGHT, - CompatibilityOperationState.AMBIGUOUS, - }: - operation = await self._reconcile(execution, operation) - if operation.state == CompatibilityOperationState.SUCCEEDED: - assert operation.result is not None - return copy.deepcopy(operation.result) - operation = await self.store.mark_in_flight(operation) - else: - raise _store_state_error( - "operation remained claimed after losing execution reservation" - ) + if latest is None: + raise + return await self._drive( + latest, record, target_binding=target_binding, recovery=recovery + ) + try: - result = await _maybe_await(self.executor(execution)) - copied = _result_payload(result) + result = await _call_callback(self.executor, _copy_execution(execution)) except asyncio.CancelledError: - try: - await asyncio.shield(self.store.mark_ambiguous(operation)) - except Exception as store_exc: - raise _ambiguous_error(operation) from store_exc + await self._mark_ambiguous_after_interruption(operation) raise except Exception as exc: - try: - await asyncio.shield(self.store.mark_ambiguous(operation)) - except Exception as store_exc: - raise _ambiguous_error(operation) from store_exc + await self._mark_ambiguous_after_interruption(operation) + raise _ambiguous_error(operation) from exc + try: + copied, result_state = _validated_result( + result, source_adcp_version=record.source_adcp_version + ) + except CompatibilityContinuationError as exc: + await self._mark_ambiguous_after_interruption(operation) + exc.details.setdefault("operation_id", operation.operation_id) + raise + except Exception as exc: + await self._mark_ambiguous_after_interruption(operation) raise _ambiguous_error(operation) from exc try: - completed = await self.store.complete(operation, copied) + completed = await _shielded_transition( + self.store.complete(operation, copied, state=result_state) + ) + except asyncio.CancelledError: + raise except Exception as store_exc: - # The seller returned after the mutation, but durable result - # persistence failed. Best-effort mark AMBIGUOUS; even if that - # write also fails, surface a typed fail-closed error rather than - # leaking the backend exception or inviting a blind replay. try: await asyncio.shield(self.store.mark_ambiguous(operation)) except Exception: @@ -566,15 +715,41 @@ async def continue_legacy_purchase( assert completed.result is not None return copy.deepcopy(completed.result) + async def _reserve_execution( + self, operation: CompatibilityPurchaseOperation + ) -> CompatibilityPurchaseOperation: + task = asyncio.create_task(self.store.mark_in_flight(operation)) + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + try: + reserved = await task + await asyncio.shield(self.store.mark_ambiguous(reserved)) + except Exception as exc: + raise _ambiguous_error(operation) from exc + raise + + async def _mark_ambiguous_after_interruption( + self, operation: CompatibilityPurchaseOperation + ) -> None: + try: + await _shielded_transition(self.store.mark_ambiguous(operation)) + except Exception as store_exc: + raise _ambiguous_error(operation) from store_exc + async def _reconcile( self, execution: LegacyPurchaseExecution, operation: CompatibilityPurchaseOperation, + *, + allow_not_applied: bool = False, ) -> CompatibilityPurchaseOperation: if self.reconciler is None: raise _ambiguous_error(operation) try: - outcome = await _maybe_await(self.reconciler(execution, operation)) + outcome = await _call_callback( + self.reconciler, _copy_execution(execution), _copy_operation(operation) + ) except asyncio.CancelledError: raise except Exception as exc: @@ -585,14 +760,20 @@ async def _reconcile( if outcome.status == ReconciliationStatus.APPLIED: if outcome.result is None: raise ValueError("applied reconciliation requires a result") - return await self.store.complete(operation, outcome.result) + copied, state = _validated_result( + outcome.result, source_adcp_version=execution.source_adcp_version + ) + return await self.store.complete(operation, copied, state=state) if outcome.status == ReconciliationStatus.NOT_APPLIED: # IN_FLIGHT may still have a live executor in another worker. An # instantaneous seller lookup can report "not applied" immediately # before that executor commits, so reopening it would permit two # calls. Only the exception/cancellation path's durably AMBIGUOUS # state proves this coordinator no longer has a live owner. - if operation.state != CompatibilityOperationState.AMBIGUOUS: + if ( + operation.state != CompatibilityOperationState.AMBIGUOUS + or not allow_not_applied + ): raise _ambiguous_error(operation) return await self.store.resume_after_not_applied(operation) raise _ambiguous_error(operation) @@ -631,7 +812,10 @@ def _validate_bindings( if not set(selected).issubset(record.product_ids): raise _binding_error("selected products are not a subset of token-bound products") accepted = _validate_loss_set( - payload["accepted_losses"], source_adcp_version=record.source_adcp_version + payload["accepted_losses"], + source_adcp_version=record.source_adcp_version, + mutation_idempotency_guaranteed=record.mutation_idempotency_guaranteed, + enforce_guarantee=False, ) if accepted != record.losses: raise _error( @@ -663,16 +847,21 @@ def _validate_bindings( if not isinstance(packages, list) or not packages or request.get("proposal_id") is not None: raise _legacy_request_error("explicit-package mode is required") package_ids: list[str] = [] - observed_pricing = _observed_pricing_options(record.observed_response) + observed_pricing = ( + _projected_pricing_options(record.projected_products) + if record.projected_products is not None + else {} + ) for package in packages: if not isinstance(package, Mapping) or not isinstance(package.get("product_id"), str): raise _legacy_request_error("every package must carry a product_id") product_id = package["product_id"] package_ids.append(product_id) pricing_option_id = package.get("pricing_option_id") - if not isinstance( - pricing_option_id, str - ) or pricing_option_id not in observed_pricing.get(product_id, frozenset()): + if not isinstance(pricing_option_id, str) or ( + record.projected_products is not None + and pricing_option_id not in observed_pricing.get(product_id, frozenset()) + ): raise _binding_error( "legacy package pricing_option_id was not observed for its product" ) @@ -730,11 +919,12 @@ def _parse_input( value: CompatibilityPurchaseCoordinatorInput | Mapping[str, Any], ) -> JsonObject: try: - model = ( - value + source = ( + value.model_dump(mode="python", by_alias=True) if isinstance(value, CompatibilityPurchaseCoordinatorInput) - else CompatibilityPurchaseCoordinatorInput.model_validate(value) + else value ) + model = CompatibilityPurchaseCoordinatorInput.model_validate(source) payload = model.model_dump(mode="json", by_alias=True, exclude_none=True) except (ValidationError, TypeError, ValueError) as exc: issues = ( @@ -781,6 +971,12 @@ def _execution_from( ) +def _execution_input(payload: JsonObject) -> JsonObject: + """Persist the validated execution fields without the bearer continuation token.""" + + return _json_copy({key: value for key, value in payload.items() if key != "continuation_token"}) + + def _validate_source_version(version: str) -> None: if not isinstance(version, str) or _SOURCE_VERSION_RE.fullmatch(version) is None: raise _invalid("source_adcp_version must be an exact 2.5.x, 3.0.x, or 3.1.x release") @@ -792,7 +988,13 @@ def _validate_source_version(version: str) -> None: ) -def _validate_loss_set(values: Any, *, source_adcp_version: str) -> frozenset[str]: +def _validate_loss_set( + values: Any, + *, + source_adcp_version: str, + mutation_idempotency_guaranteed: bool, + enforce_guarantee: bool = True, +) -> frozenset[str]: if not isinstance(values, (list, tuple, frozenset)): raise _invalid("losses must be an array") raw = [str(value) for value in values] @@ -801,8 +1003,18 @@ def _validate_loss_set(values: Any, *, source_adcp_version: str) -> frozenset[st result = frozenset(raw) if not _REQUIRED_LOSSES.issubset(result) or not result.issubset(_ALLOWED_LOSSES): raise _invalid("losses must contain both atomicity losses and no unknown values") - if source_adcp_version.startswith("2.5.") and _MUTATION_LOSS not in result: - raise _invalid("AdCP 2.5 continuations require the mutation idempotency loss") + requires_mutation_loss = source_adcp_version.startswith("2.5.") or not bool( + mutation_idempotency_guaranteed + ) + if enforce_guarantee and requires_mutation_loss and _MUTATION_LOSS not in result: + raise _invalid( + "continuations without a verified peer replay guarantee require the mutation " + "idempotency loss" + ) + if enforce_guarantee and not requires_mutation_loss and _MUTATION_LOSS in result: + raise _invalid( + "mutation idempotency loss must be omitted when a peer replay guarantee is bound" + ) return result @@ -828,8 +1040,10 @@ def _validate_observed_product_ids(response: JsonObject, expected: tuple[str, .. if not isinstance(product.get("pricing_options"), list) or not product["pricing_options"]: raise _invalid("every observed product must retain its pricing_options") observed.append(product["product_id"]) - if len(observed) != len(set(observed)) or set(observed) != set(expected): - raise _invalid("product_ids must exactly match the complete observed product set") + if len(observed) != len(set(observed)): + raise _invalid("observed product IDs must be unique") + if not set(expected).issubset(observed): + raise _invalid("product_ids must be a subset of the observed product set") def _observed_pricing_options(response: JsonObject) -> dict[str, frozenset[str]]: @@ -848,6 +1062,118 @@ def _observed_pricing_options(response: JsonObject) -> dict[str, frozenset[str]] return result +def _validate_projected_products( + values: Any, + observed_response: JsonObject, + product_ids: tuple[str, ...], + *, + source_adcp_version: str, +) -> tuple[JsonObject, ...]: + if not isinstance(values, (list, tuple)) or not values: + raise _invalid("buyer_visible_products must be a non-empty array") + projected = tuple(_json_copy(value) for value in values if isinstance(value, Mapping)) + if len(projected) != len(values): + raise _invalid("every buyer-visible product must be an object") + projected_pricing = _projected_pricing_options(projected) + if set(projected_pricing) != set(product_ids): + raise _invalid("product_ids must exactly match the buyer-visible product projection") + observed_pricing = _observed_pricing_options(observed_response) + observed_terms = _pricing_options_by_id(observed_response["products"]) + for product_id, option_ids in projected_pricing.items(): + if not option_ids.issubset(observed_pricing.get(product_id, frozenset())): + raise _invalid("buyer-visible pricing options must be present in observed discovery") + projected_terms = _pricing_options_by_id(projected) + for product_id, options in projected_terms.items(): + for option_id, option in options.items(): + observed_option = observed_terms.get(product_id, {}).get(option_id) + if observed_option is None or not _projected_option_matches_observed( + option, + observed_option, + legacy_25=source_adcp_version.startswith("2.5."), + ): + raise _binding_error("buyer-visible pricing terms differ from observed discovery") + return projected + + +def _projected_option_matches_observed( + projected: JsonObject, observed: JsonObject, *, legacy_25: bool +) -> bool: + return _normalized_pricing_terms( + projected, projected=True, legacy_25=legacy_25 + ) == _normalized_pricing_terms(observed, projected=False, legacy_25=legacy_25) + + +def _normalized_pricing_terms(value: JsonObject, *, projected: bool, legacy_25: bool) -> JsonObject: + # Preserve unknown/extension fields so a projection cannot silently alter + # commercial behavior. Only the explicit 2.5 representation differences + # are rewritten to their compact canonical equivalents. + normalized = copy.deepcopy(value) + if not projected and legacy_25: + is_fixed = normalized.pop("is_fixed", None) + rate = normalized.pop("rate", None) + if "fixed_price" not in normalized and is_fixed is True and rate is not None: + normalized["fixed_price"] = rate + guidance = normalized.get("price_guidance") + if isinstance(guidance, dict) and "floor" in guidance: + guidance = copy.deepcopy(guidance) + normalized["floor_price"] = guidance.pop("floor") + if guidance: + normalized["price_guidance"] = guidance + else: + normalized.pop("price_guidance", None) + return normalized + + +def _pricing_options_by_id( + products: list[Any] | tuple[JsonObject, ...], +) -> dict[str, dict[str, JsonObject]]: + result: dict[str, dict[str, JsonObject]] = {} + for product in products: + if not isinstance(product, Mapping) or not isinstance(product.get("product_id"), str): + raise _invalid("every product must carry product_id") + options = product.get("pricing_options") + if not isinstance(options, list): + raise _invalid("every product must carry pricing_options") + keyed: dict[str, JsonObject] = {} + for option in options: + if not isinstance(option, Mapping) or not isinstance( + option.get("pricing_option_id"), str + ): + raise _invalid("every pricing option must carry pricing_option_id") + option_id = option["pricing_option_id"] + if option_id in keyed: + raise _invalid("pricing option IDs must be unique per product") + keyed[option_id] = _json_copy(option) + result[product["product_id"]] = keyed + return result + + +def _projected_pricing_options( + products: tuple[JsonObject, ...] | list[JsonObject], +) -> dict[str, frozenset[str]]: + result: dict[str, frozenset[str]] = {} + for product in products: + product_id = product.get("product_id") + options = product.get("pricing_options") + if not isinstance(product_id, str) or not product_id: + raise _invalid("every buyer-visible product must carry product_id") + if product_id in result: + raise _invalid("buyer-visible product IDs must be unique") + if not isinstance(options, list) or not options: + raise _invalid("every buyer-visible product must retain non-empty pricing_options") + option_ids: list[str] = [] + for option in options: + if not isinstance(option, Mapping) or not isinstance( + option.get("pricing_option_id"), str + ): + raise _invalid("every buyer-visible pricing option must carry pricing_option_id") + option_ids.append(option["pricing_option_id"]) + if len(option_ids) != len(set(option_ids)): + raise _invalid("buyer-visible pricing option IDs must be unique per product") + result[product_id] = frozenset(option_ids) + return result + + def _validate_source_discovery( request: JsonObject, response: JsonObject, @@ -888,9 +1214,12 @@ def _validate_source_discovery( def _account_payload(value: Mapping[str, Any] | Any) -> JsonObject: try: - model = ( - value if isinstance(value, AccountReference) else AccountReference.model_validate(value) + source = ( + value.model_dump(mode="python", by_alias=True) + if isinstance(value, AccountReference) + else value ) + model = AccountReference.model_validate(source) payload = model.model_dump(mode="json", by_alias=True, exclude_none=True) except ValidationError as exc: raise _invalid("account must be a valid beta.4 AccountReference") from exc @@ -921,7 +1250,39 @@ def _json_copy(value: Mapping[str, Any]) -> JsonObject: return payload -def _result_payload(value: LegacyPurchaseResult) -> JsonObject: +def _raw_result_payload(value: LegacyPurchaseResult) -> JsonObject: + if isinstance(value, TaskResult): + payload = _raw_result_payload(value.data) if value.data is not None else {} + if value.status == TaskStatus.COMPLETED: + if value.data is None: + raise TypeError("completed TaskResult requires schema-shaped data") + return payload + if value.status == TaskStatus.FAILED: + if value.data is None and value.adcp_error is not None: + payload = {"errors": [value.adcp_error]} + return payload + status = { + TaskStatus.SUBMITTED: "submitted", + TaskStatus.WORKING: "working", + TaskStatus.NEEDS_INPUT: "input-required", + }.get(value.status) + if status is None: + raise TypeError(f"unsupported TaskResult status {value.status.value!r}") + payload.setdefault("status", status) + if payload.get("status") != status: + raise TypeError("TaskResult status conflicts with its data envelope") + task_id = None + if value.submitted is not None: + task_id = value.submitted.operation_id + if value.metadata is not None: + task_id = value.metadata.get("task_id", task_id) + if task_id is not None: + payload.setdefault("task_id", task_id) + if not isinstance(payload.get("task_id"), str) or not payload["task_id"]: + raise TypeError("pending TaskResult requires a non-empty task identity") + if value.message: + payload.setdefault("message", value.message) + return _json_copy(payload) if isinstance(value, BaseModel): dumped = value.model_dump(mode="json", by_alias=True, exclude_none=True) if not isinstance(dumped, dict): @@ -929,7 +1290,49 @@ def _result_payload(value: LegacyPurchaseResult) -> JsonObject: return _json_copy(dumped) if isinstance(value, Mapping): return _json_copy(value) - raise TypeError("legacy purchase executor must return a mapping or Pydantic model") + raise TypeError("legacy purchase executor must return a mapping, Pydantic model, or TaskResult") + + +def _validated_result( + value: LegacyPurchaseResult, *, source_adcp_version: str +) -> tuple[JsonObject, CompatibilityOperationState]: + try: + payload = _raw_result_payload(value) + except (TypeError, ValueError, CompatibilityContinuationError) as exc: + raise _invalid_legacy_response(source_adcp_version, []) from exc + outcome = validate_response("create_media_buy", payload, version=source_adcp_version) + if not outcome.valid or outcome.variant == "skipped": + raise _invalid_legacy_response( + source_adcp_version, + [ + { + "pointer": issue.pointer, + "keyword": issue.keyword, + "message": issue.message, + } + for issue in outcome.issues + ], + ) + if outcome.variant in {"submitted", "working", "input-required"}: + state = CompatibilityOperationState.PENDING + elif "errors" in payload: + state = CompatibilityOperationState.FAILED + else: + state = CompatibilityOperationState.SUCCEEDED + if state == CompatibilityOperationState.PENDING and ( + not isinstance(payload.get("task_id"), str) or not payload["task_id"] + ): + raise _invalid_legacy_response(source_adcp_version, []) + if isinstance(value, TaskResult): + expected = { + TaskStatus.SUBMITTED: CompatibilityOperationState.PENDING, + TaskStatus.WORKING: CompatibilityOperationState.PENDING, + TaskStatus.NEEDS_INPUT: CompatibilityOperationState.PENDING, + TaskStatus.FAILED: CompatibilityOperationState.FAILED, + }.get(value.status) + if expected is not None and state != expected: + raise _invalid_legacy_response(source_adcp_version, []) + return payload, state def _aware_utc(value: datetime, *, field: str) -> datetime: @@ -949,17 +1352,60 @@ async def _maybe_await(value: Any) -> Any: return value +async def _call_callback(callback: Callable[..., Any], *args: Any) -> Any: + call = callback + is_async = inspect.iscoroutinefunction(call) or inspect.iscoroutinefunction( + getattr(call, "__call__", None) + ) + if is_async: + return copy.deepcopy(await _maybe_await(call(*args))) + + def invoke_sync() -> Any: + value = call(*args) + return value if inspect.isawaitable(value) else copy.deepcopy(value) + + value = await asyncio.to_thread(invoke_sync) + return copy.deepcopy(await _maybe_await(value)) + + +async def _shielded_transition(value: Awaitable[Any]) -> Any: + task: asyncio.Future[Any] = asyncio.ensure_future(value) + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + # Let the durable CAS settle so cancellation cannot strand an unknown + # transition between seller completion and local persistence. + await task + raise + + def _copy_continuation(value: LegacyPurchaseContinuation) -> LegacyPurchaseContinuation: return replace( value, observed_request=copy.deepcopy(value.observed_request), observed_response=copy.deepcopy(value.observed_response), + projected_products=copy.deepcopy(value.projected_products), listed_purchase_context=copy.deepcopy(value.listed_purchase_context), ) def _copy_operation(value: CompatibilityPurchaseOperation) -> CompatibilityPurchaseOperation: - return replace(value, result=copy.deepcopy(value.result)) + return replace( + value, + execution_input=copy.deepcopy(value.execution_input), + result=copy.deepcopy(value.result), + ) + + +def _copy_execution(value: LegacyPurchaseExecution) -> LegacyPurchaseExecution: + return replace( + value, + account=copy.deepcopy(value.account), + legacy_create_request=copy.deepcopy(value.legacy_create_request), + observed_request=copy.deepcopy(value.observed_request), + observed_response=copy.deepcopy(value.observed_response), + listed_purchase_context=copy.deepcopy(value.listed_purchase_context), + ) def _error( @@ -990,6 +1436,17 @@ def _legacy_request_error(message: str) -> CompatibilityContinuationError: ) +def _invalid_legacy_response( + source_adcp_version: str, issues: list[JsonObject] +) -> CompatibilityContinuationError: + return _error( + CompatibilityContinuationErrorCode.INVALID_LEGACY_RESPONSE, + "legacy create_media_buy result failed exact source-version validation", + "Reconcile the seller mutation authoritatively; do not treat this payload as success.", + details={"source_adcp_version": source_adcp_version, "issues": issues}, + ) + + def _binding_error(message: str) -> CompatibilityContinuationError: return _error( CompatibilityContinuationErrorCode.BINDING_MISMATCH, diff --git a/src/adcp/compat/sqlite_continuation_store.py b/src/adcp/compat/sqlite_continuation_store.py index 94a97bd4..436d63e2 100644 --- a/src/adcp/compat/sqlite_continuation_store.py +++ b/src/adcp/compat/sqlite_continuation_store.py @@ -11,6 +11,7 @@ from __future__ import annotations import asyncio +import copy import json import os import secrets @@ -41,7 +42,9 @@ observed_response_json TEXT NOT NULL, observed_payload_hash TEXT NOT NULL, product_ids_json TEXT NOT NULL, + projected_products_json TEXT NOT NULL DEFAULT '[]', losses_json TEXT NOT NULL, + mutation_idempotency_guaranteed INTEGER NOT NULL DEFAULT 0, target_binding TEXT NOT NULL, listed_purchase_context_json TEXT, claimed_operation_id TEXT, @@ -58,8 +61,10 @@ token_hash TEXT NOT NULL, payload_hash TEXT NOT NULL, state TEXT NOT NULL CHECK ( - state IN ('claimed', 'in_flight', 'succeeded', 'ambiguous') + state IN ('claimed', 'in_flight', 'pending', 'succeeded', 'failed', 'ambiguous') ), + revision INTEGER NOT NULL DEFAULT 1, + execution_input_json TEXT NOT NULL DEFAULT '{}', result_json TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, @@ -86,8 +91,10 @@ def __init__( raw = str(path) if raw == ":memory:" or raw.startswith("file::memory:"): raise ValueError("SqliteCompatibilityContinuationStore requires a file-backed database") - self.path = Path(path).expanduser().resolve() - self.path.parent.mkdir(parents=True, exist_ok=True) + # Keep the lexical path so lstat() can reject symlink components; + # Path.resolve() would dereference them before the safety walk. + self.path = Path(os.path.abspath(os.path.expanduser(raw))) + self._ensure_private_parent_directory() self.timeout = timeout self._clock = clock or (lambda: datetime.now(timezone.utc)) self._ensure_private_database_file() @@ -112,6 +119,138 @@ def _ensure_timestamp_columns(self, conn: sqlite3.Connection) -> None: f"UPDATE {table} SET {column} = ? WHERE {column} IS NULL", (now,), ) + continuation_columns = { + row["name"] for row in conn.execute("PRAGMA table_info(adcp_compat_continuations)") + } + if "projected_products_json" not in continuation_columns: + conn.execute( + "ALTER TABLE adcp_compat_continuations " "ADD COLUMN projected_products_json TEXT" + ) + if "mutation_idempotency_guaranteed" not in continuation_columns: + conn.execute( + "ALTER TABLE adcp_compat_continuations " + "ADD COLUMN mutation_idempotency_guaranteed INTEGER NOT NULL DEFAULT 0" + ) + + operation_columns = { + row["name"] for row in conn.execute("PRAGMA table_info(adcp_compat_operations)") + } + if "revision" not in operation_columns: + conn.execute( + "ALTER TABLE adcp_compat_operations " + "ADD COLUMN revision INTEGER NOT NULL DEFAULT 1" + ) + if "execution_input_json" not in operation_columns: + conn.execute( + "ALTER TABLE adcp_compat_operations " + "ADD COLUMN execution_input_json TEXT NOT NULL DEFAULT '{}'" + ) + + operations_sql_row = conn.execute( + "SELECT sql FROM sqlite_master WHERE type = 'table' " + "AND name = 'adcp_compat_operations'" + ).fetchone() + operations_sql = operations_sql_row["sql"] if operations_sql_row is not None else "" + if "'pending'" not in operations_sql or "'failed'" not in operations_sql: + self._rebuild_operations_table(conn) + + @staticmethod + def _rebuild_operations_table(conn: sqlite3.Connection) -> None: + """Expand the operation-state constraint without losing ledger rows.""" + + conn.execute("ALTER TABLE adcp_compat_operations RENAME TO adcp_compat_operations_old") + conn.execute( + """ + CREATE TABLE adcp_compat_operations ( + operation_id TEXT PRIMARY KEY, + principal_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + token_hash TEXT NOT NULL, + payload_hash TEXT NOT NULL, + state TEXT NOT NULL CHECK ( + state IN ( + 'claimed', 'in_flight', 'pending', 'succeeded', 'failed', 'ambiguous' + ) + ), + revision INTEGER NOT NULL DEFAULT 1, + execution_input_json TEXT NOT NULL DEFAULT '{}', + result_json TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE (principal_id, idempotency_key), + UNIQUE (token_hash), + FOREIGN KEY (token_hash) + REFERENCES adcp_compat_continuations(token_hash) + ) + """ + ) + conn.execute( + """ + INSERT INTO adcp_compat_operations ( + operation_id, principal_id, idempotency_key, token_hash, + payload_hash, state, revision, execution_input_json, + result_json, created_at, updated_at + ) + SELECT + operation_id, principal_id, idempotency_key, token_hash, + payload_hash, state, revision, execution_input_json, + result_json, created_at, updated_at + FROM adcp_compat_operations_old + """ + ) + conn.execute("DROP TABLE adcp_compat_operations_old") + + def _ensure_private_parent_directory(self) -> None: + """Create and validate the directory used for SQLite pathname opens.""" + + parent = self.path.parent + chain = list(reversed(parent.parents)) + [parent] + for directory in chain: + try: + directory_status = directory.lstat() + except FileNotFoundError: + directory.mkdir(mode=0o700) + directory_status = directory.lstat() + except OSError as exc: + raise PermissionError(f"SQLite directory {directory} is not accessible") from exc + if not stat.S_ISDIR(directory_status.st_mode): + raise PermissionError( + f"SQLite directory {directory} must be a real directory, not a symlink" + ) + + try: + direct_status = parent.lstat() + except OSError as exc: + raise PermissionError(f"SQLite parent directory {parent} is not accessible") from exc + if not stat.S_ISDIR(direct_status.st_mode): + raise PermissionError(f"SQLite parent directory {parent} is not a directory") + if direct_status.st_uid != os.geteuid(): + raise PermissionError( + f"SQLite parent directory {parent} must be owned by the current user" + ) + direct_mode = stat.S_IMODE(direct_status.st_mode) + if direct_mode & 0o022: + raise PermissionError( + f"SQLite parent directory {parent} has mode {direct_mode:#o}; " + "remove group/world write access before opening" + ) + + for ancestor in parent.parents: + try: + ancestor_status = ancestor.lstat() + except OSError as exc: + raise PermissionError( + f"SQLite ancestor directory {ancestor} is not accessible" + ) from exc + ancestor_mode = stat.S_IMODE(ancestor_status.st_mode) + unsafe_writable = bool(ancestor_mode & 0o022) and not bool( + ancestor_status.st_mode & stat.S_ISVTX + ) + if unsafe_writable: + raise PermissionError( + f"SQLite ancestor directory {ancestor} has unsafe writable mode " + f"{ancestor_mode:#o}" + ) def _ensure_private_database_file(self) -> None: """Create the ledger as 0600 and reject an existing loose mode.""" @@ -161,6 +300,8 @@ def _ensure_private_file(path: Path, description: str) -> None: file_status = os.fstat(descriptor) if not stat.S_ISREG(file_status.st_mode): raise PermissionError(f"{description} {path} is not a regular file") + if file_status.st_uid != os.geteuid(): + raise PermissionError(f"{description} {path} must be owned by the current user") mode = stat.S_IMODE(file_status.st_mode) if mode & 0o077: raise PermissionError( @@ -173,6 +314,7 @@ def _ensure_private_file(path: Path, description: str) -> None: def _connect(self) -> sqlite3.Connection: # Re-check on every connection so sidecars removed after the previous # last close are recreated with a private mode before the next write. + self._ensure_private_parent_directory() self._ensure_private_database_file() self._ensure_private_sidecar_files() conn = sqlite3.connect(self.path, timeout=self.timeout) @@ -187,10 +329,12 @@ def _connect(self) -> sqlite3.Connection: raise async def put_continuation(self, continuation: LegacyPurchaseContinuation) -> None: - await asyncio.to_thread(self._put_continuation, continuation) + await asyncio.to_thread(self._put_continuation, copy.deepcopy(continuation)) def _put_continuation(self, value: LegacyPurchaseContinuation) -> None: now = _format_datetime(self._clock()) + if value.projected_products is None: + raise ValueError("new continuations require buyer-visible product bindings") try: with closing(self._connect()) as conn, conn: conn.execute( @@ -199,9 +343,10 @@ def _put_continuation(self, value: LegacyPurchaseContinuation) -> None: token_hash, principal_id, account_identity, source_adcp_version, expires_at, observed_request_json, observed_response_json, observed_payload_hash, - product_ids_json, losses_json, target_binding, + product_ids_json, projected_products_json, losses_json, + mutation_idempotency_guaranteed, target_binding, listed_purchase_context_json, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( value.token_hash, @@ -213,7 +358,9 @@ def _put_continuation(self, value: LegacyPurchaseContinuation) -> None: _dumps(value.observed_response), value.observed_payload_hash, _dumps(list(value.product_ids)), + _dumps(list(value.projected_products)), _dumps(sorted(value.losses)), + int(value.mutation_idempotency_guaranteed), value.target_binding, ( _dumps(value.listed_purchase_context) @@ -256,6 +403,7 @@ async def claim( principal_id: str, idempotency_key: str, payload_hash: str, + execution_input: Mapping[str, Any], now: datetime, ) -> CompatibilityPurchaseOperation: return await asyncio.to_thread( @@ -264,6 +412,7 @@ async def claim( principal_id, idempotency_key, payload_hash, + copy.deepcopy(dict(execution_input)), now, ) @@ -273,6 +422,7 @@ def _claim( principal_id: str, idempotency_key: str, payload_hash: str, + execution_input: dict[str, Any], now: datetime, ) -> CompatibilityPurchaseOperation: with closing(self._connect()) as conn, conn: @@ -293,6 +443,39 @@ def _claim( "idempotency key was already used with a different logical payload", "Use the original payload or start a new projected purchase.", ) + if not operation.execution_input: + # Pre-hardening ledgers retained the logical payload hash + # but not the sanitized execution snapshot. An exact retry + # can adopt it atomically; the revision increment fences + # every pre-migration operation object. + updated_at = _format_datetime(self._clock()) + adopted = conn.execute( + "UPDATE adcp_compat_operations " + "SET execution_input_json = ?, revision = revision + 1, updated_at = ? " + "WHERE operation_id = ? AND revision = ? " + "AND execution_input_json = '{}'", + ( + _dumps(execution_input), + updated_at, + operation.operation_id, + operation.revision, + ), + ) + if adopted.rowcount != 1: + raise _state_error("legacy execution input changed concurrently") + operation = CompatibilityPurchaseOperation( + operation_id=operation.operation_id, + principal_id=operation.principal_id, + idempotency_key=operation.idempotency_key, + token_hash=operation.token_hash, + payload_hash=operation.payload_hash, + state=operation.state, + revision=operation.revision + 1, + execution_input=copy.deepcopy(execution_input), + result=copy.deepcopy(operation.result), + ) + elif operation.execution_input != execution_input: + raise _state_error("stored execution input changed for idempotent claim") conn.commit() return operation @@ -339,8 +522,9 @@ def _claim( """ INSERT INTO adcp_compat_operations ( operation_id, principal_id, idempotency_key, token_hash, - payload_hash, state, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + payload_hash, state, revision, execution_input_json, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( operation_id, @@ -349,6 +533,8 @@ def _claim( token_hash, payload_hash, CompatibilityOperationState.CLAIMED.value, + 1, + _dumps(execution_input), _format_datetime(claim_time), _format_datetime(claim_time), ), @@ -361,14 +547,32 @@ def _claim( token_hash=token_hash, payload_hash=payload_hash, state=CompatibilityOperationState.CLAIMED, + revision=1, + execution_input=copy.deepcopy(execution_input), ) + async def get_operation( + self, operation_id: str, *, principal_id: str + ) -> CompatibilityPurchaseOperation | None: + return await asyncio.to_thread(self._get_operation, operation_id, principal_id) + + def _get_operation( + self, operation_id: str, principal_id: str + ) -> CompatibilityPurchaseOperation | None: + with closing(self._connect()) as conn: + row = conn.execute( + "SELECT * FROM adcp_compat_operations " + "WHERE operation_id = ? AND principal_id = ?", + (operation_id, principal_id), + ).fetchone() + return _decode_operation(row) if row is not None else None + async def mark_in_flight( self, operation: CompatibilityPurchaseOperation ) -> CompatibilityPurchaseOperation: return await asyncio.to_thread( self._transition, - operation, + copy.deepcopy(operation), {CompatibilityOperationState.CLAIMED}, CompatibilityOperationState.IN_FLIGHT, None, @@ -379,7 +583,7 @@ async def mark_ambiguous( ) -> CompatibilityPurchaseOperation: return await asyncio.to_thread( self._transition, - operation, + copy.deepcopy(operation), {CompatibilityOperationState.CLAIMED, CompatibilityOperationState.IN_FLIGHT}, CompatibilityOperationState.AMBIGUOUS, None, @@ -389,13 +593,36 @@ async def complete( self, operation: CompatibilityPurchaseOperation, result: Mapping[str, Any], + *, + state: CompatibilityOperationState = CompatibilityOperationState.SUCCEEDED, ) -> CompatibilityPurchaseOperation: + if state not in { + CompatibilityOperationState.PENDING, + CompatibilityOperationState.SUCCEEDED, + CompatibilityOperationState.FAILED, + }: + raise ValueError("complete state must be pending, succeeded, or failed") return await asyncio.to_thread( self._transition, - operation, - {CompatibilityOperationState.IN_FLIGHT, CompatibilityOperationState.AMBIGUOUS}, - CompatibilityOperationState.SUCCEEDED, - dict(result), + copy.deepcopy(operation), + { + CompatibilityOperationState.IN_FLIGHT, + CompatibilityOperationState.AMBIGUOUS, + CompatibilityOperationState.PENDING, + }, + state, + copy.deepcopy(dict(result)), + ) + + async def fence_in_flight( + self, operation: CompatibilityPurchaseOperation + ) -> CompatibilityPurchaseOperation: + return await asyncio.to_thread( + self._transition, + copy.deepcopy(operation), + {CompatibilityOperationState.IN_FLIGHT}, + CompatibilityOperationState.AMBIGUOUS, + None, ) async def resume_after_not_applied( @@ -403,7 +630,7 @@ async def resume_after_not_applied( ) -> CompatibilityPurchaseOperation: return await asyncio.to_thread( self._transition, - operation, + copy.deepcopy(operation), {CompatibilityOperationState.AMBIGUOUS}, CompatibilityOperationState.CLAIMED, None, @@ -430,8 +657,11 @@ def _transition( or current.idempotency_key != operation.idempotency_key or current.token_hash != operation.token_hash or current.payload_hash != operation.payload_hash + or current.execution_input != operation.execution_input ): raise _state_error("operation binding changed in continuation store") + if current.revision != operation.revision: + raise _state_error("operation revision changed concurrently") if current.state not in allowed: raise _state_error( f"cannot transition operation from {current.state.value} to {target.value}" @@ -441,8 +671,8 @@ def _transition( updated = conn.execute( """ UPDATE adcp_compat_operations - SET state = ?, result_json = ?, updated_at = ? - WHERE operation_id = ? AND state = ? + SET state = ?, result_json = ?, revision = revision + 1, updated_at = ? + WHERE operation_id = ? AND state = ? AND revision = ? """, ( target.value, @@ -450,6 +680,7 @@ def _transition( updated_at, current.operation_id, current.state.value, + current.revision, ), ) if updated.rowcount != 1: @@ -462,68 +693,133 @@ def _transition( token_hash=current.token_hash, payload_hash=current.payload_hash, state=target, - result=result, + revision=current.revision + 1, + execution_input=copy.deepcopy(current.execution_input), + result=copy.deepcopy(result), ) async def purge_resolved_before(self, cutoff: datetime) -> int: - """Delete only old succeeded or never-claimed continuations. + """Delete only old terminal or never-claimed continuations. - Claimed, in-flight, and ambiguous operations are deliberately retained - regardless of age because deleting them could permit an unsafe replay. - Returns the number of continuation records removed. + Claimed, in-flight, pending, and ambiguous operations are deliberately + retained regardless of age because deleting them could permit an unsafe + replay. Returns the number of continuation records removed. """ return await asyncio.to_thread(self._purge_resolved_before, cutoff) def _purge_resolved_before(self, cutoff: datetime) -> int: cutoff_utc = _as_utc(cutoff) - with closing(self._connect()) as conn, conn: - conn.execute("BEGIN IMMEDIATE") + with closing(self._connect()) as conn: rows = conn.execute( """ SELECT continuation.token_hash, continuation.expires_at, continuation.updated_at AS continuation_updated_at, + operation.operation_id, operation.state, operation.updated_at AS operation_updated_at FROM adcp_compat_continuations AS continuation LEFT JOIN adcp_compat_operations AS operation ON operation.token_hash = continuation.token_hash - WHERE operation.state = 'succeeded' + WHERE operation.state IN ('succeeded', 'failed') OR operation.operation_id IS NULL """, ).fetchall() - token_hashes = [ + candidates = { + row["token_hash"]: ( + row["expires_at"], + row["continuation_updated_at"], + row["operation_id"], + row["state"], + row["operation_updated_at"], + ) + for row in rows + if ( + row["state"] + in { + CompatibilityOperationState.SUCCEEDED.value, + CompatibilityOperationState.FAILED.value, + } + and _parse_datetime(row["operation_updated_at"]) < cutoff_utc + ) + or ( + row["state"] is None + and _parse_datetime(row["expires_at"]) < cutoff_utc + and _parse_datetime(row["continuation_updated_at"]) < cutoff_utc + ) + } + if not candidates: + return 0 + + # Candidate scanning and timestamp parsing happen without a write lock. + # Each small write transaction then compares the raw values again, so a + # newly claimed or otherwise updated row cannot be purged from a stale + # scan. + deleted = 0 + token_hashes = list(candidates) + for start in range(0, len(token_hashes), 200): + batch = token_hashes[start : start + 200] + deleted += self._purge_candidate_batch(batch, candidates) + return deleted + + def _purge_candidate_batch( + self, + token_hashes: list[str], + candidates: Mapping[str, tuple[str, str, str | None, str | None, str | None]], + ) -> int: + placeholders = ",".join("?" for _ in token_hashes) + with closing(self._connect()) as conn, conn: + conn.execute("BEGIN IMMEDIATE") + rows = conn.execute( + f""" + SELECT + continuation.token_hash, + continuation.expires_at, + continuation.updated_at AS continuation_updated_at, + operation.operation_id, + operation.state, + operation.updated_at AS operation_updated_at + FROM adcp_compat_continuations AS continuation + LEFT JOIN adcp_compat_operations AS operation + ON operation.token_hash = continuation.token_hash + WHERE continuation.token_hash IN ({placeholders}) + """, + token_hashes, + ).fetchall() + confirmed = [ row["token_hash"] for row in rows - if ( - row["state"] == CompatibilityOperationState.SUCCEEDED.value - and _parse_datetime(row["operation_updated_at"]) < cutoff_utc - ) - or ( - row["state"] is None - and _parse_datetime(row["expires_at"]) < cutoff_utc - and _parse_datetime(row["continuation_updated_at"]) < cutoff_utc + if candidates.get(row["token_hash"]) + == ( + row["expires_at"], + row["continuation_updated_at"], + row["operation_id"], + row["state"], + row["operation_updated_at"], ) ] - if not token_hashes: + if not confirmed: return 0 - placeholders = ",".join("?" for _ in token_hashes) + confirmed_placeholders = ",".join("?" for _ in confirmed) conn.execute( - f"DELETE FROM adcp_compat_operations WHERE token_hash IN ({placeholders})", - token_hashes, + f"DELETE FROM adcp_compat_operations " + f"WHERE token_hash IN ({confirmed_placeholders})", + confirmed, ) - deleted = conn.execute( - f"DELETE FROM adcp_compat_continuations WHERE token_hash IN ({placeholders})", - token_hashes, + removed = conn.execute( + f"DELETE FROM adcp_compat_continuations " + f"WHERE token_hash IN ({confirmed_placeholders})", + confirmed, ).rowcount conn.commit() - return deleted + return removed def _decode_continuation(row: sqlite3.Row) -> LegacyPurchaseContinuation: listed = _loads(row["listed_purchase_context_json"]) + projected = _loads(row["projected_products_json"]) return LegacyPurchaseContinuation( token_hash=row["token_hash"], principal_id=row["principal_id"], @@ -534,7 +830,13 @@ def _decode_continuation(row: sqlite3.Row) -> LegacyPurchaseContinuation: observed_response=_require_object(_loads(row["observed_response_json"])), observed_payload_hash=row["observed_payload_hash"], product_ids=tuple(_loads(row["product_ids_json"])), + projected_products=( + tuple(_require_object(product) for product in projected) + if projected is not None + else None + ), losses=frozenset(_loads(row["losses_json"])), + mutation_idempotency_guaranteed=bool(row["mutation_idempotency_guaranteed"]), target_binding=row["target_binding"], listed_purchase_context=_require_object(listed) if listed is not None else None, ) @@ -549,6 +851,8 @@ def _decode_operation(row: sqlite3.Row) -> CompatibilityPurchaseOperation: token_hash=row["token_hash"], payload_hash=row["payload_hash"], state=CompatibilityOperationState(row["state"]), + revision=row["revision"], + execution_input=_require_object(_loads(row["execution_input_json"])), result=_require_object(result) if result is not None else None, ) diff --git a/tests/test_purchase_continuation.py b/tests/test_purchase_continuation.py index 5c095d37..87d9ed73 100644 --- a/tests/test_purchase_continuation.py +++ b/tests/test_purchase_continuation.py @@ -4,6 +4,7 @@ import asyncio import copy +import hashlib import json import os import sqlite3 @@ -27,6 +28,7 @@ SqliteCompatibilityContinuationStore, ) from adcp.types import CompatibilityPurchaseCoordinatorInput +from adcp.types.core import TaskResult, TaskStatus from adcp.validation import get_bundle_adcp_version, validate_response _VECTORS = ( @@ -62,6 +64,23 @@ def _coordinator(store: Any, executor: Any, *, reconciler: Any = None) -> Legacy ) +def _success_result(source_version: str, media_buy_id: str) -> dict[str, Any]: + if source_version.startswith("2.5."): + return {"media_buy_id": media_buy_id, "buyer_ref": "buyer-ref", "packages": []} + if source_version.startswith("3.1."): + return { + "media_buy_id": media_buy_id, + "confirmed_at": "2098-01-01T00:00:00Z", + "revision": 1, + "packages": [], + } + return {"media_buy_id": media_buy_id, "packages": []} + + +def _success_for(ctx: Any, media_buy_id: str) -> dict[str, Any]: + return _success_result(ctx.source_adcp_version, media_buy_id) + + async def _issue( coordinator: LegacyPurchaseCoordinator, case: dict[str, Any], @@ -80,8 +99,10 @@ async def _issue( observed_request=case["legacy_request"], observed_response=case["legacy_response"], product_ids=continuation["product_ids"], + buyer_visible_products=case["compact_projection"]["products"], losses=continuation["losses"], target_binding=target, + mutation_idempotency_guaranteed=not case["source_version"].startswith("2.5."), ) case["continuation_input"]["continuation_token"] = token @@ -93,7 +114,7 @@ async def test_upstream_vectors_execute_and_replay(case: dict[str, Any]) -> None async def execute(ctx: Any) -> dict[str, Any]: calls.append(ctx) - return {"media_buy_id": f"mb-{ctx.selected_product_ids[0]}"} + return _success_for(ctx, f"mb-{ctx.selected_product_ids[0]}") coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), execute) await _issue(coordinator, case) @@ -121,12 +142,12 @@ async def test_concurrent_exact_retries_execute_once() -> None: release = asyncio.Event() calls = 0 - async def execute(_ctx: Any) -> dict[str, Any]: + async def execute(ctx: Any) -> dict[str, Any]: nonlocal calls calls += 1 entered.set() await release.wait() - return {"media_buy_id": "mb-once"} + return _success_for(ctx, "mb-once") coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), execute) await _issue(coordinator, case) @@ -162,12 +183,12 @@ async def test_in_flight_not_applied_reconciliation_cannot_reopen_live_executor( release = asyncio.Event() calls = 0 - async def execute(_ctx: Any) -> dict[str, Any]: + async def execute(ctx: Any) -> dict[str, Any]: nonlocal calls calls += 1 entered.set() await release.wait() - return {"media_buy_id": "mb-once"} + return _success_for(ctx, "mb-once") async def reconcile(_ctx: Any, _operation: Any) -> ReconciliationResult: return ReconciliationResult.not_applied() @@ -191,7 +212,7 @@ async def reconcile(_ctx: Any, _operation: Any) -> ReconciliationResult: target_binding="seller-session-acme", ) release.set() - assert await first == {"media_buy_id": "mb-once"} + assert await first == _success_result(case["source_version"], "mb-once") assert exc.value.code == CompatibilityContinuationErrorCode.AMBIGUOUS_MUTATION assert calls == 1 @@ -201,10 +222,10 @@ async def test_different_idempotency_keys_cannot_double_claim() -> None: case = _cases()[1] calls = 0 - async def execute(_ctx: Any) -> dict[str, Any]: + async def execute(ctx: Any) -> dict[str, Any]: nonlocal calls calls += 1 - return {"media_buy_id": "mb-once"} + return _success_for(ctx, "mb-once") coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), execute) await _issue(coordinator, case) @@ -235,7 +256,7 @@ async def test_exception_is_ambiguous_and_never_blindly_retried() -> None: case = _cases()[2] calls = 0 - async def execute(_ctx: Any) -> dict[str, Any]: + async def execute(ctx: Any) -> dict[str, Any]: nonlocal calls calls += 1 raise TimeoutError("response lost after possible commit") @@ -274,7 +295,9 @@ async def execute(_ctx: Any) -> dict[str, Any]: ) async def reconcile(_ctx: Any, _operation: Any) -> ReconciliationResult: - return ReconciliationResult.applied({"media_buy_id": "mb-reconciled"}) + return ReconciliationResult.applied( + _success_result(case["source_version"], "mb-reconciled") + ) recovered = _coordinator(store, execute, reconciler=reconcile) result = await recovered.continue_legacy_purchase( @@ -287,7 +310,7 @@ async def reconcile(_ctx: Any, _operation: Any) -> ReconciliationResult: principal_id="principal-acme", target_binding="seller-session-acme", ) - assert result == replay == {"media_buy_id": "mb-reconciled"} + assert result == replay == _success_result(case["source_version"], "mb-reconciled") assert calls == 1 @@ -296,17 +319,17 @@ async def test_authoritatively_not_applied_resumes_ambiguous_operation() -> None case = _cases()[2] calls = 0 - async def execute(_ctx: Any) -> dict[str, Any]: + async def execute(ctx: Any) -> dict[str, Any]: nonlocal calls calls += 1 if calls == 1: raise TimeoutError - return {"media_buy_id": "mb-resumed"} + return _success_for(ctx, "mb-resumed") store = InMemoryCompatibilityContinuationStore() initial = _coordinator(store, execute) await _issue(initial, case) - with pytest.raises(CompatibilityContinuationError): + with pytest.raises(CompatibilityContinuationError) as initial_error: await initial.continue_legacy_purchase( case["continuation_input"], principal_id="principal-acme", @@ -317,25 +340,75 @@ async def reconcile(_ctx: Any, _operation: Any) -> ReconciliationResult: return ReconciliationResult.not_applied() recovered = _coordinator(store, execute, reconciler=reconcile) - result = await recovered.continue_legacy_purchase( - case["continuation_input"], + snapshot = await recovered.get_legacy_purchase_operation( + initial_error.value.details["operation_id"], + principal_id="principal-acme", + ) + result = await recovered.recover_legacy_purchase( + snapshot, principal_id="principal-acme", target_binding="seller-session-acme", ) - assert result == {"media_buy_id": "mb-resumed"} + assert result == _success_result(case["source_version"], "mb-resumed") assert calls == 2 +@pytest.mark.asyncio +async def test_public_recovery_fences_in_flight_with_revision_cas() -> None: + case = copy.deepcopy(_cases()[2]) + store = InMemoryCompatibilityContinuationStore() + + async def execute(ctx: Any) -> dict[str, Any]: + return _success_for(ctx, "mb-recovered-after-crash") + + async def reconcile(_ctx: Any, _operation: Any) -> ReconciliationResult: + return ReconciliationResult.not_applied() + + coordinator = _coordinator(store, execute, reconciler=reconcile) + await _issue(coordinator, case) + token = case["continuation_input"]["continuation_token"] + execution_input = { + key: copy.deepcopy(value) + for key, value in case["continuation_input"].items() + if key != "continuation_token" + } + claimed = await store.claim( + hashlib.sha256(token.encode()).hexdigest(), + principal_id="principal-acme", + idempotency_key=case["continuation_input"]["idempotency_key"], + payload_hash="simulated-hard-loss-payload", + execution_input=execution_input, + now=_NOW, + ) + in_flight = await store.mark_in_flight(claimed) + snapshot = await coordinator.get_legacy_purchase_operation( + in_flight.operation_id, principal_id="principal-acme" + ) + result = await coordinator.recover_legacy_purchase( + snapshot, + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert result == _success_result(case["source_version"], "mb-recovered-after-crash") + with pytest.raises(CompatibilityContinuationError) as stale: + await coordinator.recover_legacy_purchase( + snapshot, + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert stale.value.code == CompatibilityContinuationErrorCode.STORE_CONFLICT + + @pytest.mark.asyncio async def test_sqlite_store_survives_restart_and_replays(tmp_path: Path) -> None: case = _cases()[1] database = tmp_path / "continuations.sqlite3" calls = 0 - async def execute(_ctx: Any) -> dict[str, Any]: + async def execute(ctx: Any) -> dict[str, Any]: nonlocal calls calls += 1 - return {"media_buy_id": "mb-durable"} + return _success_for(ctx, "mb-durable") first = _coordinator(SqliteCompatibilityContinuationStore(database), execute) await _issue(first, case) @@ -354,7 +427,7 @@ async def must_not_execute(_ctx: Any) -> dict[str, Any]: principal_id="principal-acme", target_binding="seller-session-acme", ) - assert result == replay == {"media_buy_id": "mb-durable"} + assert result == replay == _success_result(case["source_version"], "mb-durable") assert calls == 1 @@ -366,12 +439,12 @@ async def test_sqlite_stores_share_one_atomic_claim(tmp_path: Path) -> None: release = asyncio.Event() calls = 0 - async def execute(_ctx: Any) -> dict[str, Any]: + async def execute(ctx: Any) -> dict[str, Any]: nonlocal calls calls += 1 entered.set() await release.wait() - return {"media_buy_id": "mb-sqlite-once"} + return _success_for(ctx, "mb-sqlite-once") first = _coordinator(SqliteCompatibilityContinuationStore(database), execute) second = _coordinator(SqliteCompatibilityContinuationStore(database), execute) @@ -393,7 +466,7 @@ async def execute(_ctx: Any) -> dict[str, Any]: target_binding="seller-session-acme", ) release.set() - assert await first_task == {"media_buy_id": "mb-sqlite-once"} + assert await first_task == _success_result(case["source_version"], "mb-sqlite-once") assert exc.value.code == CompatibilityContinuationErrorCode.ALREADY_CLAIMED assert calls == 1 @@ -404,7 +477,7 @@ async def test_cancellation_leaves_operation_ambiguous() -> None: entered = asyncio.Event() calls = 0 - async def execute(_ctx: Any) -> dict[str, Any]: + async def execute(ctx: Any) -> dict[str, Any]: nonlocal calls calls += 1 entered.set() @@ -434,6 +507,343 @@ async def execute(_ctx: Any) -> dict[str, Any]: assert calls == 1 +@pytest.mark.asyncio +async def test_cancellation_after_reservation_commit_never_calls_executor() -> None: + case = _cases()[2] + + class DelayedReservationStore(InMemoryCompatibilityContinuationStore): + def __init__(self) -> None: + super().__init__() + self.committed = asyncio.Event() + self.release = asyncio.Event() + self.reserved_operation: Any = None + + async def mark_in_flight(self, operation: Any) -> Any: + reserved = await super().mark_in_flight(operation) + self.reserved_operation = reserved + self.committed.set() + await self.release.wait() + return reserved + + calls = 0 + + async def execute(_ctx: Any) -> dict[str, Any]: + nonlocal calls + calls += 1 + return {} + + store = DelayedReservationStore() + coordinator = _coordinator(store, execute) + await _issue(coordinator, case) + task = asyncio.create_task( + coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + ) + await store.committed.wait() + task.cancel() + store.release.set() + with pytest.raises(asyncio.CancelledError): + await task + operation = await store.get_operation( + store.reserved_operation.operation_id, principal_id="principal-acme" + ) + assert operation is not None + assert operation.state.value == "ambiguous" + assert calls == 0 + + +@pytest.mark.asyncio +async def test_invalid_executor_result_is_not_persisted_as_success() -> None: + case = _cases()[2] + coordinator = _coordinator( + InMemoryCompatibilityContinuationStore(), lambda _ctx: {"media_buy_id": "invalid"} + ) + await _issue(coordinator, case) + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_LEGACY_RESPONSE + assert exc.value.details["operation_id"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "result", + [ + {"status": "submitted", "task_id": "task-123"}, + {"status": "working", "task_id": "task-working", "percentage": 25}, + { + "status": "input-required", + "task_id": "task-input", + "reason": "APPROVAL_REQUIRED", + "errors": [{"code": "APPROVAL_REQUIRED", "message": "Approve purchase"}], + }, + {"errors": [{"code": "BUDGET_TOO_LOW", "message": "Increase budget"}]}, + TaskResult( + status=TaskStatus.SUBMITTED, + submitted={ + "webhook_url": "https://buyer.example/webhook", + "operation_id": "task-wrapper-123", + }, + ), + ], +) +async def test_non_success_legacy_results_are_validated_and_replayed(result: Any) -> None: + case = _cases()[2] + calls = 0 + + def execute(_ctx: Any) -> Any: + nonlocal calls + calls += 1 + return result + + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), execute) + await _issue(coordinator, case) + first = await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + replay = await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert first == replay + assert calls == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("case", _cases(), ids=lambda c: c["source_version"]) +@pytest.mark.parametrize("arm", ["submitted", "errors"]) +async def test_completed_task_result_accepts_any_valid_legacy_arm( + case: dict[str, Any], arm: str +) -> None: + data = ( + {"status": "submitted", "task_id": "task-completed-wrapper"} + if arm == "submitted" + else {"errors": [{"code": "INVALID_REQUEST", "message": "Rejected"}]} + ) + calls = 0 + + def execute(_ctx: Any) -> TaskResult[Any]: + nonlocal calls + calls += 1 + return TaskResult( + status=TaskStatus.COMPLETED, + success=arm != "errors", + data=data, + ) + + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), execute) + await _issue(coordinator, case) + first = await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + replay = await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert first == replay == data + assert calls == 1 + + +@pytest.mark.asyncio +async def test_task_result_status_must_match_validated_payload_arm() -> None: + case = _cases()[2] + invalid = TaskResult( + status=TaskStatus.FAILED, + success=False, + data=_success_result(case["source_version"], "mb-stale-success"), + ) + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), lambda _ctx: invalid) + await _issue(coordinator, case) + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_LEGACY_RESPONSE + + +@pytest.mark.asyncio +async def test_submitted_task_result_requires_task_identity() -> None: + case = _cases()[2] + coordinator = _coordinator( + InMemoryCompatibilityContinuationStore(), + lambda _ctx: TaskResult(status=TaskStatus.SUBMITTED), + ) + await _issue(coordinator, case) + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_LEGACY_RESPONSE + + +@pytest.mark.asyncio +async def test_synchronous_executor_runs_off_event_loop_thread() -> None: + case = _cases()[2] + event_loop_thread = threading.get_ident() + executor_thread: int | None = None + + def execute(ctx: Any) -> dict[str, Any]: + nonlocal executor_thread + executor_thread = threading.get_ident() + return _success_for(ctx, "mb-threaded") + + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), execute) + await _issue(coordinator, case) + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert executor_thread is not None + assert executor_thread != event_loop_thread + + +@pytest.mark.asyncio +async def test_buyer_visible_pricing_projection_is_authoritative() -> None: + case = copy.deepcopy(_cases()[2]) + hidden = copy.deepcopy(case["legacy_response"]["products"][0]["pricing_options"][0]) + hidden["pricing_option_id"] = "hidden-option" + case["legacy_response"]["products"][0]["pricing_options"].append(hidden) + case["continuation_input"]["legacy_create_request"]["packages"][0][ + "pricing_option_id" + ] = "hidden-option" + calls = 0 + + def execute(_ctx: Any) -> dict[str, Any]: + nonlocal calls + calls += 1 + return {} + + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), execute) + await _issue(coordinator, case) + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.BINDING_MISMATCH + assert calls == 0 + + +@pytest.mark.asyncio +async def test_buyer_visible_pricing_terms_must_match_observed_terms() -> None: + case = copy.deepcopy(_cases()[2]) + case["compact_projection"]["products"][0]["pricing_options"][0]["fixed_price"] = 1 + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), lambda _ctx: {}) + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(coordinator, case) + assert exc.value.code == CompatibilityContinuationErrorCode.BINDING_MISMATCH + + +@pytest.mark.asyncio +async def test_buyer_visible_pricing_cannot_omit_observed_commercial_terms() -> None: + case = copy.deepcopy(_cases()[2]) + del case["compact_projection"]["products"][0]["pricing_options"][0]["fixed_price"] + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), lambda _ctx: {}) + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(coordinator, case) + assert exc.value.code == CompatibilityContinuationErrorCode.BINDING_MISMATCH + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("projected_value", "observed_value"), + [(1, 2), (True, None)], +) +async def test_buyer_visible_pricing_extensions_fail_closed( + projected_value: Any, observed_value: Any +) -> None: + case = copy.deepcopy(_cases()[2]) + projected = case["compact_projection"]["products"][0]["pricing_options"][0] + observed = case["legacy_response"]["products"][0]["pricing_options"][0] + key = "x-billing-multiplier" if observed_value is not None else "max_bid" + projected[key] = projected_value + if observed_value is not None: + observed[key] = observed_value + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), lambda _ctx: {}) + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(coordinator, case) + assert exc.value.code == CompatibilityContinuationErrorCode.BINDING_MISMATCH + + +@pytest.mark.asyncio +async def test_31_pricing_does_not_apply_25_field_normalization() -> None: + case = copy.deepcopy(_cases()[2]) + observed = case["legacy_response"]["products"][0]["pricing_options"][0] + observed["rate"] = 999 + observed["is_fixed"] = True + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), lambda _ctx: {}) + with pytest.raises(CompatibilityContinuationError) as exc: + await _issue(coordinator, case) + assert exc.value.code == CompatibilityContinuationErrorCode.BINDING_MISMATCH + + +@pytest.mark.asyncio +async def test_31_without_replay_guarantee_requires_mutation_loss() -> None: + case = copy.deepcopy(_cases()[2]) + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), lambda _ctx: {}) + continuation = case["compact_projection"]["purchase_continuation"] + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.issue_legacy_create_continuation( + principal_id="principal-acme", + account=case["continuation_input"]["account"], + source_adcp_version=case["source_version"], + expires_at=datetime.fromisoformat( + continuation["continuation_expires_at"].replace("Z", "+00:00") + ), + observed_request=case["legacy_request"], + observed_response=case["legacy_response"], + product_ids=continuation["product_ids"], + buyer_visible_products=case["compact_projection"]["products"], + losses=continuation["losses"], + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_INPUT + + +@pytest.mark.asyncio +@pytest.mark.parametrize("invalid", ["false", 1, None]) +async def test_replay_guarantee_requires_strict_boolean(invalid: Any) -> None: + case = copy.deepcopy(_cases()[2]) + coordinator = _coordinator(InMemoryCompatibilityContinuationStore(), lambda _ctx: {}) + continuation = case["compact_projection"]["purchase_continuation"] + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.issue_legacy_create_continuation( + principal_id="principal-acme", + account=case["continuation_input"]["account"], + source_adcp_version=case["source_version"], + expires_at=datetime.fromisoformat( + continuation["continuation_expires_at"].replace("Z", "+00:00") + ), + observed_request=case["legacy_request"], + observed_response=case["legacy_response"], + product_ids=continuation["product_ids"], + buyer_visible_products=case["compact_projection"]["products"], + losses=continuation["losses"], + target_binding="seller-session-acme", + mutation_idempotency_guaranteed=invalid, + ) + assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_INPUT + + @pytest.mark.asyncio @pytest.mark.parametrize( ("mutation", "code"), @@ -613,10 +1023,10 @@ async def test_completed_operation_replays_after_token_expiry() -> None: mutable_now = _NOW calls = 0 - async def execute(_ctx: Any) -> dict[str, Any]: + async def execute(ctx: Any) -> dict[str, Any]: nonlocal calls calls += 1 - return {"media_buy_id": "mb-before-expiry"} + return _success_for(ctx, "mb-before-expiry") coordinator = LegacyPurchaseCoordinator( store=InMemoryCompatibilityContinuationStore(), @@ -636,7 +1046,7 @@ async def execute(_ctx: Any) -> dict[str, Any]: principal_id="principal-acme", target_binding="seller-session-acme", ) - assert first == replay == {"media_buy_id": "mb-before-expiry"} + assert first == replay == _success_result(case["source_version"], "mb-before-expiry") assert calls == 1 @@ -666,6 +1076,55 @@ def test_sqlite_store_creates_private_ledger_and_rejects_loose_existing_file( SqliteCompatibilityContinuationStore(loose) +def test_sqlite_store_creates_private_parent_directory(tmp_path: Path) -> None: + parent = tmp_path / "private-ledger" + SqliteCompatibilityContinuationStore(parent / "continuations.sqlite3") + assert stat.S_IMODE(parent.stat().st_mode) == 0o700 + + +def test_sqlite_store_rejects_writable_parent_and_ancestor(tmp_path: Path) -> None: + writable_parent = tmp_path / "writable-parent" + writable_parent.mkdir(mode=0o700) + writable_parent.chmod(0o770) + try: + with pytest.raises(PermissionError, match="remove group/world write access"): + SqliteCompatibilityContinuationStore(writable_parent / "continuations.sqlite3") + finally: + writable_parent.chmod(0o700) + + writable_ancestor = tmp_path / "writable-ancestor" + writable_ancestor.mkdir(mode=0o700) + secure_parent = writable_ancestor / "secure-parent" + secure_parent.mkdir(mode=0o700) + writable_ancestor.chmod(0o777) + try: + with pytest.raises(PermissionError, match="unsafe writable mode"): + SqliteCompatibilityContinuationStore(secure_parent / "continuations.sqlite3") + finally: + writable_ancestor.chmod(0o700) + + +def test_sqlite_store_rechecks_parent_before_every_open(tmp_path: Path) -> None: + parent = tmp_path / "ledger-parent" + parent.mkdir(mode=0o700) + store = SqliteCompatibilityContinuationStore(parent / "continuations.sqlite3") + parent.chmod(0o770) + try: + with pytest.raises(PermissionError, match="remove group/world write access"): + store._connect() + finally: + parent.chmod(0o700) + + +def test_sqlite_store_rejects_symlink_path_components(tmp_path: Path) -> None: + real_parent = tmp_path / "real-parent" + real_parent.mkdir(mode=0o700) + linked_parent = tmp_path / "linked-parent" + linked_parent.symlink_to(real_parent, target_is_directory=True) + with pytest.raises(PermissionError, match="real directory"): + SqliteCompatibilityContinuationStore(linked_parent / "continuations.sqlite3") + + def test_sqlite_store_forces_private_wal_sidecars_under_permissive_umask( tmp_path: Path, ) -> None: @@ -764,6 +1223,153 @@ def synchronized_migration( assert {"created_at", "updated_at"}.issubset(columns) +@pytest.mark.asyncio +async def test_sqlite_migrates_origin_ledger_and_adopts_exact_retry_input( + tmp_path: Path, +) -> None: + case = copy.deepcopy(_cases()[0]) + database = tmp_path / "old-ledger.sqlite3" + store = SqliteCompatibilityContinuationStore(database) + + async def uncertain(_ctx: Any) -> dict[str, Any]: + raise TimeoutError + + coordinator = _coordinator(store, uncertain) + await _issue(coordinator, case) + with pytest.raises(CompatibilityContinuationError) as initial: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + + with sqlite3.connect(database) as conn: + conn.execute("ALTER TABLE adcp_compat_continuations DROP COLUMN projected_products_json") + conn.execute( + "ALTER TABLE adcp_compat_continuations " "DROP COLUMN mutation_idempotency_guaranteed" + ) + conn.execute("ALTER TABLE adcp_compat_operations DROP COLUMN revision") + conn.execute("ALTER TABLE adcp_compat_operations DROP COLUMN execution_input_json") + + migrated_store = SqliteCompatibilityContinuationStore(database) + migrated = _coordinator(migrated_store, uncertain) + pre_adoption = await migrated.get_legacy_purchase_operation( + initial.value.details["operation_id"], principal_id="principal-acme" + ) + with pytest.raises(CompatibilityContinuationError) as missing_snapshot: + await migrated.recover_legacy_purchase( + pre_adoption, + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert missing_snapshot.value.code == CompatibilityContinuationErrorCode.STORE_CONFLICT + with pytest.raises(CompatibilityContinuationError) as replay: + await migrated.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert replay.value.code == CompatibilityContinuationErrorCode.AMBIGUOUS_MUTATION + operation = await migrated.get_legacy_purchase_operation( + initial.value.details["operation_id"], principal_id="principal-acme" + ) + assert operation.execution_input["legacy_create_request"] + assert operation.revision == 2 + with sqlite3.connect(database) as conn: + projected = conn.execute( + "SELECT projected_products_json FROM adcp_compat_continuations" + ).fetchone()[0] + assert projected is None + + +@pytest.mark.asyncio +async def test_migrated_unclaimed_token_cannot_redeem_unbound_hidden_option( + tmp_path: Path, +) -> None: + case = copy.deepcopy(_cases()[2]) + hidden = copy.deepcopy(case["legacy_response"]["products"][0]["pricing_options"][0]) + hidden["pricing_option_id"] = "seller-only-option" + case["legacy_response"]["products"][0]["pricing_options"].append(hidden) + database = tmp_path / "old-unclaimed-ledger.sqlite3" + calls = 0 + + def execute(_ctx: Any) -> dict[str, Any]: + nonlocal calls + calls += 1 + return {} + + coordinator = _coordinator(SqliteCompatibilityContinuationStore(database), execute) + await _issue(coordinator, case) + with sqlite3.connect(database) as conn: + conn.execute("ALTER TABLE adcp_compat_continuations DROP COLUMN projected_products_json") + restarted = _coordinator(SqliteCompatibilityContinuationStore(database), execute) + case["continuation_input"]["legacy_create_request"]["packages"][0][ + "pricing_option_id" + ] = "seller-only-option" + with pytest.raises(CompatibilityContinuationError) as exc: + await restarted.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_INPUT + assert calls == 0 + + +@pytest.mark.asyncio +async def test_migrated_31_ledger_without_bound_replay_guarantee_fails_closed( + tmp_path: Path, +) -> None: + case = copy.deepcopy(_cases()[2]) + database = tmp_path / "old-31-ledger.sqlite3" + coordinator = _coordinator(SqliteCompatibilityContinuationStore(database), lambda _ctx: {}) + await _issue(coordinator, case) + with sqlite3.connect(database) as conn: + conn.execute("UPDATE adcp_compat_continuations " "SET mutation_idempotency_guaranteed = 0") + with pytest.raises(CompatibilityContinuationError) as exc: + await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert exc.value.code == CompatibilityContinuationErrorCode.INVALID_INPUT + + +@pytest.mark.asyncio +async def test_migrated_31_terminal_result_still_replays_without_mutation( + tmp_path: Path, +) -> None: + case = copy.deepcopy(_cases()[2]) + database = tmp_path / "old-31-terminal-ledger.sqlite3" + calls = 0 + + def execute(ctx: Any) -> dict[str, Any]: + nonlocal calls + calls += 1 + return _success_for(ctx, "mb-before-migration") + + coordinator = _coordinator(SqliteCompatibilityContinuationStore(database), execute) + await _issue(coordinator, case) + expected = await coordinator.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + with sqlite3.connect(database) as conn: + conn.execute("ALTER TABLE adcp_compat_continuations DROP COLUMN projected_products_json") + conn.execute( + "ALTER TABLE adcp_compat_continuations " "DROP COLUMN mutation_idempotency_guaranteed" + ) + restarted = _coordinator(SqliteCompatibilityContinuationStore(database), execute) + replay = await restarted.continue_legacy_purchase( + case["continuation_input"], + principal_id="principal-acme", + target_binding="seller-session-acme", + ) + assert replay == expected + assert calls == 1 + + @pytest.mark.asyncio async def test_sqlite_store_never_persists_raw_bearer_token(tmp_path: Path) -> None: case = _cases()[1] @@ -787,7 +1393,7 @@ async def test_sqlite_cleanup_retains_unresolved_operations(tmp_path: Path) -> N store = SqliteCompatibilityContinuationStore(database, clock=lambda: mutable_now) completed_case = copy.deepcopy(_cases()[1]) - completed = _coordinator(store, lambda _ctx: {"media_buy_id": "mb-complete"}) + completed = _coordinator(store, lambda ctx: _success_for(ctx, "mb-complete")) await _issue(completed, completed_case) await completed.continue_legacy_purchase( completed_case["continuation_input"], @@ -825,7 +1431,7 @@ async def test_sqlite_cleanup_compares_fractional_timestamps_chronologically( store = SqliteCompatibilityContinuationStore( tmp_path / "continuations.sqlite3", clock=lambda: updated_at ) - coordinator = _coordinator(store, lambda _ctx: {"media_buy_id": "mb-fractional"}) + coordinator = _coordinator(store, lambda ctx: _success_for(ctx, "mb-fractional")) await _issue(coordinator, case) await coordinator.continue_legacy_purchase( case["continuation_input"], @@ -845,7 +1451,7 @@ async def test_sqlite_cleanup_does_not_depend_on_sqlite_datetime_functions( store = SqliteCompatibilityContinuationStore( tmp_path / "continuations.sqlite3", clock=lambda: _NOW ) - coordinator = _coordinator(store, lambda _ctx: {"media_buy_id": "mb-portable-cleanup"}) + coordinator = _coordinator(store, lambda ctx: _success_for(ctx, "mb-portable-cleanup")) await _issue(coordinator, case) await coordinator.continue_legacy_purchase( case["continuation_input"],