Module adcp.reporting.submissions.models

Immutable buyer intent records and closed diagnostics.

These records describe submission, not evidence selection or reconciliation. The caller must validate its receipt plan against the complete seller history. Received adjustment evidence must be retained separately, before model parsing.

Functions

def confirm_submission(submission: ReportingReceiptSubmission,
ordinal: int,
response: SyncReportingReceiptsResponse | bytes) ‑> ReportingReceiptSubmission

Expand source code

def confirm_submission(
    submission: ReportingReceiptSubmission,
    ordinal: int,
    response: SyncReportingReceiptsResponse | bytes,
) -> ReportingReceiptSubmission:
    if (
        type(ordinal) is not int
        or not 0 <= ordinal < submission.chunk_count
        or ordinal > submission.confirmed_chunks
    ):
        raise ReportingSubmissionError(ReportingSubmissionCode.HISTORY_CORRUPT)
    confirmed = _confirmation(submission, ordinal, response)
    if ordinal < submission.confirmed_chunks:
        # First confirmation is immutable. Concurrent exact replays may use
        # recorded/unchanged differently, but must agree on the actual evidence
        # and failures. A contradictory reply cannot overwrite retained results.
        previous = json.loads(submission._confirmed[ordinal])
        repeated = json.loads(confirmed)
        for results in (previous, repeated):
            for item in results:
                if item["result"] in {"recorded", "unchanged"}:
                    item["result"] = "recorded"
        if previous != repeated:
            raise ReportingSubmissionError(ReportingSubmissionCode.INVALID_RESPONSE)
        return submission
    chunks = (*submission._confirmed, confirmed)
    if _confirmation_bytes(chunks) > MAX_CONFIRMATION_BYTES:
        raise ReportingSubmissionError(ReportingSubmissionCode.INVALID_RESPONSE)
    return replace(submission, _confirmed=chunks)

def decode_submission(scope: ReportingSubmissionScope,
row: Sequence[Any]) ‑> ReportingReceiptSubmission

Expand source code

def decode_submission(
    scope: ReportingSubmissionScope, row: Sequence[Any]
) -> ReportingReceiptSubmission:
    result = None
    try:
        identifier, plan, plan_digest, confirmed, confirmed_digest, pending = row
        if (
            len(plan.encode()) > MAX_SUBMISSION_BYTES
            or len(confirmed.encode()) > MAX_CONFIRMATION_BYTES
        ):
            raise ValueError
        if (
            hashlib.sha256(plan.encode()).hexdigest() != plan_digest
            or hashlib.sha256(confirmed.encode()).hexdigest() != confirmed_digest
        ):
            raise ValueError
        candidate = ReportingReceiptSubmission(
            scope,
            identifier,
            plan.encode(),
            _decode_chunks(confirmed),
        )
        validate_submission(candidate)
        if (
            type(pending) is not bool
            or pending != candidate.pending
            or encode_submission(candidate) != (plan, plan_digest, confirmed, confirmed_digest)
        ):
            raise ValueError
        result = candidate
    except Exception:
        result = None
    if result is None:
        raise ReportingSubmissionError(ReportingSubmissionCode.HISTORY_CORRUPT)
    return result

def encode_submission(submission: ReportingReceiptSubmission) ‑> tuple[str, str, str, str]

Expand source code

def encode_submission(submission: ReportingReceiptSubmission) -> tuple[str, str, str, str]:
    """Exact text/hashes for PostgreSQL; JSONB is not the identity store."""
    confirmed = b"[" + b",".join(submission._confirmed) + b"]"
    return (
        submission._plan.decode(),
        hashlib.sha256(submission._plan).hexdigest(),
        confirmed.decode(),
        hashlib.sha256(confirmed).hexdigest(),
    )

Exact text/hashes for PostgreSQL; JSONB is not the identity store.

def freeze_response(response: SyncReportingReceiptsResponse) ‑> bytes

Expand source code

