feat(tools): detect git mutations targeting the running source checkout
When hermes runs from a source/editable install, a git checkout/reset/pull in its own repo swaps code on disk under the live interpreter. Modules imported before the switch stay old while later lazy imports load new code, producing delayed signature TypeErrors and tracebacks that don't match the source, typically losing the in-flight turn. New tools/self_repo_guard.py detects working-tree/ref mutations (checkout, switch, reset, rebase, merge, pull, restore, stash, clean, cherry-pick, revert) whose target repo is the source root the process runs from, via cwd, git -C, cd chains, and subshell segments. Read-only git, commits, fetch, and git worktree add stay allowed; packaged installs (no .git) are inert.
This commit is contained in:
Erosika
kshitij
parent
ad59d55338
commit
206531a1e1
|
|
@ -0,0 +1,157 @@
|
|||
"""Tests for tools/self_repo_guard.py — the running-source-checkout git guard."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tools.self_repo_guard import (
|
||||
detect_self_repo_git_mutation,
|
||||
get_running_source_root,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repo(tmp_path):
|
||||
"""A fake source checkout acting as the running install's repo root."""
|
||||
root = tmp_path / "hermes-agent"
|
||||
(root / ".git").mkdir(parents=True)
|
||||
(root / "agent").mkdir()
|
||||
return root.resolve()
|
||||
|
||||
|
||||
def _detect(command, cwd, root):
|
||||
return detect_self_repo_git_mutation(command, str(cwd), source_root=root)
|
||||
|
||||
|
||||
class TestBlocksMutationsInSourceRepo:
|
||||
@pytest.mark.parametrize("sub", [
|
||||
"checkout pr-51020",
|
||||
"switch main",
|
||||
"reset --hard origin/main",
|
||||
"rebase origin/main",
|
||||
"merge origin/main",
|
||||
"pull",
|
||||
"restore .",
|
||||
"stash",
|
||||
"stash pop",
|
||||
"clean -fd",
|
||||
"cherry-pick abc123",
|
||||
"revert HEAD",
|
||||
])
|
||||
def test_cwd_inside_repo(self, repo, sub):
|
||||
hit, msg = _detect(f"git {sub}", repo, repo)
|
||||
assert hit is True
|
||||
assert str(repo) in msg
|
||||
|
||||
def test_cwd_in_repo_subdirectory(self, repo):
|
||||
hit, _ = _detect("git checkout main", repo / "agent", repo)
|
||||
assert hit is True
|
||||
|
||||
def test_dash_c_targeting_repo_from_outside(self, repo, tmp_path):
|
||||
hit, _ = _detect(f"git -C {repo} checkout pr-51020", tmp_path, repo)
|
||||
assert hit is True
|
||||
|
||||
def test_cd_into_repo_then_checkout(self, repo, tmp_path):
|
||||
hit, _ = _detect(f"cd {repo} && git checkout pr-51020", tmp_path, repo)
|
||||
assert hit is True
|
||||
|
||||
def test_relative_cd_into_repo(self, repo):
|
||||
hit, _ = _detect("cd hermes-agent && git pull", repo.parent, repo)
|
||||
assert hit is True
|
||||
|
||||
def test_mutation_after_safe_command(self, repo):
|
||||
hit, _ = _detect("git status; git reset --hard HEAD~1", repo, repo)
|
||||
assert hit is True
|
||||
|
||||
def test_wrapped_in_sudo_env(self, repo):
|
||||
hit, _ = _detect("sudo env GIT_PAGER=cat git checkout main", repo, repo)
|
||||
assert hit is True
|
||||
|
||||
def test_tilde_dash_c_path(self, repo, monkeypatch, tmp_path):
|
||||
monkeypatch.setenv("HOME", str(repo.parent))
|
||||
hit, _ = _detect("git -C ~/hermes-agent checkout main", tmp_path, repo)
|
||||
assert hit is True
|
||||
|
||||
|
||||
class TestAllowsSafeCommands:
|
||||
@pytest.mark.parametrize("cmd", [
|
||||
"git status",
|
||||
"git log --oneline -5",
|
||||
"git diff main...HEAD",
|
||||
"git branch --show-current",
|
||||
"git stash list",
|
||||
"git stash show -p",
|
||||
"git commit -m 'msg'",
|
||||
"git add -A",
|
||||
"git fetch origin main",
|
||||
"git worktree add /tmp/wt feature-branch",
|
||||
"git push fork feature-branch",
|
||||
"ls -la",
|
||||
"grep -rn checkout tools/",
|
||||
])
|
||||
def test_read_only_and_dev_loop_in_repo(self, repo, cmd):
|
||||
hit, _ = _detect(cmd, repo, repo)
|
||||
assert hit is False
|
||||
|
||||
def test_mutation_in_other_repo(self, repo, tmp_path):
|
||||
other = tmp_path / "other-project"
|
||||
other.mkdir()
|
||||
hit, _ = _detect("git checkout main", other, repo)
|
||||
assert hit is False
|
||||
|
||||
def test_dash_c_redirects_out_of_repo(self, repo, tmp_path):
|
||||
hit, _ = _detect(f"git -C {tmp_path} checkout main", repo, repo)
|
||||
assert hit is False
|
||||
|
||||
def test_cd_out_of_repo_then_checkout(self, repo, tmp_path):
|
||||
hit, _ = _detect(f"cd {tmp_path} && git checkout main", repo, repo)
|
||||
assert hit is False
|
||||
|
||||
def test_mentioning_repo_path_without_targeting_it(self, repo, tmp_path):
|
||||
hit, _ = _detect(f"echo {repo} && git checkout main", tmp_path, repo)
|
||||
assert hit is False
|
||||
|
||||
def test_checkout_as_grep_pattern_not_git(self, repo):
|
||||
hit, _ = _detect("grep checkout file.txt", repo, repo)
|
||||
assert hit is False
|
||||
|
||||
def test_empty_command(self, repo):
|
||||
hit, _ = _detect("", repo, repo)
|
||||
assert hit is False
|
||||
|
||||
def test_packaged_install_is_inert(self, monkeypatch, tmp_path):
|
||||
import tools.self_repo_guard as mod
|
||||
monkeypatch.setattr(mod, "get_running_source_root", lambda: None)
|
||||
hit, msg = mod.detect_self_repo_git_mutation("git checkout main", str(tmp_path))
|
||||
assert hit is False
|
||||
assert msg is None
|
||||
|
||||
|
||||
class TestSourceRootResolution:
|
||||
def test_resolves_to_repo_when_git_dir_present(self):
|
||||
# The test suite itself runs from a source checkout, so the resolver
|
||||
# must find a root whose .git exists.
|
||||
root = get_running_source_root()
|
||||
if root is not None:
|
||||
assert (root / ".git").exists()
|
||||
|
||||
def test_worktree_git_file_counts(self, tmp_path, monkeypatch):
|
||||
import tools.self_repo_guard as mod
|
||||
root = tmp_path / "wt"
|
||||
root.mkdir()
|
||||
(root / ".git").write_text("gitdir: /somewhere/.git/worktrees/wt\n")
|
||||
(root / "tools").mkdir()
|
||||
fake_file = root / "tools" / "self_repo_guard.py"
|
||||
fake_file.write_text("")
|
||||
monkeypatch.setattr(mod, "__file__", str(fake_file))
|
||||
assert mod.get_running_source_root() == root.resolve()
|
||||
|
||||
|
||||
class TestUnparseableCommands:
|
||||
def test_unbalanced_quotes_fall_back(self, repo):
|
||||
hit, _ = _detect('git checkout "unterminated', repo, repo)
|
||||
assert hit is True
|
||||
|
||||
def test_subshell_syntax_does_not_crash(self, repo):
|
||||
hit, _ = _detect("VAL=$(git rev-parse HEAD) git checkout main", repo, repo)
|
||||
assert hit is True
|
||||
|
|
@ -0,0 +1,180 @@
|
|||
"""Guard against git commands that rewrite the running hermes source checkout.
|
||||
|
||||
When hermes runs from a source/editable install (``pip install -e`` on a git
|
||||
checkout), the interpreter keeps resolving lazy imports from that directory
|
||||
for the lifetime of the process. A ``git checkout``/``reset``/``pull`` in
|
||||
that repo swaps the code on disk under the live process: modules imported
|
||||
before the switch stay at the old version while anything imported later
|
||||
loads the new one. The resulting version skew surfaces as delayed,
|
||||
nonsensical failures — signature TypeErrors between a caller and callee of
|
||||
the same seam, tracebacks whose lines don't match the file on disk — long
|
||||
after the command that caused them, and typically eats the in-flight turn.
|
||||
|
||||
Packaged installs are immune (site-packages has no ``.git``) so the guard
|
||||
resolves to inert there. Read-only git commands (``status``/``log``/
|
||||
``diff``/``branch``), commits, and content edits to individual files stay
|
||||
allowed — editing hermes source *files* is the normal dev loop; the hazard
|
||||
is wholesale working-tree/ref switches. ``git worktree add`` also stays
|
||||
allowed: it is the recommended alternative and the block message points to
|
||||
it.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
from pathlib import Path
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
# Working-tree/ref mutations. Not listed (deliberately allowed): commit,
|
||||
# branch, tag, fetch, worktree, apply/am (file-level edits are the normal
|
||||
# hermes-on-hermes dev loop), and all read-only subcommands.
|
||||
_MUTATING_SUBCOMMANDS = frozenset({
|
||||
"checkout", "switch", "reset", "rebase", "merge", "pull",
|
||||
"restore", "stash", "clean", "cherry-pick", "revert",
|
||||
})
|
||||
|
||||
# stash invocations that only read state.
|
||||
_STASH_READONLY = frozenset({"list", "show"})
|
||||
|
||||
_WRAPPER_COMMANDS = frozenset({"sudo", "env", "exec", "nohup", "setsid", "time", "command"})
|
||||
|
||||
# Git global flags that consume the NEXT token as their argument.
|
||||
_GIT_FLAGS_WITH_ARG = frozenset({"-C", "-c", "--git-dir", "--work-tree", "--namespace", "--exec-path"})
|
||||
|
||||
# Command separators plus subshell boundaries, so `$(git ...)` and
|
||||
# `(cd repo && git ...)` are scanned as their own segments.
|
||||
_SEGMENT_SPLIT = re.compile(r"(?:&&|\|\||;|\||\n|\$\(|\(|\)|`)")
|
||||
|
||||
|
||||
def get_running_source_root() -> Optional[Path]:
|
||||
"""Repo root of the source checkout this process runs from, or None.
|
||||
|
||||
None means a packaged install (no ``.git`` beside the code) and disables
|
||||
the guard. ``.git`` may be a directory (normal clone) or a file (linked
|
||||
worktree); both count.
|
||||
"""
|
||||
try:
|
||||
root = Path(__file__).resolve().parent.parent
|
||||
except OSError:
|
||||
return None
|
||||
return root if (root / ".git").exists() else None
|
||||
|
||||
|
||||
def _tokenize(segment: str) -> List[str]:
|
||||
try:
|
||||
return shlex.split(segment, posix=True)
|
||||
except ValueError:
|
||||
return segment.split()
|
||||
|
||||
|
||||
def _resolve(path_str: str, base: Path) -> Path:
|
||||
p = Path(os.path.expanduser(path_str))
|
||||
if not p.is_absolute():
|
||||
p = base / p
|
||||
try:
|
||||
return p.resolve()
|
||||
except OSError:
|
||||
return p
|
||||
|
||||
|
||||
def _is_within(path: Path, root: Path) -> bool:
|
||||
try:
|
||||
return path == root or path.is_relative_to(root)
|
||||
except (OSError, ValueError):
|
||||
return False
|
||||
|
||||
|
||||
def _strip_wrappers(tokens: List[str]) -> List[str]:
|
||||
i = 0
|
||||
while i < len(tokens):
|
||||
tok = tokens[i]
|
||||
if tok in _WRAPPER_COMMANDS:
|
||||
i += 1
|
||||
# env/sudo may carry VAR=VAL assignments and flags before the
|
||||
# real command.
|
||||
while i < len(tokens) and ("=" in tokens[i] or tokens[i].startswith("-")):
|
||||
i += 1
|
||||
continue
|
||||
if "=" in tok and not tok.startswith("-"):
|
||||
i += 1
|
||||
continue
|
||||
break
|
||||
return tokens[i:]
|
||||
|
||||
|
||||
def _git_target_and_subcommand(
|
||||
tokens: List[str], current_dir: Path
|
||||
) -> Tuple[Optional[Path], Optional[str], List[str]]:
|
||||
"""Parse one git invocation: (target dir, subcommand, subcommand args).
|
||||
|
||||
``-C`` entries are applied cumulatively against *current_dir*, matching
|
||||
git's own sequential ``-C`` semantics.
|
||||
"""
|
||||
target = current_dir
|
||||
i = 1
|
||||
while i < len(tokens):
|
||||
tok = tokens[i]
|
||||
if tok == "-C" and i + 1 < len(tokens):
|
||||
target = _resolve(tokens[i + 1], target)
|
||||
i += 2
|
||||
elif tok in _GIT_FLAGS_WITH_ARG and i + 1 < len(tokens):
|
||||
i += 2
|
||||
elif tok.startswith("--") and "=" in tok:
|
||||
i += 1
|
||||
elif tok.startswith("-"):
|
||||
i += 1
|
||||
else:
|
||||
return target, tok, tokens[i + 1:]
|
||||
return target, None, []
|
||||
|
||||
|
||||
def detect_self_repo_git_mutation(
|
||||
command: str,
|
||||
cwd: Optional[str],
|
||||
source_root: Optional[Path] = None,
|
||||
) -> Tuple[bool, Optional[str]]:
|
||||
"""Return (True, block message) if *command* would rewrite the source repo.
|
||||
|
||||
*cwd* is the directory the command will run in; ``cd`` segments inside
|
||||
the command are tracked so ``cd <repo> && git checkout x`` is caught.
|
||||
"""
|
||||
root = source_root if source_root is not None else get_running_source_root()
|
||||
if root is None or not command:
|
||||
return False, None
|
||||
|
||||
base = _resolve(cwd, Path("/")) if cwd else Path("/")
|
||||
current_dir = base
|
||||
|
||||
for segment in _SEGMENT_SPLIT.split(command):
|
||||
tokens = _strip_wrappers(_tokenize(segment))
|
||||
if not tokens:
|
||||
continue
|
||||
if tokens[0] == "cd":
|
||||
current_dir = _resolve(tokens[1], current_dir) if len(tokens) > 1 else current_dir
|
||||
continue
|
||||
if tokens[0] != "git":
|
||||
continue
|
||||
target, sub, sub_args = _git_target_and_subcommand(tokens, current_dir)
|
||||
if sub not in _MUTATING_SUBCOMMANDS:
|
||||
continue
|
||||
if sub == "stash" and sub_args and sub_args[0] in _STASH_READONLY:
|
||||
continue
|
||||
if target is not None and _is_within(target, root):
|
||||
return True, _block_message(sub, root)
|
||||
|
||||
return False, None
|
||||
|
||||
|
||||
def _block_message(subcommand: str, root: Path) -> str:
|
||||
return (
|
||||
f"Blocked: `git {subcommand}` would rewrite the working tree of the "
|
||||
f"hermes source checkout this process is running from ({root}). "
|
||||
"Changing the code on disk under the live interpreter causes version "
|
||||
"skew: modules already imported keep the old version while later lazy "
|
||||
"imports load the new one, producing delayed crashes (mismatched "
|
||||
"signatures, tracebacks that don't match the source) that lose the "
|
||||
"in-flight turn. Work in a separate checkout instead, e.g. "
|
||||
f"`git -C {root} worktree add <tmpdir> <branch>` or a temp clone. "
|
||||
"If this checkout itself must change, ask the user to run the "
|
||||
"command outside hermes and restart hermes afterwards."
|
||||
)
|
||||
Loading…
Reference in New Issue