fix(gateway): reset _last_flushed_db_idx on cached agent reuse (#44327)

gateway/run.py

@@ -12548,6 +12548,11 @@ def _init_cached_agent_for_turn(agent: Any, interrupt_depth: int) -> None:
         if interrupt_depth == 0:
             agent._last_activity_ts = time.time()
             agent._last_activity_desc = "starting new turn (cached)"
+            # Reset the SessionDB flush cursor so the new turn's messages are
+            # fully persisted — a stale value from the previous turn would
+            # cause `_flush_messages_to_session_db` to skip new rows (#44327).
+            if hasattr(agent, "_last_flushed_db_idx"):
+                agent._last_flushed_db_idx = 0
         agent._api_call_count = 0
 
     def _release_evicted_agent_soft(self, agent: Any) -> None:

tests/gateway/test_agent_cache.py

@@ -1410,6 +1410,38 @@ def test_deep_interrupt_recursion_preserves_idle_clock(self):
 
         assert agent._last_activity_ts == old_ts
 
+    def test_fresh_turn_resets_flush_cursor(self):
+        """interrupt_depth=0: _last_flushed_db_idx resets so new-turn
+        messages are fully persisted to the session DB (#44327)."""
+        from gateway.run import GatewayRunner
+
+        agent = self._fake_agent()
+        agent._last_flushed_db_idx = 42  # stale from previous turn
+
+        with patch("gateway.run.time") as mock_time:
+            mock_time.time.return_value = _FAKE_NOW
+            GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0)
+
+        assert agent._last_flushed_db_idx == 0, (
+            "_last_flushed_db_idx must be reset on a fresh turn so that "
+            "_flush_messages_to_session_db starts from index 0"
+        )
+
+    def test_interrupt_turn_preserves_flush_cursor(self):
+        """interrupt_depth=1: _last_flushed_db_idx preserved so an
+        in-progress flush is not disrupted by interrupt re-entry."""
+        from gateway.run import GatewayRunner
+
+        agent = self._fake_agent()
+        agent._last_flushed_db_idx = 42
+
+        GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=1)
+
+        assert agent._last_flushed_db_idx == 42, (
+            "_last_flushed_db_idx must not be reset on interrupt-recursive "
+            "turns — the flush cursor tracks in-progress writes"
+        )
+
     def test_api_call_count_reset_regardless_of_depth(self):
         """_api_call_count is always reset to 0 for the new turn, at any depth."""
         from gateway.run import GatewayRunner