diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index c6ed459e93d9..78d12d007faa 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -357,6 +357,48 @@ def sanitize_tool_call_arguments( return repaired +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. + + Two turns interleaving on one session corrupt the durable transcript: + their flushes race (user rows can persist out of arrival order), a row + can be swallowed by the identity-marker dedup over shared history dicts, + and the second turn runs on a history base that never saw the first + turn's exchange. This helper does NOT prevent any of that — it names the + occurrence, with both turn ids, so the dispatch route that let the + second turn through the busy guard can be identified from logs. + + Returns the previous in-flight turn_id when an overlap is detected, + else None. Takes ownership of the in-flight slot either way, so a turn + that crashed before its persist produces at most one warning.""" + prev = getattr(agent, "_inflight_turn_id", None) + prev_started = getattr(agent, "_inflight_turn_started", 0.0) + agent._inflight_turn_id = turn_id + agent._inflight_turn_started = time.time() + if prev and prev != turn_id: + logger.warning( + "turn %s starting while turn %s (started %.0fs ago) has not " + "completed its turn-end persist (session=%s) — concurrent turns " + "on one session; transcript writes may interleave", + turn_id, + prev, + time.time() - prev_started if prev_started else -1.0, + getattr(agent, "session_id", None) or "-", + ) + return prev + return None + + +def note_turn_persisted(agent): + """Clear the in-flight marker at turn-end persist (see note_turn_start). + + Called from the single persist funnel; unconditional by design — when two + turns genuinely overlap, the first persist clears the second turn's slot + 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 + def repair_message_sequence(agent, messages: List[Dict]) -> int: """Collapse malformed role-alternation left in the live history. diff --git a/agent/turn_context.py b/agent/turn_context.py index ea150ff30a7c..75c41a848c6f 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -218,6 +218,11 @@ def build_turn_context( turn_id = f"{agent.session_id or 'session'}:{effective_task_id}:{uuid.uuid4().hex[:8]}" agent._current_turn_id = turn_id agent._current_api_request_id = "" + # Tripwire: warn (with both turn ids) when this turn starts before the + # previous turn's turn-end persist — concurrent turns on one session + # interleave transcript writes. Cleared in _persist_session. + from agent.agent_runtime_helpers import note_turn_start + note_turn_start(agent, turn_id) # Reset retry counters and iteration budget at the start of each turn. agent._invalid_tool_retries = 0 diff --git a/run_agent.py b/run_agent.py index 2d687a78c186..0b8ad99a1714 100644 --- a/run_agent.py +++ b/run_agent.py @@ -1729,12 +1729,15 @@ class AIAgent: # Close and turn-start persistence can run on separate CLI threads; the # marker test-and-append below must be one critical section or both can # observe the same unmarked dict and write duplicate durable rows. + from agent.agent_runtime_helpers import note_turn_persisted + persist_lock = getattr(self, "_session_persist_lock", None) if persist_lock is None: self._drop_trailing_empty_response_scaffolding(messages) self._session_messages = messages self._save_session_log(messages) self._flush_messages_to_session_db(messages, conversation_history) + note_turn_persisted(self) return with persist_lock: @@ -1742,6 +1745,7 @@ class AIAgent: self._session_messages = messages self._save_session_log(messages) self._flush_messages_to_session_db(messages, conversation_history) + note_turn_persisted(self) def _drop_trailing_empty_response_scaffolding(self, messages: List[Dict]) -> None: """Remove private empty-response retry/failure scaffolding from transcript tails. diff --git a/tests/agent/test_turn_overlap_tripwire.py b/tests/agent/test_turn_overlap_tripwire.py new file mode 100644 index 000000000000..faee399879cd --- /dev/null +++ b/tests/agent/test_turn_overlap_tripwire.py @@ -0,0 +1,58 @@ +"""note_turn_start / note_turn_persisted — the concurrent-turn tripwire. + +Two turns interleaving on one session corrupt the durable transcript (flush +order races, identity-dedup row loss, stale history base). The tripwire does +not prevent the overlap; it names the occurrence with both turn ids so the +dispatch route that bypassed the busy guard can be identified from logs. +""" + +import logging + +from agent.agent_runtime_helpers import note_turn_start, note_turn_persisted + + +class _FakeAgent: + session_id = "s1" + + +def test_clean_serial_turns_no_warning(caplog): + agent = _FakeAgent() + with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"): + assert note_turn_start(agent, "s1:t1:aaaa") is None + note_turn_persisted(agent) + assert note_turn_start(agent, "s1:t2:bbbb") is None + note_turn_persisted(agent) + assert not caplog.records + + +def test_overlap_warns_with_both_turn_ids(caplog): + agent = _FakeAgent() + with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"): + note_turn_start(agent, "s1:t1:aaaa") + # second turn starts before the first persisted + prev = note_turn_start(agent, "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 + + +def test_overlap_takes_ownership_no_repeat_warning(caplog): + """A turn that crashed before its persist warns at most once — the next + turn takes ownership of the in-flight slot.""" + agent = _FakeAgent() + with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"): + note_turn_start(agent, "s1:t1:aaaa") # never persists (crash) + note_turn_start(agent, "s1:t2:bbbb") # warns once, takes ownership + note_turn_persisted(agent) + note_turn_start(agent, "s1:t3:cccc") # clean again + assert len(caplog.records) == 1 + + +def test_same_turn_id_reentry_is_silent(caplog): + """Re-entering with the same turn_id (retry paths) is not an overlap.""" + agent = _FakeAgent() + with caplog.at_level(logging.WARNING, logger="agent.agent_runtime_helpers"): + note_turn_start(agent, "s1:t1:aaaa") + note_turn_start(agent, "s1:t1:aaaa") + assert not caplog.records