diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 17938e3..f784331 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -536,6 +536,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 @@ -2010,6 +2011,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 = ( @@ -2035,6 +2037,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, @@ -2045,7 +2052,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( @@ -2401,6 +2411,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, @@ -2416,6 +2427,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( @@ -2600,6 +2615,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} @@ -3197,6 +3219,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, @@ -3214,6 +3237,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: @@ -4950,6 +4977,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): " @@ -5873,6 +5902,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: @@ -5881,6 +5912,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. @@ -6936,6 +7003,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 aabb7e7..17afc36 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'.""" @@ -539,6 +545,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}, @@ -1011,8 +1020,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, @@ -1680,7 +1688,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, { @@ -1775,7 +1783,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): @@ -2168,7 +2176,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"])) @@ -2250,6 +2258,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