feat(review-workflow): raise A-bar with capability, sweep, live-state, and role-boundary proofs (#179)
Extends review_proofs.py with the four #179 proofs, the successor set to the #173 checkout/inventory proofs: - assess_capability_evidence: a capability claim (review_pr, merge_pr, ...) counts only with exact evidence citing gitea_resolve_task_capability output or equivalent runtime context; no claims at all fails closed. - assess_sweep_evidence: secret/provenance sweeps must state the exact command/script/pattern/named method, the scope scanned, and a boolean result; vague summaries are downgraded and a missing sweep fails closed. - assess_live_state_recheck: an explicit pre-mutation recheck must prove the PR is still open, the live head equals the pinned head (full 40-hex), the base branch is unchanged, and blocking review state was checked and absent; not performing it fails closed. - assess_role_boundary: a reviewer run using an author namespace (or vice versa) is clean only with an explicit justification; unreported namespace usage fails closed. build_final_report now takes the four proofs as keyword arguments: any missing or failed proof downgrades the grade, merge_allowed additionally requires the proven live-state recheck, and a merge performed without it is a blocked violation. Existing #173 semantics are unchanged otherwise; gates only get stricter. tests/test_review_proofs.py adds 29 tests covering the issue's harness assertions: capability claims without evidence downgraded, vague sweeps downgraded, missing/stale live-state recheck downgrades and blocks merge (violation when a merge is claimed anyway), unjustified author-namespace use downgraded, and the #173 positive baseline preserved. SKILL.md sections F/G and the review-pr/merge-pr templates now require the capability evidence, exact sweep, pre-verdict and pre-merge live-state rechecks, and reviewer-namespace discipline. Closes #179 Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
Claude Fable 5
+229
-12
@@ -330,22 +330,167 @@ def assess_self_review_contamination(reviewer_identity, pr_author,
|
||||
}
|
||||
|
||||
|
||||
def assess_role_boundary(proof):
|
||||
def assess_capability_evidence(capability_claims):
|
||||
"""#179 gap 1: a capability claim needs exact evidence, not assertion.
|
||||
|
||||
*capability_claims* is a list of ``{'task', 'allowed',
|
||||
'evidence_source'}`` dicts, one per capability the report claims (e.g.
|
||||
review_pr, merge_pr). Each claim must name its task, be allowed, and
|
||||
cite an exact evidence source (``gitea_resolve_task_capability`` output
|
||||
or equivalent runtime-context evidence). No claims at all fails closed.
|
||||
"""
|
||||
reasons = []
|
||||
claims = capability_claims or []
|
||||
if not claims:
|
||||
reasons.append(
|
||||
"no capability evidence provided; capability checks may not be "
|
||||
"claimed as passed"
|
||||
)
|
||||
for claim in claims:
|
||||
task = (claim.get("task") or "").strip() or "<unnamed task>"
|
||||
if claim.get("allowed") is not True:
|
||||
reasons.append(
|
||||
f"capability '{task}' is not proven allowed; fail closed"
|
||||
)
|
||||
if not (claim.get("evidence_source") or "").strip():
|
||||
reasons.append(
|
||||
f"capability '{task}' claimed without exact evidence "
|
||||
"(cite gitea_resolve_task_capability output or equivalent)"
|
||||
)
|
||||
proven = not reasons
|
||||
return {"proven": proven, "reasons": reasons, "claims": len(claims)}
|
||||
|
||||
|
||||
def assess_sweep_evidence(sweep):
|
||||
"""#179 gap 2: secret/provenance sweeps must state exact method + scope.
|
||||
|
||||
*sweep* keys: ``command`` (the exact command, script, grep pattern, or
|
||||
named sweep method), ``scope`` (what was scanned, e.g. 'full PR diff
|
||||
against prgs/master'), ``clean`` (bool result). A vague summary without
|
||||
the exact method is downgraded; a missing sweep fails closed.
|
||||
"""
|
||||
if not sweep:
|
||||
return {
|
||||
"verdict": "missing",
|
||||
"proven": False,
|
||||
"reasons": ["no secret/provenance sweep reported; fail closed"],
|
||||
}
|
||||
reasons = []
|
||||
if not (sweep.get("command") or "").strip():
|
||||
reasons.append(
|
||||
"sweep method/command not stated exactly (command, script, "
|
||||
"pattern, or named sweep method required)"
|
||||
)
|
||||
if not (sweep.get("scope") or "").strip():
|
||||
reasons.append("sweep scope not stated (what diff/files were scanned)")
|
||||
if not isinstance(sweep.get("clean"), bool):
|
||||
reasons.append("sweep result not stated as clean/not-clean")
|
||||
verdict = "exact" if not reasons else "vague"
|
||||
return {
|
||||
"verdict": verdict,
|
||||
"proven": verdict == "exact",
|
||||
"reasons": reasons,
|
||||
"clean": sweep.get("clean") if isinstance(sweep.get("clean"), bool)
|
||||
else None,
|
||||
}
|
||||
|
||||
|
||||
def assess_live_state_recheck(recheck):
|
||||
"""#179 gap 3: explicit live-state recheck before review/merge mutation.
|
||||
|
||||
*recheck* keys: ``pr_state``, ``pinned_head_sha``, ``live_head_sha``,
|
||||
``pinned_base_ref``, ``live_base_ref``, ``blocking_change_requests``.
|
||||
Proven only when the PR is still open, the live head equals the pinned
|
||||
head (full 40-hex), the base branch is unchanged, and blocking review
|
||||
state was checked and is absent. Not performing the recheck fails
|
||||
closed and blocks mutation.
|
||||
"""
|
||||
if not recheck:
|
||||
return {
|
||||
"proven": False,
|
||||
"block": True,
|
||||
"reasons": [
|
||||
"final live-state recheck not performed before mutation; "
|
||||
"fail closed"
|
||||
],
|
||||
}
|
||||
reasons = []
|
||||
if (recheck.get("pr_state") or "").strip().lower() != "open":
|
||||
reasons.append(
|
||||
f"PR state is '{recheck.get('pr_state')}', not open; stop"
|
||||
)
|
||||
|
||||
pinned = (recheck.get("pinned_head_sha") or "").strip().lower()
|
||||
live = (recheck.get("live_head_sha") or "").strip().lower()
|
||||
if not (_FULL_SHA.match(pinned) and _FULL_SHA.match(live)):
|
||||
reasons.append(
|
||||
"pinned/live head SHAs missing or not full 40-hex; fail closed"
|
||||
)
|
||||
elif pinned != live:
|
||||
reasons.append(
|
||||
"live head SHA no longer equals the pinned head; re-pin and "
|
||||
"re-validate before mutation"
|
||||
)
|
||||
|
||||
base_pinned = _normalize_ref(recheck.get("pinned_base_ref"))
|
||||
base_live = _normalize_ref(recheck.get("live_base_ref"))
|
||||
if not base_pinned or not base_live:
|
||||
reasons.append("base refs missing from live-state recheck; fail closed")
|
||||
elif base_pinned != base_live:
|
||||
reasons.append(
|
||||
f"base branch changed from '{base_pinned}' to '{base_live}'"
|
||||
)
|
||||
|
||||
blocking = recheck.get("blocking_change_requests")
|
||||
if blocking is None:
|
||||
reasons.append(
|
||||
"blocking review state not checked; fail closed"
|
||||
)
|
||||
elif blocking:
|
||||
reasons.append(
|
||||
"an undismissed REQUEST_CHANGES / blocking review state remains "
|
||||
"unresolved"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
return {"proven": proven, "block": not proven, "reasons": reasons}
|
||||
|
||||
|
||||
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
|
||||
justification=None):
|
||||
"""Assess reviewer/author role separation for blind queue workflows.
|
||||
|
||||
Issue #175 blocks a reviewer queue task from silently becoming author
|
||||
implementation work. The workflow may use both namespaces only when that
|
||||
mixed use is explicit, justified, and non-mutating; author mutations after
|
||||
a reviewer queue task require an explicit operator authorization.
|
||||
implementation work. Issue #179 also requires reviewer workflows to
|
||||
report namespace use and justify any foreign namespace calls. This helper
|
||||
accepts both forms:
|
||||
|
||||
*proof* keys:
|
||||
``task_role`` ('reviewer' or 'author'), ``task_kind`` (for example
|
||||
'blind_pr_queue_review'), ``reviewer_namespace_used``,
|
||||
``author_namespace_used``, ``author_mutations`` (list), ``review_mutations``
|
||||
(list), ``operator_authorized_author_work``, ``mixed_namespace_justification``,
|
||||
``scratch_evidence_claimed``, and ``scratch_evidence_durable``.
|
||||
- the #175 dict proof with mutation details, or
|
||||
- the #179 keyword form: ``task_role``, ``namespaces_used``,
|
||||
``justification``.
|
||||
"""
|
||||
proof = proof or {}
|
||||
if proof is None:
|
||||
namespaces_reported = namespaces_used is not None
|
||||
namespaces = list(namespaces_used or [])
|
||||
proof = {
|
||||
"task_role": task_role,
|
||||
"reviewer_namespace_used": any(
|
||||
"reviewer" in (namespace or "").lower()
|
||||
for namespace in namespaces
|
||||
),
|
||||
"author_namespace_used": any(
|
||||
"author" in (namespace or "").lower()
|
||||
for namespace in namespaces
|
||||
),
|
||||
"mixed_namespace_justification": justification,
|
||||
"author_mutations": [],
|
||||
"review_mutations": [],
|
||||
"_namespaces_used": namespaces,
|
||||
"_namespaces_reported": namespaces_reported,
|
||||
}
|
||||
else:
|
||||
proof = dict(proof or {})
|
||||
|
||||
task_role = (proof.get("task_role") or "").strip().lower()
|
||||
task_kind = (proof.get("task_kind") or "").strip().lower()
|
||||
author_mutations = list(proof.get("author_mutations") or [])
|
||||
@@ -364,6 +509,8 @@ def assess_role_boundary(proof):
|
||||
|
||||
if task_role not in {"reviewer", "author"}:
|
||||
reasons.append("task role missing or unknown; role boundary unproven")
|
||||
if proof.get("_namespaces_reported") is False:
|
||||
reasons.append("namespaces used were not reported; fail closed")
|
||||
|
||||
if task_role == "reviewer":
|
||||
if author_mutations and not authorized:
|
||||
@@ -409,18 +556,35 @@ def assess_role_boundary(proof):
|
||||
status = "clean"
|
||||
safe_next_action = "proceed"
|
||||
|
||||
namespaces = proof.get("_namespaces_used")
|
||||
if namespaces is None:
|
||||
namespaces = []
|
||||
if reviewer_used:
|
||||
namespaces.append("gitea-reviewer")
|
||||
if author_used:
|
||||
namespaces.append("gitea-author")
|
||||
foreign = [
|
||||
namespace for namespace in namespaces
|
||||
if task_role and task_role not in (namespace or "").lower()
|
||||
]
|
||||
|
||||
return {
|
||||
"status": status,
|
||||
"clean": status == "clean",
|
||||
"proven": status == "clean",
|
||||
"reasons": reasons,
|
||||
"violations": violations,
|
||||
"safe_next_action": safe_next_action,
|
||||
"foreign_namespaces": foreign,
|
||||
"justified": bool(mixed_justification),
|
||||
}
|
||||
|
||||
|
||||
def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
identity_eligible, merge_performed,
|
||||
issue_status_verified, role_boundary=None):
|
||||
issue_status_verified,
|
||||
capability_evidence=None, sweep=None, live_state=None,
|
||||
role_boundary=None):
|
||||
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
|
||||
|
||||
Combines the individual proof verdicts into the final-report fields the
|
||||
@@ -431,6 +595,13 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
- 'downgraded' — one or more proofs missing/weak; do not merge.
|
||||
- 'blocked' — a *violation*: a merge was claimed although the proofs
|
||||
did not allow one.
|
||||
|
||||
#179 raises the A bar: the report must also carry exact capability
|
||||
evidence (``assess_capability_evidence``), an exact secret/provenance
|
||||
sweep (``assess_sweep_evidence``), a pre-mutation live-state recheck
|
||||
(``assess_live_state_recheck`` — also required for ``merge_allowed``),
|
||||
and a clean role boundary (``assess_role_boundary``). Omitting any of
|
||||
them downgrades; a merge without the live recheck is a violation.
|
||||
"""
|
||||
contamination_status = contamination.get("status", "unknown")
|
||||
checkout_proven = bool(checkout_proof.get("proven"))
|
||||
@@ -443,6 +614,29 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
}
|
||||
role_status = role_boundary.get("status", "warning")
|
||||
|
||||
capability_evidence = capability_evidence or {
|
||||
"proven": False,
|
||||
"reasons": ["capability evidence not provided (#179)"],
|
||||
}
|
||||
sweep = sweep or {
|
||||
"verdict": "missing",
|
||||
"proven": False,
|
||||
"reasons": ["secret/provenance sweep evidence not provided (#179)"],
|
||||
}
|
||||
live_state = live_state or {
|
||||
"proven": False,
|
||||
"block": True,
|
||||
"reasons": ["pre-mutation live-state recheck not provided (#179)"],
|
||||
}
|
||||
role_boundary = role_boundary or {
|
||||
"proven": False,
|
||||
"reasons": ["role-boundary/namespace usage not reported (#179)"],
|
||||
}
|
||||
capability_proven = bool(capability_evidence.get("proven"))
|
||||
sweep_proven = bool(sweep.get("proven"))
|
||||
live_state_proven = bool(live_state.get("proven"))
|
||||
role_boundary_clean = bool(role_boundary.get("proven"))
|
||||
|
||||
downgrade_reasons = []
|
||||
if not identity_eligible:
|
||||
downgrade_reasons.append("identity/profile not eligible for review")
|
||||
@@ -466,6 +660,23 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
downgrade_reasons.extend(role_boundary.get("reasons", []))
|
||||
if not issue_status_verified:
|
||||
downgrade_reasons.append("linked issue status not verified")
|
||||
if not capability_proven:
|
||||
downgrade_reasons.append("exact capability evidence missing (#179)")
|
||||
downgrade_reasons.extend(capability_evidence.get("reasons", []))
|
||||
if not sweep_proven:
|
||||
downgrade_reasons.append(
|
||||
f"secret/provenance sweep evidence is "
|
||||
f"{sweep.get('verdict', 'missing')} (#179)"
|
||||
)
|
||||
downgrade_reasons.extend(sweep.get("reasons", []))
|
||||
if not live_state_proven:
|
||||
downgrade_reasons.append(
|
||||
"pre-mutation live-state recheck missing or failed (#179)"
|
||||
)
|
||||
downgrade_reasons.extend(live_state.get("reasons", []))
|
||||
if not role_boundary_clean:
|
||||
downgrade_reasons.append("role/namespace boundary not clean (#179)")
|
||||
downgrade_reasons.extend(role_boundary.get("reasons", []))
|
||||
|
||||
merge_allowed = (
|
||||
identity_eligible
|
||||
@@ -474,6 +685,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
and role_status == "clean"
|
||||
and validation_claimable
|
||||
and validation.get("verdict") != "invalid"
|
||||
# #179: no merge without a proven final live-state recheck.
|
||||
and live_state_proven
|
||||
)
|
||||
|
||||
violations = []
|
||||
@@ -508,6 +721,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
"merge_allowed": merge_allowed,
|
||||
"merge_performed": bool(merge_performed),
|
||||
"issue_status_verified": bool(issue_status_verified),
|
||||
"capability_evidence_proven": capability_proven,
|
||||
"sweep_verdict": sweep.get("verdict"),
|
||||
"live_state_recheck_proven": live_state_proven,
|
||||
"role_boundary_clean": role_boundary_clean,
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user