diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 438e97f..f173581 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -260,6 +260,7 @@ import role_session_router # noqa: E402 import role_namespace_gate # noqa: E402 import task_capability_map # noqa: E402 import review_proofs # noqa: E402 +import issue_lock_worktree # noqa: E402 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue, @@ -683,6 +684,7 @@ def gitea_lock_issue( host: str | None = None, org: str | None = None, repo: str | None = None, + worktree_path: str | None = None, ) -> dict: """Lock exactly one Gitea issue and its branch name to ensure durable tracking. @@ -693,6 +695,8 @@ def gitea_lock_issue( host: Override Gitea host. org: Override Org. repo: Override Repo. + worktree_path: Author scratch-clone path to validate (defaults to + GITEA_AUTHOR_WORKTREE or the MCP server project root). """ # 1. Enforce branch name includes issue number expected_pattern = f"issue-{issue_number}" @@ -701,6 +705,20 @@ def gitea_lock_issue( f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)" ) + resolved_worktree = issue_lock_worktree.resolve_author_worktree_path( + worktree_path, PROJECT_ROOT + ) + git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree) + lock_assessment = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path=resolved_worktree, + current_branch=git_state.get("current_branch"), + porcelain_status=git_state.get("porcelain_status") or "", + ) + if lock_assessment["block"]: + raise RuntimeError( + issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment) + ) + # 2. Check if the issue already has an open PR (reuse protection) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -737,6 +755,7 @@ def gitea_lock_issue( "remote": remote, "org": o, "repo": r, + "worktree_path": resolved_worktree, } try: @@ -747,9 +766,13 @@ def gitea_lock_issue( return { "success": True, - "message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).", + "message": ( + f"Successfully locked issue #{issue_number} to branch '{branch_name}' " + f"from worktree '{resolved_worktree}' (fail-closed check complete)." + ), "issue_number": issue_number, "branch_name": branch_name, + "worktree_path": resolved_worktree, } @@ -763,6 +786,7 @@ def gitea_create_pr( host: str | None = None, org: str | None = None, repo: str | None = None, + worktree_path: str | None = None, ) -> dict: """Create a pull request on a Gitea repository. @@ -775,6 +799,7 @@ def gitea_create_pr( host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. + worktree_path: Author worktree path; must match the path stored at lock time. Returns: dict with 'number' of the created PR ('url' only with the reveal opt-in). @@ -813,6 +838,13 @@ def gitea_create_pr( locked_issue = lock_data.get("issue_number") locked_branch = lock_data.get("branch_name") + locked_worktree = lock_data.get("worktree_path") + + worktree_check = issue_lock_worktree.verify_pr_worktree_matches_lock( + locked_worktree, worktree_path, PROJECT_ROOT + ) + if worktree_check["block"]: + raise ValueError(worktree_check["reasons"][0]) if head != locked_branch: raise ValueError( diff --git a/issue_lock_worktree.py b/issue_lock_worktree.py new file mode 100644 index 0000000..219c6b9 --- /dev/null +++ b/issue_lock_worktree.py @@ -0,0 +1,156 @@ +"""Issue-lock worktree validation (#249). + +Author issue locks must validate the caller's own scratch clone (or declared +worktree path), not the shared MCP server working directory. A clean scratch at +``master``/``main`` must remain lockable while an unrelated session leaves the +shared dev worktree dirty or on a feature branch. +""" + +from __future__ import annotations + +import os +import subprocess + +from reviewer_worktree import parse_dirty_tracked_files + +AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" +BASE_BRANCHES = frozenset({"master", "main"}) + + +def resolve_author_worktree_path( + explicit: str | None, + project_root: str, +) -> str: + """Resolve the author worktree path for lock/PR gates.""" + path = (explicit or "").strip() + if not path: + path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip() + if not path: + path = project_root + return os.path.realpath(os.path.abspath(path)) + + +def read_worktree_git_state(worktree_path: str) -> dict: + """Read branch name and porcelain status from a git worktree.""" + path = (worktree_path or "").strip() + if not path: + return {"current_branch": None, "porcelain_status": ""} + + branch_res = subprocess.run( + ["git", "-C", path, "branch", "--show-current"], + capture_output=True, + text=True, + check=False, + ) + current_branch = (branch_res.stdout or "").strip() or None + + status_res = subprocess.run( + ["git", "-C", path, "status", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + return { + "current_branch": current_branch, + "porcelain_status": status_res.stdout or "", + } + + +def assess_issue_lock_worktree( + *, + worktree_path: str, + current_branch: str | None, + porcelain_status: str, + base_branches: frozenset[str] | None = None, +) -> dict: + """Fail closed when lock preconditions are not met on the declared worktree.""" + bases = base_branches or BASE_BRANCHES + reasons: list[str] = [] + path = (worktree_path or "").strip() + if not path: + reasons.append("worktree path not declared for issue lock; fail closed") + return _assessment(False, reasons, path, None, []) + + branch = (current_branch or "").strip() + dirty_files = parse_dirty_tracked_files(porcelain_status) + + if dirty_files: + reasons.append( + "tracked file edits exist before issue lock; " + "lock must precede implementation work" + ) + if not branch: + reasons.append( + "current branch unknown (detached HEAD?); issue lock must be taken " + f"from base branch ({_base_list(bases)})" + ) + elif branch not in bases: + reasons.append( + f"issue lock must be taken from base branch ({_base_list(bases)}), " + f"not '{branch}'" + ) + + proven = not reasons + return _assessment(proven, reasons, path, branch or None, dirty_files) + + +def format_issue_lock_worktree_error(assessment: dict) -> str: + """Format a single fail-closed error for ``gitea_lock_issue``.""" + reasons = list(assessment.get("reasons") or []) + if not reasons: + reasons = ["issue lock worktree validation failed"] + return "; ".join(reasons) + " (fail closed)" + + +def verify_pr_worktree_matches_lock( + locked_worktree_path: str | None, + declared_worktree_path: str | None, + project_root: str, +) -> dict: + """PR creation must use the same worktree the lock was validated against.""" + locked = (locked_worktree_path or "").strip() + if not locked: + return {"proven": True, "block": False, "reasons": []} + + declared = resolve_author_worktree_path(declared_worktree_path, project_root) + locked_real = os.path.realpath(locked) + declared_real = os.path.realpath(declared) + if locked_real != declared_real: + return { + "proven": False, + "block": True, + "reasons": [ + f"PR worktree '{declared_real}' does not match locked worktree " + f"'{locked_real}' (fail closed)" + ], + "locked_worktree_path": locked_real, + "declared_worktree_path": declared_real, + } + return { + "proven": True, + "block": False, + "reasons": [], + "locked_worktree_path": locked_real, + "declared_worktree_path": declared_real, + } + + +def _base_list(bases: frozenset[str]) -> str: + return "/".join(sorted(bases)) + + +def _assessment( + proven: bool, + reasons: list[str], + worktree_path: str, + current_branch: str | None, + dirty_files: list[str], +) -> dict: + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "worktree_path": worktree_path or None, + "current_branch": current_branch, + "dirty_files": dirty_files, + } \ No newline at end of file diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 5d9d2fb..1a5184f 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -146,6 +146,15 @@ Worktree folder = branch with `/` replaced by `-` 2. Fetch/prune: `git fetch --prune`. 3. Confirm local `master` equals remote `master` (`git rev-list --left-right --count /master...master` → `0 0`). 4. Create/claim the issue (§A). +4b. **Issue lock from your scratch clone (#249):** when using + `gitea_lock_issue`, pass `worktree_path` pointing at your own clean + scratch clone (or set `GITEA_AUTHOR_WORKTREE`). The lock gate validates + *that* path — clean tree on `master`/`main`, no tracked edits yet — + not the shared MCP/orchestration checkout. Another session's dirty + feature branch in the shared dev worktree must not block your lock. + Never stash, reset, or checkout files in the shared worktree to satisfy + the gate. Pass the same `worktree_path` to `gitea_create_pr` so the PR + gate matches the lock record. 5. Create the isolated worktree (§B) from latest remote `master`. 6. Implement the narrow scope only — no unrelated refactors or formatting churn. 7. Add/update focused tests when behavior changes. @@ -311,7 +320,9 @@ When in doubt, stop and surface the discrepancy; do not guess or work around a g ## I. Recovery patterns - **Dirty worktree from another issue:** do not touch it. Start your issue in its - own new worktree; unrelated dirty work must not block you. + own new worktree; unrelated dirty work must not block you. For Gitea-Tools + author flows, lock the issue from your scratch clone (`worktree_path` on + `gitea_lock_issue`) — do not manipulate the shared dev checkout. - **Local `master` ahead of remote unexpectedly:** do not push `master`. Confirm the commits are preserved on a feature branch (local + remote) first, then `git reset --hard /master` to realign. Never discard commits that are diff --git a/tests/test_issue_lock_worktree.py b/tests/test_issue_lock_worktree.py new file mode 100644 index 0000000..ae7c027 --- /dev/null +++ b/tests/test_issue_lock_worktree.py @@ -0,0 +1,99 @@ +"""Tests for issue-lock worktree validation (#249).""" + +import os +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import issue_lock_worktree # noqa: E402 + + +class TestIssueLockWorktreeAssessment(unittest.TestCase): + def test_clean_base_branch_passes(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path="/scratch/wt", + current_branch="master", + porcelain_status="", + ) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_dirty_tracked_files_fail(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path="/scratch/wt", + current_branch="master", + porcelain_status=" M gitea_mcp_server.py\n", + ) + self.assertFalse(result["proven"]) + self.assertIn("tracked file edits exist before issue lock", result["reasons"][0]) + + def test_feature_branch_fails(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path="/scratch/wt", + current_branch="feat/issue-243-forbidden-git-gaps", + porcelain_status="", + ) + self.assertFalse(result["proven"]) + self.assertIn("issue lock must be taken from base branch", result["reasons"][0]) + + def test_untracked_files_do_not_block(self): + result = issue_lock_worktree.assess_issue_lock_worktree( + worktree_path="/scratch/wt", + current_branch="main", + porcelain_status="?? notes.txt\n", + ) + self.assertTrue(result["proven"]) + + +class TestIssueLockWorktreeResolution(unittest.TestCase): + def test_explicit_path_wins(self): + resolved = issue_lock_worktree.resolve_author_worktree_path( + "/tmp/scratch/wt", "/shared/dev" + ) + self.assertEqual(resolved, os.path.realpath("/tmp/scratch/wt")) + + def test_env_var_when_explicit_missing(self): + with patch.dict(os.environ, {"GITEA_AUTHOR_WORKTREE": "/tmp/from-env"}): + resolved = issue_lock_worktree.resolve_author_worktree_path( + None, "/shared/dev" + ) + self.assertEqual(resolved, os.path.realpath("/tmp/from-env")) + + def test_project_root_fallback(self): + resolved = issue_lock_worktree.resolve_author_worktree_path( + None, "/shared/dev" + ) + self.assertEqual(resolved, os.path.realpath("/shared/dev")) + + +class TestPrWorktreeMatch(unittest.TestCase): + def test_matching_paths_pass(self): + result = issue_lock_worktree.verify_pr_worktree_matches_lock( + "/tmp/scratch/wt", + "/tmp/scratch/wt", + "/shared/dev", + ) + self.assertTrue(result["proven"]) + + def test_mismatch_fails(self): + result = issue_lock_worktree.verify_pr_worktree_matches_lock( + "/tmp/scratch/wt", + "/shared/dev", + "/shared/dev", + ) + self.assertFalse(result["proven"]) + self.assertIn("does not match locked worktree", result["reasons"][0]) + + def test_missing_locked_path_skips_check(self): + result = issue_lock_worktree.verify_pr_worktree_matches_lock( + None, + "/any/path", + "/shared/dev", + ) + self.assertTrue(result["proven"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 278b96c..328e749 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -2934,26 +2934,41 @@ class TestVerifyMutationAuthority(unittest.TestCase): class TestIssueLocking(unittest.TestCase): """Test issue locking and PR gating constraints.""" + @staticmethod + def _clean_master_git_state(): + return {"current_branch": "master", "porcelain_status": ""} + def tearDown(self): if os.path.exists(ISSUE_LOCK_FILE): os.remove(ISSUE_LOCK_FILE) + @patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={"current_branch": "master", "porcelain_status": ""}, + ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_lock_issue_success(self, _auth, mock_api): + def test_lock_issue_success(self, _auth, mock_api, _git_state): mock_api.return_value = [] # no open PRs res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") self.assertTrue(res["success"]) self.assertTrue(os.path.exists(ISSUE_LOCK_FILE)) + with open(ISSUE_LOCK_FILE, encoding="utf-8") as f: + lock = json.load(f) + self.assertIn("worktree_path", lock) def test_lock_issue_mismatch_branch_fails(self): with self.assertRaises(ValueError) as ctx: gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs") self.assertIn("must contain locked issue pattern", str(ctx.exception)) + @patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={"current_branch": "master", "porcelain_status": ""}, + ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api): + def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api, _git_state): mock_api.return_value = [{ "number": 200, "head": {"ref": "feat/issue-196-boundary"}, @@ -2964,9 +2979,13 @@ class TestIssueLocking(unittest.TestCase): gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") self.assertIn("already tied to an open PR", str(ctx.exception)) + @patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={"current_branch": "master", "porcelain_status": ""}, + ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api): + def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api, _git_state): mock_api.return_value = [{ "number": 200, "head": {"ref": "feat/other-branch"}, @@ -2977,6 +2996,62 @@ class TestIssueLocking(unittest.TestCase): gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") self.assertIn("already tied to an open PR", str(ctx.exception)) + @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_from_clean_scratch_worktree(self, _auth, _api): + scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean" + with patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={"current_branch": "master", "porcelain_status": ""}, + ) as mock_git: + res = gitea_lock_issue( + issue_number=249, + branch_name="feat/issue-249-issue-lock-scratch-worktree", + remote="prgs", + worktree_path=scratch, + ) + self.assertTrue(res["success"]) + self.assertEqual(res["worktree_path"], os.path.realpath(scratch)) + mock_git.assert_called_once_with(os.path.realpath(scratch)) + + @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_fails_when_declared_worktree_dirty(self, _auth, _api): + with patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={ + "current_branch": "master", + "porcelain_status": " M gitea_mcp_server.py\n", + }, + ): + with self.assertRaises(RuntimeError) as ctx: + gitea_lock_issue( + issue_number=249, + branch_name="feat/issue-249-issue-lock-scratch-worktree", + remote="prgs", + worktree_path="/tmp/scratch/wt", + ) + self.assertIn("tracked file edits exist before issue lock", str(ctx.exception)) + + @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_lock_fails_when_declared_worktree_not_on_base(self, _auth, _api): + with patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={ + "current_branch": "feat/issue-243-forbidden-git-gaps", + "porcelain_status": "", + }, + ): + with self.assertRaises(RuntimeError) as ctx: + gitea_lock_issue( + issue_number=249, + branch_name="feat/issue-249-issue-lock-scratch-worktree", + remote="prgs", + worktree_path="/tmp/scratch/wt", + ) + self.assertIn("issue lock must be taken from base branch", str(ctx.exception)) + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -3025,6 +3100,49 @@ class TestIssueLocking(unittest.TestCase): gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs") self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception)) + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_create_pr_worktree_mismatch_fails(self, _auth, _role): + scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-pr") + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(_sample_issue_lock( + issue_number=249, + branch_name="feat/issue-249-issue-lock-scratch-worktree", + worktree_path=scratch, + ), f) + with patch.dict(os.environ, CREATE_PR_ENV, clear=True): + with self.assertRaises(ValueError) as ctx: + gitea_create_pr( + title="feat: lock scratch worktree Closes #249", + head="feat/issue-249-issue-lock-scratch-worktree", + remote="prgs", + worktree_path="/tmp/other-scratch", + ) + self.assertIn("does not match locked worktree", str(ctx.exception)) + + @patch("mcp_server.api_request") + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_create_pr_honors_scratch_worktree_lock(self, _auth, _role, mock_api): + scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-e2e") + mock_api.return_value = {"number": 250, "html_url": "https://example/pr/250"} + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(_sample_issue_lock( + issue_number=249, + branch_name="feat/issue-249-issue-lock-scratch-worktree", + worktree_path=scratch, + ), f) + with patch.dict(os.environ, CREATE_PR_ENV, clear=True): + res = gitea_create_pr( + title="feat: issue-lock scratch worktree Closes #249", + head="feat/issue-249-issue-lock-scratch-worktree", + remote="prgs", + worktree_path=scratch, + ) + self.assertEqual(res["number"], 250) + # --------------------------------------------------------------------------- # Pre-flight ordering and workspace edit block (#210)