def freeze_response(response: SyncReportingReceiptsResponse) -> bytes:
    """Detach the caller's response once, before any await or storage lock."""
    encoded = None
    try:
        encoded = canonical_json_utf8_v1(response.model_dump(mode="json", exclude_none=True))
        if len(encoded) > MAX_RESPONSE_BYTES:
            raise ValueError
    except Exception:
        encoded = None
    if encoded is None:
        raise ReportingSubmissionError(ReportingSubmissionCode.INVALID_RESPONSE)
    return encoded

Detach the caller's response once, before any await or storage lock.

def is_completed_legacy_replay(prior: ReportingReceiptSubmission,
proposed: ReportingReceiptSubmission) ‑> bool

Expand source code

def is_completed_legacy_replay(
    prior: ReportingReceiptSubmission, proposed: ReportingReceiptSubmission
) -> bool:
    """Match a completed rc.6/rc.7 intent to its 3.2 proposal without changing its bytes.

    Both plans must already have passed full submission validation. Their IDs
    exclude the request version, so compare the frozen items as well as the ID.
    Pending old requests are never eligible for a cross-version replay.
    """
    return (
        not prior.pending
        and prior.adcp_version in _LEGACY_VERSIONS
        and proposed.adcp_version == _VERSION
        and prior.scope == proposed.scope
        and prior.submission_id == proposed.submission_id
        and json.loads(prior._plan)["items"] == json.loads(proposed._plan)["items"]
    )

Match a completed rc.6/rc.7 intent to its 3.2 proposal without changing its bytes.

Both plans must already have passed full submission validation. Their IDs exclude the request version, so compare the frozen items as well as the ID. Pending old requests are never eligible for a cross-version replay.

def prepare_reporting_receipt_submission(scope: ReportingSubmissionScope,
receipts: Sequence[ReportingSubmissionReceipt]) ‑> ReportingReceiptSubmission

Expand source code

def prepare_reporting_receipt_submission(
    scope: ReportingSubmissionScope,
    receipts: Sequence[ReportingSubmissionReceipt],
) -> ReportingReceiptSubmission:
    """Freeze an already validated receipt plan, with at most 100 items per call.

    Outbound typed values are normalized once, then retained byte-for-byte.
    This is not a capture or proof of raw inbound adjustment evidence. Requests
    contain the authorized resolved account only, with no caller-provided
    account, idempotency key, context, extensions or credentials.
    """
    return _prepare_for_version(scope, receipts, _VERSION)

Freeze an already validated receipt plan, with at most 100 items per call.

Outbound typed values are normalized once, then retained byte-for-byte. This is not a capture or proof of raw inbound adjustment evidence. Requests contain the authorized resolved account only, with no caller-provided account, idempotency key, context, extensions or credentials.

def validate_submission(submission: ReportingReceiptSubmission) ‑> None

Expand source code

def validate_submission(submission: ReportingReceiptSubmission) -> None:
    """Check fresh bytes, bounds, identity and ordered confirmed prefix every time.

    Exact immutable-byte proofs avoid repeating plan and response schemas. No
    stored digest or mutable model is sufficient to hit either bounded cache.
    """
    valid = False
    try:
        requests = _validated_requests(submission)
        if (
            type(submission._confirmed) is not tuple
            or len(submission._confirmed) > len(requests)
            or _confirmation_bytes(submission._confirmed) > MAX_CONFIRMATION_BYTES
        ):
            raise ValueError
        for ordinal, chunk in enumerate(submission._confirmed):
            if type(chunk) is not bytes or len(chunk) > MAX_RESPONSE_BYTES:
                raise ValueError
            key = _confirmation_key(submission, requests[ordinal], chunk)
            if CONFIRMATIONS.get(key) is None:
                if _restore_confirmation(submission, ordinal, chunk) != chunk:
                    raise ValueError
                CONFIRMATIONS.put(key, ())
        valid = True
    except Exception:
        valid = False
    if not valid:
        raise ReportingSubmissionError(ReportingSubmissionCode.HISTORY_CORRUPT)

Check fresh bytes, bounds, identity and ordered confirmed prefix every time.

Exact immutable-byte proofs avoid repeating plan and response schemas. No stored digest or mutable model is sufficient to hit either bounded cache.

