From de72afd9a1f495de84e688ea519bf8841d8f3cde Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 20 Aug 2026 06:18:21 +0900 Subject: [PATCH] fix: clear stale internal worker in-flight state --- crates/worker/src/internal_worker.rs | 12 ++++++ .../src/lib/workspace/console/model.test.ts | 35 ++++++++++++++++- .../src/lib/workspace/console/model.ts | 39 +++++++++++++------ 3 files changed, 72 insertions(+), 14 deletions(-) diff --git a/crates/worker/src/internal_worker.rs b/crates/worker/src/internal_worker.rs index 2786788a..0a21d0d2 100644 --- a/crates/worker/src/internal_worker.rs +++ b/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); diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index f609b0ef..041fdbfb 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/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", () => { diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index 5b7538a7..2e632fff 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/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 ?? []) { - console.lines.push( - inFlightLine( - `${eventId}:internal:${snapshot.worker.session_id}:in-flight`, - block, - cwd, - ), - ); - } + appendSnapshotInFlightLines( + console, + snapshot.in_flight?.blocks ?? [], + `${eventId}:internal:${snapshot.worker.session_id}:in-flight`, + 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 ?? []) { - next.lines.push(inFlightLine(envelope.eventId, block, next.cwd)); - } + appendSnapshotInFlightLines( + 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) );