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 #381 against latest master

Browse Source
This commit is contained in:
sysadmin
2026-07-07 09:38:31 -04:00
parent 9cf1b41b77 b4c79fdbbc
commit f9bf830067
16 changed files with 2106 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+274 -1
View File
@@ -477,6 +477,7 @@ import review_proofs # noqa: E402
import agent_temp_artifacts
import issue_lock_worktree # noqa: E402
import already_landed_reconcile # noqa: E402
import issue_claim_heartbeat # noqa: E402
import merged_cleanup_reconcile # noqa: E402
@@ -5338,6 +5339,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,
@@ -5347,6 +5383,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.
@@ -5396,7 +5434,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,
@@ -5406,6 +5468,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
review_proofs.py
+285
View File
@@ -1242,6 +1242,102 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None):
}
# ── 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.
@@ -3868,6 +3964,172 @@ def assess_reviewer_baseline_validation_proof(
}
# ---------------------------------------------------------------------------
# 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)
# ---------------------------------------------------------------------------
@@ -4460,6 +4722,13 @@ def assess_merge_simulation_report(report_text, **kwargs):
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
@@ -4474,6 +4743,22 @@ 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_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_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"
),
}
task_capability_map.py
+12
View File
@@ -100,6 +100,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.pr.close",
"role": "reconciler",
},
"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_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
+19 -1
View File
@@ -142,6 +142,12 @@ def test_merge_simulation_verifier_exported():
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
@@ -154,7 +160,19 @@ 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_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
assert callable(assess_worktree_ownership_report)
assert callable(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_review_proofs.py
+135
View File
@@ -23,6 +23,8 @@ 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_validation_environment_proof,
assess_full_suite_failure_approval_gate,
@@ -2639,6 +2641,139 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
self.assertFalse(report["baseline_validation_proven"])
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."""
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()