From 8ea96e21f1287cb980fcafa7862869e62d9ff154 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 2 Oct 2026 17:49:06 +0000 Subject: [PATCH 1/4] fix(reporting): translate installed service errors by recovery policy --- docs/reliable-reporting-service.md | 11 +++ src/adcp/reporting/_errors.py | 86 ++++++++++++++++++ src/adcp/reporting/production/handler.py | 28 +++++- src/adcp/reporting/service.py | 41 +++++++-- ...st_reporting_production_service_factory.py | 34 ++++++- tests/test_reliable_reporting_lifecycle.py | 6 +- tests/test_reporting_installed_errors.py | 91 +++++++++++++++++++ .../reliable_reporting_lifecycle.py | 9 ++ 8 files changed, 289 insertions(+), 17 deletions(-) create mode 100644 src/adcp/reporting/_errors.py create mode 100644 tests/test_reporting_installed_errors.py diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 41c2452c5..5a1873e3e 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -338,3 +338,14 @@ exact-read assertions after service recreation or durable recovery. See [`examples/reliable_reporting_adapters.py`](../examples/reliable_reporting_adapters.py) for compact GAM-like and FreeWheel-like adapter definitions. + +Installed Core and managed production reporting use one error policy at service +admission. A created, stopping, closed, or failed service returns +`SERVICE_UNAVAILABLE` with transient recovery. A missing authenticated caller +returns `AUTH_REQUIRED`; explicit buyer configuration validation can raise +`ReliableReportingConfigurationError(message, kind="invalid_request")`. +Composition errors retain `INTERNAL_ERROR` and terminal recovery, with internal +configuration details omitted from buyer responses. Ledger codes are preserved +and receive the SDK's recovery classification for that code (unknown extension +codes default to terminal). Disabled optional methods and aggregate delivery +continue to delegate to the application. diff --git a/src/adcp/reporting/_errors.py b/src/adcp/reporting/_errors.py new file mode 100644 index 000000000..b1ac5c830 --- /dev/null +++ b/src/adcp/reporting/_errors.py @@ -0,0 +1,86 @@ +"""One buyer error policy for installed Core and managed reporting routes.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from typing import Literal, TypeVar + +from adcp.exceptions import ADCPTaskError +from adcp.reporting.ledger.store import LedgerConflictError +from adcp.reporting.service_lifecycle import ReliableReportingUnavailableError +from adcp.server.helpers import adcp_error +from adcp.types import Error + +_T = TypeVar("_T") + + +class ReliableReportingConfigurationError(ValueError): + """A composition failure, missing caller, or invalid buyer configuration. + + Composition failures default to terminal INTERNAL_ERROR. Only explicit + request-time failures may ask the buyer to correct authentication/input. + """ + + def __init__( + self, + message: str, + *, + kind: Literal["misconfiguration", "auth_required", "invalid_request"] = "misconfiguration", + ) -> None: + super().__init__(message) + self.kind = kind + + +def reporting_error(code: str, message: str) -> Error: + """Preserve the domain code and use the SDK's normative recovery defaults.""" + return Error.model_validate(adcp_error(code, message)["errors"][0]) + + +async def installed_reporting_call( + task: str, call: Callable[[], Awaitable[_T]], *, decisioning: bool = False +) -> _T: + """Translate at admission, including lifecycle refusals before handler entry.""" + try: + return await call() + except ReliableReportingUnavailableError as error: + wire = reporting_error( + "SERVICE_UNAVAILABLE", "Reporting service is temporarily unavailable" + ) + cause: Exception = error + except LedgerConflictError as error: + wire = reporting_error(error.code, str(error)) + cause = error + except ReliableReportingConfigurationError as error: + code, message = { + "misconfiguration": ("INTERNAL_ERROR", "Reporting service configuration failed"), + "auth_required": ("AUTH_REQUIRED", "Authenticated reporting caller is required"), + "invalid_request": ("INVALID_REQUEST", str(error)), + }[error.kind] + wire = reporting_error(code, message) + cause = error + except ADCPTaskError as error: + # Existing production routes already redact ACL/storage errors. Apply + # the same code policy without dropping their structured metadata. + errors = [ + ( + Error.model_validate( + { + **item.model_dump(exclude_none=True), + "recovery": reporting_error(item.code, item.message).recovery, + } + ) + if isinstance(item, Error) + else item + ) + for item in error.errors + ] + if not decisioning: + raise ADCPTaskError(operation=task, errors=errors) from error + wire = errors[0] + cause = error + if decisioning: + from adcp.decisioning import AdcpError + + assert wire.recovery is not None + raise AdcpError(wire.code, message=wire.message, recovery=wire.recovery.value) from cause + raise ADCPTaskError(operation=task, errors=[wire]) from cause diff --git a/src/adcp/reporting/production/handler.py b/src/adcp/reporting/production/handler.py index 6effc9bd2..808d15016 100644 --- a/src/adcp/reporting/production/handler.py +++ b/src/adcp/reporting/production/handler.py @@ -10,6 +10,11 @@ from adcp.decisioning.context import RequestContext from adcp.exceptions import ADCPTaskError +from adcp.reporting._errors import ( + ReliableReportingConfigurationError, + installed_reporting_call, + reporting_error, +) from adcp.reporting.canonical_json import canonical_json_utf8_v1 from adcp.reporting.ledger.consumer_status import ConsumerStatusIngest from adcp.reporting.ledger.delivery_models import ReportingDeliveryPrincipal @@ -21,11 +26,11 @@ ReportingReceiptHandler, _consumer, ) +from adcp.reporting.service_lifecycle import ReliableReportingUnavailableError from adcp.server.base import ADCPHandler, NotImplementedResponse, ToolContext from adcp.server.mcp_tools import ADCP_TOOL_DEFINITIONS, get_tools_for_handler from adcp.server.responses import capabilities_response from adcp.types import ( - Error, GetAdcpCapabilitiesRequest, GetMediaBuyDeliveryRequest, GetReportingStatusRequest, @@ -48,7 +53,7 @@ def _request(value: Any) -> dict[str, Any]: def _task_error(task: str, code: str, message: str) -> ADCPTaskError: - return ADCPTaskError(operation=task, errors=[Error(code=code, message=message)]) + return ADCPTaskError(operation=task, errors=[reporting_error(code, message)]) _Method = TypeVar("_Method", bound=Callable[..., Any]) @@ -76,9 +81,16 @@ async def call(self: ReportingProductionHandler, *args: Any, **kwargs: Any) -> A method.__name__ == "get_media_buy_delivery" and "reporting_revision_id" not in _request(args[0] if args else kwargs["params"]) ) - if lifecycle is None or aggregate: + if aggregate: return await method(self, *args, **kwargs) - return await lifecycle.call(lambda: method(self, *args, **kwargs)) + return await installed_reporting_call( + method.__name__, + lambda: ( + method(self, *args, **kwargs) + if lifecycle is None + else lifecycle.call(lambda: method(self, *args, **kwargs)) + ), + ) return cast(_Method, call) @@ -184,6 +196,8 @@ async def _authorize( if isinstance(context, RequestContext) and context.account.id != account: raise ReportingReceiptError("UNAUTHORIZED") return ReportingDeliveryPrincipal(account, consumer) + except (ReliableReportingConfigurationError, ReliableReportingUnavailableError): + raise except Exception: raise ReportingReceiptError("UNAUTHORIZED") from None @@ -257,6 +271,8 @@ async def sync_accounts( code, message = error.code, str(error) except ReportingReceiptError as error: code, message = error.code, str(error) + except (ReliableReportingConfigurationError, ReliableReportingUnavailableError): + raise except Exception: code, message = ( "REPORTING_CONFIGURATION_UNAVAILABLE", @@ -285,6 +301,8 @@ async def sync_reporting_status( return result except (ReportingReceiptError, LedgerConflictError) as error: code, message = error.code, str(error) + except (ReliableReportingConfigurationError, ReliableReportingUnavailableError): + raise except Exception: code, message = "REPORTING_STATUS_UNAVAILABLE", "reporting status is unavailable" raise _task_error("sync_reporting_status", code, message) @@ -412,6 +430,8 @@ async def get_media_buy_delivery( raise except (ReportingReceiptError, LedgerConflictError) as error: code, message = error.code, str(error) + except (ReliableReportingConfigurationError, ReliableReportingUnavailableError): + raise except Exception: code, message = "REPORTING_CONTENT_UNAVAILABLE", "reporting content is unavailable" raise _task_error("get_media_buy_delivery", code, message) diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index 89296d5ba..15062f45e 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -22,6 +22,10 @@ from pydantic import BaseModel +from adcp.reporting._errors import ( + ReliableReportingConfigurationError, + installed_reporting_call, +) from adcp.reporting._source_authorization import source_turn from adcp.reporting.inline_source import ( InlineFetchResult, @@ -92,10 +96,6 @@ logger = logging.getLogger(__name__) -class ReliableReportingConfigurationError(ValueError): - """The requested service composition cannot make truthful guarantees.""" - - @runtime_checkable class ReportingAdapter(Protocol): """The minimum source adapter: declaration plus one normalized slice read. @@ -1043,7 +1043,8 @@ def _default_caller(request: Any, context: Any | None) -> ReportingStatusCaller: "default caller resolution requires AccountAwareToolContext.account_id and " "authenticated caller_identity, or RequestContext.account.id and an " "authenticated current_principal or verified auth_principal; " - "provide caller_resolver for another trust model" + "provide caller_resolver for another trust model", + kind="auth_required" if not consumer_id else "misconfiguration", ) return ReportingStatusCaller(account_id=account_id, consumer_id=consumer_id) @@ -1268,26 +1269,42 @@ def install(self, platform: Any) -> Any: class ReportingInstallMixin: async def get_reporting_status(self, params: Any, context: Any | None = None) -> Any: - return await service.get_reporting_status(params, context) + return await installed_reporting_call( + "get_reporting_status", + lambda: service.get_reporting_status(params, context), + decisioning=decisioning, + ) async def sync_reporting_status(self, params: Any, context: Any | None = None) -> Any: if not service._consumer_status_enabled: return await _resolve( getattr(super(), "sync_reporting_status")(params, context) ) - return await service.sync_reporting_status(params, context) + return await installed_reporting_call( + "sync_reporting_status", + lambda: service.sync_reporting_status(params, context), + decisioning=decisioning, + ) async def sync_reporting_receipts(self, params: Any, context: Any | None = None) -> Any: if service._receipt_handler is None: return await _resolve( getattr(super(), "sync_reporting_receipts")(params, context) ) - return await service.sync_reporting_receipts(params, context) + return await installed_reporting_call( + "sync_reporting_receipts", + lambda: service.sync_reporting_receipts(params, context), + decisioning=decisioning, + ) async def _get_reporting_revision_content( self, params: Any, context: Any | None = None ) -> Any: - return await service.get_revision_content(params, context) + return await installed_reporting_call( + "get_media_buy_delivery", + lambda: service.get_revision_content(params, context), + decisioning=decisioning, + ) attrs: dict[str, Any] = {"__module__": original.__module__} if not decisioning: @@ -1305,7 +1322,11 @@ async def get_media_buy_delivery( return await _resolve( original.get_media_buy_delivery(instance, params, context) ) - return await service.get_revision_content(params, context) + return await installed_reporting_call( + "get_media_buy_delivery", + lambda: service.get_revision_content(params, context), + decisioning=decisioning, + ) attrs.update( advertised_tools=tools, diff --git a/tests/conformance/reporting/test_reporting_production_service_factory.py b/tests/conformance/reporting/test_reporting_production_service_factory.py index ffc916dc6..543ab1aa0 100644 --- a/tests/conformance/reporting/test_reporting_production_service_factory.py +++ b/tests/conformance/reporting/test_reporting_production_service_factory.py @@ -135,7 +135,7 @@ async def get_adcp_capabilities(self, params, context=None): @asynccontextmanager -async def factory_harness(backend, path, *, push=False, existing_pool=None): +async def factory_harness(backend, path, *, push=False, existing_pool=None, start=True): async with AsyncExitStack() as stack: pool = existing_pool if backend == "postgres" and pool is None: @@ -303,7 +303,8 @@ async def authorize(account, context, consumer): h.production = h.service._production h.store = h.service.store h.mount = create_mcp_server(handler) - await h.service.start() + if start: + await h.service.start() try: yield h finally: @@ -408,3 +409,32 @@ async def test_factory_revocation_discards_blocked_fetch_and_restoration_retries restored = await source_turn(h.production) assert len(restored.revisions_committed) == 1 assert len(h.source.requests) == 2 + + +@pytest.mark.parametrize("backend", ["memory", "postgres"]) +async def test_mounted_production_lifecycle_refusals_are_transient(backend, tmp_path): + from adcp.exceptions import ADCPTaskError + + async with factory_harness(backend, tmp_path / "boundary.sqlite", start=False) as h: + handler = h.production.handler + assert handler is not None + for task in ( + "get_reporting_status", + "sync_reporting_status", + "sync_reporting_receipts", + "get_media_buy_delivery", + "sync_accounts", + ): + request = ( + {"reporting_revision_id": "unavailable"} if task == "get_media_buy_delivery" else {} + ) + with pytest.raises(ADCPTaskError) as caught: + await getattr(handler, task)(request) + assert caught.value.error_codes == ["SERVICE_UNAVAILABLE"] + assert caught.value.error_info[0].recovery == "transient" + await h.service.start() + await h.service.close() + with pytest.raises(ADCPTaskError) as caught: + await handler.get_reporting_status({}) + assert caught.value.error_codes == ["SERVICE_UNAVAILABLE"] + assert caught.value.is_retryable diff --git a/tests/test_reliable_reporting_lifecycle.py b/tests/test_reliable_reporting_lifecycle.py index 4674766f0..b56a6b786 100644 --- a/tests/test_reliable_reporting_lifecycle.py +++ b/tests/test_reliable_reporting_lifecycle.py @@ -484,8 +484,12 @@ async def close_owned() -> None: with pytest.raises(ReliableReportingShutdownTimeoutError): await service.close(timeout=0) assert order == [] - with pytest.raises(ReliableReportingUnavailableError): + from adcp.exceptions import ADCPTaskError + + with pytest.raises(ADCPTaskError) as caught: await handler.sync_reporting_receipts({}) + assert caught.value.error_codes == ["SERVICE_UNAVAILABLE"] + assert caught.value.is_retryable finally: release.set() result = await receipt diff --git a/tests/test_reporting_installed_errors.py b/tests/test_reporting_installed_errors.py new file mode 100644 index 000000000..cf5ab3071 --- /dev/null +++ b/tests/test_reporting_installed_errors.py @@ -0,0 +1,91 @@ +"""Installed reporting preserves buyer recovery semantics at service admission.""" + +from unittest.mock import AsyncMock + +import pytest + +from adcp.decisioning import AdcpError +from adcp.exceptions import ADCPTaskError +from adcp.reporting import ReliableReportingConfigurationError, ReliableReportingService +from adcp.reporting.ledger import LedgerConflictError +from adcp.reporting.service_lifecycle import ( + ReliableReportingState, + ReliableReportingUnavailableError, +) +from adcp.server.base import ADCPHandler +from adcp.server.helpers import STANDARD_ERROR_CODES +from tests.test_reliable_reporting_decisioning import _SalesPlatform +from tests.test_reliable_reporting_service import _account_context + + +@pytest.mark.parametrize("decisioning", [False, True]) +@pytest.mark.parametrize( + "method,service_method", + [ + ("get_reporting_status", "get_reporting_status"), + ("sync_reporting_status", "sync_reporting_status"), + ("sync_reporting_receipts", "sync_reporting_receipts"), + ("_get_reporting_revision_content", "get_revision_content"), + ], +) +@pytest.mark.parametrize( + "failure,code", + [ + (ReliableReportingUnavailableError(ReliableReportingState.NEW), "SERVICE_UNAVAILABLE"), + (ReliableReportingConfigurationError("internal provider detail"), "INTERNAL_ERROR"), + ( + ReliableReportingConfigurationError("missing caller", kind="auth_required"), + "AUTH_REQUIRED", + ), + ( + ReliableReportingConfigurationError("invalid scope", kind="invalid_request"), + "INVALID_REQUEST", + ), + (LedgerConflictError("RATE_LIMITED", "slow down"), "RATE_LIMITED"), + (LedgerConflictError("UNAUTHORIZED", "denied"), "UNAUTHORIZED"), + (LedgerConflictError("INVALID_REQUEST", "bad input"), "INVALID_REQUEST"), + (LedgerConflictError("UNKNOWN_LEDGER_CODE", "domain detail"), "UNKNOWN_LEDGER_CODE"), + ], +) +async def test_all_installed_core_routes_translate_by_actual_code( + decisioning, method, service_method, failure, code +): + service = ReliableReportingService.memory( + account_context=_account_context, + consumer_status_enabled=True, + receipt_handler=AsyncMock(), + ) + setattr(service, service_method, AsyncMock(side_effect=failure)) + platform = service.install(_SalesPlatform() if decisioning else ADCPHandler()) + with pytest.raises(AdcpError if decisioning else ADCPTaskError) as caught: + await getattr(platform, method)({}) + error = caught.value + if decisioning: + assert error.code == code + recovery = error.recovery + else: + assert error.error_codes == [code] + recovery = error.error_info[0].recovery + assert recovery == STANDARD_ERROR_CODES.get(code, {}).get("recovery", "terminal") + if code == "INTERNAL_ERROR": + assert "internal provider detail" not in str(error) + await service.close() + + +async def test_disabled_optional_routes_and_aggregate_delivery_delegate(): + class Application(ADCPHandler): + async def sync_reporting_status(self, params, context=None): + return {"fallback": "status"} + + async def sync_reporting_receipts(self, params, context=None): + return {"fallback": "receipts"} + + async def get_media_buy_delivery(self, params, context=None): + return {"fallback": "delivery"} + + service = ReliableReportingService.memory(account_context=_account_context) + mounted = service.install(Application()) + assert await mounted.sync_reporting_status({}) == {"fallback": "status"} + assert await mounted.sync_reporting_receipts({}) == {"fallback": "receipts"} + assert await mounted.get_media_buy_delivery({}) == {"fallback": "delivery"} + await service.close() diff --git a/tests/type_checks/reliable_reporting_lifecycle.py b/tests/type_checks/reliable_reporting_lifecycle.py index f02c7a4b9..4ffe18201 100644 --- a/tests/type_checks/reliable_reporting_lifecycle.py +++ b/tests/type_checks/reliable_reporting_lifecycle.py @@ -55,3 +55,12 @@ async def serve_one_turn( except ReliableReportingServiceError as failure: component: str = failure.component assert component + + +def buyer_refusal() -> None: + from adcp.reporting import ReliableReportingConfigurationError + + refusal = ReliableReportingConfigurationError( + "invalid reporting selection", kind="invalid_request" + ) + assert refusal.kind == "invalid_request" From c707b46c405793c91a2030b86b8a624ca87304a2 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 2 Oct 2026 18:03:27 +0000 Subject: [PATCH 2/4] fix(reporting): classify reporting extensions and production refusals --- docs/reliable-reporting-service.md | 5 +- src/adcp/reporting/_errors.py | 48 ++++++++++++++- src/adcp/reporting/production/handler.py | 60 +++++++++++++++++-- ...st_reporting_production_service_factory.py | 35 +++++++++++ tests/test_reporting_installed_errors.py | 23 +++++++ 5 files changed, 161 insertions(+), 10 deletions(-) diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 5a1873e3e..5e05280b8 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -346,6 +346,7 @@ returns `AUTH_REQUIRED`; explicit buyer configuration validation can raise `ReliableReportingConfigurationError(message, kind="invalid_request")`. Composition errors retain `INTERNAL_ERROR` and terminal recovery, with internal configuration details omitted from buyer responses. Ledger codes are preserved -and receive the SDK's recovery classification for that code (unknown extension -codes default to terminal). Disabled optional methods and aggregate delivery +and receive the SDK's recovery classification for that code. Reporting cursor/ +input extensions are correctable, storage-unavailability extensions are transient, +and unknown extension or integrity failures default to terminal. Disabled optional methods and aggregate delivery continue to delegate to the application. diff --git a/src/adcp/reporting/_errors.py b/src/adcp/reporting/_errors.py index b1ac5c830..02ce3a7e8 100644 --- a/src/adcp/reporting/_errors.py +++ b/src/adcp/reporting/_errors.py @@ -8,11 +8,49 @@ from adcp.exceptions import ADCPTaskError from adcp.reporting.ledger.store import LedgerConflictError from adcp.reporting.service_lifecycle import ReliableReportingUnavailableError -from adcp.server.helpers import adcp_error +from adcp.server.helpers import STANDARD_ERROR_CODES, adcp_error from adcp.types import Error _T = TypeVar("_T") +# Reporting extensions have local meanings. Normative SDK codes always take +# precedence; unlisted extension/invariant failures retain terminal recovery. +_REPORTING_CORRECTABLE = frozenset( + { + "CURSOR_REVISION_MISMATCH", + "CURSOR_SNAPSHOT_MISMATCH", + "INVALID_CHECKPOINT", + "INVALID_CURSOR", + "INVALID_PAGE_SIZE", + "INVALID_PERIOD", + "INVALID_STATUS_PERIOD", + "INVALID_STATUS_TIME", + "INVALID_VIEW", + "MISSING_REVISION_ID", + "EMPTY_BATCH", + "UNKNOWN_CONFIGURATION_GENERATION", + "CONFIGURATION_GENERATION_IMMUTABLE", + "CONFIGURATION_GENERATION_MISMATCH", + "REPORT_DEFINITION_MISMATCH", + "OBLIGATION_IDENTITY_MISMATCH", + "SELLER_SNAPSHOT_EVIDENCE_INCOMPLETE", + "STATUS_NOT_DUE", + "STATUS_SUPERSEDES_REQUIRED", + "STATUS_SUPERSEDES_STALE", + "REVISION_NOT_CURRENTLY_REQUIRED", + "REPORTING_FEED_VERSION_MISMATCH", + } +) +_REPORTING_TRANSIENT = frozenset( + { + "REPORTING_CONFIGURATION_UNAVAILABLE", + "REPORTING_STATUS_UNAVAILABLE", + "REPORTING_CONTENT_UNAVAILABLE", + "REPORTING_FEED_STORAGE_UNAVAILABLE", + "RECEIPT_STORAGE_UNAVAILABLE", + } +) + class ReliableReportingConfigurationError(ValueError): """A composition failure, missing caller, or invalid buyer configuration. @@ -33,7 +71,13 @@ def __init__( def reporting_error(code: str, message: str) -> Error: """Preserve the domain code and use the SDK's normative recovery defaults.""" - return Error.model_validate(adcp_error(code, message)["errors"][0]) + recovery = None + if code not in STANDARD_ERROR_CODES: + if code in _REPORTING_CORRECTABLE: + recovery = "correctable" + elif code in _REPORTING_TRANSIENT: + recovery = "transient" + return Error.model_validate(adcp_error(code, message, recovery=recovery)["errors"][0]) async def installed_reporting_call( diff --git a/src/adcp/reporting/production/handler.py b/src/adcp/reporting/production/handler.py index 808d15016..1464af344 100644 --- a/src/adcp/reporting/production/handler.py +++ b/src/adcp/reporting/production/handler.py @@ -26,7 +26,10 @@ ReportingReceiptHandler, _consumer, ) -from adcp.reporting.service_lifecycle import ReliableReportingUnavailableError +from adcp.reporting.service_lifecycle import ( + ReliableReportingState, + ReliableReportingUnavailableError, +) from adcp.server.base import ADCPHandler, NotImplementedResponse, ToolContext from adcp.server.mcp_tools import ADCP_TOOL_DEFINITIONS, get_tools_for_handler from adcp.server.responses import capabilities_response @@ -83,13 +86,58 @@ async def call(self: ReportingProductionHandler, *args: Any, **kwargs: Any) -> A ) if aggregate: return await method(self, *args, **kwargs) + + async def invoke() -> Any: + support = self.production + if support._closed or support._failed or support._task is None or support._task.done(): + state = ( + ReliableReportingState.CLOSED + if support._closed + else ( + ReliableReportingState.FAILED + if support._failed + else ReliableReportingState.NEW + ) + ) + raise ReliableReportingUnavailableError(state) + try: + support._assert_components() + except ReportingNotificationError as error: + raise ReliableReportingConfigurationError( + "production composition failed" + ) from error + context = args[1] if len(args) > 1 else kwargs.get("context") + # A RequestContext's caller_identity is an account cache key. Only + # transport auth/registry evidence establishes the reporting caller. + present = context is not None and ( + any( + getattr(context, name, None) is not None + for name in ( + "auth_info", + "auth_principal", + "buyer_agent", + ) + ) + or ( + not isinstance(context, RequestContext) + and getattr(context, "caller_identity", None) is not None + ) + or any( + context.metadata.get(key) is not None + for key in ( + "adcp.auth_info", + "auth_info", + "adcp.buyer_agent", + ) + ) + ) + if not present: + raise ReliableReportingConfigurationError("missing caller", kind="auth_required") + return await method(self, *args, **kwargs) + return await installed_reporting_call( method.__name__, - lambda: ( - method(self, *args, **kwargs) - if lifecycle is None - else lifecycle.call(lambda: method(self, *args, **kwargs)) - ), + lambda: invoke() if lifecycle is None else lifecycle.call(invoke), ) return cast(_Method, call) diff --git a/tests/conformance/reporting/test_reporting_production_service_factory.py b/tests/conformance/reporting/test_reporting_production_service_factory.py index 543ab1aa0..9467c56d0 100644 --- a/tests/conformance/reporting/test_reporting_production_service_factory.py +++ b/tests/conformance/reporting/test_reporting_production_service_factory.py @@ -438,3 +438,38 @@ async def test_mounted_production_lifecycle_refusals_are_transient(backend, tmp_ await handler.get_reporting_status({}) assert caught.value.error_codes == ["SERVICE_UNAVAILABLE"] assert caught.value.is_retryable + + +@pytest.mark.parametrize("backend", ["memory", "postgres"]) +async def test_production_missing_caller_and_broken_composition_are_distinct(backend, tmp_path): + from adcp.exceptions import ADCPTaskError + from adcp.server.base import ToolContext + + async with factory_harness(backend, tmp_path / "errors.sqlite") as h: + handler = h.production.handler + tasks = ( + "get_reporting_status", + "sync_reporting_status", + "sync_reporting_receipts", + "get_media_buy_delivery", + "sync_accounts", + ) + for task in tasks: + request = ( + {"reporting_revision_id": "missing"} if task == "get_media_buy_delivery" else {} + ) + with pytest.raises(ADCPTaskError) as caught: + await getattr(handler, task)(request, ToolContext()) + assert caught.value.error_codes == ["AUTH_REQUIRED"] + assert caught.value.error_info[0].recovery == "correctable" + original = h.production.materializer.store + object.__setattr__(h.production.materializer, "store", object()) + try: + # Admission validates the concrete composition before reaching any + # buyer-input/ledger processing or broad storage-error catch. + with pytest.raises(ADCPTaskError) as caught: + await handler.sync_accounts({}, ToolContext(caller_identity="buyer")) + assert caught.value.error_codes == ["INTERNAL_ERROR"] + assert caught.value.error_info[0].recovery == "terminal" + finally: + object.__setattr__(h.production.materializer, "store", original) diff --git a/tests/test_reporting_installed_errors.py b/tests/test_reporting_installed_errors.py index cf5ab3071..93214f50f 100644 --- a/tests/test_reporting_installed_errors.py +++ b/tests/test_reporting_installed_errors.py @@ -89,3 +89,26 @@ async def get_media_buy_delivery(self, params, context=None): assert await mounted.sync_reporting_receipts({}) == {"fallback": "receipts"} assert await mounted.get_media_buy_delivery({}) == {"fallback": "delivery"} await service.close() + + +@pytest.mark.parametrize( + "code,recovery", + [ + ("CURSOR_REVISION_MISMATCH", "correctable"), + ("INVALID_PAGE_SIZE", "correctable"), + ("CONFIGURATION_GENERATION_IMMUTABLE", "correctable"), + ("HISTORY_UNAVAILABLE", "terminal"), + ("REVISION_CONTENT_MISMATCH", "terminal"), + ("REPORTING_STATUS_UNAVAILABLE", "transient"), + ("RECEIPT_STORAGE_UNAVAILABLE", "transient"), + ], +) +async def test_reporting_extension_codes_retain_their_meaning(code, recovery): + service = ReliableReportingService.memory(account_context=_account_context) + service.get_reporting_status = AsyncMock(side_effect=LedgerConflictError(code, "detail")) + handler = service.install(_SalesPlatform()) + with pytest.raises(AdcpError) as caught: + await handler.get_reporting_status({}) + assert caught.value.code == code + assert caught.value.recovery == recovery + await service.close() From 58594d7a8bf393f1134b88e8376987f64d18d9ea Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 2 Oct 2026 21:28:07 +0000 Subject: [PATCH 3/4] test(reporting): poll retained history through a running composition --- ...test_reporting_production_notifications.py | 42 ++++++++++++++++++- 1 file changed, 41 insertions(+), 1 deletion(-) diff --git a/tests/conformance/reporting/test_reporting_production_notifications.py b/tests/conformance/reporting/test_reporting_production_notifications.py index a8e98f910..901eee9b0 100644 --- a/tests/conformance/reporting/test_reporting_production_notifications.py +++ b/tests/conformance/reporting/test_reporting_production_notifications.py @@ -529,7 +529,47 @@ async def test_retry_horizon_is_immutable_across_crash_rotation_and_restart( {"account": {"account_id": "acct_a"}, "view": "periods"}, transport=transport, ) - assert polling["status"] == "completed" + assert polling["adcp_error"]["code"] == "SERVICE_UNAVAILABLE" + assert polling["adcp_error"]["recovery"] == "transient" + + # queued_production deliberately closes its owned lifecycle so these + # crash/deadline turns are deterministic. Poll retained history through + # a new running composition rather than bypassing that admission policy. + from adcp.reporting.production import ReportingProductionSupport + from adcp.server.serve import create_mcp_server + + previous = h.production + previous_mount = h.mount + current = ReportingProductionSupport( + previous.materializer, + previous.projection, + offerings=previous.offerings, + configuration_task=previous.configuration_task, + resolve_account=previous.handler._receipt_account_resolver, + notification_workers=fresh_workers(h), + poll_seconds=60, + adcp_version=previous.handler.get_adcp_version(), + source_registry=previous.source_registry, + ) + h.production = current + h.mount = create_mcp_server(current.handler) + try: + await current.start() + mount = MountedProduction(h) + mount.authorize(h.item) + async with mount.client() as client: + for transport in ("mcp", "a2a-0.3", "a2a-1.0"): + _, polling = await mount.call( + client, + "get_reporting_status", + {"account": {"account_id": "acct_a"}, "view": "periods"}, + transport=transport, + ) + assert polling.get("status") == "completed", polling + finally: + await current.aclose() + h.production = previous + h.mount = previous_mount @pytest.mark.parametrize("backend", ["memory", "postgres"]) From b326cdbb426362c077f66e8884f186d448f67fea Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sat, 3 Oct 2026 00:49:42 +0000 Subject: [PATCH 4/4] test(reporting): assert translated admission failure and ordinary delegation --- .../test_reliable_reporting_production_admission.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/tests/conformance/reporting/test_reliable_reporting_production_admission.py b/tests/conformance/reporting/test_reliable_reporting_production_admission.py index f84ca1c8c..605baac6f 100644 --- a/tests/conformance/reporting/test_reliable_reporting_production_admission.py +++ b/tests/conformance/reporting/test_reliable_reporting_production_admission.py @@ -411,7 +411,6 @@ async def broken(*args, **kwargs): async def test_service_failure_stops_admission_but_preserves_ordinary_delegate(tmp_path): from adcp.reporting.service import ( ReliableReportingServiceError, - ReliableReportingUnavailableError, ) async with production_harness( @@ -427,8 +426,12 @@ async def test_service_failure_stops_admission_but_preserves_ordinary_delegate(t with pytest.raises(ReliableReportingServiceError): await asyncio.wait_for(h.service.wait(), 2) assert not h.service.ready - with pytest.raises(ReliableReportingUnavailableError): + from adcp.exceptions import ADCPTaskError + + with pytest.raises(ADCPTaskError) as unavailable: await h.production.handler.sync_accounts({}) + assert unavailable.value.error_codes == ["SERVICE_UNAVAILABLE"] + assert unavailable.value.error_info[0].recovery == "transient" assert await h.production.handler.get_products({}) == {"products": []} assert (await h.production.handler.get_media_buy_delivery({}))["aggregated_totals"][ "impressions"