From 8a8cffbdf530c10094566b3920bf61748482a4ad Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 09:57:50 -0400 Subject: [PATCH 1/7] feat: gate reviewer mutations on canonical workflow load proof (Closes #389) Add gitea_load_review_workflow and in-process session proof required before gitea_submit_pr_review, gitea_mark_final_review_decision, and gitea_merge_pr. Capability and runtime context report workflow-load status; stale or missing proof fails closed with recovery handoff text that forbids approve/merge replay. --- gitea_mcp_server.py | 79 +++++++++++- review_workflow_load.py | 198 +++++++++++++++++++++++++++++ tests/test_llm_agent_sha.py | 2 + tests/test_mcp_server.py | 19 ++- tests/test_permission_reports.py | 2 + tests/test_review_workflow_load.py | 151 ++++++++++++++++++++++ 6 files changed, 445 insertions(+), 6 deletions(-) create mode 100644 review_workflow_load.py create mode 100644 tests/test_review_workflow_load.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 9913535..02635c7 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -474,6 +474,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 review_workflow_load # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 import issue_claim_heartbeat # noqa: E402 @@ -1636,6 +1637,7 @@ def init_review_decision_lock(remote: str | None, task: str | None): """Seed read-only-until-ready state for reviewer PR review tasks.""" if task != "review_pr": return + review_workflow_load.clear_review_workflow_load() profile = get_profile() profile_name = (profile.get("profile_name") or "").strip() session_lock = ( @@ -1661,6 +1663,11 @@ def init_review_decision_lock(remote: str | None, task: str | None): }) +def _review_workflow_load_gate_reasons() -> list[str]: + """Fail closed when canonical review workflow was not loaded (#389).""" + return review_workflow_load.review_workflow_load_blockers(PROJECT_ROOT) + + def check_review_decision_gate( pr_number: int, action: str, @@ -1671,7 +1678,10 @@ def check_review_decision_gate( repo: str | None = None, ) -> list[str]: """Fail closed unless validation completed and the final decision is ready.""" - reasons = [] + reasons = list(_review_workflow_load_gate_reasons()) + if reasons: + reasons.extend(review_workflow_load.recovery_handoff_without_replay()) + return reasons lock = _load_review_decision_lock() if lock is None: reasons.append( @@ -2027,6 +2037,7 @@ def _evaluate_pr_review_submission( """Shared gate chain for live submit and dry-run review tools.""" verify_preflight_purity(remote) action = (action or "").strip().lower() + workflow_blockers = _review_workflow_load_gate_reasons() if live else [] result = { "requested_action": action, "performed": False, @@ -2042,6 +2053,10 @@ def _evaluate_pr_review_submission( "reasons": [], } reasons = result["reasons"] + if workflow_blockers: + reasons.extend(workflow_blockers) + reasons.extend(review_workflow_load.recovery_handoff_without_replay()) + return result if action not in _REVIEW_ACTIONS: reasons.append( @@ -2226,6 +2241,13 @@ def gitea_mark_final_review_decision( } org = resolved_org repo = resolved_repo + workflow_blockers = _review_workflow_load_gate_reasons() + if workflow_blockers: + return { + "marked_ready": False, + "reasons": workflow_blockers + ( + review_workflow_load.recovery_handoff_without_replay()), + } hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready") if hard_stop: return {"marked_ready": False, "reasons": hard_stop} @@ -2809,6 +2831,7 @@ def gitea_merge_pr( available. Never secrets. """ verify_preflight_purity(remote) + workflow_blockers = _review_workflow_load_gate_reasons() do = (do or "").strip().lower() result = { "performed": False, @@ -2826,6 +2849,10 @@ def gitea_merge_pr( "reasons": [], } reasons = result["reasons"] + if workflow_blockers: + reasons.extend(workflow_blockers) + reasons.extend(review_workflow_load.recovery_handoff_without_replay()) + return result # Gate 1 — valid merge method (no API call on a bad method). if do not in _MERGE_METHODS: @@ -4175,6 +4202,8 @@ _PROJECT_SKILLS = { "steps": [ "Resolve task first: gitea_resolve_task_capability(task='review_pr') " "to confirm reviewer namespace and avoid author-profile blocks.", + "Load canonical workflow proof with gitea_load_review_workflow " + "before any review/merge mutation (#389).", "Verify reviewer identity with gitea_whoami; the PR author " "must be a different user.", "Reconcile live queue state FIRST (do not trust prior handoffs): " @@ -4978,6 +5007,8 @@ def gitea_get_runtime_context( "preflight_workspace": preflight.get("preflight_workspace"), "session_capabilities": session_capabilities, "shell_health": native_mcp_preference.shell_health_status(), + "workflow_load_proof": review_workflow_load.workflow_load_status( + PROJECT_ROOT), } if reveal and h: @@ -4986,6 +5017,42 @@ def gitea_get_runtime_context( return result +@mcp.tool() +def gitea_load_review_workflow( + prompt_text: str | None = None, +) -> dict: + """Load and record canonical review-merge workflow proof for this session (#389). + + Read-only with respect to Gitea API; records in-process workflow source/hash + proof required before reviewer review or merge mutations. + """ + try: + recorded = review_workflow_load.record_review_workflow_load( + PROJECT_ROOT, prompt_text=prompt_text) + except OSError as exc: + return { + "success": False, + "loaded": False, + "reasons": [str(exc)], + "recovery_handoff": review_workflow_load.recovery_handoff_without_replay(), + } + return { + "success": True, + "loaded": True, + "workflow_source": recorded["workflow_source"], + "task_mode": recorded["task_mode"], + "workflow_hash": recorded["workflow_hash"], + "workflow_version": recorded["workflow_version"], + "final_report_schema_path": recorded["final_report_schema_path"], + "final_report_schema_hash": recorded["final_report_schema_hash"], + "prompt_conflicts_with_workflow": recorded[ + "prompt_conflicts_with_workflow"], + "prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [], + "workflow_load_proof_present": True, + "reasons": [], + } + + @mcp.tool() def gitea_list_profiles() -> dict: """Read-only: list all Gitea MCP profiles with redacted metadata. @@ -6041,6 +6108,16 @@ def gitea_resolve_task_capability( } if reason_msg: result["reason"] = reason_msg + if task in ("review_pr", "merge_pr"): + result["workflow_load_proof"] = review_workflow_load.workflow_load_status( + PROJECT_ROOT) + if not result["workflow_load_proof"].get("workflow_load_valid"): + guidance = ( + "Call gitea_load_review_workflow before any reviewer review " + "or merge mutation." + ) + if guidance not in task_role_guidance: + task_role_guidance.append(guidance) role_session_router.sync_route_from_capability(result) was_terminal = capability_stop_terminal.is_active() terminal = capability_stop_terminal.sync_from_capability_result(result) diff --git a/review_workflow_load.py b/review_workflow_load.py new file mode 100644 index 0000000..070c29a --- /dev/null +++ b/review_workflow_load.py @@ -0,0 +1,198 @@ +"""Canonical review-merge workflow load proof for reviewer mutations (#389).""" + +from __future__ import annotations + +import hashlib +import os +import re +from pathlib import Path + +WORKFLOW_REL_PATH = ( + "skills/llm-project-workflow/workflows/review-merge-pr.md" +) +SCHEMA_REL_PATH = ( + "skills/llm-project-workflow/schemas/review-merge-final-report.md" +) +TASK_MODE = "review-merge-pr" +LOAD_TOOL_NAME = "gitea_load_review_workflow" + +_REVIEW_WORKFLOW_LOAD: dict | None = None + + +def compute_content_hash(text: str) -> str: + """Short deterministic hash for workflow/schema version proof.""" + return hashlib.sha256((text or "").encode("utf-8")).hexdigest()[:12] + + +def _read_text(path: Path) -> str: + return path.read_text(encoding="utf-8") + + +def _canonical_paths(project_root: str) -> tuple[Path, Path]: + root = Path(project_root) + workflow = root / WORKFLOW_REL_PATH + schema = root / SCHEMA_REL_PATH + if not workflow.is_file(): + raise FileNotFoundError(f"canonical workflow missing: {workflow}") + if not schema.is_file(): + raise FileNotFoundError(f"final report schema missing: {schema}") + return workflow, schema + + +def build_canonical_workflow_metadata( + project_root: str, + *, + prompt_text: str | None = None, +) -> dict: + """Load workflow + schema from disk and compute proof metadata.""" + workflow_path, schema_path = _canonical_paths(project_root) + workflow_text = _read_text(workflow_path) + schema_text = _read_text(schema_path) + workflow_hash = compute_content_hash(workflow_text) + schema_hash = compute_content_hash(schema_text) + conflict, conflict_reasons = assess_prompt_conflict(prompt_text) + return { + "workflow_source": WORKFLOW_REL_PATH, + "workflow_path": str(workflow_path), + "task_mode": TASK_MODE, + "workflow_hash": workflow_hash, + "workflow_version": workflow_hash, + "final_report_schema_path": SCHEMA_REL_PATH, + "final_report_schema_hash": schema_hash, + "prompt_conflicts_with_workflow": conflict, + "prompt_conflict_reasons": conflict_reasons, + "load_tool": LOAD_TOOL_NAME, + } + + +def assess_prompt_conflict(prompt_text: str | None) -> tuple[bool, list[str]]: + """Detect obvious task-mode conflicts between prompt and review workflow.""" + if not (prompt_text or "").strip(): + return False, [] + text = prompt_text.lower() + reasons: list[str] = [] + conflicting = ( + (r"\bwork[- ]issue\b", "work-issue author mode"), + (r"\bcreate[- ]issue\b", "create-issue mode"), + (r"\bauthor/coder\b", "author/coder mode"), + (r"\breconcile[- ]landed\b", "reconcile-landed mode"), + ) + for pattern, label in conflicting: + if re.search(pattern, text): + reasons.append( + f"active prompt appears to request {label} while loading " + f"{TASK_MODE} workflow" + ) + return bool(reasons), reasons + + +def record_review_workflow_load( + project_root: str, + *, + prompt_text: str | None = None, +) -> dict: + """Record in-process workflow load proof for the current MCP session.""" + global _REVIEW_WORKFLOW_LOAD + meta = build_canonical_workflow_metadata( + project_root, prompt_text=prompt_text) + _REVIEW_WORKFLOW_LOAD = { + **meta, + "session_pid": os.getpid(), + "loaded": True, + } + return dict(_REVIEW_WORKFLOW_LOAD) + + +def clear_review_workflow_load() -> None: + """Test helper and review_pr session reset.""" + global _REVIEW_WORKFLOW_LOAD + _REVIEW_WORKFLOW_LOAD = None + + +def workflow_load_status(project_root: str | None = None) -> dict: + """Non-throwing status for capability/runtime reports.""" + load = _REVIEW_WORKFLOW_LOAD + if load is None: + return { + "workflow_load_proof_present": False, + "workflow_load_valid": False, + "workflow_source": None, + "workflow_hash": None, + "final_report_schema_path": SCHEMA_REL_PATH, + "reasons": [ + f"{LOAD_TOOL_NAME} has not been called in this session " + "(fail closed for reviewer mutations)" + ], + } + reasons = _session_validation_reasons(load, project_root) + return { + "workflow_load_proof_present": True, + "workflow_load_valid": not reasons, + "workflow_source": load.get("workflow_source"), + "workflow_hash": load.get("workflow_hash"), + "task_mode": load.get("task_mode"), + "final_report_schema_path": load.get("final_report_schema_path"), + "final_report_schema_hash": load.get("final_report_schema_hash"), + "prompt_conflicts_with_workflow": load.get( + "prompt_conflicts_with_workflow"), + "session_pid": load.get("session_pid"), + "reasons": reasons, + } + + +def _session_validation_reasons( + load: dict, + project_root: str | None, +) -> list[str]: + reasons: list[str] = [] + if load.get("session_pid") != os.getpid(): + reasons.append( + "workflow load proof was recorded in a different process " + "(fail closed)" + ) + return reasons + if load.get("prompt_conflicts_with_workflow"): + reasons.extend(load.get("prompt_conflict_reasons") or [ + "active prompt conflicts with loaded review-merge workflow" + ]) + if project_root: + try: + current = build_canonical_workflow_metadata(project_root) + except OSError as exc: + reasons.append(f"cannot re-verify workflow hash: {exc}") + return reasons + if current["workflow_hash"] != load.get("workflow_hash"): + reasons.append( + "stored workflow hash is stale; reload via " + f"{LOAD_TOOL_NAME} (fail closed)" + ) + if current["final_report_schema_hash"] != load.get( + "final_report_schema_hash"): + reasons.append( + "stored final-report schema hash is stale; reload via " + f"{LOAD_TOOL_NAME} (fail closed)" + ) + return reasons + + +def review_workflow_load_blockers( + project_root: str | None = None, +) -> list[str]: + """Reasons reviewer mutations must fail closed.""" + status = workflow_load_status(project_root) + if not status.get("workflow_load_proof_present"): + return list(status.get("reasons") or []) + if not status.get("workflow_load_valid"): + return list(status.get("reasons") or []) + return [] + + +def recovery_handoff_without_replay() -> list[str]: + """Safe next-step lines that must not include approve/merge replay.""" + return [ + "Reload the canonical workflow via gitea_load_review_workflow, then " + "rerun the full review-merge workflow from inventory.", + "Do not call gitea_submit_pr_review, gitea_mark_final_review_decision, " + "or gitea_merge_pr until workflow-load proof is present.", + "Do not include approve/merge replay commands in the recovery handoff.", + ] \ No newline at end of file diff --git a/tests/test_llm_agent_sha.py b/tests/test_llm_agent_sha.py index 1a6ecff..e751567 100644 --- a/tests/test_llm_agent_sha.py +++ b/tests/test_llm_agent_sha.py @@ -22,6 +22,7 @@ from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) +import mcp_server # noqa: E402 from mcp_server import ( # noqa: E402 gitea_check_pr_eligibility, gitea_merge_pr, @@ -130,6 +131,7 @@ class TestShaCannotBypassSelfReview(unittest.TestCase): ] from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision init_review_decision_lock("prgs", "review_pr") + mcp_server.gitea_load_review_workflow() gitea_mark_final_review_decision(9, "approve", remote="prgs") env = self._env(SHA_WOULD_BE_REVIEWER, "reviewer") with patch.dict(os.environ, env, clear=True): diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 99d7d73..1928e2a 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -55,6 +55,12 @@ _NO_BLOCKER_FEEDBACK = { } +def _init_reviewer_session(remote="prgs"): + """Seed review decision lock and required workflow-load proof (#389).""" + init_review_decision_lock(remote, "review_pr") + mcp_server.gitea_load_review_workflow() + + def _mark_request_changes_ready(pr_number=8, **kwargs): """Mark a request_changes decision ready with the #332 duplicate- suppression feedback fetch stubbed to 'no existing blocker'.""" @@ -512,6 +518,9 @@ class TestViewPR(unittest.TestCase): class TestMergePR(unittest.TestCase): """Gated merge workflow (#16). gitea_merge_pr is the only merge path.""" + def setUp(self): + mcp_server.gitea_load_review_workflow() + def _pr(self, author, state="open", sha="abc123", mergeable=True): return { "user": {"login": author}, @@ -984,8 +993,7 @@ class TestReviewPR(unittest.TestCase): {"login": "jcwalker3"}, # /api/v1/user (submit eligibility) {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1 ] - from mcp_server import init_review_decision_lock - init_review_decision_lock("prgs", "review_pr") + _init_reviewer_session("prgs") gitea_mark_final_review_decision(1, "approve", remote="prgs") result = gitea_review_pr( pr_number=1, @@ -1643,7 +1651,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase): } def setUp(self): - init_review_decision_lock("prgs", "review_pr") + _init_reviewer_session("prgs") def _env(self): return patch.dict(os.environ, { @@ -1738,7 +1746,7 @@ class TestSubmitPrReview(unittest.TestCase): """Gated review-mutation tool (#15).""" def setUp(self): - init_review_decision_lock("prgs", "review_pr") + _init_reviewer_session("prgs") gitea_mark_final_review_decision(8, "approve", remote="prgs") def _pr(self, author, state="open", sha="abc123", mergeable=True): @@ -2131,7 +2139,7 @@ class TestSubmitPrReview(unittest.TestCase): os.remove(spoof_path) def test_mark_final_decision_rejects_remote_mismatch(self): - init_review_decision_lock("prgs", "review_pr") + _init_reviewer_session("prgs") r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools") self.assertFalse(r["marked_ready"]) self.assertTrue(any("does not match locked remote" in x for x in r["reasons"])) @@ -2213,6 +2221,7 @@ if __name__ == "__main__": class TestTrackerHygieneCleanup(unittest.TestCase): def setUp(self): + mcp_server.gitea_load_review_workflow() self.mock_api = patch("mcp_server.api_request").start() self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() patch("gitea_audit.audit_enabled", return_value=True).start() diff --git a/tests/test_permission_reports.py b/tests/test_permission_reports.py index 5483ee1..b317a00 100644 --- a/tests/test_permission_reports.py +++ b/tests/test_permission_reports.py @@ -230,6 +230,7 @@ class TestEligibilityDenialReport(PermissionReportBase): return PR_PAYLOAD mock_api.side_effect = fake_api mcp_server.init_review_decision_lock("prgs", "review_pr") + mcp_server.gitea_load_review_workflow() mcp_server.gitea_mark_final_review_decision(42, "approve", remote="prgs") with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_submit_pr_review( @@ -272,6 +273,7 @@ class TestReviewCommentPathUsesCanonicalOp(PermissionReportBase): return PR_PAYLOAD mock_api.side_effect = fake_api mcp_server.init_review_decision_lock("prgs", "review_pr") + mcp_server.gitea_load_review_workflow() mcp_server.gitea_mark_final_review_decision(42, "comment", remote="prgs") with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_submit_pr_review( diff --git a/tests/test_review_workflow_load.py b/tests/test_review_workflow_load.py new file mode 100644 index 0000000..b611422 --- /dev/null +++ b/tests/test_review_workflow_load.py @@ -0,0 +1,151 @@ +"""Tests for canonical review workflow load proof (#389).""" + +import os +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import review_workflow_load +import mcp_server + + +class TestReviewWorkflowLoadModule(unittest.TestCase): + def setUp(self): + review_workflow_load.clear_review_workflow_load() + mcp_server._save_review_decision_lock(None) + + def test_load_records_hash_and_schema(self): + root = str(__import__("pathlib").Path(__file__).resolve().parent.parent) + recorded = review_workflow_load.record_review_workflow_load(root) + self.assertEqual( + recorded["workflow_source"], + review_workflow_load.WORKFLOW_REL_PATH, + ) + self.assertEqual(recorded["task_mode"], "review-merge-pr") + self.assertRegex(recorded["workflow_hash"], r"^[0-9a-f]{12}$") + self.assertEqual( + recorded["final_report_schema_path"], + review_workflow_load.SCHEMA_REL_PATH, + ) + status = review_workflow_load.workflow_load_status(root) + self.assertTrue(status["workflow_load_proof_present"]) + self.assertTrue(status["workflow_load_valid"]) + + def test_stale_session_pid_blocks(self): + root = str(__import__("pathlib").Path(__file__).resolve().parent.parent) + review_workflow_load.record_review_workflow_load(root) + review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0 + blockers = review_workflow_load.review_workflow_load_blockers(root) + self.assertTrue(any("different process" in b for b in blockers)) + + def test_prompt_conflict_detected(self): + conflict, reasons = review_workflow_load.assess_prompt_conflict( + "Run work-issue author implementation only") + self.assertTrue(conflict) + self.assertTrue(reasons) + + +class TestReviewWorkflowLoadGates(unittest.TestCase): + def setUp(self): + review_workflow_load.clear_review_workflow_load() + mcp_server.init_review_decision_lock("prgs", "review_pr") + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", "reviewer") + + def _load_workflow(self): + return mcp_server.gitea_load_review_workflow() + + def test_mcp_helper_returns_required_fields(self): + res = self._load_workflow() + self.assertTrue(res["success"]) + self.assertTrue(res["loaded"]) + self.assertIn("workflow_source", res) + self.assertIn("workflow_hash", res) + self.assertIn("final_report_schema_path", res) + self.assertIn("final_report_schema_hash", res) + + def test_mark_final_blocked_without_load(self): + res = mcp_server.gitea_mark_final_review_decision( + 42, "approve", remote="prgs") + self.assertFalse(res["marked_ready"]) + self.assertTrue(any( + "gitea_load_review_workflow" in r for r in res["reasons"])) + self.assertTrue(any( + "approve/merge replay" in r.lower() or "Do not call" in r + for r in res["reasons"])) + + def test_submit_review_blocked_without_load(self): + with patch("mcp_server.gitea_check_pr_eligibility") as elig: + elig.return_value = { + "eligible": True, + "authenticated_user": "rev", + "profile_name": "prgs-reviewer", + "pr_author": "author", + "head_sha": "abc123", + "reasons": [], + } + res = mcp_server.gitea_submit_pr_review( + 42, + "approve", + remote="prgs", + final_review_decision_ready=True, + ) + self.assertFalse(res["performed"]) + self.assertTrue(any( + "gitea_load_review_workflow" in r for r in res["reasons"])) + + def test_merge_blocked_without_load(self): + res = mcp_server.gitea_merge_pr( + 42, + confirmation="MERGE PR 42", + remote="prgs", + ) + self.assertFalse(res["performed"]) + self.assertTrue(any( + "gitea_load_review_workflow" in r for r in res["reasons"])) + + def test_resolve_capability_reports_missing_load(self): + with patch.object(mcp_server, "_ensure_matching_profile"): + with patch.object( + mcp_server.gitea_config, "is_runtime_switching_enabled", + return_value=False): + with patch.object( + mcp_server, "_authenticated_username", + return_value="rev"): + res = mcp_server.gitea_resolve_task_capability( + "review_pr", remote="prgs") + proof = res.get("workflow_load_proof") or {} + self.assertFalse(proof.get("workflow_load_valid")) + self.assertTrue(any( + "gitea_load_review_workflow" in g + for g in res.get("task_role_guidance") or [])) + + def test_init_review_lock_clears_prior_load(self): + self._load_workflow() + mcp_server.init_review_decision_lock("prgs", "review_pr") + blockers = review_workflow_load.review_workflow_load_blockers( + str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + self.assertTrue(blockers) + + def test_dry_run_allowed_without_load(self): + with patch("mcp_server.gitea_check_pr_eligibility") as elig: + elig.return_value = { + "eligible": True, + "authenticated_user": "rev", + "profile_name": "prgs-reviewer", + "pr_author": "author", + "head_sha": "abc123", + "reasons": [], + } + res = mcp_server.gitea_dry_run_pr_review( + 42, "approve", remote="prgs") + self.assertNotIn( + "gitea_load_review_workflow", + " ".join(res.get("reasons") or []), + ) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 5512ba7710c3c11f2f3587f80aec284b3d2cbede Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 13:24:47 -0400 Subject: [PATCH 2/7] test: seed workflow load proof in reviewer mutation paths (Closes #389) --- tests/test_audit.py | 4 ++++ tests/test_permission_reports.py | 3 +++ tests/test_pr_queue_inventory.py | 2 ++ tests/test_terminal_review_hard_stop.py | 5 +++++ 4 files changed, 14 insertions(+) diff --git a/tests/test_audit.py b/tests/test_audit.py index adb9242..768f0d1 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -157,6 +157,7 @@ class _AuditWiringBase(unittest.TestCase): def tearDown(self): mcp_server._IDENTITY_CACHE.clear() + mcp_server.review_workflow_load.clear_review_workflow_load() self._dir.cleanup() def _env(self, **extra): @@ -304,6 +305,7 @@ class TestGatedToolAudit(_AuditWiringBase): env = self._env(GITEA_PROFILE_NAME="gitea-merger", GITEA_ALLOWED_OPERATIONS="read,merge") with patch.dict(os.environ, env, clear=True): + mcp_server.gitea_load_review_workflow() r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8", expected_head_sha="abc123", remote="prgs") self.assertTrue(r["performed"]) @@ -323,6 +325,7 @@ class TestGatedToolAudit(_AuditWiringBase): env = self._env(GITEA_PROFILE_NAME="gitea-merger", GITEA_ALLOWED_OPERATIONS="read,merge") with patch.dict(os.environ, env, clear=True): + mcp_server.gitea_load_review_workflow() r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8", remote="prgs") self.assertFalse(r["performed"]) recs = self._records() @@ -345,6 +348,7 @@ class TestGatedToolAudit(_AuditWiringBase): with patch.dict(os.environ, env, clear=True): from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision init_review_decision_lock("prgs", "review_pr") + mcp_server.gitea_load_review_workflow() gitea_mark_final_review_decision(8, "approve", remote="prgs") r = gitea_submit_pr_review(pr_number=8, action="approve", body="LGTM", remote="prgs", diff --git a/tests/test_permission_reports.py b/tests/test_permission_reports.py index b317a00..c560b7f 100644 --- a/tests/test_permission_reports.py +++ b/tests/test_permission_reports.py @@ -99,6 +99,7 @@ class PermissionReportBase(unittest.TestCase): def tearDown(self): self._remotes.stop() mcp_server._IDENTITY_CACHE.clear() + mcp_server.review_workflow_load.clear_review_workflow_load() gitea_config._active_profile_override = None self._dir.cleanup() @@ -249,6 +250,8 @@ class TestEligibilityDenialReport(PermissionReportBase): return {"login": "author-user"} return PR_PAYLOAD mock_api.side_effect = fake_api + mcp_server.init_review_decision_lock("prgs", "review_pr") + mcp_server.gitea_load_review_workflow() with patch.dict(os.environ, self._env("author-profile")): res = mcp_server.gitea_merge_pr( pr_number=42, confirmation="MERGE PR 42", diff --git a/tests/test_pr_queue_inventory.py b/tests/test_pr_queue_inventory.py index aeb8bc4..49d2b4d 100644 --- a/tests/test_pr_queue_inventory.py +++ b/tests/test_pr_queue_inventory.py @@ -11,6 +11,7 @@ from mcp_server import ( gitea_view_pr, gitea_review_pr, gitea_check_pr_eligibility, + gitea_load_review_workflow, ) import gitea_config @@ -157,6 +158,7 @@ class TestPRQueueInventory(unittest.TestCase): from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision init_review_decision_lock("prgs", "review_pr") + gitea_load_review_workflow() gitea_mark_final_review_decision(1, "approve", remote="prgs") result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs", final_review_decision_ready=True) self.assertTrue(result["success"]) diff --git a/tests/test_terminal_review_hard_stop.py b/tests/test_terminal_review_hard_stop.py index bd5109b..3f80fad 100644 --- a/tests/test_terminal_review_hard_stop.py +++ b/tests/test_terminal_review_hard_stop.py @@ -35,6 +35,7 @@ def _lock(mutations=None, correction=False): def _seed(mutations=None, correction=False): mcp_server._save_review_decision_lock(_lock(mutations, correction)) + mcp_server.gitea_load_review_workflow() APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1, @@ -46,6 +47,7 @@ RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2, class TestTerminalHardStopReasons(unittest.TestCase): def tearDown(self): mcp_server._save_review_decision_lock(None) + mcp_server.review_workflow_load.clear_review_workflow_load() def test_no_lock_no_reasons(self): mcp_server._save_review_decision_lock(None) @@ -91,6 +93,7 @@ class TestTerminalHardStopReasons(unittest.TestCase): class TestMergeHardStopWiring(unittest.TestCase): def tearDown(self): mcp_server._save_review_decision_lock(None) + mcp_server.review_workflow_load.clear_review_workflow_load() def test_merge_blocked_after_request_changes(self): _seed([RC_A]) @@ -112,6 +115,7 @@ class TestMergeHardStopWiring(unittest.TestCase): class TestMarkFinalHardStopWiring(unittest.TestCase): def tearDown(self): mcp_server._save_review_decision_lock(None) + mcp_server.review_workflow_load.clear_review_workflow_load() def test_mark_ready_blocked_after_terminal_mutation(self): _seed([RC_A]) @@ -134,6 +138,7 @@ def _feedback(blocking, stale=False, success=True): class TestDuplicateRequestChangesSuppression(unittest.TestCase): def tearDown(self): mcp_server._save_review_decision_lock(None) + mcp_server.review_workflow_load.clear_review_workflow_load() def test_duplicate_request_changes_blocked_at_same_head(self): _seed() From ddaf380db2c674f4dadaeaed1fadbe4488bc31f9 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 12:03:08 -0400 Subject: [PATCH 3/7] feat: gate reconciliation audit mode from unauthorized cleanup (Closes #419) Add audit vs cleanup phase tracking so reconciliation audits stay read-only unless cleanup is explicitly authorized with delete capability proof, safety proof, and before/after snapshots. Block gitea_delete_branch and merged-cleanup execution during audit phase, validate final reports for false no-mutations claims, and document the boundary in the reconcile-landed-pr workflow. --- audit_reconciliation_mode.py | 344 ++++++++++++++++++ final_report_validator.py | 17 + gitea_mcp_server.py | 75 ++++ review_proofs.py | 9 + .../workflows/reconcile-landed-pr.md | 34 ++ task_capability_map.py | 4 + tests/test_audit_reconciliation_mode.py | 291 +++++++++++++++ tests/test_llm_workflow_split.py | 8 + 8 files changed, 782 insertions(+) create mode 100644 audit_reconciliation_mode.py create mode 100644 tests/test_audit_reconciliation_mode.py diff --git a/audit_reconciliation_mode.py b/audit_reconciliation_mode.py new file mode 100644 index 0000000..52d0244 --- /dev/null +++ b/audit_reconciliation_mode.py @@ -0,0 +1,344 @@ +"""Audit vs cleanup phase gates for reconciliation workflows (#419). + +Audit/reconciliation tasks are read-only unless a separate cleanup phase is +explicitly authorized with exact capability proof, safety proof, and +before/after snapshots. Cleanup mutations must be classified in final reports; +audit reports must not claim ``no mutations`` when cleanup occurred. +""" + +from __future__ import annotations + +import re +from typing import Any + +RECONCILE_WORKFLOW_PATH = "workflows/reconcile-landed-pr.md" + +PHASE_AUDIT = "audit" +PHASE_CLEANUP = "cleanup" + +# Tasks that enter audit phase on capability resolution (read-only default). +AUDIT_PHASE_TASKS = frozenset({ + "reconcile-landed-pr", + "reconcile_landed_pr", + "reconcile_issue_claims", + "reconcile_merged_cleanups", +}) + +# Mutation tasks forbidden during audit phase (fail closed). +AUDIT_FORBIDDEN_TASKS = frozenset({ + "delete_branch", + "create_branch", + "push_branch", + "create_pr", + "commit_files", + "gitea_commit_files", + "mark_issue", + "lock_issue", + "claim_issue", + "close_pr", + "close_issue", + "create_issue", + "merge_pr", + "review_pr", + "submit_pr_review", + "comment_pr", + "comment_issue", + "set_issue_labels", +}) + +# Shell/git commands audit phase must not run. +AUDIT_FORBIDDEN_COMMAND_RE = re.compile( + r"(?:^|\s)(?:git\s+(?:push|branch\s+-D|worktree\s+remove)|" + r"gitea_delete_branch|delete_remote_branch)", + re.IGNORECASE, +) + +_NO_MUTATIONS_RE = re.compile( + r"(?:no\s+mutations|mutations\s*:\s*none|no\s+unsafe\s+mutation)", + re.IGNORECASE, +) +_CLEANUP_OCCURRED_RE = re.compile( + r"(?:delete_remote_branch|remove_local_worktree|git\s+branch\s+-D|" + r"git\s+worktree\s+remove|remote branch.*deleted|worktree.*removed|" + r"cleanup\s+phase\s*:\s*(?!none\b)\S)", + re.IGNORECASE, +) +_EXTERNAL_STATE_RE = re.compile( + r"^\s*[-*]?\s*external[- ]state mutations\s*:", + re.IGNORECASE | re.MULTILINE, +) +_GIT_REF_RE = re.compile( + r"^\s*[-*]?\s*git ref mutations\s*:", + re.IGNORECASE | re.MULTILINE, +) +_CLEANUP_MUTATIONS_RE = re.compile( + r"^\s*[-*]?\s*cleanup mutations\s*:", + re.IGNORECASE | re.MULTILINE, +) +_CLEANUP_PHASE_AUTH_RE = re.compile( + r"^\s*[-*]?\s*cleanup phase (?:authorized|authorization)\s*:\s*true", + re.IGNORECASE | re.MULTILINE, +) +_DELETE_CAPABILITY_RE = re.compile( + r"^\s*[-*]?\s*delete.?branch capability(?: proven)?\s*:\s*true", + re.IGNORECASE | re.MULTILINE, +) +_BEFORE_AFTER_RE = re.compile( + r"^\s*[-*]?\s*before/after (?:state )?snapshot\s*:", + re.IGNORECASE | re.MULTILINE, +) +_SAFETY_PROOF_RE = re.compile( + r"^\s*[-*]?\s*(?:branch|worktree) safe to remove\s*:\s*true", + re.IGNORECASE | re.MULTILINE, +) + +_session: dict[str, Any] | None = None + + +def _blank_session() -> dict[str, Any]: + return { + "phase": PHASE_AUDIT, + "entered_from_task": None, + "cleanup_authorized": False, + "cleanup_authorization": {}, + } + + +def current_phase() -> str | None: + """Return active reconciliation phase or None when unset.""" + if not _session: + return None + return _session.get("phase") + + +def active_record() -> dict[str, Any] | None: + """Return a copy of the session record, if any.""" + return dict(_session) if _session else None + + +def clear_phase() -> None: + """Clear reconciliation phase state.""" + global _session + _session = None + + +def enter_audit_phase(task: str) -> dict[str, Any]: + """Enter read-only audit phase for a reconciliation task.""" + global _session + normalized = (task or "").strip().lower() + _session = _blank_session() + _session["entered_from_task"] = normalized + return dict(_session) + + +def authorize_cleanup_phase( + *, + operator_approved: bool = False, + workflow_authorized: bool = False, + delete_capability_proven: bool = False, + safety_proof: dict[str, Any] | None = None, + before_after_snapshot: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Authorize cleanup phase after explicit approval and safety proofs.""" + reasons: list[str] = [] + if not (operator_approved or workflow_authorized): + reasons.append( + "cleanup phase requires operator approval or explicit workflow " + "authorization" + ) + if not delete_capability_proven: + reasons.append( + "cleanup phase requires exact delete_branch capability proof " + "(gitea.branch.delete)" + ) + safety = dict(safety_proof or {}) + if not safety.get("safe_to_delete_remote") and not safety.get( + "safe_to_remove_worktree" + ): + reasons.append( + "cleanup phase requires proof that branch/worktree is safe to remove" + ) + snapshot = dict(before_after_snapshot or {}) + if not snapshot.get("before") or not snapshot.get("after"): + reasons.append( + "cleanup phase requires before/after state snapshot" + ) + + if reasons: + return { + "authorized": False, + "phase": current_phase() or PHASE_AUDIT, + "reasons": reasons, + "safe_next_action": ( + "remain in audit-only mode or supply operator approval, " + "delete_branch capability proof, safety proof, and " + "before/after snapshot before cleanup" + ), + } + + global _session + if _session is None: + _session = _blank_session() + _session["phase"] = PHASE_CLEANUP + _session["cleanup_authorized"] = True + _session["cleanup_authorization"] = { + "operator_approved": operator_approved, + "workflow_authorized": workflow_authorized, + "delete_capability_proven": delete_capability_proven, + "safety_proof": safety, + "before_after_snapshot": snapshot, + } + return { + "authorized": True, + "phase": PHASE_CLEANUP, + "reasons": [], + "cleanup_authorization": dict(_session["cleanup_authorization"]), + "safe_next_action": "proceed with authorized cleanup mutations only", + } + + +def check_audit_task_enters_phase(task: str) -> bool: + """Return whether resolving *task* should enter audit phase.""" + return (task or "").strip().lower() in AUDIT_PHASE_TASKS + + +def check_audit_mutation_allowed(task: str) -> tuple[bool, list[str]]: + """Fail closed when a mutation task runs during audit phase.""" + normalized = (task or "").strip().lower() + phase = current_phase() + if phase != PHASE_AUDIT: + return True, [] + if normalized in AUDIT_FORBIDDEN_TASKS: + return False, [ + f"task '{normalized}' is forbidden in audit-only reconciliation " + "mode: switch to an explicit cleanup phase with operator approval " + "and exact delete_branch capability proof before cleanup mutations" + ] + return True, [] + + +def check_cleanup_execution_allowed() -> tuple[bool, list[str]]: + """Fail closed when cleanup execution is attempted without authorization.""" + phase = current_phase() + if phase == PHASE_CLEANUP and (_session or {}).get("cleanup_authorized"): + return True, [] + if phase is None: + return False, [ + "cleanup execution requires an active reconciliation session; " + "resolve a reconciliation audit task first" + ] + return False, [ + "cleanup execution forbidden in audit-only reconciliation mode; " + "call gitea_authorize_reconciliation_cleanup_phase with operator " + "approval, delete_branch capability proof, safety proof, and " + "before/after snapshot" + ] + + +def classify_cleanup_mutation(action: str) -> str: + """Map a cleanup action to the required mutation ledger category (#419).""" + normalized = (action or "").strip().lower() + if "delete_remote" in normalized or normalized in { + "delete_branch", + "gitea_delete_branch", + }: + return "external-state" + if "branch" in normalized and "delete" in normalized: + return "git-ref" + if "worktree" in normalized or "remove_local" in normalized: + return "cleanup" + return "cleanup" + + +def assess_audit_reconciliation_report(report_text: str) -> dict[str, Any]: + """Validate audit/cleanup reconciliation reports (fail closed).""" + text = report_text or "" + reasons: list[str] = [] + + cleanup_occurred = bool(_CLEANUP_OCCURRED_RE.search(text)) + claims_no_mutations = bool(_NO_MUTATIONS_RE.search(text)) + + if cleanup_occurred and claims_no_mutations: + reasons.append( + "report claims no mutations but documents cleanup mutations; " + "audit-only reports must not perform cleanup and cleanup reports " + "must not claim no mutations" + ) + + if cleanup_occurred: + if not _CLEANUP_PHASE_AUTH_RE.search(text): + reasons.append( + "cleanup mutations reported without " + "'Cleanup phase authorized: true'" + ) + if not _DELETE_CAPABILITY_RE.search(text): + reasons.append( + "cleanup mutations reported without delete_branch capability " + "proof" + ) + if not _BEFORE_AFTER_RE.search(text): + reasons.append( + "cleanup mutations reported without before/after state snapshot" + ) + if not _SAFETY_PROOF_RE.search(text): + reasons.append( + "cleanup mutations reported without branch/worktree safety proof" + ) + + if re.search(r"delete_remote|remote branch.*delet", text, re.I): + if not _EXTERNAL_STATE_RE.search(text): + reasons.append( + "remote branch deletion must be classified under " + "External-state mutations" + ) + if re.search(r"git\s+branch\s+-D|local branch.*delet", text, re.I): + if not _GIT_REF_RE.search(text): + reasons.append( + "local branch deletion must be classified under " + "Git ref mutations" + ) + if re.search(r"worktree.*remov|remove_local_worktree", text, re.I): + if not _CLEANUP_MUTATIONS_RE.search(text): + reasons.append( + "worktree removal must be classified under Cleanup mutations" + ) + + if ( + RECONCILE_WORKFLOW_PATH.replace("workflows/", "") in text + or "reconcile-landed-pr" in text.lower() + ): + if cleanup_occurred and "audit phase" in text.lower(): + if "cleanup phase" not in text.lower(): + reasons.append( + "report mixes audit phase with cleanup mutations without " + "documenting cleanup phase transition" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "cleanup_occurred": cleanup_occurred, + "claims_no_mutations": claims_no_mutations, + "safe_next_action": ( + "proceed" + if proven + else "fix audit/cleanup report: separate audit from cleanup phase, " + "classify mutations, and do not claim no mutations after cleanup" + ), + } + + +def assess_audit_command_allowed(command: str) -> tuple[bool, list[str]]: + """Block shell commands that perform cleanup during audit phase.""" + phase = current_phase() + if phase != PHASE_AUDIT: + return True, [] + cmd = (command or "").strip() + if AUDIT_FORBIDDEN_COMMAND_RE.search(cmd): + return False, [ + f"command forbidden in audit-only reconciliation mode: {cmd!r}; " + "authorize cleanup phase before branch/worktree deletion or push" + ] + return True, [] \ No newline at end of file diff --git a/final_report_validator.py b/final_report_validator.py index bbc4b6c..0dae1a2 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -919,6 +919,22 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st ) +def _rule_audit_reconciliation_boundary(report_text: str) -> list[dict[str, str]]: + from audit_reconciliation_mode import assess_audit_reconciliation_report + + result = assess_audit_reconciliation_report(report_text) + if result.get("proven"): + return [] + return _findings_from_reasons( + "reconcile.audit_cleanup_boundary", + result.get("reasons") or [], + field="Audit/cleanup phase", + severity="block", + safe_next_action=result.get("safe_next_action") + or "separate audit from authorized cleanup and classify mutations", + ) + + def _rule_reviewer_review_mutation( report_text: str, *, @@ -977,6 +993,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_git_fetch_readonly, _rule_reviewer_legacy_workspace_mutations, _rule_reviewer_vague_mutations_none, + _rule_audit_reconciliation_boundary, ], "author_issue": [ _rule_shared_controller_handoff, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 17938e3..2fd613c 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -546,6 +546,7 @@ import issue_work_duplicate_gate # noqa: E402 import merged_cleanup_reconcile # noqa: E402 import reconciler_profile # noqa: E402 import reconciliation_workflow # noqa: E402 +import audit_reconciliation_mode # noqa: E402 import review_merge_state_machine # noqa: E402 import native_mcp_preference # noqa: E402 @@ -3680,6 +3681,18 @@ def gitea_delete_branch( "permission_report": _permission_block_report("gitea.branch.delete"), } + audit_allowed, audit_reasons = ( + audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch") + ) + if not audit_allowed: + return { + "success": False, + "performed": False, + "required_permission": "gitea.branch.delete", + "reasons": audit_reasons, + "audit_phase": audit_reconciliation_mode.current_phase(), + } + verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -3749,6 +3762,18 @@ def gitea_reconcile_merged_cleanups( "execute_confirmed must be True when dry_run=False (fail closed)" ) + if not dry_run: + exec_allowed, exec_reasons = ( + audit_reconciliation_mode.check_cleanup_execution_allowed() + ) + if not exec_allowed: + return { + "success": False, + "performed": False, + "reasons": exec_reasons, + "audit_phase": audit_reconciliation_mode.current_phase(), + } + h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) @@ -3831,6 +3856,53 @@ def gitea_reconcile_merged_cleanups( return {"success": True, "performed": True, **report} +@mcp.tool() +def gitea_authorize_reconciliation_cleanup_phase( + operator_approved: bool = False, + workflow_authorized: bool = False, + delete_capability_proven: bool = False, + safe_to_delete_remote: bool = False, + safe_to_remove_worktree: bool = False, + before_state: str = "", + after_state: str = "", +) -> dict: + """Authorize cleanup phase after audit-only reconciliation (#419). + + Requires operator or workflow approval, exact delete_branch capability proof, + branch/worktree safety proof, and before/after state snapshots. Audit phase + forbids branch deletion, worktree removal, pushes, and issue/PR mutations. + """ + delete_gate = _profile_operation_gate("gitea.branch.delete") + capability_ok = not bool(delete_gate) + if delete_capability_proven and delete_gate: + return { + "authorized": False, + "performed": False, + "delete_capability_verified": False, + "reasons": [ + "delete_capability_proven=true but active profile lacks " + "gitea.branch.delete", + ] + delete_gate, + "audit_phase": audit_reconciliation_mode.current_phase(), + } + result = audit_reconciliation_mode.authorize_cleanup_phase( + operator_approved=operator_approved, + workflow_authorized=workflow_authorized, + delete_capability_proven=delete_capability_proven and capability_ok, + safety_proof={ + "safe_to_delete_remote": safe_to_delete_remote, + "safe_to_remove_worktree": safe_to_remove_worktree, + }, + before_after_snapshot={ + "before": (before_state or "").strip(), + "after": (after_state or "").strip(), + }, + ) + result["performed"] = bool(result.get("authorized")) + result["delete_capability_verified"] = capability_ok + return result + + @mcp.tool() def gitea_assess_already_landed_reconciliation( pr_number: int, @@ -6937,6 +7009,9 @@ def gitea_resolve_task_capability( if reason_msg: result["reason"] = reason_msg role_session_router.sync_route_from_capability(result) + if audit_reconciliation_mode.check_audit_task_enters_phase(task): + phase_record = audit_reconciliation_mode.enter_audit_phase(task) + result["reconciliation_phase"] = phase_record.get("phase") was_terminal = capability_stop_terminal.is_active() terminal = capability_stop_terminal.sync_from_capability_result(result) if terminal: diff --git a/review_proofs.py b/review_proofs.py index b74ebb8..9a37277 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -5115,6 +5115,15 @@ def assess_pr_queue_cleanup_report(report_text: str | None) -> dict: return _assess(report_text or "") +def assess_audit_reconciliation_report(report_text: str | None) -> dict: + """#419: validate audit vs cleanup reconciliation report boundaries.""" + from audit_reconciliation_mode import ( + assess_audit_reconciliation_report as _assess, + ) + + return _assess(report_text or "") + + _GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I) _NOT_APPLICABLE_VALUE = re.compile( diff --git a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md index 10ba60d..a9f4cd1 100644 --- a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md +++ b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md @@ -304,6 +304,40 @@ If any required mutation capability is missing: * include safe next action (profile switch, human close, or dedicated reconciler profile) +## 15A. Audit vs cleanup phase (#419) + +Reconciliation audits are **read-only** unless a separate cleanup phase is +explicitly authorized. + +**Audit phase forbids** (``audit_reconciliation_mode.check_audit_mutation_allowed`` +fails closed): + +* ``gitea_delete_branch`` +* ``git branch -D`` +* ``git worktree remove`` +* pushes +* issue/PR mutations +* file edits + +Dry-run merged-cleanup reconciliation (``gitea_reconcile_merged_cleanups`` with +``dry_run=True``) stays in audit phase. Execution requires: + +1. Operator approval or workflow authorization +2. Exact ``delete_branch`` capability proof (``gitea.branch.delete``) +3. Proof branch/worktree is safe to remove +4. Before/after state snapshot + +Call ``gitea_authorize_reconciliation_cleanup_phase`` before any cleanup +mutation. Final reports must not claim ``no mutations`` if cleanup occurred. +Classify cleanup mutations as: + +* remote branch deletion → **External-state mutations** +* local branch deletion → **Git ref mutations** +* worktree removal → **Cleanup mutations** + +``audit_reconciliation_mode.assess_audit_reconciliation_report`` validates +these boundaries in final reports. + ## 16. Mutation classification Use precise mutation categories in the final report: diff --git a/task_capability_map.py b/task_capability_map.py index 1d40a78..9e7d0ba 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -104,6 +104,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.read", "role": "author", }, + "reconciliation_cleanup": { + "permission": "gitea.branch.delete", + "role": "author", + }, "work_issue": { "permission": "gitea.pr.create", "role": "author", diff --git a/tests/test_audit_reconciliation_mode.py b/tests/test_audit_reconciliation_mode.py new file mode 100644 index 0000000..a117fab --- /dev/null +++ b/tests/test_audit_reconciliation_mode.py @@ -0,0 +1,291 @@ +"""Tests for audit vs cleanup reconciliation mode (#419).""" +import os +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import audit_reconciliation_mode as arm +import mcp_server +from audit_reconciliation_mode import ( + AUDIT_FORBIDDEN_TASKS, + PHASE_AUDIT, + PHASE_CLEANUP, + assess_audit_command_allowed, + assess_audit_reconciliation_report, + authorize_cleanup_phase, + check_audit_mutation_allowed, + check_cleanup_execution_allowed, + classify_cleanup_mutation, + clear_phase, + enter_audit_phase, +) +from final_report_validator import assess_final_report_validator +from review_proofs import assess_audit_reconciliation_report as proofs_assess +from task_capability_map import required_permission, required_role + +DELETE_PROFILE = { + "profile_name": "prgs-author-delete", + "allowed_operations": ["gitea.read", "gitea.branch.delete"], + "forbidden_operations": [], + "audit_label": "prgs-author-delete", +} + +READ_PROFILE = { + "profile_name": "prgs-author", + "allowed_operations": ["gitea.read", "gitea.issue.comment"], + "forbidden_operations": ["gitea.branch.delete"], + "audit_label": "prgs-author", +} + +READ_ENV = { + "GITEA_MCP_CONFIG": os.path.join( + os.path.dirname(__file__), "..", "profiles.json" + ), + "GITEA_MCP_PROFILE": "prgs-author", +} + + +def _authorize_cleanup(**kwargs): + defaults = { + "operator_approved": True, + "delete_capability_proven": True, + "safety_proof": {"safe_to_delete_remote": True}, + "before_after_snapshot": { + "before": "remote branch exists", + "after": "remote branch absent", + }, + } + defaults.update(kwargs) + return authorize_cleanup_phase(**defaults) + + +def _cleanup_report(**overrides): + base = ( + "Task mode: reconcile-landed-pr\n" + "Workflow source: workflows/reconcile-landed-pr.md\n" + "Audit phase: read-only assessment\n" + "Cleanup phase authorized: true\n" + "Delete-branch capability proven: true\n" + "Branch safe to remove: true\n" + "Before/after state snapshot: remote branch feat/x present → absent\n" + "External-state mutations: deleted remote branch feat/x\n" + "Git ref mutations: none\n" + "Cleanup mutations: removed worktree branches/feat-x\n" + ) + for key, value in overrides.items(): + base = base.replace(key, value) + return base + + +class TestAuditPhaseGates(unittest.TestCase): + def setUp(self): + clear_phase() + enter_audit_phase("reconcile-landed-pr") + + def tearDown(self): + clear_phase() + + def test_audit_blocks_delete_branch_task(self): + allowed, reasons = check_audit_mutation_allowed("delete_branch") + self.assertFalse(allowed) + self.assertTrue(reasons) + + def test_audit_blocks_worktree_shell_commands(self): + allowed, reasons = assess_audit_command_allowed( + "git worktree remove branches/feat-x" + ) + self.assertFalse(allowed) + self.assertTrue(reasons) + + def test_audit_blocks_local_branch_delete_command(self): + allowed, reasons = assess_audit_command_allowed("git branch -D feat/x") + self.assertFalse(allowed) + + def test_audit_allows_read_tasks(self): + allowed, _ = check_audit_mutation_allowed("reconcile_landed_pr") + self.assertTrue(allowed) + + def test_forbidden_set_covers_issue_and_pr_mutations(self): + for task in ("close_pr", "comment_issue", "commit_files", "push_branch"): + self.assertIn(task, AUDIT_FORBIDDEN_TASKS) + + +class TestCleanupAuthorization(unittest.TestCase): + def setUp(self): + clear_phase() + enter_audit_phase("reconcile_merged_cleanups") + + def tearDown(self): + clear_phase() + + def test_cleanup_without_approval_blocked(self): + result = authorize_cleanup_phase( + delete_capability_proven=True, + safety_proof={"safe_to_delete_remote": True}, + before_after_snapshot={"before": "a", "after": "b"}, + ) + self.assertFalse(result["authorized"]) + + def test_cleanup_without_capability_proof_blocked(self): + result = _authorize_cleanup(delete_capability_proven=False) + self.assertFalse(result["authorized"]) + + def test_cleanup_without_snapshot_blocked(self): + result = _authorize_cleanup(before_after_snapshot={"before": "", "after": ""}) + self.assertFalse(result["authorized"]) + + def test_authorized_cleanup_switches_phase(self): + result = _authorize_cleanup() + self.assertTrue(result["authorized"]) + self.assertEqual(result["phase"], PHASE_CLEANUP) + + def test_cleanup_execution_allowed_only_after_authorization(self): + self.assertFalse(check_cleanup_execution_allowed()[0]) + _authorize_cleanup() + self.assertTrue(check_cleanup_execution_allowed()[0]) + + +class TestReportVerifier(unittest.TestCase): + def test_false_no_mutations_after_cleanup_blocked(self): + report = ( + "Task mode: reconcile-landed-pr\n" + "No mutations performed.\n" + "delete_remote_branch feat/dup\n" + ) + result = assess_audit_reconciliation_report(report) + self.assertFalse(result["proven"]) + self.assertIn("no mutations", result["reasons"][0].lower()) + + def test_cleanup_without_authorization_fields_blocked(self): + report = ( + "Task mode: reconcile-landed-pr\n" + "remove_local_worktree branches/feat-x\n" + ) + result = assess_audit_reconciliation_report(report) + self.assertFalse(result["proven"]) + + def test_authorized_cleanup_report_passes(self): + result = assess_audit_reconciliation_report(_cleanup_report()) + self.assertTrue(result["proven"]) + + def test_mutation_classification_enforced(self): + report = ( + "Task mode: reconcile-landed-pr\n" + "Cleanup phase authorized: true\n" + "Delete-branch capability proven: true\n" + "Branch safe to remove: true\n" + "Before/after state snapshot: present\n" + "remove_local_worktree branches/feat-x\n" + ) + result = assess_audit_reconciliation_report(report) + self.assertFalse(result["proven"]) + + def test_proofs_export_matches_module(self): + report = "No mutations performed.\ndelete_remote_branch feat/dup" + self.assertEqual( + proofs_assess(report)["proven"], + assess_audit_reconciliation_report(report)["proven"], + ) + + def test_final_report_validator_includes_boundary_rule(self): + report = "No mutations performed.\ndelete_remote_branch feat/dup" + result = assess_final_report_validator( + report_text=report, + task_kind="reconcile_already_landed", + ) + self.assertTrue(result["blocked"]) + rule_ids = [f["rule_id"] for f in result["findings"]] + self.assertIn("reconcile.audit_cleanup_boundary", rule_ids) + + +class TestMutationClassification(unittest.TestCase): + def test_remote_delete_is_external_state(self): + self.assertEqual( + classify_cleanup_mutation("delete_remote_branch"), + "external-state", + ) + + def test_worktree_remove_is_cleanup(self): + self.assertEqual( + classify_cleanup_mutation("remove_local_worktree"), + "cleanup", + ) + + +class TestMcpGates(unittest.TestCase): + def setUp(self): + clear_phase() + enter_audit_phase("reconcile_merged_cleanups") + self.mock_api = patch("mcp_server.api_request").start() + self.mock_auth = patch( + "mcp_server.get_auth_header", return_value="token test" + ).start() + + def tearDown(self): + patch.stopall() + clear_phase() + mcp_server._IDENTITY_CACHE.clear() + + @patch.dict(os.environ, READ_ENV, clear=True) + @patch("mcp_server.get_profile", return_value=DELETE_PROFILE) + def test_delete_branch_blocked_in_audit_phase(self, _profile): + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="author") + result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs") + self.assertFalse(result["success"]) + self.assertEqual(result["audit_phase"], PHASE_AUDIT) + self.mock_api.assert_not_called() + + @patch.dict(os.environ, READ_ENV, clear=True) + @patch("mcp_server.get_profile", return_value=DELETE_PROFILE) + def test_reconcile_execute_blocked_without_cleanup_auth(self, _profile): + with self.assertRaises(ValueError): + mcp_server.gitea_reconcile_merged_cleanups( + dry_run=False, + execute_confirmed=False, + remote="prgs", + ) + + @patch.dict(os.environ, READ_ENV, clear=True) + @patch("mcp_server.get_profile", return_value=READ_PROFILE) + def test_cleanup_auth_fails_without_delete_capability(self, _profile): + result = mcp_server.gitea_authorize_reconciliation_cleanup_phase( + operator_approved=True, + delete_capability_proven=True, + safe_to_delete_remote=True, + before_state="exists", + after_state="gone", + ) + self.assertFalse(result["authorized"]) + self.assertFalse(result["delete_capability_verified"]) + + @patch.dict(os.environ, READ_ENV, clear=True) + @patch("mcp_server.get_profile", return_value=DELETE_PROFILE) + def test_delete_branch_allowed_after_cleanup_authorization(self, _profile): + mcp_server.gitea_authorize_reconciliation_cleanup_phase( + operator_approved=True, + delete_capability_proven=True, + safe_to_delete_remote=True, + before_state="exists", + after_state="gone", + ) + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", resolved_role="author") + self.mock_api.return_value = {} + result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs") + self.assertTrue(result["success"]) + + +class TestTaskCapabilityMap(unittest.TestCase): + def test_reconciliation_cleanup_maps_delete_permission(self): + self.assertEqual( + required_permission("reconciliation_cleanup"), + "gitea.branch.delete", + ) + self.assertEqual(required_role("reconciliation_cleanup"), "author") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 44d80ce..d5f1fcb 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -87,6 +87,14 @@ def test_reconcile_landed_workflow_contract(): assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text assert "RECOVERY_HANDOFF_ONLY" in text assert "resolve_partial_reconciliation_plan" in text + assert "check_audit_mutation_allowed" in text + assert "gitea_authorize_reconciliation_cleanup_phase" in text + + +def test_audit_reconciliation_verifier_exported(): + from review_proofs import assess_audit_reconciliation_report + + assert callable(assess_audit_reconciliation_report) def test_create_issue_workflow_contract(): From 551ee7d70c7968d3c43015084faa13b0b56a46e2 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 17:20:53 -0400 Subject: [PATCH 4/7] fix: resolve conflicts for PR #421 From 3d11e1f12b659bf28e8dde06e1ca73268786197b Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 03:57:38 -0400 Subject: [PATCH 5/7] feat: isolate MCP workspace binding across role namespaces (Closes #510) Namespace-scoped workspace resolution prevents GITEA_AUTHOR_WORKTREE from poisoning reviewer, merger, and reconciler purity checks. Each role uses its own worktree env var with actionable binding-source errors and safe reconnect guidance. Preserves #274/#475 author and control-checkout guards. --- .env.example | 9 + author_mutation_worktree.py | 2 + docs/llm-workflow-runbooks.md | 39 +++ gitea_mcp_server.py | 217 ++++++++++++--- namespace_workspace_binding.py | 307 ++++++++++++++++++++++ tests/test_mcp_server.py | 11 +- tests/test_namespace_workspace_binding.py | 240 +++++++++++++++++ 7 files changed, 777 insertions(+), 48 deletions(-) create mode 100644 namespace_workspace_binding.py create mode 100644 tests/test_namespace_workspace_binding.py diff --git a/.env.example b/.env.example index 5463787..4777fac 100644 --- a/.env.example +++ b/.env.example @@ -46,3 +46,12 @@ GITEA_TOKEN_SOURCE=GITEA_TOKEN # profile's values. Leave unset for pure env-based configuration. GITEA_MCP_CONFIG=/Users/jasonwalker/.config/gitea-tools/profiles.json GITEA_MCP_PROFILE=prgs + +# Namespace-scoped active task workspaces (#510). Each MCP namespace uses only +# its own role env var; foreign bindings (e.g. GITEA_AUTHOR_WORKTREE in a +# merger process) are ignored. +# GITEA_AUTHOR_WORKTREE=/path/to/repo/branches/issue-123-work +# GITEA_REVIEWER_WORKTREE=/path/to/repo/branches/review-pr456 +# GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456 +# GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456 +# GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override diff --git a/author_mutation_worktree.py b/author_mutation_worktree.py index 29dc05d..dccb94a 100644 --- a/author_mutation_worktree.py +++ b/author_mutation_worktree.py @@ -12,6 +12,8 @@ import subprocess BASE_BRANCHES = frozenset({"master", "main", "dev"}) ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE" AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" +# Author-only: reviewer/merger/reconciler namespaces use role-specific env vars +# via namespace_workspace_binding (#510). def _normalize_path(path: str) -> str: diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index d67769d..cb904c4 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -823,6 +823,45 @@ scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md --push ``` +## Namespace workspace binding (#510) + +Each MCP namespace resolves its **own** active task workspace. Foreign role +worktree environment variables must not poison another namespace's purity +checks. + +| Namespace | Workspace env vars (in priority under `GITEA_ACTIVE_WORKTREE`) | Allowed roots | +|-----------|------------------------------------------------------------------|---------------| +| author | `GITEA_AUTHOR_WORKTREE` | `branches/` worktree only (#274) | +| reviewer | `GITEA_REVIEWER_WORKTREE` | clean `branches/` worktree | +| merger | `GITEA_MERGER_WORKTREE` | clean `branches/` worktree **or** clean control checkout | +| reconciler | `GITEA_RECONCILER_WORKTREE` | clean `branches/` worktree **or** clean control checkout | + +`GITEA_AUTHOR_WORKTREE` is **author-only**. Reviewer, merger, and reconciler +MCP processes ignore it even when it points at a dirty author WIP tree. + +### Safe reconnect / rebind procedure + +When a mutation blocks on workspace binding: + +1. Read the error — it names the **resolved workspace path**, **role + namespace**, and **binding source** (tool arg, env var, or process root). +2. Reconnect or relaunch the correct namespace MCP server from the intended + workspace (or set the role-specific env var before launch). +3. Pass `worktree_path` on reviewer/merger mutation tools when the active + branches/ worktree differs from the MCP process root. +4. **Do not** clean, reset, or discard foreign role worktrees to unblock your + own namespace — that destroys another agent's WIP. + +### CTH guidance for workspace binding blockers + +When posting a Canonical Thread Handoff after a binding blocker: + +- State which namespace was active (author / reviewer / merger / reconciler). +- Quote the resolved workspace path and binding source from the error. +- Name the safe reconnect action (relaunch MCP from `branches/...`, set + `GITEA_*_WORKTREE`, or pass `worktree_path`). +- Explicitly note that foreign worktrees must not be cleaned to unblock. + ## Safety notes - Never place raw tokens or passwords in any LLM MCP config; reference secrets diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 411a80b..d9608b0 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -175,12 +175,76 @@ _preflight_reviewer_violation_files: list[str] = [] ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE" AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" +REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE" +MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE" +RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE" + +import namespace_workspace_binding as nwb # noqa: E402 def _preflight_in_test_mode() -> bool: return "pytest" in sys.modules or "unittest" in sys.modules +def _reviewer_session_worktree() -> str | None: + import reviewer_pr_lease as _reviewer_pr_lease + + session = _reviewer_pr_lease.get_session_lease() + if not session: + return None + worktree = (session.get("worktree") or "").strip() + return worktree or None + + +def _effective_workspace_role() -> str: + """Resolve the namespace key used for workspace binding (#510).""" + profile = get_profile() + role = _preflight_resolved_role + if not role: + role = _role_kind( + profile.get("allowed_operations") or [], + profile.get("forbidden_operations") or [], + ) + return nwb.normalize_role_kind( + role, + profile_name=profile.get("profile_name"), + ) + + +def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: + """Resolve the namespace-scoped workspace root inspected by pre-flight guards.""" + role = _effective_workspace_role() + workspace, _source = nwb.resolve_namespace_workspace( + role_kind=role, + worktree_path=worktree_path, + process_project_root=PROJECT_ROOT, + session_lease_worktree=( + _reviewer_session_worktree() if role in {"reviewer", "merger"} else None + ), + profile_name=get_profile().get("profile_name"), + ) + return workspace + + +def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict: + """Canonical namespace workspace + repository root for guards (#460/#510).""" + role = _effective_workspace_role() + return nwb.resolve_namespace_mutation_context( + role_kind=role, + worktree_path=worktree_path, + process_project_root=PROJECT_ROOT, + session_lease_worktree=( + _reviewer_session_worktree() if role in {"reviewer", "merger"} else None + ), + profile_name=get_profile().get("profile_name"), + ) + + +def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict: + """Backward-compatible alias for namespace workspace context.""" + return _resolve_namespace_mutation_context(worktree_path) + + def _ensure_process_start_porcelain() -> str: """Capture the shared-worktree baseline once per MCP process (#252).""" global _process_start_porcelain @@ -189,26 +253,6 @@ def _ensure_process_start_porcelain() -> str: return _process_start_porcelain -def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: - """Resolve the workspace root inspected by pre-flight guards.""" - return author_mutation_worktree.resolve_mutation_workspace( - worktree_path, - PROJECT_ROOT, - active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV), - author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV), - ) - - -def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict: - """Canonical workspace + repository root for runtime_context and guards (#460).""" - return author_mutation_worktree.resolve_author_mutation_context( - worktree_path, - PROJECT_ROOT, - active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV), - author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV), - ) - - def _get_git_root(path: str) -> str | None: try: res = subprocess.run( @@ -276,7 +320,7 @@ def _format_preflight_files(files: list[str]) -> str: def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict: - ctx = _resolve_author_mutation_context(worktree_path) + ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] inspected_root = _get_git_root(workspace) process_root = ctx["process_project_root"] @@ -294,6 +338,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st "dirty_files": list(dirty_files), "dirty_scope": dirty_scope, "workspace_roots_aligned": ctx["roots_aligned"], + "workspace_role_kind": ctx.get("workspace_role_kind"), + "workspace_binding_source": ctx.get("workspace_binding_source"), + "ignored_bindings": list(ctx.get("ignored_bindings") or []), } if not ctx["roots_aligned"]: details["workspace_root_mismatch"] = ( @@ -304,13 +351,19 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st 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')}" - ) + parts = [ + f"MCP server process root: {details.get('mcp_server_process_root')}", + f"active task workspace root: {details.get('active_task_workspace_root')}", + f"workspace role: {details.get('workspace_role_kind')}", + f"binding source: {details.get('workspace_binding_source')}", + 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')}", + ] + ignored = details.get("ignored_bindings") or [] + if ignored: + parts.append(f"ignored foreign bindings: {'; '.join(ignored)}") + return "; ".join(parts) def assess_preflight_status(worktree_path: str | None = None) -> dict: @@ -333,6 +386,25 @@ 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)})" ) + role = _effective_workspace_role() + if role in nwb.NON_AUTHOR_ROLES: + binding = nwb.assess_metadata_only_worktree_binding( + role_kind=role, + declared_worktree_path=worktree_path, + mutation_workspace=_resolve_preflight_workspace_path(worktree_path), + process_project_root=PROJECT_ROOT, + profile_name=get_profile().get("profile_name"), + ) + if binding.get("block"): + reasons.append(binding["reasons"][0]) + reasons.append( + nwb.format_namespace_workspace_binding_error( + role_kind=role, + workspace_path=binding["mutation_workspace"], + binding_source="MCP server process root (default)", + reasons=binding.get("reasons"), + ) + ) return { "preflight_ready": not reasons, "preflight_block_reasons": reasons, @@ -466,13 +538,14 @@ def record_preflight_check( def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None: """#274: author file/branch mutations must run from a branches/ worktree. - Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a - Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE`` - (#468). + Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr`` + is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE`` + (#468). Non-author namespaces use dedicated workspace env vars (#510). """ - if _preflight_resolved_role in ("reviewer", "reconciler"): + role = _effective_workspace_role() + if role in nwb.NON_AUTHOR_ROLES: return - ctx = _resolve_author_mutation_context(worktree_path) + ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] git_state = issue_lock_worktree.read_worktree_git_state(workspace) assessment = author_mutation_worktree.assess_author_mutation_worktree( @@ -520,11 +593,12 @@ def verify_preflight_purity( f"'{task}' (fail closed)" ) - ctx = _resolve_author_mutation_context(worktree_path) + ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] canonical_root = ctx["canonical_repo_root"] process_root = ctx["process_project_root"] real_workspace = os.path.realpath(workspace) + role = ctx.get("workspace_role_kind") or _effective_workspace_role() if real_workspace != process_root: if not _preflight_in_test_mode(): @@ -541,11 +615,15 @@ def verify_preflight_purity( dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace))) if dirty_files: - details = _preflight_workspace_details(workspace, 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)}" + nwb.format_namespace_workspace_binding_error( + role_kind=role, + workspace_path=workspace, + binding_source=ctx.get("workspace_binding_source") + or "unknown binding source", + dirty_files=dirty_files, + ignored_bindings=ctx.get("ignored_bindings"), + ) ) else: if _preflight_whoami_violation: @@ -561,14 +639,14 @@ def verify_preflight_purity( f"{_format_preflight_files(_preflight_capability_violation_files)}" ) - if _preflight_resolved_role == "reviewer": + if role in {"reviewer", "merger"}: current = _get_workspace_porcelain() baseline = _preflight_capability_baseline_porcelain or "" reviewer_delta = _new_tracked_changes_since(baseline, current) _preflight_reviewer_violation_files = reviewer_delta if reviewer_delta: raise RuntimeError( - "Reviewer role violation: Reviewer profile is forbidden from modifying " + f"{role.title()} role violation: profile is forbidden from modifying " "tracked workspace files (fail closed). Offending files: " f"{_format_preflight_files(reviewer_delta)}" ) @@ -576,6 +654,45 @@ def verify_preflight_purity( _enforce_branches_only_author_mutation(worktree_path) _clear_preflight_capability_state() + +def _verify_role_mutation_workspace( + remote: str | None = None, + *, + worktree_path: str | None = None, + worktree: str | None = None, + task: str | None = None, +) -> str: + """Bind reviewer/merger mutations to the active namespace workspace (#510).""" + role = _effective_workspace_role() + git_state = issue_lock_worktree.read_worktree_git_state( + _resolve_preflight_workspace_path(worktree_path) + ) + assessment = nwb.assess_namespace_mutation_workspace( + role_kind=role, + worktree_path=worktree_path, + worktree=worktree, + process_project_root=PROJECT_ROOT, + session_lease_worktree=( + _reviewer_session_worktree() if role in {"reviewer", "merger"} else None + ), + profile_name=get_profile().get("profile_name"), + current_branch=git_state.get("current_branch"), + ) + if assessment["block"]: + raise RuntimeError( + nwb.format_namespace_workspace_binding_error( + role_kind=role, + workspace_path=assessment["mutation_workspace"], + binding_source=assessment.get("workspace_binding_source") + or "unknown binding source", + reasons=assessment.get("reasons"), + ignored_bindings=assessment.get("ignored_bindings"), + ) + ) + resolved = assessment["mutation_workspace"] + verify_preflight_purity(remote, worktree_path=resolved, task=task) + return resolved + from mcp.server.fastmcp import FastMCP # noqa: E402 from gitea_auth import ( # noqa: E402 @@ -2711,9 +2828,12 @@ 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, task="review_pr") + _verify_role_mutation_workspace( + remote, worktree_path=worktree_path, task="review_pr" + ) action = (action or "").strip().lower() result = { "requested_action": action, @@ -3113,6 +3233,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( @@ -3125,6 +3246,7 @@ def gitea_dry_run_pr_review( org=org, repo=repo, live=False, + worktree_path=worktree_path, ) @@ -3140,6 +3262,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. @@ -3158,6 +3281,7 @@ def gitea_submit_pr_review( repo=repo, live=True, final_review_decision_ready=final_review_decision_ready, + worktree_path=worktree_path, ) @@ -3518,6 +3642,7 @@ def gitea_merge_pr( host: str | None = None, org: str | None = None, repo: str | None = None, + worktree_path: str | None = None, ) -> dict: """Gated merge of a Gitea pull request (#16). @@ -3564,6 +3689,10 @@ def gitea_merge_pr( host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. + worktree_path: Merger workspace path under ``branches/`` or clean + control checkout; defaults to ``GITEA_MERGER_WORKTREE``, + ``GITEA_ACTIVE_WORKTREE``, or the MCP server process root. Ignores + foreign ``GITEA_AUTHOR_WORKTREE`` bindings (#510). Returns: dict describing the attempt: performed, authenticated user, profile @@ -3571,7 +3700,9 @@ def gitea_merge_pr( reasons/gates passed or blocked, and merge result / merge commit if available. Never secrets. """ - verify_preflight_purity(remote, task="merge_pr") + _verify_role_mutation_workspace( + remote, worktree_path=worktree_path, task="merge_pr" + ) do = (do or "").strip().lower() result = { "performed": False, @@ -5054,7 +5185,7 @@ def gitea_acquire_reviewer_pr_lease( "permission_report": _permission_block_report("gitea.pr.comment"), } - verify_preflight_purity(remote, task="review_pr") + _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr") h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) profile = get_profile() diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py new file mode 100644 index 0000000..4831907 --- /dev/null +++ b/namespace_workspace_binding.py @@ -0,0 +1,307 @@ +"""Namespace-scoped MCP workspace binding (#510). + +Each role namespace (author, reviewer, merger, reconciler) resolves its own +active task workspace. Foreign role worktree environment variables must not +poison workspace purity checks in another namespace. +""" + +from __future__ import annotations + +import os + +import author_mutation_worktree as amw + +ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV +AUTHOR_WORKTREE_ENV = amw.AUTHOR_WORKTREE_ENV +REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE" +MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE" +RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE" + +ROLE_WORKTREE_ENVS: dict[str, str] = { + "author": AUTHOR_WORKTREE_ENV, + "reviewer": REVIEWER_WORKTREE_ENV, + "merger": MERGER_WORKTREE_ENV, + "reconciler": RECONCILER_WORKTREE_ENV, +} + +NON_AUTHOR_ROLES = frozenset({"reviewer", "merger", "reconciler"}) + + +def normalize_role_kind( + role_kind: str | None, + *, + profile_name: str | None = None, +) -> str: + """Map profile/task role to a workspace namespace key.""" + role = (role_kind or "author").strip().lower() + profile = (profile_name or "").strip().lower() + if role == "reviewer" and "merger" in profile: + return "merger" + if role in ROLE_WORKTREE_ENVS: + return role + return "author" + + +def _env_value(env: dict[str, str] | os._Environ, key: str) -> str | None: + text = (env.get(key) or "").strip() + return text or None + + +def resolve_namespace_workspace( + *, + role_kind: str, + worktree_path: str | None = None, + worktree: str | None = None, + process_project_root: str, + env: dict[str, str] | os._Environ | None = None, + session_lease_worktree: str | None = None, + profile_name: str | None = None, +) -> tuple[str, str]: + """Return ``(resolved_path, binding_source)`` for *role_kind*.""" + env_map = env if env is not None else os.environ + role = normalize_role_kind(role_kind, profile_name=profile_name) + role_env_key = ROLE_WORKTREE_ENVS[role] + + for candidate, source in ( + (worktree_path, "worktree_path argument"), + (worktree, "worktree argument"), + (_env_value(env_map, ACTIVE_WORKTREE_ENV), f"{ACTIVE_WORKTREE_ENV} environment variable"), + (_env_value(env_map, role_env_key), f"{role_env_key} environment variable"), + (session_lease_worktree if role in {"reviewer", "merger"} else None, + "reviewer PR lease worktree"), + ): + text = (candidate or "").strip() + if text: + return os.path.realpath(os.path.abspath(text)), source + + return os.path.realpath(process_project_root), "MCP server process root (default)" + + +def resolve_namespace_mutation_context( + *, + role_kind: str, + worktree_path: str | None, + process_project_root: str, + env: dict[str, str] | os._Environ | None = None, + session_lease_worktree: str | None = None, + worktree: str | None = None, + profile_name: str | None = None, +) -> dict: + """Shared workspace resolution for runtime_context and mutation guards.""" + workspace, binding_source = resolve_namespace_workspace( + role_kind=role_kind, + worktree_path=worktree_path, + worktree=worktree, + process_project_root=process_project_root, + env=env, + session_lease_worktree=session_lease_worktree, + profile_name=profile_name, + ) + process_root = os.path.realpath(process_project_root) + role = normalize_role_kind(role_kind, profile_name=profile_name) + pollution = assess_foreign_role_worktree_pollution( + role_kind=role, + resolved_workspace=workspace, + binding_source=binding_source, + env=env, + profile_name=profile_name, + ) + canonical_root = amw.resolve_canonical_repo_root(process_root, process_root) + return { + "workspace_path": workspace, + "workspace_binding_source": binding_source, + "workspace_role_kind": role, + "ignored_bindings": pollution.get("ignored_bindings") or [], + "process_project_root": process_root, + "canonical_repo_root": canonical_root, + "roots_aligned": canonical_root == process_root, + } + + +def assess_foreign_role_worktree_pollution( + *, + role_kind: str, + resolved_workspace: str, + binding_source: str, + env: dict[str, str] | os._Environ | None = None, + profile_name: str | None = None, +) -> dict: + """Detect when a foreign role env would have hijacked workspace binding.""" + env_map = env if env is not None else os.environ + role = normalize_role_kind(role_kind, profile_name=profile_name) + if role == "author": + return {"would_pollute": False, "ignored_bindings": []} + + ignored: list[str] = [] + author_path = _env_value(env_map, AUTHOR_WORKTREE_ENV) + if author_path: + author_real = os.path.realpath(os.path.abspath(author_path)) + resolved_real = os.path.realpath(resolved_workspace) + if author_real != resolved_real and binding_source != f"{AUTHOR_WORKTREE_ENV} environment variable": + ignored.append( + f"{AUTHOR_WORKTREE_ENV}={author_real} (ignored for {role} namespace)" + ) + return { + "would_pollute": bool(ignored), + "ignored_bindings": ignored, + } + + +def assess_metadata_only_worktree_binding( + *, + role_kind: str, + declared_worktree_path: str | None, + mutation_workspace: str, + process_project_root: str, + profile_name: str | None = None, +) -> 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) + role = normalize_role_kind(role_kind, profile_name=profile_name) + 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": [ + f"worktree_path is metadata-only for {role} 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_namespace_workspace_binding_error( + *, + role_kind: str, + workspace_path: str, + binding_source: str, + reasons: list[str] | None = None, + ignored_bindings: list[str] | None = None, + dirty_files: list[str] | None = None, +) -> str: + """Canonical error when namespace workspace binding blocks mutations.""" + role = normalize_role_kind(role_kind) + workspace = os.path.realpath(workspace_path) + parts = [ + f"Namespace workspace binding blocked ({role} namespace, #510): " + f"resolved workspace '{workspace}' via {binding_source}." + ] + if ignored_bindings: + parts.append( + "Foreign role bindings ignored: " + "; ".join(ignored_bindings) + "." + ) + if dirty_files: + parts.append( + "Dirty tracked files in active task workspace: " + + ", ".join(dirty_files) + + "." + ) + if reasons: + parts.append("Details: " + "; ".join(reasons) + ".") + parts.append( + "Remediation: reconnect or relaunch the MCP server from a clean dedicated " + f"branches/ {role} worktree, set " + f"{ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} or {ACTIVE_WORKTREE_ENV} " + "to that path, or pass worktree_path on mutation tools. Do not clean or " + "reset foreign role worktrees to unblock this namespace." + ) + return " ".join(parts) + + +def assess_namespace_mutation_workspace( + *, + role_kind: str, + worktree_path: str | None, + worktree: str | None, + process_project_root: str, + env: dict[str, str] | os._Environ | None = None, + session_lease_worktree: str | None = None, + profile_name: str | None = None, + current_branch: str | None = None, +) -> dict: + """Evaluate namespace workspace binding before preflight/mutation.""" + ctx = resolve_namespace_mutation_context( + role_kind=role_kind, + worktree_path=worktree_path, + worktree=worktree, + process_project_root=process_project_root, + env=env, + session_lease_worktree=session_lease_worktree, + profile_name=profile_name, + ) + mutation_workspace = ctx["workspace_path"] + binding_source = ctx["workspace_binding_source"] + role = ctx["workspace_role_kind"] + process_root = ctx["process_project_root"] + + metadata = assess_metadata_only_worktree_binding( + role_kind=role, + declared_worktree_path=worktree_path, + mutation_workspace=mutation_workspace, + process_project_root=process_root, + profile_name=profile_name, + ) + pollution = assess_foreign_role_worktree_pollution( + role_kind=role, + resolved_workspace=mutation_workspace, + binding_source=binding_source, + env=env, + profile_name=profile_name, + ) + + reasons = list(metadata.get("reasons") or []) + if role == "author": + branches = amw.assess_author_mutation_worktree( + workspace_path=mutation_workspace, + project_root=ctx["canonical_repo_root"], + current_branch=current_branch, + ) + if branches["block"]: + reasons.extend(branches["reasons"]) + elif ( + role == "reviewer" + and mutation_workspace == process_root + and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]) + ): + reasons.append( + f"{role} mutation blocked: workspace is the stable control checkout; " + f"create or reconnect to a session-owned worktree under branches/ " + f"or set {ROLE_WORKTREE_ENVS[role]} / {ACTIVE_WORKTREE_ENV}" + ) + elif ( + role in {"reviewer", "merger"} + and mutation_workspace != process_root + and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]) + ): + reasons.append( + f"{role} mutation blocked: workspace '{mutation_workspace}' is not under " + f"'{ctx['canonical_repo_root']}/branches/'" + ) + + block = bool(reasons) + return { + "block": block, + "reasons": reasons, + "mutation_workspace": mutation_workspace, + "workspace_binding_source": binding_source, + "workspace_role_kind": role, + "process_project_root": process_root, + "canonical_repo_root": ctx["canonical_repo_root"], + "metadata_only": metadata.get("metadata_only", False), + "declared_worktree_path": metadata.get("declared_worktree_path"), + "ignored_bindings": pollution.get("ignored_bindings") or [], + } \ No newline at end of file diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 651b253..179159d 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -3879,7 +3879,7 @@ class TestPreflightVerification(unittest.TestCase): os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n" with self.assertRaises(RuntimeError) as ctx: mcp_server.verify_preflight_purity() - self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception)) + self.assertIn("forbidden from modifying tracked workspace files", str(ctx.exception)) self.assertIn("reviewer_edit.py", str(ctx.exception)) # Foreign pre-existing dirty state does not block when unchanged. @@ -3945,7 +3945,8 @@ class TestPreflightVerification(unittest.TestCase): 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) + self.assertIn("resolved workspace", msg) + self.assertIn(worktree, msg) + self.assertIn("worktree_path argument", msg) + self.assertIn("task_file.py", msg) + self.assertIn("author namespace", msg) diff --git a/tests/test_namespace_workspace_binding.py b/tests/test_namespace_workspace_binding.py new file mode 100644 index 0000000..b9c270c --- /dev/null +++ b/tests/test_namespace_workspace_binding.py @@ -0,0 +1,240 @@ +"""Tests for namespace-scoped MCP workspace binding (#510).""" + +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 namespace_workspace_binding as nwb # 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) + +AUTHOR_DIRTY = f"{CONTROL_ROOT}/branches/mcp-author-worktree" +MERGER_CLEAN = f"{CONTROL_ROOT}/branches/merge-pr487-submit" +REVIEWER_CLEAN = f"{CONTROL_ROOT}/branches/review-pr487-submit" +RECONCILER_CLEAN = f"{CONTROL_ROOT}/branches/reconcile-pr487" +MCP_PROCESS_ROOT = CONTROL_ROOT + + +class TestNamespaceWorkspaceModule(unittest.TestCase): + def test_author_env_ignored_for_merger_namespace(self): + workspace, source = nwb.resolve_namespace_workspace( + role_kind="merger", + worktree_path=None, + process_project_root=MCP_PROCESS_ROOT, + env={ + nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY, + nwb.MERGER_WORKTREE_ENV: MERGER_CLEAN, + }, + profile_name="gitea-merger", + ) + self.assertEqual(workspace, os.path.realpath(MERGER_CLEAN)) + self.assertEqual(source, f"{nwb.MERGER_WORKTREE_ENV} environment variable") + + def test_author_env_ignored_for_reviewer_namespace(self): + workspace, source = nwb.resolve_namespace_workspace( + role_kind="reviewer", + worktree_path=None, + process_project_root=MCP_PROCESS_ROOT, + env={ + nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY, + nwb.REVIEWER_WORKTREE_ENV: REVIEWER_CLEAN, + }, + ) + self.assertEqual(workspace, os.path.realpath(REVIEWER_CLEAN)) + self.assertEqual(source, f"{nwb.REVIEWER_WORKTREE_ENV} environment variable") + + def test_author_env_ignored_for_reconciler_namespace(self): + workspace, source = nwb.resolve_namespace_workspace( + role_kind="reconciler", + worktree_path=None, + process_project_root=MCP_PROCESS_ROOT, + env={ + nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY, + nwb.RECONCILER_WORKTREE_ENV: RECONCILER_CLEAN, + }, + ) + self.assertEqual(workspace, os.path.realpath(RECONCILER_CLEAN)) + self.assertEqual(source, f"{nwb.RECONCILER_WORKTREE_ENV} environment variable") + + def test_merger_profile_maps_to_merger_namespace(self): + role = nwb.normalize_role_kind("reviewer", profile_name="gitea-merger") + self.assertEqual(role, "merger") + + def test_error_message_includes_path_and_binding_source(self): + msg = nwb.format_namespace_workspace_binding_error( + role_kind="merger", + workspace_path=AUTHOR_DIRTY, + binding_source=f"{nwb.AUTHOR_WORKTREE_ENV} environment variable", + dirty_files=["gitea_mcp_server.py"], + ignored_bindings=[f"{nwb.AUTHOR_WORKTREE_ENV}={AUTHOR_DIRTY} (ignored for merger namespace)"], + ) + self.assertIn(AUTHOR_DIRTY, msg) + self.assertIn("via", msg.lower()) + self.assertIn(nwb.AUTHOR_WORKTREE_ENV, msg) + self.assertIn("Do not clean or reset foreign role worktrees", msg) + + +class TestNamespaceWorkspaceIntegration(unittest.TestCase): + def setUp(self): + 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, + "capability_violation": srv._preflight_capability_violation, + "in_test": srv._preflight_in_test_mode, + } + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_whoami_violation = False + srv._preflight_capability_violation = False + srv._preflight_in_test_mode = lambda: False + self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False) + self._env_patch.start() + + def tearDown(self): + 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_capability_violation = self._saved["capability_violation"] + srv._preflight_in_test_mode = self._saved["in_test"] + self._env_patch.stop() + for key in ( + nwb.AUTHOR_WORKTREE_ENV, + nwb.MERGER_WORKTREE_ENV, + nwb.REVIEWER_WORKTREE_ENV, + nwb.RECONCILER_WORKTREE_ENV, + nwb.ACTIVE_WORKTREE_ENV, + ): + os.environ.pop(key, None) + + def _merger_profile(self): + return { + "profile_name": "gitea-merger", + "allowed_operations": ["gitea.pr.merge", "gitea.read"], + "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + } + + def _reviewer_profile(self): + return { + "profile_name": "prgs-reviewer", + "allowed_operations": ["gitea.pr.approve", "gitea.pr.review", "gitea.read"], + "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + } + + def _reconciler_profile(self): + return { + "profile_name": "prgs-reconciler", + "allowed_operations": ["gitea.pr.close", "gitea.read"], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.create", + ], + } + + def _author_profile(self): + return { + "profile_name": "prgs-author", + "allowed_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.read"], + "forbidden_operations": ["gitea.pr.merge", "gitea.pr.approve"], + } + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_dirty_author_worktree_does_not_block_merger_with_clean_workspace( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY + os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN + srv._preflight_resolved_role = "reviewer" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): + srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN) + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_dirty_author_worktree_does_not_block_reviewer_with_clean_workspace( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY + os.environ[nwb.REVIEWER_WORKTREE_ENV] = REVIEWER_CLEAN + srv._preflight_resolved_role = "reviewer" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._reviewer_profile()): + srv.verify_preflight_purity("prgs", worktree_path=REVIEWER_CLEAN) + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_dirty_author_worktree_does_not_block_reconciler_with_clean_workspace( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY + os.environ[nwb.RECONCILER_WORKTREE_ENV] = RECONCILER_CLEAN + srv._preflight_resolved_role = "reconciler" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._reconciler_profile()): + srv.verify_preflight_purity("prgs", worktree_path=RECONCILER_CLEAN) + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_dirty_active_task_workspace_still_blocks_mutations( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN + srv._preflight_resolved_role = "reviewer" + dirty = " M namespace_workspace_binding.py\n" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): + with mock.patch("gitea_mcp_server._get_workspace_porcelain", return_value=dirty): + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN) + self.assertIn(MERGER_CLEAN, str(ctx.exception)) + self.assertIn("binding", str(ctx.exception).lower()) + + def test_root_workspace_mutation_still_blocked_for_author(self): + srv._preflight_resolved_role = "author" + with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._author_profile()): + with mock.patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={"current_branch": "master"}, + ): + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity("prgs") + self.assertIn("stable control checkout", str(ctx.exception)) + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_pr487_style_merge_binds_clean_merger_workspace( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY + srv._preflight_resolved_role = "reviewer" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): + resolved = srv._verify_role_mutation_workspace("prgs") + self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT)) \ No newline at end of file From d56baa635e53f71be3a895816cecbea411f9363b Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 04:42:35 -0400 Subject: [PATCH 6/7] fix: register gitea_lock_issue as MCP tool (Closes #521) Move @mcp.tool() from internal _list_open_pulls helper onto gitea_lock_issue so author sessions can discover and call the issue lock required by gitea_create_pr. Add regression tests for MCP registration and create_pr lock flow. --- gitea_mcp_server.py | 2 +- tests/test_lock_issue_mcp_registration.py | 127 ++++++++++++++++++++++ 2 files changed, 128 insertions(+), 1 deletion(-) create mode 100644 tests/test_lock_issue_mcp_registration.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index d9608b0..c2514bd 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1446,7 +1446,6 @@ def gitea_create_issue( return _with_optional_url({"number": data["number"]}, data.get("html_url")) -@mcp.tool() def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]: """Fetch all OPEN pull requests for a repo (used for stacked-base proof, #484).""" try: @@ -1457,6 +1456,7 @@ def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]: ) +@mcp.tool() def gitea_lock_issue( issue_number: int, branch_name: str, diff --git a/tests/test_lock_issue_mcp_registration.py b/tests/test_lock_issue_mcp_registration.py new file mode 100644 index 0000000..4b20763 --- /dev/null +++ b/tests/test_lock_issue_mcp_registration.py @@ -0,0 +1,127 @@ +"""Regression tests for gitea_lock_issue MCP tool registration (#521).""" + +from __future__ import annotations + +import os +import tempfile +import unittest +from unittest.mock import patch + +import issue_lock_store +import mcp_server +from mcp_server import gitea_create_pr, gitea_lock_issue + +FAKE_AUTH = "Basic dGVzdDp0ZXN0" + +ISSUE_WRITE_ENV = { + "GITEA_ALLOWED_OPERATIONS": ( + "gitea.issue.create,gitea.issue.close,gitea.issue.comment" + ), +} + +CREATE_PR_ENV = { + "GITEA_PROFILE_NAME": "author-test", + "GITEA_ALLOWED_OPERATIONS": ( + "gitea.read,gitea.pr.create,gitea.branch.push," + "gitea.issue.create,gitea.issue.close,gitea.issue.comment" + ), + "GITEA_FORBIDDEN_OPERATIONS": ( + "gitea.pr.approve,gitea.pr.merge,gitea.pr.review" + ), +} + + +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", + } + + +def _registered_tool_names() -> set[str]: + manager = mcp_server.mcp._tool_manager + tools = getattr(manager, "_tools", None) or {} + return set(tools.keys()) + + +class TestLockIssueMcpRegistration(unittest.TestCase): + def test_gitea_lock_issue_registered_as_public_mcp_tool(self): + names = _registered_tool_names() + self.assertIn("gitea_lock_issue", names) + + def test_internal_list_open_pulls_not_exposed_as_mcp_tool(self): + names = _registered_tool_names() + self.assertNotIn("_list_open_pulls", names) + + +class TestCreatePrLockRegistrationFlow(unittest.TestCase): + def setUp(self): + self._lock_dir = tempfile.TemporaryDirectory() + self._env_patcher = patch.dict( + os.environ, + { + **CREATE_PR_ENV, + "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name, + }, + clear=True, + ) + self._env_patcher.start() + self._dup_fetcher_patcher = patch( + "mcp_server.issue_duplicate_context_fetcher", + return_value=([], [], {"status": "not_claimed"}), + ) + self._dup_fetcher_patcher.start() + + def tearDown(self): + self._dup_fetcher_patcher.stop() + self._env_patcher.stop() + self._lock_dir.cleanup() + + @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_still_fails_closed_without_issue_lock(self, _auth, _role): + with self.assertRaises(RuntimeError) as ctx: + gitea_create_pr( + title="feat: X Closes #521", + head="feat/issue-521-lock-issue-registration", + remote="prgs", + ) + self.assertIn("Issue lock is missing", str(ctx.exception)) + + @patch("mcp_server.api_request") + @patch( + "mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value=_clean_master_git_state_for_lock(), + ) + @patch("mcp_server.api_get_all", return_value=[]) + @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_proceeds_after_valid_issue_lock( + self, _auth, _role, _api, _git_state, mock_api_request + ): + worktree = os.path.realpath(os.getcwd()) + mock_api_request.return_value = {"number": 521, "html_url": "https://example/pr/521"} + lock_res = gitea_lock_issue( + issue_number=521, + branch_name="feat/issue-521-lock-issue-registration", + remote="prgs", + worktree_path=worktree, + ) + self.assertTrue(lock_res["success"]) + lock_path = lock_res["lock_file_path"] + self.assertTrue(os.path.exists(lock_path)) + lock = issue_lock_store.read_lock_file(lock_path) + self.assertEqual(lock["issue_number"], 521) + + res = gitea_create_pr( + title="fix: restore lock tool registration Closes #521", + head="feat/issue-521-lock-issue-registration", + remote="prgs", + worktree_path=worktree, + ) + self.assertEqual(res["number"], 521) \ No newline at end of file From 2faf18a80213d7813ab36d08d0c9efa87c12d1d0 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 04:23:17 -0400 Subject: [PATCH 7/7] feat: add safe post-merge moot lease cleanup for already-merged PRs (Closes #515) Refuse merge-oriented reviewer-lease acquisition/adoption on an already merged/closed PR, and add a read-first cleanup path for a lingering moot lease. - reviewer_pr_lease.assess_acquire_lease: new pr_merged_or_closed flag; fail closed with a post_merge_moot reason and no lease body when the PR already merged/closed. - reviewer_pr_lease.assess_post_merge_moot_lease: pure assessment. Only treats a lease as moot when the PR is merged/closed; never proposes touching an active lease on an open PR; provides a terminal phase:released (blocker: post-merge-moot) body to neutralise the moot lease append-only; idempotent once the released marker is the newest entry. - gitea_acquire_reviewer_pr_lease: fetch live PR state, pass the flag, surface post_merge_moot on refusal (no comment posted). - gitea_cleanup_post_merge_moot_lease: new read-first tool. Reports merged state, merge_commit_sha, linked-issue closure, lease-moot, cleanup performed/skipped, and no_merge_or_adoption. apply=True posts the released marker only when merged/closed with an active lease; refuses on open PRs. - tests/test_post_merge_moot_lease.py: 10 tests covering acquire refusal on merged PR (no post), safe/idempotent cleanup, and open-PR foreign-lease protection. No relaxation of #407 reviewer-lease or #16 gated-merge gates for open PRs. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 129 ++++++++++++++ reviewer_pr_lease.py | 109 +++++++++++- tests/test_post_merge_moot_lease.py | 257 ++++++++++++++++++++++++++++ 3 files changed, 494 insertions(+), 1 deletion(-) create mode 100644 tests/test_post_merge_moot_lease.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index c2514bd..ca164e0 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -5195,6 +5195,13 @@ def gitea_acquire_reviewer_pr_lease( comments = _fetch_pr_comments( pr_number, remote=remote, host=host, org=org, repo=repo) + # Refuse merge-oriented lease acquisition/adoption on an already-merged or + # closed PR: the lease is moot and adopting it for merge work is unsafe (#515). + pr_live = api_request( + "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {} + pr_merged_or_closed = bool( + pr_live.get("merged") or pr_live.get("merged_at") + ) or (str(pr_live.get("state") or "").strip().lower() == "closed") assessment = reviewer_pr_lease.assess_acquire_lease( comments, pr_number=pr_number, @@ -5207,6 +5214,7 @@ def gitea_acquire_reviewer_pr_lease( candidate_head=candidate_head, target_branch=target_branch, target_branch_sha=target_branch_sha, + pr_merged_or_closed=pr_merged_or_closed, ) if not assessment.get("acquire_allowed"): return { @@ -5214,6 +5222,7 @@ def gitea_acquire_reviewer_pr_lease( "acquired": False, "reasons": assessment.get("reasons") or [], "existing_lease": assessment.get("existing_lease"), + "post_merge_moot": assessment.get("post_merge_moot", False), } body = assessment["lease_body"] @@ -5362,6 +5371,126 @@ def gitea_assess_reviewer_pr_lease( } +@mcp.tool() +def gitea_cleanup_post_merge_moot_lease( + pr_number: int, + apply: bool = False, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Safely resolve a moot reviewer lease left on an ALREADY-MERGED/closed PR (#515). + + Read-first and fail-safe. This tool never merges and never adopts a lease. + It only acts when the live PR state is merged/closed, and it never steals or + force-cleans an active *foreign* lease while the PR is still open. When + ``apply`` is true and a lease is still active on a merged/closed PR, it posts + a terminal ``phase: released`` lease marker (``blocker: post-merge-moot``) — + an append-only comment that neutralises the moot lease without deleting any + other session's comment. + + Args: + pr_number: The PR whose lingering lease to assess/clean. + apply: When false (default) report only (read-only). When true, post the + terminal released marker if — and only if — cleanup is allowed. + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict reporting PR merged/closed state, merge_commit_sha, linked-issue + closure state, whether the lease is moot, whether cleanup was performed + or skipped (and why), and ``no_merge_or_adoption`` True — this path never + merges or adopts. + """ + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "cleanup_performed": False, + "no_merge_or_adoption": True, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + pr_live = api_request( + "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {} + comments = _fetch_pr_comments( + pr_number, remote=remote, host=host, org=org, repo=repo) + assessment = reviewer_pr_lease.assess_post_merge_moot_lease( + comments, + pr_number=pr_number, + pr_merged=bool(pr_live.get("merged") or pr_live.get("merged_at")), + pr_state=pr_live.get("state"), + merge_commit_sha=pr_live.get("merge_commit_sha"), + ) + + # Best-effort linked-issue closure state for the report. + active = assessment.get("active_lease") or {} + issue_no = active.get("issue_number") + linked_issue_state = None + if issue_no: + try: + issue = api_request( + "GET", f"{repo_api_url(h, o, r)}/issues/{issue_no}", auth) or {} + linked_issue_state = issue.get("state") + except Exception: + linked_issue_state = None + + report = { + "success": True, + "pr_number": pr_number, + "pr_state": assessment.get("pr_state"), + "pr_merged_or_closed": assessment.get("pr_merged_or_closed"), + "merge_commit_sha": assessment.get("merge_commit_sha"), + "linked_issue_number": issue_no, + "linked_issue_state": linked_issue_state, + "lease_moot": assessment.get("is_moot"), + "active_lease": assessment.get("active_lease"), + "cleanup_allowed": assessment.get("cleanup_allowed"), + "cleanup_performed": False, + "no_merge_or_adoption": True, + "mode": "apply" if apply else "read_only", + "reasons": assessment.get("reasons") or [], + } + + if not apply: + return report + if not assessment.get("cleanup_allowed"): + report["cleanup_skipped_reason"] = ( + assessment.get("reasons") or ["cleanup not allowed"] + ) + return report + + comment_block = _profile_operation_gate("gitea.pr.comment") + if comment_block: + report["success"] = False + report["reasons"] = comment_block + report["permission_report"] = _permission_block_report("gitea.pr.comment") + return report + + verify_preflight_purity(remote) + body = assessment["release_body"] + comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments" + with _audited( + "comment_pr", + host=h, + remote=remote, + org=o, + repo=r, + pr_number=pr_number, + request_metadata={"source": "cleanup_post_merge_moot_lease"}, + ): + posted = api_request("POST", comment_url, auth, {"body": body}) + + report["cleanup_performed"] = True + report["released_comment_id"] = posted.get("id") + return report + + @mcp.tool() def gitea_list_issue_comments( issue_number: int, diff --git a/reviewer_pr_lease.py b/reviewer_pr_lease.py index 6e28be5..da1169e 100644 --- a/reviewer_pr_lease.py +++ b/reviewer_pr_lease.py @@ -217,12 +217,24 @@ def assess_acquire_lease( candidate_head: str | None, target_branch: str, target_branch_sha: str | None, + pr_merged_or_closed: bool = False, now: datetime | None = None, ) -> dict[str, Any]: - """Fail closed when another session holds an active lease.""" + """Fail closed when another session holds an active lease. + + When *pr_merged_or_closed* is true the PR has already merged/closed, so any + reviewer-lease acquisition or adoption for merge work is moot: fail closed + with a ``post_merge_moot`` reason and never mint a lease body (#515). + """ now = now or datetime.now(timezone.utc) reasons: list[str] = [] existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now) + post_merge_moot = bool(pr_merged_or_closed) + if post_merge_moot: + reasons.append( + f"post_merge_moot: PR #{pr_number} is already merged/closed; reviewer " + "lease adoption for merge is moot (fail closed)" + ) if existing: owner_session = (existing.get("session_id") or "").strip() freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now) @@ -270,6 +282,101 @@ def assess_acquire_lease( "existing_lease": existing, "lease_body": body, "session_id": session_id, + "post_merge_moot": post_merge_moot, + } + + +def assess_post_merge_moot_lease( + comments: list[dict], + *, + pr_number: int, + pr_merged: bool = False, + pr_state: str | None = None, + merge_commit_sha: str | None = None, + now: datetime | None = None, +) -> dict[str, Any]: + """Assess a reviewer lease left lingering on an already-merged/closed PR (#515). + + Read-first and fail-safe: + + - Only treats a lease as moot when the live PR state is merged/closed. + - Never proposes touching an *active* lease while the PR is still open + (``cleanup_allowed`` stays false and a refusal reason is returned). + - When the PR is merged/closed and a lease is still active, ``cleanup_allowed`` + is true and a terminal ``phase: released`` lease body (``blocker: + post-merge-moot``) is provided so the moot lease can be neutralised by an + append-only comment — never by deleting a foreign session's comment, and + never by adopting or merging. + + Posting the released body makes that lease terminal, so a subsequent call + finds no active lease and reports nothing left to clean (idempotent). + """ + now = now or datetime.now(timezone.utc) + merged_or_closed = bool(pr_merged) or ( + str(pr_state or "").strip().lower() == "closed" + ) + active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now) + # The newest lease comment is authoritative: once a terminal marker + # (released/done/blocked) is the latest entry, the lease is resolved even if + # an earlier non-terminal comment from the same session still lingers. This + # keeps post-merge cleanup idempotent. + entries = _lease_entries(comments, pr_number=pr_number) + newest = entries[-1] if entries else None + newest_terminal = bool(newest) and ( + (newest.get("phase") or "").strip().lower() in _TERMINAL_PHASES + ) + reasons: list[str] = [] + cleanup_allowed = False + release_body: str | None = None + is_moot = bool(active) and merged_or_closed and not newest_terminal + + if not merged_or_closed: + if active: + reasons.append( + f"PR #{pr_number} is still open; refusing to touch active reviewer " + "lease (fail closed)" + ) + else: + reasons.append( + f"PR #{pr_number} is still open; no post-merge lease cleanup applicable" + ) + elif newest_terminal: + reasons.append( + f"PR #{pr_number} reviewer lease already released/terminal; nothing to clean" + ) + elif active: + cleanup_allowed = True + release_body = format_lease_body( + repo=active.get("repo") or "", + pr_number=pr_number, + issue_number=active.get("issue_number"), + reviewer_identity=active.get("reviewer_identity") or "", + profile=active.get("profile") or "unknown", + session_id=active.get("session_id") or "", + worktree=active.get("worktree") or "", + phase="released", + candidate_head=active.get("candidate_head"), + target_branch=active.get("target_branch") or "master", + target_branch_sha=active.get("target_branch_sha"), + last_activity=now, + blocker="post-merge-moot", + ) + else: + reasons.append( + f"PR #{pr_number} is merged/closed but no active reviewer lease remains; " + "nothing to clean" + ) + + return { + "pr_number": pr_number, + "pr_state": pr_state, + "pr_merged_or_closed": merged_or_closed, + "merge_commit_sha": merge_commit_sha, + "active_lease": active, + "is_moot": is_moot, + "cleanup_allowed": cleanup_allowed, + "release_body": release_body, + "reasons": reasons, } diff --git a/tests/test_post_merge_moot_lease.py b/tests/test_post_merge_moot_lease.py new file mode 100644 index 0000000..a71cb72 --- /dev/null +++ b/tests/test_post_merge_moot_lease.py @@ -0,0 +1,257 @@ +"""Post-merge moot reviewer-lease handling (#515). + +Covers: + a. Reviewer-lease acquisition/adoption is refused on an already-merged/closed + PR (fail closed, no mutation). + b. The post-merge moot cleanup path is safe and idempotent. + c. An active foreign lease on an *open* PR is never force-cleaned. +""" + +import os +import sys +import unittest +from datetime import datetime, timezone +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import reviewer_pr_lease as leases # noqa: E402 +from mcp_server import ( # noqa: E402 + gitea_acquire_reviewer_pr_lease, + gitea_cleanup_post_merge_moot_lease, +) + +FAKE_AUTH = "Basic dGVzdDp0ZXN0" +MERGER_ENV = { + "GITEA_PROFILE_NAME": "prgs-merger", + "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment", +} +PR = 487 +ISSUE = 485 +SESSION = "97274-676d20a825c4" + + +def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed", + candidate_head="a" * 40): + body = leases.format_lease_body( + repo="Scaled-Tech-Consulting/Gitea-Tools", + pr_number=pr_number, + issue_number=ISSUE, + reviewer_identity="sysadmin", + profile="prgs-reviewer", + session_id=session_id, + worktree="branches/review-pr487", + phase=phase, + candidate_head=candidate_head, + target_branch="master", + target_branch_sha="b" * 40, + last_activity=datetime.now(timezone.utc), + ) + return {"id": 6603, "body": body, "user": {"login": "sysadmin"}} + + +# --------------------------------------------------------------------------- # +# Pure logic +# --------------------------------------------------------------------------- # +class TestAcquireRefusedOnMergedPR(unittest.TestCase): + def test_acquire_refused_when_pr_merged_or_closed(self): + result = leases.assess_acquire_lease( + [_lease_comment()], + pr_number=PR, + reviewer_identity="sysadmin", + profile="prgs-merger", + session_id="new-session", + repo="Scaled-Tech-Consulting/Gitea-Tools", + issue_number=ISSUE, + worktree="branches/merge-pr487", + candidate_head="a" * 40, + target_branch="master", + target_branch_sha="b" * 40, + pr_merged_or_closed=True, + ) + self.assertFalse(result["acquire_allowed"]) + self.assertTrue(result["post_merge_moot"]) + self.assertIsNone(result["lease_body"]) + self.assertTrue(any("post_merge_moot" in r for r in result["reasons"])) + + def test_acquire_still_allowed_on_open_pr_without_flag(self): + result = leases.assess_acquire_lease( + [], + pr_number=PR, + reviewer_identity="sysadmin", + profile="prgs-reviewer", + session_id="s1", + repo="Scaled-Tech-Consulting/Gitea-Tools", + issue_number=ISSUE, + worktree="branches/review-pr487", + candidate_head="a" * 40, + target_branch="master", + target_branch_sha="b" * 40, + ) + self.assertTrue(result["acquire_allowed"]) + self.assertFalse(result["post_merge_moot"]) + + +class TestPostMergeMootAssessment(unittest.TestCase): + def test_merged_pr_with_active_lease_is_moot_and_cleanable(self): + a = leases.assess_post_merge_moot_lease( + [_lease_comment()], + pr_number=PR, + pr_merged=True, + pr_state="closed", + merge_commit_sha="c" * 40, + ) + self.assertTrue(a["pr_merged_or_closed"]) + self.assertTrue(a["is_moot"]) + self.assertTrue(a["cleanup_allowed"]) + self.assertIsNotNone(a["release_body"]) + self.assertIn("phase: released", a["release_body"]) + self.assertIn("blocker: post-merge-moot", a["release_body"]) + + def test_open_pr_active_lease_never_cleaned(self): + a = leases.assess_post_merge_moot_lease( + [_lease_comment()], + pr_number=PR, + pr_merged=False, + pr_state="open", + ) + self.assertFalse(a["pr_merged_or_closed"]) + self.assertFalse(a["is_moot"]) + self.assertFalse(a["cleanup_allowed"]) + self.assertIsNone(a["release_body"]) + self.assertTrue(any("still open" in r for r in a["reasons"])) + + def test_merged_pr_without_lease_nothing_to_clean(self): + a = leases.assess_post_merge_moot_lease( + [], pr_number=PR, pr_merged=True, pr_state="closed") + self.assertTrue(a["pr_merged_or_closed"]) + self.assertFalse(a["is_moot"]) + self.assertFalse(a["cleanup_allowed"]) + self.assertTrue(any("nothing to clean" in r for r in a["reasons"])) + + def test_cleanup_is_idempotent(self): + """After the released marker is posted, a re-assess finds nothing to clean.""" + first = leases.assess_post_merge_moot_lease( + [_lease_comment()], pr_number=PR, pr_merged=True, pr_state="closed") + self.assertTrue(first["cleanup_allowed"]) + released_comment = { + "id": 7000, "body": first["release_body"], "user": {"login": "sysadmin"}} + second = leases.assess_post_merge_moot_lease( + [_lease_comment(), released_comment], + pr_number=PR, pr_merged=True, pr_state="closed") + self.assertFalse(second["is_moot"]) + self.assertFalse(second["cleanup_allowed"]) + + +# --------------------------------------------------------------------------- # +# Server tools +# --------------------------------------------------------------------------- # +def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999): + """Build an api_request side effect keyed on method + url.""" + calls = {"post": []} + + def _side(method, url, auth=None, payload=None, *a, **k): + m = (method or "").upper() + if m == "POST": + calls["post"].append({"url": url, "payload": payload}) + return {"id": posted_id} + if "/comments" in url: + return list(comments) + if "/pulls/" in url: + pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40} + if pr_merged: + pr["merged"] = True + pr["merged_at"] = "2026-07-08T07:46:04Z" + return pr + if "/issues/" in url: + return {"state": "closed" if pr_merged else "open", "number": ISSUE} + return {} + + return _side, calls + + +class TestAcquireToolRefusesMergedPR(unittest.TestCase): + def setUp(self): + leases.clear_session_lease() + + @patch("mcp_server.verify_preflight_purity", return_value=None) + @patch("mcp_server._authenticated_username", return_value="sysadmin") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.api_request") + def test_acquire_tool_fails_closed_on_merged_pr_without_posting( + self, mock_api, _auth, _user, _purity): + side, calls = _api_side_effect( + pr_state="closed", pr_merged=True, comments=[]) + mock_api.side_effect = side + with patch.dict(os.environ, MERGER_ENV, clear=True): + result = gitea_acquire_reviewer_pr_lease( + pr_number=PR, + worktree="branches/merge-pr487", + candidate_head="a" * 40, + issue_number=ISSUE, + remote="prgs", + ) + self.assertFalse(result["success"]) + self.assertFalse(result["acquired"]) + self.assertTrue(result.get("post_merge_moot")) + self.assertEqual(calls["post"], [], "must not post a lease comment") + + +class TestCleanupTool(unittest.TestCase): + def setUp(self): + leases.clear_session_lease() + + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.api_request") + def test_read_only_reports_moot_without_mutating(self, mock_api, _auth): + side, calls = _api_side_effect( + pr_state="closed", pr_merged=True, comments=[_lease_comment()]) + mock_api.side_effect = side + with patch.dict(os.environ, MERGER_ENV, clear=True): + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=False, remote="prgs") + self.assertTrue(result["success"]) + self.assertTrue(result["pr_merged_or_closed"]) + self.assertTrue(result["lease_moot"]) + self.assertFalse(result["cleanup_performed"]) + self.assertTrue(result["no_merge_or_adoption"]) + self.assertEqual(result["mode"], "read_only") + self.assertEqual(calls["post"], []) + + @patch("mcp_server.verify_preflight_purity", return_value=None) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.api_request") + def test_apply_posts_released_marker_on_merged_pr( + self, mock_api, _auth, _purity): + side, calls = _api_side_effect( + pr_state="closed", pr_merged=True, comments=[_lease_comment()]) + mock_api.side_effect = side + with patch.dict(os.environ, MERGER_ENV, clear=True): + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=True, remote="prgs") + self.assertTrue(result["success"]) + self.assertTrue(result["cleanup_performed"]) + self.assertEqual(result["released_comment_id"], 9999) + self.assertEqual(len(calls["post"]), 1) + self.assertIn("phase: released", calls["post"][0]["payload"]["body"]) + self.assertIn("post-merge-moot", calls["post"][0]["payload"]["body"]) + + @patch("mcp_server.verify_preflight_purity", return_value=None) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.api_request") + def test_apply_refuses_to_clean_open_pr(self, mock_api, _auth, _purity): + side, calls = _api_side_effect( + pr_state="open", pr_merged=False, comments=[_lease_comment()]) + mock_api.side_effect = side + with patch.dict(os.environ, MERGER_ENV, clear=True): + result = gitea_cleanup_post_merge_moot_lease( + pr_number=PR, apply=True, remote="prgs") + self.assertFalse(result["cleanup_performed"]) + self.assertFalse(result["pr_merged_or_closed"]) + self.assertEqual(calls["post"], [], "never force-clean an open PR lease") + self.assertTrue( + any("still open" in r for r in result.get("cleanup_skipped_reason", []))) + + +if __name__ == "__main__": + unittest.main()