bokelley · Oct 31, 2026 · Oct 30, 2026 · Oct 30, 2026 · Oct 30, 2026 · Oct 30, 2026
@@ -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
 

@@ -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,

@@ -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',

@@ -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)

@@ -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 ""

@@ -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,

@@ -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),

@@ -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)
 

@@ -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",

@@ -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):