Module adcp.reporting.submissions.submit

Explicit low-level receipt submission with durable uncertainty semantics.

Functions

async def submit_reporting_receipts(client: ReportingReceiptSubmissionClient,
*,
authorizer: ReportingSubmissionAuthorizer,
store: ReportingSubmissionIntentStore,
receipts: Sequence[ReportingSubmissionReceipt] | None = None,
timeout_seconds: float = 30.0) ‑> ReportingSubmissionResult

Expand source code

async def submit_reporting_receipts(
    client: ReportingReceiptSubmissionClient,
    *,
    authorizer: ReportingSubmissionAuthorizer,
    store: ReportingSubmissionIntentStore,
    receipts: Sequence[ReportingSubmissionReceipt] | None = None,
    timeout_seconds: float = 30.0,
) -> ReportingSubmissionResult:
    """Reserve or resume a receipt plan without replacing uncertain work.

    Omit ``receipts`` to resume the latest intent for the freshly authorized
    scope, including its completed outcomes after a lost return. Otherwise the
    supplied plan is frozen and atomically reserved before ANY network write.
    An earlier pending scope intent takes priority, with proposal_deferred=True.
    The caller must reconsider the deferred plan against fresh seller history.

    A transport failure, timeout, failed task or malformed response leaves the
    chunk uncertain and returns pending=True. Retry within the live protocol
    version uses the exact durable body and idempotency key. A pending intent
    from a retired version remains readable and byte-identical but is refused
    before transport; it is never rewritten as a new-version request.
    Cancellation propagates with the intent still retained.
    Storage failure raises a closed error; resume the stored scope because the
    last commit may have succeeded. There is no expiry/abandon/replacement path.

    Every item failure is retained alongside successes. Completion only means
    every submission outcome is confirmed; it does not mean every receipt was
    recorded, accepted, readable, or sufficient for definitive reconciliation.
    This API does not select history, build adjustment evidence, post consumer
    statuses, or modify the original reconciliation/checkpoint interfaces.
    """
    if (
        isinstance(timeout_seconds, bool)
        or not isinstance(timeout_seconds, (int, float))
        or not isfinite(timeout_seconds)
        or timeout_seconds <= 0
    ):
        raise ReportingSubmissionError(ReportingSubmissionCode.INVALID_PLAN)
    scope = await _authorize(client, authorizer)
    proposed = (
        prepare_reporting_receipt_submission(scope, receipts) if receipts is not None else None
    )
    state = await _storage(store.reserve(proposed) if proposed is not None else store.get(scope))
    if state is None:
        raise ReportingSubmissionError(ReportingSubmissionCode.NOT_FOUND)
    _check_state(state, scope)
    deferred = proposed is not None and proposed.submission_id != state.submission_id
    if state.pending and state.adcp_version != _VERSION:
        # An unresolved older reservation retains its exact wire bytes, but
        # the current SDK must not issue a new request under a retired pin.
        raise ReportingSubmissionError(ReportingSubmissionCode.INVALID_PLAN)
    while state.pending:
        await _authorize(client, authorizer, scope)
        chunk = state.confirmed_chunks
        request = state.request(chunk)
        response = None
        try:
            response = await asyncio.wait_for(
                client.sync_reporting_receipts(request), timeout=timeout_seconds
            )
        except Exception:
            response = None
        if response is None:
            return ReportingSubmissionResult(
                state, deferred, ReportingSubmissionCode.TRANSPORT_UNCERTAIN
            )
        if (
            response.success is not True
            or response.status != "completed"
            or not isinstance(response.data, SyncReportingReceiptsResponse)
        ):
            return ReportingSubmissionResult(
                state, deferred, ReportingSubmissionCode.RESPONSE_UNCONFIRMED
            )
        invalid = False
        try:
            updated = await _storage(
                store.confirm(scope, state.submission_id, chunk, response.data)
            )
        except ReportingSubmissionError as error:
            if error.code != ReportingSubmissionCode.INVALID_RESPONSE:
                raise
            invalid = True
        if invalid:
            return ReportingSubmissionResult(
                state, deferred, ReportingSubmissionCode.INVALID_RESPONSE
            )
        _check_state(updated, scope, state)
        if updated.confirmed_chunks <= chunk:
            raise ReportingSubmissionError(ReportingSubmissionCode.HISTORY_CORRUPT)
        state = updated
    await _authorize(client, authorizer, scope)
    return ReportingSubmissionResult(state, deferred)

Reserve or resume a receipt plan without replacing uncertain work.

Omit receipts to resume the latest intent for the freshly authorized scope, including its completed outcomes after a lost return. Otherwise the supplied plan is frozen and atomically reserved before ANY network write. An earlier pending scope intent takes priority, with proposal_deferred=True. The caller must reconsider the deferred plan against fresh seller history.

A transport failure, timeout, failed task or malformed response leaves the chunk uncertain and returns pending=True. Retry within the live protocol version uses the exact durable body and idempotency key. A pending intent from a retired version remains readable and byte-identical but is refused before transport; it is never rewritten as a new-version request. Cancellation propagates with the intent still retained. Storage failure raises a closed error; resume the stored scope because the last commit may have succeeded. There is no expiry/abandon/replacement path.

Every item failure is retained alongside successes. Completion only means every submission outcome is confirmed; it does not mean every receipt was recorded, accepted, readable, or sufficient for definitive reconciliation. This API does not select history, build adjustment evidence, post consumer statuses, or modify the original reconciliation/checkpoint interfaces.

Classes

class ReportingReceiptSubmissionClient (*args, **kwargs)

Expand source code

class ReportingReceiptSubmissionClient(Protocol):
    async def sync_reporting_receipts(
        self, request: SyncReportingReceiptsRequest
    ) -> TaskResult[SyncReportingReceiptsResponse]: ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

Ancestors

• typing.Protocol
• typing.Generic

Methods

async def sync_reporting_receipts(self, request: SyncReportingReceiptsRequest) ‑> TaskResult[SyncReportingReceiptsResponse]

Expand source code

async def sync_reporting_receipts(
    self, request: SyncReportingReceiptsRequest
) -> TaskResult[SyncReportingReceiptsResponse]: ...

class ReportingSubmissionAuthorizer (*args, **kwargs)

Expand source code

class ReportingSubmissionAuthorizer(Protocol):
    """Trusted application adapter resolving the exact client's current access.

    Resolve seller identity from trusted client/registry configuration, account
    from an authorized seller account lookup, and consumer from authenticated
    credentials/registry. Never source these identities from receipt payloads,
    an asserted account reference, or a debug/raw response. Aliases must already
    be resolved. Raise on revoked access or an identity disagreement.

    Called before any store read/reservation, before each send, and before
    returning completed cached outcomes. It must authorize replay afresh too.
    """

    async def __call__(
        self, client: ReportingReceiptSubmissionClient
    ) -> ReportingSubmissionScope: ...

Trusted application adapter resolving the exact client's current access.

Resolve seller identity from trusted client/registry configuration, account from an authorized seller account lookup, and consumer from authenticated credentials/registry. Never source these identities from receipt payloads, an asserted account reference, or a debug/raw response. Aliases must already be resolved. Raise on revoked access or an identity disagreement.

Called before any store read/reservation, before each send, and before returning completed cached outcomes. It must authorize replay afresh too.

Ancestors

• typing.Protocol
• typing.Generic