feat(reporting): make the grant journal the one source of authorized_at

docs/reliable-reporting-service.md

@@ -379,6 +379,24 @@ platform = reporting.install(MySeller())  # then serve the platform as usual
   platform that cannot list accounts (`SUBSCRIPTION_ROSTER_UNAVAILABLE`). Both
   are `ReliableReportingConfigurationError` with `code` and
   `recovery="correctable"`, and each message ends with the fix.
+- **Grants and `authorized_at`.** The SDK journals every grant it learns of and
+  is the only source of `authorization.authorized_at`. It applies the chain
+  rule: a re-grant before `unauthorized_at + status_retention_days` continues
+  the chain, so `authorized_at` does not move; a later re-grant starts a new
+  one. Served `list_accounts` and `sync_accounts` responses carry the
+  journal's value, the same one coverage uses. A host value is adopted on
+  first observation; a later one that moves backwards is rejected and logged
+  as `authorized_at_regressed`. Those responses also update the journal, as
+  does `await reporting.grants.changed(principal, account_id, at=...,
+  authorized=..., allowed_tasks=...)` for sellers whose entitlement service
+  emits events. Grant checks read the journal, so a reduction shown to the
+  caller is enforced on its next reporting task.
+- **Wind down, don't revoke.** Revocation means no access at all, receipts and
+  statuses included. To end a relationship while the buyer finishes
+  reconciling, set `account.authorization = wind_down_grant(account.authorization)`
+  (from `adcp.server`). It lists only `REPORTING_WIND_DOWN_TASKS`
+  (`get_reporting_status`, `sync_reporting_receipts`, `sync_reporting_status`)
+  and keeps `authorized_at`, so coverage continues.
 - **Principals.** Reporting APIs name a principal with
   `PrincipalRef(tenant_id, principal_id)`; `grants_task` and
   `authorizes_delivery` take it. One service serves one tenant.

src/adcp/decisioning/handler.py

@@ -3314,6 +3314,9 @@ async def sync_accounts(  # type: ignore[override]
         if inspect.isawaitable(result):
             result = await result
         projected = _project_sync_accounts(result)
+        reporting = getattr(self._platform, "_reliable_reporting_service", None)
+        if reporting is not None:
+            projected = await reporting._project_accounts(projected, tool_ctx)
         await self._reporting_subscription_write(None, tool_ctx)
         return cast("SyncAccountsResponse", projected)
 
@@ -3354,9 +3357,13 @@ async def list_accounts(  # type: ignore[override]
             from adcp.reporting.outbox.activity import omit_account_activity
 
             projected = omit_account_activity(projected)
-        return cast(
-            "ListAccountsResponse", strip_credentials_from_wire_result("list_accounts", projected)
-        )
+        served = strip_credentials_from_wire_result("list_accounts", projected)
+        reporting = getattr(self._platform, "_reliable_reporting_service", None)
+        if reporting is not None:
+            # Managed reporting subscriptions serve the grant journal's
+            # authorized_at, the value coverage uses.
+            served = await reporting._project_accounts(served, tool_ctx)
+        return cast("ListAccountsResponse", served)
 
     # ----- Optional-method gate -----
 

src/adcp/reporting/service.py

@@ -778,6 +778,20 @@ async def touch(self, *, account_id: str | None = None, context: Any | None = No
         managed = self._require_managed()
         await managed.touch(account_id=account_id, context=context)
 
+    @property
+    def grants(self) -> Any:
+        """Push grant changes into the grant journal (``await reporting.grants.changed(...)``).
+
+        See :class:`~adcp.reporting.subscriptions.managed.SubscriptionGrants`.
+        """
+        return self._require_managed().grants
+
+    async def _project_accounts(self, response: Any, context: Any | None) -> Any:
+        """Serve the grant journal's ``authorized_at`` on an account response."""
+        if self._managed is None:
+            return response
+        return await self._managed.project_accounts(response, context)
+
     def _require_managed(self) -> ManagedSubscriptions:
         if self._managed is None:
             raise ReliableReportingConfigurationError(
@@ -1986,7 +2000,12 @@ def triggered(name: str) -> Callable[..., Any] | None:
 
             async def write(instance: Any, params: Any, context: Any | None = None) -> Any:
                 result = await _resolve(method(instance, params, context))
-                await service._subscription_write(_written_account(params, context), context)
+                if name in {"sync_accounts", "list_accounts"}:
+                    result = await service._project_accounts(result, context)
+                if name != "list_accounts":
+                    await service._subscription_write(
+                        _written_account(params, context), context
+                    )
                 return result
 
             write.__name__ = name
@@ -1997,7 +2016,7 @@ async def write(instance: Any, params: Any, context: Any | None = None) -> Any:
             "get_principal": get_principal,
             "sync_principal": sync_principal,
         }
-        for name in ("sync_accounts", "create_media_buy", "update_media_buy"):
+        for name in ("list_accounts", "sync_accounts", "create_media_buy", "update_media_buy"):
             wrapped = triggered(name)
             if wrapped is not None:
                 methods[name] = wrapped

src/adcp/reporting/subscriptions/__init__.py

@@ -54,6 +54,7 @@
     SubscriptionBuyPackage,
     SubscriptionException,
     SubscriptionVersionWindow,
+    authorization_chain_start,
     expected_subscription_periods,
     subscription_covers_account,
     subscription_exceptions,
@@ -113,8 +114,12 @@
     project_subscription_state,
     serves_settled_subscription_fields,
 )
+from adcp.server.reporting_grants import REPORTING_WIND_DOWN_TASKS, wind_down_grant
 
 __all__ = [
+    "REPORTING_WIND_DOWN_TASKS",
+    "authorization_chain_start",
+    "wind_down_grant",
     "COVERAGE_REASON_CODES",
     "COVERAGE_RULE_ORDER",
     "COVERAGE_VOCABULARY_VERSION",

src/adcp/reporting/subscriptions/coverage.py

@@ -575,6 +575,22 @@ class _Owed:
     media_buy_ids: set[str] = field(default_factory=set)
 
 
+def authorization_chain_start(
+    authorizations: Sequence[SubscriptionAuthorization], *, status_retention_days: int
+) -> datetime | None:
+    """The ``authorization.authorized_at`` a seller serves for a grant history.
+
+    The start of the current authority chain: a re-grant before
+    ``unauthorized_at + status_retention_days`` continues the chain, so the
+    value never moves on such a re-grant; a later re-grant starts a new chain.
+    ``None`` while the caller holds no grant. The same rule coverage applies.
+    """
+    if not authorizations or authorizations[-1].unauthorized_at is not None:
+        return None
+    chains = _chains(authorizations, timedelta(days=status_retention_days))
+    return chains[-1].start
+
+
 def _chains(
     authorizations: Sequence[SubscriptionAuthorization], retention: timedelta | None
 ) -> list[_Chain]:
@@ -1037,6 +1053,7 @@ def _subscription_id(subscription: Any, windows: Sequence[SubscriptionVersionWin
     "SubscriptionBuyPackage",
     "SubscriptionException",
     "SubscriptionVersionWindow",
+    "authorization_chain_start",
     "expected_subscription_periods",
     "subscription_covers_account",
     "subscription_exceptions",