diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index d11b5b67..2332fe2e 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -5,7 +5,7 @@ import os import select import signal import sys -from collections.abc import Callable +from collections.abc import Callable, Iterable from contextlib import nullcontext, suppress from pathlib import Path from typing import Any @@ -61,6 +61,7 @@ from nanobot.utils.restart import ( # noqa: E402 format_restart_completed_message, should_show_cli_restart_notice, ) +from nanobot.webui.sidebar_state import read_webui_sidebar_state # noqa: E402 def _sanitize_surrogates(text: str) -> str: @@ -210,6 +211,29 @@ def _heartbeat_has_active_tasks(content: str) -> bool: return True return False + +def _pick_heartbeat_target_from_sessions( + *, + enabled_channels: Iterable[str], + sessions: Iterable[dict[str, Any]], + archived_keys: Iterable[str], +) -> tuple[str, str]: + enabled = set(enabled_channels) + archived = set(archived_keys) + for item in sessions: + key = item.get("key") or "" + if key in archived: + continue + if ":" not in key: + continue + channel, chat_id = key.split(":", 1) + if channel in {"cli", "system"}: + continue + if channel in enabled and chat_id: + return channel, chat_id + return "cli", "direct" + + # --------------------------------------------------------------------------- # CLI input: prompt_toolkit for editing, paste, history, and display # --------------------------------------------------------------------------- @@ -1064,24 +1088,12 @@ def _run_gateway( def _pick_heartbeat_target() -> tuple[str, str]: """Pick a routable channel/chat target for heartbeat-triggered messages.""" - enabled = set(channels.enabled_channels) - - from nanobot.webui.sidebar_state import read_webui_sidebar_state sidebar_state = read_webui_sidebar_state() - archived_keys = set(sidebar_state.get("archived_keys", [])) - - for item in session_manager.list_sessions(): - key = item.get("key") or "" - if key in archived_keys: - continue - if ":" not in key: - continue - channel, chat_id = key.split(":", 1) - if channel in {"cli", "system"}: - continue - if channel in enabled and chat_id: - return channel, chat_id - return "cli", "direct" + return _pick_heartbeat_target_from_sessions( + enabled_channels=channels.enabled_channels, + sessions=session_manager.list_sessions(), + archived_keys=sidebar_state.get("archived_keys", []), + ) if channels.enabled_channels: console.print(f"[green]✓[/green] Channels enabled: {', '.join(channels.enabled_channels)}") diff --git a/nanobot/session/manager.py b/nanobot/session/manager.py index da62f0f0..331ccf95 100644 --- a/nanobot/session/manager.py +++ b/nanobot/session/manager.py @@ -843,7 +843,6 @@ class SessionManager: if not fallback_preview and item.get("role") == "assistant": fallback_preview = text preview = preview or fallback_preview - from datetime import datetime fallback_time = datetime.fromtimestamp(path.stat().st_mtime).isoformat() sessions.append( { diff --git a/nanobot/webui/session_list_index.py b/nanobot/webui/session_list_index.py index fdd64575..8269500c 100644 --- a/nanobot/webui/session_list_index.py +++ b/nanobot/webui/session_list_index.py @@ -272,7 +272,6 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, created_at_s = data.get("created_at") updated_at_s = data.get("updated_at") if not created_at_s or not updated_at_s: - from datetime import datetime fallback_time = datetime.fromtimestamp(signature["mtime_ns"] / 1e9).isoformat() created_at_s = created_at_s or fallback_time updated_at_s = updated_at_s or fallback_time diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index aeb67e8b..3a6ad286 100644 --- a/tests/cli/test_commands.py +++ b/tests/cli/test_commands.py @@ -1247,6 +1247,21 @@ def test_heartbeat_skips_bundled_template(): assert _heartbeat_has_active_tasks(load_bundled_template("HEARTBEAT.md")) is False +def test_heartbeat_target_skips_archived_webui_sessions(): + from nanobot.cli.commands import _pick_heartbeat_target_from_sessions + + target = _pick_heartbeat_target_from_sessions( + enabled_channels=["websocket"], + archived_keys=["websocket:archived"], + sessions=[ + {"key": "websocket:archived"}, + {"key": "websocket:active"}, + ], + ) + + assert target == ("websocket", "active") + + def _write_instance_config(tmp_path: Path) -> Path: config_file = tmp_path / "instance" / "config.json" config_file.parent.mkdir(parents=True)