fix(agent): widen composite-id alias matching to the compressor; unify variant policy owners (#63000)

Follow-up on top of the salvaged #93335:

- context_compressor._sanitize_tool_pairs now expands alias spellings on
  the RESULT side too (tool_result_id_variants), so a composite
  call|item-keyed result pairs with its split-field tool_call instead of
  being dropped and its call stripped.
- The compressor's _tool_call_id_variants staticmethod and
  agent_runtime_helpers' module-level _tool_call_id_variants are now thin
  forwarders to agent.message_sanitization.tool_call_id_variants — one
  policy owner for alias expansion, so the pre-call sanitizer, repair
  pass, dedup pass, and compression sanitizer can never drift apart.
- Preserved the #91768 SDK-object tolerance in repair pass 1 (the
  shared helper handles non-dict tool_calls via getattr; the salvaged
  commit's isinstance-dict guard was dropped in the merge resolution).

New regression tests: composite-keyed results through
sanitize_api_messages (both directions) and _sanitize_tool_pairs, with
negative controls. Sabotage-verified: compressor test fails with raw
tool_call_id tracking.

agent/context_compressor.py

@@ -33,6 +33,7 @@ from agent.auxiliary_client import (
 )
 from agent.context_engine import ContextEngine, sanitize_memory_context
 from agent.error_classifier import FailoverReason, classify_api_error
+from agent.message_sanitization import tool_result_id_variants
 from agent.model_metadata import (
     MINIMUM_CONTEXT_LENGTH,
     get_model_context_length,
@@ -5741,14 +5742,21 @@ This compaction should PRIORITISE preserving all information related to the focu
             if msg.get("role") == "tool":
                 cid = msg.get("tool_call_id")
                 if cid:
-                    result_call_ids.add(cid)
+                    # Expand alias spellings on the RESULT side too — a
+                    # composite ``call|item`` tool_call_id must match a
+                    # tool_call registered under either half (#63000).
+                    result_call_ids |= tool_result_id_variants(cid)
 
         # 1. Remove tool results whose call_id has no matching assistant tool_call
         orphaned_results = result_call_ids - surviving_call_ids
         if orphaned_results:
             messages = [
                 m for m in messages
-                if not (m.get("role") == "tool" and m.get("tool_call_id") in orphaned_results)
+                if not (
+                    m.get("role") == "tool"
+                    and (rv := tool_result_id_variants(m.get("tool_call_id")))
+                    and not (rv & surviving_call_ids)
+                )
             ]
             if not self.quiet_mode:
                 logger.info("Compression sanitizer: removed %d orphaned tool result(s)", len(orphaned_results))

tests/run_agent/test_message_sequence_repair.py

@@ -895,3 +895,64 @@ def test_sanitize_dedup_pass_rearms_constant_llamacpp_id():
 
     tool_msgs = [m for m in out if m.get("role") == "tool"]
     assert [m["content"] for m in tool_msgs] == ["round1", "round2"]
+
+
+def test_sanitize_keeps_result_keyed_on_composite_bridge_id():
+    """A tool result keyed on the composite ``call|item`` bridge spelling
+    (#63000) must pair with a tool_call carrying the split id/call_id
+    fields — and vice versa."""
+    from agent.agent_runtime_helpers import sanitize_api_messages
+
+    # Result keyed on composite; call carries split fields.
+    messages = [
+        {"role": "user", "content": "task"},
+        {"role": "assistant", "content": "", "tool_calls": [
+            {"id": "fc_1", "call_id": "call_1", "type": "function",
+             "function": {"name": "f", "arguments": "{}"}}]},
+        {"role": "tool", "tool_call_id": "call_1|fc_1", "content": "REAL"},
+    ]
+    out = sanitize_api_messages(messages)
+    tool_msgs = [m for m in out if m.get("role") == "tool"]
+    assert [m["content"] for m in tool_msgs] == ["REAL"]
+
+    # Call carries only the composite id; result keyed on the bare half.
+    messages = [
+        {"role": "user", "content": "task"},
+        {"role": "assistant", "content": "", "tool_calls": [
+            {"id": "call_2|fc_2", "type": "function",
+             "function": {"name": "f", "arguments": "{}"}}]},
+        {"role": "tool", "tool_call_id": "call_2", "content": "REAL bare"},
+    ]
+    out = sanitize_api_messages(messages)
+    tool_msgs = [m for m in out if m.get("role") == "tool"]
+    assert [m["content"] for m in tool_msgs] == ["REAL bare"]
+
+
+def test_compressor_sanitize_keeps_composite_keyed_pair():
+    """The compression sanitizer must apply the same alias expansion on the
+    RESULT side: a composite-keyed result pairs with its split-field call
+    instead of being dropped and its call stripped (#63000)."""
+    from agent.context_compressor import ContextCompressor
+
+    cc = ContextCompressor.__new__(ContextCompressor)
+    cc.quiet_mode = True
+    msgs = [
+        {"role": "assistant", "content": "", "tool_calls": [
+            {"id": "fc_7", "call_id": "call_7", "type": "function",
+             "function": {"name": "s", "arguments": "{}"}}]},
+        {"role": "tool", "tool_call_id": "call_7|fc_7", "content": "res"},
+        {"role": "user", "content": "next"},
+    ]
+    out = cc._sanitize_tool_pairs(msgs)
+    asst = next(m for m in out if m.get("role") == "assistant")
+    assert asst.get("tool_calls"), "valid tool_call must not be stripped"
+    assert [m["content"] for m in out if m.get("role") == "tool"] == ["res"]
+
+    # Negative control: composite orphan (matches nothing) still dropped.
+    msgs = [
+        {"role": "assistant", "content": "hi"},
+        {"role": "tool", "tool_call_id": "call_z|fc_z", "content": "orphan"},
+        {"role": "user", "content": "next"},
+    ]
+    out = cc._sanitize_tool_pairs(msgs)
+    assert not any(m.get("role") == "tool" for m in out)