fix(reporting): make the managed seller pass the SDK's own capability validation

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.

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:

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}

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)

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"