fix: resolve conflicts for PR #371 against current master

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

gitea_mcp_server.py

@@ -476,6 +476,7 @@ 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
 
 
@@ -5146,6 +5147,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,
@@ -5155,6 +5191,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.
 
@@ -5204,7 +5242,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,
@@ -5214,6 +5276,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

@@ -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

@@ -1242,6 +1242,121 @@ 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
@@ -4773,6 +4888,13 @@ def assess_already_landed_handoff_report(report_text, **kwargs):
     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

reviewer_infra_stop_handoff.py

@@ -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"
+        ),
+    }

task_capability_map.py

@@ -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_issue_claim_heartbeat.py

@@ -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

@@ -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

@@ -172,6 +172,12 @@ def test_already_landed_handoff_verifier_exported():
     assert callable(assess_already_landed_handoff_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
 

tests/test_mcp_server.py

@@ -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_reconciliation_handoff.py

@@ -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_reviewer_infra_stop_handoff.py

@@ -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()