diff --git a/README.md b/README.md index eb5b48340..544e20492 100644 --- a/README.md +++ b/README.md @@ -356,6 +356,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 diff --git a/docs/reporting-source-adapters.md b/docs/reporting-source-adapters.md new file mode 100644 index 000000000..da2d07ad5 --- /dev/null +++ b/docs/reporting-source-adapters.md @@ -0,0 +1,142 @@ +# 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. Declaring any cell of a metric `missing`, `delayed`, or `unsupported` +withdraws that metric's total even when rows still carry values, because the +adapter has said those values are not a measurement. Statuses the SDK derives +on its own withdraw nothing: a result with no `cell_availability` publishes +exactly the totals it published before. + +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. + +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. diff --git a/src/adcp/reporting/inline_source.py b/src/adcp/reporting/inline_source.py index 4c8eb7550..5b9209530 100644 --- a/src/adcp/reporting/inline_source.py +++ b/src/adcp/reporting/inline_source.py @@ -12,6 +12,13 @@ conforming ``basic`` manifest, complete with staged objects, coverage evidence, and byte-identical replay. +For uneven metric support, return :class:`InlineFetchResult` with +``cell_availability={constituent_id: {metric_name: MetricEvidence(...)}}``. +Omitted cells retain the constituent defaults. Explicit evidence controls each +cell independently; mixed cells make the constituent partial, and a metric +column an adapter declares incomplete receives no control total. Metric +semantics always come from the selected SDK offering. + The three answers a fetch can give --------------------------------- @@ -58,9 +65,12 @@ from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path -from typing import Any, Protocol, TypeAlias, runtime_checkable +from typing import Any, Literal, Protocol, TypeAlias, runtime_checkable + +from pydantic import TypeAdapter, ValidationError from adcp.reporting.source import ( + EvidenceReason, MediaBuyConstituentV1, PackageItemConstituentV1, ProductConstituentV1, @@ -92,9 +102,12 @@ "InlineFetch", "InlineFetchResult", "InlineReportingSource", + "MetricEvidence", "ReportingSealStore", "ReportingStagingStore", "SealedSlice", + "metric_delayed_through", + "metric_unsupported_everywhere", ] @@ -103,6 +116,90 @@ # -------------------------------------------------------------------------- +_REASON = TypeAdapter(EvidenceReason) +_AVAILABLE = frozenset({"present", "explicit_zero"}) + + +class _CellEvidenceError(ValueError): + """A deterministic adapter error, never an unclassified provider fault.""" + + +@dataclass(frozen=True) +class MetricEvidence: + """Source evidence for one requested constituent/metric cell. + + This carries availability only. The SDK binds metric semantics from the + selected offering. Prefer the named constructors; direct construction + enforces the same invariants and raises ``ValueError`` for invalid evidence. + + Watermarks must be timezone-aware. Reasons use the manifest's bounded ASCII + evidence format; use stable, redacted explanations such as + ``not_video_inventory`` rather than provider diagnostics or credentials. + """ + + status: Literal["present", "explicit_zero", "missing", "delayed", "unsupported"] + data_through: datetime | None = None + reason: str | None = None + + def __post_init__(self) -> None: + if self.status not in ("present", "explicit_zero", "missing", "delayed", "unsupported"): + raise _CellEvidenceError("MetricEvidence.status is not a supported cell status") + if self.data_through is not None: + if ( + not isinstance(self.data_through, datetime) + or self.data_through.tzinfo is None + or self.data_through.utcoffset() is None + ): + raise _CellEvidenceError( + "MetricEvidence.data_through must be a timezone-aware datetime" + ) + if self.status == "present" and self.data_through is None: + raise _CellEvidenceError("MetricEvidence present requires data_through") + if self.status in _AVAILABLE: + if self.reason is not None: + raise _CellEvidenceError(f"MetricEvidence {self.status} must not carry a reason") + else: + if self.reason is None: + raise _CellEvidenceError(f"MetricEvidence {self.status} requires a stable reason") + try: + _REASON.validate_python(self.reason, strict=True) + except ValidationError as error: + raise _CellEvidenceError( + "MetricEvidence.reason must be a wire-valid stable reason" + ) from error + if self.status in {"missing", "unsupported"} and self.data_through is not None: + raise _CellEvidenceError(f"MetricEvidence {self.status} must not carry data_through") + + @classmethod + def present(cls, data_through: datetime) -> MetricEvidence: + """The source measured this metric through the supplied watermark.""" + return cls("present", data_through=data_through) + + @classmethod + def explicit_zero(cls, *, data_through: datetime | None = None) -> MetricEvidence: + """An observed zero; omit the watermark to inherit the fetch watermark. + + Any supplied row values for this cell must be numeric zeros. An + omitted row value is not converted into a control-total contribution. + """ + return cls("explicit_zero", data_through=data_through) + + @classmethod + def missing(cls, reason: str) -> MetricEvidence: + """The source returned no answer for this cell.""" + return cls("missing", reason=reason) + + @classmethod + def delayed(cls, reason: str, *, data_through: datetime | None = None) -> MetricEvidence: + """The metric is not ready; retain its watermark when one is known.""" + return cls("delayed", data_through=data_through, reason=reason) + + @classmethod + def unavailable(cls, reason: str) -> MetricEvidence: + """The source cannot measure this cell (wire status ``unsupported``).""" + return cls("unsupported", reason=reason) + + @dataclass(frozen=True) class InlineFetchResult: """A fetch answer richer than a bare row list. @@ -149,6 +246,66 @@ class InlineFetchResult: warnings: Sequence[str] = () """Safe, redacted operator notes retained on the publication.""" + cell_availability: Mapping[str, Mapping[str, MetricEvidence]] | None = None + """Optional ``{constituent_id: {metric_name: evidence}}`` overrides. + + Keys are the frozen request's constituent IDs, which need not equal media + buy IDs. Omitted cells retain the derived constituent status and watermark. + Explicit cells override even missing/unsupported constituent defaults; + mixed availability promotes the constituent to ``partial``. The SDK still + applies the source cutoff, observation ceiling, and authoritative freshness + gate. A cell watermark may advance the batch watermark without advancing + other cells. This is an adapter surface, not the manifest's wire-format + ``metric_availability`` list. + + Unknown keys, duplicate mapping entries, invalid evidence, and contradictory + explicit zeros raise ``ValueError`` before staging. A zero-row batch must + be wholly explicit-zero or wholly unavailable under the existing contract. + """ + + @classmethod + def all_present( + cls, rows: Sequence[Mapping[str, Any]], *, data_through: datetime + ) -> InlineFetchResult: + """Cover the whole request through a watermark, using the legacy defaults. + + Constituents with rows are present; covered constituents without rows + are observed zeros. Use ``cell_availability`` for exceptions. + """ + return cls(rows=rows, data_through=data_through) + + +def metric_unsupported_everywhere( + request: ReportingSourceSliceRequestV1, metric_name: str, reason: str +) -> dict[str, dict[str, MetricEvidence]]: + """Build ``cell_availability`` for a metric unsupported across the request. + + Other metrics retain the fetch's constituent defaults. Pass the returned + map to :class:`InlineFetchResult`, alongside rows, watermarks, or warnings. + """ + return _metric_everywhere(request, metric_name, MetricEvidence.unavailable(reason)) + + +def metric_delayed_through( + request: ReportingSourceSliceRequestV1, + metric_name: str, + watermark: datetime, + *, + reason: str = "The reporting source has not finalized this metric yet", +) -> dict[str, dict[str, MetricEvidence]]: + """Build ``cell_availability`` retaining a delayed metric's known watermark.""" + return _metric_everywhere( + request, metric_name, MetricEvidence.delayed(reason, data_through=watermark) + ) + + +def _metric_everywhere( + request: ReportingSourceSliceRequestV1, metric_name: str, evidence: MetricEvidence +) -> dict[str, dict[str, MetricEvidence]]: + if metric_name not in request.requested_metrics: + raise _CellEvidenceError(f"cell_availability has unknown requested metric {metric_name!r}") + return {item.constituent_id: {metric_name: evidence} for item in request.coverage.constituents} + #: What an inline fetch may return. ``None`` is "not ready". InlineFetchReturn: TypeAlias = "InlineFetchResult | Sequence[Mapping[str, Any]] | None | object" @@ -456,6 +613,11 @@ async def execute( answer = _coerce(answer) except ReportingSourceError as error: return ReportingSourceExecutorResult.failed(error.error) + except _CellEvidenceError: + # Constructors/helpers may run inside the fetch. Keep the same + # actionable ValueError as matrix validation after the fetch, + # rather than redacting an adapter bug as a retryable provider fault. + raise except asyncio.CancelledError: raise except Exception as error: @@ -547,6 +709,7 @@ async def _publish( ) -> ReportingSourceExecutorResult: observed_at = self._clock() data_through = _clamp_watermark(request, result.data_through, observed_at) + overrides = _validate_cell_availability(request, result.cell_availability) covered = ( {item.constituent_id for item in request.coverage.constituents} if result.covered_constituent_ids is None @@ -555,7 +718,7 @@ async def _publish( covered -= set(result.unavailable_constituents) rows_by_constituent: dict[str, list[Mapping[str, Any]]] = { - constituent_id: [] for constituent_id in covered + item.constituent_id: [] for item in request.coverage.constituents } unmatched = 0 for row in result.rows: @@ -564,6 +727,8 @@ async def _publish( unmatched += 1 continue rows_by_constituent[constituent_id].append(row) + if constituent_id not in covered and not overrides.get(constituent_id): + unmatched += 1 warnings = list(result.warnings) if unmatched: @@ -575,7 +740,19 @@ async def _publish( statuses = self._derive_statuses( request, result, covered, rows_by_constituent, data_through ) - coverage_status = _roll_up(statuses.values()) + constituents, cells = self._resolve_availability( + request, + result, + overrides=overrides, + statuses=statuses, + rows_by_constituent=rows_by_constituent, + data_through=data_through, + observed_at=observed_at, + ) + # Granular watermarks are bounded by the batch watermark on the wire. + # Advancing it for an explicit cell must not advance fallback cells. + data_through = max(data_through, *(cell.data_through or data_through for cell in cells)) + coverage_status = _roll_up([item.status for item in constituents]) if request.coverage.expected == "full" and coverage_status != "full": # Never publish a partial answer to a full-coverage slice; the @@ -591,6 +768,31 @@ async def _publish( ) ) + # A partial constituent is deliberately excluded: a known zero for + # one metric must never turn unavailable neighbours into a batch zero. + explicit_zero = not result.rows and all(cell.status == "explicit_zero" for cell in cells) + if ( + not result.rows + and not explicit_zero + and any(cell.status in _AVAILABLE for cell in cells) + ): + # Reachable from a derived result too, so the message names the + # shape rather than the optional field an adopter may never have set. + raise _CellEvidenceError( + "a zero-row batch cannot mix available and unavailable cells; supply rows " + "for the observed metrics or withdraw their availability" + ) + # Only a *declared* exception withdraws a column. A control total is + # a checksum over the staged rows -- consumers recompute it from the + # revision's rows -- so an unmatched row or a legacy-derived + # unavailable constituent leaves it verifiable and unchanged. + declared_incomplete = { + cell.metric + for cell in cells + if cell.status not in _AVAILABLE + and cell.metric in overrides.get(cell.constituent_id, {}) + } + control_totals = _control_totals(request, result.rows, declared_incomplete) payload = _encode_rows(result.rows) object_ref, object_generation = await self._staging.stage( account_id=request.identity.account_id, @@ -609,23 +811,17 @@ async def _publish( row_count=len(result.rows), ) - available = { - constituent_id - for constituent_id, status in statuses.items() - if status in {"present", "explicit_zero"} - } - explicit_zero = bool(available) and not result.rows and available == set(statuses) manifest = self._seal_manifest( request, observed_at=observed_at, data_through=data_through, staged=staged, - statuses=statuses, + constituents=constituents, + cells=cells, coverage_status=coverage_status, - reasons=dict(result.unavailable_constituents), explicit_zero=explicit_zero, row_count=len(result.rows), - control_totals=_control_totals(request, result.rows), + control_totals=control_totals, warnings=warnings, ) @@ -678,6 +874,107 @@ def _derive_statuses( statuses[constituent_id] = "explicit_zero" return statuses + def _resolve_availability( + self, + request: ReportingSourceSliceRequestV1, + result: InlineFetchResult, + *, + overrides: Mapping[str, Mapping[str, MetricEvidence]], + statuses: Mapping[str, ReportingAvailabilityStatus], + rows_by_constituent: Mapping[str, Sequence[Mapping[str, Any]]], + data_through: datetime, + observed_at: datetime, + ) -> tuple[list[ReportingConstituentCoverageV1], list[ReportingMetricAvailabilityV1]]: + """Resolve cells first, then derive coverage without widening their claims.""" + offering = self._capabilities.offering(request.offering_id) + declared = {metric.name: metric for metric in offering.metrics} + constituents: list[ReportingConstituentCoverageV1] = [] + cells: list[ReportingMetricAvailabilityV1] = [] + for item in request.coverage.constituents: + constituent_id = item.constituent_id + fallback = statuses[constituent_id] + fallback_reason = ( + None + if fallback in _AVAILABLE + else result.unavailable_constituents.get(constituent_id) + or _DEFAULT_REASONS[fallback] + ) + constituent_cells: list[ReportingMetricAvailabilityV1] = [] + for metric in request.requested_metrics: + status = fallback + watermark = data_through if status in _AVAILABLE else None + reason = fallback_reason + explicit = overrides.get(constituent_id, {}).get(metric) + if explicit is not None: + _validate_cell_rows( + constituent_id, metric, explicit, rows_by_constituent[constituent_id] + ) + status, reason = explicit.status, explicit.reason + watermark = explicit.data_through + if watermark is not None: + if _utc(watermark) < _utc(request.period.start): + raise _CellEvidenceError( + f"cell_availability ({constituent_id!r}, {metric!r}) " + "data_through must not precede the reporting period" + ) + watermark = _clamp_watermark(request, watermark, observed_at) + elif status == "explicit_zero": + watermark = data_through + if ( + status in _AVAILABLE + and request.publication_class == "AUTHORITATIVE" + and watermark is not None + and _utc(watermark) < _utc(request.period.end) + ): + status, reason = "delayed", _DEFAULT_REASONS["delayed"] + # Evidence follows the cell status, never its constituent's + # roll-up. Only the selected SDK offering supplies semantics. + constituent_cells.append( + ReportingMetricAvailabilityV1( + constituent_id=constituent_id, + metric=metric, + semantic_contract_id=declared[metric].semantic_contract_id, + semantic_contract_version=declared[metric].semantic_contract_version, + semantic_contract_sha256=declared[metric].semantic_contract_sha256, + status=status, + data_through=watermark, + reason=reason, + ) + ) + cells.extend(constituent_cells) + status = fallback + watermark = data_through if status in _AVAILABLE else None + reason = fallback_reason + if overrides.get(constituent_id): + cell_statuses = {cell.status for cell in constituent_cells} + if len(cell_statuses) == 1: + status = constituent_cells[0].status + elif cell_statuses <= _AVAILABLE: + status = "present" + else: + status = "partial" + if status in _AVAILABLE: + watermark = min( + cell.data_through + for cell in constituent_cells + if cell.data_through is not None + ) + reason = None + else: + watermark = None + reasons = {cell.reason for cell in constituent_cells} + reason = ( + constituent_cells[0].reason + if len(reasons) == 1 and constituent_cells[0].reason is not None + else _DEFAULT_REASONS[status] + ) + constituents.append( + ReportingConstituentCoverageV1( + constituent=item, status=status, data_through=watermark, reason=reason + ) + ) + return constituents, cells + def _seal_manifest( self, request: ReportingSourceSliceRequestV1, @@ -685,49 +982,19 @@ def _seal_manifest( observed_at: datetime, data_through: datetime, staged: SourceBatchObjectV1, - statuses: Mapping[str, ReportingAvailabilityStatus], + constituents: list[ReportingConstituentCoverageV1], + cells: list[ReportingMetricAvailabilityV1], coverage_status: str, - reasons: Mapping[str, str], explicit_zero: bool, row_count: int, control_totals: list[SourceControlTotalV1], warnings: list[str], ) -> SourceBatchManifestV1: - available = {"present", "explicit_zero"} - - def evidence(constituent_id: str) -> dict[str, Any]: - status = statuses[constituent_id] - if status in available: - return {"data_through": data_through} - return {"reason": reasons.get(constituent_id) or _DEFAULT_REASONS[status]} - coverage = SourceBatchCoverageV1( denominator_fingerprint=request.coverage.denominator_fingerprint, status=coverage_status, # type: ignore[arg-type] - constituents=[ - ReportingConstituentCoverageV1( - constituent=item, - status=statuses[item.constituent_id], - **evidence(item.constituent_id), - ) - for item in request.coverage.constituents - ], + constituents=constituents, ) - offering = self._capabilities.offering(request.offering_id) - declared = {metric.name: metric for metric in offering.metrics} - cells = [ - ReportingMetricAvailabilityV1( - constituent_id=item.constituent_id, - metric=metric, - semantic_contract_id=declared[metric].semantic_contract_id, - semantic_contract_version=declared[metric].semantic_contract_version, - semantic_contract_sha256=declared[metric].semantic_contract_sha256, - status=statuses[item.constituent_id], - **evidence(item.constituent_id), - ) - for item in request.coverage.constituents - for metric in request.requested_metrics - ] finality_at = ( max(_utc(observed_at), _utc(request.period.end)) if request.publication_class == "AUTHORITATIVE" @@ -797,6 +1064,83 @@ def evidence(constituent_id: str) -> dict[str, Any]: # -------------------------------------------------------------------------- +def _validate_cell_availability( + request: ReportingSourceSliceRequestV1, + availability: Mapping[str, Mapping[str, MetricEvidence]] | None, +) -> dict[str, dict[str, MetricEvidence]]: + """Validate and snapshot overrides before any staging or seal side effects. + + Do not coerce keys or collapse mappings into dicts before checking them: + custom Mapping implementations can expose duplicate logical entries. + Ordinary dicts have already discarded duplicate keys before this boundary. + """ + if availability is None: + return {} + if not isinstance(availability, Mapping): + raise _CellEvidenceError("cell_availability must be a nested mapping") + requested_ids = {item.constituent_id for item in request.coverage.constituents} + requested_metrics = set(request.requested_metrics) + validated: dict[str, dict[str, MetricEvidence]] = {} + for constituent_id, metrics in availability.items(): + if not isinstance(constituent_id, str) or constituent_id not in requested_ids: + raise _CellEvidenceError( + f"cell_availability has unknown constituent_id {constituent_id!r}" + ) + if constituent_id in validated: + raise _CellEvidenceError( + f"cell_availability has duplicate constituent_id {constituent_id!r}" + ) + if not isinstance(metrics, Mapping): + raise _CellEvidenceError( + f"cell_availability for {constituent_id!r} must be a metric mapping" + ) + validated[constituent_id] = {} + for metric, evidence in metrics.items(): + if not isinstance(metric, str) or metric not in requested_metrics: + raise _CellEvidenceError( + f"cell_availability has unknown requested metric {metric!r}" + ) + if metric in validated[constituent_id]: + raise _CellEvidenceError( + f"cell_availability has duplicate cell ({constituent_id!r}, {metric!r})" + ) + if not isinstance(evidence, MetricEvidence): + raise _CellEvidenceError( + f"cell_availability ({constituent_id!r}, {metric!r}) must be MetricEvidence" + ) + validated[constituent_id][metric] = evidence + return validated + + +def _validate_cell_rows( + constituent_id: str, + metric: str, + evidence: MetricEvidence, + rows: Sequence[Mapping[str, Any]], +) -> None: + """Reject explicit assertions contradicted by the supplied measurements. + + Legacy derived cells remain permissive. Metric value types are defined by + the offering, but an explicit present cell needs a value and a claimed zero + must not conceal a nonzero, nonnumeric, or non-finite measurement. + """ + label = f"cell_availability ({constituent_id!r}, {metric!r})" + if evidence.status == "present" and (not rows or any(row.get(metric) is None for row in rows)): + raise _CellEvidenceError(f"{label} present requires a value in every constituent row") + if evidence.status == "explicit_zero": + for row in rows: + if row.get(metric) is None: + continue + try: + value = _decimal(row[metric]) + except (InvalidOperation, TypeError, ValueError) as error: + raise _CellEvidenceError( + f"{label} explicit_zero requires numeric zero row values" + ) from error + if not value.is_finite() or value != 0: + raise _CellEvidenceError(f"{label} explicit_zero requires numeric zero row values") + + def _now() -> datetime: moment = datetime.now(timezone.utc) return moment.replace(microsecond=(moment.microsecond // 1000) * 1000) @@ -857,16 +1201,26 @@ def _encode_rows(rows: Sequence[Mapping[str, Any]]) -> bytes: def _control_totals( - request: ReportingSourceSliceRequestV1, rows: Sequence[Mapping[str, Any]] + request: ReportingSourceSliceRequestV1, + rows: Sequence[Mapping[str, Any]], + declared_incomplete: Collection[str], ) -> list[SourceControlTotalV1]: - """Sum each requested metric across the rows, exactly. + """Sum each requested metric across the staged rows, exactly. Money sums through :class:`~decimal.Decimal` and is emitted as a string. A float total would round differently in two languages and turn a consumer's equality check into a flake. + + ``declared_incomplete`` names the metrics an adapter explicitly withdrew + for at least one cell. Those columns may still carry row values, but the + adapter has said they are not a measurement, so totalling them would + contradict the very evidence it supplied. Every other column keeps the + checksum a consumer recomputes from the revision's rows. """ totals: list[SourceControlTotalV1] = [] for metric in request.requested_metrics: + if metric in declared_incomplete: + continue values = [row[metric] for row in rows if row.get(metric) is not None] if len(values) != len(rows): # A metric absent from some rows has no honest total; the diff --git a/tests/conformance/reporting/test_inline_cell_availability.py b/tests/conformance/reporting/test_inline_cell_availability.py new file mode 100644 index 000000000..98014a6e0 --- /dev/null +++ b/tests/conformance/reporting/test_inline_cell_availability.py @@ -0,0 +1,865 @@ +"""Adapter evidence must survive sealing, admission, and replay without overclaiming.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Iterator, Mapping +from datetime import datetime, timedelta, timezone +from typing import Any + +import pytest + +from adcp.reporting.conformance import ( + run_reporting_source_replay_conformance, + validate_reporting_source_execution, + validate_reporting_source_failure, +) +from adcp.reporting.fixtures import ( + redacted_authoritative_request, + redacted_capabilities, + redacted_snapshot_request, +) +from adcp.reporting.inline_source import ( + InlineFetchResult, + InlineReportingSource, + InMemorySealStore, + InMemoryStagingStore, + MetricEvidence, + metric_delayed_through, + metric_unsupported_everywhere, +) +from adcp.reporting.source import ( + MediaBuyConstituentV1, + MetricOfferingV1, + ReportingSourceCapabilitiesV1, + ReportingSourceCoverageRequestV1, + ReportingSourceSliceRequestV1, + SourceBatchManifestV1, + coverage_denominator_fingerprint_v1, + reporting_source_capabilities_sha256_v1, +) + +CID = "campaign-redacted-1" # Deliberately different from media_buy_id. +SECOND_CID = "campaign-redacted-2" +METRICS = ["impressions", "clicks", "viewability", "completed_views"] +OBSERVED_AT = datetime(2026, 11, 6, 12, tzinfo=timezone.utc) +ROW = { + "media_buy_id": "media-buy-redacted", + "impressions": 10, + "clicks": 0, + "viewability": "0.75", + "completed_views": 2, +} + + +def _capabilities() -> ReportingSourceCapabilitiesV1: + base = redacted_capabilities() + offerings = [ + offering.model_copy( + update={ + "metrics": [ + MetricOfferingV1( + name=metric, + semantic_contract_id=f"fixture.{offering.offering_id}.{metric}", + semantic_contract_version=str(index + 1), + semantic_contract_sha256=str(index + 1) * 64, + ) + for index, metric in enumerate([*METRICS, "spend"]) + ] + } + ) + for offering in base.offerings + ] + draft = base.model_copy(update={"offerings": offerings}) + return ReportingSourceCapabilitiesV1.model_validate( + { + **draft.model_dump(), + "capabilities_sha256": reporting_source_capabilities_sha256_v1(draft), + } + ) + + +def _request( + *, + authoritative: bool = False, + second: bool = False, + metrics: list[str] | None = None, +) -> ReportingSourceSliceRequestV1: + base = redacted_authoritative_request() if authoritative else redacted_snapshot_request() + constituents = list(base.coverage.constituents) + if second: + constituents.append( + MediaBuyConstituentV1( + constituent_id=SECOND_CID, + media_buy_id="media-buy-second", + product_id="fixture-product", + ) + ) + return base.model_copy( + update={ + "requested_metrics": metrics or METRICS, + "coverage": ReportingSourceCoverageRequestV1( + expected="partial", + constituents=constituents, + denominator_fingerprint=coverage_denominator_fingerprint_v1(constituents), + ), + } + ) + + +def _source(fetch: Any, **kwargs: Any) -> InlineReportingSource: + return InlineReportingSource( + capabilities=_capabilities(), fetch=fetch, clock=lambda: OBSERVED_AT, **kwargs + ) + + +async def _seal( + answer: Any, request: ReportingSourceSliceRequestV1 | None = None +) -> SourceBatchManifestV1: + request = request or _request() + source = _source(lambda req: answer) + return await validate_reporting_source_execution( + capabilities=source.capabilities, + request=request, + result=await source.execute(request, cancel=asyncio.Event()), + object_reader=source.staging, + ) + + +@pytest.mark.parametrize("authoritative", [False, True], ids=["snapshot", "official"]) +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +async def test_one_constituent_has_four_independent_metric_statuses( + authoritative: bool, asynchronous: bool +) -> None: + request = _request(authoritative=authoritative) + watermark = min(request.period.end, request.period.source_read_cutoff_at) + delayed_through = watermark - timedelta(hours=4) + cases = [ + ("impressions", MetricEvidence.present(watermark), "present", watermark, None), + ("clicks", MetricEvidence.explicit_zero(), "explicit_zero", watermark, None), + ( + "viewability", + MetricEvidence.delayed("measurement_pending", data_through=delayed_through), + "delayed", + delayed_through, + "measurement_pending", + ), + ( + "completed_views", + MetricEvidence.unavailable("not_video_inventory"), + "unsupported", + None, + "not_video_inventory", + ), + ] + rows = [{key: value for key, value in ROW.items() if key != "completed_views"}] + answer = InlineFetchResult( + rows=rows, + cell_availability={CID: {metric: evidence for metric, evidence, *_ in cases}}, + ) + + async def fetch(req: ReportingSourceSliceRequestV1) -> InlineFetchResult: + return answer + + source = _source(fetch if asynchronous else lambda req: answer) + manifest = await run_reporting_source_replay_conformance( + executor=source, request=request, object_reader=source.staging + ) + assert manifest.coverage.status == "partial" + assert manifest.coverage.constituents[0].status == "partial" + assert manifest.coverage.constituents[0].reason + assert manifest.explicit_zero is False + assert manifest.row_count == 1 + assert {total.name: total.value for total in manifest.control_totals} == { + "impressions": "10", + "clicks": "0", + } + cells = {cell.metric: cell for cell in manifest.metric_availability} + declared = { + metric.name: metric for metric in source.capabilities.offering(request.offering_id).metrics + } + for metric, _, status, through, reason in cases: + cell = cells[metric] + assert (cell.status, cell.data_through, cell.reason) == (status, through, reason) + contract = declared[metric] + assert ( + cell.semantic_contract_id, + cell.semantic_contract_version, + cell.semantic_contract_sha256, + ) == ( + contract.semantic_contract_id, + contract.semantic_contract_version, + contract.semantic_contract_sha256, + ) + + +@pytest.mark.parametrize( + "fallback", ["present", "missing", "unsupported", "delayed", "stale", "partial"] +) +async def test_explicit_cells_override_constituent_defaults(fallback: str) -> None: + request = _request() + fields: dict[str, Any] = {} + if fallback == "missing": + fields["covered_constituent_ids"] = () + elif fallback != "present": + fields.update( + unavailable_constituents={CID: "constituent_default"}, unavailable_status=fallback + ) + explicit = ( + MetricEvidence.unavailable("metric_override") + if fallback == "present" + else MetricEvidence.present(request.period.source_read_cutoff_at) + ) + manifest = await _seal( + InlineFetchResult(rows=[ROW], cell_availability={CID: {"impressions": explicit}}, **fields), + request, + ) + cells = {cell.metric: cell for cell in manifest.metric_availability} + assert cells["impressions"].status == explicit.status + assert cells["impressions"].reason == explicit.reason + assert {cells[metric].status for metric in METRICS[1:]} == {fallback} + assert manifest.coverage.constituents[0].status == "partial" + + +async def test_explicit_cells_can_replace_all_missing_constituent_defaults() -> None: + request = _request() + manifest = await _seal( + InlineFetchResult( + rows=[ROW], + covered_constituent_ids=(), + cell_availability={ + CID: { + metric: MetricEvidence.present(request.period.source_read_cutoff_at) + for metric in METRICS + } + }, + ), + request, + ) + assert manifest.coverage.status == "full" + assert not manifest.warnings + assert len(manifest.control_totals) == len(METRICS) + + +async def test_present_and_explicit_zero_are_full_coverage() -> None: + manifest = await _seal( + InlineFetchResult( + rows=[ROW], cell_availability={CID: {"clicks": MetricEvidence.explicit_zero()}} + ) + ) + assert manifest.coverage.status == "full" + assert manifest.coverage.constituents[0].status == "present" + assert manifest.explicit_zero is False + + +@pytest.mark.parametrize("status", ["missing", "delayed", "unsupported"]) +async def test_uniform_unavailability_does_not_become_partial_or_zero(status: str) -> None: + evidence = MetricEvidence(status=status, reason="source_reason") + manifest = await _seal( + InlineFetchResult( + rows=[], cell_availability={CID: {metric: evidence for metric in METRICS}} + ) + ) + assert manifest.coverage.status == "none" + assert manifest.coverage.constituents[0].status == status + assert manifest.coverage.constituents[0].reason == "source_reason" + assert not manifest.explicit_zero + assert not manifest.control_totals + + +async def test_zero_row_mixed_unavailability_has_no_manufactured_measurements() -> None: + manifest = await _seal( + InlineFetchResult( + rows=[], + covered_constituent_ids=(), + cell_availability={ + CID: {"completed_views": MetricEvidence.unavailable("not_video_inventory")} + }, + ) + ) + assert manifest.coverage.status == "partial" + assert {cell.status for cell in manifest.metric_availability} == {"missing", "unsupported"} + assert manifest.row_count == 0 + assert not manifest.explicit_zero + # The declared exception withdraws its column. The derived cells keep the + # zero-row totals a legacy unavailable batch has always published, which + # the wire rule constrains to zero rather than forbidding. + assert {total.name: total.value for total in manifest.control_totals} == { + "impressions": "0", + "clicks": "0", + "viewability": "0", + } + + +async def test_full_request_with_mixed_cells_fails_without_staging() -> None: + request = _request() + request = request.model_copy( + update={"coverage": request.coverage.model_copy(update={"expected": "full"})} + ) + source = _source( + lambda req: InlineFetchResult( + rows=[ROW], + cell_availability=metric_unsupported_everywhere( + req, "completed_views", "not_video_inventory" + ), + ), + staging=_NoStaging(), + ) + error = validate_reporting_source_failure( + await source.execute(request, cancel=asyncio.Event()), "PARTIAL_RESULT" + ) + assert error.retry == "retryable" + + +class _NoStaging(InMemoryStagingStore): + async def stage(self, **kwargs: Any) -> tuple[str, str]: + pytest.fail("invalid evidence must be rejected before staging") + + +class _RepeatedMapping(Mapping[str, Any]): + """A mapping that exposes repeated entries before a dict would discard them.""" + + def __init__(self, key: str, value: Any) -> None: + self.key, self.value = key, value + + def __getitem__(self, key: str) -> Any: + return self.value + + def __iter__(self) -> Iterator[str]: + return iter([self.key, self.key]) + + def __len__(self) -> int: + return 2 + + +@pytest.mark.parametrize( + ("availability", "error"), + [ + ({"media-buy-redacted": {}}, "unknown constituent_id 'media-buy-redacted'"), + ({"other": {}}, "unknown constituent_id 'other'"), + ({CID: {"spend": MetricEvidence.explicit_zero()}}, "unknown requested metric 'spend'"), + ({CID: {"other": MetricEvidence.explicit_zero()}}, "unknown requested metric 'other'"), + ({(CID, "clicks"): MetricEvidence.explicit_zero()}, "unknown constituent_id"), + ({CID: {42: MetricEvidence.explicit_zero()}}, "unknown requested metric 42"), + ([], "must be a nested mapping"), + ({CID: []}, "must be a metric mapping"), + ( + { + CID: { + "clicks": { + "status": "explicit_zero", + "semantic_contract_id": "adapter_override", + } + } + }, + "must be MetricEvidence", + ), + ({CID: {"clicks": None}}, "must be MetricEvidence"), + ({CID: _RepeatedMapping("clicks", MetricEvidence.explicit_zero())}, "duplicate cell"), + ( + _RepeatedMapping(CID, {"clicks": MetricEvidence.explicit_zero()}), + "duplicate constituent_id", + ), + ], +) +async def test_invalid_matrices_fail_clearly_before_staging_or_sealing( + availability: Any, error: str +) -> None: + request = _request() + seals = InMemorySealStore() + source = _source( + lambda req: InlineFetchResult(rows=[ROW], cell_availability=availability), + staging=_NoStaging(), + seals=seals, + ) + with pytest.raises(ValueError, match=error): + await source.execute(request, cancel=asyncio.Event()) + assert ( + await seals.get( + account_id=request.identity.account_id, + source_execution_key=request.identity.source_execution_key, + ) + is None + ) + + +@pytest.mark.parametrize("value", [1, -1, "0.1", True, False, "not_numeric", "NaN", "Infinity"]) +async def test_explicit_zero_cannot_hide_a_contradictory_row_value(value: Any) -> None: + source = _source( + lambda req: InlineFetchResult( + rows=[{**ROW, "clicks": value}], + cell_availability={CID: {"clicks": MetricEvidence.explicit_zero()}}, + ), + staging=_NoStaging(), + ) + with pytest.raises(ValueError, match="explicit_zero requires numeric zero row values"): + await source.execute(_request(), cancel=asyncio.Event()) + + +@pytest.mark.parametrize( + "rows", + [ + [], + [{"media_buy_id": "media-buy-redacted"}], + [{**ROW, "impressions": None}], + [ROW, {"media_buy_id": "media-buy-redacted"}], + ], +) +async def test_explicit_present_requires_values_in_its_constituent_rows(rows: Any) -> None: + source = _source( + lambda req: InlineFetchResult( + rows=rows, + cell_availability={ + CID: {"impressions": MetricEvidence.present(req.period.source_read_cutoff_at)} + }, + ), + staging=_NoStaging(), + ) + with pytest.raises(ValueError, match="present requires a value in every constituent row"): + await source.execute(_request(), cancel=asyncio.Event()) + + +async def test_zero_row_available_and_unavailable_mix_is_rejected_without_a_wire_change() -> None: + source = _source( + lambda req: InlineFetchResult( + rows=[], + cell_availability={ + CID: {"completed_views": MetricEvidence.unavailable("not_video_inventory")} + }, + ), + staging=_NoStaging(), + ) + with pytest.raises( + ValueError, match="a zero-row batch cannot mix available and unavailable cells" + ): + await source.execute(_request(), cancel=asyncio.Event()) + + +async def test_the_zero_row_mixing_error_does_not_blame_an_unused_field() -> None: + """A derived result reaches the same guard without ever naming a cell.""" + source = _source( + lambda req: InlineFetchResult(rows=[], covered_constituent_ids=[CID]), + staging=_NoStaging(), + ) + with pytest.raises(ValueError, match="a zero-row batch cannot mix") as caught: + await source.execute(_request(second=True), cancel=asyncio.Event()) + assert "cell_availability" not in str(caught.value) + + +@pytest.mark.parametrize("status", ["present", "explicit_zero"]) +async def test_cell_watermarks_do_not_bypass_the_authoritative_freshness_gate(status: str) -> None: + request = _request(authoritative=True) + through = request.period.end - timedelta(hours=4) + manifest = await _seal( + InlineFetchResult( + rows=[ROW], + cell_availability={ + CID: {"clicks": MetricEvidence(status=status, data_through=through)} + }, + ), + request, + ) + cell = next(cell for cell in manifest.metric_availability if cell.metric == "clicks") + assert (cell.status, cell.data_through) == ("delayed", through) + assert cell.reason + assert manifest.coverage.status == "partial" + assert "clicks" not in {total.name for total in manifest.control_totals} + + +@pytest.mark.parametrize("authoritative", [False, True]) +async def test_newer_explicit_watermark_does_not_advance_fallback_cells( + authoritative: bool, +) -> None: + request = _request(authoritative=authoritative) + through = min(request.period.end, request.period.source_read_cutoff_at) + fallback_through = through - timedelta(hours=4) + manifest = await _seal( + InlineFetchResult( + rows=[ROW], + data_through=fallback_through, + cell_availability={CID: {"impressions": MetricEvidence.present(through)}}, + ), + request, + ) + cells = {cell.metric: cell for cell in manifest.metric_availability} + assert manifest.data_through == through + assert cells["impressions"].data_through == through + assert cells["impressions"].status == "present" + assert cells["clicks"].status == ("delayed" if authoritative else "present") + assert cells["clicks"].data_through == (None if authoritative else fallback_through) + assert manifest.coverage.constituents[0].data_through == ( + None if authoritative else fallback_through + ) + + +async def test_explicit_watermarks_are_bounded_by_the_frozen_cutoff_and_observation() -> None: + request = _request() + observed_at = request.period.source_read_cutoff_at - timedelta(hours=1) + source = InlineReportingSource( + capabilities=_capabilities(), + fetch=lambda req: InlineFetchResult( + rows=[ROW], + cell_availability={ + CID: {"impressions": MetricEvidence.present(req.period.end + timedelta(days=1))} + }, + ), + clock=lambda: observed_at, + ) + manifest = await run_reporting_source_replay_conformance( + executor=source, request=request, object_reader=source.staging + ) + assert manifest.data_through == observed_at + assert manifest.metric_availability[0].data_through == observed_at + + +async def test_explicit_watermarks_before_the_period_are_not_promoted_to_fresh_data() -> None: + source = _source( + lambda req: InlineFetchResult( + rows=[ROW], + cell_availability={ + CID: { + "impressions": MetricEvidence.present(req.period.start - timedelta(seconds=1)) + } + }, + ), + staging=_NoStaging(), + ) + with pytest.raises(ValueError, match="data_through must not precede the reporting period"): + await source.execute(_request(), cancel=asyncio.Event()) + + +@pytest.mark.parametrize("status", ["missing", "delayed", "unsupported"]) +async def test_control_total_needs_every_requested_cell_even_when_rows_have_values( + status: str, +) -> None: + request = _request(second=True) + manifest = await _seal( + InlineFetchResult( + rows=[ROW, {**ROW, "media_buy_id": "media-buy-second"}], + cell_availability={ + SECOND_CID: { + "completed_views": MetricEvidence(status=status, reason="source_reason") + } + }, + ), + request, + ) + assert {total.name: total.value for total in manifest.control_totals} == { + "impressions": "20", + "clicks": "0", + "viewability": "1.50", + } + + +@pytest.mark.parametrize("value", [None, True, "NaN", "Infinity", "not_numeric"]) +async def test_complete_cell_status_does_not_make_invalid_values_totalable(value: Any) -> None: + manifest = await _seal(InlineFetchResult(rows=[{**ROW, "viewability": value}])) + assert "viewability" not in {total.name for total in manifest.control_totals} + + +async def test_omitted_unavailable_metrics_do_not_suppress_other_metric_totals() -> None: + request = _request(second=True) + manifest = await _seal( + InlineFetchResult( + rows=[ + ROW, + { + **{key: value for key, value in ROW.items() if key != "completed_views"}, + "media_buy_id": "media-buy-second", + }, + ], + cell_availability={ + SECOND_CID: {"completed_views": MetricEvidence.missing("not_returned")} + }, + ), + request, + ) + assert {total.name for total in manifest.control_totals} == set(METRICS) - {"completed_views"} + + +async def test_unmatched_rows_are_warned_about_and_stay_in_the_staged_checksum() -> None: + manifest = await _seal([ROW, {**ROW, "media_buy_id": "not_requested"}]) + assert manifest.row_count == 2 + assert manifest.warnings + # A control total is recomputed from the revision's rows, so it must cover + # every staged row -- including one no constituent claimed. + assert {total.name: total.value for total in manifest.control_totals} == { + "impressions": "20", + "clicks": "0", + "viewability": "1.50", + "completed_views": "4", + } + + +async def test_a_derived_missing_constituent_does_not_withdraw_its_neighbour_totals() -> None: + request = _request(second=True) + manifest = await _seal(InlineFetchResult(rows=[ROW], covered_constituent_ids=[CID]), request) + assert manifest.coverage.status == "partial" + assert manifest.coverage.constituents[1].status == "missing" + assert {total.name: total.value for total in manifest.control_totals} == { + "impressions": "10", + "clicks": "0", + "viewability": "0.75", + "completed_views": "2", + } + + +async def test_a_covered_zero_row_constituent_keeps_zeros_only_for_its_available_metrics() -> None: + request = _request(second=True) + manifest = await _seal( + InlineFetchResult( + rows=[ROW], + cell_availability={ + SECOND_CID: {"completed_views": MetricEvidence.unavailable("not_video_inventory")} + }, + ), + request, + ) + assert manifest.coverage.constituents[1].status == "partial" + assert not manifest.explicit_zero + for cell in manifest.metric_availability: + if cell.constituent_id == SECOND_CID: + assert cell.status == ( + "unsupported" if cell.metric == "completed_views" else "explicit_zero" + ) + assert {total.name: total.value for total in manifest.control_totals} == { + "impressions": "10", + "clicks": "0", + "viewability": "0.75", + } + + +async def test_explicit_zero_evidence_for_every_cell_can_seal_a_real_empty_period() -> None: + manifest = await _seal( + InlineFetchResult( + rows=[], + cell_availability={CID: {metric: MetricEvidence.explicit_zero() for metric in METRICS}}, + ) + ) + assert manifest.explicit_zero + assert manifest.coverage.status == "full" + assert manifest.coverage.constituents[0].status == "explicit_zero" + assert {total.name: total.value for total in manifest.control_totals} == dict.fromkeys( + METRICS, "0" + ) + + +async def test_declared_zero_without_a_row_value_does_not_manufacture_a_total() -> None: + manifest = await _seal( + InlineFetchResult( + rows=[{key: value for key, value in ROW.items() if key != "clicks"}], + cell_availability={CID: {"clicks": MetricEvidence.explicit_zero()}}, + ) + ) + assert manifest.metric_availability[1].status == "explicit_zero" + assert "clicks" not in {total.name for total in manifest.control_totals} + + +@pytest.mark.parametrize("pattern", ["unsupported", "delayed"]) +async def test_bulk_helpers_cover_only_the_requested_metric(pattern: str) -> None: + request = _request(second=True) + through = request.period.source_read_cutoff_at - timedelta(hours=4) + overrides = ( + metric_unsupported_everywhere(request, "completed_views", "not_video_inventory") + if pattern == "unsupported" + else metric_delayed_through(request, "completed_views", through) + ) + manifest = await _seal( + InlineFetchResult( + rows=[ROW, {**ROW, "media_buy_id": "media-buy-second"}], cell_availability=overrides + ), + request, + ) + for cell in manifest.metric_availability: + if cell.metric == "completed_views": + assert cell.status == pattern + assert cell.reason + assert cell.data_through == (through if pattern == "delayed" else None) + else: + assert cell.status == "present" + + +async def test_replay_uses_sealed_cell_evidence_after_the_adapter_changes() -> None: + request = _request() + overrides = metric_unsupported_everywhere(request, "completed_views", "not_video_inventory") + rows = [dict(ROW)] + calls = 0 + + def fetch(req: ReportingSourceSliceRequestV1) -> InlineFetchResult: + nonlocal calls + calls += 1 + return InlineFetchResult(rows=rows, cell_availability=overrides) + + source = _source(fetch) + first = await source.execute(request, cancel=asyncio.Event()) + overrides[CID]["completed_views"] = MetricEvidence.present(request.period.source_read_cutoff_at) + rows[0]["completed_views"] = 999 + cancel = asyncio.Event() + cancel.set() + replay = await source.execute(request, cancel=cancel) + assert first.ok and replay.ok + assert calls == 1 + assert replay.manifest_bytes == first.manifest_bytes + assert replay.response == first.response + manifest = await validate_reporting_source_execution( + capabilities=source.capabilities, + request=request, + result=replay, + object_reader=source.staging, + ) + assert manifest.metric_availability[-1].status == "unsupported" + + +async def test_mapping_order_does_not_change_sealed_identity() -> None: + request = _request(second=True) + overrides = { + CID: { + "clicks": MetricEvidence.explicit_zero(), + "completed_views": MetricEvidence.unavailable("not_video_inventory"), + }, + SECOND_CID: {"clicks": MetricEvidence.explicit_zero()}, + } + rows = [ROW, {**ROW, "media_buy_id": "media-buy-second"}] + first = _source(lambda req: InlineFetchResult(rows=rows, cell_availability=overrides)) + reverse = { + cid: dict(reversed(list(metrics.items()))) + for cid, metrics in reversed(list(overrides.items())) + } + second = _source(lambda req: InlineFetchResult(rows=rows, cell_availability=reverse)) + a = await first.execute(request, cancel=asyncio.Event()) + b = await second.execute(request, cancel=asyncio.Event()) + assert a.ok and b.ok + assert a.manifest_bytes == b.manifest_bytes + + +async def test_status_and_reason_are_bound_by_the_content_fingerprint() -> None: + fingerprints = set() + for evidence in [ + MetricEvidence.unavailable("not_video_inventory"), + MetricEvidence.unavailable("not_measured"), + MetricEvidence.missing("not_measured"), + ]: + manifest = await _seal( + InlineFetchResult(rows=[ROW], cell_availability={CID: {"completed_views": evidence}}) + ) + fingerprints.add(manifest.content_fingerprint) + assert len(fingerprints) == 3 + + +@pytest.mark.parametrize("rows", [[ROW], []]) +async def test_legacy_rows_results_and_empty_overrides_seal_identically(rows: Any) -> None: + request = _request() + results = [] + for answer in [ + rows, + InlineFetchResult(rows), + InlineFetchResult(rows, cell_availability={}), + InlineFetchResult(rows, cell_availability={CID: {}}), + InlineFetchResult.all_present(rows, data_through=request.period.source_read_cutoff_at), + ]: + source = _source(lambda req: answer) + result = await source.execute(request, cancel=asyncio.Event()) + assert result.ok + results.append(result.manifest_bytes) + assert all(payload == results[0] for payload in results) + + +@pytest.mark.parametrize( + ("answer", "second", "totals"), + [ + pytest.param( + [ROW, {**ROW, "media_buy_id": "not_requested"}], + False, + {"impressions": "20", "clicks": "0", "viewability": "1.50", "completed_views": "4"}, + id="unmatched_row", + ), + pytest.param( + InlineFetchResult(rows=[ROW], covered_constituent_ids=[CID]), + True, + {"impressions": "10", "clicks": "0", "viewability": "0.75", "completed_views": "2"}, + id="uncovered_constituent", + ), + pytest.param( + InlineFetchResult(rows=[ROW], unavailable_constituents={SECOND_CID: "no_data"}), + True, + {"impressions": "10", "clicks": "0", "viewability": "0.75", "completed_views": "2"}, + id="unavailable_constituent", + ), + pytest.param( + InlineFetchResult( + rows=[ROW, {**ROW, "media_buy_id": "media-buy-second"}], + unavailable_constituents={SECOND_CID: "late"}, + unavailable_status="delayed", + ), + True, + {"impressions": "20", "clicks": "0", "viewability": "1.50", "completed_views": "4"}, + id="delayed_constituent", + ), + pytest.param( + None, + False, + {"impressions": "0", "clicks": "0", "viewability": "0", "completed_views": "0"}, + id="not_ready", + ), + ], +) +async def test_results_without_declared_cells_keep_their_legacy_control_totals( + answer: Any, second: bool, totals: dict[str, str] +) -> None: + """A caller that never opted in must not lose evidence it already published. + + These are the shapes where the derived constituent status is not uniformly + available. Withdrawing their totals would change sealed bytes, the content + fingerprint, and the ledger revision content hash for adopters who supplied + no ``cell_availability`` at all. + """ + manifest = await _seal(answer, _request(second=second)) + assert {total.name: total.value for total in manifest.control_totals} == totals + + +async def test_five_independent_statuses_seal_in_one_constituent_matrix() -> None: + metrics = [*METRICS, "spend"] + request = _request(metrics=metrics) + watermark = min(request.period.end, request.period.source_read_cutoff_at) + delayed_through = watermark - timedelta(hours=4) + expected = { + "impressions": ("present", watermark, None), + "clicks": ("explicit_zero", watermark, None), + "viewability": ("delayed", delayed_through, "measurement_pending"), + "completed_views": ("unsupported", None, "not_video_inventory"), + "spend": ("missing", None, "not_returned"), + } + manifest = await _seal( + InlineFetchResult( + rows=[{"media_buy_id": "media-buy-redacted", "impressions": 10, "clicks": 0}], + cell_availability={ + CID: { + "impressions": MetricEvidence.present(watermark), + "clicks": MetricEvidence.explicit_zero(), + "viewability": MetricEvidence.delayed( + "measurement_pending", data_through=delayed_through + ), + "completed_views": MetricEvidence.unavailable("not_video_inventory"), + "spend": MetricEvidence.missing("not_returned"), + } + }, + ), + request, + ) + assert manifest.coverage.constituents[0].status == "partial" + assert { + cell.metric: (cell.status, cell.data_through, cell.reason) + for cell in manifest.metric_availability + } == expected + assert {total.name for total in manifest.control_totals} == {"impressions", "clicks"} + + +async def test_existing_positional_result_arguments_keep_their_meaning() -> None: + request = _request() + result = InlineFetchResult( + [ROW], request.period.source_read_cutoff_at, None, {}, "unsupported", ("legacy_note",) + ) + assert result.cell_availability is None + manifest = await _seal(result, request) + assert manifest.coverage.status == "full" + assert manifest.warnings == ["legacy_note"] diff --git a/tests/test_reporting_metric_evidence.py b/tests/test_reporting_metric_evidence.py new file mode 100644 index 000000000..c3b860d0b --- /dev/null +++ b/tests/test_reporting_metric_evidence.py @@ -0,0 +1,136 @@ +"""Construction-time contracts for the inline adapter's typed evidence surface.""" + +import asyncio +from dataclasses import FrozenInstanceError +from datetime import datetime, timezone +from typing import Any + +import pytest + +from adcp.reporting.conformance import validate_reporting_source_failure +from adcp.reporting.fixtures import redacted_capabilities, redacted_snapshot_request +from adcp.reporting.inline_source import ( + InlineFetchResult, + InlineReportingSource, + MetricEvidence, + metric_delayed_through, + metric_unsupported_everywhere, +) + +WATERMARK = datetime(2026, 11, 1, 12, tzinfo=timezone.utc) + + +@pytest.mark.parametrize( + ("fields", "message"), + [ + ({"status": "unavailable"}, "status is not a supported cell status"), + ({"status": "partial", "reason": "partial"}, "status is not a supported cell status"), + ({"status": "present"}, "present requires data_through"), + ( + {"status": "present", "data_through": WATERMARK, "reason": "reason"}, + "present must not carry a reason", + ), + ({"status": "explicit_zero", "reason": "reason"}, "explicit_zero must not carry a reason"), + ({"status": "missing"}, "missing requires a stable reason"), + ({"status": "delayed", "data_through": WATERMARK}, "delayed requires a stable reason"), + ({"status": "unsupported"}, "unsupported requires a stable reason"), + ( + {"status": "missing", "reason": "reason", "data_through": WATERMARK}, + "missing must not carry data_through", + ), + ( + {"status": "unsupported", "reason": "reason", "data_through": WATERMARK}, + "unsupported must not carry data_through", + ), + ( + {"status": "present", "data_through": WATERMARK.replace(tzinfo=None)}, + "data_through must be a timezone-aware datetime", + ), + ( + {"status": "explicit_zero", "data_through": "2026-11-01"}, + "data_through must be a timezone-aware datetime", + ), + ], +) +def test_invalid_evidence_is_rejected_at_construction(fields: dict[str, Any], message: str) -> None: + with pytest.raises(ValueError, match=message): + MetricEvidence(**fields) + + +@pytest.mark.parametrize( + "reason", ["", " ", " leading", "trailing ", "not\nsafe", "non_ascii_é", "x" * 513, 42] +) +def test_reasons_use_the_existing_wire_evidence_contract(reason: Any) -> None: + with pytest.raises(ValueError, match="reason must be a wire-valid stable reason"): + MetricEvidence.unavailable(reason) + + +def test_reason_at_the_wire_limit_is_accepted() -> None: + assert MetricEvidence.missing("x" * 512).reason == "x" * 512 + + +@pytest.mark.parametrize( + "evidence", + [ + MetricEvidence.present(WATERMARK), + MetricEvidence.explicit_zero(), + MetricEvidence.explicit_zero(data_through=WATERMARK), + MetricEvidence.missing("not_returned"), + MetricEvidence.delayed("processing"), + MetricEvidence.delayed("processing", data_through=WATERMARK), + MetricEvidence.unavailable("not_video_inventory"), + ], +) +def test_evidence_is_immutable(evidence: MetricEvidence) -> None: + with pytest.raises(FrozenInstanceError): + setattr(evidence, "reason", "changed") + + +@pytest.mark.parametrize( + "field", ["semantic_contract_id", "semantic_contract_version", "semantic_contract_sha256"] +) +def test_adapters_cannot_supply_semantic_contract_fields(field: str) -> None: + fields = {"status": "present", "data_through": WATERMARK, field: "adapter_override"} + with pytest.raises(TypeError, match="unexpected keyword argument"): + MetricEvidence(**fields) + + +@pytest.mark.parametrize("pattern", ["unsupported", "delayed"]) +def test_bulk_helpers_reject_unrequested_metrics(pattern: str) -> None: + request = redacted_snapshot_request() + with pytest.raises(ValueError, match="unknown requested metric 'completed_views'"): + if pattern == "unsupported": + metric_unsupported_everywhere(request, "completed_views", "not_video_inventory") + else: + metric_delayed_through(request, "completed_views", WATERMARK) + + +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +async def test_constructor_errors_inside_fetch_remain_actionable(asynchronous: bool) -> None: + def invalid(req: Any) -> InlineFetchResult: + return InlineFetchResult( + rows=[], + cell_availability={"campaign-redacted-1": {"impressions": MetricEvidence("present")}}, + ) + + async def invalid_async(req: Any) -> InlineFetchResult: + return invalid(req) + + source = InlineReportingSource( + capabilities=redacted_capabilities(), fetch=invalid_async if asynchronous else invalid + ) + with pytest.raises(ValueError, match="MetricEvidence present requires data_through"): + await source.execute(redacted_snapshot_request(), cancel=asyncio.Event()) + + +async def test_ordinary_provider_value_errors_keep_their_legacy_classification() -> None: + def fetch(req: Any) -> None: + raise ValueError("provider diagnostic detail") + + source = InlineReportingSource(capabilities=redacted_capabilities(), fetch=fetch) + error = validate_reporting_source_failure( + await source.execute(redacted_snapshot_request(), cancel=asyncio.Event()), + "PROVIDER_TRANSIENT", + ) + assert error.retry == "retryable" + assert error.safe_message == "the reporting source raised ValueError" diff --git a/tests/type_checks/reporting_inline_source.py b/tests/type_checks/reporting_inline_source.py new file mode 100644 index 000000000..57f2d8dc0 --- /dev/null +++ b/tests/type_checks/reporting_inline_source.py @@ -0,0 +1,118 @@ +"""Adopter patterns: typed synchronous/asynchronous fetches and metric evidence. + +These adapters assume an offering requesting impressions, clicks, viewability, +and completed_views. The SDK supplies each metric's semantic-contract identity. +""" + +from collections.abc import Awaitable, Callable, Mapping, Sequence +from datetime import timedelta + +from adcp.reporting.inline_source import ( + InlineFetch, + InlineFetchResult, + InlineReportingSource, + MetricEvidence, + metric_delayed_through, + metric_unsupported_everywhere, +) +from adcp.reporting.source import ( + ReportingSourceCapabilitiesV1, + ReportingSourceExecutor, + ReportingSourceSliceRequestV1, +) + + +def read_rows(request: ReportingSourceSliceRequestV1) -> Sequence[Mapping[str, object]]: + return [ + {"constituent_id": item.constituent_id, "impressions": 120, "clicks": 0} + for item in request.coverage.constituents + ] + + +def fetch_sync(request: ReportingSourceSliceRequestV1) -> InlineFetchResult: + through = min(request.period.end, request.period.source_read_cutoff_at) + evidence: dict[str, dict[str, MetricEvidence]] = { + item.constituent_id: { + "impressions": MetricEvidence.present(through), + "clicks": MetricEvidence.explicit_zero(data_through=through), + "viewability": MetricEvidence.delayed( + "measurement_pending", data_through=through - timedelta(hours=1) + ), + "completed_views": MetricEvidence.unavailable("not_video_inventory"), + } + for item in request.coverage.constituents + } + return InlineFetchResult(rows=read_rows(request), cell_availability=evidence) + + +def async_source( + capabilities: ReportingSourceCapabilitiesV1, + read: Callable[ + [ReportingSourceSliceRequestV1], Awaitable[Sequence[Mapping[str, object]] | None] + ], +) -> ReportingSourceExecutor: + async def fetch(request: ReportingSourceSliceRequestV1) -> InlineFetchResult | None: + rows = await read(request) + if rows is None: + return None + return InlineFetchResult( + rows=rows, + cell_availability={ + item.constituent_id: { + "viewability": MetricEvidence.missing("not_returned"), + "completed_views": MetricEvidence.unavailable("not_video_inventory"), + } + for item in request.coverage.constituents + }, + ) + + async_fetch: InlineFetch = fetch + return InlineReportingSource(capabilities=capabilities, fetch=async_fetch) + + +def unsupported_video(request: ReportingSourceSliceRequestV1) -> InlineFetchResult: + return InlineFetchResult( + rows=[{**row, "viewability": "0.75"} for row in read_rows(request)], + cell_availability=metric_unsupported_everywhere( + request, "completed_views", "not_video_inventory" + ), + ) + + +def delayed_viewability(request: ReportingSourceSliceRequestV1) -> InlineFetchResult: + return InlineFetchResult( + rows=[{**row, "completed_views": 30} for row in read_rows(request)], + cell_availability=metric_delayed_through( + request, + "viewability", + request.period.source_read_cutoff_at - timedelta(hours=1), + reason="measurement_pending", + ), + ) + + +def all_present(request: ReportingSourceSliceRequestV1) -> InlineFetchResult: + return InlineFetchResult.all_present( + read_complete_rows(request), data_through=request.period.source_read_cutoff_at + ) + + +def read_complete_rows(request: ReportingSourceSliceRequestV1) -> Sequence[Mapping[str, object]]: + return [{**row, "viewability": "0.75", "completed_views": 30} for row in read_rows(request)] + + +def configure_sources( + capabilities: ReportingSourceCapabilitiesV1, +) -> tuple[ReportingSourceExecutor, ...]: + sync_fetch: InlineFetch = fetch_sync + # Bare row callbacks and richer results remain accepted. + return tuple( + InlineReportingSource(capabilities=capabilities, fetch=fetch) + for fetch in ( + sync_fetch, + read_complete_rows, + unsupported_video, + delayed_viewability, + all_present, + ) + )