Fix repeated Deep Agents tool calls with identical arguments

CHANGELOG.md

@@ -52,6 +52,16 @@ to include examples, links to docs, or any other relevant information.
 
 ### Fixed
 
+- `temporalio.contrib.deepagents` no longer dedups repeated identical tool,
+  model, and backend-op calls: each dispatch runs its own Activity, and the
+  continue-as-new result cache is retired for new executions (a continued run
+  resumes from the carried transcript and never re-executes prior dispatches,
+  so a carried cache entry could only serve stale results). Patch-gated
+  (`deepagents.retire-result-cache`), so histories recorded before this change
+  replay unchanged; note that deferring the patch keeps the full legacy dedup
+  cache — including the stale-result behavior this entry describes — and that
+  a chain upgraded mid-continue-as-new re-executes rather than reuses a
+  repeated identical call (the conservative direction).
 - **Experimental**: External storage metrics now report the wall-clock time storage was in flight.
   Previously each batch's duration was summed, over-reporting the time whenever storage operations
   ran concurrently.

temporalio/contrib/deepagents/README.md

@@ -294,9 +294,10 @@ class LongResearchAgent:
 Pass `continue_as_new_after=N` instead to trigger on a fixed history-event
 count.
 
-- **Carries forward:** the accumulated messages and the model/tool result cache
-  (so an LLM/tool call completed before the continue-as-new is *not* re-run
-  after it). Your `@workflow.run` must accept `state_snapshot=None` as shown.
+- **Carries forward:** the accumulated messages. Repeated identical calls are
+  NOT deduplicated: a call the agent re-issues after the boundary runs its own
+  Activity, so a genuinely new identical request is never served a stale prior
+  result. Your `@workflow.run` must accept `state_snapshot=None` as shown.
 - **Does not carry forward:** anything held only in an in-memory checkpointer's
   own structures beyond the messages/todos snapshot. The default in-workflow
   `InMemorySaver` is rehydrated for free by deterministic replay; a durable

temporalio/contrib/deepagents/_activity.py

@@ -338,9 +338,9 @@ async def invoke_tool(self, input: ToolActivityInput) -> ToolActivityOutput:
     @_auto_heartbeater
     async def backend_op(self, input: BackendOpInput) -> BackendOpOutput:
         """Run one operation against a registered (real-I/O) backend."""
-        from temporalio.contrib.deepagents._tools import registered_backends
+        from temporalio.contrib.deepagents._tools import lookup_backend
 
