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

Merge branch 'simp/r3-25-K' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-03 00:09:17 -07:00
parent eeaf3e9a11 6622c31262
commit ecc23d5c32
3 changed files with 612 additions and 877 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/web_routers/actions.py
+126 -212
View File
@@ -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.<name>`` 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()}
hermes_cli/web_routers/memory_providers.py
+243 -346
View File
@@ -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")
hermes_cli/web_routers/oauth.py
+243 -319
View File
@@ -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: