send: worktree creation rides the durable command plane; relay calls get deadlines

crates/doc/src/commands.rs

@@ -312,6 +312,7 @@ mod tests {
             sandbox: zeron_proto::SandboxLevel::WorkspaceWrite,
             auto_approve: false,
             attachments: Vec::new(),
+            worktree: None,
             resume: None,
         }
     }

crates/doc/tests/attachments_roundtrip.rs

@@ -16,6 +16,7 @@ fn run_request_attachments_survive_command_round_trip() {
         sandbox: zeron_proto::SandboxLevel::WorkspaceWrite,
         auto_approve: true,
         attachments: vec!["/tmp/a.png".into()],
+        worktree: None,
         resume: None,
     };
     doc.queue_command(&SessionCommandEntry {

crates/engine/src/doc_host.rs

@@ -172,6 +172,8 @@ struct DocHostInner {
     /// graph only drops once this back-edge is severed.
     sessions: Mutex<Option<SessionsEngine>>,
     workspace: OnceLock<WorkspaceHost>,
+    /// Worktree materialization for Run commands (see `set_repos`).
+    repos: OnceLock<crate::repos::Repos>,
     /// Cancels every worker spawned through `spawn_worker` — the loops'
     /// own exit conditions (weak handle death, closed channels) don't cover
     /// runtime replacement, where Edge-capable tasks must stop doing
@@ -387,6 +389,7 @@ impl DocHost {
                 config,
                 sessions: Mutex::new(None),
                 workspace: OnceLock::new(),
+                repos: OnceLock::new(),
                 shutdown: CancellationToken::new(),
                 tasks: TaskTracker::new(),
                 handles: Mutex::new(HashMap::new()),
@@ -482,6 +485,12 @@ impl DocHost {
         Box::new(move || weak.upgrade().is_none())
     }
 
+    /// Wire the repos engine (engine assembly) — worktree materialization for
+    /// Run commands carrying a [`zeron_proto::WorktreeSpec`].
+    pub fn set_repos(&self, repos: crate::repos::Repos) {
+        let _ = self.inner.repos.set(repos);
+    }
+
     /// Wire the workspace host (engine assembly) — the source of chat-ownership rows.
     pub fn set_workspace(&self, workspace: WorkspaceHost) {
         let chats = workspace.watch_chats();
@@ -2046,12 +2055,40 @@ impl DocHost {
                 request,
                 message_id,
             } => {
+                let mut request = request.clone();
+                // Worktree directive (WorktreeSpec): materialize on THIS host at
+                // drain time — the durable command plane replaces the sender's
+                // old blocking CreateWorktree relay RPC, whose lost reply wedged
+                // the composer on "Sending…" while the run proceeded anyway.
+                // `take()` resolves the request before dispatch, so the journal
+                // and steer→new-turn fallbacks reuse the created path instead of
+                // minting another checkout.
+                let fresh_worktree = match request.worktree.take() {
+                    Some(spec) => {
+                        let (cwd, fresh) = self.materialize_worktree(chat_id, &spec).await?;
+                        request.cwd = cwd;
+                        fresh
+                    }
+                    None => None,
+                };
                 // Claim-on-first-command: a run for a chat with no workspace row
                 // creates the row under our device id (we are about to host it).
                 if let Some(ws) = self.workspace() {
                     ws.claim_chat(chat_id, Some(&request.cwd))?;
+                    // A pre-existing row (the client's createChat raced ahead)
+                    // still carries the repo folder — repoint it at the fresh
+                    // worktree, and stamp the actual `zeron/<name>` branch so
+                    // the footer and the title-rename flow see it.
+                    if let Some(wt) = &fresh_worktree {
+                        if let Err(err) = ws.set_chat_cwd(chat_id, &wt.path) {
+                            tracing::warn!(chat = %chat_id, error = %err, "worktree cwd stamp failed");
+                        }
+                        if let Err(err) = ws.set_chat_branch(chat_id, &wt.branch) {
+                            tracing::warn!(chat = %chat_id, error = %err, "worktree branch stamp failed");
+                        }
+                    }
                 }
-                let harness = self.harness_for_request(chat_id, request);
+                let harness = self.harness_for_request(chat_id, &request);
                 // A row with no config renders no harness glyph (and every
                 // later dispatch falls back to the engine default), so stamp
                 // what this run actually executes with. Claimed rows and
@@ -2072,7 +2109,7 @@ impl DocHost {
                     }
                 }
                 sessions
-                    .dispatch(chat_id, harness, request.clone(), Some(message_id.clone()))
+                    .dispatch(chat_id, harness, request, Some(message_id.clone()))
                     .await?;
                 Ok((SessionCommandStatus::Applied, None))
             }
@@ -2186,6 +2223,43 @@ impl DocHost {
         }
     }
 
+    /// Create (or reuse) the isolated worktree a Run's [`zeron_proto::WorktreeSpec`]
+    /// asks for, returning the resolved cwd plus the fresh worktree when one was
+    /// actually created. Reuse guard: a chat whose row already points inside a
+    /// linked worktree of the same repo keeps it — a duplicate Run (client retry
+    /// after a lost ack, ledger reset) must not mint a second checkout.
+    async fn materialize_worktree(
+        &self,
+        chat_id: &str,
+        spec: &zeron_proto::WorktreeSpec,
+    ) -> Result<(String, Option<zeron_proto::Worktree>), EngineError> {
+        if let Some(ws) = self.workspace()
+            && let Ok(Some(chat)) = ws.chat(chat_id)
+            && let Some(cwd) = chat.cwd
+            && cwd != spec.repo_path
+            && crate::workspace_host::linked_worktree_root(std::path::Path::new(&cwd)).as_deref()
+                == Some(spec.repo_path.as_str())
+        {
+            tracing::info!(chat = %chat_id, cwd = %cwd, "worktree spec: reusing the chat's existing worktree");
+            return Ok((cwd, None));
+        }
+        let repos = self
+            .inner
+            .repos
+            .get()
+            .ok_or_else(|| EngineError::Other("repos engine not wired".into()))?;
+        let worktree = repos
+            .create_worktree(std::path::Path::new(&spec.repo_path), &spec.base)
+            .await?;
+        tracing::info!(
+            chat = %chat_id,
+            path = %worktree.path,
+            branch = %worktree.branch,
+            "worktree materialized for run"
+        );
+        Ok((worktree.path.clone(), Some(worktree)))
+    }
+
     /// A steer-turned-run with no in-process `last_request` (engine restarted
     /// since the last turn): rebuild the run config from the chat's workspace
     /// row — cwd from the row, model/reasoning/options/sandbox from its config
@@ -2222,6 +2296,7 @@ impl DocHost {
             auto_approve: false,
             attachments: Vec::new(),
             resume: None,
+            worktree: None,
         })
     }
 

crates/engine/src/lib.rs

@@ -229,6 +229,7 @@ impl EngineCore {
         }
         doc_host.spawn_transcript_salvage(profile.store_root().join("journals"));
         let repos = Repos::new(data_dir, &device_id);
+        doc_host.set_repos(repos.clone());
         let terminals = Terminals::new();
         let uploads = Uploads::from_root_with_fallback(
             profile.uploads_root(),

crates/engine/src/rpc.rs

@@ -612,6 +612,8 @@ impl EngineRpc {
         };
         let client = links.client(target).await?;
         if is_stream_method(method) {
+            // Streams are unbounded by design (a quiet WATCH_* is healthy);
+            // only unary calls below get the reply deadline.
             let rx = match client.subscribe(method, params).await {
                 Ok(rx) => rx,
                 Err(err) => {
@@ -627,14 +629,28 @@ impl EngineRpc {
             });
             return Ok(RpcReply::Stream(stream.boxed()));
         }
-        match client.call(method, params).await {
-            Ok(value) => Ok(RpcReply::Value(value)),
-            Err(err) => {
+        let deadline = forward_deadline(method);
+        match tokio::time::timeout(deadline, client.call(method, params)).await {
+            Ok(Ok(value)) => Ok(RpcReply::Value(value)),
+            Ok(Err(err)) => {
                 if matches!(err, RpcError::Closed | RpcError::Transport(_)) {
                     links.invalidate(target);
                 }
                 Err(err)
             }
+            Err(_) => {
+                // No reply inside the deadline. The link may be a zombie — the
+                // relay's auto-pong keeps a dead host socket looking alive
+                // (ws3 auto-pong incident) — so drop it; the next call re-dials.
+                // NOTE: the remote may still complete the forwarded work; the
+                // caller sees a retryable failure instead of hanging forever
+                // (the "Sending…" wedge, 2026-08-18).
+                links.invalidate(target);
+                Err(RpcError::Transport(format!(
+                    "no reply from device {target} for {method} within {}s",
+                    deadline.as_secs()
+                )))
+            }
         }
     }
 
@@ -760,6 +776,24 @@ impl EngineRpc {
     }
 }
 
+/// Reply deadline for a relay-forwarded unary call. The relay is WebSocket
+/// frames through a DO: a dropped frame (host socket replaced mid-call, DO
+/// restart) loses the reply SILENTLY — the DO's auto-pong keeps the client
+/// socket looking healthy — and an unbounded await wedged callers forever
+/// (the composer's permanent "Sending…", 2026-08-18). Network-bound git and
+/// update methods get a long leash; worktree creation checks out a full tree;
+/// everything else is interactive and must fail fast.
+fn forward_deadline(method: &str) -> std::time::Duration {
+    use std::time::Duration;
+    match method {
+        methods::CLONE_REPO | methods::FETCH_ALL | methods::APPLY_UPDATE => {
+            Duration::from_secs(15 * 60)
+        }
+        methods::CREATE_WORKTREE => Duration::from_secs(120),
+        _ => Duration::from_secs(30),
+    }
+}
+
 /// ControlRpc methods that honor `targetDeviceId` (feature-inventory §2.1). Extend this
 /// list (plus [`is_stream_method`] for streams) to make more of the surface
 /// device-addressable — the handlers themselves need no changes.
@@ -1748,6 +1782,30 @@ mod tests {
         assert!(forwardable(methods::FETCH_ALL));
     }
 
+    /// Every forwardable unary method gets a bounded reply deadline —
+    /// interactive calls fail fast, network-bound git/update calls get the
+    /// long leash, and nothing awaits forever (the "Sending…" wedge).
+    #[test]
+    fn forward_deadlines_are_tiered_and_bounded() {
+        use std::time::Duration;
+        assert_eq!(
+            forward_deadline(methods::CREATE_WORKTREE),
+            Duration::from_secs(120)
+        );
+        assert_eq!(
+            forward_deadline(methods::CLONE_REPO),
+            Duration::from_secs(15 * 60)
+        );
+        assert_eq!(
+            forward_deadline(methods::LIST_BRANCHES),
+            Duration::from_secs(30)
+        );
+        assert_eq!(
+            forward_deadline(methods::QUEUE_COMMAND),
+            Duration::from_secs(30)
+        );
+    }
+
     #[test]
     fn tool_file_paths_keep_workspace_activity_only() {
         assert_eq!(

crates/engine/src/sessions.rs

@@ -677,6 +677,7 @@ impl SessionsEngine {
                             auto_approve: false,
                             attachments: Vec::new(),
                             resume: None,
+                            worktree: None,
                         })
                     });
                 let Some(mut request) = request else {
@@ -939,7 +940,6 @@ impl Inner {
 
 // ── run task ────────────────────────────────────────────────────────────────
 
-
 // ── subagent docs ───────────────────────────────────────────────────────────
 
 /// The per-subagent doc id: `{chatId}--sub--{suffix}`. Constrained by the
@@ -994,21 +994,18 @@ impl SubagentSink {
                 self.written = written;
                 r
             }
-            None => match SegmentWriter::begin(
-                &self.doc,
-                &self.entry_id,
-                device_id,
-                self.started_at,
-            ) {
-                Ok(mut w) => {
-                    let r = w.sync(&rendered);
-                    let (ix, written) = w.into_state();
-                    self.entry_index = Some(ix);
-                    self.written = written;
-                    r
+            None => {
+                match SegmentWriter::begin(&self.doc, &self.entry_id, device_id, self.started_at) {
+                    Ok(mut w) => {
+                        let r = w.sync(&rendered);
+                        let (ix, written) = w.into_state();
+                        self.entry_index = Some(ix);
+                        self.written = written;
+                        r
+                    }
+                    Err(e) => Err(e),
                 }
-                Err(e) => Err(e),
-            },
+            }
         };
         if let Err(err) = result {
             // Fail soft: a broken subagent doc degrades to chip-only, never
@@ -1480,9 +1477,9 @@ async fn drive_run(
         {
             inner.publish(&chat_id, &event);
             let sub_id = subagent_doc_id(&chat_id, parent_tool_use_id);
-            let chip_streaming = folded.iter().any(
-                |p| matches!(p, MessagePart::Tool { id, .. } if id == parent_tool_use_id),
-            );
+            let chip_streaming = folded
+                .iter()
+                .any(|p| matches!(p, MessagePart::Tool { id, .. } if id == parent_tool_use_id));
             let sink_known = subagents.contains_key(parent_tool_use_id);
             if chip_streaming {
                 if !sink_known {
@@ -1507,8 +1504,7 @@ async fn drive_run(
             // A Done with NO sink (a subagent that never streamed — codex
             // turn ends can beat registration) is chip-only: minting a doc
             // just to freeze it empty helps no one.
-            let done_only = !sink_known
-                && matches!(sub_event.as_ref(), AgentEvent::Done { .. });
+            let done_only = !sink_known && matches!(sub_event.as_ref(), AgentEvent::Done { .. });
             if !sink_known && !done_only {
                 let opened = inner.doc_host().and_then(|host| match host.open(&sub_id) {
                     Ok(handle) => Some(handle.doc_arc()),

crates/engine/src/titles.rs

@@ -172,6 +172,7 @@ impl TitleGenerator {
                 auto_approve: true,
                 attachments: Vec::new(),
                 resume: None,
+                worktree: None,
             };
             match collect_text(harness.as_ref(), request).await {
                 Ok(raw) => {

crates/engine/src/workspace_host.rs

@@ -1107,7 +1107,7 @@ impl WorkspaceHostInner {
 /// checkout (`.git` is a directory), a non-repo folder, or any other layout
 /// (bare-repo worktrees have no `<root>` working copy to attribute to). Pure
 /// fs reads — no git subprocess; this runs on the synchronous claim path.
-fn linked_worktree_root(path: &std::path::Path) -> Option<String> {
+pub(crate) fn linked_worktree_root(path: &std::path::Path) -> Option<String> {
     let gitfile = path.join(".git");
     if !std::fs::metadata(&gitfile).ok()?.is_file() {
         return None;
@@ -1280,7 +1280,6 @@ fn device_name_on_boot(existing_name: Option<&str>, detected_name: &str) -> Stri
         .to_string()
 }
 
-
 /// Plain-HTTPS registry pull/push derived from the SAME WebSocket URL
 /// provider the dial uses (`wss://…/registry/{org}/ws?token=…&device=…`):
 /// swap the scheme, swap the `/ws` leaf for `/rows` or `/push`, keep the
@@ -1311,7 +1310,9 @@ impl WsDerivedRegistryTransport {
         let _ = u.set_scheme(scheme);
         let path = u.path().to_string();
         let Some(base) = path.strip_suffix("/ws") else {
-            return Err(zeron_sync::SyncError::Protocol("ws url without /ws leaf".into()));
+            return Err(zeron_sync::SyncError::Protocol(
+                "ws url without /ws leaf".into(),
+            ));
         };
         let new_path = format!("{base}/{leaf}");
         u.set_path(&new_path);

crates/engine/tests/born_chat2_race.rs

@@ -161,6 +161,7 @@ async fn transcript_survives_open_racing_create_chat() {
                         sandbox: SandboxLevel::WorkspaceWrite,
                         auto_approve: true,
                         attachments: Vec::new(),
+                        worktree: None,
                         resume: None,
                     },
                     message_id: "msg-race-1".into(),

crates/engine/tests/device_routing.rs

@@ -307,6 +307,7 @@ async fn target_device_id_routes_over_the_relay() {
             sandbox: SandboxLevel::WorkspaceWrite,
             auto_approve: true,
             attachments: Vec::new(),
+            worktree: None,
             resume: None,
         },
         message_id: "m-a-1".into(),

crates/engine/tests/e2e.rs

@@ -36,6 +36,7 @@ fn run_request(prompt: &str) -> RunRequest {
         sandbox: SandboxLevel::WorkspaceWrite,
         auto_approve: true,
         attachments: Vec::new(),
+        worktree: None,
         resume: None,
     }
 }
@@ -1659,6 +1660,7 @@ async fn real_claude_sees_uploaded_image_inline() {
         auto_approve: false,
         attachments: vec![path],
         resume: None,
+        worktree: None,
     };
     core.doc_host
         .queue_command(