diff --git a/docs/architecture/glitchtip-to-gitea-workflow-design.md b/docs/architecture/glitchtip-to-gitea-workflow-design.md index 08e4571..3106b7d 100644 --- a/docs/architecture/glitchtip-to-gitea-workflow-design.md +++ b/docs/architecture/glitchtip-to-gitea-workflow-design.md @@ -1,21 +1,34 @@ -# GlitchTip-to-Gitea Issue Filing Workflow Design +# GlitchTip-to-Gitea Issue Filing Workflow Contract -- **Status:** Design (no implementation in this repo) -- **Issue:** #74 (parent umbrella: #75) +- **Status:** Contract for the library-only implementation in + `Scaled-Tech-Consulting/mcp-control-plane` (#57) +- **Issue:** #153 (supersedes #74/#78 as the Gitea-Tools tracker) - **Related:** #78 (deduplication design, child of #74) -- **Date:** 2026-07-02 +- **Date:** 2026-07-07 ## 1. Boundary and Orchestration * **GlitchTip-to-Gitea filing is NOT a GlitchTip MCP capability.** The `glitchtip-mcp` boundary remains strictly read-only per ADR-0001. -* The filing capability lives in an **orchestrator / runbook / release workflow**. It **composes** separate GlitchTip **read** tools (from the `glitchtip-mcp` server) and Gitea **issue** tools (from the `gitea-mcp` server). +* The filing capability lives in a **library-only orchestrator** in + `mcp-control-plane` and is not exposed through `glitchtip-mcp`. +* The orchestrator composes the real GlitchTip read path with Gitea + issue-write tools. It must not use mocked GlitchTip issue data in production + filing paths. * The orchestrator **must not centralize credentials** into a single server. The GlitchTip MCP holds only GlitchTip tokens, and the Gitea MCP holds only Gitea tokens. +* If a future MCP surface exposes filing, it must be a separate write-boundary + server/profile with explicit Gitea issue-write permission and the same audit + gates. It must not be added to the read-only `glitchtip-mcp` surface. ## 2. Invocation and Safety * **Explicit invocation only:** There is no automatic, unsupervised filing in phase 1. A human or an explicitly-triggered automation must initiate the workflow. * **Dry-run / Preview required:** The orchestrator must present a preview of the drafted Gitea issue (title, body, labels) and obtain explicit confirmation before calling the Gitea mutation tool to file the issue. -* **Gitea Profile Checks & Audit Logging:** The actual Gitea issue creation relies on `gitea-mcp`, and therefore inherently subjects the mutation to Gitea profile checks and audit logging as standard. +* **Gitea Profile Checks & Audit Logging:** The actual Gitea issue creation + relies on `gitea-mcp`, and therefore must pass Gitea profile checks and + fail-closed mutation audit before create/link/comment actions. +* **Deduplication before create:** The orchestrator must run the + GlitchTip-to-Gitea dedup/linking logic before any Gitea create action. Create, + link, and skip decisions must be represented in tests. ## 3. Gitea Issue Format @@ -51,6 +64,23 @@ To prevent PII or secret leakage into Gitea, the orchestrator and the underlying The principle is: **"Link, don't dump"**. The generated issue acts as an alert/pointer, while the raw context remains protected inside GlitchTip. -## 5. Deduplication and Linking (Deferred) +## 5. Deduplication and Linking -Deduplication logic (e.g. searching existing Gitea issues, managing GlitchTip issue IDs, and race condition handling) is specifically handled by **Issue #78** and will augment this design. +Deduplication logic (e.g. searching existing Gitea issues, managing GlitchTip +issue IDs, and race condition handling) is integrated into the library-only +filing orchestrator in `mcp-control-plane`. The orchestrator must reuse that +dedup/linking path instead of duplicating or bypassing it. + +## 6. Gitea-Tools Acceptance Contract for #153 + +Gitea-Tools does not host the filing implementation. This repository owns the +operator-facing contract: + +* `glitchtip-mcp` remains read-only and exposes only GlitchTip inspection tools. +* Filing is implemented in `mcp-control-plane`, not in this Gitea MCP runtime. +* Filing uses the real GlitchTip read path, not mocked issue data. +* Dedup runs before any Gitea create action. +* Create/link/skip decisions and audit failure are covered by orchestrator tests. +* Mutation audit fails closed before any Gitea create, link, or comment mutation. +* Any future MCP exposure requires a separate write-boundary server/profile and + must not add Gitea write credentials to `glitchtip-mcp`. diff --git a/docs/mcp-client-registration.md b/docs/mcp-client-registration.md index e83d5f3..f0477dd 100644 --- a/docs/mcp-client-registration.md +++ b/docs/mcp-client-registration.md @@ -95,7 +95,14 @@ back to shell commands, raw service APIs, or unrelated MCP servers. - Do not register `jenkins-write-mcp` until an operator approves a trigger profile; no shipped profile carries trigger capability by default. - `glitchtip-mcp` remains read-only. It must not file or mutate Gitea issues. -- GlitchTip-to-Gitea filing is a separate orchestrator that composes GlitchTip - read tools with Gitea issue-write tools. +- GlitchTip-to-Gitea filing is a separate library-only orchestrator in + `mcp-control-plane` that composes the real GlitchTip read path with Gitea + issue-write tools. +- The filing orchestrator must run GlitchTip/Gitea dedup before any Gitea + create action, and its create/link/skip decisions must be covered by tests. +- The filing orchestrator must fail closed on mutation-audit failure before + any Gitea create, link, or comment mutation. +- If filing is ever MCP-exposed, it must use a separate write-boundary + server/profile. It must not be exposed by `glitchtip-mcp`. - Gitea credentials never enter Jenkins or GlitchTip runtimes. - Jenkins and GlitchTip credentials never enter the Gitea MCP runtime. diff --git a/gitea_auth.py b/gitea_auth.py index 52a34c6..5e30213 100644 --- a/gitea_auth.py +++ b/gitea_auth.py @@ -420,6 +420,43 @@ def api_get_all(url, auth_header, *, limit=None, page_size=50, max_pages=100, return results +def api_fetch_page(url, auth_header, *, page=1, limit=50, **kwargs): + """Fetch one page from a Gitea list endpoint with explicit pagination metadata. + + Returns ``(items, pagination)`` where *pagination* includes ``has_more``, + ``next_page``, and ``is_final_page`` derived from the returned page length. + """ + page = max(1, int(page)) + limit = max(1, min(50, int(limit))) + page_url = _add_query(url, page=page, limit=limit) + data = api_request("GET", page_url, auth_header, **kwargs) + if data is None: + pagination = { + "page": page, + "per_page": limit, + "returned_count": 0, + "has_more": False, + "next_page": None, + "is_final_page": True, + } + return [], pagination + if not isinstance(data, list): + raise RuntimeError( + f"expected a list page from Gitea, got {type(data).__name__}" + ) + returned = len(data) + has_more = returned >= limit + pagination = { + "page": page, + "per_page": limit, + "returned_count": returned, + "has_more": has_more, + "next_page": page + 1 if has_more else None, + "is_final_page": not has_more, + } + return data, pagination + + def gitea_url(host, path): """Build a full URL for *host* and *path*, using http for loopback and https for others.""" if not path.startswith("/"): diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 3f99bd9..b5415eb 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -461,6 +461,7 @@ from gitea_auth import ( # noqa: E402 get_auth_header, api_request, api_get_all, + api_fetch_page, repo_api_url, get_profile, gitea_url, @@ -1119,6 +1120,56 @@ def gitea_create_pr( return _with_optional_url({"number": data["number"]}, data.get("html_url")) +def _format_list_pr_entry(pr: dict) -> dict: + head_obj = pr.get("head") or {} + base_obj = pr.get("base") or {} + entry = { + "number": pr["number"], + "title": pr["title"], + "state": pr["state"], + "head": head_obj.get("ref") if isinstance(head_obj, dict) else head_obj, + "base": base_obj.get("ref") if isinstance(base_obj, dict) else base_obj, + "mergeable": pr.get("mergeable"), + "updated_at": pr.get("updated_at"), + "head_sha": head_obj.get("sha") if isinstance(head_obj, dict) else None, + "author": (pr.get("user") or {}).get("login"), + "labels": [ + label["name"] + for label in pr.get("labels", []) + if isinstance(label, dict) and label.get("name") + ], + "body": pr.get("body") or "", + } + return _with_optional_url(entry, pr.get("html_url")) + + +def _list_prs_pagination_envelope( + prs: list[dict], + *, + page: int, + per_page: int, + pages_fetched: int, + is_final_page: bool, + has_more: bool, + next_page: int | None, + inventory_complete: bool, +) -> dict: + return { + "prs": prs, + "pagination": { + "page": page, + "per_page": per_page, + "returned_count": len(prs), + "has_more": has_more, + "next_page": next_page, + "is_final_page": is_final_page, + "pages_fetched": pages_fetched, + "inventory_complete": inventory_complete, + "total_count": len(prs) if inventory_complete else None, + }, + } + + @mcp.tool() def gitea_list_prs( state: str = "open", @@ -1126,7 +1177,9 @@ def gitea_list_prs( host: str | None = None, org: str | None = None, repo: str | None = None, -) -> list[dict]: + page: int | None = None, + per_page: int = 50, +) -> dict: """List pull requests on a Gitea repository. Args: @@ -1135,11 +1188,16 @@ def gitea_list_prs( host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. + page: Optional 1-based page for explicit single-page fetch. When + omitted, all pages are traversed and ``inventory_complete`` is set. + per_page: Page size (1–50; Gitea caps at 50). Returns: - List of dicts with 'number', 'title', 'state', 'head', 'base', - 'mergeable', 'updated_at' ('url' only with the reveal opt-in). - The additional 'updated_at' aids stale/conflicting queue detection. + Dict with ``prs`` (list of PR summaries) and ``pagination`` metadata + (``has_more``, ``next_page``, ``is_final_page``, ``inventory_complete``, + ``pages_fetched``, ``total_count`` when complete). Each PR entry carries + ``number``, ``title``, ``state``, ``head``, ``base``, ``mergeable``, + ``updated_at`` ('url' only with the reveal opt-in). """ allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool( "list_prs" @@ -1149,20 +1207,48 @@ def gitea_list_prs( h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/pulls?state={state}" - prs = api_get_all(url, auth) - results = [] - for pr in prs: - entry = { - "number": pr["number"], - "title": pr["title"], - "state": pr["state"], - "head": pr["head"]["ref"], - "base": pr["base"]["ref"], - "mergeable": pr.get("mergeable"), - "updated_at": pr.get("updated_at"), - } - results.append(_with_optional_url(entry, pr.get("html_url"))) - return results + + if page is not None: + raw_page, pagination = api_fetch_page( + url, auth, page=page, limit=per_page + ) + prs = [_format_list_pr_entry(pr) for pr in raw_page] + return _list_prs_pagination_envelope( + prs, + page=pagination["page"], + per_page=pagination["per_page"], + pages_fetched=1, + is_final_page=pagination["is_final_page"], + has_more=pagination["has_more"], + next_page=pagination["next_page"], + inventory_complete=False, + ) + + all_raw: list[dict] = [] + current_page = 1 + pages_fetched = 0 + last_meta: dict = {} + while pages_fetched < 100: + raw_page, last_meta = api_fetch_page( + url, auth, page=current_page, limit=per_page + ) + pages_fetched += 1 + all_raw.extend(raw_page) + if last_meta["is_final_page"]: + break + current_page += 1 + + prs = [_format_list_pr_entry(pr) for pr in all_raw] + return _list_prs_pagination_envelope( + prs, + page=1, + per_page=per_page, + pages_fetched=pages_fetched, + is_final_page=bool(last_meta.get("is_final_page")), + has_more=False, + next_page=None, + inventory_complete=bool(last_meta.get("is_final_page")), + ) @mcp.tool() @@ -1678,6 +1764,51 @@ def record_live_review_mutation(pr_number: int, action: str, review_id: int | No _save_review_decision_lock(lock) +def terminal_review_hard_stop_reasons(pr_number: int, operation: str) -> list[str]: + """Session hard-stop after a terminal live review mutation (#332). + + After a terminal verdict is consumed in this run, the only permitted + continuation is the merge sequence for the same PR that was approved. + A REQUEST_CHANGES (or an approval of a different PR) blocks every + further review/mark-ready/merge mutation. An operator-approved + correction (#211) re-opens the review path only — never a cross-PR + merge. + + *operation* is one of 'merge', 'mark_ready', or 'review'. + Returns [] when the operation may proceed. + """ + lock = _load_review_decision_lock() + if lock is None: + return [] + terminals = [ + m for m in (lock.get("live_mutations") or []) + if m.get("action") in _TERMINAL_REVIEW_ACTIONS + ] + if not terminals: + return [] + last = terminals[-1] + if ( + operation == "merge" + and last.get("action") == "approve" + and last.get("pr_number") == pr_number + ): + return [] + if operation in ("mark_ready", "review") and lock.get("correction_authorized"): + return [] + if last.get("action") == "approve": + guidance = ( + f"only the merge sequence for approved PR " + f"#{last.get('pr_number')} may continue" + ) + else: + guidance = "the session must stop and produce a final report" + return [ + "terminal review mutation already consumed in this run " + f"({last.get('action')} on PR #{last.get('pr_number')}); {guidance} " + "(fail closed, #332)" + ] + + # Patterns scrubbed from any surfaced error text so a credential can never leak. _SECRET_PREFIXES = ("token ", "Basic ") @@ -2093,6 +2224,9 @@ def gitea_mark_final_review_decision( } org = resolved_org repo = resolved_repo + hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready") + if hard_stop: + return {"marked_ready": False, "reasons": hard_stop} if lock.get("live_mutations") and not lock.get("correction_authorized"): return { "marked_ready": False, @@ -2110,6 +2244,33 @@ def gitea_mark_final_review_decision( f"{sorted(_REVIEW_ACTIONS)}" ], } + if action == "request_changes": + # Duplicate request-changes suppression (#332): an unresolved + # REQUEST_CHANGES at the current head must not be duplicated. + feedback = gitea_get_pr_review_feedback( + pr_number, remote=remote, org=org, repo=repo) + if not feedback.get("success"): + return { + "marked_ready": False, + "reasons": [ + "could not verify existing request-changes state for " + f"PR #{pr_number}; refusing to submit a possibly " + "duplicate REQUEST_CHANGES (fail closed, #332)" + ], + } + if ( + feedback.get("has_blocking_change_requests") + and not feedback.get("review_feedback_stale") + ): + return { + "marked_ready": False, + "reasons": [ + f"PR #{pr_number} already has an unresolved " + "REQUEST_CHANGES at the current head SHA; do not " + "duplicate the request-changes review (fail closed, " + "#332)" + ], + } lock["final_review_decision_ready"] = True lock["ready_pr_number"] = pr_number lock["ready_action"] = action @@ -2679,6 +2840,14 @@ def gitea_merge_pr( ) return result + # Gate 2b — session hard-stop after a terminal review mutation (#332): + # merge may only continue for the same PR that was just approved. + # Local in-process check; zero API calls. + hard_stop = terminal_review_hard_stop_reasons(pr_number, "merge") + if hard_stop: + reasons.extend(hard_stop) + return result + # Gate 3 — reuse #14 eligibility (identity + profile + merge-allowed + # author + self-merge block + open + mergeable/unknown fail-closed). # Read-only GETs only. @@ -2929,19 +3098,25 @@ def gitea_review_pr( if can_read and auth: inventory_attempted = True try: - url = f"{repo_api_url(h, o, r)}/pulls?state=open" - prs = api_get_all(url, auth) - prs_found_count = len(prs) - for pr in prs: - labels = [l["name"] for l in pr.get("labels", [])] + prs = gitea_list_prs( + state="open", + remote=remote, + host=h, + org=o, + repo=r, + ) + prs_list = prs.get("prs") or [] + prs_found_count = len(prs_list) + for pr in prs_list: + labels = list(pr.get("labels") or []) body_text = pr.get("body") or "" linked = [] # Simple extraction of linked issues from body for match in re.findall(r'#(\d+)', body_text): linked.append(f"#{match}") - pr_head_sha = (pr.get("head") or {}).get("sha") - pr_author = (pr.get("user") or {}).get("login") + pr_head_sha = pr.get("head_sha") + pr_author = pr.get("author") # Determine review block status req_non_author = False @@ -2952,8 +3127,8 @@ def gitea_review_pr( "number": pr["number"], "title": pr["title"], "author": pr_author, - "base": pr["base"]["ref"], - "head": pr["head"]["ref"], + "base": pr["base"], + "head": pr["head"], "mergeable": pr.get("mergeable"), "linked_issues": list(set(linked)), "labels": labels, @@ -2994,6 +3169,16 @@ def gitea_review_pr( if inventory_msg: report_lines.append(inventory_msg) else: + pagination = ( + prs.get("pagination") if isinstance(prs, dict) else {} + ) or {} + has_finality_metadata = bool( + pagination.get("inventory_complete") + or ( + pagination.get("is_final_page") + and not pagination.get("has_more") + ) + ) inventory_trust_gate = review_proofs.pr_inventory_trust_gate( prs if inventory_attempted else None, remote=remote, @@ -3002,7 +3187,7 @@ def gitea_review_pr( state="open", authenticated_profile=profile, local_remote_url=_local_git_remote_url(remote), - has_finality_metadata=True, + has_finality_metadata=has_finality_metadata, ) report_lines.extend( review_proofs.format_pr_inventory_trust_gate_report( diff --git a/review_proofs.py b/review_proofs.py index 87931e9..e08bd31 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -968,6 +968,90 @@ def assess_mutation_ledger_report( } +def assess_list_prs_pagination_proof( + list_prs_response: dict | list | None, + *, + inventory_complete_claimed: bool = False, +) -> dict: + """#340: verify ``gitea_list_prs`` pagination metadata proves inventory scope. + + Accepts the wrapped ``{"prs": [...], "pagination": {...}}`` response from + ``gitea_list_prs``. Legacy bare lists fail closed when completeness is + claimed. Single-page fetches may not claim full inventory unless + ``inventory_complete`` or ``is_final_page`` is true. + """ + reasons: list[str] = [] + if list_prs_response is None: + reasons.append("gitea_list_prs response missing; fail closed") + return {"proven": False, "block": True, "reasons": reasons, "pagination": None} + + if isinstance(list_prs_response, list): + pagination = None + if inventory_complete_claimed: + reasons.append( + "bare PR list lacks pagination metadata; cannot prove inventory " + "completeness" + ) + return { + "proven": not reasons, + "block": bool(reasons), + "reasons": reasons, + "pagination": pagination, + } + + if not isinstance(list_prs_response, dict): + reasons.append("gitea_list_prs response is not a dict or list; fail closed") + return {"proven": False, "block": True, "reasons": reasons, "pagination": None} + + prs = list_prs_response.get("prs") + pagination = list_prs_response.get("pagination") + if not isinstance(prs, list): + reasons.append("gitea_list_prs response missing 'prs' list") + if not isinstance(pagination, dict): + reasons.append("gitea_list_prs response missing 'pagination' metadata") + return { + "proven": False, + "block": True, + "reasons": reasons, + "pagination": pagination if isinstance(pagination, dict) else None, + } + + required_keys = ("page", "per_page", "returned_count", "has_more", "is_final_page") + for key in required_keys: + if key not in pagination: + reasons.append(f"pagination metadata missing '{key}'") + + if inventory_complete_claimed: + complete = pagination.get("inventory_complete") is True + final_page = pagination.get("is_final_page") is True and pagination.get("has_more") is False + if not (complete or final_page): + reasons.append( + "inventory completeness claimed but pagination metadata does not " + "prove final page (inventory_complete or is_final_page with " + "has_more=false required)" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "pagination": pagination, + } + + +def _prs_from_list_response(list_prs_response: list | dict | None) -> list | None: + """Normalize ``gitea_list_prs`` output to a PR list.""" + if list_prs_response is None: + return None + if isinstance(list_prs_response, list): + return list_prs_response + if isinstance(list_prs_response, dict): + prs = list_prs_response.get("prs") + return prs if isinstance(prs, list) else None + return None + + def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None, justification=None): """Assess reviewer/author role separation for blind queue workflows. @@ -1147,7 +1231,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination, role_boundary=None, review_mutation=None, report_text=None, review_decision_lock=None, controller_handoff=None, capability_proof=None, - sweep_proof=None, worktree_proof=None): + sweep_proof=None, worktree_proof=None, + baseline_validation=None, project_root=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. Combines the individual proof verdicts into the final-report fields the @@ -1174,6 +1259,18 @@ def build_final_report(checkout_proof, inventory, validation, contamination, ) empty_queue_report = assess_empty_queue_report(report_text) + if baseline_validation is None and report_text: + baseline_validation = assess_reviewer_baseline_validation_proof( + report_text=report_text, + project_root=project_root, + ) + elif baseline_validation is None: + baseline_validation = { + "proven": True, + "block": False, + "reasons": [], + "violations": [], + } contamination_status = contamination.get("status", "unknown") checkout_proven = bool(checkout_proof.get("proven")) @@ -1303,6 +1400,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "empty-queue report missing or failed trust-gate proof (#198)" ) downgrade_reasons.extend(empty_queue_report.get("reasons", [])) + if not baseline_validation.get("proven"): + downgrade_reasons.append( + "reviewer baseline validation proof missing or failed (#325)" + ) + downgrade_reasons.extend(baseline_validation.get("reasons", [])) merge_allowed = ( identity_eligible @@ -1314,9 +1416,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, # #179: no merge without a proven final live-state recheck. and live_state_proven and worktree_proven + and baseline_validation.get("proven") ) violations = [] + violations.extend(baseline_validation.get("violations", [])) if merge_performed and not merge_allowed: violations.append( "merge was performed/claimed although the proofs did not allow " @@ -1368,6 +1472,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination, else True ), "empty_queue_trust_gate_status": empty_queue_report.get("status"), + "baseline_validation_proven": bool(baseline_validation.get("proven")), + "baseline_validation_violations": list( + baseline_validation.get("violations") or [] + ), } @@ -1571,8 +1679,55 @@ HANDOFF_ROLE_FIELDS = { )), ("Related issues", ("related issues",)), ), + "create_issue": ( + ("Active profile", ("active profile",)), + ("Runtime context", ("runtime context",)), + ("Requested issue task", ("requested issue task",)), + ("Workflow source", ("workflow source",)), + ("Capability proof", ("capability proof",)), + ("Duplicate search terms", ("duplicate search terms",)), + ("Duplicate search pagination proof", ( + "duplicate search pagination proof", + )), + ("Duplicates found", ("duplicates found",)), + ("Issues created", ("issues created",)), + ("Issues commented", ("issues commented",)), + ("Issues edited", ("issues edited",)), + ("Issues skipped as duplicates", ( + "issues skipped as duplicates", + )), + ("File edits by issue creator", ("file edits by issue creator",)), + ("Safety statement", ("safety statement",)), + ), } +CREATE_ISSUE_WORKFLOW_MARKERS = ( + "workflows/create-issue.md", + "create-issue.md", +) + +CREATE_ISSUE_TASK_MARKERS = ( + "create-issue", + "create issue", +) + +CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = ( + ("Selected issue", ("selected issue",)), + ("Branch name", ("branch name",)), + ("PR number", ("pr number",)), + ("PR URL", ("pr url",)), + ("Commit SHA", ("commit sha",)), + ("Push result", ("push result",)), + ("Baseline comparison", ("baseline comparison",)), + ("Selected PR", ("selected pr",)), + ("Review decision", ("review decision",)), + ("Merge result", ("merge result",)), + ("Merge preflight", ("merge preflight",)), + ("Already-landed gate", ("already-landed gate",)), + ("Pinned reviewed head", ("pinned reviewed head",)), + ("Terminal review mutation", ("terminal review mutation",)), +) + def _handoff_section_lines(report_text): """Return the lines of the Controller Handoff section, or None.""" @@ -1623,6 +1778,11 @@ def assess_controller_handoff(report_text, role=None, local_edits=False): fields_dict[label] = v.strip() required = list(HANDOFF_BASE_FIELDS) + if role == "create_issue": + required = [ + field for field in required + if field[0] not in ("Workspace mutations", "Mutations") + ] required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ())) missing = [] @@ -1810,14 +1970,19 @@ def pr_inventory_trust_gate( "queue_target_lock": lock_status, } - if list_prs_response is None or not isinstance(list_prs_response, list): + prs_list = _prs_from_list_response(list_prs_response) + pagination_meta = ( + list_prs_response.get("pagination") + if isinstance(list_prs_response, dict) else {} + ) or {} + if prs_list is None: return { "status": "inventory_error", - "reasons": ["PR list response is invalid (not a list or None)"], + "reasons": ["PR list response is invalid (missing prs list or None)"], "corroborated": False, } - if len(list_prs_response) > 0: + if len(prs_list) > 0: return { "status": "trusted_nonempty", "reasons": [], @@ -1848,6 +2013,13 @@ def pr_inventory_trust_gate( corroborated = False if has_finality_metadata: corroborated = True + elif pagination_meta.get("inventory_complete") is True: + corroborated = True + elif ( + pagination_meta.get("is_final_page") is True + and pagination_meta.get("has_more") is False + ): + corroborated = True elif corroboration_open_pr_counter == 0: corroborated = True else: @@ -2647,6 +2819,141 @@ def assess_issue_filing_duplicate_summary( } +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() + reasons = [] + if not any(marker in lower for marker in CREATE_ISSUE_WORKFLOW_MARKERS): + reasons.append( + "create-issue report missing workflow source " + "(workflows/create-issue.md)" + ) + if not any(marker in lower for marker in CREATE_ISSUE_TASK_MARKERS): + reasons.append( + "create-issue report missing task mode declaration " + "(create-issue)" + ) + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + +def assess_create_issue_mode_isolation(report_text: str) -> dict: + """#337: reject work-issue/review-merge handoff fields in create-issue runs.""" + section = _handoff_section_lines(report_text) + reasons = [] + if section is None: + return { + "complete": True, + "downgraded": False, + "reasons": [], + } + + labels = [] + for line in section: + stripped = line.strip().lstrip("-*").strip() + if ":" in stripped: + labels.append(stripped.split(":", 1)[0].strip().lower()) + + for name, aliases in CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS: + if any( + label.startswith(alias) + for label in labels + for alias in aliases + ): + reasons.append( + f"create-issue handoff must not include cross-mode field " + f"'{name}'" + ) + + legacy_workspace = any( + label.startswith("workspace mutations") for label in labels + ) + precise_categories = any( + label.startswith("file edits by issue creator") + for label in labels + ) + if legacy_workspace and not precise_categories: + reasons.append( + "create-issue handoff must use precise mutation categories " + "instead of legacy 'Workspace mutations'" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + +def assess_create_issue_final_report( + report_text: str, + *, + issue_number: int | None = None, + issue_title: str | None = None, + action: str = "created", + mutations: list[str] | None = None, + resolved_capabilities: dict | None = None, + issues_searched: int | None = None, + closest_matches: list[dict] | None = None, + duplicate_result: dict | None = None, + performed_mutations: list[str] | None = None, +) -> dict: + """#337: composite verifier for create-issue final reports.""" + checks = { + "workflow_source": assess_create_issue_workflow_source(report_text), + "mode_isolation": assess_create_issue_mode_isolation(report_text), + "controller_handoff": assess_controller_handoff( + report_text, role="create_issue" + ), + "issue_reference": assess_issue_filing_issue_reference( + report_text, + issue_number=issue_number, + issue_title=issue_title, + action=action, + ), + "mutation_capability": assess_issue_filing_mutation_capability( + report_text, + mutations or performed_mutations, + resolved_capabilities, + ), + "mutation_scope": assess_issue_filing_mutation_scope( + report_text, + performed_mutations=performed_mutations or mutations, + ), + "duplicate_summary": assess_issue_filing_duplicate_summary( + report_text, + issues_searched=issues_searched, + closest_matches=closest_matches, + duplicate_result=duplicate_result, + ), + } + + reasons = [] + downgraded = False + for name, result in checks.items(): + verdict = result.get("verdict") + if verdict in ("missing", "incomplete"): + downgraded = True + reasons.extend(result.get("reasons") or []) + elif result.get("downgraded") or not result.get("complete", True): + downgraded = True + reasons.extend( + f"{name}: {r}" for r in (result.get("reasons") or []) + ) + + grade = "A" if not downgraded else "downgraded" + return { + "grade": grade, + "downgraded": downgraded, + "checks": checks, + "reasons": reasons, + "complete": not downgraded, + } + + def assess_issue_filing_final_report( report_text: str, *, @@ -2737,6 +3044,180 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict: } +# --------------------------------------------------------------------------- +# Reviewer baseline validation (#325) +# --------------------------------------------------------------------------- + +_TEST_VALIDATION_COMMAND_RE = re.compile( + r"(?:^|\s)(?:venv/)?(?:bin/)?(?:python\s+-m\s+)?" + r"(?:pytest|make\s+test|npm\s+test|cargo\s+test|go\s+test\b)", + re.IGNORECASE, +) + +_PREEXISTING_FAILURE_CLAIM_RE = re.compile( + r"(?:\bsame\s+as\s+master\b|" + r"\bpre-?existing(?:\s+(?:on\s+)?master)?\b|" + r"\bfailures?\s+(?:are|is)\s+pre-?existing\b|" + r"\bmaster\s+also\s+fails?\b|" + r"\bfull-?suite\s+failures?\s+(?:are|is)\s+pre-?existing\b)", + re.IGNORECASE, +) + +_REPORT_VALIDATION_CWD_RE = re.compile( + r"(?:working\s+directory|cwd|directory)\s*:\s*(\S+)", + re.IGNORECASE, +) + + +def _normalize_path(path: str) -> str: + return (path or "").replace("\\", "/").rstrip("/") + + +def _path_under_branches(path: str, project_root: str | None = None) -> bool: + """True when *path* is inside the project's ``branches/`` directory.""" + normalized = _normalize_path(path) + if not normalized: + return False + if "/branches/" in f"{normalized}/": + return True + if normalized.endswith("/branches"): + return True + if project_root: + root = _normalize_path(project_root) + if normalized.startswith(f"{root}/"): + rel = normalized[len(root) + 1 :] + return rel == "branches" or rel.startswith("branches/") + return False + + +def _is_test_validation_command(command: str) -> bool: + return bool(_TEST_VALIDATION_COMMAND_RE.search((command or "").strip())) + + +def _is_full_sha(value: str | None) -> bool: + return bool(value and _FULL_SHA.match((value or "").strip())) + + +def assess_reviewer_baseline_validation_proof( + *, + validation_runs: list[dict] | None = None, + baseline_proof: dict | None = None, + report_text: str | None = None, + project_root: str | None = None, +) -> dict: + """#325: reviewer validation and baseline comparison must use ``branches/``. + + *validation_runs* entries: ``command``, ``working_directory`` (or ``cwd``), + optional ``project_root``. + + *baseline_proof* keys when claiming pre-existing master failures: + ``worktree_path``, ``baseline_target_sha``, ``pr_head_sha``, + ``baseline_failures``, ``pr_failures``, ``failure_signatures_match``, + ``clean_before``, ``clean_after``. + """ + reasons: list[str] = [] + violations: list[str] = [] + text = report_text or "" + + runs = list(validation_runs or []) + pending_command = None + for raw_line in text.splitlines(): + line = raw_line.strip().lstrip("-*").strip() + lower = line.lower() + if lower.startswith("validation command:"): + pending_command = line.split(":", 1)[1].strip() + continue + cwd_match = _REPORT_VALIDATION_CWD_RE.search(line) + if cwd_match: + cwd = cwd_match.group(1).strip().rstrip(",.;") + command = pending_command or line + if _is_test_validation_command(command): + runs.append( + { + "command": command, + "working_directory": cwd, + "project_root": project_root, + } + ) + pending_command = None + + for run in runs: + command = (run.get("command") or "").strip() + if not command or not _is_test_validation_command(command): + continue + cwd = (run.get("working_directory") or run.get("cwd") or "").strip() + root = run.get("project_root") or project_root + if not cwd: + reasons.append( + "test validation command stated without working directory; " + "cannot prove branches-only execution (#325)" + ) + continue + if not _path_under_branches(cwd, root): + violations.append( + f"test validation ran outside branches/ worktree: cwd={cwd!r}" + ) + reasons.append( + "reviewer workflow must not run test suites in the main " + f"checkout; cwd {cwd!r} is not under branches/ (#325)" + ) + + claims_preexisting = bool(_PREEXISTING_FAILURE_CLAIM_RE.search(text)) + if claims_preexisting: + proof = baseline_proof or {} + worktree = (proof.get("worktree_path") or "").strip() + if not worktree or not _path_under_branches(worktree, project_root): + reasons.append( + "pre-existing master failure claimed without a baseline " + "worktree path under branches/ (#325)" + ) + if not _is_full_sha(proof.get("baseline_target_sha")): + reasons.append( + "pre-existing master failure claimed without " + "baseline_target_sha proof (#325)" + ) + if not _is_full_sha(proof.get("pr_head_sha")): + reasons.append( + "pre-existing master failure claimed without pr_head_sha " + "proof (#325)" + ) + baseline_failures = proof.get("baseline_failures") + pr_failures = proof.get("pr_failures") + if baseline_failures is None or pr_failures is None: + reasons.append( + "pre-existing master failure claimed without baseline and " + "PR failure listings (#325)" + ) + if proof.get("failure_signatures_match") is not True: + reasons.append( + "pre-existing master failure claimed without proven matching " + "failure signatures (#325)" + ) + if proof.get("clean_before") is not True: + reasons.append( + "baseline worktree clean-before validation not proven (#325)" + ) + if proof.get("clean_after") is not True: + reasons.append( + "baseline worktree clean-after validation not proven (#325)" + ) + + proven = not reasons and not violations + return { + "proven": proven, + "block": bool(violations), + "claims_preexisting": claims_preexisting, + "reasons": reasons, + "violations": violations, + "safe_next_action": ( + "create a clean baseline worktree under branches/, e.g. " + "branches/baseline-master-pr, and rerun validation there" + if not proven + else "proceed" + ), + } + + # --------------------------------------------------------------------------- # Identity disclosure (#305) # --------------------------------------------------------------------------- @@ -2823,3 +3304,126 @@ def assess_final_report_validator(report_text, task_kind, **kwargs): from final_report_validator import assess_final_report_validator as _validate return _validate(report_text, task_kind, **kwargs) + + +# --------------------------------------------------------------------------- +# Git ref mutations (#297) +# --------------------------------------------------------------------------- + +_GIT_REF_MUTATIONS_FIELD_RE = re.compile( + r"^\s*git ref mutations\s*:\s*(.+?)\s*$", re.I | re.M) +_GIT_WORKTREE_MUTATIONS_NONE_RE = re.compile( + r"^\s*git/worktree mutations\s*:\s*none\b", re.I | re.M) + +_GIT_REF_MUTATING_FIRST_WORDS = frozenset( + {"fetch", "pull", "push", "merge", "update-ref"}) +_GIT_REF_MUTATING_PREFIXES = ("remote update", "branch -d", "branch -D", + "reset --hard") + + +def _command_text(entry): + if isinstance(entry, dict): + return str(entry.get("command") or "").strip() + return str(entry or "").strip() + + +def _git_subcommand_line(command): + tokens = command.split() + if not tokens or tokens[0] != "git": + return None + return " ".join(tokens[1:]) + + +def git_ref_mutating_commands(command_log): + """Return logged commands that update git refs (#297). + + ``git fetch`` and friends edit no files but move refs; they are never + read-only diagnostics. + """ + out = [] + for entry in command_log or []: + command = _command_text(entry) + rest = _git_subcommand_line(command) + if rest is None: + continue + first = rest.split()[0] if rest.split() else "" + if first in _GIT_REF_MUTATING_FIRST_WORDS: + out.append(command) + continue + if any(rest.startswith(prefix) for prefix in _GIT_REF_MUTATING_PREFIXES): + out.append(command) + return out + + +def _fetch_targets(ref_commands): + """Extract ``/`` targets from git fetch commands.""" + targets = [] + for command in ref_commands: + rest = _git_subcommand_line(command) or "" + tokens = rest.split() + if not tokens or tokens[0] != "fetch": + continue + args = [t for t in tokens[1:] if not t.startswith("-")] + if len(args) >= 2: + targets.append(f"{args[0]}/{args[1]}") + return targets + + +def assess_git_ref_mutation_report(report_text, *, command_log=None): + """#297: ref updates are Git ref mutations, not read-only diagnostics. + + When the command log holds ref-updating commands, the report must carry + a non-none ``Git ref mutations`` entry (naming ``fetched + /`` for targeted fetches), may not claim ``Git/worktree + mutations: None``, and may not list the command as read-only. + """ + text = report_text or "" + lower = text.lower() + reasons = [] + ref_commands = git_ref_mutating_commands(command_log) + fetch_targets = _fetch_targets(ref_commands) + + if ref_commands: + field = _GIT_REF_MUTATIONS_FIELD_RE.search(text) + value = field.group(1).strip().lower() if field else None + if not value or value == "none": + reasons.append( + "ref-updating commands ran but the report has no " + "'Git ref mutations' entry" + ) + for target in fetch_targets: + if "fetched" not in lower or target.lower() not in lower: + reasons.append( + "git fetch ran; report must state " + f"'Git ref mutations: fetched {target}'" + ) + if _GIT_WORKTREE_MUTATIONS_NONE_RE.search(text): + reasons.append( + "'Git/worktree mutations: None' rejected: ref-updating " + "commands ran" + ) + for line in text.splitlines(): + if "read-only" not in line.lower(): + continue + for command in ref_commands: + if command.lower() in line.lower(): + reasons.append( + "read-only diagnostics may not include ref-updating " + f"command '{command}'" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "ref_mutating_commands": ref_commands, + "fetch_targets": fetch_targets, + } + + +def assess_non_mergeable_skip_proof(report_text, **kwargs): + """#322: require conflict proof when skipping non-mergeable PRs.""" + from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_mergeability_skip.py b/reviewer_mergeability_skip.py new file mode 100644 index 0000000..565de40 --- /dev/null +++ b/reviewer_mergeability_skip.py @@ -0,0 +1,159 @@ +"""Non-mergeable PR skip conflict-proof verifier for reviewer reports (#322).""" + +from __future__ import annotations + +import re +from typing import Any + +_FULL_SHA = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE) +_SHORT_SHA = re.compile(r"\b[0-9a-f]{7,39}\b", re.IGNORECASE) +_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE) +_MERGEABILITY_FALSE_RE = re.compile( + r"(?:mergeable\s*:\s*false|not mergeable|non[- ]mergeable|mergeability\s*:\s*false|" + r"mergeability result\s*:\s*false)", + re.IGNORECASE, +) +_CONFLICT_PROOF_RE = re.compile( + r"(?:merge-tree|git merge-tree|gitea_view_pr|gitea_check_pr_eligibility|" + r"conflict proof|mergeability tool|merge simulation|conflicting files?)", + re.IGNORECASE, +) +_CONFLICTING_FILES_RE = re.compile( + r"(?:conflicting files?|conflict files?)\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_HEAD_CHANGED_RE = re.compile( + r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|" + r"head changed since|head unchanged since)", + re.IGNORECASE, +) +_MERGEABILITY_UNVERIFIED_RE = re.compile( + r"\bMERGEABILITY_UNVERIFIED\b", +) + + +def _pr_section(text: str, pr_number: int) -> str: + """Return report lines likely describing one skipped PR.""" + lines = (text or "").splitlines() + chunks: list[str] = [] + capture = False + token = f"#{pr_number}" + for line in lines: + lower = line.lower() + if token in lower or f"pr {pr_number}" in lower or f"pr#{pr_number}" in lower.replace(" ", ""): + capture = True + chunks.append(line) + continue + if capture: + if _PR_NUMBER_RE.search(line) and token not in line.lower(): + break + if line.strip() == "" and len(chunks) > 3: + break + chunks.append(line) + if chunks: + return "\n".join(chunks) + return text or "" + + +def _has_head_sha(text: str, head_sha: str | None) -> bool: + if not head_sha: + return bool(_FULL_SHA.search(text) or _SHORT_SHA.search(text)) + head = head_sha.strip().lower() + if head in text.lower(): + return True + if len(head) >= 7 and head[:7] in text.lower(): + return True + return bool(_FULL_SHA.search(text)) + + +def assess_non_mergeable_skip_proof( + report_text: str, + *, + skipped_prs: list[dict] | None = None, +) -> dict[str, Any]: + """Validate skipped non-mergeable PR documentation in reviewer reports (#322).""" + text = report_text or "" + reasons: list[str] = [] + assessments: list[dict[str, Any]] = [] + + for entry in skipped_prs or []: + pr_number = entry.get("pr_number") + if pr_number is None: + reasons.append("skipped PR entry missing pr_number") + continue + pr_number = int(pr_number) + section = _pr_section(text, pr_number) + mergeable = entry.get("mergeable") + verified = entry.get("mergeability_verified") + classification = (entry.get("classification") or "").strip().upper() + head_sha = (entry.get("head_sha") or "").strip() or None + + item_reasons: list[str] = [] + + if f"#{pr_number}" not in text.lower() and f"pr {pr_number}" not in text.lower(): + item_reasons.append(f"skipped PR #{pr_number} not documented in final report") + + if mergeable is False or verified is False: + if not _MERGEABILITY_UNVERIFIED_RE.search(section) and verified is False: + if "MERGEABILITY_UNVERIFIED" not in classification: + item_reasons.append( + f"PR #{pr_number} conflict proof unavailable; report must classify " + "MERGEABILITY_UNVERIFIED" + ) + elif verified is not False: + if not _MERGEABILITY_FALSE_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing mergeability:false proof" + ) + if not _has_head_sha(section, head_sha): + item_reasons.append( + f"PR #{pr_number} skip missing current head SHA" + ) + if not _CONFLICT_PROOF_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing conflict proof command/tool" + ) + if entry.get("head_changed_since_prior_blocker") is not None: + if not _HEAD_CHANGED_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing head-changed-since-blocker proof" + ) + + if ( + entry.get("conflicting_files") + and not _CONFLICTING_FILES_RE.search(section) + and not any( + path.lower() in section.lower() + for path in (entry.get("conflicting_files") or []) + ) + ): + item_reasons.append( + f"PR #{pr_number} has conflicting files in session proof but " + "report omits file-level conflict proof" + ) + + proven = not item_reasons + assessments.append({ + "pr_number": pr_number, + "proven": proven, + "classification": classification or ( + "MERGEABILITY_UNVERIFIED" if verified is False else "NON_MERGEABLE_SKIPPED" + ), + "reasons": item_reasons, + }) + reasons.extend(item_reasons) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "downgraded": False, + "assessments": assessments, + "reasons": reasons, + "safe_next_action": ( + "document each skipped non-mergeable PR with head SHA, mergeability result, " + "conflict proof command, conflicting files, and head-change status" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_capability_stop_terminal.py b/tests/test_capability_stop_terminal.py index 5588f33..4100630 100644 --- a/tests/test_capability_stop_terminal.py +++ b/tests/test_capability_stop_terminal.py @@ -99,12 +99,16 @@ class TestCapabilityStopTerminal(unittest.TestCase): self.assertIn("Cannot perform reviewer task", str(ctx.exception)) self.assertIn("review_pr", str(ctx.exception)) - @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request", return_value={"login": "jcwalker3"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_list_prs_allowed_after_author_task_clears_stale_denial( - self, _auth, _api, _get_all, + self, _auth, _api, mock_fetch, ): + mock_fetch.return_value = ([], { + "page": 1, "per_page": 50, "returned_count": 0, + "has_more": False, "next_page": None, "is_final_page": True, + }) with patch.dict(os.environ, self._env()): denied = mcp_server.gitea_resolve_task_capability( task="review_pr", remote="prgs" @@ -120,7 +124,7 @@ class TestCapabilityStopTerminal(unittest.TestCase): self.assertFalse(capability_stop_terminal.is_active()) prs = mcp_server.gitea_list_prs(remote="prgs") - self.assertEqual(prs, []) + self.assertEqual(prs["prs"], []) @patch("mcp_server.api_request", return_value={"login": "jcwalker3"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") diff --git a/tests/test_external_mcp_registration_docs.py b/tests/test_external_mcp_registration_docs.py index 40c8eba..8225bb6 100644 --- a/tests/test_external_mcp_registration_docs.py +++ b/tests/test_external_mcp_registration_docs.py @@ -59,6 +59,22 @@ def test_registration_doc_preserves_boundaries(): assert phrase in text +def test_registration_doc_pins_glitchtip_filing_boundary(): + text = " ".join(_doc_text().split()) + for phrase in ( + "GlitchTip-to-Gitea filing is a separate library-only orchestrator in " + "`mcp-control-plane`", + "real GlitchTip read path", + "dedup before any Gitea create action", + "create/link/skip decisions", + "fail closed on mutation-audit failure before any Gitea create, link, " + "or comment mutation", + "must use a separate write-boundary server/profile", + "must not be exposed by `glitchtip-mcp`", + ): + assert phrase in text + + def test_registration_doc_separates_jenkins_trigger_from_read_surface(): text = _doc_text() assert "jenkins_trigger_build" in text @@ -100,3 +116,25 @@ def test_related_docs_keep_jenkins_trigger_off_read_surface(): assert "jenkins-write-mcp" in text assert "jenkins_mcp.write_server" in text assert "jenkins-mcp" in text + + +def test_glitchtip_filing_contract_doc_covers_issue_153_acceptance(): + text = ( + REPO_ROOT / "docs" / "architecture" / + "glitchtip-to-gitea-workflow-design.md" + ).read_text(encoding="utf-8") + flattened = " ".join(text.split()) + for phrase in ( + "**Status:** Contract for the library-only implementation in " + "`Scaled-Tech-Consulting/mcp-control-plane` (#57)", + "**Issue:** #153", + "not exposed through `glitchtip-mcp`", + "real GlitchTip read path", + "must not use mocked GlitchTip issue data", + "separate write-boundary server/profile", + "fail-closed mutation audit before create/link/comment actions", + "Deduplication before create", + "Create, link, and skip decisions must be represented in tests", + "Mutation audit fails closed", + ): + assert phrase in flattened diff --git a/tests/test_git_ref_mutation_report.py b/tests/test_git_ref_mutation_report.py new file mode 100644 index 0000000..c8ce29c --- /dev/null +++ b/tests/test_git_ref_mutation_report.py @@ -0,0 +1,101 @@ +"""Git ref mutations must be reported, never hidden as diagnostics (#297). + +``git fetch`` updates remote-tracking refs even though it edits no files. +Reports that ran fetch (or any ref-updating command) must carry a +``Git ref mutations`` entry and may not claim ``Git/worktree mutations: +None`` or classify the fetch as read-only diagnostics. +""" +import sys +import unittest + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import review_proofs + + +def _assess(report, commands): + return review_proofs.assess_git_ref_mutation_report( + report, command_log=commands) + + +class TestGitRefMutationDetection(unittest.TestCase): + def test_fetch_only_review_with_proper_ledger_passes(self): + report = "\n".join([ + "Git ref mutations: fetched prgs/master", + "Review decision: APPROVE", + ]) + res = _assess(report, ["git fetch prgs master"]) + self.assertTrue(res["proven"], res["reasons"]) + self.assertEqual(res["ref_mutating_commands"], ["git fetch prgs master"]) + + def test_fetch_without_ref_mutation_line_is_blocked(self): + report = "Review decision: APPROVE\nMutations: None\n" + res = _assess(report, ["git fetch prgs master"]) + self.assertFalse(res["proven"]) + self.assertTrue( + any("git ref mutations" in r.lower() for r in res["reasons"])) + + def test_fetch_with_remote_branch_requires_fetched_target_in_report(self): + report = "Git ref mutations: some refs updated\n" + res = _assess(report, ["git fetch prgs master"]) + self.assertFalse(res["proven"]) + self.assertTrue( + any("prgs/master" in r for r in res["reasons"])) + + def test_git_worktree_mutations_none_rejected_when_fetch_occurred(self): + report = "\n".join([ + "Git ref mutations: fetched prgs/master", + "Git/worktree mutations: None", + ]) + res = _assess(report, ["git fetch prgs master"]) + self.assertFalse(res["proven"]) + self.assertTrue( + any("git/worktree mutations" in r.lower() for r in res["reasons"])) + + def test_fetch_classified_as_read_only_diagnostics_rejected(self): + report = "\n".join([ + "Git ref mutations: fetched prgs/master", + "Read-only diagnostics: git fetch prgs master, git status", + ]) + res = _assess(report, ["git fetch prgs master"]) + self.assertFalse(res["proven"]) + self.assertTrue( + any("read-only" in r.lower() for r in res["reasons"])) + + def test_worktree_creation_is_not_a_ref_mutation(self): + report = "Worktree mutations: created branches/review-pr9\n" + res = _assess(report, ["git worktree add branches/review-pr9 prgs/master"]) + self.assertTrue(res["proven"], res["reasons"]) + self.assertEqual(res["ref_mutating_commands"], []) + + def test_merge_command_counts_as_ref_mutation(self): + report = "Review decision: APPROVE\n" + res = _assess(report, ["git merge --no-ff feat/x"]) + self.assertFalse(res["proven"]) + self.assertIn("git merge --no-ff feat/x", res["ref_mutating_commands"]) + + def test_cleanup_branch_delete_counts_as_ref_mutation(self): + report = "Cleanup mutations: deleted branch feat/x\n" + res = _assess(report, ["git branch -d feat/x"]) + self.assertFalse(res["proven"]) + self.assertIn("git branch -d feat/x", res["ref_mutating_commands"]) + + def test_no_mutation_blocked_handoff_passes(self): + report = "\n".join([ + "Git/worktree mutations: None", + "Current status: blocked handoff, no mutations performed", + ]) + res = _assess(report, ["git status --porcelain", "git log --oneline -1"]) + self.assertTrue(res["proven"], res["reasons"]) + self.assertEqual(res["ref_mutating_commands"], []) + + def test_command_log_dict_entries_supported(self): + report = "Git ref mutations: fetched prgs/master\n" + res = _assess(report, [{"command": "git fetch prgs master"}]) + self.assertTrue(res["proven"], res["reasons"]) + self.assertEqual( + res["ref_mutating_commands"], ["git fetch prgs master"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_list_prs_pagination.py b/tests/test_list_prs_pagination.py new file mode 100644 index 0000000..2f8b78a --- /dev/null +++ b/tests/test_list_prs_pagination.py @@ -0,0 +1,153 @@ +"""Tests for gitea_list_prs pagination metadata (#340).""" +import os +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import gitea_auth # noqa: E402 +from mcp_server import gitea_list_prs # noqa: E402 +from review_proofs import assess_list_prs_pagination_proof # noqa: E402 + +FAKE_AUTH = "Basic dGVzdDp0ZXN0" +URL = "https://gitea.example.com/api/v1/repos/o/r/pulls?state=open" + +SAMPLE_PR = { + "number": 1, + "title": "PR 1", + "state": "open", + "head": {"ref": "branch1"}, + "base": {"ref": "master"}, + "mergeable": True, + "updated_at": "2026-07-05T12:00:00Z", +} + + +class TestApiFetchPage(unittest.TestCase): + @patch("gitea_auth.api_request") + def test_final_page_metadata(self, mock_req): + mock_req.return_value = [SAMPLE_PR] + items, meta = gitea_auth.api_fetch_page(URL, FAKE_AUTH, page=1, limit=50) + self.assertEqual(len(items), 1) + self.assertFalse(meta["has_more"]) + self.assertTrue(meta["is_final_page"]) + self.assertIsNone(meta["next_page"]) + + @patch("gitea_auth.api_request") + def test_full_page_has_more(self, mock_req): + mock_req.return_value = [{"id": i} for i in range(2)] + items, meta = gitea_auth.api_fetch_page(URL, FAKE_AUTH, page=1, limit=2) + self.assertEqual(len(items), 2) + self.assertTrue(meta["has_more"]) + self.assertFalse(meta["is_final_page"]) + self.assertEqual(meta["next_page"], 2) + + +class TestGiteaListPrsPagination(unittest.TestCase): + @patch("mcp_server.api_fetch_page") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_explicit_page_returns_has_more(self, _auth, mock_fetch): + mock_fetch.return_value = ( + [SAMPLE_PR], + { + "page": 1, + "per_page": 50, + "returned_count": 1, + "has_more": False, + "next_page": None, + "is_final_page": True, + }, + ) + with patch.dict(os.environ, {}, clear=True): + result = gitea_list_prs(page=1, per_page=50) + self.assertEqual(len(result["prs"]), 1) + self.assertFalse(result["pagination"]["inventory_complete"]) + self.assertTrue(result["pagination"]["is_final_page"]) + + @patch("mcp_server.api_fetch_page") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_multi_page_fetch_all_marks_inventory_complete(self, _auth, mock_fetch): + mock_fetch.side_effect = [ + ( + [dict(SAMPLE_PR, number=i) for i in (1, 2)], + { + "page": 1, + "per_page": 2, + "returned_count": 2, + "has_more": True, + "next_page": 2, + "is_final_page": False, + }, + ), + ( + [dict(SAMPLE_PR, number=3)], + { + "page": 2, + "per_page": 2, + "returned_count": 1, + "has_more": False, + "next_page": None, + "is_final_page": True, + }, + ), + ] + with patch.dict(os.environ, {}, clear=True): + result = gitea_list_prs(per_page=2) + self.assertEqual([p["number"] for p in result["prs"]], [1, 2, 3]) + self.assertTrue(result["pagination"]["inventory_complete"]) + self.assertEqual(result["pagination"]["pages_fetched"], 2) + self.assertEqual(result["pagination"]["total_count"], 3) + + +class TestListPrsPaginationProof(unittest.TestCase): + def test_wrapped_final_page_passes_completeness_claim(self): + response = { + "prs": [], + "pagination": { + "page": 1, + "per_page": 50, + "returned_count": 0, + "has_more": False, + "next_page": None, + "is_final_page": True, + "inventory_complete": True, + "pages_fetched": 1, + "total_count": 0, + }, + } + result = assess_list_prs_pagination_proof( + response, inventory_complete_claimed=True + ) + self.assertTrue(result["proven"]) + + def test_bare_list_blocks_completeness_claim(self): + result = assess_list_prs_pagination_proof( + [{"number": 1}], inventory_complete_claimed=True + ) + self.assertFalse(result["proven"]) + self.assertIn("pagination metadata", " ".join(result["reasons"])) + + def test_single_page_without_complete_blocks_claim(self): + response = { + "prs": [{"number": 1}], + "pagination": { + "page": 1, + "per_page": 50, + "returned_count": 1, + "has_more": True, + "next_page": 2, + "is_final_page": False, + "inventory_complete": False, + "pages_fetched": 1, + "total_count": None, + }, + } + result = assess_list_prs_pagination_proof( + response, inventory_complete_claimed=True + ) + self.assertFalse(result["proven"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index e790004..9a96d17 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -121,4 +121,16 @@ def test_start_issue_template_references_work_issue_workflow(): def test_final_report_validator_exported(): from review_proofs import assess_final_report_validator - assert callable(assess_final_report_validator) \ No newline at end of file + assert callable(assess_final_report_validator) + + +def test_create_issue_final_report_verifier_exported(): + from review_proofs import assess_create_issue_final_report + + assert callable(assess_create_issue_final_report) + + +def test_non_mergeable_skip_verifier_exported(): + from review_proofs import assess_non_mergeable_skip_proof + + assert callable(assess_non_mergeable_skip_proof) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index f4597ca..394ab03 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -48,6 +48,21 @@ import mcp_server FAKE_AUTH = "Basic dGVzdDp0ZXN0" +_NO_BLOCKER_FEEDBACK = { + "success": True, + "has_blocking_change_requests": False, + "review_feedback_stale": False, +} + + +def _mark_request_changes_ready(pr_number=8, **kwargs): + """Mark a request_changes decision ready with the #332 duplicate- + suppression feedback fetch stubbed to 'no existing blocker'.""" + with patch("mcp_server.gitea_get_pr_review_feedback", + return_value=dict(_NO_BLOCKER_FEEDBACK)): + return gitea_mark_final_review_decision( + pr_number, "request_changes", **kwargs) + def _formal_review(reviewer, verdict, sha="abc123", review_id=1): return { @@ -400,40 +415,49 @@ class TestMirrorRefs(unittest.TestCase): # --------------------------------------------------------------------------- class TestListPRs(unittest.TestCase): - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_list_prs(self, _auth, mock_api): - mock_api.return_value = [ - { + def test_list_prs(self, _auth, mock_fetch): + mock_fetch.return_value = ( + [{ "number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1"}, "base": {"ref": "main"}, "html_url": "http://url1", "mergeable": True, "updated_at": "2026-07-05T12:00:00Z" - } - ] + }], + { + "page": 1, "per_page": 50, "returned_count": 1, + "has_more": False, "next_page": None, "is_final_page": True, + }, + ) with patch.dict(os.environ, {}, clear=True): result = gitea_list_prs() - self.assertEqual(len(result), 1) - self.assertEqual(result[0]["number"], 1) - self.assertEqual(result[0]["head"], "branch1") - self.assertNotIn("url", result[0]) + self.assertEqual(len(result["prs"]), 1) + self.assertEqual(result["prs"][0]["number"], 1) + self.assertEqual(result["prs"][0]["head"], "branch1") + self.assertNotIn("url", result["prs"][0]) + self.assertTrue(result["pagination"]["inventory_complete"]) # Staleness fields for queue reconciliation (#166) - self.assertEqual(result[0]["updated_at"], "2026-07-05T12:00:00Z") + self.assertEqual(result["prs"][0]["updated_at"], "2026-07-05T12:00:00Z") - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_list_prs_reveal_opt_in_includes_url(self, _auth, mock_api): - mock_api.return_value = [ - { + def test_list_prs_reveal_opt_in_includes_url(self, _auth, mock_fetch): + mock_fetch.return_value = ( + [{ "number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1"}, "base": {"ref": "main"}, "html_url": "http://url1", "mergeable": True, "updated_at": "2026-07-05T12:00:00Z" - } - ] + }], + { + "page": 1, "per_page": 50, "returned_count": 1, + "has_more": False, "next_page": None, "is_final_page": True, + }, + ) with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True): result = gitea_list_prs() - self.assertEqual(result[0]["url"], "http://url1") + self.assertEqual(result["prs"][0]["url"], "http://url1") # --------------------------------------------------------------------------- @@ -1830,7 +1854,7 @@ class TestSubmitPrReview(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_request_changes_succeeds_when_eligible(self, _auth, mock_api): - gitea_mark_final_review_decision(8, "request_changes", remote="prgs") + _mark_request_changes_ready(remote="prgs") mock_api.side_effect = [ {"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 9, "state": "REQUEST_CHANGES"}, @@ -1853,7 +1877,7 @@ class TestSubmitPrReview(unittest.TestCase): self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES") def test_request_changes_blocked_without_eligibility(self): - gitea_mark_final_review_decision(8, "request_changes", remote="prgs") + _mark_request_changes_ready(remote="prgs") with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \ patch("mcp_server.api_request") as mock_api: mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")] @@ -2135,7 +2159,7 @@ class TestSubmitPrReview(unittest.TestCase): reason="operator approved correcting mistaken approve", operator_authorized=True, ) - gitea_mark_final_review_decision(8, "request_changes", remote="prgs") + _mark_request_changes_ready(remote="prgs") second = gitea_submit_pr_review( pr_number=8, action="request_changes", remote="prgs", final_review_decision_ready=True, diff --git a/tests/test_pr_queue_inventory.py b/tests/test_pr_queue_inventory.py index 82998a4..aeb8bc4 100644 --- a/tests/test_pr_queue_inventory.py +++ b/tests/test_pr_queue_inventory.py @@ -16,38 +16,50 @@ import gitea_config FAKE_AUTH = "Basic dGVzdDp0ZXN0" + +def _final_page_fetch(items): + return (items, { + "page": 1, + "per_page": 50, + "returned_count": len(items), + "has_more": False, + "next_page": None, + "is_final_page": True, + }) + + class TestPRQueueInventory(unittest.TestCase): - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_author_profile_can_list_open_prs(self, mock_get_profile, _auth, mock_get_all): + def test_author_profile_can_list_open_prs(self, mock_get_profile, _auth, mock_fetch): mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read", "gitea.pr.comment"], "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [ + mock_fetch.return_value = _final_page_fetch([ {"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True} - ] + ]) result = gitea_list_prs(state="open", remote="prgs") - self.assertEqual(len(result), 1) - self.assertEqual(result[0]["number"], 1) + self.assertEqual(len(result["prs"]), 1) + self.assertEqual(result["prs"][0]["number"], 1) - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") @patch("gitea_config.is_runtime_switching_enabled", return_value=True) - def test_author_profile_can_produce_pending_reviewer_queue_report(self, mock_switch, mock_get_profile, _auth, mock_api, mock_get_all): + def test_author_profile_can_produce_pending_reviewer_queue_report(self, mock_switch, mock_get_profile, _auth, mock_api, mock_fetch): mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [ + mock_fetch.return_value = _final_page_fetch([ { "number": 1, "title": "PR 1", @@ -72,7 +84,7 @@ class TestPRQueueInventory(unittest.TestCase): "body": "", "updated_at": "2026-07-05T11:00:00Z" } - ] + ]) mock_api.return_value = {"login": "jcwalker3"} # whoami result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs") @@ -86,18 +98,18 @@ class TestPRQueueInventory(unittest.TestCase): # updated_at surfaced for reconciliation (#166) self.assertIn("Updated: 2026-07-05T10:00:00Z", result["message"]) - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_author_profile_never_calls_mutation_tools_during_inventory(self, mock_get_profile, _auth, mock_api, mock_get_all): + def test_author_profile_never_calls_mutation_tools_during_inventory(self, mock_get_profile, _auth, mock_api, mock_fetch): mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [] + mock_fetch.return_value = _final_page_fetch([]) mock_api.return_value = {"login": "jcwalker3"} gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs") @@ -105,20 +117,20 @@ class TestPRQueueInventory(unittest.TestCase): method = call.args[0] self.assertEqual(method, "GET") - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_reviewer_profile_can_proceed_from_inventory_to_review_gates(self, mock_get_profile, _auth, mock_api, mock_get_all): + def test_reviewer_profile_can_proceed_from_inventory_to_review_gates(self, mock_get_profile, _auth, mock_api, mock_fetch): mock_get_profile.return_value = { "profile_name": "gitea-reviewer", "allowed_operations": ["read", "approve", "review"], "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [ + mock_fetch.return_value = _final_page_fetch([ {"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}} - ] + ]) # mock_api: inventory whoami, eligibility whoami, eligibility PR, # POST review (#244: state + visible-verdict GET reviews). mock_api.side_effect = [ @@ -152,19 +164,19 @@ class TestPRQueueInventory(unittest.TestCase): self.assertIn("Repository:", result["message"]) self.assertIn("Successfully submitted review", result["message"]) - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") @patch("gitea_config.is_runtime_switching_enabled", return_value=False) - def test_static_runtime_rejects_or_hides_profile_activation(self, mock_switch, mock_get_profile, _auth, mock_api, mock_get_all): + def test_static_runtime_rejects_or_hides_profile_activation(self, mock_switch, mock_get_profile, _auth, mock_api, mock_fetch): mock_get_profile.return_value = { "profile_name": "gitea-author", "allowed_operations": ["read"], "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [] + mock_fetch.return_value = _final_page_fetch([]) mock_api.return_value = {"login": "jcwalker3"} result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs") @@ -200,11 +212,11 @@ class TestPRQueueInventory(unittest.TestCase): self.assertFalse(result["eligible"]) self.assertEqual(mock_api.call_count, 0) - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_inventory_failure_output_is_redacted_no_raw_exception_leak(self, mock_get_profile, _auth, mock_api, mock_get_all): + def test_inventory_failure_output_is_redacted_no_raw_exception_leak(self, mock_get_profile, _auth, mock_api, mock_fetch): """Inventory failure path must use redaction and not leak raw exception text.""" mock_get_profile.return_value = { "profile_name": "gitea-author", @@ -213,7 +225,7 @@ class TestPRQueueInventory(unittest.TestCase): "base_url": None, } # Exception contains something that should be redacted or not leaked raw - mock_get_all.side_effect = Exception("connection error token supersecrettoken123 at https://internal.example/repo") + mock_fetch.side_effect = Exception("connection error token supersecrettoken123 at https://internal.example/repo") mock_api.return_value = {"login": "jcwalker3"} result = gitea_review_pr(pr_number=42, event="APPROVE", remote="prgs") @@ -228,11 +240,11 @@ class TestPRQueueInventory(unittest.TestCase): # uses project's redaction pattern (token prefix -> [REDACTED]) # even if url leaks, the exception is redacted per pattern - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_inventory_failure_output_redacts_real_service_urls(self, mock_get_profile, _auth, mock_api, mock_get_all): + def test_inventory_failure_output_redacts_real_service_urls(self, mock_get_profile, _auth, mock_api, mock_fetch): """Failure paths must fully redact real service hostnames and URLs from inventory output.""" mock_get_profile.return_value = { "profile_name": "gitea-author", @@ -240,7 +252,7 @@ class TestPRQueueInventory(unittest.TestCase): "forbidden_operations": [], "base_url": None, } - mock_get_all.side_effect = Exception("failed to connect to https://gitea.prgs.cc/api/v1/repos/x") + mock_fetch.side_effect = Exception("failed to connect to https://gitea.prgs.cc/api/v1/repos/x") mock_api.return_value = {"login": "jcwalker3"} result = gitea_review_pr(pr_number=42, event="APPROVE", remote="prgs") @@ -250,11 +262,11 @@ class TestPRQueueInventory(unittest.TestCase): self.assertIn("[REDACTED_URL]", msg) self.assertNotIn("gitea.prgs.cc", msg) - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_author_profiles_can_perform_read_only_pr_inventory(self, mock_get_profile, _auth, mock_api, mock_get_all): + def test_author_profiles_can_perform_read_only_pr_inventory(self, mock_get_profile, _auth, mock_api, mock_fetch): """Author profiles with read can perform read-only PR inventory (report produced, no mutations).""" mock_get_profile.return_value = { "profile_name": "gitea-author", @@ -262,9 +274,9 @@ class TestPRQueueInventory(unittest.TestCase): "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [ + mock_fetch.return_value = _final_page_fetch([ {"number": 7, "title": "Some PR", "state": "open", "head": {"ref": "f", "sha": "def"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "someone"}, "labels": [], "body": ""} - ] + ]) mock_api.return_value = {"login": "jcwalker3"} result = gitea_review_pr(pr_number=7, event="APPROVE", remote="prgs") @@ -278,12 +290,12 @@ class TestPRQueueInventory(unittest.TestCase): self.assertEqual(call.args[0], "GET") @patch("mcp_server._local_git_remote_url") - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty( - self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url + self, mock_get_profile, _auth, mock_api, mock_fetch, mock_local_url ): mock_get_profile.return_value = { "profile_name": "gitea-author", @@ -291,7 +303,7 @@ class TestPRQueueInventory(unittest.TestCase): "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [] + mock_fetch.return_value = _final_page_fetch([]) mock_api.return_value = {"login": "jcwalker3"} mock_local_url.return_value = None @@ -309,12 +321,12 @@ class TestPRQueueInventory(unittest.TestCase): self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg) @patch("mcp_server._local_git_remote_url") - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") def test_empty_inventory_trusted_empty_when_gate_passes( - self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url + self, mock_get_profile, _auth, mock_api, mock_fetch, mock_local_url ): mock_get_profile.return_value = { "profile_name": "gitea-author", @@ -322,7 +334,7 @@ class TestPRQueueInventory(unittest.TestCase): "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [] + mock_fetch.return_value = _final_page_fetch([]) mock_api.return_value = {"login": "jcwalker3"} mock_local_url.return_value = ( "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" @@ -341,11 +353,11 @@ class TestPRQueueInventory(unittest.TestCase): self.assertIn("Open PRs found: 0 (trusted_empty)", msg) @patch("mcp_server._local_git_remote_url") - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url): + def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_fetch, mock_local_url): """Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory.""" mock_local_url.return_value = ( "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" @@ -357,7 +369,7 @@ class TestPRQueueInventory(unittest.TestCase): "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [] + mock_fetch.return_value = _final_page_fetch([]) mock_api.return_value = {"login": "jcwalker3"} result = gitea_review_pr(pr_number=1, event=event, remote="prgs") @@ -377,10 +389,10 @@ class TestPRQueueInventory(unittest.TestCase): # Tests use direct list/view calls + inventory path to ensure fields and # live data are available for detection without trusting prior handoffs. - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_r5_s1_stale_prior_handoff_merged_but_live_pr_open(self, mock_get_profile, _auth, mock_get_all): + def test_r5_s1_stale_prior_handoff_merged_but_live_pr_open(self, mock_get_profile, _auth, mock_fetch): """Scenario 1: stale prior handoff says merged but live PR list says open.""" mock_get_profile.return_value = { "profile_name": "gitea-author", @@ -388,17 +400,17 @@ class TestPRQueueInventory(unittest.TestCase): "forbidden_operations": [], "base_url": None, } - mock_get_all.return_value = [{ + mock_fetch.return_value = _final_page_fetch([{ "number": 99, "title": "Stale merge claim", "state": "open", "head": {"ref": "f99", "sha": "sha99"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other"}, "labels": [], "body": "Fixes #200", "updated_at": "2026-07-05T22:00:00Z" - }] + }]) prs = gitea_list_prs(state="open", remote="prgs") - self.assertEqual(len(prs), 1) - self.assertEqual(prs[0]["state"], "open") - self.assertIn("updated_at", prs[0]) # enables staleness check vs prior "merged" + self.assertEqual(len(prs["prs"]), 1) + self.assertEqual(prs["prs"][0]["state"], "open") + self.assertIn("updated_at", prs["prs"][0]) # enables staleness check vs prior "merged" @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -424,11 +436,11 @@ class TestPRQueueInventory(unittest.TestCase): self.assertIsNotNone(pr.get("closed_at")) # live closed would contradict prior "open" handoff - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_r5_s3_list_prs_and_view_pr_disagree(self, mock_get_profile, _auth, mock_api, mock_get_all): + def test_r5_s3_list_prs_and_view_pr_disagree(self, mock_get_profile, _auth, mock_api, mock_fetch): """Scenario 3: gitea_list_prs and gitea_view_pr disagree on state/details.""" mock_get_profile.return_value = { "profile_name": "gitea-author", @@ -437,12 +449,12 @@ class TestPRQueueInventory(unittest.TestCase): "base_url": None, } # list says open - mock_get_all.return_value = [{ + mock_fetch.return_value = _final_page_fetch([{ "number": 77, "title": "Disagree", "state": "open", "head": {"ref": "f77", "sha": "s77"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "x"}, "labels": [], "body": "", "updated_at": "2026-07-05T23:00:00Z" - }] + }]) # view says closed (simulating inconsistency) mock_api.return_value = { "number": 77, "title": "Disagree", "state": "closed", @@ -453,7 +465,7 @@ class TestPRQueueInventory(unittest.TestCase): } listed = gitea_list_prs(state="open", remote="prgs") viewed = gitea_view_pr(pr_number=77, remote="prgs") - self.assertEqual(listed[0]["state"], "open") + self.assertEqual(listed["prs"][0]["state"], "open") self.assertEqual(viewed["state"], "closed") # caller must reconcile; mismatch is hard blocker per rules @@ -483,11 +495,11 @@ class TestPRQueueInventory(unittest.TestCase): self.assertIsNone(pr.get("merge_commit_sha")) # post-merge view must surface missing commit for detection - @patch("mcp_server.api_get_all") + @patch("mcp_server.api_fetch_page") @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @patch("mcp_server.get_profile") - def test_r5_s5_linked_issue_remains_open_after_claimed_merge(self, mock_get_profile, _auth, mock_api, mock_get_all): + def test_r5_s5_linked_issue_remains_open_after_claimed_merge(self, mock_get_profile, _auth, mock_api, mock_fetch): """Scenario 5: linked issue remains open after claimed merge.""" mock_get_profile.return_value = { "profile_name": "gitea-author", @@ -496,12 +508,12 @@ class TestPRQueueInventory(unittest.TestCase): "base_url": None, } # PR claims merge, links #300 - mock_get_all.return_value = [{ + mock_fetch.return_value = _final_page_fetch([{ "number": 300, "title": "PR for #300", "state": "closed", "head": {"ref": "f300", "sha": "s3"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "u"}, "labels": [], "body": "Fixes #300", "updated_at": "2026-07-05T20:20:00Z" - }] + }]) # But linked issue (via list_issues or view) still open - here we use view_pr body to confirm link # For this test we assert the PR data shows link, caller must check issue state live mock_api.return_value = { diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 7166cca..ec411d2 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -24,11 +24,15 @@ from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, assess_author_pr_report, + assess_reviewer_baseline_validation_proof, assess_capability_evidence, assess_capability_proof, assess_contradictory_no_pr_claim, assess_continuation_mode_report, assess_controller_handoff, + assess_create_issue_final_report, + assess_create_issue_mode_isolation, + assess_create_issue_workflow_source, assess_edited_pr_inventory_coverage, assess_empty_queue_report, assess_fresh_issue_selection, @@ -2042,5 +2046,240 @@ class TestIssueFilingFinalReport(unittest.TestCase): self.assertTrue(result["downgraded"]) +class TestCreateIssueFinalReport(unittest.TestCase): + """#337: create-issue final-report verifier coverage.""" + + ISSUE_TITLE = "Extract create-issue workflow from monolithic LLM project skill" + CAPABILITIES = { + "create_issue": { + "requested_task": "create_issue", + "required_operation_permission": "gitea.issue.create", + "allowed_in_current_session": True, + }, + } + CLOSEST = [{"number": 333, "title": "Split monolithic workflow", "state": "closed"}] + + def _good_report(self): + return "\n".join([ + "Task mode: create-issue", + "Workflow source: skills/llm-project-workflow/workflows/create-issue.md", + f"Updated issue #337 — `{self.ISSUE_TITLE}` (remaining verifier AC).", + "Duplicate check: searched 54 open issues.", + "Closest existing: #333 — Split monolithic workflow (partial overlap).", + "Why new issue justified: dedicated create-issue extraction.", + "Only mutation: issue creation (gitea.issue.create).", + "Confirm no labels, comments, PRs, reviews, merges, or closes.", + "## Controller Handoff", + "- Task: create-issue", + "- Repo: Scaled-Tech-Consulting/Gitea-Tools", + "- Role: author", + "- Identity: jcwalker3 / prgs-author", + "- Active profile: prgs-author", + "- Runtime context: prgs-author on prgs", + "- Requested issue task: extract create-issue workflow verifier", + "- Workflow source: workflows/create-issue.md @ fd396df", + "- Capability proof: create_issue allowed (gitea.issue.create)", + "- Duplicate search terms: create-issue, workflow extract", + "- Duplicate search pagination proof: 54 open issues < limit 500", + "- Duplicates found: #333 partial only", + "- Issues created: none (implementation PR for remaining AC)", + "- Issues commented: none", + "- Issues edited: none", + "- Issues skipped as duplicates: none", + "- File edits by issue creator: review_proofs.py, tests", + "- Worktree/index mutations: feat/issue-337 worktree created", + "- Git ref mutations: fetch prgs/master", + "- MCP/Gitea mutations: gitea_mark_issue start on #337", + "- Issue mutations: claimed #337", + "- Label/assignment/milestone mutations: none", + "- External-state mutations: none", + "- Read-only diagnostics: whoami, list issues", + "- Blockers: none", + "- Current status: implementing verifier", + "- Safe next action: open PR", + "- Safety statement: no review/merge/repo main-checkout edits", + "- Issue/PR: #337", + "- Branch/SHA: feat/issue-337-create-issue-verifier", + "- Files changed: review_proofs.py, tests", + "- Validation: pytest tests/test_review_proofs.py", + "- Mutations: create_issue capability resolved; mark_issue only", + "- Next: open PR", + "- Safety: no review/merge", + ]) + + def test_complete_create_issue_report_earns_a(self): + result = assess_create_issue_final_report( + self._good_report(), + issue_number=337, + issue_title=self.ISSUE_TITLE, + action="updated", + mutations=["create_issue"], + resolved_capabilities=self.CAPABILITIES, + issues_searched=54, + closest_matches=self.CLOSEST, + performed_mutations=["create_issue"], + ) + self.assertEqual(result["grade"], "A") + self.assertFalse(result["downgraded"]) + + def test_missing_workflow_source_downgrades(self): + report = self._good_report().replace("workflows/create-issue.md", "SKILL.md only") + result = assess_create_issue_workflow_source(report) + self.assertTrue(result["downgraded"]) + + def test_work_issue_field_in_handoff_downgrades(self): + report = self._good_report().replace( + "- Safety statement:", + "- Selected issue: #123\n- Safety statement:", + ) + result = assess_create_issue_mode_isolation(report) + self.assertTrue(result["downgraded"]) + self.assertTrue( + any("Selected issue" in r for r in result["reasons"]) + ) + + def test_review_merge_field_in_handoff_downgrades(self): + report = self._good_report().replace( + "- Safety statement:", + "- Review decision: APPROVE\n- Safety statement:", + ) + result = assess_create_issue_mode_isolation(report) + self.assertTrue(result["downgraded"]) + + def test_legacy_workspace_mutations_without_precise_categories_downgrades(self): + report = self._good_report().replace( + "- File edits by issue creator:", + "- Workspace mutations: none\n- File edits by issue creator:", + ) + lines = report.splitlines() + cleaned = [ + line for line in lines + if not line.strip().startswith("- File edits by issue creator:") + ] + report = "\n".join(cleaned) + result = assess_create_issue_mode_isolation(report) + self.assertTrue(result["downgraded"]) + + +class TestReviewerBaselineValidationProof(unittest.TestCase): + """Issue #325: baseline validation must use branches/ worktrees.""" + + ROOT = "/repo/Gitea-Tools" + + def _good_baseline_proof(self, **overrides): + proof = { + "worktree_path": f"{self.ROOT}/branches/baseline-master-pr280", + "baseline_target_sha": PINNED, + "pr_head_sha": OTHER, + "baseline_failures": ["tests/test_foo.py::test_bar"], + "pr_failures": ["tests/test_foo.py::test_bar"], + "failure_signatures_match": True, + "clean_before": True, + "clean_after": True, + } + proof.update(overrides) + return proof + + def test_main_checkout_test_run_blocks(self): + result = assess_reviewer_baseline_validation_proof( + validation_runs=[ + { + "command": "venv/bin/pytest tests/", + "working_directory": self.ROOT, + "project_root": self.ROOT, + } + ], + project_root=self.ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(result["violations"]) + + def test_branches_worktree_test_run_passes(self): + result = assess_reviewer_baseline_validation_proof( + validation_runs=[ + { + "command": "venv/bin/pytest tests/", + "working_directory": f"{self.ROOT}/branches/review-pr-280", + "project_root": self.ROOT, + } + ], + project_root=self.ROOT, + ) + self.assertTrue(result["proven"]) + + def test_preexisting_claim_without_baseline_proof_fails(self): + result = assess_reviewer_baseline_validation_proof( + report_text=( + "Validation: full-suite failures are pre-existing on master." + ), + project_root=self.ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["claims_preexisting"]) + + def test_preexisting_claim_with_complete_baseline_proof_passes(self): + result = assess_reviewer_baseline_validation_proof( + report_text="Failures are pre-existing on master.", + baseline_proof=self._good_baseline_proof(), + project_root=self.ROOT, + ) + self.assertTrue(result["proven"]) + + def test_mismatched_failures_block_preexisting_claim(self): + result = assess_reviewer_baseline_validation_proof( + report_text="Same as master.", + baseline_proof=self._good_baseline_proof( + failure_signatures_match=False, + pr_failures=["tests/test_other.py::test_other"], + ), + project_root=self.ROOT, + ) + self.assertFalse(result["proven"]) + + def test_report_parsed_main_checkout_cwd_blocks(self): + result = assess_reviewer_baseline_validation_proof( + report_text=( + "- Validation command: venv/bin/pytest tests/\n" + f"- Working directory: {self.ROOT}\n" + ), + project_root=self.ROOT, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_build_final_report_downgrades_main_checkout_validation(self): + report = build_final_report( + checkout_proof=_good_checkout(), + inventory=_good_inventory(), + validation=_good_validation(), + contamination=_good_contamination(), + identity_eligible=True, + merge_performed=False, + issue_status_verified=True, + capability_evidence=_good_capability_evidence(), + sweep=_good_sweep(), + live_state=_good_live_state(), + role_boundary=_good_role_boundary(), + review_mutation=_good_review_mutation(), + controller_handoff=_good_handoff(), + capability_proof=_good_capability_proof(), + sweep_proof=_good_secret_sweep(), + worktree_proof={ + "worktree_path": "/repo/branches/review-feat-issue-224", + "porcelain_status": "", + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": True, + }, + report_text=( + "- Validation command: venv/bin/pytest tests/\n" + f"- Working directory: {self.ROOT}\n" + ), + project_root=self.ROOT, + ) + self.assertNotEqual(report["grade"], "A") + self.assertFalse(report["baseline_validation_proven"]) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_reviewer_mergeability_skip.py b/tests/test_reviewer_mergeability_skip.py new file mode 100644 index 0000000..98fcc2d --- /dev/null +++ b/tests/test_reviewer_mergeability_skip.py @@ -0,0 +1,150 @@ +"""Tests for non-mergeable skip conflict-proof verifier (#322).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_mergeability_skip import assess_non_mergeable_skip_proof # noqa: E402 + +HEAD_A = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" +HEAD_B = "1111111111111111111111111111111111111111" + + +def _good_skip_report(pr_number: int, head: str, files: str = "review_proofs.py") -> str: + return "\n".join([ + f"Skipped PR #{pr_number} (non-mergeable).", + f"- PR number: #{pr_number}", + f"- Current head SHA: {head}", + "- Mergeability result: false", + "- Conflict proof command: git merge-tree $(git merge-base prgs/master prgs/feat) prgs/master prgs/feat", + f"- Conflicting files: {files}", + "- Head unchanged since prior blocker", + "- Blocking category: merge conflict", + ]) + + +class TestNonMergeableSkipProof(unittest.TestCase): + def test_no_skips_passes(self): + report = "Selected PR #203 for review. No earlier PRs skipped." + result = assess_non_mergeable_skip_proof(report, skipped_prs=[]) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_documented_non_mergeable_skip_passes(self): + report = _good_skip_report(282, HEAD_A) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "head_sha": HEAD_A, + "mergeability_verified": True, + "conflicting_files": ["review_proofs.py"], + "head_changed_since_prior_blocker": False, + }], + ) + self.assertTrue(result["proven"]) + + def test_skip_without_conflict_proof_blocks(self): + report = "\n".join([ + "Skipped PR #284 because it was not mergeable.", + f"Head SHA: {HEAD_B}", + ]) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 284, + "mergeable": False, + "head_sha": HEAD_B, + "mergeability_verified": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("conflict proof" in r for r in result["reasons"])) + + def test_missing_head_sha_blocks(self): + report = _good_skip_report(282, HEAD_A).replace(HEAD_A, "") + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "head_sha": HEAD_A, + "mergeability_verified": True, + }], + ) + self.assertFalse(result["proven"]) + + def test_mergeability_unverified_classification_passes(self): + report = "\n".join([ + "Skipped PR #282.", + "Classification: MERGEABILITY_UNVERIFIED", + "Conflict proof could not be retrieved in this session.", + ]) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "mergeability_verified": False, + }], + ) + self.assertTrue(result["proven"]) + + def test_unverified_without_classification_blocks(self): + report = "Skipped PR #282 due to conflicts." + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "mergeability_verified": False, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("MERGEABILITY_UNVERIFIED" in r for r in result["reasons"])) + + def test_non_mergeable_without_file_details_still_passes_with_command(self): + report = _good_skip_report(284, HEAD_B, files="not available") + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 284, + "mergeable": False, + "head_sha": HEAD_B, + "mergeability_verified": True, + "conflicting_files": [], + }], + ) + self.assertTrue(result["proven"]) + + def test_stale_head_requires_head_change_field(self): + report = _good_skip_report(282, HEAD_A).replace( + "- Head unchanged since prior blocker", + "", + ) + result = assess_non_mergeable_skip_proof( + report, + skipped_prs=[{ + "pr_number": 282, + "mergeable": False, + "head_sha": HEAD_A, + "mergeability_verified": True, + "head_changed_since_prior_blocker": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("head-changed" in r for r in result["reasons"])) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_non_mergeable_skip_proof as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_terminal_review_hard_stop.py b/tests/test_terminal_review_hard_stop.py new file mode 100644 index 0000000..bd5109b --- /dev/null +++ b/tests/test_terminal_review_hard_stop.py @@ -0,0 +1,187 @@ +"""Tests for the session hard-stop after a terminal review mutation (#332). + +Extends the single-terminal-decision machinery (#211): after a live +REQUEST_CHANGES the run must stop; after an APPROVE only the merge sequence +for that same PR may continue; a different PR can never be reviewed or +merged in the same run; duplicate REQUEST_CHANGES at an unchanged head is +suppressed. +""" +import os +import unittest +from unittest.mock import patch + +import mcp_server + + +def _lock(mutations=None, correction=False): + return { + "task": "review_pr", + "remote": "prgs", + "session_pid": os.getpid(), + "session_profile": "prgs-reviewer", + "session_profile_lock": "", + "final_review_decision_ready": False, + "ready_pr_number": None, + "ready_action": None, + "ready_expected_head_sha": None, + "ready_remote": None, + "ready_org": None, + "ready_repo": None, + "live_mutations": list(mutations or []), + "correction_authorized": correction, + "correction_reason": None, + } + + +def _seed(mutations=None, correction=False): + mcp_server._save_review_decision_lock(_lock(mutations, correction)) + + +APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1, + "review_state": "approve"} +RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2, + "review_state": "request_changes"} + + +class TestTerminalHardStopReasons(unittest.TestCase): + def tearDown(self): + mcp_server._save_review_decision_lock(None) + + def test_no_lock_no_reasons(self): + mcp_server._save_review_decision_lock(None) + self.assertEqual( + mcp_server.terminal_review_hard_stop_reasons(5, "merge"), []) + + def test_no_terminal_mutation_no_reasons(self): + _seed() + self.assertEqual( + mcp_server.terminal_review_hard_stop_reasons(5, "merge"), []) + + def test_request_changes_blocks_everything(self): + _seed([RC_A]) + for op in ("merge", "mark_ready", "review"): + for pr in (5, 6): + reasons = mcp_server.terminal_review_hard_stop_reasons(pr, op) + self.assertTrue(reasons, f"{op}/{pr}") + self.assertIn("request_changes on PR #5", reasons[0]) + self.assertIn("#332", reasons[0]) + + def test_approve_allows_same_pr_merge_only(self): + _seed([APPROVED_A]) + self.assertEqual( + mcp_server.terminal_review_hard_stop_reasons(5, "merge"), []) + self.assertTrue( + mcp_server.terminal_review_hard_stop_reasons(6, "merge")) + self.assertTrue( + mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready")) + self.assertTrue( + mcp_server.terminal_review_hard_stop_reasons(6, "review")) + + def test_correction_reopens_review_path_only(self): + _seed([RC_A], correction=True) + self.assertEqual( + mcp_server.terminal_review_hard_stop_reasons(5, "mark_ready"), []) + self.assertEqual( + mcp_server.terminal_review_hard_stop_reasons(5, "review"), []) + # correction never opens a cross-PR merge + self.assertTrue( + mcp_server.terminal_review_hard_stop_reasons(6, "merge")) + + +class TestMergeHardStopWiring(unittest.TestCase): + def tearDown(self): + mcp_server._save_review_decision_lock(None) + + def test_merge_blocked_after_request_changes(self): + _seed([RC_A]) + result = mcp_server.gitea_merge_pr( + pr_number=6, confirmation="MERGE PR 6", remote="prgs") + self.assertFalse(result["performed"]) + self.assertTrue( + any("#332" in r for r in result["reasons"]), result["reasons"]) + + def test_merge_of_other_pr_blocked_after_approve(self): + _seed([APPROVED_A]) + result = mcp_server.gitea_merge_pr( + pr_number=6, confirmation="MERGE PR 6", remote="prgs") + self.assertFalse(result["performed"]) + self.assertTrue( + any("#332" in r for r in result["reasons"]), result["reasons"]) + + +class TestMarkFinalHardStopWiring(unittest.TestCase): + def tearDown(self): + mcp_server._save_review_decision_lock(None) + + def test_mark_ready_blocked_after_terminal_mutation(self): + _seed([RC_A]) + result = mcp_server.gitea_mark_final_review_decision( + pr_number=6, action="approve", remote="prgs") + self.assertFalse(result["marked_ready"]) + self.assertTrue( + any("#332" in r for r in result["reasons"]), result["reasons"]) + + +def _feedback(blocking, stale=False, success=True): + return { + "success": success, + "has_blocking_change_requests": blocking, + "review_feedback_stale": stale, + "current_head_sha": "abc123", + } + + +class TestDuplicateRequestChangesSuppression(unittest.TestCase): + def tearDown(self): + mcp_server._save_review_decision_lock(None) + + def test_duplicate_request_changes_blocked_at_same_head(self): + _seed() + with patch.object(mcp_server, "gitea_get_pr_review_feedback", + return_value=_feedback(blocking=True, stale=False)): + result = mcp_server.gitea_mark_final_review_decision( + pr_number=6, action="request_changes", remote="prgs") + self.assertFalse(result["marked_ready"]) + self.assertTrue( + any("duplicate" in r for r in result["reasons"]), + result["reasons"]) + + def test_request_changes_allowed_when_prior_blocker_stale(self): + _seed() + with patch.object(mcp_server, "gitea_get_pr_review_feedback", + return_value=_feedback(blocking=True, stale=True)): + result = mcp_server.gitea_mark_final_review_decision( + pr_number=6, action="request_changes", remote="prgs") + self.assertTrue(result["marked_ready"], result.get("reasons")) + + def test_request_changes_allowed_when_no_blocker(self): + _seed() + with patch.object(mcp_server, "gitea_get_pr_review_feedback", + return_value=_feedback(blocking=False)): + result = mcp_server.gitea_mark_final_review_decision( + pr_number=6, action="request_changes", remote="prgs") + self.assertTrue(result["marked_ready"], result.get("reasons")) + + def test_request_changes_fails_closed_when_feedback_unavailable(self): + _seed() + with patch.object(mcp_server, "gitea_get_pr_review_feedback", + return_value=_feedback(blocking=False, + success=False)): + result = mcp_server.gitea_mark_final_review_decision( + pr_number=6, action="request_changes", remote="prgs") + self.assertFalse(result["marked_ready"]) + self.assertTrue( + any("could not verify" in r for r in result["reasons"]), + result["reasons"]) + + def test_approve_does_not_fetch_feedback(self): + _seed() + with patch.object(mcp_server, "gitea_get_pr_review_feedback", + side_effect=AssertionError("must not be called")): + result = mcp_server.gitea_mark_final_review_decision( + pr_number=6, action="approve", remote="prgs") + self.assertTrue(result["marked_ready"], result.get("reasons")) + + +if __name__ == "__main__": + unittest.main()