feat(*): pick up a turn left on a registered machine when this computer stops

CONTEXT.md

@@ -401,10 +401,14 @@ of it. The entry names only the machine's id, and the entry's `env` reaches the
 agent on the link, never on a command line. Such an entry is added the way any
 preset is, with `machine` (and
 `remote_cwd`, for an `acp` or `cli` one) on `subagents.add` — from a chat through
-`raven_config`, which lists the registered machines while the lane is on; the page's
-add passes no machine yet — refused before anything is spent when the lane is off,
+`raven_config`, which lists the usable registered machines; the page's
+add passes no machine yet — refused before anything is spent when
 the id is not usable or the preset is a built-in agent, and proved by the same
 readiness ping as a local add, run on that machine.
+Registered-machine dispatch, file access and recovery are available without an
+environment switch. An entry without `machine` still runs here. Registration,
+SSH authentication, file permissions and concurrency limits govern remote work;
+listing entries and boot capability backfills do not open machine connections.
 _Avoid_: "host" / "server" / "node" (too broad, no link to the
 `ops connection` registry that supplies the rows); "GPU box" (only some are
 GPU hosts, and the term covers any registered compute destination);
@@ -439,8 +443,17 @@ named by the code's version and digest -- its install name, `bundle.install_name
 host keeps at most one live link per machine and watches it with a heartbeat
 (`raven/node/client.py`): a link silent for 15 s is lost, the agents on it wait, and the
 link is tried again on a backoff for as long as any waits; a link that comes back attaches
