From 2df3c041313e9a8c838448b82b417596e84e8d41 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 23:48:25 +0000 Subject: [PATCH] fix(server): return task and workflow handoffs uncached from IdempotencyStore.wrap A method that may hand off to a workflow could not use the SDK's idempotency: IdempotencyStore.wrap coerced every completed result to a dict for the replay cache, and a TaskHandoff or WorkflowHandoff is a control marker for the dispatcher, not a response, so the call raised "Cannot coerce WorkflowHandoff to dict for idempotency caching" (#1781). Adopters copied the private wrapper to work around it. wrap now returns a handoff uncached, after the key, conflict, lock and cancellation handling, so the dispatcher allocates the task as usual. It also accepts cache_response=, a predicate a seller can use to keep its own control results out of the cache: @store.wrap(cache_response=...) caches a result only when the predicate returns True, and a handoff is never cached. Both forms register the wrapper for validate_idempotency_wiring. An uncached result stores no replay entry, so a retry with the same key runs the method again; the docs say to deduplicate the enqueued work on the key. Replaying the task id the dispatcher allocated needs dispatcher support and is left for a follow-up. Closes #1781 Co-Authored-By: Claude Opus 5.5 --- docs/handler-authoring.md | 9 ++ src/adcp/server/idempotency/store.py | 57 ++++++- tests/test_idempotency_handoff.py | 230 +++++++++++++++++++++++++++ 3 files changed, 294 insertions(+), 2 deletions(-) create mode 100644 tests/test_idempotency_handoff.py diff --git a/docs/handler-authoring.md b/docs/handler-authoring.md index 74fa0de0b..28adb7f5a 100644 --- a/docs/handler-authoring.md +++ b/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`): diff --git a/src/adcp/server/idempotency/store.py b/src/adcp/server/idempotency/store.py index 11c139b45..f25628026 100644 --- a/src/adcp/server/idempotency/store.py +++ b/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. diff --git a/tests/test_idempotency_handoff.py b/tests/test_idempotency_handoff.py new file mode 100644 index 000000000..316af3a91 --- /dev/null +++ b/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