Merge pull request 'fix(allocator): pre-rank exclusions and candidates_json transport (Closes #776)' (#777) from fix/issue-776-allocator-pre-rank-exclusions into master

Expose exclude_issue_numbers on gitea_allocate_next_work, remove excluded
numbers before ranking, normalize decoded-list and JSON-string
candidates_json fail-closed, and return candidate-set fingerprints for
dry-run/apply CAS. Same-owner leases on excluded issues surface a
structured resume/release blocker.

Closes #776

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

final_report_validator.py

@@ -728,6 +728,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,
@@ -1584,6 +1643,10 @@ _SHARED_CANONICAL_COMMENT_RULES = (
     _rule_shared_canonical_comment_post_claim,
 )
 
+_SHARED_MUTATION_BUDGET_RULES = (
+    _rule_shared_mutation_budget_accounting,
+)
+
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
         _rule_shared_controller_handoff,
@@ -1591,6 +1654,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _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,
@@ -1636,6 +1700,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _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,
@@ -1655,6 +1720,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _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,
     ],
@@ -1664,6 +1730,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _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,
@@ -1677,6 +1744,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_shared_email_disclosure,
         *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_CANONICAL_COMMENT_RULES,
+        *_SHARED_MUTATION_BUDGET_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
     "inventory": [
@@ -1685,6 +1753,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_shared_email_disclosure,
         *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_CANONICAL_COMMENT_RULES,
+        *_SHARED_MUTATION_BUDGET_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_pagination_proof,
     ],
@@ -1694,6 +1763,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _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
@@ -1766,6 +1836,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 +1900,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

@@ -19202,7 +19202,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 +19218,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 +19283,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 +19331,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 +19458,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 +19467,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 +19495,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 +19523,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 +19562,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 +19585,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"},
+    }

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_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))

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"]