fix: clear stale in-flight state

crates/worker/src/controller.rs

@@ -139,7 +139,13 @@ async fn finish_controller_run<C, St>(
     // 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<C, St>(
             // 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<C, St>(
             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<C, St>(
             } => 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,

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);

crates/worker/src/worker.rs

@@ -654,6 +654,12 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Worker<C, St>
         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