From 042c21afffee9a21fba84deedfcb3a3370554bf8 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 26 Oct 2026 00:03:23 +0000 Subject: [PATCH] fix(decisioning): project ADCPTaskError from platform methods to its own wire error _invoke_platform_method caught only AdcpError and pydantic ValidationError, so IdempotencyConflictError raised by IdempotencyStore.wrap reached the buyer as INTERNAL_ERROR / terminal. ADCPTaskError subclasses now keep their code and recovery through the same field extraction serve() uses, on the sync path and in handoff tasks. Fixes #1622 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/handler-authoring.md | 6 + src/adcp/decisioning/dispatch.py | 40 ++++- tests/test_decisioning_dispatch.py | 65 ++++++++ .../test_decisioning_idempotency_conflict.py | 140 ++++++++++++++++++ 4 files changed, 250 insertions(+), 1 deletion(-) create mode 100644 tests/test_decisioning_idempotency_conflict.py diff --git a/docs/handler-authoring.md b/docs/handler-authoring.md index 91e35e0ec..181f5bbee 100644 --- a/docs/handler-authoring.md +++ b/docs/handler-authoring.md @@ -708,6 +708,12 @@ re-execute and can double-allocate. Always wire `context_factory` on MCP servers so the auth middleware populates these fields before the handler runs. +The same decorator works on `DecisioningPlatform` methods. A key reused with +a different payload raises `IdempotencyConflictError`; the dispatcher +projects it, like any other `ADCPTaskError`, to its own code and recovery +(`IDEMPOTENCY_CONFLICT` / `correctable`) on MCP and A2A rather than +`INTERNAL_ERROR`. You don't need your own translator. + ## Error handling **Handler methods return error dicts — they do not raise.** Use diff --git a/src/adcp/decisioning/dispatch.py b/src/adcp/decisioning/dispatch.py index eb56ddd0b..f40c9873e 100644 --- a/src/adcp/decisioning/dispatch.py +++ b/src/adcp/decisioning/dispatch.py @@ -80,6 +80,7 @@ _extract_push_operation_id, emit_terminal_completion_webhook, ) +from adcp.exceptions import ADCPTaskError if TYPE_CHECKING: from collections.abc import Awaitable, Callable @@ -711,6 +712,30 @@ def _exception_cause_details(exc: BaseException) -> dict[str, Any]: return {"caused_by": {"type": type(exc).__name__}} +def _task_error_to_adcp_error(exc: ADCPTaskError) -> AdcpError: + """Project an AdCP-typed task error to its own wire error. + + ``IdempotencyStore.wrap`` raises + :class:`~adcp.exceptions.IdempotencyConflictError` (an + ``ADCPTaskError``) for a reused key. It is a protocol answer, not an + internal failure, so it keeps its code and recovery. The field + extraction is the one ``serve()`` applies to the same exceptions, so + both paths answer with the same code, message, recovery, field, + suggestion and sanitized details. + """ + from adcp.server.translate import _extract_structured_fields + + code, message, recovery, field, suggestion, details, _errors = _extract_structured_fields(exc) + return AdcpError( + code, + message=message, + recovery=typing.cast(Any, recovery), + field=field, + suggestion=suggestion, + details=details, + ) + + def _internal_error_details(exc: BaseException) -> dict[str, Any]: """Build the wire-side ``details`` payload for an INTERNAL_ERROR wrap. @@ -1511,7 +1536,9 @@ async def _invoke_platform_method( allocate a task_id, kick off the handoff fn, and project the Submitted envelope. - Wraps any non-:class:`AdcpError` exception to + Projects an :class:`~adcp.exceptions.ADCPTaskError` (for example + ``IdempotencyConflictError`` from ``IdempotencyStore.wrap``) to its own + code and recovery. Wraps any other non-:class:`AdcpError` exception to ``AdcpError("INTERNAL_ERROR", recovery="terminal")`` so the wire response never leaks a stack trace. Adopters get the original exception logged via the framework's observability hooks (the @@ -1655,6 +1682,14 @@ async def _invoke_platform_method( if on_failure is not None: await _safe_on_failure_call(on_failure, exc, method_name) raise + except ADCPTaskError as exc: + # AdCP-typed task errors (IdempotencyConflictError from + # ``IdempotencyStore.wrap``, account errors) carry their own wire + # code and recovery; never collapse them to INTERNAL_ERROR. + wrapped = _task_error_to_adcp_error(exc) + if on_failure is not None: + await _safe_on_failure_call(on_failure, wrapped, method_name) + raise wrapped from exc except TypeError as exc: # Most likely an arg_projector or extra_kwargs signature-drift bug. # Bare INTERNAL_ERROR would hide the cause; project to @@ -2301,6 +2336,9 @@ async def _run() -> None: except AdcpError as exc: await _fail(exc) return + except ADCPTaskError as exc: + await _fail(_task_error_to_adcp_error(exc)) + return except Exception as exc: logger.exception( "Unhandled exception in handoff fn for task %s — wrapping", diff --git a/tests/test_decisioning_dispatch.py b/tests/test_decisioning_dispatch.py index 045c77297..c4cda4f0b 100644 --- a/tests/test_decisioning_dispatch.py +++ b/tests/test_decisioning_dispatch.py @@ -803,6 +803,44 @@ async def get_products(self, req, ctx): assert exc_info.value.recovery == "correctable" +@pytest.mark.asyncio +async def test_invoke_projects_adcp_task_errors_to_their_own_code( + executor: ThreadPoolExecutor, +) -> None: + """An ``ADCPTaskError`` (``IdempotencyStore.wrap``'s conflict) keeps its + code and recovery instead of collapsing to INTERNAL_ERROR (#1622).""" + from adcp.exceptions import IdempotencyConflictError + + class _ConflictPlatform(DecisioningPlatform): + capabilities = DecisioningCapabilities() + accounts = SingletonAccounts(account_id="x") + + async def get_products(self, req, ctx): + raise IdempotencyConflictError(operation="get_products", errors=[]) + + failures: list[BaseException] = [] + + async def _on_failure(exc: BaseException) -> None: + failures.append(exc) + + ctx = _build_request_context(ToolContext(), Account(id="x"), None) + with pytest.raises(AdcpError) as exc_info: + await _invoke_platform_method( + _ConflictPlatform(), + "get_products", + _ProductsRequest(), + ctx, + executor=executor, + registry=InMemoryTaskRegistry(), + on_failure=_on_failure, + ) + assert exc_info.value.code == "IDEMPOTENCY_CONFLICT" + assert exc_info.value.recovery == "correctable" + assert exc_info.value.suggestion is not None + assert isinstance(exc_info.value.__cause__, IdempotencyConflictError) + assert failures == [exc_info.value] + + @pytest.mark.asyncio async def test_invoke_wraps_unexpected_exceptions_to_internal_error( executor: ThreadPoolExecutor, @@ -1335,6 +1373,33 @@ async def _handoff_fn(task_ctx): assert "internal bug" not in rec["error"].get("message", "") +@pytest.mark.asyncio +async def test_handoff_adcp_task_error_keeps_its_code( + executor: ThreadPoolExecutor, +) -> None: + from adcp.exceptions import IdempotencyConflictError + + registry = InMemoryTaskRegistry() + ctx = _build_request_context(ToolContext(), Account(id="acct_a"), None) + + async def _handoff_fn(task_ctx): + raise IdempotencyConflictError(operation="create_media_buy", errors=[]) + + envelope = await _project_handoff( + TaskHandoff(_handoff_fn), + ctx, + method_name="create_media_buy", + registry=registry, + executor=executor, + ) + await asyncio.sleep(0.1) + rec = await registry.get(envelope["task_id"], expected_account_id="acct_a") + assert rec is not None + assert rec["state"] == "failed" + assert rec["error"]["code"] == "IDEMPOTENCY_CONFLICT" + assert rec["error"]["recovery"] == "correctable" + + @pytest.mark.asyncio async def test_handoff_request_context_echoes_into_completed_task( executor: ThreadPoolExecutor, diff --git a/tests/test_decisioning_idempotency_conflict.py b/tests/test_decisioning_idempotency_conflict.py new file mode 100644 index 000000000..957cccf95 --- /dev/null +++ b/tests/test_decisioning_idempotency_conflict.py @@ -0,0 +1,140 @@ +"""A decisioning method wrapped by ``IdempotencyStore.wrap`` answers a reused +key with ``IDEMPOTENCY_CONFLICT``, not ``INTERNAL_ERROR`` (#1622). + +The conflict is an :class:`~adcp.exceptions.ADCPTaskError` raised inside the +platform method, so the decisioning dispatcher must project it to its own wire +error, as the handler ``serve()`` path does. Both requests run through the real +MCP and A2A servers. +""" + +from __future__ import annotations + +import secrets +from typing import Any + +import pytest + +from adcp.decisioning import DecisioningCapabilities, DecisioningPlatform, SingletonAccounts +from adcp.server import IdempotencyStore, MemoryBackend, ToolContext +from adcp.server.helpers import STANDARD_ERROR_CODES +from adcp.testing import build_test_client +from adcp.types import CreateMediaBuySuccessResponse, Package + +store = IdempotencyStore(backend=MemoryBackend(), ttl_seconds=86_400) + + +class Seller(DecisioningPlatform): + capabilities = DecisioningCapabilities( + specialisms=["sales-non-guaranteed"], channels=["display"], pricing_models=["cpm"] + ) + accounts = SingletonAccounts(account_id="acct_1") + + def __init__(self) -> None: + self.created = 0 + + @store.wrap + async def create_media_buy(self, req: Any, ctx: Any) -> CreateMediaBuySuccessResponse: + self.created += 1 + return CreateMediaBuySuccessResponse( + media_buy_id=f"mb_{self.created}", + confirmed_at="2026-05-01T00:00:00Z", + revision=1, + status="active", + packages=[Package(package_id="pkg_1", product_id="prod_1")], + ) + + def get_products(self, req: Any, ctx: Any) -> Any: + raise NotImplementedError + + def update_media_buy(self, media_buy_id: Any, patch: Any, ctx: Any) -> Any: + raise NotImplementedError + + def sync_creatives(self, req: Any, ctx: Any) -> Any: + raise NotImplementedError + + def get_media_buy_delivery(self, req: Any, ctx: Any) -> Any: + raise NotImplementedError + + def get_media_buys(self, req: Any, ctx: Any) -> Any: + raise NotImplementedError + + +def _create_request(key: str, budget: int) -> dict[str, Any]: + return { + "idempotency_key": key, + "account": {"account_id": "acct_1"}, + "brand": {"domain": "brand.example"}, + "start_time": "2026-10-01T00:00:00Z", + "end_time": "2026-10-31T00:00:00Z", + "packages": [ + { + "product_id": "prod_1", + "pricing_option_id": "cpm_1", + "budget": budget, + } + ], + } + + +async def _call(client: Any, transport: str, arguments: dict[str, Any]) -> dict[str, Any]: + if transport == "a2a": + response = await client.post( + "/", + json={ + "jsonrpc": "2.0", + "id": secrets.token_hex(4), + "method": "message/send", + "params": { + "message": { + "messageId": secrets.token_hex(8), + "role": "user", + "parts": [ + { + "kind": "data", + "data": {"skill": "create_media_buy", "parameters": arguments}, + } + ], + } + }, + }, + ) + assert response.status_code == 200, response.text + result = response.json()["result"] + return next(p["data"] for p in result["artifacts"][0]["parts"] if "data" in p) + response = await client.post( + "/mcp/", + headers={"accept": "application/json, text/event-stream"}, + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": {"name": "create_media_buy", "arguments": arguments}, + }, + ) + assert response.status_code == 200, response.text + return response.json()["result"]["structuredContent"] + + +def _caller(meta: Any) -> ToolContext: + return ToolContext(caller_identity="buyer_1") + + +@pytest.mark.parametrize("transport", ["mcp", "a2a"]) +async def test_reused_key_with_different_payload_returns_idempotency_conflict( + transport: str, +) -> None: + seller = Seller() + key = f"idem-{secrets.token_hex(8)}" + async with build_test_client( + seller, transport=transport, stateless_http=True, context_factory=_caller + ) as client: + first = await _call(client, transport, _create_request(key, 1000)) + assert "adcp_error" not in first, first + second = await _call(client, transport, _create_request(key, 2000)) + + error = second["adcp_error"] + assert error["code"] == "IDEMPOTENCY_CONFLICT", second + assert error["recovery"] == STANDARD_ERROR_CODES["IDEMPOTENCY_CONFLICT"]["recovery"] + assert error["recovery"] == "correctable" + assert "caused_by" not in error.get("details", {}) + assert seller.created == 1