Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 76 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat: align claim/lock gates with branches-only worktrees (#275)

Browse Source 
Allow issue lock from base-equivalent branches/ worktrees instead of
requiring the literal master/main branch name. Runtime preflight and
mark_issue/lock_issue now inspect the declared active task workspace so
dirty control-checkout state does not block clean task worktrees.

Closes #275

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-06 14:47:56 -04:00
co-authored by Claude Opus 4.8
parent d6f4f936e3
commit 67e34baece
6 changed files with 292 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_issue_lock_worktree.py
+27 -2
View File
@@ -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
+59 -9
View File
@@ -2931,20 +2931,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)
@@ -2964,7 +2975,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)
@@ -2981,7 +2992,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)
@@ -3002,7 +3013,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,
@@ -3022,6 +3033,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:
@@ -3041,6 +3053,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:
@@ -3050,7 +3063,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, []))
@@ -3193,7 +3206,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]
@@ -3283,3 +3301,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
+11
View File
@@ -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):