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/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/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..6c316b38f 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,61 @@ 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) + 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] = [] + 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/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", 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"(?