From 57f46e2bef99675a2405a51d495de44bfefe5ecb Mon Sep 17 00:00:00 2001 From: Pranav Mishra Date: Fri, 10 Jul 2026 18:58:11 -0700 Subject: [PATCH] fix(runner): never re-execute an already-run tool call in SessionToolRunner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _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 unconditionally re-runs the tool itself, not just the post — an at-least-once instead of at-most-once execution hazard for any side-effecting tool (bash, a file write). Add _executed, recording a call's computed result once its tool has run. _dispatch_loop now checks it before _execute: an id already in _executed goes through the new _resend (retry posting the cached result only) instead of _execute (which would run the tool again). This preserves the existing self-healing intent — a failed post keeps getting retried on later reconciles — while removing the unsafe re-execution. Reproduced deterministically with the existing FakeAsyncEvents harness (no live API needed): a permanent 4xx on the first send, followed by a stream reconnect, previously ran the tool twice for the same tool_use_id. Fixes #1749 Co-Authored-By: Claude Opus 4.8 --- .../lib/tools/_beta_session_runner.py | 54 +++++++++++++++++-- tests/lib/tools/test_session_runner.py | 53 ++++++++++++++++++ 2 files changed, 104 insertions(+), 3 deletions(-) diff --git a/src/anthropic/lib/tools/_beta_session_runner.py b/src/anthropic/lib/tools/_beta_session_runner.py index 96d943f5c..22cb8a057 100644 --- a/src/anthropic/lib/tools/_beta_session_runner.py +++ b/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( diff --git a/tests/lib/tools/test_session_runner.py b/tests/lib/tools/test_session_runner.py index d36c67ed8..72c80d5d6 100644 --- a/tests/lib/tools/test_session_runner.py +++ b/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 -----------------------------------------