fix(workflow): retire status:pr-open on every terminal PR transition (Closes #780)

status:pr-open was applied by gitea_create_pr and never removed again. Every
terminal path finished without touching it, so a repository audit found 40
closed issues still advertising an open PR.

Add terminal_pr_label_cleanup.py as the single authoritative rule and route
every sanctioned terminal path through it, so the paths cannot drift:

- merge (gitea_merge_pr)
- close without merge (gitea_edit_pr)
- supersession/abandonment (gitea_reconcile_superseded_by_merged_pr)
- already-landed reconciliation (gitea_reconcile_already_landed_pr)
- controller closure (gitea_close_issue)
- retry/recovery (new gitea_cleanup_terminal_pr_labels)

The rule removes only status:pr-open, preserves every other label, allows an
empty resulting set, is a no-op when the label is absent (so retries are
safe), and confirms the outcome by read-after-write rather than assumption.

Controller closure runs the cleanup before the state change and fails closed
if it cannot be completed and verified; closing first would bake in the stale
label with no later step to catch it. Post-merge cleanup never blocks the
merge, which already happened, and reports failures with a safe next action.

Also:
- gitea_assess_terminal_label_hygiene: read-only terminal validation that
  reports residual status:pr-open, exempting issues with a genuinely open PR.
- _put_issue_label_names now accepts Gitea's empty response body when the
  requested set is empty, so clearing the last label works.
- test_audit's close_issue fixture keys on the request instead of call order,
  since closing now also reads labels for the cleanup and its read-back.

Docs: label-taxonomy terminal-transition section, runbook pointer, and the
review-merge / reconcile-landed final-report terminal-label requirements.

Suite: 4045 passed, 11 failed, 6 skipped. The same 11 failures reproduce on
clean master df31674 (4010 passed, 11 failed) and are pre-existing.

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

allocator_service.py

@@ -18,10 +18,12 @@ after they exist as normal issues; this module never assigns incidents.
 
 from __future__ import annotations
 
+import hashlib
+import json
 import os
 import uuid
 from dataclasses import dataclass, field
-from typing import Any, Sequence
+from typing import Any, Mapping, Sequence
 
 from control_plane_db import (
     ControlPlaneDB,
@@ -40,6 +42,22 @@ OUTCOME_NEEDS_CONTROLLER = "needs_controller"
 OUTCOME_NO_SAFE = "no_safe_work"
 OUTCOME_ROLE_INELIGIBLE = "role_ineligible"
 OUTCOME_PREVIEW = "preview"  # dry-run only (apply=false)
+# #765: ownership could not be established for every remaining candidate.
+OUTCOME_OWNERSHIP_DEFECT = "allocator_ownership_defect"
+# #776: excluded issue still carries a live same-owner lease — resume or release.
+OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE = "blocked_by_excluded_own_lease"
+# #776: dry-run/apply candidate-set fingerprint mismatch (CAS drift).
+OUTCOME_CANDIDATE_SET_DRIFT = "candidate_set_drift"
+
+# #765 skip reason code for work already claimed by a different controller.
+SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session"
+# #776: controller-supplied pre-rank exclusion.
+SKIP_EXCLUDED_BY_CONTROLLER = "excluded_by_controller"
+
+# Ownership verdicts for a live claim on a candidate (#765).
+OWNERSHIP_OWN = "own"
+OWNERSHIP_FOREIGN = "foreign"
+OWNERSHIP_UNKNOWN = "unknown"
 
 # Human-readable statement of how a winner is chosen (#758 AC10). Reported
 # alongside allocator results so the flat status:ready tier and its
@@ -144,9 +162,76 @@ class SkipRecord:
     kind: str
     number: int
     reason: str
+    reason_code: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
-        return {"kind": self.kind, "number": self.number, "reason": self.reason}
+        return {
+            "kind": self.kind,
+            "number": self.number,
+            "reason": self.reason,
+            "reason_code": self.reason_code,
+        }
+
+
+CONTROLLER_INSTANCE_ENV = "GITEA_CONTROLLER_INSTANCE_ID"
+
+
+def resolve_controller_instance_id(
+    env: Mapping[str, str] | None = None,
+) -> str | None:
+    """Return this controller's stable identity, or ``None`` if undeclared.
+
+    Deliberately has no derived fallback. The obvious candidates are unsafe:
+    ``session_id`` is regenerated per invocation, and the MCP process pid is
+    shared by every controller attached to the same daemon — two independent
+    controllers really do report the same pid and profile. Guessing from either
+    would let one controller adopt another's lease, which is the failure #765
+    exists to prevent. When this returns ``None``, live claims are treated as
+    unidentified: they are excluded from selection and reported as ownership
+    defects rather than adopted.
+    """
+    source = env if env is not None else os.environ
+    return (source.get(CONTROLLER_INSTANCE_ENV) or "").strip() or None
+
+
+def classify_claim_ownership(
+    claim: dict[str, Any] | None,
+    *,
+    session_id: str | None,
+    controller_instance_id: str | None,
+) -> str | None:
+    """Classify a live claim as own / foreign / unknown ownership (#765).
+
+    Returns ``None`` when the candidate carries no live claim.
+
+    Session ids are regenerated per allocator invocation, so they only prove
+    ownership positively (an exact match is certainly this session). The
+    durable signal is ``controller_instance_id``. When either side lacks one,
+    ownership is *unknown*: the allocator must not assume that a lease sharing
+    the same profile belongs to this controller, so unknown is treated as
+    not-ours for selection purposes and reported as an ownership defect.
+    """
+    if not claim:
+        return None
+    claim_session = str(claim.get("session_id") or "").strip()
+    claim_instance = str(claim.get("controller_instance_id") or "").strip()
+    own_session = str(session_id or "").strip()
+    own_instance = str(controller_instance_id or "").strip()
+
+    if claim_session and own_session and claim_session == own_session:
+        return OWNERSHIP_OWN
+    if claim_instance and own_instance:
+        return (
+            OWNERSHIP_OWN if claim_instance == own_instance else OWNERSHIP_FOREIGN
+        )
+    if not claim_instance and not own_instance:
+        # Neither side declares a controller identity. The session ids differ
+        # (an exact match returned OWN above), so this is simply someone
+        # else's lease: foreign, and we wait rather than adopt.
+        return OWNERSHIP_FOREIGN
+    # Exactly one side is identified, so the two cannot be compared: this may
+    # or may not be our own task under a different session id. Never guess.
+    return OWNERSHIP_UNKNOWN
 
 
 def normalize_role(role: str | None, *, profile_name: str | None = None) -> str:
@@ -203,8 +288,16 @@ def classify_skip(
     *,
     role: str,
     terminal_pr: int | None,
+    claim_ownership: str | None = None,
 ) -> str | None:
-    """Return skip reason, or None if candidate is selectable for *role*."""
+    """Return skip reason, or None if candidate is selectable for *role*.
+
+    *claim_ownership* (#765) is the verdict from
+    :func:`classify_claim_ownership` for this candidate's live claim. Foreign
+    and unknown claims are excluded so one session's in-progress task can never
+    blockade the queue for a different controller; ``own`` stays selectable so
+    a controller can resume its own work.
+    """
     if c.state in ("merged", "closed"):
         return f"{c.kind}#{c.number} is {c.state}; never assign"
     if c.blocked or "status:blocked" in c.labels:
@@ -214,6 +307,16 @@ def classify_skip(
             c.dependency_reason
             or f"{c.kind}#{c.number} has unmet dependencies"
         )
+    if claim_ownership in (OWNERSHIP_FOREIGN, OWNERSHIP_UNKNOWN):
+        detail = (
+            "owned by another controller instance"
+            if claim_ownership == OWNERSHIP_FOREIGN
+            else "owner could not be identified; never adopt on a guess"
+        )
+        return (
+            f"{c.kind}#{c.number} {SKIP_CLAIMED_BY_OTHER_SESSION}: "
+            f"active lease {detail}"
+        )
     if c.already_claimed_elsewhere:
         return f"{c.kind}#{c.number} already claimed elsewhere"
     if c.kind == "pr" and not (c.head_sha or "").strip():
@@ -274,6 +377,113 @@ def sort_candidates(candidates: Sequence[WorkCandidate]) -> list[WorkCandidate]:
     )
 
 
+def _require_strict_int(value: Any, *, field: str) -> int:
+    """Parse an issue number; reject bools and non-integers (#776)."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(
+            f"{field} must be an integer (booleans and non-integers rejected; "
+            f"got {type(value).__name__})"
+        )
+    return int(value)
+
+
+def normalize_exclude_issue_numbers(
+    exclude_issue_numbers: Any = None,
+) -> list[int]:
+    """Normalize controller-supplied exclusions to a sorted unique int list (#776).
+
+    ``None`` / omitted → empty list (existing behavior). Accepts a list/tuple of
+    integers. Rejects scalars, bools-as-ints, nested structures, and strings.
+    """
+    if exclude_issue_numbers is None:
+        return []
+    if isinstance(exclude_issue_numbers, (str, bytes)) or not isinstance(
+        exclude_issue_numbers, (list, tuple)
+    ):
+        raise ValueError(
+            "exclude_issue_numbers must be a list of integers "
+            f"(got {type(exclude_issue_numbers).__name__})"
+        )
+    out: list[int] = []
+    seen: set[int] = set()
+    for idx, raw in enumerate(exclude_issue_numbers):
+        num = _require_strict_int(raw, field=f"exclude_issue_numbers[{idx}]")
+        if num not in seen:
+            seen.add(num)
+            out.append(num)
+    return sorted(out)
+
+
+def candidate_set_fingerprint(
+    candidates: Sequence[WorkCandidate],
+    *,
+    exclude_issue_numbers: Sequence[int] | None = None,
+) -> str:
+    """Stable CAS fingerprint of normalized candidate set + exclusions (#776 AC4)."""
+    payload = {
+        "candidates": sorted(
+            ({"kind": c.kind, "number": int(c.number)} for c in candidates),
+            key=lambda x: (x["kind"], x["number"]),
+        ),
+        "exclude_issue_numbers": list(
+            normalize_exclude_issue_numbers(exclude_issue_numbers)
+        ),
+    }
+    blob = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(blob.encode("utf-8")).hexdigest()
+
+
+def normalize_candidates_payload(raw: Any) -> list[WorkCandidate]:
+    """Decode MCP ``candidates_json`` from list or JSON string (#776 AC3).
+
+    Accepts:
+    * an already-decoded ``list`` of candidate dicts (native MCP transport);
+    * a valid JSON string that decodes to such a list (backward compatible).
+
+    Rejects malformed JSON, scalars, non-list containers, invalid records,
+    booleans-as-integers, and unsupported types with fail-closed ``ValueError``.
+    """
+    if raw is None:
+        raise ValueError("candidates_json is empty")
+    if isinstance(raw, (bytes, bytearray)):
+        try:
+            raw = raw.decode("utf-8")
+        except Exception as exc:  # noqa: BLE001
+            raise ValueError(
+                f"candidates_json bytes are not valid utf-8: {exc}"
+            ) from exc
+    if isinstance(raw, str):
+        text = raw.strip()
+        if not text:
+            raise ValueError("candidates_json string is empty")
+        try:
+            decoded = json.loads(text)
+        except json.JSONDecodeError as exc:
+            raise ValueError(
+                f"malformed candidates_json JSON: {exc.msg} at pos {exc.pos}"
+            ) from exc
+        raw = decoded
+    if not isinstance(raw, list):
+        raise ValueError(
+            "candidates_json must be a JSON list (or already-decoded list); "
+            f"got {type(raw).__name__}"
+        )
+    candidates: list[WorkCandidate] = []
+    for idx, item in enumerate(raw):
+        if not isinstance(item, dict):
+            raise ValueError(
+                f"invalid candidate record at index {idx}: expected object, "
+                f"got {type(item).__name__}"
+            )
+        try:
+            candidates.append(candidate_from_dict(item))
+        except (KeyError, TypeError, ValueError, InvalidWorkKindError) as exc:
+            raise ValueError(
+                f"invalid candidate record at index {idx}: {exc}"
+            ) from exc
+    return candidates
+
+
 def allocate_next_work(
     db: ControlPlaneDB,
     *,
@@ -287,12 +497,22 @@ def allocate_next_work(
     profile_name: str | None = None,
     username: str | None = None,
     lease_ttl_seconds: int | None = None,
+    controller_instance_id: str | None = None,
+    claims: Mapping[tuple[str, int], dict[str, Any]] | None = None,
+    exclude_issue_numbers: Sequence[int] | None = None,
+    expected_candidate_set_fingerprint: str | None = None,
 ) -> dict[str, Any]:
     """Select and optionally reserve the next work unit via control-plane DB.
 
     *apply=False* (default): dry-run selection only — no lease/assignment.
     *apply=True*: atomic ``assign_and_lease`` for the selected candidate.
 
+    *exclude_issue_numbers* (#776): numbers removed before ranking. Omitted /
+    empty preserves prior behavior.
+
+    *expected_candidate_set_fingerprint* (#776 AC4): when set on apply, rejects
+    material candidate-set drift vs a prior dry-run.
+
     Never uses file locks or comment-only leases as the assignment source.
     """
     if db is None:
@@ -328,6 +548,7 @@ def allocate_next_work(
             role=role_norm,
             profile=profile_name,
             pid=os.getpid(),
+            controller_instance_id=controller_instance_id,
         )
     except Exception as exc:  # noqa: BLE001 — surface structured
         return {
@@ -369,30 +590,238 @@ def allocate_next_work(
         }
     terminal_pr = int(terminal["terminal_pr"]) if terminal else None
 
+    # #765: live claims exclude work owned by a *different* controller before
+    # ranking, so one session's in-progress task cannot blockade the queue.
+    # #776 AC5: load live claims in this call path immediately before selection
+    # (and before apply reserve) so ownership is never stale within the
+    # allocation attempt. Test callers may inject *claims* explicitly.
+    if claims is None:
+        try:
+            claims = db.list_active_claims(remote=remote, org=org, repo=repo)
+        except Exception as exc:  # noqa: BLE001
+            return {
+                "success": False,
+                "outcome": OUTCOME_NO_SAFE,
+                "reasons": [
+                    f"active claim lookup failed: {exc} (fail closed, #765)"
+                ],
+                "skipped": [],
+                "assignment": None,
+                "substrate": "control_plane_db",
+            }
+
+    try:
+        exclude_nums = normalize_exclude_issue_numbers(exclude_issue_numbers)
+    except ValueError as exc:
+        return {
+            "success": False,
+            "outcome": OUTCOME_NO_SAFE,
+            "apply": bool(apply),
+            "reasons": [
+                f"invalid exclude_issue_numbers: {exc} (fail closed, #776)"
+            ],
+            "skipped": [],
+            "assignment": None,
+            "substrate": "control_plane_db",
+        }
+    exclude_set = set(exclude_nums)
+    cas_fp = candidate_set_fingerprint(
+        candidates, exclude_issue_numbers=exclude_nums
+    )
+    expected_fp = (expected_candidate_set_fingerprint or "").strip() or None
+    if expected_fp and expected_fp != cas_fp:
+        return {
+            "success": False,
+            "outcome": OUTCOME_CANDIDATE_SET_DRIFT,
+            "apply": bool(apply),
+            "reasons": [
+                "candidate-set fingerprint drift: apply rejected rather than "
+                "silently leasing a different candidate (#776 AC4)"
+            ],
+            "candidate_set_fingerprint": cas_fp,
+            "expected_candidate_set_fingerprint": expected_fp,
+            "exclude_issue_numbers": list(exclude_nums),
+            "skipped": [],
+            "assignment": None,
+            "substrate": "control_plane_db",
+            "file_lock_only": False,
+            "comment_lease_only": False,
+        }
+
     skipped: list[SkipRecord] = []
-    ordered = sort_candidates(list(candidates))
+    claims_excluded: list[dict[str, Any]] = []
+    ownership_defects: list[dict[str, Any]] = []
+    controller_excluded: list[dict[str, Any]] = []
+
+    # #776 AC2: remove excluded numbers *before* ranking / selection / lease.
+    rankable: list[WorkCandidate] = []
+    for c in candidates:
+        if int(c.number) in exclude_set:
+            reason = (
+                f"{c.kind}#{c.number} {SKIP_EXCLUDED_BY_CONTROLLER}: "
+                "controller pre-rank exclusion"
+            )
+            skipped.append(
+                SkipRecord(
+                    c.kind,
+                    c.number,
+                    reason,
+                    SKIP_EXCLUDED_BY_CONTROLLER,
+                )
+            )
+            controller_excluded.append(
+                {
+                    "kind": c.kind,
+                    "number": c.number,
+                    "reason_code": SKIP_EXCLUDED_BY_CONTROLLER,
+                }
+            )
+            # #776 AC5: same-owner live lease on an excluded issue is a
+            # structured resume/release blocker, never a silent strand.
+            claim = claims.get((c.kind, int(c.number))) if claims else None
+            ownership = classify_claim_ownership(
+                claim,
+                session_id=session_id,
+                controller_instance_id=controller_instance_id,
+            )
+            if ownership == OWNERSHIP_OWN and claim:
+                return {
+                    "success": True,
+                    "outcome": OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE,
+                    "apply": bool(apply),
+                    "role": role_norm,
+                    "profile_name": profile_name,
+                    "username": username,
+                    "session_id": session_id,
+                    "remote": remote,
+                    "org": org,
+                    "repo": repo,
+                    "selected": None,
+                    "expected_role_next": None,
+                    "reasons": [
+                        f"{c.kind}#{c.number} is excluded_by_controller but "
+                        "carries a live same-owner lease; resume or release "
+                        "that lease before allocating other work (#776 AC5)"
+                    ],
+                    "skipped": [s.as_dict() for s in skipped],
+                    "terminal_pr": terminal_pr,
+                    "assignment": None,
+                    "substrate": "control_plane_db",
+                    "file_lock_only": False,
+                    "comment_lease_only": False,
+                    "controller_instance_id": controller_instance_id,
+                    "claims_excluded": list(claims_excluded),
+                    "ownership_defects": list(ownership_defects),
+                    "controller_excluded": list(controller_excluded),
+                    "exclude_issue_numbers": list(exclude_nums),
+                    "candidate_set_fingerprint": cas_fp,
+                    "blocked_lease": {
+                        "kind": c.kind,
+                        "number": c.number,
+                        "lease_id": claim.get("lease_id"),
+                        "owner_session_id": claim.get("session_id"),
+                        "owner_controller_instance_id": claim.get(
+                            "controller_instance_id"
+                        ),
+                        "expires_at": claim.get("expires_at"),
+                        "safe_next_action": (
+                            "resume the same-owner lease or release it, then "
+                            "re-run allocation without stranding the excluded "
+                            "issue"
+                        ),
+                    },
+                }
+            continue
+        rankable.append(c)
+
+    ordered = sort_candidates(rankable)
     selected: WorkCandidate | None = None
     for c in ordered:
-        reason = classify_skip(c, role=role_norm, terminal_pr=terminal_pr)
+        claim = claims.get((c.kind, int(c.number))) if claims else None
+        ownership = classify_claim_ownership(
+            claim,
+            session_id=session_id,
+            controller_instance_id=controller_instance_id,
+        )
+        reason = classify_skip(
+            c,
+            role=role_norm,
+            terminal_pr=terminal_pr,
+            claim_ownership=ownership,
+        )
         if reason:
-            skipped.append(SkipRecord(c.kind, c.number, reason))
+            is_claim_skip = SKIP_CLAIMED_BY_OTHER_SESSION in reason
+            skipped.append(
+                SkipRecord(
+                    c.kind,
+                    c.number,
+                    reason,
+                    SKIP_CLAIMED_BY_OTHER_SESSION if is_claim_skip else None,
+                )
+            )
+            if is_claim_skip and claim:
+                record = {
+                    "kind": c.kind,
+                    "number": c.number,
+                    "ownership": ownership,
+                    "lease_id": claim.get("lease_id"),
+                    "owner_session_id": claim.get("session_id"),
+                    "owner_controller_instance_id": claim.get(
+                        "controller_instance_id"
+                    ),
+                    "expires_at": claim.get("expires_at"),
+                }
+                claims_excluded.append(record)
+                if ownership == OWNERSHIP_UNKNOWN:
+                    ownership_defects.append(record)
             continue
         selected = c
         break
 
     if selected is None:
         # If terminal lock blocks all review work, surface that explicitly.
+        owner_session_id: str | None = None
         if terminal_pr is not None and role_norm in (ROLE_REVIEWER, ROLE_MERGER):
             outcome = OUTCOME_BLOCKED_TERMINAL
             reasons = [
                 f"no safe work for role '{role_norm}': active terminal-review "
                 f"lock on PR #{terminal_pr} (resolve terminal path first, #332/#600)"
             ]
+        elif ownership_defects:
+            # #765: every remaining candidate is claimed and at least one owner
+            # could not be identified. Report the defect; never adopt.
+            outcome = OUTCOME_OWNERSHIP_DEFECT
+            reasons = [
+                f"no safe assignable work for role '{role_norm}': "
+                f"{len(ownership_defects)} candidate(s) carry an active lease "
+                "whose controller ownership could not be established. Record a "
+                "controller_instance_id on those sessions; the allocator will "
+                "not assume a shared profile means shared ownership (#765)."
+            ]
+        elif claims_excluded:
+            outcome = OUTCOME_WAIT
+            reasons = [
+                f"no unclaimed work for role '{role_norm}': "
+                f"{len(claims_excluded)} candidate(s) are actively claimed by "
+                "another controller. Waiting; their leases are not adopted (#765)."
+            ]
+            # Preserve the pre-#765 wait contract: name the blocking owner.
+            owner_session_id = claims_excluded[0].get("owner_session_id")
+        elif controller_excluded and not ordered:
+            # #776 AC7: every candidate was controller-excluded → wait / no lease.
+            outcome = OUTCOME_WAIT
+            reasons = [
+                f"no assignable work for role '{role_norm}': all "
+                f"{len(controller_excluded)} candidate(s) were removed by "
+                f"{SKIP_EXCLUDED_BY_CONTROLLER} before ranking; no assignment "
+                "or lease created (#776 AC7)"
+            ]
         else:
             outcome = OUTCOME_NO_SAFE
             reasons = [
                 f"no safe assignable work for role '{role_norm}' "
-                f"among {len(ordered)} candidates"
+                f"among {len(ordered)} rankable candidates "
+                f"({len(controller_excluded)} controller-excluded)"
             ]
         return {
             "success": True,
@@ -414,6 +843,13 @@ def allocate_next_work(
             "substrate": "control_plane_db",
             "file_lock_only": False,
             "comment_lease_only": False,
+            "controller_instance_id": controller_instance_id,
+            "claims_excluded": list(claims_excluded),
+            "ownership_defects": list(ownership_defects),
+            "controller_excluded": list(controller_excluded),
+            "exclude_issue_numbers": list(exclude_nums),
+            "candidate_set_fingerprint": cas_fp,
+            "owner_session_id": owner_session_id,
             "downstream_note": (
                 "#612 incident bridge remains downstream of #600; "
                 "allocator never assigns raw monitoring incidents"
@@ -460,6 +896,12 @@ def allocate_next_work(
             "substrate": "control_plane_db",
             "file_lock_only": False,
             "comment_lease_only": False,
+            "controller_instance_id": controller_instance_id,
+            "claims_excluded": list(claims_excluded),
+            "ownership_defects": list(ownership_defects),
+            "controller_excluded": list(controller_excluded),
+            "exclude_issue_numbers": list(exclude_nums),
+            "candidate_set_fingerprint": cas_fp,
             "downstream_note": (
                 "#612 incident bridge remains downstream of #600; "
                 "allocator never assigns raw monitoring incidents"
@@ -583,6 +1025,12 @@ def allocate_next_work(
         "substrate": "control_plane_db",
         "file_lock_only": False,
         "comment_lease_only": False,
+        "controller_instance_id": controller_instance_id,
+        "claims_excluded": list(claims_excluded),
+        "ownership_defects": list(ownership_defects),
+        "controller_excluded": list(controller_excluded),
+        "exclude_issue_numbers": list(exclude_nums),
+        "candidate_set_fingerprint": cas_fp,
         "downstream_note": (
             "#612 incident bridge remains downstream of #600; "
             "allocator never assigns raw monitoring incidents"
@@ -614,14 +1062,32 @@ def _next_command(role: str, c: WorkCandidate) -> str:
 
 
 def candidate_from_dict(data: dict[str, Any]) -> WorkCandidate:
-    """Build a WorkCandidate from a plain dict (tests / MCP inventory)."""
+    """Build a WorkCandidate from a plain dict (tests / MCP inventory).
+
+    #776: reject booleans-as-integers and non-int numbers fail-closed.
+    """
+    if "number" not in data:
+        raise KeyError("number")
+    number = _require_strict_int(data["number"], field="number")
+    priority_raw = data.get("priority") or 0
+    if isinstance(priority_raw, bool) or not isinstance(priority_raw, (int, float)):
+        # Allow numeric strings only for priority? Keep strict for bools.
+        if isinstance(priority_raw, str) and priority_raw.strip().lstrip("-").isdigit():
+            priority = int(priority_raw)
+        else:
+            raise ValueError(
+                f"priority must be numeric (booleans rejected; "
+                f"got {type(priority_raw).__name__})"
+            )
+    else:
+        priority = int(priority_raw)
     return WorkCandidate(
         kind=str(data.get("kind") or "issue"),
-        number=int(data["number"]),
+        number=number,
         state=str(data.get("state") or "open"),
         labels=tuple(data.get("labels") or ()),
         title=str(data.get("title") or ""),
-        priority=int(data.get("priority") or 0),
+        priority=priority,
         head_sha=data.get("head_sha"),
         request_changes_current_head=bool(data.get("request_changes_current_head")),
         approval_on_current_head=bool(data.get("approval_on_current_head")),

control_plane_db.py

@@ -302,6 +302,7 @@ class ControlPlaneDB:
                 conn.executescript(_SCHEMA_SQL)
                 self._migrate_incident_links_null_scope(conn)
                 self._migrate_lease_lifecycle_columns(conn)
+                self._migrate_session_ownership_columns(conn)
                 conn.execute(
                     "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)",
                     ("schema_version", str(SCHEMA_VERSION)),
@@ -492,32 +493,62 @@ class ControlPlaneDB:
         namespace: str | None = None,
         pid: int | None = None,
         status: str = "active",
+        controller_instance_id: str | None = None,
     ) -> dict[str, Any]:
+        """Register/refresh a session row.
+
+        *controller_instance_id* (#765) is the stable identity of the
+        controller that owns this session. Session ids are regenerated per
+        invocation, so they cannot express "my own in-progress task"; the
+        controller instance can. It is never overwritten with ``None``, so a
+        heartbeat from a caller that does not supply one cannot erase
+        ownership.
+        """
         now = _ts()
+        instance = (controller_instance_id or "").strip() or None
         with self._tx() as conn:
             existing = conn.execute(
                 "SELECT session_id FROM sessions WHERE session_id = ?",
                 (session_id,),
             ).fetchone()
             if existing:
-                conn.execute(
-                    """
-                    UPDATE sessions
-                    SET role = ?, profile = ?, namespace = ?, pid = ?,
-                        last_heartbeat_at = ?, status = ?
-                    WHERE session_id = ?
-                    """,
-                    (role, profile, namespace, pid, now, status, session_id),
-                )
+                if instance is None:
+                    conn.execute(
+                        """
+                        UPDATE sessions
+                        SET role = ?, profile = ?, namespace = ?, pid = ?,
+                            last_heartbeat_at = ?, status = ?
+                        WHERE session_id = ?
+                        """,
+                        (role, profile, namespace, pid, now, status, session_id),
+                    )
+                else:
+                    conn.execute(
+                        """
+                        UPDATE sessions
+                        SET role = ?, profile = ?, namespace = ?, pid = ?,
+                            last_heartbeat_at = ?, status = ?,
+                            controller_instance_id = ?
+                        WHERE session_id = ?
+                        """,
+                        (
+                            role, profile, namespace, pid, now, status,
+                            instance, session_id,
+                        ),
+                    )
             else:
                 conn.execute(
                     """
                     INSERT INTO sessions(
                         session_id, role, profile, namespace, pid,
-                        started_at, last_heartbeat_at, status
-                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                        started_at, last_heartbeat_at, status,
+                        controller_instance_id
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                     """,
-                    (session_id, role, profile, namespace, pid, now, now, status),
+                    (
+                        session_id, role, profile, namespace, pid, now, now,
+                        status, instance,
+                    ),
                 )
             row = conn.execute(
                 "SELECT * FROM sessions WHERE session_id = ?",
@@ -1215,6 +1246,73 @@ class ControlPlaneDB:
             if name not in cols:
                 conn.execute(f"ALTER TABLE leases ADD COLUMN {name} {decl}")
 
+    _SESSION_OWNERSHIP_COLUMNS: tuple[tuple[str, str], ...] = (
+        ("controller_instance_id", "TEXT"),
+    )
+
+    def _migrate_session_ownership_columns(self, conn: sqlite3.Connection) -> None:
+        """Add the stable controller identity to sessions (#765).
+
+        Pre-existing rows migrate with ``NULL``. A NULL instance is treated as
+        *unknown ownership* by the allocator and is never silently adopted.
+        """
+        cols = {
+            row[1]
+            for row in conn.execute("PRAGMA table_info(sessions)").fetchall()
+        }
+        if not cols:
+            return
+        for name, decl in self._SESSION_OWNERSHIP_COLUMNS:
+            if name not in cols:
+                conn.execute(f"ALTER TABLE sessions ADD COLUMN {name} {decl}")
+
+    def list_active_claims(
+        self,
+        *,
+        remote: str | None = None,
+        org: str | None = None,
+        repo: str | None = None,
+        role: str | None = None,
+        limit: int = 500,
+    ) -> dict[tuple[str, int], dict[str, Any]]:
+        """Map ``(work_kind, work_number)`` to its live claim (#765).
+
+        Only ``active`` leases count as claims; released/expired rows never
+        withhold work. Callers compare the returned ``controller_instance_id``
+        against their own to decide own-task vs foreign-task.
+        """
+        claims: dict[tuple[str, int], dict[str, Any]] = {}
+        for row in self.list_leases(
+            remote=remote,
+            org=org,
+            repo=repo,
+            role=role,
+            statuses=("active",),
+            limit=limit,
+        ):
+            kind = str(row.get("work_kind") or "").strip().lower()
+            number = row.get("work_number")
+            if not kind or number is None:
+                continue
+            key = (kind, int(number))
+            claim = {
+                "lease_id": row.get("lease_id"),
+                "session_id": row.get("session_id"),
+                "controller_instance_id": row.get("session_controller_instance_id"),
+                "role": row.get("role"),
+                "profile": row.get("session_profile"),
+                "expires_at": row.get("expires_at"),
+                "work_kind": kind,
+                "work_number": int(number),
+            }
+            # Keep the longest-lived claim when duplicates exist.
+            previous = claims.get(key)
+            if previous is None or str(claim["expires_at"] or "") > str(
+                previous["expires_at"] or ""
+            ):
+                claims[key] = claim
+        return claims
+
     def _lease_columns(self, conn: sqlite3.Connection) -> set[str]:
         return {
             row[1]
@@ -1256,7 +1354,8 @@ class ControlPlaneDB:
                    w.number AS work_number, w.state AS work_state,
                    w.current_head_sha AS work_head_sha,
                    s.pid AS session_pid, s.profile AS session_profile,
-                   s.status AS session_status
+                   s.status AS session_status,
+                   s.controller_instance_id AS session_controller_instance_id
             FROM leases l
             JOIN work_items w ON w.work_item_id = l.work_item_id
             LEFT JOIN sessions s ON s.session_id = l.session_id

docs/label-taxonomy.md

@@ -131,6 +131,44 @@ Suggested lifecycle:
 The helper module `issue_workflow_labels.py` is the source of truth for the
 canonical label specs and status transition replacement behavior.
 
+## Terminal PR transitions retire `status:pr-open` (#780)
+
+`status:pr-open` states that a linked PR is *currently open*. The moment that
+stops being true the label must go, whatever ended the PR:
+
+| Terminal reason | Raised by |
+|---|---|
+| `merged` | `gitea_merge_pr` |
+| `closed_without_merge` | `gitea_edit_pr` closing the PR |
+| `superseded` | `gitea_reconcile_superseded_by_merged_pr` |
+| `already_landed` | `gitea_reconcile_already_landed_pr` |
+| `controller_closure` | `gitea_close_issue` |
+| `abandoned` | abandonment handling |
+| `retry_recovery` | `gitea_cleanup_terminal_pr_labels` after a partial failure |
+
+All of these route through one rule in `terminal_pr_label_cleanup.py`, so the
+paths cannot drift apart. The rule guarantees:
+
+- only `status:pr-open` is removed — every other label is preserved verbatim;
+- an empty resulting label set is valid (it was the issue's only label);
+- an issue that no longer carries the label is a no-op, so retries are safe;
+- the result is confirmed by a read-after-write re-read, not assumed.
+
+Controller closure runs the cleanup **before** changing issue state and fails
+closed if it cannot be completed and verified — closing first would bake in the
+stale label with no later step to catch it. Post-merge cleanup never blocks the
+merge: the transition already happened, so failures are reported with a
+`safe_next_action` instead.
+
+Use `gitea_assess_terminal_label_hygiene` as terminal validation before
+declaring a transition or cleanup batch complete. It enumerates issues plus the
+live open PRs and reports any issue still carrying `status:pr-open` without an
+open PR to justify it. Issues with a genuinely open PR are exempt, not
+residual.
+
+Recovery from a partial failure is `gitea_cleanup_terminal_pr_labels` with
+`terminal_reason='retry_recovery'`.
+
 ## Discussion Issues
 
 Discussion issues must be labeled `type:discussion`.
@@ -157,6 +195,10 @@ If a discussion produces implementation work, either:
   be applied to the locked issue, then applies it after the PR is created.
 - `gitea_set_issue_labels` accepts an explicit `worktree_path` so author
   sessions can satisfy the branches-only mutation guard while changing labels.
+- `gitea_cleanup_terminal_pr_labels` retires `status:pr-open` after a terminal
+  PR transition; it is idempotent, so it is also the retry/recovery path.
+- `gitea_assess_terminal_label_hygiene` is the read-only terminal validation
+  for residual `status:pr-open`.
 
 ## Existing Non-Workflow Labels
 

docs/llm-workflow-runbooks.md

@@ -706,7 +706,9 @@ do **not** improvise shell wrappers or fall back to direct API / temp scripts.
   `fix/...` / `docs/...`); `cd` into that worktree; implement narrowly; add or
   update tests if behavior changes; run the full suite; commit with an
   issue-linked message; open a PR to `master`; move the issue to
-  `status:pr-open`. **Do not** review or merge your own PR. Include an
+  `status:pr-open` (every terminal transition later retires that label
+  automatically — see [`label-taxonomy.md`](label-taxonomy.md)). **Do not**
+  review or merge your own PR. Include an
   `LLM Handoff Metadata` block (with `LLM-Agent-SHA`) in the PR body — see
   [`llm-agent-sha.md`](llm-agent-sha.md).
 - **Prompt:** `Use an author profile to implement issue #N and open a PR to

final_report_validator.py

@@ -19,6 +19,10 @@ import reviewer_handoff_consistency
 import thread_state_ledger_validator
 from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof
 from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
+from self_propagating_handoff import (
+    HANDOFF_HEADING as SELF_PROPAGATING_HANDOFF_HEADING,
+    assess_final_report_self_propagating_handoff,
+)
 from review_proofs import (
     HANDOFF_HEADING,
     assess_controller_handoff,
@@ -728,6 +732,65 @@ def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]:
     )
 
 
+_MUTATION_ACCOUNTING_PATTERNS = {
+    "local_failed_attempts": re.compile(
+        r"local\s+failed\s+attempts\s*:\s*(\d+)", re.IGNORECASE
+    ),
+    "blocked_api_attempts": re.compile(
+        r"blocked\s+api\s+attempts\s*:\s*(\d+)", re.IGNORECASE
+    ),
+    "successful_server_mutations": re.compile(
+        r"successful\s+server(?:[-\s]side)?\s+mutations\s*:\s*(\d+)", re.IGNORECASE
+    ),
+}
+
+_READBACK_VERIFIED_PATTERN = re.compile(
+    r"read[-\s]?after[-\s]?write\s+verified\s*:\s*(yes|true)", re.IGNORECASE
+)
+
+
+def _rule_shared_mutation_budget_accounting(
+    report_text: str,
+    *,
+    mutation_attempt_ledger: list[dict] | None = None,
+) -> list[dict[str, str]]:
+    """#617: mutation budget counts server-side changes only.
+
+    No-op unless the session supplies an attempt ledger. When it does, the
+    report's three attempt categories must match the ledger exactly, so a
+    pre-API validator rejection can never be reported as a Gitea mutation and
+    a real mutation can never be hidden.
+    """
+    if mutation_attempt_ledger is None:
+        return []
+
+    from mutation_budget_classifier import assess_final_report_mutation_accounting
+
+    text = report_text or ""
+    claimed: dict[str, Any] = {}
+    for field, pattern in _MUTATION_ACCOUNTING_PATTERNS.items():
+        match = pattern.search(text)
+        if match:
+            claimed[field] = int(match.group(1))
+    if _READBACK_VERIFIED_PATTERN.search(text):
+        claimed["readback_verified"] = True
+
+    result = assess_final_report_mutation_accounting(claimed, mutation_attempt_ledger)
+    if result.get("valid"):
+        return []
+    return _findings_from_reasons(
+        "shared.mutation_budget_accounting",
+        result.get("reasons") or [],
+        field="Mutation accounting",
+        severity="block",
+        safe_next_action=(
+            "report 'Local failed attempts:', 'Blocked API attempts:', and "
+            "'Successful server-side mutations:' with counts matching the "
+            "attempt ledger; pre-API rejections are not Gitea mutations"
+        ),
+    )
+
+
 def _rule_conflict_fix_classification_proof(report_text: str) -> list[dict[str, str]]:
     from conflict_fix_classification import (
         assess_conflict_fix_classification_final_report,
@@ -1564,6 +1627,21 @@ def _rule_shared_mcp_native_cleanup_proof(report_text: str) -> list[dict[str, st
     )
 
 
+def _rule_shared_self_propagating_handoff(report_text: str) -> list[dict[str, str]]:
+    """#626: a report that adopts the handoff protocol must complete it."""
+    result = assess_final_report_self_propagating_handoff(report_text)
+    if not result.get("applicable") or not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "shared.self_propagating_handoff",
+        result.get("reasons") or ["incomplete canonical handoff"],
+        field=SELF_PROPAGATING_HANDOFF_HEADING,
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or "complete every canonical handoff field before posting",
+    )
+
+
 _SHARED_ISSUE_LOCK_RULES = (
     _rule_shared_issue_lock_external_state,
     _rule_shared_manual_lock_pr_override,
@@ -1584,13 +1662,24 @@ _SHARED_CANONICAL_COMMENT_RULES = (
     _rule_shared_canonical_comment_post_claim,
 )
 
+_SHARED_MUTATION_BUDGET_RULES = (
+    _rule_shared_mutation_budget_accounting,
+)
+
+# #626: enforced for every task kind that can continue the workflow chain.
+_SHARED_SELF_PROPAGATING_HANDOFF_RULES = (
+    _rule_shared_self_propagating_handoff,
+)
+
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
+        *_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_CANONICAL_COMMENT_RULES,
+        *_SHARED_MUTATION_BUDGET_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_legacy_workspace_mutations,
         _rule_reviewer_vague_mutations_none,
@@ -1620,6 +1709,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_stale_head_proof,
     ],
     "merge_pr": [
+        *_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
@@ -1631,11 +1721,13 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_stale_head_proof,
     ],
     "reconcile_already_landed": [
+        *_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
         _rule_reconcile_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_CANONICAL_COMMENT_RULES,
+        *_SHARED_MUTATION_BUDGET_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         *_SHARED_CLEANUP_PROOF_RULES,
         _rule_reconcile_stale_author_fields,
@@ -1650,20 +1742,24 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_audit_reconciliation_boundary,
     ],
     "author_issue": [
+        *_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_CANONICAL_COMMENT_RULES,
+        *_SHARED_MUTATION_BUDGET_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
     ],
     "work_issue": [
+        *_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_CANONICAL_COMMENT_RULES,
+        *_SHARED_MUTATION_BUDGET_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_shared_issue_acceptance_gate,
         _rule_reviewer_vague_mutations_none,
@@ -1672,28 +1768,34 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_worktree_cleanup_audit_proof,
     ],
     "issue_filing": [
+        *_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_CANONICAL_COMMENT_RULES,
+        *_SHARED_MUTATION_BUDGET_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
     "inventory": [
+        *_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_CANONICAL_COMMENT_RULES,
+        *_SHARED_MUTATION_BUDGET_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_pagination_proof,
     ],
     "issue_selection": [
+        *_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
         _rule_shared_controller_handoff,
         _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_CANONICAL_COMMENT_RULES,
+        *_SHARED_MUTATION_BUDGET_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
     # Controller issue closure (#529): a closure report must not bury an
@@ -1701,6 +1803,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     # Kept intentionally narrow so a closure pre-check does not demand the
     # full reviewer/author handoff schema.
     "controller_close": [
+        *_SHARED_SELF_PROPAGATING_HANDOFF_RULES,
         _rule_reviewer_premerge_baseline_proof,
     ],
 }
@@ -1766,6 +1869,7 @@ def assess_final_report_validator(
     session_pr_opened: bool = False,
     validation_session: dict | None = None,
     reconciler_close_lock: dict | None = None,
+    mutation_attempt_ledger: list[dict] | None = None,
 ) -> dict[str, Any]:
     """Validate final-report text against task-specific proof rules (#327).
 
@@ -1829,6 +1933,7 @@ def assess_final_report_validator(
         "session_pr_opened": session_pr_opened,
         "validation_session": validation_session,
         "reconciler_close_lock": reconciler_close_lock,
+        "mutation_attempt_ledger": mutation_attempt_ledger,
     }
 
     for rule in _RULES_BY_TASK.get(normalized_kind, ()):

gitea_mcp_server.py

@@ -2122,6 +2122,7 @@ def _seed_session_context(
     )
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
+import terminal_pr_label_cleanup  # noqa: E402  # #780 status:pr-open terminal rule
 import reviewer_pr_lease  # noqa: E402
 import post_merge_moot_lease_gate  # noqa: E402  # #745 reconciler cleanup gate
 import merger_lease_adoption  # noqa: E402
@@ -2698,6 +2699,12 @@ def _put_issue_label_names(
         auth,
         {"labels": ids},
     )
+    # Clearing every label is a legitimate full-set replacement (#780 AC4): the
+    # last label may be the one being retired. Gitea answers that PUT with an
+    # empty body, which api_request surfaces as None — treat it as the empty
+    # label set rather than a protocol violation.
+    if res is None and not names:
+        res = []
     if not isinstance(res, list):
         raise RuntimeError(
             "Post-mutation label verification failed: expected a list of labels "
@@ -2824,7 +2831,191 @@ def release_in_progress_label(issue_numbers: list[int], remote: str, host: str |
 
     return results
 
-def cleanup_in_progress_for_pr(pr_payload: dict, remote: str, host: str | None, org: str | None, repo: str | None) -> dict:
+def _clear_pr_open_label_for_issue(
+    *,
+    base: str,
+    auth: str,
+    issue_number: int,
+    terminal_reason: str,
+    resolve_label_ids,
+    audit_kwargs: dict,
+) -> dict:
+    """Apply the authoritative #780 cleanup rule to exactly one issue.
+
+    Plans through :mod:`terminal_pr_label_cleanup`, mutates via the verified
+    full-set replacement, then re-reads the issue so the caller reports a
+    read-after-write proof rather than an assumption.
+    """
+    try:
+        current = _issue_label_names(base, auth, issue_number)
+    except Exception as exc:  # noqa: BLE001 — redact before surfacing
+        return {
+            "issue_number": issue_number,
+            "status": "error",
+            "performed": False,
+            "verified": False,
+            "reasons": [f"could not read current labels: {_redact(str(exc))}"],
+        }
+
+    plan = terminal_pr_label_cleanup.plan_pr_open_cleanup(
+        current, terminal_reason=terminal_reason
+    )
+    if not plan["cleanup_required"]:
+        # Idempotent by construction: nothing to remove, nothing to verify
+        # beyond the read we already did (#780 AC5).
+        return {
+            "issue_number": issue_number,
+            "status": "not present",
+            "performed": False,
+            "verified": True,
+            "idempotent_noop": True,
+            "labels_after": plan["labels_after"],
+            "reasons": [],
+        }
+
+    try:
+        # Resolved only once a mutation is actually needed, so the no-op path
+        # stays free of extra API traffic.
+        label_ids_by_name = resolve_label_ids()
+        with _audited(
+            "set_issue_labels",
+            issue_number=issue_number,
+            request_metadata={
+                "source": f"terminal_pr_label_cleanup:{plan['terminal_reason']}",
+                "removed": plan["removed"],
+                "preserved": plan["preserved"],
+            },
+            **audit_kwargs,
+        ):
+            _put_issue_label_names(
+                base=base,
+                auth=auth,
+                issue_number=issue_number,
+                names=plan["labels_after"],
+                label_ids_by_name=label_ids_by_name,
+            )
+    except Exception as exc:  # noqa: BLE001 — redact before surfacing
+        return {
+            "issue_number": issue_number,
+            "status": "error",
+            "performed": False,
+            "verified": False,
+            "labels_before": plan["labels_before"],
+            "reasons": [f"label replacement failed: {_redact(str(exc))}"],
+            "safe_next_action": (
+                "Re-run gitea_cleanup_terminal_pr_labels with "
+                f"terminal_reason='{terminal_pr_label_cleanup.RETRY_RECOVERY}' "
+                f"for issue #{issue_number}."
+            ),
+        }
+
+    try:
+        observed = _issue_label_names(base, auth, issue_number)
+    except Exception as exc:  # noqa: BLE001 — redact before surfacing
+        return {
+            "issue_number": issue_number,
+            "status": "unverified",
+            "performed": True,
+            "verified": False,
+            "labels_before": plan["labels_before"],
+            "reasons": [f"read-after-write check failed: {_redact(str(exc))}"],
+            "safe_next_action": (
+                "Re-run gitea_cleanup_terminal_pr_labels with "
+                f"terminal_reason='{terminal_pr_label_cleanup.RETRY_RECOVERY}' "
+                f"for issue #{issue_number} to confirm the final label set."
+            ),
+        }
+
+    verification = terminal_pr_label_cleanup.verify_pr_open_cleanup(
+        observed, plan=plan
+    )
+    return {
+        "issue_number": issue_number,
+        "status": "removed" if verification["verified"] else "verification_failed",
+        "performed": True,
+        "verified": verification["verified"],
+        "labels_before": plan["labels_before"],
+        "labels_after": verification["observed_labels"],
+        "empty_label_set": verification["empty_label_set"],
+        "verification": verification,
+        "reasons": verification["reasons"],
+        "safe_next_action": verification["safe_next_action"],
+    }
+
+
+def clear_pr_open_label(
+    issue_numbers: list[int],
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    *,
+    terminal_reason: str,
+) -> dict:
+    """Authoritative ``status:pr-open`` cleanup for terminal PR states (#780).
+
+    Every sanctioned terminal path routes here — merge, close-without-merge,
+    supersession/abandonment, already-landed reconciliation, controller
+    closure, and retry/recovery — so the rule cannot drift between them. Only
+    ``status:pr-open`` is ever removed; all other labels are preserved, and an
+    empty resulting set is a valid outcome.
+    """
+    reason = terminal_pr_label_cleanup.canonical_terminal_reason(terminal_reason)
+
+    numbers: list[int] = []
+    for raw in issue_numbers or []:
+        try:
+            num = int(raw)
+        except (TypeError, ValueError):
+            continue
+        if num not in numbers:
+            numbers.append(num)
+
+    if not numbers:
+        return terminal_pr_label_cleanup.summarize_cleanup_results(
+            [], terminal_reason=reason
+        )
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    base = repo_api_url(h, o, r)
+    audit_kwargs = {"host": h, "remote": remote, "org": o, "repo": r}
+
+    # Memoized, lazily resolved label inventory: an issue that no longer
+    # carries the label needs no ids at all, so a clean repository costs one
+    # read per issue and nothing else.
+    cached: dict[str, dict[str, int]] = {}
+
+    def resolve_label_ids() -> dict[str, int]:
+        if "map" not in cached:
+            cached["map"] = _repo_label_id_map(base, auth)
+        return cached["map"]
+
+    results = [
+        _clear_pr_open_label_for_issue(
+            base=base,
+            auth=auth,
+            issue_number=num,
+            terminal_reason=reason,
+            resolve_label_ids=resolve_label_ids,
+            audit_kwargs=audit_kwargs,
+        )
+        for num in numbers
+    ]
+    return terminal_pr_label_cleanup.summarize_cleanup_results(
+        results, terminal_reason=reason
+    )
+
+
+def cleanup_in_progress_for_pr(
+    pr_payload: dict,
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    *,
+    terminal_reason: str = terminal_pr_label_cleanup.MERGED,
+) -> dict:
     body = pr_payload.get("body") or ""
     title = pr_payload.get("title") or ""
     branch = pr_payload.get("head", {}).get("ref") or ""
@@ -2833,10 +3024,38 @@ def cleanup_in_progress_for_pr(pr_payload: dict, remote: str, host: str | None,
     issues = extract_linked_issue_numbers(text, branch)
 
     if not issues:
-        return {"cleanup_status": "no linked issue found"}
+        return {
+            "cleanup_status": "no linked issue found",
+            "pr_open_label_cleanup": terminal_pr_label_cleanup.summarize_cleanup_results(
+                [], terminal_reason=terminal_reason
+            ),
+        }
 
     results = release_in_progress_label(issues, remote, host, org, repo)
-    return {"cleanup_status": results}
+    # #780: the PR has reached a terminal state, so status:pr-open must go too.
+    # Never raises — a failed cleanup is reported, never silently dropped, and
+    # never undoes the terminal transition that already happened.
+    try:
+        pr_open_cleanup = clear_pr_open_label(
+            issues, remote, host, org, repo, terminal_reason=terminal_reason
+        )
+    except Exception as exc:  # noqa: BLE001 — redact before surfacing
+        pr_open_cleanup = {
+            "label": terminal_pr_label_cleanup.PR_OPEN_LABEL,
+            "clean": False,
+            "checked": issues,
+            "removed": [],
+            "already_absent": [],
+            "failed": issues,
+            "results": [],
+            "reasons": [f"terminal label cleanup failed: {_redact(str(exc))}"],
+            "safe_next_action": (
+                "Re-run gitea_cleanup_terminal_pr_labels with "
+                f"terminal_reason='{terminal_pr_label_cleanup.RETRY_RECOVERY}' "
+                "for the linked issues."
+            ),
+        }
+    return {"cleanup_status": results, "pr_open_label_cleanup": pr_open_cleanup}
 
 # ── Helpers ───────────────────────────────────────────────────────────────────
 
@@ -8357,9 +8576,18 @@ def gitea_edit_pr(
         data = api_request("PATCH", url, auth, payload)
 
     cleanup_status = None
+    pr_open_label_cleanup = None
     if state == "closed":
-        cleanup = cleanup_in_progress_for_pr(data, remote, host, org, repo)
+        cleanup = cleanup_in_progress_for_pr(
+            data,
+            remote,
+            host,
+            org,
+            repo,
+            terminal_reason=terminal_pr_label_cleanup.CLOSED_WITHOUT_MERGE,
+        )
         cleanup_status = cleanup.get("cleanup_status")
+        pr_open_label_cleanup = cleanup.get("pr_open_label_cleanup")
         if isinstance(cleanup_status, dict):
             for issue_num, st in cleanup_status.items():
                 if st == "released":
@@ -8376,6 +8604,7 @@ def gitea_edit_pr(
         "body": data.get("body", ""),
         "state": data["state"],
         "cleanup_status": cleanup_status,
+        "pr_open_label_cleanup": pr_open_label_cleanup,
     }
     return _with_optional_url(result, data.get("html_url"))
 
@@ -9111,12 +9340,33 @@ def gitea_merge_pr(
             result["cleanup_status"] = "skipped (merge read-back failed)"
         else:
             try:
-                cleanup = cleanup_in_progress_for_pr(merged, remote, host, org, repo)
+                cleanup = cleanup_in_progress_for_pr(
+                    merged,
+                    remote,
+                    host,
+                    org,
+                    repo,
+                    terminal_reason=terminal_pr_label_cleanup.MERGED,
+                )
                 result["cleanup_status"] = cleanup.get("cleanup_status")
+                result["pr_open_label_cleanup"] = cleanup.get("pr_open_label_cleanup")
             except Exception as cleanup_exc:  # noqa: BLE001 — redact before surfacing
                 result["cleanup_status"] = (
                     f"skipped (cleanup error: {_redact(str(cleanup_exc))})"
                 )
+                result["pr_open_label_cleanup"] = {
+                    "label": terminal_pr_label_cleanup.PR_OPEN_LABEL,
+                    "clean": False,
+                    "reasons": [
+                        f"cleanup error: {_redact(str(cleanup_exc))}"
+                    ],
+                    "safe_next_action": (
+                        "Re-run gitea_cleanup_terminal_pr_labels with "
+                        f"terminal_reason="
+                        f"'{terminal_pr_label_cleanup.RETRY_RECOVERY}' for the "
+                        "linked issues."
+                    ),
+                }
     except Exception as exc:  # noqa: BLE001 — redact before surfacing
         reasons.append(f"merge failed: {_redact(str(exc))}")
         return result
@@ -11104,6 +11354,20 @@ def gitea_reconcile_already_landed_pr(
             result["issue_closed"] = True
             result["performed"] = True
 
+    # #780: once the PR is closed as already-landed, the linked issue is no
+    # longer awaiting an open PR — retire status:pr-open even when the issue
+    # itself stays open. Reported, never fatal: the reconciliation already
+    # happened and must not be reversed by a label failure.
+    if linked_issue and (result.get("pr_closed") or result.get("issue_closed")):
+        result["pr_open_label_cleanup"] = clear_pr_open_label(
+            [linked_issue],
+            remote,
+            host,
+            org,
+            repo,
+            terminal_reason=terminal_pr_label_cleanup.ALREADY_LANDED,
+        )
+
     return result
 
 
@@ -11359,6 +11623,20 @@ def gitea_reconcile_superseded_by_merged_pr(
         result["issue_closed"] = True
         result["performed"] = True
 
+    # #780: supersession/abandonment is terminal for the target PR, so the
+    # target issue must not keep advertising an open PR.
+    if target_issue_number and (
+        result.get("pr_closed") or result.get("issue_closed")
+    ):
+        result["pr_open_label_cleanup"] = clear_pr_open_label(
+            [target_issue_number],
+            remote,
+            host,
+            org,
+            repo,
+            terminal_reason=terminal_pr_label_cleanup.SUPERSEDED,
+        )
+
     return result
 
 @mcp.tool()
@@ -11417,6 +11695,33 @@ def gitea_close_issue(
             "reasons": closure_gate["reasons"],
             "safe_next_action": closure_gate["safe_next_action"],
         }
+    # #780: controller closure is a terminal transition, so status:pr-open must
+    # be retired. Run it *before* the state change and fail closed if it cannot
+    # be completed — closing first would leave the exact stale-label state this
+    # gate exists to prevent, with no remaining step to catch it.
+    pr_open_cleanup = clear_pr_open_label(
+        [issue_number],
+        remote,
+        host,
+        org,
+        repo,
+        terminal_reason=terminal_pr_label_cleanup.CONTROLLER_CLOSURE,
+    )
+    if not pr_open_cleanup["clean"]:
+        return {
+            "success": False,
+            "blocked": True,
+            "performed": False,
+            "message": (
+                f"Issue #{issue_number} close blocked (#780): the terminal "
+                f"'{terminal_pr_label_cleanup.PR_OPEN_LABEL}' cleanup could not "
+                "be completed and verified."
+            ),
+            "pr_open_label_cleanup": pr_open_cleanup,
+            "reasons": pr_open_cleanup["reasons"],
+            "safe_next_action": pr_open_cleanup["safe_next_action"],
+        }
+
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -11426,13 +11731,201 @@ def gitea_close_issue(
 
     cleanup_result = release_in_progress_label([issue_number], remote, host, org, repo)
 
+    # Terminal validation (#780): re-read the closed issue and report any
+    # residual status:pr-open rather than assuming the earlier cleanup held.
+    try:
+        closed_labels = _issue_label_names(
+            repo_api_url(h, o, r), auth, issue_number
+        )
+        terminal_validation = terminal_pr_label_cleanup.detect_residual_pr_open(
+            [{"number": issue_number, "state": "closed", "labels": closed_labels}]
+        )
+    except Exception as exc:  # noqa: BLE001 — redact before surfacing
+        terminal_validation = {
+            "label": terminal_pr_label_cleanup.PR_OPEN_LABEL,
+            "clean": False,
+            "checked_count": 0,
+            "residual_count": None,
+            "residual_issues": [],
+            "reasons": [
+                f"terminal label validation read-back failed: {_redact(str(exc))}"
+            ],
+            "safe_next_action": (
+                "Re-check the issue labels with gitea_assess_terminal_label_hygiene."
+            ),
+        }
+
     return {
         "success": True,
         "message": f"Issue #{issue_number} closed.",
-        "cleanup_status": cleanup_result
+        "cleanup_status": cleanup_result,
+        "pr_open_label_cleanup": pr_open_cleanup,
+        "terminal_label_validation": terminal_validation,
     }
 
 
+@mcp.tool()
+def gitea_cleanup_terminal_pr_labels(
+    issue_numbers: list[int],
+    terminal_reason: str = terminal_pr_label_cleanup.RETRY_RECOVERY,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Retire ``status:pr-open`` after a terminal PR transition (#780).
+
+    This is the sanctioned recovery path when a terminal transition completed
+    but its label cleanup did not — for example a merge whose read-back failed,
+    or a workflow interrupted between closing a PR and updating its issue.
+
+    The rule is the same one every terminal path uses: remove only
+    ``status:pr-open``, preserve every other label, allow an empty resulting
+    set, and verify by read-after-write. Calling it on an issue that no longer
+    carries the label is a harmless no-op, so retries are always safe.
+
+    Args:
+        issue_numbers: Issues whose terminal label state should be repaired.
+        terminal_reason: Why the PR is terminal — one of merged,
+            closed_without_merge, superseded, already_landed,
+            controller_closure, abandoned, retry_recovery.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+        worktree_path: Optional branches worktree to verify before mutation.
+
+    Returns:
+        dict with 'clean', per-issue 'results' carrying read-after-write proof,
+        'removed', 'already_absent', 'failed', and 'safe_next_action'.
+    """
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission("cleanup_terminal_pr_labels"),
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        org_explicit=org is not None,
+        repo_explicit=repo is not None,
+    )
+    if blocked:
+        return blocked
+    verify_preflight_purity(
+        remote,
+        worktree_path=worktree_path,
+        task="cleanup_terminal_pr_labels",
+        org=org,
+        repo=repo,
+    )
+
+    try:
+        reason = terminal_pr_label_cleanup.canonical_terminal_reason(terminal_reason)
+    except ValueError as exc:
+        return {
+            "success": False,
+            "clean": False,
+            "performed": False,
+            "reasons": [str(exc)],
+            "safe_next_action": (
+                "Re-call with terminal_reason set to one of: "
+                + ", ".join(terminal_pr_label_cleanup.TERMINAL_REASONS)
+            ),
+        }
+
+    summary = clear_pr_open_label(
+        issue_numbers, remote, host, org, repo, terminal_reason=reason
+    )
+    summary["success"] = summary["clean"]
+    summary["performed"] = bool(summary["removed"])
+    return summary
+
+
+@mcp.tool()
+def gitea_assess_terminal_label_hygiene(
+    state: str = "all",
+    limit: int = 500,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only terminal validation for residual ``status:pr-open`` (#780).
+
+    Enumerates issues and the live open pull requests, then reports any issue
+    still carrying ``status:pr-open`` without an open PR to justify it. Use it
+    before declaring a terminal transition — or a cleanup batch — complete.
+
+    Args:
+        state: Issue state filter — 'open', 'closed', or 'all' (default).
+        limit: Max issues to enumerate across all pages.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with 'clean', 'residual_count', 'residual_issues',
+        'open_pr_count', and 'safe_next_action'.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "clean": 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)
+
+    try:
+        issues = api_get_all(
+            f"{base}/issues?state={state}&type=issues", auth, limit=limit
+        ) or []
+    except Exception as exc:  # noqa: BLE001 — redact before surfacing
+        return {
+            "success": False,
+            "clean": False,
+            "reasons": [f"could not enumerate issues: {_redact(str(exc))}"],
+            "safe_next_action": (
+                "Retry once the Gitea API is reachable; an incomplete "
+                "enumeration cannot prove terminal label hygiene."
+            ),
+        }
+
+    try:
+        open_pulls = _list_open_pulls(h, o, r, auth)
+    except Exception as exc:  # noqa: BLE001 — redact before surfacing
+        return {
+            "success": False,
+            "clean": False,
+            "reasons": [f"could not enumerate open pull requests: {_redact(str(exc))}"],
+            "safe_next_action": (
+                "Retry once the Gitea API is reachable; without the open-PR "
+                "inventory a legitimate status:pr-open cannot be distinguished "
+                "from a stale one."
+            ),
+        }
+
+    open_pr_issues: list[int] = []
+    for pull in open_pulls:
+        head_obj = pull.get("head") or {}
+        branch = head_obj.get("ref") if isinstance(head_obj, dict) else None
+        text = f"{pull.get('title') or ''}\n{pull.get('body') or ''}"
+        open_pr_issues.extend(extract_linked_issue_numbers(text, branch))
+
+    assessment = terminal_pr_label_cleanup.detect_residual_pr_open(
+        issues, open_pr_issue_numbers=open_pr_issues
+    )
+    assessment["success"] = True
+    assessment["issue_state_filter"] = state
+    assessment["open_pr_count"] = len(open_pulls)
+    return assessment
+
+
 @mcp.tool()
 def gitea_list_issues(
     state: str = "open",
@@ -19202,7 +19695,9 @@ def gitea_allocate_next_work(
     repo: str | None = None,
     include_issues: bool = True,
     include_prs: bool = True,
-    candidates_json: str | None = None,
+    candidates_json: Any = None,
+    exclude_issue_numbers: list[int] | None = None,
+    expected_candidate_set_fingerprint: str | None = None,
     limit: int = 50,
 ) -> dict:
     """Controller-owned next-work allocator using the #613 control-plane DB (#600).
@@ -19216,10 +19711,18 @@ def gitea_allocate_next_work(
     leases as the coordination source).
 
     Outcomes include: ``assigned_work``, ``preview``, ``wait``,
-    ``blocked_by_terminal_path``, ``no_safe_work``, ``role_ineligible``.
+    ``blocked_by_terminal_path``, ``no_safe_work``, ``role_ineligible``,
+    ``blocked_by_excluded_own_lease``, ``candidate_set_drift``.
 
-    *candidates_json* may inject a JSON list of candidate dicts (tests /
-    controller overrides). When omitted, open issues/PRs are loaded from Gitea.
+    *candidates_json* may inject a candidate list as an already-decoded list
+    (native MCP transport) or a JSON string (backward compatible). When
+    omitted, open issues/PRs are loaded from Gitea.
+
+    *exclude_issue_numbers* (#776): typed list of issue/PR numbers removed
+    before ranking. Omitted/empty retains prior behavior.
+
+    *expected_candidate_set_fingerprint* (#776 AC4): optional CAS pin from a
+    prior dry-run; apply fails closed on material candidate-set drift.
 
     #612 remains downstream: raw monitoring incidents are never candidates.
     """
@@ -19273,15 +19776,12 @@ def gitea_allocate_next_work(
 
     inv_reasons: list[str] = []
     candidates: list[Any] = []
-    if candidates_json:
+    if candidates_json is not None and candidates_json != "":
         try:
-            raw = json.loads(candidates_json)
-            if not isinstance(raw, list):
-                raise ValueError("candidates_json must be a JSON list")
-            for item in raw:
-                if not isinstance(item, dict):
-                    continue
-                candidates.append(allocator_service.candidate_from_dict(item))
+            # #776 AC3: accept decoded list or JSON string; reject malformed.
+            candidates = allocator_service.normalize_candidates_payload(
+                candidates_json
+            )
         except Exception as exc:  # noqa: BLE001
             return {
                 "success": False,
@@ -19324,23 +19824,40 @@ def gitea_allocate_next_work(
     sid = (session_id or "").strip() or (
         f"{profile_name or 'session'}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
     )
-    result = allocator_service.allocate_next_work(
-        db,
-        session_id=sid,
-        role=role_in,
-        remote=remote if remote in REMOTES else remote,
-        org=o,
-        repo=r,
-        candidates=candidates,
-        apply=bool(apply),
-        profile_name=profile_name,
-        username=username,
-    )
+    try:
+        result = allocator_service.allocate_next_work(
+            db,
+            session_id=sid,
+            role=role_in,
+            remote=remote if remote in REMOTES else remote,
+            org=o,
+            repo=r,
+            candidates=candidates,
+            apply=bool(apply),
+            profile_name=profile_name,
+            username=username,
+            controller_instance_id=allocator_service.resolve_controller_instance_id(),
+            exclude_issue_numbers=exclude_issue_numbers,
+            expected_candidate_set_fingerprint=expected_candidate_set_fingerprint,
+        )
+    except ValueError as exc:
+        return {
+            "success": False,
+            "outcome": allocator_service.OUTCOME_NO_SAFE,
+            "apply": bool(apply),
+            "reasons": [
+                f"invalid allocator input: {_redact(str(exc))} (fail closed, #776)"
+            ],
+            "assignment": None,
+            "substrate": "control_plane_db",
+        }
     if inv_reasons:
         result.setdefault("inventory_warnings", inv_reasons)
     result["candidate_count"] = len(candidates)
     result["inventory_source"] = (
-        "candidates_json" if candidates_json else "gitea_live"
+        "candidates_json"
+        if candidates_json is not None and candidates_json != ""
+        else "gitea_live"
     )
     result["inventory_complete"] = True
     result["selection_policy"] = allocator_service.SELECTION_POLICY
@@ -19434,7 +19951,7 @@ def gitea_workflow_dashboard(
     include_issues: bool = True,
     include_prs: bool = True,
     include_non_active_leases: bool = True,
-    candidates_json: str | None = None,
+    candidates_json: Any = None,
     limit: int = 100,
 ) -> dict:
     """Read-only MCP workflow dashboard: queues, leases, blockers, next safe action (#605).
@@ -19443,9 +19960,10 @@ def gitea_workflow_dashboard(
     still requires ``gitea_allocate_next_work``. Never presents blocked or
     terminal-locked items as safe next work.
 
-    *candidates_json* may inject a JSON list of candidate dicts (tests /
-    offline fixtures). When omitted, open issues/PRs are loaded from Gitea.
-    Incomplete live inventory fails closed (no safe suggestions).
+    *candidates_json* may inject a candidate list as an already-decoded list
+    or JSON string (#776 transport parity). When omitted, open issues/PRs are
+    loaded from Gitea. Incomplete live inventory fails closed (no safe
+    suggestions).
     """
     read_block = _profile_operation_gate("gitea.read")
     if read_block:
@@ -19470,15 +19988,11 @@ def gitea_workflow_dashboard(
     inv_reasons: list[str] = []
     candidates: list[Any] = []
     inventory_complete = True
-    if candidates_json:
+    if candidates_json is not None and candidates_json != "":
         try:
-            raw = json.loads(candidates_json)
-            if not isinstance(raw, list):
-                raise ValueError("candidates_json must be a JSON list")
-            for item in raw:
-                if not isinstance(item, dict):
-                    continue
-                candidates.append(allocator_service.candidate_from_dict(item))
+            candidates = allocator_service.normalize_candidates_payload(
+                candidates_json
+            )
         except Exception as exc:  # noqa: BLE001
             return {
                 "success": False,
@@ -19502,6 +20016,7 @@ def gitea_workflow_dashboard(
     leases: list[dict] = []
     terminal_pr: int | None = None
     terminal_lock: dict | None = None
+    claims: dict = {}
     db, db_errs = _control_plane_db_or_error()
     if db is None:
         inv_reasons.extend(db_errs or ["control-plane DB unavailable"])
@@ -19540,6 +20055,18 @@ def gitea_workflow_dashboard(
             inv_reasons.append(
                 f"terminal lock lookup failed: {_redact(str(exc))}"
             )
+        try:
+            # #765: live claims so the dashboard never advertises another
+            # controller's active task as safe next work.
+            claims = db.list_active_claims(
+                remote=remote if remote in REMOTES else remote,
+                org=o,
+                repo=r,
+            )
+        except Exception as exc:  # noqa: BLE001
+            inv_reasons.append(
+                f"active claim lookup failed: {_redact(str(exc))}"
+            )
 
     snap = workflow_dashboard.build_workflow_dashboard(
         candidates=candidates,
@@ -19551,10 +20078,14 @@ def gitea_workflow_dashboard(
         terminal_lock=terminal_lock,
         inventory_complete=inventory_complete,
         inventory_reasons=inv_reasons,
+        claims=claims,
+        controller_instance_id=allocator_service.resolve_controller_instance_id(),
     )
     payload = snap.as_dict()
     payload["inventory_source"] = (
-        "candidates_json" if candidates_json else "gitea_live"
+        "candidates_json"
+        if candidates_json is not None and candidates_json != ""
+        else "gitea_live"
     )
     payload["limit_applies_to"] = "lease_listing_only"
     return payload

mutation_budget_classifier.py

@@ -0,0 +1,286 @@
+"""Mutation-budget classification for auto-mode attempts (#617).
+
+Mutation budget must count only *server-side* Gitea state changes. A tool call
+that fails closed before the Gitea API is reached changed nothing on the
+server, so it must not consume the budget that protects against repeated real
+mutations.
+
+The classifier separates four outcome classes plus an explicit ambiguous class:
+
+``local_validator_rejection``
+    A canonical-content validator (for example the ``[THREAD STATE LEDGER]`` or
+    ``## Canonical Issue State`` blocks) rejected the payload before any API
+    call. No server-side state exists.
+
+``capability_gate_rejection``
+    A profile/permission gate refused the operation before any API call.
+
+``transport_failure_before_api``
+    The request never reached the Gitea API (transport/EOF/connection error).
+
+``server_side_mutation``
+    The API succeeded and returned proof of durable state (comment id, review
+    id, merge commit, label result, or an issue/PR state change).
+
+``ambiguous_requires_readback``
+    The API *was* reached but the result carries no usable proof either way.
+    This fails closed: the attempt is treated as budget-consuming until a
+    read-after-write check proves otherwise, so #617 never weakens the guard
+    that prevents repeated real mutations.
+
+Only ``server_side_mutation`` consumes budget outright. Every attempt — failed
+or not — is still recorded in the local attempt ledger so a final report can
+show local failed attempts, blocked API attempts, and successful server-side
+mutations separately.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+LOCAL_VALIDATOR_REJECTION = "local_validator_rejection"
+CAPABILITY_GATE_REJECTION = "capability_gate_rejection"
+TRANSPORT_FAILURE_BEFORE_API = "transport_failure_before_api"
+SERVER_SIDE_MUTATION = "server_side_mutation"
+AMBIGUOUS_REQUIRES_READBACK = "ambiguous_requires_readback"
+
+CLASSIFICATIONS = (
+    LOCAL_VALIDATOR_REJECTION,
+    CAPABILITY_GATE_REJECTION,
+    TRANSPORT_FAILURE_BEFORE_API,
+    SERVER_SIDE_MUTATION,
+    AMBIGUOUS_REQUIRES_READBACK,
+)
+
+#: Result fields that prove durable server-side state was created.
+MUTATION_PROOF_FIELDS = (
+    "comment_id",
+    "review_id",
+    "merge_commit_sha",
+    "label_result",
+    "state_change",
+    "created_pr_number",
+)
+
+#: Classes that never consume server-side mutation budget.
+PRE_API_CLASSIFICATIONS = (
+    LOCAL_VALIDATOR_REJECTION,
+    CAPABILITY_GATE_REJECTION,
+    TRANSPORT_FAILURE_BEFORE_API,
+)
+
+FINAL_REPORT_REQUIRED_FIELDS = (
+    "local_failed_attempts",
+    "blocked_api_attempts",
+    "successful_server_mutations",
+)
+
+
+def _clean(value: Any) -> str:
+    return (value or "").strip() if isinstance(value, str) else str(value or "").strip()
+
+
+def _proof_fields_present(result: dict) -> list[str]:
+    """Return the mutation-proof fields carrying a usable value."""
+    present: list[str] = []
+    for field in MUTATION_PROOF_FIELDS:
+        value = result.get(field)
+        if value is None or value is False:
+            continue
+        if isinstance(value, str) and not value.strip():
+            continue
+        present.append(field)
+    return present
+
+
+def _decision(
+    classification: str,
+    *,
+    budget_consumed: bool,
+    requires_readback: bool,
+    reasons: list[str],
+    proof_fields: list[str],
+    api_called: bool | None,
+) -> dict:
+    return {
+        "classification": classification,
+        "budget_consumed": budget_consumed,
+        "requires_readback": requires_readback,
+        "pre_api": classification in PRE_API_CLASSIFICATIONS,
+        "api_called": api_called,
+        "proof_fields": proof_fields,
+        "reasons": reasons,
+    }
+
+
+def classify_mutation_attempt(result: dict | None) -> dict:
+    """Classify one mutation attempt and decide whether it consumes budget.
+
+    ``result`` is the raw dict a Gitea MCP tool returned. The caller does not
+    pre-interpret it: classification is driven by the explicit ``api_called``
+    signal plus the proof fields the tool reports.
+    """
+    data = dict(result or {})
+    success = bool(data.get("success"))
+    proof_fields = _proof_fields_present(data)
+    api_called = data.get("api_called")
+
+    # An unambiguous success carrying durable proof is a real mutation however
+    # the attempt was labelled upstream.
+    if success and proof_fields:
+        return _decision(
+            SERVER_SIDE_MUTATION,
+            budget_consumed=True,
+            requires_readback=False,
+            reasons=[
+                "API reported success with durable proof field(s): "
+                + ", ".join(proof_fields)
+            ],
+            proof_fields=proof_fields,
+            api_called=True,
+        )
+
+    if api_called is False:
+        # Nothing reached the server; pick the precise pre-API class.
+        if data.get("transport_error") or data.get("transport_failed"):
+            return _decision(
+                TRANSPORT_FAILURE_BEFORE_API,
+                budget_consumed=False,
+                requires_readback=False,
+                reasons=["transport failed before the Gitea API was reached"],
+                proof_fields=[],
+                api_called=False,
+            )
+        if data.get("permission_report") or data.get("capability_blocked"):
+            return _decision(
+                CAPABILITY_GATE_REJECTION,
+                budget_consumed=False,
+                requires_readback=False,
+                reasons=["capability/permission gate refused before any API call"],
+                proof_fields=[],
+                api_called=False,
+            )
+        return _decision(
+            LOCAL_VALIDATOR_REJECTION,
+            budget_consumed=False,
+            requires_readback=False,
+            reasons=[
+                "local validator rejected the payload before any API call; "
+                "no server-side state was created"
+            ],
+            proof_fields=[],
+            api_called=False,
+        )
+
+    if api_called is True:
+        if success:
+            reason = (
+                "API reported success but returned no durable proof field; "
+                "read-after-write verification required before counting budget"
+            )
+        else:
+            reason = (
+                "API was reached and the outcome carries no durable proof; "
+                "read-after-write verification required before counting budget"
+            )
+        return _decision(
+            AMBIGUOUS_REQUIRES_READBACK,
+            budget_consumed=True,
+            requires_readback=True,
+            reasons=[reason],
+            proof_fields=proof_fields,
+            api_called=True,
+        )
+
+    # ``api_called`` was not reported at all. Fail closed rather than assuming
+    # nothing happened.
+    return _decision(
+        AMBIGUOUS_REQUIRES_READBACK,
+        budget_consumed=True,
+        requires_readback=True,
+        reasons=[
+            "attempt did not report 'api_called'; cannot prove the request "
+            "stopped before the Gitea API, so the attempt fails closed"
+        ],
+        proof_fields=proof_fields,
+        api_called=None,
+    )
+
+
+def record_attempt(
+    ledger: list[dict] | None,
+    result: dict | None,
+    *,
+    operation: str = "",
+    timestamp: str | None = None,
+) -> dict:
+    """Append one classified attempt to the local ledger and return the entry.
+
+    Every attempt is recorded, including the ones that consume no budget: the
+    point of #617 is that failed local attempts stay visible without being
+    miscounted as Gitea mutations.
+    """
+    entries = ledger if isinstance(ledger, list) else []
+    entry = {
+        "operation": _clean(operation),
+        "timestamp": _clean(timestamp) or datetime.now(timezone.utc).isoformat(),
+        **classify_mutation_attempt(result),
+    }
+    entries.append(entry)
+    return entry
+
+
+def summarize_attempt_ledger(ledger: list[dict] | None) -> dict:
+    """Summarize a ledger into the categories a final report must show."""
+    entries = [e for e in (ledger or []) if isinstance(e, dict)]
+
+    def _count(*classifications: str) -> int:
+        return sum(1 for e in entries if e.get("classification") in classifications)
+
+    return {
+        "total_attempts": len(entries),
+        "local_failed_attempts": _count(LOCAL_VALIDATOR_REJECTION),
+        "blocked_api_attempts": _count(
+            CAPABILITY_GATE_REJECTION, TRANSPORT_FAILURE_BEFORE_API
+        ),
+        "successful_server_mutations": _count(SERVER_SIDE_MUTATION),
+        "ambiguous_attempts": _count(AMBIGUOUS_REQUIRES_READBACK),
+        "budget_consumed": sum(1 for e in entries if e.get("budget_consumed")),
+        "requires_readback": any(e.get("requires_readback") for e in entries),
+        "entries": entries,
+    }
+
+
+def assess_final_report_mutation_accounting(
+    report: dict | None,
+    ledger: list[dict] | None,
+) -> dict:
+    """Fail closed when a report's mutation accounting contradicts the ledger."""
+    data = dict(report or {})
+    summary = summarize_attempt_ledger(ledger)
+    reasons: list[str] = []
+
+    for field in FINAL_REPORT_REQUIRED_FIELDS:
+        if field not in data:
+            reasons.append(f"final report omits required field '{field}'")
+            continue
+        claimed = data.get(field)
+        actual = summary[field]
+        if claimed != actual:
+            reasons.append(
+                f"final report claims {field}={claimed} but the attempt ledger "
+                f"shows {actual}"
+            )
+
+    if summary["requires_readback"] and not data.get("readback_verified"):
+        reasons.append(
+            "ledger contains an ambiguous attempt; final report must record "
+            "'readback_verified' proof before claiming mutation accounting"
+        )
+
+    return {
+        "valid": not reasons,
+        "reasons": reasons,
+        "ledger_summary": {k: v for k, v in summary.items() if k != "entries"},
+    }

self_propagating_handoff.py

@@ -0,0 +1,907 @@
+"""Self-propagating canonical handoffs through final controller closure (#626).
+
+#494-#507 defined the canonical ledger, next-action comments, comment
+validation, the controller acceptance gate, and the Canonical Thread Handoff
+(CTH) shape.  What none of them enforce is the *chain*: that every actor
+consumes exactly one canonical handoff, performs exactly one authorized role,
+records the result durably in Gitea, and emits the next complete handoff until
+the controller records final closure.
+
+This module owns that systemic gap:
+
+* one canonical cross-role handoff schema (:data:`HANDOFF_FIELDS`);
+* a fail-closed validator that rejects incomplete handoffs;
+* live-state recovery so a receiving actor never trusts an inherited handoff;
+* role-limited continuation;
+* mandatory durable posting into Gitea;
+* the ``merged-awaiting-controller`` boundary and controller accept/reject
+  continuation;
+* workflow-failure escalation into separate durable issues, with duplicate
+  handling;
+* terminal closure that must *not* emit an unnecessary next prompt.
+
+Everything here is pure assessment: no network calls, no mutation.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Iterable, Mapping, Sequence
+
+MARKER = "<!-- sph:v1 -->"
+HANDOFF_HEADING = "Canonical Handoff"
+
+#: Canonical workflow states a handoff may declare.
+WORKFLOW_STATES: tuple[str, ...] = (
+    "needs-author",
+    "needs-review",
+    "approved-awaiting-merge",
+    "merged-awaiting-controller",
+    "blocked",
+    "complete",
+)
+
+TERMINAL_STATES = frozenset({"complete"})
+
+#: The single role authorized to act on each workflow state.
+NEXT_ACTOR_BY_STATE: dict[str, str] = {
+    "needs-author": "author",
+    "needs-review": "reviewer",
+    "approved-awaiting-merge": "merger",
+    "merged-awaiting-controller": "controller",
+    "blocked": "operator",
+    "complete": "none",
+}
+
+WORKFLOW_ROLES = frozenset(
+    {"author", "reviewer", "merger", "controller", "operator", "reconciler"}
+)
+
+#: What each receiving role is authorized to do when it consumes a handoff.
+ROLE_ALLOWED_ACTIONS: dict[str, tuple[str, ...]] = {
+    "author": ("implement", "commit", "push", "create_pr", "comment"),
+    "reviewer": ("review", "approve", "request_changes", "comment"),
+    "merger": ("verify_approval_parity", "merge", "comment"),
+    "controller": ("accept", "reject", "reopen", "close_issue", "comment"),
+    "operator": ("repair_infrastructure", "comment"),
+    "reconciler": ("close_superseded_pr", "cleanup_branch", "comment"),
+}
+
+ROLE_FORBIDDEN_ACTIONS: dict[str, tuple[str, ...]] = {
+    "author": ("approve", "request_changes", "merge", "close_issue"),
+    "reviewer": ("merge", "commit", "push", "create_pr"),
+    "merger": ("approve", "commit", "push", "create_pr"),
+    "controller": ("approve", "merge", "commit", "push"),
+    "operator": ("approve", "merge", "close_issue"),
+    "reconciler": ("approve", "merge", "commit", "push", "create_pr"),
+}
+
+#: Ordered canonical handoff fields.  Every one of them is required; the
+#: fields in :data:`NONE_ALLOWED_FIELDS` may legitimately carry ``none``.
+HANDOFF_FIELDS: tuple[str, ...] = (
+    "REPOSITORY",
+    "ISSUE",
+    "PR",
+    "WORKFLOW_STATE",
+    "HEAD_SHA",
+    "BASE_BRANCH",
+    "BASE_OR_MERGE_SHA",
+    "ACTING_ROLE",
+    "ACTING_IDENTITY",
+    "COMPLETED_ACTIONS",
+    "VALIDATION_EVIDENCE",
+    "MUTATION_LEDGER",
+    "BLOCKERS",
+    "NEXT_ACTOR",
+    "NEXT_ACTION",
+    "PROHIBITED_ACTIONS",
+    "NEXT_PROMPT",
+    "WORKFLOW_FAILURE_ISSUES",
+    "LAST_UPDATED",
+)
+
+NONE_ALLOWED_FIELDS = frozenset(
+    {
+        "PR",
+        "HEAD_SHA",
+        "BASE_OR_MERGE_SHA",
+        "BLOCKERS",
+        "WORKFLOW_FAILURE_ISSUES",
+        "NEXT_PROMPT",
+        "NEXT_ACTION",
+    }
+)
+
+#: States where no PR or head SHA exists yet, so ``none`` is legitimate.
+_PRE_PR_STATES = frozenset({"needs-author", "blocked"})
+
+_PLACEHOLDERS = frozenset({"", "none", "n/a", "na", "tbd", "todo", "unknown", "?"})
+
+#: A next prompt short enough to be a stub cannot be "ready to run".
+MIN_NEXT_PROMPT_CHARS = 40
+
+_FIELD_LINE_RE = re.compile(r"^([A-Z][A-Z0-9_]*)\s*:\s*(.*)$", re.MULTILINE)
+_HEADING_RE = re.compile(r"^##\s*Canonical Handoff\s*$", re.IGNORECASE | re.MULTILINE)
+_EXTERNAL_CHAT_RE = re.compile(
+    r"\b(?:previous chat|prior conversation|earlier conversation|see (?:the )?chat|"
+    r"chat history|paste (?:this )?(?:from|into) chatgpt|ask the operator to paste)\b",
+    re.IGNORECASE,
+)
+
+LIVE_DETECTION_KINDS: tuple[str, ...] = (
+    "changed_pr_head",
+    "stale_approval",
+    "issue_closed",
+    "issue_reopened",
+    "pr_merged",
+    "pr_closed_unmerged",
+    "stale_lease",
+    "foreign_lease",
+    "missing_worktree",
+    "dirty_worktree",
+    "namespace_mismatch",
+    "stale_runtime",
+    "changed_base",
+    "conflicting_canonical_comments",
+)
+
+CONTROLLER_DECISIONS = frozenset(
+    {
+        "accept",
+        "request_tests",
+        "request_proof",
+        "request_corrections",
+        "reopen",
+        "return_to_actor",
+    }
+)
+
+CONTROLLER_CLOSURE_PROOF_FIELDS = (
+    "acceptance_criteria_satisfied",
+    "cleanup_complete",
+    "canonical_final_state_posted",
+    "issue_closed_through_workflow",
+)
+
+WORKFLOW_FAILURE_FIELDS = (
+    "classification",
+    "linked_issue",
+    "temporary_impact",
+    "next_valid_actor",
+    "recovery_prompt",
+)
+
+
+def _is_placeholder(value: Any) -> bool:
+    return str(value or "").strip().lower() in _PLACEHOLDERS
+
+
+def _clean(value: Any) -> str:
+    text = str(value).strip() if value is not None else ""
+    return text or "none"
+
+
+# ---------------------------------------------------------------------------
+# rendering / parsing
+# ---------------------------------------------------------------------------
+
+
+def render_self_propagating_handoff(**values: Any) -> str:
+    """Render a canonical cross-role handoff block.
+
+    Raises ``ValueError`` for an unknown workflow state so a malformed handoff
+    can never be produced by the sanctioned renderer.
+    """
+    state = str(values.get("WORKFLOW_STATE", values.get("workflow_state", ""))).strip()
+    if state not in WORKFLOW_STATES:
+        raise ValueError(
+            f"unknown workflow state '{state}'; expected one of {list(WORKFLOW_STATES)}"
+        )
+    lines = [MARKER, f"## {HANDOFF_HEADING}", "", "```text"]
+    for name in HANDOFF_FIELDS:
+        raw = values.get(name, values.get(name.lower()))
+        lines.append(f"{name}: {_clean(raw)}")
+    lines.append("```")
+    return "\n".join(lines)
+
+
+def parse_self_propagating_handoff(text: str) -> dict[str, str] | None:
+    """Parse a canonical handoff block, or ``None`` when absent."""
+    body = text or ""
+    if MARKER not in body and not _HEADING_RE.search(body):
+        return None
+    fields = {
+        match.group(1): match.group(2).strip()
+        for match in _FIELD_LINE_RE.finditer(body)
+    }
+    if not fields:
+        return None
+    return fields
+
+
+def handoff_present(text: str) -> bool:
+    """Whether *text* carries a canonical handoff block at all."""
+    return parse_self_propagating_handoff(text) is not None
+
+
+# ---------------------------------------------------------------------------
+# handoff validation (AC: a validator rejects incomplete handoffs)
+# ---------------------------------------------------------------------------
+
+
+def assess_self_propagating_handoff(text: str) -> dict[str, Any]:
+    """Fail closed unless *text* carries one complete canonical handoff."""
+    fields = parse_self_propagating_handoff(text)
+    if fields is None:
+        return {
+            "valid": False,
+            "block": True,
+            "present": False,
+            "fields": {},
+            "missing_fields": list(HANDOFF_FIELDS),
+            "workflow_state": None,
+            "next_actor": None,
+            "terminal": False,
+            "reasons": ["report or comment carries no canonical handoff block"],
+            "safe_next_action": (
+                "add a canonical handoff block with all "
+                f"{len(HANDOFF_FIELDS)} fields before posting"
+            ),
+        }
+
+    reasons: list[str] = []
+    state = (fields.get("WORKFLOW_STATE") or "").strip()
+    terminal = state in TERMINAL_STATES
+
+    if state not in WORKFLOW_STATES:
+        reasons.append(
+            f"unknown WORKFLOW_STATE '{state or 'missing'}'; "
+            f"expected one of {list(WORKFLOW_STATES)}"
+        )
+
+    missing = [name for name in HANDOFF_FIELDS if name not in fields]
+    reasons.extend(f"handoff missing field: {name}" for name in missing)
+
+    for name in HANDOFF_FIELDS:
+        if name in missing:
+            continue
+        value = fields.get(name, "")
+        if not _is_placeholder(value):
+            continue
+        if name in NONE_ALLOWED_FIELDS:
+            continue
+        # A terminated chain names no next actor by design.
+        if name == "NEXT_ACTOR" and terminal:
+            continue
+        reasons.append(f"handoff field {name} must be concrete, got '{value or ''}'")
+
+    if state and state not in _PRE_PR_STATES and state in WORKFLOW_STATES:
+        for name in ("PR", "HEAD_SHA"):
+            if name not in missing and _is_placeholder(fields.get(name)):
+                reasons.append(
+                    f"handoff field {name} must be concrete in state '{state}'"
+                )
+
+    if state == "blocked" and _is_placeholder(fields.get("BLOCKERS")):
+        reasons.append("state 'blocked' requires a concrete BLOCKERS entry")
+
+    declared_actor = (fields.get("NEXT_ACTOR") or "").strip().lower()
+    expected_actor = NEXT_ACTOR_BY_STATE.get(state)
+    if expected_actor and declared_actor != expected_actor:
+        reasons.append(
+            f"NEXT_ACTOR '{declared_actor or 'missing'}' does not match state "
+            f"'{state}', which authorizes '{expected_actor}'"
+        )
+
+    next_prompt = (fields.get("NEXT_PROMPT") or "").strip()
+    next_action = (fields.get("NEXT_ACTION") or "").strip()
+    if terminal:
+        # A completed workflow terminates; it must not manufacture more work.
+        if not _is_placeholder(next_prompt):
+            reasons.append(
+                "terminal state 'complete' must not carry a NEXT_PROMPT; "
+                "the chain ends at controller closure"
+            )
+        if not _is_placeholder(next_action):
+            reasons.append(
+                "terminal state 'complete' must not carry a NEXT_ACTION"
+            )
+    else:
+        if _is_placeholder(next_prompt):
+            reasons.append(
+                "non-terminal handoff requires a complete ready-to-run NEXT_PROMPT"
+            )
+        elif len(next_prompt) < MIN_NEXT_PROMPT_CHARS:
+            reasons.append(
+                "NEXT_PROMPT is too short to be ready-to-run "
+                f"({len(next_prompt)} < {MIN_NEXT_PROMPT_CHARS} characters)"
+            )
+        if _is_placeholder(next_action):
+            reasons.append("non-terminal handoff requires a concrete NEXT_ACTION")
+
+    acting_role = (fields.get("ACTING_ROLE") or "").strip().lower()
+    if acting_role and acting_role not in WORKFLOW_ROLES:
+        reasons.append(
+            f"unknown ACTING_ROLE '{acting_role}'; expected one of "
+            f"{sorted(WORKFLOW_ROLES)}"
+        )
+
+    block = bool(reasons)
+    return {
+        "valid": not block,
+        "block": block,
+        "present": True,
+        "fields": fields,
+        "missing_fields": missing,
+        "workflow_state": state or None,
+        "next_actor": declared_actor or None,
+        "terminal": terminal,
+        "reasons": reasons,
+        "safe_next_action": (
+            "complete every canonical handoff field before posting"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def assess_thread_recoverability(text: str) -> dict[str, Any]:
+    """The next actor must recover from the thread alone — never outside chat."""
+    assessment = assess_self_propagating_handoff(text)
+    if assessment["block"]:
+        return {
+            "recoverable": False,
+            "block": True,
+            "reasons": assessment["reasons"],
+            "safe_next_action": assessment["safe_next_action"],
+        }
+
+    fields = assessment["fields"]
+    reasons: list[str] = []
+    if assessment["terminal"]:
+        return {
+            "recoverable": True,
+            "block": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+
+    prompt = fields.get("NEXT_PROMPT", "")
+    repository = fields.get("REPOSITORY", "").strip()
+    issue = fields.get("ISSUE", "").strip().lstrip("#")
+
+    if repository and repository.lower() not in prompt.lower():
+        reasons.append("NEXT_PROMPT must name the repository it applies to")
+    if issue and issue not in prompt:
+        reasons.append(f"NEXT_PROMPT must name issue {issue}")
+    if _EXTERNAL_CHAT_RE.search(prompt):
+        reasons.append(
+            "NEXT_PROMPT must not depend on outside chat history; the issue or "
+            "PR thread, workflow docs, and live repository state must suffice"
+        )
+
+    block = bool(reasons)
+    return {
+        "recoverable": not block,
+        "block": block,
+        "reasons": reasons,
+        "safe_next_action": (
+            "rewrite NEXT_PROMPT so it is self-contained" if block else "proceed"
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# live-state recovery (AC: head changes invalidate stale review/merge handoffs)
+# ---------------------------------------------------------------------------
+
+
+def _detection(kind: str, detail: str) -> dict[str, str]:
+    return {"kind": kind, "detail": detail}
+
+
+def assess_handoff_live_state(
+    *,
+    handoff: str | Mapping[str, str],
+    live: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Re-derive workflow truth from live state instead of trusting *handoff*.
+
+    *live* carries observed facts; absent keys are simply not checked, but any
+    fact that contradicts the inherited handoff fails closed.
+    """
+    if isinstance(handoff, Mapping):
+        fields = dict(handoff)
+    else:
+        parsed = parse_self_propagating_handoff(handoff or "")
+        if parsed is None:
+            return {
+                "block": True,
+                "detections": [_detection("missing_handoff", "no canonical handoff")],
+                "kinds": ["missing_handoff"],
+                "reasons": ["no canonical handoff to reconcile against live state"],
+                "recovered_state": None,
+                "safe_next_action": "post a canonical handoff before continuing",
+            }
+        fields = parsed
+
+    state = (fields.get("WORKFLOW_STATE") or "").strip()
+    next_actor = (fields.get("NEXT_ACTOR") or "").strip().lower()
+    detections: list[dict[str, str]] = []
+    recovered_state: str | None = None
+
+    handoff_head = (fields.get("HEAD_SHA") or "").strip()
+    live_head = str(live.get("pr_head_sha") or "").strip()
+    head_changed = bool(
+        live_head and handoff_head and not _is_placeholder(handoff_head)
+        and live_head != handoff_head
+    )
+    if head_changed:
+        detections.append(
+            _detection(
+                "changed_pr_head",
+                f"handoff pinned {handoff_head}, live head is {live_head}",
+            )
+        )
+        if next_actor in {"reviewer", "merger"}:
+            recovered_state = "needs-review"
+
+    approved_head = str(live.get("approved_head_sha") or "").strip()
+    if approved_head and live_head and approved_head != live_head:
+        detections.append(
+            _detection(
+                "stale_approval",
+                f"approval recorded at {approved_head}, live head is {live_head}",
+            )
+        )
+        if next_actor == "merger":
+            recovered_state = "needs-review"
+
+    issue_state = str(live.get("issue_state") or "").strip().lower()
+    if issue_state == "closed" and state not in TERMINAL_STATES:
+        detections.append(
+            _detection("issue_closed", "linked issue is closed but handoff is not complete")
+        )
+    if issue_state == "open" and state in TERMINAL_STATES:
+        detections.append(
+            _detection("issue_reopened", "handoff claims complete but the issue is open")
+        )
+        recovered_state = "needs-author"
+
+    pr_state = str(live.get("pr_state") or "").strip().lower()
+    if pr_state == "merged" and state in {
+        "needs-author",
+        "needs-review",
+        "approved-awaiting-merge",
+    }:
+        detections.append(
+            _detection("pr_merged", "PR is already merged; controller boundary applies")
+        )
+        recovered_state = "merged-awaiting-controller"
+    if pr_state == "closed" and state not in TERMINAL_STATES:
+        detections.append(
+            _detection("pr_closed_unmerged", "PR is closed without merge")
+        )
+
+    lease = live.get("lease") or {}
+    if isinstance(lease, Mapping) and lease:
+        lease_status = str(lease.get("status") or "").strip().lower()
+        if lease_status and lease_status != "active":
+            detections.append(
+                _detection("stale_lease", f"lease status is '{lease_status}'")
+            )
+        lease_session = str(lease.get("session_id") or "").strip()
+        actor_session = str(live.get("actor_session_id") or "").strip()
+        if lease_session and actor_session and lease_session != actor_session:
+            detections.append(
+                _detection(
+                    "foreign_lease",
+                    "lease is owned by another session; never adopt it implicitly",
+                )
+            )
+
+    worktree = live.get("worktree") or {}
+    if isinstance(worktree, Mapping) and worktree:
+        if worktree.get("present") is False:
+            detections.append(_detection("missing_worktree", "bound worktree is absent"))
+        if worktree.get("dirty") is True:
+            detections.append(
+                _detection("dirty_worktree", "bound worktree carries uncommitted changes")
+            )
+
+    namespace_role = str(live.get("namespace_role") or "").strip().lower()
+    if namespace_role and next_actor and next_actor != "none":
+        if namespace_role != next_actor:
+            detections.append(
+                _detection(
+                    "namespace_mismatch",
+                    f"live namespace role '{namespace_role}' cannot act as '{next_actor}'",
+                )
+            )
+
+    if live.get("runtime_stale") is True:
+        detections.append(
+            _detection("stale_runtime", "serving runtime is stale; reconnect required")
+        )
+
+    handoff_base = (fields.get("BASE_BRANCH") or "").strip()
+    live_base = str(live.get("base_branch") or "").strip()
+    if handoff_base and live_base and not _is_placeholder(handoff_base):
+        if handoff_base != live_base:
+            detections.append(
+                _detection(
+                    "changed_base",
+                    f"handoff base '{handoff_base}' but live base '{live_base}'",
+                )
+            )
+
+    if live.get("conflicting_canonical_comments") is True:
+        detections.append(
+            _detection(
+                "conflicting_canonical_comments",
+                "thread carries contradictory canonical comments",
+            )
+        )
+
+    kinds = [item["kind"] for item in detections]
+    reasons = [f"{item['kind']}: {item['detail']}" for item in detections]
+    block = bool(detections)
+    return {
+        "block": block,
+        "detections": detections,
+        "kinds": kinds,
+        "reasons": reasons,
+        "recovered_state": recovered_state,
+        "safe_next_action": (
+            "post a corrected canonical handoff for the recovered live state "
+            "before acting"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# role-limited continuation
+# ---------------------------------------------------------------------------
+
+
+def assess_role_continuation(
+    *,
+    handoff: str | Mapping[str, str],
+    actor_role: str,
+) -> dict[str, Any]:
+    """Only the role the current state authorizes may continue the chain."""
+    if isinstance(handoff, Mapping):
+        fields = dict(handoff)
+    else:
+        fields = parse_self_propagating_handoff(handoff or "") or {}
+
+    role = (actor_role or "").strip().lower()
+    state = (fields.get("WORKFLOW_STATE") or "").strip()
+    expected = NEXT_ACTOR_BY_STATE.get(state)
+    reasons: list[str] = []
+
+    if not fields:
+        reasons.append("no canonical handoff to continue from")
+    if role not in WORKFLOW_ROLES:
+        reasons.append(f"unknown actor role '{actor_role}'")
+    if expected is None and fields:
+        reasons.append(f"unknown workflow state '{state}'")
+    elif expected == "none":
+        reasons.append(
+            "workflow state 'complete' is terminal; no further role may continue"
+        )
+    elif expected and role != expected:
+        reasons.append(
+            f"state '{state}' authorizes '{expected}', not '{role}'"
+        )
+
+    block = bool(reasons)
+    return {
+        "allowed": not block,
+        "block": block,
+        "expected_actor": expected,
+        "actor_role": role,
+        "allowed_actions": () if block else ROLE_ALLOWED_ACTIONS.get(role, ()),
+        "forbidden_actions": ROLE_FORBIDDEN_ACTIONS.get(role, ()),
+        "reasons": reasons,
+        "safe_next_action": (
+            f"hand off to '{expected}'" if block and expected else
+            "stop; the workflow is complete" if expected == "none" else
+            "proceed"
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# durable posting (AC: a chat-only report is never sufficient)
+# ---------------------------------------------------------------------------
+
+
+def assess_durable_state_update(
+    *,
+    handoff_text: str,
+    posted_comment_id: Any = None,
+    canonical_state_posted: bool = False,
+) -> dict[str, Any]:
+    """A successful actor session must leave the handoff in Gitea, not chat."""
+    reasons: list[str] = []
+    assessment = assess_self_propagating_handoff(handoff_text)
+    if assessment["block"]:
+        reasons.extend(assessment["reasons"])
+    if not posted_comment_id:
+        reasons.append(
+            "canonical handoff was not posted to Gitea; a chat-only report is "
+            "not durable workflow state"
+        )
+    if not canonical_state_posted:
+        reasons.append(
+            "canonical issue/PR state and thread ledger were not updated"
+        )
+
+    block = bool(reasons)
+    return {
+        "durable": not block,
+        "block": block,
+        "posted_comment_id": posted_comment_id,
+        "reasons": reasons,
+        "safe_next_action": (
+            "post the canonical handoff and state update to Gitea before "
+            "ending the session"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# merge -> controller boundary and controller continuation
+# ---------------------------------------------------------------------------
+
+
+def assess_merge_completion_transition(
+    *,
+    merge_succeeded: bool,
+    controller_auto_accept: bool = False,
+) -> dict[str, Any]:
+    """A merged PR is not accepted work until the controller says so."""
+    if not merge_succeeded:
+        return {
+            "next_state": "approved-awaiting-merge",
+            "next_actor": "merger",
+            "next_prompt_required": True,
+            "reasons": ["merge did not succeed; the merger retains the work item"],
+        }
+    if controller_auto_accept:
+        return {
+            "next_state": "complete",
+            "next_actor": "none",
+            "next_prompt_required": False,
+            "reasons": ["configured workflow authorizes automatic acceptance on merge"],
+        }
+    return {
+        "next_state": "merged-awaiting-controller",
+        "next_actor": "controller",
+        "next_prompt_required": True,
+        "reasons": [
+            "merge succeeded; acceptance requires the authorized controller"
+        ],
+    }
+
+
+def assess_controller_decision(
+    *,
+    decision: str,
+    closure_proof: Mapping[str, Any] | None = None,
+    return_to: str | None = None,
+) -> dict[str, Any]:
+    """Controller acceptance or rejection produces the next or final state."""
+    normalized = (decision or "").strip().lower()
+    if normalized not in CONTROLLER_DECISIONS:
+        return {
+            "block": True,
+            "next_state": None,
+            "next_actor": None,
+            "next_prompt_required": True,
+            "reasons": [
+                f"unknown controller decision '{decision}'; expected one of "
+                f"{sorted(CONTROLLER_DECISIONS)}"
+            ],
+            "safe_next_action": "record a supported controller decision",
+        }
+
+    if normalized == "accept":
+        proof = dict(closure_proof or {})
+        missing = [
+            name
+            for name in CONTROLLER_CLOSURE_PROOF_FIELDS
+            if proof.get(name) is not True
+        ]
+        if missing:
+            return {
+                "block": True,
+                "next_state": "merged-awaiting-controller",
+                "next_actor": "controller",
+                "next_prompt_required": True,
+                "reasons": [
+                    "controller acceptance missing closure proof: " + ", ".join(missing)
+                ],
+                "safe_next_action": (
+                    "satisfy and record every closure proof field before closing"
+                ),
+            }
+        return {
+            "block": False,
+            "next_state": "complete",
+            "next_actor": "none",
+            "next_prompt_required": False,
+            "reasons": ["controller accepted; workflow chain terminates"],
+            "safe_next_action": "post the final canonical state and stop",
+        }
+
+    if normalized == "return_to_actor":
+        target = (return_to or "").strip().lower()
+        state_by_actor = {
+            "author": "needs-author",
+            "reviewer": "needs-review",
+            "merger": "approved-awaiting-merge",
+        }
+        if target not in state_by_actor:
+            return {
+                "block": True,
+                "next_state": None,
+                "next_actor": None,
+                "next_prompt_required": True,
+                "reasons": [
+                    f"return_to_actor requires a target in {sorted(state_by_actor)}"
+                ],
+                "safe_next_action": "name the actor the work returns to",
+            }
+        return {
+            "block": False,
+            "next_state": state_by_actor[target],
+            "next_actor": target,
+            "next_prompt_required": True,
+            "reasons": [f"controller returned the work item to '{target}'"],
+            "safe_next_action": f"post a complete handoff for '{target}'",
+        }
+
+    # request_tests / request_proof / request_corrections / reopen
+    return {
+        "block": False,
+        "next_state": "needs-author",
+        "next_actor": "author",
+        "next_prompt_required": True,
+        "reasons": [f"controller decision '{normalized}' returns the work to the author"],
+        "safe_next_action": "post a complete author handoff describing what is required",
+    }
+
+
+# ---------------------------------------------------------------------------
+# workflow-failure escalation
+# ---------------------------------------------------------------------------
+
+
+def assess_workflow_failure_escalation(
+    *,
+    failures: Sequence[Mapping[str, Any]] | None,
+    active_issue_number: int | str | None,
+    existing_failure_issues: Iterable[Mapping[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Tooling defects hit while working an issue become separate durable work."""
+    entries = list(failures or [])
+    known = {
+        str((item.get("signature") or "")).strip().lower(): item.get("number")
+        for item in (existing_failure_issues or [])
+        if str((item.get("signature") or "")).strip()
+    }
+    active = str(active_issue_number or "").strip().lstrip("#")
+
+    reasons: list[str] = []
+    reused: list[dict[str, Any]] = []
+    seen_signatures: dict[str, str] = {}
+
+    for index, failure in enumerate(entries):
+        label = str(failure.get("signature") or f"failure[{index}]")
+        missing = [
+            name
+            for name in WORKFLOW_FAILURE_FIELDS
+            if _is_placeholder(failure.get(name))
+        ]
+        if missing:
+            reasons.append(
+                f"{label}: workflow failure missing " + ", ".join(missing)
+            )
+
+        linked = str(failure.get("linked_issue") or "").strip().lstrip("#")
+        if linked and active and linked == active:
+            reasons.append(
+                f"{label}: workflow defects must not be folded into the active "
+                f"work item #{active}; file a separate durable issue"
+            )
+
+        signature = str(failure.get("signature") or "").strip().lower()
+        if not signature:
+            continue
+        if signature in known:
+            expected = str(known[signature] or "").strip().lstrip("#")
+            if linked and expected and linked != expected:
+                reasons.append(
+                    f"{label}: duplicate workflow-failure issue #{linked}; "
+                    f"reuse the existing issue #{expected}"
+                )
+            else:
+                reused.append({"signature": signature, "issue": expected})
+        if signature in seen_signatures:
+            reasons.append(
+                f"{label}: duplicate workflow-failure signature reported twice "
+                "in one session"
+            )
+        else:
+            seen_signatures[signature] = linked
+
+    block = bool(reasons)
+    return {
+        "escalated": not block,
+        "block": block,
+        "failure_count": len(entries),
+        "reused_issues": reused,
+        "reasons": reasons,
+        "safe_next_action": (
+            "file or reference one durable issue per distinct workflow failure"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# final-report integration
+# ---------------------------------------------------------------------------
+
+
+def assess_final_report_self_propagating_handoff(report_text: str) -> dict[str, Any]:
+    """#626 gate for final reports.
+
+    Applicability mirrors the #495 canonical-state gate: once a report adopts
+    the protocol — by carrying the marker, the ``Canonical Handoff`` heading,
+    or a ``WORKFLOW_STATE`` line — the full schema is enforced.  Reports that
+    predate the protocol are untouched here; the workflow schemas require the
+    block going forward.
+    """
+    text = report_text or ""
+    applicable = (
+        MARKER in text
+        or bool(_HEADING_RE.search(text))
+        or bool(re.search(r"^WORKFLOW_STATE\s*:", text, re.MULTILINE))
+    )
+    if not applicable:
+        return {
+            "applicable": False,
+            "valid": True,
+            "block": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+
+    assessment = assess_self_propagating_handoff(text)
+    recoverability = assess_thread_recoverability(text)
+    reasons = list(assessment["reasons"])
+    if not assessment["block"]:
+        reasons.extend(recoverability.get("reasons") or [])
+    block = bool(assessment["block"] or recoverability.get("block"))
+    return {
+        "applicable": True,
+        "valid": not block,
+        "block": block,
+        "workflow_state": assessment.get("workflow_state"),
+        "next_actor": assessment.get("next_actor"),
+        "terminal": assessment.get("terminal"),
+        "reasons": reasons,
+        "safe_next_action": (
+            assessment["safe_next_action"]
+            if assessment["block"]
+            else recoverability.get("safe_next_action", "proceed")
+        ),
+    }

skills/llm-project-workflow/SKILL.md

@@ -224,6 +224,17 @@ format canonical field set per issue #182; mode-specific schemas in
 for the loaded workflow mode — not the legacy compact block alone.
 `review_proofs.assess_controller_handoff()` validates presence.
 
+## Canonical self-propagating handoff
+
+Every workflow mode also carries the cross-role handoff block defined in
+[`schemas/self-propagating-handoff.md`](schemas/self-propagating-handoff.md)
+(#626). Each actor consumes exactly one canonical handoff, performs exactly one
+authorized role, posts the result to the Gitea issue or PR thread, and emits the
+next complete handoff — until the controller records final closure. The block
+must be posted to Gitea, not returned in chat alone, and the next prompt is not
+an optional prose section. `self_propagating_handoff.py` implements the schema;
+`final_report_validator.py` enforces it as `shared.self_propagating_handoff`.
+
 ## Prompt templates
 
 Ready-to-copy task prompts live in [`templates/`](templates/):

skills/llm-project-workflow/schemas/create-issue-final-report.md

@@ -44,3 +44,7 @@ mutations occurred).
 ```
 
 Identity format: `username / profile` (not personal email unless required — #305).
+
+The report must also carry the canonical self-propagating handoff block
+(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to
+the Gitea issue thread.

skills/llm-project-workflow/schemas/pr-queue-cleanup-final-report.md

@@ -29,3 +29,7 @@ use `none` where nothing occurred. Validated by
 * Read-only diagnostics:
 * Blockers:
 * Safe next action: (fresh run for the next PR)
+
+The report must also carry the canonical self-propagating handoff block
+(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to
+the Gitea PR thread.

skills/llm-project-workflow/schemas/reconcile-landed-final-report.md

@@ -39,6 +39,7 @@ occurred).
 - Git ref mutations:
 - MCP/Gitea mutations:
 - Reconciliation mutations:
+- Terminal label cleanup:
 - External-state mutations:
 - Read-only diagnostics:
 - Blockers:
@@ -50,4 +51,15 @@ occurred).
 
 Identity format: `username / profile` (not personal email unless required — #305).
 
-`git fetch` belongs under `Git ref mutations`, not read-only diagnostics (#297).
+`git fetch` belongs under `Git ref mutations`, not read-only diagnostics (#297).
+
+`Terminal label cleanup` (#780) reports the `pr_open_label_cleanup` record the
+reconciliation tool returned — `clean` / `failed` / `not applicable (no linked
+issue)`, with the labels removed and preserved. Reconciliation is a terminal
+transition, so a non-`clean` record blocks any "reconciled" claim; recover with
+`gitea_cleanup_terminal_pr_labels` (`terminal_reason='retry_recovery'`) and
+confirm with `gitea_assess_terminal_label_hygiene`.
+
+The report must also carry the canonical self-propagating handoff block
+(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to
+the Gitea issue or PR thread.

skills/llm-project-workflow/schemas/review-merge-final-report.md

@@ -99,6 +99,21 @@ Narrative final report and controller handoff must agree on eligibility class,
 candidate/reviewed head SHA, mutation state, worktree usage, review decision,
 terminal review mutation, merge result, and linked issue status.
 
+### Terminal label state (#780)
+
+A run that takes a PR to a terminal state — merged, closed without merge,
+superseded, or reconciled as already landed — must report what happened to the
+linked issue's `status:pr-open` label, quoting the `pr_open_label_cleanup`
+record the terminal tool returned:
+
+- Terminal label cleanup: `clean` / `failed` / `not applicable (no linked issue)`
+- Labels removed and preserved per issue, with the read-after-write read-back
+
+Never claim the transition is complete while that record is not `clean`. A
+failed cleanup does not undo the merge; the safe next action is
+`gitea_cleanup_terminal_pr_labels` with `terminal_reason='retry_recovery'`,
+confirmed by `gitea_assess_terminal_label_hygiene`.
+
 ### Proof-backed claims (#395)
 
 Proof-sensitive claims must cite explicit command/tool evidence in the report
@@ -116,4 +131,9 @@ or structured MCP metadata — not narrative alone:
 
 When a claim relies on prior-session blocker state or MCP metadata only, label
 the proof source explicitly (`command`, `MCP metadata`, `prior blocker`,
-`not checked`). Do not use `live proof` without that classification.
+`not checked`). Do not use `live proof` without that classification.
+
+The report must also carry the canonical self-propagating handoff block
+(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to
+the Gitea PR thread. A reviewer hands off to `merger`; a merger transitions to
+`merged-awaiting-controller` rather than declaring the work accepted.

skills/llm-project-workflow/schemas/self-propagating-handoff.md

@@ -0,0 +1,114 @@
+# Canonical self-propagating handoff schema (#626)
+
+**Applies to:** every workflow actor — author, reviewer, merger, controller,
+operator, reconciler.
+
+`#494`–`#507` defined the ledger, the canonical state comments, and the
+Canonical Thread Handoff shape. This schema owns the *chain*: each actor
+consumes exactly one canonical handoff, performs exactly one authorized role,
+records the result durably in Gitea, and emits the next complete handoff —
+until the controller records final closure.
+
+Implemented and enforced by `self_propagating_handoff.py`; wired into
+`final_report_validator.py` as rule `shared.self_propagating_handoff`.
+
+## The block
+
+Post this block into the Gitea issue or PR thread, and include it verbatim in
+the final report. It is not an optional prose section.
+
+```md
+<!-- sph:v1 -->
+## Canonical Handoff
+
+```text
+REPOSITORY: <org>/<repo>
+ISSUE: <number>
+PR: <number or none>
+WORKFLOW_STATE: <one of the workflow states below>
+HEAD_SHA: <current head, or none before a branch exists>
+BASE_BRANCH: <base branch>
+BASE_OR_MERGE_SHA: <base SHA, or merge commit SHA after merge>
+ACTING_ROLE: <author|reviewer|merger|controller|operator|reconciler>
+ACTING_IDENTITY: <username (profile)>
+COMPLETED_ACTIONS: <what this actor actually did>
+VALIDATION_EVIDENCE: <commands run and their results>
+MUTATION_LEDGER: <every durable mutation performed>
+BLOCKERS: <active blockers, or none>
+NEXT_ACTOR: <role authorized by WORKFLOW_STATE, or none when complete>
+NEXT_ACTION: <exact next action, or none when complete>
+PROHIBITED_ACTIONS: <what the next actor must not do>
+NEXT_PROMPT: <complete ready-to-run prompt, or none when complete>
+WORKFLOW_FAILURE_ISSUES: <durable issue refs for tooling defects, or none>
+LAST_UPDATED: <UTC timestamp>
+```
+```
+
+## Workflow states and the single authorized actor
+
+| `WORKFLOW_STATE`            | `NEXT_ACTOR` |
+| --------------------------- | ------------ |
+| `needs-author`              | `author`     |
+| `needs-review`              | `reviewer`   |
+| `approved-awaiting-merge`   | `merger`     |
+| `merged-awaiting-controller`| `controller` |
+| `blocked`                   | `operator`   |
+| `complete`                  | `none`       |
+
+A merged PR is **not** accepted work: merge transitions to
+`merged-awaiting-controller` unless the configured workflow explicitly
+authorizes automatic acceptance.
+
+## Fail-closed rules
+
+* Every field is required. Only `PR`, `HEAD_SHA`, `BASE_OR_MERGE_SHA`,
+  `BLOCKERS`, `WORKFLOW_FAILURE_ISSUES`, `NEXT_ACTION`, and `NEXT_PROMPT` may
+  carry `none`, and `PR`/`HEAD_SHA` only in `needs-author` or `blocked`.
+* `NEXT_ACTOR` must equal the actor the declared state authorizes.
+* `blocked` requires a concrete `BLOCKERS` entry.
+* A non-terminal handoff requires a concrete `NEXT_ACTION` and a
+  `NEXT_PROMPT` long enough to be ready to run.
+* `complete` must carry no `NEXT_ACTION` and no `NEXT_PROMPT`: a finished
+  workflow terminates instead of manufacturing more work.
+* `NEXT_PROMPT` must name the repository and the issue, and must not depend on
+  outside chat history. The issue or PR thread, workflow documentation, and
+  live repository state must be sufficient to recover the task.
+* The handoff must be posted to Gitea. A chat-only report is not durable
+  workflow state.
+
+## Live-state recovery before acting
+
+The receiving actor re-derives truth from live state instead of trusting the
+inherited handoff. `assess_handoff_live_state` detects and fails closed on:
+`changed_pr_head`, `stale_approval`, `issue_closed`, `issue_reopened`,
+`pr_merged`, `pr_closed_unmerged`, `stale_lease`, `foreign_lease`,
+`missing_worktree`, `dirty_worktree`, `namespace_mismatch`, `stale_runtime`,
+`changed_base`, and `conflicting_canonical_comments`.
+
+A changed head invalidates any inherited review or merge handoff; the chain
+recovers to `needs-review`.
+
+## Controller closure
+
+`accept` is only honored with all four closure proofs recorded:
+`acceptance_criteria_satisfied`, `cleanup_complete`,
+`canonical_final_state_posted`, `issue_closed_through_workflow`. Otherwise the
+work item stays at `merged-awaiting-controller`.
+
+`request_tests`, `request_proof`, `request_corrections`, and `reopen` return
+the work to the author; `return_to_actor` returns it to a named earlier actor.
+
+## Workflow-failure escalation
+
+Tooling or workflow defects found while working an item never get folded into
+the active feature issue. Each distinct failure carries `classification`,
+`linked_issue`, `temporary_impact`, `next_valid_actor`, and `recovery_prompt`.
+A failure whose signature already has a durable issue must reuse that issue
+instead of filing a duplicate.
+
+## Applicability
+
+Enforcement is applicability-gated exactly like the #495 canonical-state gate:
+once a report carries the `sph:v1` marker, the `Canonical Handoff` heading, or
+a `WORKFLOW_STATE:` line, the full schema is enforced and incomplete handoffs
+are rejected. Reports written before the protocol existed are unaffected.

skills/llm-project-workflow/schemas/work-issue-final-report.md

@@ -70,4 +70,9 @@ selected issue, and mutation ledger categories (#319, #320).
 `Read-only diagnostics` (#297).
 
 Forbidden claims without proof (#330): `next eligible issue`, `issue claimed`,
-`validation passed`, `PR created`, `worktree clean`, `all gates passed`, etc.
+`validation passed`, `PR created`, `worktree clean`, `all gates passed`, etc.
+
+The report must also carry the canonical self-propagating handoff block
+(`schemas/self-propagating-handoff.md`, #626) and that block must be posted to
+the Gitea issue or PR thread. The next prompt is not an optional prose
+section.

task_capability_map.py

@@ -36,6 +36,12 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.issue.comment",
         "role": "author",
     },
+    # #780: retire status:pr-open after a terminal PR transition. Same label
+    # authority as set_issue_labels — it is a strictly narrower operation.
+    "cleanup_terminal_pr_labels": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
     "create_label": {
         "permission": "gitea.issue.comment",
         "role": "author",
@@ -507,6 +513,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
     "gitea_create_issue_comment": "comment_issue",
     "gitea_mark_issue": "mark_issue",
     "gitea_set_issue_labels": "set_issue_labels",
+    "gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels",
     "gitea_create_label": "create_label",
     "gitea_commit_files": "commit_files",
 }

terminal_pr_label_cleanup.py

@@ -0,0 +1,308 @@
+"""Authoritative terminal-transition cleanup for ``status:pr-open`` (#780).
+
+``status:pr-open`` is applied by ``gitea_create_pr`` while a linked pull
+request is open. Nothing removed it again: the workflow's terminal paths
+(merge, close-without-merge, supersession, already-landed reconciliation,
+controller closure) each ended without touching the label, so a repository
+audit found 40 closed issues still carrying it.
+
+This module is the single source of truth for that cleanup. Every sanctioned
+terminal path plans its label mutation here rather than implementing its own
+rule, so the paths cannot drift apart:
+
+- :func:`plan_pr_open_cleanup` decides the exact resulting label set. It only
+  ever removes ``status:pr-open``; every other label is preserved verbatim,
+  including the case where the result is an empty label set.
+- :func:`verify_pr_open_cleanup` is the read-after-write check. It proves the
+  label is gone *and* that no unrelated label was dropped or added.
+- :func:`detect_residual_pr_open` is the terminal validation: given issues, it
+  reports any that still carry the label, so a controller closure or audit
+  fails loudly instead of leaving the leak behind.
+
+The rule is idempotent by construction: an issue without the label plans no
+mutation, so retries and recovery re-runs are harmless.
+
+This module performs no I/O — callers own the Gitea API calls.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Iterable, Mapping, Sequence
+
+import issue_workflow_labels
+
+#: The single label this module is responsible for retiring.
+PR_OPEN_LABEL = "status:pr-open"
+
+# Canonical terminal reasons — the sanctioned ways an issue can end up
+# associated with a pull request that is no longer open.
+MERGED = "merged"
+CLOSED_WITHOUT_MERGE = "closed_without_merge"
+SUPERSEDED = "superseded"
+ALREADY_LANDED = "already_landed"
+CONTROLLER_CLOSURE = "controller_closure"
+ABANDONED = "abandoned"
+RETRY_RECOVERY = "retry_recovery"
+
+TERMINAL_REASONS: tuple[str, ...] = (
+    MERGED,
+    CLOSED_WITHOUT_MERGE,
+    SUPERSEDED,
+    ALREADY_LANDED,
+    CONTROLLER_CLOSURE,
+    ABANDONED,
+    RETRY_RECOVERY,
+)
+
+_REASON_ALIASES: dict[str, str] = {
+    "merge": MERGED,
+    "merged": MERGED,
+    "pr_merged": MERGED,
+    "closed": CLOSED_WITHOUT_MERGE,
+    "close": CLOSED_WITHOUT_MERGE,
+    "closed_without_merge": CLOSED_WITHOUT_MERGE,
+    "pr_closed": CLOSED_WITHOUT_MERGE,
+    "supersede": SUPERSEDED,
+    "superseded": SUPERSEDED,
+    "supersession": SUPERSEDED,
+    "already_landed": ALREADY_LANDED,
+    "reconcile_already_landed": ALREADY_LANDED,
+    "controller_closure": CONTROLLER_CLOSURE,
+    "close_issue": CONTROLLER_CLOSURE,
+    "abandon": ABANDONED,
+    "abandoned": ABANDONED,
+    "retry": RETRY_RECOVERY,
+    "recovery": RETRY_RECOVERY,
+    "retry_recovery": RETRY_RECOVERY,
+}
+
+#: Human-readable phrasing used in audit comments and diagnostics.
+REASON_DESCRIPTIONS: dict[str, str] = {
+    MERGED: "the linked PR was merged",
+    CLOSED_WITHOUT_MERGE: "the linked PR was closed without merging",
+    SUPERSEDED: "the linked PR was superseded by another merged PR",
+    ALREADY_LANDED: "the linked PR's change was already on the target branch",
+    CONTROLLER_CLOSURE: "the issue reached controller closure",
+    ABANDONED: "the linked PR was abandoned",
+    RETRY_RECOVERY: "a partial terminal transition is being recovered",
+}
+
+
+def canonical_terminal_reason(reason: str | None) -> str:
+    """Normalize a terminal reason, failing closed on anything unrecognized."""
+    name = (reason or "").strip()
+    if name in TERMINAL_REASONS:
+        return name
+    normalized = name.lower().replace("-", "_").replace(" ", "_")
+    try:
+        return _REASON_ALIASES[normalized]
+    except KeyError as exc:
+        raise ValueError(
+            f"unknown terminal PR reason '{reason}' (expected one of: "
+            + ", ".join(TERMINAL_REASONS)
+            + ")"
+        ) from exc
+
+
+def plan_pr_open_cleanup(
+    current_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
+    *,
+    terminal_reason: str,
+) -> dict[str, Any]:
+    """Plan the label set an issue must carry after a terminal PR transition.
+
+    The plan removes ``status:pr-open`` and nothing else. When the label is
+    absent the plan is an explicit no-op (``cleanup_required`` False), which is
+    what makes repeated cleanup calls harmless. When it was the only label the
+    resulting set is legitimately empty.
+    """
+    reason = canonical_terminal_reason(terminal_reason)
+    before = issue_workflow_labels.label_names(current_labels)
+    after = [name for name in before if name != PR_OPEN_LABEL]
+    present = len(after) != len(before)
+    return {
+        "terminal_reason": reason,
+        "terminal_reason_description": REASON_DESCRIPTIONS[reason],
+        "label": PR_OPEN_LABEL,
+        "label_present": present,
+        "cleanup_required": present,
+        "idempotent_noop": not present,
+        "labels_before": before,
+        "labels_after": after,
+        "removed": [PR_OPEN_LABEL] if present else [],
+        "preserved": list(after),
+        "empty_label_set": not after,
+    }
+
+
+def verify_pr_open_cleanup(
+    observed_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
+    *,
+    plan: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Read-after-write check for a planned cleanup.
+
+    Verifies the label is gone and that the observed set matches the plan
+    exactly, so an unrelated label silently dropped (or re-added) by the API is
+    reported rather than accepted.
+    """
+    observed = issue_workflow_labels.label_names(observed_labels)
+    expected = list(plan.get("labels_after") or [])
+    observed_set = set(observed)
+    expected_set = set(expected)
+    residual = PR_OPEN_LABEL in observed_set
+    unexpected_removals = sorted(expected_set - observed_set)
+    unexpected_additions = sorted(observed_set - expected_set - {PR_OPEN_LABEL})
+
+    reasons: list[str] = []
+    if residual:
+        reasons.append(
+            f"'{PR_OPEN_LABEL}' is still present after terminal cleanup"
+        )
+    if unexpected_removals:
+        reasons.append(
+            "unrelated labels were dropped by the cleanup: "
+            + ", ".join(unexpected_removals)
+        )
+    if unexpected_additions:
+        reasons.append(
+            "unexpected labels appeared during the cleanup: "
+            + ", ".join(unexpected_additions)
+        )
+
+    verified = not reasons
+    return {
+        "verified": verified,
+        "residual": residual,
+        "observed_labels": observed,
+        "expected_labels": expected,
+        "unexpected_removals": unexpected_removals,
+        "unexpected_additions": unexpected_additions,
+        "empty_label_set": not observed,
+        "reasons": reasons,
+        "safe_next_action": (
+            ""
+            if verified
+            else (
+                "Re-run the terminal cleanup for this issue with "
+                f"terminal_reason='{RETRY_RECOVERY}' and confirm the read-back "
+                f"no longer reports '{PR_OPEN_LABEL}'."
+            )
+        ),
+    }
+
+
+def summarize_cleanup_results(
+    results: Sequence[Mapping[str, Any]],
+    *,
+    terminal_reason: str,
+) -> dict[str, Any]:
+    """Aggregate per-issue cleanup outcomes into one reportable record."""
+    reason = canonical_terminal_reason(terminal_reason)
+    entries = [dict(entry) for entry in results]
+    removed = [e.get("issue_number") for e in entries if e.get("status") == "removed"]
+    absent = [
+        e.get("issue_number") for e in entries if e.get("status") == "not present"
+    ]
+    failed = [
+        e.get("issue_number")
+        for e in entries
+        if e.get("status") not in ("removed", "not present") or not e.get("verified")
+    ]
+    reasons: list[str] = []
+    for entry in entries:
+        for text in entry.get("reasons") or []:
+            reasons.append(f"issue #{entry.get('issue_number')}: {text}")
+    clean = not failed
+    return {
+        "label": PR_OPEN_LABEL,
+        "terminal_reason": reason,
+        "clean": clean,
+        "checked": [e.get("issue_number") for e in entries],
+        "removed": removed,
+        "already_absent": absent,
+        "failed": failed,
+        "results": entries,
+        "reasons": reasons,
+        "safe_next_action": (
+            ""
+            if clean
+            else (
+                "Terminal label cleanup did not complete for "
+                + ", ".join(f"#{num}" for num in failed)
+                + ". Re-run gitea_cleanup_terminal_pr_labels with "
+                f"terminal_reason='{RETRY_RECOVERY}' for those issues."
+            )
+        ),
+    }
+
+
+def detect_residual_pr_open(
+    issues: Iterable[Mapping[str, Any]],
+    *,
+    open_pr_issue_numbers: Iterable[int] = (),
+) -> dict[str, Any]:
+    """Terminal validation: report issues still carrying ``status:pr-open``.
+
+    An issue with a genuinely open pull request is allowed to keep the label,
+    so *open_pr_issue_numbers* is excluded from the residual set rather than
+    being reported as a leak.
+    """
+    legitimate: set[int] = set()
+    for num in open_pr_issue_numbers or ():
+        try:
+            legitimate.add(int(num))
+        except (TypeError, ValueError):
+            continue
+
+    checked = 0
+    residual: list[dict[str, Any]] = []
+    exempt: list[int] = []
+
+    for issue in issues or []:
+        checked += 1
+        names = issue_workflow_labels.label_names(issue)
+        if PR_OPEN_LABEL not in names:
+            continue
+        try:
+            number = int(issue.get("number"))
+        except (TypeError, ValueError):
+            number = None
+        if number is not None and number in legitimate:
+            exempt.append(number)
+            continue
+        residual.append(
+            {
+                "number": number,
+                "state": issue.get("state"),
+                "labels": names,
+            }
+        )
+
+    clean = not residual
+    reasons = [
+        (
+            f"issue #{entry['number']} ({entry.get('state') or 'unknown state'}) "
+            f"still carries '{PR_OPEN_LABEL}' with no open PR"
+        )
+        for entry in residual
+    ]
+    return {
+        "label": PR_OPEN_LABEL,
+        "clean": clean,
+        "checked_count": checked,
+        "residual_count": len(residual),
+        "residual_issues": residual,
+        "exempt_open_pr_issues": sorted(exempt),
+        "reasons": reasons,
+        "safe_next_action": (
+            ""
+            if clean
+            else (
+                "Run gitea_cleanup_terminal_pr_labels with "
+                f"terminal_reason='{RETRY_RECOVERY}' for issues "
+                + ", ".join(f"#{entry['number']}" for entry in residual)
+                + " before declaring the terminal transition complete."
+            )
+        ),
+    }

tests/test_allocator_foreign_lease_exclusion.py

@@ -0,0 +1,380 @@
+"""Allocator ownership exclusion tests (#765).
+
+One session's active lease must never blockade the author queue for a
+different controller. Covers: foreign lease skipped, next unclaimed candidate
+selected, own task resumable, task-local blocker quarantined, all-claimed ->
+wait, same profile + different controller_instance_id -> different ownership,
+and claimed candidates reported in skipped results.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+
+from allocator_service import (
+    OUTCOME_OWNERSHIP_DEFECT,
+    OUTCOME_PREVIEW,
+    OUTCOME_WAIT,
+    OWNERSHIP_FOREIGN,
+    OWNERSHIP_OWN,
+    OWNERSHIP_UNKNOWN,
+    SKIP_CLAIMED_BY_OTHER_SESSION,
+    WorkCandidate,
+    allocate_next_work,
+    classify_claim_ownership,
+    resolve_controller_instance_id,
+)
+from control_plane_db import ControlPlaneDB
+
+REMOTE = "prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+MINE = "ctl-mine-0001"
+THEIRS = "ctl-theirs-0002"
+
+
+def _issue(number: int, **kwargs) -> WorkCandidate:
+    base = dict(
+        kind="issue",
+        number=number,
+        state="open",
+        labels=("status:ready", "type:bug"),
+        title=f"issue {number}",
+        priority=20,
+    )
+    base.update(kwargs)
+    return WorkCandidate(**base)
+
+
+def _claim(number: int, *, session_id: str, instance: str | None, kind: str = "issue"):
+    return {
+        "lease_id": f"lease-{number}",
+        "session_id": session_id,
+        "controller_instance_id": instance,
+        "role": "author",
+        "profile": "prgs-author",
+        "expires_at": "2026-07-20T07:06:09Z",
+        "work_kind": kind,
+        "work_number": number,
+    }
+
+
+class AllocatorOwnershipTestCase(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def _allocate(
+        self,
+        candidates,
+        *,
+        claims,
+        session_id="sess-mine",
+        instance=MINE,
+        apply=False,
+        role="author",
+    ):
+        return allocate_next_work(
+            self.db,
+            session_id=session_id,
+            role=role,
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            candidates=candidates,
+            apply=apply,
+            profile_name="prgs-author",
+            controller_instance_id=instance,
+            claims=claims,
+        )
+
+
+class TestOwnershipClassification(AllocatorOwnershipTestCase):
+    def test_no_claim_returns_none(self):
+        self.assertIsNone(
+            classify_claim_ownership(
+                None, session_id="s", controller_instance_id=MINE
+            )
+        )
+
+    def test_same_controller_instance_is_own(self):
+        claim = _claim(1, session_id="other-session", instance=MINE)
+        self.assertEqual(
+            classify_claim_ownership(
+                claim, session_id="sess-mine", controller_instance_id=MINE
+            ),
+            OWNERSHIP_OWN,
+        )
+
+    def test_same_profile_different_instance_is_foreign(self):
+        """Shared profile must not imply shared ownership."""
+        claim = _claim(1, session_id="other-session", instance=THEIRS)
+        self.assertEqual(
+            classify_claim_ownership(
+                claim, session_id="sess-mine", controller_instance_id=MINE
+            ),
+            OWNERSHIP_FOREIGN,
+        )
+
+    def test_exact_session_match_is_own(self):
+        claim = _claim(1, session_id="sess-mine", instance=None)
+        self.assertEqual(
+            classify_claim_ownership(
+                claim, session_id="sess-mine", controller_instance_id=None
+            ),
+            OWNERSHIP_OWN,
+        )
+
+    def test_legacy_claim_with_neither_side_identified_is_foreign(self):
+        """No identities anywhere: a different session id is simply not ours."""
+        claim = _claim(1, session_id="someone-else", instance=None)
+        self.assertEqual(
+            classify_claim_ownership(
+                claim, session_id="sess-mine", controller_instance_id=None
+            ),
+            OWNERSHIP_FOREIGN,
+        )
+
+    def test_claim_identified_but_local_undeclared_is_unknown(self):
+        """Only one side identified: not comparable, so never adopt."""
+        claim = _claim(1, session_id="someone-else", instance=THEIRS)
+        self.assertEqual(
+            classify_claim_ownership(
+                claim, session_id="sess-mine", controller_instance_id=None
+            ),
+            OWNERSHIP_UNKNOWN,
+        )
+
+    def test_local_identified_but_claim_undeclared_is_unknown(self):
+        """A legacy lease may be our own under an old session id; do not guess."""
+        claim = _claim(1, session_id="someone-else", instance=None)
+        self.assertEqual(
+            classify_claim_ownership(
+                claim, session_id="sess-mine", controller_instance_id=MINE
+            ),
+            OWNERSHIP_UNKNOWN,
+        )
+
+    def test_resolve_controller_instance_id_reads_env(self):
+        self.assertEqual(
+            resolve_controller_instance_id({"GITEA_CONTROLLER_INSTANCE_ID": MINE}),
+            MINE,
+        )
+        self.assertIsNone(resolve_controller_instance_id({}))
+        self.assertIsNone(
+            resolve_controller_instance_id({"GITEA_CONTROLLER_INSTANCE_ID": "  "})
+        )
+
+
+class TestForeignLeaseDoesNotBlockade(AllocatorOwnershipTestCase):
+    def test_foreign_claim_skipped_and_next_issue_selected(self):
+        """Skip the claimed issue, select the next unclaimed one."""
+        candidates = [_issue(607), _issue(615), _issue(617)]
+        claims = {
+            ("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS)
+        }
+        result = self._allocate(candidates, claims=claims)
+
+        self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(result["selected"]["number"], 615)
+        skipped_607 = [s for s in result["skipped"] if s["number"] == 607]
+        self.assertEqual(len(skipped_607), 1)
+        self.assertEqual(
+            skipped_607[0]["reason_code"], SKIP_CLAIMED_BY_OTHER_SESSION
+        )
+        self.assertIn(SKIP_CLAIMED_BY_OTHER_SESSION, skipped_607[0]["reason"])
+
+    def test_claimed_candidate_appears_in_skipped_inventory(self):
+        """Skipped reporting must reflect claimed candidates."""
+        candidates = [_issue(607), _issue(615)]
+        claims = {
+            ("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS)
+        }
+        result = self._allocate(candidates, claims=claims)
+        self.assertEqual(len(result["skipped"]), 1)
+        self.assertEqual(len(result["claims_excluded"]), 1)
+        excluded = result["claims_excluded"][0]
+        self.assertEqual(excluded["number"], 607)
+        self.assertEqual(excluded["ownership"], OWNERSHIP_FOREIGN)
+        self.assertEqual(excluded["owner_controller_instance_id"], THEIRS)
+
+    def test_task_local_blocker_does_not_freeze_unrelated_work(self):
+        """A quarantined task must not stop the rest of the queue."""
+        candidates = [_issue(607), _issue(615), _issue(617)]
+        claims = {
+            ("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS)
+        }
+        first = self._allocate(candidates, claims=claims)
+        self.assertEqual(first["selected"]["number"], 615)
+
+        # 615 then gets claimed by yet another controller; queue still advances.
+        claims[("issue", 615)] = _claim(
+            615, session_id="sess-third", instance="ctl-third-0003"
+        )
+        second = self._allocate(candidates, claims=claims)
+        self.assertEqual(second["selected"]["number"], 617)
+
+    def test_multiple_controllers_get_different_issues(self):
+        """Concurrent author sessions work on different issues."""
+        candidates = [_issue(607), _issue(615)]
+        claims = {
+            ("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS)
+        }
+        mine = self._allocate(candidates, claims=claims, instance=MINE)
+        theirs = self._allocate(
+            candidates, claims=claims, session_id="sess-theirs", instance=THEIRS
+        )
+        self.assertEqual(mine["selected"]["number"], 615)
+        # The other controller may still be handed its own in-progress task.
+        self.assertEqual(theirs["selected"]["number"], 607)
+
+    def test_unclaimed_queue_is_unaffected(self):
+        candidates = [_issue(607), _issue(615)]
+        result = self._allocate(candidates, claims={})
+        self.assertEqual(result["selected"]["number"], 607)
+        self.assertEqual(result["skipped"], [])
+        self.assertEqual(result["claims_excluded"], [])
+
+
+class TestOwnTaskResume(AllocatorOwnershipTestCase):
+    def test_controller_may_resume_its_own_active_task(self):
+        """Own claim stays selectable across a new session id."""
+        candidates = [_issue(607), _issue(615)]
+        claims = {
+            ("issue", 607): _claim(607, session_id="sess-mine-old", instance=MINE)
+        }
+        result = self._allocate(
+            candidates, claims=claims, session_id="sess-mine-new", instance=MINE
+        )
+        self.assertEqual(result["selected"]["number"], 607)
+        self.assertEqual(result["claims_excluded"], [])
+
+    def test_own_claim_by_exact_session_is_selectable(self):
+        candidates = [_issue(607)]
+        claims = {("issue", 607): _claim(607, session_id="sess-mine", instance=None)}
+        result = self._allocate(
+            candidates, claims=claims, session_id="sess-mine", instance=None
+        )
+        self.assertEqual(result["selected"]["number"], 607)
+
+
+class TestAllCandidatesClaimed(AllocatorOwnershipTestCase):
+    def test_all_claimed_returns_wait_not_a_claimed_selection(self):
+        """Never hand back a claimed issue; report waiting instead."""
+        candidates = [_issue(607), _issue(615)]
+        claims = {
+            ("issue", 607): _claim(607, session_id="sess-a", instance=THEIRS),
+            ("issue", 615): _claim(615, session_id="sess-b", instance="ctl-c-0003"),
+        }
+        result = self._allocate(candidates, claims=claims)
+        self.assertIsNone(result["selected"])
+        self.assertEqual(result["outcome"], OUTCOME_WAIT)
+        self.assertEqual(len(result["claims_excluded"]), 2)
+
+    def test_unidentifiable_owner_reports_ownership_defect(self):
+        """Refuse to adopt when ownership cannot be established."""
+        candidates = [_issue(607)]
+        claims = {("issue", 607): _claim(607, session_id="sess-legacy", instance=None)}
+        result = self._allocate(candidates, claims=claims)
+        self.assertIsNone(result["selected"])
+        self.assertEqual(result["outcome"], OUTCOME_OWNERSHIP_DEFECT)
+        self.assertEqual(len(result["ownership_defects"]), 1)
+        self.assertEqual(
+            result["ownership_defects"][0]["ownership"], OWNERSHIP_UNKNOWN
+        )
+
+
+class TestClaimsFromControlPlaneDb(AllocatorOwnershipTestCase):
+    """End-to-end against the real substrate, not injected claim dicts."""
+
+    def _seed_lease(self, number: int, *, session_id: str, instance: str | None):
+        self.db.upsert_session(
+            session_id=session_id,
+            role="author",
+            profile="prgs-author",
+            pid=4242,
+            controller_instance_id=instance,
+        )
+        return self.db.assign_and_lease(
+            session_id=session_id,
+            role="author",
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            kind="issue",
+            number=number,
+        )
+
+    def test_controller_instance_id_persists_on_session(self):
+        row = self.db.upsert_session(
+            session_id="sess-x",
+            role="author",
+            profile="prgs-author",
+            pid=1,
+            controller_instance_id=MINE,
+        )
+        self.assertEqual(row["controller_instance_id"], MINE)
+
+    def test_heartbeat_without_instance_does_not_erase_ownership(self):
+        self.db.upsert_session(
+            session_id="sess-x",
+            role="author",
+            profile="prgs-author",
+            pid=1,
+            controller_instance_id=MINE,
+        )
+        row = self.db.upsert_session(
+            session_id="sess-x", role="author", profile="prgs-author", pid=1
+        )
+        self.assertEqual(row["controller_instance_id"], MINE)
+
+    def test_list_active_claims_surfaces_owner_instance(self):
+        self._seed_lease(607, session_id="sess-theirs", instance=THEIRS)
+        claims = self.db.list_active_claims(remote=REMOTE, org=ORG, repo=REPO)
+        self.assertIn(("issue", 607), claims)
+        self.assertEqual(claims[("issue", 607)]["controller_instance_id"], THEIRS)
+
+    def test_live_foreign_lease_is_excluded_without_injected_claims(self):
+        self._seed_lease(607, session_id="sess-theirs", instance=THEIRS)
+        result = allocate_next_work(
+            self.db,
+            session_id="sess-mine",
+            role="author",
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            candidates=[_issue(607), _issue(615)],
+            apply=False,
+            profile_name="prgs-author",
+            controller_instance_id=MINE,
+        )
+        self.assertEqual(result["selected"]["number"], 615)
+        self.assertEqual(
+            result["skipped"][0]["reason_code"], SKIP_CLAIMED_BY_OTHER_SESSION
+        )
+
+    def test_apply_reserves_the_unclaimed_issue(self):
+        self._seed_lease(607, session_id="sess-theirs", instance=THEIRS)
+        result = allocate_next_work(
+            self.db,
+            session_id="sess-mine",
+            role="author",
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            candidates=[_issue(607), _issue(615)],
+            apply=True,
+            profile_name="prgs-author",
+            controller_instance_id=MINE,
+        )
+        self.assertEqual(result["outcome"], "assigned_work")
+        self.assertEqual(result["selected"]["number"], 615)
+        self.assertEqual(result["assignment"]["work_number"], 615)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_allocator_pre_rank_exclusions.py

@@ -0,0 +1,349 @@
+"""Allocator pre-rank exclusions and candidates_json transport (#776).
+
+Covers:
+* #617 excluded before ranking (never leased when exclude_issue_numbers=[617]);
+* excluded top candidate selects the next safe candidate;
+* all candidates excluded → WAIT, no lease;
+* decoded-list and JSON-string candidates_json;
+* malformed / type-invalid fail-closed cases;
+* dry-run/apply fingerprint match and drift rejection;
+* foreign lease and same-owner lease on excluded issue;
+* skipped-accounting reason parity (excluded_by_controller);
+* public MCP entry-point coverage for exclude_issue_numbers.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import gitea_mcp_server as srv
+from allocator_service import (
+    OUTCOME_ASSIGNED,
+    OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE,
+    OUTCOME_CANDIDATE_SET_DRIFT,
+    OUTCOME_NO_SAFE,
+    OUTCOME_PREVIEW,
+    OUTCOME_WAIT,
+    SKIP_CLAIMED_BY_OTHER_SESSION,
+    SKIP_EXCLUDED_BY_CONTROLLER,
+    WorkCandidate,
+    allocate_next_work,
+    candidate_from_dict,
+    candidate_set_fingerprint,
+    normalize_candidates_payload,
+    normalize_exclude_issue_numbers,
+)
+from control_plane_db import ControlPlaneDB
+
+REMOTE = "prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+MINE = "ctl-mine-776"
+THEIRS = "ctl-theirs-776"
+
+
+def _issue(number: int, **kwargs) -> WorkCandidate:
+    base = dict(
+        kind="issue",
+        number=number,
+        state="open",
+        labels=("status:ready", "type:bug"),
+        title=f"issue {number}",
+        priority=20,
+    )
+    base.update(kwargs)
+    return WorkCandidate(**base)
+
+
+def _claim(number: int, *, session_id: str, instance: str | None, kind: str = "issue"):
+    return {
+        "lease_id": f"lease-{number}",
+        "session_id": session_id,
+        "controller_instance_id": instance,
+        "role": "author",
+        "profile": "prgs-author",
+        "expires_at": "2026-07-21T12:00:00Z",
+        "work_kind": kind,
+        "work_number": number,
+    }
+
+
+def _cand_dict(number: int, **kwargs) -> dict:
+    d = {
+        "kind": "issue",
+        "number": number,
+        "state": "open",
+        "labels": ["status:ready", "type:bug"],
+        "title": f"issue {number}",
+        "priority": 20,
+    }
+    d.update(kwargs)
+    return d
+
+
+class AllocatorExcludeServiceTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def _alloc(self, candidates, **kwargs):
+        defaults = dict(
+            session_id="sess-776",
+            role="author",
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            profile_name="prgs-author",
+            controller_instance_id=MINE,
+            claims={},
+            apply=False,
+        )
+        defaults.update(kwargs)
+        return allocate_next_work(self.db, candidates=candidates, **defaults)
+
+    def test_exclude_617_before_ranking_never_selects(self) -> None:
+        """AC2/AC8: highest-ranked #617 is removed before ranking."""
+        cands = [_issue(617), _issue(700), _issue(701)]
+        res = self._alloc(cands, exclude_issue_numbers=[617])
+        self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(res["selected"]["number"], 700)
+        skipped = {s["number"]: s for s in res["skipped"]}
+        self.assertIn(617, skipped)
+        self.assertEqual(
+            skipped[617]["reason_code"], SKIP_EXCLUDED_BY_CONTROLLER
+        )
+        self.assertIn(SKIP_EXCLUDED_BY_CONTROLLER, skipped[617]["reason"])
+
+    def test_excluded_top_selects_next_safe(self) -> None:
+        """AC2: excluding the oldest ready issue promotes the next number."""
+        cands = [_issue(600), _issue(601), _issue(602)]
+        res = self._alloc(cands, exclude_issue_numbers=[600])
+        self.assertEqual(res["selected"]["number"], 601)
+
+    def test_all_candidates_excluded_wait_no_lease(self) -> None:
+        """AC7: every candidate excluded → WAIT, no assignment."""
+        cands = [_issue(617), _issue(700)]
+        res = self._alloc(cands, exclude_issue_numbers=[617, 700], apply=True)
+        self.assertEqual(res["outcome"], OUTCOME_WAIT)
+        self.assertIsNone(res["selected"])
+        self.assertIsNone(res["assignment"])
+        self.assertEqual(len(res["controller_excluded"]), 2)
+
+    def test_omit_exclude_retains_existing_behavior(self) -> None:
+        """AC1/AC9: omit exclusions → #617 still wins when oldest ready."""
+        cands = [_issue(617), _issue(700)]
+        res = self._alloc(cands)
+        self.assertEqual(res["selected"]["number"], 617)
+        self.assertEqual(res.get("exclude_issue_numbers"), [])
+
+    def test_foreign_lease_still_skipped(self) -> None:
+        """AC6/AC9: foreign claims keep SKIP_CLAIMED_BY_OTHER_SESSION."""
+        cands = [_issue(617), _issue(700)]
+        claims = {
+            ("issue", 700): _claim(700, session_id="other", instance=THEIRS),
+        }
+        res = self._alloc(
+            cands, exclude_issue_numbers=[617], claims=claims
+        )
+        # 617 excluded, 700 foreign → wait, no selection
+        self.assertEqual(res["outcome"], OUTCOME_WAIT)
+        self.assertIsNone(res["selected"])
+        codes = {s["reason_code"] for s in res["skipped"]}
+        self.assertIn(SKIP_EXCLUDED_BY_CONTROLLER, codes)
+        self.assertIn(SKIP_CLAIMED_BY_OTHER_SESSION, codes)
+
+    def test_same_owner_lease_on_excluded_blocks_resume_release(self) -> None:
+        """AC5: excluded + live same-owner lease → structured blocker."""
+        cands = [_issue(617), _issue(700)]
+        claims = {
+            ("issue", 617): _claim(617, session_id="sess-776", instance=MINE),
+        }
+        res = self._alloc(
+            cands, exclude_issue_numbers=[617], claims=claims, apply=True
+        )
+        self.assertEqual(res["outcome"], OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE)
+        self.assertIsNone(res["assignment"])
+        self.assertEqual(res["blocked_lease"]["number"], 617)
+        self.assertIn("resume", res["blocked_lease"]["safe_next_action"])
+
+    def test_dry_run_apply_fingerprint_match(self) -> None:
+        """AC4: dry-run and apply share the same fingerprint."""
+        cands = [_issue(617), _issue(700)]
+        dry = self._alloc(cands, exclude_issue_numbers=[617], apply=False)
+        apply_res = self._alloc(
+            cands,
+            exclude_issue_numbers=[617],
+            apply=True,
+            expected_candidate_set_fingerprint=dry["candidate_set_fingerprint"],
+        )
+        self.assertEqual(
+            dry["candidate_set_fingerprint"],
+            apply_res["candidate_set_fingerprint"],
+        )
+        self.assertEqual(apply_res["outcome"], OUTCOME_ASSIGNED)
+        self.assertEqual(apply_res["selected"]["number"], 700)
+
+    def test_apply_rejects_fingerprint_drift(self) -> None:
+        """AC4: material candidate-set drift fails closed on apply."""
+        cands = [_issue(617), _issue(700)]
+        res = self._alloc(
+            cands,
+            exclude_issue_numbers=[617],
+            apply=True,
+            expected_candidate_set_fingerprint="0" * 64,
+        )
+        self.assertFalse(res["success"])
+        self.assertEqual(res["outcome"], OUTCOME_CANDIDATE_SET_DRIFT)
+        self.assertIsNone(res["assignment"])
+
+    def test_fingerprint_stable_helper(self) -> None:
+        cands = [_issue(700), _issue(617)]
+        a = candidate_set_fingerprint(cands, exclude_issue_numbers=[617])
+        b = candidate_set_fingerprint(
+            list(reversed(cands)), exclude_issue_numbers=[617]
+        )
+        self.assertEqual(a, b)
+
+    def test_normalize_exclude_rejects_bool(self) -> None:
+        with self.assertRaises(ValueError):
+            normalize_exclude_issue_numbers([True])
+
+    def test_normalize_exclude_rejects_scalar(self) -> None:
+        with self.assertRaises(ValueError):
+            normalize_exclude_issue_numbers(617)
+
+
+class CandidatesJsonNormalizeTest(unittest.TestCase):
+    def test_decoded_list(self) -> None:
+        """AC3: already-decoded list from MCP transport."""
+        cands = normalize_candidates_payload([_cand_dict(617), _cand_dict(700)])
+        self.assertEqual([c.number for c in cands], [617, 700])
+
+    def test_json_string(self) -> None:
+        """AC3: backward-compatible JSON string."""
+        raw = json.dumps([_cand_dict(617)])
+        cands = normalize_candidates_payload(raw)
+        self.assertEqual(cands[0].number, 617)
+
+    def test_malformed_json_fail_closed(self) -> None:
+        with self.assertRaises(ValueError) as ctx:
+            normalize_candidates_payload("{not json")
+        self.assertIn("malformed", str(ctx.exception).lower())
+
+    def test_scalar_fail_closed(self) -> None:
+        with self.assertRaises(ValueError):
+            normalize_candidates_payload(42)
+
+    def test_bool_number_fail_closed(self) -> None:
+        with self.assertRaises(ValueError):
+            normalize_candidates_payload([_cand_dict(True)])  # type: ignore[arg-type]
+
+    def test_invalid_record_fail_closed(self) -> None:
+        with self.assertRaises(ValueError):
+            normalize_candidates_payload(["not-a-dict"])
+
+    def test_object_not_list_fail_closed(self) -> None:
+        with self.assertRaises(ValueError):
+            normalize_candidates_payload(json.dumps({"number": 1}))
+
+    def test_candidate_from_dict_rejects_bool_number(self) -> None:
+        with self.assertRaises(ValueError):
+            candidate_from_dict({"kind": "issue", "number": True})
+
+
+class AllocateNextWorkMcpExcludeTest(unittest.TestCase):
+    """Public MCP entry-point coverage (#776 AC8)."""
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _call(self, **kwargs):
+        with patch("gitea_mcp_server._profile_operation_gate", return_value=None), patch(
+            "gitea_mcp_server._resolve", return_value=("h", ORG, REPO)
+        ), patch(
+            "gitea_mcp_server.get_profile",
+            return_value={"profile_name": "prgs-author", "role": "author"},
+        ), patch(
+            "gitea_mcp_server._authenticated_username", return_value="jcwalker3"
+        ), patch(
+            "gitea_mcp_server._control_plane_db_or_error", return_value=(self.db, [])
+        ), patch(
+            "gitea_mcp_server.sentry_observability.monitor_checkin", return_value=None
+        ):
+            return srv.gitea_allocate_next_work(
+                remote="prgs", org=ORG, repo=REPO, role="author", **kwargs
+            )
+
+    def test_mcp_exclude_617_decoded_list_never_selects(self) -> None:
+        """AC8: public tool with decoded list + exclude_issue_numbers=[617]."""
+        candidates = [_cand_dict(617), _cand_dict(700)]
+        res = self._call(
+            candidates_json=candidates,
+            exclude_issue_numbers=[617],
+            apply=False,
+        )
+        self.assertTrue(res.get("success"), res)
+        self.assertEqual(res["selected"]["number"], 700)
+        skipped = {s["number"]: s for s in res["skipped"]}
+        self.assertEqual(
+            skipped[617]["reason_code"], SKIP_EXCLUDED_BY_CONTROLLER
+        )
+        self.assertNotEqual(res["selected"]["number"], 617)
+
+    def test_mcp_exclude_617_json_string_apply(self) -> None:
+        """AC8: JSON-string transport + apply never leases #617."""
+        raw = json.dumps([_cand_dict(617), _cand_dict(700)])
+        res = self._call(
+            candidates_json=raw,
+            exclude_issue_numbers=[617],
+            apply=True,
+        )
+        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertEqual(res["assignment"]["work_number"], 700)
+        self.assertNotEqual(res["selected"]["number"], 617)
+
+    def test_mcp_malformed_candidates_json_fail_closed(self) -> None:
+        res = self._call(candidates_json="{bad", apply=False)
+        self.assertFalse(res["success"])
+        self.assertIsNone(res["assignment"])
+        self.assertTrue(any("fail closed" in r for r in res["reasons"]))
+
+    def test_mcp_bool_number_fail_closed(self) -> None:
+        res = self._call(
+            candidates_json=[{"kind": "issue", "number": True, "priority": 20}],
+            apply=False,
+        )
+        self.assertFalse(res["success"])
+        self.assertIsNone(res["assignment"])
+
+    def test_mcp_fingerprint_dry_run_apply_parity(self) -> None:
+        candidates = [_cand_dict(617), _cand_dict(700)]
+        dry = self._call(
+            candidates_json=candidates,
+            exclude_issue_numbers=[617],
+            apply=False,
+        )
+        apply_res = self._call(
+            candidates_json=candidates,
+            exclude_issue_numbers=[617],
+            apply=True,
+            expected_candidate_set_fingerprint=dry["candidate_set_fingerprint"],
+        )
+        self.assertEqual(
+            dry["candidate_set_fingerprint"],
+            apply_res["candidate_set_fingerprint"],
+        )
+        self.assertEqual(apply_res["selected"]["number"], 700)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_audit.py

@@ -238,7 +238,17 @@ class TestSimpleToolAudit(_AuditWiringBase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_close_issue_audited(self, _auth, mock_api):
-        mock_api.side_effect = [{"state": "closed"}, {"login": "mgr-bot"}]
+        # Keyed rather than positional: closing an issue also reads its labels
+        # before and after the state change for the #780 terminal cleanup and
+        # its read-after-write check, so call order is not a fixed sequence.
+        def api(method, url, auth, payload=None):
+            if method == "PATCH":
+                return {"state": "closed"}
+            if "/issues/" in url:
+                return {"number": 42, "labels": []}
+            return {"login": "mgr-bot"}
+
+        mock_api.side_effect = api
         with patch.dict(os.environ, self._env(), clear=True):
             gitea_close_issue(issue_number=42, remote="prgs")
         recs = self._records()

tests/test_mutation_budget_classifier.py

@@ -0,0 +1,388 @@
+"""Tests for the #617 mutation-budget classifier.
+
+Covers every acceptance criterion on issue #617:
+
+* AC1 — the classifier distinguishes local validator rejection, capability-gate
+  rejection, transport failure before API, and successful server-side mutation.
+* AC2 — pre-API validator failures do not consume server-side mutation budget.
+* AC3 — failed attempts are still logged in the local attempt ledger.
+* AC4 — the final report separately shows local failed attempts, blocked API
+  attempts, and successful server-side mutations.
+* AC5 — the six named scenarios, including the #615 reproduction where two
+  local validator rejections precede one successful comment.
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from mutation_budget_classifier import (
+    AMBIGUOUS_REQUIRES_READBACK,
+    CAPABILITY_GATE_REJECTION,
+    LOCAL_VALIDATOR_REJECTION,
+    SERVER_SIDE_MUTATION,
+    TRANSPORT_FAILURE_BEFORE_API,
+    assess_final_report_mutation_accounting,
+    classify_mutation_attempt,
+    record_attempt,
+    summarize_attempt_ledger,
+)
+
+# The two pre-API rejections observed on the #615 comment flow.
+MISSING_LEDGER_BLOCK = {
+    "success": False,
+    "performed": False,
+    "api_called": False,
+    "reasons": ["missing [THREAD STATE LEDGER] block"],
+}
+
+MISSING_CANONICAL_STATE = {
+    "success": False,
+    "performed": False,
+    "api_called": False,
+    "reasons": ["missing ## Canonical Issue State block"],
+}
+
+# The corrected comment that actually landed as #615 comment 9137.
+SUCCESSFUL_COMMENT = {
+    "success": True,
+    "performed": True,
+    "api_called": True,
+    "comment_id": 9137,
+    "issue_number": 615,
+}
+
+
+class TestAC1Classification(unittest.TestCase):
+    """AC1: the four outcome classes are distinguished."""
+
+    def test_local_validator_rejection_is_its_own_class(self):
+        result = classify_mutation_attempt(MISSING_LEDGER_BLOCK)
+        self.assertEqual(result["classification"], LOCAL_VALIDATOR_REJECTION)
+        self.assertTrue(result["pre_api"])
+
+    def test_capability_gate_rejection_is_its_own_class(self):
+        result = classify_mutation_attempt(
+            {
+                "success": False,
+                "api_called": False,
+                "permission_report": {"missing_permission": "gitea.issue.comment"},
+            }
+        )
+        self.assertEqual(result["classification"], CAPABILITY_GATE_REJECTION)
+        self.assertTrue(result["pre_api"])
+
+    def test_transport_failure_before_api_is_its_own_class(self):
+        result = classify_mutation_attempt(
+            {"success": False, "api_called": False, "transport_error": "EOF"}
+        )
+        self.assertEqual(result["classification"], TRANSPORT_FAILURE_BEFORE_API)
+        self.assertTrue(result["pre_api"])
+
+    def test_successful_server_mutation_is_its_own_class(self):
+        result = classify_mutation_attempt(SUCCESSFUL_COMMENT)
+        self.assertEqual(result["classification"], SERVER_SIDE_MUTATION)
+        self.assertFalse(result["pre_api"])
+
+    def test_each_class_is_distinct(self):
+        classes = {
+            classify_mutation_attempt(payload)["classification"]
+            for payload in (
+                MISSING_LEDGER_BLOCK,
+                {"success": False, "api_called": False, "capability_blocked": True},
+                {"success": False, "api_called": False, "transport_failed": True},
+                SUCCESSFUL_COMMENT,
+            )
+        }
+        self.assertEqual(len(classes), 4)
+
+
+class TestAC2BudgetAccounting(unittest.TestCase):
+    """AC2: pre-API failures never consume server-side mutation budget."""
+
+    def test_missing_thread_state_ledger_not_counted_as_mutation(self):
+        result = classify_mutation_attempt(MISSING_LEDGER_BLOCK)
+        self.assertFalse(result["budget_consumed"])
+        self.assertIs(result["api_called"], False)
+
+    def test_missing_canonical_issue_state_not_counted_as_mutation(self):
+        result = classify_mutation_attempt(MISSING_CANONICAL_STATE)
+        self.assertFalse(result["budget_consumed"])
+        self.assertIs(result["api_called"], False)
+
+    def test_transport_failure_before_api_not_counted_as_mutation(self):
+        result = classify_mutation_attempt(
+            {
+                "success": False,
+                "api_called": False,
+                "transport_error": "connection reset",
+            }
+        )
+        self.assertFalse(result["budget_consumed"])
+
+    def test_capability_gate_block_not_counted_as_mutation(self):
+        result = classify_mutation_attempt(
+            {
+                "success": False,
+                "api_called": False,
+                "permission_report": {"missing_permission": "gitea.pr.merge"},
+            }
+        )
+        self.assertFalse(result["budget_consumed"])
+
+    def test_successful_comment_with_comment_id_counts_as_one_mutation(self):
+        result = classify_mutation_attempt(SUCCESSFUL_COMMENT)
+        self.assertTrue(result["budget_consumed"])
+        self.assertEqual(result["proof_fields"], ["comment_id"])
+
+
+class TestAC2FailsClosed(unittest.TestCase):
+    """AC2 must not become a loophole: ambiguity still fails closed."""
+
+    def test_api_reached_without_proof_is_ambiguous_and_consumes_budget(self):
+        result = classify_mutation_attempt({"success": True, "api_called": True})
+        self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK)
+        self.assertTrue(result["budget_consumed"])
+        self.assertTrue(result["requires_readback"])
+
+    def test_missing_api_called_signal_fails_closed(self):
+        result = classify_mutation_attempt({"success": False})
+        self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK)
+        self.assertTrue(result["budget_consumed"])
+        self.assertIsNone(result["api_called"])
+
+    def test_empty_and_none_results_fail_closed(self):
+        for payload in ({}, None):
+            result = classify_mutation_attempt(payload)
+            self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK)
+            self.assertTrue(result["budget_consumed"])
+
+    def test_success_with_proof_counts_even_when_api_called_absent(self):
+        result = classify_mutation_attempt({"success": True, "comment_id": 13320})
+        self.assertEqual(result["classification"], SERVER_SIDE_MUTATION)
+        self.assertTrue(result["budget_consumed"])
+
+    def test_blank_proof_field_is_not_proof(self):
+        result = classify_mutation_attempt(
+            {"success": True, "api_called": True, "merge_commit_sha": "   "}
+        )
+        self.assertEqual(result["classification"], AMBIGUOUS_REQUIRES_READBACK)
+
+
+class TestAC3AttemptLedger(unittest.TestCase):
+    """AC3: failed attempts are still logged locally."""
+
+    def test_failed_attempts_are_recorded(self):
+        ledger: list[dict] = []
+        record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="create_issue_comment")
+        record_attempt(ledger, MISSING_CANONICAL_STATE, operation="create_issue_comment")
+        self.assertEqual(len(ledger), 2)
+        self.assertTrue(
+            all(e["classification"] == LOCAL_VALIDATOR_REJECTION for e in ledger)
+        )
+
+    def test_recorded_entry_carries_operation_and_timestamp(self):
+        ledger: list[dict] = []
+        entry = record_attempt(
+            ledger,
+            SUCCESSFUL_COMMENT,
+            operation="create_issue_comment",
+            timestamp="2026-07-20T18:15:04+00:00",
+        )
+        self.assertEqual(entry["operation"], "create_issue_comment")
+        self.assertEqual(entry["timestamp"], "2026-07-20T18:15:04+00:00")
+
+    def test_timestamp_is_generated_when_omitted(self):
+        ledger: list[dict] = []
+        entry = record_attempt(ledger, SUCCESSFUL_COMMENT)
+        self.assertTrue(entry["timestamp"])
+
+
+class TestAC5CorrectedCommentAllowed(unittest.TestCase):
+    """AC5: the #615 reproduction — two local rejections then one success."""
+
+    def _replay_615_flow(self) -> list[dict]:
+        ledger: list[dict] = []
+        record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="create_issue_comment")
+        record_attempt(ledger, MISSING_CANONICAL_STATE, operation="create_issue_comment")
+        record_attempt(ledger, SUCCESSFUL_COMMENT, operation="create_issue_comment")
+        return ledger
+
+    def test_corrected_comment_after_two_rejections_is_allowed(self):
+        summary = summarize_attempt_ledger(self._replay_615_flow())
+        # The regression: budget must show ONE mutation, not three attempts.
+        self.assertEqual(summary["successful_server_mutations"], 1)
+        self.assertEqual(summary["budget_consumed"], 1)
+
+    def test_all_three_attempts_remain_visible(self):
+        summary = summarize_attempt_ledger(self._replay_615_flow())
+        self.assertEqual(summary["total_attempts"], 3)
+        self.assertEqual(summary["local_failed_attempts"], 2)
+
+    def test_no_readback_required_for_clean_flow(self):
+        summary = summarize_attempt_ledger(self._replay_615_flow())
+        self.assertFalse(summary["requires_readback"])
+
+
+class TestAC4FinalReportAccounting(unittest.TestCase):
+    """AC4: the report must show the three categories, and match the ledger."""
+
+    def _mixed_ledger(self) -> list[dict]:
+        ledger: list[dict] = []
+        record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="comment")
+        record_attempt(ledger, MISSING_CANONICAL_STATE, operation="comment")
+        record_attempt(
+            ledger,
+            {"success": False, "api_called": False, "transport_error": "EOF"},
+            operation="comment",
+        )
+        record_attempt(
+            ledger,
+            {"success": False, "api_called": False, "capability_blocked": True},
+            operation="merge",
+        )
+        record_attempt(ledger, SUCCESSFUL_COMMENT, operation="comment")
+        return ledger
+
+    def test_summary_separates_the_three_categories(self):
+        summary = summarize_attempt_ledger(self._mixed_ledger())
+        self.assertEqual(summary["local_failed_attempts"], 2)
+        self.assertEqual(summary["blocked_api_attempts"], 2)
+        self.assertEqual(summary["successful_server_mutations"], 1)
+
+    def test_matching_report_is_valid(self):
+        result = assess_final_report_mutation_accounting(
+            {
+                "local_failed_attempts": 2,
+                "blocked_api_attempts": 2,
+                "successful_server_mutations": 1,
+            },
+            self._mixed_ledger(),
+        )
+        self.assertTrue(result["valid"], result["reasons"])
+
+    def test_omitted_category_fails_closed(self):
+        result = assess_final_report_mutation_accounting(
+            {"local_failed_attempts": 2, "blocked_api_attempts": 2},
+            self._mixed_ledger(),
+        )
+        self.assertFalse(result["valid"])
+        self.assertTrue(
+            any("successful_server_mutations" in r for r in result["reasons"])
+        )
+
+    def test_inflated_mutation_count_fails_closed(self):
+        # The #617 bug shape: claiming three mutations when only one landed.
+        result = assess_final_report_mutation_accounting(
+            {
+                "local_failed_attempts": 2,
+                "blocked_api_attempts": 2,
+                "successful_server_mutations": 3,
+            },
+            self._mixed_ledger(),
+        )
+        self.assertFalse(result["valid"])
+        self.assertTrue(
+            any("successful_server_mutations=3" in r for r in result["reasons"])
+        )
+
+    def test_ambiguous_attempt_requires_readback_proof(self):
+        ledger: list[dict] = []
+        record_attempt(ledger, {"success": True, "api_called": True}, operation="comment")
+        report = {
+            "local_failed_attempts": 0,
+            "blocked_api_attempts": 0,
+            "successful_server_mutations": 0,
+        }
+        blocked = assess_final_report_mutation_accounting(report, ledger)
+        self.assertFalse(blocked["valid"])
+        self.assertTrue(any("readback_verified" in r for r in blocked["reasons"]))
+
+        allowed = assess_final_report_mutation_accounting(
+            {**report, "readback_verified": True}, ledger
+        )
+        self.assertTrue(allowed["valid"], allowed["reasons"])
+
+    def test_ledger_summary_is_returned_without_raw_entries(self):
+        result = assess_final_report_mutation_accounting({}, self._mixed_ledger())
+        self.assertNotIn("entries", result["ledger_summary"])
+        self.assertEqual(result["ledger_summary"]["total_attempts"], 5)
+
+
+class TestEmptyLedger(unittest.TestCase):
+    def test_empty_ledger_summarizes_to_zero(self):
+        summary = summarize_attempt_ledger([])
+        self.assertEqual(summary["total_attempts"], 0)
+        self.assertEqual(summary["successful_server_mutations"], 0)
+        self.assertFalse(summary["requires_readback"])
+
+    def test_none_ledger_is_tolerated(self):
+        self.assertEqual(summarize_attempt_ledger(None)["total_attempts"], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class TestValidatorIntegration(unittest.TestCase):
+    """The classifier is wired into the shared final-report validator (AC4)."""
+
+    def _ledger_two_rejections_one_success(self) -> list[dict]:
+        ledger: list[dict] = []
+        record_attempt(ledger, MISSING_LEDGER_BLOCK, operation="comment")
+        record_attempt(ledger, MISSING_CANONICAL_STATE, operation="comment")
+        record_attempt(ledger, SUCCESSFUL_COMMENT, operation="comment")
+        return ledger
+
+    def test_rule_is_noop_without_a_ledger(self):
+        from final_report_validator import assess_final_report_validator
+
+        result = assess_final_report_validator("some report", "review_pr")
+        self.assertFalse(
+            any(
+                f["rule_id"] == "shared.mutation_budget_accounting"
+                for f in result["findings"]
+            )
+        )
+
+    def test_report_matching_ledger_produces_no_finding(self):
+        from final_report_validator import assess_final_report_validator
+
+        report = (
+            "Local failed attempts: 2\n"
+            "Blocked API attempts: 0\n"
+            "Successful server-side mutations: 1\n"
+        )
+        result = assess_final_report_validator(
+            report,
+            "review_pr",
+            mutation_attempt_ledger=self._ledger_two_rejections_one_success(),
+        )
+        self.assertFalse(
+            any(
+                f["rule_id"] == "shared.mutation_budget_accounting"
+                for f in result["findings"]
+            )
+        )
+
+    def test_counting_rejections_as_mutations_is_blocked(self):
+        from final_report_validator import assess_final_report_validator
+
+        # The #617 bug: three attempts reported as three server-side mutations.
+        report = (
+            "Local failed attempts: 0\n"
+            "Blocked API attempts: 0\n"
+            "Successful server-side mutations: 3\n"
+        )
+        result = assess_final_report_validator(
+            report,
+            "review_pr",
+            mutation_attempt_ledger=self._ledger_two_rejections_one_success(),
+        )
+        findings = [
+            f
+            for f in result["findings"]
+            if f["rule_id"] == "shared.mutation_budget_accounting"
+        ]
+        self.assertTrue(findings)
+        self.assertTrue(all(f["severity"] == "block" for f in findings))

tests/test_self_propagating_handoff.py

@@ -0,0 +1,613 @@
+"""Tests for self-propagating canonical handoffs (#626)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from final_report_validator import assess_final_report_validator  # noqa: E402
+from self_propagating_handoff import (  # noqa: E402
+    HANDOFF_FIELDS,
+    NEXT_ACTOR_BY_STATE,
+    WORKFLOW_STATES,
+    assess_controller_decision,
+    assess_durable_state_update,
+    assess_final_report_self_propagating_handoff,
+    assess_handoff_live_state,
+    assess_merge_completion_transition,
+    assess_role_continuation,
+    assess_self_propagating_handoff,
+    assess_thread_recoverability,
+    assess_workflow_failure_escalation,
+    parse_self_propagating_handoff,
+    render_self_propagating_handoff,
+)
+
+REPO = "Scaled-Tech-Consulting/Gitea-Tools"
+
+AUTHOR_PROMPT = (
+    "Review PR #900 on Scaled-Tech-Consulting/Gitea-Tools for issue 626 at head "
+    "aaaa111. Validate the branch, then submit an independent review verdict."
+)
+REVIEWER_PROMPT = (
+    "Merge PR #900 on Scaled-Tech-Consulting/Gitea-Tools for issue 626 once the "
+    "approval at head aaaa111 still applies to the live head."
+)
+MERGER_PROMPT = (
+    "Accept or reject the merged work for issue 626 on "
+    "Scaled-Tech-Consulting/Gitea-Tools; verify acceptance criteria then close."
+)
+CONTROLLER_PROMPT = (
+    "Address the controller's requested changes for issue 626 on "
+    "Scaled-Tech-Consulting/Gitea-Tools, then hand back to an independent reviewer."
+)
+
+
+def build_handoff(**overrides):
+    """Render a valid author -> reviewer handoff, with overrides applied."""
+    values = {
+        "REPOSITORY": REPO,
+        "ISSUE": "626",
+        "PR": "900",
+        "WORKFLOW_STATE": "needs-review",
+        "HEAD_SHA": "aaaa111",
+        "BASE_BRANCH": "master",
+        "BASE_OR_MERGE_SHA": "bbbb222",
+        "ACTING_ROLE": "author",
+        "ACTING_IDENTITY": "jcwalker3 (prgs-author)",
+        "COMPLETED_ACTIONS": "implemented AC1-AC9; opened PR #900",
+        "VALIDATION_EVIDENCE": "pytest tests/test_self_propagating_handoff.py: 20 passed",
+        "MUTATION_LEDGER": "branch pushed; PR #900 opened; comment 13547 posted",
+        "BLOCKERS": "none",
+        "NEXT_ACTOR": "reviewer",
+        "NEXT_ACTION": "independently review PR #900 at head aaaa111",
+        "PROHIBITED_ACTIONS": "merge, self-approve, force-push",
+        "NEXT_PROMPT": AUTHOR_PROMPT,
+        "WORKFLOW_FAILURE_ISSUES": "none",
+        "LAST_UPDATED": "2026-07-21T03:55:00Z",
+    }
+    values.update(overrides)
+    return render_self_propagating_handoff(**values)
+
+
+class RenderAndParseTests(unittest.TestCase):
+    def test_render_emits_every_canonical_field(self):
+        body = build_handoff()
+        parsed = parse_self_propagating_handoff(body)
+        self.assertIsNotNone(parsed)
+        for name in HANDOFF_FIELDS:
+            self.assertIn(name, parsed)
+
+    def test_render_rejects_unknown_workflow_state(self):
+        with self.assertRaises(ValueError):
+            build_handoff(WORKFLOW_STATE="almost-done")
+
+    def test_every_state_maps_to_exactly_one_actor(self):
+        self.assertEqual(set(WORKFLOW_STATES), set(NEXT_ACTOR_BY_STATE))
+
+    def test_absent_block_parses_as_none(self):
+        self.assertIsNone(parse_self_propagating_handoff("no handoff here"))
+
+
+class AuthorToReviewerTests(unittest.TestCase):
+    """Scenario 1: author -> reviewer."""
+
+    def test_valid_author_handoff_passes(self):
+        result = assess_self_propagating_handoff(build_handoff())
+        self.assertTrue(result["valid"], result["reasons"])
+        self.assertEqual(result["next_actor"], "reviewer")
+        self.assertFalse(result["terminal"])
+
+    def test_reviewer_may_continue_author_handoff(self):
+        result = assess_role_continuation(
+            handoff=build_handoff(), actor_role="reviewer"
+        )
+        self.assertTrue(result["allowed"], result["reasons"])
+        self.assertIn("review", result["allowed_actions"])
+
+    def test_author_may_not_continue_its_own_handoff(self):
+        result = assess_role_continuation(
+            handoff=build_handoff(), actor_role="author"
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["expected_actor"], "reviewer")
+
+    def test_next_actor_must_match_declared_state(self):
+        result = assess_self_propagating_handoff(
+            build_handoff(NEXT_ACTOR="merger")
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("does not match state" in reason for reason in result["reasons"])
+        )
+
+
+class ReviewerToMergerTests(unittest.TestCase):
+    """Scenario 2: reviewer -> merger."""
+
+    def build(self, **overrides):
+        values = {
+            "WORKFLOW_STATE": "approved-awaiting-merge",
+            "ACTING_ROLE": "reviewer",
+            "ACTING_IDENTITY": "reviewer-bot (prgs-reviewer)",
+            "COMPLETED_ACTIONS": "review 500 APPROVED at aaaa111",
+            "NEXT_ACTOR": "merger",
+            "NEXT_ACTION": "merge PR #900 at approved head aaaa111",
+            "PROHIBITED_ACTIONS": "re-review, commit, push",
+            "NEXT_PROMPT": REVIEWER_PROMPT,
+        }
+        values.update(overrides)
+        return build_handoff(**values)
+
+    def test_reviewer_handoff_is_valid(self):
+        result = assess_self_propagating_handoff(self.build())
+        self.assertTrue(result["valid"], result["reasons"])
+        self.assertEqual(result["next_actor"], "merger")
+
+    def test_merger_may_continue(self):
+        result = assess_role_continuation(handoff=self.build(), actor_role="merger")
+        self.assertTrue(result["allowed"], result["reasons"])
+        self.assertIn("merge", result["allowed_actions"])
+
+
+class MergerToControllerTests(unittest.TestCase):
+    """Scenario 3: merger -> controller."""
+
+    def test_merge_success_stops_at_controller_boundary(self):
+        result = assess_merge_completion_transition(merge_succeeded=True)
+        self.assertEqual(result["next_state"], "merged-awaiting-controller")
+        self.assertEqual(result["next_actor"], "controller")
+        self.assertTrue(result["next_prompt_required"])
+
+    def test_configured_auto_accept_may_complete(self):
+        result = assess_merge_completion_transition(
+            merge_succeeded=True, controller_auto_accept=True
+        )
+        self.assertEqual(result["next_state"], "complete")
+        self.assertFalse(result["next_prompt_required"])
+
+    def test_failed_merge_keeps_the_work_item_with_the_merger(self):
+        result = assess_merge_completion_transition(merge_succeeded=False)
+        self.assertEqual(result["next_state"], "approved-awaiting-merge")
+
+    def test_merger_handoff_names_the_controller(self):
+        body = build_handoff(
+            WORKFLOW_STATE="merged-awaiting-controller",
+            ACTING_ROLE="merger",
+            ACTING_IDENTITY="merger-bot (prgs-merger)",
+            COMPLETED_ACTIONS="merged PR #900 as cccc333",
+            BASE_OR_MERGE_SHA="cccc333",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="verify acceptance criteria and close issue 626",
+            PROHIBITED_ACTIONS="reopen the PR, re-merge",
+            NEXT_PROMPT=MERGER_PROMPT,
+        )
+        result = assess_self_propagating_handoff(body)
+        self.assertTrue(result["valid"], result["reasons"])
+        self.assertEqual(result["next_actor"], "controller")
+
+
+class ControllerBackToAuthorTests(unittest.TestCase):
+    """Scenario 4: controller -> author."""
+
+    def test_request_corrections_returns_to_author(self):
+        result = assess_controller_decision(decision="request_corrections")
+        self.assertFalse(result["block"])
+        self.assertEqual(result["next_state"], "needs-author")
+        self.assertTrue(result["next_prompt_required"])
+
+    def test_return_to_actor_requires_a_named_target(self):
+        result = assess_controller_decision(decision="return_to_actor")
+        self.assertTrue(result["block"])
+
+    def test_return_to_reviewer_is_supported(self):
+        result = assess_controller_decision(
+            decision="return_to_actor", return_to="reviewer"
+        )
+        self.assertEqual(result["next_state"], "needs-review")
+
+    def test_unknown_decision_fails_closed(self):
+        result = assess_controller_decision(decision="looks-fine")
+        self.assertTrue(result["block"])
+
+    def test_controller_handoff_back_to_author_validates(self):
+        body = build_handoff(
+            WORKFLOW_STATE="needs-author",
+            ACTING_ROLE="controller",
+            ACTING_IDENTITY="controller (operator)",
+            COMPLETED_ACTIONS="reviewed merged work; requested corrections",
+            NEXT_ACTOR="author",
+            NEXT_ACTION="address controller corrections on issue 626",
+            PROHIBITED_ACTIONS="close the issue, merge",
+            NEXT_PROMPT=CONTROLLER_PROMPT,
+        )
+        result = assess_self_propagating_handoff(body)
+        self.assertTrue(result["valid"], result["reasons"])
+
+
+class StaleHeadRejectionTests(unittest.TestCase):
+    """Scenario 5: stale-head rejection."""
+
+    def test_changed_head_invalidates_a_merge_handoff(self):
+        body = build_handoff(
+            WORKFLOW_STATE="approved-awaiting-merge",
+            ACTING_ROLE="reviewer",
+            NEXT_ACTOR="merger",
+            NEXT_ACTION="merge PR #900 at approved head aaaa111",
+            NEXT_PROMPT=REVIEWER_PROMPT,
+        )
+        result = assess_handoff_live_state(
+            handoff=body,
+            live={"pr_head_sha": "dddd444", "pr_state": "open"},
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("changed_pr_head", result["kinds"])
+        self.assertEqual(result["recovered_state"], "needs-review")
+
+    def test_stale_approval_blocks_the_merger(self):
+        body = build_handoff(
+            WORKFLOW_STATE="approved-awaiting-merge",
+            NEXT_ACTOR="merger",
+            NEXT_ACTION="merge PR #900",
+            HEAD_SHA="dddd444",
+            NEXT_PROMPT=REVIEWER_PROMPT,
+        )
+        result = assess_handoff_live_state(
+            handoff=body,
+            live={"pr_head_sha": "dddd444", "approved_head_sha": "aaaa111"},
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("stale_approval", result["kinds"])
+
+    def test_unchanged_head_is_not_blocked(self):
+        result = assess_handoff_live_state(
+            handoff=build_handoff(),
+            live={
+                "pr_head_sha": "aaaa111",
+                "pr_state": "open",
+                "issue_state": "open",
+                "base_branch": "master",
+                "namespace_role": "reviewer",
+            },
+        )
+        self.assertFalse(result["block"], result["reasons"])
+
+    def test_merged_pr_recovers_to_the_controller_boundary(self):
+        result = assess_handoff_live_state(
+            handoff=build_handoff(),
+            live={"pr_head_sha": "aaaa111", "pr_state": "merged"},
+        )
+        self.assertIn("pr_merged", result["kinds"])
+        self.assertEqual(result["recovered_state"], "merged-awaiting-controller")
+
+    def test_reopened_issue_invalidates_a_complete_handoff(self):
+        body = build_handoff(
+            WORKFLOW_STATE="complete",
+            ACTING_ROLE="controller",
+            NEXT_ACTOR="none",
+            NEXT_ACTION="none",
+            NEXT_PROMPT="none",
+        )
+        result = assess_handoff_live_state(
+            handoff=body, live={"issue_state": "open"}
+        )
+        self.assertIn("issue_reopened", result["kinds"])
+        self.assertEqual(result["recovered_state"], "needs-author")
+
+    def test_foreign_lease_and_worktree_faults_are_detected(self):
+        result = assess_handoff_live_state(
+            handoff=build_handoff(),
+            live={
+                "pr_head_sha": "aaaa111",
+                "lease": {"status": "expired", "session_id": "other-session"},
+                "actor_session_id": "my-session",
+                "worktree": {"present": False, "dirty": True},
+                "namespace_role": "author",
+                "runtime_stale": True,
+                "base_branch": "dev",
+                "conflicting_canonical_comments": True,
+            },
+        )
+        for kind in (
+            "stale_lease",
+            "foreign_lease",
+            "missing_worktree",
+            "dirty_worktree",
+            "namespace_mismatch",
+            "stale_runtime",
+            "changed_base",
+            "conflicting_canonical_comments",
+        ):
+            self.assertIn(kind, result["kinds"])
+
+
+class BlockedInfrastructurePathTests(unittest.TestCase):
+    """Scenario 6: blocked infrastructure path."""
+
+    def build(self, **overrides):
+        values = {
+            "WORKFLOW_STATE": "blocked",
+            "PR": "none",
+            "HEAD_SHA": "none",
+            "ACTING_ROLE": "author",
+            "COMPLETED_ACTIONS": "attempted native publish; MCP mutation rejected",
+            "BLOCKERS": "gitea_create_pr rejected: namespace unreachable",
+            "NEXT_ACTOR": "operator",
+            "NEXT_ACTION": "restore the author MCP namespace",
+            "PROHIBITED_ACTIONS": "raw git push, curl, force-push",
+            "NEXT_PROMPT": (
+                "Repair the author MCP namespace for "
+                "Scaled-Tech-Consulting/Gitea-Tools so issue 626 can publish "
+                "natively, then hand back to the author."
+            ),
+            "WORKFLOW_FAILURE_ISSUES": "#640",
+        }
+        values.update(overrides)
+        return build_handoff(**values)
+
+    def test_blocked_handoff_without_pr_is_valid(self):
+        result = assess_self_propagating_handoff(self.build())
+        self.assertTrue(result["valid"], result["reasons"])
+        self.assertEqual(result["next_actor"], "operator")
+
+    def test_blocked_requires_a_concrete_blocker(self):
+        result = assess_self_propagating_handoff(self.build(BLOCKERS="none"))
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("BLOCKERS" in reason for reason in result["reasons"])
+        )
+
+    def test_operator_is_the_only_authorized_continuation(self):
+        self.assertTrue(
+            assess_role_continuation(handoff=self.build(), actor_role="operator")[
+                "allowed"
+            ]
+        )
+        self.assertTrue(
+            assess_role_continuation(handoff=self.build(), actor_role="merger")["block"]
+        )
+
+
+class FinalClosureTests(unittest.TestCase):
+    """Scenario 7: final successful closure."""
+
+    def build(self, **overrides):
+        values = {
+            "WORKFLOW_STATE": "complete",
+            "ACTING_ROLE": "controller",
+            "ACTING_IDENTITY": "controller (operator)",
+            "COMPLETED_ACTIONS": "verified acceptance criteria; closed issue 626",
+            "BASE_OR_MERGE_SHA": "cccc333",
+            "NEXT_ACTOR": "none",
+            "NEXT_ACTION": "none",
+            "PROHIBITED_ACTIONS": "reopen without new evidence",
+            "NEXT_PROMPT": "none",
+        }
+        values.update(overrides)
+        return build_handoff(**values)
+
+    def test_terminal_handoff_is_valid_without_a_next_prompt(self):
+        result = assess_self_propagating_handoff(self.build())
+        self.assertTrue(result["valid"], result["reasons"])
+        self.assertTrue(result["terminal"])
+
+    def test_terminal_handoff_must_not_manufacture_more_work(self):
+        result = assess_self_propagating_handoff(
+            self.build(NEXT_PROMPT=CONTROLLER_PROMPT)
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("must not carry a NEXT_PROMPT" in r for r in result["reasons"])
+        )
+
+    def test_no_role_may_continue_a_complete_workflow(self):
+        result = assess_role_continuation(handoff=self.build(), actor_role="author")
+        self.assertTrue(result["block"])
+
+    def test_controller_acceptance_requires_full_closure_proof(self):
+        partial = assess_controller_decision(
+            decision="accept",
+            closure_proof={"acceptance_criteria_satisfied": True},
+        )
+        self.assertTrue(partial["block"])
+        self.assertEqual(partial["next_state"], "merged-awaiting-controller")
+
+        full = assess_controller_decision(
+            decision="accept",
+            closure_proof={
+                "acceptance_criteria_satisfied": True,
+                "cleanup_complete": True,
+                "canonical_final_state_posted": True,
+                "issue_closed_through_workflow": True,
+            },
+        )
+        self.assertFalse(full["block"])
+        self.assertEqual(full["next_state"], "complete")
+        self.assertFalse(full["next_prompt_required"])
+
+
+class IncompleteHandoffRejectionTests(unittest.TestCase):
+    """Scenario 8: incomplete handoff rejection."""
+
+    def test_missing_block_is_rejected(self):
+        result = assess_self_propagating_handoff("Work is done, ping the reviewer.")
+        self.assertTrue(result["block"])
+        self.assertFalse(result["present"])
+
+    def test_missing_field_is_rejected(self):
+        body = build_handoff()
+        body = "\n".join(
+            line for line in body.splitlines() if not line.startswith("MUTATION_LEDGER:")
+        )
+        result = assess_self_propagating_handoff(body)
+        self.assertTrue(result["block"])
+        self.assertIn("MUTATION_LEDGER", result["missing_fields"])
+
+    def test_placeholder_field_is_rejected(self):
+        result = assess_self_propagating_handoff(
+            build_handoff(VALIDATION_EVIDENCE="TBD")
+        )
+        self.assertTrue(result["block"])
+
+    def test_stub_next_prompt_is_rejected(self):
+        result = assess_self_propagating_handoff(build_handoff(NEXT_PROMPT="review it"))
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("ready-to-run" in reason for reason in result["reasons"])
+        )
+
+    def test_prompt_depending_on_outside_chat_is_rejected(self):
+        prompt = (
+            "Continue issue 626 on Scaled-Tech-Consulting/Gitea-Tools using the "
+            "previous chat for the missing details."
+        )
+        result = assess_thread_recoverability(build_handoff(NEXT_PROMPT=prompt))
+        self.assertTrue(result["block"])
+
+    def test_prompt_must_name_repository_and_issue(self):
+        prompt = (
+            "Please review the pull request at the current head and submit an "
+            "independent verdict when validation passes."
+        )
+        result = assess_thread_recoverability(build_handoff(NEXT_PROMPT=prompt))
+        self.assertTrue(result["block"])
+
+    def test_self_contained_prompt_is_recoverable(self):
+        self.assertFalse(assess_thread_recoverability(build_handoff())["block"])
+
+    def test_chat_only_report_is_not_durable(self):
+        result = assess_durable_state_update(
+            handoff_text=build_handoff(),
+            posted_comment_id=None,
+            canonical_state_posted=False,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(len(result["reasons"]), 2)
+
+    def test_posted_handoff_is_durable(self):
+        result = assess_durable_state_update(
+            handoff_text=build_handoff(),
+            posted_comment_id=13550,
+            canonical_state_posted=True,
+        )
+        self.assertTrue(result["durable"], result["reasons"])
+
+
+class WorkflowFailureEscalationTests(unittest.TestCase):
+    """Scenario 9: duplicate workflow-failure issue handling."""
+
+    def failure(self, **overrides):
+        values = {
+            "signature": "lease-cleanup-internal-error",
+            "classification": "mcp-tool-defect",
+            "linked_issue": "718",
+            "temporary_impact": "lease cleanup unavailable this session",
+            "next_valid_actor": "operator",
+            "recovery_prompt": "restart the namespace and re-run lease cleanup",
+        }
+        values.update(overrides)
+        return values
+
+    def test_complete_failure_record_passes(self):
+        result = assess_workflow_failure_escalation(
+            failures=[self.failure()], active_issue_number=626
+        )
+        self.assertTrue(result["escalated"], result["reasons"])
+
+    def test_incomplete_failure_record_fails_closed(self):
+        result = assess_workflow_failure_escalation(
+            failures=[self.failure(recovery_prompt="")], active_issue_number=626
+        )
+        self.assertTrue(result["block"])
+
+    def test_folding_into_the_active_issue_is_rejected(self):
+        result = assess_workflow_failure_escalation(
+            failures=[self.failure(linked_issue="626")], active_issue_number=626
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("folded into the active work item" in r for r in result["reasons"])
+        )
+
+    def test_known_signature_reuses_the_existing_issue(self):
+        result = assess_workflow_failure_escalation(
+            failures=[self.failure()],
+            active_issue_number=626,
+            existing_failure_issues=[
+                {"signature": "lease-cleanup-internal-error", "number": 718}
+            ],
+        )
+        self.assertTrue(result["escalated"], result["reasons"])
+        self.assertEqual(result["reused_issues"], [
+            {"signature": "lease-cleanup-internal-error", "issue": "718"}
+        ])
+
+    def test_duplicate_issue_for_known_signature_is_rejected(self):
+        result = assess_workflow_failure_escalation(
+            failures=[self.failure(linked_issue="799")],
+            active_issue_number=626,
+            existing_failure_issues=[
+                {"signature": "lease-cleanup-internal-error", "number": 718}
+            ],
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("reuse the existing issue #718" in r for r in result["reasons"])
+        )
+
+    def test_same_signature_twice_in_one_session_is_rejected(self):
+        result = assess_workflow_failure_escalation(
+            failures=[self.failure(), self.failure()], active_issue_number=626
+        )
+        self.assertTrue(result["block"])
+
+    def test_no_failures_is_not_an_error(self):
+        result = assess_workflow_failure_escalation(
+            failures=[], active_issue_number=626
+        )
+        self.assertTrue(result["escalated"])
+
+
+class FinalReportIntegrationTests(unittest.TestCase):
+    def test_report_without_the_protocol_is_not_applicable(self):
+        result = assess_final_report_self_propagating_handoff("## Controller Handoff\n")
+        self.assertFalse(result["applicable"])
+        self.assertFalse(result["block"])
+
+    def test_report_with_a_complete_handoff_passes(self):
+        result = assess_final_report_self_propagating_handoff(build_handoff())
+        self.assertTrue(result["applicable"])
+        self.assertFalse(result["block"], result["reasons"])
+
+    def test_report_with_an_incomplete_handoff_blocks(self):
+        result = assess_final_report_self_propagating_handoff(
+            build_handoff(NEXT_ACTION="")
+        )
+        self.assertTrue(result["block"])
+
+    def test_validator_blocks_an_incomplete_handoff_in_a_work_issue_report(self):
+        report = build_handoff(MUTATION_LEDGER="TBD")
+        result = assess_final_report_validator(report, "work_issue")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                finding["rule_id"] == "shared.self_propagating_handoff"
+                for finding in result["findings"]
+            )
+        )
+
+    def test_validator_ignores_reports_that_predate_the_protocol(self):
+        result = assess_final_report_validator("plain legacy report", "work_issue")
+        self.assertFalse(
+            any(
+                finding["rule_id"] == "shared.self_propagating_handoff"
+                for finding in result["findings"]
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_terminal_pr_label_cleanup.py

@@ -0,0 +1,555 @@
+"""#780: ``status:pr-open`` must not survive a terminal PR transition.
+
+The leak this file locks down: ``gitea_create_pr`` applied ``status:pr-open``
+and no terminal path ever removed it, so a repository audit found 40 closed
+issues still advertising an open PR that had long since merged or closed.
+
+Coverage mirrors the issue's acceptance criteria: merge, close-without-merge,
+supersession, already-landed reconciliation, controller closure, retry /
+idempotency, unrelated-label preservation, the only-label (empty set) case,
+and terminal validation of any residual label.
+"""
+from __future__ import annotations
+
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import mcp_server
+import terminal_pr_label_cleanup as tplc
+
+
+FAKE_AUTH = "token test-token"
+PR_OPEN = tplc.PR_OPEN_LABEL
+
+
+def _lb(name: str, lid: int) -> dict:
+    return {"id": lid, "name": name, "color": "000000"}
+
+
+# ---------------------------------------------------------------------------
+# Pure rule: planning
+# ---------------------------------------------------------------------------
+class TestPlanPrOpenCleanup(unittest.TestCase):
+
+    def test_removes_only_the_pr_open_label(self):
+        plan = tplc.plan_pr_open_cleanup(
+            ["type:bug", PR_OPEN, "workflow-hardening"],
+            terminal_reason=tplc.MERGED,
+        )
+        self.assertTrue(plan["cleanup_required"])
+        self.assertEqual(plan["removed"], [PR_OPEN])
+        self.assertEqual(plan["labels_after"], ["type:bug", "workflow-hardening"])
+
+    def test_preserves_unrelated_labels_in_original_order(self):
+        labels = ["workflow-hardening", "type:bug", PR_OPEN, "role:author", "leases"]
+        plan = tplc.plan_pr_open_cleanup(labels, terminal_reason=tplc.MERGED)
+        self.assertEqual(
+            plan["labels_after"],
+            ["workflow-hardening", "type:bug", "role:author", "leases"],
+        )
+        self.assertNotIn(PR_OPEN, plan["labels_after"])
+
+    def test_only_label_yields_empty_set(self):
+        plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=tplc.MERGED)
+        self.assertTrue(plan["cleanup_required"])
+        self.assertEqual(plan["labels_after"], [])
+        self.assertTrue(plan["empty_label_set"])
+
+    def test_absent_label_is_an_idempotent_noop(self):
+        plan = tplc.plan_pr_open_cleanup(
+            ["type:bug", "status:done"], terminal_reason=tplc.RETRY_RECOVERY
+        )
+        self.assertFalse(plan["cleanup_required"])
+        self.assertTrue(plan["idempotent_noop"])
+        self.assertEqual(plan["labels_after"], ["type:bug", "status:done"])
+
+    def test_accepts_gitea_label_objects(self):
+        plan = tplc.plan_pr_open_cleanup(
+            {"labels": [{"name": PR_OPEN}, {"name": "type:bug"}]},
+            terminal_reason=tplc.SUPERSEDED,
+        )
+        self.assertEqual(plan["labels_after"], ["type:bug"])
+
+    def test_every_terminal_reason_is_planable(self):
+        for reason in tplc.TERMINAL_REASONS:
+            plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=reason)
+            self.assertEqual(plan["terminal_reason"], reason)
+            self.assertTrue(plan["terminal_reason_description"])
+
+    def test_unknown_terminal_reason_fails_closed(self):
+        with self.assertRaises(ValueError):
+            tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason="whenever")
+
+    def test_reason_aliases_normalize(self):
+        self.assertEqual(tplc.canonical_terminal_reason("merge"), tplc.MERGED)
+        self.assertEqual(
+            tplc.canonical_terminal_reason("already-landed"), tplc.ALREADY_LANDED
+        )
+        self.assertEqual(
+            tplc.canonical_terminal_reason("controller-closure"),
+            tplc.CONTROLLER_CLOSURE,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Pure rule: read-after-write verification
+# ---------------------------------------------------------------------------
+class TestVerifyPrOpenCleanup(unittest.TestCase):
+
+    def test_verified_when_observed_matches_plan(self):
+        plan = tplc.plan_pr_open_cleanup(
+            ["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
+        )
+        result = tplc.verify_pr_open_cleanup(["type:bug"], plan=plan)
+        self.assertTrue(result["verified"])
+        self.assertFalse(result["residual"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_residual_label_is_reported(self):
+        plan = tplc.plan_pr_open_cleanup(
+            ["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
+        )
+        result = tplc.verify_pr_open_cleanup(["type:bug", PR_OPEN], plan=plan)
+        self.assertFalse(result["verified"])
+        self.assertTrue(result["residual"])
+        self.assertIn(PR_OPEN, result["reasons"][0])
+        self.assertTrue(result["safe_next_action"])
+
+    def test_dropped_unrelated_label_is_reported(self):
+        plan = tplc.plan_pr_open_cleanup(
+            ["type:bug", "leases", PR_OPEN], terminal_reason=tplc.MERGED
+        )
+        result = tplc.verify_pr_open_cleanup(["type:bug"], plan=plan)
+        self.assertFalse(result["verified"])
+        self.assertEqual(result["unexpected_removals"], ["leases"])
+
+    def test_unexpected_added_label_is_reported(self):
+        plan = tplc.plan_pr_open_cleanup(
+            ["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
+        )
+        result = tplc.verify_pr_open_cleanup(["type:bug", "surprise"], plan=plan)
+        self.assertFalse(result["verified"])
+        self.assertEqual(result["unexpected_additions"], ["surprise"])
+
+    def test_empty_observed_set_verifies_for_only_label_case(self):
+        plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=tplc.MERGED)
+        result = tplc.verify_pr_open_cleanup([], plan=plan)
+        self.assertTrue(result["verified"])
+        self.assertTrue(result["empty_label_set"])
+
+
+# ---------------------------------------------------------------------------
+# Terminal validation
+# ---------------------------------------------------------------------------
+class TestDetectResidualPrOpen(unittest.TestCase):
+
+    def test_clean_repository(self):
+        issues = [
+            {"number": 1, "state": "closed", "labels": [{"name": "type:bug"}]},
+            {"number": 2, "state": "open", "labels": []},
+        ]
+        result = tplc.detect_residual_pr_open(issues)
+        self.assertTrue(result["clean"])
+        self.assertEqual(result["residual_count"], 0)
+        self.assertEqual(result["checked_count"], 2)
+
+    def test_reports_each_stale_issue(self):
+        issues = [
+            {"number": 626, "state": "closed", "labels": [{"name": PR_OPEN}]},
+            {"number": 772, "state": "closed", "labels": [{"name": PR_OPEN}]},
+            {"number": 9, "state": "open", "labels": [{"name": "type:bug"}]},
+        ]
+        result = tplc.detect_residual_pr_open(issues)
+        self.assertFalse(result["clean"])
+        self.assertEqual(result["residual_count"], 2)
+        self.assertEqual(
+            [entry["number"] for entry in result["residual_issues"]], [626, 772]
+        )
+        self.assertTrue(result["safe_next_action"])
+
+    def test_issue_with_a_live_open_pr_is_not_residual(self):
+        issues = [{"number": 42, "state": "open", "labels": [{"name": PR_OPEN}]}]
+        result = tplc.detect_residual_pr_open(issues, open_pr_issue_numbers=[42])
+        self.assertTrue(result["clean"])
+        self.assertEqual(result["exempt_open_pr_issues"], [42])
+
+
+# ---------------------------------------------------------------------------
+# Executor: one authoritative rule, with read-after-write proof
+# ---------------------------------------------------------------------------
+class _ExecutorHarness(unittest.TestCase):
+    """Drives mcp_server.clear_pr_open_label against a fake Gitea."""
+
+    def setUp(self):
+        self.issue_labels: dict[int, list[str]] = {}
+        self.repo_labels = {
+            PR_OPEN: 4,
+            "type:bug": 1,
+            "workflow-hardening": 2,
+            "leases": 3,
+            "status:done": 5,
+        }
+        self.puts: list[tuple[int, list[int]]] = []
+
+        patch("mcp_server._resolve", return_value=("h", "o", "r")).start()
+        patch("mcp_server._auth", return_value=FAKE_AUTH).start()
+        patch(
+            "mcp_server.repo_api_url",
+            return_value="https://gitea.example/api/v1/repos/o/r",
+        ).start()
+        patch("gitea_audit.audit_enabled", return_value=False).start()
+        patch("mcp_server.api_request", side_effect=self._api).start()
+        # api_get_all resolves api_request inside gitea_auth, so patching the
+        # mcp_server binding alone would let the label inventory hit the network.
+        patch("mcp_server.api_get_all", side_effect=self._api_get_all).start()
+        self.addCleanup(patch.stopall)
+
+    def _api_get_all(self, url, auth, **_kwargs):
+        if "/labels" in url:
+            return [_lb(name, lid) for name, lid in self.repo_labels.items()]
+        raise AssertionError(f"unexpected paginated GET: {url}")
+
+    def _api(self, method, url, auth, payload=None):
+        if method == "GET" and "/issues/" in url:
+            num = int(url.rsplit("/issues/", 1)[1].split("?")[0])
+            return {
+                "number": num,
+                "labels": [
+                    {"name": n, "id": self.repo_labels[n]}
+                    for n in self.issue_labels.get(num, [])
+                ],
+            }
+        if method == "PUT" and url.endswith("/labels"):
+            num = int(url.rsplit("/issues/", 1)[1].split("/")[0])
+            ids = payload["labels"]
+            by_id = {lid: name for name, lid in self.repo_labels.items()}
+            names = [by_id[i] for i in ids]
+            self.puts.append((num, ids))
+            self.issue_labels[num] = names
+            return [_lb(n, self.repo_labels[n]) for n in names]
+        raise AssertionError(f"unexpected API call: {method} {url}")
+
+    def _clear(self, numbers, reason=tplc.MERGED):
+        return mcp_server.clear_pr_open_label(
+            numbers, "prgs", None, None, None, terminal_reason=reason
+        )
+
+
+class TestClearPrOpenLabel(_ExecutorHarness):
+
+    def test_removes_label_and_preserves_the_rest(self):
+        self.issue_labels[780] = ["type:bug", PR_OPEN, "workflow-hardening"]
+        summary = self._clear([780])
+        self.assertTrue(summary["clean"])
+        self.assertEqual(summary["removed"], [780])
+        self.assertEqual(
+            self.issue_labels[780], ["type:bug", "workflow-hardening"]
+        )
+
+    def test_only_label_results_in_empty_set(self):
+        self.issue_labels[626] = [PR_OPEN]
+        summary = self._clear([626])
+        self.assertTrue(summary["clean"])
+        self.assertEqual(self.issue_labels[626], [])
+        self.assertEqual(self.puts, [(626, [])])
+        self.assertTrue(summary["results"][0]["empty_label_set"])
+
+    def test_read_after_write_proof_is_returned(self):
+        self.issue_labels[780] = ["type:bug", PR_OPEN]
+        summary = self._clear([780])
+        entry = summary["results"][0]
+        self.assertTrue(entry["verified"])
+        self.assertEqual(entry["labels_before"], ["type:bug", PR_OPEN])
+        self.assertEqual(entry["labels_after"], ["type:bug"])
+        self.assertEqual(entry["verification"]["observed_labels"], ["type:bug"])
+
+    def test_repeated_cleanup_is_harmless(self):
+        self.issue_labels[780] = ["type:bug", PR_OPEN]
+        first = self._clear([780])
+        second = self._clear([780], reason=tplc.RETRY_RECOVERY)
+        third = self._clear([780], reason=tplc.RETRY_RECOVERY)
+        self.assertTrue(first["clean"] and second["clean"] and third["clean"])
+        self.assertEqual(second["already_absent"], [780])
+        self.assertEqual(third["already_absent"], [780])
+        # Exactly one mutation across three calls.
+        self.assertEqual(len(self.puts), 1)
+        self.assertEqual(self.issue_labels[780], ["type:bug"])
+
+    def test_noop_path_never_reads_the_label_inventory(self):
+        self.issue_labels[780] = ["type:bug"]
+        with patch("mcp_server._repo_label_id_map") as mock_map:
+            summary = self._clear([780])
+        self.assertTrue(summary["clean"])
+        mock_map.assert_not_called()
+
+    def test_duplicate_issue_numbers_are_collapsed(self):
+        self.issue_labels[780] = ["type:bug", PR_OPEN]
+        summary = self._clear([780, 780, "780"])
+        self.assertEqual(summary["checked"], [780])
+        self.assertEqual(len(self.puts), 1)
+
+    def test_no_issue_numbers_is_a_clean_noop(self):
+        summary = self._clear([])
+        self.assertTrue(summary["clean"])
+        self.assertEqual(summary["checked"], [])
+
+    def test_failed_mutation_is_reported_not_swallowed(self):
+        self.issue_labels[780] = ["type:bug", PR_OPEN]
+
+        def boom(*_a, **_kw):
+            raise RuntimeError("gitea exploded")
+
+        with patch("mcp_server._put_issue_label_names", side_effect=boom):
+            summary = self._clear([780])
+        self.assertFalse(summary["clean"])
+        self.assertEqual(summary["failed"], [780])
+        self.assertTrue(summary["safe_next_action"])
+        self.assertIn(PR_OPEN, self.issue_labels[780])
+
+    def test_residual_label_after_write_fails_verification(self):
+        self.issue_labels[780] = ["type:bug", PR_OPEN]
+        real_api = self._api
+
+        # Simulate a write that reports success but leaves the label behind.
+        def sticky(method, url, auth, payload=None):
+            if method == "PUT" and url.endswith("/labels"):
+                num = int(url.rsplit("/issues/", 1)[1].split("/")[0])
+                self.puts.append((num, payload["labels"]))
+                return [_lb("type:bug", 1), _lb(PR_OPEN, 4)]
+            return real_api(method, url, auth, payload)
+
+        with patch("mcp_server.api_request", side_effect=sticky):
+            summary = self._clear([780])
+        self.assertFalse(summary["clean"])
+        self.assertEqual(summary["failed"], [780])
+
+
+# ---------------------------------------------------------------------------
+# Terminal workflow paths
+# ---------------------------------------------------------------------------
+class TestTerminalPathsUseTheSharedRule(_ExecutorHarness):
+    """Merge, close-without-merge, supersession and already-landed."""
+
+    def test_merge_path_clears_the_label_for_linked_issues(self):
+        self.issue_labels[780] = ["type:bug", PR_OPEN]
+        merged_pr = {
+            "title": "fix: terminal label cleanup",
+            "body": "Closes #780",
+            "head": {"ref": "fix/issue-780-terminal-pr-open-label-cleanup"},
+        }
+        with patch(
+            "mcp_server.release_in_progress_label", return_value={780: "released"}
+        ):
+            result = mcp_server.cleanup_in_progress_for_pr(
+                merged_pr, "prgs", None, None, None, terminal_reason=tplc.MERGED
+            )
+        cleanup = result["pr_open_label_cleanup"]
+        self.assertTrue(cleanup["clean"])
+        self.assertEqual(cleanup["terminal_reason"], tplc.MERGED)
+        self.assertEqual(self.issue_labels[780], ["type:bug"])
+
+    def test_close_without_merge_clears_the_label(self):
+        self.issue_labels[781] = ["type:bug", PR_OPEN, "leases"]
+        closed_pr = {
+            "title": "chore: abandoned",
+            "body": "Closes #781",
+            "head": {"ref": "chore/issue-781-abandoned"},
+        }
+        with patch(
+            "mcp_server.release_in_progress_label", return_value={781: "released"}
+        ):
+            result = mcp_server.cleanup_in_progress_for_pr(
+                closed_pr,
+                "prgs",
+                None,
+                None,
+                None,
+                terminal_reason=tplc.CLOSED_WITHOUT_MERGE,
+            )
+        cleanup = result["pr_open_label_cleanup"]
+        self.assertTrue(cleanup["clean"])
+        self.assertEqual(cleanup["terminal_reason"], tplc.CLOSED_WITHOUT_MERGE)
+        self.assertEqual(self.issue_labels[781], ["type:bug", "leases"])
+
+    def test_pr_without_linked_issue_reports_an_empty_cleanup(self):
+        pr = {"title": "chore: no link", "body": "", "head": {"ref": "chore/none"}}
+        result = mcp_server.cleanup_in_progress_for_pr(
+            pr, "prgs", None, None, None, terminal_reason=tplc.MERGED
+        )
+        self.assertEqual(result["cleanup_status"], "no linked issue found")
+        self.assertTrue(result["pr_open_label_cleanup"]["clean"])
+        self.assertEqual(result["pr_open_label_cleanup"]["checked"], [])
+
+    def test_supersession_reason_is_recorded(self):
+        self.issue_labels[600] = [PR_OPEN, "type:bug"]
+        summary = self._clear([600], reason=tplc.SUPERSEDED)
+        self.assertTrue(summary["clean"])
+        self.assertEqual(summary["terminal_reason"], tplc.SUPERSEDED)
+        self.assertEqual(self.issue_labels[600], ["type:bug"])
+
+    def test_already_landed_reconciliation_reason_is_recorded(self):
+        self.issue_labels[601] = [PR_OPEN]
+        summary = self._clear([601], reason=tplc.ALREADY_LANDED)
+        self.assertTrue(summary["clean"])
+        self.assertEqual(summary["terminal_reason"], tplc.ALREADY_LANDED)
+        self.assertEqual(self.issue_labels[601], [])
+
+    def test_issue_780_regression_stale_label_survived_every_terminal_path(self):
+        """Regression for the observed leak.
+
+        Before the fix each terminal path finished without touching
+        ``status:pr-open``, so the audit found closed issues still carrying it.
+        Every path now routes through the one shared rule and leaves nothing
+        behind — while preserving each issue's other labels.
+        """
+        stale = {
+            626: (["type:bug", PR_OPEN], tplc.CONTROLLER_CLOSURE),
+            772: (["workflow-hardening", PR_OPEN], tplc.MERGED),
+            768: ([PR_OPEN], tplc.CLOSED_WITHOUT_MERGE),
+            758: (["leases", PR_OPEN, "type:bug"], tplc.SUPERSEDED),
+            755: (["status:done", PR_OPEN], tplc.ALREADY_LANDED),
+        }
+        for number, (labels, _reason) in stale.items():
+            self.issue_labels[number] = list(labels)
+
+        for number, (_labels, reason) in stale.items():
+            summary = self._clear([number], reason=reason)
+            self.assertTrue(summary["clean"], msg=f"issue #{number}")
+
+        audit = tplc.detect_residual_pr_open(
+            [
+                {"number": num, "state": "closed", "labels": names}
+                for num, names in self.issue_labels.items()
+            ]
+        )
+        self.assertTrue(audit["clean"])
+        self.assertEqual(audit["residual_count"], 0)
+        # Unrelated labels survived every path.
+        self.assertEqual(self.issue_labels[626], ["type:bug"])
+        self.assertEqual(self.issue_labels[772], ["workflow-hardening"])
+        self.assertEqual(self.issue_labels[768], [])
+        self.assertEqual(self.issue_labels[758], ["leases", "type:bug"])
+        self.assertEqual(self.issue_labels[755], ["status:done"])
+
+
+# ---------------------------------------------------------------------------
+# Controller closure
+# ---------------------------------------------------------------------------
+class TestControllerClosure(unittest.TestCase):
+
+    def test_close_issue_clears_label_before_closing_and_validates(self):
+        calls: list[str] = []
+
+        def fake_clear(numbers, *_a, **kwargs):
+            calls.append(f"clear:{kwargs['terminal_reason']}")
+            return {
+                "label": PR_OPEN,
+                "clean": True,
+                "checked": list(numbers),
+                "removed": list(numbers),
+                "already_absent": [],
+                "failed": [],
+                "results": [],
+                "reasons": [],
+                "safe_next_action": "",
+                "terminal_reason": kwargs["terminal_reason"],
+            }
+
+        def fake_api(method, url, auth, payload=None):
+            if method == "PATCH":
+                calls.append("patch:closed")
+                return {"state": "closed"}
+            return {"labels": [{"name": "type:bug"}]}
+
+        with patch("mcp_server.clear_pr_open_label", side_effect=fake_clear), \
+             patch("mcp_server.api_request", side_effect=fake_api), \
+             patch("mcp_server._profile_permission_block", return_value=None), \
+             patch("mcp_server.verify_preflight_purity", return_value=None), \
+             patch("mcp_server.release_in_progress_label", return_value={}), \
+             patch("mcp_server._resolve", return_value=("h", "o", "r")), \
+             patch("mcp_server._auth", return_value=FAKE_AUTH), \
+             patch("gitea_audit.audit_enabled", return_value=False):
+            result = mcp_server.gitea_close_issue(issue_number=780, remote="prgs")
+
+        self.assertTrue(result["success"])
+        # Cleanup precedes the state change: closing first would bake in the leak.
+        self.assertEqual(calls[0], f"clear:{tplc.CONTROLLER_CLOSURE}")
+        self.assertIn("patch:closed", calls)
+        self.assertTrue(result["terminal_label_validation"]["clean"])
+
+    def test_close_issue_fails_closed_when_cleanup_cannot_complete(self):
+        def fake_clear(numbers, *_a, **kwargs):
+            return {
+                "label": PR_OPEN,
+                "clean": False,
+                "checked": list(numbers),
+                "removed": [],
+                "already_absent": [],
+                "failed": list(numbers),
+                "results": [],
+                "reasons": ["label replacement failed: boom"],
+                "safe_next_action": "retry",
+                "terminal_reason": kwargs["terminal_reason"],
+            }
+
+        def fail_on_patch(method, url, auth, payload=None):
+            if method == "PATCH":
+                raise AssertionError("issue must not be closed when cleanup failed")
+            return {}
+
+        with patch("mcp_server.clear_pr_open_label", side_effect=fake_clear), \
+             patch("mcp_server.api_request", side_effect=fail_on_patch), \
+             patch("mcp_server._profile_permission_block", return_value=None), \
+             patch("mcp_server.verify_preflight_purity", return_value=None), \
+             patch("mcp_server._resolve", return_value=("h", "o", "r")), \
+             patch("mcp_server._auth", return_value=FAKE_AUTH), \
+             patch("gitea_audit.audit_enabled", return_value=False):
+            result = mcp_server.gitea_close_issue(issue_number=780, remote="prgs")
+
+        self.assertFalse(result["success"])
+        self.assertTrue(result["blocked"])
+        self.assertFalse(result["performed"])
+        self.assertIn("#780", result["message"])
+        self.assertTrue(result["safe_next_action"])
+
+
+# ---------------------------------------------------------------------------
+# Capability wiring
+# ---------------------------------------------------------------------------
+class TestCapabilityWiring(unittest.TestCase):
+
+    def test_task_is_registered_with_label_authority(self):
+        import task_capability_map
+
+        self.assertEqual(
+            task_capability_map.required_permission("cleanup_terminal_pr_labels"),
+            "gitea.issue.comment",
+        )
+        self.assertEqual(
+            task_capability_map.required_role("cleanup_terminal_pr_labels"),
+            "author",
+        )
+        self.assertEqual(
+            task_capability_map.tool_required_permission(
+                "gitea_cleanup_terminal_pr_labels"
+            ),
+            "gitea.issue.comment",
+        )
+
+    def test_recovery_tool_rejects_an_unknown_reason_without_mutating(self):
+        with patch("mcp_server._profile_permission_block", return_value=None), \
+             patch("mcp_server.verify_preflight_purity", return_value=None), \
+             patch("mcp_server.clear_pr_open_label") as mock_clear:
+            result = mcp_server.gitea_cleanup_terminal_pr_labels(
+                issue_numbers=[780], terminal_reason="sometime", remote="prgs"
+            )
+        self.assertFalse(result["success"])
+        self.assertFalse(result["clean"])
+        mock_clear.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

workflow_dashboard.py

@@ -14,7 +14,7 @@ Design rules:
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Any, Iterable, Sequence
+from typing import Any, Iterable, Mapping, Sequence
 
 from allocator_service import (
     ROLE_AUTHOR,
@@ -22,7 +22,11 @@ from allocator_service import (
     ROLE_MERGER,
     ROLE_RECONCILER,
     ROLE_REVIEWER,
+    OWNERSHIP_FOREIGN,
+    OWNERSHIP_UNKNOWN,
+    SKIP_CLAIMED_BY_OTHER_SESSION,
     WorkCandidate,
+    classify_claim_ownership,
     classify_skip,
     expected_role_for_candidate,
     sort_candidates,
@@ -265,9 +269,13 @@ def _entry_for_candidate(
     c: WorkCandidate,
     *,
     terminal_pr: int | None,
+    claim_ownership: str | None = None,
 ) -> QueueEntry:
     expected = expected_role_for_candidate(c)
     badges = _badges_for_candidate(c, terminal_pr=terminal_pr)
+    claimed_by_other = claim_ownership in (OWNERSHIP_FOREIGN, OWNERSHIP_UNKNOWN)
+    if claimed_by_other and "claimed" not in badges:
+        badges = tuple(list(badges) + ["claimed"])
     safe_roles: list[str] = []
     block_reason: str | None = None
 
@@ -276,6 +284,12 @@ def _entry_for_candidate(
         block_reason = "status blocked"
     elif c.dependency_unmet:
         block_reason = c.dependency_reason or "unmet dependency"
+    elif claimed_by_other:
+        # #765: never advertise another controller's active task as safe work.
+        block_reason = (
+            f"{SKIP_CLAIMED_BY_OTHER_SESSION}: active lease held by another "
+            "controller"
+        )
     elif c.already_claimed_elsewhere:
         block_reason = "already claimed elsewhere"
     elif c.kind == "pr" and not (c.head_sha or "").strip():
@@ -291,7 +305,12 @@ def _entry_for_candidate(
 
     if block_reason is None:
         # Safe only for the expected role, and only when classify_skip agrees.
-        skip = classify_skip(c, role=expected, terminal_pr=terminal_pr)
+        skip = classify_skip(
+            c,
+            role=expected,
+            terminal_pr=terminal_pr,
+            claim_ownership=claim_ownership,
+        )
         if skip is None:
             safe_roles.append(expected)
         else:
@@ -450,8 +469,16 @@ def build_workflow_dashboard(
     terminal_lock: dict[str, Any] | None = None,
     inventory_complete: bool = True,
     inventory_reasons: Sequence[str] | None = None,
+    claims: Mapping[tuple[str, int], dict[str, Any]] | None = None,
+    session_id: str | None = None,
+    controller_instance_id: str | None = None,
 ) -> DashboardSnapshot:
-    """Build a full dashboard snapshot from injected inventory (pure)."""
+    """Build a full dashboard snapshot from injected inventory (pure).
+
+    *claims* (#765) maps ``(kind, number)`` to the live lease holding that work
+    item. Items claimed by a different controller are never presented as safe
+    next work for this one.
+    """
     reasons = [str(r) for r in (inventory_reasons or ()) if str(r).strip()]
     if not inventory_complete:
         reasons.append(
@@ -461,7 +488,16 @@ def build_workflow_dashboard(
 
     ranked = sort_candidates(list(candidates))
     entries = [
-        _entry_for_candidate(c, terminal_pr=terminal_pr) for c in ranked
+        _entry_for_candidate(
+            c,
+            terminal_pr=terminal_pr,
+            claim_ownership=classify_claim_ownership(
+                (claims or {}).get((c.kind, int(c.number))),
+                session_id=session_id,
+                controller_instance_id=controller_instance_id,
+            ),
+        )
+        for c in ranked
     ]
 
     open_prs = [e for e in entries if e.kind == "pr"]