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

fix: resolve conflicts for PR #365 against current master

Browse Source
This commit is contained in:
sysadmin
2026-07-07 09:17:15 -04:00
parent b71d3aeab3 e017d80256
commit a237b15559
11 changed files with 1972 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)
# ---------------------------------------------------------------------------
@@ -3363,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)
# ---------------------------------------------------------------------------
@@ -3444,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
@@ -3834,6 +4237,13 @@ 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
@@ -3846,3 +4256,10 @@ def assess_mutation_categories_report(report_text, **kwargs):
from reviewer_mutation_categories import assess_mutation_categories_report as _assess
return _assess(report_text, **kwargs)
def assess_worktree_ownership_report(report_text, **kwargs):
"""#312: prove session-owned or safe-reuse worktree before reset/validation."""
from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
return _assess(report_text, **kwargs)