Compare commits

...
Author SHA1 Message Date
sysadminandClaude Opus 4.8 9274eebfaf feat: enforce capability stop terminal mode after reviewer denial (Issue #197)
Enter terminal mode when review_pr/merge_pr capability is denied. Block
reviewer queue tools (list_prs, eligibility checks) and add report-purity
validators for forbidden PR selection, fallback, and empty-queue claims.

Closes #197

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-05 17:17:16 -04:00
sysadmin 87397230e8 Merge pull request 'Raise reviewer A-bar: capability, sweep, live-state, and role-boundary proofs (Issue #179)' (#193) from feat/issue-179-reviewer-proof-tightening into master 2026-07-05 16:17:02 -05:00
sysadminandClaude Fable 5 e2247fab85 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]>
2026-07-05 16:42:03 -04:00
8 changed files with 980 additions and 27 deletions
+202
View File
@@ -0,0 +1,202 @@
"""Hard-stop terminal mode after reviewer capability denial (#197)."""
from __future__ import annotations
import re
TERMINAL_REPORT_HEADING = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed."
)
REVIEWER_CAPABILITY_TASKS = frozenset({
"review_pr",
"merge_pr",
"blind_pr_queue_review",
"request_changes_pr",
"approve_pr",
})
BLOCKED_QUEUE_TOOLS = frozenset({
"list_prs",
"check_pr_eligibility",
"view_pr",
"submit_pr_review",
"dry_run_pr_review",
"merge_pr",
"review_pr",
})
_session_terminal: dict | None = None
def enter_from_capability_result(capability: dict) -> dict | None:
"""Enter terminal mode when a reviewer/merge task is denied."""
global _session_terminal
task = (capability or {}).get("requested_task", "")
required_role = (capability or {}).get("required_role_kind")
if not capability.get("stop_required"):
return None
if required_role != "reviewer" and task not in REVIEWER_CAPABILITY_TASKS:
return None
record = {
"active": True,
"requested_task": task,
"required_role_kind": required_role,
"active_profile": capability.get("active_profile"),
"active_identity": capability.get("active_identity"),
"stop_required": True,
"exact_safe_next_action": capability.get("exact_safe_next_action"),
"terminal_message": TERMINAL_REPORT_HEADING,
}
_session_terminal = record
return dict(record)
def enter_from_route_result(route: dict) -> dict | None:
"""Enter terminal mode from a role router wrong_role_stop (#206 compat)."""
if (route or {}).get("route_result") != "wrong_role_stop":
return None
if route.get("required_role") != "reviewer":
return None
return enter_from_capability_result({
"requested_task": route.get("task_type"),
"required_role_kind": "reviewer",
"stop_required": True,
"active_profile": route.get("active_profile"),
"active_identity": None,
"exact_safe_next_action": route.get("message"),
})
def is_active() -> bool:
return bool(_session_terminal and _session_terminal.get("active"))
def active_record() -> dict | None:
if not is_active():
return None
return dict(_session_terminal)
def clear():
global _session_terminal
_session_terminal = None
def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
"""Return (allowed, reasons). False when terminal mode blocks queue work."""
if not is_active():
return True, []
name = (tool_name or "").strip().lower().removeprefix("gitea_")
if name in BLOCKED_QUEUE_TOOLS:
return False, [
TERMINAL_REPORT_HEADING,
f"Reviewer queue tool '{tool_name}' is blocked after "
"capability denial (fail closed).",
"Relaunch a reviewer MCP namespace to perform reviewer work.",
]
return True, []
def validate_eligibility_wording(text: str) -> tuple[bool, list[str]]:
"""Reject session-based eligibility reasoning (#197)."""
lower = (text or "").lower()
violations = []
if "not authored by this session" in lower:
violations.append(
"eligibility must use authenticated account identity, not "
"'this session' wording"
)
if re.search(r"not (?:self-)?authored by (?:the )?session", lower):
violations.append("session-based eligibility reasoning is invalid")
return (len(violations) == 0), violations
def assess_capability_stop_report(
report_text: str,
*,
trust_gate_status: str | None = None,
capability_denied: bool = True,
) -> dict:
"""Validate final report purity after reviewer capability denial."""
text = report_text or ""
lower = text.lower()
violations = []
if capability_denied and TERMINAL_REPORT_HEADING.lower() not in lower:
violations.append("missing required terminal report heading")
forbidden_patterns = [
("pr selection", re.compile(
r"selected pr|pr #\d+ (?:to review|selected)|eligible pr|"
r"next pr to review", re.I)),
("sibling repo inventory", re.compile(
r"sibling repo|other repo|mcp-control-plane|gitea-tools and", re.I)),
("author fallback", re.compile(
r"rebase conflicted|author-side fallback|have me rebase|"
r"implement the fix|push a branch|open a pr for", re.I)),
("invalid session eligibility", re.compile(
r"not authored by this session", re.I)),
]
for label, pattern in forbidden_patterns:
if pattern.search(text):
violations.append(f"forbidden after hard stop: {label}")
empty_queue_patterns = re.compile(
r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
r"inventory empty",
re.I,
)
if empty_queue_patterns.search(text):
if trust_gate_status != "trusted_empty":
violations.append(
"empty-queue claim after capability stop without "
"pr_inventory_trust_gate.status == trusted_empty"
)
ok, elig_violations = validate_eligibility_wording(text)
violations.extend(elig_violations)
if violations:
return {
"pure": False,
"downgraded": True,
"violations": violations,
"reasons": violations,
}
return {
"pure": True,
"downgraded": False,
"violations": [],
"reasons": [],
}
def build_terminal_report(capability: dict) -> dict:
"""Minimal allowed report fields after hard stop."""
return {
"terminal_mode": True,
"heading": TERMINAL_REPORT_HEADING,
"authenticated_profile": capability.get("active_profile"),
"authenticated_identity": capability.get("active_identity"),
"denied_task": capability.get("requested_task"),
"required_role_kind": capability.get("required_role_kind"),
"stop_required": capability.get("stop_required"),
"required_action": capability.get("exact_safe_next_action"),
"mutations_performed": False,
"allowed_sections": [
"authenticated identity/profile",
"denied capability result",
"reason task cannot proceed",
"required reviewer profile/identity",
"mutation confirmation (none)",
],
"forbidden_sections": [
"PR selection",
"sibling-repo queue recommendations",
"author-side fallback suggestions",
"empty-queue claims without trusted_empty",
"session-based eligibility wording",
],
}
+45 -1
View File
@@ -45,6 +45,7 @@ from gitea_auth import ( # noqa: E402
)
import gitea_audit # noqa: E402
import gitea_config # noqa: E402
import capability_stop_terminal # noqa: E402
def _reveal_endpoints() -> bool:
@@ -406,6 +407,11 @@ def gitea_list_prs(
'mergeable', 'updated_at' ('url' only with the reveal opt-in).
The additional 'updated_at' aids stale/conflicting queue detection.
"""
allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
"list_prs"
)
if not allowed:
raise RuntimeError("; ".join(block_reasons))
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls?state={state}"
@@ -517,6 +523,17 @@ def gitea_check_pr_eligibility(
'permission_report' (#142).
"""
action = (action or "").strip().lower()
if action in ("review", "approve", "request_changes", "merge"):
allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
"check_pr_eligibility"
)
if not allowed:
return {
"eligible": False,
"requested_action": action,
"reasons": block_reasons,
"terminal_mode": True,
}
profile = get_profile()
result = {
"eligible": False,
@@ -3476,7 +3493,7 @@ def gitea_resolve_task_capability(
"STOP: the active profile cannot perform the requested task; "
"follow exact_safe_next_action instead of improvising.")
return {
result = {
"requested_task": task,
"required_operation_permission": required_permission,
"required_role_kind": required_role,
@@ -3491,6 +3508,33 @@ def gitea_resolve_task_capability(
"different_mcp_namespace_required": different_namespace_required,
"exact_safe_next_action": next_safe_action,
}
if stop_required:
terminal = capability_stop_terminal.enter_from_capability_result(result)
if terminal:
result["terminal_mode"] = True
result["terminal_report"] = (
capability_stop_terminal.build_terminal_report(result)
)
return result
@mcp.tool()
def gitea_capability_stop_terminal_report() -> dict:
"""Read-only: terminal report template after reviewer capability denial (#197)."""
record = capability_stop_terminal.active_record()
if not record:
return {
"terminal_mode": False,
"reasons": ["capability stop terminal mode is not active"],
}
return capability_stop_terminal.build_terminal_report({
"requested_task": record.get("requested_task"),
"required_role_kind": record.get("required_role_kind"),
"active_profile": record.get("active_profile"),
"active_identity": record.get("active_identity"),
"stop_required": record.get("stop_required"),
"exact_safe_next_action": record.get("exact_safe_next_action"),
})
# ── Entry point ───────────────────────────────────────────────────────────────
+245 -12
View File
@@ -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,
}
@@ -635,6 +852,22 @@ def assess_controller_handoff(report_text, role=None):
}
def assess_capability_stop_terminal_report(
report_text,
*,
trust_gate_status=None,
capability_denied=True,
):
"""Issue #197: reports after reviewer capability denial must stay pure."""
from capability_stop_terminal import assess_capability_stop_report
return assess_capability_stop_report(
report_text,
trust_gate_status=trust_gate_status,
capability_denied=capability_denied,
)
# ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
#
# A reviewer agent may not convert an empty PR list response into a definitive
+24 -8
View File
@@ -232,19 +232,35 @@ Worktree folder = branch with `/` replaced by `-`
validation result after the command has completed and its output has
been read (`review_proofs.assess_validation_report`).
10. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
11. The final report must distinguish (`review_proofs.build_final_report`):
11. **#179 A-bar proofs** (all fail closed when missing —
`review_proofs.assess_capability_evidence`, `assess_sweep_evidence`,
`assess_live_state_recheck`, `assess_role_boundary`):
- Capability claims must cite exact `gitea_resolve_task_capability`
output (or runtime context); a bare "capability checks passed" is
downgraded.
- The secret/provenance sweep must state the exact command/script/
pattern/named method and the scope scanned.
- Immediately before submitting a review verdict (and again before any
merge), re-read live PR state and prove: still open, live head ==
pinned head, base unchanged, no unresolved blocking review state.
- Reviewer runs stay in the reviewer namespace; any author-namespace
call requires an explicit justification in the report.
12. The final report must distinguish (`review_proofs.build_final_report`):
identity eligible; PR author different from reviewer; session
contamination absent (with evidence); role boundary clean; validation
performed on the pinned head; merge performed; issue status verified. If
any proof is missing, stop or downgrade the result instead of merging
confidently.
contamination absent (with evidence); validation performed on the pinned
head; capability evidence; sweep verdict; live-state recheck; role
boundary; merge performed; issue status verified. If any proof is
missing, stop or downgrade the result instead of merging confidently.
## G. Merge / cleanup workflow
Only an eligible (non-author) reviewer merges. Before merging: always verify
the authenticated identity **and** the PR author; respect runtime profile
gates; run independent validation (do not trust the author's reported
results); and merge with a **pinned head SHA** and, where supported, the
the authenticated identity **and** the PR author; cite exact capability
evidence for merge_pr (#179); respect runtime profile gates; run independent
validation (do not trust the author's reported results); perform the **final
live-state recheck** (#179 — PR still open, live head == pinned head, base
unchanged, no unresolved blocking review state) immediately before the merge
mutation; and merge with a **pinned head SHA** and, where supported, the
**expected changed-file set**, so a moved head or widened diff refuses the
merge. After a real merge:
@@ -20,10 +20,21 @@ Steps:
*If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
2. Verify authenticated identity + active profile.
3. Confirm PR #<pr>: author (not you), state open, mergeable, review approved. Check if PR body uses `Closes #N` or `Fixes #N`; if it uses `Implements #N` or `Refs #N`, manual closing will be needed in step 29.
4. If any gate fails → STOP and report.
4. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
optionally pinning the reviewed head SHA / changed-file set.
5. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
4. Capability evidence (#179): cite the exact gitea_resolve_task_capability
output (or runtime context) proving merge_pr is allowed — a bare
"capability checks passed" claim is downgraded.
5. Final live-state recheck (#179), immediately before the merge mutation —
re-read the live PR and prove:
- PR still open
- live head SHA still equals the pinned/reviewed head SHA
- base branch unchanged
- no undismissed REQUEST_CHANGES / blocking review state remains
If any recheck fails → STOP, re-pin, re-validate.
6. If any gate fails → STOP and report.
7. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
pinning the reviewed head SHA (expected_head_sha) and, where supported,
the changed-file set.
8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
*Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
Then run the cleanup template (worktree-cleanup.md):
@@ -36,6 +36,11 @@ Steps:
- Target task role: reviewer identity (must NOT be the PR author)
*If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
2. Verify your authenticated identity (whoami) and the active profile.
Capability evidence (#179): cite the exact gitea_resolve_task_capability
output (or runtime context) for review_pr (and merge_pr if merging later);
a bare "capability checks passed" claim is downgraded. Stay in the
reviewer namespace: any author-namespace call must be justified in the
report (#179).
3. Fetch the PR facts: PR author, head SHA, state (must be open), base branch.
Pin the head SHA in your notes; every later step validates THAT SHA.
4. If authenticated user == PR author → STOP (no self-review).
@@ -58,12 +63,23 @@ Steps:
If HEAD does not match the pinned head → STOP before review/merge.
7. Confirm the worktree is clean. Inspect the FULL diff; confirm scope matches
issue #<n>; flag any unrelated files, secrets, or formatting churn. Check that the PR body correctly uses Gitea-closing keywords (`Closes #N` or `Fixes #N`) instead of non-closing ones (`Implements #N`, `Refs #N`).
Secret/provenance sweep must be exact (#179): state the exact command,
script, grep pattern, or named sweep method AND the scope scanned (e.g.
`git diff prgs/master...HEAD | grep -inE '<pattern>'`); "checked the diff
for secrets" alone is downgraded.
8. Run the test suite; report the exact command and exact results — pass/fail
plus passed/skipped/failed counts, any ignored paths and why they are safe
to ignore, and whether the command differs from the repository's canonical
validation command. Only claim a result after the output has been read.
9. Post the review verdict: approve only if scope is clean and checks pass;
otherwise request changes with specifics. Never merge from this review step.
9. Final live-state recheck (#179), immediately before submitting the review
verdict — re-read the live PR and prove:
- PR still open
- live head SHA still equals the pinned head SHA from step 3
- base branch unchanged
- no undismissed REQUEST_CHANGES / blocking review state left unaccounted
If anything moved → STOP, re-pin, re-validate before any verdict.
10. Post the review verdict: approve only if scope is clean and checks pass;
otherwise request changes with specifics. Never merge from this review step.
Include a "Review Metadata" block (attribution only — docs/llm-agent-sha.md):
Review Metadata:
+164
View File
@@ -0,0 +1,164 @@
"""Tests for capability stop terminal mode (#197)."""
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 capability_stop_terminal
import gitea_config
import mcp_server
from review_proofs import assess_capability_stop_terminal_report
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"prgs-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "jcwalker3",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.pr.create",
"gitea.branch.push",
],
"forbidden_operations": [
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
],
"execution_profile": "prgs-author",
},
},
"rules": {"allow_runtime_switching": False},
}
class TestCapabilityStopTerminal(unittest.TestCase):
def setUp(self):
capability_stop_terminal.clear()
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {
"host": "gitea.example.com",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
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))
def tearDown(self):
self._remotes.stop()
capability_stop_terminal.clear()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir.cleanup()
def _env(self):
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": "prgs-author",
"GITEA_TOKEN_AUTHOR": "author-pass",
}
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_review_pr_stop_enters_terminal_mode(self, _auth, _api):
with patch.dict(os.environ, self._env()):
res = mcp_server.gitea_resolve_task_capability(
task="review_pr", remote="prgs"
)
self.assertTrue(res["stop_required"])
self.assertTrue(res.get("terminal_mode"))
self.assertTrue(capability_stop_terminal.is_active())
self.assertIn(
capability_stop_terminal.TERMINAL_REPORT_HEADING,
res["terminal_report"]["heading"],
)
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_list_prs_blocked_after_capability_stop(self, _auth, _api):
with patch.dict(os.environ, self._env()):
mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_list_prs(remote="prgs")
self.assertIn("Cannot perform reviewer task", str(ctx.exception))
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_eligibility_check_blocked_after_stop(self, _auth, _api):
with patch.dict(os.environ, self._env()):
mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
res = mcp_server.gitea_check_pr_eligibility(
pr_number=193, action="review", remote="prgs"
)
self.assertFalse(res["eligible"])
self.assertTrue(res.get("terminal_mode"))
def test_report_with_pr_selection_impure(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Selected PR #193 for review anyway."
)
result = assess_capability_stop_terminal_report(report)
self.assertFalse(result["pure"])
def test_session_eligibility_wording_blocked(self):
ok, violations = capability_stop_terminal.validate_eligibility_wording(
"PR 193 is not authored by this session so it is eligible."
)
self.assertFalse(ok)
self.assertTrue(violations)
def test_rebase_fallback_blocked_in_report(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Or have me rebase conflicted PR 193."
)
result = assess_capability_stop_terminal_report(report)
self.assertFalse(result["pure"])
self.assertTrue(
any("author fallback" in v for v in result["violations"])
)
def test_empty_queue_without_trusted_empty_blocked(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"The repo has 0 open PRs."
)
result = assess_capability_stop_terminal_report(
report, trust_gate_status="untrusted_empty"
)
self.assertFalse(result["pure"])
def test_pure_terminal_report_passes(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Identity: jcwalker3 / prgs-author.\n"
"Required: prgs-reviewer.\n"
"No mutations performed."
)
result = assess_capability_stop_terminal_report(report)
self.assertTrue(result["pure"])
if __name__ == "__main__":
unittest.main()
+267
View File
@@ -21,10 +21,13 @@ import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from review_proofs import ( # noqa: E402
assess_capability_evidence,
assess_controller_handoff,
assess_inventory_completeness,
assess_live_state_recheck,
assess_role_boundary,
assess_self_review_contamination,
assess_sweep_evidence,
assess_validation_report,
build_final_report,
pr_inventory_trust_gate,
@@ -116,6 +119,63 @@ def _good_role_boundary():
)
def _good_capability_evidence():
return assess_capability_evidence([
{
"task": "review_pr",
"allowed": True,
"evidence_source": (
"gitea_resolve_task_capability(review_pr) output: "
"allowed_in_current_session=true, profile prgs-reviewer"
),
},
{
"task": "merge_pr",
"allowed": True,
"evidence_source": (
"gitea_resolve_task_capability(merge_pr) output: "
"allowed_in_current_session=true, profile prgs-reviewer"
),
},
])
def _good_sweep(**overrides):
sweep = {
"command": (
"git diff prgs/master...HEAD | grep -inE "
"'password|token|secret|api[_-]?key|authorization|bearer|https?://'"
),
"scope": "full PR diff against prgs/master",
"clean": True,
}
sweep.update(overrides)
return assess_sweep_evidence(sweep)
def _good_live_state(**overrides):
recheck = {
"pr_state": "open",
"pinned_head_sha": PINNED,
"live_head_sha": PINNED,
"pinned_base_ref": "master",
"live_base_ref": "master",
"blocking_change_requests": False,
}
recheck.update(overrides)
return assess_live_state_recheck(recheck)
def _good_role_boundary_179(**overrides):
kwargs = {
"task_role": "reviewer",
"namespaces_used": ["gitea-reviewer"],
"justification": None,
}
kwargs.update(overrides)
return assess_role_boundary(**kwargs)
class TestCheckoutProof(unittest.TestCase):
"""Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
@@ -466,6 +526,9 @@ class TestFinalReport(unittest.TestCase):
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
"capability_evidence": _good_capability_evidence(),
"sweep": _good_sweep(),
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
}
kwargs.update(overrides)
@@ -879,5 +942,209 @@ class TestPRInventoryTrustGate(unittest.TestCase):
self.assertTrue(res["corroborated"])
class TestCapabilityEvidence(unittest.TestCase):
"""#179 gap 1: capability claims need exact evidence."""
def test_evidence_backed_claims_are_proven(self):
result = _good_capability_evidence()
self.assertTrue(result["proven"])
self.assertEqual(result["reasons"], [])
def test_claim_without_evidence_source_is_not_proven(self):
result = assess_capability_evidence([
{"task": "review_pr", "allowed": True, "evidence_source": ""},
])
self.assertFalse(result["proven"])
self.assertTrue(any("evidence" in r.lower() for r in result["reasons"]))
def test_no_claims_at_all_fails_closed(self):
result = assess_capability_evidence([])
self.assertFalse(result["proven"])
def test_disallowed_task_is_not_proven(self):
result = assess_capability_evidence([
{
"task": "merge_pr",
"allowed": False,
"evidence_source": "gitea_resolve_task_capability output",
},
])
self.assertFalse(result["proven"])
class TestSweepEvidence(unittest.TestCase):
"""#179 gap 2: secret/provenance sweep must be exact."""
def test_exact_sweep_is_proven(self):
result = _good_sweep()
self.assertEqual(result["verdict"], "exact")
self.assertTrue(result["proven"])
def test_vague_sweep_without_command_is_downgraded(self):
result = _good_sweep(command="")
self.assertEqual(result["verdict"], "vague")
self.assertFalse(result["proven"])
def test_sweep_without_scope_is_downgraded(self):
result = _good_sweep(scope="")
self.assertEqual(result["verdict"], "vague")
self.assertFalse(result["proven"])
def test_missing_sweep_fails_closed(self):
result = assess_sweep_evidence(None)
self.assertEqual(result["verdict"], "missing")
self.assertFalse(result["proven"])
def test_unstated_result_is_downgraded(self):
result = _good_sweep(clean=None)
self.assertFalse(result["proven"])
class TestLiveStateRecheck(unittest.TestCase):
"""#179 gap 3: explicit pre-mutation live-state recheck."""
def test_clean_recheck_is_proven(self):
result = _good_live_state()
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_missing_recheck_fails_closed(self):
result = assess_live_state_recheck(None)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_closed_pr_blocks(self):
result = _good_live_state(pr_state="closed")
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_moved_head_blocks(self):
result = _good_live_state(live_head_sha=OTHER)
self.assertFalse(result["proven"])
self.assertTrue(any("head" in r.lower() for r in result["reasons"]))
def test_changed_base_blocks(self):
result = _good_live_state(live_base_ref="develop")
self.assertFalse(result["proven"])
def test_unresolved_blocking_reviews_block(self):
result = _good_live_state(blocking_change_requests=True)
self.assertFalse(result["proven"])
def test_unchecked_blocking_state_fails_closed(self):
result = _good_live_state(blocking_change_requests=None)
self.assertFalse(result["proven"])
class TestRoleBoundary179(unittest.TestCase):
"""#179 gap 4: reviewer flows avoid unjustified author-namespace use."""
def test_native_namespace_only_is_clean(self):
result = _good_role_boundary_179()
self.assertTrue(result["proven"])
def test_foreign_namespace_without_justification_is_downgraded(self):
result = _good_role_boundary_179(
namespaces_used=["gitea-reviewer", "gitea-author"]
)
self.assertFalse(result["proven"])
self.assertTrue(any("justif" in r.lower() for r in result["reasons"]))
def test_foreign_namespace_with_justification_is_clean(self):
result = _good_role_boundary_179(
namespaces_used=["gitea-reviewer", "gitea-author"],
justification=(
"author namespace read-only whoami used to evidence "
"self-review contamination status"
),
)
self.assertTrue(result["proven"])
def test_unreported_namespaces_fail_closed(self):
result = _good_role_boundary_179(namespaces_used=None)
self.assertFalse(result["proven"])
class TestFinalReport179Bar(unittest.TestCase):
"""#179 acceptance adds capability, sweep, live-state, and role proofs."""
def _report(self, **overrides):
kwargs = {
"checkout_proof": _good_checkout(),
"inventory": _good_inventory(),
"validation": _good_validation(),
"contamination": _good_contamination(),
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
"capability_evidence": _good_capability_evidence(),
"sweep": _good_sweep(),
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
def test_all_179_proofs_present_is_grade_a(self):
report = self._report()
self.assertEqual(report["grade"], "A")
self.assertTrue(report["capability_evidence_proven"])
self.assertEqual(report["sweep_verdict"], "exact")
self.assertTrue(report["live_state_recheck_proven"])
self.assertTrue(report["role_boundary_clean"])
def test_missing_capability_evidence_downgrades(self):
report = self._report(capability_evidence=None)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["capability_evidence_proven"])
def test_unevidenced_capability_claim_downgrades(self):
report = self._report(
capability_evidence=assess_capability_evidence([
{"task": "review_pr", "allowed": True, "evidence_source": ""},
])
)
self.assertNotEqual(report["grade"], "A")
def test_vague_sweep_downgrades(self):
report = self._report(sweep=_good_sweep(command=""))
self.assertNotEqual(report["grade"], "A")
self.assertEqual(report["sweep_verdict"], "vague")
def test_missing_sweep_downgrades(self):
report = self._report(sweep=None)
self.assertNotEqual(report["grade"], "A")
def test_missing_live_state_recheck_downgrades_and_blocks_merge(self):
report = self._report(live_state=None)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
self.assertFalse(report["live_state_recheck_proven"])
def test_stale_live_state_blocks_merge(self):
report = self._report(live_state=_good_live_state(live_head_sha=OTHER))
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
def test_merge_claim_without_live_recheck_is_a_violation(self):
report = self._report(live_state=None, merge_performed=True)
self.assertEqual(report["grade"], "blocked")
self.assertTrue(report["violations"])
def test_unjustified_author_namespace_downgrades(self):
report = self._report(
role_boundary=_good_role_boundary_179(
namespaces_used=["gitea-reviewer", "gitea-author"]
)
)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["role_boundary_clean"])
def test_positive_baseline_from_173_still_holds(self):
report = self._report()
self.assertTrue(report["inventory_complete"])
self.assertTrue(report["validated_on_pinned_head"])
if __name__ == "__main__":
unittest.main()