diff --git a/src/adcp/reporting/_reconcile.py b/src/adcp/reporting/_reconcile.py index 7cfc90c48..23bf6ff77 100644 --- a/src/adcp/reporting/_reconcile.py +++ b/src/adcp/reporting/_reconcile.py @@ -9,13 +9,15 @@ from __future__ import annotations import asyncio +import hashlib import json +from collections import Counter, defaultdict from collections.abc import Awaitable, Callable, Iterable from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from enum import Enum from math import isfinite -from typing import TYPE_CHECKING, Any, Protocol, TypeVar +from typing import TYPE_CHECKING, Any, NoReturn, Protocol, TypeVar from uuid import uuid4 from pydantic import BaseModel @@ -180,6 +182,18 @@ class ReportingLedger: revision_ownership: dict[str, str] | None = None adjustments: list[ReportingAdjustment] = field(default_factory=list) adjustment_receipts: list[ReportingAdjustmentReceipt] = field(default_factory=list) + # Only the exhausted, non-incremental loader establishes this provenance. + # A mutable/manual ledger remains useful for diagnostics, never completeness. + _read_fingerprint: bytes | None = field(default=None, init=False, repr=False, compare=False) + + def __repr__(self) -> str: + # Resources and extension fields may contain private transport values. + return ( + f"ReportingLedger(obligations={len(self.obligations)}, " + f"revisions={len(self.revisions)}, materializations={len(self.materializations)}, " + f"receipts={len(self.receipts)}, adjustments={len(self.adjustments)}, " + f"adjustment_receipts={len(self.adjustment_receipts)})" + ) @dataclass(frozen=True) @@ -285,16 +299,48 @@ def _revision_matches_obligation( ) -_RecordT = TypeVar("_RecordT") +_RecordT = TypeVar("_RecordT", bound=BaseModel) + + +def _record_json(value: BaseModel) -> str: + # Preserve optional-field presence for immutable typed-page comparisons. + # This is NOT the received canonical bytes of adjustment evidence. + return _json(value.model_dump(mode="json", exclude_unset=True, exclude_none=False)) + + +def _ledger_fingerprint(ledger: ReportingLedger) -> bytes: + return hashlib.sha256( + _json( + [ + ledger.ledger_snapshot_id, + ledger.ledger_as_of.isoformat(), + ledger.account_id, + _record_json(ledger.scope), + ledger.revision_ownership, + *[ + [_record_json(record) for record in records] + for records in ( + ledger.obligations, + ledger.revisions, + ledger.materializations, + ledger.receipts, + ledger.consumer_statuses, + ledger.adjustments, + ledger.adjustment_receipts, + ) + ], + ] + ).encode("utf-8") + ).digest() def _add_immutable( target: dict[str, _RecordT], identifier: str, value: _RecordT, kind: str ) -> None: previous = target.get(identifier) - if previous is not None and _json(previous) != _json(value): + if previous is not None and _record_json(previous) != _record_json(value): raise ReportingReconciliationError( - "IMMUTABLE_RECORD_CHANGED", f"{kind} {identifier} changed within one ledger snapshot" + "IMMUTABLE_RECORD_CHANGED", f"{kind} changed within one ledger snapshot" ) target[identifier] = value @@ -306,8 +352,16 @@ async def load_reporting_ledger( max_snapshot_restarts: int = 2, max_pages: int = 2048, max_records: int = 200_000, + max_bytes: int = 64 * 1024 * 1024, ) -> ReportingLedger: - """Exhaust a stable periods cursor and verify its declared record count.""" + """Load a complete authenticated periods snapshot, never an incremental delta. + + Restart at the first page, retaining scope and the requested page size. + Incremental, health, finality and exact-revision result selectors cannot prove + completeness and are removed. Page, received-row (including replays), and + serialized typed-page byte limits bound the walk. Transport adapters must + also bound raw responses. + """ if ( type(max_snapshot_restarts) is not int or max_snapshot_restarts < 0 @@ -315,11 +369,29 @@ async def load_reporting_ledger( or max_pages < 1 or type(max_records) is not int or max_records < 1 + or type(max_bytes) is not int + or max_bytes < 1 ): raise ValueError("reporting walk bounds must be positive (restarts may be zero)") - base = request.model_dump(mode="json", exclude_none=True) - base["view"] = "periods" - base.pop("pagination", None) + prepared = None + try: + base = request.model_dump(mode="json", exclude_none=True, warnings="error") + base["view"] = "periods" + base.pop("pagination", None) + for selector in ("changes_after", "reporting_revision_id", "health", "finality"): + base.pop(selector, None) + page_size = request.pagination.max_results if request.pagination else None + requested_account = request.account.model_dump(mode="json", warnings="error").get( + "account_id" + ) + prepared = (base, page_size, requested_account) + except Exception: + prepared = None + if prepared is None: + raise ReportingReconciliationError( + "INVALID_STATUS_REQUEST", "reporting status request could not be constructed" + ) + base, page_size, requested_account = prepared for restart in range(max_snapshot_restarts + 1): try: obligations: dict[str, ReportingObligation] = {} @@ -339,20 +411,70 @@ async def load_reporting_ledger( account_id: str | None = None scope: BaseModel | None = None total_count: int | None = None + received_records = 0 + received_bytes = 0 for _page_number in range(max_pages): payload = dict(base) + pagination_request: dict[str, object] = {} + if page_size is not None: + pagination_request["max_results"] = page_size if cursor: - payload["pagination"] = {"cursor": cursor} - result = await client.get_reporting_status( - GetReportingStatusRequest.model_validate(payload) - ) + pagination_request["cursor"] = cursor + if pagination_request: + payload["pagination"] = pagination_request + page_request = None + try: + page_request = GetReportingStatusRequest.model_validate(payload) + except Exception: + page_request = None + if page_request is None: + # SDK-side construction and provider failures are distinct, + # but neither may retain validation inputs or private context. + raise ReportingReconciliationError( + "INVALID_STATUS_REQUEST", + "reporting status request could not be constructed", + ) + result = None + try: + result = await client.get_reporting_status(page_request) + except Exception: + result = None + if result is None: + # Leave the exception scope: do not retain a private provider + # exception or Pydantic input in __context__ either. + raise ReportingReconciliationError( + "STATUS_READ_FAILED", "get_reporting_status could not be read" + ) response = result.data - if not result.success or response is None or _enum(response.view) != "periods": + if ( + not result.success + or response is None + or _enum(response.view) != "periods" + or _enum(response.status) != "completed" + ): raise ReportingReconciliationError( "STATUS_READ_FAILED", "get_reporting_status did not return a completed periods view", ) + received_bytes += len(response.model_dump_json(exclude_unset=True).encode("utf-8")) + received_records += sum( + len(records or []) + for records in ( + response.periods, + response.revisions, + response.materializations, + response.receipts, + response.consumer_statuses, + response.adjustments, + response.adjustment_receipts, + ) + ) + if received_bytes > max_bytes or received_records > max_records: + raise ReportingReconciliationError( + "LEDGER_LIMIT_EXCEEDED", "ledger read budget exceeded" + ) + response = response.model_copy(deep=True) raw_page = response.model_dump(mode="json", exclude_none=True) if "ext" in response.model_fields_set and response.ext is None: raw_page["ext"] = None @@ -377,10 +499,18 @@ async def load_reporting_ledger( metadata = _json( { k: raw_page.get(k) - for k in ("changes_checkpoint", "next_expected_at", "health", "issues") + for k in ( + "changes_checkpoint", + "next_expected_at", + "health", + "issues", + "obligation_counts", + "coverage", + "data_through", + ) } ) - if mode and frozen_metadata is not None and metadata != frozen_metadata: + if frozen_metadata is not None and metadata != frozen_metadata: raise ReportingReconciliationError( "SNAPSHOT_CHANGED", "frozen projection changed" ) @@ -392,10 +522,16 @@ async def load_reporting_ledger( or not response.account_id or not response.scope or pagination is None + or pagination.total_count is None + or type(pagination.has_more) is not bool ): raise ReportingReconciliationError( "INCOMPLETE_LEDGER_PAGE", "get_reporting_status omitted ledger metadata" ) + if requested_account is not None and response.account_id != requested_account: + raise ReportingReconciliationError( + "LEDGER_SCOPE_MISMATCH", "ledger does not match the requested account" + ) if snapshot_id and snapshot_id != response.ledger_snapshot_id: raise ReportingReconciliationError("SNAPSHOT_CHANGED", "snapshot changed") if ledger_as_of and ledger_as_of != response.ledger_as_of: @@ -414,7 +550,7 @@ async def load_reporting_ledger( account_id = response.account_id scope = response.scope total_count = pagination.total_count - if total_count is not None and total_count > max_records: + if total_count > max_records: raise ReportingReconciliationError( "LEDGER_LIMIT_EXCEEDED", "ledger record limit exceeded" ) @@ -495,7 +631,7 @@ async def load_reporting_ledger( + len(adjustments) + len(adjustment_receipts) ) - if total_count is not None and total_count != count: + if total_count != count: raise ReportingReconciliationError( "LEDGER_COUNT_MISMATCH", f"ledger declared {total_count} records but returned {count}", @@ -518,8 +654,9 @@ async def load_reporting_ledger( list(adjustments.values()), list(adjustment_receipts.values()), ) - if ownership_mode: - _validate_owned_ledger(ledger) + _, partition_complete, _ = _validate_read_ledger(ledger) + if partition_complete: + ledger._read_fingerprint = _ledger_fingerprint(ledger) return ledger except ReportingReconciliationError as error: if error.code != "SNAPSHOT_CHANGED" or restart == max_snapshot_restarts: @@ -533,14 +670,13 @@ def _validate_owned_ledger(ledger: ReportingLedger) -> None: revisions = {r.reporting_revision_id: r for r in ledger.revisions} bindings = ledger.revision_ownership - def invalid() -> None: + def invalid() -> NoReturn: raise ReportingReconciliationError( "INVALID_REVISION_OWNERSHIP", "incomplete or inconsistent ownership dependencies" ) if bindings is None or set(bindings) != set(revisions): invalid() - assert bindings is not None if any(o.account_id != ledger.account_id for o in owners.values()): invalid() for revision_id, owner_id in bindings.items(): @@ -551,11 +687,9 @@ def invalid() -> None: predecessor = revisions[revision_id].supersedes_reporting_revision_id if predecessor is not None and bindings.get(predecessor) != owner_id: invalid() + counts = Counter(bindings.values()) for owner in owners.values(): - if ( - sum(o == owner.reporting_obligation_id for o in bindings.values()) - != owner.revision_count - ): + if counts[owner.reporting_obligation_id] != owner.revision_count: invalid() materials = {m.reporting_materialization_id: m for m in ledger.materializations} owned_evidence: tuple[ReportingMaterialization | ReportingReceipt, ...] = ( @@ -587,6 +721,449 @@ def invalid() -> None: invalid() +def _status_matches_obligation(status: Any, obligation: ReportingObligation) -> bool: + # Logical compatibility alone is never an ownership proof: statuses have + # no account, campaign-set or reporting-profile fields to disambiguate it. + return bool( + status.reporting_obligation_id in {None, obligation.reporting_obligation_id} + and _status_scope(status) == _status_scope(obligation) + ) + + +_Receipt = ReportingReceipt | ReportingAdjustmentReceipt +_ReceiptTarget = tuple[str, str, str] + + +def _receipt_target(receipt: _Receipt) -> _ReceiptTarget: + if isinstance(receipt, ReportingReceipt): + return ("revision", receipt.reporting_obligation_id, receipt.reporting_revision_id) + return ("adjustment", receipt.reporting_adjustment_id, receipt.adjusts_reporting_revision_id) + + +def _receipt_leaves(ledger: ReportingLedger) -> dict[_ReceiptTarget, _Receipt]: + """Linear, order-independent validation of every exact receipt chain.""" + records: list[_Receipt] = [*ledger.receipts, *ledger.adjustment_receipts] + by_id = {r.reporting_receipt_id: r for r in records} + + def invalid() -> NoReturn: + raise ReportingReconciliationError( + "INVALID_RECEIPT_CHAIN", "receipt history is not one exact predecessor chain" + ) + + if len(by_id) != len(records): + invalid() + roots: dict[_ReceiptTarget, list[str]] = {} + counts: Counter[_ReceiptTarget] = Counter() + successors: dict[str, str] = {} + for receipt in records: + key = _receipt_target(receipt) + counts[key] += 1 + roots.setdefault(key, []) + if (_enum(receipt.status) == "rejected") != bool(receipt.rejection_codes): + invalid() + predecessor_id = receipt.supersedes_reporting_receipt_id + if predecessor_id is None: + roots[key].append(receipt.reporting_receipt_id) + continue + predecessor = by_id.get(predecessor_id) + if ( + predecessor is None + or _receipt_target(predecessor) != key + or _enum(predecessor.status) != "rejected" + or predecessor_id in successors + ): + invalid() + successors[predecessor_id] = receipt.reporting_receipt_id + leaves: dict[_ReceiptTarget, _Receipt] = {} + for key, chain_roots in roots.items(): + if len(chain_roots) != 1: + invalid() + identifier = chain_roots[0] + visited: set[str] = set() + while identifier not in visited: + visited.add(identifier) + successor = successors.get(identifier) + if successor is None: + break + identifier = successor + if len(visited) != counts[key] or identifier in successors: + invalid() + leaves[key] = by_id[identifier] + return leaves + + +def _ordered_times(*values: datetime | None) -> bool: + if any(v is None or v.tzinfo is None or v.utcoffset() is None for v in values): + return False + return all(a <= b for a, b in zip(values, values[1:]) if a is not None and b is not None) + + +def _status_scope(item: Any) -> tuple[str, int, str, str]: + return ( + item.delivery_config_id, + item.delivery_config_version, + item.report_definition_id, + _json(item.period), + ) + + +def _revision_scope( + item: ReportingRevision | ReportingObligation, +) -> tuple[str, str, str, tuple[str, ...], str]: + return ( + item.account_id, + item.report_definition_id, + item.reporting_profile, + _identifiers(item.media_buy_ids), + _json(item.period), + ) + + +@dataclass +class _ObligationHistory: + revisions: list[ReportingRevision] = field(default_factory=list) + known_revision_ids: set[str] = field(default_factory=set) + materializations: list[ReportingMaterialization] = field(default_factory=list) + receipts: list[ReportingReceipt] = field(default_factory=list) + adjustments: list[ReportingAdjustment] = field(default_factory=list) + adjustment_receipts: list[ReportingAdjustmentReceipt] = field(default_factory=list) + statuses: list[Any] = field(default_factory=list) + ambiguous_revisions: bool = False + unresolved_status_count: int = 0 + + +@dataclass +class _ReadIndex: + owners: dict[str, ReportingObligation] + revisions: dict[str, ReportingRevision] + materials: dict[str, ReportingMaterialization] + adjustments: dict[str, ReportingAdjustment] + histories: dict[str, _ObligationHistory] + adjustments_by_revision: dict[str, list[ReportingAdjustment]] + unresolved_statuses: bool + selections: dict[ + str, tuple[ReportingRevision | None, ReportingMaterialization | None, list[str]] + ] = field(default_factory=dict) + + +def _index_read_ledger(ledger: ReportingLedger) -> _ReadIndex: + """Index the complete read, checking identity before partitioning evidence. + + Indexes are private to this validation, never cached on mutable ledgers. + Legacy ambiguous associations get a separate work bound so sharing one + semantic scope cannot expand a bounded record set into quadratic work. + """ + if ledger.revision_ownership is not None: + _validate_owned_ledger(ledger) + owners = {o.reporting_obligation_id: o for o in ledger.obligations} + revisions = {r.reporting_revision_id: r for r in ledger.revisions} + materials = {m.reporting_materialization_id: m for m in ledger.materializations} + adjustments = {a.reporting_adjustment_id: a for a in ledger.adjustments} + statuses = {s.reporting_status_id: s for s in ledger.consumer_statuses} + + def invalid() -> NoReturn: + raise ReportingReconciliationError( + "INVALID_LEDGER_DEPENDENCY", "ledger dependencies are incomplete or inconsistent" + ) + + if any( + len(index) != len(records) + for index, records in ( + (owners, ledger.obligations), + (revisions, ledger.revisions), + (materials, ledger.materializations), + (adjustments, ledger.adjustments), + (statuses, ledger.consumer_statuses), + ) + ): + invalid() + if any(o.account_id != ledger.account_id for o in owners.values()) or any( + r.account_id != ledger.account_id for r in revisions.values() + ): + invalid() + histories = {owner_id: _ObligationHistory() for owner_id in owners} + revision_owners = dict(ledger.revision_ownership or {}) + for material in materials.values(): + owner = owners.get(material.reporting_obligation_id) + revision = revisions.get(material.reporting_revision_id) + if ( + owner is None + or revision is None + or not _revision_matches_obligation(revision, owner) + or material.delivery_config_id != owner.delivery_config_id + or material.delivery_config_version != owner.delivery_config_version + or material.destination_ref != owner.destination_ref + or material.feed_purpose != owner.feed_purpose + or revision_owners.setdefault( + material.reporting_revision_id, material.reporting_obligation_id + ) + != material.reporting_obligation_id + ): + invalid() + histories[owner.reporting_obligation_id].materializations.append(material) + owners_by_scope: dict[tuple[str, str, str, tuple[str, ...], str], list[str]] = defaultdict(list) + for owner_id, owner in owners.items(): + owners_by_scope[_revision_scope(owner)].append(owner_id) + association_count = 0 + association_limit = max( + 200_000, + sum( + len(records) + for records in ( + ledger.obligations, + ledger.revisions, + ledger.materializations, + ledger.receipts, + ledger.adjustments, + ledger.adjustment_receipts, + ledger.consumer_statuses, + ) + ), + ) + + def account_associations(count: int) -> None: + nonlocal association_count + association_count += count + if association_count > association_limit: + raise ReportingReconciliationError( + "LEDGER_LIMIT_EXCEEDED", "legacy history association budget exceeded" + ) + + for revision_id, revision in revisions.items(): + predecessor = revision.supersedes_reporting_revision_id + if predecessor is not None and predecessor not in revisions: + invalid() + exact_owner = revision_owners.get(revision_id) + matching = ( + [exact_owner] + if exact_owner is not None + else owners_by_scope.get(_revision_scope(revision), []) + ) + if not matching: + invalid() + account_associations(len(matching)) + for owner_id in matching: + histories[owner_id].revisions.append(revision) + histories[owner_id].ambiguous_revisions |= len(matching) > 1 + if len(matching) == 1: + histories[owner_id].known_revision_ids.add(revision_id) + for receipt in ledger.receipts: + owner = owners.get(receipt.reporting_obligation_id) + revision = revisions.get(receipt.reporting_revision_id) + referenced_material = materials.get(receipt.reporting_materialization_id) + if ( + owner is None + or revision is None + or referenced_material is None + or referenced_material.reporting_revision_id != receipt.reporting_revision_id + or referenced_material.reporting_obligation_id != receipt.reporting_obligation_id + or not _revision_matches_obligation(revision, owner) + or _enum(owner.reconciliation_mode) != "consumer_receipt" + ): + invalid() + histories[owner.reporting_obligation_id].receipts.append(receipt) + adjustments_by_revision: dict[str, list[ReportingAdjustment]] = defaultdict(list) + receipts_by_adjustment: dict[str, list[ReportingAdjustmentReceipt]] = defaultdict(list) + for adjustment in adjustments.values(): + if adjustment.adjusts_reporting_revision_id not in revisions: + invalid() + adjustments_by_revision[adjustment.adjusts_reporting_revision_id].append(adjustment) + for adjustment_receipt in ledger.adjustment_receipts: + target = adjustments.get(adjustment_receipt.reporting_adjustment_id) + if target is None or target.adjusts_reporting_revision_id != ( + adjustment_receipt.adjusts_reporting_revision_id + ): + invalid() + receipts_by_adjustment[target.reporting_adjustment_id].append(adjustment_receipt) + for history in histories.values(): + for revision in history.revisions: + related = adjustments_by_revision.get(revision.reporting_revision_id, []) + account_associations(len(related)) + history.adjustments.extend(related) + for adjustment in history.adjustments: + related_receipts = receipts_by_adjustment.get(adjustment.reporting_adjustment_id, []) + account_associations(len(related_receipts)) + history.adjustment_receipts.extend(related_receipts) + + status_owners: dict[str, str | None] = {} + unresolved_by_scope: Counter[tuple[str, int, str, str]] = Counter() + for status in statuses.values(): + status_revision_id = status.reporting_revision_id + status_owner: str | None = status.reporting_obligation_id + if status_owner is None and status_revision_id is not None: + status_owner = revision_owners.get(status_revision_id) + if status_owner is None: + # Neither a logical-key match nor a seller's projected count can + # invent missing immutable ownership. Keep these rows diagnostic. + unresolved_by_scope[_status_scope(status)] += 1 + elif status_owner not in owners or not _status_matches_obligation( + status, owners[status_owner] + ): + invalid() + if status_revision_id is not None and ( + status_revision_id not in revisions + or ( + status_owner is not None + and ( + not _revision_matches_obligation( + revisions[status_revision_id], owners[status_owner] + ) + or revision_owners.get(status_revision_id, status_owner) != status_owner + ) + ) + ): + invalid() + status_owners[status.reporting_status_id] = status_owner + if status_owner is not None: + histories[status_owner].statuses.append(status) + for status in statuses.values(): + predecessor_id = status.supersedes_reporting_status_id + if predecessor_id is None: + continue + predecessor = statuses.get(predecessor_id) + if predecessor is None or _status_scope(status) != _status_scope(predecessor): + invalid() + status_owner = status_owners[status.reporting_status_id] + predecessor_owner = status_owners[predecessor_id] + if ( + status_owner is not None + and predecessor_owner is not None + and status_owner != predecessor_owner + ): + invalid() + for owner_id, owner in owners.items(): + histories[owner_id].unresolved_status_count = unresolved_by_scope[_status_scope(owner)] + current_id = owner.current_consumer_status_id + if current_id is not None: + current = statuses.get(current_id) + if ( + current is None + or not _status_matches_obligation(current, owner) + or status_owners[current_id] not in {None, owner_id} + or owner.consumer_status_count == 0 + ): + invalid() + elif owner.consumer_status_count: + invalid() + return _ReadIndex( + owners, + revisions, + materials, + adjustments, + histories, + adjustments_by_revision, + bool(unresolved_by_scope), + ) + + +def _validate_read_ledger( + ledger: ReportingLedger, +) -> tuple[dict[_ReceiptTarget, _Receipt], bool, _ReadIndex]: + """Check the full frozen denominator and dependencies before using evidence.""" + index = _index_read_ledger(ledger) + owners, revisions = index.owners, index.revisions + materials, adjustments = index.materials, index.adjustments + partition_complete = not index.unresolved_statuses + + def invalid() -> NoReturn: + raise ReportingReconciliationError( + "INVALID_LEDGER_DEPENDENCY", "ledger dependencies are incomplete or inconsistent" + ) + + for owner_id, owner in owners.items(): + selection = _select_current(owner, ledger, index) + index.selections[owner_id] = selection + if any( + reason in selection[2] + for reason in ("ASSOCIATED_HISTORY_INCOMPLETE", "AMBIGUOUS_REVISION_OWNERSHIP") + ): + partition_complete = False + for adjustment in adjustments.values(): + revision = revisions.get(adjustment.adjusts_reporting_revision_id) + if ( + revision is None + or _enum(revision.finality) != "official" + or not revision.finality_basis + or not revision.finality_policy_id + or not _ordered_times(revision.period.end, revision.finalized_at, revision.created_at) + or not _ordered_times( + revision.finalized_at, + adjustment.correction_observed_at, + adjustment.created_at, + ledger.ledger_as_of, + ) + or not _ordered_times( + adjustment.accounting_period.start, adjustment.accounting_period.end + ) + or adjustment.accounting_period.start == adjustment.accounting_period.end + ): + invalid() + for receipt_adjustment in ledger.adjustment_receipts: + target = adjustments.get(receipt_adjustment.reporting_adjustment_id) + if ( + target is None + or target.adjusts_reporting_revision_id + != receipt_adjustment.adjusts_reporting_revision_id + or not _ordered_times( + target.created_at, receipt_adjustment.observed_at, ledger.ledger_as_of + ) + or ( + receipt_adjustment.received_at is not None + and not _ordered_times( + receipt_adjustment.observed_at, + receipt_adjustment.received_at, + ledger.ledger_as_of, + ) + ) + ): + invalid() + if _enum(receipt_adjustment.status) == "accepted" and ( + not target.canonical_adjustment_sha256 + or target.canonical_adjustment_sha256 != receipt_adjustment.observed_adjustment_sha256 + ): + raise ReportingReconciliationError( + "INVALID_RECEIPT_EVIDENCE", "accepted adjustment evidence does not match its target" + ) + leaves = _receipt_leaves(ledger) + for owner in owners.values(): + if owner.pending_adjustment_count is None: + continue + selected, _, _ = index.selections[owner.reporting_obligation_id] + if selected is None: + continue + pending = sum( + ( + leaf := leaves.get( + ("adjustment", a.reporting_adjustment_id, selected.reporting_revision_id) + ) + ) + is None + or _enum(leaf.status) != "accepted" + for a in index.adjustments_by_revision.get(selected.reporting_revision_id, []) + ) + # A legacy candidate without a proven owner can contribute zero through + # all of its pending adjustments. It cannot contribute more than exist. + history = index.histories[owner.reporting_obligation_id] + minimum = pending if selected.reporting_revision_id in history.known_revision_ids else 0 + if not minimum <= owner.pending_adjustment_count <= pending: + raise ReportingReconciliationError( + "LEDGER_COUNT_MISMATCH", "frozen adjustment counts do not match current leaves" + ) + for receipt in ledger.receipts: + if _enum(receipt.status) != "accepted": + continue + owner = owners[receipt.reporting_obligation_id] + revision = revisions[receipt.reporting_revision_id] + material = materials[receipt.reporting_materialization_id] + if _materialization_reasons(owner, revision, material) or not _receipt_matches( + receipt, revision, material + ): + raise ReportingReconciliationError( + "INVALID_RECEIPT_EVIDENCE", "accepted receipt evidence does not match its target" + ) + return leaves, partition_complete, index + + def _owned_revisions( obligation: ReportingObligation, ledger: ReportingLedger ) -> list[ReportingRevision]: @@ -600,87 +1177,101 @@ def _owned_revisions( return [r for r in ledger.revisions if _revision_matches_obligation(r, obligation)] -def _select_current( - obligation: ReportingObligation, ledger: ReportingLedger -) -> tuple[ReportingRevision | None, ReportingMaterialization | None, list[str]]: +def _obligation_history( + obligation: ReportingObligation, index: _ReadIndex +) -> tuple[list[ReportingRevision], list[ReportingMaterialization], list[str]]: + history = index.histories[obligation.reporting_obligation_id] reasons: list[str] = [] - attempts = [ - item - for item in ledger.materializations - if item.reporting_obligation_id == obligation.reporting_obligation_id - ] - revision_ids = {item.reporting_revision_id for item in attempts} - managed_delivery = obligation.destination_ref is not None - # ``ReportingRevision`` carries no obligation reference, so semantic scope - # alone cannot separate two obligations that legitimately share a definition, - # profile, campaign set and period. A revision some *other* obligation has - # materialized is that obligation's publication. Anything this obligation also - # materialized stays a candidate so illegal fan-out is reported below rather - # than silently narrowed away. Revisions no obligation has materialized -- an - # unmaterialized official included -- are never excluded here. - owned_elsewhere = { - item.reporting_revision_id - for item in ledger.materializations - if item.reporting_obligation_id != obligation.reporting_obligation_id - } - revision_ids - candidates = [ - item - for item in ledger.revisions - if ( - _revision_matches_obligation(item, obligation) - or (managed_delivery and item.reporting_revision_id in revision_ids) - ) - and item.reporting_revision_id not in owned_elsewhere - ] - if ledger.revision_ownership is not None: - candidates = _owned_revisions(obligation, ledger) - elif any( - not any( - m.reporting_revision_id == item.reporting_revision_id for m in ledger.materializations - ) - and sum(_revision_matches_obligation(item, o) for o in ledger.obligations) > 1 - for item in candidates - ): - # Counts and equal semantic scopes are not an ownership declaration. - # An unmaterialized revision may still belong to either obligation. + incomplete = False + if history.ambiguous_revisions: reasons.append("AMBIGUOUS_REVISION_OWNERSHIP") - receipts = [ - item - for item in ledger.receipts - if item.reporting_obligation_id == obligation.reporting_obligation_id - ] - successful_attempts = [ - item for item in attempts if _enum(item.status) in {"available", "delivered"} - ] - accepted_receipts = [item for item in receipts if _enum(item.status) == "accepted"] - history_incomplete = len(candidates) != obligation.revision_count - if managed_delivery: - history_incomplete = history_incomplete or ( - obligation.materialization_count is None - or len(attempts) != obligation.materialization_count - or obligation.successful_materialization_count is None - or len(successful_attempts) != obligation.successful_materialization_count - ) - elif attempts: - history_incomplete = True - if obligation.receipt_count is not None: - history_incomplete = history_incomplete or len(receipts) != obligation.receipt_count - if obligation.accepted_receipt_count is not None: - history_incomplete = ( - history_incomplete or len(accepted_receipts) != obligation.accepted_receipt_count + managed = obligation.destination_ref is not None + reconciled = _enum(obligation.reconciliation_mode) == "consumer_receipt" + # Optional fields remain readable in legacy shapes. Their absence prevents + # proof where applicable; only a present, provably false count is an error. + # Applicability never depends on the unrelated revision-ownership extension. + successful = sum( + _enum(m.status) in {"available", "delivered"} for m in history.materializations + ) + accepted = sum(_enum(r.status) == "accepted" for r in history.receipts) + accepted_adjustments = [r for r in history.adjustment_receipts if _enum(r.status) == "accepted"] + # Unknown legacy owners widen a count's possible range, without excusing + # counts below known records or above the complete set of candidates. + counts = ( + (obligation.revision_count, len(history.known_revision_ids), len(history.revisions), True), + ( + obligation.materialization_count, + len(history.materializations), + len(history.materializations), + managed, + ), + ( + obligation.successful_materialization_count, + successful, + successful, + managed, + ), + (obligation.receipt_count, len(history.receipts), len(history.receipts), reconciled), + (obligation.accepted_receipt_count, accepted, accepted, reconciled), + # Reliable Reporting requires this count; its optional schema shape also + # admits older sellers, which remain diagnostic until evidence is complete. + ( + obligation.adjustment_count, + sum( + a.adjusts_reporting_revision_id in history.known_revision_ids + for a in history.adjustments + ), + len(history.adjustments), + True, + ), + ( + obligation.adjustment_receipt_count, + sum( + r.adjusts_reporting_revision_id in history.known_revision_ids + for r in history.adjustment_receipts + ), + len(history.adjustment_receipts), + reconciled, + ), + ( + obligation.accepted_adjustment_receipt_count, + sum( + r.adjusts_reporting_revision_id in history.known_revision_ids + for r in accepted_adjustments + ), + len(accepted_adjustments), + reconciled, + ), + ) + for declared, minimum, maximum, required in counts: + if declared is None: + incomplete |= required + elif not minimum <= declared <= maximum: + raise ReportingReconciliationError( + "LEDGER_COUNT_MISMATCH", "frozen obligation counts do not match the ledger" + ) + elif declared != maximum: + incomplete = True + known = len(history.statuses) + declared = obligation.consumer_status_count + if declared is not None and not known <= declared <= known + history.unresolved_status_count: + raise ReportingReconciliationError( + "LEDGER_COUNT_MISMATCH", "frozen consumer status counts do not match the ledger" ) - if _enum(obligation.reconciliation_mode) == "consumer_receipt" and ( - obligation.receipt_count is None or obligation.accepted_receipt_count is None - ): - history_incomplete = True - if history_incomplete: + # A disabled status projection does not require an optional count. A current + # status reference without its count cannot prove the projected history. + incomplete |= declared is None and obligation.current_consumer_status_id is not None + if history.unresolved_status_count: + reasons.append("AMBIGUOUS_CONSUMER_STATUS_OWNERSHIP") + if incomplete: reasons.append("ASSOCIATED_HISTORY_INCOMPLETE") - if any(not _revision_matches_obligation(item, obligation) for item in candidates) or any( - item.reporting_revision_id in revision_ids - and item.reporting_obligation_id != obligation.reporting_obligation_id - for item in ledger.materializations - ): - reasons.append("REVISION_SCOPE_MISMATCH") + return history.revisions, history.materializations, reasons + + +def _select_current( + obligation: ReportingObligation, ledger: ReportingLedger, index: _ReadIndex +) -> tuple[ReportingRevision | None, ReportingMaterialization | None, list[str]]: + candidates, attempts, reasons = _obligation_history(obligation, index) selection = select_reporting_revision( tuple( RevisionHistoryEntry( @@ -739,7 +1330,7 @@ def _select_current( elif finality_basis is not None or finality_policy_id is not None or finalized_at is not None: reasons.append("FINALITY_EVIDENCE_INVALID") - if not managed_delivery: + if obligation.destination_ref is None: if _enum(obligation.reconciliation_mode) == "consumer_receipt": reasons.append("INVALID_RECONCILIATION_TIER") return revision, None, reasons @@ -747,23 +1338,38 @@ def _select_current( successful = sorted( ( item - for item in successful_attempts + for item in attempts if item.reporting_revision_id == revision.reporting_revision_id + and _enum(item.status) in {"available", "delivered"} ), key=lambda item: item.attempt, reverse=True, ) materialization = successful[0] if successful else None + reasons.extend(_materialization_reasons(obligation, revision, materialization)) + return revision, materialization, reasons + + +def _materialization_reasons( + obligation: ReportingObligation, + revision: ReportingRevision, + materialization: ReportingMaterialization | None, +) -> list[str]: + """Immutable producer evidence; current readability/expiry is independent.""" + reasons: list[str] = [] if ( not materialization + or _enum(materialization.status) not in {"available", "delivered"} or not materialization.ready_at or not materialization.verification or not materialization.resource ): reasons.append("MISSING_VERIFIED_MATERIALIZATION") - return revision, materialization, reasons + return reasons if ( - materialization.delivery_config_id != obligation.delivery_config_id + materialization.reporting_obligation_id != obligation.reporting_obligation_id + or materialization.reporting_revision_id != revision.reporting_revision_id + or materialization.delivery_config_id != obligation.delivery_config_id or materialization.delivery_config_version != obligation.delivery_config_version or materialization.destination_ref != obligation.destination_ref or _enum(materialization.feed_purpose) != _enum(obligation.feed_purpose) @@ -826,7 +1432,7 @@ def _select_current( or not materialization.verification.physical_checksums ): reasons.append("PRODUCER_MANIFEST_EVIDENCE_MISSING") - return revision, materialization, reasons + return reasons def _receipt_matches( @@ -940,26 +1546,65 @@ def build_reporting_receipt( return ReportingReceipt.model_validate(payload) +def _expected_in_scope( + expected: ExpectedReportingPeriod, + scope: BaseModel, + generations: set[tuple[str, int, str]], + media_buy_ids: set[str], +) -> bool: + """Missing obligations can only be diagnosed inside the retained denominator.""" + start = datetime.fromisoformat(expected.period_start.replace("Z", "+00:00")) + end = datetime.fromisoformat(expected.period_end.replace("Z", "+00:00")) + return bool( + getattr(scope, "coverage_complete", False) + and _ordered_times( + getattr(scope, "period_start", None), start, end, getattr(scope, "period_end", None) + ) + and _ordered_times(getattr(scope, "ledger_retained_from", None), start) + and start < end + and ( + expected.delivery_config_id, + expected.delivery_config_version, + expected.feed_purpose, + ) + in generations + and ( + getattr(scope, "all_accessible_media_buys", False) + or set(expected.media_buy_ids) <= media_buy_ids + ) + ) + + def evaluate_reporting_ledger( ledger: ReportingLedger, *, expected_periods: list[ExpectedReportingPeriod] | None = None, now: datetime | None = None, ) -> ReportingReconciliationResult: - if ledger.revision_ownership is not None: - _validate_owned_ledger(ledger) + """Evaluate exact retained evidence without performing reads or writes. + + Definitive and missing-period claims require an unchanged full snapshot from + :func:`load_reporting_ledger`. Manually assembled ledgers are diagnostic only; + this API does not certify a caller's incremental merge or recompute raw + adjustment digests from normalized models. + """ + complete_read = ledger._read_fingerprint is not None + if complete_read and ledger._read_fingerprint != _ledger_fingerprint(ledger): + raise ReportingReconciliationError( + "LEDGER_CHANGED", "the completed ledger was changed after loading" + ) + leaves, partition_complete, index = _validate_read_ledger(ledger) + complete_read = complete_read and partition_complete now = now or datetime.now(timezone.utc) outcomes: list[ObligationReconciliation] = [] unique_revisions: dict[str, ReportingRevision] = {} for obligation in ledger.obligations: - revision, materialization, reasons = _select_current(obligation, ledger) - if obligation.adjustment_count or any( - a.adjusts_reporting_revision_id == getattr(revision, "reporting_revision_id", None) - for a in ledger.adjustments - ): - # Loading ownership/dependencies is additive. The separately owned - # buyer adjustment evidence/submission workflow remains required. - reasons.append("ADJUSTMENT_RECONCILIATION_REQUIRED") + revision, materialization, selected_reasons = index.selections[ + obligation.reporting_obligation_id + ] + reasons = list(selected_reasons) + if not complete_read: + reasons.append("UNVERIFIED_LEDGER_SNAPSHOT") if _enum(obligation.health) != "complete": reasons.append(f"OBLIGATION_{_enum(obligation.health).upper()}") if ( @@ -970,15 +1615,30 @@ def evaluate_reporting_ledger( reasons.append("RESOURCE_EXPIRED") if revision: unique_revisions[revision.reporting_revision_id] = revision - if ( - _enum(obligation.reconciliation_mode) == "consumer_receipt" - and revision - and materialization - and not any( - _receipt_matches(receipt, revision, materialization) for receipt in ledger.receipts + if _enum(obligation.reconciliation_mode) == "consumer_receipt" and revision: + if _enum(revision.finality) != "official" and "FINALITY_NOT_MET" not in reasons: + reasons.append("FINALITY_NOT_MET") + receipt = leaves.get( + ("revision", obligation.reporting_obligation_id, revision.reporting_revision_id) ) - ): - reasons.append("MISSING_MATCHING_CONSUMER_RECEIPT") + # Validation binds this leaf to the materialization it names, not + # the newest artifact selected independently for current readability. + if receipt is None or _enum(receipt.status) != "accepted": + reasons.append("MISSING_MATCHING_CONSUMER_RECEIPT") + if revision: + # Applicable corrections need accepted current evidence in either mode. + applicable = index.adjustments_by_revision.get(revision.reporting_revision_id, []) + if any( + ( + leaf := leaves.get( + ("adjustment", a.reporting_adjustment_id, revision.reporting_revision_id) + ) + ) + is None + or _enum(leaf.status) != "accepted" + for a in applicable + ): + reasons.append("MISSING_MATCHING_ADJUSTMENT_RECEIPT") outcomes.append( ObligationReconciliation( obligation.reporting_obligation_id, @@ -999,29 +1659,55 @@ def evaluate_reporting_ledger( _identifiers(item.media_buy_ids), item.period.start.isoformat(), item.period.end.isoformat(), + item.period.source_timezone, ) for item in ledger.obligations } + generations = { + (g.delivery_config_id, g.delivery_config_version, _enum(g.feed_purpose)) + for g in getattr(ledger.scope, "delivery_config_generations", []) + } + media_buy_ids = set(_identifiers(getattr(ledger.scope, "media_buy_ids", []))) + expectations = [ + ( + item, + _expected_in_scope(item, ledger.scope, generations, media_buy_ids), + ( + item.delivery_config_id, + item.delivery_config_version, + item.report_definition_id, + item.feed_purpose, + item.reporting_profile, + tuple(sorted(item.media_buy_ids)), + _iso(item.period_start), + _iso(item.period_end), + item.source_timezone, + ) + in actual, + ) + for item in expected_periods or [] + ] + # The producer describes configuration requirements here, not current + # revision finality: [official] can be a complete unfiltered denominator. + # ExpectedReportingPeriod has no trusted finality fact, however, so for an + # *absent* obligation we cannot prove membership in a proper subset. This + # intentionally withholds some valid absence claims; it is not evidence of + # seller filtering. Positive returned obligations retain their exact proof. + absence_finality_proven = {_enum(value) for value in getattr(ledger.scope, "finality", [])} == { + "snapshot", + "official", + } missing = [ item - for item in expected_periods or [] - if ( - item.delivery_config_id, - item.delivery_config_version, - item.report_definition_id, - item.feed_purpose, - item.reporting_profile, - tuple(sorted(item.media_buy_ids)), - _iso(item.period_start), - _iso(item.period_end), - ) - not in actual + for item, in_scope, present in expectations + if complete_read and absence_finality_proven and in_scope and not present ] definitive = bool( - expected_periods is not None + complete_read + and expected_periods is not None + and all(in_scope and present for _, in_scope, present in expectations) and bool(getattr(ledger.scope, "scope_closed", False)) and bool(getattr(ledger.scope, "coverage_complete", False)) - and not missing and all(item.definitive for item in outcomes) ) return ReportingReconciliationResult( @@ -1191,10 +1877,11 @@ async def reconcile_reporting( "canonical-digest verification requires the reconciled_billing tier", ) submitted: list[ReportingReceipt] = [] + _, _, index = _validate_read_ledger(ledger) for obligation in ledger.obligations: if _enum(obligation.reconciliation_mode) != "consumer_receipt": continue - revision, materialization, reasons = _select_current(obligation, ledger) + revision, materialization, reasons = index.selections[obligation.reporting_obligation_id] if not revision or not materialization or reasons: continue if any(_receipt_matches(item, revision, materialization) for item in ledger.receipts): diff --git a/tests/conformance/reporting/test_reporting_buyer_frozen_read.py b/tests/conformance/reporting/test_reporting_buyer_frozen_read.py new file mode 100644 index 000000000..798d4992a --- /dev/null +++ b/tests/conformance/reporting/test_reporting_buyer_frozen_read.py @@ -0,0 +1,179 @@ +"""Buyer full reads against reviewed, authenticated in-memory seller mounts.""" + +from dataclasses import replace + +import pytest + +from adcp.reporting import ( + ExpectedReportingPeriod, + ReportingReconciliationError, + evaluate_reporting_ledger, + load_reporting_ledger, +) +from adcp.types import GetReportingStatusRequest + +from ._durable_materializer_support import durable_case +from ._feed_support import MountedFeed, feed_request, mixed_case, second_consumer +from ._projection_support import projection_harness +from ._receipt_support import adjustment_for, receipt_case +from .test_reporting_notification_outbox import statement + + +@pytest.fixture(autouse=True) +def _a2a_compat_send_and_aggregate(): + # These mounts require the real async-generator transport, not the unit mock shim. + pass + + +@pytest.mark.parametrize("backend", ["memory", "postgres"]) +@pytest.mark.parametrize("protocol", ["mcp", "a2a"]) +async def test_delivery_only_adjustment_blocks_definitive_without_receipt_counts(backend, protocol): + async with projection_harness(backend) as h: + s = await receipt_case(h, billing=False, reconciliation_mode="delivery_only") + await adjustment_for(h, s) + await h.projection.activate(account_id=s.obligation.account_id) + mounted = MountedFeed(h) + mounted.authorize(s) + request = GetReportingStatusRequest.model_validate(feed_request(s, limit=1)) + async with mounted.sdk_clients("1.0") as (clients, observed): + ledger = await load_reporting_ledger(clients[protocol], request) + obligation = ledger.obligations[0] + assert obligation.health.value == "complete" + assert obligation.reconciliation_mode.value == "delivery_only" + assert obligation.pending_adjustment_count is None + assert obligation.adjustment_receipt_count is None + assert obligation.accepted_adjustment_receipt_count is None + assert len(ledger.adjustments) == 1 + assert ledger.adjustment_receipts == [] + assert all(params["pagination"]["max_results"] == 1 for _, _, params in observed) + result = evaluate_reporting_ledger(ledger, expected_periods=[], now=h.clock()) + assert not result.definitive + assert not result.obligations[0].definitive + assert result.obligations[0].reasons == ("MISSING_MATCHING_ADJUSTMENT_RECEIPT",) + + +@pytest.mark.parametrize("feedback", [False, True]) +@pytest.mark.parametrize("protocol", ["mcp", "a2a"]) +async def test_public_loader_keeps_exact_consumer_evidence_with_maximum_url_principals( + feedback, protocol +): + prefix = "https://buyer.example/" + consumer = prefix + "a" * (2048 - len(prefix)) + async with projection_harness("memory", feedback=feedback) as h: + s, _, _ = await mixed_case(h, consumer_id=consumer) + other = await second_consumer(h, s, consumer[:-1] + "b") + # Current wire status includes its exact owner; the pure loader also + # covers historical typed status with an omitted owner separately. + own_status = replace( + statement(s.obligation), + consumer_id=consumer, + consumer_status="received", + reporting_revision_id=s.revision.reporting_revision_id, + observed_revision_content_sha256=s.revision.revision_content_sha256, + ) + await h.store.record_consumer_status_with_lifecycle(own_status) + await h.projection.activate(account_id=s.obligation.account_id) + mounted = MountedFeed(h, feedback=feedback) + mounted.authorize(s) + mounted.authorize(other, token="token-two") + request = GetReportingStatusRequest.model_validate(feed_request(s, limit=1)) + async with mounted.sdk_clients("1.0") as (clients, observed): + ledger = await load_reporting_ledger(clients[protocol], request) + assert len(ledger.adjustment_receipts) == 1 + assert len(ledger.consumer_statuses) == 1 + assert ledger.revision_ownership == { + s.revision.reporting_revision_id: s.obligation.reporting_obligation_id + } + assert all(params["pagination"]["max_results"] == 1 for _, _, params in observed) + obligation = ledger.obligations[0] + expected = ExpectedReportingPeriod( + obligation.delivery_config_id, + obligation.delivery_config_version, + obligation.report_definition_id, + obligation.feed_purpose.value, + obligation.reporting_profile, + tuple(b.root for b in obligation.media_buy_ids), + obligation.period.start.isoformat(), + obligation.period.end.isoformat(), + obligation.period.source_timezone, + ) + result = evaluate_reporting_ledger(ledger, expected_periods=[expected], now=h.clock()) + assert result.definitive, result.obligations + async with mounted.sdk_clients("1.0", token="token-two") as (clients, _): + other_ledger = await load_reporting_ledger(clients[protocol], request) + assert other_ledger.adjustment_receipts == [] + assert other_ledger.consumer_statuses == [] + result = evaluate_reporting_ledger( + other_ledger, expected_periods=[expected], now=h.clock() + ) + assert not result.definitive + assert "MISSING_MATCHING_ADJUSTMENT_RECEIPT" in result.obligations[0].reasons + assert {consumer, other.binding.consumer_id} <= { + principal for _, principal in mounted.auth_calls + } + + +@pytest.mark.parametrize("protocol", ["mcp", "a2a"]) +async def test_revocation_between_pages_and_on_replay_never_produces_a_completed_ledger(protocol): + async with projection_harness("memory") as h: + s, _, _ = await mixed_case(h, consumer_id="https://buyer.example/authorized") + await h.projection.activate(account_id=s.obligation.account_id) + mounted = MountedFeed(h) + mounted.authorize(s) + request = GetReportingStatusRequest.model_validate(feed_request(s, limit=1)) + async with mounted.sdk_clients("1.0") as (clients, _): + + class RevokeAfterFirstPage: + calls = 0 + + async def get_reporting_status(self, request): + response = await clients[protocol].get_reporting_status(request) + self.calls += 1 + if self.calls == 1: + mounted.grants.remove((s.obligation.account_id, s.binding.consumer_id)) + return response + + client = RevokeAfterFirstPage() + for _ in range(2): + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(client, request) + assert error.value.code == "STATUS_READ_FAILED" + assert s.binding.consumer_id not in str(error.value) + assert error.value.__context__ is None + + +@pytest.mark.parametrize("protocol", ["mcp", "a2a"]) +async def test_official_configuration_scope_keeps_a_current_snapshot_obligation(protocol): + """The actual producer's scope describes required, not current, finality.""" + async with projection_harness("memory") as h: + s = await durable_case(h.store, required="official", finality="snapshot", active=False) + await h.projection.activate(account_id=s.obligation.account_id) + mounted = MountedFeed(h) + mounted.authorize(s) + request = GetReportingStatusRequest.model_validate( + feed_request(s, limit=1, finality=["official"]) + ) + async with mounted.sdk_clients("1.0") as (clients, observed): + ledger = await load_reporting_ledger(clients[protocol], request) + assert [value.value for value in ledger.scope.finality] == ["official"] + assert len(ledger.obligations) == len(ledger.revisions) == 1 + assert ledger.obligations[0].required_finality.value == "official" + assert ledger.revisions[0].finality.value == "snapshot" + assert all("finality" not in params for _, _, params in observed) + obligation = ledger.obligations[0] + expected = ExpectedReportingPeriod( + obligation.delivery_config_id, + obligation.delivery_config_version, + obligation.report_definition_id, + obligation.feed_purpose.value, + obligation.reporting_profile, + tuple(b.root for b in obligation.media_buy_ids), + obligation.period.start.isoformat(), + obligation.period.end.isoformat(), + obligation.period.source_timezone, + ) + result = evaluate_reporting_ledger(ledger, expected_periods=[expected], now=h.clock()) + assert not result.definitive + assert result.missing_expected_periods == [] + assert "FINALITY_NOT_MET" in result.obligations[0].reasons + assert "UNVERIFIED_LEDGER_SNAPSHOT" not in result.obligations[0].reasons diff --git a/tests/conformance/reporting/test_reporting_core_lifecycle.py b/tests/conformance/reporting/test_reporting_core_lifecycle.py index 505e330a8..4b70ab242 100644 --- a/tests/conformance/reporting/test_reporting_core_lifecycle.py +++ b/tests/conformance/reporting/test_reporting_core_lifecycle.py @@ -35,6 +35,7 @@ import os import secrets from collections.abc import AsyncIterator +from dataclasses import replace from datetime import datetime, timedelta, timezone from types import SimpleNamespace from typing import Any @@ -604,14 +605,16 @@ async def _restate( # -------------------------------------------------------------------------- -async def test_a_buyer_detects_a_period_the_seller_never_obligated( +@pytest.mark.parametrize("all_finalities", [False, True]) +async def test_missing_period_claim_requires_complete_finality_denominator( ledger: PgReportingLedgerStore, + all_finalities: bool, ) -> None: - """The reason the buyer derives its own expectations. + """An absent expected period prevents success, but needs proof to be claimed. - A seller that simply omits a period returns a complete-looking, internally - consistent ledger. Only the buyer's independently derived denominator - catches it. + ExpectedReportingPeriod carries no trusted finality requirement. A proper + subset in the seller's denominator cannot prove that an absent period + belongs to it, even when that subset came from an unfiltered read. """ source = SimulatedSource() await ledger.put_configuration(_configuration()) @@ -620,15 +623,27 @@ async def test_a_buyer_detects_a_period_the_seller_never_obligated( # Stop before the second period closes, so the seller's ledger is complete # and internally consistent -- exactly the shape that hides an omission. await _run_worker_at(ledger, source, now=first.end + timedelta(minutes=30)) + if all_finalities: + # Expand the real seller's configuration denominator, without rewriting + # its response or introducing an obligation for this other generation. + await ledger.put_configuration( + replace(_configuration("official"), delivery_config_id="official_delivery") + ) # The seller obligated one period; the buyer expects two. settled = await _reconcile(ledger, expected=_expected_periods(1)) assert settled.definitive is True + assert {value.value for value in settled.ledger.scope.finality} == ( + {"snapshot", "official"} if all_finalities else {"snapshot"} + ) gap = await _reconcile(ledger, expected=_expected_periods(2)) assert gap.definitive is False - assert len(gap.missing_expected_periods) == 1 - assert gap.missing_expected_periods[0].period_start == _period(1).start.isoformat() + if all_finalities: + assert len(gap.missing_expected_periods) == 1 + assert gap.missing_expected_periods[0].period_start == _period(1).start.isoformat() + else: + assert gap.missing_expected_periods == [] async def test_core_reconciliation_refuses_a_managed_delivery_ledger( diff --git a/tests/conformance/reporting/test_reporting_publication_time.py b/tests/conformance/reporting/test_reporting_publication_time.py index a10347a37..e7eafd93a 100644 --- a/tests/conformance/reporting/test_reporting_publication_time.py +++ b/tests/conformance/reporting/test_reporting_publication_time.py @@ -11,8 +11,8 @@ from adcp.reporting import ( ExpectedReportingPeriod, - ReportingLedger, evaluate_reporting_ledger, + load_reporting_ledger, ) from adcp.reporting.conformance import validate_reporting_source_execution from adcp.reporting.ledger import ( @@ -28,7 +28,8 @@ ) from adcp.reporting.materializer import ReportingWriterCapability, reference_verifier from adcp.reporting.source import SourceBatchManifestV1, parse_verified_source_batch_manifest_v1 -from adcp.types import GetReportingStatusResponse +from adcp.types import GetReportingStatusRequest, GetReportingStatusResponse +from adcp.types.core import TaskResult from adcp.validation.schema_loader import get_named_validator from ._generation_support import END, START, isolated_reporting_pool @@ -147,29 +148,35 @@ def producer(): async def public_outcome(store, config): - payload = await ReportingStatusHandler(store).handle( - { - "adcp_version": "3.2-rc.6", - "account": {"account_id": config.account_id}, - "view": "periods", - "period": {"start": START.isoformat(), "end": END.isoformat()}, - }, - caller=ReportingStatusCaller(account_id=config.account_id, consumer_id="clock-buyer"), - ) - response = GetReportingStatusResponse.model_validate(payload) + handler = ReportingStatusHandler(store) + caller = ReportingStatusCaller(account_id=config.account_id, consumer_id="clock-buyer") validator = get_named_validator("core/reporting-revision.json", version="3.2.0-rc.6") assert validator is not None - for revision in payload["revisions"]: - validator.validate(revision) - ledger = ReportingLedger( - ledger_snapshot_id=response.ledger_snapshot_id, - ledger_as_of=response.ledger_as_of, - account_id=response.account_id, - scope=response.scope, - obligations=response.periods, - revisions=response.revisions, - materializations=response.materializations, - receipts=response.receipts, + + class StatusClient: + async def get_reporting_status(self, request): + payload = await handler.handle( + request.model_dump(mode="json", exclude_none=True), caller=caller + ) + for revision in payload["revisions"]: + validator.validate(revision) + return TaskResult( + success=True, + data=GetReportingStatusResponse.model_validate(payload), + status="completed", + ) + + ledger = await load_reporting_ledger( + StatusClient(), + GetReportingStatusRequest.model_validate( + { + "adcp_version": "3.2-rc.6", + "account": {"account_id": config.account_id}, + "view": "periods", + "period": {"start": START.isoformat(), "end": END.isoformat()}, + "pagination": {"max_results": 1}, + } + ), ) expected = [ ExpectedReportingPeriod( @@ -183,7 +190,7 @@ async def public_outcome(store, config): END.isoformat(), ) ] - return response, evaluate_reporting_ledger(ledger, expected_periods=expected) + return evaluate_reporting_ledger(ledger, expected_periods=expected) @pytest.mark.parametrize("observed", [END, TURN, TURN + timedelta(seconds=1)]) @@ -201,8 +208,8 @@ async def test_creation_follows_acquisition_and_staged_read(store, tmp_path, obs object_reader=source.reader, clock=clock, ) - response, outcome = await public_outcome(store, config) - revision = response.revisions[0] + outcome = await public_outcome(store, config) + revision = outcome.ledger.revisions[0] assert outcome.definitive, [o.reasons for o in outcome.obligations] assert request.period.source_read_cutoff_at == TURN assert revision.created_at == PUBLISHED @@ -221,8 +228,8 @@ async def test_real_clock_observation_after_dispatch_is_definitive(store, tmp_pa result=result, object_reader=source.reader, ) - response, outcome = await public_outcome(store, config) - revision = response.revisions[0] + outcome = await public_outcome(store, config) + revision = outcome.ledger.revisions[0] assert outcome.definitive, [o.reasons for o in outcome.obligations] assert request.period.source_read_cutoff_at < manifest.observed_at <= revision.created_at assert revision.observed_at == revision.finalized_at == manifest.observed_at diff --git a/tests/conformance/reporting/test_reporting_reconciliation_transitions.py b/tests/conformance/reporting/test_reporting_reconciliation_transitions.py index df8303428..a59649a56 100644 --- a/tests/conformance/reporting/test_reporting_reconciliation_transitions.py +++ b/tests/conformance/reporting/test_reporting_reconciliation_transitions.py @@ -9,7 +9,7 @@ import pytest -from adcp.reporting import ReportingLedger, evaluate_reporting_ledger +from adcp.reporting import evaluate_reporting_ledger, load_reporting_ledger from adcp.reporting.canonical_json import canonical_json_utf8_v1 from adcp.reporting.ledger import ( LedgerConflictError, @@ -26,7 +26,8 @@ revision_to_wire, ) from adcp.reporting.ledger.delivery_models import DeliveryMethod, VerificationProfile -from adcp.types import GetReportingStatusResponse, ReportingReceipt +from adcp.types import GetReportingStatusRequest, GetReportingStatusResponse, ReportingReceipt +from adcp.types.core import TaskResult, TaskStatus from ._generation_support import END, NOW, configuration from ._reconciliation_support import Clock, Store, scenario @@ -66,6 +67,8 @@ async def test_generated_projections_are_accepted_by_buyer_reconciler( successful_materialization_count=1, receipt_count=1, accepted_receipt_count=1, + adjustment_receipt_count=0, + accepted_adjustment_receipt_count=0, resource_retained_until=s.delivery.resource_retained_until.isoformat(), ) result["revisions"] = [revision_to_wire(s.revision, obligation=s.obligation)] @@ -73,15 +76,18 @@ async def test_generated_projections_are_accepted_by_buyer_reconciler( result["receipts"] = [receipt_to_wire(receipt)] result["pagination"]["total_count"] = 4 response = GetReportingStatusResponse.model_validate(result) - ledger = ReportingLedger( - response.ledger_snapshot_id, - response.ledger_as_of, - response.account_id, - response.scope, - response.periods, - response.revisions, - response.materializations, - response.receipts, + + class FrozenClient: + async def get_reporting_status( + self, request: GetReportingStatusRequest + ) -> TaskResult[GetReportingStatusResponse]: + return TaskResult(status=TaskStatus.COMPLETED, data=response) + + ledger = await load_reporting_ledger( + FrozenClient(), + GetReportingStatusRequest.model_validate( + {"account": {"account_id": "acct_a"}, "view": "periods"} + ), ) verdict = evaluate_reporting_ledger(ledger, expected_periods=[], now=NOW) assert verdict.definitive, verdict.obligations diff --git a/tests/test_reporting_frozen_read_corrections.py b/tests/test_reporting_frozen_read_corrections.py new file mode 100644 index 000000000..6095771c7 --- /dev/null +++ b/tests/test_reporting_frozen_read_corrections.py @@ -0,0 +1,490 @@ +"""Regressions from the independent review of the frozen buyer read slice.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import replace +from typing import Any + +import pytest + +from adcp.reporting import ( + ReportingReconciliationError, + evaluate_reporting_ledger, + load_reporting_ledger, +) +from adcp.reporting.ownership import with_revision_ownership +from adcp.validation.schema_loader import get_named_validator +from tests.test_reporting_frozen_read_safety import ( + ARRAYS, + NOW, + SECRET, + Pages, + _adjustment_receipt, + _counts, + _expected, + _history, + _load, + _pages, + _request, + _with_status, +) + + +@pytest.mark.parametrize("explicit", [False, True]) +@pytest.mark.parametrize("declare_pending", [False, True]) +@pytest.mark.parametrize( + "evidence", ["missing", "rejected", "accepted", "accepted-leaf", "partial"] +) +async def test_delivery_only_adjustments_require_each_accepted_current_leaf( + explicit, declare_pending, evidence +): + raw = _history(adjustments=True) + raw["receipts"] = [] + if evidence == "missing": + raw["adjustment_receipts"] = [] + elif evidence == "rejected": + raw["adjustment_receipts"] = [_adjustment_receipt(status="rejected")] + elif evidence == "accepted-leaf": + raw["adjustment_receipts"] = [ + _adjustment_receipt(supersedes_reporting_receipt_id="adjustment-receipt-rejected"), + _adjustment_receipt( + reporting_receipt_id="adjustment-receipt-rejected", status="rejected" + ), + ] + elif evidence == "partial": + second = deepcopy(raw["adjustments"][0]) + second["reporting_adjustment_id"] = "adjustment-unreceipted" + raw["adjustments"].append(second) + _counts(raw) + owner = raw["periods"][0] + owner.update(reconciliation_mode="delivery_only", reconciliation_status="not_required") + for item in [owner, *raw["materializations"], *raw["scope"]["delivery_config_generations"]]: + item["feed_purpose"] = "analytics" + owner["schedule"].update(period_anchor=owner["period"]["start"], period_timezone="UTC") + # These counts do not apply in delivery-only mode. Adjustment evidence must + # remain a separate condition even when the seller omits the pending count. + for field in ( + "receipt_count", + "accepted_receipt_count", + "adjustment_receipt_count", + "accepted_adjustment_receipt_count", + ): + owner.pop(field) + accepted = evidence in {"accepted", "accepted-leaf"} + if declare_pending: + owner["pending_adjustment_count"] = int(not accepted) + validator = get_named_validator("core/reporting-obligation.json", version="3.2.0-rc.6") + assert validator is not None + assert list(validator.iter_errors(owner)) == [] + + result = evaluate_reporting_ledger( + await _load(raw, explicit=explicit), + expected_periods=[replace(_expected()[0], feed_purpose="analytics")], + now=NOW, + ) + assert result.definitive is accepted + assert result.obligations[0].definitive is accepted + assert result.obligations[0].reasons == ( + () if accepted else ("MISSING_MATCHING_ADJUSTMENT_RECEIPT",) + ) + + +def _distinct_histories(count: int, *, distinction: str = "campaigns") -> dict[str, Any]: + raw = _history() + prototype = deepcopy(raw) + raw.update({name: [] for name in ARRAYS}) + raw["scope"]["all_accessible_media_buys"] = True + raw["scope"].pop("media_buy_ids") + for number in range(count): + owner = deepcopy(prototype["periods"][0]) + revision = deepcopy(prototype["revisions"][0]) + material = deepcopy(prototype["materializations"][0]) + owner.update( + reporting_obligation_id=f"obligation-number-{number}", + reconciliation_mode="delivery_only", + reconciliation_status="not_required", + receipt_count=0, + accepted_receipt_count=0, + ) + owner["schedule"].update(period_anchor=owner["period"]["start"], period_timezone="UTC") + if distinction == "campaigns": + buys = [f"buy-{number}-a", f"buy-{number}-b"] + for item in (owner, revision): + item["media_buy_ids"] = buys + item["coverage"]["media_buy_ids"] = buys + item["coverage"]["fully_covered_media_buy_ids"] = buys + elif distinction == "profile": + owner["reporting_profile"] = revision["reporting_profile"] = f"profile-{number}" + revision["reporting_revision_id"] = f"revision-number-{number}" + material.update( + reporting_materialization_id=f"materialization-number-{number}", + reporting_obligation_id=owner["reporting_obligation_id"], + reporting_revision_id=revision["reporting_revision_id"], + ) + raw["periods"].append(owner) + raw["revisions"].append(revision) + raw["materializations"].append(material) + return raw + + +def _split(raw: dict[str, Any], explicit: bool) -> list[dict[str, Any]]: + raw["pagination"]["total_count"] = sum(len(raw[name]) for name in ARRAYS) + pages = _pages(raw, explicit=False) + if explicit: + ownership = { + material["reporting_revision_id"]: material["reporting_obligation_id"] + for material in raw["materializations"] + } + return [with_revision_ownership(page, ownership) for page in pages] + return pages + + +async def _load_distinct(raw: dict[str, Any], explicit: bool): + return await load_reporting_ledger( + Pages(_split(raw, explicit)), _request(pagination={"max_results": 1}) + ) + + +@pytest.mark.parametrize("explicit", [False, True]) +@pytest.mark.parametrize("distinction", ["campaigns", "profile", "identical"]) +async def test_ownerless_revisionless_status_does_not_certify_multiple_obligations( + explicit, distinction +): + raw = _distinct_histories(2, distinction=distinction) + status = _with_status(_history())["consumer_statuses"][0] + for key in ( + "reporting_obligation_id", + "reporting_revision_id", + "observed_revision_content_sha256", + ): + status.pop(key) + status["consumer_status"] = "obligation_missing" + validator = get_named_validator("core/reporting-consumer-status.json", version="3.2.0-rc.6") + assert validator is not None + assert list(validator.iter_errors(status)) == [] + raw["consumer_statuses"] = [status] + for owner in raw["periods"]: + owner.update( + consumer_status_count=1, current_consumer_status_id=status["reporting_status_id"] + ) + ledger = await _load_distinct(raw, explicit) + result = evaluate_reporting_ledger(ledger, expected_periods=[], now=NOW) + assert not result.definitive + assert result.missing_expected_periods == [] + assert all("UNVERIFIED_LEDGER_SNAPSHOT" in outcome.reasons for outcome in result.obligations) + # Resolving ambiguity must never rewrite the retained legacy statement. + assert ledger.consumer_statuses[0].reporting_obligation_id is None + assert ledger.consumer_statuses[0].reporting_revision_id is None + + +@pytest.mark.parametrize("explicit", [False, True]) +@pytest.mark.parametrize("binding", ["owner", "revision"]) +async def test_exact_status_binding_counts_once_even_when_logical_keys_match(explicit, binding): + raw = _distinct_histories(2) + status = _with_status(_history())["consumer_statuses"][0] + status["reporting_revision_id"] = raw["revisions"][0]["reporting_revision_id"] + if binding == "owner": + status["reporting_obligation_id"] = raw["periods"][0]["reporting_obligation_id"] + else: + # Pure already-typed legacy tolerance; rc6 received-status wire still + # requires the owner, and these tests claim no mounted compatibility. + status.pop("reporting_obligation_id") + raw["consumer_statuses"] = [status] + raw["periods"][0].update( + consumer_status_count=1, current_consumer_status_id=status["reporting_status_id"] + ) + raw["periods"][1]["consumer_status_count"] = 0 + result = evaluate_reporting_ledger( + await _load_distinct(raw, explicit), expected_periods=[], now=NOW + ) + assert result.definitive, result.obligations + + +@pytest.mark.parametrize("explicit", [False, True]) +async def test_an_explicit_current_status_cannot_be_claimed_by_a_second_owner(explicit): + raw = _distinct_histories(2, distinction="identical") + status = _with_status(_history())["consumer_statuses"][0] + status.update( + reporting_obligation_id=raw["periods"][0]["reporting_obligation_id"], + reporting_revision_id=raw["revisions"][0]["reporting_revision_id"], + ) + raw["consumer_statuses"] = [status] + for owner in raw["periods"]: + owner.update( + consumer_status_count=1, current_consumer_status_id=status["reporting_status_id"] + ) + with pytest.raises(ReportingReconciliationError): + await _load_distinct(raw, explicit) + + +@pytest.mark.parametrize("explicit", [False, True]) +async def test_exact_status_owner_and_revision_binding_cannot_disagree(explicit): + raw = _distinct_histories(2, distinction="identical") + status = _with_status(_history())["consumer_statuses"][0] + status.update( + reporting_obligation_id=raw["periods"][0]["reporting_obligation_id"], + reporting_revision_id=raw["revisions"][1]["reporting_revision_id"], + ) + raw["consumer_statuses"] = [status] + raw["periods"][0].update( + consumer_status_count=1, current_consumer_status_id=status["reporting_status_id"] + ) + raw["periods"][1]["consumer_status_count"] = 0 + with pytest.raises(ReportingReconciliationError) as error: + await _load_distinct(raw, explicit) + assert error.value.code == "INVALID_LEDGER_DEPENDENCY" + + +@pytest.mark.parametrize("explicit", [False, True]) +@pytest.mark.parametrize( + "field", + [ + "materialization_count", + "successful_materialization_count", + "receipt_count", + "accepted_receipt_count", + "adjustment_count", + "adjustment_receipt_count", + "accepted_adjustment_receipt_count", + ], +) +async def test_absent_optional_count_is_diagnostic_but_contradictory_count_is_rejected( + explicit, field +): + raw = _history(adjustments=True) + raw["periods"][0]["health"] = "waiting" + raw["periods"][0]["schedule"].update( + period_anchor=raw["periods"][0]["period"]["start"], period_timezone="UTC" + ) + actual = raw["periods"][0].pop(field) + # These tier counts are schema-optional; missing evidence is not a false count. + # Keep the conditional requirements for healthy/complete projections intact. + validator = get_named_validator("core/reporting-obligation.json", version="3.2.0-rc.6") + assert validator is not None + assert list(validator.iter_errors(raw["periods"][0])) == [] + result = evaluate_reporting_ledger( + await _load(raw, explicit=explicit), expected_periods=_expected(), now=NOW + ) + assert not result.definitive + assert "ASSOCIATED_HISTORY_INCOMPLETE" in result.obligations[0].reasons + assert result.missing_expected_periods == [] + raw["periods"][0][field] = actual + 1 + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw, explicit=explicit) + assert error.value.code == "LEDGER_COUNT_MISMATCH" + + +def _partly_owned_legacy_history() -> dict[str, Any]: + raw = _history(adjustments=True) + first = raw["periods"][0] + first["revision_count"] = 2 + second = deepcopy(first) + second.update( + reporting_obligation_id="obligation-second", + reconciliation_mode="delivery_only", + reconciliation_status="not_required", + health="waiting", + revision_count=1, + materialization_count=0, + successful_materialization_count=0, + receipt_count=0, + accepted_receipt_count=0, + adjustment_count=0, + adjustment_receipt_count=0, + accepted_adjustment_receipt_count=0, + ) + raw["periods"].append(second) + unbound = deepcopy(raw["revisions"][0]) + unbound.update(reporting_revision_id="revision-unbound", finality="snapshot") + for key in ("finalized_at", "finality_basis", "finality_policy_id"): + unbound.pop(key) + raw["revisions"].append(unbound) + return raw + + +@pytest.mark.parametrize( + ("field", "declared"), + [ + ("revision_count", 0), + ("revision_count", 3), + ("adjustment_count", 0), + ("adjustment_count", 2), + ("adjustment_receipt_count", 0), + ("adjustment_receipt_count", 2), + ("accepted_adjustment_receipt_count", 0), + ("accepted_adjustment_receipt_count", 2), + ], +) +async def test_legacy_ambiguity_does_not_hide_counts_outside_proven_bounds(field, declared): + raw = _partly_owned_legacy_history() + # The first revision and its adjustment evidence have an exact materialization + # owner. Only the extra snapshot can belong to either identical-scope period. + raw["periods"][0][field] = declared + with pytest.raises(ReportingReconciliationError) as error: + await _load_distinct(raw, explicit=False) + assert error.value.code == "LEDGER_COUNT_MISMATCH" + + +@pytest.mark.parametrize("declared", [1, 2]) +async def test_legacy_count_inside_proven_bounds_remains_diagnostic(declared): + raw = _partly_owned_legacy_history() + raw["periods"][0]["revision_count"] = declared + result = evaluate_reporting_ledger( + await _load_distinct(raw, explicit=False), expected_periods=_expected(), now=NOW + ) + assert not result.definitive + assert result.missing_expected_periods == [] + assert all("AMBIGUOUS_REVISION_OWNERSHIP" in item.reasons for item in result.obligations) + + +async def test_pending_adjustment_count_cannot_invent_ownership_of_selected_official(): + raw = _partly_owned_legacy_history() + owned, unbound = raw["revisions"] + unbound["finality"] = "official" + owned["finality"] = "snapshot" + for key in ("finalized_at", "finality_basis", "finality_policy_id"): + unbound[key] = owned.pop(key) + raw["adjustments"][0]["adjusts_reporting_revision_id"] = unbound["reporting_revision_id"] + raw["adjustment_receipts"] = [] + for owner in raw["periods"]: + owner.update( + adjustment_receipt_count=0, + accepted_adjustment_receipt_count=0, + pending_adjustment_count=0, + ) + result = evaluate_reporting_ledger( + await _load_distinct(raw, explicit=False), expected_periods=[], now=NOW + ) + assert not result.definitive + assert all("AMBIGUOUS_REVISION_OWNERSHIP" in item.reasons for item in result.obligations) + # Even ambiguous ownership cannot explain more pending rows than exist. + raw["periods"][0]["pending_adjustment_count"] = 2 + with pytest.raises(ReportingReconciliationError) as error: + await _load_distinct(raw, explicit=False) + assert error.value.code == "LEDGER_COUNT_MISMATCH" + + +@pytest.mark.parametrize("explicit", [False, True]) +async def test_ownership_extension_does_not_make_absent_adjustment_count_an_error(explicit): + raw = _distinct_histories(1) + raw["periods"][0].pop("adjustment_count") + result = evaluate_reporting_ledger( + await _load_distinct(raw, explicit), expected_periods=[], now=NOW + ) + assert not result.definitive + assert "ASSOCIATED_HISTORY_INCOMPLETE" in result.obligations[0].reasons + + +@pytest.mark.parametrize("explicit", [False, True]) +async def test_disabled_consumer_status_mode_does_not_require_optional_status_counts(explicit): + raw = _history() + assert "consumer_status_count" not in raw["periods"][0] + result = evaluate_reporting_ledger( + await _load(raw, explicit=explicit), expected_periods=_expected(), now=NOW + ) + assert result.definitive + + +@pytest.mark.parametrize("finality", [["official"], ["snapshot"], []]) +async def test_unproven_expected_finality_suppresses_absence_without_certifying_success(finality): + raw = _history() + raw["scope"]["finality"] = finality + raw.update({name: [] for name in ARRAYS}) + raw["pagination"]["total_count"] = 0 + result = evaluate_reporting_ledger(await _load(raw), expected_periods=_expected(), now=NOW) + assert result.missing_expected_periods == [] + assert not result.definitive + + +async def test_official_only_scope_still_certifies_positive_returned_evidence(): + raw = _history() + assert raw["scope"]["finality"] == ["official"] + result = evaluate_reporting_ledger(await _load(raw), expected_periods=_expected(), now=NOW) + assert result.definitive + # Missing a different profile is an unproven expectation, never a success. + unproven = replace(_expected()[0], reporting_profile="other-profile") + result = evaluate_reporting_ledger(await _load(raw), expected_periods=[unproven], now=NOW) + assert not result.definitive + assert result.missing_expected_periods == [] + + +@pytest.mark.parametrize("malformed", ["generation", "account", "pagination"]) +async def test_request_construction_failure_has_a_distinct_closed_code_and_no_remote_call( + malformed, +): + client = Pages(_pages(_history())) + changes = { + "generation": {"delivery_config_ids": [""]}, + "account": {"account": SECRET}, + "pagination": {"pagination": SECRET}, + }[malformed] + request = _request(context={"private": SECRET}).model_copy(update=changes) + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(client, request) + assert error.value.code == "INVALID_STATUS_REQUEST" + assert client.requests == [] + assert SECRET not in str(error.value) + assert SECRET not in repr(error.value) + assert error.value.__context__ is None + assert error.value.__cause__ is None + + +async def test_response_validation_failure_stays_remote_typed_and_redacted(): + raw = _with_status(_history()) + raw["consumer_statuses"][0]["reporting_status_id"] = f"{SECRET}/invalid" + client = Pages(_pages(raw)) + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(client, _request()) + assert error.value.code == "STATUS_READ_FAILED" + assert SECRET not in str(error.value) + assert SECRET not in repr(error.value) + assert error.value.__context__ is None + assert error.value.__cause__ is None + + +@pytest.mark.parametrize("count", [20, 80]) +@pytest.mark.parametrize("explicit", [False, True]) +async def test_public_read_and_evaluation_do_not_rescan_unrelated_histories( + monkeypatch, count, explicit +): + from adcp.reporting import _reconcile + + comparisons = 0 + original = _reconcile._revision_matches_obligation + + def counted(revision, obligation): + nonlocal comparisons + comparisons += 1 + return original(revision, obligation) + + monkeypatch.setattr(_reconcile, "_revision_matches_obligation", counted) + raw = _distinct_histories(count) + result = evaluate_reporting_ledger( + await _load_distinct(raw, explicit), expected_periods=[], now=NOW + ) + assert result.definitive, result.obligations + # Work grows with records, not all obligation/revision pairs; no time threshold. + assert comparisons <= 32 * count + + +async def test_ambiguous_legacy_fanout_has_a_work_bound_as_well_as_a_row_bound(): + raw = _distinct_histories(450, distinction="identical") + raw["materializations"] = [] + for owner in raw["periods"]: + owner.update( + revision_count=450, + materialization_count=0, + successful_materialization_count=0, + health="waiting", + ) + raw["pagination"]["total_count"] = 900 + # The 900 received rows could otherwise expand into over 200,000 ambiguous + # owner/revision associations despite fitting the ordinary transport budget. + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger( + Pages(_pages(raw, explicit=False)), + _request(pagination={"max_results": 1}), + max_records=1000, + ) + assert error.value.code == "LEDGER_LIMIT_EXCEEDED" diff --git a/tests/test_reporting_frozen_read_safety.py b/tests/test_reporting_frozen_read_safety.py new file mode 100644 index 000000000..09504695a --- /dev/null +++ b/tests/test_reporting_frozen_read_safety.py @@ -0,0 +1,848 @@ +"""Buyer read safety through the public full-read and pure-evaluation APIs.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import replace +from datetime import datetime, timezone +from typing import Any + +import pytest + +from adcp.reporting import ( + ExpectedReportingPeriod, + ReportingLedger, + ReportingReconciliationError, + evaluate_reporting_ledger, + load_reporting_ledger, +) +from adcp.reporting.ownership import with_revision_ownership +from adcp.types import GetReportingStatusRequest, GetReportingStatusResponse +from adcp.types.core import TaskResult, TaskStatus +from tests.test_reporting_reconciliation import ( + DIGEST, + PERIOD, + REVISION, + TOTALS, + _full_finality_scope, + _response, +) + +NOW = datetime(2026, 9, 3, tzinfo=timezone.utc) +OWNER = "obligation-billing" +REVISION_ID = "revision-august-official" +SECRET = "secret-wire-body-never-diagnostic" +ARRAYS = ( + "periods", + "revisions", + "materializations", + "receipts", + "adjustments", + "adjustment_receipts", + "consumer_statuses", +) + + +def _receipt(**changes: Any) -> dict[str, Any]: + result = { + "reporting_receipt_id": "receipt-accepted", + "reporting_obligation_id": OWNER, + "reporting_revision_id": REVISION_ID, + "reporting_materialization_id": "materialization-billing", + "status": "accepted", + "verification_profile": "canonical_digest", + "observed_row_count": 7, + "observed_control_totals": deepcopy(TOTALS), + "observed_canonical_content_digest": deepcopy(DIGEST), + "observed_at": "2026-09-02T00:01:00Z", + "received_at": "2026-09-02T00:01:01Z", + } + result.update(changes) + if result["status"] == "rejected": + result["rejection_codes"] = ["EVIDENCE_MISMATCH"] + return result + + +def _adjustment() -> dict[str, Any]: + return { + "reporting_adjustment_id": "adjustment-correction", + "adjusts_reporting_revision_id": REVISION_ID, + "reason_code": "source_correction", + "accounting_period": {"start": PERIOD["start"], "end": PERIOD["end"]}, + "control_total_deltas": [ + {"name": "spend", "value": "-1.00", "value_type": "decimal", "unit": "USD"} + ], + # This is seller-advertised evidence, not a claim of raw-byte hashing. + "canonical_adjustment_sha256": "c" * 64, + "correction_observed_at": "2026-09-02T00:01:02Z", + "created_at": "2026-09-02T00:01:03Z", + } + + +def _adjustment_receipt(**changes: Any) -> dict[str, Any]: + result = { + "reporting_receipt_id": "adjustment-receipt-accepted", + "reporting_adjustment_id": "adjustment-correction", + "adjusts_reporting_revision_id": REVISION_ID, + "status": "accepted", + "observed_adjustment_sha256": "c" * 64, + "observed_at": "2026-09-02T00:01:04Z", + "received_at": "2026-09-02T00:01:05Z", + } + result.update(changes) + if result["status"] == "rejected": + result["rejection_codes"] = ["ADJUSTMENT_DIGEST_MISMATCH"] + return result + + +def _history(*, adjustments: bool = False) -> dict[str, Any]: + raw = deepcopy(_response([_receipt()])) + raw["ledger_as_of"] = "2026-09-02T00:02:00Z" + raw["changes_checkpoint"] = "frozen-checkpoint" + raw["adjustments"] = [_adjustment()] if adjustments else [] + raw["adjustment_receipts"] = [_adjustment_receipt()] if adjustments else [] + raw["consumer_statuses"] = [] + _counts(raw) + return raw + + +def _with_status(raw: dict[str, Any]) -> dict[str, Any]: + raw["consumer_statuses"] = [ + { + "reporting_status_id": "consumer-status-current", + "delivery_config_id": "billing-feed", + "delivery_config_version": 1, + "report_definition_id": "billing-v1", + "period": deepcopy(PERIOD), + "reporting_obligation_id": OWNER, + "reporting_revision_id": REVISION_ID, + "observed_revision_content_sha256": REVISION["revision_content_sha256"], + "consumer_status": "received", + "status_as_of": "2026-09-02T00:01:00Z", + } + ] + raw["periods"][0].update( + consumer_status_count=1, current_consumer_status_id="consumer-status-current" + ) + _counts(raw) + return raw + + +def _counts(raw: dict[str, Any]) -> None: + """Recompute fixture counts; mutation tests deliberately run after this.""" + for obligation in raw["periods"]: + owner = obligation["reporting_obligation_id"] + materials = [m for m in raw["materializations"] if m["reporting_obligation_id"] == owner] + receipts = [r for r in raw["receipts"] if r["reporting_obligation_id"] == owner] + # These fixtures normally have one owner; multi-owner tests set their + # explicit per-owner counts themselves. + obligation.update( + revision_count=len(raw["revisions"]), + materialization_count=len(materials), + successful_materialization_count=sum( + m["status"] in {"available", "delivered"} for m in materials + ), + receipt_count=len(receipts), + accepted_receipt_count=sum(r["status"] == "accepted" for r in receipts), + adjustment_count=len(raw.get("adjustments", [])), + adjustment_receipt_count=len(raw.get("adjustment_receipts", [])), + accepted_adjustment_receipt_count=sum( + r["status"] == "accepted" for r in raw.get("adjustment_receipts", []) + ), + ) + raw["pagination"] = { + "has_more": False, + "total_count": sum(len(raw.get(name, [])) for name in ARRAYS), + } + + +def _pages(raw: dict[str, Any], *, explicit: bool = True) -> list[dict[str, Any]]: + rows = [(name, row) for name in ARRAYS for row in raw.get(name, [])] + pages = [] + for name, row in rows or [("periods", None)]: + page = {key: deepcopy(value) for key, value in raw.items() if key not in ARRAYS} + page.update({array: [] for array in ARRAYS}) + if row is not None: + page[name] = [deepcopy(row)] + if explicit: + page = with_revision_ownership( + page, {r["reporting_revision_id"]: OWNER for r in raw["revisions"]} + ) + pages.append(page) + for index, page in enumerate(pages): + page["pagination"]["has_more"] = index < len(pages) - 1 + if page["pagination"]["has_more"]: + page["pagination"]["cursor"] = f"page-{index + 1}" + return pages + + +class Pages: + def __init__(self, pages: list[dict[str, Any]]) -> None: + self.pages = pages + self.requests: list[GetReportingStatusRequest] = [] + + async def get_reporting_status( + self, request: GetReportingStatusRequest + ) -> TaskResult[GetReportingStatusResponse]: + page = self.pages[len(self.requests)] + self.requests.append(request) + return TaskResult( + status=TaskStatus.COMPLETED, + data=GetReportingStatusResponse.model_validate(deepcopy(page)), + ) + + +def _request(**changes: Any) -> GetReportingStatusRequest: + return GetReportingStatusRequest.model_validate( + {"account": {"account_id": "account-1"}, "view": "periods", **changes} + ) + + +async def _load(raw: dict[str, Any], *, explicit: bool = True) -> ReportingLedger: + return await load_reporting_ledger(Pages(_pages(raw, explicit=explicit)), _request()) + + +def _expected() -> list[ExpectedReportingPeriod]: + return [ + ExpectedReportingPeriod( + "billing-feed", + 1, + "billing-v1", + "billing", + "billing-v1", + ("buy-1", "buy-2"), + PERIOD["start"], + PERIOD["end"], + expected_at="2026-09-02T00:00:00Z", + ) + ] + + +async def test_full_read_discards_incremental_omissions_and_cursor_but_keeps_page_size() -> None: + class IncrementalOmissions(Pages): + async def get_reporting_status(self, request): + if request.changes_after: + raw = _history() + raw.update({name: [] for name in ARRAYS}) + raw["pagination"]["total_count"] = 0 + self.requests.append(request) + return TaskResult( + status=TaskStatus.COMPLETED, + data=GetReportingStatusResponse.model_validate(raw), + ) + return await super().get_reporting_status(request) + + client = IncrementalOmissions(_pages(_history())) + request = _request( + changes_after="incremental-checkpoint", + pagination={"cursor": "untrusted-continuation", "max_results": 1}, + ) + original = request.model_dump(mode="json") + ledger = await load_reporting_ledger(client, request) + result = evaluate_reporting_ledger(ledger, expected_periods=_expected(), now=NOW) + assert result.definitive + assert result.missing_expected_periods == [] + assert request.model_dump(mode="json") == original + assert len(client.requests) == 4 + assert all(r.changes_after is None for r in client.requests) + assert all(r.pagination.max_results == 1 for r in client.requests) + assert client.requests[0].pagination.cursor is None + + +async def test_health_finality_and_exact_revision_selectors_cannot_hide_history() -> None: + client = Pages(_pages(_history())) + request = _request( + view="revision", + reporting_revision_id=REVISION_ID, + health=["action_required"], + finality=["official"], + ) + await load_reporting_ledger(client, request) + assert all(r.view.value == "periods" for r in client.requests) + assert all( + r.health is None and r.finality is None and r.reporting_revision_id is None + for r in client.requests + ) + + +@pytest.mark.parametrize("outside", ["horizon", "generation", "media-buy", "retention"]) +async def test_expected_period_outside_the_proven_scope_is_not_an_obligation_missing_claim( + outside, +) -> None: + raw = _history() + raw.update({name: [] for name in ARRAYS}) + raw["pagination"]["total_count"] = 0 + expected = _expected()[0] + if outside == "horizon": + expected = replace( + expected, period_start="2026-07-01T00:00:00Z", period_end=PERIOD["start"] + ) + elif outside == "generation": + expected = replace(expected, delivery_config_version=2) + elif outside == "media-buy": + expected = replace(expected, media_buy_ids=("outside-authorized-scope",)) + else: + raw["scope"]["coverage_complete"] = False + result = evaluate_reporting_ledger(await _load(raw), expected_periods=[expected], now=NOW) + assert not result.definitive + assert result.missing_expected_periods == [] + + +async def test_full_retained_snapshot_can_establish_a_missing_obligation() -> None: + raw = _history() + # The expectation has no independent finality fact. Absence proof requires + # a denominator spanning both possible configuration requirements. + _full_finality_scope(raw) + raw.update({name: [] for name in ARRAYS}) + raw["pagination"]["total_count"] = 0 + result = evaluate_reporting_ledger(await _load(raw), expected_periods=_expected(), now=NOW) + assert not result.definitive + assert result.missing_expected_periods == _expected() + + +async def test_unproven_ledger_cannot_claim_definitive_or_missing_obligations() -> None: + raw = _history() + response = GetReportingStatusResponse.model_validate(raw) + manual = ReportingLedger( + response.ledger_snapshot_id, + response.ledger_as_of, + response.account_id, + response.scope, + [], + [], + [], + [], + ) + result = evaluate_reporting_ledger(manual, expected_periods=_expected(), now=NOW) + assert not result.definitive + assert result.missing_expected_periods == [] + assert not evaluate_reporting_ledger(manual, expected_periods=[], now=NOW).definitive + + +async def test_completed_snapshot_cannot_be_edited_into_different_evidence() -> None: + ledger = await _load(_history()) + ledger.receipts[0].observed_row_count = 99 + with pytest.raises(ReportingReconciliationError) as error: + evaluate_reporting_ledger(ledger, expected_periods=_expected(), now=NOW) + assert error.value.code == "LEDGER_CHANGED" + + +@pytest.mark.parametrize("explicit", [True, False]) +@pytest.mark.parametrize("field", ["changes_checkpoint", "next_expected_at", "health"]) +async def test_frozen_metadata_is_checked_including_wholly_legacy_pages(explicit, field) -> None: + pages = _pages(_history(), explicit=explicit) + pages[-1][field] = { + "changes_checkpoint": "different-checkpoint", + "next_expected_at": "2026-10-01T00:00:00Z", + "health": "waiting", + }[field] + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(Pages(pages), _request(), max_snapshot_restarts=0) + assert error.value.code == "SNAPSHOT_CHANGED" + + +async def test_every_page_requires_the_frozen_total() -> None: + pages = _pages(_history()) + pages[0]["pagination"].pop("total_count") + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(Pages(pages), _request(), max_snapshot_restarts=0) + assert error.value.code == "INCOMPLETE_LEDGER_PAGE" + + +@pytest.mark.parametrize( + "kind,field,value", + [ + ("periods", "health", "waiting"), + ("revisions", "row_count", 99), + ("materializations", "attempt", 2), + ("receipts", "observed_row_count", 99), + ("adjustments", "reason_detail", "changed correction"), + ("adjustment_receipts", "observed_adjustment_sha256", "0" * 64), + ("consumer_statuses", "consumer_commit_ref", "changed-consumer-checkpoint"), + ], +) +async def test_every_record_kind_requires_immutable_cross_page_content(kind, field, value) -> None: + pages = _pages(_with_status(_history(adjustments=True))) + index = next(i for i, page in enumerate(pages) if page[kind]) + replay = deepcopy(pages[index]) + replay[kind][0][field] = value + pages[index]["pagination"].update(has_more=True, cursor="before-conflict") + pages.insert(index + 1, replay) + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(Pages(pages), _request()) + assert error.value.code == "IMMUTABLE_RECORD_CHANGED" + + +async def test_optional_field_presence_is_part_of_typed_immutable_content() -> None: + pages = _pages(_history()) + replay = deepcopy(pages[-1]) + replay["receipts"][0]["consumer_commit_ref"] = None + pages[-1]["pagination"].update(has_more=True, cursor="before-optional-change") + pages.append(replay) + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(Pages(pages), _request()) + assert error.value.code == "IMMUTABLE_RECORD_CHANGED" + + +@pytest.mark.parametrize( + "field", + [ + "materialization_count", + "successful_materialization_count", + "receipt_count", + "accepted_receipt_count", + "consumer_status_count", + ], +) +async def test_frozen_counts_cover_materializations_receipts_and_consumer_statuses(field) -> None: + raw = _with_status(_history(adjustments=True)) + raw["periods"][0][field] += 1 + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw) + assert error.value.code == "LEDGER_COUNT_MISMATCH" + + +async def test_positive_status_count_requires_its_exact_current_status_dependency() -> None: + raw = _with_status(_history()) + raw["periods"][0]["current_consumer_status_id"] = "unknown-current-status" + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw) + assert error.value.code == "INVALID_LEDGER_DEPENDENCY" + + +async def test_legacy_typed_status_without_owner_uses_exact_revision_binding() -> None: + raw = _with_status(_history()) + raw["consumer_statuses"][0].pop("reporting_obligation_id") + ledger = await _load(raw) + assert ledger.consumer_statuses[0].reporting_obligation_id is None + assert evaluate_reporting_ledger(ledger, expected_periods=_expected(), now=NOW).definitive + + +@pytest.mark.parametrize("bounds", [{"max_bytes": 100}, {"max_records": 4}]) +async def test_byte_and_received_row_budgets_include_identical_replays(bounds) -> None: + pages = _pages(_history()) + replay = deepcopy(pages[0]) + replay["pagination"]["cursor"] = "replayed" + pages.insert(1, replay) + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(Pages(pages), _request(), **bounds) + assert error.value.code == "LEDGER_LIMIT_EXCEEDED" + + +async def test_detaches_returned_pages_before_a_client_reuses_its_models() -> None: + models = [GetReportingStatusResponse.model_validate(p) for p in _pages(_history())] + + class ReusingClient(Pages): + async def get_reporting_status(self, request): + index = len(self.requests) + if index: + models[0].periods[0].account_id = "changed-after-return" + self.requests.append(request) + return TaskResult(status=TaskStatus.COMPLETED, data=models[index]) + + ledger = await load_reporting_ledger(ReusingClient([]), _request()) + assert ledger.obligations[0].account_id == "account-1" + assert evaluate_reporting_ledger(ledger, expected_periods=_expected(), now=NOW).definitive + + +@pytest.mark.parametrize("explicit", [True, False]) +@pytest.mark.parametrize( + "field", ["adjustment_count", "adjustment_receipt_count", "accepted_adjustment_receipt_count"] +) +async def test_adjustment_frozen_counts_must_match_before_evaluation(explicit, field) -> None: + raw = _history(adjustments=True) + raw["periods"][0][field] += 1 + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw, explicit=explicit) + assert error.value.code == "LEDGER_COUNT_MISMATCH" + + +@pytest.mark.parametrize("kind", ["revision", "adjustment"]) +@pytest.mark.parametrize( + "topology", ["fork", "cycle", "disconnected-cycle", "gap", "roots", "accepted-predecessor"] +) +async def test_every_receipt_chain_is_exact_and_complete(kind, topology) -> None: + raw = _history(adjustments=kind == "adjustment") + build = _receipt if kind == "revision" else _adjustment_receipt + array = "receipts" if kind == "revision" else "adjustment_receipts" + first = build(reporting_receipt_id="reporting-receipt-first", status="rejected") + second = build( + reporting_receipt_id="reporting-receipt-second", + supersedes_reporting_receipt_id="reporting-receipt-first", + ) + records = [first, second] + if topology == "fork": + records.append( + build( + reporting_receipt_id="reporting-receipt-fork", + supersedes_reporting_receipt_id="reporting-receipt-first", + ) + ) + elif topology == "cycle": + first["supersedes_reporting_receipt_id"] = "reporting-receipt-second" + second.update(status="rejected", rejection_codes=["EVIDENCE_MISMATCH"]) + elif topology == "disconnected-cycle": + records.extend( + [ + build( + reporting_receipt_id="reporting-cycle-a", + status="rejected", + supersedes_reporting_receipt_id="reporting-cycle-b", + ), + build( + reporting_receipt_id="reporting-cycle-b", + status="rejected", + supersedes_reporting_receipt_id="reporting-cycle-a", + ), + ] + ) + elif topology == "gap": + second["supersedes_reporting_receipt_id"] = "missing-predecessor" + elif topology == "roots": + second.pop("supersedes_reporting_receipt_id") + else: + first.update(status="accepted") + first.pop("rejection_codes") + raw[array] = records + _counts(raw) + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw) + assert error.value.code == "INVALID_RECEIPT_CHAIN" + + +@pytest.mark.parametrize("kind", ["revision", "adjustment"]) +async def test_rejected_predecessors_and_only_the_accepted_current_leaf_satisfy(kind) -> None: + raw = _history(adjustments=kind == "adjustment") + build = _receipt if kind == "revision" else _adjustment_receipt + array = "receipts" if kind == "revision" else "adjustment_receipts" + raw[array] = [ + build( + reporting_receipt_id="reporting-receipt-current", + supersedes_reporting_receipt_id="receipt-rejected", + ), + build( + reporting_receipt_id="receipt-rejected", + status="rejected", + supersedes_reporting_receipt_id="reporting-receipt-root", + ), + build(reporting_receipt_id="reporting-receipt-root", status="rejected"), + ] + _counts(raw) + ledger = await _load(raw) + assert evaluate_reporting_ledger(ledger, expected_periods=_expected(), now=NOW).definitive + raw[array][0].update(status="rejected", rejection_codes=["EVIDENCE_MISMATCH"]) + _counts(raw) + ledger = await _load(raw) + result = evaluate_reporting_ledger(ledger, expected_periods=_expected(), now=NOW) + assert not result.definitive + assert ( + "MISSING_MATCHING_CONSUMER_RECEIPT" + if kind == "revision" + else "MISSING_MATCHING_ADJUSTMENT_RECEIPT" + ) in result.obligations[0].reasons + + +async def test_receipt_id_namespace_and_cross_kind_predecessors_are_not_interchangeable() -> None: + for collision in (False, True): + raw = _history(adjustments=True) + raw["receipts"][0].update(status="rejected", rejection_codes=["EVIDENCE_MISMATCH"]) + key = "reporting_receipt_id" if collision else "supersedes_reporting_receipt_id" + raw["adjustment_receipts"][0][key] = raw["receipts"][0]["reporting_receipt_id"] + _counts(raw) + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw) + assert error.value.code == "INVALID_RECEIPT_CHAIN" + + +@pytest.mark.parametrize("kind", ["revision", "adjustment"]) +async def test_receipt_predecessors_cannot_cross_exact_targets(kind) -> None: + raw = _history(adjustments=kind == "adjustment") + array = "receipts" if kind == "revision" else "adjustment_receipts" + predecessor = deepcopy(raw[array][0]) + predecessor.update( + reporting_receipt_id="receipt-other-target", + status="rejected", + rejection_codes=["EVIDENCE_MISMATCH"], + ) + if kind == "revision": + snapshot = _snapshot() + material = deepcopy(raw["materializations"][0]) + material.update( + reporting_revision_id=snapshot["reporting_revision_id"], + reporting_materialization_id="snapshot-material", + ) + raw["revisions"].append(snapshot) + raw["materializations"].append(material) + predecessor.update( + reporting_revision_id=snapshot["reporting_revision_id"], + reporting_materialization_id="snapshot-material", + ) + else: + adjustment = deepcopy(raw["adjustments"][0]) + adjustment["reporting_adjustment_id"] = "another-adjustment" + raw["adjustments"].append(adjustment) + predecessor["reporting_adjustment_id"] = "another-adjustment" + raw[array][0]["supersedes_reporting_receipt_id"] = "receipt-other-target" + raw[array].append(predecessor) + _counts(raw) + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw) + assert error.value.code == "INVALID_RECEIPT_CHAIN" + + +def _snapshot() -> dict[str, Any]: + revision = deepcopy(REVISION) + revision.update(reporting_revision_id="retained-snapshot", finality="snapshot") + for key in ("finality_basis", "finality_policy_id", "finalized_at"): + revision.pop(key) + return revision + + +@pytest.mark.parametrize("official_artifact", [False, True]) +async def test_official_precedes_snapshot_receipt_even_without_an_artifact( + official_artifact, +) -> None: + raw = _history() + snapshot = _snapshot() + raw["revisions"].insert(0, snapshot) + snapshot_material = deepcopy(raw["materializations"][0]) + snapshot_material.update( + reporting_materialization_id="snapshot-material", reporting_revision_id="retained-snapshot" + ) + snapshot_receipt = _receipt( + reporting_receipt_id="snapshot-receipt", + reporting_revision_id="retained-snapshot", + reporting_materialization_id="snapshot-material", + ) + raw["materializations"] = ( + [snapshot_material, *raw["materializations"]] if official_artifact else [snapshot_material] + ) + raw["receipts"] = [snapshot_receipt] + _counts(raw) + result = evaluate_reporting_ledger(await _load(raw), expected_periods=_expected(), now=NOW) + assert not result.definitive + assert result.obligations[0].reporting_revision_id == REVISION_ID + assert "MISSING_MATCHING_CONSUMER_RECEIPT" in result.obligations[0].reasons + + +async def test_accepted_snapshot_alone_cannot_close_reconciled_billing() -> None: + raw = _history() + raw["revisions"][0] = {**_snapshot(), "reporting_revision_id": REVISION_ID} + raw["periods"][0]["required_finality"] = "snapshot" + result = evaluate_reporting_ledger(await _load(raw), expected_periods=_expected(), now=NOW) + assert not result.definitive + assert "FINALITY_NOT_MET" in result.obligations[0].reasons + + +@pytest.mark.parametrize( + "later", ["failed", "pending", "success", "expired", "corrupt", "bad-producer-evidence"] +) +async def test_accepted_receipt_stays_bound_to_its_materialization_after_later_outcomes( + later, +) -> None: + raw = _history() + newer = deepcopy(raw["materializations"][0]) + newer.update(reporting_materialization_id="later-materialization", attempt=2) + if later in {"failed", "pending"}: + newer.update(status=later) + for key in ("ready_at", "verification", "resource"): + newer.pop(key) + if later == "failed": + newer.update(failed_at="2026-09-02T00:01:06Z", failure_code="WRITE_FAILED") + elif later == "expired": + newer["resource"]["expires_at"] = "2026-09-02T23:00:00Z" + elif later == "corrupt": + raw["periods"][0]["health"] = "action_required" + elif later == "bad-producer-evidence": + newer["verification"]["row_count"] = 999 + raw["materializations"].append(newer) + _counts(raw) + result = evaluate_reporting_ledger(await _load(raw), expected_periods=_expected(), now=NOW) + reasons = result.obligations[0].reasons + assert "MISSING_MATCHING_CONSUMER_RECEIPT" not in reasons + assert result.definitive == (later in {"failed", "pending", "success"}) + if later == "expired": + assert "RESOURCE_EXPIRED" in reasons + elif later == "corrupt": + assert "OBLIGATION_ACTION_REQUIRED" in reasons + elif later == "bad-producer-evidence": + assert "PRODUCER_CONTROL_TOTAL_MISMATCH" in reasons + + +async def test_expired_accepted_artifact_does_not_invalidate_a_readable_newer_artifact() -> None: + raw = _history() + newer = deepcopy(raw["materializations"][0]) + newer.update(reporting_materialization_id="later-materialization", attempt=2) + raw["materializations"][0]["resource"]["expires_at"] = "2026-09-02T23:00:00Z" + raw["materializations"].append(newer) + _counts(raw) + result = evaluate_reporting_ledger(await _load(raw), expected_periods=_expected(), now=NOW) + assert result.definitive + assert result.obligations[0].reporting_materialization_id == "later-materialization" + + +@pytest.mark.parametrize("field", ["row_count", "digest", "obligation", "materialization"]) +async def test_accepted_receipt_must_match_its_own_materialization_even_when_a_newer_one_is_good( + field, +) -> None: + raw = _history() + newer = deepcopy(raw["materializations"][0]) + newer.update(reporting_materialization_id="later-materialization", attempt=2) + raw["materializations"].append(newer) + if field == "row_count": + raw["materializations"][0]["verification"]["row_count"] = 999 + elif field == "digest": + raw["receipts"][0]["observed_canonical_content_digest"]["value"] = "0" * 64 + else: + raw["receipts"][0][f"reporting_{field}_id"] = "unavailable-or-foreign" + _counts(raw) + with pytest.raises(ReportingReconciliationError): + await _load(raw) + + +@pytest.mark.parametrize("explicit", [True, False]) +@pytest.mark.parametrize("target", ["obligation", "revision", "materialization"]) +async def test_all_receipt_dependencies_are_exact_in_both_ownership_modes(explicit, target) -> None: + raw = _history() + raw["receipts"][0][f"reporting_{target}_id"] = "missing-or-foreign" + with pytest.raises(ReportingReconciliationError): + await _load(raw, explicit=explicit) + + +async def test_legacy_history_cannot_hide_a_revision_without_any_owning_period() -> None: + raw = _history() + unrelated = { + **deepcopy(REVISION), + "reporting_revision_id": "unowned-revision", + "report_definition_id": "different-definition", + } + raw["revisions"].append(unrelated) + raw["pagination"]["total_count"] += 1 + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw, explicit=False) + assert error.value.code == "INVALID_LEDGER_DEPENDENCY" + + +@pytest.mark.parametrize( + "field,value", + [ + ("delivery_config_id", "foreign-config"), + ("delivery_config_version", 2), + ("destination_ref", "foreign-destination"), + ("feed_purpose", "pacing"), + ], +) +async def test_failed_materializations_must_still_have_exact_owning_scope(field, value) -> None: + raw = _history() + failed = deepcopy(raw["materializations"][0]) + failed.update( + reporting_materialization_id="failed-materialization", + status="failed", + failed_at="2026-09-02T00:01:06Z", + failure_code="WRITE_FAILED", + attempt=2, + ) + for key in ("ready_at", "resource", "verification"): + failed.pop(key) + failed[field] = value + raw["materializations"].append(failed) + _counts(raw) + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw) + assert error.value.code == "INVALID_LEDGER_DEPENDENCY" + + +@pytest.mark.parametrize("accepted", [False, True]) +async def test_pending_adjustment_count_matches_only_the_current_leaf(accepted) -> None: + raw = _history(adjustments=True) + if not accepted: + raw["adjustment_receipts"][0].update(status="rejected", rejection_codes=["MISMATCH"]) + _counts(raw) + raw["periods"][0]["pending_adjustment_count"] = int(not accepted) + result = evaluate_reporting_ledger(await _load(raw), expected_periods=_expected(), now=NOW) + assert result.definitive is accepted + raw["periods"][0]["pending_adjustment_count"] = int(accepted) + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw) + assert error.value.code == "LEDGER_COUNT_MISMATCH" + + +@pytest.mark.parametrize( + "mutation", + [ + "digest", + "correction-order", + "receipt-order", + "future-receipt", + "accounting-period", + "finality", + ], +) +async def test_adjustment_read_evidence_must_be_consistent(mutation) -> None: + raw = _history(adjustments=True) + adjustment = raw["adjustments"][0] + if mutation == "digest": + raw["adjustment_receipts"][0]["observed_adjustment_sha256"] = "0" * 64 + elif mutation == "correction-order": + adjustment["correction_observed_at"] = "2026-09-01T00:00:00Z" + elif mutation == "receipt-order": + raw["adjustment_receipts"][0]["observed_at"] = "2026-09-02T00:01:02Z" + elif mutation == "future-receipt": + raw["adjustment_receipts"][0].pop("received_at") + raw["adjustment_receipts"][0]["observed_at"] = "2026-09-03T00:00:00Z" + elif mutation == "accounting-period": + adjustment["accounting_period"]["end"] = adjustment["accounting_period"]["start"] + else: + raw["revisions"][0] = {**_snapshot(), "reporting_revision_id": REVISION_ID} + with pytest.raises(ReportingReconciliationError): + await _load(raw) + + +async def test_mismatching_resolved_account_is_rejected_without_disclosing_it() -> None: + raw = _history() + raw["account_id"] = SECRET + for record in [*raw["periods"], *raw["revisions"]]: + record["account_id"] = SECRET + with pytest.raises(ReportingReconciliationError) as error: + await _load(raw) + assert error.value.code == "LEDGER_SCOPE_MISMATCH" + assert SECRET not in str(error.value) + assert SECRET not in repr(error.value) + assert error.value.__context__ is None + + +async def test_immutable_conflict_and_ledger_repr_do_not_expose_wire_secrets() -> None: + raw = _history() + raw["materializations"][0]["resource"][ + "location" + ] = f"https://user:{SECRET}@storage.example/file" + raw["receipts"][0]["reporting_receipt_id"] = SECRET + ledger = await _load(raw) + assert SECRET not in repr(ledger) + pages = _pages(raw) + replay = deepcopy(pages[-1]) + replay["receipts"][0]["observed_row_count"] += 1 + pages[-1]["pagination"].update(has_more=True, cursor="conflicting-replay") + pages.append(replay) + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(Pages(pages), _request()) + assert error.value.code == "IMMUTABLE_RECORD_CHANGED" + assert SECRET not in str(error.value) + assert SECRET not in repr(error.value) + + +async def test_transport_or_validation_failure_is_closed_and_does_not_restart() -> None: + class PrivateFailure: + calls = 0 + + async def get_reporting_status(self, request): + self.calls += 1 + raise ValueError(f"authorization revoked for {SECRET}") + + client = PrivateFailure() + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger(client, _request()) + assert client.calls == 1 + assert error.value.code == "STATUS_READ_FAILED" + assert SECRET not in str(error.value) + assert SECRET not in repr(error.value) diff --git a/tests/test_reporting_reconciliation.py b/tests/test_reporting_reconciliation.py index 76a61fbdf..6cf1269e3 100644 --- a/tests/test_reporting_reconciliation.py +++ b/tests/test_reporting_reconciliation.py @@ -210,6 +210,9 @@ def _obligation(identifier: str = "obligation-billing") -> dict[str, object]: "successful_materialization_count": 1, "receipt_count": 0, "accepted_receipt_count": 0, + "adjustment_count": 0, + "adjustment_receipt_count": 0, + "accepted_adjustment_receipt_count": 0, "issues": [], "resource_retained_until": "2026-12-01T00:00:00Z", } @@ -294,6 +297,20 @@ def _response(receipts: list[dict[str, object]] | None = None) -> dict[str, obje } +def _full_finality_scope(raw): + # Include a second generation with a snapshot requirement; no obligation + # from that generation is expected in these billing-only vectors. + raw["scope"]["finality"] = ["snapshot", "official"] + raw["scope"]["feed_purposes"].append("pacing") + raw["scope"]["delivery_config_generations"].append( + { + "delivery_config_id": "pacing-feed", + "delivery_config_version": 1, + "feed_purpose": "pacing", + } + ) + + class _Client: def __init__(self) -> None: self.recorded_receipt: dict[str, object] | None = None @@ -1045,12 +1062,11 @@ def test_consumer_billing_mismatch_creates_rejected_receipt() -> None: @pytest.mark.asyncio async def test_missing_expected_period_prevents_definitive_result() -> None: - ledger = await load_reporting_ledger( - _Client(), - GetReportingStatusRequest.model_validate( - {"account": {"account_id": "account-1"}, "view": "periods"} - ), - ) + raw = _response() + # The requested/returned horizon must include the independently expected July period. + raw["scope"]["period_start"] = "2026-07-01T00:00:00Z" + _full_finality_scope(raw) + ledger = await _ledger_from(raw) result = evaluate_reporting_ledger( ledger, expected_periods=[ @@ -1073,12 +1089,10 @@ async def test_missing_expected_period_prevents_definitive_result() -> None: @pytest.mark.asyncio async def test_same_feed_period_cannot_hide_a_missing_campaign() -> None: - ledger = await load_reporting_ledger( - _Client(), - GetReportingStatusRequest.model_validate( - {"account": {"account_id": "account-1"}, "view": "periods"} - ), - ) + raw = _response() + raw["scope"]["media_buy_ids"].append("buy-3") + _full_finality_scope(raw) + ledger = await _ledger_from(raw) result = evaluate_reporting_ledger( ledger, expected_periods=[ @@ -1117,19 +1131,14 @@ async def get_reporting_status( data=GetReportingStatusResponse.model_validate(deepcopy(raw)), ) - ledger = await load_reporting_ledger( - ScopeMismatchClient(), - GetReportingStatusRequest.model_validate( - {"account": {"account_id": "account-1"}, "view": "periods"} - ), - ) - result = evaluate_reporting_ledger( - ledger, - expected_periods=[], - now=datetime.fromisoformat("2026-09-03T00:00:00+00:00"), - ) - assert not result.definitive - assert "REVISION_SCOPE_MISMATCH" in result.obligations[0].reasons + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger( + ScopeMismatchClient(), + GetReportingStatusRequest.model_validate( + {"account": {"account_id": "account-1"}, "view": "periods"} + ), + ) + assert error.value.code == "INVALID_LEDGER_DEPENDENCY" @pytest.mark.asyncio @@ -1207,19 +1216,14 @@ async def get_reporting_status( data=GetReportingStatusResponse.model_validate(deepcopy(raw)), ) - ledger = await load_reporting_ledger( - IncompleteClient(), - GetReportingStatusRequest.model_validate( - {"account": {"account_id": "account-1"}, "view": "periods"} - ), - ) - result = evaluate_reporting_ledger( - ledger, - expected_periods=[], - now=datetime.fromisoformat("2026-09-03T00:00:00+00:00"), - ) - assert not result.definitive - assert "ASSOCIATED_HISTORY_INCOMPLETE" in result.obligations[0].reasons + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger( + IncompleteClient(), + GetReportingStatusRequest.model_validate( + {"account": {"account_id": "account-1"}, "view": "periods"} + ), + ) + assert error.value.code == "LEDGER_COUNT_MISMATCH" @pytest.mark.asyncio @@ -1249,24 +1253,19 @@ async def get_reporting_status( data=GetReportingStatusResponse.model_validate(deepcopy(raw)), ) - ledger = await load_reporting_ledger( - FanoutClient(), - GetReportingStatusRequest.model_validate( - {"account": {"account_id": "account-1"}, "view": "periods"} - ), - ) - result = evaluate_reporting_ledger( - ledger, - expected_periods=[], - now=datetime.fromisoformat("2026-09-03T00:00:00+00:00"), - ) - assert not result.definitive - assert all("REVISION_SCOPE_MISMATCH" in item.reasons for item in result.obligations) + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger( + FanoutClient(), + GetReportingStatusRequest.model_validate( + {"account": {"account_id": "account-1"}, "view": "periods"} + ), + ) + assert error.value.code == "INVALID_LEDGER_DEPENDENCY" @pytest.mark.parametrize("finality", ["official", "snapshot"]) @pytest.mark.parametrize("delivery", ["absent", "pending", "failed", "available"]) -def test_current_publication_is_selected_before_its_materialization( +async def test_current_publication_is_selected_before_its_materialization( finality: str, delivery: str ) -> None: raw = _response() @@ -1306,17 +1305,7 @@ def test_current_publication_is_selected_before_its_materialization( materialization_count=len(raw["materializations"]), successful_materialization_count=2 if delivery == "available" else 1, ) - response = GetReportingStatusResponse.model_validate(raw) - ledger = ReportingLedger( - response.ledger_snapshot_id, - response.ledger_as_of, - response.account_id, - response.scope, - response.periods, - response.revisions, - response.materializations, - response.receipts, - ) + ledger = await _ledger_from(raw) result = evaluate_reporting_ledger( ledger, expected_periods=[], now=datetime.fromisoformat("2026-09-03T00:00:00+00:00") ) @@ -1343,17 +1332,32 @@ def _snapshot_revision(identifier: str, supersedes: str | None = None) -> dict[s return revision -def _ledger_from(raw: dict[str, object]) -> ReportingLedger: +async def _ledger_from(raw: dict[str, object]) -> ReportingLedger: response = GetReportingStatusResponse.model_validate(raw) - return ReportingLedger( - response.ledger_snapshot_id, - response.ledger_as_of, - response.account_id, - response.scope, - response.periods, - response.revisions, - response.materializations, - response.receipts, + response.pagination.total_count = sum( + len(getattr(response, name) or []) + for name in ( + "periods", + "revisions", + "materializations", + "receipts", + "adjustments", + "adjustment_receipts", + "consumer_statuses", + ) + ) + + class FrozenClient: + async def get_reporting_status( + self, request: GetReportingStatusRequest + ) -> TaskResult[GetReportingStatusResponse]: + return TaskResult(status=TaskStatus.COMPLETED, data=response) + + return await load_reporting_ledger( + FrozenClient(), + GetReportingStatusRequest.model_validate( + {"account": {"account_id": response.account_id}, "view": "periods"} + ), ) @@ -1361,7 +1365,9 @@ def _ledger_from(raw: dict[str, object]) -> ReportingLedger: "topology,reason", [("fork", "AMBIGUOUS_REVISION_CHAIN"), ("cycle", "INCOMPLETE_REVISION_CHAIN")], ) -def test_official_close_cannot_mask_a_broken_snapshot_history(topology: str, reason: str) -> None: +async def test_official_close_cannot_mask_a_broken_snapshot_history( + topology: str, reason: str +) -> None: """Snapshot topology is judged on its own, not skipped once an official exists.""" raw = _response() raw["periods"][0].update( @@ -1382,7 +1388,7 @@ def test_official_close_cannot_mask_a_broken_snapshot_history(topology: str, rea ] raw["revisions"] = [*snapshots, deepcopy(REVISION)] result = evaluate_reporting_ledger( - _ledger_from(raw), + await _ledger_from(raw), expected_periods=[], now=datetime.fromisoformat("2026-09-03T00:00:00+00:00"), ) @@ -1390,7 +1396,7 @@ def test_official_close_cannot_mask_a_broken_snapshot_history(topology: str, rea assert reason in result.obligations[0].reasons -def test_identical_scope_obligations_each_keep_their_own_materialized_publication() -> None: +async def test_identical_scope_obligations_each_keep_their_own_materialized_publication() -> None: """``ReportingRevision`` carries no obligation, so use materialization ownership.""" raw = _response() first, second = _obligation("obligation-a"), _obligation("obligation-b") @@ -1420,7 +1426,7 @@ def test_identical_scope_obligations_each_keep_their_own_materialized_publicatio materializations.append(attempt) raw.update(periods=[first, second], revisions=revisions, materializations=materializations) result = evaluate_reporting_ledger( - _ledger_from(raw), + await _ledger_from(raw), expected_periods=[], now=datetime.fromisoformat("2026-09-03T00:00:00+00:00"), ) @@ -1431,7 +1437,7 @@ def test_identical_scope_obligations_each_keep_their_own_materialized_publicatio ] -def test_unowned_publication_never_falls_back_to_an_older_materialized_snapshot() -> None: +async def test_unowned_publication_never_falls_back_to_an_older_materialized_snapshot() -> None: """A newer unmaterialized official wins; an unresolvable owner fails closed.""" raw = _response() first, second = _obligation("obligation-a"), _obligation("obligation-b") @@ -1472,7 +1478,7 @@ def test_unowned_publication_never_falls_back_to_an_older_materialized_snapshot( materializations=[stale, other], ) result = evaluate_reporting_ledger( - _ledger_from(raw), + await _ledger_from(raw), expected_periods=[], now=datetime.fromisoformat("2026-09-03T00:00:00+00:00"), ) @@ -1487,7 +1493,7 @@ def test_unowned_publication_never_falls_back_to_an_older_materialized_snapshot( assert "AMBIGUOUS_REVISION_CHAIN" in unresolved.reasons -def test_native_commit_requires_a_native_version_resource_descriptor() -> None: +async def test_native_commit_requires_a_native_version_resource_descriptor() -> None: """Matching refs and an observed path do not make a mutable location immutable.""" raw = _response() raw["periods"][0].update( @@ -1523,7 +1529,7 @@ def test_native_commit_requires_a_native_version_resource_descriptor() -> None: } raw["materializations"] = [attempt] result = evaluate_reporting_ledger( - _ledger_from(raw), + await _ledger_from(raw), expected_periods=[], now=datetime.fromisoformat("2026-09-03T00:00:00+00:00"), ) @@ -1532,7 +1538,7 @@ def test_native_commit_requires_a_native_version_resource_descriptor() -> None: @pytest.mark.parametrize("finality", ["official", "snapshot"]) -def test_publication_selection_rejects_multiple_current_revisions(finality: str) -> None: +async def test_publication_selection_rejects_multiple_current_revisions(finality: str) -> None: raw = _response() first = deepcopy(REVISION) second = deepcopy(REVISION) @@ -1544,17 +1550,7 @@ def test_publication_selection_rejects_multiple_current_revisions(finality: str) revision.pop(key) raw["periods"][0].update(required_finality=finality, revision_count=2) raw["revisions"] = [first, second] - response = GetReportingStatusResponse.model_validate(raw) - ledger = ReportingLedger( - response.ledger_snapshot_id, - response.ledger_as_of, - response.account_id, - response.scope, - response.periods, - response.revisions, - response.materializations, - response.receipts, - ) + ledger = await _ledger_from(raw) result = evaluate_reporting_ledger(ledger, expected_periods=[]) assert not result.definitive assert "AMBIGUOUS_REVISION_CHAIN" in result.obligations[0].reasons @@ -1716,20 +1712,14 @@ async def get_reporting_status( data=GetReportingStatusResponse.model_validate(deepcopy(raw)), ) - ledger = await load_reporting_ledger( - IncompleteRevisionChainClient(), - GetReportingStatusRequest.model_validate( - {"account": {"account_id": "account-1"}, "view": "periods"} - ), - ) - result = evaluate_reporting_ledger( - ledger, - expected_periods=[], - now=datetime.fromisoformat("2026-09-03T00:00:00+00:00"), - ) - - assert not result.definitive - assert "INCOMPLETE_REVISION_CHAIN" in result.obligations[0].reasons + with pytest.raises(ReportingReconciliationError) as error: + await load_reporting_ledger( + IncompleteRevisionChainClient(), + GetReportingStatusRequest.model_validate( + {"account": {"account_id": "account-1"}, "view": "periods"} + ), + ) + assert error.value.code == "INVALID_LEDGER_DEPENDENCY" @pytest.mark.asyncio