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

Merge pull request 'feat(reviewer-workflow): enforce single terminal review decision per PR review run' (#214) from feat/issue-211-single-terminal-review-decision into master

Browse Source
This commit was merged in pull request #214.
This commit is contained in:
sysadmin
2026-07-05 19:19:34 -05:00
parent 5b9fcaefbf 16f977fde0
commit adb35f12b5
12 changed files with 1008 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp_server.py
+419 -60
View File
@@ -13,6 +13,7 @@ Configuration (mcp_config.json):
"env": {}
}
"""
import json
import os
import re
import sys
@@ -921,6 +922,176 @@ _REVIEW_ACTIONS = {
"request_changes": ("request_changes", "REQUEST_CHANGES"),
}
_TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
# In-process only (#211): never persist to /tmp — host-global files are
# spoofable by any local process and go stale across sessions.
_REVIEW_DECISION_LOCK: dict | None = None
def _load_review_decision_lock():
global _REVIEW_DECISION_LOCK
return _REVIEW_DECISION_LOCK
def _save_review_decision_lock(data):
global _REVIEW_DECISION_LOCK
_REVIEW_DECISION_LOCK = data
def _review_decision_session_reasons(lock: dict | None) -> list[str]:
"""Reject locks that do not belong to this MCP process/session."""
if lock is None:
return []
reasons = []
if lock.get("session_pid") != os.getpid():
reasons.append(
"review decision lock was created in a different process "
"(fail closed)"
)
env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
stored_lock = (lock.get("session_profile_lock") or "").strip()
if env_lock and stored_lock and env_lock != stored_lock:
reasons.append(
"review decision lock session profile lock mismatch (fail closed)"
)
return reasons
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
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip()
session_lock = (
(os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
or profile_name
)
_save_review_decision_lock({
"task": task,
"remote": remote,
"session_pid": os.getpid(),
"session_profile": profile_name,
"session_profile_lock": session_lock,
"final_review_decision_ready": False,
"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,
})
def check_review_decision_gate(
pr_number: int,
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 = []
lock = _load_review_decision_lock()
if lock is None:
reasons.append(
"review decision lock missing; call gitea_resolve_task_capability "
"for review_pr before live review mutations (fail closed)"
)
return reasons
reasons.extend(_review_decision_session_reasons(lock))
if reasons:
return reasons
if remote in REMOTES:
_, org, repo = _resolve(remote, None, org, repo)
if not final_review_decision_ready:
reasons.append(
"final_review_decision_ready must be true; validation-phase live "
"review mutations are forbidden — use gitea_dry_run_pr_review "
"instead (fail closed)"
)
return reasons
if not lock.get("final_review_decision_ready"):
reasons.append(
"final review decision not marked ready; call "
"gitea_mark_final_review_decision after validation completes "
"(fail closed)"
)
return reasons
if lock.get("ready_pr_number") != pr_number:
reasons.append(
f"ready PR #{lock.get('ready_pr_number')} does not match "
f"requested PR #{pr_number} (fail closed)"
)
if lock.get("ready_action") != action:
reasons.append(
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"):
reasons.append(
"live review mutation already recorded in this run; only one live "
"review mutation is allowed unless "
"gitea_authorize_review_correction was invoked (fail closed)"
)
elif (
action in _TERMINAL_REVIEW_ACTIONS
and any(m.get("action") in _TERMINAL_REVIEW_ACTIONS for m in prior)
and not lock.get("correction_authorized")
):
reasons.append(
"terminal review decision already submitted on this PR in this "
"run; blocked unless an operator-approved correction was "
"authorized (fail closed)"
)
return reasons
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,
"review_id": review_id,
"review_state": action,
})
lock["live_mutations"] = mutations
if lock.get("correction_authorized"):
lock["correction_authorized"] = False
lock["correction_reason"] = None
_save_review_decision_lock(lock)
# Patterns scrubbed from any surfaced error text so a credential can never leak.
_SECRET_PREFIXES = ("token ", "Basic ")
@@ -1085,9 +1256,7 @@ def gitea_get_pr_review_feedback(
}
@mcp.tool()
@_audit_pr_result("submit_pr_review")
def gitea_submit_pr_review(
def _evaluate_pr_review_submission(
pr_number: int,
action: str,
body: str = "",
@@ -1096,55 +1265,17 @@ def gitea_submit_pr_review(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
*,
live: bool,
final_review_decision_ready: bool = False,
) -> dict:
"""Gated PR review mutation: comment findings, request changes, or approve.
This is the only tool that submits a Gitea PR *review*. It performs a
mutation **only after every safety gate passes**; if any gate fails it
returns ``performed=False`` and never calls the mutating endpoint.
Gate order (fail-closed at each step):
1. Validate ``action`` is one of 'comment', 'approve', 'request_changes'.
2. Reuse ``gitea_check_pr_eligibility`` (#14), which runs the authenticated
-user lookup, active-profile lookup, PR-author lookup, self-approval
block, and profile-allowed-operation check. ``approve`` requires
eligibility for 'approve', ``request_changes`` requires
'request_changes', and ``comment`` requires 'review'.
3. Redundantly block self-approval (authenticated user == PR author).
4. If ``expected_head_sha`` is supplied and the PR head has moved, abort.
5. Only then POST the review.
Endpoint: ``POST /repos/{owner}/{repo}/pulls/{n}/reviews``. This is the
*formal review* API (it records an APPROVE / COMMENT / REQUEST_CHANGES
review state tied to the head commit), chosen over the plain issue-comment
endpoint (``/issues/{n}/comments``) so that approvals and change requests
carry real review state — a plain comment cannot approve or block a PR.
Merge is intentionally NOT implemented here.
Never returns the token, Authorization header, or any credential material.
Args:
pr_number: Target PR number.
action: 'comment', 'approve', or 'request_changes'.
body: Review body / finding text.
expected_head_sha: Optional. If given and the PR head SHA differs, the
review is refused (guards against reviewing a changed PR).
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict describing the attempt: action, whether it was performed, the
authenticated user, profile name, PR author, PR number, head SHA
checked, and the reasons/gates passed or blocked. Never secrets.
"""
"""Shared gate chain for live submit and dry-run review tools."""
action = (action or "").strip().lower()
result = {
"requested_action": action,
"performed": False,
"dry_run": not live,
"would_perform": False,
"authenticated_user": None,
"profile_name": get_profile()["profile_name"],
"pr_author": None,
@@ -1156,7 +1287,6 @@ def gitea_submit_pr_review(
}
reasons = result["reasons"]
# Gate 1 — valid review action (no mutation on unknown action).
if action not in _REVIEW_ACTIONS:
reasons.append(
f"unknown review action '{action}'; expected one of "
@@ -1165,8 +1295,18 @@ def gitea_submit_pr_review(
return result
eligibility_action, event = _REVIEW_ACTIONS[action]
# Gate 2 — reuse #14 eligibility (identity + profile + author + self-approve
# + profile-allowed). This performs only read-only GETs.
if live:
reasons.extend(check_review_decision_gate(
pr_number,
action,
final_review_decision_ready=final_review_decision_ready,
remote=remote,
org=org,
repo=repo,
))
if reasons:
return result
elig = gitea_check_pr_eligibility(
pr_number=pr_number,
action=eligibility_action,
@@ -1188,30 +1328,36 @@ def gitea_submit_pr_review(
result["permission_report"] = elig["permission_report"]
return result
# Gate 3 — redundant self-approval block (belt-and-suspenders over #14).
auth_user = result["authenticated_user"]
pr_author = result["pr_author"]
if action == "approve" and auth_user and pr_author and auth_user == pr_author:
reasons.append("self-approval blocked (authenticated user is PR author)")
return result
# Gate 4 — head SHA must match if the caller pinned one.
actual_sha = result["head_sha"]
if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
pinned_sha = expected_head_sha
lock = _load_review_decision_lock() or {}
if live and lock.get("ready_expected_head_sha"):
pinned_sha = lock.get("ready_expected_head_sha")
if pinned_sha and actual_sha and pinned_sha != actual_sha:
reasons.append(
"expected head SHA does not match current PR head (fail closed)"
)
return result
if not actual_sha:
# Should be unreachable — eligibility fails closed without a head SHA —
# but never submit a review without a commit to pin it to.
reasons.append("PR head SHA unavailable (fail closed)")
return result
result["would_perform"] = True
if not live:
reasons.append(
f"dry-run only: would submit '{event}' review on PR #{pr_number}; "
"no live mutation performed"
)
return result
# Gate 5 — in-process mutation authority (#199): the last check before
# the mutating POST, using the identity the eligibility gate proved.
# A profile/identity flip or side-channel override between preflight
# and mutation fails closed here.
try:
verify_mutation_authority(remote, host, required_role="reviewer",
active_identity=auth_user)
@@ -1219,22 +1365,227 @@ def gitea_submit_pr_review(
reasons.append(str(e))
return result
# All gates passed — perform the single mutating call.
h, o, r = _resolve(remote, host, org, repo)
review_id = None
try:
auth = _auth(h)
review_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews"
payload = {"body": body, "event": event, "commit_id": actual_sha}
api_request("POST", review_url, auth, payload)
resp = api_request("POST", review_url, auth, payload)
if isinstance(resp, dict):
review_id = resp.get("id")
except Exception as exc: # noqa: BLE001 — redact before surfacing
reasons.append(f"review submission failed: {_redact(str(exc))}")
return result
record_live_review_mutation(pr_number, action, review_id)
result["performed"] = True
reasons.append(f"all gates passed; submitted '{event}' review on PR #{pr_number}")
return result
@mcp.tool()
def gitea_mark_final_review_decision(
pr_number: int,
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()
lock = _load_review_decision_lock()
if lock is None:
return {
"marked_ready": False,
"reasons": [
"review decision lock missing; resolve review_pr capability first"
],
}
session_reasons = _review_decision_session_reasons(lock)
if session_reasons:
return {"marked_ready": False, "reasons": session_reasons}
if remote != lock.get("remote"):
return {
"marked_ready": False,
"reasons": [
f"requested remote '{remote}' does not match locked remote "
f"'{lock.get('remote')}' (fail closed)"
],
}
try:
_, resolved_org, resolved_repo = _resolve(remote, None, org, repo)
except ValueError as exc:
return {"marked_ready": False, "reasons": [str(exc)]}
if org is not None and org != resolved_org:
return {
"marked_ready": False,
"reasons": [
f"requested org '{org}' does not match resolved org "
f"'{resolved_org}' (fail closed)"
],
}
if repo is not None and repo != resolved_repo:
return {
"marked_ready": False,
"reasons": [
f"requested repo '{repo}' does not match resolved repo "
f"'{resolved_repo}' (fail closed)"
],
}
org = resolved_org
repo = resolved_repo
if lock.get("live_mutations") and not lock.get("correction_authorized"):
return {
"marked_ready": False,
"reasons": [
"cannot mark final decision after a live review mutation was "
"already recorded in this run unless an operator-approved "
"correction was authorized"
],
}
if action not in _REVIEW_ACTIONS:
return {
"marked_ready": False,
"reasons": [
f"unknown review action '{action}'; expected one of "
f"{sorted(_REVIEW_ACTIONS)}"
],
}
lock["final_review_decision_ready"] = True
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(
prior_review_id: int,
prior_review_state: str,
reason: str,
operator_authorized: bool = False,
) -> 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"]}
session_reasons = _review_decision_session_reasons(lock)
if session_reasons:
return {"authorized": False, "reasons": session_reasons}
if not operator_authorized:
return {
"authorized": False,
"reasons": [
"operator authorization is required for review correction "
"(fail closed)"
],
}
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:
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)
return {"authorized": True, "correction_reason": reason, "reasons": []}
@mcp.tool()
def gitea_dry_run_pr_review(
pr_number: int,
action: str,
body: str = "",
expected_head_sha: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Validate review submission mechanics without a live PR mutation."""
return _evaluate_pr_review_submission(
pr_number=pr_number,
action=action,
body=body,
expected_head_sha=expected_head_sha,
remote=remote,
host=host,
org=org,
repo=repo,
live=False,
)
@mcp.tool()
@_audit_pr_result("submit_pr_review")
def gitea_submit_pr_review(
pr_number: int,
action: str,
body: str = "",
expected_head_sha: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
final_review_decision_ready: bool = False,
) -> dict:
"""Gated PR review mutation: comment findings, request changes, or approve.
Live mutations require ``final_review_decision_ready=True`` and a prior
``gitea_mark_final_review_decision`` call. Use ``gitea_dry_run_pr_review``
during validation instead of probing with live submissions.
"""
return _evaluate_pr_review_submission(
pr_number=pr_number,
action=action,
body=body,
expected_head_sha=expected_head_sha,
remote=remote,
host=host,
org=org,
repo=repo,
live=True,
final_review_decision_ready=final_review_decision_ready,
)
@mcp.tool()
def gitea_edit_pr(
pr_number: int,
@@ -1668,6 +2019,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).
@@ -1842,7 +2194,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
@@ -3790,6 +4143,12 @@ def gitea_resolve_task_capability(
"STOP: the active profile cannot perform the requested task; "
"follow exact_safe_next_action instead of improvising.")
if task == "review_pr":
init_review_decision_lock(
remote if remote in REMOTES else None,
task,
)
record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
return {
review_pr.py
+16 -85
View File
@@ -7,16 +7,16 @@ making any API call. Merge is handled solely by the gated `gitea_merge_pr` MCP
workflow (#16), which enforces identity/profile/eligibility, explicit
confirmation, expected head SHA checking, and self-merge protection.
Usage (review only):
Live review submission is also disabled (#211): use the gated
``gitea_submit_pr_review`` MCP workflow, which enforces validation-phase
dry-run, final decision marking, and single-terminal review mutation rules.
Usage (review only — disabled):
review_pr.py --pr-number 12 --event APPROVE --body "Approved and signed off"
"""
import os
import sys
import json
import base64
import argparse
import urllib.request
import urllib.error
# Auto-execute using the project's local virtual environment Python
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
@@ -24,7 +24,7 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3")
if os.path.exists(venv_python) and sys.executable != venv_python:
os.execv(venv_python, [venv_python] + sys.argv)
from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
from gitea_auth import add_remote_args
def main(argv=None):
@@ -42,95 +42,26 @@ def main(argv=None):
help="Ignored — CLI merge is disabled (see --merge).")
args = parser.parse_args(argv)
# Fail closed: direct CLI merge is disabled (#16). LLM automations were
# using this flag as an ungated merge bypass. Merge is only available via
# the gated `gitea_merge_pr` MCP workflow, which enforces
# identity/profile/eligibility, explicit confirmation, expected head SHA,
# and self-merge protection. No API call is made here.
if args.merge:
print(
"Direct CLI merge is disabled. Merge is only available through the "
"gated #16 workflow (MCP tool 'gitea_merge_pr'), which enforces "
"identity/profile/eligibility, explicit confirmation, expected head "
"SHA checking, and self-merge protection. Re-run without --merge to "
"submit a review only.",
"see the review-submission guard message.",
file=sys.stderr,
)
return 2
host, org, repo = resolve_remote(args)
# ── Reviewer mutation side-channel wall (#199, refs #194) ──
# The launching MCP session exports GITEA_SESSION_PROFILE_LOCK with the
# profile it was started with; child processes inherit it. If this CLI
# resolves a different profile — e.g. an ad-hoc GITEA_MCP_PROFILE
# override escalating an author-bound session to reviewer — refuse
# before any API call. No lock in the environment means no session
# context (direct operator CLI use), which stays allowed. Unlike a /tmp
# lock file, the environment is per-process-tree: other sessions cannot
# 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
body = args.body
if args.body_file:
if args.body_file == "-":
body = sys.stdin.read()
else:
with open(args.body_file, "r", encoding="utf-8") as fh:
body = fh.read()
auth = get_auth_header(host)
if not auth:
print(f"Could not get credentials or token for {host}.", file=sys.stderr)
return 1
# 1. Fetch PR to get the latest head commit SHA (required for review validation)
pr_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}"
try:
pr_data = api_request("GET", pr_url, auth)
except Exception as e:
print(f"Error fetching PR #{args.pr_number}: {e}", file=sys.stderr)
return 1
commit_sha = pr_data.get("head", {}).get("sha")
if not commit_sha:
print(f"Could not find head commit SHA for PR #{args.pr_number}.", file=sys.stderr)
return 1
# 2. Submit the PR review
review_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}/reviews"
payload = {
"body": body,
"event": args.event,
"commit_id": commit_sha
}
try:
api_request("POST", review_url, auth, payload)
print(f"Successfully submitted review for PR #{args.pr_number}: event={args.event}")
except Exception as e:
print(f"Error submitting review: {e}", file=sys.stderr)
return 1
# Merge is intentionally not performed here — see the fail-closed guard
# above. Use the gated `gitea_merge_pr` MCP workflow (#16) to merge.
return 0
print(
"Direct CLI review submission is disabled (#211). Use the gated "
"'gitea_submit_pr_review' MCP workflow, which enforces validation-phase "
"dry-run, gitea_mark_final_review_decision, and single-terminal review "
"mutation rules.",
file=sys.stderr,
)
return 2
if __name__ == "__main__":
sys.exit(main())
sys.exit(main())
review_proofs.py
+92 -1
View File
@@ -582,11 +582,60 @@ def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
}
def assess_review_mutation_final_report(report_text, review_decision_lock):
"""Require final reports to list exactly one live review mutation.
Two mutations are allowed only when an operator-approved correction flow
was invoked and explained in the report.
"""
lock = review_decision_lock or {}
text = report_text or ""
lower = text.lower()
mutations = list(lock.get("live_mutations") or [])
missing = []
count = len(mutations)
if count == 0:
if any(term in lower for term in ("submitted 'approve'", "submitted 'request_changes'", "live review mutation")):
missing.append("review mutation count")
elif count == 1:
m = mutations[0]
action = m.get("action")
pr_number = m.get("pr_number")
if action and action.lower() not in lower:
missing.append("review mutation action")
if pr_number is not None and f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
missing.append("review mutation PR number")
elif count > 1:
if not lock.get("correction_authorized") and "correction" not in lower:
missing.append("correction flow explanation")
if "review mutations:" not in lower and "review mutation" not in lower:
missing.append("review mutation listing")
if missing:
return {
"complete": False,
"downgraded": True,
"missing_fields": missing,
"reasons": [
f"final report missing review-mutation field: {field}"
for field in missing
],
}
return {
"complete": True,
"downgraded": False,
"missing_fields": [],
"reasons": [],
}
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,
report_text=None, review_decision_lock=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -604,7 +653,14 @@ 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``).
"""
if review_mutation is None and report_text is not None:
review_mutation = assess_review_mutation_final_report(
report_text, review_decision_lock
)
contamination_status = contamination.get("status", "unknown")
checkout_proven = bool(checkout_proof.get("proven"))
validation_claimable = bool(validation.get("claimable"))
@@ -634,10 +690,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:
@@ -679,6 +748,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
@@ -727,6 +799,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,
}
@@ -990,3 +1063,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": [],
}
skills/llm-project-workflow/SKILL.md
+9
View File
@@ -231,6 +231,15 @@ Worktree folder = branch with `/` replaced by `-`
differs from the repository's canonical validation command. Only claim a
validation result after the command has completed and its output has
been read (`review_proofs.assess_validation_report`).
During validation, review work is **read-only**: use
`gitea_dry_run_pr_review` to prove submission mechanics — never post live
APPROVE, REQUEST_CHANGES, or review comments to probe tool paths. After
validation completes, call `gitea_mark_final_review_decision`, then submit
exactly one live review via
`gitea_submit_pr_review(..., final_review_decision_ready=True)`.
Final reports must list exactly one review mutation
(`review_proofs.assess_review_mutation_final_report`) unless an
operator-approved correction flow was invoked and explained.
10. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
11. **#179 A-bar proofs** (all fail closed when missing —
`review_proofs.assess_capability_evidence`, `assess_sweep_evidence`,
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_audit.py
+5 -1
View File
@@ -333,8 +333,12 @@ class TestGatedToolAudit(_AuditWiringBase):
env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
GITEA_ALLOWED_OPERATIONS="read,review,approve")
with patch.dict(os.environ, env, clear=True):
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(8, "approve", remote="prgs")
r = gitea_submit_pr_review(pr_number=8, action="approve",
body="LGTM", remote="prgs")
body="LGTM", remote="prgs",
final_review_decision_ready=True)
self.assertTrue(r["performed"])
recs = self._records()
self.assertEqual(len(recs), 1)
tests/test_llm_agent_sha.py
+6 -3
View File
@@ -125,14 +125,17 @@ class TestShaCannotBypassSelfReview(unittest.TestCase):
mock_get_all.return_value = [{"number": 9, "title": "PR 9", "state": "open", "head": {"ref": "branch9", "sha": "abc1234"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "jcwalker3"}}]
mock_api.side_effect = [
{"login": "jcwalker3"}, # /user (inventory)
{"login": "jcwalker3"}, # /user (eligibility)
{"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}} # /pulls/9 (eligibility)
{"login": "jcwalker3"}, # /user (submit eligibility)
{"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/9
]
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(9, "approve", remote="prgs")
env = self._env(SHA_WOULD_BE_REVIEWER, "reviewer")
with patch.dict(os.environ, env, clear=True):
r = gitea_review_pr(
pr_number=9, event="APPROVE", body="self approve", merge=False,
remote="prgs")
remote="prgs", final_review_decision_ready=True)
self.assertFalse(r["success"])
self.assertIn("authenticated user is PR author", r["message"])
for call in mock_api.call_args_list:
tests/test_mcp_server.py
+314 -15
View File
@@ -31,6 +31,11 @@ from mcp_server import ( # noqa: E402
gitea_get_profile,
gitea_check_pr_eligibility,
gitea_submit_pr_review,
gitea_dry_run_pr_review,
gitea_mark_final_review_decision,
gitea_authorize_review_correction,
init_review_decision_lock,
gitea_list_issue_comments,
gitea_create_issue_comment,
)
@@ -813,14 +818,19 @@ class TestReviewPR(unittest.TestCase):
# mock_api responses: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility)
mock_api.side_effect = [
{"login": "jcwalker3"}, # /api/v1/user (inventory)
{"login": "jcwalker3"}, # /api/v1/user (eligibility)
{"state": "open", "head": {"sha": "abc1234"}, "mergeable": True, "user": {"login": "jcwalker3"}}, # /pulls/1
{"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")
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"])
@@ -1395,9 +1405,117 @@ class TestPrEligibility(unittest.TestCase):
self.assertNotIn(secret, blob)
class TestReviewDecisionValidationGate(unittest.TestCase):
"""Block incidental live review mutations during validation."""
PR = 203
SHA = "abc123"
def _pr(self, author, sha=SHA):
return {
"user": {"login": author},
"state": "open",
"head": {"sha": sha},
"mergeable": True,
}
def setUp(self):
init_review_decision_lock("prgs", "review_pr")
def _env(self):
return patch.dict(os.environ, {
"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes",
}, clear=True)
def _assert_no_mutation(self, mock_api):
for c in mock_api.call_args_list:
method, url = c.args[0], c.args[1]
self.assertFalse(
method == "POST" and url.endswith("/reviews"),
f"unexpected review mutation: {method} {url}",
)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_approve_during_validation_blocked(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
]
with self._env():
r = gitea_submit_pr_review(
pr_number=self.PR, action="approve", remote="prgs",
final_review_decision_ready=False,
)
self.assertFalse(r["performed"])
self.assertTrue(any(
"final_review_decision_ready must be true" in x for x in r["reasons"]
))
self._assert_no_mutation(mock_api)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_request_changes_during_validation_blocked(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
]
with self._env():
r = gitea_submit_pr_review(
pr_number=self.PR, action="request_changes", remote="prgs",
final_review_decision_ready=False,
)
self.assertFalse(r["performed"])
self.assertTrue(any(
"final_review_decision_ready must be true" in x for x in r["reasons"]
))
self._assert_no_mutation(mock_api)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_duplicate_terminal_decision_blocked(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 1},
{"login": "reviewer-bot"}, self._pr("author-bot"),
]
gitea_mark_final_review_decision(
self.PR, "approve", expected_head_sha=self.SHA, remote="prgs")
with self._env():
first = gitea_submit_pr_review(
pr_number=self.PR, action="approve", remote="prgs",
final_review_decision_ready=True,
)
second = gitea_submit_pr_review(
pr_number=self.PR, action="request_changes", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(first["performed"])
self.assertFalse(second["performed"])
self.assertTrue(any(
"live review mutation already recorded" in x for x in second["reasons"]
))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_dry_run_proves_submission_without_live_mutation(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
]
with self._env():
r = gitea_dry_run_pr_review(
pr_number=self.PR, action="approve", remote="prgs")
self.assertFalse(r["performed"])
self.assertTrue(r["would_perform"])
self.assertTrue(r["dry_run"])
self._assert_no_mutation(mock_api)
class TestSubmitPrReview(unittest.TestCase):
"""Gated review-mutation tool (#15)."""
def setUp(self):
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(8, "approve", remote="prgs")
def _pr(self, author, state="open", sha="abc123", mergeable=True):
return {
"user": {"login": author},
@@ -1427,7 +1545,10 @@ class TestSubmitPrReview(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(pr_number=8, action="approve", remote="prgs")
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertIn("authenticated user is PR author", r["reasons"])
self._assert_no_mutation(mock_api)
@@ -1442,7 +1563,9 @@ class TestSubmitPrReview(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve", body="LGTM", remote="prgs")
pr_number=8, action="approve", body="LGTM", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
self.assertEqual(r["authenticated_user"], "reviewer-bot")
self.assertEqual(r["pr_author"], "author-bot")
@@ -1459,6 +1582,7 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 9},
]
@@ -1467,11 +1591,14 @@ class TestSubmitPrReview(unittest.TestCase):
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="request_changes",
body="needs work", remote="prgs")
body="needs work", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
self.assertEqual(mock_api.call_args.args[3]["event"], "REQUEST_CHANGES")
def test_request_changes_blocked_without_eligibility(self):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \
patch("mcp_server.api_request") as mock_api:
mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
@@ -1479,7 +1606,9 @@ class TestSubmitPrReview(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,review"} # no request_changes
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="request_changes", remote="prgs")
pr_number=8, action="request_changes", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertIn("profile is not allowed to request_changes", r["reasons"])
self._assert_no_mutation(mock_api)
@@ -1489,6 +1618,7 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_comment_succeeds_when_review_eligible(self, _auth, mock_api):
gitea_mark_final_review_decision(8, "comment", remote="prgs")
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 3},
]
@@ -1496,7 +1626,9 @@ class TestSubmitPrReview(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,review"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="comment", body="finding", remote="prgs")
pr_number=8, action="comment", body="finding", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
self.assertEqual(mock_api.call_args.args[3]["event"], "COMMENT")
@@ -1510,8 +1642,11 @@ class TestSubmitPrReview(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review"}
with patch.dict(os.environ, env, clear=True):
gitea_mark_final_review_decision(8, "comment", remote="prgs")
r = gitea_submit_pr_review(
pr_number=8, action="comment", body="note", remote="prgs")
pr_number=8, action="comment", body="note", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
# -- identity / profile fail-closed ---------------------------------------
@@ -1521,7 +1656,10 @@ class TestSubmitPrReview(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(pr_number=8, action="approve", remote="prgs")
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertIsNone(r["authenticated_user"])
self.assertIn("authenticated identity could not be determined", r["reasons"])
@@ -1534,7 +1672,9 @@ class TestSubmitPrReview(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,pr.create"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs")
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertIn("profile is not allowed to approve", r["reasons"])
self._assert_no_mutation(mock_api)
@@ -1552,7 +1692,9 @@ class TestSubmitPrReview(unittest.TestCase):
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve",
expected_head_sha="deadbeef", remote="prgs")
expected_head_sha="deadbeef", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertIn(
"expected head SHA does not match current PR head (fail closed)",
@@ -1571,7 +1713,9 @@ class TestSubmitPrReview(unittest.TestCase):
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve",
expected_head_sha="abc123", remote="prgs")
expected_head_sha="abc123", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
# -- invalid action -------------------------------------------------------
@@ -1596,7 +1740,11 @@ class TestSubmitPrReview(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,review,approve",
"GITEA_TOKEN": "super-secret-token"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(pr_number=5, action="approve", remote="prgs")
gitea_mark_final_review_decision(5, "approve", remote="prgs")
r = gitea_submit_pr_review(
pr_number=5, action="approve", remote="prgs",
final_review_decision_ready=True,
)
blob = repr(r).lower()
for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
self.assertNotIn(secret, blob)
@@ -1612,12 +1760,163 @@ class TestSubmitPrReview(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(pr_number=5, action="approve", remote="prgs")
gitea_mark_final_review_decision(5, "approve", remote="prgs")
r = gitea_submit_pr_review(
pr_number=5, action="approve", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
blob = repr(r)
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",
operator_authorized=True,
)
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",
operator_authorized=True,
)
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",
operator_authorized=True,
)
self.assertFalse(r["authorized"])
self.assertTrue(any("prior review state" in x for x in r["reasons"]))
def test_authorize_review_correction_requires_operator_auth(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",
operator_authorized=False,
)
self.assertFalse(r["authorized"])
self.assertTrue(any("operator authorization" in x for x in r["reasons"]))
def test_spoofed_tmp_lock_file_does_not_bypass_gate(self):
import json
import mcp_server
mcp_server._REVIEW_DECISION_LOCK = None
spoof_path = "/tmp/gitea_review_decision.lock"
with open(spoof_path, "w", encoding="utf-8") as fh:
json.dump({"final_review_decision_ready": True}, fh)
try:
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertTrue(any("review decision lock missing" in x for x in r["reasons"]))
finally:
if os.path.exists(spoof_path):
os.remove(spoof_path)
def test_mark_final_decision_rejects_remote_mismatch(self):
init_review_decision_lock("prgs", "review_pr")
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"]))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_correction_flow_allows_second_terminal_review(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 42},
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 43},
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
gitea_mark_final_review_decision(8, "approve", remote="prgs")
with patch.dict(os.environ, env, clear=True):
first = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
auth = gitea_authorize_review_correction(
prior_review_id=42,
prior_review_state="approve",
reason="operator approved correcting mistaken approve",
operator_authorized=True,
)
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
second = gitea_submit_pr_review(
pr_number=8, action="request_changes", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(first["performed"])
self.assertTrue(auth["authorized"])
self.assertTrue(second["performed"])
@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_permission_reports.py
+8 -2
View File
@@ -229,9 +229,12 @@ 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_mark_final_review_decision(42, "approve", remote="prgs")
with patch.dict(os.environ, self._env("author-profile")):
res = mcp_server.gitea_submit_pr_review(
pr_number=42, action="approve", body="lgtm", remote="prgs")
pr_number=42, action="approve", body="lgtm", remote="prgs",
final_review_decision_ready=True)
self.assertFalse(res["performed"])
self.assertIn("permission_report", res)
self.assertEqual(res["permission_report"]["missing_permission"],
@@ -268,9 +271,12 @@ class TestReviewCommentPathUsesCanonicalOp(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_mark_final_review_decision(42, "comment", remote="prgs")
with patch.dict(os.environ, self._env("author-profile")):
res = mcp_server.gitea_submit_pr_review(
pr_number=42, action="comment", body="finding", remote="prgs")
pr_number=42, action="comment", body="finding", remote="prgs",
final_review_decision_ready=True)
self.assertFalse(res["performed"])
report = res["permission_report"]
self._assert_report_safe(report)
tests/test_pr_queue_inventory.py
+7 -4
View File
@@ -122,12 +122,15 @@ class TestPRQueueInventory(unittest.TestCase):
# mock_api: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility), 4) /pulls/1/reviews (POST review)
mock_api.side_effect = [
{"login": "reviewer1"}, # inventory whoami
{"login": "reviewer1"}, # eligibility whoami
{"state": "open", "head": {"sha": "abc1"}, "mergeable": True, "user": {"login": "other_user"}}, # eligibility PR details
{"id": 100} # POST review
{"login": "reviewer1"}, # submit eligibility whoami
{"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
{"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
+29 -76
View File
@@ -34,8 +34,7 @@ class TestArgParsing(unittest.TestCase):
self.exists_patcher.start()
self.addCleanup(self.exists_patcher.stop)
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_missing_pr_number_exits(self, _auth):
def test_missing_pr_number_exits(self):
with self.assertRaises(SystemExit):
review_pr.main([])
@@ -47,37 +46,21 @@ class TestAPIPayload(unittest.TestCase):
self.exists_patcher.start()
self.addCleanup(self.exists_patcher.stop)
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_payload_fields_and_workflow(self, _auth, mock_api):
# Setup mock api_request to return PR details, then review response
mock_api.side_effect = [FAKE_PR_DATA, {}]
def test_review_submission_fails_closed_without_api_call(self):
# #211: live review submission must use gated MCP tools, not CLI POST.
import io
buf = io.StringIO()
with patch.object(sys, "stderr", buf):
rc = review_pr.main([
"--pr-number", "81",
"--event", "APPROVE",
"--body", "Approved and ready to merge",
])
self.assertEqual(rc, 2)
self.assertIn("disabled", buf.getvalue().lower())
self.assertIn("gitea_submit_pr_review", buf.getvalue())
rc = review_pr.main([
"--pr-number", "81",
"--event", "APPROVE",
"--body", "Approved and ready to merge",
])
self.assertEqual(rc, 0)
self.assertEqual(mock_api.call_count, 2)
# Verify first call: GET PR
first_call_args = mock_api.call_args_list[0]
self.assertEqual(first_call_args[0][0], "GET")
self.assertEqual(first_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81")
# Verify second call: POST review
second_call_args = mock_api.call_args_list[1]
self.assertEqual(second_call_args[0][0], "POST")
self.assertEqual(second_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81/reviews")
payload = second_call_args[0][3]
self.assertEqual(payload["event"], "APPROVE")
self.assertEqual(payload["body"], "Approved and ready to merge")
self.assertEqual(payload["commit_id"], "abcdef1234567890")
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_merge_flag_fails_closed_without_api_call(self, _auth, mock_api):
def test_merge_flag_fails_closed_without_api_call(self):
# --merge is an ungated bypass and is disabled (#16). It must fail
# closed BEFORE any API call — no review, no merge.
rc = review_pr.main([
@@ -88,39 +71,25 @@ class TestAPIPayload(unittest.TestCase):
"--merge-method", "squash",
])
self.assertEqual(rc, 2)
self.assertEqual(mock_api.call_count, 0)
def test_merge_flag_message_points_to_gated_workflow(self):
from _pytest.monkeypatch import MonkeyPatch
import io
with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \
patch("review_pr.api_request") as mock_api:
buf = io.StringIO()
monkeypatch = MonkeyPatch()
monkeypatch.setattr(sys, "stderr", buf)
try:
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE", "--merge",
])
finally:
monkeypatch.undo()
buf = io.StringIO()
with patch.object(sys, "stderr", buf):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE", "--merge",
])
self.assertEqual(rc, 2)
self.assertEqual(mock_api.call_count, 0)
msg = buf.getvalue().lower()
self.assertIn("disabled", msg)
self.assertIn("gitea_merge_pr", msg)
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,36 +98,20 @@ 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)
self.assertIn("disabled", msg)
self.assertIn("gitea_submit_pr_review", 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)
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_cli_allowed_without_session_lock(self, _auth, mock_api):
# No lock in the environment = direct operator CLI use; the wall
# does not apply and the normal flow proceeds.
mock_api.side_effect = [FAKE_PR_DATA, {}]
def test_cli_disabled_even_without_session_lock(self):
# #211: CLI review submission is fail-closed regardless of session lock.
env = {k: v for k, v in os.environ.items()
if k != "GITEA_SESSION_PROFILE_LOCK"}
with patch.dict(os.environ, env, clear=True):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
self.assertEqual(rc, 0)
self.assertEqual(rc, 2)
if __name__ == "__main__":
tests/test_review_proofs.py
+102
View File
@@ -25,6 +25,7 @@ from review_proofs import ( # noqa: E402
assess_controller_handoff,
assess_inventory_completeness,
assess_live_state_recheck,
assess_review_mutation_final_report,
assess_role_boundary,
assess_self_review_contamination,
assess_sweep_evidence,
@@ -166,6 +167,19 @@ def _good_live_state(**overrides):
return assess_live_state_recheck(recheck)
def _good_review_mutation():
report = (
"Review complete on PR #203.\n"
"Review mutations: one request_changes on #203.\n"
"Review decision: request_changes."
)
lock = {
"live_mutations": [{"pr_number": 203, "action": "request_changes"}],
"correction_authorized": False,
}
return assess_review_mutation_final_report(report, lock)
def _good_role_boundary_179(**overrides):
kwargs = {
"task_role": "reviewer",
@@ -530,6 +544,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)
@@ -624,6 +639,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")
@@ -860,6 +876,81 @@ class TestControllerHandoff(unittest.TestCase):
self.assertIn("issue #182", skill)
class TestReviewMutationFinalReport(unittest.TestCase):
"""Final reports must list exactly one live review mutation."""
LOCK_ONE = {
"live_mutations": [{"pr_number": 203, "action": "request_changes"}],
"correction_authorized": False,
}
def test_single_mutation_report_complete(self):
report = (
"Review complete on PR #203.\n"
"Review mutations: one request_changes on #203.\n"
"Review decision: request_changes."
)
result = assess_review_mutation_final_report(report, self.LOCK_ONE)
self.assertTrue(result["complete"])
self.assertFalse(result["downgraded"])
def test_missing_mutation_details_downgraded(self):
result = assess_review_mutation_final_report(
"Review complete. No details.", self.LOCK_ONE
)
self.assertFalse(result["complete"])
self.assertTrue(result["downgraded"])
def test_two_mutations_require_correction_explanation(self):
lock = {
"live_mutations": [
{"pr_number": 203, "action": "approve"},
{"pr_number": 203, "action": "request_changes"},
],
"correction_authorized": True,
"correction_reason": "operator approved correcting mistaken approve",
}
incomplete = assess_review_mutation_final_report(
"Submitted approve then request_changes on #203.", lock
)
self.assertFalse(incomplete["complete"])
complete = assess_review_mutation_final_report(
"\n".join([
"Review mutations: approve (mistake), then request_changes (final).",
"Correction flow: operator approved correcting mistaken approve on PR #203.",
]),
lock,
)
self.assertTrue(complete["complete"])
def test_build_final_report_wires_review_mutation_from_report_text(self):
report_text = (
"Review complete on PR #203.\n"
"Review mutations: one request_changes on #203."
)
lock = {
"live_mutations": [{"pr_number": 203, "action": "request_changes"}],
"correction_authorized": False,
}
final = build_final_report(
checkout_proof=_good_checkout(),
inventory=_good_inventory(),
validation=_good_validation(),
contamination=_good_contamination(),
identity_eligible=True,
merge_performed=False,
issue_status_verified=True,
capability_evidence=_good_capability_evidence(),
sweep=_good_sweep(),
live_state=_good_live_state(),
role_boundary=_good_role_boundary(),
report_text=report_text,
review_decision_lock=lock,
)
self.assertTrue(final["review_mutation_complete"])
class TestPRInventoryTrustGate(unittest.TestCase):
"""Issue #194: unit tests for the PR inventory trust gate."""
@@ -1081,6 +1172,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)
@@ -1145,6 +1237,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()