diff --git a/engine/hooks/auto-pr/detect.py b/engine/hooks/auto-pr/detect.py index c40fe62a..69fa398f 100644 --- a/engine/hooks/auto-pr/detect.py +++ b/engine/hooks/auto-pr/detect.py @@ -20,6 +20,7 @@ from __future__ import annotations import hashlib +import json import os import subprocess import sys @@ -32,6 +33,8 @@ HERE = os.path.dirname(os.path.realpath(__file__)) OWN_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(HERE))) +TRANSCRIPT_TAIL_LINES = 4000 + STATE_DIR = os.environ.get( "AUTO_PR_STATE_DIR", os.path.join(os.path.expanduser("~"), ".cache", "catstack-auto-pr"), @@ -74,14 +77,9 @@ def _run_git(root: str, *args: str) -> str | None: return result.stdout -def repo_root(payload: dict) -> str | None: - """The catstack checkout/worktree root, only for this repository.""" - cwd = payload.get("cwd") or payload.get("workspace_roots") - if isinstance(cwd, list): - cwd = cwd[0] if cwd else "" - if not isinstance(cwd, str) or not cwd: - return None - out = _run_git(cwd, "rev-parse", "--show-toplevel") +def _accept_root(directory: str) -> str | None: + """`directory`'s repo root, but only when that repo is catstack.""" + out = _run_git(directory, "rev-parse", "--show-toplevel") if not out: return None try: @@ -101,6 +99,72 @@ def repo_root(payload: dict) -> str | None: return candidate if candidate_common_path == own_common_path else None +def touched_dirs(payload: dict) -> list[str]: + """Directories this session's tool calls wrote or read, newest first. + + A session whose cwd is another repository can still edit catstack through + an absolute path or a worktree, and cwd alone cannot see that. The paths + the session actually named can. + """ + path = payload.get("transcript_path") or payload.get("transcriptPath") or "" + if not isinstance(path, str) or not path: + return [] + try: + with open(path, encoding="utf-8") as handle: + lines = handle.readlines() + except OSError as exc: + sys.stderr.write(f"auto-pr: cannot read transcript {path}: {exc}\n") + return [] + + seen: list[str] = [] + for raw in reversed(lines[-TRANSCRIPT_TAIL_LINES:]): + try: + data = json.loads(raw) + except (json.JSONDecodeError, TypeError): + continue + for named in _tool_input_paths(data): + directory = named if os.path.isdir(named) else os.path.dirname(named) + if directory and directory not in seen: + seen.append(directory) + return seen + + +def _tool_input_paths(entry) -> list[str]: + if not isinstance(entry, dict): + return [] + found: list[str] = [] + content = entry.get("message", {}).get("content") if isinstance(entry.get("message"), dict) else None + blocks = content if isinstance(content, list) else [] + for block in blocks: + if not isinstance(block, dict) or block.get("type") != "tool_use": + continue + tool_input = block.get("input") + if not isinstance(tool_input, dict): + continue + for key in ("file_path", "path", "notebook_path"): + value = tool_input.get(key) + if isinstance(value, str) and value.startswith("/"): + found.append(os.path.realpath(value)) + return found + + +def repo_root(payload: dict) -> str | None: + """The catstack checkout/worktree root, only for this repository.""" + cwd = payload.get("cwd") or payload.get("workspace_roots") + if isinstance(cwd, list): + cwd = cwd[0] if cwd else "" + if isinstance(cwd, str) and cwd: + from_cwd = _accept_root(cwd) + if from_cwd: + return from_cwd + + for directory in touched_dirs(payload): + from_touch = _accept_root(directory) + if from_touch: + return from_touch + return None + + def current_branch(root: str) -> str: out = _run_git(root, "branch", "--show-current") return (out or "").strip() or "HEAD" diff --git a/engine/hooks/auto-pr/tests/test_hooks.py b/engine/hooks/auto-pr/tests/test_hooks.py index c234e7c5..4fd97186 100644 --- a/engine/hooks/auto-pr/tests/test_hooks.py +++ b/engine/hooks/auto-pr/tests/test_hooks.py @@ -321,3 +321,60 @@ def test_subagent_payload_is_silent_and_leaves_the_marker_for_the_parent(self): blocked, err = run_claude({"cwd": self.repo}) self.assertTrue(blocked) self.assertIn("catstack changes detected", err) + + +class ForeignSessionRootTests(unittest.TestCase): + """A session whose cwd is another repo can still edit catstack. cwd alone + cannot see that, so the paths the session named are the fallback signal. + """ + + def _transcript(self, *paths: str) -> str: + handle = tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) + for path in paths: + handle.write(json.dumps({ + "type": "assistant", + "message": {"content": [ + {"type": "tool_use", "name": "Edit", "input": {"file_path": path}}, + ]}, + }) + "\n") + handle.close() + self.addCleanup(os.unlink, handle.name) + return handle.name + + def test_cwd_inside_catstack_still_wins(self) -> None: + self.assertEqual( + detect.repo_root({"cwd": detect.OWN_REPO_ROOT}), + detect.OWN_REPO_ROOT, + ) + + def test_foreign_cwd_with_no_transcript_is_silent(self) -> None: + self.assertIsNone(detect.repo_root({"cwd": tempfile.gettempdir()})) + + def test_foreign_cwd_detects_catstack_from_a_touched_path(self) -> None: + touched = os.path.join(detect.OWN_REPO_ROOT, "install.sh") + root = detect.repo_root({ + "cwd": tempfile.gettempdir(), + "transcript_path": self._transcript(touched), + }) + self.assertEqual(root, detect.OWN_REPO_ROOT) + + def test_touched_paths_outside_catstack_stay_silent(self) -> None: + outside = os.path.join(tempfile.gettempdir(), "somewhere", "file.py") + self.assertIsNone(detect.repo_root({ + "cwd": tempfile.gettempdir(), + "transcript_path": self._transcript(outside), + })) + + def test_unreadable_transcript_is_reported_not_swallowed(self) -> None: + buffer = io.StringIO() + with redirect_stderr(buffer): + dirs = detect.touched_dirs({"transcript_path": "/nope/missing.jsonl"}) + self.assertEqual(dirs, []) + self.assertIn("cannot read transcript", buffer.getvalue()) + + def test_malformed_transcript_line_does_not_crash(self) -> None: + handle = tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) + handle.write("{not json}\n") + handle.close() + self.addCleanup(os.unlink, handle.name) + self.assertEqual(detect.touched_dirs({"transcript_path": handle.name}), [])