diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 12ba8d8a..002b91dc 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -363,6 +363,10 @@ pub enum Event { max_attempts: u32, reason: String, }, + /// Any transient streaming blocks that were not committed to history were + /// discarded at a run boundary. Clients should remove live in-flight lines + /// without modifying persisted transcript entries. + InFlightCleared, TextDelta { text: String, }, diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index af2abaf0..0469967d 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -1030,6 +1030,12 @@ impl App { "LLM stream interrupted; continuing generation ({attempt}/{max_attempts}): {reason}" )); } + Event::InFlightCleared => { + self.assistant_streaming = false; + self.latest_llm_wait_event = None; + self.mark_orphan_tool_calls_incomplete(); + self.current_tool = None; + } Event::TextDelta { text } => { self.latest_llm_wait_event = None; self.append_assistant_text(&text); diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 0ec1277d..3f18f901 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -139,7 +139,13 @@ async fn finish_controller_run( // history / user_segments are no longer mirrored on WorkerSharedState — // clients reconstruct them from `Event::Snapshot` + live // `Event::Entry` deliveries driven by the session-log sink. We - // only flip the status and kick post-run memory jobs here. + // flip the status and kick post-run memory jobs here. + // + // In-flight blocks are run-local streaming state, not durable transcript. + // Any block not cleared by a committed AssistantItem must be discarded at + // the terminal run boundary so reconnect snapshots cannot append stale + // partial text/tool arguments after newer entries. + worker.clear_in_flight_events(); set_controller_status(shared_state, runtime_dir, event_tx, new_status).await; worker.spawn_post_run_memory_jobs(); } @@ -807,6 +813,10 @@ async fn controller_loop( // after this point is delivered to the turn and must not be discarded by // the Engine at run start. worker.engine_mut().clear_pending_cancel(); + // In-flight display state belongs to the active run only. Defensive + // clear at run start prevents stale partial output left by an older + // interrupted/error turn from being carried into the next snapshot. + worker.clear_in_flight_events(); set_controller_status( &shared_state, &runtime_dir, @@ -943,6 +953,7 @@ async fn controller_loop( Method::Cancel => match shared_state.get_status() { WorkerStatus::Paused => match worker.cancel_paused_turn() { Ok(()) => { + worker.clear_in_flight_events(); set_controller_status( &shared_state, &runtime_dir, @@ -1028,6 +1039,7 @@ async fn controller_loop( } => match shared_state.get_status() { WorkerStatus::Idle => { if apply_rewind(&mut worker, &event_tx, target, expected_head_entries) { + worker.clear_in_flight_events(); shared_state.set_status(WorkerStatus::Idle); let _ = event_tx.send(Event::Status { status: WorkerStatus::Idle, diff --git a/crates/worker/src/in_flight.rs b/crates/worker/src/in_flight.rs index 2fa5119f..5b545f81 100644 --- a/crates/worker/src/in_flight.rs +++ b/crates/worker/src/in_flight.rs @@ -201,6 +201,16 @@ impl InFlightEvents { f() } + pub(crate) fn clear(&self) { + let cleared = { + let mut inner = self.lock(); + inner.clear() + }; + if cleared { + let _ = self.event_tx.send(Event::InFlightCleared); + } + } + fn lock(&self) -> MutexGuard<'_, InFlightInner> { self.inner.lock().expect("in-flight event mutex poisoned") } @@ -271,6 +281,15 @@ impl InFlightInner { } } + fn clear(&mut self) -> bool { + if self.blocks.is_empty() { + false + } else { + self.blocks.clear(); + true + } + } + fn remove_first_text_matching(&mut self, committed: &str) -> bool { if let Some(index) = self.blocks.iter().position(|block| match block { TrackedBlock::Text { text, .. } => text == committed, @@ -569,6 +588,37 @@ mod tests { ); } + #[test] + fn clear_discards_uncommitted_blocks_and_notifies_clients() { + let (event_tx, _) = broadcast::channel(16); + let mut rx = event_tx.subscribe(); + let in_flight = InFlightEvents::new(event_tx); + let text = in_flight.start_text_block(); + in_flight.text_delta(text, "stale".into()); + let tool = in_flight.tool_call_start("call-1".into(), "Bash".into()); + in_flight.tool_call_args_delta(tool, "call-1".into(), "{\"command\":".into()); + + in_flight.clear(); + + let guard = in_flight.snapshot_guard(); + assert!(snapshot_from_guard(&guard).is_empty()); + drop(guard); + assert!(matches!( + rx.try_recv().unwrap(), + Event::TextDelta { text } if text == "stale" + )); + assert!(matches!( + rx.try_recv().unwrap(), + Event::ToolCallStart { .. } + )); + assert!(matches!( + rx.try_recv().unwrap(), + Event::ToolCallArgsDelta { .. } + )); + assert!(matches!(rx.try_recv().unwrap(), Event::InFlightCleared)); + assert!(rx.try_recv().is_err()); + } + #[test] fn snapshot_omits_empty_finished_thinking_blocks() { let (event_tx, _) = broadcast::channel(16); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index b2aa804e..8c55dd58 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -654,6 +654,12 @@ impl Worker self.in_flight = Some(in_flight); } + pub fn clear_in_flight_events(&self) { + if let Some(in_flight) = &self.in_flight { + in_flight.clear(); + } + } + /// Wire `Engine::on_history_append` to commit each appended item /// directly as a singular `LogEntry::AssistantItem` / `ToolResult` /// through the writer. The controller calls this once per spawned diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index b639802a..9d619e65 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -225,6 +225,7 @@ export type Event = reason: string; }; } + | { "event": "in_flight_cleared" } | { "event": "text_delta"; "data": { text: string } } | { "event": "text_done"; "data": { text: string } } | { "event": "thinking_start" } diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index d1c5253c..faea91c7 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -859,6 +859,53 @@ Deno.test("projectConsole renders snapshot entries and in-flight output", () => ); }); +Deno.test("projectConsole clears transient in-flight rows", () => { + const projection = projectConsole([ + { + eventId: "20", + event: { + event: "snapshot", + data: { + entries: [ + { + kind: "assistant_item", + ts: 1, + item: { + kind: "message", + role: "assistant", + content: [{ kind: "text", text: "latest committed" }], + }, + }, + ], + greeting: { + worker_name: "Worker", + cwd: "/repo", + provider: "provider", + model: "model", + scope_summary: "bounded", + tools: [], + context_window: 100, + context_tokens: 20, + }, + status: "running", + in_flight: { + blocks: [{ kind: "text", text: "stale partial" }], + }, + }, + } satisfies Event, + }, + { + eventId: "21", + event: { event: "in_flight_cleared" } satisfies Event, + }, + ]); + + assertEquals( + projection.lines.map((line) => `${line.kind}:${line.body}`), + ["assistant:latest committed"], + ); +}); + Deno.test("projectConsole reseeds visible rows from segment rotation", () => { const projection = projectConsole([ { diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index abfa3d52..218708d1 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -303,6 +303,9 @@ export function applyProtocolEvent( next.lines.push(inFlightLine(envelope.eventId, block, next.cwd)); } break; + case "in_flight_cleared": + next.lines = next.lines.filter((line) => line.kind !== "in_flight"); + break; case "status": next.status = event.data.status; break;