fix(loop): the turn-boundary export skips preflight-timeout envelopes and stops re-anchoring the persist index

Follow-up to #106312. _preflight_timeout_result carries the prior history without this turn's
user row (#7100); with a repeated prompt ("continue") the verbatim scan resolved to the
historical copy and exported it as this turn's proven boundary — the exact relabeling the export
exists to prevent. Nothing is exported for that envelope now.

The trailing `agent._persist_user_message_idx = idx` ran after finalize_turn had already flushed
the transcript, so it never influenced a persist and the next turn reset it: dead state, removed.

agent/turn_context.py

@@ -254,10 +254,11 @@ def export_current_turn_boundary(agent: Any, result: Any, user_message: Any) ->
     identical historical prompt and claim its old answer as this turn's. So the producer
     exports the coordinate, computed on the final list, only when the addressed row is this
     turn's user message verbatim. Otherwise the keys are omitted and hosts fail closed.
-    Also mirrors the final index into ``_persist_user_message_idx`` so the persist override
-    targets the surviving row after post-turn micro-compaction.
+
+    A preflight-timeout envelope carries the prior history without this turn's row (#7100), so a
+    repeated prompt would resolve to its historical copy: nothing is exported there.
     """
-    if not isinstance(result, dict):
+    if not isinstance(result, dict) or result.get("turn_exit_reason") == "context_compression_timeout":
         return result
     messages = result.get("messages")
     turn_id = str(getattr(agent, "_current_turn_id", "") or "")
@@ -278,7 +279,6 @@ def export_current_turn_boundary(agent: Any, result: Any, user_message: Any) ->
         return result  # rewritten (merge-into-tail) row: not a proven boundary
     result["turn_id"] = turn_id
     result["current_turn_user_idx"] = idx
-    agent._persist_user_message_idx = idx
     return result
 
 

tests/run_agent/test_export_current_turn_boundary.py

@@ -44,11 +44,21 @@ def test_boundary_is_exported_only_for_the_verbatim_current_row(user_message, me
     result = export_current_turn_boundary(agent, {"messages": messages}, user_message)
     if expected_idx is None:
         assert "current_turn_user_idx" not in result and "turn_id" not in result
-        assert agent._persist_user_message_idx is None
     else:
         assert result["current_turn_user_idx"] == expected_idx
         assert result["turn_id"] == agent._current_turn_id
-        assert agent._persist_user_message_idx == expected_idx
+    # the persist funnel has already run by the time the envelope is stamped: never re-anchor it here
+    assert agent._persist_user_message_idx is None
+
+
+def test_preflight_timeout_envelope_exports_nothing():
+    """The preflight-timeout result carries the prior history without this turn's row (#7100);
+    a repeated prompt must not be resolved to its historical copy."""
+    agent = _Agent()
+    result = {"messages": [{"role": "user", "content": "continue"}, {"role": "assistant", "content": "old"}],
+              "turn_exit_reason": "context_compression_timeout"}
+    out = export_current_turn_boundary(agent, result, "continue")
+    assert "current_turn_user_idx" not in out and "turn_id" not in out
 
 
 @pytest.fixture()