From 042aea505d253b9b4ced4e19e16031f6f303eb08 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:13:55 -0700 Subject: [PATCH] refactor(agent/subdirectory_hints,skill_preprocessing): fold hint-file lookup, compact exception ladders and docs --- agent/skill_preprocessing.py | 89 +++++----------- agent/subdirectory_hints.py | 198 +++++++++++++---------------------- 2 files changed, 97 insertions(+), 190 deletions(-) diff --git a/agent/skill_preprocessing.py b/agent/skill_preprocessing.py index 44c5714b9b..eadc425aa9 100644 --- a/agent/skill_preprocessing.py +++ b/agent/skill_preprocessing.py @@ -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 diff --git a/agent/subdirectory_hints.py b/agent/subdirectory_hints.py index 6ad23a2372..e4b805cefe 100644 --- a/agent/subdirectory_hints.py +++ b/agent/subdirectory_hints.py @@ -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)