feat: recover stale sessions on gateway startup
Test Suite / Detect changes (push) Has been cancelled
Test Suite / Python (Windows, 3.14) (push) Has been cancelled
Test Suite / Python (minimum, 3.11) (push) Has been cancelled
Test Suite / Python (latest, 3.14 + coverage) (push) Has been cancelled
Test Suite / webui (push) Has been cancelled
Test Suite / docker (push) Has been cancelled

Scan all session files on startup and recover any with stale turn state
(pending_user_turn or runtime_checkpoint metadata) left from a previous
crash or restart. Previously this only happened lazily when a new message
arrived for the affected session.

- Add SessionManager.list_session_keys() to enumerate session keys from disk
- Add AgentLoop.recover_stale_sessions() to scan and recover all sessions
- Call recover_stale_sessions() during gateway startup in _run_gateway
This commit is contained in:
nanobot
2026-07-27 21:27:29 +08:00
parent e819b7eea4
commit 99b86a22c8
3 changed files with 55 additions and 0 deletions
+31
View File
@@ -2038,3 +2038,34 @@ class AgentLoop:
finally: finally:
await self._runtime_events().run_status_changed(msg, session_key, "idle") await self._runtime_events().run_status_changed(msg, session_key, "idle")
self._runtime_events().clear_turn(session_key) self._runtime_events().clear_turn(session_key)
def recover_stale_sessions(self) -> int:
"""Scan all sessions on startup and recover any with stale turn state.
Returns the number of sessions that were recovered.
"""
recovered = 0
for key in self.sessions.list_session_keys():
try:
session = self.sessions.get_or_create(key)
changed = False
if self._restore_runtime_checkpoint(session):
changed = True
if self._restore_pending_user_turn(session):
changed = True
if changed:
self.sessions.save(session)
recovered += 1
logger.info(
"Recovered stale session {} on startup",
key,
)
except Exception:
logger.debug(
"Could not recover stale session {}",
key,
exc_info=True,
)
if recovered:
logger.info("Recovered {} stale session(s) on startup", recovered)
return recovered
+4
View File
@@ -1744,6 +1744,10 @@ def _run_gateway(
local_trigger_store=trigger_store, local_trigger_store=trigger_store,
hook_factories=[create_file_edit_activity_hook], hook_factories=[create_file_edit_activity_hook],
) )
# Recover any sessions left in a stale state from a previous crash/restart.
agent.recover_stale_sessions()
webui_turn_coordinator = WebuiTurnCoordinator( webui_turn_coordinator = WebuiTurnCoordinator(
bus=bus, bus=bus,
sessions=session_manager, sessions=session_manager,
+20
View File
@@ -965,3 +965,23 @@ class SessionManager:
) )
continue continue
return sorted(sessions, key=lambda x: x.get("updated_at", ""), reverse=True) return sorted(sessions, key=lambda x: x.get("updated_at", ""), reverse=True)
def list_session_keys(self) -> list[str]:
"""List all session keys from disk without loading full sessions."""
keys: list[str] = []
for path in self.sessions_dir.glob("*.jsonl"):
storage_key = self._session_key_from_path(path)
if storage_key is None:
continue
try:
with open(path, encoding="utf-8") as f:
first_line = f.readline().strip()
if first_line:
data = json.loads(first_line)
if isinstance(data, dict) and data.get("_type") == "metadata":
keys.append(data.get("key") or storage_key)
else:
keys.append(storage_key)
except (OSError, _SESSION_DATA_ERRORS):
continue
return keys