Module adcp.compat

AdCP compatibility surfaces for buyers on older spec versions.

The framework natively validates against the SDK's pinned major (3.x). Buyers on pre-3 wire shapes are handled by the per-tool adapter registry in :mod:adcp.compat.legacy — see that module's docstring for the AdapterPair pattern and the JS-SDK parity notes.

AdCP 3.2 products-only brief projections use the stateful :class:LegacyPurchaseCoordinator. It lives outside the pure wire adapter registry because redemption requires durable state and crash reconciliation.

Sub-modules

adcp.compat.legacy

Per-tool adapters for buyers on legacy AdCP wire shapes …

adcp.compat.negotiated_catalog_purchase

Negotiated catalog purchase across AdCP 3.0, 3.1, and 3.2 …

adcp.compat.purchase_continuation

Durable continuation of lossy AdCP 3.2 legacy purchases …

adcp.compat.reporting_version

Release gates for reporting controls added in AdCP 3.2 beta.6.

adcp.compat.sqlite_continuation_store

SQLite durable store for :mod:adcp.compat.purchase_continuation …

Functions

def canonical_account_identity(account: Mapping[str, Any] | Any) ‑> str

Expand source code

def canonical_account_identity(account: Mapping[str, Any] | Any) -> str:
    """Return the AdCP account natural-key identity as canonical JSON.

    Mutable/display-only account fields are excluded.  In particular,
    ``operator_unit.name`` and brand governance/creative overrides do not
    participate in identity.
    """

    payload = _account_payload(account)
    if "account_id" in payload:
        identity: JsonObject = {"account_id": payload["account_id"]}
    else:
        brand = payload.get("brand")
        if not isinstance(brand, Mapping):
            raise _invalid("natural-key account requires brand")
        brand_identity: JsonObject = {"domain": brand.get("domain")}
        if brand.get("brand_id") is not None:
            brand_identity["brand_id"] = brand["brand_id"]
        if brand.get("countries") is not None:
            brand_identity["countries"] = sorted(str(v) for v in brand["countries"])
        identity = {
            "brand": brand_identity,
            "operator": payload.get("operator"),
            "sandbox": bool(payload.get("sandbox", False)),
        }
        operator_unit = payload.get("operator_unit")
        if isinstance(operator_unit, Mapping):
            identity["operator_unit_id"] = operator_unit.get("id")
        for key in ("currency", "timezone"):
            if payload.get(key) is not None:
                identity[key] = payload[key]
    try:
        return rfc8785.dumps(identity).decode("utf-8")
    except (TypeError, ValueError) as exc:
        raise _invalid("account identity is not JSON-canonicalizable") from exc

Return the AdCP account natural-key identity as canonical JSON.

Mutable/display-only account fields are excluded. In particular, operator_unit.name and brand governance/creative overrides do not participate in identity.

Classes

class CompatibilityContinuationError (code: CompatibilityContinuationErrorCode,
message: str,
*,
recovery_guidance: str,
details: Mapping[str, Any] | None = None)

Expand source code

class CompatibilityContinuationError(Exception):
    """Typed SDK-local failure raised before or around legacy execution."""

    code: CompatibilityContinuationErrorCode
    recovery_guidance: str
    details: JsonObject

    def __init__(
        self,
        code: CompatibilityContinuationErrorCode,
        message: str,
        *,
        recovery_guidance: str,
        details: Mapping[str, Any] | None = None,
    ) -> None:
        self.code = code
        self.recovery_guidance = recovery_guidance
        self.details = dict(details or {})
        super().__init__(message)

Typed SDK-local failure raised before or around legacy execution.

Ancestors

• builtins.Exception
• builtins.BaseException

Class variables

var code : CompatibilityContinuationErrorCode

var details : dict[str, typing.Any]

var recovery_guidance : str

class CompatibilityContinuationErrorCode (*args, **kwds)

Expand source code

class CompatibilityContinuationErrorCode(str, Enum):
    """SDK-local error categories for continuation failures.

    These names are not AdCP wire error codes and are never sent to a seller.
    """

    NOT_FOUND = "continuation_not_found"
    EXPIRED = "continuation_expired"
    BINDING_MISMATCH = "continuation_binding_mismatch"
    INVALID_INPUT = "invalid_continuation_input"
    INVALID_LEGACY_REQUEST = "invalid_legacy_create_request"
    LOSS_MISMATCH = "loss_acceptance_mismatch"
    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"
    PERSISTENCE_POLICY = "continuation_persistence_policy"
    STORE_QUOTA_EXCEEDED = "continuation_store_quota_exceeded"
    PENDING_RESOLUTION_REQUIRED = "pending_legacy_resolution_required"

SDK-local error categories for continuation failures.

These names are not AdCP wire error codes and are never sent to a seller.

Ancestors

• builtins.str
• enum.Enum

Class variables

var ALREADY_CLAIMED

var AMBIGUOUS_MUTATION

var BINDING_MISMATCH

var EXPIRED

var IDEMPOTENCY_CONFLICT

var INVALID_INPUT

var INVALID_LEGACY_REQUEST

var INVALID_LEGACY_RESPONSE

var LOSS_MISMATCH

var NOT_FOUND

var PENDING_RESOLUTION_REQUIRED

var PERSISTENCE_POLICY

var STORE_CONFLICT

var STORE_QUOTA_EXCEEDED

class CompatibilityContinuationStore (*args, **kwargs)

Expand source code

