feat(reporting): persist managed delivery reconciliation records

README.md

@@ -12,7 +12,7 @@ This README serves both sides of an AdCP integration. Jump to what you're doing:
 
 - **Connect as a buyer** → [Quick Start: Test Helpers](#quick-start-test-helpers) and [Quick Start: Distributed Operations](#quick-start-distributed-operations). Entry point: `from adcp import ADCPClient, AgentConfig`; start with the `client.simple.*` API.
 - **Build a seller / agent** → [Building an AdCP Agent](#building-an-adcp-agent). Entry point: `from adcp.server import ADCPHandler, serve`; use the [production seller path](docs/production-seller.md) when adding tenants, durable tasks, and webhooks.
-- **Run Reliable Reporting** → [Account currencies](docs/reporting-currency.md) and [ledger migrations](docs/reporting-ledger-migration.md).
+- **Run Reliable Reporting** → [Account currencies](docs/reporting-currency.md), [ledger migrations](docs/reporting-ledger-migration.md), and the [optional reconciliation storage contract](docs/reporting-reconciliation-storage.md).
 - **Understand the type system & imports** → [Type Safety](#type-safety) (import surface, partial modules, cold-start note).
 - **Test against reference agents** → [Quick Start: Test Helpers](#quick-start-test-helpers) and [Test Helpers](#test-helpers). Entry point: `from adcp.testing import test_agent, creative_agent`.
 

docs/reporting-ledger-migration.md

@@ -49,27 +49,34 @@ obligation IDs, the named obligations must exist in the requested account.
    mixing old and new workers is unsafe once accounts reuse a config ID.
 2. With the upgraded SDK, run `await store.create_schema()` before starting
    reporting work. It creates missing tables and applies the bundled
-   `reporting_ledger_account_generations.sql` and
-   `reporting_ledger_obligation_currency.sql` migrations in one transaction.
+   `reporting_ledger_account_generations.sql`,
+   `reporting_ledger_obligation_currency.sql`, and
+   `reporting_ledger_reconciliation.sql` migrations in one transaction.
 3. Restart reporting work with the upgraded SDK on every instance.
 
 For deployments managed by a migration tool, the standalone migration is
 [`reporting_ledger_account_generations.sql`](../src/adcp/reporting/ledger/reporting_ledger_account_generations.sql).
 It upgrades an existing beta.15 ledger by itself, including in autocommit mode.
-For a combined bootstrap and upgrade, run all three bundled files in one transaction:
+For a combined bootstrap and upgrade, run all four bundled files in one transaction:
 
 ```sh
 psql "$REPORTING_DATABASE_URL" --set=ON_ERROR_STOP=1 --single-transaction \
   -f src/adcp/reporting/ledger/reporting_ledger.sql \
   -f src/adcp/reporting/ledger/reporting_ledger_account_generations.sql \
-  -f src/adcp/reporting/ledger/reporting_ledger_obligation_currency.sql
+  -f src/adcp/reporting/ledger/reporting_ledger_obligation_currency.sql \
+  -f src/adcp/reporting/ledger/reporting_ledger_reconciliation.sql
 ```
 
-Use the ledger's existing `search_path` and a role that owns its tables. Both
+Use the ledger's existing `search_path` and a role that owns its tables. All
 SQL migrations are also included as resources in the installed SDK's
 `adcp.reporting.ledger` package. Running only `CREATE TABLE IF NOT EXISTS`
 leaves the old primary key in place and does not perform this upgrade.
 
+The reconciliation migration adds empty optional evidence tables and nullable,
+immutable managed digest/total evidence on revisions and adjustments. It does not infer historical
+evidence or enable a delivery tier. See the [storage contract](reporting-reconciliation-storage.md)
+for the records, migration invariants, and deferred writer/handler work.
+
 The migration inspects the primary-key columns under an exclusive table lock
 and replaces the beta.15 global primary key with the account-qualified key.
 It preserves the constraint's name, including a renamed beta.15 key. Existing

src/adcp/reporting/_reconcile.py

@@ -437,14 +437,26 @@ def _select_current(
     ]
     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 (
-            item.reporting_revision_id in revision_ids
-            if managed_delivery
-            else _revision_matches_obligation(item, obligation)
+            _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
     ]
     receipts = [
         item
@@ -483,19 +495,48 @@ def _select_current(
         if item.supersedes_reporting_revision_id
     }
     candidate_ids = {item.reporting_revision_id for item in candidates}
-    if any(
-        item.supersedes_reporting_revision_id
-        and item.supersedes_reporting_revision_id not in candidate_ids
-        for item in candidates
+    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")
+    by_id = {item.reporting_revision_id: item for item in candidates}
+    leaves = [item for item in candidates if item.reporting_revision_id not in superseded]
+    # Walk every leaf back through its predecessors. A supersession cycle leaves
+    # its members unreachable, so a broken history cannot hide behind an official
+    # close the way a leaf-only count would let it.
+    reachable: set[str] = set()
+    for leaf in leaves:
+        node: ReportingRevision | None = leaf
+        while node is not None and node.reporting_revision_id not in reachable:
+            reachable.add(node.reporting_revision_id)
+            predecessor = node.supersedes_reporting_revision_id
+            node = by_id.get(predecessor) if predecessor else None
+    if (
+        any(
+            item.supersedes_reporting_revision_id
+            and item.supersedes_reporting_revision_id not in candidate_ids
+            for item in candidates
+        )
+        or reachable != candidate_ids
     ):
         reasons.append("INCOMPLETE_REVISION_CHAIN")
-    current = [item for item in candidates if item.reporting_revision_id not in superseded]
+    # Publication selection precedes destination selection. An official close
+    # coexists with retained snapshots; it does not supersede their histories.
+    # A newer unmaterialized publication must never reveal an older snapshot as
+    # the current deliverable merely because that snapshot has a ready resource.
+    official = [item for item in candidates if _enum(item.finality) == "official"]
+    # Snapshot topology is judged on its own. Selecting the official close must
+    # never excuse a forked snapshot history the buyer cannot reconcile.
+    snapshot_leaves = [item for item in leaves if _enum(item.finality) != "official"]
+    if official and len(snapshot_leaves) > 1:
+        reasons.append("AMBIGUOUS_REVISION_CHAIN")
+    current = official or snapshot_leaves
     if len(current) != 1:
         reasons.append("MISSING_CURRENT_REVISION" if not current else "AMBIGUOUS_REVISION_CHAIN")
         return None, None, reasons
     revision = current[0]
-    if any(not _revision_matches_obligation(item, obligation) for item in candidates):
-        reasons.append("REVISION_SCOPE_MISMATCH")
     if (
         not _coverage_is_full(obligation.coverage, obligation.media_buy_ids)
         or obligation.coverage.evaluated_at != obligation.scope_resolved_at
@@ -587,6 +628,9 @@ def _select_current(
         evidence = materialization.verification.native_commit_evidence
         if (
             not evidence
+            # A matching reference and path prove nothing unless the retained
+            # descriptor itself declares the resource immutable by native version.
+            or _enum(materialization.resource.immutability) != "native_version"
             or not materialization.resource.native_version_ref
             or evidence.native_version_ref != materialization.resource.native_version_ref
             or _enum(evidence.observed_through)