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