From 211c0db057e90dbf47b142165a930e95b84c64b3 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 02:55:03 +0000 Subject: [PATCH 1/4] feat(reporting): hold reporting-subscription owed sets in the reporting ledger A principal's reporting subscription is now executable by the ledger. The engine's frozen owed sets become ordinary obligations of one configuration generation per account, owning principal, subscription_id, version and resolved offering_id, aligned with the TypeScript SDK. - ReportingConfigurationGenerationKey gains an optional offering_id, set only for subscription generations. ledger_asdict omits it when unset, so every digest, work key and persisted scope of an account generation is byte-identical to before; a guard test forbids bare asdict in reporting. - SubscriptionObligationPlanner writes sparse, immutable obligations for periods ending by the engine's horizon, then advances the owed set's completeness horizon. Stores opt in through ReportingSubscriptionLedger; the in-memory store implements it, and the planner refuses other stores. - The producer never grid-fills an owed set and reads an owed obligation only once its period has ended. Owed sets are not leased; the service runs them through run_configuration. - get_reporting_status reports offering_id per period, keeps one wire generation per id, version and purpose, and leaves scope_closed false while the scope ends after the owed set's horizon. - sync_reporting_status resolves a statement through the caller's owed obligation for its period and report definition; a period the owed set does not hold fails with INVALID_STATUS_PERIOD. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/adcp_reporting_storage/ledger/pg.py | 8 +- .../reporting_ledger_reconciliation.sql | 3 +- .../outbox/_activity_pg.py | 6 +- .../src/adcp_reporting_storage/outbox/pg.py | 5 +- .../outbox/status_pg.py | 14 +- src/adcp/reporting/ledger/__init__.py | 4 + src/adcp/reporting/ledger/_delivery_state.py | 10 +- src/adcp/reporting/ledger/consumer_status.py | 124 +++++- src/adcp/reporting/ledger/delivery.py | 7 +- src/adcp/reporting/ledger/delivery_changes.py | 5 +- src/adcp/reporting/ledger/models.py | 84 +++- .../reporting/ledger/notification_models.py | 11 +- src/adcp/reporting/ledger/producer.py | 51 +++ .../ledger/reconciliation_projection.py | 5 +- src/adcp/reporting/ledger/status.py | 33 +- .../reporting/ledger/status_projection.py | 11 +- src/adcp/reporting/ledger/store.py | 78 +++- src/adcp/reporting/materializer/contracts.py | 7 +- .../reporting/materializer/verification.py | 9 +- src/adcp/reporting/materializer/work.py | 7 +- src/adcp/reporting/outbox/memory.py | 5 +- src/adcp/reporting/outbox/routing.py | 13 +- src/adcp/reporting/subscriptions/__init__.py | 10 + src/adcp/reporting/subscriptions/engine.py | 59 +++ src/adcp/reporting/subscriptions/ledger.py | 311 +++++++++++++++ tests/test_reporting_subscription_ledger.py | 367 ++++++++++++++++++ 26 files changed, 1172 insertions(+), 75 deletions(-) create mode 100644 src/adcp/reporting/subscriptions/ledger.py create mode 100644 tests/test_reporting_subscription_ledger.py diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py index 8ab1f7ee2..786454498 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py @@ -738,7 +738,7 @@ async def _dirty_issue( *, enqueue: bool = True, ) -> None: - from dataclasses import asdict + from adcp.reporting.ledger.models import ledger_asdict has_scope_storage = await self._issue_scope_storage_on(connection) row = ( @@ -814,7 +814,7 @@ async def _dirty_issue( "INSERT INTO reporting_issue_status_scopes (account_id, issue_id, scope)" " VALUES (%s,%s,%s::jsonb) ON CONFLICT (account_id, issue_id)" " DO UPDATE SET scope = EXCLUDED.scope", - (issue.account_id, issue.issue_id, _json(asdict(scope))), + (issue.account_id, issue.issue_id, _json(ledger_asdict(scope))), ) if enqueue and (before != issue or existing != scope): await self._dirty_status( @@ -1318,7 +1318,11 @@ async def find_obligation( delivery_config_version: int, period_start: datetime, period_end: datetime, + offering_id: str | None = None, ) -> ReportingObligationRecord | None: + if offering_id is not None: + # This store does not hold reporting-subscription owed sets yet. + return None key = ReportingConfigurationGenerationKey( account_id=account_id, consumer_id=consumer_id, diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql index d0876f42b..e0ab49762 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_ledger_reconciliation.sql @@ -511,7 +511,8 @@ BEGIN 'native_version_ref', 'object_ref', 'object_refs', 'observed_adjustment_sha256', 'observed_at', 'observed_canonical_content_digest', 'observed_control_totals', 'observed_manifest_sha256', 'observed_native_version_ref', 'observed_row_count', - 'physical_checksums', 'reader_compatibility', 'received_at', 'reconciliation_mode', + 'offering_id', 'physical_checksums', 'reader_compatibility', 'received_at', + 'reconciliation_mode', 'rejection_codes', 'reporting_adjustment_id', 'reporting_materialization_id', 'reporting_obligation_id', 'reporting_receipt_id', 'reporting_revision_id', 'resource', 'resource_ref', 'resource_retained_until', 'resource_retention_days', 'row_count', diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/_activity_pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/_activity_pg.py index 79b368102..f391684da 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/_activity_pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/_activity_pg.py @@ -5,11 +5,11 @@ import json from collections.abc import AsyncIterator, Callable from contextlib import asynccontextmanager -from dataclasses import asdict from datetime import datetime from secrets import token_hex from typing import TYPE_CHECKING, Any +from adcp.reporting.ledger.models import ledger_asdict from adcp.reporting.ledger.notification_models import ReportingNotificationError from adcp.reporting.outbox.activity import ( ActivityOutcome, @@ -150,7 +150,7 @@ async def _reserve_attempt_on( b.consumer_namespace, lease.token, reservation, - json.dumps(asdict(b)), + json.dumps(ledger_asdict(b)), at, request.url, request.payload_size_bytes, @@ -222,7 +222,7 @@ async def complete_attempt( attempt.request.url, attempt.request.payload_size_bytes, attempt.lease_token, - json.dumps(asdict(b)), + json.dumps(ledger_asdict(b)), ), ) return bool(cursor.rowcount) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/pg.py index be74e7831..46aaa519b 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/pg.py @@ -5,13 +5,14 @@ import hashlib import json from collections.abc import Callable -from dataclasses import asdict, astuple, fields +from dataclasses import astuple, fields from datetime import datetime, timedelta from secrets import token_hex from typing import TYPE_CHECKING, Any from adcp.reporting.canonical_json import canonical_json_utf8_v1 from adcp.reporting.evidence import aware_utc +from adcp.reporting.ledger.models import ledger_asdict from adcp.reporting.ledger.notification_models import ( DirtyReason, ReportingDomainEvent, @@ -117,7 +118,7 @@ async def mark_dirty( from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore await PgReportingLedgerStore._lock_account(connection, scope.account_id) - scope_hash = hashlib.sha256(canonical_json_utf8_v1(asdict(scope))).hexdigest() + scope_hash = hashlib.sha256(canonical_json_utf8_v1(ledger_asdict(scope))).hexdigest() evidence = after or before cause_id = evidence.record_id if evidence is not None else scope_hash consumer = scope.consumer_id or "" diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py index 63f4e91f3..cf78d9a91 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/status_pg.py @@ -9,7 +9,7 @@ import json from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from dataclasses import asdict, replace +from dataclasses import replace from datetime import timedelta from importlib.resources import files from secrets import token_hex @@ -19,7 +19,11 @@ from adcp._deferred_adapters import deferred_adapter from adcp.reporting.ledger.delivery_models import ReportingDeliveryRecord -from adcp.reporting.ledger.models import ReportingDeliveryEscalation, ReportingIssueLifecycle +from adcp.reporting.ledger.models import ( + ReportingDeliveryEscalation, + ReportingIssueLifecycle, + ledger_asdict, +) from adcp.reporting.ledger.notification_models import ( ReportingDomainEvent, ReportingNotificationError, @@ -78,7 +82,7 @@ def _replay_storage(snapshot: ReportingStatusSnapshot) -> str: [ { "lifecycle": _lifecycle_adapter().dump_python(i, mode="json"), - "scope": asdict(scopes[i.issue_id]) if i.issue_id in scopes else None, + "scope": ledger_asdict(scopes[i.issue_id]) if i.issue_id in scopes else None, } for i in snapshot.lifecycles ] @@ -293,7 +297,7 @@ async def _lock_scopes_on(self, connection: Any, snapshot: ReportingStatusSnapsh " snapshot, source_sequence, baseline, publishable, selector_writer_floor)" ' VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb,%s,\'{"health":"waiting"}\',0,FALSE,FALSE,2)' f" ON CONFLICT ({_KEY}) DO NOTHING", # nosec B608 - (*scope.checkpoint_key, json.dumps(asdict(scope)), "0" * 64), + (*scope.checkpoint_key, json.dumps(ledger_asdict(scope)), "0" * 64), ) await ( await connection.execute( @@ -312,7 +316,7 @@ async def _write_on(self, connection: Any, checkpoint: StatusCheckpoint) -> None " selector_writer_floor=%s" f" WHERE {_WHERE}", # nosec B608 ( - json.dumps(asdict(checkpoint.scope)), + json.dumps(ledger_asdict(checkpoint.scope)), checkpoint.fingerprint, checkpoint.generation, json.dumps(checkpoint.snapshot), diff --git a/src/adcp/reporting/ledger/__init__.py b/src/adcp/reporting/ledger/__init__.py index 52c46d7ef..7b5b3b774 100644 --- a/src/adcp/reporting/ledger/__init__.py +++ b/src/adcp/reporting/ledger/__init__.py @@ -154,6 +154,7 @@ ReportingProductionStatus, ReportingRevisionRecord, ReportingScheduleSpec, + ReportingSubscriptionGeneration, derive_period, iso_duration_to_timedelta, ) @@ -193,6 +194,7 @@ ReportingRowPage, ReportingRowStagingStore, ReportingSourcePublicationStore, + ReportingSubscriptionLedger, ReportingWorkerRetryPolicy, RestatementCheckpoint, RestatementCheckpointStore, @@ -311,6 +313,8 @@ "RetryScheduleEntry", "RetryScheduleStore", "ReportingScheduleSpec", + "ReportingSubscriptionGeneration", + "ReportingSubscriptionLedger", "ReportingStatusCaller", "ReportingStatusHandler", "ReportingStatusView", diff --git a/src/adcp/reporting/ledger/_delivery_state.py b/src/adcp/reporting/ledger/_delivery_state.py index 71f0cb5b8..b9d213881 100644 --- a/src/adcp/reporting/ledger/_delivery_state.py +++ b/src/adcp/reporting/ledger/_delivery_state.py @@ -4,7 +4,7 @@ import hashlib import json -from dataclasses import asdict, dataclass, replace +from dataclasses import dataclass, replace from datetime import datetime, timedelta from typing import Annotated, Any, NoReturn, TypeVar, cast @@ -33,6 +33,7 @@ ReportingConfiguration, ReportingObligationRecord, ReportingRevisionRecord, + ledger_asdict, ) from adcp.reporting.ledger.store import ( LedgerConflictError, @@ -72,7 +73,7 @@ def iso(value: datetime) -> str: def payload(record: ReportingDeliveryRecord) -> dict[str, Any]: # Dataclass fields only: never __dict__, a provider object, or a wire extras bag. - result: dict[str, Any] = json.loads(json.dumps(asdict(record), default=iso)) + result: dict[str, Any] = json.loads(json.dumps(ledger_asdict(record), default=iso)) return result @@ -108,7 +109,10 @@ def principal(record: ReportingDeliveryRecord) -> ReportingDeliveryPrincipal: def record_identity(record: ReportingDeliveryRecord) -> tuple[str, str]: if isinstance(record, ReportingDestinationBinding): generation = record.generation_key - return record.kind, hashlib.sha256(canonical_json_utf8_v1(asdict(generation))).hexdigest() + return ( + record.kind, + hashlib.sha256(canonical_json_utf8_v1(ledger_asdict(generation))).hexdigest(), + ) if isinstance(record, ReportingObligationDeliveryRecord): return record.kind, record.scope.reporting_obligation_id if isinstance(record, ReportingMaterializationCheck): diff --git a/src/adcp/reporting/ledger/consumer_status.py b/src/adcp/reporting/ledger/consumer_status.py index 0fe72a0a9..f7e016ec0 100644 --- a/src/adcp/reporting/ledger/consumer_status.py +++ b/src/adcp/reporting/ledger/consumer_status.py @@ -63,7 +63,7 @@ import hashlib from collections.abc import Callable, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timedelta, timezone from typing import Any, Literal, cast @@ -201,7 +201,7 @@ async def handle( if replay is not None: results.append({"result": "unchanged", "consumer_status": _to_wire(replay)}) continue - await self._validate_against_configuration(record) + record = await self._validate_against_configuration(record) from adcp.reporting.ledger.status_snapshot import ReportingStatusParticipant atomic = isinstance(self.store, ReportingStatusParticipant) @@ -224,6 +224,55 @@ async def handle( ) return {"status": "completed", "results": results} + async def _find_period_obligation( + self, record: ConsumerStatusRecord + ) -> ReportingObligationRecord | None: + """The obligation a statement names by generation and period. + + A reporting subscription keeps one generation per resolved offering, + so the same (id, version, period) can name several. Only the caller's + own owed-set generations are searched, and the report definition must + match too. + """ + found = await self.store.find_obligation( + account_id=record.account_id, + consumer_id=record.consumer_id, + delivery_config_id=record.delivery_config_id, + delivery_config_version=record.delivery_config_version, + period_start=record.period_start, + period_end=record.period_end, + ) + if found is not None: + return found + from adcp.reporting.ledger.delivery_models import ReportingDeliveryPrincipal + + configurations = await self.store.list_configurations( + caller=ReportingDeliveryPrincipal(record.account_id, record.consumer_id), + delivery_config_ids=[record.delivery_config_id], + ) + for configuration in sorted( + ( + item + for item in configurations + if item.owed_set + and item.delivery_config_version == record.delivery_config_version + and item.report_definition_id == record.report_definition_id + ), + key=lambda item: (item.subscription.offering_id if item.subscription else ""), + ): + candidate = await self.store.find_obligation( + account_id=record.account_id, + consumer_id=record.consumer_id, + delivery_config_id=record.delivery_config_id, + delivery_config_version=record.delivery_config_version, + period_start=record.period_start, + period_end=record.period_end, + offering_id=configuration.generation_key.offering_id, + ) + if candidate is not None: + return candidate + return None + async def _open_issue_on_first_observation(self, stored: ConsumerStatusRecord) -> None: """Start the escalation clock when the statement arrives, not when polled. @@ -247,14 +296,7 @@ async def _open_issue_on_first_observation(self, stored: ConsumerStatusRecord) - reporting_obligation_id=stored.reporting_obligation_id, ) if obligation is None: - obligation = await self.store.find_obligation( - account_id=stored.account_id, - consumer_id=stored.consumer_id, - delivery_config_id=stored.delivery_config_id, - delivery_config_version=stored.delivery_config_version, - period_start=stored.period_start, - period_end=stored.period_end, - ) + obligation = await self._find_period_obligation(stored) revisions = ( await self.store.list_revisions( account_id=stored.account_id, @@ -305,7 +347,9 @@ async def _open_issue_on_first_observation(self, stored: ConsumerStatusRecord) - observed_at=stored.recorded_at, ) - async def _validate_against_configuration(self, record: ConsumerStatusRecord) -> None: + async def _validate_against_configuration( + self, record: ConsumerStatusRecord + ) -> ConsumerStatusRecord: """Resolve every identifier the statement names, within caller + account. Deliberately does *not* require an obligation to exist *when none is @@ -330,6 +374,29 @@ async def _validate_against_configuration(self, record: ConsumerStatusRecord) -> (item for item in configurations if item.generation_key == record.generation_key), None, ) + if generation is None and any(item.owed_set for item in configurations): + # A reporting subscription's generations differ by offering. A + # statement names a period of the caller's own owed set, never a + # bare grid slot: the owed obligation decides the generation. + owed = await self._find_period_obligation(record) + generation = next( + ( + item + for item in configurations + if owed is not None and item.generation_key == owed.generation_key + ), + None, + ) + if generation is None and any( + _same_generation(item.generation_key, record.generation_key) + for item in configurations + ): + raise LedgerConflictError( + "INVALID_STATUS_PERIOD", + "the statement names no period of this subscription's owed set", + ) + if generation is not None: + record = replace(record, offering_id=generation.generation_key.offering_id) if generation is None: raise LedgerConflictError( "UNKNOWN_CONFIGURATION_GENERATION", @@ -350,6 +417,7 @@ async def _validate_against_configuration(self, record: ConsumerStatusRecord) -> ) await self._resolve_named_records(record) validate_consumer_status_timing(record, generation, as_of=self._now()) + return record async def _resolve_named_records(self, record: ConsumerStatusRecord) -> None: """Resolve the optional obligation and revision ids the caller supplied. @@ -371,7 +439,7 @@ async def _resolve_named_records(self, record: ConsumerStatusRecord) -> None: "resolve for this caller and account", ) if ( - obligation.generation_key != record.generation_key + not _same_generation(obligation.generation_key, record.generation_key) or obligation.report_definition_id != record.report_definition_id or _utc(obligation.period.start) != _utc(record.period_start) or _utc(obligation.period.end) != _utc(record.period_end) @@ -387,14 +455,7 @@ async def _resolve_named_records(self, record: ConsumerStatusRecord) -> None: ) if obligation is None: - obligation = await self.store.find_obligation( - account_id=record.account_id, - consumer_id=record.consumer_id, - delivery_config_id=record.delivery_config_id, - delivery_config_version=record.delivery_config_version, - period_start=record.period_start, - period_end=record.period_end, - ) + obligation = await self._find_period_obligation(record) required = None if obligation is not None: revisions = await self.store.list_revisions( @@ -490,6 +551,25 @@ def _to_record( ) +def _same_generation( + key: ReportingConfigurationGenerationKey, record_key: ReportingConfigurationGenerationKey +) -> bool: + """Same generation, where a statement's key carries no offering. + + The wire identifies a generation by (account, caller, delivery_config_id, + delivery_config_version). A reporting subscription's generations also + differ by offering, which a statement resolves through its obligation. + """ + if record_key.offering_id is not None: + return key == record_key + return ( + key.account_id == record_key.account_id + and key.consumer_id == record_key.consumer_id + and key.delivery_config_id == record_key.delivery_config_id + and key.delivery_config_version == record_key.delivery_config_version + ) + + def validate_consumer_status_timing( record: ConsumerStatusRecord, generation: ReportingConfiguration, *, as_of: datetime ) -> None: @@ -513,6 +593,10 @@ def validate_consumer_status_timing( raise LedgerConflictError( "INVALID_STATUS_TIME", "status observation is outside its valid time range" ) + if generation.owed_set: + # An owed set has no grid: the statement was matched to an existing + # owed obligation's exact period, which carries its own calendar. + return zone_name = generation.schedule.timezone_name(generation.account_timezone) if record.period_source_timezone != zone_name: raise LedgerConflictError( diff --git a/src/adcp/reporting/ledger/delivery.py b/src/adcp/reporting/ledger/delivery.py index 35ce5e321..92800889d 100644 --- a/src/adcp/reporting/ledger/delivery.py +++ b/src/adcp/reporting/ledger/delivery.py @@ -7,7 +7,7 @@ from __future__ import annotations -from dataclasses import asdict, dataclass +from dataclasses import dataclass from datetime import datetime from typing import Any, NoReturn, Protocol, cast, runtime_checkable @@ -61,6 +61,7 @@ ReportingConfigurationGenerationKey, ReportingObligationRecord, ReportingRevisionRecord, + ledger_asdict, ) from adcp.reporting.ledger.store import InMemoryReportingLedgerStore, LedgerConflictError @@ -192,7 +193,7 @@ def to_wire(self) -> dict[str, Any]: assert resource is not None and verification is not None descriptor = { key: value - for key, value in asdict(resource).items() + for key, value in ledger_asdict(resource).items() if value is not None and key != "object_refs" } descriptor["expires_at"] = iso(resource.expires_at) @@ -210,7 +211,7 @@ def to_wire(self) -> dict[str, Any]: evidence["canonical_content_digest"] = verification.canonical_content_digest.to_wire() if verification.physical_checksums: evidence["physical_checksums"] = [ - asdict(item) for item in verification.physical_checksums + ledger_asdict(item) for item in verification.physical_checksums ] if verification.native_version_ref is not None: evidence["native_commit_evidence"] = { diff --git a/src/adcp/reporting/ledger/delivery_changes.py b/src/adcp/reporting/ledger/delivery_changes.py index e230ead9d..f8b11979c 100644 --- a/src/adcp/reporting/ledger/delivery_changes.py +++ b/src/adcp/reporting/ledger/delivery_changes.py @@ -3,7 +3,7 @@ from __future__ import annotations import hashlib -from dataclasses import asdict, dataclass +from dataclasses import dataclass from datetime import datetime from typing import Any, NewType, Protocol, get_args, runtime_checkable @@ -16,6 +16,7 @@ ReportingDestinationBinding, ReportingReconciliationRecordKind, ) +from adcp.reporting.ledger.models import ledger_asdict from adcp.reporting.ledger.store import LedgerConflictError, decode_cursor, encode_cursor _FEED = "reporting-reconciliation-v1" @@ -43,7 +44,7 @@ def __post_init__(self) -> None: @property def fingerprint(self) -> str: - return hashlib.sha256(canonical_json_utf8_v1(asdict(self))).hexdigest() + return hashlib.sha256(canonical_json_utf8_v1(ledger_asdict(self))).hexdigest() def matches(self, record: ReportingDeliveryRecord) -> bool: return (not self.record_kinds or record.kind in self.record_kinds) and ( diff --git a/src/adcp/reporting/ledger/models.py b/src/adcp/reporting/ledger/models.py index eebffbbfb..7826a9041 100644 --- a/src/adcp/reporting/ledger/models.py +++ b/src/adcp/reporting/ledger/models.py @@ -23,7 +23,7 @@ from __future__ import annotations -from dataclasses import dataclass, field +from dataclasses import asdict, dataclass, field from datetime import datetime, timedelta, timezone from typing import Any, Literal, Protocol from zoneinfo import ZoneInfo @@ -337,6 +337,33 @@ def account_id(self) -> str: ... def consumer_id(self) -> str: ... +_GENERATION_KEY_FIELDS = frozenset( + {"account_id", "consumer_id", "delivery_config_id", "delivery_config_version", "offering_id"} +) + + +def _ledger_dict(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + if ( + len(pairs) == len(_GENERATION_KEY_FIELDS) + and {name for name, _ in pairs} == _GENERATION_KEY_FIELDS + and dict(pairs)["offering_id"] is None + ): + return {name: value for name, value in pairs if name != "offering_id"} + return dict(pairs) + + +def ledger_asdict(value: Any) -> dict[str, Any]: + """``dataclasses.asdict`` that keeps account-level generation keys byte-stable. + + A generation key's ``offering_id`` is set only for a reporting + subscription's generation. Left unset it is omitted, so every digest, + work key and persisted scope derived from an account-level generation is + exactly what it was before subscriptions existed. Use this, never bare + ``asdict``, for any value that may contain a generation key. + """ + return asdict(value, dict_factory=_ledger_dict) + + @dataclass(frozen=True) class ReportingConfigurationGenerationKey: """The caller/account-qualified identity of one accepted configuration generation. @@ -350,12 +377,43 @@ class ReportingConfigurationGenerationKey: consumer_id: str delivery_config_id: str delivery_config_version: int + #: Set only for a principal-level reporting subscription's generation: one + #: generation per resolved offering of a subscription version, since + #: ``match: all`` and per-product resolution give one version several. + offering_id: str | None = None def __post_init__(self) -> None: principal_reference(self.account_id) consumer_reference(self.consumer_id) +@dataclass(frozen=True) +class ReportingSubscriptionGeneration: + """Marks a configuration generation as a reporting subscription's owed set. + + ``delivery_config_id`` is the ``subscription_id`` and + ``delivery_config_version`` the ``subscription_version``; the consumer is + the owning principal. The generation holds exactly the obligations the + subscription engine froze (an owed set). The ledger never fills a period + grid for it: its obligations are sparse by construction, because a + subscription owes only the periods its buys' flights and exceptions leave. + """ + + subscription_id: str + offering_id: str + + def __post_init__(self) -> None: + if not self.subscription_id or not self.offering_id: + raise ValueError("subscription generations need subscription_id and offering_id") + + def to_storage(self) -> dict[str, str]: + return { + "subscription_id": self.subscription_id, + "offering_id": self.offering_id, + "ownership": "owed_set", + } + + @dataclass(frozen=True) class ReportingConfiguration: """One accepted reporting configuration generation. @@ -390,10 +448,25 @@ class ReportingConfiguration: authoritative_party: Literal["seller", "consumer"] = "seller" # Maintenance imports are read-only; recovery requires a new generation. quarantined: bool = False + #: Set for a reporting subscription's generation (an owed set); see + #: :class:`ReportingSubscriptionGeneration`. + subscription: ReportingSubscriptionGeneration | None = None def __post_init__(self) -> None: principal_reference(self.account_id) consumer_reference(self.consumer_id) + if ( + self.subscription is not None + and self.subscription.subscription_id != self.delivery_config_id + ): + raise ValueError( + "a subscription generation's delivery_config_id is its subscription_id" + ) + + @property + def owed_set(self) -> bool: + """Whether the generation holds a subscription's sparse owed set.""" + return self.subscription is not None @property def generation_key(self) -> ReportingConfigurationGenerationKey: @@ -402,6 +475,7 @@ def generation_key(self) -> ReportingConfigurationGenerationKey: consumer_id=self.consumer_id, delivery_config_id=self.delivery_config_id, delivery_config_version=self.delivery_config_version, + offering_id=None if self.subscription is None else self.subscription.offering_id, ) @@ -436,6 +510,9 @@ class ReportingObligationRecord: # 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 + #: The resolved offering of a reporting subscription's obligation; ``None`` + #: for account-level configurations. Authoritative for its timing. + offering_id: str | None = None @property def generation_key(self) -> ReportingConfigurationGenerationKey: @@ -444,6 +521,7 @@ def generation_key(self) -> ReportingConfigurationGenerationKey: consumer_id=self.consumer_id, delivery_config_id=self.delivery_config_id, delivery_config_version=self.delivery_config_version, + offering_id=self.offering_id, ) def __post_init__(self) -> None: @@ -615,6 +693,9 @@ class ConsumerStatusRecord: seller_ledger_snapshot_id: str | None = None seller_ledger_as_of: datetime | None = None superseded: bool = False + #: A reporting subscription's resolved offering, set when ingest matched + #: the statement to an owed obligation. Not part of the chain key. + offering_id: str | None = None def __post_init__(self) -> None: principal_reference(self.account_id) @@ -627,6 +708,7 @@ def generation_key(self) -> ReportingConfigurationGenerationKey: consumer_id=self.consumer_id, delivery_config_id=self.delivery_config_id, delivery_config_version=self.delivery_config_version, + offering_id=self.offering_id, ) @property diff --git a/src/adcp/reporting/ledger/notification_models.py b/src/adcp/reporting/ledger/notification_models.py index bdeb5de7a..4f94e16cb 100644 --- a/src/adcp/reporting/ledger/notification_models.py +++ b/src/adcp/reporting/ledger/notification_models.py @@ -4,7 +4,7 @@ import hashlib import json -from dataclasses import asdict, dataclass, field +from dataclasses import dataclass, field from datetime import datetime from typing import Any, Literal, TypeAlias from uuid import uuid4 @@ -26,6 +26,7 @@ ReportingHealth, ReportingIssueLifecycle, ReportingObligationRecord, + ledger_asdict, ) from adcp.validation.schema_loader import get_named_validator @@ -333,7 +334,7 @@ def for_obligation( return decode_status_scope( { "account_id": obligation.account_id, - "generation_key": asdict(obligation.generation_key), + "generation_key": ledger_asdict(obligation.generation_key), "reporting_obligation_id": obligation.reporting_obligation_id, "consumer_id": obligation.consumer_id if consumer_id is None else consumer_id, "feed_purpose": obligation.feed_purpose, @@ -456,7 +457,7 @@ def iso(value: datetime) -> str: def event_storage(event: ReportingDomainEvent) -> dict[str, Any]: - return dict(json.loads(json.dumps(asdict(event), default=iso))) + return dict(json.loads(json.dumps(ledger_asdict(event), default=iso))) @deferred_adapter @@ -475,7 +476,7 @@ def _dirty_adapter() -> TypeAdapter[ReportingStatusDirty]: def dirty_storage(record: ReportingStatusDirty) -> dict[str, Any]: - return dict(json.loads(json.dumps(asdict(record), default=iso))) + return dict(json.loads(json.dumps(ledger_asdict(record), default=iso))) def decode_dirty(value: object) -> ReportingStatusDirty: @@ -504,7 +505,7 @@ def decode_event(value: object) -> ReportingDomainEvent: def decode_status_scope(value: object) -> ReportingStatusScope: try: scope = _scope_adapter().validate_python(value) - if asdict(scope) == value: + if ledger_asdict(scope) == value: return scope except (ValidationError, ValueError, TypeError): # Leave the parsing exception context before emitting the closed classification. diff --git a/src/adcp/reporting/ledger/producer.py b/src/adcp/reporting/ledger/producer.py index 449acde58..210614db9 100644 --- a/src/adcp/reporting/ledger/producer.py +++ b/src/adcp/reporting/ledger/producer.py @@ -594,6 +594,10 @@ async def _close_elapsed_periods( ) -> list[ReportingObligationRecord]: from adcp.reporting.ledger.producer_progress import ReportingProducerProgress + if configuration.owed_set: + # A subscription's owed set is written by its planner, never filled + # from a period grid. + return [] progress = self._store if isinstance(self._store, ReportingProducerProgress) else None after = None if progress is None else await progress.producer_closed_through(configuration) committed: list[ReportingObligationRecord] = [] @@ -712,6 +716,9 @@ async def _acquire_pending( ) -> None: from adcp.reporting.ledger.producer_progress import ReportingProducerProgress + if configuration.owed_set: + await self._acquire_owed_set(configuration, turn, now=now) + return if isinstance(self._store, ReportingProducerProgress): await self._acquire_progress(self._store, configuration, turn, now=now) return @@ -754,6 +761,50 @@ async def _acquire_pending( turn.slices_failed.append(obligation.reporting_obligation_id) self._note_escalation(obligation, turn, now=now) + async def _acquire_owed_set( + self, configuration: ReportingConfiguration, turn: WorkerTurn, *, now: datetime + ) -> None: + """Acquire a subscription's frozen obligations that have elapsed. + + The owed set is exactly the stored obligations; there is no grid to + walk. Satisfied obligations are left alone by ``acquire_obligation``, + and those with an official revision are skipped before dispatch. + """ + from adcp.reporting.ledger.store import ReportingSubscriptionLedger + + if not isinstance(self._store, ReportingSubscriptionLedger): + raise LedgerConflictError( + "UNSUPPORTED_STORE", "this store cannot hold reporting-subscription owed sets" + ) + worked = 0 + for obligation in await self._store.list_generation_obligations( + configuration.generation_key + ): + if worked >= self._max_periods_per_turn or _utc(obligation.period.end) > _utc(now): + break + revisions = await self._store.list_revisions( + account_id=obligation.account_id, + reporting_obligation_id=obligation.reporting_obligation_id, + ) + if any(item.finality == "official" for item in revisions): + continue + if not await self._automatic_read_ready(configuration, obligation, now=now): + continue + worked += 1 + try: + policy = self._settling_policy(configuration, obligation) + if policy is None: + await self.acquire_obligation(configuration, obligation, turn=turn, now=now) + else: + await self._acquire_with_settling_policy( + configuration, obligation, policy=policy, turn=turn, now=now + ) + except (ReportingCurrencyError, LedgerConflictError) as error: + if isinstance(error, LedgerConflictError) and error.code != "OBSERVATION_CONFLICT": + raise + turn.slices_failed.append(obligation.reporting_obligation_id) + self._note_escalation(obligation, turn, now=now) + async def _acquire_progress( self, progress: ReportingProducerProgress, diff --git a/src/adcp/reporting/ledger/reconciliation_projection.py b/src/adcp/reporting/ledger/reconciliation_projection.py index 6a670c6df..4aefa6681 100644 --- a/src/adcp/reporting/ledger/reconciliation_projection.py +++ b/src/adcp/reporting/ledger/reconciliation_projection.py @@ -9,7 +9,7 @@ import hashlib import json -from dataclasses import asdict, dataclass +from dataclasses import dataclass from datetime import datetime from typing import Any, Literal @@ -42,6 +42,7 @@ ReportingIssue, ReportingObligationRecord, ReportingRevisionRecord, + ledger_asdict, ) from adcp.reporting.ledger.store import LedgerConflictError from adcp.reporting.revision_selection import select_reporting_revision @@ -103,7 +104,7 @@ def project_reconciliation( [ a.reporting_adjustment_id, hashlib.sha256( - json.dumps(asdict(a), default=iso, sort_keys=True).encode() + json.dumps(ledger_asdict(a), default=iso, sort_keys=True).encode() ).hexdigest(), ] for a in owned_adjustments diff --git a/src/adcp/reporting/ledger/status.py b/src/adcp/reporting/ledger/status.py index bb3b564c7..ba4a0103d 100644 --- a/src/adcp/reporting/ledger/status.py +++ b/src/adcp/reporting/ledger/status.py @@ -30,7 +30,7 @@ from __future__ import annotations import hashlib -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from dataclasses import dataclass, replace from datetime import datetime, timezone from typing import Any, Literal @@ -47,6 +47,7 @@ ConsumerStatusRecord, ReportingAdjustmentRecord, ReportingConfiguration, + ReportingConfigurationGenerationKey, ReportingDeliveryEscalation, ReportingHealth, ReportingIssue, @@ -388,6 +389,7 @@ def render_snapshot( ledger_as_of=snapshot.as_of, request=request, obligations=tuple(p.obligation for p in result.obligations), + owed_through=dict(getattr(snapshot, "owed_through", ()) or ()), ) obligations = tuple(p.obligation for p in result.obligations) if view == "summary": @@ -639,6 +641,7 @@ def _scope_to_wire( ledger_as_of: datetime, request: dict[str, Any], obligations: Sequence[ReportingObligationRecord] = (), + owed_through: Mapping[ReportingConfigurationGenerationKey, datetime] | None = None, ) -> dict[str, Any]: """The denominator this read's health was computed over. @@ -657,13 +660,31 @@ def _scope_to_wire( retained_from = status_retained_from(configurations, ledger_as_of) horizon_start = _parse(requested.get("start")) or retained_from horizon_end = _parse(requested.get("end")) or _utc(ledger_as_of) - generations = {item.generation_key: item for item in configurations} + # A reporting subscription keeps one generation per resolved offering; + # they share (id, version, feed_purpose), which the wire lists once. + generations = { + ( + item.account_id, + item.delivery_config_id, + item.delivery_config_version, + item.feed_purpose, + ): item + for item in configurations + } + owed = owed_through or {} return { "period_start": _iso(horizon_start), "period_end": _iso(horizon_end), - # No further obligation can enter a horizon that has already elapsed. + # No further obligation can enter a horizon that has already elapsed, + # and a subscription's owed set must also be frozen through it: until + # then a period may still be added, which is not missing history. "scope_closed": horizon_end <= _utc(ledger_as_of) - and all(o.period.end <= ledger_as_of for o in obligations), + and all(o.period.end <= ledger_as_of for o in obligations) + and all( + owed.get(item.generation_key) is not None and owed[item.generation_key] >= horizon_end + for item in configurations + if item.owed_set + ), # True when the caller named no media buys, so the scope is every buy # it can reach rather than an enumerated subset. "all_accessible_media_buys": not request.get("media_buy_ids"), @@ -825,6 +846,10 @@ def _obligation_to_wire( "adjustment_count": 0, "issues": [issue.to_wire() for issue in issues], } + if obligation.offering_id is not None: + # 3.3 reporting-obligation.json: the subscription's resolved offering, + # authoritative for the obligation's timing. + payload["offering_id"] = obligation.offering_id if statuses: current = next((item for item in statuses if not item.superseded), None) payload["consumer_status_count"] = len(statuses) diff --git a/src/adcp/reporting/ledger/status_projection.py b/src/adcp/reporting/ledger/status_projection.py index 2e12ccfb3..181393827 100644 --- a/src/adcp/reporting/ledger/status_projection.py +++ b/src/adcp/reporting/ledger/status_projection.py @@ -9,7 +9,7 @@ import hashlib from collections.abc import Sequence -from dataclasses import asdict, dataclass, replace +from dataclasses import dataclass, replace from datetime import datetime, timedelta from typing import TYPE_CHECKING, Any, Literal, cast @@ -37,6 +37,7 @@ ConsumerStatusRecord, ReportingAdjustmentRecord, ReportingConfiguration, + ReportingConfigurationGenerationKey, ReportingDeliveryEscalation, ReportingHealth, ReportingIssue, @@ -44,6 +45,7 @@ ReportingObligationRecord, ReportingRevisionRecord, iso_duration_to_timedelta, + ledger_asdict, ) from adcp.reporting.ledger.notification_models import ( ReportingNotificationError, @@ -70,6 +72,9 @@ class ReportingStatusSnapshot: adjustments: tuple[ReportingAdjustmentRecord, ...] = () # Sequence, record kind, ID, authenticated generation/status owner. changes: tuple[tuple[int, str, str, str], ...] = () + #: Reporting-subscription owed sets: the horizon each owed-set generation's + #: obligations are frozen through. A scope ending after it is not closed. + owed_through: tuple[tuple[ReportingConfigurationGenerationKey, datetime], ...] = () @property def max_sequence(self) -> int: @@ -562,7 +567,7 @@ def _project(value: StatusProjectionInput) -> tuple[StatusProjectionResult, set[ ReportingIssue( issue_id=issue_id_for( "retained-scope-history-unavailable-v2", - canonical_json_utf8_v1(asdict(scope)).hex(), + canonical_json_utf8_v1(ledger_asdict(scope)).hex(), ), code="HISTORY_UNAVAILABLE", severity="action_required", @@ -852,7 +857,7 @@ def _project(value: StatusProjectionInput) -> tuple[StatusProjectionResult, set[ health = "delayed" canonical = { "version": 1, - "scope": asdict(scope), + "scope": ledger_asdict(scope), "health": health, "issues": [i.to_wire() for i in full_issues], "filters": { diff --git a/src/adcp/reporting/ledger/store.py b/src/adcp/reporting/ledger/store.py index b91762212..db43c8d37 100644 --- a/src/adcp/reporting/ledger/store.py +++ b/src/adcp/reporting/ledger/store.py @@ -408,6 +408,7 @@ async def find_obligation( delivery_config_version: int, period_start: datetime, period_end: datetime, + offering_id: str | None = None, ) -> ReportingObligationRecord | None: ... # -- revisions ------------------------------------------------------- @@ -773,6 +774,29 @@ def check_issue_state_transition(current: str, requested: str) -> None: ) +@runtime_checkable +class ReportingSubscriptionLedger(Protocol): + """Optional store capability for reporting-subscription owed sets. + + A subscription generation holds a sparse owed set: exactly the obligations + the subscription engine froze, never a period grid. The planner and the + producer use these operations; a store without them cannot hold + subscription obligations and is refused rather than half-supported. + """ + + async def list_generation_obligations( + self, generation_key: ReportingConfigurationGenerationKey + ) -> tuple[ReportingObligationRecord, ...]: ... + + async def record_owed_set_progress( + self, generation_key: ReportingConfigurationGenerationKey, through: datetime + ) -> datetime: ... + + async def owed_set_through( + self, generation_key: ReportingConfigurationGenerationKey + ) -> datetime | None: ... + + class InMemoryReportingLedgerStore: """Process-local reference store. Correct, ordered, and not durable. @@ -810,6 +834,9 @@ def __init__( # letting it win every turn. self._lease_turns: dict[ReportingConfigurationGenerationKey, int] = {} self._lease_turn = 0 + # Reporting-subscription owed sets: the horizon each generation's + # obligations are frozen through. + self._owed_through: dict[ReportingConfigurationGenerationKey, datetime] = {} # Live occurrence per (account, issue_key), plus the retired generation # high-water mark so a recurrence never reuses an id. self._issues: dict[tuple[str, str], ReportingIssueLifecycle] = {} @@ -1092,6 +1119,7 @@ async def find_obligation( delivery_config_version: int, period_start: datetime, period_end: datetime, + offering_id: str | None = None, ) -> ReportingObligationRecord | None: key = ( ReportingConfigurationGenerationKey( @@ -1099,6 +1127,7 @@ async def find_obligation( consumer_id=consumer_id, delivery_config_id=delivery_config_id, delivery_config_version=delivery_config_version, + offering_id=offering_id, ), _utc(period_start).isoformat(), _utc(period_end).isoformat(), @@ -1110,6 +1139,39 @@ async def find_obligation( else None ) + # -- reporting-subscription owed sets --------------------------------- + + async def list_generation_obligations( + self, generation_key: ReportingConfigurationGenerationKey + ) -> tuple[ReportingObligationRecord, ...]: + """Every obligation of one generation, oldest period first.""" + found = [ + self._obligations[identifier] + for (key, _start, _end), identifier in self._obligation_by_period.items() + if key == generation_key + ] + return tuple(sorted(found, key=lambda item: _utc(item.period.start))) + + async def record_owed_set_progress( + self, generation_key: ReportingConfigurationGenerationKey, through: datetime + ) -> datetime: + """Advance the horizon an owed set is frozen through; it never regresses.""" + async with self._mutation(): + configuration = self._configurations.get(generation_key) + if configuration is None or not configuration.owed_set: + raise LedgerConflictError( + "UNKNOWN_CONFIGURATION_GENERATION", "owed-set generation is unavailable" + ) + current = self._owed_through.get(generation_key) + value = _utc(through) if current is None else max(current, _utc(through)) + self._owed_through[generation_key] = value + return value + + async def owed_set_through( + self, generation_key: ReportingConfigurationGenerationKey + ) -> datetime | None: + return self._owed_through.get(generation_key) + # -- revisions ------------------------------------------------------- async def commit_revision( @@ -1624,10 +1686,16 @@ async def read_status_snapshot(self, *, caller: ReportingCaller) -> ReportingSta from adcp.reporting.ledger.delivery_models import ReportingDeliveryPrincipal from adcp.reporting.materializer.capture import private_snapshot - return private_snapshot( + snapshot = private_snapshot( settle_memory_snapshot(self, caller.account_id), ReportingDeliveryPrincipal(caller.account_id, caller.consumer_id), ) + owed = tuple( + (item.generation_key, self._owed_through[item.generation_key]) + for item in snapshot.configurations + if item.owed_set and item.generation_key in self._owed_through + ) + return replace(snapshot, owed_through=owed) if owed else snapshot def _current_status_leaf(self, chain_key: tuple[Any, ...]) -> ConsumerStatusRecord | None: for item in self._statuses.values(): @@ -1999,7 +2067,8 @@ def _obligation_for( # -- leasing --------------------------------------------------------- def _configuration_lease_eligible(self, configuration: ReportingConfiguration) -> bool: - return not configuration.quarantined + # An owed set is never grid-filled; its owner runs it explicitly. + return not configuration.quarantined and not configuration.owed_set async def lease_period_close( self, *, worker_id: str, now: datetime, lease_seconds: float @@ -2126,6 +2195,11 @@ def _config_payload(configuration: ReportingConfiguration) -> dict[str, Any]: "authoritative_party": configuration.authoritative_party, "media_buy_ids": sorted(configuration.media_buy_ids), "definition": configuration.definition.to_storage() if configuration.definition else None, + **( + {"subscription": configuration.subscription.to_storage()} + if configuration.subscription is not None + else {} + ), "schedule": { "period_duration": schedule.period_duration, "delivery_sla": schedule.delivery_sla, diff --git a/src/adcp/reporting/materializer/contracts.py b/src/adcp/reporting/materializer/contracts.py index be70ca58e..326fa4ae9 100644 --- a/src/adcp/reporting/materializer/contracts.py +++ b/src/adcp/reporting/materializer/contracts.py @@ -14,7 +14,7 @@ import re from abc import ABC, abstractmethod from collections.abc import AsyncIterator, Awaitable, Callable -from dataclasses import asdict, dataclass, field, replace +from dataclasses import dataclass, field, replace from datetime import datetime, timezone from types import TracebackType from typing import Any, Literal, Protocol, TypeVar @@ -47,6 +47,7 @@ ReportingDefinitionBinding, ReportingObligationRecord, ReportingRevisionRecord, + ledger_asdict, ) ReportingIOPhase = Literal["write", "readback"] @@ -203,7 +204,7 @@ def __post_init__(self) -> None: def binding_fingerprint(binding: ReportingDestinationBinding) -> str: - value = asdict(binding) + value = ledger_asdict(binding) value["created_at"] = binding.created_at.isoformat() return hashlib.sha256(canonical_json_utf8_v1(value)).hexdigest() @@ -276,7 +277,7 @@ def from_binding( @property def external_id(self) -> str: """Stable across pending retries; isolated across tenants and revision attempts.""" - return "rwm_" + hashlib.sha256(canonical_json_utf8_v1(asdict(self))).hexdigest() + return "rwm_" + hashlib.sha256(canonical_json_utf8_v1(ledger_asdict(self))).hexdigest() @dataclass(frozen=True, slots=True) diff --git a/src/adcp/reporting/materializer/verification.py b/src/adcp/reporting/materializer/verification.py index a63f6a842..820bd19d3 100644 --- a/src/adcp/reporting/materializer/verification.py +++ b/src/adcp/reporting/materializer/verification.py @@ -16,7 +16,7 @@ import secrets from collections.abc import AsyncIterator, Sequence from contextlib import asynccontextmanager -from dataclasses import asdict, dataclass, field, replace +from dataclasses import dataclass, field, replace from datetime import datetime, timedelta, timezone from decimal import Decimal, localcontext from typing import Any, Protocol, cast @@ -38,6 +38,7 @@ ReportingDefinitionBinding, ReportingObligationRecord, ReportingRevisionRecord, + ledger_asdict, ) from adcp.reporting.ledger.producer import revision_content_sha256 from adcp.reporting.ledger.store import LedgerConflictError, ReportingRowPage, revision_row_offset @@ -107,9 +108,9 @@ def _verification_seal(value: ReportingVerifiedDestination) -> bytes: content = json.loads( json.dumps( [ - asdict(value.request), - asdict(value.resource), - asdict(value.verification), + ledger_asdict(value.request), + ledger_asdict(value.resource), + ledger_asdict(value.verification), getattr(value, "_materializer_fence", None), ], default=lambda at: at.isoformat(), diff --git a/src/adcp/reporting/materializer/work.py b/src/adcp/reporting/materializer/work.py index 9bed997e6..eeb983504 100644 --- a/src/adcp/reporting/materializer/work.py +++ b/src/adcp/reporting/materializer/work.py @@ -3,7 +3,7 @@ from __future__ import annotations import hashlib -from dataclasses import asdict, dataclass, field +from dataclasses import dataclass, field from datetime import datetime from typing import Literal, Protocol, get_args, runtime_checkable from uuid import UUID @@ -24,6 +24,7 @@ ReportingConfiguration, ReportingObligationRecord, ReportingRevisionRecord, + ledger_asdict, ) from adcp.reporting.materializer._errors import ReportingMaterializerUsageError from adcp.reporting.materializer.contracts import ( @@ -59,11 +60,11 @@ def scope_id(scope: ReportingDeliveryScope) -> str: - return hashlib.sha256(canonical_json_utf8_v1(asdict(scope))).hexdigest() + return hashlib.sha256(canonical_json_utf8_v1(ledger_asdict(scope))).hexdigest() def verification_key_id(key: ReportingVerificationKey) -> str: - return hashlib.sha256(canonical_json_utf8_v1(asdict(key))).hexdigest() + return hashlib.sha256(canonical_json_utf8_v1(ledger_asdict(key))).hexdigest() def key_for( diff --git a/src/adcp/reporting/outbox/memory.py b/src/adcp/reporting/outbox/memory.py index 0d49637cf..0e2dcf6a3 100644 --- a/src/adcp/reporting/outbox/memory.py +++ b/src/adcp/reporting/outbox/memory.py @@ -3,13 +3,14 @@ from __future__ import annotations from copy import deepcopy -from dataclasses import asdict, dataclass, field, replace +from dataclasses import dataclass, field, replace from datetime import datetime, timedelta from secrets import token_hex from typing import TYPE_CHECKING from adcp.reporting.canonical_json import canonical_json_utf8_v1 from adcp.reporting.evidence import aware_utc +from adcp.reporting.ledger.models import ledger_asdict from adcp.reporting.ledger.notification_models import ( DirtyReason, ReportingDomainEvent, @@ -124,7 +125,7 @@ def mark_dirty( cause_id = ( evidence.record_id if evidence is not None - else hashlib.sha256(canonical_json_utf8_v1(asdict(scope))).hexdigest() + else hashlib.sha256(canonical_json_utf8_v1(ledger_asdict(scope))).hexdigest() ) generation = ( sum( diff --git a/src/adcp/reporting/outbox/routing.py b/src/adcp/reporting/outbox/routing.py index d5cfaffaa..26eb91f4d 100644 --- a/src/adcp/reporting/outbox/routing.py +++ b/src/adcp/reporting/outbox/routing.py @@ -7,7 +7,7 @@ import json import secrets from collections.abc import Mapping -from dataclasses import asdict, dataclass, field +from dataclasses import dataclass, field from typing import Any, Literal, Protocol from uuid import uuid4 @@ -17,6 +17,7 @@ from adcp.reporting.canonical_json import canonical_json_utf8_v1 from adcp.reporting.evidence import principal_reference, reporting_identifier +from adcp.reporting.ledger.models import ledger_asdict from adcp.reporting.ledger.notification_models import ( ReportingDomainEvent, ReportingNotificationError, @@ -132,7 +133,7 @@ def auth_mode(self) -> str: @property def fingerprint(self) -> str: - return hashlib.sha256(canonical_json_utf8_v1(asdict(self))).hexdigest() + return hashlib.sha256(canonical_json_utf8_v1(ledger_asdict(self))).hexdigest() def matches(self, event: ReportingDomainEvent) -> bool: return ( @@ -212,7 +213,7 @@ def _decode_subscription(value: Any) -> ReportingNotificationSubscription: if auth is not None: candidate["authentication"] = ReportingLegacyAuthentication(**auth) subscription = ReportingNotificationSubscription(**candidate) - normalized = json.loads(canonical_json_utf8_v1(asdict(subscription))) + normalized = json.loads(canonical_json_utf8_v1(ledger_asdict(subscription))) if normalized != value: raise ReportingNotificationError("integrity_failure") return subscription @@ -243,7 +244,9 @@ def __init__( @staticmethod def _aad(binding: DeliveryBinding) -> bytes: - return canonical_json_utf8_v1({"purpose": "adcp.reporting.delivery", **asdict(binding)}) + return canonical_json_utf8_v1( + {"purpose": "adcp.reporting.delivery", **ledger_asdict(binding)} + ) def prepare( self, @@ -282,7 +285,7 @@ def prepare( ) plaintext = canonical_json_utf8_v1( { - "subscription": asdict(subscription), + "subscription": ledger_asdict(subscription), "event": event_storage(event), "body": base64.b64encode(body).decode("ascii"), } diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index b76c8b7b1..2ace83dcc 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -72,6 +72,12 @@ decode_principal_state, encode_principal_state, ) +from adcp.reporting.subscriptions.ledger import ( + FrozenOwedSet, + SubscriptionObligationPlanner, + SubscriptionPlanResult, + subscription_obligation_id, +) from adcp.reporting.subscriptions.registry import ( ReportingSubscriptionError, apply_reporting_subscriptions, @@ -96,6 +102,10 @@ ) __all__ = [ + "FrozenOwedSet", + "SubscriptionObligationPlanner", + "SubscriptionPlanResult", + "subscription_obligation_id", "REPORTING_SUBSCRIPTION_CHANGED", "ReportingSubscriptionClientError", "ReportingSubscriptionReadback", diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index 79d179fba..015324ea7 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -75,6 +75,7 @@ _MAX_STATE_WRITE_ATTEMPTS = 8 if TYPE_CHECKING: + from adcp.reporting.subscriptions.ledger import FrozenOwedSet from adcp.server.principal import ( DeliveryConfigIdsHook, PrincipalIdentity, @@ -1371,6 +1372,64 @@ def _owed( timezone_changes=list(account.timezone_changes), ) + async def frozen_owed_sets( + self, + identity: PrincipalIdentity, + *, + account_id: str | None = None, + now: datetime | None = None, + ) -> list[FrozenOwedSet]: + """Every (account, subscription) owed set frozen through ``now``, for the ledger. + + Only periods that have ended are returned: a period's buys are frozen + at its end, and ledger obligations are immutable. A subscription that + cannot be computed for an account is skipped and recorded in + :attr:`evaluation_errors`; the others are returned. + """ + from adcp.reporting.subscriptions.ledger import FrozenOwedSet + + instant = self._now(now) + states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] + destinations = await self._destinations(identity) + async with self._store.lock: + principal = self._store.principal(identity.subject) + accounts = { + key: value + for key, value in principal.accounts.items() + if account_id is None or key == account_id + } + buys = {key: await self._host.account_buys(identity, key) for key in accounts} + result: list[FrozenOwedSet] = [] + async with self._store.lock: + principal = self._store.principal(identity.subject) + for state in states: + subscription_id = str(state["subscription_id"]) + versions = { + version: configuration + for (sub_id, version), configuration in principal.versions.items() + if sub_id == subscription_id + } + for key, account in sorted(accounts.items()): + try: + periods = self._owed( + principal, state, key, account, buys[key], destinations, instant + ) + except Exception as error: # noqa: BLE001 - isolate one owed set + self._errors[(identity.subject, subscription_id)] = error + continue + if periods: + result.append( + FrozenOwedSet( + principal_id=identity.subject, + account_id=key, + subscription_id=subscription_id, + through=instant, + periods=tuple(periods), + versions=versions, + ) + ) + return result + async def receipt_roster(self, identity: PrincipalIdentity, account_id: str) -> list[str]: """Active ``consumer_receipt`` subscriptions covering the account. diff --git a/src/adcp/reporting/subscriptions/ledger.py b/src/adcp/reporting/subscriptions/ledger.py new file mode 100644 index 000000000..3a2d75c8c --- /dev/null +++ b/src/adcp/reporting/subscriptions/ledger.py @@ -0,0 +1,311 @@ +"""Write a reporting subscription's owed set into the reporting ledger. + +The subscription engine decides which periods a subscription owes an account +(:meth:`~adcp.reporting.subscriptions.ReportingSubscriptionEngine.frozen_owed_sets`). +:class:`SubscriptionObligationPlanner` makes them ordinary ledger obligations, +so the producer executes them, ``get_reporting_status`` reports them and the +consumer-status loop accepts statements about them. + +Identity (aligned with the TypeScript SDK): one configuration generation per +(account, owning principal, ``subscription_id``, ``subscription_version``, +resolved ``offering_id``). ``delivery_config_id`` is the ``subscription_id`` +and ``delivery_config_version`` the ``subscription_version``; the generation +key carries the ``offering_id``, so several offerings of one version (``match: +all``, per-product resolution) never collide. Account-level generations keep +their identity exactly. + +Each generation holds an owed set, never a period grid: + +* obligations are written only for periods the engine froze (ended by the + planning horizon), each with its own period, the ``media_buy_ids`` owed that + period (frozen at ``scope_resolved_at``, the period end) and ``expected_at`` + = period end + the offering's ``delivery_sla``; +* obligations are immutable: a later cycle never rewrites or removes one; +* after the obligations are durable, the owed set's completeness horizon + advances on every generation the subscription owns on the account, and a + ``get_reporting_status`` scope ending after it is not closed. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from typing import Any + +from adcp.reporting.canonical_json import canonical_json_utf8_v1 +from adcp.reporting.currency import validate_currency +from adcp.reporting.ledger.delivery_models import ReportingDeliveryPrincipal +from adcp.reporting.ledger.models import ( + ReportingConfiguration, + ReportingConfigurationGenerationKey, + ReportingDefinitionBinding, + ReportingObligationRecord, + ReportingPeriodBoundary, + ReportingScheduleSpec, + ReportingSubscriptionGeneration, +) +from adcp.reporting.ledger.store import ReportingLedgerStore, ReportingSubscriptionLedger +from adcp.reporting.subscriptions._clock import parse_delivery_sla, utc_instant +from adcp.reporting.subscriptions.coverage import ExpectedSubscriptionPeriod +from adcp.reporting.subscriptions.resolution import SubscriptionOffering, _text, _value + +CurrencyResolver = Callable[[str], str] + + +@dataclass(frozen=True) +class FrozenOwedSet: + """One subscription's frozen owed set on one account. + + ``periods`` are every owed period ending by ``through``; ``versions`` maps + each retained ``subscription_version`` to its configuration, which supplies + the generation's ``feed_purpose`` and ``required_finality``. + """ + + principal_id: str + account_id: str + subscription_id: str + through: datetime + periods: tuple[ExpectedSubscriptionPeriod, ...] + versions: Mapping[int, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class SubscriptionPlanResult: + generations: tuple[ReportingConfigurationGenerationKey, ...] = () + obligations_written: tuple[str, ...] = () + through: datetime | None = None + + +def _iso(value: datetime) -> str: + return utc_instant(value).isoformat() + + +def _parse(value: str) -> datetime: + return utc_instant(datetime.fromisoformat(value.replace("Z", "+00:00"))) + + +def subscription_obligation_id( + *, + account_id: str, + principal_id: str, + subscription_id: str, + subscription_version: int, + offering_id: str, + period_start: datetime, + period_end: datetime, +) -> str: + """The deterministic id of one subscription obligation.""" + digest = hashlib.sha256( + canonical_json_utf8_v1( + [ + "subscription", + account_id, + principal_id, + subscription_id, + subscription_version, + offering_id, + _iso(period_start), + _iso(period_end), + ] + ) + ).hexdigest() + return f"rpo_{digest[:40]}" + + +def subscription_schedule(offering: SubscriptionOffering, zone: str) -> ReportingScheduleSpec: + """The ledger's record of an offering's schedule identity, as resolved for one zone. + + Kept verbatim for readback; owed-set generations never derive periods from + it (the engine's calendar did). + """ + if offering.alignment == "utc": + return ReportingScheduleSpec( + period_duration=offering.period_duration, + delivery_sla=offering.delivery_sla, + alignment="utc", + ) + if offering.alignment == "account_timezone": + return ReportingScheduleSpec( + period_duration=offering.period_duration, + delivery_sla=offering.delivery_sla, + alignment="account_timezone", + ) + return ReportingScheduleSpec( + period_duration=offering.period_duration, + delivery_sla=offering.delivery_sla, + alignment="custom_timezone", + period_timezone=zone, + period_anchor=offering.period_anchor if offering.alignment == "billing_cycle" else None, + ) + + +class SubscriptionObligationPlanner: + """Install owed-set generations and write their frozen obligations. + + ``currency`` gives each account's frozen reporting currency (a code or a + resolver). ``definitions`` binds offerings to their content-addressed + report definitions; a seller advertising ``reporting.core`` supplies them. + """ + + def __init__( + self, + store: ReportingLedgerStore, + *, + offerings: Sequence[SubscriptionOffering], + currency: str | CurrencyResolver, + definitions: Mapping[str, ReportingDefinitionBinding] | None = None, + automated_recovery_window: timedelta = timedelta(hours=6), + status_retention_days: int = 400, + ) -> None: + if not isinstance(store, ReportingSubscriptionLedger): + raise TypeError( + "this ledger store cannot hold reporting-subscription owed sets; it needs " + "list_generation_obligations, record_owed_set_progress and owed_set_through" + ) + self._store = store + self._ledger = store + self._offerings = {item.offering_id: item for item in offerings} + self._currency = currency + self._definitions = dict(definitions or {}) + self._window = automated_recovery_window + self._retention_days = status_retention_days + + def _currency_for(self, account_id: str) -> str: + value = self._currency if isinstance(self._currency, str) else self._currency(account_id) + return validate_currency(value) + + def _configuration( + self, + owed: FrozenOwedSet, + version: int, + offering: SubscriptionOffering, + zone: str, + ) -> ReportingConfiguration: + configuration = owed.versions.get(version) + if configuration is None: + raise ValueError( + f"subscription {owed.subscription_id!r} version {version} has no known " + "configuration" + ) + if hasattr(configuration, "model_dump"): + configuration = configuration.model_dump(mode="json", exclude_none=True) + selector = _value(configuration, "selector") + return ReportingConfiguration( + delivery_config_id=owed.subscription_id, + delivery_config_version=version, + account_id=owed.account_id, + consumer_id=owed.principal_id, + report_definition_id=offering.report_definition_id, + reporting_profile=offering.reporting_profile, + feed_purpose=_text(_value(selector, "feed_purpose")), + schedule=subscription_schedule(offering, zone), + required_finality=_text(_value(selector, "required_finality")), # type: ignore[arg-type] + account_timezone=zone, + activated_at=None, + automated_recovery_window=self._window, + status_retention_days=self._retention_days, + definition=self._definitions.get(offering.offering_id), + subscription=ReportingSubscriptionGeneration( + owed.subscription_id, offering.offering_id + ), + ) + + async def plan(self, owed: FrozenOwedSet) -> SubscriptionPlanResult: + """Install generations, write new obligations, then advance completeness. + + Idempotent: a repeated cycle writes nothing new. A generation whose + content would change (an offering edited under the same id) raises the + store's immutability conflict instead of silently diverging. + """ + through = utc_instant(owed.through, field="through") + installed: dict[ReportingConfigurationGenerationKey, ReportingConfiguration] = {} + written: list[str] = [] + for period in owed.periods: + start, end = _parse(period.period_start), _parse(period.period_end) + if end > through: + continue # not frozen yet: its buys may still change + offering = self._offerings.get(period.offering_id) + if offering is None: + raise ValueError(f"offering {period.offering_id!r} is not registered") + configuration = self._configuration( + owed, period.delivery_config_version, offering, period.source_timezone + ) + key = configuration.generation_key + if key not in installed: + await self._store.put_configuration(configuration) + installed[key] = configuration + existing = await self._store.find_obligation( + account_id=owed.account_id, + consumer_id=owed.principal_id, + delivery_config_id=owed.subscription_id, + delivery_config_version=period.delivery_config_version, + period_start=start, + period_end=end, + offering_id=offering.offering_id, + ) + if existing is not None: + continue # immutable: never rewritten + expected_at = end + parse_delivery_sla(offering.delivery_sla) + obligation = ReportingObligationRecord( + reporting_obligation_id=subscription_obligation_id( + account_id=owed.account_id, + principal_id=owed.principal_id, + subscription_id=owed.subscription_id, + subscription_version=period.delivery_config_version, + offering_id=offering.offering_id, + period_start=start, + period_end=end, + ), + account_id=owed.account_id, + consumer_id=owed.principal_id, + delivery_config_id=owed.subscription_id, + delivery_config_version=period.delivery_config_version, + report_definition_id=offering.report_definition_id, + reporting_profile=offering.reporting_profile, + feed_purpose=configuration.feed_purpose, + period=ReportingPeriodBoundary( + period_key=( + f"{start.strftime('%Y-%m-%dT%H:%M:%SZ')}_{offering.period_duration}" + ), + start=start, + end=end, + source_timezone=period.source_timezone, + expected_at=expected_at, + ), + scope_resolved_at=end, + media_buy_ids=tuple(period.media_buy_ids), + required_finality=configuration.required_finality, + automated_recovery_deadline_at=expected_at + self._window, + schedule=configuration.schedule, + definition=configuration.definition, + currency=self._currency_for(owed.account_id), + offering_id=offering.offering_id, + ) + stored = await self._store.commit_obligation(obligation) + written.append(stored.reporting_obligation_id) + owned = [ + item.generation_key + for item in await self._store.list_configurations( + caller=ReportingDeliveryPrincipal(owed.account_id, owed.principal_id), + delivery_config_ids=[owed.subscription_id], + ) + if item.owed_set + ] + # Completeness advances only after every obligation is durable. + for key in owned: + await self._ledger.record_owed_set_progress(key, through) + return SubscriptionPlanResult( + generations=tuple(owned), obligations_written=tuple(written), through=through + ) + + +__all__ = [ + "CurrencyResolver", + "FrozenOwedSet", + "SubscriptionObligationPlanner", + "SubscriptionPlanResult", + "subscription_obligation_id", + "subscription_schedule", +] diff --git a/tests/test_reporting_subscription_ledger.py b/tests/test_reporting_subscription_ledger.py new file mode 100644 index 000000000..1f7e2d99e --- /dev/null +++ b/tests/test_reporting_subscription_ledger.py @@ -0,0 +1,367 @@ +"""Reporting-subscription owed sets in the reporting ledger (AdCP 3.3). + +The engine decides what a subscription owes; the planner writes it into the +ledger as ordinary obligations of one generation per (account, principal, +subscription_id, version, offering_id). These tests drive the in-memory +ledger end to end: planning, idempotency, several offerings of one version, +two principals with the same subscription_id, per-principal status reads, +owed-set completeness and status ingest. +""" + +from __future__ import annotations + +from datetime import timedelta + +import pytest + +from adcp.reporting.ledger import ( + InMemoryReportingLedgerStore, + ProducerOfferings, + ReportingProducer, + ReportingStatusCaller, + ReportingStatusHandler, +) +from adcp.reporting.ledger.consumer_status import ConsumerStatusIngest +from adcp.reporting.subscriptions import SubscriptionObligationPlanner +from adcp.server import PrincipalIdentity +from tests.test_reporting_subscription_engine import ( + OFFERINGS, + World, + at, + buy, + subscription, +) + +AGENCY = PrincipalIdentity("https://agency.example/agent", "buyer_agent") +OTHER = PrincipalIdentity("https://other.example/agent", "buyer_agent") + + +class _NoSource: + @property + def capabilities(self): # pragma: no cover - never reached + raise AssertionError("not reached") + + async def execute(self, request, *, cancel, heartbeat=None): # pragma: no cover + raise AssertionError("not reached") + + +async def _world_with(identity: PrincipalIdentity, buys, *, selector=None): + world = World() + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.host.buys["acct-1"] = buys + config = subscription() if selector is None else subscription(selector=selector) + world._key += 1 + response = await world.principals.sync_principal( + identity, + { + "idempotency_key": f"ledger-test-key-{world._key:06d}", + "configuration": {"reporting_subscriptions": [config]}, + }, + ) + assert response.result.kind == "applied" + world.clock.set("2026-10-01T00:00:00Z") + await world.engine.reconcile(identity) + return world + + +def _planner(store) -> SubscriptionObligationPlanner: + return SubscriptionObligationPlanner(store, offerings=OFFERINGS, currency="USD") + + +async def _plan(world: World, identity: PrincipalIdentity, store, now: str): + planner = _planner(store) + results = [] + for owed in await world.engine.frozen_owed_sets(identity, now=at(now)): + results.append(await planner.plan(owed)) + return results + + +@pytest.mark.asyncio +async def test_owed_set_is_written_sparse_and_idempotently() -> None: + world = await _world_with( + AGENCY, + [ + buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-03T04:00:00Z"), + buy("mb-2", "2026-10-05T04:00:00Z", "2026-10-06T04:00:00Z"), + ], + ) + store = InMemoryReportingLedgerStore(clock=lambda: at("2026-10-08T00:00:00Z")) + (first,) = await _plan(world, AGENCY, store, "2026-10-08T00:00:00Z") + assert len(first.obligations_written) == 3 + (key,) = first.generations + assert key.offering_id == "pkg" and key.delivery_config_id == "analytics" + obligations = await store.list_generation_obligations(key) + # Sparse: Oct 3 and Oct 4 are owed by no buy, so no obligation exists. + assert [(o.period.start, o.media_buy_ids) for o in obligations] == [ + (at("2026-10-01T04:00:00Z"), ("mb-1",)), + (at("2026-10-02T04:00:00Z"), ("mb-1",)), + (at("2026-10-05T04:00:00Z"), ("mb-2",)), + ] + assert {o.offering_id for o in obligations} == {"pkg"} + assert obligations[0].period.expected_at == at("2026-10-02T10:00:00Z") + assert await store.owed_set_through(key) == at("2026-10-08T00:00:00Z") + # A second cycle writes nothing and never regresses the horizon. + (again,) = await _plan(world, AGENCY, store, "2026-10-08T00:00:00Z") + assert again.obligations_written == () + await store.record_owed_set_progress(key, at("2026-10-07T00:00:00Z")) + assert await store.owed_set_through(key) == at("2026-10-08T00:00:00Z") + # The producer never grid-fills an owed set. + (configuration,) = [ + c + for c in await store.list_configurations( + caller=ReportingStatusCaller("acct-1", AGENCY.subject) + ) + ] + producer = ReportingProducer(source=_NoSource(), offerings=ProducerOfferings(), store=store) + assert await producer.close_elapsed_periods(configuration, now=at("2026-10-08T00:00:00Z")) == [] + + +@pytest.mark.asyncio +async def test_several_offerings_of_one_version_never_collide() -> None: + world = await _world_with( + AGENCY, + [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-03T04:00:00Z")], + selector={ + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "official", + "dimensions": ["package"], + "match": "all", + }, + ) + store = InMemoryReportingLedgerStore(clock=lambda: at("2026-10-04T00:00:00Z")) + (result,) = await _plan(world, AGENCY, store, "2026-10-04T00:00:00Z") + assert sorted(key.offering_id or "" for key in result.generations) == [ + "pkg", + "pkg-country", + "pkg-creative", + "pkg-region-utc", + ] + assert {(k.delivery_config_id, k.delivery_config_version) for k in result.generations} == { + ("analytics", 1) + } + payload = await ReportingStatusHandler(store).handle( + {"view": "periods"}, caller=ReportingStatusCaller("acct-1", AGENCY.subject) + ) + # Every offering's obligations, one wire generation per (id, version, purpose). + assert {o["offering_id"] for o in payload["periods"]} == { + "pkg", + "pkg-country", + "pkg-creative", + "pkg-region-utc", + } + assert payload["scope"]["delivery_config_generations"] == [ + { + "delivery_config_id": "analytics", + "delivery_config_version": 1, + "feed_purpose": "analytics", + } + ] + + +@pytest.mark.asyncio +async def test_two_principals_with_one_subscription_id_see_only_their_own() -> None: + agency = await _world_with( + AGENCY, [buy("mb-a", "2026-10-01T04:00:00Z", "2026-10-03T04:00:00Z")] + ) + other = await _world_with(OTHER, [buy("mb-b", "2026-10-02T04:00:00Z", "2026-10-03T04:00:00Z")]) + store = InMemoryReportingLedgerStore(clock=lambda: at("2026-10-04T00:00:00Z")) + await _plan(agency, AGENCY, store, "2026-10-04T00:00:00Z") + await _plan(other, OTHER, store, "2026-10-04T00:00:00Z") + handler = ReportingStatusHandler(store) + mine = await handler.handle( + {"view": "periods"}, caller=ReportingStatusCaller("acct-1", AGENCY.subject) + ) + theirs = await handler.handle( + {"view": "periods"}, caller=ReportingStatusCaller("acct-1", OTHER.subject) + ) + nobody = await handler.handle( + {"view": "periods"}, caller=ReportingStatusCaller("acct-1", "https://third.example") + ) + assert {tuple(o["media_buy_ids"]) for o in mine["periods"]} == {("mb-a",)} + assert len(mine["periods"]) == 2 + assert {tuple(o["media_buy_ids"]) for o in theirs["periods"]} == {("mb-b",)} + assert nobody["periods"] == [] + + +@pytest.mark.asyncio +async def test_scope_stays_open_until_the_owed_set_is_frozen_through_it() -> None: + world = await _world_with(AGENCY, [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-10T04:00:00Z")]) + store = InMemoryReportingLedgerStore(clock=lambda: at("2026-10-08T00:00:00Z")) + await _plan(world, AGENCY, store, "2026-10-04T00:00:00Z") + caller = ReportingStatusCaller("acct-1", AGENCY.subject) + handler = ReportingStatusHandler(store) + frozen = await handler.handle( + { + "view": "periods", + "period": {"start": "2026-10-01T04:00:00Z", "end": "2026-10-03T04:00:00Z"}, + }, + caller=caller, + ) + assert frozen["scope"]["scope_closed"] is True + open_scope = await handler.handle( + { + "view": "periods", + "period": {"start": "2026-10-01T04:00:00Z", "end": "2026-10-07T04:00:00Z"}, + }, + caller=caller, + ) + # The owed set is frozen only through 10-04: later periods may still be + # added, which is not missing history. + assert open_scope["scope"]["scope_closed"] is False + assert open_scope["scope"]["coverage_complete"] is True + + +@pytest.mark.asyncio +async def test_status_ingest_finds_the_owed_obligation_by_period() -> None: + world = await _world_with(AGENCY, [buy("mb-1", "2026-10-01T04:00:00Z", "2026-10-03T04:00:00Z")]) + store = InMemoryReportingLedgerStore(clock=lambda: at("2026-10-04T00:00:00Z")) + await _plan(world, AGENCY, store, "2026-10-04T00:00:00Z") + handler = ConsumerStatusIngest(store, enabled=True, clock=lambda: at("2026-10-04T00:00:00Z")) + + def statement(identifier: str, start: str, end: str) -> dict: + return { + "reporting_status_id": identifier, + "delivery_config_id": "analytics", + "delivery_config_version": 1, + "report_definition_id": "def-pkg", + "period": {"start": start, "end": end, "source_timezone": "America/New_York"}, + "consumer_status": "obligation_missing", + "status_as_of": "2026-10-03T12:00:00Z", + } + + response = await handler.handle( + { + "statuses": [ + # A period of the owed set: matched to its owed obligation. + statement("rps_subscription_0001", "2026-10-01T04:00:00Z", "2026-10-02T04:00:00Z"), + # Not owed (no buy): never a bare grid slot. + statement("rps_subscription_0002", "2026-10-03T04:00:00Z", "2026-10-04T04:00:00Z"), + ] + }, + account_id="acct-1", + consumer_id=AGENCY.subject, + ) + owed, not_owed = response["results"] + assert owed["result"] in {"recorded", "unchanged"}, owed + assert not_owed["result"] == "failed" + assert not_owed["errors"][0]["code"] in {"INVALID_STATUS_PERIOD", "INVALID_CONSUMER_STATUS"} + + +def test_the_planner_refuses_a_store_without_owed_set_support() -> None: + class _AccountOnlyStore: + pass + + with pytest.raises(TypeError, match="owed sets"): + SubscriptionObligationPlanner(_AccountOnlyStore(), offerings=OFFERINGS, currency="USD") + + +@pytest.mark.asyncio +async def test_account_generations_keep_their_identity() -> None: + from tests.test_reporting_ledger import _configuration + + configuration = _configuration() + assert configuration.generation_key.offering_id is None + assert not configuration.owed_set + store = InMemoryReportingLedgerStore() + await store.put_configuration(configuration) + await store.put_configuration(configuration) # unchanged content is idempotent + assert configuration.generation_key in { + c.generation_key + for c in await store.list_configurations( + caller=ReportingStatusCaller(configuration.account_id, configuration.consumer_id) + ) + } + + +@pytest.mark.asyncio +async def test_the_producer_reads_only_elapsed_owed_obligations() -> None: + from dataclasses import replace + + from adcp.reporting.ledger import ReportingSubscriptionGeneration + from tests.test_reporting_ledger import _obligation + from tests.test_reporting_producer_retry import NOW, FailingSource, configuration, producer + + account_generation = configuration(config="analytics") + owed = replace( + account_generation, + activated_at=None, + media_buy_ids=(), + subscription=ReportingSubscriptionGeneration("analytics", "pkg"), + ) + store = InMemoryReportingLedgerStore(clock=lambda: NOW) + await store.put_configuration(owed) + hour = timedelta(hours=1) + for identifier, start in (("rpo_elapsed", NOW - 2 * hour), ("rpo_later", NOW + hour)): + template = _obligation(account_generation) + start = start.replace(minute=0) + await store.commit_obligation( + replace( + template, + reporting_obligation_id=identifier, + media_buy_ids=("mb-1",), + offering_id="pkg", + period=replace( + template.period, + period_key=start.strftime("%Y-%m-%dT%H:%M:%SZ") + "_PT1H", + start=start, + end=start + hour, + expected_at=start + hour + timedelta(minutes=10), + ), + scope_resolved_at=start + hour, + created_at=NOW, + ) + ) + source = FailingSource() + turn = await producer(source, store).run_configuration(owed, now=NOW) + # The owed set is not grid-filled, and a period that has not ended is not read. + assert len(await store.list_generation_obligations(owed.generation_key)) == 2 + assert source.calls == 1 + assert turn.slices_failed in ([], ["rpo_elapsed"]) + assert not await store.lease_period_close(worker_id="w", now=NOW, lease_seconds=30) + + +def test_account_generation_bytes_are_unchanged_by_the_offering_dimension() -> None: + """Every digest and persisted scope of an account generation stays byte-stable.""" + from dataclasses import asdict + + from adcp.reporting.ledger.models import ( + ReportingConfigurationGenerationKey, + ledger_asdict, + ) + from adcp.reporting.ledger.notification_models import ( + ReportingStatusScope, + decode_status_scope, + ) + + key = ReportingConfigurationGenerationKey("acct-1", "buyer", "daily", 1) + assert ledger_asdict(key) == { + "account_id": "acct-1", + "consumer_id": "buyer", + "delivery_config_id": "daily", + "delivery_config_version": 1, + } + owed = ReportingConfigurationGenerationKey("acct-1", "buyer", "daily", 1, "pkg") + assert ledger_asdict(owed) == {**ledger_asdict(key), "offering_id": "pkg"} + assert owed != key and len({owed, key}) == 2 + # A scope persisted before subscriptions existed still decodes. + scope = ReportingStatusScope("acct-1", key, consumer_id="buyer") + legacy = asdict(scope) + del legacy["generation_key"]["offering_id"] + assert decode_status_scope(legacy) == scope + assert ledger_asdict(scope) == legacy + + +def test_reporting_code_never_serializes_a_generation_key_with_bare_asdict() -> None: + import re + from pathlib import Path + + root = Path(__file__).resolve().parents[1] + offenders = [ + str(path.relative_to(root)) + for base in ("src/adcp/reporting", "packages/adcp-reporting-storage/src") + for path in (root / base).rglob("*.py") + if not path.as_posix().endswith("ledger/models.py") + and re.search(r"(? Date: Fri, 30 Oct 2026 03:14:52 +0000 Subject: [PATCH 2/4] fix(reporting): repin the reconciliation guard and the public API snapshot The closed-payload allowlist in reporting_reconciliation_validate now names offering_id, so the required-schema manifest carries the function's new fingerprint, computed from a clean migration chain. The public API snapshot records ReportingSubscriptionGeneration and ReportingSubscriptionLedger. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/adcp_reporting_storage/outbox/required_schema.json | 2 +- tests/fixtures/public_api_snapshot.json | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_schema.json b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_schema.json index 38cbc3623..a9af8299b 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_schema.json +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/outbox/required_schema.json @@ -1437,7 +1437,7 @@ }, "function:reporting_reconciliation_validate(r reporting_reconciliation_records)": { "enabled": true, - "fingerprint": "c4790a050f91451c7b7d5974c4513a8f9d2c5703c520604c4b87ba7aea443e8a" + "fingerprint": "6e1be3ec1aa59ba6b7b3e0fdc78deba9d36ff37f72f81373a64c3bda13832d65" }, "function:reporting_sorted_strings(document jsonb)": { "enabled": true, diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 1dabac09d..0bd0ff4a7 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -898,6 +898,8 @@ "ReportingStatusParticipant": "adcp.reporting.ledger.status_snapshot", "ReportingStatusSnapshot": "adcp.reporting.ledger.status_projection", "ReportingStatusView": "union[value:'periods',value:'revision',value:'summary']", + "ReportingSubscriptionGeneration": "adcp.reporting.ledger.models", + "ReportingSubscriptionLedger": "adcp.reporting.ledger.store", "ReportingVerificationRecord": "adcp.reporting.ledger.delivery_models", "ReportingWorkerRetryPolicy": "adcp.reporting.ledger.store", "RestatementCheckpoint": "adcp.reporting.ledger.store", From ff039e496320ac7cf7db9ec6f5eeacf920384278 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 03:22:35 +0000 Subject: [PATCH 3/4] fix(reporting): read frozen owed sets from the engine state store The engine now persists its state through a compare-and-swap store, so frozen_owed_sets reads one snapshot of the principal's state, as owed_periods and subscription_coverage do, instead of the removed process-local lock. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/engine.py | 65 +++++++++++----------- 1 file changed, 31 insertions(+), 34 deletions(-) diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index 015324ea7..6c316b38f 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -1391,43 +1391,40 @@ async def frozen_owed_sets( instant = self._now(now) states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] destinations = await self._destinations(identity) - async with self._store.lock: - principal = self._store.principal(identity.subject) - accounts = { - key: value - for key, value in principal.accounts.items() - if account_id is None or key == account_id - } + principal = await self._read(identity.subject) + accounts = { + key: value + for key, value in principal.accounts.items() + if account_id is None or key == account_id + } buys = {key: await self._host.account_buys(identity, key) for key in accounts} result: list[FrozenOwedSet] = [] - async with self._store.lock: - principal = self._store.principal(identity.subject) - for state in states: - subscription_id = str(state["subscription_id"]) - versions = { - version: configuration - for (sub_id, version), configuration in principal.versions.items() - if sub_id == subscription_id - } - for key, account in sorted(accounts.items()): - try: - periods = self._owed( - principal, state, key, account, buys[key], destinations, instant - ) - except Exception as error: # noqa: BLE001 - isolate one owed set - self._errors[(identity.subject, subscription_id)] = error - continue - if periods: - result.append( - FrozenOwedSet( - principal_id=identity.subject, - account_id=key, - subscription_id=subscription_id, - through=instant, - periods=tuple(periods), - versions=versions, - ) + for state in states: + subscription_id = str(state["subscription_id"]) + versions = { + version: configuration + for (sub_id, version), configuration in principal.versions.items() + if sub_id == subscription_id + } + for key, account in sorted(accounts.items()): + try: + periods = self._owed( + principal, state, key, account, buys[key], destinations, instant + ) + except Exception as error: # noqa: BLE001 - isolate one owed set + self._errors[(identity.subject, subscription_id)] = error + continue + if periods: + result.append( + FrozenOwedSet( + principal_id=identity.subject, + account_id=key, + subscription_id=subscription_id, + through=instant, + periods=tuple(periods), + versions=versions, ) + ) return result async def receipt_roster(self, identity: PrincipalIdentity, account_id: str) -> list[str]: From 171d6a8d35706e5a3bce234a35488efa316ea239 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 03:42:13 +0000 Subject: [PATCH 4/4] feat(reporting): serve reporting subscriptions from ReliableReportingService ReliableReportingService(reporting_subscriptions=...) now owes and serves a principal's reporting subscriptions. Without the option, capabilities, responses and worker turns are unchanged. - Each worker turn plans every listed principal's frozen owed set, then executes it with the account configurations. A subscription generation is installed through the same trusted configure path: account_context receives it with configuration.subscription set, and the service validates the source route, currency and advertised offering as for an account configuration. The horizon is the earlier of the engine's and the turn clock. - A subscription that cannot be planned is isolated, recorded in turn.subscription_errors and reported to planning_error_handler or the worker error handler; it never stops another subscription or an account configuration. - Capabilities gain configuration_tasks, the subscriptions block, standard_dimensions, the feature id and the principal section at the served release; callers below 3.3 see the pre-3.3 block, and adcp#8585 offering fields reach only 3.3.0-beta.2 callers. - get_reporting_status summary and periods views gain subscription_coverage. Ledger reads stay scoped to the caller, and resolve_principal is checked against the caller resolver. - The planner takes an install hook and resolves currency per generation; the engine exposes its offerings and status_retention_days. Production options, managed or receipt-reconciled subscription offerings and PostgreSQL owed sets are refused or not yet served. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 66 ++- src/adcp/decisioning/handler.py | 5 +- src/adcp/reporting/service.py | 170 +++++++- src/adcp/reporting/service_subscriptions.py | 248 ++++++++++++ src/adcp/reporting/subscriptions/engine.py | 9 + src/adcp/reporting/subscriptions/ledger.py | 40 +- tests/test_reporting_subscription_service.py | 404 +++++++++++++++++++ 7 files changed, 919 insertions(+), 23 deletions(-) create mode 100644 src/adcp/reporting/service_subscriptions.py create mode 100644 tests/test_reporting_subscription_service.py diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 17da23fa6..0a19fd50f 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -414,9 +414,69 @@ dropped `status_retention_days` after they close. `subscription_capabilities(...)` and `advertise_reporting_subscriptions(...)` project `configuration_tasks`, the deprecated `configuration_task`, the -`subscriptions` block, the feature id and the principal section. The engine -does not yet write ledger obligations, so do not advertise the capability -until obligations are materialized for the periods it owes. +`subscriptions` block, the feature id and the principal section. + +**Serving subscriptions from the service.** Pass +`reporting_subscriptions=ReportingSubscriptionServiceOptions(...)` to +`ReliableReportingService`. Without it, capabilities, responses and worker +turns are unchanged. + +```python +from adcp.reporting.service_subscriptions import ReportingSubscriptionServiceOptions + +reporting = ReliableReportingService( + store=store, + account_context=resolve_account_context, # also receives subscription generations + reporting_subscriptions=ReportingSubscriptionServiceOptions( + engine=engine, + list_principals=list_subscribing_principals, + resolve_principal=resolve_authenticated_principal, + definitions={"daily_v1": daily_definition}, # content-addressed, per offering + ), +) +``` + +What the service does with it: + +- **Planning.** Each worker turn first plans every listed principal's frozen + owed set, then executes those obligations alongside the account + configurations. + - A subscription generation is one per (account, principal, + `subscription_id`, version, resolved `offering_id`). Its `consumer_id` is + the principal's `subject`. + - Each generation is installed through the same trusted path as + `configure()`. `account_context` receives it with + `configuration.subscription` set and returns the source route, currency + and advertised offering. The service validates these as it does for an + account configuration. + - Obligations are written only for periods ending by + `min(engine through, turn clock)`. Each carries exactly that period's + media buys and is never rewritten. + - A subscription that cannot be planned is recorded in + `turn.subscription_errors` and reported to `planning_error_handler` (or + the worker error handler). It never stops another subscription or an + account configuration. +- **Capabilities.** The service adds `configuration_tasks`, the + `subscriptions` block, `standard_dimensions` on every offering, the feature + id and the principal section, at the release each caller is served. + - Callers served below 3.3 see the pre-3.3 block. + - The adcp#8585 offering fields reach only 3.3.0-beta.2 callers. + - Advertise `adcp.principal` yourself. The service refuses to advertise + subscriptions without it. +- **Status.** `get_reporting_status` summary and periods views gain + `subscription_coverage[]`. +- **Visibility.** Ledger reads and `sync_reporting_status` are scoped by the + caller's `consumer_id`, so a principal sees only its own subscription + periods. The caller resolver must therefore yield the principal's `subject`; + the service checks `resolve_principal` against it. + +**Limits.** These are not served yet: + +- the service refuses `production` options; +- subscriptions that resolve a managed or receipt-reconciled offering; +- PostgreSQL ledger storage of owed sets. + +Keep the engine's `status_retention_days` equal to the service's. ## Advanced composition diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index 6f734d9a3..c863e9153 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -2269,7 +2269,10 @@ async def get_adcp_capabilities( reporting = getattr(self._platform, "_reliable_reporting_service", None) if reporting is not None: - response = reporting.inject_capabilities(response) + response = reporting.inject_capabilities( + response, + adcp_version=context.resolved_adcp_version if context is not None else None, + ) await validate_status_claims(response, support=self._reporting_status, handler=self) diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index ba973cc65..18f1a2b4d 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -71,6 +71,10 @@ from adcp.reporting.production.contracts import ReportingProductionSourceBinding from adcp.reporting.production.service import ReportingProductionSupport from adcp.reporting.service_production import ReportingProductionOptions + from adcp.reporting.service_subscriptions import ( + ReportingSubscriptionServiceOptions, + SubscriptionPlanningFailure, + ) __all__ = [ "AdapterRegistration", @@ -436,6 +440,12 @@ class ReliableReportingTurn: ) extension_results: list[Any] = field(default_factory=list) extension_errors: dict[str, BaseException] = field(default_factory=dict) + #: Reporting-subscription obligations written this turn. + subscription_obligations_written: int = 0 + #: (principal subject, account_id, subscription_id) -> why it was not planned. + subscription_errors: dict[tuple[str, str | None, str | None], BaseException] = field( + default_factory=dict + ) @property def did_work(self) -> bool: @@ -444,6 +454,8 @@ def did_work(self) -> bool: or bool(self.extension_results) or bool(self.configuration_errors) or bool(self.extension_errors) + or bool(self.subscription_obligations_written) + or bool(self.subscription_errors) ) @@ -481,10 +493,14 @@ def __init__( staging: ReportingStagingStore | None = None, seals: ReportingSealStore | None = None, initialize_source_storage: Callable[[], Awaitable[None]] | None = None, + reporting_subscriptions: ReportingSubscriptionServiceOptions | None = None, ) -> None: """``staging``/``seals`` are the default source stores adapters register with; ``initialize_source_storage`` runs once at startup, after ``store.create_schema()``, - so a durable storage distribution can install its own schema.""" + so a durable storage distribution can install its own schema. + ``reporting_subscriptions`` serves AdCP 3.3 principal-level reporting + subscriptions (see :mod:`adcp.reporting.service_subscriptions`); without + it the service is unchanged.""" if production is not None and ( caller_resolver is not None or producer_factory is not ReportingProducer @@ -500,6 +516,10 @@ def __init__( raise ReliableReportingConfigurationError( "production options compose their own workers, admission and receipt handlers" ) + if production is not None and reporting_subscriptions is not None: + raise ReliableReportingConfigurationError( + "reporting subscriptions are not yet served with production options" + ) effective_clock = clock or (lambda: datetime.now(timezone.utc)) self.store = store self._production: ReportingProductionSupport | None = None @@ -544,6 +564,23 @@ def __init__( ReportingConfigurationGenerationKey, ReportingConfiguration ] = {} self._turn_lock = asyncio.Lock() + self._subscriptions = reporting_subscriptions + self._subscription_planner: Any | None = None + if reporting_subscriptions is not None: + from adcp.reporting.subscriptions.ledger import SubscriptionObligationPlanner + + try: + self._subscription_planner = SubscriptionObligationPlanner( + store, + offerings=reporting_subscriptions.engine.offerings, + currency=self._subscription_currency, + definitions=reporting_subscriptions.definitions, + automated_recovery_window=automated_recovery_window, + status_retention_days=status_retention_days, + install=self._install_subscription_generation, + ) + except TypeError as error: + raise ReliableReportingConfigurationError(str(error)) from None self._lifecycle = _ServiceLifecycle( self._initialize, resources=( @@ -693,7 +730,12 @@ async def _configure( raise ReliableReportingConfigurationError( "resolved account context does not match the configuration account" ) - if context.account_timezone != configuration.account_timezone: + # An owed set never derives periods from its schedule: each + # obligation carries its own period and source timezone. + if ( + not configuration.owed_set + and context.account_timezone != configuration.account_timezone + ): raise ReliableReportingConfigurationError( "resolved account timezone does not match the configuration timezone" ) @@ -753,6 +795,45 @@ def initialization_errors(self) -> Mapping[ReportingConfigurationGenerationKey, """ return MappingProxyType(self._initialization_errors) + async def _install_subscription_generation(self, configuration: ReportingConfiguration) -> None: + """Install one subscription generation through the trusted configure path.""" + async with self._configuration_lock: + await self._configure(configuration) + + def _subscription_currency(self, configuration: ReportingConfiguration) -> str: + return self._bindings[configuration.generation_key].context.currency + + async def _plan_subscriptions(self, turn: ReliableReportingTurn, *, now: datetime) -> None: + from adcp.reporting.service_subscriptions import plan_subscription_cycle + + assert self._subscriptions is not None + written, failures = await plan_subscription_cycle( + self._subscriptions, + self._subscription_planner, + now=now, + stopping=lambda: self._lifecycle.stopping, + ) + turn.subscription_obligations_written += written + for failure in failures: + turn.subscription_errors[failure.key] = failure.error + await self._report_planning_failure(failure) + + async def _report_planning_failure(self, failure: SubscriptionPlanningFailure) -> None: + assert self._subscriptions is not None + handler = self._subscriptions.planning_error_handler + if handler is None: + await self._report_worker_error( + "subscription:" + f"{failure.principal.subject}:{failure.account_id or ''}:" + f"{failure.subscription_id or ''}", + failure.error, + ) + return + try: + await _resolve(handler(failure)) + except Exception: + logger.error("Reliable Reporting subscription planning error handler failed") + async def _recover_configuration( self, configuration: ReportingConfiguration, *, retrying: bool = False ) -> None: @@ -877,6 +958,8 @@ def validate(self) -> None: raise ReliableReportingConfigurationError( "a notification worker requires durable notification attempt storage" ) + if self._subscriptions is not None: + self._validate_subscriptions() if self._receipt_handler is not None and self._materialization_worker is None: raise ReliableReportingConfigurationError( "receipt handling requires managed delivery/materialization" @@ -902,6 +985,30 @@ def validate(self) -> None: "consumer_receipt offerings require reconciled billing" ) + def _validate_subscriptions(self) -> None: + assert self._subscriptions is not None + engine = self._subscriptions.engine + if engine.status_retention_days != self._status_retention_days: + raise ReliableReportingConfigurationError( + "the reporting subscription engine must use the status_retention_days this " + "service advertises" + ) + mapped = {item.offering_id for item in engine.offerings} + for name in self.sources.names: + for offering in self.sources.get(name).capability_offerings: + # adcp#8585: every advertised offering publishes standard_dimensions, + # so buyers resolve a selector from the wire as the engine does. + if offering["offering_id"] not in mapped: + raise ReliableReportingConfigurationError( + f"offering {offering['offering_id']!r} has no subscription dimension " + "mapping; register it with the reporting subscription engine" + ) + if offering.get("method") or offering["reconciliation_mode"] != "delivery_only": + raise ReliableReportingConfigurationError( + "reporting subscriptions do not yet serve managed or receipt-reconciled " + "offerings" + ) + async def initialize(self) -> None: """Prepare resources once; start/run_worker opens reporting admission.""" await self._lifecycle.initialize() @@ -1032,6 +1139,9 @@ async def _run_worker(self, *, now: datetime | None) -> ReliableReportingTurn: break await self._recover_configuration(configuration, retrying=True) turn.configuration_errors.update(self._initialization_errors) + if self._subscriptions is not None and not self._lifecycle.stopping: + # Plan first, so newly frozen obligations run in this turn. + await self._plan_subscriptions(turn, now=now or self._clock()) with source_turn(): for key, binding in sorted( self._bindings.items(), @@ -1039,6 +1149,8 @@ async def _run_worker(self, *, now: datetime | None) -> ReliableReportingTurn: item[0].account_id, item[0].delivery_config_id, item[0].delivery_config_version, + item[0].consumer_id, + item[0].offering_id or "", ), ): if self._lifecycle.stopping: @@ -1121,11 +1233,23 @@ async def get_reporting_status( async def _get_reporting_status(self, request: Any, context: Any | None) -> dict[str, Any]: caller = await self.caller_for(request, context) - return await ReportingStatusHandler( + response = await ReportingStatusHandler( self.store, consumer_status_enabled=self._consumer_status_enabled, escalation=self._escalation, ).handle(_wire(request), caller=caller) + if self._subscriptions is None: + return response + from adcp.reporting.service_subscriptions import with_subscription_coverage + + return await with_subscription_coverage( + response, + self._subscriptions, + request=request, + context=context, + account_id=caller.account_id, + consumer_id=caller.consumer_id, + ) async def sync_reporting_status( self, request: Any, context: Any | None = None @@ -1221,8 +1345,25 @@ async def _get_revision_content(self, request: Any, context: Any | None) -> dict response["context"] = payload["context"] return response - def capability_block(self) -> dict[str, Any]: - """Project only components and offerings this service actually installed.""" + def capability_block(self, *, adcp_version: str | None = None) -> dict[str, Any]: + """Project only components and offerings this service actually installed. + + ``adcp_version`` is the release the response is served at; it decides + which reporting-subscription fields appear when subscriptions are served. + """ + block = self._capability_block() + if self._subscriptions is None or not block: + return block + from adcp.reporting.service_subscriptions import subscription_capability_block + + try: + return subscription_capability_block( + block, self._subscriptions, adcp_version=adcp_version + ) + except (KeyError, ValueError) as error: + raise ReliableReportingConfigurationError(str(error)) from None + + def _capability_block(self) -> dict[str, Any]: if self._production is not None: return dict(self._production._declared_capabilities()["reporting_delivery"]) if self._production_options is not None: @@ -1278,10 +1419,12 @@ def capability_block(self) -> dict[str, Any]: payload["receipt_task"] = "sync_reporting_receipts" return payload - def inject_capabilities(self, response: Any) -> dict[str, Any]: + def inject_capabilities( + self, response: Any, *, adcp_version: str | None = None + ) -> dict[str, Any]: """Merge the truthful reporting block into a base capability response.""" payload = _wire(response) - block = self.capability_block() + block = self.capability_block(adcp_version=adcp_version) if not block: return payload media_buy = dict(payload.get("media_buy") or {}) @@ -1295,6 +1438,15 @@ def inject_capabilities(self, response: Any) -> dict[str, Any]: if "media_buy.reporting_delivery" not in experimental: experimental.append("media_buy.reporting_delivery") payload["experimental_features"] = experimental + if self._subscriptions is not None: + from adcp.reporting.service_subscriptions import advertise_subscriptions + + try: + payload = advertise_subscriptions( + payload, self._subscriptions, adcp_version=adcp_version + ) + except ValueError as error: + raise ReliableReportingConfigurationError(str(error)) from None return payload def install(self, platform: Any) -> Any: @@ -1384,7 +1536,9 @@ async def get_adcp_capabilities( instance: Any, params: Any, context: Any | None = None ) -> Any: result = await _resolve(original.get_adcp_capabilities(instance, params, context)) - return service.inject_capabilities(result) + return service.inject_capabilities( + result, adcp_version=getattr(context, "resolved_adcp_version", None) + ) async def get_media_buy_delivery( instance: Any, params: Any, context: Any | None = None diff --git a/src/adcp/reporting/service_subscriptions.py b/src/adcp/reporting/service_subscriptions.py new file mode 100644 index 000000000..09ed90fed --- /dev/null +++ b/src/adcp/reporting/service_subscriptions.py @@ -0,0 +1,248 @@ +"""Reporting subscriptions in :class:`~adcp.reporting.service.ReliableReportingService`. + +Opt in with ``ReliableReportingService(reporting_subscriptions=...)``. Without +it, capabilities, responses and worker turns are exactly what they were. + +With it, the service: + +* **plans** every listed principal's frozen owed sets at the start of each + worker turn, through :class:`~adcp.reporting.subscriptions.SubscriptionObligationPlanner`, + and then executes them with the account generations. Each (version, + offering) generation is installed through the same trusted path as + :meth:`~adcp.reporting.service.ReliableReportingService.configure`: the + ``account_context`` resolver receives the generation (its ``subscription`` + is set) and returns its source route, currency and advertised offering, and + the service validates them as for an account configuration. The horizon is + ``min(engine through, turn clock)``. One subscription's failure never stops + another's or an account configuration's; it is recorded on the turn and + reported to ``planning_error_handler`` (or the worker error handler). +* **advertises** ``configuration_tasks``, the ``subscriptions`` block, + ``standard_dimensions`` on every offering, the + ``media_buy.reporting_subscriptions`` feature and the principal section, at + the release each caller is served. Callers served below 3.3 see the + pre-3.3 block. +* **adds** ``subscription_coverage[]`` to completed ``summary`` and + ``periods`` views of ``get_reporting_status``. + +Ownership: a subscription generation belongs to its principal, whose +``subject`` is the generation's ``consumer_id``. Ledger reads and status +ingest are scoped by the resolved caller's ``consumer_id``, so the caller +resolver must yield the principal's subject; ``resolve_principal`` is checked +against it. +""" + +from __future__ import annotations + +import inspect +from collections.abc import Awaitable, Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime +from typing import TYPE_CHECKING, Any + +from adcp._version import is_adcp_version_at_least, resolve_adcp_version_alias +from adcp.reporting._errors import ReliableReportingConfigurationError +from adcp.reporting.ledger.models import ReportingDefinitionBinding + +if TYPE_CHECKING: + from adcp.reporting.subscriptions.engine import ReportingSubscriptionEngine + from adcp.server.principal import PrincipalIdentity + +#: Every principal whose subscription obligations the service writes. +PrincipalLister = Callable[ + [], "Sequence[PrincipalIdentity] | Awaitable[Sequence[PrincipalIdentity]]" +] +#: The authenticated principal of a request; never derived from request content. +PrincipalResolver = Callable[[Any, Any], "PrincipalIdentity | Awaitable[PrincipalIdentity]"] + +_FIRST_RELEASE = "3.3.0-beta.1" + + +@dataclass(frozen=True) +class SubscriptionPlanningFailure: + """One subscription (or one principal's listing) a turn could not plan.""" + + principal: PrincipalIdentity + error: Exception + account_id: str | None = None + subscription_id: str | None = None + + @property + def key(self) -> tuple[str, str | None, str | None]: + return (self.principal.subject, self.account_id, self.subscription_id) + + +PlanningErrorHandler = Callable[[SubscriptionPlanningFailure], "Awaitable[None] | None"] + + +@dataclass(frozen=True) +class ReportingSubscriptionServiceOptions: + """Serve principal-level reporting subscriptions from the reliable-reporting service. + + ``engine`` is the :class:`~adcp.reporting.subscriptions.ReportingSubscriptionEngine` + linked to the seller's ``PrincipalService``; the host keeps it reconciled + (``reconcile`` on ``principal.changed`` and authorization changes). + ``list_principals`` names every principal whose obligations each turn + writes. ``resolve_principal`` is the same authenticated mapping the + principal runtime uses. ``definitions`` binds offering ids to their + content-addressed report definitions; a seller advertising + ``reporting.core`` supplies them, as it does on account configurations. + """ + + engine: ReportingSubscriptionEngine + list_principals: PrincipalLister + resolve_principal: PrincipalResolver + planning_error_handler: PlanningErrorHandler | None = None + definitions: Mapping[str, ReportingDefinitionBinding] | None = None + + def __post_init__(self) -> None: + for name in ("list_principals", "resolve_principal"): + if not callable(getattr(self, name)): + raise TypeError(f"reporting_subscriptions requires {name}") + if not hasattr(self.engine, "frozen_owed_sets") or not hasattr( + self.engine, "subscription_coverage" + ): + raise TypeError("reporting_subscriptions requires a reporting subscription engine") + + +async def _resolve(value: Any) -> Any: + return await value if inspect.isawaitable(value) else value + + +def serves_subscriptions(adcp_version: str | None) -> bool: + """Whether a response served at ``adcp_version`` may carry subscription fields. + + An absent version is the SDK's pinned default (3.3.0-beta.1); an + unparseable one is treated as older. + """ + if adcp_version is None: + return True + try: + return is_adcp_version_at_least(resolve_adcp_version_alias(adcp_version), _FIRST_RELEASE) + except ValueError: + return False + + +async def plan_subscription_cycle( + options: ReportingSubscriptionServiceOptions, + planner: Any, + *, + now: datetime, + stopping: Callable[[], bool], +) -> tuple[int, list[SubscriptionPlanningFailure]]: + """Plan every listed principal's frozen owed sets, isolating each subscription.""" + principals = await _resolve(options.list_principals()) + if isinstance(principals, (str, bytes)) or not isinstance(principals, Sequence): + raise TypeError("list_principals must return a sequence of PrincipalIdentity") + written = 0 + failures: list[SubscriptionPlanningFailure] = [] + for principal in principals: + if stopping(): + break + try: + owed_sets = await options.engine.frozen_owed_sets(principal, now=now) + except Exception as error: # noqa: BLE001 - isolate one principal + failures.append(SubscriptionPlanningFailure(principal, error)) + continue + for owed in owed_sets: + if stopping(): + break + try: + result = await planner.plan(owed, through=now) + except Exception as error: # noqa: BLE001 - isolate one subscription + failures.append( + SubscriptionPlanningFailure( + principal, + error, + account_id=owed.account_id, + subscription_id=owed.subscription_id, + ) + ) + continue + written += len(result.obligations_written) + return written, failures + + +def subscription_capability_block( + block: Mapping[str, Any], + options: ReportingSubscriptionServiceOptions, + *, + adcp_version: str | None, +) -> dict[str, Any]: + """``reporting_delivery`` with the subscription registration fields, as served.""" + from adcp.reporting.subscriptions.capabilities import ( + subscription_capabilities, + with_subscription_dimensions, + ) + from adcp.reporting.subscriptions.wire import project_reporting_delivery + + payload = dict(block) + if not payload or not serves_subscriptions(adcp_version): + return payload + mappings = {item.offering_id: item for item in options.engine.offerings} + payload["offerings"] = [ + with_subscription_dimensions(item, mappings[str(item["offering_id"])]) + for item in payload.get("offerings") or () + ] + payload = subscription_capabilities(payload, options.engine.support) + return project_reporting_delivery(payload, adcp_version) + + +def advertise_subscriptions( + response: Mapping[str, Any], + options: ReportingSubscriptionServiceOptions, + *, + adcp_version: str | None, +) -> dict[str, Any]: + """Add the feature id and the principal section, when subscriptions are served.""" + from adcp.reporting.subscriptions.capabilities import advertise_reporting_subscriptions + + if not serves_subscriptions(adcp_version): + return dict(response) + return advertise_reporting_subscriptions(response, options.engine.support) + + +async def with_subscription_coverage( + response: dict[str, Any], + options: ReportingSubscriptionServiceOptions, + *, + request: Any, + context: Any, + account_id: str, + consumer_id: str, +) -> dict[str, Any]: + """Add ``subscription_coverage[]`` to a completed summary or periods view.""" + if response.get("status") != "completed" or response.get("view") not in { + "summary", + "periods", + }: + return response + adcp_version = getattr(context, "resolved_adcp_version", None) + if not serves_subscriptions(adcp_version): + return response + principal = await _resolve(options.resolve_principal(request, context)) + if getattr(principal, "subject", None) != consumer_id: + raise ReliableReportingConfigurationError( + "resolve_principal and the caller resolver disagree on the requesting principal" + ) + coverage = await options.engine.subscription_coverage( + principal, account_id, adcp_version=adcp_version + ) + if not coverage: + return response + return { + **response, + "subscription_coverage": [ + item.model_dump(mode="json", exclude_none=True) if hasattr(item, "model_dump") else item + for item in coverage + ], + } + + +__all__ = [ + "PlanningErrorHandler", + "PrincipalLister", + "PrincipalResolver", + "ReportingSubscriptionServiceOptions", + "SubscriptionPlanningFailure", + "serves_subscriptions", +] diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index 6c316b38f..5591c65b0 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -571,6 +571,15 @@ def evaluation_errors(self) -> Mapping[tuple[str, str], Exception]: def support(self) -> ReportingSubscriptionSupport: return self._support + @property + def offerings(self) -> tuple[SubscriptionOffering, ...]: + """The offerings subscriptions resolve against, with their dimension mappings.""" + return self._offerings + + @property + def status_retention_days(self) -> int: + return self._retention_days + def principal_options( self, *, delivery_config_ids: DeliveryConfigIdsHook | None = None ) -> ReportingSubscriptionOptions: diff --git a/src/adcp/reporting/subscriptions/ledger.py b/src/adcp/reporting/subscriptions/ledger.py index 3a2d75c8c..ff7838607 100644 --- a/src/adcp/reporting/subscriptions/ledger.py +++ b/src/adcp/reporting/subscriptions/ledger.py @@ -29,7 +29,7 @@ from __future__ import annotations import hashlib -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Any @@ -51,7 +51,12 @@ from adcp.reporting.subscriptions.coverage import ExpectedSubscriptionPeriod from adcp.reporting.subscriptions.resolution import SubscriptionOffering, _text, _value -CurrencyResolver = Callable[[str], str] +#: The frozen reporting currency of one generation, as its account context gives it. +CurrencyResolver = Callable[[ReportingConfiguration], str] +#: Installs one generation through the host's trusted path (for example +#: ``ReliableReportingService``, which resolves and validates the account +#: context and persists the generation). Defaults to ``put_configuration``. +GenerationInstaller = Callable[[ReportingConfiguration], Awaitable[None]] @dataclass(frozen=True) @@ -144,9 +149,11 @@ def subscription_schedule(offering: SubscriptionOffering, zone: str) -> Reportin class SubscriptionObligationPlanner: """Install owed-set generations and write their frozen obligations. - ``currency`` gives each account's frozen reporting currency (a code or a - resolver). ``definitions`` binds offerings to their content-addressed - report definitions; a seller advertising ``reporting.core`` supplies them. + ``currency`` gives each generation's frozen reporting currency (a code, + or a resolver called after the generation is installed). ``definitions`` + binds offerings to their content-addressed report definitions; a seller + advertising ``reporting.core`` supplies them. ``install`` replaces the + plain ``put_configuration`` with the host's trusted installation path. """ def __init__( @@ -158,6 +165,7 @@ def __init__( definitions: Mapping[str, ReportingDefinitionBinding] | None = None, automated_recovery_window: timedelta = timedelta(hours=6), status_retention_days: int = 400, + install: GenerationInstaller | None = None, ) -> None: if not isinstance(store, ReportingSubscriptionLedger): raise TypeError( @@ -171,9 +179,10 @@ def __init__( self._definitions = dict(definitions or {}) self._window = automated_recovery_window self._retention_days = status_retention_days + self._install = install or store.put_configuration - def _currency_for(self, account_id: str) -> str: - value = self._currency if isinstance(self._currency, str) else self._currency(account_id) + def _currency_for(self, configuration: ReportingConfiguration) -> str: + value = self._currency if isinstance(self._currency, str) else self._currency(configuration) return validate_currency(value) def _configuration( @@ -212,14 +221,22 @@ def _configuration( ), ) - async def plan(self, owed: FrozenOwedSet) -> SubscriptionPlanResult: + async def plan( + self, owed: FrozenOwedSet, *, through: datetime | None = None + ) -> SubscriptionPlanResult: """Install generations, write new obligations, then advance completeness. + ``through`` caps the engine's horizon (a service passes its cycle + clock), so nothing past it is written or claimed complete. + Idempotent: a repeated cycle writes nothing new. A generation whose content would change (an offering edited under the same id) raises the store's immutability conflict instead of silently diverging. """ - through = utc_instant(owed.through, field="through") + horizon = utc_instant(owed.through, field="through") + if through is not None: + horizon = min(horizon, utc_instant(through, field="through")) + through = horizon installed: dict[ReportingConfigurationGenerationKey, ReportingConfiguration] = {} written: list[str] = [] for period in owed.periods: @@ -234,7 +251,7 @@ async def plan(self, owed: FrozenOwedSet) -> SubscriptionPlanResult: ) key = configuration.generation_key if key not in installed: - await self._store.put_configuration(configuration) + await self._install(configuration) installed[key] = configuration existing = await self._store.find_obligation( account_id=owed.account_id, @@ -280,7 +297,7 @@ async def plan(self, owed: FrozenOwedSet) -> SubscriptionPlanResult: automated_recovery_deadline_at=expected_at + self._window, schedule=configuration.schedule, definition=configuration.definition, - currency=self._currency_for(owed.account_id), + currency=self._currency_for(configuration), offering_id=offering.offering_id, ) stored = await self._store.commit_obligation(obligation) @@ -304,6 +321,7 @@ async def plan(self, owed: FrozenOwedSet) -> SubscriptionPlanResult: __all__ = [ "CurrencyResolver", "FrozenOwedSet", + "GenerationInstaller", "SubscriptionObligationPlanner", "SubscriptionPlanResult", "subscription_obligation_id", diff --git a/tests/test_reporting_subscription_service.py b/tests/test_reporting_subscription_service.py new file mode 100644 index 000000000..7b8f0dd98 --- /dev/null +++ b/tests/test_reporting_subscription_service.py @@ -0,0 +1,404 @@ +"""Reporting subscriptions served by ReliableReportingService (AdCP 3.3). + +The service plans every listed principal's frozen owed sets each turn, +executes them with the account generations, advertises the subscription +capability fields at the served release and adds ``subscription_coverage[]`` +to status reads. Without the option it is unchanged. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from typing import Any, cast + +import pytest + +from adcp.reporting.fixtures import SNAPSHOT_OFFERING_ID, redacted_capabilities +from adcp.reporting.ledger import ( + InMemoryReportingLedgerStore, + ReportingConfiguration, + ReportingStatusCaller, +) +from adcp.reporting.service import ( + ReliableReportingConfigurationError, + ReliableReportingService, + ReportingAccountContext, +) +from adcp.reporting.service_subscriptions import ReportingSubscriptionServiceOptions +from adcp.reporting.subscriptions import ( + AuthorizedAccount, + ReportingSubscriptionEngine, + ReportingSubscriptionSupport, + SubscriptionBuy, + SubscriptionOffering, +) +from adcp.reporting.testing import DeterministicReportingClock +from adcp.server import PrincipalIdentity, PrincipalService +from tests.test_reliable_reporting_service import ( + DEFINITION, + _capability_offering, + _configuration, +) +from tests.test_reporting_subscription_engine import Clock, Host + +NOW = datetime(2026, 11, 1, 3, 30, tzinfo=timezone.utc) +ACCOUNT = "account-redacted" +AGENCY = PrincipalIdentity("https://agency.example/agent", "buyer_agent") +OTHER = PrincipalIdentity("https://other.example/agent", "buyer_agent") +STRANGER = PrincipalIdentity("https://stranger.example/agent", "buyer_agent") +OFFERING_ID = "GAM_SNAPSHOT_V1" +HOURLY = SubscriptionOffering( + offering_id=OFFERING_ID, + report_definition_id="PAID_MEDIA_DAILY_V1", + reporting_profile="paid_media_delivery", + feed_purpose="analytics", + period_duration="PT1H", + alignment="utc", + delivery_sla="PT10M", + supported_finality=frozenset({"snapshot"}), + reconciliation_mode="delivery_only", + dimensions=frozenset({"package"}), +) +SUBSCRIPTION = { + "subscription_id": "analytics-hourly", + "subscription_version": 1, + "active": True, + "scope": {"all_authorized_accounts": True}, + "selector": { + "feed_purpose": "analytics", + "period_duration": "PT1H", + "required_finality": "snapshot", + "dimensions": ["package"], + }, + "reconciliation_mode": "delivery_only", +} + + +class _ByBuySource: + """Answers every slice with one row per requested media buy.""" + + def __init__(self) -> None: + self.calls: list[Any] = [] + + @property + def capabilities(self): + return redacted_capabilities() + + def fetch_slice(self, request): + self.calls.append(request) + return [ + { + "media_buy_id": item.media_buy_id, + "campaign_id": "campaign-redacted-1", + "impressions": 10, + "spend": "1.25", + } + for item in request.coverage.constituents + ] + + +def _context(configuration: ReportingConfiguration) -> ReportingAccountContext: + return ReportingAccountContext( + account_id=configuration.account_id, + adapter="gam", + currency="USD", + account_timezone="UTC", + source_scope=redacted_capabilities().source_scope, + snapshot_offering_id=SNAPSHOT_OFFERING_ID, + requested_dimensions=("campaign_id",), + capability_offering=_capability_offering("gam"), + publication_namespace="reporting-source:fixture", + ) + + +def _buy(media_buy_id: str, start: str, end: str) -> SubscriptionBuy: + def at(text: str) -> datetime: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + + return SubscriptionBuy( + media_buy_id=media_buy_id, + confirmed_at=at(start), + start_time=at(start), + end_time=at(end), + reporting_delivery_offering_ids=(OFFERING_ID,), + ) + + +class _Seller: + def __init__(self, *, principals=(AGENCY, OTHER), **options: Any) -> None: + self.clock = DeterministicReportingClock(NOW) + # Subscriptions are accepted before the buys run; the engine owes + # nothing before acceptance. + self.engine_clock = Clock("2026-11-01T00:00:00Z") + self.host = Host() + self.principals = PrincipalService() + self.engine = ReportingSubscriptionEngine( + principals=self.principals, + host=self.host, + offerings=[HOURLY], + support=ReportingSubscriptionSupport(dimensions=("package",)), + clock=self.engine_clock, + ) + self.principals.set_reporting_subscriptions(self.engine.principal_options()) + self.listed = list(principals) + self.caller = AGENCY + self.source = _ByBuySource() + self.options = ReportingSubscriptionServiceOptions( + engine=self.engine, + list_principals=lambda: self.listed, + resolve_principal=lambda _request, _context: self.caller, + definitions={OFFERING_ID: DEFINITION}, + **options, + ) + self.service = ReliableReportingService.memory( + account_context=_context, + caller_resolver=lambda _request, _context: ReportingStatusCaller( + ACCOUNT, self.caller.subject + ), + clock=self.clock, + read_jitter_window=timedelta(0), + reporting_subscriptions=self.options, + ) + self.service.sources.register("gam", self.source) + self._key = 0 + + async def subscribe(self, identity: PrincipalIdentity, buys) -> None: + self.engine_clock.set("2026-11-01T00:00:00Z") + self.host.accounts[ACCOUNT] = AuthorizedAccount( + ACCOUNT, timezone="UTC", authorized_at=datetime(2026, 10, 1, tzinfo=timezone.utc) + ) + self.host.buys[ACCOUNT] = list(buys) + self._key += 1 + response = await self.principals.sync_principal( + identity, + { + "idempotency_key": f"service-test-key-{self._key:06d}", + "configuration": {"reporting_subscriptions": [SUBSCRIPTION]}, + }, + ) + assert response.result.kind == "applied", response + await self.engine.reconcile(identity) + self.engine_clock.set("2026-11-01T03:30:00Z") + + async def status(self, identity: PrincipalIdentity, view: str = "periods", context=None): + self.caller = identity + return await self.service.get_reporting_status({"view": view}, context) + + +class _Buys: + """Each principal sees only its own buys on the shared account.""" + + def __init__(self, seller: _Seller, by_principal: dict[str, list[SubscriptionBuy]]) -> None: + self.by_principal = by_principal + seller.host.account_buys = self.account_buys # type: ignore[method-assign] + + async def account_buys(self, principal, account_id): + return list(self.by_principal.get(principal.subject, ())) + + +@pytest.mark.asyncio +async def test_a_turn_plans_and_executes_each_principals_owed_set() -> None: + seller = _Seller() + mine = [ + _buy("media-buy-a1", "2026-11-01T01:00:00Z", "2026-11-01T03:00:00Z"), + _buy("media-buy-a2", "2026-11-01T02:00:00Z", "2026-11-01T03:00:00Z"), + ] + theirs = [_buy("media-buy-b1", "2026-11-01T02:00:00Z", "2026-11-01T03:00:00Z")] + await seller.subscribe(AGENCY, mine) + await seller.subscribe(OTHER, theirs) + _Buys(seller, {AGENCY.subject: mine, OTHER.subject: theirs}) + + turn = await seller.service.run_worker(now=NOW) + + assert turn.subscription_errors == {} + assert turn.subscription_obligations_written == 3 + # Each slice carries exactly that period's frozen media buys. + requested = sorted( + (call.period.start.isoformat(), tuple(c.media_buy_id for c in call.coverage.constituents)) + for call in seller.source.calls + ) + assert requested == [ + ("2026-11-01T01:00:00+00:00", ("media-buy-a1",)), + ("2026-11-01T02:00:00+00:00", ("media-buy-a1", "media-buy-a2")), + ("2026-11-01T02:00:00+00:00", ("media-buy-b1",)), + ] + assert sum(len(item.revisions_committed) for item in turn.configurations.values()) == 3 + + # Per-principal visibility: one wire generation, its own periods only. + mine_view = await seller.status(AGENCY) + assert [p["offering_id"] for p in mine_view["periods"]] == [OFFERING_ID, OFFERING_ID] + assert {b for p in mine_view["periods"] for b in p["media_buy_ids"]} == { + "media-buy-a1", + "media-buy-a2", + } + assert mine_view["scope"]["delivery_config_generations"] == [ + { + "delivery_config_id": "analytics-hourly", + "delivery_config_version": 1, + "feed_purpose": "analytics", + } + ] + assert mine_view["subscription_coverage"][0]["subscription_id"] == "analytics-hourly" + theirs_view = await seller.status(OTHER) + assert [p["media_buy_ids"] for p in theirs_view["periods"]] == [["media-buy-b1"]] + stranger = await seller.status(STRANGER) + assert stranger["periods"] == [] + assert "subscription_coverage" not in stranger + + # A second turn writes and reads nothing new. + again = await seller.service.run_worker(now=NOW) + assert again.subscription_obligations_written == 0 + assert len(seller.source.calls) == 3 + + +@pytest.mark.asyncio +async def test_the_horizon_never_passes_the_turn_clock() -> None: + seller = _Seller(principals=(AGENCY,)) + await seller.subscribe( + AGENCY, [_buy("media-buy-a1", "2026-11-01T01:00:00Z", "2026-11-01T05:00:00Z")] + ) + turn = await seller.service.run_worker(now=NOW) + # Periods ending at 02:00 and 03:00 are frozen; 03:00-04:00 has not ended. + assert turn.subscription_obligations_written == 2 + + +@pytest.mark.asyncio +async def test_one_failing_subscription_never_stops_the_others() -> None: + failures: list[Any] = [] + seller = _Seller(planning_error_handler=failures.append) + mine = [_buy("media-buy-a1", "2026-11-01T02:00:00Z", "2026-11-01T03:00:00Z")] + theirs = [_buy("media-buy-b1", "2026-11-01T02:00:00Z", "2026-11-01T03:00:00Z")] + await seller.subscribe(AGENCY, mine) + await seller.subscribe(OTHER, theirs) + _Buys(seller, {AGENCY.subject: mine, OTHER.subject: theirs}) + original = seller.engine.frozen_owed_sets + + async def frozen_owed_sets(identity, **kwargs): + if identity == AGENCY: + raise RuntimeError("engine unavailable for this principal") + return await original(identity, **kwargs) + + seller.engine.frozen_owed_sets = frozen_owed_sets # type: ignore[method-assign] + # An account configuration on the same service keeps running too. + await seller.service.configure(_configuration()) + turn = await seller.service.run_worker(now=NOW) + + assert list(turn.subscription_errors) == [(AGENCY.subject, None, None)] + assert [failure.principal for failure in failures] == [AGENCY] + assert turn.subscription_obligations_written == 1 + assert {c.identity.delivery_config_id for c in seller.source.calls} == { + "analytics-hourly", + "gam-delivery", + } + + +def _capabilities() -> dict[str, Any]: + return { + "adcp": {"major_versions": [3], "principal": {"supported_sections": ["destinations"]}}, + "supported_protocols": ["media_buy"], + } + + +@pytest.mark.asyncio +async def test_capabilities_follow_the_served_release() -> None: + seller = _Seller() + await seller.service.configure(_configuration()) + beta_1 = seller.service.inject_capabilities(_capabilities(), adcp_version="3.3-beta.1") + block = beta_1["media_buy"]["reporting_delivery"] + assert block["configuration_tasks"] == ["sync_principal", "sync_accounts"] + assert block["configuration_task"] == "sync_accounts" + assert block["subscriptions"] == {"supported": True, "dimensions": ["package"]} + # adcp#8585 offering fields reach only 3.3.0-beta.2 callers. + assert "standard_dimensions" not in block["offerings"][0] + assert "media_buy.reporting_subscriptions" in beta_1["experimental_features"] + assert "reporting_subscriptions" in beta_1["adcp"]["principal"]["supported_sections"] + beta_2 = seller.service.inject_capabilities(_capabilities(), adcp_version="3.3-beta.2") + assert beta_2["media_buy"]["reporting_delivery"]["offerings"][0]["standard_dimensions"] == [ + "package" + ] + # A 3.2 caller sees the pre-3.3 block. + older = seller.service.inject_capabilities(_capabilities(), adcp_version="3.2") + assert "configuration_tasks" not in older["media_buy"]["reporting_delivery"] + assert "subscriptions" not in older["media_buy"]["reporting_delivery"] + assert "media_buy.reporting_subscriptions" not in older["experimental_features"] + assert older["adcp"]["principal"]["supported_sections"] == ["destinations"] + + +@pytest.mark.asyncio +async def test_coverage_is_served_from_3_3_only() -> None: + seller = _Seller(principals=(AGENCY,)) + await seller.subscribe( + AGENCY, [_buy("media-buy-a1", "2026-11-01T01:00:00Z", "2026-11-01T03:00:00Z")] + ) + await seller.service.run_worker(now=NOW) + served = await seller.status( + AGENCY, "summary", SimpleNamespace(resolved_adcp_version="3.3-beta.1") + ) + assert served["subscription_coverage"][0]["status"] == "active" + older = await seller.status(AGENCY, "summary", SimpleNamespace(resolved_adcp_version="3.2")) + assert "subscription_coverage" not in older + + +@pytest.mark.asyncio +async def test_resolve_principal_must_agree_with_the_caller() -> None: + seller = _Seller(principals=(AGENCY,)) + await seller.service.run_worker(now=NOW) + seller.service._caller_resolver = lambda _r, _c: ReportingStatusCaller( # type: ignore[assignment] + ACCOUNT, OTHER.subject + ) + with pytest.raises(ReliableReportingConfigurationError, match="disagree"): + await seller.service.get_reporting_status({"view": "summary"}) + + +@pytest.mark.asyncio +async def test_construction_refuses_what_it_cannot_serve() -> None: + seller = _Seller() + with pytest.raises(ReliableReportingConfigurationError, match="production"): + ReliableReportingService( + store=InMemoryReportingLedgerStore(), + account_context=_context, + production=cast(Any, object()), + reporting_subscriptions=seller.options, + ) + mismatched = ReliableReportingService.memory( + account_context=_context, + status_retention_days=30, + reporting_subscriptions=seller.options, + ) + with pytest.raises(ReliableReportingConfigurationError, match="status_retention_days"): + mismatched.validate() + unmapped = ReliableReportingService.memory( + account_context=_context, reporting_subscriptions=seller.options + ) + other = dict(_capability_offering("gam"), offering_id="FREEWHEEL_SNAPSHOT_V1") + unmapped.sources.register("freewheel", _ByBuySource(), capability_offerings=[other]) + with pytest.raises(ReliableReportingConfigurationError, match="dimension mapping"): + unmapped.validate() + with pytest.raises(TypeError, match="list_principals"): + ReportingSubscriptionServiceOptions( + engine=seller.engine, list_principals=None, resolve_principal=lambda r, c: AGENCY # type: ignore[arg-type] + ) + + +@pytest.mark.asyncio +async def test_without_the_option_nothing_changes() -> None: + clock = DeterministicReportingClock(NOW) + service = ReliableReportingService.memory( + account_context=_context, + caller_resolver=lambda _r, _c: ReportingStatusCaller(ACCOUNT, "buyer-1"), + clock=clock, + read_jitter_window=timedelta(0), + ) + service.sources.register("gam", _ByBuySource()) + await service.configure(_configuration()) + turn = await service.run_worker(now=NOW) + assert turn.subscription_obligations_written == 0 and turn.subscription_errors == {} + capabilities = service.inject_capabilities(_capabilities(), adcp_version="3.3-beta.2") + block = capabilities["media_buy"]["reporting_delivery"] + assert "configuration_tasks" not in block and "subscriptions" not in block + assert "standard_dimensions" not in block["offerings"][0] + assert capabilities["adcp"]["principal"]["supported_sections"] == ["destinations"] + assert "media_buy.reporting_subscriptions" not in capabilities["experimental_features"] + status = await service.get_reporting_status({"view": "summary"}) + assert "subscription_coverage" not in status