Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

fix: resolve conflicts for PR #368 against current master

Browse Source
This commit is contained in:
sysadmin
2026-07-07 09:15:26 -04:00
parent 876f1ee3a3 c21e756a30
commit 5131fc595a
10 changed files with 1951 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+418 -1
View File
@@ -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)
@@ -1351,6 +1422,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,
queue_ordering=None,
baseline_validation=None, project_root=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
@@ -1378,6 +1450,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
)
empty_queue_report = assess_empty_queue_report(report_text)
if queue_ordering is None and report_text:
queue_ordering_proof = assess_queue_ordering_proof(
report_text=report_text,
)
elif queue_ordering is not None:
queue_ordering_proof = assess_queue_ordering_proof(
report_text=report_text,
queue_ordering=queue_ordering,
)
else:
queue_ordering_proof = {
"proven": True,
"block": False,
"reasons": [],
"violations": [],
}
queue_status_report = (
assess_queue_status_report(report_text)
if report_text and _QUEUE_STATUS_REPORT_HINT.search(report_text)
@@ -1529,6 +1617,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 queue_ordering_proof.get("proven"):
downgrade_reasons.append(
"queue ordering proof missing or failed (#321)"
)
downgrade_reasons.extend(queue_ordering_proof.get("reasons", []))
if not proof_wording.get("proven"):
downgrade_reasons.append(
"unsupported proof wording in final report (#330)"
@@ -1555,10 +1648,12 @@ 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 queue_ordering_proof.get("proven")
and baseline_validation.get("proven")
)
violations = []
violations.extend(queue_ordering_proof.get("violations", []))
violations.extend(baseline_validation.get("violations", []))
if merge_performed and not merge_allowed:
violations.append(
@@ -1611,6 +1706,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
else True
),
"empty_queue_trust_gate_status": empty_queue_report.get("status"),
"queue_ordering_proven": bool(queue_ordering_proof.get("proven")),
"queue_ordering_policy": queue_ordering_proof.get("policy"),
"queue_ordering_violations": list(
queue_ordering_proof.get("violations") or []
),
"proof_wording_proven": bool(proof_wording.get("proven")),
"proof_wording_violations": list(proof_wording.get("violations") or []),
"queue_status_report_proven": bool(queue_status_report.get("proven")),
@@ -1762,6 +1862,20 @@ HANDOFF_BASE_FIELDS = (
("Safety", ("safety",)),
)
# Issue #320: reviewer handoffs replace the legacy ambiguous
# "Workspace mutations" field with these precise mutation categories.
HANDOFF_REVIEW_MUTATION_FIELDS = (
("File edits by reviewer", ("file edits by reviewer",)),
("Worktree/index mutations", ("worktree/index mutations",)),
("Git ref mutations", ("git ref mutations",)),
("MCP/Gitea mutations", ("mcp/gitea mutations",)),
("Review mutations", ("review mutations",)),
("Merge mutations", ("merge mutations",)),
("Cleanup mutations", ("cleanup mutations",)),
("External-state mutations", ("external-state mutations",)),
("Read-only diagnostics", ("read-only diagnostics",)),
)
HANDOFF_ROLE_FIELDS = {
"review": (
("Selected PR", ("selected pr",)),
@@ -1777,7 +1891,7 @@ HANDOFF_ROLE_FIELDS = {
("Merge result", ("merge result",)),
("Linked issue status", ("linked issue status", "linked issue")),
("Cleanup status", ("cleanup status", "cleanup")),
),
) + HANDOFF_REVIEW_MUTATION_FIELDS,
"author": (
("Selected issue", ("selected issue",)),
("Issue lock proof", ("issue lock proof", "lock before diff")),
@@ -1928,6 +2042,25 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
field for field in required
if field[0] not in ("Workspace mutations", "Mutations")
]
if role == "review":
# Issue #320: reviewer handoffs use the precise mutation categories
# in HANDOFF_REVIEW_MUTATION_FIELDS instead of the legacy ambiguous
# "Workspace mutations" field, which is rejected below.
required = [
field for field in required
if field[0] != "Workspace mutations"
]
if any(label.startswith("workspace mutations") for label in labels):
return {
"verdict": "incomplete",
"downgraded": True,
"missing_fields": [],
"reasons": [
"review handoff must not include legacy "
"'Workspace mutations' field; report the precise "
"mutation categories instead (issue #320)"
],
}
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
missing = []
@@ -3189,6 +3322,162 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict:
}
# ---------------------------------------------------------------------------
# Reviewer queue ordering proof (#321)
# ---------------------------------------------------------------------------
_QUEUE_SELECTION_CLAIM_RE = re.compile(
r"\b(?:next|oldest)\s+eligible\s+pr\b",
re.IGNORECASE,
)
_ORDERING_POLICY_LINE_RE = re.compile(
r"(?:queue\s+ordering\s+policy|selection\s+policy|ordering\s+policy)"
r"\s*:\s*(.+)",
re.IGNORECASE,
)
_SKIPPED_PR_LINE_RE = re.compile(
r"\bskipped\s+pr\s*#?(\d+)\b",
re.IGNORECASE,
)
def _parse_ordering_policy(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _ORDERING_POLICY_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(".")
return ""
def _skipped_pr_numbers(report_text: str) -> set[int]:
skipped: set[int] = set()
for match in _SKIPPED_PR_LINE_RE.finditer(report_text or ""):
try:
skipped.add(int(match.group(1)))
except ValueError:
continue
return skipped
def assess_queue_ordering_proof(
*,
report_text: str | None = None,
queue_ordering: dict | None = None,
) -> dict:
"""#321: reviewer queue selection must prove ordering before claiming next PR."""
text = report_text or ""
lower = text.lower()
proof = queue_ordering or {}
reasons: list[str] = []
violations: list[str] = []
claims_selection = bool(_QUEUE_SELECTION_CLAIM_RE.search(text))
selected = proof.get("selected_pr_number")
if selected is None and "selected pr #" in lower:
for token in lower.split():
if token.startswith("#") and token[1:].isdigit():
selected = int(token[1:])
break
if not claims_selection and selected is None:
return {
"proven": True,
"block": False,
"claims_selection": False,
"policy": None,
"reasons": [],
"violations": [],
"safe_next_action": "proceed",
}
policy = (proof.get("policy") or _parse_ordering_policy(text)).strip()
if not policy:
reasons.append(
"queue selection claimed without stated ordering policy (#321)"
)
policy_lower = policy.lower()
api_numbers: list[int] = []
for n in proof.get("api_pr_numbers") or []:
if isinstance(n, int):
api_numbers.append(n)
elif isinstance(n, str) and n.isdigit():
api_numbers.append(int(n))
skipped_entries = list(proof.get("skipped_earlier") or [])
skipped_numbers = {
int(entry["pr_number"])
for entry in skipped_entries
if isinstance(entry, dict) and entry.get("pr_number") is not None
}
skipped_numbers.update(_skipped_pr_numbers(text))
if selected is not None and api_numbers:
if "oldest" in policy_lower and "number" in policy_lower:
earlier_actionable = [
n for n in api_numbers
if n < int(selected) and n not in skipped_numbers
]
for pr_number in earlier_actionable:
entry = next(
(
e for e in skipped_entries
if isinstance(e, dict)
and int(e.get("pr_number", -1)) == pr_number
),
None,
)
reason = (entry or {}).get("reason", "").strip()
if not reason and f"skipped pr #{pr_number}" not in lower:
violations.append(
f"earlier actionable PR #{pr_number} skipped without "
"proof-backed reason (#321)"
)
reasons.append(
f"oldest-first policy requires skip reasoning for "
f"earlier PR #{pr_number} (#321)"
)
if api_numbers and api_numbers[0] != min(api_numbers):
if (
"api order" not in lower
and "sorted" not in lower
and "newest-first" not in lower
and not skipped_entries
):
reasons.append(
"API response order differs from oldest-first "
"selection but report does not prove explicit sort "
"or skip reasoning (#321)"
)
if "priority" in policy_lower:
override = (proof.get("priority_override") or "").strip()
if not override and "priority" not in lower:
reasons.append(
"priority-label ordering policy requires override "
"explanation in report (#321)"
)
proven = not reasons and not violations
return {
"proven": proven,
"block": bool(violations),
"claims_selection": True,
"policy": policy or None,
"selected_pr_number": selected,
"reasons": reasons,
"violations": violations,
"safe_next_action": (
"state queue ordering policy and proof-backed skip reasoning "
"for every earlier actionable PR"
if not proven
else "proceed"
),
}
# ---------------------------------------------------------------------------
# Reviewer baseline validation (#325)
# ---------------------------------------------------------------------------
@@ -3529,6 +3818,113 @@ def assess_already_landed_report_state(
}
# ---------------------------------------------------------------------------
# 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)
# ---------------------------------------------------------------------------
@@ -3610,6 +4006,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
@@ -4000,8 +4403,22 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
}
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)
reviewer_blocker_skip.py
+164
View File
@@ -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"
),
}
reviewer_fallback.py
+213
View File
@@ -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"
)
),
}
reviewer_worktree_ownership.py
+207
View File
@@ -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<N>-* "
"or document full safe-reuse proof before destructive reset"
if reasons
else "proceed"
),
}
tests/test_linked_issue_consistency.py
+118
View File
@@ -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()
tests/test_llm_workflow_split.py
+18
View File
@@ -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
@@ -130,7 +136,19 @@ def test_create_issue_final_report_verifier_exported():
assert callable(assess_create_issue_final_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)
tests/test_review_proofs.py
+425 -2
View File
@@ -26,6 +26,8 @@ from review_proofs import ( # noqa: E402
assess_already_landed_report_state,
assess_already_landed_review_gate,
assess_author_pr_report,
assess_full_suite_failure_approval_gate,
assess_queue_ordering_proof,
assess_reviewer_baseline_validation_proof,
assess_capability_evidence,
assess_capability_proof,
@@ -942,13 +944,18 @@ class TestControllerHandoff(unittest.TestCase):
self.assertIn("Safety", result["missing_fields"])
def test_review_role_requires_review_fields(self):
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
# Issue #320: review handoffs must not carry the legacy
# 'Workspace mutations' line, so strip it from the shared base.
review_base = "\n".join(
line for line in self.BASE_HANDOFF.splitlines()
if not line.startswith("- Workspace mutations:"))
result = assess_controller_handoff(review_base, role="review")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Pinned reviewed head", result["missing_fields"])
self.assertIn("Worktree path", result["missing_fields"])
self.assertIn("Merge result", result["missing_fields"])
complete = self.BASE_HANDOFF + "\n" + "\n".join([
complete = review_base + "\n" + "\n".join([
"- Selected PR: #999",
"- Reviewer eligibility: passed",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
@@ -960,6 +967,15 @@ class TestControllerHandoff(unittest.TestCase):
"- Merge result: merged",
"- Linked issue status: closed",
"- Cleanup status: branch deleted",
"- File edits by reviewer: none",
"- Worktree/index mutations: created and removed review worktree",
"- Git ref mutations: git fetch prgs",
"- MCP/Gitea mutations: none",
"- Review mutations: one approve on #999",
"- Merge mutations: PR #999 merged",
"- Cleanup mutations: removed review worktree",
"- External-state mutations: none",
"- Read-only diagnostics: git log, git diff",
])
result = assess_controller_handoff(complete, role="review")
self.assertEqual(result["verdict"], "complete")
@@ -1082,6 +1098,144 @@ class TestControllerHandoff(unittest.TestCase):
self.assertEqual(res2["verdict"], "complete")
class TestReviewHandoffPreciseMutationCategories(unittest.TestCase):
"""Issue #320: reviewer handoffs drop legacy 'Workspace mutations'."""
REVIEW_BASE = "\n".join([
"## Controller Handoff",
"",
"- Task: review PR #999",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: reviewer",
"- Identity: jcwalker3 / prgs-reviewer",
"- Issue/PR: PR #999",
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Files changed: review_proofs.py",
"- Validation: 700 passed, 6 skipped",
"- Mutations: one review submitted",
"- Current status: review complete",
"- Blockers: none",
"- Next: controller decides",
"- Safety: no self-review; no self-merge; no secrets",
"- Selected PR: #999",
"- Reviewer eligibility: passed",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Worktree path: /repo/branches/review-pr-999",
"- Worktree dirty: no",
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
"- Unrelated local mutations: none",
"- Review decision: approve",
"- Merge result: none",
"- Linked issue status: open",
"- Cleanup status: worktree removed",
])
PRECISE_CATEGORIES = "\n".join([
"- File edits by reviewer: none",
"- Worktree/index mutations: created and removed review worktree",
"- Git ref mutations: git fetch prgs",
"- MCP/Gitea mutations: none",
"- Review mutations: one approve on #999",
"- Merge mutations: none",
"- Cleanup mutations: removed review worktree",
"- External-state mutations: none",
"- Read-only diagnostics: git log, git diff",
])
def _complete_handoff(self, **overrides):
text = self.REVIEW_BASE + "\n" + self.PRECISE_CATEGORIES
for old, new in overrides.items():
text = text.replace(old, new)
return text
def test_review_handoff_with_precise_categories_is_complete(self):
result = assess_controller_handoff(
self._complete_handoff(), role="review")
self.assertEqual(result["verdict"], "complete")
self.assertFalse(result["downgraded"])
def test_review_handoff_rejects_legacy_workspace_mutations_none(self):
text = self._complete_handoff() + "\n- Workspace mutations: none"
result = assess_controller_handoff(text, role="review")
self.assertEqual(result["verdict"], "incomplete")
self.assertTrue(result["downgraded"])
self.assertTrue(any(
"workspace mutations" in reason.lower()
for reason in result["reasons"]
))
def test_review_handoff_rejects_legacy_workspace_mutations_any_value(self):
text = (self._complete_handoff()
+ "\n- Workspace mutations: edited review_proofs.py")
result = assess_controller_handoff(text, role="review")
self.assertEqual(result["verdict"], "incomplete")
self.assertTrue(any(
"workspace mutations" in reason.lower()
for reason in result["reasons"]
))
def test_review_handoff_missing_precise_categories_is_incomplete(self):
result = assess_controller_handoff(self.REVIEW_BASE, role="review")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("File edits by reviewer", result["missing_fields"])
self.assertIn("Worktree/index mutations", result["missing_fields"])
self.assertIn("Read-only diagnostics", result["missing_fields"])
def test_review_handoff_does_not_require_workspace_mutations(self):
result = assess_controller_handoff(
self._complete_handoff(), role="review")
self.assertNotIn("Workspace mutations", result["missing_fields"])
def test_request_changes_handoff_with_precise_categories_is_complete(self):
text = self._complete_handoff(**{
"- Review decision: approve": "- Review decision: request-changes",
"- Review mutations: one approve on #999":
"- Review mutations: one request_changes on #999",
})
result = assess_controller_handoff(text, role="review")
self.assertEqual(result["verdict"], "complete")
def test_merge_handoff_with_precise_categories_is_complete(self):
text = self._complete_handoff(**{
"- Merge result: none": "- Merge result: merged",
"- Merge mutations: none": "- Merge mutations: PR #999 merged",
})
result = assess_controller_handoff(text, role="review")
self.assertEqual(result["verdict"], "complete")
def test_fetch_only_handoff_with_precise_categories_is_complete(self):
text = self._complete_handoff(**{
"- Review decision: approve": "- Review decision: none",
"- Review mutations: one approve on #999": "- Review mutations: none",
"- Worktree/index mutations: created and removed review worktree":
"- Worktree/index mutations: none",
"- Cleanup mutations: removed review worktree":
"- Cleanup mutations: none",
"- Mutations: one review submitted": "- Mutations: fetch only",
})
result = assess_controller_handoff(text, role="review")
self.assertEqual(result["verdict"], "complete")
def test_blocked_handoff_with_precise_categories_is_complete(self):
text = self._complete_handoff(**{
"- Blockers: none": "- Blockers: infra_stop (registry unreachable)",
"- Review decision: approve": "- Review decision: none",
"- Review mutations: one approve on #999": "- Review mutations: none",
})
result = assess_controller_handoff(text, role="review")
self.assertEqual(result["verdict"], "complete")
def test_author_role_still_requires_workspace_mutations(self):
# Non-review roles keep the legacy contract until their own issues
# migrate them.
author = TestControllerHandoff.BASE_HANDOFF
stripped = "\n".join(
line for line in author.splitlines()
if not line.startswith("- Workspace mutations:"))
result = assess_controller_handoff(stripped, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Workspace mutations", result["missing_fields"])
class TestReviewMutationFinalReport(unittest.TestCase):
"""Final reports must list exactly one live review mutation."""
@@ -2163,6 +2317,111 @@ class TestCreateIssueFinalReport(unittest.TestCase):
self.assertTrue(result["downgraded"])
class TestQueueOrderingProof(unittest.TestCase):
"""Issue #321: reviewer queue selection must prove ordering."""
def test_next_eligible_claim_without_policy_fails(self):
result = assess_queue_ordering_proof(
report_text="Selected the next eligible PR: PR #286.",
)
self.assertFalse(result["proven"])
self.assertTrue(result["claims_selection"])
def test_next_eligible_claim_with_policy_passes(self):
result = assess_queue_ordering_proof(
report_text=(
"Queue ordering policy: oldest PR number first\n"
"Selected the next eligible PR: PR #286."
),
)
self.assertTrue(result["proven"])
self.assertEqual(result["policy"], "oldest PR number first")
def test_newest_first_api_with_oldest_first_selection_passes(self):
result = assess_queue_ordering_proof(
report_text=(
"Queue ordering policy: oldest PR number first\n"
"API order was newest-first (PR #291 before PR #286).\n"
"Selected the next eligible PR: PR #286."
),
queue_ordering={
"policy": "oldest PR number first",
"selected_pr_number": 286,
"api_pr_numbers": [291, 286],
},
)
self.assertTrue(result["proven"])
def test_earlier_actionable_pr_without_skip_reason_fails(self):
result = assess_queue_ordering_proof(
report_text=(
"Queue ordering policy: oldest PR number first\n"
"Selected the next eligible PR: PR #300."
),
queue_ordering={
"policy": "oldest PR number first",
"selected_pr_number": 300,
"api_pr_numbers": [300, 286],
},
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_priority_override_requires_explanation(self):
result = assess_queue_ordering_proof(
report_text="Selected the next eligible PR: PR #300.",
queue_ordering={
"policy": "priority label",
"selected_pr_number": 300,
"api_pr_numbers": [286, 300],
},
)
self.assertFalse(result["proven"])
def test_priority_override_with_explanation_passes(self):
result = assess_queue_ordering_proof(
report_text=(
"Queue ordering policy: priority label\n"
"Priority override: security label on PR #300.\n"
"Selected the next eligible PR: PR #300."
),
queue_ordering={
"policy": "priority label",
"selected_pr_number": 300,
"api_pr_numbers": [286, 300],
"priority_override": "security label on PR #300",
},
)
self.assertTrue(result["proven"])
def test_build_final_report_downgrades_missing_ordering_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="Selected the next eligible PR: PR #286.",
)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["queue_ordering_proven"])
class TestReviewerBaselineValidationProof(unittest.TestCase):
"""Issue #325: baseline validation must use branches/ worktrees."""
@@ -2414,6 +2673,170 @@ class TestAlreadyLandedReportState(unittest.TestCase):
"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"])
tests/test_reviewer_blocker_skip.py
+139
View File
@@ -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()
tests/test_reviewer_fallback.py
+112
View File
@@ -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()
tests/test_reviewer_worktree_ownership.py
+137
View File
@@ -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()