diff --git a/skills/kvb-org-organize/scripts/council.py b/skills/kvb-org-organize/scripts/council.py index 63280b3..84956cb 100755 --- a/skills/kvb-org-organize/scripts/council.py +++ b/skills/kvb-org-organize/scripts/council.py @@ -11,7 +11,9 @@ Environment: Exit codes: 0 done, 2 usage error, 3 office held by a live session, 4 office held by a dead session (take it over with `takeover`), - gate: 2 blocks the tool call. + gate: 2 blocks the tool call: an edit without a claim of this session + below a directory whose .git is a directory holding HEAD or a file + starting with gitdir:. """ from __future__ import annotations @@ -284,8 +286,21 @@ def cmd_list(args: argparse.Namespace) -> int: return 0 +def is_git_marker(marker: Path) -> bool: + if marker.is_dir(): + return (marker / "HEAD").is_file() + # is_file() first: opening a planted FIFO named .git would hang this hook. + if not marker.is_file(): + return False + try: + with marker.open("rb") as handle: + return handle.read(7) == b"gitdir:" + except OSError: + return False + + def in_git_worktree(target: Path) -> bool: - return any((path / ".git").exists() for path in (target, *target.parents)) + return any(is_git_marker(path / ".git") for path in (target, *target.parents)) def cmd_gate(_: argparse.Namespace) -> int: diff --git a/tests/test_council.py b/tests/test_council.py index 1bd9b0e..77297f9 100644 --- a/tests/test_council.py +++ b/tests/test_council.py @@ -48,8 +48,15 @@ class TestCouncil(unittest.TestCase): input=stdin, cwd=cwd, check=False, + timeout=30, ) + def _repo(self) -> Path: + repo = Path(self.tmp.name) / "repo" + (repo / ".git").mkdir(parents=True) + (repo / ".git" / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8") + return repo + def test_first_claim_wins_and_second_is_refused(self): self.assertEqual(self._run("s1", "claim", "work/x", "--work", "w").returncode, 0) second = self._run("s2", "claim", "work/x", "--work", "w") @@ -131,8 +138,7 @@ class TestCouncil(unittest.TestCase): self.assertEqual(sorted(self.root.rglob("*")), []) def test_gate_blocks_unclaimed_edits_in_a_git_worktree_only(self): - repo = Path(self.tmp.name) / "repo" - (repo / ".git").mkdir(parents=True) + repo = self._repo() inside = json.dumps({"session_id": "s1", "tool_input": {"file_path": str(repo / "f")}}) outside = json.dumps( {"session_id": "s1", "tool_input": {"file_path": str(self.home / "f")}} @@ -142,9 +148,20 @@ class TestCouncil(unittest.TestCase): self._run("s1", "claim", "work/x", "--work", "w") self.assertEqual(self._run("s1", "gate", stdin=inside).returncode, 0) + def test_gate_counts_only_a_git_dir_holding_head_or_a_gitdir_file(self): + base = Path(self.tmp.name) + (base / "stray" / ".git").mkdir(parents=True) + (base / "linked").mkdir() + (base / "linked" / ".git").write_text("gitdir: /elsewhere/.git/worktrees/linked\n", encoding="utf-8") + (base / "fifo").mkdir() + os.mkfifo(base / "fifo" / ".git", 0o600) + for folder, code in (("stray", 0), ("linked", 2), ("fifo", 0)): + with self.subTest(folder=folder): + edit = json.dumps({"session_id": "s1", "tool_input": {"file_path": str(base / folder / "f")}}) + self.assertEqual(self._run("s1", "gate", stdin=edit).returncode, code) + def test_gate_still_blocks_past_a_planted_fifo_or_malformed_claim(self): - repo = Path(self.tmp.name) / "repo" - (repo / ".git").mkdir(parents=True) + repo = self._repo() inside = json.dumps({"session_id": "s1", "tool_input": {"file_path": str(repo / "f")}}) (self.root / "work" / "fifo").mkdir(parents=True) os.mkfifo(self.root / "work" / "fifo" / "claim.json", 0o600)