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

refactor(hermes_cli): copilot_auth/container_boot/dashboard_register — hoist poll fields, suppress-based cleanup, compact rationale comments (every WHY kept)

Browse Source
This commit is contained in:
Teknium
2026-09-03 00:04:01 -07:00
parent 07617a8f92
commit d70bc4185a
3 changed files with 139 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
hermes_cli/container_boot.py
View File

@@ -16,24 +16,16 @@ from typing import Literal, Sequence
log = logging.getLogger(__name__)
# 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.
# missing) registers the slot down and waits for the user — no crash-loop of a broken gateway
# across `docker restart`. Older installs only have gateway_state; newer lifecycle commands
# persist desired_state separately so a transient runtime state can't erase operator intent.
_AUTOSTART_STATES = frozenset({"running"})
# 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 sub-states of a RUNNING gateway (not an operator stop, not a failed boot). A gateway
# hard-killed in one of them with no `desired_state` would otherwise stay DOWN on every later boot
# (observed: staging stranded at `draining`); map them to `running`, mirroring gateway/run.py.
# `starting` / `startup_failed` are excluded: auto-restarting a mid-boot death is the crash-loop.
_TRANSIENT_RUNNING_STATES = frozenset({"draining", "degraded"})
# 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.
# Container-namespaced state is garbage post-restart (an equal PID is a different process).
_STALE_RUNTIME_FILES = ("gateway.pid", "processes.json")
ReconcileActionLabel = Literal["started", "registered", "skipped"]
@@ -45,9 +37,8 @@ class ReconcileAction:
profile: str
prior_state: str | None
action: ReconcileActionLabel
# "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).
# "clean" / "unclean" (sentinel still says running — SIGKILL/OOM/VM death) / "unknown". Boot
# is the one place that can stamp a violent death into the volume-persisted log.
prior_exit: str = "unknown"
@@ -70,10 +61,8 @@ def reconcile_profile_gateways(
so it is what ``hermes gateway start`` (no ``-p``) targets.
"""
actions: list[ReconcileAction] = []
# 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.
# Under a multiplexing root gateway named slots are still registered 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:
@@ -83,9 +72,7 @@ def reconcile_profile_gateways(
"GATEWAY_MULTIPLEX_PROFILES override if set.", exc_info=True)
multiplex_profiles = is_truthy_value(os.environ.get("GATEWAY_MULTIPLEX_PROFILES"))
# 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.
# A legacy `gateway run` container with no state yet seeds `running` (pre-s6 behavior).
legacy_default_state = _maybe_migrate_legacy_gateway_run_state(
hermes_home, container_argv=container_argv, dry_run=dry_run)
default_prior_state = legacy_default_state or _read_desired_state(hermes_home)
@@ -98,12 +85,10 @@ def reconcile_profile_gateways(
profiles_root = hermes_home / "profiles"
if profiles_root.is_dir():
for entry in sorted(profiles_root.iterdir()):
# 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).
# SOUL.md (seeded by `hermes profile create`) is the "real profile" marker.
if not entry.is_dir() or not (entry / "SOUL.md").exists():
continue
# "default" is reserved for the root profile (above); skip a stray
# ``profiles/default/`` rather than collide on the slot.
# "default" is reserved for the root profile slot above.
if entry.name == "default":
log.warning("profiles/default/ exists — skipping to avoid colliding with the "
"reserved root-profile s6 slot")
@@ -111,13 +96,10 @@ def reconcile_profile_gateways(
prior_state = _read_desired_state(entry)
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(_slot_action(entry.name, entry, prior_state, should_start))
if not dry_run:
_write_reconcile_log(hermes_home, actions)
return actions
@@ -126,22 +108,16 @@ def reconcile_profile_gateways(
def _maybe_migrate_legacy_gateway_run_state(
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 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.
"""
"""Seed root gateway_state for pre-s6 `gateway run` containers (the tini image let users run
the gateway as the container command; post-s6 it would register down and never start)."""
state_file = hermes_home / "gateway_state.json"
if state_file.exists():
return None
if os.environ.get("HERMES_GATEWAY_NO_SUPERVISE", "").lower() in ("1", "true", "yes"):
return None
argv = tuple(container_argv) if container_argv is not None else _read_container_argv()
if not _is_legacy_gateway_run_request(argv):
return None
if not dry_run:
import time
state_file.write_text(json.dumps({
@@ -159,11 +135,8 @@ def _cmdline_argv(cmdline: Path) -> tuple[str, ...]:
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``.
"""
"""Best-effort argv of the container's main program (the one holding ``main-wrapper.sh``):
PID 1 first (s6 v2 ``/init``), then every ``/proc/*/cmdline`` (s6 v3 ``s6-svscan``)."""
def _cmdlines():
yield Path("/proc/1/cmdline")
try:
@@ -190,20 +163,14 @@ def _strip_container_argv_prefix(argv: Sequence[str]) -> list[str]:
owns — rather than peeling tokens positionally (which broke on the s6 v2→v3 bump).
"""
args = list(argv)
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":
# Defensive: an `init` prefix with no wrapper token in argv.
elif args and Path(args[0]).name == "init": # defensive: `init` with no wrapper token
args = args[1:]
# Non-PID-1 entrypoints go through the dispatch shim instead of /init.
if args and args[0].endswith("entrypoint-dispatch.sh"):
if args and args[0].endswith("entrypoint-dispatch.sh"): # non-PID-1 dispatch shim
args = args[1:]
# The wrapper re-execs `hermes <subcommand>`; peel an explicit hermes.
if args and Path(args[0]).name == "hermes":
if args and Path(args[0]).name == "hermes": # the wrapper re-execs `hermes <subcommand>`
args = args[1:]
return args
@@ -223,11 +190,8 @@ def _is_dashboard_container(argv: Sequence[str]) -> bool:
def _read_desired_state(profile_dir: Path) -> str | None:
"""Persisted gateway desired state: ``desired_state`` (operator intent), else ``gateway_state``.
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.
"""
"""Persisted ``desired_state`` (operator intent), else legacy ``gateway_state``; missing or
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
@@ -240,9 +204,7 @@ def _read_desired_state(profile_dir: Path) -> str | None:
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
return "running" if gateway_state in _TRANSIENT_RUNNING_STATES else gateway_state
def _cleanup_stale_runtime_files(profile_dir: Path) -> None:
@@ -269,9 +231,8 @@ 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: 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.
marker (cont-init.d runs before s6-svscan has a control socket). Built in a sibling temp
dir and ``Path.replace``d into place so an interrupted write never leaves a half-built dir.
"""
import shutil
@@ -281,83 +242,61 @@ def _register_service(scandir: Path, profile: str, *, start: bool) -> None:
validate_profile_name(profile)
service_dir = scandir / f"gateway-{profile}"
# 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.
# Dot-prefixed so s6-svscan skips the staging dir: a non-dotted name gets supervised AS ROOT
# by a concurrent rescan, creating a root-owned ``supervise/`` → EACCES in the seed below.
tmp_dir = service_dir.with_name("." + service_dir.name + ".tmp")
if tmp_dir.exists():
shutil.rmtree(tmp_dir, ignore_errors=True)
tmp_dir.mkdir(parents=True)
try:
(tmp_dir / "type").write_text("longrun\n", encoding="utf-8")
# 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.
# Manager's own rendering keeps both registration paths consistent; per-profile env
# comes from the profile's config.yaml, so extra_env is empty.
_write_exec(tmp_dir / "run", S6ServiceManager._render_run_script(profile, extra_env={}))
_write_exec(tmp_dir / "finish", S6ServiceManager._render_finish_script())
# Persistent log rotation.
(tmp_dir / "log").mkdir()
_write_exec(tmp_dir / "log" / "run", S6ServiceManager._render_log_run(profile))
# `down` tells s6-supervise NOT to start on pickup; `hermes -p <profile> gateway start`
# brings it up (→ `s6-svc -u`).
if not start:
if not start: # `hermes -p <profile> gateway start` brings it up later (s6-svc -u)
(tmp_dir / "down").touch()
# 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.
# Pre-create supervise/ with hermes ownership BEFORE publishing so s6-supervise inherits
# it and runtime s6-svc calls as the hermes user won't EACCES.
_seed_supervise_skeleton(tmp_dir)
# 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)
tmp_dir.replace(service_dir) # atomic publish
except Exception:
shutil.rmtree(tmp_dir, ignore_errors=True)
raise
# 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.
# ~3000 lines ≈ a year of daily reboots on a 5-profile container; rotated to .1 when crossed.
_LOG_ROTATE_BYTES = 256 * 1024
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".
"""
"""Append one line per profile to $HERMES_HOME/logs/container-boot.log (rotated to ``.1``) —
a separate greppable file for "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"
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:
# Non-fatal — keep appending rather than lose the entry entirely.
except OSError as exc: # non-fatal — keep appending rather than lose the entry
log.warning("could not rotate %s: %s", log_path, exc)
ts = time.strftime("%Y-%m-%dT%H:%M:%S%z")
with log_path.open("a", encoding="utf-8") as f:
for a in actions:
f.write(
f"{ts} profile={a.profile} prior_state={a.prior_state} "
f"action={a.action} prior_exit={a.prior_exit}\n"
)
f.write(f"{ts} profile={a.profile} prior_state={a.prior_state} "
f"action={a.action} prior_exit={a.prior_exit}\n")
def main() -> int:
"""Entry point invoked from /etc/cont-init.d/02-reconcile-profiles."""
# 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.
# A dashboard-only container must not reconcile: with a shared bind-mounted HERMES_HOME both
# containers race to flock() the same s6-log 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)")
return 0

243
hermes_cli/copilot_auth.py
View File

@@ -1,11 +1,9 @@
"""GitHub Copilot authentication utilities.
Credential search order (matching Copilot CLI behaviour): 1. COPILOT_GITHUB_TOKEN env var 2.
GH_TOKEN env var 3. GITHUB_TOKEN env var 4. gh auth token CLI fallback
"""
"""GitHub Copilot authentication utilities (credential order matches the Copilot CLI:
COPILOT_GITHUB_TOKEN, GH_TOKEN, GITHUB_TOKEN, then ``gh auth token``)."""
from __future__ import annotations
import contextlib
import hashlib
import json
import logging
@@ -24,17 +22,11 @@ from hermes_cli._subprocess_compat import IS_WINDOWS, windows_hide_flags
logger = logging.getLogger(__name__)
# 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.
# VS Code's GitHub App client ID: mints ghu_* tokens exchangeable for Copilot API JWTs (needed for
# internal-only models / enterprise endpoints). The opencode App ID mints gho_* tokens that 404.
COPILOT_OAUTH_CLIENT_ID = "Iv1.b507a08c87ecfe98"
# ghp_ classic PATs are rejected by the Copilot API (gho_ / github_pat_ / ghu_ work).
_CLASSIC_PAT_PREFIX = "ghp_"
# Env var search order (matches Copilot CLI)
_CLASSIC_PAT_PREFIX = "ghp_" # rejected by the Copilot API (gho_ / github_pat_ / ghu_ work)
COPILOT_ENV_VARS = ("COPILOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN")
# Polling constants
_DEVICE_CODE_POLL_INTERVAL = 5 # seconds
_DEVICE_CODE_POLL_SAFETY_MARGIN = 3 # seconds
@@ -44,7 +36,6 @@ def validate_copilot_token(token: str) -> tuple[bool, str]:
token = token.strip()
if not token:
return False, "Empty token"
if token.startswith(_CLASSIC_PAT_PREFIX):
return False, (
"Classic Personal Access Tokens (ghp_*) are not supported by the "
@@ -53,7 +44,6 @@ def validate_copilot_token(token: str) -> tuple[bool, str]:
" → A fine-grained PAT (github_pat_*) with Copilot Requests permission\n"
" → `gh auth login` with the default device code flow (produces gho_* tokens)"
)
return True, "OK"
@@ -72,24 +62,20 @@ def resolve_copilot_token() -> tuple[str, str]:
if valid:
return val, env_var
logger.warning("Token from %s is not supported: %s", env_var, msg)
# 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.
# `gh auth token` fallback ONLY when no Copilot env var was set: an exported GITHUB_TOKEN
# (even a classic PAT) means the user intends *that* token; skipping also avoids a slow
# subprocess (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).")
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}")
return token, "gh auth token"
return "", ""
@@ -103,11 +89,10 @@ def _gh_cli_candidates() -> list[str]:
return candidates
# ``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 auth token`` cache (misses too). With no credential store the probe blocks its full 5s on
# keyring / D-Bus, and provider inventory probes Copilot several times per request — an uncached
# miss made one settings page a 4×5s stall past Desktop's 15s IPC budget. 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
@@ -121,12 +106,10 @@ def _invalidate_gh_cli_token_cache() -> None:
def _try_gh_cli_token() -> Optional[str]:
"""Token from ``gh auth token`` when available; the result (incl. a miss) is cached per TTL."""
global _gh_cli_token_cache
now = time.monotonic()
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)
return token
@@ -135,18 +118,14 @@ def _try_gh_cli_token() -> Optional[str]:
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()
# 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.
# gh must not short-circuit on GITHUB_TOKEN / GH_TOKEN, nor 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")
_popen_kwargs = {"creationflags": windows_hide_flags()} if IS_WINDOWS else {}
host_args = ["--hostname", hostname] if hostname else []
for gh_path in _gh_cli_candidates():
cmd = [gh_path, "auth", "token"]
if hostname:
cmd += ["--hostname", hostname]
cmd = [gh_path, "auth", "token", *host_args]
try:
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8',
errors='replace', timeout=5, env=clean_env,
@@ -159,21 +138,15 @@ def _probe_gh_cli_token() -> Optional[str]:
return None
# ─── OAuth Device Code Flow ────────────────────────────────────────────────
_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",
},
url, data=urllib.parse.urlencode(fields).encode(),
headers={"Accept": "application/json", "User-Agent": "HermesAgent/1.0",
"Content-Type": "application/x-www-form-urlencoded"},
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode())
@@ -184,11 +157,8 @@ def copilot_device_code_login(
) -> Optional[str]:
"""Run the GitHub OAuth device code flow for Copilot."""
domain = host.rstrip("/")
device_code_url = f"https://{domain}/login/device/code"
access_token_url = f"https://{domain}/login/oauth/access_token"
try:
device_data = _post_form(device_code_url,
device_data = _post_form(f"https://{domain}/login/device/code",
{"client_id": COPILOT_OAUTH_CLIENT_ID, "scope": "read:user"}, 15)
except Exception as exc:
logger.error("Failed to initiate device authorization: %s", exc)
@@ -199,38 +169,25 @@ def copilot_device_code_login(
user_code = device_data.get("user_code", "")
device_code = device_data.get("device_code", "")
interval = max(device_data.get("interval", _DEVICE_CODE_POLL_INTERVAL), 1)
if not device_code or not user_code:
print(" ✗ GitHub did not return a device code.")
return None
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)
poll_fields = {"client_id": COPILOT_OAUTH_CLIENT_ID, "device_code": device_code,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code"}
deadline = time.monotonic() + timeout_seconds
while time.monotonic() < deadline:
time.sleep(interval + _DEVICE_CODE_POLL_SAFETY_MARGIN)
try:
result = _post_form(
access_token_url,
{
"client_id": COPILOT_OAUTH_CLIENT_ID,
"device_code": device_code,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
},
10,
)
result = _post_form(f"https://{domain}/login/oauth/access_token", poll_fields, 10)
except Exception:
print(".", end="", flush=True)
continue
if result.get("access_token"):
print(" ✓")
return result["access_token"]
error = result.get("error", "")
if error == "slow_down":
# RFC 8628: add 5 seconds to polling interval (or honor a server-supplied one)
@@ -244,41 +201,32 @@ def copilot_device_code_login(
print("\n" + _DEVICE_CODE_TERMINAL_ERRORS.get(error,
f" ✗ Authorization failed: {error}"))
return None
print("\n ✗ Timed out waiting for authorization.")
return None
# ─── Copilot Token Exchange ────────────────────────────────────────────────
# In-process cache of exchanged Copilot API tokens:
# raw_token_fingerprint -> (api_token, expires_at_epoch, base_url).
# In-process cache: 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
# Token exchange endpoint and headers (matching VS Code / Copilot CLI)
# Exchange endpoint and headers (matching VS Code / Copilot CLI)
_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. 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.
# Transient-failure hardening: gateway startup races network readiness, and a single-shot
# exchange failing there silently degrades to the RAW GitHub token, whose integrator 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: 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).
# it). Without it a permanently-rejected token 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 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.
# Single-flight per fingerprint: concurrent callers (dashboard polls every few seconds) wait on
# the ONE in-flight exchange instead of each spawning a hung resolver thread during a DNS outage.
_exchange_locks: dict[str, threading.Lock] = {}
_exchange_locks_guard = threading.Lock()
@@ -293,8 +241,7 @@ def _exchange_lock_for(fp: str) -> threading.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 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.
# The token itself is rejected — retrying with backoff just blocks the caller.
_EXCHANGE_PERMANENT_HTTP_STATUSES = frozenset({401, 403, 404})
@@ -304,12 +251,8 @@ def _token_fingerprint(raw_token: str) -> str:
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 (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.
"""
"""Bounded read of the on-disk JWT store → dict, or None if missing/unusable (a store over
the 1 MiB cap or non-dict can't balloon memory or get rewritten back out)."""
if not path.exists():
return None
try:
@@ -328,10 +271,8 @@ def _write_jwt_store(path: Path, store: dict) -> None:
"""Atomically write the JWT store (tmp + os.replace), best-effort 0o600."""
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(json.dumps(store), encoding="utf-8")
try:
with contextlib.suppress(Exception):
os.chmod(tmp, 0o600)
except Exception:
pass
os.replace(tmp, path)
@@ -357,17 +298,13 @@ def _with_jwt_store(verb: str, op):
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.
"""
"""Drop any cached exchanged JWT for ``raw_token`` (in-process + on-disk) — the runtime
stale-credential recovery path for ``model_not_available_for_integrator`` 400s."""
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.
# Eviction = "force a fresh exchange": the negative-cache entry must go too.
_exchange_failure_cache.pop(fp, None)
def _evict(path, store):
@@ -405,18 +342,14 @@ def _save_jwt_to_disk(fp: str, api_token: str, expires_at: float, base_url: Opti
_with_jwt_store("persist", _save)
# 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).
# urllib's ``timeout`` only bounds socket ops AFTER DNS; getaddrinfo ignores it, so a networkless
# Windows host can hang for minutes (observed: a 17-minute event-loop stall).
_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. Raises the worker's exception, or TimeoutError on the cap.
"""
"""urlopen() on a daemon thread, abandoned after timeout + _DNS_GRACE_SECONDS so a
DNS/getaddrinfo hang cannot block the caller. Raises the worker's exception or TimeoutError."""
box: dict = {}
abandoned = threading.Event()
@@ -426,12 +359,9 @@ def _urlopen_bounded(req, timeout: float):
except BaseException as exc: # re-raised on the caller's thread
box["exc"] = exc
return
if abandoned.is_set():
# Caller already timed out; nobody will read this response — release its socket.
try:
if abandoned.is_set(): # caller already timed out — release the socket
with contextlib.suppress(Exception):
resp.close()
except Exception:
pass
return
box["resp"] = resp
@@ -440,10 +370,8 @@ def _urlopen_bounded(req, timeout: float):
t.join(timeout + _DNS_GRACE_SECONDS)
if t.is_alive():
abandoned.set()
raise TimeoutError(
"copilot token exchange exceeded hard cap of "
f"{timeout + _DNS_GRACE_SECONDS:.0f}s (DNS/getaddrinfo hang?)"
)
raise TimeoutError("copilot token exchange exceeded hard cap of "
f"{timeout + _DNS_GRACE_SECONDS:.0f}s (DNS/getaddrinfo hang?)")
if "exc" in box:
raise box["exc"]
if "resp" not in box:
@@ -454,9 +382,8 @@ def _urlopen_bounded(req, timeout: float):
def _fetch_exchange_with_retry(req, timeout: float, fp: str) -> dict:
"""GET the exchange with backoff for startup network races; 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.
Permanent rejections (401/403/404) skip the retry loop. Failures populate the negative
cache (long TTL for permanent, short for transient); success clears it.
"""
last_exc: Optional[Exception] = None
permanent_failure = False
@@ -478,12 +405,11 @@ def _fetch_exchange_with_retry(req, timeout: float, fp: str) -> dict:
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
_exchange_failure_cache[fp] = time.time() + (
_EXCHANGE_FAILURE_TTL_PERMANENT_SECONDS if permanent_failure
else _EXCHANGE_FAILURE_TTL_TRANSIENT_SECONDS)
raise ValueError(f"Copilot token exchange failed after {_EXCHANGE_MAX_ATTEMPTS} attempts: "
f"{last_exc}") from last_exc
def _cache_entry_fresh(cached) -> bool:
@@ -501,12 +427,9 @@ def exchange_copilot_token(
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)
cached = _jwt_cache.get(fp) # fast path outside the lock
if _cache_entry_fresh(cached):
return cached
with _exchange_lock_for(fp):
return _exchange_copilot_token_locked(raw_token, fp, timeout=timeout)
@@ -514,52 +437,35 @@ def exchange_copilot_token(
def _exchange_copilot_token_locked(
raw_token: str, fp: str, *, timeout: float,
) -> tuple[str, float, Optional[str]]:
# 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.
# Re-check in-process under the lock (a queued-behind caller may have just exchanged), then
# on-disk: a fresh process may hold a still-valid persisted JWT, avoiding 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 _cache_entry_fresh(cached):
_jwt_cache[fp] = cached
return cached
# Negative cache: 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.
_fail_until = _exchange_failure_cache.get(fp, 0.0)
if time.time() < _fail_until:
raise ValueError(
"Copilot token exchange recently failed; skipping re-attempt "
f"for another {int(_fail_until - time.time())}s"
)
raise ValueError("Copilot token exchange recently failed; skipping re-attempt "
f"for another {int(_fail_until - time.time())}s")
req = urllib.request.Request(
_TOKEN_EXCHANGE_URL,
method="GET",
headers={
"Authorization": f"token {raw_token}",
"User-Agent": _EXCHANGE_USER_AGENT,
"Accept": "application/json",
"Editor-Version": _EDITOR_VERSION,
},
_TOKEN_EXCHANGE_URL, method="GET",
headers={"Authorization": f"token {raw_token}", "User-Agent": _EXCHANGE_USER_AGENT,
"Accept": "application/json", "Editor-Version": _EDITOR_VERSION},
)
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 = float(data.get("expires_at") or 0) or time.time() + 1800
# 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.api`` is authoritative (Copilot Enterprise / proxied accounts); else derive from
# the token's ``proxy-ep``. Individual accounts have neither → None (registry default).
endpoints = data.get("endpoints")
base_url: Optional[str] = (
str(endpoints.get("api") or "").strip().rstrip("/") if isinstance(endpoints, dict) else ""
) or _derive_base_url_from_proxy_ep(api_token)
_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)
@@ -567,10 +473,7 @@ def _exchange_copilot_token_locked(
def _derive_base_url_from_proxy_ep(token: str) -> Optional[str]:
"""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;…``.
"""
"""Copilot API base URL from the token's ``proxy-ep=proxy.<host>`` field (→ ``api.``)."""
m = re.search(r'(?:^|;)\s*proxy-ep=([^;\s]+)', token)
if not m:
return None
@@ -580,11 +483,8 @@ def _derive_base_url_from_proxy_ep(token: str) -> Optional[str]:
def get_copilot_api_token(raw_token: str) -> tuple[str, Optional[str]]:
"""``(api_token, base_url)`` from the exchange, or ``(raw_token, None)`` when it fails.
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.
"""
"""``(api_token, base_url)`` from the exchange, or ``(raw_token, None)`` when it fails
(accounts that don't need exchange keep working)."""
if not raw_token:
return raw_token, None
try:
@@ -595,8 +495,6 @@ def get_copilot_api_token(raw_token: str) -> tuple[str, Optional[str]]:
return raw_token, None
# ─── Copilot API Headers ───────────────────────────────────────────────────
def copilot_request_headers(
*, is_agent_turn: bool = True, is_vision: bool = False,
) -> dict[str, str]:
@@ -607,5 +505,4 @@ def copilot_request_headers(
"x-initiator": "agent" if is_agent_turn else "user"}
if is_vision:
headers["Copilot-Vision-Request"] = "true"
return headers

79
hermes_cli/dashboard_register.py
View File

@@ -1,10 +1,9 @@
"""``hermes dashboard register`` — register a self-hosted dashboard OAuth client.
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.
Automates the Nous Portal ``/local-dashboards`` flow: resolve a fresh Nous access token, POST
``{portal}/api/oauth/self-hosted-client`` (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.
"""
from __future__ import annotations
@@ -20,8 +19,7 @@ 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.
# Docker-style adjective_noun names; the portal keys on row id, so collisions are harmless.
_NAME_ADJECTIVES = (
"amber", "bold", "brave", "bright", "calm", "clever", "cosmic", "crisp",
"dreamy", "eager", "electric", "fancy", "gentle", "golden", "happy",
@@ -46,8 +44,8 @@ def _generate_dashboard_name() -> str:
def _resolve_portal_base_url(override: Optional[str] = None) -> str:
"""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."""
"""Portal base URL: explicit *override* (must be the token's issuer), then the login's stored
``portal_base_url``, then production."""
if isinstance(override, str) and override.strip():
return override.rstrip("/")
try:
@@ -65,27 +63,24 @@ 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 — 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.
``existing_client_id`` makes the portal update that record in place (idempotent re-runs;
the portal mints 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"
fields = (("name", name), ("custom_redirect_uri", custom_redirect_uri), ("client_id", existing_client_id))
body = {k: v for k, v in fields if v}
fields = (("name", name), ("custom_redirect_uri", custom_redirect_uri),
("client_id", existing_client_id))
req = urllib.request.Request(
url, data=json.dumps(body).encode("utf-8"), method="POST",
f"{portal_base_url.rstrip('/')}/api/oauth/self-hosted-client",
data=json.dumps({k: v for k, v in fields if v}).encode("utf-8"), method="POST",
headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json",
"Accept": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
payload = json.loads(resp.read().decode())
except urllib.error.HTTPError as exc:
# Structured JSON errors: {error, error_description}.
try:
try: # structured JSON errors: {error, error_description}
err_body = json.loads(exc.read().decode())
detail = err_body.get("error_description") or err_body.get("error") or ""
except Exception:
@@ -100,7 +95,6 @@ def _register_self_hosted_client(
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
if not isinstance(payload, dict) or not payload.get("client_id"):
raise RuntimeError("Portal returned an unexpected response (no client_id).")
return payload
@@ -123,8 +117,7 @@ def _print_post_register_hint(
" without auth, which is fine for your own machine.\n"
)
if custom_redirect_uri:
# Example host matches the one the user registered.
try:
try: # example host matches the one the user registered
host = urlparse(custom_redirect_uri).hostname or "your-host"
except Exception:
host = "your-host"
@@ -165,12 +158,8 @@ def _save_env_quietly(key: str, value: str) -> bool:
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.
"""
"""Origin (``scheme://host[:port]``) of *redirect_uri*, or ``""`` — the runtime appends
``/auth/callback`` to HERMES_DASHBOARD_PUBLIC_URL, so the raw URI would double the path."""
try:
parsed = urlparse(redirect_uri or "")
if parsed.scheme in ("http", "https") and parsed.netloc:
@@ -184,14 +173,12 @@ 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 installs get the client id stamped in by the orchestrator (save_env_value refuses).
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.")
sys.exit(1)
# 1. Fresh Nous access token (refreshes near expiry).
try:
access_token = resolve_nous_access_token()
except Exception as exc:
@@ -201,25 +188,17 @@ def cmd_dashboard_register(args) -> None:
else:
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
# write-only-if-absent behaviour so .env isn't cluttered for the common production case.
# An explicitly supplied portal (flag or env) is persisted in place; an inferred one is
# written only if absent 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).
# Re-sending a locally held client_id makes the portal UPDATE that record (idempotent).
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-name ONLY a first registration; a re-run without --name keeps 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)
# 2. Register with the portal.
try:
result = _register_self_hosted_client(
access_token=access_token, portal_base_url=portal_base_url, name=name,
@@ -234,17 +213,13 @@ def cmd_dashboard_register(args) -> None:
# The portal echoes back the same client_id when it updated in place.
verb = "Updated" if existing_client_id and client_id == existing_client_id else "Registered"
print(f'✓ {verb} dashboard "{registered_name}"')
# 3. Write env vars. The client_id is always set and is fatal on failure.
try:
try: # client_id is load-bearing: fatal on failure
save_env_value("HERMES_DASHBOARD_OAUTH_CLIENT_ID", client_id)
except Exception as exc:
print(f"✗ Failed to write HERMES_DASHBOARD_OAUTH_CLIENT_ID to .env: {exc}\n"
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.
# Explicit portal → always persist (the user asked); inferred → only if unset AND non-default.
existing_portal = _env_value("HERMES_DASHBOARD_PORTAL_URL")
should_write_portal = (
existing_portal != portal_base_url
@@ -252,17 +227,13 @@ def cmd_dashboard_register(args) -> None:
else not existing_portal and portal_base_url.rstrip("/") != _DEFAULT_PORTAL
)
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 from --redirect-uri: written when supplied and different; never localhost-only.
public_url = _public_url_from_redirect(custom_redirect_uri)
wrote_public_url = bool(
public_url
and _env_value("HERMES_DASHBOARD_PUBLIC_URL") != public_url
and _save_env_quietly("HERMES_DASHBOARD_PUBLIC_URL", public_url)
)
# 4. Hint.
_print_post_register_hint(
client_id=client_id, portal_base_url=portal_base_url, custom_redirect_uri=custom_redirect_uri,
wrote_portal_url=wrote_portal_url, public_url=public_url if wrote_public_url else "",