@runtime_checkable
class CompatibilityContinuationStore(Protocol):
    """Atomic persistence contract for compatibility continuations.

    Production implementations must set :attr:`is_durable` to ``True`` and
    make :meth:`claim` atomic across every process that can execute a purchase.
    The supplied ``now`` is only a lower bound: after taking the transaction
    lock, the store must refresh time from an authoritative clock before
    checking expiry. A claimed token is never made available again merely
    because time passed.
    """

    is_durable: ClassVar[bool]

    async def put_continuation(self, continuation: LegacyPurchaseContinuation) -> None: ...

    async def get_continuation(
        self, token_hash: str, *, principal_id: str
    ) -> LegacyPurchaseContinuation | None: ...

    async def claim(
        self,
        token_hash: str,
        *,
        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 get_operation_by_idempotency_key(
        self, idempotency_key: str, *, principal_id: str
    ) -> CompatibilityPurchaseOperation | None: ...

    async def mark_in_flight(
        self, operation: CompatibilityPurchaseOperation
    ) -> CompatibilityPurchaseOperation: ...

    async def mark_ambiguous(
        self, operation: CompatibilityPurchaseOperation
    ) -> CompatibilityPurchaseOperation: ...

    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:
        """Atomically resume only ``AMBIGUOUS`` after authoritative absence."""
        ...

Atomic persistence contract for compatibility continuations.

Production implementations must set :attr:is_durable to True and make :meth:claim atomic across every process that can execute a purchase. The supplied now is only a lower bound: after taking the transaction lock, the store must refresh time from an authoritative clock before checking expiry. A claimed token is never made available again merely because time passed.

Ancestors

• typing.Protocol
• typing.Generic

Class variables

var is_durable : ClassVar[bool]

Methods

async def claim(self,
token_hash: str,
*,
principal_id: str,
idempotency_key: str,
payload_hash: str,
execution_input: Mapping[str, Any],
now: datetime) ‑> CompatibilityPurchaseOperation

Expand source code

async def claim(
    self,
    token_hash: str,
    *,
    principal_id: str,
    idempotency_key: str,
    payload_hash: str,
    execution_input: Mapping[str, Any],
    now: datetime,
) -> CompatibilityPurchaseOperation: ...

async def complete(self,
operation: CompatibilityPurchaseOperation,
result: Mapping[str, Any],
*,
state: CompatibilityOperationState = CompatibilityOperationState.SUCCEEDED) ‑> CompatibilityPurchaseOperation

Expand source code

async def complete(
    self,
    operation: CompatibilityPurchaseOperation,
    result: Mapping[str, Any],
    *,
    state: CompatibilityOperationState = CompatibilityOperationState.SUCCEEDED,
) -> CompatibilityPurchaseOperation: ...

async def fence_in_flight(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def fence_in_flight(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation:
    """CAS ``IN_FLIGHT`` to ``AMBIGUOUS`` using the operation revision."""
    ...

CAS IN_FLIGHT to AMBIGUOUS using the operation revision.

async def get_continuation(self, token_hash: str, *, principal_id: str) ‑> LegacyPurchaseContinuation | None

Expand source code

async def get_continuation(
    self, token_hash: str, *, principal_id: str
) -> LegacyPurchaseContinuation | None: ...

async def get_operation(self, operation_id: str, *, principal_id: str) ‑> CompatibilityPurchaseOperation | None

Expand source code

async def get_operation(
    self, operation_id: str, *, principal_id: str
) -> CompatibilityPurchaseOperation | None: ...

async def get_operation_by_idempotency_key(self, idempotency_key: str, *, principal_id: str) ‑> CompatibilityPurchaseOperation | None

Expand source code

async def get_operation_by_idempotency_key(
    self, idempotency_key: str, *, principal_id: str
) -> CompatibilityPurchaseOperation | None: ...

async def mark_ambiguous(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def mark_ambiguous(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation: ...

async def mark_in_flight(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def mark_in_flight(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation: ...

async def put_continuation(self,
continuation: LegacyPurchaseContinuation) ‑> None

Expand source code

async def put_continuation(self, continuation: LegacyPurchaseContinuation) -> None: ...

async def resume_after_not_applied(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def resume_after_not_applied(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation:
    """Atomically resume only ``AMBIGUOUS`` after authoritative absence."""
    ...

Atomically resume only AMBIGUOUS after authoritative absence.

class CompatibilityOperationState (*args, **kwds)

Expand source code

class CompatibilityOperationState(str, Enum):
    """Durable operation states used by continuation stores."""

    CLAIMED = "claimed"
    IN_FLIGHT = "in_flight"
    PENDING = "pending"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    AMBIGUOUS = "ambiguous"

Durable operation states used by continuation stores.

Ancestors

• builtins.str
• enum.Enum

Class variables

var AMBIGUOUS

var CLAIMED

var FAILED

var IN_FLIGHT

var PENDING

var SUCCEEDED

class CompatibilityPurchaseOperation (operation_id: str,
principal_id: str,
idempotency_key: str,
token_hash: str,
payload_hash: str,
state: CompatibilityOperationState,
revision: int,
execution_input: JsonObject,
reserved_result_bytes: int = 0,
result: JsonObject | None = None)

Expand source code

@dataclass(frozen=True)
class CompatibilityPurchaseOperation:
    """Durable single-use operation returned by a continuation store."""

    operation_id: str
    principal_id: str
    idempotency_key: str
    token_hash: str
    payload_hash: str
    state: CompatibilityOperationState
    revision: int
    execution_input: JsonObject
    reserved_result_bytes: int = 0
    result: JsonObject | None = None

Durable single-use operation returned by a continuation store.

Instance variables

var execution_input : dict[str, typing.Any]

var idempotency_key : str

var operation_id : str

var payload_hash : str

var principal_id : str

var reserved_result_bytes : int

var result : dict[str, typing.Any] | None

var revision : int

var state : CompatibilityOperationState

var token_hash : str

class CompatibleCatalog (products: tuple[JsonObject, ...],
compatibility: MediaBuyCompatibilityReport,
raw: Any,
feed_version: str | None = None,
pricing_version: str | None = None,
next_cursor: str | None = None,
cache_scope: str | None = None,
purchase_continuation: LegacyCatalogContinuation | None = None)

Expand source code

@dataclass(frozen=True)
class CompatibleCatalog:
    """Lifecycle-neutral completed catalog page."""

    products: tuple[JsonObject, ...]
    compatibility: MediaBuyCompatibilityReport
    raw: Any
    feed_version: str | None = None
    pricing_version: str | None = None
    next_cursor: str | None = None
    cache_scope: str | None = None
    purchase_continuation: LegacyCatalogContinuation | None = None
    _account: JsonObject | None = None
    _coordinator_identity: int = 0
    _source_schema_version: str | None = None

Lifecycle-neutral completed catalog page.

Instance variables

var cache_scope : str | None

var compatibility : MediaBuyCompatibilityReport

var feed_version : str | None

var next_cursor : str | None

var pricing_version : str | None

var products : tuple[dict[str, typing.Any], ...]

var purchase_continuation : LegacyCatalogContinuation | None

var raw : Any

class CompatiblePurchaseResult (result: TaskResult[Any] | JsonObject,
compatibility: MediaBuyCompatibilityReport)

Expand source code

@dataclass(frozen=True)
class CompatiblePurchaseResult:
    """Lifecycle-neutral wrapper around a native or established purchase result."""

    result: TaskResult[Any] | JsonObject
    compatibility: MediaBuyCompatibilityReport

    @property
    def success(self) -> bool:
        """Whether the selected lifecycle reported a successful purchase."""

        if isinstance(self.result, TaskResult):
            return self.result.success
        return "errors" not in self.result and self.result.get("status") != "failed"

    @property
    def status(self) -> TaskStatus:
        """Normalized terminal status for ordinary direct purchases."""

        if isinstance(self.result, TaskResult):
            return self.result.status
        status = self.result.get("status")
        if not isinstance(status, str):
            return TaskStatus.COMPLETED if self.success else TaskStatus.FAILED
        return {
            "submitted": TaskStatus.SUBMITTED,
            "working": TaskStatus.WORKING,
            "input-required": TaskStatus.NEEDS_INPUT,
            "failed": TaskStatus.FAILED,
        }.get(status, TaskStatus.COMPLETED)

    @property
    def data(self) -> Any:
        """Lifecycle-neutral response data; ``result`` retains the SDK source."""

        if isinstance(self.result, TaskResult):
            return self.result.data
        return self.result

Lifecycle-neutral wrapper around a native or established purchase result.

Instance variables

var compatibility : MediaBuyCompatibilityReport

prop data : Any

Expand source code

@property
def data(self) -> Any:
    """Lifecycle-neutral response data; ``result`` retains the SDK source."""

    if isinstance(self.result, TaskResult):
        return self.result.data
    return self.result

Lifecycle-neutral response data; result retains the SDK source.

var result : TaskResult[Any] | dict[str, typing.Any]

prop status : TaskStatus

Expand source code

@property
def status(self) -> TaskStatus:
    """Normalized terminal status for ordinary direct purchases."""

    if isinstance(self.result, TaskResult):
        return self.result.status
    status = self.result.get("status")
    if not isinstance(status, str):
        return TaskStatus.COMPLETED if self.success else TaskStatus.FAILED
    return {
        "submitted": TaskStatus.SUBMITTED,
        "working": TaskStatus.WORKING,
        "input-required": TaskStatus.NEEDS_INPUT,
        "failed": TaskStatus.FAILED,
    }.get(status, TaskStatus.COMPLETED)

Normalized terminal status for ordinary direct purchases.

prop success : bool

Expand source code

@property
def success(self) -> bool:
    """Whether the selected lifecycle reported a successful purchase."""

    if isinstance(self.result, TaskResult):
        return self.result.success
    return "errors" not in self.result and self.result.get("status") != "failed"

Whether the selected lifecycle reported a successful purchase.

class CoordinatorBuyProductsInput (*args, **kwargs)

Expand source code

class CoordinatorBuyProductsInput(TypedDict, total=False):
    """Buyer-owned fields for :meth:`MediaBuyLifecycleCoordinator.buy_products`.

    This mirrors the buyer-owned JSON fields of ``BuyProductsRequest``. The
    coordinator owns the negotiated protocol version and injects feed/pricing
    evidence from the bound catalog result, so none of those fields are caller
    inputs.
    """

    idempotency_key: Required[str]
    account: Required[JsonMapping]
    brand: NotRequired[JsonMapping | None]
    advertiser_industry: NotRequired[str | None]
    purchases: Required[list[JsonMapping]]
    total_budget: NotRequired[JsonMapping | None]
    daily_budget_cap: NotRequired[float | None]
    budget_cap_timezone: NotRequired[str | None]
    budget_allocation: NotRequired[JsonMapping | None]
    start_time: Required[str | JsonMapping]
    end_time: Required[str]
    pacing: NotRequired[str | None]
    bidding: NotRequired[JsonMapping | None]
    paused: NotRequired[bool | None]
    purchase_order_ref: NotRequired[str | None]
    agency_estimate_number: NotRequired[str | None]
    invoice_recipient: NotRequired[JsonMapping | None]
    governance_context: NotRequired[str | None]
    push_notification_config: NotRequired[JsonMapping | None]
    reporting_webhook: NotRequired[JsonMapping | None]
    opportunity: NotRequired[JsonMapping | None]
    context: NotRequired[JsonMapping | None]
    ext: NotRequired[JsonMapping | None]

Buyer-owned fields for :meth:MediaBuyLifecycleCoordinator.buy_products().

This mirrors the buyer-owned JSON fields of BuyProductsRequest. The coordinator owns the negotiated protocol version and injects feed/pricing evidence from the bound catalog result, so none of those fields are caller inputs.

Ancestors

• builtins.dict

Class variables

var account : Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]]

var advertiser_industry : str | None

var agency_estimate_number : str | None

var bidding : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

var brand : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

var budget_allocation : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

var budget_cap_timezone : str | None

var context : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

var daily_budget_cap : float | None

var end_time : str

var ext : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

var governance_context : str | None

var idempotency_key : str

var invoice_recipient : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

var opportunity : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

var pacing : str | None

var paused : bool | None

var purchase_order_ref : str | None

var purchases : list[Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]]]

var push_notification_config : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

var reporting_webhook : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

var start_time : str | collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]]

var total_budget : collections.abc.Mapping[str, str | int | float | bool | None | list[str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]] | dict[str, str | int | float | bool | None | list[JsonValue] | dict[str, JsonValue]]] | None

class InMemoryCompatibilityContinuationStore (*, max_records: int = 20000, max_bytes: int = 67108864)

Expand source code

class InMemoryCompatibilityContinuationStore:
    """Process-local reference store for tests and development only."""

    is_durable: ClassVar[bool] = False

    def __init__(self, *, max_records: int = 20_000, max_bytes: int = 64 * 1024 * 1024) -> None:
        if type(max_records) is not int or max_records <= 0:
            raise ValueError("max_records must be a positive integer")
        if type(max_bytes) is not int or max_bytes <= 0:
            raise ValueError("max_bytes must be a positive integer")
        self._continuations: dict[str, LegacyPurchaseContinuation] = {}
        self._claimed_by: dict[str, str] = {}
        self._operations: dict[tuple[str, str], CompatibilityPurchaseOperation] = {}
        self._lock = asyncio.Lock()
        self._clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc)
        self._max_records = max_records
        self._max_bytes = max_bytes

    async def put_continuation(self, continuation: LegacyPurchaseContinuation) -> None:
        async with self._lock:
            copied = _copy_continuation(continuation)
            _validate_continuation_persistence(copied)
            if any(
                value.issuance_fingerprint is None
                and value.principal_id == copied.principal_id
                and value.observed_payload_hash == copied.observed_payload_hash
                and value.account_identity == copied.account_identity
                and value.target_binding == copied.target_binding
                for value in self._continuations.values()
            ):
                raise _error(
                    CompatibilityContinuationErrorCode.STORE_CONFLICT,
                    "equivalent pre-migration authorization requires operator resolution",
                    "Resolve or quarantine the legacy continuation before reissuing.",
                )
            existing = self._continuations.get(continuation.token_hash)
            by_fingerprint = next(
                (
                    value
                    for value in self._continuations.values()
                    if value.issuance_fingerprint is not None
                    and value.issuance_fingerprint == continuation.issuance_fingerprint
                    and value.principal_id == continuation.principal_id
                ),
                None,
            )
            if existing == copied and (by_fingerprint is None or by_fingerprint == existing):
                return
            if existing is not None or by_fingerprint is not None:
                raise _error(
                    CompatibilityContinuationErrorCode.STORE_CONFLICT,
                    "continuation issuance fingerprint is already registered differently",
                    "Use the same token derivation key and exact issuance inputs.",
                )
            self._check_quota(additional=copied)
            self._continuations[continuation.token_hash] = copied

    async def get_continuation(
        self, token_hash: str, *, principal_id: str
    ) -> LegacyPurchaseContinuation | None:
        async with self._lock:
            record = self._continuations.get(token_hash)
            if record is None or record.principal_id != principal_id:
                return None
            return _copy_continuation(record)

    async def claim(
        self,
        token_hash: str,
        *,
        principal_id: str,
        idempotency_key: str,
        payload_hash: str,
        execution_input: Mapping[str, Any],
        now: datetime,
    ) -> CompatibilityPurchaseOperation:
        async with self._lock:
            claim_time = max(_aware_utc(now, field="claim time"), self._clock())
            key = (principal_id, idempotency_key)
            existing = self._operations.get(key)
            if existing is not None:
                if existing.payload_hash != payload_hash or existing.token_hash != token_hash:
                    raise _error(
                        CompatibilityContinuationErrorCode.IDEMPOTENCY_CONFLICT,
                        "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)
            if record is None or record.principal_id != principal_id:
                raise _not_found()
            if claim_time >= record.expires_at:
                raise _error(
                    CompatibilityContinuationErrorCode.EXPIRED,
                    "continuation expired before it could be claimed",
                    "Repeat product discovery and obtain a new continuation.",
                )
            if token_hash in self._claimed_by:
                raise _error(
                    CompatibilityContinuationErrorCode.ALREADY_CLAIMED,
                    "continuation was already claimed by another operation",
                    "Replay the original idempotency key, or restart product discovery.",
                )

            operation = CompatibilityPurchaseOperation(
                operation_id=secrets.token_urlsafe(24),
                principal_id=principal_id,
                idempotency_key=idempotency_key,
                token_hash=token_hash,
                payload_hash=payload_hash,
                state=CompatibilityOperationState.CLAIMED,
                revision=1,
                execution_input=_json_copy(execution_input),
            )
            _validate_persistable_payload(operation.execution_input, context="execution input")
            self._check_quota(additional=operation)
            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 get_operation_by_idempotency_key(
        self, idempotency_key: str, *, principal_id: str
    ) -> CompatibilityPurchaseOperation | None:
        async with self._lock:
            operation = self._operations.get((principal_id, idempotency_key))
            return _copy_operation(operation) if operation is not None else None

    async def mark_in_flight(
        self, operation: CompatibilityPurchaseOperation
    ) -> CompatibilityPurchaseOperation:
        return await self._transition(
            operation,
            allowed={CompatibilityOperationState.CLAIMED},
            target=CompatibilityOperationState.IN_FLIGHT,
        )

    async def mark_ambiguous(
        self, operation: CompatibilityPurchaseOperation
    ) -> CompatibilityPurchaseOperation:
        return await self._transition(
            operation,
            allowed={CompatibilityOperationState.CLAIMED, CompatibilityOperationState.IN_FLIGHT},
            target=CompatibilityOperationState.AMBIGUOUS,
        )

    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)
        _validate_persistable_payload(copied, context="legacy result")
        return await self._transition(
            operation,
            allowed={
                CompatibilityOperationState.IN_FLIGHT,
                CompatibilityOperationState.AMBIGUOUS,
                CompatibilityOperationState.PENDING,
            },
            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:
        return await self._transition(
            operation,
            allowed={CompatibilityOperationState.AMBIGUOUS},
            target=CompatibilityOperationState.CLAIMED,
        )

    async def _transition(
        self,
        operation: CompatibilityPurchaseOperation,
        *,
        allowed: set[CompatibilityOperationState],
        target: CompatibilityOperationState,
        result: JsonObject | None = None,
    ) -> CompatibilityPurchaseOperation:
        async with self._lock:
            key = (operation.principal_id, operation.idempotency_key)
            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,
                revision=current.revision + 1,
                result=copy.deepcopy(result),
            )
            self._operations[key] = updated
            try:
                self._check_current_quota()
            except BaseException:
                self._operations[key] = current
                raise
            return _copy_operation(updated)

    def _check_quota(
        self,
        *,
        additional: LegacyPurchaseContinuation | CompatibilityPurchaseOperation,
    ) -> None:
        records = len(self._continuations) + len(self._operations) + 1
        values: list[Any] = [*self._continuations.values(), *self._operations.values(), additional]
        logical_bytes = sum(len(repr(value).encode("utf-8")) for value in values)
        if records > self._max_records or logical_bytes > self._max_bytes:
            raise _quota_error(self._max_records, self._max_bytes)

    def _check_current_quota(self) -> None:
        values: list[Any] = [*self._continuations.values(), *self._operations.values()]
        logical_bytes = sum(len(repr(value).encode("utf-8")) for value in values)
        if len(values) > self._max_records or logical_bytes > self._max_bytes:
            raise _quota_error(self._max_records, self._max_bytes)

Process-local reference store for tests and development only.

Class variables

var is_durable : ClassVar[bool]

Methods

async def claim(self,
token_hash: str,
*,
principal_id: str,
idempotency_key: str,
payload_hash: str,
execution_input: Mapping[str, Any],
now: datetime) ‑> CompatibilityPurchaseOperation

Expand source code

async def claim(
    self,
    token_hash: str,
    *,
    principal_id: str,
    idempotency_key: str,
    payload_hash: str,
    execution_input: Mapping[str, Any],
    now: datetime,
) -> CompatibilityPurchaseOperation:
    async with self._lock:
        claim_time = max(_aware_utc(now, field="claim time"), self._clock())
        key = (principal_id, idempotency_key)
        existing = self._operations.get(key)
        if existing is not None:
            if existing.payload_hash != payload_hash or existing.token_hash != token_hash:
                raise _error(
                    CompatibilityContinuationErrorCode.IDEMPOTENCY_CONFLICT,
                    "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)
        if record is None or record.principal_id != principal_id:
            raise _not_found()
        if claim_time >= record.expires_at:
            raise _error(
                CompatibilityContinuationErrorCode.EXPIRED,
                "continuation expired before it could be claimed",
                "Repeat product discovery and obtain a new continuation.",
            )
        if token_hash in self._claimed_by:
            raise _error(
                CompatibilityContinuationErrorCode.ALREADY_CLAIMED,
                "continuation was already claimed by another operation",
                "Replay the original idempotency key, or restart product discovery.",
            )

        operation = CompatibilityPurchaseOperation(
            operation_id=secrets.token_urlsafe(24),
            principal_id=principal_id,
            idempotency_key=idempotency_key,
            token_hash=token_hash,
            payload_hash=payload_hash,
            state=CompatibilityOperationState.CLAIMED,
            revision=1,
            execution_input=_json_copy(execution_input),
        )
        _validate_persistable_payload(operation.execution_input, context="execution input")
        self._check_quota(additional=operation)
        self._operations[key] = operation
        self._claimed_by[token_hash] = operation.operation_id
        return _copy_operation(operation)

async def complete(self,
operation: CompatibilityPurchaseOperation,
result: Mapping[str, Any],
*,
state: CompatibilityOperationState = CompatibilityOperationState.SUCCEEDED) ‑> CompatibilityPurchaseOperation

Expand source code

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)
    _validate_persistable_payload(copied, context="legacy result")
    return await self._transition(
        operation,
        allowed={
            CompatibilityOperationState.IN_FLIGHT,
            CompatibilityOperationState.AMBIGUOUS,
            CompatibilityOperationState.PENDING,
        },
        target=state,
        result=copied,
    )

async def fence_in_flight(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def fence_in_flight(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation:
    return await self._transition(
        operation,
        allowed={CompatibilityOperationState.IN_FLIGHT},
        target=CompatibilityOperationState.AMBIGUOUS,
    )

async def get_continuation(self, token_hash: str, *, principal_id: str) ‑> LegacyPurchaseContinuation | None

Expand source code

async def get_continuation(
    self, token_hash: str, *, principal_id: str
) -> LegacyPurchaseContinuation | None:
    async with self._lock:
        record = self._continuations.get(token_hash)
        if record is None or record.principal_id != principal_id:
            return None
        return _copy_continuation(record)

async def get_operation(self, operation_id: str, *, principal_id: str) ‑> CompatibilityPurchaseOperation | None

Expand source code

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 get_operation_by_idempotency_key(self, idempotency_key: str, *, principal_id: str) ‑> CompatibilityPurchaseOperation | None

Expand source code

async def get_operation_by_idempotency_key(
    self, idempotency_key: str, *, principal_id: str
) -> CompatibilityPurchaseOperation | None:
    async with self._lock:
        operation = self._operations.get((principal_id, idempotency_key))
        return _copy_operation(operation) if operation is not None else None

async def mark_ambiguous(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def mark_ambiguous(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation:
    return await self._transition(
        operation,
        allowed={CompatibilityOperationState.CLAIMED, CompatibilityOperationState.IN_FLIGHT},
        target=CompatibilityOperationState.AMBIGUOUS,
    )

async def mark_in_flight(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def mark_in_flight(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation:
    return await self._transition(
        operation,
        allowed={CompatibilityOperationState.CLAIMED},
        target=CompatibilityOperationState.IN_FLIGHT,
    )

async def put_continuation(self,
continuation: LegacyPurchaseContinuation) ‑> None

Expand source code

async def put_continuation(self, continuation: LegacyPurchaseContinuation) -> None:
    async with self._lock:
        copied = _copy_continuation(continuation)
        _validate_continuation_persistence(copied)
        if any(
            value.issuance_fingerprint is None
            and value.principal_id == copied.principal_id
            and value.observed_payload_hash == copied.observed_payload_hash
            and value.account_identity == copied.account_identity
            and value.target_binding == copied.target_binding
            for value in self._continuations.values()
        ):
            raise _error(
                CompatibilityContinuationErrorCode.STORE_CONFLICT,
                "equivalent pre-migration authorization requires operator resolution",
                "Resolve or quarantine the legacy continuation before reissuing.",
            )
        existing = self._continuations.get(continuation.token_hash)
        by_fingerprint = next(
            (
                value
                for value in self._continuations.values()
                if value.issuance_fingerprint is not None
                and value.issuance_fingerprint == continuation.issuance_fingerprint
                and value.principal_id == continuation.principal_id
            ),
            None,
        )
        if existing == copied and (by_fingerprint is None or by_fingerprint == existing):
            return
        if existing is not None or by_fingerprint is not None:
            raise _error(
                CompatibilityContinuationErrorCode.STORE_CONFLICT,
                "continuation issuance fingerprint is already registered differently",
                "Use the same token derivation key and exact issuance inputs.",
            )
        self._check_quota(additional=copied)
        self._continuations[continuation.token_hash] = copied

async def resume_after_not_applied(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def resume_after_not_applied(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation:
    return await self._transition(
        operation,
        allowed={CompatibilityOperationState.AMBIGUOUS},
        target=CompatibilityOperationState.CLAIMED,
    )

class LegacyCatalogContinuation (kind: "Literal['legacy_create']",
token: str,
losses: tuple[str, ...],
expires_at: datetime)

Expand source code

@dataclass(frozen=True)
class LegacyCatalogContinuation:
    """SDK-local authorization for purchasing from an established catalog result."""

    kind: Literal["legacy_create"]
    token: str = dataclass_field(repr=False)
    losses: tuple[str, ...]
    expires_at: datetime

SDK-local authorization for purchasing from an established catalog result.

Instance variables

var expires_at : datetime.datetime

var kind : Literal['legacy_create']

var losses : tuple[str, ...]

var token : str

class LegacyPurchaseContinuation (token_hash: str,
issuance_fingerprint: str | None,
issuance_binding_hash: str | None,
principal_id: str,
account_identity: str,
source_adcp_version: str,
expires_at: datetime,
observed_request: JsonObject,
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)

Expand source code

@dataclass(frozen=True)
class LegacyPurchaseContinuation:
    """Immutable, token-bound compatibility context stored at projection time.

    ``token_hash`` is SHA-256 of the opaque token.  The raw bearer token must
    never be persisted.  ``observed_request`` and ``observed_response`` retain
    the complete legacy discovery transaction, including every observed product
    and pricing option, rather than a reconstructable subset.
    """

    token_hash: str
    issuance_fingerprint: str | None
    issuance_binding_hash: str | None
    principal_id: str
    account_identity: str
    source_adcp_version: str
    expires_at: datetime
    observed_request: JsonObject
    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

Immutable, token-bound compatibility context stored at projection time.

token_hash is SHA-256 of the opaque token. The raw bearer token must never be persisted. observed_request and observed_response retain the complete legacy discovery transaction, including every observed product and pricing option, rather than a reconstructable subset.

Instance variables

var account_identity : str

var expires_at : datetime.datetime

var issuance_binding_hash : str | None

var issuance_fingerprint : str | None

var listed_purchase_context : dict[str, typing.Any] | None

var losses : frozenset[str]

var mutation_idempotency_guaranteed : bool

var observed_payload_hash : str

var observed_request : dict[str, typing.Any]

var observed_response : dict[str, typing.Any]

var principal_id : str

var product_ids : tuple[str, ...]

var projected_products : tuple[dict[str, typing.Any], ...] | None

var source_adcp_version : str

var target_binding : str

var token_hash : str

class LegacyPurchaseCoordinator (*,
store: CompatibilityContinuationStore,
executor: LegacyPurchaseExecutor,
reconciler: LegacyPurchaseReconciler | None = None,
pending_poller: LegacyPurchasePendingPoller | None = None,
token_derivation_key: bytes | bytearray | memoryview | None = None,
allow_non_durable_store: bool = False,
clock: Callable[[], datetime] | None = None)

Expand source code

class LegacyPurchaseCoordinator:
    """Validate, atomically claim, execute, and replay a legacy purchase."""

    def __init__(
        self,
        *,
        store: CompatibilityContinuationStore,
        executor: LegacyPurchaseExecutor,
        reconciler: LegacyPurchaseReconciler | None = None,
        pending_poller: LegacyPurchasePendingPoller | None = None,
        token_derivation_key: bytes | bytearray | memoryview | None = None,
        allow_non_durable_store: bool = False,
        clock: Callable[[], datetime] | None = None,
    ) -> None:
        if not isinstance(store, CompatibilityContinuationStore):
            raise TypeError("store must implement CompatibilityContinuationStore")
        if not store.is_durable and not allow_non_durable_store:
            raise ValueError(
                "production continuation coordination requires a durable store; "
                "set allow_non_durable_store=True only for tests or local development"
            )
        self.store = store
        self.executor = executor
        self.reconciler = reconciler
        self.pending_poller = pending_poller
        if token_derivation_key is None:
            if store.is_durable:
                raise ValueError(
                    "durable continuation coordination requires a stable "
                    "token_derivation_key of at least 32 bytes"
                )
            token_derivation_key = secrets.token_bytes(_MIN_TOKEN_DERIVATION_KEY_BYTES)
        if not isinstance(token_derivation_key, (bytes, bytearray, memoryview)):
            raise TypeError("token_derivation_key must be bytes-like")
        key = bytes(token_derivation_key)
        if len(key) < _MIN_TOKEN_DERIVATION_KEY_BYTES or not any(key):
            raise ValueError("token_derivation_key must be a high-entropy secret of 32+ bytes")
        self._token_derivation_key = key
        self._clock = clock or (lambda: datetime.now(timezone.utc))

    async def issue_legacy_create_continuation(
        self,
        *,
        principal_id: str,
        issuance_idempotency_key: str,
        account: Mapping[str, Any] | Any,
        source_adcp_version: str,
        expires_at: datetime,
        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(issuance_idempotency_key, "issuance_idempotency_key")
        _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")
        if expires_at <= now:
            raise _invalid("expires_at must be in the future")

        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,
            mutation_idempotency_guaranteed=mutation_idempotency_guaranteed,
        )
        observed_req = _json_copy(observed_request)
        observed_resp = _json_copy(observed_response)
        _validate_source_discovery(
            observed_req,
            observed_resp,
            source_adcp_version=source_adcp_version,
            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
        )
        for context, value in (
            ("observed request", observed_req),
            ("observed response", observed_resp),
            ("buyer-visible products", {"products": list(projected)}),
            ("listed purchase context", listed),
        ):
            if value is not None:
                _validate_persistable_payload(value, context=context)

        issuance_fingerprint = _full_hash(
            {
                "principal_id": principal_id,
                "issuance_idempotency_key": issuance_idempotency_key,
            }
        )
        issuance_binding_hash = _full_hash(
            {
                "account_identity": account_identity,
                "source_adcp_version": source_adcp_version,
                "expires_at": expires_at.isoformat(),
                "observed_request": observed_req,
                "observed_response": observed_resp,
                "product_ids": list(ids),
                "projected_products": list(projected),
                "losses": sorted(loss_set),
                "mutation_idempotency_guaranteed": mutation_idempotency_guaranteed,
                "target_binding": target_binding,
                "listed_purchase_context": listed,
            }
        )
        token = _derive_token(
            self._token_derivation_key, issuance_fingerprint, issuance_binding_hash
        )
        token_hash = _token_hash(token)
        observed_payload_hash = _full_hash({"request": observed_req, "response": observed_resp})
        record = LegacyPurchaseContinuation(
            token_hash=token_hash,
            issuance_fingerprint=issuance_fingerprint,
            issuance_binding_hash=issuance_binding_hash,
            principal_id=principal_id,
            account_identity=account_identity,
            source_adcp_version=source_adcp_version,
            expires_at=expires_at,
            observed_request=observed_req,
            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,
        )
        await self.store.put_continuation(record)
        return token

    async def continue_legacy_purchase(
        self,
        purchase_input: CompatibilityPurchaseCoordinatorInput | Mapping[str, Any],
        *,
        principal_id: str,
        target_binding: str,
    ) -> JsonObject:
        """Redeem a ``legacy_create`` continuation or deterministically replay it."""

        _require_text(principal_id, "principal_id")
        _require_text(target_binding, "target_binding")
        payload = _parse_input(purchase_input)
        token_hash = _token_hash(payload["continuation_token"])
        record = await self.store.get_continuation(token_hash, principal_id=principal_id)
        if record is None:
            raise _not_found()
        self._validate_bindings(payload, record, target_binding=target_binding)

        payload_hash = _full_hash({"input": payload, "target_binding": target_binding})
        now = _aware_utc(self._clock(), field="clock result")
        operation = await self.store.claim(
            token_hash,
            principal_id=principal_id,
            idempotency_key=payload["idempotency_key"],
            payload_hash=payload_hash,
            execution_input=_execution_input(payload),
            now=now,
        )
        return await self._drive(operation, record, target_binding=target_binding)

    async def refresh_pending_legacy_purchase(
        self,
        operation: CompatibilityPurchaseOperation,
        *,
        principal_id: str,
        target_binding: str,
    ) -> JsonObject:
        """Poll and CAS-advance a pending seller task through an application callback.

        The poller must be an idempotent, read-only lookup of the already-created
        seller task. Input/approval submission happens outside this callback.
        The returned task identity is checked for pending and terminal results.
        """

        _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 pending refresh")
        if current.state != CompatibilityOperationState.PENDING or current.result is None:
            raise _error(
                CompatibilityContinuationErrorCode.PENDING_RESOLUTION_REQUIRED,
                "operation is not a revision-bearing pending seller task",
                "Look up a fresh pending operation snapshot before refreshing it.",
                details={"operation_id": current.operation_id},
            )
        if self.pending_poller is None:
            raise _error(
                CompatibilityContinuationErrorCode.PENDING_RESOLUTION_REQUIRED,
                "no pending seller task poller is configured",
                "Configure pending_poller to read the original seller task state.",
                details={"operation_id": current.operation_id},
            )
        record = await self.store.get_continuation(current.token_hash, principal_id=principal_id)
        if record is None:
            raise _not_found()
        self._validate_bindings(current.execution_input, record, target_binding=target_binding)
        execution = _execution_from(current, record, current.execution_input, target_binding)
        resolution = await _call_callback(
            self.pending_poller, _copy_execution(execution), _copy_operation(current)
        )
        previous_task_id = current.result.get("task_id")
        if not isinstance(resolution, PendingTaskResolution):
            raise TypeError("pending_poller must return PendingTaskResolution")
        if resolution.task_id != previous_task_id:
            raise _invalid_legacy_response(record.source_adcp_version, [])
        copied, state = _validated_result(
            resolution.result, source_adcp_version=record.source_adcp_version
        )
        if (
            state == CompatibilityOperationState.PENDING
            and copied.get("task_id") != previous_task_id
        ):
            raise _invalid_legacy_response(record.source_adcp_version, [])
        completed = await _shielded_transition(self.store.complete(current, copied, state=state))
        assert completed.result is not None
        return copy.deepcopy(completed.result)

    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 get_legacy_purchase_operation_by_idempotency_key(
        self, idempotency_key: str, *, principal_id: str
    ) -> CompatibilityPurchaseOperation:
        """Look up a principal-scoped operation when only the buyer key is known."""

        _require_text(idempotency_key, "idempotency_key")
        _require_text(principal_id, "principal_id")
        operation = await self.store.get_operation_by_idempotency_key(
            idempotency_key, 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)

    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("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, 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._reserve_execution(operation)
        except CompatibilityContinuationError as exc:
            if exc.code != CompatibilityContinuationErrorCode.STORE_CONFLICT:
                raise
            latest = await self.store.get_operation(
                operation.operation_id, principal_id=operation.principal_id
            )
            if latest is None:
                raise
            return await self._drive(
                latest, record, target_binding=target_binding, recovery=recovery
            )

        try:
            result = await _call_callback(self.executor, _copy_execution(execution))
        except asyncio.CancelledError:
            await self._mark_ambiguous_after_interruption(operation)
            raise
        except Exception as 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 _shielded_transition(
                self.store.complete(operation, copied, state=result_state)
            )
        except asyncio.CancelledError:
            raise
        except Exception as store_exc:
            try:
                await asyncio.shield(self.store.mark_ambiguous(operation))
            except Exception:  # nosec B110 - surface the original ambiguous commit
                pass
            raise _ambiguous_error(operation) from store_exc
        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 _call_callback(
                self.reconciler, _copy_execution(execution), _copy_operation(operation)
            )
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            raise _ambiguous_error(operation) from exc
        try:
            if not isinstance(outcome, ReconciliationResult):
                raise TypeError("legacy purchase reconciler must return ReconciliationResult")
            if outcome.status == ReconciliationStatus.APPLIED:
                if outcome.result is None:
                    raise ValueError("applied reconciliation requires a 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
                    or not allow_not_applied
                ):
                    raise _ambiguous_error(operation)
                return await self.store.resume_after_not_applied(operation)
            raise _ambiguous_error(operation)
        except CompatibilityContinuationError as exc:
            if exc.code == CompatibilityContinuationErrorCode.AMBIGUOUS_MUTATION:
                raise
            raise _ambiguous_error(operation) from exc
        except Exception as exc:
            raise _ambiguous_error(operation) from exc

    def _validate_bindings(
        self,
        payload: JsonObject,
        record: LegacyPurchaseContinuation,
        *,
        target_binding: str,
    ) -> None:
        if (
            _full_hash(
                {
                    "request": record.observed_request,
                    "response": record.observed_response,
                }
            )
            != record.observed_payload_hash
        ):
            raise _store_state_error("stored observed transaction failed its payload binding")
        if target_binding != record.target_binding:
            raise _binding_error("seller target/session does not match the issued continuation")
        if canonical_account_identity(payload["account"]) != record.account_identity:
            raise _binding_error("input account does not match the issued continuation")

        selected = _unique_nonempty_strings(
            payload["selected_product_ids"], field="selected_product_ids"
        )
        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,
            mutation_idempotency_guaranteed=record.mutation_idempotency_guaranteed,
            enforce_guarantee=False,
        )
        if accepted != record.losses:
            raise _error(
                CompatibilityContinuationErrorCode.LOSS_MISMATCH,
                "accepted_losses does not exactly match the issued loss set",
                "Accept the complete current loss set exactly, or restart discovery.",
            )

        request = payload["legacy_create_request"]
        outcome = validate_request("create_media_buy", request, version=record.source_adcp_version)
        if not _source_schema_valid(outcome, record.source_adcp_version):
            raise _error(
                CompatibilityContinuationErrorCode.INVALID_LEGACY_REQUEST,
                "legacy_create_request does not validate against its exact source version",
                "Construct the request using the source-version create_media_buy schema.",
                details={
                    "source_adcp_version": record.source_adcp_version,
                    "issues": [
                        {
                            "pointer": issue.pointer,
                            "keyword": issue.keyword,
                            "message": issue.message,
                        }
                        for issue in outcome.issues
                    ],
                },
            )
        packages = request.get("packages")
        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 = (
            _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 (
                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"
                )
        if set(package_ids) != set(selected):
            raise _binding_error(
                "distinct legacy package product IDs do not equal selected_product_ids"
            )

        if not record.source_adcp_version.startswith("2.5."):
            request_account = request.get("account")
            if not isinstance(request_account, Mapping):
                raise _legacy_request_error("source-version request account is required")
            if canonical_account_identity(request_account) != record.account_identity:
                raise _binding_error("legacy request account does not match the continuation")

Validate, atomically claim, execute, and replay a legacy purchase.

Methods

async def continue_legacy_purchase(self,
purchase_input: CompatibilityPurchaseCoordinatorInput | Mapping[str, Any],
*,
principal_id: str,
target_binding: str) ‑> dict[str, typing.Any]

Expand source code

async def continue_legacy_purchase(
    self,
    purchase_input: CompatibilityPurchaseCoordinatorInput | Mapping[str, Any],
    *,
    principal_id: str,
    target_binding: str,
) -> JsonObject:
    """Redeem a ``legacy_create`` continuation or deterministically replay it."""

    _require_text(principal_id, "principal_id")
    _require_text(target_binding, "target_binding")
    payload = _parse_input(purchase_input)
    token_hash = _token_hash(payload["continuation_token"])
    record = await self.store.get_continuation(token_hash, principal_id=principal_id)
    if record is None:
        raise _not_found()
    self._validate_bindings(payload, record, target_binding=target_binding)

    payload_hash = _full_hash({"input": payload, "target_binding": target_binding})
    now = _aware_utc(self._clock(), field="clock result")
    operation = await self.store.claim(
        token_hash,
        principal_id=principal_id,
        idempotency_key=payload["idempotency_key"],
        payload_hash=payload_hash,
        execution_input=_execution_input(payload),
        now=now,
    )
    return await self._drive(operation, record, target_binding=target_binding)

Redeem a legacy_create continuation or deterministically replay it.

async def get_legacy_purchase_operation(self, operation_id: str, *, principal_id: str) ‑> CompatibilityPurchaseOperation

Expand source code

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)

Return a principal-scoped operation snapshot carrying its CAS revision.

async def get_legacy_purchase_operation_by_idempotency_key(self, idempotency_key: str, *, principal_id: str) ‑> CompatibilityPurchaseOperation

Expand source code

async def get_legacy_purchase_operation_by_idempotency_key(
    self, idempotency_key: str, *, principal_id: str
) -> CompatibilityPurchaseOperation:
    """Look up a principal-scoped operation when only the buyer key is known."""

    _require_text(idempotency_key, "idempotency_key")
    _require_text(principal_id, "principal_id")
    operation = await self.store.get_operation_by_idempotency_key(
        idempotency_key, principal_id=principal_id
    )
    if operation is None:
        raise _not_found()
    return _copy_operation(operation)

Look up a principal-scoped operation when only the buyer key is known.

async def issue_legacy_create_continuation(self,
*,
principal_id: str,
issuance_idempotency_key: str,
account: Mapping[str, Any] | Any,
source_adcp_version: str,
expires_at: datetime,
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

Expand source code

async def issue_legacy_create_continuation(
    self,
    *,
    principal_id: str,
    issuance_idempotency_key: str,
    account: Mapping[str, Any] | Any,
    source_adcp_version: str,
    expires_at: datetime,
    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(issuance_idempotency_key, "issuance_idempotency_key")
    _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")
    if expires_at <= now:
        raise _invalid("expires_at must be in the future")

    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,
        mutation_idempotency_guaranteed=mutation_idempotency_guaranteed,
    )
    observed_req = _json_copy(observed_request)
    observed_resp = _json_copy(observed_response)
    _validate_source_discovery(
        observed_req,
        observed_resp,
        source_adcp_version=source_adcp_version,
        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
    )
    for context, value in (
        ("observed request", observed_req),
        ("observed response", observed_resp),
        ("buyer-visible products", {"products": list(projected)}),
        ("listed purchase context", listed),
    ):
        if value is not None:
            _validate_persistable_payload(value, context=context)

    issuance_fingerprint = _full_hash(
        {
            "principal_id": principal_id,
            "issuance_idempotency_key": issuance_idempotency_key,
        }
    )
    issuance_binding_hash = _full_hash(
        {
            "account_identity": account_identity,
            "source_adcp_version": source_adcp_version,
            "expires_at": expires_at.isoformat(),
            "observed_request": observed_req,
            "observed_response": observed_resp,
            "product_ids": list(ids),
            "projected_products": list(projected),
            "losses": sorted(loss_set),
            "mutation_idempotency_guaranteed": mutation_idempotency_guaranteed,
            "target_binding": target_binding,
            "listed_purchase_context": listed,
        }
    )
    token = _derive_token(
        self._token_derivation_key, issuance_fingerprint, issuance_binding_hash
    )
    token_hash = _token_hash(token)
    observed_payload_hash = _full_hash({"request": observed_req, "response": observed_resp})
    record = LegacyPurchaseContinuation(
        token_hash=token_hash,
        issuance_fingerprint=issuance_fingerprint,
        issuance_binding_hash=issuance_binding_hash,
        principal_id=principal_id,
        account_identity=account_identity,
        source_adcp_version=source_adcp_version,
        expires_at=expires_at,
        observed_request=observed_req,
        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,
    )
    await self.store.put_continuation(record)
    return token

Persist all projection bindings and return the opaque bearer token.

async def recover_legacy_purchase(self,
operation: CompatibilityPurchaseOperation,
*,
principal_id: str,
target_binding: str) ‑> dict[str, typing.Any]

Expand source code

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)

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.

async def refresh_pending_legacy_purchase(self,
operation: CompatibilityPurchaseOperation,
*,
principal_id: str,
target_binding: str) ‑> dict[str, typing.Any]

Expand source code

async def refresh_pending_legacy_purchase(
    self,
    operation: CompatibilityPurchaseOperation,
    *,
    principal_id: str,
    target_binding: str,
) -> JsonObject:
    """Poll and CAS-advance a pending seller task through an application callback.

    The poller must be an idempotent, read-only lookup of the already-created
    seller task. Input/approval submission happens outside this callback.
    The returned task identity is checked for pending and terminal results.
    """

    _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 pending refresh")
    if current.state != CompatibilityOperationState.PENDING or current.result is None:
        raise _error(
            CompatibilityContinuationErrorCode.PENDING_RESOLUTION_REQUIRED,
            "operation is not a revision-bearing pending seller task",
            "Look up a fresh pending operation snapshot before refreshing it.",
            details={"operation_id": current.operation_id},
        )
    if self.pending_poller is None:
        raise _error(
            CompatibilityContinuationErrorCode.PENDING_RESOLUTION_REQUIRED,
            "no pending seller task poller is configured",
            "Configure pending_poller to read the original seller task state.",
            details={"operation_id": current.operation_id},
        )
    record = await self.store.get_continuation(current.token_hash, principal_id=principal_id)
    if record is None:
        raise _not_found()
    self._validate_bindings(current.execution_input, record, target_binding=target_binding)
    execution = _execution_from(current, record, current.execution_input, target_binding)
    resolution = await _call_callback(
        self.pending_poller, _copy_execution(execution), _copy_operation(current)
    )
    previous_task_id = current.result.get("task_id")
    if not isinstance(resolution, PendingTaskResolution):
        raise TypeError("pending_poller must return PendingTaskResolution")
    if resolution.task_id != previous_task_id:
        raise _invalid_legacy_response(record.source_adcp_version, [])
    copied, state = _validated_result(
        resolution.result, source_adcp_version=record.source_adcp_version
    )
    if (
        state == CompatibilityOperationState.PENDING
        and copied.get("task_id") != previous_task_id
    ):
        raise _invalid_legacy_response(record.source_adcp_version, [])
    completed = await _shielded_transition(self.store.complete(current, copied, state=state))
    assert completed.result is not None
    return copy.deepcopy(completed.result)

Poll and CAS-advance a pending seller task through an application callback.

The poller must be an idempotent, read-only lookup of the already-created seller task. Input/approval submission happens outside this callback. The returned task identity is checked for pending and terminal results.

class LegacyPurchaseExecution (operation_id: str,
principal_id: str,
idempotency_key: str,
source_adcp_version: str,
account: JsonObject,
target_binding: str,
selected_product_ids: tuple[str, ...],
legacy_create_request: JsonObject,
observed_request: JsonObject,
observed_response: JsonObject,
listed_purchase_context: JsonObject | None)

Expand source code

@dataclass(frozen=True)
class LegacyPurchaseExecution:
    """Execution context passed to the application-owned legacy executor."""

    operation_id: str
    principal_id: str
    idempotency_key: str
    source_adcp_version: str
    account: JsonObject
    target_binding: str
    selected_product_ids: tuple[str, ...]
    legacy_create_request: JsonObject
    observed_request: JsonObject
    observed_response: JsonObject
    listed_purchase_context: JsonObject | None

Execution context passed to the application-owned legacy executor.

Instance variables

var account : dict[str, typing.Any]

var idempotency_key : str

var legacy_create_request : dict[str, typing.Any]

var listed_purchase_context : dict[str, typing.Any] | None

var observed_request : dict[str, typing.Any]

var observed_response : dict[str, typing.Any]

var operation_id : str

var principal_id : str

var selected_product_ids : tuple[str, ...]

var source_adcp_version : str

var target_binding : str

class MediaBuyCompatibility (*args, **kwds)

Expand source code

class MediaBuyCompatibility(str, Enum):
    """Strength of an operation projection."""

    NATIVE = "native"
    LOSSLESS_PROJECTION = "lossless_projection"
    LOSSY_PROJECTION = "lossy_projection"

Strength of an operation projection.

Ancestors

• builtins.str
• enum.Enum

Class variables

var LOSSLESS_PROJECTION

var LOSSY_PROJECTION

var NATIVE

class MediaBuyCompatibilityReport (negotiated_version: str,
lifecycle: MediaBuyLifecycle,
tools_used: tuple[str, ...],
compatibility: MediaBuyCompatibility,
warnings: tuple[str, ...] = (),
losses: tuple[str, ...] = ())

Expand source code

@dataclass(frozen=True)
class MediaBuyCompatibilityReport:
    """Exact route and guarantee changes for one coordinator operation."""

    negotiated_version: str
    lifecycle: MediaBuyLifecycle
    tools_used: tuple[str, ...]
    compatibility: MediaBuyCompatibility
    warnings: tuple[str, ...] = ()
    losses: tuple[str, ...] = ()

Exact route and guarantee changes for one coordinator operation.

Instance variables

var compatibility : MediaBuyCompatibility

var lifecycle : MediaBuyLifecycle

var losses : tuple[str, ...]

var negotiated_version : str

var tools_used : tuple[str, ...]

var warnings : tuple[str, ...]

class MediaBuyLifecycle (*args, **kwds)

Expand source code

class MediaBuyLifecycle(str, Enum):
    """Wire lifecycle selected for an operation."""

    COMPACT = "compact"
    ESTABLISHED = "established"

Wire lifecycle selected for an operation.

Ancestors

• builtins.str
• enum.Enum

Class variables

var COMPACT

var ESTABLISHED

class MediaBuyLifecycleCompatibilityError (*,
operation: str,
negotiated_version: str,
lifecycle: MediaBuyLifecycle,
feature: str,
message: str,
losses: Sequence[str] = (),
recovery: str | None = None,
code: CompatibilityErrorCode = 'UNSUPPORTED_FEATURE')

Expand source code

class MediaBuyLifecycleCompatibilityError(Exception):
    """Structured preflight failure raised before an incompatible dispatch."""

    def __init__(
        self,
        *,
        operation: str,
        negotiated_version: str,
        lifecycle: MediaBuyLifecycle,
        feature: str,
        message: str,
        losses: Sequence[str] = (),
        recovery: str | None = None,
        code: CompatibilityErrorCode = "UNSUPPORTED_FEATURE",
    ) -> None:
        self.code = code
        self.operation = operation
        self.negotiated_version = negotiated_version
        self.lifecycle = lifecycle
        self.feature = feature
        self.losses = tuple(losses)
        self.recovery = recovery or (
            "Remove the unsupported requirement, select a compatible seller, or explicitly "
            "accept the named compatibility loss."
        )
        super().__init__(message)

Structured preflight failure raised before an incompatible dispatch.

Ancestors

• builtins.Exception
• builtins.BaseException

class MediaBuyLifecycleCoordinator (client: ADCPClient,
capabilities: GetAdcpCapabilitiesResponse,
advertised_tools: Sequence[str],
*,
legacy_purchase_coordinator: LegacyPurchaseCoordinator | None = None,
principal_id: str | None = None,
target_binding: str | None = None,
preferred_lifecycle: LifecyclePreference = 'auto',
allowed_losses: Sequence[str] = (),
continuation_ttl: timedelta = datetime.timedelta(seconds=300),
clock: Callable[[], datetime] | None = None)

Expand source code

class MediaBuyLifecycleCoordinator:
    """Negotiate one catalog-to-purchase workflow across AdCP 3.x."""

    def __init__(
        self,
        client: ADCPClient,
        capabilities: GetAdcpCapabilitiesResponse,
        advertised_tools: Sequence[str],
        *,
        legacy_purchase_coordinator: LegacyPurchaseCoordinator | None = None,
        principal_id: str | None = None,
        target_binding: str | None = None,
        preferred_lifecycle: LifecyclePreference = "auto",
        allowed_losses: Sequence[str] = (),
        continuation_ttl: timedelta = timedelta(minutes=5),
        clock: Callable[[], datetime] | None = None,
    ) -> None:
        if preferred_lifecycle not in {"auto", "compact", "established"}:
            raise ValueError("preferred_lifecycle must be auto, compact, or established")
        if continuation_ttl <= timedelta(0):
            raise ValueError("continuation_ttl must be positive")
        if principal_id is not None:
            if not isinstance(principal_id, str) or not principal_id.strip():
                raise TypeError("principal_id must be a non-empty string when provided")
            principal_id = principal_id.strip()
            if re.search(r"[\x00-\x1f\x7f]", principal_id):
                raise ValueError("principal_id must not contain control characters")
            if len(principal_id.encode("utf-8")) > _MAX_PRINCIPAL_BYTES:
                raise ValueError(f"principal_id must be at most {_MAX_PRINCIPAL_BYTES} UTF-8 bytes")
        if target_binding is not None:
            if not isinstance(target_binding, str) or not target_binding.strip():
                raise TypeError("target_binding must be a non-empty string when provided")
            target_binding = target_binding.strip()
        if legacy_purchase_coordinator is not None and (not principal_id or not target_binding):
            raise ValueError(
                "compatibility persistence requires non-empty principal_id and target_binding"
            )

        self.client = client
        self.capabilities = capabilities
        self.negotiated_version = _negotiate_version(capabilities, client.get_adcp_version())
        self._tools = frozenset(str(tool) for tool in advertised_tools)
        media_buy = getattr(capabilities, "media_buy", None)
        self._tools = self._tools | frozenset(
            _root_value(tool) for tool in (getattr(media_buy, "lifecycle_tools", None) or ())
        )
        self._buying_modes = frozenset(
            _root_value(mode) for mode in (getattr(media_buy, "buying_modes", None) or ())
        )
        self._preferred_lifecycle = preferred_lifecycle
        self._allowed_losses = frozenset(str(loss) for loss in allowed_losses)
        self._legacy = legacy_purchase_coordinator
        self._principal_id = principal_id
        self._target_binding = target_binding
        self._continuation_ttl = continuation_ttl
        self._clock = clock or (lambda: datetime.now(timezone.utc))
        idempotency = getattr(getattr(capabilities, "adcp", None), "idempotency", None)
        idempotency_supported = getattr(idempotency, "supported", False) is True
        replay_ttl_seconds = getattr(idempotency, "replay_ttl_seconds", None)
        if idempotency_supported and replay_ttl_seconds is not None:
            if (
                isinstance(replay_ttl_seconds, bool)
                or not isinstance(replay_ttl_seconds, int)
                or not _MIN_IDEMPOTENCY_REPLAY_TTL_SECONDS
                <= replay_ttl_seconds
                <= _MAX_IDEMPOTENCY_REPLAY_TTL_SECONDS
            ):
                raise ValueError(
                    "capabilities adcp.idempotency.replay_ttl_seconds must be an integer "
                    f"in [{_MIN_IDEMPOTENCY_REPLAY_TTL_SECONDS}, "
                    f"{_MAX_IDEMPOTENCY_REPLAY_TTL_SECONDS}]"
                )
        self._mutation_idempotency_guaranteed = (
            idempotency_supported and replay_ttl_seconds is not None
        )
        # ``lifecycle`` describes the negotiated protocol family. The complete
        # catalog-purchase route is preflighted when an operation is requested.
        self.lifecycle = (
            MediaBuyLifecycle.COMPACT
            if _is_compact_release(self.negotiated_version)
            else MediaBuyLifecycle.ESTABLISHED
        )

    @classmethod
    async def negotiate(
        cls,
        client: ADCPClient,
        *,
        capabilities: GetAdcpCapabilitiesResponse | None = None,
        advertised_tools: Sequence[str] | None = None,
        legacy_purchase_coordinator: LegacyPurchaseCoordinator | None = None,
        principal_id: str | None = None,
        target_binding: str | None = None,
        preferred_lifecycle: LifecyclePreference = "auto",
        allowed_losses: Sequence[str] = (),
        continuation_ttl: timedelta = timedelta(minutes=5),
        clock: Callable[[], datetime] | None = None,
    ) -> MediaBuyLifecycleCoordinator:
        """Fetch seller evidence and construct a stable negotiated coordinator."""

        resolved_capabilities = capabilities or await client.fetch_capabilities()
        resolved_tools = (
            tuple(advertised_tools)
            if advertised_tools is not None
            else tuple(await client.list_tools())
        )
        return cls(
            client,
            resolved_capabilities,
            resolved_tools,
            legacy_purchase_coordinator=legacy_purchase_coordinator,
            principal_id=principal_id,
            target_binding=target_binding,
            preferred_lifecycle=preferred_lifecycle,
            allowed_losses=allowed_losses,
            continuation_ttl=continuation_ttl,
            clock=clock,
        )

    def report(self, operation: str | None = None) -> MediaBuyCompatibilityReport:
        """Describe a viable operation route without claiming a dispatch.

        With no operation this reports only the negotiated protocol family.
        Passing ``list_products`` or ``buy_products`` performs the same route
        preflight as that operation while leaving ``tools_used`` empty.
        """

        lifecycle = self.lifecycle
        if operation is not None:
            if operation not in _COMPACT_TO_ESTABLISHED:
                raise ValueError("operation must be list_products or buy_products")
            lifecycle = self._select_lifecycle(operation)
        return self._report(lifecycle, ())

    async def list_products(
        self,
        request: ListProductsRequest,
        *,
        issuance_idempotency_key: str | None = None,
    ) -> CompatibleCatalog:
        """List a catalog through the negotiated lifecycle.

        Established results issue a short-lived SDK continuation.  They never
        synthesize ``feed_version`` or ``pricing_version`` values.
        """

        request_payload = request.model_dump(mode="json", exclude_none=True)
        conditional = tuple(
            field for field in ("if_feed_version", "if_pricing_version") if field in request_payload
        )
        if conditional:
            raise self._unsupported(
                "list_products",
                self.lifecycle,
                "conditional_catalog_reuse",
                "The coordinator cannot safely bind a purchase to an unchanged response; "
                "omit " + ", ".join(conditional) + " and request a complete catalog page.",
            )
        lifecycle = self._select_lifecycle("list_products")
        account = _json_object(request.account)
        if lifecycle is MediaBuyLifecycle.COMPACT:
            pinned = request.model_copy(update={"adcp_version": self.negotiated_version})
            result = await self.client.list_products(pinned)
            payload = self._completed_payload("list_products", result, lifecycle)
            self._validate_served_version("list_products", payload, lifecycle)
            products = _products(payload, operation="list_products", coordinator=self)
            feed_version = _optional_text(payload.get("feed_version"))
            if feed_version is None:
                raise self._unsupported(
                    "list_products",
                    lifecycle,
                    "feed_version",
                    "The compact seller did not return the required real feed_version.",
                )
            return CompatibleCatalog(
                products=products,
                feed_version=feed_version,
                pricing_version=_optional_text(payload.get("pricing_version")),
                next_cursor=_optional_text(payload.get("next_cursor")),
                cache_scope=_optional_text(payload.get("cache_scope")),
                compatibility=self._report(lifecycle, ("list_products",)),
                raw=result,
                _account=account,
                _coordinator_identity=id(self),
                _source_schema_version=self.negotiated_version,
            )

        legacy = self._require_legacy("list_products", lifecycle)
        if account is None:
            raise self._unsupported(
                "list_products",
                lifecycle,
                "account",
                "Established catalog purchase coordination requires an account-bound listing.",
            )
        issue_key = issuance_idempotency_key or request.idempotency_key
        if not issue_key:
            raise self._unsupported(
                "list_products",
                lifecycle,
                "issuance_idempotency_key",
                "Established listing requires a stable issuance_idempotency_key.",
            )
        wire_request = self._legacy_list_request(request)
        source_version = self._exact_source_schema_version("list_products", lifecycle)
        try:
            _assert_declared_legacy_fields(
                wire_request,
                tool="get_products",
                source_version=source_version,
                nested=("filters",),
            )
        except ValueError as exc:
            raise self._unsupported(
                "list_products",
                lifecycle,
                "legacy_request_projection",
                str(exc),
            ) from exc
        validation = validate_request("get_products", wire_request, version=source_version)
        if not validation.valid or validation.variant == "skipped":
            raise self._unsupported(
                "list_products",
                lifecycle,
                "legacy_request_projection",
                "The listing requirements cannot be represented by the negotiated "
                f"{self.negotiated_version} get_products contract.",
            )
        model = LegacyGetProductsRequest.model_validate(wire_request)
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            legacy_result = await self.client.get_products_legacy(model)
        payload = self._completed_payload("list_products", legacy_result, lifecycle)
        self._validate_served_version("list_products", payload, lifecycle)
        response_validation = validate_response("get_products", payload, version=source_version)
        if not response_validation.valid or response_validation.variant == "skipped":
            raise self._unsupported(
                "list_products",
                lifecycle,
                "response_projection",
                "Established get_products returned an invalid exact-version response: "
                f"{format_issues(response_validation.issues)}",
            )
        products = _products(payload, operation="list_products", coordinator=self)
        continuation = (
            await self._issue_legacy_continuation(
                legacy=legacy,
                issuance_idempotency_key=issue_key,
                account=account,
                source_version=source_version,
                observed_request=wire_request,
                observed_response=payload,
                products=products,
            )
            if products
            else None
        )
        return CompatibleCatalog(
            products=products,
            feed_version=_optional_text(payload.get("wholesale_feed_version")),
            pricing_version=_optional_text(payload.get("pricing_version")),
            next_cursor=_optional_text(_mapping(payload.get("pagination")).get("cursor")),
            cache_scope=_optional_text(payload.get("cache_scope")),
            purchase_continuation=continuation,
            compatibility=self._report(lifecycle, ("get_products",)),
            raw=legacy_result,
            _account=account,
            _coordinator_identity=id(self),
            _source_schema_version=source_version,
        )

    async def buy_products(
        self,
        listing: CompatibleCatalog,
        request: CoordinatorBuyProductsInput | BaseModel,
        *,
        accepted_losses: Sequence[str] | None = None,
    ) -> CompatiblePurchaseResult:
        """Purchase selected products without changing either native signature."""

        self._validate_listing(listing)
        lifecycle = self._select_lifecycle("buy_products")
        if lifecycle is not listing.compatibility.lifecycle:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "lifecycle_changed",
                "The catalog route no longer matches the purchase route; restart discovery.",
            )
        payload = _json_object(request)
        assert payload is not None
        # Feed and pricing versions are seller-issued listing evidence. Never
        # accept caller substitutions, including when the listing omitted an
        # independent pricing version.
        payload.pop("feed_version", None)
        payload.pop("pricing_version", None)
        account = _mapping(payload.get("account"))
        if listing._account is not None and (
            not account
            or canonical_account_identity(account) != canonical_account_identity(listing._account)
        ):
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "account_binding",
                "The purchase account does not match the account used for listing.",
            )
        purchases = payload.get("purchases")
        try:
            selected_ids = _selected_product_ids(
                purchases, unique=lifecycle is MediaBuyLifecycle.ESTABLISHED
            )
        except ValueError as exc:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "purchase_selection",
                str(exc),
            ) from exc
        listed_ids = {product["product_id"] for product in listing.products}
        if not set(selected_ids).issubset(listed_ids):
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "selection_binding",
                "The purchase selects a product that was not present in this catalog result.",
            )

        if lifecycle is MediaBuyLifecycle.COMPACT:
            if listing.feed_version is None:
                raise self._unsupported(
                    "buy_products",
                    lifecycle,
                    "feed_version",
                    "A compact purchase requires the seller's real listing feed_version.",
                )
            compact_payload = {
                **payload,
                "feed_version": listing.feed_version,
                "adcp_version": self.negotiated_version,
            }
            if listing.pricing_version is not None:
                compact_payload["pricing_version"] = listing.pricing_version
            model = BuyProductsRequest.model_validate(compact_payload)
            result = await self.client.buy_products(model)
            return CompatiblePurchaseResult(
                result=result,
                compatibility=self._report(lifecycle, ("buy_products",)),
            )

        legacy = self._require_legacy("buy_products", lifecycle)
        if not _optional_text(payload.get("idempotency_key")):
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "idempotency_key",
                "Established purchase requires a non-empty idempotency_key.",
            )
        continuation = listing.purchase_continuation
        if continuation is None:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "purchase_continuation",
                "The established listing has no SDK purchase continuation.",
            )
        exact_losses = tuple(continuation.losses if accepted_losses is None else accepted_losses)
        if set(exact_losses) != set(continuation.losses) or len(exact_losses) != len(
            continuation.losses
        ):
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "accepted_losses",
                "accepted_losses must exactly match the listing continuation losses.",
                losses=continuation.losses,
            )
        refused = tuple(loss for loss in continuation.losses if loss not in self._allowed_losses)
        if refused:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "compatibility_losses",
                "No purchase was sent because required compatibility losses were not allowed.",
                losses=refused,
            )
        source_version = listing._source_schema_version
        if source_version is None:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "served_version_schema_unavailable",
                "The established listing did not retain its exact source schema version.",
            )
        try:
            legacy_create = _legacy_create_request(payload, self.negotiated_version, source_version)
        except ValueError as exc:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "legacy_request_projection",
                str(exc),
            ) from exc
        legacy_result = await legacy.continue_legacy_purchase(
            {
                "idempotency_key": payload.get("idempotency_key"),
                "continuation_token": continuation.token,
                "account": account,
                "selected_product_ids": selected_ids,
                "accepted_losses": list(exact_losses),
                "legacy_create_request": legacy_create,
            },
            principal_id=cast(str, self._principal_id),
            target_binding=cast(str, self._target_binding),
        )
        return CompatiblePurchaseResult(
            result=legacy_result,
            compatibility=self._report(
                lifecycle,
                ("create_media_buy",),
                losses=continuation.losses,
                warnings=(
                    "The established mutation cannot atomically fence the listed feed or pricing.",
                ),
            ),
        )

    async def recover_legacy_purchase(
        self, operation: CompatibilityPurchaseOperation
    ) -> CompatiblePurchaseResult:
        """Resume a fenced established purchase through the configured reconciler."""

        lifecycle = MediaBuyLifecycle.ESTABLISHED
        legacy = self._require_legacy("recover_legacy_purchase", lifecycle)
        record = await legacy.store.get_continuation(
            operation.token_hash,
            principal_id=cast(str, self._principal_id),
        )
        losses = (
            tuple(loss for loss in (*_PURCHASE_LOSSES, _MUTATION_LOSS) if loss in record.losses)
            if record is not None
            else ()
        )
        result = await legacy.recover_legacy_purchase(
            operation,
            principal_id=cast(str, self._principal_id),
            target_binding=cast(str, self._target_binding),
        )
        return CompatiblePurchaseResult(
            result=result,
            compatibility=self._report(
                lifecycle,
                ("create_media_buy",),
                losses=losses,
                warnings=(
                    "The recovered established mutation did not gain atomic feed or pricing "
                    "fencing.",
                ),
            ),
        )

    async def continue_legacy_purchase(
        self,
        continuation: LegacyCatalogContinuation | str,
        request: CoordinatorBuyProductsInput | BaseModel,
        *,
        accepted_losses: Sequence[str] | None = None,
    ) -> CompatiblePurchaseResult:
        """Redeem an established catalog continuation after coordinator restart."""

        lifecycle = self._select_lifecycle("buy_products")
        if lifecycle is not MediaBuyLifecycle.ESTABLISHED:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "legacy_continuation",
                "A legacy catalog continuation can only be redeemed on the established lane.",
            )
        token = (
            continuation.token
            if isinstance(continuation, LegacyCatalogContinuation)
            else continuation
        )
        if not isinstance(token, str) or not token:
            raise ValueError("continuation token must be non-empty")
        payload = _json_object(request)
        assert payload is not None
        payload.pop("feed_version", None)
        payload.pop("pricing_version", None)
        account = _mapping(payload.get("account"))
        if not account:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "account_binding",
                "Established continuation redemption requires an account.",
            )
        if not _optional_text(payload.get("idempotency_key")):
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "idempotency_key",
                "Established continuation redemption requires a non-empty idempotency_key.",
            )
        try:
            selected_ids = _selected_product_ids(payload.get("purchases"), unique=True)
        except ValueError as exc:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "purchase_selection",
                str(exc),
            ) from exc
        legacy = self._require_legacy("continue_legacy_purchase", lifecycle)
        record = await legacy.store.get_continuation(
            _token_hash(token), principal_id=cast(str, self._principal_id)
        )
        if record is None:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "purchase_continuation",
                "No continuation exists in this authenticated principal scope.",
            )
        source_version = self._exact_source_schema_version("continue_legacy_purchase", lifecycle)
        if record.source_adcp_version != source_version:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "served_version_changed",
                "The continuation was issued under a different served AdCP contract; "
                "reconstruct the coordinator with the original seller binding.",
            )
        loss_order = (*_PURCHASE_LOSSES, _MUTATION_LOSS)
        losses = tuple(loss for loss in loss_order if loss in record.losses)
        if set(losses) != set(record.losses):
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "compatibility_losses",
                "The continuation contains an unsupported compatibility loss set.",
            )
        exact_losses = tuple(losses if accepted_losses is None else accepted_losses)
        if set(exact_losses) != set(losses) or len(exact_losses) != len(losses):
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "accepted_losses",
                "accepted_losses must exactly match the established continuation losses.",
                losses=losses,
            )
        refused = tuple(loss for loss in losses if loss not in self._allowed_losses)
        if refused:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "compatibility_losses",
                "No purchase was sent because required compatibility losses were not allowed.",
                losses=refused,
            )
        try:
            legacy_create = _legacy_create_request(payload, self.negotiated_version, source_version)
        except ValueError as exc:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "legacy_request_projection",
                str(exc),
            ) from exc
        result = await legacy.continue_legacy_purchase(
            {
                "idempotency_key": payload["idempotency_key"],
                "continuation_token": token,
                "account": account,
                "selected_product_ids": selected_ids,
                "accepted_losses": list(exact_losses),
                "legacy_create_request": legacy_create,
            },
            principal_id=cast(str, self._principal_id),
            target_binding=cast(str, self._target_binding),
        )
        return CompatiblePurchaseResult(
            result=result,
            compatibility=self._report(
                lifecycle,
                ("create_media_buy",),
                losses=losses,
                warnings=(
                    "The established mutation cannot atomically fence the listed feed or pricing.",
                ),
            ),
        )

    async def _issue_legacy_continuation(
        self,
        *,
        legacy: LegacyPurchaseCoordinator,
        issuance_idempotency_key: str,
        account: JsonObject,
        source_version: str,
        observed_request: JsonObject,
        observed_response: JsonObject,
        products: Sequence[JsonObject],
    ) -> LegacyCatalogContinuation:
        losses = list(_PURCHASE_LOSSES)
        if not self._mutation_idempotency_guaranteed:
            losses.append(_MUTATION_LOSS)
        expires_at = _aware_utc(self._clock()) + self._continuation_ttl
        token = await legacy.issue_legacy_create_continuation(
            principal_id=cast(str, self._principal_id),
            issuance_idempotency_key=issuance_idempotency_key,
            account=account,
            source_adcp_version=source_version,
            expires_at=expires_at,
            observed_request=observed_request,
            observed_response=observed_response,
            product_ids=[product["product_id"] for product in products],
            buyer_visible_products=list(products),
            losses=losses,
            target_binding=cast(str, self._target_binding),
            mutation_idempotency_guaranteed=self._mutation_idempotency_guaranteed,
            listed_purchase_context={
                "negotiated_version": self.negotiated_version,
                "tools_used": ["get_products"],
            },
        )
        return LegacyCatalogContinuation(
            kind="legacy_create",
            token=token,
            losses=tuple(losses),
            expires_at=expires_at,
        )

    def _select_lifecycle(self, operation: str) -> MediaBuyLifecycle:
        return self._select_catalog_purchase_lifecycle(operation)

    def _select_catalog_purchase_lifecycle(self, operation: str) -> MediaBuyLifecycle:
        compact_contract = _is_compact_release(self.negotiated_version)
        _, release_minor = _release_minor(self.negotiated_version)
        # The 3.0 capabilities contract did not expose buying_modes. Wholesale
        # support is therefore discovered by the call on that exact legacy
        # contract; 3.1+ has an explicit capability and must advertise it.
        wholesale_supported = release_minor == 0 or "wholesale" in self._buying_modes
        compact_pair = compact_contract and {
            "list_products",
            "buy_products",
        }.issubset(self._tools)
        established_pair = (
            not compact_contract
            and {
                "get_products",
                "create_media_buy",
            }.issubset(self._tools)
            and wholesale_supported
        )

        if self._preferred_lifecycle == "compact":
            if compact_pair:
                return MediaBuyLifecycle.COMPACT
            raise self._unsupported(
                operation,
                MediaBuyLifecycle.COMPACT,
                "compact_catalog_purchase_not_advertised",
                "A compact catalog purchase requires both list_products and buy_products.",
            )
        if self._preferred_lifecycle == "established":
            if established_pair:
                return MediaBuyLifecycle.ESTABLISHED
            raise self._unsupported(
                operation,
                MediaBuyLifecycle.ESTABLISHED,
                "established_catalog_purchase_not_advertised",
                "An established catalog purchase requires get_products, create_media_buy, "
                "and the wholesale buying mode.",
            )
        if compact_pair:
            return MediaBuyLifecycle.COMPACT
        if established_pair:
            return MediaBuyLifecycle.ESTABLISHED
        raise self._unsupported(
            operation,
            MediaBuyLifecycle.COMPACT if compact_contract else MediaBuyLifecycle.ESTABLISHED,
            "catalog_purchase_route_not_advertised",
            "The seller advertises no complete compact or established catalog-purchase route.",
        )

    def _legacy_list_request(self, request: ListProductsRequest) -> JsonObject:
        value = request.model_dump(mode="json", exclude_none=True)
        criteria = _mapping(value.get("criteria"))
        unsupported = set(criteria) - {"offer_filters"}
        if unsupported:
            raise self._unsupported(
                "list_products",
                MediaBuyLifecycle.ESTABLISHED,
                "criteria",
                "Established listing cannot preserve criteria fields: "
                + ", ".join(sorted(unsupported)),
            )
        for field in ("governance_context", "context_id"):
            if field in value:
                raise self._unsupported(
                    "list_products",
                    MediaBuyLifecycle.ESTABLISHED,
                    field,
                    f"Established listing cannot preserve {field}.",
                )
        wire: JsonObject = {
            "adcp_version": self.negotiated_version,
            "adcp_major_version": 3,
            "buying_mode": "wholesale",
        }
        for field in ("account", "brand", "context", "push_notification_config"):
            if field in value:
                wire[field] = value[field]
        if "offer_filters" in criteria:
            wire["filters"] = criteria["offer_filters"]
        if "fields" in value:
            wire["fields"] = value["fields"]
        pagination = {key: value[key] for key in ("cursor", "max_results") if key in value}
        if pagination:
            wire["pagination"] = pagination
        if "if_feed_version" in value:
            wire["if_wholesale_feed_version"] = value["if_feed_version"]
        if "if_pricing_version" in value:
            wire["if_pricing_version"] = value["if_pricing_version"]
        return wire

    def _completed_payload(
        self,
        operation: str,
        result: TaskResult[Any],
        lifecycle: MediaBuyLifecycle,
    ) -> JsonObject:
        if not result.success or result.status is not TaskStatus.COMPLETED or result.data is None:
            raise self._unsupported(
                operation,
                lifecycle,
                "completed_catalog_required",
                f"{operation} did not return a completed successful response: "
                f"{result.error or result.message or result.status.value}",
            )
        return cast(
            JsonObject,
            result.data.model_dump(mode="json", exclude_none=True, exclude_unset=True),
        )

    def _validate_served_version(
        self, operation: str, payload: JsonObject, lifecycle: MediaBuyLifecycle
    ) -> None:
        served = payload.get("adcp_version")
        if served is None:
            # The capabilities response is the served-contract binding. Some
            # task response schemas do not repeat the envelope field.
            return
        if not isinstance(served, str) or _parse_release(served) is None:
            raise self._unsupported(
                operation,
                lifecycle,
                "served_version_invalid",
                "The seller returned an invalid response adcp_version.",
            )
        normalized = normalize_to_release_precision(served)
        if normalized != self.negotiated_version:
            raise self._unsupported(
                operation,
                lifecycle,
                "served_version_mismatch",
                f"The seller served AdCP {normalized}, but "
                f"{self.negotiated_version} was negotiated.",
            )

    def _exact_source_schema_version(self, operation: str, lifecycle: MediaBuyLifecycle) -> str:
        exact = get_bundle_adcp_version(version=self.negotiated_version)
        if exact is None:
            raise self._unsupported(
                operation,
                lifecycle,
                "served_version_schema_unavailable",
                f"No bundled schema can validate AdCP {self.negotiated_version}.",
            )
        return exact

    def _validate_listing(self, listing: CompatibleCatalog) -> None:
        if not isinstance(listing, CompatibleCatalog) or listing._coordinator_identity != id(self):
            raise TypeError("listing must be a CompatibleCatalog issued by this coordinator")

    def _require_legacy(
        self, operation: str, lifecycle: MediaBuyLifecycle
    ) -> LegacyPurchaseCoordinator:
        if self._legacy is None:
            raise self._unsupported(
                operation,
                lifecycle,
                "legacy_purchase_coordinator",
                "Established purchase coordination requires a LegacyPurchaseCoordinator.",
            )
        return self._legacy

    def _report(
        self,
        lifecycle: MediaBuyLifecycle,
        tools: Sequence[str],
        *,
        losses: Sequence[str] = (),
        warnings: Sequence[str] = (),
    ) -> MediaBuyCompatibilityReport:
        compatibility = (
            MediaBuyCompatibility.NATIVE
            if lifecycle is MediaBuyLifecycle.COMPACT
            else (
                MediaBuyCompatibility.LOSSY_PROJECTION
                if losses
                else MediaBuyCompatibility.LOSSLESS_PROJECTION
            )
        )
        return MediaBuyCompatibilityReport(
            negotiated_version=self.negotiated_version,
            lifecycle=lifecycle,
            tools_used=tuple(tools),
            compatibility=compatibility,
            warnings=tuple(warnings),
            losses=tuple(losses),
        )

    def _unsupported(
        self,
        operation: str,
        lifecycle: MediaBuyLifecycle,
        feature: str,
        message: str,
        *,
        losses: Sequence[str] = (),
        code: CompatibilityErrorCode = "UNSUPPORTED_FEATURE",
    ) -> MediaBuyLifecycleCompatibilityError:
        return MediaBuyLifecycleCompatibilityError(
            operation=operation,
            negotiated_version=self.negotiated_version,
            lifecycle=lifecycle,
            feature=feature,
            message=message,
            losses=losses,
            code=code,
        )

