diff --git a/gateway/session.py b/gateway/session.py index 2edd3c4d30ae..7a1273e93cf0 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1043,6 +1043,21 @@ class SessionStore: self._db = SessionDB() except Exception as e: print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}") + + def _has_active_processes_safe(self, session_key: str, *, context: str) -> bool: + """Return whether a session has active work, failing closed on registry errors.""" + if self._has_active_processes_fn is None: + return False + try: + return bool(self._has_active_processes_fn(session_key)) + except Exception as exc: + logger.warning( + "has_active_processes_fn raised during %s for %s; keeping session alive: %s", + context, + session_key, + exc, + ) + return True def _ensure_loaded(self) -> None: """Load sessions index from disk if not already loaded.""" @@ -1606,13 +1621,12 @@ class SessionStore: Used by the background expiry watcher to proactively flush memories. Sessions with active background processes are never considered expired. """ - if self._has_active_processes_fn: - if self._has_active_processes_fn(entry.session_key): - logger.debug( - "Session %s not expired — active background processes", - entry.session_key, - ) - return False + if self._has_active_processes_safe(entry.session_key, context="expiry"): + logger.debug( + "Session %s not expired — active background processes", + entry.session_key, + ) + return False policy = self.config.get_reset_policy( platform=entry.platform, @@ -1707,14 +1721,13 @@ class SessionStore: Sessions with active background processes are never reset. """ - if self._has_active_processes_fn: - session_key = self._generate_session_key(source) - if self._has_active_processes_fn(session_key): - logger.debug( - "Session reset skipped for %s — active background processes", - session_key, - ) - return None + session_key = self._generate_session_key(source) + if self._has_active_processes_safe(session_key, context="reset"): + logger.debug( + "Session reset skipped for %s — active background processes", + session_key, + ) + return None policy = self.config.get_reset_policy( platform=source.platform, @@ -2221,19 +2234,8 @@ class SessionStore: # The callback is keyed by session_key (see process_registry. # has_active_for_session); passing session_id here used to # never match, so active sessions got pruned anyway. - if self._has_active_processes_fn is not None: - try: - if self._has_active_processes_fn(entry.session_key): - continue - except Exception as exc: - logger.debug( - "has_active_processes_fn raised during prune for %s: %s", - entry.session_key, exc, - ) - # Fail safe: if we can't tell whether a background - # process is attached, keep the entry rather than - # risk orphaning live work. - continue + if self._has_active_processes_safe(entry.session_key, context="prune"): + continue if entry.updated_at < cutoff: removed_keys.append(key) for key in removed_keys: diff --git a/tests/gateway/test_session_reset_notify.py b/tests/gateway/test_session_reset_notify.py index 3ecdc60840cb..6db74fe1b338 100644 --- a/tests/gateway/test_session_reset_notify.py +++ b/tests/gateway/test_session_reset_notify.py @@ -31,11 +31,15 @@ def _make_source(platform=Platform.TELEGRAM, chat_id="123", user_id="u1"): ) -def _make_store(policy=None, tmp_path=None): +def _make_store(policy=None, tmp_path=None, has_active_processes_fn=None): config = GatewayConfig() if policy: config.default_reset_policy = policy - store = SessionStore(sessions_dir=tmp_path or "/tmp/test-sessions", config=config) + store = SessionStore( + sessions_dir=tmp_path or "/tmp/test-sessions", + config=config, + has_active_processes_fn=has_active_processes_fn, + ) return store @@ -101,6 +105,45 @@ class TestShouldResetReason: source = _make_source() assert store._should_reset(entry, source) is None + def test_returns_none_when_active_process_check_raises(self, tmp_path): + def _raise(_session_key): + raise RuntimeError("process registry unavailable") + + store = _make_store( + SessionResetPolicy(mode="idle", idle_minutes=30), + tmp_path, + has_active_processes_fn=_raise, + ) + entry = SessionEntry( + session_key="test", + session_id="s1", + created_at=datetime.now() - timedelta(hours=2), + updated_at=datetime.now() - timedelta(hours=1), + ) + source = _make_source() + + assert store._should_reset(entry, source) is None + + def test_is_session_expired_fails_closed_when_active_process_check_raises(self, tmp_path): + def _raise(_session_key): + raise RuntimeError("process registry unavailable") + + store = _make_store( + SessionResetPolicy(mode="idle", idle_minutes=30), + tmp_path, + has_active_processes_fn=_raise, + ) + entry = SessionEntry( + session_key="test", + session_id="s1", + platform=Platform.TELEGRAM, + chat_type="dm", + created_at=datetime.now() - timedelta(hours=2), + updated_at=datetime.now() - timedelta(hours=1), + ) + + assert store._is_session_expired(entry) is False + # --------------------------------------------------------------------------- # SessionEntry captures reason