Merge pull request 'feat: align claim/lock gates with branches-only worktrees (Closes #275)' (#280) from feat/issue-275-claim-lock-branches-worktree into master

tests/test_issue_lock_worktree.py

@@ -36,7 +36,32 @@ class TestIssueLockWorktreeAssessment(unittest.TestCase):
             porcelain_status="",
         )
         self.assertFalse(result["proven"])
-        self.assertIn("issue lock must be taken from base branch", result["reasons"][0])
+        self.assertIn("base-equivalence could not be proven", result["reasons"][0])
+
+    def test_base_equivalent_feature_branch_passes(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path="/repo/branches/issue-275",
+            current_branch="feat/issue-275-claim-lock-branches-worktree",
+            porcelain_status="",
+            base_equivalent=True,
+            inspected_git_root="/repo/branches/issue-275",
+            base_branch="origin/master",
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+        self.assertEqual(result["base_branch"], "origin/master")
+
+    def test_non_base_equivalent_branch_fails(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path="/repo/branches/issue-275",
+            current_branch="feat/issue-275-claim-lock-branches-worktree",
+            porcelain_status="",
+            base_equivalent=False,
+            inspected_git_root="/repo/branches/issue-275",
+            base_branch=None,
+        )
+        self.assertFalse(result["proven"])
+        self.assertIn("must be base-equivalent", result["reasons"][0])
 
     def test_untracked_files_do_not_block(self):
         result = issue_lock_worktree.assess_issue_lock_worktree(
@@ -96,4 +121,4 @@ class TestPrWorktreeMatch(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

tests/test_mcp_server.py

@@ -2937,20 +2937,31 @@ class TestVerifyMutationAuthority(unittest.TestCase):
             mcp_server.verify_mutation_authority("prgs")
 
 
+def _clean_master_git_state_for_lock():
+    return {
+        "current_branch": "master",
+        "porcelain_status": "",
+        "base_equivalent": True,
+        "inspected_git_root": "/scratch/wt",
+        "base_branch": "origin/master",
+    }
+
+
 class TestIssueLocking(unittest.TestCase):
     """Test issue locking and PR gating constraints."""
 
-    @staticmethod
-    def _clean_master_git_state():
-        return {"current_branch": "master", "porcelain_status": ""}
+    def setUp(self):
+        self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
+        self._env_patcher.start()
 
     def tearDown(self):
+        self._env_patcher.stop()
         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": ""},
+        return_value=_clean_master_git_state_for_lock(),
     )
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -2970,7 +2981,7 @@ class TestIssueLocking(unittest.TestCase):
 
     @patch(
         "mcp_server.issue_lock_worktree.read_worktree_git_state",
-        return_value={"current_branch": "master", "porcelain_status": ""},
+        return_value=_clean_master_git_state_for_lock(),
     )
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -2987,7 +2998,7 @@ class TestIssueLocking(unittest.TestCase):
 
     @patch(
         "mcp_server.issue_lock_worktree.read_worktree_git_state",
-        return_value={"current_branch": "master", "porcelain_status": ""},
+        return_value=_clean_master_git_state_for_lock(),
     )
     @patch("mcp_server.api_get_all")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -3008,7 +3019,7 @@ class TestIssueLocking(unittest.TestCase):
         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": ""},
+            return_value=_clean_master_git_state_for_lock(),
         ) as mock_git:
             res = gitea_lock_issue(
                 issue_number=249,
@@ -3028,6 +3039,7 @@ class TestIssueLocking(unittest.TestCase):
             return_value={
                 "current_branch": "master",
                 "porcelain_status": " M gitea_mcp_server.py\n",
+                "base_equivalent": True,
             },
         ):
             with self.assertRaises(RuntimeError) as ctx:
@@ -3047,6 +3059,7 @@ class TestIssueLocking(unittest.TestCase):
             return_value={
                 "current_branch": "feat/issue-243-forbidden-git-gaps",
                 "porcelain_status": "",
+                "base_equivalent": False,
             },
         ):
             with self.assertRaises(RuntimeError) as ctx:
@@ -3056,7 +3069,7 @@ class TestIssueLocking(unittest.TestCase):
                     remote="prgs",
                     worktree_path="/tmp/scratch/wt",
                 )
-        self.assertIn("issue lock must be taken from base branch", str(ctx.exception))
+        self.assertIn("base-equivalent", str(ctx.exception))
 
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []))
@@ -3199,7 +3212,12 @@ class TestPreflightVerification(unittest.TestCase):
         mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
         mcp_server._preflight_capability_violation_files = self.orig_capability_files
         mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
-        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+        for key in (
+            "GITEA_TEST_FORCE_DIRTY",
+            "GITEA_TEST_PORCELAIN",
+            "GITEA_ACTIVE_WORKTREE",
+            "GITEA_AUTHOR_WORKTREE",
+        ):
             if key in os.environ:
                 del os.environ[key]
 
@@ -3289,3 +3307,35 @@ class TestPreflightVerification(unittest.TestCase):
         status = mcp_server.assess_preflight_status()
         self.assertFalse(status["preflight_ready"])
         self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])
+
+    def test_declared_clean_task_worktree_ignores_control_checkout_violation(self):
+        """#275: active branches/ worktree is the inspected mutation workspace."""
+        import mcp_server
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_whoami_violation = True
+        mcp_server._preflight_whoami_violation_files = ["mcp_server.py"]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+
+        worktree = "/repo/branches/issue-275-clean"
+        status = mcp_server.assess_preflight_status(worktree_path=worktree)
+        self.assertTrue(status["preflight_ready"])
+        self.assertEqual(status["preflight_block_reasons"], [])
+        self.assertIn("preflight_workspace", status)
+        mcp_server.verify_preflight_purity(worktree_path=worktree)
+
+    def test_declared_dirty_task_worktree_reports_workspace_diagnostics(self):
+        """#275: dirty failures name the inspected task workspace, not just files."""
+        import mcp_server
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        os.environ["GITEA_TEST_PORCELAIN"] = " M task_file.py\n"
+
+        worktree = "/repo/branches/issue-275-dirty"
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(worktree_path=worktree)
+        msg = str(ctx.exception)
+        self.assertIn("active task workspace root", msg)
+        self.assertIn("inspected git root", msg)
+        self.assertIn("dirty files: task_file.py", msg)
+        self.assertIn("dirty scope:", msg)

tests/test_resolve_task_capability.py

@@ -141,6 +141,17 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertTrue(res["allowed_in_current_session"])
             self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
 
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_lock_issue_author_profile_allowed(self, _auth, _api):
+        # #275: lock_issue must be an exact resolver task for claim/lock gates.
+        with patch.dict(os.environ, self._env("author-profile")):
+            res = mcp_server.gitea_resolve_task_capability(task="lock_issue", remote="prgs")
+            self.assertEqual(res["requested_task"], "lock_issue")
+            self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
+            self.assertTrue(res["allowed_in_current_session"])
+            self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
+
     def test_resolve_unknown_task_fails_closed(self):
         with patch.dict(os.environ, self._env("author-profile")):
             with self.assertRaises(ValueError):