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

fix: bind reviewer mutations to active branches worktree (Closes #503)

Browse Source 
Reviewer mutation tools now resolve workspace from worktree_path,
GITEA_ACTIVE_WORKTREE, or the session reviewer lease before preflight
and branches-only validation. Preflight rejects metadata-only worktree_path
bindings that would mislead runtime_context while mutations still inspect
the MCP process root.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-08 03:42:07 -04:00
co-authored by Claude Opus 4.8
parent 9ada4762c5
commit dad1dc8d51
3 changed files with 508 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+134 -5
View File
@@ -208,6 +208,71 @@ def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
)
def _reviewer_session_worktree() -> str | None:
session = reviewer_pr_lease.get_session_lease()
if not session:
return None
worktree = (session.get("worktree") or "").strip()
return worktree or None
def _resolve_reviewer_mutation_worktree_path(
worktree_path: str | None = None,
worktree: str | None = None,
) -> str:
"""Resolve the workspace inspected before reviewer mutations (#503)."""
return reviewer_mutation_workspace.resolve_reviewer_mutation_workspace(
worktree_path=worktree_path,
worktree=worktree,
process_project_root=PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
session_lease_worktree=_reviewer_session_worktree(),
)
def _assess_reviewer_mutation_workspace(
worktree_path: str | None = None,
worktree: str | None = None,
) -> dict:
git_state = issue_lock_worktree.read_worktree_git_state(
_resolve_reviewer_mutation_worktree_path(worktree_path, worktree)
)
return reviewer_mutation_workspace.assess_reviewer_mutation_workspace(
worktree_path=worktree_path,
worktree=worktree,
process_project_root=PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
session_lease_worktree=_reviewer_session_worktree(),
current_branch=git_state.get("current_branch"),
)
def _verify_reviewer_mutation_workspace(
remote: str | None = None,
worktree_path: str | None = None,
worktree: str | None = None,
) -> str:
"""Bind reviewer mutations to the active branches/ worktree (#503)."""
assessment = _assess_reviewer_mutation_workspace(worktree_path, worktree)
if assessment["block"]:
raise RuntimeError(
reviewer_mutation_workspace.format_stale_mcp_workspace_binding_error(
process_project_root=assessment["process_project_root"],
mutation_workspace=assessment["mutation_workspace"],
declared_worktree_path=assessment.get("declared_worktree_path"),
metadata_only=assessment.get("metadata_only", False),
)
+ (
f" Details: {'; '.join(assessment['reasons'])}"
if assessment.get("reasons")
else ""
)
)
resolved = assessment["mutation_workspace"]
verify_preflight_purity(remote, worktree_path=resolved)
return resolved
def _get_git_root(path: str) -> str | None:
try:
res = subprocess.run(
@@ -332,6 +397,22 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
"Active task workspace has tracked file edits before mutation "
f"({_format_preflight_workspace_details(workspace_details)})"
)
if _preflight_resolved_role == "reviewer":
binding = reviewer_mutation_workspace.assess_metadata_only_worktree_binding(
declared_worktree_path=worktree_path,
mutation_workspace=_resolve_reviewer_mutation_worktree_path(),
process_project_root=PROJECT_ROOT,
)
if binding.get("block"):
reasons.append(binding["reasons"][0])
reasons.append(
reviewer_mutation_workspace.format_stale_mcp_workspace_binding_error(
process_project_root=PROJECT_ROOT,
mutation_workspace=binding["mutation_workspace"],
declared_worktree_path=binding.get("declared_worktree_path"),
metadata_only=True,
)
)
return {
"preflight_ready": not reasons,
"preflight_block_reasons": reasons,
@@ -542,6 +623,7 @@ import author_mutation_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402
import issue_work_duplicate_gate # noqa: E402
import reviewer_pr_lease # noqa: E402
import reviewer_mutation_workspace # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402
@@ -2639,9 +2721,28 @@ def _evaluate_pr_review_submission(
*,
live: bool,
final_review_decision_ready: bool = False,
worktree_path: str | None = None,
) -> dict:
"""Shared gate chain for live submit and dry-run review tools."""
verify_preflight_purity(remote)
try:
_verify_reviewer_mutation_workspace(
remote, worktree_path=worktree_path
)
except RuntimeError as exc:
return {
"requested_action": (action or "").strip().lower(),
"performed": False,
"dry_run": not live,
"would_perform": False,
"authenticated_user": None,
"profile_name": get_profile()["profile_name"],
"pr_author": None,
"pr_number": pr_number,
"head_sha": None,
"expected_head_sha": expected_head_sha,
"remote": remote if remote in REMOTES else None,
"reasons": [str(exc)],
}
action = (action or "").strip().lower()
result = {
"requested_action": action,
@@ -2832,9 +2933,16 @@ def gitea_mark_final_review_decision(
remote: str = "dadeschools",
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Mark validation complete; the final review decision is ready to submit."""
action = (action or "").strip().lower()
try:
_verify_reviewer_mutation_workspace(
remote, worktree_path=worktree_path
)
except RuntimeError as exc:
return {"marked_ready": False, "reasons": [str(exc)]}
lock = _load_review_decision_lock()
if lock is None:
return {
@@ -3041,6 +3149,7 @@ def gitea_dry_run_pr_review(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Validate review submission mechanics without a live PR mutation."""
return _evaluate_pr_review_submission(
@@ -3053,6 +3162,7 @@ def gitea_dry_run_pr_review(
org=org,
repo=repo,
live=False,
worktree_path=worktree_path,
)
@@ -3068,6 +3178,7 @@ def gitea_submit_pr_review(
org: str | None = None,
repo: str | None = None,
final_review_decision_ready: bool = False,
worktree_path: str | None = None,
) -> dict:
"""Gated PR review mutation: comment findings, request changes, or approve.
@@ -3086,6 +3197,7 @@ def gitea_submit_pr_review(
repo=repo,
live=True,
final_review_decision_ready=final_review_decision_ready,
worktree_path=worktree_path,
)
@@ -4982,7 +5094,17 @@ def gitea_acquire_reviewer_pr_lease(
"permission_report": _permission_block_report("gitea.pr.comment"),
}
verify_preflight_purity(remote)
try:
resolved_worktree = _verify_reviewer_mutation_workspace(
remote, worktree=worktree
)
except RuntimeError as exc:
return {
"success": False,
"acquired": False,
"reasons": [str(exc)],
"permission_report": None,
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
profile = get_profile()
@@ -5000,7 +5122,7 @@ def gitea_acquire_reviewer_pr_lease(
session_id=sid,
repo=repo_label,
issue_number=issue_number,
worktree=worktree,
worktree=resolved_worktree,
candidate_head=candidate_head,
target_branch=target_branch,
target_branch_sha=target_branch_sha,
@@ -5032,7 +5154,7 @@ def gitea_acquire_reviewer_pr_lease(
"session_id": sid,
"reviewer_identity": identity,
"profile": profile.get("profile_name"),
"worktree": worktree,
"worktree": resolved_worktree,
"phase": "claimed",
"candidate_head": candidate_head,
"target_branch": target_branch,
@@ -5083,7 +5205,14 @@ def gitea_heartbeat_reviewer_pr_lease(
],
}
verify_preflight_purity(remote)
try:
_verify_reviewer_mutation_workspace(
remote,
worktree=worktree,
worktree_path=session.get("worktree"),
)
except RuntimeError as exc:
return {"success": False, "posted": False, "reasons": [str(exc)]}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
body = reviewer_pr_lease.format_lease_body(
reviewer_mutation_workspace.py
+157
View File
@@ -0,0 +1,157 @@
"""Reviewer mutation workspace binding (#503).
Reviewer mutation tools must validate the same branches/ worktree that
read-only preflight and runtime_context inspect — not only the MCP server
process root.
"""
from __future__ import annotations
import os
import author_mutation_worktree as amw
ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV
def resolve_reviewer_mutation_workspace(
*,
worktree_path: str | None,
worktree: str | None,
process_project_root: str,
active_worktree_env: str | None = None,
session_lease_worktree: str | None = None,
) -> str:
"""Resolve the workspace inspected before reviewer mutations."""
for candidate in (
worktree_path,
worktree,
active_worktree_env,
session_lease_worktree,
):
text = (candidate or "").strip()
if text:
return os.path.realpath(os.path.abspath(text))
return os.path.realpath(process_project_root)
def assess_metadata_only_worktree_binding(
*,
declared_worktree_path: str | None,
mutation_workspace: str,
process_project_root: str,
) -> dict:
"""Fail closed when declared worktree_path would not redirect mutations."""
declared = (declared_worktree_path or "").strip()
process_root = os.path.realpath(process_project_root)
mutation_root = os.path.realpath(mutation_workspace)
if not declared:
return {"block": False, "reasons": [], "metadata_only": False}
declared_root = os.path.realpath(os.path.abspath(declared))
if declared_root == mutation_root:
return {"block": False, "reasons": [], "metadata_only": False}
if declared_root != process_root and mutation_root == process_root:
return {
"block": True,
"metadata_only": True,
"reasons": [
"worktree_path is metadata-only for reviewer mutations: preflight "
f"inspected '{declared_root}' but mutation tools would still "
f"validate MCP server process root '{process_root}'"
],
"declared_worktree_path": declared_root,
"mutation_workspace": mutation_root,
"process_project_root": process_root,
}
return {"block": False, "reasons": [], "metadata_only": False}
def format_stale_mcp_workspace_binding_error(
*,
process_project_root: str,
mutation_workspace: str,
declared_worktree_path: str | None = None,
metadata_only: bool = False,
) -> str:
"""Canonical handoff when reviewer mutations bind to the wrong workspace."""
process_root = os.path.realpath(process_project_root)
mutation_root = os.path.realpath(mutation_workspace)
declared = (declared_worktree_path or "").strip()
declared_root = (
os.path.realpath(os.path.abspath(declared)) if declared else None
)
if metadata_only and declared_root:
lead = (
"Reviewer mutation workspace mismatch (#503): worktree_path affected "
f"read-only preflight only ('{declared_root}') while mutation tools "
f"still inspect MCP server process root '{process_root}'."
)
elif mutation_root == process_root:
lead = (
"Reviewer mutation workspace mismatch (#503): mutation tools are "
f"bound to the stable control checkout '{process_root}' instead of "
"an active branches/ review worktree."
)
else:
lead = (
"Reviewer mutation workspace mismatch (#503): mutation workspace "
f"'{mutation_root}' does not match the declared active task worktree."
)
fixes = (
"Fix: relaunch or reconnect the MCP server from the branches/ review "
"worktree, set GITEA_ACTIVE_WORKTREE to that path, pass worktree_path "
"to reviewer mutation tools, or acquire a reviewer PR lease from the "
"same branches/ worktree so mutations inherit the validated workspace."
)
return f"{lead} {fixes}"
def assess_reviewer_mutation_workspace(
*,
worktree_path: str | None,
worktree: str | None,
process_project_root: str,
active_worktree_env: str | None = None,
session_lease_worktree: str | None = None,
current_branch: str | None = None,
) -> dict:
"""Evaluate reviewer mutation workspace binding before preflight/mutation."""
mutation_workspace = resolve_reviewer_mutation_workspace(
worktree_path=worktree_path,
worktree=worktree,
process_project_root=process_project_root,
active_worktree_env=active_worktree_env,
session_lease_worktree=session_lease_worktree,
)
process_root = os.path.realpath(process_project_root)
metadata = assess_metadata_only_worktree_binding(
declared_worktree_path=worktree_path,
mutation_workspace=mutation_workspace,
process_project_root=process_root,
)
reasons = list(metadata.get("reasons") or [])
canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
branches = amw.assess_author_mutation_worktree(
workspace_path=mutation_workspace,
project_root=canonical_root,
current_branch=current_branch,
)
if branches["block"]:
reasons.extend(branches["reasons"])
block = bool(reasons)
return {
"block": block,
"reasons": reasons,
"mutation_workspace": mutation_workspace,
"process_project_root": process_root,
"canonical_repo_root": canonical_root,
"metadata_only": metadata.get("metadata_only", False),
"declared_worktree_path": metadata.get("declared_worktree_path"),
"under_branches": branches.get("under_branches", False),
}
tests/test_reviewer_mutation_workspace.py
+217
View File
@@ -0,0 +1,217 @@
"""Tests for reviewer mutation workspace binding (#503)."""
from __future__ import annotations
import os
import sys
import unittest
from pathlib import Path
from unittest import mock
from unittest.mock import MagicMock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import gitea_mcp_server as srv # noqa: E402
import reviewer_mutation_workspace as rmw # noqa: E402
import reviewer_pr_lease # noqa: E402
REPO_ROOT = Path(__file__).resolve().parent.parent
if REPO_ROOT.parent.name == "branches":
CONTROL_ROOT = str(REPO_ROOT.parent.parent)
else:
CONTROL_ROOT = str(REPO_ROOT)
BRANCHES_WORKTREE = f"{CONTROL_ROOT}/branches/review-pr503-submit"
MCP_PROCESS_ROOT = CONTROL_ROOT
class TestReviewerMutationWorkspaceModule(unittest.TestCase):
def test_resolve_prefers_explicit_worktree_path(self):
resolved = rmw.resolve_reviewer_mutation_workspace(
worktree_path=BRANCHES_WORKTREE,
worktree="/ignored",
process_project_root=MCP_PROCESS_ROOT,
active_worktree_env="/env/path",
session_lease_worktree="/lease/path",
)
self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE))
def test_resolve_falls_back_to_session_lease(self):
resolved = rmw.resolve_reviewer_mutation_workspace(
worktree_path=None,
worktree=None,
process_project_root=MCP_PROCESS_ROOT,
session_lease_worktree=BRANCHES_WORKTREE,
)
self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE))
def test_metadata_only_binding_detected(self):
assessment = rmw.assess_metadata_only_worktree_binding(
declared_worktree_path=BRANCHES_WORKTREE,
mutation_workspace=MCP_PROCESS_ROOT,
process_project_root=MCP_PROCESS_ROOT,
)
self.assertTrue(assessment["block"])
self.assertTrue(assessment["metadata_only"])
self.assertIn("metadata-only", assessment["reasons"][0])
def test_canonical_error_mentions_fix_paths(self):
msg = rmw.format_stale_mcp_workspace_binding_error(
process_project_root=MCP_PROCESS_ROOT,
mutation_workspace=MCP_PROCESS_ROOT,
declared_worktree_path=BRANCHES_WORKTREE,
metadata_only=True,
)
self.assertIn("GITEA_ACTIVE_WORKTREE", msg)
self.assertIn("worktree_path", msg)
self.assertIn("branches/", msg)
class TestReviewerMutationWorkspaceIntegration(unittest.TestCase):
def setUp(self):
reviewer_pr_lease.clear_session_lease()
self._saved = {
"whoami_called": srv._preflight_whoami_called,
"capability_called": srv._preflight_capability_called,
"resolved_role": srv._preflight_resolved_role,
"whoami_violation": srv._preflight_whoami_violation,
"whoami_files": list(srv._preflight_whoami_violation_files),
"capability_violation": srv._preflight_capability_violation,
"capability_files": list(srv._preflight_capability_violation_files),
"reviewer_files": list(srv._preflight_reviewer_violation_files),
"in_test": srv._preflight_in_test_mode,
}
srv._preflight_whoami_called = True
srv._preflight_capability_called = True
srv._preflight_resolved_role = "reviewer"
srv._preflight_whoami_violation = True
srv._preflight_whoami_violation_files = ["gitea_mcp_server.py"]
srv._preflight_capability_violation = False
srv._preflight_capability_violation_files = []
srv._preflight_reviewer_violation_files = []
srv._preflight_in_test_mode = lambda: False
self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False)
self._env_patch.start()
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
def tearDown(self):
reviewer_pr_lease.clear_session_lease()
srv._preflight_whoami_called = self._saved["whoami_called"]
srv._preflight_capability_called = self._saved["capability_called"]
srv._preflight_resolved_role = self._saved["resolved_role"]
srv._preflight_whoami_violation = self._saved["whoami_violation"]
srv._preflight_whoami_violation_files = self._saved["whoami_files"]
srv._preflight_capability_violation = self._saved["capability_violation"]
srv._preflight_capability_violation_files = self._saved["capability_files"]
srv._preflight_reviewer_violation_files = self._saved["reviewer_files"]
srv._preflight_in_test_mode = self._saved["in_test"]
self._env_patch.stop()
@mock.patch("subprocess.run")
@mock.patch("os.path.isdir", return_value=True)
@mock.patch("os.path.exists", return_value=True)
def test_reviewer_mutation_uses_declared_branches_worktree(
self, _exists, _isdir, mock_run
):
mock_run.return_value = MagicMock(
returncode=0,
stdout=f"{CONTROL_ROOT}/.git\n",
)
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
resolved = srv._verify_reviewer_mutation_workspace(
"prgs", worktree_path=BRANCHES_WORKTREE
)
self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE))
def test_control_checkout_blocks_reviewer_mutations(self):
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
with self.assertRaises(RuntimeError) as ctx:
srv._verify_reviewer_mutation_workspace("prgs")
self.assertIn("stable control checkout", str(ctx.exception))
self.assertIn("GITEA_ACTIVE_WORKTREE", str(ctx.exception))
def test_preflight_and_mutation_agree_on_workspace(self):
os.environ["GITEA_ACTIVE_WORKTREE"] = BRANCHES_WORKTREE
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
with mock.patch.object(
srv,
"_get_git_root",
return_value=BRANCHES_WORKTREE,
):
status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE)
resolved = srv._resolve_reviewer_mutation_worktree_path(
worktree_path=BRANCHES_WORKTREE
)
self.assertTrue(status["preflight_ready"])
self.assertEqual(
status["preflight_workspace"]["active_task_workspace_root"],
os.path.realpath(BRANCHES_WORKTREE),
)
self.assertEqual(resolved, os.path.realpath(BRANCHES_WORKTREE))
def test_metadata_only_worktree_path_blocks_preflight_ready(self):
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE)
self.assertFalse(status["preflight_ready"])
self.assertTrue(
any("metadata-only" in reason for reason in status["preflight_block_reasons"])
)
@mock.patch("subprocess.run")
@mock.patch("os.path.isdir", return_value=True)
@mock.patch("os.path.exists", return_value=True)
def test_dry_run_inherits_session_lease_worktree(
self, _exists, _isdir, mock_run
):
mock_run.return_value = MagicMock(
returncode=0,
stdout=f"{CONTROL_ROOT}/.git\n",
)
reviewer_pr_lease.record_session_lease({
"pr_number": 503,
"session_id": "sess-503",
"worktree": BRANCHES_WORKTREE,
})
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
with mock.patch.object(
srv,
"gitea_check_pr_eligibility",
return_value={
"eligible": False,
"reasons": ["stub"],
"authenticated_user": "reviewer",
"profile_name": "prgs-reviewer",
"pr_author": "author",
"head_sha": "abc123",
},
):
result = srv.gitea_dry_run_pr_review(
pr_number=503,
action="comment",
remote="prgs",
)
self.assertNotIn("stable control checkout", " ".join(result.get("reasons") or []))
self.assertNotIn("metadata-only", " ".join(result.get("reasons") or []))
@mock.patch("subprocess.run")
@mock.patch("os.path.isdir", return_value=True)
@mock.patch("os.path.exists", return_value=True)
def test_dry_run_fails_closed_from_control_checkout_without_binding(
self, _exists, _isdir, mock_run
):
mock_run.return_value = MagicMock(
returncode=0,
stdout=f"{CONTROL_ROOT}/.git\n",
)
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
result = srv.gitea_dry_run_pr_review(
pr_number=503,
action="comment",
remote="prgs",
)
reasons = " ".join(result.get("reasons") or [])
self.assertIn("stable control checkout", reasons)
if __name__ == "__main__":
unittest.main()