Negotiate one catalog-to-purchase workflow across AdCP 3.x.

Static methods

async def negotiate(client: ADCPClient,
*,
capabilities: GetAdcpCapabilitiesResponse | None = None,
advertised_tools: Sequence[str] | None = None,
legacy_purchase_coordinator: LegacyPurchaseCoordinator | None = None,
principal_id: str | None = None,
target_binding: str | None = None,
preferred_lifecycle: LifecyclePreference = 'auto',
allowed_losses: Sequence[str] = (),
continuation_ttl: timedelta = datetime.timedelta(seconds=300),
clock: Callable[[], datetime] | None = None)

Fetch seller evidence and construct a stable negotiated coordinator.

Methods

async def buy_products(self,
listing: CompatibleCatalog,
request: CoordinatorBuyProductsInput | BaseModel,
*,
accepted_losses: Sequence[str] | None = None) ‑> CompatiblePurchaseResult

Expand source code

async def buy_products(
    self,
    listing: CompatibleCatalog,
    request: CoordinatorBuyProductsInput | BaseModel,
    *,
    accepted_losses: Sequence[str] | None = None,
) -> CompatiblePurchaseResult:
    """Purchase selected products without changing either native signature."""

    self._validate_listing(listing)
    lifecycle = self._select_lifecycle("buy_products")
    if lifecycle is not listing.compatibility.lifecycle:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "lifecycle_changed",
            "The catalog route no longer matches the purchase route; restart discovery.",
        )
    payload = _json_object(request)
    assert payload is not None
    # Feed and pricing versions are seller-issued listing evidence. Never
    # accept caller substitutions, including when the listing omitted an
    # independent pricing version.
    payload.pop("feed_version", None)
    payload.pop("pricing_version", None)
    account = _mapping(payload.get("account"))
    if listing._account is not None and (
        not account
        or canonical_account_identity(account) != canonical_account_identity(listing._account)
    ):
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "account_binding",
            "The purchase account does not match the account used for listing.",
        )
    purchases = payload.get("purchases")
    try:
        selected_ids = _selected_product_ids(
            purchases, unique=lifecycle is MediaBuyLifecycle.ESTABLISHED
        )
    except ValueError as exc:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "purchase_selection",
            str(exc),
        ) from exc
    listed_ids = {product["product_id"] for product in listing.products}
    if not set(selected_ids).issubset(listed_ids):
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "selection_binding",
            "The purchase selects a product that was not present in this catalog result.",
        )

    if lifecycle is MediaBuyLifecycle.COMPACT:
        if listing.feed_version is None:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "feed_version",
                "A compact purchase requires the seller's real listing feed_version.",
            )
        compact_payload = {
            **payload,
            "feed_version": listing.feed_version,
            "adcp_version": self.negotiated_version,
        }
        if listing.pricing_version is not None:
            compact_payload["pricing_version"] = listing.pricing_version
        model = BuyProductsRequest.model_validate(compact_payload)
        result = await self.client.buy_products(model)
        return CompatiblePurchaseResult(
            result=result,
            compatibility=self._report(lifecycle, ("buy_products",)),
        )

    legacy = self._require_legacy("buy_products", lifecycle)
    if not _optional_text(payload.get("idempotency_key")):
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "idempotency_key",
            "Established purchase requires a non-empty idempotency_key.",
        )
    continuation = listing.purchase_continuation
    if continuation is None:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "purchase_continuation",
            "The established listing has no SDK purchase continuation.",
        )
    exact_losses = tuple(continuation.losses if accepted_losses is None else accepted_losses)
    if set(exact_losses) != set(continuation.losses) or len(exact_losses) != len(
        continuation.losses
    ):
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "accepted_losses",
            "accepted_losses must exactly match the listing continuation losses.",
            losses=continuation.losses,
        )
    refused = tuple(loss for loss in continuation.losses if loss not in self._allowed_losses)
    if refused:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "compatibility_losses",
            "No purchase was sent because required compatibility losses were not allowed.",
            losses=refused,
        )
    source_version = listing._source_schema_version
    if source_version is None:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "served_version_schema_unavailable",
            "The established listing did not retain its exact source schema version.",
        )
    try:
        legacy_create = _legacy_create_request(payload, self.negotiated_version, source_version)
    except ValueError as exc:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "legacy_request_projection",
            str(exc),
        ) from exc
    legacy_result = await legacy.continue_legacy_purchase(
        {
            "idempotency_key": payload.get("idempotency_key"),
            "continuation_token": continuation.token,
            "account": account,
            "selected_product_ids": selected_ids,
            "accepted_losses": list(exact_losses),
            "legacy_create_request": legacy_create,
        },
        principal_id=cast(str, self._principal_id),
        target_binding=cast(str, self._target_binding),
    )
    return CompatiblePurchaseResult(
        result=legacy_result,
        compatibility=self._report(
            lifecycle,
            ("create_media_buy",),
            losses=continuation.losses,
            warnings=(
                "The established mutation cannot atomically fence the listed feed or pricing.",
            ),
        ),
    )

