fix(agent): a pick-up binds a cli instance the way a dispatch does

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

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(

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

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

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,30 +216,56 @@ 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]:
             best = (started, path.name[: -len(".meta.json")], meta)
     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"]

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"

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