From 4032ade23be8f1455261b93aad93578fd665e524 Mon Sep 17 00:00:00 2001 From: qbit-mirror-bot Date: Thu, 2 Jul 2026 13:31:13 +0000 Subject: [PATCH] fix(agent): close orphaned assistant(tool_calls) in repair_message_sequence (#56980) --- agent/agent_runtime_helpers.py | 110 +++++++++++------- .../run_agent/test_message_sequence_repair.py | 109 +++++++++++++++++ 2 files changed, 174 insertions(+), 45 deletions(-) diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index 18ed3102c272..1e3c5d0b3dc0 100644 --- a/agent/agent_runtime_helpers.py +++ b/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:`` (see - # CUSTOM_POOL_PREFIX). Resolve the primary's base_url to its - # ``custom:`` 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 diff --git a/tests/run_agent/test_message_sequence_repair.py b/tests/run_agent/test_message_sequence_repair.py index 93e65193756d..6c970fcc05cd 100644 --- a/tests/run_agent/test_message_sequence_repair.py +++ b/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