diff --git a/mcp_server.py b/mcp_server.py index 0218616..7fc5cc4 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -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 { diff --git a/review_pr.py b/review_pr.py index b4fe4dc..48b6168 100755 --- a/review_pr.py +++ b/review_pr.py @@ -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()) \ No newline at end of file diff --git a/review_proofs.py b/review_proofs.py index bd4b817..cdc31c4 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -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": [], + } diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index fd94eab..c24670c 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -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`, diff --git a/tests/conftest.py b/tests/conftest.py index e08e2df..41c72c4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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 diff --git a/tests/test_audit.py b/tests/test_audit.py index 2db3079..6594739 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -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) diff --git a/tests/test_llm_agent_sha.py b/tests/test_llm_agent_sha.py index 4c827b1..1a6ecff 100644 --- a/tests/test_llm_agent_sha.py +++ b/tests/test_llm_agent_sha.py @@ -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: diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 5c2806f..124ef0e 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -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() diff --git a/tests/test_permission_reports.py b/tests/test_permission_reports.py index b1a4d2d..5483ee1 100644 --- a/tests/test_permission_reports.py +++ b/tests/test_permission_reports.py @@ -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) diff --git a/tests/test_pr_queue_inventory.py b/tests/test_pr_queue_inventory.py index ab17370..4866f99 100644 --- a/tests/test_pr_queue_inventory.py +++ b/tests/test_pr_queue_inventory.py @@ -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"]) diff --git a/tests/test_review_pr.py b/tests/test_review_pr.py index 1344745..9db4a6f 100644 --- a/tests/test_review_pr.py +++ b/tests/test_review_pr.py @@ -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__": diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 8bd3276..847de9d 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -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()