diff --git a/crates/doc/src/registry.rs b/crates/doc/src/registry.rs index 9761ee0eaf..62ec7a6c83 100644 --- a/crates/doc/src/registry.rs +++ b/crates/doc/src/registry.rs @@ -540,19 +540,36 @@ impl RegistryDoc { self.clock.next(Self::now_ms(), &device) } - fn enqueue_ops(&mut self, ops: Vec) { + fn enqueue_ops(&mut self, mut ops: Vec) { if ops.is_empty() { return; } + // The registry room rejects any batch over its op cap (500), and a + // rejected batch is a PERMANENT wedge: error frames carry no batch + // id, so the client can never retire it — it replays and fails on + // every reconnect, and every write queued behind it (session-status + // rows included) never propagates again. `delete_space` on a + // ≥250-chat space minted exactly such a batch. Chunk here so no + // writer can ever produce one; chunks apply in order, each + // atomically (only cross-chunk atomicity is given up). + const MAX_OPS_PER_BATCH: usize = 400; self.generation += 1; - // Batch ids only need device-lifetime uniqueness; the HLC of a fresh - // tick provides exactly that without a rng dependency. - let batch = format!("b-{}", self.next_hlc()); - self.pending.push(PendingBatch { - batch, - ops, - in_flight: false, - }); + while !ops.is_empty() { + let tail = if ops.len() > MAX_OPS_PER_BATCH { + ops.split_off(MAX_OPS_PER_BATCH) + } else { + Vec::new() + }; + // Batch ids only need device-lifetime uniqueness; the HLC of a + // fresh tick provides exactly that without a rng dependency. + let batch = format!("b-{}", self.next_hlc()); + self.pending.push(PendingBatch { + batch, + ops, + in_flight: false, + }); + ops = tail; + } } fn write(&mut self, kind: &str, id: &str, op: OpKind, set: BTreeMap) { diff --git a/crates/engine/src/chat2_host.rs b/crates/engine/src/chat2_host.rs index bb9c2a7e6a..eb7fe1c088 100644 --- a/crates/engine/src/chat2_host.rs +++ b/crates/engine/src/chat2_host.rs @@ -28,15 +28,21 @@ pub const CHAT2_DOC_EPOCH: u32 = 2; /// the existing change plumbing — this type only owns import + same-tx /// persistence. pub struct EngineChatSink { - doc: Arc, + /// WEAK: the sink lives inside the handle's `ChatClient` for the + /// client's whole life — a strong ref here kept + /// `Arc::strong_count(&handle.doc) > 1` permanently, which reads as + /// "live writer" to `pinned()` and made every chat2 handle immune to + /// LRU eviction (unbounded warm-doc growth). Callbacks upgrade per + /// call; a dead doc (evicted handle) is a no-op. + doc: std::sync::Weak, store: Arc, chat_id: String, } impl EngineChatSink { - pub fn new(doc: Arc, store: Arc, chat_id: impl Into) -> Self { + pub fn new(doc: &Arc, store: Arc, chat_id: impl Into) -> Self { Self { - doc, + doc: Arc::downgrade(doc), store, chat_id: chat_id.into(), } @@ -44,7 +50,10 @@ impl EngineChatSink { /// Export the CURRENT doc and persist it with `cursor` in one tx. fn persist_with_cursor(&self, cursor: u64) { - match self.doc.export_snapshot() { + let Some(doc) = self.doc.upgrade() else { + return; + }; + match doc.export_snapshot() { Ok(bytes) => { if let Err(err) = self.store.save_snapshot_with_cursor( &self.chat_id, @@ -66,7 +75,10 @@ impl EngineChatSink { impl ChatDocSink for EngineChatSink { fn apply_row(&self, bytes: &[u8], cursor: u64) { - if let Err(err) = self.doc.doc().import(bytes) { + let Some(doc) = self.doc.upgrade() else { + return; + }; + if let Err(err) = doc.doc().import(bytes) { // Malformed remote bytes cost the row, never the doc (the same // skip-not-fail rule as transcript reads). The cursor still // advances: replaying a poison row forever is the wedge class. @@ -77,8 +89,8 @@ impl ChatDocSink for EngineChatSink { } fn apply_checkpoint(&self, bytes: &[u8], cursor: u64) -> Result<(), String> { - self.doc - .doc() + let doc = self.doc.upgrade().ok_or("doc evicted")?; + doc.doc() .import(bytes) .map_err(|e| format!("checkpoint import: {e}"))?; self.persist_with_cursor(cursor); @@ -86,6 +98,9 @@ impl ChatDocSink for EngineChatSink { } fn contains_frontier(&self, frontier: &[u8]) -> bool { + let Some(doc) = self.doc.upgrade() else { + return true; // evicted: claim contained so the client idles, not refetches + }; if frontier.is_empty() { return true; } @@ -95,7 +110,7 @@ impl ChatDocSink for EngineChatSink { // merge), never silently skips history. return false; }; - self.doc.doc().oplog_vv().includes_vv(&vv) + doc.doc().oplog_vv().includes_vv(&vv) } fn advance_cursor(&self, cursor: u64) { @@ -134,6 +149,7 @@ impl CheckpointFetcher for EdgeCheckpointFetcher { ); Box::pin(async move { let mut got: Vec = Vec::new(); + let mut seen_seq: Option = None; // Range-resume loop: each attempt continues at the byte where // the last one stopped. Attempt count bounds a flapping link; // the ChatClient's own deadline bounds wall clock. @@ -153,6 +169,26 @@ impl CheckpointFetcher for EdgeCheckpointFetcher { continue; } }; + // Resume validator: a NEW checkpoint can commit between + // attempts, and a Range against it would splice two different + // blobs (the import fails and burns a whole redial cycle). + // The DO stamps every response with the checkpoint's seq — + // on change, restart the download from byte 0. + let seq = res + .headers() + .get("x-chat2-checkpoint-seq") + .and_then(|v| v.to_str().ok()) + .map(str::to_string); + if seq.is_some() && seen_seq.is_some() && seq != seen_seq { + tracing::info!(resumed_at = got.len(), + "chat2 checkpoint replaced mid-download; restarting from 0"); + got.clear(); + seen_seq = seq; + continue; + } + if seq.is_some() { + seen_seq = seq; + } match res.status().as_u16() { 200 => got.clear(), 206 => {} diff --git a/crates/engine/src/doc_host.rs b/crates/engine/src/doc_host.rs index 7d215e747c..ffc3e14883 100644 --- a/crates/engine/src/doc_host.rs +++ b/crates/engine/src/doc_host.rs @@ -243,6 +243,9 @@ impl ChatDocHandle { if let Some(room) = lock(&self.room).as_ref() { room.probe(); } + if let Some(chat2) = lock(&self.chat2).as_ref() { + chat2.probe(); + } // Subscribe BEFORE the dirty check: a commit racing this attach then // sees a live receiver and publishes, instead of re-marking dirty // after our refresh and leaving the new watcher a cleared mirror. @@ -380,7 +383,55 @@ impl DocHost { /// Wire the workspace host (engine assembly) — the source of chat-ownership rows. pub fn set_workspace(&self, workspace: WorkspaceHost) { - let _ = self.inner.workspace.set(workspace); + let chats = workspace.watch_chats(); + if self.inner.workspace.set(workspace).is_ok() { + self.spawn_cutover_watcher(chats); + } + } + + /// Live cutover convergence: when a registry chat row flips to roomGen 2 + /// while this device holds an s2-mode handle, retire it NOW — a viewer + /// watching the chat at flip time otherwise stays frozen on the dead s2 + /// room until they happen to reopen (the host writes only to chat2 from + /// the flip on). Dropping the handle ends the watch streams cleanly; the + /// UI's transcript/standing watches resubscribe and the fresh open takes + /// the chat2 adopt path. A handle with a LIVE local writer (a running + /// turn's doc ref) is left alone — the host never flips mid-run, and a + /// racing writer must never lose its doc out from under it. + fn spawn_cutover_watcher(&self, mut chats: watch::Receiver>) { + let host = self.clone(); + tokio::spawn(async move { + loop { + if chats.changed().await.is_err() { + return; // workspace host gone (shutdown) + } + let flipped: Vec = chats + .borrow_and_update() + .iter() + .filter(|c| c.room_gen.unwrap_or(1) >= 2) + .map(|c| c.id.clone()) + .collect(); + if flipped.is_empty() { + continue; + } + let mut handles = lock(&host.inner.handles); + for chat_id in flipped { + let Some(handle) = handles.get(&chat_id) else { + continue; + }; + if lock(&handle.chat2_local_sub).is_some() { + continue; // already chat2-mode + } + handle.retired.store(true, Ordering::Relaxed); + let live_writer = Arc::strong_count(&handle.doc) > 1; + if !live_writer { + handles.remove(&chat_id); + tracing::info!(chat = %chat_id, + "s2 handle dropped on chat2 cutover; watchers resubscribe onto the new room"); + } + } + } + }); } /// The workspace host, once wired (tests may assemble a DocHost without one). @@ -395,9 +446,25 @@ impl DocHost { /// Open (or return) the chat's doc handle: load the local snapshot (or init fresh), /// start the change-driven task, and join the edge room when configured. pub fn open(&self, chat_id: &str) -> Result, EngineError> { + // The registry names the sync room generation (docs/chat2-sync.md + // M2): absent row / absent field = legacy s2. Read it BEFORE the + // cached-handle check — a cached s2-mode handle for a chat another + // device has since cut over to chat2 would otherwise serve its frozen + // fat lineage forever (the host writes only to the chat2 room now; + // this device's s2 room has gone permanently silent). + let chat_row = self + .workspace() + .and_then(|w| w.chat(chat_id).ok().flatten()); + let registry_gen = chat_row.as_ref().and_then(|c| c.room_gen).unwrap_or(1); { let mut handles = lock(&self.inner.handles); if let Some(handle) = handles.get(chat_id) { + // s2-mode marker: the chat2 local-update subscription is + // installed before the chat2 dial, so its absence means this + // handle was built for the legacy room. + if registry_gen >= 2 && lock(&handle.chat2_local_sub).is_none() { + handle.retired.store(true, Ordering::Relaxed); + } if handle.retired.load(Ordering::Relaxed) && !self.pinned(handle) { // A seed flipped this chat under a cached fat handle // (review B1): drop it so this open converges onto the @@ -409,13 +476,6 @@ impl DocHost { } } } - // The registry names the sync room generation (docs/chat2-sync.md - // M2): absent row / absent field = legacy s2. Read it BEFORE the doc - // load — chat2 chats adopt (M3) and dial differently. - let chat_row = self - .workspace() - .and_then(|w| w.chat(chat_id).ok().flatten()); - let registry_gen = chat_row.as_ref().and_then(|c| c.room_gen).unwrap_or(1); // B2/M5 guard: the LOCAL epoch is the second cutover signal. A crash // between the thin save and the registry flip (or a not-yet-synced // registry) must NOT route an epoch-2 doc back onto s2 — the s2 @@ -510,12 +570,6 @@ impl DocHost { None => SessionDoc::init(chat_id)?, } }; - // Re-queue survives the adopt: our own pending commands become fresh - // entries in the new lineage (the processed_commands ledger still - // guards against double execution regardless). - for command in &requeue_commands { - let _ = doc.queue_command(command); - } let doc = Arc::new(doc); let (changed_tx, changed_rx) = watch::channel(0u64); @@ -583,6 +637,16 @@ impl DocHost { true })); *lock(&handle.chat2_local_sub) = Some(sub); + // Re-queue survives the adopt: our own pending commands + // become fresh entries in the new lineage (the + // processed_commands ledger still guards double execution). + // Committed AFTER the local-update subscription above — a + // commit before it never enters the pending buffer or the + // client, so the requeued command would sit in the local doc + // and never reach the room (the host would never see it). + for command in &requeue_commands { + let _ = doc.queue_command(command); + } self.spawn_chat2_join(edge.clone(), &handle, chat2_cursor); } else { // Host-side lazy migration (M1/M2): seeding runs in the @@ -596,9 +660,15 @@ impl DocHost { // Quiescent-only (review B1): a seed under a live run or // watched transcript would strand everything written // after the rebuild instant in a retired fat lineage. - // This open is brand new — nothing watches yet; the - // pinned re-check happens again right before the flip. - self.spawn_chat2_seed(edge.clone(), chat_id, doc.clone()); + // Deferred until the s2 room has JOINED and the doc has + // gone quiet: seeding straight from open() rebuilt from + // the host's LOCAL doc before the room backfill landed — + // rows contributed by other devices (a queued command, a + // phone steer) forked away into the retired lineage, and + // a nudge-triggered open seeded+dropped the handle out + // from under the very command it was opened to execute + // (e2e smoke, 2026-08-10). + self.spawn_chat2_seed_when_quiet(edge.clone(), chat_id, &handle); } self.spawn_s2_join(edge, chat_id, &doc, &handle); } @@ -688,12 +758,13 @@ impl DocHost { let http = self.inner.http.clone(); let device = self.inner.config.device_id.clone(); let weak = Arc::downgrade(handle); + let host = self.clone(); tokio::spawn(async move { - let sink = Arc::new(crate::chat2_host::EngineChatSink::new( - doc.clone(), - store, - chat.clone(), - )); + let sink = Arc::new(crate::chat2_host::EngineChatSink::new(&doc, store, chat.clone())); + // The sink holds only a Weak doc ref (a strong one made every + // chat2 handle read as perma-pinned — LRU eviction dead); this + // task's own strong ref dies when the join resolves. + drop(doc); let fetcher = Arc::new(crate::chat2_host::EdgeCheckpointFetcher::new( http, edge.clone(), @@ -722,6 +793,7 @@ impl DocHost { let Some(handle) = weak.upgrade() else { return; // evicted mid-dial: drop leaves the room }; + let mut events = client.events(); { // Store + drain under ONE client-lock critical // section: the subscription pushes to the buffer @@ -737,6 +809,35 @@ impl DocHost { *client_slot = Some(client); } tracing::info!(chat = %chat, "chat2 room joined (converged)"); + // Host recovery duties (C3): a wiped room needs a + // seed checkpoint or fresh readers see only + // post-reset rows; rejected pushes reach peers only + // through a checkpoint. Watcher dies with the handle. + if host.is_host(&chat) { + let host = host.clone(); + let weak = weak.clone(); + let chat = chat.clone(); + tokio::spawn(async move { + use comet_sync::chat_client::ChatEvent; + loop { + match events.recv().await { + Ok(ChatEvent::ServerReset) => { + let Some(handle) = weak.upgrade() else { return }; + tracing::warn!(chat = %chat, "chat2 room reset; posting seed checkpoint"); + host.spawn_chat2_checkpoint(&handle, "server-reset"); + } + Ok(ChatEvent::PushRejected) => { + let Some(handle) = weak.upgrade() else { return }; + tracing::warn!(chat = %chat, "chat2 push rejected; compensating via checkpoint"); + host.spawn_chat2_checkpoint(&handle, "push-rejected"); + } + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {} + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, + } + } + }); + } return; } Ok(Err(err)) => { @@ -766,6 +867,68 @@ impl DocHost { /// the seed checkpoint, persist the thin lineage locally, THEN flip the /// registry — each step idempotent, a crash before the flip leaves the /// chat on s2 and the next open retries (M5). + /// Defer a chat2 seed until the chat is verifiably quiet: the s2 room has + /// joined (the rebuild must include every row the room holds — seeding + /// from the pre-backfill local doc forked other devices' rows into the + /// retired lineage) and the doc frontier has stopped moving (a + /// nudge-burst in flight — an incoming queued command — must land and + /// execute first). Holds only a weak handle: eviction ends the wait. + fn spawn_chat2_seed_when_quiet( + &self, + edge: EdgeConfig, + chat_id: &str, + handle: &Arc, + ) { + const TICK: std::time::Duration = std::time::Duration::from_millis(500); + const JOIN_WAIT_TICKS: u32 = 120; // ≤60s for the room; offline stays s2 + const QUIET_TICKS: u32 = 4; // 2s of frontier silence + let host = self.clone(); + let chat = chat_id.to_string(); + let weak = Arc::downgrade(handle); + tokio::spawn(async move { + let mut ticks = 0u32; + loop { + { + let Some(handle) = weak.upgrade() else { return }; + if handle.retired.load(Ordering::Relaxed) { + return; // another path already seeded + } + if lock(&handle.room).is_some() { + break; + } + } + ticks += 1; + if ticks >= JOIN_WAIT_TICKS { + tracing::debug!(chat = %chat, "chat2 seed skipped: s2 room never joined"); + return; + } + tokio::time::sleep(TICK).await; + } + let mut quiet = 0u32; + let mut last_vv: Option> = None; + loop { + tokio::time::sleep(TICK).await; + let Some(handle) = weak.upgrade() else { return }; + if handle.retired.load(Ordering::Relaxed) { + return; + } + let vv = handle.doc.doc().oplog_vv().encode(); + if last_vv.as_ref() == Some(&vv) { + quiet += 1; + if quiet >= QUIET_TICKS { + let doc = handle.doc.clone(); + drop(handle); + host.spawn_chat2_seed(edge, &chat, doc); + return; + } + } else { + quiet = 0; + last_vv = Some(vv); + } + } + }); + } + fn spawn_chat2_seed(&self, edge: EdgeConfig, chat_id: &str, doc: Arc) { { let mut seeding = lock(&self.inner.seeding); @@ -797,7 +960,14 @@ impl DocHost { doc: Arc, ) -> Result<(), String> { use base64::Engine as _; + let vv_at_rebuild = doc.doc().oplog_vv().encode(); let rebuilt = comet_doc::rebuild::rebuild_thin_doc(&doc).map_err(|e| e.to_string())?; + // The seed's own doc ref must be gone before the pinned re-check + // below: `pinned` reads `Arc::strong_count(&handle.doc) > 1`, and + // holding this clone made that true unconditionally — every seed + // aborted "became active mid-seed" and the cutover never flipped a + // single chat (v0.1.32 DOA). + drop(doc); if !rebuilt.sidecar.is_empty() { // Sidecar PARKED (docs/chat2-sync.md A2): full outputs are not // uploaded; they survive in the rollback snapshot + run journal. @@ -836,10 +1006,18 @@ impl DocHost { // quiet open retries. { let handles = lock(&self.inner.handles); - if let Some(handle) = handles.get(chat_id) - && self.pinned(handle) - { - return Err("chat became active mid-seed; aborted before flip".into()); + if let Some(handle) = handles.get(chat_id) { + if self.pinned(handle) { + return Err("chat became active mid-seed; aborted before flip".into()); + } + // Frontier seal (review B1's TOCTOU): ANY doc movement since + // the rebuild — a synced remote row, a local write that has + // already released its doc ref — means the thin lineage is + // missing it. Borrowed read (no Arc clone: that would trip + // the pinned check we just passed). + if handle.doc.doc().oplog_vv().encode() != vv_at_rebuild { + return Err("doc advanced mid-seed; aborted before flip".into()); + } } } // Rollback copy of the fat lineage BEFORE the thin one replaces it — @@ -898,7 +1076,6 @@ impl DocHost { /// - tail sidecar: publish the last-64 transcript JSON for thin/instant /// readers (the iOS fallback path). async fn chat2_maintenance(&self, handle: &Arc) { - use base64::Engine as _; if handle.retired.load(Ordering::Relaxed) { return; } @@ -941,6 +1118,24 @@ impl DocHost { if stats.row_bytes <= 512 * 1024 && stats.row_count <= 200 { return; } + self.spawn_chat2_checkpoint(handle, "threshold"); + } + + /// POST a full checkpoint for a chat2 room (one in flight per handle). + /// Callers: the quiesce-tick threshold above, and the client recovery + /// events (`ServerReset` — a wiped room needs a seed checkpoint or every + /// fresh reader sees only post-reset rows; `PushRejected` — the rejected + /// ops reach peers only through a checkpoint). + fn spawn_chat2_checkpoint(&self, handle: &Arc, reason: &'static str) { + use base64::Engine as _; + let Some(edge) = self.inner.config.edge.clone() else { + return; + }; + let stats = match &*lock(&handle.chat2) { + Some(client) => client.stats(), + None => return, + }; + let chat_id = handle.chat_id.clone(); if handle .checkpointing .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) @@ -981,7 +1176,7 @@ impl DocHost { .await { Ok(res) if res.status().is_success() => { - tracing::info!(chat = %chat_id, seq_covered, "chat2 threshold checkpoint posted"); + tracing::info!(chat = %chat_id, seq_covered, reason, "chat2 checkpoint posted"); if let Some(handle) = weak_note.upgrade() && let Some(client) = &*lock(&handle.chat2) { @@ -1084,6 +1279,12 @@ impl DocHost { if let Some(room) = lock(&handle.room).as_ref() { room.probe(); } + // chat2 rooms verify liveness on the same user signals — a + // deaf-but-ponging DO otherwise freezes a watched transcript + // for the whole background probe quiet window. + if let Some(chat2) = lock(&handle.chat2).as_ref() { + chat2.probe(); + } } } diff --git a/crates/engine/src/sessions.rs b/crates/engine/src/sessions.rs index 4e03229e90..00610c46fb 100644 --- a/crates/engine/src/sessions.rs +++ b/crates/engine/src/sessions.rs @@ -78,6 +78,20 @@ struct RunHandle { cancel: watch::Sender, engine_tx: mpsc::UnboundedSender, pending_inputs: PendingInputs, + /// Steers accepted into the mailbox but not yet confirmed by a `Steered` + /// event — the at-least-once ledger. A run can die with accepted steers + /// still in its mailbox (idle reaper vs. a routed send; a mid-turn error + /// discarding queued boundary steers): the run task drains this at exit + /// and re-dispatches each entry as a fresh turn, so an accepted message + /// can never silently evaporate from a transcript that shows it as sent. + routed_steers: Arc>>, +} + +/// One accepted-but-unconfirmed steer: enough to re-dispatch it verbatim. +#[derive(Debug, Clone)] +struct RoutedSteer { + prompt: String, + message_id: String, } struct Inner { @@ -243,38 +257,63 @@ impl SessionsEngine { chat_id: &str, harness_id: HarnessId, mut request: RunRequest, - message_id: Option, + mut message_id: Option, inject_resume: bool, ) -> Result { - let routed = lock(&self.inner.runs) - .get(chat_id) - .map(|h| (h.run_id.clone(), h.steerable, h.steer_tx.clone())); - if let Some((run_id, steerable, steer_tx)) = routed { + let routed = lock(&self.inner.runs).get(chat_id).map(|h| { + ( + h.run_id.clone(), + h.steerable, + h.steer_tx.clone(), + h.routed_steers.clone(), + ) + }); + if let Some((run_id, steerable, steer_tx, ledger)) = routed { let message = SteerMessage { prompt: request.prompt.clone(), message_id: message_id.clone(), }; - // The run can vanish between the map read and the send (drive_run - // removes the handle, then stamps the final status): re-check the - // registration AFTER the send, or a steer routed into a dying - // mailbox marks the session Working with no run behind it. - if steerable - && steer_tx.try_send(message).is_ok() - && self.is_live(chat_id, &run_id) - { - let user_id = message_id.unwrap_or_else(new_id); + if steerable && steer_tx.try_send(message).is_ok() { + // The run can vanish between the send and here (the idle + // reaper, a parked child death): the ledger entry below is + // the at-least-once guarantee — the run task's exit drain + // re-dispatches any accepted steer no `Steered` confirmed. + let user_id = message_id.clone().unwrap_or_else(new_id); + lock(&ledger).push_back(RoutedSteer { + prompt: request.prompt.clone(), + message_id: user_id.clone(), + }); let handle = self.doc_handle(chat_id)?; handle.write_user_message(&user_id, &request.prompt, now_ms())?; - // Working BEFORE the lastMessageAt bump: both ride the - // workspace doc from this one peer, so causal order makes it - // impossible for an observer to hold [new message, old status] - // — that gap read as unseen-with-no-live-run = a phantom - // "completed" flash on every remote send (2026-07-31). - self.set_status(chat_id, SessionStatus::Working, false); - self.inner.note_message(chat_id, &request.prompt); - return Ok(run_id); + if self.is_live(chat_id, &run_id) { + // Working BEFORE the lastMessageAt bump: both ride the + // workspace doc from this one peer, so causal order makes it + // impossible for an observer to hold [new message, old status] + // — that gap read as unseen-with-no-live-run = a phantom + // "completed" flash on every remote send (2026-07-31). + self.set_status(chat_id, SessionStatus::Working, false); + self.inner.note_message(chat_id, &request.prompt); + return Ok(run_id); + } + // The run died around the send. If its exit drain already + // claimed the entry, that re-dispatch owns the message — + // otherwise reclaim it and fall through to a fresh run. + let reclaimed = { + let mut ledger = lock(&ledger); + let before = ledger.len(); + ledger.retain(|s| s.message_id != user_id); + ledger.len() != before + }; + if !reclaimed { + self.inner.note_message(chat_id, &request.prompt); + return Ok(run_id); + } + // Keep the already-written doc entry's id for the fresh run + // below (write_user_message dedupes by id). + message_id = Some(user_id); } - // Mailbox closed (runtime mid-teardown / non-steering harness): replace it. + // Mailbox closed (runtime mid-teardown / non-steering harness) or + // the routed run died with the message reclaimed: replace it. self.interrupt(chat_id).await?; } @@ -333,6 +372,7 @@ impl SessionsEngine { cancel: cancel_tx, engine_tx, pending_inputs, + routed_steers: Arc::new(Mutex::new(std::collections::VecDeque::new())), }, ); self.set_status(chat_id, SessionStatus::Working, true); @@ -378,8 +418,8 @@ impl SessionsEngine { let target = lock(&self.inner.runs) .get(chat_id) .filter(|h| h.steerable) - .map(|h| h.steer_tx.clone()); - let Some(steer_tx) = target else { + .map(|h| (h.run_id.clone(), h.steer_tx.clone(), h.routed_steers.clone())); + let Some((run_id, steer_tx, ledger)) = target else { return Ok(SteerOutcome::NotSteerable); }; let message = SteerMessage { @@ -389,10 +429,36 @@ impl SessionsEngine { if steer_tx.try_send(message).is_err() { return Ok(SteerOutcome::NotSteerable); } - // Accepted: the steer prompt becomes a user entry immediately (client-minted id). + // Accepted: ledger first (at-least-once across a dying run), then the + // user entry (client-minted id), then Working BEFORE the + // lastMessageAt bump — same causal-order invariant as the dispatch + // route (an observer must never hold [new message, settled status]: + // the phantom "completed" flash, 2026-07-31). let user_id = message_id.unwrap_or_else(new_id); + lock(&ledger).push_back(RoutedSteer { + prompt: prompt.to_string(), + message_id: user_id.clone(), + }); let handle = self.doc_handle(chat_id)?; handle.write_user_message(&user_id, prompt, now_ms())?; + if self.is_live(chat_id, &run_id) { + self.set_status(chat_id, SessionStatus::Working, false); + self.inner.note_message(chat_id, prompt); + return Ok(SteerOutcome::Accepted); + } + // The run died around the send. Exit drain claimed the entry → its + // re-dispatch owns the message; still ours → reclaim and report + // NotSteerable so the executor falls back to a fresh dispatch + // (same message id — the doc entry dedupes). + let reclaimed = { + let mut ledger = lock(&ledger); + let before = ledger.len(); + ledger.retain(|s| s.message_id != user_id); + ledger.len() != before + }; + if reclaimed { + return Ok(SteerOutcome::NotSteerable); + } self.inner.note_message(chat_id, prompt); Ok(SteerOutcome::Accepted) } @@ -668,10 +734,26 @@ impl Inner { started_at: None, updated_at: now, }); + // `started_at` is the elapsed-timer base and must only ever mean + // "this turn". Entering Working from a settled state always + // restamps (a steer into a parked session is a NEW turn — reusing + // the old base showed the previous turn's 30min on send), and a + // settled session drops its base entirely so no later reader can + // resurrect a stale elapsed. + let was_active = matches!( + entry.status, + SessionStatus::Working | SessionStatus::AwaitingInput + ); entry.status = status; entry.updated_at = now; - if fresh_start { - entry.started_at = Some(now); + match status { + SessionStatus::Working if fresh_start || !was_active => { + entry.started_at = Some(now); + } + SessionStatus::Working | SessionStatus::AwaitingInput => {} + SessionStatus::Idle | SessionStatus::Errored => { + entry.started_at = None; + } } let session = entry.clone(); let mut list: Vec = statuses.values().cloned().collect(); @@ -1085,6 +1167,29 @@ async fn drive_run( // Any stream activity proves the run is alive — keep the session's // freshness inside the UI's 45s staleness window (throttled). inner.touch_session(&chat_id); + // The engine's input bridge is the sole authority on input requests: + // it mints the id and parks the resolver BEFORE emitting the event, + // so a legitimate id is always pending here. A harness emitting its + // own copy (a different id no resolver knows) would fold an + // unanswerable twin chip into the doc — and answering the twin would + // never resume the run. Dropped BEFORE the parked gate below: letting + // it un-park the session and then dropping it would disarm the idle + // reaper with no turn running (a leaked child no reap ever ends). + if let AgentEvent::InputRequested { request_id, .. } = &event { + let pending = lock(&inner.runs) + .get(&chat_id) + .map(|h| h.pending_inputs.clone()); + let known = pending.is_some_and(|p| lock(&p).contains_key(request_id)); + if !known { + tracing::warn!( + chat = %chat_id, + request = %request_id, + "dropping harness-emitted InputRequested (unknown id; \ + the engine input bridge owns this lifecycle)" + ); + continue; + } + } // PARKED: only a steer boundary (or an input round-trip / terminal // Done) re-opens the session. The ACP child keeps forwarding // `session/update` frames after a turn completes — late @@ -1098,11 +1203,26 @@ async fn drive_run( idle_since = None; inner.set_status(&chat_id, SessionStatus::Working, true); } - AgentEvent::InputRequested { .. } - | AgentEvent::InputResolved { .. } - | AgentEvent::Done { .. } => { + AgentEvent::Done { .. } => { idle_since = None; } + // A question with NO turn behind it (post-turn permission + // noise): answer it empty right here. Un-parking into + // AwaitingInput would disarm the idle reaper with no Done + // ever coming — stranded status, leaked warm child. + AgentEvent::InputRequested { request_id, .. } => { + let resolver = lock(&inner.runs) + .get(&chat_id) + .and_then(|h| lock(&h.pending_inputs).remove(request_id)); + if let Some(tx) = resolver { + let _ = tx.send(Vec::new()); + } + tracing::debug!(chat = %chat_id, "parked session: post-turn input request auto-declined"); + continue; + } + // A stale answer settling after its turn already closed: + // nothing is running — stay parked. + AgentEvent::InputResolved { .. } => continue, _ => continue, } } @@ -1126,6 +1246,13 @@ async fn drive_run( .any(|p| matches!(p, MessagePart::Tool { id: pid, .. } if pid == id)) }; match &event { + // The live plan chip is a deliberate singleton (every update + // reuses `LIVE_PLAN_TOOL_ID` so the fold refreshes in place): + // treating its reappearance after a park/steer reset as a stale + // echo dropped the todo list for the rest of the run — from the + // first boundary on, plans never rendered again. + AgentEvent::ToolCall { id, .. } if id == comet_proto::LIVE_PLAN_TOOL_ID => {} + AgentEvent::ToolResult { id, .. } if id == comet_proto::LIVE_PLAN_TOOL_ID => {} AgentEvent::ToolCall { id, .. } => { if !in_segment(&folded, id) && seen_tools.contains(id) { continue; @@ -1211,6 +1338,18 @@ async fn drive_run( dirty = false; entry_id = next_assistant_message_id.clone().unwrap_or_else(new_id); segment_started = now_ms(); + // The elapsed timer is per user message, not per child process: a + // steer boundary restarts it (matches the parked-resume path and + // the composer's optimistic overlay, which already reads 0:00). + inner.set_status(&chat_id, SessionStatus::Working, true); + // The boundary confirms delivery of the oldest accepted steer — + // retire its at-least-once ledger entry. + if let Some(h) = lock(&inner.runs) + .get(&chat_id) + .filter(|h| h.run_id == run_id) + { + lock(&h.routed_steers).pop_front(); + } continue; } @@ -1229,27 +1368,9 @@ async fn drive_run( } => { inner.remember_harness_session(&chat_id, session_id, &run_cwd); } - AgentEvent::InputRequested { request_id, .. } => { - // The engine's input bridge is the sole authority on input - // requests: it mints the id and parks the resolver BEFORE - // emitting the event, so a legitimate id is always pending - // here. A harness emitting its own copy (a different id no - // resolver knows) would fold an unanswerable twin chip into - // the doc — and answering the twin would never resume the - // run. Drop such events. - let pending = lock(&inner.runs) - .get(&chat_id) - .map(|h| h.pending_inputs.clone()); - let known = pending.is_some_and(|p| lock(&p).contains_key(request_id)); - if !known { - tracing::warn!( - chat = %chat_id, - request = %request_id, - "dropping harness-emitted InputRequested (unknown id; \ - the engine input bridge owns this lifecycle)" - ); - continue; - } + AgentEvent::InputRequested { .. } => { + // Known-id guaranteed: the unknown-id twin was dropped above, + // before the parked gate. inner.set_status(&chat_id, SessionStatus::AwaitingInput, false); } AgentEvent::InputResolved { .. } => { @@ -1274,6 +1395,21 @@ async fn drive_run( } if let AgentEvent::Done { status, .. } = &event { + // A question still pending at turn end can never be legitimately + // answered (its turn is over): drain the resolvers NOW, or a late + // `respond_input` finds one, emits InputResolved, and un-parks + // the session into Working with no turn behind it — stranded + // Working, timer forever, reaper disarmed. Empty answers unblock + // the harness-side bridge like an interrupt does. + let pending = lock(&inner.runs) + .get(&chat_id) + .filter(|h| h.run_id == run_id) + .map(|h| h.pending_inputs.clone()); + if let Some(pending) = pending { + for (_, tx) in lock(&pending).drain() { + let _ = tx.send(Vec::new()); + } + } let message_status = match status { DoneStatus::Interrupted => MessageStatus::Aborted, DoneStatus::Completed | DoneStatus::Errored => MessageStatus::Complete, @@ -1344,6 +1480,48 @@ async fn drive_run( } }; + // Claim any accepted-but-unconfirmed steers BEFORE the handle goes away: + // a routed send that raced this exit either finds its entry gone (we own + // it — re-dispatched below) or reclaims it and starts a fresh run itself. + let orphans: Vec = lock(&inner.runs) + .get(&chat_id) + .filter(|h| h.run_id == run_id) + .map(|h| std::mem::take(&mut *lock(&h.routed_steers)).into()) + .unwrap_or_default(); inner.remove_run(&chat_id, &run_id); inner.set_status(&chat_id, final_status, false); + if !interrupted && !orphans.is_empty() { + // The dying run accepted these into its mailbox but never confirmed a + // Steered boundary (idle-reaper race, a mid-turn error discarding + // queued boundary steers, a parked child death). Their user entries + // are already in the transcript — a message that shows as sent must + // never silently not run. Re-dispatch each as a fresh turn + // (write_user_message dedupes by id; resume is engine-injected). + let engine = SessionsEngine { + inner: inner.clone(), + }; + let chat = chat_id.clone(); + tokio::spawn(async move { + for steer in orphans { + let Some(mut request) = engine.last_request(&chat) else { + tracing::warn!(chat = %chat, "orphaned steer lost: no run config to re-dispatch"); + break; + }; + request.prompt = steer.prompt.clone(); + request.resume = None; + request.attachments = Vec::new(); + tracing::info!(chat = %chat, "re-dispatching steer orphaned by a dying run"); + if let Err(err) = engine + .dispatch(&chat, harness_id, request, Some(steer.message_id.clone())) + .await + { + tracing::warn!(chat = %chat, error = %err, "orphaned steer re-dispatch failed"); + engine + .inner + .set_status(&chat, SessionStatus::Errored, false); + break; + } + } + }); + } } diff --git a/crates/engine/tests/e2e.rs b/crates/engine/tests/e2e.rs index 1071ad434b..b0df0a9ce8 100644 --- a/crates/engine/tests/e2e.rs +++ b/crates/engine/tests/e2e.rs @@ -1925,3 +1925,122 @@ async fn stale_tool_echo_after_steer_boundary_does_not_split_text() { "stale echo must not split the streaming text" ); } + +/// The elapsed timer's base (`started_at`) is per user message: a settled +/// session drops it (no reader can resurrect the previous turn's elapsed — +/// the "timer opens at 30:00 on send" bug), and a steer into a PARKED +/// persistent session restamps it fresh for the new turn. +#[tokio::test] +async fn parked_steer_restamps_started_at_and_idle_clears_it() { + // Steerable harness whose stream stays open after the turn's Done — the + // engine parks the session — and whose steering mailbox drives turn two. + struct ParkingHarness; + #[async_trait] + impl Harness for ParkingHarness { + fn id(&self) -> HarnessId { + HarnessId::Mock + } + fn display_name(&self) -> &str { + "Parking" + } + fn supports_steering(&self) -> bool { + true + } + fn steering_mode(&self) -> SteeringMode { + SteeringMode::StepBoundary + } + fn reasoning_levels(&self) -> &[ReasoningLevel] { + &[] + } + async fn models(&self) -> Result, HarnessError> { + Ok(vec![]) + } + async fn run( + &self, + _request: RunRequest, + mut controls: RunControls, + ) -> Result>, HarnessError> { + let (tx, rx) = tokio::sync::mpsc::channel::>(16); + tokio::spawn(async move { + let _ = tx + .send(Ok(AgentEvent::TextDelta { + text: "turn one".into(), + })) + .await; + let _ = tx.send(Ok(done(DoneStatus::Completed))).await; + // Parked. The next steer is turn two. + if let Some(_msg) = controls.steering.recv().await { + let _ = tx + .send(Ok(AgentEvent::Steered { + assistant_message_id: None, + next_assistant_message_id: None, + })) + .await; + let _ = tx + .send(Ok(AgentEvent::TextDelta { + text: "turn two".into(), + })) + .await; + // Hold the turn open so the test's poll observes Working + // (the transition is otherwise sub-millisecond). + tokio::time::sleep(Duration::from_millis(400)).await; + let _ = tx.send(Ok(done(DoneStatus::Completed))).await; + } + }); + Ok(futures::stream::unfold(rx, |mut rx| async move { + rx.recv().await.map(|event| (event, rx)) + }) + .boxed()) + } + } + + let dir = tempfile::tempdir().unwrap(); + let core = assemble(dir.path(), Arc::new(ParkingHarness)); + let handle = core.doc_host.open(CHAT).unwrap(); + queue_as_viewer( + handle.doc(), + "cmd-run-park-timer", + SessionCommandPayload::Run { + request: run_request("first"), + message_id: "m-1".into(), + }, + ); + wait_for( + || core.sessions.session_status(CHAT).map(|s| s.status) == Some(SessionStatus::Idle), + "turn one to park", + ) + .await; + let parked = core.sessions.session_status(CHAT).unwrap(); + assert_eq!( + parked.started_at, None, + "a settled session must drop its timer base" + ); + + let before_steer = chrono::Utc::now(); + queue_as_viewer( + handle.doc(), + "cmd-steer-park-timer", + SessionCommandPayload::Steer { + prompt: "next".into(), + message_id: Some("m-2".into()), + }, + ); + wait_for( + || core.sessions.session_status(CHAT).map(|s| s.status) == Some(SessionStatus::Working), + "turn two working", + ) + .await; + let working = core.sessions.session_status(CHAT).unwrap(); + let started = working.started_at.expect("Working carries a timer base"); + assert!( + started >= before_steer, + "steer into a parked session must restamp started_at (got {started}, steered at {before_steer})" + ); + + wait_for( + || core.sessions.session_status(CHAT).map(|s| s.status) == Some(SessionStatus::Idle), + "turn two to settle", + ) + .await; + assert_eq!(core.sessions.session_status(CHAT).unwrap().started_at, None); +} diff --git a/crates/harness/src/acp/mod.rs b/crates/harness/src/acp/mod.rs index 66a0e1df8f..2b4eae32e9 100644 --- a/crates/harness/src/acp/mod.rs +++ b/crates/harness/src/acp/mod.rs @@ -1642,6 +1642,56 @@ async fn run_session(session: Session) { tokio::select! { res = async { turn.as_mut().expect("guarded by if").await }, if turn.is_some() => { turn = None; + // Settle an in-flight `_session/steering` call BEFORE closing + // the turn: its response rides the same stdout as the prompt + // response, so by now it is (nearly always) already parsed — + // the select just hadn't polled it yet. Deciding it here keeps + // the ordering deterministic: an injection that landed in this + // turn emits its Steered boundary now, ahead of the drained + // tail and the Done (a Steered AFTER Done re-armed the + // consumer with no next turn — the stranded-Working bug); a + // rejected/unsettled call redelivers as the next turn. The + // timeout guards the flooded-incoming edge (reader blocked on + // a full channel never parses the response): past it the call + // is abandoned and the steer redelivered. + if let Some((text, mut fut)) = steering_call.take() { + let outcome = match tokio::time::timeout( + Duration::from_millis(1000), + &mut fut, + ) + .await + { + Ok(Ok(resp)) => resp + .get("outcome") + .and_then(Value::as_str) + .unwrap_or("injected") + .to_owned(), + Ok(Err(_)) | Err(_) => "promptRequired".to_owned(), + }; + if interrupted { + // Winding down; abandoned like any queued steer. + } else if outcome != "promptRequired" { + let (prev, next) = rotate(&mut assistant_message_id); + if !send( + &event_tx, + AgentEvent::Steered { + assistant_message_id: Some(prev), + next_assistant_message_id: Some(next), + }, + ) + .await + { + break 'main; + } + } else { + queued_steers.push_back(text); + } + // Followers waiting on the settled call have no live turn + // to inject into anymore: boundary delivery. + while let Some(next_text) = steer_backlog.pop_front() { + queued_steers.push_back(next_text); + } + } // Updates streamed before the prompt response are already // queued in stdout order — fold them into the turn before // closing it (responses bypass the incoming queue). @@ -1816,17 +1866,63 @@ async fn run_session(session: Session) { // The run is winding down; the steer is abandoned like // any queued steer at interrupt. } else if outcome != "promptRequired" { - let (prev, next) = rotate(&mut assistant_message_id); - if !send( - &event_tx, - AgentEvent::Steered { - assistant_message_id: Some(prev), - next_assistant_message_id: Some(next), - }, - ) - .await - { - break 'main; + // Injected into a live turn → a Steered boundary. But if + // the turn ended while the call was in flight, the + // injection was consumed by THAT turn — its output + // already streamed and the turn's Done already closed the + // segment. Emitting Steered after that Done re-armed the + // consumer (parked session → Working) with no next turn + // and no Done ever coming — the stranded-Working / + // eternal-timer bug. Post-turn: nothing left to do. + if turn.is_some() { + // Pre-injection updates can still sit in `incoming` + // (responses bypass that queue): drain them into the + // CURRENT segment first, or text the agent streamed + // before the injection landed folds after the split — + // the transcript attributes it to the reply-to-steer. + let mut consumer_gone = false; + while let Ok(inc) = incoming.try_recv() { + match inc { + Incoming::Notification { method, params } + if method == "session/update" => + { + for ev in session_update_events(¶ms, &session_id) { + if !send(&event_tx, ev).await { + consumer_gone = true; + break; + } + } + } + Incoming::Request { id, method, params } => { + handle_server_request_live( + &client, + id, + &method, + ¶ms, + &request_input, + ); + } + _ => {} + } + if consumer_gone { + break; + } + } + if consumer_gone { + break 'main; + } + let (prev, next) = rotate(&mut assistant_message_id); + if !send( + &event_tx, + AgentEvent::Steered { + assistant_message_id: Some(prev), + next_assistant_message_id: Some(next), + }, + ) + .await + { + break 'main; + } } } else if turn.is_some() { // Raced the turn end: redeliver at the boundary the @@ -1959,10 +2055,13 @@ async fn run_session(session: Session) { } } - shutdown_child(&mut child, kill_grace).await; + // Escalation dies BEFORE the child is reaped: after `shutdown_child` + // waits the pid, a still-armed SIGTERM/SIGKILL timer would fire at a + // freed (reusable) pid. if let Some(handle) = escalation { handle.abort(); } + shutdown_child(&mut child, kill_grace).await; } #[cfg(test)] diff --git a/crates/harness/src/acp/normalize.rs b/crates/harness/src/acp/normalize.rs index cafcbeb883..493cdd2208 100644 --- a/crates/harness/src/acp/normalize.rs +++ b/crates/harness/src/acp/normalize.rs @@ -255,11 +255,11 @@ pub(crate) fn map_update(update: &Value) -> Vec { // update refresh the same chip (fold refreshes in place by id). vec![ AgentEvent::ToolCall { - id: "acp-plan".into(), + id: comet_proto::LIVE_PLAN_TOOL_ID.into(), call: ToolCall::Todo { items }, }, AgentEvent::ToolResult { - id: "acp-plan".into(), + id: comet_proto::LIVE_PLAN_TOOL_ID.into(), is_error: false, output: None, diff: None, @@ -492,7 +492,7 @@ mod tests { assert_eq!( events[0], AgentEvent::ToolCall { - id: "acp-plan".into(), + id: comet_proto::LIVE_PLAN_TOOL_ID.into(), call: ToolCall::Todo { items: vec![ TodoItem { diff --git a/crates/harness/tests/acp.rs b/crates/harness/tests/acp.rs index 2a34246120..30e8f98000 100644 --- a/crates/harness/tests/acp.rs +++ b/crates/harness/tests/acp.rs @@ -366,6 +366,52 @@ async fn steering_extension_injects_mid_turn() { assert_eq!(dones(&events), vec![(DoneStatus::Completed, None)]); } +/// The steering response racing the turn's own end: the injection landed in +/// the dying turn, and the prompt response reached the wire first. The +/// boundary must still be emitted BEFORE the Done — a Steered after Done +/// re-armed the consumer (parked session → Working) with no next turn and no +/// Done ever coming (the stranded-Working / eternal-timer bug). +#[tokio::test] +async fn steer_racing_the_turn_end_never_emits_steered_after_done() { + let (controls, steer, _token) = controls(); + let harness = harness(); + let stream = harness + .run(request("scenario:steer-race"), controls) + .await + .expect("run starts"); + let events = tokio::time::timeout(Duration::from_secs(10), async move { + let mut events = Vec::new(); + let mut stream = stream; + while let Some(ev) = stream.next().await { + let ev = ev.expect("stream event"); + if matches!(ev, AgentEvent::TextDelta { ref text } if text == "first") { + steer + .send(SteerMessage { + prompt: "redirect please".into(), + message_id: None, + }) + .await + .expect("steer sent"); + } + events.push(ev); + } + events + }) + .await + .expect("run finished in time"); + + assert_eq!(dones(&events), vec![(DoneStatus::Completed, None)], "{events:?}"); + let steered = events + .iter() + .position(|e| matches!(e, AgentEvent::Steered { .. })) + .expect("steer landed in the turn: a Steered boundary must exist"); + let done = events + .iter() + .position(|e| matches!(e, AgentEvent::Done { .. })) + .expect("checked above"); + assert!(steered < done, "Steered after Done strands the session: {events:?}"); +} + #[tokio::test] async fn rejected_steer_queues_and_delivers_at_the_turn_boundary() { let (controls, steer, _token) = controls(); diff --git a/crates/harness/tests/fixtures/fake-acp.sh b/crates/harness/tests/fixtures/fake-acp.sh index 58f3b09375..e1fb837f18 100755 --- a/crates/harness/tests/fixtures/fake-acp.sh +++ b/crates/harness/tests/fixtures/fake-acp.sh @@ -162,6 +162,21 @@ case "$promptline" in fi ;; +*scenario:steer-race*) + update '{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"first"}}' + read -r steerline || exit 1 + sid=$(rid "$steerline") + has "$steerline" '"method":"_session/steering"' || exit 1 + # The exact turn-end race: the injection lands in the turn's tail and the + # PROMPT response hits the wire before the steering response does. The + # harness must settle the steering call at the boundary — Steered before + # Done, never after — instead of stranding the session or re-prompting. + update '{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"steered"}}' + emit "{\"id\":$pid,\"result\":{\"stopReason\":\"end_turn\"}}" + sleep 0.2 + emit "{\"id\":$sid,\"result\":{\"outcome\":\"injected\"}}" + ;; + *scenario:steer-queue*) update '{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"first"}}' read -r steerline || exit 1 diff --git a/crates/proto/src/agent.rs b/crates/proto/src/agent.rs index b6c3f4a49c..efd4df19f4 100644 --- a/crates/proto/src/agent.rs +++ b/crates/proto/src/agent.rs @@ -113,6 +113,14 @@ pub struct RunRequest { pub attachments: Vec, } +/// The session-scoped singleton id for the live plan/todo chip. ACP plan +/// updates carry no wire id; adapters emit every update under this one id so +/// the fold refreshes the same chip in place. Consumers that de-duplicate +/// tool ids across segment boundaries (the engine's stale-echo filter) must +/// EXEMPT this id — it legitimately reappears in every segment for the whole +/// life of a run. +pub const LIVE_PLAN_TOOL_ID: &str = "acp-plan"; + /// A decoded tool invocation, reduced to the fields each kind renders. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "camelCase")] diff --git a/crates/rpc/examples/e2e_driver.rs b/crates/rpc/examples/e2e_driver.rs index 33ce1d0bbc..998e390a99 100644 --- a/crates/rpc/examples/e2e_driver.rs +++ b/crates/rpc/examples/e2e_driver.rs @@ -44,6 +44,11 @@ async fn device_id(client: &RpcClient, label: &str) -> String { } /// Subscribe to a watch-stream and poll items until `predicate` returns `Some`. +/// +/// Stream end is NOT a failure: the engine ends watch streams deliberately at +/// lifecycle boundaries (the chat2 cutover drops the s2 handle so watchers +/// converge onto the new room), and the client contract — the UI does exactly +/// this — is to resubscribe. The step deadline still bounds the whole wait. async fn wait_stream( client: &RpcClient, method: &str, @@ -51,30 +56,42 @@ async fn wait_stream( what: &str, predicate: impl Fn(&serde_json::Value) -> Option, ) -> T { - let mut rx = match client.subscribe(method, params).await { - Ok(rx) => rx, - Err(err) => fail(&format!("{what}: subscribe {method} failed: {err}")), - }; let deadline = Instant::now() + STEP_TIMEOUT; - loop { - let remaining = deadline.saturating_duration_since(Instant::now()); - if remaining.is_zero() { + 'resubscribe: loop { + if deadline.saturating_duration_since(Instant::now()).is_zero() { fail(&format!( "{what}: timed out after {}s", STEP_TIMEOUT.as_secs() )); } - match tokio::time::timeout(remaining, rx.recv()).await { - Ok(Some(item)) => { - if let Some(found) = predicate(&item) { - return found; + let mut rx = match client.subscribe(method, params.clone()).await { + Ok(rx) => rx, + Err(err) => fail(&format!("{what}: subscribe {method} failed: {err}")), + }; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + fail(&format!( + "{what}: timed out after {}s", + STEP_TIMEOUT.as_secs() + )); + } + match tokio::time::timeout(remaining, rx.recv()).await { + Ok(Some(item)) => { + if let Some(found) = predicate(&item) { + return found; + } + } + Ok(None) => { + // Lifecycle boundary (e.g. chat2 cutover): resubscribe. + tokio::time::sleep(Duration::from_millis(300)).await; + continue 'resubscribe; } + Err(_) => fail(&format!( + "{what}: timed out after {}s", + STEP_TIMEOUT.as_secs() + )), } - Ok(None) => fail(&format!("{what}: stream ended early")), - Err(_) => fail(&format!( - "{what}: timed out after {}s", - STEP_TIMEOUT.as_secs() - )), } } } @@ -209,14 +226,26 @@ async fn main() { .unwrap_or_else(|err| fail(&format!("QueueCommand on B: {err}"))); pass("run command queued on B via the doc command queue"); - // 5a. Assistant entry executed by A arrives back on B, complete, with the mock text. + // 5a. Assistant entry executed by A arrives back on B, complete, with the + // mock text. WatchDocMessages speaks the framed protocol: a full `reset` + // list on attach, then `upsert`/`append` deltas (each wrapping an + // `entry`) — collect candidates from both shapes. let (by_device, text) = wait_stream( &b, methods::WATCH_DOC_MESSAGES, serde_json::json!({ "chatId": chat_id }), "assistant transcript on B", |item| { - let entry = item.as_array()?.iter().find(|entry| { + let mut candidates: Vec<&serde_json::Value> = Vec::new(); + if let Some(reset) = item.get("reset").and_then(|v| v.as_array()) { + candidates.extend(reset.iter()); + } + for key in ["upsert", "append"] { + if let Some(deltas) = item.get(key).and_then(|v| v.as_array()) { + candidates.extend(deltas.iter().filter_map(|d| d.get("entry"))); + } + } + let entry = candidates.into_iter().find(|entry| { entry.get("role").and_then(|v| v.as_str()) == Some("assistant") && entry.get("status").and_then(|v| v.as_str()) == Some("complete") })?; diff --git a/crates/ui/src/shell.rs b/crates/ui/src/shell.rs index 6f706a25c3..a606a21c1f 100644 --- a/crates/ui/src/shell.rs +++ b/crates/ui/src/shell.rs @@ -3069,10 +3069,18 @@ impl Shell { return strip.into_any_element(); }; let indicator = state.indicator_for(&chat_id, now); - let elapsed_secs = state + // Timer base: the freshest of the session row's turn start and the + // in-flight send. During the send→ack window the row (if any) still + // carries the PREVIOUS turn's start, and using it opened the timer at + // the old turn's elapsed instead of 0:00. + let started = state .session_for(&chat_id) .and_then(|s| s.started_at) - .map(|t| now.signed_duration_since(t).num_seconds()) + .into_iter() + .chain(state.pending_send_started(&chat_id, now)) + .max(); + let elapsed_secs = started + .map(|t| now.signed_duration_since(t).num_seconds().max(0)) .unwrap_or(0); let sending = self.composer.read(cx).is_sending(); diff --git a/crates/ui/src/state.rs b/crates/ui/src/state.rs index 3108d1b320..9a0ddb6779 100644 --- a/crates/ui/src/state.rs +++ b/crates/ui/src/state.rs @@ -579,6 +579,18 @@ impl AppState { }) } + /// When the in-flight send (if any, inside the TTL) was fired — the + /// elapsed-timer base while the overlay reads as Working. The session + /// row's `started_at` still belongs to the PREVIOUS turn during this + /// window, and showing it made a fresh send open at the old turn's + /// half-hour mark. + pub fn pending_send_started(&self, chat_id: &str, now: DateTime) -> Option> { + self.pending_sends + .get(chat_id) + .filter(|p| now.signed_duration_since(p.started).num_milliseconds() <= PENDING_SEND_TTL_MS) + .map(|p| p.started) + } + /// The host executed the queued command iff the sent message's id showed /// up in the transcript (it writes the message before — causally with — /// the Working status; sessions.rs dispatch paths). @@ -885,43 +897,59 @@ impl AppState { /// on the first frame when nothing is selected yet (manual selection wins). fn spawn_chats_watch(cx: &mut Context, handle: EngineHandle) -> Task<()> { cx.spawn(async move |this, cx| { - let mut rx = match handle - .client() - .subscribe(methods::WATCH_CHATS, serde_json::json!({})) - .await - { - Ok(rx) => rx, - Err(err) => { - tracing::debug!(error = %err, "chats watch unavailable"); - return; - } - }; - while let Some(value) = rx.recv().await { - let parsed: Vec = match serde_json::from_value(value) { - Ok(parsed) => parsed, + // Resubscribe loop (same contract as the transcript watch): a daemon + // restart or RPC drop ends the stream, and a bare return here froze + // the sidebar until app restart — new chats, renames and archives + // from every device silently stopped arriving. + const RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(2); + loop { + let mut rx = match handle + .client() + .subscribe(methods::WATCH_CHATS, serde_json::json!({})) + .await + { + Ok(rx) => rx, Err(err) => { - tracing::warn!(error = %err, "dropping malformed chats frame"); + tracing::debug!(error = %err, "chats watch unavailable; retrying"); + if this.update(cx, |_, _| {}).is_err() { + return; + } + cx.background_executor().timer(RETRY_DELAY).await; continue; } }; - let alive = this.update(cx, |state, cx| { - state.apply_chats(parsed); - if state.selected_chat.is_none() && !state.auto_selected { - let most_recent = state - .chats - .iter() - .find(|c| !c.archived) - .map(|c| c.id.clone()); - if let Some(chat_id) = most_recent { - state.auto_selected = true; - state.select_chat(Some(chat_id), cx); + while let Some(value) = rx.recv().await { + let parsed: Vec = match serde_json::from_value(value) { + Ok(parsed) => parsed, + Err(err) => { + tracing::warn!(error = %err, "dropping malformed chats frame"); + continue; + } + }; + let alive = this.update(cx, |state, cx| { + state.apply_chats(parsed); + if state.selected_chat.is_none() && !state.auto_selected { + let most_recent = state + .chats + .iter() + .find(|c| !c.archived) + .map(|c| c.id.clone()); + if let Some(chat_id) = most_recent { + state.auto_selected = true; + state.select_chat(Some(chat_id), cx); + } } + cx.notify(); + }); + if alive.is_err() { + return; } - cx.notify(); - }); - if alive.is_err() { - break; } + tracing::debug!("chats stream ended; resubscribing"); + if this.update(cx, |_, _| {}).is_err() { + return; + } + cx.background_executor().timer(RETRY_DELAY).await; } }) } @@ -933,32 +961,49 @@ fn spawn_watch( apply: fn(&mut AppState, T), ) -> Task<()> { cx.spawn(async move |this, cx| { - let mut rx = match handle - .client() - .subscribe(method, serde_json::json!({})) - .await - { - Ok(rx) => rx, - Err(err) => { - tracing::debug!(method, error = %err, "watch unavailable"); - return; - } - }; - while let Some(value) = rx.recv().await { - let parsed: T = match serde_json::from_value(value) { - Ok(parsed) => parsed, + // Resubscribe loop: these are the standing Sessions/Devices/Spaces + // watches — a daemon restart ended the stream and a bare return froze + // them for the rest of the app's life (remote Working dots staled out + // to nothing after 45s, and Idle/Completed transitions from other + // devices never arrived again — "the session never completes"). + const RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(2); + loop { + let mut rx = match handle + .client() + .subscribe(method, serde_json::json!({})) + .await + { + Ok(rx) => rx, Err(err) => { - tracing::warn!(method, error = %err, "dropping malformed watch frame"); + tracing::debug!(method, error = %err, "watch unavailable; retrying"); + if this.update(cx, |_, _| {}).is_err() { + return; + } + cx.background_executor().timer(RETRY_DELAY).await; continue; } }; - let alive = this.update(cx, |state, cx| { - apply(state, parsed); - cx.notify(); - }); - if alive.is_err() { - break; + while let Some(value) = rx.recv().await { + let parsed: T = match serde_json::from_value(value) { + Ok(parsed) => parsed, + Err(err) => { + tracing::warn!(method, error = %err, "dropping malformed watch frame"); + continue; + } + }; + let alive = this.update(cx, |state, cx| { + apply(state, parsed); + cx.notify(); + }); + if alive.is_err() { + return; + } } + tracing::debug!(method, "watch stream ended; resubscribing"); + if this.update(cx, |_, _| {}).is_err() { + return; + } + cx.background_executor().timer(RETRY_DELAY).await; } }) } diff --git a/scripts/e2e-smoke.sh b/scripts/e2e-smoke.sh index f72480f3c1..cfd6f57cae 100755 --- a/scripts/e2e-smoke.sh +++ b/scripts/e2e-smoke.sh @@ -84,7 +84,11 @@ fi # ── 2. Build the binaries (workspace target is warm in CI/dev) ───────────────── echo "build: comet + e2e_driver" -(cd "$ROOT" && cargo build -q -p comet -p comet-rpc --example e2e_driver) +# Two invocations: a single one with `--example` builds ONLY the example +# (the target filter applies across every -p), silently skipping the comet +# bin — the smoke then dies on "No such file or directory". +(cd "$ROOT" && cargo build -q -p comet) +(cd "$ROOT" && cargo build -q -p comet-rpc --example e2e_driver) COMET="$ROOT/target/debug/comet" DRIVER="$ROOT/target/debug/examples/e2e_driver"