From 563415d70f35135c812efd63f28b995ae65d0713 Mon Sep 17 00:00:00 2001 From: Pranav Mishra Date: Mon, 3 Aug 2026 00:04:10 -0700 Subject: [PATCH] fix(runner): never re-execute an already-run tool call in SessionToolRunner `_reconcile()` re-enqueues every `agent.tool_use` / `agent.custom_tool_use` not yet in `_answered`, and runs on every stream reconnect. A call whose result post permanently failed (or exhausted SEND_RETRIES) is exactly that: still unanswered. On the next reconnect it was dispatched through `_execute` again, which re-runs the tool itself and not just the post, so a side-effecting tool such as bash or a file write executes twice for one `tool_use_id`. Add `_executed`, recording a call's computed result once its tool has run. `_dispatch_loop` checks it before `_execute`: an id already in `_executed` goes through the new `_resend`, which retries posting the cached result only. That keeps the existing self-healing behavior, where a failed post is retried on later reconciles, while removing the re-execution. Rebased onto main after the confirmation-gating refactor. `_executed` also carries the call's confirmation so a resent `DispatchedToolCall` reports the same verdict as the original, and `_resend` reuses the new `_surface_call` helper rather than duplicating the consumer-gone handling. Reproduced with the existing FakeAsyncEvents harness, no live API: a permanent 4xx on the first send followed by a stream reconnect ran the tool twice. Fixes #1749 --- .../lib/tools/_beta_session_runner.py | 45 ++++++++++++++++++- tests/lib/tools/test_session_runner.py | 45 +++++++++++++++++++ 2 files changed, 89 insertions(+), 1 deletion(-) diff --git a/src/anthropic/lib/tools/_beta_session_runner.py b/src/anthropic/lib/tools/_beta_session_runner.py index ec85da807..2f17692a5 100644 --- a/src/anthropic/lib/tools/_beta_session_runner.py +++ b/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, diff --git a/tests/lib/tools/test_session_runner.py b/tests/lib/tools/test_session_runner.py index e9d26d5dc..bdc093d43 100644 --- a/tests/lib/tools/test_session_runner.py +++ b/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