From 580824a15ae9832432fa0e9884a7a87644935f14 Mon Sep 17 00:00:00 2001 From: chengyongru <61816729+chengyongru@users.noreply.github.com> Date: Mon, 3 Aug 2026 09:51:35 +0800 Subject: [PATCH] perf(webui): accelerate JSONL session list and thread loading (#5194) --- .../tests/test_websocket_http_routes.py | 160 ++++++++++++++- .../websocket/tests/ws_test_client.py | 2 +- nanobot/webui/http_utils.py | 61 +++++- nanobot/webui/session_list_index.py | 105 ++++++++-- nanobot/webui/transcript.py | 192 ++++++++++++------ nanobot/webui/workspaces.py | 43 +++- nanobot/webui/ws_http.py | 60 ++++-- tests/utils/test_webui_thread_disk.py | 1 + tests/utils/test_webui_transcript.py | 141 ++++++++++++- tests/utils/test_webui_workspaces.py | 83 +++++++- tests/webui/test_http_utils.py | 38 ++++ tests/webui/test_session_list_index.py | 103 +++++++++- webui/src/hooks/useSessions.ts | 2 +- webui/src/tests/useSessions.test.tsx | 2 +- 14 files changed, 864 insertions(+), 129 deletions(-) create mode 100644 tests/webui/test_http_utils.py diff --git a/nanobot/channels/websocket/tests/test_websocket_http_routes.py b/nanobot/channels/websocket/tests/test_websocket_http_routes.py index 2371ca53b..bcba8a7a9 100644 --- a/nanobot/channels/websocket/tests/test_websocket_http_routes.py +++ b/nanobot/channels/websocket/tests/test_websocket_http_routes.py @@ -24,6 +24,7 @@ from nanobot.runtime_context import ( RuntimeContextBlock, append_runtime_context, ) +from nanobot.security.workspace_access import WORKSPACE_SCOPE_METADATA_KEY from nanobot.session.keys import UNIFIED_SESSION_KEY from nanobot.session.manager import Session, SessionManager from nanobot.triggers.local_store import LocalTriggerStore @@ -2201,7 +2202,7 @@ async def test_mcp_presets_routes_require_token_and_return_payload( @pytest.mark.asyncio async def test_sessions_list_only_returns_websocket_sessions_by_default( - bus: MagicMock, tmp_path: Path + bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: # Seed a realistic multi-channel disk state: CLI, Slack, Lark and # websocket sessions all live in the same ``sessions/`` directory. @@ -2215,7 +2216,20 @@ async def test_sessions_list_only_returns_websocket_sessions_by_default( "websocket:beta", ], ) - channel = _ch(bus, session_manager=sm, port=29906) + project = tmp_path / "project" + project.mkdir() + scoped = sm.get_or_create("websocket:beta") + scoped.metadata[WORKSPACE_SCOPE_METADATA_KEY] = { + "project_path": str(project), + "access_mode": "restricted", + } + sm.save(scoped) + + def fail_metadata_read(_key: str) -> None: + raise AssertionError("the session list must use its own index metadata") + + monkeypatch.setattr(sm, "read_session_metadata", fail_metadata_read) + channel = _ch(bus, session_manager=sm, workspace_path=tmp_path, port=29906) server_task = asyncio.create_task(channel.start()) try: token = channel.gateway.tokens.issue_api_token(300) @@ -2225,10 +2239,17 @@ async def test_sessions_list_only_returns_websocket_sessions_by_default( "http://127.0.0.1:29906/api/sessions", headers=auth ) assert listing.status_code == 200 - keys = {s["key"] for s in listing.json()["sessions"]} + sessions = listing.json()["sessions"] + keys = {s["key"] for s in sessions} # Only websocket-channel sessions are part of the webui surface; CLI / # Slack / Lark rows would be non-resumable from the browser. assert keys == {"websocket:alpha", "websocket:beta"} + rows = {row["key"]: row for row in sessions} + assert rows["websocket:beta"]["workspace_scope"]["project_path"] == str( + project.resolve() + ) + assert rows["websocket:beta"]["workspace_scope"]["access_mode"] == "restricted" + assert all(not any(key.startswith("_") for key in row) for row in sessions) finally: await channel.stop() await server_task @@ -2956,6 +2977,139 @@ async def test_webui_thread_resigns_assistant_media_urls( await server_task +@pytest.mark.asyncio +async def test_sessions_list_negotiates_gzip_across_repeated_headers( + bus: MagicMock, tmp_path: Path +) -> None: + sm = _seed_many(tmp_path, [f"websocket:gzip-{index:03d}" for index in range(80)]) + port = _free_port() + channel = _ch(bus, session_manager=sm, workspace_path=tmp_path, port=port) + server_task = asyncio.create_task(channel.start()) + try: + token = channel.gateway.tokens.issue_api_token(300) + response = await _http_get( + f"http://127.0.0.1:{port}/api/sessions", + headers=[ + ("Authorization", f"Bearer {token}"), + ("Accept-Encoding", "identity;q=0"), + ("Accept-Encoding", "gzip"), + ], + ) + + assert response.status_code == 200 + assert response.headers["Content-Encoding"] == "gzip" + assert response.headers["Vary"] == "Accept-Encoding" + assert len(response.json()["sessions"]) == 80 + finally: + await channel.stop() + await server_task + + +@pytest.mark.asyncio +async def test_webui_thread_complete_transcript_skips_session_history_read( + bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from nanobot.webui.transcript import append_transcript_object + + monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) + key = "websocket:fast-thread" + sm = _seed_session(tmp_path, key=key) + for event in ( + {"event": "user", "chat_id": "fast-thread", "text": "hi"}, + {"event": "message", "chat_id": "fast-thread", "text": "hello back"}, + {"event": "turn_end", "chat_id": "fast-thread"}, + ): + append_transcript_object(key, event) + + read_session_file = MagicMock( + side_effect=AssertionError("complete transcripts must not read canonical history") + ) + monkeypatch.setattr(sm, "read_session_file", read_session_file) + port = _free_port() + channel = _ch( + bus, + session_manager=sm, + workspace_path=tmp_path, + port=port, + ) + server_task = asyncio.create_task(channel.start()) + try: + token = channel.gateway.tokens.issue_api_token(300) + response = await _http_get( + f"http://127.0.0.1:{port}/api/sessions/" + "websocket%3Afast-thread/webui-thread?limit=160&direction=latest", + headers={"Authorization": f"Bearer {token}"}, + ) + + assert response.status_code == 200 + assert [message["content"] for message in response.json()["messages"]] == [ + "hi", + "hello back", + ] + read_session_file.assert_not_called() + finally: + await channel.stop() + await server_task + + +@pytest.mark.asyncio +async def test_webui_thread_negotiates_gzip_for_large_payloads( + bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from nanobot.webui.transcript import append_transcript_object + + monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) + sm = SessionManager(tmp_path) + append_transcript_object( + "websocket:gzip-thread", + { + "event": "user", + "chat_id": "gzip-thread", + "text": "compress me " * 1_000, + }, + ) + port = _free_port() + channel = _ch(bus, session_manager=sm, workspace_path=tmp_path, port=port) + server_task = asyncio.create_task(channel.start()) + try: + token = channel.gateway.tokens.issue_api_token(300) + url = ( + f"http://127.0.0.1:{port}/api/sessions/" + "websocket%3Agzip-thread/webui-thread?limit=80&direction=latest" + ) + compressed = await _http_get( + url, + headers={ + "Authorization": f"Bearer {token}", + "Accept-Encoding": "br, gzip", + }, + ) + + assert compressed.status_code == 200 + assert compressed.headers["Content-Encoding"] == "gzip" + assert compressed.headers["Vary"] == "Accept-Encoding" + assert int(compressed.headers["Content-Length"]) < len(compressed.content) + assert compressed.json()["messages"][0]["content"].startswith("compress me") + + identity = await _http_get( + url, + headers={ + "Authorization": f"Bearer {token}", + "Accept-Encoding": "gzip;q=0, br", + }, + ) + assert identity.status_code == 200 + assert "Content-Encoding" not in identity.headers + assert identity.json() == compressed.json() + + unauthorized = await _http_get(url, headers={"Accept-Encoding": "gzip"}) + assert unauthorized.status_code == 401 + assert "Content-Encoding" not in unauthorized.headers + finally: + await channel.stop() + await server_task + + @pytest.mark.asyncio async def test_session_routes_reject_non_websocket_keys( bus: MagicMock, tmp_path: Path diff --git a/nanobot/channels/websocket/tests/ws_test_client.py b/nanobot/channels/websocket/tests/ws_test_client.py index ecdbbf0ad..ae3fbc490 100644 --- a/nanobot/channels/websocket/tests/ws_test_client.py +++ b/nanobot/channels/websocket/tests/ws_test_client.py @@ -248,7 +248,7 @@ class WsTestClient: async def http_get( url: str, - headers: dict[str, str] | None = None, + headers: dict[str, str] | list[tuple[str, str]] | None = None, ) -> httpx.Response: """GET a local test server without loading an unused TLS trust store.""" request = httpx.Request("GET", url, headers=headers or {}) diff --git a/nanobot/webui/http_utils.py b/nanobot/webui/http_utils.py index e261b6f5b..375dfaeac 100644 --- a/nanobot/webui/http_utils.py +++ b/nanobot/webui/http_utils.py @@ -3,6 +3,7 @@ from __future__ import annotations import email.utils +import gzip import hmac import http import ipaddress @@ -16,6 +17,9 @@ from websockets.http11 import Response QueryParams = dict[str, list[str]] +_JSON_GZIP_MIN_BYTES = 4 * 1024 +_JSON_GZIP_LEVEL = 5 + def strip_trailing_slash(path: str) -> str: if len(path) > 1 and path.endswith("/"): @@ -41,6 +45,15 @@ def case_insensitive_header(headers: Any, key: str) -> str: return str(value or "").strip() +def combined_list_header(headers: Any, key: str) -> str: + """Combine repeated values for a comma-separated HTTP list header.""" + try: + values = headers.get_all(key) + except (AttributeError, KeyError): + return case_insensitive_header(headers, key) + return ", ".join(str(value).strip() for value in values if str(value).strip()) + + def safe_host_header(value: str) -> str: """Return a safe Host header value, or empty when it should not be echoed.""" value = value.strip() @@ -62,18 +75,46 @@ def host_for_url(host: str, port: int) -> str: return f"{host}:{port}" -def http_json_response(data: dict[str, Any], *, status: int = 200) -> Response: +def _accepts_gzip(value: str) -> bool: + wildcard_quality: float | None = None + for item in value.split(","): + name, *params = (part.strip() for part in item.split(";")) + quality = 1.0 + for param in params: + key, separator, raw_value = param.partition("=") + if separator and key.strip().lower() == "q": + try: + quality = float(raw_value.strip()) + except ValueError: + quality = 0.0 + break + if name.lower() == "gzip": + return quality > 0 + if name == "*": + wildcard_quality = quality + return wildcard_quality is not None and wildcard_quality > 0 + + +def http_json_response( + data: dict[str, Any], + *, + status: int = 200, + accept_encoding: str | None = None, +) -> Response: body = json.dumps(data, ensure_ascii=False).encode("utf-8") - headers = Headers( - [ - ("Date", email.utils.formatdate(usegmt=True)), - ("Connection", "close"), - ("Content-Length", str(len(body))), - ("Content-Type", "application/json; charset=utf-8"), - ] - ) + headers = [ + ("Date", email.utils.formatdate(usegmt=True)), + ("Connection", "close"), + ("Content-Type", "application/json; charset=utf-8"), + ] + if accept_encoding is not None: + headers.append(("Vary", "Accept-Encoding")) + if len(body) >= _JSON_GZIP_MIN_BYTES and _accepts_gzip(accept_encoding): + body = gzip.compress(body, compresslevel=_JSON_GZIP_LEVEL, mtime=0) + headers.append(("Content-Encoding", "gzip")) + headers.append(("Content-Length", str(len(body)))) reason = http.HTTPStatus(status).phrase - return Response(status, reason, headers, body) + return Response(status, reason, Headers(headers), body) def http_response( diff --git a/nanobot/webui/session_list_index.py b/nanobot/webui/session_list_index.py index 9354406b6..a078295b3 100644 --- a/nanobot/webui/session_list_index.py +++ b/nanobot/webui/session_list_index.py @@ -16,6 +16,7 @@ from typing import Any, cast from loguru import logger from nanobot.config.paths import get_webui_dir +from nanobot.security.workspace_access import WORKSPACE_SCOPE_METADATA_KEY from nanobot.session.history_visibility import is_hidden_history_message from nanobot.session.manager import ( _PROVIDER_STATE_RECORD_TYPE, # pyright: ignore[reportPrivateUsage] @@ -29,9 +30,16 @@ from nanobot.session.manager import ( ) from nanobot.session.model_selection import model_preset_from_metadata -_INDEX_VERSION = 4 +_INDEX_VERSION = 6 _INDEX_FILENAME = ".webui_session_index.json" _MODEL_PRESET_FIELD = "model_preset" +_WORKSPACE_SCOPE_PRESENT_FIELD = "_workspace_scope_present" +_WORKSPACE_SCOPE_VALUE_FIELD = "_workspace_scope_value" +WEBUI_SESSION_INDEX_INTERNAL_FIELDS = frozenset( + {_WORKSPACE_SCOPE_PRESENT_FIELD, _WORKSPACE_SCOPE_VALUE_FIELD} +) +_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" _VISIBLE_TRANSCRIPT_ROLES = {"user", "assistant"} @@ -61,17 +69,21 @@ def _reconcile_index(session_manager: SessionManager) -> tuple[list[dict[str, An 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() rows: list[dict[str, Any]] = [] changed = existing_rows is None for path in paths: row = existing_by_file.get(path.name) - if row is not None and _indexed_row_matches_file(row, path): + if row is not None and _indexed_row_matches_file(row, path, webui_dir): rows.append(row) continue changed = True - scanned = _scan_session_row(session_manager, path) + scanned = _scan_session_row(session_manager, path, webui_dir) if scanned is not None: rows.append(scanned) @@ -125,18 +137,20 @@ def _file_signature(path: Path) -> dict[str, int]: return {"mtime_ns": stat.st_mtime_ns, "size": stat.st_size} -def _indexed_row_matches_file(row: dict[str, Any], path: Path) -> bool: +def _indexed_row_matches_file(row: dict[str, Any], path: Path, webui_dir: Path) -> bool: if not all(isinstance(row.get(key), str) for key in ("key", "created_at", "updated_at")): 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 if row.get("file") != path.name: return False try: signature = _file_signature(path) except OSError: return False - activity_signature = _webui_activity_signature(str(row.get("key"))) + activity_signature = _webui_activity_signature(str(row.get("key")), webui_dir) return ( row.get("mtime_ns") == signature["mtime_ns"] and row.get("size") == signature["size"] @@ -153,10 +167,57 @@ def _public_row(sessions_dir: Path, row: dict[str, Any]) -> dict[str, Any]: "title": row.get("title", ""), "preview": row.get("preview", ""), _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", ""))), } +def indexed_workspace_scope(row: dict[str, Any]) -> tuple[bool, object]: + """Return the cached sidebar scope value while preserving missing vs null.""" + return ( + row.get(_WORKSPACE_SCOPE_PRESENT_FIELD) is True, + cast(object, row.get(_WORKSPACE_SCOPE_VALUE_FIELD)), + ) + + +def _indexed_workspace_scope_fields(metadata: object) -> dict[str, object]: + if not isinstance(metadata, dict): + return { + _WORKSPACE_SCOPE_PRESENT_FIELD: False, + _WORKSPACE_SCOPE_VALUE_FIELD: None, + } + metadata_data = cast(dict[str, Any], metadata) + if WORKSPACE_SCOPE_METADATA_KEY not in metadata_data: + return { + _WORKSPACE_SCOPE_PRESENT_FIELD: False, + _WORKSPACE_SCOPE_VALUE_FIELD: None, + } + + raw_scope = metadata_data.get(WORKSPACE_SCOPE_METADATA_KEY) + indexed_scope: object = False + if raw_scope is None: + indexed_scope = None + elif isinstance(raw_scope, dict): + scope_data = cast(dict[object, object], raw_scope) + recognized = { + key: scope_data[key] + for key in _INDEXED_WORKSPACE_SCOPE_KEYS + if key in scope_data + } + try: + encoded = json.dumps(recognized, ensure_ascii=False) + except (TypeError, ValueError): + pass + else: + if len(encoded.encode("utf-8")) <= _MAX_INDEXED_WORKSPACE_SCOPE_BYTES: + indexed_scope = cast(object, json.loads(encoded)) + return { + _WORKSPACE_SCOPE_PRESENT_FIELD: True, + _WORKSPACE_SCOPE_VALUE_FIELD: indexed_scope, + } + + def _preview_from_messages(messages: list[dict[str, Any]]) -> str: fallback_preview = "" scanned_records = 0 @@ -181,19 +242,18 @@ def _preview_from_messages(messages: list[dict[str, Any]]) -> str: return fallback_preview -def _webui_activity_paths(session_key: str) -> list[Path]: +def _webui_activity_paths(session_key: str, webui_dir: Path) -> list[Path]: stem = SessionManager.safe_key(session_key) - webui_dir = get_webui_dir() return [ webui_dir / f"{stem}.jsonl", webui_dir / f"{stem}.json", ] -def _webui_activity_signature(session_key: str) -> dict[str, int]: +def _webui_activity_signature(session_key: str, webui_dir: Path) -> dict[str, int]: latest_mtime_ns = 0 total_size = 0 - for path in _webui_activity_paths(session_key): + for path in _webui_activity_paths(session_key, webui_dir): try: stat = path.stat() except OSError: @@ -231,10 +291,10 @@ def _latest_updated_at(stored: str | None, activity: str | None) -> str | None: def _visible_message_timestamp(item: dict[str, Any]) -> str | None: - if is_hidden_history_message(item): - return None if item.get("role") not in _VISIBLE_TRANSCRIPT_ROLES: return None + if is_hidden_history_message(item): + return None timestamp = item.get("timestamp") return timestamp if isinstance(timestamp, str) else None @@ -256,9 +316,9 @@ def _visible_activity_updated_at( return _latest_updated_at(visible_message_at, webui_activity) or stored -def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]: +def _indexed_row_for_session(session: Session, path: Path, webui_dir: Path) -> dict[str, Any]: signature = _file_signature(path) - activity_signature = _webui_activity_signature(session.key) + activity_signature = _webui_activity_signature(session.key, webui_dir) activity_updated_at = _webui_activity_updated_at(activity_signature) visible_message_at = _last_visible_message_at(session.messages) return { @@ -272,6 +332,7 @@ def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]: "title": _metadata_title(session.metadata), "preview": _preview_from_messages(session.messages), _MODEL_PRESET_FIELD: model_preset_from_metadata(session.metadata), + **_indexed_workspace_scope_fields(session.metadata), "file": path.name, "mtime_ns": signature["mtime_ns"], "size": signature["size"], @@ -279,11 +340,16 @@ def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]: } -def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, Any] | None: +def _scan_session_row( + session_manager: SessionManager, + path: Path, + webui_dir: Path, +) -> dict[str, Any] | None: storage_key = SessionManager._session_key_from_path(path) # pyright: ignore[reportPrivateUsage] if storage_key is None: return None try: + signature = _file_signature(path) with open(path, encoding="utf-8") as f: first_line = f.readline().strip() if not first_line: @@ -330,7 +396,6 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, continue if not fallback_preview and item.get("role") == "assistant": fallback_preview = text - signature = _file_signature(path) created_at_s = data.get("created_at") updated_at_s = data.get("updated_at") if not created_at_s or not updated_at_s: @@ -338,7 +403,8 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, created_at_s = created_at_s or fallback_time updated_at_s = updated_at_s or fallback_time key = data.get("key") or storage_key - activity_signature = _webui_activity_signature(key) + metadata = data.get("metadata", {}) + activity_signature = _webui_activity_signature(key, webui_dir) activity_updated_at = _webui_activity_updated_at(activity_signature) return { "key": key, @@ -348,9 +414,10 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, visible_message_at, activity_updated_at, ), - "title": _metadata_title(data.get("metadata", {})), + "title": _metadata_title(metadata), "preview": preview or fallback_preview, - _MODEL_PRESET_FIELD: model_preset_from_metadata(data.get("metadata", {})), + _MODEL_PRESET_FIELD: model_preset_from_metadata(metadata), + **_indexed_workspace_scope_fields(metadata), "file": path.name, "mtime_ns": signature["mtime_ns"], "size": signature["size"], @@ -360,4 +427,4 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, repaired = session_manager._repair(storage_key) # pyright: ignore[reportPrivateUsage] if repaired is None: return None - return _indexed_row_for_session(repaired, path) + return _indexed_row_for_session(repaired, path, webui_dir) diff --git a/nanobot/webui/transcript.py b/nanobot/webui/transcript.py index f9c43d415..0c31b90ae 100644 --- a/nanobot/webui/transcript.py +++ b/nanobot/webui/transcript.py @@ -28,7 +28,8 @@ WEBUI_TRANSCRIPT_SCHEMA_VERSION = 3 WEBUI_FORK_MARKER_EVENT = "fork_marker" WEBUI_TRANSCRIPT_INCOMPLETE_KEY = "transcript_incomplete" _MAX_TRANSCRIPT_FILE_BYTES = 8 * 1024 * 1024 -_TARGET_ACTIVE_TRANSCRIPT_BYTES = _MAX_TRANSCRIPT_FILE_BYTES // 2 +_ACTIVE_TRANSCRIPT_ROTATE_BYTES = 2 * 1024 * 1024 +_TARGET_ACTIVE_TRANSCRIPT_BYTES = _ACTIVE_TRANSCRIPT_ROTATE_BYTES // 2 _TRANSCRIPT_SEGMENT_MANIFEST_VERSION = 2 _TRANSCRIPT_ACTIVE_CHUNK_ID = "active" _TRANSCRIPT_SEGMENT_RE = re.compile(r"^\d{6}\.jsonl$") @@ -284,12 +285,12 @@ def _normalize_manifest_entry(session_key: str, entry: Any) -> dict[str, Any] | } -def _write_segment_manifest(session_key: str, segment_ids: list[str]) -> None: +def _write_segment_manifest(session_key: str, entries: list[dict[str, Any]]) -> None: directory = webui_transcript_segments_dir(session_key) directory.mkdir(parents=True, exist_ok=True) data = { "version": _TRANSCRIPT_SEGMENT_MANIFEST_VERSION, - "segments": [_segment_manifest_entry(session_key, segment_id) for segment_id in segment_ids], + "segments": entries, } path = _webui_transcript_manifest_path(session_key) tmp_path = path.with_suffix(".json.tmp") @@ -301,17 +302,14 @@ def _write_segment_manifest(session_key: str, segment_ids: list[str]) -> None: raise -def _rebuild_segment_manifest(session_key: str) -> list[str]: +def _rebuild_segment_manifest(session_key: str) -> list[dict[str, Any]]: segment_ids = _segment_ids_on_disk(session_key) - if segment_ids: - _write_segment_manifest(session_key, segment_ids) + entries = [_segment_manifest_entry(session_key, segment_id) for segment_id in segment_ids] + if entries: + _write_segment_manifest(session_key, entries) else: _webui_transcript_manifest_path(session_key).unlink(missing_ok=True) - return segment_ids - - -def _rebuilt_segment_manifest_entries(session_key: str) -> list[dict[str, Any]]: - return [_segment_manifest_entry(session_key, segment_id) for segment_id in _rebuild_segment_manifest(session_key)] + return entries def _read_segment_manifest_entries(session_key: str) -> list[dict[str, Any]]: @@ -320,7 +318,7 @@ def _read_segment_manifest_entries(session_key: str) -> list[dict[str, Any]]: return [] path = _webui_transcript_manifest_path(session_key) if not path.is_file(): - return _rebuilt_segment_manifest_entries(session_key) + return _rebuild_segment_manifest(session_key) try: data = json.loads(path.read_text(encoding="utf-8")) manifest = cast(dict[str, Any], data) if isinstance(data, dict) else None @@ -330,18 +328,18 @@ def _read_segment_manifest_entries(session_key: str) -> list[dict[str, Any]]: or manifest.get("version") != _TRANSCRIPT_SEGMENT_MANIFEST_VERSION or not isinstance(raw_segments, list) ): - return _rebuilt_segment_manifest_entries(session_key) + return _rebuild_segment_manifest(session_key) entries: list[dict[str, Any]] = [] for entry in cast(list[Any], raw_segments): normalized = _normalize_manifest_entry(session_key, entry) if normalized is None: - return _rebuilt_segment_manifest_entries(session_key) + return _rebuild_segment_manifest(session_key) entries.append(normalized) if [entry["id"] for entry in entries] != _segment_ids_on_disk(session_key): - return _rebuilt_segment_manifest_entries(session_key) + return _rebuild_segment_manifest(session_key) return entries except (OSError, json.JSONDecodeError, TypeError, AttributeError): - return _rebuilt_segment_manifest_entries(session_key) + return _rebuild_segment_manifest(session_key) def _read_segment_ids(session_key: str) -> list[str]: @@ -351,26 +349,40 @@ def _read_segment_ids(session_key: str) -> list[str]: def _append_segment_turns(session_key: str, turns: list[list[dict[str, Any]]]) -> None: if not turns: return - segment_ids = _read_segment_ids(session_key) - next_id = int(segment_ids[-1]) + 1 if segment_ids else 1 + entries = _read_segment_manifest_entries(session_key) + next_id = int(entries[-1]["id"]) + 1 if entries else 1 batch: list[list[dict[str, Any]]] = [] batch_bytes = 0 + + def write_batch() -> None: + nonlocal next_id + segment_id = f"{next_id:06d}" + path = _segment_file_path(session_key, segment_id) + _write_records_to_path(path, _flatten_turns(batch)) + entries.append({ + "id": segment_id, + "bytes": path.stat().st_size, + "turn_count": len(batch), + "user_count": sum( + 1 + for turn in batch + for row in turn + if _is_user_transcript_row(row) + ), + }) + next_id += 1 + for turn in turns: turn_bytes = _records_bytes(turn) if batch and batch_bytes + turn_bytes > _MAX_TRANSCRIPT_FILE_BYTES: - segment_id = f"{next_id:06d}" - _write_records_to_path(_segment_file_path(session_key, segment_id), _flatten_turns(batch)) - segment_ids.append(segment_id) - next_id += 1 + write_batch() batch = [] batch_bytes = 0 batch.append(turn) batch_bytes += turn_bytes if batch: - segment_id = f"{next_id:06d}" - _write_records_to_path(_segment_file_path(session_key, segment_id), _flatten_turns(batch)) - segment_ids.append(segment_id) - _write_segment_manifest(session_key, segment_ids) + write_batch() + _write_segment_manifest(session_key, entries) def _rotate_active_transcript_if_needed(session_key: str) -> None: @@ -378,7 +390,7 @@ def _rotate_active_transcript_if_needed(session_key: str) -> None: if not path.is_file(): return try: - if path.stat().st_size <= _MAX_TRANSCRIPT_FILE_BYTES: + if path.stat().st_size <= _ACTIVE_TRANSCRIPT_ROTATE_BYTES: return except OSError: return @@ -426,6 +438,16 @@ def _read_chunk_turns(session_key: str, chunk_id: str) -> list[list[dict[str, An return _split_transcript_turns(_read_transcript_file(path)) +def _cached_chunk_turns( + session_key: str, + chunk_id: str, + turn_cache: dict[str, list[list[dict[str, Any]]]], +) -> list[list[dict[str, Any]]]: + if chunk_id not in turn_cache: + turn_cache[chunk_id] = _read_chunk_turns(session_key, chunk_id) + return turn_cache[chunk_id] + + def _encode_page_cursor(before_turn_ordinal: int) -> str: raw = json.dumps( {"before_turn": before_turn_ordinal}, @@ -462,7 +484,10 @@ def _coerce_page_limit(limit: int | None) -> int: return max(1, min(_MAX_TRANSCRIPT_PAGE_LIMIT, int(limit))) -def _chunk_turn_refs(session_key: str) -> list[_TranscriptChunkRef]: +def _chunk_turn_refs( + session_key: str, + turn_cache: dict[str, list[list[dict[str, Any]]]], +) -> list[_TranscriptChunkRef]: _rotate_active_transcript_if_needed(session_key) refs: list[_TranscriptChunkRef] = [] ordinal = 0 @@ -474,7 +499,11 @@ def _chunk_turn_refs(session_key: str) -> list[_TranscriptChunkRef]: refs.append(_TranscriptChunkRef(chunk_id, ordinal, turn_count, int(entry["user_count"]))) ordinal += turn_count if webui_transcript_path(session_key).is_file(): - active_turns = _read_chunk_turns(session_key, _TRANSCRIPT_ACTIVE_CHUNK_ID) + active_turns = _cached_chunk_turns( + session_key, + _TRANSCRIPT_ACTIVE_CHUNK_ID, + turn_cache, + ) active_turn_count = len(active_turns) if active_turn_count > 0: refs.append( @@ -492,6 +521,7 @@ def _count_user_messages_before_ordinal( session_key: str, chunks: list[_TranscriptChunkRef], before_ordinal: int, + turn_cache: dict[str, list[list[dict[str, Any]]]], ) -> int: total = 0 for chunk in chunks: @@ -503,7 +533,7 @@ def _count_user_messages_before_ordinal( if local_end >= chunk.turn_count: total += chunk.user_count continue - turns = _read_chunk_turns(session_key, chunk.chunk_id) + turns = _cached_chunk_turns(session_key, chunk.chunk_id, turn_cache) total += sum( 1 for turn in turns[:local_end] @@ -521,7 +551,8 @@ def _select_transcript_page( _manifest_rebuilt: bool = False, ) -> tuple[list[dict[str, Any]], dict[str, Any]]: page_limit = _coerce_page_limit(limit) - chunks = _chunk_turn_refs(session_key) + turn_cache: dict[str, list[list[dict[str, Any]]]] = {} + chunks = _chunk_turn_refs(session_key, turn_cache) total_turns = sum(chunk.turn_count for chunk in chunks) before_ordinal = _decode_page_cursor(before) upper_ordinal = total_turns if before_ordinal is None else min(before_ordinal, total_turns) @@ -534,7 +565,7 @@ def _select_transcript_page( local_upper = min(chunk.turn_count, upper_ordinal - chunk.start_ordinal) if local_upper <= 0: continue - turns = _read_chunk_turns(session_key, chunk.chunk_id) + turns = _cached_chunk_turns(session_key, chunk.chunk_id, turn_cache) if ( chunk.chunk_id != _TRANSCRIPT_ACTIVE_CHUNK_ID and len(turns) != chunk.turn_count @@ -585,6 +616,7 @@ def _select_transcript_page( session_key, chunks, first_ref.ordinal, + turn_cache, ), } return lines, page @@ -1182,19 +1214,18 @@ def _is_recoverable_answer_record(record: dict[str, Any]) -> bool: } -def recover_incomplete_turns_from_session( - lines: list[dict[str, Any]], - session_messages: list[dict[str, Any]] | None, - *, - session_key: str, -) -> list[dict[str, Any]]: - """Recover marked transcript answers only when one durable session turn matches.""" - if not lines or not session_messages: - return lines - session_turns = _session_backfill_turns(session_key, session_messages) - if not session_turns: - return lines +def _needs_incomplete_turn_recovery(lines: list[dict[str, Any]]) -> bool: + return any( + record.get("event") == "turn_end" + and record.get(WEBUI_TRANSCRIPT_INCOMPLETE_KEY) is True + for record in lines + ) + +def _recover_incomplete_turns( + lines: list[dict[str, Any]], + session_turns: list[_SessionBackfillTurn], +) -> list[dict[str, Any]]: recovered: list[dict[str, Any]] = [] for turn in _split_transcript_turns(lines): turn_end = turn[-1] if turn else None @@ -1244,6 +1275,21 @@ def recover_incomplete_turns_from_session( return recovered +def recover_incomplete_turns_from_session( + lines: list[dict[str, Any]], + session_messages: list[dict[str, Any]] | None, + *, + session_key: str, +) -> list[dict[str, Any]]: + """Recover marked transcript answers only when one durable session turn matches.""" + if not lines or not session_messages or not _needs_incomplete_turn_recovery(lines): + return lines + session_turns = _session_backfill_turns(session_key, session_messages) + if not session_turns: + return lines + return _recover_incomplete_turns(lines, session_turns) + + def _with_backfilled_user( records: list[dict[str, Any]], user_event: dict[str, Any], @@ -1254,18 +1300,19 @@ def _with_backfilled_user( return records -def inject_missing_user_events_from_session( - session_key: str, - lines: list[dict[str, Any]], - session_messages: list[dict[str, Any]] | None, -) -> list[dict[str, Any]]: - """Backfill user rows for legacy WebUI transcripts that only stored assistant streams.""" - if not lines or not session_messages: - return lines - session_turns = _session_backfill_turns(session_key, session_messages) - if not session_turns: - return lines +def _needs_user_event_backfill(lines: list[dict[str, Any]]) -> bool: + for turn in _split_transcript_turns(lines): + if any(record.get("event") == "user" for record in turn): + continue + if _transcript_turn_signature(turn): + return True + return False + +def _inject_missing_user_events( + lines: list[dict[str, Any]], + session_turns: list[_SessionBackfillTurn], +) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] session_cursor = 0 for turn in _split_transcript_turns(lines): @@ -1280,6 +1327,20 @@ def inject_missing_user_events_from_session( return out +def inject_missing_user_events_from_session( + session_key: str, + lines: list[dict[str, Any]], + session_messages: list[dict[str, Any]] | None, +) -> list[dict[str, Any]]: + """Backfill user rows for legacy WebUI transcripts that only stored assistant streams.""" + if not lines or not session_messages or not _needs_user_event_backfill(lines): + return lines + session_turns = _session_backfill_turns(session_key, session_messages) + if not session_turns: + return lines + return _inject_missing_user_events(lines, session_turns) + + def _format_tool_call_trace(call: Any) -> str | None: if not call or not isinstance(call, dict): return None @@ -2358,6 +2419,7 @@ def build_webui_thread_response( augment_assistant_media: Callable[[list[str]], list[dict[str, Any]]] | None = None, augment_assistant_text: Callable[[str], str] | None = None, session_messages: list[dict[str, Any]] | None = None, + session_messages_loader: Callable[[], list[dict[str, Any]] | None] | None = None, active_turn_started_at: float | None = None, active_turn_id: str | None = None, active_turn_transcript_persistence_failed: bool = False, @@ -2374,12 +2436,20 @@ def build_webui_thread_response( lines = _annotate_replay_identities(read_transcript_lines(session_key)) if not lines and active_turn_started_at is None: return None - lines = inject_missing_user_events_from_session(session_key, lines, session_messages) - lines = recover_incomplete_turns_from_session( - lines, - session_messages, - session_key=session_key, - ) + needs_user_backfill = _needs_user_event_backfill(lines) + needs_incomplete_recovery = _needs_incomplete_turn_recovery(lines) + if ( + session_messages is None + and session_messages_loader is not None + and (needs_user_backfill or needs_incomplete_recovery) + ): + session_messages = session_messages_loader() + if session_messages and (needs_user_backfill or needs_incomplete_recovery): + session_turns = _session_backfill_turns(session_key, session_messages) + if needs_user_backfill: + lines = _inject_missing_user_events(lines, session_turns) + if needs_incomplete_recovery: + lines = _recover_incomplete_turns(lines, session_turns) lines = _ensure_replay_identities(lines) fork_boundary = fork_boundary_message_count(lines) msgs = replay_transcript_to_ui_messages( diff --git a/nanobot/webui/workspaces.py b/nanobot/webui/workspaces.py index 56dc205f7..c4b8e46a1 100644 --- a/nanobot/webui/workspaces.py +++ b/nanobot/webui/workspaces.py @@ -191,24 +191,47 @@ class WebUIWorkspaceController: self._default_restrict_to_workspace, ) - def scope_for_session_key(self, session_key: str) -> WorkspaceScope: - if self._sessions is None: - return self.default_scope() - data = self._sessions.read_session_metadata(session_key) - session_data = data if data is not None else {} - metadata = session_data.get("metadata", {}) - if not isinstance(metadata, dict) or WORKSPACE_SCOPE_METADATA_KEY not in metadata: - return self.default_scope() - metadata = cast(dict[str, Any], metadata) + def _scope_from_metadata_value( + self, + raw_scope: object, + *, + default_scope: WorkspaceScope | None = None, + ) -> WorkspaceScope: try: return validate_workspace_scope_payload( - metadata.get(WORKSPACE_SCOPE_METADATA_KEY), + raw_scope, default_workspace=self._default_workspace, default_restrict_to_workspace=self._default_restrict_to_workspace, source_channel=_WEBUI_SCOPE_CHANNEL, ) except WorkspaceScopeError: + return default_scope if default_scope is not None else self.default_scope() + + def scope_for_indexed_metadata( + self, + raw_scope: object, + *, + scope_present: bool, + default_scope: WorkspaceScope, + ) -> WorkspaceScope: + """Resolve a sidebar-only metadata snapshot without an authority-store read.""" + if not scope_present: + return default_scope + return self._scope_from_metadata_value(raw_scope, default_scope=default_scope) + + def scope_for_session_key(self, session_key: str) -> WorkspaceScope: + if self._sessions is None: return self.default_scope() + data = self._sessions.read_session_metadata(session_key) + if not isinstance(data, dict): + return self.default_scope() + metadata = data.get("metadata", {}) + if not isinstance(metadata, dict) or WORKSPACE_SCOPE_METADATA_KEY not in metadata: + return self.default_scope() + metadata_data = cast(dict[str, Any], metadata) + return self._scope_from_metadata_value( + cast(object, metadata_data.get(WORKSPACE_SCOPE_METADATA_KEY)) + ) def payload(self, *, controls_available: bool) -> dict[str, Any]: return workspaces_payload( diff --git a/nanobot/webui/ws_http.py b/nanobot/webui/ws_http.py index 9dcfdfbfe..1f895b17a 100644 --- a/nanobot/webui/ws_http.py +++ b/nanobot/webui/ws_http.py @@ -27,6 +27,7 @@ from nanobot.command.builtin import builtin_command_palette from nanobot.cron.session_turns import is_bound_cron_job from nanobot.cron.types import CronJob, CronSchedule from nanobot.runtime_context import public_history_messages +from nanobot.security.workspace_access import WorkspaceScope from nanobot.triggers.local_types import LocalTrigger from nanobot.utils.subagent_channel_display import scrub_subagent_messages_for_channel from nanobot.webui.file_preview import ( @@ -38,6 +39,9 @@ from nanobot.webui.gateway_tokens import GatewayTokenStore, token_response_paylo from nanobot.webui.http_utils import ( case_insensitive_header as _case_insensitive_header, ) +from nanobot.webui.http_utils import ( + combined_list_header as _combined_list_header, +) from nanobot.webui.http_utils import ( host_for_url as _host_for_url, ) @@ -82,7 +86,11 @@ from nanobot.webui.session_automations import ( session_automation_jobs, session_automations_payload, ) -from nanobot.webui.session_list_index import list_webui_sessions +from nanobot.webui.session_list_index import ( + WEBUI_SESSION_INDEX_INTERNAL_FIELDS, + indexed_workspace_scope, + list_webui_sessions, +) from nanobot.webui.sidebar_state import ( read_webui_sidebar_state, write_webui_sidebar_state, @@ -422,7 +430,10 @@ class GatewayHTTPHandler: if self.session_manager is None: return _http_error(503, "session manager unavailable") payload = await asyncio.to_thread(self._sessions_list_payload) - return _http_json_response(payload) + return _http_json_response( + payload, + accept_encoding=_combined_list_header(request.headers, "Accept-Encoding"), + ) def _sessions_list_payload(self) -> dict[str, Any]: assert self.session_manager is not None @@ -430,16 +441,28 @@ class GatewayHTTPHandler: from nanobot.session.webui_turns import websocket_turn_wall_started_at cleaned: list[dict[str, Any]] = [] + default_scope: WorkspaceScope | None = None for s in sessions: key = s.get("key") if not (isinstance(key, str) and key.startswith("websocket:")): continue - row = {k: v for k, v in s.items() if k != "path"} + row = { + k: v + for k, v in s.items() + if k != "path" and k not in WEBUI_SESSION_INDEX_INTERNAL_FIELDS + } chat_id = key.split(":", 1)[1] started_at = websocket_turn_wall_started_at(chat_id) if started_at is not None: row["run_started_at"] = started_at - scope = self.workspaces.scope_for_session_key(key) + if default_scope is None: + default_scope = self.workspaces.default_scope() + scope_present, raw_scope = indexed_workspace_scope(s) + scope = self.workspaces.scope_for_indexed_metadata( + raw_scope, + scope_present=scope_present, + default_scope=default_scope, + ) row["workspace_scope"] = scope.payload() cleaned.append(row) return {"sessions": cleaned} @@ -481,17 +504,21 @@ class GatewayHTTPHandler: if not _is_websocket_channel_session_key(decoded_key): return _http_error(404, "session not found") scope = self.workspaces.scope_for_session_key(decoded_key) - session_messages: list[dict[str, Any]] | None = None - if self.session_manager is not None: + + def load_session_messages() -> list[dict[str, Any]] | None: + if self.session_manager is None: + return None session_data = self.session_manager.read_session_file(decoded_key) raw_messages = session_data.get("messages") if isinstance(session_data, dict) else None - if isinstance(raw_messages, list): - raw_session_messages = cast(list[Any], raw_messages) - session_messages = [ - cast(dict[str, Any], raw_message) - for raw_message in raw_session_messages - if isinstance(raw_message, dict) - ] + if not isinstance(raw_messages, list): + return None + raw_session_messages = cast(list[Any], raw_messages) + return [ + cast(dict[str, Any], raw_message) + for raw_message in raw_session_messages + if isinstance(raw_message, dict) + ] + query = _parse_query(request.path) raw_limit = _query_first(query, "limit") limit: int | None = None @@ -524,7 +551,7 @@ class GatewayHTTPHandler: text, workspace_path=scope.project_path, ), - session_messages=session_messages, + session_messages_loader=load_session_messages, active_turn_started_at=active_turn_started_at, active_turn_id=active_turn_id, active_turn_transcript_persistence_failed=( @@ -537,7 +564,10 @@ class GatewayHTTPHandler: if data is None: return _http_error(404, "webui thread not found") data["workspace_scope"] = scope.payload() - return _http_json_response(data) + return _http_json_response( + data, + accept_encoding=_combined_list_header(request.headers, "Accept-Encoding"), + ) def _handle_file_preview(self, request: WsRequest, key: str) -> Response: if not self.check_api_token(request): diff --git a/tests/utils/test_webui_thread_disk.py b/tests/utils/test_webui_thread_disk.py index ee825dc42..8486beb74 100644 --- a/tests/utils/test_webui_thread_disk.py +++ b/tests/utils/test_webui_thread_disk.py @@ -13,6 +13,7 @@ from nanobot.webui.transcript import ( def test_delete_webui_thread_removes_legacy_json_and_transcript(tmp_path, monkeypatch) -> None: monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) monkeypatch.setattr("nanobot.webui.transcript._MAX_TRANSCRIPT_FILE_BYTES", 520) + monkeypatch.setattr("nanobot.webui.transcript._ACTIVE_TRANSCRIPT_ROTATE_BYTES", 520) monkeypatch.setattr("nanobot.webui.transcript._TARGET_ACTIVE_TRANSCRIPT_BYTES", 260) key = "websocket:k1" json_path = webui_thread_file_path(key) diff --git a/tests/utils/test_webui_transcript.py b/tests/utils/test_webui_transcript.py index 1fe3fa6e1..45ba5252d 100644 --- a/tests/utils/test_webui_transcript.py +++ b/tests/utils/test_webui_transcript.py @@ -2,6 +2,7 @@ from __future__ import annotations +import nanobot.webui.transcript as transcript_module from nanobot.session.history_visibility import HIDDEN_HISTORY_META from nanobot.webui.transcript import ( WEBUI_TRANSCRIPT_SCHEMA_VERSION, @@ -38,6 +39,7 @@ def test_append_stamps_created_at_ms(tmp_path, monkeypatch) -> None: def _force_small_transcript_budget(monkeypatch, *, limit: int = 520, target: int = 260) -> None: monkeypatch.setattr("nanobot.webui.transcript._MAX_TRANSCRIPT_FILE_BYTES", limit) + monkeypatch.setattr("nanobot.webui.transcript._ACTIVE_TRANSCRIPT_ROTATE_BYTES", limit) monkeypatch.setattr("nanobot.webui.transcript._TARGET_ACTIVE_TRANSCRIPT_BYTES", target) @@ -122,6 +124,28 @@ def test_segmented_transcript_paginates_latest_and_older_without_overlap( ] +def test_latest_page_reads_active_chunk_once(tmp_path, monkeypatch) -> None: + monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) + key = "websocket:single-active-read" + for idx in range(1, 7): + _append_numbered_turn(key, "single-active-read", idx) + + original = transcript_module._read_chunk_turns + read_chunk_ids: list[str] = [] + + def track_read(session_key: str, chunk_id: str) -> list[list[dict]]: + read_chunk_ids.append(chunk_id) + return original(session_key, chunk_id) + + monkeypatch.setattr(transcript_module, "_read_chunk_turns", track_read) + + latest = build_webui_thread_response(key, limit=4, direction="latest") + + assert latest is not None + assert _message_contents(latest) == _numbered_turn_texts(5, 6) + assert read_chunk_ids == ["active"] + + def test_page_cursor_survives_active_rotation_after_latest_page( tmp_path, monkeypatch, @@ -148,15 +172,53 @@ def test_segment_manifest_can_be_rebuilt_when_missing_or_corrupt(tmp_path, monke key = "websocket:manifest" _write_segmented_turns(tmp_path, monkeypatch, key, "manifest", 4) - manifest = webui_transcript_segments_dir(key) / "manifest.json" + segment_dir = webui_transcript_segments_dir(key) + segment_names = sorted(path.name for path in segment_dir.glob("*.jsonl")) + assert segment_names + original = transcript_module._read_transcript_file + segment_reads: list[str] = [] + + def track_read(path): + if path.parent == segment_dir and path.suffix == ".jsonl": + segment_reads.append(path.name) + return original(path) + + monkeypatch.setattr(transcript_module, "_read_transcript_file", track_read) + manifest = segment_dir / "manifest.json" manifest.write_text("{not json", encoding="utf-8") + entries = transcript_module._read_segment_manifest_entries(key) + + assert [entry["id"] for entry in entries] == [path.removesuffix(".jsonl") for path in segment_names] + assert segment_reads == segment_names + lines = read_transcript_lines(key) assert len([line for line in lines if line.get("event") == "user"]) == 4 assert manifest.read_text(encoding="utf-8").lstrip().startswith("{") +def test_rotation_does_not_reread_existing_segments(tmp_path, monkeypatch) -> None: + key = "websocket:manifest-append" + _write_segmented_turns(tmp_path, monkeypatch, key, "manifest-append", 4) + segment_dir = webui_transcript_segments_dir(key) + assert list(segment_dir.glob("*.jsonl")) + + original = transcript_module._read_transcript_file + segment_reads: list[str] = [] + + def track_read(path): + if path.parent == segment_dir and path.suffix == ".jsonl": + segment_reads.append(path.name) + return original(path) + + monkeypatch.setattr(transcript_module, "_read_transcript_file", track_read) + for idx in range(5, 9): + _append_numbered_turn(key, "manifest-append", idx) + + assert segment_reads == [] + + def test_delete_webui_transcript_removes_segments(tmp_path, monkeypatch) -> None: from nanobot.webui.thread_disk import webui_thread_file_path from nanobot.webui.transcript import delete_webui_transcript, webui_transcript_path @@ -786,6 +848,83 @@ def test_build_response_restores_session_users_for_legacy_transcript( ] +def test_complete_transcript_does_not_load_session_messages(tmp_path, monkeypatch) -> None: + monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) + key = "websocket:complete-fast-path" + for event in ( + {"event": "user", "chat_id": "complete-fast-path", "text": "question"}, + {"event": "message", "chat_id": "complete-fast-path", "text": "answer"}, + {"event": "turn_end", "chat_id": "complete-fast-path"}, + ): + append_transcript_object(key, event) + + def fail_if_loaded() -> list[dict]: + raise AssertionError("complete transcripts must not read canonical session history") + + out = build_webui_thread_response( + key, + limit=4, + direction="latest", + session_messages_loader=fail_if_loaded, + ) + + assert out is not None + assert [(message["role"], message["content"]) for message in out["messages"]] == [ + ("user", "question"), + ("assistant", "answer"), + ] + + +def test_legacy_recovery_loads_session_and_builds_backfill_turns_once( + tmp_path, + monkeypatch, +) -> None: + monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path) + key = "websocket:lazy-legacy-recovery" + append_transcript_object( + key, + {"event": "message", "chat_id": "lazy-legacy-recovery", "text": "answer"}, + ) + append_transcript_object( + key, + { + "event": "turn_end", + "chat_id": "lazy-legacy-recovery", + "transcript_incomplete": True, + }, + ) + + loader_calls = 0 + backfill_calls = 0 + original = transcript_module._session_backfill_turns + + def load_session_messages() -> list[dict]: + nonlocal loader_calls + loader_calls += 1 + return [ + {"role": "user", "content": "question"}, + {"role": "assistant", "content": "answer"}, + ] + + def track_backfill(session_key: str, session_messages: list[dict]): + nonlocal backfill_calls + backfill_calls += 1 + return original(session_key, session_messages) + + monkeypatch.setattr(transcript_module, "_session_backfill_turns", track_backfill) + + out = build_webui_thread_response(key, session_messages_loader=load_session_messages) + + assert out is not None + assert loader_calls == 1 + assert backfill_calls == 1 + assert [(message["role"], message["content"]) for message in out["messages"]] == [ + ("user", "question"), + ("assistant", "answer"), + ] + assert out["has_pending_tool_calls"] is False + + def test_build_response_restores_session_users_without_duplicating_new_transcript_users( tmp_path, monkeypatch, diff --git a/tests/utils/test_webui_workspaces.py b/tests/utils/test_webui_workspaces.py index a6688ae5d..634375fcf 100644 --- a/tests/utils/test_webui_workspaces.py +++ b/tests/utils/test_webui_workspaces.py @@ -1,9 +1,14 @@ import json +from unittest.mock import MagicMock import pytest -from nanobot.security.workspace_access import WorkspaceScopeError, default_workspace_scope -from nanobot.session.manager import SessionManager +from nanobot.security.workspace_access import ( + WORKSPACE_SCOPE_METADATA_KEY, + WorkspaceScopeError, + default_workspace_scope, +) +from nanobot.session.manager import SessionManager, SessionStore from nanobot.webui.workspaces import ( WebUIWorkspaceController, read_webui_default_access_mode, @@ -135,6 +140,33 @@ def test_webui_default_access_applies_to_unscoped_old_sessions(tmp_path, monkeyp assert new_scope.access_mode == "full" +def test_indexed_scope_preserves_missing_and_explicit_null_semantics(tmp_path, monkeypatch) -> None: + monkeypatch.setattr("nanobot.webui.workspaces.get_webui_dir", lambda: tmp_path / "webui") + default = tmp_path / "default" + default.mkdir() + write_webui_default_access_mode("full") + controller = WebUIWorkspaceController( + session_manager=None, + default_workspace=default, + default_restrict_to_workspace=True, + ) + webui_default = controller.default_scope() + + missing = controller.scope_for_indexed_metadata( + None, + scope_present=False, + default_scope=webui_default, + ) + explicit_null = controller.scope_for_indexed_metadata( + None, + scope_present=True, + default_scope=webui_default, + ) + + assert missing.access_mode == "full" + assert explicit_null.access_mode == "restricted" + + def test_webui_default_access_does_not_override_explicit_session_scope(tmp_path, monkeypatch) -> None: monkeypatch.setattr("nanobot.webui.workspaces.get_webui_dir", lambda: tmp_path / "webui") default = tmp_path / "default" @@ -185,6 +217,53 @@ def test_scope_for_session_key_reads_metadata_without_full_history( assert scope.access_mode == "full" +def test_scope_for_session_key_always_reads_the_active_store(tmp_path, monkeypatch) -> None: + monkeypatch.setattr("nanobot.webui.workspaces.get_webui_dir", lambda: tmp_path / "webui") + default = tmp_path / "default" + project = tmp_path / "project" + default.mkdir() + project.mkdir() + workspace = tmp_path / "session-data" + full_scope = default_workspace_scope(project, restrict_to_workspace=False) + restricted_scope = default_workspace_scope(project, restrict_to_workspace=True) + + residual_sessions = SessionManager(workspace) + residual = residual_sessions.get_or_create("websocket:cached") + residual.metadata[WORKSPACE_SCOPE_METADATA_KEY] = full_scope.metadata() + residual_sessions.save(residual) + + store = MagicMock(spec=SessionStore) + store.read_metadata.side_effect = [ + { + "key": "websocket:cached", + "created_at": None, + "updated_at": None, + "metadata": {WORKSPACE_SCOPE_METADATA_KEY: full_scope.metadata()}, + }, + { + "key": "websocket:cached", + "created_at": None, + "updated_at": None, + "metadata": {WORKSPACE_SCOPE_METADATA_KEY: restricted_scope.metadata()}, + }, + ] + sessions = SessionManager(workspace, store=store) + controller = WebUIWorkspaceController( + session_manager=sessions, + default_workspace=default, + default_restrict_to_workspace=True, + ) + + first = controller.scope_for_session_key("websocket:cached") + second = controller.scope_for_session_key("websocket:cached") + + assert first.project_path == project.resolve() + assert first.access_mode == "full" + assert second.project_path == project.resolve() + assert second.access_mode == "restricted" + assert store.read_metadata.call_count == 2 + + def test_remote_existing_chat_can_reduce_its_workspace_access(tmp_path, monkeypatch) -> None: monkeypatch.setattr("nanobot.webui.workspaces.get_webui_dir", lambda: tmp_path / "webui") default = tmp_path / "default" diff --git a/tests/webui/test_http_utils.py b/tests/webui/test_http_utils.py new file mode 100644 index 000000000..e23710332 --- /dev/null +++ b/tests/webui/test_http_utils.py @@ -0,0 +1,38 @@ +"""Tests for shared embedded WebUI HTTP helpers.""" + +import gzip +import json + +from nanobot.webui.http_utils import http_json_response + + +def test_http_json_response_compresses_large_payload_when_gzip_is_accepted() -> None: + payload = {"message": "响应内容" * 2_000} + + response = http_json_response(payload, accept_encoding="br, gzip; q=0.5") + + assert response.headers["Content-Encoding"] == "gzip" + assert response.headers["Vary"] == "Accept-Encoding" + assert int(response.headers["Content-Length"]) == len(response.body) + assert json.loads(gzip.decompress(response.body)) == payload + + +def test_http_json_response_preserves_identity_when_gzip_is_rejected() -> None: + payload = {"message": "x" * 8_000} + + response = http_json_response(payload, accept_encoding="gzip;q=0, br") + + assert "Content-Encoding" not in response.headers + assert response.headers["Vary"] == "Accept-Encoding" + assert int(response.headers["Content-Length"]) == len(response.body) + assert json.loads(response.body) == payload + + +def test_http_json_response_does_not_compress_small_payload() -> None: + payload = {"ok": True} + + response = http_json_response(payload, accept_encoding="gzip") + + assert "Content-Encoding" not in response.headers + assert response.headers["Vary"] == "Accept-Encoding" + assert json.loads(response.body) == payload diff --git a/tests/webui/test_session_list_index.py b/tests/webui/test_session_list_index.py index 47a1164ce..939819cc5 100644 --- a/tests/webui/test_session_list_index.py +++ b/tests/webui/test_session_list_index.py @@ -1,5 +1,6 @@ from __future__ import annotations +import io import os from datetime import datetime from pathlib import Path @@ -9,6 +10,7 @@ import pytest import nanobot.webui.session_list_index as session_list_index from nanobot.cron.session_turns import CRON_HISTORY_META from nanobot.providers.base import ProviderConversationState +from nanobot.security.workspace_access import WORKSPACE_SCOPE_METADATA_KEY from nanobot.session.automation_turns import AUTOMATION_HISTORY_META from nanobot.session.history_visibility import HIDDEN_HISTORY_META from nanobot.session.manager import SessionManager @@ -28,7 +30,7 @@ def test_webui_session_list_reuses_valid_index_without_scanning_files( assert list_webui_sessions(manager)[0]["preview"] == "indexed preview" assert list_webui_sessions(manager)[0]["model_preset"] == "fast" - def fail_scan(session_manager: SessionManager, path: Path) -> None: + def fail_scan(session_manager: SessionManager, path: Path, webui_dir: Path) -> None: raise AssertionError(f"unexpected session file scan: {path}") monkeypatch.setattr(session_list_index, "_scan_session_row", fail_scan) @@ -40,6 +42,89 @@ def test_webui_session_list_reuses_valid_index_without_scanning_files( assert rows[0]["model_preset"] == "fast" +def test_webui_session_list_indexes_workspace_scope_and_preserves_null( + tmp_path: Path, +) -> None: + manager = SessionManager(tmp_path) + project = tmp_path / "project" + project.mkdir() + + scoped = manager.get_or_create("websocket:scoped") + scoped.metadata[WORKSPACE_SCOPE_METADATA_KEY] = { + "project_path": str(project), + "access_mode": "full", + "future_extension": "x" * 5000, + } + manager.save(scoped) + explicit_null = manager.get_or_create("websocket:null") + explicit_null.metadata[WORKSPACE_SCOPE_METADATA_KEY] = None + manager.save(explicit_null) + manager.save(manager.get_or_create("websocket:missing")) + + rows = {row["key"]: row for row in list_webui_sessions(manager)} + + assert session_list_index.indexed_workspace_scope(rows["websocket:scoped"]) == ( + True, + {"project_path": str(project), "access_mode": "full"}, + ) + assert session_list_index.indexed_workspace_scope(rows["websocket:null"]) == (True, None) + assert session_list_index.indexed_workspace_scope(rows["websocket:missing"]) == (False, None) + + scoped.metadata[WORKSPACE_SCOPE_METADATA_KEY]["access_mode"] = "restricted" + manager.save(scoped) + + refreshed = {row["key"]: row for row in list_webui_sessions(manager)} + assert session_list_index.indexed_workspace_scope(refreshed["websocket:scoped"])[1] == { + "project_path": str(project), + "access_mode": "restricted", + } + + +def test_webui_session_list_does_not_cache_old_snapshot_with_new_signature( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manager = SessionManager(tmp_path) + session_key = "websocket:scope-race" + session = manager.get_or_create(session_key) + session.metadata[WORKSPACE_SCOPE_METADATA_KEY] = { + "project_path": str(tmp_path), + "access_mode": "full", + } + session.add_message("user", "hello") + manager.save(session) + session_path = manager._get_session_path(session_key) + original_open = open + scope_changed = False + + class RacingReader(io.StringIO): + def __next__(self) -> str: + nonlocal scope_changed + if not scope_changed: + scope_changed = True + current = manager.get_or_create(session_key) + current.metadata[WORKSPACE_SCOPE_METADATA_KEY] = { + "project_path": str(tmp_path), + "access_mode": "restricted", + } + manager.save(current) + return super().__next__() + + def racing_open(path, *args, **kwargs): + if Path(path) == session_path: + 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)[0] + second = list_webui_sessions(manager)[0] + + assert session_list_index.indexed_workspace_scope(first)[1]["access_mode"] == "full" + assert session_list_index.indexed_workspace_scope(second)[1]["access_mode"] == "restricted" + + def test_webui_session_list_rejects_invalid_internal_model_preset_metadata( tmp_path: Path, ) -> None: @@ -74,9 +159,13 @@ def test_webui_session_list_rescans_only_changed_file(tmp_path: Path, monkeypatc original_scan = session_list_index._scan_session_row scanned: list[str] = [] - def record_scan(session_manager: SessionManager, path: Path) -> dict | None: + def record_scan( + session_manager: SessionManager, + path: Path, + webui_dir: Path, + ) -> dict | None: scanned.append(path.name) - return original_scan(session_manager, path) + return original_scan(session_manager, path, webui_dir) monkeypatch.setattr(session_list_index, "_scan_session_row", record_scan) @@ -247,9 +336,13 @@ def test_webui_session_list_rescans_when_transcript_changes( original_scan = session_list_index._scan_session_row scanned: list[str] = [] - def record_scan(session_manager: SessionManager, path: Path) -> dict | None: + def record_scan( + session_manager: SessionManager, + path: Path, + webui_dir: Path, + ) -> dict | None: scanned.append(path.name) - return original_scan(session_manager, path) + return original_scan(session_manager, path, webui_dir) monkeypatch.setattr(session_list_index, "_scan_session_row", record_scan) diff --git a/webui/src/hooks/useSessions.ts b/webui/src/hooks/useSessions.ts index cb59a9dfc..fd63457cc 100644 --- a/webui/src/hooks/useSessions.ts +++ b/webui/src/hooks/useSessions.ts @@ -20,7 +20,7 @@ import type { } from "@/lib/types"; const EMPTY_MESSAGES: UIMessage[] = []; -const INITIAL_HISTORY_PAGE_LIMIT = 160; +const INITIAL_HISTORY_PAGE_LIMIT = 80; const OLDER_HISTORY_PAGE_LIMIT = 120; const CHAT_CREATE_TIMEOUT_MS = 60_000; diff --git a/webui/src/tests/useSessions.test.tsx b/webui/src/tests/useSessions.test.tsx index 9946379fd..3a302e512 100644 --- a/webui/src/tests/useSessions.test.tsx +++ b/webui/src/tests/useSessions.test.tsx @@ -685,7 +685,7 @@ describe("useSessions", () => { "tok", "websocket:paged", expect.objectContaining({ - limit: 160, + limit: 80, direction: "latest", signal: expect.any(AbortSignal), }),