Skip to content

fix(gateway): notify user when stale-routing self-heal can't recover the old session #59597

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hejuntt1014 wants to merge 1 commit into
base: main
Choose a base branch
from
+345 −63
Open

fix(gateway): notify user when stale-routing self-heal can't recover the old session #59597

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
168 changes: 108 additions & 60 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1911,6 +1911,7 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor
)
from gateway.session import (
AsyncSessionStore,
SessionEntry,
SessionStore,
SessionSource,
SessionContext,
Expand Down Expand Up @@ -12098,68 +12099,11 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
# so the agent knows this is a fresh conversation (not an intentional /reset).
if _was_auto_reset:
reset_reason = getattr(session_entry, 'auto_reset_reason', None) or 'idle'
if reset_reason == "suspended":
context_note = "[System note: The user's previous session was stopped and suspended. This is a fresh conversation with no prior context.]"
elif reset_reason == "daily":
context_note = "[System note: The user's session was automatically reset by the daily schedule. This is a fresh conversation with no prior context.]"
elif reset_reason == "resume_pending_expired":
context_note = "[System note: The previous gateway session could not be recovered after a restart (API recovery timed out). This is a fresh conversation — use /resume to restore history if needed.]"
else:
context_note = "[System note: The user's previous session expired due to inactivity. This is a fresh conversation with no prior context.]"
turn_sidecar_notes.append(context_note)

# Send a user-facing notification explaining the reset, unless:
# - notifications are disabled in config
# - the platform is excluded (e.g. api_server, webhook)
# - the expired session had no activity (nothing was cleared)
turn_sidecar_notes.append(self._auto_reset_context_note(reset_reason))
try:
policy = self.session_store.config.get_reset_policy(
platform=source.platform,
session_type=getattr(source, 'chat_type', 'dm'),
)
platform_name = source.platform.value if source.platform else ""
had_activity = getattr(session_entry, 'reset_had_activity', False)
# Suspended and restart-recovery-expired sessions always notify
# regardless of policy.notify — the user had an active session
# that was silently replaced, so they need to know they can
# /resume it. Idle/daily resets respect the policy flag.
should_notify = reset_reason in {"suspended", "resume_pending_expired"} or (
policy.notify
and had_activity
and platform_name not in policy.notify_exclude_platforms
await self._maybe_send_auto_reset_notice(
source, session_entry, reset_reason,
)
if should_notify:
adapter = self._adapter_for_source(source)
if adapter:
if reset_reason == "suspended":
reason_text = "previous session was stopped or interrupted"
elif reset_reason == "resume_pending_expired":
reason_text = "gateway restart recovery timed out"
elif reset_reason == "daily":
reason_text = f"daily schedule at {policy.at_hour}:00"
else:
hours = policy.idle_minutes // 60
mins = policy.idle_minutes % 60
duration = f"{hours}h" if not mins else f"{hours}h {mins}m" if hours else f"{mins}m"
reason_text = f"inactive for {duration}"
notice = (
f"◐ Session automatically reset ({reason_text}). "
f"Conversation history cleared.\n"
f"Use /resume to browse and restore a previous session.\n"
f"Adjust reset timing in config.yaml under session_reset."
)
try:
session_info = await asyncio.to_thread(
self._reset_notice_session_info, source
)
if session_info:
notice = f"{notice}\n\n{session_info}"
except Exception:
pass
await adapter.send(
source.chat_id, notice,
metadata=self._thread_metadata_for_source(source),
)
except Exception as e:
logger.debug("Auto-reset notification failed (non-fatal): %s", e)

Expand Down Expand Up @@ -13569,6 +13513,110 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
# Restore session context variables to their pre-handler state
self._clear_session_env(_session_env_tokens)

@staticmethod
def _auto_reset_context_note(reset_reason: str) -> str:
"""Agent-facing system note for an auto-reset turn."""
if reset_reason == "suspended":
return (
"[System note: The user's previous session was stopped and "
"suspended. This is a fresh conversation with no prior context.]"
)
if reset_reason == "daily":
return (
"[System note: The user's session was automatically reset by "
"the daily schedule. This is a fresh conversation with no prior context.]"
)
if reset_reason == "resume_pending_expired":
return (
"[System note: The previous gateway session could not be "
"recovered after a restart (API recovery timed out). This is a "
"fresh conversation — use /resume to restore history if needed.]"
)
if reset_reason == "stale_routing_recovered":
return (
"[System note: The previous session ended in a way that could "
"not be automatically resumed (e.g. the client disconnected). "
"This is a fresh conversation — use /resume to restore history "
"if needed.]"
)
return (
"[System note: The user's previous session expired due to "
"inactivity. This is a fresh conversation with no prior context.]"
)

async def _maybe_send_auto_reset_notice(
self,
source: SessionSource,
session_entry: SessionEntry,
reset_reason: str,
) -> None:
"""Send a user-facing auto-reset notice via the platform adapter.

Suspended, restart-recovery-expired, and stale-routing-reset sessions
always notify regardless of ``policy.notify`` — the user had an active
session that was silently replaced, so they need to know they can
``/resume`` it. Idle/daily resets respect the policy flag.
"""
policy = self.session_store.config.get_reset_policy(
platform=source.platform,
session_type=getattr(source, "chat_type", "dm"),
)
platform_name = source.platform.value if source.platform else ""
had_activity = getattr(session_entry, "reset_had_activity", False)
should_notify = reset_reason in {
"suspended",
"resume_pending_expired",
"stale_routing_recovered",
} or (
policy.notify
and had_activity
and platform_name not in policy.notify_exclude_platforms
)
if not should_notify:
return

adapter = self._adapter_for_source(source)
if not adapter:
return

if reset_reason == "suspended":
reason_text = "previous session was stopped or interrupted"
elif reset_reason == "resume_pending_expired":
reason_text = "gateway restart recovery timed out"
elif reset_reason == "stale_routing_recovered":
reason_text = "previous session ended and could not be auto-resumed"
elif reset_reason == "daily":
reason_text = f"daily schedule at {policy.at_hour}:00"
else:
hours = policy.idle_minutes // 60
mins = policy.idle_minutes % 60
duration = (
f"{hours}h" if not mins
else f"{hours}h {mins}m" if hours
else f"{mins}m"
)
reason_text = f"inactive for {duration}"

notice = (
f"◐ Session automatically reset ({reason_text}). "
f"Conversation history cleared.\n"
f"Use /resume to browse and restore a previous session.\n"
f"Adjust reset timing in config.yaml under session_reset."
)
try:
session_info = await asyncio.to_thread(
self._reset_notice_session_info, source
)
if session_info:
notice = f"{notice}\n\n{session_info}"
except Exception:
pass
await adapter.send(
source.chat_id,
notice,
metadata=self._thread_metadata_for_source(source),
)

def _reset_notice_session_info(self, source: SessionSource) -> str:
"""Session-info block for the auto-reset notice, profile-scoped.

Expand Down
36 changes: 34 additions & 2 deletions gateway/session.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1976,6 +1976,15 @@ def _get_or_create_session_impl(
was_auto_reset = False
auto_reset_reason = None
reset_had_activity = False
# Set when the #54878 self-heal drops a routing entry whose session
# was already ended in state.db under a reason the recovery finder
# doesn't consider resumable (e.g. "tui_shutdown" — anything other
# than "agent_close" / still-open). Used after the recovery attempt
# to decide whether the user needs to be told their old thread is
# gone, since that path otherwise falls through to "Create new
# session" in total silence (#59580).
stale_routing_dropped_session_id = None
stale_routing_had_activity = False

with self._lock:
self._ensure_loaded_locked()
Expand All @@ -2001,15 +2010,23 @@ def _get_or_create_session_impl(
"(#54878)",
session_key, entry.session_id,
)
# Snapshot before pop — recovery may still reopen the row
# (agent_close); if it can't, these drive the user-facing
# stale_routing_recovered notice (#59580).
_dropped_session_id = entry.session_id
_dropped_had_activity = entry.last_prompt_tokens > 0
self._entries.pop(session_key, None)
# If an expiry watcher (daily/idle reset) already finalized
# this session, honour the reset decision instead of silently
# reopening it via recovery.
if _reset_reason:
was_auto_reset = True
auto_reset_reason = _reset_reason
reset_had_activity = entry.last_prompt_tokens > 0
db_end_session_id = entry.session_id
reset_had_activity = _dropped_had_activity
db_end_session_id = _dropped_session_id
else:
stale_routing_dropped_session_id = _dropped_session_id
stale_routing_had_activity = _dropped_had_activity
entry = None
_needs_recover = True
elif entry.session_id != _stale_session_id:
Expand Down Expand Up @@ -2049,6 +2066,21 @@ def _get_or_create_session_impl(
_needs_save = True

if entry is None:
# If the #54878 self-heal dropped a stale routing entry above and
# recovery just failed to reopen it (end_reason was outside the
# ended_at IS NULL / 'agent_close' whitelist — e.g. 'tui_shutdown'),
# the user is about to receive a brand-new, empty session in place
# of a thread they may believe is still live. That's the same
# "your previous session was silently replaced" situation as any
# other auto-reset, so surface it the same way instead of staying
# silent. NOTE: db_end_session_id is intentionally left unset here
# — the old session was already ended (with its real reason) by
# whatever finalized it, and we must not overwrite that reason.
if stale_routing_dropped_session_id is not None and not was_auto_reset:
was_auto_reset = True
auto_reset_reason = "stale_routing_recovered"
reset_had_activity = stale_routing_had_activity

# Create a candidate outside the lock, then publish only if another
# worker has not already populated this routing key.
session_id = f"{now.strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}"
Expand Down
Loading
Loading