from __future__ import annotations import os from datetime import datetime from pathlib import Path import nanobot.webui.session_list_index as session_list_index from nanobot.cron.session_turns import CRON_HISTORY_META from nanobot.session.automation_turns import AUTOMATION_HISTORY_META from nanobot.session.history_visibility import HIDDEN_HISTORY_META from nanobot.session.manager import SessionManager def test_webui_session_list_reuses_valid_index_without_scanning_files( tmp_path: Path, monkeypatch, ) -> None: manager = SessionManager(tmp_path) session = manager.get_or_create("websocket:indexed") session.add_message("user", "indexed preview") manager.save(session) assert list_webui_sessions(manager)[0]["preview"] == "indexed preview" def fail_scan(session_manager: SessionManager, path: Path) -> None: raise AssertionError(f"unexpected session file scan: {path}") monkeypatch.setattr(session_list_index, "_scan_session_row", fail_scan) rows = list_webui_sessions(manager) assert rows[0]["key"] == "websocket:indexed" assert rows[0]["preview"] == "indexed preview" def test_webui_session_list_rescans_only_changed_file(tmp_path: Path, monkeypatch) -> None: manager = SessionManager(tmp_path) first = manager.get_or_create("websocket:first") first.add_message("user", "first") manager.save(first) second = manager.get_or_create("websocket:second") second.add_message("user", "second before") manager.save(second) assert {row["preview"] for row in list_webui_sessions(manager)} == {"first", "second before"} second.messages.clear() second.add_message("user", "second after") manager.save(second) original_scan = session_list_index._scan_session_row scanned: list[str] = [] def record_scan(session_manager: SessionManager, path: Path) -> dict | None: scanned.append(path.name) return original_scan(session_manager, path) monkeypatch.setattr(session_list_index, "_scan_session_row", record_scan) rows = list_webui_sessions(manager) assert scanned == [manager._get_session_path("websocket:second").name] assert {row["preview"] for row in rows} == {"first", "second after"} def test_webui_session_list_drops_deleted_index_rows(tmp_path: Path) -> None: manager = SessionManager(tmp_path) session = manager.get_or_create("websocket:deleted") session.add_message("user", "gone") manager.save(session) assert list_webui_sessions(manager)[0]["key"] == "websocket:deleted" assert manager.delete_session("websocket:deleted") is True assert list_webui_sessions(manager) == [] def test_webui_session_list_skips_cron_internal_user_preview(tmp_path: Path) -> None: manager = SessionManager(tmp_path) session = manager.get_or_create("websocket:cron-preview") session.add_message( "user", "Scheduled cron job triggered: 30s-test\n\nInternal reminder prompt", **{CRON_HISTORY_META: True}, ) session.add_message("assistant", "提醒已经到期。") manager.save(session) assert list_webui_sessions(manager)[0]["preview"] == "提醒已经到期。" def test_webui_session_list_skips_trigger_internal_user_preview(tmp_path: Path) -> None: manager = SessionManager(tmp_path) session = manager.get_or_create("websocket:trigger-preview") session.add_message( "user", "Local trigger received: PR review", **{AUTOMATION_HISTORY_META: {"kind": "local_trigger", "trigger_id": "trg_123"}}, ) session.add_message("assistant", "PR #4502 已经开始 review。") manager.save(session) assert list_webui_sessions(manager)[0]["preview"] == "PR #4502 已经开始 review。" def test_webui_session_list_skips_hidden_history_user_preview(tmp_path: Path) -> None: manager = SessionManager(tmp_path) session = manager.get_or_create("websocket:hidden-preview") session.add_message( "user", "internal subagent result", **{HIDDEN_HISTORY_META: {"kind": "subagent_result", "subagent_task_id": "sub-1"}}, ) session.add_message("assistant", "subagent summary") manager.save(session) assert list_webui_sessions(manager)[0]["preview"] == "subagent summary" def test_webui_session_list_uses_webui_transcript_activity_for_sort( tmp_path: Path, monkeypatch, ) -> None: webui_dir = tmp_path / "webui" webui_dir.mkdir() monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) manager = SessionManager(tmp_path) old_session = manager.get_or_create("websocket:old-metadata") old_session.created_at = datetime(2026, 6, 15, 10, 0, 0) old_session.updated_at = datetime(2026, 6, 15, 10, 0, 0) old_session.add_message("user", "old metadata") old_session.messages[-1]["timestamp"] = "2026-06-15T10:00:00" old_session.updated_at = datetime(2026, 6, 15, 10, 0, 0) manager.save(old_session) newer_metadata = manager.get_or_create("websocket:newer-metadata") newer_metadata.created_at = datetime(2026, 6, 15, 11, 0, 0) newer_metadata.updated_at = datetime(2026, 6, 15, 11, 0, 0) newer_metadata.add_message("user", "newer metadata") newer_metadata.messages[-1]["timestamp"] = "2026-06-15T11:00:00" newer_metadata.updated_at = datetime(2026, 6, 15, 11, 0, 0) manager.save(newer_metadata) transcript = webui_dir / "websocket_old-metadata.jsonl" transcript.write_text( '{"event":"turn_end","chat_id":"old-metadata"}\n', encoding="utf-8", ) activity_ns = int(datetime(2026, 6, 15, 12, 0, 0).timestamp() * 1_000_000_000) os.utime(transcript, ns=(activity_ns, activity_ns)) rows = list_webui_sessions(manager) assert [row["key"] for row in rows] == [ "websocket:old-metadata", "websocket:newer-metadata", ] assert rows[0]["updated_at"].startswith("2026-06-15T12:00:00") def test_webui_session_list_rescans_when_transcript_changes( tmp_path: Path, monkeypatch, ) -> None: webui_dir = tmp_path / "webui" webui_dir.mkdir() monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) manager = SessionManager(tmp_path) session = manager.get_or_create("websocket:transcript-change") session.created_at = datetime(2026, 6, 15, 10, 0, 0) session.updated_at = datetime(2026, 6, 15, 10, 0, 0) session.add_message("user", "preview") session.messages[-1]["timestamp"] = "2026-06-15T10:00:00" session.updated_at = datetime(2026, 6, 15, 10, 0, 0) manager.save(session) assert list_webui_sessions(manager)[0]["preview"] == "preview" transcript = webui_dir / "websocket_transcript-change.jsonl" transcript.write_text( '{"event":"turn_end","chat_id":"transcript-change"}\n', encoding="utf-8", ) activity_ns = int(datetime(2026, 6, 15, 12, 30, 0).timestamp() * 1_000_000_000) os.utime(transcript, ns=(activity_ns, activity_ns)) original_scan = session_list_index._scan_session_row scanned: list[str] = [] def record_scan(session_manager: SessionManager, path: Path) -> dict | None: scanned.append(path.name) return original_scan(session_manager, path) monkeypatch.setattr(session_list_index, "_scan_session_row", record_scan) rows = list_webui_sessions(manager) assert scanned == [manager._get_session_path("websocket:transcript-change").name] assert rows[0]["updated_at"].startswith("2026-06-15T12:30:00") def test_webui_session_list_sorts_by_message_activity_not_maintenance_timestamp( tmp_path: Path, ) -> None: manager = SessionManager(tmp_path) old = manager.get_or_create("websocket:old") old.created_at = datetime(2026, 6, 1, 10, 0, 0) old.add_message("user", "old first visible activity") old.messages[-1]["timestamp"] = "2026-06-01T10:00:00" old.add_message("assistant", "automation result") old.messages[-1]["timestamp"] = "2026-06-05T10:00:00" old.updated_at = datetime(2026, 6, 30, 17, 40, 0) manager.save(old) newer = manager.get_or_create("websocket:newer") newer.created_at = datetime(2026, 6, 4, 10, 0, 0) newer.add_message("user", "newer real activity") newer.messages[-1]["timestamp"] = "2026-06-04T10:00:00" newer.updated_at = datetime(2026, 6, 4, 10, 0, 0) manager.save(newer) rows = list_webui_sessions(manager) assert [row["key"] for row in rows] == ["websocket:old", "websocket:newer"] assert rows[0]["updated_at"] == "2026-06-05T10:00:00" def list_webui_sessions(manager: SessionManager) -> list[dict]: return session_list_index.list_webui_sessions(manager) def test_webui_session_list_fallback_time_when_missing(tmp_path: Path) -> None: manager = SessionManager(tmp_path) path = manager._get_session_path("websocket:missing-time") path.write_text( '{"_type": "metadata", "key": "websocket:missing-time"}\n' '{"_type": "message", "role": "user", "content": "hello"}\n', encoding="utf-8", ) rows = list_webui_sessions(manager) assert len(rows) == 1 assert rows[0]["key"] == "websocket:missing-time" assert rows[0]["created_at"] is not None assert rows[0]["updated_at"] is not None datetime.fromisoformat(rows[0]["created_at"]) datetime.fromisoformat(rows[0]["updated_at"]) def test_session_manager_list_sessions_fallback_time_when_missing(tmp_path: Path) -> None: manager = SessionManager(tmp_path) path = manager._get_session_path("websocket:missing-time2") path.write_text( '{"_type": "metadata", "key": "websocket:missing-time2"}\n' '{"_type": "message", "role": "user", "content": "hello"}\n', encoding="utf-8", ) sessions = manager.list_sessions() assert len(sessions) == 1 assert sessions[0]["key"] == "websocket:missing-time2" assert sessions[0]["created_at"] is not None assert sessions[0]["updated_at"] is not None datetime.fromisoformat(sessions[0]["created_at"]) datetime.fromisoformat(sessions[0]["updated_at"])