feat(reporting): persist buyer receipt submission intents

.github/workflows/ci.yml

@@ -135,13 +135,18 @@ jobs:
   pg-conformance:
     name: Postgres conformance tests (Postgres 16, ${{ matrix.lane }})
     runs-on: ubuntu-latest
-    timeout-minutes: 15
+    timeout-minutes: ${{ matrix.timeout_minutes }}
     permissions:
       contents: read
     strategy:
       fail-fast: false
       matrix:
-        lane: [core, process]
+        include:
+          # Core also builds and installs the reporting wheel/sdist fixtures.
+          - lane: core
+            timeout_minutes: 30
+          - lane: process
+            timeout_minutes: 15
     services:
       postgres:
         # CI-local ephemeral database. POSTGRES_HOST_AUTH_METHOD=trust
@@ -184,8 +189,8 @@ jobs:
           ADCP_PG_TEST_URL: postgresql://postgres@localhost:5432/adcp_test
           PG_LANE: ${{ matrix.lane }}
         run: |
-          # Keep every case and its deadline. Separate process-crash controls
-          # so setup and teardown also fit inside each unchanged job budget.
+          # Keep every case and its per-case deadline. The core lane includes
+          # installed artifact coverage and has a larger overall job budget.
           case "$PG_LANE" in
             core)
               python scripts/reporting_test_harness.py pytest tests/conformance/signing/test_pg_replay_store.py \

src/adcp/reporting/ledger/reporting_buyer_submissions.sql

@@ -0,0 +1,33 @@
+-- Buyer-only additive persistence. Use a dedicated buyer schema/pool.
+-- Never expire/delete a pending intent: uncertainty has no safe time limit.
+CREATE TABLE IF NOT EXISTS reporting_buyer_submission_scopes (
+    scope_sha256 text NOT NULL,
+    canonical_identity text NOT NULL,
+    current_submission_id text,
+    CONSTRAINT reporting_buyer_submission_scopes_pkey PRIMARY KEY (scope_sha256),
+    CONSTRAINT reporting_buyer_scope_digest CHECK (scope_sha256 ~ '^[a-f0-9]{64}$'),
+    CONSTRAINT reporting_buyer_scope_bound CHECK (octet_length(canonical_identity) <= 32768)
+);
+
+CREATE TABLE IF NOT EXISTS reporting_buyer_submission_intents (
+    scope_sha256 text NOT NULL,
+    submission_id text NOT NULL,
+    canonical_plan text NOT NULL,
+    plan_sha256 text NOT NULL,
+    confirmed_results text NOT NULL,
+    confirmed_sha256 text NOT NULL,
+    pending boolean NOT NULL,
+    CONSTRAINT reporting_buyer_submission_intents_pkey PRIMARY KEY (scope_sha256, submission_id),
+    CONSTRAINT reporting_buyer_submission_scope_fk FOREIGN KEY (scope_sha256)
+        REFERENCES reporting_buyer_submission_scopes (scope_sha256),
+    CONSTRAINT reporting_buyer_submission_id CHECK (
+        submission_id ~ '^reporting-submission:[a-f0-9]{64}$'
+    ),
+    CONSTRAINT reporting_buyer_plan_digest CHECK (plan_sha256 ~ '^[a-f0-9]{64}$'),
+    CONSTRAINT reporting_buyer_confirmed_digest CHECK (confirmed_sha256 ~ '^[a-f0-9]{64}$'),
+    CONSTRAINT reporting_buyer_plan_bound CHECK (octet_length(canonical_plan) <= 16777216),
+    CONSTRAINT reporting_buyer_confirmed_bound CHECK (octet_length(confirmed_results) <= 16777216)
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS reporting_buyer_one_pending_scope
+    ON reporting_buyer_submission_intents (scope_sha256) WHERE pending;

src/adcp/reporting/submissions/__init__.py

@@ -0,0 +1,55 @@
+"""Buyer-only durable receipt submission, separate from reconciliation planning.
+
+Public imports are additive and work without the optional PostgreSQL driver.
+The application supplies a trusted authorizer, a receipt client, and an explicit
+intent store. Use PgReportingSubmissionIntentStore for restart durability; the
+memory implementation is a volatile reference for tests. No checkpoint, seller
+service, consumer-status, or client.reporting facade behavior is changed.
+
+Rollout: use a dedicated buyer schema/pool; explicitly call create_schema before
+enabling submissions. The migration adds only reporting_buyer_submission_*.
+Retain pending intents indefinitely and resume them after any uncertain result.
+Do not drop the tables on rollback or replace pending plans with new receipt IDs.
+Disable new planning while recovering, then resume the same stored scope with
+current authorization. Exact final-head review precedes facade integration.
+"""
+
+from adcp.reporting.submissions.models import (
+    ReportingReceiptFailureCode,
+    ReportingReceiptOutcome,
+    ReportingReceiptSubmission,
+    ReportingSubmissionCode,
+    ReportingSubmissionError,
+    ReportingSubmissionReceipt,
+    ReportingSubmissionResult,
+    ReportingSubmissionScope,
+    prepare_reporting_receipt_submission,
+)
+from adcp.reporting.submissions.pg import PgReportingSubmissionIntentStore
+from adcp.reporting.submissions.store import (
+    InMemoryReportingSubmissionIntentStore,
+    ReportingSubmissionIntentStore,
+)
+from adcp.reporting.submissions.submit import (
+    ReportingReceiptSubmissionClient,
+    ReportingSubmissionAuthorizer,
+    submit_reporting_receipts,
+)
+
+__all__ = [
+    "InMemoryReportingSubmissionIntentStore",
+    "PgReportingSubmissionIntentStore",
+    "ReportingReceiptFailureCode",
+    "ReportingReceiptOutcome",
+    "ReportingReceiptSubmission",
+    "ReportingReceiptSubmissionClient",
+    "ReportingSubmissionAuthorizer",
+    "ReportingSubmissionCode",
+    "ReportingSubmissionError",
+    "ReportingSubmissionIntentStore",
+    "ReportingSubmissionReceipt",
+    "ReportingSubmissionResult",
+    "ReportingSubmissionScope",
+    "prepare_reporting_receipt_submission",
+    "submit_reporting_receipts",
+]

src/adcp/reporting/submissions/_validation_cache.py

@@ -0,0 +1,65 @@
+"""Bounded exact-byte validation proofs; no caller-owned objects are retained."""
+
+from __future__ import annotations
+
+import sys
+from collections import OrderedDict
+from threading import RLock
+
+
+class ValidationCache:
+    """An internal LRU of immutable keys and values, bounded in two dimensions.
+
+    Count shared bytes again when sizing entries: this deliberately overcounts
+    retained payloads. The per-entry allowance also covers mapping/lock metadata.
+    A cache hit is equality of the complete bytes, never equality of a digest.
+    Eviction or an oversized entry merely causes full validation on the next use.
+    """
+
+    def __init__(self, *, entries: int, byte_budget: int) -> None:
+        if entries < 1 or byte_budget < 1:
+            raise ValueError("positive validation cache bounds required")
+        self._entries = entries
+        self._byte_budget = byte_budget
+        self._retained_bytes = 0
+        self._values: OrderedDict[tuple[bytes, ...], tuple[tuple[bytes, ...], int]] = OrderedDict()
+        self._lock = RLock()
+
+    def get(self, key: tuple[bytes, ...]) -> tuple[bytes, ...] | None:
+        with self._lock:
+            entry = self._values.get(key)
+            if entry is None:
+                return None
+            self._values.move_to_end(key)
+            return entry[0]
+
+    def put(self, key: tuple[bytes, ...], value: tuple[bytes, ...]) -> None:
+        if (
+            type(key) is not tuple
+            or type(value) is not tuple
+            or any(type(part) is not bytes for part in (*key, *value))
+        ):
+            raise TypeError("immutable validation proof required")
+        size = 1024 + sum(sys.getsizeof(parts) for parts in (key, value))
+        size += sum(sys.getsizeof(part) for part in (*key, *value))
+        with self._lock:
+            previous = self._values.pop(key, None)
+            if previous is not None:
+                self._retained_bytes -= previous[1]
+            if size > self._byte_budget:
+                return
+            while self._values and (
+                len(self._values) >= self._entries
+                or self._retained_bytes + size > self._byte_budget
+            ):
+                _, (_, released) = self._values.popitem(last=False)
+                self._retained_bytes -= released
+            self._values[key] = value, size
+            self._retained_bytes += size
+
+
+# At most 64 MiB of conservatively counted proofs across both caches. They are
+# performance aids, not durable state, and contain neither auth grants nor raw
+# seller diagnostics. A cold process still validates the complete stored bytes.
+PLANS = ValidationCache(entries=16, byte_budget=32 * 1024 * 1024)
+CONFIRMATIONS = ValidationCache(entries=256, byte_budget=32 * 1024 * 1024)