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

src/anthropic/lib/tools/_beta_session_runner.py

@@ -487,6 +487,12 @@ async def _run(self) -> AsyncIterator[AsyncIterator[DispatchedToolCall]]:
         # by :meth:`_note_confirmation` / the next reconcile pass. Like ``_seen``
         # and ``_answered``, ``_confirmations`` is per-session O(tool calls):
         # recorded verdicts persist for the life of the run.
+        # ``_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._executed: dict[str, tuple[DispatchedToolResultParams, bool, Literal["allow"] | None]] = {}
         self._confirmations: dict[str, Literal["allow", "deny"]] = {}
         self._awaiting_confirmation: dict[str, DispatchedToolUseEvent] = {}
         self._stop = anyio.Event()
@@ -775,6 +781,28 @@ async def _resolve_denied(self, ev: DispatchedToolUseEvent) -> None:
             )
         )
 
+    async def _resend(self, ev: DispatchedToolUseEvent) -> None:
+        """Retry posting an already-computed result without re-running the tool.
+
+        Reached from :meth:`_dispatch_loop` when reconcile re-enqueues a call
+        whose tool already ran but whose result was not yet confirmed posted.
+        """
+        tool_result, is_error, confirmation = self._executed[ev.id]
+        sent = await self._send_result(tool_result, ev.id)
+        if sent:
+            self._executed.pop(ev.id, None)
+        await self._surface_call(
+            DispatchedToolCall(
+                event=ev,
+                result=tool_result,
+                tool_use_id=ev.id,
+                name=ev.name,
+                is_error=is_error,
+                posted=sent,
+                confirmation=confirmation,
+            )
+        )
+
     async def _surface_call(self, call: DispatchedToolCall) -> None:
         """Yield ``call`` to the consumer, tolerating a consumer that left early.
 
@@ -807,7 +835,16 @@ async def _dispatch_loop(self) -> None:
                         # posted and the DispatchedToolCall enqueued before the
                         # cancel propagates.
                         with anyio.CancelScope(shield=True):
-                            await self._execute(ev, confirmation)
+                            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, which would
+                                # re-run the tool itself and is unsafe for a
+                                # side-effecting tool such as bash or a file write.
+                                await self._resend(ev)
+                            else:
+                                await self._execute(ev, confirmation)
                 finally:
                     if confirmation == "allow":
                         # The user-approved call is fully disposed of (executed,
@@ -872,7 +909,13 @@ async def _execute(self, ev: DispatchedToolUseEvent, confirmation: Literal["allo
                 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, confirmation)
             sent = await self._send_result(tool_result, ev.id)
+            if sent:
+                self._executed.pop(ev.id, None)
         await self._surface_call(
             DispatchedToolCall(
                 event=ev,

tests/lib/tools/test_session_runner.py

@@ -1595,3 +1595,48 @@ def test_to_session_content_tool_reference_stringified() -> None:
     block = {"type": "tool_reference", "tool_name": "weather"}
     out = _to_session_content([block])
     assert out == [{"type": "text", "text": session_runner_mod.json.dumps(block)}]
+
+
+@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`` hit a permanent
+    4xx. The other duplicate-delivery sources this file guards against are
+    duplicates of already-settled work; this one is not, so the same event was
+    dispatched again and ``_execute`` re-ran the tool rather than just retrying
+    the post. For a side-effecting tool that is 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)
+    # Both reconcile passes see the same still-unanswered agent.tool_use, since
+    # the first send permanently fails. The retry on the second pass has no
+    # scripted failure, so it succeeds.
+    events = FakeAsyncEvents(
+        list_events=[_tool_use("tu_1", "inc", {})],
+        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 for the same tool_use_id"
+    # First yield reports the failed post; the reconcile-triggered retry succeeds
+    # without recomputing the 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