fix(decisioning): project ADCPTaskError from platform methods to its own wire error

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

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",

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,

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