Purchase selected products without changing either native signature.

async def continue_legacy_purchase(self,
continuation: LegacyCatalogContinuation | str,
request: CoordinatorBuyProductsInput | BaseModel,
*,
accepted_losses: Sequence[str] | None = None) ‑> CompatiblePurchaseResult

Expand source code

async def continue_legacy_purchase(
    self,
    continuation: LegacyCatalogContinuation | str,
    request: CoordinatorBuyProductsInput | BaseModel,
    *,
    accepted_losses: Sequence[str] | None = None,
) -> CompatiblePurchaseResult:
    """Redeem an established catalog continuation after coordinator restart."""

    lifecycle = self._select_lifecycle("buy_products")
    if lifecycle is not MediaBuyLifecycle.ESTABLISHED:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "legacy_continuation",
            "A legacy catalog continuation can only be redeemed on the established lane.",
        )
    token = (
        continuation.token
        if isinstance(continuation, LegacyCatalogContinuation)
        else continuation
    )
    if not isinstance(token, str) or not token:
        raise ValueError("continuation token must be non-empty")
    payload = _json_object(request)
    assert payload is not None
    payload.pop("feed_version", None)
    payload.pop("pricing_version", None)
    account = _mapping(payload.get("account"))
    if not account:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "account_binding",
            "Established continuation redemption requires an account.",
        )
    if not _optional_text(payload.get("idempotency_key")):
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "idempotency_key",
            "Established continuation redemption requires a non-empty idempotency_key.",
        )
    try:
        selected_ids = _selected_product_ids(payload.get("purchases"), unique=True)
    except ValueError as exc:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "purchase_selection",
            str(exc),
        ) from exc
    legacy = self._require_legacy("continue_legacy_purchase", lifecycle)
    record = await legacy.store.get_continuation(
        _token_hash(token), principal_id=cast(str, self._principal_id)
    )
    if record is None:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "purchase_continuation",
            "No continuation exists in this authenticated principal scope.",
        )
    source_version = self._exact_source_schema_version("continue_legacy_purchase", lifecycle)
    if record.source_adcp_version != source_version:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "served_version_changed",
            "The continuation was issued under a different served AdCP contract; "
            "reconstruct the coordinator with the original seller binding.",
        )
    loss_order = (*_PURCHASE_LOSSES, _MUTATION_LOSS)
    losses = tuple(loss for loss in loss_order if loss in record.losses)
    if set(losses) != set(record.losses):
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "compatibility_losses",
            "The continuation contains an unsupported compatibility loss set.",
        )
    exact_losses = tuple(losses if accepted_losses is None else accepted_losses)
    if set(exact_losses) != set(losses) or len(exact_losses) != len(losses):
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "accepted_losses",
            "accepted_losses must exactly match the established continuation losses.",
            losses=losses,
        )
    refused = tuple(loss for loss in losses if loss not in self._allowed_losses)
    if refused:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "compatibility_losses",
            "No purchase was sent because required compatibility losses were not allowed.",
            losses=refused,
        )
    try:
        legacy_create = _legacy_create_request(payload, self.negotiated_version, source_version)
    except ValueError as exc:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "legacy_request_projection",
            str(exc),
        ) from exc
    result = await legacy.continue_legacy_purchase(
        {
            "idempotency_key": payload["idempotency_key"],
            "continuation_token": token,
            "account": account,
            "selected_product_ids": selected_ids,
            "accepted_losses": list(exact_losses),
            "legacy_create_request": legacy_create,
        },
        principal_id=cast(str, self._principal_id),
        target_binding=cast(str, self._target_binding),
    )
    return CompatiblePurchaseResult(
        result=result,
        compatibility=self._report(
            lifecycle,
            ("create_media_buy",),
            losses=losses,
            warnings=(
                "The established mutation cannot atomically fence the listed feed or pricing.",
            ),
        ),
    )

