"""Hermes Web UI -- Session model and in-memory session store."""import collectionsimport copyimport datetimeimport hashlibimport inspectimport jsonimport loggingimport osimport reimport threadingimport timeimport uuidfrom contextlib import closingfrom pathlib import Path import api.config as _cfgfrom api.compression_anchor import is_context_compression_markerfrom api.config import ( SESSION_DIR, SESSION_INDEX_FILE, SESSIONS, SESSIONS_MAX, LOCK, STREAMS, STREAMS_LOCK, DEFAULT_WORKSPACE, DEFAULT_MODEL, PROJECTS_FILE, HOME, get_effective_default_model, _get_session_agent_lock,)from api.workspace import get_last_workspacefrom api.usage import prompt_cache_hit_percentfrom api.agent_sessions import ( _is_continuation_session, is_cli_session_row, normalize_agent_session_source, open_state_db_readonly, read_importable_agent_session_rows, read_session_lineage_metadata,) logger = logging.getLogger(__name__)CLI_VISIBLE_SESSION_LIMIT = 20# How many messageful cron sessions to surface in the project-chip layer.# Needs to exceed CLI_VISIBLE_SESSION_LIMIT so older cron runs stay# addressable even when many newer non-cron sessions dominate the default# sidebar window (#3172).CRON_PROJECT_CHIP_LIMIT = 200WEBHOOK_PROJECT_CHIP_LIMIT = 200_CLI_SESSIONS_CACHE_TTL_SECONDS = 5.0# While a turn is actively streaming, hold the CLI/cron projection longer than# one poll interval (mirrors the route-level #4808 hold-down). The frontend# polls /api/sessions every ~5s during a stream; without a wider window the# CLI cache key advances on every streamed message row (see below) and the# expensive state.db CLI/cron projection is re-run on every poll. (#4842)_CLI_SESSIONS_CACHE_STREAMING_TTL_SECONDS = 30.0_CLI_SESSIONS_CACHE_LOCK = threading.Lock()_CLI_SESSIONS_CACHE_INFLIGHT: "dict[tuple, threading.Event]" = {}_CLI_SESSIONS_CACHE_INVALIDATION_VERSION = 0_CLI_SESSIONS_CACHE = {}_CLI_SESSIONS_CACHE_WAIT_SECONDS = 0.25# Event waits that keep stale rows visible while a rebuild is in flight._CLI_SESSIONS_CACHE_STALE_WAIT_SECONDS = 0.10 # Per-file parse cache for Claude Code JSONL transcripts (#4718/#4662 phase 4).# ``~/.claude/projects`` is a GLOBAL, profile-independent directory, but the# sidebar re-derives every Claude Code row from scratch on each /api/sessions# build — fully re-reading and JSON-parsing up to CLAUDE_CODE_MAX_FILES# transcripts line-by-line (hundreds of MB) just to recover a title + message# count. That parse dominates the cold sidebar build (~650-1000ms measured on a# 200-file / ~130MB tree) and it repeats on every profile switch, on the 5s# CLI-cache expiry, and on every sidebar poll, because the higher CLI cache is# keyed per active profile while the underlying transcripts never change between# switches. This cache memoizes the EXPENSIVE per-file parse result keyed by the# file's (path, mtime_ns, size, ctime_ns); a warm sidebar build then re-stats the# files (~4ms for 200) instead of re-parsing them. Any external edit/append to a# transcript changes mtime_ns/size/ctime_ns and transparently invalidates just# that one file's entry. Bounded so a pathological projects tree can't grow it unbounded._CLAUDE_CODE_PARSE_CACHE_LOCK = threading.Lock()_CLAUDE_CODE_PARSE_CACHE: "collections.OrderedDict[tuple, tuple]" = collections.OrderedDict()_CLAUDE_CODE_PARSE_CACHE_MAX = 1000 # Per-file cache for the UI-owned sidecar metadata (title + archived) that the# state.db sidebar projection overlays onto each CLI/cron row (#4842). The# projection calls _state_projection_sidecar_metadata() once per row in BOTH# the main visible pass AND the higher-capped (CRON_PROJECT_CHIP_LIMIT=200)# cron-only second pass, and each call was an uncached open() + 64KB prefix# read + a pure-Python JSON-key scan. On a cron-heavy profile that is up to# ~200 sidecar file reads per /api/sessions build — and because the enclosing# _CLI_SESSIONS_CACHE is keyed on a state.db content fingerprint that advances# on every streamed message row, that whole scan was re-paid on essentially# every 5s poll during a live turn (the "100% CPU / multi-second get_cli_sessions"# in #4842/#4808/#4672). This memoizes the parse result keyed by the sidecar's# (path, mtime_ns, size, ctime_ns) stat signature: a warm projection re-stats# each file (~1 stat) instead of re-reading+parsing it, while any genuine# rename/archive/edit bumps the signature and transparently invalidates just# that one entry. Bounded so a pathological session store can't grow it without# limit. Mirrors the Claude Code parse cache (#4718)._SIDECAR_METADATA_CACHE_LOCK = threading.Lock()_SIDECAR_METADATA_CACHE: "collections.OrderedDict[tuple, dict]" = collections.OrderedDict()_SIDECAR_METADATA_CACHE_MAX = 2000 # #5854: authoritative facts for a LEGACY (pre-#5854) sidecar whose scenes# serialize before `messages`, so the cheap metadata-prefix read can't recover# its message_count or scene fingerprint. Without this, an unchanged legacy# large-scene session would full-parse on every poll (recreating the #4633# churn for legacy files) and could not be LRU-evicted. Populated once per file# from a full Session.load(); keyed by the sidecar's stat signature so any edit# invalidates it. Bounded. Value: {"message_count": int, "scene_index": dict}._LEGACY_SIDECAR_FACTS_LOCK = threading.Lock()_LEGACY_SIDECAR_FACTS: "collections.OrderedDict[tuple, dict]" = collections.OrderedDict()_LEGACY_SIDECAR_FACTS_MAX = 2000 # ---------------------------------------------------------------------------# Stale temp-file cleanup# ---------------------------------------------------------------------------# Both Session.save() and _write_session_index() use the atomic-write pattern:# write to <path>.tmp.<pid>.<tid> → os.replace() to final path# If the process crashes between write and replace the .tmp file is left# behind. Because the name embeds pid + tid, leftover files can never be# reused by a different process/thread, so they are safe to remove on the# next startup. _cleanup_stale_tmp_files() is called from the full-rebuild# path of _write_session_index (i.e. at first index access / startup) and# removes any *.tmp.* file whose mtime is older than one hour.# --------------------------------------------------------------------------- _STALE_TMP_AGE_SECONDS = 3600 # 1 hour # ---------------------------------------------------------------------------# Windows-safe os.replace() with retry# ---------------------------------------------------------------------------# On Windows, os.replace() raises WinError 5 (ERROR_ACCESS_DENIED) when the# target file is momentarily locked by another process (antivirus scanner,# browser polling the session JSON, etc.). This helper retries with# exponential backoff on PermissionError, which is the Python exception# mapped from WinError 5. On non-Windows platforms it is a thin wrapper# (one attempt, no delay).# --------------------------------------------------------------------------- _WINDOWS_REPLACE_MAX_RETRIES = 5_WINDOWS_REPLACE_INITIAL_DELAY = 0.05 # 50 ms def _safe_replace(src: Path, dst: Path) -> None: """Atomic replace with retries on Windows file-locking errors.""" if os.name != 'nt': os.replace(src, dst) return delay = _WINDOWS_REPLACE_INITIAL_DELAY for attempt in range(_WINDOWS_REPLACE_MAX_RETRIES): try: os.replace(src, dst) return except PermissionError: if attempt == _WINDOWS_REPLACE_MAX_RETRIES - 1: raise time.sleep(delay) delay *= 2 # 50 -> 100 -> 200 -> 400 -> 800 ms # Serializes index writers so concurrent Session.save() calls cannot race on# stale baselines while still allowing LOCK to be released before disk I/O._INDEX_WRITE_LOCK = threading.RLock()_SESSION_INDEX_REBUILD_LOCK = threading.Lock()_SESSION_INDEX_REBUILD_THREAD = None_SESSION_INDEX_REBUILD_THREAD_TARGET: tuple[Path, Path] | None = None # Serializes ``_record_webui_zero_message_orphan_tombstone`` /# ``_clear_webui_zero_message_orphan_tombstone`` so two concurrent sidebar# polls (or a poll racing ``Session.save`` / ``new_session`` /# ``import_cli_session``) cannot lose each other's load-modify-write/unlink.# Without this lock each operation rewrites the entire tombstone file from# scratch, so a concurrent recorder and clearer can land last-writer-wins and# silently drop each other's update — defeating the self-healing invariant# that ``Session.save`` clears the tombstone the same poll that re-prunes# would otherwise re-add the row for. ``threading.Lock`` is sufficient (the# WebUI sidebar polling path is single-process) but must wrap the WHOLE# load-modify-write/unlink sequence in both helpers._WEBUI_ZERO_MESSAGE_ORPHAN_TOMBSTONE_LOCK = threading.Lock()_WEBUI_DELETED_SESSION_TOMBSTONE_LOCK = threading.Lock() # Path-safety contract for session IDs. Accept alphanumerics, underscore, and# hyphen so API/gateway-issued ids (``api-*``, ``reachy-voice-*``) round-trip# through filesystem load/save/delete/worktree paths without traversal risk.# Dots and slashes are rejected so the id can never name a parent directory# or hide an unexpected extension._SAFE_SID_CHARS = frozenset( '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-') def is_safe_session_id(sid) -> bool: """Return True iff ``sid`` is a non-empty path-safe session id. Centralizes the validation previously duplicated across ``Session.load``, ``Session.load_metadata_only``, ``_repair_stale_pending``, ``/api/session/worktree/remove``, and ``/api/session/delete`` so every call site agrees on what characters are allowed. See #3023. """ if not sid or not isinstance(sid, str): return False return all(c in _SAFE_SID_CHARS for c in sid) def _cleanup_stale_tmp_files() -> None: """Best-effort removal of stale ``*.tmp.*`` files from SESSION_DIR. Only files whose mtime is older than ``_STALE_TMP_AGE_SECONDS`` are removed so that in-flight writes from a long-running sibling process are not disturbed. Errors are logged and swallowed — this must never prevent startup. """ cutoff = time.time() - _STALE_TMP_AGE_SECONDS try: for p in SESSION_DIR.glob('*.tmp.*'): try: if p.stat().st_mtime < cutoff: p.unlink(missing_ok=True) logger.debug("Cleaned up stale tmp file: %s", p.name) except OSError: pass # best-effort except Exception: pass # SESSION_DIR may not exist yet; that's fine _PERSISTED_SESSION_IDS_CACHE: tuple[Path | None, int | None, frozenset[str]] = (None, None, frozenset()) def _persisted_session_ids_snapshot() -> frozenset[str]: """Return persisted session ids, caching the directory snapshot by mtime. `/api/sessions` and incremental index writes may run every few seconds. A full `SESSION_DIR.glob('*.json')` on a large session directory is expensive, and doing that scan while request threads contend on LOCK makes the sidebar look like it was designed by a committee of glaciers. Cache the listing until the directory mtime changes, and let callers take the snapshot before entering critical sections. """ global _PERSISTED_SESSION_IDS_CACHE try: dir_mtime_ns = SESSION_DIR.stat().st_mtime_ns except Exception: dir_mtime_ns = None cached_dir, cached_mtime_ns, cached_ids = _PERSISTED_SESSION_IDS_CACHE if cached_dir == SESSION_DIR and cached_mtime_ns == dir_mtime_ns: return cached_ids try: ids = frozenset( p.stem for p in SESSION_DIR.glob('*.json') if not p.name.startswith('_') ) except Exception: ids = frozenset() _PERSISTED_SESSION_IDS_CACHE = (SESSION_DIR, dir_mtime_ns, ids) return ids def _session_dir_has_persisted_session_files() -> bool: """Return True when the current session dir has at least one session JSON file.""" try: return any(not p.name.startswith('_') for p in SESSION_DIR.glob('*.json')) except Exception: return False def _rebuild_session_index_background(expected_session_dir: Path, expected_index_file: Path) -> None: global _SESSION_INDEX_REBUILD_THREAD, _SESSION_INDEX_REBUILD_THREAD_TARGET current_thread = threading.current_thread() try: with _SESSION_INDEX_REBUILD_LOCK: if SESSION_DIR != expected_session_dir or SESSION_INDEX_FILE != expected_index_file: return _write_session_index( updates=None, session_dir=expected_session_dir, session_index_file=expected_index_file, ) except Exception: logger.debug("Background session-index rebuild failed", exc_info=True) finally: with _SESSION_INDEX_REBUILD_LOCK: if _SESSION_INDEX_REBUILD_THREAD is current_thread and _SESSION_INDEX_REBUILD_THREAD_TARGET == ( expected_session_dir, expected_index_file, ): _SESSION_INDEX_REBUILD_THREAD = None _SESSION_INDEX_REBUILD_THREAD_TARGET = None def _start_session_index_rebuild_thread() -> None: """Start one background full-index rebuild if the index is missing.""" global _SESSION_INDEX_REBUILD_THREAD, _SESSION_INDEX_REBUILD_THREAD_TARGET target = (SESSION_DIR, SESSION_INDEX_FILE) with _SESSION_INDEX_REBUILD_LOCK: if SESSION_INDEX_FILE.exists(): return if ( _SESSION_INDEX_REBUILD_THREAD is not None and _SESSION_INDEX_REBUILD_THREAD.is_alive() and _SESSION_INDEX_REBUILD_THREAD_TARGET == target ): return _SESSION_INDEX_REBUILD_THREAD_TARGET = target _SESSION_INDEX_REBUILD_THREAD = threading.Thread( target=_rebuild_session_index_background, args=target, name="session-index-rebuild", daemon=True, ) _SESSION_INDEX_REBUILD_THREAD.start() def _index_entry_exists(session_id: str, in_memory_ids=None) -> bool: """Return True if an index entry still has backing state. A session can legitimately exist either as a persisted JSON file or as an in-memory Session object that has not been flushed yet. This helper is used to prune stale `_index.json` rows left behind after session-id rotation or file removal. """ if not session_id: return False if in_memory_ids is None: with LOCK: in_memory_ids = set(SESSIONS.keys()) if session_id in in_memory_ids: return True p = SESSION_DIR / f'{session_id}.json' return p.exists() def _write_session_index(updates=None, *, session_dir: Path | None = None, session_index_file: Path | None = None): """Update the session index file. When *updates* is provided (a list of Session objects whose compact entries should be refreshed), this does a targeted in-place update of the existing index — O(1) for single-session changes. When *updates* is None, a full rebuild is performed (used on startup / first call). LOCK protects only in-memory session snapshots. JSON parsing, payload construction, and disk I/O run outside LOCK so active-stream saves do not block ordinary session reads longer than necessary. The on-disk index read-modify-write is NOT unsynchronized: it stays fully serialized by ``_INDEX_WRITE_LOCK`` (held across this whole function), so narrowing LOCK cannot introduce a lost-update or index-corruption race between writers. """ session_dir = session_dir or SESSION_DIR session_index_file = session_index_file or SESSION_INDEX_FILE _tmp = session_index_file.with_suffix(f'.tmp.{os.getpid()}.{threading.current_thread().ident}') with _INDEX_WRITE_LOCK: # Lazy full-rebuild path — used when index doesn't exist yet. if updates is None or not session_index_file.exists(): _cleanup_stale_tmp_files() # best-effort sweep on startup / first call entry_map: dict[str, dict] = {} for p in session_dir.glob('*.json'): if p.name.startswith('_'): continue try: s = _load_session_from_path(p) if s: c = s.compact() sid = c.get('session_id') if sid: # Dedup by session_id: prefer entry with more messages # (handles old-format session_xxx.json files alongside # WebUI-format xxx.json with the same session_id) existing = entry_map.get(sid) if existing is None or ( c.get('message_count', 0) > existing.get('message_count', 0) ): entry_map[sid] = c except Exception: logger.debug("Failed to load session from %s", p) entries = list(entry_map.values()) existing_ids = set(entry_map.keys()) with LOCK: in_memory_entries = [ s.compact() for s in SESSIONS.values() if s.session_id not in existing_ids ] entries.extend(in_memory_entries) entries.sort(key=lambda s: s.get('updated_at', 0), reverse=True) _payload = json.dumps(entries, ensure_ascii=False, indent=2) try: with open(_tmp, 'w', encoding='utf-8') as f: f.write(_payload) f.flush() os.fsync(f.fileno()) _safe_replace(_tmp, session_index_file) except Exception: # Best-effort cleanup of stale tmp on failure try: _tmp.unlink(missing_ok=True) except Exception: pass raise return # Fast path: patch existing index with updated sessions. # This avoids loading every session file on every single save(). _fallback = False try: # Avoid N filesystem exists() checks under LOCK by collecting # on-disk IDs once before entering the critical section. on_disk_ids = _persisted_session_ids_snapshot() existing = json.loads(session_index_file.read_bytes()) if not isinstance(existing, list): raise ValueError("session index must be a list") with LOCK: in_memory_ids = set(SESSIONS.keys()) updated_map = {s.session_id: s.compact() for s in updates} existing = [ e for e in existing if (e.get('session_id') in in_memory_ids or e.get('session_id') in on_disk_ids) ] existing_ids = {e.get('session_id') for e in existing} # Add any updated entries not yet in the index. for sid, entry in updated_map.items(): if sid not in existing_ids: existing.append(entry) # Replace matching entries in-place. for i, e in enumerate(existing): sid = e.get('session_id') if sid in updated_map: existing[i] = updated_map[sid] existing.sort(key=lambda s: s.get('updated_at', 0), reverse=True) _payload = json.dumps(existing, ensure_ascii=False, indent=2) try: with open(_tmp, 'w', encoding='utf-8') as f: f.write(_payload) f.flush() os.fsync(f.fileno()) _safe_replace(_tmp, session_index_file) except Exception: try: _tmp.unlink(missing_ok=True) except Exception: pass raise except Exception: _fallback = True if _fallback: # Corrupt or missing index — fall back to full rebuild (called outside LOCK to avoid deadlock). # Propagate the resolved target so a rebuild scoped to a specific session dir # (the background rebuild thread) falls back to rebuilding THAT dir's index, # not the global SESSION_DIR (Opus advisor, stage-344 — defensive; today the # only kwargs-caller passes updates=None and never reaches the fast path). _write_session_index( updates=None, session_dir=session_dir, session_index_file=session_index_file, ) def prune_session_from_index(session_id: str) -> None: """Remove one session row from the persisted sidebar index if present.""" sid = str(session_id or "") if not sid or not SESSION_INDEX_FILE.exists(): return _tmp = SESSION_INDEX_FILE.with_suffix(f'.tmp.{os.getpid()}.{threading.current_thread().ident}') _fallback = False with _INDEX_WRITE_LOCK: try: with LOCK: existing = json.loads(SESSION_INDEX_FILE.read_bytes()) if not isinstance(existing, list): raise ValueError("session index must be a list") pruned = [e for e in existing if e.get('session_id') != sid] if len(pruned) == len(existing): return _payload = json.dumps(pruned, ensure_ascii=False, indent=2) try: with open(_tmp, 'w', encoding='utf-8') as f: f.write(_payload) f.flush() os.fsync(f.fileno()) _safe_replace(_tmp, SESSION_INDEX_FILE) except Exception: try: _tmp.unlink(missing_ok=True) except Exception: pass raise except Exception: _fallback = True if _fallback: _write_session_index(updates=None) # ---------------------------------------------------------------------------# #4985 webui zero-message orphan tombstone# ---------------------------------------------------------------------------Lines 1–500 of 8555. Load more
"""Hermes Web UI -- Session model and in-memory session store.""" import collections import copy import datetime import hashlib import inspect import json import logging import os import re import threading import time import uuid from contextlib import closing from pathlib import Path import api.config as _cfg from api.compression_anchor import is_context_compression_marker from api.config import ( SESSION_DIR, SESSION_INDEX_FILE, SESSIONS, SESSIONS_MAX, LOCK, STREAMS, STREAMS_LOCK, DEFAULT_WORKSPACE, DEFAULT_MODEL, PROJECTS_FILE, HOME, get_effective_default_model, _get_session_agent_lock, ) from api.workspace import get_last_workspace from api.usage import prompt_cache_hit_percent from api.agent_sessions import ( _is_continuation_session, is_cli_session_row, normalize_agent_session_source, open_state_db_readonly, read_importable_agent_session_rows, read_session_lineage_metadata, ) logger = logging.getLogger(__name__) CLI_VISIBLE_SESSION_LIMIT = 20 # How many messageful cron sessions to surface in the project-chip layer. # Needs to exceed CLI_VISIBLE_SESSION_LIMIT so older cron runs stay # addressable even when many newer non-cron sessions dominate the default # sidebar window (#3172). CRON_PROJECT_CHIP_LIMIT = 200 WEBHOOK_PROJECT_CHIP_LIMIT = 200 _CLI_SESSIONS_CACHE_TTL_SECONDS = 5.0 # While a turn is actively streaming, hold the CLI/cron projection longer than # one poll interval (mirrors the route-level #4808 hold-down). The frontend # polls /api/sessions every ~5s during a stream; without a wider window the # CLI cache key advances on every streamed message row (see below) and the # expensive state.db CLI/cron projection is re-run on every poll. (#4842) _CLI_SESSIONS_CACHE_STREAMING_TTL_SECONDS = 30.0 _CLI_SESSIONS_CACHE_LOCK = threading.Lock() _CLI_SESSIONS_CACHE_INFLIGHT: "dict[tuple, threading.Event]" = {} _CLI_SESSIONS_CACHE_INVALIDATION_VERSION = 0 _CLI_SESSIONS_CACHE = {} _CLI_SESSIONS_CACHE_WAIT_SECONDS = 0.25 # Event waits that keep stale rows visible while a rebuild is in flight. _CLI_SESSIONS_CACHE_STALE_WAIT_SECONDS = 0.10 # Per-file parse cache for Claude Code JSONL transcripts (#4718/#4662 phase 4). # ``~/.claude/projects`` is a GLOBAL, profile-independent directory, but the # sidebar re-derives every Claude Code row from scratch on each /api/sessions # build — fully re-reading and JSON-parsing up to CLAUDE_CODE_MAX_FILES # transcripts line-by-line (hundreds of MB) just to recover a title + message # count. That parse dominates the cold sidebar build (~650-1000ms measured on a # 200-file / ~130MB tree) and it repeats on every profile switch, on the 5s # CLI-cache expiry, and on every sidebar poll, because the higher CLI cache is # keyed per active profile while the underlying transcripts never change between # switches. This cache memoizes the EXPENSIVE per-file parse result keyed by the # file's (path, mtime_ns, size, ctime_ns); a warm sidebar build then re-stats the # files (~4ms for 200) instead of re-parsing them. Any external edit/append to a # transcript changes mtime_ns/size/ctime_ns and transparently invalidates just # that one file's entry. Bounded so a pathological projects tree can't grow it unbounded. _CLAUDE_CODE_PARSE_CACHE_LOCK = threading.Lock() _CLAUDE_CODE_PARSE_CACHE: "collections.OrderedDict[tuple, tuple]" = collections.OrderedDict() _CLAUDE_CODE_PARSE_CACHE_MAX = 1000 # Per-file cache for the UI-owned sidecar metadata (title + archived) that the # state.db sidebar projection overlays onto each CLI/cron row (#4842). The # projection calls _state_projection_sidecar_metadata() once per row in BOTH # the main visible pass AND the higher-capped (CRON_PROJECT_CHIP_LIMIT=200) # cron-only second pass, and each call was an uncached open() + 64KB prefix # read + a pure-Python JSON-key scan. On a cron-heavy profile that is up to # ~200 sidecar file reads per /api/sessions build — and because the enclosing # _CLI_SESSIONS_CACHE is keyed on a state.db content fingerprint that advances # on every streamed message row, that whole scan was re-paid on essentially # every 5s poll during a live turn (the "100% CPU / multi-second get_cli_sessions" # in #4842/#4808/#4672). This memoizes the parse result keyed by the sidecar's # (path, mtime_ns, size, ctime_ns) stat signature: a warm projection re-stats # each file (~1 stat) instead of re-reading+parsing it, while any genuine # rename/archive/edit bumps the signature and transparently invalidates just # that one entry. Bounded so a pathological session store can't grow it without # limit. Mirrors the Claude Code parse cache (#4718). _SIDECAR_METADATA_CACHE_LOCK = threading.Lock() _SIDECAR_METADATA_CACHE: "collections.OrderedDict[tuple, dict]" = collections.OrderedDict() _SIDECAR_METADATA_CACHE_MAX = 2000 # #5854: authoritative facts for a LEGACY (pre-#5854) sidecar whose scenes # serialize before `messages`, so the cheap metadata-prefix read can't recover # its message_count or scene fingerprint. Without this, an unchanged legacy # large-scene session would full-parse on every poll (recreating the #4633 # churn for legacy files) and could not be LRU-evicted. Populated once per file # from a full Session.load(); keyed by the sidecar's stat signature so any edit # invalidates it. Bounded. Value: {"message_count": int, "scene_index": dict}. _LEGACY_SIDECAR_FACTS_LOCK = threading.Lock() _LEGACY_SIDECAR_FACTS: "collections.OrderedDict[tuple, dict]" = collections.OrderedDict() _LEGACY_SIDECAR_FACTS_MAX = 2000 # --------------------------------------------------------------------------- # Stale temp-file cleanup # --------------------------------------------------------------------------- # Both Session.save() and _write_session_index() use the atomic-write pattern: # write to <path>.tmp.<pid>.<tid> → os.replace() to final path # If the process crashes between write and replace the .tmp file is left # behind. Because the name embeds pid + tid, leftover files can never be # reused by a different process/thread, so they are safe to remove on the # next startup. _cleanup_stale_tmp_files() is called from the full-rebuild # path of _write_session_index (i.e. at first index access / startup) and # removes any *.tmp.* file whose mtime is older than one hour. # --------------------------------------------------------------------------- _STALE_TMP_AGE_SECONDS = 3600 # 1 hour # --------------------------------------------------------------------------- # Windows-safe os.replace() with retry # --------------------------------------------------------------------------- # On Windows, os.replace() raises WinError 5 (ERROR_ACCESS_DENIED) when the # target file is momentarily locked by another process (antivirus scanner, # browser polling the session JSON, etc.). This helper retries with # exponential backoff on PermissionError, which is the Python exception # mapped from WinError 5. On non-Windows platforms it is a thin wrapper # (one attempt, no delay). # --------------------------------------------------------------------------- _WINDOWS_REPLACE_MAX_RETRIES = 5 _WINDOWS_REPLACE_INITIAL_DELAY = 0.05 # 50 ms def _safe_replace(src: Path, dst: Path) -> None: """Atomic replace with retries on Windows file-locking errors.""" if os.name != 'nt': os.replace(src, dst) return delay = _WINDOWS_REPLACE_INITIAL_DELAY for attempt in range(_WINDOWS_REPLACE_MAX_RETRIES): try: os.replace(src, dst) return except PermissionError: if attempt == _WINDOWS_REPLACE_MAX_RETRIES - 1: raise time.sleep(delay) delay *= 2 # 50 -> 100 -> 200 -> 400 -> 800 ms # Serializes index writers so concurrent Session.save() calls cannot race on # stale baselines while still allowing LOCK to be released before disk I/O. _INDEX_WRITE_LOCK = threading.RLock() _SESSION_INDEX_REBUILD_LOCK = threading.Lock() _SESSION_INDEX_REBUILD_THREAD = None _SESSION_INDEX_REBUILD_THREAD_TARGET: tuple[Path, Path] | None = None # Serializes ``_record_webui_zero_message_orphan_tombstone`` / # ``_clear_webui_zero_message_orphan_tombstone`` so two concurrent sidebar # polls (or a poll racing ``Session.save`` / ``new_session`` / # ``import_cli_session``) cannot lose each other's load-modify-write/unlink. # Without this lock each operation rewrites the entire tombstone file from # scratch, so a concurrent recorder and clearer can land last-writer-wins and # silently drop each other's update — defeating the self-healing invariant # that ``Session.save`` clears the tombstone the same poll that re-prunes # would otherwise re-add the row for. ``threading.Lock`` is sufficient (the # WebUI sidebar polling path is single-process) but must wrap the WHOLE # load-modify-write/unlink sequence in both helpers. _WEBUI_ZERO_MESSAGE_ORPHAN_TOMBSTONE_LOCK = threading.Lock() _WEBUI_DELETED_SESSION_TOMBSTONE_LOCK = threading.Lock() # Path-safety contract for session IDs. Accept alphanumerics, underscore, and # hyphen so API/gateway-issued ids (``api-*``, ``reachy-voice-*``) round-trip # through filesystem load/save/delete/worktree paths without traversal risk. # Dots and slashes are rejected so the id can never name a parent directory # or hide an unexpected extension. _SAFE_SID_CHARS = frozenset( '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-' ) def is_safe_session_id(sid) -> bool: """Return True iff ``sid`` is a non-empty path-safe session id. Centralizes the validation previously duplicated across ``Session.load``, ``Session.load_metadata_only``, ``_repair_stale_pending``, ``/api/session/worktree/remove``, and ``/api/session/delete`` so every call site agrees on what characters are allowed. See #3023. """ if not sid or not isinstance(sid, str): return False return all(c in _SAFE_SID_CHARS for c in sid) def _cleanup_stale_tmp_files() -> None: """Best-effort removal of stale ``*.tmp.*`` files from SESSION_DIR. Only files whose mtime is older than ``_STALE_TMP_AGE_SECONDS`` are removed so that in-flight writes from a long-running sibling process are not disturbed. Errors are logged and swallowed — this must never prevent startup. """ cutoff = time.time() - _STALE_TMP_AGE_SECONDS try: for p in SESSION_DIR.glob('*.tmp.*'): try: if p.stat().st_mtime < cutoff: p.unlink(missing_ok=True) logger.debug("Cleaned up stale tmp file: %s", p.name) except OSError: pass # best-effort except Exception: pass # SESSION_DIR may not exist yet; that's fine _PERSISTED_SESSION_IDS_CACHE: tuple[Path | None, int | None, frozenset[str]] = (None, None, frozenset()) def _persisted_session_ids_snapshot() -> frozenset[str]: """Return persisted session ids, caching the directory snapshot by mtime. `/api/sessions` and incremental index writes may run every few seconds. A full `SESSION_DIR.glob('*.json')` on a large session directory is expensive, and doing that scan while request threads contend on LOCK makes the sidebar look like it was designed by a committee of glaciers. Cache the listing until the directory mtime changes, and let callers take the snapshot before entering critical sections. """ global _PERSISTED_SESSION_IDS_CACHE try: dir_mtime_ns = SESSION_DIR.stat().st_mtime_ns except Exception: dir_mtime_ns = None cached_dir, cached_mtime_ns, cached_ids = _PERSISTED_SESSION_IDS_CACHE if cached_dir == SESSION_DIR and cached_mtime_ns == dir_mtime_ns: return cached_ids try: ids = frozenset( p.stem for p in SESSION_DIR.glob('*.json') if not p.name.startswith('_') ) except Exception: ids = frozenset() _PERSISTED_SESSION_IDS_CACHE = (SESSION_DIR, dir_mtime_ns, ids) return ids def _session_dir_has_persisted_session_files() -> bool: """Return True when the current session dir has at least one session JSON file.""" try: return any(not p.name.startswith('_') for p in SESSION_DIR.glob('*.json')) except Exception: return False def _rebuild_session_index_background(expected_session_dir: Path, expected_index_file: Path) -> None: global _SESSION_INDEX_REBUILD_THREAD, _SESSION_INDEX_REBUILD_THREAD_TARGET current_thread = threading.current_thread() try: with _SESSION_INDEX_REBUILD_LOCK: if SESSION_DIR != expected_session_dir or SESSION_INDEX_FILE != expected_index_file: return _write_session_index( updates=None, session_dir=expected_session_dir, session_index_file=expected_index_file, ) except Exception: logger.debug("Background session-index rebuild failed", exc_info=True) finally: with _SESSION_INDEX_REBUILD_LOCK: if _SESSION_INDEX_REBUILD_THREAD is current_thread and _SESSION_INDEX_REBUILD_THREAD_TARGET == ( expected_session_dir, expected_index_file, ): _SESSION_INDEX_REBUILD_THREAD = None _SESSION_INDEX_REBUILD_THREAD_TARGET = None def _start_session_index_rebuild_thread() -> None: """Start one background full-index rebuild if the index is missing.""" global _SESSION_INDEX_REBUILD_THREAD, _SESSION_INDEX_REBUILD_THREAD_TARGET target = (SESSION_DIR, SESSION_INDEX_FILE) with _SESSION_INDEX_REBUILD_LOCK: if SESSION_INDEX_FILE.exists(): return if ( _SESSION_INDEX_REBUILD_THREAD is not None and _SESSION_INDEX_REBUILD_THREAD.is_alive() and _SESSION_INDEX_REBUILD_THREAD_TARGET == target ): return _SESSION_INDEX_REBUILD_THREAD_TARGET = target _SESSION_INDEX_REBUILD_THREAD = threading.Thread( target=_rebuild_session_index_background, args=target, name="session-index-rebuild", daemon=True, ) _SESSION_INDEX_REBUILD_THREAD.start() def _index_entry_exists(session_id: str, in_memory_ids=None) -> bool: """Return True if an index entry still has backing state. A session can legitimately exist either as a persisted JSON file or as an in-memory Session object that has not been flushed yet. This helper is used to prune stale `_index.json` rows left behind after session-id rotation or file removal. """ if not session_id: return False if in_memory_ids is None: with LOCK: in_memory_ids = set(SESSIONS.keys()) if session_id in in_memory_ids: return True p = SESSION_DIR / f'{session_id}.json' return p.exists() def _write_session_index(updates=None, *, session_dir: Path | None = None, session_index_file: Path | None = None): """Update the session index file. When *updates* is provided (a list of Session objects whose compact entries should be refreshed), this does a targeted in-place update of the existing index — O(1) for single-session changes. When *updates* is None, a full rebuild is performed (used on startup / first call). LOCK protects only in-memory session snapshots. JSON parsing, payload construction, and disk I/O run outside LOCK so active-stream saves do not block ordinary session reads longer than necessary. The on-disk index read-modify-write is NOT unsynchronized: it stays fully serialized by ``_INDEX_WRITE_LOCK`` (held across this whole function), so narrowing LOCK cannot introduce a lost-update or index-corruption race between writers. """ session_dir = session_dir or SESSION_DIR session_index_file = session_index_file or SESSION_INDEX_FILE _tmp = session_index_file.with_suffix(f'.tmp.{os.getpid()}.{threading.current_thread().ident}') with _INDEX_WRITE_LOCK: # Lazy full-rebuild path — used when index doesn't exist yet. if updates is None or not session_index_file.exists(): _cleanup_stale_tmp_files() # best-effort sweep on startup / first call entry_map: dict[str, dict] = {} for p in session_dir.glob('*.json'): if p.name.startswith('_'): continue try: s = _load_session_from_path(p) if s: c = s.compact() sid = c.get('session_id') if sid: # Dedup by session_id: prefer entry with more messages # (handles old-format session_xxx.json files alongside # WebUI-format xxx.json with the same session_id) existing = entry_map.get(sid) if existing is None or ( c.get('message_count', 0) > existing.get('message_count', 0) ): entry_map[sid] = c except Exception: logger.debug("Failed to load session from %s", p) entries = list(entry_map.values()) existing_ids = set(entry_map.keys()) with LOCK: in_memory_entries = [ s.compact() for s in SESSIONS.values() if s.session_id not in existing_ids ] entries.extend(in_memory_entries) entries.sort(key=lambda s: s.get('updated_at', 0), reverse=True) _payload = json.dumps(entries, ensure_ascii=False, indent=2) try: with open(_tmp, 'w', encoding='utf-8') as f: f.write(_payload) f.flush() os.fsync(f.fileno()) _safe_replace(_tmp, session_index_file) except Exception: # Best-effort cleanup of stale tmp on failure try: _tmp.unlink(missing_ok=True) except Exception: pass raise return # Fast path: patch existing index with updated sessions. # This avoids loading every session file on every single save(). _fallback = False try: # Avoid N filesystem exists() checks under LOCK by collecting # on-disk IDs once before entering the critical section. on_disk_ids = _persisted_session_ids_snapshot() existing = json.loads(session_index_file.read_bytes()) if not isinstance(existing, list): raise ValueError("session index must be a list") with LOCK: in_memory_ids = set(SESSIONS.keys()) updated_map = {s.session_id: s.compact() for s in updates} existing = [ e for e in existing if (e.get('session_id') in in_memory_ids or e.get('session_id') in on_disk_ids) ] existing_ids = {e.get('session_id') for e in existing} # Add any updated entries not yet in the index. for sid, entry in updated_map.items(): if sid not in existing_ids: existing.append(entry) # Replace matching entries in-place. for i, e in enumerate(existing): sid = e.get('session_id') if sid in updated_map: existing[i] = updated_map[sid] existing.sort(key=lambda s: s.get('updated_at', 0), reverse=True) _payload = json.dumps(existing, ensure_ascii=False, indent=2) try: with open(_tmp, 'w', encoding='utf-8') as f: f.write(_payload) f.flush() os.fsync(f.fileno()) _safe_replace(_tmp, session_index_file) except Exception: try: _tmp.unlink(missing_ok=True) except Exception: pass raise except Exception: _fallback = True if _fallback: # Corrupt or missing index — fall back to full rebuild (called outside LOCK to avoid deadlock). # Propagate the resolved target so a rebuild scoped to a specific session dir # (the background rebuild thread) falls back to rebuilding THAT dir's index, # not the global SESSION_DIR (Opus advisor, stage-344 — defensive; today the # only kwargs-caller passes updates=None and never reaches the fast path). _write_session_index( updates=None, session_dir=session_dir, session_index_file=session_index_file, ) def prune_session_from_index(session_id: str) -> None: """Remove one session row from the persisted sidebar index if present.""" sid = str(session_id or "") if not sid or not SESSION_INDEX_FILE.exists(): return _tmp = SESSION_INDEX_FILE.with_suffix(f'.tmp.{os.getpid()}.{threading.current_thread().ident}') _fallback = False with _INDEX_WRITE_LOCK: try: with LOCK: existing = json.loads(SESSION_INDEX_FILE.read_bytes()) if not isinstance(existing, list): raise ValueError("session index must be a list") pruned = [e for e in existing if e.get('session_id') != sid] if len(pruned) == len(existing): return _payload = json.dumps(pruned, ensure_ascii=False, indent=2) try: with open(_tmp, 'w', encoding='utf-8') as f: f.write(_payload) f.flush() os.fsync(f.fileno()) _safe_replace(_tmp, SESSION_INDEX_FILE) except Exception: try: _tmp.unlink(missing_ok=True) except Exception: pass raise except Exception: _fallback = True if _fallback: _write_session_index(updates=None) # --------------------------------------------------------------------------- # #4985 webui zero-message orphan tombstone # ---------------------------------------------------------------------------