feat(reporting): apply the AdCP 3.3 reporting rules

docs/reporting-release-notes.md

@@ -1,5 +1,63 @@
 # Reliable Reporting upgrade and release notes
 
+## AdCP 3.3 reporting rules
+
+AdCP 3.3 changes several reporting rules
+([#1669](https://github.com/adcontextprotocol/adcp-client-python/issues/1669)).
+Here is what the SDK does for each, and what stays with the code that produces
+delivery.
+
+- **Delivery rows may omit `impressions`.** The generated delivery row types make
+  the field optional. Absence means "not measured", never zero. Only
+  `aggregated_totals` still requires it. `InlineReportingSource` stages such a
+  row without an `impressions` key. It publishes no `impressions` control total
+  unless every staged row carries the metric, so an unmeasured row is never
+  summed as zero. No SDK reader derives `cpm`, `ctr`, `completion_rate` or
+  `viewable_rate` from delivery rows. A seller that has no impression count for
+  a row omits the field, along with that row's impression-denominated rates. In a
+  response served below 3.3, it omits the row instead of sending
+  `impressions: 0`.
+- **Multi-qualifier metrics omit flat totals.** The SDK builds no buy or window
+  `totals`. A seller whose packages report one `metric_id` under more than one
+  qualifier set omits that metric's flat carrier, such as `viewability`, from
+  buy totals, window totals and per-buy webhook totals. Buyers read the package
+  rows instead.
+- **Receivers ignore `aggregated_totals` in delivery webhooks.** `WebhookReceiver`
+  and `ADCPClient.handle_webhook` accept a `media_buy_delivery` result that
+  carries it. Nothing in the SDK reads it. On `WebhookReceiver`'s typed result
+  it is an unmodelled extra; `handle_webhook` returns the result mapping as
+  received. Do not treat it as authoritative. Reconcile from
+  `get_media_buy_delivery` or the per-buy rows. Senders still must not include it.
+- **Property-grain rollups.** Under the `property_delivery_reporting` and
+  `installment_property_delivery_reporting` storyboards, a seller builds
+  `by_property`, `by_installment` and `by_installment_property` only from the
+  rows that `simulate_delivery` injects. The test-controller dispatcher rejects
+  these arrays with `INVALID_PARAMS` when they come with `delivery_date`,
+  because the arrays are undated. It also rejects them when the store's
+  `simulate_delivery` override does not declare the parameter: dropping the rows
+  would report success for rows the seller never received. The store rejects
+  multi-package buys itself.
+- **`source.custody` on report definitions.** A definition's `source` may claim
+  `system_of_record` or `upstream_derived`. The SDK hashes installed definition
+  bytes exactly as given, so `report_definition_sha256` covers the claim.
+  Adding the claim to a published definition therefore needs a new digest.
+  `ReportingRevisionVerifier.custody` returns the claim, or `None` when the
+  definition makes none. A value outside the enum fails construction with
+  `UNSUPPORTED_VERIFICATION`. The buyer-side `ManifestReportingInspector`
+  validates definitions with the generated `ReportingReportDefinition`, which
+  models the field.
+- **Superseded snapshot compaction is not permitted under 3.3.** The notice
+  takes effect no earlier than the first minor after November 20, 2026. The
+  ledger keeps every revision's content for its full retention window, which
+  stays conformant. When compaction becomes permitted, an exact read of a
+  compacted revision must return the same nondisclosing `REFERENCE_NOT_FOUND`
+  as an unknown ID, with `field: "reporting_revision_id"`. The SDK's not-found
+  answers do not set `field` yet. A revision still owed whose content cannot be
+  served stays a `REVISION_UNREADABLE` issue, served as `SERVICE_UNAVAILABLE`,
+  never a not-found. Buyers that need a superseded snapshot's rows should
+  persist them when read, or record a receipt or status statement naming the
+  revision.
+
 ## Protocol-shape control totals
 
 New Core revisions bind `{name, value, value_type[, unit]}` control totals in

src/adcp/reporting/materializer/verification.py

@@ -65,6 +65,7 @@
     object_path,
 )
 from adcp.reporting.revision_selection import select_reporting_revision
+from adcp.types.domains.enums.reporting_custody import ReportingCustody
 
 _VERIFICATION_SEAL_KEY = secrets.token_bytes(32)
 
@@ -167,6 +168,10 @@ class ReportingRevisionVerifier(_ClosedValue):
     (value_type, optional unit, decimal scale). This explicit pinned subset has
     no expression evaluator, custom callbacks, mutable registration or network.
     Unsupported formats and definition semantics fail during construction.
+
+    The definition bytes are hashed exactly as installed, so every field they
+    carry, including AdCP 3.3's optional ``source.custody``, is covered by
+    ``report_definition_sha256``. :attr:`custody` exposes that claim.
     """
 
     key: ReportingVerificationKey
@@ -207,6 +212,7 @@ def _runtime(self) -> _Canonicalizer:
             if hashlib.sha256(raw).hexdigest() != expected.lower():
                 raise failure("UNSUPPORTED_VERIFICATION")
         report = _mapping(parse_reporting_json(self.definition_bytes, self.limits))
+        _definition_custody(report)
         schema = _mapping(parse_reporting_json(self.schema_bytes, self.limits))
         contract = _mapping(parse_reporting_json(self.canonicalization_bytes, self.limits))
         if (
@@ -337,6 +343,22 @@ def canonicalize(
         """Useful to trusted publishers preparing expected evidence before storage."""
         return self._runtime().canonicalize(rows)
 
+    @property
+    def custody(self) -> ReportingCustody | None:
+        """The installed definition's ``source.custody`` claim, or ``None`` if it makes none.
+
+        AdCP 3.3 adds the optional claim: ``system_of_record`` when the producer
+        controls the numbers, ``upstream_derived`` when it obtained them from the
+        provider and system its ``source`` names. The definition digest fixes it
+        for every revision, so no revision carries its own copy. Read it with a
+        revision's ``finality_basis``: an ``upstream_derived`` definition that
+        finalizes by ``contractual_cutoff`` closes figures the producer does not
+        control. Construction already rejected a value outside the enum.
+        """
+        return _definition_custody(
+            _mapping(parse_reporting_json(self.definition_bytes, self.limits))
+        )
+
 
 @dataclass(frozen=True, slots=True)
 class ReportingRevisionVerifierRegistry(_ClosedValue):
@@ -722,6 +744,17 @@ def _mapping(value: object) -> dict[str, Any]:
     return cast(dict[str, Any], value)
 
 
+def _definition_custody(report: dict[str, Any]) -> ReportingCustody | None:
+    """``source.custody`` from a parsed report definition; fail closed on a malformed claim."""
+    source = report.get("source")
+    if type(source) is not dict or "custody" not in source:
+        return None
+    custody = source["custody"]
+    if type(custody) is not str or custody not in {c.value for c in ReportingCustody}:
+        raise failure("UNSUPPORTED_VERIFICATION")
+    return ReportingCustody(custody)
+
+
 def _local_schema(schema: dict[str, Any]) -> None:
     stack: list[object] = [schema]
     while stack:

src/adcp/server/test_controller.py

@@ -169,6 +169,16 @@ def __call__(self, ref: dict[str, Any] | None, *, auth_info: AuthInfo | None = N
 }
 
 
+#: ``simulate_delivery`` params whose rows the seller must report verbatim
+#: (AdCP 3.3): ``by_property``, ``by_installment`` and ``by_installment_property``
+#: are built only from them.
+_INJECTED_DELIVERY_ROWS = (
+    "property_delivery",
+    "installment_delivery",
+    "installment_property_delivery",
+)
+
+
 class TestControllerError(Exception):
     """Typed error for test controller store methods.
 
@@ -491,7 +501,20 @@ async def simulate_delivery(
         ``installment_property_delivery`` (AdCP 3.3) carry the property-grain,
         installment and installment-by-property rows the seller reports under
         ``supports_property_breakdown`` /
-        ``supports_installment_property_breakdown``.
+        ``supports_installment_property_breakdown``. Build ``by_property``,
+        ``by_installment`` and ``by_installment_property`` from exactly these
+        rows (applying ``limit``, ``sort_by`` and ``sort_direction``), never
+        from catalog eligibility. Each array replaces the previous one, is not
+        dated and is not added to totals. Reject a multi-package buy with
+        ``TestControllerError("INVALID_PARAMS", ...)``. The dispatcher already
+        rejects these arrays alongside ``delivery_date``, and rejects them when
+        an override does not declare the parameter, rather than dropping rows
+        the seller would then fail to report.
+
+        ``impressions`` is optional. When a batch carries none, record no
+        impression count rather than zero: AdCP 3.3 delivery rows omit
+        ``impressions`` when it was not measured, and a response served below
+        3.3 omits such a row instead of reporting ``impressions: 0``.
 
         Returns:
             {"simulated": {...}, "cumulative": {...} | None}
@@ -1382,6 +1405,22 @@ async def _handle_test_controller(
                 method_kwargs["result"] = result_value
         elif scenario == "simulate_delivery":
             _require_scenario_params(scenario_params, "media_buy_id")
+            injected = [name for name in _INJECTED_DELIVERY_ROWS if name in scenario_params]
+            if injected and "delivery_date" in scenario_params:
+                # The injected arrays replace the previous rows and are never
+                # dated (AdCP 3.3), so a dated batch cannot carry them.
+                return _controller_error(
+                    "INVALID_PARAMS",
+                    f"{injected[0]} rows are undated; delivery_date is not allowed with them",
+                )
+            unsupported = [name for name in injected if name not in method_kwargs]
+            if unsupported:
+                # The seller must build the breakdown from exactly these rows.
+                # Dropping them would report success for rows it never received.
+                return _controller_error(
+                    "INVALID_PARAMS",
+                    f"{unsupported[0]} is not supported by this agent's simulate_delivery",
+                )
         elif scenario == "force_media_buy_confirmation":
             _require_scenario_params(scenario_params, "action")
             action = scenario_params["action"]