fix(agent): drop stale api_content sidecar and unpaired tool results

Rebased onto current main to drop the empty-tool_calls fix (already on
main via #86654, cherry-picked from #77944 with @webtecnica's
authorship). This PR now carries only the two fixes unique to it:

1. A pre-existing api_content sidecar left stale on the consecutive-
   assistant merge. The sidecar takes priority over content at
   API-build time, so a merge could silently discard its own freshly
   concatenated content on the next call. Only dropped when the merge
   actually changes the resulting value (wz-heng, #78063 review) --
   content_rewritten compares before/after value, not just whether an
   assignment branch fired, so a falsy new_content (e.g. "") that
   strips to nothing no longer trips a spurious sidecar drop.

2. sanitize_api_messages never flagged a tool result with a missing/
   empty tool_call_id -- its orphan-detection set only ever collected
   truthy ids, so an unpaired result with no id passed the final
   chokepoint untouched.

Addresses teknium1's rebase request and wz-heng's review findings on

agent/agent_runtime_helpers.py

@@ -683,19 +683,46 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
             # blocks — fall back to keeping the existing content.
             prev_content = prev.get("content")
             new_content = msg.get("content")
+            content_rewritten = False
             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
+                # A falsy ``new_content`` (e.g. "") strips to nothing and
+                # ``joined`` collapses back to ``prev_content`` unchanged --
+                # that must NOT count as a rewrite (wz-heng, #78063 review).
+                content_rewritten = joined != prev_content
             elif not prev_content and new_content is not None:
                 prev["content"] = new_content
+                content_rewritten = new_content != prev_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"]
+            # ``prev`` may carry an ``api_content`` sidecar (the exact bytes
+            # previously sent to the API, e.g. a sanitize-divergence stamp —
+            # see ``_flush_messages_to_session_db``) from BEFORE this merge.
+            # The sidecar takes priority over ``content`` at API-build time
+            # (``conversation_loop``'s ``api_messages`` build substitutes it
+            # back in for role ``assistant``), so leaving it in place while
+            # ``prev["content"]`` changes would silently replay the pre-merge
+            # bytes and discard everything this merge just concatenated on —
+            # the same stale-field-survives-the-merge shape as the
+            # ``tool_calls`` gap above, just for a different field. Only drop
+            # it when the merge actually changed the resulting value (e.g.
+            # the later turn's content is ``None``, or either side is
+            # multimodal/list — both branches skip the reassignment and
+            # ``prev["content"]`` is untouched; a falsy ``new_content`` that
+            # strips to nothing also leaves ``joined`` equal to the original
+            # ``prev_content``): in those cases the sidecar is still the
+            # exact bytes previously sent for the UNCHANGED content, and
+            # dropping it would break the prompt-cache replay invariant for
+            # no reason (wz-heng, #78063 review).
+            if content_rewritten:
+                drop_stale_api_content(prev)
             repairs += 1
             continue
         collapsed.append(msg)
@@ -3826,6 +3853,31 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
             elif isinstance(tc, dict):
                 tc["function"] = {"name": _EMPTY_NAME_SENTINEL, "arguments": "{}"}
 
+    # --- Drop tool results with a missing/empty tool_call_id ---
+    # The orphan-sweep below only ever adds a TRUTHY ``tool_call_id`` to
+    # ``result_call_ids``, so a message with a missing/empty id is never
+    # added to that set and can therefore never land in ``orphaned_results``
+    # (a set-difference against ``surviving_call_ids``) either — it silently
+    # passes through this chokepoint untouched and can reach the provider
+    # with no ``tool_call_id`` at all, which strict OpenAI-compatible
+    # providers reject as a schema violation. ``repair_message_sequence``'s
+    # Pass 1 already drops this shape (`if tc_id and tc_id in
+    # known_tool_ids`) when it runs first on the same list, but any caller
+    # that reaches this function without going through
+    # ``repair_message_sequence`` first has no such guard. Drop explicitly
+    # here so this "final chokepoint" claim (see module docstring) actually
+    # holds regardless of caller (#78071).
+    _pre_id_filter_count = len(messages)
+    messages = [
+        m for m in messages
+        if not (m.get("role") == "tool" and not (m.get("tool_call_id") or "").strip())
+    ]
+    if len(messages) != _pre_id_filter_count:
+        _ra().logger.debug(
+            "Pre-call sanitizer: dropped %d tool result(s) with missing/empty tool_call_id",
+            _pre_id_filter_count - len(messages),
+        )
+
     assistant_call_variants: List[tuple[Any, frozenset[str]]] = []
     surviving_call_ids: set[str] = set()
     for msg in messages:

tests/run_agent/test_message_sequence_repair.py

@@ -288,6 +288,106 @@ def test_repair_keeps_two_parallel_calls_answered_by_mixed_variants():
     ]
 
 
