chore(agent): tripwire — warn when a turn starts before the previous turn's persist

Two turns interleaving on one session corrupt the durable transcript:
flushes race (user rows persist out of arrival order), the identity-marker
dedup over shared history dicts can swallow a row, and the second turn
runs on a history base that never saw the first turn's exchange. The
dispatch route that lets the second turn through the busy guard is not
yet identified.

Add note_turn_start (build_turn_context) / note_turn_persisted
(_persist_session funnel): one WARNING naming both turn_ids when a turn
starts before the previous turn's turn-end persist. Ownership transfer
keeps a crashed turn from warning more than once; the unconditional clear
makes the tripwire under-report rather than double-report under a real
overlap. Log-only, no behavior change.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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.

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

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.

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