diff --git a/README.md b/README.md index eb5b48340..9fc00e55a 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,7 @@ This README serves both sides of an AdCP integration. Jump to what you're doing: - **Connect as a buyer** → [Quick Start: Test Helpers](#quick-start-test-helpers) and [Quick Start: Distributed Operations](#quick-start-distributed-operations). Entry point: `from adcp import ADCPClient, AgentConfig`; start with the `client.simple.*` API. - **Build a seller / agent** → [Building an AdCP Agent](#building-an-adcp-agent). Entry point: `from adcp.server import ADCPHandler, serve`; use the [production seller path](docs/production-seller.md) when adding tenants, durable tasks, and webhooks. +- **Run Reliable Reporting** → [Account currencies](docs/reporting-currency.md) and [ledger migrations](docs/reporting-ledger-migration.md). - **Understand the type system & imports** → [Type Safety](#type-safety) (import surface, partial modules, cold-start note). - **Test against reference agents** → [Quick Start: Test Helpers](#quick-start-test-helpers) and [Test Helpers](#test-helpers). Entry point: `from adcp.testing import test_agent, creative_agent`. diff --git a/docs/reporting-currency.md b/docs/reporting-currency.md new file mode 100644 index 000000000..f98c2e5c6 --- /dev/null +++ b/docs/reporting-currency.md @@ -0,0 +1,164 @@ +# Currency on reporting obligations + +One `ReportingProducer` can serve USD and EUR accounts concurrently. Currency +is resolved once, from trusted seller state, when the SDK creates an obligation. +It is stored before source acquisition and reused for retries, process restarts, +snapshot restatements, official publications and subsequent adjustments. + +Pass a synchronous or asynchronous `CurrencyResolver` to the producer: + +```python +import asyncio + +from adcp.reporting.ledger import ( + ProducerOfferings, + ReportingConfiguration, + ReportingObligationRecord, + ReportingProducer, + require_single_currency, +) + +# Illustrative historical seller records, keyed by account and media buy. +# In production, read their accepted value as of candidate.scope_resolved_at. +accepted_currencies = { + ("us-account", "us-buy"): "USD", + ("eu-account", "eu-buy"): "EUR", +} + +def resolve_currency( + configuration: ReportingConfiguration, + candidate: ReportingObligationRecord, +) -> str: + return require_single_currency( + accepted_currencies[(candidate.account_id, buy_id)] + for buy_id in candidate.media_buy_ids + ) + +producer = ReportingProducer( + store=ledger, + source=source, + object_reader=staging, + offerings=ProducerOfferings(snapshot_offering_id="ACCOUNT_CURRENCY_SNAPSHOT"), + currency_resolver=resolve_currency, +) +# Accepted us-account/daily@1 and eu-account/daily@1 configurations each +# contain their own media buy. The #1169 account-qualified keys isolate them. +await asyncio.gather(producer.run_worker(), producer.run_worker()) +``` + +`ledger`, `source` and `staging` are the seller's existing reporting components; +the offering must support the accepted report definition and source scope. +The resolver receives SDK-owned configuration/obligation records, including the +account-qualified generation, period-end timestamp and frozen media-buy/package +scope. The candidate's currency is initially `None`. An async resolver can load +the same historical account context that a future `ReliableReportingService` +uses. A resolver must establish **one currency for the entire scope** and raise +on unknown or mixed currencies. `require_single_currency` validates each code +and rejects empty/mixed input without aggregating it. + +Never derive this value from a buyer's request `context`, a live mutable default, +or an adapter response. A resolver should perform bounded, read-only historical +lookups. It can run more than once if workers race before commit; both workers +use the store's immutable winning obligation. An existing obligation never +calls the resolver again. Resolver failure occurs before obligation creation +and leaves source work untouched; it must be surfaced by the worker supervisor. + +The existing `ProducerOfferings(currency="EUR")` option remains a convenient +single-currency configuration, implemented by `FixedCurrencyResolver`. Its +default remains USD. A custom resolver takes precedence. This convenience is +appropriate only when every scope it serves has that currency. Codes must match +three uppercase ASCII letters exactly: `EUR` is accepted; `eur`, whitespace, +non-ASCII letters, and non-string values fail with `INVALID_CURRENCY`. Validation +checks ISO 4217 **shape**, not membership in a periodically changing registry. + +## Pinned monetary semantics + +`ReportingDefinitionBinding` can retain immutable monetary unit declarations +projected by trusted seller code from its verified, content-addressed definition: + +```python +from dataclasses import replace + +binding = replace( + verified_binding, + monetary_metric_units=(("spend", "EUR"),), + monetary_control_total_units=(("spend", "EUR"),), +) +``` + +These are tuples, copied into immutable pairs on construction and persisted +with the obligation. They are not an alternate definition download or evidence +obtained from the source. Verify the definition bytes against the retained +digest before projecting them. Do not supply a unit that its pinned definition +does not establish. If the definition leaves currency account-scoped, the +trusted resolver establishes it; unitless values inherit the frozen currency. + +Conflicting pinned currencies fail at freeze time. Both ledger stores check +flat monetary columns and same-name additive totals using exact decimal values. +The existing `spend` column is treated as monetary even without extra declarations; +declare other monetary columns and totals explicitly. This convenience check +is for flat additive reporting, matching `InlineReportingSource`; it is not a +general evaluator for arbitrary definition expressions or nested row schemas. +Non-additive/custom metrics require an appropriate seller/source validator. +Nonmonetary control-total units are preserved; three capital letters alone do +not make a unit a currency. + +A metric that some rows omit has no honest sum, because an omitted cell and a +measured zero are different facts. That is exactly when `InlineReportingSource` +declines to publish a control total for it, and those rows still publish: each +reported value is checked and the column is left unsummed. A control total that +*is* published must reconcile against every row; a column every row carries must +come with one. A period with no rows carries no money column, so it owes no +derived total -- `monetary_control_total_units` is the declaration meaning +"always present", and it applies rows or none. `null` reads as "not reported" +for both money columns and row `currency` labels. + +Source slice requests use only the stored currency. A manifest's currency and +explicit monetary total units must match; its definition binding must also +match any retained pin before staging is read or a revision is committed. The +source cannot override the obligation. `InlineFetchResult(currency="EUR", rows=...)` +supplies optional corroboration. Any explicit row `currency` must match too. +When a fetch returns `GetMediaBuyDeliveryResponse`, response, media-buy and +package currency labels are checked before flattening. Mixed/mismatched rows +are rejected before inline aggregation or staging; no conversion to USD occurs. +The adapter copies rows before validation so a shared cache cannot change the +checked money during staging. +Absent unit labels inherit the obligation/definition, preserving existing +unitless low-level row and total formats. Inline `spend` totals explicitly carry +the frozen currency in their source manifests. + +`ReportingCurrencyError.code` distinguishes `CURRENCY_UNRESOLVED`, +`CURRENCY_MISMATCH`, `MIXED_CURRENCY_SCOPE`, `INVALID_CURRENCY`, and +`MONETARY_TOTAL_MISMATCH`. An inline adapter returns currency failures as terminal +`INTEGRITY_FAILED` with that reason in `safe_message`. + +## Low-level use and retained history + +Low-level writers can still construct records and call `commit_obligation`, +`commit_revision` and `commit_adjustment` directly. Set `currency="EUR"` on each +new obligation, using trusted historical evidence. `currency=None` remains +representable for loading old records, but new writes without currency fail. +Public producer acquisition/manifest commit calls reload the stored obligation; +passing a modified copy cannot substitute its currency. Exact legacy record +replays remain idempotent. Adjustments inherit units through their official +revision's obligation and cannot request a different currency. + +No new currency field is invented on the AdCP status wire schema. Status and +exact content reads retain the original definition binding, rows and digests; +audits can inspect the obligation's currency through the store. The consumer's +existing `currency_mismatch` classification remains available for contradictory +observed metric/control-total units. + +Upgrading from beta.15 or #1169 preserves unknown legacy currencies as `NULL`, +blocks new acquisition/publication/adjustment for them, and projects +`HISTORY_UNAVAILABLE` / `action_required` while keeping their history readable. +See [the migration policy and deployment instructions](reporting-ledger-migration.md). + +The shared memory/Postgres scenarios in +[`test_reporting_currency.py`](../tests/conformance/reporting/test_reporting_currency.py) +publish USD and EUR through one producer, change its resolver/default between +attempts, and reconcile the frozen results with the buyer-side SDK. The +[strict adopter fixture](../tests/type_checks/reporting_currency.py) demonstrates +both callback forms and low-level writes without typing suppressions. +`adcp.reporting.fixtures.redacted_multi_currency_requests()` supplies USD/EUR +slice requests for adopter replay-conformance tests. diff --git a/docs/reporting-ledger-migration.md b/docs/reporting-ledger-migration.md index 6690eb113..dd5c91c61 100644 --- a/docs/reporting-ledger-migration.md +++ b/docs/reporting-ledger-migration.md @@ -1,4 +1,4 @@ -# Account-qualified reporting generations +# Reporting ledger migrations The fix for [#1169](https://github.com/adcontextprotocol/adcp-client-python/issues/1169) changes a reporting configuration generation's identity to @@ -49,22 +49,24 @@ obligation IDs, the named obligations must exist in the requested account. mixing old and new workers is unsafe once accounts reuse a config ID. 2. With the upgraded SDK, run `await store.create_schema()` before starting reporting work. It creates missing tables and applies the bundled - `reporting_ledger_account_generations.sql` migration in one transaction. + `reporting_ledger_account_generations.sql` and + `reporting_ledger_obligation_currency.sql` migrations in one transaction. 3. Restart reporting work with the upgraded SDK on every instance. For deployments managed by a migration tool, the standalone migration is [`reporting_ledger_account_generations.sql`](../src/adcp/reporting/ledger/reporting_ledger_account_generations.sql). It upgrades an existing beta.15 ledger by itself, including in autocommit mode. -For a combined bootstrap and upgrade, run both bundled files in one transaction: +For a combined bootstrap and upgrade, run all three bundled files in one transaction: ```sh psql "$REPORTING_DATABASE_URL" --set=ON_ERROR_STOP=1 --single-transaction \ -f src/adcp/reporting/ledger/reporting_ledger.sql \ - -f src/adcp/reporting/ledger/reporting_ledger_account_generations.sql + -f src/adcp/reporting/ledger/reporting_ledger_account_generations.sql \ + -f src/adcp/reporting/ledger/reporting_ledger_obligation_currency.sql ``` Use the ledger's existing `search_path` and a role that owns its tables. Both -SQL files are also included as resources in the installed SDK's +SQL migrations are also included as resources in the installed SDK's `adcp.reporting.ledger` package. Running only `CREATE TABLE IF NOT EXISTS` leaves the old primary key in place and does not perform this upgrade. @@ -98,3 +100,76 @@ account-qualified implementation. The in-memory store needs no schema migration. Restart it with the upgraded SDK and reload accepted configurations from the adopter's source of truth. + +## Frozen currency: upgrading beta.15 or the #1169 schema + +[#1171](https://github.com/adcontextprotocol/adcp-client-python/issues/1171) +adds `reporting_obligations.currency`. The migration runs after #1169 under the +same advisory lock and transaction. It is an idempotent, in-place nullable +column addition, with uppercase three-letter validation and a trigger that +rejects changes to a stored currency, including `NULL` to a guessed value. +The account-qualified keys and their migration remain intact. + +**Legacy policy: preserve unknown, fail closed.** Neither beta.15 nor #1169 +persisted the process currency on obligations. A definition URI/digest does not +contain its definition bytes; retained Core totals contain name/value pairs +without units. Row data may include a currency, but it is adapter-supplied +corroboration rather than proof of the originally accepted scope. Empty rows +and unfulfilled obligations carry even less evidence. Therefore the migration +leaves **every existing obligation's currency `NULL`**, including apparently +USD rows. It does not copy a current account setting, use a new resolver, read +buyer context, infer from an adapter row, or default historical obligations to USD. + +The options considered are: + +1. Backfill USD or today's account currency: rejected. Either can mislabel + non-USD history, and even historically USD deployments require evidence + beyond the retained ledger to prove that choice. +2. Recover from external, authenticated historical configuration/definition + evidence: possible only through a separate, reviewed adopter migration. + Corroborate the entire frozen account/media-buy/package scope, including + every existing revision and adjustment; retain the provenance and audit + trail. The SDK does not perform that repair or supply an unchecked backfill + API. Its immutability trigger deliberately requires explicit operator work. +3. Retain `NULL`, prevent new monetary work and preserve readable history: + **the implemented recommendation**. This makes no lossy assumption and is + safe when historical evidence is unavailable. + +Upgraded workers refuse acquisition, snapshot restatement, new revision and +new adjustment writes for unknown obligations with `CURRENCY_UNRESOLVED`. An +obligation that is already satisfied or officially closed has no acquisition +work to refuse, so it stays the no-op it was before the upgrade. Inside +`run_worker`, an unresolved obligation is reported in `WorkerTurn.slices_failed` +and escalated like any other stuck slice: every *other* period under the same +configuration still closes and publishes on that same turn. A direct +`acquire_obligation` call still raises, so an operator driving one period by +hand sees the failure. +Exact legacy obligation/revision/adjustment replays still return the retained +record without appending evidence. Status and content reads remain available; +unknown obligations project `HISTORY_UNAVAILABLE`, `action_required` and +`contact_seller`. Reads never resolve or backfill currency. A new accepted +generation can support **future** work with a proven currency; it must not be +used to relabel old periods or erase a missing historical obligation. + +No existing rows, hashes, statuses, leases, revisions, adjustments or change-feed +sequences are rewritten. The new column has **no database default**. Existing +indexes/constraints are preserved; a currency check and immutability trigger +are added. The currency migration locks `reporting_obligations` exclusively +while adding/validating the column and check. Plan a maintenance window for +large tables; production-sized lock time has not been benchmarked. Unexpected +adopter column types/defaults fail and roll back rather than silently adapting. + +Stop and drain **all** beta.15 and #1169-only reporting writers before upgrading; +they do not supply frozen currency. Run `create_schema()` or the three-file SQL +command above, then start upgraded writers. If #1169 is already installed, its +primary-key migration recognizes the account-qualified key without rebuilding +it. The standalone currency SQL also runs atomically on an installed #1169 +schema, including with autocommit. Do not run older code against this schema: +it can ignore the new invariant or create unresolved obligations. + +New low-level `ReportingObligationRecord` writes must explicitly include a +trusted `currency`. The optional Python field exists to deserialize legacy +history, not to authorize an unfrozen new publication. High-level users of +`ProducerOfferings(currency=...)` keep their fixed-currency convenience; new +obligations freeze that value. See [multi-account currency resolution and +monetary validation](reporting-currency.md) for the resolver and type examples. diff --git a/src/adcp/reporting/__init__.py b/src/adcp/reporting/__init__.py index bf6974368..b2f68881b 100644 --- a/src/adcp/reporting/__init__.py +++ b/src/adcp/reporting/__init__.py @@ -71,13 +71,14 @@ if TYPE_CHECKING: from adcp.reporting import canonical_json as canonical_json from adcp.reporting import conformance as conformance + from adcp.reporting import currency as currency from adcp.reporting import fixtures as fixtures from adcp.reporting import inline_source as inline_source from adcp.reporting import ledger as ledger from adcp.reporting import source as source _LAZY_SUBMODULES = frozenset( - {"canonical_json", "conformance", "fixtures", "inline_source", "ledger", "source"} + {"canonical_json", "conformance", "currency", "fixtures", "inline_source", "ledger", "source"} ) diff --git a/src/adcp/reporting/conformance.py b/src/adcp/reporting/conformance.py index d2a86d33f..6680c8c49 100644 --- a/src/adcp/reporting/conformance.py +++ b/src/adcp/reporting/conformance.py @@ -378,6 +378,14 @@ def _validate_manifest_against_request( f"({actual!r} != {expected!r})", ) + for total in manifest.control_totals: + # Other monetary names need the ledger's trusted definition binding; + # a three-letter unit alone does not establish monetary semantics. + if total.unit is not None and total.name == "spend" and total.unit != request.currency: + raise _fail( + "MANIFEST_MISMATCH", "monetary control total unit contradicts frozen currency" + ) + requested_metrics = set(request.requested_metrics) requested_constituents = {item.constituent_id for item in request.coverage.constituents} declared = {metric.name: metric for metric in offering.metrics} diff --git a/src/adcp/reporting/currency.py b/src/adcp/reporting/currency.py new file mode 100644 index 000000000..238b4fd0b --- /dev/null +++ b/src/adcp/reporting/currency.py @@ -0,0 +1,177 @@ +"""Currency checks shared by trusted obligation writers and source adapters. + +These helpers never choose a currency from report rows. A row or a manifest +can corroborate the seller's frozen currency, but cannot establish it. +""" + +from __future__ import annotations + +import math +import re +from collections.abc import Iterable, Mapping, Sequence +from decimal import Decimal, InvalidOperation, localcontext +from typing import Any + +__all__ = ["ReportingCurrencyError", "require_single_currency", "validate_currency"] + + +class ReportingCurrencyError(ValueError): + """Reporting money cannot be interpreted safely; ``code`` is stable.""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(f"{code}: {message}") + self.code = code + + +def validate_currency(value: object) -> str: + """Require an ISO 4217-shaped code, without coercion or a registry lookup.""" + if not isinstance(value, str) or re.fullmatch(r"[A-Z]{3}", value) is None: + raise ReportingCurrencyError( + "INVALID_CURRENCY", "reporting currency must be three uppercase ASCII letters" + ) + return value + + +def require_single_currency(currencies: Iterable[str]) -> str: + """Resolve trusted constituent currencies, rejecting an empty or mixed scope. + + Call this inside a currency resolver with historical account/media-buy + values, before an obligation is committed. It does not convert money. + """ + resolved = {validate_currency(value) for value in currencies} + if not resolved: + return require_frozen_currency(None) + if len(resolved) != 1: + raise ReportingCurrencyError( + "MIXED_CURRENCY_SCOPE", "one reporting obligation cannot aggregate multiple currencies" + ) + return next(iter(resolved)) + + +def require_frozen_currency(currency: str | None) -> str: + if currency is None: + raise ReportingCurrencyError( + "CURRENCY_UNRESOLVED", + "the obligation has no proven currency; retain its history and use verified " + "historical evidence for an explicit repair, never current defaults", + ) + return validate_currency(currency) + + +def validate_currency_units(currency: str, units: Iterable[tuple[str, str]]) -> None: + """Check trusted monetary unit declarations against the resolved currency.""" + pinned = {validate_currency(unit) for _, unit in units} + if len(pinned) > 1: + raise ReportingCurrencyError( + "MIXED_CURRENCY_SCOPE", "the pinned definition declares multiple monetary currencies" + ) + if pinned and pinned != {currency}: + raise ReportingCurrencyError( + "CURRENCY_MISMATCH", "the currency disagrees with the pinned monetary definition" + ) + + +def _carries(row: Mapping[str, Any], name: str) -> bool: + """Whether a row actually reports ``name``. + + ``None`` is "not reported", not zero and not an empty label -- the same + reading the inline adapter's control totals take, so the two cannot + disagree about which cells exist. + """ + return row.get(name) is not None + + +def validate_row_currencies(currency: str, rows: Sequence[Mapping[str, Any]]) -> None: + """Reject mixed/mismatched row evidence before any aggregation takes place.""" + observed = {validate_currency(row["currency"]) for row in rows if _carries(row, "currency")} + if len(observed) > 1: + raise ReportingCurrencyError( + "MIXED_CURRENCY_SCOPE", "source rows contain multiple currencies" + ) + if observed and observed != {currency}: + raise ReportingCurrencyError( + "CURRENCY_MISMATCH", "source row currency disagrees with the frozen currency" + ) + + +def monetary_decimal(value: object) -> Decimal: + """An exact, finite decimal representation of a normalized metric value.""" + if isinstance(value, bool) or not isinstance(value, (str, int, float, Decimal)): + raise ReportingCurrencyError("MONETARY_TOTAL_MISMATCH", "invalid monetary value") + if isinstance(value, float) and not math.isfinite(value): + raise ReportingCurrencyError("MONETARY_TOTAL_MISMATCH", "non-finite monetary value") + try: + result = Decimal(str(value)) + except InvalidOperation as error: + raise ReportingCurrencyError("MONETARY_TOTAL_MISMATCH", "invalid monetary value") from error + if not result.is_finite(): + raise ReportingCurrencyError("MONETARY_TOTAL_MISMATCH", "non-finite monetary value") + return result + + +def validate_monetary_content( + *, + currency: str, + rows: Sequence[Mapping[str, Any]], + totals: Sequence[tuple[str, str]], + metric_units: Sequence[tuple[str, str]] = (), + total_units: Sequence[tuple[str, str]] = (), +) -> None: + """Check flat monetary columns and their same-name additive totals. + + ``spend`` is the built-in money column for the existing single-currency + API. Other monetary columns/totals must be declared by the trusted pinned + definition. Unitless values inherit that declaration, never a source hint. + + A metric absent from *some* rows has no honest additive total -- the same + rule :func:`adcp.reporting.inline_source._control_totals` applies when it + declines to publish one -- so a sparse money column is checked value by + value and left unsummed. A published total is still reconciled exactly, and + a column every row carries still has to come with one. Rows are what create + that demand, so a period with no rows owes no derived total; ``total_units`` + is the declaration that demands one unconditionally. + """ + validate_currency_units(currency, (*metric_units, *total_units)) + validate_row_currencies(currency, rows) + declared_metrics = dict(metric_units) + declared_totals = dict(total_units) + if "spend" in dict(totals) or any(_carries(row, "spend") for row in rows): + declared_metrics.setdefault("spend", currency) + total_values = dict(totals) + if len(total_values) != len(totals): + raise ReportingCurrencyError("MONETARY_TOTAL_MISMATCH", "duplicate control total names") + for name in declared_totals: + if name not in total_values: + raise ReportingCurrencyError( + "MONETARY_TOTAL_MISMATCH", f"missing pinned monetary control total {name!r}" + ) + monetary_decimal(total_values[name]) + for name in declared_metrics: + values = [monetary_decimal(row[name]) for row in rows if _carries(row, name)] + complete = len(values) == len(rows) + if name not in total_values: + if complete and rows: + raise ReportingCurrencyError( + "MONETARY_TOTAL_MISMATCH", + f"monetary metric {name!r} needs rows and a control total", + ) + continue + if not complete: + raise ReportingCurrencyError( + "MONETARY_TOTAL_MISMATCH", + f"control total {name!r} cannot be reconciled against rows missing it", + ) + # Do not let an adopter's Decimal context round an otherwise exact sum. + # Decimal exponents are integers after the finite check above. + precision = sum(len(value.as_tuple().digits) for value in values) + 1 + if values: + precision += max(value.adjusted() for value in values) - min( + int(value.as_tuple().exponent) for value in values + ) + with localcontext() as context: + context.prec = max(28, precision) + observed = sum(values, Decimal(0)) + if observed != monetary_decimal(total_values[name]): + raise ReportingCurrencyError( + "MONETARY_TOTAL_MISMATCH", f"control total {name!r} disagrees with monetary rows" + ) diff --git a/src/adcp/reporting/fixtures.py b/src/adcp/reporting/fixtures.py index 50abedbb2..6acb010a0 100644 --- a/src/adcp/reporting/fixtures.py +++ b/src/adcp/reporting/fixtures.py @@ -68,6 +68,7 @@ "redacted_capabilities", "redacted_completed_result", "redacted_contract_identity", + "redacted_multi_currency_requests", "redacted_snapshot_request", ] @@ -211,17 +212,31 @@ def redacted_capabilities( ) +def redacted_multi_currency_requests() -> tuple[ReportingSourceSliceRequestV1, ...]: + """Two frozen account scopes for adopters testing a shared USD/EUR adapter.""" + return tuple( + redacted_snapshot_request( + account_id=f"account-{currency.lower()}", + currency=currency, + source_execution_key=f"currency-{currency.lower()}-001", + ) + for currency in ("USD", "EUR") + ) + + def redacted_snapshot_request( *, source_execution_key: str = "snapshot-execution-001", run_id: str = "run-redacted-1", source_read_cutoff_at: datetime | None = None, trigger: str = "scheduled_poll", + currency: str = "USD", + account_id: str = "account-redacted", ) -> ReportingSourceSliceRequestV1: """A frozen ``PROVISIONAL_SNAPSHOT`` slice over one media-buy constituent.""" return ReportingSourceSliceRequestV1( identity=ReportingSourceIdentityV1( - account_id="account-redacted", + account_id=account_id, delivery_config_id="config-redacted", delivery_config_version=1, report_definition_id="PAID_MEDIA_DAILY_V1", @@ -257,7 +272,7 @@ def redacted_snapshot_request( ), requested_metrics=["impressions", "spend"], requested_dimensions=["campaign_id"], - currency="USD", + currency=currency, deadline_at=datetime(2099, 11, 2, 10, 0, tzinfo=timezone.utc), ) diff --git a/src/adcp/reporting/inline_source.py b/src/adcp/reporting/inline_source.py index 4c8eb7550..33e656299 100644 --- a/src/adcp/reporting/inline_source.py +++ b/src/adcp/reporting/inline_source.py @@ -54,12 +54,18 @@ import os import tempfile from collections.abc import Awaitable, Callable, Collection, Mapping, Sequence -from dataclasses import dataclass, field +from copy import deepcopy +from dataclasses import dataclass, field, replace from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any, Protocol, TypeAlias, runtime_checkable +from adcp.reporting.currency import ( + ReportingCurrencyError, + validate_currency, + validate_row_currencies, +) from adcp.reporting.source import ( MediaBuyConstituentV1, PackageItemConstituentV1, @@ -149,6 +155,9 @@ class InlineFetchResult: warnings: Sequence[str] = () """Safe, redacted operator notes retained on the publication.""" + currency: str | None = None + """Optional source corroboration; must match the already frozen request.""" + #: What an inline fetch may return. ``None`` is "not ready". InlineFetchReturn: TypeAlias = "InlineFetchResult | Sequence[Mapping[str, Any]] | None | object" @@ -453,7 +462,26 @@ async def execute( # Coerce inside the guard: a fetch that returns something # unrecognizable is a source failure like any other, not an # exception escaping the executor. - answer = _coerce(answer) + answer = _coerce(answer, currency=request.currency) + # Retain the rows we checked across staging awaits. A fetch + # may return a shared cache whose mappings later change. + answer = replace(answer, rows=tuple(deepcopy(dict(row)) for row in answer.rows)) + if ( + answer.currency is not None + and validate_currency(answer.currency) != request.currency + ): + raise ReportingCurrencyError( + "CURRENCY_MISMATCH", + "inline source currency disagrees with the frozen request", + ) + # Before staging and, crucially, before _control_totals aggregates. + validate_row_currencies(request.currency, answer.rows) + except ReportingCurrencyError as error: + return ReportingSourceExecutorResult.failed( + ReportingSourceErrorV1( + code="INTEGRITY_FAILED", retry="terminal", safe_message=str(error) + ) + ) except ReportingSourceError as error: return ReportingSourceExecutorResult.failed(error.error) except asyncio.CancelledError: @@ -890,11 +918,21 @@ def _control_totals( scaled = any(_scale_of(item) < 0 for item in decimals) if scaled: totals.append( - SourceControlTotalV1(name=metric, value=_plain(total), value_type="decimal") + SourceControlTotalV1( + name=metric, + value=_plain(total), + value_type="decimal", + unit=request.currency if metric == "spend" else None, + ) ) else: totals.append( - SourceControlTotalV1(name=metric, value=str(int(total)), value_type="integer") + SourceControlTotalV1( + name=metric, + value=str(int(total)), + value_type="integer", + unit=request.currency if metric == "spend" else None, + ) ) return totals @@ -974,11 +1012,11 @@ def _is_async_callable(fetch: InlineFetch) -> bool: return call is not None and asyncio.iscoroutinefunction(call) -def _coerce(answer: InlineFetchReturn) -> InlineFetchResult: +def _coerce(answer: InlineFetchReturn, *, currency: str) -> InlineFetchResult: """Normalize whatever the fetch returned into an :class:`InlineFetchResult`.""" if isinstance(answer, InlineFetchResult): return answer - rows = _delivery_rows(answer) + rows = _delivery_rows(answer, currency=currency) if rows is not None: return rows if isinstance(answer, Sequence) and not isinstance(answer, (str, bytes)): @@ -990,7 +1028,22 @@ def _coerce(answer: InlineFetchReturn) -> InlineFetchResult: ) -def _delivery_rows(answer: Any) -> InlineFetchResult | None: +def _declared_currency(holder: Any) -> Any: + """The currency a delivery object labels itself with, or ``None``. + + ``GetMediaBuyDeliveryResponse.currency`` is deprecated in AdCP 3.2 and + Pydantic warns on every attribute read of it. Checking a legacy + response-wide label against the frozen request is deliberate -- a + contradiction there still has to fail the slice -- so read the stored + value rather than emitting a DeprecationWarning per fetch. + """ + stored = getattr(holder, "__dict__", None) + if isinstance(stored, dict) and "currency" in stored: + return stored["currency"] + return getattr(holder, "currency", None) + + +def _delivery_rows(answer: Any, *, currency: str) -> InlineFetchResult | None: """Project an AdCP ``GetMediaBuyDeliveryResponse`` into normalized rows. Prefers the response's own ``reporting_rows`` when present -- those are @@ -1002,6 +1055,23 @@ def _delivery_rows(answer: Any) -> InlineFetchResult | None: reporting_rows = getattr(answer, "reporting_rows", None) if deliveries is None and reporting_rows is None: return None + # Flattening must not discard a response/media-buy/package denomination + # before it can contradict the frozen request. These are source hints, + # never a way to choose the obligation's currency. + currency_holders = [answer] + for delivery in deliveries or (): + currency_holders.append(delivery) + currency_holders.extend(getattr(delivery, "by_package", None) or ()) + for window in getattr(delivery, "windows", None) or (): + currency_holders.extend(getattr(window, "by_package", None) or ()) + validate_row_currencies( + currency, + [ + {"currency": observed} + for holder in currency_holders + if (observed := _declared_currency(holder)) is not None + ], + ) period = getattr(answer, "reporting_period", None) data_through = getattr(period, "end", None) if period is not None else None if reporting_rows: diff --git a/src/adcp/reporting/ledger/__init__.py b/src/adcp/reporting/ledger/__init__.py index bb785ff5f..8769ce372 100644 --- a/src/adcp/reporting/ledger/__init__.py +++ b/src/adcp/reporting/ledger/__init__.py @@ -59,6 +59,11 @@ from __future__ import annotations +from adcp.reporting.currency import ( + ReportingCurrencyError, + require_single_currency, + validate_currency, +) from adcp.reporting.ledger.consumer_status import ( ConsumerMismatch, ConsumerStatusDisabledError, @@ -102,6 +107,8 @@ iso_duration_to_timedelta, ) from adcp.reporting.ledger.producer import ( + CurrencyResolver, + FixedCurrencyResolver, ProducerOfferings, ReportingProducer, WorkerTurn, @@ -130,6 +137,8 @@ "ConsumerStatusIngest", "ConsumerStatusRecord", "ConsumerStatusValue", + "CurrencyResolver", + "FixedCurrencyResolver", "InMemoryReportingLedgerStore", "LeasedConfiguration", "LedgerChange", @@ -142,6 +151,7 @@ "ReportingAdjustmentRecord", "ReportingConfiguration", "ReportingConfigurationGenerationKey", + "ReportingCurrencyError", "ReportingDefinitionBinding", "ReportingDeliveryEscalation", "ReportingFinality", @@ -176,8 +186,10 @@ "project_consumer_mismatch", "project_obligation_health", "reject_reserved_authoritative_party", + "require_single_currency", "revision_content_sha256", "stale_received_grace_deadline", + "validate_currency", ] diff --git a/src/adcp/reporting/ledger/health.py b/src/adcp/reporting/ledger/health.py index 0f986394a..d24cd7ef4 100644 --- a/src/adcp/reporting/ledger/health.py +++ b/src/adcp/reporting/ledger/health.py @@ -114,6 +114,36 @@ def project_obligation_health( ] current = _current_revision(qualifying) + if obligation.currency is None: + return ObligationProjection( + health="action_required", + production_status="published" if revisions else "pending", + issues=( + ReportingIssue( + issue_id=issue_id_for( + "core-currency-unresolved-v1", obligation.reporting_obligation_id + ), + code="HISTORY_UNAVAILABLE", + severity="action_required", + responsible_party="seller", + recommended_action="contact_seller", + reporting_obligation_id=obligation.reporting_obligation_id, + delivery_config_id=obligation.delivery_config_id, + delivery_config_version=obligation.delivery_config_version, + feed_purpose=obligation.feed_purpose, + media_buy_ids=obligation.media_buy_ids, + period_start=obligation.period.start, + period_end=obligation.period.end, + message=( + "Currency was not retained for this legacy obligation; " + "verified historical evidence is required." + ), + ), + ), + satisfied=False, + current_revision=current, + ) + readable = [revision for revision in qualifying if revision.readable] if readable: return ObligationProjection( diff --git a/src/adcp/reporting/ledger/models.py b/src/adcp/reporting/ledger/models.py index 75fa3ee39..40d83bafe 100644 --- a/src/adcp/reporting/ledger/models.py +++ b/src/adcp/reporting/ledger/models.py @@ -27,6 +27,8 @@ from datetime import datetime, timedelta, timezone from typing import Any, Literal +from adcp.reporting.currency import validate_currency, validate_currency_units + __all__ = [ "ConsumerStatusRecord", "ConsumerStatusValue", @@ -149,6 +151,28 @@ class ReportingDefinitionBinding: schema_sha256: str schema_dialect: str = "https://json-schema.org/draft/2020-12/schema" schema_ref_policy: str = "local_fragment_only" + # Trusted projections of the content-addressed definition, not adapter or + # buyer context. Tuples keep these declarations immutable after acceptance. + monetary_metric_units: tuple[tuple[str, str], ...] = () + monetary_control_total_units: tuple[tuple[str, str], ...] = () + + def __post_init__(self) -> None: + for field_name in ("monetary_metric_units", "monetary_control_total_units"): + units = tuple( + (name, validate_currency(unit)) for name, unit in getattr(self, field_name) + ) + if len(dict(units)) != len(units): + raise ValueError(f"duplicate names in {field_name}") + object.__setattr__(self, field_name, units) + + def to_storage(self) -> dict[str, Any]: + """Retain monetary semantics without changing AdCP wire records or old hashes.""" + payload = self.to_wire() + if self.monetary_metric_units: + payload["monetary_metric_units"] = list(self.monetary_metric_units) + if self.monetary_control_total_units: + payload["monetary_control_total_units"] = list(self.monetary_control_total_units) + return payload def to_wire(self) -> dict[str, Any]: return { @@ -370,6 +394,9 @@ class ReportingObligationRecord: package_ids: tuple[str, ...] = () definition: ReportingDefinitionBinding | None = None created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + # None describes legacy evidence only. New writes must freeze a currency; + # neither a restart nor a source response may fill an unknown historical one. + currency: str | None = None @property def generation_key(self) -> ReportingConfigurationGenerationKey: @@ -380,6 +407,16 @@ def generation_key(self) -> ReportingConfigurationGenerationKey: ) def __post_init__(self) -> None: + if self.currency is not None: + validate_currency(self.currency) + if self.definition is not None: + validate_currency_units( + self.currency, + ( + *self.definition.monetary_metric_units, + *self.definition.monetary_control_total_units, + ), + ) if _utc(self.scope_resolved_at) != _utc(self.period.end): raise ValueError( "scope_resolved_at must equal the period end; the denominator froze there " diff --git a/src/adcp/reporting/ledger/pg.py b/src/adcp/reporting/ledger/pg.py index efd53b996..cbd94f849 100644 --- a/src/adcp/reporting/ledger/pg.py +++ b/src/adcp/reporting/ledger/pg.py @@ -26,7 +26,8 @@ :meth:`create_schema` creates or upgrades the schema transactionally, including the account-qualified configuration primary key for beta.15 installations. The raw DDL ships in :file:`reporting_ledger.sql` followed by -:file:`reporting_ledger_account_generations.sql`; run both in one transaction +:file:`reporting_ledger_account_generations.sql` and +:file:`reporting_ledger_obligation_currency.sql`; run all three in one transaction when using Alembic, Flyway, or psql. See :file:`docs/reporting-ledger-migration.md` for deployment and compatibility notes. @@ -77,6 +78,7 @@ from typing import TYPE_CHECKING, Any, ClassVar, Literal from adcp.reporting.canonical_json import canonical_json_utf8_v1 +from adcp.reporting.currency import require_frozen_currency from adcp.reporting.ledger.health import issue_id_for_occurrence from adcp.reporting.ledger.models import ( ConsumerStatusRecord, @@ -102,6 +104,8 @@ encode_cursor, issue_is_retirable, reject_reserved_authoritative_party, + validate_adjustment_currency, + validate_revision_currency, ) if TYPE_CHECKING: @@ -121,6 +125,7 @@ _DDL_PATH = Path(__file__).parent / "reporting_ledger.sql" _ACCOUNT_GENERATIONS_DDL_PATH = Path(__file__).parent / "reporting_ledger_account_generations.sql" +_CURRENCY_DDL_PATH = Path(__file__).parent / "reporting_ledger_obligation_currency.sql" __all__ = ["PG_AVAILABLE", "PgReportingLedgerStore"] @@ -170,6 +175,7 @@ async def create_schema(self) -> None: async with connection.transaction(): await connection.execute(_DDL_PATH.read_text()) await connection.execute(_ACCOUNT_GENERATIONS_DDL_PATH.read_text()) + await connection.execute(_CURRENCY_DDL_PATH.read_text()) # -- change feed ------------------------------------------------------ @@ -279,6 +285,23 @@ async def commit_obligation( key = obligation.generation_key async with self._pool.connection() as connection: await self._lock_account(connection, key.account_id) + existing_row = await ( + await connection.execute( + f"SELECT {_OBLIGATION_COLUMNS} FROM reporting_obligations" # noqa: S608 # nosec B608 + " WHERE account_id = %s AND delivery_config_id = %s" + " AND delivery_config_version = %s AND period_start = %s AND period_end = %s", + ( + key.account_id, + key.delivery_config_id, + key.delivery_config_version, + obligation.period.start, + obligation.period.end, + ), + ) + ).fetchone() + if existing_row is not None: + return _obligation_from_row(existing_row) + require_frozen_currency(obligation.currency) try: inserted = await ( await connection.execute( @@ -288,9 +311,9 @@ async def commit_obligation( " feed_purpose, period_key, period_start, period_end, source_timezone," " expected_at, scope_resolved_at, automated_recovery_deadline_at," " required_finality, coverage_status, media_buy_ids, package_ids," - " schedule, definition, created_at)" + " schedule, definition, created_at, currency)" " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s," - " %s::jsonb, %s::jsonb, %s::jsonb, %s::jsonb, %s)" + " %s::jsonb, %s::jsonb, %s::jsonb, %s::jsonb, %s, %s)" " ON CONFLICT (account_id, delivery_config_id, delivery_config_version," " period_start, period_end) DO NOTHING" " RETURNING reporting_obligation_id", @@ -320,6 +343,7 @@ async def commit_obligation( else None ), obligation.created_at, + obligation.currency, ), ) ).fetchone() @@ -427,7 +451,7 @@ async def commit_revision( await self._lock_account(connection, revision.account_id) obligation = await ( await connection.execute( - "SELECT 1 FROM reporting_obligations" + f"SELECT {_OBLIGATION_COLUMNS} FROM reporting_obligations" # noqa: S608 # nosec B608 " WHERE reporting_obligation_id = %s AND account_id = %s", (revision.reporting_obligation_id, revision.account_id), ) @@ -437,6 +461,7 @@ async def commit_revision( "OBLIGATION_NOT_FOUND", "a revision must attach to an obligation committed at the period close", ) + validate_revision_currency(_obligation_from_row(obligation), revision, rows) if revision.supersedes_reporting_revision_id: await self._require_current_leaf(connection, revision) try: @@ -604,9 +629,18 @@ async def commit_adjustment( ) -> ReportingAdjustmentRecord: async with self._pool.connection() as connection: await self._lock_account(connection, adjustment.account_id) + existing = await ( + await connection.execute( + f"SELECT {_ADJUSTMENT_COLUMNS} FROM reporting_adjustments" # noqa: S608 # nosec B608 + " WHERE reporting_adjustment_id = %s AND account_id = %s", + (adjustment.reporting_adjustment_id, adjustment.account_id), + ) + ).fetchone() + if existing is not None: + return _adjustment_from_row(existing) revision = await ( await connection.execute( - "SELECT finality FROM reporting_revisions" + "SELECT finality, reporting_obligation_id FROM reporting_revisions" " WHERE reporting_revision_id = %s AND account_id = %s", (adjustment.adjusts_reporting_revision_id, adjustment.account_id), ) @@ -621,6 +655,18 @@ async def commit_adjustment( "adjustments correct an official revision; restate a snapshot with a " "superseding snapshot revision instead", ) + obligation = await ( + await connection.execute( + f"SELECT {_OBLIGATION_COLUMNS} FROM reporting_obligations" # noqa: S608 # nosec B608 + " WHERE reporting_obligation_id = %s AND account_id = %s", + (revision[1], adjustment.account_id), + ) + ).fetchone() + if obligation is None: + raise LedgerConflictError( + "OBLIGATION_NOT_FOUND", "the adjustment target has no obligation" + ) + validate_adjustment_currency(_obligation_from_row(obligation), adjustment) inserted = await ( await connection.execute( "INSERT INTO reporting_adjustments" @@ -1206,7 +1252,7 @@ def _translate_integrity_error(error: Exception) -> LedgerConflictError: " report_definition_id, reporting_profile, feed_purpose, period_key, period_start," " period_end, source_timezone, expected_at, scope_resolved_at," " automated_recovery_deadline_at, required_finality, coverage_status, media_buy_ids," - " package_ids, schedule, definition, created_at" + " package_ids, schedule, definition, created_at, currency" ) _REVISION_COLUMNS = ( @@ -1268,7 +1314,7 @@ def _schedule_payload(schedule: ReportingScheduleSpec) -> dict[str, Any]: def _definition_payload( definition: ReportingDefinitionBinding | None, ) -> dict[str, Any] | None: - return definition.to_wire() if definition is not None else None + return definition.to_storage() if definition is not None else None def _definition_from_payload(payload: dict[str, Any] | None) -> ReportingDefinitionBinding | None: @@ -1346,6 +1392,7 @@ def _obligation_from_row(row: Sequence[Any]) -> ReportingObligationRecord: schedule=_schedule_from_payload(row[18]), definition=_definition_from_payload(row[19]), created_at=_utc(row[20]), + currency=row[21], ) diff --git a/src/adcp/reporting/ledger/producer.py b/src/adcp/reporting/ledger/producer.py index 3801b8926..3c192dfc4 100644 --- a/src/adcp/reporting/ledger/producer.py +++ b/src/adcp/reporting/ledger/producer.py @@ -30,13 +30,19 @@ import asyncio import hashlib +import inspect import logging -from collections.abc import Callable, Mapping, Sequence -from dataclasses import dataclass, field +from collections.abc import Awaitable, Callable, Mapping, Sequence +from dataclasses import dataclass, field, replace from datetime import datetime, timedelta, timezone -from typing import Any +from typing import Any, TypeAlias from adcp.reporting.canonical_json import canonical_json_utf8_v1 +from adcp.reporting.currency import ( + ReportingCurrencyError, + require_frozen_currency, + validate_currency, +) from adcp.reporting.ledger.models import ( ReportingConfiguration, ReportingDeliveryEscalation, @@ -53,6 +59,7 @@ from adcp.reporting.source import ( MediaBuyConstituentV1, ReportingConstituent, + ReportingContractIdentityV1, ReportingPublicationClass, ReportingSourceCoverageRequestV1, ReportingSourceExecutor, @@ -67,6 +74,8 @@ ) __all__ = [ + "CurrencyResolver", + "FixedCurrencyResolver", "ProducerOfferings", "ReportingProducer", "WorkerTurn", @@ -76,6 +85,34 @@ logger = logging.getLogger(__name__) +CurrencyResolver: TypeAlias = Callable[ + [ReportingConfiguration, ReportingObligationRecord], Awaitable[str] | str +] +"""Resolve from trusted account/definition state at the obligation's period end. + +The candidate obligation supplies the account-qualified generation, historical +scope and boundary. It has no currency yet. No buyer context or source response +is passed. Implementations must reject mixed-currency media-buy/package scope; +``require_single_currency`` can check their historical constituent values. +An async resolver may load the account context for a future reporting service. +""" + + +@dataclass(frozen=True) +class FixedCurrencyResolver: + """The backward-compatible single-currency default, also usable explicitly.""" + + currency: str = "USD" + + def __post_init__(self) -> None: + validate_currency(self.currency) + + def __call__( + self, configuration: ReportingConfiguration, obligation: ReportingObligationRecord + ) -> str: + return self.currency + + def _utc(value: datetime) -> datetime: return value.astimezone(timezone.utc) @@ -172,6 +209,7 @@ def __init__( lease_seconds: float = 60.0, max_periods_per_turn: int = 64, clock: Callable[[], datetime] | None = None, + currency_resolver: CurrencyResolver | None = None, ) -> None: self._source = source self._offerings = offerings @@ -182,6 +220,11 @@ def __init__( self._lease_seconds = lease_seconds self._max_periods_per_turn = max_periods_per_turn self._clock = clock or (lambda: datetime.now(timezone.utc)) + self._currency_resolver = ( + currency_resolver + if currency_resolver is not None + else FixedCurrencyResolver(offerings.currency) + ) @property def store(self) -> ReportingLedgerStore: @@ -341,6 +384,9 @@ async def _close_elapsed_periods( definition=configuration.definition, created_at=now, ) + resolved = self._currency_resolver(configuration, obligation) + currency = await resolved if inspect.isawaitable(resolved) else resolved + obligation = replace(obligation, currency=validate_currency(currency)) stored = await self._store.commit_obligation(obligation) committed.append(stored) turn.obligations_committed.append(stored.reporting_obligation_id) @@ -418,7 +464,21 @@ async def _acquire_pending( ) if obligation is None: continue - await self.acquire_obligation(configuration, obligation, turn=turn, now=now) + try: + await self.acquire_obligation(configuration, obligation, turn=turn, now=now) + except ReportingCurrencyError as error: + # One obligation whose money cannot be interpreted -- a legacy + # period with no retained currency, or a source contradicting + # the frozen one -- is a stuck slice, not a broken worker. + # Raising here would starve every later period under this + # configuration on every turn, forever. + logger.info( + "reporting slice failed obligation=%s code=%s", + obligation.reporting_obligation_id, + error.code, + ) + turn.slices_failed.append(obligation.reporting_obligation_id) + self._note_escalation(obligation, turn, now=now) async def acquire_obligation( self, @@ -444,6 +504,7 @@ async def acquire_obligation( "CONFIGURATION_GENERATION_MISMATCH", "the source configuration must belong to the obligation's account and generation", ) + obligation = await self._stored_obligation(obligation) turn = turn or WorkerTurn() now = now or self._clock() revisions = await self._store.list_revisions( @@ -457,6 +518,11 @@ async def acquire_obligation( satisfied = any(item.readable for item in revisions) if satisfied and not restate: return None + # Everything below needs the frozen code: the slice request carries it, + # the manifest is checked against it, and the revision is written under + # it. Gate here rather than earlier so a settled legacy obligation stays + # the no-op it already was instead of becoming an error on every turn. + require_frozen_currency(obligation.currency) finality = obligation.required_finality offering_id = self._offerings.offering_for(finality) @@ -495,6 +561,7 @@ async def acquire_obligation( return None manifest = self._verified_manifest(result) + self._validate_manifest_currency(obligation, manifest) return await self.commit_revision_from_manifest( obligation, manifest, @@ -567,6 +634,8 @@ async def commit_revision_from_manifest( edit path. An official close is terminal, so a later source correction must arrive as an adjustment instead. """ + obligation = await self._stored_obligation(obligation) + self._validate_manifest_currency(obligation, manifest) now = now or self._clock() turn = turn or WorkerTurn() existing = await self._store.list_revisions( @@ -610,6 +679,71 @@ async def commit_revision_from_manifest( turn.revisions_committed.append(committed.reporting_revision_id) return committed + async def _stored_obligation( + self, obligation: ReportingObligationRecord + ) -> ReportingObligationRecord: + """Always use the durable winner, including for public low-level calls.""" + stored = await self._store.get_obligation( + account_id=obligation.account_id, + reporting_obligation_id=obligation.reporting_obligation_id, + ) + if stored is None: + raise LedgerConflictError( + "OBLIGATION_NOT_FOUND", "commit the obligation before source work" + ) + if stored.generation_key != obligation.generation_key: + raise LedgerConflictError( + "CONFIGURATION_GENERATION_MISMATCH", "the obligation's retained generation differs" + ) + return stored + + @staticmethod + def _validate_manifest_currency( + obligation: ReportingObligationRecord, manifest: SourceBatchManifestV1 + ) -> None: + currency = require_frozen_currency(obligation.currency) + if manifest.currency != currency: + raise ReportingCurrencyError( + "CURRENCY_MISMATCH", "source manifest currency disagrees with the frozen obligation" + ) + identity = manifest.identity + if ( + identity.account_id != obligation.account_id + or identity.reporting_obligation_id != obligation.reporting_obligation_id + or identity.delivery_config_id != obligation.delivery_config_id + or identity.delivery_config_version != obligation.delivery_config_version + or identity.report_definition_id != obligation.report_definition_id + ): + raise LedgerConflictError("MANIFEST_MISMATCH", "manifest does not bind this obligation") + definition = obligation.definition + units = {"spend": currency} + ReportingProducer._validate_definition_binding(obligation, manifest.contract) + if definition is not None: + units.update(definition.monetary_metric_units) + units.update(definition.monetary_control_total_units) + for total in manifest.control_totals: + expected = units.get(total.name) + if expected is not None and total.unit is not None and total.unit != expected: + raise ReportingCurrencyError( + "CURRENCY_MISMATCH", + "source control total unit disagrees with the frozen currency", + ) + + @staticmethod + def _validate_definition_binding( + obligation: ReportingObligationRecord, contract: ReportingContractIdentityV1 + ) -> None: + definition = obligation.definition + if definition is not None and ( + contract.report_definition_id != obligation.report_definition_id + or contract.reporting_profile != obligation.reporting_profile + or any(getattr(contract, name) != value for name, value in definition.to_wire().items()) + ): + raise LedgerConflictError( + "REPORT_DEFINITION_MISMATCH", + "the source definition differs from the obligation's pin", + ) + @staticmethod def _current_snapshot_leaf(revisions: Sequence[ReportingRevisionRecord]) -> str | None: snapshots = [item for item in revisions if item.finality == "snapshot"] @@ -652,10 +786,11 @@ def _build_slice( as a new immutable observation. The ordinal comes from durable state, not from the clock, so neither behavior depends on wall time. """ + offering = self._source.capabilities.offering(offering_id) constituents: list[ReportingConstituent] = [ MediaBuyConstituentV1( constituent_id=media_buy_id, - product_id=configuration.report_definition_id, + product_id=obligation.report_definition_id, media_buy_id=media_buy_id, ) for media_buy_id in obligation.media_buy_ids @@ -696,7 +831,7 @@ def _build_slice( offering_id=offering_id, publication_namespace=self._offerings.publication_namespace, publication_class=publication_class, - contract=self._source.capabilities.offering(offering_id).contract, + contract=offering.contract, period=ReportingSourcePeriodV1( period_key=obligation.period.period_key, source_local_date=_source_local_date( @@ -722,7 +857,7 @@ def _build_slice( ), requested_metrics=list(self._offerings.requested_metrics), requested_dimensions=list(self._offerings.requested_dimensions), - currency=self._offerings.currency, + currency=require_frozen_currency(obligation.currency), deadline_at=_utc(now) + self._offerings.slice_timeout, ) @@ -740,6 +875,7 @@ def _logical_slice_fingerprint(obligation: ReportingObligationRecord, offering_i "period_start": _utc(obligation.period.start).isoformat(), "period_end": _utc(obligation.period.end).isoformat(), "media_buy_ids": sorted(obligation.media_buy_ids), + "currency": require_frozen_currency(obligation.currency), } ) diff --git a/src/adcp/reporting/ledger/reporting_ledger.sql b/src/adcp/reporting/ledger/reporting_ledger.sql index 839408c2f..3d391954f 100644 --- a/src/adcp/reporting/ledger/reporting_ledger.sql +++ b/src/adcp/reporting/ledger/reporting_ledger.sql @@ -1,6 +1,7 @@ -- AdCP Reliable Reporting ledger — durable obligations, revisions, and status. -- --- Run this followed by reporting_ledger_account_generations.sql in ONE +-- Run this followed by reporting_ledger_account_generations.sql and +-- reporting_ledger_obligation_currency.sql in ONE -- transaction (psql --single-transaction -f ... -f ...), or call -- PgReportingLedgerStore.create_schema(). CREATE TABLE IF NOT EXISTS alone -- does not upgrade the global configuration primary key from 8.0.0-beta.15. @@ -81,7 +82,10 @@ CREATE TABLE IF NOT EXISTS reporting_obligations ( -- definition that changes later must not retroactively re-describe a -- period that already closed. definition JSONB, - created_at TIMESTAMPTZ NOT NULL + created_at TIMESTAMPTZ NOT NULL, + -- NULL is unknown legacy currency, never an implicit USD default. The + -- currency migration installs validation and immutability on all schemas. + currency TEXT COLLATE "C" ); -- One obligation per logical period. Without this, two workers racing a period diff --git a/src/adcp/reporting/ledger/reporting_ledger_obligation_currency.sql b/src/adcp/reporting/ledger/reporting_ledger_obligation_currency.sql new file mode 100644 index 000000000..19f17c4df --- /dev/null +++ b/src/adcp/reporting/ledger/reporting_ledger_obligation_currency.sql @@ -0,0 +1,63 @@ +-- #1171: retain the currency frozen before source work. Apply after the #1169 +-- account-generations migration. This DO statement is atomic in autocommit too. +-- Stop older writers first: they cannot freeze currency for new obligations. +-- +-- There is deliberately NO backfill and NO USD default. Neither beta.15 nor +-- #1169 retained sufficient evidence to prove arbitrary historical currency. +-- NULL means unknown: history remains readable, but new source work, revisions +-- and adjustments must fail closed. See docs/reporting-ledger-migration.md. + +DO $obligation_currency$ +DECLARE + currency_type OID; + currency_default TEXT; +BEGIN + PERFORM pg_advisory_xact_lock(hashtext('adcp.reporting.schema'), hashtext(current_schema())); + LOCK TABLE reporting_obligations IN ACCESS EXCLUSIVE MODE; + + ALTER TABLE reporting_obligations ADD COLUMN IF NOT EXISTS currency TEXT COLLATE "C"; + + SELECT a.atttypid, pg_get_expr(d.adbin, d.adrelid) + INTO currency_type, currency_default + FROM pg_attribute a + LEFT JOIN pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum + WHERE a.attrelid = 'reporting_obligations'::regclass AND a.attname = 'currency'; + IF currency_type <> 'text'::regtype OR currency_default IS NOT NULL THEN + RAISE EXCEPTION 'Unexpected reporting_obligations.currency type or default' + USING HINT = 'Inspect the adopter schema; currency must be text with no inferred default.'; + END IF; + + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conrelid = 'reporting_obligations'::regclass + AND conname = 'reporting_obligations_currency_code' + ) THEN + ALTER TABLE reporting_obligations ADD CONSTRAINT reporting_obligations_currency_code + CHECK (currency IS NULL OR (octet_length(currency) = 3 AND currency COLLATE "C" ~ '^[A-Z]{3}$')); + END IF; + + -- Protect the new immutable fact even from a direct UPDATE/upsert. A + -- historical repair needs its own reviewed, evidence-backed migration. + CREATE OR REPLACE FUNCTION reporting_obligation_currency_immutable() + RETURNS TRIGGER LANGUAGE plpgsql AS $function$ + BEGIN + IF NEW.currency IS DISTINCT FROM OLD.currency THEN + RAISE EXCEPTION 'reporting obligation currency is immutable' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; + END + $function$; + + IF NOT EXISTS ( + SELECT 1 FROM pg_trigger + WHERE tgrelid = 'reporting_obligations'::regclass + AND tgname = 'reporting_obligation_currency_immutable' + AND NOT tgisinternal + ) THEN + CREATE TRIGGER reporting_obligation_currency_immutable + BEFORE UPDATE OF currency ON reporting_obligations + FOR EACH ROW EXECUTE FUNCTION reporting_obligation_currency_immutable(); + END IF; +END +$obligation_currency$; diff --git a/src/adcp/reporting/ledger/store.py b/src/adcp/reporting/ledger/store.py index b190f879d..4f9a9ca16 100644 --- a/src/adcp/reporting/ledger/store.py +++ b/src/adcp/reporting/ledger/store.py @@ -41,6 +41,12 @@ from typing import Any, Literal, Protocol, runtime_checkable from adcp.reporting.canonical_json import canonical_json_utf8_v1 +from adcp.reporting.currency import ( + monetary_decimal, + require_frozen_currency, + validate_currency_units, + validate_monetary_content, +) from adcp.reporting.ledger.health import issue_id_for_occurrence from adcp.reporting.ledger.models import ( ConsumerStatusRecord, @@ -195,6 +201,8 @@ async def commit_obligation( Idempotent by ``(account, config generation, period)``, not by id: two workers racing a period close must converge on one obligation. + New records require an explicit, validated currency. A legacy record + with unknown currency can be read/replayed but cannot be filled here. """ ... @@ -613,6 +621,7 @@ async def commit_obligation( "OBLIGATION_IDENTITY_CONFLICT", "the obligation identifier already belongs to a different logical period", ) + require_frozen_currency(obligation.currency) self._obligations[obligation.reporting_obligation_id] = obligation self._obligation_by_period[key] = obligation.reporting_obligation_id self._append(obligation.account_id, "obligation", obligation.reporting_obligation_id) @@ -654,6 +663,14 @@ async def find_obligation( async def commit_revision( self, revision: ReportingRevisionRecord, rows: Sequence[dict[str, Any]] ) -> ReportingRevisionRecord: + # Checked first, exactly as PgReportingLedgerStore does: a caller whose + # rows do not match its own declared count must get the same code from + # both stores, not whichever invariant that store happens to reach. + if revision.row_count != len(rows): + raise LedgerConflictError( + "ROW_COUNT_MISMATCH", + f"revision declares {revision.row_count} rows but {len(rows)} were supplied", + ) async with self._lock: identity = _revision_identity(revision) existing = self._revisions.get(revision.reporting_revision_id) @@ -671,6 +688,7 @@ async def commit_revision( "OBLIGATION_NOT_FOUND", "a revision must attach to an obligation committed at the period close", ) + validate_revision_currency(obligation, revision, rows) siblings = [ item for item in self._revisions.values() @@ -686,11 +704,6 @@ async def commit_revision( ) if revision.supersedes_reporting_revision_id: self._require_current_leaf(revision, siblings) - if revision.row_count != len(rows): - raise LedgerConflictError( - "ROW_COUNT_MISMATCH", - f"revision declares {revision.row_count} rows but {len(rows)} were supplied", - ) self._revisions[revision.reporting_revision_id] = revision self._revision_identity[revision.reporting_revision_id] = identity self._rows[revision.reporting_revision_id] = tuple(dict(row) for row in rows) @@ -792,6 +805,9 @@ async def commit_adjustment( "adjustments correct an official revision; restate a snapshot with a " "superseding snapshot revision instead", ) + validate_adjustment_currency( + self._obligations[revision.reporting_obligation_id], adjustment + ) self._adjustments[adjustment.reporting_adjustment_id] = adjustment self._append(adjustment.account_id, "adjustment", adjustment.reporting_adjustment_id) return adjustment @@ -1186,6 +1202,7 @@ def _config_payload(configuration: ReportingConfiguration) -> dict[str, Any]: "account_timezone": configuration.account_timezone, "authoritative_party": configuration.authoritative_party, "media_buy_ids": sorted(configuration.media_buy_ids), + "definition": configuration.definition.to_storage() if configuration.definition else None, "schedule": { "period_duration": schedule.period_duration, "delivery_sla": schedule.delivery_sla, @@ -1196,3 +1213,35 @@ def _config_payload(configuration: ReportingConfiguration) -> dict[str, Any]: ), }, } + + +def validate_revision_currency( + obligation: ReportingObligationRecord, + revision: ReportingRevisionRecord, + rows: Sequence[dict[str, Any]], +) -> None: + """Shared write gate; low-level stores enforce the same monetary invariant.""" + currency = require_frozen_currency(obligation.currency) + definition = obligation.definition + validate_monetary_content( + currency=currency, + rows=rows, + totals=revision.control_totals, + metric_units=definition.monetary_metric_units if definition else (), + total_units=definition.monetary_control_total_units if definition else (), + ) + + +def validate_adjustment_currency( + obligation: ReportingObligationRecord, adjustment: ReportingAdjustmentRecord +) -> None: + """Deltas inherit units from the target's obligation; they cannot re-resolve.""" + currency = require_frozen_currency(obligation.currency) + units = {"spend": currency} + if obligation.definition is not None: + units.update(obligation.definition.monetary_metric_units) + units.update(obligation.definition.monetary_control_total_units) + validate_currency_units(currency, units.items()) + for name, value in adjustment.control_total_deltas: + if name in units: + monetary_decimal(value) diff --git a/tests/conformance/reporting/_generation_support.py b/tests/conformance/reporting/_generation_support.py index 8bcd1059a..3ad45f96b 100644 --- a/tests/conformance/reporting/_generation_support.py +++ b/tests/conformance/reporting/_generation_support.py @@ -109,6 +109,7 @@ def obligation_for(config: ReportingConfiguration) -> ReportingObligationRecord: schedule=config.schedule, definition=config.definition, created_at=END, + currency="USD", ) diff --git a/tests/conformance/reporting/test_pg_reporting_ledger.py b/tests/conformance/reporting/test_pg_reporting_ledger.py index 8a973c6d1..fdaa4907e 100644 --- a/tests/conformance/reporting/test_pg_reporting_ledger.py +++ b/tests/conformance/reporting/test_pg_reporting_ledger.py @@ -149,6 +149,7 @@ def _obligation(configuration: ReportingConfiguration, ordinal: int = 0, **overr automated_recovery_deadline_at=boundary.expected_at + timedelta(hours=6), schedule=configuration.schedule, created_at=boundary.end, + currency="USD", ) defaults.update(overrides) return ReportingObligationRecord(**defaults) # type: ignore[arg-type] diff --git a/tests/conformance/reporting/test_reporting_core_lifecycle.py b/tests/conformance/reporting/test_reporting_core_lifecycle.py index bf45cb899..9c077d9fe 100644 --- a/tests/conformance/reporting/test_reporting_core_lifecycle.py +++ b/tests/conformance/reporting/test_reporting_core_lifecycle.py @@ -77,6 +77,7 @@ ) from adcp.reporting.ledger.models import first_ordinal_after # noqa: E402 from adcp.reporting.ledger.pg import PgReportingLedgerStore # noqa: E402 +from adcp.reporting.source import reporting_source_capabilities_sha256_v1 # noqa: E402 from adcp.types import ( # noqa: E402 GetReportingStatusRequest, GetReportingStatusResponse, @@ -248,8 +249,25 @@ def _producer( store: PgReportingLedgerStore, source: SimulatedSource, *, now: datetime ) -> tuple[ReportingProducer, InlineReportingSource]: staging = InMemoryStagingStore() + # The source must advertise the definition pinned by these configurations; + # the generic fixture advertises a different example definition. + capabilities = redacted_capabilities() + contract = capabilities.offerings[0].contract.model_copy( + update={"report_definition_id": DEFINITION_ID, **DEFINITION.to_wire()} + ) + capabilities = capabilities.model_copy( + update={ + "offerings": [ + offering.model_copy(update={"contract": contract}) + for offering in capabilities.offerings + ] + } + ) + capabilities = capabilities.model_copy( + update={"capabilities_sha256": reporting_source_capabilities_sha256_v1(capabilities)} + ) executor = InlineReportingSource( - capabilities=redacted_capabilities(), + capabilities=capabilities, fetch=source, staging=staging, seals=InMemorySealStore(), diff --git a/tests/conformance/reporting/test_reporting_currency.py b/tests/conformance/reporting/test_reporting_currency.py new file mode 100644 index 000000000..63c7031b7 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_currency.py @@ -0,0 +1,1159 @@ +"""The same currency trust boundary against memory and real PostgreSQL.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator, Callable +from dataclasses import replace +from datetime import datetime, timedelta, timezone +from decimal import Decimal, localcontext +from typing import Any + +import pytest + +from adcp.reporting import ( + ReportingContentReading, + ReportingPinnedDefinition, + classify_content_mismatch, +) +from adcp.reporting.conformance import ( + ReportingSourceConformanceError, + run_reporting_source_replay_conformance, + validate_reporting_source_execution, +) +from adcp.reporting.fixtures import ( + OFFICIAL_OFFERING_ID, + SNAPSHOT_OFFERING_ID, + redacted_capabilities, + redacted_contract_identity, + redacted_multi_currency_requests, +) +from adcp.reporting.inline_source import ( + InlineFetch, + InlineFetchResult, + InlineReportingSource, + InMemorySealStore, + InMemoryStagingStore, +) +from adcp.reporting.ledger import ( + CurrencyResolver, + FixedCurrencyResolver, + InMemoryReportingLedgerStore, + LedgerConflictError, + ProducerOfferings, + ReportingAdjustmentRecord, + ReportingConfiguration, + ReportingCurrencyError, + ReportingDefinitionBinding, + ReportingLedgerStore, + ReportingObligationRecord, + ReportingProducer, + ReportingScheduleSpec, + ReportingStatusCaller, + ReportingStatusHandler, + require_single_currency, + revision_content_sha256, +) +from adcp.reporting.ledger.pg import PgReportingLedgerStore +from adcp.reporting.source import ( + ReportingSourceCapabilitiesV1, + ReportingSourceExecutorResult, + ReportingSourceSliceRequestV1, + SourceBatchManifestV1, + encode_source_batch_manifest_v1, + parse_verified_source_batch_manifest_v1, + publication_content_fingerprint_v1, + reporting_source_capabilities_sha256_v1, + source_batch_manifest_reference_v1, +) +from adcp.types import GetMediaBuyDeliveryResponse, GetReportingStatusResponse + +from ._generation_support import START, UncalledSource, isolated_reporting_pool, revision_for + +END = START + timedelta(days=1) +NOW = END + timedelta(days=2) + + +@pytest.fixture(params=["memory", "postgres"]) +async def store(request: pytest.FixtureRequest) -> AsyncIterator[ReportingLedgerStore]: + if request.param == "memory": + yield InMemoryReportingLedgerStore(clock=lambda: NOW) + else: + async with isolated_reporting_pool() as pool: + ledger = PgReportingLedgerStore(pool=pool, clock=lambda: NOW) + await ledger.create_schema() + yield ledger + + +def configuration( + account: str = "eur", *, pinned_currency: str | None = None +) -> ReportingConfiguration: + contract = redacted_contract_identity + units = (("spend", pinned_currency),) if pinned_currency else () + return ReportingConfiguration( + account_id=account, + delivery_config_id="daily", + delivery_config_version=1, + report_definition_id=contract.report_definition_id, + reporting_profile=contract.reporting_profile, + feed_purpose="analytics", + schedule=ReportingScheduleSpec("P1D", "PT1H", period_anchor=START), + required_finality="snapshot", + activated_at=START, + deactivated_at=END, + media_buy_ids=(f"mb_{account}",), + definition=ReportingDefinitionBinding( + report_definition_uri=contract.report_definition_uri, + report_definition_sha256=contract.report_definition_sha256, + schema_version=contract.schema_version, + schema_uri=contract.schema_uri, + schema_sha256=contract.schema_sha256, + monetary_metric_units=units, + monetary_control_total_units=units, + ), + ) + + +def capabilities() -> ReportingSourceCapabilitiesV1: + base = redacted_capabilities() + draft = base.model_copy( + update={ + "offerings": [ + offering.model_copy( + update={ + "source_timezone": "UTC", + "product_ids": [configuration().report_definition_id], + } + ) + for offering in base.offerings + ] + } + ) + return ReportingSourceCapabilitiesV1.model_validate( + draft.model_copy( + update={"capabilities_sha256": reporting_source_capabilities_sha256_v1(draft)} + ) + ) + + +def source_for(fetch: InlineFetch) -> tuple[InlineReportingSource, InMemoryStagingStore]: + staging = InMemoryStagingStore() + return ( + InlineReportingSource( + capabilities=capabilities(), + fetch=fetch, + staging=staging, + seals=InMemorySealStore(), + clock=lambda: NOW, + ), + staging, + ) + + +def producer_for( + store: ReportingLedgerStore, + source: Any, + staging: InMemoryStagingStore | None = None, + *, + currency: str = "USD", + resolver: CurrencyResolver | None = None, +) -> ReportingProducer: + return ReportingProducer( + store=store, + source=source, + object_reader=staging, + offerings=ProducerOfferings( + snapshot_offering_id=SNAPSHOT_OFFERING_ID, + official_offering_id=OFFICIAL_OFFERING_ID, + publication_namespace="reporting-source:fixture", + source_scope=dict(redacted_capabilities().source_scope), + currency=currency, + ), + currency_resolver=resolver, + clock=lambda: NOW, + ) + + +def money_rows(request: ReportingSourceSliceRequestV1) -> list[dict[str, Any]]: + return [ + { + "media_buy_id": request.coverage.constituents[0].media_buy_id, + "impressions": 5, + "spend": "0.10", + "currency": request.currency, + }, + { + "media_buy_id": request.coverage.constituents[0].media_buy_id, + "impressions": 10, + "spend": "0.20", + "currency": request.currency, + }, + ] + + +async def test_concurrent_accounts_freeze_before_acquisition_and_survive_restarts( + store: ReportingLedgerStore, +) -> None: + configs = [configuration("usd"), configuration("eur")] + currencies = {"usd": "USD", "eur": "EUR"} + calls: list[str] = [] + + async def resolver(config: ReportingConfiguration, candidate: ReportingObligationRecord) -> str: + assert candidate.currency is None + assert candidate.generation_key == config.generation_key + assert candidate.scope_resolved_at == END + calls.append(candidate.account_id) + await asyncio.sleep(0) + return currencies[candidate.account_id] + + requests: list[ReportingSourceSliceRequestV1] = [] + ready = False + + async def fetch(request: ReportingSourceSliceRequestV1) -> InlineFetchResult | None: + requests.append(request) + # Model concurrent I/O even in memory (completed coroutines need not + # yield on Python 3.12), while each worker still holds its lease. + await asyncio.sleep(0) + stored = await store.get_obligation( + account_id=request.identity.account_id, + reporting_obligation_id=request.identity.reporting_obligation_id, + ) + assert stored is not None and stored.currency == request.currency + if not ready: + return None + return InlineFetchResult(rows=money_rows(request), currency=request.currency) + + source, staging = source_for(fetch) + producer = producer_for(store, source, staging, resolver=resolver) + await asyncio.gather(*(store.put_configuration(config) for config in configs)) + # One producer concurrently leases both accounts' same-name generation. + turns = await asyncio.gather(producer.run_worker(), producer.run_worker()) + assert {turn.leased.account_id for turn in turns if turn.leased} == {"usd", "eur"} + assert all( + len(turn.obligations_committed) == 1 and len(turn.slices_failed) == 1 for turn in turns + ) + assert sorted(calls) == ["eur", "usd"] + assert {request.currency for request in requests} == {"EUR", "USD"} + first_keys = { + request.identity.account_id: request.identity.source_execution_key for request in requests + } + + # The resolver's account state and the process default both change before + # retry. Neither is consulted for an already committed obligation. + currencies.update(usd="GBP", eur="JPY") + ready = True + restarted_store = ( + PgReportingLedgerStore(pool=store._pool, clock=lambda: NOW) + if isinstance(store, PgReportingLedgerStore) + else store + ) + + def must_not_resolve( + config: ReportingConfiguration, candidate: ReportingObligationRecord + ) -> str: + raise AssertionError("retry/restart/restatement must not resolve currency") + + restarted = producer_for( + restarted_store, source, staging, currency="GBP", resolver=must_not_resolve + ) + results = await asyncio.gather(restarted.run_worker(), restarted.run_worker()) + assert all(len(turn.revisions_committed) == 1 for turn in results) + assert all( + request.identity.source_execution_key == first_keys[request.identity.account_id] + for request in requests + ) + + for config in configs: + obligation = await restarted_store.find_obligation( + account_id=config.account_id, + delivery_config_id="daily", + delivery_config_version=1, + period_start=START, + period_end=END, + ) + assert obligation is not None + expected = "EUR" if config.account_id == "eur" else "USD" + assert obligation.currency == expected + # A replaced caller object cannot replace the stored currency. + restated = await restarted.acquire_obligation( + config, + replace(obligation, currency="GBP"), + restate=True, + ) + assert restated is not None and restated.supersedes_reporting_revision_id + assert requests[-1].currency == expected + assert requests[-1].identity.source_execution_key != first_keys[config.account_id] + assert await restarted.close_elapsed_periods(config) == [] + handler = ReportingStatusHandler(restarted_store) + payload = await handler.handle( + {"view": "periods", "context": {"currency": "JPY"}}, + caller=ReportingStatusCaller(account_id=config.account_id, consumer_id="buyer"), + ) + parsed = GetReportingStatusResponse.model_validate(payload) + current = next( + item + for item in parsed.revisions + if item.reporting_revision_id == restated.reporting_revision_id + ) + wire_obligation = parsed.periods[0] + rows = await restarted_store.read_revision_rows( + account_id=config.account_id, + reporting_revision_id=restated.reporting_revision_id, + ) + assert {row["currency"] for row in rows.rows} == {expected} + assert dict(restated.control_totals)["spend"] == "0.30" + assert restated.revision_content_sha256 == revision_content_sha256( + reporting_revision_id=restated.reporting_revision_id, + row_count=2, + control_totals=restated.control_totals, + reporting_rows=rows.rows, + ) + reading = ReportingContentReading( + reporting_revision_id=restated.reporting_revision_id, + media_buy_ids=config.media_buy_ids, + metric_names=("impressions", "spend"), + metric_units={"spend": expected}, + control_total_units={"spend": expected}, + ) + definition = ReportingPinnedDefinition( + metric_names=("impressions", "spend"), + metric_units={"spend": expected}, + control_total_units={"spend": expected}, + ) + assert ( + classify_content_mismatch( + obligation=wire_obligation, + revision=current, + reading=reading, + definition=definition, + ) + is None + ) + assert ( + classify_content_mismatch( + obligation=wire_obligation, + revision=current, + reading=replace(reading, metric_units={"spend": "JPY"}), + definition=definition, + ) + == "currency_mismatch" + ) + assert ( + await restarted_store.get_obligation( + account_id=config.account_id, + reporting_obligation_id=obligation.reporting_obligation_id, + ) + ) == obligation + + +@pytest.mark.parametrize( + "invalid", ["usd", "Usd", "US", "EURO", "US1", " EUR", "EUR\n", "ÅBC", "", None, ["USD", "EUR"]] +) +async def test_invalid_resolution_cannot_commit_or_touch_source( + store: ReportingLedgerStore, + invalid: Any, +) -> None: + producer = producer_for(store, UncalledSource(), resolver=lambda config, candidate: invalid) + with pytest.raises(ReportingCurrencyError, match="INVALID_CURRENCY"): + await producer.close_elapsed_periods(configuration()) + snapshot = await store.open_snapshot(account_id="eur", filters_fingerprint="") + assert snapshot.max_sequence == 0 + + +@pytest.mark.parametrize("currency", ["USD", "EUR"]) +async def test_fixed_currency_option_and_sync_resolver_are_supported( + store: ReportingLedgerStore, + currency: str, +) -> None: + for account, resolver in [("default", None), ("explicit", FixedCurrencyResolver(currency))]: + producer = producer_for(store, UncalledSource(), currency=currency, resolver=resolver) + (obligation,) = await producer.close_elapsed_periods(configuration(account)) + assert obligation.currency == currency + + +async def test_concurrent_resolutions_converge_on_one_immutable_winner( + store: ReportingLedgerStore, +) -> None: + entered = asyncio.Event() + count = 0 + + async def resolve(config: ReportingConfiguration, candidate: ReportingObligationRecord) -> str: + nonlocal count + count += 1 + result = "EUR" if count == 1 else "USD" + if count == 2: + entered.set() + await entered.wait() + return result + + producer = producer_for(store, UncalledSource(), resolver=resolve) + left, right = await asyncio.gather( + producer.close_elapsed_periods(configuration()), + producer.close_elapsed_periods(configuration()), + ) + assert left == right and left[0].currency in {"USD", "EUR"} + assert (await store.open_snapshot(account_id="eur", filters_fingerprint="")).max_sequence == 1 + + +async def test_trusted_mixed_scope_is_rejected_at_freeze(store: ReportingLedgerStore) -> None: + config = replace(configuration(), media_buy_ids=("mb_usd", "mb_eur")) + currencies = {"mb_usd": "USD", "mb_eur": "EUR"} + producer = producer_for( + store, + UncalledSource(), + resolver=lambda accepted, candidate: require_single_currency( + currencies[buy] for buy in candidate.media_buy_ids + ), + ) + with pytest.raises(ReportingCurrencyError, match="MIXED_CURRENCY_SCOPE"): + await producer.close_elapsed_periods(config) + assert (await store.open_snapshot(account_id="eur", filters_fingerprint="")).max_sequence == 0 + + +@pytest.mark.parametrize("mixed", [False, True]) +async def test_pinned_definition_units_are_checked_before_source( + store: ReportingLedgerStore, + mixed: bool, +) -> None: + config = configuration(pinned_currency="USD") + assert config.definition is not None + if mixed: + config = replace( + config, + definition=replace( + config.definition, + monetary_control_total_units=(("spend", "EUR"),), + ), + ) + producer = producer_for(store, UncalledSource(), currency="EUR") + with pytest.raises( + ReportingCurrencyError, match="MIXED_CURRENCY_SCOPE" if mixed else "CURRENCY_MISMATCH" + ): + await producer.close_elapsed_periods(config) + + +class CorruptingSource: + def __init__( + self, source: InlineReportingSource, corrupt: Callable[[dict[str, Any]], None] + ) -> None: + self.source = source + self.corrupt = corrupt + self.request: ReportingSourceSliceRequestV1 | None = None + self.result: ReportingSourceExecutorResult | None = None + + @property + def capabilities(self) -> ReportingSourceCapabilitiesV1: + return self.source.capabilities + + async def execute( + self, + request: ReportingSourceSliceRequestV1, + *, + cancel: asyncio.Event, + heartbeat: Callable[[], None] | None = None, + ) -> ReportingSourceExecutorResult: + result = await self.source.execute(request, cancel=cancel) + assert result.response is not None and result.manifest_bytes is not None + manifest = parse_verified_source_batch_manifest_v1( + result.response.manifest, result.manifest_bytes + ) + payload = manifest.model_dump(mode="json", exclude_none=True) + self.corrupt(payload) + payload["content_fingerprint"] = publication_content_fingerprint_v1(payload) + raw = encode_source_batch_manifest_v1(SourceBatchManifestV1.model_validate(payload)) + self.request = request + self.result = ReportingSourceExecutorResult.completed( + request=request, + manifest=source_batch_manifest_reference_v1("staged-corrupt", raw), + manifest_bytes=raw, + ) + return self.result + + +@pytest.mark.parametrize("corruption", ["manifest_currency", "total_unit", "total_value"]) +async def test_adapter_currency_and_monetary_totals_are_only_evidence( + store: ReportingLedgerStore, + corruption: str, +) -> None: + def corrupt(payload: dict[str, Any]) -> None: + if corruption == "manifest_currency": + payload["currency"] = "USD" + else: + spend = next(item for item in payload["control_totals"] if item["name"] == "spend") + spend["unit" if corruption == "total_unit" else "value"] = ( + "USD" if corruption == "total_unit" else "99.99" + ) + + source, staging = source_for(money_rows) + adapter = CorruptingSource(source, corrupt) + producer = producer_for(store, adapter, staging, currency="EUR") + config = configuration(pinned_currency="EUR") + (obligation,) = await producer.close_elapsed_periods(config) + code = "MONETARY_TOTAL_MISMATCH" if corruption == "total_value" else "CURRENCY_MISMATCH" + with pytest.raises(ReportingCurrencyError, match=code): + await producer.acquire_obligation(config, obligation) + assert ( + await store.list_revisions( + account_id="eur", reporting_obligation_id=obligation.reporting_obligation_id + ) + == () + ) + if corruption != "total_value": + assert adapter.request is not None and adapter.result is not None + with pytest.raises(ReportingSourceConformanceError, match="MANIFEST_MISMATCH"): + await validate_reporting_source_execution( + capabilities=adapter.capabilities, + request=adapter.request.model_copy( + update={"deadline_at": datetime.now(timezone.utc) + timedelta(hours=1)} + ), + result=adapter.result, + object_reader=staging, + ) + + +@pytest.mark.parametrize("corruption", ["row_currency", "mixed_rows", "source_currency"]) +async def test_inline_source_rejects_currency_before_aggregation( + store: ReportingLedgerStore, + monkeypatch: pytest.MonkeyPatch, + corruption: str, +) -> None: + def fetch(request: ReportingSourceSliceRequestV1) -> InlineFetchResult: + rows = money_rows(request) + if corruption == "mixed_rows": + rows[0]["currency"] = "USD" + elif corruption == "row_currency": + for row in rows: + row["currency"] = "USD" + return InlineFetchResult( + rows=rows, currency="USD" if corruption == "source_currency" else None + ) + + def must_not_aggregate(*args: Any) -> None: + raise AssertionError("mixed/mismatched money was aggregated") + + monkeypatch.setattr("adcp.reporting.inline_source._control_totals", must_not_aggregate) + source, staging = source_for(fetch) + producer = producer_for(store, source, staging, currency="EUR") + config = configuration() + (obligation,) = await producer.close_elapsed_periods(config) + request = producer._build_slice(config, obligation, SNAPSHOT_OFFERING_ID, now=NOW) + result = await source.execute(request, cancel=asyncio.Event()) + assert not result.ok and result.error is not None + assert result.error.code == "INTEGRITY_FAILED" + assert ( + "MIXED_CURRENCY_SCOPE" if corruption == "mixed_rows" else "CURRENCY_MISMATCH" + ) in result.error.safe_message + + +async def test_official_and_adjustments_keep_frozen_units(store: ReportingLedgerStore) -> None: + config = replace(configuration(pinned_currency="EUR"), required_finality="official") + source, staging = source_for(money_rows) + first = producer_for(store, source, staging, currency="EUR") + (obligation,) = await first.close_elapsed_periods(config) + restarted = producer_for(store, source, staging, currency="USD") + official = await restarted.acquire_obligation(config, obligation) + assert official is not None and official.finality == "official" + adjustment = ReportingAdjustmentRecord( + reporting_adjustment_id="adj_eur", + account_id="eur", + adjusts_reporting_revision_id=official.reporting_revision_id, + reason_code="source_correction", + accounting_period_start=END, + accounting_period_end=END + timedelta(days=1), + control_total_deltas=(("spend", "-0.10"),), + correction_observed_at=NOW, + created_at=NOW, + ) + assert await store.commit_adjustment(adjustment) == adjustment + assert await restarted.acquire_obligation(config, obligation, restate=True) is None + assert ( + await store.get_obligation( + account_id="eur", reporting_obligation_id=obligation.reporting_obligation_id + ) + == obligation + ) + with pytest.raises(ReportingCurrencyError, match="MONETARY_TOTAL_MISMATCH"): + await store.commit_adjustment( + replace( + adjustment, + reporting_adjustment_id="bad_delta", + control_total_deltas=(("spend", "NaN"),), + ) + ) + + +async def test_low_level_writes_require_currency_and_validate_frozen_money( + store: ReportingLedgerStore, +) -> None: + config = configuration() + (obligation,) = await producer_for( + store, UncalledSource(), currency="EUR" + ).close_elapsed_periods(config) + assert await store.commit_obligation(replace(obligation, currency="USD")) == obligation + unknown = replace( + obligation, reporting_obligation_id="new_unknown", account_id="other", currency=None + ) + with pytest.raises(ReportingCurrencyError, match="CURRENCY_UNRESOLVED"): + await store.commit_obligation(unknown) + revision, _ = revision_for(obligation) + rows = [{"media_buy_id": "mb_eur", "spend": "0.10", "currency": "USD"}] + with pytest.raises(ReportingCurrencyError, match="CURRENCY_MISMATCH"): + await store.commit_revision(replace(revision, control_totals=(("spend", "0.10"),)), rows) + rows[0]["currency"] = "EUR" + with pytest.raises(ReportingCurrencyError, match="MONETARY_TOTAL_MISMATCH"): + await store.commit_revision(replace(revision, control_totals=(("spend", "0.11"),)), rows) + # The check itself is independent of an adopter's Decimal precision. + rows[0]["spend"] = "123456789012345678901234567890.12" + totals = (("spend", rows[0]["spend"]),) + valid = replace( + revision, + control_totals=totals, + revision_content_sha256=revision_content_sha256( + reporting_revision_id=revision.reporting_revision_id, + row_count=len(rows), + control_totals=totals, + reporting_rows=rows, + ), + ) + with localcontext() as context: + context.prec = 4 + committed = await store.commit_revision(valid, rows) + assert committed.control_totals[0][1] == rows[0]["spend"] + + +async def test_pinned_monetary_semantics_are_immutable_in_both_stores( + store: ReportingLedgerStore, +) -> None: + config = configuration(pinned_currency="EUR") + await store.put_configuration(config) + await store.put_configuration(config) + assert await store.list_configurations(account_id="eur") == (config,) + with pytest.raises(LedgerConflictError, match="different content"): + await store.put_configuration(configuration(pinned_currency="USD")) + + +async def test_sync_callable_returning_a_coroutine_is_awaited(store: ReportingLedgerStore) -> None: + async def answer() -> str: + return "EUR" + + producer = producer_for(store, UncalledSource(), resolver=lambda config, candidate: answer()) + (obligation,) = await producer.close_elapsed_periods(configuration()) + assert obligation.currency == "EUR" + + +async def test_sealed_source_retry_after_commit_failure_keeps_currency( + store: ReportingLedgerStore, + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = 0 + + def fetch(request: ReportingSourceSliceRequestV1) -> list[dict[str, Any]]: + nonlocal calls + calls += 1 + return money_rows(request) + + source, staging = source_for(fetch) + config = configuration(pinned_currency="EUR") + producer = producer_for(store, source, staging, currency="EUR") + (obligation,) = await producer.close_elapsed_periods(config) + original_commit = store.commit_revision + + async def crash(*args: Any) -> Any: + raise RuntimeError("crash after source seal") + + monkeypatch.setattr(store, "commit_revision", crash) + with pytest.raises(RuntimeError, match="crash after source seal"): + await producer.acquire_obligation(config, obligation) + monkeypatch.setattr(store, "commit_revision", original_commit) + restarted = producer_for(store, source, staging, currency="USD") + committed = await restarted.acquire_obligation(config, obligation) + assert committed is not None + assert calls == 1 # The sealed source publication replays without refetch. + assert dict(committed.control_totals)["spend"] == "0.30" + rows = await store.read_revision_rows( + account_id="eur", reporting_revision_id=committed.reporting_revision_id + ) + assert {row["currency"] for row in rows.rows} == {"EUR"} + + +async def test_restored_memory_history_is_readable_but_cannot_be_resolved() -> None: + store = InMemoryReportingLedgerStore(clock=lambda: NOW) + config = configuration() + producer = producer_for(store, UncalledSource(), currency="EUR") + (current,) = await producer.close_elapsed_periods(config) + # Simulate rehydrating a record written before the optional Python field + # existed; regular new writes deliberately refuse this representation. + old = replace(current, currency=None) + store._obligations[old.reporting_obligation_id] = old + + def must_not_resolve( + config: ReportingConfiguration, candidate: ReportingObligationRecord + ) -> str: + raise AssertionError("legacy history cannot use a current account lookup") + + restarted = producer_for(store, UncalledSource(), resolver=must_not_resolve) + assert await restarted.close_elapsed_periods(config) == [] + assert await store.commit_obligation(replace(old, currency="USD")) == old + with pytest.raises(ReportingCurrencyError, match="CURRENCY_UNRESOLVED"): + await restarted.acquire_obligation(config, replace(old, currency="USD")) + revision, rows = revision_for(old) + with pytest.raises(ReportingCurrencyError, match="CURRENCY_UNRESOLVED"): + await store.commit_revision(revision, rows) + payload = await ReportingStatusHandler(store).handle( + {"view": "summary"}, + caller=ReportingStatusCaller(account_id="eur", consumer_id="buyer"), + ) + assert payload["health"] == "action_required" + assert payload["issues"][0]["code"] == "HISTORY_UNAVAILABLE" + assert ( + await store.get_obligation( + account_id="eur", reporting_obligation_id=old.reporting_obligation_id + ) + == old + ) + + +async def test_multi_currency_source_fixtures_pass_replay_conformance() -> None: + staging = InMemoryStagingStore() + requests = redacted_multi_currency_requests() + source = InlineReportingSource( + capabilities=redacted_capabilities(), + fetch=money_rows, + staging=staging, + clock=lambda: requests[0].period.source_read_cutoff_at, + ) + manifests = await asyncio.gather( + *( + run_reporting_source_replay_conformance( + executor=source, request=request, object_reader=staging + ) + for request in requests + ) + ) + assert {manifest.currency for manifest in manifests} == {"USD", "EUR"} + assert len({manifest.publication_id for manifest in manifests}) == 2 + for manifest in manifests: + spend = next(total for total in manifest.control_totals if total.name == "spend") + assert spend.unit == manifest.currency and spend.value == "0.30" + + +async def test_non_usd_zero_rows_keep_the_frozen_currency(store: ReportingLedgerStore) -> None: + source, staging = source_for(lambda request: []) + producer = producer_for(store, source, staging, currency="EUR") + config = configuration(pinned_currency="EUR") + (obligation,) = await producer.close_elapsed_periods(config) + revision = await producer.acquire_obligation(config, obligation) + assert revision is not None and revision.row_count == 0 + assert dict(revision.control_totals)["spend"] == "0" + assert obligation.currency == "EUR" + + +async def test_manifest_cannot_substitute_a_different_pinned_definition( + store: ReportingLedgerStore, +) -> None: + def corrupt(payload: dict[str, Any]) -> None: + payload["contract"]["report_definition_sha256"] = "f" * 64 + + source, staging = source_for(money_rows) + producer = producer_for(store, CorruptingSource(source, corrupt), staging, currency="EUR") + config = configuration(pinned_currency="EUR") + (obligation,) = await producer.close_elapsed_periods(config) + with pytest.raises(LedgerConflictError) as caught: + await producer.acquire_obligation(config, obligation) + assert caught.value.code == "REPORT_DEFINITION_MISMATCH" + assert ( + await store.list_revisions( + account_id="eur", reporting_obligation_id=obligation.reporting_obligation_id + ) + == () + ) + + +async def test_postgres_currency_survives_closing_every_application_connection() -> None: + async with isolated_reporting_pool() as original_pool: + from psycopg_pool import AsyncConnectionPool + + store = PgReportingLedgerStore(pool=original_pool, clock=lambda: NOW) + await store.create_schema() + configs = [configuration("usd"), configuration("eur")] + currencies = {"usd": "USD", "eur": "EUR"} + producer = producer_for( + store, + UncalledSource(), + resolver=lambda config, candidate: currencies[candidate.account_id], + ) + for config in configs: + await store.put_configuration(config) + await producer.close_elapsed_periods(config) + async with original_pool.connection() as connection: + row = await (await connection.execute("SELECT current_schema()")).fetchone() + assert row is not None + schema = row[0] + await original_pool.close() + currencies.update(usd="GBP", eur="JPY") + + async with AsyncConnectionPool( + original_pool.conninfo, + kwargs={"options": f"-csearch_path={schema} -cstatement_timeout=15000"}, + open=False, + ) as restarted_pool: + await restarted_pool.wait(timeout=10) + restarted_store = PgReportingLedgerStore(pool=restarted_pool, clock=lambda: NOW) + await restarted_store.create_schema() + source, staging = source_for(money_rows) + + def must_not_resolve( + config: ReportingConfiguration, candidate: ReportingObligationRecord + ) -> str: + raise AssertionError("restart must use PostgreSQL's frozen currency") + + restarted = producer_for( + restarted_store, source, staging, currency="GBP", resolver=must_not_resolve + ) + for config in configs: + assert await restarted.close_elapsed_periods(config) == [] + obligation = await restarted_store.find_obligation( + account_id=config.account_id, + delivery_config_id=config.delivery_config_id, + delivery_config_version=config.delivery_config_version, + period_start=START, + period_end=END, + ) + assert obligation is not None + expected = "EUR" if config.account_id == "eur" else "USD" + assert obligation.currency == expected + for restate in (False, True): + revision = await restarted.acquire_obligation( + config, obligation, restate=restate + ) + assert revision is not None + content = await restarted_store.read_revision_rows( + account_id=config.account_id, + reporting_revision_id=revision.reporting_revision_id, + ) + assert {row["currency"] for row in content.rows} == {expected} + + +@pytest.mark.parametrize( + ("shape", "expected_error"), + [ + ("response", "CURRENCY_MISMATCH"), + ("media_buy", "CURRENCY_MISMATCH"), + ("package", "CURRENCY_MISMATCH"), + ("mixed_packages", "MIXED_CURRENCY_SCOPE"), + ("reporting_rows", "CURRENCY_MISMATCH"), + ("matching_eur", None), + ], +) +async def test_delivery_response_currency_is_checked_before_flattening( + store: ReportingLedgerStore, + monkeypatch: pytest.MonkeyPatch, + shape: str, + expected_error: str | None, +) -> None: + def fetch(request: ReportingSourceSliceRequestV1) -> GetMediaBuyDeliveryResponse: + delivery: dict[str, Any] = { + "media_buy_id": "mb_eur", + "status": "active", + "totals": {"impressions": 15, "spend": 0.30}, + "by_package": [], + } + payload: dict[str, Any] = { + "reporting_period": {"start": START, "end": END}, + "media_buy_deliveries": [delivery], + } + if shape in {"response", "reporting_rows", "matching_eur"}: + payload["currency"] = "EUR" if shape == "matching_eur" else "USD" + if shape in {"reporting_rows", "matching_eur"}: + # Core's canonical row format uses exact decimal strings; the + # ordinary delivery surface models its legacy metrics as floats. + payload["reporting_rows"] = money_rows(request) + if shape == "media_buy": + delivery["currency"] = "USD" + if shape in {"package", "mixed_packages"}: + codes = ["USD", "EUR"] if shape == "mixed_packages" else ["USD"] + delivery["totals"] = {"impressions": 15} + delivery["by_package"] = [ + { + "package_id": f"pkg_{code}", + "pricing_model": "cpm", + "rate": 1.0, + "currency": code, + "impressions": 5, + "spend": 0.10, + } + for code in codes + ] + return GetMediaBuyDeliveryResponse.model_validate(payload) + + source, staging = source_for(fetch) + producer = producer_for(store, source, staging, currency="EUR") + config = configuration() + (obligation,) = await producer.close_elapsed_periods(config) + if expected_error is None: + revision = await producer.acquire_obligation(config, obligation) + assert revision is not None + assert Decimal(dict(revision.control_totals)["spend"]) == Decimal("0.30") + assert obligation.currency == "EUR" + else: + + def must_not_aggregate(*args: Any) -> None: + raise AssertionError("delivery-response currency was discarded before aggregation") + + monkeypatch.setattr("adcp.reporting.inline_source._control_totals", must_not_aggregate) + request = producer._build_slice(config, obligation, SNAPSHOT_OFFERING_ID, now=NOW) + result = await source.execute(request, cancel=asyncio.Event()) + assert result.error is not None and result.error.code == "INTEGRITY_FAILED" + assert result.error.retry == "terminal" + assert expected_error in result.error.safe_message + + +async def test_inline_money_is_frozen_before_staging_awaits( + store: ReportingLedgerStore, + monkeypatch: pytest.MonkeyPatch, +) -> None: + shared: list[dict[str, Any]] = [] + + def fetch(request: ReportingSourceSliceRequestV1) -> InlineFetchResult: + shared.extend(money_rows(request)) + return InlineFetchResult(rows=shared, currency="EUR") + + source, staging = source_for(fetch) + original_stage = staging.stage + + async def stage(**kwargs: Any) -> tuple[str, str]: + shared[0]["currency"] = "USD" + shared[0]["spend"] = "99.99" + return await original_stage(**kwargs) + + monkeypatch.setattr(staging, "stage", stage) + producer = producer_for(store, source, staging, currency="EUR") + config = configuration(pinned_currency="EUR") + (obligation,) = await producer.close_elapsed_periods(config) + revision = await producer.acquire_obligation(config, obligation) + assert revision is not None and dict(revision.control_totals)["spend"] == "0.30" + content = await store.read_revision_rows( + account_id="eur", reporting_revision_id=revision.reporting_revision_id + ) + assert {row["currency"] for row in content.rows} == {"EUR"} + assert content.rows[0]["spend"] == "0.10" + assert shared[0]["currency"] == "USD" and shared[0]["spend"] == "99.99" + + +@pytest.mark.parametrize("monetary", [False, True]) +async def test_control_total_units_use_trusted_semantics_instead_of_guessing_from_shape( + store: ReportingLedgerStore, + monetary: bool, +) -> None: + def corrupt(payload: dict[str, Any]) -> None: + payload["control_totals"].append( + { + "name": "custom_total", + "value": "1", + "value_type": "integer", + "unit": "USD" if monetary else "GRP", + } + ) + + source, staging = source_for(money_rows) + adapter = CorruptingSource(source, corrupt) + producer = producer_for(store, adapter, staging, currency="EUR") + config = configuration(pinned_currency="EUR") + if monetary: + assert config.definition is not None + config = replace( + config, + definition=replace( + config.definition, + monetary_control_total_units=(("spend", "EUR"), ("custom_total", "EUR")), + ), + ) + (obligation,) = await producer.close_elapsed_periods(config) + if monetary: + with pytest.raises(ReportingCurrencyError, match="CURRENCY_MISMATCH"): + await producer.acquire_obligation(config, obligation) + else: + # Nonmonetary units remain valid even if they have three capital letters. + revision = await producer.acquire_obligation(config, obligation) + assert revision is not None and dict(revision.control_totals)["custom_total"] == "1" + assert adapter.request is not None and adapter.result is not None + manifest = await validate_reporting_source_execution( + capabilities=adapter.capabilities, + request=adapter.request.model_copy( + update={"deadline_at": datetime.now(timezone.utc) + timedelta(hours=1)} + ), + result=adapter.result, + object_reader=staging, + ) + assert next(t for t in manifest.control_totals if t.name == "custom_total").unit == "GRP" + + +async def _forget_currency( + store: ReportingLedgerStore, obligation: ReportingObligationRecord +) -> None: + """Rewrite one obligation the way a pre-#1171 ledger retained it.""" + if isinstance(store, InMemoryReportingLedgerStore): + store._obligations[obligation.reporting_obligation_id] = replace(obligation, currency=None) + return + trigger = "reporting_obligation_currency_immutable" + async with store._pool.connection() as connection: + await connection.execute(f"ALTER TABLE reporting_obligations DISABLE TRIGGER {trigger}") + await connection.execute( + "UPDATE reporting_obligations SET currency = NULL WHERE reporting_obligation_id = %s", + (obligation.reporting_obligation_id,), + ) + await connection.execute(f"ALTER TABLE reporting_obligations ENABLE TRIGGER {trigger}") + + +@pytest.mark.parametrize("settled", [True, False]) +async def test_an_unresolved_legacy_obligation_cannot_starve_later_periods( + store: ReportingLedgerStore, settled: bool +) -> None: + """Upgrading must not turn one unknown period into a permanently dead worker.""" + config = replace(configuration(), deactivated_at=END + timedelta(days=1)) + await store.put_configuration(config) + source, staging = source_for(money_rows) + producer = producer_for(store, source, staging, currency="EUR") + legacy, later = await producer.close_elapsed_periods(config) + if settled: + settled_revision = await producer.acquire_obligation(config, legacy) + assert settled_revision is not None and settled_revision.readable + await _forget_currency(store, legacy) + + turn = await producer.run_worker() + + assert turn.leased is not None and turn.leased.generation_key == config.generation_key + # The period that *does* have a proven currency is published on this turn. + published = await store.list_revisions( + account_id="eur", reporting_obligation_id=later.reporting_obligation_id + ) + assert [item.reporting_revision_id for item in published] == turn.revisions_committed + assert len(turn.revisions_committed) == 1 + # A settled legacy period is the no-op it always was; an unfulfilled one is + # a stuck slice the supervisor can see, not an exception out of the turn. + assert turn.slices_failed == ([] if settled else [legacy.reporting_obligation_id]) + retained = await store.get_obligation( + account_id="eur", reporting_obligation_id=legacy.reporting_obligation_id + ) + assert retained is not None and retained.currency is None + assert len( + await store.list_revisions( + account_id="eur", reporting_obligation_id=legacy.reporting_obligation_id + ) + ) == (1 if settled else 0) + # Repeating the turn stays stable rather than compounding the quarantine. + assert not (await producer.run_worker()).revisions_committed + + +async def test_a_metric_absent_from_some_rows_needs_no_invented_total( + store: ReportingLedgerStore, +) -> None: + """A sparse money column has no honest sum; it must not block publication. + + An omitted cell and an explicit ``null`` say the same thing -- not + reported -- for both a money column and a row's ``currency`` label, so + neither an unlabeled row nor a missing price is evidence of a second + currency or of a measured zero. + """ + + def fetch(request: ReportingSourceSliceRequestV1) -> InlineFetchResult: + buy = request.coverage.constituents[0].media_buy_id + return InlineFetchResult( + rows=[ + {"media_buy_id": buy, "impressions": 5, "spend": "0.10", "currency": "EUR"}, + {"media_buy_id": buy, "impressions": 7, "currency": None}, + ] + ) + + source, staging = source_for(fetch) + producer = producer_for(store, source, staging, currency="EUR") + # No pinned monetary control total: a definition that pins one is entitled + # to demand it, but the built-in spend handling must not invent that demand. + config = configuration() + (obligation,) = await producer.close_elapsed_periods(config) + revision = await producer.acquire_obligation(config, obligation) + assert revision is not None + # The inline adapter declines to publish a spend total here, and the ledger + # agrees rather than demanding one it cannot derive. + assert dict(revision.control_totals) == {"impressions": "12"} + content = await store.read_revision_rows( + account_id="eur", reporting_revision_id=revision.reporting_revision_id + ) + assert [row.get("spend") for row in content.rows] == ["0.10", None] + + # A total that *is* published still has to reconcile against every row. + # The null cell is what makes this irreconcilable: read as a zero it would + # sum to the published 0.10 and commit. + sparse = [ + {"media_buy_id": "mb_eur", "spend": "0.10"}, + {"media_buy_id": "mb_eur", "impressions": 7, "spend": None}, + ] + unreconcilable = replace( + revision_for(obligation)[0], + reporting_revision_id="rpr_eur_sparse_total", + row_count=2, + control_totals=(("spend", "0.10"),), + ) + with pytest.raises(ReportingCurrencyError, match="MONETARY_TOTAL_MISMATCH"): + await store.commit_revision(unreconcilable, sparse) + + +async def test_row_count_is_rejected_before_money_in_both_stores( + store: ReportingLedgerStore, +) -> None: + """Both stores answer a miscounted revision with the same code.""" + config = configuration() + (obligation,) = await producer_for( + store, UncalledSource(), currency="EUR" + ).close_elapsed_periods(config) + revision, _ = revision_for(obligation) + with pytest.raises(LedgerConflictError) as caught: + await store.commit_revision( + replace(revision, control_totals=(("spend", "9.99"),)), + [ + {"media_buy_id": "mb_eur", "spend": "0.10"}, + {"media_buy_id": "mb_eur", "spend": "0.20"}, + ], + ) + assert caught.value.code == "ROW_COUNT_MISMATCH" + + +@pytest.mark.parametrize("pinned_total", [False, True]) +async def test_a_period_with_no_rows_owes_no_derived_monetary_total( + store: ReportingLedgerStore, pinned_total: bool +) -> None: + """The demand for a total is derived from rows, so an empty period owes none. + + Two rows that both omit ``spend`` report no spend, and a period with no + rows at all reports no less than that -- deriving a demand from the empty + case would make zero rows stricter than two. A definition that pins a + monetary *control total* is the declaration meaning "always present", and + it still fails closed here. + """ + config = configuration(pinned_currency="EUR") + assert config.definition is not None + config = replace( + config, + definition=replace( + config.definition, + monetary_control_total_units=(("spend", "EUR"),) if pinned_total else (), + ), + ) + (obligation,) = await producer_for( + store, UncalledSource(), currency="EUR" + ).close_elapsed_periods(config) + revision_id = "rpr_eur_empty" + empty = replace( + revision_for(obligation)[0], + reporting_revision_id=revision_id, + row_count=0, + control_totals=(), + revision_content_sha256=revision_content_sha256( + reporting_revision_id=revision_id, row_count=0, control_totals=(), reporting_rows=[] + ), + ) + if pinned_total: + with pytest.raises(ReportingCurrencyError, match="MONETARY_TOTAL_MISMATCH"): + await store.commit_revision(empty, []) + return + assert (await store.commit_revision(empty, [])).row_count == 0 diff --git a/tests/conformance/reporting/test_reporting_currency_migration.py b/tests/conformance/reporting/test_reporting_currency_migration.py new file mode 100644 index 000000000..5c1a5d3c8 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_currency_migration.py @@ -0,0 +1,258 @@ +"""No invented currency when upgrading literal beta.15 and #1169 ledgers.""" + +from __future__ import annotations + +import asyncio +import hashlib +from dataclasses import replace +from importlib.resources import files +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import pytest + +from adcp.reporting.ledger import ( + ProducerOfferings, + ReportingConfiguration, + ReportingCurrencyError, + ReportingObligationRecord, + ReportingProducer, + ReportingStatusCaller, + ReportingStatusHandler, +) +from adcp.reporting.ledger.pg import PgReportingLedgerStore +from adcp.types import GetReportingStatusResponse + +from ._generation_support import ( + NOW, + UncalledSource, + configuration, + isolated_reporting_pool, + revision_for, +) +from .test_reporting_generation_migration import _TABLES, _primary_key + +if TYPE_CHECKING: + from psycopg_pool import AsyncConnectionPool + +FIXTURES = Path(__file__).resolve().parents[2] / "fixtures" +MIGRATION = files("adcp.reporting.ledger").joinpath("reporting_ledger_obligation_currency.sql") +ACCOUNT_MIGRATION = files("adcp.reporting.ledger").joinpath( + "reporting_ledger_account_generations.sql" +) + + +def test_1169_schema_fixture_is_the_reviewed_stacked_base() -> None: + # 91fa2786efdd3117ce2c1a875ede7431767d7e87, byte for byte. + assert hashlib.sha256((FIXTURES / "reporting_ledger_1169.sql").read_bytes()).hexdigest() == ( + "6617d1c840b50530266f5126ec20f86c4b032735f51ecc78f983db5ac070a911" + ) + + +async def retained(pool: AsyncConnectionPool) -> dict[str, list[Any]]: + from psycopg import sql + + result = {} + async with pool.connection() as connection: + for table in _TABLES: + rows = await ( + await connection.execute( + sql.SQL("SELECT to_jsonb(t) FROM {} t ORDER BY to_jsonb(t)::text").format( + sql.Identifier(table) + ) + ) + ).fetchall() + result[table] = [row[0] for row in rows] + return result + + +async def raw_upgrade(pool: AsyncConnectionPool) -> None: + async with pool.connection() as connection: + async with connection.transaction(): + await connection.execute(ACCOUNT_MIGRATION.read_text()) + await connection.execute(MIGRATION.read_text()) + + +@pytest.mark.parametrize("schema", ["reporting_ledger_beta15.sql", "reporting_ledger_1169.sql"]) +@pytest.mark.parametrize("autocommit", [False, True]) +async def test_migration_preserves_evidence_and_quarantines_unknown_currency( + schema: str, + autocommit: bool, +) -> None: + async with isolated_reporting_pool(autocommit=autocommit) as pool: + async with pool.connection() as connection: + await connection.execute((FIXTURES / schema).read_text()) + await connection.execute((FIXTURES / "reporting_ledger_beta15_data.sql").read_text()) + # An unfulfilled old obligation demonstrates that neither a live + # account lookup nor new process options can fill historical gaps. + await connection.execute( + "INSERT INTO reporting_obligations SELECT (jsonb_populate_record(" + "NULL::reporting_obligations, to_jsonb(o) || " + '\'{"reporting_obligation_id":"legacy_pending","delivery_config_id":"pending"}\'::jsonb)).*' + " FROM reporting_obligations o WHERE reporting_obligation_id = 'rpo_acct_a'" + ) + before = await retained(pool) + primary = await _primary_key(pool) + await asyncio.gather( + *(PgReportingLedgerStore(pool=pool).create_schema() for _ in range(3)), + *(raw_upgrade(pool) for _ in range(3)), + ) + after = await retained(pool) + # The ONLY record change is the explicit unknown column; all hashes, + # rows, units, issues, leases and sequence numbers are otherwise exact. + for record in before["reporting_obligations"]: + record["currency"] = None + assert after == before + assert (await _primary_key(pool))[ + 3 + ] == "PRIMARY KEY (account_id, delivery_config_id, delivery_config_version)" + if schema == "reporting_ledger_1169.sql": + assert await _primary_key(pool) == primary + await raw_upgrade(pool) + await PgReportingLedgerStore(pool=pool).create_schema() + assert await retained(pool) == after + + store = PgReportingLedgerStore(pool=pool, clock=lambda: NOW) + old = await store.get_obligation(account_id="acct_a", reporting_obligation_id="rpo_acct_a") + pending = await store.get_obligation( + account_id="acct_a", reporting_obligation_id="legacy_pending" + ) + assert old is not None and old.currency is None + assert pending is not None and pending.currency is None + assert await store.commit_obligation(replace(old, currency="EUR")) == old + revision = await store.get_revision( + account_id="acct_a", reporting_revision_id="rpr_acct_a_official" + ) + assert revision is not None + rows = await store.read_revision_rows( + account_id="acct_a", reporting_revision_id=revision.reporting_revision_id + ) + assert rows.rows == ({"media_buy_id": "mb_acct_a", "impressions": 5},) + assert await store.commit_revision(revision, rows.rows) == revision + adjustments = await store.list_adjustments( + account_id="acct_a", reporting_revision_ids=[revision.reporting_revision_id] + ) + assert len(adjustments) == 1 + assert await store.commit_adjustment(adjustments[0]) == adjustments[0] + + def must_not_resolve( + config: ReportingConfiguration, candidate: ReportingObligationRecord + ) -> str: + raise AssertionError("migration/retry cannot resolve a legacy obligation") + + producer = ReportingProducer( + store=store, + source=UncalledSource(), + offerings=ProducerOfferings(currency="EUR"), + currency_resolver=must_not_resolve, + clock=lambda: NOW, + ) + config = replace( + configuration(), delivery_config_id="pending", required_finality="official" + ) + assert await producer.close_elapsed_periods(config) == [] + # ``old`` closed officially before the upgrade: acquisition is already + # terminal, so it stays the no-op it was rather than becoming an error + # the worker loop would hit on every turn. Nothing is read or written. + assert ( + await producer.acquire_obligation( + replace(config, delivery_config_id=old.delivery_config_id), old, restate=True + ) + is None + ) + for candidate in (pending, replace(pending, currency="EUR")): + with pytest.raises(ReportingCurrencyError, match="CURRENCY_UNRESOLVED"): + await producer.acquire_obligation( + replace(config, delivery_config_id=candidate.delivery_config_id), + candidate, + restate=True, + ) + fresh, fresh_rows = revision_for(pending) + with pytest.raises(ReportingCurrencyError, match="CURRENCY_UNRESOLVED"): + await store.commit_revision(fresh, fresh_rows) + with pytest.raises(ReportingCurrencyError, match="CURRENCY_UNRESOLVED"): + await store.commit_adjustment( + replace(adjustments[0], reporting_adjustment_id="new_legacy_delta") + ) + + # Reads remain schema-valid and actionable, without re-labeling history. + handler = ReportingStatusHandler(store) + payload = await handler.handle( + {"view": "summary"}, + caller=ReportingStatusCaller(account_id="acct_a", consumer_id="buyer"), + ) + GetReportingStatusResponse.model_validate(payload) + assert payload["health"] == "action_required" + assert any(issue["code"] == "HISTORY_UNAVAILABLE" for issue in payload["issues"]) + assert await retained(pool) == after + + # Ordinary upgraded accounts still publish after the legacy quarantine. + new_config = configuration("eur") + await store.put_configuration(new_config) + next_producer = ReportingProducer( + store=store, + source=UncalledSource(), + offerings=ProducerOfferings(currency="EUR"), + clock=lambda: NOW, + ) + (new,) = await next_producer.close_elapsed_periods(new_config) + assert new.currency == "EUR" + + +async def test_currency_constraints_and_immutability_cover_direct_sql() -> None: + from psycopg.errors import CheckViolation + + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, clock=lambda: NOW) + await store.create_schema() + producer = ReportingProducer( + store=store, + source=UncalledSource(), + offerings=ProducerOfferings(currency="EUR"), + clock=lambda: NOW, + ) + (obligation,) = await producer.close_elapsed_periods(configuration()) + for value in ("USD", None, "eur", "EUR\n", "ÅBC", "US1", "EURO"): + with pytest.raises(CheckViolation): + async with pool.connection() as connection: + await connection.execute( + "UPDATE reporting_obligations SET currency = %s", (value,) + ) + async with pool.connection() as connection: + await connection.execute("UPDATE reporting_obligations SET currency = currency") + assert ( + await store.get_obligation( + account_id=obligation.account_id, + reporting_obligation_id=obligation.reporting_obligation_id, + ) + == obligation + ) + for value in ("eur", "EUR\n", "ÅBC", "US1", "EURO"): + with pytest.raises(CheckViolation): + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO reporting_obligations SELECT (jsonb_populate_record(" + "NULL::reporting_obligations, to_jsonb(o) || jsonb_build_object(" + "'currency', %s::text, 'account_id', 'invalid', " + "'reporting_obligation_id', 'invalid'))).*" + " FROM reporting_obligations o", + (value,), + ) + + +async def test_unexpected_default_rolls_back_without_rewriting_evidence() -> None: + from psycopg.errors import RaiseException + + async with isolated_reporting_pool() as pool: + async with pool.connection() as connection: + await connection.execute((FIXTURES / "reporting_ledger_beta15.sql").read_text()) + await connection.execute((FIXTURES / "reporting_ledger_beta15_data.sql").read_text()) + await connection.execute( + "ALTER TABLE reporting_obligations ADD COLUMN currency TEXT DEFAULT 'USD'" + ) + before = await retained(pool) + primary = await _primary_key(pool) + with pytest.raises(RaiseException, match="Unexpected reporting_obligations.currency"): + await PgReportingLedgerStore(pool=pool).create_schema() + assert await retained(pool) == before + assert await _primary_key(pool) == primary diff --git a/tests/conformance/reporting/test_reporting_generation_migration.py b/tests/conformance/reporting/test_reporting_generation_migration.py index 888b5225a..b953e9968 100644 --- a/tests/conformance/reporting/test_reporting_generation_migration.py +++ b/tests/conformance/reporting/test_reporting_generation_migration.py @@ -74,6 +74,12 @@ async def _retained_rows(pool: AsyncConnectionPool) -> dict[str, list[Any]]: ) ).fetchall() result[table] = [row[0] for row in rows] + if table == "reporting_obligations": + # #1171 adds an explicit unknown currency. This test still + # compares every byte of the pre-existing #1169 evidence. + for record in result[table]: + if record.get("currency") is None: + record.pop("currency", None) return result @@ -97,6 +103,7 @@ async def _other_constraints(pool: AsyncConnectionPool) -> list[Any]: "SELECT oid, conname, pg_get_constraintdef(oid) FROM pg_constraint" " WHERE connamespace = current_schema()::regnamespace" " AND NOT (conrelid = 'reporting_configurations'::regclass AND contype = 'p')" + " AND conname <> 'reporting_obligations_currency_code'" " ORDER BY oid" ) ).fetchall() diff --git a/tests/fixtures/reporting_ledger_1169.sql b/tests/fixtures/reporting_ledger_1169.sql new file mode 100644 index 000000000..839408c2f --- /dev/null +++ b/tests/fixtures/reporting_ledger_1169.sql @@ -0,0 +1,291 @@ +-- AdCP Reliable Reporting ledger — durable obligations, revisions, and status. +-- +-- Run this followed by reporting_ledger_account_generations.sql in ONE +-- transaction (psql --single-transaction -f ... -f ...), or call +-- PgReportingLedgerStore.create_schema(). CREATE TABLE IF NOT EXISTS alone +-- does not upgrade the global configuration primary key from 8.0.0-beta.15. +-- See docs/reporting-ledger-migration.md before upgrading live workers. +-- +-- COLLATE "C" on identifier columns avoids locale-dependent case folding — on +-- some locales "Buy-A" and "buy-a" compare equal, which would collapse two +-- distinct accounts or revisions into one. "C" is the byte-for-byte comparison +-- reporting evidence actually requires. +-- +-- Every immutable write also appends to reporting_ledger_changes in the same +-- transaction. That feed is what makes `changes_after` exact: a consumer that +-- persists a checkpoint and replays from it cannot miss a record or see the +-- same record twice under a different identity. + +-- Serialize bootstrap/upgrade before touching catalog objects: IF NOT EXISTS +-- by itself can still race another CREATE TABLE on an empty schema. The +-- standalone account-generations migration uses the same advisory lock. +SELECT pg_advisory_xact_lock(hashtext('adcp.reporting.schema'), hashtext(current_schema())); + +CREATE TABLE IF NOT EXISTS reporting_configurations ( + delivery_config_id TEXT COLLATE "C" NOT NULL, + delivery_config_version INTEGER NOT NULL, + account_id TEXT COLLATE "C" NOT NULL, + report_definition_id TEXT COLLATE "C" NOT NULL, + reporting_profile TEXT NOT NULL, + feed_purpose TEXT NOT NULL, + required_finality TEXT NOT NULL, + account_timezone TEXT NOT NULL DEFAULT 'UTC', + schedule JSONB NOT NULL, + media_buy_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + activated_at TIMESTAMPTZ, + deactivated_at TIMESTAMPTZ, + automated_recovery_seconds DOUBLE PRECISION NOT NULL DEFAULT 21600, + status_retention_days INTEGER NOT NULL DEFAULT 400, + -- Content-addressed report definition and row schema (URI + digest). Core + -- wire records are self-describing; without this a retained revision + -- cannot name what it was produced under. + definition JSONB, + -- Binds the whole generation so a re-put with changed content is a + -- detectable conflict rather than a silent edit of retained evidence. + content_sha256 TEXT COLLATE "C" NOT NULL, + -- Period-close leasing. A worker that dies mid-close releases its work by + -- expiry instead of wedging the period forever. + lease_worker_id TEXT COLLATE "C", + lease_expires_at TIMESTAMPTZ, + PRIMARY KEY (account_id, delivery_config_id, delivery_config_version) +); + +CREATE INDEX IF NOT EXISTS reporting_configurations_account_idx + ON reporting_configurations (account_id); + +-- Supports "lease the least recently worked generation" without a full scan. +CREATE INDEX IF NOT EXISTS reporting_configurations_lease_idx + ON reporting_configurations (lease_expires_at NULLS FIRST); + +CREATE TABLE IF NOT EXISTS reporting_obligations ( + reporting_obligation_id TEXT COLLATE "C" NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + delivery_config_id TEXT COLLATE "C" NOT NULL, + delivery_config_version INTEGER NOT NULL, + report_definition_id TEXT COLLATE "C" NOT NULL, + reporting_profile TEXT NOT NULL, + feed_purpose TEXT NOT NULL, + period_key TEXT NOT NULL, + period_start TIMESTAMPTZ NOT NULL, + period_end TIMESTAMPTZ NOT NULL, + source_timezone TEXT NOT NULL, + expected_at TIMESTAMPTZ NOT NULL, + scope_resolved_at TIMESTAMPTZ NOT NULL, + automated_recovery_deadline_at TIMESTAMPTZ NOT NULL, + required_finality TEXT NOT NULL, + coverage_status TEXT NOT NULL DEFAULT 'full', + media_buy_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + package_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + schedule JSONB NOT NULL, + -- Frozen with the obligation, not read from the live configuration: a + -- definition that changes later must not retroactively re-describe a + -- period that already closed. + definition JSONB, + created_at TIMESTAMPTZ NOT NULL +); + +-- One obligation per logical period. Without this, two workers racing a period +-- close could commit two obligations and a seller could quietly publish twice +-- and pick a winner. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_obligations_period_key + ON reporting_obligations + (account_id, delivery_config_id, delivery_config_version, period_start, period_end); + +CREATE INDEX IF NOT EXISTS reporting_obligations_account_idx + ON reporting_obligations (account_id, period_end DESC); + +CREATE TABLE IF NOT EXISTS reporting_revisions ( + reporting_revision_id TEXT COLLATE "C" NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + reporting_obligation_id TEXT COLLATE "C" NOT NULL + REFERENCES reporting_obligations (reporting_obligation_id), + finality TEXT NOT NULL, + revision_content_sha256 TEXT COLLATE "C" NOT NULL, + row_count BIGINT NOT NULL, + control_totals JSONB NOT NULL DEFAULT '[]'::jsonb, + observed_at TIMESTAMPTZ NOT NULL, + data_through TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL, + supersedes_reporting_revision_id TEXT COLLATE "C", + finality_basis TEXT, + finality_policy_id TEXT, + finalized_at TIMESTAMPTZ, + -- Core promises a committed revision stays readable for + -- status_retention_days. `readable` records reality; `readable_at_commit` + -- remembers that it once was, so the health projection can name the + -- revision that opened the gap rather than an arbitrary one. + readable BOOLEAN NOT NULL DEFAULT TRUE, + readable_at_commit BOOLEAN NOT NULL DEFAULT TRUE, + source_publication_id TEXT COLLATE "C", + source_manifest_sha256 TEXT COLLATE "C", + -- Binds what was published, excluding mutable readability. + content_sha256 TEXT COLLATE "C" NOT NULL +); + +-- An official revision is terminal: at most one per obligation. A later source +-- correction is an adjustment, never a second official close. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_revisions_one_official + ON reporting_revisions (reporting_obligation_id) + WHERE finality = 'official'; + +-- Supersession must not fork: a given revision may be superseded at most once. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_revisions_one_successor + ON reporting_revisions (supersedes_reporting_revision_id) + WHERE supersedes_reporting_revision_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS reporting_revisions_obligation_idx + ON reporting_revisions (account_id, reporting_obligation_id); + +CREATE TABLE IF NOT EXISTS reporting_revision_rows ( + reporting_revision_id TEXT COLLATE "C" NOT NULL + REFERENCES reporting_revisions (reporting_revision_id), + ordinal BIGINT NOT NULL, + row_payload JSONB NOT NULL, + PRIMARY KEY (reporting_revision_id, ordinal) +); + +CREATE TABLE IF NOT EXISTS reporting_adjustments ( + reporting_adjustment_id TEXT COLLATE "C" NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + adjusts_reporting_revision_id TEXT COLLATE "C" NOT NULL + REFERENCES reporting_revisions (reporting_revision_id), + reason_code TEXT NOT NULL, + reason_detail TEXT, + accounting_period_start TIMESTAMPTZ NOT NULL, + accounting_period_end TIMESTAMPTZ NOT NULL, + control_total_deltas JSONB NOT NULL DEFAULT '[]'::jsonb, + correction_observed_at TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL +); + +CREATE INDEX IF NOT EXISTS reporting_adjustments_revision_idx + ON reporting_adjustments (account_id, adjusts_reporting_revision_id); + +-- PREVIEW: sync_reporting_status. Created unconditionally because DDL is +-- cheap and a migration mid-rollout is not; the ingest that writes here is off +-- until an adopter enables it. See adcp/reporting/ledger/consumer_status.py. +CREATE TABLE IF NOT EXISTS reporting_consumer_statuses ( + reporting_status_id TEXT COLLATE "C" NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + -- Derived from authenticated transport, never from the request body. + consumer_id TEXT COLLATE "C" NOT NULL, + delivery_config_id TEXT COLLATE "C" NOT NULL, + delivery_config_version INTEGER NOT NULL, + report_definition_id TEXT COLLATE "C" NOT NULL, + period_start TIMESTAMPTZ NOT NULL, + period_end TIMESTAMPTZ NOT NULL, + period_source_timezone TEXT NOT NULL, + consumer_status TEXT NOT NULL, + status_as_of TIMESTAMPTZ NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL, + supersedes_reporting_status_id TEXT COLLATE "C", + -- Nullable by design: obligation_missing is filed precisely when the + -- seller's ledger omitted the period, so requiring a seller-issued + -- obligation id would make the first missing report invisible again. + reporting_obligation_id TEXT COLLATE "C", + reporting_revision_id TEXT COLLATE "C", + observed_revision_content_sha256 TEXT COLLATE "C", + failure_code TEXT, + consumer_commit_ref TEXT, + seller_ledger_snapshot_id TEXT, + seller_ledger_as_of TIMESTAMPTZ, + superseded BOOLEAN NOT NULL DEFAULT FALSE, + content_sha256 TEXT COLLATE "C" NOT NULL +); + +-- Exactly one unsuperseded leaf per logical chain. This is what makes +-- supersession atomic: a concurrent update naming a stale leaf hits this +-- constraint instead of forking the chain, so a successful retry cannot erase +-- a recorded outage. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_consumer_statuses_one_leaf + ON reporting_consumer_statuses + (account_id, consumer_id, delivery_config_id, delivery_config_version, + report_definition_id, period_start, period_end) + WHERE superseded = FALSE; + +CREATE UNIQUE INDEX IF NOT EXISTS reporting_consumer_statuses_one_successor + ON reporting_consumer_statuses (supersedes_reporting_status_id) + WHERE supersedes_reporting_status_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS reporting_consumer_statuses_chain_idx + ON reporting_consumer_statuses (account_id, consumer_id, period_end DESC); + +-- AdCP 3.2.0-rc.3 additions to tables created by an earlier SDK. ADD COLUMN +-- IF NOT EXISTS keeps create_schema() an upgrade path, not just a bootstrap: +-- an adopter that installed the rc.2 schema gets these on the next boot +-- without a hand-written migration, and a fresh install is unaffected. +ALTER TABLE reporting_consumer_statuses + ADD COLUMN IF NOT EXISTS mismatch_code TEXT; + +ALTER TABLE reporting_configurations + ADD COLUMN IF NOT EXISTS authoritative_party TEXT NOT NULL DEFAULT 'seller'; + +-- Durable issue lifecycle. Every other issue in this ledger has a *derived* +-- identity, because the conditions they name are monotone for one immutable +-- obligation: once a qualifying revision is associated, REPORT_OVERDUE cannot +-- recur. A consumer mismatch is not monotone -- the buyer can supersede, the +-- seller can restate, the disagreement can clear and come back -- and rc.3 +-- requires opened_at to survive every re-emission because it anchors the +-- escalation clock. A derived timestamp would reset on each poll and an +-- unattended mismatch would never escalate. +-- +-- issue_key identifies the condition; issue_id identifies one occurrence of +-- it. Retirement bumps generation, so a recurrence after resolved/waived gets +-- a new issue_id and a new opened_at, while an unresolved condition keeps both +-- across a severity change from delayed to action_required. +CREATE TABLE IF NOT EXISTS reporting_issue_lifecycle ( + issue_key TEXT COLLATE "C" NOT NULL, + account_id TEXT COLLATE "C" NOT NULL, + generation INTEGER NOT NULL, + issue_id TEXT COLLATE "C" NOT NULL, + -- Caller-scoped issues (every consumer mismatch) carry the consumer whose + -- statement caused them; NULL is a seller-wide condition. + consumer_id TEXT COLLATE "C", + opened_at TIMESTAMPTZ NOT NULL, + issue_state TEXT NOT NULL DEFAULT 'open', + -- Inert correlation text for the party's own tracker. Never dereferenced. + external_ref TEXT, + retired_at TIMESTAMPTZ, + PRIMARY KEY (account_id, issue_key, generation) +); + +-- At most one live occurrence per condition. 'waived' counts as live: waiving +-- records an agreement to stop *acting*, not a finding that the reporting is +-- fine, so the occurrence must keep blocking a new one -- otherwise the next +-- poll would open a fresh occurrence and republish the very issue the parties +-- agreed to stop acting on. Only 'resolved' frees the condition to recur. +-- +-- This is also what makes +-- ensure_issue_opened() safe under concurrent reads: two readers of the same +-- condition collide on this index and converge on one row instead of opening +-- two occurrences with two different opened_at values -- which would give the +-- same disagreement two escalation clocks. +CREATE UNIQUE INDEX IF NOT EXISTS reporting_issue_lifecycle_one_live + ON reporting_issue_lifecycle (account_id, issue_key) + WHERE issue_state IN ('open', 'acknowledged', 'waived'); + +CREATE UNIQUE INDEX IF NOT EXISTS reporting_issue_lifecycle_issue_id + ON reporting_issue_lifecycle (issue_id); + +-- The per-account change feed. `seq` orders every immutable record across +-- kinds so `changes_after` is exact. +-- +-- Appends take a transaction-scoped advisory lock on the account, which makes +-- sequence order equal commit order *within an account*. Without it, a +-- transaction that grabbed a low sequence but committed late would be invisible +-- to a consumer that had already checkpointed past it — a silently lost record, +-- which is the one failure a reporting ledger must not have. Contention is +-- per-account and the appends are short. +CREATE TABLE IF NOT EXISTS reporting_ledger_changes ( + seq BIGSERIAL NOT NULL PRIMARY KEY, + account_id TEXT COLLATE "C" NOT NULL, + record_kind TEXT NOT NULL, + record_id TEXT COLLATE "C" NOT NULL, + committed_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE UNIQUE INDEX IF NOT EXISTS reporting_ledger_changes_record + ON reporting_ledger_changes (account_id, record_kind, record_id); + +CREATE INDEX IF NOT EXISTS reporting_ledger_changes_feed_idx + ON reporting_ledger_changes (account_id, seq); diff --git a/tests/test_reporting_inline_source.py b/tests/test_reporting_inline_source.py index cacd30c41..11287e7ef 100644 --- a/tests/test_reporting_inline_source.py +++ b/tests/test_reporting_inline_source.py @@ -11,6 +11,7 @@ import asyncio import threading import time +import warnings from datetime import datetime, timedelta, timezone from typing import Any @@ -518,17 +519,18 @@ async def test_row_encoding_is_key_order_independent() -> None: # -- delivery-response projection ------------------------------------------- -async def test_a_get_media_buy_delivery_response_projects_into_rows() -> None: +@pytest.mark.parametrize("currency", ["USD", "EUR"]) +async def test_a_get_media_buy_delivery_response_projects_into_rows(currency: str) -> None: from adcp.types import GetMediaBuyDeliveryResponse - request = redacted_snapshot_request() + request = redacted_snapshot_request(currency=currency) response = GetMediaBuyDeliveryResponse.model_validate( { "reporting_period": { "start": request.period.start.isoformat(), "end": request.period.source_read_cutoff_at.isoformat(), }, - "currency": "USD", + "currency": currency, "media_buy_deliveries": [ { "media_buy_id": "media-buy-redacted", @@ -539,7 +541,7 @@ async def test_a_get_media_buy_delivery_response_projects_into_rows() -> None: "package_id": "pkg-1", "pricing_model": "cpm", "rate": 1.0, - "currency": "USD", + "currency": currency, "impressions": 6, "spend": 0.75, }, @@ -547,7 +549,7 @@ async def test_a_get_media_buy_delivery_response_projects_into_rows() -> None: "package_id": "pkg-2", "pricing_model": "cpm", "rate": 1.0, - "currency": "USD", + "currency": currency, "impressions": 4, "spend": 0.5, }, @@ -565,11 +567,47 @@ async def test_a_get_media_buy_delivery_response_projects_into_rows() -> None: ) assert manifest.row_count == 2 assert manifest.coverage.status == "full" + assert manifest.currency == currency totals = {total.name: total.value for total in manifest.control_totals} assert totals["impressions"] == "10" assert totals["spend"] == "1.25" +async def test_the_deprecated_response_currency_is_checked_without_warning() -> None: + """The legacy response-wide label still has to be read -- silently. + + ``GetMediaBuyDeliveryResponse.currency`` is deprecated in AdCP 3.2, so a + naive attribute read emits a DeprecationWarning on every single fetch. An + adopter running warnings as errors would see that surface as an opaque + ``PROVIDER_TRANSIENT`` failure instead of a published slice. + """ + from adcp.types import GetMediaBuyDeliveryResponse + + request = redacted_snapshot_request(currency="EUR") + response = GetMediaBuyDeliveryResponse.model_validate( + { + "reporting_period": { + "start": request.period.start.isoformat(), + "end": request.period.source_read_cutoff_at.isoformat(), + }, + "currency": "EUR", + "media_buy_deliveries": [ + { + "media_buy_id": "media-buy-redacted", + "status": "active", + "totals": {"impressions": 10, "spend": 1.25}, + "by_package": [], + } + ], + } + ) + source = _source(lambda _request: response) + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + result = await _run(source, request) + assert result.ok + + async def test_an_unrecognized_return_value_is_a_typed_failure() -> None: request = redacted_snapshot_request() source = _source(lambda _request: 42) diff --git a/tests/test_reporting_ledger.py b/tests/test_reporting_ledger.py index 1c69bd2cf..8a7767846 100644 --- a/tests/test_reporting_ledger.py +++ b/tests/test_reporting_ledger.py @@ -103,6 +103,7 @@ def _obligation( + configuration.automated_recovery_window, schedule=configuration.schedule, created_at=boundary.end, + currency="USD", ) defaults.update(overrides) return ReportingObligationRecord(**defaults) # type: ignore[arg-type] diff --git a/tests/type_checks/reporting_currency.py b/tests/type_checks/reporting_currency.py new file mode 100644 index 000000000..16bd65259 --- /dev/null +++ b/tests/type_checks/reporting_currency.py @@ -0,0 +1,91 @@ +"""Trusted account currency callbacks and explicit low-level obligation writes.""" + +from __future__ import annotations + +from dataclasses import replace +from datetime import datetime +from typing import Protocol + +from typing_extensions import assert_type + +from adcp.reporting.ledger import ( + CurrencyResolver, + FixedCurrencyResolver, + ProducerOfferings, + ReportingConfiguration, + ReportingConfigurationGenerationKey, + ReportingDefinitionBinding, + ReportingLedgerStore, + ReportingObligationRecord, + ReportingProducer, + require_single_currency, + validate_currency, +) +from adcp.reporting.source import ReportingSourceExecutor + + +class TrustedAccountHistory(Protocol): + async def currencies_for_scope( + self, + *, + generation: ReportingConfigurationGenerationKey, + at: datetime, + media_buy_ids: tuple[str, ...], + package_ids: tuple[str, ...], + ) -> tuple[str, ...]: ... + + +def configure( + store: ReportingLedgerStore, source: ReportingSourceExecutor, history: TrustedAccountHistory +) -> tuple[ReportingProducer, ReportingProducer, ReportingProducer]: + # A later ReliableReportingService account-context resolver can implement + # this lookup. No buyer context or source response participates. + async def historical_currency( + configuration: ReportingConfiguration, candidate: ReportingObligationRecord + ) -> str: + assert_type(candidate.currency, str | None) + values = await history.currencies_for_scope( + generation=configuration.generation_key, + at=candidate.scope_resolved_at, + media_buy_ids=candidate.media_buy_ids, + package_ids=candidate.package_ids, + ) + return assert_type(require_single_currency(values), str) + + def synchronous_currency( + configuration: ReportingConfiguration, candidate: ReportingObligationRecord + ) -> str: + return {"us-account": "USD", "eu-account": "EUR"}[candidate.account_id] + + async_resolver: CurrencyResolver = historical_currency + sync_resolver: CurrencyResolver = synchronous_currency + fixed_resolver: CurrencyResolver = FixedCurrencyResolver("EUR") + offerings = ProducerOfferings(snapshot_offering_id="SNAPSHOT") + return ( + ReportingProducer( + store=store, source=source, offerings=offerings, currency_resolver=async_resolver + ), + ReportingProducer( + store=store, source=source, offerings=offerings, currency_resolver=sync_resolver + ), + ReportingProducer( + store=store, source=source, offerings=offerings, currency_resolver=fixed_resolver + ), + ) + + +async def explicit_low_level_currency( + store: ReportingLedgerStore, + candidate: ReportingObligationRecord, + verified_definition: ReportingDefinitionBinding, +) -> None: + # Unit declarations are derived from the verified content-addressed + # definition by trusted seller code, then retained with the obligation. + definition = replace( + verified_definition, + monetary_metric_units=(("spend", "EUR"),), + monetary_control_total_units=(("spend", "EUR"),), + ) + frozen = replace(candidate, definition=definition, currency=validate_currency("EUR")) + assert_type(await store.commit_obligation(frozen), ReportingObligationRecord) + assert_type(frozen.currency, str | None) # None remains representable for legacy reads.