diff --git a/hermes_cli/web_routers/actions.py b/hermes_cli/web_routers/actions.py index 39e35f05f2..e6f8f0bb06 100644 --- a/hermes_cli/web_routers/actions.py +++ b/hermes_cli/web_routers/actions.py @@ -4,20 +4,22 @@ Extracted from ``hermes_cli.web_server``; helpers/state that tests monkeypatch o ``web_server`` stay there and are late-bound (cycle-safe). """ -import logging import asyncio +import contextlib +import logging +import re import secrets import subprocess -from fastapi import APIRouter -from hermes_cli.web_routers._common import http_failure -from hermes_cli.web_deps import LateState, late -from fastapi import HTTPException, Request +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +from fastapi import APIRouter, HTTPException, Request + from hermes_cli import __version__ from hermes_cli.config import format_docker_update_message, recommended_update_command_for_method -from typing import Any, Dict, List, Optional -from pathlib import Path -import re -import time +from hermes_cli.web_deps import LateState, late +from hermes_cli.web_routers._common import http_failure _log = logging.getLogger("hermes_cli.web_server") router = APIRouter() @@ -36,15 +38,11 @@ _ACTION_PROCS = LateState("_ACTION_PROCS") _ACTION_RESULTS = LateState("_ACTION_RESULTS") -def _action_log_dir() -> Path: - """Live ``web_server._ACTION_LOG_DIR`` (a Path value, so not proxied by LateState).""" - from hermes_cli.web_server import _ACTION_LOG_DIR - return _ACTION_LOG_DIR - - -def _project_root() -> Path: - from hermes_cli.web_server import PROJECT_ROOT - return PROJECT_ROOT +def _server_path(name: str) -> Path: + """Live ``web_server.`` Path value (``_ACTION_LOG_DIR`` / ``PROJECT_ROOT``; plain + values are not proxied by LateState).""" + import hermes_cli.web_server as ws + return getattr(ws, name) _ACTION_LOG_TAIL_MAX_BYTES = 256 * 1024 @@ -53,29 +51,29 @@ _ACTION_LOG_TAIL_MAX_CHUNK_BYTES = 64 * 1024 _UPDATE_ACTION_COMPLETED_RE = re.compile(r"^=== hermes-update completed ([0-9a-f]{32}) ===$") -_MANAGED_EXTERNALLY_MESSAGE = ( - "Hermes updates are managed outside this dashboard in " - "containerized environments." -) +_MANAGED_EXTERNALLY_MESSAGE = "Hermes updates are managed outside this dashboard in containerized environments." + +# Per-kind dashboard error codes the UI keys on, by admission-refusal code. +_UPDATE_REFUSAL_ERROR_CODES = { + "docker": "docker_update_unsupported", "image-marker": "docker_update_unsupported", + "image-marker-invalid": "docker_update_unsupported", "apt": "apt_update_required", + "nix": "nix_update_unsupported", +} def _finish_action(name: str, exit_code: Optional[int], pid: Optional[int]) -> None: """Record a terminal result and drop the live-process registries for ``name``.""" _ACTION_RESULTS[name] = {"exit_code": exit_code, "pid": pid} - _ACTION_PROCS.pop(name, None) - _ACTION_COMMANDS.pop(name, None) - _ACTION_IDS.pop(name, None) + for registry in (_ACTION_PROCS, _ACTION_COMMANDS, _ACTION_IDS): + registry.pop(name, None) def _record_completed_action(name: str, message: str, exit_code: int = 1) -> None: """Record a non-spawned action result and write it to the action log.""" - log_dir = _action_log_dir() + log_dir = _server_path("_ACTION_LOG_DIR") log_dir.mkdir(parents=True, exist_ok=True) - log_path = log_dir / _ACTION_LOG_FILES[name] - with open(log_path, "ab", buffering=0) as log_file: - log_file.write( - f"\n=== {name} completed {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n".encode() - ) + with open(log_dir / _ACTION_LOG_FILES[name], "ab", buffering=0) as log_file: + log_file.write(f"\n=== {name} completed {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n".encode()) log_file.write(message.encode("utf-8", errors="replace")) if not message.endswith("\n"): log_file.write(b"\n") @@ -84,13 +82,11 @@ def _record_completed_action(name: str, message: str, exit_code: int = 1) -> Non def _tail_lines(path: Path, n: int) -> List[str]: """Return the last ``n`` lines of ``path`` without loading huge logs.""" - if n <= 0 or not path.exists(): - return [] try: size = path.stat().st_size except OSError: return [] - if size <= 0: + if n <= 0 or size <= 0: return [] min_offset = max(0, size - _ACTION_LOG_TAIL_MAX_BYTES) @@ -99,7 +95,6 @@ def _tail_lines(path: Path, n: int) -> List[str]: newline_count = 0 chunks: List[bytes] = [] drop_partial_first_line = False - try: with path.open("rb") as handle: while offset > min_offset and newline_count <= n: @@ -123,30 +118,20 @@ def _tail_lines(path: Path, n: int) -> List[str]: def _durable_completed_update_action_id(lines: List[str]) -> Optional[str]: - """Recover the latest successful update identity from ``update.log``. - - The dashboard action process can restart the dashboard that spawned it, - losing the in-memory ``Popen``/result registries while the durable log - survives. Only a completion marker after the latest update-start marker - counts, so a stale success cannot mask a newer failed attempt. - """ - last_start = -1 - last_completed = -1 + """Latest successful update id from ``update.log`` — the durable record that survives + the update restarting the dashboard (losing the in-memory ``Popen``/result registries). + Only a completion marker after the latest start marker counts, so a stale success + cannot mask a newer failed attempt.""" + last_start = last_completed = -1 completed_action_id: Optional[str] = None - for index, line in enumerate(lines): if line.startswith("=== hermes update started "): last_start = index - match = _UPDATE_ACTION_COMPLETED_RE.fullmatch(line.strip()) if match: last_completed = index completed_action_id = match.group(1) - - if completed_action_id and last_completed > last_start: - return completed_action_id - - return None + return completed_action_id if completed_action_id and last_completed > last_start else None @router.post("/api/gateway/restart") @@ -161,178 +146,135 @@ async def restart_gateway(profile: Optional[str] = None): async def gateway_drain(request: Request): """Begin or cancel an external (NAS-driven) gateway drain. - Authenticated by the non-interactive token-auth seam: the - ``dashboard_auth/drain`` plugin registers this path as a token route and - verifies the bearer secret. Without that plugin (no - ``HERMES_DASHBOARD_DRAIN_SECRET``) the cookie gate covers a gated bind and - the legacy session-token gate a loopback bind — never unauthenticated on a - network-exposed bind. + Authenticated by the non-interactive token-auth seam (the ``dashboard_auth/drain`` + plugin registers this path as a token route and verifies the bearer secret); + without that plugin the cookie gate covers a gated bind and the legacy + session-token gate a loopback bind — never unauthenticated on a network bind. - Body: ``{"action": "drain"}`` or ``{"action": "cancel"}``. Begin writes the - ``.drain_request.json`` marker the gateway's ``_drain_control_watcher`` - observes (flip to ``draining`` + refuse new turns); cancel removes it. - Idempotent on both sides. Only the marker is written here — the gateway - process owns the state transition (the marker IS the control channel). The - force-override is ``POST /api/gateway/restart``, which supersedes a drain. + Body ``{"action": "drain"|"cancel"}``. Only the ``.drain_request.json`` marker is + written/removed here — the gateway's ``_drain_control_watcher`` owns the state + transition (the marker IS the control channel). Idempotent on both sides; + ``POST /api/gateway/restart`` is the force-override that supersedes a drain. """ - from gateway.drain_control import ( - clear_drain_request, drain_requested, write_drain_request, - ) + from gateway.drain_control import clear_drain_request, drain_requested, write_drain_request try: body = await request.json() except Exception: body = {} - action = str((body or {}).get("action", "drain")).strip().lower() - - # Attribute the request to the verified token principal when present - # (token-auth seam attaches it); fall back to a generic label otherwise. - principal_obj = getattr(request.state, "token_principal", None) - principal = getattr(principal_obj, "principal", None) or "dashboard" + body = body or {} + action = str(body.get("action", "drain")).strip().lower() + # Attribute to the verified token principal when the token-auth seam attached one. + principal = getattr(getattr(request.state, "token_principal", None), "principal", None) or "dashboard" if action == "cancel": existed = clear_drain_request() _log.info("Gateway drain CANCEL requested by %s (existed=%s)", principal, existed) return {"ok": True, "action": "cancel", "was_draining": existed} - if action != "drain": - raise HTTPException( - status_code=400, - detail=f"Unknown drain action {action!r}; expected 'drain' or 'cancel'", - ) + raise HTTPException(status_code=400, detail=f"Unknown drain action {action!r}; expected 'drain' or 'cancel'") payload = write_drain_request( - principal=str(principal), - suppress_notification=bool((body or {}).get("suppress_notification", False)), + principal=str(principal), suppress_notification=bool(body.get("suppress_notification", False)), ) _log.info( - "Gateway drain BEGIN requested by %s (suppress_notification=%s)", principal, - payload["suppress_notification"], + "Gateway drain BEGIN requested by %s (suppress_notification=%s)", principal, payload["suppress_notification"], ) return { - "ok": True, - "action": "drain", - "requested_at": payload["requested_at"], + "ok": True, "action": "drain", "requested_at": payload["requested_at"], # Echo so a caller polling /api/status knows the marker is now set; # the gateway watcher flips gateway_state -> draining within ~1s. - "draining": drain_requested(), - "suppress_notification": payload["suppress_notification"], + "draining": drain_requested(), "suppress_notification": payload["suppress_notification"], } def _update_refused(error: str, message: str, update_command: str) -> Dict[str, Any]: + _record_completed_action("hermes-update", message, exit_code=1) return { "ok": False, "pid": None, "name": "hermes-update", "error": error, "message": message, "update_command": update_command, } -# Per-kind dashboard error codes the UI keys on, by admission-refusal code. -_UPDATE_REFUSAL_ERROR_CODES = { - "docker": "docker_update_unsupported", "image-marker": "docker_update_unsupported", - "image-marker-invalid": "docker_update_unsupported", "apt": "apt_update_required", - "nix": "nix_update_unsupported", -} - - @router.post("/api/hermes/update") async def update_hermes(): """Kick off ``hermes update`` in the background.""" if _dashboard_local_update_managed_externally(): message = _MANAGED_EXTERNALLY_MESSAGE + " The built-in local updater is disabled here." - _record_completed_action("hermes-update", message, exit_code=1) return _update_refused("dashboard_update_managed_externally", message, "managed outside dashboard") # Shared admission gate: marker-first, then the docker/nix/apt heuristics — # one decision with the CLI paths. - from hermes_cli.update_contract import ( - evaluate_update_admission, record_refusal_receipt, - ) + from hermes_cli.update_contract import evaluate_update_admission, record_refusal_receipt - refusal = evaluate_update_admission(_project_root()) + refusal = evaluate_update_admission(_server_path("PROJECT_ROOT")) if refusal is not None: - _record_completed_action("hermes-update", refusal.message, exit_code=1) + response = _update_refused( + _UPDATE_REFUSAL_ERROR_CODES.get(refusal.code, "update_not_in_place"), refusal.message, refusal.update_command, + ) record_refusal_receipt(refusal) - error_code = _UPDATE_REFUSAL_ERROR_CODES.get(refusal.code, "update_not_in_place") - return _update_refused(error_code, refusal.message, refusal.update_command) + return response existing = _ACTION_PROCS.get("hermes-update") if existing is not None and existing.poll() is None: - response = { - "ok": True, "pid": existing.pid, "name": "hermes-update", "already_running": True, - } + response = {"ok": True, "pid": existing.pid, "name": "hermes-update", "already_running": True} action_id = _ACTION_IDS.get("hermes-update") if action_id: response["action_id"] = action_id return response action_id = secrets.token_hex(16) - try: - proc = _spawn_hermes_action( - ["update"], "hermes-update", env_overrides={"HERMES_ACTION_ID": action_id}, - ) - except Exception as exc: - _log.exception("Failed to spawn hermes update") - raise HTTPException(status_code=500, detail=f"Failed to start update: {exc}") + with http_failure("Failed to spawn hermes update", 500, "Failed to start update"): + proc = _spawn_hermes_action(["update"], "hermes-update", env_overrides={"HERMES_ACTION_ID": action_id}) return {"ok": True, "pid": proc.pid, "name": "hermes-update", "action_id": action_id} def _recent_upstream_commits(n: int = 20) -> List[Dict[str, Any]]: - """Commits the local checkout is behind ``origin/main`` by, newest first. + """Commits the local checkout is behind ``origin/main`` by, newest first; [] on any failure. - Logs the SAME range the behind-count uses (``HEAD..origin/main`` — see - ``banner._check_via_local_git``), NOT ``@{upstream}``: on a feature-branch - checkout that is the branch's own tip (zero commits), leaving the changelog - empty while the count is non-zero. Best-effort: [] if not a git checkout, - origin/main unreachable, or git unavailable. Never raises. + Logs the SAME range the behind-count uses (``HEAD..origin/main``, see + ``banner._check_via_local_git``), NOT ``@{upstream}``: on a feature branch that is + the branch's own tip (zero commits), leaving the changelog empty while the count is non-zero. """ try: + # git log emits UTF-8 (emoji/CJK subjects). On Windows text=True defaults to + # the ANSI code page; an undefined cp1252 byte crashed the stdlib + # _readerthread and killed the desktop backend — hence encoding="utf-8". out = subprocess.run( [ - "git", "-C", str(_project_root()), "log", "--format=%H%x1f%s%x1f%an%x1f%ct", + "git", "-C", str(_server_path("PROJECT_ROOT")), "log", "--format=%H%x1f%s%x1f%an%x1f%ct", "HEAD..origin/main", f"-n{int(n)}", ], - capture_output=True, - text=True, - # git log emits UTF-8 (emoji/CJK subjects). On Windows text=True - # defaults to the ANSI code page; an undefined cp1252 byte crashed - # the stdlib _readerthread and killed the desktop backend. - encoding="utf-8", - errors="replace", - timeout=5, + capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5, ) if out.returncode != 0: return [] rows: List[Dict[str, Any]] = [] for line in out.stdout.splitlines(): - if not line.strip(): - continue - sha, summary, author, at = (line.split("\x1f") + ["", "", "", "0"])[:4] - rows.append({"sha": sha[:7], "summary": summary, "author": author, "at": int(at or 0)}) + if line.strip(): + sha, summary, author, at = (line.split("\x1f") + ["", "", "", "0"])[:4] + rows.append({"sha": sha[:7], "summary": summary, "author": author, "at": int(at or 0)}) return rows except Exception: return [] +_NON_APPLYABLE_MESSAGES = { + "docker": format_docker_update_message, + "apt": lambda: "Hermes is managed by Termux APT; run `pkg upgrade hermes-agent`.", +} + + @router.get("/api/hermes/update/check") async def check_hermes_update(force: bool = False): """Report whether a Hermes update is available, without applying it. - Powers the dashboard's "check before you update" flow. - - Returns: - install_method: 'apt' | 'git' | 'docker' | 'nix' | 'nixos' | 'unknown' - current_version: installed Hermes version string - behind: commits behind upstream (>=1), 0 if up to date, -1 if behind by - an unknown count, or null if the check could not run - update_available: convenience bool (behind is non-zero and not null) - can_apply: True when the dashboard's update button can apply it in - place (git); False where the user must update out-of-band - update_command: the recommended command for this install method - message: human-readable guidance for non-applyable methods - commits: for git installs that are behind, the commits the checkout is - behind by — each {sha, summary, author, at}. Absent/empty - otherwise; additive, existing consumers ignore it. + Returns install_method ('apt'|'git'|'docker'|'nix'|'nixos'|'unknown'), + current_version, behind (commits behind, 0 = up to date, -1 = unknown count, + null = check could not run), update_available, can_apply (git only — the + dashboard button can apply in place), update_command, message (guidance for + non-applyable methods) and, for git installs that are behind, commits + [{sha, summary, author, at}] (additive; existing consumers ignore it). """ if _dashboard_local_update_managed_externally(): return { @@ -341,20 +283,15 @@ async def check_hermes_update(force: bool = False): "update_command": "managed outside dashboard", "message": _MANAGED_EXTERNALLY_MESSAGE, } - install_method = detect_install_method(_project_root()) - update_command = recommended_update_command_for_method(install_method) - + install_method = detect_install_method(_server_path("PROJECT_ROOT")) payload: Dict[str, Any] = { "install_method": install_method, "current_version": __version__, "behind": None, "update_available": False, "can_apply": install_method == "git", - "update_command": update_command, "message": None, + "update_command": recommended_update_command_for_method(install_method), "message": None, } - - if install_method == "docker": - payload["message"] = format_docker_update_message() - return payload - if install_method == "apt": - payload["message"] = "Hermes is managed by Termux APT; run `pkg upgrade hermes-agent`." + non_applyable = _NON_APPLYABLE_MESSAGES.get(install_method) + if non_applyable is not None: + payload["message"] = non_applyable() return payload # banner.check_for_updates() handles git / nix-revision paths and caches @@ -363,11 +300,8 @@ async def check_hermes_update(force: bool = False): from hermes_cli.banner import check_for_updates if force: - try: + with contextlib.suppress(OSError): (get_hermes_home() / ".update_check").unlink() - except OSError: - pass - behind = await asyncio.to_thread(check_for_updates) except Exception: _log.exception("Update check failed") @@ -384,10 +318,25 @@ async def check_hermes_update(force: bool = False): # best-effort (empty list on any failure). if install_method == "git": payload["commits"] = await asyncio.to_thread(_recent_upstream_commits) - return payload +def _completed_exit_code( + result: Optional[Dict[str, Any]], durable_action_id: Optional[str], receipt: Optional[Dict[str, Any]], +) -> Optional[int]: + """Exit code for an action with no live process: in-memory result, else durable evidence.""" + if result is not None: + return result.get("exit_code") + if durable_action_id: + return 0 + if receipt is not None and receipt.get("outcome") in ("success", "partial"): + # No in-memory result and no log marker (e.g. log rotated), but the + # receipt proves a completed run: report its outcome rather than a + # null clients time out on. ``partial`` maps to exit 1 like the CLI. + return 0 if receipt["outcome"] == "success" else 1 + return None + + @status_router.get("/api/actions/{name}/status") async def get_action_status(name: str, lines: int = 200): """Tail an action log and report whether the process is still running.""" @@ -395,15 +344,14 @@ async def get_action_status(name: str, lines: int = 200): if log_file_name is None: raise HTTPException(status_code=404, detail=f"Unknown action: {name}") - log_dir = _action_log_dir() + log_dir = _server_path("_ACTION_LOG_DIR") requested_lines = min(max(lines, 1), 2000) tail = _tail_lines(log_dir / log_file_name, requested_lines) durable_update_action_id = None update_receipt_summary = None if name == "hermes-update": - durable_lines = _tail_lines(log_dir / "update.log", 2000) - durable_update_action_id = _durable_completed_update_action_id(durable_lines) + durable_update_action_id = _durable_completed_update_action_id(_tail_lines(log_dir / "update.log", 2000)) if durable_update_action_id: marker = f"=== hermes-update completed {durable_update_action_id} ===" if marker not in tail: @@ -418,34 +366,18 @@ async def get_action_status(name: str, lines: int = 200): if proc is None: result = _ACTION_RESULTS.get(name) running = False - exit_code = result.get("exit_code") if result else None pid = result.get("pid") if result else None - if result is None and durable_update_action_id: - exit_code = 0 - if ( - result is None - and exit_code is None - and update_receipt_summary is not None - and update_receipt_summary.get("outcome") in ("success", "partial") - ): - # No in-memory result and no log marker (e.g. log rotated), but the - # receipt proves a completed run: report its outcome rather than a - # null clients time out on. ``partial`` maps to exit 1 like the CLI. - exit_code = 0 if update_receipt_summary["outcome"] == "success" else 1 + exit_code = _completed_exit_code(result, durable_update_action_id, update_receipt_summary) else: exit_code = proc.poll() running = exit_code is None pid = proc.pid if exit_code is not None: - try: + with contextlib.suppress(Exception): proc.wait(timeout=1) - except Exception: - pass _finish_action(name, exit_code, pid) - response = { - "name": name, "running": running, "exit_code": exit_code, "pid": pid, "lines": tail, - } + response = {"name": name, "running": running, "exit_code": exit_code, "pid": pid, "lines": tail} if durable_update_action_id: response["action_id"] = durable_update_action_id if update_receipt_summary is not None: @@ -457,36 +389,24 @@ def _read_latest_receipt() -> Optional[Dict[str, Any]]: """Latest update receipt, or None on any failure (never raises).""" try: from hermes_cli.update_receipt import read_latest_receipt - return read_latest_receipt() or None except Exception: return None def _latest_update_receipt_summary() -> Optional[Dict[str, Any]]: - """Compact summary of the most recent update receipt, or None. - - The receipt (written by EVERY ``hermes update`` run, including refused and - failed ones, with a ``latest.json`` pointer) is the durable success signal - the Desktop and dashboard read instead of inferring outcomes from liveness - probes across the update's stop/start gap. Summary only — steps and skips - stay in the full receipt endpoint. Never raises. - """ + """Compact summary of the latest receipt (written by EVERY ``hermes update`` run, + incl. refused/failed), or None; never raises. Steps/skips stay in the full endpoint.""" receipt = _read_latest_receipt() if not receipt: return None try: - fleet = receipt.get("fleet") or [] + post = receipt.get("post_update") or {} return { - "outcome": receipt.get("outcome"), - "started_at": receipt.get("started_at"), - "finished_at": receipt.get("finished_at"), + **{k: receipt.get(k) for k in ("outcome", "started_at", "finished_at")}, "pre_sha": (receipt.get("pre_update") or {}).get("sha"), - "post_sha": (receipt.get("post_update") or {}).get("sha"), - "post_version": (receipt.get("post_update") or {}).get("version"), - "fleet_states": sorted( - {str(e.get("state")) for e in fleet if isinstance(e, dict)} - ), + "post_sha": post.get("sha"), "post_version": post.get("version"), + "fleet_states": sorted({str(e.get("state")) for e in receipt.get("fleet") or [] if isinstance(e, dict)}), } except Exception: return None @@ -494,17 +414,11 @@ def _latest_update_receipt_summary() -> Optional[Dict[str, Any]]: @status_router.get("/api/hermes/update/receipt") async def get_update_receipt(): - """The most recent update receipt — the durable update-outcome record. - - Dashboards and the Desktop read this instead of inferring update success - from backend liveness (which misread the update's own restart gap as a - failed update/boot). Returns the FULL receipt (steps, skips, gateway - restart outcome, fleet matrix) plus a compact ``summary``; 404 when no - update has run since receipts landed. - """ + """The FULL latest update receipt (steps, skips, gateway restart outcome, fleet + matrix) plus a compact ``summary``; 404 when no update has run since receipts landed. + Clients read this instead of inferring success from backend liveness, which misread + the update's own restart gap as a failed update/boot.""" receipt = _read_latest_receipt() if not receipt: - raise HTTPException( - status_code=404, detail="No update receipt found (no `hermes update` run recorded).", - ) + raise HTTPException(status_code=404, detail="No update receipt found (no `hermes update` run recorded).") return {"receipt": receipt, "summary": _latest_update_receipt_summary()} diff --git a/hermes_cli/web_routers/memory_providers.py b/hermes_cli/web_routers/memory_providers.py index d0730abdef..77a2a288f2 100644 --- a/hermes_cli/web_routers/memory_providers.py +++ b/hermes_cli/web_routers/memory_providers.py @@ -4,21 +4,24 @@ Extracted from ``hermes_cli.web_server``; helpers/state that tests monkeypatch o ``web_server`` stay there and are late-bound (cycle-safe). """ +import contextlib +import json import logging import math import re import shlex -from fastapi import APIRouter -from hermes_cli.web_deps import late -from hermes_cli.web_routers._common import scoped_to_thread -from fastapi import HTTPException -from hermes_cli.web_models import MemoryProviderConfigUpdate, MemoryProviderSetupRequest -from plugins.memory.config_schema import get_provider_config_schema -from typing import Any, Dict, List, Optional -from plugins.memory.config_schema import ProviderConfigSchema, ProviderField, STORAGE_HONCHO_HOST_BLOCK import subprocess -import json from pathlib import Path +from typing import Any, Dict, List, Optional + +from fastapi import APIRouter, HTTPException + +from hermes_cli.web_deps import late +from hermes_cli.web_models import MemoryProviderConfigUpdate, MemoryProviderSetupRequest +from hermes_cli.web_routers._common import scoped_to_thread +from plugins.memory.config_schema import ( + STORAGE_HONCHO_HOST_BLOCK, ProviderConfigSchema, ProviderField, get_provider_config_schema, +) _log = logging.getLogger("hermes_cli.web_server") router = APIRouter() @@ -49,76 +52,86 @@ load_env = late("load_env") # Sentinel: remove this key so it falls back to the host or built-in default. _UNSET: Any = object() +_MEMORY_PROVIDER_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$") + + +def _unknown_provider(name: str) -> HTTPException: + return HTTPException(status_code=404, detail=f"Unknown memory provider: {name}") + + +@contextlib.contextmanager +def _value_errors_as_http(log_msg: str, name: str, *, passthrough_http: bool = True): + """``ValueError`` -> 400 with its text; any other error -> logged 500 (an + ``HTTPException`` passes through unless ``passthrough_http`` is False).""" + try: + yield + except HTTPException: + if passthrough_http: + raise + _log.exception(log_msg, name) + raise HTTPException(status_code=500, detail="Internal server error") + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except Exception: + _log.exception(log_msg, name) + raise HTTPException(status_code=500, detail="Internal server error") + + +# ── Declared-schema surface (plugins.memory.config_schema) ──────────────────── def _coerce_field_value(field: ProviderField, raw: str) -> Any: """Coerce a submitted non-secret string to its native JSON type. - A bool is stored as JSON ``false`` rather than ``"false"`` (truthy). Returns - ``_UNSET`` when the field should be removed; raises ``ValueError`` on - malformed input. + A bool is stored as JSON ``false`` rather than ``"false"`` (truthy). Blank + number/json/text clears the key (``_UNSET``); raises ``ValueError`` on malformed input. """ value = (raw or "").strip() kind = field.kind - if kind == "select": - if not value: - value = field.default + value = value or field.default if value not in field.allowed_values(): raise ValueError(f"Invalid value for '{field.key}'") return value - if kind == "bool": from utils import is_truthy_value - return is_truthy_value(value) - if not value: - # number / json / text / secret — blank clears the key. return _UNSET - if kind == "number": try: number = float(value) except ValueError as exc: raise ValueError(f"Invalid number for '{field.key}'") from exc return int(number) if number.is_integer() else number - if kind == "json": try: parsed = json.loads(value) except (ValueError, TypeError) as exc: raise ValueError(f"Invalid JSON for '{field.key}'") from exc - if not isinstance(parsed, (dict, list)): - raise ValueError(f"'{field.key}' must be a JSON object or array") - return parsed - + if isinstance(parsed, (dict, list)): + return parsed + raise ValueError(f"'{field.key}' must be a JSON object or array") return value -def _read_json_dict(path: Path, log_msg: str) -> Dict[str, Any]: +def _read_json_dict(path: Path, what: str) -> Dict[str, Any]: """Read a JSON object from ``path``; missing/unreadable/non-dict -> ``{}``.""" if not path.exists(): return {} try: data = json.loads(path.read_text(encoding="utf-8")) except Exception: - _log.warning(log_msg, path, exc_info=True) + _log.warning("Failed to read %s from %s", what, path, exc_info=True) return {} return data if isinstance(data, dict) else {} -# — flat-json backend (default; reusable for simple providers) — - - def _flat_json_path(provider: ProviderConfigSchema) -> Path: return get_hermes_home() / provider.name / "config.json" def _read_flat_json(provider: ProviderConfigSchema) -> Dict[str, Any]: - return _read_json_dict(_flat_json_path(provider), "Failed to read memory provider config from %s") - - -# — honcho host-block backend — + return _read_json_dict(_flat_json_path(provider), "memory provider config") def _honcho_resolvers(): @@ -127,8 +140,18 @@ def _honcho_resolvers(): return resolve_active_host, resolve_config_path, _host_block -def _read_honcho_config(path: Path) -> Dict[str, Any]: - return _read_json_dict(path, "Failed to read Honcho config from %s") +def _save_submitted_secrets(provider: ProviderConfigSchema, values: Dict[str, str]) -> list: + """Persist each non-blank secret submission to the env store (when the field has an + ``env_key``); return the ``(field, submitted)`` pairs for backend-specific handling.""" + saved = [] + for field in provider.fields: + submitted = (values.get(field.key) or "").strip() if field.is_secret else "" + if not submitted: + continue + if field.env_key: + save_env_value(field.env_key, submitted) + saved.append((field, submitted)) + return saved def _apply_field_values(provider: ProviderConfigSchema, values: Dict[str, str], target_for) -> None: @@ -150,46 +173,56 @@ def _apply_field_values(provider: ProviderConfigSchema, values: Dict[str, str], target[field.key] = coerced -def _trim_setup_output(value: Optional[str], limit: int = 4000) -> str: - text = str(value or "") - if len(text) <= limit: - return text - return f"{text[:limit]}\n... truncated ..." +def _write_json_0600(path: Path, data: Dict[str, Any]) -> None: + from utils import atomic_json_write + path.parent.mkdir(parents=True, exist_ok=True) + atomic_json_write(path, data, mode=0o600) -# ── Memory provider config: one generic GET/PUT pair, dispatching on storage ── +def _write_provider_flat(provider: ProviderConfigSchema, values: Dict[str, str]) -> None: + existing = _read_flat_json(provider) + _save_submitted_secrets(provider, values) + _apply_field_values(provider, values, lambda field: existing) + _write_json_0600(_flat_json_path(provider), existing) -def _provider_field_entry(field: ProviderField) -> Dict[str, Any]: - """Static, storage-independent shape of one field for the UI payload.""" - return { - "key": field.key, - "label": field.label, - "kind": field.kind, - "description": field.description, - "info": field.info, - "placeholder": field.placeholder, - "inline": field.inline, - "group": field.group, - "options": [ - {"value": opt.value, "label": opt.label, "description": opt.description} - for opt in field.options - ], - } +def _write_provider_honcho(provider: ProviderConfigSchema, values: Dict[str, str]) -> None: + """Persist submitted fields to Honcho's real config for the active host (partial + saves touch only submitted keys; blank text clears a key — see ``_apply_field_values``).""" + from plugins.memory.honcho.oauth import ACCESS_TOKEN_PREFIX, _config_refresh_lock + + resolve_active_host, resolve_config_path, host_block_of = _honcho_resolvers() + host = resolve_active_host() + # Write the file reads resolve, or a save shadows it with a sparse copy. + path = resolve_config_path() + + # OAuth rotation is single-use; an unlocked RMW here can revoke the grant. + with _config_refresh_lock(path): + cfg = _read_json_dict(path, "Honcho config") + hosts = cfg.get("hosts") + cfg["hosts"] = hosts = hosts if isinstance(hosts, dict) else {} + # Update the block reads resolve (legacy dot-form included), never shadow it. + existing = host_block_of(cfg, host) + host_key = next((k for k, v in hosts.items() if v is existing), host) if existing else host + host_block = hosts.setdefault(host_key, existing) + + for field, submitted in _save_submitted_secrets(provider, values): + # Persist where the client reads first; an OAuth token owns that slot. + stored = host_block.get(field.key) + if not (isinstance(stored, str) and stored.startswith(ACCESS_TOKEN_PREFIX)): + host_block[field.key] = submitted + + _apply_field_values(provider, values, lambda field: host_block if field.scope == "host" else cfg) + _write_json_0600(path, cfg) def _serialize_field_value(field: ProviderField, value: Any) -> str: - """Render a stored native value as the string the generic UI edits. - - ``None`` (key absent) yields the field's declared default; bools become - ``"true"``/``"false"``, JSON objects/arrays are re-encoded, numbers are - stringified. - """ + """Render a stored native value as the string the generic UI edits (``None`` = key + absent -> declared default; bools -> "true"/"false"; JSON containers re-encoded).""" if value is None: return field.default if field.kind == "bool": from utils import is_truthy_value - return "true" if is_truthy_value(value) else "false" if field.kind == "json" and isinstance(value, (dict, list)): return json.dumps(value) @@ -197,7 +230,7 @@ def _serialize_field_value(field: ProviderField, value: Any) -> str: def _read_field(field: ProviderField, sources: tuple, env: Dict[str, str]) -> Any: - """Return the stored native value from the first source holding it, or ``None``. + """Stored native value from the first source holding it, else ``None``. Presence (``key in source``) decides, not truthiness, so a stored ``False`` or ``0`` survives instead of being mistaken for "unset". @@ -207,87 +240,92 @@ def _read_field(field: ProviderField, sources: tuple, env: Dict[str, str]) -> An if source_key in source and source[source_key] is not None: return source[source_key] for env_key in field.env_fallbacks: - value = env.get(env_key) - if value: - return value + if env.get(env_key): + return env[env_key] return None def _declared_field_is_set(field: ProviderField, sources: tuple, env: Dict[str, str]) -> bool: - for env_key in (field.env_key, *field.env_fallbacks): - if env_key and env.get(env_key): - return True + if any(env_key and env.get(env_key) for env_key in (field.env_key, *field.env_fallbacks)): + return True return any(source.get(k) for source in sources for k in (field.key, *field.aliases)) -def _honcho_read_sources() -> tuple[Dict[str, Any], str, Dict[str, Any]]: - """Return (root config, active host key, host block) for the current profile.""" - resolve_active_host, resolve_config_path, host_block_of = _honcho_resolvers() - host = resolve_active_host() - raw = _read_honcho_config(resolve_config_path()) - return raw, host, host_block_of(raw, host) +def _declared_provider_payload(provider: ProviderConfigSchema) -> Dict[str, Any]: + env = load_env() + is_honcho = provider.storage == STORAGE_HONCHO_HOST_BLOCK + if is_honcho: + resolve_active_host, resolve_config_path, host_block_of = _honcho_resolvers() + host = resolve_active_host() + raw = _read_json_dict(resolve_config_path(), "Honcho config") + host_block = host_block_of(raw, host) + def sources_for(field: ProviderField) -> tuple: + return (host_block, raw) if field.scope == "host" else (raw,) + else: + host, data = "", _read_flat_json(provider) -def _write_provider_flat(provider: ProviderConfigSchema, values: Dict[str, str]) -> None: - from utils import atomic_json_write - - existing = _read_flat_json(provider) + def sources_for(field: ProviderField) -> tuple: + return (data,) + fields: List[Dict[str, Any]] = [] for field in provider.fields: + entry = {k: getattr(field, k) for k in ("key", "label", "kind", "description", "info", "placeholder", "inline", "group")} + entry["options"] = [{"value": o.value, "label": o.label, "description": o.description} for o in field.options] + sources = sources_for(field) if field.is_secret: - submitted = (values.get(field.key) or "").strip() - if submitted and field.env_key: - save_env_value(field.env_key, submitted) - - _apply_field_values(provider, values, lambda field: existing) - - path = _flat_json_path(provider) - path.parent.mkdir(parents=True, exist_ok=True) - atomic_json_write(path, existing, mode=0o600) + entry["value"] = "" # secrets are write-only over the API + entry["is_set"] = _declared_field_is_set(field, sources, env) + fields.append(entry) + continue + native = _read_field(field, sources, env) + if is_honcho and not field.placeholder and field.key in {"workspace", "aiPeer"}: + # Blank fields surface the resolved host Honcho will actually use. + entry["placeholder"] = host + value = _serialize_field_value(field, native) + if field.kind == "select" and value not in field.allowed_values(): + value = field.default + entry["value"] = value + # Presence, not truthiness — a stored False/0 is still "set". + entry["is_set"] = native is not None if is_honcho else bool(value) + fields.append(entry) + return {"name": provider.name, "label": provider.label, "docs_url": provider.docs_url, "fields": fields} -def _write_provider_honcho(provider: ProviderConfigSchema, values: Dict[str, str]) -> None: - """Persist submitted fields to Honcho's real config for the active host. +def _stringify_submitted(value: Any) -> str: + """The declared-schema path edits strings; the dashboard may send natives.""" + if value is None: + return "" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, (dict, list)): + return json.dumps(value) + return str(value) - Only keys present in ``values`` are touched, so a partial save (e.g. the - inline panel) never clobbers fields owned by the full-config editor. Blank - text clears a key so it falls back to the host/default mapping. - """ - from plugins.memory.honcho.oauth import ACCESS_TOKEN_PREFIX, _config_refresh_lock - from utils import atomic_json_write - resolve_active_host, resolve_config_path, host_block_of = _honcho_resolvers() - host = resolve_active_host() - # Write the file reads resolve, or a save shadows it with a sparse copy. - path = resolve_config_path() +def _memory_section(config: Dict[str, Any]) -> Dict[str, Any]: + """Return ``config["memory"]`` as a dict, creating/replacing a non-dict value.""" + memory_config = config.get("memory") + if not isinstance(memory_config, dict): + memory_config = config["memory"] = {} + return memory_config - # OAuth rotation is single-use; an unlocked RMW here can revoke the grant. - with _config_refresh_lock(path): - cfg = _read_honcho_config(path) - hosts = cfg.get("hosts") - cfg["hosts"] = hosts = hosts if isinstance(hosts, dict) else {} - # Update the block reads resolve (legacy dot-form included), never shadow it. - existing = host_block_of(cfg, host) - host_key = next((k for k, v in hosts.items() if v is existing), host) if existing else host - host_block = hosts.setdefault(host_key, existing) - for field in provider.fields: - if not field.is_secret: - continue - submitted = (values.get(field.key) or "").strip() - if not submitted: - continue - if field.env_key: - save_env_value(field.env_key, submitted) - # Persist where the client reads first; an OAuth token owns that slot. - stored = host_block.get(field.key) - if not (isinstance(stored, str) and stored.startswith(ACCESS_TOKEN_PREFIX)): - host_block[field.key] = submitted +def _update_memory_provider_config(provider: ProviderConfigSchema, values: Dict[str, str]) -> None: + writer = _write_provider_honcho if provider.storage == STORAGE_HONCHO_HOST_BLOCK else _write_provider_flat + writer(provider, values) + config = load_config() + memory_config = _memory_section(config) + if memory_config.get("provider") != provider.name: + memory_config["provider"] = provider.name + save_config(config) - _apply_field_values(provider, values, lambda field: host_block if field.scope == "host" else cfg) - path.parent.mkdir(parents=True, exist_ok=True) - atomic_json_write(path, cfg, mode=0o600) +# ── Setup: dependency installation ──────────────────────────────────────────── + +def _trim_setup_output(value: Optional[str], limit: int = 4000) -> str: + text = str(value or "") + return text if len(text) <= limit else f"{text[:limit]}\n... truncated ..." def _command_result( @@ -302,99 +340,12 @@ def _command_result( } -def _declared_provider_payload(provider: ProviderConfigSchema) -> Dict[str, Any]: - fields: List[Dict[str, Any]] = [] - env = load_env() - is_honcho = provider.storage == STORAGE_HONCHO_HOST_BLOCK - - if is_honcho: - raw, host, host_block = _honcho_read_sources() - - def sources_for(field: ProviderField) -> tuple: - return (host_block, raw) if field.scope == "host" else (raw,) - else: - host = "" - data = _read_flat_json(provider) - - def sources_for(field: ProviderField) -> tuple: - return (data,) - - for field in provider.fields: - entry = _provider_field_entry(field) - sources = sources_for(field) - - if field.is_secret: - entry["value"] = "" # secrets are write-only over the API - entry["is_set"] = _declared_field_is_set(field, sources, env) - fields.append(entry) - continue - - native = _read_field(field, sources, env) - if is_honcho and not field.placeholder and field.key in {"workspace", "aiPeer"}: - # Blank fields surface the resolved host Honcho will actually use. - entry["placeholder"] = host - - value = _serialize_field_value(field, native) - if field.kind == "select" and value not in field.allowed_values(): - value = field.default - entry["value"] = value - # Presence, not truthiness — a stored False/0 is still "set". - entry["is_set"] = native is not None if is_honcho else bool(value) - fields.append(entry) - - return {"name": provider.name, "label": provider.label, "docs_url": provider.docs_url, "fields": fields} - - -def _stringify_submitted_values(values: Dict[str, Any]) -> Dict[str, str]: - """The declared-schema path edits strings; the dashboard may send natives.""" - out: Dict[str, str] = {} - for key, value in values.items(): - if value is None: - out[key] = "" - elif isinstance(value, str): - out[key] = value - elif isinstance(value, bool): - out[key] = "true" if value else "false" - elif isinstance(value, (dict, list)): - out[key] = json.dumps(value) - else: - out[key] = str(value) - return out - - -def _memory_section(config: Dict[str, Any]) -> Dict[str, Any]: - """Return ``config["memory"]`` as a dict, creating/replacing a non-dict value.""" - memory_config = config.get("memory") - if not isinstance(memory_config, dict): - memory_config = {} - config["memory"] = memory_config - return memory_config - - -def _update_memory_provider_config(provider: ProviderConfigSchema, values: Dict[str, str]) -> None: - if provider.storage == STORAGE_HONCHO_HOST_BLOCK: - _write_provider_honcho(provider, values) - else: - _write_provider_flat(provider, values) - - config = load_config() - memory_config = _memory_section(config) - if memory_config.get("provider") != provider.name: - memory_config["provider"] = provider.name - save_config(config) - - -def _memory_provider_label(name: str) -> str: - return name.replace("_", " ").replace("-", " ").title() - - def _install_memory_provider_pip_dependencies(dependencies: List[str]) -> List[Dict[str, Any]]: - missing = [dep for dep in dependencies if not _dependency_importable(dep)] if not dependencies: return [] + missing = [dep for dep in dependencies if not _dependency_importable(dep)] if not missing: return [_command_result(kind="pip", name=", ".join(dependencies), status="already_installed")] - # Route through the lazy-install pipeline rather than pip against # sys.executable: on hosted/immutable images the agent venv is sealed # read-only and installs must go to HERMES_LAZY_INSTALL_TARGET, which @@ -402,60 +353,51 @@ def _install_memory_provider_pip_dependencies(dependencies: List[str]) -> List[D name = ", ".join(missing) try: from tools.lazy_deps import install_specs - outcome = install_specs(missing, timeout=240) except Exception as exc: return [_command_result(kind="pip", name=name, status="failed", error=str(exc))] - if outcome.blocked: return [_command_result(kind="pip", name=name, status="failed", command=outcome.command, error=outcome.reason)] - - return [ - _command_result( - kind="pip", - name=name, - status="installed" if outcome.ok else "failed", - command=outcome.command, - completed=subprocess.CompletedProcess( - args=outcome.command, returncode=0 if outcome.ok else 1, stdout=outcome.stdout, - stderr=outcome.stderr, - ), - ) - ] + return [_command_result( + kind="pip", name=name, status="installed" if outcome.ok else "failed", command=outcome.command, + completed=subprocess.CompletedProcess( + args=outcome.command, returncode=0 if outcome.ok else 1, stdout=outcome.stdout, stderr=outcome.stderr, + ), + )] -def _install_memory_provider_external_dependencies( - dependencies: List[Dict[str, str]], -) -> List[Dict[str, Any]]: +def _run_setup_step(results: list, kind: str, name: str, command: str, status_of, **kwargs) -> Optional[int]: + """Run a setup command, append its result row; returncode or None on spawn failure.""" + try: + completed = _run_setup_command(command if kwargs.get("shell") else shlex.split(command), display=command, **kwargs) + except Exception as exc: + results.append(_command_result(kind=kind, name=name, status=status_of(None), command=command, error=str(exc))) + return None + results.append(_command_result(kind=kind, name=name, status=status_of(completed.returncode == 0), command=command, completed=completed)) + return completed.returncode + + +def _install_memory_provider_external_dependencies(dependencies: List[Dict[str, str]]) -> List[Dict[str, Any]]: results: List[Dict[str, Any]] = [] for dep in dependencies: name = dep.get("name") or "dependency" check_cmd = dep.get("check") or "" install_cmd = dep.get("install") or "" - - def _run(kind: str, command: str, status_of, **kwargs) -> Optional[int]: - """Run a setup command, append its result row; returncode or None on spawn failure.""" - try: - completed = _run_setup_command(command if kwargs.get("shell") else shlex.split(command), display=command, **kwargs) - except Exception as exc: - results.append(_command_result(kind=kind, name=name, status=status_of(None), command=command, error=str(exc))) - return None - results.append(_command_result(kind=kind, name=name, status=status_of(completed.returncode == 0), command=command, completed=completed)) - return completed.returncode - # Check first: "already_installed" short-circuits; a failed check is # "missing" when an install step can fix it, "failed" otherwise. - if check_cmd and _run( - "external_check", check_cmd, + if check_cmd and _run_setup_step( + results, "external_check", name, check_cmd, lambda ok: "already_installed" if ok else ("missing" if install_cmd else "failed"), timeout=20, ) == 0: continue if not install_cmd: continue - rc = _run("external_install", install_cmd, lambda ok: "installed" if ok else "failed", shell=True, timeout=300) + rc = _run_setup_step( + results, "external_install", name, install_cmd, lambda ok: "installed" if ok else "failed", + shell=True, timeout=300, + ) if check_cmd and rc == 0: - _run("external_check", check_cmd, lambda ok: "verified" if ok else "failed", timeout=20) - + _run_setup_step(results, "external_check", name, check_cmd, lambda ok: "verified" if ok else "failed", timeout=20) return results @@ -463,82 +405,67 @@ def _install_memory_provider_setup(name: str) -> Dict[str, Any]: provider = _load_memory_provider(name) manifest = _memory_provider_manifest(name) if provider is None and not manifest: - raise HTTPException(status_code=404, detail=f"Unknown memory provider: {name}") - + raise _unknown_provider(name) setup = _memory_provider_setup_manifest(name) results = _install_memory_provider_pip_dependencies(setup["pip_dependencies"]) results.extend(_install_memory_provider_external_dependencies(setup["external_dependencies"])) - if not results: results.append(_command_result(kind="setup", name=name, status="no_declared_steps")) - - ok = all(result["status"] not in {"failed"} for result in results) + ok = all(result["status"] != "failed" for result in results) statuses = {row["name"]: row for row in _discover_memory_provider_statuses()} return {"ok": ok, "provider": name, "results": results, "status": statuses.get(name)} -def _public_memory_provider_field(field: Dict[str, Any], data: Dict[str, Any]) -> Dict[str, Any]: - return { - "key": field["key"], "label": field["label"], "kind": field["kind"], - "description": field["description"], "placeholder": field["placeholder"], - "required": field["required"], - "value": "" if field["kind"] == "secret" else _field_value(field, data), - "is_set": _field_is_set(field, data), "options": field.get("options", []), - "url": field.get("url", ""), "when": field.get("when"), "minimum": field.get("minimum"), - "maximum": field.get("maximum"), "step": field.get("step"), - } - +# ── Legacy provider surface (provider.config_schema()) ──────────────────────── def _memory_provider_payload(name: str, provider: Any) -> Dict[str, Any]: data = _read_memory_provider_existing_values(name) fields = [ - _public_memory_provider_field(field, data) + { + **{k: field[k] for k in ("key", "label", "kind", "description", "placeholder", "required")}, + "value": "" if field["kind"] == "secret" else _field_value(field, data), + "is_set": _field_is_set(field, data), "options": field.get("options", []), "url": field.get("url", ""), + **{k: field.get(k) for k in ("when", "minimum", "maximum", "step")}, + } for field in _normalize_memory_provider_schema(name, provider) ] return { - "name": name, "label": _memory_provider_label(name), "fields": fields, + "name": name, "label": name.replace("_", " ").replace("-", " ").title(), "fields": fields, "setup": _memory_provider_setup_info(name), } +def _coerce_schema_number(field: Dict[str, Any], raw: Any) -> "int | float": + value = raw if raw is not None and raw != "" else _field_default(field) + try: + if isinstance(value, bool) or not math.isfinite(result := float(value)): + raise ValueError + if field["kind"] == "integer": + if not result.is_integer(): + raise ValueError + result = int(result) + except (TypeError, ValueError, OverflowError) as exc: + raise ValueError(f"Invalid numeric value for '{field['key']}'") from exc + minimum, maximum = field.get("minimum"), field.get("maximum") + if minimum is not None and result < minimum: + raise ValueError(f"'{field['key']}' must be at least {minimum}") + if maximum is not None and result > maximum: + raise ValueError(f"'{field['key']}' must be at most {maximum}") + return result + + def _coerce_schema_field(field: Dict[str, Any], raw: Any) -> Any: - if field["kind"] == "boolean": + kind = field["kind"] + if kind == "boolean": return _coerce_bool(raw, default=_coerce_bool(_field_default(field), default=False)) - - if field["kind"] in {"integer", "number"}: - value = raw if raw is not None and raw != "" else _field_default(field) - try: - if isinstance(value, bool): - raise ValueError - parsed = float(value) - if not math.isfinite(parsed): - raise ValueError - if field["kind"] == "integer": - if not parsed.is_integer(): - raise ValueError - result: int | float = int(parsed) - else: - result = parsed - except (TypeError, ValueError, OverflowError) as exc: - raise ValueError(f"Invalid numeric value for '{field['key']}'") from exc - - minimum = field.get("minimum") - maximum = field.get("maximum") - if minimum is not None and result < minimum: - raise ValueError(f"'{field['key']}' must be at least {minimum}") - if maximum is not None and result > maximum: - raise ValueError(f"'{field['key']}' must be at most {maximum}") - return result - + if kind in {"integer", "number"}: + return _coerce_schema_number(field, raw) value = str(raw if raw is not None else "").strip() - if field["kind"] == "select": - if not value: - value = str(_field_default(field)) - allowed = {opt["value"] for opt in field.get("options", [])} - if value not in allowed: + if kind == "select": + value = value or str(_field_default(field)) + if value not in {opt["value"] for opt in field.get("options", [])}: raise ValueError(f"Invalid value for '{field['key']}'") return value - return value or _field_default(field) @@ -547,56 +474,40 @@ def _save_memory_provider_native_config(name: str, provider: Any, values: Dict[s try: from agent.memory_provider import MemoryProvider as _BaseMemoryProvider except Exception: + _BaseMemoryProvider = None + if _BaseMemoryProvider is None or type(provider).save_config is not _BaseMemoryProvider.save_config: provider.save_config(values, str(get_hermes_home())) return - if type(provider).save_config is not _BaseMemoryProvider.save_config: - provider.save_config(values, str(get_hermes_home())) - return - cfg = load_config() memory_cfg = _memory_section(cfg) current = memory_cfg.get(name) - if not isinstance(current, dict): - current = {} - current.update(values) - memory_cfg[name] = current + memory_cfg[name] = {**(current if isinstance(current, dict) else {}), **values} save_config(cfg) -def _write_memory_provider_config_values( - name: str, provider: Any, values: Dict[str, Any], -) -> None: +def _write_memory_provider_config_values(name: str, provider: Any, values: Dict[str, Any]) -> None: existing = _read_memory_provider_existing_values(name) fields = _normalize_memory_provider_schema(name, provider) fields_by_key = {field["key"]: field for field in fields} config_values: Dict[str, Any] = {} secrets: Dict[str, str] = {} - for field in fields: if not _field_visible(field, {**existing, **config_values}, fields_by_key): continue - + key = field["key"] if field["kind"] == "secret": - submitted = str(values.get(field["key"]) or "").strip() + submitted = str(values.get(key) or "").strip() if submitted and field.get("_env_key"): secrets[str(field["_env_key"])] = submitted continue - - raw = ( - values[field["key"]] - if field["key"] in values - else existing.get(field["key"], _field_default(field)) - ) - config_values[field["key"]] = _coerce_schema_field(field, raw) - + raw = values[key] if key in values else existing.get(key, _field_default(field)) + config_values[key] = _coerce_schema_field(field, raw) _save_memory_provider_native_config(name, provider, config_values) - for env_key, secret in secrets.items(): save_env_value(env_key, secret) -_MEMORY_PROVIDER_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$") - +# ── Routes ──────────────────────────────────────────────────────────────────── def _require_valid_memory_provider_name(name: str) -> None: """Reject provider names that could traverse outside the plugin dirs. @@ -606,7 +517,7 @@ def _require_valid_memory_provider_name(name: str) -> None: allowlist (no path separators, no dots) makes traversal impossible. """ if not _MEMORY_PROVIDER_NAME_RE.fullmatch(name or ""): - raise HTTPException(status_code=404, detail=f"Unknown memory provider: {name}") + raise _unknown_provider(name) @router.get("/api/memory/providers/{name}/config") @@ -621,7 +532,6 @@ async def get_memory_provider_config(name: str, surface: Optional[str] = None, p if declared is None: return {"name": name, "label": name, "docs_url": "", "fields": []} return _declared_provider_payload(declared) - provider = _load_memory_provider(name) if provider is None: return {"name": name, "label": name, "fields": [], "setup": _memory_provider_setup_info(name)} @@ -639,15 +549,10 @@ async def setup_memory_provider(name: str, body: MemoryProviderSetupRequest): # setup commands; refuse before the command-running path. (provider # may be None with a manifest present when its pip deps aren't # installed yet — that's the setup use case.) - raise HTTPException(status_code=404, detail=f"Unknown memory provider: {name}") + raise _unknown_provider(name) if provider is not None and body.values: - try: + with _value_errors_as_http("Failed to persist memory provider setup values for %s", name, passthrough_http=False): _write_memory_provider_config_values(name, provider, body.values) - except ValueError as exc: - raise HTTPException(status_code=400, detail=str(exc)) from exc - except Exception: - _log.exception("Failed to persist memory provider setup values for %s", name) - raise HTTPException(status_code=500, detail="Internal server error") _invalidate_plugins_hub_cache() return _install_memory_provider_setup(name) @@ -663,14 +568,13 @@ async def update_memory_provider_config( if surface == "declared": declared = get_provider_config_schema(name) if declared is None: - raise HTTPException(status_code=404, detail=f"Unknown memory provider: {name}") - _update_memory_provider_config(declared, _stringify_submitted_values(values)) + raise _unknown_provider(name) + _update_memory_provider_config(declared, {k: _stringify_submitted(v) for k, v in values.items()}) _invalidate_plugins_hub_cache() return {"ok": True} - provider = _load_memory_provider(name) if provider is None: - raise HTTPException(status_code=404, detail=f"Unknown memory provider: {name}") + raise _unknown_provider(name) _write_memory_provider_config_values(name, provider, values) _require_memory_provider_ready(name) config = load_config() @@ -679,12 +583,5 @@ async def update_memory_provider_config( _invalidate_plugins_hub_cache() return {"ok": True, "active": name} - try: + with _value_errors_as_http("PUT /api/memory/providers/%s/config failed", name): return await scoped_to_thread(profile, _run) - except HTTPException: - raise - except ValueError as exc: - raise HTTPException(status_code=400, detail=str(exc)) from exc - except Exception: - _log.exception("PUT /api/memory/providers/%s/config failed", name) - raise HTTPException(status_code=500, detail="Internal server error") diff --git a/hermes_cli/web_routers/oauth.py b/hermes_cli/web_routers/oauth.py index ac8905ca3c..3157b8f2b9 100644 --- a/hermes_cli/web_routers/oauth.py +++ b/hermes_cli/web_routers/oauth.py @@ -4,19 +4,20 @@ Extracted from ``hermes_cli.web_server``; helpers/state that tests monkeypatch o ``web_server`` stay there and are late-bound (cycle-safe). """ -import logging import asyncio -import sys -import time -from fastapi import APIRouter -from hermes_cli.web_deps import LateState, late -from hermes_cli.web_routers._common import scoped_to_thread -from fastapi import HTTPException, Request -from hermes_cli.web_models import OAuthSubmitBody -from typing import Any, Callable, Dict, Optional -import threading +import logging import os import secrets +import sys +import threading +import time +from typing import Any, Callable, Dict, Optional + +from fastapi import APIRouter, HTTPException, Request + +from hermes_cli.web_deps import LateState, late +from hermes_cli.web_models import OAuthSubmitBody +from hermes_cli.web_routers._common import scoped_to_thread _log = logging.getLogger("hermes_cli.web_server") router = APIRouter() @@ -35,6 +36,12 @@ _oauth_sessions = LateState("_oauth_sessions") _oauth_sessions_lock = LateState("_oauth_sessions_lock") _OAUTH_PROVIDER_CATALOG = LateState("_OAUTH_PROVIDER_CATALOG") +_CODEX_ISSUER = "https://auth.openai.com" +_JSON_HEADERS = {"Content-Type": "application/json"} +_CODEX_EXPIRES_IN = 15 * 60 # OpenAI's effective limit +_CANCELLED: Any = object() +_OAUTH_SESSION_TTL_SECONDS = 15 * 60 + def _http_response_error_detail(resp: Any) -> str: """Best-effort extraction of a short provider error detail.""" @@ -45,21 +52,13 @@ def _http_response_error_detail(resp: Any) -> str: if isinstance(payload, dict): error = payload.get("error") if isinstance(error, dict): - parts = [ - str(error.get(key, "")).strip() - for key in ("message", "error_description", "code", "type") - if str(error.get(key, "")).strip() - ] + parts = [s for s in (str(error.get(k, "")).strip() for k in ("message", "error_description", "code", "type")) if s] if parts: return ": ".join(parts) - if isinstance(error, str) and error.strip(): - return error.strip() - for key in ("detail", "message", "error_description"): - value = payload.get(key) + for value in (error, *(payload.get(k) for k in ("detail", "message", "error_description"))): if isinstance(value, str) and value.strip(): return value.strip() - text = str(getattr(resp, "text", "") or "").strip() - return text[:500] + return str(getattr(resp, "text", "") or "").strip()[:500] def _codex_device_code_start_error(resp: Any) -> str: @@ -79,21 +78,14 @@ def _codex_device_code_start_error(resp: Any) -> str: "OpenAI rejected the device-code login request. Please try Login " "again from the dashboard after checking your OpenAI account settings." ) - if detail: - return f"{message} (HTTP {status}: {detail})" - return f"{message} (HTTP {status})" + return f"{message} (HTTP {status}: {detail})" if detail else f"{message} (HTTP {status})" -def _new_oauth_session( - provider_id: str, flow: str, profile: Optional[str] = None, -) -> tuple[str, Dict[str, Any]]: +def _new_oauth_session(provider_id: str, flow: str, profile: Optional[str] = None) -> tuple[str, Dict[str, Any]]: """Create + register a new OAuth session, return (session_id, session_dict).""" sid = secrets.token_urlsafe(16) sess = { - "session_id": sid, - "provider": provider_id, - "flow": flow, - "profile": _oauth_profile_name(profile), + "session_id": sid, "provider": provider_id, "flow": flow, "profile": _oauth_profile_name(profile), "created_at": time.time(), "status": "pending", # pending | approved | denied | expired | error "error_message": None, @@ -103,120 +95,145 @@ def _new_oauth_session( return sid, sess -def _start_poller(target, sid: str) -> None: - threading.Thread(target=target, args=(sid,), daemon=True, name=f"oauth-poll-{sid[:6]}").start() +def _start_poller(target, sid: str, prefix: str = "oauth-poll") -> None: + threading.Thread(target=target, args=(sid,), daemon=True, name=f"{prefix}-{sid[:6]}").start() -def _device_code_started(sid: str, user_code, verification_url, expires_in: int, poll_interval: int) -> Dict[str, Any]: - """The /start response shape shared by every device-code flow.""" +def _device_session_started( + provider_id: str, profile: Optional[str], poller, fields: Dict[str, Any], + user_code, verification_url, expires_in: int, poll_interval: int, +) -> Dict[str, Any]: + """Register a device-code session carrying ``fields``, start its poller, return the /start body.""" + sid, sess = _new_oauth_session(provider_id, "device_code", profile=profile) + sess.update(fields) + _start_poller(poller, sid) return { "session_id": sid, "flow": "device_code", "user_code": user_code, "verification_url": verification_url, "expires_in": expires_in, "poll_interval": poll_interval, } -def _codex_full_login_worker(session_id: str) -> None: - """Run the complete OpenAI Codex device-code flow. +async def _httpx_call(fn: Callable[[Any], Any], timeout: float = 15.0, **client_kwargs) -> Any: + """Run ``fn(client)`` off-loop with a short-lived JSON-accepting ``httpx.Client``.""" + import httpx - Codex has its own ``/api/accounts/deviceauth/usercode`` (returns - ``device_auth_id``) and ``/api/accounts/deviceauth/token`` (polled until - 200) endpoints; success yields ``authorization_code`` + ``code_verifier`` - exchanged at CODEX_OAUTH_TOKEN_URL. Replicated inline rather than calling - ``_codex_device_code_login`` because that helper prints/blocks/polls in one - function — the dashboard needs the user_code before polling completes. - """ + def _call(): + with httpx.Client( + timeout=httpx.Timeout(timeout), headers={"Accept": "application/json"}, **client_kwargs + ) as client: + return fn(client) + + return await asyncio.get_running_loop().run_in_executor(None, _call) + + +# OpenAI Codex device-code worker. Codex's own deviceauth/usercode (returns +# device_auth_id) + deviceauth/token (polled until 200) endpoints yield +# authorization_code + code_verifier exchanged at CODEX_OAUTH_TOKEN_URL. Replicated +# here rather than calling ``_codex_device_code_login``, which prints/blocks/polls in +# one function — the dashboard needs the user_code before polling completes. + + +def _codex_cancelled(sess: Dict[str, Any], session_id: str, stage: str = "") -> bool: + if not sess.get("cancelled"): + return False + _log.info("oauth/device: openai-codex login cancelled%s (session=%s)", stage, session_id) + return True + + +def _codex_request_user_code(httpx) -> Dict[str, Any]: + """Step 1: request device code; returns device_data with ``interval`` clamped (>= 3s).""" + from hermes_cli.auth import CODEX_OAUTH_CLIENT_ID + + with httpx.Client(timeout=httpx.Timeout(15.0)) as client: + resp = client.post( + f"{_CODEX_ISSUER}/api/accounts/deviceauth/usercode", json={"client_id": CODEX_OAUTH_CLIENT_ID}, + headers=_JSON_HEADERS, + ) + if resp.status_code != 200: + raise RuntimeError(_codex_device_code_start_error(resp)) + device_data = resp.json() + device_data["interval"] = max(3, int(device_data.get("interval", "5"))) + if not device_data.get("user_code") or not device_data.get("device_auth_id"): + raise RuntimeError("device-code response missing user_code or device_auth_id") + return device_data + + +def _codex_poll_authorization(httpx, sess: Dict[str, Any], session_id: str) -> Any: + """Step 2: poll until authorized. ``None`` = expired; ``_CANCELLED`` = user cancelled.""" + deadline = time.monotonic() + sess["expires_in"] + payload = {"device_auth_id": sess["device_auth_id"], "user_code": sess["user_code"]} + with httpx.Client(timeout=httpx.Timeout(15.0)) as client: + while time.monotonic() < deadline: + if _codex_cancelled(sess, session_id): + return _CANCELLED + time.sleep(sess["interval"]) + if _codex_cancelled(sess, session_id): + return _CANCELLED + poll = client.post(f"{_CODEX_ISSUER}/api/accounts/deviceauth/token", json=payload, headers=_JSON_HEADERS) + if poll.status_code == 200: + return poll.json() + if poll.status_code in {403, 404}: + continue # user hasn't authorized yet + raise RuntimeError(f"deviceauth/token poll returned {poll.status_code}") + return None + + +def _codex_exchange_tokens(httpx, code_resp: Dict[str, Any]) -> Dict[str, str]: + """Step 3: exchange authorization_code for tokens.""" + from hermes_cli.auth import CODEX_OAUTH_CLIENT_ID, CODEX_OAUTH_TOKEN_URL + + authorization_code = code_resp.get("authorization_code", "") + code_verifier = code_resp.get("code_verifier", "") + if not authorization_code or not code_verifier: + raise RuntimeError("device-auth response missing authorization_code/code_verifier") + with httpx.Client(timeout=httpx.Timeout(15.0)) as client: + token_resp = client.post( + CODEX_OAUTH_TOKEN_URL, + data={ + "grant_type": "authorization_code", "code": authorization_code, + "redirect_uri": f"{_CODEX_ISSUER}/deviceauth/callback", + "client_id": CODEX_OAUTH_CLIENT_ID, "code_verifier": code_verifier, + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + if token_resp.status_code != 200: + raise RuntimeError(f"token exchange returned {token_resp.status_code}") + tokens = token_resp.json() + if not tokens.get("access_token"): + raise RuntimeError("token exchange did not return access_token") + return {"access_token": tokens.get("access_token", ""), "refresh_token": tokens.get("refresh_token", "")} + + +def _codex_full_login_worker(session_id: str) -> None: + """Run the complete OpenAI Codex device-code flow (see comment above).""" try: import httpx - from hermes_cli.auth import CODEX_OAUTH_CLIENT_ID, CODEX_OAUTH_TOKEN_URL - issuer = "https://auth.openai.com" - # Step 1: request device code - with httpx.Client(timeout=httpx.Timeout(15.0)) as client: - resp = client.post( - f"{issuer}/api/accounts/deviceauth/usercode", - json={"client_id": CODEX_OAUTH_CLIENT_ID}, - headers={"Content-Type": "application/json"}, - ) - if resp.status_code != 200: - raise RuntimeError(_codex_device_code_start_error(resp)) - device_data = resp.json() - user_code = device_data.get("user_code", "") - device_auth_id = device_data.get("device_auth_id", "") - poll_interval = max(3, int(device_data.get("interval", "5"))) - if not user_code or not device_auth_id: - raise RuntimeError("device-code response missing user_code or device_auth_id") + device_data = _codex_request_user_code(httpx) with _oauth_sessions_lock: sess = _oauth_sessions.get(session_id) if not sess: return - sess["user_code"] = user_code - sess["verification_url"] = f"{issuer}/codex/device" - sess["device_auth_id"] = device_auth_id - sess["interval"] = poll_interval - sess["expires_in"] = 15 * 60 # OpenAI's effective limit - sess["expires_at"] = time.time() + sess["expires_in"] + sess.update( + user_code=device_data["user_code"], verification_url=f"{_CODEX_ISSUER}/codex/device", + device_auth_id=device_data["device_auth_id"], interval=device_data["interval"], + expires_in=_CODEX_EXPIRES_IN, expires_at=time.time() + _CODEX_EXPIRES_IN, + ) # Captured now (not re-derived after cancel pops the session) so a # cancelled session can never fall back to the caller's current # profile scope at save time. session_profile = sess.get("profile") - # Step 2: poll until authorized - deadline = time.monotonic() + sess["expires_in"] - code_resp = None - with httpx.Client(timeout=httpx.Timeout(15.0)) as client: - while time.monotonic() < deadline: - if sess.get("cancelled"): - _log.info("oauth/device: openai-codex login cancelled (session=%s)", session_id) - return - time.sleep(poll_interval) - if sess.get("cancelled"): - _log.info("oauth/device: openai-codex login cancelled (session=%s)", session_id) - return - poll = client.post( - f"{issuer}/api/accounts/deviceauth/token", - json={"device_auth_id": device_auth_id, "user_code": user_code}, - headers={"Content-Type": "application/json"}, - ) - if poll.status_code == 200: - code_resp = poll.json() - break - if poll.status_code in {403, 404}: - continue # user hasn't authorized yet - raise RuntimeError(f"deviceauth/token poll returned {poll.status_code}") - + code_resp = _codex_poll_authorization(httpx, sess, session_id) if code_resp is None: with _oauth_sessions_lock: sess["status"] = "expired" sess["error_message"] = "Device code expired before approval" return - - if sess.get("cancelled"): - _log.info("oauth/device: openai-codex login cancelled before token exchange (session=%s)", session_id) + if code_resp is _CANCELLED or _codex_cancelled(sess, session_id, " before token exchange"): return - # Step 3: exchange authorization_code for tokens - authorization_code = code_resp.get("authorization_code", "") - code_verifier = code_resp.get("code_verifier", "") - if not authorization_code or not code_verifier: - raise RuntimeError("device-auth response missing authorization_code/code_verifier") - with httpx.Client(timeout=httpx.Timeout(15.0)) as client: - token_resp = client.post( - CODEX_OAUTH_TOKEN_URL, - data={ - "grant_type": "authorization_code", "code": authorization_code, - "redirect_uri": f"{issuer}/deviceauth/callback", - "client_id": CODEX_OAUTH_CLIENT_ID, "code_verifier": code_verifier, - }, - headers={"Content-Type": "application/x-www-form-urlencoded"}, - ) - if token_resp.status_code != 200: - raise RuntimeError(f"token exchange returned {token_resp.status_code}") - tokens = token_resp.json() - access_token = tokens.get("access_token", "") - refresh_token = tokens.get("refresh_token", "") - if not access_token: - raise RuntimeError("token exchange did not return access_token") - + tokens = _codex_exchange_tokens(httpx, code_resp) from hermes_cli.auth import _save_codex_tokens # The cancellation check and the save are one atomic critical section @@ -224,11 +241,10 @@ def _codex_full_login_worker(session_id: str) -> None: # flip "cancelled" between the check and the save and tokens would be # persisted after the user believed the login was aborted. with _oauth_sessions_lock: - if sess.get("cancelled"): - _log.info("oauth/device: openai-codex login cancelled before token save (session=%s)", session_id) + if _codex_cancelled(sess, session_id, " before token save"): return with _profile_scope(session_profile): - _save_codex_tokens({"access_token": access_token, "refresh_token": refresh_token}) + _save_codex_tokens(tokens) sess["status"] = "approved" _log.info("oauth/device: openai-codex login completed (session=%s)", session_id) except Exception as e: @@ -285,12 +301,9 @@ _PROVIDER_STATUS: Dict[str, tuple[str, Callable[[dict], dict]]] = { def _resolve_provider_status(provider_id: str, status_fn) -> Dict[str, Any]: """Dispatch to the right status helper for an OAuth provider entry.""" - if status_fn is not None: - try: - return status_fn() - except Exception as e: - return {"logged_in": False, "error": str(e)} try: + if status_fn is not None: + return status_fn() from hermes_cli import auth as hauth entry = _PROVIDER_STATUS.get(provider_id) if entry is not None: @@ -316,46 +329,32 @@ def _resolve_provider_status(provider_id: str, status_fn) -> Dict[str, Any]: async def _start_nous_device_code(profile: Optional[str]) -> Dict[str, Any]: - from hermes_cli.auth import _request_device_code, PROVIDER_REGISTRY - import httpx + from hermes_cli.auth import PROVIDER_REGISTRY, _request_device_code pconfig = PROVIDER_REGISTRY["nous"] portal_base_url = ( - os.getenv("HERMES_PORTAL_BASE_URL") - or os.getenv("NOUS_PORTAL_BASE_URL") - or pconfig.portal_base_url + os.getenv("HERMES_PORTAL_BASE_URL") or os.getenv("NOUS_PORTAL_BASE_URL") or pconfig.portal_base_url ).rstrip("/") - client_id = pconfig.client_id - scope = pconfig.scope - - def _do_nous_device_request(): - with httpx.Client(timeout=httpx.Timeout(15.0), headers={"Accept": "application/json"}) as client: - return ( - _request_device_code( - client=client, portal_base_url=portal_base_url, client_id=client_id, scope=scope - ), - scope, - ) - - device_data, effective_scope = await asyncio.get_running_loop().run_in_executor(None, _do_nous_device_request) - sid, sess = _new_oauth_session("nous", "device_code", profile=profile) - sess.update( - device_code=str(device_data["device_code"]), interval=int(device_data["interval"]), - expires_at=time.time() + int(device_data["expires_in"]), portal_base_url=portal_base_url, - client_id=client_id, scope=effective_scope, - ) - _start_poller(_nous_poller, sid) - return _device_code_started( - sid, str(device_data["user_code"]), str(device_data["verification_uri_complete"]), + device_data = await _httpx_call(lambda client: _request_device_code( + client=client, portal_base_url=portal_base_url, client_id=pconfig.client_id, scope=pconfig.scope, + )) + return _device_session_started( + "nous", profile, _nous_poller, + dict( + device_code=str(device_data["device_code"]), interval=int(device_data["interval"]), + expires_at=time.time() + int(device_data["expires_in"]), portal_base_url=portal_base_url, + client_id=pconfig.client_id, scope=pconfig.scope, + ), + str(device_data["user_code"]), str(device_data["verification_uri_complete"]), int(device_data["expires_in"]), int(device_data["interval"]), ) async def _start_codex_device_code(profile: Optional[str]) -> Dict[str, Any]: # The full Codex helper polls inline, so it runs in a worker thread and - # proxies user_code + verification_url back via the session dict. + # proxies user_code + verification_url back via the session dict; block + # briefly until the worker has populated the user_code, OR errored. sid, _ = _new_oauth_session("openai-codex", "device_code", profile=profile) - threading.Thread(target=_codex_full_login_worker, args=(sid,), daemon=True, name=f"oauth-codex-{sid[:6]}").start() - # Block briefly until the worker has populated the user_code, OR error. + _start_poller(_codex_full_login_worker, sid, prefix="oauth-codex") deadline = time.monotonic() + 10 while time.monotonic() < deadline: with _oauth_sessions_lock: @@ -369,75 +368,59 @@ async def _start_codex_device_code(profile: Optional[str]) -> Dict[str, Any]: raise HTTPException(status_code=500, detail=s.get("error_message") or "device-auth failed") if not s.get("user_code"): raise HTTPException(status_code=504, detail="device-auth timed out before returning a user code") - return _device_code_started( - sid, s["user_code"], s["verification_url"], int(s.get("expires_in") or 900), int(s.get("interval") or 5) - ) + return { + "session_id": sid, "flow": "device_code", "user_code": s["user_code"], "verification_url": s["verification_url"], + "expires_in": int(s.get("expires_in") or 900), "poll_interval": int(s.get("interval") or 5), + } async def _start_minimax_device_code(profile: Optional[str]) -> Dict[str, Any]: # Device-code flow with a PKCE extension: verifier + challenge from # _minimax_pkce_pair bind the token exchange to the original session. from hermes_cli.auth import ( - _minimax_pkce_pair, _minimax_request_user_code, MINIMAX_OAUTH_CLIENT_ID, MINIMAX_OAUTH_GLOBAL_BASE, + MINIMAX_OAUTH_CLIENT_ID, MINIMAX_OAUTH_GLOBAL_BASE, _minimax_pkce_pair, _minimax_request_user_code, ) - import httpx verifier, challenge, state = _minimax_pkce_pair() portal_base_url = (os.getenv("MINIMAX_PORTAL_BASE_URL") or MINIMAX_OAUTH_GLOBAL_BASE).rstrip("/") - - def _do_minimax_request(): - with httpx.Client( - timeout=httpx.Timeout(15.0), headers={"Accept": "application/json"}, follow_redirects=True - ) as client: - return _minimax_request_user_code( - client=client, portal_base_url=portal_base_url, client_id=MINIMAX_OAUTH_CLIENT_ID, - code_challenge=challenge, state=state, - ) - - device_data = await asyncio.get_event_loop().run_in_executor(None, _do_minimax_request) - sid, sess = _new_oauth_session("minimax-oauth", "device_code", profile=profile) + device_data = await _httpx_call(lambda client: _minimax_request_user_code( + client=client, portal_base_url=portal_base_url, client_id=MINIMAX_OAUTH_CLIENT_ID, + code_challenge=challenge, state=state, + ), follow_redirects=True) # MiniMax's `interval` is in milliseconds (defensive default 2000ms in _minimax_poll_token). interval_raw = device_data.get("interval") - sess.update( - interval_ms=int(interval_raw) if interval_raw is not None else None, - user_code=str(device_data["user_code"]), code_verifier=verifier, state=state, - portal_base_url=portal_base_url, client_id=MINIMAX_OAUTH_CLIENT_ID, region="global", - ) + interval_ms = int(interval_raw) if interval_raw is not None else None # `expired_in` is overloaded — a unix-ms timestamp OR seconds-from-now. # Mirror the heuristic in _minimax_poll_token; keep the raw value for the # poller and derive expires_at + UI-friendly expires_in seconds. expired_in_raw = int(device_data["expired_in"]) - sess["expired_in_raw"] = expired_in_raw if expired_in_raw > 1_000_000_000_000: # likely unix-ms expires_at_ts = expired_in_raw / 1000.0 expires_in_seconds = max(0, int(expires_at_ts - time.time())) else: expires_at_ts = time.time() + expired_in_raw expires_in_seconds = expired_in_raw - sess["expires_at"] = expires_at_ts - _start_poller(_minimax_poller, sid) - return _device_code_started( - sid, str(device_data["user_code"]), str(device_data["verification_uri"]), expires_in_seconds, - max(2, (sess["interval_ms"] or 2000) // 1000), + return _device_session_started( + "minimax-oauth", profile, _minimax_poller, + dict( + interval_ms=interval_ms, user_code=str(device_data["user_code"]), code_verifier=verifier, state=state, + portal_base_url=portal_base_url, client_id=MINIMAX_OAUTH_CLIENT_ID, region="global", + expired_in_raw=expired_in_raw, expires_at=expires_at_ts, + ), + str(device_data["user_code"]), str(device_data["verification_uri"]), expires_in_seconds, + max(2, (interval_ms or 2000) // 1000), ) async def _start_xai_device_code(profile: Optional[str]) -> Dict[str, Any]: from hermes_cli.auth import _xai_oauth_request_device_code - import httpx - - def _do_xai_device_request(): - with httpx.Client(timeout=httpx.Timeout(20.0), headers={"Accept": "application/json"}) as client: - return _xai_oauth_request_device_code(client) - - device_data = await asyncio.get_running_loop().run_in_executor(None, _do_xai_device_request) - sid, sess = _new_oauth_session("xai-oauth", "device_code", profile=profile) - sess.update( - device_code=str(device_data["device_code"]), interval=int(device_data["interval"]), - expires_at=time.time() + int(device_data["expires_in"]), - ) - _start_poller(_xai_device_poller, sid) - return _device_code_started( - sid, str(device_data["user_code"]), + device_data = await _httpx_call(_xai_oauth_request_device_code, timeout=20.0) + return _device_session_started( + "xai-oauth", profile, _xai_device_poller, + dict( + device_code=str(device_data["device_code"]), interval=int(device_data["interval"]), + expires_at=time.time() + int(device_data["expires_in"]), + ), + str(device_data["user_code"]), str(device_data.get("verification_uri_complete") or device_data["verification_uri"]), int(device_data["expires_in"]), int(device_data["interval"]), ) @@ -449,15 +432,8 @@ _DEVICE_CODE_STARTERS = { } -async def _start_device_code_flow( - provider_id: str, profile: Optional[str] = None, -) -> Dict[str, Any]: - """Initiate a device-code flow (Nous, OpenAI Codex, MiniMax, or xAI). - - Calls the provider's device-auth endpoint via the CLI helpers, then spawns - a background poller. Returns the user-facing display fields (verification - link + user code). - """ +async def _start_device_code_flow(provider_id: str, profile: Optional[str] = None) -> Dict[str, Any]: + """Hit the provider's device-auth endpoint, spawn its poller, return the display fields.""" starter = _DEVICE_CODE_STARTERS.get(provider_id) if starter is None: raise HTTPException(status_code=400, detail=f"Provider {provider_id} does not support device-code flow") @@ -465,25 +441,19 @@ async def _start_device_code_flow( def _oauth_provider_disconnect_command(provider: Dict[str, Any]) -> Optional[str]: - """Shell command that clears an external provider's credentials. + """Shell command that clears an external provider's credentials, or None. - External providers store credentials outside Hermes, so the disconnect API - refuses them (never silently delete files another CLI owns); instead the - GUI gets a command to run in the embedded terminal, so the user sees - exactly what executes. Claude Code has no scriptable logout, so we remove - the credential the way logout does: the macOS Keychain entry - (``Claude Code-credentials``) and/or ``~/.claude/.credentials.json`` — the - two sources ``read_claude_code_credentials()`` consults. None for providers - we can't safely clear (the GUI shows a manual hint). + The disconnect API never silently deletes files another CLI owns; the GUI runs + this in its embedded terminal so the user sees exactly what executes. Claude Code + has no scriptable logout, so remove what logout would: the macOS Keychain entry + and/or ``~/.claude/.credentials.json`` (the two ``read_claude_code_credentials()`` sources). """ - if provider.get("flow") != "external": + if provider.get("flow") != "external" or provider.get("id") != "claude-code": return None - if provider.get("id") == "claude-code": - rm_file = "rm -f ~/.claude/.credentials.json" - if sys.platform == "darwin": - return f'security delete-generic-password -s "Claude Code-credentials" 2>/dev/null; {rm_file}' - return rm_file - return None + rm_file = "rm -f ~/.claude/.credentials.json" + if sys.platform == "darwin": + return f'security delete-generic-password -s "Claude Code-credentials" 2>/dev/null; {rm_file}' + return rm_file def _oauth_provider_disconnect_hint(provider: Dict[str, Any], status: Dict[str, Any]) -> Optional[str]: @@ -503,24 +473,15 @@ def _oauth_provider_disconnect_hint(provider: Dict[str, Any], status: Dict[str, def _build_oauth_catalog() -> list[Dict[str, Any]]: - """Build the Accounts-tab provider list. - - Membership is the union of ``_OAUTH_PROVIDER_CATALOG`` (hand-tuned cards - with bespoke flow / status_fn / cli_command, incl. the Anthropic PKCE card - and the synthetic claude-code row) and every accounts-tab provider in the - unified ``provider_catalog()``, so a plugin-added OAuth/external provider - appears automatically. Explicit cards win on metadata and come first in - curated order; catalog-only providers follow in ``hermes model`` order. - """ + """Accounts-tab provider list: ``_OAUTH_PROVIDER_CATALOG`` cards first (curated + order, win on metadata), then every other accounts-tab ``provider_catalog()`` entry + in ``hermes model`` order, so plugin-added OAuth/external providers appear automatically.""" rows: list[Dict[str, Any]] = [] seen: set[str] = set() - for entry in _OAUTH_PROVIDER_CATALOG: - if entry["id"] in seen: - continue - seen.add(entry["id"]) - rows.append(dict(entry)) - + if entry["id"] not in seen: + seen.add(entry["id"]) + rows.append(dict(entry)) try: from hermes_cli.provider_catalog import provider_catalog for d in provider_catalog(): @@ -529,25 +490,17 @@ def _build_oauth_catalog() -> list[Dict[str, Any]]: seen.add(d.slug) rows.append({ "id": d.slug, "name": d.label, "flow": "external", - "cli_command": f"hermes auth add {d.slug}", "docs_url": d.signup_url or "", - "status_fn": None, + "cli_command": f"hermes auth add {d.slug}", "docs_url": d.signup_url or "", "status_fn": None, }) except Exception: pass - return rows @router.get("/api/providers/oauth") async def list_oauth_providers(profile: Optional[str] = None): - """Enumerate every OAuth-capable LLM provider with current status. - - Per provider: id (used in the DELETE path), name, flow - ("device_code" | "external"), cli_command (manual fallback), - disconnect_command (shell command for external providers, else null), - docs_url, and status {logged_in, source slug, source_label, token_preview - (last N chars, never the full token), expires_at, has_refresh_token}. - """ + """Every OAuth-capable provider with current status (token_preview is the last + N chars, never the full token; disconnect_command only for external providers).""" def _run(): providers = [] for p in _build_oauth_catalog(): @@ -568,16 +521,30 @@ async def list_oauth_providers(profile: Optional[str] = None): def _reject_if_not_disconnectable(provider: Dict[str, Any], status: Dict[str, Any]) -> None: disconnect_hint = _oauth_provider_disconnect_hint(provider, status) if disconnect_hint: - raise HTTPException( - status_code=400, - detail=f"{provider['name']} cannot be disconnected automatically. {disconnect_hint}", - ) + raise HTTPException(400, f"{provider['name']} cannot be disconnected automatically. {disconnect_hint}") + + +def _clear_anthropic_auth() -> bool: + """Clear only the Hermes-managed PKCE file and auth-store entry (never ~/.claude/*).""" + cleared = False + try: + from agent.anthropic_adapter import _get_hermes_oauth_file + oauth_file = _get_hermes_oauth_file() + if oauth_file.exists(): + oauth_file.unlink() + cleared = True + except Exception: + pass + try: + from hermes_cli.auth import clear_provider_auth + cleared = clear_provider_auth("anthropic") or cleared + except Exception: + pass + return cleared @router.delete("/api/providers/oauth/{provider_id}") -async def disconnect_oauth_provider( - provider_id: str, request: Request, profile: Optional[str] = None, -): +async def disconnect_oauth_provider(provider_id: str, request: Request, profile: Optional[str] = None): """Disconnect an OAuth provider. Token-protected (matches /env/reveal).""" _require_token(request) @@ -585,38 +552,15 @@ async def disconnect_oauth_provider( catalog_by_id = {p["id"]: p for p in _build_oauth_catalog()} provider = catalog_by_id.get(provider_id) if provider is None: - raise HTTPException( - status_code=400, - detail=f"Unknown provider: {provider_id}. " - f"Available: {', '.join(sorted(catalog_by_id))}", - ) - + raise HTTPException(400, f"Unknown provider: {provider_id}. Available: {', '.join(sorted(catalog_by_id))}") + # Flow-only rejection first so external providers never reach status resolution. _reject_if_not_disconnectable(provider, {}) - _reject_if_not_disconnectable( - provider, _resolve_provider_status(provider_id, provider.get("status_fn")) - ) + _reject_if_not_disconnectable(provider, _resolve_provider_status(provider_id, provider.get("status_fn"))) - # Anthropic clears only the Hermes-managed PKCE file and auth-store - # entry; the external claude-code row was rejected above so we never - # pretend to remove ~/.claude/* credentials owned by the CLI. if provider_id == "anthropic": - cleared = False - try: - from agent.anthropic_adapter import _get_hermes_oauth_file - oauth_file = _get_hermes_oauth_file() - if oauth_file.exists(): - oauth_file.unlink() - cleared = True - except Exception: - pass - try: - from hermes_cli.auth import clear_provider_auth - cleared = clear_provider_auth("anthropic") or cleared - except Exception: - pass + cleared = _clear_anthropic_auth() _log.info("oauth/disconnect: %s", provider_id) return {"ok": bool(cleared), "provider": provider_id} - try: from hermes_cli.auth import clear_provider_auth, invalidate_nous_auth_status_cache cleared = clear_provider_auth(provider_id) @@ -631,42 +575,31 @@ async def disconnect_oauth_provider( return await scoped_to_thread(profile, _run) -# In-browser device-code flows (Nous, OpenAI Codex, MiniMax, xAI): -# 1. POST /api/providers/oauth/{provider}/start hits the provider's -# device-auth endpoint, spawns a poller thread that polls the token -# endpoint every `interval` seconds, and returns -# {session_id, flow, user_code, verification_url, expires_in, poll_interval}. -# 2. UI opens verification_url and shows user_code, then polls -# GET /api/providers/oauth/{provider}/poll/{session_id} until status != "pending". -# 3. On "approved" the poller has already saved creds; UI refreshes the list. -# Anthropic has NO dashboard PKCE flow: an unattended endpoint minting Claude -# subscription tokens outside Anthropic's own client violates its OAuth usage -# policy; that card is flow == "external" (`hermes auth add anthropic`). -# Sessions are in-memory (single-process) and expire after 15 minutes; /start -# GCs expired ones so the dict is bounded. - -_OAUTH_SESSION_TTL_SECONDS = 15 * 60 +# In-browser device-code sessions: /start spawns a poller thread and returns the +# display fields; the UI polls .../poll/{session_id} until status != "pending" (on +# "approved" the poller has already saved creds). Anthropic has NO dashboard PKCE +# flow — an unattended endpoint minting Claude subscription tokens outside +# Anthropic's own client violates its OAuth usage policy; that card is "external". +# Sessions are in-memory (single-process), expire after 15 min, GC'd on /start. def _gc_oauth_sessions() -> None: - """Drop expired sessions. Called opportunistically on /start.""" cutoff = time.time() - _OAUTH_SESSION_TTL_SECONDS with _oauth_sessions_lock: - stale = [sid for sid, sess in _oauth_sessions.items() if sess["created_at"] < cutoff] - for sid in stale: + for sid in [sid for sid, sess in _oauth_sessions.items() if sess["created_at"] < cutoff]: _oauth_sessions.pop(sid, None) -def _validate_oauth_profile(profile: Optional[str]) -> None: +def _validate_oauth_profile(profile: Optional[str]) -> str: + """Validate the requested profile (404 via ``_resolve_profile_dir``) and return its name.""" profile_name = _oauth_profile_name(profile) if profile_name: _resolve_profile_dir(profile_name) + return profile_name @router.post("/api/providers/oauth/{provider_id}/start") -async def start_oauth_login( - provider_id: str, request: Request, profile: Optional[str] = None, -): +async def start_oauth_login(provider_id: str, request: Request, profile: Optional[str] = None): """Initiate an OAuth login flow. Token-protected.""" _require_token(request) _gc_oauth_sessions() @@ -675,10 +608,7 @@ async def start_oauth_login( if catalog_entry is None: raise HTTPException(status_code=400, detail=f"Unknown provider {provider_id}") if catalog_entry["flow"] == "external": - raise HTTPException( - status_code=400, - detail=f"{provider_id} uses an external CLI; run `{catalog_entry['cli_command']}` manually", - ) + raise HTTPException(400, f"{provider_id} uses an external CLI; run `{catalog_entry['cli_command']}` manually") try: if catalog_entry["flow"] == "device_code": return await _start_device_code_flow(provider_id, profile=profile) @@ -700,13 +630,10 @@ async def submit_oauth_code( @router.get("/api/providers/oauth/{provider_id}/poll/{session_id}") -async def poll_oauth_session( - provider_id: str, session_id: str, profile: Optional[str] = None, -): +async def poll_oauth_session(provider_id: str, session_id: str, profile: Optional[str] = None): """Poll a session's status (no auth — read-only state). One endpoint serves every device-code flow: all report progress via the worker-updated ``status``.""" - _validate_oauth_profile(profile) - requested_profile = _oauth_profile_name(profile) + requested_profile = _validate_oauth_profile(profile) with _oauth_sessions_lock: sess = _oauth_sessions.get(session_id) if not sess: @@ -722,9 +649,7 @@ async def poll_oauth_session( @router.delete("/api/providers/oauth/sessions/{session_id}") -async def cancel_oauth_session( - session_id: str, request: Request, profile: Optional[str] = None, -): +async def cancel_oauth_session(session_id: str, request: Request, profile: Optional[str] = None): """Cancel a pending OAuth session. Token-protected. Marks the session dict ``cancelled`` before popping it so a background @@ -733,8 +658,7 @@ async def cancel_oauth_session( believed it was aborted. """ _require_token(request) - _validate_oauth_profile(profile) - requested_profile = _oauth_profile_name(profile) + requested_profile = _validate_oauth_profile(profile) with _oauth_sessions_lock: sess = _oauth_sessions.get(session_id) if sess is not None: