fix(agent): recover from orphaned tool_use blocks with a one-shot retry

agent/conversation_loop.py

@@ -2614,6 +2614,91 @@ def _perform_api_call(next_api_kwargs):
                     )
                     continue
 
+                # ── Orphaned tool_use recovery ─────────────────────────
+                # Anthropic rejects messages where a tool_use block is not
+                # immediately followed by a matching tool_result.  Two
+                # known causes:
+                #   1. Context compression inserts messages between a
+                #      tool_use and its tool_result.
+                #   2. A cron/subagent session is interrupted before the
+                #      tool_result is appended (e.g. execute_code blocked
+                #      by the approval guard).
+                # The canonical ``messages`` list uses OpenAI-style
+                # role=tool / tool_calls — not the Anthropic wire format.
+                # _strip_orphaned_tool_blocks operates on Anthropic-style
+                # api_messages, so we strip there and then signal the
+                # outer loop to rebuild api_messages from the cleaned
+                # canonical list by removing the orphaned tool_calls
+                # entries and their matching role=tool messages.
+                # One-shot to avoid an infinite strip loop.
+                if (
+                    classified.reason == FailoverReason.orphaned_tool_use
+                    and not _retry.orphaned_tool_use_retry_attempted
+                ):
+                    _retry.orphaned_tool_use_retry_attempted = True
+                    try:
+                        # Parse the orphaned tool_use IDs directly from the
+                        # Anthropic error message.  The error looks like:
+                        #   "messages.N: `tool_use` ids were found without
+                        #    `tool_result` blocks immediately after:
+                        #    toolu_xxx, toolu_yyy."
+                        # We cannot rely on detecting orphans from the
+                        # canonical messages (OpenAI-style tool_calls) because
+                        # the pair IS present there — the adjacency breaks
+                        # during Anthropic adapter conversion, e.g. when
+                        # context compaction injects a synthetic user message
+                        # between an assistant tool_use and its tool_result.
+                        import re as _re
+                        _err_str = str(api_error or "")
+                        _orphaned_ids: set = set(
+                            _re.findall(r"toolu_[A-Za-z0-9]+", _err_str)
+                        )
+
+                        # Remove these IDs from canonical messages so the
+                        # next api_messages rebuild produces valid adjacency.
+                        _stripped_canonical = 0
+                        if _orphaned_ids:
+                            _i = 0
+                            while _i < len(messages):
+                                _cm = messages[_i]
+                                if not isinstance(_cm, dict):
+                                    _i += 1
+                                    continue
+                                if _cm.get("role") == "assistant" and isinstance(_cm.get("tool_calls"), list):
+                                    _kept = [
+                                        tc for tc in _cm["tool_calls"]
+                                        if tc.get("id") not in _orphaned_ids
+                                    ]
+                                    if len(_kept) != len(_cm["tool_calls"]):
+                                        _stripped_canonical += len(_cm["tool_calls"]) - len(_kept)
+                                        if _kept:
+                                            _cm["tool_calls"] = _kept
+                                        else:
+                                            _cm.pop("tool_calls", None)
+                                if _cm.get("role") == "tool" and _cm.get("tool_call_id") in _orphaned_ids:
+                                    messages.pop(_i)
+                                    _stripped_canonical += 1
+                                    continue
+                                _i += 1
+                    except Exception as _strip_exc:
+                        logger.warning(
+                            "%sOrphaned tool_use recovery: strip failed: %s",
+                            agent.log_prefix, _strip_exc,
+                        )
+                        _orphaned_ids = set()
+                        _stripped_canonical = 0
+                    agent._vprint(
+                        f"{agent.log_prefix}⚠️  Orphaned tool_use detected — "
+                        f"stripped {len(_orphaned_ids)} id(s) from api_messages "
+                        f"and {_stripped_canonical} canonical entry/entries, retrying...",
+                        force=True,
+                    )
+                    logger.warning(
+                        "%sOrphaned tool_use recovery: stripped ids=%s canonical_entries=%d",
+                        agent.log_prefix, _orphaned_ids, _stripped_canonical,
+                    )
+                    continue
+
                 # ── llama.cpp grammar-parse recovery ──────────────────
                 # llama.cpp's ``json-schema-to-grammar`` converter rejects
                 # regex escape classes (``\d``, ``\w``, ``\s``) and most

