fix(gateway): remove expired session entries after flush to prevent restart re-flushing

gateway/run.py

@@ -1045,8 +1045,6 @@ async def _session_expiry_watcher(self, interval: int = 300):
             try:
                 self.session_store._ensure_loaded()
                 for key, entry in list(self.session_store._entries.items()):
-                    if entry.session_id in self.session_store._pre_flushed_sessions:
-                        continue  # already flushed this session
                     if not self.session_store._is_session_expired(entry):
                         continue  # session still active
                     # Session has expired — flush memories in the background
@@ -1057,7 +1055,11 @@ async def _session_expiry_watcher(self, interval: int = 300):
                     try:
                         await self._async_flush_memories(entry.session_id, key)
                         self._shutdown_gateway_honcho(key)
-                        self.session_store._pre_flushed_sessions.add(entry.session_id)
+                        # Remove the entry so it is not re-flushed on restart.
+                        # _pre_flushed_sessions was an in-memory set that reset
+                        # on every restart — persisting the removal is the fix.
+                        self.session_store._entries.pop(key, None)
+                        self.session_store._save()
                     except Exception as e:
                         logger.debug("Proactive memory flush failed for %s: %s", entry.session_id, e)
             except Exception as e:

gateway/session.py

@@ -472,9 +472,6 @@ def __init__(self, sessions_dir: Path, config: GatewayConfig,
         self._entries: Dict[str, SessionEntry] = {}
         self._loaded = False
         self._has_active_processes_fn = has_active_processes_fn
-        # on_auto_reset is deprecated — memory flush now runs proactively
-        # via the background session expiry watcher in GatewayRunner.
-        self._pre_flushed_sessions: set = set()  # session_ids already flushed by watcher
 
         # Initialize SQLite session database
         self._db = None
@@ -665,14 +662,10 @@ def get_or_create_session(
                 self._save()
                 return entry
             else:
-                # Session is being auto-reset.  The background expiry watcher
-                # should have already flushed memories proactively; discard
-                # the marker so it doesn't accumulate.
                 was_auto_reset = True
                 auto_reset_reason = reset_reason
                 # Track whether the expired session had any real conversation
                 reset_had_activity = entry.total_tokens > 0
-                self._pre_flushed_sessions.discard(entry.session_id)
                 if self._db:
                     try:
                         self._db.end_session(entry.session_id, "session_reset")

tests/gateway/test_async_memory_flush.py

@@ -3,7 +3,7 @@
 Verifies that:
 1. _is_session_expired() works from a SessionEntry alone (no source needed)
 2. The sync callback is no longer called in get_or_create_session
-3. _pre_flushed_sessions tracking works correctly
+3. Expired session entries are removed from sessions.json after flush (not re-flushed on restart)
 4. The background watcher can detect expired sessions
 """
 
@@ -115,32 +115,25 @@ def test_daily_mode_expired(self, tmp_path):
 class TestGetOrCreateSessionNoCallback:
     """get_or_create_session should NOT call a sync flush callback."""
 
-    def test_auto_reset_cleans_pre_flushed_marker(self, idle_store):
-        """When a session auto-resets, the pre_flushed marker should be discarded."""
+    def test_auto_reset_creates_new_session(self, idle_store):
+        """When a session auto-resets, a new session_id should be created."""
         source = SessionSource(
             platform=Platform.TELEGRAM,
             chat_id="123",
             chat_type="dm",
         )
-        # Create initial session
         entry1 = idle_store.get_or_create_session(source)
         old_sid = entry1.session_id
 
-        # Simulate the watcher having flushed it
-        idle_store._pre_flushed_sessions.add(old_sid)
-
         # Simulate the session going idle
         entry1.updated_at = datetime.now() - timedelta(minutes=120)
         idle_store._save()
 
-        # Next call should auto-reset
+        # Next call should auto-reset with a fresh session_id
         entry2 = idle_store.get_or_create_session(source)
         assert entry2.session_id != old_sid
         assert entry2.was_auto_reset is True
 
-        # The old session_id should be removed from pre_flushed
-        assert old_sid not in idle_store._pre_flushed_sessions
-
     def test_no_sync_callback_invoked(self, idle_store):
         """No synchronous callback should block during auto-reset."""
         source = SessionSource(
@@ -159,22 +152,6 @@ def test_no_sync_callback_invoked(self, idle_store):
         entry2 = idle_store.get_or_create_session(source)
         assert entry2.was_auto_reset is True
 
-
-class TestPreFlushedSessionsTracking:
-    """The _pre_flushed_sessions set should prevent double-flushing."""
-
-    def test_starts_empty(self, idle_store):
-        assert len(idle_store._pre_flushed_sessions) == 0
-
-    def test_add_and_check(self, idle_store):
-        idle_store._pre_flushed_sessions.add("sid_old")
-        assert "sid_old" in idle_store._pre_flushed_sessions
-        assert "sid_other" not in idle_store._pre_flushed_sessions
-
-    def test_discard_on_reset(self, idle_store):
-        """discard should remove without raising if not present."""
-        idle_store._pre_flushed_sessions.add("sid_a")
-        idle_store._pre_flushed_sessions.discard("sid_a")
-        assert "sid_a" not in idle_store._pre_flushed_sessions
-        # discard on non-existent should not raise
-        idle_store._pre_flushed_sessions.discard("sid_nonexistent")
+    def test_pre_flushed_sessions_attr_removed(self, idle_store):
+        """_pre_flushed_sessions in-memory set is removed — entry deletion is the fix."""
+        assert not hasattr(idle_store, '_pre_flushed_sessions')