Classes

class ReportingReceiptFailureCode (*args, **kwds)

Expand source code

class ReportingReceiptFailureCode(str, Enum):
    """Known item failures; unrecognized seller codes map to UNKNOWN.

    Seller messages/details are deliberately not persisted or exposed. Each
    failed item and each error position is retained, including unknown codes.
    """

    UNKNOWN = "UNKNOWN"
    INVALID_REQUEST = "INVALID_REQUEST"
    UNAUTHORIZED = "UNAUTHORIZED"
    RATE_LIMITED = "RATE_LIMITED"
    NOT_SUPPORTED = "NOT_SUPPORTED"
    IDEMPOTENCY_CONFLICT = "IDEMPOTENCY_CONFLICT"
    INVALID_REPORTING_RECORD = "INVALID_REPORTING_RECORD"
    REPORTING_RECORD_UNAVAILABLE = "REPORTING_RECORD_UNAVAILABLE"
    REPORTING_HISTORY_CORRUPT = "REPORTING_HISTORY_CORRUPT"
    REPORTING_IDENTITY_CONFLICT = "REPORTING_IDENTITY_CONFLICT"
    REPORTING_TIME_INVALID = "REPORTING_TIME_INVALID"
    RECEIPTS_NOT_ENABLED = "RECEIPTS_NOT_ENABLED"
    RECEIVED_AT_READ_ONLY = "RECEIVED_AT_READ_ONLY"
    RECEIPT_PROFILE_MISMATCH = "RECEIPT_PROFILE_MISMATCH"
    RECEIPT_TOTALS_MISMATCH = "RECEIPT_TOTALS_MISMATCH"
    RECEIPT_EVIDENCE_MISMATCH = "RECEIPT_EVIDENCE_MISMATCH"
    MATERIALIZATION_UNREADABLE = "MATERIALIZATION_UNREADABLE"
    ADJUSTMENT_REQUIRES_OFFICIAL = "ADJUSTMENT_REQUIRES_OFFICIAL"
    ADJUSTMENT_ORDER_INVALID = "ADJUSTMENT_ORDER_INVALID"
    ADJUSTMENT_DIGEST_MISMATCH = "ADJUSTMENT_DIGEST_MISMATCH"
    ACCEPTED_RECEIPT_TERMINAL = "ACCEPTED_RECEIPT_TERMINAL"

Known item failures; unrecognized seller codes map to UNKNOWN.

Seller messages/details are deliberately not persisted or exposed. Each failed item and each error position is retained, including unknown codes.

Ancestors

• builtins.str
• enum.Enum

Class variables

var ACCEPTED_RECEIPT_TERMINAL

var ADJUSTMENT_DIGEST_MISMATCH

var ADJUSTMENT_ORDER_INVALID

var ADJUSTMENT_REQUIRES_OFFICIAL

var IDEMPOTENCY_CONFLICT

var INVALID_REPORTING_RECORD

var INVALID_REQUEST

var MATERIALIZATION_UNREADABLE

var NOT_SUPPORTED

var RATE_LIMITED

var RECEIPTS_NOT_ENABLED

var RECEIPT_EVIDENCE_MISMATCH

var RECEIPT_PROFILE_MISMATCH

var RECEIPT_TOTALS_MISMATCH

var RECEIVED_AT_READ_ONLY

var REPORTING_HISTORY_CORRUPT

var REPORTING_IDENTITY_CONFLICT

var REPORTING_RECORD_UNAVAILABLE

var REPORTING_TIME_INVALID

var UNAUTHORIZED

var UNKNOWN

class ReportingReceiptOutcome (ordinal: int,
kind: ReceiptKind,
result: ReceiptResult,
error_codes: tuple[ReportingReceiptFailureCode, ...] = ())

Expand source code

