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

refactor(tools): unify worktree probes, fold terminal_scope config-yaml guard, rewrap hint tables, compact docstrings (group I)

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:27:29 -07:00
parent 922083ab93
commit 1b889dbc1e
5 changed files with 172 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/slash_confirm.py
+18 -28
View File
@@ -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
tools/subagent_worktree.py
+36 -67
View File
@@ -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 ``<repo>/.worktrees/subagent-<id>`` on branch ``hermes-subagent/<id>``.
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 ``<repo>/.worktrees/subagent-<id>``, branch ``hermes-subagent/<id>``;
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."
)
tools/terminal_hints.py
+53 -89
View File
@@ -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"(?<!\|)\|(?!\|)\s*" + _PASSTHROUGH_CONSUMERS + r"\b[^|]*$"),
"exit_code 0 here is the status of the last pipeline command "
"(tail/head/cat/...), NOT of the command before the pipe — and "
"the output contains failure indicators. Treat this run as "
"FAILED until proven otherwise: re-run the command WITHOUT the "
"pipe (output is auto-truncated and the full text is saved to a "
"file, so piping through tail/head is never needed) to get the "
"real exit code.",
),
(re.compile(r"(?<!\|)\|(?!\|)\s*" + _PASSTHROUGH_CONSUMERS + r"\b[^|]*$"),
"exit_code 0 here is the status of the last pipeline command (tail/head/cat/...), NOT of "
"the command before the pipe — and the output contains failure indicators. Treat this run "
"as FAILED until proven otherwise: re-run the command WITHOUT the pipe (output is "
"auto-truncated and the full text is saved to a file, so piping through tail/head is never "
"needed) to get the real exit code."),
# `cmd || echo ...` / `cmd || true` — fallback swallows the failure status.
(
re.compile(r"\|\|\s*(?:echo\b|printf\b|true\b|:\s|:$)"),
"exit_code 0 here is the status of the `||` fallback (echo/true), "
"NOT of the command before it — and the output contains failure "
"indicators. Treat this run as FAILED until proven otherwise: "
"re-run the command bare to get its real exit code.",
),
(re.compile(r"\|\|\s*(?:echo\b|printf\b|true\b|:\s|:$)"),
"exit_code 0 here is the status of the `||` fallback (echo/true), NOT of the command before "
"it — and the output contains failure indicators. Treat this run as FAILED until proven "
"otherwise: re-run the command bare to get its real exit code."),
]
_READONLY_HEADS = frozenset({
"grep", "rg", "ag", "find", "ls", "cat", "head", "tail", "jq", "awk",
"sed", "strings", "zcat", "journalctl", "dmesg", "echo", "printf",
})
"sed", "strings", "zcat", "journalctl", "dmesg", "echo", "printf"})
# Strong failure shapes keyed to specific tools so that error-mentioning
# *content* (diffs, logs, commit messages) rarely matches.
@@ -160,16 +125,14 @@ _FAILURE_SHAPES = re.compile(
r"|BUILD FAILED|Build FAILED" # gradle/msbuild/echoed fallbacks
r"|FAILED: " # ninja
r"|(?m:^make(?:\[\d+\])?: \*\*\*)" # make
r")"
)
r")")
def _first_token(command: str) -> 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)
tools/terminal_scope.py
+46 -75
View File
@@ -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
tools/thread_context.py
+19 -27
View File
@@ -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)