refactor(hclib): shared roots — hermes_constants, hermes_logging, utils, _subprocess_compat compaction and dedupe

2026-09-02 14:38:33 -07:00
parent 2c7f5d12b5
commit 78335adfec
4 changed files with 901 additions and 2104 deletions
--- a/hermes_cli/_subprocess_compat.py
+++ b/hermes_cli/_subprocess_compat.py
@@ -1,29 +1,12 @@
 """Windows subprocess compatibility helpers.
-Hermes is developed on Linux / macOS and tested natively on Windows too.
+* ``["npm", "install", ...]`` — on Windows ``npm`` is ``npm.cmd``, a batch shim.
-Several common subprocess patterns break silently-or-loudly on Windows:
+``subprocess.Popen(["npm", ...])`` fails with WinError 193 ("not a valid Win32 application") because
 CreateProcessW can't run a ``.cmd`` file without ``shell=True`` or PATHEXT resolution.
-* ``["npm", "install", ...]`` — on Windows ``npm`` is ``npm.cmd``, a batch
+* ``start_new_session=True`` — on POSIX, this maps to ``os.setsid()`` and actually detaches the
-  shim.  ``subprocess.Popen(["npm", ...])`` fails with WinError 193
+child. On Windows it's silently ignored; the Windows equivalent is the ``CREATE_NEW_PROCESS_GROUP |
-  ("not a valid Win32 application") because CreateProcessW can't run a
+CREATE_NO_WINDOW`` creationflags bundle, which Python only applies when you pass it explicitly.
  ``.cmd`` file without ``shell=True`` or PATHEXT resolution.
 * ``start_new_session=True`` — on POSIX, this maps to ``os.setsid()`` and
  actually detaches the child.  On Windows it's silently ignored; the
  Windows equivalent is the ``CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW``
  creationflags bundle, which Python only applies when you pass it
  explicitly.
 * Console-window flashes — every ``subprocess.Popen`` of a ``.exe`` on
  Windows spawns a cmd window briefly unless ``CREATE_NO_WINDOW`` is
  passed.  Cosmetic but jarring for background daemons.
 This module centralizes the platform-branching logic so the rest of the
 codebase doesn't sprinkle ``if sys.platform == "win32":`` everywhere.
 **All helpers are no-ops on non-Windows** — calling them in Linux/macOS
 code paths is safe by design.  That's the "do no damage on POSIX"
 guarantee.
 """
 from __future__ import annotations
@@ -115,20 +98,10 @@ _WINDOWS_GATEWAY_BREAKAWAY_ENV = "_HERMES_GATEWAY_BREAKAWAY"
 def split_command_line(line: str) -> list[str]:
    """Split a user-supplied command line into tokens, Windows-safely.
-    ``shlex.split(line)`` (posix=True) treats every backslash as an escape
+    ``shlex.split`` (posix=True) treats every backslash as an escape, silently mangling Windows
-    character, so Windows paths are silently mangled: ``C:\\Users\\me\\out.txt``
+    paths (the separators vanish, leaving a wrong relative name). On Windows use ``posix=False``
-    becomes ``C:Usersmeout.txt`` — no error, just a wrong path that then
+    and strip one layer of matching double quotes per token; on POSIX this is exactly
-    "succeeds" against a mangled relative filename (#83934) or makes a valid
+    ``shlex.split``. Raises ValueError on unbalanced quotes.
    hook script report "not executable" (#78293).
    On Windows this uses ``posix=False``, which preserves backslashes while
    still honoring double-quoted tokens ("path with spaces"). The trade-off
    is that posix=False keeps surrounding quotes on quoted tokens, so we
    strip one layer of matching double quotes per token — that matches how
    Windows command lines are conventionally parsed. On POSIX the behavior
    is exactly ``shlex.split``.
    Raises ValueError for unbalanced quotes, same as ``shlex.split``.
    """
    if not IS_WINDOWS:
        import shlex
@@ -153,34 +126,13 @@ def split_command_line(line: str) -> list[str]:
 def resolve_node_command(name: str, argv: Sequence[str]) -> list[str]:
    """Resolve a Node-ecosystem command name to an absolute-path argv.
-    On Windows, commands like ``npm``, ``npx``, ``yarn``, ``pnpm``,
+    On Windows, commands like ``npm``, ``npx``, ``yarn``, ``pnpm``, ``playwright``, ``prettier``
-    ``playwright``, ``prettier`` ship as ``.cmd`` files (batch shims).
+    ship as ``.cmd`` files (batch shims). ``subprocess.Popen(["npm", "install"])`` fails with
-    ``subprocess.Popen(["npm", "install"])`` fails with WinError 193
+    WinError 193 because CreateProcessW doesn't execute batch files directly.
    because CreateProcessW doesn't execute batch files directly.
-    ``shutil.which(name)`` *does* resolve ``.cmd`` via PATHEXT and returns
+    ``shutil.which(name)`` *does* resolve ``.cmd`` via PATHEXT and returns the fully-qualified path
-    the fully-qualified path — which CreateProcessW accepts because the
+    — which CreateProcessW accepts because the extension tells Windows to route through ``cmd.exe
-    extension tells Windows to route through ``cmd.exe /c``.
+    /c``.
    On POSIX ``shutil.which`` also returns a fully-qualified path when
    found.  That's a small change from bare-name resolution (the OS does
    its own PATH search) but functionally identical and has the side
    benefit of making the argv reproducible in logs.
    Behavior when the command is not on PATH:
    - On Windows: return the bare name — caller can still try with
      ``shell=True`` as a last resort, OR the subsequent Popen will
      raise FileNotFoundError with a readable error we want to surface.
    - On POSIX: same.  Bare ``npm`` on a Linux box without npm installed
      fails the same way it did before this function existed.
    Args:
        name: The command name to resolve (``npm``, ``npx``, ``node`` …).
        argv: The remaining arguments.  Must NOT include ``name`` itself —
            this function builds the full argv list.
    Returns:
        A list suitable for passing to subprocess.Popen/run/call.
    """
    resolved = shutil.which(name)
    if resolved:
@@ -228,42 +180,14 @@ _CREATE_BREAKAWAY_FROM_JOB = 0x01000000
 def windows_detach_flags() -> int:
-    """Return Win32 creationflags that detach a child from the parent
+    """Return Win32 creationflags detaching a child from the parent console/group; 0 elsewhere.
    console and process group without leaving it console-less.  0 on
    non-Windows.
