fix(webui): restore transcript-only session history

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

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 core ``SessionManager`` owns model context while the WebUI transcript owns
-the WebUI sidebar optimization so core session writes stay independent from UI
+durable display history. The sidebar discovers both without reconstructing one
-presentation caches.
+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 = {
+    existing_by_source = {
-        row.get("file"): row
+        (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:
+    for key, path in sorted(session_paths.items()):
-        row = existing_by_file.get(path.name)
+        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"],

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)
+        session_deleted = self.session_manager.delete_session(decoded_key)
-        delete_webui_thread(decoded_key)
+        transcript_deleted = delete_webui_thread(decoded_key)
-        return _http_json_response({"deleted": bool(deleted)})
+        return _http_json_response({"deleted": bool(session_deleted or transcript_deleted)})
 
     # -- Automation routes --------------------------------------------------
 

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"])
-