From 97ebe3ee4b983927ee906c50574249b14a9e1c52 Mon Sep 17 00:00:00 2001 From: serxa Date: Fri, 17 Jul 2026 22:29:43 +0000 Subject: [PATCH 1/2] Add opt-in interactive-session idle auto-close (sessions.archive_after_hours) Interactive chat sessions were only cleaned by the 30-day staleness backstop and the max_sessions cap. Leaving them open for days delays memory consolidation (a session is memorized on close), keeps stale context around to be reloaded/re-cached on resume, silently resumes day-old threads, and lets an unanswered ask_user question dangle. This adds an opt-in short idle cutoff to address that. sessions.archive_after_hours (default 0 = disabled) archives interactive sessions (web/telegram/discord/slack/whatsapp) idle beyond it, memorizing them and expiring any pending question; cron/persistent sessions are excluded. Default behavior is unchanged: the block is skipped when 0, and the cleanup cadence stays 6h unless the feature is enabled (then hourly). Adds additive SessionStore.get_stale_interactive_sessions + NotificationStore.expire_pending_questions_for_session, and tests incl. a default-unchanged case. Co-Authored-By: Claude Opus 4.8 --- nerve/agent/sessions.py | 35 +++++++++++++++++++++++++++++-- nerve/config.py | 2 ++ nerve/db/notifications.py | 23 +++++++++++++++++++++ nerve/db/sessions.py | 21 +++++++++++++++++++ nerve/gateway/server.py | 16 ++++++++++++--- tests/test_sessions.py | 43 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 135 insertions(+), 5 deletions(-) diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py index 758f1f0..340951b 100644 --- a/nerve/agent/sessions.py +++ b/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, + archive_after_hours: int = 0, ) -> dict: """Auto-archive stale sessions and enforce limits. - Returns dict with cleanup statistics. + When ``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 archive_after_hours == 0, so default behavior is unchanged. + archived_interactive = 0 + if archive_after_hours and archive_after_hours > 0: + icutoff = (now - timedelta(hours=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 diff --git a/nerve/config.py b/nerve/config.py index a97e0d7..2081d0d 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -635,6 +635,7 @@ def from_dict(cls, d: dict) -> BackupConfig: @dataclass class SessionsConfig: archive_after_days: int = 30 + 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), + archive_after_hours=d.get("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), diff --git a/nerve/db/notifications.py b/nerve/db/notifications.py index 6204bc8..17239d1 100644 --- a/nerve/db/notifications.py +++ b/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: diff --git a/nerve/db/sessions.py b/nerve/db/sessions.py index 9064bd7..e511756 100644 --- a/nerve/db/sessions.py +++ b/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( diff --git a/nerve/gateway/server.py b/nerve/gateway/server.py index e569ab6..df34e28 100644 --- a/nerve/gateway/server.py +++ b/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.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.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, + archive_after_hours=config.sessions.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) diff --git a/tests/test_sessions.py b/tests/test_sessions.py index fb5a0f5..64f508d 100644 --- a/tests/test_sessions.py +++ b/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 (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, 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, 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, archive_after_hours=1) + + assert (await db.get_session("idle-cron"))["status"] == "idle" + @pytest.mark.asyncio class TestOrphanRecovery: From a28f66c6ae3f3ce28cdd56ae3b6321025b1794eb Mon Sep 17 00:00:00 2001 From: serxa Date: Fri, 17 Jul 2026 22:42:25 +0000 Subject: [PATCH 2/2] Rename archive_after_hours -> interactive_archive_after_hours for clarity The knob is scoped to interactive sources only (distinct from the global archive_after_days backstop that covers cron/persistent/external too), so the explicit name avoids it reading like a finer-grained archive_after_days. Co-Authored-By: Claude Opus 4.8 --- nerve/agent/sessions.py | 10 +++++----- nerve/config.py | 4 ++-- nerve/gateway/server.py | 6 +++--- tests/test_sessions.py | 8 ++++---- 4 files changed, 14 insertions(+), 14 deletions(-) diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py index 340951b..501d7a4 100644 --- a/nerve/agent/sessions.py +++ b/nerve/agent/sessions.py @@ -618,11 +618,11 @@ async def run_cleanup( self, archive_after_days: int = DEFAULT_ARCHIVE_AFTER_DAYS, max_sessions: int = DEFAULT_MAX_SESSIONS, - archive_after_hours: int = 0, + interactive_archive_after_hours: int = 0, ) -> dict: """Auto-archive stale sessions and enforce limits. - When ``archive_after_hours`` > 0 (opt-in; 0 disables and leaves the + 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 @@ -633,10 +633,10 @@ async def run_cleanup( now = datetime.now(timezone.utc) # Opt-in short idle cutoff for interactive sessions. Skipped entirely - # when archive_after_hours == 0, so default behavior is unchanged. + # when interactive_archive_after_hours == 0, so default behavior is unchanged. archived_interactive = 0 - if archive_after_hours and archive_after_hours > 0: - icutoff = (now - timedelta(hours=archive_after_hours)).isoformat() + 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, ) diff --git a/nerve/config.py b/nerve/config.py index 2081d0d..eaee664 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -635,7 +635,7 @@ def from_dict(cls, d: dict) -> BackupConfig: @dataclass class SessionsConfig: archive_after_days: int = 30 - archive_after_hours: int = 0 # Interactive (web/telegram/…) sessions auto-close after this many idle hours (0 = disabled; opt in via config) + 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 @@ -646,7 +646,7 @@ class SessionsConfig: def from_dict(cls, d: dict) -> SessionsConfig: return cls( archive_after_days=d.get("archive_after_days", 30), - archive_after_hours=d.get("archive_after_hours", 0), + 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), diff --git a/nerve/gateway/server.py b/nerve/gateway/server.py index df34e28..81cae50 100644 --- a/nerve/gateway/server.py +++ b/nerve/gateway/server.py @@ -239,11 +239,11 @@ async def lifespan(app: FastAPI): # Periodic session cleanup. Default cadence is every 6 hours (unchanged); # it tightens to hourly only when the opt-in interactive idle auto-close - # (sessions.archive_after_hours > 0) is enabled and needs finer resolution. + # (sessions.interactive_archive_after_hours > 0) is enabled and needs finer resolution. async def _periodic_cleanup(): while True: interval = ( - 3600 if config.sessions.archive_after_hours > 0 else 6 * 3600 + 3600 if config.sessions.interactive_archive_after_hours > 0 else 6 * 3600 ) await asyncio.sleep(interval) try: @@ -251,7 +251,7 @@ async def _periodic_cleanup(): stats = await _engine.sessions.run_cleanup( archive_after_days=config.sessions.archive_after_days, max_sessions=config.sessions.max_sessions, - archive_after_hours=config.sessions.archive_after_hours, + interactive_archive_after_hours=config.sessions.interactive_archive_after_hours, ) if ( stats.get("archived_stale") diff --git a/tests/test_sessions.py b/tests/test_sessions.py index 64f508d..2d574c2 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -426,11 +426,11 @@ async def _set_idle_hours_ago(self, db: Database, sid: str, hours: int): async def test_cleanup_hours_disabled_leaves_interactive_untouched( self, sm: SessionManager, db: Database, ): - """Default (archive_after_hours=0) must NOT close idle interactive sessions.""" + """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, archive_after_hours=0) + 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" @@ -441,7 +441,7 @@ async def test_cleanup_hours_enabled_archives_idle_interactive( 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, archive_after_hours=1) + 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" @@ -453,7 +453,7 @@ async def test_cleanup_hours_excludes_cron_sessions( 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, archive_after_hours=1) + await sm.run_cleanup(archive_after_days=30, interactive_archive_after_hours=1) assert (await db.get_session("idle-cron"))["status"] == "idle"