fix(auto-pr): detect catstack work from a session in another repo

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"

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}), [])