Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
db032ef93c | ||
|
|
1a1e679246 | ||
|
|
9fde4f3e76 | ||
|
|
6b5d2ad080 | ||
|
|
af7131abf1 | ||
|
|
71ccbca10f | ||
|
|
6220304f8c | ||
|
|
c1d85b621a |
@@ -457,6 +457,40 @@ def _rule_reviewer_git_fetch_readonly(
|
||||
return []
|
||||
|
||||
|
||||
def _rule_reviewer_validation_failure_history(
|
||||
report_text: str,
|
||||
*,
|
||||
validation_session: dict | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
from reviewer_validation_failure_history import (
|
||||
assess_validation_failure_history_report,
|
||||
)
|
||||
|
||||
session = validation_session or {}
|
||||
observed = session.get("observed_failures") or []
|
||||
if not observed:
|
||||
return []
|
||||
|
||||
result = assess_validation_failure_history_report(
|
||||
report_text,
|
||||
validation_session=session,
|
||||
)
|
||||
if result.get("proven"):
|
||||
return []
|
||||
severity = "block" if result.get("violations") else "downgrade"
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.validation_failure_history",
|
||||
severity,
|
||||
"Validation failure history",
|
||||
reason,
|
||||
result.get("safe_next_action")
|
||||
or "document every observed validation failure before claiming pass",
|
||||
)
|
||||
for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not _BARE_PYTEST_RE.search(text):
|
||||
@@ -893,6 +927,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
_rule_reviewer_mutation_categories,
|
||||
_rule_reviewer_git_fetch_readonly,
|
||||
_rule_reviewer_validation_command,
|
||||
_rule_reviewer_validation_failure_history,
|
||||
_rule_reviewer_validation_structured,
|
||||
_rule_reviewer_linked_issue,
|
||||
_rule_reviewer_baseline_on_failure,
|
||||
@@ -1000,6 +1035,7 @@ def assess_final_report_validator(
|
||||
local_edits: bool = False,
|
||||
issue_filing_lock: dict | None = None,
|
||||
session_pr_opened: bool = False,
|
||||
validation_session: dict | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate final-report text against task-specific proof rules (#327).
|
||||
|
||||
@@ -1055,6 +1091,7 @@ def assess_final_report_validator(
|
||||
"mutations_observed": mutations_observed,
|
||||
"local_edits": local_edits,
|
||||
"session_pr_opened": session_pr_opened,
|
||||
"validation_session": validation_session,
|
||||
}
|
||||
|
||||
for rule in _RULES_BY_TASK.get(normalized_kind, ()):
|
||||
|
||||
+21
-3
@@ -1245,7 +1245,7 @@ def gitea_create_pr(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
|
||||
# ── Issue Lock Validation (Issue #194 / #196) ──
|
||||
@@ -3464,16 +3464,32 @@ def gitea_delete_branch(
|
||||
repo: Override the repository name.
|
||||
|
||||
Returns:
|
||||
dict with 'success' and 'message'.
|
||||
dict with 'success' and 'message'; on a permission block,
|
||||
'success'/'performed' False, 'reasons', and a structured
|
||||
'permission_report' (#142) with no API call made.
|
||||
"""
|
||||
gate_reasons = _profile_operation_gate("gitea.branch.delete")
|
||||
if gate_reasons:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"reasons": gate_reasons,
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
import urllib.parse
|
||||
encoded_branch = urllib.parse.quote(branch, safe="")
|
||||
url = f"{repo_api_url(h, o, r)}/branches/{encoded_branch}"
|
||||
request_metadata = {
|
||||
"branch": branch,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
}
|
||||
with _audited("delete_branch", host=h, remote=remote, org=o, repo=r,
|
||||
target_branch=branch, request_metadata={"branch": branch}):
|
||||
target_branch=branch, request_metadata=request_metadata):
|
||||
api_request("DELETE", url, auth)
|
||||
return {"success": True, "message": f"Remote branch '{branch}' deleted."}
|
||||
|
||||
@@ -5510,6 +5526,7 @@ def gitea_validate_review_final_report(
|
||||
action_log: list[dict] | None = None,
|
||||
mutations_observed: bool = False,
|
||||
local_edits: bool = False,
|
||||
validation_session: dict | None = None,
|
||||
) -> dict:
|
||||
"""Read-only: machine-validate reviewer final report schema before output (#391).
|
||||
|
||||
@@ -5527,6 +5544,7 @@ def gitea_validate_review_final_report(
|
||||
action_log=action_log,
|
||||
mutations_observed=mutations_observed,
|
||||
local_edits=local_edits,
|
||||
validation_session=validation_session,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -301,6 +301,7 @@ def assess_review_final_report_schema(
|
||||
action_log: list[dict] | None = None,
|
||||
mutations_observed: bool = False,
|
||||
local_edits: bool = False,
|
||||
validation_session: dict | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate reviewer final report text before session completion (#391)."""
|
||||
base = assess_final_report_validator(
|
||||
@@ -312,6 +313,7 @@ def assess_review_final_report_schema(
|
||||
action_log=action_log,
|
||||
mutations_observed=mutations_observed,
|
||||
local_edits=local_edits,
|
||||
validation_session=validation_session,
|
||||
)
|
||||
extra: list[dict[str, str]] = []
|
||||
for rule in _SCHEMA_RULES:
|
||||
|
||||
@@ -1735,6 +1735,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
if report_text
|
||||
else {"proven": True, "block": False, "reasons": []}
|
||||
)
|
||||
proof_backed_handoff = (
|
||||
assess_proof_backed_handoff_report(report_text)
|
||||
if report_text and _PROOF_BACKED_HANDOFF_HINT.search(report_text)
|
||||
else {"proven": True, "block": False, "reasons": []}
|
||||
)
|
||||
if baseline_validation is None and report_text:
|
||||
baseline_validation = assess_reviewer_baseline_validation_proof(
|
||||
report_text=report_text,
|
||||
@@ -1937,6 +1942,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
"reviewer baseline validation proof missing or failed (#325)"
|
||||
)
|
||||
downgrade_reasons.extend(baseline_validation.get("reasons", []))
|
||||
if not proof_backed_handoff.get("proven"):
|
||||
downgrade_reasons.append(
|
||||
"proof-sensitive handoff claims lack command/tool evidence (#395)"
|
||||
)
|
||||
downgrade_reasons.extend(proof_backed_handoff.get("reasons", []))
|
||||
if not already_landed_classification.get("proven"):
|
||||
downgrade_reasons.append(
|
||||
"already-landed classification wording missing or invalid (#295)"
|
||||
@@ -5088,6 +5098,12 @@ _PR_QUEUE_CLEANUP_REPORT_HINT = re.compile(
|
||||
re.I,
|
||||
)
|
||||
|
||||
_PROOF_BACKED_HANDOFF_HINT = re.compile(
|
||||
r"task:\s*review-merge-pr|task mode:\s*review-merge-pr|"
|
||||
r"review-merge-pr|controller handoff",
|
||||
re.I,
|
||||
)
|
||||
|
||||
|
||||
def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
|
||||
"""#390: validate PR-only cleanup final reports (one PR per run)."""
|
||||
@@ -5487,6 +5503,15 @@ def assess_validation_integrity_report(report_text, **kwargs):
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
|
||||
def assess_validation_failure_history_report(report_text, **kwargs):
|
||||
"""#396: final reports must account for transient validation failures."""
|
||||
from reviewer_validation_failure_history import (
|
||||
assess_validation_failure_history_report as _assess,
|
||||
)
|
||||
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
|
||||
def assess_already_landed_classification_report(report_text, **kwargs):
|
||||
"""#295: already-landed PRs are reconciliation-only, not review eligible."""
|
||||
from reviewer_already_landed_classification import (
|
||||
@@ -5566,3 +5591,10 @@ def assess_worktree_ownership_report(report_text, **kwargs):
|
||||
from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
|
||||
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
|
||||
def assess_proof_backed_handoff_report(report_text, **kwargs):
|
||||
"""#395: proof-sensitive review handoff claims require explicit evidence."""
|
||||
from reviewer_proof_backed_handoff import assess_proof_backed_handoff_report as _assess
|
||||
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
@@ -0,0 +1,217 @@
|
||||
"""Proof-backed reviewer handoff claim verifier (#395)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
_PAGINATION_FINALITY = re.compile(
|
||||
r"has_more\s*:\s*false|is_final_page\s*:\s*true|"
|
||||
r"inventory_complete\s*:\s*true|pages_fetched|total_count\s*:",
|
||||
re.I,
|
||||
)
|
||||
_PAGE_SIZE_ONLY = re.compile(
|
||||
r"(?:less|fewer) than (?:the )?(?:default )?page[- ]?size|"
|
||||
r"under (?:the )?50[- ]?(?:item )?limit|"
|
||||
r"returned \d+ (?:open )?prs?.*less than 50",
|
||||
re.I,
|
||||
)
|
||||
_INVENTORY_COMPLETE_CLAIM = re.compile(
|
||||
r"\b(?:inventory (?:is )?complete|inventory exhaustive|"
|
||||
r"complete (?:pr )?inventory)\b",
|
||||
re.I,
|
||||
)
|
||||
_SKIP_CLAIM = re.compile(
|
||||
r"\b(?:earlier prs? skipped|skipped (?:earlier )?pr|"
|
||||
r"skip(?:ped)? pr #?\d+|non-mergeable|prior request.changes)\b",
|
||||
re.I,
|
||||
)
|
||||
_CONFLICT_PROOF = re.compile(
|
||||
r"merge simulation|git merge --no-commit|conflicting files|"
|
||||
r"conflict proof|merge_exit|non-mergeable",
|
||||
re.I,
|
||||
)
|
||||
_BASELINE_CLAIM = re.compile(
|
||||
r"\b(?:baseline (?:validation|worktree|comparison)|"
|
||||
r"same as master|pre-existing (?:on )?master|"
|
||||
r"failure signatures match)\b",
|
||||
re.I,
|
||||
)
|
||||
_BASELINE_PATH = re.compile(r"baseline worktree path\s*:\s*(\S+)", re.I)
|
||||
_BASELINE_SHA = re.compile(r"baseline (?:target )?sha\s*:\s*([0-9a-f]{7,40})", re.I)
|
||||
_BASELINE_DIRTY_BEFORE = re.compile(
|
||||
r"baseline.*dirty before|dirty before.*baseline", re.I
|
||||
)
|
||||
_BASELINE_DIRTY_AFTER = re.compile(
|
||||
r"baseline.*dirty after|dirty after.*baseline", re.I
|
||||
)
|
||||
_BASELINE_COMMAND = re.compile(
|
||||
r"baseline.*(?:validation )?command|pytest.*baseline", re.I
|
||||
)
|
||||
_BASELINE_RESULT = re.compile(
|
||||
r"baseline.*(?:validation )?result|baseline_exit|baseline failures", re.I
|
||||
)
|
||||
_MASTER_INTEGRATION = re.compile(
|
||||
r"\b(?:merged master into|merge(?:d)? (?:remote[- ]tracking )?branch.*master|"
|
||||
r"master integration|integrated master|rebase.*master)\b",
|
||||
re.I,
|
||||
)
|
||||
_CLEANUP_CLAIM = re.compile(
|
||||
r"\b(?:worktree(?:s)? (?:were )?cleaned|cleanup (?:result|mutations)|"
|
||||
r"removed (?:session[- ]owned )?worktree|git worktree remove)\b",
|
||||
re.I,
|
||||
)
|
||||
_WORKTREE_LIST = re.compile(r"git worktree list|worktree list proof", re.I)
|
||||
_PROOF_SOURCE = re.compile(
|
||||
r"proof source\s*:\s*(command|mcp metadata|prior blocker|not checked)",
|
||||
re.I,
|
||||
)
|
||||
_HANDOFF_SECTION = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)
|
||||
|
||||
|
||||
def _handoff_fields(report_text: str) -> dict[str, str]:
|
||||
text = report_text or ""
|
||||
match = _HANDOFF_SECTION.search(text)
|
||||
if not match:
|
||||
return {}
|
||||
fields: dict[str, str] = {}
|
||||
for line in text[match.end() :].splitlines():
|
||||
stripped = line.strip().lstrip("-*").strip()
|
||||
if ":" not in stripped:
|
||||
continue
|
||||
key, value = stripped.split(":", 1)
|
||||
fields[key.strip().lower()] = value.strip()
|
||||
return fields
|
||||
|
||||
|
||||
def _command_text(entry: Any) -> str:
|
||||
if isinstance(entry, dict):
|
||||
return str(entry.get("command") or entry.get("tool") or "").strip()
|
||||
return str(entry or "").strip()
|
||||
|
||||
|
||||
def _action_log_has(action_log: list | None, pattern: re.Pattern[str]) -> bool:
|
||||
for entry in action_log or []:
|
||||
blob = " ".join(
|
||||
filter(
|
||||
None,
|
||||
[
|
||||
_command_text(entry),
|
||||
str(entry.get("result") or "") if isinstance(entry, dict) else "",
|
||||
str(entry.get("reason") or "") if isinstance(entry, dict) else "",
|
||||
],
|
||||
)
|
||||
)
|
||||
if pattern.search(blob):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def assess_proof_backed_handoff_report(
|
||||
report_text: str,
|
||||
*,
|
||||
action_log: list | None = None,
|
||||
inventory_session: dict | None = None,
|
||||
baseline_session: dict | None = None,
|
||||
skip_session: list[dict] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Require explicit command/tool evidence for proof-sensitive review claims (#395)."""
|
||||
text = report_text or ""
|
||||
fields = _handoff_fields(text)
|
||||
reasons: list[str] = []
|
||||
inventory = dict(inventory_session or {})
|
||||
baseline = dict(baseline_session or {})
|
||||
skips = list(skip_session or [])
|
||||
|
||||
pagination_field = fields.get("inventory pagination proof", "")
|
||||
if _INVENTORY_COMPLETE_CLAIM.search(text) or _PAGE_SIZE_ONLY.search(text):
|
||||
has_meta = bool(
|
||||
inventory.get("inventory_complete")
|
||||
or inventory.get("pagination_complete")
|
||||
or _PAGINATION_FINALITY.search(pagination_field)
|
||||
or _PAGINATION_FINALITY.search(text)
|
||||
or _action_log_has(action_log, re.compile(r"gitea_list_prs", re.I))
|
||||
)
|
||||
if _PAGE_SIZE_ONLY.search(text) and not has_meta:
|
||||
reasons.append(
|
||||
"inventory completeness claimed from page-size assumption only; "
|
||||
"require has_more=false, is_final_page=true, or inventory_complete=true"
|
||||
)
|
||||
elif _INVENTORY_COMPLETE_CLAIM.search(text) and not has_meta:
|
||||
reasons.append(
|
||||
"inventory complete claim lacks explicit pagination metadata proof"
|
||||
)
|
||||
|
||||
if _SKIP_CLAIM.search(text) or fields.get("earlier prs skipped", "").lower() not in {
|
||||
"",
|
||||
"none",
|
||||
"n/a",
|
||||
}:
|
||||
skip_proof = bool(
|
||||
skips
|
||||
or _CONFLICT_PROOF.search(text)
|
||||
or _action_log_has(
|
||||
action_log, re.compile(r"git merge --no-commit|gitea_get_pr_review_feedback", re.I)
|
||||
)
|
||||
)
|
||||
if not skip_proof:
|
||||
reasons.append(
|
||||
"earlier PR skip claim lacks command/tool conflict or blocker proof"
|
||||
)
|
||||
|
||||
if _BASELINE_CLAIM.search(text) or fields.get("baseline worktree used", "").lower() == "true":
|
||||
baseline_ok = bool(
|
||||
baseline.get("complete")
|
||||
or (
|
||||
_BASELINE_PATH.search(text)
|
||||
and _BASELINE_SHA.search(text)
|
||||
and (_BASELINE_DIRTY_BEFORE.search(text) or baseline.get("clean_before"))
|
||||
and (_BASELINE_COMMAND.search(text) or baseline.get("command"))
|
||||
and (_BASELINE_RESULT.search(text) or baseline.get("result"))
|
||||
and (_BASELINE_DIRTY_AFTER.search(text) or baseline.get("clean_after"))
|
||||
)
|
||||
)
|
||||
if not baseline_ok:
|
||||
reasons.append(
|
||||
"baseline validation claim missing worktree path, target SHA, "
|
||||
"dirty-before/after status, command, or result proof"
|
||||
)
|
||||
|
||||
if _MASTER_INTEGRATION.search(text):
|
||||
if not _action_log_has(
|
||||
action_log,
|
||||
re.compile(r"git merge.*master|git rebase.*master", re.I),
|
||||
) and not re.search(r"merge_exit\s*=\s*\d+|merge simulation", text, re.I):
|
||||
reasons.append(
|
||||
"master integration claim lacks exact merge/rebase command and result"
|
||||
)
|
||||
|
||||
if _CLEANUP_CLAIM.search(text) or fields.get("cleanup mutations", "").lower() not in {
|
||||
"",
|
||||
"none",
|
||||
"n/a",
|
||||
}:
|
||||
if not (_WORKTREE_LIST.search(text) or _action_log_has(action_log, _WORKTREE_LIST)):
|
||||
reasons.append(
|
||||
"cleanup claim lacks final git worktree list or equivalent proof"
|
||||
)
|
||||
|
||||
if re.search(r"\blive proof\b", text, re.I) and not _PROOF_SOURCE.search(text):
|
||||
if not action_log:
|
||||
reasons.append(
|
||||
"live proof wording requires proof source classification "
|
||||
"(command, MCP metadata, prior blocker, or not checked)"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"cite explicit command/tool evidence or structured MCP pagination metadata "
|
||||
"for each proof-sensitive claim"
|
||||
if not proven
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
@@ -0,0 +1,198 @@
|
||||
"""Transient validation failure history verifier (#396).
|
||||
|
||||
Reviewer sessions may observe validation failures that later pass on rerun.
|
||||
Final reports must document every failure observed during the session, not
|
||||
only the last passing result.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
_SECTION_RE = re.compile(
|
||||
r"validation failure history",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_FAILURE_ENTRY_RE = re.compile(
|
||||
r"(?:failure\s*(?:#|entry)?\s*\d+|validation failure)\s*:",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_COMMAND_RE = re.compile(
|
||||
r"(?:command|validation command)\s*:\s*(.+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_FAILING_TEST_RE = re.compile(
|
||||
r"(?:failing test|failure|error)\s*:\s*(.+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CAUSE_RE = re.compile(
|
||||
r"(?:suspected cause|cause)\s*:\s*(.+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_REPRODUCED_RE = re.compile(
|
||||
r"(?:reproduced|reproduces)\s*:\s*(yes|no|unknown)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_BASELINE_RE = re.compile(
|
||||
r"(?:on baseline master|baseline master|exists on baseline)\s*:\s*(yes|no|unknown|not checked)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_PR_CAUSED_RE = re.compile(
|
||||
r"(?:pr[- ]caused|pr caused)\s*:\s*(yes|no|unknown|not proven)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_TRANSIENT_STATUS_RE = re.compile(
|
||||
r"(?:passed after transient failure investigation|transient failure investigation)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_PLAIN_PASS_RE = re.compile(
|
||||
r"validation\s*:\s*(?:pass|passed|strong|ok|green)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_ENV_CLEANUP_RE = re.compile(
|
||||
r"(?:environmental cleanup|state cleaned|what changed between runs)\s*:",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_UNKNOWN_CAUSE_RE = re.compile(
|
||||
r"(?:suspected cause|cause)\s*:\s*(?:unknown|unexplained|not determined)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def _failure_documented_in_text(text: str, failure: dict[str, Any]) -> bool:
|
||||
"""Return True when *failure* appears documented in free-form report text."""
|
||||
command = (failure.get("command") or "").strip()
|
||||
failing = (failure.get("failing_test") or failure.get("error") or "").strip()
|
||||
if command and command not in text:
|
||||
return False
|
||||
if failing and failing not in text:
|
||||
return False
|
||||
return bool(command or failing)
|
||||
|
||||
|
||||
def _section_has_structured_fields(text: str) -> bool:
|
||||
if not _SECTION_RE.search(text):
|
||||
return False
|
||||
section_start = _SECTION_RE.search(text).start()
|
||||
section = text[section_start:]
|
||||
has_command = bool(_COMMAND_RE.search(section))
|
||||
has_failure = bool(_FAILING_TEST_RE.search(section))
|
||||
return has_command and has_failure
|
||||
|
||||
|
||||
def assess_validation_failure_history_report(
|
||||
report_text: str,
|
||||
*,
|
||||
validation_session: dict | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Require final reports to account for transient validation failures (#396)."""
|
||||
text = report_text or ""
|
||||
session = dict(validation_session or {})
|
||||
reasons: list[str] = []
|
||||
violations: list[str] = []
|
||||
|
||||
observed = list(session.get("observed_failures") or [])
|
||||
final_status = (session.get("final_validation_status") or "").strip().lower()
|
||||
cause_unknown = any(
|
||||
(f.get("suspected_cause") or "").strip().lower() in {
|
||||
"unknown", "unexplained", "not determined", ""
|
||||
}
|
||||
and not (f.get("pr_caused") or "").strip().lower() in {"yes", "no"}
|
||||
for f in observed
|
||||
) or bool(session.get("cause_unknown"))
|
||||
|
||||
if not observed:
|
||||
return {
|
||||
"proven": True,
|
||||
"block": False,
|
||||
"reasons": [],
|
||||
"violations": [],
|
||||
"observed_failure_count": 0,
|
||||
"safe_next_action": "proceed",
|
||||
}
|
||||
|
||||
documented = _section_has_structured_fields(text)
|
||||
if not documented:
|
||||
for failure in observed:
|
||||
if _failure_documented_in_text(text, failure):
|
||||
documented = True
|
||||
break
|
||||
|
||||
if not documented:
|
||||
violations.append(
|
||||
"session observed validation failure(s) but final report omits "
|
||||
"Validation failure history"
|
||||
)
|
||||
reasons.append(
|
||||
"every validation failure observed during the session must appear "
|
||||
"in a Validation failure history section"
|
||||
)
|
||||
|
||||
if documented and not _SECTION_RE.search(text):
|
||||
reasons.append(
|
||||
"failure details must appear under an explicit "
|
||||
"'Validation failure history' heading"
|
||||
)
|
||||
|
||||
for idx, failure in enumerate(observed, start=1):
|
||||
prefix = f"failure #{idx}"
|
||||
if not _failure_documented_in_text(text, failure) and documented:
|
||||
reasons.append(
|
||||
f"{prefix}: command and failing test/error not documented in report"
|
||||
)
|
||||
command = (failure.get("command") or "").strip()
|
||||
failing = (failure.get("failing_test") or failure.get("error") or "").strip()
|
||||
if not command:
|
||||
reasons.append(f"{prefix}: missing command in session failure record")
|
||||
if not failing:
|
||||
reasons.append(
|
||||
f"{prefix}: missing failing test or error in session failure record"
|
||||
)
|
||||
|
||||
plain_pass = bool(_PLAIN_PASS_RE.search(text))
|
||||
transient_wording = bool(_TRANSIENT_STATUS_RE.search(text))
|
||||
final_passed = final_status in {
|
||||
"passed",
|
||||
"pass",
|
||||
"passed_after_transient_failure_investigation",
|
||||
}
|
||||
|
||||
if (final_passed or plain_pass) and observed:
|
||||
if cause_unknown and not transient_wording:
|
||||
violations.append(
|
||||
"unknown transient failure cause cannot be erased as plain 'passed'"
|
||||
)
|
||||
reasons.append(
|
||||
"when cause is unknown, use status "
|
||||
"'passed after transient failure investigation'"
|
||||
)
|
||||
elif not transient_wording and final_status != "passed_after_transient_failure_investigation":
|
||||
if not _ENV_CLEANUP_RE.search(text):
|
||||
reasons.append(
|
||||
"later passing rerun must document what changed between runs "
|
||||
"or environmental cleanup performed"
|
||||
)
|
||||
|
||||
if session.get("environmental_contamination") and not _ENV_CLEANUP_RE.search(text):
|
||||
reasons.append(
|
||||
"environmental /tmp state contamination must document cleanup or "
|
||||
"what changed before the passing rerun"
|
||||
)
|
||||
|
||||
proven = not reasons and not violations
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": bool(violations) or not proven,
|
||||
"reasons": reasons,
|
||||
"violations": violations,
|
||||
"observed_failure_count": len(observed),
|
||||
"documented": documented,
|
||||
"safe_next_action": (
|
||||
"add Validation failure history with command, failing test, cause, "
|
||||
"reproduction, baseline comparison, and PR-caused evidence; use "
|
||||
"'passed after transient failure investigation' when appropriate"
|
||||
if not proven
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
@@ -91,4 +91,23 @@ Verifier: `review_proofs.assess_queue_status_report()`.
|
||||
|
||||
Narrative final report and controller handoff must agree on eligibility class,
|
||||
candidate/reviewed head SHA, mutation state, worktree usage, review decision,
|
||||
terminal review mutation, merge result, and linked issue status.
|
||||
terminal review mutation, merge result, and linked issue status.
|
||||
|
||||
### Proof-backed claims (#395)
|
||||
|
||||
Proof-sensitive claims must cite explicit command/tool evidence in the report
|
||||
or structured MCP metadata — not narrative alone:
|
||||
|
||||
- **Inventory complete:** `has_more=false`, `is_final_page=true`,
|
||||
`inventory_complete=true`, and/or `total_count` from `gitea_list_prs`.
|
||||
- **Skipped earlier PRs:** merge-simulation command output, conflict file list,
|
||||
or live `gitea_get_pr_review_feedback` / mergeability fields.
|
||||
- **Baseline validation:** baseline worktree path, baseline target SHA,
|
||||
dirty-before/after, exact command, exact result.
|
||||
- **Master integration:** exact `git merge` / `git rebase` command and exit status.
|
||||
- **Cleanup:** final `git worktree list` (or remove commands) proving session
|
||||
worktrees were removed.
|
||||
|
||||
When a claim relies on prior-session blocker state or MCP metadata only, label
|
||||
the proof source explicitly (`command`, `MCP metadata`, `prior blocker`,
|
||||
`not checked`). Do not use `live proof` without that classification.
|
||||
@@ -538,7 +538,37 @@ Official validation integrity status must be one of:
|
||||
|
||||
Do not invent a softer status.
|
||||
|
||||
## 21A. Validation status taxonomy (#406)
|
||||
## 21A. Validation failure history rule
|
||||
|
||||
Every validation failure observed during the review session must appear in the
|
||||
final report, even if a later rerun passes.
|
||||
|
||||
Before claiming `Validation: passed`, list every failed validation command from
|
||||
the session under `Validation failure history`.
|
||||
|
||||
For each failure, report:
|
||||
|
||||
* command
|
||||
* failing test or error
|
||||
* suspected cause
|
||||
* whether it reproduced
|
||||
* whether it exists on baseline master
|
||||
* evidence for whether it is or is not PR-caused
|
||||
|
||||
If a later rerun passes, also report:
|
||||
|
||||
* what changed between runs
|
||||
* whether environmental state was cleaned (for example `/tmp` lock files)
|
||||
* why the pass is trustworthy
|
||||
|
||||
If the cause is unknown, do not erase the earlier failure with plain
|
||||
`Validation: passed`. Use a status such as
|
||||
`passed after transient failure investigation`.
|
||||
|
||||
`gitea_validate_review_final_report` rejects reports that omit known earlier
|
||||
validation failures when `validation_session.observed_failures` is supplied.
|
||||
|
||||
## 21B. Validation status taxonomy (#406)
|
||||
|
||||
When the final report summarizes how validation concluded, use one of these
|
||||
**validation status** labels (distinct from per-command pass/fail entries):
|
||||
@@ -1118,6 +1148,7 @@ Controller Handoff:
|
||||
* Baseline worktree path:
|
||||
* Files reviewed:
|
||||
* Validation:
|
||||
* Validation failure history:
|
||||
* Official validation integrity status:
|
||||
* Terminal review mutation:
|
||||
* Review decision:
|
||||
|
||||
@@ -0,0 +1,217 @@
|
||||
"""Tests for gitea_delete_branch capability gate (Issue #408).
|
||||
|
||||
``gitea_delete_branch`` requires the exact ``gitea.branch.delete`` operation:
|
||||
without it the delete fails closed (no preflight, no auth lookup, no API call,
|
||||
structured permission report). With it, deletion proceeds through existing
|
||||
preflight and audit unchanged.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import mcp_server
|
||||
from mcp_server import gitea_delete_branch
|
||||
|
||||
FAKE_AUTH = "token fake"
|
||||
|
||||
AUTHOR_NO_DELETE = {
|
||||
"profile_name": "prgs-author",
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.pr.create", "gitea.pr.comment",
|
||||
"gitea.branch.push", "gitea.issue.comment",
|
||||
],
|
||||
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
|
||||
"audit_label": "prgs-author",
|
||||
}
|
||||
|
||||
AUTHOR_WITH_DELETE = {
|
||||
"profile_name": "prgs-author-deleter",
|
||||
"allowed_operations": AUTHOR_NO_DELETE["allowed_operations"] + [
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
|
||||
"audit_label": "prgs-author-deleter",
|
||||
}
|
||||
|
||||
CONFIG = {
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"ctx": {
|
||||
"enabled": True,
|
||||
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"author-no-delete": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "author-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": AUTHOR_NO_DELETE["allowed_operations"],
|
||||
"forbidden_operations": [],
|
||||
"execution_profile": "author-no-delete",
|
||||
},
|
||||
"author-with-delete": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "deleter-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": AUTHOR_WITH_DELETE["allowed_operations"],
|
||||
"forbidden_operations": [],
|
||||
"execution_profile": "author-with-delete",
|
||||
},
|
||||
"reviewer-profile": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "reviewer",
|
||||
"username": "reviewer-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.pr.review", "gitea.pr.merge",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.branch.delete", "gitea.branch.push", "gitea.pr.create",
|
||||
],
|
||||
"execution_profile": "reviewer-profile",
|
||||
},
|
||||
},
|
||||
"rules": {"allow_runtime_switching": False},
|
||||
}
|
||||
|
||||
|
||||
class TestDeleteBranchToolGate(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self._preflight_snapshot = (
|
||||
mcp_server._preflight_whoami_called,
|
||||
mcp_server._preflight_capability_called,
|
||||
)
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
||||
"repo": "Example-Repo"},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
patch("gitea_config.load_config", return_value={}).start()
|
||||
patch(
|
||||
"gitea_config.is_runtime_switching_enabled", return_value=False
|
||||
).start()
|
||||
patch("gitea_audit.audit_enabled", return_value=False).start()
|
||||
self.mock_api = patch("mcp_server.api_request").start()
|
||||
self.mock_auth = patch(
|
||||
"mcp_server.get_auth_header", return_value=FAKE_AUTH
|
||||
).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
|
||||
self._preflight_snapshot
|
||||
)
|
||||
|
||||
def _set_profile(self, profile):
|
||||
patch("mcp_server.get_profile", return_value=profile).start()
|
||||
|
||||
def test_blocked_without_delete_capability(self):
|
||||
self._set_profile(AUTHOR_NO_DELETE)
|
||||
res = gitea_delete_branch(branch="feat/branch", remote="prgs")
|
||||
self.assertFalse(res["success"])
|
||||
self.assertFalse(res["performed"])
|
||||
self.assertEqual(res["required_permission"], "gitea.branch.delete")
|
||||
self.assertTrue(res["reasons"])
|
||||
self.assertEqual(
|
||||
res["permission_report"]["missing_permission"],
|
||||
"gitea.branch.delete",
|
||||
)
|
||||
self.mock_api.assert_not_called()
|
||||
self.mock_auth.assert_not_called()
|
||||
|
||||
def test_allowed_delete_proceeds(self):
|
||||
self._set_profile(AUTHOR_WITH_DELETE)
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
res = gitea_delete_branch(branch="feat/branch", remote="prgs")
|
||||
self.assertTrue(res["success"])
|
||||
self.assertIn("deleted", res["message"])
|
||||
delete_calls = [
|
||||
c for c in self.mock_api.call_args_list if c.args[0] == "DELETE"
|
||||
]
|
||||
self.assertTrue(delete_calls)
|
||||
|
||||
def test_allowed_delete_audited_with_capability_proof(self):
|
||||
self._set_profile(AUTHOR_WITH_DELETE)
|
||||
patch("gitea_audit.audit_enabled", return_value=True).start()
|
||||
mock_write = patch("gitea_audit.write_event").start()
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
self.mock_api.return_value = {}
|
||||
gitea_delete_branch(branch="feat/branch", remote="prgs")
|
||||
mock_write.assert_called()
|
||||
event = mock_write.call_args[0][0]
|
||||
self.assertEqual(event["action"], "delete_branch")
|
||||
self.assertEqual(
|
||||
event["request_metadata"]["required_permission"],
|
||||
"gitea.branch.delete",
|
||||
)
|
||||
|
||||
|
||||
class TestDeleteBranchResolverParity(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
||||
"repo": "Example-Repo"},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(CONFIG))
|
||||
patch(
|
||||
"gitea_config.is_runtime_switching_enabled", return_value=False
|
||||
).start()
|
||||
patch("gitea_audit.audit_enabled", return_value=False).start()
|
||||
self.mock_api = patch("mcp_server.api_request").start()
|
||||
patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
self._dir.cleanup()
|
||||
|
||||
def _env(self, profile: str) -> dict:
|
||||
return {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
||||
}
|
||||
|
||||
def test_reviewer_resolver_denial_blocks_raw_tool(self):
|
||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
||||
resolve = mcp_server.gitea_resolve_task_capability(
|
||||
task="delete_branch", remote="prgs")
|
||||
self.assertFalse(resolve["allowed_in_current_session"])
|
||||
res = gitea_delete_branch(branch="feat/branch", remote="prgs")
|
||||
self.assertFalse(res["success"])
|
||||
self.assertEqual(
|
||||
res["permission_report"]["missing_permission"],
|
||||
resolve["required_operation_permission"],
|
||||
)
|
||||
delete_calls = [
|
||||
c for c in self.mock_api.call_args_list if c.args[0] == "DELETE"
|
||||
]
|
||||
self.assertFalse(delete_calls)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -207,6 +207,12 @@ def test_validation_integrity_verifier_exported():
|
||||
assert callable(assess_validation_integrity_report)
|
||||
|
||||
|
||||
def test_validation_failure_history_verifier_exported():
|
||||
from review_proofs import assess_validation_failure_history_report
|
||||
|
||||
assert callable(assess_validation_failure_history_report)
|
||||
|
||||
|
||||
def test_prior_blocker_skip_verifier_exported():
|
||||
from review_proofs import assess_prior_blocker_skip_proof
|
||||
|
||||
@@ -249,6 +255,12 @@ def test_inventory_worktree_verifier_exported():
|
||||
assert callable(assess_inventory_worktree_report)
|
||||
|
||||
|
||||
def test_proof_backed_handoff_verifier_exported():
|
||||
from review_proofs import assess_proof_backed_handoff_report
|
||||
|
||||
assert callable(assess_proof_backed_handoff_report)
|
||||
|
||||
|
||||
def test_infra_stop_handoff_verifier_exported():
|
||||
from review_proofs import assess_infra_stop_handoff_report
|
||||
|
||||
|
||||
@@ -1005,9 +1005,19 @@ class TestReviewPR(unittest.TestCase):
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestDeleteBranch(unittest.TestCase):
|
||||
|
||||
DELETE_PROFILE = {
|
||||
"profile_name": "test-deleter",
|
||||
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "test-deleter",
|
||||
}
|
||||
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_delete_branch(self, _auth, mock_api):
|
||||
def test_delete_branch(self, _auth, mock_api, _profile):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
mock_api.return_value = {}
|
||||
result = gitea_delete_branch(branch="feat/branch")
|
||||
self.assertTrue(result["success"])
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Tests for proof-backed reviewer handoff verifier (#395)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_proof_backed_handoff import ( # noqa: E402
|
||||
assess_proof_backed_handoff_report,
|
||||
)
|
||||
|
||||
|
||||
def _good_report() -> str:
|
||||
return "\n".join([
|
||||
"Inventory pagination proof: is_final_page: true, has_more: false, total_count: 11",
|
||||
"Earlier PRs skipped: #372 non-mergeable — merge simulation conflicts in review_proofs.py",
|
||||
"Baseline worktree path: branches/baseline-master-pr383",
|
||||
"Baseline target SHA: 4243b60ca32d79f1ee5e6b0f10d7570e9dea2937",
|
||||
"Baseline dirty before validation: false",
|
||||
"Baseline validation command: venv/bin/python -m pytest tests/ -q",
|
||||
"Baseline validation result: 1 failed (test_clean_reviewer_report_passes)",
|
||||
"Baseline dirty after validation: false",
|
||||
"Cleanup mutations: git worktree list showed session worktrees removed",
|
||||
"## Controller Handoff",
|
||||
"- Inventory pagination proof: inventory_complete: true, total_count: 11",
|
||||
"- Earlier PRs skipped: #372 conflict proof via merge simulation",
|
||||
"- Baseline worktree used: true",
|
||||
"- Baseline worktree path: branches/baseline-master-pr383",
|
||||
"- Cleanup mutations: git worktree remove branches/review-pr383",
|
||||
])
|
||||
|
||||
|
||||
class TestProofBackedHandoff(unittest.TestCase):
|
||||
def test_fully_proof_backed_report_passes(self):
|
||||
result = assess_proof_backed_handoff_report(
|
||||
_good_report(),
|
||||
action_log=[{"command": "gitea_list_prs", "result": "inventory_complete=true"}],
|
||||
baseline_session={
|
||||
"complete": True,
|
||||
"clean_before": True,
|
||||
"clean_after": True,
|
||||
"command": "pytest",
|
||||
"result": "passed",
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_inventory_page_size_assumption_blocks(self):
|
||||
report = (
|
||||
"Inventory is complete because 13 results are less than page size 50."
|
||||
)
|
||||
result = assess_proof_backed_handoff_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("page-size" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_skip_without_conflict_proof_blocks(self):
|
||||
report = "\n".join([
|
||||
"Earlier PRs skipped: #372, #374",
|
||||
"## Controller Handoff",
|
||||
"- Earlier PRs skipped: #372, #374",
|
||||
])
|
||||
result = assess_proof_backed_handoff_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("skip" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_baseline_without_command_blocks(self):
|
||||
report = "Baseline validation passed; same as master failures."
|
||||
result = assess_proof_backed_handoff_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("baseline" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_cleanup_without_worktree_list_blocks(self):
|
||||
report = "Worktrees were cleaned after merge."
|
||||
result = assess_proof_backed_handoff_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("cleanup" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_cleanup_with_worktree_list_passes(self):
|
||||
report = "Cleanup: git worktree list confirmed removal."
|
||||
result = assess_proof_backed_handoff_report(
|
||||
report,
|
||||
action_log=[{"command": "git worktree list"}],
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_master_integration_requires_command(self):
|
||||
report = "Merged master into the review worktree before validation."
|
||||
result = assess_proof_backed_handoff_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("master integration" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_master_integration_with_action_log_passes(self):
|
||||
report = "Merged master into review worktree; merge_exit=0"
|
||||
result = assess_proof_backed_handoff_report(
|
||||
report,
|
||||
action_log=[{"command": "git merge --no-commit prgs/master"}],
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_live_proof_requires_source_classification(self):
|
||||
report = "Live proof shows inventory complete."
|
||||
result = assess_proof_backed_handoff_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("proof source" in r.lower() for r in result["reasons"]))
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_proof_backed_handoff_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
result = exported(_good_report(), inventory_session={"inventory_complete": True})
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,147 @@
|
||||
"""Tests for transient validation failure history verifier (#396)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from final_report_validator import assess_final_report_validator # noqa: E402
|
||||
from reviewer_validation_failure_history import ( # noqa: E402
|
||||
assess_validation_failure_history_report,
|
||||
)
|
||||
|
||||
|
||||
def _full_history_report() -> str:
|
||||
return "\n".join([
|
||||
"Validation failure history",
|
||||
"Failure #1:",
|
||||
"Command: venv/bin/python -m pytest tests/test_worktrees.py -q",
|
||||
"Failing test: tests/test_worktrees.py::TestWorktreeStart::test_rejects_untraceable_branches",
|
||||
"Suspected cause: stale /tmp/gitea_issue_lock.json from prior session",
|
||||
"Reproduced: no",
|
||||
"On baseline master: no",
|
||||
"PR-caused: no",
|
||||
"What changed between runs: removed /tmp/gitea_issue_lock.json",
|
||||
"Environmental cleanup: lock file removed before rerun",
|
||||
"Validation: passed after transient failure investigation",
|
||||
"Final validation command: venv/bin/python -m pytest tests/ -q",
|
||||
"Final result: 1497 passed, 6 skipped",
|
||||
])
|
||||
|
||||
|
||||
class TestValidationFailureHistory(unittest.TestCase):
|
||||
def test_no_failures_observed_passes(self):
|
||||
result = assess_validation_failure_history_report(
|
||||
"Validation: passed",
|
||||
validation_session={"observed_failures": []},
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_omitted_failure_blocks(self):
|
||||
result = assess_validation_failure_history_report(
|
||||
"Validation: passed\n1497 passed, 6 skipped",
|
||||
validation_session={
|
||||
"observed_failures": [{
|
||||
"command": "pytest tests/test_worktrees.py -q",
|
||||
"failing_test": (
|
||||
"tests/test_worktrees.py::TestWorktreeStart::"
|
||||
"test_rejects_untraceable_branches"
|
||||
),
|
||||
}],
|
||||
"final_validation_status": "passed",
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_full_history_with_transient_investigation_passes(self):
|
||||
result = assess_validation_failure_history_report(
|
||||
_full_history_report(),
|
||||
validation_session={
|
||||
"observed_failures": [{
|
||||
"command": (
|
||||
"venv/bin/python -m pytest tests/test_worktrees.py -q"
|
||||
),
|
||||
"failing_test": (
|
||||
"tests/test_worktrees.py::TestWorktreeStart::"
|
||||
"test_rejects_untraceable_branches"
|
||||
),
|
||||
"suspected_cause": "stale /tmp/gitea_issue_lock.json",
|
||||
"reproduced": "no",
|
||||
"on_baseline_master": "no",
|
||||
"pr_caused": "no",
|
||||
}],
|
||||
"final_validation_status": "passed_after_transient_failure_investigation",
|
||||
"environmental_contamination": True,
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_baseline_equivalent_failure_documented(self):
|
||||
report = "\n".join([
|
||||
"Validation failure history",
|
||||
"Command: pytest tests/test_example.py -q",
|
||||
"Failing test: tests/test_example.py::test_flaky",
|
||||
"Suspected cause: pre-existing master failure",
|
||||
"On baseline master: yes",
|
||||
"PR-caused: no",
|
||||
"What changed between runs: none; failure matches baseline",
|
||||
"Validation: passed after transient failure investigation",
|
||||
])
|
||||
result = assess_validation_failure_history_report(
|
||||
report,
|
||||
validation_session={
|
||||
"observed_failures": [{
|
||||
"command": "pytest tests/test_example.py -q",
|
||||
"failing_test": "tests/test_example.py::test_flaky",
|
||||
"on_baseline_master": "yes",
|
||||
"pr_caused": "no",
|
||||
}],
|
||||
"final_validation_status": "passed_after_transient_failure_investigation",
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_unknown_cause_plain_pass_blocks(self):
|
||||
result = assess_validation_failure_history_report(
|
||||
"Validation: passed",
|
||||
validation_session={
|
||||
"observed_failures": [{
|
||||
"command": "pytest tests/ -q",
|
||||
"failing_test": "tests/test_foo.py::test_bar",
|
||||
"suspected_cause": "unknown",
|
||||
}],
|
||||
"final_validation_status": "passed",
|
||||
"cause_unknown": True,
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("transient" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_final_report_validator_rejects_omitted_failure(self):
|
||||
result = assess_final_report_validator(
|
||||
"Validation: passed",
|
||||
"review_pr",
|
||||
validation_session={
|
||||
"observed_failures": [{
|
||||
"command": "pytest tests/ -q",
|
||||
"failing_test": "tests/test_foo.py::test_bar",
|
||||
}],
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["blocked"] or result["downgraded"])
|
||||
self.assertTrue(
|
||||
any(
|
||||
f.get("rule_id") == "reviewer.validation_failure_history"
|
||||
for f in result.get("findings") or []
|
||||
)
|
||||
)
|
||||
|
||||
def test_exported_from_review_proofs(self):
|
||||
from review_proofs import assess_validation_failure_history_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user