fix(runner): never re-execute an already-run tool call in SessionToolRunner

src/anthropic/lib/tools/_beta_session_runner.py

@@ -202,8 +202,12 @@ class DispatchedToolCall:
     posted: bool = True
     """``True`` if the result event made it to the session. ``False`` if all
     retries were exhausted or the server returned a permanent 4xx — in which
-    case the session-side agent will *not* see this result and the consumer may
-    want to surface that or retry at a higher level — and also ``False``, with
+    case the session-side agent will *not yet* see this result. The runner
+    keeps retrying the post (never re-running the tool) on every later
+    reconcile pass until it succeeds or the session ends, so the consumer may
+    see a later :class:`DispatchedToolCall` for the same ``tool_use_id`` with
+    ``posted=True`` — but may still want to surface a ``False`` here, since
+    there's no bound on how long that can take. Also ``False``, with
     ``result`` left ``None``, when the tool name is not one this runner owns and
     it deliberately posted nothing, leaving the ``tool_use_id`` pending for its
     owner (the split-client partial-fulfilment behavior)."""
@@ -391,8 +395,14 @@ async def _run(self) -> AsyncIterator[AsyncIterator[DispatchedToolCall]]:
         # ``_seen`` dedups tool-call events across the stream and the reconcile
         # pass (by event id); ``_answered`` holds the ids whose result post has
         # actually landed, so a failed post is retried on the next reconcile.
+        # ``_executed`` holds the computed (unconfirmed) result for an id whose
+        # tool has already run: reconcile re-enqueues anything not yet in
+        # ``_answered``, and a call already in ``_executed`` must only have its
+        # result re-posted, never re-run the tool itself (see ``_dispatch_loop``
+        # / ``_resend``).
         self._seen: set[str] = set()
         self._answered: set[str] = set()
+        self._executed: dict[str, tuple[DispatchedToolResultParams, bool]] = {}
         self._stop = anyio.Event()
         self._idle_clock = _IdleClock()
 
@@ -567,7 +577,16 @@ async def _dispatch_loop(self) -> None:
                 # an in-flight tool. The result will still be posted and the
                 # DispatchedToolCall enqueued before the cancel propagates.
                 with anyio.CancelScope(shield=True):
-                    await self._execute(ev)
+                    if ev.id in self._executed:
+                        # Reconcile re-enqueued a call whose tool already ran
+                        # (the earlier post failed or was never confirmed) —
+                        # retry posting the result we already have. Never call
+                        # _execute again: that would re-run the tool itself,
+                        # which is unsafe for a side-effecting tool (bash, a
+                        # file write).
+                        await self._resend(ev)
+                    else:
+                        await self._execute(ev)
         finally:
             # Closing the results stream signals the iterator that no more
             # results will arrive. Wrapped in a shield because we're often in a
@@ -619,7 +638,36 @@ async def _execute(self, ev: DispatchedToolUseEvent) -> None:
                 content = tool_error_content(e)
                 is_error = True
             tool_result = _build_result_event(ev, content, is_error)
+            # Recorded before the send attempt: the tool has now run, so a
+            # later reconcile must never dispatch this id through _execute
+            # again, regardless of whether the send below succeeds.
+            self._executed[ev.id] = (tool_result, is_error)
             sent = await self._send_result(tool_result, ev.id)
+            if sent:
+                self._executed.pop(ev.id, None)
+        await self._post_result(ev, tool_result, is_error, sent)
+
+    async def _resend(self, ev: DispatchedToolUseEvent) -> None:
+        """Retry posting an already-computed result without re-running the tool.
+
+        Reached from ``_dispatch_loop`` when reconcile re-enqueues a call whose
+        tool already ran (via ``_execute``) but whose result was not yet
+        confirmed posted.
+        """
+        tool_result, is_error = self._executed[ev.id]
+        sent = await self._send_result(tool_result, ev.id)
+        if sent:
+            self._executed.pop(ev.id, None)
+        await self._post_result(ev, tool_result, is_error, sent)
+
+    async def _post_result(
+        self,
+        ev: DispatchedToolUseEvent,
+        tool_result: DispatchedToolResultParams | None,
+        is_error: bool,
+        sent: bool,
+    ) -> None:
+        """Surface a completed dispatch (fresh or retried) to the consumer."""
         try:
             await self._send_results.send(
                 DispatchedToolCall(

tests/lib/tools/test_session_runner.py

@@ -787,6 +787,59 @@ async def increment(_input: dict[str, Any]) -> str:
     assert counter["calls"] == 0
 
 
+@pytest.mark.asyncio()
+async def test_reconnect_does_not_re_execute_tool_after_failed_send(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A tool whose result-post permanently failed must not be re-executed when
+    a later stream reconnect re-runs ``_reconcile``.
+
+    ``_reconcile`` re-enqueues every ``agent.tool_use`` not yet in
+    ``_answered`` — which is exactly the state of a call whose ``_send_result``
+    exhausted its retries or hit a permanent 4xx (mirroring
+    ``test_yields_with_posted_false_on_permanent_4xx``). Every *other* source of
+    duplicate delivery this file guards against (the live stream re-seeing an
+    already-answered id, a partial list-pagination re-read) is a duplicate of
+    already-settled work. This one is not: nothing has told the runner the
+    result landed, so on the next reconnect it dispatches the same
+    ``agent.tool_use`` again — and ``_execute`` unconditionally re-runs the
+    tool, not just the post. For a side-effecting tool (``bash``, a file
+    write) that is a real correctness hazard: at-least-once instead of
+    at-most-once execution.
+    """
+    monkeypatch.setattr(session_runner_mod, "STREAM_BACKOFF_START", 0.001)
+    counter = {"calls": 0}
+
+    async def increment(_input: dict[str, Any]) -> str:
+        counter["calls"] += 1
+        return "done"
+
+    tool = _FakeTool("inc", increment)
+    # Every reconcile pass (both connections) sees the same still-unanswered
+    # agent.tool_use — the server's own record, since the first send attempt
+    # below permanently fails and the runner never posts a result. The second
+    # send attempt (the reconcile-triggered retry) has no scripted failure, so
+    # it succeeds.
+    events = FakeAsyncEvents(
+        list_events=[_tool_use("tu_1", "inc", {})],
+        # First connection: a filler event (so `raise_after` has something to
+        # count past) then a transient disconnect, forcing a reconnect and a
+        # second `_reconcile()` pass. Second connection: ends the run.
+        streams=[
+            _FakeStream([_StubEvent("noop")], raise_after=1, raise_with=_api_status_error(500)),
+            _FakeStream([_terminated()]),
+        ],
+        send_failures=[_api_status_error(400)],
+    )
+
+    items = [item async for item in _run_with_fakes(events=events, tools=[tool])]
+
+    assert counter["calls"] == 1, "the tool must not be re-executed once it has already run for this tool_use_id"
+    # The first yield reports the failed post; the retry on the second
+    # reconcile succeeds without recomputing the tool's result.
+    assert [item.posted for item in items] == [False, True]
+    assert all(item.tool_use_id == "tu_1" and _result_text(item) == "done" for item in items)
+    assert len(events.send_calls) == 2
+
+
 # ---------- environment-key auth -----------------------------------------