fix(server): return task and workflow handoffs uncached from IdempotencyStore.wrap

docs/handler-authoring.md

@@ -704,6 +704,15 @@ class MySeller(ADCPHandler):
         return capabilities_response(["media_buy"], idempotency=idempotency.capability())
 ```
 
+A wrapped method may hand off: a `TaskHandoff` or `WorkflowHandoff`
+(`ctx.handoff_to_task(...)` / `ctx.handoff_to_workflow(...)`) is returned to
+the dispatcher uncached, because it is a control marker rather than a
+response. Nothing is stored for that key, so a retry with the same
+`idempotency_key` runs the method again; make the work the handoff enqueues
+deduplicate on the key. To keep other results out of the replay cache, pass a
+predicate: `@idempotency.wrap(cache_response=lambda result: ...)` caches a
+result only when it returns `True`.
+
 For production, swap `MemoryBackend()` for `PgBackend` (note the
 import path — `PgBackend` lives in `adcp.server.idempotency`, not the
 top-level `adcp.server`):

src/adcp/server/idempotency/store.py

@@ -39,7 +39,7 @@
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from contextvars import ContextVar
 from functools import wraps
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, overload
 
 from pydantic import BaseModel
 
@@ -123,6 +123,17 @@ def is_wrapped(fn: Any) -> bool:
 _MAX_TTL_SECONDS = 604800
 
 HandlerFn = Callable[..., Awaitable[Any]]
+#: Predicate deciding whether a completed handler result is cached for replay.
+CacheResponsePredicate = Callable[[Any], bool]
+
+
+def _is_handoff(response: Any) -> bool:
+    """A ``TaskHandoff`` / ``WorkflowHandoff`` control marker, not a response."""
+    # Local import: the server middleware must not import the decisioning
+    # package at module import time.
+    from adcp.decisioning.types import is_task_handoff, is_workflow_handoff
+
+    return is_task_handoff(response) or is_workflow_handoff(response)
 
 
 class IdempotencyStore:
@@ -237,7 +248,22 @@ def reserve(
             operation=operation,
         )
 
-    def wrap(self, handler: HandlerFn) -> HandlerFn:
+    @overload
+    def wrap(
+        self, handler: HandlerFn, *, cache_response: CacheResponsePredicate | None = None
+    ) -> HandlerFn: ...
+
+    @overload
+    def wrap(
+        self, handler: None = None, *, cache_response: CacheResponsePredicate | None = None
+    ) -> Callable[[HandlerFn], HandlerFn]: ...
+
+    def wrap(
+        self,
+        handler: HandlerFn | None = None,
+        *,
+        cache_response: CacheResponsePredicate | None = None,
+    ) -> HandlerFn | Callable[[HandlerFn], HandlerFn]:
         """Decorator that adds idempotency semantics to an AdCP handler method.
 
         Supports three calling conventions the framework dispatches with:
@@ -265,7 +291,24 @@ def wrap(self, handler: HandlerFn) -> HandlerFn:
         re-validation on a hit (when the handler's declared return
         type exposes ``model_validate``). Callers that return raw
         dicts get dicts back.
+
+        **Uncached results.** A :class:`~adcp.decisioning.TaskHandoff` or
+        :class:`~adcp.decisioning.WorkflowHandoff` is a control marker for
+        the dispatcher, not a response, so it is returned uncached: the
+        dispatcher allocates the task and owns what happens next. Use
+        ``@store.wrap(cache_response=predicate)`` to mark other results
+        as not cacheable; ``predicate(result)`` returning ``False``
+        returns the result uncached (a handoff is never cached, whatever
+        the predicate says). An uncached result holds no replay entry, so
+        a retry with the same ``idempotency_key`` runs the method again:
+        deduplicate the work a handoff enqueues on that key.
         """
+        if handler is None:
+
+            def _decorate(fn: HandlerFn) -> HandlerFn:
+                return self.wrap(fn, cache_response=cache_response)
+
+            return _decorate
 
         @wraps(handler)
         async def _wrapped(*args: Any, **kwargs: Any) -> Any:
@@ -343,6 +386,16 @@ async def _execute_locked() -> Any:
                         return _replay_cached(cached)
 
                     response = await handler(*args, **kwargs)
+                    if _is_handoff(response) or (
+                        cache_response is not None and not cache_response(response)
+                    ):
+                        logger.debug(
+                            "idempotency: %s result not cached: scope=%s key_prefix=%s",
+                            operation,
+                            _scope_log_id(scope_key),
+                            idempotency_key[:8],
+                        )
+                        return response
                     # Deep-copy when caching so post-return mutation of the caller's
                     # copy can't poison future replays. `_clone_response` also deep-
                     # copies on the hit path, giving independent objects per replay.

tests/test_idempotency_handoff.py

@@ -0,0 +1,230 @@
+"""``IdempotencyStore.wrap`` returns task and workflow handoffs uncached (#1781).
+
+A handoff is a control marker for the dispatcher, not a response, so the
+store cannot cache it. ``cache_response=`` lets a seller mark other results
+as not cacheable.
+"""
+
+from __future__ import annotations
+
+import secrets
+from typing import Any
+
+import pytest
+
+from adcp.decisioning import DecisioningCapabilities, DecisioningPlatform, SingletonAccounts
+from adcp.decisioning.types import TaskHandoff, WorkflowHandoff
+from adcp.decisioning.validate_idempotency import is_wrapped
+from adcp.server import IdempotencyStore, MemoryBackend, ToolContext
+from adcp.testing import build_test_client
+from adcp.types import CreateMediaBuySuccessResponse, Package
+
+CONTEXT = ToolContext(caller_identity="buyer_1")
+
+
+def _params(key: str | None = None, **extra: Any) -> dict[str, Any]:
+    return {"idempotency_key": key or "k" * 16, "media_buy_id": "mb_1", **extra}
+
+
+@pytest.mark.parametrize(
+    "make_handoff",
+    [lambda: WorkflowHandoff(lambda task_ctx: None), lambda: TaskHandoff(lambda task_ctx: {})],
+    ids=["workflow", "task"],
+)
+async def test_handoff_is_returned_uncached(make_handoff: Any) -> None:
+    store = IdempotencyStore(MemoryBackend())
+    returned: list[Any] = []
+
+    class Platform:
+        @store.wrap
+        async def update_media_buy(self, params: Any, context: Any = None) -> Any:
+            handoff = make_handoff()
+            returned.append(handoff)
+            return handoff
+
+    platform = Platform()
+    first = await platform.update_media_buy(_params(), CONTEXT)
+    assert first is returned[0]
+    # Nothing was cached, so a retry with the same key runs the method again
+    # rather than replaying a marker the dispatcher already consumed.
+    second = await platform.update_media_buy(_params(), CONTEXT)
+    assert second is returned[1]
+    assert len(returned) == 2
+
+
+async def test_a_completed_response_after_a_handoff_is_cached() -> None:
+    store = IdempotencyStore(MemoryBackend())
+    calls = 0
+
+    class Platform:
+        @store.wrap
+        async def update_media_buy(self, params: Any, context: Any = None) -> Any:
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                return WorkflowHandoff(lambda task_ctx: None)
+            return {"media_buy_id": "mb_1", "revision": calls}
+
+    platform = Platform()
+    assert isinstance(await platform.update_media_buy(_params(), CONTEXT), WorkflowHandoff)
+    assert await platform.update_media_buy(_params(), CONTEXT) == {
+        "media_buy_id": "mb_1",
+        "revision": 2,
+    }
+    replay = await platform.update_media_buy(_params(), CONTEXT)
+    assert replay == {"media_buy_id": "mb_1", "revision": 2, "replayed": True}
+    assert calls == 2
+
+
+async def test_cache_response_predicate_marks_results_uncached() -> None:
+    store = IdempotencyStore(MemoryBackend())
+    calls = 0
+
+    class Pending:
+        """A seller's own control result, not a wire response."""
+
+    class Platform:
+        @store.wrap(cache_response=lambda result: not isinstance(result, Pending))
+        async def update_media_buy(self, params: Any, context: Any = None) -> Any:
+            nonlocal calls
+            calls += 1
+            return Pending() if params.get("paused") else {"media_buy_id": "mb_1"}
+
+    platform = Platform()
+    assert is_wrapped(platform.update_media_buy)
+    assert isinstance(await platform.update_media_buy(_params(paused=True), CONTEXT), Pending)
+    assert isinstance(await platform.update_media_buy(_params(paused=True), CONTEXT), Pending)
+    assert calls == 2
+
+    key = "c" * 16
+    assert await platform.update_media_buy(_params(key), CONTEXT) == {"media_buy_id": "mb_1"}
+    assert (await platform.update_media_buy(_params(key), CONTEXT))["replayed"] is True
+    assert calls == 3
+
+
+async def test_predicate_cannot_cache_a_handoff() -> None:
+    store = IdempotencyStore(MemoryBackend())
+    calls = 0
+
+    class Platform:
+        @store.wrap(cache_response=lambda result: True)
+        async def update_media_buy(self, params: Any, context: Any = None) -> Any:
+            nonlocal calls
+            calls += 1
+            return WorkflowHandoff(lambda task_ctx: None)
+
+    platform = Platform()
+    await platform.update_media_buy(_params(), CONTEXT)
+    await platform.update_media_buy(_params(), CONTEXT)
+    assert calls == 2
+
+
+async def test_uncached_result_still_checks_conflicts_against_cached_ones() -> None:
+    """A key first used for a cached response keeps its conflict detection."""
+    from adcp.exceptions import IdempotencyConflictError
+
+    store = IdempotencyStore(MemoryBackend())
+
+    class Platform:
+        @store.wrap
+        async def update_media_buy(self, params: Any, context: Any = None) -> Any:
+            if params.get("paused"):
+                return WorkflowHandoff(lambda task_ctx: None)
+            return {"media_buy_id": "mb_1"}
+
+    platform = Platform()
+    await platform.update_media_buy(_params(), CONTEXT)
+    with pytest.raises(IdempotencyConflictError):
+        await platform.update_media_buy(_params(paused=True), CONTEXT)
+
+
+# End to end: a decisioning method that may hand off to a workflow keeps the
+# SDK's idempotency, and the buyer gets the framework's Submitted envelope.
+
+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.enqueued: list[str] = []
+        self.created = 0
+
+    @store.wrap
+    async def create_media_buy(self, req: Any, ctx: Any) -> Any:
+        if req.packages[0].budget > 5000:
+            # Needs a person: hand the call to an external review workflow.
+            return ctx.handoff_to_workflow(lambda task_ctx: self.enqueued.append(task_ctx.id))
+        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 _create(client: Any, arguments: dict[str, Any]) -> dict[str, Any]:
+    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"]  # type: ignore[no-any-return]
+
+
+async def test_wrapped_decisioning_method_can_hand_off_to_a_workflow() -> None:
+    seller = Seller()
+    async with build_test_client(
+        seller,
+        transport="mcp",
+        stateless_http=True,
+        context_factory=lambda meta: ToolContext(caller_identity="buyer_1"),
+    ) as client:
+        submitted = await _create(client, _create_request(f"idem-{secrets.token_hex(8)}", 9000))
+        assert submitted["status"] == "submitted", submitted
+        assert seller.enqueued == [submitted["task_id"]]
+
+        # The synchronous path on the same wrapper still replays.
+        key = f"idem-{secrets.token_hex(8)}"
+        first = await _create(client, _create_request(key, 1000))
+        replay = await _create(client, _create_request(key, 1000))
+    assert first["media_buy_id"] == replay["media_buy_id"] == "mb_1"
+    assert seller.created == 1