Merging consecutive assistant tool_calls messages during session replay

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #55603 fix(agent): merge consecutive assistant messages before API replay (#29148, #49147)

  merged to main

  files: agent/agent_runtime_helpers.py, tests/run_agent/test_message_sequence_repair.py

  Show the diff

  • agent/agent_runtime_helpers.py

    @@ -368,6 +368,18 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
         host code) can feed in already-broken histories.
     
         Repairs applied:
    +      0. Consecutive ``assistant`` messages with no intervening
    +         ``tool``/``user`` turn — merged into a single assistant turn
    +         (union of ``tool_calls``, concatenated ``content``). Strict
    +         OpenAI-compatible providers (DeepSeek v4, Moonshot/Kimi) reject
    +         a history where an ``assistant`` message carrying ``tool_calls``
    +         is immediately followed by another ``assistant`` message instead
    +         of its ``tool`` results — HTTP 400 "An assistant message with
    +         'tool_calls' must be followed by tool messages…". The split
    +         shape is produced by recovery/continuation paths that append an
    +         interim assistant turn (thinking-prefill, codex
    +         incomplete-continuation) or by host-fed / legacy-persisted /
    +         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
    @@ -387,12 +399,74 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
     
         repairs = 0
     
    +    # Pass 0: merge consecutive assistant messages. Runs BEFORE Pass 1 so
    +    # the merged turn's union of tool_call ids is known when Pass 1
    +    # validates which tool-result messages are orphans. Two assistant
    +    # messages are only adjacent here when nothing (no tool result, no
    +    # user turn) separates them — an intervening ``tool`` message means
    +    # two distinct, valid tool-call rounds that must NOT be merged.
    +    #
    +    # Codex Responses interim turns are exempt: the codex_responses
    +    # api_mode legitimately keeps multiple consecutive incomplete
    +    # assistant turns in history, each carrying its own encrypted
    +    # continuation state (codex_reasoning_items / codex_message_items)
    +    # that must be replayed verbatim. Collapsing them corrupts the
    +    # Responses replay chain (the duplicate-detection logic at
    +    # conversation_loop.py already de-dups identical codex interims).
    +    def _is_codex_interim(m: Dict) -> bool:
    +        return bool(
    +            m.get("codex_reasoning_items")
    +            or m.get("codex_message_items")
    +            or m.get("finish_reason") == "incomplete"
    +        )
    +
    +    collapsed: List[Dict] = []
    +    for msg in messages:
    +        if (
    +            collapsed
    +            and isinstance(msg, dict)
    +            and msg.get("role") == "assistant"
    +            and isinstance(collapsed[-1], dict)
    +            and collapsed[-1].get("role") == "assistant"
    +            and not _is_codex_interim(msg)
    +            and not _is_codex_interim(collapsed[-1])
    +        ):
    +            prev = collapsed[-1]
    +            # Union tool_calls (preserve order, both may carry them).
    +            prev_calls = list(prev.get("tool_calls") or [])
    +            new_calls = list(msg.get("tool_calls") or [])
    +            if new_calls:
    +                prev["tool_calls"] = prev_calls + new_calls
    +            elif prev_calls:
    +                prev["tool_calls"] = prev_calls
    +            # Concatenate plain-text content; leave multimodal (list)
    +            # content on either side alone to avoid mangling attachment
    +            # blocks — fall back to keeping the existing content.
    +            prev_content = prev.get("content")
    +            new_content = msg.get("content")
    +            if isinstance(prev_content, str) and isinstance(new_content, str):
    +                joined = "\n".join(
    +                    p for p in (prev_content.strip(), new_content.strip()) if p
    +                )
    +                prev["content"] = joined
    +            elif not prev_content and new_content is not None:
    +                prev["content"] = new_content
    +            # Carry reasoning_content from the later turn only if the
    +            # earlier turn lacks it (strict thinking providers require a
    +            # reasoning_content on the merged tool-call turn; the first
    +            # non-empty one suffices).
    +            if not prev.get("reasoning_content") and msg.get("reasoning_content"):
    +                prev["reasoning_content"] = msg["reasoning_content"]
    +            repairs += 1
    +            continue
    +        collapsed.append(msg)
    +
         # Pass 1: drop stray tool messages that don't follow a known
         # assistant tool_call_id. Uses a rolling set of known ids refreshed
         # on each assistant message.
         known_tool_ids: set = set()
         filtered: List[Dict] = []
    -    for msg in messages:
    +    for msg in collapsed:
             if not isinstance(msg, dict):
                 filtered.append(msg)
                 continue

  • tests/run_agent/test_message_sequence_repair.py

    @@ -298,3 +298,229 @@ def append_message(self, **kw):
     
         # min(5, 2) = 2 → nothing skipped below start_idx, cursor settles at 2
         assert agent._last_flushed_db_idx == 2
    +
    +
    +# ── Pass 0: merge consecutive assistant messages (issue #29148, #49147) ─────
    +
    +def test_repair_merges_parallel_tool_calls_split_across_assistants():
    +    """Two adjacent assistant(tool_calls) collapse into one turn (#29148).
    +
    +    DeepSeek v4 rejects a replayed history where parallel calls appear as
    +    separate assistant turns:
    +        assistant(tc=[A]) → assistant(tc=[B]) → tool(A) → tool(B)
    +    The repair must produce:
    +        assistant(tc=[A, B]) → tool(A) → tool(B)
    +    """
    +    agent = _bare_agent()
    +    messages = [
    +        {"role": "user", "content": "run both"},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [{"id": "call_A", "type": "function",
    +                         "function": {"name": "session_search", "arguments": "{}"}}]},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [{"id": "call_B", "type": "function",
    +                         "function": {"name": "search_files", "arguments": "{}"}}]},
    +        {"role": "tool", "tool_call_id": "call_A", "content": "A"},
    +        {"role": "tool", "tool_call_id": "call_B", "content": "B"},
    +    ]
    +
    +    repairs = AIAgent._repair_message_sequence(agent, messages)
    +
    +    assert repairs >= 1
    +    assistant_msgs = [m for m in messages if m.get("role") == "assistant"]
    +    assert len(assistant_msgs) == 1
    +    assert {tc["id"] for tc in assistant_msgs[0]["tool_calls"]} == {"call_A", "call_B"}
    +    # Both tool results survive Pass 1 (their ids are in the merged union).
    +    assert sum(1 for m in messages if m.get("role") == "tool") == 2
    +
    +
    +def test_repair_merges_content_then_toolcalls_split():
    +    """content-only assistant followed by tool_calls-only assistant merge (#49147).
    +
    +    The recovery/continuation paths can leave:
    +        assistant(content="Let me search") → assistant(tool_calls=[A]) → tool(A)
    +    which must become:
    +        assistant(content="Let me search", tool_calls=[A]) → tool(A)
    +    """
    +    agent = _bare_agent()
    +    messages = [
    +        {"role": "user", "content": "search"},
    +        {"role": "assistant", "content": "Let me search for that."},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [{"id": "call_1", "type": "function",
    +                         "function": {"name": "session_search", "arguments": "{}"}}]},
    +        {"role": "tool", "tool_call_id": "call_1", "content": "found"},
    +    ]
    +
    +    repairs = AIAgent._repair_message_sequence(agent, messages)
    +
    +    assert repairs >= 1
    +    assistant_msgs = [m for m in messages if m.get("role") == "assistant"]
    +    assert len(assistant_msgs) == 1
    +    merged = assistant_msgs[0]
    +    assert merged["content"] == "Let me search for that."
    +    assert len(merged["tool_calls"]) == 1
    +    assert merged["tool_calls"][0]["id"] == "call_1"
    +    # Tool result still follows immediately.
    +    assert messages[-1]["role"] == "tool"
    +
    +
    +def test_repair_merges_three_consecutive_assistant_tool_calls():
    +    """Three adjacent assistant(tool_calls) turns all collapse into one."""
    +    agent = _bare_agent()
    +    messages = [
    +        {"role": "user", "content": "run three"},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [{"id": "c1", "type": "function",
    +                         "function": {"name": "x", "arguments": "{}"}}]},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [{"id": "c2", "type": "function",
    +                         "function": {"name": "y", "arguments": "{}"}}]},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [{"id": "c3", "type": "function",
    +                         "function": {"name": "z", "arguments": "{}"}}]},
    +        {"role": "tool", "tool_call_id": "c1", "content": "r1"},
    +        {"role": "tool", "tool_call_id": "c2", "content": "r2"},
    +        {"role": "tool", "tool_call_id": "c3", "content": "r3"},
    +    ]
    +
    +    repairs = AIAgent._repair_message_sequence(agent, messages)
    +
    +    assert repairs >= 2
    +    assistant_msgs = [m for m in messages if m.get("role") == "assistant"]
    +    assert len(assistant_msgs) == 1
    +    assert len(assistant_msgs[0]["tool_calls"]) == 3
    +    assert sum(1 for m in messages if m.get("role") == "tool") == 3
    +
    +
    +def test_repair_does_NOT_merge_tool_calls_separated_by_tool_result():
    +    """A tool result between two assistant(tool_calls) marks distinct rounds.
    +
    +    This is the critical guard: two sequential tool-call rounds must NOT be
    +    collapsed, or the second round's tool result would orphan.
    +    """
    +    agent = _bare_agent()
    +    messages = [
    +        {"role": "user", "content": "go"},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [{"id": "t1", "type": "function",
    +                         "function": {"name": "f", "arguments": "{}"}}]},
    +        {"role": "tool", "tool_call_id": "t1", "content": "done"},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [{"id": "t2", "type": "function",
    +                         "function": {"name": "g", "arguments": "{}"}}]},
    +        {"role": "tool", "tool_call_id": "t2", "content": "done2"},
    +    ]
    +    before = sum(1 for m in messages if m.get("role") == "assistant")
    +
    +    AIAgent._repair_message_sequence(agent, messages)
    +
    +    assert sum(1 for m in messages if m.get("role") == "assistant") == before
    +    # Both tool results survive (neither orphaned).
    +    assert sum(1 for m in messages if m.get("role") == "tool") == 2
    +
    +
    +def test_repair_does_NOT_merge_assistant_separated_by_user():
    +    """A user turn between two assistants blocks the merge (normal dialog)."""
    +    agent = _bare_agent()
    +    messages = [
    +        {"role": "user", "content": "q1"},
    +        {"role": "assistant", "content": "a1"},
    +        {"role": "user", "content": "q2"},
    +        {"role": "assistant", "content": "a2"},
    +    ]
    +
    +    AIAgent._repair_message_sequence(agent, messages)
    +
    +    assert sum(1 for m in messages if m.get("role") == "assistant") == 2
    +
    +
    +def test_repair_merges_two_text_only_assistants():
    +    """Two consecutive text-only assistants (no tool_calls) still merge.
    +
    +    The empty-response / thinking-prefill paths can leave two adjacent
    +    text assistants; strict providers reject consecutive same-role turns.
    +    """
    +    agent = _bare_agent()
    +    messages = [
    +        {"role": "user", "content": "q"},
    +        {"role": "assistant", "content": "First part."},
    +        {"role": "assistant", "content": "Second part."},
    +    ]
    +
    +    repairs = AIAgent._repair_message_sequence(agent, messages)
    +
    +    assert repairs >= 1
    +    assistant_msgs = [m for m in messages if m.get("role") == "assistant"]
    +    assert len(assistant_msgs) == 1
    +    assert assistant_msgs[0]["content"] == "First part.\nSecond part."
    +
    +
    +def test_repair_preserves_reasoning_content_on_merge():
    +    """Merged tool-call turn keeps a reasoning_content (DeepSeek/Kimi replay)."""
    +    agent = _bare_agent()
    +    messages = [
    +        {"role": "user", "content": "go"},
    +        {"role": "assistant", "content": "", "reasoning_content": "thinking A",
    +         "tool_calls": [{"id": "a", "type": "function",
    +                         "function": {"name": "f", "arguments": "{}"}}]},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [{"id": "b", "type": "function",
    +                         "function": {"name": "g", "arguments": "{}"}}]},
    +        {"role": "tool", "tool_call_id": "a", "content": "ra"},
    +        {"role": "tool", "tool_call_id": "b", "content": "rb"},
    +    ]
    +
    +    AIAgent._repair_message_sequence(agent, messages)
    +
    +    merged = [m for m in messages if m.get("role") == "assistant"][0]
    +    assert merged.get("reasoning_content") == "thinking A"
    +
    +
    +def test_repair_noop_on_valid_parallel_format():
    +    """A correctly-formatted single assistant with multiple tool_calls is unchanged."""
    +    agent = _bare_agent()
    +    messages = [
    +        {"role": "user", "content": "run both"},
    +        {"role": "assistant", "content": "",
    +         "tool_calls": [
    +             {"id": "call_A", "type": "function",
    +              "function": {"name": "session_search", "arguments": "{}"}},
    +             {"id": "call_B", "type": "function",
    +              "function": {"name": "search_files", "arguments": "{}"}},
    +         ]},
    +        {"role": "tool", "tool_call_id": "call_A", "content": "A"},
    +        {"role": "tool", "tool_call_id": "call_B", "content": "B"},
    +    ]
    +    original_len = len(messages)
    +
    +    repairs = AIAgent._repair_message_sequence(agent, messages)
    +
    +    assert repairs == 0
    +    assert len(messages) == original_len
    +
    +
    +def test_repair_does_NOT_merge_codex_interim_assistants():
    +    """Codex Responses interim turns stay separate (encrypted replay state).
    +
    +    The codex_responses api_mode keeps multiple consecutive incomplete
    +    assistant turns, each carrying distinct codex_reasoning_items /
    +    codex_message_items that must replay verbatim. Pass 0 must exempt them.
    +    Refs test_run_agent_codex_responses.py duplicate-detection tests.
    +    """
    +    agent = _bare_agent()
    +    messages = [
    +        {"role": "user", "content": "think hard"},
    +        {"role": "assistant", "content": "", "finish_reason": "incomplete",
    +         "codex_reasoning_items": [{"encrypted_content": "enc_first"}]},
    +        {"role": "assistant", "content": "", "finish_reason": "incomplete",
    +         "codex_reasoning_items": [{"encrypted_content": "enc_second"}]},
    +        {"role": "assistant", "content": "Final answer."},
    +    ]
    +
    +    AIAgent._repair_message_sequence(agent, messages)
    +
    +    interim = [m for m in messages if m.get("finish_reason") == "incomplete"]
    +    assert len(interim) == 2
    +    encs = [m["codex_reasoning_items"][0]["encrypted_content"] for m in interim]
    +    assert "enc_first" in encs and "enc_second" in encs