+def test_repair_merge_drops_stale_api_content_sidecar_on_surviving_turn():
+    """A pre-existing ``api_content`` sidecar on the surviving (first)
+    assistant turn must be dropped when the merge rewrites ``content`` —
+    otherwise the sidecar (which takes priority over ``content`` at
+    API-build time, see ``conversation_loop``'s ``api_messages`` build)
+    replays the STALE pre-merge bytes on the next call, silently discarding
+    everything the merge just concatenated on. Same stale-field-survives-
+    the-merge shape as the ``tool_calls`` gap above (#77921), for the
+    ``api_content`` field instead.
+    """
+    agent = _bare_agent()
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {
+            "role": "assistant",
+            "content": "first reply",
+            "api_content": "first reply (stale pre-merge bytes)",
+        },
+        {"role": "assistant", "content": "second reply"},
+    ]
+
+    repairs = AIAgent._repair_message_sequence(agent, messages)
+
+    assert repairs == 1
+    assert len(messages) == 2
+    assert "api_content" not in messages[1]
+    assert messages[1]["content"] == "first reply\nsecond reply"
+
+
+def test_repair_merge_preserves_api_content_sidecar_when_content_unchanged():
+    """Negative control (#78063 review): ``api_content`` must NOT be dropped
+    when the merge does not actually rewrite ``prev["content"]``.
+
+    When the later assistant turn's content is ``None``, neither the
+    both-str branch nor the ``not prev_content`` branch fires (``prev_content``
+    is a truthy string, so ``not prev_content`` is False) -- ``prev["content"]``
+    is left completely untouched. The sidecar is still the exact bytes
+    previously sent for that UNCHANGED content, so dropping it here would
+    diverge replay bytes and break the prompt-cache invariant for no reason.
+    """
+    agent = _bare_agent()
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "clean", "api_content": "wire bytes"},
+        {"role": "assistant", "content": None},
+    ]
+
+    repairs = AIAgent._repair_message_sequence(agent, messages)
+
+    assert repairs == 1
+    assert len(messages) == 2
+    assert messages[1]["content"] == "clean"
+    assert messages[1]["api_content"] == "wire bytes"
+
+
+def test_repair_merge_preserves_api_content_sidecar_when_content_unchanged_by_empty_string():
+    """Negative control (wz-heng, #78063 review): ``content_rewritten`` must
+    mean "the value changed", not "entered the assignment branch".
+
+    Later turn's content is ``""`` -- the both-str branch fires and
+    ``prev["content"]`` IS reassigned, but ``joined`` strips the falsy
+    empty string away and collapses back to the original ``prev_content``
+    unchanged. The sidecar must survive because no byte of ``content``
+    actually moved.
+    """
+    agent = _bare_agent()
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "clean", "api_content": "wire bytes"},
+        {"role": "assistant", "content": ""},
+    ]
+
+    repairs = AIAgent._repair_message_sequence(agent, messages)
+
+    assert repairs == 1
+    assert len(messages) == 2
+    assert messages[1]["content"] == "clean"
+    assert messages[1]["api_content"] == "wire bytes"
+
+
+def test_repair_merge_preserves_api_content_sidecar_with_multimodal_content():
+    """Same negative control, multimodal (list) content on the later turn --
+    the merge intentionally leaves list content alone (see the merge's
+    docstring), so ``prev["content"]`` is untouched and the sidecar must
+    survive."""
+    agent = _bare_agent()
+    messages = [
+        {"role": "user", "content": "Q1"},
+        {"role": "assistant", "content": "clean", "api_content": "wire bytes"},
+        {"role": "assistant", "content": [{"type": "text", "text": "img context"}]},
+    ]
+
+    AIAgent._repair_message_sequence(agent, messages)
+
+    assert len(messages) == 2
+    assert messages[1]["content"] == "clean"
+    assert messages[1]["api_content"] == "wire bytes"
+
+
+
 def test_sanitize_consumes_all_responses_id_variants_for_duplicate_result():
     """A sibling-id replay must not replace the first real result."""
     from agent.agent_runtime_helpers import sanitize_api_messages
@@ -496,6 +596,34 @@ def test_sanitize_preserves_distinct_tool_call_ids():
     assert sorted(m["tool_call_id"] for m in out if m.get("role") == "tool") == ["call_A", "call_B"]
 
 
+def test_sanitize_drops_tool_result_with_missing_tool_call_id():
+    """A tool message with NO ``tool_call_id`` must be dropped, not silently
+    passed through.
+
+    Before the fix: ``result_call_ids`` only ever collects TRUTHY ids, so a
+    missing/empty id is never added to that set and can therefore never land
+    in ``orphaned_results`` (a set-difference against ``surviving_call_ids``)
+    either -- the message survives sanitize_api_messages untouched and can
+    reach the provider with no ``tool_call_id`` at all, a schema violation
+    on strict OpenAI-compatible providers (#78071).
+    """
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    messages = [
+        {"role": "user", "content": "hi"},
+        {"role": "assistant", "content": "", "tool_calls": [
+            {"id": "call_Z", "type": "function",
+             "function": {"name": "f", "arguments": "{}"}},
+        ]},
+        {"role": "tool", "tool_call_id": "call_Z", "content": "real result"},
+        {"role": "tool", "tool_call_id": "", "content": "no id"},
+        {"role": "tool", "content": "id key entirely absent"},
+    ]
+    out = sanitize_api_messages(list(messages))
+    tool_ids = [m.get("tool_call_id") for m in out if m.get("role") == "tool"]
+    assert tool_ids == ["call_Z"]  # only the properly-paired result survives
+
+
 # ── tool_call_id reuse by local servers (#70724) ────────────────────────────
 # llama.cpp emits ONE constant tool_call_id for every tool call it returns, so
 # ``tool_call_id`` is not globally unique in practice. The #58327 dedup pass