From 206ab0943cf6bd121d16c14f52ac999d34327735 Mon Sep 17 00:00:00 2001 From: nanobot Date: Tue, 28 Jul 2026 14:06:26 +0800 Subject: [PATCH] --- nanobot/agent/loop.py | 121 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 105 insertions(+), 16 deletions(-) diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 90effcb1..b5fe774c 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -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, + )