diff --git a/tools/delegation_live_log.py b/tools/delegation_live_log.py index 13638025ac..55d9409a83 100644 --- a/tools/delegation_live_log.py +++ b/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) diff --git a/tools/environments/base.py b/tools/environments/base.py index dbe848248f..ad0577afb9 100644 --- a/tools/environments/base.py +++ b/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() diff --git a/tools/environments/base_output.py b/tools/environments/base_output.py index c7d8aab9bb..460568cd25 100644 --- a/tools/environments/base_output.py +++ b/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: diff --git a/tools/environments/base_session_env.py b/tools/environments/base_session_env.py index ff29649b6f..12a8f85b47 100644 --- a/tools/environments/base_session_env.py +++ b/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. diff --git a/tools/environments/docker.py b/tools/environments/docker.py index a07e618291..ce2bc690d3 100644 --- a/tools/environments/docker.py +++ b/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. diff --git a/tools/environments/docker_egress.py b/tools/environments/docker_egress.py index 1f555f96aa..4778e6ec19 100644 --- a/tools/environments/docker_egress.py +++ b/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)) diff --git a/tools/environments/file_sync.py b/tools/environments/file_sync.py index cee07113e2..c9cb048e58 100644 --- a/tools/environments/file_sync.py +++ b/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) diff --git a/tools/interrupt.py b/tools/interrupt.py index 8a15b4946f..f779ecdc4b 100644 --- a/tools/interrupt.py +++ b/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: diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index 2530880034..3a9ae789e7 100644 --- a/tools/kanban_tools.py +++ b/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() diff --git a/tools/lazy_deps.py b/tools/lazy_deps.py index ed369a18e0..a9264b989a 100644 --- a/tools/lazy_deps.py +++ b/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 diff --git a/tools/mcp_oauth.py b/tools/mcp_oauth.py index 610d6e19a7..52a19b4e86 100644 --- a/tools/mcp_oauth.py +++ b/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.