Redeem an established catalog continuation after coordinator restart.

async def list_products(self,
request: ListProductsRequest,
*,
issuance_idempotency_key: str | None = None) ‑> CompatibleCatalog

Expand source code

async def list_products(
    self,
    request: ListProductsRequest,
    *,
    issuance_idempotency_key: str | None = None,
) -> CompatibleCatalog:
    """List a catalog through the negotiated lifecycle.

    Established results issue a short-lived SDK continuation.  They never
    synthesize ``feed_version`` or ``pricing_version`` values.
    """

    request_payload = request.model_dump(mode="json", exclude_none=True)
    conditional = tuple(
        field for field in ("if_feed_version", "if_pricing_version") if field in request_payload
    )
    if conditional:
        raise self._unsupported(
            "list_products",
            self.lifecycle,
            "conditional_catalog_reuse",
            "The coordinator cannot safely bind a purchase to an unchanged response; "
            "omit " + ", ".join(conditional) + " and request a complete catalog page.",
        )
    lifecycle = self._select_lifecycle("list_products")
    account = _json_object(request.account)
    if lifecycle is MediaBuyLifecycle.COMPACT:
        pinned = request.model_copy(update={"adcp_version": self.negotiated_version})
        result = await self.client.list_products(pinned)
        payload = self._completed_payload("list_products", result, lifecycle)
        self._validate_served_version("list_products", payload, lifecycle)
        products = _products(payload, operation="list_products", coordinator=self)
        feed_version = _optional_text(payload.get("feed_version"))
        if feed_version is None:
            raise self._unsupported(
                "list_products",
                lifecycle,
                "feed_version",
                "The compact seller did not return the required real feed_version.",
            )
        return CompatibleCatalog(
            products=products,
            feed_version=feed_version,
            pricing_version=_optional_text(payload.get("pricing_version")),
            next_cursor=_optional_text(payload.get("next_cursor")),
            cache_scope=_optional_text(payload.get("cache_scope")),
            compatibility=self._report(lifecycle, ("list_products",)),
            raw=result,
            _account=account,
            _coordinator_identity=id(self),
            _source_schema_version=self.negotiated_version,
        )

    legacy = self._require_legacy("list_products", lifecycle)
    if account is None:
        raise self._unsupported(
            "list_products",
            lifecycle,
            "account",
            "Established catalog purchase coordination requires an account-bound listing.",
        )
    issue_key = issuance_idempotency_key or request.idempotency_key
    if not issue_key:
        raise self._unsupported(
            "list_products",
            lifecycle,
            "issuance_idempotency_key",
            "Established listing requires a stable issuance_idempotency_key.",
        )
    wire_request = self._legacy_list_request(request)
    source_version = self._exact_source_schema_version("list_products", lifecycle)
    try:
        _assert_declared_legacy_fields(
            wire_request,
            tool="get_products",
            source_version=source_version,
            nested=("filters",),
        )
    except ValueError as exc:
        raise self._unsupported(
            "list_products",
            lifecycle,
            "legacy_request_projection",
            str(exc),
        ) from exc
    validation = validate_request("get_products", wire_request, version=source_version)
    if not validation.valid or validation.variant == "skipped":
        raise self._unsupported(
            "list_products",
            lifecycle,
            "legacy_request_projection",
            "The listing requirements cannot be represented by the negotiated "
            f"{self.negotiated_version} get_products contract.",
        )
    model = LegacyGetProductsRequest.model_validate(wire_request)
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", DeprecationWarning)
        legacy_result = await self.client.get_products_legacy(model)
    payload = self._completed_payload("list_products", legacy_result, lifecycle)
    self._validate_served_version("list_products", payload, lifecycle)
    response_validation = validate_response("get_products", payload, version=source_version)
    if not response_validation.valid or response_validation.variant == "skipped":
        raise self._unsupported(
            "list_products",
            lifecycle,
            "response_projection",
            "Established get_products returned an invalid exact-version response: "
            f"{format_issues(response_validation.issues)}",
        )
    products = _products(payload, operation="list_products", coordinator=self)
    continuation = (
        await self._issue_legacy_continuation(
            legacy=legacy,
            issuance_idempotency_key=issue_key,
            account=account,
            source_version=source_version,
            observed_request=wire_request,
            observed_response=payload,
            products=products,
        )
        if products
        else None
    )
    return CompatibleCatalog(
        products=products,
        feed_version=_optional_text(payload.get("wholesale_feed_version")),
        pricing_version=_optional_text(payload.get("pricing_version")),
        next_cursor=_optional_text(_mapping(payload.get("pagination")).get("cursor")),
        cache_scope=_optional_text(payload.get("cache_scope")),
        purchase_continuation=continuation,
        compatibility=self._report(lifecycle, ("get_products",)),
        raw=legacy_result,
        _account=account,
        _coordinator_identity=id(self),
        _source_schema_version=source_version,
    )

List a catalog through the negotiated lifecycle.

Established results issue a short-lived SDK continuation. They never synthesize feed_version or pricing_version values.

async def recover_legacy_purchase(self,
operation: CompatibilityPurchaseOperation) ‑> CompatiblePurchaseResult

Expand source code

async def recover_legacy_purchase(
    self, operation: CompatibilityPurchaseOperation
) -> CompatiblePurchaseResult:
    """Resume a fenced established purchase through the configured reconciler."""

    lifecycle = MediaBuyLifecycle.ESTABLISHED
    legacy = self._require_legacy("recover_legacy_purchase", lifecycle)
    record = await legacy.store.get_continuation(
        operation.token_hash,
        principal_id=cast(str, self._principal_id),
    )
    losses = (
        tuple(loss for loss in (*_PURCHASE_LOSSES, _MUTATION_LOSS) if loss in record.losses)
        if record is not None
        else ()
    )
    result = await legacy.recover_legacy_purchase(
        operation,
        principal_id=cast(str, self._principal_id),
        target_binding=cast(str, self._target_binding),
    )
    return CompatiblePurchaseResult(
        result=result,
        compatibility=self._report(
            lifecycle,
            ("create_media_buy",),
            losses=losses,
            warnings=(
                "The recovered established mutation did not gain atomic feed or pricing "
                "fencing.",
            ),
        ),
    )

Resume a fenced established purchase through the configured reconciler.

def report(self, operation: str | None = None) ‑> MediaBuyCompatibilityReport

Expand source code

def report(self, operation: str | None = None) -> MediaBuyCompatibilityReport:
    """Describe a viable operation route without claiming a dispatch.

    With no operation this reports only the negotiated protocol family.
    Passing ``list_products`` or ``buy_products`` performs the same route
    preflight as that operation while leaving ``tools_used`` empty.
    """

    lifecycle = self.lifecycle
    if operation is not None:
        if operation not in _COMPACT_TO_ESTABLISHED:
            raise ValueError("operation must be list_products or buy_products")
        lifecycle = self._select_lifecycle(operation)
    return self._report(lifecycle, ())

Describe a viable operation route without claiming a dispatch.

With no operation this reports only the negotiated protocol family. Passing list_products or buy_products performs the same route preflight as that operation while leaving tools_used empty.

class NegotiatedCatalogBuyer (client: ADCPClient,
capabilities: GetAdcpCapabilitiesResponse,
advertised_tools: Sequence[str],
*,
legacy_purchase_coordinator: LegacyPurchaseCoordinator | None = None,
principal_id: str | None = None,
target_binding: str | None = None,
preferred_lifecycle: LifecyclePreference = 'auto',
allowed_losses: Sequence[str] = (),
continuation_ttl: timedelta = datetime.timedelta(seconds=300),
clock: Callable[[], datetime] | None = None)

Expand source code

