diff --git a/nanobot/channels/websocket/tests/test_websocket_http_routes.py b/nanobot/channels/websocket/tests/test_websocket_http_routes.py index 59a1dd31d..fd93b9074 100644 --- a/nanobot/channels/websocket/tests/test_websocket_http_routes.py +++ b/nanobot/channels/websocket/tests/test_websocket_http_routes.py @@ -31,6 +31,11 @@ from .ws_test_client import http_get as _http_get _PORT = 29900 +@pytest.fixture(autouse=True) +def _isolate_runtime_data(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) + + class _MatrixChannel(BaseChannel): name = "matrix" display_name = "Matrix" @@ -283,6 +288,53 @@ async def test_sessions_list_requires_bearer_token( await server_task +@pytest.mark.asyncio +async def test_sessions_list_and_thread_restore_transcript_without_canonical_file( + bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) + sm = SessionManager(tmp_path / "workspace") + from nanobot.webui.transcript import append_transcript_object + + key = "websocket:restored-history" + append_transcript_object( + key, + {"event": "user", "chat_id": "restored-history", "text": "original question"}, + ) + append_transcript_object( + key, + {"event": "message", "chat_id": "restored-history", "text": "original answer"}, + ) + assert not sm._get_session_path(key).exists() + + port = _free_port() + channel = _ch(bus, session_manager=sm, port=port) + server_task = asyncio.create_task(channel.start()) + try: + token = channel.gateway.tokens.issue_api_token(300) + auth = {"Authorization": f"Bearer {token}"} + + listing = await _http_get(f"http://127.0.0.1:{port}/api/sessions", headers=auth) + thread = await _http_get( + f"http://127.0.0.1:{port}/api/sessions/" + "websocket%3Arestored-history/webui-thread", + headers=auth, + ) + + assert listing.status_code == 200 + assert [row["key"] for row in listing.json()["sessions"]] == [key] + assert listing.json()["sessions"][0]["preview"] == "original question" + assert thread.status_code == 200 + assert [message["content"] for message in thread.json()["messages"]] == [ + "original question", + "original answer", + ] + assert not sm._get_session_path(key).exists() + finally: + await channel.stop() + await server_task + + @pytest.mark.asyncio async def test_legacy_session_messages_route_is_not_exposed( bus: MagicMock, tmp_path: Path @@ -2267,6 +2319,40 @@ async def test_session_delete_removes_file( await server_task +@pytest.mark.asyncio +async def test_session_delete_removes_transcript_without_canonical_file( + bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) + sm = SessionManager(tmp_path / "workspace") + from nanobot.webui.transcript import append_transcript_object + + key = "websocket:transcript-only" + append_transcript_object( + key, + {"event": "user", "chat_id": "transcript-only", "text": "recover me"}, + ) + assert not sm._get_session_path(key).exists() + webui_path = tmp_path / "webui" / f"{SessionManager.safe_key(key)}.jsonl" + assert webui_path.is_file() + + channel = _ch(bus, session_manager=sm, port=_free_port()) + server_task = asyncio.create_task(channel.start()) + try: + response = await _webui_mutate( + channel, + "session.delete", + {"key": key}, + ) + + assert response.status_code == 200 + assert response.json()["deleted"] is True + assert not webui_path.exists() + finally: + await channel.stop() + await server_task + + @pytest.mark.asyncio async def test_webui_automations_route_lists_all_jobs_and_allows_user_actions( bus: MagicMock, tmp_path: Path diff --git a/nanobot/webui/session_list_index.py b/nanobot/webui/session_list_index.py index a078295b3..df0ac8357 100644 --- a/nanobot/webui/session_list_index.py +++ b/nanobot/webui/session_list_index.py @@ -1,14 +1,15 @@ """Cache-only WebUI session list index. -The core ``SessionManager`` owns durable conversation history. This module owns -the WebUI sidebar optimization so core session writes stay independent from UI -presentation caches. +The core ``SessionManager`` owns model context while the WebUI transcript owns +durable display history. The sidebar discovers both without reconstructing one +store from the other, so core session writes stay independent from UI state. """ from __future__ import annotations import json import os +import re from datetime import datetime from pathlib import Path from typing import Any, cast @@ -30,9 +31,12 @@ from nanobot.session.manager import ( ) from nanobot.session.model_selection import model_preset_from_metadata -_INDEX_VERSION = 6 +_INDEX_VERSION = 7 _INDEX_FILENAME = ".webui_session_index.json" _MODEL_PRESET_FIELD = "model_preset" +_ROW_SOURCE_FIELD = "_source" +_SESSION_SOURCE = "session" +_TRANSCRIPT_SOURCE = "webui_transcript" _WORKSPACE_SCOPE_PRESENT_FIELD = "_workspace_scope_present" _WORKSPACE_SCOPE_VALUE_FIELD = "_workspace_scope_value" WEBUI_SESSION_INDEX_INTERNAL_FIELDS = frozenset( @@ -42,7 +46,12 @@ _INDEXED_WORKSPACE_SCOPE_KEYS = ("project_path", "path", "access_mode") _MAX_INDEXED_WORKSPACE_SCOPE_BYTES = 4096 _WEBUI_ACTIVITY_MTIME_NS = "webui_activity_mtime_ns" _WEBUI_ACTIVITY_SIZE = "webui_activity_size" +_WEBUI_ACTIVITY_FILES = "webui_activity_files" _VISIBLE_TRANSCRIPT_ROLES = {"user", "assistant"} +_WEBUI_SESSION_STEM_PREFIX = SessionManager.safe_key("websocket:") +_WEBUI_CHAT_ID_RE = re.compile(r"^[A-Za-z0-9_:-]{1,64}$") +_TRANSCRIPT_SEGMENTS_SUFFIX = ".segments" +_TRANSCRIPT_NON_ANSWER_KINDS = {"progress", "reasoning", "tool_hint"} def list_webui_sessions(session_manager: SessionManager) -> list[dict[str, Any]]: @@ -53,41 +62,81 @@ def list_webui_sessions(session_manager: SessionManager) -> list[dict[str, Any]] _write_index_rows(session_manager.sessions_dir, rows) except Exception as e: logger.debug("Failed to write WebUI session list index: {}", e) - sessions = [_public_row(session_manager.sessions_dir, row) for row in rows] + sessions = [ + _public_row(session_manager.sessions_dir, get_webui_dir(), row) + for row in rows + ] return sorted(sessions, key=lambda row: row.get("updated_at", ""), reverse=True) def _reconcile_index(session_manager: SessionManager) -> tuple[list[dict[str, Any]], bool]: existing_rows = _read_index_rows(session_manager.sessions_dir) - existing_by_file = { - row.get("file"): row + existing_by_source = { + (row.get(_ROW_SOURCE_FIELD), row.get("file")): row for row in existing_rows or [] - if isinstance(row.get("file"), str) + if isinstance(row.get(_ROW_SOURCE_FIELD), str) + and isinstance(row.get("file"), str) } - paths = sorted( - path - for path in session_manager.sessions_dir.glob("*.jsonl") - if SessionManager._session_key_from_path(path) is not None # pyright: ignore[reportPrivateUsage] - ) - if not paths: - return [], existing_rows != [] - webui_dir = get_webui_dir() + session_paths: dict[str, Path] = {} + for path in sorted(session_manager.sessions_dir.glob("*.jsonl")): + key = SessionManager._session_key_from_path(path) # pyright: ignore[reportPrivateUsage] + if key is not None: + session_paths[key] = path + + transcript_sources: dict[str, tuple[str, tuple[Path, ...]]] = {} + session_keys_by_stem = { + SessionManager.safe_key(key): key + for key in session_paths + if key.startswith("websocket:") + } + for stem, paths in _webui_transcript_sources(webui_dir).items(): + key = session_keys_by_stem.get(stem) + if key is None: + cached = existing_by_source.get((_TRANSCRIPT_SOURCE, stem)) + cached_key = cached.get("key") if cached is not None else None + if isinstance(cached_key, str) and _valid_transcript_session_key(cached_key, stem): + key = cached_key + else: + key = _session_key_from_transcript(stem, paths) + if key is not None: + transcript_sources[key] = (stem, paths) + rows: list[dict[str, Any]] = [] changed = existing_rows is None + expected_sources: set[tuple[str, str]] = set() - for path in paths: - row = existing_by_file.get(path.name) + for key, path in sorted(session_paths.items()): + identity = (_SESSION_SOURCE, path.name) + row = existing_by_source.get(identity) if row is not None and _indexed_row_matches_file(row, path, webui_dir): rows.append(row) + expected_sources.add(identity) continue changed = True scanned = _scan_session_row(session_manager, path, webui_dir) if scanned is not None: rows.append(scanned) + expected_sources.add(identity) - if set(existing_by_file) != {path.name for path in paths}: + for key, (stem, paths) in sorted(transcript_sources.items()): + if key in session_paths: + continue + identity = (_TRANSCRIPT_SOURCE, stem) + row = existing_by_source.get(identity) + if row is not None and _indexed_transcript_row_matches(row, key, webui_dir): + rows.append(row) + expected_sources.add(identity) + continue + + changed = True + scanned = _scan_transcript_row(key, stem, paths, webui_dir) + if scanned is not None: + rows.append(scanned) + expected_sources.add(identity) + + if set(existing_by_source) != expected_sources: changed = True if existing_rows is not None and rows != existing_rows: changed = True @@ -144,7 +193,7 @@ def _indexed_row_matches_file(row: dict[str, Any], path: Path, webui_dir: Path) return False if not isinstance(row.get(_WORKSPACE_SCOPE_PRESENT_FIELD), bool): return False - if row.get("file") != path.name: + if row.get(_ROW_SOURCE_FIELD) != _SESSION_SOURCE or row.get("file") != path.name: return False try: signature = _file_signature(path) @@ -156,10 +205,39 @@ def _indexed_row_matches_file(row: dict[str, Any], path: Path, webui_dir: Path) and row.get("size") == signature["size"] and row.get(_WEBUI_ACTIVITY_MTIME_NS) == activity_signature[_WEBUI_ACTIVITY_MTIME_NS] and row.get(_WEBUI_ACTIVITY_SIZE) == activity_signature[_WEBUI_ACTIVITY_SIZE] + and row.get(_WEBUI_ACTIVITY_FILES) == activity_signature[_WEBUI_ACTIVITY_FILES] ) -def _public_row(sessions_dir: Path, row: dict[str, Any]) -> dict[str, Any]: +def _indexed_transcript_row_matches( + row: dict[str, Any], + session_key: str, + webui_dir: Path, +) -> bool: + if not all(isinstance(row.get(key), str) for key in ("key", "created_at", "updated_at")): + return False + if row.get(_ROW_SOURCE_FIELD) != _TRANSCRIPT_SOURCE: + return False + if row.get("key") != session_key or row.get("file") != SessionManager.safe_key(session_key): + return False + if not isinstance(row.get("title", ""), str) or not isinstance(row.get("preview", ""), str): + return False + if not isinstance(row.get(_WORKSPACE_SCOPE_PRESENT_FIELD), bool): + return False + signature = _webui_activity_signature(session_key, webui_dir) + return ( + row.get(_WEBUI_ACTIVITY_MTIME_NS) == signature[_WEBUI_ACTIVITY_MTIME_NS] + and row.get(_WEBUI_ACTIVITY_SIZE) == signature[_WEBUI_ACTIVITY_SIZE] + and row.get(_WEBUI_ACTIVITY_FILES) == signature[_WEBUI_ACTIVITY_FILES] + ) + + +def _public_row(sessions_dir: Path, webui_dir: Path, row: dict[str, Any]) -> dict[str, Any]: + file = str(row.get("file", "")) + if row.get(_ROW_SOURCE_FIELD) == _TRANSCRIPT_SOURCE: + path = webui_dir / f"{file}.jsonl" + else: + path = sessions_dir / file return { "key": row.get("key"), "created_at": row.get("created_at"), @@ -169,7 +247,7 @@ def _public_row(sessions_dir: Path, row: dict[str, Any]) -> dict[str, Any]: _MODEL_PRESET_FIELD: row.get(_MODEL_PRESET_FIELD), _WORKSPACE_SCOPE_PRESENT_FIELD: row.get(_WORKSPACE_SCOPE_PRESENT_FIELD, False), _WORKSPACE_SCOPE_VALUE_FIELD: row.get(_WORKSPACE_SCOPE_VALUE_FIELD), - "path": str(sessions_dir / str(row.get("file", ""))), + "path": str(path), } @@ -242,17 +320,127 @@ def _preview_from_messages(messages: list[dict[str, Any]]) -> str: return fallback_preview +def _webui_transcript_record_paths(stem: str, webui_dir: Path) -> tuple[Path, ...]: + paths: list[Path] = [] + segments_dir = webui_dir / f"{stem}{_TRANSCRIPT_SEGMENTS_SUFFIX}" + if segments_dir.is_dir() and not segments_dir.is_symlink(): + try: + paths.extend( + sorted( + path + for path in segments_dir.glob("*.jsonl") + if path.is_file() and not path.is_symlink() + ) + ) + except OSError: + pass + active = webui_dir / f"{stem}.jsonl" + if active.is_file() and not active.is_symlink(): + paths.append(active) + return tuple(paths) + + +def _webui_transcript_sources(webui_dir: Path) -> dict[str, tuple[Path, ...]]: + stems: set[str] = set() + try: + entries = tuple(webui_dir.iterdir()) + except OSError: + return {} + for path in entries: + if path.is_symlink(): + continue + if path.is_file() and path.suffix == ".jsonl": + stem = path.stem + elif path.is_dir() and path.name.endswith(_TRANSCRIPT_SEGMENTS_SUFFIX): + stem = path.name.removesuffix(_TRANSCRIPT_SEGMENTS_SUFFIX) + else: + continue + if stem.startswith(_WEBUI_SESSION_STEM_PREFIX): + stems.add(stem) + return { + stem: paths + for stem in sorted(stems) + if (paths := _webui_transcript_record_paths(stem, webui_dir)) + } + + +def _transcript_record(line: str) -> dict[str, Any] | None: + try: + value: object = json.loads(line) + except json.JSONDecodeError: + return None + return cast(dict[str, Any], value) if isinstance(value, dict) else None + + +def _session_key_from_transcript(stem: str, paths: tuple[Path, ...]) -> str | None: + scanned_records = 0 + scanned_chars = 0 + for path in paths: + try: + with open(path, encoding="utf-8") as handle: + for line in handle: + if not line.strip(): + continue + scanned_records += 1 + scanned_chars += len(line) + record = _transcript_record(line) + chat_id = record.get("chat_id") if record is not None else None + if isinstance(chat_id, str) and chat_id.strip(): + key = f"websocket:{chat_id.strip()}" + if _valid_transcript_session_key(key, stem): + return key + if ( + scanned_records >= _SESSION_LIST_PREVIEW_MAX_RECORDS + or scanned_chars >= _SESSION_LIST_PREVIEW_MAX_CHARS + ): + break + except OSError: + continue + if ( + scanned_records >= _SESSION_LIST_PREVIEW_MAX_RECORDS + or scanned_chars >= _SESSION_LIST_PREVIEW_MAX_CHARS + ): + break + + chat_id = stem.removeprefix(_WEBUI_SESSION_STEM_PREFIX) + fallback = f"websocket:{chat_id}" + if _valid_transcript_session_key(fallback, stem): + return fallback + return None + + +def _valid_transcript_session_key(key: str, stem: str) -> bool: + if not key.startswith("websocket:"): + return False + chat_id = key.split(":", 1)[1] + return _WEBUI_CHAT_ID_RE.fullmatch(chat_id) is not None and SessionManager.safe_key(key) == stem + + def _webui_activity_paths(session_key: str, webui_dir: Path) -> list[Path]: stem = SessionManager.safe_key(session_key) - return [ + paths = [ webui_dir / f"{stem}.jsonl", webui_dir / f"{stem}.json", ] + segments_dir = webui_dir / f"{stem}{_TRANSCRIPT_SEGMENTS_SUFFIX}" + if segments_dir.is_dir() and not segments_dir.is_symlink(): + try: + paths.extend( + sorted( + path + for path in segments_dir.iterdir() + if path.is_file() and not path.is_symlink() + ) + ) + except OSError: + pass + return paths def _webui_activity_signature(session_key: str, webui_dir: Path) -> dict[str, int]: latest_mtime_ns = 0 total_size = 0 + file_count = 0 for path in _webui_activity_paths(session_key, webui_dir): try: stat = path.stat() @@ -260,11 +448,13 @@ def _webui_activity_signature(session_key: str, webui_dir: Path) -> dict[str, in continue if not path.is_file(): continue + file_count += 1 latest_mtime_ns = max(latest_mtime_ns, stat.st_mtime_ns) total_size += stat.st_size return { _WEBUI_ACTIVITY_MTIME_NS: latest_mtime_ns, _WEBUI_ACTIVITY_SIZE: total_size, + _WEBUI_ACTIVITY_FILES: file_count, } @@ -333,6 +523,7 @@ def _indexed_row_for_session(session: Session, path: Path, webui_dir: Path) -> d "preview": _preview_from_messages(session.messages), _MODEL_PRESET_FIELD: model_preset_from_metadata(session.metadata), **_indexed_workspace_scope_fields(session.metadata), + _ROW_SOURCE_FIELD: _SESSION_SOURCE, "file": path.name, "mtime_ns": signature["mtime_ns"], "size": signature["size"], @@ -340,6 +531,108 @@ def _indexed_row_for_session(session: Session, path: Path, webui_dir: Path) -> d } +def _transcript_preview(record: dict[str, Any]) -> tuple[str, str]: + text = record.get("text") + if not isinstance(text, str) or not text.strip(): + return "", "" + event = record.get("event") + if event == "user" or record.get("role") == "user": + return text, "" + if ( + event == "message" + and record.get("kind") not in _TRANSCRIPT_NON_ANSWER_KINDS + ) or record.get("role") == "assistant": + return "", text + return "", "" + + +def _transcript_created_at(record: dict[str, Any]) -> str | None: + value = record.get("created_at_ms") + if ( + not isinstance(value, int | float) + or isinstance(value, bool) + or value < 0 + ): + return None + try: + return datetime.fromtimestamp(value / 1000).isoformat() + except (OSError, OverflowError, ValueError): + return None + + +def _scan_transcript_row( + session_key: str, + stem: str, + paths: tuple[Path, ...], + webui_dir: Path, +) -> dict[str, Any] | None: + signature = _webui_activity_signature(session_key, webui_dir) + activity_updated_at = _webui_activity_updated_at(signature) + if activity_updated_at is None: + return None + + preview = "" + fallback_preview = "" + created_at: str | None = None + saw_record = False + scanned_records = 0 + scanned_chars = 0 + for path in paths: + try: + with open(path, encoding="utf-8") as handle: + for line in handle: + if not line.strip(): + continue + scanned_records += 1 + scanned_chars += len(line) + record = _transcript_record(line) + if record is not None: + saw_record = True + if created_at is None: + created_at = _transcript_created_at(record) + user_preview, assistant_preview = _transcript_preview(record) + if user_preview: + preview = user_preview + break + if not fallback_preview and assistant_preview: + fallback_preview = assistant_preview + if ( + scanned_records >= _SESSION_LIST_PREVIEW_MAX_RECORDS + or scanned_chars >= _SESSION_LIST_PREVIEW_MAX_CHARS + ): + break + except OSError: + continue + if preview or ( + scanned_records >= _SESSION_LIST_PREVIEW_MAX_RECORDS + or scanned_chars >= _SESSION_LIST_PREVIEW_MAX_CHARS + ): + break + if not saw_record: + return None + + if created_at is None: + try: + earliest_mtime = min(path.stat().st_mtime for path in paths) + created_at = datetime.fromtimestamp(earliest_mtime).isoformat() + except (OSError, OverflowError, ValueError): + created_at = activity_updated_at + return { + "key": session_key, + "created_at": created_at, + "updated_at": activity_updated_at, + "title": "", + "preview": preview or fallback_preview, + _MODEL_PRESET_FIELD: None, + **_indexed_workspace_scope_fields({}), + _ROW_SOURCE_FIELD: _TRANSCRIPT_SOURCE, + "file": stem, + "mtime_ns": signature[_WEBUI_ACTIVITY_MTIME_NS], + "size": signature[_WEBUI_ACTIVITY_SIZE], + **signature, + } + + def _scan_session_row( session_manager: SessionManager, path: Path, @@ -418,6 +711,7 @@ def _scan_session_row( "preview": preview or fallback_preview, _MODEL_PRESET_FIELD: model_preset_from_metadata(metadata), **_indexed_workspace_scope_fields(metadata), + _ROW_SOURCE_FIELD: _SESSION_SOURCE, "file": path.name, "mtime_ns": signature["mtime_ns"], "size": signature["size"], diff --git a/nanobot/webui/ws_http.py b/nanobot/webui/ws_http.py index 3b95a0e23..c74f6f51d 100644 --- a/nanobot/webui/ws_http.py +++ b/nanobot/webui/ws_http.py @@ -855,9 +855,9 @@ class GatewayHTTPHandler: self.local_trigger_store.delete(job.id) elif self.cron_service is not None: self.cron_service.remove_job(job.id) - deleted = self.session_manager.delete_session(decoded_key) - delete_webui_thread(decoded_key) - return _http_json_response({"deleted": bool(deleted)}) + session_deleted = self.session_manager.delete_session(decoded_key) + transcript_deleted = delete_webui_thread(decoded_key) + return _http_json_response({"deleted": bool(session_deleted or transcript_deleted)}) # -- Automation routes -------------------------------------------------- diff --git a/tests/webui/test_session_list_index.py b/tests/webui/test_session_list_index.py index 939819cc5..6e1d71215 100644 --- a/tests/webui/test_session_list_index.py +++ b/tests/webui/test_session_list_index.py @@ -17,6 +17,13 @@ from nanobot.session.manager import SessionManager from nanobot.session.model_selection import SESSION_MODEL_PRESET_METADATA_KEY +@pytest.fixture(autouse=True) +def _isolate_webui_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + webui_dir = tmp_path / "webui" + webui_dir.mkdir(exist_ok=True) + monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) + + def test_webui_session_list_reuses_valid_index_without_scanning_files( tmp_path: Path, monkeypatch, @@ -208,6 +215,247 @@ def test_webui_session_list_drops_deleted_index_rows(tmp_path: Path) -> None: assert list_webui_sessions(manager) == [] +def test_webui_session_list_recovers_transcript_without_canonical_session( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + webui_dir = tmp_path / "webui" + webui_dir.mkdir(exist_ok=True) + monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) + key = "websocket:restored" + transcript = webui_dir / f"{SessionManager.safe_key(key)}.jsonl" + transcript.write_text( + '{"event":"user","chat_id":"restored","text":"original question",' + '"created_at_ms":1785502800000}\n' + '{"event":"message","chat_id":"restored","text":"original answer",' + '"created_at_ms":1785502801000}\n' + '{"event":"turn_end","chat_id":"restored","created_at_ms":1785502802000}\n', + encoding="utf-8", + ) + manager = SessionManager(tmp_path / "workspace") + + [row] = list_webui_sessions(manager) + + assert row["key"] == key + assert row["preview"] == "original question" + assert row["created_at"] == datetime.fromtimestamp(1785502800).isoformat() + assert not manager._get_session_path(key).exists() + assert manager.list_sessions() == [] + + reloaded = SessionManager(tmp_path / "workspace") + assert [row["key"] for row in list_webui_sessions(reloaded)] == [key] + + +def test_webui_session_list_recovers_colon_chat_id_from_transcript( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + webui_dir = tmp_path / "webui" + webui_dir.mkdir(exist_ok=True) + monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) + key = "websocket:scope:child" + transcript = webui_dir / f"{SessionManager.safe_key(key)}.jsonl" + transcript.write_text( + '{"event":"user","chat_id":"scope:child","text":"scoped history"}\n', + encoding="utf-8", + ) + + [row] = list_webui_sessions(SessionManager(tmp_path / "workspace")) + + assert row["key"] == key + assert row["preview"] == "scoped history" + + +def test_webui_session_list_tolerates_invalid_transcript_timestamp( + tmp_path: Path, +) -> None: + key = "websocket:bad-time" + transcript = tmp_path / "webui" / f"{SessionManager.safe_key(key)}.jsonl" + transcript.write_text( + '{"event":"user","chat_id":"bad-time","text":"still visible",' + '"created_at_ms":1e100}\n', + encoding="utf-8", + ) + + [row] = list_webui_sessions(SessionManager(tmp_path / "workspace")) + + assert row["preview"] == "still visible" + datetime.fromisoformat(row["created_at"]) + + +def test_webui_session_list_ignores_invalid_transcript_chat_id(tmp_path: Path) -> None: + transcript = tmp_path / "webui" / "websocket_.._outside.jsonl" + transcript.write_text( + '{"event":"user","chat_id":"../outside","text":"do not expose"}\n', + encoding="utf-8", + ) + + assert list_webui_sessions(SessionManager(tmp_path / "workspace")) == [] + + +def test_webui_session_list_recovers_segment_only_transcript( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + webui_dir = tmp_path / "webui" + webui_dir.mkdir(exist_ok=True) + monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) + key = "websocket:segmented" + segments = webui_dir / f"{SessionManager.safe_key(key)}.segments" + segments.mkdir() + (segments / "000001.jsonl").write_text( + '{"event":"user","chat_id":"segmented","text":"older segment"}\n' + '{"event":"turn_end","chat_id":"segmented"}\n', + encoding="utf-8", + ) + + [row] = list_webui_sessions(SessionManager(tmp_path / "workspace")) + + assert row["key"] == key + assert row["preview"] == "older segment" + + +def test_webui_session_list_prefers_canonical_metadata_without_duplicate( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + webui_dir = tmp_path / "webui" + webui_dir.mkdir(exist_ok=True) + monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) + key = "websocket:canonical" + (webui_dir / f"{SessionManager.safe_key(key)}.jsonl").write_text( + '{"event":"user","chat_id":"canonical","text":"display copy"}\n', + encoding="utf-8", + ) + manager = SessionManager(tmp_path / "workspace") + session = manager.get_or_create(key) + session.metadata["title"] = "Canonical title" + session.add_message("user", "canonical preview") + manager.save(session) + + rows = list_webui_sessions(manager) + + assert len(rows) == 1 + assert rows[0]["key"] == key + assert rows[0]["preview"] == "canonical preview" + + +def test_webui_session_list_reuses_unchanged_transcript_index( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + webui_dir = tmp_path / "webui" + webui_dir.mkdir(exist_ok=True) + monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) + key = "websocket:cached-transcript" + (webui_dir / f"{SessionManager.safe_key(key)}.jsonl").write_text( + '{"event":"user","chat_id":"cached-transcript","text":"cached"}\n', + encoding="utf-8", + ) + manager = SessionManager(tmp_path / "workspace") + assert list_webui_sessions(manager)[0]["preview"] == "cached" + + def fail_scan(*args, **kwargs): + raise AssertionError("unchanged transcript should reuse its index row") + + monkeypatch.setattr(session_list_index, "_scan_transcript_row", fail_scan) + + assert list_webui_sessions(manager)[0]["preview"] == "cached" + + +def test_webui_session_list_does_not_cache_changed_transcript_with_old_signature( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + webui_dir = tmp_path / "webui" + key = "websocket:transcript-race" + transcript = webui_dir / f"{SessionManager.safe_key(key)}.jsonl" + transcript.write_text( + '{"event":"user","chat_id":"transcript-race","text":"initial"}\n', + encoding="utf-8", + ) + manager = SessionManager(tmp_path / "workspace") + assert list_webui_sessions(manager)[0]["preview"] == "initial" + transcript.write_text( + '{"event":"user","chat_id":"transcript-race","text":"first scan"}\n', + encoding="utf-8", + ) + + original_open = open + changed = False + + class RacingReader(io.StringIO): + def __next__(self) -> str: + nonlocal changed + if not changed: + changed = True + transcript.write_text( + '{"event":"user","chat_id":"transcript-race","text":"second scan"}\n', + encoding="utf-8", + ) + return super().__next__() + + def racing_open(path, *args, **kwargs): + if Path(path) == transcript: + with original_open(path, *args, **kwargs) as source: + return RacingReader(source.read()) + return original_open(path, *args, **kwargs) + + monkeypatch.setattr(session_list_index, "open", racing_open, raising=False) + + first = list_webui_sessions(manager) + second = list_webui_sessions(manager) + + assert first[0]["preview"] == "first scan" + assert second[0]["preview"] == "second scan" + + +def test_webui_session_list_drops_deleted_transcript_index_row( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + webui_dir = tmp_path / "webui" + webui_dir.mkdir(exist_ok=True) + monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) + key = "websocket:deleted-transcript" + transcript = webui_dir / f"{SessionManager.safe_key(key)}.jsonl" + transcript.write_text( + '{"event":"user","chat_id":"deleted-transcript","text":"delete me"}\n', + encoding="utf-8", + ) + manager = SessionManager(tmp_path / "workspace") + assert list_webui_sessions(manager)[0]["key"] == key + + transcript.unlink() + + assert list_webui_sessions(manager) == [] + + +def test_webui_session_list_keeps_runtime_instances_isolated( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + first_dir = tmp_path / "instance-a" / "webui" + second_dir = tmp_path / "instance-b" / "webui" + first_dir.mkdir(parents=True) + second_dir.mkdir(parents=True) + (first_dir / "websocket_first.jsonl").write_text( + '{"event":"user","chat_id":"first","text":"first instance"}\n', + encoding="utf-8", + ) + (second_dir / "websocket_second.jsonl").write_text( + '{"event":"user","chat_id":"second","text":"second instance"}\n', + encoding="utf-8", + ) + manager = SessionManager(tmp_path / "workspace") + + monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: first_dir) + assert [row["key"] for row in list_webui_sessions(manager)] == ["websocket:first"] + + monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: second_dir) + assert [row["key"] for row in list_webui_sessions(manager)] == ["websocket:second"] + + def test_webui_session_list_ignores_legacy_stem(tmp_path: Path) -> None: manager = SessionManager(tmp_path) legacy_path = manager.sessions_dir / "websocket_legacy.jsonl" @@ -269,7 +517,7 @@ def test_webui_session_list_uses_webui_transcript_activity_for_sort( monkeypatch, ) -> None: webui_dir = tmp_path / "webui" - webui_dir.mkdir() + webui_dir.mkdir(exist_ok=True) monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) manager = SessionManager(tmp_path) @@ -311,7 +559,7 @@ def test_webui_session_list_rescans_when_transcript_changes( monkeypatch, ) -> None: webui_dir = tmp_path / "webui" - webui_dir.mkdir() + webui_dir.mkdir(exist_ok=True) monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir) manager = SessionManager(tmp_path) @@ -416,4 +664,3 @@ def test_session_manager_list_sessions_fallback_time_when_missing(tmp_path: Path assert sessions[0]["updated_at"] is not None datetime.fromisoformat(sessions[0]["created_at"]) datetime.fromisoformat(sessions[0]["updated_at"]) -