diff --git a/tools/slash_confirm.py b/tools/slash_confirm.py index b7bdbe3f3a..a44b9d0dab 100644 --- a/tools/slash_confirm.py +++ b/tools/slash_confirm.py @@ -1,12 +1,11 @@ """Generic slash-command confirmation primitive (gateway-side). -Slash commands with an expensive side effect (currently only ``/reload-mcp``, -which invalidates the provider prompt cache) route through here. Button-UI -adapters render Approve Once / Always Approve / Cancel and call ``resolve()``; -text-only adapters get a prompt and the gateway intercepts ``/approve``, -``/always``, ``/cancel``. State is module-level (like ``tools.approval``) so -adapters can resolve without a ``GatewayRunner`` backreference. The CLI has its -own synchronous variant (``_prompt_slash_confirm`` in ``cli.py``). +Slash commands with an expensive side effect (currently only ``/reload-mcp``, which invalidates +the provider prompt cache) route through here. Button-UI adapters render Approve Once / Always +Approve / Cancel and call ``resolve()``; text-only adapters get a prompt and the gateway +intercepts ``/approve``, ``/always``, ``/cancel``. State is module-level (like ``tools.approval``) +so adapters can resolve without a ``GatewayRunner`` backreference. The CLI has its own +synchronous variant (``_prompt_slash_confirm`` in ``cli.py``). """ from __future__ import annotations @@ -22,17 +21,13 @@ logger = logging.getLogger(__name__) _pending: Dict[str, Dict[str, Any]] = {} _lock = threading.RLock() -# A pending confirm older than this is discarded when the next message arrives -# for the same session (buttons live as long as the adapter keeps callback_data). +# Older pending confirms are discarded when the session's next message arrives (buttons live +# as long as the adapter keeps callback_data). DEFAULT_TIMEOUT_SECONDS = 300 -def register( - session_key: str, - confirm_id: str, - command: str, - handler: Callable[[str], Awaitable[Optional[str]]], -) -> None: +def register(session_key: str, confirm_id: str, command: str, + handler: Callable[[str], Awaitable[Optional[str]]]) -> None: """Register a pending confirm, superseding any prior one for the session.""" with _lock: _pending[session_key] = {"confirm_id": confirm_id, "command": command, @@ -60,29 +55,24 @@ def clear_if_stale(session_key: str, timeout: float = DEFAULT_TIMEOUT_SECONDS) - """Drop the pending confirm if older than ``timeout`` seconds; True if dropped.""" with _lock: entry = _pending.get(session_key) - if entry and _is_stale(entry, timeout): + stale = bool(entry and _is_stale(entry, timeout)) + if stale: _pending.pop(session_key, None) - return True - return False + return stale -async def resolve( - session_key: str, - confirm_id: str, - choice: str, - timeout: float = DEFAULT_TIMEOUT_SECONDS, -) -> Optional[str]: +async def resolve(session_key: str, confirm_id: str, choice: str, + timeout: float = DEFAULT_TIMEOUT_SECONDS) -> Optional[str]: """Run the pending handler with ``choice`` ("once" / "always" / "cancel"). - Returns the handler's output string, or None if the confirm was stale, - already resolved, or the confirm_id doesn't match (superseded prompt). + Returns the handler's output string, or None if the confirm was stale, already resolved, + or the confirm_id doesn't match (superseded prompt). """ with _lock: entry = _pending.get(session_key) if not entry or entry.get("confirm_id") != confirm_id: return None - # Pop before running so duplicate callbacks (button double-click) - # cannot run the handler twice. + # Pop before running so duplicate callbacks (button double-click) cannot run it twice. _pending.pop(session_key, None) if _is_stale(entry, timeout): return None diff --git a/tools/subagent_worktree.py b/tools/subagent_worktree.py index edeb76dbe1..709049441c 100644 --- a/tools/subagent_worktree.py +++ b/tools/subagent_worktree.py @@ -1,14 +1,9 @@ -"""Opt-in git worktree isolation for delegated subagents -(``delegation.worktree_isolation: true``, default false). +"""Opt-in git worktree isolation for delegated subagents (``delegation.worktree_isolation``). -Git-only: in a non-git workspace the setting is ignored and children share -the parent's cwd. One worktree per child, branched from the parent's ``HEAD`` -under ``/.worktrees/subagent-`` on branch ``hermes-subagent/``. -Each result entry reports path, branch, commit count and dirty state; a -worktree is pruned only on affirmative proof (zero commits AND clean tree, both -probes succeeded) — otherwise it is kept with ``inspection_failed`` + ``note``. -Local terminal backend only: on docker/ssh/modal the host worktree is invisible -inside the sandbox, so isolation is skipped rather than half-applied. +Git-only (outside a repo children share the parent's cwd); local terminal backend only (on +docker/ssh/modal the host worktree is invisible in the sandbox, so isolation is skipped). One +worktree per child under ``/.worktrees/subagent-``, branch ``hermes-subagent/``; +pruned only on proof (zero commits AND clean, both probes ok), else kept + ``inspection_failed``. """ from __future__ import annotations @@ -30,9 +25,9 @@ _GIT_TIMEOUT = 30 def _run_git(args, cwd: str, timeout: int = _GIT_TIMEOUT): """Run git capturing output; never raises on non-zero exit. - Runs under :func:`noninteractive_git_env` (GHSA-7x36-8jrh-v4pw): this runs - automatically against whatever repo the parent sits in and ``worktree add`` - runs checkout hooks, so a malicious ``.git/config`` must not execute. + :func:`noninteractive_git_env` (GHSA-7x36-8jrh-v4pw): this runs unattended against whatever + repo the parent sits in and ``worktree add`` runs hooks, so a malicious ``.git/config`` must + not execute. """ return subprocess.run(["git", *harden_git_argv(args)], cwd=cwd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, @@ -73,15 +68,14 @@ def _ensure_gitignore_entry(repo_root: str) -> None: existing = gitignore.read_text(encoding="utf-8-sig", errors="replace") if gitignore.exists() else "" if ".worktrees/" not in existing.splitlines(): with open(gitignore, "a", encoding="utf-8") as f: - if existing and not existing.endswith("\n"): - f.write("\n") - f.write(".worktrees/\n") + sep = "\n" if existing and not existing.endswith("\n") else "" + f.write(f"{sep}.worktrees/\n") except Exception as exc: logger.debug("subagent worktree: could not update .gitignore: %s", exc) def create_subagent_worktree(parent_cwd: Optional[str], subagent_id: Optional[str] = None) -> Optional[Dict[str, str]]: - """Create an isolated worktree for one child; None (silent downgrade) when not a git repo or on failure.""" + """Create an isolated worktree for one child; None (silent downgrade) if not git / on failure.""" repo_root = resolve_repo_root(parent_cwd) if not repo_root: return None @@ -89,16 +83,9 @@ def create_subagent_worktree(parent_cwd: Optional[str], subagent_id: Optional[st wt_name = f"subagent-{(subagent_id or uuid.uuid4().hex[:8]).replace('/', '-')}" branch = f"hermes-subagent/{wt_name}" wt_path = Path(repo_root) / ".worktrees" / wt_name - try: wt_path.parent.mkdir(parents=True, exist_ok=True) - except Exception as exc: - logger.warning("subagent worktree: cannot create %s: %s", wt_path.parent, exc) - return None - - _ensure_gitignore_entry(repo_root) - - try: + _ensure_gitignore_entry(repo_root) base = _run_git(["rev-parse", "HEAD"], cwd=repo_root) base_commit = base.stdout.strip() if base.returncode == 0 else "" result = _run_git(["worktree", "add", str(wt_path), "-b", branch, "HEAD"], cwd=repo_root) @@ -109,7 +96,6 @@ def create_subagent_worktree(parent_cwd: Optional[str], subagent_id: Optional[st # Common on repos with zero commits (unborn HEAD) — degrade silently. logger.warning("subagent worktree: git worktree add failed: %s", result.stderr.strip()) return None - logger.info("subagent worktree created: %s (branch %s)", wt_path, branch) return {"path": str(wt_path), "branch": branch, "repo_root": repo_root, "base_commit": base_commit} @@ -120,64 +106,51 @@ def _base_payload(info: Dict[str, str]) -> Dict[str, Any]: "commits": 0, "dirty": False, "pruned": False} -def mark_worktree_payload_unproven( - payload: Dict[str, Any], reason: str, *, unmeasured: str = "commits/dirty" -) -> Dict[str, Any]: +def mark_worktree_payload_unproven(payload: Dict[str, Any], reason: str, *, + unmeasured: str = "commits/dirty") -> Dict[str, Any]: """Flag a worktree result payload as un-inspected, in place. - The parent only sees this dict, so the uncertainty must travel in it or - "0 commits, clean" reads as "the child produced nothing". *unmeasured* - names only the fields actually left unproven (one probe can succeed while - the other fails). Shared by ``finalize_subagent_worktree`` and - ``delegate_tool``'s fallback so the two producers cannot drift. + The parent only sees this dict, so the uncertainty must travel in it or "0 commits, clean" + reads as "the child produced nothing". *unmeasured* names only the fields actually left + unproven (one probe can succeed while the other fails). """ - path = payload.get("path", "") - branch = payload.get("branch", "") + path, branch = payload.get("path", ""), payload.get("branch", "") payload["inspection_failed"] = True - payload["note"] = ( - f"git inspection failed ({reason}): {unmeasured} UNKNOWN — not " - "proven zero/clean. The worktree and branch were preserved " - f"— inspect {path} (branch {branch}) before assuming no work." - ) - logger.warning("subagent worktree: git inspection failed (%s) — keeping %s (branch %s) for manual review", - reason, path, branch) + payload["note"] = (f"git inspection failed ({reason}): {unmeasured} UNKNOWN — not proven " + f"zero/clean. The worktree and branch were preserved — inspect {path} " + f"(branch {branch}) before assuming no work.") + logger.warning("subagent worktree: git inspection failed (%s) — keeping %s (branch %s) " + "for manual review", reason, path, branch) return payload def unproven_worktree_payload(info: Dict[str, str], reason: str) -> Dict[str, Any]: - """Complete un-inspected payload for ``delegate_tool`` when ``finalize_subagent_worktree`` raises.""" + """Complete un-inspected payload for ``delegate_tool`` when finalize raises.""" return mark_worktree_payload_unproven(_base_payload(info), reason) def finalize_subagent_worktree(info: Dict[str, str], *, prune: bool = True) -> Dict[str, Any]: """Inspect (and possibly prune) a child worktree after the child finishes. - Prunes only when *prune*, commits==0, clean tree AND both git probes - succeeded. If a probe exits non-zero or raises the worktree/branch are kept - and the payload carries ``inspection_failed`` + ``note``; ``commits`` / - ``dirty`` are then defaults, NOT measurements. + Prunes only when *prune*, commits==0, clean tree AND both git probes succeeded; otherwise + keeps it with ``inspection_failed`` + ``note`` (``commits``/``dirty`` then are defaults). """ - path = info.get("path", "") - branch = info.get("branch", "") + path, branch = info.get("path", ""), info.get("branch", "") base_commit = info.get("base_commit", "") payload = _base_payload(info) if not path or not os.path.isdir(path): payload["pruned"] = True # nothing on disk to review return payload - # Without a base commit the count is an unproven default, and the prune # condition reads payload["commits"] — so it must not prune either. if not base_commit: return mark_worktree_payload_unproven( - payload, "no base_commit recorded — commit count unmeasurable", unmeasured="commits" - ) + payload, "no base_commit recorded — commit count unmeasurable", unmeasured="commits") - failed: list = [] - unmeasured: list = [] - probes = ( - ("commits", "rev-list", ["rev-list", "--count", f"{base_commit}..HEAD"], lambda out: int(out or 0)), - ("dirty", "status", ["status", "--porcelain"], bool), - ) + failed, unmeasured = [], [] + probes = (("commits", "rev-list", ["rev-list", "--count", f"{base_commit}..HEAD"], + lambda s: int(s or 0)), + ("dirty", "status", ["status", "--porcelain"], bool)) try: for field, label, args, parse in probes: res = _run_git(args, cwd=path) @@ -190,7 +163,6 @@ def finalize_subagent_worktree(info: Dict[str, str], *, prune: bool = True) -> D # Timeout, OSError or non-numeric rev-list stdout: which probe raised is # unknowable, so neither value is trustworthy — keep the worktree. return mark_worktree_payload_unproven(payload, f"inspection raised: {exc}") - if failed: # Destructive cleanup requires affirmative proof; defaults prove nothing. return mark_worktree_payload_unproven(payload, "; ".join(failed), unmeasured="/".join(unmeasured)) @@ -207,7 +179,6 @@ def finalize_subagent_worktree(info: Dict[str, str], *, prune: bool = True) -> D logger.debug("subagent worktree: prune failed: %s", removed.stderr.strip()) except Exception as exc: logger.debug("subagent worktree: prune failed: %s", exc) - return payload @@ -217,10 +188,8 @@ def build_worktree_context_note(info: Dict[str, str]) -> str: "\n\n[WORKTREE ISOLATION] You are working in an isolated git worktree " f"at: {info.get('path')}\n" f"Your dedicated branch is: {info.get('branch')}\n" - "All file edits and shell commands must happen inside this worktree " - "directory (your terminal already starts there). Do NOT cd to the " - "main repository checkout. Commit your changes to your branch when " - "done; the parent agent will review and merge your branch. If you " - "make no commits and leave the tree clean, the worktree is discarded " - "automatically." + "All file edits and shell commands must happen inside this worktree directory (your " + "terminal already starts there). Do NOT cd to the main repository checkout. Commit your " + "changes to your branch when done; the parent agent will review and merge your branch. If " + "you make no commits and leave the tree clean, the worktree is discarded automatically." ) diff --git a/tools/terminal_hints.py b/tools/terminal_hints.py index 071808fad4..396838ef2f 100644 --- a/tools/terminal_hints.py +++ b/tools/terminal_hints.py @@ -1,11 +1,10 @@ """Output-pattern failure hints for the terminal tool. -Extends the exit-code semantics table in ``terminal_tool`` with a bounded scan -of failed-command output mapped to ONE short, actionable recovery hint. -Rules: only fires on non-zero exit; first match wins, patterns ordered by -observed production frequency; scans only the first ``_SCAN_CHARS`` so hints -key on error headers, not deep context; hints state the *next action* in 1-2 -sentences; pure function, no I/O or config reads. +Extends the exit-code semantics table in ``terminal_tool`` with a bounded scan of failed-command +output mapped to ONE short, actionable recovery hint. Rules: only fires on non-zero exit; first +match wins, patterns ordered by observed production frequency; scans only the first +``_SCAN_CHARS`` so hints key on error headers, not deep context; hints state the *next action* +in 1-2 sentences; pure function, no I/O or config reads. """ from __future__ import annotations @@ -17,85 +16,59 @@ _SCAN_CHARS = 4000 def _regex_hint(pattern: str, message: str | Callable[[str], str], flags: int = 0) -> Callable[[str, str], Optional[str]]: - """Hint firing when ``pattern`` matches; ``{0}`` = first group, or ``message(group1)`` if callable.""" + """Hint firing when ``pattern`` matches; ``{0}`` = first group, or ``message(group1)``.""" rx = re.compile(pattern, flags) def hint(command: str, output: str) -> Optional[str]: m = rx.search(output) - if not m: - return None - return message(m.group(1)) if callable(message) else message.format(*m.groups()) + if m: + return message(m.group(1)) if callable(message) else message.format(*m.groups()) + return None return hint # Most `command not found` hits are bare `python` on python3-only distros. _MISSING_COMMAND_HINTS = { - "python": ( - "This system has no bare `python` — use `python3`, or the " - "project venv's interpreter (e.g. .venv/bin/python)." - ), - "pip": ( - "This system has no bare `pip` — use `pip3`, `python3 -m pip`, " - "or the project venv's pip (e.g. .venv/bin/pip)." - ), + "python": "This system has no bare `python` — use `python3`, or the project venv's " + "interpreter (e.g. .venv/bin/python).", + "pip": "This system has no bare `pip` — use `pip3`, `python3 -m pip`, or the project venv's " + "pip (e.g. .venv/bin/pip).", } def _missing_command_hint(missing: str) -> str: return _MISSING_COMMAND_HINTS.get(missing) or ( - f"`{missing}` is not installed or not on PATH. Verify with " - f"`which {missing}`; install it or use an absolute path instead of " - "retrying the same command." - ) + f"`{missing}` is not installed or not on PATH. Verify with `which {missing}`; install it " + "or use an absolute path instead of retrying the same command.") # Ordered by production frequency — first match wins. _OUTPUT_HINTS: list[Callable[[str, str], Optional[str]]] = [ # gh version drift; gh already prints the valid field list. - _regex_hint( - r'Unknown JSON field: "?(\w+)', - "The installed gh does not support the JSON field '{0}'. " - "The valid field list is printed in the output above — retry using " - "only fields from that list.", - ), - _regex_hint( - r"^CONFLICT |Automatic merge failed|needs merge", - "Git merge conflict. Do not retry this command. Resolve the " - "conflicted files listed above (edit, then `git add`), then continue " - "(`git rebase --continue` / commit the merge) — or abort with " - "`--abort`.", - re.M, - ), - _regex_hint( - r"(?:bash: line \d+: |bash: |sh: \d*:? ?)?([\w.+-]+): command not found", - _missing_command_hint, - ), + _regex_hint(r'Unknown JSON field: "?(\w+)', + "The installed gh does not support the JSON field '{0}'. The valid field list is " + "printed in the output above — retry using only fields from that list."), + _regex_hint(r"^CONFLICT |Automatic merge failed|needs merge", + "Git merge conflict. Do not retry this command. Resolve the conflicted files " + "listed above (edit, then `git add`), then continue (`git rebase --continue` / " + "commit the merge) — or abort with `--abort`.", re.M), + _regex_hint(r"(?:bash: line \d+: |bash: |sh: \d*:? ?)?([\w.+-]+): command not found", + _missing_command_hint), # Almost always a venv-activation slip, not a missing dependency. - _regex_hint( - r"(?:ModuleNotFoundError|ImportError): No module named '?([\w.]+)", - "Python cannot import '{0}'. Most often the wrong " - "interpreter is running: activate the project venv (e.g. `source " - ".venv/bin/activate`) or invoke its python directly. Only pip " - "install if the package is genuinely absent from that venv.", - ), - _regex_hint( - r"(?:fatal|error):.*?'([^']+)' already exists", - "'{0}' already exists — retrying unchanged will keep " - "failing. Reuse it, choose another name, or delete it first if it is " - "genuinely stale.", - ), - _regex_hint( - r"API rate limit|was submitted too quickly", - "GitHub API rate limit hit — immediate retries will keep failing. " - "Continue with other work and retry this operation later.", - ), - _regex_hint( - r"Permission denied|EACCES", - "Permission denied. Check ownership/mode of the target path " - "(`ls -la`); prefer a user-writable location. Only escalate to sudo " - "if the task genuinely requires it.", - ), + _regex_hint(r"(?:ModuleNotFoundError|ImportError): No module named '?([\w.]+)", + "Python cannot import '{0}'. Most often the wrong interpreter is running: " + "activate the project venv (e.g. `source .venv/bin/activate`) or invoke its python " + "directly. Only pip install if the package is genuinely absent from that venv."), + _regex_hint(r"(?:fatal|error):.*?'([^']+)' already exists", + "'{0}' already exists — retrying unchanged will keep failing. Reuse it, choose " + "another name, or delete it first if it is genuinely stale."), + _regex_hint(r"API rate limit|was submitted too quickly", + "GitHub API rate limit hit — immediate retries will keep failing. Continue with " + "other work and retry this operation later."), + _regex_hint(r"Permission denied|EACCES", + "Permission denied. Check ownership/mode of the target path (`ls -la`); prefer a " + "user-writable location. Only escalate to sudo if the task genuinely requires it."), ] # Exit-code-only hints for codes the terminal_tool semantics table does not @@ -120,30 +93,22 @@ _PASSTHROUGH_CONSUMERS = r"(?:tail|head|cat|tee|less|more|wc|sort|uniq)" # Command shapes that swallow an upstream status -> warning, checked in order. _MASKING_SHAPES: list[tuple[re.Pattern[str], str]] = [ # Top-level `... | tail -20` (not `||`); consumer must be the LAST segment. - ( - re.compile(r"(? str: """Basename of the command head, skipping leading env-var assignments.""" for tok in (command or "").strip().split(): - if "=" in tok and not tok.startswith(("=", "./", "/")): - continue - return tok.rsplit("/", 1)[-1] + if "=" not in tok or tok.startswith(("=", "./", "/")): + return tok.rsplit("/", 1)[-1] return "" @@ -177,7 +140,8 @@ def annotate_masked_success(command: str, output: str) -> Optional[str]: """Warning note when an exit-0 result likely masks a failure (caller gates on exit 0).""" cmd = command or "" window = (output or "")[:_SCAN_CHARS] - if not cmd or not window or _first_token(cmd) in _READONLY_HEADS or not _FAILURE_SHAPES.search(window): + if (not cmd or not window or _first_token(cmd) in _READONLY_HEADS + or not _FAILURE_SHAPES.search(window)): return None return next((note for rx, note in _MASKING_SHAPES if rx.search(cmd)), None) diff --git a/tools/terminal_scope.py b/tools/terminal_scope.py index 1cd8807b57..a5dc6e4f08 100644 --- a/tools/terminal_scope.py +++ b/tools/terminal_scope.py @@ -1,13 +1,11 @@ """Per-turn terminal scope: profile-scoped TERMINAL_* policy. -Multiplexed surfaces (gateway, dashboard/TUI, cron) serve several profiles from -one process; mirroring terminal settings into ``os.environ`` let the first -profile pin its backend onto everyone else (sandbox escape). Like -``agent/secret_scope.py``, a ContextVar holds the active profile's COMPLETE -effective ``TERMINAL_*`` policy. While bound, ``terminal_env`` resolves ONLY -from it (omitted keys -> defined default, never ambient env). If the policy -cannot be resolved a *refusal* scope is installed and terminal execution raises -:class:`TerminalPolicyUnavailable` instead of falling back to ambient authority. +Multiplexed surfaces (gateway, dashboard/TUI, cron) serve several profiles from one process; +mirroring terminal settings into ``os.environ`` let the first profile pin its backend onto +everyone else (sandbox escape). Like ``agent/secret_scope.py``, a ContextVar holds the active +profile's COMPLETE ``TERMINAL_*`` policy; while bound, ``terminal_env`` resolves ONLY from it +(omitted keys -> defined default, never ambient env). If the policy cannot be resolved a +*refusal* scope is installed and terminal execution raises :class:`TerminalPolicyUnavailable`. """ from __future__ import annotations @@ -21,24 +19,15 @@ from typing import Any, Dict, Iterator, Optional logger = logging.getLogger(__name__) -# None = no scope bound (process-env behavior); dict = active profile's complete -# policy; TerminalPolicyRefusal = resolution failed. +# None = no scope bound (process-env behavior); dict = complete policy; Refusal = resolution failed. _terminal_scope_var: ContextVar = ContextVar("hermes_terminal_scope", default=None) -# Terminal keys whose config default lives in terminal_tool.py rather than -# DEFAULT_CONFIG (DEFAULT_CONFIG wins on overlap). Without them the projection -# is not total. +# Keys whose default lives in terminal_tool.py, not DEFAULT_CONFIG (which wins on overlap); +# without them the projection is not total. _TOOL_LEVEL_DEFAULTS: Dict[str, Any] = { - "cwd": ".", - "ssh_host": "", - "ssh_user": "", - "ssh_port": 22, - "ssh_key": "", - "docker_orphan_reaper": True, - "docker_persist_across_processes": True, - "sandbox_dir": "", - "lifetime_seconds": 300, - "docker_shared_container_key": "", + "cwd": ".", "ssh_host": "", "ssh_user": "", "ssh_port": 22, "ssh_key": "", + "docker_orphan_reaper": True, "docker_persist_across_processes": True, + "sandbox_dir": "", "lifetime_seconds": 300, "docker_shared_container_key": "", "home_mode": "auto", } @@ -60,11 +49,6 @@ def set_terminal_scope(mapping: Optional[Dict[str, str]]) -> Token: return _terminal_scope_var.set(mapping) -def install_refusal_scope(reason: str) -> Token: - """Install a refusal scope; terminal execution under it is rejected.""" - return _terminal_scope_var.set(TerminalPolicyRefusal(reason)) - - def reset_terminal_scope(token: Token) -> None: _terminal_scope_var.reset(token) @@ -84,8 +68,8 @@ def enforce_no_refusal() -> None: def terminal_env(name: str, default: str = "") -> str: """Authoritative read of a ``TERMINAL_*`` variable. - No scope: process env, then *default*. Refusal scope: raise. Policy scope: - ONLY the policy; a missing key yields *default*, never os.environ. + No scope: process env, then *default*. Refusal scope: raise. Policy scope: ONLY the + policy; a missing key yields *default*, never os.environ. """ scope = _terminal_scope_var.get() if scope is None: @@ -98,69 +82,56 @@ def terminal_env(name: str, default: str = "") -> str: def build_profile_terminal_scope(hermes_home: "Any") -> Dict[str, str]: """Build the COMPLETE effective ``TERMINAL_*`` policy for a profile home. - Projection: ``DEFAULT_CONFIG['terminal']`` <- profile ``.env`` TERMINAL_* <- - profile ``config.yaml`` ``terminal:`` keys. Total by construction, so a bound - scope never widens back to ambient authority. Raises - :class:`TerminalPolicyUnavailable` when either file exists but cannot be - read/parsed. + Projection: ``DEFAULT_CONFIG['terminal']`` <- profile ``.env`` TERMINAL_* <- profile + ``config.yaml`` ``terminal:``. Total by construction, so a bound scope never widens back to + ambient authority. Raises :class:`TerminalPolicyUnavailable` if a present file is unreadable. """ - home = Path(hermes_home) - from hermes_cli.config import TERMINAL_CONFIG_ENV_MAP from hermes_cli.config_defaults import DEFAULT_CONFIG - defaults = DEFAULT_CONFIG.get("terminal") if isinstance(DEFAULT_CONFIG, dict) else None - defaults = {**_TOOL_LEVEL_DEFAULTS, **(defaults if isinstance(defaults, dict) else {})} - + home = Path(hermes_home) scope: Dict[str, str] = {} - def _apply(cfg_key: str, value: Any) -> None: - # cwd placeholders are resolved per-surface later; not a policy value. - if value is None or (cfg_key == "cwd" and str(value).strip() in {".", "auto", "cwd"}): - return - env_var = TERMINAL_CONFIG_ENV_MAP.get(cfg_key) - if env_var: - scope[env_var] = str(value) - - for cfg_key, value in defaults.items(): - _apply(cfg_key, value) + def _apply(mapping: Dict[str, Any]) -> None: + for cfg_key, value in mapping.items(): + # cwd placeholders are resolved per-surface later; not a policy value. + if value is None or (cfg_key == "cwd" and str(value).strip() in {".", "auto", "cwd"}): + continue + env_var = TERMINAL_CONFIG_ENV_MAP.get(cfg_key) + if env_var: + scope[env_var] = str(value) + _apply({**_TOOL_LEVEL_DEFAULTS, **(DEFAULT_CONFIG.get("terminal") or {})}) env_path = home / ".env" if env_path.exists(): - # load_env_file swallows OSError by design (secret scope fails soft); - # an unreadable profile .env must fail closed here. + # load_env_file swallows OSError by design (secret scope fails soft); an unreadable + # profile .env must fail closed here. try: env_path.read_bytes() except Exception as exc: raise TerminalPolicyUnavailable(f"cannot read {env_path}: {exc}") from exc from agent.secret_scope import load_env_file - for key, value in load_env_file(env_path).items(): - if key.startswith("TERMINAL_"): - scope[key] = str(value) - - # The profile's own config.yaml is read directly (not read_raw_config(): it - # collapses "missing" and "unparseable" into {}); present-but-unparseable - # fails closed. + scope.update((k, str(v)) for k, v in load_env_file(env_path).items() + if k.startswith("TERMINAL_")) + # Read config.yaml directly, not via read_raw_config() (which collapses "missing" and + # "unparseable" into {}): present-but-unparseable must fail closed. + config_path = home / "config.yaml" try: - config_path = home / "config.yaml" - if config_path.exists(): - from hermes_cli.config import fast_safe_load - - try: - with open(config_path, encoding="utf-8") as f: - raw = fast_safe_load(f) - except Exception as exc: - raise TerminalPolicyUnavailable(f"cannot parse {config_path}: {exc}") from exc - raw_terminal = raw.get("terminal") if isinstance(raw, dict) else None - if isinstance(raw_terminal, dict): - for cfg_key, value in raw_terminal.items(): - _apply(cfg_key, value) - except TerminalPolicyUnavailable: - raise + config_exists = config_path.exists() except Exception as exc: raise TerminalPolicyUnavailable(f"cannot resolve terminal config in {home}: {exc}") from exc + if config_exists: + from hermes_cli.config import fast_safe_load + try: + with open(config_path, encoding="utf-8") as f: + raw = fast_safe_load(f) + except Exception as exc: + raise TerminalPolicyUnavailable(f"cannot parse {config_path}: {exc}") from exc + raw_terminal = raw.get("terminal") if isinstance(raw, dict) else None + if isinstance(raw_terminal, dict): + _apply(raw_terminal) return scope @@ -170,7 +141,7 @@ def install_profile_terminal_scope(hermes_home: "Any") -> Token: return set_terminal_scope(build_profile_terminal_scope(hermes_home)) except TerminalPolicyUnavailable as exc: logger.warning("terminal policy unavailable: %s", exc) - return install_refusal_scope(str(exc)) + return _terminal_scope_var.set(TerminalPolicyRefusal(str(exc))) @contextmanager diff --git a/tools/thread_context.py b/tools/thread_context.py index b48867fa98..9e5ae6a7a3 100644 --- a/tools/thread_context.py +++ b/tools/thread_context.py @@ -1,13 +1,12 @@ """Propagate agent-turn context into worker threads that dispatch Hermes tools. -A bare ``threading.Thread`` / ``ThreadPoolExecutor`` worker starts with an -empty ``contextvars.Context`` and no thread-local approval/sudo callbacks, so -tool dispatch inside it silently loses the approval ContextVars (gateway -sessions then auto-approve dangerous commands) and the CLI approval/sudo -callbacks (``prompt_dangerous_approval`` cannot reach the user, -GHSA-qg5c-hvr5-hjgr). Call :func:`propagate_context_to_thread` **on the parent -thread** (it snapshots at call time) and use the result as the worker target. -Callbacks are installed for the worker's lifetime and always cleared on exit. +A bare ``threading.Thread`` / ``ThreadPoolExecutor`` worker starts with an empty +``contextvars.Context`` and no thread-local approval/sudo callbacks, so tool dispatch inside it +silently loses the approval ContextVars (gateway sessions then auto-approve dangerous commands) +and the CLI callbacks (``prompt_dangerous_approval`` cannot reach the user, GHSA-qg5c-hvr5-hjgr). +Call :func:`propagate_context_to_thread` **on the parent thread** (it snapshots at call time) and +use the result as the worker target; callbacks are installed for the worker's lifetime and +always cleared on exit. """ from __future__ import annotations @@ -20,26 +19,19 @@ logger = logging.getLogger(__name__) def _callback_api(): - """Resolve the terminal_tool callback getters/setters. + """Resolve the terminal_tool callback getters/setters (lazy: terminal_tool imports + tools.approval at load, so a top-level import risks a cycle for tools.approval callers).""" + from tools import terminal_tool as tt - Lazy: ``tools.terminal_tool`` imports ``tools.approval`` at module load, so a - top-level import would risk a cycle for callers in ``tools.approval``. - """ - from tools.terminal_tool import ( - _get_approval_callback, - _get_sudo_password_callback, - set_approval_callback, - set_sudo_password_callback, - ) - return (_get_approval_callback, _get_sudo_password_callback, set_approval_callback, set_sudo_password_callback) + return (tt._get_approval_callback, tt._get_sudo_password_callback, + tt.set_approval_callback, tt.set_sudo_password_callback) def propagate_context_to_thread(target: Callable) -> Callable: """Wrap *target* to run with the *current* thread's ContextVars and approval/sudo callbacks. - Fail-closed: if callback installation raises, the callbacks stay unset - (``None``) — ``prompt_dangerous_approval`` then denies dangerous commands - and the gateway approval queue blocks. + Fail-closed: if callback installation raises they stay ``None`` — dangerous commands are then + denied by ``prompt_dangerous_approval`` and the gateway approval queue blocks. """ ctx = contextvars.copy_context() parent_approval_cb = parent_sudo_cb = None @@ -58,10 +50,9 @@ def propagate_context_to_thread(target: Callable) -> Callable: return target(*args, **kwargs) set_approval, set_sudo = setters try: - if parent_approval_cb is not None: - set_approval(parent_approval_cb) - if parent_sudo_cb is not None: - set_sudo(parent_sudo_cb) + for setter, cb in ((set_approval, parent_approval_cb), (set_sudo, parent_sudo_cb)): + if cb is not None: + setter(cb) except Exception: logger.debug("Failed to install propagated approval/sudo callbacks; " "dangerous-command approval will fail closed", exc_info=True) @@ -72,7 +63,8 @@ def propagate_context_to_thread(target: Callable) -> Callable: set_approval(None) set_sudo(None) except Exception: - logger.debug("Failed to clear propagated approval/sudo callbacks", exc_info=True) + logger.debug("Failed to clear propagated approval/sudo callbacks", + exc_info=True) return ctx.run(_inner)