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/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) 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"