merge: r3-18 group D into simp/r3-18

This commit is contained in:
Teknium
2026-09-02 23:15:47 -07:00
6 changed files with 1127 additions and 1945 deletions
+137 -295
View File
@@ -1,8 +1,8 @@
"""Container-boot reconciliation of per-profile gateway s6 services.
Wired into the image as /etc/cont-init.d/02-reconcile-profiles by the Dockerfile (Phase 4 Task 4.0).
Runs as root after 01-hermes-setup (the stage2 hook) has chowned the volume and seeded $HERMES_HOME,
but before s6-rc starts user services.
Wired into the image as /etc/cont-init.d/02-reconcile-profiles. Runs as root after
01-hermes-setup (the stage2 hook) has chowned the volume and seeded $HERMES_HOME, but
before s6-rc starts user services.
"""
from __future__ import annotations
@@ -15,50 +15,25 @@ from typing import Literal, Sequence
log = logging.getLogger(__name__)
# Only this desired state triggers automatic restart. Everything else
# (startup_failed, starting, stopped, missing) registers the slot in
# the down state and waits for explicit user action — this avoids the
# crash-loop where a broken gateway keeps being restarted across
# `docker restart` cycles. Older installs only have gateway_state;
# newer lifecycle commands persist desired_state separately so a transient
# runtime state (draining/startup_failed) does not erase the operator's
# durable start/stop intent across pod/container recreation.
# Only this desired state auto-restarts; everything else (startup_failed, starting, stopped,
# missing) registers the slot down and waits for the user — avoiding the crash-loop where a
# broken gateway keeps restarting across `docker restart`. Older installs only have
# gateway_state; newer lifecycle commands persist desired_state separately so a transient
# runtime state does not erase the operator's durable intent.
_AUTOSTART_STATES = frozenset({"running"})
# Transient runtime sub-states of a RUNNING gateway. A gateway only ever
# reaches these while it is up and serving, so they are NOT an operator stop
# and NOT a failed boot:
# - `draining` — written by the drain watcher / scale-to-zero go-dormant
# path when an in-flight quiesce begins (gateway/run.py).
# - `degraded` — written when the gateway comes up with some platforms
# queued for retry, then "falls through to the normal
# running state" (gateway/run.py #5196): the process is up,
# serving cron + whatever platforms connected, and the
# reconnect watcher takes the rest from there.
#
# When a gateway is hard-killed *while in one of these states* (a container/VM
# recreate SIGTERMs it before `_stop_impl` reaches its terminal-state persist),
# the last value left in gateway_state.json is the transient sub-state. With no
# explicit `desired_state` to fall back to, treating that literal value as the
# autostart intent would leave the gateway DOWN on every subsequent boot — the
# gateway never comes back, the dashboard is up but messaging stays dark
# (observed on a relay-opted-in staging instance stranded at `draining`,
# 2026-06; `degraded` is the same wedge class). Map these transient sub-states
# to `running` so a stranded marker reads as the run-intent it actually
# represents. This mirrors gateway/run.py's #42675 handling, which persists
# `running` (not the mid-shutdown `draining`) when an unexpected signal tears
# the gateway down — extended here to the case where the gateway died before it
# could persist anything at all.
#
# `starting` / `startup_failed` are deliberately NOT included: those mean the
# gateway died mid-boot or failed to come up, so auto-restarting them would
# reintroduce the crash-loop the down-marker guard exists to prevent.
# Transient sub-states of a RUNNING gateway (`draining`: in-flight quiesce; `degraded`: up
# with some platforms queued for reconnect) — NOT an operator stop, NOT a failed boot. When a
# gateway is hard-killed in one of them (container recreate SIGTERMs it before `_stop_impl`
# persists a terminal state) and there is no `desired_state`, reading the marker literally
# would leave the gateway DOWN on every later boot (observed: staging instance stranded at
# `draining`). Map them to `running`, mirroring gateway/run.py's unexpected-signal handling.
# `starting` / `startup_failed` are deliberately excluded: auto-restarting a gateway that died
# mid-boot reintroduces the crash-loop.
_TRANSIENT_RUNNING_STATES = frozenset({"draining", "degraded"})
# Stale runtime files we sweep before recreating service slots. These
# all hold container-namespaced state (PIDs, process tables) that's
# garbage post-restart — a numerically-equal PID in the new container
# is a different process. See the Risk Register in the plan.
# Swept before recreating slots: container-namespaced state (PIDs, process tables) is garbage
# post-restart — a numerically-equal PID in the new container is a different process.
_STALE_RUNTIME_FILES = ("gateway.pid", "processes.json")
ReconcileActionLabel = Literal["started", "registered", "skipped"]
@@ -70,116 +45,78 @@ class ReconcileAction:
profile: str
prior_state: str | None
action: ReconcileActionLabel
# How the profile's previous gateway life ended: "clean" (exit path ran),
# "unclean" (sentinel still says running — SIGKILL/OOM/VM death), or
# "unknown" (no sentinel / never ran). See gateway.lifecycle_ledger
# (NS-608): at container boot this is the one place that can stamp
# "the previous container life ended violently" into a durable,
# volume-persisted log line.
# "clean" (exit path ran) / "unclean" (sentinel still says running — SIGKILL/OOM/VM death)
# / "unknown". Boot is the one place that can stamp a violent previous death into a
# durable, volume-persisted log line (gateway.lifecycle_ledger).
prior_exit: str = "unknown"
def _slot_action(
profile: str, profile_dir: Path, prior_state: str | None, start: bool,
) -> ReconcileAction:
return ReconcileAction(profile=profile, prior_state=prior_state,
action="started" if start else "registered",
prior_exit=_read_prior_exit_label(profile_dir))
def reconcile_profile_gateways(
*,
hermes_home: Path,
scandir: Path,
dry_run: bool = False,
*, hermes_home: Path, scandir: Path, dry_run: bool = False,
container_argv: Sequence[str] | None = None,
) -> list[ReconcileAction]:
"""Recreate s6 service registrations for every persistent profile.
Always registers a ``gateway-default`` slot for the root profile (the implicit profile that
lives at the top of ``$HERMES_HOME``, not under ``profiles/``). The dispatcher in
``hermes_cli.gateway`` maps an empty profile suffix to ``gateway-default``, so this slot is what
``hermes gateway start`` (no ``-p``) targets.
Always registers a ``gateway-default`` slot for the root profile (the implicit profile at
the top of ``$HERMES_HOME``): ``hermes_cli.gateway`` maps an empty profile suffix to it,
so it is what ``hermes gateway start`` (no ``-p``) targets.
"""
actions: list[ReconcileAction] = []
# A multiplexing root/default gateway owns inbound platform connections
# for every profile. Named slots must still be registered (so explicit
# lifecycle management remains available), but booting them from their
# persisted run intent would create additional multiplex owners.
# Keep the boot reconciler aligned with the gateway that will own these
# slots. The runtime resolver gives a recognized environment override
# precedence over config.yaml and otherwise preserves the configured value.
# A multiplexing root gateway owns inbound platform connections for every profile: named
# slots are still registered (lifecycle management stays available) but must not boot
# from their persisted run intent, or they would become additional multiplex owners.
from gateway.config import load_gateway_config
from utils import is_truthy_value
try:
multiplex_profiles = load_gateway_config().multiplex_profiles
except Exception:
log.warning(
"Unable to load gateway configuration during container boot; "
"using the GATEWAY_MULTIPLEX_PROFILES override if set.",
exc_info=True,
)
multiplex_profiles = is_truthy_value(
os.environ.get("GATEWAY_MULTIPLEX_PROFILES"),
)
log.warning("Unable to load gateway configuration during container boot; using the "
"GATEWAY_MULTIPLEX_PROFILES override if set.", exc_info=True)
multiplex_profiles = is_truthy_value(os.environ.get("GATEWAY_MULTIPLEX_PROFILES"))
# Default profile — always register, even if nothing has ever
# populated the root profile dir. The slot exists so
# ``hermes gateway start`` (no ``-p``) has somewhere to land;
# auto-up only when the prior state was "running" (same rule as
# named profiles). If the container was launched with the legacy
# `gateway run` command and no state exists yet, seed that intent
# as `running` so the s6 reconciler preserves the pre-s6 behavior.
# Default profile — always registered; auto-up only when the prior state was "running"
# (same rule as named profiles). A legacy `gateway run` container with no state yet seeds
# that intent as `running` so the pre-s6 behavior is preserved.
legacy_default_state = _maybe_migrate_legacy_gateway_run_state(
hermes_home,
container_argv=container_argv,
dry_run=dry_run,
)
hermes_home, container_argv=container_argv, dry_run=dry_run)
default_prior_state = legacy_default_state or _read_desired_state(hermes_home)
default_should_start = default_prior_state in _AUTOSTART_STATES
if not dry_run:
_cleanup_stale_runtime_files(hermes_home)
_register_service(scandir, "default", start=default_should_start)
actions.append(ReconcileAction(
profile="default",
prior_state=default_prior_state,
action="started" if default_should_start else "registered",
prior_exit=_read_prior_exit_label(hermes_home),
))
actions.append(_slot_action("default", hermes_home, default_prior_state, default_should_start))
profiles_root = hermes_home / "profiles"
if profiles_root.is_dir():
for entry in sorted(profiles_root.iterdir()):
if not entry.is_dir():
# SOUL.md is always seeded by `hermes profile create` (config.yaml comes later via
# `hermes setup`): the "real profile" marker that skips stray dirs (backups, mkdir).
if not entry.is_dir() or not (entry / "SOUL.md").exists():
continue
# SOUL.md is always seeded by `hermes profile create` (config.yaml
# is not — that comes later via `hermes setup`). Use it as the
# "real profile" marker so stray dirs (backups, manual mkdir)
# aren't picked up.
if not (entry / "SOUL.md").exists():
continue
# The "default" service name is reserved for the root
# profile (above) — if a user has somehow created a
# ``profiles/default/`` directory, skip it to avoid the
# slot collision. Their gateway would still be reachable
# via ``hermes -p default-named gateway start`` if they
# rename the directory; we don't try to disambiguate here.
# "default" is reserved for the root profile (above); skip a stray
# ``profiles/default/`` rather than collide on the slot.
if entry.name == "default":
log.warning(
"profiles/default/ exists — skipping to avoid colliding "
"with the reserved root-profile s6 slot",
)
log.warning("profiles/default/ exists — skipping to avoid colliding with the "
"reserved root-profile s6 slot")
continue
prior_state = _read_desired_state(entry)
should_start = (
not multiplex_profiles and prior_state in _AUTOSTART_STATES
)
should_start = not multiplex_profiles and prior_state in _AUTOSTART_STATES
if not dry_run:
_cleanup_stale_runtime_files(entry)
_register_service(scandir, entry.name, start=should_start)
actions.append(ReconcileAction(
profile=entry.name,
prior_state=prior_state,
action="started" if should_start else "registered",
prior_exit=_read_prior_exit_label(entry),
))
actions.append(_slot_action(entry.name, entry, prior_state, should_start))
if not dry_run:
_write_reconcile_log(hermes_home, actions)
@@ -187,17 +124,12 @@ def reconcile_profile_gateways(
def _maybe_migrate_legacy_gateway_run_state(
hermes_home: Path,
*,
container_argv: Sequence[str] | None,
dry_run: bool,
hermes_home: Path, *, container_argv: Sequence[str] | None, dry_run: bool
) -> str | None:
"""Seed root gateway_state for pre-s6 `gateway run` containers.
The tini image let Docker users run the gateway as the container command (`docker run ...
gateway run`). After the s6 migration, profile gateways are restored from persisted
gateway_state.json; a legacy container with no state file would therefore register the default
service down and never start.
The tini image let users run the gateway as the container command; post-s6, gateways are
restored from gateway_state.json, so such a container would register down and never start.
"""
state_file = hermes_home / "gateway_state.json"
if state_file.exists():
@@ -221,65 +153,45 @@ def _maybe_migrate_legacy_gateway_run_state(
return "running"
def _read_container_argv() -> tuple[str, ...]:
"""Best-effort read of the container's main program argv.
def _cmdline_argv(cmdline: Path) -> tuple[str, ...]:
raw = cmdline.read_bytes()
return tuple(part.decode("utf-8", "replace") for part in raw.split(b"\0") if part)
s6-overlay v2: PID 1 is ``/init`` and its argv holds ``main-wrapper.sh``. v3: PID 1 is
``s6-svscan`` and the real command lives on another PID, so after the PID 1 fast path we scan
``/proc/*/cmdline`` for a process whose argv contains ``main-wrapper.sh``.
def _read_container_argv() -> tuple[str, ...]:
"""Best-effort read of the container's main program argv (the one holding ``main-wrapper.sh``).
s6-overlay v2: PID 1 is ``/init``. v3: PID 1 is ``s6-svscan`` and the real command lives on
another PID, so after the PID 1 fast path we scan ``/proc/*/cmdline``.
"""
# Fast path: PID 1 is the command itself (s6-overlay v2 / tini).
try:
raw = Path("/proc/1/cmdline").read_bytes()
argv = tuple(
part.decode("utf-8", "replace") for part in raw.split(b"\0") if part
)
def _cmdlines():
yield Path("/proc/1/cmdline")
try:
for entry in Path("/proc").iterdir():
if entry.name.isdigit():
yield entry / "cmdline"
except OSError:
return
for cmdline in _cmdlines():
try:
argv = _cmdline_argv(cmdline)
except OSError:
continue
if any("main-wrapper.sh" in part for part in argv):
return argv
except OSError:
pass
# Slow path: s6-overlay v3 — PID 1 is s6-svscan; find the
# rc.init-launched process whose argv contains main-wrapper.sh.
try:
proc_dir = Path("/proc")
for entry in proc_dir.iterdir():
if not entry.name.isdigit():
continue
try:
raw = (entry / "cmdline").read_bytes()
except OSError:
continue
argv = tuple(
part.decode("utf-8", "replace")
for part in raw.split(b"\0")
if part
)
if any("main-wrapper.sh" in part for part in argv):
return argv
except OSError:
pass
return ()
def _strip_container_argv_prefix(argv: Sequence[str]) -> list[str]:
"""Strip the s6/wrapper prefix off the container argv, leaving the hermes args.
Rather than peel each leading token positionally (which silently breaks the moment s6 changes
its launcher shape again — exactly what happened in the v2→v3 bump), drop everything up to and
including the ``main-wrapper.sh`` token: that wrapper path is the stable boundary the image
owns, and the subcommand always follows it.
Drops everything through the ``main-wrapper.sh`` token — the stable boundary the image
owns — rather than peeling tokens positionally (which broke on the s6 v2→v3 bump).
"""
args = list(argv)
# Preferred boundary: everything through main-wrapper.sh is launcher
# prefix. Covers s6-overlay v2 (`/init …main-wrapper.sh …`) and v3
# (`/bin/sh -e …rc.init top …main-wrapper.sh …`) with one rule.
wrapper_idx = next(
(i for i, a in enumerate(args) if a.endswith("main-wrapper.sh")),
None,
)
wrapper_idx = next((i for i, a in enumerate(args) if a.endswith("main-wrapper.sh")), None)
if wrapper_idx is not None:
args = args[wrapper_idx + 1 :]
elif args and Path(args[0]).name == "init":
@@ -297,7 +209,7 @@ def _strip_container_argv_prefix(argv: Sequence[str]) -> list[str]:
def _is_legacy_gateway_run_request(argv: Sequence[str]) -> bool:
"""Return True for Docker commands equivalent to `gateway run`."""
"""True for Docker commands equivalent to `gateway run`."""
args = _strip_container_argv_prefix(argv)
if "--no-supervise" in args:
return False
@@ -305,58 +217,42 @@ def _is_legacy_gateway_run_request(argv: Sequence[str]) -> bool:
def _is_dashboard_container(argv: Sequence[str]) -> bool:
"""Return True when the container's command is the dashboard.
A dashboard-only container (``hermes dashboard ...``) never spawns or supervises per-profile
gateways — that is the gateway container's job.
"""
"""True when the container's command is the dashboard (which never supervises gateways)."""
args = _strip_container_argv_prefix(argv)
return bool(args) and args[0] == "dashboard"
def _read_desired_state(profile_dir: Path) -> str | None:
"""Read the persisted gateway desired state for reconciliation.
"""Persisted gateway desired state: ``desired_state`` (operator intent), else ``gateway_state``.
Newer state files carry ``desired_state``: operator intent written by s6 lifecycle commands.
Older files only carry ``gateway_state``; keep that as a compatibility fallback so existing
running/stopped profiles preserve their behavior until the next explicit start/stop.
Missing or unparseable files count as "no desired state" so we don't bork the whole
reconciliation on a corrupt file.
The older key is a fallback so existing profiles keep their behavior until the next explicit
start/stop. Missing/unparseable files count as "no state" so a corrupt file can't bork boot.
"""
state_file = profile_dir / "gateway_state.json"
if not state_file.exists():
return None
try:
data = json.loads(state_file.read_text(encoding="utf-8"))
desired_state = data.get("desired_state")
if desired_state is not None:
return desired_state
gateway_state = data.get("gateway_state")
if gateway_state in _TRANSIENT_RUNNING_STATES:
return "running"
return gateway_state
except (OSError, json.JSONDecodeError):
log.warning(
"could not read %s; treating as no prior state", state_file,
)
log.warning("could not read %s; treating as no prior state", state_file)
return None
desired_state = data.get("desired_state")
if desired_state is not None:
return desired_state
gateway_state = data.get("gateway_state")
if gateway_state in _TRANSIENT_RUNNING_STATES:
return "running"
return gateway_state
def _cleanup_stale_runtime_files(profile_dir: Path) -> None:
"""Remove gateway.pid and processes.json — they reference PIDs in the dead container's process
namespace and would otherwise confuse the newly-started gateway's process-mismatch checks.
"""
"""Remove PID-namespace-bound runtime files that would confuse the new gateway's checks."""
for name in _STALE_RUNTIME_FILES:
(profile_dir / name).unlink(missing_ok=True)
def _read_prior_exit_label(profile_dir: Path) -> str:
"""How the profile's previous gateway life ended (clean/unclean/unknown).
Thin, exception-free wrapper over :func:`gateway.lifecycle_ledger.read_prior_exit_label` — cont-
init runs in a minimal environment and forensics must never block reconciliation (NS-608).
"""
"""Exception-free ``lifecycle_ledger.read_prior_exit_label`` — forensics never block boot."""
try:
from gateway.lifecycle_ledger import read_prior_exit_label
return read_prior_exit_label(profile_dir)
@@ -364,36 +260,32 @@ def _read_prior_exit_label(profile_dir: Path) -> str:
return "unknown"
def _write_exec(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
path.chmod(0o755)
def _register_service(scandir: Path, profile: str, *, start: bool) -> None:
"""Recreate the s6 service slot for one profile.
Mirrors ``S6ServiceManager.register_profile_gateway`` but sets start state via the ``down``
marker directly: cont-init.d runs as root before s6-svscan scans the dynamic scandir, so the
manager's ``s6-svscanctl -a`` would fail with no control socket. Built in a sibling temp dir
and ``Path.replace``d into place so an interrupted write never leaves a half-populated dir.
marker: cont-init.d runs before s6-svscan scans the scandir, so ``s6-svscanctl -a`` has no
control socket yet. Built in a sibling temp dir and ``Path.replace``d into place so an
interrupted write never leaves a half-populated dir.
"""
import shutil
from hermes_cli.service_manager import (
S6ServiceManager,
_seed_supervise_skeleton,
validate_profile_name,
S6ServiceManager, _seed_supervise_skeleton, validate_profile_name,
)
validate_profile_name(profile)
service_dir = scandir / f"gateway-{profile}"
# Dot-prefix the staging dir so s6-svscan skips it while half-built
# (s6-svscan ignores scandir entries whose name starts with ".").
# A non-dotted ``.tmp`` staging name is supervised AS ROOT by any
# concurrent ``s6-svscanctl -a`` rescan the moment it has a valid
# ``type``/``run``, creating a root-owned ``supervise/`` that makes
# ``_seed_supervise_skeleton`` EACCES — see the matching comment in
# ``S6ServiceManager.register_profile_gateway``. The atomic
# ``tmp_dir.replace(service_dir)`` below renames to the dotless live
# name, so the published slot is unchanged.
# Dot-prefix the staging dir so s6-svscan skips it while half-built: a non-dotted name is
# supervised AS ROOT by any concurrent rescan the moment it has ``type``/``run``, creating
# a root-owned ``supervise/`` that makes ``_seed_supervise_skeleton`` EACCES.
tmp_dir = service_dir.with_name("." + service_dir.name + ".tmp")
# Wipe any leftover tmp from a previous interrupted run.
if tmp_dir.exists():
shutil.rmtree(tmp_dir, ignore_errors=True)
tmp_dir.mkdir(parents=True)
@@ -401,49 +293,25 @@ def _register_service(scandir: Path, profile: str, *, start: bool) -> None:
try:
(tmp_dir / "type").write_text("longrun\n", encoding="utf-8")
# Reuse the manager's run-script rendering — single source of
# truth so register_profile_gateway and reconcile_profile_gateways
# stay consistent. extra_env is empty here; users who need
# per-profile env can set it via the profile's config.yaml
# (which the gateway itself loads).
run = tmp_dir / "run"
run.write_text(S6ServiceManager._render_run_script(profile, extra_env={}), encoding="utf-8")
run.chmod(0o755)
# Reuse the manager's script rendering so both registration paths stay consistent.
# extra_env is empty: per-profile env comes from the profile's config.yaml.
_write_exec(tmp_dir / "run", S6ServiceManager._render_run_script(profile, extra_env={}))
_write_exec(tmp_dir / "finish", S6ServiceManager._render_finish_script())
finish = tmp_dir / "finish"
finish.write_text(S6ServiceManager._render_finish_script(), encoding="utf-8")
finish.chmod(0o755)
# Persistent log rotation.
(tmp_dir / "log").mkdir()
_write_exec(tmp_dir / "log" / "run", S6ServiceManager._render_log_run(profile))
# Persistent log rotation (OQ8-C).
log_subdir = tmp_dir / "log"
log_subdir.mkdir()
log_run = log_subdir / "run"
log_run.write_text(S6ServiceManager._render_log_run(profile), encoding="utf-8")
log_run.chmod(0o755)
# The presence of a `down` file tells s6-supervise to NOT
# start the service when s6-svscan picks it up. User brings
# it up explicitly with `hermes -p <profile> gateway start`
# (which routes through the Phase 4
# _dispatch_via_service_manager_if_s6 helper to `s6-svc -u`).
# `down` tells s6-supervise NOT to start on pickup; `hermes -p <profile> gateway start`
# brings it up (→ `s6-svc -u`).
if not start:
(tmp_dir / "down").touch()
# Pre-create the supervise/ skeleton with hermes ownership
# BEFORE we publish the slot. Mirrors the same pre-creation
# step in S6ServiceManager.register_profile_gateway — when
# s6-svscan picks the published slot up, the s6-supervise it
# spawns will EEXIST our dirs/FIFOs and inherit hermes
# ownership, so runtime s6-svc / s6-svstat / s6-svwait calls
# (all dispatched as the hermes user) won't hit EACCES. See
# ``_seed_supervise_skeleton`` in service_manager.py for the
# full rationale.
# Pre-create supervise/ with hermes ownership BEFORE publishing: s6-supervise will EEXIST
# our dirs/FIFOs and inherit it, so runtime s6-svc calls as the hermes user won't EACCES.
_seed_supervise_skeleton(tmp_dir)
# Publish atomically. Path.replace handles the existing-target
# case the same way os.rename does on POSIX: the target is
# silently replaced, so a previous reconcile pass's slot is
# cleanly overwritten in one operation.
# Publish atomically (Path.replace overwrites a previous pass's slot in one operation).
if service_dir.exists():
shutil.rmtree(service_dir)
tmp_dir.replace(service_dir)
@@ -452,32 +320,27 @@ def _register_service(scandir: Path, profile: str, *, start: bool) -> None:
raise
def _write_reconcile_log(
hermes_home: Path, actions: list[ReconcileAction],
) -> None:
"""Append one line per profile to $HERMES_HOME/logs/container-boot.log.
# 256 KiB soft cap on container-boot.log (~3000 lines ≈ a year of daily reboots on a
# 5-profile container), rotated to .1 when crossed. Tuned for grep-ability, not space.
_LOG_ROTATE_BYTES = 256 * 1024
Operators inspect this to debug "why didn't my profile come back up". Keeping a separate log
file (vs. mixing into agent.log) lets troubleshooters grep for "profile=foo" without wading
through unrelated activity.
Size-bounded: when the file exceeds ``_LOG_ROTATE_BYTES`` (defaults to 256 KiB ≈ 3000 reconcile
lines), the current file is renamed to ``container-boot.log.1`` (replacing any previous
rotation) before the new entries are appended.
def _write_reconcile_log(hermes_home: Path, actions: list[ReconcileAction]) -> None:
"""Append one line per profile to $HERMES_HOME/logs/container-boot.log (rotated to ``.1``).
A separate file (vs. agent.log) lets operators grep "profile=foo" when debugging "why
didn't my profile come back up".
"""
import time
log_dir = hermes_home / "logs"
log_dir.mkdir(parents=True, exist_ok=True)
log_path = log_dir / "container-boot.log"
# Rotate before opening to append, so the new entries always land
# in a fresh file when we crossed the threshold last time.
try:
if log_path.exists() and log_path.stat().st_size >= _LOG_ROTATE_BYTES:
log_path.replace(log_dir / "container-boot.log.1")
except OSError as exc:
# Rotation failure is non-fatal — keep appending to the
# existing file rather than losing the entry entirely.
# Non-fatal — keep appending rather than lose the entry entirely.
log.warning("could not rotate %s: %s", log_path, exc)
ts = time.strftime("%Y-%m-%dT%H:%M:%S%z")
@@ -489,42 +352,21 @@ def _write_reconcile_log(
)
# 256 KiB soft cap on container-boot.log; rotated to .1 when crossed.
# At ~80 B per reconcile-action line this is ~3000 lines, or about a
# year of daily reboots on a 5-profile container. Two files = ~512 KiB
# worst case. Tuned for visibility (small enough to grep / cat without
# scrolling forever) more than space (the persistent volume has GB).
_LOG_ROTATE_BYTES = 256 * 1024
def main() -> int:
"""Entry point invoked from /etc/cont-init.d/02-reconcile-profiles."""
# A dashboard-only container never spawns or supervises per-profile
# gateways, so reconciling their s6 slots here is pure waste — and
# actively harmful: when the gateway and dashboard containers share a
# bind-mounted HERMES_HOME, both race to flock() the same s6-log lock
# files under logs/gateways/<profile>/lock, producing "Resource busy"
# failures and a restart storm. Detect the role from PID 1 argv and
# skip reconciliation in the dashboard container. No operator flag:
# the role is a fact about the container's command, and a flag can be
# forgotten in a hand-written manifest, reintroducing the storm.
# A dashboard-only container never supervises gateways, and reconciling here is harmful:
# with a shared bind-mounted HERMES_HOME both containers race to flock() the same s6-log
# lock files → "Resource busy" restart storm. Detected from PID 1 argv, not an operator
# flag — a flag can be forgotten in a hand-written manifest.
if _is_dashboard_container(_read_container_argv()):
print(
"reconcile: skipping (dashboard container — does not need "
"per-profile gateways)"
)
print("reconcile: skipping (dashboard container — does not need per-profile gateways)")
return 0
hermes_home = Path(os.environ.get("HERMES_HOME", "/opt/data"))
scandir = Path(os.environ.get("S6_PROFILE_GATEWAY_SCANDIR", "/run/service"))
actions = reconcile_profile_gateways(
hermes_home=hermes_home, scandir=scandir,
)
actions = reconcile_profile_gateways(hermes_home=hermes_home, scandir=scandir)
for a in actions:
print(
f"reconcile: profile={a.profile} "
f"prior_state={a.prior_state} action={a.action}"
)
print(f"reconcile: profile={a.profile} prior_state={a.prior_state} action={a.action}")
return 0
+220 -323
View File
@@ -6,13 +6,17 @@ GH_TOKEN env var 3. GITHUB_TOKEN env var 4. gh auth token CLI fallback
from __future__ import annotations
import hashlib
import json
import logging
import os
import re
import shutil
import subprocess
import threading
import time
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Optional
@@ -20,13 +24,9 @@ from hermes_cli._subprocess_compat import IS_WINDOWS, windows_hide_flags
logger = logging.getLogger(__name__)
# OAuth device code flow constants — VS Code's GitHub App client ID.
# The previous opencode OAuth App ID (Ov23li8tweQw6odWQebz) produces gho_*
# tokens that cannot be exchanged for Copilot API JWTs (404 on
# /copilot_internal/v2/token). VS Code's App ID produces ghu_* tokens
# that support exchange, which is required to access internal-only models
# (e.g. claude-opus-4.6-1m) and enterprise endpoints.
# Tested on Individual and Enterprise accounts.
# OAuth device code flow — VS Code's GitHub App client ID: it mints ghu_* tokens that can be
# exchanged for Copilot API JWTs (required for internal-only models and enterprise endpoints).
# The opencode App ID (Ov23li8tweQw6odWQebz) mints gho_* tokens that 404 on exchange.
COPILOT_OAUTH_CLIENT_ID = "Iv1.b507a08c87ecfe98"
# ghp_ classic PATs are rejected by the Copilot API (gho_ / github_pat_ / ghu_ work).
_CLASSIC_PAT_PREFIX = "ghp_"
@@ -58,77 +58,56 @@ def validate_copilot_token(token: str) -> tuple[bool, str]:
def resolve_copilot_token() -> tuple[str, str]:
"""Resolve a GitHub token suitable for Copilot API use.
"""Resolve a GitHub token suitable for Copilot API use → (token, source); ("", "") if none.
Returns (token, source) where source describes where the token came from. Raises ValueError if
only a classic PAT is available.
Raises ValueError if only a classic PAT is available.
"""
# 1. Check env vars in priority order
any_env_var_set = False
for env_var in COPILOT_ENV_VARS:
val = os.getenv(env_var, "").strip()
if val:
any_env_var_set = True
valid, msg = validate_copilot_token(val)
if not valid:
logger.warning(
"Token from %s is not supported: %s", env_var, msg
)
continue
if not val:
continue
any_env_var_set = True
valid, msg = validate_copilot_token(val)
if valid:
return val, env_var
logger.warning("Token from %s is not supported: %s", env_var, msg)
# 2. Fall back to gh auth token — but ONLY when no Copilot env var was
# explicitly set. When the user exported GITHUB_TOKEN (even an
# unsupported classic PAT), their intent is to use *that* token, not
# to silently substitute one from the gh CLI credential store.
# Skipping the subprocess here also avoids a slow `gh auth token`
# call (up to 5s timeout on Windows) on every cold start that scans
# Copilot auth state — a measurable contributor to the ~14s
# cold-start stall (#60800). The user can run `copilot login` or
# set a supported token (gho_*/github_pat_*/ghu_) explicitly.
# Fall back to gh auth token ONLY when no Copilot env var was explicitly set: an exported
# GITHUB_TOKEN (even an unsupported classic PAT) means the user intends *that* token, not
# one silently substituted from the gh credential store. Skipping the subprocess also
# avoids a slow `gh auth token` call (up to 5s on Windows) on every cold start.
if any_env_var_set:
logger.debug(
"Copilot env var(s) set but none held a supported token; "
"skipping `gh auth token` fallback to honor explicit env-var "
"intent (and avoid the subprocess cost on cold start, #60800)."
)
logger.debug("Copilot env var(s) set but none held a supported token; skipping `gh auth "
"token` fallback to honor explicit env-var intent (and avoid the subprocess "
"cost on cold start, #60800).")
return "", ""
token = _try_gh_cli_token()
if token:
valid, msg = validate_copilot_token(token)
if not valid:
raise ValueError(
f"Token from `gh auth token` is a classic PAT (ghp_*). {msg}"
)
raise ValueError(f"Token from `gh auth token` is a classic PAT (ghp_*). {msg}")
return token, "gh auth token"
return "", ""
def _gh_cli_candidates() -> list[str]:
"""Return candidate ``gh`` binary paths, including common Homebrew installs."""
"""Candidate ``gh`` binary paths, including common Homebrew installs."""
candidates: list[str] = [c for c in (shutil.which("gh"),) if c]
for candidate in (
"/opt/homebrew/bin/gh",
"/usr/local/bin/gh",
str(Path.home() / ".local" / "bin" / "gh"),
):
if candidate not in candidates and os.path.isfile(candidate) and os.access(candidate, os.X_OK):
candidates.append(candidate)
candidates += [
c for c in ("/opt/homebrew/bin/gh", "/usr/local/bin/gh", str(Path.home() / ".local/bin/gh"))
if c not in candidates and os.path.isfile(c) and os.access(c, os.X_OK)
]
return candidates
# ``gh auth token`` result cache. The probe shells out to the gh CLI, and when
# gh has no credential store for this HOME (fresh profile, desktop-spawned
# backend, CI) it can block for its full 5s subprocess timeout — on keyring /
# D-Bus prompts rather than returning immediately. Provider inventory builds
# (``/api/model/options``, ``hermes tools``) probe Copilot auth several times
# per request, so an uncached miss turns one settings-page load into a 4×5s
# stall that exceeds the Desktop renderer's 15s IPC budget and paints an error
# (observed Aug 2026: Models/Providers settings pages timing out on every
# open). Successes and failures are both cached; a short TTL keeps a freshly
# run ``gh auth login`` discoverable without restarting the backend.
# ``gh auth token`` result cache. With no credential store for this HOME (fresh profile, CI)
# the probe blocks for its full 5s timeout on keyring / D-Bus prompts, and provider inventory
# builds probe Copilot auth several times per request — an uncached miss made one settings
# page a 4×5s stall past the Desktop renderer's 15s IPC budget. Misses are cached too; a short
# TTL keeps a fresh ``gh auth login`` discoverable.
_GH_CLI_TOKEN_CACHE_TTL_SECONDS = 300.0
_gh_cli_token_cache: tuple[float, Optional[str]] | None = None
@@ -140,17 +119,13 @@ def _invalidate_gh_cli_token_cache() -> None:
def _try_gh_cli_token() -> Optional[str]:
"""Return a token from ``gh auth token`` when the GitHub CLI is available.
The result (including a miss) is cached for a short TTL — see the cache comment above. Callers
that just re-authenticated can call ``_invalidate_gh_cli_token_cache()`` to re-probe
immediately.
"""
"""Token from ``gh auth token`` when available; the result (incl. a miss) is cached per TTL."""
global _gh_cli_token_cache
now = time.monotonic()
if _gh_cli_token_cache is not None and now - _gh_cli_token_cache[0] < _GH_CLI_TOKEN_CACHE_TTL_SECONDS:
return _gh_cli_token_cache[1]
cache = _gh_cli_token_cache
if cache is not None and now - cache[0] < _GH_CLI_TOKEN_CACHE_TTL_SECONDS:
return cache[1]
token = _probe_gh_cli_token()
_gh_cli_token_cache = (now, token)
@@ -161,10 +136,9 @@ def _probe_gh_cli_token() -> Optional[str]:
"""Uncached ``gh auth token`` subprocess probe (see ``_try_gh_cli_token``)."""
hostname = os.getenv("COPILOT_GH_HOST", "").strip()
# Build a clean env so gh doesn't short-circuit on GITHUB_TOKEN / GH_TOKEN
clean_env = {k: v for k, v in os.environ.items()
if k not in {"GITHUB_TOKEN", "GH_TOKEN"}}
# Never let gh open an interactive prompt from a backend process.
# Clean env so gh doesn't short-circuit on GITHUB_TOKEN / GH_TOKEN, and never let it open
# an interactive prompt from a backend process.
clean_env = {k: v for k, v in os.environ.items() if k not in {"GITHUB_TOKEN", "GH_TOKEN"}}
clean_env.setdefault("GH_PROMPT_DISABLED", "1")
clean_env.setdefault("GH_NO_UPDATE_NOTIFIER", "1")
@@ -174,15 +148,9 @@ def _probe_gh_cli_token() -> Optional[str]:
if hostname:
cmd += ["--hostname", hostname]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True, encoding='utf-8', errors='replace',
timeout=5,
env=clean_env,
stdin=subprocess.DEVNULL,
**_popen_kwargs,
)
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8',
errors='replace', timeout=5, env=clean_env,
stdin=subprocess.DEVNULL, **_popen_kwargs)
except (FileNotFoundError, subprocess.TimeoutExpired) as exc:
logger.debug("gh CLI token lookup failed (%s): %s", gh_path, exc)
continue
@@ -193,42 +161,35 @@ def _probe_gh_cli_token() -> Optional[str]:
# ─── OAuth Device Code Flow ────────────────────────────────────────────────
_DEVICE_CODE_TERMINAL_ERRORS = {
"expired_token": " ✗ Device code expired. Please try again.",
"access_denied": " ✗ Authorization was denied.",
}
_DEVICE_CODE_TERMINAL_ERRORS = {"expired_token": " ✗ Device code expired. Please try again.",
"access_denied": " ✗ Authorization was denied."}
def _post_form(url: str, fields: dict, timeout: float) -> dict:
req = urllib.request.Request(
url,
data=urllib.parse.urlencode(fields).encode(),
headers={
"Accept": "application/json",
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": "HermesAgent/1.0",
},
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode())
def copilot_device_code_login(
*,
host: str = "github.com",
timeout_seconds: float = 300,
*, host: str = "github.com", timeout_seconds: float = 300,
) -> Optional[str]:
"""Run the GitHub OAuth device code flow for Copilot."""
import urllib.request
import urllib.parse
domain = host.rstrip("/")
device_code_url = f"https://{domain}/login/device/code"
access_token_url = f"https://{domain}/login/oauth/access_token"
def _post_form(url: str, fields: dict, timeout: float) -> dict:
req = urllib.request.Request(
url,
data=urllib.parse.urlencode(fields).encode(),
headers={
"Accept": "application/json",
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": "HermesAgent/1.0",
},
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode())
# Step 1: Request device code
try:
device_data = _post_form(
device_code_url, {"client_id": COPILOT_OAUTH_CLIENT_ID, "scope": "read:user"}, 15
)
device_data = _post_form(device_code_url,
{"client_id": COPILOT_OAUTH_CLIENT_ID, "scope": "read:user"}, 15)
except Exception as exc:
logger.error("Failed to initiate device authorization: %s", exc)
print(f" ✗ Failed to start device authorization: {exc}")
@@ -243,14 +204,10 @@ def copilot_device_code_login(
print(" ✗ GitHub did not return a device code.")
return None
# Step 2: Show instructions
print()
print(f" Open this URL in your browser: {verification_uri}")
print(f" Enter this code: {user_code}")
print()
print(f"\n Open this URL in your browser: {verification_uri}\n"
f" Enter this code: {user_code}\n")
print(" Waiting for authorization...", end="", flush=True)
# Step 3: Poll for completion
deadline = time.monotonic() + timeout_seconds
while time.monotonic() < deadline:
@@ -276,29 +233,26 @@ def copilot_device_code_login(
error = result.get("error", "")
if error == "slow_down":
# RFC 8628: add 5 seconds to polling interval
# RFC 8628: add 5 seconds to polling interval (or honor a server-supplied one)
server_interval = result.get("interval")
if isinstance(server_interval, (int, float)) and server_interval > 0:
interval = int(server_interval)
else:
interval += 5
is_num = isinstance(server_interval, (int, float)) and server_interval > 0
interval = int(server_interval) if is_num else interval + 5
if error in ("authorization_pending", "slow_down"):
print(".", end="", flush=True)
continue
if error:
print()
print(_DEVICE_CODE_TERMINAL_ERRORS.get(error, f" ✗ Authorization failed: {error}"))
print("\n" + _DEVICE_CODE_TERMINAL_ERRORS.get(error,
f" ✗ Authorization failed: {error}"))
return None
print()
print(" ✗ Timed out waiting for authorization.")
print("\n ✗ Timed out waiting for authorization.")
return None
# ─── Copilot Token Exchange ────────────────────────────────────────────────
# Module-level cache for exchanged Copilot API tokens.
# Maps raw_token_fingerprint -> (api_token, expires_at_epoch, base_url).
# In-process cache of exchanged Copilot API tokens:
# raw_token_fingerprint -> (api_token, expires_at_epoch, base_url).
_jwt_cache: dict[str, tuple[str, float, Optional[str]]] = {}
_JWT_REFRESH_MARGIN_SECONDS = 120 # refresh 2 min before expiry
@@ -307,32 +261,24 @@ _TOKEN_EXCHANGE_URL = "https://api.github.com/copilot_internal/v2/token"
_EDITOR_VERSION = "vscode/1.104.1"
_EXCHANGE_USER_AGENT = "GitHubCopilotChat/0.26.7"
# Transient-failure hardening for the token exchange. Gateway startup often
# races network readiness (launchd relaunch, DHCP/VPN settling); a single-shot
# exchange that fails there silently degrades to the RAW GitHub token, which the
# Copilot server routes to the "copilot-language-server" integrator whose model
# allowlist omits enterprise-only models (e.g. claude-opus-4.8) → HTTP 400 on
# every turn until the next restart. Retry a few times, and persist the last
# good exchanged JWT to disk so a restart during a blip reuses the still-valid
# ~30-min token instead of degrading.
# Transient-failure hardening. Gateway startup races network readiness (launchd relaunch,
# DHCP/VPN settling); a single-shot exchange failing there silently degrades to the RAW GitHub
# token, which Copilot routes to the "copilot-language-server" integrator whose allowlist omits
# enterprise-only models → HTTP 400 every turn until restart. Retry, and persist the last good
# JWT so a restart during a blip reuses the still-valid ~30-min token.
_EXCHANGE_MAX_ATTEMPTS = 3
_EXCHANGE_BACKOFF_BASE_SECONDS = 1.5 # sleeps ~1.5s, ~3.0s between attempts
_JWT_DISK_FILENAME = ".copilot_jwt.json"
_JWT_DISK_MAX_BYTES = 1_048_576 # 1 MiB cap on the persisted JWT store read
# Negative cache for failed exchanges. Without it, every load_pool("copilot")
# call re-runs the full exchange — and on a permanently-rejected token
# (HTTP 403: account not Copilot-entitled, expired grant, org policy) the
# retry backoff burned ~4.5s of time.sleep() on EVERY provider-discovery
# pass. The /model picker, delegation child spawns, and the web dashboard
# all walk that path, so a single bad Copilot token made all of them crawl.
# Maps raw-token fingerprint -> epoch until which exchange attempts are
# skipped (raise immediately). Success clears the entry.
# Negative cache: fingerprint -> epoch until which attempts raise immediately (success clears
# it). Without it a permanently-rejected token (403: not entitled, expired grant, org policy)
# burned ~4.5s of retry backoff on EVERY provider-discovery pass (/model picker, delegation
# spawns, dashboard).
_exchange_failure_cache: dict[str, float] = {}
# Single-flight guard per token fingerprint: concurrent callers (the dashboard
# polls /api/credentials/pool every few seconds, each poll off-loop) wait on
# the ONE in-flight exchange and then hit the positive/negative cache, instead
# of each spawning their own hung resolver thread during a DNS outage.
# Single-flight guard per fingerprint: concurrent callers (dashboard polls /api/credentials/
# pool every few seconds) wait on the ONE in-flight exchange and then hit the cache, instead
# of each spawning its own hung resolver thread during a DNS outage.
_exchange_locks: dict[str, threading.Lock] = {}
_exchange_locks_guard = threading.Lock()
@@ -343,34 +289,33 @@ def _exchange_lock_for(fp: str) -> threading.Lock:
if lock is None:
lock = _exchange_locks[fp] = threading.Lock()
return lock
_EXCHANGE_FAILURE_TTL_TRANSIENT_SECONDS = 60.0 # network blips: retry soon
_EXCHANGE_FAILURE_TTL_PERMANENT_SECONDS = 1800.0 # 401/403/404: won't heal
# HTTP statuses that indicate the token itself is rejected — retrying with
# backoff is pointless (the retry loop exists for startup network races,
# not for auth rejections) and sleeping on them just blocks the caller.
# HTTP statuses meaning the token itself is rejected — retrying with backoff is pointless
# (the retry loop exists for startup network races) and sleeping on them just blocks the caller.
_EXCHANGE_PERMANENT_HTTP_STATUSES = frozenset({401, 403, 404})
def _token_fingerprint(raw_token: str) -> str:
"""Short fingerprint of a raw token for cache keying (avoids storing full token)."""
import hashlib
return hashlib.sha256(raw_token.encode()).hexdigest()[:16]
def _read_jwt_store(path: Path) -> Optional[dict]:
"""Bounded read of the on-disk JWT store → dict, or None if missing/unusable.
Single chokepoint for every read of the persisted store (load, eviction, save-merge). A well-
formed store is a few KB; a file over the 1 MiB cap or with non-dict content is treated as
unusable so a corrupt/oversized file can't balloon memory or get rewritten back out.
Single chokepoint for every read (load, eviction, save-merge). A well-formed store is a
few KB; a file over the 1 MiB cap or with non-dict content is unusable so a corrupt or
oversized file can't balloon memory or get rewritten back out.
"""
if not path.exists():
return None
try:
if path.stat().st_size > _JWT_DISK_MAX_BYTES:
logger.debug(
"Persisted Copilot JWT store exceeds %d bytes; ignoring", _JWT_DISK_MAX_BYTES
)
logger.debug("Persisted Copilot JWT store exceeds %d bytes; ignoring",
_JWT_DISK_MAX_BYTES)
return None
loaded = json.loads(path.read_text(encoding="utf-8"))
return loaded if isinstance(loaded, dict) else None
@@ -390,33 +335,6 @@ def _write_jwt_store(path: Path, store: dict) -> None:
os.replace(tmp, path)
def evict_cached_exchanged_token(raw_token: str) -> None:
"""Drop any cached exchanged JWT for ``raw_token`` (in-process + on-disk).
Used by the runtime stale-credential recovery path: when a live request starts failing with a
Copilot ``model_not_available_for_integrator`` / ``model_not_supported`` 400, the cached
exchanged token (or a degraded raw fallback that was cached in its place) is stale.
"""
if not raw_token:
return
fp = _token_fingerprint(raw_token)
_jwt_cache.pop(fp, None)
# Also clear any negative-cache entry: eviction is an explicit "force a
# fresh exchange" signal from the stale-credential recovery path, so the
# next exchange_copilot_token() must be allowed to hit the network.
_exchange_failure_cache.pop(fp, None)
path = _jwt_disk_path()
if not path:
return
try:
store = _read_jwt_store(path)
if store is not None and fp in store:
del store[fp]
_write_jwt_store(path, store)
except Exception as exc:
logger.debug("Failed to evict cached Copilot JWT: %s", exc)
def _jwt_disk_path() -> Optional[Path]:
"""Path to the on-disk exchanged-JWT cache (profile-aware), or None."""
try:
@@ -426,69 +344,79 @@ def _jwt_disk_path() -> Optional[Path]:
return None
def _load_jwt_from_disk(fp: str) -> Optional[tuple[str, float, Optional[str]]]:
"""Load a persisted exchanged JWT for ``fp`` → (api_token, expires_at, base_url).
An oversized/corrupt store is treated as unusable — the caller re-exchanges (bound shared with
eviction/save via _read_jwt_store).
"""
def _with_jwt_store(verb: str, op):
"""Run ``op(path, store_or_None)`` against the disk store; failures are logged, never raised."""
path = _jwt_disk_path()
if not path:
return None
try:
entry = (_read_jwt_store(path) or {}).get(fp)
return op(path, _read_jwt_store(path))
except Exception as exc:
logger.debug("Failed to %s Copilot JWT: %s", verb, exc)
return None
def evict_cached_exchanged_token(raw_token: str) -> None:
"""Drop any cached exchanged JWT for ``raw_token`` (in-process + on-disk).
Used by the runtime stale-credential recovery path when a live request starts failing with
a Copilot ``model_not_available_for_integrator`` / ``model_not_supported`` 400.
"""
if not raw_token:
return
fp = _token_fingerprint(raw_token)
_jwt_cache.pop(fp, None)
# Eviction is an explicit "force a fresh exchange" signal, so the negative-cache entry
# must go too — the next exchange_copilot_token() must be allowed to hit the network.
_exchange_failure_cache.pop(fp, None)
def _evict(path, store):
if store is not None and fp in store:
del store[fp]
_write_jwt_store(path, store)
_with_jwt_store("evict cached", _evict)
def _load_jwt_from_disk(fp: str) -> Optional[tuple[str, float, Optional[str]]]:
"""Persisted exchanged JWT for ``fp`` → (api_token, expires_at, base_url), or None."""
def _load(path, store):
entry = (store or {}).get(fp)
if not isinstance(entry, dict):
return None
api_token = entry.get("api_token", "")
expires_at = float(entry.get("expires_at", 0) or 0)
if api_token and expires_at:
return api_token, expires_at, entry.get("base_url")
except Exception as exc:
logger.debug("Failed to load persisted Copilot JWT: %s", exc)
return None
return (api_token, expires_at, entry.get("base_url")) if api_token and expires_at else None
return _with_jwt_store("load persisted", _load)
def _save_jwt_to_disk(
fp: str, api_token: str, expires_at: float, base_url: Optional[str]
) -> None:
def _save_jwt_to_disk(fp: str, api_token: str, expires_at: float, base_url: Optional[str]) -> None:
"""Persist an exchanged JWT (0o600), pruning expired entries."""
path = _jwt_disk_path()
if not path:
return
try:
def _save(path, store):
now = time.time()
store = {
k: v
for k, v in (_read_jwt_store(path) or {}).items()
kept = {
k: v for k, v in (store or {}).items()
if isinstance(v, dict) and float(v.get("expires_at", 0) or 0) > now
}
store[fp] = {
"api_token": api_token,
"expires_at": expires_at,
"base_url": base_url,
}
_write_jwt_store(path, store)
except Exception as exc:
logger.debug("Failed to persist Copilot JWT: %s", exc)
kept[fp] = {"api_token": api_token, "expires_at": expires_at, "base_url": base_url}
_write_jwt_store(path, kept)
_with_jwt_store("persist", _save)
# Hard wall-clock cap for the token-exchange HTTP call. urllib's ``timeout``
# only bounds socket operations AFTER DNS resolution succeeds; getaddrinfo
# blocks in C and ignores it entirely, so on a networkless Windows host the
# resolver can hang for many minutes (observed: a 17-minute event-loop stall
# on 2026-08-22 that took the whole backend down with it).
# Hard wall-clock cap for the exchange call: urllib's ``timeout`` only bounds socket ops AFTER
# DNS succeeds; getaddrinfo blocks in C and ignores it, so a networkless Windows host can hang
# for many minutes (observed: a 17-minute event-loop stall that took the backend down).
_DNS_GRACE_SECONDS = 5.0
def _urlopen_bounded(req, timeout: float):
"""urlopen() with a hard wall-clock cap of timeout + _DNS_GRACE_SECONDS.
Runs the call on a daemon thread and abandons it if the cap fires, so a
DNS/getaddrinfo hang cannot block the caller indefinitely. Raises the
worker's exception, or TimeoutError when the cap fires.
Runs the call on a daemon thread and abandons it if the cap fires, so a DNS/getaddrinfo
hang cannot block the caller. Raises the worker's exception, or TimeoutError on the cap.
"""
import urllib.request
box: dict = {}
abandoned = threading.Event()
@@ -499,8 +427,7 @@ def _urlopen_bounded(req, timeout: float):
box["exc"] = exc
return
if abandoned.is_set():
# The caller already timed out; nobody will read this response,
# so release its socket instead of leaking it with the thread.
# Caller already timed out; nobody will read this response — release its socket.
try:
resp.close()
except Exception:
@@ -508,9 +435,7 @@ def _urlopen_bounded(req, timeout: float):
return
box["resp"] = resp
t = threading.Thread(
target=_worker, name="copilot-token-exchange", daemon=True
)
t = threading.Thread(target=_worker, name="copilot-token-exchange", daemon=True)
t.start()
t.join(timeout + _DNS_GRACE_SECONDS)
if t.is_alive():
@@ -526,19 +451,60 @@ def _urlopen_bounded(req, timeout: float):
return box["resp"]
def exchange_copilot_token(raw_token: str, *, timeout: float = 10.0) -> tuple[str, float, Optional[str]]:
"""Exchange a raw GitHub token for a short-lived Copilot API token.
def _fetch_exchange_with_retry(req, timeout: float, fp: str) -> dict:
"""GET the exchange with backoff for startup network races; raises ValueError on failure.
The token is a semicolon-separated string (not a JWT) used as a Bearer token. ``base_url`` is
the account-specific host: the exchange's ``endpoints.api`` (enterprise/proxied accounts), else
derived from the token's ``proxy-ep``; individual accounts have neither, so it is None. Cached
in-process until close to expiry. Raises ``ValueError`` on failure.
Permanent HTTP rejections (401/403/404) skip the retry loop entirely — sleeping on an auth
rejection blocks the caller for ~4.5s with an identical outcome. Failures populate the
negative cache (long TTL for permanent, short for transient); success clears it.
"""
last_exc: Optional[Exception] = None
permanent_failure = False
for attempt in range(1, _EXCHANGE_MAX_ATTEMPTS + 1):
try:
with _urlopen_bounded(req, timeout) as resp:
data = json.loads(resp.read().decode())
_exchange_failure_cache.pop(fp, None)
return data
except Exception as exc: # noqa: BLE001 — retry all, re-raise below
last_exc = exc
status = getattr(exc, "code", None) or getattr(exc, "status", None)
permanent_failure = status in _EXCHANGE_PERMANENT_HTTP_STATUSES
if permanent_failure:
logger.debug("Copilot token exchange rejected (HTTP %s); not retrying", status)
break
if attempt < _EXCHANGE_MAX_ATTEMPTS:
sleep_s = _EXCHANGE_BACKOFF_BASE_SECONDS * attempt
logger.debug("Copilot token exchange attempt %d/%d failed (%s); retrying in %.1fs",
attempt, _EXCHANGE_MAX_ATTEMPTS, exc, sleep_s)
time.sleep(sleep_s)
ttl = (_EXCHANGE_FAILURE_TTL_PERMANENT_SECONDS if permanent_failure
else _EXCHANGE_FAILURE_TTL_TRANSIENT_SECONDS)
_exchange_failure_cache[fp] = time.time() + ttl
raise ValueError(
f"Copilot token exchange failed after {_EXCHANGE_MAX_ATTEMPTS} attempts: {last_exc}"
) from last_exc
def _cache_entry_fresh(cached) -> bool:
return bool(cached) and time.time() < cached[1] - _JWT_REFRESH_MARGIN_SECONDS
def exchange_copilot_token(
raw_token: str, *, timeout: float = 10.0,
) -> tuple[str, float, Optional[str]]:
"""Exchange a raw GitHub token for a Copilot API token → (token, expires_at, base_url).
The token is a semicolon-separated string (not a JWT) used as a Bearer token. ``base_url``
is the account-specific host: the exchange's ``endpoints.api`` (enterprise/proxied
accounts), else derived from the token's ``proxy-ep``; individual accounts have neither,
so it is None. Cached in-process until close to expiry. Raises ``ValueError`` on failure.
"""
fp = _token_fingerprint(raw_token)
# Fast path outside the lock: a valid in-process JWT needs no exchange.
cached = _jwt_cache.get(fp)
if cached and time.time() < cached[1] - _JWT_REFRESH_MARGIN_SECONDS:
if _cache_entry_fresh(cached):
return cached
with _exchange_lock_for(fp):
@@ -546,26 +512,20 @@ def exchange_copilot_token(raw_token: str, *, timeout: float = 10.0) -> tuple[st
def _exchange_copilot_token_locked(
raw_token: str, fp: str, *, timeout: float
raw_token: str, fp: str, *, timeout: float,
) -> tuple[str, float, Optional[str]]:
import urllib.request
# Re-check the in-process cache under the lock (a concurrent caller may have
# just completed the exchange we were queued behind), then the on-disk
# cache: a fresh process (e.g. gateway restart) has an empty in-process
# cache but may have a still-valid persisted JWT. Reusing it avoids a
# network round-trip at startup — precisely when the network is most likely
# to be flaky and the single-shot exchange would degrade to the raw token.
# Re-check the in-process cache under the lock (a concurrent caller may have just completed
# the exchange we were queued behind), then the on-disk cache: a fresh process (gateway
# restart) has an empty in-process cache but may hold a still-valid persisted JWT, and
# reusing it avoids a network round-trip precisely when the network is most likely flaky.
for lookup in (_jwt_cache.get, _load_jwt_from_disk):
cached = lookup(fp)
if cached and time.time() < cached[1] - _JWT_REFRESH_MARGIN_SECONDS:
if _cache_entry_fresh(cached):
_jwt_cache[fp] = cached
return cached
# Negative cache: a recent exchange failure for this token means the
# network round-trip (and its retry backoff) would just repeat. Fail
# fast so provider discovery / picker opens don't block on a token we
# already know is rejected or unreachable.
# Negative cache: fail fast so provider discovery / picker opens don't block on a token
# we already know is rejected or unreachable.
_fail_until = _exchange_failure_cache.get(fp, 0.0)
if time.time() < _fail_until:
raise ValueError(
@@ -584,59 +544,17 @@ def _exchange_copilot_token_locked(
},
)
# Retry with backoff. Startup network races (launchd relaunch, VPN/DHCP
# settling) make the first attempt flaky; without this the sole failure
# silently degrades to the raw token for the whole process lifetime.
# Permanent HTTP rejections (401/403/404 — token not Copilot-entitled,
# revoked, or org-blocked) skip the retry loop entirely: backoff exists
# for transient network races, and sleeping on an auth rejection just
# blocks the caller for ~4.5s with an identical outcome.
data = None
last_exc: Optional[Exception] = None
permanent_failure = False
for attempt in range(1, _EXCHANGE_MAX_ATTEMPTS + 1):
try:
with _urlopen_bounded(req, timeout) as resp:
data = json.loads(resp.read().decode())
break
except Exception as exc: # noqa: BLE001 — retry all, re-raise below
last_exc = exc
status = getattr(exc, "code", None) or getattr(exc, "status", None)
permanent_failure = status in _EXCHANGE_PERMANENT_HTTP_STATUSES
if permanent_failure:
logger.debug("Copilot token exchange rejected (HTTP %s); not retrying", status)
break
if attempt < _EXCHANGE_MAX_ATTEMPTS:
sleep_s = _EXCHANGE_BACKOFF_BASE_SECONDS * attempt
logger.debug(
"Copilot token exchange attempt %d/%d failed (%s); retrying in %.1fs",
attempt, _EXCHANGE_MAX_ATTEMPTS, exc, sleep_s,
)
time.sleep(sleep_s)
if data is None:
ttl = (
_EXCHANGE_FAILURE_TTL_PERMANENT_SECONDS
if permanent_failure
else _EXCHANGE_FAILURE_TTL_TRANSIENT_SECONDS
)
_exchange_failure_cache[fp] = time.time() + ttl
raise ValueError(
f"Copilot token exchange failed after {_EXCHANGE_MAX_ATTEMPTS} attempts: {last_exc}"
) from last_exc
_exchange_failure_cache.pop(fp, None)
data = _fetch_exchange_with_retry(req, timeout, fp)
api_token = data.get("token", "")
if not api_token:
raise ValueError("Copilot token exchange returned empty token")
expires_at = data.get("expires_at", 0)
expires_at = float(expires_at) if expires_at else time.time() + 1800
expires_at = float(data.get("expires_at") or 0) or time.time() + 1800
# Resolve the account-specific API base URL. GitHub advertises the
# authoritative endpoint under ``endpoints.api`` in the exchange response
# (it differs for Copilot Enterprise / proxied accounts). When the
# response omits it, fall back to deriving the host from the ``proxy-ep``
# field embedded in the exchanged token. Individual accounts have neither,
# so ``base_url`` stays None and callers use the registry default.
# Account-specific API base URL: GitHub advertises the authoritative endpoint under
# ``endpoints.api`` (differs for Copilot Enterprise / proxied accounts); when omitted,
# derive the host from the ``proxy-ep`` field embedded in the exchanged token. Individual
# accounts have neither, so ``base_url`` stays None and callers use the registry default.
endpoints = data.get("endpoints")
base_url: Optional[str] = (
str(endpoints.get("api") or "").strip().rstrip("/") if isinstance(endpoints, dict) else ""
@@ -644,44 +562,28 @@ def _exchange_copilot_token_locked(
_jwt_cache[fp] = (api_token, expires_at, base_url)
_save_jwt_to_disk(fp, api_token, expires_at, base_url)
logger.debug(
"Copilot token exchanged, expires_at=%s, base_url=%s",
expires_at,
base_url,
)
logger.debug("Copilot token exchanged, expires_at=%s, base_url=%s", expires_at, base_url)
return api_token, expires_at, base_url
def _derive_base_url_from_proxy_ep(token: str) -> Optional[str]:
"""Derive the Copilot API base URL from a proxy-ep field in the token.
"""Copilot API base URL from the token's ``proxy-ep`` (``proxy.`` host → ``api.``), or None.
The token looks like ``tid=…;exp=…;proxy-ep=proxy.enterprise.githubcopilot.com;…``; the API host
is that value with the leading ``proxy.`` replaced by ``api.``. None if proxy-ep is absent.
The token looks like ``tid=…;exp=…;proxy-ep=proxy.enterprise.githubcopilot.com;…``.
"""
import re
m = re.search(r'(?:^|;)\s*proxy-ep=([^;\s]+)', token)
if not m:
return None
# Strip scheme if present
proxy_ep = re.sub(r"^https?://", "", m.group(1), count=1).rstrip("/")
# Replace leading "proxy." with "api."
if proxy_ep.startswith("proxy."):
proxy_ep = "api." + proxy_ep[len("proxy."):]
proxy_ep = re.sub(r"^proxy\.", "api.", proxy_ep, count=1)
return f"https://{proxy_ep}"
def get_copilot_api_token(raw_token: str) -> tuple[str, Optional[str]]:
"""Exchange a raw GitHub token for a Copilot API token, with fallback.
"""``(api_token, base_url)`` from the exchange, or ``(raw_token, None)`` when it fails.
Convenience wrapper: returns ``(api_token, base_url)`` on success, or ``(raw_token, None)`` if
the exchange fails (e.g. network error, unsupported account type). This preserves existing
behaviour for accounts that don't need exchange while enabling access to internal-only models
for those that do.
``base_url`` is the account-specific API endpoint advertised by the exchange (``endpoints.api``,
with a ``proxy-ep`` fallback), or None for individual accounts.
The fallback preserves behaviour for accounts that don't need exchange (network error,
unsupported account type) while enabling internal-only models for those that do.
"""
if not raw_token:
return raw_token, None
@@ -696,18 +598,13 @@ def get_copilot_api_token(raw_token: str) -> tuple[str, Optional[str]]:
# ─── Copilot API Headers ───────────────────────────────────────────────────
def copilot_request_headers(
*,
is_agent_turn: bool = True,
is_vision: bool = False,
*, is_agent_turn: bool = True, is_vision: bool = False,
) -> dict[str, str]:
"""Build the standard headers for Copilot API requests."""
headers: dict[str, str] = {
"Editor-Version": _EDITOR_VERSION,
"User-Agent": "HermesAgent/1.0",
"Copilot-Integration-Id": "vscode-chat",
"Openai-Intent": "conversation-edits",
"x-initiator": "agent" if is_agent_turn else "user",
}
headers: dict[str, str] = {"Editor-Version": _EDITOR_VERSION, "User-Agent": "HermesAgent/1.0",
"Copilot-Integration-Id": "vscode-chat",
"Openai-Intent": "conversation-edits",
"x-initiator": "agent" if is_agent_turn else "user"}
if is_vision:
headers["Copilot-Vision-Request"] = "true"
+358 -558
View File
File diff suppressed because it is too large Load Diff
+195 -323
View File
@@ -1,29 +1,28 @@
"""Dashboard process-hygiene helpers — extracted from ``hermes_cli/main.py``.
Helpers that STAY in ``hermes_cli.main`` are reached through the lazy ``_m()``
reference so monkeypatches on ``hermes_cli.main.<name>`` keep working and
imports stay one-way (main.py imports this module, never the reverse).
Helpers that STAY in ``hermes_cli.main`` are reached through the lazy ``_m()`` reference so
monkeypatches on ``hermes_cli.main.<name>`` keep working and imports stay one-way.
"""
import contextlib
import os
import subprocess
import sys
from pathlib import Path
# Cmdline substrings identifying the long-lived server. ``hermes serve`` is the
# same server under the headless name the desktop app spawns; it is reaped on
# update for the same frontend/backend-mismatch reason as ``dashboard``.
# Cmdline substrings identifying the long-lived server; ``hermes serve`` is the same server
# under the headless name the desktop app spawns, reaped on update for the same reason.
_DASHBOARD_PATTERNS = tuple(
f"{launcher} {cmd}"
for cmd in ("dashboard", "serve")
for launcher in ("hermes", "hermes_cli.main", "hermes_cli/main.py")
)
_PS_RUN_KWARGS = dict(capture_output=True, text=True, encoding="utf-8", errors="replace")
def _m():
"""Lazy ``hermes_cli.main`` reference (call-time; keeps patches working)."""
from hermes_cli import main
return main
@@ -31,92 +30,74 @@ def _empty_result() -> dict[str, list]:
return {"matched": [], "killed": [], "failed": []}
def _append_row(rows: list[tuple[int, str]], pid_text: str, command: str) -> None:
try:
rows.append((int(pid_text), command))
except ValueError:
pass
def _iter_process_table() -> list[tuple[int, str]]:
"""``(pid, cmdline)`` for every process, via wmic (Windows) or ps. Raises on scan failure."""
rows: list[tuple[int, str]] = []
if sys.platform == "win32":
# errors="ignore": wmic may emit the system code page (a decode error leaves stdout
# None). bounded_probe_run, not run(): run()'s post-timeout cleanup joins pipe readers
# unbounded and a conhost descendant holding duplicated handles wedges it forever.
from hermes_cli._subprocess_compat import bounded_probe_run
result = bounded_probe_run(
["wmic", "process", "get", "ProcessId,CommandLine", "/FORMAT:LIST"],
timeout=10, errors="ignore",
)
if result is None or result.returncode != 0 or result.stdout is None:
return rows
current_cmd = ""
for line in result.stdout.split("\n"):
line = line.strip()
if line.startswith("CommandLine="):
current_cmd = line[len("CommandLine=") :]
elif line.startswith("ProcessId="):
_append_row(rows, line[len("ProcessId=") :], current_cmd)
return rows
# ps (not `pgrep -f "hermes.*dashboard"`) keeps us consistent with gateway._scan_gateway_pids
# and avoids a greedy regex matching unrelated cmdlines that merely contain both words.
result = subprocess.run(["ps", "-A", "-o", "pid=,command="], timeout=10, **_PS_RUN_KWARGS)
if result.returncode == 0:
for line in getattr(result, "stdout", "").split("\n"):
parts = line.strip().split(None, 1)
if len(parts) == 2 and "grep" not in line:
_append_row(rows, parts[0], parts[1])
return rows
def _scan_dashboard_processes(*, exclude_pids: set[int] | None = None) -> list[tuple[int, str]]:
"""Return matching ``dashboard``/``serve`` processes with their cmdlines.
"""``(pid, cmdline)`` of running ``dashboard``/``serve`` processes; empty on any scan error.
``hermes update`` swaps files on disk while a forgotten dashboard keeps the
old Python backend in memory against the new JS bundle — a silent mismatch
(new auth headers → every API call 401s). *exclude_pids* must never be
returned: Hermes Desktop sets ``HERMES_DESKTOP_CHILD_PID`` on the backend
it spawns so an auto-update never kills the backend it manages itself.
Returns an empty list on any scan error (missing ps/wmic, timeout, etc.).
A forgotten dashboard keeps the old Python backend in memory against the new JS bundle
after ``hermes update`` (new auth headers → every API call 401s). *exclude_pids* must never
be returned: Desktop marks the backend it manages via ``HERMES_DESKTOP_CHILD_PID``.
"""
self_pid = os.getpid()
found: list[tuple[int, str]] = []
def _consider(pid: int, command: str) -> None:
if pid != self_pid and any(p in command for p in _DASHBOARD_PATTERNS):
found.append((pid, command))
skip = {self_pid, *(exclude_pids or ())}
try:
if sys.platform == "win32":
# errors="ignore": wmic may emit the system code page; a decode
# error would leave stdout=None. bounded_probe_run (not run()):
# run()'s post-timeout cleanup joins pipe readers unbounded and a
# conhost descendant holding duplicated handles wedges it forever.
# It also passes CREATE_NO_WINDOW for the pythonw.exe backend.
from hermes_cli._subprocess_compat import bounded_probe_run
result = bounded_probe_run(
["wmic", "process", "get", "ProcessId,CommandLine", "/FORMAT:LIST"],
timeout=10,
errors="ignore",
)
if result is None or result.returncode != 0 or result.stdout is None:
return []
current_cmd = ""
for line in result.stdout.split("\n"):
line = line.strip()
if line.startswith("CommandLine="):
current_cmd = line[len("CommandLine=") :]
elif line.startswith("ProcessId="):
try:
_consider(int(line[len("ProcessId=") :]), current_cmd)
except ValueError:
pass
else:
# ps (not `pgrep -f "hermes.*dashboard"`) keeps us consistent with
# gateway._scan_gateway_pids and avoids a greedy regex matching
# unrelated cmdlines that merely contain both words.
result = subprocess.run(
["ps", "-A", "-o", "pid=,command="],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10,
)
if result.returncode == 0:
for line in getattr(result, "stdout", "").split("\n"):
parts = line.strip().split(None, 1)
if len(parts) != 2 or "grep" in line:
continue
try:
_consider(int(parts[0]), parts[1])
except ValueError:
continue
found = [
(pid, cmd) for pid, cmd in _iter_process_table()
if pid not in skip and any(p in cmd for p in _DASHBOARD_PATTERNS)
]
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
return []
if exclude_pids:
found = [proc for proc in found if proc[0] not in exclude_pids]
# Spawn-ledger augmentation: substring patterns miss profiled launches
# (`hermes --profile p serve ...`). Every serve/dashboard registers itself
# in the spawn ledger with live-verified (pid, create_time) — positive
# identity. Add entries the scan missed, preferring the ledger's full argv.
# Spawn-ledger augmentation: substring patterns miss profiled launches (`hermes --profile
# p serve`); the ledger holds live-verified (pid, create_time) — positive identity.
try:
from hermes_cli.process_identity import ledger_entries
seen = {pid for pid, _ in found}
seen = {pid for pid, _ in found} | skip
for entry in ledger_entries():
pid = entry.get("pid")
if (
entry.get("purpose") not in ("serve", "dashboard")
or not isinstance(pid, int)
or pid == self_pid
or pid in seen
or (exclude_pids and pid in exclude_pids)
):
if entry.get("purpose") not in ("serve", "dashboard") or not isinstance(pid, int):
continue
found.append((pid, str(entry.get("argv") or "")))
if pid not in seen:
found.append((pid, str(entry.get("argv") or "")))
except Exception:
pass # ledger unavailable → scan-only behavior
@@ -125,14 +106,10 @@ def _scan_dashboard_processes(*, exclude_pids: set[int] | None = None) -> list[t
def _hermes_home_for_pid(pid: int) -> str | None:
"""Best-effort ``HERMES_HOME`` from *pid*'s environment (psutil, then /proc)."""
try:
with contextlib.suppress(Exception):
import psutil
home = psutil.Process(pid).environ().get("HERMES_HOME")
if home:
if home := psutil.Process(pid).environ().get("HERMES_HOME"):
return home
except Exception:
pass
try:
raw = Path(f"/proc/{pid}/environ").read_bytes()
except OSError:
@@ -163,11 +140,8 @@ def _profile_flag_value(argv: list[str]) -> str | None:
def _is_ephemeral_port_zero_backend(argv: list[str]) -> bool:
"""True for Desktop-style ``serve|dashboard --port 0`` backends.
Ephemeral-port backends are owned by Hermes Desktop (or are PPID-1 orphans
of a prior update respawn). Replaying them after ``hermes update``
multiplies listening backends because ``--port 0`` always binds a fresh
port. Covers both ``serve`` and the legacy ``dashboard --no-open``
fallback older Desktop runtimes use.
Owned by Hermes Desktop (or PPID-1 orphans of a prior respawn); replaying them after
``hermes update`` multiplies listening backends because ``--port 0`` binds a fresh port.
"""
if _dashboard_subcommand_index(argv) is None:
return False
@@ -206,17 +180,16 @@ def _resolved_home(home: str) -> Path:
def _normalized_home_for_compare(home: str) -> str:
"""Install-identity key for *home*: symlinked / differently-spelled roots
compare equal (same normalization as ``home:`` respawn keys)."""
"""Install-identity key for *home*: symlinked / differently-spelled roots compare equal."""
return os.path.normcase(str(_resolved_home(home)))
def _profile_key_for_respawn(argv: list[str], hermes_home: str | None = None) -> str:
"""Stable owner key: ``HERMES_HOME`` when known, else ``--profile`` / ``-p``.
``HERMES_HOME`` ending in ``profiles/<name>`` → ``profile:<name>`` so it
shares a cap with an explicit ``--profile``; other homes keep a resolved
``home:`` key so unrelated installs never collapse together.
``HERMES_HOME`` ending in ``profiles/<name>`` → ``profile:<name>`` so it shares a cap with
an explicit ``--profile``; other homes keep a resolved ``home:`` key so unrelated installs
never collapse together.
"""
if hermes_home:
parts = _resolved_home(hermes_home).parts
@@ -227,30 +200,20 @@ def _profile_key_for_respawn(argv: list[str], hermes_home: str | None = None) ->
def _filter_dashboard_respawn_candidates(
candidates: list[tuple[int, list[str], str | None]],
*,
own_home: str | None = None,
candidates: list[tuple[int, list[str], str | None]], *, own_home: str | None = None
) -> list[list[str]]:
"""Select which killed manual backends to respawn after ``hermes update``.
"""Select which killed manual backends ``(pid, argv, hermes_home)`` to respawn after update.
Candidates are ``(pid, argv, hermes_home)``; *own_home* (default
``get_hermes_home()``) is a parameter so tests can pin it. Rules:
1. Never resurrect Desktop ephemeral ``--port 0`` backends — Desktop owns
their lifecycle; they are the PPID-1 orphans that multiplied across updates.
2. Never replay a backend from a **foreign** ``HERMES_HOME``: the respawn
is argv-only (no ``env=``), so it would come back on the *updating*
install's home and steal the foreign install's fixed port, leaving its
supervisor to crash-loop on ``EADDRINUSE``. Unreadable (``None``) stays eligible.
3. Dedupe by normalized cmdline. 4. At most one backend per profile / home.
Does **not** blanket-skip PPID-1: a prior update respawn detaches with
``start_new_session=True``, so fixed-port manual backends sit under init
and must stay eligible next update.
Rules: never resurrect Desktop ``--port 0`` backends (Desktop owns them); never replay a
backend from a **foreign** ``HERMES_HOME`` (the respawn is argv-only, so it would come back
on the updating install's home and steal the foreign install's fixed port → its supervisor
crash-loops on ``EADDRINUSE``; unreadable ``None`` stays eligible); dedupe by normalized
cmdline; at most one backend per profile / home. PPID-1 is NOT skipped: a prior respawn
detaches with ``start_new_session=True``, so fixed-port manual backends sit under init.
"""
if own_home is None:
try:
from hermes_constants import get_hermes_home
own_home = str(get_hermes_home())
except Exception:
own_home = ""
@@ -277,20 +240,11 @@ def _filter_dashboard_respawn_candidates(
def _exclude_pids_from_env() -> set[int]:
"""PIDs Desktop marks as live backends (``HERMES_DESKTOP_CHILD_PID``).
Desktop may manage several backends (one per active profile) and passes
them comma-separated; a lone int still parses for back-compat.
"""
"""PIDs Desktop marks as live backends (``HERMES_DESKTOP_CHILD_PID``, comma-separated)."""
out: set[int] = set()
for part in os.environ.get("HERMES_DESKTOP_CHILD_PID", "").split(","):
part = part.strip()
if not part:
continue
try:
with contextlib.suppress(ValueError):
out.add(int(part))
except ValueError:
continue
return out
@@ -298,29 +252,25 @@ def _kill_pids_windows(pids: list[int], killed: list[int], failed: list[tuple[in
"""``taskkill /F`` each PID after re-verifying its identity."""
from gateway.status import get_process_start_time
from hermes_cli._subprocess_compat import pid_is_hermes, windows_hide_flags
# Capture identity immediately after discovery: a PID reused before the
# destructive action fails the start-time check.
# Capture identity immediately after discovery: a PID reused before the destructive
# action fails the start-time check.
pid_start_times = {pid: get_process_start_time(pid) for pid in pids}
for pid in pids:
try:
expected_start_time = pid_start_times.get(pid)
if expected_start_time is None:
failed.append((pid, "could not verify process identity"))
continue
if not pid_is_hermes(pid, expected_start_time=expected_start_time):
elif not pid_is_hermes(pid, expected_start_time=expected_start_time):
failed.append((pid, "not hermes-owned or process identity changed"))
continue
result = subprocess.run(
["taskkill", "/PID", str(pid), "/F"],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL,
text=True, encoding="utf-8", errors="replace", timeout=10,
creationflags=windows_hide_flags(),
)
if result.returncode == 0:
killed.append(pid)
else:
failed.append((pid, (result.stderr or result.stdout or "").strip()))
result = subprocess.run(
["taskkill", "/PID", str(pid), "/F"], stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stdin=subprocess.DEVNULL, text=True, encoding="utf-8",
errors="replace", timeout=10, creationflags=windows_hide_flags())
if result.returncode == 0:
killed.append(pid)
else:
failed.append((pid, (result.stderr or result.stdout or "").strip()))
except (FileNotFoundError, subprocess.TimeoutExpired, OSError) as e:
failed.append((pid, str(e)))
@@ -330,6 +280,8 @@ def _kill_pids_posix(pids: list[int], killed: list[int], failed: list[tuple[int,
import signal as _signal
import time as _time
from gateway.status import _pid_exists
def _send(pid: int, sig) -> None:
try:
os.kill(pid, sig)
@@ -348,7 +300,6 @@ def _kill_pids_posix(pids: list[int], killed: list[int], failed: list[tuple[int,
while pending and _time.monotonic() < deadline:
_time.sleep(0.1)
# os.kill(pid, 0) is NOT a no-op on Windows; use the portable check.
from gateway.status import _pid_exists
alive = [p for p in pending if _pid_exists(p)]
killed.extend(p for p in pending if p not in alive)
pending = alive
@@ -358,48 +309,37 @@ def _kill_pids_posix(pids: list[int], killed: list[int], failed: list[tuple[int,
def _kill_stale_dashboard_processes(
reason: str = "the running backend no longer matches the updated frontend",
*,
restart_managed: bool = False,
already_restarted_units: "set[str] | None" = None,
reason: str = "the running backend no longer matches the updated frontend", *,
restart_managed: bool = False, already_restarted_units: "set[str] | None" = None,
) -> dict[str, list]:
"""Kill running ``hermes dashboard`` / ``hermes serve`` processes.
"""Kill running ``hermes dashboard`` / ``hermes serve`` processes (update end, ``--stop``).
Called at the end of ``hermes update`` (default ``reason``) and from
``hermes dashboard --stop``; after an update the running process serves
stale Python against a fresh JS bundle. POSIX: SIGTERM, ~3s grace, SIGKILL
survivors. Windows: ``taskkill /F``.
With ``restart_managed`` (update path only — ``--stop`` never restarts) a
detected ``hermes-dashboard.service`` is restarted through systemd, any
other killed PID owned by a systemd unit has that unit restarted after the
kill (systemd treats our SIGTERM as a clean stop, so ``Restart=on-failure``
never fires), and manual PIDs are respawned from their captured argv.
*already_restarted_units* (no ``.service`` suffix) were restarted by the
caller already; PIDs they own are left untouched, not killed twice.
With ``restart_managed`` (update only) systemd-owned PIDs get their unit restarted after the
kill (systemd treats our SIGTERM as a clean stop, so ``Restart=on-failure`` never fires) and
manual PIDs are respawned from captured argv. PIDs owned by *already_restarted_units* (no
``.service`` suffix) are left untouched, not killed twice.
"""
if restart_managed and _m()._restart_managed_dashboard_service(reason):
# The dashboard unit is handled but every OTHER backend is not (a host
# may also run hermes-serve.service hosting tui_gateway): record the
# unit as handled (the filter below drops PIDs it owns) and keep going.
# The dashboard unit is handled but every OTHER backend is not (a host may also run
# hermes-serve.service hosting tui_gateway): record the unit as handled (the filter
# below drops PIDs it owns) and keep going.
_dash_unit = getattr(_m(), "_DASHBOARD_SYSTEMD_UNIT", "hermes-dashboard.service")
already_restarted_units = set(already_restarted_units or ()) | {
str(_dash_unit).removesuffix(".service")
}
str(_dash_unit).removesuffix(".service")}
exclude = _exclude_pids_from_env()
if restart_managed:
# An SSH-owned backend belongs to an attached Desktop client even when
# the updater runs from an unrelated shell; killing it strands that
# client's fixed SSH port-forward. Same ownership records as the reaper.
# An SSH-owned backend belongs to an attached Desktop client even when the updater
# runs from an unrelated shell; killing it strands that client's fixed SSH
# port-forward. Same ownership records as the reaper.
exclude |= _lock_owned_serve_pids()
pids = _m()._find_stale_dashboard_pids(exclude_pids=exclude or None)
if not pids:
return _empty_result()
# Snapshot systemd cgroup/unit and argv BEFORE killing (the cgroup
# disappears with the process). Linux + update path only.
# Snapshot systemd cgroup/unit and argv BEFORE killing (the cgroup disappears with the
# process). Linux + update path only.
pid_cgroup: dict[int, str | None] = {}
pid_service: dict[int, str | None] = {}
pid_cmdline: dict[int, list[str]] = {}
@@ -408,14 +348,11 @@ def _kill_stale_dashboard_processes(
for pid in pids:
pid_cgroup[pid] = _m()._get_pid_cgroup_path(pid)
pid_service[pid] = _m()._get_systemd_service_for_pid(pid)
if not pid_service[pid]:
# Manual process: keep exact argv + HERMES_HOME for the
# post-update respawn and its per-profile cap.
cmdline = _m()._dashboard_cmdline_for_pid(pid)
if cmdline:
pid_cmdline[pid] = cmdline
pid_home[pid] = _hermes_home_for_pid(pid)
if not pid_service[pid] and (cmdline := _m()._dashboard_cmdline_for_pid(pid)):
# Manual process: keep exact argv + HERMES_HOME for the post-update respawn
# and its per-profile cap.
pid_cmdline[pid] = cmdline
pid_home[pid] = _hermes_home_for_pid(pid)
if already_restarted_units:
pids = [
pid for pid in pids
@@ -428,10 +365,7 @@ def _kill_stale_dashboard_processes(
killed: list[int] = []
failed: list[tuple[int, str]] = []
if sys.platform == "win32":
_kill_pids_windows(pids, killed, failed)
else:
_kill_pids_posix(pids, killed, failed)
(_kill_pids_windows if sys.platform == "win32" else _kill_pids_posix)(pids, killed, failed)
for pid in killed:
print(f" ✓ stopped PID {pid}")
@@ -445,18 +379,18 @@ def _kill_stale_dashboard_processes(
if killed:
print(" Restart the dashboard when you're ready:\n hermes dashboard --port <port>")
return {"matched": list(pids), "killed": list(killed), "failed": list(failed), "unrecovered": list(unrecovered)}
return {"matched": list(pids), "killed": list(killed), "failed": list(failed),
"unrecovered": list(unrecovered)}
def _restart_killed_backends(
killed: list[int], pid_service: dict[int, str | None], pid_cgroup: dict[int, str | None],
pid_cmdline: dict[int, list[str]], pid_home: dict[int, str | None],
) -> list[int]:
"""Update path: restart systemd-owned units, respawn manual argv.
"""Update path: restart systemd-owned units, respawn manual argv. Returns PIDs not brought back.
Respawns are detached, headless, logged to logs/dashboard-restart.log;
Desktop ``--port 0`` backends are filtered out and duplicates collapse to
one per profile. Returns the PIDs that were not brought back.
Respawns are detached, headless, logged to logs/dashboard-restart.log; Desktop ``--port 0``
backends are filtered out and duplicates collapse to one per profile.
"""
unrecovered: list[int] = []
failed_restarts: list[tuple[str, str]] = []
@@ -482,10 +416,9 @@ def _restart_killed_backends(
print(f" ⚠ {svc}: {err}")
respawn_cmds = _filter_dashboard_respawn_candidates(respawn_candidates)
if respawn_cmds:
failed_cmds = _m()._respawn_dashboard_processes(respawn_cmds)
if failed_cmds:
unrecovered.extend(p for p in killed if pid_cmdline.get(p) in failed_cmds)
failed_cmds = _m()._respawn_dashboard_processes(respawn_cmds) if respawn_cmds else None
if failed_cmds:
unrecovered.extend(p for p in killed if pid_cmdline.get(p) in failed_cmds)
if failed_restarts or unrecovered:
print(" Restart anything not auto-restarted when you're ready:\n hermes dashboard --port <port>")
@@ -503,21 +436,13 @@ def _norm_exe(path) -> str:
def _detect_concurrent_hermes_instances(
scripts_dir: Path, *, exclude_pid: int | None = None
) -> list[tuple[int, str]]:
"""Find other live processes whose .exe is one of our entry-point shims.
"""``(pid, name)`` of other live processes whose .exe is one of our entry-point shims.
Windows blocks DELETE/REPLACE on a running .exe (and RENAME when opened
without ``FILE_SHARE_DELETE``); Desktop spawns ``hermes.EXE`` as a backend
child, so the update's quarantine rename fails with ``[WinError 32]``.
Returns ``(pid, process_name)`` for processes whose ``exe`` matches a venv
shim (``hermes.exe`` / ``hermes-gateway.exe``). Excludes our own PID and
every *shim* ancestor: the setuptools launcher is a separate native process
from the ``python.exe`` it loads, so otherwise every update reports its own
launcher. ``proc.parents()`` (whole chain at once) because a per-hop loop
bailed on the first AccessDenied. Only shim ancestors are excluded so a
second hermes.exe under a non-Hermes parent (Desktop child) is still flagged.
Empty off-Windows, without psutil, or with no other instances. Never raises.
Windows blocks DELETE/REPLACE on a running .exe, so a Desktop-spawned ``hermes.EXE`` makes
the update's quarantine rename fail with ``[WinError 32]``. Excludes our PID and every
*shim* ancestor (the setuptools launcher is a separate native process from the
``python.exe`` it loads); ``proc.parents()`` at once because a per-hop loop bailed on the
first AccessDenied. Empty off-Windows / without psutil. Never raises.
"""
if not _m()._is_windows():
return []
@@ -534,44 +459,36 @@ def _detect_concurrent_hermes_instances(
seed = int(exclude_pid) if exclude_pid is not None else os.getpid()
exclude_pids: set[int] = {seed}
# Broad ``except Exception``: psutil may be partially stubbed in tests.
try:
with contextlib.suppress(Exception):
for ancestor in psutil.Process(seed).parents():
try:
with contextlib.suppress(Exception):
anc_exe = ancestor.exe()
if anc_exe and _norm_exe(anc_exe) in shim_paths:
exclude_pids.add(int(ancestor.pid))
except Exception:
continue
except Exception:
pass
matches: list[tuple[int, str]] = []
try:
proc_iter = psutil.process_iter(["pid", "exe", "name"])
except Exception:
return []
for proc in proc_iter:
try:
info = proc.info
except Exception:
continue
pid = info.get("pid")
exe = info.get("exe")
if not exe or pid is None or pid in exclude_pids:
continue
if _norm_exe(exe) in shim_paths:
name = info.get("name") or Path(exe).name
matches.append((int(pid), str(name)))
pid, exe = info.get("pid"), info.get("exe")
if exe and pid is not None and pid not in exclude_pids and _norm_exe(exe) in shim_paths:
matches.append((int(pid), str(info.get("name") or Path(exe).name)))
return matches
def _is_desktop_local_serve_cmdline(command: str) -> bool:
"""True for the Desktop-local serve shape ``hermes serve [--isolated]
--host 127.0.0.1 --port 0``. Long-lived headless serves (``--host
<tailscale-ip> --port 9119``) must never match — those are operator-managed
remote backends that legitimately run with ppid 1 under launchd/nohup."""
"""True for the Desktop-local shape ``hermes serve [--isolated] --host 127.0.0.1 --port 0``.
Long-lived headless serves (``--host <tailscale-ip> --port 9119``) must never match —
those are operator-managed remote backends that legitimately run with ppid 1.
"""
cmd = command.lower()
if "serve" not in cmd or ("hermes" not in cmd and "hermes_cli" not in cmd):
return False
@@ -584,15 +501,11 @@ def _is_desktop_local_serve_cmdline(command: str) -> bool:
def _process_ppid(pid: int) -> int | None:
"""Best-effort parent pid lookup. None on failure (and always on Windows,
where orphan reap is handled by the desktop tree-kill)."""
"""Best-effort parent pid; None on failure (always None on Windows: desktop tree-kill reaps)."""
try:
if sys.platform == "win32":
return None
result = subprocess.run(
["ps", "-o", "ppid=", "-p", str(pid)],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
)
result = subprocess.run(["ps", "-o", "ppid=", "-p", str(pid)], timeout=5, **_PS_RUN_KWARGS)
if result.returncode != 0 or not result.stdout:
return None
return int(result.stdout.strip().split()[0])
@@ -601,13 +514,11 @@ def _process_ppid(pid: int) -> int | None:
# --- SSH remote-backend lock ownership -------------------------------------
# ``backend.lock.json`` is written by the Desktop SSH runtime on the *remote*
# host for every ``hermes serve`` it spawns (apps/desktop/electron/
# remote-lifecycle.ts). A backend another client/machine started is legitimate
# and lock-owned even with no parent here (sshd exited → ppid 1). The reap must
# NEVER kill a PID a valid lock claims — that once killed a production backend.
# Schema constants mirror the writer; a mismatched record is simply ignored
# (the reap only ever *spares*).
# ``backend.lock.json`` is written by the Desktop SSH runtime on the *remote* host for every
# ``hermes serve`` it spawns (apps/desktop/electron/remote-lifecycle.ts). Such a backend is
# legitimate even with no parent here (sshd exited → ppid 1); the reap must NEVER kill a PID a
# valid lock claims — that once killed a production backend. Schema mirrors the writer; a
# mismatched record is ignored (the reap only *spares*).
_LOCKFILE_SCHEMA_VERSION = 2
_PROTOCOL_VERSION = 1
_REMOTE_LOCK_SUBDIR = "desktop-ssh"
@@ -625,11 +536,7 @@ def _is_hex(value: object, length: int) -> bool:
def _valid_lockfile_payload(parsed: object, ownership_id: str) -> bool:
"""Validate a parsed ``backend.lock.json`` body, mirroring readLockfile().
An invalid lock is "no ownership claim", which never causes a kill — the
reap only ever *adds* lock-owned PIDs to its spare-set.
"""
"""Validate a parsed ``backend.lock.json`` body, mirroring readLockfile()."""
if (
not isinstance(parsed, dict)
or parsed.get("schemaVersion") != _LOCKFILE_SCHEMA_VERSION
@@ -639,35 +546,30 @@ def _valid_lockfile_payload(parsed: object, ownership_id: str) -> bool:
or not _is_hex(parsed.get("tokenFingerprint"), 32)
):
return False
pid = parsed.get("pid")
port = parsed.get("port")
if not isinstance(pid, int) or not 0 < pid <= 4194304:
pid, port = parsed.get("pid"), parsed.get("port")
if not (isinstance(pid, int) and 0 < pid <= 4194304):
return False
if not isinstance(port, int) or not 0 <= port <= 65535:
if not (isinstance(port, int) and 0 <= port <= 65535):
return False
# String fields must be present and bounded (the writer enforces <=1024).
for field in ("profile", "hermesPath", "hermesHome", "logPath", "startedAt"):
value = parsed.get(field)
if not isinstance(value, str) or len(value) > 1024:
return False
# logPath is ``{lock_root}/{ownershipId}/{spawnNonce}.log``. Only the
# suffix is checked so a relocated HERMES_HOME doesn't falsely reject a
# legitimate remote-owned backend (a false reject re-introduces the kill).
if any(not isinstance(parsed.get(f), str) or len(parsed[f]) > 1024
for f in ("profile", "hermesPath", "hermesHome", "logPath", "startedAt")):
return False
# logPath is ``{lock_root}/{ownershipId}/{spawnNonce}.log``; only the suffix is checked so a
# relocated HERMES_HOME can't falsely reject a legitimate backend (= re-introduce the kill).
return parsed["logPath"].endswith(f"/{ownership_id}/{parsed['spawnNonce']}.log")
def _lock_owned_serve_pids(base_dir: Path | None = None) -> set[int]:
"""PIDs claimed by valid ``{hermes_home}/desktop-ssh/<ownershipId>/backend.lock.json``
records — legitimately owned (incl. SSH backends other clients started) and
spared by the reap. Best-effort: a bad record contributes no PID; never raises."""
import json
"""PIDs claimed by valid ``{hermes_home}/desktop-ssh/<ownershipId>/backend.lock.json`` records.
Best-effort: a bad record contributes no PID; never raises.
"""
import json
root = base_dir if base_dir is not None else _hermes_home_dir() / _REMOTE_LOCK_SUBDIR
owned: set[int] = set()
if not root.is_dir():
return owned
try:
entries = list(root.iterdir())
entries = list(root.iterdir()) if root.is_dir() else []
except OSError:
return owned
for entry in entries:
@@ -677,26 +579,19 @@ def _lock_owned_serve_pids(base_dir: Path | None = None) -> set[int]:
# Mirror validateOwnershipId(): exactly 32 lowercase hex chars.
if not entry.is_dir() or not _is_hex(ownership_id, 32) or not lock_path.is_file():
continue
with open(lock_path, "rb") as handle:
data = handle.read()
except OSError:
continue
if len(data) > 65536:
continue
try:
data = lock_path.read_bytes()
if len(data) > 65536:
continue
parsed = json.loads(data)
except (UnicodeDecodeError, ValueError):
except (OSError, UnicodeDecodeError, ValueError):
continue
if _valid_lockfile_payload(parsed, ownership_id):
try:
owned.add(int(parsed["pid"]))
except (TypeError, ValueError):
continue
owned.add(parsed["pid"]) # validated as int above
return owned
# Grace window before an orphaned-looking backend may be reaped. Covers the
# gap between process start and the Desktop client writing backend.lock.json.
# Grace window before an orphaned-looking backend may be reaped: covers the gap between
# process start and the Desktop client writing backend.lock.json.
_REAP_MIN_AGE_SECONDS = 180.0
@@ -705,7 +600,6 @@ def _process_age_seconds(pid: int) -> float:
import time as _time
import psutil as _psutil
return max(0.0, _time.time() - _psutil.Process(pid).create_time())
@@ -713,33 +607,23 @@ def _reap_orphaned_desktop_local_serves(
*, reason: str = "orphaned desktop-local hermes serve", signal_term=None, signal_kill=None,
sleep_fn=None, lock_owned_pids_fn=None, process_age_seconds_fn=None,
) -> dict[str, list]:
"""Kill leftover Desktop-local ``hermes serve`` backends with no parent.
"""Kill leftover Desktop-local ``hermes serve`` backends with no parent. Never raises.
When Electron dies uncleanly, ``serve --host 127.0.0.1 --port 0`` children
get reparented to pid 1 with their MCP trees alive; each Desktop boot then
stacks a fresh backend on the corpses until EMFILE. The parent-death
watchdog (HERMES_PARENT_PID) prevents *future* orphans; this clears
*already* orphaned ones when a new Desktop backend starts.
A candidate is reaped only if ALL hold: Desktop-local shape (never a
fixed-port remote serve); ppid 1 (or 0 on some supervisors); not self /
parent / a HERMES_DESKTOP_CHILD_PID; not claimed by a valid
``backend.lock.json`` (SSH backends other clients started legitimately sit
at ppid 1 — killing them is an incident, not cleanup); older than
``_REAP_MIN_AGE_SECONDS`` with a determinable age — Desktop writes the lock
only after HERMES_BACKEND_READY, so during concurrent multi-profile startup
a live sibling is briefly unowned and indistinguishable from a corpse
(mutual-reap storm); a real corpse just waits for a later scan.
Best-effort; failures never raise to the caller.
When Electron dies uncleanly its ``serve --host 127.0.0.1 --port 0`` children are
reparented to pid 1 with their MCP trees alive; each Desktop boot then stacks a fresh
backend on the corpses until EMFILE. Reaped only if ALL hold: Desktop-local shape; ppid
0/1; not self / parent / HERMES_DESKTOP_CHILD_PID; not claimed by a valid
``backend.lock.json`` (SSH backends other clients started legitimately sit at ppid 1);
older than ``_REAP_MIN_AGE_SECONDS`` with a determinable age (Desktop writes the lock only
after HERMES_BACKEND_READY, so a live sibling mid-startup is briefly unowned).
"""
import signal as _signal
import time as _time
signal_term = _signal.SIGTERM if signal_term is None else signal_term
signal_kill = getattr(_signal, "SIGKILL", _signal.SIGTERM) if signal_kill is None else signal_kill
sleep_fn = _time.sleep if sleep_fn is None else sleep_fn
lock_owned_pids_fn = _lock_owned_serve_pids if lock_owned_pids_fn is None else lock_owned_pids_fn
process_age_seconds_fn = _process_age_seconds if process_age_seconds_fn is None else process_age_seconds_fn
sleep_fn = sleep_fn or _time.sleep
lock_owned_pids_fn = lock_owned_pids_fn or _lock_owned_serve_pids
process_age_seconds_fn = process_age_seconds_fn or _process_age_seconds
if sys.platform == "win32":
# Windows desktop uses taskkill tree teardown; orphan scan is POSIX.
@@ -751,13 +635,11 @@ def _reap_orphaned_desktop_local_serves(
except Exception:
return set() # never let lock scanning block or widen the reap
exclude = _exclude_pids_from_env()
exclude.add(os.getpid())
exclude = _exclude_pids_from_env() | {os.getpid()} | _owned_pids()
try:
exclude.add(os.getppid()) # the desktop / sshd wrapper
except Exception:
pass
exclude |= _owned_pids()
try:
scanned = _scan_dashboard_processes(exclude_pids=exclude)
@@ -767,27 +649,23 @@ def _reap_orphaned_desktop_local_serves(
# Re-read ownership: a lock may have been written between scan and now.
owned_now = _owned_pids()
targets: list[tuple[int, str]] = []
for pid, cmd in scanned:
if not _is_desktop_local_serve_cmdline(cmd) or pid in owned_now:
continue
if _process_ppid(pid) not in (0, 1):
continue
try:
if process_age_seconds_fn(pid) < _REAP_MIN_AGE_SECONDS:
continue
def _is_stale_orphan(pid: int) -> bool:
try: # never let a liveness probe failure widen the reap
return process_age_seconds_fn(pid) >= _REAP_MIN_AGE_SECONDS
except Exception:
continue # never let a liveness probe failure widen the reap
targets.append((pid, cmd))
return False
if not targets:
matched = [
pid for pid, cmd in scanned
if _is_desktop_local_serve_cmdline(cmd) and pid not in owned_now
and _process_ppid(pid) in (0, 1) and _is_stale_orphan(pid)
]
if not matched:
return _empty_result()
matched = [pid for pid, _ in targets]
killed: list[int] = []
failed: list[int] = []
for pid, _cmd in targets:
for pid in matched:
try:
os.kill(pid, signal_term)
except ProcessLookupError:
@@ -795,28 +673,22 @@ def _reap_orphaned_desktop_local_serves(
except OSError:
failed.append(pid)
# Brief grace, then SIGKILL survivors. psutil.pid_exists rather than
# os.kill(pid, 0), which is a Windows footgun the linter blocks everywhere.
# Brief grace, then SIGKILL survivors. psutil.pid_exists rather than os.kill(pid, 0),
# which is a Windows footgun the linter blocks everywhere.
sleep_fn(1.5)
import psutil
for pid, _cmd in targets:
for pid in matched:
if pid in failed:
continue
if not psutil.pid_exists(pid):
killed.append(pid)
continue
try:
os.kill(pid, signal_kill)
if psutil.pid_exists(pid):
os.kill(pid, signal_kill)
killed.append(pid)
except ProcessLookupError:
killed.append(pid)
except OSError:
failed.append(pid)
try:
with contextlib.suppress(Exception):
print(f"⟲ Reaped {len(killed)} orphaned desktop-local serve backend(s) ({reason}): {killed or matched}")
except Exception:
pass
return {"matched": matched, "killed": killed, "failed": failed}
+38 -65
View File
@@ -1,11 +1,9 @@
"""``hermes dashboard register`` — register a self-hosted dashboard OAuth client.
Automates the manual flow (Nous Portal ``/local-dashboards`` → "register" →
paste the ``agent:{id}`` client ID into ``~/.hermes/.env``): resolve a fresh
Nous access token from the stored login, POST
``{portal}/api/oauth/self-hosted-client`` (the portal creates a SELF_HOSTED
client in the caller's org; the ``agent:`` prefix is applied server-side),
write ``HERMES_DASHBOARD_OAUTH_CLIENT_ID`` (+ portal/public URL when
Automates the manual Nous Portal ``/local-dashboards`` flow: resolve a fresh Nous access
token from the stored login, POST ``{portal}/api/oauth/self-hosted-client`` (the portal
creates a SELF_HOSTED client in the caller's org; the ``agent:`` prefix is applied
server-side), write ``HERMES_DASHBOARD_OAUTH_CLIENT_ID`` (+ portal/public URL when
warranted) into ``.env`` idempotently, then print the gate-engagement hint.
"""
@@ -18,12 +16,12 @@ import sys
import urllib.error
import urllib.request
from typing import Optional
from urllib.parse import urlparse
_DEFAULT_PORTAL = "https://portal.nousresearch.com"
# Docker-style adjective_noun names (underscore-joined so they drop into a
# label field). The portal has no uniqueness constraint (the row id is the
# key), so collisions are harmless and never retried.
# Docker-style adjective_noun names (underscore-joined so they drop into a label field). The
# portal has no uniqueness constraint (the row id is the key), so collisions are harmless.
_NAME_ADJECTIVES = (
"amber", "bold", "brave", "bright", "calm", "clever", "cosmic", "crisp",
"dreamy", "eager", "electric", "fancy", "gentle", "golden", "happy",
@@ -48,17 +46,12 @@ def _generate_dashboard_name() -> str:
def _resolve_portal_base_url(override: Optional[str] = None) -> str:
"""Portal base URL for the registration request.
Precedence: explicit *override* (``--portal-url`` / env — the token must
have been minted by that same portal), then the ``portal_base_url`` stored
on the Nous login (the issuer, so the correct default), then production.
"""
"""Portal base URL: explicit *override* (the token must have been minted by that same
portal), then the ``portal_base_url`` stored on the Nous login (the issuer), then production."""
if isinstance(override, str) and override.strip():
return override.rstrip("/")
try:
from hermes_cli.auth import DEFAULT_NOUS_PORTAL_URL, get_provider_auth_state
base = (get_provider_auth_state("nous") or {}).get("portal_base_url")
chosen = base if isinstance(base, str) and base.strip() else str(DEFAULT_NOUS_PORTAL_URL)
return chosen.rstrip("/")
@@ -72,14 +65,10 @@ def _register_self_hosted_client(
) -> dict:
"""POST to the portal's self-hosted-client endpoint and return the JSON body.
``existing_client_id`` (persisted from a prior run) makes the portal update
that record in place instead of minting a duplicate — this is what makes
re-running idempotent. The portal falls back to creating a fresh client if
the id no longer resolves in the caller's org, so passing it is always safe.
``name`` is ``None`` on the update path without an explicit ``--name``;
omitting it tells the portal to keep the stored name. Required on create.
``existing_client_id`` (persisted from a prior run) makes the portal update that record in
place instead of minting a duplicate — what makes re-running idempotent; the portal falls
back to creating a fresh client if the id no longer resolves, so passing it is always safe.
``name`` is ``None`` on the update path without ``--name`` (portal keeps the stored name).
Raises RuntimeError with a user-facing message on non-2xx or transport failure.
"""
url = f"{portal_base_url.rstrip('/')}/api/oauth/self-hosted-client"
@@ -96,24 +85,19 @@ def _register_self_hosted_client(
payload = json.loads(resp.read().decode())
except urllib.error.HTTPError as exc:
# Structured JSON errors: {error, error_description}.
detail = ""
try:
err_body = json.loads(exc.read().decode())
detail = err_body.get("error_description") or err_body.get("error") or ""
except Exception:
pass
detail = ""
if exc.code == 401:
raise RuntimeError(
"Nous Portal rejected the access token (401). "
"Try `hermes auth add nous` to re-authenticate."
) from exc
if exc.code == 403:
raise RuntimeError(
detail or "Your account is not permitted to register a self-hosted dashboard."
) from exc
raise RuntimeError(
f"Portal returned HTTP {exc.code}" + (f": {detail}" if detail else "")
) from exc
message = ("Nous Portal rejected the access token (401). "
"Try `hermes auth add nous` to re-authenticate.")
elif exc.code == 403:
message = detail or "Your account is not permitted to register a self-hosted dashboard."
else:
message = f"Portal returned HTTP {exc.code}" + (f": {detail}" if detail else "")
raise RuntimeError(message) from exc
except urllib.error.URLError as exc:
raise RuntimeError(f"Could not reach Nous Portal at {portal_base_url}: {exc.reason}") from exc
@@ -128,9 +112,7 @@ def _print_post_register_hint(
) -> None:
"""Print the success summary + the gate-engagement caveat."""
from hermes_cli.config import get_env_path
print(f"\n Wrote to {get_env_path()}:")
print(" HERMES_DASHBOARD_OAUTH_CLIENT_ID=" + str(client_id))
print(f"\n Wrote to {get_env_path()}:\n HERMES_DASHBOARD_OAUTH_CLIENT_ID={client_id}")
if wrote_portal_url:
print(" HERMES_DASHBOARD_PORTAL_URL=" + str(portal_base_url))
if public_url:
@@ -143,8 +125,6 @@ def _print_post_register_hint(
if custom_redirect_uri:
# Example host matches the one the user registered.
try:
from urllib.parse import urlparse
host = urlparse(custom_redirect_uri).hostname or "your-host"
except Exception:
host = "your-host"
@@ -168,7 +148,6 @@ def _print_post_register_hint(
def _env_value(key: str) -> Optional[str]:
"""Stored ``.env`` value, or ``None`` on any read failure."""
from hermes_cli.config import get_env_value
try:
return get_env_value(key)
except Exception:
@@ -176,9 +155,8 @@ def _env_value(key: str) -> Optional[str]:
def _save_env_quietly(key: str, value: str) -> bool:
"""Persist *key*; False on failure (non-fatal: client_id is load-bearing)."""
"""Persist *key*; False on failure (non-fatal: only client_id is load-bearing)."""
from hermes_cli.config import save_env_value
try:
save_env_value(key, value)
return True
@@ -190,12 +168,10 @@ def _public_url_from_redirect(redirect_uri: Optional[str]) -> str:
"""Origin (``scheme://host[:port]``) of *redirect_uri*, or ``""``.
``dashboard_auth/routes._redirect_uri`` rebuilds the callback as
``HERMES_DASHBOARD_PUBLIC_URL + "/auth/callback"``, so the runtime consumes
the ORIGIN — persisting the raw redirect URI would double up the path.
``HERMES_DASHBOARD_PUBLIC_URL + "/auth/callback"``, so the runtime consumes the ORIGIN —
persisting the raw redirect URI would double up the path.
"""
try:
from urllib.parse import urlparse
parsed = urlparse(redirect_uri or "")
if parsed.scheme in ("http", "https") and parsed.netloc:
return f"{parsed.scheme}://{parsed.netloc}"
@@ -208,9 +184,8 @@ def cmd_dashboard_register(args) -> None:
"""Register a self-hosted dashboard OAuth client with Nous Portal."""
from hermes_cli.auth import AuthError, resolve_nous_access_token
from hermes_cli.config import is_managed, save_env_value
# Managed (Docker/hosted) installs get HERMES_DASHBOARD_OAUTH_CLIENT_ID
# stamped in by the orchestrator; save_env_value refuses to write anyway.
# Managed (Docker/hosted) installs get HERMES_DASHBOARD_OAUTH_CLIENT_ID stamped in by the
# orchestrator; save_env_value refuses to write anyway.
if is_managed():
print("✗ `hermes dashboard register` is not available in a managed/hosted install.\n"
" The dashboard OAuth client is provisioned by the hosting platform.")
@@ -227,21 +202,20 @@ def cmd_dashboard_register(args) -> None:
print(f"✗ Could not resolve a Nous Portal access token: {exc}")
sys.exit(1)
# An *explicitly supplied* portal (flag or env) is an intentional choice we
# persist (overwriting in place); a portal merely inferred from the stored
# login keeps the conservative write-only-if-absent behaviour so .env isn't
# cluttered for the common production case.
# An *explicitly supplied* portal (flag or env) is an intentional choice we persist
# (overwriting in place); a portal merely inferred from the stored login keeps the
# write-only-if-absent behaviour so .env isn't cluttered for the common production case.
portal_override = getattr(args, "portal_url", None) or os.environ.get("HERMES_DASHBOARD_PORTAL_URL")
custom_portal_supplied = bool(isinstance(portal_override, str) and portal_override.strip())
portal_base_url = _resolve_portal_base_url(portal_override)
# Idempotency: re-send a locally held client_id so the portal UPDATES that
# record instead of creating a duplicate (no id = new dashboard).
# Idempotency: re-send a locally held client_id so the portal UPDATES that record instead
# of creating a duplicate (no id = new dashboard).
stored = _env_value("HERMES_DASHBOARD_OAUTH_CLIENT_ID")
existing_client_id = (stored.strip() or None) if isinstance(stored, str) else None
# Auto-generate a name ONLY for a first registration; on a re-run without
# --name leave it unset so the portal preserves the stored name.
# Auto-generate a name ONLY for a first registration; on a re-run without --name leave it
# unset so the portal preserves the stored name.
name = getattr(args, "name", None) or (None if existing_client_id else _generate_dashboard_name())
custom_redirect_uri = getattr(args, "redirect_uri", None)
@@ -269,9 +243,8 @@ def cmd_dashboard_register(args) -> None:
f" Set it manually: HERMES_DASHBOARD_OAUTH_CLIENT_ID={client_id}")
sys.exit(1)
# Portal URL: explicit custom portal → always persist (even when it equals
# production; the user asked). Inferred portal → only when unset AND it
# differs from the production default.
# Portal URL: explicit custom portal → always persist (even when it equals production; the
# user asked). Inferred portal → only when unset AND it differs from the production default.
existing_portal = _env_value("HERMES_DASHBOARD_PORTAL_URL")
should_write_portal = (
existing_portal != portal_base_url
@@ -280,8 +253,8 @@ def cmd_dashboard_register(args) -> None:
)
wrote_portal_url = should_write_portal and _save_env_quietly("HERMES_DASHBOARD_PORTAL_URL", portal_base_url)
# Public URL derived from --redirect-uri: written in place when supplied,
# a no-op when it already matches, never on a localhost-only install.
# Public URL derived from --redirect-uri: written in place when supplied, a no-op when it
# already matches, never on a localhost-only install.
public_url = _public_url_from_redirect(custom_redirect_uri)
wrote_public_url = bool(
public_url
+179 -381
View File
@@ -1,5 +1,6 @@
"""``hermes debug`` debug tools for Hermes Agent."""
import contextlib
import datetime
import gzip
import io
@@ -11,6 +12,7 @@ import time
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from types import SimpleNamespace
from typing import Optional
from hermes_constants import get_hermes_home
@@ -18,9 +20,8 @@ from utils import atomic_replace
logger = logging.getLogger(__name__)
# Banner prepended to upload-bound log content when redaction is enabled.
# Visible in the public paste so reviewers know the content was sanitized.
# Kept short; the trailing newline guarantees the banner sits on its own line.
# Prepended to upload-bound content when redaction is enabled so reviewers of the public
# paste know it was sanitized; trailing newline keeps it on its own line.
_REDACTION_BANNER = (
"[hermes debug share: log content redacted at upload time. "
"run with --no-redact to disable]\n"
@@ -32,48 +33,36 @@ _EMAIL_ADDRESS_RE = re.compile(
r"(?![A-Za-z0-9._%+-])"
)
# ---------------------------------------------------------------------------
# Paste services — try paste.rs first, dpaste.com as fallback.
# ---------------------------------------------------------------------------
# Paste services — paste.rs first, dpaste.com as fallback.
_PASTE_RS_URL = "https://paste.rs/"
_DPASTE_COM_URL = "https://dpaste.com/api/"
_USER_AGENT = "hermes-agent/debug-share"
# Maximum bytes to read from a single log file for upload.
# paste.rs caps at ~1 MB; we stay under that with headroom.
# Max bytes read from one log file for upload (paste.rs caps at ~1 MB; keep headroom).
_MAX_LOG_BYTES = 512_000
# Auto-delete pastes after this many seconds (6 hours).
# Auto-delete pastes after 6 hours.
_AUTO_DELETE_SECONDS = 21600
# ---------------------------------------------------------------------------
# Pending-deletion tracking (replaces the old fork-and-sleep subprocess).
# ---------------------------------------------------------------------------
# ── Pending-deletion tracking ──────────────────────────────────────────────
# Deletion is driven by the gateway's cron ticker (``gateway/run.py::_start_cron_ticker``
# calls ``_sweep_expired_pastes`` hourly); ``hermes debug`` also sweeps opportunistically on
# entry for CLI-only users who never start the gateway. Replaces the old fork-and-sleep
# subprocess that leaked ~20 MB of resident interpreter per share.
def _pending_file() -> Path:
"""Path to ``~/.hermes/pastes/pending.json``.
Deletion is now driven by the gateway's cron ticker (``gateway/run.py::_start_cron_ticker``)
which calls ``_sweep_expired_pastes`` once per hour. ``hermes debug share`` also runs an
opportunistic sweep on entry as a fallback for CLI-only users who never start the gateway.
"""
return get_hermes_home() / "pastes" / "pending.json"
def _load_pending() -> list[dict]:
path = _pending_file()
if not path.exists():
return []
try:
data = json.loads(path.read_text(encoding="utf-8"))
data = json.loads(_pending_file().read_text(encoding="utf-8"))
except (OSError, ValueError):
return []
if not isinstance(data, list):
return []
# Filter to well-formed entries only
return [e for e in data if isinstance(e, dict) and "url" in e and "expire_at" in e]
return [e for e in data if isinstance(e, dict) and {"url", "expire_at"} <= e.keys()]
def _save_pending(entries: list[dict]) -> None:
@@ -84,16 +73,14 @@ def _save_pending(entries: list[dict]) -> None:
tmp.write_text(json.dumps(entries, indent=2), encoding="utf-8")
atomic_replace(tmp, path)
except OSError:
# Non-fatal — worst case the user has to run ``hermes debug delete``
# manually.
pass
pass # non-fatal — worst case the user runs ``hermes debug delete`` manually
def _sweep_expired_pastes(now: Optional[float] = None) -> tuple[int, int]:
"""Synchronously DELETE any pending pastes whose ``expire_at`` has passed.
"""Synchronously DELETE pending pastes whose ``expire_at`` has passed → (deleted, remaining).
Returns ``(deleted, remaining)``. Best-effort: failed deletes stay in the pending file for
the next sweep. Silent, since it runs on every ``hermes debug`` invocation.
Best-effort and silent: failed deletes stay pending for the next sweep, up to 24h past
expiration (paste.rs GCs them eventually).
"""
entries = _load_pending()
if not entries:
@@ -111,22 +98,14 @@ def _sweep_expired_pastes(now: Optional[float] = None) -> tuple[int, int]:
if expire_at > current:
remaining.append(entry)
continue
url = entry.get("url", "")
try:
if delete_paste(url):
deleted += 1
continue
gone = delete_paste(entry.get("url", ""))
except Exception:
# Network hiccup, 404 (already gone), etc. — drop the entry
# after a grace period; don't retry forever.
pass
# Retain failed deletes for up to 24h past expiration, then give up.
if expire_at + 86400 > current:
remaining.append(entry)
gone = False # network hiccup, 404 (already gone), ...
if gone or expire_at + 86400 <= current:
deleted += 1 # deleted, or given up on → count as reaped
else:
deleted += 1 # count as reaped (paste.rs will GC eventually)
remaining.append(entry)
if deleted:
_save_pending(remaining)
@@ -135,16 +114,14 @@ def _sweep_expired_pastes(now: Optional[float] = None) -> tuple[int, int]:
def _best_effort_sweep_expired_pastes() -> None:
"""Attempt pending-paste cleanup without letting /debug fail offline."""
"""Pending-paste cleanup that never lets /debug fail offline."""
try:
_sweep_expired_pastes()
except Exception:
pass
# ---------------------------------------------------------------------------
# Privacy / delete helpers
# ---------------------------------------------------------------------------
# ── Privacy / delete helpers ───────────────────────────────────────────────
_PRIVACY_NOTICE = """\
⚠️ This will upload system info + logs to a PUBLIC paste service.
@@ -172,7 +149,7 @@ _GATEWAY_PRIVACY_NOTICE = (
def _extract_paste_id(url: str) -> Optional[str]:
"""Extract the paste ID from a paste.rs or dpaste.com URL."""
"""Paste ID from a paste.rs URL (dpaste.com pastes have no deletable ID)."""
url = url.strip().rstrip("/")
for prefix in ("https://paste.rs/", "http://paste.rs/"):
if url.startswith(prefix):
@@ -181,40 +158,24 @@ def _extract_paste_id(url: str) -> Optional[str]:
def delete_paste(url: str) -> bool:
"""Delete a paste from paste.rs. Returns True on success.
Only paste.rs supports unauthenticated DELETE. dpaste.com pastes expire automatically but cannot
be deleted via API.
"""
"""Delete a paste.rs paste (the only service with unauthenticated DELETE). True on success."""
paste_id = _extract_paste_id(url)
if not paste_id:
raise ValueError(
f"Cannot delete: only paste.rs URLs are supported. Got: {url}"
)
raise ValueError(f"Cannot delete: only paste.rs URLs are supported. Got: {url}")
target = f"{_PASTE_RS_URL}{paste_id}"
req = urllib.request.Request(
target, method="DELETE",
headers={"User-Agent": "hermes-agent/debug-share"},
)
req = urllib.request.Request(f"{_PASTE_RS_URL}{paste_id}", method="DELETE",
headers={"User-Agent": _USER_AGENT})
with urllib.request.urlopen(req, timeout=30) as resp:
return 200 <= resp.status < 300
def _schedule_auto_delete(urls: list[str], delay_seconds: int = _AUTO_DELETE_SECONDS):
"""Record *urls* for deletion ``delay_seconds`` from now.
Only paste.rs URLs are recorded (dpaste.com auto-expires); entries are merged into any existing
pending.json. The old fork-and-sleep subprocesses leaked ~20 MB of resident interpreter per
``hermes debug share``. This replacement is stateless: the gateway's cron ticker sweeps expired
entries once per hour, ``hermes debug share`` runs an opportunistic sweep as a fallback for
CLI-only users, and paste.rs's own retention policy is the last resort.
"""
"""Record paste.rs *urls* for deletion ``delay_seconds`` from now (merged into pending.json)."""
paste_rs_urls = [u for u in urls if _extract_paste_id(u)]
if not paste_rs_urls:
return
# Dedupe by URL: keep the later expire_at if same URL appears twice
# Dedupe by URL, keeping the later expire_at.
by_url: dict[str, float] = {e["url"]: float(e["expire_at"]) for e in _load_pending()}
expire_at = time.time() + delay_seconds
for u in paste_rs_urls:
@@ -224,10 +185,8 @@ def _schedule_auto_delete(urls: list[str], delay_seconds: int = _AUTO_DELETE_SEC
def _post_paste(service: str, endpoint: str, body: bytes, content_type: str) -> str:
"""POST *body* to a paste service and return the paste URL it echoes back."""
req = urllib.request.Request(
endpoint, data=body, method="POST",
headers={"Content-Type": content_type, "User-Agent": "hermes-agent/debug-share"},
)
req = urllib.request.Request(endpoint, data=body, method="POST",
headers={"Content-Type": content_type, "User-Agent": _USER_AGENT})
with urllib.request.urlopen(req, timeout=30) as resp:
url = resp.read().decode("utf-8").strip()
if not url.startswith("http"):
@@ -236,21 +195,14 @@ def _post_paste(service: str, endpoint: str, body: bytes, content_type: str) ->
def _upload_paste_rs(content: str) -> str:
"""Upload to paste.rs. Returns the paste URL."""
return _post_paste("paste.rs", _PASTE_RS_URL, content.encode("utf-8"), "text/plain; charset=utf-8")
def _upload_dpaste_com(content: str, expiry_days: int = 7) -> str:
"""Upload to dpaste.com. Returns the paste URL."""
boundary = "----HermesDebugBoundary9f3c"
def _field(name: str, value: str) -> str:
return (
f"--{boundary}\r\n"
f'Content-Disposition: form-data; name="{name}"\r\n'
f"\r\n"
f"{value}\r\n"
)
return f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n'
body = (
_field("content", content)
@@ -264,28 +216,18 @@ def _upload_dpaste_com(content: str, expiry_days: int = 7) -> str:
def upload_to_pastebin(content: str, expiry_days: int = 7) -> str:
"""Upload *content* to a paste service, trying paste.rs then dpaste.com."""
errors: list[str] = []
# Try paste.rs first (simple, fast)
try:
return _upload_paste_rs(content)
except Exception as exc:
errors.append(f"paste.rs: {exc}")
# Fallback: dpaste.com (supports expiry)
try:
return _upload_dpaste_com(content, expiry_days=expiry_days)
except Exception as exc:
errors.append(f"dpaste.com: {exc}")
raise RuntimeError(
"Failed to upload to any paste service:\n " + "\n ".join(errors)
)
for service, upload in (
("paste.rs", lambda: _upload_paste_rs(content)),
("dpaste.com", lambda: _upload_dpaste_com(content, expiry_days=expiry_days)),
):
try:
return upload()
except Exception as exc:
errors.append(f"{service}: {exc}")
raise RuntimeError("Failed to upload to any paste service:\n " + "\n ".join(errors))
# ---------------------------------------------------------------------------
# Log file reading
# ---------------------------------------------------------------------------
# ── Log file reading ───────────────────────────────────────────────────────
@dataclass
class LogSnapshot:
@@ -299,16 +241,13 @@ class LogSnapshot:
def _primary_log_path(log_name: str) -> Optional[Path]:
"""Where *log_name* would live if present. Doesn't check existence."""
from hermes_cli.logs import LOG_FILES
filename = LOG_FILES.get(log_name)
return (get_hermes_home() / "logs" / filename) if filename else None
# Logs written by a client process rather than by this backend. When the
# desktop app talks to a remote/docker/SSH backend, `hermes debug share` runs
# on the *backend* and can never see them — a bare "(file not found)" then
# reads as "the app logged nothing" and sends triage down a dead end, which is
# exactly the wrong answer when the client is the thing being debugged.
# Logs written by a client process rather than this backend. When the desktop app talks to a
# remote/docker/SSH backend, `hermes debug share` runs on the *backend* and can never see
# them — a bare "(file not found)" would read as "the app logged nothing" and misdirect triage.
_CLIENT_SIDE_LOGS = {
"desktop": (
"written by Hermes Desktop on the machine running the app, not by this "
@@ -323,65 +262,71 @@ def _missing_log_note(log_name: str) -> str:
reason = _CLIENT_SIDE_LOGS.get(log_name)
if reason is None:
return "(file not found)"
primary = _primary_log_path(log_name)
where = f" — expected at {primary}" if primary else ""
return f"(not on this host: {reason}{where})"
return f"(not on this host: {reason}{f' — expected at {primary}' if primary else ''})"
def _resolve_log_path(log_name: str) -> Optional[Path]:
"""Find the log file for *log_name*, falling back to the .1 rotation.
Returns the first non-empty candidate (primary, then .1), or None. Callers distinguish
'empty primary' from 'truly missing' via :func:`_primary_log_path`.
"""
"""First non-empty candidate for *log_name* (primary, then the .1 rotation), or None."""
primary = _primary_log_path(log_name)
if primary is None:
return None
if primary.exists() and primary.stat().st_size > 0:
return primary
rotated = primary.parent / f"{primary.name}.1"
if rotated.exists() and rotated.stat().st_size > 0:
return rotated
for candidate in (primary, primary.parent / f"{primary.name}.1"):
if candidate.exists() and candidate.stat().st_size > 0:
return candidate
return None
def _redact_log_text(text: str) -> str:
"""Run ``redact_sensitive_text`` with ``force=True`` over upload-bound text.
"""Force-mode ``redact_sensitive_text`` (+ email scrub) over upload-bound text.
Uses ``force=True`` so redaction fires regardless of the operator's ``security.redact_secrets``
setting. The local on-disk log file is not modified; only the in-memory copy headed for the
public paste service is sanitized. Returns the redacted text (or the original when empty / non-
string).
``force=True`` so redaction fires regardless of the operator's ``security.redact_secrets``
setting; only the in-memory copy headed for the paste service is sanitized.
"""
if not text:
return text
from agent.redact import redact_sensitive_text
text = redact_sensitive_text(text, force=True)
return _EMAIL_ADDRESS_RE.sub("[REDACTED_EMAIL]", text)
def _read_tail_bytes(
log_path: Path, size: int, max_bytes: int, tail_lines: int,
) -> tuple[bytes, bool]:
"""Read the whole file, or enough of its tail for both views → (raw, truncated).
For oversized files, read backwards until we have ``max_bytes`` for the standalone upload
AND enough newline context to render the summary tail from the same snapshot.
"""
with open(log_path, "rb") as f:
if size <= max_bytes:
return f.read(), False
chunk_size = 8192
pos = size
chunks: list[bytes] = []
total = 0
newline_count = 0
while (pos > 0 and total < max_bytes * 2
and (total < max_bytes or newline_count <= tail_lines + 1)):
read_size = min(chunk_size, pos)
pos -= read_size
f.seek(pos)
chunk = f.read(read_size)
chunks.insert(0, chunk)
total += len(chunk)
newline_count += chunk.count(b"\n")
chunk_size = min(chunk_size * 2, 65536)
return b"".join(chunks), pos > 0
def _capture_log_snapshot(
log_name: str,
*,
tail_lines: int,
max_bytes: int = _MAX_LOG_BYTES,
redact: bool = True,
log_name: str, *, tail_lines: int, max_bytes: int = _MAX_LOG_BYTES, redact: bool = True,
) -> LogSnapshot:
"""Capture a log once and derive summary/full-log views from it.
"""Capture a log once and derive the summary tail and full-log views from it.
The report tail and standalone log upload must come from the same file snapshot. Otherwise a
rotation/truncate between reads can make the report look newer than the uploaded ``agent.log``
paste.
When ``redact`` is True (the default), both ``tail_text`` and ``full_text`` are run through
``_redact_log_text`` so the snapshot returned is upload-safe. The on-disk log file is never
modified. Pass ``redact=False`` to capture original log content (used by ``hermes debug share
--no-redact``).
Both views must come from the same read: a rotation/truncate between reads would make the
report look newer than the uploaded ``agent.log`` paste. With ``redact`` both texts are
upload-safe; the on-disk file is never modified.
"""
log_path = _resolve_log_path(log_name)
if log_path is None:
@@ -395,40 +340,13 @@ def _capture_log_snapshot(
# race: file was truncated between _resolve_log_path and stat
return LogSnapshot(path=log_path, tail_text="(file empty)", full_text=None)
with open(log_path, "rb") as f:
if size <= max_bytes:
raw = f.read()
truncated = False
else:
# Read from the end until we have enough bytes for the
# standalone upload and enough newline context to render the
# summary tail from the same snapshot.
chunk_size = 8192
pos = size
chunks: list[bytes] = []
total = 0
newline_count = 0
while pos > 0 and (total < max_bytes or newline_count <= tail_lines + 1) and total < max_bytes * 2:
read_size = min(chunk_size, pos)
pos -= read_size
f.seek(pos)
chunk = f.read(read_size)
chunks.insert(0, chunk)
total += len(chunk)
newline_count += chunk.count(b"\n")
chunk_size = min(chunk_size * 2, 65536)
raw = b"".join(chunks)
truncated = pos > 0
raw, truncated = _read_tail_bytes(log_path, size, max_bytes, tail_lines)
full_raw = raw
if truncated and len(full_raw) > max_bytes:
cut = len(full_raw) - max_bytes
# Check whether the cut lands exactly on a line boundary. If the
# byte just before the cut position is a newline the first retained
# byte starts a complete line and we should keep it. Only drop a
# partial first line when we're genuinely mid-line.
# Only drop a partial first line when the cut lands genuinely mid-line (the byte
# before the cut is not a newline).
on_boundary = cut > 0 and full_raw[cut - 1 : cut] == b"\n"
full_raw = full_raw[cut:]
if not on_boundary and b"\n" in full_raw:
@@ -456,52 +374,38 @@ _REPORT_LOGS = ("agent", "errors", "gateway", "gui", "desktop")
_FULL_LOGS = ("agent", "gateway", "gui", "desktop")
def _tail_budget(name: str, log_lines: int) -> int:
return log_lines if name == "agent" else min(log_lines, 100)
def _capture_default_log_snapshots(
log_lines: int, *, redact: bool = True
log_lines: int, *, redact: bool = True,
) -> dict[str, LogSnapshot]:
"""Capture all logs used by debug-share exactly once."""
errors_lines = min(log_lines, 100)
return {
name: _capture_log_snapshot(
name, tail_lines=log_lines if name == "agent" else errors_lines, redact=redact
)
name: _capture_log_snapshot(name, tail_lines=_tail_budget(name, log_lines), redact=redact)
for name in _REPORT_LOGS
}
# ---------------------------------------------------------------------------
# Debug report collection
# ---------------------------------------------------------------------------
# ── Debug report collection ────────────────────────────────────────────────
def _capture_dump() -> str:
"""Run ``hermes dump`` and return its stdout as a string."""
from hermes_cli.dump import run_dump
class _FakeArgs:
show_keys = False
old_stdout = sys.stdout
sys.stdout = capture = io.StringIO()
try:
run_dump(_FakeArgs())
except SystemExit:
pass
finally:
sys.stdout = old_stdout
capture = io.StringIO()
with contextlib.redirect_stdout(capture), contextlib.suppress(SystemExit):
run_dump(SimpleNamespace(show_keys=False))
return capture.getvalue()
def collect_debug_report(
*,
log_lines: int = 200,
dump_text: str = "",
*, log_lines: int = 200, dump_text: str = "",
log_snapshots: Optional[dict[str, LogSnapshot]] = None,
) -> str:
"""Build the summary debug report: system dump + log tails.
"""Build the summary debug report (system dump + log tails) as upload-ready text.
``dump_text`` is pre-captured dump output; when empty, ``hermes dump`` is run internally.
Returns plain text ready for upload.
"""
buf = io.StringIO()
@@ -512,14 +416,11 @@ def collect_debug_report(
if log_snapshots is None:
log_snapshots = _capture_default_log_snapshots(log_lines)
# ── Sanitiser heal counters (#96870) ─────────────────────────────────
# In-process, in-memory counters: populated when this report is built
# inside a process that ran agent turns (gateway /debug share); empty
# from a fresh CLI process, where the errors.log tail below carries the
# same escalation lines instead.
# Sanitiser heal counters: in-process, in-memory — populated when this report is built
# inside a process that ran agent turns (gateway /debug share); empty from a fresh CLI
# process, where the errors.log tail below carries the same escalation lines instead.
try:
from agent.agent_runtime_helpers import get_sanitizer_heal_stats
heal_stats = get_sanitizer_heal_stats()
if heal_stats:
buf.write("\n\n--- transcript sanitiser heal counters ---\n")
@@ -532,82 +433,55 @@ def collect_debug_report(
except Exception:
pass
# ── Recent log tails (summary only) ──────────────────────────────────
errors_lines = min(log_lines, 100)
buf.write("\n")
for name in _REPORT_LOGS:
lines = log_lines if name == "agent" else errors_lines
buf.write(f"\n--- {name}.log (last {lines} lines) ---\n")
buf.write(log_snapshots[name].tail_text)
buf.write("\n")
buf.write(f"\n--- {name}.log (last {_tail_budget(name, log_lines)} lines) ---\n"
f"{log_snapshots[name].tail_text}\n")
return buf.getvalue()
# ---------------------------------------------------------------------------
# Shared bundle collection (used by both the paste.rs and Nous-S3 paths)
# ---------------------------------------------------------------------------
# ── Shared bundle collection (paste.rs and Nous-S3 paths) ──────────────────
# Bundle format identifier embedded in the Nous-S3 JSON envelope. The
# discord-support viewer keys off this string to parse the bundle.
# Bundle format identifier in the Nous-S3 JSON envelope; the discord-support viewer keys off it.
_NOUS_BUNDLE_FORMAT = "hermes-debug-share/1"
def collect_share_bundle(
log_lines: int = 200,
redact: bool = True,
) -> dict[str, str]:
def collect_share_bundle(log_lines: int = 200, redact: bool = True) -> dict[str, str]:
"""Collect the debug report + full logs as a label→text mapping.
The dump header is prepended to each full log (mirroring the historical paste behaviour) so
every file is self-contained, and the redaction banner is prepended when ``redact`` is True.
The dump header is prepended to each full log so every file is self-contained, and the
redaction banner is prepended when ``redact`` is True.
"""
dump_text = _capture_dump()
log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
report = collect_debug_report(
log_lines=log_lines,
dump_text=dump_text,
log_snapshots=log_snapshots,
)
# Visible banner so reviewers know redaction was applied at upload time.
report = collect_debug_report(log_lines=log_lines, dump_text=dump_text,
log_snapshots=log_snapshots)
banner = _REDACTION_BANNER if redact else ""
bundle: dict[str, str] = {"report": banner + report}
# Prepend dump header to each full log so every file is self-contained.
for name in _FULL_LOGS:
full = log_snapshots[name].full_text
if full:
if full := log_snapshots[name].full_text:
bundle[f"{name}.log"] = banner + dump_text + f"\n\n--- full {name}.log ---\n" + full
return bundle
def build_nous_bundle(bundle: dict[str, str], redact: bool = True) -> bytes:
"""Gzip-compress a :func:`collect_share_bundle` mapping into the Nous envelope.
"""Gzip a :func:`collect_share_bundle` mapping into the Nous envelope.
The JSON shape (``format: hermes-debug-share/1``, ``redacted``, ``created``, ``files``) is
what the discord-support viewer parses — keep it stable.
The JSON shape (``format``, ``redacted``, ``created``, ``files``) is what the
discord-support viewer parses — keep it stable.
"""
created = datetime.datetime.now(datetime.timezone.utc).isoformat()
envelope = {
"format": _NOUS_BUNDLE_FORMAT,
"redacted": bool(redact),
"created": created,
"files": bundle,
}
envelope = {"format": _NOUS_BUNDLE_FORMAT, "redacted": bool(redact),
"created": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"files": bundle}
return gzip.compress(json.dumps(envelope).encode("utf-8"))
# ---------------------------------------------------------------------------
# CLI entry points
# ---------------------------------------------------------------------------
# ── CLI entry points ───────────────────────────────────────────────────────
@dataclass
class DebugShareResult:
"""Structured outcome of a ``debug share`` upload.
Returned by :func:`build_debug_share` so non-CLI callers (dashboard web server, gateway) can
render the uploaded paste URLs as real links instead of scraping printed text.
"""
"""Outcome of a ``debug share`` upload, so non-CLI callers can render real links."""
urls: dict # label -> paste URL (e.g. {"Report": "...", "agent.log": "..."})
failures: list # human-readable "label: error" strings for optional uploads
@@ -617,147 +491,97 @@ class DebugShareResult:
def build_debug_share(
*,
log_lines: int = 200,
expiry: int = 7,
redact: bool = True,
*, log_lines: int = 200, expiry: int = 7, redact: bool = True,
) -> DebugShareResult:
"""Collect the debug report + full logs, upload each, return the URLs.
This is the shared core behind ``hermes debug share`` (CLI) and the dashboard ``POST
/api/ops/debug-share`` endpoint. It performs blocking network I/O (paste uploads) — callers
inside an event loop must run it in a worker thread.
Shared core behind ``hermes debug share`` and the dashboard ``POST /api/ops/debug-share``.
Blocking network I/O — callers inside an event loop must run it in a worker thread.
"""
_best_effort_sweep_expired_pastes()
# Collect the report + full logs (force-redacted when redact=True) via the
# shared collector so the paste.rs and Nous-S3 paths build identical,
# identically-redacted bundles. The dump header + redaction banner are
# applied inside collect_share_bundle.
bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
if redact:
logger.info(
"hermes debug share: applied force-mode redaction to log snapshots before upload"
)
report = bundle["report"]
urls: dict[str, str] = {}
failures: list[str] = []
# 1. Summary report (required — raises on failure so callers can fall back)
urls["Report"] = upload_to_pastebin(report, expiry_days=expiry)
# 2-5. Full logs (optional — failures are collected, not raised)
for name in _FULL_LOGS:
label = f"{name}.log"
content = bundle.get(label)
if not content:
# Summary report is required — raises on failure so callers can fall back; full logs are
# optional — failures are collected, not raised.
urls = {"Report": upload_to_pastebin(report, expiry_days=expiry)}
for label, content in bundle.items():
if label == "report":
continue
try:
urls[label] = upload_to_pastebin(content, expiry_days=expiry)
except Exception as exc:
failures.append(f"{label}: {exc}")
# Schedule auto-deletion after 6 hours.
_schedule_auto_delete(list(urls.values()))
return DebugShareResult(
urls=urls,
failures=failures,
redacted=redact,
auto_delete_seconds=_AUTO_DELETE_SECONDS,
report=report,
)
return DebugShareResult(urls=urls, failures=failures, redacted=redact,
auto_delete_seconds=_AUTO_DELETE_SECONDS, report=report)
def _confirm_upload(args) -> bool:
"""Require explicit consent before any debug-share upload.
The privacy notice is printed by the caller. This gates the actual upload: with ``--yes`` (or
``-y``) we proceed unprompted; otherwise we ask an interactive ``[y/N]`` question.
"""
"""Gate the actual upload: ``--yes`` proceeds unprompted, else ask an interactive [y/N]."""
if bool(getattr(args, "yes", False)):
return True
if not sys.stdin.isatty():
print(
"ERROR: Non-interactive mode requires --yes to confirm upload.\n"
" This prevents accidental exposure of personal data.\n"
" Use --local to view the report without uploading.",
file=sys.stderr,
)
print("ERROR: Non-interactive mode requires --yes to confirm upload.\n"
" This prevents accidental exposure of personal data.\n"
" Use --local to view the report without uploading.", file=sys.stderr)
sys.exit(1)
try:
answer = input("Upload debug report? [y/N] ").strip().lower()
except (EOFError, KeyboardInterrupt):
answer = ""
if answer not in ("y", "yes"):
print("Aborted.")
return False
return True
if answer in ("y", "yes"):
return True
print("Aborted.")
return False
def run_debug_share(args):
"""Collect debug report + full logs, upload each, print URLs."""
log_lines = getattr(args, "lines", 200)
expiry = getattr(args, "expire", 7)
local_only = getattr(args, "local", False)
nous = getattr(args, "nous", False)
redact = not getattr(args, "no_redact", False)
if local_only:
# Local-only path never uploads — render the report to stdout and bail
# before any network I/O. Reuses the shared collector so the rendered
# output matches exactly what would be uploaded.
if getattr(args, "local", False):
# Never uploads — render via the shared collector so the output matches exactly what
# would be uploaded, and bail before any network I/O.
_best_effort_sweep_expired_pastes()
print("Collecting debug report...")
bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
print(bundle["report"])
for name in _FULL_LOGS:
body = bundle.get(f"{name}.log")
if body:
print(f"\n\n{'=' * 60}\nFULL {name}.log\n{'=' * 60}\n")
print(body)
for label, body in bundle.items():
if label != "report":
print(f"\n\n{'=' * 60}\nFULL {label}\n{'=' * 60}\n\n{body}")
return
if nous:
if getattr(args, "nous", False):
_run_debug_share_nous(args, log_lines=log_lines, redact=redact)
return
print(_PRIVACY_NOTICE)
if not _confirm_upload(args):
return
print("Collecting debug report...")
print("Uploading...")
print("Collecting debug report...\nUploading...")
try:
result = build_debug_share(
log_lines=log_lines,
expiry=expiry,
redact=redact,
)
result = build_debug_share(log_lines=log_lines, expiry=expiry, redact=redact)
except RuntimeError as exc:
print(f"\nUpload failed: {exc}", file=sys.stderr)
print("\nRun `hermes debug share --local` to print the report instead.\n")
sys.exit(1)
# Print results
label_width = max(len(k) for k in result.urls)
print("\nDebug report uploaded:")
for label, url in result.urls.items():
print(f" {label:<{label_width}} {url}")
if result.failures:
print(f"\n (failed to upload: {', '.join(result.failures)})")
hours = result.auto_delete_seconds // 3600
print(f"\n⏱ Pastes will auto-delete in {hours} hours.")
# Manual delete fallback
print("To delete now: hermes debug delete <url>")
print("\nShare these links with the Hermes team for support.")
print(f"\n⏱ Pastes will auto-delete in {result.auto_delete_seconds // 3600} hours.\n"
"To delete now: hermes debug delete <url>\n"
"\nShare these links with the Hermes team for support.")
_NOUS_PRIVACY_NOTICE = """\
@@ -776,35 +600,21 @@ _NOUS_PRIVACY_NOTICE = """\
def _run_debug_share_nous(args, *, log_lines: int, redact: bool) -> None:
"""Handle ``hermes debug share --nous``: upload the bundle to Nous-S3.
Collects the same force-redacted bundle as the paste path, gzips it into the Nous envelope,
requests a signed URL from NAS, uploads, and prints the private viewer link. On any failure
falls back to a clear error that suggests ``--local``.
"""
"""``hermes debug share --nous``: gzip the same bundle into the Nous envelope → Nous-S3."""
from hermes_cli.diagnostics_upload import share_to_nous
print(_NOUS_PRIVACY_NOTICE)
if not _confirm_upload(args):
return
if not redact:
print(
"⚠️ --no-redact is set: secrets in your logs will NOT be redacted "
"before upload.\n"
)
print("⚠️ --no-redact is set: secrets in your logs will NOT be redacted before upload.\n")
print("Collecting debug report...")
_best_effort_sweep_expired_pastes()
bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
if redact:
logger.info(
"hermes debug share --nous: applied force-mode redaction before upload"
)
blob = build_nous_bundle(bundle, redact=redact)
logger.info("hermes debug share --nous: applied force-mode redaction before upload")
print("Uploading to Nous diagnostics storage...")
try:
res = share_to_nous(blob)
res = share_to_nous(build_nous_bundle(bundle, redact=redact))
except Exception as exc:
print(
f"\nNous upload failed: {exc}\n"
@@ -817,36 +627,27 @@ def _run_debug_share_nous(args, *, log_lines: int, redact: bool) -> None:
sys.exit(1)
view_url = res.get("viewUrl") or res.get("view_url")
print("\nDebug bundle uploaded to Nous (private):")
if view_url:
print(f" View URL {view_url}")
else:
print(f" (no view URL returned; upload id: {res.get('id', '?')})")
expires_at = res.get("expiresAt") or res.get("expires_at")
if expires_at:
print(f"\n⏱ Auto-deletes at {expires_at} (14-day retention).")
else:
print("\n⏱ Auto-deletes after 14 days.")
print("\nDebug bundle uploaded to Nous (private):")
print(f" View URL {view_url}" if view_url
else f" (no view URL returned; upload id: {res.get('id', '?')})")
print(f"\n⏱ Auto-deletes at {expires_at} (14-day retention)." if expires_at
else "\n⏱ Auto-deletes after 14 days.")
print(
"\nShare this private link with the Nous team — only Nous staff "
"(via Google login) can open it."
)
print(
"\nPick up the discussion in:\n"
" GitHub Issues https://github.com/NousResearch/hermes-agent/issues\n"
" Nous Portal Support https://portal.nousresearch.com/help\n"
" Discord https://discord.gg/NousResearch"
)
print("\nShare this private link with the Nous team — only Nous staff "
"(via Google login) can open it.\n"
"\nPick up the discussion in:\n"
" GitHub Issues https://github.com/NousResearch/hermes-agent/issues\n"
" Nous Portal Support https://portal.nousresearch.com/help\n"
" Discord https://discord.gg/NousResearch")
def run_debug_delete(args):
"""Delete one or more paste URLs uploaded by /debug."""
urls = getattr(args, "urls", [])
if not urls:
print("Usage: hermes debug delete <url> [<url> ...]")
print(" Deletes paste.rs pastes uploaded by 'hermes debug share'.")
print("Usage: hermes debug delete <url> [<url> ...]\n"
" Deletes paste.rs pastes uploaded by 'hermes debug share'.")
return
for url in urls:
@@ -864,17 +665,14 @@ def run_debug_delete(args):
def run_debug(args):
"""Route debug subcommands."""
# Opportunistic sweep of expired pastes on every ``hermes debug`` call.
# Replaces the old per-paste sleeping subprocess that used to leak as
# one orphaned Python interpreter per scheduled deletion.
_best_effort_sweep_expired_pastes()
subcmd = getattr(args, "debug_command", None)
if subcmd == "share":
run_debug_share(args)
elif subcmd == "delete":
run_debug_delete(args)
else:
handlers = {"share": run_debug_share, "delete": run_debug_delete}
handler = handlers.get(getattr(args, "debug_command", None))
if handler is None:
print(_DEBUG_USAGE)
else:
handler(args)
_DEBUG_USAGE = """\