fix(reporting): validate buyer frozen histories

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

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(

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