feat: notify user and re-trigger interrupted turns on session recovery
recover_stale_sessions now publishes a restart notification to the user channel and re-publishes the last user message for sessions with a pending turn, so interrupted work continues automatically after a crash or restart. Also skip checkpoint restore during shutdown so pending markers survive for startup recovery.
This commit is contained in:
nanobot
+105
-16
@@ -74,7 +74,11 @@ from nanobot.session.goal_state import (
|
||||
sustained_goal_active,
|
||||
)
|
||||
from nanobot.session.history_visibility import HIDDEN_HISTORY_META
|
||||
from nanobot.session.keys import UNIFIED_SESSION_KEY, remember_last_channel
|
||||
from nanobot.session.keys import (
|
||||
UNIFIED_SESSION_KEY,
|
||||
last_channel_from_metadata,
|
||||
remember_last_channel,
|
||||
)
|
||||
from nanobot.session.manager import (
|
||||
Session,
|
||||
SessionManager,
|
||||
@@ -1219,22 +1223,28 @@ class AgentLoop:
|
||||
# _emit_checkpoint during tool execution; materializing
|
||||
# it into session history now makes it visible in the
|
||||
# next conversation turn.
|
||||
try:
|
||||
key = self._effective_session_key(msg)
|
||||
session = self.sessions.get_or_create(key)
|
||||
if self._restore_runtime_checkpoint(session):
|
||||
self._clear_pending_user_turn(session)
|
||||
self.sessions.save(session)
|
||||
logger.info(
|
||||
"Restored partial context for cancelled session {}",
|
||||
key,
|
||||
#
|
||||
# During gateway shutdown (self._running is False), skip
|
||||
# this so the pending markers survive and
|
||||
# recover_stale_sessions() can notify the user and
|
||||
# re-trigger the interrupted turn on next startup.
|
||||
if self._running:
|
||||
try:
|
||||
key = self._effective_session_key(msg)
|
||||
session = self.sessions.get_or_create(key)
|
||||
if self._restore_runtime_checkpoint(session):
|
||||
self._clear_pending_user_turn(session)
|
||||
self.sessions.save(session)
|
||||
logger.info(
|
||||
"Restored partial context for cancelled session {}",
|
||||
key,
|
||||
)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Could not restore checkpoint for cancelled session {}",
|
||||
session_key,
|
||||
exc_info=True,
|
||||
)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Could not restore checkpoint for cancelled session {}",
|
||||
session_key,
|
||||
exc_info=True,
|
||||
)
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.exception("Error processing message for session {}", session_key)
|
||||
@@ -2042,6 +2052,10 @@ class AgentLoop:
|
||||
def recover_stale_sessions(self) -> int:
|
||||
"""Scan all sessions on startup and recover any with stale turn state.
|
||||
|
||||
For each recovered session, publishes a restart notification to the
|
||||
user and re-triggers any interrupted turn so work continues
|
||||
automatically after a crash or restart.
|
||||
|
||||
Returns the number of sessions that were recovered.
|
||||
"""
|
||||
recovered = 0
|
||||
@@ -2049,10 +2063,12 @@ class AgentLoop:
|
||||
try:
|
||||
session = self.sessions.get_or_create(key)
|
||||
changed = False
|
||||
had_pending = False
|
||||
if self._restore_runtime_checkpoint(session):
|
||||
changed = True
|
||||
if self._restore_pending_user_turn(session):
|
||||
changed = True
|
||||
had_pending = True
|
||||
if changed:
|
||||
self.sessions.save(session)
|
||||
recovered += 1
|
||||
@@ -2060,6 +2076,8 @@ class AgentLoop:
|
||||
"Recovered stale session {} on startup",
|
||||
key,
|
||||
)
|
||||
# Notify the user and re-trigger interrupted work.
|
||||
self._notify_session_recovered(key, session, had_pending)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Could not recover stale session {}",
|
||||
@@ -2069,3 +2087,74 @@ class AgentLoop:
|
||||
if recovered:
|
||||
logger.info("Recovered {} stale session(s) on startup", recovered)
|
||||
return recovered
|
||||
|
||||
def _notify_session_recovered(
|
||||
self,
|
||||
session_key: str,
|
||||
session: Session,
|
||||
had_pending_turn: bool,
|
||||
) -> None:
|
||||
"""Publish restart notification and re-trigger interrupted turns.
|
||||
|
||||
Uses the bus queues directly so messages are delivered once the
|
||||
outbound dispatcher and agent loop start.
|
||||
"""
|
||||
route = last_channel_from_metadata(session.metadata)
|
||||
if route is None:
|
||||
# Fall back to parsing the session key (format: channel:chat_id).
|
||||
if ":" in session_key and not session_key.startswith("unified:"):
|
||||
parts = session_key.split(":", 1)
|
||||
if parts[0] and parts[1]:
|
||||
route = (parts[0], parts[1])
|
||||
if route is None:
|
||||
logger.debug(
|
||||
"Cannot determine route for recovered session {}",
|
||||
session_key,
|
||||
)
|
||||
return
|
||||
channel, chat_id = route
|
||||
# Notify the user that a restart happened and the session was recovered.
|
||||
try:
|
||||
self.bus.outbound.put_nowait(
|
||||
OutboundMessage(
|
||||
channel=channel,
|
||||
chat_id=chat_id,
|
||||
content=(
|
||||
"🔄 nanobot was restarted (crash or deploy). "
|
||||
"Your session has been recovered."
|
||||
),
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Could not enqueue restart notification for {}",
|
||||
session_key,
|
||||
exc_info=True,
|
||||
)
|
||||
# Re-trigger the interrupted turn so work continues automatically.
|
||||
if had_pending_turn:
|
||||
last_user_msg = None
|
||||
for msg in reversed(session.messages):
|
||||
if msg.get("role") == "user":
|
||||
last_user_msg = msg
|
||||
break
|
||||
if last_user_msg and last_user_msg.get("content"):
|
||||
try:
|
||||
self.bus.inbound.put_nowait(
|
||||
InboundMessage(
|
||||
channel=channel,
|
||||
sender_id=chat_id,
|
||||
chat_id=chat_id,
|
||||
content=last_user_msg["content"],
|
||||
)
|
||||
)
|
||||
logger.info(
|
||||
"Re-published user message for recovered session {}",
|
||||
session_key,
|
||||
)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Could not re-publish message for {}",
|
||||
session_key,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user