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

refactor(agent/subdirectory_hints,skill_preprocessing): fold hint-file lookup, compact exception ladders and docs

Browse Source
This commit is contained in:
Teknium
2026-09-02 19:13:55 -07:00
parent 82aa392f1c
commit 042aea505d
2 changed files with 97 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/skill_preprocessing.py
+25 -64
View File
@@ -1,4 +1,5 @@
"""Shared SKILL.md preprocessing helpers."""
"""Shared SKILL.md preprocessing helpers: ``${HERMES_*}`` template tokens and
inline ``!`cmd``` shell expansion."""
import logging
import re
@@ -9,15 +10,11 @@ from hermes_cli._subprocess_compat import IS_WINDOWS, windows_hide_flags
logger = logging.getLogger(__name__)
# Matches ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} tokens in SKILL.md.
# Tokens that don't resolve (e.g. ${HERMES_SESSION_ID} with no session) are
# left as-is so the user can debug them.
# ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} tokens. Unresolvable ones (e.g. no
# session) are left as-is so the author can spot them.
_SKILL_TEMPLATE_RE = re.compile(r"\$\{(HERMES_SKILL_DIR|HERMES_SESSION_ID)\}")
# Matches inline shell snippets like: !`date +%Y-%m-%d`
# Non-greedy, single-line only -- no newlines inside the backticks.
# Inline shell snippets like !`date +%Y-%m-%d` — single-line only.
_INLINE_SHELL_RE = re.compile(r"!`([^`\n]+)`")
# Cap inline-shell output so a runaway command can't blow out the context.
_INLINE_SHELL_MAX_OUTPUT = 4000
@@ -27,8 +24,7 @@ def load_skills_config() -> dict:
try:
from hermes_cli.config import load_config_readonly
cfg = load_config_readonly() or {}
skills_cfg = cfg.get("skills")
skills_cfg = (load_config_readonly() or {}).get("skills")
if isinstance(skills_cfg, dict):
return skills_cfg
except Exception:
@@ -36,38 +32,21 @@ def load_skills_config() -> dict:
return {}
def substitute_template_vars(
content: str,
skill_dir: Path | None,
session_id: str | None,
) -> str:
"""Replace ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID} in skill content.
Only substitutes tokens for which a concrete value is available --
unresolved tokens are left in place so the author can spot them.
"""
def substitute_template_vars(content: str, skill_dir: Path | None, session_id: str | None) -> str:
"""Replace ${HERMES_SKILL_DIR} / ${HERMES_SESSION_ID}; tokens without a value stay in place."""
if not content:
return content
skill_dir_str = str(skill_dir) if skill_dir else None
def _replace(match: re.Match) -> str:
token = match.group(1)
if token == "HERMES_SKILL_DIR" and skill_dir_str:
return skill_dir_str
if token == "HERMES_SESSION_ID" and session_id:
return str(session_id)
return match.group(0)
return _SKILL_TEMPLATE_RE.sub(_replace, content)
values = {
"HERMES_SKILL_DIR": str(skill_dir) if skill_dir else None,
"HERMES_SESSION_ID": str(session_id) if session_id else None,
}
return _SKILL_TEMPLATE_RE.sub(lambda m: values[m.group(1)] or m.group(0), content)
def run_inline_shell(command: str, cwd: Path | None, timeout: int) -> str:
"""Execute a single inline-shell snippet and return its stdout (trimmed).
Failures return a short ``[inline-shell error: ...]`` marker instead of
raising, so one bad snippet can't wreck the whole skill message.
"""
"""Run one inline-shell snippet and return its stdout (trimmed; stderr when
stdout is empty). Failures return an ``[inline-shell ...]`` marker instead
of raising, so one bad snippet can't wreck the whole skill message."""
_popen_kwargs = {"creationflags": windows_hide_flags()} if IS_WINDOWS else {}
try:
completed = subprocess.run(
@@ -84,43 +63,27 @@ def run_inline_shell(command: str, cwd: Path | None, timeout: int) -> str:
return f"[inline-shell timeout after {timeout}s: {command}]"
except FileNotFoundError:
return "[inline-shell error: bash not found]"
except RuntimeError as exc:
# tests/conftest.py installs a live-system guard that blocks real
# os.kill on out-of-tree PIDs. subprocess.run(timeout=...) may trip
# that guard while trying to clean up the timed-out shell; treat that
# as the same timeout outcome instead of surfacing the guard error.
if "live-system guard: blocked os.kill" in str(exc):
except Exception as exc:
# tests/conftest.py's live-system guard may block the os.kill that
# subprocess.run uses to clean up a timed-out shell; report the timeout.
if isinstance(exc, RuntimeError) and "live-system guard: blocked os.kill" in str(exc):
return f"[inline-shell timeout after {timeout}s: {command}]"
return f"[inline-shell error: {exc}]"
except Exception as exc:
return f"[inline-shell error: {exc}]"
output = (completed.stdout or "").rstrip("\n")
if not output and completed.stderr:
output = completed.stderr.rstrip("\n")
output = (completed.stdout or "").rstrip("\n") or (completed.stderr or "").rstrip("\n")
if len(output) > _INLINE_SHELL_MAX_OUTPUT:
output = output[:_INLINE_SHELL_MAX_OUTPUT] + "...[truncated]"
return output
def expand_inline_shell(
content: str,
skill_dir: Path | None,
timeout: int,
) -> str:
"""Replace every !`cmd` snippet in ``content`` with its stdout.
Runs each snippet with the skill directory as CWD so relative paths in
the snippet work the way the author expects.
"""
def expand_inline_shell(content: str, skill_dir: Path | None, timeout: int) -> str:
"""Replace every !`cmd` snippet with its stdout, run with the skill dir as CWD."""
if "!`" not in content:
return content
def _replace(match: re.Match) -> str:
cmd = match.group(1).strip()
if not cmd:
return ""
return run_inline_shell(cmd, skill_dir, timeout)
return run_inline_shell(cmd, skill_dir, timeout) if cmd else ""
return _INLINE_SHELL_RE.sub(_replace, content)
@@ -134,11 +97,9 @@ def preprocess_skill_content(
"""Apply configured SKILL.md template and inline-shell preprocessing."""
if not content:
return content
cfg = skills_cfg if isinstance(skills_cfg, dict) else load_skills_config()
if cfg.get("template_vars", True):
content = substitute_template_vars(content, skill_dir, session_id)
if cfg.get("inline_shell", False):
timeout = int(cfg.get("inline_shell_timeout", 10) or 10)
content = expand_inline_shell(content, skill_dir, timeout)
content = expand_inline_shell(content, skill_dir, int(cfg.get("inline_shell_timeout", 10) or 10))
return content
agent/subdirectory_hints.py
+72 -126
View File
@@ -1,11 +1,10 @@
"""Progressive subdirectory hint discovery.
As the agent navigates into subdirectories via tool calls, this module loads
project context files (AGENTS.md, CLAUDE.md, .cursorrules) from those
directories and appends them to the tool result — context arrives without
touching the system prompt (preserving prompt caching). Complements the
startup CWD-only loading in ``prompt_builder.py``. Inspired by goose's
SubdirectoryHintTracker.
As the agent navigates into subdirectories via tool calls, load project
context files (AGENTS.md, CLAUDE.md, .cursorrules) from those directories and
append them to the tool result — context arrives without touching the system
prompt (preserving prompt caching). Complements the startup CWD-only loading
in ``prompt_builder.py``.
"""
import hashlib
@@ -29,8 +28,7 @@ _HINT_FILENAMES = [
_MAX_HINT_CHARS = 8_000
_PATH_ARG_KEYS = {"path", "file_path", "workdir"}
_COMMAND_TOOLS = {"terminal"}
# Ancestor levels walked per path — bounds the scan for deeply nested paths.
_MAX_ANCESTOR_WALK = 5
_MAX_ANCESTOR_WALK = 5 # ancestor levels walked per path — bounds deep-path scans
# Directories that hold *copies* of context files (backups, vendored deps,
# VCS internals, caches), never authoritative project context.
@@ -44,13 +42,23 @@ _EXCLUDED_DIR_NAMES = frozenset({
})
def _is_ancestor_or_same(a: Path, b: Path) -> bool:
"""True if *a* is *b* or one of its ancestors."""
try:
b.relative_to(a)
return True
except ValueError:
return False
def _digest(content: str) -> str:
return hashlib.sha256(content.encode("utf-8")).hexdigest()
def _first_hint_file(directory: Path):
"""``(path, stripped content)`` of the first readable non-empty hint file
in *directory* (priority order), or None. Unreadable files are skipped."""
for filename in _HINT_FILENAMES:
candidate = directory / filename
try:
if not candidate.is_file():
continue
content = candidate.read_text(encoding="utf-8").strip()
except (OSError, UnicodeDecodeError):
continue
return candidate, content
return None
class SubdirectoryHintTracker:
@@ -65,63 +73,34 @@ class SubdirectoryHintTracker:
# The working dir is pre-marked loaded (startup context handles it).
self._loaded_dirs: Set[Path] = {self.working_dir}
# Content digests already injected: the same file reached through
# symlinks/hardlinks/copies is never re-sent.
# symlinks/hardlinks/copies is never re-sent. Seeded with the CWD hint
# file prompt_builder already loaded.
self._loaded_digests: Set[str] = set()
self._seed_working_dir_digest()
found = _first_hint_file(self.working_dir)
if found and found[1]:
self._loaded_digests.add(_digest(found[1]))
def _seed_working_dir_digest(self) -> None:
"""Record the CWD context file's digest (prompt_builder already loaded it)."""
for filename in _HINT_FILENAMES:
candidate = self.working_dir / filename
try:
if not candidate.is_file():
continue
content = candidate.read_text(encoding="utf-8").strip()
except (OSError, UnicodeDecodeError):
continue
if content:
self._loaded_digests.add(
hashlib.sha256(content.encode("utf-8")).hexdigest()
)
break # first match wins, mirroring startup loading
def check_tool_call(
self,
tool_name: str,
tool_args: Dict[str, Any],
) -> Optional[str]:
def check_tool_call(self, tool_name: str, tool_args: Dict[str, Any]) -> Optional[str]:
"""Return formatted hint text for newly visited directories, or None."""
all_hints = []
for d in self._extract_directories(tool_name, tool_args):
hints = self._load_hints_for_directory(d)
if hints:
all_hints.append(hints)
if not all_hints:
return None
return "\n\n" + "\n\n".join(all_hints)
all_hints = [h for d in self._extract_directories(tool_name, tool_args) if (h := self._load_hints_for_directory(d))]
return "\n\n" + "\n\n".join(all_hints) if all_hints else None
def _extract_directories(
self, tool_name: str, args: Dict[str, Any]
) -> list:
def _extract_directories(self, tool_name: str, args: Dict[str, Any]) -> list:
"""Extract directory paths from tool call arguments."""
candidates: Set[Path] = set()
for key in _PATH_ARG_KEYS:
val = args.get(key)
if isinstance(val, str) and val.strip():
self._add_path_candidate(val, candidates)
if tool_name in _COMMAND_TOOLS:
cmd = args.get("command", "")
if isinstance(cmd, str):
self._extract_paths_from_command(cmd, candidates)
cmd = args.get("command", "") if tool_name in _COMMAND_TOOLS else None
if isinstance(cmd, str):
self._extract_paths_from_command(cmd, candidates)
return list(candidates)
def _add_path_candidate(self, raw_path: str, candidates: Set[Path]):
"""Add a raw path's directory and its ancestors to candidates.
Walks up toward the root, stopping at the first already-loaded
directory or after ``_MAX_ANCESTOR_WALK`` levels, so reading
``project/src/main.py`` still discovers ``project/AGENTS.md``.
"""
"""Add a raw path's directory and its ancestors (up to ``_MAX_ANCESTOR_WALK``
levels, stopping at the first already-loaded dir) so reading
``project/src/main.py`` still discovers ``project/AGENTS.md``."""
try:
p = Path(raw_path).expanduser()
if not p.is_absolute():
@@ -134,10 +113,9 @@ class SubdirectoryHintTracker:
break
if self._is_valid_subdir(p):
candidates.add(p)
parent = p.parent
if parent == p:
if p.parent == p:
break # filesystem root
p = parent
p = p.parent
except (OSError, ValueError, RuntimeError):
pass
@@ -148,25 +126,22 @@ class SubdirectoryHintTracker:
except ValueError:
tokens = cmd.split()
for token in tokens:
if token.startswith("-"):
continue
if "/" not in token and "." not in token:
continue
if token.startswith(("http://", "https://", "git@")):
if token.startswith(("-", "http://", "https://", "git@")) or ("/" not in token and "." not in token):
continue
self._add_path_candidate(token, candidates)
def _within_working_dir(self, path: Path) -> bool:
"""Reject paths outside the working-dir tree.
Loading ~/.codex/AGENTS.md or ~/.claude/CLAUDE.md would mix another
agent's instructions into this session. ``is_relative_to`` handles
symlinked paths; the ancestor check is a best-effort fallback.
"""
"""Reject paths outside the working-dir tree: loading ~/.codex/AGENTS.md
or ~/.claude/CLAUDE.md would mix another agent's instructions into this
session. Falls back to an ancestor check when ``is_relative_to`` fails."""
try:
return path.is_relative_to(self.working_dir)
except (OSError, ValueError):
return _is_ancestor_or_same(self.working_dir, path)
try:
path.relative_to(self.working_dir)
return True
except ValueError:
return False
def _is_valid_subdir(self, path: Path) -> bool:
"""Directory inside the working-dir tree, not yet loaded, not an excluded copy dir."""
@@ -175,18 +150,11 @@ class SubdirectoryHintTracker:
return False
except OSError:
return False
if path in self._loaded_dirs:
return False
if not self._within_working_dir(path):
return False
return not self._is_excluded(path)
return path not in self._loaded_dirs and self._within_working_dir(path) and not self._is_excluded(path)
def _is_excluded(self, path: Path) -> bool:
"""True when a segment *below* the working dir is an excluded copy dir.
Only segments under ``working_dir`` are screened: a user deliberately
working inside ``vendor/`` keeps that segment legitimate.
"""
"""True when a segment *below* the working dir is an excluded copy dir
(a user deliberately working inside ``vendor/`` keeps that segment legitimate)."""
try:
rel_parts = path.relative_to(self.working_dir).parts
except ValueError:
@@ -197,13 +165,8 @@ class SubdirectoryHintTracker:
"""Load the first hint file in *directory*; formatted text or None."""
self._loaded_dirs.add(directory)
if not self._within_working_dir(directory):
logger.debug(
"Skipping hint files in %s — outside working_dir %s",
directory, self.working_dir,
)
logger.debug("Skipping hint files in %s — outside working_dir %s", directory, self.working_dir)
return None
found_hints = []
for filename in _HINT_FILENAMES:
hint_path = directory / filename
try:
@@ -215,47 +178,30 @@ class SubdirectoryHintTracker:
content = hint_path.read_text(encoding="utf-8").strip()
if not content:
continue
digest = hashlib.sha256(content.encode("utf-8")).hexdigest()
digest = _digest(content)
if digest in self._loaded_digests:
logger.debug(
"Skipping duplicate hint content at %s (digest %s)",
hint_path,
digest[:12],
)
break
logger.debug("Skipping duplicate hint content at %s (digest %s)", hint_path, digest[:12])
return None
self._loaded_digests.add(digest)
# Same security scan as startup context loading.
content = _scan_context_content(content, filename)
if len(content) > _MAX_HINT_CHARS:
content = (
content[:_MAX_HINT_CHARS]
+ f"\n\n[...truncated {filename}: {len(content):,} chars total]"
)
rel_path = str(hint_path)
try:
rel_path = str(hint_path.relative_to(self.working_dir))
except (ValueError, RuntimeError):
try:
# as_posix: "~/" shorthand implies POSIX rendering
# (avoids ~/AppData\Local\... chimeras on Windows).
rel_path = "~/" + hint_path.relative_to(Path.home()).as_posix()
except (ValueError, RuntimeError):
pass # keep absolute
found_hints.append((rel_path, content))
break # first match wins per directory (like startup loading)
content = content[:_MAX_HINT_CHARS] + f"\n\n[...truncated {filename}: {len(content):,} chars total]"
rel_path = self._display_path(hint_path)
logger.debug("Loaded subdirectory hints from %s: %s", directory, [rel_path])
return f"[Subdirectory context discovered: {rel_path}]\n{content}" # first match wins per directory
except Exception as exc:
logger.debug("Could not read %s: %s", hint_path, exc)
return None
if not found_hints:
return None
sections = [
f"[Subdirectory context discovered: {rel_path}]\n{content}"
for rel_path, content in found_hints
]
logger.debug(
"Loaded subdirectory hints from %s: %s",
directory,
[h[0] for h in found_hints],
)
return "\n\n".join(sections)
def _display_path(self, hint_path: Path) -> str:
"""Working-dir-relative, else ``~/``-relative (POSIX rendering so Windows
never shows ``~/AppData\\Local\\...`` chimeras), else absolute."""
try:
return str(hint_path.relative_to(self.working_dir))
except (ValueError, RuntimeError):
pass
try:
return "~/" + hint_path.relative_to(Path.home()).as_posix()
except (ValueError, RuntimeError):
return str(hint_path)