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 #361

Browse Source
This commit is contained in:
sysadmin
2026-07-07 05:33:47 -04:00
parent 5fa5760dd0 4d3e90e7c5
commit a0d68ef66e
9 changed files with 2229 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+604
View File
@@ -1052,6 +1052,125 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non
return None
# ── Workspace-vs-worktree mutation consistency (Issue #313) ──────────────────
#
# A worktree reset/checkout/clean is a workspace mutation even when the
# reviewer edited no files by hand. Final reports must therefore never say
# "Workspace mutations: none" once a worktree mutation occurred, and every
# observed worktree / git ref mutation must appear under its precise
# category field.
WORKTREE_MUTATION_MARKERS = (
"reset",
"checkout",
"switch",
"restore",
"clean",
"stash",
"merge",
"rebase",
"cherry-pick",
"worktree add",
"worktree remove",
"worktree prune",
)
GIT_REF_MUTATION_MARKERS = (
"fetch",
"pull",
)
def _report_labeled_fields(report_text):
"""Return {lowercase label: value} for every 'Label: value' line."""
fields = {}
for line in (report_text or "").splitlines():
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
k, v = stripped.split(":", 1)
fields[k.strip().lower()] = v.strip()
return fields
def _field_claims_none(fields, label):
"""True when *label* is present and its value is empty or 'none...'."""
value = fields.get(label)
if value is None:
return False
value = value.strip().lower()
return not value or value.startswith("none")
def _classify_git_commands(observed_commands):
"""Split observed git commands into worktree and ref mutations."""
worktree_cmds = []
ref_cmds = []
for cmd in observed_commands or []:
lowered = " ".join((cmd or "").lower().split())
if "git" not in lowered:
continue
if any(marker in lowered for marker in WORKTREE_MUTATION_MARKERS):
worktree_cmds.append(cmd)
elif any(marker in lowered for marker in GIT_REF_MUTATION_MARKERS):
ref_cmds.append(cmd)
return worktree_cmds, ref_cmds
def assess_workspace_mutation_consistency(report_text, observed_commands=None):
"""#313: reject 'Workspace mutations: none' after worktree mutations.
*observed_commands* is the optional list of git commands the workflow
actually ran. Even without it, a report that itself lists a non-none
``Worktree mutations`` value while claiming ``Workspace mutations:
none`` is internally contradictory and fails.
Returns {'complete', 'downgraded', 'reasons'}; fails closed on any
contradiction or unreported observed mutation.
"""
fields = _report_labeled_fields(report_text)
worktree_cmds, ref_cmds = _classify_git_commands(observed_commands)
reported_worktree = fields.get("worktree mutations")
reported_worktree_nonnone = bool(
reported_worktree and not reported_worktree.strip().lower().startswith("none")
)
reasons = []
workspace_none = _field_claims_none(fields, "workspace mutations")
if workspace_none and (worktree_cmds or reported_worktree_nonnone):
detail = worktree_cmds[0] if worktree_cmds else reported_worktree
reasons.append(
"report claims 'Workspace mutations: none' but worktree "
f"mutations occurred ({detail}); use precise categories such as "
"'File edits by reviewer: none' plus 'Worktree mutations: <cmd>'"
)
if worktree_cmds and (
"worktree mutations" not in fields
or _field_claims_none(fields, "worktree mutations")
):
reasons.append(
"observed worktree mutation not reported under 'Worktree "
f"mutations': {worktree_cmds[0]}"
)
if ref_cmds and (
"git ref mutations" not in fields
or _field_claims_none(fields, "git ref mutations")
):
reasons.append(
"observed git ref mutation not reported under 'Git ref "
f"mutations': {ref_cmds[0]}"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
justification=None):
"""Assess reviewer/author role separation for blind queue workflows.
@@ -1232,6 +1351,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.
@@ -1259,6 +1379,32 @@ 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)
else {"proven": True, "block": False, "reasons": [], "violations": []}
)
proof_wording = (
assess_proof_wording(report_text)
if report_text
else {"proven": True, "block": False, "reasons": [], "violations": []}
)
if baseline_validation is None and report_text:
baseline_validation = assess_reviewer_baseline_validation_proof(
report_text=report_text,
@@ -1400,6 +1546,21 @@ 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)"
)
downgrade_reasons.extend(proof_wording.get("reasons", []))
if not queue_status_report.get("proven"):
downgrade_reasons.append(
"queue-status report violates loaded workflow proof gates (#339)"
)
downgrade_reasons.extend(queue_status_report.get("reasons", []))
if not baseline_validation.get("proven"):
downgrade_reasons.append(
"reviewer baseline validation proof missing or failed (#325)"
@@ -1416,10 +1577,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(
@@ -1472,6 +1635,17 @@ 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")),
"queue_status_violations": list(
queue_status_report.get("violations") or []
),
"baseline_validation_proven": bool(baseline_validation.get("proven")),
"baseline_validation_violations": list(
baseline_validation.get("violations") or []
@@ -3044,6 +3218,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)
# ---------------------------------------------------------------------------
@@ -3406,6 +3736,280 @@ def assess_email_disclosure(
}
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
return _validate(report_text, task_kind, **kwargs)
_QUEUE_STATUS_REPORT_HINT = re.compile(
r"queue[- ]status|selected pr:\s*none|no pr selected|"
r"queue[- ]status[- ]only",
re.I,
)
_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
_NOT_APPLICABLE_VALUE = re.compile(
r"\b(?:none|not applicable|n/?a|not run|not verified|—|-)\b",
re.I,
)
_PRIOR_PROOF_LABEL = re.compile(
r"prior (?:blocker|proof|request[- ]changes|feedback)|"
r"head sha unchanged since|prior blocker reused|labeled as prior",
re.I,
)
_PAGINATION_FINALITY_EVIDENCE = re.compile(
r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|"
r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|"
r"no next page|final[- ]page|pr_inventory_trust_gate\.status|"
r"total_count\s*:|pagination.*(?:final|complete)|"
r"inventory pagination proof:",
re.I,
)
_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile(
r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|"
r"default (?:gitea )?page[- ]?size|under (?:the )?50[- ]?(?:item )?limit|"
r"(?:complete|exhaustive).*(?:default )?page[- ]?size|"
r"page[- ]?size assumption",
re.I,
)
_PROOF_WORDING_RULES: tuple[dict, ...] = (
{
"id": "live_proof",
"pattern": re.compile(r"\blive (?:blocker )?proof\b", re.I),
"session_keys": ("live_session_proof", "session_proof_commands"),
"text_evidence": re.compile(
r"current[- ]session|ran in (?:the )?current session|"
r"proof (?:command|tool).*(?:current|this) session|"
r"revalidated in (?:the )?current session",
re.I,
),
"allow_prior_label": True,
},
{
"id": "inventory_complete",
"pattern": re.compile(
r"\binventory (?:is )?complete\b|\binventory completeness\b",
re.I,
),
"session_keys": ("pagination_complete", "inventory_complete"),
"text_evidence": _PAGINATION_FINALITY_EVIDENCE,
"allow_prior_label": False,
},
{
"id": "same_as_master",
"pattern": re.compile(r"\bsame as master\b", re.I),
"session_keys": ("baseline_worktree_proof", "clean_baseline_proof"),
"text_evidence": re.compile(
r"baseline worktree|clean baseline|diff base.*master|"
r"base[- ]equivalent.*master|worktree proof",
re.I,
),
"allow_prior_label": False,
},
{
"id": "no_file_edits",
"pattern": re.compile(
r"\b(?:no file edits|file edits none|workspace mutations none)\b",
re.I,
),
"session_keys": ("no_file_edits_proven", "mutation_ledger_clean"),
"text_evidence": re.compile(
r"mutation ledger|worktree mutations:\s*none|"
r"file edits:\s*none|tracked file edits:\s*none|"
r"no tracked (?:file )?edits",
re.I,
),
"allow_prior_label": False,
},
)
_QUEUE_STATUS_GATES_NO_PR = (
"already-landed gate",
"author-safety result",
"merge preflight",
)
_REVIEW_WORKTREE_DETAIL_FIELDS = (
"review worktree dirty before validation",
"review worktree dirty after validation",
"review worktree head state",
)
def _controller_handoff_field_map(report_text: str | None) -> dict[str, str]:
"""Parse ``- Field: value`` lines from the Controller Handoff section."""
section = _handoff_section_lines(report_text)
if section is None:
return {}
fields: dict[str, str] = {}
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" not in stripped:
continue
key, value = stripped.split(":", 1)
fields[key.strip().lower()] = value.strip()
return fields
def _queue_status_only_run(fields: dict[str, str], report_text: str) -> bool:
selected = fields.get("selected pr", "")
if selected and _NOT_APPLICABLE_VALUE.search(selected):
return True
if re.search(r"\bnone\b", selected, re.I):
return True
if re.search(
r"no pr selected|queue[- ]status[- ]only|selected pr:\s*none",
report_text or "",
re.I,
):
return True
return not selected
def assess_proof_wording(
report_text: str | None,
*,
session_evidence: dict | None = None,
) -> dict:
"""Issue #330: reject unsupported proof phrases in final reports.
Forbidden wording such as ``inventory complete``, ``live proof``, and
``same as master`` is allowed only when matching current-session evidence
appears in the report text or in *session_evidence*.
"""
text = report_text or ""
evidence = dict(session_evidence or {})
violations: list[str] = []
flagged_rules: list[str] = []
if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text):
if not _PAGINATION_FINALITY_EVIDENCE.search(text) and not (
evidence.get("pagination_complete") or evidence.get("inventory_complete")
):
violations.append(
"inventory pagination assumed from default page size without "
"final-page/no-next-page/total-count/traversal proof"
)
flagged_rules.append("default_page_size_assumption")
for rule in _PROOF_WORDING_RULES:
if not rule["pattern"].search(text):
continue
if rule.get("allow_prior_label") and _PRIOR_PROOF_LABEL.search(text):
continue
if any(evidence.get(key) for key in rule.get("session_keys") or ()):
continue
if rule["text_evidence"].search(text):
continue
violations.append(
f"forbidden proof phrase ({rule['id']}) without matching evidence"
)
flagged_rules.append(rule["id"])
proven = not violations
return {
"proven": proven,
"block": not proven,
"violations": violations,
"flagged_rules": flagged_rules,
"reasons": violations,
}
def assess_queue_status_report(
report_text: str | None,
*,
session_evidence: dict | None = None,
) -> dict:
"""Issue #339: reject contradictory reviewer queue-status-only reports."""
text = report_text or ""
fields = _controller_handoff_field_map(text)
violations: list[str] = []
proof = assess_proof_wording(text, session_evidence=session_evidence)
violations.extend(proof.get("violations", []))
if re.search(r"prior diagnostic", text, re.I) and not _PRIOR_PROOF_LABEL.search(
text
):
violations.append(
"prior diagnostic worktree simulation used as current conflict proof"
)
queue_only = _queue_status_only_run(fields, text)
if queue_only:
for gate in _QUEUE_STATUS_GATES_NO_PR:
value = fields.get(gate, "")
if value and _GATE_PASSED_VALUE.search(value):
if not _NOT_APPLICABLE_VALUE.search(value):
violations.append(
f"{gate} cannot be 'passed' when no PR is selected (#339)"
)
worktree_used = fields.get("review worktree used", "")
if worktree_used and re.search(r"\bfalse\b", worktree_used, re.I):
for detail_field in _REVIEW_WORKTREE_DETAIL_FIELDS:
detail_value = fields.get(detail_field, "")
if (
detail_value
and not _NOT_APPLICABLE_VALUE.search(detail_value)
):
violations.append(
f"{detail_field} must be 'not applicable' or 'none' "
"when no review worktree was created"
)
blockers = fields.get("blockers", "")
if blockers and re.search(r"\bnone\b", blockers, re.I):
if re.search(
r"all (?:open )?prs? (?:are |is )?(?:conflicted|blocked|unverified)|"
r"every (?:open )?pr (?:is )?(?:conflicted|blocked|unverified)",
text,
re.I,
):
violations.append(
"Blockers: none contradicts report stating all PRs are "
"conflicted, blocked, or unverified"
)
if re.search(r"skipped (?:pr|#)|earlier pr.*skipped", text, re.I):
has_skip_proof = bool(
re.search(
r"current[- ]session|prior blocker reused|conflict proof:|"
r"gitea_view_pr|review[- ]feedback proof|unchanged head",
text,
re.I,
)
or (session_evidence or {}).get("skip_proof_per_pr")
)
if not has_skip_proof:
violations.append(
"skipped PRs listed without current-session or labeled prior proof"
)
if re.search(r"workflows/review-merge-pr\.md", text, re.I) and violations:
violations.append(
"canonical workflow cited but mandatory queue-status proof gates violated"
)
deduped = list(dict.fromkeys(violations))
proven = not deduped
return {
"proven": proven,
"block": not proven,
"queue_status_only": queue_only,
"violations": deduped,
"reasons": deduped,
}
# ---------------------------------------------------------------------------
# Git ref mutations (#297)
# ---------------------------------------------------------------------------