fix: exclude foreign-claimed work from allocation instead of blockading the queue (Closes #765)

Recovered preserved candidate d06198b onto current master 0c2f45a via clean
cherry-pick. Active foreign leases are excluded before allocator ranking;
controller_instance_id ownership is persisted; dashboard matches allocator
exclusion. Stable patch-id bacafc5f… preserved.

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

allocator_service.py

@@ -21,7 +21,7 @@ from __future__ import annotations
 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 +40,16 @@ 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"
+
+# #765 skip reason code for work already claimed by a different controller.
+SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session"
+
+# 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 +154,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 +280,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 +299,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():
@@ -287,6 +382,8 @@ 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,
 ) -> dict[str, Any]:
     """Select and optionally reserve the next work unit via control-plane DB.
 
@@ -328,6 +425,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,25 +467,99 @@ 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.
+    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",
+            }
+
     skipped: list[SkipRecord] = []
+    claims_excluded: list[dict[str, Any]] = []
+    ownership_defects: list[dict[str, Any]] = []
     ordered = sort_candidates(list(candidates))
     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")
         else:
             outcome = OUTCOME_NO_SAFE
             reasons = [
@@ -414,6 +586,10 @@ 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),
+            "owner_session_id": owner_session_id,
             "downstream_note": (
                 "#612 incident bridge remains downstream of #600; "
                 "allocator never assigns raw monitoring incidents"
@@ -460,6 +636,9 @@ 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),
             "downstream_note": (
                 "#612 incident bridge remains downstream of #600; "
                 "allocator never assigns raw monitoring incidents"
@@ -583,6 +762,9 @@ 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),
         "downstream_note": (
             "#612 incident bridge remains downstream of #600; "
             "allocator never assigns raw monitoring incidents"

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

gitea_mcp_server.py

@@ -19211,6 +19211,7 @@ def gitea_allocate_next_work(
         apply=bool(apply),
         profile_name=profile_name,
         username=username,
+        controller_instance_id=allocator_service.resolve_controller_instance_id(),
     )
     if inv_reasons:
         result.setdefault("inventory_warnings", inv_reasons)
@@ -19378,6 +19379,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"])
@@ -19416,6 +19418,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,
@@ -19427,6 +19441,8 @@ 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"] = (

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

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