-        backend = registered_backends().get(input.backend_ref)
+        backend = lookup_backend(input.backend_ref)
         if backend is None:
             raise ApplicationError(
                 f"Backend {input.backend_ref!r} is not registered on this worker.",

temporalio/contrib/deepagents/_tools.py

@@ -55,6 +55,11 @@
 
 _TOOL_REGISTRY: dict[str, "BaseTool"] = {}
 _BACKEND_REGISTRY: dict[str, Any] = {}
+# Recently retired backends (insertion-ordered, oldest evicted first). Keeps a
+# GC'd wrapper's backend reachable for the eviction -> activity-start window;
+# see _unregister_backend. Bounded so retired backends cannot accumulate.
+_RETIRED_BACKENDS: dict[str, Any] = {}
+_RETIRED_BACKENDS_MAX = 512
 # Serializes registration against the GC-time unregister in
 # _unregister_backend, which may run on another thread.
 _BACKEND_REGISTRY_LOCK = threading.Lock()
@@ -154,25 +159,49 @@ def register_backend(ref: str, backend: Any) -> None:
 
 
 def _unregister_backend(ref: str, inner: Any) -> None:
-    """Drop ``ref`` from the registry if it still maps to ``inner``.
+    """Retire ``ref`` from the registry if it still maps to ``inner``.
 
     GC hook for :class:`TemporalBackend` (via ``weakref.finalize``): a wrapper
     is typically constructed per workflow run, so without cleanup a long-lived
     worker accumulates one registry entry per run. The identity guard is
     load-bearing: refs are deterministic per run, so after a cache eviction a
     replay re-registers the *same* ref with a fresh inner backend — the evicted
     wrapper's finalizer must not remove that live registration.
+
+    Retired entries move to the bounded :data:`_RETIRED_BACKENDS` store instead
+    of vanishing: a ``backend_op`` activity scheduled just before a cache
+    eviction can be delivered *after* the evicted wrapper is collected, and the
+    replay that would re-register the ref only happens once that very activity
+    completes. :func:`lookup_backend` still resolves the ref in that window.
     """
     with _BACKEND_REGISTRY_LOCK:
         if _BACKEND_REGISTRY.get(ref) is inner:
             del _BACKEND_REGISTRY[ref]
+            _RETIRED_BACKENDS.pop(ref, None)
+            _RETIRED_BACKENDS[ref] = inner
+            while len(_RETIRED_BACKENDS) > _RETIRED_BACKENDS_MAX:
+                _RETIRED_BACKENDS.pop(next(iter(_RETIRED_BACKENDS)))
 
 
 def registered_backends() -> dict[str, Any]:
     """Return the live backend registry (read by the plugin at worker build)."""
     return _BACKEND_REGISTRY
 
 
+def lookup_backend(ref: str) -> Any | None:
+    """Resolve ``ref`` for a ``backend_op`` activity.
+
+    Prefers the live registry, then falls back to recently retired entries so
+    an activity dispatched before a cache eviction still resolves (see
+    :func:`_unregister_backend`).
+    """
+    with _BACKEND_REGISTRY_LOCK:
+        backend = _BACKEND_REGISTRY.get(ref)
+        if backend is not None:
+            return backend
+        return _RETIRED_BACKENDS.get(ref)
+
+
 # ---------------------------------------------------------------------------
 # activity_as_tool
 # ---------------------------------------------------------------------------

temporalio/contrib/deepagents/workflow.py

@@ -3,8 +3,9 @@
 Everything here runs *inside* the workflow. The dispatch helpers
 (:func:`call_model` / :func:`call_tool` / :func:`call_backend_op`) are the single
 choke point through which the in-workflow model / tool / backend stubs reach
-their activities; they also consult the continue-as-new result cache so work
-done before a ``continue_as_new`` is reused rather than repeated after it.
+their activities. Every dispatch runs its own Activity; only executions
+recorded before the ``deepagents.retire-result-cache`` patch consult the
+legacy continue-as-new result cache during replay.
 
 :func:`run_deep_agent` is the optional driver that adds continue-as-new
 state-carry around a native ``agent.ainvoke(...)`` — plain ``agent.ainvoke(...)``
@@ -77,20 +78,47 @@ def __init__(self, message: str, *, non_retryable: bool = True) -> None:
 # ---------------------------------------------------------------------------
 
 
+def _legacy_lookup(kind: str, name: str, payload: Any) -> tuple[str | None, bool, Any]:
+    """Legacy-cache lookup: ``(key, hit, value)``; key is None on new executions."""
+    if not _legacy_result_cache():
+        return None, False, None
+    key = _serde.cache_key(kind, name, payload)
+    hit, value = _serde.cache_lookup(key)
+    return key, hit, value
+
+
+def _legacy_result_cache() -> bool:
+    """Whether this execution uses the legacy continue-as-new result cache.
+
+    New executions do not cache at all: repeated identical calls are
+    legitimate work (a re-issued tool call, a deliberate model resample, a
+    re-read after a write), and under the resume-from-transcript
+    continue-as-new semantics a continued run never re-executes prior
+    dispatches — so a carried cache entry could only ever serve a stale
+    result to a genuinely new call. Replay of a single run needs no cache:
+    history supplies recorded activity results.
+
+    Patch-gated because histories recorded under the legacy cache contain
+    dedup decisions (a repeated call answered with no activity scheduled);
+    replaying them without the cache would emit commands history does not
+    have.
+    """
+    return not workflow.patched("deepagents.retire-result-cache")
+
+
 async def call_model(
     activity_name: str,
     activity_input: _activity.ModelActivityInput,
     *,
     summary: str,
     **opts: Any,
 ) -> _activity.ModelActivityOutput:
-    """Dispatch one model call, reusing a cached result across continue-as-new."""
-    key = _serde.cache_key(
+    """Dispatch one model call as its own Activity."""
+    legacy_key, hit, cached = _legacy_lookup(
         "model",
         activity_input.model_name,
         [activity_input.messages, activity_input.tool_schemas],
     )
-    hit, cached = _serde.cache_lookup(key)
     if hit:
         return _activity.ModelActivityOutput(message=cached)
     output = await workflow.execute_activity(
@@ -100,7 +128,8 @@ async def call_model(
         summary=summary,
         **opts,
     )
-    _serde.cache_put(key, output.message)
+    if legacy_key is not None:
+        _serde.cache_put(legacy_key, output.message)
     return output
 
 
@@ -110,9 +139,10 @@ async def call_tool(
     summary: str,
     **opts: Any,
 ) -> _activity.ToolActivityOutput:
-    """Dispatch one tool call, reusing a cached result across continue-as-new."""
-    key = _serde.cache_key("tool", activity_input.tool_name, activity_input.args)
-    hit, cached = _serde.cache_lookup(key)
+    """Dispatch one tool call as its own Activity."""
+    legacy_key, hit, cached = _legacy_lookup(
+        "tool", activity_input.tool_name, activity_input.args
+    )
     if hit:
         return _activity.ToolActivityOutput(message=cached)
     output = await workflow.execute_activity(
@@ -122,7 +152,8 @@ async def call_tool(
         summary=summary,
         **opts,
     )
-    _serde.cache_put(key, output.message)
+    if legacy_key is not None:
+        _serde.cache_put(legacy_key, output.message)
     return output
 
 
@@ -132,13 +163,12 @@ async def call_backend_op(
     summary: str,
     **opts: Any,
 ) -> _activity.BackendOpOutput:
-    """Dispatch one backend op, reusing a cached result across continue-as-new."""
-    key = _serde.cache_key(
+    """Dispatch one backend op as its own Activity."""
+    legacy_key, hit, cached = _legacy_lookup(
         f"backend:{activity_input.backend_ref}",
         activity_input.op,
         [activity_input.args, activity_input.kwargs],
     )
-    hit, cached = _serde.cache_lookup(key)
     if hit:
         return _activity.BackendOpOutput(result=cached)
     output = await workflow.execute_activity(
@@ -148,7 +178,8 @@ async def call_backend_op(
         summary=summary,
         **opts,
     )
-    _serde.cache_put(key, output.result)
+    if legacy_key is not None:
+        _serde.cache_put(legacy_key, output.result)
     return output
 
 
@@ -204,7 +235,14 @@ async def run_deep_agent(
     """
     # Resume path: rehydrate the result cache and fold carried messages in.
     if state_snapshot is not None:
-        _serde.set_result_cache(dict(state_snapshot.get(_CACHE_KEY) or {}))
+        # Only legacy executions consult the carried cache; on new
+        # executions the inbound legacy entries are dead weight, and at the
+        # upgrade hop a repeated identical call re-executes (conservative
+        # direction) rather than being served a possibly-stale carried result.
+        if _legacy_result_cache():
+            _serde.set_result_cache(dict(state_snapshot.get(_CACHE_KEY) or {}))
+        else:
+            _serde.set_result_cache({})
         input = _merge_snapshot(input, state_snapshot)
     else:
         _serde.set_result_cache({})
@@ -230,10 +268,9 @@ async def run_deep_agent(
             workflow.info().get_current_history_length() >= continue_as_new_after
         )
     if should_continue and _has_pending_work(result):
-        snapshot = {
-            "messages": _extract_messages(result),
-            _CACHE_KEY: _serde.result_cache_snapshot() or {},
-        }
+        snapshot: dict[str, Any] = {"messages": _extract_messages(result)}
+        if _legacy_result_cache():
+            snapshot[_CACHE_KEY] = _serde.result_cache_snapshot() or {}
         # ``continue_as_new`` threads positional args into the next run via
         # ``args=``; the enclosing ``@workflow.run`` receives them as
         # ``(input, state_snapshot)``.

tests/contrib/deepagents/histories/legacy_cache_carried_across_can.json

@@ -0,0 +1,129 @@
+{
+  "events": [
+    {
+      "eventId": "1",
+      "eventTime": "2026-09-11T04:57:21.100741Z",
+      "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED",
+      "taskId": "1048617",
+      "workflowExecutionStartedEventAttributes": {
+        "workflowType": {
+          "name": "LegacyCanDedupWorkflow"
+        },
+        "taskQueue": {
+          "name": "lcan",
+          "kind": "TASK_QUEUE_KIND_NORMAL"
+        },
+        "input": {
+          "payloads": [
+            {
+              "metadata": {
+                "encoding": "anNvbi9wbGFpbg=="
+              },
+              "data": "eyJtZXNzYWdlcyI6W119"
+            },
+            {
+              "metadata": {
+                "encoding": "anNvbi9wbGFpbg=="
+              },
+              "data": "eyJfX3RlbXBvcmFsX2NhY2hlX18iOnsiZTA4MDJlZDU4ZWEzZmNiZmE4NmYzOGNhNDFhMWVlMWI2ODI4NmIzYTFkMWFjZGQwYWU2MDM5OWRiMzIwZjY0ZCI6eyJpZCI6WyJsYW5nY2hhaW4iLCJzY2hlbWEiLCJtZXNzYWdlcyIsIlRvb2xNZXNzYWdlIl0sImt3YXJncyI6eyJjb250ZW50IjoiZWNobzoxIiwibmFtZSI6ImxlZ2FjeV9lY2hvIiwic3RhdHVzIjoic3VjY2VzcyIsInRvb2xfY2FsbF9pZCI6InRjLTEiLCJ0eXBlIjoidG9vbCJ9LCJsYyI6MSwidHlwZSI6ImNvbnN0cnVjdG9yIn19LCJtZXNzYWdlcyI6WyJ0dXJuIl19"
+            }
+          ]
+        },
+        "workflowRunTimeout": "0s",
+        "workflowTaskTimeout": "10s",
+        "continuedExecutionRunId": "01a08ed3-ce0b-7514-a4a7-0fb1fc20592b",
+        "initiator": "CONTINUE_AS_NEW_INITIATOR_WORKFLOW",
+        "originalExecutionRunId": "f5508d0e-4d4f-4c66-964c-9e9b1205f09f",
+        "firstExecutionRunId": "01a08ed3-ce0b-7514-a4a7-0fb1fc20592b",
+        "attempt": 1,
+        "firstWorkflowTaskBackoff": "0.550550s",
+        "prevAutoResetPoints": {
+          "points": [
+            {
+              "runId": "01a08ed3-ce0b-7514-a4a7-0fb1fc20592b",
+              "firstWorkflowTaskCompletedId": "4",
+              "createTime": "2026-09-11T04:57:21.088610Z",
+              "expireTime": "2026-09-12T04:57:21.100741Z",
+              "resettable": true,
+              "buildId": "28a196b25282242baeb8f6d083356dde"
+            }
+          ]
+        },
+        "workflowId": "lcan-5866553c-638c-47ff-85ca-f07589d77cbe",
+        "priority": {}
+      }
+    },
+    {
+      "eventId": "2",
+      "eventTime": "2026-09-11T04:57:22.096483Z",
+      "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED",
+      "taskId": "1048624",
+      "workflowTaskScheduledEventAttributes": {
+        "taskQueue": {
+          "name": "lcan",
+          "kind": "TASK_QUEUE_KIND_NORMAL"
+        },
+        "startToCloseTimeout": "10s",
+        "attempt": 1
+      }
+    },
+    {
+      "eventId": "3",
+      "eventTime": "2026-09-11T04:57:22.098235Z",
+      "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED",
+      "taskId": "1048627",
+      "workflowTaskStartedEventAttributes": {
+        "scheduledEventId": "2",
+        "identity": "54291@Davids-MacBook-Pro.local",
+        "requestId": "a5fc4f0a-3162-4b03-b55c-e4966d2ecfba",
+        "historySizeBytes": "769",
+        "workerVersion": {
+          "buildId": "28a196b25282242baeb8f6d083356dde"
+        }
+      }
+    },
+    {
+      "eventId": "4",
+      "eventTime": "2026-09-11T04:57:22.111140Z",
+      "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED",
+      "taskId": "1048631",
+      "workflowTaskCompletedEventAttributes": {
+        "scheduledEventId": "2",
+        "startedEventId": "3",
+        "identity": "54291@Davids-MacBook-Pro.local",
+        "workerVersion": {
+          "buildId": "28a196b25282242baeb8f6d083356dde"
+        },
+        "sdkMetadata": {
+          "coreUsedFlags": [
+            2,
+            1,
+            3
+          ],
+          "sdkName": "temporal-python",
+          "sdkVersion": "1.32.0"
+        },
+        "meteringMetadata": {}
+      }
+    },
+    {
+      "eventId": "5",
+      "eventTime": "2026-09-11T04:57:22.111208Z",
+      "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED",
+      "taskId": "1048632",
+      "workflowExecutionCompletedEventAttributes": {
+        "result": {
+          "payloads": [
+            {
+              "metadata": {
+                "encoding": "anNvbi9wbGFpbg=="
+              },
+              "data": "eyJtZXNzYWdlcyI6WyJ0dXJuIiwidHVybiJdLCJ0b2RvcyI6W3siY29udGVudCI6InciLCJzdGF0dXMiOiJjb21wbGV0ZWQifV19"
+            }
+          ]
+        },
+        "workflowTaskCompletedEventId": "4"
+      }
+    }
+  ]
+}