From 4d9007cd1712084b7b13de982f81915ed0c3a367 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:12:22 -0700 Subject: [PATCH] refactor(web): simplify sessions/tools/skills/chat routers and session-db helpers - sessions: table-driven prune filters, shared scope kwargs, _project_for_display, _compact_json, _is_compression_edge, dedup 404 detail, flag-setter table - web_server_sessions: descendants CTE via db._conn + dict(zip) rows, _is_stale_schema_error, compact docs - tools: _BACKEND_PROBES dispatch table, _env_value/_category_providers helpers - skills: _hub_sources/_resolve_hub_skill/_hub_lookup/_flag helpers, _API_SOURCE_IDS - web_server_chat: _ws_client_is_allowed delegates to _ws_client_reason, _server_internal_ws_url unifies gateway/sidecar URL builders, _reject/_stamp_identity - layout compaction (AST-neutral), docstring compaction keeping every WHY --- hermes_cli/web_routers/sessions.py | 497 +++++++++---------------- hermes_cli/web_routers/skills.py | 210 +++++------ hermes_cli/web_routers/tools.py | 419 ++++++++------------- hermes_cli/web_server_chat.py | 575 +++++++++-------------------- hermes_cli/web_server_sessions.py | 200 ++++------ 5 files changed, 663 insertions(+), 1238 deletions(-) diff --git a/hermes_cli/web_routers/sessions.py b/hermes_cli/web_routers/sessions.py index b5440fef7f..634571d1a1 100644 --- a/hermes_cli/web_routers/sessions.py +++ b/hermes_cli/web_routers/sessions.py @@ -13,7 +13,7 @@ import json import re import sqlite3 import time -from typing import Callable, List, Optional +from typing import Any, Callable, Dict, List, Optional from fastapi import APIRouter, HTTPException, Query, Request from fastapi.encoders import jsonable_encoder @@ -21,15 +21,9 @@ from fastapi.responses import StreamingResponse from hermes_cli.web_deps import late from hermes_cli.web_models import ( - BulkDeleteSessions, - SessionImport, - SessionOwnerBackfill, - SessionPrune, - SessionRename, -) + BulkDeleteSessions, SessionImport, SessionOwnerBackfill, SessionPrune, SessionRename) from hermes_cli.web_routers._common import log as _log, http_failure from hermes_state import is_malformed_db_error, is_transient_sqlite_error -from typing import Any, Dict list_router = APIRouter() search_router = APIRouter() @@ -42,20 +36,16 @@ _open_session_db_for_profile = late("_open_session_db_for_profile") _session_latest_descendant = late("_session_latest_descendant") _strip_session_list_rows = late("_strip_session_list_rows") +_NOT_FOUND = "Session not found" -# CRITICAL — every literal-path route below MUST be declared BEFORE the -# templated ``/api/sessions/{session_id}`` family that follows. FastAPI/ -# Starlette match routes in registration order, and the ``{session_id}`` -# pattern is unconstrained — it would otherwise swallow e.g. -# ``DELETE /api/sessions/empty``, ``POST /api/sessions/bulk-delete``, or -# ``GET /api/sessions/stats`` as "operate on the session with id -# 'empty'" / "'bulk-delete'" / "'stats'", which would 404 (or worse, -# succeed and delete the wrong row). Same story as the older -# ``/api/sessions/search`` endpoint up at line ~1191. If you split or -# reorder this block, move every route in it together. -# Keep the dashboard import endpoint stream-safe: FastAPI otherwise parses and -# buffers an arbitrarily large JSON body before SessionDB can enforce its own -# per-session and transaction-work limits. +# CRITICAL — every literal-path route on ``manage_router`` MUST be declared +# BEFORE the templated ``/api/sessions/{session_id}`` family. Starlette matches +# in registration order and ``{session_id}`` is unconstrained, so e.g. +# ``DELETE /api/sessions/empty`` would otherwise be taken as "the session with +# id 'empty'" (404, or worse, deleting the wrong row). Move the block as a unit. + +# Stream-safe import: FastAPI otherwise buffers an arbitrarily large JSON body +# before SessionDB can enforce its own per-session and transaction limits. _SESSION_IMPORT_MAX_BYTES = 25 * 1024 * 1024 @@ -76,58 +66,39 @@ def _import_sessions_for_profile(profile: Optional[str], sessions: List[Dict[str db.close() +# Prune filters forwarded to SessionDB; string filters map "" -> None. +_PRUNE_STR_FILTERS = ( + "source", "title_like", "end_reason", "cwd_prefix", "model_like", "provider", + "user_id", "chat_id", "chat_type", "branch_like") +_PRUNE_NUM_FILTERS = ( + "min_messages", "max_messages", "min_tokens", "max_tokens", "min_cost", "max_cost", + "min_tool_calls", "max_tool_calls") + + def _prune_sessions(body: SessionPrune): """Delete ended sessions matching filters (mirrors `hermes sessions prune`).""" from hermes_cli.web_server import get_hermes_home - has_window = ( - body.started_before is not None or body.started_after is not None - ) + has_window = body.started_before is not None or body.started_after is not None if body.older_than_days is not None and body.older_than_days < 1 and not has_window: raise HTTPException(status_code=400, detail="older_than_days must be >= 1") # Mirror the CLI: the implicit 90-day cutoff only applies to a truly bare - # prune. Any attribute filter (source, title, model, ...) suppresses it - # unless the caller explicitly sent older_than_days. - _attr_filters_set = any( - getattr(body, f) is not None - for f in ( - "source", "title_like", "end_reason", "cwd_prefix", - "min_messages", "max_messages", "model_like", "provider", - "user_id", "chat_id", "chat_type", "branch_like", - "min_tokens", "max_tokens", "min_cost", "max_cost", - "min_tool_calls", "max_tool_calls", - ) - ) - _older_than_explicit = "older_than_days" in body.model_fields_set - _effective_older_than = body.older_than_days - if has_window or (_attr_filters_set and not _older_than_explicit): - _effective_older_than = None + # prune. Any attribute filter suppresses it unless older_than_days was + # explicitly sent. + attr_filters_set = any( + getattr(body, f) is not None for f in _PRUNE_STR_FILTERS + _PRUNE_NUM_FILTERS) + effective_older_than = body.older_than_days + if has_window or (attr_filters_set and "older_than_days" not in body.model_fields_set): + effective_older_than = None profile_home = _cron_profile_home(body.profile)[1] if body.profile else get_hermes_home() db = _open_session_db_for_profile(body.profile, read_only=False) try: - filters = dict( - older_than_days=_effective_older_than, - source=(body.source or None), - started_before=body.started_before, - started_after=body.started_after, - title_like=(body.title_like or None), - end_reason=(body.end_reason or None), - cwd_prefix=(body.cwd_prefix or None), - min_messages=body.min_messages, - max_messages=body.max_messages, - model_like=(body.model_like or None), - provider=(body.provider or None), - user_id=(body.user_id or None), - chat_id=(body.chat_id or None), - chat_type=(body.chat_type or None), - branch_like=(body.branch_like or None), - min_tokens=body.min_tokens, - max_tokens=body.max_tokens, - min_cost=body.min_cost, - max_cost=body.max_cost, - min_tool_calls=body.min_tool_calls, - max_tool_calls=body.max_tool_calls, - archived=None if body.include_archived else False, - ) + filters = { + "older_than_days": effective_older_than, + "started_before": body.started_before, + "started_after": body.started_after, + "archived": None if body.include_archived else False, + **{f: (getattr(body, f) or None) for f in _PRUNE_STR_FILTERS}, + **{f: getattr(body, f) for f in _PRUNE_NUM_FILTERS}} skipped_open = db.count_open_prune_matches(**filters) if body.dry_run: rows = db.list_prune_candidates(**filters) @@ -139,34 +110,22 @@ def _prune_sessions(body: SessionPrune): # Rows are ordered by last activity, not creation time. "oldest_last_active": rows[0]["last_active"] if rows else None, "newest_last_active": rows[-1]["last_active"] if rows else None, - "oldest_started_at": ( - min(r["started_at"] for r in rows) if rows else None - ), - "newest_started_at": ( - max(r["started_at"] for r in rows) if rows else None - ), + "oldest_started_at": min(r["started_at"] for r in rows) if rows else None, + "newest_started_at": max(r["started_at"] for r in rows) if rows else None, "sessions": [ { - "id": r["id"], - "source": r["source"], - "title": r.get("title"), - "model": r.get("model"), - "started_at": r["started_at"], - "last_active": r["last_active"], - "message_count": r["message_count"], - } - for r in rows - ], - } + "id": r["id"], "source": r["source"], "title": r.get("title"), + "model": r.get("model"), "started_at": r["started_at"], + "last_active": r["last_active"], "message_count": r["message_count"]} + for r in rows]} sessions_dir = profile_home / "sessions" removed = db.prune_sessions( - sessions_dir=sessions_dir if sessions_dir.exists() else None, - **filters, - ) + sessions_dir=sessions_dir if sessions_dir.exists() else None, **filters) return {"ok": True, "removed": removed, "skipped_open": skipped_open} finally: db.close() + _ACTIVE_WINDOW_S = 300 @@ -178,8 +137,7 @@ def _csv(value: Optional[str]) -> List[str]: def _is_active(row: dict, now: float) -> bool: return ( row.get("ended_at") is None - and (now - row.get("last_active", row.get("started_at", 0))) < _ACTIVE_WINDOW_S - ) + and (now - row.get("last_active", row.get("started_at", 0))) < _ACTIVE_WINDOW_S) def _with_db(profile: Optional[str], fn: Callable, *, read_only: bool): @@ -200,26 +158,22 @@ def _serving_profile(profile: Optional[str]) -> str: def _resolve_session_id(db, session_id: str) -> Optional[str]: """Resolve *session_id*, distinguishing "absent" from "unreadable". - On a corrupt ``state.db`` the exact-match lookup (primary-key index) just - misses, while the prefix fallback scans the b-tree and raises "database - disk image is malformed". Both used to end at 404; report corruption as - 503 with the actual problem instead of an empty store. + On a corrupt ``state.db`` the exact-match lookup just misses while the + prefix fallback scans the b-tree and raises "malformed"; report corruption + as 503 with the actual problem instead of an empty store (404). """ try: return db.resolve_session_id(session_id) except sqlite3.DatabaseError as exc: if not is_malformed_db_error(exc): raise - _log.error( - "state.db is corrupt while resolving session %s: %s", session_id, exc - ) + _log.error("state.db is corrupt while resolving session %s: %s", session_id, exc) raise HTTPException( status_code=503, detail=( "Session store is corrupt (database disk image is malformed). " "Sessions cannot be read until it is repaired — run " - "`hermes doctor` for diagnosis." - ), + "`hermes doctor` for diagnosis."), ) from exc @@ -237,8 +191,7 @@ def get_sessions( exclude_sources: str = None, cwd_prefix: str = None, full: bool = False, - profile: Optional[str] = None, -): + profile: Optional[str] = None): """List sessions. ``archived``: ``exclude`` (default) / ``only`` / ``include``. ``order``: @@ -248,15 +201,9 @@ def get_sessions( ``model_config`` unless ``full=1``. """ if archived not in ("exclude", "only", "include"): - raise HTTPException( - status_code=400, - detail="archived must be one of: exclude, only, include", - ) + raise HTTPException(status_code=400, detail="archived must be one of: exclude, only, include") if order not in ("created", "recent"): - raise HTTPException( - status_code=400, - detail="order must be one of: created, recent", - ) + raise HTTPException(status_code=400, detail="order must be one of: created, recent") profile_name: Optional[str] = None if profile: profile_name, _ = _cron_profile_home(profile) @@ -273,32 +220,24 @@ def get_sessions( # the cron-jobs section (source=cron) into two independent lists. source_list = _csv(sources) exclude_list = _csv(exclude_sources) - sessions = db.list_sessions_rich( + scope = dict( source=source or None, sources=source_list 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, + archived_only=archived_only) + sessions = db.list_sessions_rich( + limit=limit, + offset=offset, order_by_last_active=order == "recent", # Skip the system_prompt blob inside SQLite too (pairs with # _strip_session_list_rows below). compact_rows=not full, include_pinned=True, - ) - total = db.session_count( - source=source or None, - sources=source_list 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, - ) + **scope) + total = db.session_count(exclude_children=True, **scope) now = time.time() row_profile = profile_name or _cron_default_profile() for s in sessions: @@ -326,14 +265,23 @@ def get_sessions( detail=( "Session store is busy (disk I/O or lock). Retry; the list was not cleared." if transient - else "Internal server error" - ), + else "Internal server error"), ) from exc except Exception: _log.exception("GET /api/sessions failed") raise HTTPException(status_code=500, detail="Internal server error") +def _is_compression_edge(child: dict, parent: dict) -> bool: + parent_ended_at = parent.get("ended_at") + started_at = child.get("started_at") + return ( + parent.get("end_reason") == "compression" + and parent_ended_at is not None + and started_at is not None + and started_at >= parent_ended_at) + + @search_router.get("/api/sessions/search") async def search_sessions( q: str = "", @@ -341,8 +289,7 @@ async def search_sessions( profile: Optional[str] = None, source: str = None, sources: str = None, - exclude_sources: str = None, -): + exclude_sources: str = None): """Search sessions by ID plus FTS5 message content. ID matches first, then content matches. Results are deduped by @@ -363,6 +310,12 @@ async def search_sessions( exclude_list = _csv(exclude_sources) now = time.time() + def get_session(sid): + try: + return db.get_session(sid) + except Exception: + return None + # Walk parent_session_id to the compression root, memoized per # chain; stops at branch/delegate edges (those stay searchable). root_cache: dict = {} @@ -382,33 +335,10 @@ async def search_sessions( 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 + s = get_session(cur) 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: + parent_session = get_session(parent) if parent else None + if not parent_session or not _is_compression_edge(s, parent_session): root = cur break cur = parent @@ -423,9 +353,7 @@ async def search_sessions( return tip_cache[root_id] tip = root_id try: - resolved = db.get_compression_tip(root_id) - if resolved: - tip = resolved + tip = db.get_compression_tip(root_id) or tip except Exception: pass tip_cache[root_id] = tip @@ -450,92 +378,59 @@ async def search_sessions( except Exception: row = None if row: - payload.update( - { - "id": row.get("id") or sid, - "source": row.get("source"), - "model": row.get("model"), - "title": row.get("title"), - "started_at": row.get("started_at"), - "ended_at": row.get("ended_at"), - "last_active": row.get("last_active") or row.get("started_at"), - "is_active": ( - row.get("ended_at") is None - and (now - (row.get("last_active") or row.get("started_at") or 0)) < 300 - ), - "message_count": row.get("message_count") or 0, - "tool_call_count": row.get("tool_call_count") or 0, - "input_tokens": row.get("input_tokens") or 0, - "output_tokens": row.get("output_tokens") or 0, - "preview": row.get("preview"), - "parent_session_id": row.get("parent_session_id"), - "archived": bool(row.get("archived")), - } - ) + last_active = row.get("last_active") or row.get("started_at") + payload.update({ + "id": row.get("id") or sid, + "source": row.get("source"), + "model": row.get("model"), + "title": row.get("title"), + "started_at": row.get("started_at"), + "ended_at": row.get("ended_at"), + "last_active": last_active, + "is_active": row.get("ended_at") is None and (now - (last_active or 0)) < 300, + "message_count": row.get("message_count") or 0, + "tool_call_count": row.get("tool_call_count") or 0, + "input_tokens": row.get("input_tokens") or 0, + "output_tokens": row.get("output_tokens") or 0, + "preview": row.get("preview"), + "parent_session_id": row.get("parent_session_id"), + "archived": bool(row.get("archived"))}) else: payload["id"] = sid seen[root] = payload + def hit_payload(row: dict, snippet: str, role) -> dict: + return { + "snippet": snippet, "role": role, "source": row.get("source"), + "model": row.get("model"), "session_started": row.get("session_started")} + # Direct ID matches first (pasted ids never appear in message text). for row in db.search_sessions_by_id( - q, - limit=safe_limit, - include_archived=True, - source=source_filter, - sources=source_list or None, - exclude_sources=exclude_list or None, - ): + q, limit=safe_limit, include_archived=True, source=source_filter, + sources=source_list or None, exclude_sources=exclude_list or None): 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"), - }, - ) + payload = hit_payload(row, preview or f"Session ID: {sid}", None) + payload["session_started"] = row.get("started_at") + add_lineage_result(sid, payload) # Prefix wildcards so partial words match ("nimb" -> "nimb*"); # quoted phrases and existing wildcards are kept as-is. prefix_query = " ".join( tok if tok.startswith('"') or tok.endswith("*") else tok + "*" - for tok in re.findall(r'"[^"]*"|\S+', q.strip()) - ) + for tok in re.findall(r'"[^"]*"|\S+', q.strip())) # Over-fetch so lineage dedup can still surface `limit` distinct # conversations when several hits collapse onto one root. - fetch_limit = max(safe_limit * 5, 50) matches = db.search_messages( query=prefix_query, source_filter=include_sources, exclude_sources=exclude_list or None, - limit=fetch_limit, - fields=( - "session_id", - "role", - "snippet", - "source", - "model", - "session_started", - ), - ) - + limit=max(safe_limit * 5, 50), + fields=("session_id", "role", "snippet", "source", "model", "session_started")) 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"), - }, - ) + add_lineage_result(m["session_id"], hit_payload(m, m.get("snippet", ""), m.get("role"))) return {"results": list(seen.values())} finally: db.close() @@ -555,13 +450,9 @@ async def bulk_delete_sessions_endpoint(body: BulkDeleteSessions): # Hard cap so a runaway selection can't lock the writer for long; 500 # covers "select all on every page of a reasonable scrollback". if len(body.ids) > 500: - raise HTTPException( - status_code=400, - detail="ids must contain at most 500 entries", - ) + raise HTTPException(status_code=400, detail="ids must contain at most 500 entries") deleted = await asyncio.to_thread( - _with_db, body.profile, lambda db: db.delete_sessions(body.ids), read_only=False - ) + _with_db, body.profile, lambda db: db.delete_sessions(body.ids), read_only=False) return {"ok": True, "deleted": deleted} @@ -592,8 +483,7 @@ async def count_empty_sessions_endpoint(profile: Optional[str] = None): """Count of empty, ended, non-archived sessions (drives the "Delete empty (N)" button, hidden when N is 0).""" count = await asyncio.to_thread( - _with_db, profile, lambda db: db.count_empty_sessions(), read_only=True - ) + _with_db, profile, lambda db: db.count_empty_sessions(), read_only=True) return {"count": count} @@ -608,8 +498,7 @@ async def delete_empty_sessions_endpoint(profile: Optional[str] = None): orphaned; on-disk cleanup is left to the next prune pass. """ deleted = await asyncio.to_thread( - _with_db, profile, lambda db: db.delete_empty_sessions(), read_only=False - ) + _with_db, profile, lambda db: db.delete_empty_sessions(), read_only=False) return {"ok": True, "deleted": deleted} @@ -623,13 +512,9 @@ async def get_session_stats(profile: Optional[str] = None): "active_store": db.session_count(include_archived=False), "archived": db.session_count(archived_only=True), "messages": db.message_count(), - "by_source": {}, - } + "by_source": {}} try: - out["by_source"] = db.session_count_by_source( - include_archived=True, - exclude_children=True, - ) + out["by_source"] = db.session_count_by_source(include_archived=True, exclude_children=True) except Exception: pass return out @@ -643,7 +528,7 @@ async def get_session_detail(session_id: str, profile: Optional[str] = None): sid = _resolve_session_id(db, session_id) session = db.get_session(sid) if sid else None if not session: - raise HTTPException(status_code=404, detail="Session not found") + raise HTTPException(status_code=404, detail=_NOT_FOUND) # Always stamp the owner: stamping only on explicit ``?profile=`` left # default-profile rows unowned, so multi-profile clients resolved them # to whichever gateway happened to be active. @@ -655,61 +540,20 @@ async def get_session_detail(session_id: str, profile: Optional[str] = None): @manage_router.get("/api/sessions/{session_id}/latest-descendant") -async def get_session_latest_descendant( - session_id: str, - profile: Optional[str] = None, -): +async def get_session_latest_descendant(session_id: str, profile: Optional[str] = None): latest, path = await asyncio.to_thread( - _with_db, profile, lambda db: _session_latest_descendant(session_id, db), read_only=True - ) + _with_db, profile, lambda db: _session_latest_descendant(session_id, db), read_only=True) if not latest: - raise HTTPException(status_code=404, detail="Session not found") + raise HTTPException(status_code=404, detail=_NOT_FOUND) return { "requested_session_id": path[0] if path else session_id, "session_id": latest, "path": path, - "changed": bool(path and latest != path[0]), - } + "changed": bool(path and latest != path[0])} -@manage_router.get("/api/sessions/{session_id}/messages") -async def get_session_messages( - session_id: str, - profile: Optional[str] = None, - limit: Optional[int] = Query(None, ge=0), - offset: int = Query(0, ge=0), - order: Optional[str] = Query(None), - include_compacted: bool = Query(False), -): - if order not in (None, "oldest", "latest"): - raise HTTPException( - status_code=400, - detail="order must be one of: oldest, latest", - ) - - def _read(db): - sid = _resolve_session_id(db, session_id) - if not sid: - return None - sid = db.resolve_resume_session_id(sid) - # Always page: an omitted limit used to load whole transcripts (hundreds - # of thousands of rows for a runaway session). Explicit pagination - # anchors at the start; the default view is the latest page. - default_page = limit is None - latest_page = order == "latest" or (order is None and default_page) - _limit = 500 if default_page else min(limit, 500) - return sid, _limit, db.get_messages( - sid, - limit=_limit, - offset=offset, - latest=latest_page, - include_compacted=include_compacted, - ) - - result = await asyncio.to_thread(_with_db, profile, _read, read_only=True) - if result is None: - raise HTTPException(status_code=404, detail="Session not found") - sid, _limit, messages = result +def _project_for_display(messages: list) -> list: + """Replace compaction summaries with their display-only projection.""" from agent.compaction_display import project_compaction_message_for_display from agent.context_compressor import is_compaction_summary_message @@ -730,6 +574,40 @@ async def get_session_messages( projected["display_content"] = display_view.get("content") projected.pop("display_kind", None) projected_messages.append(projected) + return projected_messages + + +@manage_router.get("/api/sessions/{session_id}/messages") +async def get_session_messages( + session_id: str, + profile: Optional[str] = None, + limit: Optional[int] = Query(None, ge=0), + offset: int = Query(0, ge=0), + order: Optional[str] = Query(None), + include_compacted: bool = Query(False)): + if order not in (None, "oldest", "latest"): + raise HTTPException(status_code=400, detail="order must be one of: oldest, latest") + + def _read(db): + sid = _resolve_session_id(db, session_id) + if not sid: + return None + sid = db.resolve_resume_session_id(sid) + # Always page: an omitted limit used to load whole transcripts (hundreds + # of thousands of rows for a runaway session). Explicit pagination + # anchors at the start; the default view is the latest page. + default_page = limit is None + latest_page = order == "latest" or (order is None and default_page) + _limit = 500 if default_page else min(limit, 500) + return sid, _limit, db.get_messages( + sid, limit=_limit, offset=offset, latest=latest_page, include_compacted=include_compacted + ) + + result = await asyncio.to_thread(_with_db, profile, _read, read_only=True) + if result is None: + raise HTTPException(status_code=404, detail=_NOT_FOUND) + sid, _limit, messages = result + projected_messages = _project_for_display(messages) return { "session_id": sid, "messages": projected_messages, @@ -737,9 +615,7 @@ async def get_session_messages( "limit": _limit, "offset": offset, "order": order or ("latest" if limit is None else "oldest"), - "returned": len(projected_messages), - }, - } + "returned": len(projected_messages)}} @manage_router.delete("/api/sessions/{session_id}") @@ -780,12 +656,16 @@ async def backfill_session_owner_profiles(body: SessionOwnerBackfill): if stamped: _log.info( "owner-backfill: stamped %d legacy NULL-profile session row(s) with profile %r", - stamped, - stamp, - ) + stamped, stamp) return {"ok": True, "stamped": stamped, "profile": stamp} +_RENAME_FLAG_SETTERS = ( + ("archived", "set_session_archived"), + ("hidden", "set_session_hidden"), + ("pinned", "set_session_pinned")) + + @manage_router.patch("/api/sessions/{session_id}") async def rename_session_endpoint(session_id: str, body: SessionRename): """Update a session: ``title`` (empty clears), ``archived``, ``hidden``, @@ -796,7 +676,7 @@ async def rename_session_endpoint(session_id: str, body: SessionRename): def _update(db): sid = _resolve_session_id(db, session_id) if not sid: - raise HTTPException(status_code=404, detail="Session not found") + raise HTTPException(status_code=404, detail=_NOT_FOUND) if body.title is None and all(getattr(body, f) is None for f in flags): raise HTTPException( status_code=400, @@ -808,12 +688,10 @@ async def rename_session_endpoint(session_id: str, body: SessionRename): 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 body.hidden is not None: - db.set_session_hidden(sid, body.hidden) - if body.pinned is not None: - db.set_session_pinned(sid, body.pinned) + for flag, setter in _RENAME_FLAG_SETTERS: + value = getattr(body, flag) + if value is not None: + getattr(db, setter)(sid, value) if body.unread is not None: db.set_session_read(sid, read=not body.unread) result = {"ok": True, "title": db.get_session_title(sid) or ""} @@ -825,6 +703,10 @@ async def rename_session_endpoint(session_id: str, body: SessionRename): return _with_db(body.profile, _update, read_only=False) +def _compact_json(obj) -> str: + return json.dumps(jsonable_encoder(obj), ensure_ascii=False, separators=(",", ":")) + + @manage_router.get("/api/sessions/{session_id}/export") async def export_session_endpoint(session_id: str, profile: Optional[str] = None): """Stream a single session (metadata + messages) as JSON.""" @@ -834,53 +716,36 @@ async def export_session_endpoint(session_id: str, profile: Optional[str] = None prepared = await asyncio.to_thread(_with_db, profile, _prepare_export, read_only=True) if prepared is None or prepared[1] is None: - raise HTTPException(status_code=404, detail="Session not found") + raise HTTPException(status_code=404, detail=_NOT_FOUND) sid, session = prepared def _stream_export(): db = _open_session_db_for_profile(profile, read_only=True) try: - metadata = json.dumps( - jsonable_encoder(session), - ensure_ascii=False, - separators=(",", ":"), - ) - yield metadata[:-1] + ',"messages":[' - + yield _compact_json(session)[:-1] + ',"messages":[' # Keyset pagination (id > last_seen): O(n) total over the # transcript, vs OFFSET's O(n²) on huge sessions. last_id = None first = True while True: messages = db.get_messages( - sid, - limit=500, - after_id=last_id if last_id is not None else 0, - ) + sid, limit=500, after_id=last_id if last_id is not None else 0) for message in messages: if not first: yield "," - yield json.dumps( - jsonable_encoder(message), - ensure_ascii=False, - separators=(",", ":"), - ) + yield _compact_json(message) first = False if len(messages) < 500: break last_id = messages[-1].get("id") if last_id is None: break # defensive: cannot keyset without row ids - yield "]}" finally: db.close() - return StreamingResponse( - _stream_export(), - media_type="application/json", - ) + return StreamingResponse(_stream_export(), media_type="application/json") @manage_router.post("/api/sessions/prune") diff --git a/hermes_cli/web_routers/skills.py b/hermes_cli/web_routers/skills.py index 72f571a046..efd2e205f8 100644 --- a/hermes_cli/web_routers/skills.py +++ b/hermes_cli/web_routers/skills.py @@ -14,22 +14,11 @@ from fastapi import APIRouter, HTTPException from hermes_cli.web_deps import late from hermes_cli.web_models import ( - SkillContentUpdate, - SkillCreate, - SkillInstallRequest, - SkillToggle, - SkillUninstallRequest, - SkillsUpdateRequest, -) + SkillContentUpdate, SkillCreate, SkillInstallRequest, SkillToggle, SkillUninstallRequest, + SkillsUpdateRequest) from hermes_cli.web_routers._common import ( - _profile_scope, - config_write_scope, - http_failure, - log as _log, - require, - scoped_to_thread, - spawn_profile_action, -) + _profile_scope, config_write_scope, http_failure, log as _log, require, scoped_to_thread, + spawn_profile_action) hub_router = APIRouter() router = APIRouter() @@ -40,8 +29,8 @@ _installed_hub_identifiers = late("_installed_hub_identifiers") load_config = late("load_config") -# Human-readable labels for each hub source id (matches `hermes skills search` -# provenance). Keep in sync with create_source_router()'s source list. +# Labels per hub source id (matches `hermes skills search` provenance); keep in +# sync with create_source_router()'s source list. _SKILL_HUB_SOURCE_LABELS = { "official": "Official (Nous)", "hermes-index": "Hermes Index", @@ -51,8 +40,40 @@ _SKILL_HUB_SOURCE_LABELS = { "github": "GitHub", "clawhub": "ClawHub", "lobehub": "LobeHub", - "browse-sh": "browse.sh", -} + "browse-sh": "browse.sh"} + + +def _hub_sources(profile: Optional[str]): + """Source router built under ``profile``'s config scope.""" + from tools.skills_hub import create_source_router + + with _config_profile_scope(profile): + return create_source_router() + + +def _resolve_hub_skill(ident: str, profile: Optional[str]): + """``(meta, bundle)`` for a hub identifier, resolved under ``profile``'s scope.""" + from hermes_cli.skills_hub import _resolve_source_meta_and_bundle + from tools.skills_hub import create_source_router + + with _config_profile_scope(profile): + sources = create_source_router() + meta, bundle, _src = _resolve_source_meta_and_bundle(ident, sources) + return meta, bundle + + +# Sources subsumed by an available hermes-index (progressive per-source fan-out +# skips them: ~70 GitHub calls per keystroke saved). Keep in sync with +# parallel_search_sources' _api_source_ids. +_API_SOURCE_IDS = frozenset({"github", "skills-sh", "clawhub", "lobehub", "well-known"}) + + +def _flag(obj, attr: str) -> bool: + """``bool(getattr(obj, attr, False))``; a raising property reads as False.""" + try: + return bool(getattr(obj, attr, False)) + except Exception: + return False def _skill_meta_to_payload(m) -> dict: @@ -63,8 +84,7 @@ def _skill_meta_to_payload(m) -> dict: "identifier": m.identifier, "trust_level": m.trust_level, "repo": m.repo, - "tags": list(m.tags or []), - } + "tags": list(m.tags or [])} def _clear_skills_prompt_cache() -> None: @@ -88,8 +108,7 @@ async def install_skill_hub(body: SkillInstallRequest, profile: Optional[str] = ["skills", "install", identifier, "--yes"], _hub_action_name("install", identifier), log_msg="Failed to spawn skills install", - prefix="Failed to install skill", - ) + prefix="Failed to install skill") @hub_router.post("/api/skills/hub/uninstall") @@ -100,21 +119,18 @@ async def uninstall_skill_hub(body: SkillUninstallRequest, profile: Optional[str ["skills", "uninstall", name, "--yes"], _hub_action_name("uninstall", name), log_msg="Failed to spawn skills uninstall", - prefix="Failed to uninstall skill", - ) + prefix="Failed to uninstall skill") @hub_router.post("/api/skills/hub/update") async def update_skills_hub( - body: Optional[SkillsUpdateRequest] = None, profile: Optional[str] = None -): + body: Optional[SkillsUpdateRequest] = None, profile: Optional[str] = None): return spawn_profile_action( (body.profile if body else None) or profile, ["skills", "update"], "skills-update", log_msg="Failed to spawn skills update", - prefix="Failed to update skills", - ) + prefix="Failed to update skills") @hub_router.get("/api/skills/hub/official") @@ -149,55 +165,35 @@ async def list_skills_hub_sources(profile: Optional[str] = None): scopes the installed-skill provenance.""" def _run(): - from tools.skills_hub import create_source_router - - with _config_profile_scope(profile): - sources = create_source_router() + sources = _hub_sources(profile) out = [] index_available = False featured = [] for src in sources: sid = src.source_id() - entry = { - "id": sid, - "label": _SKILL_HUB_SOURCE_LABELS.get(sid, sid), - } + entry = {"id": sid, "label": _SKILL_HUB_SOURCE_LABELS.get(sid, sid)} # GitHub exposes a rate-limit flag; the index an availability flag. if sid == "github": - try: - entry["rate_limited"] = bool(getattr(src, "is_rate_limited", False)) - except Exception: - entry["rate_limited"] = False + entry["rate_limited"] = _flag(src, "is_rate_limited") if sid == "hermes-index": - try: - index_available = bool(getattr(src, "is_available", False)) - except Exception: - index_available = False + index_available = _flag(src, "is_available") entry["available"] = index_available # Empty-query search on the index returns featured/popular skills. if index_available: try: featured = [ - _skill_meta_to_payload(m) for m in src.search("", limit=12) - ] + _skill_meta_to_payload(m) for m in src.search("", limit=12)] except Exception: featured = [] out.append(entry) - # Which sources are worth searching individually (progressive per-source - # fan-out). Mirrors parallel_search_sources: an available index already - # subsumes the external API sources, so skipping them saves ~70 GitHub - # calls per keystroke. Keep in sync with that function's _api_source_ids. - _api_source_ids = frozenset( - {"github", "skills-sh", "clawhub", "lobehub", "well-known"} - ) + # Which sources are worth searching individually (see _API_SOURCE_IDS). for entry in out: - entry["searchable"] = not (index_available and entry["id"] in _api_source_ids) + entry["searchable"] = not (index_available and entry["id"] in _API_SOURCE_IDS) return { "sources": out, "index_available": index_available, "featured": featured, - "installed": _installed_hub_identifiers(profile), - } + "installed": _installed_hub_identifiers(profile)} with http_failure("skills hub sources listing failed", 502, "Hub sources failed"): return await asyncio.to_thread(_run) @@ -205,8 +201,7 @@ async def list_skills_hub_sources(profile: Optional[str] = None): @hub_router.get("/api/skills/hub/search") async def search_skills_hub( - q: str = "", source: str = "all", limit: int = 20, profile: Optional[str] = None -): + q: str = "", source: str = "all", limit: int = 20, profile: Optional[str] = None): """Search the skill hub across all configured sources (network-bound, runs in a thread). Results install by identifier via /hub/install.""" query = (q or "").strip() @@ -214,22 +209,18 @@ async def search_skills_hub( return {"results": [], "source_counts": {}, "timed_out": [], "installed": {}} def _run(): - from tools.skills_hub import create_source_router, parallel_search_sources + from tools.skills_hub import parallel_search_sources - with _config_profile_scope(profile): - sources = create_source_router() + sources = _hub_sources(profile) capped = min(max(limit, 1), 50) all_results, source_counts, timed_out = parallel_search_sources( - sources, query=query, source_filter=source or "all", overall_timeout=30 - ) + sources, query=query, source_filter=source or "all", overall_timeout=30) # Dedupe by identifier, preferring higher trust (mirrors unified_search). _rank = {"builtin": 2, "trusted": 1, "community": 0} seen = {} for r in all_results: - if r.identifier not in seen: - seen[r.identifier] = r - elif _rank.get(r.trust_level, 0) > _rank.get(seen[r.identifier].trust_level, 0): + if r.identifier not in seen or _rank.get(r.trust_level, 0) > _rank.get(seen[r.identifier].trust_level, 0): seen[r.identifier] = r deduped = list(seen.values())[:capped] @@ -237,13 +228,21 @@ async def search_skills_hub( "results": [_skill_meta_to_payload(m) for m in deduped], "source_counts": source_counts, "timed_out": timed_out, - "installed": _installed_hub_identifiers(profile), - } + "installed": _installed_hub_identifiers(profile)} with http_failure("skills hub search failed", 502, "Hub search failed"): return await asyncio.to_thread(_run) +async def _hub_lookup(fn, ident: str, log_msg: str, prefix: str): + """Run ``fn`` off-loop; any failure -> 502 ``": "``, None -> 404.""" + with http_failure(log_msg, 502, prefix): + result = await asyncio.to_thread(fn) + if result is None: + raise HTTPException(status_code=404, detail=f"Skill not found: {ident}") + return result + + @hub_router.get("/api/skills/hub/preview") async def preview_skill_hub(identifier: str = "", profile: Optional[str] = None): """A hub skill's SKILL.md + file manifest WITHOUT installing it. Scoped to @@ -252,12 +251,7 @@ async def preview_skill_hub(identifier: str = "", profile: Optional[str] = None) ident = require(identifier, "identifier is required") def _run(): - from hermes_cli.skills_hub import _resolve_source_meta_and_bundle - from tools.skills_hub import create_source_router - - with _config_profile_scope(profile): - sources = create_source_router() - meta, bundle, _src = _resolve_source_meta_and_bundle(ident, sources) + meta, bundle = _resolve_hub_skill(ident, profile) if not bundle and not meta: return None @@ -286,17 +280,9 @@ async def preview_skill_hub(identifier: str = "", profile: Optional[str] = None) "repo": getattr(m, "repo", None), "tags": list(getattr(m, "tags", None) or []), "skill_md": skill_md, - "files": sorted(files.keys()), - } + "files": sorted(files.keys())} - try: - result = await asyncio.to_thread(_run) - except Exception as exc: - _log.exception("skills hub preview failed") - raise HTTPException(status_code=502, detail=f"Hub preview failed: {exc}") - if result is None: - raise HTTPException(status_code=404, detail=f"Skill not found: {ident}") - return result + return await _hub_lookup(_run, ident, "skills hub preview failed", "Hub preview failed") @hub_router.get("/api/skills/hub/scan") @@ -310,24 +296,17 @@ async def scan_skill_hub(identifier: str = "", profile: Optional[str] = None): def _run(): import shutil as _shutil - from hermes_cli.skills_hub import _resolve_source_meta_and_bundle - from tools.skills_hub import create_source_router, quarantine_bundle + from tools.skills_hub import quarantine_bundle from tools.skills_guard import scan_skill, should_allow_install - with _config_profile_scope(profile): - sources = create_source_router() - meta, bundle, _src = _resolve_source_meta_and_bundle(ident, sources) + meta, bundle = _resolve_hub_skill(ident, profile) if not bundle: return None if bundle.source == "official": scan_source = "official" else: - scan_source = ( - getattr(bundle, "identifier", "") - or getattr(meta, "identifier", "") - or ident - ) + scan_source = getattr(bundle, "identifier", "") or getattr(meta, "identifier", "") or ident q_path = None tier1 = None @@ -337,9 +316,7 @@ async def scan_skill_hub(identifier: str = "", profile: Optional[str] = None): # Advisory SkillEvaluator Tier 1 second opinion: optional binary, # never blocks, errors degrade to no data (same as the CLI installer). try: - from tools.skillevaluator_scan import ( - run_tier1_scan, tier1_advisory_enabled, - ) + from tools.skillevaluator_scan import run_tier1_scan, tier1_advisory_enabled if tier1_advisory_enabled(): t1 = run_tier1_scan(q_path) if t1.available: @@ -354,11 +331,8 @@ async def scan_skill_hub(identifier: str = "", profile: Optional[str] = None): "message": f.message, "file": f.file, "line": f.line, - "secrets_class": f.is_secrets_class, - } - for f in t1.findings - ], - } + "secrets_class": f.is_secrets_class} + for f in t1.findings]} except Exception: _log.debug("Tier 1 advisory scan skipped", exc_info=True) finally: @@ -373,11 +347,9 @@ async def scan_skill_hub(identifier: str = "", profile: Optional[str] = None): "category": f.category, "file": f.file, "line": f.line, - "description": f.description, - } - for f in result.findings - ] - counts = {"critical": 0, "high": 0, "medium": 0, "low": 0} + "description": f.description} + for f in result.findings] + counts = {sev: 0 for sev in ("critical", "high", "medium", "low")} for f in result.findings: if f.severity in counts: counts[f.severity] += 1 @@ -396,14 +368,7 @@ async def scan_skill_hub(identifier: str = "", profile: Optional[str] = None): "tier1": tier1, # None when the optional scanner isn't installed/enabled } - try: - result = await asyncio.to_thread(_run) - except Exception as exc: - _log.exception("skills hub scan failed") - raise HTTPException(status_code=502, detail=f"Hub scan failed: {exc}") - if result is None: - raise HTTPException(status_code=404, detail=f"Skill not found: {ident}") - return result + return await _hub_lookup(_run, ident, "skills hub scan failed", "Hub scan failed") @router.get("/api/skills") @@ -411,11 +376,8 @@ async def get_skills(profile: Optional[str] = None): from tools.skills_tool import _find_all_skills from hermes_cli.skills_config import get_disabled_skills from tools.skill_usage import ( - _read_bundled_manifest_names, - _read_hub_installed_names, - activity_count, - load_usage, - ) + _read_bundled_manifest_names, _read_hub_installed_names, activity_count, load_usage) + def _run(): with _profile_scope(profile): config = load_config() @@ -434,8 +396,7 @@ async def get_skills(profile: Optional[str] = None): s["provenance"] = ( "hub" if s["name"] in hub_names else "bundled" if s["name"] in bundled_names - else "agent" - ) + else "agent") return skills return await asyncio.to_thread(_run) @@ -488,8 +449,7 @@ async def create_skill(body: SkillCreate): from tools.skill_manager_tool import _create_skill result = await scoped_to_thread( - body.profile, lambda: _create_skill(body.name, body.content, body.category or None) - ) + body.profile, lambda: _create_skill(body.name, body.content, body.category or None)) if not result.get("success"): raise HTTPException(status_code=400, detail=result.get("error", "Failed to create skill.")) _clear_skills_prompt_cache() diff --git a/hermes_cli/web_routers/tools.py b/hermes_cli/web_routers/tools.py index 1fb02092a7..037b0c4095 100644 --- a/hermes_cli/web_routers/tools.py +++ b/hermes_cli/web_routers/tools.py @@ -15,23 +15,11 @@ from fastapi import APIRouter, HTTPException from hermes_cli.web_deps import late from hermes_cli.web_models import ( - TerminalBackendSelect, - ToolsetEnvUpdate, - ToolsetModelSelect, - ToolsetPostSetup, - ToolsetProviderSelect, - ToolsetToggle, -) + TerminalBackendSelect, ToolsetEnvUpdate, ToolsetModelSelect, ToolsetPostSetup, + ToolsetProviderSelect, ToolsetToggle) from hermes_cli.web_routers._common import ( - _CONFIG_MUTATION_LOCK, - _profile_cli_args, - _profile_scope, - _spawn_hermes_action, - config_write_scope, - log as _log, - scoped_to_thread, - spawn_profile_action, -) + _CONFIG_MUTATION_LOCK, _profile_cli_args, _profile_scope, _spawn_hermes_action, + config_write_scope, log as _log, scoped_to_thread, spawn_profile_action) router = APIRouter() @@ -41,73 +29,57 @@ run_in_threadpool = late("run_in_threadpool") _plugin_terminal_backend_rows = late("_plugin_terminal_backend_rows") +def _env_value(name: str) -> str: + """``get_env_value`` that never raises (empty string on any failure).""" + try: + from hermes_cli.config import get_env_value + + return get_env_value(name) or "" + except Exception: + return "" + + def _terminal_cfg_value(terminal_cfg: dict, key: str, env_var: str) -> str: """Read a terminal.* setting from config.yaml, falling back to its env var.""" value = terminal_cfg.get(key) if value is not None and str(value).strip(): return str(value).strip() - try: - from hermes_cli.config import get_env_value - - return (get_env_value(env_var) or "").strip() - except Exception: - return "" + return _env_value(env_var).strip() def _terminal_backend_rows() -> List[Dict[str, str]]: - """Built-in picker rows plus plugin-registered backends (request time). - - Computed per request (mirrors ``_schema_with_dynamic_provider_options``) - so a plugin installed after server start still shows up. - """ + """Built-in picker rows plus plugin-registered backends, computed per request + so a plugin installed after server start still shows up.""" from hermes_cli.web_server import _TERMINAL_BACKENDS return [*_TERMINAL_BACKENDS, *_plugin_terminal_backend_rows()] -def _probe_docker_backend() -> tuple: +def _probe_docker_backend(_cfg) -> tuple: if not shutil.which("docker"): - return ( - "needs_setup", - "Docker CLI not found — install Docker Desktop or docker-ce.", - ) + return ("needs_setup", "Docker CLI not found — install Docker Desktop or docker-ce.") try: proc = subprocess.run( ["docker", "info", "--format", "{{.ServerVersion}}"], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=2, - ) + capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=2) if proc.returncode == 0: return ("ready", "") - return ( - "needs_setup", - "Docker daemon not reachable — start Docker and retry.", - ) + return ("needs_setup", "Docker daemon not reachable — start Docker and retry.") except subprocess.TimeoutExpired: return ("needs_setup", "Docker daemon not responding (timed out).") except Exception as exc: return ("unavailable", f"Docker probe failed: {exc}") -def _probe_singularity_backend() -> tuple: +def _probe_singularity_backend(_cfg) -> tuple: if shutil.which("singularity") or shutil.which("apptainer"): return ("ready", "") - return ( - "needs_setup", - "Neither singularity nor apptainer found on PATH.", - ) + return ("needs_setup", "Neither singularity nor apptainer found on PATH.") def _probe_ssh_backend(terminal_cfg: dict) -> tuple: host = _terminal_cfg_value(terminal_cfg, "ssh_host", "TERMINAL_SSH_HOST") user = _terminal_cfg_value(terminal_cfg, "ssh_user", "TERMINAL_SSH_USER") - missing = [] - if not host: - missing.append("terminal.ssh_host") - if not user: - missing.append("terminal.ssh_user") + missing = [k for k, v in (("terminal.ssh_host", host), ("terminal.ssh_user", user)) if not v] if missing: return ( "needs_setup", @@ -116,7 +88,7 @@ def _probe_ssh_backend(terminal_cfg: dict) -> tuple: return ("ready", f"{user}@{host}") -def _probe_modal_backend() -> tuple: +def _probe_modal_backend(_cfg) -> tuple: try: from tools.tool_backend_helpers import has_direct_modal_credentials @@ -124,109 +96,35 @@ def _probe_modal_backend() -> tuple: return ("ready", "") except Exception: pass - try: - from hermes_cli.config import get_env_value - - if get_env_value("MODAL_TOKEN_ID") and get_env_value("MODAL_TOKEN_SECRET"): - return ("ready", "") - except Exception: - pass + if _env_value("MODAL_TOKEN_ID") and _env_value("MODAL_TOKEN_SECRET"): + return ("ready", "") return ( "needs_setup", "Modal credentials not found — set MODAL_TOKEN_ID and MODAL_TOKEN_SECRET (or run `modal setup`).", ) -def _probe_daytona_backend() -> tuple: - try: - from hermes_cli.config import get_env_value - - if get_env_value("DAYTONA_API_KEY"): - return ("ready", "") - except Exception: - pass +def _probe_daytona_backend(_cfg) -> tuple: + if _env_value("DAYTONA_API_KEY"): + return ("ready", "") return ("needs_setup", "Set DAYTONA_API_KEY to use the Daytona backend.") -# Built-ins + plugin-registered backends, computed per request so a plugin -# installed after server start still shows up. -# Toolsets whose backends carry a selectable model catalog, mapped to the -# config.yaml section their `model` key lives in. Mirrors the CLI's -# post-selection model pickers (`_configure_imagegen_model_for_plugin` / -# `_configure_videogen_model_for_plugin` in tools_config.py). -_MODEL_CATALOG_TOOLSETS = { - "image_gen": "image_gen", - "video_gen": "video_gen", -} - - -def _resolve_toolset_model_plugin(ts_key: str, provider_row: dict) -> Optional[str]: - """Map a provider picker row to its model-catalog plugin name. - - Plugin-backed rows carry ``image_gen_plugin_name`` / ``video_gen_plugin_name``; - the managed "Nous Subscription" image row instead carries the legacy - ``imagegen_backend: "fal"`` marker (same underlying FAL catalog). - """ - if ts_key == "image_gen": - return provider_row.get("image_gen_plugin_name") or ( - "fal" if provider_row.get("imagegen_backend") else None - ) - if ts_key == "video_gen": - return provider_row.get("video_gen_plugin_name") - return None - - -def _toolset_model_catalog(ts_key: str, plugin_name: str): - """Return ``(catalog_dict, default_model)`` for a toolset's plugin backend.""" - from hermes_cli.tools_config import ( - _plugin_image_gen_catalog, - _plugin_video_gen_catalog, - ) - - if ts_key == "image_gen": - return _plugin_image_gen_catalog(plugin_name) - return _plugin_video_gen_catalog(plugin_name) - - -def _find_toolset_provider_row(ts_key: str, config: dict, provider: Optional[str]) -> Optional[dict]: - """Resolve a provider picker row by name, or the active row when omitted.""" - from hermes_cli.tools_config import ( - TOOL_CATEGORIES, - _is_provider_active, - _visible_providers, - ) - - cat = TOOL_CATEGORIES.get(ts_key) - if cat is None: - return None - rows = _visible_providers(cat, config, force_fresh=True) - if provider: - return next((p for p in rows if p.get("name") == provider), None) - return next( - (p for p in rows if _is_provider_active(p, config, force_fresh=True)), None - ) - - -def _terminal_backend_names() -> set: - """Valid ``terminal.backend`` values, including plugin backends.""" - return {row["name"] for row in _terminal_backend_rows()} +_BACKEND_PROBES = { + "local": lambda _cfg: ("ready", ""), + "docker": _probe_docker_backend, + "singularity": _probe_singularity_backend, + "ssh": _probe_ssh_backend, + "modal": _probe_modal_backend, + "daytona": _probe_daytona_backend} def _probe_terminal_backend(name: str, terminal_cfg: dict) -> tuple: """Return ``(status, detail)`` for one backend. Never raises.""" try: - if name == "local": - return ("ready", "") - if name == "docker": - return _probe_docker_backend() - if name == "singularity": - return _probe_singularity_backend() - if name == "ssh": - return _probe_ssh_backend(terminal_cfg) - if name == "modal": - return _probe_modal_backend() - if name == "daytona": - return _probe_daytona_backend() + probe = _BACKEND_PROBES.get(name) + if probe is not None: + return probe(terminal_cfg) try: from agent.terminal_env_registry import get_provider @@ -240,6 +138,59 @@ def _probe_terminal_backend(name: str, terminal_cfg: dict) -> tuple: return ("unavailable", f"Probe failed: {exc}") +# Toolsets whose backends carry a selectable model catalog, mapped to the +# config.yaml section their `model` key lives in. Mirrors the CLI's +# post-selection model pickers in tools_config.py. +_MODEL_CATALOG_TOOLSETS = {"image_gen": "image_gen", "video_gen": "video_gen"} + + +def _resolve_toolset_model_plugin(ts_key: str, provider_row: dict) -> Optional[str]: + """Map a provider picker row to its model-catalog plugin name. + + Plugin-backed rows carry ``image_gen_plugin_name`` / ``video_gen_plugin_name``; + the managed "Nous Subscription" image row instead carries the legacy + ``imagegen_backend: "fal"`` marker (same underlying FAL catalog). + """ + if ts_key == "image_gen": + return provider_row.get("image_gen_plugin_name") or ( + "fal" if provider_row.get("imagegen_backend") else None) + if ts_key == "video_gen": + return provider_row.get("video_gen_plugin_name") + return None + + +def _toolset_model_catalog(ts_key: str, plugin_name: str): + """Return ``(catalog_dict, default_model)`` for a toolset's plugin backend.""" + from hermes_cli.tools_config import _plugin_image_gen_catalog, _plugin_video_gen_catalog + + if ts_key == "image_gen": + return _plugin_image_gen_catalog(plugin_name) + return _plugin_video_gen_catalog(plugin_name) + + +def _category_providers(ts_key: str, config: dict) -> list: + """Visible provider rows for a toolset's category (fresh entitlement read).""" + from hermes_cli.tools_config import TOOL_CATEGORIES, _visible_providers + + cat = TOOL_CATEGORIES.get(ts_key) + return _visible_providers(cat, config, force_fresh=True) if cat else [] + + +def _find_toolset_provider_row(ts_key: str, config: dict, provider: Optional[str]) -> Optional[dict]: + """Resolve a provider picker row by name, or the active row when omitted.""" + from hermes_cli.tools_config import _is_provider_active + + rows = _category_providers(ts_key, config) + if provider: + return next((p for p in rows if p.get("name") == provider), None) + return next((p for p in rows if _is_provider_active(p, config, force_fresh=True)), None) + + +def _terminal_backend_names() -> set: + """Valid ``terminal.backend`` values, including plugin backends.""" + return {row["name"] for row in _terminal_backend_rows()} + + def _require_known_toolset(name: str) -> None: """400 for toolset keys outside the effective configurable set.""" from hermes_cli.tools_config import _get_effective_configurable_toolsets @@ -261,6 +212,13 @@ def _no_models(name: str) -> dict: return {"name": name, "has_models": False, "models": [], "current": None, "default": None} +def _model_catalog_section(name: str) -> str: + section = _MODEL_CATALOG_TOOLSETS.get(name) + if section is None: + raise HTTPException(status_code=400, detail=f"Toolset has no model catalog: {name}") + return section + + @router.get("/api/tools/toolsets") async def get_toolsets(profile: Optional[str] = None): from hermes_cli.tools_config import ( @@ -270,8 +228,7 @@ async def get_toolsets(profile: Optional[str] = None): _toolset_configuration_platform, _toolset_has_keys, get_nous_subscription_features, - gui_toolset_label, - ) + gui_toolset_label) from hermes_cli.platforms import platform_label from toolsets import resolve_toolset @@ -279,17 +236,10 @@ async def get_toolsets(profile: Optional[str] = None): with _profile_scope(profile): config = load_config() toolset_rows = _get_effective_configurable_toolsets() - target_platforms = { - _toolset_configuration_platform(name) for name, _, _ in toolset_rows - } + target_platforms = {_toolset_configuration_platform(name) for name, _, _ in toolset_rows} enabled_by_platform = { - platform: _get_platform_tools( - config, - platform, - include_default_mcp_servers=False, - ) - for platform in target_platforms - } + platform: _get_platform_tools(config, platform, include_default_mcp_servers=False) + for platform in target_platforms} features = get_nous_subscription_features(config) return config, toolset_rows, enabled_by_platform, features @@ -316,14 +266,11 @@ async def get_toolsets(profile: Optional[str] = None): "label": gui_toolset_label(label), "description": desc, "platform": target_platform, - "platform_label": gui_toolset_label( - platform_label(target_platform, target_platform) - ), + "platform_label": gui_toolset_label(platform_label(target_platform, target_platform)), "enabled": is_enabled, "available": is_enabled, "configured": _toolset_has_keys(name, config, features=features), - "tools": tools, - }) + "tools": tools}) return result @@ -336,14 +283,14 @@ async def toggle_toolset(name: str, body: ToolsetToggle, profile: Optional[str] _CONFIG_ONLY_TOOLSETS, _get_platform_tools, _save_platform_tools, - _toolset_configuration_platform, - ) + _toolset_configuration_platform) _require_known_toolset(name) target_platform = _toolset_configuration_platform(name) + scope_profile = body.profile or profile def _run(): - with config_write_scope(body.profile or profile): + with config_write_scope(scope_profile): config = load_config() if name in _CONFIG_ONLY_TOOLSETS: # Config-only capabilities (stt) toggle their own section's @@ -351,13 +298,7 @@ async def toggle_toolset(name: str, body: ToolsetToggle, profile: Optional[str] _dict_section(config, name)["enabled"] = bool(body.enabled) save_config(config) return - enabled = set( - _get_platform_tools( - config, - target_platform, - include_default_mcp_servers=False, - ) - ) + enabled = set(_get_platform_tools(config, target_platform, include_default_mcp_servers=False)) if body.enabled: enabled.add(name) else: @@ -375,15 +316,12 @@ async def toggle_toolset(name: str, body: ToolsetToggle, profile: Optional[str] if body.enabled and name not in _CONFIG_ONLY_TOOLSETS: def _pending_install_key() -> Optional[str]: from hermes_cli.tools_config import ( - TOOL_CATEGORIES, - _post_setup_already_installed, - _visible_providers, - ) + TOOL_CATEGORIES, _post_setup_already_installed, _visible_providers) cat = TOOL_CATEGORIES.get(name) if not cat: return None - with _profile_scope(body.profile or profile): + with _profile_scope(scope_profile): config = load_config() for prov in _visible_providers(cat, config): key = prov.get("post_setup") @@ -395,10 +333,8 @@ async def toggle_toolset(name: str, body: ToolsetToggle, profile: Optional[str] pending_key = await asyncio.to_thread(_pending_install_key) if pending_key: _spawn_hermes_action( - _profile_cli_args(body.profile or profile) - + ["tools", "post-setup", pending_key], - "tools-post-setup", - ) + _profile_cli_args(scope_profile) + ["tools", "post-setup", pending_key], + "tools-post-setup") post_setup_started = pending_key except Exception: _log.exception("install-on-enable post-setup spawn failed for %s", name) @@ -408,8 +344,7 @@ async def toggle_toolset(name: str, body: ToolsetToggle, profile: Optional[str] "name": name, "platform": target_platform, "enabled": body.enabled, - "post_setup_started": post_setup_started, - } + "post_setup_started": post_setup_started} @router.get("/api/tools/toolsets/{name}/config") @@ -423,8 +358,7 @@ async def get_toolset_config(name: str, profile: Optional[str] = None): _is_provider_active, _visible_providers, provider_readiness_status, - web_provider_capabilities, - ) + web_provider_capabilities) from hermes_cli.config import get_env_value from hermes_cli.nous_subscription import get_nous_subscription_features @@ -448,10 +382,8 @@ async def get_toolset_config(name: str, profile: Optional[str] = None): "prompt": e.get("prompt", e["key"]), "url": e.get("url"), "default": e.get("default"), - "is_set": bool(get_env_value(e["key"])), - } - for e in prov.get("env_vars", []) - ] + "is_set": bool(get_env_value(e["key"]))} + for e in prov.get("env_vars", [])] # Same active-provider determination as the CLI picker, so the # GUI highlights the provider actually written to config. is_active = _is_provider_active(prov, config, force_fresh=True) @@ -469,9 +401,7 @@ async def get_toolset_config(name: str, profile: Optional[str] = None): # automatically ready (logged-out Nous rows, never-run # post_setup installs). "status": provider_readiness_status( - prov, config, features=features, is_active=is_active - ), - } + prov, config, features=features, is_active=is_active)} if name == "web" and prov.get("web_backend"): # web is two capabilities (search/extract); surface each # row's backend key + capabilities for per-capability selection. @@ -501,8 +431,7 @@ async def get_toolset_config(name: str, profile: Optional[str] = None): "name": name, "has_category": cat is not None, "providers": providers, - "active_provider": active_provider, - } + "active_provider": active_provider} if name == "web": payload["active_search_backend"] = active_search_backend payload["active_extract_backend"] = active_extract_backend @@ -511,8 +440,7 @@ async def get_toolset_config(name: str, profile: Optional[str] = None): @router.get("/api/tools/toolsets/{name}/models") async def get_toolset_models( - name: str, provider: Optional[str] = None, profile: Optional[str] = None -): + name: str, provider: Optional[str] = None, profile: Optional[str] = None): """Model catalog for a toolset backend (image/video gen) — the GUI counterpart of the CLI model picker. ``provider`` names a picker row (default: the active provider); no catalog -> ``has_models: false``.""" @@ -550,10 +478,8 @@ async def get_toolset_models( "display": meta.get("display", model_id), "speed": meta.get("speed", ""), "strengths": meta.get("strengths", ""), - "price": meta.get("price", ""), - } - for model_id, meta in catalog.items() - ] + "price": meta.get("price", "")} + for model_id, meta in catalog.items()] return { "name": name, "has_models": bool(models), @@ -561,22 +487,15 @@ async def get_toolset_models( "plugin": plugin, "models": models, "current": current, - "default": default_model, - } + "default": default_model} @router.put("/api/tools/toolsets/{name}/model") async def select_toolset_model( - name: str, body: ToolsetModelSelect, profile: Optional[str] = None -): + name: str, body: ToolsetModelSelect, profile: Optional[str] = None): """Persist a backend model selection (``image_gen.model`` / ``video_gen.model``), validated against the resolved backend's catalog.""" - section = _MODEL_CATALOG_TOOLSETS.get(name) - if section is None: - raise HTTPException( - status_code=400, detail=f"Toolset has no model catalog: {name}" - ) - + section = _model_catalog_section(name) model_id = (body.model or "").strip() if not model_id: raise HTTPException(status_code=400, detail="model is required") @@ -588,16 +507,12 @@ async def select_toolset_model( plugin = _resolve_toolset_model_plugin(name, row) if row else None if not plugin: raise HTTPException( - status_code=400, - detail=f"No model-capable backend is active for {name}", - ) + status_code=400, detail=f"No model-capable backend is active for {name}") catalog, _default = _toolset_model_catalog(name, plugin) if model_id not in catalog: raise HTTPException( - status_code=400, - detail=f"Unknown model {model_id!r} for backend {plugin!r}", - ) + status_code=400, detail=f"Unknown model {model_id!r} for backend {plugin!r}") _dict_section(config, section)["model"] = model_id save_config(config) @@ -609,8 +524,7 @@ async def select_toolset_model( @router.put("/api/tools/toolsets/{name}/provider") async def select_toolset_provider( - name: str, body: ToolsetProviderSelect, profile: Optional[str] = None -): + name: str, body: ToolsetProviderSelect, profile: Optional[str] = None): """Persist a provider selection via ``apply_provider_selection`` (the non-interactive core shared with ``hermes tools``, so both write identical config keys). Keys and post-setup are separate endpoints. @@ -625,35 +539,25 @@ async def select_toolset_provider( response adds ``needs_nous_auth: true`` + ``feature`` so the client can drive the Nous Portal OAuth flow and refetch. """ - from hermes_cli.tools_config import ( - TOOL_CATEGORIES, - apply_provider_selection, - web_provider_capabilities, - _visible_providers, - ) + from hermes_cli.tools_config import apply_provider_selection, web_provider_capabilities from hermes_cli.nous_subscription import ( - MANAGED_FEATURE_COVERAGE_CATEGORY, - get_nous_subscription_features, - ) + MANAGED_FEATURE_COVERAGE_CATEGORY, get_nous_subscription_features) _require_known_toolset(name) if body.capability is not None: if name != "web": raise HTTPException( - status_code=400, - detail="capability selection is only supported for the web toolset", + status_code=400, detail="capability selection is only supported for the web toolset" ) if body.capability not in ("search", "extract"): raise HTTPException( status_code=400, - detail=f"Unknown capability: {body.capability!r} (expected 'search' or 'extract')", - ) + detail=f"Unknown capability: {body.capability!r} (expected 'search' or 'extract')") def _provider_row(config): - cat = TOOL_CATEGORIES.get(name) - providers = _visible_providers(cat, config, force_fresh=True) if cat else [] - return next((p for p in providers if p.get("name") == body.provider), None) + return next( + (p for p in _category_providers(name, config) if p.get("name") == body.provider), None) def _run(): with _profile_scope(body.profile or profile): @@ -667,19 +571,16 @@ async def select_toolset_provider( if prov is None: raise HTTPException( status_code=400, - detail=f"Unknown provider {body.provider!r} for toolset {name!r}", - ) + detail=f"Unknown provider {body.provider!r} for toolset {name!r}") backend = prov.get("web_backend") if not backend: raise HTTPException( status_code=400, - detail=f"Provider {body.provider!r} has no web backend key", - ) + detail=f"Provider {body.provider!r} has no web backend key") if body.capability not in web_provider_capabilities(backend): raise HTTPException( status_code=400, - detail=f"{body.provider} does not support {body.capability}", - ) + detail=f"{body.provider} does not support {body.capability}") _dict_section(config, "web")[f"{body.capability}_backend"] = backend else: try: @@ -706,9 +607,7 @@ async def select_toolset_provider( and ( acct.tool_gateway_entitled_for(category) if category - else acct.tool_gateway_entitled - ) - ) + else acct.tool_gateway_entitled)) if not entitled: response["needs_nous_auth"] = True response["feature"] = managed_feature @@ -722,7 +621,6 @@ async def save_toolset_env(name: str, body: ToolsetEnvUpdate, profile: Optional[ """Persist API keys to ``.env`` via ``save_env_value``. Keys are validated against the union of the category's visible-provider ``env_vars`` so this can't write arbitrary env vars; a blank value means "leave unchanged".""" - from hermes_cli.tools_config import TOOL_CATEGORIES, _visible_providers from hermes_cli.config import get_env_value, save_env_value _require_known_toolset(name) @@ -730,19 +628,16 @@ async def save_toolset_env(name: str, body: ToolsetEnvUpdate, profile: Optional[ def _run(): with _profile_scope(body.profile or profile): config = load_config() - cat = TOOL_CATEGORIES.get(name) - allowed: set[str] = set() - if cat: - for prov in _visible_providers(cat, config, force_fresh=True): - for e in prov.get("env_vars", []): - allowed.add(e["key"]) + allowed: set[str] = { + e["key"] + for prov in _category_providers(name, config) + for e in prov.get("env_vars", [])} unknown = [k for k in body.env if k not in allowed] if unknown: raise HTTPException( status_code=400, - detail=f"Unknown env var(s) for toolset {name}: {', '.join(sorted(unknown))}", - ) + detail=f"Unknown env var(s) for toolset {name}: {', '.join(sorted(unknown))}") saved: List[str] = [] skipped: List[str] = [] @@ -765,8 +660,7 @@ async def save_toolset_env(name: str, body: ToolsetEnvUpdate, profile: Optional[ @router.post("/api/tools/toolsets/{name}/post-setup") async def run_toolset_post_setup( - name: str, body: ToolsetPostSetup, profile: Optional[str] = None -): + name: str, body: ToolsetPostSetup, profile: Optional[str] = None): """Spawn ``hermes tools post-setup `` (long-running installs: npm, pip, cua-driver fetch) as a background action the frontend tails via ``GET /api/actions/tools-post-setup/status``. The key is validated @@ -776,17 +670,14 @@ async def run_toolset_post_setup( _require_known_toolset(name) if body.key not in valid_post_setup_keys(): - raise HTTPException( - status_code=400, detail=f"Unknown post-setup key: {body.key}" - ) + raise HTTPException(status_code=400, detail=f"Unknown post-setup key: {body.key}") result = spawn_profile_action( body.profile or profile, ["tools", "post-setup", body.key], "tools-post-setup", log_msg="Failed to spawn tools post-setup", - prefix="Failed to run post-setup", - ) + prefix="Failed to run post-setup") result["key"] = body.key return result @@ -816,8 +707,7 @@ async def get_terminal_backends(profile: Optional[str] = None): "description": row["description"], "active": row["name"] == active, "status": status, - "detail": detail, - }) + "detail": detail}) return {"active": active, "backends": backends} return await asyncio.to_thread(_read) @@ -825,8 +715,7 @@ async def get_terminal_backends(profile: Optional[str] = None): @router.put("/api/tools/terminal/backend") async def select_terminal_backend( - body: TerminalBackendSelect, profile: Optional[str] = None -): + body: TerminalBackendSelect, profile: Optional[str] = None): """Persist ``terminal.backend``. A backend that still needs setup is allowed — the picker shows guidance instead of blocking, like the CLI.""" backend = (body.backend or "").strip().lower() @@ -835,8 +724,7 @@ async def select_terminal_backend( raise HTTPException( status_code=400, detail=f"Unknown terminal backend: {body.backend!r}. " - f"Use one of: {', '.join(sorted(valid_names))}", - ) + f"Use one of: {', '.join(sorted(valid_names))}") def _run(): with config_write_scope(body.profile or profile): @@ -864,13 +752,10 @@ async def grant_computer_use_permissions(profile: Optional[str] = None): The frontend polls ``GET /api/actions/computer-use-grant/status``.""" if sys.platform != "darwin": raise HTTPException( - status_code=400, - detail="Computer Use permission grants are a macOS concept.", - ) + status_code=400, detail="Computer Use permission grants are a macOS concept.") return spawn_profile_action( profile, ["computer-use", "permissions", "grant"], "computer-use-grant", log_msg="Failed to spawn computer-use permissions grant", - prefix="Failed to request permissions", - ) + prefix="Failed to request permissions") diff --git a/hermes_cli/web_server_chat.py b/hermes_cli/web_server_chat.py index 53a21fdc32..c5e51f67d1 100644 --- a/hermes_cli/web_server_chat.py +++ b/hermes_cli/web_server_chat.py @@ -1,4 +1,5 @@ -"""Chat/terminal WebSocket plumbing: PTY bridge selection and registry, WS client/origin/auth gates, chat argv resolution, gateway/sidecar URL building. +"""Chat/terminal WebSocket plumbing: PTY bridge selection and registry, WS +client/origin/auth gates, chat argv resolution, gateway/sidecar URL building. Split out of ``hermes_cli.web_server``; every externally used name is re-imported there, so ``web_server.`` keeps resolving (and monkeypatching) as before. @@ -25,24 +26,15 @@ from hermes_cli.pty_session import PtySessionRegistry _log = logging.getLogger("hermes_cli.web_server") -# --------------------------------------------------------------------------- -# /api/pty — PTY-over-WebSocket bridge for the dashboard "Chat" tab. -# -# The endpoint spawns the same ``hermes --tui`` binary the CLI uses, behind -# a POSIX pseudo-terminal, and forwards bytes + resize escapes across a -# WebSocket. The browser renders the ANSI through xterm.js (see -# web/src/pages/ChatPage.tsx). -# -# Auth: ``?token=`` query param (browsers can't set -# Authorization on the WS upgrade). Same ephemeral ``_SESSION_TOKEN`` as -# REST. Localhost-only — we defensively reject non-loopback clients even -# though uvicorn binds to 127.0.0.1. -# --------------------------------------------------------------------------- +# /api/pty — PTY-over-WebSocket bridge for the dashboard "Chat" tab: spawns the +# same ``hermes --tui`` binary the CLI uses behind a pseudo-terminal and forwards +# bytes + resize escapes; the browser renders the ANSI through xterm.js. +# Auth: ``?token=`` query param (browsers can't set Authorization +# on the WS upgrade), same ephemeral ``_SESSION_TOKEN`` as REST. # PTY bridge: POSIX uses pty_bridge (fcntl/termios/ptyprocess); native Windows -# uses win_pty_bridge (pywinpty/ConPTY, already a declared dependency). Both -# expose the same public surface — spawn/read/write/resize/close/is_available — -# so the /api/pty WebSocket handler needs no platform guards. +# uses win_pty_bridge (pywinpty/ConPTY). Both expose the same surface — +# spawn/read/write/resize/close/is_available — so the handler needs no guards. if sys.platform.startswith("win"): try: from hermes_cli.win_pty_bridge import WinPtyBridge as PtyBridge, PtyUnavailableError @@ -53,7 +45,6 @@ if sys.platform.startswith("win"): class PtyUnavailableError(RuntimeError): # type: ignore[no-redef] """Stub when win_pty_bridge cannot be imported.""" - pass else: try: from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError @@ -64,40 +55,25 @@ else: class PtyUnavailableError(RuntimeError): # type: ignore[no-redef] """Stub on platforms where pty_bridge can't be imported.""" - pass _RESIZE_RE = re.compile(rb"\x1b\[RESIZE:(\d+);(\d+)\]") _PTY_READ_CHUNK_TIMEOUT = 0.2 -# Back-off delay between idle PTY reads so a quiet terminal does not spin -# the event loop. A positive sleep lets other coroutines run and keeps -# dashboard idle CPU low (#42627). +# Back-off between idle PTY reads so a quiet terminal does not spin the event +# loop (keeps dashboard idle CPU low). _PTY_IDLE_BACKOFF = 0.05 PTY_REGISTRY = PtySessionRegistry( - ttl=30 * 60, - max_sessions=16, - buffer_cap=1 * 1024 * 1024, - read_timeout=_PTY_READ_CHUNK_TIMEOUT, -) + ttl=30 * 60, max_sessions=16, buffer_cap=1 * 1024 * 1024, read_timeout=_PTY_READ_CHUNK_TIMEOUT) async def _legacy_pump(ws: "WebSocket", bridge) -> None: """Original 1:1 socket<->PTY pump: stream until disconnect, then close the - bridge. Used when no ``?attach=`` token is supplied (keep-alive opt-in). - - Behavior is identical to the pre-keep-alive ``pty_ws`` body, including the - #54028 half-open-socket protection (reader EOF → close the WS so the - writer's ``ws.receive()`` unparks) and the #53227 ``to_thread`` offloads - for the blocking ``bridge.close()``. - """ + bridge. Used when no ``?attach=`` token is supplied (keep-alive opt-in).""" loop = asyncio.get_running_loop() - # --- reader task: PTY master → WebSocket ---------------------------- async def pump_pty_to_ws() -> None: try: while True: - chunk = await loop.run_in_executor( - None, bridge.read, _PTY_READ_CHUNK_TIMEOUT - ) + chunk = await loop.run_in_executor(None, bridge.read, _PTY_READ_CHUNK_TIMEOUT) if chunk is None: # EOF return if not chunk: # no data this tick; yield control and retry @@ -108,19 +84,13 @@ async def _legacy_pump(ws: "WebSocket", bridge) -> None: except Exception: return finally: - # The child has exited (EOF) or the send side broke. Close the - # WebSocket so the writer loop's ``ws.receive()`` returns instead - # of blocking forever — otherwise, when the browser's socket is - # half-open (no FIN delivered, common on macOS/launchd) the - # handler never reaches its ``finally`` and the PTY's fds leak. - # With dashboard auto-reconnect (#52962) every dropped socket then - # stacks a fresh PTY on top of the orphaned one, exhausting fds. - # - # Reap the bridge here too (close() is idempotent): on child EOF the - # writer loop's ``finally`` is the usual closer, but if the handler - # task is cancelled the instant we close the WS, that ``finally`` - # can be skipped, leaking the PTY. Closing from the EOF path makes - # the reap independent of that cancellation race (#54028). + # Child exited (EOF) or the send side broke. Close the WebSocket so + # the writer loop's ``ws.receive()`` returns instead of blocking + # forever on a half-open browser socket (no FIN, common on + # macOS/launchd) — otherwise the PTY's fds leak and auto-reconnect + # stacks a fresh PTY on each orphan. Reap the bridge here too + # (close() is idempotent): if the handler task is cancelled the + # instant we close the WS, the writer's ``finally`` can be skipped. try: await asyncio.to_thread(bridge.close) except Exception: @@ -132,14 +102,13 @@ async def _legacy_pump(ws: "WebSocket", bridge) -> None: reader_task = asyncio.create_task(pump_pty_to_ws()) - # --- writer loop: WebSocket → PTY master ---------------------------- try: while True: try: msg = await ws.receive() except RuntimeError: - # Raised when ws.receive() is called after the socket is - # already disconnected (e.g. closed by the reader task above). + # ws.receive() after the socket is already disconnected + # (e.g. closed by the reader task above). break if msg.get("type") == "websocket.disconnect": break @@ -172,13 +141,16 @@ _LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost", "testclient"}) def _ws_client_reason(ws: "WebSocket") -> Optional[str]: - """Return a rejection reason for the client IP, or None when allowed. + """Return a rejection reason token for the peer IP, or None when allowed. - Reasons are short machine-parseable tokens logged on the rejection path - so a "WS keeps closing" report can be diagnosed from agent.log without a - repro. ``None`` means the peer IP passed this gate. - - See :func:`_ws_client_is_allowed` for the full policy rationale. + Loopback bind: only loopback clients — the legacy ``?token=`` is the only + auth, so LAN hosts must not get to guess it. Explicit non-loopback bind + (``--host 0.0.0.0``/``::``/LAN IP, always with ``--insecure``): any peer; + DNS-rebinding is still blocked by :func:`_ws_host_origin_reason`. Gated + mode: any peer — ``proxy_headers=True`` rewrites ``ws.client.host`` to the + X-Forwarded-For value and the OAuth gate + ``?ticket=`` is the auth. + An empty peer on a loopback bind fails closed (misconfigured proxy / unix + socket must not reach a loopback-only surface). """ from hermes_cli.web_server import app if getattr(app.state, "auth_required", False): @@ -188,11 +160,6 @@ def _ws_client_reason(ws: "WebSocket") -> Optional[str]: return None client_host = ws.client.host if ws.client else "" if not client_host: - # Fail-closed: a loopback-bound dashboard with auth disabled must - # not accept a WebSocket with no identifiable peer. ASGI servers - # behind a misconfigured proxy or unix socket can deliver - # ws.client == None or "" — treating that as "allowed" would let - # an unidentified peer reach a loopback-only surface. return f"missing_or_empty_peer bound={bound_host or '?'}" if client_host in _LOOPBACK_HOSTS: return None @@ -200,69 +167,29 @@ def _ws_client_reason(ws: "WebSocket") -> Optional[str]: def _ws_client_is_allowed(ws: "WebSocket") -> bool: - """Check if the WebSocket client IP is acceptable. - - Loopback bind: only loopback clients allowed — the legacy - ``?token=<_SESSION_TOKEN>`` path is the only auth we have, so we - don't want LAN hosts guessing tokens. - - Explicit non-loopback bind (``--host 0.0.0.0``, ``--host ::``, or a - specific address such as a Tailscale/LAN IP, always with - ``--insecure``): allow any peer. The operator explicitly opted into - non-loopback exposure, so the loopback-only peer restriction does not - apply. DNS-rebinding is still blocked by the Host/Origin guard in - :func:`_ws_host_origin_is_allowed`, which mirrors the HTTP layer and - requires the Host header to match the bound interface — the same - defence ``_is_accepted_host`` applies to non-loopback HTTP requests. - - Gated mode: any peer is allowed — uvicorn's ``proxy_headers=True`` - (enabled when the OAuth gate is active so cookies can pick up - ``X-Forwarded-Proto``) rewrites ``ws.client.host`` to the - X-Forwarded-For value, which is the real internet client IP. The - OAuth gate + single-use ``?ticket=`` is the auth at that point; the - Host/Origin guard in :func:`_ws_host_origin_is_allowed` is what - blocks DNS-rebinding here, not the peer IP. - """ - from hermes_cli.web_server import app - if getattr(app.state, "auth_required", False): - return True - # Any explicit non-loopback bind (0.0.0.0, ::, or a specific LAN / - # Tailscale address) means the operator opted into non-loopback - # access via --insecure. The loopback-only peer gate only applies to - # an actual loopback bind; otherwise the WS handshake is rejected even - # though same-bind HTTP requests pass _is_accepted_host. - bound_host = (getattr(app.state, "bound_host", "") or "").strip().lower() - if bound_host and bound_host not in _LOOPBACK_HOSTS: - return True - client_host = ws.client.host if ws.client else "" - if not client_host: - # Fail-closed: see _ws_client_reason for rationale. An empty - # client_host on a loopback-bound dashboard with auth disabled - # must be rejected, not accepted as a default-allow. - return False - return client_host in _LOOPBACK_HOSTS + """True when the peer IP passes :func:`_ws_client_reason`.""" + return _ws_client_reason(ws) is None def _ws_host_origin_reason(ws: "WebSocket") -> Optional[str]: - """Return a Host/Origin rejection reason, or None when allowed. + """Return a Host/Origin rejection reason (``host_mismatch …`` / + ``origin_mismatch …``), or None when allowed. - Mirrors :func:`_ws_host_origin_is_allowed` but yields a short - machine-parseable token (``host_mismatch …`` / ``origin_mismatch …``) - on rejection so the close path can log *why* the upgrade was refused. + HTTP middleware does not run for WebSocket routes, so the DNS-rebinding + Host check is repeated here before accepting the upgrade; a browser Origin + header, when present, must target the same bound host. Non-web origins + (packaged Electron: file://, null, app://) are trusted — the upstream + credential check is the real auth boundary there. """ from hermes_cli.web_server import _is_accepted_host, app bound_host = getattr(app.state, "bound_host", None) if not bound_host: return None - trusted_public_hosts = getattr( - app.state, "trusted_public_hosts", frozenset() - ) + trusted_public_hosts = getattr(app.state, "trusted_public_hosts", frozenset()) host_header = ws.headers.get("host", "") - if not _is_accepted_host( - host_header, bound_host, trusted_public_hosts - ): + if not _is_accepted_host(host_header, bound_host, trusted_public_hosts): return f"host_mismatch host={host_header or '?'} bound={bound_host}" origin = ws.headers.get("origin", "") @@ -271,30 +198,15 @@ def _ws_host_origin_reason(ws: "WebSocket") -> Optional[str]: parsed = urllib.parse.urlparse(origin) if parsed.scheme not in {"http", "https"}: - # Non-web origin (packaged Electron: file://, null, app://). The - # upstream credential check is the real auth boundary; trust it. - # See _ws_host_origin_is_allowed for the full rationale. return None - if not parsed.netloc: - return f"origin_mismatch origin={origin} bound={bound_host}" - - if not _is_accepted_host( - parsed.netloc, bound_host, trusted_public_hosts - ): + if not parsed.netloc or not _is_accepted_host(parsed.netloc, bound_host, trusted_public_hosts): return f"origin_mismatch origin={origin} bound={bound_host}" return None def _ws_host_origin_is_allowed(ws: "WebSocket") -> bool: - """Apply the dashboard Host/Origin guard to WebSocket upgrades. - - FastAPI HTTP middleware does not run for WebSocket routes, so the - DNS-rebinding Host check used for normal dashboard HTTP requests must be - repeated here before accepting the upgrade. Browsers also send an Origin - header on WebSocket handshakes; when present, require it to target the - same bound dashboard host. - """ + """True when the upgrade passes the dashboard Host/Origin guard.""" from hermes_cli.web_server import _ws_host_origin_reason return _ws_host_origin_reason(ws) is None @@ -313,9 +225,7 @@ def _gateway_ws_ticket_from_subprotocol(ws: "WebSocket") -> tuple[str, str]: raw = str(ws.headers.get("sec-websocket-protocol", "") or "") protocols = [value.strip() for value in raw.split(",") if value.strip()] ticket_protocols = [ - value for value in protocols - if value.startswith(_GATEWAY_WS_TICKET_PROTOCOL_PREFIX) - ] + value for value in protocols if value.startswith(_GATEWAY_WS_TICKET_PROTOCOL_PREFIX)] if not ticket_protocols: return "", "none" if _GATEWAY_WS_PROTOCOL not in protocols or len(ticket_protocols) != 1: @@ -327,34 +237,19 @@ def _gateway_ws_ticket_from_subprotocol(ws: "WebSocket") -> tuple[str, str]: def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]: """Validate WS-upgrade auth; return ``(reason, credential)``. - ``reason`` is None when the credential is accepted, else a short - machine-parseable token explaining the rejection (``no_credential``, - ``token_mismatch``, ``ticket_invalid``, ``internal_invalid``). - ``credential`` names which credential type was presented (``ticket``, - ``internal``, ``token``, or ``none``) so the accepted path can log *how* - a peer authed, not just that it did. + ``reason`` is None when accepted, else ``no_credential`` / ``token_mismatch`` + / ``ticket_invalid`` / ``internal_invalid``; ``credential`` names what was + presented (``ticket``, ``ticket-subprotocol``, ``internal``, ``token``, + ``none``) so the accept path can log *how* a peer authed. - Loopback / ``--insecure``: legacy ``?token=<_SESSION_TOKEN>`` query - parameter, constant-time compared. - - Gated (public bind, no ``--insecure``): one of two credentials — - - * ``?ticket=`` — a browser-minted, single-use, 30s-TTL ticket - consumed against the dashboard-auth ticket store. This is what the SPA - (and native clients) use. - * ``?internal=`` — the process-lifetime internal - credential, used only by WS clients the server spawns itself (the - embedded-TUI PTY child attaching to ``/api/ws`` and ``/api/pub``). It - is multi-use and never expires so the child can reconnect, and is never - injected into the SPA — see ``dashboard_auth.ws_tickets`` for the - threat model. - - The legacy ``?token=`` path is unconditionally rejected in gated mode - (the SPA bundle isn't carrying the token any longer, and a leaked - ``_SESSION_TOKEN`` must not grant WS access once the gate is engaged). - - Audit-logs the rejection so operators can debug "WS keeps closing" - issues from the log. + Loopback / ``--insecure``: legacy ``?token=<_SESSION_TOKEN>``, constant-time + compared. Gated: ``?ticket=`` (browser-minted, single-use, 30s TTL) or + ``?internal=`` (process-lifetime credential used only by WS clients the + server spawns itself — multi-use so the PTY child can reconnect; never + injected into the SPA, see ``dashboard_auth.ws_tickets``). The legacy + token is unconditionally rejected in gated mode: a leaked ``_SESSION_TOKEN`` + must not grant WS access once the gate is engaged. Rejections are + audit-logged so "WS keeps closing" can be debugged from the log. """ from hermes_cli.web_server import _SESSION_TOKEN, app auth_required = bool(getattr(app.state, "auth_required", False)) @@ -363,35 +258,31 @@ def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]: # that don't bring in the dashboard_auth layer. from hermes_cli.dashboard_auth.audit import AuditEvent, audit_log from hermes_cli.dashboard_auth.ws_tickets import ( - TicketInvalid, - consume_internal_credential, - consume_ticket, - ) + TicketInvalid, consume_internal_credential, consume_ticket) + + def _reject(reason: str) -> None: + audit_log( + AuditEvent.WS_TICKET_REJECTED, + reason=reason, + ip=(ws.client.host if ws.client else ""), + path=ws.url.path) + + def _stamp_identity(info) -> None: + # Server-minted {user_id, provider} stamped onto the WS object is the + # sole identity authority downstream (gateway transport / controller + # registration); a client can never supply it through RPC params. + # Only the two identity fields are carried — bookkeeping such as + # ``minted_at`` is not part of the identity contract. + ws._hermes_auth_identity = { + "user_id": info.get("user_id"), "provider": info.get("provider")} - # Server-spawned children (PTY child → /api/ws, /api/pub) present the - # multi-use internal credential rather than a single-use ticket, so - # they survive reconnects and slow cold boots. internal = ws.query_params.get("internal", "") if internal: try: - info = consume_internal_credential(internal) - # Stamp the server-minted identity onto the WS object so the - # connection (and any transport built from it) can never be - # impersonated by RPC params. Internal peers are marked - # ``server-internal`` and are excluded from privileged - # controller registration downstream. - ws._hermes_auth_identity = { - "user_id": info.get("user_id"), - "provider": info.get("provider"), - } + _stamp_identity(consume_internal_credential(internal)) return None, "internal" except TicketInvalid as exc: - audit_log( - AuditEvent.WS_TICKET_REJECTED, - reason=f"internal: {exc}", - ip=(ws.client.host if ws.client else ""), - path=ws.url.path, - ) + _reject(f"internal: {exc}") return "internal_invalid", "internal" protocol_ticket, protocol_reason = _gateway_ws_ticket_from_subprotocol(ws) @@ -402,18 +293,7 @@ def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]: return "no_credential", "none" try: - info = consume_ticket(ticket) - # The ticket binds a server-minted {user_id, provider}; stamp it - # onto the WS object so ``gateway_ws`` can hand it to the gateway - # transport, where it is the sole identity authority for - # browser-controller registration. A client can never supply or - # spoof this value through RPC params. Only the two identity - # fields are carried — bookkeeping (e.g. ``minted_at``) is not - # part of the identity contract. - ws._hermes_auth_identity = { - "user_id": info.get("user_id"), - "provider": info.get("provider"), - } + _stamp_identity(consume_ticket(ticket)) if protocol_ticket: # Select only the stable public protocol during accept. The # ticket-bearing protocol is a credential and must never be @@ -422,12 +302,7 @@ def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]: return None, "ticket-subprotocol" return None, "ticket" except TicketInvalid as exc: - audit_log( - AuditEvent.WS_TICKET_REJECTED, - reason=str(exc), - ip=(ws.client.host if ws.client else ""), - path=ws.url.path, - ) + _reject(str(exc)) return "ticket_invalid", "ticket" token = ws.query_params.get("token", "") @@ -444,61 +319,34 @@ def _ws_auth_ok(ws: "WebSocket") -> bool: return _ws_auth_reason(ws)[0] is None -# Per-channel subscriber registry used by /api/pub (PTY-side gateway → dashboard) -# and /api/events (dashboard → browser sidebar). Keyed by an opaque channel id -# the chat tab generates on mount; entries auto-evict when the last subscriber -# drops AND the publisher has disconnected. -# (Channel state and the chat-argv lock are initialised in _lifespan on app -# startup — see _get_event_state / _get_chat_argv_lock above.) - - def _resolve_chat_argv( resume: Optional[str] = None, sidecar_url: Optional[str] = None, profile: Optional[str] = None, - active_session_file: Optional[str] = None, -) -> tuple[list[str], Optional[str], Optional[dict]]: - """Resolve the argv + cwd + env for the chat PTY. + active_session_file: Optional[str] = None) -> tuple[list[str], Optional[str], Optional[dict]]: + """Resolve the argv + cwd + env for the chat PTY (what ``hermes --tui`` runs). - Default: whatever ``hermes --tui`` would run. Tests monkeypatch this - function to inject a tiny fake command (``cat``, ``sh -c 'printf …'``) - so nothing has to build Node or the TUI bundle. + Tests monkeypatch this to inject a tiny fake command so nothing has to + build the TUI bundle. Env contract with the child: - Session resume is propagated via the ``HERMES_TUI_RESUME`` env var — - matching what ``hermes_cli.main._launch_tui`` does for the CLI path. - Appending ``--resume `` to argv doesn't work because ``ui-tui`` does - not parse its argv. - - ``HERMES_TUI_GATEWAY_URL`` is injected so the PTY child can attach to - this process's in-memory ``tui_gateway`` instance instead of spawning - its own Python gateway subprocess. - - `sidecar_url` (when set) is forwarded as ``HERMES_TUI_SIDECAR_URL`` so - the spawned ``tui_gateway.entry`` can mirror dispatcher emits to the - dashboard's ``/api/pub`` endpoint (see :func:`pub_ws`). - - `active_session_file` (when set) is forwarded as - ``HERMES_TUI_ACTIVE_SESSION_FILE``. The TUI writes the current session id - there whenever it creates/resumes/switches sessions, giving the dashboard a - small cross-process breadcrumb for reconnecting after an unexpected browser - WebSocket close. - - `profile` (when set) scopes the ENTIRE chat to that profile by pointing - ``HERMES_HOME`` at the profile dir in the child env. Every spawned - process (the TUI and the ``tui_gateway.entry`` it launches) resolves - ``get_hermes_home()`` from that env var at its own import, so the child - binds the profile's config, skills, memory, and state.db from the start - — the same propagation ``hermes -p `` performs. The in-process - ``HERMES_TUI_GATEWAY_URL`` attach is SKIPPED for scoped chats: the - dashboard's in-memory gateway runs under the dashboard's own profile, - so a profile-scoped chat must spawn its own gateway subprocess. + * ``HERMES_TUI_RESUME`` — session resume (``ui-tui`` does not parse argv, so + ``--resume`` cannot be appended); resolved to the newest descendant first. + * ``HERMES_TUI_GATEWAY_URL`` — attach to this process's in-memory + ``tui_gateway`` instead of spawning a Python gateway subprocess. SKIPPED + for profile-scoped chats: the dashboard's gateway runs under the + dashboard's own profile, so a scoped chat must spawn its own. + * ``HERMES_TUI_SIDECAR_URL`` — mirror dispatcher emits to ``/api/pub``. + * ``HERMES_TUI_ACTIVE_SESSION_FILE`` — the TUI writes its current session id + there, a cross-process breadcrumb for reconnecting after a WS drop. + * ``profile`` scopes the ENTIRE chat by pointing ``HERMES_HOME`` at the + profile dir; every spawned process resolves ``get_hermes_home()`` from + that at import, the same propagation ``hermes -p `` performs. """ from hermes_cli.web_server import ( _config_profile_scope, _open_session_db_for_profile, _resolve_profile_dir, - _session_latest_descendant, - ) + _session_latest_descendant) from hermes_cli.main import PROJECT_ROOT, _apply_tui_python_env, _make_tui_argv profile_dir: Optional[Path] = None @@ -507,27 +355,22 @@ def _resolve_chat_argv( profile_dir = _resolve_profile_dir(requested) argv, cwd = _make_tui_argv(PROJECT_ROOT / "ui-tui", tui_dev=False) - # Hermes TUI child: build via the single spawn-env factory (profile-home - # contract applied; secrets kept — the spawned agent needs provider creds). - # An explicit profile scope still overrides HERMES_HOME before config is - # bridged into the child environment. + # Build via the single spawn-env factory (profile-home contract applied; + # secrets kept — the spawned agent needs provider creds). An explicit + # profile scope overrides HERMES_HOME before config is bridged into the env. from tools.environments.local import build_subprocess_env env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=True) if profile_dir is not None: env["HERMES_HOME"] = str(profile_dir) try: from hermes_cli.config import ( - apply_terminal_config_to_env, - read_raw_config, - terminal_config_owned_env_vars, - ) + apply_terminal_config_to_env, read_raw_config, terminal_config_owned_env_vars) if profile_dir is not None: - # The dashboard process already bridged its own terminal config - # into os.environ at startup. Remove only keys explicitly owned by - # that launch profile before applying the selected profile. Values - # exported by the operator for keys omitted from the launch profile - # remain valid fallbacks, matching apply_terminal_config_to_env(). + # The dashboard already bridged its own terminal config into + # os.environ at startup. Remove only keys explicitly owned by that + # launch profile before applying the selected profile; operator + # exports for keys the launch profile omits remain valid fallbacks. raw_launch_terminal = read_raw_config().get("terminal") for env_var in terminal_config_owned_env_vars(raw_launch_terminal): env.pop(env_var, None) @@ -539,32 +382,20 @@ def _resolve_chat_argv( _log.warning("Failed to apply terminal config bridge for dashboard chat", exc_info=True) _apply_tui_python_env(env) env.setdefault("NODE_ENV", "production") - # Browser-embedded chat should prefer stable wheel-based scrollback over - # native terminal mouse tracking. When mouse tracking is enabled, wheel - # events are consumed by the TUI and forwarded as terminal input, which - # makes browser-side transcript scrolling feel broken. Keep the terminal - # build unchanged for native CLI usage; only disable mouse tracking for - # the dashboard PTY path. + # Mouse tracking would swallow wheel events the browser needs for + # transcript scrolling; disable it for the dashboard PTY only. env.setdefault("HERMES_TUI_DISABLE_MOUSE", "1") env.setdefault("HERMES_TUI_INLINE", "1") - # The dashboard terminal is xterm.js, which always renders 24-bit RGB. - # But chalk inside the TUI child decides its color depth from the - # SERVER process env — and hosted/cloud deploys run the dashboard under - # a process manager (container init, systemd) with no COLORTERM, so - # chalk downgrades every hex color to the xterm 256 palette. The skin's - # bronze border #CD7F32 snaps to palette 173 (#D7875F, salmon-red) and - # the banner reads red/yellow instead of gold. Local launches dodge - # this only because the operator's interactive terminal leaks - # COLORTERM=truecolor into os.environ. Backfill it for the PTY child; - # setdefault so an explicit operator value still wins. + # xterm.js always renders 24-bit RGB, but chalk in the child picks its + # depth from the SERVER env — hosted deploys under a process manager have + # no COLORTERM, so hex colors snap to the 256 palette (bronze -> salmon). + # Backfill; setdefault so an explicit operator value still wins. env.setdefault("COLORTERM", "truecolor") env["HERMES_TUI_DASHBOARD"] = "1" if resume: _resume_db = _open_session_db_for_profile( - requested if profile_dir is not None else None, - read_only=True, - ) + requested if profile_dir is not None else None, read_only=True) try: latest_resume, _latest_path = _session_latest_descendant(resume, _resume_db) finally: @@ -579,40 +410,27 @@ def _resolve_chat_argv( if active_session_file: env["HERMES_TUI_ACTIVE_SESSION_FILE"] = active_session_file - # Profile-scoped chats must NOT attach to the dashboard's in-memory - # gateway — it runs under the dashboard's own profile. Without the - # attach URL, gatewayClient spawns its own `tui_gateway.entry`, which - # inherits the profile HERMES_HOME set above. - if profile_dir is None: - if gateway_ws_url := _build_gateway_ws_url(): - env["HERMES_TUI_GATEWAY_URL"] = gateway_ws_url + # Without the attach URL, gatewayClient spawns its own `tui_gateway.entry`, + # which inherits the profile HERMES_HOME set above. + if profile_dir is None and (gateway_ws_url := _build_gateway_ws_url()): + env["HERMES_TUI_GATEWAY_URL"] = gateway_ws_url return list(argv), str(cwd) if cwd else None, env -# Hosts that mean "listen on every interface" — the server should bind to -# them, but an in-container client must NOT dial them: dialing 0.0.0.0 -# resolves to "any local interface", which on most platforms routes through -# the kernel's wildcard stack and behind a forward proxy (HTTPS_PROXY with -# a NO_PROXY that doesn't list 0.0.0.0) gets MITM'd into a failed handshake -# (issue #58993). The fix is to use a loopback address for the client -# netloc while leaving the bind host alone. +# Hosts that mean "listen on every interface" — bind to them, but an +# in-container client must NOT dial them: 0.0.0.0 routes through the wildcard +# stack and behind a forward proxy (HTTPS_PROXY without 0.0.0.0 in NO_PROXY) +# gets MITM'd into a failed handshake. Clients dial loopback instead. _WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"}) def _resolve_client_ws_host() -> Optional[str]: """Return the host the in-container WS client should dial. - Resolution order: - - 1. Explicit ``HERMES_DASHBOARD_WS_HOST`` env var — wins always. Operators - running the dashboard behind a forward proxy can pin a routable host - (e.g. ``127.0.0.1``, the container's internal IP, or a sidecar DNS - name) and bypass auto-detection entirely. - 2. The configured bind host — if it's a wildcard (``0.0.0.0`` / ``::``), - substitute ``127.0.0.1`` since both the dashboard and its TUI child - run in the same container. - 3. Any other bind host (loopback or LAN IP) — preserved verbatim. + ``HERMES_DASHBOARD_WS_HOST`` wins always (operators behind a forward proxy + pin a routable host); a wildcard bind becomes ``127.0.0.1`` (dashboard and + TUI child share the container); any other bind host is preserved verbatim. """ from hermes_cli.web_server import app explicit = os.environ.get("HERMES_DASHBOARD_WS_HOST", "").strip() @@ -629,85 +447,15 @@ def _resolve_client_ws_host() -> Optional[str]: return host -def _build_gateway_ws_url() -> Optional[str]: - """ws:// URL the PTY child should attach to for JSON-RPC gateway traffic. +def _server_internal_ws_url(path: str, **extra_qs) -> Optional[str]: + """``ws://:?&`` for server-spawned WS + clients, or None when unbound. - Loopback / ``--insecure``: ``?token=<_SESSION_TOKEN>``. - - Gated mode: the legacy token path is rejected by ``_ws_auth_ok``, so the - server-spawned PTY child authenticates with the process-lifetime internal - credential (``?internal=``). It must NOT use a single-use browser ticket: - the child reads this URL once at startup and reuses it on every reconnect, - and a 30s-TTL ticket can expire before a slow cold boot even dials. - """ - from hermes_cli.web_server import _SESSION_TOKEN, app - host = _resolve_client_ws_host() - port = getattr(app.state, "bound_port", None) - - if not host or not port: - return None - - netloc = ( - f"[{host}]:{port}" - if ":" in host and not host.startswith("[") - else f"{host}:{port}" - ) - - if getattr(app.state, "auth_required", False): - from hermes_cli.dashboard_auth.ws_tickets import internal_ws_credential - - qs = urllib.parse.urlencode({"internal": internal_ws_credential()}) - else: - qs = urllib.parse.urlencode({"token": _SESSION_TOKEN}) - - return f"ws://{netloc}/api/ws?{qs}" - - -async def _resolve_chat_argv_async( - resume: Optional[str] = None, - sidecar_url: Optional[str] = None, - profile: Optional[str] = None, - active_session_file: Optional[str] = None, -) -> tuple[list[str], Optional[str], Optional[dict]]: - """Resolve chat argv without blocking the dashboard event loop. - - ``_resolve_chat_argv`` may run ``npm install`` / ``npm run build`` through - ``_make_tui_argv``. Keep that synchronous work off the WebSocket event - loop so reverse proxies and existing dashboard connections can continue - to exchange keepalives while the TUI launch command is prepared. The - async lock preserves the previous one-build-at-a-time behavior when - multiple browser tabs connect at once without occupying worker threads - while queued connections wait. - """ - from hermes_cli.web_server import _get_chat_argv_lock, _resolve_chat_argv, app - kwargs = { - "resume": resume, - "sidecar_url": sidecar_url, - "profile": profile, - } - if active_session_file is not None: - kwargs["active_session_file"] = active_session_file - - async with _get_chat_argv_lock(app): - return await asyncio.to_thread( - _resolve_chat_argv, - **kwargs, - ) - - -def _build_sidecar_url(channel: str) -> Optional[str]: - """ws:// URL the PTY child should publish events to, or None when unbound. - - Loopback / ``--insecure``: uses ``?token=<_SESSION_TOKEN>``. - - Gated mode: authenticates with the process-lifetime internal credential - (``?internal=``), the same one ``_build_gateway_ws_url`` uses. The PTY - child is a server-spawned process we trust; the credential is multi-use - and never expires, so the child can reconnect ``/api/pub`` without a new - URL. (This previously minted a single-use 30s ticket, which meant the - child could not reconnect and could miss the window on a slow cold boot.) - Connections authenticated this way are recorded under the - ``server-internal`` identity in the audit log. + Loopback / ``--insecure``: ``?token=<_SESSION_TOKEN>``. Gated: the legacy + token is rejected by ``_ws_auth_ok``, so the PTY child authenticates with + the process-lifetime internal credential (``?internal=``) — NOT a single-use + browser ticket: the child reads the URL once and reuses it on every + reconnect, and a 30s-TTL ticket can expire before a slow cold boot dials. """ from hermes_cli.web_server import _SESSION_TOKEN, app host = _resolve_client_ws_host() @@ -719,17 +467,44 @@ def _build_sidecar_url(channel: str) -> Optional[str]: netloc = f"[{host}]:{port}" if ":" in host and not host.startswith("[") else f"{host}:{port}" if getattr(app.state, "auth_required", False): - # Gated mode — use the internal credential so the WS upgrade survives - # _ws_auth_ok and the child can reconnect. from hermes_cli.dashboard_auth.ws_tickets import internal_ws_credential - qs = urllib.parse.urlencode( - {"internal": internal_ws_credential(), "channel": channel} - ) + auth = {"internal": internal_ws_credential()} else: - qs = urllib.parse.urlencode({"token": _SESSION_TOKEN, "channel": channel}) + auth = {"token": _SESSION_TOKEN} - return f"ws://{netloc}/api/pub?{qs}" + return f"ws://{netloc}{path}?{urllib.parse.urlencode({**auth, **extra_qs})}" + + +def _build_gateway_ws_url() -> Optional[str]: + """ws:// URL the PTY child attaches to for JSON-RPC gateway traffic.""" + return _server_internal_ws_url("/api/ws") + + +def _build_sidecar_url(channel: str) -> Optional[str]: + """ws:// URL the PTY child publishes events to, or None when unbound.""" + return _server_internal_ws_url("/api/pub", channel=channel) + + +async def _resolve_chat_argv_async( + resume: Optional[str] = None, + sidecar_url: Optional[str] = None, + profile: Optional[str] = None, + active_session_file: Optional[str] = None) -> tuple[list[str], Optional[str], Optional[dict]]: + """Resolve chat argv without blocking the dashboard event loop. + + ``_resolve_chat_argv`` may run ``npm install`` / ``npm run build``; keep + that off the WebSocket loop so keepalives keep flowing. The async lock + preserves one-build-at-a-time when several tabs connect at once without + occupying worker threads while queued connections wait. + """ + from hermes_cli.web_server import _get_chat_argv_lock, _resolve_chat_argv, app + kwargs = {"resume": resume, "sidecar_url": sidecar_url, "profile": profile} + if active_session_file is not None: + kwargs["active_session_file"] = active_session_file + + async with _get_chat_argv_lock(app): + return await asyncio.to_thread(_resolve_chat_argv, **kwargs) def _active_session_file_for_channel(app: "FastAPI", channel: str) -> Path: @@ -747,12 +522,10 @@ def _active_session_file_for_channel(app: "FastAPI", channel: str) -> Path: return path -# Console commands run in a worker thread. On a timeout, asyncio.wait_for cancels -# the *awaitable*, but Python threads aren't preemptible, so a genuinely stuck -# worker keeps running to completion. To keep that from exhausting the shared -# default thread pool (asyncio.to_thread), we run console commands on a small -# dedicated, bounded pool: a leaked worker is capped, and concurrent console -# execution is bounded to a fixed number of threads regardless of reconnects. +# Console commands run in a worker thread; on timeout asyncio cancels the +# awaitable but the thread keeps running, so a stuck worker would exhaust the +# shared default pool. A small dedicated pool caps the leak and bounds +# concurrent console execution regardless of reconnects. _CONSOLE_EXECUTOR_MAX_WORKERS = 4 _console_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None _console_executor_lock = threading.Lock() @@ -765,14 +538,10 @@ def _get_console_executor() -> concurrent.futures.ThreadPoolExecutor: with _console_executor_lock: if _console_executor is None: _console_executor = concurrent.futures.ThreadPoolExecutor( - max_workers=_CONSOLE_EXECUTOR_MAX_WORKERS, - thread_name_prefix="hermes-console", - ) - # Ensure the pool is torn down on interpreter exit. Don't wait on - # in-flight workers: a stuck 60s console command must not block - # shutdown (cancel_futures drops anything not yet started). + max_workers=_CONSOLE_EXECUTOR_MAX_WORKERS, thread_name_prefix="hermes-console") + # Tear down on interpreter exit without waiting on in-flight + # workers: a stuck 60s console command must not block shutdown. atexit.register( lambda: _console_executor - and _console_executor.shutdown(wait=False, cancel_futures=True) - ) + and _console_executor.shutdown(wait=False, cancel_futures=True)) return _console_executor diff --git a/hermes_cli/web_server_sessions.py b/hermes_cli/web_server_sessions.py index f6c8192983..6aa6cedd91 100644 --- a/hermes_cli/web_server_sessions.py +++ b/hermes_cli/web_server_sessions.py @@ -1,4 +1,5 @@ -"""Session-DB access for the dashboard: per-profile SessionDB opening with schema heal, latest-descendant lookup and the auto-archive ticker. +"""Session-DB access for the dashboard: per-profile SessionDB opening with schema +heal, latest-descendant lookup and the auto-archive ticker. Split out of ``hermes_cli.web_server``; every externally used name is re-imported there, so ``web_server.`` keeps resolving (and monkeypatching) as before. @@ -15,44 +16,7 @@ from typing import Dict, Optional # Same logger the code used before extraction (record parity). _log = logging.getLogger("hermes_cli.web_server") - -# --------------------------------------------------------------------------- -# Session detail endpoints -# --------------------------------------------------------------------------- - - -def _session_latest_descendant(session_id: str, db): - """Resolve a session id to the newest child leaf session. - - /model may create child sessions. Dashboard refresh should continue the - newest child instead of reopening the old parent. - """ - def row_get(row, key, index): - if isinstance(row, dict): - return row.get(key) - try: - return row[key] - except Exception: - try: - return row[index] - except Exception: - return None - - sid = db.resolve_session_id(session_id) - if not sid or not db.get_session(sid): - return None, [] - - conn = ( - getattr(db, "conn", None) - or getattr(db, "_conn", None) - or getattr(db, "connection", None) - or getattr(db, "_connection", None) - ) - - rows = [] - if conn is not None: - raw_rows = conn.execute( - """ +_DESCENDANTS_SQL = """ WITH RECURSIVE descendants(id, parent_session_id, started_at) AS ( SELECT id, parent_session_id, started_at FROM sessions WHERE id = ? UNION @@ -61,15 +25,23 @@ def _session_latest_descendant(session_id: str, db): JOIN descendants d ON s.parent_session_id = d.id ) SELECT id, parent_session_id, started_at FROM descendants - """, - (sid,), - ).fetchall() - for row in raw_rows: - rows.append({ - "id": row_get(row, "id", 0), - "parent_session_id": row_get(row, "parent_session_id", 1), - "started_at": row_get(row, "started_at", 2), - }) + """ + + +def _session_latest_descendant(session_id: str, db): + """Resolve a session id to the newest child leaf session. + + /model may create child sessions; a dashboard refresh should continue the + newest child instead of reopening the old parent. Returns ``(leaf, path)``. + """ + sid = db.resolve_session_id(session_id) + if not sid or not db.get_session(sid): + return None, [] + + conn = getattr(db, "_conn", None) + if conn is not None: + keys = ("id", "parent_session_id", "started_at") + rows = [dict(zip(keys, row)) for row in conn.execute(_DESCENDANTS_SQL, (sid,)).fetchall()] else: rows = db.list_sessions_rich(limit=10000, offset=0, compact_rows=True) @@ -89,7 +61,6 @@ def _session_latest_descendant(session_id: str, db): current = sid path = [sid] seen = {sid} - while children.get(current): candidates = [r for r in children[current] if r.get("id") not in seen] if not candidates: @@ -98,67 +69,59 @@ def _session_latest_descendant(session_id: str, db): current = candidates[0]["id"] path.append(current) seen.add(current) - return current, path -# Serialises the one-time writable schema bootstrap for read-only opens. -# Concurrent first-load polls otherwise race sqlite file creation: the losers -# open mode=ro against a store whose schema is still being written and every -# query raises "no such table: sessions". +# Serialises the one-time writable schema bootstrap for read-only opens, so +# concurrent first-load polls don't open mode=ro against a half-written schema +# ("no such table: sessions"). _session_db_bootstrap_lock = threading.Lock() def _session_db_read_probe_statements() -> tuple: """Stale-schema probes for read-only opens, derived from SCHEMA_SQL. - Read-only opens skip _reconcile_columns(), so an older store would - otherwise 500 on every poll until something opened it writable. Derived - from the same schema the writable reconciler applies, so any column - added there is probed here automatically — the previous hand-written - probe listed four columns and went stale the first time a new column - (sessions.last_activity_at) shipped, leaving the desktop sidebar empty - after `hermes update` until the first message forced a writable open. + Read-only opens skip _reconcile_columns(), so an older store would 500 on + every poll until something opened it writable. Deriving the probe from the + schema means any new column is probed automatically (a hand-written list + went stale once and left the sidebar empty after `hermes update`). """ from hermes_state_schema import schema_read_probe_statements return schema_read_probe_statements() -# Stores where a heal WRITABLE OPEN SUCCEEDED and the read probe still -# failed afterwards: the schema problem is one reconciliation cannot fix -# (e.g. a NOT-NULL-without-default column SQLite refuses to ADD). Retrying -# the full writable init on every poll would hammer a live DB for nothing, -# so such stores fall back to the raw read-only open until restart. A -# FAILED writable open (transient lock) is deliberately NOT recorded — -# the next poll retries the heal. +# Stores where a heal WRITABLE OPEN SUCCEEDED but the read probe still failed: +# one reconciliation cannot fix them (e.g. a NOT-NULL-without-default column), +# so they fall back to the raw read-only open until restart instead of paying +# a writable init per poll. A FAILED writable open (transient lock) is NOT +# recorded — the next poll retries the heal. _session_db_heal_exhausted: set = set() -# Deduplicates the heal-failure warning per store per process, so a -# persistent problem is loud once instead of once per sidebar poll. +# Deduplicates the heal-failure warning per store per process. _session_db_heal_warned: set = set() +def _is_stale_schema_error(exc: BaseException) -> bool: + message = str(exc).lower() + return "no such table" in message or "no such column" in message + + def _open_session_db_at_path(db_path: Path, *, read_only: bool): """Open a SessionDB at an explicit path with an explicit access mode. - Writable opens keep the full init and repair path. Read-only opens - bootstrap a missing or zero-byte store once, and heal an older or - malformed schema through one writable open before reopening read-only. - The healthy read path never takes a write lock or requests a checkpoint. + Writable opens keep the full init and repair path. Read-only opens bootstrap + a missing or zero-byte store once, and heal an older or malformed schema + through one writable open before reopening read-only; the healthy read path + never takes a write lock or requests a checkpoint. - Scope of the heal: the probe checks every table/column declared in - SCHEMA_SQL (see ``schema_read_probe_statements``), so ANY schema - addition escalates a stale store to a one-time writable open — the same - reconcile the store's own backend runs at startup. Tables created - outside SCHEMA_SQL (telemetry ``tel_*``, FTS shadow tables) are - deliberately outside both the probe and the heal. + The probe covers every table/column in SCHEMA_SQL, so ANY schema addition + escalates a stale store to a one-time writable open (the same reconcile the + store's backend runs at startup). Tables created outside SCHEMA_SQL + (telemetry ``tel_*``, FTS shadow tables) are outside both probe and heal. """ from hermes_cli.web_server import ( - _session_db_heal_exhausted, - _session_db_heal_warned, - _session_db_read_probe_statements, - ) + _session_db_heal_exhausted, _session_db_heal_warned, _session_db_read_probe_statements) import sqlite3 from hermes_state import SessionDB, is_malformed_schema_error @@ -196,28 +159,24 @@ def _open_session_db_at_path(db_path: Path, *, read_only: bool): try: return _open_probed() except (sqlite3.DatabaseError, UnicodeDecodeError) as exc: - message = str(exc).lower() - stale_schema = "no such table" in message or "no such column" in message - if not stale_schema and not ( - # UnicodeDecodeError = pysqlite could not decode SQLite's own - # error message because corrupt file bytes were embedded in it - # (#98924). The one-writable-open heal is the only repair path, - # so route it through the same dispatch as malformed schema. - is_malformed_schema_error(exc) or isinstance(exc, UnicodeDecodeError) - ): + # UnicodeDecodeError = pysqlite could not decode SQLite's own error + # message because corrupt file bytes were embedded in it; the + # one-writable-open heal is the only repair path, so treat it as + # malformed schema. + if not ( + _is_stale_schema_error(exc) + or is_malformed_schema_error(exc) + or isinstance(exc, UnicodeDecodeError)): raise SessionDB(db_path=db_path, read_only=False).close() try: return _open_probed() except (sqlite3.DatabaseError, UnicodeDecodeError) as still_stale: - message = str(still_stale).lower() - if "no such table" not in message and "no such column" not in message: + if not _is_stale_schema_error(still_stale): raise - # The writable open succeeded but the store is STILL behind the - # probe: reconciliation cannot fix this one. Serve reads without - # the probe (queries touching the broken part will still fail, - # everything else works) and stop paying the writable init per - # poll. + # Writable open succeeded but the store is STILL behind the probe: + # serve reads without the probe (only queries touching the broken + # part fail) and stop paying the writable init per poll. _session_db_heal_exhausted.add(str(db_path)) if str(db_path) not in _session_db_heal_warned: _session_db_heal_warned.add(str(db_path)) @@ -226,17 +185,14 @@ def _open_session_db_at_path(db_path: Path, *, read_only: bool): "reconcile could not add (%s); read paths may partially " "fail until the store is repaired", db_path, - still_stale, - ) + still_stale) return _open_probed() def _open_session_db_for_profile(profile: Optional[str], *, read_only: bool): - """Open a SessionDB with an explicit access mode for a profile. + """Open a SessionDB for ``profile`` (None/empty = this process's own state.db). - ``profile`` None/empty selects this process's own ``state.db``. A named - profile opens that profile's on-disk store directly. Access-mode - semantics are documented on :func:`_open_session_db_at_path`. + Access-mode semantics: see :func:`_open_session_db_at_path`. """ from hermes_cli.web_server import _cron_profile_home from hermes_state import _default_db_path @@ -250,21 +206,18 @@ def _open_session_db_for_profile(profile: Optional[str], *, read_only: bool): # In-process throttle for the opportunistic auto-archive trigger, keyed by -# profile. Bounds the config.yaml read to at most once per this window per -# profile; the actual sweep is throttled far more coarsely by state_meta -# (sessions.min_interval_hours) inside maybe_auto_archive. +# profile: bounds the config.yaml read to once per window; the sweep itself is +# throttled far more coarsely by state_meta (sessions.min_interval_hours). _AUTO_ARCHIVE_CHECK_INTERVAL_S = 300.0 _last_auto_archive_check: Dict[str, float] = {} def _maybe_auto_archive_for_profile(profile: Optional[str]) -> None: - """Run the config-gated stale-session auto-archive for ``profile``. + """Run the config-gated stale-session auto-archive for ``profile``. Never raises. - The Desktop backend is spawned as ``hermes serve`` — it runs neither the - interactive CLI nor the messaging gateway, so neither of those startup - hooks fire for Desktop users. Triggering the (double-throttled, config-off - by default) sweep from the session-list path is what makes - ``sessions.auto_archive`` take effect there. Never raises. + The Desktop backend (``hermes serve``) runs neither the CLI nor the gateway + startup hooks, so the session-list path is what makes + ``sessions.auto_archive`` take effect there. """ from hermes_cli.web_server import _open_session_db_for_profile try: @@ -283,8 +236,7 @@ def _maybe_auto_archive_for_profile(profile: Optional[str]) -> None: try: db.maybe_auto_archive( idle_days=float(cfg.get("auto_archive_days", 3)), - min_interval_hours=int(cfg.get("min_interval_hours", 24)), - ) + min_interval_hours=int(cfg.get("min_interval_hours", 24))) finally: db.close() except Exception as exc: @@ -292,16 +244,10 @@ def _maybe_auto_archive_for_profile(profile: Optional[str]) -> None: async def _auto_archive_ticker_loop( - interval_s: float = 3600.0, initial_delay_s: float = 90.0 -) -> None: - """Live timer for the stale-session auto-archive (primary profile). - - A long-running Desktop/serve backend must keep sweeping on schedule even - when no ``/api/sessions`` request arrives to fire the opportunistic - trigger — e.g. the app sits open for days on an idle chat. The real - cadence is still owned by state_meta (``sessions.min_interval_hours``) - inside ``maybe_auto_archive``; this loop is only the poll rate. - """ + interval_s: float = 3600.0, initial_delay_s: float = 90.0) -> None: + """Poll-rate timer for the auto-archive sweep (primary profile), so a + long-idle Desktop keeps sweeping without any ``/api/sessions`` request. + The real cadence is still owned by state_meta inside ``maybe_auto_archive``.""" def _sweep() -> None: _maybe_auto_archive_for_profile(None)