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

agent/agent_runtime_helpers.py

@@ -26,10 +26,11 @@
 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 _prepend_marker(tool_msg: dict) -> None:
     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,39 @@ 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.
+    # Persist-disabled agents (background-review forks) deliberately share
+    # the live parent's session_id for prompt-cache warmth but can never
+    # write to the transcript — they must not register here (would warn a
+    # false overlap against the parent's real turn) nor pop the parent's
+    # slot at their persist (note_turn_persisted skips them symmetrically).
+    session_id = getattr(agent, "session_id", None)
+    if session_id and not getattr(agent, "_persist_disabled", False):
+        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 +448,18 @@ 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
+    # Symmetric with note_turn_start's cross-agent leg: persist-disabled
+    # forks never registered a session slot, and their persist funnel still
+    # runs — popping here would steal the live parent turn's slot and make
+    # the tripwire under-report the real overlap it exists to catch.
+    if not getattr(agent, "_persist_disabled", False):
+        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:

contributors/emails/hotragn.pettugani_2024@woxsen.edu.in

@@ -0,0 +1 @@
+Hotragn

tests/agent/test_turn_overlap_tripwire.py

@@ -8,13 +8,26 @@
 
 import logging
 
+import pytest
+
+from agent import agent_runtime_helpers as _helpers
 from agent.agent_runtime_helpers import note_turn_start, note_turn_persisted
 
 
 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,111 @@ 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
+
+
+def test_persist_disabled_fork_neither_registers_nor_warns(caplog):
+    """Background-review forks share the live parent's session_id for
+    prompt-cache warmth but are _persist_disabled — they can never write
+    to the transcript, so they must not trip the cross-agent warning
+    against the parent's real in-flight turn (in either direction)."""
+    parent, fork = _FakeAgent(), _FakeAgent()
+    fork._persist_disabled = True
+    with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"):
+        note_turn_start(parent, "s1:t1:aaaa")     # real turn in flight
+        assert note_turn_start(fork, "s1:tr:ffff") is None  # fork: silent
+        note_turn_persisted(fork)                 # fork's funnel still runs
+        # Reverse order on the next cycle: fork in flight, then real turn.
+        note_turn_persisted(parent)
+        note_turn_start(fork, "s1:tr:gggg")
+        assert note_turn_start(parent, "s1:t2:bbbb") is None
+    assert not caplog.records
+
+
+def test_persist_disabled_fork_persist_does_not_steal_parent_slot(caplog):
+    """The fork's persist funnel still runs; it must not pop the parent's
+    session slot, or a real cross-agent overlap right after a review fork
+    would go unreported."""
+    parent, fork, intruder = _FakeAgent(), _FakeAgent(), _FakeAgent()
+    fork._persist_disabled = True
+    with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"):
+        note_turn_start(parent, "s1:t1:aaaa")     # real turn holds the slot
+        note_turn_start(fork, "s1:tr:ffff")
+        note_turn_persisted(fork)                 # must NOT release s1
+        prev = note_turn_start(intruder, "s1:t2:bbbb")
+    assert prev == "s1:t1:aaaa"
+    assert len(caplog.records) == 1