Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

fix: resolve conflicts for PR #378

Browse Source 
Merge prgs/master and keep both review-merge state machine (#290) and
native MCP preference (#270) imports, skill entries, and MCP tools.
This commit is contained in:
sysadmin
2026-07-07 09:57:01 -04:00
parent 8d32af14d6 47b4ee4791
commit af064dd06b
27 changed files with 4646 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+346 -1
View File
@@ -476,8 +476,10 @@ import task_capability_map # noqa: E402
import review_proofs # noqa: E402
import agent_temp_artifacts
import issue_lock_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import review_merge_state_machine # noqa: E402
import native_mcp_preference # noqa: E402
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
@@ -4091,6 +4093,14 @@ _GUIDE_RULES = {
"states forbid downstream approve/merge. infra_stop and capability "
"blocks stop immediately. Blocked recovery handoffs must restart the "
"full workflow — never replay approve/merge commands (#290)."),
"native_mcp_preference": (
"Gitea mutations must use native MCP tools first. When MCP is "
"available, block shell scripts, direct API calls, WebFetch, "
"Playwright, helper encoders, and profile-secret reads unless "
"explicit recovery-mode proof is complete. If MCP transport is "
"broken or the shell circuit breaker trips, emit a terminal recovery "
"report instead of improvising unsafe fallback. Reviewers must not "
"touch or restart MCP server files/processes (#270)."),
}
_COMMON_WORKFLOWS = [
@@ -4855,6 +4865,68 @@ def gitea_assess_review_merge_state_machine(
return result
@mcp.tool()
def gitea_assess_gitea_operation_path(
task: str,
path_kind: str | None = None,
command_or_detail: str | None = None,
mcp_available: bool = True,
mcp_tool_visible: bool = True,
recovery_mode: bool = False,
recovery_proof_complete: bool = False,
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Read-only: assess whether a proposed Gitea mutation path is allowed (#270).
Native MCP must be preferred when available. Shell scripts, direct API
calls, WebFetch, Playwright, and helper encoders are blocked unless
explicit recovery-mode proof is supplied.
"""
profile = get_profile()
role = _role_kind(
profile.get("allowed_operations") or [],
profile.get("forbidden_operations") or [],
)
return native_mcp_preference.assess_gitea_operation_path(
task=task,
path_kind=path_kind,
command_or_detail=command_or_detail,
mcp_available=mcp_available,
mcp_tool_visible=mcp_tool_visible,
recovery_mode=recovery_mode,
recovery_proof_complete=recovery_proof_complete,
role=role,
active_profile=profile.get("profile_name"),
)
@mcp.tool()
def gitea_record_shell_spawn_outcome(
exit_code: int | None = None,
stdout: str = "",
stderr: str = "",
spawn_failure: bool | None = None,
probe_attempted: bool = False,
probe_succeeded: bool | None = None,
) -> dict:
"""Record shell spawn outcome and update the session circuit breaker (#270)."""
return native_mcp_preference.record_shell_spawn_outcome(
exit_code=exit_code,
stdout=stdout,
stderr=stderr,
spawn_failure=spawn_failure,
probe_attempted=probe_attempted,
probe_succeeded=probe_succeeded,
)
@mcp.tool()
def gitea_get_shell_health() -> dict:
"""Read-only: return shell spawn circuit-breaker state for this MCP session (#270)."""
return native_mcp_preference.shell_health_status()
@mcp.tool()
def gitea_get_runtime_context(
remote: str = "dadeschools",
@@ -4973,6 +5045,7 @@ def gitea_get_runtime_context(
"preflight_block_reasons": preflight["preflight_block_reasons"],
"preflight_workspace": preflight.get("preflight_workspace"),
"session_capabilities": session_capabilities,
"shell_health": native_mcp_preference.shell_health_status(),
}
if reveal and h:
@@ -5215,6 +5288,41 @@ def gitea_audit_config() -> dict:
return report
def _post_structured_issue_comment(
*,
issue_number: int,
body: str,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
audit_op: str = "create_issue_comment",
) -> dict:
"""Post an issue-thread comment after permission gates already passed."""
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
with _audited(
audit_op,
host=h,
remote=remote,
org=o,
repo=r,
issue_number=issue_number,
request_metadata={"body_chars": len(body)},
):
data = api_request("POST", api, auth, {"body": body})
result = {
"success": True,
"performed": True,
"comment_id": data["id"],
"issue_number": issue_number,
}
if _reveal_endpoints():
result["url"] = data.get("html_url")
return result
@mcp.tool()
def gitea_mark_issue(
issue_number: int,
@@ -5224,6 +5332,8 @@ def gitea_mark_issue(
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
branch_name: str | None = None,
profile: str | None = None,
) -> dict:
"""Claim or release an issue via the status:in-progress label.
@@ -5273,7 +5383,31 @@ def gitea_mark_issue(
request_metadata={"op": "add", "label": "status:in-progress"}):
api_request("POST", f"{base}/issues/{issue_number}/labels", auth,
{"labels": [label_id]})
return {"success": True, "message": f"Issue #{issue_number} claimed."}
active_profile = (profile or get_profile().get("profile_name") or "unknown")
branch = (branch_name or "pending").strip() or "pending"
heartbeat_body = issue_claim_heartbeat.format_heartbeat_body(
kind="claim",
issue_number=issue_number,
branch=branch,
phase="claimed",
profile=active_profile,
next_action="create worktree and begin implementation",
)
heartbeat = _post_structured_issue_comment(
issue_number=issue_number,
body=heartbeat_body,
remote=remote,
host=host,
org=org,
repo=repo,
audit_op="claim_heartbeat",
)
return {
"success": True,
"message": f"Issue #{issue_number} claimed.",
"heartbeat_posted": heartbeat.get("success", False),
"heartbeat_comment_id": heartbeat.get("comment_id"),
}
else:
with _audited("unlabel_issue", host=h, remote=remote, org=o, repo=r,
issue_number=issue_number,
@@ -5283,6 +5417,217 @@ def gitea_mark_issue(
return {"success": True, "message": f"Issue #{issue_number} released."}
@mcp.tool()
def gitea_post_heartbeat(
issue_number: int,
branch: str,
phase: str,
pr: str = "none",
next_action: str = "none",
blocker: str = "none",
profile: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Post a structured progress heartbeat on a claimed issue (#268)."""
blocked = _profile_permission_block(
task_capability_map.required_permission("post_heartbeat"))
if blocked:
return blocked
verify_preflight_purity(remote)
active_profile = profile or get_profile().get("profile_name")
body = issue_claim_heartbeat.format_heartbeat_body(
kind="progress",
issue_number=issue_number,
branch=branch,
phase=phase,
profile=active_profile,
pr=pr,
next_action=next_action,
blocker=blocker,
)
return _post_structured_issue_comment(
issue_number=issue_number,
body=body,
remote=remote,
host=host,
org=org,
repo=repo,
audit_op="progress_heartbeat",
)
@mcp.tool()
def gitea_reconcile_issue_claims(
state: str = "open",
stale_after_hours: int = 24,
heartbeat_lease_minutes: int = 30,
limit: int = 100,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only inventory of issue claims and heartbeat lease status (#268)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
issues = api_get_all(f"{base}/issues?state={state}&type=issues", auth, limit=limit)
open_prs = api_get_all(f"{base}/pulls?state=open", auth)
branches = api_get_all(f"{base}/branches", auth, limit=limit)
branch_names = [b.get("name") for b in branches if b.get("name")]
comments_by_issue: dict[int, list[dict]] = {}
for issue in issues:
if not issue_claim_heartbeat.issue_has_in_progress_label(issue):
continue
number = int(issue["number"])
api = f"{base}/issues/{number}/comments"
comments_by_issue[number] = api_request("GET", api, auth) or []
reclaim_after_minutes = max(heartbeat_lease_minutes * 2, 60)
inventory = issue_claim_heartbeat.build_claim_inventory(
issues=issues,
comments_by_issue=comments_by_issue,
open_prs=open_prs,
branch_names=branch_names,
heartbeat_lease_minutes=heartbeat_lease_minutes,
reclaim_after_minutes=reclaim_after_minutes,
)
inventory["cleanup_plan"] = issue_claim_heartbeat.build_cleanup_plan(inventory)
inventory["success"] = True
inventory["performed"] = False
return inventory
@mcp.tool()
def gitea_cleanup_stale_claims(
dry_run: bool = True,
execute_confirmed: bool = False,
heartbeat_lease_minutes: int = 30,
limit: int = 100,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Propose or execute stale-claim cleanup for phantom/reclaimable issues (#268)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"performed": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
inventory = gitea_reconcile_issue_claims(
heartbeat_lease_minutes=heartbeat_lease_minutes,
limit=limit,
remote=remote,
host=host,
org=org,
repo=repo,
)
if not inventory.get("success"):
return inventory
plan = [
entry
for entry in inventory.get("cleanup_plan") or []
if entry.get("action") == "remove_status_in_progress_and_comment"
]
report = {
"success": True,
"dry_run": dry_run,
"performed": False,
"planned_actions": plan,
"inventory_counts": inventory.get("counts"),
}
if dry_run:
return report
if not execute_confirmed:
raise ValueError(
"execute_confirmed must be True when dry_run=False (fail closed)"
)
blocked = _profile_permission_block(
task_capability_map.required_permission("cleanup_stale_claims"))
if blocked:
return blocked
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
labels = api_request("GET", f"{base}/labels?limit=100", auth)
label_id = next(
(lb["id"] for lb in labels if lb.get("name") == "status:in-progress"),
None,
)
if label_id is None:
raise RuntimeError("Label 'status:in-progress' not found")
actions: list[dict] = []
active_profile = get_profile().get("profile_name")
for entry in plan:
issue_number = int(entry["issue_number"])
with _audited(
"unlabel_issue",
host=h,
remote=remote,
org=o,
repo=r,
issue_number=issue_number,
request_metadata={"op": "cleanup_stale_claim", "label": "status:in-progress"},
):
api_request(
"DELETE",
f"{base}/issues/{issue_number}/labels/{label_id}",
auth,
)
cleanup_body = issue_claim_heartbeat.format_heartbeat_body(
kind="cleanup",
issue_number=issue_number,
branch="none",
phase="stale-claim-cleanup",
profile=active_profile,
next_action="issue reclaimable by queue",
blocker=entry.get("status") or "stale",
)
comment = _post_structured_issue_comment(
issue_number=issue_number,
body=cleanup_body,
remote=remote,
host=host,
org=org,
repo=repo,
audit_op="cleanup_heartbeat",
)
actions.append(
{
"issue_number": issue_number,
"label_removed": True,
"cleanup_comment_id": comment.get("comment_id"),
}
)
report["performed"] = True
report["actions"] = actions
return report
@mcp.tool()
def gitea_list_labels(
remote: str = "dadeschools",
issue_claim_heartbeat.py
+295
View File
@@ -0,0 +1,295 @@
"""Issue claim heartbeat leases and stale-claim reconciliation (#268).
Structured issue-thread comments prove live ownership beyond the
``status:in-progress`` label alone. Queue inventory can classify claims as
active, stale, reclaimable, PR-backed, or phantom.
"""
from __future__ import annotations
import re
from datetime import datetime, timedelta, timezone
from typing import Any
MARKER = "<!-- gitea-issue-claim-heartbeat:v1 -->"
IN_PROGRESS_LABEL = "status:in-progress"
_KIND_CLAIM = "claim"
_KIND_PROGRESS = "progress"
_KIND_CLEANUP = "cleanup"
_FIELD_RE = re.compile(
r"^\s*-\s*([a-z_]+)\s*:\s*(.+?)\s*$",
re.IGNORECASE | re.MULTILINE,
)
_ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
def _parse_timestamp(value: str | None) -> datetime | None:
if not value:
return None
text = value.strip()
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is None:
return parsed.replace(tzinfo=timezone.utc)
return parsed
def format_heartbeat_body(
*,
kind: str,
issue_number: int,
branch: str,
phase: str,
profile: str | None = None,
pr: str = "none",
next_action: str = "none",
blocker: str = "none",
) -> str:
"""Return a structured, machine-parseable issue comment body."""
profile_value = (profile or "unknown").strip() or "unknown"
lines = [
MARKER,
"**Issue claim heartbeat**",
f"- kind: {kind}",
f"- issue: #{issue_number}",
f"- branch: {branch}",
f"- phase: {phase}",
f"- profile: {profile_value}",
f"- pr: {pr}",
f"- blocker: {blocker}",
f"- next_action: {next_action}",
]
return "\n".join(lines)
def parse_heartbeat_comment(body: str) -> dict[str, Any] | None:
"""Parse one structured heartbeat comment, or None when not a heartbeat."""
text = body or ""
if MARKER not in text:
return None
fields: dict[str, str] = {}
for match in _FIELD_RE.finditer(text):
fields[match.group(1).strip().lower()] = match.group(2).strip()
if not fields:
return None
issue_raw = fields.get("issue", "")
issue_digits = re.sub(r"[^\d]", "", issue_raw)
issue_number = int(issue_digits) if issue_digits.isdigit() else None
return {
"kind": fields.get("kind"),
"issue_number": issue_number,
"branch": fields.get("branch"),
"phase": fields.get("phase"),
"profile": fields.get("profile"),
"pr": fields.get("pr"),
"blocker": fields.get("blocker"),
"next_action": fields.get("next_action"),
"raw_fields": fields,
}
def extract_issue_heartbeats(
comments: list[dict],
*,
issue_number: int | None = None,
) -> list[dict]:
"""Return parsed heartbeats newest-last, optionally filtered to *issue_number*."""
heartbeats: list[dict] = []
for comment in comments or []:
parsed = parse_heartbeat_comment(comment.get("body") or "")
if not parsed:
continue
if issue_number is not None and parsed.get("issue_number") != issue_number:
continue
heartbeats.append(
{
**parsed,
"comment_id": comment.get("id"),
"author": (comment.get("user") or {}).get("login")
or comment.get("author"),
"created_at": comment.get("created_at"),
"updated_at": comment.get("updated_at"),
}
)
return heartbeats
def issue_has_in_progress_label(issue: dict) -> bool:
labels = issue.get("labels") or []
for label in labels:
name = label if isinstance(label, str) else label.get("name")
if name == IN_PROGRESS_LABEL:
return True
return False
def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None:
pattern = f"issue-{issue_number}"
closes = f"closes #{issue_number}"
fixes = f"fixes #{issue_number}"
for pr in open_prs or []:
head = (pr.get("head") or {}).get("ref") or ""
text = f"{pr.get('title', '')} {pr.get('body', '')}".lower()
if pattern in head.lower():
return pr
if closes in text or fixes in text:
return pr
return None
def _matching_branch_names(issue_number: int, branch_names: list[str]) -> list[str]:
pattern = f"issue-{issue_number}"
return [name for name in branch_names if pattern in (name or "").lower()]
def classify_issue_claim(
*,
issue: dict,
comments: list[dict],
open_prs: list[dict] | None = None,
branch_names: list[str] | None = None,
now: datetime | None = None,
heartbeat_lease_minutes: int = 30,
reclaim_after_minutes: int = 60,
) -> dict[str, Any]:
"""Classify one issue's claim state from labels, heartbeats, PRs, and branches."""
issue_number = int(issue.get("number") or 0)
open_prs = open_prs or []
branch_names = branch_names or []
current = now or datetime.now(timezone.utc)
has_label = issue_has_in_progress_label(issue)
heartbeats = extract_issue_heartbeats(comments, issue_number=issue_number)
latest = heartbeats[-1] if heartbeats else None
latest_at = _parse_timestamp(
(latest or {}).get("updated_at") or (latest or {}).get("created_at")
)
age_minutes = None
if latest_at:
age_minutes = int((current - latest_at).total_seconds() // 60)
linked_pr = _linked_open_pr(issue_number, open_prs)
matching_branches = _matching_branch_names(issue_number, branch_names)
if not has_label:
status = "not_claimed"
reasons = ["issue lacks status:in-progress label"]
elif linked_pr:
status = "awaiting_review"
reasons = [f"open PR #{linked_pr.get('number')} covers this issue"]
elif not heartbeats:
status = "phantom"
reasons = [
"status:in-progress label present without structured claim heartbeat"
]
elif latest_at is None:
status = "phantom"
reasons = ["heartbeat comments present but timestamps could not be parsed"]
elif age_minutes is not None and age_minutes <= heartbeat_lease_minutes:
status = "active"
reasons = [f"heartbeat age {age_minutes}m within {heartbeat_lease_minutes}m lease"]
elif matching_branches and age_minutes is not None and age_minutes <= reclaim_after_minutes:
status = "active"
reasons = [
f"matching branch(es) {', '.join(matching_branches)} with heartbeat "
f"age {age_minutes}m"
]
elif age_minutes is not None and age_minutes > reclaim_after_minutes and not matching_branches:
status = "reclaimable"
reasons = [
f"no heartbeat within {reclaim_after_minutes}m and no matching branch"
]
elif age_minutes is not None and age_minutes > heartbeat_lease_minutes:
status = "stale"
reasons = [
f"heartbeat age {age_minutes}m exceeds {heartbeat_lease_minutes}m lease"
]
else:
status = "active"
reasons = ["claim has structured heartbeat proof"]
return {
"issue_number": issue_number,
"title": issue.get("title"),
"status": status,
"has_in_progress_label": has_label,
"heartbeat_count": len(heartbeats),
"latest_heartbeat": latest,
"heartbeat_age_minutes": age_minutes,
"linked_open_pr": linked_pr.get("number") if linked_pr else None,
"matching_branches": matching_branches,
"reasons": reasons,
"reclaimable": status == "reclaimable",
"stale": status in {"stale", "phantom", "reclaimable"},
}
def build_claim_inventory(
*,
issues: list[dict],
comments_by_issue: dict[int, list[dict]],
open_prs: list[dict],
branch_names: list[str],
now: datetime | None = None,
heartbeat_lease_minutes: int = 30,
reclaim_after_minutes: int = 60,
) -> dict[str, Any]:
"""Build a queue inventory report for in-progress issue claims."""
entries: list[dict] = []
for issue in issues or []:
if not issue_has_in_progress_label(issue):
continue
number = int(issue["number"])
entry = classify_issue_claim(
issue=issue,
comments=comments_by_issue.get(number, []),
open_prs=open_prs,
branch_names=branch_names,
now=now,
heartbeat_lease_minutes=heartbeat_lease_minutes,
reclaim_after_minutes=reclaim_after_minutes,
)
entries.append(entry)
counts: dict[str, int] = {}
for entry in entries:
counts[entry["status"]] = counts.get(entry["status"], 0) + 1
return {
"entries": entries,
"counts": counts,
"heartbeat_lease_minutes": heartbeat_lease_minutes,
"reclaim_after_minutes": reclaim_after_minutes,
"in_progress_total": len(entries),
}
def build_cleanup_plan(inventory: dict[str, Any]) -> list[dict]:
"""Return stale/reclaimable/phantom claims that may be cleaned up."""
plan: list[dict] = []
for entry in inventory.get("entries") or []:
if entry.get("status") in {"reclaimable", "phantom"}:
plan.append(
{
"issue_number": entry["issue_number"],
"status": entry["status"],
"action": "remove_status_in_progress_and_comment",
"reasons": list(entry.get("reasons") or []),
}
)
elif entry.get("status") == "stale" and not entry.get("linked_open_pr"):
plan.append(
{
"issue_number": entry["issue_number"],
"status": entry["status"],
"action": "report_only",
"reasons": list(entry.get("reasons") or []),
}
)
return plan
native_mcp_preference.py
+369
View File
@@ -0,0 +1,369 @@
"""Native MCP preference gate and shell health circuit breaker (#270).
Gitea mutations must use native MCP tools first. Shell scripts, direct API
calls, browser helpers, and improvised encoders are blocked when MCP is
available unless explicit recovery-mode proof is supplied.
"""
from __future__ import annotations
import re
from typing import Any
from reviewer_fallback import LOCAL_GITEA_SCRIPT_NAMES
# Tasks that must prefer native MCP over shell/API/helper fallbacks.
GITEA_MUTATION_TASKS = frozenset({
"comment_issue",
"mark_issue",
"lock_issue",
"set_issue_labels",
"create_issue",
"close_issue",
"create_branch",
"push_branch",
"create_pr",
"close_pr",
"comment_pr",
"review_pr",
"merge_pr",
"approve_pr",
"request_changes_pr",
"commit_files",
"gitea_commit_files",
"delete_branch",
"address_pr_change_requests",
})
ALLOWED_PATH_KINDS = frozenset({
"mcp_native",
"recovery_fallback",
})
BLOCKED_PATH_KINDS = frozenset({
"shell_script",
"direct_api",
"webfetch",
"playwright",
"helper_script",
"unsafe_helper",
"mcp_server_touch",
})
SHELL_SPAWN_FAILURE_THRESHOLD = 2
TERMINAL_REPORT_HEADING = (
"MCP transport unavailable or shell circuit breaker tripped. "
"No unsafe Gitea fallback performed."
)
_RECOVERY_MODE_RE = re.compile(
r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|"
r"mcp tools unavailable|no mcp path)\b",
re.IGNORECASE,
)
_LOCAL_SCRIPT_RE = re.compile(
r"(?:^|[\s\"'`/])(?:python3?|bash|sh)?\s*(?:"
+ "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES)
+ r")",
re.IGNORECASE,
)
_WEBFETCH_RE = re.compile(r"\b(?:webfetch|mcp_web_fetch|fetch\s+url)\b", re.IGNORECASE)
_PLAYWRIGHT_RE = re.compile(r"\b(?:playwright|browser_navigate|browser_click)\b", re.IGNORECASE)
_HELPER_SCRIPT_RE = re.compile(
r"(?:^|[\s\"'`/])(?:_encode_|_emit_|_inline_)[\w-]+\.py",
re.IGNORECASE,
)
_MANUAL_BASE64_RE = re.compile(
r"\b(?:manual\s+base64|llm-generated\s+base64|base64-encode\s+in\s+chat)\b",
re.IGNORECASE,
)
_DIRECT_API_RE = re.compile(
r"\b(?:api_request|urllib\.request|requests\.(?:post|patch|put)|curl\s+-X\s+POST)\b",
re.IGNORECASE,
)
_MCP_SERVER_TOUCH_RE = re.compile(
r"\b(?:kill|pkill|restart|reload|touch|edit|modify|write)\b.{0,40}\b(?:mcp_server|mcp-server|gitea_mcp_server)\b",
re.IGNORECASE,
)
_CLI_AUTH_DIVERGENCE_RE = re.compile(
r"GITEA_MCP_PROFILE\s*=\s*['\"]?([^\s'\";]+)",
re.IGNORECASE,
)
_session_shell: dict[str, Any] = {
"consecutive_spawn_failures": 0,
"shell_unavailable": False,
"hard_stopped": False,
"last_exit_code": None,
}
def _clean(value: str | None) -> str:
return (value or "").strip()
def is_spawn_failure(
*,
exit_code: int | None = None,
stdout: str | None = None,
stderr: str | None = None,
spawn_failure: bool | None = None,
) -> bool:
"""True for executor spawn failures (exit_code -1, empty output)."""
if spawn_failure is True:
return True
if spawn_failure is False:
return False
if exit_code != -1:
return False
return not (_clean(stdout) or _clean(stderr))
def record_shell_spawn_outcome(
*,
exit_code: int | None = None,
stdout: str | None = None,
stderr: str | None = None,
spawn_failure: bool | None = None,
probe_attempted: bool = False,
probe_succeeded: bool | None = None,
) -> dict[str, Any]:
"""Track shell spawn outcomes and trip the session circuit breaker (#270 AC4)."""
global _session_shell
failed = is_spawn_failure(
exit_code=exit_code,
stdout=stdout,
stderr=stderr,
spawn_failure=spawn_failure,
)
if failed:
_session_shell["consecutive_spawn_failures"] = (
int(_session_shell.get("consecutive_spawn_failures") or 0) + 1
)
_session_shell["last_exit_code"] = exit_code
if probe_attempted and probe_succeeded is False:
_session_shell["shell_unavailable"] = True
else:
_session_shell["consecutive_spawn_failures"] = 0
_session_shell["shell_unavailable"] = False
_session_shell["hard_stopped"] = False
_session_shell["last_exit_code"] = exit_code
failures = int(_session_shell["consecutive_spawn_failures"])
if failures >= SHELL_SPAWN_FAILURE_THRESHOLD:
_session_shell["shell_unavailable"] = True
_session_shell["hard_stopped"] = True
return shell_health_status()
def shell_health_status() -> dict[str, Any]:
"""Return the current shell health / circuit-breaker state."""
failures = int(_session_shell.get("consecutive_spawn_failures") or 0)
hard_stopped = bool(_session_shell.get("hard_stopped"))
shell_unavailable = bool(_session_shell.get("shell_unavailable"))
return {
"consecutive_spawn_failures": failures,
"shell_unavailable": shell_unavailable,
"hard_stopped": hard_stopped,
"threshold": SHELL_SPAWN_FAILURE_THRESHOLD,
"shell_use_allowed": not hard_stopped,
"last_exit_code": _session_shell.get("last_exit_code"),
"safe_next_action": (
"emit terminal recovery report; prefer native MCP for remaining Gitea mutations"
if hard_stopped
else (
"probe shell once with echo/pwd; if probe fails mark shell unavailable"
if failures == 1
else "shell healthy"
)
),
}
def clear_shell_health_for_tests() -> None:
"""Reset session shell state (tests only)."""
global _session_shell
_session_shell = {
"consecutive_spawn_failures": 0,
"shell_unavailable": False,
"hard_stopped": False,
"last_exit_code": None,
}
def classify_command_path(command_or_detail: str | None) -> str:
"""Classify a proposed command/detail into a path kind."""
text = command_or_detail or ""
if _MCP_SERVER_TOUCH_RE.search(text):
return "mcp_server_touch"
if _WEBFETCH_RE.search(text):
return "webfetch"
if _PLAYWRIGHT_RE.search(text):
return "playwright"
if _HELPER_SCRIPT_RE.search(text) or _MANUAL_BASE64_RE.search(text):
return "unsafe_helper"
if _LOCAL_SCRIPT_RE.search(text):
return "shell_script"
if _DIRECT_API_RE.search(text):
return "direct_api"
return "mcp_native"
def detect_cli_auth_divergence(
command_or_detail: str | None,
*,
active_profile: str | None,
) -> list[str]:
"""Flag shell commands that override GITEA_MCP_PROFILE away from the session."""
reasons: list[str] = []
text = command_or_detail or ""
active = _clean(active_profile)
match = _CLI_AUTH_DIVERGENCE_RE.search(text)
if not match or not active:
return reasons
requested = _clean(match.group(1))
if requested and requested != active:
reasons.append(
f"CLI auth divergence: command sets GITEA_MCP_PROFILE='{requested}' "
f"but active session profile is '{active}' (fail closed)"
)
return reasons
def format_mcp_unavailable_terminal_report(
*,
task: str | None = None,
reasons: list[str] | None = None,
) -> str:
"""Terminal report when MCP is broken and unsafe recovery is forbidden (#270 AC3)."""
lines = [TERMINAL_REPORT_HEADING, ""]
if task:
lines.append(f"Blocked task: {task}")
if reasons:
lines.extend(["", "Reasons:"])
lines.extend(f"- {reason}" for reason in reasons)
lines.extend([
"",
"Required recovery (no improvised fallback):",
"- Restart the MCP session",
"- Kill hung background terminals holding the shell executor",
"- Retry the native MCP tool once after reconnect",
"- If MCP remains unavailable, stop and hand off to the operator",
"- Do not run local Gitea scripts, WebFetch, Playwright, or manual base64 encoders",
])
return "\n".join(lines)
def assess_gitea_operation_path(
*,
task: str,
path_kind: str | None = None,
command_or_detail: str | None = None,
mcp_available: bool = True,
mcp_tool_visible: bool = True,
recovery_mode: bool = False,
recovery_proof_complete: bool = False,
role: str | None = None,
active_profile: str | None = None,
) -> dict[str, Any]:
"""Fail closed when a Gitea mutation would bypass native MCP (#270)."""
task_name = _clean(task)
resolved_kind = _clean(path_kind) or classify_command_path(command_or_detail)
reasons: list[str] = []
if task_name and task_name not in GITEA_MUTATION_TASKS:
reasons.append(
f"unknown or non-mutation Gitea task '{task_name}'; "
"native MCP preference gate applies only to Gitea mutations"
)
shell = shell_health_status()
if shell["hard_stopped"] and resolved_kind != "mcp_native":
reasons.append(
"shell circuit breaker is hard-stopped after consecutive spawn failures; "
"use native MCP or emit terminal recovery report"
)
recovery_declared = recovery_mode or bool(
_RECOVERY_MODE_RE.search(command_or_detail or "")
)
recovery_allowed = (
recovery_declared
and recovery_proof_complete
and not mcp_available
and resolved_kind in BLOCKED_PATH_KINDS - {"mcp_server_touch"}
)
if resolved_kind in BLOCKED_PATH_KINDS and not recovery_allowed:
reasons.append(f"path kind '{resolved_kind}' is not an approved Gitea mutation path")
if resolved_kind == "mcp_server_touch":
if _clean(role) == "reviewer" or (
active_profile and "reviewer" in active_profile.lower()
):
reasons.append(
"reviewer workflows must not touch, restart, or kill MCP server files/processes"
)
else:
reasons.append(
"MCP server files/processes must not be modified during normal Gitea workflows"
)
reasons.extend(
detect_cli_auth_divergence(command_or_detail, active_profile=active_profile)
)
if (
mcp_available
and mcp_tool_visible
and resolved_kind != "mcp_native"
and not recovery_allowed
):
if not recovery_declared:
reasons.append(
"native MCP tools are available; shell/API/helper fallback is forbidden"
)
elif not recovery_proof_complete:
reasons.append(
"recovery-mode fallback requires complete recovery proof before proceeding"
)
if not mcp_available and resolved_kind != "mcp_native" and not recovery_allowed:
if not recovery_declared:
reasons.append(
"MCP transport unavailable; produce terminal recovery report instead of "
"improvising unsafe fallback"
)
block = bool(reasons)
terminal_report = None
if block and (not mcp_available or shell["hard_stopped"]):
terminal_report = format_mcp_unavailable_terminal_report(
task=task_name or None,
reasons=reasons,
)
return {
"task": task_name or None,
"path_kind": resolved_kind,
"mcp_available": mcp_available,
"mcp_tool_visible": mcp_tool_visible,
"recovery_mode": recovery_declared,
"shell_health": shell,
"block": block,
"allowed": not block,
"reasons": reasons,
"terminal_report": terminal_report,
"safe_next_action": (
"use the native MCP tool for this task"
if mcp_available and mcp_tool_visible and block
else (
terminal_report or "proceed with native MCP"
if block
else "proceed with native MCP"
)
),
}
review_proofs.py
+816
View File
@@ -1242,6 +1242,217 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None):
}
# ── Reconciliation controller-handoff schema (Issue #303) ────────────────────
#
# Already-landed PR reconciliation is neither author issue work nor
# reviewer merge work; its handoff needs its own schema. Stale
# author/reviewer fields fail validation, and observed git ref
# mutations (fetch) must be reported under the precise category.
RECONCILIATION_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED"
RECONCILIATION_HANDOFF_FIELDS = (
("Task", ("task",)),
("Repo", ("repo",)),
("Role/profile", ("role/profile", "role", "profile")),
("Identity", ("identity",)),
("Selected PR", ("selected pr",)),
("PR live state", ("pr live state",)),
("Candidate head SHA", ("candidate head sha",)),
("Target branch", ("target branch",)),
("Target branch SHA", ("target branch sha",)),
("Ancestor proof", ("ancestor proof",)),
("Linked issue", ("linked issue",)),
("Linked issue live status", ("linked issue live status",)),
("Eligibility class", ("eligibility class",)),
("Capabilities proven", ("capabilities proven",)),
("Missing capabilities", ("missing capabilities",)),
("PR comments posted", ("pr comments posted",)),
("Issue comments posted", ("issue comments posted",)),
("PRs closed", ("prs closed",)),
("Issues closed", ("issues closed",)),
("Git ref mutations", ("git ref mutations",)),
("Worktree mutations", ("worktree mutations",)),
("MCP/Gitea mutations", ("mcp/gitea mutations",)),
("External-state mutations", ("external-state mutations",)),
("Read-only diagnostics", ("read-only diagnostics",)),
("Blocker", ("blocker",)),
("Safe next action", ("safe next action",)),
("No review/merge confirmation", ("no review/merge",)),
)
RECONCILIATION_FORBIDDEN_FIELDS = (
"pr number opened",
"pinned reviewed head",
"scratch worktree used",
"workspace mutations",
"issue lock proof",
"claim/comment status",
)
def assess_reconciliation_handoff(report_text, observed_commands=None):
"""#303: validate the dedicated reconciliation handoff schema.
Requires a Controller Handoff section carrying every reconciliation
field, ``Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED``, and
no stale author/reviewer fields. *observed_commands* is the git
command log; an observed fetch/pull must be reported under ``Git ref
mutations`` (never claimed none).
Returns {'complete', 'downgraded', 'reasons'}; fails closed.
"""
section = _handoff_section_lines(report_text)
if section is None:
return {
"complete": False,
"downgraded": True,
"reasons": [
"reconciliation report has no section titled exactly "
f"'{HANDOFF_HEADING}'"
],
}
fields = {}
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
k, v = stripped.split(":", 1)
fields[k.strip().lower()] = v.strip()
reasons = []
for name, aliases in RECONCILIATION_HANDOFF_FIELDS:
if not any(label.startswith(alias)
for label in fields for alias in aliases):
reasons.append(
f"reconciliation handoff missing required field: {name}"
)
for stale in RECONCILIATION_FORBIDDEN_FIELDS:
if any(label.startswith(stale) for label in fields):
reasons.append(
f"reconciliation handoff must not include stale field "
f"'{stale}'"
)
eligibility = fields.get("eligibility class", "")
if eligibility and RECONCILIATION_ELIGIBILITY_CLASS not in eligibility.upper():
reasons.append(
"reconciliation handoff eligibility class must be "
f"'{RECONCILIATION_ELIGIBILITY_CLASS}' (got '{eligibility}')"
)
_, ref_cmds = _classify_git_commands(observed_commands)
if ref_cmds and _field_claims_none(fields, "git ref mutations"):
reasons.append(
"observed git ref mutation not reported under 'Git ref "
f"mutations': {ref_cmds[0]}"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
# ── Already-landed report wording (Issue #298) ───────────────────────────────
#
# An already-landed PR is reconciliation-only: it must never be called
# eligible, its head SHA must never be called reviewed, and the legacy
# eligible/reviewed/scratch-worktree handoff fields must not appear.
# Reviewed-head claims in any report require validation + diff review to
# have actually passed.
ALREADY_LANDED_FORBIDDEN_PHRASES = (
"oldest eligible pr",
"next eligible pr",
"pinned reviewed head",
"scratch worktree used",
)
ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED"
def assess_already_landed_report_wording(report_text, *, already_landed,
review_validated=False):
"""#298: reject eligible/reviewed wording for already-landed PRs.
*already_landed* states whether the already-landed gate fired for the
selected PR. *review_validated* states whether validation and diff
review actually passed this session; only then may a reviewed head
SHA be populated in a non-already-landed report.
Returns {'complete', 'downgraded', 'reasons'}; fails closed.
"""
text_lower = (report_text or "").lower()
fields = _report_labeled_fields(report_text)
reasons = []
reviewed_head = fields.get("reviewed head sha")
reviewed_head_populated = bool(
reviewed_head and not reviewed_head.strip().lower().startswith("none")
)
pinned_head_present = "pinned reviewed head" in fields
if already_landed:
for phrase in ALREADY_LANDED_FORBIDDEN_PHRASES:
if phrase in text_lower:
reasons.append(
f"already-landed report must not use '{phrase}'; the PR "
"is reconciliation-only, not review/merge eligible"
)
eligibility = fields.get("eligibility class", "")
if ALREADY_LANDED_ELIGIBILITY_CLASS not in eligibility.upper():
reasons.append(
"already-landed report missing 'Eligibility class: "
f"{ALREADY_LANDED_ELIGIBILITY_CLASS}'"
)
for required in ("oldest open pr requiring action",
"candidate head sha"):
if required not in fields:
reasons.append(
f"already-landed report missing '{required}' field"
)
if "reviewed head sha" not in fields:
reasons.append(
"already-landed report must state 'Reviewed head SHA: none'"
)
elif reviewed_head_populated:
reasons.append(
"already-landed report must not claim a reviewed head SHA; "
"no validation or diff review passed for this PR"
)
worktree_used = fields.get("review worktree used", "")
if worktree_used.strip().lower() not in ("false", "no"):
reasons.append(
"already-landed report must state 'Review worktree used: "
"false'"
)
elif not review_validated:
if pinned_head_present:
reasons.append(
"'Pinned reviewed head' populated before validation and "
"diff review passed"
)
if reviewed_head_populated:
reasons.append(
"reviewed head SHA populated before validation and diff "
"review passed"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"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.
@@ -1422,6 +1633,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None, worktree_proof=None,
validation_environment=None,
queue_ordering=None,
baseline_validation=None, project_root=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
@@ -1450,6 +1662,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
)
empty_queue_report = assess_empty_queue_report(report_text)
if validation_environment is None and report_text:
validation_env_proof = assess_validation_environment_proof(
report_text=report_text,
)
elif validation_environment is not None:
validation_env_proof = assess_validation_environment_proof(
report_text=report_text,
validation_environment=validation_environment,
)
else:
validation_env_proof = {
"proven": True,
"block": False,
"reasons": [],
"violations": [],
}
if queue_ordering is None and report_text:
queue_ordering_proof = assess_queue_ordering_proof(
report_text=report_text,
@@ -1476,6 +1704,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
if report_text
else {"proven": True, "block": False, "reasons": [], "violations": []}
)
inventory_worktree = (
assess_inventory_worktree_report(report_text)
if 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,
@@ -1617,6 +1850,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"empty-queue report missing or failed trust-gate proof (#198)"
)
downgrade_reasons.extend(empty_queue_report.get("reasons", []))
if not validation_env_proof.get("proven"):
downgrade_reasons.append(
"validation environment proof missing or failed (#311)"
)
downgrade_reasons.extend(validation_env_proof.get("reasons", []))
if not queue_ordering_proof.get("proven"):
downgrade_reasons.append(
"queue ordering proof missing or failed (#321)"
@@ -1632,6 +1870,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"queue-status report violates loaded workflow proof gates (#339)"
)
downgrade_reasons.extend(queue_status_report.get("reasons", []))
if not inventory_worktree.get("proven"):
downgrade_reasons.append(
"inventory pagination or worktree fields inconsistent (#293)"
)
downgrade_reasons.extend(inventory_worktree.get("reasons", []))
if not baseline_validation.get("proven"):
downgrade_reasons.append(
"reviewer baseline validation proof missing or failed (#325)"
@@ -1648,11 +1891,13 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
# #179: no merge without a proven final live-state recheck.
and live_state_proven
and worktree_proven
and validation_env_proof.get("proven")
and queue_ordering_proof.get("proven")
and baseline_validation.get("proven")
)
violations = []
violations.extend(validation_env_proof.get("violations", []))
violations.extend(queue_ordering_proof.get("violations", []))
violations.extend(baseline_validation.get("violations", []))
if merge_performed and not merge_allowed:
@@ -1706,6 +1951,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
else True
),
"empty_queue_trust_gate_status": empty_queue_report.get("status"),
"validation_environment_proven": bool(
validation_env_proof.get("proven")
),
"validation_environment_violations": list(
validation_env_proof.get("violations") or []
),
"queue_ordering_proven": bool(queue_ordering_proof.get("proven")),
"queue_ordering_policy": queue_ordering_proof.get("policy"),
"queue_ordering_violations": list(
@@ -1717,6 +1968,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"queue_status_violations": list(
queue_status_report.get("violations") or []
),
"inventory_worktree_proven": bool(inventory_worktree.get("proven")),
"inventory_worktree_violations": list(
inventory_worktree.get("reasons") or []
),
"baseline_validation_proven": bool(baseline_validation.get("proven")),
"baseline_validation_violations": list(
baseline_validation.get("violations") or []
@@ -3322,6 +3577,192 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict:
}
# ---------------------------------------------------------------------------
# Validation environment proof (#311)
# ---------------------------------------------------------------------------
_VALIDATION_CLAIM_RE = re.compile(
r"(?:validation\s+command|pytest\s+(?:executed|passed|failed)|"
r"\d+\s+passed|\d+\s+failed|tests?\s+passed)",
re.IGNORECASE,
)
_VALIDATION_CMD_LINE_RE = re.compile(
r"validation\s+command\s*:\s*(.+)",
re.IGNORECASE,
)
_VALIDATION_CWD_LINE_RE = re.compile(
r"working\s+directory\s*:\s*(\S+)",
re.IGNORECASE,
)
_WHICH_PYTEST_LINE_RE = re.compile(
r"which\s+pytest\s*:\s*(.+)",
re.IGNORECASE,
)
_PYTEST_VERSION_LINE_RE = re.compile(
r"pytest\s+--version\s*:\s*(.+)",
re.IGNORECASE,
)
_BARE_PYTEST_CMD_RE = re.compile(
r"^(?:python\s+-m\s+)?pytest(?:\s|$)",
re.IGNORECASE,
)
_VAGUE_PYTEST_CLAIM_RE = re.compile(
r"pytest\s+(?:executed|passed|failed)\b",
re.IGNORECASE,
)
def _parse_validation_command(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _VALIDATION_CMD_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(".")
return ""
def _parse_validation_cwd(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _VALIDATION_CWD_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(",.;")
return ""
def _is_bare_pytest_command(command: str) -> bool:
text = (command or "").strip()
if not text:
return False
if "/" in text or "\\" in text:
return False
return bool(_BARE_PYTEST_CMD_RE.match(text))
def assess_validation_environment_proof(
*,
report_text: str | None = None,
validation_environment: dict | None = None,
) -> dict:
"""#311: reviewer reports must include exact validation command/environment."""
text = report_text or ""
lower = text.lower()
env = validation_environment or {}
reasons: list[str] = []
violations: list[str] = []
claims_validation = bool(_VALIDATION_CLAIM_RE.search(text))
if not claims_validation and not env.get("command"):
return {
"proven": True,
"block": False,
"claims_validation": False,
"reasons": [],
"violations": [],
"safe_next_action": "proceed",
}
command = (env.get("command") or _parse_validation_command(text)).strip()
working_directory = (
env.get("working_directory") or env.get("cwd")
or _parse_validation_cwd(text)
).strip()
if not command:
reasons.append(
"validation result claimed without exact validation command (#311)"
)
elif _VAGUE_PYTEST_CLAIM_RE.search(text) and command == "":
violations.append(
"vague pytest summary without exact validation command (#311)"
)
if not working_directory:
reasons.append(
"validation result claimed without working directory (#311)"
)
has_result = any(
token in lower
for token in ("passed", "failed", "skipped", "error")
) or any(
env.get(key) is not None
for key in ("passed", "failed", "skipped", "result")
)
if not has_result:
reasons.append(
"validation result claimed without pass/fail result evidence (#311)"
)
if command and _is_bare_pytest_command(command):
which_pytest = (
(env.get("which_pytest") or env.get("executable_path") or "")
.strip()
)
if not which_pytest:
for line in text.splitlines():
match = _WHICH_PYTEST_LINE_RE.search(line)
if match:
which_pytest = match.group(1).strip()
break
if not which_pytest:
reasons.append(
"bare pytest command requires which-pytest/executable proof (#311)"
)
pytest_version = (env.get("pytest_version") or "").strip()
if not pytest_version:
for line in text.splitlines():
match = _PYTEST_VERSION_LINE_RE.search(line)
if match:
pytest_version = match.group(1).strip()
break
if not pytest_version:
reasons.append(
"bare pytest command requires pytest --version proof (#311)"
)
venv_resolved = env.get("venv_resolved")
if venv_resolved is None:
venv_resolved = any(
marker in lower
for marker in (
"project venv",
"venv/bin/pytest",
"resolves to the project venv",
"venv resolved: yes",
)
)
if not venv_resolved and which_pytest and "venv" not in which_pytest:
reasons.append(
"bare pytest command requires proof whether executable "
"resolves to the project venv (#311)"
)
proven = not reasons and not violations
return {
"proven": proven,
"block": bool(violations),
"claims_validation": True,
"command": command or None,
"working_directory": working_directory or None,
"reasons": reasons,
"violations": violations,
"safe_next_action": (
"report exact validation command, working directory, result, and "
"for bare pytest also which-pytest, pytest --version, and venv proof"
if not proven
else "proceed"
),
}
# ---------------------------------------------------------------------------
# Reviewer queue ordering proof (#321)
# ---------------------------------------------------------------------------
@@ -3652,6 +4093,337 @@ def assess_reviewer_baseline_validation_proof(
}
# ---------------------------------------------------------------------------
# Partial reconciliation policy (#302)
# ---------------------------------------------------------------------------
# Durable policy choice for already-landed PR reconciliation when
# ``gitea.pr.close`` is unavailable: Option B, comment-then-stop.
PARTIAL_RECONCILIATION_POLICY = "comment_then_stop"
PARTIAL_RECONCILIATION_COMMENT_FIELDS = (
"PR head SHA",
"target branch SHA",
"ancestor proof",
"linked issue status",
"required missing capability",
)
def resolve_partial_reconciliation_plan(
*,
ancestor_proven: bool,
capabilities: dict | None,
) -> dict:
"""#302: decide reconciliation mutations from proven capabilities.
Implements the comment-then-stop policy (Option B): with ancestor
proof but no ``close_pr`` capability, a reconciliation comment with
the full proof is posted when ``comment_pr`` is proven, then the
workflow stops for an authorized close. Missing comment capability
degrades to a recovery handoff with no Gitea mutation. Unproven
ancestry blocks every mutation regardless of capability.
*capabilities* maps ``close_pr``/``comment_pr``/``close_issue``/
``comment_issue`` to proven booleans; ``None`` or missing keys fail
closed.
"""
caps = capabilities or {}
if not ancestor_proven:
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "GATE_NOT_PROVEN",
"allowed_mutations": (),
"required_comment_fields": (),
"report_requirements": (
"state that ancestry was not proven and no Gitea mutation "
"was performed",
),
"reasons": [
"ancestor proof missing; no reconciliation mutation may "
"run (#302)"
],
"safe_next_action": (
"run the already-landed ancestry check before any "
"reconciliation mutation"
),
}
if caps.get("close_pr") is True:
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "FULL_RECONCILE_CLOSE_ALLOWED",
"allowed_mutations": ("close_pr", "comment_pr"),
"required_comment_fields": (),
"report_requirements": (
"report the PR close result",
),
"reasons": [],
"safe_next_action": (
"close the already-landed PR with the proven close_pr "
"capability and report the close result"
),
}
if caps.get("comment_pr") is True:
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "PARTIAL_RECONCILE_COMMENT_THEN_STOP",
"allowed_mutations": ("comment_pr",),
"required_comment_fields": PARTIAL_RECONCILIATION_COMMENT_FIELDS,
"report_requirements": (
"report that the reconciliation comment was posted",
"report the missing close capability that prevented the "
"PR close",
),
"reasons": [
"close_pr capability missing; policy is comment-then-stop "
"(#302)"
],
"safe_next_action": (
"post one reconciliation comment carrying the ancestor "
"proof, then stop for a human or authorized close"
),
}
return {
"policy": PARTIAL_RECONCILIATION_POLICY,
"outcome": "RECOVERY_HANDOFF_ONLY",
"allowed_mutations": (),
"required_comment_fields": (),
"report_requirements": (
"report that no reconciliation comment was posted and name "
"the missing capability",
"report that no Gitea mutation was performed",
),
"reasons": [
"close_pr and comment_pr capabilities missing; recovery "
"handoff only (#302)"
],
"safe_next_action": (
"produce a recovery handoff recording the intended comment "
"and required capabilities; perform no Gitea mutation"
),
}
def assess_partial_reconciliation_report(
report_text: str | None,
*,
plan: dict,
) -> dict:
"""#302: final reports must explain why comments were or were not posted."""
text = (report_text or "").lower()
outcome = (plan or {}).get("outcome", "")
reasons: list[str] = []
if outcome == "FULL_RECONCILE_CLOSE_ALLOWED":
if "close" not in text or "result" not in text:
reasons.append(
"full reconciliation report must state the PR close "
"result (#302)"
)
elif outcome == "PARTIAL_RECONCILE_COMMENT_THEN_STOP":
if "comment" not in text:
reasons.append(
"partial reconciliation report must state that the "
"reconciliation comment was posted (#302)"
)
if "capability" not in text and "close_pr" not in text:
reasons.append(
"partial reconciliation report must name the missing "
"close capability (#302)"
)
elif outcome == "RECOVERY_HANDOFF_ONLY":
if "comment" not in text or "capability" not in text:
reasons.append(
"recovery handoff report must explain that no comment was "
"posted and name the missing capability (#302)"
)
if "no gitea mutation" not in text and "no mutation" not in text:
reasons.append(
"recovery handoff report must state that no Gitea "
"mutation was performed (#302)"
)
return {
"complete": not reasons,
"block": bool(reasons),
"reasons": reasons,
}
# ---------------------------------------------------------------------------
# Already-landed review gate (#292)
# ---------------------------------------------------------------------------
ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED"
ALREADY_LANDED_HANDOFF_FIELDS = (
"PR number/title",
"candidate head SHA",
"target branch",
"target branch SHA",
"ancestor proof",
"linked issue status",
"recommended reconciliation action",
"capability proof for close/comment mutations",
)
_FORBIDDEN_LANDED_STATES_RE = re.compile(
r"review decision\s*:\s*approved?\b|"
r"merge result\s*:\s*merged\b|"
r"\bready[_ ]to[_ ]merge\b|"
r"\bmerge[d]?\s*:\s*success\b",
re.IGNORECASE,
)
def assess_already_landed_review_gate(
*,
pr_number: int | None,
candidate_head_sha: str | None,
target_branch: str | None,
target_branch_sha: str | None,
head_is_ancestor_of_target: bool | None,
live_head_sha: str | None = None,
real_blocker: bool = False,
mergeable: bool | None = None,
) -> dict:
"""#292: hard gate before any review mutation on an already-landed PR.
Runs after PR selection and head pinning, before approval, request-
changes, or the merge API. *head_is_ancestor_of_target* is the result
of an ancestry check of *candidate_head_sha* against the freshly
fetched target branch at *target_branch_sha* (e.g. ``git merge-base
--is-ancestor``). ``None`` means the check was not run — the gate then
fails closed.
*live_head_sha*, when provided, must equal *candidate_head_sha*; a
changed head invalidates the pinned ancestry result until re-checked.
*mergeable* is accepted for caller convenience but never consulted:
conflict/mergeability handling is a separate gate.
"""
del mergeable # ancestry gate only; mergeability is a separate gate
reasons: list[str] = []
if not isinstance(pr_number, int) or pr_number <= 0:
reasons.append("PR number missing or invalid (#292)")
if not _FULL_SHA.match((candidate_head_sha or "").strip()):
reasons.append(
"candidate head SHA is not a full 40-hex commit SHA (#292)"
)
if not (target_branch or "").strip():
reasons.append("target branch missing (#292)")
if not _FULL_SHA.match((target_branch_sha or "").strip()):
reasons.append(
"target branch SHA is not a full 40-hex commit SHA; fetch the "
"target branch and record its SHA before the ancestry check "
"(#292)"
)
if live_head_sha is not None and candidate_head_sha and (
live_head_sha.strip() != candidate_head_sha.strip()
):
reasons.append(
"PR head changed since the ancestry check was pinned; re-fetch "
"and re-run the already-landed gate (#292)"
)
if head_is_ancestor_of_target is None:
reasons.append(
"ancestry of the PR head against the target branch was not "
"checked; the already-landed gate must run before any review "
"mutation (#292)"
)
if reasons:
return {
"state": "GATE_NOT_PROVEN",
"approve_allowed": False,
"request_changes_allowed": False,
"merge_api_allowed": False,
"reconciliation_handoff_required": False,
"required_handoff_fields": (),
"reasons": reasons,
"safe_next_action": (
"fetch the target branch, pin the live PR head, run the "
"ancestry check, then re-run this gate"
),
}
if head_is_ancestor_of_target:
return {
"state": ALREADY_LANDED_STATE,
"approve_allowed": False,
"request_changes_allowed": bool(real_blocker),
"merge_api_allowed": False,
"reconciliation_handoff_required": True,
"required_handoff_fields": ALREADY_LANDED_HANDOFF_FIELDS,
"reasons": [
f"PR #{pr_number} head {candidate_head_sha} is already an "
f"ancestor of {target_branch} @ {target_branch_sha}; the PR "
"is reconciliation-only (#292)"
],
"safe_next_action": (
"stop before review mutation and emit an "
f"{ALREADY_LANDED_STATE} reconciliation handoff; any PR/issue "
"close or comment requires exact capability proof"
),
}
return {
"state": "NORMAL_REVIEW_CANDIDATE",
"approve_allowed": True,
"request_changes_allowed": True,
"merge_api_allowed": True,
"reconciliation_handoff_required": False,
"required_handoff_fields": (),
"reasons": [],
"safe_next_action": "proceed with the normal review workflow gates",
}
def assess_already_landed_report_state(
report_text: str | None,
*,
gate_fired: bool,
) -> dict:
"""#292: reject APPROVED/MERGED/READY_TO_MERGE after the gate fired.
*gate_fired* comes from the session's own gate result, so a report
that omits the already-landed markers entirely still fails closed —
unlike the text-only #327 wording rules this does not depend on the
report admitting the PR was already landed.
"""
if not gate_fired:
return {"complete": True, "block": False, "reasons": []}
text = report_text or ""
reasons: list[str] = []
if _FORBIDDEN_LANDED_STATES_RE.search(text):
reasons.append(
"report claims APPROVED/MERGED/READY_TO_MERGE although the "
"already-landed gate fired (#292)"
)
if ALREADY_LANDED_STATE.lower() not in text.lower():
reasons.append(
f"report must state {ALREADY_LANDED_STATE} when the "
"already-landed gate fired (#292)"
)
return {
"complete": not reasons,
"block": bool(reasons),
"reasons": reasons,
}
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# Full-suite failure approval gate (#323)
# ---------------------------------------------------------------------------
@@ -4237,6 +5009,20 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
}
def assess_merge_simulation_report(report_text, **kwargs):
"""#317: classify local merge simulation as worktree/index mutations."""
from reviewer_merge_simulation import assess_merge_simulation_report as _assess
return _assess(report_text, **kwargs)
def assess_validation_integrity_report(report_text, **kwargs):
"""#316: separate official PR-head validation from diagnostic experiments."""
from reviewer_validation_integrity import assess_validation_integrity_report as _assess
return _assess(report_text, **kwargs)
def assess_prior_blocker_skip_proof(report_text, **kwargs):
"""#318: require live blocker proof before skipping earlier open PRs."""
from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess
@@ -4251,6 +5037,36 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs):
return _assess(report_text, **kwargs)
def assess_validation_worktree_edit_report(report_text, **kwargs):
"""#315: block edits in PR validation worktrees; require diagnostic separation."""
from reviewer_validation_worktree_edits import (
assess_validation_worktree_edit_report as _assess,
)
return _assess(report_text, **kwargs)
def assess_inventory_worktree_report(report_text, **kwargs):
"""#293: prove inventory pagination and consistent worktree reporting."""
from reviewer_inventory_worktree import assess_inventory_worktree_report as _assess
return _assess(report_text, **kwargs)
def assess_infra_stop_handoff_report(report_text, **kwargs):
"""#289: repair handoff purity when infra_stop blocks reviewer capability."""
from reviewer_infra_stop_handoff import assess_infra_stop_handoff_report as _assess
return _assess(report_text, **kwargs)
def assess_mutation_categories_report(report_text, **kwargs):
"""#319: require precise mutation categories in reviewer controller handoffs."""
from reviewer_mutation_categories import assess_mutation_categories_report as _assess
return _assess(report_text, **kwargs)
def assess_worktree_ownership_report(report_text, **kwargs):
"""#312: prove session-owned or safe-reuse worktree before reset/validation."""
from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
reviewer_infra_stop_handoff.py
+162
View File
@@ -0,0 +1,162 @@
"""Infra-stop repair handoff verifier for blocked reviewer workflows (#289)."""
from __future__ import annotations
import re
from typing import Any
from capability_stop_terminal import (
TERMINAL_REPORT_HEADING,
assess_capability_stop_report,
)
_REPAIR_HANDOFF_RE = re.compile(
r"repair handoff|control-checkout repair mode",
re.IGNORECASE,
)
_CONTROL_REPAIR_RE = re.compile(r"control-checkout repair mode", re.IGNORECASE)
_PR_QUEUE_ADVANCE_RE = re.compile(
r"(?:selected pr|pr #\d+ (?:to review|selected)|eligible pr|"
r"next (?:eligible )?pr(?: to review)?|oldest eligible pr|pinned (?:review )?head)",
re.IGNORECASE,
)
_REVIEW_MUTATION_RE = re.compile(
r"(?:gitea_review_pr|gitea_merge_pr|request_changes|submitted\s+approve|"
r"merge result|review decision\s*:\s*approve)",
re.IGNORECASE,
)
_BACKGROUND_TOOL_RE = re.compile(r"\b(?:schedule|manage_task)\b", re.IGNORECASE)
_PR_REVIEW_REPORT_RE = re.compile(
r"(?:review summary|merge recommendation|validation result\s*:\s*pass|"
r"queue status report with selected pr)",
re.IGNORECASE,
)
_DIAGNOSTIC_FIELDS = {
"mcp process root": re.compile(r"mcp process root\s*:", re.IGNORECASE),
"inspected git root": re.compile(r"inspected git root\s*:", re.IGNORECASE),
"conflict marker path": re.compile(
r"(?:conflict marker path|exact conflict marker path)\s*:",
re.IGNORECASE,
),
"merge/rebase control path": re.compile(
r"(?:merge/rebase control path|exact merge/rebase control path)\s*:",
re.IGNORECASE,
),
"safe next repair action": re.compile(
r"safe next (?:repair )?action\s*:",
re.IGNORECASE,
),
}
def assess_infra_stop_handoff_report(
report_text: str,
*,
stop_session: dict | None = None,
) -> dict[str, Any]:
"""Validate repair handoff purity when infra_stop blocks reviewer capability (#289)."""
text = report_text or ""
session = dict(stop_session or {})
reasons: list[str] = []
infra_stop = bool(
session.get("infra_stop")
or session.get("infra_stop_blocked")
or session.get("route_result") == "infra_stop"
)
if not infra_stop:
return {
"proven": True,
"block": False,
"reasons": [],
"infra_stop": False,
"safe_next_action": "proceed",
}
lower = text.lower()
repair_handoff = bool(
_REPAIR_HANDOFF_RE.search(text)
or TERMINAL_REPORT_HEADING.lower() in lower
)
if not repair_handoff:
reasons.append(
"infra_stop requires a repair handoff, not a PR review report"
)
if _PR_REVIEW_REPORT_RE.search(text) and not _REPAIR_HANDOFF_RE.search(text):
reasons.append(
"final output must be a repair handoff instead of stale PR review state"
)
if _PR_QUEUE_ADVANCE_RE.search(text):
reasons.append(
"infra_stop blocks PR queue advancement; do not select or advance next PR"
)
if _REVIEW_MUTATION_RE.search(text):
reasons.append(
"review, approval, request-changes, merge, or comment mutations "
"forbidden while infra_stop is active"
)
if session.get("pinned_head_sha") or re.search(
r"pinned (?:review )?head sha\s*:", text, re.IGNORECASE
):
if session.get("capability_cleared") is not True:
reasons.append(
"cannot pin PR head SHA while review capability never cleared"
)
if session.get("main_checkout_diagnostics") and not _CONTROL_REPAIR_RE.search(text):
reasons.append(
"main-checkout diagnostics must be labeled CONTROL-CHECKOUT REPAIR MODE"
)
if _BACKGROUND_TOOL_RE.search(text):
reasons.append(
"background schedule/manage_task tools must not be used during "
"blocked infra_stop recovery"
)
assessment = session.get("infra_stop_assessment") or {}
if assessment.get("infra_stop") or infra_stop:
missing = [
label
for label, pattern in _DIAGNOSTIC_FIELDS.items()
if not pattern.search(text)
]
if missing and session.get("require_infra_diagnostics", True):
reasons.append(
"infra_stop repair handoff missing diagnostic fields: "
+ ", ".join(missing)
)
conflict_file = assessment.get("conflict_file")
if conflict_file and conflict_file.lower() not in lower:
reasons.append(
f"infra_stop assessment conflict file {conflict_file!r} "
"not reflected in repair handoff"
)
capability = assess_capability_stop_report(
text,
trust_gate_status=session.get("trust_gate_status"),
capability_denied=True,
)
if not capability.get("pure"):
reasons.extend(capability.get("reasons") or [])
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": list(dict.fromkeys(reasons)),
"infra_stop": True,
"repair_handoff": repair_handoff,
"safe_next_action": (
"stop PR queue work; emit CONTROL-CHECKOUT REPAIR MODE handoff "
"with infra diagnostics and safe next repair action"
if reasons
else "proceed"
),
}
reviewer_inventory_worktree.py
+308
View File
@@ -0,0 +1,308 @@
"""Reviewer inventory completeness and worktree state verifier (#293)."""
from __future__ import annotations
import re
from typing import Any
_PAGINATION_FINALITY_EVIDENCE = re.compile(
r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|"
r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|"
r"no next page|final[- ]page|pr_inventory_trust_gate\.status|"
r"total_count\s*:|pagination.*(?:final|complete)",
re.I,
)
_INCOMPLETE_PAGINATION_PROOF = re.compile(
r"first page only|partial page|page 1 only|truncated|"
r"returned \d+ open prs?(?:\s*$|\s*[,;])",
re.I,
)
_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile(
r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|"
r"default (?:gitea )?page[- ]?size|under (?:the )?50[- ]?(?:item )?limit|"
r"(?:complete|exhaustive).*(?:default )?page[- ]?size|"
r"page[- ]?size assumption|assumed complete",
re.I,
)
_ELIGIBILITY_CLAIM_RE = re.compile(
r"\b(?:oldest eligible pr|next eligible pr|next pr to review|"
r"inventory (?:is )?complete|inventory exhaustive|exhaustive inventory)\b",
re.I,
)
_EXACT_PAGE_LIMIT_RE = re.compile(
r"(?:returned|listed|fetched)\s+(\d+)\s+open prs?|"
r"open pr count\s*:\s*(\d+)|"
r"page[- ]?size\s*(?:=|:)\s*(\d+)",
re.I,
)
_LEGACY_SCRATCH_FALSE_RE = re.compile(
r"scratch worktree used\s*:\s*false",
re.I,
)
_BRANCHES_WORKTREE_RE = re.compile(r"\bbranches/", re.I)
_CONTRADICTORY_HEAD_RE = re.compile(
r"detached head\s*/\s*branch\s+master|"
r"branch\s+master\s*/\s*detached head|"
r"detached head.*branch\s+(?:master|main|dev)\b.*detached|"
r"checkout\s*:\s*detached head\s*/\s*branch",
re.I,
)
_MAIN_CHECKOUT_BRANCH_RE = re.compile(
r"main checkout branch\s*:\s*(\S+)",
re.I,
)
_REVIEW_HEAD_STATE_RE = re.compile(
r"review worktree head state\s*:\s*(\S+)",
re.I,
)
_HANDOFF_SECTION_RE = re.compile(
r"^##\s*Controller Handoff\s*$",
re.I | re.M,
)
def _handoff_field_map(report_text: str) -> dict[str, str]:
"""Parse ``- Field: value`` lines from the Controller Handoff section."""
text = report_text or ""
match = _HANDOFF_SECTION_RE.search(text)
if not match:
return {}
section = text[match.end() :]
fields: dict[str, str] = {}
for line in section.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 _truthy_field(value: str) -> bool:
lowered = (value or "").strip().lower()
if not lowered:
return False
if lowered in {"false", "no", "none", "not applicable", "n/a", "—", "-"}:
return False
return True
def _field_proves_pagination(value: str) -> bool:
proof = (value or "").strip()
if not proof or proof.lower() in {"none", "n/a", "not applicable", "unknown"}:
return False
if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(proof):
return False
if _INCOMPLETE_PAGINATION_PROOF.search(proof):
return False
return bool(_PAGINATION_FINALITY_EVIDENCE.search(proof))
def _pagination_proven(text: str, session: dict, *, pagination_field: str = "") -> bool:
if session.get("pagination_complete") or session.get("inventory_complete"):
return True
session_proof = session.get("inventory_pagination_proof") or ""
if _field_proves_pagination(str(session_proof)):
return True
if _field_proves_pagination(pagination_field):
return True
if _PAGINATION_FINALITY_EVIDENCE.search(text):
return True
return False
def _exact_page_limit_unproven(
text: str, session: dict, *, pagination_proven: bool = False
) -> bool:
"""True when a full page was returned without final-page proof."""
if pagination_proven:
return False
requested = session.get("requested_page_size")
returned = session.get("returned_page_size")
if isinstance(requested, int) and isinstance(returned, int):
if returned >= requested > 0:
return True
for match in _EXACT_PAGE_LIMIT_RE.finditer(text):
count = next((g for g in match.groups() if g), None)
if not count:
continue
try:
n = int(count)
except ValueError:
continue
if n in {10, 20, 50}:
return True
return False
def assess_inventory_worktree_report(
report_text: str,
*,
inventory_session: dict | None = None,
) -> dict[str, Any]:
"""Validate PR inventory pagination proof and worktree field consistency (#293)."""
text = report_text or ""
session = dict(inventory_session or {})
reasons: list[str] = []
fields = _handoff_field_map(text)
pagination_proof_field = fields.get("inventory pagination proof", "")
pagination_proven = _pagination_proven(
text, session, pagination_field=pagination_proof_field
)
if _ELIGIBILITY_CLAIM_RE.search(text):
if not pagination_proven:
reasons.append(
"oldest/next eligible PR or inventory-complete claim requires "
"final-page/no-next-page/pagination_complete proof"
)
if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text):
if not pagination_proven:
reasons.append(
"inventory pagination assumed from default page size without "
"final-page/no-next-page/total-count/traversal proof"
)
if pagination_proof_field:
if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(pagination_proof_field):
reasons.append(
"Inventory pagination proof field relies on page-size assumption"
)
elif _INCOMPLETE_PAGINATION_PROOF.search(pagination_proof_field):
reasons.append(
"Inventory pagination proof field does not prove final page"
)
elif not _field_proves_pagination(pagination_proof_field):
if _ELIGIBILITY_CLAIM_RE.search(text) or _EXACT_PAGE_LIMIT_RE.search(text):
reasons.append(
"Inventory pagination proof field missing final-page metadata"
)
if _exact_page_limit_unproven(text, session, pagination_proven=pagination_proven):
reasons.append(
"exactly-full first page returned without final-page or "
"no-next-page proof"
)
review_worktree_used = fields.get("review worktree used", "")
review_worktree_path = fields.get("review worktree path", "")
scratch_used = fields.get("scratch worktree used", "")
inside_branches = fields.get("review worktree inside branches:", "") or fields.get(
"review worktree inside branches", ""
)
branches_path_in_text = bool(
_BRANCHES_WORKTREE_RE.search(review_worktree_path)
or _BRANCHES_WORKTREE_RE.search(text)
)
if branches_path_in_text:
if review_worktree_used and not _truthy_field(review_worktree_used):
reasons.append(
"reports using a branches/ review worktree must set "
"Review worktree used: true"
)
if scratch_used and re.search(r"\bfalse\b", scratch_used, re.I):
reasons.append(
"Scratch worktree used: false rejected when a branches/ "
"review worktree was created or used"
)
if _LEGACY_SCRATCH_FALSE_RE.search(text) and not _truthy_field(
review_worktree_used
):
reasons.append(
"legacy Scratch worktree used: false contradicts branches/ "
"review worktree usage"
)
if review_worktree_path and _truthy_field(review_worktree_used):
if "branches/" not in review_worktree_path.replace("\\", "/").lower():
reasons.append(
"Review worktree path must be under branches/ when "
"Review worktree used is true"
)
if inside_branches and re.search(r"\bfalse\b", inside_branches, re.I):
reasons.append(
"Review worktree inside branches must be true when path is "
"under branches/"
)
main_branch = (
fields.get("main checkout branch", "")
or _MAIN_CHECKOUT_BRANCH_RE.search(text).group(1)
if _MAIN_CHECKOUT_BRANCH_RE.search(text)
else ""
)
head_state = (
fields.get("review worktree head state", "")
or (
_REVIEW_HEAD_STATE_RE.search(text).group(1)
if _REVIEW_HEAD_STATE_RE.search(text)
else ""
)
)
if main_branch and head_state:
main_norm = main_branch.strip().lower()
head_norm = head_state.strip().lower()
if head_norm in {"branch", "on branch"} and main_norm in {
"master",
"main",
"dev",
}:
if "detached" in text.lower() and "review worktree" in text.lower():
reasons.append(
"review worktree HEAD state must not reuse main checkout "
"branch wording"
)
if _CONTRADICTORY_HEAD_RE.search(text):
reasons.append(
"contradictory checkout wording such as 'Detached HEAD / Branch master'"
)
if review_worktree_used and _truthy_field(review_worktree_used):
if not review_worktree_path or review_worktree_path.lower() in {
"none",
"n/a",
"not applicable",
}:
reasons.append(
"Review worktree used: true requires Review worktree path"
)
if not head_state or head_state.lower() in {
"none",
"n/a",
"not applicable",
"unknown",
}:
reasons.append(
"Review worktree used: true requires Review worktree HEAD state"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": list(dict.fromkeys(reasons)),
"pagination_proven": pagination_proven,
"branches_worktree_used": branches_path_in_text,
"safe_next_action": (
"prove inventory pagination with final-page metadata; align "
"Review worktree used/path/HEAD state with branches/ usage"
if reasons
else "proceed"
),
}
reviewer_merge_simulation.py
+148
View File
@@ -0,0 +1,148 @@
"""Merge-simulation worktree mutation verifier for reviewer reports (#317)."""
from __future__ import annotations
import re
from typing import Any
_WORKTREE_INDEX_FIELD_RE = re.compile(
r"^\s*worktree/index mutations\s*:\s*(.+?)\s*$",
re.IGNORECASE | re.MULTILINE,
)
_READONLY_DIAG_RE = re.compile(
r"read[- ]only diagnostics\s*:\s*(.+)$",
re.IGNORECASE | re.MULTILINE,
)
_WORKTREE_PATH_RE = re.compile(
r"(?:worktree path|diagnostic worktree|simulation worktree)\s*:",
re.IGNORECASE,
)
_PRE_CLEAN_RE = re.compile(
r"(?:pre[- ]simulation|before simulation|clean before).*(?:clean|status)",
re.IGNORECASE,
)
_MERGE_RESULT_RE = re.compile(
r"(?:merge result|simulation result|conflict status|merge conflict)",
re.IGNORECASE,
)
_ABORT_RE = re.compile(
r"(?:merge --abort|abort/reset command|abort command|git merge --abort)",
re.IGNORECASE,
)
_POST_CLEAN_RE = re.compile(
r"(?:post[- ]abort|after abort|clean after).*(?:clean|status)",
re.IGNORECASE,
)
def _command_text(entry: Any) -> str:
if isinstance(entry, dict):
return str(entry.get("command") or "").strip()
return str(entry or "").strip()
def _git_subcommand_line(command: str) -> str | None:
tokens = command.split()
if not tokens or tokens[0] != "git":
return None
return " ".join(tokens[1:])
def merge_simulation_commands(command_log: list | None) -> list[str]:
"""Return logged commands that perform local merge simulation (#317)."""
out: list[str] = []
for entry in command_log or []:
command = _command_text(entry)
rest = _git_subcommand_line(command)
if rest is None:
continue
lower = command.lower()
if rest.startswith("merge"):
if "--no-commit" in lower or (
"--no-ff" in lower and "merge" in lower and "--commit" not in lower
):
out.append(command)
elif "--abort" in lower:
out.append(command)
return out
def assess_merge_simulation_report(
report_text: str,
*,
command_log: list | None = None,
) -> dict[str, Any]:
"""Reject merge simulation classified as read-only; require worktree proof (#317)."""
text = report_text or ""
lower = text.lower()
reasons: list[str] = []
simulation_commands = merge_simulation_commands(command_log)
has_abort = any("--abort" in c.lower() for c in simulation_commands)
has_simulation = any(
"--no-commit" in c.lower() or (
"--no-ff" in c.lower() and "--abort" not in c.lower()
)
for c in simulation_commands
)
if not simulation_commands:
return {
"proven": True,
"block": False,
"reasons": [],
"simulation_commands": [],
"safe_next_action": "proceed",
}
field = _WORKTREE_INDEX_FIELD_RE.search(text)
value = (field.group(1).strip().lower() if field else "") or ""
if not value or value in {"none", "n/a", "no"}:
reasons.append(
"merge simulation commands ran but report lacks "
"'Worktree/index mutations' entry"
)
elif "merge" not in value and "simulation" not in value:
if not any(cmd.lower() in lower for cmd in simulation_commands):
reasons.append(
"Worktree/index mutations must document merge simulation commands"
)
readonly = _READONLY_DIAG_RE.search(text)
if readonly:
readonly_value = readonly.group(1)
for command in simulation_commands:
if command.lower() in readonly_value.lower():
reasons.append(
"merge simulation may not be listed under read-only diagnostics"
)
if has_simulation and "merge simulation" in readonly_value.lower():
reasons.append(
"merge simulation may not be classified as read-only diagnostics"
)
if has_simulation:
if not _WORKTREE_PATH_RE.search(text):
reasons.append("merge simulation report missing worktree path")
if not _PRE_CLEAN_RE.search(text):
reasons.append("merge simulation report missing pre-simulation clean status")
if not _MERGE_RESULT_RE.search(text):
reasons.append("merge simulation report missing merge result/conflict status")
if has_abort or has_simulation:
if has_abort and not _ABORT_RE.search(text):
reasons.append("merge simulation report missing abort/reset command proof")
if has_abort and not _POST_CLEAN_RE.search(text):
reasons.append("merge simulation report missing post-abort clean status")
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"simulation_commands": simulation_commands,
"safe_next_action": (
"classify merge simulation under Worktree/index mutations with full "
"pre/merge/abort/post-clean proof; never list as read-only diagnostics"
if reasons
else "proceed"
),
}
reviewer_mutation_categories.py
+130
View File
@@ -0,0 +1,130 @@
"""Precise mutation category verifier for reviewer controller handoffs (#319)."""
from __future__ import annotations
import re
from typing import Any
_CONTROLLER_HANDOFF_RE = re.compile(r"controller handoff", re.IGNORECASE)
_WORKSPACE_MUTATIONS_RE = re.compile(
r"^\s*[-*]?\s*workspace\s+mutations\s*:",
re.IGNORECASE | re.MULTILINE,
)
_VAGUE_MUTATIONS_NONE_RE = re.compile(
r"^\s*[-*]?\s*mutations\s*:\s*none\s*$",
re.IGNORECASE | re.MULTILINE,
)
_REQUIRED_CATEGORIES = (
"file edits by reviewer",
"worktree/index mutations",
"git ref mutations",
)
_WORKTREE_FIELD_ALIASES = (
"worktree/index mutations",
"worktree mutations",
)
def _has_category(text: str, category: str) -> bool:
pattern = re.compile(
rf"^\s*[-*]?\s*{re.escape(category)}\s*:",
re.IGNORECASE | re.MULTILINE,
)
return bool(pattern.search(text))
def _has_worktree_category(text: str) -> bool:
return any(_has_category(text, alias) for alias in _WORKTREE_FIELD_ALIASES)
def _normalize_for_consistency_check(text: str) -> str:
"""Map #319 precise labels to #313 field names for consistency delegation."""
normalized = text
if _has_category(text, "worktree/index mutations") and not _has_category(
text, "worktree mutations"
):
normalized = re.sub(
r"(worktree/index mutations\s*:)",
"Worktree mutations:",
normalized,
flags=re.IGNORECASE,
)
return normalized
def assess_mutation_categories_report(
report_text: str,
*,
handoff_session: dict | None = None,
observed_commands: list | None = None,
) -> dict[str, Any]:
"""Require precise mutation categories in reviewer controller handoffs (#319)."""
text = report_text or ""
session = dict(handoff_session or {})
reasons: list[str] = []
lower = text.lower()
is_controller = bool(
session.get("controller_handoff")
or _CONTROLLER_HANDOFF_RE.search(text)
)
if not is_controller and not session.get("require_categories"):
return {
"proven": True,
"block": False,
"reasons": [],
"controller_handoff": False,
"safe_next_action": "proceed",
}
if _WORKSPACE_MUTATIONS_RE.search(text):
reasons.append(
"controller handoffs must not use legacy 'Workspace mutations'; "
"use precise mutation category fields"
)
if _VAGUE_MUTATIONS_NONE_RE.search(text):
reasons.append(
"controller handoffs must not use vague 'Mutations: none'; "
"report each precise category separately"
)
missing = [
category
for category in _REQUIRED_CATEGORIES
if category != "worktree/index mutations" and not _has_category(text, category)
]
if not _has_worktree_category(text):
missing.append("worktree/index mutations")
if missing:
reasons.append(
"controller handoff missing precise mutation categories: "
+ ", ".join(missing)
)
commands = observed_commands or session.get("observed_commands")
if commands:
from review_proofs import assess_workspace_mutation_consistency
consistency = assess_workspace_mutation_consistency(
_normalize_for_consistency_check(text),
commands,
)
if not consistency.get("complete"):
reasons.extend(consistency.get("reasons") or [])
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"controller_handoff": True,
"safe_next_action": (
"replace Workspace mutations with file edits, worktree/index, git ref, "
"and other precise mutation category fields"
if reasons
else "proceed"
),
}
reviewer_validation_integrity.py
+148
View File
@@ -0,0 +1,148 @@
"""PR-head vs diagnostic validation integrity verifier (#316)."""
from __future__ import annotations
import re
from typing import Any
_DIAGNOSTIC_LABEL_RE = re.compile(
r"diagnostic local experiment|not pr-head validation|diagnostic-only validation",
re.IGNORECASE,
)
_OFFICIAL_VALIDATION_RE = re.compile(
r"(?:official validation|pr-head validation|validation integrity)",
re.IGNORECASE,
)
_OFFICIAL_COMMAND_RE = re.compile(
r"(?:official validation command|pr-head validation command|validation command)",
re.IGNORECASE,
)
_OFFICIAL_RESULT_RE = re.compile(
r"(?:official validation result|pr-head validation result|validation result|validation integrity status)",
re.IGNORECASE,
)
_DIRTY_AFTER_RE = re.compile(
r"(?:worktree dirty after (?:official )?validation|dirty (?:state )?after (?:official )?validation|"
r"validation worktree dirty after)",
re.IGNORECASE,
)
_DIAGNOSTIC_EDIT_RE = re.compile(
r"(?:diagnostic edit(?:ed)? path|file edits by reviewer|diagnostic local edit)",
re.IGNORECASE,
)
_DIAGNOSTIC_COMMAND_RE = re.compile(
r"(?:diagnostic (?:validation )?command|diagnostic test run|diagnostic result)",
re.IGNORECASE,
)
_SUGGESTED_FIX_RE = re.compile(
r"(?:diagnostic results? (?:were )?used only for suggested fix|suggested fix only|"
r"not used as official validation)",
re.IGNORECASE,
)
_INTEGRITY_STATUS_RE = re.compile(
r"validation integrity status\s*:\s*(passed|failed|not run|contaminated)",
re.IGNORECASE,
)
_POST_EDIT_AS_OFFICIAL_RE = re.compile(
r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit|"
r"full suite passed after diagnostic edit)",
re.IGNORECASE,
)
def _performed_edits(action_log: list[dict] | None) -> list[str]:
paths: list[str] = []
for entry in action_log or []:
if entry.get("gated_rejected") or entry.get("performed") is False:
continue
path = entry.get("path")
if path and entry.get("kind", "file_edit") in {"file_edit", "edit", "write"}:
paths.append(str(path))
return paths
def assess_validation_integrity_report(
report_text: str,
*,
validation_session: dict | None = None,
action_log: list[dict] | None = None,
) -> dict[str, Any]:
"""Separate official PR-head validation from diagnostic edited-worktree tests (#316)."""
text = report_text or ""
session = dict(validation_session or {})
reasons: list[str] = []
diagnostic_edits = list(session.get("diagnostic_edits") or [])
if not diagnostic_edits:
diagnostic_edits = _performed_edits(action_log)
diagnostic_ran = bool(
session.get("diagnostic_validation_ran")
or session.get("diagnostic_runs")
)
official_ran = bool(session.get("official_validation_ran", True))
official_result = (session.get("official_result") or "").strip().lower()
diagnostic_result = (session.get("diagnostic_result") or "").strip().lower()
dirty_after = session.get("worktree_dirty_after_official")
contaminated = bool(session.get("contaminated"))
if official_ran:
if not _OFFICIAL_VALIDATION_RE.search(text) and not _OFFICIAL_COMMAND_RE.search(text):
reasons.append("report missing official PR-head validation section")
if not _OFFICIAL_COMMAND_RE.search(text) and "validation:" not in text.lower():
reasons.append("report missing official validation command")
if not _OFFICIAL_RESULT_RE.search(text):
reasons.append("report missing official validation result or integrity status")
if dirty_after is not None and not _DIRTY_AFTER_RE.search(text):
reasons.append(
"report missing worktree dirty state after official validation"
)
elif dirty_after is None and official_ran and not _DIRTY_AFTER_RE.search(text):
reasons.append(
"report missing worktree dirty state after official validation"
)
if diagnostic_edits or diagnostic_ran:
if not _DIAGNOSTIC_LABEL_RE.search(text):
reasons.append(
"post-edit diagnostic runs must be labeled "
"'Diagnostic local experiment — not PR-head validation'"
)
if diagnostic_edits and not _DIAGNOSTIC_EDIT_RE.search(text):
reasons.append("report missing diagnostic edit path")
if diagnostic_ran and not _DIAGNOSTIC_COMMAND_RE.search(text):
reasons.append("report missing diagnostic command/result")
if not _SUGGESTED_FIX_RE.search(text):
reasons.append(
"report must state diagnostic results were used only for suggested fix"
)
if _POST_EDIT_AS_OFFICIAL_RE.search(text):
reasons.append(
"post-edit diagnostic results cannot be reported as official PR-head validation"
)
if diagnostic_result == "pass" and official_result == "fail":
if "request_changes" not in text.lower() and "failed" not in text.lower():
reasons.append(
"request-changes must cite unmodified PR-head failure, not diagnostic pass"
)
if contaminated:
status = _INTEGRITY_STATUS_RE.search(text)
if not status or "contaminated" not in status.group(1).lower():
reasons.append(
"contaminated validation must report integrity status "
"'contaminated — recovery required'"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"diagnostic_edits": diagnostic_edits,
"safe_next_action": (
"separate official PR-head validation from diagnostic experiments; "
"report dirty-after-validation state"
if reasons
else "proceed"
),
}
reviewer_validation_worktree_edits.py
+184
View File
@@ -0,0 +1,184 @@
"""PR validation worktree read-only edit verifier (#315)."""
from __future__ import annotations
import re
from typing import Any
_VALIDATION_WORKTREE_RE = re.compile(
r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)",
re.IGNORECASE,
)
_DIAGNOSTIC_WORKTREE_RE = re.compile(
r"branches/(?:diagnostic[\w/-]*|scratch[\w/-]*)",
re.IGNORECASE,
)
_FILE_EDITS_NONE_RE = re.compile(
r"file edits by reviewer\s*:\s*none\b",
re.IGNORECASE,
)
_FILE_EDITS_FIELD_RE = re.compile(
r"file edits by reviewer\s*:\s*(.+)",
re.IGNORECASE,
)
_VALIDATION_WORKTREE_PATH_RE = re.compile(
r"(?:validation worktree path|review worktree path)\s*:\s*(\S+)",
re.IGNORECASE,
)
_DIAGNOSTIC_WORKTREE_PATH_RE = re.compile(
r"(?:diagnostic (?:scratch )?worktree path|diagnostic worktree)\s*:\s*(\S+)",
re.IGNORECASE,
)
_DIAGNOSTIC_LABEL_RE = re.compile(
r"diagnostic local experiment|not pr-head validation|diagnostic-only",
re.IGNORECASE,
)
_OFFICIAL_VALIDATION_RE = re.compile(
r"(?:official (?:pr-head )?validation|pr-head validation result)",
re.IGNORECASE,
)
_POST_EDIT_OFFICIAL_RE = re.compile(
r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit)",
re.IGNORECASE,
)
_PERFORMED_EDIT_KINDS = frozenset({"file_edit", "edit", "write", "edited", "created", "wrote"})
def _normalize_path(path: str) -> str:
return (path or "").replace("\\", "/").strip()
def _is_validation_worktree(path: str) -> bool:
return bool(_VALIDATION_WORKTREE_RE.search(_normalize_path(path)))
def _is_diagnostic_worktree(path: str) -> bool:
normalized = _normalize_path(path)
return bool(
_DIAGNOSTIC_WORKTREE_RE.search(normalized)
or "diagnostic" in normalized.lower()
)
def _performed_edits(action_log: list[dict] | None) -> list[dict[str, str]]:
edits: list[dict[str, str]] = []
for entry in action_log or []:
if entry.get("gated_rejected") or entry.get("performed") is False:
continue
kind = (entry.get("kind") or entry.get("action") or "file_edit").strip().lower()
if kind not in _PERFORMED_EDIT_KINDS:
continue
path = (entry.get("path") or "").strip()
if not path:
continue
worktree = _normalize_path(str(entry.get("worktree_path") or entry.get("cwd") or ""))
edits.append({"path": path, "worktree_path": worktree})
return edits
def _extract_validation_path(text: str, session: dict) -> str:
path = _normalize_path(str(session.get("validation_worktree_path") or ""))
if path:
return path
match = _VALIDATION_WORKTREE_PATH_RE.search(text or "")
return _normalize_path(match.group(1)) if match else ""
def _extract_diagnostic_path(text: str, session: dict) -> str:
path = _normalize_path(str(session.get("diagnostic_worktree_path") or ""))
if path:
return path
match = _DIAGNOSTIC_WORKTREE_PATH_RE.search(text or "")
return _normalize_path(match.group(1)) if match else ""
def assess_validation_worktree_edit_report(
report_text: str,
*,
validation_session: dict | None = None,
action_log: list[dict] | None = None,
) -> dict[str, Any]:
"""Block edits in PR validation worktrees; require diagnostic separation (#315)."""
text = report_text or ""
session = dict(validation_session or {})
reasons: list[str] = []
validation_path = _extract_validation_path(text, session)
diagnostic_path = _extract_diagnostic_path(text, session)
validation_edits = list(session.get("validation_worktree_edits") or [])
diagnostic_edits = list(session.get("diagnostic_edits") or [])
edits = _performed_edits(action_log)
if not validation_edits:
for entry in edits:
wt = entry.get("worktree_path") or validation_path
if wt and _is_validation_worktree(wt) and not _is_diagnostic_worktree(wt):
validation_edits.append(entry["path"])
if not diagnostic_edits:
for entry in edits:
wt = entry.get("worktree_path") or ""
if wt and _is_diagnostic_worktree(wt):
diagnostic_edits.append(entry["path"])
elif entry["path"] and diagnostic_path and not validation_edits:
if _is_diagnostic_worktree(diagnostic_path):
diagnostic_edits.append(entry["path"])
claimed_none = bool(_FILE_EDITS_NONE_RE.search(text))
any_edits = bool(validation_edits or diagnostic_edits or edits)
if validation_edits:
reasons.append(
"reviewer must not edit files in the PR validation worktree; "
f"observed edits: {', '.join(validation_edits)}"
)
if diagnostic_edits or session.get("diagnostic_experiment"):
if not diagnostic_path and not _DIAGNOSTIC_WORKTREE_PATH_RE.search(text):
reasons.append(
"diagnostic experiments require a separate diagnostic scratch worktree path"
)
if not _DIAGNOSTIC_LABEL_RE.search(text):
reasons.append(
"diagnostic experiments must be labeled "
"'Diagnostic local experiment — not PR-head validation'"
)
if not _FILE_EDITS_FIELD_RE.search(text) or claimed_none:
reasons.append(
"diagnostic edits must be reported under 'File edits by reviewer'"
)
if any_edits and claimed_none:
reasons.append(
"report claims 'File edits by reviewer: none' but reviewer file edits occurred"
)
if session.get("official_validation_after_edit") or _POST_EDIT_OFFICIAL_RE.search(text):
if validation_edits or (validation_path and diagnostic_edits):
reasons.append(
"post-edit test results cannot be reported as official PR-head validation"
)
if validation_edits and _OFFICIAL_VALIDATION_RE.search(text):
if not _DIAGNOSTIC_LABEL_RE.search(text):
reasons.append(
"validation worktree was edited; official PR-head validation is no longer valid"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"validation_worktree_path": validation_path or None,
"diagnostic_worktree_path": diagnostic_path or None,
"validation_worktree_edits": validation_edits,
"diagnostic_edits": diagnostic_edits,
"safe_next_action": (
"keep PR validation worktrees read-only; use a separate diagnostic "
"scratch worktree and report all reviewer file edits"
if reasons
else "proceed"
),
}
skills/llm-project-workflow/workflows/reconcile-landed-pr.md
+22
View File
@@ -248,6 +248,28 @@ Post a reconciliation comment only if:
If comment capability is missing, record the intended comment in the final
handoff only.
## 12A. Partial reconciliation policy (#302)
The durable policy when `close_pr` capability is missing is
**comment-then-stop** (`resolve_partial_reconciliation_plan` in
`review_proofs.py` enforces it):
* `close_pr` proven → full reconciliation: close the PR and report the
close result (`FULL_RECONCILE_CLOSE_ALLOWED`).
* `close_pr` missing, `comment_pr` proven → post exactly one
reconciliation comment carrying PR head SHA, target branch SHA,
ancestor proof, linked issue status, and the required missing
capability, then stop for a human or authorized close
(`PARTIAL_RECONCILE_COMMENT_THEN_STOP`).
* `comment_pr` also missing → no Gitea mutation; produce a recovery
handoff recording the intended comment and required capabilities
(`RECOVERY_HANDOFF_ONLY`).
* Ancestry not proven → no mutation regardless of capability
(`GATE_NOT_PROVEN`).
Final reports must explain why the comment was or was not posted and
name the missing capability (`assess_partial_reconciliation_report`).
## 13. PR close rules
Close a PR only if:
task_capability_map.py
+12
View File
@@ -96,6 +96,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.read",
"role": "author",
},
"post_heartbeat": {
"permission": "gitea.issue.comment",
"role": "author",
},
"reconcile_issue_claims": {
"permission": "gitea.read",
"role": "author",
},
"cleanup_stale_claims": {
"permission": "gitea.issue.comment",
"role": "author",
},
}
# Issue-mutating MCP tools and their resolver task keys.
tests/test_agent_temp_artifacts.py
+20
View File
@@ -62,7 +62,24 @@ class TestPreflightWarnings(unittest.TestCase):
self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0])
# Issue-write tools are profile-gated (#69); gitea_lock_issue requires
# gitea.issue.comment (see task_capability_map), so the gate must be
# seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359).
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": (
"gitea.issue.create,gitea.issue.close,gitea.issue.comment"
),
}
class TestIssueLockArtifactWarning(unittest.TestCase):
def setUp(self):
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
self._env_patcher.start()
def tearDown(self):
self._env_patcher.stop()
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server._auth", return_value="token x")
@patch("mcp_server._resolve", return_value=("h", "o", "r"))
@@ -72,6 +89,9 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
mock_state.return_value = {
"current_branch": "master",
"porcelain_status": "?? _emit_payload.py\n",
"base_equivalent": True,
"inspected_git_root": "/scratch/wt",
"base_branch": "origin/master",
}
result = mcp_server.gitea_lock_issue(
issue_number=261,
tests/test_already_landed_report_wording.py
+151
View File
@@ -0,0 +1,151 @@
"""Tests for already-landed final-report wording validator (#298).
An already-landed PR is reconciliation-only, never review/merge
eligible, and a head SHA may not be called reviewed unless validation
and diff review actually passed. Final reports and controller handoffs
must use the reconciliation wording, not the legacy eligible/reviewed
fields.
"""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from review_proofs import assess_already_landed_report_wording # noqa: E402
GOOD_ALREADY_LANDED = (
"Controller Handoff\n"
"- Oldest open PR requiring action: PR #278\n"
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n"
"- Candidate head SHA: 2a544b7\n"
"- Reviewed head SHA: none\n"
"- Review worktree used: false\n"
)
class TestAlreadyLandedWording(unittest.TestCase):
def test_correct_reconciliation_wording_passes(self):
result = assess_already_landed_report_wording(
GOOD_ALREADY_LANDED, already_landed=True
)
self.assertTrue(result["complete"])
self.assertFalse(result["downgraded"])
self.assertEqual(result["reasons"], [])
def test_oldest_eligible_wording_fails(self):
report = GOOD_ALREADY_LANDED + "- oldest eligible PR: PR #278\n"
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
self.assertTrue(
any("oldest eligible pr" in r.lower() for r in result["reasons"])
)
def test_next_eligible_wording_fails(self):
report = GOOD_ALREADY_LANDED + "- next eligible PR: PR #278\n"
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
def test_pinned_reviewed_head_fails(self):
report = GOOD_ALREADY_LANDED + "- Pinned reviewed head: 2a544b7\n"
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
self.assertTrue(
any("pinned reviewed head" in r.lower() for r in result["reasons"])
)
def test_scratch_worktree_field_fails(self):
report = GOOD_ALREADY_LANDED + "- Scratch worktree used: False\n"
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
def test_missing_eligibility_class_fails(self):
report = (
"Controller Handoff\n"
"- Oldest open PR requiring action: PR #278\n"
"- Candidate head SHA: 2a544b7\n"
"- Reviewed head SHA: none\n"
"- Review worktree used: false\n"
)
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
self.assertTrue(
any("eligibility class" in r.lower() for r in result["reasons"])
)
def test_populated_reviewed_head_sha_fails(self):
report = GOOD_ALREADY_LANDED.replace(
"- Reviewed head SHA: none\n",
"- Reviewed head SHA: 2a544b7\n",
)
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
self.assertTrue(
any("reviewed head" in r.lower() for r in result["reasons"])
)
class TestNonAlreadyLandedReports(unittest.TestCase):
def test_normal_reviewed_report_passes(self):
report = (
"- Selected PR: 281\n"
"- Pinned reviewed head: abc1234\n"
"- Review decision: approve\n"
)
result = assess_already_landed_report_wording(
report, already_landed=False, review_validated=True
)
self.assertTrue(result["complete"])
def test_blocked_infra_report_with_pinned_head_fails(self):
report = (
"- Selected PR: 281\n"
"- Pinned reviewed head: abc1234\n"
"- Current status: blocked on infra_stop\n"
)
result = assess_already_landed_report_wording(
report, already_landed=False, review_validated=False
)
self.assertFalse(result["complete"])
self.assertTrue(
any("before validation" in r.lower() for r in result["reasons"])
)
def test_validation_failed_report_with_reviewed_sha_fails(self):
report = (
"- Selected PR: 281\n"
"- Reviewed head SHA: abc1234\n"
"- Validation: full suite failed\n"
)
result = assess_already_landed_report_wording(
report, already_landed=False, review_validated=False
)
self.assertFalse(result["complete"])
def test_validation_failed_report_with_reviewed_none_passes(self):
report = (
"- Selected PR: 281\n"
"- Reviewed head SHA: none\n"
"- Validation: full suite failed\n"
)
result = assess_already_landed_report_wording(
report, already_landed=False, review_validated=False
)
self.assertTrue(result["complete"])
if __name__ == "__main__":
unittest.main()
tests/test_issue_claim_heartbeat.py
+116
View File
@@ -0,0 +1,116 @@
"""Tests for issue claim heartbeat leases (#268)."""
from __future__ import annotations
import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_claim_heartbeat as ich # noqa: E402
class TestHeartbeatFormatting(unittest.TestCase):
def test_format_and_parse_roundtrip(self):
body = ich.format_heartbeat_body(
kind="claim",
issue_number=268,
branch="feat/issue-268-heartbeat-leases",
phase="claimed",
profile="prgs-author",
next_action="implement module",
)
parsed = ich.parse_heartbeat_comment(body)
self.assertIsNotNone(parsed)
assert parsed is not None
self.assertEqual(parsed["issue_number"], 268)
self.assertEqual(parsed["branch"], "feat/issue-268-heartbeat-leases")
self.assertEqual(parsed["phase"], "claimed")
class TestClaimClassification(unittest.TestCase):
NOW = datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc)
def _comment(self, *, minutes_ago: int, kind: str = "progress") -> dict:
ts = (self.NOW - timedelta(minutes=minutes_ago)).isoformat()
body = ich.format_heartbeat_body(
kind=kind,
issue_number=268,
branch="feat/issue-268-heartbeat-leases",
phase="implementation",
profile="prgs-author",
)
return {"id": 1, "body": body, "created_at": ts, "updated_at": ts}
def test_active_claim_within_lease(self):
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
result = ich.classify_issue_claim(
issue=issue,
comments=[self._comment(minutes_ago=5)],
now=self.NOW,
)
self.assertEqual(result["status"], "active")
def test_stale_claim_without_branch(self):
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
result = ich.classify_issue_claim(
issue=issue,
comments=[self._comment(minutes_ago=45)],
now=self.NOW,
)
self.assertEqual(result["status"], "stale")
def test_reclaimable_after_sixty_minutes_without_branch(self):
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
result = ich.classify_issue_claim(
issue=issue,
comments=[self._comment(minutes_ago=90)],
now=self.NOW,
)
self.assertEqual(result["status"], "reclaimable")
def test_awaiting_review_when_open_pr_exists(self):
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
open_prs = [
{
"number": 400,
"title": "feat",
"body": "Closes #268",
"head": {"ref": "feat/issue-268-heartbeat-leases"},
}
]
result = ich.classify_issue_claim(
issue=issue,
comments=[self._comment(minutes_ago=120)],
open_prs=open_prs,
now=self.NOW,
)
self.assertEqual(result["status"], "awaiting_review")
def test_phantom_claim_without_heartbeat(self):
issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
result = ich.classify_issue_claim(
issue=issue,
comments=[{"id": 1, "body": "working on it"}],
now=self.NOW,
)
self.assertEqual(result["status"], "phantom")
class TestInventory(unittest.TestCase):
def test_build_cleanup_plan_flags_phantom(self):
inventory = {
"entries": [
{"issue_number": 1, "status": "phantom", "reasons": ["no heartbeat"]},
{"issue_number": 2, "status": "active", "reasons": []},
]
}
plan = ich.build_cleanup_plan(inventory)
self.assertEqual(len(plan), 1)
self.assertEqual(plan[0]["issue_number"], 1)
if __name__ == "__main__":
unittest.main()
tests/test_issue_write_tool_gates.py
+2
View File
@@ -172,6 +172,8 @@ class TestAllowedIssueWritesProceed(IssueWriteGateBase):
return [{"id": 10, "name": "status:in-progress"}]
if method == "POST" and "/issues/9/labels" in url:
return [{"name": "status:in-progress"}]
if method == "POST" and "/issues/9/comments" in url:
return {"id": 501}
if method == "GET" and url.endswith("/user"):
return {"login": "author-user"}
return {}
tests/test_llm_workflow_split.py
+42
View File
@@ -78,6 +78,12 @@ def test_reconcile_landed_workflow_contract():
assert "canonical: true" in text
assert "Do not review or merge normal PRs" in text
assert "Already-landed proof" in text
# Issue #302: partial reconciliation policy must stay documented.
assert "## 12A. Partial reconciliation policy (#302)" in text
assert "comment-then-stop" in text
assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text
assert "RECOVERY_HANDOFF_ONLY" in text
assert "resolve_partial_reconciliation_plan" in text
def test_create_issue_workflow_contract():
@@ -136,6 +142,18 @@ def test_create_issue_final_report_verifier_exported():
assert callable(assess_create_issue_final_report)
def test_merge_simulation_verifier_exported():
from review_proofs import assess_merge_simulation_report
assert callable(assess_merge_simulation_report)
def test_validation_integrity_verifier_exported():
from review_proofs import assess_validation_integrity_report
assert callable(assess_validation_integrity_report)
def test_prior_blocker_skip_verifier_exported():
from review_proofs import assess_prior_blocker_skip_proof
@@ -148,6 +166,30 @@ def test_non_mergeable_skip_verifier_exported():
assert callable(assess_non_mergeable_skip_proof)
def test_validation_worktree_edit_verifier_exported():
from review_proofs import assess_validation_worktree_edit_report
assert callable(assess_validation_worktree_edit_report)
def test_inventory_worktree_verifier_exported():
from review_proofs import assess_inventory_worktree_report
assert callable(assess_inventory_worktree_report)
def test_infra_stop_handoff_verifier_exported():
from review_proofs import assess_infra_stop_handoff_report
assert callable(assess_infra_stop_handoff_report)
def test_mutation_categories_verifier_exported():
from review_proofs import assess_mutation_categories_report
assert callable(assess_mutation_categories_report)
def test_worktree_ownership_verifier_exported():
from review_proofs import assess_worktree_ownership_report
tests/test_mcp_server.py
+3 -1
View File
@@ -321,15 +321,17 @@ class TestMarkIssue(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_start_adds_label(self, _auth, mock_api):
# First call: get labels; second call: add label
# labels, add label, post claim heartbeat comment
mock_api.side_effect = [
[{"id": 10, "name": "status:in-progress"}],
[{"name": "status:in-progress"}],
{"id": 99},
]
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_mark_issue(issue_number=5, action="start")
self.assertTrue(result["success"])
self.assertIn("claimed", result["message"])
self.assertTrue(result.get("heartbeat_posted"))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
tests/test_native_mcp_preference.py
+121
View File
@@ -0,0 +1,121 @@
"""Tests for native MCP preference gate and shell circuit breaker (#270)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import native_mcp_preference as nmp # noqa: E402
class TestShellHealthCircuitBreaker(unittest.TestCase):
def setUp(self):
nmp.clear_shell_health_for_tests()
def test_two_spawn_failures_hard_stop(self):
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
status = nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
self.assertTrue(status["hard_stopped"])
self.assertFalse(status["shell_use_allowed"])
self.assertEqual(status["consecutive_spawn_failures"], 2)
def test_success_resets_breaker(self):
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
status = nmp.record_shell_spawn_outcome(exit_code=0, stdout="ok", stderr="")
self.assertFalse(status["hard_stopped"])
self.assertEqual(status["consecutive_spawn_failures"], 0)
class TestNativeMcpPreferenceGate(unittest.TestCase):
def setUp(self):
nmp.clear_shell_health_for_tests()
def test_mcp_available_blocks_local_script(self):
result = nmp.assess_gitea_operation_path(
task="create_pr",
command_or_detail="python create_pr.py --remote prgs --title T --head h",
mcp_available=True,
mcp_tool_visible=True,
)
self.assertTrue(result["block"])
self.assertEqual(result["path_kind"], "shell_script")
def test_mcp_native_path_allowed(self):
result = nmp.assess_gitea_operation_path(
task="comment_issue",
path_kind="mcp_native",
mcp_available=True,
mcp_tool_visible=True,
)
self.assertFalse(result["block"])
self.assertTrue(result["allowed"])
def test_mcp_unavailable_requires_terminal_report(self):
result = nmp.assess_gitea_operation_path(
task="merge_pr",
path_kind="shell_script",
mcp_available=False,
mcp_tool_visible=False,
)
self.assertTrue(result["block"])
self.assertTrue(
any("terminal recovery report" in r for r in result["reasons"])
)
self.assertIn(nmp.TERMINAL_REPORT_HEADING, result["terminal_report"])
def test_recovery_fallback_allowed_with_proof(self):
result = nmp.assess_gitea_operation_path(
task="merge_pr",
path_kind="shell_script",
command_or_detail="Recovery mode: MCP tools unavailable in this session.",
mcp_available=False,
recovery_mode=True,
recovery_proof_complete=True,
)
self.assertFalse(result["block"])
def test_cli_auth_divergence_blocked(self):
result = nmp.assess_gitea_operation_path(
task="create_pr",
command_or_detail="GITEA_MCP_PROFILE=prgs-reviewer python create_pr.py",
mcp_available=True,
active_profile="prgs-author",
)
self.assertTrue(result["block"])
self.assertTrue(any("CLI auth divergence" in r for r in result["reasons"]))
def test_unsafe_helper_fallback_denied(self):
result = nmp.assess_gitea_operation_path(
task="commit_files",
command_or_detail="python _encode_payload.py",
mcp_available=True,
mcp_tool_visible=True,
)
self.assertTrue(result["block"])
self.assertEqual(result["path_kind"], "unsafe_helper")
def test_reviewer_mcp_server_touch_blocked(self):
result = nmp.assess_gitea_operation_path(
task="review_pr",
command_or_detail="pkill -f gitea_mcp_server.py",
mcp_available=True,
role="reviewer",
active_profile="prgs-reviewer",
)
self.assertTrue(result["block"])
self.assertEqual(result["path_kind"], "mcp_server_touch")
def test_hard_stopped_shell_blocks_non_mcp_path(self):
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
result = nmp.assess_gitea_operation_path(
task="lock_issue",
path_kind="shell_script",
mcp_available=True,
)
self.assertTrue(result["block"])
self.assertTrue(any("circuit breaker" in r for r in result["reasons"]))
if __name__ == "__main__":
unittest.main()
tests/test_reconciliation_handoff.py
+162
View File
@@ -0,0 +1,162 @@
"""Tests for the dedicated reconciliation controller-handoff schema (#303).
Already-landed PR reconciliation runs must emit the reconciliation
schema, not stale author/reviewer handoff fields; observed git ref
mutations (fetch) must be reported, and legacy fields fail validation.
"""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from review_proofs import assess_reconciliation_handoff # noqa: E402
def _good_handoff(**overrides):
fields = {
"Task": "Reconcile already-landed open PRs",
"Repo": "prgs/Scaled-Tech-Consulting/Gitea-Tools",
"Role/profile": "prgs-reconciler",
"Identity": "jcwalker3",
"Selected PR": "278",
"PR live state": "open",
"Candidate head SHA": "2a544b7",
"Target branch": "master",
"Target branch SHA": "fa0cb12",
"Ancestor proof": "candidate head is ancestor of target branch SHA",
"Linked issue": "263",
"Linked issue live status": "open (fetched live this session)",
"Eligibility class": "ALREADY_LANDED_RECONCILE_REQUIRED",
"Capabilities proven": "gitea.pr.comment",
"Missing capabilities": "gitea.pr.close",
"PR comments posted": "1 (PR #278 reconciliation notice)",
"Issue comments posted": "none",
"PRs closed": "none",
"Issues closed": "none",
"Git ref mutations": "git fetch prgs master",
"Worktree mutations": "none",
"MCP/Gitea mutations": "PR comment on #278",
"External-state mutations": "none",
"Read-only diagnostics": "gitea_view_pr 278, gitea_view_issue 263",
"Blocker": "PR close capability missing",
"Safe next action": "operator closes PR #278 or grants close capability",
"No review/merge confirmation": "no review or merge mutations performed",
}
fields.update(overrides)
lines = ["Controller Handoff"]
lines += [f"- {k}: {v}" for k, v in fields.items() if v is not None]
return "\n".join(lines) + "\n"
class TestReconciliationSchemaPasses(unittest.TestCase):
def test_blocked_reconciliation_passes(self):
result = assess_reconciliation_handoff(
_good_handoff(),
observed_commands=["git fetch prgs master"],
)
self.assertTrue(result["complete"])
self.assertEqual(result["reasons"], [])
def test_successful_pr_close_passes(self):
report = _good_handoff(**{
"Missing capabilities": "none",
"PRs closed": "PR #278",
"Blocker": "none",
"Safe next action": "none; reconciliation complete",
})
result = assess_reconciliation_handoff(
report, observed_commands=["git fetch prgs master"]
)
self.assertTrue(result["complete"])
def test_comment_only_reconciliation_passes(self):
report = _good_handoff(**{
"Git ref mutations": "none",
})
result = assess_reconciliation_handoff(report, observed_commands=[])
self.assertTrue(result["complete"])
def test_noop_already_closed_passes(self):
report = _good_handoff(**{
"PR live state": "closed",
"PR comments posted": "none",
"MCP/Gitea mutations": "none",
"Git ref mutations": "none",
"Blocker": "none",
"Safe next action": "none; PR already closed",
})
result = assess_reconciliation_handoff(report, observed_commands=[])
self.assertTrue(result["complete"])
class TestSchemaViolationsFail(unittest.TestCase):
def test_missing_linked_issue_proof_fails(self):
report = _good_handoff(**{"Linked issue live status": None})
result = assess_reconciliation_handoff(report, observed_commands=[])
self.assertFalse(result["complete"])
self.assertTrue(
any("linked issue live status" in r.lower()
for r in result["reasons"])
)
def test_wrong_eligibility_class_fails(self):
report = _good_handoff(**{"Eligibility class": "REVIEW_ELIGIBLE"})
result = assess_reconciliation_handoff(report, observed_commands=[])
self.assertFalse(result["complete"])
def test_missing_handoff_section_fails(self):
result = assess_reconciliation_handoff(
"no handoff here", observed_commands=[]
)
self.assertFalse(result["complete"])
class TestStaleFieldsFail(unittest.TestCase):
def test_pr_number_opened_fails(self):
report = _good_handoff() + "- PR number opened: 278\n"
result = assess_reconciliation_handoff(report, observed_commands=[])
self.assertFalse(result["complete"])
self.assertTrue(
any("pr number opened" in r.lower() for r in result["reasons"])
)
def test_pinned_reviewed_head_fails(self):
report = _good_handoff() + "- Pinned reviewed head: 2a544b7\n"
result = assess_reconciliation_handoff(report, observed_commands=[])
self.assertFalse(result["complete"])
def test_scratch_worktree_used_fails(self):
report = _good_handoff() + "- Scratch worktree used: False\n"
result = assess_reconciliation_handoff(report, observed_commands=[])
self.assertFalse(result["complete"])
def test_workspace_mutations_fails(self):
report = _good_handoff() + "- Workspace mutations: None\n"
result = assess_reconciliation_handoff(report, observed_commands=[])
self.assertFalse(result["complete"])
def test_author_lock_fields_fail(self):
report = (
_good_handoff()
+ "- Issue lock proof: locked before diff\n"
+ "- Claim/comment status: claimed\n"
)
result = assess_reconciliation_handoff(report, observed_commands=[])
self.assertFalse(result["complete"])
class TestObservedFetchMustBeReported(unittest.TestCase):
def test_fetch_with_ref_mutations_none_fails(self):
report = _good_handoff(**{"Git ref mutations": "none"})
result = assess_reconciliation_handoff(
report, observed_commands=["git fetch prgs master"]
)
self.assertFalse(result["complete"])
self.assertTrue(
any("git ref mutation" in r.lower() for r in result["reasons"])
)
if __name__ == "__main__":
unittest.main()
tests/test_review_proofs.py
+369 -3
View File
@@ -23,7 +23,12 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par
from review_proofs import ( # noqa: E402
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
assess_already_landed_report_state,
assess_already_landed_review_gate,
assess_author_pr_report,
assess_partial_reconciliation_report,
resolve_partial_reconciliation_plan,
assess_validation_environment_proof,
assess_full_suite_failure_approval_gate,
assess_queue_ordering_proof,
assess_reviewer_baseline_validation_proof,
@@ -2315,10 +2320,108 @@ class TestCreateIssueFinalReport(unittest.TestCase):
self.assertTrue(result["downgraded"])
class TestQueueOrderingProof(unittest.TestCase):
"""Issue #321: reviewer queue selection must prove ordering."""
class TestValidationEnvironmentProof(unittest.TestCase):
"""Issue #311: exact validation command and execution environment."""
ROOT = "/repo/Gitea-Tools/branches/review-pr-280"
def test_venv_pytest_with_full_report_passes(self):
result = assess_validation_environment_proof(
report_text=(
"Validation command: venv/bin/pytest tests/ --ignore=branches\n"
f"Working directory: {self.ROOT}\n"
"Result: 1014 passed, 6 skipped"
),
)
self.assertTrue(result["proven"])
def test_bare_pytest_without_path_proof_fails(self):
result = assess_validation_environment_proof(
report_text=(
"Validation command: pytest\n"
f"Working directory: {self.ROOT}\n"
"1014 passed, 6 skipped"
),
)
self.assertFalse(result["proven"])
def test_bare_pytest_with_path_and_version_proof_passes(self):
result = assess_validation_environment_proof(
report_text=(
"Validation command: pytest tests/\n"
f"Working directory: {self.ROOT}\n"
"which pytest: /repo/Gitea-Tools/venv/bin/pytest\n"
"pytest --version: pytest 8.3.4\n"
"Resolves to the project venv: yes\n"
"1014 passed, 6 skipped"
),
)
self.assertTrue(result["proven"])
def test_vague_pytest_summary_without_command_fails(self):
result = assess_validation_environment_proof(
report_text=(
f"Working directory: {self.ROOT}\n"
"pytest executed inside the worktree: 1014 passed, 6 skipped"
),
)
self.assertFalse(result["proven"])
def test_structured_environment_proof_passes(self):
result = assess_validation_environment_proof(
validation_environment={
"command": "pytest tests/",
"working_directory": self.ROOT,
"which_pytest": "/repo/Gitea-Tools/venv/bin/pytest",
"pytest_version": "pytest 8.3.4",
"venv_resolved": True,
"passed": 1014,
"skipped": 6,
},
)
self.assertTrue(result["proven"])
def test_missing_working_directory_fails(self):
result = assess_validation_environment_proof(
report_text=(
"Validation command: venv/bin/pytest tests/\n"
"1014 passed, 6 skipped"
),
)
self.assertFalse(result["proven"])
def test_build_final_report_downgrades_missing_environment_proof(self):
report = build_final_report(
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(),
review_mutation=_good_review_mutation(),
controller_handoff=_good_handoff(),
capability_proof=_good_capability_proof(),
sweep_proof=_good_secret_sweep(),
worktree_proof={
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
},
report_text=(
f"Working directory: {self.ROOT}\n"
"pytest executed: 1014 passed, 6 skipped"
),
)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["validation_environment_proven"])
def test_next_eligible_claim_without_policy_fails(self):
result = assess_queue_ordering_proof(
report_text="Selected the next eligible PR: PR #286.",
)
@@ -2540,6 +2643,266 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
self.assertFalse(report["baseline_validation_proven"])
class TestPartialReconciliationPlan(unittest.TestCase):
"""Issue #302: policy when PR-close capability is missing (Option B)."""
def _plan(self, **overrides):
kwargs = {
"ancestor_proven": True,
"capabilities": {
"close_pr": False,
"comment_pr": True,
"close_issue": True,
"comment_issue": True,
},
}
kwargs.update(overrides)
return resolve_partial_reconciliation_plan(**kwargs)
def test_close_capability_present_allows_full_reconcile(self):
plan = self._plan(capabilities={
"close_pr": True, "comment_pr": True,
"close_issue": True, "comment_issue": True,
})
self.assertEqual(plan["outcome"], "FULL_RECONCILE_CLOSE_ALLOWED")
self.assertIn("close_pr", plan["allowed_mutations"])
def test_close_missing_with_comment_posts_comment_then_stops(self):
plan = self._plan()
self.assertEqual(
plan["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP")
self.assertEqual(plan["allowed_mutations"], ("comment_pr",))
for field in (
"PR head SHA",
"target branch SHA",
"ancestor proof",
"linked issue status",
"required missing capability",
):
self.assertIn(field, plan["required_comment_fields"])
def test_comment_capability_missing_produces_handoff_only(self):
plan = self._plan(capabilities={
"close_pr": False, "comment_pr": False,
"close_issue": True, "comment_issue": True,
})
self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY")
self.assertEqual(plan["allowed_mutations"], ())
def test_all_capabilities_missing_produces_handoff_only(self):
plan = self._plan(capabilities={
"close_pr": False, "comment_pr": False,
"close_issue": False, "comment_issue": False,
})
self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY")
self.assertEqual(plan["allowed_mutations"], ())
def test_unproven_ancestry_blocks_all_mutations(self):
plan = self._plan(ancestor_proven=False, capabilities={
"close_pr": True, "comment_pr": True,
"close_issue": True, "comment_issue": True,
})
self.assertEqual(plan["outcome"], "GATE_NOT_PROVEN")
self.assertEqual(plan["allowed_mutations"], ())
def test_missing_capability_dict_fails_closed(self):
plan = self._plan(capabilities=None)
self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY")
self.assertEqual(plan["allowed_mutations"], ())
class TestPartialReconciliationReport(unittest.TestCase):
"""Issue #302: reports must explain why comments were or were not posted."""
def _plan(self, outcome):
capabilities = {
"FULL_RECONCILE_CLOSE_ALLOWED": {
"close_pr": True, "comment_pr": True,
"close_issue": True, "comment_issue": True,
},
"PARTIAL_RECONCILE_COMMENT_THEN_STOP": {
"close_pr": False, "comment_pr": True,
"close_issue": True, "comment_issue": True,
},
"RECOVERY_HANDOFF_ONLY": {
"close_pr": False, "comment_pr": False,
"close_issue": False, "comment_issue": False,
},
}[outcome]
return resolve_partial_reconciliation_plan(
ancestor_proven=True, capabilities=capabilities)
def test_partial_report_requires_comment_and_missing_capability(self):
plan = self._plan("PARTIAL_RECONCILE_COMMENT_THEN_STOP")
good = (
"Reconciliation comment posted on PR #278 with ancestor proof. "
"PR close skipped: close capability gitea.pr.close missing; "
"stopped for authorized close."
)
result = assess_partial_reconciliation_report(good, plan=plan)
self.assertTrue(result["complete"])
bad = "Stopped. Nothing done."
result = assess_partial_reconciliation_report(bad, plan=plan)
self.assertTrue(result["block"])
def test_handoff_only_report_requires_no_comment_explanation(self):
plan = self._plan("RECOVERY_HANDOFF_ONLY")
good = (
"No reconciliation comment posted: comment capability missing. "
"No Gitea mutation performed; recovery handoff produced."
)
result = assess_partial_reconciliation_report(good, plan=plan)
self.assertTrue(result["complete"])
bad = "Recovery handoff produced."
result = assess_partial_reconciliation_report(bad, plan=plan)
self.assertTrue(result["block"])
def test_full_reconcile_report_requires_close_result(self):
plan = self._plan("FULL_RECONCILE_CLOSE_ALLOWED")
good = "PR close result: closed PR #278 after ancestor proof."
result = assess_partial_reconciliation_report(good, plan=plan)
self.assertTrue(result["complete"])
bad = "Reconciliation done."
result = assess_partial_reconciliation_report(bad, plan=plan)
self.assertTrue(result["block"])
class TestAlreadyLandedReviewGate(unittest.TestCase):
"""Issue #292: PR head already on target blocks approval and merge."""
def _gate(self, **overrides):
kwargs = {
"pr_number": 278,
"candidate_head_sha": PINNED,
"target_branch": "master",
"target_branch_sha": OTHER,
"head_is_ancestor_of_target": True,
"live_head_sha": PINNED,
}
kwargs.update(overrides)
return assess_already_landed_review_gate(**kwargs)
def test_already_landed_blocks_approval_and_merge(self):
result = self._gate()
self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
self.assertFalse(result["approve_allowed"])
self.assertFalse(result["merge_api_allowed"])
self.assertFalse(result["request_changes_allowed"])
self.assertTrue(result["reconciliation_handoff_required"])
def test_already_landed_with_real_blocker_allows_request_changes(self):
result = self._gate(real_blocker=True)
self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
self.assertFalse(result["approve_allowed"])
self.assertFalse(result["merge_api_allowed"])
self.assertTrue(result["request_changes_allowed"])
def test_normal_candidate_passes_gate(self):
result = self._gate(head_is_ancestor_of_target=False)
self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
self.assertTrue(result["approve_allowed"])
self.assertTrue(result["merge_api_allowed"])
self.assertFalse(result["reconciliation_handoff_required"])
def test_non_mergeable_normal_pr_still_passes_this_gate(self):
# Mergeability/conflicts are separate gates; ancestry gate only
# decides already-landed vs normal candidate.
result = self._gate(
head_is_ancestor_of_target=False, mergeable=False)
self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
self.assertTrue(result["request_changes_allowed"])
def test_unchecked_ancestry_blocks_review_mutations(self):
result = self._gate(head_is_ancestor_of_target=None)
self.assertEqual(result["state"], "GATE_NOT_PROVEN")
self.assertFalse(result["approve_allowed"])
self.assertFalse(result["merge_api_allowed"])
def test_changed_head_since_pin_blocks_until_recheck(self):
result = self._gate(
head_is_ancestor_of_target=False, live_head_sha=OTHER)
self.assertEqual(result["state"], "GATE_NOT_PROVEN")
self.assertFalse(result["approve_allowed"])
self.assertTrue(any(
"head" in reason.lower() for reason in result["reasons"]
))
def test_invalid_shas_block_gate(self):
result = self._gate(candidate_head_sha="deadbeef")
self.assertEqual(result["state"], "GATE_NOT_PROVEN")
self.assertFalse(result["approve_allowed"])
result = self._gate(target_branch_sha=None)
self.assertEqual(result["state"], "GATE_NOT_PROVEN")
def test_already_landed_handoff_lists_required_fields(self):
result = self._gate()
for field in (
"PR number/title",
"candidate head SHA",
"target branch",
"target branch SHA",
"ancestor proof",
"linked issue status",
"recommended reconciliation action",
"capability proof for close/comment mutations",
):
self.assertIn(field, result["required_handoff_fields"])
class TestAlreadyLandedReportState(unittest.TestCase):
"""Issue #292: reports cannot say APPROVED after the gate fired."""
GOOD_REPORT = (
"Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n"
"Candidate head SHA: " + PINNED + "\n"
"Review decision: none\n"
"Merge result: none\n"
)
def test_gate_fired_with_reconcile_state_passes(self):
result = assess_already_landed_report_state(
self.GOOD_REPORT, gate_fired=True)
self.assertTrue(result["complete"])
self.assertFalse(result["block"])
def test_gate_fired_with_approved_state_blocks(self):
report = self.GOOD_REPORT.replace(
"Review decision: none", "Review decision: APPROVED")
result = assess_already_landed_report_state(report, gate_fired=True)
self.assertTrue(result["block"])
def test_gate_fired_with_merged_state_blocks(self):
report = self.GOOD_REPORT.replace(
"Merge result: none", "Merge result: MERGED")
result = assess_already_landed_report_state(report, gate_fired=True)
self.assertTrue(result["block"])
def test_gate_fired_with_ready_to_merge_blocks(self):
result = assess_already_landed_report_state(
self.GOOD_REPORT + "Current status: READY_TO_MERGE\n",
gate_fired=True)
self.assertTrue(result["block"])
def test_gate_fired_without_reconcile_state_blocks(self):
result = assess_already_landed_report_state(
"Current status: review complete.\n", gate_fired=True)
self.assertTrue(result["block"])
self.assertTrue(any(
"already_landed_reconcile_required" in reason.lower()
for reason in result["reasons"]
))
def test_gate_not_fired_reports_pass_untouched(self):
result = assess_already_landed_report_state(
"Review decision: APPROVED\nMerge result: MERGED\n",
gate_fired=False)
self.assertTrue(result["complete"])
class TestFullSuiteFailureApprovalGate(unittest.TestCase):
"""Issue #323: full-suite failure blocks approval without baseline proof."""
@@ -2707,3 +3070,6 @@ class TestFullSuiteFailureApprovalGate(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_infra_stop_handoff.py
+106
View File
@@ -0,0 +1,106 @@
"""Tests for infra-stop repair handoff verifier (#289)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from capability_stop_terminal import TERMINAL_REPORT_HEADING
from reviewer_infra_stop_handoff import assess_infra_stop_handoff_report # noqa: E402
def _repair_handoff() -> str:
return "\n".join([
TERMINAL_REPORT_HEADING,
"Repair handoff for infra_stop blocked reviewer workflow.",
"CONTROL-CHECKOUT REPAIR MODE",
"MCP process root: /Users/dev/Gitea-Tools",
"Inspected git root: /Users/dev/Gitea-Tools",
"Conflict marker path: gitea_mcp_server.py",
"Merge/rebase control path: none",
"Safe next repair action: resolve conflict markers and restart MCP",
"infra_stop: true",
])
class TestInfraStopHandoff(unittest.TestCase):
def test_repair_handoff_passes(self):
result = assess_infra_stop_handoff_report(
_repair_handoff(),
stop_session={
"infra_stop": True,
"infra_stop_assessment": {
"infra_stop": True,
"conflict_file": "gitea_mcp_server.py",
},
},
)
self.assertTrue(result["proven"], result["reasons"])
def test_next_pr_selection_blocks(self):
report = "\n".join([
TERMINAL_REPORT_HEADING,
"Next eligible PR to review: PR #276",
"Pinned review head SHA: abc123",
])
result = assess_infra_stop_handoff_report(
report,
stop_session={"infra_stop": True, "require_infra_diagnostics": False},
)
self.assertFalse(result["proven"])
self.assertTrue(any("queue" in r.lower() for r in result["reasons"]))
def test_missing_repair_handoff_blocks(self):
result = assess_infra_stop_handoff_report(
"Validation result: pass\nReview summary for PR #276",
stop_session={"infra_stop": True},
)
self.assertFalse(result["proven"])
def test_main_checkout_diagnostics_without_label_blocks(self):
report = "\n".join([
TERMINAL_REPORT_HEADING,
"Repair handoff",
"Ran git status in main checkout for MCP diagnostics.",
])
result = assess_infra_stop_handoff_report(
report,
stop_session={
"infra_stop": True,
"main_checkout_diagnostics": True,
"require_infra_diagnostics": False,
},
)
self.assertFalse(result["proven"])
self.assertTrue(any("control-checkout" in r.lower() for r in result["reasons"]))
def test_background_scheduling_blocks(self):
report = "\n".join([
TERMINAL_REPORT_HEADING,
"Repair handoff",
"Used schedule/manage_task while waiting for MCP recovery.",
])
result = assess_infra_stop_handoff_report(
report,
stop_session={"infra_stop": True, "require_infra_diagnostics": False},
)
self.assertFalse(result["proven"])
self.assertTrue(any("schedule" in r.lower() for r in result["reasons"]))
def test_non_infra_stop_skips(self):
result = assess_infra_stop_handoff_report(
"Selected PR #1 for review",
stop_session={"infra_stop": False},
)
self.assertTrue(result["proven"])
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_infra_stop_handoff_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_inventory_worktree.py
+126
View File
@@ -0,0 +1,126 @@
"""Tests for reviewer inventory/worktree verifier (#293)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_inventory_worktree import assess_inventory_worktree_report # noqa: E402
def _good_handoff() -> str:
return "\n".join([
"## Controller Handoff",
"- Selected PR: 280",
"- Inventory pagination proof: is_final_page: true, has_more: false, total_count: 6",
"- Review worktree used: true",
"- Review worktree path: branches/review-pr280-feat-issue-275-claim",
"- Review worktree inside branches: true",
"- Review worktree HEAD state: detached",
"- Main checkout branch: master",
"- Current status: reviewed PR #280",
])
class TestInventoryPaginationProof(unittest.TestCase):
def test_no_next_page_proof_passes(self):
result = assess_inventory_worktree_report(_good_handoff())
self.assertTrue(result["proven"], result["reasons"])
self.assertTrue(result["pagination_proven"])
def test_rejects_partial_first_page_eligibility_claim(self):
report = "\n".join([
"Oldest eligible PR is #280.",
"gitea_list_prs returned 10 open PRs on the first page.",
"## Controller Handoff",
"- Inventory pagination proof: first page only",
])
result = assess_inventory_worktree_report(report)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(any("eligible" in r.lower() for r in result["reasons"]))
def test_rejects_default_page_size_assumption(self):
report = (
"Inventory exhaustive because fewer than the default Gitea page size "
"were returned."
)
result = assess_inventory_worktree_report(report)
self.assertFalse(result["proven"])
self.assertTrue(any("page size" in r.lower() for r in result["reasons"]))
def test_rejects_exactly_full_first_page_without_finality(self):
report = "\n".join([
"Next eligible PR: #290 based on complete inventory.",
"gitea_list_prs returned 50 open PRs.",
"## Controller Handoff",
"- Inventory pagination proof: returned 50 open PRs",
])
result = assess_inventory_worktree_report(
report,
inventory_session={"requested_page_size": 50, "returned_page_size": 50},
)
self.assertFalse(result["proven"])
self.assertTrue(any("full first page" in r.lower() for r in result["reasons"]))
def test_allows_exact_page_with_session_pagination_complete(self):
report = "Oldest eligible PR: #12 after queue inventory."
result = assess_inventory_worktree_report(
report,
inventory_session={"pagination_complete": True},
)
self.assertTrue(result["proven"], result["reasons"])
class TestWorktreeConsistency(unittest.TestCase):
def test_branches_worktree_requires_review_worktree_used_true(self):
report = "\n".join([
"## Controller Handoff",
"- Inventory pagination proof: is_final_page: true",
"- Review worktree used: false",
"- Review worktree path: branches/review-pr280-feat",
"- Scratch worktree used: false",
])
result = assess_inventory_worktree_report(report)
self.assertFalse(result["proven"])
joined = " ".join(result["reasons"]).lower()
self.assertIn("review worktree used", joined)
self.assertIn("scratch worktree", joined)
def test_rejects_contradictory_detached_and_branch_wording(self):
report = "\n".join([
"Checkout: Detached HEAD / Branch master",
"## Controller Handoff",
"- Inventory pagination proof: is_final_page: true",
"- Review worktree used: true",
"- Review worktree path: branches/review-pr280-feat",
"- Review worktree HEAD state: detached",
"- Main checkout branch: master",
])
result = assess_inventory_worktree_report(report)
self.assertFalse(result["proven"])
self.assertTrue(
any("contradictory" in r.lower() for r in result["reasons"])
)
def test_requires_head_state_when_review_worktree_used(self):
report = "\n".join([
"## Controller Handoff",
"- Inventory pagination proof: pagination_complete: true",
"- Review worktree used: true",
"- Review worktree path: branches/review-pr280-feat",
])
result = assess_inventory_worktree_report(report)
self.assertFalse(result["proven"])
self.assertTrue(any("head state" in r.lower() for r in result["reasons"]))
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_inventory_worktree_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_merge_simulation.py
+135
View File
@@ -0,0 +1,135 @@
"""Tests for merge-simulation worktree mutation verifier (#317)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_merge_simulation import ( # noqa: E402
assess_merge_simulation_report,
merge_simulation_commands,
)
def _good_simulation_report(**overrides) -> str:
lines = [
"Worktree/index mutations: merge simulation in branches/review-pr281",
"Worktree path: branches/review-pr281",
"Pre-simulation clean status: clean (git status --porcelain empty)",
"Merge result: conflicts in review_proofs.py",
"Abort command: git merge --abort",
"Post-abort clean status: clean after abort (git status --porcelain empty)",
]
text = "\n".join(lines)
for key, value in overrides.items():
text = text.replace(f"{key}:", f"{key}: {value}")
return text
class TestMergeSimulationDetection(unittest.TestCase):
def test_detects_no_commit_merge(self):
commands = merge_simulation_commands([
"git merge prgs/master --no-commit --no-ff",
"git status",
])
self.assertEqual(commands, ["git merge prgs/master --no-commit --no-ff"])
def test_detects_merge_abort(self):
commands = merge_simulation_commands([
"git merge prgs/master --no-commit --no-ff",
"git merge --abort",
])
self.assertIn("git merge --abort", commands)
class TestMergeSimulationReport(unittest.TestCase):
def test_no_simulation_passes(self):
result = assess_merge_simulation_report(
"Review complete. Worktree/index mutations: none",
command_log=["git status --porcelain", "git diff prgs/master...HEAD"],
)
self.assertTrue(result["proven"])
def test_documented_simulation_passes(self):
report = _good_simulation_report()
result = assess_merge_simulation_report(
report,
command_log=[
"git merge prgs/master --no-commit --no-ff",
"git merge --abort",
],
)
self.assertTrue(result["proven"], result["reasons"])
def test_readonly_classification_blocks(self):
report = "\n".join([
"Read-only diagnostics: git merge prgs/master --no-commit --no-ff, git status",
"Worktree/index mutations: none",
])
result = assess_merge_simulation_report(
report,
command_log=["git merge prgs/master --no-commit --no-ff"],
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(any("read-only" in r.lower() for r in result["reasons"]))
def test_missing_worktree_field_blocks(self):
report = "Merge simulation ran but only noted in prose."
result = assess_merge_simulation_report(
report,
command_log=["git merge prgs/master --no-commit --no-ff"],
)
self.assertFalse(result["proven"])
self.assertTrue(any("worktree/index mutations" in r.lower() for r in result["reasons"]))
def test_conflict_simulation_requires_merge_result(self):
report = _good_simulation_report().replace(
"Merge result: conflicts in review_proofs.py",
"",
)
result = assess_merge_simulation_report(
report,
command_log=["git merge prgs/master --no-commit --no-ff"],
)
self.assertFalse(result["proven"])
self.assertTrue(any("merge result" in r.lower() for r in result["reasons"]))
def test_aborted_simulation_requires_post_clean_proof(self):
report = _good_simulation_report().replace(
"Post-abort clean status: clean after abort (git status --porcelain empty)",
"",
)
result = assess_merge_simulation_report(
report,
command_log=[
"git merge prgs/master --no-commit --no-ff",
"git merge --abort",
],
)
self.assertFalse(result["proven"])
self.assertTrue(any("post-abort" in r.lower() for r in result["reasons"]))
def test_successful_simulation_without_abort_omits_abort_fields(self):
report = "\n".join([
"Worktree/index mutations: merge simulation completed cleanly",
"Worktree path: branches/review-pr281",
"Pre-simulation clean status: clean before simulation",
"Merge result: clean fast-forward simulation",
])
result = assess_merge_simulation_report(
report,
command_log=["git merge prgs/master --no-commit --no-ff"],
)
self.assertTrue(result["proven"], result["reasons"])
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_merge_simulation_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_mutation_categories.py
+88
View File
@@ -0,0 +1,88 @@
"""Tests for precise mutation category verifier (#319)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_mutation_categories import assess_mutation_categories_report # noqa: E402
def _precise_handoff() -> str:
return "\n".join([
"Controller Handoff",
"File edits by reviewer: none",
"Worktree/index mutations: git worktree add branches/review-pr1",
"Git ref mutations: git fetch prgs master",
"MCP/Gitea mutations: gitea_view_pr",
"Review mutations: gitea_review_pr request_changes",
"Read-only diagnostics: git status, git diff",
])
class TestMutationCategories(unittest.TestCase):
def test_precise_categories_pass(self):
result = assess_mutation_categories_report(_precise_handoff())
self.assertTrue(result["proven"], result["reasons"])
def test_legacy_workspace_mutations_blocks(self):
report = "\n".join([
"Controller Handoff",
"Workspace mutations: none (no local file changes)",
"File edits by reviewer: none",
"Worktree/index mutations: none",
"Git ref mutations: none",
])
result = assess_mutation_categories_report(report)
self.assertFalse(result["proven"])
self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
def test_workspace_none_with_worktree_command_blocks(self):
report = "\n".join([
"Controller Handoff",
"Workspace mutations: none (no local file changes)",
"File edits by reviewer: none",
"Worktree/index mutations: none",
"Git ref mutations: none",
])
result = assess_mutation_categories_report(
report,
observed_commands=["git reset --hard FETCH_HEAD"],
)
self.assertFalse(result["proven"])
def test_file_edits_none_with_worktree_mutation_passes(self):
report = "\n".join([
"Controller Handoff",
"File edits by reviewer: none",
"Worktree/index mutations: git reset --hard FETCH_HEAD",
"Git ref mutations: git fetch prgs",
])
result = assess_mutation_categories_report(
report,
observed_commands=["git reset --hard FETCH_HEAD", "git fetch prgs"],
)
self.assertTrue(result["proven"], result["reasons"])
def test_missing_required_categories_blocks(self):
report = "Controller Handoff\nFile edits by reviewer: none\n"
result = assess_mutation_categories_report(report)
self.assertFalse(result["proven"])
self.assertTrue(any("missing" in r.lower() for r in result["reasons"]))
def test_non_controller_handoff_skips(self):
result = assess_mutation_categories_report(
"Workspace mutations: none\n",
)
self.assertTrue(result["proven"])
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_mutation_categories_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_validation_integrity.py
+130
View File
@@ -0,0 +1,130 @@
"""Tests for PR-head vs diagnostic validation integrity verifier (#316)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_validation_integrity import assess_validation_integrity_report # noqa: E402
def _official_only_report() -> str:
return "\n".join([
"Official PR-head validation on unmodified worktree.",
"Official validation command: pytest tests/ -q",
"Official validation result: failed (3 failed)",
"Validation integrity status: failed",
"Worktree dirty after official validation: clean",
"Review decision: request_changes",
])
def _diagnostic_report() -> str:
return "\n".join([
_official_only_report(),
"Diagnostic local experiment — not PR-head validation",
"Diagnostic edit path: tests/test_example.py",
"File edits by reviewer: tests/test_example.py",
"Diagnostic command: pytest tests/test_example.py -q",
"Diagnostic result: passed after local fix",
"Diagnostic results used only for suggested fix; blocker remains PR-head failure.",
])
class TestValidationIntegrity(unittest.TestCase):
def test_official_only_passes(self):
result = assess_validation_integrity_report(
_official_only_report(),
validation_session={
"official_validation_ran": True,
"official_result": "fail",
"worktree_dirty_after_official": False,
},
)
self.assertTrue(result["proven"], result["reasons"])
def test_failing_pr_head_with_diagnostic_pass_passes(self):
result = assess_validation_integrity_report(
_diagnostic_report(),
validation_session={
"official_validation_ran": True,
"official_result": "fail",
"diagnostic_validation_ran": True,
"diagnostic_result": "pass",
"diagnostic_edits": ["tests/test_example.py"],
"worktree_dirty_after_official": False,
},
)
self.assertTrue(result["proven"], result["reasons"])
def test_post_edit_tests_as_official_blocks(self):
report = "\n".join([
"Official validation result: passed after diagnostic edit",
"pytest passed after local fix in validation worktree",
])
result = assess_validation_integrity_report(
report,
validation_session={
"official_validation_ran": True,
"diagnostic_edits": ["tests/test_example.py"],
"diagnostic_validation_ran": True,
},
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_missing_dirty_after_validation_blocks(self):
report = _official_only_report().replace(
"Worktree dirty after official validation: clean",
"",
)
result = assess_validation_integrity_report(
report,
validation_session={"official_validation_ran": True, "official_result": "fail"},
)
self.assertFalse(result["proven"])
self.assertTrue(any("dirty" in r.lower() for r in result["reasons"]))
def test_diagnostic_without_label_blocks(self):
report = _diagnostic_report().replace(
"Diagnostic local experiment — not PR-head validation",
"",
)
result = assess_validation_integrity_report(
report,
validation_session={
"official_validation_ran": True,
"diagnostic_edits": ["tests/test_example.py"],
"diagnostic_validation_ran": True,
},
)
self.assertFalse(result["proven"])
self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"]))
def test_contaminated_requires_integrity_status(self):
report = "Validation worktree was edited before official validation completed."
result = assess_validation_integrity_report(
report,
validation_session={"contaminated": True, "official_validation_ran": True},
)
self.assertFalse(result["proven"])
self.assertTrue(any("contaminated" in r.lower() for r in result["reasons"]))
def test_action_log_edits_trigger_diagnostic_rules(self):
result = assess_validation_integrity_report(
_official_only_report(),
validation_session={"official_validation_ran": True, "official_result": "fail"},
action_log=[{"path": "tests/test_example.py", "kind": "file_edit", "performed": True}],
)
self.assertFalse(result["proven"])
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_validation_integrity_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_validation_worktree_edits.py
+134
View File
@@ -0,0 +1,134 @@
"""Tests for PR validation worktree no-edit verifier (#315)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_validation_worktree_edits import ( # noqa: E402
assess_validation_worktree_edit_report,
)
def _read_only_report() -> str:
return "\n".join([
"Validation worktree path: branches/review-pr280-feat-issue-275-claim",
"Official PR-head validation on unmodified worktree.",
"Official validation result: failed (1 failed)",
"File edits by reviewer: none",
"Review decision: request_changes",
])
def _diagnostic_report() -> str:
return "\n".join([
"Validation worktree path: branches/review-pr280-feat-issue-275-claim",
"Official PR-head validation on unmodified worktree.",
"Official validation result: failed (1 failed)",
"Review decision: request_changes",
"Diagnostic local experiment — not PR-head validation",
"Diagnostic scratch worktree path: branches/diagnostic-pr280-fix-test",
"File edits by reviewer: tests/test_agent_temp_artifacts.py (diagnostic only)",
"Diagnostic result: passed after local fix",
])
class TestValidationWorktreeEdits(unittest.TestCase):
def test_read_only_review_passes(self):
result = assess_validation_worktree_edit_report(
_read_only_report(),
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
},
)
self.assertTrue(result["proven"], result["reasons"])
def test_validation_worktree_edit_blocks(self):
result = assess_validation_worktree_edit_report(
_read_only_report(),
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
"validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"],
},
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(any("validation worktree" in r.lower() for r in result["reasons"]))
def test_file_edits_none_contradiction_blocks(self):
result = assess_validation_worktree_edit_report(
_read_only_report(),
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
},
action_log=[
{
"path": "tests/test_agent_temp_artifacts.py",
"kind": "file_edit",
"performed": True,
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
}
],
)
self.assertFalse(result["proven"])
self.assertTrue(any("file edits" in r.lower() for r in result["reasons"]))
def test_diagnostic_scratch_with_reporting_passes(self):
result = assess_validation_worktree_edit_report(
_diagnostic_report(),
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
"diagnostic_worktree_path": "branches/diagnostic-pr280-fix-test",
"diagnostic_edits": ["tests/test_agent_temp_artifacts.py"],
"diagnostic_experiment": True,
},
)
self.assertTrue(result["proven"], result["reasons"])
def test_diagnostic_without_scratch_worktree_blocks(self):
result = assess_validation_worktree_edit_report(
"File edits by reviewer: tests/test_example.py",
validation_session={
"diagnostic_edits": ["tests/test_example.py"],
"diagnostic_experiment": True,
},
)
self.assertFalse(result["proven"])
self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"]))
def test_post_edit_official_validation_blocks(self):
report = "\n".join([
"Validation worktree path: branches/review-pr280-feat-issue-275-claim",
"Official validation result: passed after local edit",
"File edits by reviewer: tests/test_agent_temp_artifacts.py",
])
result = assess_validation_worktree_edit_report(
report,
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
"validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"],
"official_validation_after_edit": True,
},
)
self.assertFalse(result["proven"])
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_validation_worktree_edit_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()