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 339

Browse Source
This commit is contained in:
sysadmin
2026-07-07 05:18:29 -04:00
parent 6c653ce32a 6bf0210128
commit 240e7adf46
14 changed files with 1761 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+601 -4
View File
@@ -968,6 +968,90 @@ def assess_mutation_ledger_report(
}
def assess_list_prs_pagination_proof(
list_prs_response: dict | list | None,
*,
inventory_complete_claimed: bool = False,
) -> dict:
"""#340: verify ``gitea_list_prs`` pagination metadata proves inventory scope.
Accepts the wrapped ``{"prs": [...], "pagination": {...}}`` response from
``gitea_list_prs``. Legacy bare lists fail closed when completeness is
claimed. Single-page fetches may not claim full inventory unless
``inventory_complete`` or ``is_final_page`` is true.
"""
reasons: list[str] = []
if list_prs_response is None:
reasons.append("gitea_list_prs response missing; fail closed")
return {"proven": False, "block": True, "reasons": reasons, "pagination": None}
if isinstance(list_prs_response, list):
pagination = None
if inventory_complete_claimed:
reasons.append(
"bare PR list lacks pagination metadata; cannot prove inventory "
"completeness"
)
return {
"proven": not reasons,
"block": bool(reasons),
"reasons": reasons,
"pagination": pagination,
}
if not isinstance(list_prs_response, dict):
reasons.append("gitea_list_prs response is not a dict or list; fail closed")
return {"proven": False, "block": True, "reasons": reasons, "pagination": None}
prs = list_prs_response.get("prs")
pagination = list_prs_response.get("pagination")
if not isinstance(prs, list):
reasons.append("gitea_list_prs response missing 'prs' list")
if not isinstance(pagination, dict):
reasons.append("gitea_list_prs response missing 'pagination' metadata")
return {
"proven": False,
"block": True,
"reasons": reasons,
"pagination": pagination if isinstance(pagination, dict) else None,
}
required_keys = ("page", "per_page", "returned_count", "has_more", "is_final_page")
for key in required_keys:
if key not in pagination:
reasons.append(f"pagination metadata missing '{key}'")
if inventory_complete_claimed:
complete = pagination.get("inventory_complete") is True
final_page = pagination.get("is_final_page") is True and pagination.get("has_more") is False
if not (complete or final_page):
reasons.append(
"inventory completeness claimed but pagination metadata does not "
"prove final page (inventory_complete or is_final_page with "
"has_more=false required)"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"pagination": pagination,
}
def _prs_from_list_response(list_prs_response: list | dict | None) -> list | None:
"""Normalize ``gitea_list_prs`` output to a PR list."""
if list_prs_response is None:
return None
if isinstance(list_prs_response, list):
return list_prs_response
if isinstance(list_prs_response, dict):
prs = list_prs_response.get("prs")
return prs if isinstance(prs, list) else None
return None
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
justification=None):
"""Assess reviewer/author role separation for blind queue workflows.
@@ -1147,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
@@ -1184,6 +1269,18 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
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,
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"))
@@ -1323,6 +1420,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"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)"
)
downgrade_reasons.extend(baseline_validation.get("reasons", []))
merge_allowed = (
identity_eligible
@@ -1334,9 +1436,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 "
@@ -1394,6 +1498,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"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 []
),
}
@@ -1597,8 +1705,55 @@ HANDOFF_ROLE_FIELDS = {
)),
("Related issues", ("related issues",)),
),
"create_issue": (
("Active profile", ("active profile",)),
("Runtime context", ("runtime context",)),
("Requested issue task", ("requested issue task",)),
("Workflow source", ("workflow source",)),
("Capability proof", ("capability proof",)),
("Duplicate search terms", ("duplicate search terms",)),
("Duplicate search pagination proof", (
"duplicate search pagination proof",
)),
("Duplicates found", ("duplicates found",)),
("Issues created", ("issues created",)),
("Issues commented", ("issues commented",)),
("Issues edited", ("issues edited",)),
("Issues skipped as duplicates", (
"issues skipped as duplicates",
)),
("File edits by issue creator", ("file edits by issue creator",)),
("Safety statement", ("safety statement",)),
),
}
CREATE_ISSUE_WORKFLOW_MARKERS = (
"workflows/create-issue.md",
"create-issue.md",
)
CREATE_ISSUE_TASK_MARKERS = (
"create-issue",
"create issue",
)
CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = (
("Selected issue", ("selected issue",)),
("Branch name", ("branch name",)),
("PR number", ("pr number",)),
("PR URL", ("pr url",)),
("Commit SHA", ("commit sha",)),
("Push result", ("push result",)),
("Baseline comparison", ("baseline comparison",)),
("Selected PR", ("selected pr",)),
("Review decision", ("review decision",)),
("Merge result", ("merge result",)),
("Merge preflight", ("merge preflight",)),
("Already-landed gate", ("already-landed gate",)),
("Pinned reviewed head", ("pinned reviewed head",)),
("Terminal review mutation", ("terminal review mutation",)),
)
def _handoff_section_lines(report_text):
"""Return the lines of the Controller Handoff section, or None."""
@@ -1649,6 +1804,11 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
fields_dict[label] = v.strip()
required = list(HANDOFF_BASE_FIELDS)
if role == "create_issue":
required = [
field for field in required
if field[0] not in ("Workspace mutations", "Mutations")
]
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
missing = []
@@ -1836,14 +1996,19 @@ def pr_inventory_trust_gate(
"queue_target_lock": lock_status,
}
if list_prs_response is None or not isinstance(list_prs_response, list):
prs_list = _prs_from_list_response(list_prs_response)
pagination_meta = (
list_prs_response.get("pagination")
if isinstance(list_prs_response, dict) else {}
) or {}
if prs_list is None:
return {
"status": "inventory_error",
"reasons": ["PR list response is invalid (not a list or None)"],
"reasons": ["PR list response is invalid (missing prs list or None)"],
"corroborated": False,
}
if len(list_prs_response) > 0:
if len(prs_list) > 0:
return {
"status": "trusted_nonempty",
"reasons": [],
@@ -1874,6 +2039,13 @@ def pr_inventory_trust_gate(
corroborated = False
if has_finality_metadata:
corroborated = True
elif pagination_meta.get("inventory_complete") is True:
corroborated = True
elif (
pagination_meta.get("is_final_page") is True
and pagination_meta.get("has_more") is False
):
corroborated = True
elif corroboration_open_pr_counter == 0:
corroborated = True
else:
@@ -2673,6 +2845,141 @@ def assess_issue_filing_duplicate_summary(
}
def assess_create_issue_workflow_source(report_text: str) -> dict:
"""#337: create-issue reports must cite the canonical workflow file."""
lower = (report_text or "").lower()
reasons = []
if not any(marker in lower for marker in CREATE_ISSUE_WORKFLOW_MARKERS):
reasons.append(
"create-issue report missing workflow source "
"(workflows/create-issue.md)"
)
if not any(marker in lower for marker in CREATE_ISSUE_TASK_MARKERS):
reasons.append(
"create-issue report missing task mode declaration "
"(create-issue)"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_create_issue_mode_isolation(report_text: str) -> dict:
"""#337: reject work-issue/review-merge handoff fields in create-issue runs."""
section = _handoff_section_lines(report_text)
reasons = []
if section is None:
return {
"complete": True,
"downgraded": False,
"reasons": [],
}
labels = []
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
labels.append(stripped.split(":", 1)[0].strip().lower())
for name, aliases in CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS:
if any(
label.startswith(alias)
for label in labels
for alias in aliases
):
reasons.append(
f"create-issue handoff must not include cross-mode field "
f"'{name}'"
)
legacy_workspace = any(
label.startswith("workspace mutations") for label in labels
)
precise_categories = any(
label.startswith("file edits by issue creator")
for label in labels
)
if legacy_workspace and not precise_categories:
reasons.append(
"create-issue handoff must use precise mutation categories "
"instead of legacy 'Workspace mutations'"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_create_issue_final_report(
report_text: str,
*,
issue_number: int | None = None,
issue_title: str | None = None,
action: str = "created",
mutations: list[str] | None = None,
resolved_capabilities: dict | None = None,
issues_searched: int | None = None,
closest_matches: list[dict] | None = None,
duplicate_result: dict | None = None,
performed_mutations: list[str] | None = None,
) -> dict:
"""#337: composite verifier for create-issue final reports."""
checks = {
"workflow_source": assess_create_issue_workflow_source(report_text),
"mode_isolation": assess_create_issue_mode_isolation(report_text),
"controller_handoff": assess_controller_handoff(
report_text, role="create_issue"
),
"issue_reference": assess_issue_filing_issue_reference(
report_text,
issue_number=issue_number,
issue_title=issue_title,
action=action,
),
"mutation_capability": assess_issue_filing_mutation_capability(
report_text,
mutations or performed_mutations,
resolved_capabilities,
),
"mutation_scope": assess_issue_filing_mutation_scope(
report_text,
performed_mutations=performed_mutations or mutations,
),
"duplicate_summary": assess_issue_filing_duplicate_summary(
report_text,
issues_searched=issues_searched,
closest_matches=closest_matches,
duplicate_result=duplicate_result,
),
}
reasons = []
downgraded = False
for name, result in checks.items():
verdict = result.get("verdict")
if verdict in ("missing", "incomplete"):
downgraded = True
reasons.extend(result.get("reasons") or [])
elif result.get("downgraded") or not result.get("complete", True):
downgraded = True
reasons.extend(
f"{name}: {r}" for r in (result.get("reasons") or [])
)
grade = "A" if not downgraded else "downgraded"
return {
"grade": grade,
"downgraded": downgraded,
"checks": checks,
"reasons": reasons,
"complete": not downgraded,
}
def assess_issue_filing_final_report(
report_text: str,
*,
@@ -2763,6 +3070,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)
# ---------------------------------------------------------------------------
@@ -3093,3 +3574,119 @@ def assess_queue_status_report(
"violations": deduped,
"reasons": deduped,
}
# ---------------------------------------------------------------------------
# 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,
}