-    Pair with ``start_new_session=False`` (default) when calling
+    Pair with the default ``start_new_session=False`` (POSIX uses ``start_new_session=True``).
-    subprocess.Popen — on POSIX use ``start_new_session=True`` instead,
+    CREATE_NEW_PROCESS_GROUP stops Ctrl+C propagating; CREATE_NO_WINDOW gives the child a hidden
-    which maps to ``os.setsid()`` in the child.
+    console that descendants inherit, avoiding per-descendant console flashes (DETACHED_PROCESS
-
+    would make CREATE_NO_WINDOW ignored and re-create that bug); CREATE_BREAKAWAY_FROM_JOB escapes
-    Rationale:
+    Electron/Tauri job objects that would otherwise kill the child with the parent. A job that
-    - ``CREATE_NEW_PROCESS_GROUP`` — child has its own process group so
+    forbids breakaway yields PermissionError from Popen -- callers catch OSError and fall back.
      Ctrl+C in the parent console doesn't propagate.
    - ``CREATE_NO_WINDOW`` — the child gets its own fresh console that is
      never shown.  This both detaches it from the parent's console
      lifetime (closing the launching terminal doesn't CTRL_CLOSE it) AND
      gives every console-subsystem descendant (git, gh, cmd, node, …) a
      console to inherit, so they don't allocate visible flashing ones.
      This deliberately replaces the old ``DETACHED_PROCESS`` approach:
      MSDN specifies CREATE_NO_WINDOW is *ignored* when combined with
      DETACHED_PROCESS, and a truly console-less daemon re-creates the
      per-descendant console-flash bug (#54220/#56747) at every spawn —
      see the note on ``_DETACHED_PROCESS`` above.
    - ``CREATE_BREAKAWAY_FROM_JOB`` — escape any job object the parent is
      in.  Electron (Desktop app) and Tauri (bootstrap installer) wrap
      their children in job objects; without breakaway, those children
      die when the parent process exits even though they have their own
      console.  This was the missing flag that made the post-update
      gateway respawn watcher silently die alongside the Tauri updater
      after the Electron Desktop's update flow finished.
    If a process is in a job that disallows breakaway (rare —
    JOB_OBJECT_LIMIT_BREAKAWAY_OK isn't set), CreateProcess returns
    ERROR_ACCESS_DENIED.  Python surfaces that as ``PermissionError``
    on the ``subprocess.Popen`` call.  Callers in this codebase already
    wrap detached spawns in ``try/except OSError`` and fall back to a
    cmd.exe wrapper, so the breakaway-denied case degrades gracefully
    rather than crashing.
    """
    if not IS_WINDOWS:
        return 0
@@ -277,27 +201,8 @@ def windows_detach_flags() -> int:
 def windows_detach_flags_without_breakaway() -> int:
    """Same as :func:`windows_detach_flags` minus ``CREATE_BREAKAWAY_FROM_JOB``.
-    The docstring on :func:`windows_detach_flags` notes that a process in
+    Retry with this when the breakaway variant raises OSError (job disallows breakaway), instead
-    a job which disallows breakaway (no ``JOB_OBJECT_LIMIT_BREAKAWAY_OK``)
+    of hand-coding the bit mask at every site. Returns 0 on non-Windows.
    will see ``ERROR_ACCESS_DENIED`` from CreateProcess, surfacing as
    ``OSError`` (``PermissionError``) on the ``subprocess.Popen`` call.
    Callers that want to recover — by retrying without the breakaway
    bit — can pair the two helpers symbolically rather than coding the
    ``& ~0x01000000`` magic at every site:
    .. code-block:: python
        try:
            subprocess.Popen(argv, creationflags=windows_detach_flags(), …)
        except OSError:
            subprocess.Popen(
                argv,
                creationflags=windows_detach_flags_without_breakaway(),
                …,
            )
    See ``gateway_windows.py::_spawn_detached`` for the canonical
    implementation of this pattern.  Returns 0 on non-Windows.
    """
    if not IS_WINDOWS:
        return 0
@@ -305,19 +210,11 @@ def windows_detach_flags_without_breakaway() -> int:
 def windows_hide_flags() -> int:
-    """Return Win32 creationflags that merely hide the child's console
+    """Return Win32 creationflags hiding the child's console without detaching it; 0 elsewhere.
    window without detaching the child.  0 on non-Windows.
-    Use for short-lived console apps spawned as part of a larger
+    For short-lived helpers (``taskkill``, ``where``, version probes) run synchronously: no
-    operation (``taskkill``, ``where``, version probes) where we want no
+    flash, but the child stays in the parent's process group and job so Ctrl+C and job teardown
-    flash but also want to collect stdout/exit code synchronously.
+    still propagate. Stdio is inherited, so ``capture_output=True`` works.
    The difference from :func:`windows_detach_flags`: no
    ``CREATE_NEW_PROCESS_GROUP`` / ``CREATE_BREAKAWAY_FROM_JOB`` — the
    child stays in the parent's process group and job so Ctrl+C and job
    teardown propagate normally, as a short-lived helper wants.  Stdio
    handles are inherited either way, so ``capture_output=True`` works
    with both bundles.
    """
    if not IS_WINDOWS:
        return 0
@@ -325,25 +222,12 @@ def windows_hide_flags() -> int:
 def suppress_platform_ver_console() -> None:
-    """Stub out ``platform._syscmd_ver`` on Windows so it can never flash a
+    """Stub ``platform._syscmd_ver`` on Windows so it never flashes a console. No-op elsewhere.
    console window.  No-op on non-Windows.
-    CPython's ``platform.win32_ver()`` — reached by ``platform.uname()``,
+    ``platform.win32_ver()`` shells out ``cmd /c ver`` without CREATE_NO_WINDOW, so a windowless
-    ``platform.version()``, and ``platform.platform()`` — unconditionally
+    parent (pythonw gateway, kanban workers) flashes a visible cmd window whenever a dependency
-    shells out ``cmd /c ver`` via ``subprocess.check_output(..., shell=True)``
+    touches ``platform.uname()`` at import. With the stub, ``win32_ver()`` takes its documented
-    with no ``CREATE_NO_WINDOW``.  From a windowless parent (the pythonw
+    fallback to ``sys.getwindowsversion()`` -- same data, in-process. Call before heavy imports.
    gateway and every kanban worker it spawns) that allocates a fresh
    *visible* console: one flashing ``cmd`` window per process, triggered by
    any dependency that merely touches ``platform.uname()`` at import time.
    With ``_syscmd_ver`` stubbed to return its inputs, ``win32_ver()`` hits
    the documented ``ValueError`` fallback and reads the version from
    ``sys.getwindowsversion().platform_version`` — same information, queried
    in-process, no subprocess, no window.  Verified equivalent on
    CPython 3.11 (``platform()`` → ``Windows-10-10.0.xxxxx-SP0`` either way).
    Call early, before heavyweight imports — the flash typically happens
    during a dependency's import, not from Hermes' own code.
    """
    if not IS_WINDOWS:
        return
@@ -362,32 +246,10 @@ def suppress_platform_ver_console() -> None:
 def windows_detach_popen_kwargs() -> dict:
-    """Return a dict of Popen kwargs that detach a child on Windows and
+    """Return Popen kwargs detaching a child on Windows, or ``start_new_session=True`` on POSIX.
    fall back to the POSIX equivalent (``start_new_session=True``) on
    Linux/macOS.
-    Usage pattern:
+    Replaces bare ``start_new_session=True``, which is accepted but has no effect on Windows: the
-
+    child stays attached to the parent console and dies when it closes.
    .. code-block:: python
        subprocess.Popen(
            argv,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            stdin=subprocess.DEVNULL,
            close_fds=True,
            **windows_detach_popen_kwargs(),
        )
    This replaces the unsafe-on-Windows pattern:
    .. code-block:: python
        subprocess.Popen(..., start_new_session=True)
    which silently fails to detach on Windows (the flag is accepted but
    has no effect — the child stays attached to the parent's console
    and dies when the console closes).
    """
    if IS_WINDOWS:
        return {"creationflags": windows_detach_flags()}
@@ -404,38 +266,20 @@ def noninteractive_git_env(
 ) -> dict[str, str]:
    """Environment for *internal* git invocations that must never prompt.
-    Hermes shells out to git from many non-interactive contexts — MCP catalog
+    * ``GIT_TERMINAL_PROMPT=0`` — git fails with "terminal prompts disabled" instead of prompting
-    installs, plugin install/update, profile distribution staging, worktree
+    for credentials. * ``GCM_INTERACTIVE=Never`` — Git Credential Manager (the default credential
-    base fetches, desktop review-pane fetch/push. When the remote is private,
+    helper on Windows installs) never pops its own dialog.
    misconfigured, or requires auth, git's default behavior is to prompt on
    the inherited terminal (or via an askpass helper), which silently hangs
    the operation until its timeout — or forever at call sites without one.
    Ported from openai/codex#34540 / #34612 ("detach non-interactive
    subprocesses from stdin"): a background tool invocation must fail fast
    with a readable error, not wait for input nobody can type.
-    Returns a copy of ``base`` (default ``os.environ``) with:
+    Returns a copy of ``base`` (default ``os.environ``) with ``GIT_TERMINAL_PROMPT=0`` (fail instead
    of prompting), ``GCM_INTERACTIVE=Never`` (no Git Credential Manager dialog), and isolated git
    config: inherited ``GIT_CONFIG_*`` overrides, global/system config, pagers, editors, fsmonitor,
    external diff, and hooks are all disabled for the child so a user's repo/global config cannot
    hang or mutate Hermes's internal plumbing calls.
-    * ``GIT_TERMINAL_PROMPT=0`` — git fails with "terminal prompts disabled"
+    ``GIT_ASKPASS`` / ``SSH_ASKPASS`` are deliberately left alone: when the user has a *working*
-      instead of prompting for credentials.
+    askpass helper or ssh-agent configured, auth should still succeed non-interactively. The env
-    * ``GCM_INTERACTIVE=Never`` — Git Credential Manager (the default
+    only disables paths that block on a human. Pair with ``stdin=subprocess.DEVNULL``. Internal
-      credential helper on Windows installs) never pops its own dialog.
+    plumbing only — the agent-facing terminal tool has its own policy layer and visible PTY.
    * isolated git config — inherited ``GIT_CONFIG_*`` overrides, global/system
      config, pagers, editors, fsmonitor, external diff, and hooks are disabled
      for the child process. A user's repo/global config should not be able to
      hang or mutate Hermes's internal plumbing calls.
    ``GIT_ASKPASS`` / ``SSH_ASKPASS`` are deliberately left alone: when the
    user has a *working* askpass helper or ssh-agent configured, auth should
    still succeed non-interactively. The env only disables paths that block
    on a human.
    Pair with ``stdin=subprocess.DEVNULL`` so git (and any credential helper
    it spawns) also can't read the parent's inherited stdin.
    This is for internal plumbing calls only — the agent-facing terminal tool
    has its own policy layer and user-visible PTY, where prompting can be
    legitimate.
    """
    env = dict(base if base is not None else os.environ)
    env["GIT_TERMINAL_PROMPT"] = "0"
@@ -485,7 +329,6 @@ def noninteractive_git_env(
 # -----------------------------------------------------------------------------
 def _process_start_time(pid: int) -> int | None:
    """Return the repository's stable process-start fingerprint, if available."""
    try:
@@ -499,13 +342,9 @@ def _process_start_time(pid: int) -> int | None:
 def _text_names_hermes(text: str) -> bool:
    """True when *text* names Hermes at a path-segment / token boundary.
-    A bare ``"hermes" in text`` substring test would also match unrelated
+    A bare ``"hermes" in text`` substring test would also match unrelated processes whose paths
-    processes whose paths merely contain the letters (``...\\shermesa\\...``),
+    merely contain the letters (``...\shermesa\...``), which is exactly the false-positive class
-    which is exactly the false-positive class this guard exists to prevent.
+    this guard exists to prevent.
    Instead, split on path separators and whitespace and require a segment
    that *starts with* ``hermes`` (``hermes``, ``hermes.exe``, ``hermes_cli``,
    ``hermes-agent``, ``hermes-runtime``) or the hidden-dir form
    ``.hermes``/``.hermes-runtime``.
    """
    for token in re.split(r"[\\/\s=,;\"']+", text.lower()):
        if token.startswith("hermes") or token.startswith(".hermes"):
@@ -533,13 +372,9 @@ def pid_is_hermes(
 ) -> bool:
    """Return whether it is safe to use ``taskkill`` for *pid*.
-    The PID must be valid, currently exist, and identify a Hermes process. When
+    The PID must be valid, currently exist, and identify a Hermes process. When the caller captured
-    the caller captured a start-time fingerprint before the destructive action,
+    a start-time fingerprint before the destructive action, the live process must still have the
-    the live process must still have the same ``(pid, start_time)`` identity.
+    same ``(pid, start_time)`` identity. Any ambiguity fails closed.
    Any ambiguity fails closed. Non-Windows callers have no ``taskkill`` path,
    so a valid PID with no (or a matching) explicit expectation is accepted
    there — but a caller-provided fingerprint that no longer matches is a
    recycled PID on every platform and is always refused.
    """
    if not isinstance(pid, int) or isinstance(pid, bool) or pid <= 0:
        return False
@@ -571,36 +406,11 @@ def pid_is_hermes(
 def kill_process_tree(proc: "subprocess.Popen") -> None:
    """Best-effort terminate *proc* and its descendants on both platforms.
-    ``proc.kill()`` alone only terminates the direct child. On Windows a
+    ``proc.kill()`` alone only terminates the direct child. On Windows a suspended descendant (e.g.
    suspended descendant (e.g. ``git.exe``) can survive holding duplicates of the
    captured pipe handles, which keeps the pipes from reaching EOF and leaks two
    reader threads + the process per fired timeout — ``taskkill /T /F`` takes the
    whole tree down so the bounded drain that follows can actually reach EOF.
    On POSIX the same class exists: killing the launcher leaves descendants
    (credential helpers, ``git-remote-https``, hook children) running and
    holding the pipe write ends. Callers spawn the child in its own process
    group (``process_group=0``, Python ≥3.11), so when — and only
    when — the child leads its own group (``pgid == pid``), the entire group is
    signalled with ``os.killpg``. The ownership check means a fallback spawn
    that shares our group can never cause us to kill unrelated processes.
    Ported from openai/codex#36793 ("Terminate timed-out Git process trees");
    generalized for the shell-hook runner via openai/codex#37527
    ("Terminate timed-out hook process trees").
-    All failures are swallowed — this is cleanup on an already-failing path, and
+    All failures are swallowed — this is cleanup on an already-failing path, and the caller's
-    the caller's contract is to fail open. ``kill()`` can raise (access denied,
+    contract is to fail open. ``kill()`` can raise (access denied, already reaped); an unhandled
-    already reaped); an unhandled raise here would escape the caller's ``except``
+    raise here would escape the caller's ``except`` handler and break that contract.
    handler and break that contract. The ``taskkill`` spawn itself cannot
    re-enter the deadlock class it fixes: it captures no pipes (DEVNULL), so its
    own timeout cleanup has no reader threads to join.
    Delegates the tree-kill to :func:`agent.deadline.kill_process_tree`
    (#85125 4d) — same taskkill /T /F on Windows and killpg-when-leader on
    POSIX, plus a psutil descendant sweep that also reaches descendants that
    ``setsid``'d into their own sessions. On any import/delegation failure it
    falls back to the original local implementation
    (:func:`_legacy_kill_process_tree`), so the fail-open contract holds even
    in stripped environments.
    """
    try:
        from agent.deadline import kill_process_tree as _deadline_kill_tree
@@ -620,9 +430,8 @@ def kill_process_tree(proc: "subprocess.Popen") -> None:
 def _legacy_kill_process_tree(proc: "subprocess.Popen") -> None:
    """Pre-#85125 local tree-kill — fallback when agent.deadline is unavailable.
-    Kept verbatim so ``kill_process_tree`` can honor its swallow-everything
+    Kept verbatim so ``kill_process_tree`` can honor its swallow-everything contract even when the
-    contract even when the delegation path itself fails (partial install,
+    delegation path itself fails (partial install, import cycle during teardown).
    import cycle during teardown).
    """
    if not IS_WINDOWS:
        # Group-kill first: verify the child actually leads its own process
@@ -667,31 +476,9 @@ def bounded_probe_run(
    errors: str = "replace",
    env: "Mapping[str, str] | None" = None,
 ) -> "subprocess.CompletedProcess[str] | None":
-    """Deadlock-safe ``subprocess.run(argv, capture_output=True, timeout=...)``
+    """Deadlock-safe ``subprocess.run(argv, capture_output=True, timeout=...)`` for fail-open probe
-    for fail-open probe call sites. Returns a ``CompletedProcess`` when the
+    call sites. Returns a ``CompletedProcess`` when the child finished within *timeout* (any exit
-    child finished within *timeout* (any exit code), or ``None`` on spawn
+    code), or ``None`` on spawn failure or timeout.
    failure or timeout.
    Why not ``subprocess.run``: on Windows, ``run()``'s post-timeout cleanup
    calls an *unbounded* ``communicate()`` after killing the direct child.
    Killing it can leave a descendant (``git.exe`` under a launcher shim,
    ``conhost.exe`` under wmic/powershell) holding duplicates of the captured
    stdout/stderr handles, so the pipes never reach EOF and the reader-thread
    join blocks forever. The wmic / ``Get-CimInstance Win32_Process`` gateway
    scan hit exactly this during ``hermes update`` on slow-WMI machines
    (#87134); the git probes hit it first (#68609 / #66037).
    The bounded flow: an explicit ``communicate(timeout)``, then on any
    failure a tree-kill (see :func:`kill_process_tree`) plus a bounded 1s
    post-kill drain; if the pipes are still held after that, they're abandoned
    (the orphaned reader threads are daemonic and cost nothing).
    The spawn contract mirrors the ``run`` calls it replaces: PIPE/PIPE/DEVNULL,
    ``text`` with UTF-8 decoding (*errors* configurable — the process scans use
    ``"ignore"``), and the hidden-window ``creationflags`` on Windows only. On
    POSIX the child is placed in its own process group (``process_group=0``,
    Python ≥3.11) so timeout cleanup can take down descendants with the
    launcher instead of orphaning them.
    """
    _popen_kwargs: dict = {"creationflags": windows_hide_flags()} if IS_WINDOWS else {"process_group": 0}
    try:
@@ -724,50 +511,21 @@ def bounded_probe_run(
 def bounded_git_probe(argv: Sequence[str], *, timeout: float) -> str:
-    """Run a short, throwaway ``git`` probe and return stripped stdout, or ``""``
+    """Run a short ``git`` probe and return stripped stdout, or ``""`` on ANY failure.
    on ANY failure (nonzero exit, timeout, spawn error, decode error).
-    This is the shared, deadlock-safe replacement for
+    Deadlock-safe replacement for ``subprocess.run(["git", ...], timeout=...)`` at fail-open
-    ``subprocess.run(["git", ...], timeout=...)`` at fail-open probe call sites
+    probe sites. On Windows ``run()``'s post-timeout cleanup calls an unbounded ``communicate()``;
-    (``tui_gateway.git_probe.run_git``, ``agent.coding_context._git``).
+    a suspended descendant git.exe holding the pipe handles then blocks forever. Here: bounded
    ``communicate``, then tree-kill plus a 1s drain, then abandon the pipes. Spawn contract
    matches ``run`` byte-for-byte; on POSIX the probe gets its own process group so cleanup
    also takes down credential helpers and remote helpers.
-    **Security (GHSA-7x36-8jrh-v4pw):** these probes run automatically against
+    Security (GHSA-7x36-8jrh-v4pw): these probes run automatically against whatever directory
-    whatever directory the session sits in — the coding-workspace snapshot and
+    the session sits in, before any tool call or trust prompt, and an index refresh executes the
-    the gateway project-tree build fire ``git status`` / ``git branch`` before
+    repo-configured ``core.fsmonitor`` program (hooks/pager/editor/credential helper are sinks
-    any tool call, approval, or trust prompt. An index refresh executes the
+    too). A repo delivered with its ``.git`` intact would get host code execution. Every probe
-    repository-configured ``core.fsmonitor`` program, and other config keys
+    therefore runs under :func:`noninteractive_git_env`; diff-rendering callers additionally pass
-    (hooks, pager, editor, credential helper) are execution sinks too. A repo
+    :data:`NO_DRIVER_DIFF_FLAGS` (attribute-scoped drivers can't be disabled via env).
    delivered as files with its ``.git`` directory intact (a shared zip, sync
    folder, or USB stick — ``git clone`` never transfers ``.git/config``) would
    otherwise get host code execution as the user. Every probe now runs under
    :func:`noninteractive_git_env`, which pins those keys to inert values via
    ``GIT_CONFIG_*`` and ignores global/system config. Diff-rendering callers
    additionally pass :data:`NO_DRIVER_DIFF_FLAGS` (attribute-scoped drivers
    can't be disabled through env overrides).
    Why not ``subprocess.run``: on Windows, ``run()``'s post-timeout cleanup
    calls an *unbounded* ``communicate()`` after killing git. Killing the
    PATH-resolved launcher can leave a suspended descendant ``git.exe`` holding
    duplicates of the captured stdout/stderr handles, so the pipes never reach
    EOF and the reader-thread join blocks forever. On the Desktop agent-build
    path (``_start_agent_build → _session_info → branch() → run_git``) that turned
    an optional branch label into ``agent initialization timed out``
    (issues #68609 / #66037).
    The bounded flow: an explicit ``communicate(timeout)``, then on any failure a
    tree-kill (see :func:`_kill_git_process_tree`) plus a bounded 1s post-kill
    drain; if the pipes are still held after that, they're abandoned (the orphaned
    reader threads are daemonic and cost nothing).
    The normal-path spawn contract mirrors the previous ``run`` call byte-for-byte:
    PIPE/PIPE/DEVNULL, ``text`` with UTF-8 ``errors="replace"`` decoding, and the
    hidden-window ``creationflags`` on Windows only. On POSIX the probe is
    additionally placed in its own process group (``process_group=0``,
    Python ≥3.11) so timeout cleanup can take down descendants — credential
    helpers, ``git-remote-https``, hook children — with the launcher instead of
    orphaning them (see :func:`_kill_git_process_tree`; port of
    openai/codex#36793). ``process_group`` only changes which group the child
    belongs to; it does not detach the terminal or alter the fast path.
    """
    result = bounded_probe_run(argv, timeout=timeout, env=noninteractive_git_env())
    if result is None or result.returncode != 0:
--- a/hermes_constants.py
+++ b/hermes_constants.py
--- a/hermes_logging.py
+++ b/hermes_logging.py
@@ -1,30 +1,12 @@
 """Centralized logging setup for Hermes Agent.
-Provides a single ``setup_logging()`` entry point that both the CLI and
+Log files produced: agent.log — INFO+, all agent/tool/session activity (the main log) errors.log —
-gateway call early in their startup path.  All log files live under
+WARNING+, errors and warnings only (quick triage) gateway.log — INFO+, gateway-only events (created
-``~/.hermes/logs/`` (profile-aware via ``get_hermes_home()``).
+when mode="gateway") gui.log — INFO+, dashboard/websocket/TUI-gateway events (created when
 mode="gui")
-Log files produced:
+All files use ``RotatingFileHandler`` with ``RedactingFormatter`` so secrets are never written to
-    agent.log   — INFO+, all agent/tool/session activity (the main log)
+disk.
    errors.log  — WARNING+, errors and warnings only (quick triage)
    gateway.log — INFO+, gateway-only events (created when mode="gateway")
    gui.log     — INFO+, dashboard/websocket/TUI-gateway events
                  (created when mode="gui")
 All files use ``RotatingFileHandler`` with ``RedactingFormatter`` so
 secrets are never written to disk.
 Component separation:
    gateway.log only receives records from ``gateway.*`` loggers —
    platform adapters, session management, slash commands, delivery.
    gui.log receives dashboard-side records from ``hermes_cli.web_server``,
    ``hermes_cli.pty_bridge``, ``tui_gateway.*``, and ``uvicorn.*``.
    agent.log remains the catch-all (everything goes there).
 Session context:
    Call ``set_session_context(session_id)`` at the start of a conversation
    and ``clear_session_context()`` when done.  All log lines emitted on
    that thread will include ``[session_id]`` for filtering/correlation.
 """
 import atexit
@@ -69,7 +51,7 @@ else:
    from logging.handlers import RotatingFileHandler  # noqa: E402
-from hermes_constants import get_config_path, get_hermes_home
+from hermes_constants import get_config_path, get_hermes_home, mkdir_under_hermes_home
 # Sentinel to track whether setup_logging() has already run.  The function
 # is idempotent — calling it twice is safe but the second call is a no-op
@@ -89,12 +71,8 @@ _LOG_FORMAT_VERBOSE = "%(asctime)s - %(name)s - %(levelname)s%(session_tag)s - %
 def _safe_stderr():  # type: ignore[return]
    """Return a stderr stream that tolerates Unicode on all platforms.
-    On Windows the console encoding is often a legacy MBCS codec
+    We wrap ``sys.stderr`` in a ``TextIOWrapper`` with ``errors='replace'`` so log lines are never
-    (cp949, cp1252, …) that raises ``UnicodeEncodeError`` for characters
+    lost — un-encodable characters are replaced with ``?`` instead of crashing the process.
    like the em-dash (U+2014).  We wrap ``sys.stderr`` in a
    ``TextIOWrapper`` with ``errors='replace'`` so log lines are never
    lost — un-encodable characters are replaced with ``?`` instead of
    crashing the process.
    """
    stream = sys.stderr
    encoding = getattr(stream, "encoding", None) or "utf-8"
@@ -102,72 +80,47 @@ def _safe_stderr():  # type: ignore[return]
    if encoding.lower().replace("-", "") in ("utf8", "utf8surrogateescape"):
        return stream
    try:
-        buf = getattr(stream, "buffer", None)
+        wrapped = io.TextIOWrapper(stream.buffer, encoding="utf-8", errors="replace", line_buffering=True)
-        if buf is not None:
+        # Prevent the wrapper from closing the underlying buffer when it is garbage-collected.
-            wrapped = io.TextIOWrapper(
+        wrapped.close = lambda: None  # type: ignore[assignment]
-                buf,
+        return wrapped
                encoding="utf-8",
                errors="replace",
                line_buffering=True,
            )
            # Prevent the wrapper from closing the underlying buffer
            # when it is garbage-collected.
            wrapped.close = lambda: None  # type: ignore[assignment]
            return wrapped
    except Exception:
-        pass
+        return stream  # best-effort: no buffer / wrapping failed -> original stream
    # Best-effort: if wrapping fails, return the original stream.
    return stream
 _CONCURRENT_LOG_LOCK_TIMEOUT = "Cannot acquire lock after 20 attempts"
 def _is_windows_concurrent_log_lock_timeout(exc: BaseException | None) -> bool:
    """Return True for concurrent-log-handler's Windows lock timeout.
-    On Windows Desktop, slash-command workers and the gateway can all write to
+    On Windows Desktop, slash-command workers and the gateway can all write to the same rotating log
-    the same rotating log files. ``concurrent-log-handler`` serializes rollover
+    files. ``concurrent-log-handler`` serializes rollover with a cross-process lock, but when
-    with a cross-process lock, but when another process holds that lock too
+    another process holds that lock too long it raises this RuntimeError. Logging failures should
-    long it raises this RuntimeError. Logging failures should not escape into
+    not escape into Desktop chat output.
    Desktop chat output.
    """
    return (
        sys.platform == "win32"
        and isinstance(exc, RuntimeError)
-        and _CONCURRENT_LOG_LOCK_TIMEOUT in str(exc)
+        and "Cannot acquire lock after 20 attempts" in str(exc)
    )
 # Third-party loggers that are noisy at DEBUG/INFO level.
 _NOISY_LOGGERS = (
-    "openai",
+    "openai", "openai._base_client", "httpx", "httpcore", "asyncio", "hpack", "hpack.hpack",
-    "openai._base_client",
+    "grpc", "modal", "urllib3", "urllib3.connectionpool", "websockets", "charset_normalizer",
    "httpx",
    "httpcore",
    "asyncio",
    "hpack",
    "hpack.hpack",
    "grpc",
    "modal",
    "urllib3",
    "urllib3.connectionpool",
    "websockets",
    "charset_normalizer",
    "markdown_it",
 )
-# ---------------------------------------------------------------------------
+def _quiet_noisy_loggers() -> None:
    """Pin noisy third-party loggers at WARNING."""
    for name in _NOISY_LOGGERS:
        logging.getLogger(name).setLevel(logging.WARNING)
 # Public session context API
 # ---------------------------------------------------------------------------
 def set_session_context(session_id: str) -> None:
-    """Set the session ID for the current thread.
+    """Set the session ID for the current thread."""
    All subsequent log records on this thread will include ``[session_id]``
    in the formatted output.  Call at the start of ``run_conversation()``.
    """
    _session_context.session_id = session_id
@@ -176,22 +129,14 @@ def clear_session_context() -> None:
    _session_context.session_id = None
 # ---------------------------------------------------------------------------
 # Record factory — injects session_tag into every LogRecord at creation
 # ---------------------------------------------------------------------------
 def _install_session_record_factory() -> None:
    """Replace the global LogRecord factory with one that adds ``session_tag``.
-    Unlike a ``logging.Filter`` on a handler or logger, the record factory
+    Unlike a handler/logger ``Filter``, the record factory runs for EVERY record in the process,
-    runs for EVERY record in the process — including records that propagate
+    including propagated and third-party-handled ones, so ``%(session_tag)s`` is always available
-    from child loggers and records handled by third-party handlers.  This
+    and never KeyErrors. Idempotent: a marker attribute prevents double-wrapping on reload.
    guarantees ``%(session_tag)s`` is always available in format strings,
    eliminating the KeyError that would occur if a handler used our format
    without having a ``_SessionFilter`` attached.
    Idempotent — checks for a marker attribute to avoid double-wrapping if
    the module is reloaded.
    """
    current_factory = logging.getLogRecordFactory()
    if getattr(current_factory, "_hermes_session_injector", False):
@@ -220,16 +165,10 @@ def _install_session_record_factory() -> None:
 _install_session_record_factory()
 # ---------------------------------------------------------------------------
 # Filters
 # ---------------------------------------------------------------------------
 class _ComponentFilter(logging.Filter):
-    """Only pass records whose logger name starts with one of *prefixes*.
+    """Only pass records whose logger name starts with one of *prefixes*."""
    Used to route gateway-specific records to ``gateway.log`` while
    keeping ``agent.log`` as the catch-all.
    """
    def __init__(self, prefixes: Sequence[str]) -> None:
        super().__init__()
@@ -251,18 +190,11 @@ COMPONENT_PREFIXES = {
    "tools": ("tools",),
    "cli": ("hermes_cli", "cli"),
    "cron": ("cron",),
-    "gui": (
+    "gui": ("hermes_cli.web_server", "hermes_cli.pty_bridge", "tui_gateway", "uvicorn"),
        "hermes_cli.web_server",
        "hermes_cli.pty_bridge",
        "tui_gateway",
        "uvicorn",
    ),
 }
 # ---------------------------------------------------------------------------
 # Main setup
 # ---------------------------------------------------------------------------
 def setup_logging(
    *,
@@ -275,41 +207,13 @@ def setup_logging(
 ) -> Path:
    """Configure the Hermes logging subsystem.
-    Safe to call multiple times — the second call is a no-op unless
+    Safe to call multiple times; the second call is a no-op unless *force* is ``True``. Level and
-    *force* is ``True``.
+    rotation defaults come from config.yaml ``logging.*``. ``mode="gateway"`` adds ``gateway.log``
-
+    (gateway components only) and ``mode="gui"`` adds ``gui.log`` (dashboard / TUI-gateway).
-    Parameters
+    Returns the ``logs/`` directory.
    ----------
    hermes_home
        Override for the Hermes home directory.  Falls back to
        ``get_hermes_home()`` (profile-aware).
    log_level
        Minimum level for the ``agent.log`` file handler.  Accepts any
        standard Python level name (``"DEBUG"``, ``"INFO"``, ``"WARNING"``).
        Defaults to ``"INFO"`` or the value from config.yaml ``logging.level``.
    max_size_mb
        Maximum size of each log file in megabytes before rotation.
        Defaults to 5 or the value from config.yaml ``logging.max_size_mb``.
    backup_count
        Number of rotated backup files to keep.
        Defaults to 3 or the value from config.yaml ``logging.backup_count``.
    mode
        Caller context: ``"cli"``, ``"gateway"``, ``"gui"``, ``"cron"``.
        When ``"gateway"``, an additional ``gateway.log`` file is created
        that receives only gateway-component records.
        When ``"gui"``, an additional ``gui.log`` file is created that
        receives dashboard and TUI-gateway component records.
    force
        Re-run setup even if it has already been called.
    Returns
    -------
    Path
        The ``logs/`` directory where files are written.
    """
    global _logging_initialized
    home = hermes_home or get_hermes_home()
    from hermes_constants import mkdir_under_hermes_home
    log_dir = mkdir_under_hermes_home(home / "logs")
    # Read config defaults (best-effort — config may not be loaded yet).
@@ -325,48 +229,21 @@ def setup_logging(
    root = logging.getLogger()
-    # --- agent.log (INFO+) — the main activity log -------------------------
+    # (filename, level, max_bytes, backup_count, component) — component gates on ``mode`` and
-    _add_rotating_handler(
+    # restricts the file to that component's logger prefixes.
-        root,
+    handler_specs = (
-        log_dir / "agent.log",
+        ("agent.log", level, max_bytes, backups, None),
-        level=level,
+        ("errors.log", logging.WARNING, 2 * 1024 * 1024, 2, None),
-        max_bytes=max_bytes,
+        ("gateway.log", logging.INFO, 5 * 1024 * 1024, 3, "gateway"),
-        backup_count=backups,
+        ("gui.log", logging.INFO, 10 * 1024 * 1024, 5, "gui"),
        formatter=RedactingFormatter(_LOG_FORMAT),
    )
-
+    for filename, lvl, size, count, component in handler_specs:
-    # --- errors.log (WARNING+) — quick triage log --------------------------
+        if component is not None and mode != component:
-    _add_rotating_handler(
+            continue
        root,
        log_dir / "errors.log",
        level=logging.WARNING,
        max_bytes=2 * 1024 * 1024,
        backup_count=2,
        formatter=RedactingFormatter(_LOG_FORMAT),
    )
    # --- gateway.log (INFO+, gateway component only) ------------------------
    if mode == "gateway":
        _add_rotating_handler(
-            root,
+            log_dir / filename, level=lvl, max_bytes=size, backup_count=count,
            log_dir / "gateway.log",
            level=logging.INFO,
            max_bytes=5 * 1024 * 1024,
            backup_count=3,
            formatter=RedactingFormatter(_LOG_FORMAT),
-            log_filter=_ComponentFilter(COMPONENT_PREFIXES["gateway"]),
+            log_filter=_ComponentFilter(COMPONENT_PREFIXES[component]) if component else None,
        )
    # --- gui.log (INFO+, dashboard/tui-gateway components) -----------------
    if mode == "gui":
        _add_rotating_handler(
            root,
            log_dir / "gui.log",
            level=logging.INFO,
            max_bytes=10 * 1024 * 1024,
            backup_count=5,
            formatter=RedactingFormatter(_LOG_FORMAT),
            log_filter=_ComponentFilter(COMPONENT_PREFIXES["gui"]),
        )
    if _logging_initialized and not force:
@@ -376,28 +253,21 @@ def setup_logging(
    if root.level == logging.NOTSET or root.level > level:
        root.setLevel(level)
-    # Suppress noisy third-party loggers.
+    _quiet_noisy_loggers()
    for name in _NOISY_LOGGERS:
        logging.getLogger(name).setLevel(logging.WARNING)
    _logging_initialized = True
    return log_dir
 def setup_verbose_logging() -> None:
-    """Enable DEBUG-level console logging for ``--verbose`` / ``-v`` mode.
+    """Enable DEBUG-level console logging for ``--verbose`` / ``-v`` mode."""
    Called by ``AIAgent.__init__()`` when ``verbose_logging=True``.
    """
    from agent.redact import RedactingFormatter
    root = logging.getLogger()
    # Avoid adding duplicate stream handlers.
-    for h in root.handlers:
+    if any(getattr(h, "_hermes_verbose", False) for h in root.handlers):
-        if isinstance(h, logging.StreamHandler) and not isinstance(h, RotatingFileHandler):
+        return
            if getattr(h, "_hermes_verbose", False):
                return
    handler = logging.StreamHandler(_safe_stderr())
    handler.setLevel(logging.DEBUG)
@@ -409,40 +279,29 @@ def setup_verbose_logging() -> None:
    if root.level > logging.DEBUG:
        root.setLevel(logging.DEBUG)
-    # Keep third-party libraries at WARNING to reduce noise.
+    _quiet_noisy_loggers()
    for name in _NOISY_LOGGERS:
        logging.getLogger(name).setLevel(logging.WARNING)
    # rex-deploy at INFO for sandbox status.
    logging.getLogger("rex-deploy").setLevel(logging.INFO)
 # ---------------------------------------------------------------------------
 # Internal helpers
-# ---------------------------------------------------------------------------
+
 def _quietly(fn) -> None:
    """Call *fn* (a ``close``/``stop`` bound method) swallowing errors — teardown must never raise."""
    try:
        fn()
    except Exception:
        pass
 class _ManagedRotatingFileHandler(RotatingFileHandler):
    """RotatingFileHandler that ensures group-writable perms in managed mode
    AND survives external rotation.
-    Two responsibilities:
+    In managed mode (NixOS) the setgid stateDir needs group-readable files, but ``_open()`` and
-
+    ``doRollover()`` honor the umask (0644), so ``chmod 0660`` is applied after both. Also, a
-    1.  In managed mode (NixOS), the stateDir uses setgid (2770) so new files
+    rotating handler holds an fd: if the file is rotated externally (logrotate, ``mv``) writes
-        inherit the hermes group. However, both ``_open()`` (initial creation)
+    silently go to the old inode, so before each emit the path's inode is compared to the open
-        and ``doRollover()`` create files via ``open()``, which uses the
+    stream's and the file reopened on mismatch (the ``WatchedFileHandler`` pattern).
        process umask — typically 0022, producing 0644. This subclass applies
        ``chmod 0660`` after both operations so the gateway and interactive
        users can share log files.
    2.  ``RotatingFileHandler`` keeps an open file descriptor.  If anything
        rotates the file *externally* (``logrotate``, manual ``mv``,
        another process rotating under us, a transient unlink), our fd
        keeps pointing at the renamed/unlinked inode and every subsequent
        write goes to ``gateway.log.1`` instead of ``gateway.log`` — silent
        log loss for the file every operator expects to read.  Before each
        emit we ``stat`` ``baseFilename`` and compare it against the open
        stream's inode; on mismatch we reopen.  This is the same pattern
        as stdlib ``WatchedFileHandler.reopenIfNeeded()``, adapted for
        rotating handlers.
    """
    def __init__(self, *args, **kwargs):
@@ -451,8 +310,6 @@ class _ManagedRotatingFileHandler(RotatingFileHandler):
        super().__init__(*args, **kwargs)
        # Snapshot the inode of the currently open stream so emit() can
        # detect external rotation without an extra fstat per write.
        self._stat_dev: Optional[int] = None
        self._stat_ino: Optional[int] = None
        self._record_stream_stat()
    def _chmod_if_managed(self):
@@ -462,62 +319,51 @@ class _ManagedRotatingFileHandler(RotatingFileHandler):
            except OSError:
                pass
-    def _record_stream_stat(self) -> None:
+    def _record_stream_stat(self, st: Optional[os.stat_result] = None) -> None:
        """Snapshot dev/ino of ``baseFilename`` so we can detect external rotation."""
        try:
-            st = os.stat(self.baseFilename)
+            st = st or os.stat(self.baseFilename)
            self._stat_dev, self._stat_ino = st.st_dev, st.st_ino
        except OSError:
            self._stat_dev, self._stat_ino = None, None
    def _reopen_stream(self, stat_result=None) -> None:
        """Close the current stream and open ``baseFilename`` afresh (best-effort).
        On failure the stream is left ``None`` so the next emit bails rather than writing to a
        stale inode.
        """
        if self.stream is not None:
            _quietly(self.stream.close)
        self.stream = None  # type: ignore[assignment]
        try:
            self.stream = self._open()
        except Exception:
            return
        self._record_stream_stat(stat_result)
    def _reopen_if_externally_rotated(self) -> None:
        """Reopen the stream when ``baseFilename`` no longer matches our fd.
-        Triggered when ``baseFilename`` was renamed (logrotate), unlinked,
+        Triggered when ``baseFilename`` was renamed (logrotate), unlinked, or replaced by a
-        or replaced by a different inode.  Silent + best-effort: any error
+        different inode. Silent + best-effort: any error falls back to the existing (possibly stale)
-        falls back to the existing (possibly stale) stream so logging keeps
+        stream so logging keeps working instead of dying on a stat failure.
        working instead of dying on a stat failure.
        """
        try:
            st = os.stat(self.baseFilename)
        except FileNotFoundError:
-            # File was rotated/unlinked underneath us.  Close + reopen so a
+            # File was rotated/unlinked underneath us: reopen so a fresh inode
-            # fresh inode is created at the expected path.
+            # is created at the expected path.
-            try:
+            self._reopen_stream()
                if self.stream is not None:
                    self.stream.close()
            except Exception:
                pass
            self.stream = None  # type: ignore[assignment]
            try:
                self.stream = self._open()
                self._record_stream_stat()
            except Exception:
                # Couldn't reopen — leave stream=None; next emit will
                # bail rather than write to a stale inode.
                pass
            return
        except OSError:
            return  # transient — try again on the next emit
        if self._stat_dev is None or self._stat_ino is None:
-            self._stat_dev, self._stat_ino = st.st_dev, st.st_ino
+            self._record_stream_stat(st)
-            return
+        elif (st.st_dev, st.st_ino) != (self._stat_dev, self._stat_ino):
-
+            # baseFilename now points at a DIFFERENT inode than the one we hold open.
-        if (st.st_dev, st.st_ino) != (self._stat_dev, self._stat_ino):
+            self._reopen_stream(st)
            # baseFilename now points at a DIFFERENT inode than the one we
            # hold open.  Close the old stream and open the new file.
            try:
                if self.stream is not None:
                    self.stream.close()
            except Exception:
                pass
            self.stream = None  # type: ignore[assignment]
            try:
                self.stream = self._open()
                self._stat_dev, self._stat_ino = st.st_dev, st.st_ino
            except Exception:
                pass
    def emit(self, record: logging.LogRecord) -> None:
        # Cheap-ish stat-per-record check; the kernel caches inode metadata
@@ -528,19 +374,13 @@ class _ManagedRotatingFileHandler(RotatingFileHandler):
    def handleError(self, record: logging.LogRecord) -> None:
        """Suppress the known Windows ``concurrent-log-handler`` lock timeout
        instead of printing a traceback.
-        CLH's own ``emit()`` wraps its body in ``try/except Exception:
+        CLH's ``emit()`` catches the ``"Cannot acquire lock after N attempts"`` RuntimeError and
-        self.handleError(record)``, so the ``"Cannot acquire lock after N
+        routes it here, so this override is the single point to silence it before stdlib prints to
-        attempts"`` RuntimeError raised in ``_do_lock()`` is caught inside CLH
+        stderr (which the Desktop slash-worker captures and surfaces into chat output).
-        and routed here — it never propagates out of ``super().emit()``.  This
+        """
-        override is the single point where that timeout can be silenced before
+        if not _is_windows_concurrent_log_lock_timeout(sys.exc_info()[1]):
-        the stdlib handler prints it to stderr (which, under the Desktop
+            super().handleError(record)
        slash-worker, is captured and surfaced into chat output)."""
        exc = sys.exc_info()[1]
        if _is_windows_concurrent_log_lock_timeout(exc):
            return
        super().handleError(record)
    def _open(self):
        stream = super()._open()
@@ -555,48 +395,42 @@ class _ManagedRotatingFileHandler(RotatingFileHandler):
        self._record_stream_stat()
 def _new_file_handler(
    path: Path, *, level: int, max_bytes: int, backup_count: int, formatter
 ) -> "_ManagedRotatingFileHandler":
    """Create the ``logs/`` directory and a configured ``_ManagedRotatingFileHandler``."""
    mkdir_under_hermes_home(path.parent)
    handler = _ManagedRotatingFileHandler(
        str(path), maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8"
    )
    handler.setLevel(level)
    handler.setFormatter(formatter)
    return handler
 class _ProfileRoutingFileHandler(logging.Handler):
    """Route queued records to the log file for their Hermes home.
-    Dashboard logging is initialized once for the process that launched it,
+    The handler itself is used only behind the existing QueueListener, so its small routing lock
-    while the desktop cron ticker can execute jobs for several profile homes.
+    never blocks an agent or dashboard event loop. The underlying handlers retain the existing
-    A normal ``RotatingFileHandler`` therefore pins every cron record to the
+    rotation, redaction, and managed permission behavior.
    dashboard profile. This handler keeps one rotating file handler per live
    profile and selects it from the home captured by the record factory.
    The handler itself is used only behind the existing QueueListener, so its
    small routing lock never blocks an agent or dashboard event loop. The
    underlying handlers retain the existing rotation, redaction, and managed
    permission behavior.
    """
-    def __init__(
+    def __init__(self, existing: RotatingFileHandler, profile_homes: Sequence[Path]) -> None:
-        self,
+        """Take over *existing*'s path, level, rotation, formatter and filters."""
-        *,
+        super().__init__(level=existing.level)
-        default_path: Path,
+        resolved = Path(existing.baseFilename).resolve()
-        profile_homes: Sequence[Path],
+        self.baseFilename = str(resolved)
-        level: int,
+        self._hermes_routed_log_path = resolved
-        max_bytes: int,
+        self._default_home = resolved.parent.parent.resolve()
-        backup_count: int,
+        self._profile_homes = {Path(home).expanduser().resolve() for home in profile_homes}
-        formatter: logging.Formatter | None,
+        self._filename = resolved.name
-        log_filters: Sequence[logging.Filter],
+        self._max_bytes = getattr(existing, "maxBytes", 0)
-    ) -> None:
+        self._backup_count = getattr(existing, "backupCount", 0)
        super().__init__(level=level)
        self.baseFilename = str(default_path.resolve())
        self._hermes_routed_log_path = Path(self.baseFilename)
        self._default_home = Path(self.baseFilename).parent.parent.resolve()
        self._profile_homes = {
            Path(home).expanduser().resolve()
            for home in profile_homes
        }
        self._filename = Path(self.baseFilename).name
        self._max_bytes = max_bytes
        self._backup_count = backup_count
        self._profile_handlers: dict[Path, _ManagedRotatingFileHandler] = {}
        self._profile_handlers_lock = threading.RLock()
-        if formatter is not None:
+        self.setFormatter(existing.formatter)
-            self.setFormatter(formatter)
+        for log_filter in existing.filters:
        for log_filter in log_filters:
            self.addFilter(log_filter)
    def _home_for_record(self, record: logging.LogRecord) -> Path:
@@ -609,24 +443,12 @@ class _ProfileRoutingFileHandler(logging.Handler):
    def _handler_for_home(self, home: Path) -> _ManagedRotatingFileHandler:
        with self._profile_handlers_lock:
-            handler = self._profile_handlers.get(home)
+            if home not in self._profile_handlers:
-            if handler is not None:
+                self._profile_handlers[home] = _new_file_handler(
-                return handler
+                    home / "logs" / self._filename, level=self.level, max_bytes=self._max_bytes,
-
+                    backup_count=self._backup_count, formatter=self.formatter,
-            path = home / "logs" / self._filename
+                )
-            from hermes_constants import mkdir_under_hermes_home
+            return self._profile_handlers[home]
            mkdir_under_hermes_home(path.parent)
            handler = _ManagedRotatingFileHandler(
                str(path),
                maxBytes=self._max_bytes,
                backupCount=self._backup_count,
                encoding="utf-8",
            )
            handler.setLevel(self.level)
            handler.setFormatter(self.formatter)
            self._profile_handlers[home] = handler
            return handler
    def emit(self, record: logging.LogRecord) -> None:
        try:
@@ -639,14 +461,10 @@ class _ProfileRoutingFileHandler(logging.Handler):
            handlers = list(self._profile_handlers.values())
            self._profile_handlers.clear()
        for handler in handlers:
-            try:
+            _quietly(handler.close)
                handler.close()
            except Exception:
                pass
        super().close()
 # ---------------------------------------------------------------------------
 # Asynchronous file logging — keep the cross-process rotation lock off the loop
 #
 # The rotating file handlers serialize rollover with a cross-process lock (see
@@ -656,7 +474,6 @@ class _ProfileRoutingFileHandler(logging.Handler):
 # WebSocket clients.  To keep file I/O off the hot path, every file handler is
 # driven by a single ``QueueListener`` on a dedicated thread; loggers only touch
 # an in-memory queue (a non-blocking enqueue).
 # ---------------------------------------------------------------------------
 _log_queue: "Optional[queue.SimpleQueue]" = None
 _queue_listener: Optional[QueueListener] = None
@@ -674,49 +491,47 @@ _queue_state_lock = threading.Lock()
 class _NonFormattingQueueHandler(QueueHandler):
    """``QueueHandler`` for an in-process queue.
-    Stdlib ``prepare()`` formats the record and drops ``args``/``exc_info`` so it
+    Stdlib ``prepare()`` formats and strips ``args``/``exc_info`` for pickling across processes;
-    can be pickled to another process.  Our queue is in-process, so we skip that
+    our queue is in-process, so the target handlers get an unformatted record and apply their own
-    and hand the target file handlers an unformatted record — they apply their
+    ``RedactingFormatter`` on the listener thread. A shallow copy is returned because the emitting
-    own ``RedactingFormatter`` and component filters on the listener thread.
+    thread's synchronous handlers may mutate ``record.message`` while the listener reads it.
    We return a **shallow copy** rather than the original record: the same
    record is still owned by the emitting thread (and any synchronous handler
    on it, e.g. a ``StreamHandler``), which may format/mutate ``record.message``
    while our listener thread reads it. Copying preserves ``msg``/``args``/
    ``exc_info`` for the deferred format while removing the cross-thread
    mutation race on a shared object.
    """
    def prepare(self, record: logging.LogRecord) -> logging.LogRecord:
        return copy.copy(record)
 def _stop_queue_listener_locked() -> None:
    """Stop the listener assuming ``_queue_state_lock`` is already held."""
    global _queue_listener
    listener, _queue_listener = _queue_listener, None
    if listener is not None:
        try:
            listener.stop()
        except Exception:
            pass
 def _stop_queue_listener() -> None:
    """Flush and stop the background log listener (idempotent, thread-safe).
    This is the atexit hook, so it must acquire the state lock itself.
    """
    global _queue_listener
    with _queue_state_lock:
-        _stop_queue_listener_locked()
+        listener, _queue_listener = _queue_listener, None
        if listener is not None:
            _quietly(listener.stop)
 def _start_queue_listener_locked() -> None:
    """(Re)build + start a listener over the current handler set (``_queue_state_lock`` held).
    A running listener is stopped first; this only happens while handlers are being added
    (queue empty), so ``stop()`` returns immediately.
    """
    global _queue_listener
    if _queue_listener is not None:
        _queue_listener.stop()
    _queue_listener = QueueListener(_log_queue, *_queued_file_handlers, respect_handler_level=True)
    _queue_listener.start()
 def _register_queued_handler(handler: logging.Handler) -> None:
-    """Route *handler* through the shared async queue instead of attaching it to
+    """Route *handler* through the shared async queue instead of attaching it to *root* directly, so
-    *root* directly, so emitting threads never block on file I/O or the
+    emitting threads never block on file I/O or the cross-process rotation lock. The
-    cross-process rotation lock.  The ``QueueListener`` applies each handler's
+    ``QueueListener`` applies each handler's own level and filters on its worker thread.
-    own level and filters on its worker thread."""
+    """
-    global _log_queue, _queue_listener, _queue_atexit_registered
+    global _log_queue, _queue_atexit_registered
    with _queue_state_lock:
        if _log_queue is None:
            _log_queue = queue.SimpleQueue()
@@ -727,15 +542,7 @@ def _register_queued_handler(handler: logging.Handler) -> None:
            # queue via propagation.
            logging.getLogger().addHandler(qh)
        _queued_file_handlers.append(handler)
-        # Rebuild the listener with the full target set.  This only happens
+        _start_queue_listener_locked()
        # while init_logging() adds handlers (2-3 times, queue empty), so
        # stop() returns immediately.
        if _queue_listener is not None:
            _queue_listener.stop()
        _queue_listener = QueueListener(
            _log_queue, *_queued_file_handlers, respect_handler_level=True
        )
        _queue_listener.start()
        if not _queue_atexit_registered:
            # Runs before logging.shutdown (registered earlier at import time),
            # so the listener stops before its file handlers are closed.
@@ -746,14 +553,12 @@ def _register_queued_handler(handler: logging.Handler) -> None:
 def flush_log_queue() -> None:
    """Block until all queued records have been written, then resume.
-    Draining is done by stopping the listener (which processes every pending
+    Draining is done by stopping the listener (which processes every pending record before joining)
-    record before joining) and restarting it.  Used by tests that read a log
+    and restarting it. Used by tests that read a log file right after emitting to it.
    file right after emitting to it.
-    NOTE: ``stop()`` joins the worker thread, so this blocks until the queue
+    NOTE: ``stop()`` joins the worker thread, so this blocks until the queue is empty. Do NOT call
-    is empty. Do NOT call this on a hard-exit path where the listener may be
+    this on a hard-exit path where the listener may be wedged on the rotation lock — use
-    wedged on the rotation lock — use ``drain_log_queue()`` there instead,
+    ``drain_log_queue()`` there instead, which bounds the wait.
    which bounds the wait.
    """
    with _queue_state_lock:
        listener = _queue_listener
@@ -765,59 +570,34 @@ def flush_log_queue() -> None:
 def drain_log_queue(timeout: float = 1.0) -> None:
    """Best-effort, time-bounded drain for hard-exit paths (no restart).
-    Unlike ``flush_log_queue()``, this stops the listener WITHOUT restarting it
+    Unlike ``flush_log_queue()``, this stops the listener WITHOUT restarting it (the process is
-    (the process is about to exit) and bounds the drain: if the listener's
+    about to exit) and bounds the drain: if the listener's worker thread is wedged on the cross-
-    worker thread is wedged on the cross-process rotation lock — the very
+    process rotation lock — the very failure this async-logging change exists to survive — an
-    failure this async-logging change exists to survive — an unbounded
+    unbounded ``stop()``/join would re-freeze the shutdown path.
    ``stop()``/join would re-freeze the shutdown path. We run ``stop()`` on a
    throwaway thread and only wait ``timeout`` seconds for it; if it hasn't
    drained by then we abandon the last few records and let ``os._exit``
    proceed. Availability beats the last log line when the disk is already
    wedged.
    """
    listener = _queue_listener
    if listener is None:
        return
-
+    t = threading.Thread(target=lambda: _quietly(listener.stop), name="hermes-log-drain", daemon=True)
    def _drain() -> None:
        try:
            listener.stop()
        except Exception:
            pass
    t = threading.Thread(target=_drain, name="hermes-log-drain", daemon=True)
    t.start()
    t.join(timeout)
 def rotating_file_handlers() -> list:
    """Return the live rotating file handlers.
    They are attached to the async ``QueueListener`` rather than the root
    logger, so callers/tests must use this instead of scanning
    ``logging.getLogger().handlers``."""
    return list(_queued_file_handlers)
 def enable_profile_log_routing(profile_homes: Sequence[str | Path]) -> bool:
    """Make the queued file logs follow a desktop profile context.
-    ``setup_logging`` normally binds handlers to one process home. The
+    ``setup_logging`` normally binds handlers to one process home. The desktop dashboard is the
-    desktop dashboard is the exception: its embedded cron ticker may run
+    exception: its embedded cron ticker may run jobs for every profile. Replace the existing static
-    jobs for every profile. Replace the existing static file handlers with
+    file handlers with profile routers after that profile list is known.
    profile routers after that profile list is known.
-    Returns ``True`` when routing is enabled or was already enabled. A
+    Returns ``True`` when routing is enabled or was already enabled. A single-profile caller is left
-    single-profile caller is left untouched because its existing handlers are
+    untouched because its existing handlers are already correctly scoped.
    already correctly scoped.
    """
    global _queue_listener
-
+    homes: list[Path] = []
    homes = []
    for entry in profile_homes:
        home = entry[1] if isinstance(entry, tuple) else entry
        try:
-            resolved = Path(home).expanduser().resolve()
+            resolved = Path(entry[1] if isinstance(entry, tuple) else entry).expanduser().resolve()
        except (TypeError, ValueError, OSError):
            continue
        if resolved not in homes:
@@ -834,58 +614,38 @@ def enable_profile_log_routing(profile_homes: Sequence[str | Path]) -> bool:
        listener = _queue_listener
        if listener is not None:
            listener.stop()
            _queue_listener = None
        replacement = []
        for existing in _queued_file_handlers:
-            if not isinstance(existing, RotatingFileHandler):
+            if isinstance(existing, RotatingFileHandler):
                replacement.append(_ProfileRoutingFileHandler(existing, homes))
                _quietly(existing.close)
            else:
                replacement.append(existing)
                continue
            default_path = Path(existing.baseFilename)
            router = _ProfileRoutingFileHandler(
                default_path=default_path,
                profile_homes=homes,
                level=existing.level,
                max_bytes=getattr(existing, "maxBytes", 0),
                backup_count=getattr(existing, "backupCount", 0),
                formatter=existing.formatter,
                log_filters=list(existing.filters),
            )
            replacement.append(router)
            try:
                existing.close()
            except Exception:
                pass
        _queued_file_handlers[:] = replacement
        if listener is not None:
-            _queue_listener = QueueListener(
+            _start_queue_listener_locked()
                _log_queue, *_queued_file_handlers, respect_handler_level=True
            )
            _queue_listener.start()
        return True
 def _reset_queued_handlers() -> None:
    """Tear down the async logging queue + listener (test-isolation helper)."""
    global _log_queue
    _stop_queue_listener()
    with _queue_state_lock:
        _stop_queue_listener_locked()
        root = logging.getLogger()
        for h in list(root.handlers):
            if getattr(h, "_hermes_queue", False):
                root.removeHandler(h)
        for h in list(_queued_file_handlers):
-            try:
+            _quietly(h.close)
                h.close()
            except Exception:
                pass
        _queued_file_handlers.clear()
        _log_queue = None
 def _add_rotating_handler(
    logger: logging.Logger,
    path: Path,
    *,
    level: int,
@@ -894,33 +654,21 @@ def _add_rotating_handler(
    formatter: logging.Formatter,
    log_filter: Optional[logging.Filter] = None,
 ) -> None:
-    """Add a ``RotatingFileHandler`` to *logger*, skipping if one already
+    """Register a queued ``RotatingFileHandler`` for *path*, skipping if one already exists for the
-    exists for the same resolved file path (idempotent).
+    same resolved file path (idempotent).
    Parameters
    ----------
    log_filter
        Optional filter to attach to the handler (e.g. ``_ComponentFilter``
        for gateway.log).
    """
    resolved = path.resolve()
    for existing in _queued_file_handlers:
-        if (
+        # Already attached directly, or already covered by the profile router.
        if getattr(existing, "_hermes_routed_log_path", None) == resolved or (
            isinstance(existing, RotatingFileHandler)
            and Path(getattr(existing, "baseFilename", "")).resolve() == resolved
        ):
-            return  # already attached
+            return
        if getattr(existing, "_hermes_routed_log_path", None) == resolved:
            return  # already covered by the profile router
-    from hermes_constants import mkdir_under_hermes_home
+    handler = _new_file_handler(
-    mkdir_under_hermes_home(path.parent)
+        path, level=level, max_bytes=max_bytes, backup_count=backup_count, formatter=formatter,
    handler = _ManagedRotatingFileHandler(
        str(path), maxBytes=max_bytes, backupCount=backup_count,
        encoding="utf-8",
    )
    handler.setLevel(level)
    handler.setFormatter(formatter)
    if log_filter is not None:
        handler.addFilter(log_filter)
    # Route through the async queue instead of ``logger.addHandler(handler)`` so
@@ -929,10 +677,7 @@ def _add_rotating_handler(
 def _read_logging_config():
-    """Best-effort read of ``logging.*`` from config.yaml.
+    """Best-effort read of ``logging.*`` from config.yaml."""
    Returns ``(level, max_size_mb, backup_count)`` — any may be ``None``.
    """
    try:
        # Prefer the shared (mtime, size)-keyed raw-config cache so this read
        # reuses the parse hermes_cli.main's early bridge already did (one
@@ -945,25 +690,22 @@ def _read_logging_config():
        except Exception:
            from utils import fast_safe_load
            config_path = get_config_path()
-            if not config_path.exists():
+            cfg = {}
-                return (None, None, None)
+            if config_path.exists():
-            with open(config_path, "r", encoding="utf-8") as f:
+                with open(config_path, "r", encoding="utf-8") as f:
-                cfg = fast_safe_load(f) or {}
+                    cfg = fast_safe_load(f) or {}
-        if cfg:
+        if not cfg:
-            # Managed scope: an administrator can pin logging.* too. Overlay via
+            return (None, None, None)
-            # the shared helper (fail-open) since this reads config.yaml directly.
+        # Managed scope: an administrator can pin logging.* too. Overlay via
-            try:
+        # the shared helper (fail-open) since this reads config.yaml directly.
-                from hermes_cli import managed_scope
+        try:
-                cfg = managed_scope.apply_managed_overlay(cfg)
+            from hermes_cli import managed_scope
-            except Exception:
+            cfg = managed_scope.apply_managed_overlay(cfg)
-                pass
+        except Exception:
-            log_cfg = cfg.get("logging", {})
+            pass
-            if isinstance(log_cfg, dict):
+        log_cfg = cfg.get("logging", {})
-                return (
+        if isinstance(log_cfg, dict):
-                    log_cfg.get("level"),
+            return (log_cfg.get("level"), log_cfg.get("max_size_mb"), log_cfg.get("backup_count"))
                    log_cfg.get("max_size_mb"),
                    log_cfg.get("backup_count"),
                )
    except Exception:
        pass
    return (None, None, None)
--- a/utils.py
+++ b/utils.py
@@ -24,8 +24,6 @@ def is_truthy_value(value: Any, default: bool = False) -> bool:
    """Coerce bool-ish values using the project's shared truthy string set."""
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() in TRUTHY_STRINGS
    return bool(value)
@@ -46,47 +44,40 @@ def _preserve_file_mode(path: Path) -> "int | None":
 def _preserve_file_owner(path: Path) -> "tuple[int, int] | None":
    """Capture the owning uid/gid of *path* if the platform supports it."""
    if os.name != "posix":
        return None
    try:
-        st = path.stat()
+        st = path.stat() if os.name == "posix" else None
    except OSError:
        return None
-    return st.st_uid, st.st_gid
+    return (st.st_uid, st.st_gid) if st else None
 def _restore_file_metadata(path: Path, owner: "tuple[int, int] | None", mode: "int | None") -> None:
    """Best-effort re-apply of uid/gid and permission bits after an atomic replace.
    Docker/NAS installs often run some commands as root while the volume is owned by the runtime
    user; ``os.replace`` swaps in the temp file's owner, leaving ``config.yaml`` root-owned, so
    privileged callers chown it back (harmless otherwise). ``tempfile.mkstemp`` creates files 0o600;
    without re-applying *mode* the target would inherit that and break volume mounts relying on
    broader permissions.
    """
    if owner is not None and hasattr(os, "chown"):
        try:
            os.chown(path, owner[0], owner[1])
        except OSError:
            pass
    if mode is not None:
        try:
            os.chmod(path, mode)
        except OSError:
            pass
 def _restore_file_owner(path: Path, owner: "tuple[int, int] | None") -> None:
-    """Re-apply uid/gid after an atomic replace when permitted.
+    _restore_file_metadata(path, owner, None)
    Docker and NAS-backed installs often run some commands as root while the
    persistent volume is owned by the runtime user. ``os.replace`` swaps in the
    temp file's owner, so a root-run config write can leave ``config.yaml`` owned
    by root. Best-effort chown preserves the existing owner for privileged
    callers and is harmless for unprivileged callers that cannot chown.
    """
    if owner is None or not hasattr(os, "chown"):
        return
    try:
        os.chown(path, owner[0], owner[1])
    except OSError:
        pass
 def _restore_file_mode(path: Path, mode: "int | None") -> None:
-    """Re-apply *mode* to *path* after an atomic replace.
+    _restore_file_metadata(path, None, mode)
    ``tempfile.mkstemp`` creates files with 0o600 (owner-only).  After
    ``os.replace`` swaps the temp file into place the target inherits
    those restrictive permissions, breaking Docker / NAS volume mounts
    that rely on broader permissions set by the user.  Calling this
    right after ``os.replace`` restores the original permissions.
    """
    if mode is None:
        return
    try:
        os.chmod(path, mode)
    except OSError:
        pass
 _IS_WINDOWS = os.name == "nt"
@@ -127,42 +118,27 @@ _REPLACE_RETRY_MAX_DELAY_S = 0.1
 def _is_contended_windows_replace_error(exc: OSError) -> bool:
    """Return True for Windows rename failures a retry might clear.
-    Only a *candidate* classification: ``ERROR_ACCESS_DENIED`` covers both a
+    Only a *candidate* classification: ``ERROR_ACCESS_DENIED`` covers both a concurrent handle and a
-    concurrent handle and a real ACL denial, and the two are not reliably
+    real ACL denial, and the two are not reliably distinguishable up front.
    distinguishable up front.  Probing the target with ``os.access`` does not
    work — ``os.replace`` needs delete-child rights on the *parent directory*,
    so a directory-level denial reports the target as writable.  Instead of
    guessing, both cases enter the same bounded retry and a genuine denial
    falls through to the caller with its original error.
    """
-    return _IS_WINDOWS and getattr(exc, "winerror", None) in (
+    return _IS_WINDOWS and getattr(exc, "winerror", None) in _WINDOWS_CONTENDED_REPLACE_ERRORS
        _WINDOWS_CONTENDED_REPLACE_ERRORS
    )
 def _rewrite_in_place(tmp_str: str, real_path: str) -> None:
    """Overwrite *real_path* with the contents of *tmp_str*, in place.
-    Last-resort path for a target whose handle is still held after the retry
+    Last-resort path for a target whose handle is still held after the retry budget: writing through
-    budget: writing through the existing file works where renaming onto it
+    the existing file works where renaming onto it does not.
    does not.  Unlike ``shutil.copyfile`` this never truncates the target to
    zero first — a concurrent reader would otherwise be able to observe an
    empty ``auth.json`` / ``gateway_state.json`` mid-write (measured: a
    4-thread poller sees a 0-byte read during a plain copyfile).  A single
    ``os.write`` of the full payload followed by ``ftruncate`` keeps the
    visible content going straight from old to new.
-    This is still not atomic — it is a strictly smaller window than a copy,
+    This is still not atomic — it is a strictly smaller window than a copy, not the absence of one —
-    not the absence of one — so it runs only after the rename has genuinely
+    so it runs only after the rename has genuinely failed. Writing through the target also preserves
-    failed.  Writing through the target also preserves its ACL, which
+    its ACL, which ``os.replace`` does not (the temp file's inherited ACL wins there).
    ``os.replace`` does not (the temp file's inherited ACL wins there).
    """
    with open(tmp_str, "rb") as src:
        data = src.read()
    flags = os.O_WRONLY | getattr(os, "O_BINARY", 0)
    fd = os.open(real_path, flags)
    try:
        os.lseek(fd, 0, os.SEEK_SET)
        written = 0
        while written < len(data):
            written += os.write(fd, data[written:])
@@ -194,34 +170,13 @@ def _copy_fallback(tmp_str: str, real_path: str) -> None:
 def atomic_replace(tmp_path: Union[str, Path], target: Union[str, Path]) -> str:
    """Atomically move *tmp_path* onto *target*, preserving symlinks.
-    ``os.replace(tmp, target)`` atomically swaps ``tmp`` into place at
+    This helper resolves the symlink first so ``os.replace`` writes to the real file in-place while
-    ``target``.  When ``target`` is a symlink, the symlink itself is
+    the symlink survives. For non-symlink and non-existent paths the behavior is identical to a
-    replaced with a regular file — silently detaching managed deployments
+    plain ``os.replace`` call unless the rename fails with:
    that symlink ``config.yaml`` / ``SOUL.md`` / ``auth.json`` etc. from
    ``~/.hermes/`` to a git-tracked profile package or dotfiles repo
    (GitHub #16743).
-    This helper resolves the symlink first so ``os.replace`` writes to
+    * ``EXDEV`` / ``EBUSY`` (any platform) — cross-device, bind-mount, and busy-file deployments
-    the real file in-place while the symlink survives.  For non-symlink
+    fall back to copy/fsync/unlink immediately. These never clear on retry. * A Windows rename
-    and non-existent paths the behavior is identical to a plain
+    contended by another open handle (winerror 5/32/33).
    ``os.replace`` call unless the rename fails with:
    * ``EXDEV`` / ``EBUSY`` (any platform) — cross-device, bind-mount, and
      busy-file deployments fall back to copy/fsync/unlink immediately.
      These never clear on retry.
    * A Windows rename contended by another open handle (winerror 5/32/33).
      CPython opens files without ``FILE_SHARE_DELETE``, so *any* concurrent
      reader of the target blocks the rename.  The rename is retried with
      jittered backoff first — a retry that wins keeps the write atomic —
      and only a target whose handle outlives the budget is rewritten in
      place, so the update lands instead of being silently dropped.
    A genuine Windows permission failure produces the same winerror as a
    contended one, so it is not classified up front: it exhausts the retry
    budget, fails the in-place rewrite too, and is re-raised unchanged.
    Returns the resolved real path used for the replace, so callers that
    need to re-apply permissions can target it instead of the symlink.
    """
    target_str = str(target)
    real_path = os.path.realpath(target_str) if os.path.islink(target_str) else target_str
@@ -239,13 +194,9 @@ def atomic_replace(tmp_path: Union[str, Path], target: Union[str, Path]) -> str:
            from agent.retry_utils import jittered_backoff
            for attempt in range(1, _REPLACE_RETRY_ATTEMPTS + 1):
-                time.sleep(
+                time.sleep(jittered_backoff(
-                    jittered_backoff(
+                    attempt, base_delay=_REPLACE_RETRY_BASE_DELAY_S, max_delay=_REPLACE_RETRY_MAX_DELAY_S
-                        attempt,
+                ))
                        base_delay=_REPLACE_RETRY_BASE_DELAY_S,
                        max_delay=_REPLACE_RETRY_MAX_DELAY_S,
                    )
                )
                try:
                    os.replace(tmp_str, real_path)
                    return real_path
@@ -260,10 +211,8 @@ def atomic_replace(tmp_path: Union[str, Path], target: Union[str, Path]) -> str:
                    exc = retry_exc
        logger.debug(
            "atomic_replace: %s -> %s failed with %s; falling back to %s",
-            tmp_str,
+            tmp_str, real_path,
-            real_path,
+            getattr(exc, "winerror", None) or errno.errorcode.get(exc.errno or 0, exc.errno),
            getattr(exc, "winerror", None)
            or errno.errorcode.get(exc.errno or 0, exc.errno),
            "in-place rewrite" if contended else "copy",
        )
        if contended:
@@ -276,6 +225,53 @@ def atomic_replace(tmp_path: Union[str, Path], target: Union[str, Path]) -> str:
    return real_path
 def _atomic_write(
    path: Path,
    write,
    *,
    prefix: str,
    encoding: str = "utf-8",
    mode: "int | None" = None,
    preserve_owner: bool = True,
 ) -> None:
    """Temp file + fsync + :func:`atomic_replace`, then re-apply owner/mode.
    *write(f)* emits the payload into the open text handle. *mode* (when not ``None``) is fchmod'd
    onto the temp fd BEFORE the replace so the target never transits through mkstemp's 0600 (fchmod
    is Unix-only; the post-replace chmod is the sole path on Windows and harmless elsewhere). The
    temp file is removed on any failure — ``BaseException`` on purpose, so KeyboardInterrupt /
    SystemExit still clean up before re-raising.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    original_owner = _preserve_file_owner(path) if preserve_owner else None
    fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), prefix=prefix, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding=encoding) as f:
            if mode is not None and hasattr(os, "fchmod"):
                os.fchmod(f.fileno(), mode)
            write(f)
            f.flush()
            os.fsync(f.fileno())
        # Preserve symlinks — swap in-place on the real file (GitHub #16743).
        _restore_file_metadata(Path(atomic_replace(tmp_path, path)), original_owner, mode)
    except BaseException:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
 def _mode_for_write(
    path: Path, create_mode: "int | None", preserve: bool = True
 ) -> "int | None":
    """Existing permission bits of *path* (when *preserve*), else *create_mode* for a new file."""
    mode = _preserve_file_mode(path) if preserve else None
    if mode is None and create_mode is not None and not path.exists():
        mode = create_mode
    return mode
 def atomic_write_text(
    path: Union[str, Path],
    content: str,
@@ -287,60 +283,22 @@ def atomic_write_text(
 ) -> None:
    """Write *content* to *path* via temp file + fsync + atomic rename.
-    Ensures the target file is never left in a partially-written state if
+    Ensures the target file is never left in a partially-written state if the process crashes or is
-    the process crashes or is interrupted.  ``atomic_replace`` preserves
+    interrupted. ``atomic_replace`` preserves symlinks and handles cross-device / busy-file
-    symlinks and handles cross-device / busy-file fallbacks.
+    fallbacks.
-    Used by the memory store, skill manager, and agent importer so that
+    Used by the memory store, skill manager, and agent importer so that every destructive file
-    every destructive file rewrite in the codebase shares one implementation.
+    rewrite in the codebase shares one implementation.
    Args:
        preserve_mode: When True, carry an existing target's permission bits
            and (POSIX, best-effort) owner across the replace, like
            ``atomic_yaml_write`` does unconditionally.  ``os.replace`` swaps
            in mkstemp's 0600 temp file owned by the writing user, so without
            this a root-run rewrite of a user-owned file flips its owner and
            tightens its mode.  The mode is applied to the temp fd *before*
            the replace, so the file never transits through 0600.  Off by
            default: the historical callers (memory store, skill manager,
            cron) own their 0600-is-fine files.
        create_mode: Permission bits to apply when the target does not yet
            exist (otherwise the new file keeps mkstemp's 0600).  Never
            applied to an existing file.
    """
    path = Path(path)
-    path.parent.mkdir(parents=True, exist_ok=True)
+    _atomic_write(
-
+        path,
-    original_mode = _preserve_file_mode(path) if preserve_mode else None
+        lambda f: f.write(content),
-    original_owner = _preserve_file_owner(path) if preserve_mode else None
+        prefix=tmp_prefix,
-    effective_mode = original_mode
+        encoding=encoding,
-    if effective_mode is None and create_mode is not None and not path.exists():
+        mode=_mode_for_write(path, create_mode, preserve=preserve_mode),
-        effective_mode = create_mode
+        preserve_owner=preserve_mode,
    fd, tmp_path = tempfile.mkstemp(
        dir=str(path.parent), prefix=tmp_prefix, suffix=".tmp"
    )
    try:
        with os.fdopen(fd, "w", encoding=encoding) as handle:
            if effective_mode is not None and hasattr(os, "fchmod"):
                # fchmod the temp fd BEFORE the replace so the target never
                # transits through mkstemp's 0600. fchmod is Unix-only; on
                # Windows the post-replace chmod below applies the mode.
                os.fchmod(handle.fileno(), effective_mode)
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
        real_path = atomic_replace(tmp_path, path)
        if preserve_mode:
            _restore_file_owner(Path(real_path), original_owner)
        if effective_mode is not None and not hasattr(os, "fchmod"):
            _restore_file_mode(Path(real_path), effective_mode)
    except BaseException:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
 def atomic_json_write(
@@ -353,66 +311,17 @@ def atomic_json_write(
 ) -> None:
    """Write JSON data to a file atomically.
-    Uses temp file + fsync + os.replace to ensure the target file is never
+    Uses temp file + fsync + os.replace to ensure the target file is never left in a partially-
-    left in a partially-written state. If the process crashes mid-write,
+    written state. If the process crashes mid-write, the previous version of the file remains
-    the previous version of the file remains intact.
+    intact.
    Args:
        path: Target file path (will be created or overwritten).
        data: JSON-serializable data to write.
        indent: JSON indentation (default 2).
        mode: Optional final permission mode. When set, the temp file is
            created and replaced with this mode, avoiding chmod-after-write
            TOCTOU exposure for secret-bearing files.
        **dump_kwargs: Additional keyword args forwarded to json.dump(), such
            as default=str for non-native types.
    """
    path = Path(path)
-    path.parent.mkdir(parents=True, exist_ok=True)
+    _atomic_write(
-
+        path,
-    original_mode = None if mode is not None else _preserve_file_mode(path)
+        lambda f: json.dump(data, f, indent=indent, ensure_ascii=False, **dump_kwargs),
    original_owner = _preserve_file_owner(path)
    fd, tmp_path = tempfile.mkstemp(
        dir=str(path.parent),
        prefix=f".{path.stem}_",
-        suffix=".tmp",
+        mode=mode if mode is not None else _preserve_file_mode(path),
    )
    try:
        if mode is not None and hasattr(os, "fchmod"):
            # fchmod is Unix-only; Windows' os module has no fchmod. Skipping it
            # here is safe — mkstemp already created the temp file as 0o600, and
            # the post-replace os.chmod below applies the final mode durably.
            os.fchmod(fd, mode)
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(
                data,
                f,
                indent=indent,
                ensure_ascii=False,
                **dump_kwargs,
            )
            f.flush()
            os.fsync(f.fileno())
        # Preserve symlinks — swap in-place on the real file (GitHub #16743).
        real_path = atomic_replace(tmp_path, path)
        real_path_obj = Path(real_path)
        _restore_file_owner(real_path_obj, original_owner)
        if mode is not None:
            try:
                os.chmod(real_path_obj, mode)
            except OSError:
                pass
        else:
            _restore_file_mode(real_path_obj, original_mode)
    except BaseException:
        # Intentionally catch BaseException so temp-file cleanup still runs for
        # KeyboardInterrupt/SystemExit before re-raising the original signal.
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
 def warn_if_credential_file_broadly_readable(
@@ -423,29 +332,24 @@ def warn_if_credential_file_broadly_readable(
 ) -> bool:
    """Warn (once per call) when a credential file is group/world-readable.
-    Secret-bearing files that users create by hand (or that older Hermes
+    Secret-bearing files that users create by hand (or that older Hermes versions wrote without an
-    versions wrote without an explicit mode) commonly end up 0o644 under the
+    explicit mode) commonly end up 0o644 under the default umask. This helper is the shared read-
-    default umask. This helper is the shared read-time check for that class:
+    time check for that class: call it before loading any token/credential file so the owner gets a
    call it before loading any token/credential file so the owner gets a
    remediation hint in the logs.
-    Returns True when a warning was emitted. No-ops (returns False) on
+    Returns True when a warning was emitted. No-ops (returns False) on platforms without POSIX
-    platforms without POSIX permission bits semantics (best effort), when the
+    permission bits semantics (best effort), when the file is missing, or when permissions are
-    file is missing, or when permissions are already tight.
+    already tight.
    """
    p = Path(path)
    _log = log or logger
    try:
        file_mode = p.stat().st_mode
    except OSError:
        return False
-    if os.name != "posix":
+    # Windows ACLs don't map onto POSIX group/other bits; st_mode there is synthesized.
-        # Windows ACLs don't map onto POSIX group/other bits; st_mode there
+    if os.name != "posix" or not (file_mode & (stat.S_IRGRP | stat.S_IROTH)):
        # is synthesized and would false-positive.
        return False
-    if not (file_mode & (stat.S_IRGRP | stat.S_IROTH)):
+    (log or logger).warning(
        return False
    _log.warning(
        "%s%s is group/world-readable (mode 0%o) and contains secrets. "
        "Run: chmod 600 %s",
        f"{label} " if label else "",
@@ -459,13 +363,10 @@ def warn_if_credential_file_broadly_readable(
 class IndentDumper(yaml.SafeDumper):
    """PyYAML dumper that indents list items under mapping keys (2-space).
-    Default PyYAML emits "indentless" sequences — list items start at the
+    Default PyYAML emits "indentless" sequences while ``ruamel.yaml`` (used by
-    same column as their parent mapping key.  ``ruamel.yaml`` (used by
+    :func:`atomic_roundtrip_yaml_update`) indents them; mixing both in one ``config.yaml`` makes
-    :func:`atomic_roundtrip_yaml_update`) emits 2-space-indented sequences.
+    stricter parsers like ``js-yaml`` reject it. Forcing ``indentless=False`` keeps every write
-    Mixing both styles in the same ``config.yaml`` produces a file that
+    path byte-identical.
    stricter parsers like ``js-yaml`` reject with ``bad indentation of a
    mapping entry``.  Forcing ``indentless=False`` aligns the two
    serializers so all write paths emit byte-identical layouts (#31999).
    """
    def increase_indent(self, flow=False, indentless=False):  # noqa: ARG002
@@ -483,74 +384,57 @@ def atomic_yaml_write(
 ) -> None:
    """Write YAML data to a file atomically.
-    Uses temp file + fsync + os.replace to ensure the target file is never
+    Uses temp file + fsync + os.replace to ensure the target file is never left in a partially-
-    left in a partially-written state.  If the process crashes mid-write,
+    written state. If the process crashes mid-write, the previous version of the file remains
-    the previous version of the file remains intact.
+    intact.
    Args:
        path: Target file path (will be created or overwritten).
        data: YAML-serializable data to write.
        default_flow_style: YAML flow style (default False).
        sort_keys: Whether to sort dict keys (default False).
        extra_content: Optional string to append after the YAML dump
            (e.g. commented-out sections for user reference).
        create_mode: Permission bits to apply when the target does not yet
            exist (a created file otherwise keeps mkstemp's 0600).  Never
            applied to an existing file, whose mode is always preserved.
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
-    original_mode = _preserve_file_mode(path)
+    def _write(f) -> None:
-    original_owner = _preserve_file_owner(path)
+        # allow_unicode=True writes emoji/kaomoji (e.g. personalities, skin
-    if original_mode is None and create_mode is not None and not path.exists():
+        # cursors) as real UTF-8 instead of fragile escape sequences. Without
-        original_mode = create_mode
+        # it, PyYAML emits astral-plane chars as `\UXXXXXXXX` (8-digit) escapes
        # inside multi-line double-quoted strings wrapped with `\`
        # continuations — a structure that stricter/non-PyYAML parsers and
        # hand-edits routinely break into unclosed quotes, corrupting the whole
        # config (GitHub #51356).
        yaml.dump(
            data,
            f,
            Dumper=IndentDumper,
            default_flow_style=default_flow_style,
            sort_keys=sort_keys,
            allow_unicode=True,
        )
        if extra_content:
            f.write(extra_content)
-    fd, tmp_path = tempfile.mkstemp(
+    _atomic_write(
-        dir=str(path.parent),
+        path, _write, prefix=f".{path.stem}_", mode=_mode_for_write(path, create_mode)
        prefix=f".{path.stem}_",
        suffix=".tmp",
    )
-    try:
+
-        with os.fdopen(fd, "w", encoding="utf-8") as f:
+
-            if original_mode is not None and hasattr(os, "fchmod"):
+def _roundtrip_yaml():
-                # Apply the mode to the temp fd BEFORE the replace so the
+    """ruamel round-trip ``YAML`` configured to keep quotes/Unicode with 2-space indents."""
-                # target never transits through mkstemp's 0600 (the
+    from ruamel.yaml import YAML
-                # post-replace _restore_file_mode below then re-applies it
+
-                # harmlessly, and remains the sole path on Windows).
+    yaml_rt = YAML(typ="rt")
-                os.fchmod(f.fileno(), original_mode)
+    yaml_rt.preserve_quotes = True
-            # allow_unicode=True writes emoji/kaomoji (e.g. personalities, skin
+    yaml_rt.allow_unicode = True
-            # cursors) as real UTF-8 instead of fragile escape sequences. Without
+    yaml_rt.default_flow_style = False
-            # it, PyYAML emits astral-plane chars as `\UXXXXXXXX` (8-digit) escapes
+    yaml_rt.indent(mapping=2, sequence=4, offset=2)
-            # inside multi-line double-quoted strings wrapped with `\`
+    return yaml_rt
-            # continuations — a structure that stricter/non-PyYAML parsers and
+
-            # hand-edits routinely break into unclosed quotes, corrupting the whole
+
-            # config (GitHub #51356).
+def _load_commented_map(yaml_rt, path: Path):
-            yaml.dump(
+    """Load *path* with *yaml_rt* as a ``CommentedMap`` (empty when missing/blank)."""
-                data,
+    from ruamel.yaml.comments import CommentedMap
-                f,
+
-                Dumper=IndentDumper,
+    data = None
-                default_flow_style=default_flow_style,
+    if path.exists():
-                sort_keys=sort_keys,
+        with path.open("r", encoding="utf-8") as f:
-                allow_unicode=True,
+            data = yaml_rt.load(f)
-            )
+    return data if isinstance(data, CommentedMap) else CommentedMap(data or {})
            if extra_content:
                f.write(extra_content)
            f.flush()
            os.fsync(f.fileno())
        # Preserve symlinks — swap in-place on the real file (GitHub #16743).
        real_path = atomic_replace(tmp_path, path)
        real_path_obj = Path(real_path)
        _restore_file_owner(real_path_obj, original_owner)
        _restore_file_mode(real_path_obj, original_mode)
    except BaseException:
        # Match atomic_json_write: cleanup must also happen for process-level
        # interruptions before we re-raise them.
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
 def atomic_roundtrip_yaml_update(
@@ -560,31 +444,16 @@ def atomic_roundtrip_yaml_update(
 ) -> None:
    """Update one dotted YAML key while preserving comments and readable text.
-    This is intentionally narrower than :func:`atomic_yaml_write`: it is for
+    Narrower than :func:`atomic_yaml_write` on purpose: for user-edited config files where
-    user-edited config files where comments, ordering, quoting, and Unicode
+    comments, ordering, quoting and Unicode must survive a single setting mutation. Still writes
-    should survive a single setting mutation.  Writes still use the same temp
+    via temp file + fsync + atomic replace.
    file + fsync + atomic replace pattern.
    """
    from ruamel.yaml import YAML
    from ruamel.yaml.comments import CommentedMap
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
-
+    yaml_rt = _roundtrip_yaml()
-    yaml_rt = YAML(typ="rt")
+    config = _load_commented_map(yaml_rt, path)
    yaml_rt.preserve_quotes = True
    yaml_rt.allow_unicode = True
    yaml_rt.default_flow_style = False
    yaml_rt.indent(mapping=2, sequence=4, offset=2)
    if path.exists():
        with path.open("r", encoding="utf-8") as f:
            config = yaml_rt.load(f) or CommentedMap()
    else:
        config = CommentedMap()
    if not isinstance(config, CommentedMap):
        config = CommentedMap(config)
    current = config
    # Honor escaped dots and prefer existing literal dotted keys (e.g. model
@@ -612,28 +481,12 @@ def atomic_roundtrip_yaml_update(
        current = next_value
        i += consumed
-    original_mode = _preserve_file_mode(path)
+    _atomic_write(
-    original_owner = _preserve_file_owner(path)
+        path,
-    fd, tmp_path = tempfile.mkstemp(
+        lambda f: yaml_rt.dump(config, f),
        dir=str(path.parent),
        prefix=f".{path.stem}_",
-        suffix=".tmp",
+        mode=_preserve_file_mode(path),
    )
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            yaml_rt.dump(config, f)
            f.flush()
            os.fsync(f.fileno())
        real_path = atomic_replace(tmp_path, path)
        real_path_obj = Path(real_path)
        _restore_file_owner(real_path_obj, original_owner)
        _restore_file_mode(real_path_obj, original_mode)
    except BaseException:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
 def atomic_roundtrip_yaml_save(
@@ -642,34 +495,13 @@ def atomic_roundtrip_yaml_save(
 ) -> None:
    """Persist a full config-state dict while preserving comments and ordering.
-    Behaves like ``atomic_yaml_write`` (writes the whole file in one shot from
+    Behaves like ``atomic_yaml_write`` (writes the whole file in one shot from ``new_state``), but
-    ``new_state``), but routes through ruamel.yaml round-trip mode so existing
+    routes through ruamel.yaml round-trip mode so existing comments, key order, quotes, and readable
-    comments, key order, quotes, and readable Unicode survive.
+    Unicode survive.
-    Reconciliation rules against the on-disk YAML:
+    This is the comment-safe replacement for ``yaml.safe_dump(cfg, f)`` in callers that mutate a
-
+    deep-loaded config dict and want to persist the whole thing.
    * Keys present in both are updated in-place via assignment, which keeps
      ruamel's CommentedMap anchors (and their attached comments) attached to
      their original positions.
    * Keys missing from ``new_state`` are deleted.
    * Keys added in ``new_state`` are appended at the end of their parent map.
    * Nested ``dict`` values recurse with the same rules.
    * Non-dict values (lists, scalars) are overwritten wholesale — list
      element comments are not individually preserved, matching ruamel's
      semantics.
    This is the comment-safe replacement for ``yaml.safe_dump(cfg, f)`` in
    callers that mutate a deep-loaded config dict and want to persist the
    whole thing.
    Shares the fail-closed contract ``hermes_cli.config.atomic_config_write``
    enforces for plain (non-comment-preserving) full-document writes: an
    existing-but-unreadable ``config.yaml`` (permission error, broken mount,
    transient I/O) raises rather than being silently replaced with only
    ``new_state``. Imported lazily to avoid a module-level circular import —
    ``hermes_cli.config`` itself imports from this module.
    """
    from ruamel.yaml import YAML
    from ruamel.yaml.comments import CommentedMap
    from ruamel.yaml.scalarstring import DoubleQuotedScalarString
@@ -678,20 +510,8 @@ def atomic_roundtrip_yaml_save(
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    require_readable_config_before_write(path)
-
+    yaml_rt = _roundtrip_yaml()
-    yaml_rt = YAML(typ="rt")
+    existing = _load_commented_map(yaml_rt, path)
    yaml_rt.preserve_quotes = True
    yaml_rt.allow_unicode = True
    yaml_rt.default_flow_style = False
    yaml_rt.indent(mapping=2, sequence=4, offset=2)
    if path.exists():
        with path.open("r", encoding="utf-8") as f:
            existing = yaml_rt.load(f)
        if not isinstance(existing, CommentedMap):
            existing = CommentedMap(existing or {})
    else:
        existing = CommentedMap()
    # ruamel's round-trip dumper resolves plain scalars against the YAML 1.2
    # core schema, where only true/false/null are reserved words — so a plain
@@ -702,9 +522,7 @@ def atomic_roundtrip_yaml_save(
    # `approvals.mode: off` silently round-trips back as `False` under
    # yaml.safe_load. Force-quote any new string value that YAML 1.1 would
    # otherwise misparse as bool/null.
-    _YAML11_AMBIGUOUS_WORDS = {
+    _YAML11_AMBIGUOUS_WORDS = {"y", "n", "yes", "no", "true", "false", "on", "off", "null", "~"}
        "y", "n", "yes", "no", "true", "false", "on", "off", "null", "~",
    }
    def _quote_if_yaml11_ambiguous(value):
        if isinstance(value, str) and value.lower() in _YAML11_AMBIGUOUS_WORDS:
@@ -725,45 +543,24 @@ def atomic_roundtrip_yaml_save(
        # Delete keys missing from src — preserves "explicit absence" semantics
        # of the old _save_cfg(cfg) pattern (e.g. cfg.pop("custom_prompt", None)
        # then _save_cfg must actually remove the key from disk).
-        for key in [k for k in dst.keys() if k not in src]:
+        for key in [k for k in dst if k not in src]:
            del dst[key]
    _merge(existing, new_state)
-    original_mode = _preserve_file_mode(path)
+    _atomic_write(
-    original_owner = _preserve_file_owner(path)
+        path,
-    fd, tmp_path = tempfile.mkstemp(
+        lambda f: yaml_rt.dump(existing, f),
        dir=str(path.parent),
        prefix=f".{path.stem}_",
-        suffix=".tmp",
+        mode=_preserve_file_mode(path),
    )
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            yaml_rt.dump(existing, f)
            f.flush()
            os.fsync(f.fileno())
        real_path = atomic_replace(tmp_path, path)
        real_path_obj = Path(real_path)
        _restore_file_owner(real_path_obj, original_owner)
        _restore_file_mode(real_path_obj, original_mode)
    except BaseException:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
 # ─── JSON Helpers ─────────────────────────────────────────────────────────────
 def safe_json_loads(text: str, default: Any = None) -> Any:
-    """Parse JSON, returning *default* on any parse error.
+    """Parse JSON, returning *default* on any parse error."""
    Replaces the ``try: json.loads(x) except (JSONDecodeError, TypeError)``
    pattern duplicated across display.py, anthropic_adapter.py,
    auxiliary_client.py, and others.
    """
    try:
        return json.loads(text)
    except (json.JSONDecodeError, TypeError, ValueError):
@@ -777,50 +574,41 @@ def safe_json_loads(text: str, default: Any = None) -> Any:
 # manifest with the slow path, costing ~0.9s of cold-start time. The C loader
 # is a true drop-in for ``safe_load`` (same restricted tag set), so prefer it
 # and fall back to the pure-Python loader only when libyaml isn't compiled in.
-_fast_yaml_loader = None
+_fast_yaml_loader = getattr(yaml, "CSafeLoader", None) or yaml.SafeLoader
 def _get_fast_yaml_loader():
    global _fast_yaml_loader
    if _fast_yaml_loader is None:
        _fast_yaml_loader = getattr(yaml, "CSafeLoader", None) or yaml.SafeLoader
    return _fast_yaml_loader
 def fast_safe_load(stream: Any) -> Any:
    """``yaml.safe_load`` using the libyaml C loader when available.
-    Accepts the same inputs as ``yaml.safe_load`` (a ``str``/``bytes`` document
+    Accepts the same inputs as ``yaml.safe_load`` (a ``str``/``bytes`` document or a readable file
-    or a readable file object) and returns the same parsed structure. Falls
+    object) and returns the same parsed structure. Falls back to PyYAML's pure-Python ``SafeLoader``
-    back to PyYAML's pure-Python ``SafeLoader`` when ``CSafeLoader`` isn't
+    when ``CSafeLoader`` isn't available, so behavior is identical everywhere — only the speed
-    available, so behavior is identical everywhere — only the speed differs.
+    differs.
    """
-    return yaml.load(stream, Loader=_get_fast_yaml_loader())
+    return yaml.load(stream, Loader=_fast_yaml_loader)
 # ─── Environment Variable Helpers ─────────────────────────────────────────────
-def env_int(key: str, default: int = 0) -> int:
+def _env_number(key: str, default, cast):
    """Read an environment variable as an integer, with fallback."""
    raw = os.getenv(key, "").strip()
    if not raw:
        return default
    try:
-        return int(raw)
+        return cast(raw)
    except (ValueError, TypeError):
        return default
 def env_int(key: str, default: int = 0) -> int:
    """Read an environment variable as an integer, with fallback."""
    return _env_number(key, default, int)
 def env_float(key: str, default: float = 0.0) -> float:
    """Read an environment variable as a float, with fallback."""
-    raw = os.getenv(key, "").strip()
+    return _env_number(key, default, float)
    if not raw:
        return default
    try:
        return float(raw)
    except (ValueError, TypeError):
        return default
 def env_bool(key: str, default: bool = False) -> bool:
@@ -840,16 +628,13 @@ _PROXY_ENV_KEYS = (
 def normalize_proxy_url(proxy_url: str | None) -> str | None:
    """Normalize proxy URLs for httpx/aiohttp compatibility.
-    WSL/Clash-style environments often export SOCKS proxies as
+    WSL/Clash-style environments export SOCKS proxies as ``socks://host:port``; httpx rejects
-    ``socks://127.0.0.1:PORT``. httpx rejects that alias and expects the
+    that alias and needs the explicit ``socks5://`` scheme.
    explicit ``socks5://`` scheme instead.
    """
    candidate = str(proxy_url or "").strip()
    if not candidate:
        return None
    if candidate.lower().startswith("socks://"):
        return f"socks5://{candidate[len('socks://'):]}"
-    return candidate
+    return candidate or None
 def normalize_proxy_env_vars() -> None:
@@ -864,21 +649,23 @@ def normalize_proxy_env_vars() -> None:
 # ─── URL Parsing Helpers ──────────────────────────────────────────────────────
 def _parse_base_url(base_url: str):
    """``urlparse`` that tolerates a bare ``host[:port][/path]`` (no scheme)."""
    raw = (base_url or "").strip()
    if not raw:
        return None
    return urlparse(raw if "://" in raw else f"//{raw}")
 def base_url_hostname(base_url: str) -> str:
    """Return the lowercased hostname for a base URL, or ``""`` if absent.
-    Use exact-hostname comparisons against known provider hosts
+    Compare exact hostnames against known provider hosts instead of substring-matching the raw
-    (``api.openai.com``, ``api.x.ai``, ``api.anthropic.com``) instead of
+    URL: substring checks treat ``https://api.openai.com.example/v1`` or
-    substring matches on the raw URL. Substring checks treat attacker- or
+    ``https://proxy.test/api.openai.com/v1`` as native endpoints, mis-routing api_mode and auth.
    proxy-controlled paths/hosts like ``https://api.openai.com.example/v1``
    or ``https://proxy.test/api.openai.com/v1`` as native endpoints, which
    leads to wrong api_mode / auth routing.
    """
-    raw = (base_url or "").strip()
+    parsed = _parse_base_url(base_url)
-    if not raw:
+    return (parsed.hostname or "").lower().rstrip(".") if parsed else ""
        return ""
    parsed = urlparse(raw if "://" in raw else f"//{raw}")
    return (parsed.hostname or "").lower().rstrip(".")
 # ─── Model Capability Detection ──────────────────────────────────────────────
@@ -887,58 +674,27 @@ def base_url_hostname(base_url: str) -> str:
 def model_forces_max_completion_tokens(model: str) -> bool:
    """Return True for model families that require ``max_completion_tokens``.
-    OpenAI's newer families reject ``max_tokens`` on /v1/chat/completions with
+    OpenAI's newer families reject ``max_tokens`` on /v1/chat/completions with HTTP 400
-    HTTP 400 ``unsupported_parameter`` — the caller must send
+    ``unsupported_parameter`` — the caller must send ``max_completion_tokens`` instead. This covers:
    ``max_completion_tokens`` instead. This covers:
    - ``gpt-4o`` / ``gpt-4o-mini`` / ``gpt-4o-*``
    - ``gpt-4.1`` / ``gpt-4.1-*``
    - ``gpt-5`` / ``gpt-5.x`` / ``gpt-5-*``
    - ``o1`` / ``o1-*``
    - ``o3`` / ``o3-*``
    - ``o4`` / ``o4-*``
    Handles vendor prefixes like ``openai/gpt-5.4`` by stripping to the tail.
    The URL-based check (``base_url_hostname == "api.openai.com"``) misses
    third-party OpenAI-compatible endpoints (custom OpenAI gateways,
    OpenRouter) that front these models and enforce the same parameter
    constraint, so name-based detection is required as a fallback.
    """
-    m = (model or "").strip().lower()
+    m = (model or "").strip().lower().rsplit("/", 1)[-1]
-    if not m:
+    return m.startswith(("gpt-4o", "gpt-4.1", "gpt-5", "o1", "o3", "o4"))
        return False
    if "/" in m:
        m = m.rsplit("/", 1)[-1]
    return (
        m.startswith("gpt-4o")
        or m.startswith("gpt-4.1")
        or m.startswith("gpt-5")
        or m.startswith("o1")
        or m.startswith("o3")
        or m.startswith("o4")
    )
 def base_url_origin(base_url: str) -> tuple[str, str, int]:
    """Return ``(scheme, hostname, effective_port)`` for a base URL.
-    Origin, not just host. ``https://h/v1`` and ``http://h/v1`` are different
+    Origin, not just host: ``https://h`` vs ``http://h`` and two ports on one host are different
-    trust boundaries, and so are two ports on the same host, so any decision
+    trust boundaries, so any decision to hand a bearer secret to a new URL must compare all
-    about handing a bearer secret to a new URL has to compare all three —
+    three — hostname alone would authorise an HTTPS→HTTP downgrade. The port defaults to 443/80
-    hostname equality alone would authorise an HTTPS→HTTP downgrade.
+    when absent so ``https://h`` equals ``https://h:443``. Returns ``("", "", 0)`` on no
-
+    hostname or a bad port.
    The port is normalised to the scheme default (443/80) when absent, so
    ``https://h`` and ``https://h:443`` compare equal. Returns
    ``("", "", 0)`` when the URL yields no usable hostname or a bad port.
    """
-    raw = (base_url or "").strip()
+    parsed = _parse_base_url(base_url)
-    if not raw:
+    hostname = (parsed.hostname or "").lower().rstrip(".") if parsed else ""
        return ("", "", 0)
    parsed = urlparse(raw if "://" in raw else f"//{raw}")
    scheme = (parsed.scheme or "").lower()
    hostname = (parsed.hostname or "").lower().rstrip(".")
    if not hostname:
        return ("", "", 0)
    scheme = (parsed.scheme or "").lower()
    try:
        port = parsed.port
    except ValueError:
@@ -952,19 +708,10 @@ def base_url_origin(base_url: str) -> tuple[str, str, int]:
 def base_url_host_matches(base_url: str, domain: str) -> bool:
    """Return True when the base URL's hostname is ``domain`` or a subdomain.
-    Safer counterpart to ``domain in base_url``, which is the substring
+    Safer counterpart to ``domain in base_url``, which has the substring false-positive class
-    false-positive class documented on ``base_url_hostname``. Accepts bare
+    noted on ``base_url_hostname`` (``evil.com/moonshot.ai`` or ``moonshot.ai.evil`` must not
-    hosts, full URLs, and URLs with paths.
+    match). Accepts bare hosts, full URLs, and URLs with paths.
        base_url_host_matches("https://api.moonshot.ai/v1", "moonshot.ai") == True
        base_url_host_matches("https://moonshot.ai", "moonshot.ai")        == True
        base_url_host_matches("https://evil.com/moonshot.ai/v1", "moonshot.ai") == False
        base_url_host_matches("https://moonshot.ai.evil/v1", "moonshot.ai")     == False
    """
    hostname = base_url_hostname(base_url)
    if not hostname:
        return False
    domain = (domain or "").strip().lower().rstrip(".")
-    if not domain:
+    return bool(hostname and domain) and (hostname == domain or hostname.endswith("." + domain))
        return False
    return hostname == domain or hostname.endswith("." + domain)