test(reporting): integrate currency and metric evidence

README.md

@@ -357,6 +357,7 @@ forward traffic degrades gracefully rather than failing.
 - **[Handler authoring](docs/handler-authoring.md)** - Building an AdCP-compliant agent on `adcp.server`
 - **[Production seller path](docs/production-seller.md)** - Choose the server abstraction and wire durable multi-tenant tasks and webhook delivery
 - **[Validation contract](docs/validation-contract.md)** - Canonical wire validation versus structural Pydantic models
+- **[Reporting source adapters](docs/reporting-source-adapters.md)** - Per-metric availability evidence, coverage, and control totals for inline delivery fetches
 - **[Migrating from SDK 6 to 7](https://github.com/adcontextprotocol/adcp-client-python/blob/main/MIGRATION_v6_to_v7.md)** - Breaking API, security, concurrency, and webhook changes
 - **[Migrating from SDK 7 to 8](https://github.com/adcontextprotocol/adcp-client-python/blob/main/MIGRATION_v7_to_v8.md)** - Secure webhook defaults and telemetry changes
 - **[Migrating from AdCP 3.1 to 3.2 beta](MIGRATION_ADCP_3.1_TO_3.2.md)** - Compact lifecycle adoption and old/new compatibility matrix

docs/reporting-source-adapters.md

@@ -0,0 +1,192 @@
+# Per-metric evidence for inline reporting sources
+
+`InlineReportingSource` wraps a synchronous or asynchronous delivery fetch in a
+sealed, replayable reporting publication. Return `InlineFetchResult` when metric
+availability differs within a constituent. Its `cell_availability` map is keyed
+by **request constituent ID, then metric name**. Constituent IDs are not
+necessarily media buy IDs; read them from `request.coverage.constituents`.
+
+```python
+from adcp.reporting.inline_source import InlineFetchResult, MetricEvidence
+
+return InlineFetchResult(
+    rows=[{"constituent_id": constituent_id, "impressions": 120, "clicks": 0}],
+    cell_availability={
+        constituent_id: {
+            "impressions": MetricEvidence.present(data_through=watermark),
+            "clicks": MetricEvidence.explicit_zero(),
+            "viewability": MetricEvidence.delayed(
+                "measurement_pending", data_through=viewability_watermark
+            ),
+            "completed_views": MetricEvidence.unavailable("not_video_inventory"),
+        },
+    },
+)
+```
+
+This publishes one `partial` constituent with four independent metric cells.
+It can complete a request whose `coverage.expected` is `partial`. A `full`
+request receives retryable `PARTIAL_RESULT` and no publication until every
+requested cell is available.
+
+| Constructor | Meaning and required evidence |
+| --- | --- |
+| `MetricEvidence.present(data_through)` | Measured values through a timezone-aware watermark. Every matched row for the constituent must carry a non-null value for this metric. |
+| `MetricEvidence.explicit_zero(data_through=...)` | An observed zero. The watermark defaults to the fetch watermark. Any supplied values must be finite numeric zeros. |
+| `MetricEvidence.missing(reason)` | No answer for this metric. A stable reason is required; a watermark is forbidden. |
+| `MetricEvidence.delayed(reason, data_through=...)` | Not ready yet. A stable reason is required; retain a known watermark when available. |
+| `MetricEvidence.unavailable(reason)` | The source cannot measure this metric for this constituent. Emits the existing wire status `unsupported`; a reason is required and a watermark is forbidden. |
+
+Evidence is immutable. Direct `MetricEvidence(...)` construction enforces the
+same invariants. Reasons use the existing manifest format: bounded ASCII,
+at most 512 characters, beginning and ending with an alphanumeric character.
+Use redacted explanations such as `not_video_inventory`, not provider error
+payloads. Available statuses do not accept reasons.
+
+The adapter supplies availability, watermarks, and measurements. The SDK alone
+copies semantic-contract ID, version, and digest from the **selected offering**.
+`MetricEvidence` has no semantic-contract fields. `cell_availability` is an
+adapter input; the sealed manifest continues to use its existing
+`metric_availability` list and statuses.
+
+## Defaults and freshness
+
+Omit `cell_availability` (or pass `None` or `{}`) to retain the existing derived
+behavior. Omitted cells inherit their constituent's status, reason, and
+watermark. Explicit cells take precedence over `covered_constituent_ids` and
+`unavailable_constituents`, including explicit measurements for an otherwise
+missing constituent. Coverage is then reconciled from the resolved cells:
+uniform statuses stay uniform, present plus explicit-zero is `present`, and
+other mixtures are `partial`.
+
+Explicit watermarks are bounded by period end, source read cutoff, and the
+observation instant. A watermark before the period is rejected. An explicit
+cell may advance the batch watermark while other cells keep their earlier
+evidence. A fully available constituent uses its earliest cell watermark.
+For an authoritative publication, an available cell whose bounded watermark
+does not reach period end becomes `delayed`, with a reason and its watermark.
+Cell evidence cannot bypass that freshness gate.
+
+Rows can omit missing, unsupported, or delayed metrics; available measurements
+for other metrics are retained. A control total is a checksum over the staged
+rows -- a consumer recomputes it from the revision's rows -- so it covers every
+staged row, including rows outside the requested coverage, which stay staged
+with a warning. It is emitted when every staged row carries a valid finite
+numeric value for the metric. Missing fields, nulls, booleans, and invalid
+numbers prevent a total, and an omitted explicit-zero row value is not filled
+in. Statuses the SDK derives on its own withdraw nothing: a result with no
+`cell_availability` publishes exactly the totals it published before.
+
+### The withdrawal invariant
+
+A declared `missing`, `delayed`, or `unsupported` cell asserts that the source
+produced **no measurement** for it. A control total is the exact sum of a metric
+over every staged row, so the only question a withdrawal raises is whether it
+can put a disclaimed value in that sum. Two shapes follow, and they are
+deliberately not treated alike:
+
+- **A withdrawal by a constituent that staged rows removes that metric's
+  total.** Its own rows carry values the adapter has disclaimed, so summing
+  them would contradict the evidence it supplied. The rows stay staged
+  byte-for-byte -- an adapter may publish a provider payload unchanged and
+  declare per cell which of its columns are measurements.
+- **A withdrawal by a constituent that staged no rows removes nothing.** It
+  reaches no sum, so the checksum over the rows that *were* measured is
+  retained. A buy that delivered nothing and whose billing is pending is the
+  answer per-metric evidence exists to give; withdrawing its neighbours'
+  subtotal would destroy a checksum the consumer recomputes.
+
+A batch with no rows at all is the one case where a withdrawal removes a total
+on its own: a `0` there would publish unavailability as an observed zero.
+Unavailability is otherwise carried by the cell's own evidence, never by a
+missing total.
+
+### Monetary columns have no third option
+
+`spend`, plus every metric the trusted report definition froze through
+`monetary_metric_units` / `monetary_control_total_units`, is reconciled by the
+obligation ledger against the rows a revision retains. For those columns the
+first shape above has nowhere to go: dropping the total makes the ledger refuse
+the revision with `MONETARY_TOTAL_MISMATCH` on the immutable replay of every
+retry, and keeping it would sum a value the adapter disclaims. So a withdrawn
+monetary cell whose own constituent's rows report that metric raises
+`ValueError` before anything is staged or sealed. Omit the metric from those
+rows, or declare the cell measured.
+
+The frozen slice request cannot carry those unit declarations --
+`ReportingDefinitionBinding.to_wire()` keeps them off the wire so retained
+contract hashes do not move -- so tell the adapter which columns they are:
+
+```python
+InlineReportingSource(
+    capabilities=capabilities,
+    fetch=fetch,
+    monetary_metrics=[name for name, _ in obligation.definition.monetary_metric_units],
+)
+```
+
+`spend` is always included. A custom money column that is **not** declared here
+is treated as non-monetary, which can still wedge the obligation for that
+column -- declare it whenever the obligation's definition does.
+
+The existing zero-row wire rule is unchanged: an empty batch must be wholly
+explicit-zero or wholly unavailable. It cannot mix available cells with
+unavailable cells. To report a measured zero alongside an unavailable metric,
+supply the source's normalized zero measurement row. The adapter does not
+manufacture rows or silently convert unavailable metrics to zero.
+
+## Common patterns
+
+For complete source data, the existing row-list shorthand still works. To
+declare its freshness explicitly:
+
+```python
+return InlineFetchResult.all_present(rows, data_through=watermark)
+```
+
+This uses the constituent defaults: constituents with rows are present and
+covered constituents without rows are observed zeros. Bare `[]` still means
+an observed zero; `None` still means not ready. Existing positional
+`InlineFetchResult` arguments retain their meaning, and so do their control
+totals -- only an explicitly declared cell withdraws one, under the
+[withdrawal invariant](#the-withdrawal-invariant).
+
+Bulk helpers return maps to pass to `cell_availability`, leaving the result's
+other options available:
+
+```python
+from adcp.reporting.inline_source import (
+    metric_delayed_through,
+    metric_unsupported_everywhere,
+)
+
+return InlineFetchResult(
+    rows=rows,
+    cell_availability=metric_unsupported_everywhere(
+        request, "completed_views", "not_video_inventory"
+    ),
+)
+
+# Or retain a delayed metric's watermark across every requested constituent:
+return InlineFetchResult(
+    rows=rows,
+    cell_availability=metric_delayed_through(
+        request, "viewability", watermark, reason="measurement_pending"
+    ),
+)
+```
+
+Only keys in the frozen requested matrix are accepted, even if the selected
+offering declares additional metrics. Unknown keys, duplicate entries
+exposed by a mapping, and non-`MetricEvidence` values raise `ValueError` before
+staging or sealing. Evidence construction errors inside a fetch also propagate
+as `ValueError`; they are not classified as transient provider failures.
+Keys are not coerced or normalized. Ordinary Python dicts already discard
+repeated keys; reject duplicates while parsing provider input if it can contain
+them.
+
+See the [sync and async type-check examples](../tests/type_checks/reporting_inline_source.py)
+and the [conformance tests](../tests/conformance/reporting/test_inline_cell_availability.py).
+Run `run_reporting_source_replay_conformance` against your adapter and staging
+store to verify that the same execution key returns the original sealed
+evidence even if source measurements or availability later change.

src/adcp/reporting/conformance.py

@@ -110,6 +110,7 @@ async def _with_deadline(
     *,
     deadline_at: datetime,
     cancel: asyncio.Event,
+    clock: Callable[[], datetime] | None = None,
 ) -> _T:
     """Run ``operation`` under the slice deadline, cancelling cooperatively first.
 
@@ -118,7 +119,8 @@ async def _with_deadline(
     and settle.  An executor that ignores the event is then hard-cancelled --
     the harness stays bounded even when the thing it is grading does not.
     """
-    remaining = (_utc(deadline_at) - datetime.now(timezone.utc)).total_seconds()
+    now = clock() if clock is not None else datetime.now(timezone.utc)
+    remaining = (_utc(deadline_at) - _utc(now)).total_seconds()
     if cancel.is_set() or remaining <= 0:
         raise _fail(
             "EXECUTION_FAILED",
@@ -438,12 +440,14 @@ async def validate_reporting_source_execution(
     result: ReportingSourceExecutorResult,
     object_reader: ReportingSourceStagedObjectReader,
     cancel: asyncio.Event | None = None,
+    clock: Callable[[], datetime] | None = None,
 ) -> SourceBatchManifestV1:
     """Validate one execution end to end and return its verified manifest.
 
     Reads every staged object the manifest names and checks its bytes against
     the declared digest and size.  A manifest whose objects cannot be read, or
     read differently than claimed, is not evidence of anything.
+    ``clock`` permits deterministic deadline checks for retained test slices.
     """
     offering = _validate_request_against_capabilities(capabilities, request)
     if not result.ok:
@@ -477,6 +481,7 @@ async def validate_reporting_source_execution(
                 ),
                 deadline_at=request.deadline_at,
                 cancel=cancel,
+                clock=clock,
             )
         except ReportingSourceConformanceError:
             raise
@@ -499,13 +504,15 @@ async def run_reporting_source_replay_conformance(
     request: ReportingSourceSliceRequestV1,
     object_reader: ReportingSourceStagedObjectReader,
     cancel: asyncio.Event | None = None,
+    clock: Callable[[], datetime] | None = None,
 ) -> SourceBatchManifestV1:
     """Execute the same slice twice and require an identical immutable publication.
 
     Reusing a ``source_execution_key`` must return byte-identical manifest bytes
     and the same staged object set.  This is the check that catches the two most
     common non-conformances: a ``now()`` timestamp baked into the manifest, and a
-    fresh UUID minted per attempt.
+    fresh UUID minted per attempt. ``clock`` supplies the deadline-check instant
+    for both executions and their staged-object reads.
     """
     capabilities = executor.capabilities
     cancel = cancel or asyncio.Event()
@@ -515,13 +522,15 @@ async def once() -> tuple[ReportingSourceExecutorResult, SourceBatchManifestV1]:
             lambda: executor.execute(request, cancel=cancel),
             deadline_at=request.deadline_at,
             cancel=cancel,
+            clock=clock,
         )
         manifest = await validate_reporting_source_execution(
             capabilities=capabilities,
             request=request,
             result=result,
             object_reader=object_reader,
             cancel=cancel,
+            clock=clock,
         )
         return result, manifest