From 1458a39260b80a22915a479f1e5b7d595293e181 Mon Sep 17 00:00:00 2001 From: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> Date: Thu, 2 Jul 2026 14:28:20 -0700 Subject: [PATCH] fix(dashboard): offload session read endpoints Offloads the remaining dashboard SessionDB read sites from async handlers through sync helpers and adds Semaphore(2) bounding for heavyweight reads. Adds thread-assert coverage plus an AST-aware regression lint with bare/lambda/partial/inner-def/to_thread self-tests. Verified: scripts/run_tests.sh tests/test_web_server_sessiondb_eventloop.py tests/test_web_server.py tests/test_hermes_state.py -q; scripts/run_tests.sh tests/hermes_cli/test_dashboard_auth_status_endpoint.py -q. --- hermes_cli/web_server.py | 503 +++++++++++-------- tests/test_web_server_sessiondb_eventloop.py | 433 ++++++++++++++++ 2 files changed, 730 insertions(+), 206 deletions(-) create mode 100644 tests/test_web_server_sessiondb_eventloop.py diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 3378c0006b116..c447af1101a5d 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -1938,6 +1938,27 @@ async def _blocking_io(fn, *args): return await loop.run_in_executor(None, fn, *args) +_SESSION_DB_HEAVY_READ_SEMAPHORES: Dict[int, asyncio.Semaphore] = {} + + +def _session_db_heavy_read_semaphore() -> asyncio.Semaphore: + loop = asyncio.get_running_loop() + key = id(loop) + semaphore = _SESSION_DB_HEAVY_READ_SEMAPHORES.get(key) + if semaphore is None: + semaphore = asyncio.Semaphore(2) + _SESSION_DB_HEAVY_READ_SEMAPHORES[key] = semaphore + return semaphore + + +async def _session_db_read(fn, *args, heavy: bool = False): + """Run a blocking SessionDB read off-loop, bounding heavyweight scans.""" + if heavy: + async with _session_db_heavy_read_semaphore(): + return await _blocking_io(fn, *args) + return await _blocking_io(fn, *args) + + def _git_path(path: str) -> str: return str(_fs_path(path)) @@ -2178,18 +2199,12 @@ async def get_status(profile: Optional[str] = None): active_sessions = 0 try: - from hermes_state import SessionDB - db = SessionDB() - try: - sessions = db.list_sessions_rich(limit=50) - now = time.time() - active_sessions = sum( - 1 for s in sessions - if s.get("ended_at") is None - and (now - s.get("last_active", s.get("started_at", 0))) < 300 - ) - finally: - db.close() + active_sessions = await _session_db_read( + _read_active_session_count, + requested_profile + if requested_profile and requested_profile.lower() != "current" + else None, + ) except Exception: pass @@ -2282,6 +2297,21 @@ async def get_status(profile: Optional[str] = None): status_scope.__exit__(*sys.exc_info()) +def _read_active_session_count(profile: Optional[str] = None) -> int: + """Blocking active-session status read; callers must offload it.""" + db = _open_session_db_for_profile(profile) + try: + sessions = db.list_sessions_rich(limit=50) + now = time.time() + return sum( + 1 for s in sessions + if s.get("ended_at") is None + and (now - s.get("last_active", s.get("started_at", 0))) < 300 + ) + finally: + db.close() + + _WINDOWS_11_MIN_BUILD = 22000 @@ -3456,50 +3486,40 @@ async def get_sessions( if profile: profile_name, _ = _cron_profile_home(profile) try: - db = _open_session_db_for_profile(profile) - try: - min_message_count = max(0, min_messages) - archived_only = archived == "only" - include_archived = archived == "include" - # Optional source scoping: ``source`` includes a single class, - # ``exclude_sources`` (comma-separated) drops classes. The desktop - # uses these to split recents (exclude=cron) from the cron-jobs - # section (source=cron) into two independent lists. - exclude_list = [s for s in (exclude_sources or "").split(",") if s.strip()] - sessions = db.list_sessions_rich( - source=source or None, - exclude_sources=exclude_list or None, - cwd_prefix=(cwd_prefix or None), - limit=limit, - offset=offset, - min_message_count=min_message_count, - include_archived=include_archived, - archived_only=archived_only, - order_by_last_active=order == "recent", - ) - total = db.session_count( - source=source or None, - cwd_prefix=(cwd_prefix or None), - exclude_sources=exclude_list or None, - min_message_count=min_message_count, - include_archived=include_archived, - archived_only=archived_only, - exclude_children=True, + min_message_count = max(0, min_messages) + archived_only = archived == "only" + include_archived = archived == "include" + # Optional source scoping: ``source`` includes a single class, + # ``exclude_sources`` (comma-separated) drops classes. The desktop + # uses these to split recents (exclude=cron) from the cron-jobs + # section (source=cron) into two independent lists. + exclude_list = [s for s in (exclude_sources or "").split(",") if s.strip()] + sessions, total = await _session_db_read( + _read_sessions_page, + profile, + source or None, + exclude_list or None, + cwd_prefix or None, + limit, + offset, + min_message_count, + include_archived, + archived_only, + order == "recent", + heavy=int(limit or 0) >= 1000, + ) + now = time.time() + for s in sessions: + s["is_active"] = ( + s.get("ended_at") is None + and (now - s.get("last_active", s.get("started_at", 0))) < 300 ) - now = time.time() - for s in sessions: - s["is_active"] = ( - s.get("ended_at") is None - and (now - s.get("last_active", s.get("started_at", 0))) < 300 - ) - if profile_name: - s["profile"] = profile_name - s["is_default_profile"] = profile_name == "default" - # SQLite stores the flag as 0/1; expose a real JSON boolean. - s["archived"] = bool(s.get("archived")) - return {"sessions": sessions, "total": total, "limit": limit, "offset": offset} - finally: - db.close() + if profile_name: + s["profile"] = profile_name + s["is_default_profile"] = profile_name == "default" + # SQLite stores the flag as 0/1; expose a real JSON boolean. + s["archived"] = bool(s.get("archived")) + return {"sessions": sessions, "total": total, "limit": limit, "offset": offset} except HTTPException: raise except Exception: @@ -3507,6 +3527,46 @@ async def get_sessions( raise HTTPException(status_code=500, detail="Internal server error") +def _read_sessions_page( + profile: Optional[str], + source: Optional[str], + exclude_sources: Optional[List[str]], + cwd_prefix: Optional[str], + limit: int, + offset: int, + min_message_count: int, + include_archived: bool, + archived_only: bool, + order_by_last_active: bool, +): + """Blocking sessions-page read; callers must offload it.""" + db = _open_session_db_for_profile(profile) + try: + sessions = db.list_sessions_rich( + source=source, + exclude_sources=exclude_sources, + cwd_prefix=cwd_prefix, + limit=limit, + offset=offset, + min_message_count=min_message_count, + include_archived=include_archived, + archived_only=archived_only, + order_by_last_active=order_by_last_active, + ) + total = db.session_count( + source=source, + cwd_prefix=cwd_prefix, + exclude_sources=exclude_sources, + min_message_count=min_message_count, + include_archived=include_archived, + archived_only=archived_only, + exclude_children=True, + ) + return sessions, total + finally: + db.close() + + @app.get("/api/profiles/sessions") def get_profiles_sessions( limit: int = 20, @@ -3641,153 +3701,162 @@ async def search_sessions(q: str = "", limit: int = 20, profile: Optional[str] = if not q or not q.strip(): return {"results": []} try: - db = _open_session_db_for_profile(profile) - try: - safe_limit = max(1, min(int(limit or 20), 100)) - - # Walk parent_session_id to the compression root, memoized so a - # chain of compression segments only costs one walk. We deliberately - # stop at branch/delegate edges: those sessions may diverge from the - # parent and should remain searchable on their own. - root_cache: dict = {} - - def compression_root(session_id: str) -> str: - if not session_id: - return session_id - if session_id in root_cache: - return root_cache[session_id] - chain = [] - cur = session_id - visited = set() - root = session_id - while cur and cur not in visited: - visited.add(cur) - chain.append(cur) - if cur in root_cache: - root = root_cache[cur] - break - try: - s = db.get_session(cur) - except Exception: - s = None - if not s: - root = cur - break - parent = s.get("parent_session_id") if isinstance(s, dict) else None - if not parent: - root = cur - break - try: - parent_session = db.get_session(parent) - except Exception: - parent_session = None - if not parent_session: - root = cur - break - parent_ended_at = parent_session.get("ended_at") - started_at = s.get("started_at") - is_compression_edge = ( - parent_session.get("end_reason") == "compression" - and parent_ended_at is not None - and started_at is not None - and started_at >= parent_ended_at - ) - if not is_compression_edge: - root = cur - break - cur = parent - for node in chain: - root_cache[node] = root - return root + safe_limit = max(1, min(int(limit or 20), 100)) + return await _session_db_read( + _read_session_search, + profile, + q, + safe_limit, + heavy=True, + ) + except HTTPException: + raise + except Exception: + _log.exception("GET /api/sessions/search failed") + raise HTTPException(status_code=500, detail="Search failed") - tip_cache: dict = {} - def lineage_tip(root_id: str) -> str: - if root_id in tip_cache: - return tip_cache[root_id] - tip = root_id +def _read_session_search(profile: Optional[str], q: str, safe_limit: int): + """Blocking search-session read; callers must offload it.""" + db = _open_session_db_for_profile(profile) + try: + # Walk parent_session_id to the compression root, memoized so a + # chain of compression segments only costs one walk. We deliberately + # stop at branch/delegate edges: those sessions may diverge from the + # parent and should remain searchable on their own. + root_cache: dict = {} + + def compression_root(session_id: str) -> str: + if not session_id: + return session_id + if session_id in root_cache: + return root_cache[session_id] + chain = [] + cur = session_id + visited = set() + root = session_id + while cur and cur not in visited: + visited.add(cur) + chain.append(cur) + if cur in root_cache: + root = root_cache[cur] + break try: - resolved = db.get_compression_tip(root_id) - if resolved: - tip = resolved + s = db.get_session(cur) except Exception: - pass - tip_cache[root_id] = tip - return tip - - # Both ID matches and content matches share one keyspace, keyed by - # compression lineage root, so an id-hit and a content-hit on the - # same logical conversation collapse to a single result. The first - # hit for a lineage wins; ID matches run first and take priority. - seen: dict = {} - - def add_lineage_result(raw_sid: str, payload: dict) -> None: - if not raw_sid: - return - root = compression_root(raw_sid) - if root in seen or len(seen) >= safe_limit: - return - payload = dict(payload) - payload["session_id"] = lineage_tip(root) - payload["lineage_root"] = root - seen[root] = payload - - # Direct ID matches first: users often paste a session id from CLI, - # logs, or another Hermes surface. FTS can't find those unless the - # id happens to appear in message text. search_sessions_by_id is - # SQL-bounded, so this stays cheap even with thousands of sessions. - for row in db.search_sessions_by_id(q, limit=safe_limit, include_archived=True): - sid = row.get("id") - preview = (row.get("preview") or "").strip() - snippet = preview or f"Session ID: {sid}" - add_lineage_result( - sid, - { - "snippet": snippet, - "role": None, - "source": row.get("source"), - "model": row.get("model"), - "session_started": row.get("started_at"), - }, - ) - - # Auto-add prefix wildcards so partial words match - # e.g. "nimb" → "nimb*" matches "nimby" - # Preserve quoted phrases and existing wildcards as-is - import re - terms = [] - for token in re.findall(r'"[^"]*"|\S+', q.strip()): - if token.startswith('"') or token.endswith("*"): - terms.append(token) - else: - terms.append(token + "*") - prefix_query = " ".join(terms) - # Over-fetch so lineage dedup can still surface `limit` distinct - # conversations even when several hits collapse onto one root. - fetch_limit = max(safe_limit * 5, 50) - matches = db.search_messages(query=prefix_query, limit=fetch_limit) - - for m in matches: - if len(seen) >= safe_limit: + s = None + if not s: + root = cur break - add_lineage_result( - m["session_id"], - { - "snippet": m.get("snippet", ""), - "role": m.get("role"), - "source": m.get("source"), - "model": m.get("model"), - "session_started": m.get("session_started"), - }, + parent = s.get("parent_session_id") if isinstance(s, dict) else None + if not parent: + root = cur + break + try: + parent_session = db.get_session(parent) + except Exception: + parent_session = None + if not parent_session: + root = cur + break + parent_ended_at = parent_session.get("ended_at") + started_at = s.get("started_at") + is_compression_edge = ( + parent_session.get("end_reason") == "compression" + and parent_ended_at is not None + and started_at is not None + and started_at >= parent_ended_at ) - return {"results": list(seen.values())} - finally: - db.close() - except HTTPException: - raise - except Exception: - _log.exception("GET /api/sessions/search failed") - raise HTTPException(status_code=500, detail="Search failed") + if not is_compression_edge: + root = cur + break + cur = parent + for node in chain: + root_cache[node] = root + return root + + tip_cache: dict = {} + + def lineage_tip(root_id: str) -> str: + if root_id in tip_cache: + return tip_cache[root_id] + tip = root_id + try: + resolved = db.get_compression_tip(root_id) + if resolved: + tip = resolved + except Exception: + pass + tip_cache[root_id] = tip + return tip + + # Both ID matches and content matches share one keyspace, keyed by + # compression lineage root, so an id-hit and a content-hit on the + # same logical conversation collapse to a single result. The first + # hit for a lineage wins; ID matches run first and take priority. + seen: dict = {} + + def add_lineage_result(raw_sid: str, payload: dict) -> None: + if not raw_sid: + return + root = compression_root(raw_sid) + if root in seen or len(seen) >= safe_limit: + return + payload = dict(payload) + payload["session_id"] = lineage_tip(root) + payload["lineage_root"] = root + seen[root] = payload + + # Direct ID matches first: users often paste a session id from CLI, + # logs, or another Hermes surface. FTS can't find those unless the + # id happens to appear in message text. search_sessions_by_id is + # SQL-bounded, so this stays cheap even with thousands of sessions. + for row in db.search_sessions_by_id(q, limit=safe_limit, include_archived=True): + sid = row.get("id") + preview = (row.get("preview") or "").strip() + snippet = preview or f"Session ID: {sid}" + add_lineage_result( + sid, + { + "snippet": snippet, + "role": None, + "source": row.get("source"), + "model": row.get("model"), + "session_started": row.get("started_at"), + }, + ) + + # Auto-add prefix wildcards so partial words match + # e.g. "nimb" → "nimb*" matches "nimby" + # Preserve quoted phrases and existing wildcards as-is + terms = [] + for token in re.findall(r'"[^"]*"|\S+', q.strip()): + if token.startswith('"') or token.endswith("*"): + terms.append(token) + else: + terms.append(token + "*") + prefix_query = " ".join(terms) + # Over-fetch so lineage dedup can still surface `limit` distinct + # conversations even when several hits collapse onto one root. + fetch_limit = max(safe_limit * 5, 50) + matches = db.search_messages(query=prefix_query, limit=fetch_limit) + + for m in matches: + if len(seen) >= safe_limit: + break + add_lineage_result( + m["session_id"], + { + "snippet": m.get("snippet", ""), + "role": m.get("role"), + "source": m.get("source"), + "model": m.get("model"), + "session_started": m.get("session_started"), + }, + ) + return {"results": list(seen.values())} + finally: + db.close() def _normalize_config_for_web(config: Dict[str, Any]) -> Dict[str, Any]: @@ -8051,14 +8120,20 @@ def _open_session_db_for_profile(profile: Optional[str]): @app.get("/api/sessions/{session_id}") async def get_session_detail(session_id: str, profile: Optional[str] = None): + session = await _session_db_read(_read_session_detail, session_id, profile) + if profile: + session["profile"] = _cron_profile_home(profile)[0] + return session + + +def _read_session_detail(session_id: str, profile: Optional[str] = None): + """Blocking session-detail read; callers must offload it.""" db = _open_session_db_for_profile(profile) try: sid = db.resolve_session_id(session_id) session = db.get_session(sid) if sid else None if not session: raise HTTPException(status_code=404, detail="Session not found") - if profile: - session["profile"] = _cron_profile_home(profile)[0] return session finally: db.close() @@ -8132,27 +8207,43 @@ async def rename_session_endpoint(session_id: str, body: SessionRename): restores the session. Either field may be omitted. ``profile`` targets another profile's session. """ - db = _open_session_db_for_profile(body.profile) + return await _session_db_read( + _rename_session_record, + session_id, + body.profile, + body.title, + body.archived, + ) + + +def _rename_session_record( + session_id: str, + profile: Optional[str], + title: Optional[str], + archived: Optional[bool], +): + """Blocking session update/read; callers must offload it.""" + db = _open_session_db_for_profile(profile) try: sid = db.resolve_session_id(session_id) if not sid: raise HTTPException(status_code=404, detail="Session not found") - if body.title is None and body.archived is None: + if title is None and archived is None: raise HTTPException( status_code=400, detail="Nothing to update; provide 'title' and/or 'archived'.", ) - if body.title is not None: + if title is not None: try: - db.set_session_title(sid, body.title or "") + db.set_session_title(sid, title or "") except ValueError as e: # Title too long, invalid characters, or already in use. raise HTTPException(status_code=400, detail=str(e)) - if body.archived is not None: - db.set_session_archived(sid, body.archived) + if archived is not None: + db.set_session_archived(sid, archived) result = {"ok": True, "title": db.get_session_title(sid) or ""} - if body.archived is not None: - result["archived"] = bool(body.archived) + if archived is not None: + result["archived"] = bool(archived) return result finally: db.close() diff --git a/tests/test_web_server_sessiondb_eventloop.py b/tests/test_web_server_sessiondb_eventloop.py new file mode 100644 index 0000000000000..dc7c58ee1ce4b --- /dev/null +++ b/tests/test_web_server_sessiondb_eventloop.py @@ -0,0 +1,433 @@ +import ast +import asyncio +import threading +import time +import textwrap +from pathlib import Path + +import hermes_state + +from hermes_cli import web_server + + +SESSIONDB_READ_METHODS = { + "get_session", + "get_session_title", + "list_sessions_rich", + "message_count", + "search_messages", + "search_sessions_by_id", + "session_count", + "session_counts_by_source", +} + +OFFLOAD_CALL_NAMES = { + "_blocking_io", + "_session_db_read", + "run_in_executor", + "to_thread", +} + +# Pass-4 affinity enumeration record for the PRD's nine SessionDB read sites, +# plus the paired ``get_sessions`` total count that shares that offloaded page +# helper. Each site opens a request-local SessionDB inside a synchronous helper +# that the async route awaits via the executor. No site reuses a loop-thread +# connection. +PRD_NINE_SITE_AFFINITY_RECORD = { + "get_status:list_sessions_rich(limit=50)": "_read_active_session_count -> _open_session_db_for_profile", + "get_sessions:list_sessions_rich(page)": "_read_sessions_page -> _open_session_db_for_profile", + "get_sessions:session_count(total)": "_read_sessions_page -> _open_session_db_for_profile", + "search_sessions:search_sessions_by_id": "_read_session_search -> _open_session_db_for_profile", + "search_sessions:get_session(compression_root/current)": "_read_session_search -> _open_session_db_for_profile", + "search_sessions:get_session(compression_root/parent)": "_read_session_search -> _open_session_db_for_profile", + "search_sessions:search_messages(FTS)": "_read_session_search -> _open_session_db_for_profile", + "get_session_detail:get_session": "_read_session_detail -> _open_session_db_for_profile", + "rename_session_endpoint:get_session_title": "_rename_session_record -> _open_session_db_for_profile", + "get_session_stats:session_counts_by_source": "_read_session_stats -> _open_session_db_for_profile", +} + + +def test_affinity_record_covers_prd_sites_and_paired_total_count(): + assert len(PRD_NINE_SITE_AFFINITY_RECORD) == 10 + assert all( + target.endswith("_open_session_db_for_profile") + for target in PRD_NINE_SITE_AFFINITY_RECORD.values() + ) + + +def _parent_map(tree: ast.AST) -> dict[ast.AST, ast.AST]: + parents: dict[ast.AST, ast.AST] = {} + for parent in ast.walk(tree): + for child in ast.iter_child_nodes(parent): + parents[child] = parent + return parents + + +def _func_name(func: ast.AST) -> str | None: + if isinstance(func, ast.Name): + return func.id + if isinstance(func, ast.Attribute): + return func.attr + return None + + +def _nearest_callable( + node: ast.AST, parents: dict[ast.AST, ast.AST] +) -> ast.AsyncFunctionDef | ast.FunctionDef | ast.Lambda | None: + current = node + while current in parents: + current = parents[current] + if isinstance(current, (ast.AsyncFunctionDef, ast.FunctionDef, ast.Lambda)): + return current + return None + + +def _enclosing_async( + node: ast.AST, parents: dict[ast.AST, ast.AST] +) -> ast.AsyncFunctionDef | None: + current = node + while current in parents: + current = parents[current] + if isinstance(current, ast.AsyncFunctionDef): + return current + return None + + +def _is_offload_call(call: ast.Call) -> bool: + return _func_name(call.func) in OFFLOAD_CALL_NAMES + + +def _lambda_is_offloaded(lambda_node: ast.Lambda, parents: dict[ast.AST, ast.AST]) -> bool: + parent = parents.get(lambda_node) + return isinstance(parent, ast.Call) and _is_offload_call(parent) + + +def _named_inner_is_offloaded(inner: ast.FunctionDef, async_fn: ast.AsyncFunctionDef) -> bool: + for node in ast.walk(async_fn): + if isinstance(node, ast.Call) and _is_offload_call(node): + if any(isinstance(arg, ast.Name) and arg.id == inner.name for arg in node.args): + return True + return False + + +def _bare_sessiondb_reads(source: str) -> list[tuple[int, str, str]]: + tree = ast.parse(textwrap.dedent(source)) + parents = _parent_map(tree) + violations: list[tuple[int, str, str]] = [] + + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + if not isinstance(node.func, ast.Attribute): + continue + if node.func.attr not in SESSIONDB_READ_METHODS: + continue + if not isinstance(node.func.value, ast.Name) or node.func.value.id != "db": + continue + + owner = _nearest_callable(node, parents) + if owner is None: + continue + if isinstance(owner, ast.AsyncFunctionDef): + violations.append((node.lineno, owner.name, node.func.attr)) + continue + + async_fn = _enclosing_async(owner, parents) + if async_fn is None: + # Top-level sync helpers are safe when the async route awaits a + # single executor offload around the helper call. + continue + if isinstance(owner, ast.Lambda) and _lambda_is_offloaded(owner, parents): + continue + if isinstance(owner, ast.FunctionDef) and _named_inner_is_offloaded(owner, async_fn): + continue + violations.append((node.lineno, async_fn.name, node.func.attr)) + + return violations + + +def test_ast_lint_self_test_distinguishes_bare_reads_from_offloaded_forms(): + source = """ + import asyncio + import functools + + async def bad(db): + return db.get_session("x") + + async def ok_lambda(db, loop): + return await loop.run_in_executor(None, lambda: db.get_session("x")) + + async def ok_partial(db, loop): + return await loop.run_in_executor(None, functools.partial(db.get_session, "x")) + + async def ok_inner_def(db, loop): + def _read(): + return db.get_session("x") + return await loop.run_in_executor(None, _read) + + async def ok_to_thread(db): + return await asyncio.to_thread(db.get_session, "x") + + def sync_helper(db): + return db.get_session("x") + """ + + assert _bare_sessiondb_reads(source) == [(6, "bad", "get_session")] + + +def test_web_server_has_no_bare_sessiondb_reads_in_async_handlers(): + source = Path(web_server.__file__).read_text(encoding="utf-8") + assert _bare_sessiondb_reads(source) == [] + + +def test_session_db_heavy_reads_are_limited_to_two_executor_calls(monkeypatch): + in_flight = 0 + max_in_flight = 0 + + async def fake_blocking_io(fn, *args): + nonlocal in_flight, max_in_flight + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + try: + await asyncio.sleep(0.01) + return fn(*args) + finally: + in_flight -= 1 + + monkeypatch.setattr(web_server, "_blocking_io", fake_blocking_io) + + async def run_reads(): + return await asyncio.gather( + *( + web_server._session_db_read(lambda value=value: value, heavy=True) + for value in range(3) + ) + ) + + assert asyncio.run(run_reads()) == [0, 1, 2] + assert max_in_flight == 2 + + +def test_get_status_offloads_active_session_scan(monkeypatch): + loop_thread = threading.get_ident() + db_threads: list[int] = [] + + class _GatewayConfig: + def get_connected_platforms(self): + return [] + + class _DB: + def list_sessions_rich(self, *, limit): + assert limit == 50 + db_threads.append(threading.get_ident()) + return [{"ended_at": None, "last_active": time.time()}] + + def close(self): + db_threads.append(threading.get_ident()) + + import gateway.config as gateway_config + + monkeypatch.setattr(web_server, "check_config_version", lambda: (1, 1)) + monkeypatch.setattr(web_server, "get_running_pid", lambda: None) + monkeypatch.setattr(web_server, "_GATEWAY_HEALTH_URL", None) + monkeypatch.setattr(web_server, "read_runtime_status", lambda: None) + monkeypatch.setattr(gateway_config, "load_gateway_config", lambda: _GatewayConfig()) + monkeypatch.setattr(web_server, "_resolve_restart_drain_timeout", lambda: 0) + monkeypatch.setattr(web_server, "_dashboard_local_update_managed_externally", lambda: False) + monkeypatch.setattr(web_server.app.state, "auth_required", True, raising=False) + monkeypatch.setattr(hermes_state, "SessionDB", _DB) + + status = asyncio.run(web_server.get_status()) + + assert status["active_sessions"] == 1 + assert db_threads + assert all(thread_id != loop_thread for thread_id in db_threads) + + +def test_get_sessions_offloads_sessiondb_page(monkeypatch): + loop_thread = threading.get_ident() + db_threads: list[int] = [] + + class _DB: + def _record(self): + db_threads.append(threading.get_ident()) + + def list_sessions_rich(self, **kwargs): + self._record() + assert kwargs["limit"] == 5 + assert kwargs["offset"] == 1 + assert kwargs["order_by_last_active"] is True + return [{"id": "s1", "ended_at": None, "last_active": 0, "archived": 0}] + + def session_count(self, **kwargs): + self._record() + assert kwargs["exclude_children"] is True + return 1 + + def close(self): + self._record() + + monkeypatch.setattr(web_server, "_open_session_db_for_profile", lambda profile=None: _DB()) + + result = asyncio.run(web_server.get_sessions(limit=5, offset=1, order="recent")) + + assert result == { + "sessions": [ + {"id": "s1", "ended_at": None, "last_active": 0, "archived": False, "is_active": False} + ], + "total": 1, + "limit": 5, + "offset": 1, + } + assert db_threads + assert all(thread_id != loop_thread for thread_id in db_threads) + + +def test_search_sessions_offloads_all_sessiondb_reads(monkeypatch): + loop_thread = threading.get_ident() + db_threads: list[int] = [] + + class _DB: + def _record(self): + db_threads.append(threading.get_ident()) + + def search_sessions_by_id(self, query, *, limit, include_archived): + self._record() + assert query == "root" + assert limit == 2 + assert include_archived is True + return [ + { + "id": "root", + "preview": "", + "source": "cli", + "model": "m", + "started_at": 1, + } + ] + + def search_messages(self, *, query, limit): + self._record() + assert query == "root*" + assert limit == 50 + return [ + { + "session_id": "child", + "snippet": "hit", + "role": "user", + "source": "cli", + "model": "m", + "session_started": 2, + } + ] + + def get_session(self, session_id): + self._record() + return {"id": session_id, "parent_session_id": None, "started_at": 1} + + def get_compression_tip(self, session_id): + self._record() + return session_id + + def close(self): + self._record() + + monkeypatch.setattr(web_server, "_open_session_db_for_profile", lambda profile=None: _DB()) + + result = asyncio.run(web_server.search_sessions(q="root", limit=2)) + + assert result == { + "results": [ + { + "snippet": "Session ID: root", + "role": None, + "source": "cli", + "model": "m", + "session_started": 1, + "session_id": "root", + "lineage_root": "root", + }, + { + "snippet": "hit", + "role": "user", + "source": "cli", + "model": "m", + "session_started": 2, + "session_id": "child", + "lineage_root": "child", + }, + ] + } + assert db_threads + assert all(thread_id != loop_thread for thread_id in db_threads) + + +def test_get_session_detail_offloads_sessiondb_lookup(monkeypatch): + loop_thread = threading.get_ident() + db_threads: list[int] = [] + + class _DB: + def _record(self): + db_threads.append(threading.get_ident()) + + def resolve_session_id(self, session_id): + self._record() + assert session_id == "s" + return "sid" + + def get_session(self, session_id): + self._record() + assert session_id == "sid" + return {"id": "sid", "title": "Session"} + + def close(self): + self._record() + + monkeypatch.setattr(web_server, "_open_session_db_for_profile", lambda profile=None: _DB()) + + assert asyncio.run(web_server.get_session_detail("s")) == {"id": "sid", "title": "Session"} + assert db_threads + assert all(thread_id != loop_thread for thread_id in db_threads) + + +def test_rename_session_offloads_sessiondb_update_and_title_read(monkeypatch): + loop_thread = threading.get_ident() + db_threads: list[int] = [] + + class _DB: + def _record(self): + db_threads.append(threading.get_ident()) + + def resolve_session_id(self, session_id): + self._record() + assert session_id == "s" + return "sid" + + def set_session_title(self, session_id, title): + self._record() + assert (session_id, title) == ("sid", "New title") + + def set_session_archived(self, session_id, archived): + self._record() + assert (session_id, archived) == ("sid", True) + + def get_session_title(self, session_id): + self._record() + assert session_id == "sid" + return "New title" + + def close(self): + self._record() + + monkeypatch.setattr(web_server, "_open_session_db_for_profile", lambda profile=None: _DB()) + + result = asyncio.run( + web_server.rename_session_endpoint( + "s", + web_server.SessionRename(title="New title", archived=True), + ) + ) + + assert result == {"ok": True, "title": "New title", "archived": True} + assert db_threads + assert all(thread_id != loop_thread for thread_id in db_threads)