fix(agent): make sanitize_api_messages a fixpoint so dedup cannot re-wedge sessions (#83312)

agent/agent_runtime_helpers.py

@@ -3272,6 +3272,51 @@ def repair_empty_non_final_messages(
     return messages
 
 
+def drop_empty_tool_calls_arrays(
+    messages: List[Dict[str, Any]],
+) -> List[Dict[str, Any]]:
+    """Strip empty / malformed ``tool_calls`` from assistant messages.
+
+    An assistant message carrying ``tool_calls: []`` (an empty array) — or a
+    non-list value under the key — is semantically identical to an assistant
+    message with no tool calls, but strict OpenAI-compatible providers reject
+    the empty array outright: DeepSeek v4 returns HTTP 400 "Invalid
+    'messages[N].tool_calls': empty array. Expected an array with minimum
+    length 1, but got an empty array instead." (#58755, follow-up to #56980).
+
+    Empty arrays reach here from session resume, host-fed histories, the
+    consecutive-assistant merge in ``repair_message_sequence`` (which preserves
+    a pre-existing ``[]`` on the surviving turn), and — the reason this lives in
+    a reusable helper rather than inline — from ``sanitize_api_messages``' own
+    later dedup pass, which can empty an array it did not create (#83312).
+
+    Per the #56980 review this normalization belongs on the per-call copy, not
+    in ``repair_message_sequence``, which would destructively rewrite the
+    persisted trajectory. Shallow-copy the message before dropping the key so
+    stored history (and prompt caching) stays byte-stable.
+    """
+    normalized: List[Dict[str, Any]] = []
+    dropped = 0
+    for msg in messages:
+        if (
+            isinstance(msg, dict)
+            and msg.get("role") == "assistant"
+            and "tool_calls" in msg
+            and not (isinstance(msg["tool_calls"], list) and msg["tool_calls"])
+        ):
+            msg = {k: v for k, v in msg.items() if k != "tool_calls"}
+            dropped += 1
+        normalized.append(msg)
+    if dropped:
+        _ra().logger.debug(
+            "Pre-call sanitizer: dropped empty/invalid tool_calls on %d "
+            "assistant message(s)",
+            dropped,
+        )
+        return normalized
+    return messages
+
+
 def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     """Fix orphaned tool_call / tool_result pairs before every LLM call.
 
@@ -3302,39 +3347,7 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
     messages = repair_empty_non_final_messages(messages)
 
     # --- Drop empty / malformed tool_calls arrays on assistant messages ---
-    # An assistant message carrying ``tool_calls: []`` (an empty array) — or a
-    # non-list value under the key — is semantically identical to an assistant
-    # message with no tool calls, but strict OpenAI-compatible providers reject
-    # the empty array outright: DeepSeek v4 returns HTTP 400 "Invalid
-    # 'messages[N].tool_calls': empty array. Expected an array with minimum
-    # length 1, but got an empty array instead." (#58755, follow-up to #56980).
-    # Empty arrays reach here from session resume, host-fed histories, or the
-    # consecutive-assistant merge in ``repair_message_sequence`` (which
-    # preserves a pre-existing ``[]`` on the surviving turn). This is the final
-    # pre-API chokepoint, so normalize defensively — and, per the #56980
-    # review, do it HERE on the per-call copy rather than in
-    # ``repair_message_sequence``, which would destructively rewrite the
-    # persisted trajectory. Shallow-copy the message before dropping the key so
-    # stored history (and prompt caching) stays byte-stable.
-    normalized: List[Dict[str, Any]] = []
-    dropped_empty_tool_calls = 0
-    for msg in messages:
-        if (
-            isinstance(msg, dict)
-            and msg.get("role") == "assistant"
-            and "tool_calls" in msg
-            and not (isinstance(msg["tool_calls"], list) and msg["tool_calls"])
-        ):
-            msg = {k: v for k, v in msg.items() if k != "tool_calls"}
-            dropped_empty_tool_calls += 1
-        normalized.append(msg)
-    if dropped_empty_tool_calls:
-        messages = normalized
-        _ra().logger.debug(
-            "Pre-call sanitizer: dropped empty/invalid tool_calls on %d "
-            "assistant message(s)",
-            dropped_empty_tool_calls,
-        )
+    messages = drop_empty_tool_calls_arrays(messages)
 
     # --- Repair tool_calls whose function.name is empty/missing ---
     # Some providers (and partially-streamed responses) emit a tool_call with
@@ -3476,6 +3489,30 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
             "Pre-call sanitizer: removed %d duplicate tool_call_id reference(s)",
             removed_dupes,
         )
+        # --- Re-establish the invariants the passes above already enforced ---
+        # This sanitizer is only useful if it is a fixpoint: every invariant it
+        # claims must still hold on the value it returns. The dedup pass breaks
+        # that. When EVERY call on an assistant turn is a duplicate (the normal
+        # shape after ``repair_message_sequence`` merges consecutive assistant
+        # turns and unions their call lists onto the text turn), ``kept_tcs``
+        # is empty and the turn is rewritten to ``tool_calls: []`` — re-creating
+        # the exact payload the empty-array pass deleted a few steps earlier,
+        # after that pass can no longer see it. DeepSeek then 400s on every
+        # send, and because the poisoned turn is persisted the session is
+        # wedged permanently (#83312).
+        #
+        # Healing at this one site would only cover the empty-array half. A
+        # collapsed turn that carried no text also comes back with empty
+        # content, which is a second, independent 400 ("messages must have
+        # non-empty content"). So re-run both invariant passes over the deduped
+        # list instead: cheap (only on the rare dedup path), idempotent, and it
+        # covers both of those invariants no matter where above they were
+        # broken. It does NOT cover every possible future pass — a new pass
+        # inserted before this return would need the same re-run treatment.
+        # Order matters: drop the empty arrays first so the content healer
+        # sees the turn as genuinely payload-less and substitutes a placeholder.
+        messages = drop_empty_tool_calls_arrays(messages)
+        messages = repair_empty_non_final_messages(messages)
     return messages
 
 
@@ -4076,6 +4113,7 @@ def force_close_tcp_sockets(client: Any) -> int:
     "invoke_tool",
     "repair_tool_call",
     "sanitize_api_messages",
+    "drop_empty_tool_calls_arrays",
     "looks_like_codex_intermediate_ack",
     "copy_reasoning_content_for_api",
     "cleanup_dead_connections",

tests/run_agent/test_message_sequence_repair.py

@@ -380,3 +380,86 @@ def test_sanitize_drops_empty_tool_calls_array():
 
 
 
+
+
+# ── Sanitizer fixpoint: dedup must not re-break earlier invariants ──────────
+# The tool_call_id dedup pass runs AFTER the empty-array and empty-content
+# passes. When every call on an assistant turn is a duplicate, dedup rewrites
+# that turn to ``tool_calls: []`` — re-creating the exact payload the earlier
+# pass deleted, where nothing downstream can see it. DeepSeek 400s on every
+# subsequent send and the persisted turn wedges the session (#83312).
+
+
+def _wedge_transcript(text: str | None) -> list[dict]:
+    """Transcript whose last assistant turn re-uses an already-seen call id.
+
+    This is the shape ``repair_message_sequence`` produces when it merges two
+    consecutive assistant turns and unions their tool_calls onto the survivor.
+    """
+    call = {
+        "id": "call_dup",
+        "type": "function",
+        "function": {"name": "read_file", "arguments": "{}"},
+    }
+    return [
+        {"role": "user", "content": "hi"},
+        {"role": "assistant", "content": None, "tool_calls": [call]},
+        {"role": "tool", "tool_call_id": "call_dup", "content": "file body"},
+        {"role": "assistant", "content": text, "tool_calls": [dict(call)]},
+        {"role": "user", "content": "and now?"},
+    ]
+
+
+def test_dedup_does_not_reintroduce_empty_tool_calls_array():
+    """Collapsing every call on a turn must drop the key, not leave ``[]``."""
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    out = sanitize_api_messages(_wedge_transcript("here is the file"))
+
+    assert not any(
+        m.get("role") == "assistant" and m.get("tool_calls") == [] for m in out
+    ), "dedup re-introduced the empty tool_calls array DeepSeek rejects"
+    survivor = [m for m in out if m.get("content") == "here is the file"][0]
+    assert "tool_calls" not in survivor
+
+
+def test_dedup_collapse_heals_contentless_turn():
+    """A collapsed turn that carried no text must not be sent with empty
+    content either — that is a second, independent 400 ("messages must have
+    non-empty content"), so the content healer has to re-run after dedup."""
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    out = sanitize_api_messages(_wedge_transcript(""))
+
+    assistants = [m for m in out if m.get("role") == "assistant"]
+    assert all(
+        m.get("content") or m.get("tool_calls") for m in assistants[:-1]
+    ), "a non-final assistant turn survived with neither content nor tool_calls"
+    assert not any(m.get("tool_calls") == [] for m in assistants)
+
+
+def test_sanitize_is_a_fixpoint_over_the_wedge_transcript():
+    """Sanitizing twice must equal sanitizing once. Any invariant a pass
+    enforces has to still hold on the value the function returns, otherwise a
+    later pass can silently undo an earlier one."""
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    for text in ("here is the file", "", None):
+        once = sanitize_api_messages(_wedge_transcript(text))
+        twice = sanitize_api_messages([dict(m) for m in once])
+        assert once == twice, f"sanitizer not idempotent for content={text!r}"
+
+        surviving_call_ids = {
+            call.get("id")
+            for m in once
+            if m.get("role") == "assistant"
+            for call in (m.get("tool_calls") or [])
+        }
+        orphaned = [
+            m
+            for m in once
+            if m.get("role") == "tool" and m.get("tool_call_id") not in surviving_call_ids
+        ]
+        assert not orphaned, (
+            f"dedup left orphaned tool result(s) with no matching call: {orphaned!r}"
+        )