diff --git a/CHANGELOG.md b/CHANGELOG.md
index a85459eb7..6ba949b3d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -270,7 +270,11 @@ All notable changes to Raven are documented here.
   failed remote run whose announce the draining scheduler refused used to
   vanish; its record now stays, like a reply's, for the next start to report.
   A cli run picked up after a restart now binds its instance to the session
-  id in its output, so the next turn resumes it. A cancel while the machine's
+  id in its output, so the next turn resumes it; a stateless run binds
+  nothing, and a binding a dispatch made while the pick-up waited is kept,
+  for a minted id as for one read off the output, without the pick-up
+  waiting for that turn, and the reply picked up finishes that dispatch's
+  own record, not a newer turn's to the same instance. A cancel while the machine's
   link is down no longer waits out the settle budget. An openai row's tunnel
   gives a local port that was taken in between up for another instead of
   sending the request to whoever took it. A run record that is not JSON, or not
diff --git a/raven/agent/subagent/backends/cli_agent.py b/raven/agent/subagent/backends/cli_agent.py
index fdf0f52a0..04ebb7f29 100644
--- a/raven/agent/subagent/backends/cli_agent.py
+++ b/raven/agent/subagent/backends/cli_agent.py
@@ -1384,9 +1384,26 @@ async def _bind_picked_up(self, run: Any, jsonl_id: str | None, stdout: str, std
         the transcript's id, else the pattern over stdout then stderr -- so
         the next dispatch to the handle resumes this session instead of
         opening a fresh one. Nothing for a minted id (bound at the launch)
-        or a stateless run.
+        or a stateless run -- the record's ``stateful`` says which, as the
+        handle cannot: every run carries one (``instance`` or ``task_id``),
+        and a stateless entry may declare ``idSource: derived`` too, which
+        the dispatch path then reads nothing from and binds nothing with.
+
+        Bound only where nothing is bound yet (``InstanceRegistry.bind_if_absent``,
+        one step under the registry's own lock): a dispatch to the same named
+        instance while this pick-up waited on the machine's command looked
+        the handle up, found nothing, and bound the session it created; that
+        binding is the newer turn's thread and is kept, this one's session
+        left unbound, where committing over it would have moved the instance
+        back to a session the newer turn never saw. Not under ``hold_handle``:
+        the dispatch holds that across its whole turn, with no timeout, and
+        the pick-ups run one after another -- a pick-up waiting for it would
+        hold back every run picked up after it. The dispatch commits last
+        whichever way the two land, so the handle ends on its session either
+        way, and the pick-up never waits for that turn. The minted id of a
+        ``provisioned`` entry is bound the same way, by ``resume_remote._conclude``.
         """
-        if run.session_id or not run.handle or self.id_source != "derived":
+        if run.session_id or not run.stateful or self.id_source != "derived":
             return ""
         agent_id = jsonl_id
         if agent_id is None and self._session_id_re is not None:
@@ -1395,11 +1412,13 @@ async def _bind_picked_up(self, run: Any, jsonl_id: str | None, stdout: str, std
                     agent_id = m.group(1)
                     break
         if agent_id is None:
+            if await self._registry.lookup(run.session_key, self.name, run.handle) is not None:
+                return ""  # bound meanwhile by a newer turn: resumable through that
             return (
                 "\n\n[raven] Warning: no session id could be extracted from the create "
                 "output, so this instance is not resumable. Use a new handle to recreate."
             )
-        await self._registry.commit(run.session_key, self.name, run.handle, agent_id)
+        await self._registry.bind_if_absent(run.session_key, self.name, run.handle, agent_id)
         return ""
 
     async def _attempt(
diff --git a/raven/agent/subagent/instances.py b/raven/agent/subagent/instances.py
index d2cff85a3..3ff77bf5d 100644
--- a/raven/agent/subagent/instances.py
+++ b/raven/agent/subagent/instances.py
@@ -149,29 +149,63 @@ async def lookup(self, session_key: str, agent: str, handle: str, *, kind: str =
         return rec.get("agentId")
 
     async def commit(self, session_key: str, agent: str, handle: str, agent_id: str, *, kind: str = "cli") -> None:
+        async with self._lock:
+            with write_transaction(self._path):
+                self._bind(self._load(), (session_key, agent, handle), agent_id, kind)
+
+    async def bind_if_absent(
+        self, session_key: str, agent: str, handle: str, agent_id: str, *, kind: str = "cli"
+    ) -> str:
+        """Bind ``handle`` to ``agent_id`` unless it is bound to another id already; the id it is bound to after this.
+
+        One step under the registry's own lock, for a writer that must neither
+        replace a binding made meanwhile nor wait for the handle's lock: the
+        pick-up of a run a stopped host left on a machine
+        (``CliAgentBackend._bind_picked_up``, ``resume_remote._conclude``).
+        A dispatch to the same named instance holds ``hold_handle`` across
+        its whole turn and commits last; whichever order the two land in,
+        the handle ends bound to the dispatch's session, and the pick-up
+        never waits for that turn. A row of another ``kind``, or a status row
+        with no id, counts as absent, as :meth:`lookup` reads it. A row
+        bound to ``agent_id`` itself is rebuilt as a binding alone, as
+        :meth:`commit` rebuilds one: a left turn that resumed the session
+        its handle is bound to owns that row, and the rebuild closes the
+        ``running`` the manager put on it when the turn began -- or the
+        instance reads as interrupted after a reply that was announced.
+        """
         async with self._lock:
             with write_transaction(self._path):
                 records = self._load()
                 key = (session_key, agent, handle)
-                now = int(time.time() * 1000)
-                existing = records.get(key) or {}
-                records[key] = {
-                    "kind": kind,
-                    "sessionKey": session_key,
-                    "agent": agent,
-                    "handle": handle,
-                    "agentId": agent_id,
-                    "createdAtMs": existing.get("createdAtMs", now),
-                    "updatedAtMs": now,
-                    **_dag_origin(existing),
-                }
-                try:
-                    self._flush()
-                except OSError as e:  # noqa: BLE001 - persistence is best-effort
-                    logger.warning(
-                        "subagent instance registry write failed (mapping live in-process, will not survive restart): {}",
-                        e,
-                    )
+                rec = records.get(key)
+                if rec and rec.get("kind") == kind and rec.get("agentId") and rec["agentId"] != agent_id:
+                    return str(rec["agentId"])
+                # Unbound, or bound to this very id: this run's own row either way.
+                self._bind(records, key, agent_id, kind)
+                return agent_id
+
+    def _bind(self, records: dict[_Key, dict[str, Any]], key: _Key, agent_id: str, kind: str) -> None:
+        """Write ``key``'s binding into ``records`` and to disk; the caller holds the lock and the transaction."""
+        session_key, agent, handle = key
+        now = int(time.time() * 1000)
+        existing = records.get(key) or {}
+        records[key] = {
+            "kind": kind,
+            "sessionKey": session_key,
+            "agent": agent,
+            "handle": handle,
+            "agentId": agent_id,
+            "createdAtMs": existing.get("createdAtMs", now),
+            "updatedAtMs": now,
+            **_dag_origin(existing),
+        }
+        try:
+            self._flush()
+        except OSError as e:  # noqa: BLE001 - persistence is best-effort
+            logger.warning(
+                "subagent instance registry write failed (mapping live in-process, will not survive restart): {}",
+                e,
+            )
 
     async def upsert_spawn(
         self, session_key: str, agent: str, handle: str, status: str, agent_id: str | None = None
diff --git a/raven/agent/subagent/manager.py b/raven/agent/subagent/manager.py
index a5f2490f1..fd5a60229 100644
--- a/raven/agent/subagent/manager.py
+++ b/raven/agent/subagent/manager.py
@@ -1805,9 +1805,10 @@ async def _run_subagent_inner(
         # the same question.
         cancelled = False
         left = False
-        # Whether the reply reached the conversation: only a refused announce
-        # of a run that returned says no, and the run's record on its machine
-        # then stays for the next start to report.
+        # Whether the outcome reached the conversation: only an announce the
+        # draining scheduler refused says no -- of a reply, or of a failure
+        # -- and the run's record on its machine then stays for the next
+        # start to report.
         announced = True
         # The record's own id, not its directory's name: the artifacts are a
         # filename prefix in the shared node root now, so the directory names
@@ -1976,7 +1977,8 @@ async def _run_subagent_inner(
                 if not left:
                     await self._finalize_node_claim(session_key, record)
                     # The run's record on its machine, when the backend left
-                    # one with the reply: gone once the announce was taken. An
+                    # one -- with the reply, or with the failure of a prompt
+                    # that went out: gone once the announce was taken. An
                     # announce the draining scheduler refused leaves it, with
                     # what the announce after a restart needs, for the next
                     # start to report (raven.agent.subagent.resume_remote).
diff --git a/raven/agent/subagent/resume_remote.py b/raven/agent/subagent/resume_remote.py
index b513ab8ed..a2a523ecc 100644
--- a/raven/agent/subagent/resume_remote.py
+++ b/raven/agent/subagent/resume_remote.py
@@ -161,7 +161,11 @@ async def _conclude(manager: Any, run: remote_runs.RemoteRun, status: str, reply
         await manager._finalize_node_claim(run.session_key, record)
     try:
         if ok and run.stateful and run.session_id:
-            await get_registry().commit(run.session_key, run.agent, run.handle, run.session_id, kind=run.kind)
+            # Where nothing is bound yet: a dispatch to the same named
+            # instance that ran while the pick-up waited bound its own
+            # session, and that newer binding stays (``bind_if_absent``;
+            # the derived case is ``CliAgentBackend._bind_picked_up``).
+            await get_registry().bind_if_absent(run.session_key, run.agent, run.handle, run.session_id, kind=run.kind)
         else:
             await get_registry().upsert_spawn(run.session_key, run.agent, run.handle, "completed" if ok else "failed")
     except Exception:  # noqa: BLE001 - the registry row must not keep the result from its reader
@@ -212,17 +216,26 @@ def _origin_of(
 
 
 def _spawn_record(session_dir: Path, run: remote_runs.RemoteRun) -> tuple[SpawnRecord | None, dict[str, Any]]:
-    """The dispatch's record under the session, still open: the latest of this agent and handle left running."""
+    """The dispatch's record under the session, still open.
+
+    The one the manager named when it left the run -- ``node_id`` in the
+    run's origin (:func:`raven.acp_client.remote_runs.annotate`): a newer
+    turn to the same named instance opens its own record, ``running`` and
+    under the handle, as its dispatch starts, and that record is the newer
+    turn's to finish. A run a host killed outright carries no origin; for it
+    the latest of this agent and handle left running is the best guess.
+    """
     root = nodes_root(session_dir)
+    origin = run.origin if isinstance(run.origin, dict) else {}
+    named = str(origin.get("node_id") or "")
+    if named:
+        meta = _open_meta(root / f"{named}.meta.json", run)
+        if meta is not None:
+            return _record(root, session_dir, named), meta
     best: tuple[int, str, dict[str, Any]] | None = None
     for path in root.glob("*.meta.json") if root.is_dir() else []:
-        try:
-            meta = json.loads(path.read_text(encoding="utf-8"))
-        except (OSError, ValueError):
-            continue
-        if not isinstance(meta, dict) or meta.get("status") != "running":
-            continue
-        if meta.get("agent") != run.agent or str(meta.get("handle") or "") != run.handle:
+        meta = _open_meta(path, run)
+        if meta is None:
             continue
         started = int(meta.get("started_at_ms") or 0)
         if best is None or started > best[0]:
@@ -230,12 +243,29 @@ def _spawn_record(session_dir: Path, run: remote_runs.RemoteRun) -> tuple[SpawnR
     if best is None:
         return None, {}
     _, node_id, meta = best
+    return _record(root, session_dir, node_id), meta
+
+
+def _open_meta(path: Path, run: remote_runs.RemoteRun) -> dict[str, Any] | None:
+    """The record at ``path``, when it is this agent and handle's and still running; else ``None``."""
+    try:
+        meta = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+    if not isinstance(meta, dict) or meta.get("status") != "running":
+        return None
+    if meta.get("agent") != run.agent or str(meta.get("handle") or "") != run.handle:
+        return None
+    return meta
+
+
+def _record(root: Path, session_dir: Path, node_id: str) -> SpawnRecord:
     task = ""
     try:
         task = (root / f"{node_id}.prompt.md").read_text(encoding="utf-8")
     except OSError:
         pass
-    return SpawnRecord(root, session_dir=session_dir, task=task, node_id=node_id), meta
+    return SpawnRecord(root, session_dir=session_dir, task=task, node_id=node_id)
 
 
 __all__ = ["START_DELAY_S", "resume_remote_runs", "schedule_remote_resume"]
diff --git a/tests/test_resume_remote.py b/tests/test_resume_remote.py
index 1cf94d9cf..5ea50b64f 100644
--- a/tests/test_resume_remote.py
+++ b/tests/test_resume_remote.py
@@ -4,6 +4,7 @@
 
 import asyncio
 import json
+import time
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
@@ -37,12 +38,20 @@ def logs():
 
 
 class _Registry:
-    def __init__(self) -> None:
+    def __init__(self, bound: str | None = None) -> None:
+        self.bound = bound
         self.committed: list[tuple[str, str, str, str, str]] = []
         self.statuses: list[tuple[str, str, str, str]] = []
 
-    async def commit(self, session_key, agent, handle, agent_id, *, kind="cli"):
+    async def bind_if_absent(self, session_key, agent, handle, agent_id, *, kind="cli"):
+        if self.bound is not None:
+            return self.bound
         self.committed.append((session_key, agent, handle, agent_id, kind))
+        self.bound = agent_id
+        return agent_id
+
+    async def commit(self, session_key, agent, handle, agent_id, *, kind="cli"):
+        raise AssertionError("a pick-up binds where nothing is bound; it never commits over a binding")
 
     async def upsert_spawn(self, session_key, agent, handle, status, agent_id=None):
         self.statuses.append((session_key, agent, handle, status))
@@ -505,3 +514,90 @@ async def pick_up(saved, backend):
     assert manager.announced[0]["status"] == "ok"
     assert manager.announced[0]["result"] == "recovered answer"
     assert remote_runs.pending() == []
+
+
+async def test_a_minted_id_is_bound_only_where_nothing_is_bound_so_a_newer_turn_s_binding_is_kept(
+    tmp_path, monkeypatch
+):
+    """The provisioned twin of the cli pick-up's rule: `_conclude` binds the record's session id where the handle is
+    unbound, and keeps the binding a dispatch made while the pick-up waited on the machine."""
+    kept = _Registry(bound="newer-turn")
+    monkeypatch.setattr(resume_remote, "get_registry", lambda: kept)
+    manager = _Manager(_backend(), tmp_path / "s1")
+    _open_record(tmp_path / "s1")
+
+    assert (
+        await resume_remote._conclude(manager, _run(stateful=True, session_id="minted-X"), "ok", "the answer") is True
+    )
+
+    assert kept.committed == [] and kept.bound == "newer-turn", "the newer binding stays"
+    assert manager.announced[0]["status"] == "ok"
+
+    fresh = _Registry()
+    monkeypatch.setattr(resume_remote, "get_registry", lambda: fresh)
+    _open_record(tmp_path / "s2")
+    assert (
+        await resume_remote._conclude(
+            _Manager(_backend(), tmp_path / "s2"), _run(stateful=True, session_id="minted-X"), "ok", "x"
+        )
+        is True
+    )
+    assert fresh.committed == [("tui:s1", "stub@box", "t1", "minted-X", "acp")], "unbound: the minted id binds"
+
+
+async def test_a_left_turn_that_resumed_the_session_its_handle_is_bound_to_closes_the_instance_s_row(
+    tmp_path, monkeypatch
+):
+    """The manager wrote ``running`` on the row when the turn began, and nothing wrote it since; the pick-up's bind,
+    which finds the handle bound to this very session, is what closes the row -- or the instance reads as interrupted
+    after its reply was announced."""
+    from raven.agent.subagent.instances import InstanceRegistry
+
+    registry = InstanceRegistry(path=tmp_path / "instances.json")
+    monkeypatch.setattr(resume_remote, "get_registry", lambda: registry)
+    await registry.commit("tui:s1", "stub@box", "t1", "X", kind="acp")
+    await registry.upsert_spawn("tui:s1", "stub@box", "t1", "running")
+    manager = _Manager(_backend(), tmp_path / "s1")
+    _open_record(tmp_path / "s1")
+
+    assert await resume_remote._conclude(manager, _run(stateful=True, session_id="X"), "ok", "the answer") is True
+
+    row = registry._load()[("tui:s1", "stub@box", "t1")]
+    assert row["agentId"] == "X" and row["kind"] == "acp" and "status" not in row
+    assert manager.announced[0]["status"] == "ok"
+
+
+async def test_a_pick_up_concludes_the_record_the_run_s_origin_names_not_the_newest_running_one_of_the_handle(
+    tmp_path, monkeypatch
+):
+    """A newer turn to the same named instance opens its own record, ``running`` and under the handle, as its dispatch
+    starts; the left run's reply goes to the record the manager named when it left the run, and the newer turn's stays
+    open for that turn to finish. With no origin -- a host killed outright annotated nothing -- the latest left running
+    is the best guess, as before."""
+    registry = _Registry()
+    monkeypatch.setattr(resume_remote, "get_registry", lambda: registry)
+    session_dir = tmp_path / "s1"
+    _open_record(session_dir, node_id="n-left")
+    time.sleep(0.02)
+    _open_record(session_dir, node_id="n-newer")
+    manager = _Manager(_backend(), session_dir)
+
+    assert (
+        await resume_remote._conclude(manager, _run(origin={"node_id": "n-left"}), "ok", "THE LEFT RUN'S REPLY") is True
+    )
+
+    root = nodes_root(session_dir)
+
+    def meta(node_id: str) -> dict[str, Any]:
+        return json.loads((root / f"{node_id}.meta.json").read_text(encoding="utf-8"))
+
+    assert meta("n-left")["status"] == "completed" and meta("n-newer")["status"] == "running"
+    assert "THE LEFT RUN'S REPLY" in (root / "n-left.out.md").read_text(encoding="utf-8")
+    assert manager.finalized == ["n-left"] and str(manager.announced[0]["record"]).endswith("n-left.out.md")
+
+    _open_record(tmp_path / "s2", node_id="n-old")
+    time.sleep(0.02)
+    _open_record(tmp_path / "s2", node_id="n-new")
+    later = _Manager(_backend(), tmp_path / "s2")
+    assert await resume_remote._conclude(later, _run(), "ok", "x") is True
+    assert later.finalized == ["n-new"], "no origin: the latest left running"
diff --git a/tests/test_subagent_acp_remote.py b/tests/test_subagent_acp_remote.py
index 15062f4db..ab5a66009 100644
--- a/tests/test_subagent_acp_remote.py
+++ b/tests/test_subagent_acp_remote.py
@@ -590,6 +590,10 @@ def __init__(self) -> None:
     async def commit(self, session_key, agent, handle, agent_id, *, kind="cli"):
         self.rows.append(("commit", session_key, agent, handle, agent_id))
 
+    async def bind_if_absent(self, session_key, agent, handle, agent_id, *, kind="cli"):
+        self.rows.append(("bind", session_key, agent, handle, agent_id))
+        return agent_id
+
     async def upsert_spawn(self, session_key, agent, handle, status, agent_id=None):
         self.rows.append(("status", session_key, agent, handle, status))
 
diff --git a/tests/test_subagent_cli_remote.py b/tests/test_subagent_cli_remote.py
index 970659d52..66e71d587 100644
--- a/tests/test_subagent_cli_remote.py
+++ b/tests/test_subagent_cli_remote.py
@@ -1028,37 +1028,175 @@ async def test_a_left_cli_run_is_read_whole_off_the_daemon_and_its_exit_ends_it(
     assert remote_resume.PICKED_UP_NOTE in picked.reply and f"scp://worker@{HOST}:{PORT}//home/worker" in picked.reply
 
 
+def _derived_config(**kw: Any) -> ThirdPartyCliSubagentConfig:
+    """A cli entry whose session id is read off the create's output; stateful unless ``resume_command`` is cleared."""
+    kw.setdefault("resume_command", "sh -c 'cat' {agent_id}")
+    if kw["resume_command"] is None:
+        del kw["resume_command"]
+    return cli_config("sh -c 'cat'", id_source="derived", session_id_pattern=r"session (\S+)", **kw)
+
+
+class _Recorder:
+    """A registry stand-in: what a pick-up bound, looked up, and what it found."""
+
+    def __init__(self, bound: str | None = None) -> None:
+        self.bound = bound
+        self.rows: list[tuple[str, ...]] = []
+        self.lookups = 0
+
+    async def lookup(self, session_key: str, agent: str, handle: str, *, kind: str = "cli") -> str | None:
+        self.lookups += 1
+        return self.bound
+
+    async def bind_if_absent(
+        self, session_key: str, agent: str, handle: str, agent_id: str, *, kind: str = "cli"
+    ) -> str:
+        if self.bound is not None:
+            return self.bound
+        self.rows.append((session_key, agent, handle, agent_id))
+        self.bound = agent_id
+        return agent_id
+
+    async def commit(self, session_key: str, agent: str, handle: str, agent_id: str, *, kind: str = "cli") -> None:
+        raise AssertionError("a pick-up binds where nothing is bound; it never commits over a binding")
+
+
 async def test_a_picked_up_create_binds_the_instance_to_the_session_id_its_output_names(box):
     """The dispatch's deferred commit never ran (this computer stopped first): the pick-up reads the id the way the
     dispatch would have and binds the handle, so the next turn to it resumes this session."""
-    cfg = cli_config(
-        "sh -c 'cat'", id_source="derived", session_id_pattern=r"session (\S+)", resume_command="sh -c 'cat' {agent_id}"
+    backend = build_third_party_backend(_derived_config())
+    registry = _Recorder()
+    backend._registry = registry  # type: ignore[attr-defined]
+
+    picked = await backend.pick_up(
+        _left_run(handle="t1", session_id="", stateful=True),
+        pool=_PickUpLink(_left_endpoint("session abc123", "done", rc=0)),
+    )
+    assert picked.status == "ok" and registry.rows == [("tui:s1", "where@box", "t1", "abc123")]
+    assert "not resumable" not in picked.reply
+
+    registry = _Recorder()
+    backend._registry = registry  # type: ignore[attr-defined]
+    picked = await backend.pick_up(
+        _left_run(handle="t1", session_id="", stateful=True), pool=_PickUpLink(_left_endpoint("done", rc=0))
+    )
+    assert picked.status == "ok" and registry.rows == [] and "not resumable" in picked.reply, (
+        "no id in the output: said so"
     )
-    backend = build_third_party_backend(cfg)
-    rows: list[tuple[str, ...]] = []
 
-    class _Recorder:
-        async def commit(self, session_key: str, agent: str, handle: str, agent_id: str, *, kind: str = "cli") -> None:
-            rows.append((session_key, agent, handle, agent_id))
+    picked = await backend.pick_up(
+        _left_run(handle="t1", session_id="bound-before", stateful=True),
+        pool=_PickUpLink(_left_endpoint("session zzz", "done", rc=0)),
+    )
+    assert picked.status == "ok" and registry.rows == [], "a run that resumed a bound session binds nothing"
+
 
-    backend._registry = _Recorder()  # type: ignore[attr-defined]
+async def test_a_picked_up_stateless_run_binds_nothing_and_warns_of_nothing_as_the_dispatch_path_does(box):
+    """A stateless entry may declare ``idSource: derived``; the dispatch reads no id from it and binds nothing, and
+    the record's ``stateful`` is what tells the pick-up so -- the handle cannot, every run carries one."""
+    backend = build_third_party_backend(_derived_config(resume_command=None))
+    assert backend.is_stateful is False
+    registry = _Recorder()
+    backend._registry = registry  # type: ignore[attr-defined]
+
+    with_id = await backend.pick_up(
+        _left_run(handle="t1", session_id="", stateful=False),
+        pool=_PickUpLink(_left_endpoint("session abc123", "done", rc=0)),
+    )
+    without = await backend.pick_up(
+        _left_run(handle="t1", session_id="", stateful=False), pool=_PickUpLink(_left_endpoint("done", rc=0))
+    )
+
+    assert with_id.status == "ok" and without.status == "ok"
+    assert registry.rows == [] and registry.lookups == 0, "nothing bound, nothing looked up"
+    assert "not resumable" not in with_id.reply and "not resumable" not in without.reply
+
+
+async def test_a_binding_a_dispatch_made_while_the_pick_up_waited_is_kept(box):
+    """A session the newer turn bound while this pick-up waited on the machine is the instance's thread now, and the
+    picked-up create's id does not replace it; with no id in the output there is no warning either."""
+    backend = build_third_party_backend(_derived_config())
+    registry = _Recorder(bound="newer-turn")
+    backend._registry = registry  # type: ignore[attr-defined]
 
     picked = await backend.pick_up(
-        _left_run(handle="t1", session_id=""), pool=_PickUpLink(_left_endpoint("session abc123", "done", rc=0))
+        _left_run(handle="t1", session_id="", stateful=True),
+        pool=_PickUpLink(_left_endpoint("session first", "done", rc=0)),
     )
-    assert picked.status == "ok" and rows == [("tui:s1", "where@box", "t1", "abc123")]
+    assert picked.status == "ok" and registry.rows == [] and registry.bound == "newer-turn"
     assert "not resumable" not in picked.reply
 
-    rows.clear()
     picked = await backend.pick_up(
-        _left_run(handle="t1", session_id=""), pool=_PickUpLink(_left_endpoint("done", rc=0))
+        _left_run(handle="t1", session_id="", stateful=True), pool=_PickUpLink(_left_endpoint("done", rc=0))
     )
-    assert picked.status == "ok" and rows == [] and "not resumable" in picked.reply, "no id in the output: said so"
+    assert picked.status == "ok" and "not resumable" not in picked.reply, "resumable, through the newer binding"
 
-    picked = await backend.pick_up(
-        _left_run(handle="t1", session_id="bound-before"), pool=_PickUpLink(_left_endpoint("session zzz", "done", rc=0))
+
+async def test_the_pick_up_does_not_wait_for_a_dispatch_holding_the_handle_and_its_binding_still_wins(box, tmp_path):
+    """The interleaving itself, with the real registry: the dispatch holds the handle through its create and
+    commit while the picked-up command ends; the pick-up binds at once -- it waits for no turn, and no run picked
+    up after it does -- and the dispatch's commit, which comes last, is what the handle ends on."""
+    from raven.agent.subagent.instances import hold_handle
+
+    backend = build_third_party_backend(_derived_config())
+    registry = InstanceRegistry(path=tmp_path / "instances.json")
+    backend._registry = registry  # type: ignore[attr-defined]
+    endpoint = _left_endpoint("session first", rc=None)  # the command on the machine has not ended
+    picking = asyncio.create_task(
+        backend.pick_up(_left_run(handle="t1", session_id="", stateful=True), pool=_PickUpLink(endpoint))
     )
-    assert picked.status == "ok" and rows == [], "a run that resumed a bound session binds nothing"
+    await asyncio.sleep(0.05)
+    assert not picking.done(), "waiting on the machine's command"
+
+    async with hold_handle("tui:s1", "where@box", "t1"):
+        # A dispatch to the same instance: looks up, finds nothing, creates -- and holds the handle throughout.
+        assert await registry.lookup("tui:s1", "where@box", "t1") is None
+        endpoint.feed_frame("done", 2)
+        endpoint.exited(0)  # the picked-up command ends while the dispatch still holds the handle
+        picked = await asyncio.wait_for(picking, timeout=5)
+        assert picked.status == "ok", "the pick-up did not wait for the handle"
+        assert await registry.lookup("tui:s1", "where@box", "t1") == "first", "bound where nothing was bound"
+        await registry.commit("tui:s1", "where@box", "t1", "second")  # the dispatch's deferred commit
+
+    assert await registry.lookup("tui:s1", "where@box", "t1") == "second", "the dispatch commits last and wins"
+
+
+async def test_bind_if_absent_binds_once_and_reads_a_status_row_or_another_kind_as_absent(tmp_path):
+    registry = InstanceRegistry(path=tmp_path / "instances.json")
+
+    assert await registry.bind_if_absent("tui:s1", "a", "h", "first") == "first"
+    assert await registry.bind_if_absent("tui:s1", "a", "h", "second") == "first", "bound already: the first stays"
+    assert await registry.lookup("tui:s1", "a", "h") == "first"
+
+    await registry.upsert_spawn("tui:s1", "a", "idle-one", "idle")  # a status row with no id
+    assert await registry.bind_if_absent("tui:s1", "a", "idle-one", "x") == "x"
+    assert await registry.lookup("tui:s1", "a", "idle-one") == "x"
+
+    await registry.commit("tui:s1", "a", "other", "acp-sess", kind="acp")
+    assert await registry.bind_if_absent("tui:s1", "a", "other", "cli-sess") == "cli-sess", "another kind: absent"
+    assert await registry.lookup("tui:s1", "a", "other") == "cli-sess"
+
+
+async def test_bind_if_absent_closes_a_row_bound_to_this_very_id_and_leaves_one_bound_to_another_alone(tmp_path):
+    """A left turn that resumed the session its handle is bound to: the manager wrote ``running`` on the row when the
+    turn began, and the pick-up's bind is what closes it -- as ``commit``'s rebuild did before -- or the instance
+    reads as interrupted after a reply that was announced. A row bound to another turn's session is that turn's to
+    close."""
+    registry = InstanceRegistry(path=tmp_path / "instances.json")
+    await registry.commit("tui:s1", "a", "own", "X")
+    await registry.upsert_spawn("tui:s1", "a", "own", "running")
+    await registry.commit("tui:s1", "a", "other", "X")
+    await registry.upsert_spawn("tui:s1", "a", "other", "running")
+    was = {key: dict(row) for key, row in registry._load().items()}
+
+    assert await registry.bind_if_absent("tui:s1", "a", "own", "X") == "X"
+    assert await registry.bind_if_absent("tui:s1", "a", "other", "Y") == "X"
+
+    rows = registry._load()
+    own, other = rows[("tui:s1", "a", "own")], rows[("tui:s1", "a", "other")]
+    assert own["agentId"] == "X" and "status" not in own, "bound to this id: the row is this run's, and closed"
+    assert own["createdAtMs"] == was[("tui:s1", "a", "own")]["createdAtMs"]
+    assert other == was[("tui:s1", "a", "other")], "bound to another id: untouched, that turn closes it"
 
 
 async def test_a_left_cli_run_that_exited_non_zero_is_an_error_with_its_one_line(box):