Drop orphan tool messages and guard pair helper

Two regressions surfaced by a 5000-iteration fuzz pass on
sim_pr5710.py (deterministic seed 0xC0FFEE):

1) _assistant_tool_call_ids called tc.get("id") without an
   isinstance(tc, dict) guard, so a malformed pre-pydantic tool_calls
   entry (a bare string, a None) crashed _pair_linked_indices and the
   whole compactor with AttributeError. Mirrors the same guard the
   round-2 commit added to estimate_tokens.

2) When a tool message arrives after a user / system boundary and
   references an assistant tool_call_id that the boundary-clearing
   logic in _pair_linked_indices no longer treats as the pair root,
   the assistant gets dropped by the main sweep but the tool survives
   into the output. The result is a tool message whose tool_call_id
   has no matching assistant tool_calls earlier in the kept output;
   llama-server returns 400 on that template. Add a final invariant
   sweep that walks the surviving indices in order and drops any tool
   message whose tool_call_id is not introduced by a surviving
   assistant earlier in the output. Anchored (multimodal-content)
   tools stay regardless, matching the existing
   leak-rather-than-violate-anchor rule.

Tests grow from 86 to 89.

studio/backend/core/inference/context_compaction.py

@@ -101,11 +101,16 @@ def _is_tool_message(msg: dict) -> bool:
 def _assistant_tool_call_ids(msg: dict) -> set[str]:
     """Return the set of ``id`` values from an assistant message's
     ``tool_calls``. Empty set when the message has no tool calls.
+    Mirrors ``estimate_tokens``: skip non-dict entries so malformed
+    pre-pydantic inputs (a string or ``None`` in the list) don't crash
+    ``_pair_linked_indices`` mid-compaction.
     """
     out: set[str] = set()
     tcs = msg.get("tool_calls")
     if isinstance(tcs, list):
         for tc in tcs:
+            if not isinstance(tc, dict):
+                continue
             tcid = tc.get("id")
             if isinstance(tcid, str) and tcid:
                 out.add(tcid)
@@ -279,6 +284,25 @@ class SlidingWindowCompact(CompactStrategy):
                 if asst_idx not in anchor_idx:
                     dropped.add(asst_idx)
 
+        # Final invariant sweep: drop any surviving tool message whose
+        # tool_call_id has no matching assistant ``tool_calls`` earlier
+        # in the kept output. This catches orphans pair_map could not
+        # link -- e.g. a tool that arrives after a user boundary and
+        # references an assistant that the boundary-clearing logic in
+        # ``_pair_linked_indices`` no longer treats as the pair root.
+        # Anchored tool messages (multimodal content) stay regardless,
+        # matching the existing leak-rather-than-violate-anchor rule.
+        seen_ids: set[str] = set()
+        for i, m in enumerate(messages):
+            if i in dropped:
+                continue
+            if m.get("role") == "assistant":
+                seen_ids |= _assistant_tool_call_ids(m)
+            elif m.get("role") == "tool" and i not in anchor_idx:
+                tcid = m.get("tool_call_id")
+                if isinstance(tcid, str) and tcid and tcid not in seen_ids:
+                    dropped.add(i)
+
         return [m for i, m in enumerate(messages) if i not in dropped]
 
 

studio/backend/tests/test_context_compaction_edge.py

@@ -653,3 +653,80 @@ def test_estimate_tokens_counts_compaction_part():
     ]
     # 5 + 4000 chars -> ceil-divided by 4 ~= 1002 tokens.
     assert estimate_tokens(msgs) >= 1000, estimate_tokens(msgs)
+
+
+# ── Anomaly regressions surfaced by sim_pr5710.py fuzzing ────
+
+
+def test_pair_linked_indices_skips_non_dict_tool_call_entries():
+    """``_assistant_tool_call_ids`` ran inside ``_pair_linked_indices``,
+    so a malformed string / None entry in ``tool_calls`` used to crash
+    the compactor mid-call. Mirrors the ``estimate_tokens`` guard.
+    """
+    msgs = [
+        {"role": "assistant", "content": "x", "tool_calls": ["bare-string"]},
+        {"role": "assistant", "content": "y", "tool_calls": [None, {"id": "t1"}]},
+        {"role": "tool", "tool_call_id": "t1", "content": "ok"},
+    ]
+    pm = _pair_linked_indices(msgs)
+    assert pm[0] == set()
+    assert pm[1] == {2}
+
+
+def test_compact_drops_orphan_tool_left_by_user_boundary():
+    """Tool message arrives after a user / system boundary, references
+    an assistant tool_call_id that the boundary-clearing logic no
+    longer treats as the pair root. The assistant gets dropped by the
+    main sweep; without the final invariant pass the orphan tool
+    survives into the output and llama-server rejects the template.
+    Repro distilled from sim_pr5710.py fuzz iter 1407.
+    """
+    msgs = [
+        {"role": "system", "content": "sys"},
+        {"role": "user", "content": "task " * 20},
+        {"role": "assistant", "content": "a", "tool_calls": [_tool_call("t1")]},
+        {"role": "tool", "tool_call_id": "t1", "content": "first"},
+        {"role": "user", "content": "next"},
+        {"role": "user", "content": "again"},
+        {"role": "tool", "tool_call_id": "t1", "content": "stale"},
+    ]
+    out = SlidingWindowCompact(keep_recent=2).compact(msgs, budget_tokens=1)
+    # Any surviving tool message must have its assistant earlier in
+    # the output.
+    seen_ids: set[str] = set()
+    for m in out:
+        if m.get("role") == "assistant":
+            for tc in m.get("tool_calls") or []:
+                tcid = tc.get("id") if isinstance(tc, dict) else None
+                if isinstance(tcid, str) and tcid:
+                    seen_ids.add(tcid)
+        elif m.get("role") == "tool":
+            tcid = m.get("tool_call_id")
+            assert isinstance(tcid, str) and tcid in seen_ids, (
+                f"orphan tool {tcid!r} survived; seen={seen_ids}"
+            )
+
+
+def test_compact_does_not_drop_anchored_multimodal_tool():
+    """An anchored tool (multimodal content) still survives even when
+    its matching assistant gets dropped -- the existing
+    leak-rather-than-violate-anchor rule applies in this direction too.
+    """
+    msgs = [
+        {"role": "system", "content": "sys"},
+        {"role": "user", "content": "task " * 20},
+        {"role": "assistant", "content": "a", "tool_calls": [_tool_call("t1")]},
+        # Multimodal tool message -- anchored.
+        {
+            "role": "tool",
+            "tool_call_id": "t1",
+            "content": [{"type": "text", "text": "ok"}],
+        },
+        {"role": "user", "content": "more"},
+        {"role": "user", "content": "even more"},
+    ]
+    out = SlidingWindowCompact(keep_recent=2).compact(msgs, budget_tokens=1)
+    # The anchored tool survives.
+    assert any(
+        m.get("role") == "tool" and m.get("tool_call_id") == "t1" for m in out
+    )