diff --git a/tests/tools/test_self_repo_guard.py b/tests/tools/test_self_repo_guard.py new file mode 100644 index 0000000000000..e4e957222043d --- /dev/null +++ b/tests/tools/test_self_repo_guard.py @@ -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 diff --git a/tools/self_repo_guard.py b/tools/self_repo_guard.py new file mode 100644 index 0000000000000..f2e06de579be1 --- /dev/null +++ b/tools/self_repo_guard.py @@ -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 && 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 ` or a temp clone. " + "If this checkout itself must change, ask the user to run the " + "command outside hermes and restart hermes afterwards." + )