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

Merge branch 'simp/r3-34-I' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:41:41 -07:00
parent 37696eabfc 3149ad5ef7
commit b3053edee7
5 changed files with 217 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/slash_confirm.py
+20 -41
View File
@@ -1,13 +1,10 @@
"""Generic slash-command confirmation primitive (gateway-side).
Slash commands with an expensive side effect worth surfacing (currently only
``/reload-mcp``, which invalidates the provider prompt cache) route through
here. Adapters with button UIs render Approve Once / Always Approve / Cancel
and call ``resolve()`` from the callback; text-only adapters get a prompt and
the gateway intercepts ``/approve``, ``/always``, ``/cancel`` replies.
State is module-level (like ``tools.approval``) so adapters can resolve
callbacks without a ``GatewayRunner`` backreference. The CLI has its own
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``).
"""
@@ -24,26 +21,17 @@ 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
# the 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,
"handler": handler,
"created_at": time.time(),
}
_pending[session_key] = {"confirm_id": confirm_id, "command": command,
"handler": handler, "created_at": time.time()}
def get_pending(session_key: str) -> Optional[Dict[str, Any]]:
@@ -67,30 +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).
Safe from an asyncio button callback or the gateway's message intercept.
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
@@ -102,9 +84,6 @@ async def resolve(
try:
result = await handler(choice)
except Exception as exc:
logger.error(
"Slash-confirm handler for /%s raised: %s",
command, exc, exc_info=True,
)
logger.error("Slash-confirm handler for /%s raised: %s", command, exc, exc_info=True)
return f"❌ Error handling confirmation: {exc}"
return result if isinstance(result, str) else None
tools/subagent_worktree.py
+51 -103
View File
@@ -1,18 +1,9 @@
"""Opt-in git worktree isolation for delegated subagents (clean-room port of
Muse Code's documented ``--subagent-worktree-isolation`` semantics).
Enable with ``delegation.worktree_isolation: true`` (default false).
"""Opt-in git worktree isolation for delegated subagents (``delegation.worktree_isolation``).
Contract: git-only — in a non-git workspace the setting is ignored without
error 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>``. The parent reviews/merges: 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); if a
probe fails the state is unknown, so it is kept and the entry carries
``inspection_failed`` + ``note``.
Local terminal backend only: on docker/ssh/modal the host worktree is invisible
inside the sandbox, so isolation is skipped (debug log) 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
@@ -29,18 +20,14 @@ from hermes_cli._subprocess_compat import harden_git_argv, noninteractive_git_en
logger = logging.getLogger(__name__)
_GIT_TIMEOUT = 30
_WORKTREES_DIRNAME = ".worktrees"
_BRANCH_NAMESPACE = "hermes-subagent"
def _run_git(args, cwd: str, timeout: int = _GIT_TIMEOUT):
"""Run a git command, capturing output. Never raises on non-zero exit.
"""Run git capturing output; never raises on non-zero exit.
Runs under :func:`noninteractive_git_env` (GHSA-7x36-8jrh-v4pw): worktree
isolation runs automatically for delegated subagents against whatever repo
the parent sits in, and ``worktree add`` runs checkout hooks. Disabling the
fsmonitor/hooks/pager/credential config sinks keeps a malicious ``.git/config``
from executing on the host.
: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,
@@ -61,12 +48,10 @@ def local_backend_active() -> bool:
def resolve_repo_root(path: Optional[str]) -> Optional[str]:
"""Return the git toplevel for *path*, or None when not in a work tree."""
if not path:
candidate = os.path.abspath(os.path.expanduser(str(path))) if path else ""
if not candidate or not os.path.isdir(candidate):
return None
try:
candidate = os.path.abspath(os.path.expanduser(str(path)))
if not os.path.isdir(candidate):
return None
result = _run_git(["rev-parse", "--show-toplevel"], cwd=candidate)
except Exception as exc:
logger.debug("subagent worktree: rev-parse failed: %s", exc)
@@ -77,39 +62,27 @@ def resolve_repo_root(path: Optional[str]) -> Optional[str]:
def _ensure_gitignore_entry(repo_root: str) -> None:
"""Best-effort: keep ``.worktrees/`` out of git status."""
gitignore = Path(repo_root) / ".gitignore"
entry = f"{_WORKTREES_DIRNAME}/"
try:
existing = gitignore.read_text(encoding="utf-8-sig", errors="replace") if gitignore.exists() else ""
if entry not in existing.splitlines():
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(f"{entry}\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 agent. Returns
``path``/``branch``/``repo_root``/``base_commit``, or ``None`` when not a git
repo or creation fails — absence of git downgrades silently to shared workspace."""
"""Create an isolated worktree for one child; None (silent downgrade) outside git/on failure."""
repo_root = resolve_repo_root(parent_cwd)
if not repo_root:
return None
wt_name = f"subagent-{(subagent_id or uuid.uuid4().hex[:8]).replace('/', '-')}"
branch = f"{_BRANCH_NAMESPACE}/{wt_name}"
wt_path = Path(repo_root) / _WORKTREES_DIRNAME / wt_name
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)
@@ -120,7 +93,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}
@@ -131,86 +103,65 @@ 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.
A failed probe proves nothing, so the fields it would have filled keep
their defaults. The parent only sees this dict (not logs), so the
uncertainty must travel in the payload 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, and calling a
measured value UNKNOWN would be its own misreport. Shared by
``finalize_subagent_worktree`` and ``delegate_tool``'s finalize-raised
fallback so the two producers of this schema 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 callers that never got one back
(``delegate_tool``'s fallback when ``finalize_subagent_worktree`` raises).
Emits exactly the parent-facing schema, WITHOUT ``repo_root``/``base_commit``."""
"""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.
Returns path, branch, ``commits`` ahead of base, ``dirty``, ``pruned``. 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: True`` + ``note``; ``commits``/``dirty`` are then
defaults, NOT measurements."""
path = info.get("path", "")
branch = info.get("branch", "")
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, 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"
)
failed: list = []
unmeasured: list = []
payload, "no base_commit recorded — commit count unmeasurable", unmeasured="commits")
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:
counted = _run_git(["rev-list", "--count", f"{base_commit}..HEAD"], cwd=path)
if counted.returncode == 0:
payload["commits"] = int(counted.stdout.strip() or 0)
else:
failed.append(f"rev-list exit {counted.returncode}: {counted.stderr.strip()[:200]}")
unmeasured.append("commits")
status = _run_git(["status", "--porcelain"], cwd=path)
if status.returncode == 0:
payload["dirty"] = bool(status.stdout.strip())
else:
failed.append(f"status exit {status.returncode}: {status.stderr.strip()[:200]}")
unmeasured.append("dirty")
for field, label, args, parse in probes:
res = _run_git(args, cwd=path)
if res.returncode == 0:
payload[field] = parse(res.stdout.strip())
else:
failed.append(f"{label} exit {res.returncode}: {res.stderr.strip()[:200]}")
unmeasured.append(field)
except Exception as exc:
# 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))
if prune and payload["commits"] == 0 and not payload["dirty"]:
cwd = info.get("repo_root", "") or path
try:
@@ -223,7 +174,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
@@ -233,10 +183,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
+66 -118
View File
@@ -1,14 +1,10 @@
"""Output-pattern failure hints for the terminal tool.
Extends the exit-code semantics table in ``terminal_tool`` with an
output-pattern tier: a bounded scan of failed-command output mapped to one
short, actionable recovery hint.
Rules (keep when adding patterns): only fires on non-zero exit; at most ONE
hint, 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, not a diagnosis;
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
@@ -16,95 +12,67 @@ from __future__ import annotations
import re
from typing import Callable, Optional
# Bounded scan window: error headers appear early; deep output is noise.
_SCAN_CHARS = 4000
def _regex_hint(pattern: str, message: str | Callable[[str], str], flags: int = 0) -> Callable[[str, str], Optional[str]]:
"""Build a hint that fires when ``pattern`` matches; ``{0}`` = first capture group,
or ``message(group1)`` when a callable is given."""
"""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 semantics table in terminal_tool does
# not cover per-command. Checked after output patterns.
# Exit-code-only hints for codes the terminal_tool semantics table does not
# cover per-command. Checked after output patterns.
_EXIT_CODE_HINTS: dict[int, str] = {
126: "Exit 126: the file was found but is not executable — `chmod +x` it or invoke it via its interpreter (e.g. `bash script.sh`).",
137: "Exit 137: the process was SIGKILLed — usually out-of-memory or an external kill. Reduce memory use or check `dmesg | tail` before retrying.",
@@ -112,14 +80,10 @@ _EXIT_CODE_HINTS: dict[int, str] = {
}
# ---------------------------------------------------------------------------
# Masked-success detection (exit 0 that probably isn't a success)
# ---------------------------------------------------------------------------
# `cargo build 2>&1 | tail -20` exits with tail's 0 (no pipefail) and
# `cargo build || echo FAILED` exits with echo's 0, so the model can conclude
# a build passed while the output says it failed. Conservative: BOTH the
# command shape (top-level pipe into a passthrough consumer, or `|| <cheap
# fallback>`) AND a strong tool-specific failure shape must hold, and
# Masked-success detection: `cargo build 2>&1 | tail -20` exits with tail's 0
# (no pipefail) and `cargo build || echo FAILED` with echo's 0, so the model can
# conclude a build passed while the output says it failed. Conservative: BOTH a
# masking command shape AND a strong tool-specific failure shape must hold, and
# read-only heads (`grep ... | head`) are excluded because their output
# legitimately contains error text. Advisory only — exit_code is never changed.
@@ -129,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.
@@ -169,29 +125,23 @@ _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():
# Skip env-var assignments and common wrappers.
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 ""
def annotate_masked_success(command: str, output: str) -> Optional[str]:
"""Return a warning note when an exit-0 result likely masks a failure.
Caller gates on exit_code == 0. Advisory only; returns None otherwise.
"""
"""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:
return None
if 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)
@@ -201,12 +151,10 @@ def annotate_failure(command: str, exit_code: int, output: str) -> Optional[str]
if exit_code == 0:
return None
window = (output or "")[:_SCAN_CHARS]
if window:
for fn in _OUTPUT_HINTS:
try:
hint = fn(command or "", window)
except Exception:
continue
if hint:
for fn in _OUTPUT_HINTS if window else ():
try:
if hint := fn(command or "", window):
return hint
except Exception:
continue
return _EXIT_CODE_HINTS.get(exit_code)
tools/terminal_scope.py
+53 -111
View File
@@ -1,20 +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`` for credentials, a ContextVar holds the active
profile's COMPLETE effective ``TERMINAL_*`` policy, installed at each
in-process profile boundary.
- **Authoritative projection.** While a scope is bound, ``terminal_env``
resolves ONLY from the policy (defaults + profile ``.env`` + its
``config.yaml``); omitted keys yield the defined default, never ambient
``os.environ`` (#68559).
- **Fail closed.** If the policy cannot be resolved, callers install a
*refusal* scope and terminal execution under it 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
@@ -28,23 +19,15 @@ from typing import Any, Dict, Iterator, Optional
logger = logging.getLogger(__name__)
# None = no scope bound (historical process-env behavior); dict = the 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 the consuming tool
# (terminal_tool.py) rather than DEFAULT_CONFIG; DEFAULT_CONFIG wins on overlap.
# 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",
}
@@ -66,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)
@@ -80,23 +58,24 @@ def get_terminal_scope() -> Optional[Dict[str, str]]:
return _terminal_scope_var.get()
def _raise_if_refusal(scope: Any) -> None:
def enforce_no_refusal() -> None:
"""Raise when the active scope is a refusal scope (fail closed)."""
scope = _terminal_scope_var.get()
if isinstance(scope, TerminalPolicyRefusal):
raise TerminalPolicyUnavailable(
f"terminal policy unavailable for this profile: {scope.reason}"
)
f"terminal policy unavailable for this profile: {scope.reason}")
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:
return os.environ.get(name, default)
_raise_if_refusal(scope)
enforce_no_refusal()
value = scope.get(name)
return default if value is None else str(value)
@@ -104,107 +83,70 @@ 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 process 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
# Without the tool-level defaults the projection is not total.
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)
# Read config.yaml through the HERMES_HOME override so the profile's own
# file is consulted; a present-but-unparseable file fails closed.
from hermes_constants import (
get_hermes_home_override,
reset_hermes_home_override,
set_hermes_home_override,
)
override_token = None
if get_hermes_home_override() != str(home):
override_token = set_hermes_home_override(home)
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():
# Not read_raw_config(): it collapses "missing" and "unparseable"
# into {}; the file exists, so a parse failure must fail closed.
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
finally:
if override_token is not None:
reset_hermes_home_override(override_token)
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
def install_profile_terminal_scope(hermes_home: "Any") -> Token:
"""Build AND install a profile's policy; on failure install the refusal scope.
Never raises. Returns the token for ``reset_terminal_scope``.
"""
"""Build AND install a profile's policy; on failure install the refusal scope. Never raises."""
try:
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))
def enforce_no_refusal() -> None:
"""Raise when the active scope is a refusal scope (fail closed, #68559)."""
_raise_if_refusal(_terminal_scope_var.get())
return _terminal_scope_var.set(TerminalPolicyRefusal(str(exc)))
@contextmanager
def install_and_reset_profile_terminal_scope(
hermes_home: "Any",
) -> Iterator[None]:
def install_and_reset_profile_terminal_scope(hermes_home: "Any") -> Iterator[None]:
"""Install the profile's terminal policy for a bounded turn/fire. Never raises."""
token = install_profile_terminal_scope(hermes_home)
try:
tools/thread_context.py
+27 -67
View File
@@ -1,29 +1,12 @@
#!/usr/bin/env python3
"""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 session/platform ContextVars (``tools.approval`` /
``gateway.session_context``) — gateway sessions then fall into
``check_dangerous_command``'s non-interactive auto-approve branch and
dangerous commands run without prompting;
* the thread-local CLI approval/sudo callbacks (``tools.terminal_tool``) —
``prompt_dangerous_approval`` then cannot reach the user (GHSA-qg5c-hvr5-hjgr).
One audited capture/install/clear lifecycle for every place that fans tool
dispatch onto worker threads (``agent.tool_executor``, ``execute_code`` RPC threads).
Call :func:`propagate_context_to_thread` **on the parent thread** (it snapshots
the parent's ContextVars and callbacks at call time) and use the result as the
worker target::
t = threading.Thread(target=propagate_context_to_thread(loop_fn), args=(...))
executor.submit(propagate_context_to_thread(worker_fn), *args)
Callbacks are installed for the worker's lifetime and **always cleared on exit**,
so a recycled thread never holds a stale reference to a disposed CLI instance.
``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
@@ -36,72 +19,49 @@ 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
Imported lazily: ``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* so it runs on a worker thread with the *current* thread's
ContextVars and approval/sudo callbacks. The wrapper forwards args/kwargs.
"""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`` denies dangerous commands with no
callback in an interactive context, and the gateway approval queue blocks
when its notify callback is absent.
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
setters = None
# (setter, parent callback) pairs; None when the callback API could not be captured.
installs = None
try:
get_approval, get_sudo, set_approval, set_sudo = _callback_api()
parent_approval_cb = get_approval()
parent_sudo_cb = get_sudo()
setters = (set_approval, set_sudo)
installs = ((set_approval, get_approval()), (set_sudo, get_sudo()))
except Exception:
logger.debug("Could not capture parent approval/sudo callbacks", exc_info=True)
def _runner(*args, **kwargs):
def _inner():
if setters is None:
if installs is None:
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 installs:
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,
)
logger.debug("Failed to install propagated approval/sudo callbacks; "
"dangerous-command approval will fail closed", exc_info=True)
try:
return target(*args, **kwargs)
finally:
try:
set_approval(None)
set_sudo(None)
for setter, _cb in installs:
setter(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)