-each agent again by its task id. Only `--stdio` (tests, a fallback) ties the agents to the
-connection.
+each agent again by its task id. A turn running on a machine when this computer stops is
+left with the daemon (`remote.leave_agents_running()`), remembered while in flight in
+`<raven home>/machines/runs/` (`raven/acp_client/remote_runs.py`), and picked up by the
+next start (`raven/agent/subagent/resume_remote.py`): attached from the frame its prompt
+went out after, so the turn is read again whole (a cli entry's command from its first line,
+its whole output being the reply), its reply read off the machine and announced into the
+conversation that asked -- the task quoted as the model wrote it, the record removed only
+once the announce was taken. A turn that returned here keeps its record the same way until
+the manager's announce of the reply was taken (`remote_runs.forget_dispatch`), so a host
+that stops between the two reads the reply again. Only
+`--stdio` (tests, a fallback) ties the agents to the connection.
 _Avoid_: "node" alone (a DAG node is a `run_subagent_dag` step, and the Machine
 entry already rules the bare word out for a computer); "node id" for the install
 name (a **Node id** is a delegated task's address); "agent" or "server" (it

docs-site/docs/agent-collaboration.md

@@ -7,6 +7,28 @@ tasks and stateful agent instances; they are different things.
 Set up agents first with [Agent Integrations](agent-integrations.md). This page
 covers using them, not installing their transports.
 
+## Work on a registered machine
+
+Register the destination with `raven ops connection add`, then install its node
+with `raven ops connection install-node <machine-id>`. Set an agent entry's
+`machine` to that registry id, or ask Raven to add a preset on that machine.
+Start Raven normally, for example with `raven serve`; no environment switch is
+required. Entries without `machine` continue to run locally.
+
+ACP and CLI entries run under the machine's node. An OpenAI entry reaches its
+HTTP endpoint through an SSH tunnel from that machine; built-in entries run
+locally. The registered SSH credentials, file permission prompts and concurrency
+limits still apply. Attachments are named by address, not copied automatically.
+
+If Raven stops during a remote ACP or CLI task, the machine keeps working. The
+next start uses the records in the same Raven home to recover the result into
+the original conversation. Keep the machine registration and agent entry
+available for that recovery.
+
+`RAVEN_DEBUG_MACHINES` was a default-off trial switch for machine dispatch and
+remote file access. It is no longer read; an old value of `0` does not disable
+these features or prevent task recovery.
+
 ## Choose a collaboration shape
 
 | Need | Use | Completion evidence |

docs-site/docs/agent-collaboration.zh.md

@@ -5,6 +5,24 @@ Raven 同时提供委派任务与有状态 Agent 实例，两者不是一回事
 
 先按[Agent 集成](agent-integrations.md)完成配置。本页介绍如何协作，不介绍传输安装。
 
+## 在登记机器上工作 { #work-on-a-registered-machine }
+
+先用 `raven ops connection add` 登记设备，再用
+`raven ops connection install-node <机器 id>` 安装节点。把 Agent 配置的
+`machine` 设为该登记 id，或在对话里让 Raven 把预设 Agent 添加到那台机器。
+正常运行 `raven serve` 即可，不需要额外环境开关；未配置 `machine` 的 Agent 仍在本机运行。
+
+ACP 和 CLI 类型由设备上的节点启动；OpenAI 类型通过 SSH 隧道访问设备那边的
+HTTP 端点；内置 Agent 仍在本机运行。SSH 登录凭据、文件审批和并发限制照常生效。
+附件只提供地址，不自动复制。
+
+远端 ACP 或 CLI 任务运行时退出 Raven，设备上的任务会继续。下次使用同一个
+Raven 主目录启动时，会根据保存的记录接回结果并通告到原会话。接回时需保留
+对应的机器登记和 Agent 配置。
+
+`RAVEN_DEBUG_MACHINES` 原来是默认关闭的试用开关，控制远端派发和按地址访问文件。
+现在不再读取这个变量；旧环境里即使还设为 `0`，也不会禁用这些功能或阻止任务恢复。
+
 ## 选择协作形式 { #choose-a-collaboration-shape }
 
 | 需求 | 方式 | 完成证据 |

raven/acp_client/__init__.py

@@ -16,6 +16,9 @@
 - :mod:`raven.acp_client.capabilities` -- turning one ``initialize`` handshake into
   a stored snapshot of what the agent can actually do, so the roster advertises
   measurements rather than hand-typed declarations.
+- :mod:`raven.acp_client.remote_pickup` -- the types a picked-up run is reported
+  with, in a leaf of their own so the cli backend need not import the pick-up
+  module (see the guard in ``tests/test_import_cycle_budget.py``).
 """
 
 from raven.acp_client.capabilities import (

raven/acp_client/acp_agent.py

@@ -36,7 +36,7 @@
 
 from loguru import logger
 
-from raven.acp_client import remote
+from raven.acp_client import remote, remote_runs
 from raven.acp_client.acp_dialects import AcpDialect, ToolCall, content_texts, dialect_for
 from raven.acp_client.ask_user import AskUserResponder, clarify_responder
 from raven.acp_client.capabilities import (
@@ -1536,6 +1536,11 @@ async def run(
         authored_task: str | None = None,
         on_delta: Callable[[str], Awaitable[None]] | None = None,
         media: Sequence[Media] = (),
+        # Whether the caller reports the reply itself (the manager's announce)
+        # and removes the run's record on its machine then
+        # (`remote_runs.forget_dispatch`); otherwise the record goes as this
+        # returns, as for a caller that takes the reply as its own result.
+        reported_by_caller: bool = False,
     ) -> str:
         # The parent binding is forwarded when the pooled ACP worker launches.
         # Without this, a long-lived worker keeps the provider/model captured
@@ -1658,6 +1663,23 @@ async def run(
                         }
                     )
                 record_session(span, session_id=session_id, resumed=resumed)
+                # A run on a machine is remembered here while it is in flight,
+                # so a process after this one can pick it up where this one
+                # left it (remote_runs): removed when the turn ends here, kept
+                # when this process stops and leaves the agent to the machine.
+                tracker = (
+                    self._track_on_machine(
+                        client,
+                        connection,
+                        session_id=session_id,
+                        skey=skey,
+                        handle=handle,
+                        task_id=task_id,
+                        session_cwd=session_cwd,
+                    )
+                    if launched is not None
+                    else None
+                )
 
                 # Snapshotted before the prompt so the error below names what this
                 # turn ran into rather than what the connection has ever refused.
@@ -1700,6 +1722,7 @@ async def run(
                     connection.responders.attach(session_id, responder)
                     if self.can_steer or steer_offered(connection.initialize):
                         activity.offer_steer(collector._run, self._steerer(client, session_id))
+                    answered = False
                     try:
                         result = await client.request(
                             "session/prompt",
@@ -1719,8 +1742,15 @@ async def run(
                             },
                             timeout=_prompt_deadline(self.timeout),
                             cancel_session=session_id,
+                            sent=tracker.prompt_sent if tracker is not None else None,
                         )
+                        answered = True
                     except asyncio.CancelledError:
+                        if tracker is not None and remote.leaving():
+                            # This process is stopping: the turn goes on under
+                            # the machine's daemon, and the manager reads this
+                            # dispatch as left, not cancelled.
+                            remote_runs.mark_left(task_id)
                         # An UNSETTLED cancel means the turn outlived its cancel
                         # budget, so it is still running on the agent while this lock
                         # is about to be released -- prompting the same session again
@@ -1752,6 +1782,14 @@ async def run(
                         record_frames(span, cancelled)
                         raise
                     finally:
+                        if tracker is not None:
+                            # The record goes with a turn that ended here with no
+                            # reply. It stays while this process leaves the turn
+                            # running, and with the reply for a caller that reports
+                            # it (`remote_runs.forget_dispatch`): a host that stops
+                            # between the answer and the announce has no other way
+                            # back to the run.
+                            self._untrack(client, tracker, keep=remote.leaving() or (answered and reported_by_caller))
                         activity.offer_steer(collector._run, None)
                         connection.router.detach(session_id, collector)
                         connection.elicitors.detach(session_id, elicitor)
@@ -1879,6 +1917,60 @@ async def launcher() -> Any:
 
         return _OnMachine(target=target, root=root, session_cwd=session_cwd, since=since, link=link, launcher=launcher)
 
+    def _track_on_machine(
+        self,
+        client: Any,
+        connection: Any,
+        *,
+        session_id: str,
+        skey: str,
+        handle: str,
+        task_id: str,
+        session_cwd: str,
+    ) -> remote_runs.Tracker | None:
+        """The record of this turn on its machine, kept current until the prompt is out (``remote_runs.Tracker``)."""
+        endpoint = client.endpoint
+        task = getattr(endpoint, "task", None)
+        watchers = getattr(endpoint, "watchers", None)
+        if not isinstance(task, str) or watchers is None:
+            return None
+        run = remote_runs.RemoteRun(
+            machine=self.machine or "",
+            task=task,
+            agent=self.name,
+            session_key=skey,
+            handle=handle,
+            task_id=task_id,
+            session_id=session_id,
+            prompt_id=0,
+            next_id=client.next_request_id,
+            initialize=connection.initialize,
+            launch_key=connection.launch_key,
+            config_key=connection.config_key,
+            session_cwd=session_cwd,
+            stateful=self.is_stateful,
+            received=int(getattr(endpoint, "received", 0) or 0),
+        )
+        try:
+            tracker = remote_runs.Tracker(run)
+        except OSError as exc:
+            logger.warning(
+                "acp agent {!r}: the run on machine {} could not be recorded: {}", self.name, self.machine, exc
+            )
+            return None
+        watchers.append(tracker.received)
+        return tracker
+
+    @staticmethod
+    def _untrack(client: Any, tracker: remote_runs.Tracker, *, keep: bool) -> None:
+        watchers = getattr(client.endpoint, "watchers", None)
+        if watchers is not None:
+            try:
+                watchers.remove(tracker.received)
+            except ValueError:
+                pass
+        tracker.close(keep=keep)
+
     @property
     def can_steer(self) -> bool:
         """Whether this agent takes text mid-turn (``_raven/session/steer``).

raven/acp_client/client.py

@@ -630,9 +630,13 @@ async def request(
         *,
         timeout: float | None = None,
         cancel_session: str | None = None,
+        sent: Callable[[int], None] | None = None,
     ) -> Any:
         """Send a request and await its result.
 
+        ``sent`` hears the request's id as it goes out: what a run on a
+        machine records, so a process after this one can await the answer.
+
         Raises :class:`AcpRemoteError` if the agent answers with an error,
         :class:`AcpTimeoutError` on budget expiry, :class:`AcpConnectionError` if
         the connection is gone.
@@ -667,6 +671,8 @@ async def request(
         self._request_sessions[request_id] = (params or {}).get("sessionId") if isinstance(params, dict) else None
         self._request_methods[request_id] = method
         try:
+            if sent is not None:
+                sent(request_id)
             await self._send(protocol.request(request_id, method, params))
             if timeout is None:
                 return await asyncio.shield(future)
@@ -684,6 +690,51 @@ async def request(
             self._request_sessions.pop(request_id, None)
             self._request_methods.pop(request_id, None)
 
+    @property
+    def endpoint(self) -> Endpoint:
+        """What this connection runs over: a child process here, or a channel to an agent on a machine."""
+        return self._endpoint
+
+    @property
+    def next_request_id(self) -> int:
+        """The id the next request from here carries."""
+        return self._next_id + 1
+
+    def continue_ids_from(self, next_id: int) -> None:
+        """Carry request ids on from ``next_id``: for a connection taken over from a process that used the ones before it."""
+        self._next_id = max(self._next_id, int(next_id) - 1)
+
+    async def await_response(
+        self, request_id: int, *, method: str, session: str | None = None, timeout: float | None = None
+    ) -> Any:
+        """The answer to a request another process of this Raven sent on this connection: awaited, never sent.
+
+        What a run picked up after a restart waits on: the ``session/prompt``
+        the process before this one sent, whose answer the agent still owes.
+        Raises as :meth:`request` does.
+        """
+        if not self.alive:
+            if self._closed:
+                raise AcpConnectionError(f"acp agent {self.name!r}: connection is not open")
+            await self._settle_exit()
+            raise self._ended_error()
+        future: asyncio.Future[Any] = asyncio.get_running_loop().create_future()
+        self._pending[request_id] = future
+        self._request_sessions[request_id] = session
+        self._request_methods[request_id] = method
+        try:
+            if timeout is None:
+                return await asyncio.shield(future)
+            return await asyncio.wait_for(asyncio.shield(future), timeout=timeout)
+        except asyncio.TimeoutError:
+            raise AcpTimeoutError(
+                f"acp agent {self.name!r}: {method} timed out after {timeout}s", method=method
+            ) from None
+        finally:
+            self._pending.pop(request_id, None)
+            self._request_sessions.pop(request_id, None)
+            self._request_methods.pop(request_id, None)
+
     @property
     def prompting(self) -> bool:
         """Whether a turn is in flight on this connection right now.
@@ -738,6 +789,10 @@ async def _cancel_turn(self, session_id: str, future: asyncio.Future) -> None:
         this codebase cancels once and then gathers -- the same property
         ``CliAgentBackend._kill_process_group`` relies on to await the child.
         """
+        if getattr(self._endpoint, "stays", False):
+            # This process is stopping and the agent stays with its machine's
+            # daemon: the turn goes on there, and the next process picks it up.
+            return
         try:
             await self.notify("session/cancel", {"sessionId": session_id})
         except Exception:  # noqa: BLE001 - a connection already gone has nothing to settle
@@ -810,6 +865,12 @@ async def _read_stdout(self) -> None:
                 if self._journal is not None:
                     self._journal.note("in", frame=frame, session=self._frame_session(frame))
                 await self._dispatch(frame)
+                # Told after the frame is dealt with, so what an endpoint
+                # records as read is what was read: a channel to a machine
+                # keeps the count a later attach goes on from.
+                consumed = getattr(self._endpoint, "consumed", None)
+                if consumed is not None:
+                    consumed()
         except asyncio.CancelledError:
             raise
         except Exception as exc:  # noqa: BLE001 - the loop must not die silently