Add opt-in interactive-session idle auto-close (sessions.interactive_archive_after_hours)

nerve/agent/sessions.py

@@ -25,6 +25,9 @@
 # Cleanup defaults
 DEFAULT_ARCHIVE_AFTER_DAYS = 30
 DEFAULT_MAX_SESSIONS = 500
+# Sources treated as interactive for the opt-in short idle auto-close. Cron and
+# source-runner sessions are excluded so their context continuity is preserved.
+INTERACTIVE_SOURCES = ["web", "telegram", "discord", "slack", "whatsapp"]
 
 
 class SessionStatus(StrEnum):
@@ -615,12 +618,39 @@ async def run_cleanup(
         self,
         archive_after_days: int = DEFAULT_ARCHIVE_AFTER_DAYS,
         max_sessions: int = DEFAULT_MAX_SESSIONS,
+        interactive_archive_after_hours: int = 0,
     ) -> dict:
         """Auto-archive stale sessions and enforce limits.
 
-        Returns dict with cleanup statistics.
+        When ``interactive_archive_after_hours`` > 0 (opt-in; 0 disables and leaves the
+        default behavior unchanged), interactive sessions (web/telegram/…)
+        idle longer than that are archived and memorized promptly — including
+        ones parked on an unanswered ``ask_user`` question, whose pending
+        notification is expired so it doesn't dangle on a closed session.
+        Cron/persistent sessions are excluded from the short cutoff and only
+        hit the ``archive_after_days`` backstop. Returns cleanup statistics.
         """
         now = datetime.now(timezone.utc)
+
+        # Opt-in short idle cutoff for interactive sessions. Skipped entirely
+        # when interactive_archive_after_hours == 0, so default behavior is unchanged.
+        archived_interactive = 0
+        if interactive_archive_after_hours and interactive_archive_after_hours > 0:
+            icutoff = (now - timedelta(hours=interactive_archive_after_hours)).isoformat()
+            interactive = await self.db.get_stale_interactive_sessions(
+                icutoff, INTERACTIVE_SOURCES,
+            )
+            for s in interactive:
+                try:
+                    await self.db.expire_pending_questions_for_session(s["id"])
+                except Exception as e:
+                    logger.warning(
+                        "Expire pending questions failed for %s: %s", s["id"], e,
+                    )
+                await self.archive_session(s["id"])
+            archived_interactive = len(interactive)
+
+        # Long backstop cutoff for everything (incl. non-interactive).
         cutoff = (now - timedelta(days=archive_after_days)).isoformat()
 
         # Archive idle/stopped sessions older than cutoff
