test: cover archived heartbeat target selection

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", []))
+        return _pick_heartbeat_target_from_sessions(
-
+            enabled_channels=channels.enabled_channels,
-        for item in session_manager.list_sessions():
+            sessions=session_manager.list_sessions(),
-            key = item.get("key") or ""
+            archived_keys=sidebar_state.get("archived_keys", []),
-            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"
 
     if channels.enabled_channels:
         console.print(f"[green]✓[/green] Channels enabled: {', '.join(channels.enabled_channels)}")

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(
                                 {

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

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)