From a0e52d8f16c45f4fd4c1e9dd1e8e7cba60b8b1b0 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:08:08 +0000 Subject: [PATCH 1/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) From c6a2575075c07601b22e5c11a9547e7b68f7cc75 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 18:59:43 +0000 Subject: [PATCH 2/2] fix(reporting): make the managed seller pass the SDK's own capability validation adcp-client-python#1771: a managed reporting-subscriptions seller advertised capabilities the SDK's own client rejected, and needed undocumented wiring over MCP. The principal block now carries optimistic_concurrency true and lists reporting_destinations only when the seller supplies the destination facts the schema requires with it. supported_versions is advertised when the seller does not state it, so a buyer negotiates the 3.3 line. A reporting task with an authenticated caller but no resolved account, as a plain ToolContext over MCP has, reads the account the request names, behind the subscription grant check. The seller guide lists what the host still supplies: adcp.idempotency, the destination facts, and the context factory. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reporting-subscriptions-seller.md | 14 ++++++ src/adcp/reporting/service.py | 40 ++++++++++++++++- src/adcp/reporting/subscriptions/managed.py | 27 ++++++++++-- tests/test_reporting_subscription_managed.py | 46 ++++++++++++++++++-- 4 files changed, 119 insertions(+), 8 deletions(-) diff --git a/docs/reporting-subscriptions-seller.md b/docs/reporting-subscriptions-seller.md index b440aa2e9..20d338944 100644 --- a/docs/reporting-subscriptions-seller.md +++ b/docs/reporting-subscriptions-seller.md @@ -31,6 +31,20 @@ The SDK reads these surfaces under each principal's identity. A principal sees e A decisioning `Account` carries `timezone=` and `operator=`. An account without a timezone is refused rather than read as UTC. +### Capabilities and serving + +The service fills in what it serves and leaves the rest to you: +- **What the service adds.** + - The `media_buy.reporting_delivery` block. + - The `adcp.principal` block, with `optimistic_concurrency: true` and `reporting_subscriptions` in `supported_sections`. + - `adcp.supported_versions`, when your capabilities do not state it. Without it a buyer negotiates down from `major_versions` and never reaches 3.3. +- **What your capabilities must carry.** + - `adcp.idempotency`, which says whether you honor `idempotency_key` replay. The schema requires it, and only you know it. + - To accept `reporting_destinations`, `adcp.principal.max_reporting_destinations`, `reporting_destination_offerings` and `suspension_interval_seconds`. Without all three, that section is not advertised. +- **Serving over MCP.** Pass `serve(..., context_factory=auth_context_factory)` (or your own factory), so `list_accounts` and `get_media_buys` see the authenticated `caller_identity`. On the managed path, a reporting task with no resolved account reads the account the request names. The subscription grant check then refuses an account the caller holds no grant on, so no `caller_resolver` is needed for `get_reporting_status` over plain MCP. + +The reference seller (`python -m adcp reference-seller reporting`) is a complete example. + ## AdCP 3.3.0-beta.3 rules the engine applies - **Admission under `max_accounts`.** Candidates are ordered by `authorization.authorized_at`, then `account_id` in byte order, and an admitted account keeps its slot. A waiting candidate carries a `capacity` exception. diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index d1afcbc5f..c3143e5c4 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -600,7 +600,9 @@ def __init__( self._initialize_source_storage = initialize_source_storage self.sources = ReportingAdapterRegistry(clock=effective_clock, staging=staging, seals=seals) self._context_resolver = account_context - self._caller_resolver = caller_resolver or self._default_caller + self._caller_resolver = caller_resolver or ( + self._managed_caller if subscriptions is not None else self._default_caller + ) self._consumer_status_enabled = consumer_status_enabled self._escalation = escalation or ReportingDeliveryEscalation() self._clock = effective_clock @@ -1572,6 +1574,29 @@ async def caller_for(self, request: Any, context: Any | None) -> ReportingStatus raise TypeError("caller_resolver must resolve ReportingStatusCaller") return caller + @classmethod + def _managed_caller(cls, request: Any, context: Any | None) -> ReportingStatusCaller: + """The managed subscription path's default caller, which also works over plain MCP. + + As :meth:`_default_caller`; when the context carries an authenticated + caller but no resolved account (a plain ``ToolContext`` from + ``serve()``), the account is the one the request names. The + subscription grant check then refuses an account the caller holds no + grant on, as for any other read (adcp-client-python#1771). + """ + try: + return cls._default_caller(request, context) + except ReliableReportingConfigurationError: + from adcp.reporting.subscriptions.managed import request_principal_id + + consumer_id = request_principal_id(context) + payload = _wire(request) if request is not None else {} + account = payload.get("account") if isinstance(payload, Mapping) else None + account_id = account.get("account_id") if isinstance(account, Mapping) else None + if not consumer_id or not account_id: + raise + return ReportingStatusCaller(account_id=str(account_id), consumer_id=consumer_id) + @staticmethod def _default_caller(request: Any, context: Any | None) -> ReportingStatusCaller: del request @@ -1847,6 +1872,19 @@ def inject_capabilities( serves_subscriptions, ) + if self._managed is not None: + # Subscriptions are served from 3.3; without supported_versions + # a buyer negotiates down from major_versions and never reaches + # them (adcp-client-python#1771). Advertise the releases this + # SDK serves unless the seller states its own. + adcp = dict(payload.get("adcp") or {}) + if not adcp.get("supported_versions"): + from adcp._version import get_supported_adcp_versions + + adcp["supported_versions"] = [ + item for item in get_supported_adcp_versions() if "-rc" not in item + ] + payload["adcp"] = adcp if self._managed is not None and serves_subscriptions(adcp_version): payload = self._managed.advertise_principal(payload) try: diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index ef88eb143..ee444afa0 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -1096,16 +1096,35 @@ async def _project_page(self, principal: PrincipalRef, page: list[Any]) -> list[ @staticmethod def advertise_principal(payload: dict[str, Any]) -> dict[str, Any]: - """Advertise the principal layer the service serves, merged with the seller's.""" + """Advertise the principal layer the service serves, merged with the seller's. + + ``reporting_subscriptions`` is always added, and ``optimistic_concurrency`` + is ``true``: the principal runtime enforces + ``expected_configuration_version``. ``reporting_destinations`` is added + only when the seller's own block carries the facts the schema requires + with it (``max_reporting_destinations``, + ``reporting_destination_offerings`` and ``suspension_interval_seconds``), + so the advertisement always validates. + """ adcp = dict(payload.get("adcp") or {}) principal = dict(adcp.get("principal") or {}) principal.setdefault("supported", True) principal.setdefault("sync_task", "sync_principal") principal.setdefault("read_task", "get_principal") + principal.setdefault("optimistic_concurrency", True) sections = [str(item) for item in principal.get("supported_sections") or ()] - for section in ("reporting_destinations", "reporting_subscriptions"): - if section not in sections: - sections.append(section) + destination_facts = ( + "max_reporting_destinations", + "reporting_destination_offerings", + "suspension_interval_seconds", + ) + if all(principal.get(name) for name in destination_facts): + if "reporting_destinations" not in sections: + sections.append("reporting_destinations") + elif "reporting_destinations" in sections: + sections.remove("reporting_destinations") + if "reporting_subscriptions" not in sections: + sections.append("reporting_subscriptions") principal["supported_sections"] = sections adcp["principal"] = principal payload = {**payload, "adcp": adcp} diff --git a/tests/test_reporting_subscription_managed.py b/tests/test_reporting_subscription_managed.py index 6a71e2e69..249d33fad 100644 --- a/tests/test_reporting_subscription_managed.py +++ b/tests/test_reporting_subscription_managed.py @@ -183,9 +183,9 @@ async def test_capabilities_advertise_the_principal_layer_the_service_serves() - capabilities = await world.platform.get_adcp_capabilities({}, _ctx()) principal = capabilities["adcp"]["principal"] assert principal["sync_task"] == "sync_principal" and principal["read_task"] == "get_principal" - assert {"reporting_destinations", "reporting_subscriptions"} <= set( - principal["supported_sections"] - ) + # reporting_destinations needs seller facts this fixture does not advertise. + assert principal["supported_sections"] == ["reporting_subscriptions"] + assert principal["optimistic_concurrency"] is True assert {"protocol.principal", "media_buy.reporting_subscriptions"} <= set( capabilities["experimental_features"] ) @@ -572,3 +572,43 @@ async def hung(params: Any, context: Any = None) -> Any: [failure] = world.failures assert failure.error.code == "SUBSCRIPTION_ROSTER_UNAVAILABLE" assert "RuntimeError" in str(failure.error) + + +@pytest.mark.asyncio +async def test_issue_1771_managed_capabilities_pass_the_sdk_response_validation( + monkeypatch, +) -> None: + from adcp.validation.schema_loader import get_named_validator + + base = Seller.get_adcp_capabilities + + async def capabilities(self: Any, params: Any, context: Any = None) -> Any: + payload = await base(self, params, context) + # The seller's own fact: whether it honors idempotency_key replay. + payload["adcp"]["idempotency"] = {"supported": True, "replay_ttl_seconds": 86400} + return payload + + monkeypatch.setattr(Seller, "get_adcp_capabilities", capabilities) + world = _World() + served = await world.platform.get_adcp_capabilities({"adcp_version": "3.3-beta.3"}, _ctx()) + validator = get_named_validator("protocol/get-adcp-capabilities-response.json") + payload = {"status": "completed", **served} + errors = [f"{list(e.absolute_path)} {e.message}" for e in validator.iter_errors(payload)] + assert errors == [] + # A buyer negotiates the 3.3 line instead of falling back to 3.0. + assert "3.3-beta.3" in served["adcp"]["supported_versions"] + + +@pytest.mark.asyncio +async def test_issue_1771_status_is_served_to_a_plain_mcp_caller() -> None: + from adcp.server.base import ToolContext + + world = _World() + await world.subscribe() + world.clock.set(START + timedelta(hours=3, minutes=30)) + await world.service.run_worker() + status = await world.platform.get_reporting_status( + {"account": {"account_id": ACCOUNT}, "view": "summary"}, + ToolContext(caller_identity=AGENCY), + ) + assert status["subscription_coverage"][0]["subscription_id"] == "analytics-hourly"