fix(agent): catch cross-agent turn overlaps in the tripwire (#64934)

note_turn_start kept its in-flight marker on the agent object, but the
gateway caches agents per routing key (_agent_cache) while transcripts
are owned by session_id — and switch_session (/resume from a second
surface, CLI-continuity rebinding, async-delegation pinning,
topic-binding tip-walks) maps multiple routing keys onto one session_id
without any cross-key check. Two keys mapped to one session run
concurrent turns on two different agent objects, so the per-agent
tripwire could never fire for exactly the dispatch route #64934 is
waiting to identify.

Add a module-level session_id-keyed in-flight registry alongside the
per-agent marker. Same philosophy as the original tripwire: log-only,
takes ownership on overlap, under-reports rather than double-reports
(a same-agent overlap warns once, not twice). The persist-time clear
pops the session id stamped at turn start, so a mid-turn compression
rotation of agent.session_id cannot strand the slot.

agent/agent_runtime_helpers.py

@@ -26,10 +26,11 @@ import copy
 import json
 import logging
 import re
+import threading
 import time
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 from hermes_cli.timeouts import get_provider_request_timeout
 from agent.prompt_builder import format_steer_marker
@@ -358,9 +359,25 @@ def sanitize_tool_call_arguments(
     return repaired
 
 
+# Session-scoped in-flight registry backing note_turn_start's cross-agent
+# check.  The per-agent marker catches a second turn on the SAME AIAgent
+# object, but the gateway caches agents per *routing key* (``_agent_cache``
+# in gateway/run.py) while the durable transcript is keyed by *session_id* —
+# and the key→id mapping is many-to-one (``switch_session``: /resume from a
+# second chat/topic, CLI-continuity rebinding, async-delegation pinning,
+# topic-binding tip-walks).  Two routing keys mapped to one session_id run
+# concurrent turns on two different agent objects, which per-agent state can
+# never see (#64934).  Keyed by session_id so that route produces the same
+# named warning.  Process-local by design — same visibility scope as the
+# per-agent marker it extends.
+_INFLIGHT_TURNS_BY_SESSION: Dict[str, Tuple[str, float]] = {}
+_INFLIGHT_TURNS_LOCK = threading.Lock()
+
+
 def note_turn_start(agent, turn_id: str):
-    """Tripwire: detect a turn starting while the previous turn of the SAME
-    agent/session has not completed its turn-end persist.
+    """Tripwire: detect a turn starting while a previous turn of the same
+    agent — or of the same underlying *session* on a different agent object —
+    has not completed its turn-end persist.
 
     Two turns interleaving on one session corrupt the durable transcript:
     their flushes race (user rows can persist out of arrival order), a row
@@ -377,6 +394,7 @@ def note_turn_start(agent, turn_id: str):
     prev_started = getattr(agent, "_inflight_turn_started", 0.0)
     agent._inflight_turn_id = turn_id
     agent._inflight_turn_started = time.time()
+    overlap = None
     if prev and prev != turn_id:
         logger.warning(
             "turn %s starting while turn %s (started %.0fs ago) has not "
@@ -387,8 +405,34 @@ def note_turn_start(agent, turn_id: str):
             time.time() - prev_started if prev_started else -1.0,
             getattr(agent, "session_id", None) or "-",
         )
-        return prev
-    return None
+        overlap = prev
+
+    # Cross-agent leg: same session_id in flight under a different agent
+    # object means two routing keys resolve to one durable session — the
+    # busy guard (keyed by routing key) cannot see this overlap at all.
+    session_id = getattr(agent, "session_id", None)
+    if session_id:
+        now = time.time()
+        with _INFLIGHT_TURNS_LOCK:
+            entry = _INFLIGHT_TURNS_BY_SESSION.get(session_id)
+            _INFLIGHT_TURNS_BY_SESSION[session_id] = (turn_id, now)
+        # Stamp the session id this turn registered under: compression can
+        # rotate agent.session_id mid-turn, and the persist-time clear must
+        # pop the slot the turn actually holds, not the rotated id.
+        agent._inflight_turn_session_id = session_id
+        if entry and entry[0] not in (turn_id, prev):
+            logger.warning(
+                "turn %s starting while turn %s (started %.0fs ago) is still "
+                "in flight on session %s under a different agent object — "
+                "two routing keys are mapped to one session_id; concurrent "
+                "turns on one session; transcript writes may interleave",
+                turn_id,
+                entry[0],
+                now - entry[1] if entry[1] else -1.0,
+                session_id,
+            )
+            overlap = overlap or entry[0]
+    return overlap
 
 
 def note_turn_persisted(agent):
@@ -399,6 +443,13 @@ def note_turn_persisted(agent):
     and the tripwire under-reports instead of double-reporting. A diagnostic
     must never be noisier than the defect it hunts."""
     agent._inflight_turn_id = None
+    session_id = getattr(agent, "_inflight_turn_session_id", None) or getattr(
+        agent, "session_id", None
+    )
+    if session_id:
+        with _INFLIGHT_TURNS_LOCK:
+            _INFLIGHT_TURNS_BY_SESSION.pop(session_id, None)
+    agent._inflight_turn_session_id = None
 
 
 def repair_message_sequence(agent, messages: List[Dict]) -> int:

tests/agent/test_turn_overlap_tripwire.py

@@ -8,6 +8,9 @@ dispatch route that bypassed the busy guard can be identified from logs.
 
 import logging
 
+import pytest
+
+from agent import agent_runtime_helpers as _helpers
 from agent.agent_runtime_helpers import note_turn_start, note_turn_persisted
 
 
@@ -15,6 +18,16 @@ class _FakeAgent:
     session_id = "s1"
 
 
+@pytest.fixture(autouse=True)
+def _clear_inflight_registry():
+    """Isolate the module-level session registry between tests."""
+    with _helpers._INFLIGHT_TURNS_LOCK:
+        _helpers._INFLIGHT_TURNS_BY_SESSION.clear()
+    yield
+    with _helpers._INFLIGHT_TURNS_LOCK:
+        _helpers._INFLIGHT_TURNS_BY_SESSION.clear()
+
+
 def test_clean_serial_turns_no_warning(caplog):
     agent = _FakeAgent()
     with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"):
@@ -56,3 +69,78 @@ def test_same_turn_id_reentry_is_silent(caplog):
         note_turn_start(agent, "s1:t1:aaaa")
         note_turn_start(agent, "s1:t1:aaaa")
     assert not caplog.records
+
+
+def test_cross_agent_same_session_overlap_warns(caplog):
+    """#64934 route: two routing keys mapped to one session_id run their
+    turns on two different agent objects (the gateway agent cache is keyed
+    by routing key), so per-agent state alone can never see the overlap."""
+    agent_a, agent_b = _FakeAgent(), _FakeAgent()
+    with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"):
+        assert note_turn_start(agent_a, "s1:t1:aaaa") is None
+        prev = note_turn_start(agent_b, "s1:t2:bbbb")
+    assert prev == "s1:t1:aaaa"
+    assert len(caplog.records) == 1
+    msg = caplog.records[0].getMessage()
+    assert "s1:t1:aaaa" in msg and "s1:t2:bbbb" in msg and "s1" in msg
+    assert "different agent object" in msg
+
+
+def test_cross_agent_serial_turns_are_silent(caplog):
+    """A persisted turn releases the session slot — a later turn on another
+    agent object for the same session is normal (e.g. cache eviction)."""
+    agent_a, agent_b = _FakeAgent(), _FakeAgent()
+    with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"):
+        note_turn_start(agent_a, "s1:t1:aaaa")
+        note_turn_persisted(agent_a)
+        note_turn_start(agent_b, "s1:t2:bbbb")
+        note_turn_persisted(agent_b)
+    assert not caplog.records
+
+
+def test_distinct_sessions_never_cross_warn(caplog):
+    """Concurrent turns on different session_ids are legitimate parallelism."""
+    agent_a, agent_b = _FakeAgent(), _FakeAgent()
+    agent_b.session_id = "s2"
+    with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"):
+        note_turn_start(agent_a, "s1:t1:aaaa")
+        assert note_turn_start(agent_b, "s2:t2:bbbb") is None
+    assert not caplog.records
+
+
+def test_same_agent_overlap_warns_once_not_twice(caplog):
+    """A same-agent overlap must not double-report through the session leg."""
+    agent = _FakeAgent()
+    with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"):
+        note_turn_start(agent, "s1:t1:aaaa")
+        prev = note_turn_start(agent, "s1:t2:bbbb")
+    assert prev == "s1:t1:aaaa"
+    assert len(caplog.records) == 1
+
+
+def test_persist_clears_start_session_after_mid_turn_rotation(caplog):
+    """Compression rotates agent.session_id mid-turn; the persist must
+    release the slot the turn registered under, not the rotated id."""
+    agent = _FakeAgent()
+    agent.session_id = "s-parent"
+    with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"):
+        note_turn_start(agent, "sp:t1:aaaa")
+        agent.session_id = "s-child"  # mid-turn compression rotation
+        note_turn_persisted(agent)
+        # A fresh turn on the parent id must find the slot released.
+        other = _FakeAgent()
+        other.session_id = "s-parent"
+        assert note_turn_start(other, "sp:t2:bbbb") is None
+    assert not caplog.records
+
+
+def test_crashed_cross_agent_turn_warns_once_then_recovers(caplog):
+    """A turn that never persists (crash) yields one warning; the next turn
+    takes ownership of the session slot and the tripwire goes quiet."""
+    agent_a, agent_b, agent_c = _FakeAgent(), _FakeAgent(), _FakeAgent()
+    with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"):
+        note_turn_start(agent_a, "s1:t1:aaaa")   # never persists (crash)
+        note_turn_start(agent_b, "s1:t2:bbbb")   # warns once, takes ownership
+        note_turn_persisted(agent_b)
+        note_turn_start(agent_c, "s1:t3:cccc")   # clean again
+    assert len(caplog.records) == 1