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

feat(review-workflow): enforce single terminal review decision per review run (#211)

Browse Source
This commit is contained in:
sysadmin
2026-07-05 17:43:50 -04:00
parent 5d0845df90
commit 7489107768
8 changed files with 221 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp_server.py
+72 -18
View File
@@ -925,24 +925,17 @@ _REVIEW_ACTIONS = {
_TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
REVIEW_DECISION_FILE = "/tmp/gitea_review_decision.lock"
_REVIEW_DECISION_LOCK: dict | None = None
def _load_review_decision_lock():
if not os.path.exists(REVIEW_DECISION_FILE):
return None
try:
with open(REVIEW_DECISION_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return None
global _REVIEW_DECISION_LOCK
return _REVIEW_DECISION_LOCK
def _save_review_decision_lock(data):
try:
with open(REVIEW_DECISION_FILE, "w", encoding="utf-8") as f:
json.dump(data, f)
except Exception:
pass
global _REVIEW_DECISION_LOCK
_REVIEW_DECISION_LOCK = data
def init_review_decision_lock(remote: str | None, task: str | None):
@@ -956,6 +949,9 @@ def init_review_decision_lock(remote: str | None, task: str | None):
"ready_pr_number": None,
"ready_action": None,
"ready_expected_head_sha": None,
"ready_remote": None,
"ready_org": None,
"ready_repo": None,
"live_mutations": [],
"correction_authorized": False,
"correction_reason": None,
@@ -967,6 +963,9 @@ def check_review_decision_gate(
action: str,
*,
final_review_decision_ready: bool,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> list[str]:
"""Fail closed unless validation completed and the final decision is ready."""
reasons = []
@@ -1004,6 +1003,21 @@ def check_review_decision_gate(
f"ready action '{lock.get('ready_action')}' does not match "
f"requested action '{action}' (fail closed)"
)
if lock.get("ready_remote") != remote:
reasons.append(
f"ready remote '{lock.get('ready_remote')}' does not match "
f"requested remote '{remote}' (fail closed)"
)
if lock.get("ready_org") != org:
reasons.append(
f"ready org '{lock.get('ready_org')}' does not match "
f"requested org '{org}' (fail closed)"
)
if lock.get("ready_repo") != repo:
reasons.append(
f"ready repo '{lock.get('ready_repo')}' does not match "
f"requested repo '{repo}' (fail closed)"
)
prior = list(lock.get("live_mutations") or [])
if prior and not lock.get("correction_authorized"):
@@ -1026,10 +1040,15 @@ def check_review_decision_gate(
return reasons
def record_live_review_mutation(pr_number: int, action: str):
def record_live_review_mutation(pr_number: int, action: str, review_id: int | None = None):
lock = _load_review_decision_lock() or {}
mutations = list(lock.get("live_mutations") or [])
mutations.append({"pr_number": pr_number, "action": action})
mutations.append({
"pr_number": pr_number,
"action": action,
"review_id": review_id,
"review_state": action,
})
lock["live_mutations"] = mutations
if lock.get("correction_authorized"):
lock["correction_authorized"] = False
@@ -1245,6 +1264,9 @@ def _evaluate_pr_review_submission(
pr_number,
action,
final_review_decision_ready=final_review_decision_ready,
remote=remote,
org=org,
repo=repo,
))
if reasons:
return result
@@ -1332,6 +1354,8 @@ def gitea_mark_final_review_decision(
action: str,
expected_head_sha: str | None = None,
remote: str = "dadeschools",
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Mark validation complete; the final review decision is ready to submit."""
action = (action or "").strip().lower()
@@ -1363,31 +1387,59 @@ def gitea_mark_final_review_decision(
lock["ready_pr_number"] = pr_number
lock["ready_action"] = action
lock["ready_expected_head_sha"] = expected_head_sha
lock["ready_remote"] = remote
lock["ready_org"] = org
lock["ready_repo"] = repo
_save_review_decision_lock(lock)
return {
"marked_ready": True,
"pr_number": pr_number,
"action": action,
"expected_head_sha": expected_head_sha,
"remote": remote,
"org": org,
"repo": repo,
"final_review_decision_ready": True,
"reasons": [],
}
@mcp.tool()
def gitea_authorize_review_correction(reason: str) -> dict:
def gitea_authorize_review_correction(
prior_review_id: int,
prior_review_state: str,
reason: str,
) -> dict:
"""Authorize one operator-approved correction after a mistaken live review."""
reason = (reason or "").strip()
prior_review_state = (prior_review_state or "").strip().lower()
lock = _load_review_decision_lock()
if lock is None:
return {"authorized": False, "reasons": ["review decision lock missing"]}
if not lock.get("live_mutations"):
prior = list(lock.get("live_mutations") or [])
if not prior:
return {
"authorized": False,
"reasons": ["no prior live review mutation to correct"],
}
last_mutation = prior[-1]
last_review_id = last_mutation.get("review_id")
last_review_state = last_mutation.get("action")
reasons = []
if last_review_id is not None and last_review_id != prior_review_id:
reasons.append(
f"prior review ID '{prior_review_id}' does not match last "
f"recorded review ID '{last_review_id}' (fail closed)"
)
if last_review_state != prior_review_state:
reasons.append(
f"prior review state '{prior_review_state}' does not match last "
f"recorded review state '{last_review_state}' (fail closed)"
)
if not reason:
return {"authorized": False, "reasons": ["correction reason is required"]}
reasons.append("correction reason is required")
if reasons:
return {"authorized": False, "reasons": reasons}
lock["correction_authorized"] = True
lock["correction_reason"] = reason
_save_review_decision_lock(lock)
@@ -1885,6 +1937,7 @@ def gitea_review_pr(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
final_review_decision_ready: bool = False,
) -> dict:
"""Submit a review on a Gitea pull request (Legacy wrapper).
@@ -2059,7 +2112,8 @@ def gitea_review_pr(
remote=remote,
host=host,
org=org,
repo=repo
repo=repo,
final_review_decision_ready=final_review_decision_ready
)
# Include the inventory report in the response message
review_pr.py
+14 -14
View File
@@ -28,6 +28,14 @@ from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_req
def main(argv=None):
if os.environ.get("GITEA_SESSION_PROFILE_LOCK"):
print(
"Direct CLI review submission is disabled within MCP sessions. "
"Use MCP tools instead.",
file=sys.stderr,
)
return 2
parser = argparse.ArgumentParser(description="Review and sign-off on a Gitea pull request.")
add_remote_args(parser)
parser.add_argument("--pr-number", type=int, required=True, help="PR number/index to review.")
@@ -71,20 +79,12 @@ def main(argv=None):
# spoof it and it cannot go stale across sessions.
session_lock = (os.environ.get("GITEA_SESSION_PROFILE_LOCK") or "").strip()
if session_lock:
try:
cli_profile = (get_profile().get("profile_name") or "").strip()
except Exception as e:
print(f"Mutation authority check failed: {e}", file=sys.stderr)
return 3
if cli_profile != session_lock:
print(
f"Mismatched active profile vs session profile lock "
f"(CLI override rejected): CLI profile '{cli_profile}' does "
f"not match locked session profile '{session_lock}' "
f"(fail closed)",
file=sys.stderr,
)
return 3
print(
"Direct CLI review submission is disabled within MCP sessions. "
"Use the gated 'gitea_submit_pr_review' MCP tool instead.",
file=sys.stderr,
)
return 2
body = args.body
if args.body_file:
review_proofs.py
+38 -1
View File
@@ -634,7 +634,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
identity_eligible, merge_performed,
issue_status_verified,
capability_evidence=None, sweep=None, live_state=None,
role_boundary=None):
role_boundary=None, review_mutation=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -652,6 +652,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
(``assess_live_state_recheck`` — also required for ``merge_allowed``),
and a clean role boundary (``assess_role_boundary``). Omitting any of
them downgrades; a merge without the live recheck is a violation.
#211: the report must also carry the review mutation proof (``assess_review_mutation_final_report``).
"""
contamination_status = contamination.get("status", "unknown")
checkout_proven = bool(checkout_proof.get("proven"))
@@ -682,10 +684,23 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"proven": False,
"reasons": ["role-boundary/namespace usage not reported (#179)"],
}
review_mutation = review_mutation or {
"complete": False,
"downgraded": True,
"reasons": ["review mutation proof not provided (#211)"],
}
capability_proven = bool(capability_evidence.get("proven"))
sweep_proven = bool(sweep.get("proven"))
live_state_proven = bool(live_state.get("proven"))
role_boundary_clean = bool(role_boundary.get("proven"))
review_mutation_complete = bool(review_mutation.get("complete"))
review_mutation = review_mutation or {
"complete": False,
"reasons": ["review mutation proof missing"],
}
review_mutation_complete = bool(review_mutation.get("complete"))
downgrade_reasons = []
if not identity_eligible:
@@ -727,6 +742,9 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
if not role_boundary_clean:
downgrade_reasons.append("role/namespace boundary not clean (#179)")
downgrade_reasons.extend(role_boundary.get("reasons", []))
if not review_mutation_complete:
downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
downgrade_reasons.extend(review_mutation.get("reasons", []))
merge_allowed = (
identity_eligible
@@ -775,6 +793,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"sweep_verdict": sweep.get("verdict"),
"live_state_recheck_proven": live_state_proven,
"role_boundary_clean": role_boundary_clean,
"review_mutation_complete": review_mutation_complete,
}
@@ -1038,3 +1057,21 @@ def assess_duplicate_search_proof(report_text, matches):
return issue_duplicate_gate.assess_duplicate_search_proof(
report_text, matches
)
def build_review_mutation_proof(run_log: list[dict]) -> dict:
"""Assess the live review mutations recorded during this run."""
mutations = [e for e in (run_log or []) if e.get("kind") == "live_review_mutation"]
if not mutations:
return {
"complete": False,
"downgraded": True,
"missing_fields": ["live review mutation"],
"reasons": ["no live review mutation recorded in run log"],
}
return {
"complete": True,
"downgraded": False,
"missing_fields": [],
"reasons": [],
}
tests/conftest.py
+1
View File
@@ -25,4 +25,5 @@ def _reset_mutation_authority(monkeypatch):
return
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
yield
tests/test_mcp_server.py
+65 -1
View File
@@ -821,11 +821,16 @@ class TestReviewPR(unittest.TestCase):
{"login": "jcwalker3"}, # /api/v1/user (eligibility)
{"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1
]
from mcp_server import init_review_decision_lock
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(1, "approve", remote="prgs")
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
body="Self approve",
merge=False
merge=False,
remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(result["success"])
self.assertIn("Review submission failed eligibility gates", result["message"])
@@ -1773,6 +1778,65 @@ class TestSubmitPrReview(unittest.TestCase):
self.assertIn("[REDACTED]", blob)
self.assertNotIn("abc-secret-xyz", blob)
def test_authorize_review_correction_success(self):
from mcp_server import _load_review_decision_lock, _save_review_decision_lock
lock = _load_review_decision_lock() or {}
lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
_save_review_decision_lock(lock)
r = gitea_authorize_review_correction(prior_review_id=42, prior_review_state="approve", reason="typo")
self.assertTrue(r["authorized"])
lock = _load_review_decision_lock()
self.assertTrue(lock["correction_authorized"])
def test_authorize_review_correction_mismatch(self):
from mcp_server import _load_review_decision_lock, _save_review_decision_lock
lock = _load_review_decision_lock() or {}
lock["live_mutations"] = [{"pr_number": 8, "action": "approve", "review_id": 42, "review_state": "approve"}]
_save_review_decision_lock(lock)
# Mismatched ID
r = gitea_authorize_review_correction(prior_review_id=99, prior_review_state="approve", reason="typo")
self.assertFalse(r["authorized"])
self.assertTrue(any("prior review ID" in x for x in r["reasons"]))
# Mismatched State
r = gitea_authorize_review_correction(prior_review_id=42, prior_review_state="request_changes", reason="typo")
self.assertFalse(r["authorized"])
self.assertTrue(any("prior review state" in x for x in r["reasons"]))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_remote_org_repo_validation_mismatch(self, _auth, mock_api):
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
# Mark decision for specific remote, org, repo
gitea_mark_final_review_decision(8, "approve", remote="prgs", org="MyOrg", repo="MyRepo")
# Mismatched remote
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="dadeschools", org="MyOrg", repo="MyRepo",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertTrue(any("ready remote" in x for x in r["reasons"]))
# Mismatched org
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs", org="OtherOrg", repo="MyRepo",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertTrue(any("ready org" in x for x in r["reasons"]))
# Mismatched repo
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs", org="MyOrg", repo="OtherRepo",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertTrue(any("ready repo" in x for x in r["reasons"]))
if __name__ == "__main__":
unittest.main()
tests/test_pr_queue_inventory.py
+4 -1
View File
@@ -127,7 +127,10 @@ class TestPRQueueInventory(unittest.TestCase):
{"id": 100} # POST review
]
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
init_review_decision_lock("prgs", "review_pr")
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"])
self.assertIn("=== PR Queue Inventory ===", result["message"])
self.assertIn("Repository:", result["message"])
tests/test_review_pr.py
+5 -23
View File
@@ -112,15 +112,10 @@ class TestAPIPayload(unittest.TestCase):
class TestMutationAuthorityLock(unittest.TestCase):
"""#199 (refs #194): the CLI refuses to run under a profile that differs
from the session profile lock exported by the launching MCP session."""
"""#199 (refs #194): the CLI refuses to run under active MCP sessions."""
@patch("review_pr.get_profile")
def test_cli_blocked_on_session_lock_mismatch(self, mock_get_profile):
# An author-bound session exported the lock; the CLI resolves a
# reviewer profile (GITEA_MCP_PROFILE side-channel override) — reject
# before any API call.
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
def test_cli_disabled_on_session_lock(self):
# When running inside an MCP session, direct review submission via CLI is disabled entirely.
import io
buf = io.StringIO()
with patch.dict(os.environ,
@@ -129,22 +124,9 @@ class TestMutationAuthorityLock(unittest.TestCase):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
self.assertEqual(rc, 3)
self.assertEqual(rc, 2)
msg = buf.getvalue().lower()
self.assertIn("cli override rejected", msg)
@patch("review_pr.get_profile")
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
mock_api.side_effect = [FAKE_PR_DATA, {}]
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
self.assertEqual(rc, 0)
self.assertIn("disabled within mcp sessions", msg)
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
tests/test_review_proofs.py
+22
View File
@@ -167,6 +167,15 @@ def _good_live_state(**overrides):
return assess_live_state_recheck(recheck)
def _good_review_mutation():
return {
"complete": True,
"downgraded": False,
"missing_fields": [],
"reasons": [],
}
def _good_role_boundary_179(**overrides):
kwargs = {
"task_role": "reviewer",
@@ -531,6 +540,7 @@ class TestFinalReport(unittest.TestCase):
"sweep": _good_sweep(),
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
"review_mutation": _good_review_mutation(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -625,6 +635,7 @@ class TestFinalReport(unittest.TestCase):
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
"review_mutation": _good_review_mutation(),
}
report = build_final_report(**kwargs)
self.assertNotEqual(report["grade"], "A")
@@ -1131,6 +1142,7 @@ class TestFinalReport179Bar(unittest.TestCase):
"sweep": _good_sweep(),
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
"review_mutation": _good_review_mutation(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -1195,6 +1207,16 @@ class TestFinalReport179Bar(unittest.TestCase):
self.assertTrue(report["inventory_complete"])
self.assertTrue(report["validated_on_pinned_head"])
def test_missing_review_mutation_downgrades(self):
report = self._report(review_mutation=None)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["review_mutation_complete"])
def test_incomplete_review_mutation_downgrades(self):
report = self._report(review_mutation={"complete": False, "downgraded": True, "reasons": []})
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["review_mutation_complete"])
if __name__ == "__main__":
unittest.main()