class MediaBuyLifecycleCoordinator:
    """Negotiate one catalog-to-purchase workflow across AdCP 3.x."""

    def __init__(
        self,
        client: ADCPClient,
        capabilities: GetAdcpCapabilitiesResponse,
        advertised_tools: Sequence[str],
        *,
        legacy_purchase_coordinator: LegacyPurchaseCoordinator | None = None,
        principal_id: str | None = None,
        target_binding: str | None = None,
        preferred_lifecycle: LifecyclePreference = "auto",
        allowed_losses: Sequence[str] = (),
        continuation_ttl: timedelta = timedelta(minutes=5),
        clock: Callable[[], datetime] | None = None,
    ) -> None:
        if preferred_lifecycle not in {"auto", "compact", "established"}:
            raise ValueError("preferred_lifecycle must be auto, compact, or established")
        if continuation_ttl <= timedelta(0):
            raise ValueError("continuation_ttl must be positive")
        if principal_id is not None:
            if not isinstance(principal_id, str) or not principal_id.strip():
                raise TypeError("principal_id must be a non-empty string when provided")
            principal_id = principal_id.strip()
            if re.search(r"[\x00-\x1f\x7f]", principal_id):
                raise ValueError("principal_id must not contain control characters")
            if len(principal_id.encode("utf-8")) > _MAX_PRINCIPAL_BYTES:
                raise ValueError(f"principal_id must be at most {_MAX_PRINCIPAL_BYTES} UTF-8 bytes")
        if target_binding is not None:
            if not isinstance(target_binding, str) or not target_binding.strip():
                raise TypeError("target_binding must be a non-empty string when provided")
            target_binding = target_binding.strip()
        if legacy_purchase_coordinator is not None and (not principal_id or not target_binding):
            raise ValueError(
                "compatibility persistence requires non-empty principal_id and target_binding"
            )

        self.client = client
        self.capabilities = capabilities
        self.negotiated_version = _negotiate_version(capabilities, client.get_adcp_version())
        self._tools = frozenset(str(tool) for tool in advertised_tools)
        media_buy = getattr(capabilities, "media_buy", None)
        self._tools = self._tools | frozenset(
            _root_value(tool) for tool in (getattr(media_buy, "lifecycle_tools", None) or ())
        )
        self._buying_modes = frozenset(
            _root_value(mode) for mode in (getattr(media_buy, "buying_modes", None) or ())
        )
        self._preferred_lifecycle = preferred_lifecycle
        self._allowed_losses = frozenset(str(loss) for loss in allowed_losses)
        self._legacy = legacy_purchase_coordinator
        self._principal_id = principal_id
        self._target_binding = target_binding
        self._continuation_ttl = continuation_ttl
        self._clock = clock or (lambda: datetime.now(timezone.utc))
        idempotency = getattr(getattr(capabilities, "adcp", None), "idempotency", None)
        idempotency_supported = getattr(idempotency, "supported", False) is True
        replay_ttl_seconds = getattr(idempotency, "replay_ttl_seconds", None)
        if idempotency_supported and replay_ttl_seconds is not None:
            if (
                isinstance(replay_ttl_seconds, bool)
                or not isinstance(replay_ttl_seconds, int)
                or not _MIN_IDEMPOTENCY_REPLAY_TTL_SECONDS
                <= replay_ttl_seconds
                <= _MAX_IDEMPOTENCY_REPLAY_TTL_SECONDS
            ):
                raise ValueError(
                    "capabilities adcp.idempotency.replay_ttl_seconds must be an integer "
                    f"in [{_MIN_IDEMPOTENCY_REPLAY_TTL_SECONDS}, "
                    f"{_MAX_IDEMPOTENCY_REPLAY_TTL_SECONDS}]"
                )
        self._mutation_idempotency_guaranteed = (
            idempotency_supported and replay_ttl_seconds is not None
        )
        # ``lifecycle`` describes the negotiated protocol family. The complete
        # catalog-purchase route is preflighted when an operation is requested.
        self.lifecycle = (
            MediaBuyLifecycle.COMPACT
            if _is_compact_release(self.negotiated_version)
            else MediaBuyLifecycle.ESTABLISHED
        )

    @classmethod
    async def negotiate(
        cls,
        client: ADCPClient,
        *,
        capabilities: GetAdcpCapabilitiesResponse | None = None,
        advertised_tools: Sequence[str] | None = None,
        legacy_purchase_coordinator: LegacyPurchaseCoordinator | None = None,
        principal_id: str | None = None,
        target_binding: str | None = None,
        preferred_lifecycle: LifecyclePreference = "auto",
        allowed_losses: Sequence[str] = (),
        continuation_ttl: timedelta = timedelta(minutes=5),
        clock: Callable[[], datetime] | None = None,
    ) -> MediaBuyLifecycleCoordinator:
        """Fetch seller evidence and construct a stable negotiated coordinator."""

        resolved_capabilities = capabilities or await client.fetch_capabilities()
        resolved_tools = (
            tuple(advertised_tools)
            if advertised_tools is not None
            else tuple(await client.list_tools())
        )
        return cls(
            client,
            resolved_capabilities,
            resolved_tools,
            legacy_purchase_coordinator=legacy_purchase_coordinator,
            principal_id=principal_id,
            target_binding=target_binding,
            preferred_lifecycle=preferred_lifecycle,
            allowed_losses=allowed_losses,
            continuation_ttl=continuation_ttl,
            clock=clock,
        )

    def report(self, operation: str | None = None) -> MediaBuyCompatibilityReport:
        """Describe a viable operation route without claiming a dispatch.

        With no operation this reports only the negotiated protocol family.
        Passing ``list_products`` or ``buy_products`` performs the same route
        preflight as that operation while leaving ``tools_used`` empty.
        """

        lifecycle = self.lifecycle
        if operation is not None:
            if operation not in _COMPACT_TO_ESTABLISHED:
                raise ValueError("operation must be list_products or buy_products")
            lifecycle = self._select_lifecycle(operation)
        return self._report(lifecycle, ())

    async def list_products(
        self,
        request: ListProductsRequest,
        *,
        issuance_idempotency_key: str | None = None,
    ) -> CompatibleCatalog:
        """List a catalog through the negotiated lifecycle.

        Established results issue a short-lived SDK continuation.  They never
        synthesize ``feed_version`` or ``pricing_version`` values.
        """

        request_payload = request.model_dump(mode="json", exclude_none=True)
        conditional = tuple(
            field for field in ("if_feed_version", "if_pricing_version") if field in request_payload
        )
        if conditional:
            raise self._unsupported(
                "list_products",
                self.lifecycle,
                "conditional_catalog_reuse",
                "The coordinator cannot safely bind a purchase to an unchanged response; "
                "omit " + ", ".join(conditional) + " and request a complete catalog page.",
            )
        lifecycle = self._select_lifecycle("list_products")
        account = _json_object(request.account)
        if lifecycle is MediaBuyLifecycle.COMPACT:
            pinned = request.model_copy(update={"adcp_version": self.negotiated_version})
            result = await self.client.list_products(pinned)
            payload = self._completed_payload("list_products", result, lifecycle)
            self._validate_served_version("list_products", payload, lifecycle)
            products = _products(payload, operation="list_products", coordinator=self)
            feed_version = _optional_text(payload.get("feed_version"))
            if feed_version is None:
                raise self._unsupported(
                    "list_products",
                    lifecycle,
                    "feed_version",
                    "The compact seller did not return the required real feed_version.",
                )
            return CompatibleCatalog(
                products=products,
                feed_version=feed_version,
                pricing_version=_optional_text(payload.get("pricing_version")),
                next_cursor=_optional_text(payload.get("next_cursor")),
                cache_scope=_optional_text(payload.get("cache_scope")),
                compatibility=self._report(lifecycle, ("list_products",)),
                raw=result,
                _account=account,
                _coordinator_identity=id(self),
                _source_schema_version=self.negotiated_version,
            )

        legacy = self._require_legacy("list_products", lifecycle)
        if account is None:
            raise self._unsupported(
                "list_products",
                lifecycle,
                "account",
                "Established catalog purchase coordination requires an account-bound listing.",
            )
        issue_key = issuance_idempotency_key or request.idempotency_key
        if not issue_key:
            raise self._unsupported(
                "list_products",
                lifecycle,
                "issuance_idempotency_key",
                "Established listing requires a stable issuance_idempotency_key.",
            )
        wire_request = self._legacy_list_request(request)
        source_version = self._exact_source_schema_version("list_products", lifecycle)
        try:
            _assert_declared_legacy_fields(
                wire_request,
                tool="get_products",
                source_version=source_version,
                nested=("filters",),
            )
        except ValueError as exc:
            raise self._unsupported(
                "list_products",
                lifecycle,
                "legacy_request_projection",
                str(exc),
            ) from exc
        validation = validate_request("get_products", wire_request, version=source_version)
        if not validation.valid or validation.variant == "skipped":
            raise self._unsupported(
                "list_products",
                lifecycle,
                "legacy_request_projection",
                "The listing requirements cannot be represented by the negotiated "
                f"{self.negotiated_version} get_products contract.",
            )
        model = LegacyGetProductsRequest.model_validate(wire_request)
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            legacy_result = await self.client.get_products_legacy(model)
        payload = self._completed_payload("list_products", legacy_result, lifecycle)
        self._validate_served_version("list_products", payload, lifecycle)
        response_validation = validate_response("get_products", payload, version=source_version)
        if not response_validation.valid or response_validation.variant == "skipped":
            raise self._unsupported(
                "list_products",
                lifecycle,
                "response_projection",
                "Established get_products returned an invalid exact-version response: "
                f"{format_issues(response_validation.issues)}",
            )
        products = _products(payload, operation="list_products", coordinator=self)
        continuation = (
            await self._issue_legacy_continuation(
                legacy=legacy,
                issuance_idempotency_key=issue_key,
                account=account,
                source_version=source_version,
                observed_request=wire_request,
                observed_response=payload,
                products=products,
            )
            if products
            else None
        )
        return CompatibleCatalog(
            products=products,
            feed_version=_optional_text(payload.get("wholesale_feed_version")),
            pricing_version=_optional_text(payload.get("pricing_version")),
            next_cursor=_optional_text(_mapping(payload.get("pagination")).get("cursor")),
            cache_scope=_optional_text(payload.get("cache_scope")),
            purchase_continuation=continuation,
            compatibility=self._report(lifecycle, ("get_products",)),
            raw=legacy_result,
            _account=account,
            _coordinator_identity=id(self),
            _source_schema_version=source_version,
        )

    async def buy_products(
        self,
        listing: CompatibleCatalog,
        request: CoordinatorBuyProductsInput | BaseModel,
        *,
        accepted_losses: Sequence[str] | None = None,
    ) -> CompatiblePurchaseResult:
        """Purchase selected products without changing either native signature."""

        self._validate_listing(listing)
        lifecycle = self._select_lifecycle("buy_products")
        if lifecycle is not listing.compatibility.lifecycle:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "lifecycle_changed",
                "The catalog route no longer matches the purchase route; restart discovery.",
            )
        payload = _json_object(request)
        assert payload is not None
        # Feed and pricing versions are seller-issued listing evidence. Never
        # accept caller substitutions, including when the listing omitted an
        # independent pricing version.
        payload.pop("feed_version", None)
        payload.pop("pricing_version", None)
        account = _mapping(payload.get("account"))
        if listing._account is not None and (
            not account
            or canonical_account_identity(account) != canonical_account_identity(listing._account)
        ):
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "account_binding",
                "The purchase account does not match the account used for listing.",
            )
        purchases = payload.get("purchases")
        try:
            selected_ids = _selected_product_ids(
                purchases, unique=lifecycle is MediaBuyLifecycle.ESTABLISHED
            )
        except ValueError as exc:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "purchase_selection",
                str(exc),
            ) from exc
        listed_ids = {product["product_id"] for product in listing.products}
        if not set(selected_ids).issubset(listed_ids):
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "selection_binding",
                "The purchase selects a product that was not present in this catalog result.",
            )

        if lifecycle is MediaBuyLifecycle.COMPACT:
            if listing.feed_version is None:
                raise self._unsupported(
                    "buy_products",
                    lifecycle,
                    "feed_version",
                    "A compact purchase requires the seller's real listing feed_version.",
                )
            compact_payload = {
                **payload,
                "feed_version": listing.feed_version,
                "adcp_version": self.negotiated_version,
            }
            if listing.pricing_version is not None:
                compact_payload["pricing_version"] = listing.pricing_version
            model = BuyProductsRequest.model_validate(compact_payload)
            result = await self.client.buy_products(model)
            return CompatiblePurchaseResult(
                result=result,
                compatibility=self._report(lifecycle, ("buy_products",)),
            )

        legacy = self._require_legacy("buy_products", lifecycle)
        if not _optional_text(payload.get("idempotency_key")):
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "idempotency_key",
                "Established purchase requires a non-empty idempotency_key.",
            )
        continuation = listing.purchase_continuation
        if continuation is None:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "purchase_continuation",
                "The established listing has no SDK purchase continuation.",
            )
        exact_losses = tuple(continuation.losses if accepted_losses is None else accepted_losses)
        if set(exact_losses) != set(continuation.losses) or len(exact_losses) != len(
            continuation.losses
        ):
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "accepted_losses",
                "accepted_losses must exactly match the listing continuation losses.",
                losses=continuation.losses,
            )
        refused = tuple(loss for loss in continuation.losses if loss not in self._allowed_losses)
        if refused:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "compatibility_losses",
                "No purchase was sent because required compatibility losses were not allowed.",
                losses=refused,
            )
        source_version = listing._source_schema_version
        if source_version is None:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "served_version_schema_unavailable",
                "The established listing did not retain its exact source schema version.",
            )
        try:
            legacy_create = _legacy_create_request(payload, self.negotiated_version, source_version)
        except ValueError as exc:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "legacy_request_projection",
                str(exc),
            ) from exc
        legacy_result = await legacy.continue_legacy_purchase(
            {
                "idempotency_key": payload.get("idempotency_key"),
                "continuation_token": continuation.token,
                "account": account,
                "selected_product_ids": selected_ids,
                "accepted_losses": list(exact_losses),
                "legacy_create_request": legacy_create,
            },
            principal_id=cast(str, self._principal_id),
            target_binding=cast(str, self._target_binding),
        )
        return CompatiblePurchaseResult(
            result=legacy_result,
            compatibility=self._report(
                lifecycle,
                ("create_media_buy",),
                losses=continuation.losses,
                warnings=(
                    "The established mutation cannot atomically fence the listed feed or pricing.",
                ),
            ),
        )

    async def recover_legacy_purchase(
        self, operation: CompatibilityPurchaseOperation
    ) -> CompatiblePurchaseResult:
        """Resume a fenced established purchase through the configured reconciler."""

        lifecycle = MediaBuyLifecycle.ESTABLISHED
        legacy = self._require_legacy("recover_legacy_purchase", lifecycle)
        record = await legacy.store.get_continuation(
            operation.token_hash,
            principal_id=cast(str, self._principal_id),
        )
        losses = (
            tuple(loss for loss in (*_PURCHASE_LOSSES, _MUTATION_LOSS) if loss in record.losses)
            if record is not None
            else ()
        )
        result = await legacy.recover_legacy_purchase(
            operation,
            principal_id=cast(str, self._principal_id),
            target_binding=cast(str, self._target_binding),
        )
        return CompatiblePurchaseResult(
            result=result,
            compatibility=self._report(
                lifecycle,
                ("create_media_buy",),
                losses=losses,
                warnings=(
                    "The recovered established mutation did not gain atomic feed or pricing "
                    "fencing.",
                ),
            ),
        )

    async def continue_legacy_purchase(
        self,
        continuation: LegacyCatalogContinuation | str,
        request: CoordinatorBuyProductsInput | BaseModel,
        *,
        accepted_losses: Sequence[str] | None = None,
    ) -> CompatiblePurchaseResult:
        """Redeem an established catalog continuation after coordinator restart."""

        lifecycle = self._select_lifecycle("buy_products")
        if lifecycle is not MediaBuyLifecycle.ESTABLISHED:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "legacy_continuation",
                "A legacy catalog continuation can only be redeemed on the established lane.",
            )
        token = (
            continuation.token
            if isinstance(continuation, LegacyCatalogContinuation)
            else continuation
        )
        if not isinstance(token, str) or not token:
            raise ValueError("continuation token must be non-empty")
        payload = _json_object(request)
        assert payload is not None
        payload.pop("feed_version", None)
        payload.pop("pricing_version", None)
        account = _mapping(payload.get("account"))
        if not account:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "account_binding",
                "Established continuation redemption requires an account.",
            )
        if not _optional_text(payload.get("idempotency_key")):
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "idempotency_key",
                "Established continuation redemption requires a non-empty idempotency_key.",
            )
        try:
            selected_ids = _selected_product_ids(payload.get("purchases"), unique=True)
        except ValueError as exc:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "purchase_selection",
                str(exc),
            ) from exc
        legacy = self._require_legacy("continue_legacy_purchase", lifecycle)
        record = await legacy.store.get_continuation(
            _token_hash(token), principal_id=cast(str, self._principal_id)
        )
        if record is None:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "purchase_continuation",
                "No continuation exists in this authenticated principal scope.",
            )
        source_version = self._exact_source_schema_version("continue_legacy_purchase", lifecycle)
        if record.source_adcp_version != source_version:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "served_version_changed",
                "The continuation was issued under a different served AdCP contract; "
                "reconstruct the coordinator with the original seller binding.",
            )
        loss_order = (*_PURCHASE_LOSSES, _MUTATION_LOSS)
        losses = tuple(loss for loss in loss_order if loss in record.losses)
        if set(losses) != set(record.losses):
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "compatibility_losses",
                "The continuation contains an unsupported compatibility loss set.",
            )
        exact_losses = tuple(losses if accepted_losses is None else accepted_losses)
        if set(exact_losses) != set(losses) or len(exact_losses) != len(losses):
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "accepted_losses",
                "accepted_losses must exactly match the established continuation losses.",
                losses=losses,
            )
        refused = tuple(loss for loss in losses if loss not in self._allowed_losses)
        if refused:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "compatibility_losses",
                "No purchase was sent because required compatibility losses were not allowed.",
                losses=refused,
            )
        try:
            legacy_create = _legacy_create_request(payload, self.negotiated_version, source_version)
        except ValueError as exc:
            raise self._unsupported(
                "continue_legacy_purchase",
                lifecycle,
                "legacy_request_projection",
                str(exc),
            ) from exc
        result = await legacy.continue_legacy_purchase(
            {
                "idempotency_key": payload["idempotency_key"],
                "continuation_token": token,
                "account": account,
                "selected_product_ids": selected_ids,
                "accepted_losses": list(exact_losses),
                "legacy_create_request": legacy_create,
            },
            principal_id=cast(str, self._principal_id),
            target_binding=cast(str, self._target_binding),
        )
        return CompatiblePurchaseResult(
            result=result,
            compatibility=self._report(
                lifecycle,
                ("create_media_buy",),
                losses=losses,
                warnings=(
                    "The established mutation cannot atomically fence the listed feed or pricing.",
                ),
            ),
        )

    async def _issue_legacy_continuation(
        self,
        *,
        legacy: LegacyPurchaseCoordinator,
        issuance_idempotency_key: str,
        account: JsonObject,
        source_version: str,
        observed_request: JsonObject,
        observed_response: JsonObject,
        products: Sequence[JsonObject],
    ) -> LegacyCatalogContinuation:
        losses = list(_PURCHASE_LOSSES)
        if not self._mutation_idempotency_guaranteed:
            losses.append(_MUTATION_LOSS)
        expires_at = _aware_utc(self._clock()) + self._continuation_ttl
        token = await legacy.issue_legacy_create_continuation(
            principal_id=cast(str, self._principal_id),
            issuance_idempotency_key=issuance_idempotency_key,
            account=account,
            source_adcp_version=source_version,
            expires_at=expires_at,
            observed_request=observed_request,
            observed_response=observed_response,
            product_ids=[product["product_id"] for product in products],
            buyer_visible_products=list(products),
            losses=losses,
            target_binding=cast(str, self._target_binding),
            mutation_idempotency_guaranteed=self._mutation_idempotency_guaranteed,
            listed_purchase_context={
                "negotiated_version": self.negotiated_version,
                "tools_used": ["get_products"],
            },
        )
        return LegacyCatalogContinuation(
            kind="legacy_create",
            token=token,
            losses=tuple(losses),
            expires_at=expires_at,
        )

    def _select_lifecycle(self, operation: str) -> MediaBuyLifecycle:
        return self._select_catalog_purchase_lifecycle(operation)

    def _select_catalog_purchase_lifecycle(self, operation: str) -> MediaBuyLifecycle:
        compact_contract = _is_compact_release(self.negotiated_version)
        _, release_minor = _release_minor(self.negotiated_version)
        # The 3.0 capabilities contract did not expose buying_modes. Wholesale
        # support is therefore discovered by the call on that exact legacy
        # contract; 3.1+ has an explicit capability and must advertise it.
        wholesale_supported = release_minor == 0 or "wholesale" in self._buying_modes
        compact_pair = compact_contract and {
            "list_products",
            "buy_products",
        }.issubset(self._tools)
        established_pair = (
            not compact_contract
            and {
                "get_products",
                "create_media_buy",
            }.issubset(self._tools)
            and wholesale_supported
        )

        if self._preferred_lifecycle == "compact":
            if compact_pair:
                return MediaBuyLifecycle.COMPACT
            raise self._unsupported(
                operation,
                MediaBuyLifecycle.COMPACT,
                "compact_catalog_purchase_not_advertised",
                "A compact catalog purchase requires both list_products and buy_products.",
            )
        if self._preferred_lifecycle == "established":
            if established_pair:
                return MediaBuyLifecycle.ESTABLISHED
            raise self._unsupported(
                operation,
                MediaBuyLifecycle.ESTABLISHED,
                "established_catalog_purchase_not_advertised",
                "An established catalog purchase requires get_products, create_media_buy, "
                "and the wholesale buying mode.",
            )
        if compact_pair:
            return MediaBuyLifecycle.COMPACT
        if established_pair:
            return MediaBuyLifecycle.ESTABLISHED
        raise self._unsupported(
            operation,
            MediaBuyLifecycle.COMPACT if compact_contract else MediaBuyLifecycle.ESTABLISHED,
            "catalog_purchase_route_not_advertised",
            "The seller advertises no complete compact or established catalog-purchase route.",
        )

    def _legacy_list_request(self, request: ListProductsRequest) -> JsonObject:
        value = request.model_dump(mode="json", exclude_none=True)
        criteria = _mapping(value.get("criteria"))
        unsupported = set(criteria) - {"offer_filters"}
        if unsupported:
            raise self._unsupported(
                "list_products",
                MediaBuyLifecycle.ESTABLISHED,
                "criteria",
                "Established listing cannot preserve criteria fields: "
                + ", ".join(sorted(unsupported)),
            )
        for field in ("governance_context", "context_id"):
            if field in value:
                raise self._unsupported(
                    "list_products",
                    MediaBuyLifecycle.ESTABLISHED,
                    field,
                    f"Established listing cannot preserve {field}.",
                )
        wire: JsonObject = {
            "adcp_version": self.negotiated_version,
            "adcp_major_version": 3,
            "buying_mode": "wholesale",
        }
        for field in ("account", "brand", "context", "push_notification_config"):
            if field in value:
                wire[field] = value[field]
        if "offer_filters" in criteria:
            wire["filters"] = criteria["offer_filters"]
        if "fields" in value:
            wire["fields"] = value["fields"]
        pagination = {key: value[key] for key in ("cursor", "max_results") if key in value}
        if pagination:
            wire["pagination"] = pagination
        if "if_feed_version" in value:
            wire["if_wholesale_feed_version"] = value["if_feed_version"]
        if "if_pricing_version" in value:
            wire["if_pricing_version"] = value["if_pricing_version"]
        return wire

    def _completed_payload(
        self,
        operation: str,
        result: TaskResult[Any],
        lifecycle: MediaBuyLifecycle,
    ) -> JsonObject:
        if not result.success or result.status is not TaskStatus.COMPLETED or result.data is None:
            raise self._unsupported(
                operation,
                lifecycle,
                "completed_catalog_required",
                f"{operation} did not return a completed successful response: "
                f"{result.error or result.message or result.status.value}",
            )
        return cast(
            JsonObject,
            result.data.model_dump(mode="json", exclude_none=True, exclude_unset=True),
        )

    def _validate_served_version(
        self, operation: str, payload: JsonObject, lifecycle: MediaBuyLifecycle
    ) -> None:
        served = payload.get("adcp_version")
        if served is None:
            # The capabilities response is the served-contract binding. Some
            # task response schemas do not repeat the envelope field.
            return
        if not isinstance(served, str) or _parse_release(served) is None:
            raise self._unsupported(
                operation,
                lifecycle,
                "served_version_invalid",
                "The seller returned an invalid response adcp_version.",
            )
        normalized = normalize_to_release_precision(served)
        if normalized != self.negotiated_version:
            raise self._unsupported(
                operation,
                lifecycle,
                "served_version_mismatch",
                f"The seller served AdCP {normalized}, but "
                f"{self.negotiated_version} was negotiated.",
            )

    def _exact_source_schema_version(self, operation: str, lifecycle: MediaBuyLifecycle) -> str:
        exact = get_bundle_adcp_version(version=self.negotiated_version)
        if exact is None:
            raise self._unsupported(
                operation,
                lifecycle,
                "served_version_schema_unavailable",
                f"No bundled schema can validate AdCP {self.negotiated_version}.",
            )
        return exact

    def _validate_listing(self, listing: CompatibleCatalog) -> None:
        if not isinstance(listing, CompatibleCatalog) or listing._coordinator_identity != id(self):
            raise TypeError("listing must be a CompatibleCatalog issued by this coordinator")

    def _require_legacy(
        self, operation: str, lifecycle: MediaBuyLifecycle
    ) -> LegacyPurchaseCoordinator:
        if self._legacy is None:
            raise self._unsupported(
                operation,
                lifecycle,
                "legacy_purchase_coordinator",
                "Established purchase coordination requires a LegacyPurchaseCoordinator.",
            )
        return self._legacy

    def _report(
        self,
        lifecycle: MediaBuyLifecycle,
        tools: Sequence[str],
        *,
        losses: Sequence[str] = (),
        warnings: Sequence[str] = (),
    ) -> MediaBuyCompatibilityReport:
        compatibility = (
            MediaBuyCompatibility.NATIVE
            if lifecycle is MediaBuyLifecycle.COMPACT
            else (
                MediaBuyCompatibility.LOSSY_PROJECTION
                if losses
                else MediaBuyCompatibility.LOSSLESS_PROJECTION
            )
        )
        return MediaBuyCompatibilityReport(
            negotiated_version=self.negotiated_version,
            lifecycle=lifecycle,
            tools_used=tuple(tools),
            compatibility=compatibility,
            warnings=tuple(warnings),
            losses=tuple(losses),
        )

    def _unsupported(
        self,
        operation: str,
        lifecycle: MediaBuyLifecycle,
        feature: str,
        message: str,
        *,
        losses: Sequence[str] = (),
        code: CompatibilityErrorCode = "UNSUPPORTED_FEATURE",
    ) -> MediaBuyLifecycleCompatibilityError:
        return MediaBuyLifecycleCompatibilityError(
            operation=operation,
            negotiated_version=self.negotiated_version,
            lifecycle=lifecycle,
            feature=feature,
            message=message,
            losses=losses,
            code=code,
        )

Negotiate one catalog-to-purchase workflow across AdCP 3.x.

Static methods

async def negotiate(client: ADCPClient,
*,
capabilities: GetAdcpCapabilitiesResponse | None = None,
advertised_tools: Sequence[str] | None = None,
legacy_purchase_coordinator: LegacyPurchaseCoordinator | None = None,
principal_id: str | None = None,
target_binding: str | None = None,
preferred_lifecycle: LifecyclePreference = 'auto',
allowed_losses: Sequence[str] = (),
continuation_ttl: timedelta = datetime.timedelta(seconds=300),
clock: Callable[[], datetime] | None = None)

Fetch seller evidence and construct a stable negotiated coordinator.

Methods

async def buy_products(self,
listing: CompatibleCatalog,
request: CoordinatorBuyProductsInput | BaseModel,
*,
accepted_losses: Sequence[str] | None = None) ‑> CompatiblePurchaseResult

Expand source code

async def buy_products(
    self,
    listing: CompatibleCatalog,
    request: CoordinatorBuyProductsInput | BaseModel,
    *,
    accepted_losses: Sequence[str] | None = None,
) -> CompatiblePurchaseResult:
    """Purchase selected products without changing either native signature."""

    self._validate_listing(listing)
    lifecycle = self._select_lifecycle("buy_products")
    if lifecycle is not listing.compatibility.lifecycle:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "lifecycle_changed",
            "The catalog route no longer matches the purchase route; restart discovery.",
        )
    payload = _json_object(request)
    assert payload is not None
    # Feed and pricing versions are seller-issued listing evidence. Never
    # accept caller substitutions, including when the listing omitted an
    # independent pricing version.
    payload.pop("feed_version", None)
    payload.pop("pricing_version", None)
    account = _mapping(payload.get("account"))
    if listing._account is not None and (
        not account
        or canonical_account_identity(account) != canonical_account_identity(listing._account)
    ):
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "account_binding",
            "The purchase account does not match the account used for listing.",
        )
    purchases = payload.get("purchases")
    try:
        selected_ids = _selected_product_ids(
            purchases, unique=lifecycle is MediaBuyLifecycle.ESTABLISHED
        )
    except ValueError as exc:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "purchase_selection",
            str(exc),
        ) from exc
    listed_ids = {product["product_id"] for product in listing.products}
    if not set(selected_ids).issubset(listed_ids):
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "selection_binding",
            "The purchase selects a product that was not present in this catalog result.",
        )

    if lifecycle is MediaBuyLifecycle.COMPACT:
        if listing.feed_version is None:
            raise self._unsupported(
                "buy_products",
                lifecycle,
                "feed_version",
                "A compact purchase requires the seller's real listing feed_version.",
            )
        compact_payload = {
            **payload,
            "feed_version": listing.feed_version,
            "adcp_version": self.negotiated_version,
        }
        if listing.pricing_version is not None:
            compact_payload["pricing_version"] = listing.pricing_version
        model = BuyProductsRequest.model_validate(compact_payload)
        result = await self.client.buy_products(model)
        return CompatiblePurchaseResult(
            result=result,
            compatibility=self._report(lifecycle, ("buy_products",)),
        )

    legacy = self._require_legacy("buy_products", lifecycle)
    if not _optional_text(payload.get("idempotency_key")):
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "idempotency_key",
            "Established purchase requires a non-empty idempotency_key.",
        )
    continuation = listing.purchase_continuation
    if continuation is None:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "purchase_continuation",
            "The established listing has no SDK purchase continuation.",
        )
    exact_losses = tuple(continuation.losses if accepted_losses is None else accepted_losses)
    if set(exact_losses) != set(continuation.losses) or len(exact_losses) != len(
        continuation.losses
    ):
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "accepted_losses",
            "accepted_losses must exactly match the listing continuation losses.",
            losses=continuation.losses,
        )
    refused = tuple(loss for loss in continuation.losses if loss not in self._allowed_losses)
    if refused:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "compatibility_losses",
            "No purchase was sent because required compatibility losses were not allowed.",
            losses=refused,
        )
    source_version = listing._source_schema_version
    if source_version is None:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "served_version_schema_unavailable",
            "The established listing did not retain its exact source schema version.",
        )
    try:
        legacy_create = _legacy_create_request(payload, self.negotiated_version, source_version)
    except ValueError as exc:
        raise self._unsupported(
            "buy_products",
            lifecycle,
            "legacy_request_projection",
            str(exc),
        ) from exc
    legacy_result = await legacy.continue_legacy_purchase(
        {
            "idempotency_key": payload.get("idempotency_key"),
            "continuation_token": continuation.token,
            "account": account,
            "selected_product_ids": selected_ids,
            "accepted_losses": list(exact_losses),
            "legacy_create_request": legacy_create,
        },
        principal_id=cast(str, self._principal_id),
        target_binding=cast(str, self._target_binding),
    )
    return CompatiblePurchaseResult(
        result=legacy_result,
        compatibility=self._report(
            lifecycle,
            ("create_media_buy",),
            losses=continuation.losses,
            warnings=(
                "The established mutation cannot atomically fence the listed feed or pricing.",
            ),
        ),
    )

Purchase selected products without changing either native signature.

async def continue_legacy_purchase(self,
continuation: LegacyCatalogContinuation | str,
request: CoordinatorBuyProductsInput | BaseModel,
*,
accepted_losses: Sequence[str] | None = None) ‑> CompatiblePurchaseResult

Expand source code

async def continue_legacy_purchase(
    self,
    continuation: LegacyCatalogContinuation | str,
    request: CoordinatorBuyProductsInput | BaseModel,
    *,
    accepted_losses: Sequence[str] | None = None,
) -> CompatiblePurchaseResult:
    """Redeem an established catalog continuation after coordinator restart."""

    lifecycle = self._select_lifecycle("buy_products")
    if lifecycle is not MediaBuyLifecycle.ESTABLISHED:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "legacy_continuation",
            "A legacy catalog continuation can only be redeemed on the established lane.",
        )
    token = (
        continuation.token
        if isinstance(continuation, LegacyCatalogContinuation)
        else continuation
    )
    if not isinstance(token, str) or not token:
        raise ValueError("continuation token must be non-empty")
    payload = _json_object(request)
    assert payload is not None
    payload.pop("feed_version", None)
    payload.pop("pricing_version", None)
    account = _mapping(payload.get("account"))
    if not account:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "account_binding",
            "Established continuation redemption requires an account.",
        )
    if not _optional_text(payload.get("idempotency_key")):
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "idempotency_key",
            "Established continuation redemption requires a non-empty idempotency_key.",
        )
    try:
        selected_ids = _selected_product_ids(payload.get("purchases"), unique=True)
    except ValueError as exc:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "purchase_selection",
            str(exc),
        ) from exc
    legacy = self._require_legacy("continue_legacy_purchase", lifecycle)
    record = await legacy.store.get_continuation(
        _token_hash(token), principal_id=cast(str, self._principal_id)
    )
    if record is None:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "purchase_continuation",
            "No continuation exists in this authenticated principal scope.",
        )
    source_version = self._exact_source_schema_version("continue_legacy_purchase", lifecycle)
    if record.source_adcp_version != source_version:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "served_version_changed",
            "The continuation was issued under a different served AdCP contract; "
            "reconstruct the coordinator with the original seller binding.",
        )
    loss_order = (*_PURCHASE_LOSSES, _MUTATION_LOSS)
    losses = tuple(loss for loss in loss_order if loss in record.losses)
    if set(losses) != set(record.losses):
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "compatibility_losses",
            "The continuation contains an unsupported compatibility loss set.",
        )
    exact_losses = tuple(losses if accepted_losses is None else accepted_losses)
    if set(exact_losses) != set(losses) or len(exact_losses) != len(losses):
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "accepted_losses",
            "accepted_losses must exactly match the established continuation losses.",
            losses=losses,
        )
    refused = tuple(loss for loss in losses if loss not in self._allowed_losses)
    if refused:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "compatibility_losses",
            "No purchase was sent because required compatibility losses were not allowed.",
            losses=refused,
        )
    try:
        legacy_create = _legacy_create_request(payload, self.negotiated_version, source_version)
    except ValueError as exc:
        raise self._unsupported(
            "continue_legacy_purchase",
            lifecycle,
            "legacy_request_projection",
            str(exc),
        ) from exc
    result = await legacy.continue_legacy_purchase(
        {
            "idempotency_key": payload["idempotency_key"],
            "continuation_token": token,
            "account": account,
            "selected_product_ids": selected_ids,
            "accepted_losses": list(exact_losses),
            "legacy_create_request": legacy_create,
        },
        principal_id=cast(str, self._principal_id),
        target_binding=cast(str, self._target_binding),
    )
    return CompatiblePurchaseResult(
        result=result,
        compatibility=self._report(
            lifecycle,
            ("create_media_buy",),
            losses=losses,
            warnings=(
                "The established mutation cannot atomically fence the listed feed or pricing.",
            ),
        ),
    )

