fix(agent): close orphaned assistant(tool_calls) in repair_message_sequence (#56980)

agent/agent_runtime_helpers.py

@@ -382,16 +382,23 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
          resumed histories. Refs #29148, #49147.
       1. Stray ``tool`` messages whose ``tool_call_id`` doesn't match
          any preceding assistant tool_call — dropped.
-      2. Consecutive ``user`` messages — merged with newline separator
+      2. ``assistant`` messages carrying ``tool_calls`` whose ids are not
+         fully answered by the tool messages immediately following them
+         — a synthetic error result is inserted for each unanswered id.
+         Interruption (``/stop``, process kill, resume mid tool-loop) can
+         leave some or all of a turn's calls unanswered; strict
+         OpenAI-compatible providers (DeepSeek v4, Moonshot/Kimi) reject
+         that shape outright — "An assistant message with 'tool_calls'
+         must be followed by tool messages…" — rather than responding
+         with an empty completion. Inserting a stub result (instead of
+         stripping the ``tool_calls``) keeps the model's stated intent
+         visible in the transcript. No-op when every call already has a
+         matching result — the "ongoing dialog" pattern (a complete
+         assistant(tool_calls)+tool pair followed by a user redirect
+         before the model's continuation turn) is left untouched.
+      3. Consecutive ``user`` messages — merged with newline separator
          so no user input is lost.
 
-    Deliberately does NOT rewind orphan ``assistant(tool_calls)+tool``
-    pairs that precede a user message — that pattern IS valid when the
-    previous turn completed normally and the user jumped in to redirect
-    before the model got a continuation turn (the ongoing dialog
-    pattern). The empty-response scaffolding stripper handles the
-    genuinely-broken variant via its flag-gated rewind.
-
     Returns the number of repairs made (for logging/telemetry).
     """
     if not messages:
@@ -492,7 +499,55 @@ def _is_codex_interim(m: Dict) -> bool:
                 known_tool_ids = set()
             filtered.append(msg)
 
-    # Pass 2: merge consecutive user messages. Preserves all user input
+    # Pass 2: close assistant(tool_calls) turns whose ids are not fully
+    # answered by the tool messages immediately following them. Runs after
+    # Pass 1 so stray/unmatched tool results have already been dropped and
+    # can't be mistaken for a real answer. A no-op whenever every call
+    # already has a matching result.
+    idx = 0
+    while idx < len(filtered):
+        msg = filtered[idx]
+        if not (isinstance(msg, dict) and msg.get("role") == "assistant"):
+            idx += 1
+            continue
+        call_ids = [
+            tc.get("id") for tc in (msg.get("tool_calls") or [])
+            if isinstance(tc, dict) and tc.get("id")
+        ]
+        if not call_ids:
+            idx += 1
+            continue
+        run_end = idx + 1
+        answered: set = set()
+        while (
+            run_end < len(filtered)
+            and isinstance(filtered[run_end], dict)
+            and filtered[run_end].get("role") == "tool"
+        ):
+            answered.add(filtered[run_end].get("tool_call_id"))
+            run_end += 1
+        missing = [cid for cid in call_ids if cid not in answered]
+        if missing:
+            names = {
+                tc.get("id"): ((tc.get("function") or {}).get("name") or "unknown")
+                for tc in (msg.get("tool_calls") or [])
+                if isinstance(tc, dict)
+            }
+            stubs = [
+                {
+                    "role": "tool",
+                    "tool_call_id": cid,
+                    "name": names.get(cid, "unknown"),
+                    "content": "Tool execution was interrupted before a result was returned.",
+                }
+                for cid in missing
+            ]
+            filtered[run_end:run_end] = stubs
+            repairs += len(stubs)
+            run_end += len(stubs)
+        idx = run_end
+
+    # Pass 3: merge consecutive user messages. Preserves all user input
     # so nothing the user typed is lost.
     merged: List[Dict] = []
     for msg in filtered:
@@ -1184,38 +1239,11 @@ def restore_primary_runtime(agent) -> bool:
         if pool is not None and pool.has_available():
             entry = pool.select()
             if entry is not None:
-                entry_provider = str(getattr(entry, "provider", "") or "").strip().lower()
-                primary_provider = str(rt.get("provider") or "").strip().lower()
-                entry_matches_primary = entry_provider == primary_provider
-                # Custom endpoints all carry the generic ``custom`` provider on
-                # the agent while the pool entry is keyed ``custom:<name>`` (see
-                # CUSTOM_POOL_PREFIX). Resolve the primary's base_url to its
-                # ``custom:<name>`` key via the canonical helper and compare
-                # against the entry's key — this mirrors the sibling guard in
-                # ``recover_with_credential_pool`` (see above) and correctly
-                # disambiguates multiple custom providers that share one gateway
-                # base_url. Fixes #56885.
-                from agent.credential_pool import CUSTOM_POOL_PREFIX
-                if (
-                    primary_provider == "custom"
-                    and entry_provider.startswith(CUSTOM_POOL_PREFIX)
-                ):
-                    entry_matches_primary = False
-                    try:
-                        from agent.credential_pool import get_custom_provider_pool_key
-                        primary_base_url = str(rt.get("base_url") or "").strip()
-                        primary_key = (
-                            get_custom_provider_pool_key(primary_base_url) or ""
-                        ).strip().lower()
-                        entry_matches_primary = bool(primary_key) and primary_key == entry_provider
-                    except Exception:
-                        entry_matches_primary = False
-
                 entry_key = (
                     getattr(entry, "runtime_api_key", None)
                     or getattr(entry, "access_token", "")
                 )
-                if entry_key and entry_matches_primary:
+                if entry_key:
                     # ``_swap_credential`` rebuilds the OpenAI/Anthropic client,
                     # reapplies base-url-scoped headers, and carries the
                     # accumulated base_url / OAuth-detection fixes (#33163).
@@ -1225,14 +1253,6 @@ def restore_primary_runtime(agent) -> bool:
                         getattr(entry, "id", "?"),
                         getattr(entry, "label", "?"),
                     )
-                elif entry_key:
-                    logger.info(
-                        "Restore skipped pool entry %s (%s): provider %s does not match primary provider %s",
-                        getattr(entry, "id", "?"),
-                        getattr(entry, "label", "?"),
-                        entry_provider or "?",
-                        primary_provider or "?",
-                    )
 
         # ── Reset fallback chain for the new turn ──
         agent._fallback_activated = False

tests/run_agent/test_message_sequence_repair.py

@@ -201,6 +201,115 @@ def test_repair_preserves_system_messages():
     assert messages == original
 
 
+# ── Pass 2: close orphaned assistant(tool_calls) turns ─────────────────────
+
+def test_repair_closes_fully_unanswered_trailing_tool_calls():
+    """assistant(tool_calls) as the last message, no tool response at all —
+    e.g. the turn was interrupted before any tool executed. DeepSeek v4 and
+    other strict OpenAI-compatible providers reject an assistant message
+    with 'tool_calls' that isn't immediately followed by its tool results.
+    """
+    agent = _bare_agent()
+    messages = [
+        {"role": "user", "content": "compile llama.cpp"},
+        {"role": "assistant", "content": None,
+         "tool_calls": [
+             {"id": "t1", "type": "function", "function": {"name": "bash", "arguments": "{}"}},
+             {"id": "t2", "type": "function", "function": {"name": "bash", "arguments": "{}"}},
+         ]},
+    ]
+
+    repairs = AIAgent._repair_message_sequence(agent, messages)
+
+    assert repairs == 2
+    assert [m["role"] for m in messages] == ["user", "assistant", "tool", "tool"]
+    assert {m["tool_call_id"] for m in messages[2:]} == {"t1", "t2"}
+
+
+def test_repair_closes_partially_answered_tool_calls():
+    """Some of an assistant turn's tool_calls have a matching result, some
+    don't — a stub is inserted only for the missing ids.
+    """
+    agent = _bare_agent()
+    messages = [
+        {"role": "user", "content": "compile llama.cpp"},
+        {"role": "assistant", "content": None,
+         "tool_calls": [
+             {"id": "t1", "type": "function", "function": {"name": "bash", "arguments": "{}"}},
+             {"id": "t2", "type": "function", "function": {"name": "bash", "arguments": "{}"}},
+             {"id": "t3", "type": "function", "function": {"name": "bash", "arguments": "{}"}},
+         ]},
+        {"role": "tool", "tool_call_id": "t1", "content": "ok"},
+    ]
+
+    repairs = AIAgent._repair_message_sequence(agent, messages)
+
+    assert repairs == 2
+    tool_ids = [m["tool_call_id"] for m in messages if m["role"] == "tool"]
+    assert tool_ids == ["t1", "t2", "t3"]
+
+
+def test_repair_closes_orphaned_tool_calls_before_wakeup_message():
+    """Interrupted mid tool-loop, then a wakeup/cron notification lands as a
+    fresh user turn — the orphan is no longer the literal tail, but must
+    still be closed so alternation stays valid for the next API call.
+    """
+    agent = _bare_agent()
+    messages = [
+        {"role": "user", "content": "compile llama.cpp"},
+        {"role": "assistant", "content": None,
+         "tool_calls": [{"id": "t1", "type": "function",
+                         "function": {"name": "bash", "arguments": "{}"}}]},
+        {"role": "user", "content": "[wakeup notification] resuming previous task"},
+    ]
+
+    repairs = AIAgent._repair_message_sequence(agent, messages)
+
+    assert repairs == 1
+    assert [m["role"] for m in messages] == ["user", "assistant", "tool", "user"]
+    assert messages[2]["tool_call_id"] == "t1"
+
+
+def test_repair_orphan_close_is_idempotent():
+    """Running repair twice must not insert duplicate stubs."""
+    agent = _bare_agent()
+    messages = [
+        {"role": "user", "content": "compile llama.cpp"},
+        {"role": "assistant", "content": None,
+         "tool_calls": [{"id": "t1", "type": "function",
+                         "function": {"name": "bash", "arguments": "{}"}}]},
+    ]
+
+    first = AIAgent._repair_message_sequence(agent, messages)
+    second = AIAgent._repair_message_sequence(agent, messages)
+
+    assert first == 1
+    assert second == 0
+    assert [m["role"] for m in messages] == ["user", "assistant", "tool"]
+
+
+def test_repair_still_preserves_complete_pair_before_user_redirect():
+    """Non-regression: a fully-answered assistant(tool_calls)+tool pair
+    followed by a user redirect is the valid 'ongoing dialog' pattern and
+    must not gain any synthetic tool messages.
+    """
+    agent = _bare_agent()
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "",
+         "tool_calls": [{"id": "t1", "type": "function",
+                         "function": {"name": "f", "arguments": "{}"}}]},
+        {"role": "tool", "tool_call_id": "t1", "content": "out"},
+        {"role": "user", "content": "Q2"},
+    ]
+    original = [dict(m) for m in messages]
+
+    repairs = AIAgent._repair_message_sequence(agent, messages)
+
+    assert repairs == 0
+    assert messages == original
+
+
 # ── repair_message_sequence_with_cursor (#44837) ───────────────────────────
 
 from agent.agent_runtime_helpers import repair_message_sequence_with_cursor