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 issue 324

Browse Source
This commit is contained in:
sysadmin
2026-07-07 05:19:41 -04:00
parent 24788d3b4c deb228d629
commit 38b35534d4
9 changed files with 1148 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+322 -1
View File
@@ -1231,7 +1231,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
role_boundary=None, review_mutation=None,
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None, worktree_proof=None):
sweep_proof=None, worktree_proof=None,
baseline_validation=None, project_root=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -1258,6 +1259,18 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
)
empty_queue_report = assess_empty_queue_report(report_text)
if baseline_validation is None and report_text:
baseline_validation = assess_reviewer_baseline_validation_proof(
report_text=report_text,
project_root=project_root,
)
elif baseline_validation is None:
baseline_validation = {
"proven": True,
"block": False,
"reasons": [],
"violations": [],
}
contamination_status = contamination.get("status", "unknown")
checkout_proven = bool(checkout_proof.get("proven"))
@@ -1387,6 +1400,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 baseline_validation.get("proven"):
downgrade_reasons.append(
"reviewer baseline validation proof missing or failed (#325)"
)
downgrade_reasons.extend(baseline_validation.get("reasons", []))
merge_allowed = (
identity_eligible
@@ -1398,9 +1416,11 @@ 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 baseline_validation.get("proven")
)
violations = []
violations.extend(baseline_validation.get("violations", []))
if merge_performed and not merge_allowed:
violations.append(
"merge was performed/claimed although the proofs did not allow "
@@ -1452,6 +1472,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
else True
),
"empty_queue_trust_gate_status": empty_queue_report.get("status"),
"baseline_validation_proven": bool(baseline_validation.get("proven")),
"baseline_validation_violations": list(
baseline_validation.get("violations") or []
),
}
@@ -3020,6 +3044,180 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict:
}
# ---------------------------------------------------------------------------
# Reviewer baseline validation (#325)
# ---------------------------------------------------------------------------
_TEST_VALIDATION_COMMAND_RE = re.compile(
r"(?:^|\s)(?:venv/)?(?:bin/)?(?:python\s+-m\s+)?"
r"(?:pytest|make\s+test|npm\s+test|cargo\s+test|go\s+test\b)",
re.IGNORECASE,
)
_PREEXISTING_FAILURE_CLAIM_RE = re.compile(
r"(?:\bsame\s+as\s+master\b|"
r"\bpre-?existing(?:\s+(?:on\s+)?master)?\b|"
r"\bfailures?\s+(?:are|is)\s+pre-?existing\b|"
r"\bmaster\s+also\s+fails?\b|"
r"\bfull-?suite\s+failures?\s+(?:are|is)\s+pre-?existing\b)",
re.IGNORECASE,
)
_REPORT_VALIDATION_CWD_RE = re.compile(
r"(?:working\s+directory|cwd|directory)\s*:\s*(\S+)",
re.IGNORECASE,
)
def _normalize_path(path: str) -> str:
return (path or "").replace("\\", "/").rstrip("/")
def _path_under_branches(path: str, project_root: str | None = None) -> bool:
"""True when *path* is inside the project's ``branches/`` directory."""
normalized = _normalize_path(path)
if not normalized:
return False
if "/branches/" in f"{normalized}/":
return True
if normalized.endswith("/branches"):
return True
if project_root:
root = _normalize_path(project_root)
if normalized.startswith(f"{root}/"):
rel = normalized[len(root) + 1 :]
return rel == "branches" or rel.startswith("branches/")
return False
def _is_test_validation_command(command: str) -> bool:
return bool(_TEST_VALIDATION_COMMAND_RE.search((command or "").strip()))
def _is_full_sha(value: str | None) -> bool:
return bool(value and _FULL_SHA.match((value or "").strip()))
def assess_reviewer_baseline_validation_proof(
*,
validation_runs: list[dict] | None = None,
baseline_proof: dict | None = None,
report_text: str | None = None,
project_root: str | None = None,
) -> dict:
"""#325: reviewer validation and baseline comparison must use ``branches/``.
*validation_runs* entries: ``command``, ``working_directory`` (or ``cwd``),
optional ``project_root``.
*baseline_proof* keys when claiming pre-existing master failures:
``worktree_path``, ``baseline_target_sha``, ``pr_head_sha``,
``baseline_failures``, ``pr_failures``, ``failure_signatures_match``,
``clean_before``, ``clean_after``.
"""
reasons: list[str] = []
violations: list[str] = []
text = report_text or ""
runs = list(validation_runs or [])
pending_command = None
for raw_line in text.splitlines():
line = raw_line.strip().lstrip("-*").strip()
lower = line.lower()
if lower.startswith("validation command:"):
pending_command = line.split(":", 1)[1].strip()
continue
cwd_match = _REPORT_VALIDATION_CWD_RE.search(line)
if cwd_match:
cwd = cwd_match.group(1).strip().rstrip(",.;")
command = pending_command or line
if _is_test_validation_command(command):
runs.append(
{
"command": command,
"working_directory": cwd,
"project_root": project_root,
}
)
pending_command = None
for run in runs:
command = (run.get("command") or "").strip()
if not command or not _is_test_validation_command(command):
continue
cwd = (run.get("working_directory") or run.get("cwd") or "").strip()
root = run.get("project_root") or project_root
if not cwd:
reasons.append(
"test validation command stated without working directory; "
"cannot prove branches-only execution (#325)"
)
continue
if not _path_under_branches(cwd, root):
violations.append(
f"test validation ran outside branches/ worktree: cwd={cwd!r}"
)
reasons.append(
"reviewer workflow must not run test suites in the main "
f"checkout; cwd {cwd!r} is not under branches/ (#325)"
)
claims_preexisting = bool(_PREEXISTING_FAILURE_CLAIM_RE.search(text))
if claims_preexisting:
proof = baseline_proof or {}
worktree = (proof.get("worktree_path") or "").strip()
if not worktree or not _path_under_branches(worktree, project_root):
reasons.append(
"pre-existing master failure claimed without a baseline "
"worktree path under branches/ (#325)"
)
if not _is_full_sha(proof.get("baseline_target_sha")):
reasons.append(
"pre-existing master failure claimed without "
"baseline_target_sha proof (#325)"
)
if not _is_full_sha(proof.get("pr_head_sha")):
reasons.append(
"pre-existing master failure claimed without pr_head_sha "
"proof (#325)"
)
baseline_failures = proof.get("baseline_failures")
pr_failures = proof.get("pr_failures")
if baseline_failures is None or pr_failures is None:
reasons.append(
"pre-existing master failure claimed without baseline and "
"PR failure listings (#325)"
)
if proof.get("failure_signatures_match") is not True:
reasons.append(
"pre-existing master failure claimed without proven matching "
"failure signatures (#325)"
)
if proof.get("clean_before") is not True:
reasons.append(
"baseline worktree clean-before validation not proven (#325)"
)
if proof.get("clean_after") is not True:
reasons.append(
"baseline worktree clean-after validation not proven (#325)"
)
proven = not reasons and not violations
return {
"proven": proven,
"block": bool(violations),
"claims_preexisting": claims_preexisting,
"reasons": reasons,
"violations": violations,
"safe_next_action": (
"create a clean baseline worktree under branches/, e.g. "
"branches/baseline-master-pr<N>, and rerun validation there"
if not proven
else "proceed"
),
}
# ---------------------------------------------------------------------------
# Identity disclosure (#305)
# ---------------------------------------------------------------------------
@@ -3106,3 +3304,126 @@ def assess_reviewer_fallback_report(report_text, **kwargs):
from reviewer_fallback import assess_reviewer_fallback_report as _assess
return _assess(report_text, **kwargs)
# ---------------------------------------------------------------------------
# Git ref mutations (#297)
# ---------------------------------------------------------------------------
_GIT_REF_MUTATIONS_FIELD_RE = re.compile(
r"^\s*git ref mutations\s*:\s*(.+?)\s*$", re.I | re.M)
_GIT_WORKTREE_MUTATIONS_NONE_RE = re.compile(
r"^\s*git/worktree mutations\s*:\s*none\b", re.I | re.M)
_GIT_REF_MUTATING_FIRST_WORDS = frozenset(
{"fetch", "pull", "push", "merge", "update-ref"})
_GIT_REF_MUTATING_PREFIXES = ("remote update", "branch -d", "branch -D",
"reset --hard")
def _command_text(entry):
if isinstance(entry, dict):
return str(entry.get("command") or "").strip()
return str(entry or "").strip()
def _git_subcommand_line(command):
tokens = command.split()
if not tokens or tokens[0] != "git":
return None
return " ".join(tokens[1:])
def git_ref_mutating_commands(command_log):
"""Return logged commands that update git refs (#297).
``git fetch`` and friends edit no files but move refs; they are never
read-only diagnostics.
"""
out = []
for entry in command_log or []:
command = _command_text(entry)
rest = _git_subcommand_line(command)
if rest is None:
continue
first = rest.split()[0] if rest.split() else ""
if first in _GIT_REF_MUTATING_FIRST_WORDS:
out.append(command)
continue
if any(rest.startswith(prefix) for prefix in _GIT_REF_MUTATING_PREFIXES):
out.append(command)
return out
def _fetch_targets(ref_commands):
"""Extract ``<remote>/<branch>`` targets from git fetch commands."""
targets = []
for command in ref_commands:
rest = _git_subcommand_line(command) or ""
tokens = rest.split()
if not tokens or tokens[0] != "fetch":
continue
args = [t for t in tokens[1:] if not t.startswith("-")]
if len(args) >= 2:
targets.append(f"{args[0]}/{args[1]}")
return targets
def assess_git_ref_mutation_report(report_text, *, command_log=None):
"""#297: ref updates are Git ref mutations, not read-only diagnostics.
When the command log holds ref-updating commands, the report must carry
a non-none ``Git ref mutations`` entry (naming ``fetched
<remote>/<branch>`` for targeted fetches), may not claim ``Git/worktree
mutations: None``, and may not list the command as read-only.
"""
text = report_text or ""
lower = text.lower()
reasons = []
ref_commands = git_ref_mutating_commands(command_log)
fetch_targets = _fetch_targets(ref_commands)
if ref_commands:
field = _GIT_REF_MUTATIONS_FIELD_RE.search(text)
value = field.group(1).strip().lower() if field else None
if not value or value == "none":
reasons.append(
"ref-updating commands ran but the report has no "
"'Git ref mutations' entry"
)
for target in fetch_targets:
if "fetched" not in lower or target.lower() not in lower:
reasons.append(
"git fetch ran; report must state "
f"'Git ref mutations: fetched {target}'"
)
if _GIT_WORKTREE_MUTATIONS_NONE_RE.search(text):
reasons.append(
"'Git/worktree mutations: None' rejected: ref-updating "
"commands ran"
)
for line in text.splitlines():
if "read-only" not in line.lower():
continue
for command in ref_commands:
if command.lower() in line.lower():
reasons.append(
"read-only diagnostics may not include ref-updating "
f"command '{command}'"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"ref_mutating_commands": ref_commands,
"fetch_targets": fetch_targets,
}
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)