fix(gateway): fail closed on active-process check errors

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:

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