hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 29 Packages Projects Releases Wiki Activity

refactor(web): tools _bad_request helper, skills hub helpers, sessions flag table, docstring compaction

Browse Source 
- tools: _bad_request() for 16 status_code=400 raises; payload built in worker;
  drop _terminal_backend_names/_model_catalog_section one-shot helpers
- sessions: _RENAME_FLAG_SETTERS table (4 flags incl. unread), keyset export
  loop tightened, compression_root/lineage_tip compacted
- web_server_chat: single try/except for both PTY bridge imports,
  contextlib.suppress for swallow-all blocks
- docstrings/comments compacted by hand (every WHY kept); AST-neutral packing
This commit is contained in:
Teknium
2026-09-02 22:21:05 -07:00
parent 4d9007cd17
commit 92e52684d1
5 changed files with 320 additions and 593 deletions
Download Patch File Download Diff File Expand all files Collapse all files

288
hermes_cli/web_routers/sessions.py
View File

@@ -1,11 +1,9 @@
"""Session dashboard routes.
Three routers because global route order matters: ``list_router``
(GET /api/sessions) was registered before the profiles ``sessions_router``
include, ``search_router`` (GET /api/sessions/search) right after it, and
``manage_router`` (mutation/detail endpoints) much later — each is mounted at
its original registration point. web_server-owned helpers are reached via
the late-binding seam so ``monkeypatch.setattr(web_server, ...)`` keeps working.
Three routers because global route order matters: ``list_router`` (GET
/api/sessions) mounts before the profiles ``sessions_router``, ``search_router``
right after it, ``manage_router`` (mutation/detail) much later. web_server-owned
helpers are reached via the late-binding seam so monkeypatching keeps working.
"""
import asyncio
@@ -13,7 +11,7 @@ import json
import re
import sqlite3
import time
from typing import Any, Callable, Dict, List, Optional
from typing import Callable, List, Optional
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.encoders import jsonable_encoder
@@ -58,14 +56,6 @@ async def _read_session_import_body(request: Request) -> bytes:
return bytes(body)
def _import_sessions_for_profile(profile: Optional[str], sessions: List[Dict[str, Any]]) -> Dict[str, Any]:
db = _open_session_db_for_profile(profile, read_only=False)
try:
return db.import_sessions(sessions)
finally:
db.close()
# Prune filters forwarded to SessionDB; string filters map "" -> None.
_PRUNE_STR_FILTERS = (
"source", "title_like", "end_reason", "cwd_prefix", "model_like", "provider",
@@ -75,6 +65,9 @@ _PRUNE_NUM_FILTERS = (
"min_tool_calls", "max_tool_calls")
_PRUNE_ROW_KEYS = ("id", "source", "title", "model", "started_at", "last_active", "message_count")
def _prune_sessions(body: SessionPrune):
"""Delete ended sessions matching filters (mirrors `hermes sessions prune`)."""
from hermes_cli.web_server import get_hermes_home
@@ -93,8 +86,7 @@ def _prune_sessions(body: SessionPrune):
db = _open_session_db_for_profile(body.profile, read_only=False)
try:
filters = {
"older_than_days": effective_older_than,
"started_before": body.started_before,
"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},
@@ -112,12 +104,7 @@ def _prune_sessions(body: SessionPrune):
"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,
"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]}
"sessions": [{k: r.get(k) for k in _PRUNE_ROW_KEYS} for r in rows]}
sessions_dir = profile_home / "sessions"
removed = db.prune_sessions(
sessions_dir=sessions_dir if sessions_dir.exists() else None, **filters)
@@ -156,12 +143,8 @@ 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 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).
"""
"""Resolve *session_id*; a corrupt store (prefix scan raises "malformed") is
reported as 503 with the actual problem instead of a misleading 404."""
try:
return db.resolve_session_id(session_id)
except sqlite3.DatabaseError as exc:
@@ -177,36 +160,26 @@ def _resolve_session_id(db, session_id: str) -> Optional[str]:
) from exc
# ``le=100`` on limit: an unbounded limit lets one request drag every session
# row (plus correlated-subquery preview work) out of SQLite in a single hit.
@list_router.get("/api/sessions")
def get_sessions(
# ``le=100``: an unbounded limit lets one request drag every session row
# (plus correlated-subquery preview work) out of SQLite in a single hit.
limit: int = Query(20, ge=0, le=100),
offset: int = Query(0, ge=0),
min_messages: int = 0,
archived: str = "exclude",
order: str = "created",
source: str = None,
sources: str = None,
exclude_sources: str = None,
cwd_prefix: str = None,
full: bool = False,
limit: int = Query(20, ge=0, le=100), offset: int = Query(0, ge=0), min_messages: int = 0,
archived: str = "exclude", order: str = "created", source: str = None, sources: str = None,
exclude_sources: str = None, cwd_prefix: str = None, full: bool = False,
profile: Optional[str] = None):
"""List sessions.
``archived``: ``exclude`` (default) / ``only`` / ``include``. ``order``:
``created`` (start time) or ``recent`` (latest activity across the
compression chain, so a long-running chat stays on page one after it
auto-compresses onto a fresh id). Rows omit ``system_prompt`` /
``model_config`` unless ``full=1``.
``order=recent`` sorts by latest activity across the compression chain, so
a long-running chat stays on page one after it auto-compresses onto a fresh
id. Rows omit ``system_prompt`` / ``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")
profile_name: Optional[str] = None
if profile:
profile_name, _ = _cron_profile_home(profile)
profile_name = _cron_profile_home(profile)[0] if profile else None
try:
# Auto-archive is the only write on this GET path: run it on its own
# maintenance connection, then open the listing connection read-only.
@@ -221,12 +194,9 @@ def get_sessions(
source_list = _csv(sources)
exclude_list = _csv(exclude_sources)
scope = dict(
source=source or None,
sources=source_list or None,
exclude_sources=exclude_list or None,
cwd_prefix=(cwd_prefix or None),
min_message_count=min_message_count,
include_archived=include_archived,
source=source or None, sources=source_list or None,
exclude_sources=exclude_list or None, cwd_prefix=(cwd_prefix or None),
min_message_count=min_message_count, include_archived=include_archived,
archived_only=archived_only)
sessions = db.list_sessions_rich(
limit=limit,
@@ -258,7 +228,6 @@ def get_sessions(
_log.exception("GET /api/sessions failed")
# 503, not 500: the store is busy, not gone — the desktop keeps its
# sidebar instead of reading a 500 as an authoritative empty list.
# The bounded open-retry lives in SessionDB's read-only constructor.
transient = is_transient_sqlite_error(exc)
raise HTTPException(
status_code=503 if transient else 500,
@@ -284,19 +253,14 @@ def _is_compression_edge(child: dict, parent: dict) -> bool:
@search_router.get("/api/sessions/search")
async def search_sessions(
q: str = "",
limit: int = 20,
profile: Optional[str] = None,
source: str = None,
sources: str = None,
exclude_sources: str = None):
"""Search sessions by ID plus FTS5 message content.
q: str = "", limit: int = 20, profile: Optional[str] = None, source: str = None,
sources: str = None, exclude_sources: str = None):
"""Search sessions by ID (first) plus FTS5 message content.
ID matches first, then content matches. Results are deduped by
compression lineage, not raw ``session_id``: auto-compression rotates a
chat onto a fresh id and leaves the old segment in the FTS index, so one
logical chat owns many rows. Branches also use ``parent_session_id`` but
are real alternate conversations — they are NOT collapsed into the parent.
Results are deduped by compression lineage, not raw ``session_id``:
auto-compression rotates a chat onto a fresh id and leaves the old segment
in the FTS index. Branches also use ``parent_session_id`` but are real
alternate conversations — they are NOT collapsed into the parent.
"""
if not q or not q.strip():
return {"results": []}
@@ -321,20 +285,12 @@ async def search_sessions(
root_cache: dict = {}
def compression_root(session_id: str) -> str:
if not session_id:
return session_id
if session_id in root_cache:
return root_cache[session_id]
chain = []
cur = session_id
visited = set()
root = session_id
while cur and cur not in visited:
visited.add(cur)
chain.append(cur)
chain, cur, root = [], session_id, session_id
while cur and cur not in chain: # ``not in chain`` guards parent cycles
if cur in root_cache:
root = root_cache[cur]
break
chain.append(cur)
s = get_session(cur)
parent = s.get("parent_session_id") if isinstance(s, dict) else None
parent_session = get_session(parent) if parent else None
@@ -349,15 +305,12 @@ async def search_sessions(
tip_cache: dict = {}
def lineage_tip(root_id: str) -> str:
if root_id in tip_cache:
return tip_cache[root_id]
tip = root_id
try:
tip = db.get_compression_tip(root_id) or tip
except Exception:
pass
tip_cache[root_id] = tip
return tip
if root_id not in tip_cache:
try:
tip_cache[root_id] = db.get_compression_tip(root_id) or root_id
except Exception:
tip_cache[root_id] = root_id
return tip_cache[root_id]
# One keyspace for id-hits and content-hits, keyed by lineage root;
# first hit wins, and ID matches run first.
@@ -387,7 +340,8 @@ async def search_sessions(
"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,
"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,
@@ -399,10 +353,10 @@ async def search_sessions(
payload["id"] = sid
seen[root] = payload
def hit_payload(row: dict, snippet: str, role) -> dict:
def hit_payload(row: dict, snippet: str, role, session_started) -> dict:
return {
"snippet": snippet, "role": role, "source": row.get("source"),
"model": row.get("model"), "session_started": row.get("session_started")}
"model": row.get("model"), "session_started": session_started}
# Direct ID matches first (pasted ids never appear in message text).
for row in db.search_sessions_by_id(
@@ -410,9 +364,8 @@ async def search_sessions(
sources=source_list or None, exclude_sources=exclude_list or None):
sid = row.get("id")
preview = (row.get("preview") or "").strip()
payload = hit_payload(row, preview or f"Session ID: {sid}", None)
payload["session_started"] = row.get("started_at")
add_lineage_result(sid, payload)
snippet = preview or f"Session ID: {sid}"
add_lineage_result(sid, hit_payload(row, snippet, None, row.get("started_at")))
# Prefix wildcards so partial words match ("nimb" -> "nimb*");
# quoted phrases and existing wildcards are kept as-is.
@@ -422,15 +375,15 @@ async def search_sessions(
# Over-fetch so lineage dedup can still surface `limit` distinct
# conversations when several hits collapse onto one root.
matches = db.search_messages(
query=prefix_query,
source_filter=include_sources,
exclude_sources=exclude_list or None,
limit=max(safe_limit * 5, 50),
query=prefix_query, source_filter=include_sources,
exclude_sources=exclude_list or None, 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"], hit_payload(m, m.get("snippet", ""), m.get("role")))
add_lineage_result(
m["session_id"],
hit_payload(m, m.get("snippet", ""), m.get("role"), m.get("session_started")))
return {"results": list(seen.values())}
finally:
db.close()
@@ -438,17 +391,14 @@ async def search_sessions(
@manage_router.post("/api/sessions/bulk-delete")
async def bulk_delete_sessions_endpoint(body: BulkDeleteSessions):
"""Delete every session in ``body.ids`` in one transaction (POST because
many clients refuse a DELETE body).
"""Delete every session in ``body.ids`` in one transaction (POST: many
clients refuse a DELETE body).
Contract matches :meth:`SessionDB.delete_sessions`: unknown ids are
skipped (``deleted`` reports what really happened — selection state can
race another tab's delete), children are orphaned not cascaded, active
and archived rows ARE deleted because the user hand-picked them, and
on-disk transcript cleanup is left to the next prune pass.
Per :meth:`SessionDB.delete_sessions`: unknown ids are skipped (``deleted``
reports what really happened), children are orphaned, active/archived rows
ARE deleted (hand-picked), on-disk cleanup is left to the next prune.
"""
# Hard cap so a runaway selection can't lock the writer for long; 500
# covers "select all on every page of a reasonable scrollback".
# Hard cap so a runaway selection can't lock the writer for long.
if len(body.ids) > 500:
raise HTTPException(status_code=400, detail="ids must contain at most 500 entries")
deleted = await asyncio.to_thread(
@@ -469,7 +419,8 @@ async def import_sessions_endpoint(request: Request):
raise HTTPException(status_code=400, detail="Invalid session import payload") from exc
try:
result = await asyncio.to_thread(_import_sessions_for_profile, body.profile, body.sessions)
result = await asyncio.to_thread(
_with_db, body.profile, lambda db: db.import_sessions(body.sessions), read_only=False)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@@ -480,8 +431,7 @@ async def import_sessions_endpoint(request: Request):
@manage_router.get("/api/sessions/empty/count")
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 of empty, ended, non-archived sessions (the "Delete empty (N)" button)."""
count = await asyncio.to_thread(
_with_db, profile, lambda db: db.count_empty_sessions(), read_only=True)
return {"count": count}
@@ -491,11 +441,9 @@ async def count_empty_sessions_endpoint(profile: Optional[str] = None):
async def delete_empty_sessions_endpoint(profile: Optional[str] = None):
"""Delete every empty, ended, non-archived session in one transaction.
Mirrors :meth:`SessionDB.delete_empty_sessions`: "empty" means NO
``messages`` rows at all (a rewound/compacted chat reads
``message_count == 0`` while its soft-archived rows are the only copy of
the transcript); active and archived sessions are skipped; children are
orphaned; on-disk cleanup is left to the next prune pass.
"Empty" means NO ``messages`` rows at all — a rewound/compacted chat reads
``message_count == 0`` while its soft-archived rows are the only transcript
copy (see :meth:`SessionDB.delete_empty_sessions`).
"""
deleted = await asyncio.to_thread(
_with_db, profile, lambda db: db.delete_empty_sessions(), read_only=False)
@@ -504,17 +452,16 @@ async def delete_empty_sessions_endpoint(profile: Optional[str] = None):
@manage_router.get("/api/sessions/stats")
async def get_session_stats(profile: Optional[str] = None):
"""Session-store statistics (mirrors `hermes sessions stats`). Registered
before ``/api/sessions/{session_id}`` so ``stats`` isn't taken as an id."""
"""Session-store statistics (mirrors `hermes sessions stats`)."""
def _stats(db):
out = {
"total": db.session_count(include_archived=True),
"active_store": db.session_count(include_archived=False),
"archived": db.session_count(archived_only=True),
"messages": db.message_count(),
"archived": db.session_count(archived_only=True), "messages": db.message_count(),
"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
@@ -529,9 +476,8 @@ async def get_session_detail(session_id: str, profile: Optional[str] = None):
session = db.get_session(sid) if sid else None
if not session:
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.
# Always stamp the owner: unowned default-profile rows made multi-profile
# clients resolve them to whichever gateway happened to be active.
session["profile"] = _serving_profile(profile)
session["is_default_profile"] = session["profile"] == "default"
return session
@@ -546,9 +492,7 @@ async def get_session_latest_descendant(session_id: str, profile: Optional[str]
if not latest:
raise HTTPException(status_code=404, detail=_NOT_FOUND)
return {
"requested_session_id": path[0] if path else session_id,
"session_id": latest,
"path": path,
"requested_session_id": path[0] if path else session_id, "session_id": latest, "path": path,
"changed": bool(path and latest != path[0])}
@@ -579,11 +523,8 @@ def _project_for_display(messages: list) -> list:
@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),
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")
@@ -593,15 +534,14 @@ async def get_session_messages(
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.
# Always page (an omitted limit used to load whole transcripts). 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
)
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:
@@ -612,21 +552,17 @@ async def get_session_messages(
"session_id": sid,
"messages": projected_messages,
"pagination": {
"limit": _limit,
"offset": offset,
"limit": _limit, "offset": offset,
"order": order or ("latest" if limit is None else "oldest"),
"returned": len(projected_messages)}}
@manage_router.delete("/api/sessions/{session_id}")
async def delete_session_endpoint(session_id: str, profile: Optional[str] = None):
# ``profile`` opens another local profile's state.db directly (remote
# profiles never reach here — the desktop routes those to the remote backend).
def _delete(db):
# An already-absent session is an idempotent success: the desktop
# optimistically removes the row and RESTORES it on any error, so a 404
# here resurrected ghost rows (transient empties from /goal +
# auto-compression churn racing the sidebar snapshot).
# Already-absent is an idempotent success: the desktop optimistically
# removes the row and RESTORES it on any error, so a 404 resurrected
# ghost rows (transient empties racing the sidebar snapshot).
sid = _resolve_session_id(db, session_id)
if not sid:
return {"ok": True, "already_absent": True}
@@ -638,20 +574,19 @@ async def delete_session_endpoint(session_id: str, profile: Optional[str] = None
@manage_router.post("/api/sessions/owner-backfill")
async def backfill_session_owner_profiles(body: SessionOwnerBackfill):
"""Stamp legacy ``profile_name = NULL`` rows with this store's own
serving-profile identity.
"""Stamp legacy ``profile_name = NULL`` rows with the serving-profile identity.
A multi-connection Desktop fails closed on unowned rows, leaving legacy
sessions unresumable. Each ``state.db`` belongs to exactly one profile,
so this is a single-match backfill (the same identity ``get_sessions``
stamps on outgoing rows), idempotent: non-NULL owners are never overwritten.
A multi-connection Desktop fails closed on unowned rows. Each ``state.db``
belongs to exactly one profile, so this is a single-match, idempotent
backfill (non-NULL owners are never overwritten).
"""
stamp = _serving_profile(body.profile)
with http_failure("POST /api/sessions/owner-backfill failed", 500, detail="Internal server error"):
with http_failure(
"POST /api/sessions/owner-backfill failed", 500, detail="Internal server error"):
stamped = await asyncio.to_thread(
_with_db, body.profile, lambda db: db.backfill_null_session_profiles(stamp), read_only=False
)
_with_db, body.profile, lambda db: db.backfill_null_session_profiles(stamp),
read_only=False)
if stamped:
_log.info(
@@ -660,18 +595,20 @@ async def backfill_session_owner_profiles(body: SessionOwnerBackfill):
return {"ok": True, "stamped": stamped, "profile": stamp}
# PATCH /api/sessions/{id} flag -> SessionDB setter, applied in this order.
_RENAME_FLAG_SETTERS = (
("archived", "set_session_archived"),
("hidden", "set_session_hidden"),
("pinned", "set_session_pinned"))
("archived", lambda db, sid, v: db.set_session_archived(sid, v)),
("hidden", lambda db, sid, v: db.set_session_hidden(sid, v)),
("pinned", lambda db, sid, v: db.set_session_pinned(sid, v)),
("unread", lambda db, sid, v: db.set_session_read(sid, read=not v)),
)
@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``,
``pinned`` (exempts from the auto-archive sweep), ``unread`` (True =
explicitly unread, False = read up to now). Any field may be omitted."""
flags = ("archived", "hidden", "pinned", "unread")
"""Update ``title`` (empty clears) and/or the flags; ``pinned`` exempts from
the auto-archive sweep, ``unread=False`` marks read up to now."""
flags = [flag for flag, _ in _RENAME_FLAG_SETTERS]
def _update(db):
sid = _resolve_session_id(db, session_id)
@@ -688,16 +625,13 @@ 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))
result = {"ok": True, "title": None}
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 ""}
for f in flags:
if getattr(body, f) is not None:
result[f] = bool(getattr(body, f))
setter(db, sid, value)
result[flag] = bool(value)
result["title"] = db.get_session_title(sid) or ""
return result
return _with_db(body.profile, _update, read_only=False)
@@ -726,21 +660,15 @@ async def export_session_endpoint(session_id: str, profile: Optional[str] = None
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
last_id, first = 0, True
while True:
messages = db.get_messages(
sid, limit=500, after_id=last_id if last_id is not None else 0)
messages = db.get_messages(sid, limit=500, after_id=last_id)
for message in messages:
if not first:
yield ","
yield _compact_json(message)
yield ("" if first else ",") + _compact_json(message)
first = False
if len(messages) < 500:
last_id = messages[-1].get("id") if len(messages) == 500 else None
if last_id is None: # short page, or cannot keyset without row ids
break
last_id = messages[-1].get("id")
if last_id is None:
break # defensive: cannot keyset without row ids
yield "]}"
finally:
db.close()

105
hermes_cli/web_routers/skills.py
View File

@@ -40,7 +40,8 @@ _SKILL_HUB_SOURCE_LABELS = {
"github": "GitHub",
"clawhub": "ClawHub",
"lobehub": "LobeHub",
"browse-sh": "browse.sh"}
"browse-sh": "browse.sh",
}
def _hub_sources(profile: Optional[str]):
@@ -78,12 +79,8 @@ def _flag(obj, attr: str) -> bool:
def _skill_meta_to_payload(m) -> dict:
return {
"name": m.name,
"description": m.description,
"source": m.source,
"identifier": m.identifier,
"trust_level": m.trust_level,
"repo": m.repo,
"name": m.name, "description": m.description, "source": m.source,
"identifier": m.identifier, "trust_level": m.trust_level, "repo": m.repo,
"tags": list(m.tags or [])}
@@ -104,10 +101,8 @@ def _clear_skills_prompt_cache() -> None:
async def install_skill_hub(body: SkillInstallRequest, profile: Optional[str] = None):
identifier = require(body.identifier, "identifier is required")
return spawn_profile_action(
body.profile or profile,
["skills", "install", identifier, "--yes"],
_hub_action_name("install", identifier),
log_msg="Failed to spawn skills install",
body.profile or profile, ["skills", "install", identifier, "--yes"],
_hub_action_name("install", identifier), log_msg="Failed to spawn skills install",
prefix="Failed to install skill")
@@ -115,10 +110,8 @@ async def install_skill_hub(body: SkillInstallRequest, profile: Optional[str] =
async def uninstall_skill_hub(body: SkillUninstallRequest, profile: Optional[str] = None):
name = require(body.name, "name is required")
return spawn_profile_action(
body.profile or profile,
["skills", "uninstall", name, "--yes"],
_hub_action_name("uninstall", name),
log_msg="Failed to spawn skills uninstall",
body.profile or profile, ["skills", "uninstall", name, "--yes"],
_hub_action_name("uninstall", name), log_msg="Failed to spawn skills uninstall",
prefix="Failed to uninstall skill")
@@ -126,17 +119,13 @@ async def uninstall_skill_hub(body: SkillUninstallRequest, profile: Optional[str
async def update_skills_hub(
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")
(body.profile if body else None) or profile, ["skills", "update"], "skills-update",
log_msg="Failed to spawn skills update", prefix="Failed to update skills")
@hub_router.get("/api/skills/hub/official")
async def list_official_skills(profile: Optional[str] = None):
"""The ENTIRE built-in optional-skills catalog (local scan, no network),
each row marked installed-or-not for ``profile``."""
"""The ENTIRE optional-skills catalog (local scan), marked installed for ``profile``."""
def _run():
from tools.skills_hub import OptionalSkillSource
@@ -160,9 +149,8 @@ async def list_official_skills(profile: Optional[str] = None):
@hub_router.get("/api/skills/hub/sources")
async def list_skills_hub_sources(profile: Optional[str] = None):
"""Configured skill-hub sources + installed-skill provenance, so the
Browse-hub tab has something to show before a search runs. ``profile``
scopes the installed-skill provenance."""
"""Configured skill-hub sources + installed-skill provenance (scoped to
``profile``), so the Browse-hub tab has something before a search runs."""
def _run():
sources = _hub_sources(profile)
@@ -190,9 +178,7 @@ async def list_skills_hub_sources(profile: Optional[str] = None):
for entry in out:
entry["searchable"] = not (index_available and entry["id"] in _API_SOURCE_IDS)
return {
"sources": out,
"index_available": index_available,
"featured": featured,
"sources": out, "index_available": index_available, "featured": featured,
"installed": _installed_hub_identifiers(profile)}
with http_failure("skills hub sources listing failed", 502, "Hub sources failed"):
@@ -202,8 +188,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):
"""Search the skill hub across all configured sources (network-bound,
runs in a thread). Results install by identifier via /hub/install."""
"""Search the skill hub across all configured sources (network-bound)."""
query = (q or "").strip()
if not query:
return {"results": [], "source_counts": {}, "timed_out": [], "installed": {}}
@@ -220,15 +205,14 @@ async def search_skills_hub(
_rank = {"builtin": 2, "trusted": 1, "community": 0}
seen = {}
for r in all_results:
if r.identifier not in seen or _rank.get(r.trust_level, 0) > _rank.get(seen[r.identifier].trust_level, 0):
prev = seen.get(r.identifier)
if prev is None or _rank.get(r.trust_level, 0) > _rank.get(prev.trust_level, 0):
seen[r.identifier] = r
deduped = list(seen.values())[:capped]
return {
"results": [_skill_meta_to_payload(m) for m in deduped],
"source_counts": source_counts,
"timed_out": timed_out,
"installed": _installed_hub_identifiers(profile)}
"results": [_skill_meta_to_payload(m) for m in deduped], "source_counts": source_counts,
"timed_out": timed_out, "installed": _installed_hub_identifiers(profile)}
with http_failure("skills hub search failed", 502, "Hub search failed"):
return await asyncio.to_thread(_run)
@@ -245,9 +229,8 @@ async def _hub_lookup(fn, ident: str, log_msg: str, prefix: str):
@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
``profile`` so a profile with different hub taps resolves against ITS
source router."""
"""A hub skill's SKILL.md + file manifest WITHOUT installing it; scoped to
``profile`` so different hub taps resolve against THAT source router."""
ident = require(identifier, "identifier is required")
def _run():
@@ -272,25 +255,21 @@ async def preview_skill_hub(identifier: str = "", profile: Optional[str] = None)
m = meta or bundle
return {
"name": getattr(m, "name", ident),
"description": getattr(m, "description", "") or "",
"name": getattr(m, "name", ident), "description": getattr(m, "description", "") or "",
"source": getattr(m, "source", "") or "",
"identifier": getattr(m, "identifier", ident) or ident,
"trust_level": getattr(m, "trust_level", "community") or "community",
"repo": getattr(m, "repo", None),
"tags": list(getattr(m, "tags", None) or []),
"skill_md": skill_md,
"files": sorted(files.keys())}
"repo": getattr(m, "repo", None), "tags": list(getattr(m, "tags", None) or []),
"skill_md": skill_md, "files": sorted(files.keys())}
return await _hub_lookup(_run, ident, "skills hub preview failed", "Hub preview failed")
@hub_router.get("/api/skills/hub/scan")
async def scan_skill_hub(identifier: str = "", profile: Optional[str] = None):
"""Run the install-time security scan on a hub skill WITHOUT installing it
(same ``scan_skill`` / ``should_allow_install`` pipeline as the CLI, on a
quarantined bundle that is cleaned up afterwards). Scoped to ``profile``
so the bundle resolves where an install would pull it from."""
"""Install-time security scan of a hub skill WITHOUT installing it (the CLI's
``scan_skill`` / ``should_allow_install`` pipeline on a quarantined bundle);
scoped to ``profile`` so the bundle resolves where an install would."""
ident = require(identifier, "identifier is required")
def _run():
@@ -306,12 +285,12 @@ async def scan_skill_hub(identifier: str = "", profile: Optional[str] = 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
q_path = quarantine_bundle(bundle)
try:
q_path = quarantine_bundle(bundle)
result = scan_skill(q_path, source=scan_source)
# Advisory SkillEvaluator Tier 1 second opinion: optional binary,
# never blocks, errors degrade to no data (same as the CLI installer).
@@ -325,29 +304,22 @@ async def scan_skill_hub(identifier: str = "", profile: Optional[str] = None):
"incomplete_checks": t1.incomplete_checks,
"findings": [
{
"check": f.check,
"validator": f.validator,
"severity": f.severity,
"message": f.message,
"file": f.file,
"line": f.line,
"check": f.check, "validator": f.validator,
"severity": f.severity, "message": f.message,
"file": f.file, "line": f.line,
"secrets_class": f.is_secrets_class}
for f in t1.findings]}
except Exception:
_log.debug("Tier 1 advisory scan skipped", exc_info=True)
finally:
if q_path is not None:
_shutil.rmtree(q_path, ignore_errors=True)
_shutil.rmtree(q_path, ignore_errors=True)
# `allowed` may be None ("ask") for agent-created/dangerous gates.
allowed, reason = should_allow_install(result, force=False)
findings = [
{
"severity": f.severity,
"category": f.category,
"file": f.file,
"line": f.line,
"description": f.description}
"severity": f.severity, "category": f.category, "file": f.file,
"line": f.line, "description": f.description}
for f in result.findings]
counts = {sev: 0 for sev in ("critical", "high", "medium", "low")}
for f in result.findings:
@@ -443,9 +415,8 @@ async def get_skill_content(name: str, profile: Optional[str] = None):
@router.post("/api/skills")
async def create_skill(body: SkillCreate):
"""Create a custom skill via the same validated write path as the agent's
``skill_manage`` tool, minus the agent write-approval gate — a write from
the authenticated dashboard IS the user acting directly."""
"""Create a skill via the agent's ``skill_manage`` write path, minus the
write-approval gate — an authenticated dashboard write IS the user."""
from tools.skill_manager_tool import _create_skill
result = await scoped_to_thread(

254
hermes_cli/web_routers/tools.py
View File

@@ -116,7 +116,8 @@ _BACKEND_PROBES = {
"singularity": _probe_singularity_backend,
"ssh": _probe_ssh_backend,
"modal": _probe_modal_backend,
"daytona": _probe_daytona_backend}
"daytona": _probe_daytona_backend,
}
def _probe_terminal_backend(name: str, terminal_cfg: dict) -> tuple:
@@ -176,7 +177,8 @@ def _category_providers(ts_key: str, config: dict) -> list:
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]:
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
@@ -186,17 +188,12 @@ def _find_toolset_provider_row(ts_key: str, config: dict, provider: Optional[str
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
if name not in {ts_key for ts_key, _, _ in _get_effective_configurable_toolsets()}:
raise HTTPException(status_code=400, detail=f"Unknown toolset: {name}")
raise _bad_request(f"Unknown toolset: {name}")
def _dict_section(config: dict, key: str) -> dict:
@@ -208,35 +205,30 @@ def _dict_section(config: dict, key: str) -> dict:
return section
def _bad_request(detail: str) -> HTTPException:
return HTTPException(status_code=400, detail=detail)
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 (
_CONFIG_ONLY_TOOLSETS,
_get_effective_configurable_toolsets,
_get_platform_tools,
_toolset_configuration_platform,
_toolset_has_keys,
get_nous_subscription_features,
_CONFIG_ONLY_TOOLSETS, _get_effective_configurable_toolsets, _get_platform_tools,
_toolset_configuration_platform, _toolset_has_keys, get_nous_subscription_features,
gui_toolset_label)
from hermes_cli.platforms import platform_label
from toolsets import resolve_toolset
from utils import is_truthy_value
def _read():
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}
@@ -254,23 +246,17 @@ async def get_toolsets(profile: Optional[str] = None):
if name in _CONFIG_ONLY_TOOLSETS:
# Config-only capabilities (stt) have no per-platform toolset —
# their switch is their own config section (e.g. stt.enabled).
from utils import is_truthy_value
section = config.get(name)
section = section if isinstance(section, dict) else {}
is_enabled = is_truthy_value(section.get("enabled", True), default=True)
else:
is_enabled = name in enabled_by_platform[target_platform]
result.append({
"name": name,
"label": gui_toolset_label(label),
"description": desc,
"name": name, "label": gui_toolset_label(label), "description": desc,
"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})
"enabled": is_enabled, "available": is_enabled,
"configured": _toolset_has_keys(name, config, features=features), "tools": tools})
return result
@@ -280,9 +266,7 @@ async def toggle_toolset(name: str, body: ToolsetToggle, profile: Optional[str]
(``platform_toolsets.cli`` for most; platform-restricted toolsets target
their own platform) via the same ``_save_platform_tools`` the CLI uses."""
from hermes_cli.tools_config import (
_CONFIG_ONLY_TOOLSETS,
_get_platform_tools,
_save_platform_tools,
_CONFIG_ONLY_TOOLSETS, _get_platform_tools, _save_platform_tools,
_toolset_configuration_platform)
_require_known_toolset(name)
@@ -298,7 +282,8 @@ 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:
@@ -307,11 +292,10 @@ async def toggle_toolset(name: str, body: ToolsetToggle, profile: Optional[str]
await asyncio.to_thread(_run)
# Install-on-enable: a provider with a post_setup hook whose install-state
# predicate is UNSATISFIED (cua-driver binary missing, etc.) gets the same
# background install `hermes tools` runs interactively — otherwise the
# toggle "saves" but the tool never appears because its check_fn can't
# find the binary. Best-effort: a spawn failure never fails the toggle.
# Install-on-enable: a provider whose post_setup install predicate is
# UNSATISFIED (cua-driver binary missing, etc.) gets the same background
# install `hermes tools` runs — otherwise the toggle "saves" but the tool
# never appears. Best-effort: a spawn failure never fails the toggle.
post_setup_started: Optional[str] = None
if body.enabled and name not in _CONFIG_ONLY_TOOLSETS:
def _pending_install_key() -> Optional[str]:
@@ -323,11 +307,8 @@ async def toggle_toolset(name: str, body: ToolsetToggle, profile: Optional[str]
return None
with _profile_scope(scope_profile):
config = load_config()
for prov in _visible_providers(cat, config):
key = prov.get("post_setup")
if key and not _post_setup_already_installed(key):
return key
return None
keys = [prov.get("post_setup") for prov in _visible_providers(cat, config)]
return next((k for k in keys if k and not _post_setup_already_installed(k)), None)
try:
pending_key = await asyncio.to_thread(_pending_install_key)
@@ -340,24 +321,16 @@ async def toggle_toolset(name: str, body: ToolsetToggle, profile: Optional[str]
_log.exception("install-on-enable post-setup spawn failed for %s", name)
return {
"ok": True,
"name": name,
"platform": target_platform,
"enabled": body.enabled,
"ok": True, "name": name, "platform": target_platform, "enabled": body.enabled,
"post_setup_started": post_setup_started}
@router.get("/api/tools/toolsets/{name}/config")
async def get_toolset_config(name: str, profile: Optional[str] = None):
"""Provider matrix + key status for a toolset's config panel: the same rows
the CLI ``hermes tools`` picker shows, each env var annotated with
``is_set``. No ``TOOL_CATEGORIES`` entry -> empty list, ``has_category:
false``."""
"""Provider matrix + key status for a toolset's config panel (the CLI picker
rows, each env var annotated ``is_set``); no category -> ``has_category: false``."""
from hermes_cli.tools_config import (
TOOL_CATEGORIES,
_is_provider_active,
_visible_providers,
provider_readiness_status,
TOOL_CATEGORIES, _is_provider_active, _visible_providers, provider_readiness_status,
web_provider_capabilities)
from hermes_cli.config import get_env_value
from hermes_cli.nous_subscription import get_nous_subscription_features
@@ -370,18 +343,14 @@ async def get_toolset_config(name: str, profile: Optional[str] = None):
cat = TOOL_CATEGORIES.get(name)
providers = []
active_provider = None
active_search_backend = None
active_extract_backend = None
if cat:
# Entitlement state fetched once for the whole matrix.
features = get_nous_subscription_features(config, force_fresh=True)
for prov in _visible_providers(cat, config, force_fresh=True):
env_vars = [
{
"key": e["key"],
"prompt": e.get("prompt", e["key"]),
"url": e.get("url"),
"default": e.get("default"),
"key": e["key"], "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", [])]
# Same active-provider determination as the CLI picker, so the
@@ -412,30 +381,24 @@ async def get_toolset_config(name: str, profile: Optional[str] = None):
# (tts.<key>.*) holding the provider's voice/model settings.
row["tts_provider"] = prov["tts_provider"]
providers.append(row)
payload = {
"name": name, "has_category": cat is not None, "providers": providers,
"active_provider": active_provider}
if name == "web":
# Resolve active backends exactly as the web_search/web_extract
# dispatchers do, so badges reflect what a call would hit now.
try:
from tools.web_tools import _get_extract_backend, _get_search_backend
active_search_backend = _get_search_backend()
active_extract_backend = _get_extract_backend()
search_backend = _get_search_backend()
extract_backend = _get_extract_backend()
except Exception:
active_search_backend = None
active_extract_backend = None
return cat, providers, active_provider, active_search_backend, active_extract_backend
search_backend = extract_backend = None
payload["active_search_backend"] = search_backend
payload["active_extract_backend"] = extract_backend
return payload
cat, providers, active_provider, active_search_backend, active_extract_backend = await asyncio.to_thread(_read)
payload = {
"name": name,
"has_category": cat is not None,
"providers": providers,
"active_provider": active_provider}
if name == "web":
payload["active_search_backend"] = active_search_backend
payload["active_extract_backend"] = active_extract_backend
return payload
return await asyncio.to_thread(_read)
@router.get("/api/tools/toolsets/{name}/models")
@@ -465,29 +428,17 @@ async def get_toolset_models(
current = raw.strip()
if current not in catalog:
current = default_model if default_model in catalog else None
return row, plugin, catalog, default_model, current
models = [
{
"id": model_id, "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()]
return {
"name": name, "has_models": bool(models), "provider": row.get("name"),
"plugin": plugin, "models": models, "current": current, "default": default_model}
resolved = await asyncio.to_thread(_read)
if resolved is None:
return _no_models(name)
row, plugin, catalog, default_model, current = resolved
models = [
{
"id": model_id,
"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()]
return {
"name": name,
"has_models": bool(models),
"provider": row.get("name") if row else None,
"plugin": plugin,
"models": models,
"current": current,
"default": default_model}
return await asyncio.to_thread(_read) or _no_models(name)
@router.put("/api/tools/toolsets/{name}/model")
@@ -495,10 +446,12 @@ async def select_toolset_model(
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_section(name)
section = _MODEL_CATALOG_TOOLSETS.get(name)
if section is None:
raise _bad_request(f"Toolset has no model catalog: {name}")
model_id = (body.model or "").strip()
if not model_id:
raise HTTPException(status_code=400, detail="model is required")
raise _bad_request("model is required")
def _run():
with config_write_scope(body.profile or profile):
@@ -506,13 +459,11 @@ async def select_toolset_model(
row = _find_toolset_provider_row(name, config, body.provider)
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}")
raise _bad_request(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}")
raise _bad_request(f"Unknown model {model_id!r} for backend {plugin!r}")
_dict_section(config, section)["model"] = model_id
save_config(config)
@@ -525,19 +476,14 @@ 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):
"""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.
"""Persist a provider selection via ``apply_provider_selection`` (shared with
``hermes tools``, so both write identical keys).
``web`` only: ``capability`` ('search' | 'extract') scopes the write to
``web.<capability>_backend`` (the per-capability override the runtime
dispatchers resolve first); the provider must support that capability.
Omitted -> legacy whole-provider write of ``web.backend``.
Managed Nous rows report Portal entitlement: the GUI has no inline login,
so an unentitled selection would write config and never activate. The
response adds ``needs_nous_auth: true`` + ``feature`` so the client can
drive the Nous Portal OAuth flow and refetch.
``web`` only: ``capability`` ('search' | 'extract') writes
``web.<capability>_backend`` (the override the dispatchers resolve first);
omitted -> legacy ``web.backend``. Managed Nous rows report Portal
entitlement (``needs_nous_auth`` + ``feature``): the GUI has no inline
login, so an unentitled selection would write config and never activate.
"""
from hermes_cli.tools_config import apply_provider_selection, web_provider_capabilities
from hermes_cli.nous_subscription import (
@@ -547,13 +493,10 @@ async def select_toolset_provider(
if body.capability is not None:
if name != "web":
raise HTTPException(
status_code=400, detail="capability selection is only supported for the web toolset"
)
raise _bad_request("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')")
raise _bad_request(
f"Unknown capability: {body.capability!r} (expected 'search' or 'extract')")
def _provider_row(config):
return next(
@@ -569,24 +512,19 @@ async def select_toolset_provider(
# resolving through the shared fallback chain.
prov = _provider_row(config)
if prov is None:
raise HTTPException(
status_code=400,
detail=f"Unknown provider {body.provider!r} for toolset {name!r}")
raise _bad_request(
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")
raise _bad_request(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}")
raise _bad_request(f"{body.provider} does not support {body.capability}")
_dict_section(config, "web")[f"{body.capability}_backend"] = backend
else:
try:
apply_provider_selection(name, body.provider, config)
except KeyError as exc:
raise HTTPException(status_code=400, detail=str(exc).strip('"'))
raise _bad_request(str(exc).strip('"'))
save_config(config)
response: Dict[str, Any] = {"ok": True, "name": name, "provider": body.provider}
if body.capability is not None:
@@ -635,9 +573,8 @@ async def save_toolset_env(name: str, body: ToolsetEnvUpdate, profile: Optional[
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))}")
raise _bad_request(
f"Unknown env var(s) for toolset {name}: {', '.join(sorted(unknown))}")
saved: List[str] = []
skipped: List[str] = []
@@ -646,7 +583,7 @@ async def save_toolset_env(name: str, body: ToolsetEnvUpdate, profile: Optional[
try:
save_env_value(key, value.strip())
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
raise _bad_request(str(exc))
saved.append(key)
else:
skipped.append(key)
@@ -661,23 +598,18 @@ 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):
"""Spawn ``hermes tools post-setup <key>`` (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
against the declared allowlist; ``profile`` is threaded so hooks that
touch config see the same HERMES_HOME as the drawer's other writes."""
"""Spawn ``hermes tools post-setup <key>`` (long-running installs) as a
background action tailed via ``GET /api/actions/tools-post-setup/status``;
``profile`` is threaded so hooks see the drawer's HERMES_HOME."""
from hermes_cli.tools_config import valid_post_setup_keys
_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 _bad_request(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")
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")
result["key"] = body.key
return result
@@ -702,12 +634,8 @@ async def get_terminal_backends(profile: Optional[str] = None):
for row in rows:
status, detail = _probe_terminal_backend(row["name"], terminal_cfg)
backends.append({
"name": row["name"],
"label": row["label"],
"description": row["description"],
"active": row["name"] == active,
"status": status,
"detail": detail})
"name": row["name"], "label": row["label"], "description": row["description"],
"active": row["name"] == active, "status": status, "detail": detail})
return {"active": active, "backends": backends}
return await asyncio.to_thread(_read)
@@ -719,11 +647,10 @@ async def select_terminal_backend(
"""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()
valid_names = _terminal_backend_names()
valid_names = {row["name"] for row in _terminal_backend_rows()}
if backend not in valid_names:
raise HTTPException(
status_code=400,
detail=f"Unknown terminal backend: {body.backend!r}. "
raise _bad_request(
f"Unknown terminal backend: {body.backend!r}. "
f"Use one of: {', '.join(sorted(valid_names))}")
def _run():
@@ -751,11 +678,8 @@ async def grant_computer_use_permissions(profile: Optional[str] = None):
CuaDriver via LaunchServices so the TCC dialog is attributed correctly).
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.")
raise _bad_request("Computer Use permission grants are a macOS concept.")
return spawn_profile_action(
profile,
["computer-use", "permissions", "grant"],
"computer-use-grant",
profile, ["computer-use", "permissions", "grant"], "computer-use-grant",
log_msg="Failed to spawn computer-use permissions grant",
prefix="Failed to request permissions")

230
hermes_cli/web_server_chat.py
View File

@@ -10,6 +10,7 @@ import logging
import asyncio
import atexit
import concurrent.futures
import contextlib
import hmac
import os
import re
@@ -26,35 +27,21 @@ 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: 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=<session_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). Both expose the same surface —
# spawn/read/write/resize/close/is_available — so the handler needs no guards.
if sys.platform.startswith("win"):
try:
# /api/pty spawns ``hermes --tui`` behind a pseudo-terminal and forwards bytes +
# resize escapes to xterm.js. POSIX uses pty_bridge (fcntl/termios); native
# Windows uses win_pty_bridge (pywinpty/ConPTY); same surface, no handler guards.
try:
if sys.platform.startswith("win"):
from hermes_cli.win_pty_bridge import WinPtyBridge as PtyBridge, PtyUnavailableError
_PTY_BRIDGE_AVAILABLE = True
except ImportError: # pragma: no cover - pywinpty missing
PtyBridge = None # type: ignore[assignment]
_PTY_BRIDGE_AVAILABLE = False
class PtyUnavailableError(RuntimeError): # type: ignore[no-redef]
"""Stub when win_pty_bridge cannot be imported."""
else:
try:
else:
from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError
_PTY_BRIDGE_AVAILABLE = True
except ImportError: # pragma: no cover - dev env without ptyprocess
PtyBridge = None # type: ignore[assignment]
_PTY_BRIDGE_AVAILABLE = False
_PTY_BRIDGE_AVAILABLE = True
except ImportError: # pragma: no cover - pywinpty / ptyprocess missing
PtyBridge = None # type: ignore[assignment]
_PTY_BRIDGE_AVAILABLE = False
class PtyUnavailableError(RuntimeError): # type: ignore[no-redef]
"""Stub on platforms where pty_bridge can't be imported."""
class PtyUnavailableError(RuntimeError): # type: ignore[no-redef]
"""Stub when the platform PTY bridge cannot be imported."""
_RESIZE_RE = re.compile(rb"\x1b\[RESIZE:(\d+);(\d+)\]")
_PTY_READ_CHUNK_TIMEOUT = 0.2
@@ -84,21 +71,15 @@ async def _legacy_pump(ws: "WebSocket", bridge) -> None:
except Exception:
return
finally:
# 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:
# Close the WS so the writer's ``ws.receive()`` returns instead of
# blocking forever on a half-open browser socket (fds would leak and
# auto-reconnect stacks a fresh PTY on each orphan). Reap the bridge
# here too (idempotent): cancelling the handler the instant the WS
# closes can skip the writer's ``finally``.
with contextlib.suppress(Exception):
await asyncio.to_thread(bridge.close)
except Exception:
pass
try:
with contextlib.suppress(Exception):
await ws.close()
except Exception:
pass
reader_task = asyncio.create_task(pump_pty_to_ws())
@@ -128,10 +109,8 @@ async def _legacy_pump(ws: "WebSocket", bridge) -> None:
pass
finally:
reader_task.cancel()
try:
with contextlib.suppress(asyncio.CancelledError, Exception):
await reader_task
except (asyncio.CancelledError, Exception):
pass
await asyncio.to_thread(bridge.close)
@@ -143,14 +122,11 @@ _LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost", "testclient"})
def _ws_client_reason(ws: "WebSocket") -> Optional[str]:
"""Return a rejection reason token for the peer IP, or None when allowed.
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).
Loopback bind: only loopback peers (the legacy ``?token=`` is the only auth,
LAN hosts must not get to guess it); an empty peer fails closed. Explicit
non-loopback bind (``--insecure``) or gated mode: any peer — DNS-rebinding is
blocked by :func:`_ws_host_origin_reason`, and in gated mode
``ws.client.host`` is the X-Forwarded-For value anyway.
"""
from hermes_cli.web_server import app
if getattr(app.state, "auth_required", False):
@@ -172,34 +148,27 @@ def _ws_client_is_allowed(ws: "WebSocket") -> bool:
def _ws_host_origin_reason(ws: "WebSocket") -> Optional[str]:
"""Return a Host/Origin rejection reason (``host_mismatch …`` /
``origin_mismatch …``), or None when allowed.
"""Return ``host_mismatch …`` / ``origin_mismatch …``, or None when allowed.
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.
Host check is repeated here; an Origin header, when present, must target the
bound host. Non-web origins (packaged Electron: file://, null, app://) are
trusted — the 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())
host_header = ws.headers.get("host", "")
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", "")
if not origin:
return None
parsed = urllib.parse.urlparse(origin)
if parsed.scheme not in {"http", "https"}:
return None
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
@@ -237,19 +206,15 @@ 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 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.
``reason`` is None when accepted, else a short token (``no_credential``,
``token_mismatch``, ``ticket_invalid``, ``internal_invalid``);
``credential`` names what was presented so the accept path can log *how*.
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.
Loopback / ``--insecure``: legacy ``?token=`` (constant-time compared).
Gated: ``?ticket=`` (browser-minted, single-use, 30s TTL) or ``?internal=``
(process-lifetime, multi-use, only for server-spawned WS clients so the PTY
child can reconnect; never injected into the SPA). The legacy token is
rejected in gated mode: a leaked ``_SESSION_TOKEN`` must not grant access.
"""
from hermes_cli.web_server import _SESSION_TOKEN, app
auth_required = bool(getattr(app.state, "auth_required", False))
@@ -262,10 +227,8 @@ def _ws_auth_reason(ws: "WebSocket") -> tuple[Optional[str], str]:
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)
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
@@ -320,32 +283,20 @@ def _ws_auth_ok(ws: "WebSocket") -> bool:
def _resolve_chat_argv(
resume: Optional[str] = None,
sidecar_url: Optional[str] = None,
profile: Optional[str] = None,
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 (what ``hermes --tui`` runs).
Tests monkeypatch this to inject a tiny fake command so nothing has to
build the TUI bundle. Env contract with the child:
* ``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 <name>`` performs.
Tests monkeypatch this with a tiny fake command. Env contract: resume goes
through ``HERMES_TUI_RESUME`` (``ui-tui`` does not parse argv), resolved to
the newest descendant; ``HERMES_TUI_GATEWAY_URL`` attaches to this process's
in-memory gateway but is SKIPPED for profile-scoped chats (that gateway runs
under the dashboard's own profile, so a scoped chat spawns its own);
``profile`` scopes the ENTIRE chat by pointing ``HERMES_HOME`` at the profile
dir, the same propagation ``hermes -p <name>`` performs.
"""
from hermes_cli.web_server import (
_config_profile_scope,
_open_session_db_for_profile,
_resolve_profile_dir,
_config_profile_scope, _open_session_db_for_profile, _resolve_profile_dir,
_session_latest_descendant)
from hermes_cli.main import PROJECT_ROOT, _apply_tui_python_env, _make_tui_argv
@@ -355,9 +306,8 @@ def _resolve_chat_argv(
profile_dir = _resolve_profile_dir(requested)
argv, cwd = _make_tui_argv(PROJECT_ROOT / "ui-tui", tui_dev=False)
# 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.
# 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:
@@ -367,10 +317,8 @@ def _resolve_chat_argv(
apply_terminal_config_to_env, read_raw_config, terminal_config_owned_env_vars)
if profile_dir is not None:
# 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.
# Drop only the terminal keys the launch profile owns before applying
# the selected profile; operator exports for other keys stay valid.
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)
@@ -386,10 +334,8 @@ def _resolve_chat_argv(
# transcript scrolling; disable it for the dashboard PTY only.
env.setdefault("HERMES_TUI_DISABLE_MOUSE", "1")
env.setdefault("HERMES_TUI_INLINE", "1")
# 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.
# chalk in the child picks its color depth from the SERVER env; hosted
# deploys have no COLORTERM, so hex colors would snap to the 256 palette.
env.setdefault("COLORTERM", "truecolor")
env["HERMES_TUI_DASHBOARD"] = "1"
@@ -418,61 +364,44 @@ def _resolve_chat_argv(
return list(argv), str(cwd) if cwd else None, env
# 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 bind hosts an in-container client must NOT dial: behind a forward
# proxy (HTTPS_PROXY without 0.0.0.0 in NO_PROXY) the handshake gets MITM'd.
_WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"})
def _resolve_client_ws_host() -> Optional[str]:
"""Return the host the in-container WS client should dial.
``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.
"""
"""Host the in-container WS client should dial: ``HERMES_DASHBOARD_WS_HOST``
wins always; a wildcard bind becomes ``127.0.0.1``; others verbatim."""
from hermes_cli.web_server import app
explicit = os.environ.get("HERMES_DASHBOARD_WS_HOST", "").strip()
if explicit:
return explicit
host = getattr(app.state, "bound_host", None)
if not host:
return None
if host in _WILDCARD_HOSTS:
return "127.0.0.1"
return host
return "127.0.0.1" if host in _WILDCARD_HOSTS else host
def _server_internal_ws_url(path: str, **extra_qs) -> Optional[str]:
"""``ws://<client host>:<port><path>?<auth>&<extra>`` for server-spawned WS
clients, or None when unbound.
"""``ws://<host>:<port><path>?<auth>&<extra>`` for server-spawned WS clients,
or None when unbound.
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
Gated mode uses the process-lifetime internal credential, 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()
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
auth = {"internal": internal_ws_credential()}
else:
auth = {"token": _SESSION_TOKEN}
return f"ws://{netloc}{path}?{urllib.parse.urlencode({**auth, **extra_qs})}"
@@ -487,17 +416,10 @@ def _build_sidecar_url(channel: str) -> Optional[str]:
async def _resolve_chat_argv_async(
resume: Optional[str] = None,
sidecar_url: Optional[str] = None,
profile: Optional[str] = None,
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.
"""
"""Resolve chat argv off the event loop (it may run ``npm run build``); the
async lock keeps one-build-at-a-time without parking worker threads."""
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:
@@ -511,21 +433,15 @@ def _active_session_file_for_channel(app: "FastAPI", channel: str) -> Path:
"""Return the per-channel file where a dashboard TUI writes its active sid."""
from hermes_cli.web_server import _get_pty_active_session_files
files = _get_pty_active_session_files(app)
existing = files.get(channel)
if existing is not None:
return existing
fd, raw_path = tempfile.mkstemp(prefix="hermes-pty-active-", suffix=".json")
os.close(fd)
path = Path(raw_path)
files[channel] = path
return path
if files.get(channel) is None:
fd, raw_path = tempfile.mkstemp(prefix="hermes-pty-active-", suffix=".json")
os.close(fd)
files[channel] = Path(raw_path)
return files[channel]
# 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.
# On timeout asyncio cancels the awaitable but the console thread keeps running;
# a small dedicated pool caps the leak instead of exhausting the default pool.
_CONSOLE_EXECUTOR_MAX_WORKERS = 4
_console_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None
_console_executor_lock = threading.Lock()

36
hermes_cli/web_server_sessions.py
View File

@@ -1,9 +1,9 @@
"""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.<name>`` keeps resolving (and monkeypatching) as before.
Helpers that tests patch on ``web_server`` are reached lazily through it.
Split out of ``hermes_cli.web_server``, which re-imports every external name so
``web_server.<name>`` keeps resolving (and monkeypatching); helpers tests patch
there are reached lazily through it.
"""
import logging
@@ -79,13 +79,9 @@ _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 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`).
"""
"""Stale-schema probes for read-only opens (which skip _reconcile_columns()).
Derived from SCHEMA_SQL so a new column is probed automatically — a
hand-written list once went stale and emptied the sidebar after update."""
from hermes_state_schema import schema_read_probe_statements
return schema_read_probe_statements()
@@ -110,14 +106,9 @@ def _is_stale_schema_error(exc: BaseException) -> bool:
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.
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
Read-only opens bootstrap a missing/zero-byte store once and heal a stale or
malformed schema through ONE writable open before reopening read-only; the
healthy read path never takes a write lock. Tables outside SCHEMA_SQL
(telemetry ``tel_*``, FTS shadow tables) are outside both probe and heal.
"""
from hermes_cli.web_server import (
@@ -213,12 +204,9 @@ _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``. 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.
"""
"""Config-gated stale-session auto-archive for ``profile``; never raises.
``hermes serve`` runs neither CLI nor gateway startup hooks, so this
session-list trigger is what makes ``sessions.auto_archive`` work there."""
from hermes_cli.web_server import _open_session_db_for_profile
try:
key = profile or ""