diff --git a/capability_stop_terminal.py b/capability_stop_terminal.py index 70ff264..d7f9751 100644 --- a/capability_stop_terminal.py +++ b/capability_stop_terminal.py @@ -13,6 +13,8 @@ REVIEWER_CAPABILITY_TASKS = frozenset({ "review_pr", "merge_pr", "blind_pr_queue_review", + "pr_queue_cleanup", + "pr-queue-cleanup", "request_changes_pr", "approve_pr", }) diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index a5a51b2..f163ae6 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -749,3 +749,22 @@ scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md --push - [`credential-isolation.md`](credential-isolation.md) — credential handling. - [`release-workflows.md`](release-workflows.md) — release/merge workflow. - [`../README.md`](../README.md) — canonical config, thin launchers, the menu. + +## PR-only queue cleanup mode (#390) + +Use `pr-queue-cleanup` mode when the queue holds many open PRs and the goal +is to drain reviews deterministically. One run = one PR = one canonical +review (`skills/llm-project-workflow/workflows/pr-queue-cleanup.md`). + +* Cleanup runs are reviewer-role only (`pr_queue_cleanup` resolver task); + author sessions get `wrong_role_stop`. +* Forbidden in cleanup mode: issue claiming, branch creation, implementation + edits, new issue filing, and any second-PR review after a terminal + mutation. +* Terminal chain: stop after `REQUEST_CHANGES`; after `APPROVED` continue + only to same-PR merge with explicit per-PR operator authorization and + passing gates; stop after merge or merge blocker. +* Every run reports the next suggested PR without continuing to it; the next + PR requires a fresh run with fresh identity/capability/inventory proof. +* Use full `review-merge-pr` mode instead when the operator wants a single + targeted review, and `work-issue` mode for any authoring. diff --git a/final_report_validator.py b/final_report_validator.py index bff9467..b6cc93c 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -87,14 +87,34 @@ _ALREADY_LANDED_RE = re.compile( r"already[- ]landed|ancestor proof\s*:\s*passed|eligibility class\s*:\s*already_landed", re.IGNORECASE, ) +_ALREADY_LANDED_GATE_FIRED_RE = re.compile( + r"\b(?:fired|passed|true|yes|already_landed_reconcile_required)\b", + re.IGNORECASE, +) _ELIGIBLE_REVIEW_RE = re.compile( r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr", re.IGNORECASE, ) +_APPROVED_STATE_RE = re.compile(r"\bapproved?\b", re.IGNORECASE) +_MERGED_STATE_RE = re.compile(r"\bmerged\b", re.IGNORECASE) +_READY_TO_MERGE_STATE_RE = re.compile(r"\bready_to_merge\b|\bready to merge\b", re.IGNORECASE) +_NEGATED_STATE_RE = re.compile( + r"\b(?:not|no|none|n/a|never)\b|request_changes|not attempted", + re.IGNORECASE, +) _REVIEWED_LANDED_RE = re.compile( r"(?:reviewed|approved).{0,40}already[- ]landed|already[- ]landed.{0,40}(?:reviewed|eligible)", re.IGNORECASE | re.DOTALL, ) +_TARGET_BRANCH_UP_TO_DATE_RE = re.compile( + r"\btarget branch (?:is )?up[- ]to[- ]date\b", + re.IGNORECASE, +) +_TARGET_BRANCH_SHA_RE = re.compile( + r"target branch sha\s*:\s*[0-9a-f]{40}", + re.IGNORECASE, +) +_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE) _RECONCILE_STALE_FIELDS = ( "pr number opened", "pinned reviewed head", @@ -190,6 +210,54 @@ def _handoff_fields(report_text: str) -> dict[str, str]: return fields +def _already_landed_gate_fired(report_text: str) -> bool: + text = report_text or "" + fields = _handoff_fields(text) + + gate_value = fields.get("already-landed gate", "") + if gate_value and _ALREADY_LANDED_GATE_FIRED_RE.search(gate_value): + return True + + ancestor_value = fields.get("ancestor proof", "") + if re.search(r"\bpassed\b", ancestor_value, re.IGNORECASE): + return True + + eligibility_value = fields.get("eligibility class", "") + if "already_landed" in eligibility_value.lower(): + return True + + if re.search( + r"\bpr\b.{0,60}\balready[- ]landed\b|\balready[- ]landed\b.{0,60}\bpr\b", + text, + re.IGNORECASE | re.DOTALL, + ): + return True + + return bool(_ALREADY_LANDED_RE.search(text)) + + +def _positive_review_state_terms(fields: dict[str, str]) -> list[str]: + terms: list[str] = [] + for key, value in fields.items(): + lowered_key = key.lower() + lowered_value = value.lower() + if _NEGATED_STATE_RE.search(value): + continue + if lowered_key == "review decision" and _APPROVED_STATE_RE.search(value): + terms.append("APPROVED") + elif lowered_key == "merge result" and _MERGED_STATE_RE.search(value): + terms.append("MERGED") + elif _READY_TO_MERGE_STATE_RE.search(value): + terms.append("READY_TO_MERGE") + elif lowered_value in {"approved", "approve"}: + terms.append("APPROVED") + elif lowered_value == "merged": + terms.append("MERGED") + elif lowered_value == "ready_to_merge": + terms.append("READY_TO_MERGE") + return sorted(set(terms)) + + def _rule_shared_controller_handoff( report_text: str, task_kind: str, @@ -492,7 +560,7 @@ def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]: def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, str]]: text = report_text or "" - if not _ALREADY_LANDED_RE.search(text): + if not _already_landed_gate_fired(text): return [] if not _ELIGIBLE_REVIEW_RE.search(text): return [] @@ -507,6 +575,65 @@ def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, s ] +def _rule_reviewer_already_landed_state(report_text: str) -> list[dict[str, str]]: + text = report_text or "" + if not _already_landed_gate_fired(text): + return [] + + terms = _positive_review_state_terms(_handoff_fields(text)) + if not terms: + return [] + + return [ + validator_finding( + "reviewer.already_landed_review_state", + "block", + "Review decision", + "already-landed gate fired but final report claims " + f"{', '.join(terms)} state", + "stop normal review/merge; classify as ALREADY_LANDED_RECONCILE_REQUIRED", + ) + ] + + +def _rule_reviewer_target_branch_freshness( + report_text: str, + *, + action_log: list[dict] | None = None, +) -> list[dict[str, str]]: + text = report_text or "" + if not _TARGET_BRANCH_UP_TO_DATE_RE.search(text): + return [] + + fields = _handoff_fields(text) + fetch_reported = bool(_GIT_FETCH_RE.search(text)) or any( + _GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or "")) + for e in (action_log or []) + ) + target_sha_reported = bool(_TARGET_BRANCH_SHA_RE.search(text)) or any( + "target branch" in key and "sha" in key and _FULL_SHA_RE.search(value) + for key, value in fields.items() + ) + if fetch_reported and target_sha_reported: + return [] + + missing = [] + if not fetch_reported: + missing.append("fresh git fetch") + if not target_sha_reported: + missing.append("target branch SHA") + return [ + validator_finding( + "reviewer.target_branch_freshness_proof", + "block", + "Target branch SHA", + "target branch up-to-date claim lacks " + " and ".join(missing), + "fetch the target branch and report the exact target branch SHA before " + "claiming it is up to date", + ) + ] + + def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]: from review_proofs import _handoff_section_lines @@ -706,6 +833,8 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_main_checkout_baseline, _rule_reviewer_main_checkout_path, _rule_reviewer_already_landed_eligible, + _rule_reviewer_already_landed_state, + _rule_reviewer_target_branch_freshness, _rule_reviewer_mutation_ledger, _rule_reviewer_review_mutation, ], @@ -875,4 +1004,4 @@ def assess_final_report_validator( "safe_next_action": _primary_safe_next_action(findings), "complete": grade == "A", "handoff_heading": HANDOFF_HEADING, - } \ No newline at end of file + } diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 2e90211..35867ad 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -479,6 +479,7 @@ import issue_lock_worktree # noqa: E402 import issue_claim_heartbeat # noqa: E402 import merged_cleanup_reconcile # noqa: E402 import reconciliation_workflow # noqa: E402 +import review_merge_state_machine # noqa: E402 import native_mcp_preference # noqa: E402 @@ -4257,6 +4258,14 @@ _GUIDE_RULES = { "inventory, summarize) needs no explicit authorization. Enforced " "fail closed via subagent_gate.assess_subagent_delegation and " "validate_subagent_report (#266)."), + "review_merge_state_machine": ( + "PR review/merge must follow the enforced state machine: PRECHECK → " + "INVENTORY → SELECT_PR → PIN_HEAD_SHA → CREATE_BRANCHES_WORKTREE → " + "VALIDATE → REVIEW_DECISION → APPROVE_OR_REQUEST_CHANGES → " + "PRE_MERGE_RECHECK → MERGE → POST_MERGE_REPORT. Failed upstream " + "states forbid downstream approve/merge. infra_stop and capability " + "blocks stop immediately. Blocked recovery handoffs must restart the " + "full workflow — never replay approve/merge commands (#290)."), "native_mcp_preference": ( "Gitea mutations must use native MCP tools first. When MCP is " "available, block shell scripts, direct API calls, WebFetch, " @@ -4992,6 +5001,65 @@ def _build_runtime_task_capabilities( @mcp.tool() +def gitea_assess_review_merge_state_machine( + target_state: str | None = None, + state_completion: dict[str, bool] | None = None, + pre_merge_gates: dict[str, bool] | None = None, + infra_stop: bool = False, + capability_blocked: bool = False, + recovery_handoff_text: str | None = None, + final_report_text: str | None = None, +) -> dict: + """Read-only: assess enforced PR review/merge workflow state (#290).""" + completion = state_completion or {} + blockers = review_merge_state_machine.assess_workflow_blockers( + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ) + result = { + "workflow": review_merge_state_machine.workflow_status( + completion, + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ), + "blockers": blockers, + "approve": review_merge_state_machine.can_approve( + completion, + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ), + "merge": review_merge_state_machine.can_merge( + completion, + pre_merge_gates=pre_merge_gates, + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ), + } + if target_state: + result["advancement"] = review_merge_state_machine.assess_state_advancement( + completion, + target_state=target_state, + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ) + if recovery_handoff_text is not None: + result["recovery_handoff"] = ( + review_merge_state_machine.assess_blocked_recovery_handoff( + recovery_handoff_text + ) + ) + if final_report_text is not None: + result["final_report_claims"] = ( + review_merge_state_machine.assess_final_report_state_claims( + final_report_text, + state_completion=completion, + approve_completed=bool(completion.get("APPROVE_OR_REQUEST_CHANGES")), + merge_completed=bool(completion.get("MERGE")), + ) + ) + return result + + @mcp.tool() def gitea_assess_gitea_operation_path( task: str, diff --git a/pr_queue_cleanup.py b/pr_queue_cleanup.py new file mode 100644 index 0000000..c1237aa --- /dev/null +++ b/pr_queue_cleanup.py @@ -0,0 +1,225 @@ +"""PR-only queue cleanup mode gates and report verifier (#390). + +Cleanup mode dispatches exactly one canonical review run per PR. It forbids +author-side mutations (issue claiming, branch creation, implementation edits, +issue filing) and enforces the terminal-mutation chain: stop after +``REQUEST_CHANGES``; after ``APPROVED`` continue only to same-PR merge when +merge is explicitly authorized for that PR and merge gates pass; stop after +merge or a merge blocker. The next PR always requires a fresh run. +""" + +from __future__ import annotations + +import re + +CLEANUP_WORKFLOW_PATH = "workflows/pr-queue-cleanup.md" + +# Author-side resolver tasks that must never run inside cleanup mode. +CLEANUP_FORBIDDEN_TASKS = frozenset({ + "claim_issue", + "mark_issue", + "lock_issue", + "create_issue", + "create_branch", + "push_branch", + "create_pr", + "commit_files", + "gitea_commit_files", + "address_pr_change_requests", +}) + +# Run-state outcomes for one cleanup dispatch. +STOP_AFTER_REQUEST_CHANGES = "STOP_AFTER_REQUEST_CHANGES" +STOP_AFTER_DECISION = "STOP_AFTER_DECISION" +CONTINUE_TO_SAME_PR_MERGE = "CONTINUE_TO_SAME_PR_MERGE" +STOP_APPROVED_NO_MERGE_AUTH = "STOP_APPROVED_NO_MERGE_AUTH" +STOP_APPROVED_MERGE_GATES_FAILED = "STOP_APPROVED_MERGE_GATES_FAILED" +STOP_AFTER_MERGE = "STOP_AFTER_MERGE" +STOP_AFTER_MERGE_BLOCKER = "STOP_AFTER_MERGE_BLOCKER" +STOP_GATE_NOT_PROVEN = "STOP_GATE_NOT_PROVEN" + +_TERMINAL_DECISIONS = frozenset({"approved", "request_changes", "comment", "skip"}) + + +def resolve_cleanup_run_state( + decision: str | None, + *, + merge_authorized_for_pr: bool = False, + merge_gates_passed: bool | None = None, + merge_completed: bool = False, + merge_blocker: bool = False, +) -> dict: + """Resolve what one cleanup run may do after its terminal review decision. + + Fail closed: unknown decisions stop the run with no further mutation. + """ + normalized = (decision or "").strip().lower() + + if normalized not in _TERMINAL_DECISIONS: + return { + "outcome": STOP_GATE_NOT_PROVEN, + "further_mutation_allowed": False, + "reasons": [ + f"unknown terminal decision {decision!r}; cleanup run stops " + "(fail closed)" + ], + } + + if normalized == "request_changes": + return { + "outcome": STOP_AFTER_REQUEST_CHANGES, + "further_mutation_allowed": False, + "reasons": ["REQUEST_CHANGES is terminal in cleanup mode"], + } + + if normalized in {"comment", "skip"}: + return { + "outcome": STOP_AFTER_DECISION, + "further_mutation_allowed": False, + "reasons": [f"{normalized} decision ends the cleanup run"], + } + + # normalized == "approved" + if merge_completed: + return { + "outcome": STOP_AFTER_MERGE, + "further_mutation_allowed": False, + "reasons": ["merge completed; cleanup run stops"], + } + if merge_blocker: + return { + "outcome": STOP_AFTER_MERGE_BLOCKER, + "further_mutation_allowed": False, + "reasons": ["merge blocker recorded; cleanup run stops"], + } + if not merge_authorized_for_pr: + return { + "outcome": STOP_APPROVED_NO_MERGE_AUTH, + "further_mutation_allowed": False, + "reasons": [ + "APPROVED without explicit per-PR merge authorization; " + "cleanup run stops" + ], + } + if merge_gates_passed is not True: + return { + "outcome": STOP_APPROVED_MERGE_GATES_FAILED, + "further_mutation_allowed": False, + "reasons": [ + "merge gates not proven passed; cleanup run stops before merge" + ], + } + return { + "outcome": CONTINUE_TO_SAME_PR_MERGE, + "further_mutation_allowed": True, + "reasons": [], + "allowed_mutation": "merge same PR only", + } + + +def check_cleanup_task_allowed(task: str) -> tuple[bool, list[str]]: + """Fail closed on any author-side mutation task inside cleanup mode.""" + normalized = (task or "").strip().lower() + if normalized in CLEANUP_FORBIDDEN_TASKS: + return False, [ + f"task '{normalized}' is forbidden in PR-only cleanup mode: " + "no issue claiming, branch creation, implementation edits, or " + "issue filing" + ] + return True, [] + + +_SELECTED_PR_RE = re.compile( + r"^\s*[-*]?\s*selected pr\s*:\s*#?(\d+)", re.IGNORECASE | re.MULTILINE +) +_NEXT_SUGGESTED_RE = re.compile( + r"^\s*[-*]?\s*next suggested pr\s*:", re.IGNORECASE | re.MULTILINE +) +_TERMINAL_MUTATION_RE = re.compile( + r"^\s*[-*]?\s*(?:review (?:decision|verdict)|terminal (?:review )?" + r"(?:decision|mutation))\s*:\s*" + r"(approved|request_changes|request changes|merged|comment)", + re.IGNORECASE | re.MULTILINE, +) +_PAGINATION_HINT_RE = re.compile( + r"inventory_complete|final page|has_more\s*[=:]\s*false|total[_ ]count", + re.IGNORECASE, +) +_MERGE_RESULT_RE = re.compile( + r"^\s*[-*]?\s*merge result\s*:\s*(?!none\b|not attempted\b)\S", + re.IGNORECASE | re.MULTILINE, +) +_MERGE_AUTHORIZED_RE = re.compile( + r"^\s*[-*]?\s*merge authorized(?: for pr)?\s*:\s*true", + re.IGNORECASE | re.MULTILINE, +) +_ISSUE_MUTATION_CLAIM_RE = re.compile( + r"^\s*[-*]?\s*issue mutations\s*:\s*(?!none\b)\S", + re.IGNORECASE | re.MULTILINE, +) +_BRANCH_MUTATION_CLAIM_RE = re.compile( + r"^\s*[-*]?\s*branch mutations\s*:\s*(?!none\b)\S", + re.IGNORECASE | re.MULTILINE, +) + + +def assess_pr_queue_cleanup_report(report_text: str) -> dict: + """Validate a PR-only cleanup run report (single dispatch, fail closed).""" + reasons: list[str] = [] + text = report_text or "" + + if CLEANUP_WORKFLOW_PATH not in text: + reasons.append( + f"report must cite the canonical cleanup workflow " + f"({CLEANUP_WORKFLOW_PATH})" + ) + + selected = _SELECTED_PR_RE.findall(text) + if len(selected) == 0: + reasons.append("report must name exactly one Selected PR") + elif len(set(selected)) > 1: + reasons.append( + "cleanup run selected multiple PRs " + f"({', '.join(sorted(set(selected)))}); one canonical review per " + "PR per run" + ) + + terminal = _TERMINAL_MUTATION_RE.findall(text) + if len(terminal) > 1: + reasons.append( + "multiple terminal review mutations reported in one cleanup run" + ) + + if not _PAGINATION_HINT_RE.search(text): + reasons.append( + "report missing PR inventory pagination proof " + "(inventory_complete / final page / total_count)" + ) + + if not _NEXT_SUGGESTED_RE.search(text): + reasons.append( + "report must include 'Next suggested PR' (without continuing to it)" + ) + + if _MERGE_RESULT_RE.search(text) and not _MERGE_AUTHORIZED_RE.search(text): + reasons.append( + "merge reported without explicit per-PR merge authorization " + "('Merge authorized: true')" + ) + + if _ISSUE_MUTATION_CLAIM_RE.search(text): + reasons.append("issue mutations are forbidden in PR-only cleanup mode") + if _BRANCH_MUTATION_CLAIM_RE.search(text): + reasons.append("branch mutations are forbidden in PR-only cleanup mode") + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "safe_next_action": ( + "proceed" if proven else + "fix the cleanup report: one PR, one terminal mutation, pagination " + "proof, next-suggested-PR field, and no issue/branch mutations" + ), + } diff --git a/review_merge_state_machine.py b/review_merge_state_machine.py new file mode 100644 index 0000000..33c9705 --- /dev/null +++ b/review_merge_state_machine.py @@ -0,0 +1,332 @@ +"""Enforced PR review/merge workflow state machine (#290). + +Review and merge must advance through explicit states. Any failed upstream +gate forbids downstream approve/merge mutations until the full workflow is +restarted after blockers clear. +""" + +from __future__ import annotations + +import re +from typing import Any + +REVIEW_MERGE_STATES: tuple[str, ...] = ( + "PRECHECK", + "INVENTORY", + "SELECT_PR", + "PIN_HEAD_SHA", + "CREATE_BRANCHES_WORKTREE", + "VALIDATE", + "REVIEW_DECISION", + "APPROVE_OR_REQUEST_CHANGES", + "PRE_MERGE_RECHECK", + "MERGE", + "POST_MERGE_REPORT", +) + +TERMINAL_BLOCKED_HEADING = ( + "PR review/merge workflow blocked. Restart the full workflow after blockers clear." +) + +_FORBIDDEN_RECOVERY_REPLAY_RE = re.compile( + r"\b(?:approve(?:\s+pr)?\s*#?\d+|merge(?:\s+pr)?\s*#?\d+|gitea_merge_pr|" + r"gitea_submit_pr_review|submit\s+approve|run\s+merge)\b", + re.IGNORECASE, +) +_RESTART_WORKFLOW_RE = re.compile( + r"restart(?:\s+the)?\s+full\s+workflow|rerun\s+the\s+full\s+workflow", + re.IGNORECASE, +) +_READY_TO_MERGE_RE = re.compile( + r"\bready\s+to\s+merge\b", + re.IGNORECASE, +) +_REVIEWED_VALIDATED_RE = re.compile( + r"\b(?:reviewed|validated|approval\s+submitted)\b", + re.IGNORECASE, +) + +_PRE_MERGE_REQUIRED_GATES = ( + "whoami_verified", + "profile_runtime_verified", + "merge_capability_verified", + "pr_refetched", + "reviewed_head_sha_unchanged", + "pr_mergeable", + "checks_passed", + "reviewer_not_author", + "worktree_clean", +) + + +def _clean(value: str | None) -> str: + return (value or "").strip() + + +def state_index(state: str) -> int: + name = _clean(state).upper() + try: + return REVIEW_MERGE_STATES.index(name) + except ValueError as exc: + raise ValueError(f"unknown review/merge state '{state}' (fail closed)") from exc + + +def downstream_states(from_state: str) -> list[str]: + idx = state_index(from_state) + return list(REVIEW_MERGE_STATES[idx + 1 :]) + + +def _completed_through(state_completion: dict[str, bool], state: str) -> bool: + return bool(state_completion.get(state)) + + +def _first_incomplete_state(state_completion: dict[str, bool]) -> str | None: + for state in REVIEW_MERGE_STATES: + if not _completed_through(state_completion, state): + return state + return None + + +def assess_workflow_blockers( + *, + infra_stop: bool = False, + capability_blocked: bool = False, + mcp_reconnect_failed: bool = False, + stale_capability_state: bool = False, +) -> dict[str, Any]: + """Return hard blockers that forbid all PR queue work (#290 AC3).""" + reasons: list[str] = [] + if infra_stop: + reasons.append("infra_stop is active; PR selection/review/merge is forbidden") + if capability_blocked: + reasons.append("gitea_resolve_task_capability returned blocked/stop_required") + if mcp_reconnect_failed: + reasons.append("MCP reconnect failed; stale session state cannot be reused") + if stale_capability_state: + reasons.append("stale MCP capability state detected after reconnect failure") + return { + "block": bool(reasons), + "reasons": reasons, + "forbidden_states": list(REVIEW_MERGE_STATES) if reasons else [], + "safe_next_action": ( + TERMINAL_BLOCKED_HEADING if reasons else "proceed with PRECHECK" + ), + } + + +def assess_state_advancement( + state_completion: dict[str, bool] | None, + *, + target_state: str, + infra_stop: bool = False, + capability_blocked: bool = False, +) -> dict[str, Any]: + """Fail closed when *target_state* is requested before upstream gates pass.""" + completion = dict(state_completion or {}) + target = _clean(target_state).upper() + blockers = assess_workflow_blockers( + infra_stop=infra_stop, + capability_blocked=capability_blocked, + ) + reasons = list(blockers["reasons"]) + + try: + target_idx = state_index(target) + except ValueError as exc: + return { + "target_state": target, + "allowed": False, + "block": True, + "reasons": [str(exc)], + "next_allowed_state": None, + "safe_next_action": TERMINAL_BLOCKED_HEADING, + } + + if blockers["block"]: + return { + "target_state": target, + "allowed": False, + "block": True, + "reasons": reasons, + "next_allowed_state": None, + "safe_next_action": blockers["safe_next_action"], + } + + next_allowed = _first_incomplete_state(completion) + if next_allowed is None: + allowed = target_idx == len(REVIEW_MERGE_STATES) - 1 + if not allowed: + reasons.append("workflow already completed through POST_MERGE_REPORT") + else: + allowed = state_index(next_allowed) >= target_idx + if not allowed: + reasons.append( + f"state '{target}' is forbidden until '{next_allowed}' completes" + ) + + return { + "target_state": target, + "allowed": allowed and not reasons, + "block": bool(reasons), + "reasons": reasons, + "next_allowed_state": next_allowed, + "completed_states": [s for s in REVIEW_MERGE_STATES if completion.get(s)], + "safe_next_action": ( + f"complete state '{next_allowed}' before advancing" + if next_allowed and not allowed + else ( + TERMINAL_BLOCKED_HEADING + if reasons + else f"advance to {target}" + ) + ), + } + + +def can_approve(state_completion: dict[str, bool] | None, **blocker_kwargs) -> dict[str, Any]: + """Approval requires all states through REVIEW_DECISION (#290 AC).""" + required = REVIEW_MERGE_STATES[: REVIEW_MERGE_STATES.index("REVIEW_DECISION") + 1] + completion = dict(state_completion or {}) + advance = assess_state_advancement( + completion, + target_state="APPROVE_OR_REQUEST_CHANGES", + **blocker_kwargs, + ) + missing = [s for s in required if not completion.get(s)] + reasons = list(advance["reasons"]) + if missing: + reasons.append( + "approval blocked: incomplete states: " + ", ".join(missing) + ) + block = bool(reasons) or advance["block"] + return { + "allowed": not block, + "block": block, + "reasons": reasons, + "missing_states": missing, + "safe_next_action": advance["safe_next_action"], + } + + +def can_merge( + state_completion: dict[str, bool] | None, + *, + pre_merge_gates: dict[str, bool] | None = None, + **blocker_kwargs, +) -> dict[str, Any]: + """Merge requires approve path plus fresh PRE_MERGE_RECHECK gates (#290 AC6).""" + completion = dict(state_completion or {}) + approve = can_approve(completion, **blocker_kwargs) + reasons = list(approve["reasons"]) + + if not completion.get("APPROVE_OR_REQUEST_CHANGES"): + reasons.append("merge blocked: APPROVE_OR_REQUEST_CHANGES not completed") + + gate_map = dict(pre_merge_gates or {}) + missing_gates = [g for g in _PRE_MERGE_REQUIRED_GATES if not gate_map.get(g)] + if missing_gates: + reasons.append( + "merge blocked: pre-merge gates incomplete: " + ", ".join(missing_gates) + ) + + advance = assess_state_advancement( + completion, + target_state="MERGE", + **blocker_kwargs, + ) + reasons.extend(advance["reasons"]) + + block = bool(reasons) + return { + "allowed": not block, + "block": block, + "reasons": reasons, + "missing_pre_merge_gates": missing_gates, + "safe_next_action": ( + "complete PRE_MERGE_RECHECK with fresh whoami/capability/PR re-fetch " + "before merge" + if block + else "merge allowed" + ), + } + + +def assess_blocked_recovery_handoff(report_text: str) -> dict[str, Any]: + """Blocked handoffs must not replay approve/merge commands (#290 AC4).""" + text = report_text or "" + reasons: list[str] = [] + if _FORBIDDEN_RECOVERY_REPLAY_RE.search(text): + reasons.append( + "blocked recovery handoff contains direct approve/merge replay command" + ) + if reasons and not _RESTART_WORKFLOW_RE.search(text): + reasons.append( + "blocked recovery handoff must direct operator to restart full workflow" + ) + return { + "block": bool(reasons), + "allowed": not reasons, + "reasons": reasons, + "safe_next_action": ( + "remove approve/merge replay commands and say to restart full workflow " + "after blockers clear" + if reasons + else "recovery handoff wording acceptable" + ), + } + + +def assess_final_report_state_claims( + report_text: str, + *, + state_completion: dict[str, bool] | None = None, + approve_completed: bool = False, + merge_completed: bool = False, +) -> dict[str, Any]: + """Reports must not claim reviewed/ready-to-merge without gate proof (#290 AC10).""" + text = report_text or "" + completion = dict(state_completion or {}) + reasons: list[str] = [] + + if _READY_TO_MERGE_RE.search(text) and not ( + merge_completed or completion.get("PRE_MERGE_RECHECK") + ): + reasons.append( + "report claims ready-to-merge without PRE_MERGE_RECHECK completion" + ) + + if _REVIEWED_VALIDATED_RE.search(text): + if not (approve_completed or completion.get("VALIDATE")): + reasons.append( + "report claims reviewed/validated without VALIDATE/APPROVE proof" + ) + + return { + "block": bool(reasons), + "allowed": not reasons, + "reasons": reasons, + "safe_next_action": ( + "remove stale reviewed/ready-to-merge claims unless gates passed" + if reasons + else "final report state claims consistent with workflow" + ), + } + + +def workflow_status( + state_completion: dict[str, bool] | None, + **blocker_kwargs, +) -> dict[str, Any]: + """Summarize current state-machine position for MCP/runtime reporting.""" + completion = dict(state_completion or {}) + blockers = assess_workflow_blockers(**blocker_kwargs) + next_state = _first_incomplete_state(completion) + return { + "states": list(REVIEW_MERGE_STATES), + "completed_states": [s for s in REVIEW_MERGE_STATES if completion.get(s)], + "next_required_state": next_state, + "workflow_complete": next_state is None, + "approve_allowed": can_approve(completion, **blocker_kwargs)["allowed"], + "merge_allowed": can_merge(completion, **blocker_kwargs)["allowed"], + "blockers": blockers, + } \ No newline at end of file diff --git a/review_proofs.py b/review_proofs.py index d3741f7..51b1cbf 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -14,6 +14,7 @@ is usable from prompts, harness assertions, and tests. The MCP-level gates here weakens or replaces them. """ +import hashlib import re import issue_duplicate_gate @@ -1704,11 +1705,26 @@ def build_final_report(checkout_proof, inventory, validation, contamination, if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text) else {"proven": True, "block": False, "reasons": [], "violations": []} ) + pr_queue_cleanup_report = ( + assess_pr_queue_cleanup_report(report_text) + if report_text and _PR_QUEUE_CLEANUP_REPORT_HINT.search(report_text) + else {"proven": True, "block": False, "reasons": [], "violations": []} + ) proof_wording = ( assess_proof_wording(report_text) if report_text else {"proven": True, "block": False, "reasons": [], "violations": []} ) + reconcile_linked_issue = ( + assess_reconcile_linked_issue_report(report_text) + if report_text and _RECONCILE_LINKED_ISSUE_HINT.search(report_text) + else {"proven": True, "block": False, "reasons": []} + ) + already_landed_handoff = ( + assess_already_landed_handoff_report(report_text) + if report_text + else {"proven": True, "block": False, "reasons": []} + ) inventory_worktree = ( assess_inventory_worktree_report(report_text) if report_text @@ -1875,6 +1891,21 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "queue-status report violates loaded workflow proof gates (#339)" ) downgrade_reasons.extend(queue_status_report.get("reasons", [])) + if not reconcile_linked_issue.get("proven"): + downgrade_reasons.append( + "reconciliation linked issue status lacks live fetch proof (#300)" + ) + downgrade_reasons.extend(reconcile_linked_issue.get("reasons", [])) + if not pr_queue_cleanup_report.get("proven"): + downgrade_reasons.append( + "PR-only cleanup report violates cleanup-mode proof gates (#390)" + ) + downgrade_reasons.extend(pr_queue_cleanup_report.get("reasons", [])) + if not already_landed_handoff.get("proven"): + downgrade_reasons.append( + "already-landed controller handoff has stale or inconsistent fields (#299)" + ) + downgrade_reasons.extend(already_landed_handoff.get("reasons", [])) if not inventory_worktree.get("proven"): downgrade_reasons.append( "inventory pagination or worktree fields inconsistent (#293)" @@ -1973,6 +2004,20 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "queue_status_violations": list( queue_status_report.get("violations") or [] ), + "reconcile_linked_issue_proven": bool(reconcile_linked_issue.get("proven")), + "reconcile_linked_issue_violations": list( + reconcile_linked_issue.get("reasons") or [] + ), + "pr_queue_cleanup_report_proven": bool( + pr_queue_cleanup_report.get("proven") + ), + "pr_queue_cleanup_violations": list( + pr_queue_cleanup_report.get("reasons") or [] + ), + "already_landed_handoff_proven": bool(already_landed_handoff.get("proven")), + "already_landed_handoff_violations": list( + already_landed_handoff.get("reasons") or [] + ), "inventory_worktree_proven": bool(inventory_worktree.get("proven")), "inventory_worktree_violations": list( inventory_worktree.get("reasons") or [] @@ -3357,6 +3402,67 @@ def assess_issue_filing_duplicate_summary( } +# ── Canonical review-workflow citation and version proof (Issue #296) ──────── +# +# The canonical PR review/merge workflow lives in +# skills/llm-project-workflow/workflows/review-merge-pr.md and is exposed +# via mcp_get_skill_guide. Review reports must prove they loaded it and +# cite the version hash used, so stale or shortened prompt copies are +# rejected. + +REVIEW_WORKFLOW_MARKERS = ( + "workflows/review-merge-pr.md", + "review-merge-pr.md", +) + + +def compute_workflow_hash(workflow_text): + """Return the short (12-hex) sha256 version hash of a workflow text.""" + digest = hashlib.sha256((workflow_text or "").encode("utf-8")) + return digest.hexdigest()[:12] + + +def assess_review_workflow_source(report_text, *, canonical_hash=None): + """#296: review reports must cite the canonical workflow and version. + + The report must reference the canonical workflow file and carry a + populated ``Workflow version`` (or ``Workflow hash``) field. When + *canonical_hash* — ``compute_workflow_hash`` of the current + workflows/review-merge-pr.md content — is supplied, the cited hash + must match it, rejecting stale copies. + + Returns {'complete', 'downgraded', 'reasons'}; fails closed. + """ + lower = (report_text or "").lower() + reasons = [] + + if not any(marker in lower for marker in REVIEW_WORKFLOW_MARKERS): + reasons.append( + "review report missing canonical workflow source citation " + "(workflows/review-merge-pr.md)" + ) + + fields = _report_labeled_fields(report_text) + version = fields.get("workflow version") or fields.get("workflow hash") + if not version or version.strip().lower().startswith(("none", "unknown")): + reasons.append( + "review report missing populated 'Workflow version' field " + "(version/hash of the canonical workflow used)" + ) + elif canonical_hash and canonical_hash.lower() not in version.lower(): + reasons.append( + f"review report cites stale workflow version '{version}'; " + f"current canonical hash is '{canonical_hash}' — reload the " + "workflow via mcp_get_skill_guide" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + def assess_create_issue_workflow_source(report_text: str) -> dict: """#337: create-issue reports must cite the canonical workflow file.""" lower = (report_text or "").lower() @@ -4800,6 +4906,25 @@ _QUEUE_STATUS_REPORT_HINT = re.compile( re.I, ) +_RECONCILE_LINKED_ISSUE_HINT = re.compile( + r"already[- ]landed|reconcil|linked issue(?:\s+live)?\s+status", + re.I, +) + +_PR_QUEUE_CLEANUP_REPORT_HINT = re.compile( + r"workflows/pr-queue-cleanup\.md|task:\s*pr-queue-cleanup|" + r"pr[- ]only queue cleanup", + re.I, +) + + +def assess_pr_queue_cleanup_report(report_text: str | None) -> dict: + """#390: validate PR-only cleanup final reports (one PR per run).""" + from pr_queue_cleanup import assess_pr_queue_cleanup_report as _assess + + return _assess(report_text or "") + + _GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I) _NOT_APPLICABLE_VALUE = re.compile( @@ -5214,6 +5339,20 @@ def assess_validation_worktree_edit_report(report_text, **kwargs): return _assess(report_text, **kwargs) +def assess_reconcile_linked_issue_report(report_text, **kwargs): + """#300: prove linked issue status was fetched live in reconciliation handoffs.""" + from reviewer_reconcile_linked_issue import assess_reconcile_linked_issue_report as _assess + + return _assess(report_text, **kwargs) + + +def assess_already_landed_handoff_report(report_text, **kwargs): + """#299: reject stale fields after the already-landed gate fires.""" + from reviewer_already_landed_handoff import assess_already_landed_handoff_report as _assess + + return _assess(report_text, **kwargs) + + def assess_inventory_worktree_report(report_text, **kwargs): """#293: prove inventory pagination and consistent worktree reporting.""" from reviewer_inventory_worktree import assess_inventory_worktree_report as _assess diff --git a/reviewer_already_landed_handoff.py b/reviewer_already_landed_handoff.py new file mode 100644 index 0000000..0642644 --- /dev/null +++ b/reviewer_already_landed_handoff.py @@ -0,0 +1,216 @@ +"""Already-landed controller handoff consistency verifier (#299).""" + +from __future__ import annotations + +import re +from typing import Any + +from review_proofs import git_ref_mutating_commands + +ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED" + +_HANDOFF_SECTION_RE = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M) + +_STALE_HANDOFF_FIELDS = ( + "pinned reviewed head", + "scratch worktree used", +) + +_LEGACY_MUTATIONS_NONE_RE = re.compile( + r"^\s*[-*]?\s*mutations\s*:\s*none\s*$", + re.I | re.M, +) +_LEGACY_WORKSPACE_NONE_RE = re.compile( + r"^\s*[-*]?\s*workspace\s+mutations\s*:\s*none\s*$", + re.I | re.M, +) + +_FORBIDDEN_REVIEW_STATES_RE = re.compile( + r"review decision\s*:\s*approved?\b|" + r"merge result\s*:\s*merged\b|" + r"\bready[_ ]to[_ ]merge\b", + re.I, +) + +_NARRATIVE_ELIGIBILITY_RE = re.compile( + r"eligibility class\s*:\s*([^\n]+)", + re.I, +) +_NARRATIVE_REVIEWED_SHA_RE = re.compile( + r"reviewed head sha\s*:\s*([^\n]+)", + re.I, +) +_NARRATIVE_WORKTREE_RE = re.compile( + r"review worktree used\s*:\s*([^\n]+)", + re.I, +) + + +def _handoff_field_map(report_text: str) -> dict[str, str]: + text = report_text or "" + match = _HANDOFF_SECTION_RE.search(text) + if not match: + return {} + fields: dict[str, str] = {} + for line in text[match.end() :].splitlines(): + stripped = line.strip().lstrip("-*").strip() + if ":" not in stripped: + continue + key, value = stripped.split(":", 1) + fields[key.strip().lower()] = value.strip() + return fields + + +def _narrative_before_handoff(report_text: str) -> str: + text = report_text or "" + match = _HANDOFF_SECTION_RE.search(text) + if not match: + return text + return text[: match.start()] + + +def _is_truthy(value: str) -> bool: + lowered = (value or "").strip().lower() + return lowered not in {"", "none", "n/a", "not applicable", "false", "no", "—", "-"} + + +def _gate_active(text: str, fields: dict[str, str], session: dict) -> bool: + if session.get("gate_fired") or session.get("already_landed_gate_fired"): + return True + eligibility = (fields.get("eligibility class") or "").upper() + if ALREADY_LANDED_STATE in eligibility: + return True + return ALREADY_LANDED_STATE.lower() in text.lower() + + +def assess_already_landed_handoff_report( + report_text: str, + *, + handoff_session: dict | None = None, + command_log: list | None = None, +) -> dict[str, Any]: + """Reject stale handoff fields and narrative/handoff drift after the gate (#299).""" + text = report_text or "" + session = dict(handoff_session or {}) + fields = _handoff_field_map(text) + reasons: list[str] = [] + + if not _gate_active(text, fields, session): + return { + "proven": True, + "block": False, + "reasons": [], + "gate_active": False, + "safe_next_action": "proceed", + } + + for stale_field in _STALE_HANDOFF_FIELDS: + if stale_field in fields: + reasons.append( + f"already-landed handoff must not include stale field " + f"'{stale_field.title()}' (#299)" + ) + + if _LEGACY_WORKSPACE_NONE_RE.search(text): + reasons.append( + "already-landed handoff must not include legacy " + "'Workspace mutations: None' (#299)" + ) + + ref_commands = git_ref_mutating_commands(command_log or session.get("command_log")) + mutations_observed = bool( + ref_commands + or session.get("mutations_observed") + or session.get("mcp_mutations") + or session.get("review_mutations") + ) + if mutations_observed and _LEGACY_MUTATIONS_NONE_RE.search(text): + reasons.append( + "already-landed handoff must not claim 'Mutations: None' when " + "git ref, MCP, review, merge, or cleanup mutations occurred (#299)" + ) + + git_ref_value = fields.get("git ref mutations", "").strip().lower() + if ref_commands and (not git_ref_value or git_ref_value == "none"): + reasons.append( + "git fetch/ref updates must be reported under Git ref mutations (#299)" + ) + + reviewed_sha = fields.get("reviewed head sha", "") + if reviewed_sha and _is_truthy(reviewed_sha): + reasons.append( + "already-landed handoff must use 'Reviewed head SHA: none' (#299)" + ) + + worktree_used = fields.get("review worktree used", "") + if worktree_used and _is_truthy(worktree_used): + reasons.append( + "already-landed handoff must set Review worktree used: false (#299)" + ) + + candidate_sha = fields.get("candidate head sha", "") + if not candidate_sha or candidate_sha.lower() in {"none", "n/a", "unknown"}: + reasons.append( + "already-landed handoff must include Candidate head SHA (#299)" + ) + + if _FORBIDDEN_REVIEW_STATES_RE.search(text): + reasons.append( + "already-landed handoff must not claim approved/merged/ready-to-merge (#299)" + ) + + narrative = _narrative_before_handoff(text) + handoff_eligibility = (fields.get("eligibility class") or "").strip() + narrative_eligibility = ( + _NARRATIVE_ELIGIBILITY_RE.search(narrative).group(1).strip() + if _NARRATIVE_ELIGIBILITY_RE.search(narrative) + else "" + ) + if handoff_eligibility and narrative_eligibility: + if handoff_eligibility.upper() != narrative_eligibility.upper(): + reasons.append( + "narrative Eligibility class disagrees with controller handoff (#299)" + ) + + narrative_reviewed = ( + _NARRATIVE_REVIEWED_SHA_RE.search(narrative).group(1).strip() + if _NARRATIVE_REVIEWED_SHA_RE.search(narrative) + else "" + ) + if narrative_reviewed and reviewed_sha: + if narrative_reviewed.lower() != reviewed_sha.lower(): + reasons.append( + "narrative Reviewed head SHA disagrees with controller handoff (#299)" + ) + + narrative_worktree = ( + _NARRATIVE_WORKTREE_RE.search(narrative).group(1).strip() + if _NARRATIVE_WORKTREE_RE.search(narrative) + else "" + ) + if narrative_worktree and worktree_used: + if narrative_worktree.lower() != worktree_used.lower(): + reasons.append( + "narrative Review worktree used disagrees with controller handoff (#299)" + ) + + git_ref_narrative = "git ref mutations" in narrative.lower() + if git_ref_narrative and git_ref_value: + if "none" in git_ref_value and ref_commands: + reasons.append( + "narrative and handoff disagree on git ref mutation reporting (#299)" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": list(dict.fromkeys(reasons)), + "gate_active": True, + "safe_next_action": ( + "emit canonical ALREADY_LANDED_RECONCILE_REQUIRED handoff without " + "stale review/merge fields; align narrative and controller handoff" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/reviewer_reconcile_linked_issue.py b/reviewer_reconcile_linked_issue.py new file mode 100644 index 0000000..8a01ee0 --- /dev/null +++ b/reviewer_reconcile_linked_issue.py @@ -0,0 +1,191 @@ +"""Reconciliation linked-issue live proof verifier (#300).""" + +from __future__ import annotations + +import re +from typing import Any + +_HANDOFF_SECTION_RE = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M) + +_LINKED_ISSUE_FIELD_RE = re.compile( + r"linked issue\s*:\s*(.+)$", + re.I | re.M, +) +_LINKED_ISSUE_STATUS_RE = re.compile( + r"linked issue(?:\s+live)?\s+status\s*:\s*(.+)$", + re.I | re.M, +) + +_STATUS_CLAIM_RE = re.compile( + r"\b(?:issue\s+#?\d+\s*\()?(open|closed|resolved)\b|" + r"issue\s+(?:is\s+)?(open|closed|resolved)\b|" + r"linked issue.*\b(open|closed|resolved)\b", + re.I, +) +_NOT_VERIFIED_RE = re.compile(r"not verified in this session", re.I) + +_LIVE_FETCH_PROOF_RE = re.compile( + r"gitea_view_issue|issue fetched live|live issue fetch|" + r"fetched linked issue.*(?:current|this) session|" + r"linked issue (?:fetch|proof).*(?:current|this) session|" + r"live linked issue proof", + re.I, +) + +_ISSUE_ACTION_RECOMMEND_RE = re.compile( + r"(?:recommend(?:ed)?|should|must)\s+(?:close|reopen|comment on)\s+" + r"(?:the\s+)?(?:linked\s+)?issue|" + r"(?:close|reopen|comment on)\s+linked issue\s+#?\d+", + re.I, +) +_CAPABILITY_PROOF_RE = re.compile( + r"capability proof|exact capability|gitea_close_issue|" + r"gitea_mark_issue|gitea_create_issue_comment|gitea_edit_issue", + re.I, +) + +_NO_LINKED_ISSUE_VALUES = frozenset({ + "", + "none", + "n/a", + "not applicable", + "no linked issue", + "unknown", +}) + + +def _handoff_field_map(report_text: str) -> dict[str, str]: + text = report_text or "" + match = _HANDOFF_SECTION_RE.search(text) + if not match: + return {} + fields: dict[str, str] = {} + for line in text[match.end() :].splitlines(): + stripped = line.strip().lstrip("-*").strip() + if ":" not in stripped: + continue + key, value = stripped.split(":", 1) + fields[key.strip().lower()] = value.strip() + return fields + + +def _extract_linked_issue_number(text: str, fields: dict[str, str]) -> int | None: + linked = fields.get("linked issue", "") + if linked.lower() in _NO_LINKED_ISSUE_VALUES: + return None + match = re.search(r"#?(\d+)", linked) + if match: + return int(match.group(1)) + field_match = _LINKED_ISSUE_FIELD_RE.search(text) + if field_match: + num = re.search(r"#?(\d+)", field_match.group(1)) + if num: + return int(num.group(1)) + return None + + +def _status_value(text: str, fields: dict[str, str]) -> str: + for key in ("linked issue live status", "linked issue status"): + if fields.get(key): + return fields[key] + match = _LINKED_ISSUE_STATUS_RE.search(text) + return match.group(1).strip() if match else "" + + +def _live_proof_present(text: str, session: dict, issue_number: int | None) -> bool: + if session.get("live_fetched") or session.get("verified_live"): + return True + if session.get("issue_fetch_proof"): + return True + if _LIVE_FETCH_PROOF_RE.search(text): + return True + if issue_number is not None: + pattern = re.compile( + rf"gitea_view_issue.*#?{issue_number}|" + rf"issue\s+#?{issue_number}.*(?:fetched|viewed).*(?:current|this) session", + re.I, + ) + if pattern.search(text): + return True + return False + + +def assess_reconcile_linked_issue_report( + report_text: str, + *, + reconcile_session: dict | None = None, +) -> dict[str, Any]: + """Validate linked issue status claims in reconciliation handoffs (#300).""" + text = report_text or "" + session = dict(reconcile_session or {}) + fields = _handoff_field_map(text) + reasons: list[str] = [] + + issue_number = session.get("linked_issue_number") + if issue_number is None: + issue_number = _extract_linked_issue_number(text, fields) + + if issue_number is None: + status = _status_value(text, fields) + if status and status.lower() not in _NO_LINKED_ISSUE_VALUES: + if _STATUS_CLAIM_RE.search(status): + reasons.append( + "linked issue status reported without identifying the linked issue (#300)" + ) + if not reasons: + return { + "proven": True, + "block": False, + "reasons": [], + "linked_issue_number": None, + "safe_next_action": "proceed", + } + + status = _status_value(text, fields) + status_lower = status.lower() + + if session.get("issue_missing"): + if status_lower and "not verified" not in status_lower: + reasons.append( + "missing linked issue must be reported as " + "'not verified in this session' (#300)" + ) + elif status: + if _NOT_VERIFIED_RE.search(status): + pass + elif _STATUS_CLAIM_RE.search(status): + if not _live_proof_present(text, session, issue_number): + reasons.append( + "linked issue open/closed/resolved status requires live " + "gitea_view_issue proof in the current session (#300)" + ) + elif status_lower not in _NO_LINKED_ISSUE_VALUES: + if not _live_proof_present(text, session, issue_number): + reasons.append( + "linked issue status claim requires live fetch proof or " + "'not verified in this session' (#300)" + ) + + if _ISSUE_ACTION_RECOMMEND_RE.search(text): + if not _CAPABILITY_PROOF_RE.search(text) and not session.get( + "issue_action_capability_proven" + ): + reasons.append( + "recommended linked issue close/reopen/comment requires " + "exact capability proof (#300)" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": list(dict.fromkeys(reasons)), + "linked_issue_number": issue_number, + "live_proof_present": _live_proof_present(text, session, issue_number), + "safe_next_action": ( + "fetch linked issue with gitea_view_issue in this session or " + "report 'Linked issue status: not verified in this session'" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/role_session_router.py b/role_session_router.py index 9c2b544..41b5134 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -57,6 +57,8 @@ REVIEWER_TASKS = frozenset({ "review_pr", "merge_pr", "blind_pr_queue_review", + "pr_queue_cleanup", + "pr-queue-cleanup", "request_changes_pr", "approve_pr", }) @@ -90,6 +92,8 @@ TASK_REQUIRED_ROLE = { "review_pr": "reviewer", "merge_pr": "reviewer", "blind_pr_queue_review": "reviewer", + "pr_queue_cleanup": "reviewer", + "pr-queue-cleanup": "reviewer", "request_changes_pr": "reviewer", "approve_pr": "reviewer", "reconcile_landed_pr": "author", diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 0ef7db1..dc2f961 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -22,6 +22,7 @@ workflow file. | Reconcile already-landed open PRs | [`workflows/reconcile-landed-pr.md`](workflows/reconcile-landed-pr.md) | [`schemas/reconcile-landed-final-report.md`](schemas/reconcile-landed-final-report.md) | | Create or update Gitea issues | [`workflows/create-issue.md`](workflows/create-issue.md) | [`schemas/create-issue-final-report.md`](schemas/create-issue-final-report.md) | | Work on an assigned issue / author code | [`workflows/work-issue.md`](workflows/work-issue.md) | [`schemas/work-issue-final-report.md`](schemas/work-issue-final-report.md) | +| PR-only queue cleanup (one canonical review per PR) | [`workflows/pr-queue-cleanup.md`](workflows/pr-queue-cleanup.md) | [`schemas/pr-queue-cleanup-final-report.md`](schemas/pr-queue-cleanup-final-report.md) | ## Universal rules @@ -50,6 +51,10 @@ changes, merge, implement fixes, create branches, commit, push, or create PRs. A run that starts in `work-issue` mode may not review, approve, request changes, merge, close PRs, or act as reviewer. +A run that starts in `pr-queue-cleanup` mode may not claim issues, create +branches, edit implementation files, file new issues, or review a second PR +after any terminal review mutation. + If the task requires a different mode, stop and produce a handoff for the correct workflow. @@ -156,6 +161,7 @@ Ready-to-copy task prompts live in [`templates/`](templates/): - [`start-issue.md`](templates/start-issue.md) — author work (loads `work-issue.md`) - [`review-pr.md`](templates/review-pr.md) — review (loads `review-merge-pr.md`) +- [`pr-queue-cleanup.md`](templates/pr-queue-cleanup.md) — one PR per cleanup run - [`merge-pr.md`](templates/merge-pr.md) — merge (loads `review-merge-pr.md`) - [`recover-bad-state.md`](templates/recover-bad-state.md) - [`reconcile-closed-not-merged-pr.md`](templates/reconcile-closed-not-merged-pr.md) diff --git a/skills/llm-project-workflow/schemas/pr-queue-cleanup-final-report.md b/skills/llm-project-workflow/schemas/pr-queue-cleanup-final-report.md new file mode 100644 index 0000000..c7abe8c --- /dev/null +++ b/skills/llm-project-workflow/schemas/pr-queue-cleanup-final-report.md @@ -0,0 +1,31 @@ +# PR-queue-cleanup final report schema + +One report per cleanup run (one run = one PR). Fields must all be present; +use `none` where nothing occurred. Validated by +`pr_queue_cleanup.assess_pr_queue_cleanup_report` (fail closed). + +* Task: pr-queue-cleanup +* Workflow source: workflows/pr-queue-cleanup.md (+ version/commit/hash) +* Repo: +* Role/profile: +* Identity: +* PR inventory pagination proof: (inventory_complete / final page / total_count) +* Queue ordering proof: +* Earlier PRs skipped: (with live per-PR proof) +* Selected PR: (exactly one) +* Pinned head SHA: +* Review decision: (single terminal decision) +* Merge authorized for PR: true/false (explicit per-PR operator authorization) +* Merge gates result: +* Merge result: (none / not attempted / merged SHA / blocker) +* Run stop point: (which §4 chain rule ended the run) +* Next suggested PR: (named, not continued to) +* File edits by reviewer: +* Worktree/index mutations: +* Git ref mutations: +* MCP/Gitea mutations: +* Issue mutations: none (required — forbidden in cleanup mode) +* Branch mutations: none (required — forbidden in cleanup mode) +* Read-only diagnostics: +* Blockers: +* Safe next action: (fresh run for the next PR) diff --git a/skills/llm-project-workflow/templates/pr-queue-cleanup.md b/skills/llm-project-workflow/templates/pr-queue-cleanup.md new file mode 100644 index 0000000..d3913f1 --- /dev/null +++ b/skills/llm-project-workflow/templates/pr-queue-cleanup.md @@ -0,0 +1,25 @@ +# Template: PR-only queue cleanup (one PR per run) + +Copy, fill the `<...>` fields, and paste as the task prompt. + +```text +Task: PR-only queue cleanup for . + +Load workflows/pr-queue-cleanup.md and workflows/review-merge-pr.md before any +mutation. Route pr_queue_cleanup through gitea_route_task_session (reviewer +profile required). + +Rules: +- One run = exactly one selected PR = one terminal review decision. +- Build full open-PR inventory with pagination proof before selection. +- Forbidden: issue claiming, branch creation, implementation edits, issue filing, + reviewing a second PR after a terminal mutation in this run. +- After REQUEST_CHANGES: stop. After APPROVED: merge only this PR and only if + operator explicitly authorized merge for this PR in this run. +- Report Next suggested PR without continuing to it. + +Operator PR list (optional): +Merge authorized for selected PR in this run: + +End with the pr-queue-cleanup final report schema. +``` \ No newline at end of file diff --git a/skills/llm-project-workflow/workflows/pr-queue-cleanup.md b/skills/llm-project-workflow/workflows/pr-queue-cleanup.md new file mode 100644 index 0000000..8a1be98 --- /dev/null +++ b/skills/llm-project-workflow/workflows/pr-queue-cleanup.md @@ -0,0 +1,86 @@ +--- +task_mode: pr-queue-cleanup +canonical: true +final_report_schema: ../schemas/pr-queue-cleanup-final-report.md +--- + +# PR-only queue cleanup workflow (canonical) + +**Task mode:** `pr-queue-cleanup` + +Reviewer-role mode for cleanup periods when the queue holds many open PRs. +Each run dispatches **exactly one** canonical review for **exactly one** PR, +then stops after any terminal review mutation. The next PR always requires a +new run with fresh identity, capability, and inventory proof. + +This mode composes with the canonical review workflow: for the selected PR, +load and follow [`workflows/review-merge-pr.md`](review-merge-pr.md) in full. +This file adds the cleanup-mode boundaries around that per-PR run; it does +not replace the review workflow. + +## 0. Load the canonical workflow first + +Load this file and `workflows/review-merge-pr.md` before any mutation and +report source, version/commit/hash, and any conflict with the operator +prompt. If either cannot be loaded, stop and produce a recovery handoff. + +## 1. Mode isolation — forbidden actions + +PR-only cleanup mode forbids, with no exceptions: + +* issue claiming (`claim_issue`, `mark_issue`, `lock_issue`) +* branch creation or push (`create_branch`, `push_branch`) +* implementation edits of any kind +* new issue filing (`create_issue`) +* PR creation (`create_pr`) and commit tools (`commit_files`) +* reviewing a second PR after any terminal review mutation + +`pr_queue_cleanup.check_cleanup_task_allowed` fails closed on these tasks. +If author-side work is needed, stop and hand off to `work-issue` mode in a +separate session. + +## 2. Identity, capability, and routing + +Route `pr_queue_cleanup` through `gitea_route_task_session` — reviewer role +required; author sessions receive `wrong_role_stop`. Prove `gitea.pr.review` +capability before selection. Merge additionally requires `gitea.pr.merge` +plus the explicit per-PR authorization in §4. + +## 3. Inventory and selection + +Build the complete open-PR inventory with pagination proof +(`inventory_complete`, final page, `total_count`) before any selection +claim. Select exactly one PR according to project queue ordering rules +(oldest-first unless the operator queue says otherwise), skipping earlier +PRs only with live per-PR proof. No multi-PR validation and no batch report +may substitute for per-PR proof. + +## 4. Terminal mutation chain + +`pr_queue_cleanup.resolve_cleanup_run_state` is the authority: + +* After `REQUEST_CHANGES` → the run stops. +* After `COMMENT` or a proof-backed skip → the run stops. +* After `APPROVED` → the run may continue **only** to same-PR merge, and only + when the operator explicitly authorized merge for that specific PR in this + run (`Merge authorized: true`) and every merge gate passes. +* After merge, or on any merge blocker → the run stops. +* Any terminal mutation targeting a PR other than the selected PR is a hard + stop and must be reported as a violation. + +## 5. Fresh run per PR + +The next PR requires a new run with fresh identity, capability, inventory, +and ordering proof. Do not carry pinned SHAs, eligibility classes, or +validation results across runs. + +## 6. Final report + +Use the schema in +[`schemas/pr-queue-cleanup-final-report.md`](../schemas/pr-queue-cleanup-final-report.md). +The report must include the **Next suggested PR** (from the proven ordering) +without continuing to it, exactly one **Selected PR**, the single terminal +decision, pagination proof, and `Issue mutations: none` / `Branch mutations: +none`. `pr_queue_cleanup.assess_pr_queue_cleanup_report` validates these +fields and fails closed on batch reviews, missing pagination proof, missing +next-suggested-PR, unauthorized merges, or any issue/branch mutation. diff --git a/task_capability_map.py b/task_capability_map.py index 4ea2ecf..1fc1a86 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -72,6 +72,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.review", "role": "reviewer", }, + "pr_queue_cleanup": { + "permission": "gitea.pr.review", + "role": "reviewer", + }, + "pr-queue-cleanup": { + "permission": "gitea.pr.review", + "role": "reviewer", + }, "request_changes_pr": { "permission": "gitea.pr.request_changes", "role": "reviewer", diff --git a/tests/test_final_report_validator.py b/tests/test_final_report_validator.py index f8e8848..72ef13d 100644 --- a/tests/test_final_report_validator.py +++ b/tests/test_final_report_validator.py @@ -25,11 +25,15 @@ def _review_handoff(**overrides): "- Files changed: review_proofs.py", "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203", "- Mutations: review only", - "- Workspace mutations: none", "- File edits by reviewer: none", "- Worktree/index mutations: none", "- Git ref mutations: git fetch prgs master", "- MCP/Gitea mutations: review submitted", + "- Review mutations: submitted request_changes review", + "- Merge mutations: none", + "- Cleanup mutations: none", + "- External-state mutations: none", + "- Read-only diagnostics: issue and PR metadata inspected", "- Current status: PR open", "- Blockers: none", "- Next: await author", @@ -190,11 +194,63 @@ class TestReviewPrRules(unittest.TestCase): ) ) + def test_already_landed_gate_blocks_review_terminal_states(self): + cases = { + "APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"), + "MERGED": ("- Merge result: not attempted", "- Merge result: MERGED"), + "READY_TO_MERGE": ("- Current status: PR open", "- Current status: READY_TO_MERGE"), + } + for state, (old, new) in cases.items(): + with self.subTest(state=state): + report = ( + _review_handoff() + .replace("- Reviewer eligibility: eligible", "- Reviewer eligibility: blocked") + .replace(old, new) + + "\n- Already-landed gate: fired" + + "\n- Ancestor proof: passed" + + "\n- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED" + ) + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + f["rule_id"] == "reviewer.already_landed_review_state" + for f in result["findings"] + ) + ) + + def test_target_branch_up_to_date_requires_fetch_and_sha(self): + report = ( + _review_handoff() + .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none") + + "\nTarget branch up to date." + ) + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + f["rule_id"] == "reviewer.target_branch_freshness_proof" + for f in result["findings"] + ) + ) + + def test_target_branch_up_to_date_with_fetch_and_sha_passes(self): + report = ( + _review_handoff() + + "\n- Target branch SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" + + "\nTarget branch up to date." + ) + result = assess_final_report_validator(report, "review_pr") + self.assertEqual(result["grade"], "A") + def test_git_fetch_must_not_be_readonly_only(self): report = ( _review_handoff() .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none") - + "\n- Read-only diagnostics: git fetch prgs master" + .replace( + "- Read-only diagnostics: issue and PR metadata inspected", + "- Read-only diagnostics: git fetch prgs master", + ) ) result = assess_final_report_validator( report, @@ -270,4 +326,4 @@ class TestEntryPoint(unittest.TestCase): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index e09c0f9..3b5e215 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -16,6 +16,7 @@ def test_all_workflow_files_exist(): "reconcile-landed-pr.md", "create-issue.md", "work-issue.md", + "pr-queue-cleanup.md", ): assert (SKILL_DIR / "workflows" / name).is_file(), name @@ -26,6 +27,7 @@ def test_skill_references_all_workflow_files(): "workflows/reconcile-landed-pr.md", "workflows/create-issue.md", "workflows/work-issue.md", + "workflows/pr-queue-cleanup.md", ): assert name in SKILL @@ -36,6 +38,7 @@ def test_skill_contains_mode_isolation_language(): assert "reconcile-landed-pr" in SKILL assert "create-issue" in SKILL assert "work-issue" in SKILL + assert "pr-queue-cleanup" in SKILL assert "Do not mix modes" in SKILL or "do not mix modes" in SKILL.lower() @@ -99,12 +102,24 @@ def test_work_issue_workflow_contract(): assert "## 21. PR creation rules" in text +def test_pr_queue_cleanup_workflow_contract(): + text = (SKILL_DIR / "workflows" / "pr-queue-cleanup.md").read_text( + encoding="utf-8" + ) + assert "canonical: true" in text + assert "exactly one" in text + assert "check_cleanup_task_allowed" in text + assert "resolve_cleanup_run_state" in text + assert "Next suggested PR" in text + + def test_final_report_schemas_exist(): for name in ( "review-merge-final-report.md", "reconcile-landed-final-report.md", "create-issue-final-report.md", "work-issue-final-report.md", + "pr-queue-cleanup-final-report.md", ): assert (SKILL_DIR / "schemas" / name).is_file(), name @@ -134,6 +149,20 @@ def test_reconcile_skill_registered_in_mcp_server(): assert "gitea_assess_already_landed_reconciliation" in joined +def test_pr_queue_cleanup_template_references_workflow(): + text = (SKILL_DIR / "templates" / "pr-queue-cleanup.md").read_text( + encoding="utf-8" + ) + assert "workflows/pr-queue-cleanup.md" in text + assert "pr_queue_cleanup" in text + + +def test_pr_queue_cleanup_verifier_exported(): + from review_proofs import assess_pr_queue_cleanup_report + + assert callable(assess_pr_queue_cleanup_report) + + def test_reviewer_fallback_verifier_exported(): from review_proofs import assess_reviewer_fallback_report @@ -182,6 +211,18 @@ def test_validation_worktree_edit_verifier_exported(): assert callable(assess_validation_worktree_edit_report) +def test_reconcile_linked_issue_verifier_exported(): + from review_proofs import assess_reconcile_linked_issue_report + + assert callable(assess_reconcile_linked_issue_report) + + +def test_already_landed_handoff_verifier_exported(): + from review_proofs import assess_already_landed_handoff_report + + assert callable(assess_already_landed_handoff_report) + + def test_inventory_worktree_verifier_exported(): from review_proofs import assess_inventory_worktree_report @@ -203,4 +244,23 @@ def test_mutation_categories_verifier_exported(): def test_worktree_ownership_verifier_exported(): from review_proofs import assess_worktree_ownership_report - assert callable(assess_worktree_ownership_report) \ No newline at end of file + assert callable(assess_worktree_ownership_report) + + +def test_pr_queue_cleanup_verifier_exported(): + from review_proofs import assess_pr_queue_cleanup_report + + assert callable(assess_pr_queue_cleanup_report) + + +def test_pr_queue_cleanup_workflow_contract(): + text = (SKILL_DIR / "workflows" / "pr-queue-cleanup.md").read_text(encoding="utf-8") + assert "canonical: true" in text + assert "task_mode: pr-queue-cleanup" in text + assert "## 1. Mode isolation" in text + assert "## 4. Terminal mutation chain" in text + assert "## 5. Fresh run per PR" in text + assert "exactly one" in text.lower() + assert "review-merge-pr.md" in text + assert "Next suggested PR" in text + assert "schemas/pr-queue-cleanup-final-report.md" in text diff --git a/tests/test_pr_queue_cleanup.py b/tests/test_pr_queue_cleanup.py new file mode 100644 index 0000000..e26a026 --- /dev/null +++ b/tests/test_pr_queue_cleanup.py @@ -0,0 +1,269 @@ +"""Tests for PR-only queue cleanup mode (#390).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from pr_queue_cleanup import ( + CLEANUP_FORBIDDEN_TASKS, + CLEANUP_WORKFLOW_PATH, + CONTINUE_TO_SAME_PR_MERGE, + STOP_AFTER_DECISION, + STOP_AFTER_MERGE, + STOP_AFTER_MERGE_BLOCKER, + STOP_AFTER_REQUEST_CHANGES, + STOP_APPROVED_MERGE_GATES_FAILED, + STOP_APPROVED_NO_MERGE_AUTH, + STOP_GATE_NOT_PROVEN, + assess_pr_queue_cleanup_report, + check_cleanup_task_allowed, + resolve_cleanup_run_state, +) +from review_proofs import assess_pr_queue_cleanup_report as proofs_assess +from role_session_router import REVIEWER_TASKS, route_task_session +from task_capability_map import required_permission, required_role + + +class TestCleanupCapabilityMapping(unittest.TestCase): + def test_cleanup_task_is_reviewer_role(self): + for key in ("pr_queue_cleanup", "pr-queue-cleanup"): + self.assertEqual(required_permission(key), "gitea.pr.review") + self.assertEqual(required_role(key), "reviewer") + + def test_cleanup_task_registered_as_reviewer_route(self): + self.assertIn("pr_queue_cleanup", REVIEWER_TASKS) + + def test_author_session_gets_wrong_role_stop(self): + result = route_task_session( + "pr_queue_cleanup", + active_profile="prgs-author", + active_role_kind="author", + allowed_in_current_session=False, + ) + self.assertEqual(result["route_result"], "wrong_role_stop") + self.assertFalse(result["downstream_allowed"]) + + def test_reviewer_session_allowed(self): + result = route_task_session( + "pr_queue_cleanup", + active_profile="prgs-reviewer", + active_role_kind="reviewer", + allowed_in_current_session=True, + ) + self.assertEqual(result["route_result"], "allowed_current_session") + self.assertTrue(result["downstream_allowed"]) + + +class TestCleanupTaskGate(unittest.TestCase): + def test_author_tasks_blocked(self): + for task in ( + "claim_issue", + "mark_issue", + "lock_issue", + "create_issue", + "create_branch", + "push_branch", + "create_pr", + "commit_files", + ): + allowed, reasons = check_cleanup_task_allowed(task) + self.assertFalse(allowed, task) + self.assertTrue(reasons, task) + + def test_review_task_allowed(self): + allowed, reasons = check_cleanup_task_allowed("review_pr") + self.assertTrue(allowed) + self.assertEqual(reasons, []) + + def test_forbidden_set_covers_issue_filing_and_impl(self): + self.assertIn("create_issue", CLEANUP_FORBIDDEN_TASKS) + self.assertIn("create_branch", CLEANUP_FORBIDDEN_TASKS) + self.assertIn("commit_files", CLEANUP_FORBIDDEN_TASKS) + + +class TestCleanupRunState(unittest.TestCase): + def test_request_changes_stops_run(self): + state = resolve_cleanup_run_state("request_changes") + self.assertEqual(state["outcome"], STOP_AFTER_REQUEST_CHANGES) + self.assertFalse(state["further_mutation_allowed"]) + + def test_comment_and_skip_stop_run(self): + for decision in ("comment", "skip"): + state = resolve_cleanup_run_state(decision) + self.assertEqual(state["outcome"], STOP_AFTER_DECISION) + self.assertFalse(state["further_mutation_allowed"]) + + def test_approved_without_merge_auth_stops(self): + state = resolve_cleanup_run_state("approved", merge_gates_passed=True) + self.assertEqual(state["outcome"], STOP_APPROVED_NO_MERGE_AUTH) + self.assertFalse(state["further_mutation_allowed"]) + + def test_approved_with_merge_auth_continues_to_merge(self): + state = resolve_cleanup_run_state( + "approved", + merge_authorized_for_pr=True, + merge_gates_passed=True, + ) + self.assertEqual(state["outcome"], CONTINUE_TO_SAME_PR_MERGE) + self.assertTrue(state["further_mutation_allowed"]) + self.assertEqual(state["allowed_mutation"], "merge same PR only") + + def test_approved_with_auth_but_failed_gates_stops(self): + state = resolve_cleanup_run_state( + "approved", + merge_authorized_for_pr=True, + merge_gates_passed=False, + ) + self.assertEqual(state["outcome"], STOP_APPROVED_MERGE_GATES_FAILED) + self.assertFalse(state["further_mutation_allowed"]) + + def test_merge_completed_stops(self): + state = resolve_cleanup_run_state( + "approved", + merge_authorized_for_pr=True, + merge_gates_passed=True, + merge_completed=True, + ) + self.assertEqual(state["outcome"], STOP_AFTER_MERGE) + self.assertFalse(state["further_mutation_allowed"]) + + def test_merge_blocker_stops(self): + state = resolve_cleanup_run_state( + "approved", + merge_authorized_for_pr=True, + merge_gates_passed=True, + merge_blocker=True, + ) + self.assertEqual(state["outcome"], STOP_AFTER_MERGE_BLOCKER) + self.assertFalse(state["further_mutation_allowed"]) + + def test_unknown_decision_fails_closed(self): + state = resolve_cleanup_run_state("ship_it") + self.assertEqual(state["outcome"], STOP_GATE_NOT_PROVEN) + self.assertFalse(state["further_mutation_allowed"]) + + +def _clean_report(**overrides): + lines = { + "task": "Task: pr-queue-cleanup", + "workflow": f"Workflow source: {CLEANUP_WORKFLOW_PATH} (hash abc123def456)", + "pagination": ( + "PR inventory pagination proof: inventory_complete=true, final " + "page, total_count 15" + ), + "selected": "Selected PR: #371", + "decision": "Review decision: approved", + "merge_auth": "Merge authorized for PR: false", + "merge_result": "Merge result: not attempted", + "next": "Next suggested PR: #372", + "issue_mut": "Issue mutations: none", + "branch_mut": "Branch mutations: none", + } + lines.update(overrides) + return "\n".join(v for v in lines.values() if v) + + +class TestCleanupReportVerifier(unittest.TestCase): + def test_clean_single_pr_report_passes(self): + result = assess_pr_queue_cleanup_report(_clean_report()) + self.assertTrue(result["proven"], result["reasons"]) + self.assertFalse(result["block"]) + + def test_approved_and_merged_with_authorization_passes(self): + report = _clean_report( + merge_auth="Merge authorized for PR: true", + merge_result="Merge result: merged 0123456789ab", + ) + result = assess_pr_queue_cleanup_report(report) + self.assertTrue(result["proven"], result["reasons"]) + + def test_request_changes_then_stop_passes(self): + report = _clean_report(decision="Review decision: request_changes") + result = assess_pr_queue_cleanup_report(report) + self.assertTrue(result["proven"], result["reasons"]) + + def test_multi_pr_selection_blocked(self): + report = _clean_report() + "\nSelected PR: #403\n" + result = assess_pr_queue_cleanup_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("multiple prs" in r.lower() for r in result["reasons"]), + result["reasons"], + ) + + def test_two_terminal_mutations_blocked(self): + report = _clean_report() + "\nReview decision: request_changes" + result = assess_pr_queue_cleanup_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("terminal review mutations" in r for r in result["reasons"]), + result["reasons"], + ) + + def test_missing_pagination_proof_blocked(self): + report = _clean_report( + pagination="PR inventory pagination proof: inventory listed" + ) + result = assess_pr_queue_cleanup_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("pagination" in r.lower() for r in result["reasons"]), + result["reasons"], + ) + + def test_missing_next_suggested_pr_blocked(self): + report = _clean_report(next="") + result = assess_pr_queue_cleanup_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("Next suggested PR" in r for r in result["reasons"]), + result["reasons"], + ) + + def test_merge_without_authorization_blocked(self): + report = _clean_report( + merge_result="Merge result: merged abc123abc123", + ) + result = assess_pr_queue_cleanup_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("authorization" in r.lower() for r in result["reasons"]), + result["reasons"], + ) + + def test_issue_mutations_forbidden(self): + report = _clean_report(issue_mut="Issue mutations: gitea_mark_issue") + result = assess_pr_queue_cleanup_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("issue mutations" in r.lower() for r in result["reasons"]), + result["reasons"], + ) + + def test_branch_mutations_forbidden(self): + report = _clean_report(branch_mut="Branch mutations: created feat/x") + result = assess_pr_queue_cleanup_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("branch mutations" in r.lower() for r in result["reasons"]), + result["reasons"], + ) + + def test_missing_workflow_citation_blocked(self): + report = _clean_report(workflow="Workflow source: none") + result = assess_pr_queue_cleanup_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("pr-queue-cleanup.md" in r for r in result["reasons"]), + result["reasons"], + ) + + def test_review_proofs_wrapper_matches_module(self): + direct = assess_pr_queue_cleanup_report(_clean_report()) + wrapped = proofs_assess(_clean_report()) + self.assertEqual(direct["proven"], wrapped["proven"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index 3545675..fca97aa 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -275,6 +275,28 @@ class TestResolveTaskCapability(unittest.TestCase): self.assertEqual(res["required_role_kind"], "author") self.assertIn("author", res["exact_safe_next_action"].lower()) + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_pr_queue_cleanup_author_profile_blocked(self, _auth, _api): + with patch.dict(os.environ, self._env("author-profile")): + res = mcp_server.gitea_resolve_task_capability( + task="pr_queue_cleanup", remote="prgs" + ) + self.assertEqual(res["required_role_kind"], "reviewer") + self.assertFalse(res["allowed_in_current_session"]) + self.assertTrue(res["stop_required"]) + + @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_resolve_pr_queue_cleanup_reviewer_profile_allowed(self, _auth, _api): + with patch.dict(os.environ, self._env("reviewer-profile")): + res = mcp_server.gitea_resolve_task_capability( + task="pr_queue_cleanup", remote="prgs" + ) + self.assertEqual(res["required_operation_permission"], "gitea.pr.review") + self.assertTrue(res["allowed_in_current_session"]) + self.assertFalse(res["stop_required"]) + @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api): diff --git a/tests/test_review_merge_state_machine.py b/tests/test_review_merge_state_machine.py new file mode 100644 index 0000000..b3930d6 --- /dev/null +++ b/tests/test_review_merge_state_machine.py @@ -0,0 +1,121 @@ +"""Tests for enforced PR review/merge state machine (#290).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import review_merge_state_machine as rmsm # noqa: E402 + +FULL_REVIEW_COMPLETION = {state: True for state in rmsm.REVIEW_MERGE_STATES} + + +def _completion_through(state_name: str) -> dict[str, bool]: + idx = rmsm.state_index(state_name) + return {state: i <= idx for i, state in enumerate(rmsm.REVIEW_MERGE_STATES)} + + +class TestWorkflowBlockers(unittest.TestCase): + def test_infra_stop_blocks_all_states(self): + result = rmsm.assess_workflow_blockers(infra_stop=True) + self.assertTrue(result["block"]) + self.assertEqual(result["forbidden_states"], list(rmsm.REVIEW_MERGE_STATES)) + + def test_infra_stop_forbids_advancement(self): + result = rmsm.assess_state_advancement( + {}, + target_state="INVENTORY", + infra_stop=True, + ) + self.assertTrue(result["block"]) + self.assertFalse(result["allowed"]) + + +class TestStateAdvancement(unittest.TestCase): + def test_cannot_skip_inventory(self): + result = rmsm.assess_state_advancement( + {"PRECHECK": True}, + target_state="SELECT_PR", + ) + self.assertTrue(result["block"]) + self.assertEqual(result["next_allowed_state"], "INVENTORY") + + def test_sequential_advance_allowed(self): + completion = _completion_through("INVENTORY") + result = rmsm.assess_state_advancement( + completion, + target_state="SELECT_PR", + ) + self.assertFalse(result["block"]) + self.assertTrue(result["allowed"]) + + +class TestApproveAndMergeGates(unittest.TestCase): + def test_approve_blocked_without_validate(self): + completion = _completion_through("PIN_HEAD_SHA") + result = rmsm.can_approve(completion) + self.assertTrue(result["block"]) + self.assertIn("VALIDATE", ",".join(result["missing_states"])) + + def test_approve_allowed_when_review_path_complete(self): + completion = _completion_through("REVIEW_DECISION") + result = rmsm.can_approve(completion) + self.assertFalse(result["block"]) + self.assertTrue(result["allowed"]) + + def test_merge_blocked_without_pre_merge_gates(self): + completion = _completion_through("APPROVE_OR_REQUEST_CHANGES") + result = rmsm.can_merge(completion) + self.assertTrue(result["block"]) + self.assertTrue(result["missing_pre_merge_gates"]) + + def test_merge_allowed_with_pre_merge_gates(self): + completion = _completion_through("PRE_MERGE_RECHECK") + gates = {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES} + result = rmsm.can_merge(completion, pre_merge_gates=gates) + self.assertFalse(result["block"]) + self.assertTrue(result["allowed"]) + + def test_merge_blocked_when_head_sha_gate_fails(self): + completion = _completion_through("PRE_MERGE_RECHECK") + gates = {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES} + gates["reviewed_head_sha_unchanged"] = False + result = rmsm.can_merge(completion, pre_merge_gates=gates) + self.assertTrue(result["block"]) + self.assertIn("reviewed_head_sha_unchanged", result["missing_pre_merge_gates"]) + + +class TestRecoveryHandoff(unittest.TestCase): + def test_blocked_handoff_without_replay_passes(self): + report = ( + "Blocked recovery handoff.\n" + "Restart the full workflow after infra_stop clears.\n" + "No approve/merge performed." + ) + result = rmsm.assess_blocked_recovery_handoff(report) + self.assertFalse(result["block"]) + + def test_blocked_handoff_with_merge_replay_fails(self): + report = "Please merge PR #12 now using gitea_merge_pr." + result = rmsm.assess_blocked_recovery_handoff(report) + self.assertTrue(result["block"]) + + +class TestFinalReportClaims(unittest.TestCase): + def test_ready_to_merge_claim_without_gates_fails(self): + result = rmsm.assess_final_report_state_claims( + "PR is ready to merge after validation.", + state_completion=_completion_through("VALIDATE"), + ) + self.assertTrue(result["block"]) + + def test_reviewed_claim_without_validate_fails(self): + result = rmsm.assess_final_report_state_claims( + "PR reviewed and looks good.", + state_completion={"PRECHECK": True}, + ) + self.assertTrue(result["block"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_review_workflow_source.py b/tests/test_review_workflow_source.py new file mode 100644 index 0000000..07552ca --- /dev/null +++ b/tests/test_review_workflow_source.py @@ -0,0 +1,108 @@ +"""Tests for canonical review-workflow citation and version proof (#296). + +PR review/merge reports must prove they loaded the canonical workflow +(workflows/review-merge-pr.md) and cite the workflow version/hash used; +a stale or missing hash fails so prompt drift is caught. +""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import ( # noqa: E402 + assess_review_workflow_source, + compute_workflow_hash, +) + + +CANONICAL_TEXT = "# review-merge-pr workflow v1\nstep one\n" +CANONICAL_HASH = compute_workflow_hash(CANONICAL_TEXT) + + +def _report(citation=True, version=CANONICAL_HASH): + lines = ["Task: review next eligible PR"] + if citation: + lines.append( + "Workflow source: skills/llm-project-workflow/" + "workflows/review-merge-pr.md (loaded via mcp_get_skill_guide)" + ) + if version is not None: + lines.append(f"- Workflow version: {version}") + return "\n".join(lines) + "\n" + + +class TestComputeWorkflowHash(unittest.TestCase): + def test_deterministic(self): + self.assertEqual( + compute_workflow_hash(CANONICAL_TEXT), + compute_workflow_hash(CANONICAL_TEXT), + ) + + def test_changes_with_content(self): + self.assertNotEqual( + compute_workflow_hash(CANONICAL_TEXT), + compute_workflow_hash(CANONICAL_TEXT + "extra rule\n"), + ) + + def test_short_hex(self): + h = compute_workflow_hash(CANONICAL_TEXT) + self.assertEqual(len(h), 12) + int(h, 16) # raises if not hex + + +class TestReviewWorkflowSource(unittest.TestCase): + def test_citation_and_matching_hash_passes(self): + result = assess_review_workflow_source( + _report(), canonical_hash=CANONICAL_HASH + ) + self.assertTrue(result["complete"]) + self.assertEqual(result["reasons"], []) + + def test_missing_citation_fails(self): + result = assess_review_workflow_source( + _report(citation=False), canonical_hash=CANONICAL_HASH + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("review-merge-pr" in r.lower() for r in result["reasons"]) + ) + + def test_missing_version_field_fails(self): + result = assess_review_workflow_source( + _report(version=None), canonical_hash=CANONICAL_HASH + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("workflow version" in r.lower() for r in result["reasons"]) + ) + + def test_stale_hash_fails(self): + result = assess_review_workflow_source( + _report(version="deadbeef0000"), canonical_hash=CANONICAL_HASH + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("stale" in r.lower() or "does not match" in r.lower() + for r in result["reasons"]) + ) + + def test_version_none_fails(self): + result = assess_review_workflow_source( + _report(version="none"), canonical_hash=CANONICAL_HASH + ) + self.assertFalse(result["complete"]) + + def test_without_canonical_hash_field_still_required(self): + # No canonical hash supplied (e.g. offline validation): citation + # and a populated version field are still required. + self.assertTrue( + assess_review_workflow_source(_report())["complete"] + ) + self.assertFalse( + assess_review_workflow_source(_report(version=None))["complete"] + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_reviewer_already_landed_handoff.py b/tests/test_reviewer_already_landed_handoff.py new file mode 100644 index 0000000..cf1c779 --- /dev/null +++ b/tests/test_reviewer_already_landed_handoff.py @@ -0,0 +1,131 @@ +"""Tests for already-landed handoff verifier (#299).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_already_landed_handoff import ( # noqa: E402 + ALREADY_LANDED_STATE, + assess_already_landed_handoff_report, +) + + +def _good_report() -> str: + return "\n".join([ + "## PR reconciliation summary", + f"Eligibility class: {ALREADY_LANDED_STATE}", + "Candidate head SHA: 2a544f582026b72a229d59a172c0a63ac4aaeaf9", + "Reviewed head SHA: none", + "Review worktree used: false", + "Ancestor proof: PR head is ancestor of prgs/master", + "", + "## Controller Handoff", + "- Selected PR: #278", + f"- Eligibility class: {ALREADY_LANDED_STATE}", + "- Candidate head SHA: 2a544f582026b72a229d59a172c0a63ac4aaeaf9", + "- Reviewed head SHA: none", + "- Target branch: master", + "- Target branch SHA: abc123def4567890abcdef1234567890abcdef12", + "- Review worktree used: false", + "- Git ref mutations: git fetch prgs master", + "- Review decision: none", + "- Merge result: none", + "- Linked issue status: open", + "- Safe next action: reconcile open PR and linked issue", + ]) + + +class TestAlreadyLandedHandoff(unittest.TestCase): + def test_clean_handoff_passes(self): + result = assess_already_landed_handoff_report(_good_report()) + self.assertTrue(result["proven"], result["reasons"]) + self.assertTrue(result["gate_active"]) + + def test_inactive_gate_skips_checks(self): + report = "## Controller Handoff\n- Selected PR: #12\n- Review decision: approve" + result = assess_already_landed_handoff_report(report) + self.assertTrue(result["proven"]) + self.assertFalse(result["gate_active"]) + + def test_rejects_stale_pinned_reviewed_head(self): + report = _good_report().replace( + "- Candidate head SHA:", + "- Pinned reviewed head: 2a544f582026b72a229d59a172c0a63ac4aaeaf9\n" + "- Candidate head SHA:", + ) + result = assess_already_landed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("pinned reviewed head" in r.lower() for r in result["reasons"])) + + def test_rejects_scratch_worktree_used(self): + report = _good_report() + "\n- Scratch worktree used: false" + result = assess_already_landed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("scratch worktree" in r.lower() for r in result["reasons"])) + + def test_rejects_mutations_none_with_git_fetch(self): + report = _good_report().replace( + "- Git ref mutations: git fetch prgs master", + "- Git ref mutations: none\n- Mutations: None", + ) + result = assess_already_landed_handoff_report( + report, + command_log=[{"command": "git fetch prgs master", "performed": True}], + ) + self.assertFalse(result["proven"]) + joined = " ".join(result["reasons"]).lower() + self.assertIn("mutations", joined) + self.assertIn("git ref", joined) + + def test_rejects_workspace_mutations_none(self): + report = _good_report() + "\n- Workspace mutations: None" + result = assess_already_landed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"])) + + def test_rejects_reviewed_head_sha_when_gate_fired(self): + report = _good_report().replace("- Reviewed head SHA: none", "- Reviewed head SHA: abc123") + result = assess_already_landed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("reviewed head sha" in r.lower() for r in result["reasons"])) + + def test_rejects_narrative_handoff_eligibility_mismatch(self): + report = _good_report().replace( + f"Eligibility class: {ALREADY_LANDED_STATE}", + "Eligibility class: NORMAL_REVIEW_CANDIDATE", + 1, + ) + result = assess_already_landed_handoff_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("eligibility class" in r.lower() for r in result["reasons"])) + + def test_session_gate_fired_without_text_marker(self): + report = "## Controller Handoff\n- Selected PR: #1\n- Pinned reviewed head: abc" + result = assess_already_landed_handoff_report( + report, + handoff_session={"gate_fired": True}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["gate_active"]) + + def test_infra_stop_style_normal_handoff_not_blocked(self): + report = "\n".join([ + "## Controller Handoff", + "- Selected PR: #12", + "- Review decision: request_changes", + "- Pinned reviewed head: abc123def4567890abcdef1234567890abcdef12", + ]) + result = assess_already_landed_handoff_report(report) + self.assertTrue(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_already_landed_handoff_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reviewer_reconcile_linked_issue.py b/tests/test_reviewer_reconcile_linked_issue.py new file mode 100644 index 0000000..1640f4f --- /dev/null +++ b/tests/test_reviewer_reconcile_linked_issue.py @@ -0,0 +1,117 @@ +"""Tests for reconciliation linked-issue live proof verifier (#300).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_reconcile_linked_issue import assess_reconcile_linked_issue_report # noqa: E402 + + +def _good_handoff() -> str: + return "\n".join([ + "## PR reconciliation summary", + "Fetched linked issue #263 live with gitea_view_issue in this session.", + "", + "## Controller Handoff", + "- Selected PR: #278", + "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED", + "- Linked issue: #263", + "- Linked issue live status: Issue #263 (open)", + "- Safe next action: reconcile open PR", + ]) + + +class TestReconcileLinkedIssueLiveProof(unittest.TestCase): + def test_live_fetch_with_open_status_passes(self): + result = assess_reconcile_linked_issue_report( + _good_handoff(), + reconcile_session={"live_fetched": True, "linked_issue_number": 263}, + ) + self.assertTrue(result["proven"], result["reasons"]) + self.assertEqual(result["linked_issue_number"], 263) + + def test_no_linked_issue_passes(self): + report = "\n".join([ + "## Controller Handoff", + "- Selected PR: #99", + "- Linked issue: none", + "- Linked issue live status: not applicable", + ]) + result = assess_reconcile_linked_issue_report(report) + self.assertTrue(result["proven"]) + self.assertIsNone(result["linked_issue_number"]) + + def test_closed_status_without_live_proof_blocks(self): + report = "\n".join([ + "## Controller Handoff", + "- Linked issue: #263", + "- Linked issue live status: Issue #263 (closed)", + ]) + result = assess_reconcile_linked_issue_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("live" in r.lower() for r in result["reasons"])) + + def test_not_verified_wording_passes_without_live_proof(self): + report = "\n".join([ + "## Controller Handoff", + "- Linked issue: #263", + "- Linked issue live status: not verified in this session", + ]) + result = assess_reconcile_linked_issue_report(report) + self.assertTrue(result["proven"], result["reasons"]) + + def test_missing_issue_requires_not_verified(self): + report = "\n".join([ + "## Controller Handoff", + "- Linked issue: #999", + "- Linked issue live status: Issue #999 (open)", + ]) + result = assess_reconcile_linked_issue_report( + report, + reconcile_session={"issue_missing": True, "linked_issue_number": 999}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("not verified" in r.lower() for r in result["reasons"])) + + def test_text_live_proof_without_session_lock_passes(self): + report = _good_handoff() + result = assess_reconcile_linked_issue_report(report) + self.assertTrue(result["proven"], result["reasons"]) + self.assertTrue(result["live_proof_present"]) + + def test_issue_action_recommendation_requires_capability_proof(self): + report = "\n".join([ + _good_handoff(), + "Recommended: close linked issue #263 after PR reconciliation.", + ]) + result = assess_reconcile_linked_issue_report( + report, + reconcile_session={"live_fetched": True}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("capability" in r.lower() for r in result["reasons"])) + + def test_issue_action_with_capability_proof_passes(self): + report = "\n".join([ + _good_handoff(), + "Recommended: close linked issue #263.", + "Capability proof: gitea_close_issue allowed for prgs-author.", + ]) + result = assess_reconcile_linked_issue_report( + report, + reconcile_session={"live_fetched": True}, + ) + self.assertTrue(result["proven"], result["reasons"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_reconcile_linked_issue_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py index 6204b6d..2099407 100644 --- a/tests/test_role_session_router.py +++ b/tests/test_role_session_router.py @@ -99,6 +99,24 @@ class TestRoleSessionRouter(unittest.TestCase): self.assertFalse(route["downstream_allowed"]) self.assertIn("Wrong role/session for reviewer task", route["message"]) + def test_pr_queue_cleanup_under_author_profile_wrong_role_stop(self): + with patch.dict(os.environ, self._env("prgs-author")): + route = mcp_server.gitea_route_task_session( + task_type="pr_queue_cleanup", remote="prgs" + ) + self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE) + self.assertFalse(route["downstream_allowed"]) + + @patch("mcp_server.api_request", return_value={"login": "sysadmin"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_pr_queue_cleanup_under_reviewer_profile_allowed(self, _auth, _api): + with patch.dict(os.environ, self._env("prgs-reviewer")): + route = mcp_server.gitea_route_task_session( + task_type="pr_queue_cleanup", remote="prgs" + ) + self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED) + self.assertTrue(route["downstream_allowed"]) + @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_issue_creation_blocked_after_reviewer_wrong_role_stop(