From 8d32af14d6e67243c670bdd3b4a4155fde40a98a Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 09:14:29 -0400 Subject: [PATCH] feat: enforce PR review/merge workflow state machine gates (Closes #290) Add fail-closed review/merge state progression helpers, pre-merge gate requirements, blocked recovery handoff checks, and MCP assessment tool. Closes #290 --- gitea_mcp_server.py | 69 +++++ review_merge_state_machine.py | 332 +++++++++++++++++++++++ tests/test_review_merge_state_machine.py | 121 +++++++++ 3 files changed, 522 insertions(+) create mode 100644 review_merge_state_machine.py create mode 100644 tests/test_review_merge_state_machine.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index b5415eb..9e8a6e8 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -477,6 +477,7 @@ import review_proofs # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 import merged_cleanup_reconcile # noqa: E402 +import review_merge_state_machine # noqa: E402 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue, @@ -4082,6 +4083,14 @@ _GUIDE_RULES = { "inventory, summarize) needs no explicit authorization. Enforced " "fail closed via subagent_gate.assess_subagent_delegation and " "validate_subagent_report (#266)."), + "review_merge_state_machine": ( + "PR review/merge must follow the enforced state machine: PRECHECK → " + "INVENTORY → SELECT_PR → PIN_HEAD_SHA → CREATE_BRANCHES_WORKTREE → " + "VALIDATE → REVIEW_DECISION → APPROVE_OR_REQUEST_CHANGES → " + "PRE_MERGE_RECHECK → MERGE → POST_MERGE_REPORT. Failed upstream " + "states forbid downstream approve/merge. infra_stop and capability " + "blocks stop immediately. Blocked recovery handoffs must restart the " + "full workflow — never replay approve/merge commands (#290)."), } _COMMON_WORKFLOWS = [ @@ -4786,6 +4795,66 @@ def _build_runtime_task_capabilities( } +@mcp.tool() +def gitea_assess_review_merge_state_machine( + target_state: str | None = None, + state_completion: dict[str, bool] | None = None, + pre_merge_gates: dict[str, bool] | None = None, + infra_stop: bool = False, + capability_blocked: bool = False, + recovery_handoff_text: str | None = None, + final_report_text: str | None = None, +) -> dict: + """Read-only: assess enforced PR review/merge workflow state (#290).""" + completion = state_completion or {} + blockers = review_merge_state_machine.assess_workflow_blockers( + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ) + result = { + "workflow": review_merge_state_machine.workflow_status( + completion, + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ), + "blockers": blockers, + "approve": review_merge_state_machine.can_approve( + completion, + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ), + "merge": review_merge_state_machine.can_merge( + completion, + pre_merge_gates=pre_merge_gates, + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ), + } + if target_state: + result["advancement"] = review_merge_state_machine.assess_state_advancement( + completion, + target_state=target_state, + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ) + if recovery_handoff_text is not None: + result["recovery_handoff"] = ( + review_merge_state_machine.assess_blocked_recovery_handoff( + recovery_handoff_text + ) + ) + if final_report_text is not None: + result["final_report_claims"] = ( + review_merge_state_machine.assess_final_report_state_claims( + final_report_text, + state_completion=completion, + approve_completed=bool(completion.get("APPROVE_OR_REQUEST_CHANGES")), + merge_completed=bool(completion.get("MERGE")), + ) + ) + return result + + @mcp.tool() def gitea_get_runtime_context( remote: str = "dadeschools", diff --git a/review_merge_state_machine.py b/review_merge_state_machine.py new file mode 100644 index 0000000..33c9705 --- /dev/null +++ b/review_merge_state_machine.py @@ -0,0 +1,332 @@ +"""Enforced PR review/merge workflow state machine (#290). + +Review and merge must advance through explicit states. Any failed upstream +gate forbids downstream approve/merge mutations until the full workflow is +restarted after blockers clear. +""" + +from __future__ import annotations + +import re +from typing import Any + +REVIEW_MERGE_STATES: tuple[str, ...] = ( + "PRECHECK", + "INVENTORY", + "SELECT_PR", + "PIN_HEAD_SHA", + "CREATE_BRANCHES_WORKTREE", + "VALIDATE", + "REVIEW_DECISION", + "APPROVE_OR_REQUEST_CHANGES", + "PRE_MERGE_RECHECK", + "MERGE", + "POST_MERGE_REPORT", +) + +TERMINAL_BLOCKED_HEADING = ( + "PR review/merge workflow blocked. Restart the full workflow after blockers clear." +) + +_FORBIDDEN_RECOVERY_REPLAY_RE = re.compile( + r"\b(?:approve(?:\s+pr)?\s*#?\d+|merge(?:\s+pr)?\s*#?\d+|gitea_merge_pr|" + r"gitea_submit_pr_review|submit\s+approve|run\s+merge)\b", + re.IGNORECASE, +) +_RESTART_WORKFLOW_RE = re.compile( + r"restart(?:\s+the)?\s+full\s+workflow|rerun\s+the\s+full\s+workflow", + re.IGNORECASE, +) +_READY_TO_MERGE_RE = re.compile( + r"\bready\s+to\s+merge\b", + re.IGNORECASE, +) +_REVIEWED_VALIDATED_RE = re.compile( + r"\b(?:reviewed|validated|approval\s+submitted)\b", + re.IGNORECASE, +) + +_PRE_MERGE_REQUIRED_GATES = ( + "whoami_verified", + "profile_runtime_verified", + "merge_capability_verified", + "pr_refetched", + "reviewed_head_sha_unchanged", + "pr_mergeable", + "checks_passed", + "reviewer_not_author", + "worktree_clean", +) + + +def _clean(value: str | None) -> str: + return (value or "").strip() + + +def state_index(state: str) -> int: + name = _clean(state).upper() + try: + return REVIEW_MERGE_STATES.index(name) + except ValueError as exc: + raise ValueError(f"unknown review/merge state '{state}' (fail closed)") from exc + + +def downstream_states(from_state: str) -> list[str]: + idx = state_index(from_state) + return list(REVIEW_MERGE_STATES[idx + 1 :]) + + +def _completed_through(state_completion: dict[str, bool], state: str) -> bool: + return bool(state_completion.get(state)) + + +def _first_incomplete_state(state_completion: dict[str, bool]) -> str | None: + for state in REVIEW_MERGE_STATES: + if not _completed_through(state_completion, state): + return state + return None + + +def assess_workflow_blockers( + *, + infra_stop: bool = False, + capability_blocked: bool = False, + mcp_reconnect_failed: bool = False, + stale_capability_state: bool = False, +) -> dict[str, Any]: + """Return hard blockers that forbid all PR queue work (#290 AC3).""" + reasons: list[str] = [] + if infra_stop: + reasons.append("infra_stop is active; PR selection/review/merge is forbidden") + if capability_blocked: + reasons.append("gitea_resolve_task_capability returned blocked/stop_required") + if mcp_reconnect_failed: + reasons.append("MCP reconnect failed; stale session state cannot be reused") + if stale_capability_state: + reasons.append("stale MCP capability state detected after reconnect failure") + return { + "block": bool(reasons), + "reasons": reasons, + "forbidden_states": list(REVIEW_MERGE_STATES) if reasons else [], + "safe_next_action": ( + TERMINAL_BLOCKED_HEADING if reasons else "proceed with PRECHECK" + ), + } + + +def assess_state_advancement( + state_completion: dict[str, bool] | None, + *, + target_state: str, + infra_stop: bool = False, + capability_blocked: bool = False, +) -> dict[str, Any]: + """Fail closed when *target_state* is requested before upstream gates pass.""" + completion = dict(state_completion or {}) + target = _clean(target_state).upper() + blockers = assess_workflow_blockers( + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ) + reasons = list(blockers["reasons"]) + + try: + target_idx = state_index(target) + except ValueError as exc: + return { + "target_state": target, + "allowed": False, + "block": True, + "reasons": [str(exc)], + "next_allowed_state": None, + "safe_next_action": TERMINAL_BLOCKED_HEADING, + } + + if blockers["block"]: + return { + "target_state": target, + "allowed": False, + "block": True, + "reasons": reasons, + "next_allowed_state": None, + "safe_next_action": blockers["safe_next_action"], + } + + next_allowed = _first_incomplete_state(completion) + if next_allowed is None: + allowed = target_idx == len(REVIEW_MERGE_STATES) - 1 + if not allowed: + reasons.append("workflow already completed through POST_MERGE_REPORT") + else: + allowed = state_index(next_allowed) >= target_idx + if not allowed: + reasons.append( + f"state '{target}' is forbidden until '{next_allowed}' completes" + ) + + return { + "target_state": target, + "allowed": allowed and not reasons, + "block": bool(reasons), + "reasons": reasons, + "next_allowed_state": next_allowed, + "completed_states": [s for s in REVIEW_MERGE_STATES if completion.get(s)], + "safe_next_action": ( + f"complete state '{next_allowed}' before advancing" + if next_allowed and not allowed + else ( + TERMINAL_BLOCKED_HEADING + if reasons + else f"advance to {target}" + ) + ), + } + + +def can_approve(state_completion: dict[str, bool] | None, **blocker_kwargs) -> dict[str, Any]: + """Approval requires all states through REVIEW_DECISION (#290 AC).""" + required = REVIEW_MERGE_STATES[: REVIEW_MERGE_STATES.index("REVIEW_DECISION") + 1] + completion = dict(state_completion or {}) + advance = assess_state_advancement( + completion, + target_state="APPROVE_OR_REQUEST_CHANGES", + **blocker_kwargs, + ) + missing = [s for s in required if not completion.get(s)] + reasons = list(advance["reasons"]) + if missing: + reasons.append( + "approval blocked: incomplete states: " + ", ".join(missing) + ) + block = bool(reasons) or advance["block"] + return { + "allowed": not block, + "block": block, + "reasons": reasons, + "missing_states": missing, + "safe_next_action": advance["safe_next_action"], + } + + +def can_merge( + state_completion: dict[str, bool] | None, + *, + pre_merge_gates: dict[str, bool] | None = None, + **blocker_kwargs, +) -> dict[str, Any]: + """Merge requires approve path plus fresh PRE_MERGE_RECHECK gates (#290 AC6).""" + completion = dict(state_completion or {}) + approve = can_approve(completion, **blocker_kwargs) + reasons = list(approve["reasons"]) + + if not completion.get("APPROVE_OR_REQUEST_CHANGES"): + reasons.append("merge blocked: APPROVE_OR_REQUEST_CHANGES not completed") + + gate_map = dict(pre_merge_gates or {}) + missing_gates = [g for g in _PRE_MERGE_REQUIRED_GATES if not gate_map.get(g)] + if missing_gates: + reasons.append( + "merge blocked: pre-merge gates incomplete: " + ", ".join(missing_gates) + ) + + advance = assess_state_advancement( + completion, + target_state="MERGE", + **blocker_kwargs, + ) + reasons.extend(advance["reasons"]) + + block = bool(reasons) + return { + "allowed": not block, + "block": block, + "reasons": reasons, + "missing_pre_merge_gates": missing_gates, + "safe_next_action": ( + "complete PRE_MERGE_RECHECK with fresh whoami/capability/PR re-fetch " + "before merge" + if block + else "merge allowed" + ), + } + + +def assess_blocked_recovery_handoff(report_text: str) -> dict[str, Any]: + """Blocked handoffs must not replay approve/merge commands (#290 AC4).""" + text = report_text or "" + reasons: list[str] = [] + if _FORBIDDEN_RECOVERY_REPLAY_RE.search(text): + reasons.append( + "blocked recovery handoff contains direct approve/merge replay command" + ) + if reasons and not _RESTART_WORKFLOW_RE.search(text): + reasons.append( + "blocked recovery handoff must direct operator to restart full workflow" + ) + return { + "block": bool(reasons), + "allowed": not reasons, + "reasons": reasons, + "safe_next_action": ( + "remove approve/merge replay commands and say to restart full workflow " + "after blockers clear" + if reasons + else "recovery handoff wording acceptable" + ), + } + + +def assess_final_report_state_claims( + report_text: str, + *, + state_completion: dict[str, bool] | None = None, + approve_completed: bool = False, + merge_completed: bool = False, +) -> dict[str, Any]: + """Reports must not claim reviewed/ready-to-merge without gate proof (#290 AC10).""" + text = report_text or "" + completion = dict(state_completion or {}) + reasons: list[str] = [] + + if _READY_TO_MERGE_RE.search(text) and not ( + merge_completed or completion.get("PRE_MERGE_RECHECK") + ): + reasons.append( + "report claims ready-to-merge without PRE_MERGE_RECHECK completion" + ) + + if _REVIEWED_VALIDATED_RE.search(text): + if not (approve_completed or completion.get("VALIDATE")): + reasons.append( + "report claims reviewed/validated without VALIDATE/APPROVE proof" + ) + + return { + "block": bool(reasons), + "allowed": not reasons, + "reasons": reasons, + "safe_next_action": ( + "remove stale reviewed/ready-to-merge claims unless gates passed" + if reasons + else "final report state claims consistent with workflow" + ), + } + + +def workflow_status( + state_completion: dict[str, bool] | None, + **blocker_kwargs, +) -> dict[str, Any]: + """Summarize current state-machine position for MCP/runtime reporting.""" + completion = dict(state_completion or {}) + blockers = assess_workflow_blockers(**blocker_kwargs) + next_state = _first_incomplete_state(completion) + return { + "states": list(REVIEW_MERGE_STATES), + "completed_states": [s for s in REVIEW_MERGE_STATES if completion.get(s)], + "next_required_state": next_state, + "workflow_complete": next_state is None, + "approve_allowed": can_approve(completion, **blocker_kwargs)["allowed"], + "merge_allowed": can_merge(completion, **blocker_kwargs)["allowed"], + "blockers": blockers, + } \ No newline at end of file diff --git a/tests/test_review_merge_state_machine.py b/tests/test_review_merge_state_machine.py new file mode 100644 index 0000000..b3930d6 --- /dev/null +++ b/tests/test_review_merge_state_machine.py @@ -0,0 +1,121 @@ +"""Tests for enforced PR review/merge state machine (#290).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import review_merge_state_machine as rmsm # noqa: E402 + +FULL_REVIEW_COMPLETION = {state: True for state in rmsm.REVIEW_MERGE_STATES} + + +def _completion_through(state_name: str) -> dict[str, bool]: + idx = rmsm.state_index(state_name) + return {state: i <= idx for i, state in enumerate(rmsm.REVIEW_MERGE_STATES)} + + +class TestWorkflowBlockers(unittest.TestCase): + def test_infra_stop_blocks_all_states(self): + result = rmsm.assess_workflow_blockers(infra_stop=True) + self.assertTrue(result["block"]) + self.assertEqual(result["forbidden_states"], list(rmsm.REVIEW_MERGE_STATES)) + + def test_infra_stop_forbids_advancement(self): + result = rmsm.assess_state_advancement( + {}, + target_state="INVENTORY", + infra_stop=True, + ) + self.assertTrue(result["block"]) + self.assertFalse(result["allowed"]) + + +class TestStateAdvancement(unittest.TestCase): + def test_cannot_skip_inventory(self): + result = rmsm.assess_state_advancement( + {"PRECHECK": True}, + target_state="SELECT_PR", + ) + self.assertTrue(result["block"]) + self.assertEqual(result["next_allowed_state"], "INVENTORY") + + def test_sequential_advance_allowed(self): + completion = _completion_through("INVENTORY") + result = rmsm.assess_state_advancement( + completion, + target_state="SELECT_PR", + ) + self.assertFalse(result["block"]) + self.assertTrue(result["allowed"]) + + +class TestApproveAndMergeGates(unittest.TestCase): + def test_approve_blocked_without_validate(self): + completion = _completion_through("PIN_HEAD_SHA") + result = rmsm.can_approve(completion) + self.assertTrue(result["block"]) + self.assertIn("VALIDATE", ",".join(result["missing_states"])) + + def test_approve_allowed_when_review_path_complete(self): + completion = _completion_through("REVIEW_DECISION") + result = rmsm.can_approve(completion) + self.assertFalse(result["block"]) + self.assertTrue(result["allowed"]) + + def test_merge_blocked_without_pre_merge_gates(self): + completion = _completion_through("APPROVE_OR_REQUEST_CHANGES") + result = rmsm.can_merge(completion) + self.assertTrue(result["block"]) + self.assertTrue(result["missing_pre_merge_gates"]) + + def test_merge_allowed_with_pre_merge_gates(self): + completion = _completion_through("PRE_MERGE_RECHECK") + gates = {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES} + result = rmsm.can_merge(completion, pre_merge_gates=gates) + self.assertFalse(result["block"]) + self.assertTrue(result["allowed"]) + + def test_merge_blocked_when_head_sha_gate_fails(self): + completion = _completion_through("PRE_MERGE_RECHECK") + gates = {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES} + gates["reviewed_head_sha_unchanged"] = False + result = rmsm.can_merge(completion, pre_merge_gates=gates) + self.assertTrue(result["block"]) + self.assertIn("reviewed_head_sha_unchanged", result["missing_pre_merge_gates"]) + + +class TestRecoveryHandoff(unittest.TestCase): + def test_blocked_handoff_without_replay_passes(self): + report = ( + "Blocked recovery handoff.\n" + "Restart the full workflow after infra_stop clears.\n" + "No approve/merge performed." + ) + result = rmsm.assess_blocked_recovery_handoff(report) + self.assertFalse(result["block"]) + + def test_blocked_handoff_with_merge_replay_fails(self): + report = "Please merge PR #12 now using gitea_merge_pr." + result = rmsm.assess_blocked_recovery_handoff(report) + self.assertTrue(result["block"]) + + +class TestFinalReportClaims(unittest.TestCase): + def test_ready_to_merge_claim_without_gates_fails(self): + result = rmsm.assess_final_report_state_claims( + "PR is ready to merge after validation.", + state_completion=_completion_through("VALIDATE"), + ) + self.assertTrue(result["block"]) + + def test_reviewed_claim_without_validate_fails(self): + result = rmsm.assess_final_report_state_claims( + "PR reviewed and looks good.", + state_completion={"PRECHECK": True}, + ) + self.assertTrue(result["block"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file