fix(gateway): carry _db_persisted stamp in _build_replay_entry to prevent duplicate transcript flushes (#123462)

(cherry picked from commit 7366ba26e91907641527d698a72f0f57de530eba)

gateway/run.py

@@ -1148,6 +1148,8 @@ def _build_replay_entry(
             entry[_rkey] = _rval
     if preserve_timestamp and msg.get("timestamp"):
         entry["timestamp"] = msg["timestamp"]
+    if msg.get("_db_persisted"):
+        entry["_db_persisted"] = True
     return entry
 
 

tests/gateway/test_cached_agent_history_guard.py

@@ -31,7 +31,7 @@ def test_select_cached_history_keeps_expected_replay_cleanup():
 
     assert observed is None
     assert len(persisted) < len(live)
-    assert persisted == [{"role": "user", "content": "check status"}]
+    assert persisted == [{"role": "user", "content": "check status", "_db_persisted": True}]
     assert _select_cached_agent_history(persisted, live) is persisted
 
 

tests/gateway/test_gateway_history_persistence_stamp.py

@@ -0,0 +1,40 @@
+from types import SimpleNamespace
+from agent.session_persistence import _db_flush_collect
+from gateway.run import _build_gateway_agent_history
+
+
+def test_build_gateway_agent_history_preserves_db_persisted_marker():
+    """Verify plain user and assistant rows preserve the _db_persisted marker (#123462)."""
+    stored = [
+        {"role": "user", "content": "first", "_db_persisted": True, "timestamp": 1.0},
+        {"role": "assistant", "content": "reply", "_db_persisted": True},
+    ]
+    history, _ = _build_gateway_agent_history(stored)
+
+    assert len(history) == 2
+    assert history[0]["role"] == "user"
+    assert history[0].get("_db_persisted") is True
+    assert history[1]["role"] == "assistant"
+    assert history[1].get("_db_persisted") is True
+
+
+def test_db_flush_collect_does_not_recollect_gateway_history_rows():
+    """Verify tool turn flushes do not re-append rebuilt user/assistant rows to state.db (#123462)."""
+    stored = [
+        {"role": "user", "content": "first", "_db_persisted": True, "timestamp": 1.0},
+        {"role": "assistant", "content": "reply", "_db_persisted": True},
+    ]
+    history, _ = _build_gateway_agent_history(stored)
+    new = {"role": "user", "content": "new"}
+
+    agent = SimpleNamespace(
+        session_id="s",
+        _flushed_db_message_ids=None,
+        _last_flushed_db_idx=0,
+        _db_flush_scan_prefix=None,
+        _persist_user_message_idx=None,
+        _pending_cli_user_message=None,
+    )
+    rows, msgs = _db_flush_collect(agent, list(history) + [new], None)
+
+    assert msgs == [new]