diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 5209f27..7e13834 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -500,6 +500,7 @@ import review_proofs # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 import author_mutation_worktree # noqa: E402 +import issue_claim_heartbeat # noqa: E402 import merged_cleanup_reconcile # noqa: E402 @@ -5170,6 +5171,41 @@ def gitea_audit_config() -> dict: return report +def _post_structured_issue_comment( + *, + issue_number: int, + body: str, + remote: str, + host: str | None, + org: str | None, + repo: str | None, + audit_op: str = "create_issue_comment", +) -> dict: + """Post an issue-thread comment after permission gates already passed.""" + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments" + with _audited( + audit_op, + host=h, + remote=remote, + org=o, + repo=r, + issue_number=issue_number, + request_metadata={"body_chars": len(body)}, + ): + data = api_request("POST", api, auth, {"body": body}) + result = { + "success": True, + "performed": True, + "comment_id": data["id"], + "issue_number": issue_number, + } + if _reveal_endpoints(): + result["url"] = data.get("html_url") + return result + + @mcp.tool() def gitea_mark_issue( issue_number: int, @@ -5179,6 +5215,8 @@ def gitea_mark_issue( org: str | None = None, repo: str | None = None, worktree_path: str | None = None, + branch_name: str | None = None, + profile: str | None = None, ) -> dict: """Claim or release an issue via the status:in-progress label. @@ -5228,7 +5266,31 @@ def gitea_mark_issue( request_metadata={"op": "add", "label": "status:in-progress"}): api_request("POST", f"{base}/issues/{issue_number}/labels", auth, {"labels": [label_id]}) - return {"success": True, "message": f"Issue #{issue_number} claimed."} + active_profile = (profile or get_profile().get("profile_name") or "unknown") + branch = (branch_name or "pending").strip() or "pending" + heartbeat_body = issue_claim_heartbeat.format_heartbeat_body( + kind="claim", + issue_number=issue_number, + branch=branch, + phase="claimed", + profile=active_profile, + next_action="create worktree and begin implementation", + ) + heartbeat = _post_structured_issue_comment( + issue_number=issue_number, + body=heartbeat_body, + remote=remote, + host=host, + org=org, + repo=repo, + audit_op="claim_heartbeat", + ) + return { + "success": True, + "message": f"Issue #{issue_number} claimed.", + "heartbeat_posted": heartbeat.get("success", False), + "heartbeat_comment_id": heartbeat.get("comment_id"), + } else: with _audited("unlabel_issue", host=h, remote=remote, org=o, repo=r, issue_number=issue_number, @@ -5238,6 +5300,217 @@ def gitea_mark_issue( return {"success": True, "message": f"Issue #{issue_number} released."} +@mcp.tool() +def gitea_post_heartbeat( + issue_number: int, + branch: str, + phase: str, + pr: str = "none", + next_action: str = "none", + blocker: str = "none", + profile: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Post a structured progress heartbeat on a claimed issue (#268).""" + blocked = _profile_permission_block( + task_capability_map.required_permission("post_heartbeat")) + if blocked: + return blocked + verify_preflight_purity(remote) + active_profile = profile or get_profile().get("profile_name") + body = issue_claim_heartbeat.format_heartbeat_body( + kind="progress", + issue_number=issue_number, + branch=branch, + phase=phase, + profile=active_profile, + pr=pr, + next_action=next_action, + blocker=blocker, + ) + return _post_structured_issue_comment( + issue_number=issue_number, + body=body, + remote=remote, + host=host, + org=org, + repo=repo, + audit_op="progress_heartbeat", + ) + + +@mcp.tool() +def gitea_reconcile_issue_claims( + state: str = "open", + stale_after_hours: int = 24, + heartbeat_lease_minutes: int = 30, + limit: int = 100, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Read-only inventory of issue claims and heartbeat lease status (#268).""" + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + issues = api_get_all(f"{base}/issues?state={state}&type=issues", auth, limit=limit) + open_prs = api_get_all(f"{base}/pulls?state=open", auth) + branches = api_get_all(f"{base}/branches", auth, limit=limit) + branch_names = [b.get("name") for b in branches if b.get("name")] + + comments_by_issue: dict[int, list[dict]] = {} + for issue in issues: + if not issue_claim_heartbeat.issue_has_in_progress_label(issue): + continue + number = int(issue["number"]) + api = f"{base}/issues/{number}/comments" + comments_by_issue[number] = api_request("GET", api, auth) or [] + + reclaim_after_minutes = max(heartbeat_lease_minutes * 2, 60) + inventory = issue_claim_heartbeat.build_claim_inventory( + issues=issues, + comments_by_issue=comments_by_issue, + open_prs=open_prs, + branch_names=branch_names, + heartbeat_lease_minutes=heartbeat_lease_minutes, + reclaim_after_minutes=reclaim_after_minutes, + ) + inventory["cleanup_plan"] = issue_claim_heartbeat.build_cleanup_plan(inventory) + inventory["success"] = True + inventory["performed"] = False + return inventory + + +@mcp.tool() +def gitea_cleanup_stale_claims( + dry_run: bool = True, + execute_confirmed: bool = False, + heartbeat_lease_minutes: int = 30, + limit: int = 100, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Propose or execute stale-claim cleanup for phantom/reclaimable issues (#268).""" + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "performed": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + inventory = gitea_reconcile_issue_claims( + heartbeat_lease_minutes=heartbeat_lease_minutes, + limit=limit, + remote=remote, + host=host, + org=org, + repo=repo, + ) + if not inventory.get("success"): + return inventory + + plan = [ + entry + for entry in inventory.get("cleanup_plan") or [] + if entry.get("action") == "remove_status_in_progress_and_comment" + ] + report = { + "success": True, + "dry_run": dry_run, + "performed": False, + "planned_actions": plan, + "inventory_counts": inventory.get("counts"), + } + if dry_run: + return report + + if not execute_confirmed: + raise ValueError( + "execute_confirmed must be True when dry_run=False (fail closed)" + ) + + blocked = _profile_permission_block( + task_capability_map.required_permission("cleanup_stale_claims")) + if blocked: + return blocked + verify_preflight_purity(remote) + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + labels = api_request("GET", f"{base}/labels?limit=100", auth) + label_id = next( + (lb["id"] for lb in labels if lb.get("name") == "status:in-progress"), + None, + ) + if label_id is None: + raise RuntimeError("Label 'status:in-progress' not found") + + actions: list[dict] = [] + active_profile = get_profile().get("profile_name") + for entry in plan: + issue_number = int(entry["issue_number"]) + with _audited( + "unlabel_issue", + host=h, + remote=remote, + org=o, + repo=r, + issue_number=issue_number, + request_metadata={"op": "cleanup_stale_claim", "label": "status:in-progress"}, + ): + api_request( + "DELETE", + f"{base}/issues/{issue_number}/labels/{label_id}", + auth, + ) + cleanup_body = issue_claim_heartbeat.format_heartbeat_body( + kind="cleanup", + issue_number=issue_number, + branch="none", + phase="stale-claim-cleanup", + profile=active_profile, + next_action="issue reclaimable by queue", + blocker=entry.get("status") or "stale", + ) + comment = _post_structured_issue_comment( + issue_number=issue_number, + body=cleanup_body, + remote=remote, + host=host, + org=org, + repo=repo, + audit_op="cleanup_heartbeat", + ) + actions.append( + { + "issue_number": issue_number, + "label_removed": True, + "cleanup_comment_id": comment.get("comment_id"), + } + ) + + report["performed"] = True + report["actions"] = actions + return report + + @mcp.tool() def gitea_list_labels( remote: str = "dadeschools", diff --git a/issue_claim_heartbeat.py b/issue_claim_heartbeat.py new file mode 100644 index 0000000..3d16e1d --- /dev/null +++ b/issue_claim_heartbeat.py @@ -0,0 +1,295 @@ +"""Issue claim heartbeat leases and stale-claim reconciliation (#268). + +Structured issue-thread comments prove live ownership beyond the +``status:in-progress`` label alone. Queue inventory can classify claims as +active, stale, reclaimable, PR-backed, or phantom. +""" + +from __future__ import annotations + +import re +from datetime import datetime, timedelta, timezone +from typing import Any + +MARKER = "" +IN_PROGRESS_LABEL = "status:in-progress" + +_KIND_CLAIM = "claim" +_KIND_PROGRESS = "progress" +_KIND_CLEANUP = "cleanup" + +_FIELD_RE = re.compile( + r"^\s*-\s*([a-z_]+)\s*:\s*(.+?)\s*$", + re.IGNORECASE | re.MULTILINE, +) +_ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE) + + +def _parse_timestamp(value: str | None) -> datetime | None: + if not value: + return None + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed + + +def format_heartbeat_body( + *, + kind: str, + issue_number: int, + branch: str, + phase: str, + profile: str | None = None, + pr: str = "none", + next_action: str = "none", + blocker: str = "none", +) -> str: + """Return a structured, machine-parseable issue comment body.""" + profile_value = (profile or "unknown").strip() or "unknown" + lines = [ + MARKER, + "**Issue claim heartbeat**", + f"- kind: {kind}", + f"- issue: #{issue_number}", + f"- branch: {branch}", + f"- phase: {phase}", + f"- profile: {profile_value}", + f"- pr: {pr}", + f"- blocker: {blocker}", + f"- next_action: {next_action}", + ] + return "\n".join(lines) + + +def parse_heartbeat_comment(body: str) -> dict[str, Any] | None: + """Parse one structured heartbeat comment, or None when not a heartbeat.""" + text = body or "" + if MARKER not in text: + return None + fields: dict[str, str] = {} + for match in _FIELD_RE.finditer(text): + fields[match.group(1).strip().lower()] = match.group(2).strip() + if not fields: + return None + issue_raw = fields.get("issue", "") + issue_digits = re.sub(r"[^\d]", "", issue_raw) + issue_number = int(issue_digits) if issue_digits.isdigit() else None + return { + "kind": fields.get("kind"), + "issue_number": issue_number, + "branch": fields.get("branch"), + "phase": fields.get("phase"), + "profile": fields.get("profile"), + "pr": fields.get("pr"), + "blocker": fields.get("blocker"), + "next_action": fields.get("next_action"), + "raw_fields": fields, + } + + +def extract_issue_heartbeats( + comments: list[dict], + *, + issue_number: int | None = None, +) -> list[dict]: + """Return parsed heartbeats newest-last, optionally filtered to *issue_number*.""" + heartbeats: list[dict] = [] + for comment in comments or []: + parsed = parse_heartbeat_comment(comment.get("body") or "") + if not parsed: + continue + if issue_number is not None and parsed.get("issue_number") != issue_number: + continue + heartbeats.append( + { + **parsed, + "comment_id": comment.get("id"), + "author": (comment.get("user") or {}).get("login") + or comment.get("author"), + "created_at": comment.get("created_at"), + "updated_at": comment.get("updated_at"), + } + ) + return heartbeats + + +def issue_has_in_progress_label(issue: dict) -> bool: + labels = issue.get("labels") or [] + for label in labels: + name = label if isinstance(label, str) else label.get("name") + if name == IN_PROGRESS_LABEL: + return True + return False + + +def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None: + pattern = f"issue-{issue_number}" + closes = f"closes #{issue_number}" + fixes = f"fixes #{issue_number}" + for pr in open_prs or []: + head = (pr.get("head") or {}).get("ref") or "" + text = f"{pr.get('title', '')} {pr.get('body', '')}".lower() + if pattern in head.lower(): + return pr + if closes in text or fixes in text: + return pr + return None + + +def _matching_branch_names(issue_number: int, branch_names: list[str]) -> list[str]: + pattern = f"issue-{issue_number}" + return [name for name in branch_names if pattern in (name or "").lower()] + + +def classify_issue_claim( + *, + issue: dict, + comments: list[dict], + open_prs: list[dict] | None = None, + branch_names: list[str] | None = None, + now: datetime | None = None, + heartbeat_lease_minutes: int = 30, + reclaim_after_minutes: int = 60, +) -> dict[str, Any]: + """Classify one issue's claim state from labels, heartbeats, PRs, and branches.""" + issue_number = int(issue.get("number") or 0) + open_prs = open_prs or [] + branch_names = branch_names or [] + current = now or datetime.now(timezone.utc) + + has_label = issue_has_in_progress_label(issue) + heartbeats = extract_issue_heartbeats(comments, issue_number=issue_number) + latest = heartbeats[-1] if heartbeats else None + latest_at = _parse_timestamp( + (latest or {}).get("updated_at") or (latest or {}).get("created_at") + ) + age_minutes = None + if latest_at: + age_minutes = int((current - latest_at).total_seconds() // 60) + + linked_pr = _linked_open_pr(issue_number, open_prs) + matching_branches = _matching_branch_names(issue_number, branch_names) + + if not has_label: + status = "not_claimed" + reasons = ["issue lacks status:in-progress label"] + elif linked_pr: + status = "awaiting_review" + reasons = [f"open PR #{linked_pr.get('number')} covers this issue"] + elif not heartbeats: + status = "phantom" + reasons = [ + "status:in-progress label present without structured claim heartbeat" + ] + elif latest_at is None: + status = "phantom" + reasons = ["heartbeat comments present but timestamps could not be parsed"] + elif age_minutes is not None and age_minutes <= heartbeat_lease_minutes: + status = "active" + reasons = [f"heartbeat age {age_minutes}m within {heartbeat_lease_minutes}m lease"] + elif matching_branches and age_minutes is not None and age_minutes <= reclaim_after_minutes: + status = "active" + reasons = [ + f"matching branch(es) {', '.join(matching_branches)} with heartbeat " + f"age {age_minutes}m" + ] + elif age_minutes is not None and age_minutes > reclaim_after_minutes and not matching_branches: + status = "reclaimable" + reasons = [ + f"no heartbeat within {reclaim_after_minutes}m and no matching branch" + ] + elif age_minutes is not None and age_minutes > heartbeat_lease_minutes: + status = "stale" + reasons = [ + f"heartbeat age {age_minutes}m exceeds {heartbeat_lease_minutes}m lease" + ] + else: + status = "active" + reasons = ["claim has structured heartbeat proof"] + + return { + "issue_number": issue_number, + "title": issue.get("title"), + "status": status, + "has_in_progress_label": has_label, + "heartbeat_count": len(heartbeats), + "latest_heartbeat": latest, + "heartbeat_age_minutes": age_minutes, + "linked_open_pr": linked_pr.get("number") if linked_pr else None, + "matching_branches": matching_branches, + "reasons": reasons, + "reclaimable": status == "reclaimable", + "stale": status in {"stale", "phantom", "reclaimable"}, + } + + +def build_claim_inventory( + *, + issues: list[dict], + comments_by_issue: dict[int, list[dict]], + open_prs: list[dict], + branch_names: list[str], + now: datetime | None = None, + heartbeat_lease_minutes: int = 30, + reclaim_after_minutes: int = 60, +) -> dict[str, Any]: + """Build a queue inventory report for in-progress issue claims.""" + entries: list[dict] = [] + for issue in issues or []: + if not issue_has_in_progress_label(issue): + continue + number = int(issue["number"]) + entry = classify_issue_claim( + issue=issue, + comments=comments_by_issue.get(number, []), + open_prs=open_prs, + branch_names=branch_names, + now=now, + heartbeat_lease_minutes=heartbeat_lease_minutes, + reclaim_after_minutes=reclaim_after_minutes, + ) + entries.append(entry) + + counts: dict[str, int] = {} + for entry in entries: + counts[entry["status"]] = counts.get(entry["status"], 0) + 1 + + return { + "entries": entries, + "counts": counts, + "heartbeat_lease_minutes": heartbeat_lease_minutes, + "reclaim_after_minutes": reclaim_after_minutes, + "in_progress_total": len(entries), + } + + +def build_cleanup_plan(inventory: dict[str, Any]) -> list[dict]: + """Return stale/reclaimable/phantom claims that may be cleaned up.""" + plan: list[dict] = [] + for entry in inventory.get("entries") or []: + if entry.get("status") in {"reclaimable", "phantom"}: + plan.append( + { + "issue_number": entry["issue_number"], + "status": entry["status"], + "action": "remove_status_in_progress_and_comment", + "reasons": list(entry.get("reasons") or []), + } + ) + elif entry.get("status") == "stale" and not entry.get("linked_open_pr"): + plan.append( + { + "issue_number": entry["issue_number"], + "status": entry["status"], + "action": "report_only", + "reasons": list(entry.get("reasons") or []), + } + ) + return plan \ No newline at end of file diff --git a/review_proofs.py b/review_proofs.py index 77986a7..9c572c3 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1052,6 +1052,15 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non return None +# ── Linked-issue consistency (Issue #314) ──────────────────────────────────── +# +# Stale linked-issue text from a previous PR must not leak into a final +# report: the "Linked issue status" field must match the issue(s) the +# selected PR actually links, verified live this session, and no other +# "Issue #N" mention may appear unless it is a verified linked issue. + +_ISSUE_MENTION_RE = re.compile(r"\bissue\s+#(\d+)", re.IGNORECASE) + # ── Workspace-vs-worktree mutation consistency (Issue #313) ────────────────── # # A worktree reset/checkout/clean is a workspace mutation even when the @@ -1092,6 +1101,68 @@ def _report_labeled_fields(report_text): return fields +def _issue_numbers_mentioned(text): + """Return the set of ints referenced as 'Issue #N' in *text*.""" + return {int(n) for n in _ISSUE_MENTION_RE.findall(text or "")} + + +def assess_linked_issue_consistency(report_text, *, selected_pr=None, + linked_issues=None): + """#314: linked-issue status must match the selected PR, live-verified. + + *linked_issues* is the list of issue numbers the selected PR was + live-verified to link this session, or None when no live verification + happened. Without live proof the report may only say the status was + not verified; with proof, the ``Linked issue status`` field must name + every linked issue and no stale issue number may appear anywhere in + the report. + + Returns {'complete', 'downgraded', 'reasons'}; fails closed. + """ + fields = _report_labeled_fields(report_text) + status_value = fields.get("linked issue status") + + reasons = [] + + if status_value is None: + reasons.append( + "final report missing 'Linked issue status' field for the " + "selected PR" + ) + elif linked_issues is None: + if not status_value.strip().lower().startswith("not verified"): + reasons.append( + "linked issue status claimed without live proof; report " + "'Linked issue status: not verified in this session' or " + "verify the linked issue live" + ) + else: + allowed = set(linked_issues) + status_mentions = _issue_numbers_mentioned(status_value) + for missing in sorted(allowed - status_mentions): + reasons.append( + f"linked issue #{missing} of selected PR " + f"#{selected_pr} not reported in 'Linked issue status'" + ) + for stale in sorted(status_mentions - allowed): + reasons.append( + f"'Linked issue status' names issue #{stale}, which is " + f"not a live-verified linked issue of selected PR " + f"#{selected_pr}" + ) + for stale in sorted(_issue_numbers_mentioned(report_text) - allowed): + reasons.append( + f"stale issue #{stale} mentioned in final report but not " + f"live-verified as linked to selected PR #{selected_pr}" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + def _field_claims_none(fields, label): """True when *label* is present and its value is empty or 'none...'.""" value = fields.get(label) @@ -1171,6 +1242,102 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None): } +# ── Already-landed report wording (Issue #298) ─────────────────────────────── +# +# An already-landed PR is reconciliation-only: it must never be called +# eligible, its head SHA must never be called reviewed, and the legacy +# eligible/reviewed/scratch-worktree handoff fields must not appear. +# Reviewed-head claims in any report require validation + diff review to +# have actually passed. + +ALREADY_LANDED_FORBIDDEN_PHRASES = ( + "oldest eligible pr", + "next eligible pr", + "pinned reviewed head", + "scratch worktree used", +) + +ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED" + + +def assess_already_landed_report_wording(report_text, *, already_landed, + review_validated=False): + """#298: reject eligible/reviewed wording for already-landed PRs. + + *already_landed* states whether the already-landed gate fired for the + selected PR. *review_validated* states whether validation and diff + review actually passed this session; only then may a reviewed head + SHA be populated in a non-already-landed report. + + Returns {'complete', 'downgraded', 'reasons'}; fails closed. + """ + text_lower = (report_text or "").lower() + fields = _report_labeled_fields(report_text) + reasons = [] + + reviewed_head = fields.get("reviewed head sha") + reviewed_head_populated = bool( + reviewed_head and not reviewed_head.strip().lower().startswith("none") + ) + pinned_head_present = "pinned reviewed head" in fields + + if already_landed: + for phrase in ALREADY_LANDED_FORBIDDEN_PHRASES: + if phrase in text_lower: + reasons.append( + f"already-landed report must not use '{phrase}'; the PR " + "is reconciliation-only, not review/merge eligible" + ) + + eligibility = fields.get("eligibility class", "") + if ALREADY_LANDED_ELIGIBILITY_CLASS not in eligibility.upper(): + reasons.append( + "already-landed report missing 'Eligibility class: " + f"{ALREADY_LANDED_ELIGIBILITY_CLASS}'" + ) + + for required in ("oldest open pr requiring action", + "candidate head sha"): + if required not in fields: + reasons.append( + f"already-landed report missing '{required}' field" + ) + + if "reviewed head sha" not in fields: + reasons.append( + "already-landed report must state 'Reviewed head SHA: none'" + ) + elif reviewed_head_populated: + reasons.append( + "already-landed report must not claim a reviewed head SHA; " + "no validation or diff review passed for this PR" + ) + + worktree_used = fields.get("review worktree used", "") + if worktree_used.strip().lower() not in ("false", "no"): + reasons.append( + "already-landed report must state 'Review worktree used: " + "false'" + ) + elif not review_validated: + if pinned_head_present: + reasons.append( + "'Pinned reviewed head' populated before validation and " + "diff review passed" + ) + if reviewed_head_populated: + reasons.append( + "reviewed head SHA populated before validation and diff " + "review passed" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None, justification=None): """Assess reviewer/author role separation for blind queue workflows. @@ -1351,6 +1518,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, report_text=None, review_decision_lock=None, controller_handoff=None, capability_proof=None, sweep_proof=None, worktree_proof=None, + validation_environment=None, queue_ordering=None, baseline_validation=None, project_root=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. @@ -1379,6 +1547,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination, ) empty_queue_report = assess_empty_queue_report(report_text) + if validation_environment is None and report_text: + validation_env_proof = assess_validation_environment_proof( + report_text=report_text, + ) + elif validation_environment is not None: + validation_env_proof = assess_validation_environment_proof( + report_text=report_text, + validation_environment=validation_environment, + ) + else: + validation_env_proof = { + "proven": True, + "block": False, + "reasons": [], + "violations": [], + } if queue_ordering is None and report_text: queue_ordering_proof = assess_queue_ordering_proof( report_text=report_text, @@ -1546,6 +1730,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 validation_env_proof.get("proven"): + downgrade_reasons.append( + "validation environment proof missing or failed (#311)" + ) + downgrade_reasons.extend(validation_env_proof.get("reasons", [])) if not queue_ordering_proof.get("proven"): downgrade_reasons.append( "queue ordering proof missing or failed (#321)" @@ -1577,11 +1766,13 @@ 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 validation_env_proof.get("proven") and queue_ordering_proof.get("proven") and baseline_validation.get("proven") ) violations = [] + violations.extend(validation_env_proof.get("violations", [])) violations.extend(queue_ordering_proof.get("violations", [])) violations.extend(baseline_validation.get("violations", [])) if merge_performed and not merge_allowed: @@ -1635,6 +1826,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination, else True ), "empty_queue_trust_gate_status": empty_queue_report.get("status"), + "validation_environment_proven": bool( + validation_env_proof.get("proven") + ), + "validation_environment_violations": list( + validation_env_proof.get("violations") or [] + ), "queue_ordering_proven": bool(queue_ordering_proof.get("proven")), "queue_ordering_policy": queue_ordering_proof.get("policy"), "queue_ordering_violations": list( @@ -3251,6 +3448,192 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict: } +# --------------------------------------------------------------------------- +# Validation environment proof (#311) +# --------------------------------------------------------------------------- + +_VALIDATION_CLAIM_RE = re.compile( + r"(?:validation\s+command|pytest\s+(?:executed|passed|failed)|" + r"\d+\s+passed|\d+\s+failed|tests?\s+passed)", + re.IGNORECASE, +) + +_VALIDATION_CMD_LINE_RE = re.compile( + r"validation\s+command\s*:\s*(.+)", + re.IGNORECASE, +) + +_VALIDATION_CWD_LINE_RE = re.compile( + r"working\s+directory\s*:\s*(\S+)", + re.IGNORECASE, +) + +_WHICH_PYTEST_LINE_RE = re.compile( + r"which\s+pytest\s*:\s*(.+)", + re.IGNORECASE, +) + +_PYTEST_VERSION_LINE_RE = re.compile( + r"pytest\s+--version\s*:\s*(.+)", + re.IGNORECASE, +) + +_BARE_PYTEST_CMD_RE = re.compile( + r"^(?:python\s+-m\s+)?pytest(?:\s|$)", + re.IGNORECASE, +) + +_VAGUE_PYTEST_CLAIM_RE = re.compile( + r"pytest\s+(?:executed|passed|failed)\b", + re.IGNORECASE, +) + + +def _parse_validation_command(report_text: str) -> str: + for line in (report_text or "").splitlines(): + match = _VALIDATION_CMD_LINE_RE.search(line) + if match: + return match.group(1).strip().rstrip(".") + return "" + + +def _parse_validation_cwd(report_text: str) -> str: + for line in (report_text or "").splitlines(): + match = _VALIDATION_CWD_LINE_RE.search(line) + if match: + return match.group(1).strip().rstrip(",.;") + return "" + + +def _is_bare_pytest_command(command: str) -> bool: + text = (command or "").strip() + if not text: + return False + if "/" in text or "\\" in text: + return False + return bool(_BARE_PYTEST_CMD_RE.match(text)) + + +def assess_validation_environment_proof( + *, + report_text: str | None = None, + validation_environment: dict | None = None, +) -> dict: + """#311: reviewer reports must include exact validation command/environment.""" + text = report_text or "" + lower = text.lower() + env = validation_environment or {} + reasons: list[str] = [] + violations: list[str] = [] + + claims_validation = bool(_VALIDATION_CLAIM_RE.search(text)) + if not claims_validation and not env.get("command"): + return { + "proven": True, + "block": False, + "claims_validation": False, + "reasons": [], + "violations": [], + "safe_next_action": "proceed", + } + + command = (env.get("command") or _parse_validation_command(text)).strip() + working_directory = ( + env.get("working_directory") or env.get("cwd") + or _parse_validation_cwd(text) + ).strip() + + if not command: + reasons.append( + "validation result claimed without exact validation command (#311)" + ) + elif _VAGUE_PYTEST_CLAIM_RE.search(text) and command == "": + violations.append( + "vague pytest summary without exact validation command (#311)" + ) + + if not working_directory: + reasons.append( + "validation result claimed without working directory (#311)" + ) + + has_result = any( + token in lower + for token in ("passed", "failed", "skipped", "error") + ) or any( + env.get(key) is not None + for key in ("passed", "failed", "skipped", "result") + ) + if not has_result: + reasons.append( + "validation result claimed without pass/fail result evidence (#311)" + ) + + if command and _is_bare_pytest_command(command): + which_pytest = ( + (env.get("which_pytest") or env.get("executable_path") or "") + .strip() + ) + if not which_pytest: + for line in text.splitlines(): + match = _WHICH_PYTEST_LINE_RE.search(line) + if match: + which_pytest = match.group(1).strip() + break + if not which_pytest: + reasons.append( + "bare pytest command requires which-pytest/executable proof (#311)" + ) + + pytest_version = (env.get("pytest_version") or "").strip() + if not pytest_version: + for line in text.splitlines(): + match = _PYTEST_VERSION_LINE_RE.search(line) + if match: + pytest_version = match.group(1).strip() + break + if not pytest_version: + reasons.append( + "bare pytest command requires pytest --version proof (#311)" + ) + + venv_resolved = env.get("venv_resolved") + if venv_resolved is None: + venv_resolved = any( + marker in lower + for marker in ( + "project venv", + "venv/bin/pytest", + "resolves to the project venv", + "venv resolved: yes", + ) + ) + if not venv_resolved and which_pytest and "venv" not in which_pytest: + reasons.append( + "bare pytest command requires proof whether executable " + "resolves to the project venv (#311)" + ) + + proven = not reasons and not violations + return { + "proven": proven, + "block": bool(violations), + "claims_validation": True, + "command": command or None, + "working_directory": working_directory or None, + "reasons": reasons, + "violations": violations, + "safe_next_action": ( + "report exact validation command, working directory, result, and " + "for bare pytest also which-pytest, pytest --version, and venv proof" + if not proven + else "proceed" + ), + } + + + + # --------------------------------------------------------------------------- # Reviewer queue ordering proof (#321) # --------------------------------------------------------------------------- @@ -3581,6 +3964,279 @@ def assess_reviewer_baseline_validation_proof( } +# --------------------------------------------------------------------------- +# Already-landed review gate (#292) +# --------------------------------------------------------------------------- + +ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED" + +ALREADY_LANDED_HANDOFF_FIELDS = ( + "PR number/title", + "candidate head SHA", + "target branch", + "target branch SHA", + "ancestor proof", + "linked issue status", + "recommended reconciliation action", + "capability proof for close/comment mutations", +) + +_FORBIDDEN_LANDED_STATES_RE = re.compile( + r"review decision\s*:\s*approved?\b|" + r"merge result\s*:\s*merged\b|" + r"\bready[_ ]to[_ ]merge\b|" + r"\bmerge[d]?\s*:\s*success\b", + re.IGNORECASE, +) + + +def assess_already_landed_review_gate( + *, + pr_number: int | None, + candidate_head_sha: str | None, + target_branch: str | None, + target_branch_sha: str | None, + head_is_ancestor_of_target: bool | None, + live_head_sha: str | None = None, + real_blocker: bool = False, + mergeable: bool | None = None, +) -> dict: + """#292: hard gate before any review mutation on an already-landed PR. + + Runs after PR selection and head pinning, before approval, request- + changes, or the merge API. *head_is_ancestor_of_target* is the result + of an ancestry check of *candidate_head_sha* against the freshly + fetched target branch at *target_branch_sha* (e.g. ``git merge-base + --is-ancestor``). ``None`` means the check was not run — the gate then + fails closed. + + *live_head_sha*, when provided, must equal *candidate_head_sha*; a + changed head invalidates the pinned ancestry result until re-checked. + + *mergeable* is accepted for caller convenience but never consulted: + conflict/mergeability handling is a separate gate. + """ + del mergeable # ancestry gate only; mergeability is a separate gate + reasons: list[str] = [] + + if not isinstance(pr_number, int) or pr_number <= 0: + reasons.append("PR number missing or invalid (#292)") + if not _FULL_SHA.match((candidate_head_sha or "").strip()): + reasons.append( + "candidate head SHA is not a full 40-hex commit SHA (#292)" + ) + if not (target_branch or "").strip(): + reasons.append("target branch missing (#292)") + if not _FULL_SHA.match((target_branch_sha or "").strip()): + reasons.append( + "target branch SHA is not a full 40-hex commit SHA; fetch the " + "target branch and record its SHA before the ancestry check " + "(#292)" + ) + if live_head_sha is not None and candidate_head_sha and ( + live_head_sha.strip() != candidate_head_sha.strip() + ): + reasons.append( + "PR head changed since the ancestry check was pinned; re-fetch " + "and re-run the already-landed gate (#292)" + ) + if head_is_ancestor_of_target is None: + reasons.append( + "ancestry of the PR head against the target branch was not " + "checked; the already-landed gate must run before any review " + "mutation (#292)" + ) + + if reasons: + return { + "state": "GATE_NOT_PROVEN", + "approve_allowed": False, + "request_changes_allowed": False, + "merge_api_allowed": False, + "reconciliation_handoff_required": False, + "required_handoff_fields": (), + "reasons": reasons, + "safe_next_action": ( + "fetch the target branch, pin the live PR head, run the " + "ancestry check, then re-run this gate" + ), + } + + if head_is_ancestor_of_target: + return { + "state": ALREADY_LANDED_STATE, + "approve_allowed": False, + "request_changes_allowed": bool(real_blocker), + "merge_api_allowed": False, + "reconciliation_handoff_required": True, + "required_handoff_fields": ALREADY_LANDED_HANDOFF_FIELDS, + "reasons": [ + f"PR #{pr_number} head {candidate_head_sha} is already an " + f"ancestor of {target_branch} @ {target_branch_sha}; the PR " + "is reconciliation-only (#292)" + ], + "safe_next_action": ( + "stop before review mutation and emit an " + f"{ALREADY_LANDED_STATE} reconciliation handoff; any PR/issue " + "close or comment requires exact capability proof" + ), + } + + return { + "state": "NORMAL_REVIEW_CANDIDATE", + "approve_allowed": True, + "request_changes_allowed": True, + "merge_api_allowed": True, + "reconciliation_handoff_required": False, + "required_handoff_fields": (), + "reasons": [], + "safe_next_action": "proceed with the normal review workflow gates", + } + + +def assess_already_landed_report_state( + report_text: str | None, + *, + gate_fired: bool, +) -> dict: + """#292: reject APPROVED/MERGED/READY_TO_MERGE after the gate fired. + + *gate_fired* comes from the session's own gate result, so a report + that omits the already-landed markers entirely still fails closed — + unlike the text-only #327 wording rules this does not depend on the + report admitting the PR was already landed. + """ + if not gate_fired: + return {"complete": True, "block": False, "reasons": []} + + text = report_text or "" + reasons: list[str] = [] + + if _FORBIDDEN_LANDED_STATES_RE.search(text): + reasons.append( + "report claims APPROVED/MERGED/READY_TO_MERGE although the " + "already-landed gate fired (#292)" + ) + if ALREADY_LANDED_STATE.lower() not in text.lower(): + reasons.append( + f"report must state {ALREADY_LANDED_STATE} when the " + "already-landed gate fired (#292)" + ) + + return { + "complete": not reasons, + "block": bool(reasons), + "reasons": reasons, + } + + +# --------------------------------------------------------------------------- +# Full-suite failure approval gate (#323) +# --------------------------------------------------------------------------- + + +def assess_full_suite_failure_approval_gate( + *, + review_decision: str, + full_suite_passed: bool | None, + baseline_validation: dict | None = None, + baseline_proof: dict | None = None, + report_text: str | None = None, + project_root: str | None = None, +) -> dict: + """#323: approving a PR whose full suite fails requires baseline proof. + + Default action on a full-suite failure is REQUEST_CHANGES. Approval is + allowed only when the #325 baseline proof passes in full (branches/ + worktree, pinned SHAs, matching failure signatures) and the extra #323 + fields prove the failures are pre-existing: ``pr_suite_command``, + ``baseline_suite_command``, ``new_tests_passed``, and a non-empty + ``unrelated_explanation``. + + *review_decision* is the decision the workflow intends to submit; only + ``approve`` can be blocked. ``block`` is True when an approval was + requested that the proofs do not allow. + """ + decision = (review_decision or "").strip().lower().replace("-", "_") + wants_approval = decision == "approve" + reasons: list[str] = [] + + if full_suite_passed is True: + return { + "approve_allowed": True, + "block": False, + "default_action": "proceed", + "reasons": [], + } + + default_action = "request_changes" + + if full_suite_passed is None: + reasons.append( + "full-suite result not proven; approval requires a recorded " + "full-suite command and result (#323)" + ) + else: + if baseline_validation is None: + baseline_validation = assess_reviewer_baseline_validation_proof( + baseline_proof=baseline_proof, + report_text=report_text, + project_root=project_root, + ) + proof = baseline_proof or {} + + if not proof: + reasons.append( + "full suite failed but no baseline proof was provided; " + "default action is REQUEST_CHANGES (#323)" + ) + if not baseline_validation.get("proven"): + reasons.append( + "baseline validation proof missing or failed (#323/#325)" + ) + reasons.extend(baseline_validation.get("reasons", [])) + + if proof: + worktree = (proof.get("worktree_path") or "").strip() + if not _path_under_branches(worktree, project_root): + reasons.append( + "baseline comparison must run in a clean baseline " + "worktree under branches/, not the main checkout (#323)" + ) + if not (proof.get("pr_suite_command") or "").strip(): + reasons.append( + "PR full-suite command/result missing from baseline " + "proof (#323)" + ) + if not (proof.get("baseline_suite_command") or "").strip(): + reasons.append( + "baseline full-suite command/result missing from " + "baseline proof (#323)" + ) + if proof.get("new_tests_passed") is not True: + reasons.append( + "proof that new/changed tests passed is missing (#323)" + ) + if not (proof.get("unrelated_explanation") or "").strip(): + reasons.append( + "explanation why failures are unrelated to the PR is " + "missing (#323)" + ) + if proof.get("failure_signatures_match") is not True: + reasons.append( + "PR and baseline failure signatures do not match; " + "request changes (#323)" + ) + + approve_allowed = not reasons + return { + "approve_allowed": approve_allowed, + "block": wants_approval and not approve_allowed, + "default_action": "proceed" if approve_allowed else default_action, + "reasons": reasons, + } + + # --------------------------------------------------------------------------- # Identity disclosure (#305) # --------------------------------------------------------------------------- @@ -4059,6 +4715,20 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None): } +def assess_merge_simulation_report(report_text, **kwargs): + """#317: classify local merge simulation as worktree/index mutations.""" + from reviewer_merge_simulation import assess_merge_simulation_report as _assess + + return _assess(report_text, **kwargs) + + +def assess_validation_integrity_report(report_text, **kwargs): + """#316: separate official PR-head validation from diagnostic experiments.""" + from reviewer_validation_integrity import assess_validation_integrity_report as _assess + + return _assess(report_text, **kwargs) + + def assess_prior_blocker_skip_proof(report_text, **kwargs): """#318: require live blocker proof before skipping earlier open PRs.""" from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess @@ -4071,3 +4741,26 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess return _assess(report_text, **kwargs) + + +def assess_validation_worktree_edit_report(report_text, **kwargs): + """#315: block edits in PR validation worktrees; require diagnostic separation.""" + from reviewer_validation_worktree_edits import ( + assess_validation_worktree_edit_report as _assess, + ) + + return _assess(report_text, **kwargs) + + +def assess_mutation_categories_report(report_text, **kwargs): + """#319: require precise mutation categories in reviewer controller handoffs.""" + from reviewer_mutation_categories import assess_mutation_categories_report as _assess + + return _assess(report_text, **kwargs) + + +def assess_worktree_ownership_report(report_text, **kwargs): + """#312: prove session-owned or safe-reuse worktree before reset/validation.""" + from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_merge_simulation.py b/reviewer_merge_simulation.py new file mode 100644 index 0000000..5397c4f --- /dev/null +++ b/reviewer_merge_simulation.py @@ -0,0 +1,148 @@ +"""Merge-simulation worktree mutation verifier for reviewer reports (#317).""" + +from __future__ import annotations + +import re +from typing import Any + +_WORKTREE_INDEX_FIELD_RE = re.compile( + r"^\s*worktree/index mutations\s*:\s*(.+?)\s*$", + re.IGNORECASE | re.MULTILINE, +) +_READONLY_DIAG_RE = re.compile( + r"read[- ]only diagnostics\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_WORKTREE_PATH_RE = re.compile( + r"(?:worktree path|diagnostic worktree|simulation worktree)\s*:", + re.IGNORECASE, +) +_PRE_CLEAN_RE = re.compile( + r"(?:pre[- ]simulation|before simulation|clean before).*(?:clean|status)", + re.IGNORECASE, +) +_MERGE_RESULT_RE = re.compile( + r"(?:merge result|simulation result|conflict status|merge conflict)", + re.IGNORECASE, +) +_ABORT_RE = re.compile( + r"(?:merge --abort|abort/reset command|abort command|git merge --abort)", + re.IGNORECASE, +) +_POST_CLEAN_RE = re.compile( + r"(?:post[- ]abort|after abort|clean after).*(?:clean|status)", + re.IGNORECASE, +) + + +def _command_text(entry: Any) -> str: + if isinstance(entry, dict): + return str(entry.get("command") or "").strip() + return str(entry or "").strip() + + +def _git_subcommand_line(command: str) -> str | None: + tokens = command.split() + if not tokens or tokens[0] != "git": + return None + return " ".join(tokens[1:]) + + +def merge_simulation_commands(command_log: list | None) -> list[str]: + """Return logged commands that perform local merge simulation (#317).""" + out: list[str] = [] + for entry in command_log or []: + command = _command_text(entry) + rest = _git_subcommand_line(command) + if rest is None: + continue + lower = command.lower() + if rest.startswith("merge"): + if "--no-commit" in lower or ( + "--no-ff" in lower and "merge" in lower and "--commit" not in lower + ): + out.append(command) + elif "--abort" in lower: + out.append(command) + return out + + +def assess_merge_simulation_report( + report_text: str, + *, + command_log: list | None = None, +) -> dict[str, Any]: + """Reject merge simulation classified as read-only; require worktree proof (#317).""" + text = report_text or "" + lower = text.lower() + reasons: list[str] = [] + simulation_commands = merge_simulation_commands(command_log) + has_abort = any("--abort" in c.lower() for c in simulation_commands) + has_simulation = any( + "--no-commit" in c.lower() or ( + "--no-ff" in c.lower() and "--abort" not in c.lower() + ) + for c in simulation_commands + ) + + if not simulation_commands: + return { + "proven": True, + "block": False, + "reasons": [], + "simulation_commands": [], + "safe_next_action": "proceed", + } + + field = _WORKTREE_INDEX_FIELD_RE.search(text) + value = (field.group(1).strip().lower() if field else "") or "" + if not value or value in {"none", "n/a", "no"}: + reasons.append( + "merge simulation commands ran but report lacks " + "'Worktree/index mutations' entry" + ) + elif "merge" not in value and "simulation" not in value: + if not any(cmd.lower() in lower for cmd in simulation_commands): + reasons.append( + "Worktree/index mutations must document merge simulation commands" + ) + + readonly = _READONLY_DIAG_RE.search(text) + if readonly: + readonly_value = readonly.group(1) + for command in simulation_commands: + if command.lower() in readonly_value.lower(): + reasons.append( + "merge simulation may not be listed under read-only diagnostics" + ) + if has_simulation and "merge simulation" in readonly_value.lower(): + reasons.append( + "merge simulation may not be classified as read-only diagnostics" + ) + + if has_simulation: + if not _WORKTREE_PATH_RE.search(text): + reasons.append("merge simulation report missing worktree path") + if not _PRE_CLEAN_RE.search(text): + reasons.append("merge simulation report missing pre-simulation clean status") + if not _MERGE_RESULT_RE.search(text): + reasons.append("merge simulation report missing merge result/conflict status") + if has_abort or has_simulation: + if has_abort and not _ABORT_RE.search(text): + reasons.append("merge simulation report missing abort/reset command proof") + if has_abort and not _POST_CLEAN_RE.search(text): + reasons.append("merge simulation report missing post-abort clean status") + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "simulation_commands": simulation_commands, + "safe_next_action": ( + "classify merge simulation under Worktree/index mutations with full " + "pre/merge/abort/post-clean proof; never list as read-only diagnostics" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/reviewer_mutation_categories.py b/reviewer_mutation_categories.py new file mode 100644 index 0000000..a09e73b --- /dev/null +++ b/reviewer_mutation_categories.py @@ -0,0 +1,130 @@ +"""Precise mutation category verifier for reviewer controller handoffs (#319).""" + +from __future__ import annotations + +import re +from typing import Any + +_CONTROLLER_HANDOFF_RE = re.compile(r"controller handoff", re.IGNORECASE) +_WORKSPACE_MUTATIONS_RE = re.compile( + r"^\s*[-*]?\s*workspace\s+mutations\s*:", + re.IGNORECASE | re.MULTILINE, +) +_VAGUE_MUTATIONS_NONE_RE = re.compile( + r"^\s*[-*]?\s*mutations\s*:\s*none\s*$", + re.IGNORECASE | re.MULTILINE, +) + +_REQUIRED_CATEGORIES = ( + "file edits by reviewer", + "worktree/index mutations", + "git ref mutations", +) +_WORKTREE_FIELD_ALIASES = ( + "worktree/index mutations", + "worktree mutations", +) + + +def _has_category(text: str, category: str) -> bool: + pattern = re.compile( + rf"^\s*[-*]?\s*{re.escape(category)}\s*:", + re.IGNORECASE | re.MULTILINE, + ) + return bool(pattern.search(text)) + + +def _has_worktree_category(text: str) -> bool: + return any(_has_category(text, alias) for alias in _WORKTREE_FIELD_ALIASES) + + +def _normalize_for_consistency_check(text: str) -> str: + """Map #319 precise labels to #313 field names for consistency delegation.""" + normalized = text + if _has_category(text, "worktree/index mutations") and not _has_category( + text, "worktree mutations" + ): + normalized = re.sub( + r"(worktree/index mutations\s*:)", + "Worktree mutations:", + normalized, + flags=re.IGNORECASE, + ) + return normalized + + +def assess_mutation_categories_report( + report_text: str, + *, + handoff_session: dict | None = None, + observed_commands: list | None = None, +) -> dict[str, Any]: + """Require precise mutation categories in reviewer controller handoffs (#319).""" + text = report_text or "" + session = dict(handoff_session or {}) + reasons: list[str] = [] + lower = text.lower() + + is_controller = bool( + session.get("controller_handoff") + or _CONTROLLER_HANDOFF_RE.search(text) + ) + if not is_controller and not session.get("require_categories"): + return { + "proven": True, + "block": False, + "reasons": [], + "controller_handoff": False, + "safe_next_action": "proceed", + } + + if _WORKSPACE_MUTATIONS_RE.search(text): + reasons.append( + "controller handoffs must not use legacy 'Workspace mutations'; " + "use precise mutation category fields" + ) + + if _VAGUE_MUTATIONS_NONE_RE.search(text): + reasons.append( + "controller handoffs must not use vague 'Mutations: none'; " + "report each precise category separately" + ) + + missing = [ + category + for category in _REQUIRED_CATEGORIES + if category != "worktree/index mutations" and not _has_category(text, category) + ] + if not _has_worktree_category(text): + missing.append("worktree/index mutations") + + if missing: + reasons.append( + "controller handoff missing precise mutation categories: " + + ", ".join(missing) + ) + + commands = observed_commands or session.get("observed_commands") + if commands: + from review_proofs import assess_workspace_mutation_consistency + + consistency = assess_workspace_mutation_consistency( + _normalize_for_consistency_check(text), + commands, + ) + if not consistency.get("complete"): + reasons.extend(consistency.get("reasons") or []) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "controller_handoff": True, + "safe_next_action": ( + "replace Workspace mutations with file edits, worktree/index, git ref, " + "and other precise mutation category fields" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/reviewer_validation_integrity.py b/reviewer_validation_integrity.py new file mode 100644 index 0000000..038c457 --- /dev/null +++ b/reviewer_validation_integrity.py @@ -0,0 +1,148 @@ +"""PR-head vs diagnostic validation integrity verifier (#316).""" + +from __future__ import annotations + +import re +from typing import Any + +_DIAGNOSTIC_LABEL_RE = re.compile( + r"diagnostic local experiment|not pr-head validation|diagnostic-only validation", + re.IGNORECASE, +) +_OFFICIAL_VALIDATION_RE = re.compile( + r"(?:official validation|pr-head validation|validation integrity)", + re.IGNORECASE, +) +_OFFICIAL_COMMAND_RE = re.compile( + r"(?:official validation command|pr-head validation command|validation command)", + re.IGNORECASE, +) +_OFFICIAL_RESULT_RE = re.compile( + r"(?:official validation result|pr-head validation result|validation result|validation integrity status)", + re.IGNORECASE, +) +_DIRTY_AFTER_RE = re.compile( + r"(?:worktree dirty after (?:official )?validation|dirty (?:state )?after (?:official )?validation|" + r"validation worktree dirty after)", + re.IGNORECASE, +) +_DIAGNOSTIC_EDIT_RE = re.compile( + r"(?:diagnostic edit(?:ed)? path|file edits by reviewer|diagnostic local edit)", + re.IGNORECASE, +) +_DIAGNOSTIC_COMMAND_RE = re.compile( + r"(?:diagnostic (?:validation )?command|diagnostic test run|diagnostic result)", + re.IGNORECASE, +) +_SUGGESTED_FIX_RE = re.compile( + r"(?:diagnostic results? (?:were )?used only for suggested fix|suggested fix only|" + r"not used as official validation)", + re.IGNORECASE, +) +_INTEGRITY_STATUS_RE = re.compile( + r"validation integrity status\s*:\s*(passed|failed|not run|contaminated)", + re.IGNORECASE, +) +_POST_EDIT_AS_OFFICIAL_RE = re.compile( + r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit|" + r"full suite passed after diagnostic edit)", + re.IGNORECASE, +) + + +def _performed_edits(action_log: list[dict] | None) -> list[str]: + paths: list[str] = [] + for entry in action_log or []: + if entry.get("gated_rejected") or entry.get("performed") is False: + continue + path = entry.get("path") + if path and entry.get("kind", "file_edit") in {"file_edit", "edit", "write"}: + paths.append(str(path)) + return paths + + +def assess_validation_integrity_report( + report_text: str, + *, + validation_session: dict | None = None, + action_log: list[dict] | None = None, +) -> dict[str, Any]: + """Separate official PR-head validation from diagnostic edited-worktree tests (#316).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + + diagnostic_edits = list(session.get("diagnostic_edits") or []) + if not diagnostic_edits: + diagnostic_edits = _performed_edits(action_log) + diagnostic_ran = bool( + session.get("diagnostic_validation_ran") + or session.get("diagnostic_runs") + ) + official_ran = bool(session.get("official_validation_ran", True)) + official_result = (session.get("official_result") or "").strip().lower() + diagnostic_result = (session.get("diagnostic_result") or "").strip().lower() + dirty_after = session.get("worktree_dirty_after_official") + contaminated = bool(session.get("contaminated")) + + if official_ran: + if not _OFFICIAL_VALIDATION_RE.search(text) and not _OFFICIAL_COMMAND_RE.search(text): + reasons.append("report missing official PR-head validation section") + if not _OFFICIAL_COMMAND_RE.search(text) and "validation:" not in text.lower(): + reasons.append("report missing official validation command") + if not _OFFICIAL_RESULT_RE.search(text): + reasons.append("report missing official validation result or integrity status") + if dirty_after is not None and not _DIRTY_AFTER_RE.search(text): + reasons.append( + "report missing worktree dirty state after official validation" + ) + elif dirty_after is None and official_ran and not _DIRTY_AFTER_RE.search(text): + reasons.append( + "report missing worktree dirty state after official validation" + ) + + if diagnostic_edits or diagnostic_ran: + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "post-edit diagnostic runs must be labeled " + "'Diagnostic local experiment — not PR-head validation'" + ) + if diagnostic_edits and not _DIAGNOSTIC_EDIT_RE.search(text): + reasons.append("report missing diagnostic edit path") + if diagnostic_ran and not _DIAGNOSTIC_COMMAND_RE.search(text): + reasons.append("report missing diagnostic command/result") + if not _SUGGESTED_FIX_RE.search(text): + reasons.append( + "report must state diagnostic results were used only for suggested fix" + ) + if _POST_EDIT_AS_OFFICIAL_RE.search(text): + reasons.append( + "post-edit diagnostic results cannot be reported as official PR-head validation" + ) + if diagnostic_result == "pass" and official_result == "fail": + if "request_changes" not in text.lower() and "failed" not in text.lower(): + reasons.append( + "request-changes must cite unmodified PR-head failure, not diagnostic pass" + ) + + if contaminated: + status = _INTEGRITY_STATUS_RE.search(text) + if not status or "contaminated" not in status.group(1).lower(): + reasons.append( + "contaminated validation must report integrity status " + "'contaminated — recovery required'" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "diagnostic_edits": diagnostic_edits, + "safe_next_action": ( + "separate official PR-head validation from diagnostic experiments; " + "report dirty-after-validation state" + if reasons + else "proceed" + ), + } diff --git a/reviewer_validation_worktree_edits.py b/reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..c4e904f --- /dev/null +++ b/reviewer_validation_worktree_edits.py @@ -0,0 +1,184 @@ +"""PR validation worktree read-only edit verifier (#315).""" + +from __future__ import annotations + +import re +from typing import Any + +_VALIDATION_WORKTREE_RE = re.compile( + r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_RE = re.compile( + r"branches/(?:diagnostic[\w/-]*|scratch[\w/-]*)", + re.IGNORECASE, +) +_FILE_EDITS_NONE_RE = re.compile( + r"file edits by reviewer\s*:\s*none\b", + re.IGNORECASE, +) +_FILE_EDITS_FIELD_RE = re.compile( + r"file edits by reviewer\s*:\s*(.+)", + re.IGNORECASE, +) +_VALIDATION_WORKTREE_PATH_RE = re.compile( + r"(?:validation worktree path|review worktree path)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_PATH_RE = re.compile( + r"(?:diagnostic (?:scratch )?worktree path|diagnostic worktree)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_LABEL_RE = re.compile( + r"diagnostic local experiment|not pr-head validation|diagnostic-only", + re.IGNORECASE, +) +_OFFICIAL_VALIDATION_RE = re.compile( + r"(?:official (?:pr-head )?validation|pr-head validation result)", + re.IGNORECASE, +) +_POST_EDIT_OFFICIAL_RE = re.compile( + r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit)", + re.IGNORECASE, +) + +_PERFORMED_EDIT_KINDS = frozenset({"file_edit", "edit", "write", "edited", "created", "wrote"}) + + +def _normalize_path(path: str) -> str: + return (path or "").replace("\\", "/").strip() + + +def _is_validation_worktree(path: str) -> bool: + return bool(_VALIDATION_WORKTREE_RE.search(_normalize_path(path))) + + +def _is_diagnostic_worktree(path: str) -> bool: + normalized = _normalize_path(path) + return bool( + _DIAGNOSTIC_WORKTREE_RE.search(normalized) + or "diagnostic" in normalized.lower() + ) + + +def _performed_edits(action_log: list[dict] | None) -> list[dict[str, str]]: + edits: list[dict[str, str]] = [] + for entry in action_log or []: + if entry.get("gated_rejected") or entry.get("performed") is False: + continue + kind = (entry.get("kind") or entry.get("action") or "file_edit").strip().lower() + if kind not in _PERFORMED_EDIT_KINDS: + continue + path = (entry.get("path") or "").strip() + if not path: + continue + worktree = _normalize_path(str(entry.get("worktree_path") or entry.get("cwd") or "")) + edits.append({"path": path, "worktree_path": worktree}) + return edits + + +def _extract_validation_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("validation_worktree_path") or "")) + if path: + return path + match = _VALIDATION_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def _extract_diagnostic_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("diagnostic_worktree_path") or "")) + if path: + return path + match = _DIAGNOSTIC_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def assess_validation_worktree_edit_report( + report_text: str, + *, + validation_session: dict | None = None, + action_log: list[dict] | None = None, +) -> dict[str, Any]: + """Block edits in PR validation worktrees; require diagnostic separation (#315).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + + validation_path = _extract_validation_path(text, session) + diagnostic_path = _extract_diagnostic_path(text, session) + validation_edits = list(session.get("validation_worktree_edits") or []) + diagnostic_edits = list(session.get("diagnostic_edits") or []) + edits = _performed_edits(action_log) + + if not validation_edits: + for entry in edits: + wt = entry.get("worktree_path") or validation_path + if wt and _is_validation_worktree(wt) and not _is_diagnostic_worktree(wt): + validation_edits.append(entry["path"]) + + if not diagnostic_edits: + for entry in edits: + wt = entry.get("worktree_path") or "" + if wt and _is_diagnostic_worktree(wt): + diagnostic_edits.append(entry["path"]) + elif entry["path"] and diagnostic_path and not validation_edits: + if _is_diagnostic_worktree(diagnostic_path): + diagnostic_edits.append(entry["path"]) + + claimed_none = bool(_FILE_EDITS_NONE_RE.search(text)) + any_edits = bool(validation_edits or diagnostic_edits or edits) + + if validation_edits: + reasons.append( + "reviewer must not edit files in the PR validation worktree; " + f"observed edits: {', '.join(validation_edits)}" + ) + + if diagnostic_edits or session.get("diagnostic_experiment"): + if not diagnostic_path and not _DIAGNOSTIC_WORKTREE_PATH_RE.search(text): + reasons.append( + "diagnostic experiments require a separate diagnostic scratch worktree path" + ) + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "diagnostic experiments must be labeled " + "'Diagnostic local experiment — not PR-head validation'" + ) + if not _FILE_EDITS_FIELD_RE.search(text) or claimed_none: + reasons.append( + "diagnostic edits must be reported under 'File edits by reviewer'" + ) + + if any_edits and claimed_none: + reasons.append( + "report claims 'File edits by reviewer: none' but reviewer file edits occurred" + ) + + if session.get("official_validation_after_edit") or _POST_EDIT_OFFICIAL_RE.search(text): + if validation_edits or (validation_path and diagnostic_edits): + reasons.append( + "post-edit test results cannot be reported as official PR-head validation" + ) + + if validation_edits and _OFFICIAL_VALIDATION_RE.search(text): + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "validation worktree was edited; official PR-head validation is no longer valid" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "validation_worktree_path": validation_path or None, + "diagnostic_worktree_path": diagnostic_path or None, + "validation_worktree_edits": validation_edits, + "diagnostic_edits": diagnostic_edits, + "safe_next_action": ( + "keep PR validation worktrees read-only; use a separate diagnostic " + "scratch worktree and report all reviewer file edits" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/reviewer_worktree_ownership.py b/reviewer_worktree_ownership.py new file mode 100644 index 0000000..d30c48d --- /dev/null +++ b/reviewer_worktree_ownership.py @@ -0,0 +1,207 @@ +"""Reviewer worktree ownership and safe-reuse proof verifier (#312).""" + +from __future__ import annotations + +import re +from typing import Any + +_SESSION_OWNED_RE = re.compile( + r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)", + re.IGNORECASE, +) +_WORKTREE_PATH_RE = re.compile( + r"(?:review worktree path|worktree path|session-owned worktree)\s*:\s*(\S+)", + re.IGNORECASE, +) +_BRANCHES_PATH_RE = re.compile(r"/branches/|\bbranches/", re.IGNORECASE) +_MAIN_CHECKOUT_RE = re.compile( + r"main checkout|not (?:the )?main checkout|outside branches/", + re.IGNORECASE, +) +_SAFE_REUSE_RE = re.compile(r"safe[- ]reuse proof", re.IGNORECASE) +_REUSE_POLICY_RE = re.compile( + r"(?:project policy|policy) (?:allows|allowing) (?:reuse|reset)", + re.IGNORECASE, +) +_CLEAN_TRACKED_RE = re.compile( + r"(?:clean tracked state|tracked state\s*:\s*clean|no uncommitted tracked)", + re.IGNORECASE, +) +_CLEAN_UNTRACKED_RE = re.compile( + r"(?:clean untracked state|untracked state\s*:\s*clean|no untracked files)", + re.IGNORECASE, +) +_NOT_OTHER_SESSION_RE = re.compile( + r"not owned by (?:another|other) (?:active )?(?:task|session)", + re.IGNORECASE, +) +_BRANCH_BEFORE_RESET_RE = re.compile( + r"(?:branch/head before reset|head before reset|branch before reset)\s*:\s*(\S+)", + re.IGNORECASE, +) +_RESET_TARGET_RE = re.compile( + r"(?:reset target sha|reset target)\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE, +) +_DESTRUCTIVE_CMD_RE = re.compile( + r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?" + r"(?:reset\s+--hard|clean(?:\s+-[A-Za-z]+)*|checkout\s+(?!HEAD\s+--)|switch\s+)", + re.IGNORECASE, +) +_WORKSPACE_NONE_RE = re.compile(r"workspace mutations\s*:\s*none", re.IGNORECASE) +_WORKTREE_MUTATION_RE = re.compile( + r"(?:worktree(?:/index)? mutations|destructive reset)\s*:\s*(?!none\b).+", + re.IGNORECASE, +) +_RESET_IN_REPORT_RE = re.compile(r"reset\s+--hard", re.IGNORECASE) + + +def _command_text(entry) -> str: + if isinstance(entry, dict): + return str(entry.get("command") or "").strip() + return str(entry or "").strip() + + +def _destructive_commands(command_log: list | None) -> list[str]: + return [ + cmd + for cmd in (_command_text(entry) for entry in (command_log or [])) + if cmd and _DESTRUCTIVE_CMD_RE.search(cmd) + ] + + +def _extract_worktree_path(text: str, session: dict) -> str: + path = (session.get("worktree_path") or "").strip() + if path: + return path + match = _WORKTREE_PATH_RE.search(text or "") + return match.group(1).strip() if match else "" + + +def _is_session_owned_path(path: str) -> bool: + normalized = (path or "").replace("\\", "/") + return bool(_SESSION_OWNED_RE.search(normalized)) + + +def _safe_reuse_fields_present(text: str) -> list[str]: + missing: list[str] = [] + if not _WORKTREE_PATH_RE.search(text): + missing.append("exact worktree path") + if not _BRANCHES_PATH_RE.search(text): + missing.append("worktree inside branches/") + if not _MAIN_CHECKOUT_RE.search(text): + missing.append("worktree is not the main checkout") + if not _NOT_OTHER_SESSION_RE.search(text): + missing.append("worktree not owned by another active session") + if not _CLEAN_TRACKED_RE.search(text): + missing.append("clean tracked state") + if not _CLEAN_UNTRACKED_RE.search(text): + missing.append("clean untracked state") + if not _BRANCH_BEFORE_RESET_RE.search(text): + missing.append("branch/head before reset") + if not _RESET_TARGET_RE.search(text): + missing.append("reset target SHA") + if not _REUSE_POLICY_RE.search(text): + missing.append("explicit project policy allowing reuse/reset") + return missing + + +def assess_worktree_ownership_report( + report_text: str, + *, + ownership_session: dict | None = None, + command_log: list | None = None, +) -> dict[str, Any]: + """Prove reviewer worktree ownership before reset or validation (#312).""" + text = report_text or "" + session = dict(ownership_session or {}) + reasons: list[str] = [] + + worktree_path = _extract_worktree_path(text, session) + destructive = _destructive_commands(command_log or session.get("command_log")) + if not destructive and session.get("destructive_reset"): + destructive = ["git reset --hard (session)"] + + session_owned = bool( + session.get("session_owned") + or (worktree_path and _is_session_owned_path(worktree_path)) + ) + safe_reuse = bool(session.get("safe_reuse") or _SAFE_REUSE_RE.search(text)) + main_checkout = bool(session.get("main_checkout")) + dirty_tracked = session.get("dirty_tracked") + dirty_untracked = session.get("dirty_untracked") + other_session_owned = bool(session.get("other_session_owned")) + + if worktree_path or destructive or session.get("validation_ran"): + if not worktree_path: + reasons.append("report missing exact review worktree path") + elif main_checkout or "branches/" not in worktree_path.replace("\\", "/"): + reasons.append("review worktree must be inside branches/, not the main checkout") + elif not _BRANCHES_PATH_RE.search(worktree_path.replace("\\", "/")): + reasons.append("review worktree path must be under branches/") + + if other_session_owned and not safe_reuse: + reasons.append( + "reused worktree appears owned by another active task/session; " + "safe-reuse proof required" + ) + + if dirty_tracked: + reasons.append( + "worktree has uncommitted tracked changes before reset or validation" + ) + if dirty_untracked and safe_reuse: + reasons.append("safe-reuse proof requires clean untracked state") + + if safe_reuse and not session_owned: + reasons.extend( + f"safe-reuse proof missing {field}" + for field in _safe_reuse_fields_present(text) + ) + + if destructive: + if not session_owned and not safe_reuse: + reasons.append( + "destructive worktree commands forbidden without session-owned " + "worktree or safe-reuse proof" + ) + if _WORKSPACE_NONE_RE.search(text) and ( + _RESET_IN_REPORT_RE.search(text) or any("reset" in c.lower() for c in destructive) + ): + reasons.append( + "final report must not claim 'Workspace mutations: none' when " + "git reset --hard occurred" + ) + if not _WORKTREE_MUTATION_RE.search(text) and not _RESET_IN_REPORT_RE.search(text): + reasons.append( + "destructive reset must be reported under Worktree/index mutations " + "or destructive reset operations" + ) + + if ( + worktree_path + and not session_owned + and not safe_reuse + and (destructive or session.get("validation_ran")) + and not _is_session_owned_path(worktree_path) + ): + reasons.append( + "reused branch-named worktree requires safe-reuse proof before reset or validation" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "worktree_path": worktree_path or None, + "session_owned": session_owned, + "safe_reuse": safe_reuse, + "destructive_commands": destructive, + "safe_next_action": ( + "use a fresh session-owned review worktree under branches/review-pr-* " + "or document full safe-reuse proof before destructive reset" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/task_capability_map.py b/task_capability_map.py index 9d162c1..2b794e7 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -96,6 +96,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.read", "role": "author", }, + "post_heartbeat": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "reconcile_issue_claims": { + "permission": "gitea.read", + "role": "author", + }, + "cleanup_stale_claims": { + "permission": "gitea.issue.comment", + "role": "author", + }, } # Issue-mutating MCP tools and their resolver task keys. diff --git a/tests/test_agent_temp_artifacts.py b/tests/test_agent_temp_artifacts.py index 2e14385..903eab8 100644 --- a/tests/test_agent_temp_artifacts.py +++ b/tests/test_agent_temp_artifacts.py @@ -62,7 +62,24 @@ class TestPreflightWarnings(unittest.TestCase): self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0]) +# Issue-write tools are profile-gated (#69); gitea_lock_issue requires +# gitea.issue.comment (see task_capability_map), so the gate must be +# seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359). +ISSUE_WRITE_ENV = { + "GITEA_ALLOWED_OPERATIONS": ( + "gitea.issue.create,gitea.issue.close,gitea.issue.comment" + ), +} + + class TestIssueLockArtifactWarning(unittest.TestCase): + def setUp(self): + self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True) + self._env_patcher.start() + + def tearDown(self): + self._env_patcher.stop() + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server._auth", return_value="token x") @patch("mcp_server._resolve", return_value=("h", "o", "r")) @@ -72,6 +89,9 @@ class TestIssueLockArtifactWarning(unittest.TestCase): mock_state.return_value = { "current_branch": "master", "porcelain_status": "?? _emit_payload.py\n", + "base_equivalent": True, + "inspected_git_root": "/scratch/wt", + "base_branch": "origin/master", } result = mcp_server.gitea_lock_issue( issue_number=261, diff --git a/tests/test_already_landed_report_wording.py b/tests/test_already_landed_report_wording.py new file mode 100644 index 0000000..ac0379f --- /dev/null +++ b/tests/test_already_landed_report_wording.py @@ -0,0 +1,151 @@ +"""Tests for already-landed final-report wording validator (#298). + +An already-landed PR is reconciliation-only, never review/merge +eligible, and a head SHA may not be called reviewed unless validation +and diff review actually passed. Final reports and controller handoffs +must use the reconciliation wording, not the legacy eligible/reviewed +fields. +""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import assess_already_landed_report_wording # noqa: E402 + + +GOOD_ALREADY_LANDED = ( + "Controller Handoff\n" + "- Oldest open PR requiring action: PR #278\n" + "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n" + "- Candidate head SHA: 2a544b7\n" + "- Reviewed head SHA: none\n" + "- Review worktree used: false\n" +) + + +class TestAlreadyLandedWording(unittest.TestCase): + def test_correct_reconciliation_wording_passes(self): + result = assess_already_landed_report_wording( + GOOD_ALREADY_LANDED, already_landed=True + ) + self.assertTrue(result["complete"]) + self.assertFalse(result["downgraded"]) + self.assertEqual(result["reasons"], []) + + def test_oldest_eligible_wording_fails(self): + report = GOOD_ALREADY_LANDED + "- oldest eligible PR: PR #278\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("oldest eligible pr" in r.lower() for r in result["reasons"]) + ) + + def test_next_eligible_wording_fails(self): + report = GOOD_ALREADY_LANDED + "- next eligible PR: PR #278\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + + def test_pinned_reviewed_head_fails(self): + report = GOOD_ALREADY_LANDED + "- Pinned reviewed head: 2a544b7\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("pinned reviewed head" in r.lower() for r in result["reasons"]) + ) + + def test_scratch_worktree_field_fails(self): + report = GOOD_ALREADY_LANDED + "- Scratch worktree used: False\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + + def test_missing_eligibility_class_fails(self): + report = ( + "Controller Handoff\n" + "- Oldest open PR requiring action: PR #278\n" + "- Candidate head SHA: 2a544b7\n" + "- Reviewed head SHA: none\n" + "- Review worktree used: false\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("eligibility class" in r.lower() for r in result["reasons"]) + ) + + def test_populated_reviewed_head_sha_fails(self): + report = GOOD_ALREADY_LANDED.replace( + "- Reviewed head SHA: none\n", + "- Reviewed head SHA: 2a544b7\n", + ) + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("reviewed head" in r.lower() for r in result["reasons"]) + ) + + +class TestNonAlreadyLandedReports(unittest.TestCase): + def test_normal_reviewed_report_passes(self): + report = ( + "- Selected PR: 281\n" + "- Pinned reviewed head: abc1234\n" + "- Review decision: approve\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=True + ) + self.assertTrue(result["complete"]) + + def test_blocked_infra_report_with_pinned_head_fails(self): + report = ( + "- Selected PR: 281\n" + "- Pinned reviewed head: abc1234\n" + "- Current status: blocked on infra_stop\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("before validation" in r.lower() for r in result["reasons"]) + ) + + def test_validation_failed_report_with_reviewed_sha_fails(self): + report = ( + "- Selected PR: 281\n" + "- Reviewed head SHA: abc1234\n" + "- Validation: full suite failed\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertFalse(result["complete"]) + + def test_validation_failed_report_with_reviewed_none_passes(self): + report = ( + "- Selected PR: 281\n" + "- Reviewed head SHA: none\n" + "- Validation: full suite failed\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertTrue(result["complete"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_issue_claim_heartbeat.py b/tests/test_issue_claim_heartbeat.py new file mode 100644 index 0000000..c8a154e --- /dev/null +++ b/tests/test_issue_claim_heartbeat.py @@ -0,0 +1,116 @@ +"""Tests for issue claim heartbeat leases (#268).""" + +from __future__ import annotations + +import sys +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import issue_claim_heartbeat as ich # noqa: E402 + + +class TestHeartbeatFormatting(unittest.TestCase): + def test_format_and_parse_roundtrip(self): + body = ich.format_heartbeat_body( + kind="claim", + issue_number=268, + branch="feat/issue-268-heartbeat-leases", + phase="claimed", + profile="prgs-author", + next_action="implement module", + ) + parsed = ich.parse_heartbeat_comment(body) + self.assertIsNotNone(parsed) + assert parsed is not None + self.assertEqual(parsed["issue_number"], 268) + self.assertEqual(parsed["branch"], "feat/issue-268-heartbeat-leases") + self.assertEqual(parsed["phase"], "claimed") + + +class TestClaimClassification(unittest.TestCase): + NOW = datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc) + + def _comment(self, *, minutes_ago: int, kind: str = "progress") -> dict: + ts = (self.NOW - timedelta(minutes=minutes_ago)).isoformat() + body = ich.format_heartbeat_body( + kind=kind, + issue_number=268, + branch="feat/issue-268-heartbeat-leases", + phase="implementation", + profile="prgs-author", + ) + return {"id": 1, "body": body, "created_at": ts, "updated_at": ts} + + def test_active_claim_within_lease(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=5)], + now=self.NOW, + ) + self.assertEqual(result["status"], "active") + + def test_stale_claim_without_branch(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=45)], + now=self.NOW, + ) + self.assertEqual(result["status"], "stale") + + def test_reclaimable_after_sixty_minutes_without_branch(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=90)], + now=self.NOW, + ) + self.assertEqual(result["status"], "reclaimable") + + def test_awaiting_review_when_open_pr_exists(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + open_prs = [ + { + "number": 400, + "title": "feat", + "body": "Closes #268", + "head": {"ref": "feat/issue-268-heartbeat-leases"}, + } + ] + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=120)], + open_prs=open_prs, + now=self.NOW, + ) + self.assertEqual(result["status"], "awaiting_review") + + def test_phantom_claim_without_heartbeat(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[{"id": 1, "body": "working on it"}], + now=self.NOW, + ) + self.assertEqual(result["status"], "phantom") + + +class TestInventory(unittest.TestCase): + def test_build_cleanup_plan_flags_phantom(self): + inventory = { + "entries": [ + {"issue_number": 1, "status": "phantom", "reasons": ["no heartbeat"]}, + {"issue_number": 2, "status": "active", "reasons": []}, + ] + } + plan = ich.build_cleanup_plan(inventory) + self.assertEqual(len(plan), 1) + self.assertEqual(plan[0]["issue_number"], 1) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_issue_write_tool_gates.py b/tests/test_issue_write_tool_gates.py index afe865d..736b295 100644 --- a/tests/test_issue_write_tool_gates.py +++ b/tests/test_issue_write_tool_gates.py @@ -172,6 +172,8 @@ class TestAllowedIssueWritesProceed(IssueWriteGateBase): return [{"id": 10, "name": "status:in-progress"}] if method == "POST" and "/issues/9/labels" in url: return [{"name": "status:in-progress"}] + if method == "POST" and "/issues/9/comments" in url: + return {"id": 501} if method == "GET" and url.endswith("/user"): return {"login": "author-user"} return {} diff --git a/tests/test_linked_issue_consistency.py b/tests/test_linked_issue_consistency.py new file mode 100644 index 0000000..cf325bd --- /dev/null +++ b/tests/test_linked_issue_consistency.py @@ -0,0 +1,118 @@ +"""Tests for linked-issue consistency verifier (#314). + +Reviewer reports must verify the linked issue live for the selected PR; +stale issue numbers from a previous PR must not leak into the final +report, and linked-issue status must never be claimed without live proof. +""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import assess_linked_issue_consistency # noqa: E402 + + +class TestCorrectLinkedIssue(unittest.TestCase): + def test_matching_linked_issue_passes(self): + report = ( + "Selected PR: 280\n" + "Linked issue status: Issue #275 open, closes on merge\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertTrue(result["complete"]) + self.assertFalse(result["downgraded"]) + self.assertEqual(result["reasons"], []) + + def test_multiple_linked_issues_all_reported_passes(self): + report = ( + "Linked issue status: Issue #275 and Issue #276 both open\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275, 276] + ) + self.assertTrue(result["complete"]) + + def test_multiple_linked_issues_one_missing_fails(self): + report = "Linked issue status: Issue #275 open\n" + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275, 276] + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("276" in r for r in result["reasons"]) + ) + + +class TestStaleIssueLeak(unittest.TestCase): + def test_stale_issue_in_status_field_fails(self): + # #314 observed behavior: PR #280 links Issue #275 but the report + # carries Issue #260 from the previous PR. + report = "Linked issue status: Issue #260 open\n" + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertFalse(result["complete"]) + self.assertTrue(any("260" in r for r in result["reasons"])) + + def test_stale_issue_in_diagnostics_fails(self): + report = ( + "Linked issue status: Issue #275 open\n" + "Read-only diagnostics: viewed Issue #260 status\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("stale" in r.lower() and "260" in r for r in result["reasons"]) + ) + + def test_selected_pr_number_mention_is_not_stale(self): + # "PR #280" mentions must not be confused with issue mentions. + report = ( + "Selected PR: 280\n" + "Validation: ran tests at PR #280 head\n" + "Linked issue status: Issue #275 open\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertTrue(result["complete"]) + + +class TestMissingProof(unittest.TestCase): + def test_status_claim_without_live_proof_fails(self): + report = "Linked issue status: Issue #275 open\n" + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=None + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("live proof" in r.lower() for r in result["reasons"]) + ) + + def test_unverified_wording_without_proof_passes(self): + report = ( + "Linked issue status: not verified in this session\n" + ) + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=None + ) + self.assertTrue(result["complete"]) + + def test_missing_field_fails(self): + report = "Selected PR: 280\n" + result = assess_linked_issue_consistency( + report, selected_pr=280, linked_issues=[275] + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("linked issue status" in r.lower() for r in result["reasons"]) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 8f208c4..609906a 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -136,6 +136,18 @@ def test_create_issue_final_report_verifier_exported(): assert callable(assess_create_issue_final_report) +def test_merge_simulation_verifier_exported(): + from review_proofs import assess_merge_simulation_report + + assert callable(assess_merge_simulation_report) + + +def test_validation_integrity_verifier_exported(): + from review_proofs import assess_validation_integrity_report + + assert callable(assess_validation_integrity_report) + + def test_prior_blocker_skip_verifier_exported(): from review_proofs import assess_prior_blocker_skip_proof @@ -146,3 +158,21 @@ def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof assert callable(assess_non_mergeable_skip_proof) + + +def test_validation_worktree_edit_verifier_exported(): + from review_proofs import assess_validation_worktree_edit_report + + assert callable(assess_validation_worktree_edit_report) + + +def test_mutation_categories_verifier_exported(): + from review_proofs import assess_mutation_categories_report + + assert callable(assess_mutation_categories_report) + + +def test_worktree_ownership_verifier_exported(): + from review_proofs import assess_worktree_ownership_report + + assert callable(assess_worktree_ownership_report) \ No newline at end of file diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 394ab03..99d7d73 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -321,15 +321,17 @@ class TestMarkIssue(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_start_adds_label(self, _auth, mock_api): - # First call: get labels; second call: add label + # labels, add label, post claim heartbeat comment mock_api.side_effect = [ [{"id": 10, "name": "status:in-progress"}], [{"name": "status:in-progress"}], + {"id": 99}, ] with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): result = gitea_mark_issue(issue_number=5, action="start") self.assertTrue(result["success"]) self.assertIn("claimed", result["message"]) + self.assertTrue(result.get("heartbeat_posted")) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 3a3ea7a..0d42359 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -23,7 +23,11 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, + assess_already_landed_report_state, + assess_already_landed_review_gate, assess_author_pr_report, + assess_validation_environment_proof, + assess_full_suite_failure_approval_gate, assess_queue_ordering_proof, assess_reviewer_baseline_validation_proof, assess_capability_evidence, @@ -2314,10 +2318,108 @@ class TestCreateIssueFinalReport(unittest.TestCase): self.assertTrue(result["downgraded"]) -class TestQueueOrderingProof(unittest.TestCase): - """Issue #321: reviewer queue selection must prove ordering.""" +class TestValidationEnvironmentProof(unittest.TestCase): + """Issue #311: exact validation command and execution environment.""" + + ROOT = "/repo/Gitea-Tools/branches/review-pr-280" + + def test_venv_pytest_with_full_report_passes(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: venv/bin/pytest tests/ --ignore=branches\n" + f"Working directory: {self.ROOT}\n" + "Result: 1014 passed, 6 skipped" + ), + ) + self.assertTrue(result["proven"]) + + def test_bare_pytest_without_path_proof_fails(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: pytest\n" + f"Working directory: {self.ROOT}\n" + "1014 passed, 6 skipped" + ), + ) + self.assertFalse(result["proven"]) + + def test_bare_pytest_with_path_and_version_proof_passes(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: pytest tests/\n" + f"Working directory: {self.ROOT}\n" + "which pytest: /repo/Gitea-Tools/venv/bin/pytest\n" + "pytest --version: pytest 8.3.4\n" + "Resolves to the project venv: yes\n" + "1014 passed, 6 skipped" + ), + ) + self.assertTrue(result["proven"]) + + def test_vague_pytest_summary_without_command_fails(self): + result = assess_validation_environment_proof( + report_text=( + f"Working directory: {self.ROOT}\n" + "pytest executed inside the worktree: 1014 passed, 6 skipped" + ), + ) + self.assertFalse(result["proven"]) + + def test_structured_environment_proof_passes(self): + result = assess_validation_environment_proof( + validation_environment={ + "command": "pytest tests/", + "working_directory": self.ROOT, + "which_pytest": "/repo/Gitea-Tools/venv/bin/pytest", + "pytest_version": "pytest 8.3.4", + "venv_resolved": True, + "passed": 1014, + "skipped": 6, + }, + ) + self.assertTrue(result["proven"]) + + def test_missing_working_directory_fails(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: venv/bin/pytest tests/\n" + "1014 passed, 6 skipped" + ), + ) + self.assertFalse(result["proven"]) + + def test_build_final_report_downgrades_missing_environment_proof(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=( + f"Working directory: {self.ROOT}\n" + "pytest executed: 1014 passed, 6 skipped" + ), + ) + self.assertNotEqual(report["grade"], "A") + self.assertFalse(report["validation_environment_proven"]) + - def test_next_eligible_claim_without_policy_fails(self): result = assess_queue_ordering_proof( report_text="Selected the next eligible PR: PR #286.", ) @@ -2539,5 +2641,306 @@ class TestReviewerBaselineValidationProof(unittest.TestCase): self.assertFalse(report["baseline_validation_proven"]) +class TestAlreadyLandedReviewGate(unittest.TestCase): + """Issue #292: PR head already on target blocks approval and merge.""" + + def _gate(self, **overrides): + kwargs = { + "pr_number": 278, + "candidate_head_sha": PINNED, + "target_branch": "master", + "target_branch_sha": OTHER, + "head_is_ancestor_of_target": True, + "live_head_sha": PINNED, + } + kwargs.update(overrides) + return assess_already_landed_review_gate(**kwargs) + + def test_already_landed_blocks_approval_and_merge(self): + result = self._gate() + self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED") + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["merge_api_allowed"]) + self.assertFalse(result["request_changes_allowed"]) + self.assertTrue(result["reconciliation_handoff_required"]) + + def test_already_landed_with_real_blocker_allows_request_changes(self): + result = self._gate(real_blocker=True) + self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED") + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["merge_api_allowed"]) + self.assertTrue(result["request_changes_allowed"]) + + def test_normal_candidate_passes_gate(self): + result = self._gate(head_is_ancestor_of_target=False) + self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE") + self.assertTrue(result["approve_allowed"]) + self.assertTrue(result["merge_api_allowed"]) + self.assertFalse(result["reconciliation_handoff_required"]) + + def test_non_mergeable_normal_pr_still_passes_this_gate(self): + # Mergeability/conflicts are separate gates; ancestry gate only + # decides already-landed vs normal candidate. + result = self._gate( + head_is_ancestor_of_target=False, mergeable=False) + self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE") + self.assertTrue(result["request_changes_allowed"]) + + def test_unchecked_ancestry_blocks_review_mutations(self): + result = self._gate(head_is_ancestor_of_target=None) + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["merge_api_allowed"]) + + def test_changed_head_since_pin_blocks_until_recheck(self): + result = self._gate( + head_is_ancestor_of_target=False, live_head_sha=OTHER) + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + self.assertFalse(result["approve_allowed"]) + self.assertTrue(any( + "head" in reason.lower() for reason in result["reasons"] + )) + + def test_invalid_shas_block_gate(self): + result = self._gate(candidate_head_sha="deadbeef") + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + self.assertFalse(result["approve_allowed"]) + + result = self._gate(target_branch_sha=None) + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + + def test_already_landed_handoff_lists_required_fields(self): + result = self._gate() + for field in ( + "PR number/title", + "candidate head SHA", + "target branch", + "target branch SHA", + "ancestor proof", + "linked issue status", + "recommended reconciliation action", + "capability proof for close/comment mutations", + ): + self.assertIn(field, result["required_handoff_fields"]) + + +class TestAlreadyLandedReportState(unittest.TestCase): + """Issue #292: reports cannot say APPROVED after the gate fired.""" + + GOOD_REPORT = ( + "Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n" + "Candidate head SHA: " + PINNED + "\n" + "Review decision: none\n" + "Merge result: none\n" + ) + + def test_gate_fired_with_reconcile_state_passes(self): + result = assess_already_landed_report_state( + self.GOOD_REPORT, gate_fired=True) + self.assertTrue(result["complete"]) + self.assertFalse(result["block"]) + + def test_gate_fired_with_approved_state_blocks(self): + report = self.GOOD_REPORT.replace( + "Review decision: none", "Review decision: APPROVED") + result = assess_already_landed_report_state(report, gate_fired=True) + self.assertTrue(result["block"]) + + def test_gate_fired_with_merged_state_blocks(self): + report = self.GOOD_REPORT.replace( + "Merge result: none", "Merge result: MERGED") + result = assess_already_landed_report_state(report, gate_fired=True) + self.assertTrue(result["block"]) + + def test_gate_fired_with_ready_to_merge_blocks(self): + result = assess_already_landed_report_state( + self.GOOD_REPORT + "Current status: READY_TO_MERGE\n", + gate_fired=True) + self.assertTrue(result["block"]) + + def test_gate_fired_without_reconcile_state_blocks(self): + result = assess_already_landed_report_state( + "Current status: review complete.\n", gate_fired=True) + self.assertTrue(result["block"]) + self.assertTrue(any( + "already_landed_reconcile_required" in reason.lower() + for reason in result["reasons"] + )) + + def test_gate_not_fired_reports_pass_untouched(self): + result = assess_already_landed_report_state( + "Review decision: APPROVED\nMerge result: MERGED\n", + gate_fired=False) + self.assertTrue(result["complete"]) + + +class TestFullSuiteFailureApprovalGate(unittest.TestCase): + """Issue #323: full-suite failure blocks approval without baseline proof.""" + + ROOT = "/repo/Gitea-Tools" + + def _good_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, + "pr_suite_command": "venv/bin/pytest tests/", + "baseline_suite_command": "venv/bin/pytest tests/", + "new_tests_passed": True, + "unrelated_explanation": ( + "failures touch agent_temp_artifacts only; PR changes " + "review_proofs handoff fields" + ), + } + proof.update(overrides) + return proof + + def _good_report(self): + return ( + "Full suite failed on PR worktree and on baseline.\n" + "- Validation command: venv/bin/pytest tests/\n" + f"- Working directory: {self.ROOT}/branches/review-pr-280\n" + "Failures are pre-existing on master.\n" + ) + + def test_full_suite_pass_allows_approval_without_baseline(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=True, + project_root=self.ROOT, + ) + self.assertTrue(result["approve_allowed"]) + self.assertFalse(result["block"]) + self.assertEqual(result["default_action"], "proceed") + + def test_full_suite_failure_defaults_to_request_changes(self): + result = assess_full_suite_failure_approval_gate( + review_decision="request_changes", + full_suite_passed=False, + project_root=self.ROOT, + ) + self.assertFalse(result["block"]) + self.assertEqual(result["default_action"], "request_changes") + + def test_approval_with_failure_and_no_proof_blocks(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + self.assertEqual(result["default_action"], "request_changes") + + def test_approval_with_matching_baseline_proof_allowed(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof(), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertTrue(result["approve_allowed"]) + self.assertFalse(result["block"]) + + def test_mismatched_failure_signatures_block_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof( + failure_signatures_match=False, + pr_failures=["tests/test_other.py::test_other"], + ), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + + def test_main_checkout_baseline_blocks_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof(worktree_path=self.ROOT), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + + def test_missing_new_tests_proof_blocks_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof(new_tests_passed=None), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(any( + "new" in reason.lower() and "test" in reason.lower() + for reason in result["reasons"] + )) + + def test_missing_unrelated_explanation_blocks_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof(unrelated_explanation=""), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + + def test_missing_suite_commands_block_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + baseline_proof=self._good_proof( + pr_suite_command="", baseline_suite_command=""), + report_text=self._good_report(), + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(any( + "command" in reason.lower() for reason in result["reasons"] + )) + + def test_unknown_suite_result_blocks_approval(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=None, + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + + def test_vague_same_as_master_claim_without_proof_blocks(self): + result = assess_full_suite_failure_approval_gate( + review_decision="approve", + full_suite_passed=False, + report_text="Validation: failures same as master; approving.", + project_root=self.ROOT, + ) + self.assertFalse(result["approve_allowed"]) + self.assertTrue(result["block"]) + + def test_non_approve_decision_never_blocks(self): + result = assess_full_suite_failure_approval_gate( + review_decision="comment", + full_suite_passed=None, + project_root=self.ROOT, + ) + self.assertFalse(result["block"]) + + +if __name__ == "__main__": + unittest.main() + if __name__ == "__main__": unittest.main() diff --git a/tests/test_reviewer_merge_simulation.py b/tests/test_reviewer_merge_simulation.py new file mode 100644 index 0000000..e3101f8 --- /dev/null +++ b/tests/test_reviewer_merge_simulation.py @@ -0,0 +1,135 @@ +"""Tests for merge-simulation worktree mutation verifier (#317).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_merge_simulation import ( # noqa: E402 + assess_merge_simulation_report, + merge_simulation_commands, +) + + +def _good_simulation_report(**overrides) -> str: + lines = [ + "Worktree/index mutations: merge simulation in branches/review-pr281", + "Worktree path: branches/review-pr281", + "Pre-simulation clean status: clean (git status --porcelain empty)", + "Merge result: conflicts in review_proofs.py", + "Abort command: git merge --abort", + "Post-abort clean status: clean after abort (git status --porcelain empty)", + ] + text = "\n".join(lines) + for key, value in overrides.items(): + text = text.replace(f"{key}:", f"{key}: {value}") + return text + + +class TestMergeSimulationDetection(unittest.TestCase): + def test_detects_no_commit_merge(self): + commands = merge_simulation_commands([ + "git merge prgs/master --no-commit --no-ff", + "git status", + ]) + self.assertEqual(commands, ["git merge prgs/master --no-commit --no-ff"]) + + def test_detects_merge_abort(self): + commands = merge_simulation_commands([ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ]) + self.assertIn("git merge --abort", commands) + + +class TestMergeSimulationReport(unittest.TestCase): + def test_no_simulation_passes(self): + result = assess_merge_simulation_report( + "Review complete. Worktree/index mutations: none", + command_log=["git status --porcelain", "git diff prgs/master...HEAD"], + ) + self.assertTrue(result["proven"]) + + def test_documented_simulation_passes(self): + report = _good_simulation_report() + result = assess_merge_simulation_report( + report, + command_log=[ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_readonly_classification_blocks(self): + report = "\n".join([ + "Read-only diagnostics: git merge prgs/master --no-commit --no-ff, git status", + "Worktree/index mutations: none", + ]) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("read-only" in r.lower() for r in result["reasons"])) + + def test_missing_worktree_field_blocks(self): + report = "Merge simulation ran but only noted in prose." + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("worktree/index mutations" in r.lower() for r in result["reasons"])) + + def test_conflict_simulation_requires_merge_result(self): + report = _good_simulation_report().replace( + "Merge result: conflicts in review_proofs.py", + "", + ) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("merge result" in r.lower() for r in result["reasons"])) + + def test_aborted_simulation_requires_post_clean_proof(self): + report = _good_simulation_report().replace( + "Post-abort clean status: clean after abort (git status --porcelain empty)", + "", + ) + result = assess_merge_simulation_report( + report, + command_log=[ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("post-abort" in r.lower() for r in result["reasons"])) + + def test_successful_simulation_without_abort_omits_abort_fields(self): + report = "\n".join([ + "Worktree/index mutations: merge simulation completed cleanly", + "Worktree path: branches/review-pr281", + "Pre-simulation clean status: clean before simulation", + "Merge result: clean fast-forward simulation", + ]) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_merge_simulation_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reviewer_mutation_categories.py b/tests/test_reviewer_mutation_categories.py new file mode 100644 index 0000000..055b174 --- /dev/null +++ b/tests/test_reviewer_mutation_categories.py @@ -0,0 +1,88 @@ +"""Tests for precise mutation category verifier (#319).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_mutation_categories import assess_mutation_categories_report # noqa: E402 + + +def _precise_handoff() -> str: + return "\n".join([ + "Controller Handoff", + "File edits by reviewer: none", + "Worktree/index mutations: git worktree add branches/review-pr1", + "Git ref mutations: git fetch prgs master", + "MCP/Gitea mutations: gitea_view_pr", + "Review mutations: gitea_review_pr request_changes", + "Read-only diagnostics: git status, git diff", + ]) + + +class TestMutationCategories(unittest.TestCase): + def test_precise_categories_pass(self): + result = assess_mutation_categories_report(_precise_handoff()) + self.assertTrue(result["proven"], result["reasons"]) + + def test_legacy_workspace_mutations_blocks(self): + report = "\n".join([ + "Controller Handoff", + "Workspace mutations: none (no local file changes)", + "File edits by reviewer: none", + "Worktree/index mutations: none", + "Git ref mutations: none", + ]) + result = assess_mutation_categories_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"])) + + def test_workspace_none_with_worktree_command_blocks(self): + report = "\n".join([ + "Controller Handoff", + "Workspace mutations: none (no local file changes)", + "File edits by reviewer: none", + "Worktree/index mutations: none", + "Git ref mutations: none", + ]) + result = assess_mutation_categories_report( + report, + observed_commands=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + + def test_file_edits_none_with_worktree_mutation_passes(self): + report = "\n".join([ + "Controller Handoff", + "File edits by reviewer: none", + "Worktree/index mutations: git reset --hard FETCH_HEAD", + "Git ref mutations: git fetch prgs", + ]) + result = assess_mutation_categories_report( + report, + observed_commands=["git reset --hard FETCH_HEAD", "git fetch prgs"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_missing_required_categories_blocks(self): + report = "Controller Handoff\nFile edits by reviewer: none\n" + result = assess_mutation_categories_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("missing" in r.lower() for r in result["reasons"])) + + def test_non_controller_handoff_skips(self): + result = assess_mutation_categories_report( + "Workspace mutations: none\n", + ) + self.assertTrue(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_mutation_categories_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reviewer_validation_integrity.py b/tests/test_reviewer_validation_integrity.py new file mode 100644 index 0000000..5137057 --- /dev/null +++ b/tests/test_reviewer_validation_integrity.py @@ -0,0 +1,130 @@ +"""Tests for PR-head vs diagnostic validation integrity verifier (#316).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_validation_integrity import assess_validation_integrity_report # noqa: E402 + + +def _official_only_report() -> str: + return "\n".join([ + "Official PR-head validation on unmodified worktree.", + "Official validation command: pytest tests/ -q", + "Official validation result: failed (3 failed)", + "Validation integrity status: failed", + "Worktree dirty after official validation: clean", + "Review decision: request_changes", + ]) + + +def _diagnostic_report() -> str: + return "\n".join([ + _official_only_report(), + "Diagnostic local experiment — not PR-head validation", + "Diagnostic edit path: tests/test_example.py", + "File edits by reviewer: tests/test_example.py", + "Diagnostic command: pytest tests/test_example.py -q", + "Diagnostic result: passed after local fix", + "Diagnostic results used only for suggested fix; blocker remains PR-head failure.", + ]) + + +class TestValidationIntegrity(unittest.TestCase): + def test_official_only_passes(self): + result = assess_validation_integrity_report( + _official_only_report(), + validation_session={ + "official_validation_ran": True, + "official_result": "fail", + "worktree_dirty_after_official": False, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_failing_pr_head_with_diagnostic_pass_passes(self): + result = assess_validation_integrity_report( + _diagnostic_report(), + validation_session={ + "official_validation_ran": True, + "official_result": "fail", + "diagnostic_validation_ran": True, + "diagnostic_result": "pass", + "diagnostic_edits": ["tests/test_example.py"], + "worktree_dirty_after_official": False, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_post_edit_tests_as_official_blocks(self): + report = "\n".join([ + "Official validation result: passed after diagnostic edit", + "pytest passed after local fix in validation worktree", + ]) + result = assess_validation_integrity_report( + report, + validation_session={ + "official_validation_ran": True, + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_validation_ran": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_missing_dirty_after_validation_blocks(self): + report = _official_only_report().replace( + "Worktree dirty after official validation: clean", + "", + ) + result = assess_validation_integrity_report( + report, + validation_session={"official_validation_ran": True, "official_result": "fail"}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("dirty" in r.lower() for r in result["reasons"])) + + def test_diagnostic_without_label_blocks(self): + report = _diagnostic_report().replace( + "Diagnostic local experiment — not PR-head validation", + "", + ) + result = assess_validation_integrity_report( + report, + validation_session={ + "official_validation_ran": True, + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_validation_ran": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"])) + + def test_contaminated_requires_integrity_status(self): + report = "Validation worktree was edited before official validation completed." + result = assess_validation_integrity_report( + report, + validation_session={"contaminated": True, "official_validation_ran": True}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("contaminated" in r.lower() for r in result["reasons"])) + + def test_action_log_edits_trigger_diagnostic_rules(self): + result = assess_validation_integrity_report( + _official_only_report(), + validation_session={"official_validation_ran": True, "official_result": "fail"}, + action_log=[{"path": "tests/test_example.py", "kind": "file_edit", "performed": True}], + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_validation_integrity_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_reviewer_validation_worktree_edits.py b/tests/test_reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..adedaca --- /dev/null +++ b/tests/test_reviewer_validation_worktree_edits.py @@ -0,0 +1,134 @@ +"""Tests for PR validation worktree no-edit verifier (#315).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_validation_worktree_edits import ( # noqa: E402 + assess_validation_worktree_edit_report, +) + + +def _read_only_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "File edits by reviewer: none", + "Review decision: request_changes", + ]) + + +def _diagnostic_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "Review decision: request_changes", + "Diagnostic local experiment — not PR-head validation", + "Diagnostic scratch worktree path: branches/diagnostic-pr280-fix-test", + "File edits by reviewer: tests/test_agent_temp_artifacts.py (diagnostic only)", + "Diagnostic result: passed after local fix", + ]) + + +class TestValidationWorktreeEdits(unittest.TestCase): + def test_read_only_review_passes(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_validation_worktree_edit_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("validation worktree" in r.lower() for r in result["reasons"])) + + def test_file_edits_none_contradiction_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + action_log=[ + { + "path": "tests/test_agent_temp_artifacts.py", + "kind": "file_edit", + "performed": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + } + ], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("file edits" in r.lower() for r in result["reasons"])) + + def test_diagnostic_scratch_with_reporting_passes(self): + result = assess_validation_worktree_edit_report( + _diagnostic_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "diagnostic_worktree_path": "branches/diagnostic-pr280-fix-test", + "diagnostic_edits": ["tests/test_agent_temp_artifacts.py"], + "diagnostic_experiment": True, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_diagnostic_without_scratch_worktree_blocks(self): + result = assess_validation_worktree_edit_report( + "File edits by reviewer: tests/test_example.py", + validation_session={ + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_experiment": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"])) + + def test_post_edit_official_validation_blocks(self): + report = "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official validation result: passed after local edit", + "File edits by reviewer: tests/test_agent_temp_artifacts.py", + ]) + result = assess_validation_worktree_edit_report( + report, + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + "official_validation_after_edit": True, + }, + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_validation_worktree_edit_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reviewer_worktree_ownership.py b/tests/test_reviewer_worktree_ownership.py new file mode 100644 index 0000000..3b9dad3 --- /dev/null +++ b/tests/test_reviewer_worktree_ownership.py @@ -0,0 +1,137 @@ +"""Tests for reviewer worktree ownership verifier (#312).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_worktree_ownership import assess_worktree_ownership_report # noqa: E402 + + +def _fresh_review_report() -> str: + return "\n".join([ + "Review worktree path: branches/review-pr280-feat-issue-275-claim", + "Worktree is inside branches/ and not the main checkout.", + "Tracked state: clean; untracked state: clean.", + "Official PR-head validation on session-owned worktree.", + "Worktree mutations: none", + ]) + + +def _safe_reuse_report() -> str: + return "\n".join([ + "Safe-reuse proof for existing review worktree.", + "Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks", + "Worktree inside branches/; not the main checkout.", + "Not owned by another active task/session.", + "Clean tracked state; clean untracked state.", + "Branch/head before reset: feat/issue-260-example", + "Reset target SHA: abc123def4567890abcdef1234567890abcdef12", + "Project policy allows reuse/reset for this clean review worktree.", + "Worktree mutations: git reset --hard FETCH_HEAD", + "Destructive reset operations: git reset --hard FETCH_HEAD", + ]) + + +class TestWorktreeOwnership(unittest.TestCase): + def test_fresh_session_owned_passes(self): + result = assess_worktree_ownership_report( + _fresh_review_report(), + ownership_session={ + "validation_ran": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + self.assertTrue(result["session_owned"]) + + def test_safe_reuse_with_reset_passes(self): + result = assess_worktree_ownership_report( + _safe_reuse_report(), + ownership_session={ + "safe_reuse": True, + "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_destructive_reset_without_proof_blocks(self): + report = "\n".join([ + "Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks", + "Worktree mutations: git reset --hard FETCH_HEAD", + ]) + result = assess_worktree_ownership_report( + report, + ownership_session={ + "validation_ran": True, + "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_reused_dirty_tracked_blocks(self): + result = assess_worktree_ownership_report( + _fresh_review_report(), + ownership_session={ + "validation_ran": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + "dirty_tracked": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("tracked" in r.lower() for r in result["reasons"])) + + def test_safe_reuse_dirty_untracked_blocks(self): + result = assess_worktree_ownership_report( + _safe_reuse_report(), + ownership_session={ + "safe_reuse": True, + "dirty_untracked": True, + "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("untracked" in r.lower() for r in result["reasons"])) + + def test_workspace_none_with_reset_blocks(self): + report = "\n".join([ + "Review worktree path: branches/review-pr280-feat-issue-275-claim", + "Workspace mutations: none", + "Worktree mutations: git reset --hard FETCH_HEAD", + ]) + result = assess_worktree_ownership_report( + report, + ownership_session={ + "session_owned": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"])) + + def test_main_checkout_blocks(self): + result = assess_worktree_ownership_report( + "Review worktree path: /Users/dev/Gitea-Tools", + ownership_session={ + "validation_ran": True, + "main_checkout": True, + "worktree_path": "/Users/dev/Gitea-Tools", + }, + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_worktree_ownership_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file