@dataclass(frozen=True)
class ReportingReceiptOutcome:
    """One confirmed item, in original input order, with fresh model views."""

    ordinal: int
    kind: ReceiptKind
    result: ReceiptResult
    error_codes: tuple[ReportingReceiptFailureCode, ...] = ()
    _submitted: bytes = field(default=b"", repr=False)
    _stored: bytes | None = field(default=None, repr=False)

    @property
    def submitted_receipt(self) -> ReportingSubmissionReceipt:
        return _receipt(self.kind, json.loads(self._submitted))

    @property
    def receipt(self) -> ReportingSubmissionReceipt | None:
        """The seller's actual stored receipt on success, including received_at."""
        return _receipt(self.kind, json.loads(self._stored)) if self._stored is not None else None

One confirmed item, in original input order, with fresh model views.

Instance variables

var error_codes : tuple[ReportingReceiptFailureCode, ...]

var kind : Literal['receipt', 'adjustment_receipt']

var ordinal : int

prop receipt : ReportingSubmissionReceipt | None

Expand source code

@property
def receipt(self) -> ReportingSubmissionReceipt | None:
    """The seller's actual stored receipt on success, including received_at."""
    return _receipt(self.kind, json.loads(self._stored)) if self._stored is not None else None

The seller's actual stored receipt on success, including received_at.

var result : Literal['recorded', 'unchanged', 'failed']

prop submitted_receipt : ReportingSubmissionReceipt

Expand source code

@property
def submitted_receipt(self) -> ReportingSubmissionReceipt:
    return _receipt(self.kind, json.loads(self._submitted))

class ReportingReceiptSubmission (scope: ReportingSubmissionScope,
submission_id: str,
_plan: bytes)

Expand source code

@dataclass(frozen=True)
class ReportingReceiptSubmission:
    """Exact reserved plan and its immutable, fully confirmed chunk prefix.

    Construct with :func:`prepare_reporting_receipt_submission`. Stores validate
    persisted records before returning them. Private bytes keep mutable caller
    models, driver results, and representations out of the retained identity.
    """

    scope: ReportingSubmissionScope = field(repr=False)
    submission_id: str = field(repr=False)
    _plan: bytes = field(repr=False)
    _confirmed: tuple[bytes, ...] = field(default=(), repr=False)

    @property
    def chunk_count(self) -> int:
        return len(_validated_requests(self))

    @property
    def confirmed_chunks(self) -> int:
        return len(self._confirmed)

    @property
    def pending(self) -> bool:
        return self.confirmed_chunks < self.chunk_count

    @property
    def adcp_version(self) -> str:
        """The immutable request version, including for completed older plans."""
        return str(json.loads(_validated_requests(self)[0])["adcp_version"])

    def request(self, ordinal: int) -> SyncReportingReceiptsRequest:
        """Return a fresh typed view of one exact persisted request body."""
        body = json.loads(_validated_requests(self)[ordinal])
        return SyncReportingReceiptsRequest.model_validate(body)

    @property
    def outcomes(self) -> tuple[ReportingReceiptOutcome, ...]:
        plan = json.loads(self._plan)
        confirmed = {
            item["reporting_receipt_id"]: item
            for chunk in self._confirmed
            for item in json.loads(chunk)
        }
        outcomes = []
        for ordinal, item in enumerate(plan["items"]):
            result = confirmed.get(item["body"]["reporting_receipt_id"])
            if result is not None:
                stored = result.get("receipt")
                outcomes.append(
                    ReportingReceiptOutcome(
                        ordinal,
                        item["kind"],
                        result["result"],
                        tuple(ReportingReceiptFailureCode(code) for code in result["errors"]),
                        canonical_json_utf8_v1(item["body"]),
                        canonical_json_utf8_v1(stored) if stored is not None else None,
                    )
                )
        return tuple(outcomes)

Exact reserved plan and its immutable, fully confirmed chunk prefix.

Construct with :func:prepare_reporting_receipt_submission(). Stores validate persisted records before returning them. Private bytes keep mutable caller models, driver results, and representations out of the retained identity.

Instance variables

prop adcp_version : str

Expand source code

@property
def adcp_version(self) -> str:
    """The immutable request version, including for completed older plans."""
    return str(json.loads(_validated_requests(self)[0])["adcp_version"])