agent/error_classifier.py

@@ -52,6 +52,7 @@ class FailoverReason(enum.Enum):
     # Request format
     format_error = "format_error"        # 400 bad request — abort or strip + retry
     invalid_encrypted_content = "invalid_encrypted_content"  # Responses replay blob rejected — strip replay state and retry
+    orphaned_tool_use = "orphaned_tool_use"  # tool_use block has no adjacent tool_result — strip orphans and retry
     multimodal_tool_content_unsupported = "multimodal_tool_content_unsupported"  # Provider rejected list-type content in tool messages (e.g. Xiaomi MiMo) — downgrade to text and retry
 
     # Provider-specific
@@ -1096,6 +1097,18 @@ def _classify_400(
             should_compress=True,
         )
 
+    # Anthropic rejects messages where a tool_use block is not immediately
+    # followed by a matching tool_result.  This can happen when:
+    #   • context compression inserts messages between the pair, or
+    #   • a cron/subagent session is interrupted before the tool_result
+    #     is appended (e.g. execute_code blocked by the approval guard).
+    # Recovery: strip orphaned tool_use/tool_result blocks and retry once.
+    if "tool_use" in error_msg and "tool_result" in error_msg:
+        return result_fn(
+            FailoverReason.orphaned_tool_use,
+            retryable=True,
+        )
+
     # Non-retryable format error
     return result_fn(
         FailoverReason.format_error,

agent/turn_retry_state.py

@@ -49,6 +49,7 @@ class TurnRetryState:
     # ── Format / payload recovery guards ─────────────────────────────────
     thinking_sig_retry_attempted: bool = False
     invalid_encrypted_content_retry_attempted: bool = False
+    orphaned_tool_use_retry_attempted: bool = False
     image_shrink_retry_attempted: bool = False
     multimodal_tool_content_retry_attempted: bool = False
     oauth_1m_beta_retry_attempted: bool = False

tests/run_agent/test_orphaned_tool_use_recovery.py

@@ -0,0 +1,235 @@
+"""Stress tests for orphaned tool_use recovery (PR #53236).
+
+Tests three layers:
+1. error_classifier: detects the Anthropic 400 and emits orphaned_tool_use
+2. _strip_orphaned_tool_blocks: actually cleans the canonical messages list
+3. TurnRetryState: one-shot guard prevents infinite retry loops
+"""
+
+from __future__ import annotations
+
+import pytest
+from agent.error_classifier import FailoverReason, _classify_400
+from agent.anthropic_adapter import _strip_orphaned_tool_blocks
+from agent.turn_retry_state import TurnRetryState
+
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+def _result_fn(reason, **kwargs):
+    """Minimal stand-in for the result_fn closure in classify_api_error."""
+    from agent.error_classifier import ClassifiedError
+    obj = ClassifiedError(reason=reason, status_code=400)
+    for k, v in kwargs.items():
+        setattr(obj, k, v)
+    return obj
+
+
+def _classify(msg: str):
+    body = {"type": "error", "error": {"type": "invalid_request_error", "message": msg}}
+    return _classify_400(
+        msg.lower(), "", body,
+        provider="anthropic", model="claude-sonnet-4-6",
+        approx_tokens=1000, context_length=200_000,
+        num_messages=10, result_fn=_result_fn,
+    )
+
+
+def _make_messages(*, orphaned: bool = True):
+    """Build a minimal canonical messages list.
+
+    With orphaned=True:  assistant has tool_use with no following tool_result
+    With orphaned=False: well-formed pair
+    """
+    tool_use_block = {
+        "type": "tool_use",
+        "id": "toolu_017QgFu6YZ8WbbMGDb1Z9FtP",
+        "name": "execute_code",
+        "input": {"code": "print('hi')"},
+    }
+    tool_result_msg = {
+        "role": "user",
+        "content": [
+            {
+                "type": "tool_result",
+                "tool_use_id": "toolu_017QgFu6YZ8WbbMGDb1Z9FtP",
+                "content": "Blocked by approval guard",
+            }
+        ],
+    }
+    messages = [
+        {"role": "user", "content": "run some code"},
+        {"role": "assistant", "content": [{"type": "text", "text": "Sure!"}, tool_use_block]},
+    ]
+    if not orphaned:
+        messages.append(tool_result_msg)
+        messages.append({"role": "assistant", "content": [{"type": "text", "text": "Done."}]})
+    return messages
+
+
+# ── Layer 1: error classifier ─────────────────────────────────────────────────
+
+ANTHROPIC_ERROR_MESSAGES = [
+    # Exact wording from real Anthropic API response
+    "messages.2: `tool_use` ids were found without `tool_result` blocks immediately after: "
+    "toolu_017QgFu6YZ8WbbMGDb1Z9FtP. Each `tool_use` block must have a corresponding "
+    "`tool_result` block in the next message.",
+    # Shorter variant
+    "tool_use ids found without tool_result blocks",
+    # Multi-tool variant
+    "tool_use ids toolu_abc toolu_def were found without tool_result blocks",
+]
+
+
+@pytest.mark.parametrize("msg", ANTHROPIC_ERROR_MESSAGES)
+def test_classifier_detects_orphaned_tool_use(msg):
+    result = _classify(msg)
+    assert result.reason == FailoverReason.orphaned_tool_use, (
+        f"Expected orphaned_tool_use, got {result.reason} for: {msg!r}"
+    )
+    assert result.retryable is True, "Must be retryable so the loop can recover"
+
+
+def test_classifier_does_not_trigger_on_unrelated_400():
+    result = _classify("invalid model name: claude-fake-99")
+    assert result.reason != FailoverReason.orphaned_tool_use
+
+
+def test_classifier_does_not_trigger_on_context_overflow():
+    # Simulate a large session that triggers context_overflow before our check
+    body = {"type": "error", "error": {"type": "invalid_request_error", "message": "error"}}
+    result = _classify_400(
+        "error", "", body,
+        provider="anthropic", model="claude-sonnet-4-6",
+        approx_tokens=90_000, context_length=200_000,
+        num_messages=90, result_fn=_result_fn,
+    )
+    assert result.reason == FailoverReason.context_overflow
+
+
+# ── Layer 2: strip function cleans canonical messages ─────────────────────────
+
+def test_strip_removes_orphaned_tool_use():
+    messages = _make_messages(orphaned=True)
+    assert len(messages) == 2  # user + orphaned assistant
+
+    _strip_orphaned_tool_blocks(messages)
+
+    # The assistant message should no longer contain the tool_use block
+    assistant_blocks = [
+        b for m in messages
+        if m.get("role") == "assistant"
+        for b in (m.get("content") if isinstance(m.get("content"), list) else [])
+        if isinstance(b, dict) and b.get("type") == "tool_use"
+    ]
+    assert assistant_blocks == [], f"Expected no tool_use blocks, found: {assistant_blocks}"
+
+
+def test_strip_leaves_well_formed_messages_intact():
+    messages = _make_messages(orphaned=False)
+    original_len = len(messages)
+    _strip_orphaned_tool_blocks(messages)
+    # Well-formed pair: tool_use + tool_result both present, nothing to strip
+    tool_uses = [
+        b for m in messages
+        if m.get("role") == "assistant"
+        for b in (m.get("content") if isinstance(m.get("content"), list) else [])
+        if isinstance(b, dict) and b.get("type") == "tool_use"
+    ]
+    assert tool_uses != [] or len(messages) == original_len, (
+        "Well-formed pair should survive the strip"
+    )
+
+
+def test_strip_is_idempotent():
+    """Calling strip twice should not crash or over-strip."""
+    messages = _make_messages(orphaned=True)
+    _strip_orphaned_tool_blocks(messages)
+    snapshot = [m.copy() for m in messages]
+    _strip_orphaned_tool_blocks(messages)
+    assert messages == snapshot, "Second strip changed messages"
+
+
+def test_strip_handles_empty_list():
+    messages = []
+    _strip_orphaned_tool_blocks(messages)  # must not raise
+    assert messages == []
+
+
+def test_strip_handles_no_tool_use():
+    messages = [
+        {"role": "user", "content": "hello"},
+        {"role": "assistant", "content": [{"type": "text", "text": "hi"}]},
+    ]
+    _strip_orphaned_tool_blocks(messages)
+    assert len(messages) == 2
+
+
+# ── Layer 3: TurnRetryState one-shot guard ────────────────────────────────────
+
+def test_retry_state_has_orphaned_tool_use_flag():
+    state = TurnRetryState()
+    assert hasattr(state, "orphaned_tool_use_retry_attempted"), (
+        "TurnRetryState must have orphaned_tool_use_retry_attempted field"
+    )
+    assert state.orphaned_tool_use_retry_attempted is False
+
+
+def test_retry_state_flag_prevents_second_recovery():
+    """Simulate the guard: second HTTP 400 should NOT trigger recovery again."""
+    state = TurnRetryState()
+
+    # First recovery fires
+    assert not state.orphaned_tool_use_retry_attempted
+    state.orphaned_tool_use_retry_attempted = True
+
+    # Second time: guard is already set — recovery must NOT fire
+    assert state.orphaned_tool_use_retry_attempted is True
+
+
+def test_retry_state_fresh_instance_resets_flag():
+    """Each new API call attempt gets a fresh TurnRetryState."""
+    state1 = TurnRetryState()
+    state1.orphaned_tool_use_retry_attempted = True
+
+    state2 = TurnRetryState()
+    assert state2.orphaned_tool_use_retry_attempted is False, (
+        "New TurnRetryState must start clean"
+    )
+
+
+# ── Integration: full recovery simulation ────────────────────────────────────
+
+def test_full_recovery_flow():
+    """Simulate the complete recovery path end-to-end without a real API call."""
+    # 1. Build a broken canonical messages list (cron job interrupted mid-tool)
+    messages = _make_messages(orphaned=True)
+
+    # 2. Verify the classifier recognizes the error
+    error_msg = (
+        "messages.2: `tool_use` ids were found without `tool_result` blocks immediately after: "
+        "toolu_017QgFu6YZ8WbbMGDb1Z9FtP."
+    )
+    classified = _classify(error_msg)
+    assert classified.reason == FailoverReason.orphaned_tool_use
+    assert classified.retryable
+
+    # 3. Verify guard not yet set
+    retry_state = TurnRetryState()
+    assert not retry_state.orphaned_tool_use_retry_attempted
+
+    # 4. Run the recovery (as conversation_loop.py would)
+    retry_state.orphaned_tool_use_retry_attempted = True
+    _strip_orphaned_tool_blocks(messages)
+
+    # 5. Verify messages are now clean
+    remaining_tool_uses = [
+        b for m in messages
+        if m.get("role") == "assistant"
+        for b in (m.get("content") if isinstance(m.get("content"), list) else [])
+        if isinstance(b, dict) and b.get("type") == "tool_use"
+    ]
+    assert remaining_tool_uses == [], "After recovery, no orphaned tool_use blocks should remain"
+
+    # 6. Verify a second identical error would NOT trigger recovery again
+    assert retry_state.orphaned_tool_use_retry_attempted is True