Redeem an established catalog continuation after coordinator restart.

async def list_products(self,
request: ListProductsRequest,
*,
issuance_idempotency_key: str | None = None) ‑> CompatibleCatalog

Expand source code

async def list_products(
    self,
    request: ListProductsRequest,
    *,
    issuance_idempotency_key: str | None = None,
) -> CompatibleCatalog:
    """List a catalog through the negotiated lifecycle.

    Established results issue a short-lived SDK continuation.  They never
    synthesize ``feed_version`` or ``pricing_version`` values.
    """

    request_payload = request.model_dump(mode="json", exclude_none=True)
    conditional = tuple(
        field for field in ("if_feed_version", "if_pricing_version") if field in request_payload
    )
    if conditional:
        raise self._unsupported(
            "list_products",
            self.lifecycle,
            "conditional_catalog_reuse",
            "The coordinator cannot safely bind a purchase to an unchanged response; "
            "omit " + ", ".join(conditional) + " and request a complete catalog page.",
        )
    lifecycle = self._select_lifecycle("list_products")
    account = _json_object(request.account)
    if lifecycle is MediaBuyLifecycle.COMPACT:
        pinned = request.model_copy(update={"adcp_version": self.negotiated_version})
        result = await self.client.list_products(pinned)
        payload = self._completed_payload("list_products", result, lifecycle)
        self._validate_served_version("list_products", payload, lifecycle)
        products = _products(payload, operation="list_products", coordinator=self)
        feed_version = _optional_text(payload.get("feed_version"))
        if feed_version is None:
            raise self._unsupported(
                "list_products",
                lifecycle,
                "feed_version",
                "The compact seller did not return the required real feed_version.",
            )
        return CompatibleCatalog(
            products=products,
            feed_version=feed_version,
            pricing_version=_optional_text(payload.get("pricing_version")),
            next_cursor=_optional_text(payload.get("next_cursor")),
            cache_scope=_optional_text(payload.get("cache_scope")),
            compatibility=self._report(lifecycle, ("list_products",)),
            raw=result,
            _account=account,
            _coordinator_identity=id(self),
            _source_schema_version=self.negotiated_version,
        )

    legacy = self._require_legacy("list_products", lifecycle)
    if account is None:
        raise self._unsupported(
            "list_products",
            lifecycle,
            "account",
            "Established catalog purchase coordination requires an account-bound listing.",
        )
    issue_key = issuance_idempotency_key or request.idempotency_key
    if not issue_key:
        raise self._unsupported(
            "list_products",
            lifecycle,
            "issuance_idempotency_key",
            "Established listing requires a stable issuance_idempotency_key.",
        )
    wire_request = self._legacy_list_request(request)
    source_version = self._exact_source_schema_version("list_products", lifecycle)
    try:
        _assert_declared_legacy_fields(
            wire_request,
            tool="get_products",
            source_version=source_version,
            nested=("filters",),
        )
    except ValueError as exc:
        raise self._unsupported(
            "list_products",
            lifecycle,
            "legacy_request_projection",
            str(exc),
        ) from exc
    validation = validate_request("get_products", wire_request, version=source_version)
    if not validation.valid or validation.variant == "skipped":
        raise self._unsupported(
            "list_products",
            lifecycle,
            "legacy_request_projection",
            "The listing requirements cannot be represented by the negotiated "
            f"{self.negotiated_version} get_products contract.",
        )
    model = LegacyGetProductsRequest.model_validate(wire_request)
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", DeprecationWarning)
        legacy_result = await self.client.get_products_legacy(model)
    payload = self._completed_payload("list_products", legacy_result, lifecycle)
    self._validate_served_version("list_products", payload, lifecycle)
    response_validation = validate_response("get_products", payload, version=source_version)
    if not response_validation.valid or response_validation.variant == "skipped":
        raise self._unsupported(
            "list_products",
            lifecycle,
            "response_projection",
            "Established get_products returned an invalid exact-version response: "
            f"{format_issues(response_validation.issues)}",
        )
    products = _products(payload, operation="list_products", coordinator=self)
    continuation = (
        await self._issue_legacy_continuation(
            legacy=legacy,
            issuance_idempotency_key=issue_key,
            account=account,
            source_version=source_version,
            observed_request=wire_request,
            observed_response=payload,
            products=products,
        )
        if products
        else None
    )
    return CompatibleCatalog(
        products=products,
        feed_version=_optional_text(payload.get("wholesale_feed_version")),
        pricing_version=_optional_text(payload.get("pricing_version")),
        next_cursor=_optional_text(_mapping(payload.get("pagination")).get("cursor")),
        cache_scope=_optional_text(payload.get("cache_scope")),
        purchase_continuation=continuation,
        compatibility=self._report(lifecycle, ("get_products",)),
        raw=legacy_result,
        _account=account,
        _coordinator_identity=id(self),
        _source_schema_version=source_version,
    )

List a catalog through the negotiated lifecycle.

Established results issue a short-lived SDK continuation. They never synthesize feed_version or pricing_version values.

async def recover_legacy_purchase(self,
operation: CompatibilityPurchaseOperation) ‑> CompatiblePurchaseResult

Expand source code

async def recover_legacy_purchase(
    self, operation: CompatibilityPurchaseOperation
) -> CompatiblePurchaseResult:
    """Resume a fenced established purchase through the configured reconciler."""

    lifecycle = MediaBuyLifecycle.ESTABLISHED
    legacy = self._require_legacy("recover_legacy_purchase", lifecycle)
    record = await legacy.store.get_continuation(
        operation.token_hash,
        principal_id=cast(str, self._principal_id),
    )
    losses = (
        tuple(loss for loss in (*_PURCHASE_LOSSES, _MUTATION_LOSS) if loss in record.losses)
        if record is not None
        else ()
    )
    result = await legacy.recover_legacy_purchase(
        operation,
        principal_id=cast(str, self._principal_id),
        target_binding=cast(str, self._target_binding),
    )
    return CompatiblePurchaseResult(
        result=result,
        compatibility=self._report(
            lifecycle,
            ("create_media_buy",),
            losses=losses,
            warnings=(
                "The recovered established mutation did not gain atomic feed or pricing "
                "fencing.",
            ),
        ),
    )

Resume a fenced established purchase through the configured reconciler.

def report(self, operation: str | None = None) ‑> MediaBuyCompatibilityReport

Expand source code

def report(self, operation: str | None = None) -> MediaBuyCompatibilityReport:
    """Describe a viable operation route without claiming a dispatch.

    With no operation this reports only the negotiated protocol family.
    Passing ``list_products`` or ``buy_products`` performs the same route
    preflight as that operation while leaving ``tools_used`` empty.
    """

    lifecycle = self.lifecycle
    if operation is not None:
        if operation not in _COMPACT_TO_ESTABLISHED:
            raise ValueError("operation must be list_products or buy_products")
        lifecycle = self._select_lifecycle(operation)
    return self._report(lifecycle, ())

Describe a viable operation route without claiming a dispatch.

With no operation this reports only the negotiated protocol family. Passing list_products or buy_products performs the same route preflight as that operation while leaving tools_used empty.

class PendingTaskResolution (task_id: str, result: LegacyPurchaseResult)

Expand source code

@dataclass(frozen=True)
class PendingTaskResolution:
    """Task-bound result returned by a read-only pending-task poller."""

    task_id: str
    result: LegacyPurchaseResult

Task-bound result returned by a read-only pending-task poller.

Instance variables

var result : collections.abc.Mapping[str, typing.Any] | pydantic.main.BaseModel | TaskResult[Any]

var task_id : str

class ReconciliationResult (status: ReconciliationStatus,
result: LegacyPurchaseResult | None = None)

Expand source code

@dataclass(frozen=True)
class ReconciliationResult:
    """Authoritative result of reconciling an interrupted seller mutation."""

    status: ReconciliationStatus
    result: LegacyPurchaseResult | None = None

    @classmethod
    def applied(cls, result: LegacyPurchaseResult) -> ReconciliationResult:
        return cls(ReconciliationStatus.APPLIED, copy.deepcopy(result))

    @classmethod
    def not_applied(cls) -> ReconciliationResult:
        return cls(ReconciliationStatus.NOT_APPLIED)

    @classmethod
    def ambiguous(cls) -> ReconciliationResult:
        return cls(ReconciliationStatus.AMBIGUOUS)

Authoritative result of reconciling an interrupted seller mutation.

Static methods

def ambiguous() ‑> ReconciliationResult

def applied(result: LegacyPurchaseResult) ‑> ReconciliationResult

def not_applied() ‑> ReconciliationResult

Instance variables

var result : collections.abc.Mapping[str, typing.Any] | pydantic.main.BaseModel | TaskResult[Any] | None

var status : ReconciliationStatus

class ReconciliationStatus (*args, **kwds)

Expand source code

class ReconciliationStatus(str, Enum):
    APPLIED = "authoritatively_applied"
    NOT_APPLIED = "authoritatively_not_applied"
    AMBIGUOUS = "ambiguous"

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

Ancestors

• builtins.str
• enum.Enum

Class variables

var AMBIGUOUS

var APPLIED

var NOT_APPLIED

class SqliteCompatibilityContinuationStore (path: str | Path,
*,
timeout: float = 30.0,
clock: Callable[[], datetime] | None = None,
max_records: int = 20000,
max_bytes: int = 67108864,
max_payload_bytes: int = 1048576,
max_records_per_principal: int = 2000,
max_bytes_per_principal: int = 8388608)

Expand source code

class SqliteCompatibilityContinuationStore:
    """Durable local continuation ledger backed by a SQLite file."""

    is_durable: ClassVar[bool] = True

    def __init__(
        self,
        path: str | Path,
        *,
        timeout: float = 30.0,
        clock: Callable[[], datetime] | None = None,
        max_records: int = 20_000,
        max_bytes: int = 64 * 1024 * 1024,
        max_payload_bytes: int = 1024 * 1024,
        max_records_per_principal: int = 2_000,
        max_bytes_per_principal: int = 8 * 1024 * 1024,
    ) -> None:
        raw = str(path)
        if raw == ":memory:" or raw.startswith("file::memory:"):
            raise ValueError("SqliteCompatibilityContinuationStore requires a file-backed database")
        # 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))
        for name, value in (
            ("max_records", max_records),
            ("max_bytes", max_bytes),
            ("max_payload_bytes", max_payload_bytes),
            ("max_records_per_principal", max_records_per_principal),
            ("max_bytes_per_principal", max_bytes_per_principal),
        ):
            if type(value) is not int or value <= 0:
                raise ValueError(f"{name} must be a positive integer")
        self.max_records = max_records
        self.max_bytes = max_bytes
        self.max_payload_bytes = max_payload_bytes
        self.max_records_per_principal = min(max_records_per_principal, max_records)
        self.max_bytes_per_principal = min(max_bytes_per_principal, max_bytes)
        self._ensure_private_database_file()
        with closing(self._connect()) as conn, conn:
            conn.executescript(_SCHEMA)
            self._ensure_timestamp_columns(conn)

    def _ensure_timestamp_columns(self, conn: sqlite3.Connection) -> None:
        """Migrate ledgers created by pre-release coordinator builds."""

        # Serialize the inspect/alter/backfill sequence across processes. Without
        # the write lock, two starters can both observe a missing column and the
        # second ALTER then fails with ``duplicate column name``.
        conn.execute("BEGIN IMMEDIATE")
        now = _format_datetime(self._clock())
        for table in ("adcp_compat_continuations", "adcp_compat_operations"):
            columns = {row["name"] for row in conn.execute(f"PRAGMA table_info({table})")}
            for column in ("created_at", "updated_at"):
                if column not in columns:
                    conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} TEXT")
                conn.execute(
                    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 "issuance_fingerprint" not in continuation_columns:
            conn.execute(
                "ALTER TABLE adcp_compat_continuations ADD COLUMN issuance_fingerprint TEXT"
            )
        if "issuance_binding_hash" not in continuation_columns:
            conn.execute(
                "ALTER TABLE adcp_compat_continuations ADD COLUMN issuance_binding_hash TEXT"
            )
        conn.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS adcp_compat_continuations_issuance_idx "
            "ON adcp_compat_continuations (principal_id, issuance_fingerprint) "
            "WHERE issuance_fingerprint IS NOT NULL"
        )
        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)
        operation_columns = {
            row["name"] for row in conn.execute("PRAGMA table_info(adcp_compat_operations)")
        }
        if "reserved_result_bytes" not in operation_columns:
            conn.execute(
                "ALTER TABLE adcp_compat_operations "
                "ADD COLUMN reserved_result_bytes INTEGER NOT NULL DEFAULT 0 "
                "CHECK (reserved_result_bytes >= 0)"
            )
        # Pre-reservation ledgers may already contain a seller mutation that
        # requires a terminal write. Give each unresolved row a durable budget
        # once, preserving any larger pending result already on disk.
        conn.execute(
            "UPDATE adcp_compat_operations "
            "SET reserved_result_bytes = MAX(?, "
            "length(CAST(COALESCE(result_json, '') AS BLOB))) "
            "WHERE state IN ('in_flight', 'pending', 'ambiguous') "
            "AND reserved_result_bytes = 0",
            (self.max_payload_bytes,),
        )
        self._audit_legacy_payloads(conn)
        # The table, index, and guards share this migration's write transaction.
        # Otherwise an older process could delete a continuation after the table
        # became visible but before the delete guard committed.
        self._migrate_replay_fence_schema(conn)
        for statement in _REPLAY_FENCE_SCHEMA:
            conn.execute(statement)

    @staticmethod
    def _migrate_replay_fence_schema(conn: sqlite3.Connection) -> None:
        """Remove the pre-release write-only equivalence column atomically."""

        columns = {
            row["name"]
            for row in conn.execute("PRAGMA table_info(adcp_compat_issuance_tombstones)")
        }
        if "legacy_equivalence_hash" not in columns:
            return

        for trigger_name in _REPLAY_FENCE_TRIGGER_NAMES:
            conn.execute(f'DROP TRIGGER IF EXISTS "{trigger_name}"')
        conn.execute(
            "ALTER TABLE adcp_compat_issuance_tombstones "
            "RENAME TO adcp_compat_issuance_tombstones_legacy"
        )
        conn.execute(_REPLAY_FENCE_SCHEMA[0])
        conn.execute(
            """
            INSERT INTO adcp_compat_issuance_tombstones (
                token_hash, principal_id, issuance_fingerprint,
                issuance_binding_hash, retired_at
            )
            SELECT
                token_hash, principal_id, issuance_fingerprint,
                issuance_binding_hash, retired_at
            FROM adcp_compat_issuance_tombstones_legacy
            """
        )
        conn.execute("DROP TABLE adcp_compat_issuance_tombstones_legacy")

    @staticmethod
    def _audit_legacy_payloads(conn: sqlite3.Connection) -> None:
        audited = conn.execute(
            "SELECT value FROM adcp_compat_metadata WHERE key = 'persistence_policy_version'"
        ).fetchone()
        if audited is not None and audited["value"] == "1":
            return
        for row in conn.execute("SELECT * FROM adcp_compat_continuations"):
            _validate_continuation_persistence(_decode_continuation(row))
        for row in conn.execute("SELECT * FROM adcp_compat_operations"):
            operation = _decode_operation(row)
            _validate_persistable_payload(operation.execution_input, context="execution input")
            if operation.result is not None:
                _validate_persistable_payload(operation.result, context="legacy result")
        conn.execute(
            "INSERT OR REPLACE INTO adcp_compat_metadata (key, value) VALUES (?, ?)",
            ("persistence_policy_version", "1"),
        )

    @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, exist_ok=True)
                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."""

        self._ensure_private_file(self.path, "continuation database")

    def _ensure_private_sidecar_files(self) -> None:
        """Pre-create SQLite WAL files privately before SQLite can open them.

        SQLite normally inherits the database mode for ``-wal`` and ``-shm``
        files.  Creating and validating them explicitly makes that guarantee
        independent of the process umask and SQLite build behavior.
        """

        for suffix in ("-wal", "-shm"):
            self._ensure_private_file(Path(f"{self.path}{suffix}"), "SQLite sidecar")

    @staticmethod
    def _ensure_private_file(path: Path, description: str) -> None:
        """Atomically create *path* as 0600 or validate the existing file."""

        create_flags = os.O_CREAT | os.O_EXCL | os.O_RDWR
        existing_flags = os.O_RDWR
        if hasattr(os, "O_NOFOLLOW"):
            create_flags |= os.O_NOFOLLOW
            existing_flags |= os.O_NOFOLLOW
        while True:
            try:
                descriptor = os.open(path, create_flags, 0o600)
                break
            except FileExistsError:
                try:
                    descriptor = os.open(path, existing_flags)
                    break
                except FileNotFoundError:
                    # SQLite removes sidecars when the last WAL connection
                    # closes.  If that happens between the existence check and
                    # this open, retry the atomic create path.
                    continue
                except OSError as exc:
                    raise PermissionError(
                        f"{description} {path} is not a safe regular file"
                    ) from exc
            except OSError as exc:
                raise PermissionError(f"{description} {path} is not a safe regular file") from exc
        try:
            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(
                    f"{description} {path} has mode {mode:#o}; restrict it to 0o600 before opening"
                )
        finally:
            os.close(descriptor)

    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)
        try:
            conn.row_factory = sqlite3.Row
            conn.execute("PRAGMA foreign_keys = ON")
            deadline = time.monotonic() + self.timeout
            while True:
                try:
                    conn.execute("PRAGMA journal_mode = WAL")
                    break
                except sqlite3.OperationalError as exc:
                    if "locked" not in str(exc).lower() or time.monotonic() >= deadline:
                        raise
                    time.sleep(min(0.01, max(0.0, deadline - time.monotonic())))
            conn.execute("PRAGMA synchronous = FULL")
            return conn
        except BaseException:
            conn.close()
            raise

    async def put_continuation(self, continuation: LegacyPurchaseContinuation) -> None:
        await asyncio.to_thread(self._put_continuation, copy.deepcopy(continuation))

    def _put_continuation(self, value: LegacyPurchaseContinuation) -> None:
        _validate_continuation_persistence(value)
        now = _format_datetime(self._clock())
        if value.projected_products is None:
            raise ValueError("new continuations require buyer-visible product bindings")
        payloads = (
            _dumps(value.observed_request),
            _dumps(value.observed_response),
            _dumps(list(value.product_ids)),
            _dumps(list(value.projected_products)),
            _dumps(sorted(value.losses)),
            (
                _dumps(value.listed_purchase_context)
                if value.listed_purchase_context is not None
                else None
            ),
        )
        self._enforce_payload_quota(*(payload for payload in payloads if payload is not None))
        try:
            with closing(self._connect()) as conn, conn:
                conn.execute("BEGIN IMMEDIATE")
                retired = conn.execute(
                    "SELECT 1 FROM adcp_compat_issuance_tombstones "
                    "WHERE token_hash = ? OR "
                    "(principal_id = ? AND issuance_fingerprint = ?)",
                    (value.token_hash, value.principal_id, value.issuance_fingerprint),
                ).fetchone()
                if retired is not None:
                    raise _error(
                        CompatibilityContinuationErrorCode.STORE_CONFLICT,
                        "continuation issuance identity was already retired",
                        "Start a genuinely new discovery with a new issuance identity.",
                    )
                legacy = conn.execute(
                    """
                    SELECT 1 FROM adcp_compat_continuations
                    WHERE issuance_fingerprint IS NULL
                      AND principal_id = ?
                      AND observed_payload_hash = ?
                      AND account_identity = ?
                      AND target_binding = ?
                    LIMIT 1
                    """,
                    (
                        value.principal_id,
                        value.observed_payload_hash,
                        value.account_identity,
                        value.target_binding,
                    ),
                ).fetchone()
                if legacy is not None:
                    raise _error(
                        CompatibilityContinuationErrorCode.STORE_CONFLICT,
                        "equivalent pre-migration authorization requires operator resolution",
                        "Resolve or quarantine the legacy continuation before reissuing.",
                    )
                conn.execute(
                    """
                    INSERT INTO adcp_compat_continuations (
                        token_hash, issuance_fingerprint, issuance_binding_hash,
                        principal_id, account_identity,
                        source_adcp_version, expires_at, observed_request_json,
                        observed_response_json, observed_payload_hash,
                        product_ids_json, projected_products_json, losses_json,
                        mutation_idempotency_guaranteed, target_binding,
                        listed_purchase_context_json, created_at, updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        value.token_hash,
                        value.issuance_fingerprint,
                        value.issuance_binding_hash,
                        value.principal_id,
                        value.account_identity,
                        value.source_adcp_version,
                        _format_datetime(value.expires_at),
                        payloads[0],
                        payloads[1],
                        value.observed_payload_hash,
                        payloads[2],
                        payloads[3],
                        payloads[4],
                        int(value.mutation_idempotency_guaranteed),
                        value.target_binding,
                        payloads[5],
                        now,
                        now,
                    ),
                )
                self._enforce_ledger_quota(conn, principal_id=value.principal_id)
        except sqlite3.IntegrityError as exc:
            with closing(self._connect()) as conn:
                row = conn.execute(
                    "SELECT * FROM adcp_compat_continuations "
                    "WHERE token_hash = ? OR (principal_id = ? AND issuance_fingerprint = ?)",
                    (value.token_hash, value.principal_id, value.issuance_fingerprint),
                ).fetchone()
            if row is not None and _decode_continuation(row) == value:
                return
            raise _error(
                CompatibilityContinuationErrorCode.STORE_CONFLICT,
                "continuation issuance fingerprint is already registered differently",
                "Use the same token derivation key and exact issuance inputs.",
            ) from exc

    async def get_continuation(
        self, token_hash: str, *, principal_id: str
    ) -> LegacyPurchaseContinuation | None:
        return await asyncio.to_thread(self._get_continuation, token_hash, principal_id)

    def _get_continuation(
        self, token_hash: str, principal_id: str
    ) -> LegacyPurchaseContinuation | None:
        with closing(self._connect()) as conn, conn:
            row = conn.execute(
                """
                SELECT * FROM adcp_compat_continuations
                WHERE token_hash = ? AND principal_id = ?
                """,
                (token_hash, principal_id),
            ).fetchone()
        return _decode_continuation(row) if row is not None else None

    async def claim(
        self,
        token_hash: str,
        *,
        principal_id: str,
        idempotency_key: str,
        payload_hash: str,
        execution_input: Mapping[str, Any],
        now: datetime,
    ) -> CompatibilityPurchaseOperation:
        snapshot = copy.deepcopy(dict(execution_input))
        _validate_persistable_payload(snapshot, context="execution input")
        return await asyncio.to_thread(
            self._claim,
            token_hash,
            principal_id,
            idempotency_key,
            payload_hash,
            snapshot,
            now,
        )

    def _claim(
        self,
        token_hash: str,
        principal_id: str,
        idempotency_key: str,
        payload_hash: str,
        execution_input: dict[str, Any],
        now: datetime,
    ) -> CompatibilityPurchaseOperation:
        execution_input_json = _dumps(execution_input)
        with closing(self._connect()) as conn, conn:
            conn.execute("BEGIN IMMEDIATE")
            claim_time = max(_as_utc(now), _as_utc(self._clock()))
            existing = conn.execute(
                """
                SELECT * FROM adcp_compat_operations
                WHERE principal_id = ? AND idempotency_key = ?
                """,
                (principal_id, idempotency_key),
            ).fetchone()
            if existing is not None:
                operation = _decode_operation(existing)
                if operation.token_hash != token_hash or operation.payload_hash != payload_hash:
                    raise _error(
                        CompatibilityContinuationErrorCode.IDEMPOTENCY_CONFLICT,
                        "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.
                    self._enforce_payload_quota(execution_input_json)
                    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 = '{}'",
                        (
                            execution_input_json,
                            updated_at,
                            operation.operation_id,
                            operation.revision,
                        ),
                    )
                    if adopted.rowcount != 1:
                        raise _state_error("legacy execution input changed concurrently")
                    # This bounded, one-time migration write may be required to
                    # reconcile a seller mutation that already executed. Global
                    # fullness must not prevent recovery of that existing row.
                    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),
                        reserved_result_bytes=operation.reserved_result_bytes,
                        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

            continuation = conn.execute(
                """
                SELECT * FROM adcp_compat_continuations
                WHERE token_hash = ? AND principal_id = ?
                """,
                (token_hash, principal_id),
            ).fetchone()
            if continuation is None:
                raise _not_found()
            expires_at = _parse_datetime(continuation["expires_at"])
            if claim_time >= expires_at:
                raise _error(
                    CompatibilityContinuationErrorCode.EXPIRED,
                    "continuation expired before it could be claimed",
                    "Repeat product discovery and obtain a new continuation.",
                )
            if continuation["claimed_operation_id"] is not None:
                raise _error(
                    CompatibilityContinuationErrorCode.ALREADY_CLAIMED,
                    "continuation was already claimed by another operation",
                    "Replay the original idempotency key, or restart product discovery.",
                )

            self._enforce_payload_quota(execution_input_json)
            operation_id = secrets.token_urlsafe(24)
            updated = conn.execute(
                """
                UPDATE adcp_compat_continuations
                SET claimed_operation_id = ?, updated_at = ?
                WHERE token_hash = ? AND principal_id = ?
                  AND claimed_operation_id IS NULL
                """,
                (operation_id, _format_datetime(claim_time), token_hash, principal_id),
            )
            if updated.rowcount != 1:
                raise _error(
                    CompatibilityContinuationErrorCode.ALREADY_CLAIMED,
                    "continuation was concurrently claimed by another operation",
                    "Replay the original idempotency key, or restart product discovery.",
                )
            conn.execute(
                """
                INSERT INTO adcp_compat_operations (
                    operation_id, principal_id, idempotency_key, token_hash,
                    payload_hash, state, revision, execution_input_json,
                    created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    operation_id,
                    principal_id,
                    idempotency_key,
                    token_hash,
                    payload_hash,
                    CompatibilityOperationState.CLAIMED.value,
                    1,
                    execution_input_json,
                    _format_datetime(claim_time),
                    _format_datetime(claim_time),
                ),
            )
            self._enforce_ledger_quota(conn, principal_id=principal_id)
            conn.commit()
            return CompatibilityPurchaseOperation(
                operation_id=operation_id,
                principal_id=principal_id,
                idempotency_key=idempotency_key,
                token_hash=token_hash,
                payload_hash=payload_hash,
                state=CompatibilityOperationState.CLAIMED,
                revision=1,
                execution_input=copy.deepcopy(execution_input),
                reserved_result_bytes=0,
            )

    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 get_operation_by_idempotency_key(
        self, idempotency_key: str, *, principal_id: str
    ) -> CompatibilityPurchaseOperation | None:
        return await asyncio.to_thread(
            self._get_operation_by_idempotency_key, idempotency_key, principal_id
        )

    def _get_operation_by_idempotency_key(
        self, idempotency_key: str, principal_id: str
    ) -> CompatibilityPurchaseOperation | None:
        with closing(self._connect()) as conn:
            row = conn.execute(
                "SELECT * FROM adcp_compat_operations "
                "WHERE idempotency_key = ? AND principal_id = ?",
                (idempotency_key, 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,
            copy.deepcopy(operation),
            {CompatibilityOperationState.CLAIMED},
            CompatibilityOperationState.IN_FLIGHT,
            None,
        )

    async def mark_ambiguous(
        self, operation: CompatibilityPurchaseOperation
    ) -> CompatibilityPurchaseOperation:
        return await asyncio.to_thread(
            self._transition,
            copy.deepcopy(operation),
            {CompatibilityOperationState.CLAIMED, CompatibilityOperationState.IN_FLIGHT},
            CompatibilityOperationState.AMBIGUOUS,
            None,
        )

    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")
        snapshot = copy.deepcopy(dict(result))
        _validate_persistable_payload(snapshot, context="legacy result")
        return await asyncio.to_thread(
            self._transition,
            copy.deepcopy(operation),
            {
                CompatibilityOperationState.IN_FLIGHT,
                CompatibilityOperationState.AMBIGUOUS,
                CompatibilityOperationState.PENDING,
            },
            state,
            snapshot,
        )

    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(
        self, operation: CompatibilityPurchaseOperation
    ) -> CompatibilityPurchaseOperation:
        return await asyncio.to_thread(
            self._transition,
            copy.deepcopy(operation),
            {CompatibilityOperationState.AMBIGUOUS},
            CompatibilityOperationState.CLAIMED,
            None,
        )

    def _transition(
        self,
        operation: CompatibilityPurchaseOperation,
        allowed: set[CompatibilityOperationState],
        target: CompatibilityOperationState,
        result: dict[str, Any] | None,
    ) -> CompatibilityPurchaseOperation:
        with closing(self._connect()) as conn, conn:
            conn.execute("BEGIN IMMEDIATE")
            row = conn.execute(
                "SELECT * FROM adcp_compat_operations WHERE operation_id = ?",
                (operation.operation_id,),
            ).fetchone()
            if row is None:
                raise _state_error("operation is missing from continuation store")
            current = _decode_operation(row)
            if (
                current.principal_id != operation.principal_id
                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
                or current.reserved_result_bytes != operation.reserved_result_bytes
            ):
                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}"
                )
            result_json = _dumps(result) if result is not None else None
            if (
                result_json is not None
                and len(result_json.encode("utf-8")) > current.reserved_result_bytes
            ):
                raise CompatibilityContinuationError(
                    CompatibilityContinuationErrorCode.STORE_QUOTA_EXCEEDED,
                    "operation result exceeds its reserved durable capacity",
                    recovery_guidance=(
                        "Reconcile using a result within the operation's "
                        "original durable reservation."
                    ),
                    details={"reserved_result_bytes": current.reserved_result_bytes},
                )
            reserved_result_bytes = current.reserved_result_bytes
            if current.state is CompatibilityOperationState.CLAIMED and target in {
                CompatibilityOperationState.IN_FLIGHT,
                CompatibilityOperationState.AMBIGUOUS,
            }:
                reserved_result_bytes = self.max_payload_bytes
            elif target in {
                CompatibilityOperationState.CLAIMED,
                CompatibilityOperationState.SUCCEEDED,
                CompatibilityOperationState.FAILED,
            }:
                reserved_result_bytes = 0
            updated_at = _format_datetime(self._clock())
            updated = conn.execute(
                """
                UPDATE adcp_compat_operations
                SET state = ?, result_json = ?, reserved_result_bytes = ?,
                    revision = revision + 1, updated_at = ?
                WHERE operation_id = ? AND state = ? AND revision = ?
                """,
                (
                    target.value,
                    result_json,
                    reserved_result_bytes,
                    updated_at,
                    current.operation_id,
                    current.state.value,
                    current.revision,
                ),
            )
            if updated.rowcount != 1:
                raise _state_error("operation state changed concurrently")
            if current.state is CompatibilityOperationState.CLAIMED and target in {
                CompatibilityOperationState.IN_FLIGHT,
                CompatibilityOperationState.AMBIGUOUS,
            }:
                # Reserve one full result payload before the seller mutation can
                # run. Later pending/terminal writes consume that reservation,
                # so another principal cannot strand an executed mutation by
                # filling the global quota between execution and completion.
                self._enforce_ledger_quota(conn, principal_id=current.principal_id)
            conn.commit()
            return CompatibilityPurchaseOperation(
                operation_id=current.operation_id,
                principal_id=current.principal_id,
                idempotency_key=current.idempotency_key,
                token_hash=current.token_hash,
                payload_hash=current.payload_hash,
                state=target,
                revision=current.revision + 1,
                execution_input=copy.deepcopy(current.execution_input),
                reserved_result_bytes=reserved_result_bytes,
                result=copy.deepcopy(result),
            )

    def _enforce_payload_quota(self, *serialized_values: str) -> None:
        payload_bytes = sum(len(value.encode("utf-8")) for value in serialized_values)
        if payload_bytes > self.max_payload_bytes:
            raise _quota_error(self.max_records, self.max_bytes)

    def _enforce_ledger_quota(self, conn: sqlite3.Connection, *, principal_id: str) -> None:
        records = conn.execute(
            "SELECT "
            "(SELECT COUNT(*) FROM adcp_compat_continuations) + "
            "(SELECT COUNT(*) FROM adcp_compat_operations) + "
            "(SELECT COUNT(*) FROM adcp_compat_issuance_tombstones)"
        ).fetchone()[0]
        continuation_bytes = conn.execute(
            """
            SELECT COALESCE(SUM(
                length(CAST(token_hash AS BLOB)) +
                length(CAST(COALESCE(issuance_fingerprint, '') AS BLOB)) +
                length(CAST(COALESCE(issuance_binding_hash, '') AS BLOB)) +
                length(CAST(principal_id AS BLOB)) +
                length(CAST(account_identity AS BLOB)) +
                length(CAST(source_adcp_version AS BLOB)) +
                length(CAST(expires_at AS BLOB)) +
                length(CAST(observed_request_json AS BLOB)) +
                length(CAST(observed_response_json AS BLOB)) +
                length(CAST(observed_payload_hash AS BLOB)) +
                length(CAST(product_ids_json AS BLOB)) +
                length(CAST(COALESCE(projected_products_json, '') AS BLOB)) +
                length(CAST(losses_json AS BLOB)) +
                length(CAST(target_binding AS BLOB)) +
                length(CAST(COALESCE(listed_purchase_context_json, '') AS BLOB))
            ), 0) FROM adcp_compat_continuations
            """
        ).fetchone()[0]
        tombstone_bytes = conn.execute(
            """
            SELECT COALESCE(SUM(
                length(CAST(token_hash AS BLOB)) +
                length(CAST(principal_id AS BLOB)) +
                length(CAST(COALESCE(issuance_fingerprint, '') AS BLOB)) +
                length(CAST(COALESCE(issuance_binding_hash, '') AS BLOB))
            ), 0) FROM adcp_compat_issuance_tombstones
            """
        ).fetchone()[0]
        operation_bytes = conn.execute(
            """
            SELECT COALESCE(SUM(
                length(CAST(operation_id AS BLOB)) +
                length(CAST(principal_id AS BLOB)) +
                length(CAST(idempotency_key AS BLOB)) +
                length(CAST(token_hash AS BLOB)) +
                length(CAST(payload_hash AS BLOB)) +
                CASE
                    WHEN state = 'pending' THEN length(CAST('succeeded' AS BLOB))
                    ELSE length(CAST(state AS BLOB))
                END +
                length(CAST(execution_input_json AS BLOB)) +
                CASE
                    WHEN state IN ('in_flight', 'pending', 'ambiguous')
                        THEN reserved_result_bytes
                    ELSE length(CAST(COALESCE(result_json, '') AS BLOB))
                END
            ), 0) FROM adcp_compat_operations
            """
        ).fetchone()[0]
        if (
            records > self.max_records
            or continuation_bytes + operation_bytes + tombstone_bytes > self.max_bytes
        ):
            raise _quota_error(self.max_records, self.max_bytes)
        principal_records = conn.execute(
            "SELECT "
            "(SELECT COUNT(*) FROM adcp_compat_continuations WHERE principal_id = ?) + "
            "(SELECT COUNT(*) FROM adcp_compat_operations WHERE principal_id = ?) + "
            "(SELECT COUNT(*) FROM adcp_compat_issuance_tombstones WHERE principal_id = ?)",
            (principal_id, principal_id, principal_id),
        ).fetchone()[0]
        principal_continuation_bytes = conn.execute(
            """
            SELECT COALESCE(SUM(
                length(CAST(token_hash AS BLOB)) +
                length(CAST(COALESCE(issuance_fingerprint, '') AS BLOB)) +
                length(CAST(COALESCE(issuance_binding_hash, '') AS BLOB)) +
                length(CAST(principal_id AS BLOB)) +
                length(CAST(account_identity AS BLOB)) +
                length(CAST(source_adcp_version AS BLOB)) +
                length(CAST(expires_at AS BLOB)) +
                length(CAST(observed_request_json AS BLOB)) +
                length(CAST(observed_response_json AS BLOB)) +
                length(CAST(observed_payload_hash AS BLOB)) +
                length(CAST(product_ids_json AS BLOB)) +
                length(CAST(COALESCE(projected_products_json, '') AS BLOB)) +
                length(CAST(losses_json AS BLOB)) +
                length(CAST(target_binding AS BLOB)) +
                length(CAST(COALESCE(listed_purchase_context_json, '') AS BLOB))
            ), 0) FROM adcp_compat_continuations WHERE principal_id = ?
            """,
            (principal_id,),
        ).fetchone()[0]
        principal_operation_bytes = conn.execute(
            """
            SELECT COALESCE(SUM(
                length(CAST(operation_id AS BLOB)) +
                length(CAST(principal_id AS BLOB)) +
                length(CAST(idempotency_key AS BLOB)) +
                length(CAST(token_hash AS BLOB)) +
                length(CAST(payload_hash AS BLOB)) +
                CASE
                    WHEN state = 'pending' THEN length(CAST('succeeded' AS BLOB))
                    ELSE length(CAST(state AS BLOB))
                END +
                length(CAST(execution_input_json AS BLOB)) +
                CASE
                    WHEN state IN ('in_flight', 'pending', 'ambiguous')
                        THEN reserved_result_bytes
                    ELSE length(CAST(COALESCE(result_json, '') AS BLOB))
                END
            ), 0) FROM adcp_compat_operations WHERE principal_id = ?
            """,
            (principal_id,),
        ).fetchone()[0]
        principal_tombstone_bytes = conn.execute(
            """
            SELECT COALESCE(SUM(
                length(CAST(token_hash AS BLOB)) +
                length(CAST(principal_id AS BLOB)) +
                length(CAST(COALESCE(issuance_fingerprint, '') AS BLOB)) +
                length(CAST(COALESCE(issuance_binding_hash, '') AS BLOB))
            ), 0) FROM adcp_compat_issuance_tombstones WHERE principal_id = ?
            """,
            (principal_id,),
        ).fetchone()[0]
        if (
            principal_records > self.max_records_per_principal
            or principal_continuation_bytes + principal_operation_bytes + principal_tombstone_bytes
            > self.max_bytes_per_principal
        ):
            raise _quota_error(self.max_records_per_principal, self.max_bytes_per_principal)

    async def purge_resolved_before(self, cutoff: datetime) -> int:
        """Delete only old terminal or never-claimed continuations.

        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)
        now_utc = _as_utc(self._clock())
        with closing(self._connect()) as conn:
            rows = conn.execute(
                """
                SELECT
                    continuation.token_hash,
                    continuation.issuance_fingerprint,
                    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.issuance_fingerprint IS NOT NULL
                  AND (
                       operation.state IN ('succeeded', 'failed')
                       OR operation.operation_id IS NULL
                  )
                """,
            ).fetchall()
        candidates = {
            row["token_hash"]: (
                row["expires_at"],
                row["continuation_updated_at"],
                row["operation_id"],
                row["state"],
                row["operation_updated_at"],
                row["issuance_fingerprint"],
            )
            for row in rows
            if row["issuance_fingerprint"] is not None
            and (
                (
                    row["state"]
                    in {
                        CompatibilityOperationState.SUCCEEDED.value,
                        CompatibilityOperationState.FAILED.value,
                    }
                    and _parse_datetime(row["expires_at"]) <= now_utc
                    and _parse_datetime(row["operation_updated_at"]) < cutoff_utc
                )
                or (
                    row["state"] is None
                    and _parse_datetime(row["expires_at"]) <= now_utc
                    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, now_utc=now_utc)
        return deleted

    def _purge_candidate_batch(
        self,
        token_hashes: list[str],
        candidates: Mapping[
            str,
            tuple[str, str, str | None, str | None, str | None, str],
        ],
        *,
        now_utc: datetime,
    ) -> 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.principal_id,
                    continuation.issuance_fingerprint,
                    continuation.issuance_binding_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_rows = [
                row
                for row in rows
                if candidates.get(row["token_hash"])
                == (
                    row["expires_at"],
                    row["continuation_updated_at"],
                    row["operation_id"],
                    row["state"],
                    row["operation_updated_at"],
                    row["issuance_fingerprint"],
                )
                and row["issuance_fingerprint"] is not None
                and _parse_datetime(row["expires_at"]) <= now_utc
            ]
            if not confirmed_rows:
                return 0
            confirmed = [row["token_hash"] for row in confirmed_rows]
            confirmed_placeholders = ",".join("?" for _ in confirmed)
            conn.executemany(
                """
                INSERT INTO adcp_compat_issuance_tombstones (
                    token_hash, principal_id, issuance_fingerprint,
                    issuance_binding_hash, retired_at
                ) VALUES (?, ?, ?, ?, ?)
                """,
                [
                    (
                        row["token_hash"],
                        row["principal_id"],
                        row["issuance_fingerprint"],
                        row["issuance_binding_hash"],
                        _format_datetime(now_utc),
                    )
                    for row in confirmed_rows
                ],
            )
            conn.execute(
                f"DELETE FROM adcp_compat_operations "
                f"WHERE token_hash IN ({confirmed_placeholders})",
                confirmed,
            )
            removed = conn.execute(
                f"DELETE FROM adcp_compat_continuations "
                f"WHERE token_hash IN ({confirmed_placeholders})",
                confirmed,
            ).rowcount
            conn.commit()
            return removed

Durable local continuation ledger backed by a SQLite file.

Class variables

var is_durable : ClassVar[bool]

Methods

async def claim(self,
token_hash: str,
*,
principal_id: str,
idempotency_key: str,
payload_hash: str,
execution_input: Mapping[str, Any],
now: datetime) ‑> CompatibilityPurchaseOperation

Expand source code

async def claim(
    self,
    token_hash: str,
    *,
    principal_id: str,
    idempotency_key: str,
    payload_hash: str,
    execution_input: Mapping[str, Any],
    now: datetime,
) -> CompatibilityPurchaseOperation:
    snapshot = copy.deepcopy(dict(execution_input))
    _validate_persistable_payload(snapshot, context="execution input")
    return await asyncio.to_thread(
        self._claim,
        token_hash,
        principal_id,
        idempotency_key,
        payload_hash,
        snapshot,
        now,
    )

async def complete(self,
operation: CompatibilityPurchaseOperation,
result: Mapping[str, Any],
*,
state: CompatibilityOperationState = CompatibilityOperationState.SUCCEEDED) ‑> CompatibilityPurchaseOperation

Expand source code

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")
    snapshot = copy.deepcopy(dict(result))
    _validate_persistable_payload(snapshot, context="legacy result")
    return await asyncio.to_thread(
        self._transition,
        copy.deepcopy(operation),
        {
            CompatibilityOperationState.IN_FLIGHT,
            CompatibilityOperationState.AMBIGUOUS,
            CompatibilityOperationState.PENDING,
        },
        state,
        snapshot,
    )

async def fence_in_flight(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

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 get_continuation(self, token_hash: str, *, principal_id: str) ‑> LegacyPurchaseContinuation | None

Expand source code

async def get_continuation(
    self, token_hash: str, *, principal_id: str
) -> LegacyPurchaseContinuation | None:
    return await asyncio.to_thread(self._get_continuation, token_hash, principal_id)

async def get_operation(self, operation_id: str, *, principal_id: str) ‑> CompatibilityPurchaseOperation | None

Expand source code

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)

async def get_operation_by_idempotency_key(self, idempotency_key: str, *, principal_id: str) ‑> CompatibilityPurchaseOperation | None

Expand source code

async def get_operation_by_idempotency_key(
    self, idempotency_key: str, *, principal_id: str
) -> CompatibilityPurchaseOperation | None:
    return await asyncio.to_thread(
        self._get_operation_by_idempotency_key, idempotency_key, principal_id
    )

async def mark_ambiguous(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def mark_ambiguous(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation:
    return await asyncio.to_thread(
        self._transition,
        copy.deepcopy(operation),
        {CompatibilityOperationState.CLAIMED, CompatibilityOperationState.IN_FLIGHT},
        CompatibilityOperationState.AMBIGUOUS,
        None,
    )

async def mark_in_flight(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def mark_in_flight(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation:
    return await asyncio.to_thread(
        self._transition,
        copy.deepcopy(operation),
        {CompatibilityOperationState.CLAIMED},
        CompatibilityOperationState.IN_FLIGHT,
        None,
    )

async def purge_resolved_before(self, cutoff: datetime) ‑> int

Expand source code

async def purge_resolved_before(self, cutoff: datetime) -> int:
    """Delete only old terminal or never-claimed continuations.

    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)

Delete only old terminal or never-claimed continuations.

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.

async def put_continuation(self,
continuation: LegacyPurchaseContinuation) ‑> None

Expand source code

async def put_continuation(self, continuation: LegacyPurchaseContinuation) -> None:
    await asyncio.to_thread(self._put_continuation, copy.deepcopy(continuation))

async def resume_after_not_applied(self,
operation: CompatibilityPurchaseOperation) ‑> CompatibilityPurchaseOperation

Expand source code

async def resume_after_not_applied(
    self, operation: CompatibilityPurchaseOperation
) -> CompatibilityPurchaseOperation:
    return await asyncio.to_thread(
        self._transition,
        copy.deepcopy(operation),
        {CompatibilityOperationState.AMBIGUOUS},
        CompatibilityOperationState.CLAIMED,
        None,
    )