The immutable request version, including for completed older plans.

prop chunk_count : int

Expand source code

@property
def chunk_count(self) -> int:
    return len(_validated_requests(self))

prop confirmed_chunks : int

Expand source code

@property
def confirmed_chunks(self) -> int:
    return len(self._confirmed)

prop outcomes : tuple[ReportingReceiptOutcome, ...]

Expand source code

@property
def outcomes(self) -> tuple[ReportingReceiptOutcome, ...]:
    plan = json.loads(self._plan)
    confirmed = {
        item["reporting_receipt_id"]: item
        for chunk in self._confirmed
        for item in json.loads(chunk)
    }
    outcomes = []
    for ordinal, item in enumerate(plan["items"]):
        result = confirmed.get(item["body"]["reporting_receipt_id"])
        if result is not None:
            stored = result.get("receipt")
            outcomes.append(
                ReportingReceiptOutcome(
                    ordinal,
                    item["kind"],
                    result["result"],
                    tuple(ReportingReceiptFailureCode(code) for code in result["errors"]),
                    canonical_json_utf8_v1(item["body"]),
                    canonical_json_utf8_v1(stored) if stored is not None else None,
                )
            )
    return tuple(outcomes)

prop pending : bool

Expand source code

@property
def pending(self) -> bool:
    return self.confirmed_chunks < self.chunk_count

var scope : ReportingSubmissionScope

var submission_id : str

Methods

def request(self, ordinal: int) ‑> SyncReportingReceiptsRequest

Expand source code

def request(self, ordinal: int) -> SyncReportingReceiptsRequest:
    """Return a fresh typed view of one exact persisted request body."""
    body = json.loads(_validated_requests(self)[ordinal])
    return SyncReportingReceiptsRequest.model_validate(body)

Return a fresh typed view of one exact persisted request body.

class ReportingSubmissionCode (*args, **kwds)

Expand source code

class ReportingSubmissionCode(str, Enum):
    INVALID_SCOPE = "INVALID_SUBMISSION_SCOPE"
    INVALID_PLAN = "INVALID_SUBMISSION_PLAN"
    UNAUTHORIZED = "SUBMISSION_NOT_AUTHORIZED"
    NOT_FOUND = "SUBMISSION_NOT_FOUND"
    HISTORY_CORRUPT = "SUBMISSION_HISTORY_CORRUPT"
    STORAGE_UNAVAILABLE = "SUBMISSION_STORAGE_UNAVAILABLE"
    PG_REQUIRED = "SUBMISSION_PG_REQUIRED"
    INVALID_RESPONSE = "SUBMISSION_RESPONSE_INVALID"
    TRANSPORT_UNCERTAIN = "SUBMISSION_TRANSPORT_UNCERTAIN"
    RESPONSE_UNCONFIRMED = "SUBMISSION_RESPONSE_UNCONFIRMED"

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 HISTORY_CORRUPT

var INVALID_PLAN

var INVALID_RESPONSE

var INVALID_SCOPE

var NOT_FOUND

var PG_REQUIRED

var RESPONSE_UNCONFIRMED

var STORAGE_UNAVAILABLE

var TRANSPORT_UNCERTAIN

var UNAUTHORIZED

class ReportingSubmissionError (code: ReportingSubmissionCode)

Expand source code

class ReportingSubmissionError(RuntimeError):
    """Closed code only: no provider, authentication, SQL or wire body details."""

    def __init__(self, code: ReportingSubmissionCode) -> None:
        self.code = (
            code
            if isinstance(code, ReportingSubmissionCode)
            else ReportingSubmissionCode.INVALID_PLAN
        )
        super().__init__(self.code.value)

Closed code only: no provider, authentication, SQL or wire body details.

Ancestors

• builtins.RuntimeError
• builtins.Exception
• builtins.BaseException

class ReportingSubmissionResult (submission: ReportingReceiptSubmission,
proposal_deferred: bool = False,
diagnostic: ReportingSubmissionCode | None = None)

Expand source code

