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 #369 against latest master

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-07 09:31:42 -04:00
co-authored by Claude Opus 4.8
parent a5fb7d2472 7b71113b4d
commit 51d0c14009
8 changed files with 1498 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+494
View File
@@ -1242,6 +1242,102 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None):
}
# ── Already-landed report wording (Issue #298) ───────────────────────────────
#
# An already-landed PR is reconciliation-only: it must never be called
# eligible, its head SHA must never be called reviewed, and the legacy
# eligible/reviewed/scratch-worktree handoff fields must not appear.
# Reviewed-head claims in any report require validation + diff review to
# have actually passed.
ALREADY_LANDED_FORBIDDEN_PHRASES = (
"oldest eligible pr",
"next eligible pr",
"pinned reviewed head",
"scratch worktree used",
)
ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED"
def assess_already_landed_report_wording(report_text, *, already_landed,
review_validated=False):
"""#298: reject eligible/reviewed wording for already-landed PRs.
*already_landed* states whether the already-landed gate fired for the
selected PR. *review_validated* states whether validation and diff
review actually passed this session; only then may a reviewed head
SHA be populated in a non-already-landed report.
Returns {'complete', 'downgraded', 'reasons'}; fails closed.
"""
text_lower = (report_text or "").lower()
fields = _report_labeled_fields(report_text)
reasons = []
reviewed_head = fields.get("reviewed head sha")
reviewed_head_populated = bool(
reviewed_head and not reviewed_head.strip().lower().startswith("none")
)
pinned_head_present = "pinned reviewed head" in fields
if already_landed:
for phrase in ALREADY_LANDED_FORBIDDEN_PHRASES:
if phrase in text_lower:
reasons.append(
f"already-landed report must not use '{phrase}'; the PR "
"is reconciliation-only, not review/merge eligible"
)
eligibility = fields.get("eligibility class", "")
if ALREADY_LANDED_ELIGIBILITY_CLASS not in eligibility.upper():
reasons.append(
"already-landed report missing 'Eligibility class: "
f"{ALREADY_LANDED_ELIGIBILITY_CLASS}'"
)
for required in ("oldest open pr requiring action",
"candidate head sha"):
if required not in fields:
reasons.append(
f"already-landed report missing '{required}' field"
)
if "reviewed head sha" not in fields:
reasons.append(
"already-landed report must state 'Reviewed head SHA: none'"
)
elif reviewed_head_populated:
reasons.append(
"already-landed report must not claim a reviewed head SHA; "
"no validation or diff review passed for this PR"
)
worktree_used = fields.get("review worktree used", "")
if worktree_used.strip().lower() not in ("false", "no"):
reasons.append(
"already-landed report must state 'Review worktree used: "
"false'"
)
elif not review_validated:
if pinned_head_present:
reasons.append(
"'Pinned reviewed head' populated before validation and "
"diff review passed"
)
if reviewed_head_populated:
reasons.append(
"reviewed head SHA populated before validation and diff "
"review passed"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
justification=None):
"""Assess reviewer/author role separation for blind queue workflows.
@@ -1422,6 +1518,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None, worktree_proof=None,
validation_environment=None,
queue_ordering=None,
baseline_validation=None, project_root=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
@@ -1450,6 +1547,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
)
empty_queue_report = assess_empty_queue_report(report_text)
if validation_environment is None and report_text:
validation_env_proof = assess_validation_environment_proof(
report_text=report_text,
)
elif validation_environment is not None:
validation_env_proof = assess_validation_environment_proof(
report_text=report_text,
validation_environment=validation_environment,
)
else:
validation_env_proof = {
"proven": True,
"block": False,
"reasons": [],
"violations": [],
}
if queue_ordering is None and report_text:
queue_ordering_proof = assess_queue_ordering_proof(
report_text=report_text,
@@ -1622,6 +1735,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"empty-queue report missing or failed trust-gate proof (#198)"
)
downgrade_reasons.extend(empty_queue_report.get("reasons", []))
if not validation_env_proof.get("proven"):
downgrade_reasons.append(
"validation environment proof missing or failed (#311)"
)
downgrade_reasons.extend(validation_env_proof.get("reasons", []))
if not queue_ordering_proof.get("proven"):
downgrade_reasons.append(
"queue ordering proof missing or failed (#321)"
@@ -1658,11 +1776,13 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
# #179: no merge without a proven final live-state recheck.
and live_state_proven
and worktree_proven
and validation_env_proof.get("proven")
and queue_ordering_proof.get("proven")
and baseline_validation.get("proven")
)
violations = []
violations.extend(validation_env_proof.get("violations", []))
violations.extend(queue_ordering_proof.get("violations", []))
violations.extend(baseline_validation.get("violations", []))
if merge_performed and not merge_allowed:
@@ -1716,6 +1836,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
else True
),
"empty_queue_trust_gate_status": empty_queue_report.get("status"),
"validation_environment_proven": bool(
validation_env_proof.get("proven")
),
"validation_environment_violations": list(
validation_env_proof.get("violations") or []
),
"queue_ordering_proven": bool(queue_ordering_proof.get("proven")),
"queue_ordering_policy": queue_ordering_proof.get("policy"),
"queue_ordering_violations": list(
@@ -3336,6 +3462,192 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict:
}
# ---------------------------------------------------------------------------
# Validation environment proof (#311)
# ---------------------------------------------------------------------------
_VALIDATION_CLAIM_RE = re.compile(
r"(?:validation\s+command|pytest\s+(?:executed|passed|failed)|"
r"\d+\s+passed|\d+\s+failed|tests?\s+passed)",
re.IGNORECASE,
)
_VALIDATION_CMD_LINE_RE = re.compile(
r"validation\s+command\s*:\s*(.+)",
re.IGNORECASE,
)
_VALIDATION_CWD_LINE_RE = re.compile(
r"working\s+directory\s*:\s*(\S+)",
re.IGNORECASE,
)
_WHICH_PYTEST_LINE_RE = re.compile(
r"which\s+pytest\s*:\s*(.+)",
re.IGNORECASE,
)
_PYTEST_VERSION_LINE_RE = re.compile(
r"pytest\s+--version\s*:\s*(.+)",
re.IGNORECASE,
)
_BARE_PYTEST_CMD_RE = re.compile(
r"^(?:python\s+-m\s+)?pytest(?:\s|$)",
re.IGNORECASE,
)
_VAGUE_PYTEST_CLAIM_RE = re.compile(
r"pytest\s+(?:executed|passed|failed)\b",
re.IGNORECASE,
)
def _parse_validation_command(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _VALIDATION_CMD_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(".")
return ""
def _parse_validation_cwd(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _VALIDATION_CWD_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(",.;")
return ""
def _is_bare_pytest_command(command: str) -> bool:
text = (command or "").strip()
if not text:
return False
if "/" in text or "\\" in text:
return False
return bool(_BARE_PYTEST_CMD_RE.match(text))
def assess_validation_environment_proof(
*,
report_text: str | None = None,
validation_environment: dict | None = None,
) -> dict:
"""#311: reviewer reports must include exact validation command/environment."""
text = report_text or ""
lower = text.lower()
env = validation_environment or {}
reasons: list[str] = []
violations: list[str] = []
claims_validation = bool(_VALIDATION_CLAIM_RE.search(text))
if not claims_validation and not env.get("command"):
return {
"proven": True,
"block": False,
"claims_validation": False,
"reasons": [],
"violations": [],
"safe_next_action": "proceed",
}
command = (env.get("command") or _parse_validation_command(text)).strip()
working_directory = (
env.get("working_directory") or env.get("cwd")
or _parse_validation_cwd(text)
).strip()
if not command:
reasons.append(
"validation result claimed without exact validation command (#311)"
)
elif _VAGUE_PYTEST_CLAIM_RE.search(text) and command == "":
violations.append(
"vague pytest summary without exact validation command (#311)"
)
if not working_directory:
reasons.append(
"validation result claimed without working directory (#311)"
)
has_result = any(
token in lower
for token in ("passed", "failed", "skipped", "error")
) or any(
env.get(key) is not None
for key in ("passed", "failed", "skipped", "result")
)
if not has_result:
reasons.append(
"validation result claimed without pass/fail result evidence (#311)"
)
if command and _is_bare_pytest_command(command):
which_pytest = (
(env.get("which_pytest") or env.get("executable_path") or "")
.strip()
)
if not which_pytest:
for line in text.splitlines():
match = _WHICH_PYTEST_LINE_RE.search(line)
if match:
which_pytest = match.group(1).strip()
break
if not which_pytest:
reasons.append(
"bare pytest command requires which-pytest/executable proof (#311)"
)
pytest_version = (env.get("pytest_version") or "").strip()
if not pytest_version:
for line in text.splitlines():
match = _PYTEST_VERSION_LINE_RE.search(line)
if match:
pytest_version = match.group(1).strip()
break
if not pytest_version:
reasons.append(
"bare pytest command requires pytest --version proof (#311)"
)
venv_resolved = env.get("venv_resolved")
if venv_resolved is None:
venv_resolved = any(
marker in lower
for marker in (
"project venv",
"venv/bin/pytest",
"resolves to the project venv",
"venv resolved: yes",
)
)
if not venv_resolved and which_pytest and "venv" not in which_pytest:
reasons.append(
"bare pytest command requires proof whether executable "
"resolves to the project venv (#311)"
)
proven = not reasons and not violations
return {
"proven": proven,
"block": bool(violations),
"claims_validation": True,
"command": command or None,
"working_directory": working_directory or None,
"reasons": reasons,
"violations": violations,
"safe_next_action": (
"report exact validation command, working directory, result, and "
"for bare pytest also which-pytest, pytest --version, and venv proof"
if not proven
else "proceed"
),
}
# ---------------------------------------------------------------------------
# Reviewer queue ordering proof (#321)
# ---------------------------------------------------------------------------
@@ -3666,6 +3978,172 @@ def assess_reviewer_baseline_validation_proof(
}
# ---------------------------------------------------------------------------
# Already-landed review gate (#292)
# ---------------------------------------------------------------------------
ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED"
ALREADY_LANDED_HANDOFF_FIELDS = (
"PR number/title",
"candidate head SHA",
"target branch",
"target branch SHA",
"ancestor proof",
"linked issue status",
"recommended reconciliation action",
"capability proof for close/comment mutations",
)
_FORBIDDEN_LANDED_STATES_RE = re.compile(
r"review decision\s*:\s*approved?\b|"
r"merge result\s*:\s*merged\b|"
r"\bready[_ ]to[_ ]merge\b|"
r"\bmerge[d]?\s*:\s*success\b",
re.IGNORECASE,
)
def assess_already_landed_review_gate(
*,
pr_number: int | None,
candidate_head_sha: str | None,
target_branch: str | None,
target_branch_sha: str | None,
head_is_ancestor_of_target: bool | None,
live_head_sha: str | None = None,
real_blocker: bool = False,
mergeable: bool | None = None,
) -> dict:
"""#292: hard gate before any review mutation on an already-landed PR.
Runs after PR selection and head pinning, before approval, request-
changes, or the merge API. *head_is_ancestor_of_target* is the result
of an ancestry check of *candidate_head_sha* against the freshly
fetched target branch at *target_branch_sha* (e.g. ``git merge-base
--is-ancestor``). ``None`` means the check was not run — the gate then
fails closed.
*live_head_sha*, when provided, must equal *candidate_head_sha*; a
changed head invalidates the pinned ancestry result until re-checked.
*mergeable* is accepted for caller convenience but never consulted:
conflict/mergeability handling is a separate gate.
"""
del mergeable # ancestry gate only; mergeability is a separate gate
reasons: list[str] = []
if not isinstance(pr_number, int) or pr_number <= 0:
reasons.append("PR number missing or invalid (#292)")
if not _FULL_SHA.match((candidate_head_sha or "").strip()):
reasons.append(
"candidate head SHA is not a full 40-hex commit SHA (#292)"
)
if not (target_branch or "").strip():
reasons.append("target branch missing (#292)")
if not _FULL_SHA.match((target_branch_sha or "").strip()):
reasons.append(
"target branch SHA is not a full 40-hex commit SHA; fetch the "
"target branch and record its SHA before the ancestry check "
"(#292)"
)
if live_head_sha is not None and candidate_head_sha and (
live_head_sha.strip() != candidate_head_sha.strip()
):
reasons.append(
"PR head changed since the ancestry check was pinned; re-fetch "
"and re-run the already-landed gate (#292)"
)
if head_is_ancestor_of_target is None:
reasons.append(
"ancestry of the PR head against the target branch was not "
"checked; the already-landed gate must run before any review "
"mutation (#292)"
)
if reasons:
return {
"state": "GATE_NOT_PROVEN",
"approve_allowed": False,
"request_changes_allowed": False,
"merge_api_allowed": False,
"reconciliation_handoff_required": False,
"required_handoff_fields": (),
"reasons": reasons,
"safe_next_action": (
"fetch the target branch, pin the live PR head, run the "
"ancestry check, then re-run this gate"
),
}
if head_is_ancestor_of_target:
return {
"state": ALREADY_LANDED_STATE,
"approve_allowed": False,
"request_changes_allowed": bool(real_blocker),
"merge_api_allowed": False,
"reconciliation_handoff_required": True,
"required_handoff_fields": ALREADY_LANDED_HANDOFF_FIELDS,
"reasons": [
f"PR #{pr_number} head {candidate_head_sha} is already an "
f"ancestor of {target_branch} @ {target_branch_sha}; the PR "
"is reconciliation-only (#292)"
],
"safe_next_action": (
"stop before review mutation and emit an "
f"{ALREADY_LANDED_STATE} reconciliation handoff; any PR/issue "
"close or comment requires exact capability proof"
),
}
return {
"state": "NORMAL_REVIEW_CANDIDATE",
"approve_allowed": True,
"request_changes_allowed": True,
"merge_api_allowed": True,
"reconciliation_handoff_required": False,
"required_handoff_fields": (),
"reasons": [],
"safe_next_action": "proceed with the normal review workflow gates",
}
def assess_already_landed_report_state(
report_text: str | None,
*,
gate_fired: bool,
) -> dict:
"""#292: reject APPROVED/MERGED/READY_TO_MERGE after the gate fired.
*gate_fired* comes from the session's own gate result, so a report
that omits the already-landed markers entirely still fails closed —
unlike the text-only #327 wording rules this does not depend on the
report admitting the PR was already landed.
"""
if not gate_fired:
return {"complete": True, "block": False, "reasons": []}
text = report_text or ""
reasons: list[str] = []
if _FORBIDDEN_LANDED_STATES_RE.search(text):
reasons.append(
"report claims APPROVED/MERGED/READY_TO_MERGE although the "
"already-landed gate fired (#292)"
)
if ALREADY_LANDED_STATE.lower() not in text.lower():
reasons.append(
f"report must state {ALREADY_LANDED_STATE} when the "
"already-landed gate fired (#292)"
)
return {
"complete": not reasons,
"block": bool(reasons),
"reasons": reasons,
}
# ---------------------------------------------------------------------------
# Full-suite failure approval gate (#323)
# ---------------------------------------------------------------------------
@@ -4251,6 +4729,13 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
}
def assess_merge_simulation_report(report_text, **kwargs):
"""#317: classify local merge simulation as worktree/index mutations."""
from reviewer_merge_simulation import assess_merge_simulation_report as _assess
return _assess(report_text, **kwargs)
def assess_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
@@ -4265,6 +4750,15 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs):
return _assess(report_text, **kwargs)
def assess_validation_worktree_edit_report(report_text, **kwargs):
"""#315: block edits in PR validation worktrees; require diagnostic separation."""
from reviewer_validation_worktree_edits import (
assess_validation_worktree_edit_report as _assess,
)
return _assess(report_text, **kwargs)
def assess_inventory_worktree_report(report_text, **kwargs):
"""#293: prove inventory pagination and consistent worktree reporting."""
from reviewer_inventory_worktree import assess_inventory_worktree_report as _assess