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

Durable continuation of lossy AdCP 3.2 legacy purchases …

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

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,
    )