fix(session runner): retry the sub-thread tool_use commit-race 400 instead of stranding (#1744)

src/anthropic/lib/tools/_beta_session_runner.py

@@ -27,6 +27,7 @@
 
 from .._retry import TRANSIENT_ERRORS, is_fatal_status_error
 from ..._types import Headers
+from ..._exceptions import APIStatusError
 from ._tool_dispatch import tool_registry, run_runnable_tool, tool_error_content
 from .._scoped_client import _copy_client_with_bearer_auth
 from ._beta_functions import (
@@ -105,6 +106,25 @@
 # tests/lib/tools/test_session_runner.py::test_tool_timeout_exceeds_bash_default.
 TOOL_TIMEOUT = 150.0
 SEND_RETRIES = 3
+# The sub-thread tool_use commit race (#1744): posting a tool_result for a tool that ran
+# on a non-primary ``session_thread_id`` can 400 with "does not match any agent.tool_use
+# event" until that originating ``agent.tool_use`` event becomes matchable server-side.
+# That 400 is transient (it clears once the event commits), not a permanent bad request, so
+# it gets its own, more patient retry budget instead of the generic fatal-4xx bailout below.
+# Matched on the message because the API exposes no distinct error code for it.
+SUBTHREAD_TOOL_USE_RACE_MARKER = "does not match any agent.tool_use event"
+SUBTHREAD_RACE_SEND_RETRIES = 8
+
+
+def _is_subthread_tool_use_race(err: Exception) -> bool:
+    """True for the transient #1744 sub-thread ``agent.tool_use`` commit-race 400."""
+    return (
+        isinstance(err, APIStatusError)
+        and err.status_code == 400
+        and SUBTHREAD_TOOL_USE_RACE_MARKER in str(getattr(err, "message", None) or err)
+    )
+
+
 # Grace period, in seconds, that the runner keeps running after the session goes
 # idle with stop_reason ``end_turn`` before it stops; any new event in that
 # window resets it. ``max_idle=None`` disables it (run until the session ends).
@@ -645,7 +665,9 @@ async def _send_result(self, tool_result: DispatchedToolResultParams, tool_use_i
         (``tool_use_id`` vs ``custom_tool_use_id``) depending on the kind.
         """
         last_err: Exception | None = None
-        for i in range(SEND_RETRIES):
+        attempt = 0
+        retries = SEND_RETRIES
+        while attempt < retries:
             try:
                 await self._events.send(
                     self.session_id,
@@ -656,11 +678,18 @@ async def _send_result(self, tool_result: DispatchedToolResultParams, tool_use_i
                 return True
             except TRANSIENT_ERRORS as e:
                 last_err = e
-                if is_fatal_status_error(e):
+                if _is_subthread_tool_use_race(e):
+                    # Transient commit race (#1744): the originating sub-thread
+                    # ``agent.tool_use`` event has not committed server-side yet. Keep
+                    # retrying on a more patient budget instead of bailing as fatal, so the
+                    # result lands once the event is matchable rather than stranding the call.
+                    retries = SUBTHREAD_RACE_SEND_RETRIES
+                elif is_fatal_status_error(e):
                     break
+                attempt += 1
                 # Don't sleep after the final attempt — there is no retry to wait for.
-                if i < SEND_RETRIES - 1:
-                    await anyio.sleep(i + 1)
+                if attempt < retries:
+                    await anyio.sleep(min(attempt, STREAM_BACKOFF_CAP))
         log.error("failed to send tool result tool_use_id=%s error=%s", tool_use_id, last_err)
         return False
 

tests/lib/tools/test_session_runner.py

@@ -111,6 +111,13 @@ def _api_status_error(code: int) -> APIStatusError:
     return APIStatusError("boom", response=response, body=None)
 
 
+def _subthread_race_400() -> APIStatusError:
+    # The transient #1744 400: a sub-thread agent.tool_use event not yet matchable.
+    request = httpx.Request("POST", "https://api.example/x")
+    response = httpx.Response(status_code=400, request=request, content=b"{}")
+    return APIStatusError("does not match any agent.tool_use event", response=response, body=None)
+
+
 class _FakeStream:
     """Stand-in for the AsyncStream returned by ``events.stream()``.
 
@@ -1013,3 +1020,72 @@ 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)}]
+
+
+# ---------- #1744: sub-thread tool_use commit-race 400 ----------------------
+
+
+@pytest.mark.asyncio()
+async def test_subthread_race_400_is_retried_not_fatal() -> None:
+    """The #1744 sub-thread commit-race 400 is transient and must be retried.
+
+    Before the fix ``is_fatal_status_error`` classified it as fatal, so the runner
+    bailed on the first attempt and left the tool call permanently stranded.
+    """
+
+    async def echo(_input: dict[str, Any]) -> str:
+        return "ok"
+
+    tool = _FakeTool("echo", echo)
+    events = FakeAsyncEvents(
+        stream_events=[_tool_use("tu_1", "echo", {"x": 1}), _terminated()],
+        send_failures=[_subthread_race_400(), None],
+    )
+
+    items = [item async for item in _run_with_fakes(events=events, tools=[tool])]
+
+    assert len(events.send_calls) == 2, "retried past the first race 400 instead of bailing"
+    assert items[0].posted is True
+
+
+@pytest.mark.asyncio()
+async def test_subthread_race_400_uses_extended_budget(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The race 400 keeps retrying past ``SEND_RETRIES`` (3) on its patient budget."""
+
+    async def _no_sleep(_seconds: float) -> None:
+        return None
+
+    monkeypatch.setattr(session_runner_mod.anyio, "sleep", _no_sleep)
+
+    async def echo(_input: dict[str, Any]) -> str:
+        return "ok"
+
+    tool = _FakeTool("echo", echo)
+    events = FakeAsyncEvents(
+        stream_events=[_tool_use("tu_1", "echo", {"x": 1}), _terminated()],
+        send_failures=[_subthread_race_400()] * 4 + [None],
+    )
+
+    items = [item async for item in _run_with_fakes(events=events, tools=[tool])]
+
+    assert len(events.send_calls) == 5, "retried the race 400 past SEND_RETRIES"
+    assert items[0].posted is True
+
+
+@pytest.mark.asyncio()
+async def test_non_race_400_still_fatal() -> None:
+    """A generic 400 (no race marker) must still bail immediately — no over-broadening."""
+
+    async def echo(_input: dict[str, Any]) -> str:
+        return "ok"
+
+    tool = _FakeTool("echo", echo)
+    events = FakeAsyncEvents(
+        stream_events=[_tool_use("tu_1", "echo", {"x": 1}), _terminated()],
+        send_failures=[_api_status_error(400)],
+    )
+
+    items = [item async for item in _run_with_fakes(events=events, tools=[tool])]
+
+    assert len(events.send_calls) == 1, "a non-race 400 must not be retried"
+    assert items[0].posted is False