fix: clear stale internal worker in-flight state

crates/worker/src/internal_worker.rs

@@ -555,6 +555,7 @@ pub(crate) async fn prepare_internal_worker_session(
     spawn_internal_log_event_bridge(sink.clone(), event_tx.clone());
     let alerter = Alerter::new(event_tx.clone());
     let in_flight = InFlightEvents::new(event_tx.clone());
+    let actor_in_flight = in_flight.clone();
     worker.attach_alerter(alerter.clone());
     worker.attach_event_tx(event_tx.clone());
     worker.attach_in_flight_events(in_flight.clone());
@@ -587,6 +588,7 @@ pub(crate) async fn prepare_internal_worker_session(
         while let Some(command) = command_rx.recv().await {
             match command {
                 InternalWorkerSessionCommand::Run(input) => {
+                    actor_in_flight.clear();
                     let cancel_sender = worker.engine_mut().cancel_sender();
                     let mut run = std::pin::pin!(worker.run_text(&input));
                     loop {
@@ -599,6 +601,7 @@ pub(crate) async fn prepare_internal_worker_session(
                                         Some(error.to_string()),
                                     ),
                                 };
+                                actor_in_flight.clear();
                                 status.store(turn_status.encode(), std::sync::atomic::Ordering::Release);
                                 if let Some(message) = error {
                                     *last_error.lock().unwrap() = Some(message.clone());
@@ -622,6 +625,7 @@ pub(crate) async fn prepare_internal_worker_session(
                                     Some(InternalWorkerSessionCommand::Stop(done)) => {
                                         let _ = cancel_sender.send(()).await;
                                         let _ = (&mut run).await;
+                                        actor_in_flight.clear();
                                         status.store(InternalWorkerSessionStatus::Stopped.encode(), std::sync::atomic::Ordering::Release);
                                         let _ = event_tx.send(Event::Status { status: WorkerStatus::Paused });
                                         let _ = event_tx.send(Event::Shutdown);
@@ -634,6 +638,7 @@ pub(crate) async fn prepare_internal_worker_session(
                                     }
                                     None => {
                                         let _ = cancel_sender.send(()).await;
+                                        actor_in_flight.clear();
                                         return;
                                     }
                                 }
@@ -642,6 +647,7 @@ pub(crate) async fn prepare_internal_worker_session(
                     }
                 }
                 InternalWorkerSessionCommand::Stop(done) => {
+                    actor_in_flight.clear();
                     status.store(
                         InternalWorkerSessionStatus::Stopped.encode(),
                         std::sync::atomic::Ordering::Release,
@@ -656,6 +662,7 @@ pub(crate) async fn prepare_internal_worker_session(
                 }
             }
         }
+        actor_in_flight.clear();
     });
 
     Ok(handle)
@@ -1068,6 +1075,10 @@ permission = "write"
             handle.wait_until_idle().await,
             InternalWorkerSessionStatus::Idle
         );
+        handle
+            .in_flight
+            .tool_call_start("stale-call".to_string(), "Read".to_string());
+        assert_eq!(handle.protocol_snapshot().in_flight.blocks.len(), 1);
         let entries_after_first = handle.entries().len();
         assert!(entries_after_first >= 4);
         handle.send("follow-up").await.expect("send follow-up turn");
@@ -1075,6 +1086,7 @@ permission = "write"
             handle.wait_until_idle().await,
             InternalWorkerSessionStatus::Idle
         );
+        assert!(handle.protocol_snapshot().in_flight.blocks.is_empty());
         assert_eq!(calls.load(Ordering::SeqCst), 2);
         assert!(handle.entries().len() > entries_after_first);
 

web/workspace/src/lib/workspace/console/model.test.ts

@@ -1314,9 +1314,36 @@ Deno.test("parent snapshot authoritatively replaces Internal Worker projections"
       kind: "sub_worker",
     },
     revision: 4,
-    entries: [],
+    entries: [{
+      kind: "assistant_item",
+      ts: 1,
+      item: {
+        kind: "tool_call",
+        call_id: "committed-call",
+        name: "Read",
+        arguments: JSON.stringify({ file_path: "/repo/a.md" }),
+      },
+    }, {
+      kind: "tool_result",
+      ts: 2,
+      item: {
+        kind: "tool_result",
+        call_id: "committed-call",
+        summary: "read file",
+        content: "content",
+        is_error: false,
+      },
+    }],
     status: "idle",
-    in_flight: { blocks: [] },
+    in_flight: {
+      blocks: [{
+        kind: "tool_call",
+        id: "committed-call",
+        name: "Read",
+        args: JSON.stringify({ file_path: "/repo/a.md" }),
+        state: "done",
+      }],
+    },
     internal_workers: [],
   }];
   const projector = createConsoleProjector();
@@ -1340,6 +1367,10 @@ Deno.test("parent snapshot authoritatively replaces Internal Worker projections"
   assertEquals(projection.internalWorkers.map((worker) => worker.worker.session_id), [
     "replacement",
   ]);
+  const childLines = projection.internalWorkers[0].console.lines;
+  assertEquals(childLines.length, 1);
+  assertEquals(new Set(childLines.map((line) => line.id)).size, 1);
+  assertEquals(childLines[0].kind, "tool");
 });
 
 Deno.test("snapshot restores TaskStore state from system history", () => {

web/workspace/src/lib/workspace/console/model.ts

@@ -224,6 +224,21 @@ function projectVisibleConsole(
   };
 }
 
+function appendSnapshotInFlightLines(
+  projection: ConsoleProjection,
+  blocks: InFlightBlock[],
+  eventId: string,
+  cwd: string | null,
+): void {
+  const lineIds = new Set(projection.lines.map((line) => line.id));
+  blocks.forEach((block, index) => {
+    const pending = inFlightLine(`${eventId}:${index}`, block, cwd);
+    if (lineIds.has(pending.id)) return;
+    projection.lines.push(pending);
+    lineIds.add(pending.id);
+  });
+}
+
 function projectInternalWorkerSnapshot(
   snapshot: InternalWorkerSnapshot,
   eventId: string,
@@ -235,15 +250,12 @@ function projectInternalWorkerSnapshot(
     cwd,
   );
   console.status = snapshot.status;
-  for (const block of snapshot.in_flight?.blocks ?? []) {
+  appendSnapshotInFlightLines(
-    console.lines.push(
+    console,
-      inFlightLine(
+    snapshot.in_flight?.blocks ?? [],
-        `${eventId}:internal:${snapshot.worker.session_id}:in-flight`,
+    `${eventId}:internal:${snapshot.worker.session_id}:in-flight`,
-        block,
+    cwd,
-        cwd,
+  );
-      ),
-    );
-  }
   if (snapshot.error) {
     console.lines.push({
       id: `${eventId}:internal:${snapshot.worker.session_id}:error`,
@@ -385,9 +397,12 @@ export function applyProtocolEvent(
       next.lines = snapshot.lines;
       next.tasks = snapshot.tasks;
       next.taskNextId = snapshot.taskNextId;
-      for (const block of event.data.in_flight?.blocks ?? []) {
+      appendSnapshotInFlightLines(
-        next.lines.push(inFlightLine(envelope.eventId, block, next.cwd));
+        next,
-      }
+        event.data.in_flight?.blocks ?? [],
+        `${envelope.eventId}:snapshot-in-flight`,
+        next.cwd,
+      );
       next.internalWorkers = (event.data.internal_workers ?? []).map((worker) =>
         projectInternalWorkerSnapshot(worker, envelope.eventId, next.cwd)
       );