From 064e29b58c67d8efa24a102b7a64ee41fdc67b30 Mon Sep 17 00:00:00 2001 From: teyrebaz33 Date: Sun, 22 Mar 2026 20:08:35 +0300 Subject: [PATCH] fix(gateway): remove expired session entries after flush to prevent restart re-flushing --- gateway/run.py | 8 +++-- gateway/session.py | 7 ----- tests/gateway/test_async_memory_flush.py | 37 +++++------------------- 3 files changed, 12 insertions(+), 40 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index a1a73475927c..c69db327d947 100644 --- a/gateway/run.py +++ b/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: diff --git a/gateway/session.py b/gateway/session.py index 58e8d584d537..dd5849f5ed4d 100644 --- a/gateway/session.py +++ b/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") diff --git a/tests/gateway/test_async_memory_flush.py b/tests/gateway/test_async_memory_flush.py index 675746920f86..59b8616cc8e1 100644 --- a/tests/gateway/test_async_memory_flush.py +++ b/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')