Merge branch 'simp/r3-32' into simp/integration3

tools/delegation_live_log.py

@@ -112,8 +112,7 @@ class LiveTranscriptWriter:
             logger.debug("Live transcript write failed (%s): %s", self.path, exc)
 
     def _line(self, role: str, text: str, limit: int) -> None:
-        t = _one_line(text, limit)
-        if t:
+        if t := _one_line(text, limit):
             self.event(role, t)
 
     def assistant_text(self, text: str) -> None:
@@ -177,8 +176,8 @@ class LiveTranscriptWriter:
         "subagent.start": lambda s, n, p, a, kw: s.event("start", _one_line(p, _KICKOFF_MAX)),
         "subagent.complete": _on_complete}
 
-    def observe(self, event_type: Any, tool_name: Any = None,
-                preview: Any = None, args: Any = None, **kwargs: Any) -> None:
+    def observe(self, event_type: Any, tool_name: Any = None, preview: Any = None,
+                args: Any = None, **kwargs: Any) -> None:
         """Map a child tool_progress_callback event onto transcript lines.
         Unknown events are ignored. Never raises (event() swallows I/O)."""
         handler = self._OBSERVERS.get(str(event_type or ""))
@@ -228,10 +227,7 @@ def create_live_transcripts(
     Returns ``(delegation_id, writers, paths)``; on any top-level failure
     ``(None, [None]*n, [])`` so delegation proceeds untouched."""
     n = len(task_list)
-    try:
-        prune_stale_live_dirs()
-    except Exception:
-        pass
+    prune_stale_live_dirs()  # best-effort; never raises
     try:
         # Same id shape as async_delegation's so the dir name matches the handle.
         deleg_id = delegation_id or f"deleg_{uuid.uuid4().hex[:8]}"
@@ -257,18 +253,14 @@ def _write_manifest(delegation_id: str, task_list: List[Dict[str, Any]],
                     provider: Optional[str] = None) -> None:
     try:
         _dump_json(_manifest_path(delegation_id), {
-            "delegation_id": delegation_id,
-            "started": time.strftime(_TIME_FMT),
-            "task_count": len(task_list),
-            "model": model,
-            "provider": provider,
+            "delegation_id": delegation_id, "started": time.strftime(_TIME_FMT),
+            "task_count": len(task_list), "model": model, "provider": provider,
             "tasks": [{
                 "index": i,
                 # Same mounted dir as the .log files, so the goal needs the same redaction.
                 "goal": _redact(str(t.get("goal", ""))[:500]),
                 "log": paths[i] if i < len(paths) else None,
-                "status": "running",
-            } for i, t in enumerate(task_list)]})
+                "status": "running"} for i, t in enumerate(task_list)]})
     except Exception as exc:
         logger.debug("Live transcript manifest write failed: %s", exc)
 

tools/environments/base.py

@@ -22,31 +22,18 @@ from typing import Callable, Iterable
 from hermes_constants import get_hermes_home
 from tools.interrupt import is_interrupted, is_thread_interrupted
 from tools.environments.base_output import (  # noqa: F401
-    ProcessHandle,
-    _BoundedOutputCollector,
-    _ThreadedProcessHandle,
-    _UNBOUNDED_CAPTURE_CHARS,
-    _finalize_wait_result,
-    _new_output_collector,
-    _pipe_stdin,
-    _popen_bash,
-    _start_drain_thread)
+    ProcessHandle, _BoundedOutputCollector, _ThreadedProcessHandle, _UNBOUNDED_CAPTURE_CHARS,
+    _finalize_wait_result, _new_output_collector, _pipe_stdin, _popen_bash, _start_drain_thread,
+)
 from tools.environments.base_session_env import (  # noqa: F401
-    _SHELL_ENV_NAME_RE,
-    _SNAP_TMP,
-    _SNAP_TMP_SUFFIX,
-    _SNAPSHOT_EXCLUDED_ENV_REGEX,
-    _cwd_marker,
-    _export_dump_excluding_session_vars,
-    _snapshot_bootstrap_script,
-    _split_cwd_marker,
-    _wrap_command_script)
+    _SHELL_ENV_NAME_RE, _SNAP_TMP, _SNAP_TMP_SUFFIX, _SNAPSHOT_EXCLUDED_ENV_REGEX, _cwd_marker,
+    _export_dump_excluding_session_vars, _snapshot_bootstrap_script, _split_cwd_marker,
+    _wrap_command_script,
+)
 from tools.environments.base_wait import _WaitTrace
 from tools.environments.path_utils import (  # noqa: F401
-    _SANDBOX_DIR_HASH_LEN,
-    _SANDBOX_DIR_MAX_LEN,
-    _SANDBOX_DIR_UNSAFE_RE,
-    sanitize_task_id_for_path)
+    _SANDBOX_DIR_HASH_LEN, _SANDBOX_DIR_MAX_LEN, _SANDBOX_DIR_UNSAFE_RE, sanitize_task_id_for_path,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -183,10 +170,7 @@ class BaseEnvironment(ABC):
         # so execute() must fall back to non-login ``bash -c``, not ``bash -l``.
         self._prefer_nonlogin = False
 
-    # ------------------------------------------------------------------
-    # Abstract methods
-    # ------------------------------------------------------------------
-
+    # --- Abstract methods ---
     def _run_bash(
         self, cmd_string: str, *, login: bool = False, timeout: int = 120, stdin_data: str | None = None,
     ) -> ProcessHandle:
@@ -198,10 +182,7 @@ class BaseEnvironment(ABC):
         """Release backend resources (container, instance, connection)."""
         ...
 
-    # ------------------------------------------------------------------
-    # Session snapshot (init_session)
-    # ------------------------------------------------------------------
-
+    # --- Session snapshot (init_session) ---
     def _additional_profile_scoped_passthrough_names(self) -> Iterable[str]:
         """Return backend-specific names that must not persist in snapshots."""
         return ()
@@ -258,8 +239,7 @@ class BaseEnvironment(ABC):
                     self._session_id, exc)
             else:
                 logger.warning(
-                    "init_session failed (session=%s): %s — "
-                    "falling back to bash -l per command",
+                    "init_session failed (session=%s): %s — falling back to bash -l per command",
                     self._session_id, detail)
 
     def _probe_nonlogin_fallback(self, detail: str) -> tuple[bool, str]:
@@ -275,10 +255,7 @@ class BaseEnvironment(ABC):
         except Exception as probe_exc:
             return False, f"{detail}; non-login probe: {probe_exc}"
 
-    # ------------------------------------------------------------------
-    # Command wrapping
-    # ------------------------------------------------------------------
-
+    # --- Command wrapping ---
     @staticmethod
     def _quote_cwd_for_cd(cwd: str) -> str:
         """Quote a ``cd`` target while preserving ``~`` expansion (``~/...``
@@ -310,29 +287,24 @@ class BaseEnvironment(ABC):
         delimiter = f"HERMES_STDIN_{uuid.uuid4().hex[:12]}"
         return f"{command} << '{delimiter}'\n{stdin_data}\n{delimiter}"
 
-    # ------------------------------------------------------------------
-    # Process lifecycle
-    # ------------------------------------------------------------------
-
+    # --- Process lifecycle ---
     def _wait_for_process(
         self, proc: ProcessHandle, timeout: int = 120, *,
         bounded_capture: bool = False, watch_interrupt_tid: int | None = None) -> dict:
         """Poll-based wait with interrupt checking and stdout draining (shared, not overridden).
-
         ``bounded_capture=True`` (foreground terminal-tool path only) retains at most
         ``tool_output.max_bytes`` in a head/tail window so a verbose subprocess cannot OOM the
-        process; the default keeps full fidelity for internal consumers where truncation
-        corrupts data. Fires the activity callback every 10s so the gateway's inactivity
-        timeout doesn't kill long commands. ``watch_interrupt_tid`` is the tool-worker thread
-        that submitted this wait: ``execute()`` may move the wait onto a ``run_bounded_sync``
-        worker while ``/stop`` still interrupts the original tid, so both bits are honored.
-        ``KeyboardInterrupt``/``SystemExit`` mid-poll kills the process first — the local
-        backend spawns into its own process group, so an unkilled child would be orphaned.
-        """
+        process; the default keeps full fidelity for internal consumers. Fires the activity
+        callback every 10s so the gateway's inactivity timeout doesn't kill long commands.
+        ``watch_interrupt_tid`` is the tool-worker thread that submitted this wait: ``execute()``
+        may move the wait onto a ``run_bounded_sync`` worker while ``/stop`` still interrupts the
+        original tid, so both bits are honored. ``KeyboardInterrupt``/``SystemExit`` mid-poll
+        kills the process first — the local backend spawns into its own process group, so an
+        unkilled child would be orphaned."""
         output = _new_output_collector(proc, bounded_capture)
         drain_thread = _start_drain_thread(proc, output)
-        deadline = time.monotonic() + timeout
         _now = time.monotonic()
+        deadline = _now + timeout
         _activity_state = {"last_touch": _now, "start": _now}
         trace = _WaitTrace(proc, timeout, enabled=_DEBUG_INTERRUPT, logger=logger)
         trace.enter()
@@ -388,10 +360,8 @@ class BaseEnvironment(ABC):
             stdin_thread.join(timeout=5)
         rendered = output.render()
         result = self._finalize_wait_result(output, rendered, proc.returncode)
-        stdin_errors = getattr(proc, "_hermes_stdin_errors", None)
-        if stdin_errors:
-            err = str(stdin_errors[0])
-            result["stdin_error"] = err
+        if stdin_errors := getattr(proc, "_hermes_stdin_errors", None):
+            result["stdin_error"] = err = str(stdin_errors[0])
             result["output"] = rendered + f"\n[stdin write failed: {err}]"
         return result
 
@@ -404,10 +374,7 @@ class BaseEnvironment(ABC):
         except (ProcessLookupError, PermissionError, OSError):
             pass
 
-    # ------------------------------------------------------------------
-    # CWD extraction
-    # ------------------------------------------------------------------
-
+    # --- CWD extraction ---
     def _update_cwd(self, result: dict):
         """Extract CWD from command output. Override for local file-based read."""
         self._extract_cwd_from_output(result)
@@ -428,19 +395,13 @@ class BaseEnvironment(ABC):
             result["cwd"] = cwd_path
         result["output"] = cleaned
 
-    # ------------------------------------------------------------------
-    # Hooks
-    # ------------------------------------------------------------------
-
+    # --- Hooks ---
     def _before_execute(self) -> None:
         """Hook before each command. Remote backends (SSH, Modal, Daytona)
         trigger their FileSyncManager here; bind-mount backends and Local don't."""
         pass
 
-    # ------------------------------------------------------------------
-    # Unified execute()
-    # ------------------------------------------------------------------
-
+    # --- Unified execute() ---
     def execute(
         self,
         command: str,
@@ -450,15 +411,12 @@ class BaseEnvironment(ABC):
         stdin_data: str | None = None,
         rewrite_compound_background: bool = True,
         bounded_capture: bool = False) -> dict:
-        """Execute a command, return {"output": str, "returncode": int}.
-
-        ``bounded_capture=True`` caps retention at ``tool_output.max_bytes`` WHILE draining;
-        only the foreground terminal tool may set it — internal full-fidelity consumers
-        (file-op ``cat`` reads feeding the patch engine, RPC reads, log reads) MUST leave it
-        False or data is corrupted. The wait is bounded by ``agent.deadline.run_bounded_sync``
-        so a wedged poll loop cannot hang past ``timeout`` and silently disable every asyncio
-        timer in the process.
-        """
+        """Execute a command, return {"output": str, "returncode": int}. ``bounded_capture=True``
+        caps retention at ``tool_output.max_bytes`` WHILE draining; only the foreground terminal
+        tool may set it — internal full-fidelity consumers (file-op ``cat`` reads feeding the
+        patch engine, RPC reads, log reads) MUST leave it False or data is corrupted. The wait is
+        bounded by ``agent.deadline.run_bounded_sync`` so a wedged poll loop cannot hang past
+        ``timeout`` and silently disable every asyncio timer in the process."""
         self._before_execute()
 
         exec_command, sudo_stdin = self._prepare_command(command)
@@ -508,10 +466,10 @@ class BaseEnvironment(ABC):
         from agent.deadline import run_bounded_sync
 
         try:
-            bound_s = float(effective_timeout) + _EXECUTE_WAIT_BOUND_GRACE_S
+            bound_s = float(effective_timeout)
         except (TypeError, ValueError):
-            # A non-numeric timeout must not disable the backstop; use the 120s default.
-            bound_s = 120.0 + _EXECUTE_WAIT_BOUND_GRACE_S
+            bound_s = 120.0  # a non-numeric timeout must not disable the backstop
+        bound_s += _EXECUTE_WAIT_BOUND_GRACE_S
 
         try:
             bounded = run_bounded_sync(
@@ -520,11 +478,9 @@ class BaseEnvironment(ABC):
             _on_timeout()
             raise
 
-        if bounded.timed_out:
-            timeout_msg = f"\n[Command timed out after {effective_timeout}s]"
-            result = {"output": timeout_msg.lstrip(), "returncode": 124}
-        else:
-            result = bounded.value
+        result = (
+            {"output": f"[Command timed out after {effective_timeout}s]", "returncode": 124}
+            if bounded.timed_out else bounded.value)
         self._update_cwd(result)
         return result
 
@@ -543,10 +499,7 @@ class BaseEnvironment(ABC):
         except Exception:
             logger.debug("terminal wait-bound kill_process_tree failed", exc_info=True)
 
-    # ------------------------------------------------------------------
-    # Shared helpers
-    # ------------------------------------------------------------------
-
+    # --- Shared helpers ---
     def stop(self):
         """Alias for cleanup (compat with older callers)."""
         self.cleanup()

tools/environments/base_output.py

@@ -128,8 +128,7 @@ class _BoundedOutputCollector:
             chunk = text[start:]
             self._tail.append(chunk)
             self._tail_chars += len(chunk)
-            while self._tail_chars > self._tail_limit:
-                excess = self._tail_chars - self._tail_limit
+            while (excess := self._tail_chars - self._tail_limit) > 0:
                 first = self._tail[0]
                 if len(first) <= excess:
                     self._tail.popleft()
@@ -154,9 +153,7 @@ class _BoundedOutputCollector:
             # the notice length; iterate to a fixed point (converges quickly).
             notice = ""
             for _ in range(4):
-                content_budget = max(0, available - len(notice))
-                head_chars = int(content_budget * 0.4)
-                omitted = max(0, self._total_chars - content_budget)
+                omitted = max(0, self._total_chars - max(0, available - len(notice)))
                 updated = (
                     f"\n\n... [OUTPUT TRUNCATED - {omitted:,} chars omitted "
                     f"out of {self._total_chars:,} total] ...\n\n")
@@ -211,14 +208,9 @@ def _finalize_wait_result(collector: _BoundedOutputCollector, rendered: str, ret
     return result
 
 
-# ---------------------------------------------------------------------------
-# Stdin / spawn helpers
-# ---------------------------------------------------------------------------
-
-
+# --- Stdin / spawn helpers ---
 def _pipe_stdin(proc: subprocess.Popen, data: str) -> None:
     """Write *data* to proc.stdin on a daemon thread to avoid pipe-buffer deadlocks.
-
     Writes go through ``proc.stdin.buffer`` as UTF-8 bytes we encode ourselves: Windows
     text-mode stdin would translate ``\\n`` -> ``\\r\\n`` and corrupt every write_file/patch
     payload. Encoding uses ``surrogateescape`` (exact inverse of the read-side decode);
@@ -275,11 +267,7 @@ def _popen_bash(cmd: list[str], stdin_data: str | None = None, **kwargs) -> subp
     return proc
 
 
-# ---------------------------------------------------------------------------
-# ProcessHandle protocol
-# ---------------------------------------------------------------------------
-
-
+# --- ProcessHandle protocol ---
 class ProcessHandle(Protocol):
     """Duck type every backend's _run_bash() must return. subprocess.Popen satisfies this
     natively; SDK backends (Modal, Daytona) return _ThreadedProcessHandle."""
@@ -304,7 +292,6 @@ class _ThreadedProcessHandle:
         self._cancel_fn = cancel_fn
         self._done = threading.Event()
         self._returncode: int | None = None
-        self._error: Exception | None = None
 
         # Pipe for stdout — the drain thread in _wait_for_process reads the read end.
         read_fd, write_fd = os.pipe()
@@ -319,8 +306,7 @@ class _ThreadedProcessHandle:
                     os.write(self._write_fd, output.encode("utf-8", errors="replace"))
                 except OSError:
                     pass
-            except Exception as exc:
-                self._error = exc
+            except Exception:
                 self._returncode = 1
             finally:
                 try:
@@ -354,14 +340,9 @@ class _ThreadedProcessHandle:
         return self._returncode
 
 
-# ---------------------------------------------------------------------------
-# Stdout drain thread
-# ---------------------------------------------------------------------------
-
-
+# --- Stdout drain thread ---
 def _drain_stdout(proc: ProcessHandle, output: _BoundedOutputCollector) -> None:
     """Drain ``proc.stdout`` into *output* until EOF or shortly after exit.
-
     ``for line in proc.stdout`` would block on ``readline()`` until EOF, and a backgrounded
     grandchild (``cmd &``, ``setsid cmd & disown``) inherits the pipe's write end — so the
     tool would hang for the grandchild's lifetime. Instead we ``select()`` with a short poll
@@ -376,10 +357,9 @@ def _drain_stdout(proc: ProcessHandle, output: _BoundedOutputCollector) -> None:
     if stream is None:
         return
     decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
-    fileno = getattr(stream, "fileno", None)
     try:
-        fd = fileno() if callable(fileno) else None
-    except Exception:
+        fd = stream.fileno()
+    except Exception:  # mocks / in-memory adapters without a real descriptor
         fd = None
     try:
         if not isinstance(fd, int) or fd < 0:

tools/environments/base_session_env.py

@@ -36,50 +36,42 @@ def _cwd_marker_printf(marker: str) -> str:
 
 
 def _export_dump_excluding_session_vars(tmp_path: str, excluded_names: Iterable[str] = ()) -> str:
-    """Shell snippet dumping ``export -p`` to *tmp_path* minus the per-session bridged vars
-    (see ``_SNAPSHOT_EXCLUDED_ENV_REGEX``) and *excluded_names*.
-
-    The vars are ``unset`` in a subshell BEFORE ``export -p``: a line-based ``grep -vE`` is
-    unsafe because bash 3.2 prints a value containing a newline as a multi-line ``declare -x``
-    block, so smuggled continuation lines would survive into the snapshot and execute on the
-    next ``source``. ``|| true`` keeps the success contract. The dump is a brace group with the
-    redirection on the group: *tmp_path* is usually a shell-variable expansion, and a redirect
-    on a pipeline segment would expand it inside that segment's subshell, inconsistently with
-    the parent that expands the follow-up ``mv``.
-    """
+    """Shell snippet dumping ``export -p`` to *tmp_path* minus the per-session bridged vars (see
+    ``_SNAPSHOT_EXCLUDED_ENV_REGEX``) and *excluded_names*. The vars are ``unset`` in a subshell
+    BEFORE ``export -p``: a line-based ``grep -vE`` is unsafe because bash 3.2 prints a value
+    containing a newline as a multi-line ``declare -x`` block, so smuggled continuation lines would
+    survive into the snapshot and execute on the next ``source``. ``|| true`` keeps the success
+    contract. The dump is a brace group with the redirection on the group: *tmp_path* is usually a
+    shell-variable expansion, and a redirect on a pipeline segment would expand it inside that
+    segment's subshell, inconsistently with the parent that expands the follow-up ``mv``."""
     # ${!PREFIX*} is bash 3.2+ name-prefix expansion; empty matches are ignored
     # under 2>/dev/null. Caller names are quoted so malformed config can never
     # become shell syntax (valid names stay unquoted by shlex.quote()).
     safe_names = {name for name in excluded_names if isinstance(name, str) and name}
     extra_unset = "".join(f" {shlex.quote(name)}" for name in sorted(safe_names))
     return (
-        "{ ( "
-        "unset ${!HERMES_SESSION_*} ${!HERMES_CRON_AUTO_DELIVER_*} "
+        "{ ( unset ${!HERMES_SESSION_*} ${!HERMES_CRON_AUTO_DELIVER_*} "
         "${!HERMES_BROWSER_CONTROL_*} "
         # AI_AGENT / HERMES_AGENT are per-command attribution markers re-exported
         # by every wrapper with ${VAR:-default} semantics; persisting them would
         # let the FIRST command's value override a later outer-harness value.
         "AI_AGENT HERMES_AGENT "
         f"HERMES_UI_SESSION_ID{extra_unset} 2>/dev/null; "
-        "export -p; "
-        ") || true; } "
+        "export -p; ) || true; } "
         f"> {tmp_path}")
 
 
 def _snapshot_bootstrap_script(
     *, quoted_cwd: str, quoted_snap: str, snap_tmp_template: str, excluded_names: Iterable[str], cwd_marker: str,
 ) -> str:
-    """Login-shell bootstrap that captures env/functions/aliases into the snapshot.
-
-    Atomic publish: assemble in a ``mktemp`` file, then ``mv`` over the final path so a
-    concurrent ``source`` never reads a half-written snapshot (``$$`` is the parent PID in
-    ``&``-launched subshells and macOS bash 3.2 lacks ``$BASHPID``, so only ``mktemp`` is
-    portable). Functions are filtered by NAME via ``declare -F`` (a line-based
-    ``declare -f | grep -v`` strips the header and leaves an orphaned body that breaks every
-    sourced command); the non-empty guard matters because bare ``declare -f`` dumps ALL
-    functions. The trailing ``cd`` restores the configured cwd after profile scripts
-    (e.g. ``cd ~``) so ``pwd -P`` reports terminal.cwd, not the profile's directory.
-    """
+    """Login-shell bootstrap that captures env/functions/aliases into the snapshot. Atomic publish:
+    assemble in a ``mktemp`` file, then ``mv`` over the final path so a concurrent ``source`` never
+    reads a half-written snapshot (``$$`` is the parent PID in ``&``-launched subshells and macOS
+    bash 3.2 lacks ``$BASHPID``, so only ``mktemp`` is portable). Functions are filtered by NAME via
+    ``declare -F`` (a line-based ``declare -f | grep -v`` strips the header and leaves an orphaned
+    body that breaks every sourced command); the non-empty guard matters because bare ``declare -f``
+    dumps ALL functions. The trailing ``cd`` restores the configured cwd after profile scripts (e.g.
+    ``cd ~``) so ``pwd -P`` reports terminal.cwd, not the profile's directory."""
     return (
         "umask 077\n"
         f"__hermes_snap_tmp=$(mktemp {snap_tmp_template}) || exit 1\n"
@@ -116,7 +108,6 @@ def _wrap_command_script(
     command: str, *, quoted_cwd: str, quoted_snap: str, snap_tmp_template: str,
     passthrough_names: Iterable[str], snapshot_ready: bool, cwd_marker: str) -> str:
     """Per-command bash script: source snapshot, cd, run, re-dump env, emit CWD marker.
-
     ``source`` stdout goes to /dev/null because macOS bash 3.2 / some Homebrew builds echo
     ``declare -x`` lines when sourcing. AI_AGENT/HERMES_AGENT advertise the harness to remote
     backends (whose env is not inherited); ``${VAR:-default}`` never clobbers an outer harness.

tools/environments/docker.py

@@ -20,18 +20,12 @@ import uuid
 from pathlib import Path
 from typing import Optional
 
-from tools.environments.base import BaseEnvironment, EnvironmentConnectionError, _popen_bash
+from tools.environments.base import BaseEnvironment, EnvironmentConnectionError, _SHELL_ENV_NAME_RE, _popen_bash
 from tools.environments.docker_egress import (  # noqa: F401 — re-exported for tests/patch targets
-    _EGRESS_LABEL_KEY,
-    _critical_egress_env_names,
-    _egress_enforce_on_docker,
-    _egress_proxy_args_for_docker,
-    _egress_reuse_fingerprint,
-    _extra_args_egress_collisions,
-    check_docker_env_collisions,
-    check_extra_args_collisions,
-    check_forward_env_collisions,
-    merge_egress_env)
+    _EGRESS_LABEL_KEY, _critical_egress_env_names, _egress_enforce_on_docker, _egress_proxy_args_for_docker,
+    _egress_reuse_fingerprint, _extra_args_egress_collisions, check_docker_env_collisions,
+    check_extra_args_collisions, check_forward_env_collisions, merge_egress_env,
+)
 from tools.environments.path_utils import sanitize_task_id_for_path
 from tools.environments.remote_common import bash_argv, run_capture
 from tools.environments.local import _HERMES_PROVIDER_ENV_BLOCKLIST, _is_hermes_internal_secret
@@ -41,12 +35,11 @@ logger = logging.getLogger(__name__)
 # Docker Desktop install paths checked when 'docker' is not in PATH
 # (macOS Intel / Apple Silicon Homebrew / app bundle).
 _DOCKER_SEARCH_PATHS = [
-    "/usr/local/bin/docker",
-    "/opt/homebrew/bin/docker",
-    "/Applications/Docker.app/Contents/Resources/bin/docker"]
+    "/usr/local/bin/docker", "/opt/homebrew/bin/docker", "/Applications/Docker.app/Contents/Resources/bin/docker",
+]
 
 _docker_executable: Optional[str] = None  # resolved once, cached
-_ENV_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+_ENV_VAR_NAME_RE = _SHELL_ENV_NAME_RE
 
 
 def _normalize_forward_env_names(forward_env: list[str] | None) -> list[str]:
@@ -122,13 +115,10 @@ def _get_active_profile_name() -> str:
 
 
 def _container_identity(shared_key: str = "") -> str:
-    """Profile label used for container reuse and orphan reaping.
-
-    Profiles are isolated by default; an explicit shared key lets trusted profiles share
-    one Docker identity. Label sanitization is lossy and reuse is label-keyed, so a digest
-    of the raw key disambiguates colliding keys. Plain profile names keep their historical
-    un-suffixed labels.
-    """
+    """Profile label used for container reuse and orphan reaping. Profiles are isolated by default; an
+    explicit shared key lets trusted profiles share one Docker identity. Label sanitization is lossy
+    and reuse is label-keyed, so a digest of the raw key disambiguates colliding keys. Plain profile
+    names keep their historical un-suffixed labels."""
     if not shared_key:
         return _sanitize_label_value(_get_active_profile_name())
     digest = hashlib.sha256(shared_key.encode("utf-8")).hexdigest()[:12]
@@ -149,13 +139,10 @@ def reap_orphan_containers(
     if profile_filter:
         filters.extend(["--filter", f"label=hermes-profile={_sanitize_label_value(profile_filter)}"])
 
-    try:
-        listing = run_capture([docker, "ps", "-a", *filters, "--format", "{{.ID}}"], timeout=15)
-    except (subprocess.TimeoutExpired, OSError) as e:
-        logger.debug("orphan reaper docker ps failed: %s", e)
-        return 0
-    if listing.returncode != 0:
-        logger.debug("orphan reaper docker ps returned %d: %s", listing.returncode, listing.stderr.strip())
+    listing = _docker_query(
+        [docker, "ps", "-a", *filters, "--format", "{{.ID}}"], timeout=15,
+        fail="orphan reaper docker ps failed: %s", nonzero="orphan reaper docker ps returned %d: %s")
+    if listing is None:
         return 0
 
     # Per-container inspect keeps the failure blast radius to one container.
@@ -168,29 +155,28 @@ def reap_orphan_containers(
         age = (now - finished_at).total_seconds()
         if age < max_age_seconds:
             continue
-        try:
-            result = run_capture([docker, "rm", "-f", cid], timeout=30)
-            if result.returncode == 0:
-                removed += 1
-                logger.info("Reaped orphan container %s (exited %d seconds ago)", cid[:12], int(age))
-            else:
-                logger.debug("docker rm -f %s failed: %s", cid[:12], result.stderr.strip())
-        except (subprocess.TimeoutExpired, OSError) as e:
-            logger.debug("orphan reaper docker rm %s failed: %s", cid[:12], e)
+        result = _docker_query(
+            [docker, "rm", "-f", cid], timeout=30, fail="orphan reaper docker rm %s failed: %s", fail_args=(cid[:12],))
+        if result is None:
+            continue
+        if result.returncode == 0:
+            removed += 1
+            logger.info("Reaped orphan container %s (exited %d seconds ago)", cid[:12], int(age))
+        else:
+            logger.debug("docker rm -f %s failed: %s", cid[:12], result.stderr.strip())
     return removed
 
 
 def _container_finished_at(docker_exe: str, container_id: str):
     """``docker inspect`` FinishedAt as an aware datetime; ``None`` (= don't reap) when
     missing/unparseable or Docker's never-finished zero value ``0001-01-01T00:00:00Z``."""
-    try:
-        result = run_capture(
-            [docker_exe, "inspect", "--format", "{{.State.FinishedAt}}", container_id], timeout=10)
-    except (subprocess.TimeoutExpired, OSError) as e:
-        logger.debug("orphan reaper docker inspect %s failed: %s", container_id[:12], e)
+    result = _docker_query(
+        [docker_exe, "inspect", "--format", "{{.State.FinishedAt}}", container_id], timeout=10,
+        fail="orphan reaper docker inspect %s failed: %s", fail_args=(container_id[:12],))
+    if result is None or result.returncode != 0:
         return None
     raw = result.stdout.strip()
-    if result.returncode != 0 or not raw or raw.startswith("0001-01-01"):
+    if not raw or raw.startswith("0001-01-01"):
         return None
     # Docker emits RFC3339 with nanoseconds; fromisoformat only takes microseconds.
     raw = re.sub(r"(\.\d{6})\d+", r"\1", raw).replace("Z", "+00:00")
@@ -205,6 +191,24 @@ def _is_executable(path: str) -> bool:
     return os.path.isfile(path) and os.access(path, os.X_OK)
 
 
+def _docker_query(
+    argv: list[str], *, timeout: float, fail: str, fail_args: tuple = (), nonzero: str | None = None,
+    exc_types: tuple = (subprocess.TimeoutExpired, OSError),
+):
+    """Best-effort ``run_capture`` for probes: logs *fail* (``*fail_args, exc``) at debug and
+    returns ``None`` when the CLI raises; with *nonzero* set, a nonzero exit also logs
+    (``*fail_args, returncode, stderr``) and returns ``None``."""
+    try:
+        result = run_capture(argv, timeout=timeout)
+    except exc_types as e:
+        logger.debug(fail, *fail_args, e)
+        return None
+    if nonzero is not None and result.returncode != 0:
+        logger.debug(nonzero, *fail_args, result.returncode, result.stderr.strip())
+        return None
+    return result
+
+
 def find_docker() -> Optional[str]:
     """Locate the docker/podman CLI (cached): ``HERMES_DOCKER_BINARY`` override, ``docker``
     on PATH, ``podman`` on PATH, then macOS Docker Desktop locations; ``None`` if absent."""
@@ -281,18 +285,12 @@ def _image_uses_init_entrypoint(docker_exe: str, image: str) -> bool:
     """True if ``image``'s entrypoint is the s6-overlay ``/init`` (own PID 1: incompatible
     with ``--init`` and a noexec /run). Any inspection failure, including an image not yet
     pulled, returns False and keeps the hardened defaults."""
-    try:
-        result = run_capture(
-            [docker_exe, "image", "inspect", image, "--format", "{{json .Config.Entrypoint}}"], timeout=15)
-    except (subprocess.SubprocessError, OSError) as e:
-        logger.debug("Docker: could not inspect entrypoint for %s: %s", image, e)
-        return False
-    if result.returncode != 0:
-        logger.debug(
-            "Docker: image inspect for %s returned %d (stderr=%s)",
-            image, result.returncode, result.stderr.strip())
-        return False
-    raw = (result.stdout or "").strip()
+    result = _docker_query(
+        [docker_exe, "image", "inspect", image, "--format", "{{json .Config.Entrypoint}}"], timeout=15,
+        fail="Docker: could not inspect entrypoint for %s: %s", fail_args=(image,),
+        nonzero="Docker: image inspect for %s returned %d (stderr=%s)",
+        exc_types=(subprocess.SubprocessError, OSError))
+    raw = (result.stdout or "").strip() if result is not None else ""
     if not raw or raw == "null":
         return False
     try:
@@ -348,8 +346,7 @@ def _cgroup_limits_available(image: str) -> bool:
                 "Cgroup resource limits (--cpus/--memory/--pids-limit) not "
                 "available in this environment. Containers will run without "
                 "CPU, memory or PID limits. To enable, delegate the cpu, "
-                "memory and pids cgroup controllers to this container. "
-                "Probe stderr: %s",
+                "memory and pids cgroup controllers to this container. Probe stderr: %s",
                 (result.stderr or "").strip()[:500])
     except Exception as e:
         _cgroup_limits_ok = False
@@ -368,8 +365,7 @@ def _ensure_docker_available() -> None:
     if not docker_exe:
         raise _docker_unavailable(
             "Docker backend selected but no docker executable was found in PATH "
-            "or known install locations. Install Docker Desktop and ensure the "
-            "CLI is available.",
+            "or known install locations. Install Docker Desktop and ensure the CLI is available.",
             error="Docker executable not found in PATH or known install locations. "
                   "Install Docker and ensure the 'docker' command is available.",
             hint="Install Docker (or fix PATH) and retry, or switch terminal.backend to 'local'.")
@@ -414,29 +410,23 @@ _RO_MOUNT_SOURCES = (
 
 
 def _readonly_skill_mount_args() -> list[str]:
-    """``-v host:container:ro`` args for credential files, skill dirs and cache dirs.
-
-    Read-only so the container can authenticate/read but never modify host state. Missing
-    or wrong-kind sources are skipped with a warning (Docker-in-Docker auto-creates a
-    missing file source as a directory, which would exit 125).
-    """
+    """``-v host:container:ro`` args for credential files, skill dirs and cache dirs. Read-only so the
+    container can authenticate/read but never modify host state. Missing or wrong-kind sources are
+    skipped with a warning (Docker-in-Docker auto-creates a missing file source as a directory,
+    which would exit 125)."""
     args: list[str] = []
     try:
         import tools.credential_files as cf
         for getter, expects_file, noun in _RO_MOUNT_SOURCES:
             for entry in getattr(cf, getter)():
                 src = Path(entry["host_path"])
-                if expects_file and src.is_dir():
-                    logger.warning(
-                        "Docker: skipping credential mount — source is a directory "
-                        "(likely Docker-in-Docker auto-creation): %s", src)
-                    continue
-                if expects_file and not src.is_file():
-                    logger.warning("Docker: skipping credential mount — source not found: %s", src)
-                    continue
-                if not expects_file and not src.is_dir():
-                    logger.warning("Docker: skipping %s mount — source is not a directory: %s",
-                                   noun.split()[0], src)
+                if expects_file:
+                    problem = ("source is a directory (likely Docker-in-Docker auto-creation)" if src.is_dir()
+                               else None if src.is_file() else "source not found")
+                else:
+                    problem = None if src.is_dir() else "source is not a directory"
+                if problem:
+                    logger.warning("Docker: skipping %s mount — %s: %s", noun.split()[0], problem, src)
                     continue
                 args.extend(["-v", f"{entry['host_path']}:{entry['container_path']}:ro"])
                 logger.info("Docker: mounting %s %s -> %s", noun, entry["host_path"], entry["container_path"])
@@ -456,8 +446,7 @@ def _host_user_args(run_as_host_user: bool) -> list[str]:
         return ["--user", user_spec]
     logger.warning(
         "docker_run_as_host_user is enabled but this platform does "
-        "not expose POSIX uid/gid; container will start as its "
-        "image default user.")
+        "not expose POSIX uid/gid; container will start as its image default user.")
     return []
 
 
@@ -568,10 +557,7 @@ class DockerEnvironment(BaseEnvironment):
         self._init_env_args = self._build_init_env_args()
         self.init_session()
 
-    # ------------------------------------------------------------------
-    # __init__ helpers
-    # ------------------------------------------------------------------
-
+    # --- __init__ helpers ---
     def _egress_and_env_args(self, extra_args) -> tuple[str, list[str], list[str], list[str], list[str]]:
         """Egress credential-injection proxy plumbing (CA mount + HTTPS_PROXY/CA-bundle env so
         outbound traffic routes through the host-side proxy and the sandbox receives proxy tokens
@@ -603,11 +589,11 @@ class DockerEnvironment(BaseEnvironment):
     def _resource_args(self, image, cpu, memory, disk, network, shm_size, extra_args) -> list[str]:
         """cgroup-gated CPU/memory/pids limits, shm size, disk quota and network mode."""
         args: list[str] = []
-        if cpu > 0 and _cgroup_limits_available(image):
-            args.extend(["--cpus", str(cpu)])
-        if memory > 0 and _cgroup_limits_available(image):
-            args.extend(["--memory", f"{memory}m"])
         if _cgroup_limits_available(image):
+            if cpu > 0:
+                args.extend(["--cpus", str(cpu)])
+            if memory > 0:
+                args.extend(["--memory", f"{memory}m"])
             args.extend(["--pids-limit", _DEFAULT_PIDS_LIMIT])
         # --shm-size is a tmpfs option, not cgroup-gated. Skipped when the user
         # sets it in docker_extra_args or opts out with empty/"0".
@@ -629,7 +615,6 @@ class DockerEnvironment(BaseEnvironment):
         """``(volume_args, writable_args)`` for user volumes, host cwd and /workspace,/root.
         Persistent mode bind-mounts from TERMINAL_SANDBOX_DIR (default ~/.hermes/sandboxes/)."""
         volume_args: list[str] = []
-        workspace_explicitly_mounted = False
         for vol in (volumes or []):
             if not isinstance(vol, str):
                 logger.warning("Docker volume entry is not a string: %r", vol)
@@ -641,8 +626,7 @@ class DockerEnvironment(BaseEnvironment):
                 logger.warning("Docker volume '%s' missing colon, skipping", vol)
                 continue
             volume_args.extend(["-v", vol])
-            if ":/workspace" in vol:
-                workspace_explicitly_mounted = True
+        workspace_explicitly_mounted = any(":/workspace" in v for v in volume_args)
 
         host_cwd_abs = os.path.abspath(os.path.expanduser(host_cwd)) if host_cwd else ""
         bind_host_cwd = (
@@ -666,8 +650,7 @@ class DockerEnvironment(BaseEnvironment):
                 os.makedirs(self._workspace_dir, exist_ok=True)
                 writable_args += ["-v", f"{self._workspace_dir}:/workspace"]
         else:
-            if mount_workspace:
-                writable_args += ["--tmpfs", "/workspace:rw,exec,size=10g"]
+            writable_args += ["--tmpfs", "/workspace:rw,exec,size=10g"] if mount_workspace else []
             writable_args += ["--tmpfs", "/home:rw,exec,size=1g", "--tmpfs", "/root:rw,exec,size=1g"]
 
         if bind_host_cwd:
@@ -693,8 +676,7 @@ class DockerEnvironment(BaseEnvironment):
                 logger.warning(
                     "Existing container %s has NetworkMode=%s but "
                     "docker_network=false requests an air-gapped "
-                    "container — removing it and starting fresh "
-                    "(task=%s, profile=%s).",
+                    "container — removing it and starting fresh (task=%s, profile=%s).",
                     container_id[:12], actual_mode or "unknown", task_label, profile_name)
                 try:
                     run_capture([self._docker_exe, "rm", "-f", container_id], timeout=30)
@@ -703,13 +685,12 @@ class DockerEnvironment(BaseEnvironment):
                 return False
 
         if state != "running":
-            try:
-                run_capture([self._docker_exe, "start", container_id], timeout=30, check=True)
-            except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
+            err = self._start_container(container_id)
+            if err is not None:
                 logger.warning(
                     "Failed to start existing container %s (state=%s): "
                     "%s — falling back to a fresh container.",
-                    container_id[:12], state, e)
+                    container_id[:12], state, err)
                 return False
         self._container_id = container_id
         logger.info(
@@ -717,6 +698,14 @@ class DockerEnvironment(BaseEnvironment):
             container_id[:12], task_label, profile_name, state)
         return True
 
+    def _start_container(self, container_id: str) -> Exception | None:
+        """``docker start`` a stopped container; returns the failure instead of raising."""
+        try:
+            run_capture([self._docker_exe, "start", container_id], timeout=30, check=True)
+            return None
+        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
+            return e
+
     def _run_command(self, name: str, workdir: str) -> list[str]:
         """``docker run -d`` argv for a fresh ``sleep infinity`` container (idle reaper handles
         lifetime). s6-overlay images already provide PID 1, so ``--init`` is skipped for them."""
@@ -752,10 +741,7 @@ class DockerEnvironment(BaseEnvironment):
         logger.info("Started container %s (%s)", container_name, container_id[:12])
         return container_id
 
-    # ------------------------------------------------------------------
-    # Env forwarding
-    # ------------------------------------------------------------------
-
+    # --- Env forwarding ---
     def _docker_client_env(self, values: dict[str, str]) -> dict[str, str] | None:
         """Env for the docker-client subprocess carrying forwarded values (pairs with name-only
         ``-e KEY`` flags to keep secrets out of cmdline); ``None`` = inherit when empty."""
@@ -783,9 +769,8 @@ class DockerEnvironment(BaseEnvironment):
         is_global_env = lambda _name: False  # noqa: E731
         try:
             from tools.env_passthrough import get_all_passthrough, resolve_passthrough_value
-            from agent.secret_scope import _is_global_env, is_multiplex_active as _is_multiplex_active
-            is_global_env = _is_global_env
-            multiplex_active = _is_multiplex_active()
+            from agent.secret_scope import _is_global_env as is_global_env, is_multiplex_active
+            multiplex_active = is_multiplex_active()
             passthrough_keys = set(get_all_passthrough())
         except Exception:
             pass
@@ -832,19 +817,12 @@ class DockerEnvironment(BaseEnvironment):
         if unset_names:
             quoted_names = " ".join(shlex.quote(name) for name in unset_names)
             cmd_string = f"unset {quoted_names} 2>/dev/null || true\n{cmd_string}"
-
-        cmd.append(self._container_id)
-        cmd.extend(bash_argv(cmd_string, login))
+        cmd += [self._container_id, *bash_argv(cmd_string, login)]
 
         client_env = self._docker_client_env(env_values)
-        if client_env is not None:
-            return _popen_bash(cmd, stdin_data, env=client_env)
-        return _popen_bash(cmd, stdin_data)
-
-    # ------------------------------------------------------------------
-    # "No such container" recovery
-    # ------------------------------------------------------------------
+        return _popen_bash(cmd, stdin_data, env=client_env) if client_env is not None else _popen_bash(cmd, stdin_data)
 
+    # --- "No such container" recovery ---
     _NO_CONTAINER_PATTERNS = ("No such container", "is not running", "no such container")
 
     def _is_container_gone(self, output: str) -> bool:
@@ -866,13 +844,11 @@ class DockerEnvironment(BaseEnvironment):
             if state == "running":
                 self._container_id = cid
                 logger.info("Recovery: reusing running container %s", cid[:12])
+            elif (err := self._start_container(cid)) is None:
+                self._container_id = cid
+                logger.info("Recovery: restarted container %s", cid[:12])
             else:
-                try:
-                    run_capture([self._docker_exe, "start", cid], timeout=30, check=True)
-                    self._container_id = cid
-                    logger.info("Recovery: restarted container %s", cid[:12])
-                except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
-                    logger.warning("Recovery: failed to start container %s: %s", cid[:12], e)
+                logger.warning("Recovery: failed to start container %s: %s", cid[:12], err)
 
         if not self._container_id:
             if not self._image:
@@ -937,17 +913,10 @@ class DockerEnvironment(BaseEnvironment):
     def _container_network_mode(self, container_id: str) -> Optional[str]:
         """``HostConfig.NetworkMode`` of a container, or ``None`` when inspection fails (callers
         treat ``None`` as a mismatch under lockdown, so a failed inspect fails closed)."""
-        try:
-            result = run_capture(
-                [self._docker_exe, "inspect", "--format", "{{.HostConfig.NetworkMode}}", container_id],
-                timeout=10)
-        except (subprocess.TimeoutExpired, OSError) as e:
-            logger.debug("docker inspect NetworkMode failed: %s", e)
-            return None
-        if result.returncode != 0:
-            logger.debug("docker inspect NetworkMode returned %d: %s", result.returncode, result.stderr.strip())
-            return None
-        return result.stdout.strip() or None
+        result = _docker_query(
+            [self._docker_exe, "inspect", "--format", "{{.HostConfig.NetworkMode}}", container_id], timeout=10,
+            fail="docker inspect NetworkMode failed: %s", nonzero="docker inspect NetworkMode returned %d: %s")
+        return (result.stdout.strip() or None) if result is not None else None
 
     def _find_reusable_container(
         self, task_label: str, profile_label: str, egress_label: str) -> Optional[tuple[str, str]]:
@@ -966,15 +935,11 @@ class DockerEnvironment(BaseEnvironment):
         else:
             filters.extend(["--filter", f"label={_EGRESS_LABEL_KEY}={egress_label}"])
             fmt = "{{.ID}}\t{{.State}}"
-        try:
-            result = run_capture([self._docker_exe, "ps", "-a", *filters, "--format", fmt], timeout=10)
-        except (subprocess.TimeoutExpired, OSError) as e:
-            logger.debug("docker ps probe failed: %s — will start a fresh container", e)
-            return None
-        if result.returncode != 0:
-            logger.debug(
-                "docker ps probe returned %d: %s — will start a fresh container",
-                result.returncode, result.stderr.strip())
+        result = _docker_query(
+            [self._docker_exe, "ps", "-a", *filters, "--format", fmt], timeout=10,
+            fail="docker ps probe failed: %s — will start a fresh container",
+            nonzero="docker ps probe returned %d: %s — will start a fresh container")
+        if result is None:
             return None
         # Multiple matches can happen after a crash mid-cleanup: prefer a running
         # one, else the first listed; stale duplicates are the orphan reaper's job.

tools/environments/docker_egress.py

@@ -146,27 +146,20 @@ def _critical_egress_env_names(env_overrides: dict[str, str]) -> set[str]:
 def _extra_args_egress_collisions(extra_args: list[str], critical_names: set[str]) -> list[str]:
     """Return docker_extra_args entries that can override egress controls."""
     collisions: list[str] = []
-    env_flags = {"-e", "--env", "--env-file"}
-    network_flags = {"--network", "--net"}
     i = 0
     while i < len(extra_args):
         arg = extra_args[i]
-        nxt = extra_args[i + 1] if i + 1 < len(extra_args) else ""
-        if arg in env_flags:
-            if arg == "--env-file":
-                collisions.append(arg)
-            elif nxt.split("=", 1)[0] in critical_names:
-                collisions.append(nxt.split("=", 1)[0])
-            i += 2
+        flag, sep, inline_value = arg.partition("=")  # ``-e NAME=v`` vs ``-e=NAME=v`` / ``--env-file=f``
+        if flag in ("-e", "--env", "--env-file"):
+            value = inline_value if sep else (extra_args[i + 1] if i + 1 < len(extra_args) else "")
+            name = value.split("=", 1)[0]
+            if flag == "--env-file":
+                collisions.append(flag)
+            elif name in critical_names:
+                collisions.append(name)
+            i += 1 if sep else 2
             continue
-        if any(arg.startswith(f"{flag}=") for flag in env_flags):
-            if arg.startswith("--env-file="):
-                collisions.append("--env-file")
-            else:
-                name = arg.split("=", 1)[1].split("=", 1)[0]
-                if name in critical_names:
-                    collisions.append(name)
-        elif arg in network_flags or any(arg.startswith(f"{flag}=") for flag in network_flags):
+        if flag in ("--network", "--net"):
             collisions.append(arg)
         i += 1
     return sorted(set(collisions))

tools/environments/file_sync.py

@@ -211,10 +211,7 @@ class FileSyncManager:
             self._delete_fn(to_delete)
             logger.debug("file_sync: deleted %s", to_delete)
 
-    # ------------------------------------------------------------------
-    # Sync-back: pull remote changes to host on teardown
-    # ------------------------------------------------------------------
-
+    # --- Sync-back: pull remote changes to host on teardown ---
     def sync_back(self, hermes_home: Path | None = None) -> None:
         """Pull remote changes back to the host: download the remote ``.hermes/`` as a tar and
         apply only files whose SHA-256 differs from what was pushed. SIGINT is deferred until
@@ -360,8 +357,7 @@ class FileSyncManager:
         if pushed_hash is not None and os.path.exists(host_path) and _sha256_file(host_path) != pushed_hash:
             logger.warning(
                 "sync_back: conflict on %s — host modified "
-                "since push, remote also changed. Applying "
-                "remote version (last-write-wins).",
+                "since push, remote also changed. Applying remote version (last-write-wins).",
                 remote_path)
 
         os.makedirs(os.path.dirname(host_path), exist_ok=True)

tools/interrupt.py

@@ -12,17 +12,14 @@ import threading
 
 logger = logging.getLogger(__name__)
 
-# Opt-in debug tracing — pairs with HERMES_DEBUG_INTERRUPT in
-# tools/environments/base.py; logs caller/target thread and state per set/check.
+# Opt-in debug tracing — pairs with HERMES_DEBUG_INTERRUPT in tools/environments/base.py.
 _DEBUG_INTERRUPT = bool(os.getenv("HERMES_DEBUG_INTERRUPT"))
-
 if _DEBUG_INTERRUPT:
     # AIAgent's quiet_mode forces the `tools` logger to ERROR on CLI startup;
     # force ours back to INFO so the trace is visible in agent.log.
     logger.setLevel(logging.INFO)
 
-# Interrupted thread idents, plus an optional user-safe cause per signal.  The
-# cause deliberately never contains an incoming user's message text.
+# Interrupted thread idents + optional user-safe cause (never the user's message text).
 _interrupted_threads: set[int] = set()
 _interrupt_reasons: dict[int, str] = {}
 _lock = threading.Lock()
@@ -33,14 +30,10 @@ def set_interrupt(active: bool, thread_id: int | None = None, *, reason: str | N
     CLI/tests).  ``reason`` is an optional user-safe cause."""
     tid = thread_id if thread_id is not None else threading.current_thread().ident
     with _lock:
-        if active:
-            _interrupted_threads.add(tid)
-            if reason:
-                _interrupt_reasons[tid] = reason
-            else:
-                _interrupt_reasons.pop(tid, None)
+        (_interrupted_threads.add if active else _interrupted_threads.discard)(tid)
+        if active and reason:
+            _interrupt_reasons[tid] = reason
         else:
-            _interrupted_threads.discard(tid)
             _interrupt_reasons.pop(tid, None)
         _snapshot = set(_interrupted_threads) if _DEBUG_INTERRUPT else None
     if _DEBUG_INTERRUPT:

tools/kanban_tools.py

@@ -114,10 +114,9 @@ def _kanban_handler(tool_name: str) -> Callable:
                 return fn(args, **kw)
             except _Reject as e:
                 return e.args[0]
-            except ValueError as e:
-                return tool_error(f"{tool_name}: {e}")
             except Exception as e:
-                logger.exception(f"{tool_name} failed")
+                if not isinstance(e, ValueError):
+                    logger.exception(f"{tool_name} failed")
                 return tool_error(f"{tool_name}: {e}")
         return wrapper
     return deco
@@ -152,11 +151,11 @@ def _require_task_id(args: dict) -> str:
 
 def _worker_run_id(task_id: str) -> Optional[int]:
     """This worker's dispatcher run id when it is scoped to task_id."""
-    if os.environ.get("HERMES_KANBAN_TASK") != task_id:
-        return None
     raw = os.environ.get("HERMES_KANBAN_RUN_ID")
+    if os.environ.get("HERMES_KANBAN_TASK") != task_id or not raw:
+        return None
     try:
-        return int(raw) if raw else None
+        return int(raw)
     except ValueError:
         return None
 
@@ -233,6 +232,10 @@ def _redact(value: Any) -> str:
     return redact_sensitive_text(str(value), force=True)
 
 
+def _redact_opt(value: Any) -> Any:
+    return _redact(value) if value else value
+
+
 def _redact_metadata(metadata: dict) -> Optional[dict]:
     """Redact via a JSON round-trip; None if the result can't be re-parsed."""
     try:
@@ -283,38 +286,34 @@ def _require_text(args: dict, name: str, message: Optional[str] = None) -> Any:
     return value
 
 
+_BOOL_WORDS = {"true": True, "1": True, "yes": True, "false": False, "0": False, "no": False}
+
+
 def _parse_bool_arg(args: dict, name: str) -> bool:
     value = args.get(name)
-    if value is None:
-        return False
-    if isinstance(value, bool):
-        return value
-    text = str(value).strip().lower()
-    if text in {"true", "1", "yes"}:
-        return True
-    if text in {"false", "0", "no"}:
-        return False
-    raise _Reject(f"{name} must be a boolean or 'true'/'false'")
+    if value is None or isinstance(value, bool):
+        return bool(value)
+    parsed = _BOOL_WORDS.get(str(value).strip().lower())
+    _check(parsed is not None, f"{name} must be a boolean or 'true'/'false'")
+    return parsed
 
 
 def _opt_int(value: Any, default: Optional[int] = None) -> Optional[int]:
     return int(value) if value is not None else default
 
 
-_TASK_FIELDS = (
-    "id", "title", "body", "assignee", "status", "tenant", "priority", "workspace_kind",
-    "workspace_path", "created_by", "created_at", "started_at", "completed_at", "result",
-    "current_run_id", "model_override", "provider_override")
-_TASK_SUMMARY_FIELDS = (
-    "id", "title", "assignee", "status", "priority", "tenant", "workspace_kind", "workspace_path",
-    "project_id", "created_by", "created_at", "started_at", "completed_at", "current_run_id",
-    "model_override", "provider_override")
-_RUN_FIELDS = (
-    "id", "profile", "status", "outcome", "summary", "error", "metadata", "started_at", "ended_at")
+_TASK_FIELDS = tuple(
+    "id title body assignee status tenant priority workspace_kind workspace_path created_by "
+    "created_at started_at completed_at result current_run_id model_override "
+    "provider_override".split())
+_TASK_SUMMARY_FIELDS = tuple(
+    "id title assignee status priority tenant workspace_kind workspace_path project_id created_by "
+    "created_at started_at completed_at current_run_id model_override provider_override".split())
+_RUN_FIELDS = tuple("id profile status outcome summary error metadata started_at ended_at".split())
 _COMMENT_FIELDS = ("author", "body", "created_at")
 _EVENT_FIELDS = ("kind", "payload", "created_at", "run_id")
-_ATTACHMENT_FIELDS = (
-    "id", "filename", "content_type", "size", "uploaded_by", "stored_path", "created_at")
+_ATTACHMENT_FIELDS = tuple(
+    "id filename content_type size uploaded_by stored_path created_at".split())
 
 
 def _fields(obj: Any, names: tuple[str, ...]) -> dict[str, Any]:
@@ -403,22 +402,19 @@ def heartbeat_current_worker_from_env() -> bool:
     heartbeated; ``HERMES_KANBAN_CLAIM_LOCK`` absent -> default claimer (local workers)."""
     global _auto_heartbeat_last_attempt
     tid = os.environ.get("HERMES_KANBAN_TASK")
-    if not tid:
-        return False
     now = time.monotonic()
-    if (now - _auto_heartbeat_last_attempt) < _AUTO_HEARTBEAT_MIN_INTERVAL_SECONDS:
+    if not tid or (now - _auto_heartbeat_last_attempt) < _AUTO_HEARTBEAT_MIN_INTERVAL_SECONDS:
         return False
     _auto_heartbeat_last_attempt = now
     try:
         with _board(None, quiet_close=True) as (kb, conn):
-            try:
-                kb.heartbeat_claim(conn, tid, claimer=os.environ.get("HERMES_KANBAN_CLAIM_LOCK"))
-            except Exception:
-                logger.debug("auto-heartbeat: heartbeat_claim failed", exc_info=True)
-            try:
-                kb.heartbeat_worker(conn, tid, note=None, expected_run_id=_worker_run_id(tid))
-            except Exception:
-                logger.debug("auto-heartbeat: heartbeat_worker failed", exc_info=True)
+            ops = (("heartbeat_claim", {"claimer": os.environ.get("HERMES_KANBAN_CLAIM_LOCK")}),
+                   ("heartbeat_worker", {"note": None, "expected_run_id": _worker_run_id(tid)}))
+            for op, kwargs in ops:
+                try:
+                    getattr(kb, op)(conn, tid, **kwargs)
+                except Exception:
+                    logger.debug("auto-heartbeat: %s failed", op, exc_info=True)
         return True
     except Exception:
         logger.debug("auto-heartbeat: bridge failed", exc_info=True)
@@ -436,15 +432,13 @@ _comment_watermark: dict[str, int] = {}
 def inject_new_comments_from_env(agent: Any) -> bool:
     """Steer new operator comments on the worker's task into ``agent``; True iff a
     steer was injected; never raises. Own comments (``HERMES_PROFILE``) are skipped."""
-    tid = os.environ.get("HERMES_KANBAN_TASK")
-    if not tid or agent is None or not hasattr(agent, "steer"):
-        return False
     global _comment_poll_last_attempt
+    tid = os.environ.get("HERMES_KANBAN_TASK")
     now = time.monotonic()
-    if (now - _comment_poll_last_attempt) < _COMMENT_POLL_MIN_INTERVAL_SECONDS:
+    if (not tid or agent is None or not hasattr(agent, "steer")
+            or (now - _comment_poll_last_attempt) < _COMMENT_POLL_MIN_INTERVAL_SECONDS):
         return False
     _comment_poll_last_attempt = now
-
     seen = _comment_watermark.get(tid)
     try:
         with _board(None, quiet_close=True) as (kb, conn):
@@ -452,27 +446,20 @@ def inject_new_comments_from_env(agent: Any) -> bool:
     except Exception:
         logger.debug("comment-inject: bridge failed", exc_info=True)
         return False
-
     if seen is None:
         _comment_watermark[tid] = max((c.id for c in rows), default=0)
-        return False
-    if not rows:
+    if seen is None or not rows:
         return False
     # Advance past everything read (including our own notes) so nothing is re-injected.
     _comment_watermark[tid] = max(c.id for c in rows)
-
     own = (os.environ.get("HERMES_PROFILE") or "").strip()
     fresh = [c for c in rows if (c.author or "").strip() != own and (c.body or "").strip()]
     if not fresh:
         return False
-
     lines = [f"- {c.author or 'operator'}: {c.body.strip()}" for c in fresh]
-    note = (
-        "New note"
-        + ("s" if len(fresh) > 1 else "")
-        + " on your kanban task from the operator (delivered mid-run). "
-        + "Take it into account for the work you're doing right now:\n"
-        + "\n".join(lines))
+    note = ("New note" + ("s" if len(fresh) > 1 else "")
+            + " on your kanban task from the operator (delivered mid-run). "
+            + "Take it into account for the work you're doing right now:\n" + "\n".join(lines))
     try:
         return bool(agent.steer(note))
     except Exception:
@@ -535,18 +522,12 @@ def _handle_list(args: dict, **kw) -> str:
 def _handle_complete(args: dict, **kw) -> str:
     """Mark the current task done with a structured handoff."""
     tid = _worker_guard("kanban_complete", args)
-    summary = args.get("summary")
+    summary = _redact_opt(args.get("summary"))
+    result = _redact_opt(args.get("result"))
     metadata = args.get("metadata")
-    result = args.get("result")
-    if summary:
-        summary = _redact(summary)
-    if result:
-        result = _redact(result)
     if isinstance(metadata, dict):
         # Keep the unredacted dict if the redacted JSON cannot be re-parsed.
-        redacted = _redact_metadata(metadata)
-        if redacted is not None:
-            metadata = redacted
+        metadata = _redact_metadata(metadata) or metadata
     created_cards = _coerce_str_list(
         args.get("created_cards"), "created_cards", "task ids", strip=True)
     artifacts = _coerce_str_list(args.get("artifacts"), "artifacts", "file paths", strip=True)
@@ -672,11 +653,10 @@ def _handle_comment(args: dict, **kw) -> str:
     _check(tid, "task_id is required (use the current task id if that's what "
                 "you mean — pulls from env but kept explicit here)")
     body = _redact(_require_text(args, "body"))
-    # Author comes from the worker's runtime identity, never caller args:
-    # comments are injected into future workers' system prompts as
-    # ``**{author}** (timestamp): {body}``, so an args["author"] override
-    # could forge a directive from a name like ``hermes-system``. Cross-task
-    # commenting stays unrestricted — it is the handoff channel between tasks.
+    # Author comes from the worker's runtime identity, never caller args: comments are
+    # injected into future workers' system prompts, so an args["author"] override could
+    # forge a directive from ``hermes-system``. Cross-task commenting stays unrestricted —
+    # it is the handoff channel between tasks.
     author = os.environ.get("HERMES_PROFILE") or "worker"
     with _board(args.get("board")) as (kb, conn):
         cid = kb.add_comment(conn, tid, author=author, body=str(body))
@@ -722,7 +702,6 @@ def _download_url_with_cap(url: str, max_bytes: int) -> tuple[bytes, Optional[st
     from urllib.parse import urljoin, urlparse
     import httpx
     from tools.url_safety import is_safe_url
-
     current_url = url
     for _ in range(_MAX_ATTACH_URL_REDIRECTS + 1):
         scheme = (urlparse(current_url).scheme or "").lower()
@@ -756,7 +735,6 @@ def _download_url_with_cap(url: str, max_bytes: int) -> tuple[bytes, Optional[st
 def _handle_attach_url(args: dict, **kw) -> str:
     """Attach a file fetched server-side from an http(s) URL (shared size cap)."""
     from hermes_cli import kanban_db as kb
-
     tid = _worker_guard("kanban_attach_url", args)
     url = str(_require_text(args, "url")).strip()
     filename = args.get("filename") or args.get("title")
@@ -796,19 +774,16 @@ def _handle_create(args: dict, **kw) -> str:
     assignee = args.get("assignee")
     _check(assignee, "assignee is required — name the profile that should execute this "
                      "task (the dispatcher will only spawn tasks with an assignee)")
-    # Prefer the request-scoped api_server origin binding over HERMES_SESSION_ID:
-    # the env var is clobbered with a subagent's internal id whenever a child
-    # agent is constructed in-process, which would stamp — and later wake —
-    # the wrong session. NULL on CLI/dashboard paths that set neither.
+    # Prefer the request-scoped api_server origin binding over HERMES_SESSION_ID: the env
+    # var is clobbered with a subagent's internal id whenever a child agent is constructed
+    # in-process, which would stamp — and later wake — the wrong session.
     from tools.async_delegation import _current_origin_session_id
-
     session_id = (args.get("session_id") or _current_origin_session_id()
                   or os.environ.get("HERMES_SESSION_ID"))
-    # Workspace sharing is always explicit: omitted fields mean a fresh scratch
-    # workspace even for a dispatcher-spawned creator — reusing the parent's
-    # literal path would let a child mutate review evidence or race its
-    # checkout. Project identity is the one safe thing to inherit implicitly
-    # (the DB turns it into a fresh per-task worktree).
+    # Workspace sharing is always explicit: omitted fields mean a fresh scratch workspace
+    # even for a dispatcher-spawned creator (reusing the parent's path would let a child
+    # mutate review evidence or race its checkout). Project identity is the one safe thing
+    # to inherit implicitly (the DB turns it into a fresh per-task worktree).
     workspace_kind = args.get("workspace_kind")
     workspace_path = args.get("workspace_path")
     project_id = args.get("project") or args.get("project_id")
@@ -829,28 +804,18 @@ def _handle_create(args: dict, **kw) -> str:
                 project_id = self_task.project_id
                 project_source_task_id = self_task.id
         new_tid = kb.create_task(
-            conn,
-            title=str(title).strip(),
-            body=args.get("body"),
-            assignee=str(assignee),
-            parents=tuple(parents),
-            tenant=args.get("tenant") or os.environ.get("HERMES_TENANT"),
+            conn, title=str(title).strip(), body=args.get("body"), assignee=str(assignee),
+            parents=tuple(parents), tenant=args.get("tenant") or os.environ.get("HERMES_TENANT"),
             priority=_opt_int(args.get("priority"), 0),
             workspace_kind=str(workspace_kind if workspace_kind is not None else "scratch"),
-            workspace_path=workspace_path,
-            project_id=project_id,
-            project_source_task_id=project_source_task_id,
-            triage=triage,
+            workspace_path=workspace_path, project_id=project_id,
+            project_source_task_id=project_source_task_id, triage=triage,
             idempotency_key=args.get("idempotency_key"),
-            max_runtime_seconds=_opt_int(args.get("max_runtime_seconds")),
-            skills=skills,
-            model_override=model_override,
-            provider_override=provider_override,
-            goal_mode=goal_mode,
-            goal_max_turns=_opt_int(args.get("goal_max_turns")),
+            max_runtime_seconds=_opt_int(args.get("max_runtime_seconds")), skills=skills,
+            model_override=model_override, provider_override=provider_override,
+            goal_mode=goal_mode, goal_max_turns=_opt_int(args.get("goal_max_turns")),
             initial_status=str(args.get("initial_status") or "running"),
-            created_by=os.environ.get("HERMES_PROFILE") or "worker",
-            session_id=session_id)
+            created_by=os.environ.get("HERMES_PROFILE") or "worker", session_id=session_id)
         new_task = kb.get_task(conn, new_tid)
         subscribed = _maybe_auto_subscribe(conn, new_tid)
         return _ok(
@@ -869,7 +834,6 @@ def _resolve_notify_target() -> Optional[dict[str, Any]]:
     every CLI/ACP invocation and would auto-subscribe every CLI run.
     """
     from gateway.session_context import get_session_env
-
     platform = get_session_env("HERMES_SESSION_PLATFORM", "")
     chat_id = get_session_env("HERMES_SESSION_CHAT_ID", "")
     if not platform or not chat_id:
@@ -894,10 +858,8 @@ def _resolve_notify_target() -> Optional[dict[str, Any]]:
         delivery_metadata["thread_id"] = thread_id
     if chat_type:
         delivery_metadata["chat_type"] = chat_type
-    if (
-        platform.lower() == "telegram"
-        and thread_id
-        and (chat_type or "").lower() in {"dm", "direct", "private"}):
+    if (platform.lower() == "telegram" and thread_id
+            and (chat_type or "").lower() in {"dm", "direct", "private"}):
         delivery_metadata["telegram_dm_topic_reply_fallback"] = True
         if str(thread_id) not in {"", "1"}:
             delivery_metadata["direct_messages_topic_id"] = str(thread_id)
@@ -923,7 +885,6 @@ def _maybe_auto_subscribe(conn: Any, task_id: str) -> bool:
             return False
     except Exception:
         pass  # unreadable config keeps the user-friendly default (True)
-
     target = None
     try:
         target = _resolve_notify_target()

tools/lazy_deps.py

@@ -75,10 +75,9 @@ LAZY_DEPS: dict[str, tuple[str, ...]] = {
     "stt.silk": ("pilk==0.2.4",),
 
     # ─── Wake word ("Hey Hermes") engines (sync with the `wake` extra) ──────
-    # openWakeWord's ONNX model scores ~0 on macOS ARM64, so macOS uses the tflite
-    # backend (ai-edge-litert, bridged in tools/wake_word.py). Separate feature
-    # because specs cannot carry PEP 508 markers (";" is rejected) — the caller
-    # applies the platform gate.
+    # openWakeWord's ONNX model scores ~0 on macOS ARM64, so macOS uses the tflite backend
+    # (ai-edge-litert, bridged in tools/wake_word.py). Separate feature because specs cannot
+    # carry PEP 508 markers (";" is rejected) — the caller applies the platform gate.
     "wake.openwakeword.tflite": (
         "ai-edge-litert==2.1.6",
     ),
@@ -108,17 +107,16 @@ LAZY_DEPS: dict[str, tuple[str, ...]] = {
     # ─── Memory providers ──────────────────────────────────────────────────
     "memory.honcho": ("honcho-ai==2.2.0",),
     "memory.hindsight": ("hindsight-client==0.6.1",),
-    # Cloud memory SDKs MUST be allowlisted + ensure()'d at the import site, or
-    # they never install on the sealed Docker image (durable-target only).
+    # Cloud memory SDKs MUST be allowlisted + ensure()'d at the import site, or they never
+    # install on the sealed Docker image (durable-target only).
     "memory.supermemory": ("supermemory==3.50.0",),
     "memory.mem0": ("mem0ai==2.0.10",),
 
     # ─── Messaging platforms (lazy-installable on demand) ──────────────────
     "platform.telegram": ("python-telegram-bot[webhooks]==22.8",),
-    # brotlicffi: aiohttp needs its 2-arg Decompressor for Discord CDN's
-    # Brotli attachments; google's `Brotli` (1-arg) fails "Can not decode br".
-    # aiohttp is only capped transitively by these adapters, so a vulnerable
-    # already-installed copy would satisfy them — pin the patched floor explicitly.
+    # brotlicffi: aiohttp needs its 2-arg Decompressor for Discord CDN Brotli attachments
+    # (google's 1-arg `Brotli` fails "Can not decode br"). aiohttp is only capped transitively
+    # by these adapters, so pin the patched floor explicitly.
     "platform.discord": (
         "discord.py[voice]==2.7.1",
         "brotlicffi==1.2.0.1",
@@ -176,23 +174,19 @@ LAZY_DEPS: dict[str, tuple[str, ...]] = {
         "starlette==1.3.1",
         "python-multipart==0.0.32",  # FastAPI UploadFile/Form streaming uploads
     ),
-    # Pillow and firecrawl-anydoc are CORE deps; these entries are the self-heal
-    # path for lean/partial installs. Call sites use prompt=False so read_file /
-    # vision can never block on an input() prompt mid-session.
+    # Pillow and firecrawl-anydoc are CORE deps; these entries self-heal lean/partial installs.
+    # Call sites use prompt=False so read_file / vision never block on input() mid-session.
     "tool.vision": ("Pillow==12.3.0",),
     "tool.doc_extract": ("firecrawl-anydoc==0.2.4",),  # imports as `anydoc`; lockstep with pyproject
-    # MCP client SDK for the cua-driver; covers lean/broken-extra installs so
-    # computer_use never dead-ends on `No module named 'mcp'`.
+    # MCP client SDK for the cua-driver, so computer_use never dead-ends on `No module named 'mcp'`.
     "tool.computer_use": (
         "mcp==2.0.0",
         "httpx2==2.7.0",  # mcp 2.x HTTP stack — sync with pyproject [computer-use]
         "starlette==1.3.1",
     ),
-    # huggingface-hub is SHARED with transformers (>=1.5.0,<2 via Hindsight) and
-    # active_features() marks it active on mere presence, so `hermes update`
-    # re-asserts this pin everywhere hub exists. It MUST stay inside transformers'
-    # window and match uv.lock (tests/test_project_metadata.py enforces both);
-    # bump with `uv lock --upgrade-package huggingface-hub` in lockstep.
+    # huggingface-hub is SHARED with transformers (>=1.5.0,<2 via Hindsight) and marked active
+    # on mere presence, so `hermes update` re-asserts this pin everywhere hub exists. MUST stay
+    # inside transformers' window and match uv.lock (tests/test_project_metadata.py enforces).
     "tool.trace_upload": ("huggingface-hub==1.24.0",),
 }
 
@@ -256,10 +250,9 @@ def _ensure_target_ready(target: Path) -> Optional[str]:
     stamp = target / _TARGET_STAMP_NAME
     try:
         if target.exists():
-            try:
+            have = ""
+            with contextlib.suppress(OSError):
                 have = stamp.read_text(encoding="utf-8").strip()
-            except OSError:
-                have = ""
             if have and have != want:
                 logger.info("Lazy install target %s was built for ABI %r but running ABI is %r; wiping stale packages.", target, have, want)
                 for child in target.iterdir():
@@ -495,8 +488,7 @@ def _venv_pip_install(specs: tuple[str, ...], *, timeout: int = 300) -> _Install
     constraints: Optional[Path] = None
     extra_args: list[str] = []
     if target is not None:
-        err = _ensure_target_ready(target)
-        if err:
+        if err := _ensure_target_ready(target):
             return _InstallResult(False, "", err)
         constraints = _core_constraints_file()
         extra_args += ["--target", str(target)]
@@ -766,9 +758,8 @@ def ensure_and_bind(feature: str, importer: Callable[[], dict[str, Any]], target
         logger.warning("Failed to ensure feature %r: %s", feature, exc)
         return False
     try:
-        bindings = importer()
+        target_globals.update(importer())
     except ImportError as exc:
         logger.warning("Failed to import feature %r after install: %s", feature, exc)
         return False
-    target_globals.update(bindings)
     return True

tools/mcp_oauth.py

@@ -478,7 +478,7 @@ def _paste_callback_reader(result: dict) -> None:
         line = sys.stdin.readline()
     except (KeyboardInterrupt, OSError, ValueError):
         return
-    line = line.strip() if line else ""
+    line = (line or "").strip()
     if not line or _result_taken(result):
         return  # EOF / blank, or the HTTP listener already won
     if line.lower() in _SKIP_TOKENS:
@@ -620,8 +620,7 @@ def _make_callback_waiter(port: int, cimd_url: str | None = None, timeout: float
         dashboard_flow = get_dashboard_oauth_flow()
         if dashboard_flow is not None:
             # Dashboard flow speaks the legacy tuple; normalize to one shape.
-            dash_code, dash_state = await dashboard_flow.wait_for_callback()
-            return _authorization_code_result(dash_code, dash_state)
+            return _authorization_code_result(*await dashboard_flow.wait_for_callback())
         # The SDK entered the authorization-code flow, so any cached token is unusable. Reject
         # BEFORE binding: binding would block for the full timeout and collide with the
         # TIME_WAIT port on retry.