diff --git a/crates/doc/src/commands.rs b/crates/doc/src/commands.rs index 0832081d67..469e6b1b5c 100644 --- a/crates/doc/src/commands.rs +++ b/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, } } diff --git a/crates/doc/tests/attachments_roundtrip.rs b/crates/doc/tests/attachments_roundtrip.rs index 493ef41598..c463468afe 100644 --- a/crates/doc/tests/attachments_roundtrip.rs +++ b/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 { diff --git a/crates/engine/src/doc_host.rs b/crates/engine/src/doc_host.rs index 99904bf913..6516e32aa5 100644 --- a/crates/engine/src/doc_host.rs +++ b/crates/engine/src/doc_host.rs @@ -172,6 +172,8 @@ struct DocHostInner { /// graph only drops once this back-edge is severed. sessions: Mutex>, workspace: OnceLock, + /// Worktree materialization for Run commands (see `set_repos`). + repos: OnceLock, /// 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/` 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), 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, }) } diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index 6a3d9ee558..ff98109c18 100644 --- a/crates/engine/src/lib.rs +++ b/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(), diff --git a/crates/engine/src/rpc.rs b/crates/engine/src/rpc.rs index 49d870add4..e188c3c443 100644 --- a/crates/engine/src/rpc.rs +++ b/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!( diff --git a/crates/engine/src/sessions.rs b/crates/engine/src/sessions.rs index 5e90f01a06..24717d9608 100644 --- a/crates/engine/src/sessions.rs +++ b/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()), diff --git a/crates/engine/src/titles.rs b/crates/engine/src/titles.rs index 8139989036..f87120ab5f 100644 --- a/crates/engine/src/titles.rs +++ b/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) => { diff --git a/crates/engine/src/workspace_host.rs b/crates/engine/src/workspace_host.rs index bbba093981..492e8d15ee 100644 --- a/crates/engine/src/workspace_host.rs +++ b/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 `` 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 { +pub(crate) fn linked_worktree_root(path: &std::path::Path) -> Option { 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); diff --git a/crates/engine/tests/born_chat2_race.rs b/crates/engine/tests/born_chat2_race.rs index 0674a4edf5..07fbadf89b 100644 --- a/crates/engine/tests/born_chat2_race.rs +++ b/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(), diff --git a/crates/engine/tests/device_routing.rs b/crates/engine/tests/device_routing.rs index eacea1657c..e16cff6aab 100644 --- a/crates/engine/tests/device_routing.rs +++ b/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(), diff --git a/crates/engine/tests/e2e.rs b/crates/engine/tests/e2e.rs index 3470d3bef7..104617bf84 100644 --- a/crates/engine/tests/e2e.rs +++ b/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( diff --git a/crates/engine/tests/m5c_accounts_uploads_titles.rs b/crates/engine/tests/m5c_accounts_uploads_titles.rs index eedec34a20..8029d3620f 100644 --- a/crates/engine/tests/m5c_accounts_uploads_titles.rs +++ b/crates/engine/tests/m5c_accounts_uploads_titles.rs @@ -572,6 +572,7 @@ async fn titling_e2e_names_chat_and_renames_worktree_branch() { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, }; core.sessions @@ -616,6 +617,7 @@ async fn titling_e2e_names_chat_and_renames_worktree_branch() { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, }; core.sessions @@ -954,7 +956,10 @@ exit 0 .await .expect("start"); assert_eq!(start.mode, AgentLoginMode::Browser); - assert_eq!(start.url, "https://cursor.com/loginDeepControl?challenge=fake"); + assert_eq!( + start.url, + "https://cursor.com/loginDeepControl?challenge=fake" + ); let deadline = tokio::time::Instant::now() + Duration::from_secs(15); loop { @@ -964,10 +969,7 @@ exit 0 AgentLoginStatus::Pending => {} AgentLoginStatus::Error => panic!("login errored: {:?}", poll.message), } - assert!( - tokio::time::Instant::now() < deadline, - "login never landed" - ); + assert!(tokio::time::Instant::now() < deadline, "login never landed"); tokio::time::sleep(Duration::from_millis(50)).await; } diff --git a/crates/engine/tests/projectless.rs b/crates/engine/tests/projectless.rs index 8ae7c80091..ca89f38d02 100644 --- a/crates/engine/tests/projectless.rs +++ b/crates/engine/tests/projectless.rs @@ -167,6 +167,7 @@ async fn projectless_chat_runs_from_home_and_mints_no_space() { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, }, message_id: "msg-np-1".into(), diff --git a/crates/engine/tests/restart_resume.rs b/crates/engine/tests/restart_resume.rs index 2a3bb48462..bf5dfe50e1 100644 --- a/crates/engine/tests/restart_resume.rs +++ b/crates/engine/tests/restart_resume.rs @@ -46,6 +46,7 @@ fn run_request(prompt: &str, cwd: &str) -> RunRequest { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, } } @@ -848,6 +849,7 @@ async fn real_claude_remembers_codeword_across_engine_restart() { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: false, attachments: Vec::new(), + worktree: None, resume: None, }; let assemble_real = || { diff --git a/crates/engine/tests/self_continued_quiesce.rs b/crates/engine/tests/self_continued_quiesce.rs index 2ee70d11be..35d4b7b0a4 100644 --- a/crates/engine/tests/self_continued_quiesce.rs +++ b/crates/engine/tests/self_continued_quiesce.rs @@ -57,6 +57,7 @@ fn run_request(prompt: &str) -> RunRequest { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, } } diff --git a/crates/engine/tests/turn_quiesce.rs b/crates/engine/tests/turn_quiesce.rs index 5bd7c15494..e73619732d 100644 --- a/crates/engine/tests/turn_quiesce.rs +++ b/crates/engine/tests/turn_quiesce.rs @@ -58,6 +58,7 @@ fn run_request(prompt: &str) -> RunRequest { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, } } diff --git a/crates/engine/tests/unarchive_on_send.rs b/crates/engine/tests/unarchive_on_send.rs index c332530785..5f9d5c484c 100644 --- a/crates/engine/tests/unarchive_on_send.rs +++ b/crates/engine/tests/unarchive_on_send.rs @@ -109,6 +109,7 @@ fn run_payload(message_id: &str) -> SessionCommandPayload { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, }, message_id: message_id.into(), diff --git a/crates/engine/tests/workspace_sync.rs b/crates/engine/tests/workspace_sync.rs index 30859e7eab..632a9e188e 100644 --- a/crates/engine/tests/workspace_sync.rs +++ b/crates/engine/tests/workspace_sync.rs @@ -154,6 +154,7 @@ fn run_request(prompt: &str) -> RunRequest { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, } } diff --git a/crates/engine/tests/worktree_on_run.rs b/crates/engine/tests/worktree_on_run.rs new file mode 100644 index 0000000000..8f1aef85e0 --- /dev/null +++ b/crates/engine/tests/worktree_on_run.rs @@ -0,0 +1,243 @@ +//! Host-side worktree materialization: a Run command carrying a +//! `WorktreeSpec` creates the isolated worktree on the HOST at drain time +//! (the durable replacement for the composer's old blocking CreateWorktree +//! relay RPC), runs there, and stamps the chat row's cwd + `zeron/` +//! branch. A second spec-carrying Run for the same chat REUSES the checkout +//! instead of minting another. + +use std::path::PathBuf; +use std::process::Command; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use async_trait::async_trait; +use futures::StreamExt; +use futures::stream::BoxStream; + +use zeron_doc::{MessageRole, MessageStatus, SessionCommandPayload, SessionMessageEntry}; +use zeron_engine::{EngineCore, HarnessRegistry}; +use zeron_harness::{Harness, HarnessError, RunControls}; +use zeron_proto::{ + AgentEvent, DoneStatus, HarnessId, Model, ReasoningLevel, RunRequest, SandboxLevel, + SteeringMode, WorktreeSpec, +}; + +const CHAT: &str = "chat-worktree-run"; + +/// Completes a one-line turn and records the cwd each run spawned with. +struct RecordingHarness { + cwds: Arc>>, +} + +#[async_trait] +impl Harness for RecordingHarness { + fn id(&self) -> HarnessId { + HarnessId::Mock + } + fn display_name(&self) -> &str { + "Recorder" + } + fn supports_steering(&self) -> bool { + false + } + fn steering_mode(&self) -> SteeringMode { + SteeringMode::TurnBoundary + } + fn reasoning_levels(&self) -> &[ReasoningLevel] { + &[ReasoningLevel::Medium] + } + async fn models(&self) -> Result, HarnessError> { + Ok(vec![]) + } + async fn run( + &self, + request: RunRequest, + _controls: RunControls, + ) -> Result>, HarnessError> { + self.cwds.lock().unwrap().push(request.cwd.clone()); + let events: Vec> = vec![ + Ok(AgentEvent::SessionStarted { + harness: HarnessId::Mock, + model: "mock-1".into(), + tools: vec![], + cwd: request.cwd.clone(), + session_id: "sess-wt".into(), + assistant_message_id: "a-1".into(), + }), + Ok(AgentEvent::TextDelta { + text: format!("ack: {}", request.prompt), + }), + Ok(AgentEvent::Done { + status: DoneStatus::Completed, + result: None, + error: None, + session_id: Some("sess-wt".into()), + }), + ]; + Ok(futures::stream::iter(events).boxed()) + } +} + +async fn wait_for(mut predicate: F, what: &str) +where + F: FnMut() -> bool, +{ + let deadline = tokio::time::Instant::now() + Duration::from_secs(20); + while !predicate() { + assert!( + tokio::time::Instant::now() < deadline, + "timed out waiting for {what}" + ); + tokio::time::sleep(Duration::from_millis(15)).await; + } +} + +fn complete_assistant_count(core: &EngineCore) -> usize { + let entries: Vec = core + .doc_host + .open(CHAT) + .ok() + .and_then(|h| h.doc().read_entries().ok()) + .unwrap_or_default(); + entries + .iter() + .filter(|e| e.role == MessageRole::Assistant && e.status == Some(MessageStatus::Complete)) + .count() +} + +fn run_payload(message_id: &str, repo_path: &str) -> SessionCommandPayload { + SessionCommandPayload::Run { + request: RunRequest { + prompt: "isolated please".into(), + harness: None, + model: None, + reasoning: None, + model_options: Default::default(), + // Fallback for hosts that predate the spec: the repo's own folder. + cwd: repo_path.into(), + sandbox: SandboxLevel::WorkspaceWrite, + auto_approve: true, + attachments: Vec::new(), + resume: None, + worktree: Some(WorktreeSpec { + repo_path: repo_path.into(), + base: "main".into(), + }), + }, + message_id: message_id.into(), + } +} + +fn git(cwd: &std::path::Path, args: &[&str]) { + let out = Command::new("git") + .args(args) + .current_dir(cwd) + .output() + .expect("git runs"); + assert!( + out.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&out.stderr) + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn run_with_worktree_spec_materializes_on_host_and_reuses() { + let tmp = tempfile::tempdir().unwrap(); + // Canonicalize: git records canonical paths in worktree gitdir links, and + // macOS tempdirs live behind the /var → /private/var symlink. + let tmp_path = tmp.path().canonicalize().unwrap(); + let worktrees_root = tmp_path.join("worktrees"); + unsafe { std::env::set_var("ZERON_WORKTREES_DIR", &worktrees_root) }; + + let repo_dir = tmp_path.join("repo"); + std::fs::create_dir_all(&repo_dir).unwrap(); + git(&repo_dir, &["init", "-b", "main"]); + git(&repo_dir, &["config", "user.email", "t@example.com"]); + git(&repo_dir, &["config", "user.name", "Test"]); + std::fs::write(repo_dir.join("README.md"), "hello\n").unwrap(); + git(&repo_dir, &["add", "."]); + git(&repo_dir, &["commit", "-m", "init"]); + let repo_path = repo_dir.to_string_lossy().to_string(); + + let cwds: Arc>> = Arc::new(Mutex::new(Vec::new())); + let registry = HarnessRegistry::new(); + registry.register(Arc::new(RecordingHarness { cwds: cwds.clone() })); + let core = EngineCore::assemble( + &tmp_path.join("data"), + Arc::new(registry), + HarnessId::Mock, + None, + ) + .expect("engine core assembles"); + + // Mirror the composer: createChat lands first (cwd-less; the engine + // resolves the project folder), then the queued Run carries the spec. + let client = zeron_rpc::memory_client(core.rpc_service()); + client + .call( + zeron_rpc::methods::MUTATE, + serde_json::json!({ + "op": "createChat", + "chatId": CHAT, + "deviceId": core.device_id, + }), + ) + .await + .expect("createChat"); + // Pre-title so the auto-titler's harness request stays out of the flow. + core.workspace + .rename_chat(CHAT, "Pre-titled") + .expect("rename chat"); + + core.doc_host + .queue_command(CHAT, run_payload("msg-wt-1", &repo_path)) + .expect("queue run command"); + wait_for(|| complete_assistant_count(&core) == 1, "first turn").await; + + let first_cwd = cwds.lock().unwrap().first().cloned().expect("run recorded"); + assert_ne!( + first_cwd, repo_path, + "the run must execute in a fresh worktree, not the repo folder" + ); + let first = PathBuf::from(&first_cwd); + assert!( + first.starts_with(&worktrees_root), + "worktree lands under the worktrees root: {first_cwd}" + ); + assert!( + first.join(".git").is_file(), + "a linked worktree has a .git FILE" + ); + + // The chat row follows: cwd repointed at the worktree, branch stamped + // with the actual zeron/ (the composer only knew the base). + let chat = core + .workspace + .chat(CHAT) + .expect("read chat row") + .expect("chat row exists"); + assert_eq!(chat.cwd.as_deref(), Some(first_cwd.as_str())); + let branch = chat.branch.expect("branch stamped"); + assert!( + branch.starts_with("zeron/"), + "stamped branch is the worktree's own: {branch}" + ); + + // A duplicate spec-carrying Run (client retry) REUSES the checkout. + core.doc_host + .queue_command(CHAT, run_payload("msg-wt-2", &repo_path)) + .expect("queue second run"); + wait_for(|| complete_assistant_count(&core) == 2, "second turn").await; + let second_cwd = cwds.lock().unwrap().get(1).cloned().expect("second run"); + assert_eq!( + second_cwd, first_cwd, + "a second Run with the spec must reuse the chat's worktree" + ); + let minted = std::fs::read_dir(worktrees_root.join("repo")) + .map(|entries| entries.count()) + .unwrap_or(0); + assert_eq!(minted, 1, "exactly one worktree minted for the chat"); + + core.shutdown().await; +} diff --git a/crates/harness/examples/grok_subagent_probe.rs b/crates/harness/examples/grok_subagent_probe.rs index 847cf0fa2e..ea4895abeb 100644 --- a/crates/harness/examples/grok_subagent_probe.rs +++ b/crates/harness/examples/grok_subagent_probe.rs @@ -46,6 +46,7 @@ async fn main() { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, }; let mut stream = AcpHarness::grok() diff --git a/crates/harness/tests/acp.rs b/crates/harness/tests/acp.rs index 149677bac7..c30f1d6e85 100644 --- a/crates/harness/tests/acp.rs +++ b/crates/harness/tests/acp.rs @@ -7,9 +7,7 @@ use std::time::Duration; use futures::StreamExt; use tokio::sync::{mpsc, oneshot}; -use zeron_harness::{ - AcpHarness, CancellationToken, Harness, RunControls, SteerMessage, -}; +use zeron_harness::{AcpHarness, CancellationToken, Harness, RunControls, SteerMessage}; use zeron_proto::{ AgentEvent, DoneStatus, HarnessId, RunRequest, SandboxLevel, SteeringMode, TodoItem, ToolCall, UserInputAnswer, @@ -43,6 +41,7 @@ fn request(prompt: &str) -> RunRequest { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, } } @@ -764,8 +763,9 @@ async fn grok_subagent_lifecycle_tails_the_disk_transcript_into_tagged_events() ) }) .expect("tool result tailed"); - let text = pos(&|e| matches!(e, AgentEvent::TextDelta { text } if text.starts_with("two files"))) - .expect("mid-run append tailed"); + let text = + pos(&|e| matches!(e, AgentEvent::TextDelta { text } if text.starts_with("two files"))) + .expect("mid-run append tailed"); let done = pos(&|e| { matches!( e, diff --git a/crates/harness/tests/acp_quiet.rs b/crates/harness/tests/acp_quiet.rs index 66dfab4bd4..02ec35f473 100644 --- a/crates/harness/tests/acp_quiet.rs +++ b/crates/harness/tests/acp_quiet.rs @@ -51,6 +51,7 @@ fn request(prompt: &str) -> RunRequest { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, } } diff --git a/crates/harness/tests/acp_stall.rs b/crates/harness/tests/acp_stall.rs index 1748df7391..aaec2496bc 100644 --- a/crates/harness/tests/acp_stall.rs +++ b/crates/harness/tests/acp_stall.rs @@ -53,6 +53,7 @@ async fn silent_agent_errors_via_the_prompt_stall_watchdog() { sandbox: SandboxLevel::DangerFullAccess, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, }; let harness = AcpHarness::grok().with_executable(fixture_path()); diff --git a/crates/harness/tests/claude.rs b/crates/harness/tests/claude.rs index 9d536c4699..5abc646acf 100644 --- a/crates/harness/tests/claude.rs +++ b/crates/harness/tests/claude.rs @@ -45,6 +45,7 @@ fn request(prompt: &str) -> RunRequest { sandbox: SandboxLevel::DangerFullAccess, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, } } @@ -226,7 +227,11 @@ async fn eager_done_forwards_wake_turn_as_second_done() { .enumerate() .filter_map(|(i, e)| matches!(e, AgentEvent::Done { .. }).then_some(i)) .collect(); - assert_eq!(done_positions.len(), 2, "eager done + wake done: {events:?}"); + assert_eq!( + done_positions.len(), + 2, + "eager done + wake done: {events:?}" + ); // One SessionStarted total — the wake init is deduped. assert_eq!( @@ -475,7 +480,10 @@ async fn captured_live_background_subagent_frames_replay_correctly() { let cli = dir.join("replay.sh"); std::fs::write( &cli, - format!("#!/bin/sh\nread -r _first || exit 1\ncat '{}'\n", frames.display()), + format!( + "#!/bin/sh\nread -r _first || exit 1\ncat '{}'\n", + frames.display() + ), ) .expect("replayer written"); #[cfg(unix)] diff --git a/crates/harness/tests/codex.rs b/crates/harness/tests/codex.rs index 1d732bec10..82271356c9 100644 --- a/crates/harness/tests/codex.rs +++ b/crates/harness/tests/codex.rs @@ -44,6 +44,7 @@ fn request(prompt: &str) -> RunRequest { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, } } @@ -692,12 +693,14 @@ async fn child_thread_routing_tags_and_never_settles_parent() { // parent delta that follows it in the script (not only at thread/closed). let child_done = events .iter() - .position(|e| matches!( - e, - AgentEvent::Subagent { parent_tool_use_id, event } - if parent_tool_use_id == "call_alpha" - && matches!(event.as_ref(), AgentEvent::Done { .. }) - )) + .position(|e| { + matches!( + e, + AgentEvent::Subagent { parent_tool_use_id, event } + if parent_tool_use_id == "call_alpha" + && matches!(event.as_ref(), AgentEvent::Done { .. }) + ) + }) .expect("tagged done"); let late_parent_delta = events .iter() diff --git a/crates/harness/tests/cursor.rs b/crates/harness/tests/cursor.rs index 1ebfbce9ea..f77c25245f 100644 --- a/crates/harness/tests/cursor.rs +++ b/crates/harness/tests/cursor.rs @@ -38,6 +38,7 @@ fn request(prompt: &str) -> RunRequest { sandbox: SandboxLevel::DangerFullAccess, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, } } diff --git a/crates/harness/tests/managed_install.rs b/crates/harness/tests/managed_install.rs index 3dbd576bb2..8a20945068 100644 --- a/crates/harness/tests/managed_install.rs +++ b/crates/harness/tests/managed_install.rs @@ -43,6 +43,7 @@ async fn managed_install_reaches_session_started() { sandbox: zeron_proto::SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, }; diff --git a/crates/harness/tests/managed_install_failure.rs b/crates/harness/tests/managed_install_failure.rs index a63952b2eb..54483a1c09 100644 --- a/crates/harness/tests/managed_install_failure.rs +++ b/crates/harness/tests/managed_install_failure.rs @@ -52,6 +52,7 @@ async fn silent_npm_enoent_death_surfaces_decoded_error() { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, }; @@ -65,8 +66,5 @@ async fn silent_npm_enoent_death_surfaces_decoded_error() { assert!(message.contains("exit code 254"), "{message}"); assert!(message.contains("ENOENT"), "{message}"); assert!(message.contains("failed silently"), "{message}"); - assert!( - message.contains("pi-acp"), - "names the package: {message}" - ); + assert!(message.contains("pi-acp"), "names the package: {message}"); } diff --git a/crates/harness/tests/real_quiet_survey.rs b/crates/harness/tests/real_quiet_survey.rs index 8a0f7118e5..05fca6b232 100644 --- a/crates/harness/tests/real_quiet_survey.rs +++ b/crates/harness/tests/real_quiet_survey.rs @@ -68,6 +68,7 @@ async fn probe_once(harness: AcpHarness) -> ProbeOutcome { sandbox: SandboxLevel::WorkspaceWrite, auto_approve: true, attachments: Vec::new(), + worktree: None, resume: None, }; let mut stream = match harness.run(req, controls).await { diff --git a/crates/proto/src/agent.rs b/crates/proto/src/agent.rs index 45341421bb..b1ec984d26 100644 --- a/crates/proto/src/agent.rs +++ b/crates/proto/src/agent.rs @@ -111,6 +111,26 @@ pub struct RunRequest { /// content blocks. Additive + serde-defaulted for wire compat. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub attachments: Vec, + /// Host-side isolated-worktree creation (see [`WorktreeSpec`]): when set, + /// the HOST materializes the worktree at command-drain time and runs there + /// instead of `cwd`. Additive + serde-defaulted for wire compat — an old + /// host ignores it and runs in `cwd` (the repo's main checkout). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worktree: Option, +} + +/// Isolated-worktree directive riding [`RunRequest`]. The worktree is created +/// by the HOST while draining the queued Run — not by the sender over a +/// blocking CreateWorktree RPC — so the send path stays durable: a lost relay +/// frame can't wedge the composer on "Sending…" while the session runs anyway +/// (2026-08-18 user report). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct WorktreeSpec { + /// The repo whose worktree to create (the space's folder on the host). + pub repo_path: String, + /// Base ref the fresh `zeron/` branch is created off. + pub base: String, } /// The session-scoped singleton id for the live plan/todo chip. ACP plan @@ -369,6 +389,29 @@ mod tests { assert_eq!(round.attachments, vec!["/tmp/a.png".to_string()]); } + #[test] + fn run_request_worktree_default_and_round_trip() { + // Old-wire JSON without the field parses (additive compat)… + let old = r#"{"prompt":"p","model":null,"reasoning":null,"cwd":".","sandbox":"workspace-write","resume":null}"#; + let req: RunRequest = serde_json::from_str(old).unwrap(); + assert!(req.worktree.is_none()); + // …and `None` serializes away (old readers never see it). + let json = serde_json::to_value(&req).unwrap(); + assert!(json.get("worktree").is_none()); + // A populated spec round-trips camelCased. + let req = RunRequest { + worktree: Some(WorktreeSpec { + repo_path: "/repos/comet".into(), + base: "main".into(), + }), + ..req + }; + let json = serde_json::to_value(&req).unwrap(); + assert_eq!(json["worktree"]["repoPath"], "/repos/comet"); + let round: RunRequest = serde_json::from_value(json).unwrap(); + assert_eq!(round.worktree, req.worktree); + } + #[test] fn harness_id_uses_kebab_case() { assert_eq!( diff --git a/crates/sync/src/chat_client.rs b/crates/sync/src/chat_client.rs index dfbb578ed3..4b1f577c87 100644 --- a/crates/sync/src/chat_client.rs +++ b/crates/sync/src/chat_client.rs @@ -126,8 +126,11 @@ pub trait CheckpointFetcher: Send + Sync + 'static { /// Both are safe at-least-once: batchId dedupe and Loro re-import no-ops. pub trait ChatTransport: Send + Sync + 'static { fn fetch_rows(&self, after: u64) -> BoxFuture<'static, Result, SyncError>>; - fn push(&self, batch_id: String, bytes: Vec) - -> BoxFuture<'static, Result>; + fn push( + &self, + batch_id: String, + bytes: Vec, + ) -> BoxFuture<'static, Result>; } // ── catch-up planning (pure — the client-side precision rule) ─────────────── @@ -413,8 +416,16 @@ impl ChatClient { initial_cursor: u64, tuning: ChatTuning, ) -> Result { - Self::connect_with_transport(connector, sink, fetcher, device_id, initial_cursor, tuning, None) - .await + Self::connect_with_transport( + connector, + sink, + fetcher, + device_id, + initial_cursor, + tuning, + None, + ) + .await } pub(crate) async fn connect_with_transport( @@ -1187,16 +1198,13 @@ impl Actor { serde_json::from_value::(state_frame.header.clone()) { lock(&shared).server = Some(state); - let contained = state.checkpoint_size == 0 - || sink.contains_frontier(&state_frame.payload); + let contained = + state.checkpoint_size == 0 || sink.contains_frontier(&state_frame.payload); if let CatchUpPlan::CheckpointThenRows { after } = plan_catch_up(cursor, &state, contained) { - let fetched = tokio::time::timeout( - CHECKPOINT_FETCH_DEADLINE, - fetcher.fetch(), - ) - .await; + let fetched = + tokio::time::timeout(CHECKPOINT_FETCH_DEADLINE, fetcher.fetch()).await; match fetched { Ok(Ok(bytes)) => { if sink.apply_checkpoint(&bytes, state.checkpoint_seq).is_err() { diff --git a/crates/sync/src/chat_client/tests.rs b/crates/sync/src/chat_client/tests.rs index 7f93ffbade..7cebab0704 100644 --- a/crates/sync/src/chat_client/tests.rs +++ b/crates/sync/src/chat_client/tests.rs @@ -788,7 +788,13 @@ async fn checkpoint_fetch_overlaps_row_backfill() { b"r7", ) .await; - send(&end, frame_type::ROWS_DONE, serde_json::json!({"headSeq": 7}), &[]).await; + send( + &end, + frame_type::ROWS_DONE, + serde_json::json!({"headSeq": 7}), + &[], + ) + .await; // Only now does the "download" complete. let _ = gate_tx.send(()); end @@ -849,7 +855,13 @@ async fn pending_push_flushes_before_backfill_completes() { &[], ) .await; - send(&end_b, frame_type::ROWS_DONE, serde_json::json!({"headSeq": 1}), &[]).await; + send( + &end_b, + frame_type::ROWS_DONE, + serde_json::json!({"headSeq": 1}), + &[], + ) + .await; end_b }); @@ -871,7 +883,11 @@ async fn pending_push_flushes_before_backfill_completes() { let _ = events.recv().await; } let _end = server.await.unwrap(); - assert_eq!(client.stats().cursor, 1, "early replay acked before ROWS_DONE"); + assert_eq!( + client.stats().cursor, + 1, + "early replay acked before ROWS_DONE" + ); client.shutdown().await; } diff --git a/crates/ui/src/composer.rs b/crates/ui/src/composer.rs index c6955333cc..ef8602ac93 100644 --- a/crates/ui/src/composer.rs +++ b/crates/ui/src/composer.rs @@ -4503,9 +4503,6 @@ impl Composer { }; let space_id = space.as_ref().map(|s| s.id.clone()); let space_path = space.as_ref().map(|s| s.path.clone()); - let space_remote = space - .as_ref() - .is_some_and(|s| local_device_id.as_deref() != Some(s.device_id.as_str())); // Snapshot-and-clear NOW (use-attachments.ts takeAttachments): the // strip empties the instant you hit send; a failure hands the files // back into the chat's stash. @@ -4609,6 +4606,12 @@ impl Composer { } .unwrap_or_else(|| ".".to_string()); let mut worktree_cwd: Option = None; + // Fresh-worktree plans ride the QUEUED Run command (a + // WorktreeSpec the HOST materializes at drain time) instead of + // a blocking CreateWorktree relay RPC here: the RPC had no + // timeout, so a lost relay frame wedged the send on "Sending…" + // forever while the session ran remotely anyway (2026-08-18). + let mut run_worktree: Option = None; // The picked ref rides createChat so the session footer names // it from the first frame (it read "Select ref" until the // host's diff reconciler got around to stamping the branch). @@ -4624,29 +4627,17 @@ impl Composer { chat_branch = Some(branch.clone()); } crate::pickers::CheckoutPlan::NewWorktree { base } => { + // Footer shows the base until the host stamps the + // actual zeron/ branch post-creation. cwd + // stays the repo folder — an old host that doesn't + // know the spec degrades to the main checkout + // instead of failing the run. chat_branch = base.clone(); if let (Some(repo_path), Some(base)) = (&space_path, base) { - let mut params = serde_json::json!({ - "repoPath": repo_path, - "branch": base, + run_worktree = Some(zeron_proto::WorktreeSpec { + repo_path: repo_path.clone(), + base: base.clone(), }); - if space_remote - && let Some(object) = params.as_object_mut() - { - object.insert( - "targetDeviceId".into(), - serde_json::Value::String(device_id.clone()), - ); - } - let value = engine - .client() - .call(methods::CREATE_WORKTREE, params) - .await - .map_err(|e| format!("Worktree failed: {e}"))?; - let worktree: zeron_proto::Worktree = serde_json::from_value(value) - .map_err(|e| format!("Worktree reply malformed: {e}"))?; - cwd = worktree.path.clone(); - worktree_cwd = Some(worktree.path); } } } @@ -4783,6 +4774,7 @@ impl Composer { auto_approve: false, resume: None, attachments: attachment_paths, + worktree: run_worktree, }, message_id: message_id.clone(), }