@dataclass(frozen=True)
class ReportingSubmissionResult:
    """Submission outcomes only; completion does not establish reconciliation.

    ``proposal_deferred`` means an earlier pending scope reservation was resumed
    instead. Inspect its outcomes before planning any subsequent submission.
    """

    submission: ReportingReceiptSubmission
    proposal_deferred: bool = False
    diagnostic: ReportingSubmissionCode | None = None

    @property
    def pending(self) -> bool:
        return self.submission.pending

    @property
    def outcomes(self) -> tuple[ReportingReceiptOutcome, ...]:
        return self.submission.outcomes

    @property
    def submitted_receipts(self) -> tuple[ReportingSubmissionReceipt, ...]:
        return tuple(
            receipt for outcome in self.outcomes if (receipt := outcome.receipt) is not None
        )

Submission outcomes only; completion does not establish reconciliation.

proposal_deferred means an earlier pending scope reservation was resumed instead. Inspect its outcomes before planning any subsequent submission.

Instance variables

var diagnostic : ReportingSubmissionCode | None

prop outcomes : tuple[ReportingReceiptOutcome, ...]

Expand source code

@property
def outcomes(self) -> tuple[ReportingReceiptOutcome, ...]:
    return self.submission.outcomes

prop pending : bool

Expand source code

@property
def pending(self) -> bool:
    return self.submission.pending

var proposal_deferred : bool

var submission : ReportingReceiptSubmission

prop submitted_receipts : tuple[ReportingSubmissionReceipt, ...]

Expand source code

@property
def submitted_receipts(self) -> tuple[ReportingSubmissionReceipt, ...]:
    return tuple(
        receipt for outcome in self.outcomes if (receipt := outcome.receipt) is not None
    )

class ReportingSubmissionScope (seller_id: str, account_id: str, consumer_id: str)

Expand source code

@dataclass(frozen=True)
class ReportingSubmissionScope:
    """Identity resolved by a trusted authorization adapter, never from a request.

    ``seller_id`` identifies the configured seller, ``account_id`` is the
    seller-resolved account and ``consumer_id`` is the canonical authenticated
    principal. Syntax checks cannot establish provenance: the required
    authorizer must bind all three to the exact client and its current access.
    No credential or natural-key account assertion belongs in these fields.
    """

    seller_id: str = field(repr=False)
    account_id: str = field(repr=False)
    consumer_id: str = field(repr=False)

    def __post_init__(self) -> None:
        valid = False
        try:
            consumer_reference(self.seller_id)
            principal_reference(self.account_id)
            canonical_consumer(self.consumer_id)
            valid = all(value == value.strip() for value in (self.seller_id, self.account_id))
        except (ValueError, TypeError, RuntimeError):
            # Map provider validation details to the closed INVALID_SCOPE error below.
            pass
        if not valid:
            raise ReportingSubmissionError(ReportingSubmissionCode.INVALID_SCOPE)

    @property
    def canonical_identity(self) -> bytes:
        return canonical_json_utf8_v1([self.seller_id, self.account_id, self.consumer_id])

    @property
    def storage_key(self) -> str:
        # Index a fixed-size digest, not a possibly 2048-character principal.
        # Stores also compare the full identity to fail closed on collisions.
        return hashlib.sha256(self.canonical_identity).hexdigest()

Identity resolved by a trusted authorization adapter, never from a request.

seller_id identifies the configured seller, account_id is the seller-resolved account and consumer_id is the canonical authenticated principal. Syntax checks cannot establish provenance: the required authorizer must bind all three to the exact client and its current access. No credential or natural-key account assertion belongs in these fields.

Instance variables

var account_id : str

prop canonical_identity : bytes

Expand source code

@property
def canonical_identity(self) -> bytes:
    return canonical_json_utf8_v1([self.seller_id, self.account_id, self.consumer_id])

var consumer_id : str

var seller_id : str

prop storage_key : str

Expand source code

@property
def storage_key(self) -> str:
    # Index a fixed-size digest, not a possibly 2048-character principal.
    # Stores also compare the full identity to fail closed on collisions.
    return hashlib.sha256(self.canonical_identity).hexdigest()