diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index e54cfff71946..354de4116010 100644 --- a/agent/agent_runtime_helpers.py +++ b/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: diff --git a/tests/agent/test_turn_overlap_tripwire.py b/tests/agent/test_turn_overlap_tripwire.py index faee399879cd..314c8fffa81a 100644 --- a/tests/agent/test_turn_overlap_tripwire.py +++ b/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