diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index ca164e0..c131813 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -714,6 +714,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_lock_provenance # noqa: E402 @@ -2384,6 +2385,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 = ( @@ -2410,6 +2412,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, @@ -2420,7 +2427,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( @@ -2835,6 +2845,7 @@ def _evaluate_pr_review_submission( remote, worktree_path=worktree_path, task="review_pr" ) action = (action or "").strip().lower() + workflow_blockers = _review_workflow_load_gate_reasons() if live else [] result = { "requested_action": action, "performed": False, @@ -2850,6 +2861,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( @@ -3068,6 +3083,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} @@ -3703,6 +3725,7 @@ def gitea_merge_pr( _verify_role_mutation_workspace( remote, worktree_path=worktree_path, task="merge_pr" ) + workflow_blockers = _review_workflow_load_gate_reasons() do = (do or "").strip().lower() result = { "performed": False, @@ -3720,6 +3743,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: @@ -5890,6 +5917,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): " @@ -6813,6 +6842,8 @@ def gitea_get_runtime_context( ), "role_kind": _role_kind(allowed, forbidden), "shell_health": native_mcp_preference.shell_health_status(), + "workflow_load_proof": review_workflow_load.workflow_load_status( + PROJECT_ROOT), } if reveal and h: @@ -6821,6 +6852,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. @@ -7989,6 +8056,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_audit.py b/tests/test_audit.py index b233b15..2d3a1e9 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): @@ -291,12 +292,11 @@ class TestGatedToolAudit(_AuditWiringBase): def setUp(self): super().setUp() - from mcp_server import init_review_decision_lock - from tests.test_mcp_server import _install_owned_reviewer_lease + from tests.test_mcp_server import _init_reviewer_session, _install_owned_reviewer_lease import reviewer_pr_lease - # init_review_decision_lock clears any prior session lease (#407). - init_review_decision_lock("prgs", "review_pr") + # Session init clears any prior session lease (#407) and loads workflow (#389). + _init_reviewer_session("prgs") self._lease_patch = _install_owned_reviewer_lease(8) self._lease_patch.start() self._auth_identity_patch = patch( @@ -337,6 +337,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"]) @@ -356,6 +357,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() diff --git a/tests/test_llm_agent_sha.py b/tests/test_llm_agent_sha.py index 4d74dfa..2751d9d 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, diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 179159d..d5a5035 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -58,6 +58,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'.""" @@ -179,6 +185,7 @@ def _seed_ready_review_decision( "correction_authorized": False, "correction_reason": None, }) + _m.gitea_load_review_workflow() # Issue-write tools are profile-gated (#69). @@ -671,6 +678,7 @@ class TestMergePR(unittest.TestCase): def setUp(self): import reviewer_pr_lease + mcp_server.gitea_load_review_workflow() self._lease_patch = _install_owned_reviewer_lease(8) self._lease_patch.start() self._auth_identity_patch = patch( @@ -1876,7 +1884,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase): def setUp(self): import reviewer_pr_lease - init_review_decision_lock("prgs", "review_pr") + _init_reviewer_session("prgs") self._lease_patch = _install_owned_reviewer_lease( self.PR, head_sha=self.SHA, ) @@ -1996,7 +2004,7 @@ class TestSubmitPrReview(unittest.TestCase): def setUp(self): import reviewer_pr_lease - init_review_decision_lock("prgs", "review_pr") + _init_reviewer_session("prgs") self._lease_patch = _install_owned_reviewer_lease(8) self._lease_patch.start() self._auth_identity_patch = patch( @@ -2416,7 +2424,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", expected_head_sha="abc123") self.assertFalse(r["marked_ready"]) self.assertTrue(any("does not match locked remote" in x for x in r["reasons"])) @@ -2500,6 +2508,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 279dc2b..2978af5 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() @@ -250,6 +251,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 a61cc3f..ddd9902 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 diff --git a/tests/test_review_workflow_load.py b/tests/test_review_workflow_load.py new file mode 100644 index 0000000..e59bf1a --- /dev/null +++ b/tests/test_review_workflow_load.py @@ -0,0 +1,153 @@ +"""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.get_auth_header", return_value="Basic dGVzdA=="), \ + patch("mcp_server._list_pr_lease_comments", return_value=[]), \ + 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 diff --git a/tests/test_terminal_review_hard_stop.py b/tests/test_terminal_review_hard_stop.py index 4c65130..2a7b45a 100644 --- a/tests/test_terminal_review_hard_stop.py +++ b/tests/test_terminal_review_hard_stop.py @@ -37,6 +37,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, @@ -48,6 +49,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) @@ -93,6 +95,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]) @@ -114,6 +117,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]) @@ -146,6 +150,7 @@ def _mark(action, pr_number=6, **kwargs): 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()