@@ -640,10 +670,11 @@ async def run_cleanup(
             overflow = len(excess)
 
         stats = {
+            "archived_interactive": archived_interactive,
             "archived_stale": len(stale),
             "archived_overflow": overflow,
         }
-        if stale or overflow:
+        if archived_interactive or stale or overflow:
             logger.info("Session cleanup: %s", stats)
         return stats
 

nerve/config.py

@@ -635,6 +635,7 @@ def from_dict(cls, d: dict) -> BackupConfig:
 @dataclass
 class SessionsConfig:
     archive_after_days: int = 30
+    interactive_archive_after_hours: int = 0  # Interactive (web/telegram/…) sessions auto-close after this many idle hours (0 = disabled; opt in via config)
     max_sessions: int = 500
     cron_session_mode: str = "per_run"  # "per_run" or "reuse"
     memorize_interval_minutes: int = 30  # Background memorization sweep interval
@@ -645,6 +646,7 @@ class SessionsConfig:
     def from_dict(cls, d: dict) -> SessionsConfig:
         return cls(
             archive_after_days=d.get("archive_after_days", 30),
+            interactive_archive_after_hours=d.get("interactive_archive_after_hours", 0),
             max_sessions=d.get("max_sessions", 500),
             cron_session_mode=d.get("cron_session_mode", "per_run"),
             memorize_interval_minutes=d.get("memorize_interval_minutes", 30),

nerve/db/notifications.py

@@ -187,6 +187,29 @@ async def expire_notification(self, notification_id: str) -> bool:
             )
         return True
 
+    async def expire_pending_questions_for_session(self, session_id: str) -> int:
+        """Expire a session's pending ``question`` notifications.
+
+        Called when an idle session is auto-archived so the user isn't left
+        with a phantom pending question on a closed session (and the periodic
+        expiry pass has nothing to inject into the now-archived session).
+        Returns the number of questions expired.
+        """
+        async with self._atomic():
+            async with self.db.execute(
+                """SELECT id FROM notifications
+                   WHERE session_id = ? AND status = 'pending' AND type = 'question'""",
+                (session_id,),
+            ) as cursor:
+                ids = [row[0] async for row in cursor]
+            if ids:
+                ph = ",".join("?" for _ in ids)
+                await self.db.execute(
+                    f"UPDATE notifications SET status = 'expired' WHERE id IN ({ph})",
+                    tuple(ids),
+                )
+        return len(ids)
+
     async def snooze_notification(
         self, notification_id: str, redeliver_at: str, new_expires_at: str,
     ) -> bool:

nerve/db/sessions.py

@@ -303,6 +303,27 @@ async def get_stale_sessions(
         async with self.db.execute(query, params) as cursor:
             return [dict(row) async for row in cursor]
 
+    async def get_stale_interactive_sessions(
+        self, before_iso: str, sources: list[str],
+    ) -> list[dict]:
+        """Idle/stopped/error *interactive* sessions not updated since before_iso.
+
+        Scoped by ``source`` (web/telegram/…) so cron and persistent sessions
+        keep their own long rotation and are never archived at the short
+        interactive cutoff. Returns [] when ``sources`` is empty.
+        """
+        if not sources:
+            return []
+        src = ",".join("?" for _ in sources)
+        query = f"""
+            SELECT * FROM sessions
+            WHERE status IN ('idle', 'stopped', 'error')
+              AND source IN ({src})
+              AND updated_at < ?
+        """
+        async with self.db.execute(query, (*sources, before_iso)) as cursor:
+            return [dict(row) async for row in cursor]
+
     async def count_active_sessions(self) -> int:
         """Count non-archived sessions."""
         async with self.db.execute(

nerve/gateway/server.py

@@ -237,17 +237,27 @@ async def lifespan(app: FastAPI):
     except Exception as e:
         logger.warning("Cron service failed to start: %s", e)
 
-    # Periodic session cleanup (every 6 hours)
+    # Periodic session cleanup. Default cadence is every 6 hours (unchanged);
+    # it tightens to hourly only when the opt-in interactive idle auto-close
+    # (sessions.interactive_archive_after_hours > 0) is enabled and needs finer resolution.
     async def _periodic_cleanup():
         while True:
-            await asyncio.sleep(6 * 3600)
+            interval = (
+                3600 if config.sessions.interactive_archive_after_hours > 0 else 6 * 3600
+            )
+            await asyncio.sleep(interval)
             try:
                 if _engine:
                     stats = await _engine.sessions.run_cleanup(
                         archive_after_days=config.sessions.archive_after_days,
                         max_sessions=config.sessions.max_sessions,
+                        interactive_archive_after_hours=config.sessions.interactive_archive_after_hours,
                     )
-                    if stats.get("archived_stale") or stats.get("archived_overflow"):
+                    if (
+                        stats.get("archived_stale")
+                        or stats.get("archived_overflow")
+                        or stats.get("archived_interactive")
+                    ):
                         logger.info("Session cleanup: %s", stats)
             except Exception as e:
                 logger.error("Session cleanup failed: %s", e)

tests/test_sessions.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import contextlib
+from datetime import datetime, timedelta, timezone
 
 import pytest
 import pytest_asyncio
@@ -414,6 +415,48 @@ async def test_cleanup_archives_all_stale_sessions(self, sm: SessionManager, db:
         session = await db.get_session("cleanup-any")
         assert session["status"] == "archived"
 
+    async def _set_idle_hours_ago(self, db: Database, sid: str, hours: int):
+        ts = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
+        await db.update_session_fields(sid, {"status": "idle"})
+        await db.db.execute(
+            "UPDATE sessions SET updated_at = ? WHERE id = ?", (ts, sid),
+        )
+        await db.db.commit()
+
+    async def test_cleanup_hours_disabled_leaves_interactive_untouched(
+        self, sm: SessionManager, db: Database,
+    ):
+        """Default (interactive_archive_after_hours=0) must NOT close idle interactive sessions."""
+        await sm.get_or_create("idle-web", source="web")
+        await self._set_idle_hours_ago(db, "idle-web", hours=5)
+
+        stats = await sm.run_cleanup(archive_after_days=30, interactive_archive_after_hours=0)
+
+        assert stats["archived_interactive"] == 0
+        assert (await db.get_session("idle-web"))["status"] == "idle"
+
+    async def test_cleanup_hours_enabled_archives_idle_interactive(
+        self, sm: SessionManager, db: Database,
+    ):
+        await sm.get_or_create("idle-web2", source="web")
+        await self._set_idle_hours_ago(db, "idle-web2", hours=5)
+
+        stats = await sm.run_cleanup(archive_after_days=30, interactive_archive_after_hours=1)
+
+        assert stats["archived_interactive"] >= 1
+        assert (await db.get_session("idle-web2"))["status"] == "archived"
+
+    async def test_cleanup_hours_excludes_cron_sessions(
+        self, sm: SessionManager, db: Database,
+    ):
+        """Cron sessions are never subject to the short interactive cutoff."""
+        await sm.get_or_create("idle-cron", source="cron")
+        await self._set_idle_hours_ago(db, "idle-cron", hours=5)
+
+        await sm.run_cleanup(archive_after_days=30, interactive_archive_after_hours=1)
+
+        assert (await db.get_session("idle-cron"))["status"] == "idle"
+
 
 @pytest.mark.asyncio
 class TestOrphanRecovery: