Module adcp.compat.purchase_continuation

Durable continuation of lossy AdCP 3.2 legacy purchases.

products_available compatibility projections can expose a legacy_create continuation when an established 2.5/3.0/3.1 seller returned products without an atomic 3.2 proposal. This module redeems that continuation without weakening its security boundary: all bindings are checked before the first seller mutation and the token is claimed exactly once in durable state.

The coordinator is application-owned. It does not choose credentials, derive the authenticated principal, or route a seller connection. Applications must provide those values and an executor bound to the original seller session.

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 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 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:
                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 outcome.valid or outcome.variant == "skipped":
            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 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