From f1479f5d79e7c5813c7dc207804ef3878b999e5c Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 16:32:54 +0000 Subject: [PATCH 1/2] fix(reporting): read each package offering snapshot from get_media_buys Without a buy_offerings hook, the managed path treated an empty packages[].reporting_delivery_offering_ids snapshot as absent and reported the package by every registered offering. In 3.3.0-beta.3 a seller serving subscriptions includes the snapshot on every package, fixed at confirmation, and [] means no offering applies. The snapshot is now honoured as served, including [], and only a package served without one falls back to every registered offering, matching the TypeScript SDK. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 2 +- docs/reporting-subscriptions-seller.md | 2 +- src/adcp/reporting/subscriptions/managed.py | 30 +++++++++++++----- tests/test_reporting_subscription_managed.py | 33 ++++++++++++++++++++ 4 files changed, 58 insertions(+), 9 deletions(-) diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index e0b7f8d80..90038b8d9 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -345,7 +345,7 @@ platform = reporting.install(MySeller()) # then serve the platform as usual | The principal runtime, `get_principal`/`sync_principal` and the principal section of capabilities | the service's own `PrincipalService` | `principals`, `principal_store` | | The accounts a principal may read | the platform's `list_accounts` (a decisioning platform's `accounts.list`), read under the principal's identity | `accounts(principal, *, cursor)` | | Each account's buys | the platform's `get_media_buys`, under the same identity | `buys(principal, account_id)` | -| Each package's offerings | every registered offering | `buy_offerings(media_buy)` | +| Each package's offerings | the package's `reporting_delivery_offering_ids` snapshot from `get_media_buys` (`[]` means none), else every registered offering | `buy_offerings(media_buy)` | | Offerings | the registered `capability_offerings`, each publishing `standard_dimensions` | none | | Report content per generation | `account_context` | `configure(generation)` | | The authenticated principal | the caller the reporting tasks resolve, with `ToolContext.tenant_id` | `resolve_principal(request, context)` | diff --git a/docs/reporting-subscriptions-seller.md b/docs/reporting-subscriptions-seller.md index b440aa2e9..f18881af0 100644 --- a/docs/reporting-subscriptions-seller.md +++ b/docs/reporting-subscriptions-seller.md @@ -25,7 +25,7 @@ The SDK reads these surfaces under each principal's identity. A principal sees e |---|---|---|---| | Authorized accounts | `list_accounts` (a decisioning platform's `accounts.list`) | `account_id`, `timezone` (IANA), `operator`, `authorization.allowed_tasks` | `SubscriptionOptions.accounts` | | Visible buys | `get_media_buys` | `confirmed_at`, `start_time`, `end_time`, `cancellation.canceled_at`, `cancellation_effective_at`, `completed_at`, `flight_history`, `packages[].package_id` | `SubscriptionOptions.buys` | -| Package offerings | `packages[].reporting_delivery_offering_ids` (fixed at confirmation), else every registered offering | | `SubscriptionOptions.buy_offerings` | +| Package offerings | `packages[].reporting_delivery_offering_ids` (fixed at confirmation; `[]` means none), else, only when the seller omits it, every registered offering | | `SubscriptionOptions.buy_offerings` | | Offerings | `capability_offerings` on each registered source | `standard_dimensions` on every offering; `available_from` / `withdrawn_at` when they apply; `account_id` in `reporting_profile.primary_keys` on every offering delivered to destinations (`method`), required by 3.3.0-beta.3 and logged as `reporting.subscription.offering_primary_keys` when missing | none | | Report content | `account_context`, called with `configuration.subscription` set | | `SubscriptionOptions.configure` | diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index ef88eb143..0f6ed3934 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -104,6 +104,19 @@ _MAX_PAGES = 10_000 +def _package_offering_ids(package: Mapping[str, Any], every: Sequence[str]) -> list[str]: + """A package's ``reporting_delivery_offering_ids`` snapshot (3.3.0-beta.3). + + The seller fixes it when the package is confirmed and serves ``[]`` when + no offering applies, so an empty snapshot is honoured. Only a package + served without one falls back to every registered offering. + """ + snapshot = package.get("reporting_delivery_offering_ids") + if isinstance(snapshot, (list, tuple)): + return [str(item) for item in snapshot] + return list(every) + + @dataclass(frozen=True) class SubscriptionOptions: """Serve ``media_buy.reporting_subscriptions`` from ``ReliableReportingService``. @@ -121,8 +134,11 @@ class SubscriptionOptions: * ``buys(principal, account_id)``: an account's buys, instead of ``get_media_buys``. * ``buy_offerings(media_buy)``: each package's offering ids, keyed by - ``package_id``. Without it every package is reported by every offering - the service registered; supply it when products differ. + ``package_id``. Without it each package's own + ``reporting_delivery_offering_ids`` snapshot from ``get_media_buys`` is + used (3.3.0-beta.3, fixed at confirmation; ``[]`` means no offering), and + only a package without one is reported by every offering the service + registered. * ``configure(generation)``: the report content for a subscription generation, instead of the service's ``account_context``. * ``account_delivery_configs(principal, account_id)``: the caller's own @@ -986,8 +1002,10 @@ def _buy(self, raw: Mapping[str, Any]) -> SubscriptionBuy | None: The flight, ``cancellation_effective_at`` and ``flight_history`` come from the wire. Each package's offering set is ``buy_offerings`` when - given, else the package's ``reporting_delivery_offering_ids`` (fixed at - confirmation), else every registered offering. + given, else the package's ``reporting_delivery_offering_ids`` snapshot + (fixed at confirmation; an empty one means no offering reports the + package), else, only when the seller omits the snapshot, every + registered offering. """ if raw.get("start_time") is None or raw.get("end_time") is None: return None # not yet flighted: nothing can be owed for it @@ -1003,9 +1021,7 @@ def _buy(self, raw: Mapping[str, Any]) -> SubscriptionBuy | None: media_buy["packages"] = [ { **package, - "reporting_delivery_offering_ids": list( - package.get("reporting_delivery_offering_ids") or every - ), + "reporting_delivery_offering_ids": _package_offering_ids(package, every), } for package in raw.get("packages") or () if package.get("package_id") diff --git a/tests/test_reporting_subscription_managed.py b/tests/test_reporting_subscription_managed.py index 6a71e2e69..8cfa11f5f 100644 --- a/tests/test_reporting_subscription_managed.py +++ b/tests/test_reporting_subscription_managed.py @@ -322,6 +322,39 @@ async def test_derived_buys_use_buy_offerings_per_package() -> None: assert turn.subscription_obligations_written == 0 +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("snapshot", "owed"), + [ + # The snapshot names the registered offering: the package is reported. + ([OFFERING["offering_id"]], True), + # 3.3.0-beta.3: [] means no offering applies, never "every offering". + ([], False), + # The snapshot fixed at confirmation names another offering. + (["SOME_OTHER_OFFERING"], False), + ], +) +async def test_derived_buys_read_each_package_offering_snapshot( + snapshot: list[str], owed: bool +) -> None: + world = _World() + world.seller.buys[0]["packages"][0]["reporting_delivery_offering_ids"] = snapshot + await world.subscribe() + world.clock.set(START + timedelta(hours=3, minutes=30)) + turn = await world.service.run_worker() + assert (turn.subscription_obligations_written > 0) is owed + + +@pytest.mark.asyncio +async def test_a_package_without_a_snapshot_falls_back_to_every_registered_offering() -> None: + world = _World() + assert "reporting_delivery_offering_ids" not in world.seller.buys[0]["packages"][0] + await world.subscribe() + world.clock.set(START + timedelta(hours=3, minutes=30)) + turn = await world.service.run_worker() + assert turn.subscription_obligations_written > 0 + + @pytest.mark.asyncio async def test_replace_account_configs_is_refused_without_a_host_that_applies_it() -> None: world = _World() From 41a0a5c862ff188310c1b378dc014f271183228e Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:08:08 +0000 Subject: [PATCH 2/2] test(reporting): run the row-upload lease test at the wall clock The producer row-upload test stamps its worker lease with the settling harness clock, fixed at 2026-11-01T02:10Z, while the row store checks the lease against the database's clock_timestamp. Once the wall clock passed that instant plus the lease, every run failed with STATE_UNAVAILABLE: no live worker lease covers this row upload, on next and on every pull request. The test now runs its turn at the wall clock, so the lease lies in the real future. Co-Authored-By: Claude Opus 5.5 (1M context) (cherry picked from commit ae15bab6e8a73f974ff6adc30cf761454b089b44) --- .../reporting/test_reporting_row_storage_object_pg.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/conformance/reporting/test_reporting_row_storage_object_pg.py b/tests/conformance/reporting/test_reporting_row_storage_object_pg.py index aa2410045..f4cab4017 100644 --- a/tests/conformance/reporting/test_reporting_row_storage_object_pg.py +++ b/tests/conformance/reporting/test_reporting_row_storage_object_pg.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -442,13 +443,18 @@ async def test_the_producer_stages_object_uploads_outside_its_publication_transa from tests.test_reporting_settling import _capabilities, _harness, _revisions async with isolated_reporting_pool() as pool: - producer, store, _, _ = await _harness( + producer, store, _, clock = await _harness( _capabilities(restatement_window="P3D"), store_factory=lambda clock: PgReportingLedgerStore( pool=pool, clock=clock, row_storage=options(tmp_path) ), ) assert isinstance(producer, ReportingProducer) + # The row upload checks the worker lease against the database's + # clock_timestamp(), so the lease the injected clock stamps must lie in + # the real future: run the turn at the wall clock, not the harness's + # fixed instant, which is already in the past. + clock[0] = datetime.now(timezone.utc) turn = await producer.run_worker() assert turn.revisions_committed (revision,) = await _revisions(store)