Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 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:49:39 -04:00
co-authored by Claude Opus 4.8
parent d6f4f936e3
commit 5e64c96851
6 changed files with 294 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+111 -8
View File
@@ -171,6 +171,9 @@ _preflight_whoami_violation_files: list[str] = []
_preflight_capability_violation_files: list[str] = []
_preflight_reviewer_violation_files: list[str] = []
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
def _preflight_in_test_mode() -> bool:
return "pytest" in sys.modules or "unittest" in sys.modules
@@ -184,19 +187,47 @@ def _ensure_process_start_porcelain() -> str:
return _process_start_porcelain
def _get_workspace_porcelain() -> str:
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
"""Resolve the workspace root inspected by pre-flight guards."""
path = (worktree_path or "").strip()
if not path:
path = (os.environ.get(ACTIVE_WORKTREE_ENV) 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 _get_git_root(path: str) -> str | None:
try:
res = subprocess.run(
["git", "-C", path, "rev-parse", "--show-toplevel"],
capture_output=True,
text=True,
check=False,
)
except Exception:
return None
if res.returncode != 0:
return None
return (res.stdout or "").strip() or None
def _get_workspace_porcelain(worktree_path: str | None = None) -> str:
"""Return tracked-workspace porcelain for pre-flight attribution."""
if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
return " M __gitea_test_force_dirty__.py\n"
override = os.environ.get("GITEA_TEST_PORCELAIN")
if override is not None:
return override
workspace = _resolve_preflight_workspace_path(worktree_path)
try:
res = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True,
text=True,
cwd=PROJECT_ROOT,
cwd=workspace,
)
return res.stdout or ""
except Exception:
@@ -234,7 +265,35 @@ def _format_preflight_files(files: list[str]) -> str:
return ", ".join(files)
def assess_preflight_status() -> dict:
def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
workspace = _resolve_preflight_workspace_path(worktree_path)
inspected_root = _get_git_root(workspace)
control_root = os.path.realpath(PROJECT_ROOT)
active_root = os.path.realpath(inspected_root or workspace)
if active_root == control_root:
dirty_scope = "control checkout"
else:
dirty_scope = "active task workspace"
return {
"mcp_server_process_root": control_root,
"active_task_workspace_root": active_root,
"inspected_git_root": inspected_root,
"dirty_files": list(dirty_files),
"dirty_scope": dirty_scope,
}
def _format_preflight_workspace_details(details: dict) -> str:
return (
f"MCP server process root: {details.get('mcp_server_process_root')}; "
f"active task workspace root: {details.get('active_task_workspace_root')}; "
f"inspected git root: {details.get('inspected_git_root')}; "
f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; "
f"dirty scope: {details.get('dirty_scope')}"
)
def assess_preflight_status(worktree_path: str | None = None) -> dict:
"""Non-throwing pre-flight readiness for runtime-context alignment (#252)."""
reasons: list[str] = []
if not _preflight_whoami_called:
@@ -245,6 +304,25 @@ def assess_preflight_status() -> dict:
reasons.append(
"Task capability (gitea_resolve_task_capability) has not been resolved"
)
workspace_details = None
if worktree_path:
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path)))
workspace_details = _preflight_workspace_details(worktree_path, dirty_files)
if dirty_files:
reasons.append(
"Active task workspace has tracked file edits before mutation "
f"({_format_preflight_workspace_details(workspace_details)})"
)
return {
"preflight_ready": not reasons,
"preflight_block_reasons": reasons,
"preflight_whoami_verified": _preflight_whoami_called,
"preflight_capability_resolved": _preflight_capability_called,
"preflight_whoami_violation_files": [],
"preflight_capability_violation_files": [],
"preflight_reviewer_violation_files": [],
"preflight_workspace": workspace_details,
}
if _preflight_whoami_violation:
reasons.append(
"Workspace file edits occurred before gitea_whoami verification "
@@ -268,6 +346,7 @@ def assess_preflight_status() -> dict:
"preflight_whoami_violation_files": list(_preflight_whoami_violation_files),
"preflight_capability_violation_files": list(_preflight_capability_violation_files),
"preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files),
"preflight_workspace": _preflight_workspace_details(None, []),
}
@@ -308,7 +387,7 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
_preflight_resolved_role = resolved_role
def verify_preflight_purity(remote: str | None = None):
def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
"""Verify that identity and capability were verified prior to session edits."""
global _preflight_reviewer_violation_files
@@ -328,6 +407,17 @@ def verify_preflight_purity(remote: str | None = None):
"Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
)
if worktree_path:
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path)))
if dirty_files:
details = _preflight_workspace_details(worktree_path, dirty_files)
raise RuntimeError(
"Pre-flight order violation: Active task workspace has tracked "
"file edits before mutation (fail closed). "
f"{_format_preflight_workspace_details(details)}"
)
return
if _preflight_whoami_violation:
raise RuntimeError(
"Pre-flight order violation: Workspace file edits occurred before "
@@ -728,6 +818,8 @@ def gitea_create_issue(
Returns:
dict with 'number' of the created issue ('url' only with the reveal opt-in).
"""
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
"create_issue"
)
@@ -749,8 +841,6 @@ def gitea_create_issue(
if blocked:
return blocked
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
closed_issues = api_get_all(
@@ -818,14 +908,23 @@ def gitea_lock_issue(
f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
)
blocked = _profile_permission_block(
task_capability_map.required_permission("lock_issue"))
if blocked:
return blocked
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
worktree_path, PROJECT_ROOT
)
git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
verify_preflight_purity(remote, worktree_path=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 "",
base_equivalent=git_state.get("base_equivalent"),
inspected_git_root=git_state.get("inspected_git_root"),
base_branch=git_state.get("base_branch"),
)
if lock_assessment["block"]:
raise RuntimeError(
@@ -4191,6 +4290,7 @@ def _build_runtime_task_capabilities(
def gitea_get_runtime_context(
remote: str = "dadeschools",
host: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Read-only: explicit visibility into active profile, configuration model, and eligibility.
@@ -4278,7 +4378,7 @@ def gitea_get_runtime_context(
allowed, forbidden, config
)
preflight = assess_preflight_status()
preflight = assess_preflight_status(worktree_path)
if not preflight["preflight_ready"]:
safe_next_action = (
"Complete pre-flight verification before mutating: call gitea_whoami, then "
@@ -4302,6 +4402,7 @@ def gitea_get_runtime_context(
"safe_next_action": safe_next_action,
"preflight_ready": preflight["preflight_ready"],
"preflight_block_reasons": preflight["preflight_block_reasons"],
"preflight_workspace": preflight.get("preflight_workspace"),
"session_capabilities": session_capabilities,
}
@@ -4553,6 +4654,7 @@ def gitea_mark_issue(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Claim or release an issue via the status:in-progress label.
@@ -4565,6 +4667,7 @@ def gitea_mark_issue(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Active task worktree to inspect for pre-flight purity.
Returns:
dict with 'success' boolean and 'message'.
@@ -4576,7 +4679,7 @@ def gitea_mark_issue(
task_capability_map.required_permission("mark_issue"))
if blocked:
return blocked
verify_preflight_purity(remote)
verify_preflight_purity(remote, worktree_path=worktree_path)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
issue_lock_worktree.py
+81 -14
View File
@@ -14,7 +14,7 @@ import subprocess
from reviewer_worktree import parse_dirty_tracked_files
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
BASE_BRANCHES = frozenset({"master", "main"})
BASE_BRANCHES = frozenset({"master", "main", "dev"})
def resolve_author_worktree_path(
@@ -50,9 +50,28 @@ def read_worktree_git_state(worktree_path: str) -> dict:
text=True,
check=False,
)
root_res = subprocess.run(
["git", "-C", path, "rev-parse", "--show-toplevel"],
capture_output=True,
text=True,
check=False,
)
head_res = subprocess.run(
["git", "-C", path, "rev-parse", "HEAD"],
capture_output=True,
text=True,
check=False,
)
head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None
base_branch, base_sha = _find_matching_base_ref(path, head_sha)
return {
"current_branch": current_branch,
"porcelain_status": status_res.stdout or "",
"inspected_git_root": (root_res.stdout or "").strip() if root_res.returncode == 0 else None,
"head_sha": head_sha,
"base_branch": base_branch,
"base_sha": base_sha,
"base_equivalent": bool(head_sha and base_sha and head_sha == base_sha),
}
@@ -61,6 +80,9 @@ def assess_issue_lock_worktree(
worktree_path: str,
current_branch: str | None,
porcelain_status: str,
base_equivalent: bool | None = None,
inspected_git_root: str | None = None,
base_branch: str | None = None,
base_branches: frozenset[str] | None = None,
) -> dict:
"""Fail closed when lock preconditions are not met on the declared worktree."""
@@ -77,21 +99,39 @@ def assess_issue_lock_worktree(
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}'"
f"lock must precede implementation work in '{path}' "
f"(dirty files: {', '.join(dirty_files)})"
)
if base_equivalent is False:
reasons.append(
"issue lock worktree must be base-equivalent to one of "
f"{_base_list(bases)} before implementation work; inspected "
f"branch '{branch or '(detached)'}' at '{path}'"
)
elif base_equivalent is None:
if not branch:
reasons.append(
"current branch unknown (detached HEAD?); issue lock base-equivalence "
f"to {_base_list(bases)} could not be proven"
)
elif branch not in bases:
reasons.append(
"issue lock worktree base-equivalence could not be proven; "
f"branch '{branch}' is not {_base_list(bases)}"
)
proven = not reasons
return _assessment(proven, reasons, path, branch or None, dirty_files)
return _assessment(
proven,
reasons,
path,
branch or None,
dirty_files,
inspected_git_root=inspected_git_root,
base_branch=base_branch,
base_equivalent=base_equivalent,
)
def format_issue_lock_worktree_error(assessment: dict) -> str:
@@ -145,12 +185,39 @@ def _assessment(
worktree_path: str,
current_branch: str | None,
dirty_files: list[str],
*,
inspected_git_root: str | None = None,
base_branch: str | None = None,
base_equivalent: bool | None = None,
) -> dict:
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"worktree_path": worktree_path or None,
"inspected_git_root": inspected_git_root,
"current_branch": current_branch,
"dirty_files": dirty_files,
}
"base_branch": base_branch,
"base_equivalent": base_equivalent,
}
def _find_matching_base_ref(path: str, head_sha: str | None) -> tuple[str | None, str | None]:
"""Return the stable branch ref whose commit matches HEAD, if any."""
if not head_sha:
return None, None
candidates: list[str] = []
for branch in sorted(BASE_BRANCHES):
candidates.extend((f"origin/{branch}", branch))
for ref in candidates:
res = subprocess.run(
["git", "-C", path, "rev-parse", "--verify", ref],
capture_output=True,
text=True,
check=False,
)
sha = (res.stdout or "").strip()
if res.returncode == 0 and sha == head_sha:
return ref, sha
return None, None
task_capability_map.py
+5 -1
View File
@@ -28,6 +28,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.issue.comment",
"role": "author",
},
"lock_issue": {
"permission": "gitea.issue.comment",
"role": "author",
},
"set_issue_labels": {
"permission": "gitea.issue.comment",
"role": "author",
@@ -110,4 +114,4 @@ def required_role(task: str) -> str:
def tool_required_permission(tool_name: str) -> str:
"""Return the operation an issue-mutating tool must gate on."""
return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name])
return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name])
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):