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

feat: gate reviewer mutations on canonical workflow load proof (Closes #389)

Browse Source 
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.
This commit is contained in:
sysadmin
2026-07-07 12:54:10 -04:00
parent 1a1e679246
commit 2aae877637
6 changed files with 445 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+78 -1
View File
@@ -498,6 +498,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 already_landed_reconcile # noqa: E402
@@ -1822,6 +1823,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 = (
@@ -1847,6 +1849,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,
@@ -1857,7 +1864,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(
@@ -2213,6 +2223,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,
@@ -2228,6 +2239,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(
@@ -2412,6 +2427,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}
@@ -2995,6 +3017,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,
@@ -3012,6 +3035,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:
@@ -4748,6 +4775,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): "
@@ -5671,6 +5700,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:
@@ -5679,6 +5710,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.
@@ -6734,6 +6801,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)
review_workflow_load.py
+198
View File
@@ -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.",
]
tests/test_llm_agent_sha.py
+2
View File
@@ -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):
tests/test_mcp_server.py
+14 -5
View File
@@ -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,
@@ -1653,7 +1661,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, {
@@ -1748,7 +1756,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):
@@ -2141,7 +2149,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"]))
@@ -2223,6 +2231,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()
tests/test_permission_reports.py
+2
View File
@@ -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(
tests/test_review_workflow_load.py
+151
View File
@@ -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()