Compare commits

..
15 Commits
Author SHA1 Message Date
sysadminandClaude Opus 4.8 1320a55a7e feat: require validation cwd and HEAD proof in reviewer reports (Closes #398)
Rebased onto current prgs/master; merged with #396 validation failure
history by keeping both validator rules and workflow handoff fields.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 13:09:58 -04:00
sysadmin 1a1e679246 Merge pull request 'feat: gate gitea_delete_branch on gitea.branch.delete capability (Closes #408)' (#410) from feat/issue-408-delete-branch-capability-gate into master 2026-07-07 10:48:51 -05:00
sysadmin 9fde4f3e76 Merge pull request 'feat: require transient validation failure history in reviewer reports (Closes #396)' (#409) from feat/issue-396-transient-validation-failure-history into master 2026-07-07 10:42:56 -05:00
sysadmin 6b5d2ad080 Merge pull request 'feat: require proof-backed claims in reviewer handoff reports (Closes #395)' (#397) from feat/issue-395-proof-backed-review-handoff into master 2026-07-07 10:34:35 -05:00
sysadmin f5654963eb Merge pull request 'feat: add prgs-reconciler profile model and role detection (Closes #304)' (#388) from feat/issue-304-reconciler-profile into master 2026-07-07 10:27:12 -05:00
sysadminandClaude Opus 4.8 af7131abf1 feat: gate gitea_delete_branch on gitea.branch.delete capability (Closes #408)
Add fail-closed profile gate at tool entry before preflight or API calls,
return structured permission reports on block, and record required_permission
in delete_branch audit metadata. Regression tests prove resolver-denied
sessions cannot bypass the gate through the raw tool.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 11:26:37 -04:00
sysadminandClaude Opus 4.8 71ccbca10f feat: require transient validation failure history in reviewer reports (Closes #396)
Add assess_validation_failure_history_report so reviewer final reports must
document every validation failure observed during a session, not only the
final passing result. Wire the verifier into final_report_validator,
gitea_validate_review_final_report, and the review-merge workflow template.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 11:22:00 -04:00
sysadmin 6220304f8c fix: resolve conflicts for PR #397
Merge prgs/master into feat/issue-395-proof-backed-review-handoff; keep
both proof-backed handoff (#395) and already-landed classification (#295)
downgrade gates in review_proofs.py.
2026-07-07 11:21:22 -04:00
sysadmin 6a37f7c8d5 fix: resolve conflicts for PR #388
Merge prgs/master into feat/issue-304-reconciler-profile; keep both
reconciler_profile and reconciliation_workflow imports in gitea_mcp_server.py.
2026-07-07 11:19:57 -04:00
sysadmin d814a9ca81 fix: resolve conflicts for PR #388
Merge prgs/master; unify reconciler profile docs (#304) with master's
already-landed close tool guidance (#310).
2026-07-07 10:34:57 -04:00
sysadmin c1d85b621a feat: require proof-backed claims in reviewer handoff reports (Closes #395) 2026-07-07 10:24:54 -04:00
sysadmin 9ff861a1f3 fix: resolve conflicts for PR #388
Merge prgs/master into feat/issue-304-reconciler-profile; keep both
reconciler_profile and review_merge_state_machine imports.

Closes conflict with master landing review_merge_state_machine (#389 stack).
2026-07-07 10:19:40 -04:00
sysadmin 7ef4c53cd8 Merge remote-tracking branch 'prgs/master' into feat/issue-304-reconciler-profile 2026-07-07 09:57:58 -04:00
sysadmin b604b468b5 fix: resolve conflicts for PR #388
Merge prgs/master and keep reconciler profile runtime fields (#304)
alongside native MCP shell health (#270).
2026-07-07 09:57:01 -04:00
sysadminandClaude Opus 4.8 21ff12cef7 feat: add prgs-reconciler profile model and role detection (Closes #304)
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 09:39:52 -04:00
16 changed files with 1259 additions and 12 deletions
+1 -1
View File
@@ -23,7 +23,7 @@ launched with exactly one static execution profile:
|-----------------------------|----------------|-------------|
| `gitea-author` | an author profile | implement issues, push branches, open PRs, comment |
| `gitea-reviewer` | a reviewer profile | review, approve/request changes, merge |
| `gitea-reconciler` | a reconciler profile | close already-landed open PRs after ancestry proof (#310) |
| `gitea-reconciler` | a reconciler profile | close already-landed open PRs after ancestry proof (#304 profile; #310 close tool) |
Properties:
+13 -6
View File
@@ -226,11 +226,12 @@ explicit operator-directed closure of a contaminated PR). If `close_pr` ever
resolves as unknown, agents must fail closed rather than fall back to the
edit path.
## Reconciler profile for already-landed open PRs (#310)
## Reconciler profile for already-landed open PRs (#304 / #310)
Normal author and reviewer profiles must not gain broad PR-close authority.
Already-landed open PRs (head SHA is an ancestor of the target branch) need a
dedicated reconciler profile such as `prgs-reconciler` with:
Normal author and reviewer profiles must not gain broad `gitea.pr.close`
authority. Already-landed open PRs (head SHA is an ancestor of the target
branch) need a dedicated reconciler profile such as `prgs-reconciler` with a
narrow operation set:
- `gitea.read`
- `gitea.pr.comment`
@@ -238,8 +239,14 @@ dedicated reconciler profile such as `prgs-reconciler` with:
- `gitea.issue.close`
- `gitea.pr.close`
Use the `gitea-reconciler` MCP namespace (static profile launch) and the
`gitea_reconcile_already_landed_pr` tool. The resolver task is
Forbidden on reconciler profiles: `gitea.pr.approve`, `gitea.pr.merge`,
`gitea.pr.review`, `gitea.pr.create`, `gitea.branch.push`, and
`gitea.repo.commit`.
Launch a static `gitea-reconciler` MCP namespace with
`GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
`reconciler_profile.assess_reconciler_profile` (#304). Use the
`gitea_reconcile_already_landed_pr` tool (#310). The resolver task is
`reconcile_already_landed_pr`. PR close is allowed only after live PR fetch,
fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose
heads are not already landed cannot be closed through this path.
+35
View File
@@ -456,6 +456,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_cwd_proof(
report_text: str,
*,
@@ -900,6 +934,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_cwd_proof,
_rule_reviewer_validation_structured,
_rule_reviewer_linked_issue,
+27 -3
View File
@@ -504,6 +504,7 @@ import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402
import review_merge_state_machine # noqa: E402
import native_mcp_preference # noqa: E402
@@ -1244,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) ──
@@ -3463,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."}
@@ -4472,12 +4489,15 @@ def gitea_create_issue_comment(
def _role_kind(allowed, forbidden) -> str:
"""Classify the active profile from its normalized operations.
'reconciler' can close already-landed PRs without review/author powers;
'author' can create PRs / push branches; 'reviewer' can approve/merge;
'reconciler' can close already-landed PRs via ``gitea.pr.close`` without
review/author powers; 'mixed' can do both (a config smell the
reviewer-deadlock invariant forbids it in v2 configs); 'limited' can do
neither.
"""
if reconciler_profile.is_reconciler_profile(allowed, forbidden):
return "reconciler"
def can(op):
return gitea_config.check_operation(op, allowed, forbidden)[0]
review = can("gitea.pr.approve") or can("gitea.pr.merge")
@@ -5646,6 +5666,10 @@ def gitea_get_runtime_context(
"preflight_block_reasons": preflight["preflight_block_reasons"],
"preflight_workspace": preflight.get("preflight_workspace"),
"session_capabilities": session_capabilities,
"reconciler_profile_assessment": reconciler_profile.assess_reconciler_profile(
allowed, forbidden
),
"role_kind": _role_kind(allowed, forbidden),
"shell_health": native_mcp_preference.shell_health_status(),
}
+94
View File
@@ -0,0 +1,94 @@
"""Reconciler profile model for already-landed PR closure (#304).
Defines the narrowly scoped operation set for a dedicated reconciler profile
such as ``prgs-reconciler``. Close MCP tooling and ancestry gates ship in the
#310 stack; this module validates profile shape only.
"""
from __future__ import annotations
import gitea_config
RECONCILER_REQUIRED_OPERATIONS = (
"gitea.read",
"gitea.pr.close",
)
RECONCILER_RECOMMENDED_OPERATIONS = (
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.issue.close",
)
RECONCILER_FORBIDDEN_OPERATIONS = (
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.review",
"gitea.pr.create",
"gitea.branch.push",
"gitea.repo.commit",
)
def _normalized_allowed(allowed: list[str]) -> set[str]:
normalized: set[str] = set()
for entry in allowed or []:
try:
normalized.add(gitea_config.normalize_operation(entry))
except gitea_config.ConfigError:
continue
return normalized
def _forbidden_in_allowed(allowed: list[str], ops: tuple[str, ...]) -> list[str]:
"""Return forbidden ops that are explicitly listed in *allowed*."""
allowed_n = _normalized_allowed(allowed)
present: list[str] = []
for op in ops:
try:
if gitea_config.normalize_operation(op) in allowed_n:
present.append(op)
except gitea_config.ConfigError:
continue
return present
def is_reconciler_profile(allowed: list[str], forbidden: list[str]) -> bool:
"""Return True when *allowed*/*forbidden* describe a reconciler profile."""
def can(op: str) -> bool:
return gitea_config.check_operation(op, allowed, forbidden)[0]
if not can("gitea.pr.close"):
return False
if _forbidden_in_allowed(allowed, RECONCILER_FORBIDDEN_OPERATIONS):
return False
return True
def assess_reconciler_profile(allowed: list[str], forbidden: list[str]) -> dict:
"""Validate reconciler profile operations (read-only, fail closed)."""
allowed = list(allowed or [])
forbidden = list(forbidden or [])
reasons: list[str] = []
for op in RECONCILER_REQUIRED_OPERATIONS:
ok, _ = gitea_config.check_operation(op, allowed, forbidden)
if not ok:
reasons.append(f"missing required operation {op}")
for op in _forbidden_in_allowed(allowed, RECONCILER_FORBIDDEN_OPERATIONS):
reasons.append(f"forbidden operation must not be allowed: {op}")
missing_recommended = [
op for op in RECONCILER_RECOMMENDED_OPERATIONS
if not gitea_config.check_operation(op, allowed, forbidden)[0]
]
return {
"is_reconciler_profile": is_reconciler_profile(allowed, forbidden),
"valid": not reasons and is_reconciler_profile(allowed, forbidden),
"reasons": reasons,
"missing_recommended_operations": missing_recommended,
"required_operations": list(RECONCILER_REQUIRED_OPERATIONS),
"forbidden_operations": list(RECONCILER_FORBIDDEN_OPERATIONS),
}
+32
View File
@@ -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_validation_cwd_proof_report(report_text, **kwargs):
"""#398: validation commands require explicit worktree cwd and HEAD proof."""
from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report as _assess
@@ -5573,3 +5598,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)
+217
View File
@@ -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"
),
}
+198
View File
@@ -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,6 +538,36 @@ Official validation integrity status must be one of:
Do not invent a softer status.
## 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.
## 22. Baseline validation rule
Do not run tests in the main checkout.
@@ -1118,6 +1148,7 @@ Controller Handoff:
* Baseline worktree path:
* Files reviewed:
* Validation:
* Validation failure history:
* Validation cwd/HEAD proof:
* Official validation integrity status:
* Terminal review mutation:
+217
View File
@@ -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()
+12
View File
@@ -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_validation_cwd_proof_verifier_exported():
from review_proofs import assess_validation_cwd_proof_report
@@ -255,6 +261,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
+11 -1
View File
@@ -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"])
+87
View File
@@ -0,0 +1,87 @@
"""Tests for reconciler profile model (#304)."""
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import gitea_mcp_server as mcp_server
import migrate_profiles
import reconciler_profile
PRGS_RECONCILER_ALLOWED = [
"gitea.read",
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.issue.close",
"gitea.pr.close",
]
PRGS_RECONCILER_FORBIDDEN = [
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.create",
"gitea.branch.push",
]
class TestReconcilerProfileModel(unittest.TestCase):
def test_prgs_reconciler_shape_valid(self):
result = reconciler_profile.assess_reconciler_profile(
PRGS_RECONCILER_ALLOWED,
PRGS_RECONCILER_FORBIDDEN,
)
self.assertTrue(result["is_reconciler_profile"])
self.assertTrue(result["valid"])
self.assertEqual(result["reasons"], [])
def test_author_profile_not_reconciler(self):
allowed = [
"gitea.read",
"gitea.pr.create",
"gitea.branch.push",
"gitea.issue.comment",
]
forbidden = ["gitea.pr.approve", "gitea.pr.merge"]
self.assertFalse(
reconciler_profile.is_reconciler_profile(allowed, forbidden)
)
def test_reviewer_profile_not_reconciler(self):
allowed = [
"gitea.read",
"gitea.pr.review",
"gitea.pr.approve",
"gitea.pr.merge",
]
forbidden = ["gitea.pr.create", "gitea.branch.push"]
self.assertFalse(
reconciler_profile.is_reconciler_profile(allowed, forbidden)
)
def test_broad_close_on_author_invalid(self):
allowed = PRGS_RECONCILER_ALLOWED + ["gitea.pr.create"]
result = reconciler_profile.assess_reconciler_profile(
allowed,
PRGS_RECONCILER_FORBIDDEN,
)
self.assertFalse(result["valid"])
self.assertFalse(result["is_reconciler_profile"])
def test_role_kind_detects_reconciler(self):
role = mcp_server._role_kind(
PRGS_RECONCILER_ALLOWED,
PRGS_RECONCILER_FORBIDDEN,
)
self.assertEqual(role, "reconciler")
def test_migrate_infer_role_reconciler(self):
self.assertEqual(
migrate_profiles.infer_role("prgs-reconciler", "prgs-reconciler"),
"reconciler",
)
if __name__ == "__main__":
unittest.main()
+117
View File
@@ -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()