diff --git a/review_proofs.py b/review_proofs.py index ec36656..b723246 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) @@ -3581,6 +3652,113 @@ def assess_reviewer_baseline_validation_proof( } +# --------------------------------------------------------------------------- +# 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) # --------------------------------------------------------------------------- @@ -3662,6 +3840,13 @@ def assess_email_disclosure( } +def assess_reviewer_fallback_report(report_text, **kwargs): + """#324: block local Gitea fallbacks during normal reviewer workflows.""" + from reviewer_fallback import assess_reviewer_fallback_report as _assess + + return _assess(report_text, **kwargs) + + def assess_final_report_validator(report_text, task_kind, **kwargs): """#327: single entry point for task-specific final-report validation.""" from final_report_validator import assess_final_report_validator as _validate @@ -4059,8 +4244,22 @@ def assess_validation_integrity_report(report_text, **kwargs): 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 + + return _assess(report_text, **kwargs) + + def assess_non_mergeable_skip_proof(report_text, **kwargs): """#322: require conflict proof when skipping non-mergeable PRs.""" from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess return _assess(report_text, **kwargs) + + +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_blocker_skip.py b/reviewer_blocker_skip.py new file mode 100644 index 0000000..a95d368 --- /dev/null +++ b/reviewer_blocker_skip.py @@ -0,0 +1,164 @@ +"""Prior-blocker skip proof verifier for reviewer queue reports (#318).""" + +from __future__ import annotations + +import re +from typing import Any + +_FULL_SHA = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE) +_SHORT_SHA = re.compile(r"\b[0-9a-f]{7,39}\b", re.IGNORECASE) +_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE) +_BLOCKING_DECISION_RE = re.compile( + r"(?:blocking review decision|review decision|blocking decision)\s*:\s*request[_ ]changes|" + r"request[_ ]changes", + re.IGNORECASE, +) +_BLOCKING_HEAD_RE = re.compile( + r"(?:blocking review head(?:\s+sha)?|blocker head(?:\s+sha)?|review head at blocker)", + re.IGNORECASE, +) +_HEAD_CHANGED_RE = re.compile( + r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|" + r"head changed (?:after|since) (?:the )?blocker|head unchanged since blocker)", + re.IGNORECASE, +) +_BLOCKER_REASON_RE = re.compile( + r"(?:reason (?:it )?remains blocked|remains blocked because|blocking category)", + re.IGNORECASE, +) +_LIVE_BLOCKER_PROOF_RE = re.compile( + r"(?:blocker revalidated live|live proof|gitea_get_pr_review_feedback|" + r"gitea_view_pr|review feedback fetched|current review state)", + re.IGNORECASE, +) +_BLOCKER_UNVERIFIED_RE = re.compile(r"\bBLOCKER_STATUS_UNVERIFIED\b") + + +def _pr_section(text: str, pr_number: int) -> str: + lines = (text or "").splitlines() + chunks: list[str] = [] + capture = False + token = f"#{pr_number}" + for line in lines: + lower = line.lower() + if token in lower or f"pr {pr_number}" in lower or f"pr#{pr_number}" in lower.replace(" ", ""): + capture = True + chunks.append(line) + continue + if capture: + if _PR_NUMBER_RE.search(line) and token not in line.lower(): + break + if line.strip() == "" and len(chunks) > 3: + break + chunks.append(line) + if chunks: + return "\n".join(chunks) + return text or "" + + +def _has_head_sha(text: str, head_sha: str | None) -> bool: + if not head_sha: + return bool(_FULL_SHA.search(text) or _SHORT_SHA.search(text)) + head = head_sha.strip().lower() + if head in text.lower(): + return True + if len(head) >= 7 and head[:7] in text.lower(): + return True + return bool(_FULL_SHA.search(text)) + + +def assess_prior_blocker_skip_proof( + report_text: str, + *, + skipped_prs: list[dict] | None = None, +) -> dict[str, Any]: + """Validate live blocker proof for skipped earlier open PRs (#318).""" + text = report_text or "" + reasons: list[str] = [] + assessments: list[dict[str, Any]] = [] + + for entry in skipped_prs or []: + pr_number = entry.get("pr_number") + if pr_number is None: + reasons.append("skipped PR entry missing pr_number") + continue + pr_number = int(pr_number) + section = _pr_section(text, pr_number) + verified = entry.get("blocker_verified") + head_changed = entry.get("head_changed_since_blocker") + blocking_head = (entry.get("blocking_review_head_sha") or "").strip() or None + current_head = (entry.get("head_sha") or "").strip() or None + skip_reason = (entry.get("skip_reason") or entry.get("blocking_decision") or "").lower() + + item_reasons: list[str] = [] + + if head_changed is True: + item_reasons.append( + f"PR #{pr_number} head changed after blocker; it cannot be skipped " + "based on stale REQUEST_CHANGES" + ) + + if f"#{pr_number}" not in text.lower() and f"pr {pr_number}" not in text.lower(): + item_reasons.append(f"skipped PR #{pr_number} not documented in final report") + + if "request_changes" in skip_reason or entry.get("blocking_decision") == "request_changes": + if verified is False: + if not _BLOCKER_UNVERIFIED_RE.search(section): + item_reasons.append( + f"PR #{pr_number} blocker proof unavailable; report must classify " + "BLOCKER_STATUS_UNVERIFIED" + ) + else: + if not _BLOCKING_DECISION_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing blocking review decision proof" + ) + if not _has_head_sha(section, current_head): + item_reasons.append( + f"PR #{pr_number} skip missing current head SHA" + ) + if blocking_head and not _BLOCKING_HEAD_RE.search(section): + if blocking_head.lower() not in section.lower(): + item_reasons.append( + f"PR #{pr_number} skip missing blocking review head SHA" + ) + if head_changed is False and not _HEAD_CHANGED_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing head-changed-since-blocker proof" + ) + if not _BLOCKER_REASON_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing reason-it-remains-blocked" + ) + if not _LIVE_BLOCKER_PROOF_RE.search(section): + item_reasons.append( + f"PR #{pr_number} skip missing live blocker proof for this session" + ) + + proven = not item_reasons + assessments.append({ + "pr_number": pr_number, + "proven": proven, + "classification": ( + "BLOCKER_STATUS_UNVERIFIED" + if verified is False + else "BLOCKED_SKIPPED" + ), + "reasons": item_reasons, + }) + reasons.extend(item_reasons) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "downgraded": False, + "assessments": assessments, + "reasons": reasons, + "safe_next_action": ( + "fetch live review feedback and document blocker proof for each skipped PR, " + "or classify BLOCKER_STATUS_UNVERIFIED" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/reviewer_fallback.py b/reviewer_fallback.py new file mode 100644 index 0000000..9b8ef6c --- /dev/null +++ b/reviewer_fallback.py @@ -0,0 +1,213 @@ +"""Reviewer local-fallback detection for normal PR review workflows (#324). + +Normal reviewer runs must use MCP tools. Reading profile secret files or +running local Gitea helper scripts while MCP is available is a fail-closed +violation. Explicit recovery mode may use local fallback only with full proof. +""" + +from __future__ import annotations + +import re +from typing import Any + +LOCAL_GITEA_SCRIPT_NAMES = ( + "list_prs.py", + "view_pr.py", + "create_pr.py", + "merge_pr.py", + "edit_pr.py", + "list_issues.py", + "delete_branch.py", + "create_issue.py", + "close_issue.py", + "review_pr.py", + "mark_issue.py", + "mirror_refs.sh", +) + +PROFILE_SECRET_MARKERS = ( + "profiles.json", + ".config/gitea-tools/profiles", + "gitea_auth.py", + "gitea_config.py", + "keychain", + "credential fill", + "token store", +) + +_RECOVERY_MODE_RE = re.compile( + r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|" + r"mcp tools unavailable|no mcp path)\b", + re.IGNORECASE, +) +_FALLBACK_CLASSIFICATION_RE = re.compile( + r"\b(?:local fallback|fallback mode|used local gitea|ran local script)\b", + re.IGNORECASE, +) +_MCP_TOOL_USE_RE = re.compile( + r"\b(?:gitea_list_prs|gitea_view_pr|gitea_review_pr|gitea_merge_pr|" + r"gitea_resolve_task_capability|gitea_whoami|mcp tool)\b", + re.IGNORECASE, +) +_LOCAL_SCRIPT_RE = re.compile( + r"(?:^|[\s\"'`/])(?:" + "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES) + r")", + re.IGNORECASE, +) +_PROFILE_ACCESS_RE = re.compile( + r"(?:read|open|inspect|cat|view|access(?:ed)?|loaded?)\s+(?:file\s+)?[`'\"]?" + r"[^`'\"]*profiles\.json|profiles\.json[`'\"]?\s+(?:read|opened|inspected|accessed)|" + r"~/?\.config/gitea-tools/profiles", + re.IGNORECASE, +) +_LOCAL_ENV_SCRIPT_RE = re.compile( + r"GITEA_MCP_(?:CONFIG|PROFILE)=.*\b(?:python\s+)?(?:list_prs|view_pr|create_pr|merge_pr)\.py", + re.IGNORECASE, +) +_RECOVERY_PROOF_FIELDS = ( + ("identity proof", ("identity proof", "exact identity proof")), + ("profile proof", ("profile proof", "exact profile proof")), + ("repo proof", ("repo proof", "exact repo proof")), + ("capability proof", ("capability proof", "exact capability proof")), + ("mcp unavailable reason", ( + "why mcp was unavailable", + "mcp unavailable", + "mcp unavailability", + )), +) + + +def _collect_observed_text( + report_text: str, + action_log: list[dict] | None, +) -> str: + chunks = [report_text or ""] + for entry in action_log or []: + for key in ("command", "action", "detail", "path", "script"): + value = entry.get(key) + if value: + chunks.append(str(value)) + return "\n".join(chunks) + + +def _detect_profile_secret_access(text: str) -> list[str]: + reasons = [] + lower = text.lower() + if _PROFILE_ACCESS_RE.search(text): + reasons.append("report or action log shows profiles.json/profile secret access") + for marker in PROFILE_SECRET_MARKERS: + if marker == "profiles.json": + continue + if marker in lower and "do not" not in lower and "must not" not in lower: + if any(verb in lower for verb in ("read ", "open ", "inspect ", "cat ", "loaded ")): + reasons.append(f"profile secret surface '{marker}' accessed during review") + return reasons + + +def _detect_local_script_use(text: str) -> list[str]: + reasons = [] + match = _LOCAL_SCRIPT_RE.search(text) + if match: + reasons.append( + f"local Gitea helper script invoked ({match.group(0).strip()})" + ) + if _LOCAL_ENV_SCRIPT_RE.search(text): + reasons.append( + "local Gitea script run with GITEA_MCP_CONFIG/GITEA_MCP_PROFILE env overrides" + ) + return reasons + + +def _recovery_proof_missing(text: str) -> list[str]: + lower = text.lower() + missing = [] + for label, aliases in _RECOVERY_PROOF_FIELDS: + if not any(alias in lower for alias in aliases): + missing.append(label) + if not _FALLBACK_CLASSIFICATION_RE.search(text): + missing.append("fallback classification") + if not _RECOVERY_MODE_RE.search(text): + missing.append("recovery mode declaration") + return missing + + +def assess_reviewer_fallback_report( + report_text: str, + *, + action_log: list[dict] | None = None, + recovery_mode: bool = False, + mcp_available: bool = True, + mcp_tools_used: bool | None = None, +) -> dict[str, Any]: + """Fail closed when normal reviewer workflows use local Gitea fallbacks (#324).""" + text = _collect_observed_text(report_text, action_log) + lower = (report_text or "").lower() + + profile_violations = _detect_profile_secret_access(text) + script_violations = _detect_local_script_use(text) + violations = profile_violations + script_violations + + if mcp_tools_used is None: + mcp_tools_used = bool(_MCP_TOOL_USE_RE.search(report_text or "")) + elif mcp_tools_used is False and _MCP_TOOL_USE_RE.search(report_text or ""): + mcp_tools_used = True + + reasons: list[str] = [] + recovery_declared = recovery_mode or bool(_RECOVERY_MODE_RE.search(report_text or "")) + + if violations and mcp_available and not recovery_declared: + reasons.extend(violations) + if mcp_tools_used: + reasons.append( + "MCP tools were available and used; local fallback/profile access is forbidden" + ) + else: + reasons.append( + "MCP path is available; normal review must not use local Gitea fallbacks" + ) + + if violations and recovery_declared: + missing = _recovery_proof_missing(report_text or "") + if missing: + reasons.append( + "recovery-mode fallback missing proof fields: " + + ", ".join(missing) + ) + + if ( + not violations + and recovery_declared + and _FALLBACK_CLASSIFICATION_RE.search(report_text or "") + and not recovery_mode + ): + missing = _recovery_proof_missing(report_text or "") + if missing: + reasons.append( + "report claims local fallback but recovery proof is incomplete: " + + ", ".join(missing) + ) + + proven = not reasons + blocked = bool(reasons) and not recovery_declared or any( + "recovery-mode fallback missing" in r or "recovery proof is incomplete" in r + for r in reasons + ) + return { + "proven": proven, + "block": blocked or (bool(reasons) and mcp_available and not recovery_declared), + "downgraded": bool(reasons) and not blocked, + "recovery_mode": recovery_declared, + "mcp_available": mcp_available, + "mcp_tools_used": mcp_tools_used, + "profile_violations": profile_violations, + "script_violations": script_violations, + "reasons": reasons, + "safe_next_action": ( + "use MCP reviewer tools only; do not read profiles.json or run local Gitea scripts" + if reasons and not recovery_declared + else ( + "complete recovery-mode fallback proof before claiming local fallback" + if reasons + else "proceed with MCP tools" + ) + ), + } \ 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/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 c20eaf4..bbb8c05 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -118,6 +118,12 @@ def test_start_issue_template_references_work_issue_workflow(): assert "workflows/work-issue.md" in text +def test_reviewer_fallback_verifier_exported(): + from review_proofs import assess_reviewer_fallback_report + + assert callable(assess_reviewer_fallback_report) + + def test_final_report_validator_exported(): from review_proofs import assess_final_report_validator @@ -136,7 +142,19 @@ def test_validation_integrity_verifier_exported(): assert callable(assess_validation_integrity_report) +def test_prior_blocker_skip_verifier_exported(): + from review_proofs import assess_prior_blocker_skip_proof + + assert callable(assess_prior_blocker_skip_proof) + + 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_worktree_ownership_verifier_exported(): + from review_proofs import assess_worktree_ownership_report + + assert callable(assess_worktree_ownership_report) diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 3a3ea7a..8721545 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, assess_author_pr_report, + assess_full_suite_failure_approval_gate, assess_queue_ordering_proof, assess_reviewer_baseline_validation_proof, assess_capability_evidence, @@ -2539,5 +2540,170 @@ class TestReviewerBaselineValidationProof(unittest.TestCase): self.assertFalse(report["baseline_validation_proven"]) +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() diff --git a/tests/test_reviewer_blocker_skip.py b/tests/test_reviewer_blocker_skip.py new file mode 100644 index 0000000..b037ede --- /dev/null +++ b/tests/test_reviewer_blocker_skip.py @@ -0,0 +1,139 @@ +"""Tests for prior-blocker skip proof verifier (#318).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_blocker_skip import assess_prior_blocker_skip_proof # noqa: E402 + +HEAD_CURRENT = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" +HEAD_BLOCKER = "1111111111111111111111111111111111111111" + + +def _good_blocker_skip_report(pr_number: int) -> str: + return "\n".join([ + f"Skipped PR #{pr_number} due to prior REQUEST_CHANGES.", + f"- PR number: #{pr_number}", + f"- Current head SHA: {HEAD_CURRENT}", + "- Blocking review decision: request_changes", + f"- Blocking review head SHA: {HEAD_CURRENT}", + "- Head unchanged since blocker", + "- Reason remains blocked: unresolved review feedback at same head", + "- Blocker revalidated live in this session via gitea_get_pr_review_feedback", + ]) + + +class TestPriorBlockerSkipProof(unittest.TestCase): + def test_no_skips_passes(self): + result = assess_prior_blocker_skip_proof( + "Selected PR #281 for review.", + skipped_prs=[], + ) + self.assertTrue(result["proven"]) + + def test_blocked_unchanged_with_live_proof_passes(self): + report = _good_blocker_skip_report(280) + result = assess_prior_blocker_skip_proof( + report, + skipped_prs=[{ + "pr_number": 280, + "head_sha": HEAD_CURRENT, + "blocking_decision": "request_changes", + "blocking_review_head_sha": HEAD_CURRENT, + "head_changed_since_blocker": False, + "blocker_verified": True, + }], + ) + self.assertTrue(result["proven"]) + + def test_head_changed_after_blocker_blocks_skip(self): + result = assess_prior_blocker_skip_proof( + _good_blocker_skip_report(280), + skipped_prs=[{ + "pr_number": 280, + "head_sha": HEAD_CURRENT, + "blocking_decision": "request_changes", + "blocking_review_head_sha": HEAD_BLOCKER, + "head_changed_since_blocker": True, + "blocker_verified": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("head changed" in r for r in result["reasons"])) + + def test_missing_live_proof_blocks(self): + report = _good_blocker_skip_report(280).replace( + "- Blocker revalidated live in this session via gitea_get_pr_review_feedback", + "", + ) + result = assess_prior_blocker_skip_proof( + report, + skipped_prs=[{ + "pr_number": 280, + "head_sha": HEAD_CURRENT, + "blocking_decision": "request_changes", + "blocking_review_head_sha": HEAD_CURRENT, + "head_changed_since_blocker": False, + "blocker_verified": True, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("live blocker proof" in r for r in result["reasons"])) + + def test_blocker_unverified_classification_passes(self): + report = "\n".join([ + "Skipped PR #280.", + "Classification: BLOCKER_STATUS_UNVERIFIED", + "Review feedback could not be fetched live in this session.", + ]) + result = assess_prior_blocker_skip_proof( + report, + skipped_prs=[{ + "pr_number": 280, + "blocking_decision": "request_changes", + "blocker_verified": False, + }], + ) + self.assertTrue(result["proven"]) + + def test_unverified_without_classification_blocks(self): + result = assess_prior_blocker_skip_proof( + "Skipped PR #280 based on memory from last session.", + skipped_prs=[{ + "pr_number": 280, + "blocking_decision": "request_changes", + "blocker_verified": False, + }], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("BLOCKER_STATUS_UNVERIFIED" in r for r in result["reasons"])) + + def test_missing_blocking_head_sha_blocks(self): + report = _good_blocker_skip_report(280).replace( + f"- Blocking review head SHA: {HEAD_CURRENT}", + "", + ) + result = assess_prior_blocker_skip_proof( + report, + skipped_prs=[{ + "pr_number": 280, + "head_sha": HEAD_CURRENT, + "blocking_decision": "request_changes", + "blocking_review_head_sha": HEAD_CURRENT, + "head_changed_since_blocker": False, + "blocker_verified": True, + }], + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_prior_blocker_skip_proof as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reviewer_fallback.py b/tests/test_reviewer_fallback.py new file mode 100644 index 0000000..59fce4e --- /dev/null +++ b/tests/test_reviewer_fallback.py @@ -0,0 +1,112 @@ +"""Tests for reviewer local-fallback verifier (#324).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_fallback import assess_reviewer_fallback_report # noqa: E402 + + +class TestReviewerFallbackNormalMode(unittest.TestCase): + def test_clean_mcp_report_passes(self): + report = ( + "Used gitea_list_prs and gitea_view_pr via MCP reviewer namespace.\n" + "Capability evidence: gitea_resolve_task_capability(review_pr)." + ) + result = assess_reviewer_fallback_report(report, mcp_available=True) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_profiles_json_access_blocks(self): + report = ( + "Inspected ~/.config/gitea-tools/profiles.json for reviewer credentials.\n" + "Then used gitea_view_pr." + ) + result = assess_reviewer_fallback_report(report, mcp_available=True) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("profiles.json" in r for r in result["reasons"])) + + def test_local_list_prs_script_blocks_when_mcp_available(self): + report = ( + "Ran `python list_prs.py --remote prgs` with GITEA_MCP_CONFIG set.\n" + "MCP tools were also available." + ) + result = assess_reviewer_fallback_report( + report, + mcp_available=True, + action_log=[{"command": "GITEA_MCP_PROFILE=prgs-reviewer python list_prs.py"}], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(result["script_violations"]) + + def test_action_log_profile_access_blocks(self): + result = assess_reviewer_fallback_report( + "Review complete via MCP.", + action_log=[{"path": "/Users/me/.config/gitea-tools/profiles.json", "action": "read"}], + mcp_available=True, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + +class TestReviewerFallbackRecoveryMode(unittest.TestCase): + def _recovery_report(self, **extra): + base = "\n".join([ + "Recovery mode: MCP tools unavailable in this session.", + "Local fallback: ran python view_pr.py after capability proof.", + "Identity proof: sysadmin / prgs-reviewer", + "Profile proof: prgs-reviewer", + "Repo proof: Scaled-Tech-Consulting/Gitea-Tools", + "Capability proof: gitea.pr.review allowed for recovery command", + "Why MCP was unavailable: reviewer MCP namespace failed to start", + ]) + return base + ("\n" + extra.get("append", "") if extra.get("append") else "") + + def test_recovery_fallback_with_proof_passes(self): + result = assess_reviewer_fallback_report( + self._recovery_report(), + action_log=[{"command": "python view_pr.py --remote prgs"}], + mcp_available=False, + recovery_mode=True, + ) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_recovery_fallback_without_proof_blocks(self): + report = ( + "Recovery mode engaged.\n" + "Local fallback: python list_prs.py\n" + ) + result = assess_reviewer_fallback_report( + report, + mcp_available=False, + recovery_mode=True, + action_log=[{"command": "python list_prs.py"}], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue( + any("recovery" in r.lower() for r in result["reasons"]) + ) + + def test_mcp_unavailable_allows_recovery_attempt(self): + result = assess_reviewer_fallback_report( + self._recovery_report(), + mcp_available=False, + recovery_mode=True, + ) + self.assertTrue(result["proven"]) + + +class TestReviewerFallbackExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_reviewer_fallback_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