From 3a7a3307ef3a96f1a8bf4ce64846d334e1698cf3 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 27 Aug 2026 12:27:55 +0900 Subject: [PATCH] fix: map internal worker terminal lifecycles --- crates/session-store/tests/session_test.rs | 38 ++++-- crates/worker/examples/worker_cli.rs | 1 + crates/worker/src/controller.rs | 6 +- crates/worker/src/internal_worker.rs | 121 ++++++++++++++---- crates/worker/src/worker.rs | 30 ++++- .../sidebar/WorkersNavSection.svelte | 9 +- .../sidebar/worker-subscription.test.ts | 34 ++++- .../src/lib/workspace/sidebar/workers.test.ts | 19 +++ .../src/lib/workspace/sidebar/workers.ts | 19 +++ 9 files changed, 225 insertions(+), 52 deletions(-) diff --git a/crates/session-store/tests/session_test.rs b/crates/session-store/tests/session_test.rs index dda7205d..2116370f 100644 --- a/crates/session-store/tests/session_test.rs +++ b/crates/session-store/tests/session_test.rs @@ -102,7 +102,7 @@ async fn run_and_persist( session_id: session_store::SessionId, segment_id: session_store::SegmentId, input: &str, -) -> (Engine, agen::EngineResult) { +) -> (Engine, agen::EngineRunExit) { // Mirror Worker's run-entry contract: log the user input as segments // before the worker pushes its flattened user_message; save_delta // skips the resulting user_message item to avoid double-write. @@ -125,31 +125,49 @@ async fn run_and_persist( session_store::save_turn_end(store, session_id, segment_id, worker.turn_count()).unwrap(); match &result { - Ok(r) => { + agen::EngineRunExit::Finished + | agen::EngineRunExit::Paused + | agen::EngineRunExit::Yielded => { + let (legacy_result, interrupted) = match &result { + agen::EngineRunExit::Finished => (agen::EngineResult::Finished, false), + agen::EngineRunExit::Paused => (agen::EngineResult::Paused, true), + agen::EngineRunExit::Yielded => (agen::EngineResult::Yielded, true), + agen::EngineRunExit::Interrupted(_) => unreachable!(), + }; session_store::save_run_completed( store, session_id, segment_id, - r.clone(), - worker.last_run_interrupted(), + legacy_result, + interrupted, worker.active_run_turn_count(), ) .unwrap(); } - Err(e) => { + agen::EngineRunExit::Interrupted(agen::StopReason::LimitReached) => { + session_store::save_run_completed( + store, + session_id, + segment_id, + agen::EngineResult::LimitReached, + false, + worker.active_run_turn_count(), + ) + .unwrap(); + } + agen::EngineRunExit::Interrupted(reason) => { session_store::save_run_errored( store, session_id, segment_id, - e.to_string(), - worker.last_run_interrupted(), + format!("{reason:?}"), + true, ) .unwrap(); } } - let r = result.unwrap(); - (worker, r) + (worker, result) } // ============================================================================= @@ -292,7 +310,7 @@ async fn session_resume_after_pause() { .unwrap(); let (_worker, result) = run_and_persist(worker, &store, sid, segid, "Weather?").await; - assert!(matches!(result, agen::EngineResult::Paused)); + assert!(matches!(result, agen::EngineRunExit::Paused)); // Check RunCompleted is Paused let entries = store.read_all(sid, segid).unwrap(); diff --git a/crates/worker/examples/worker_cli.rs b/crates/worker/examples/worker_cli.rs index d7e26f0c..ccc2acd1 100644 --- a/crates/worker/examples/worker_cli.rs +++ b/crates/worker/examples/worker_cli.rs @@ -66,6 +66,7 @@ async fn main() -> Result<(), Box> { WorkerRunResult::Finished => println!("(finished)"), WorkerRunResult::Paused => println!("(paused)"), WorkerRunResult::LimitReached => println!("(turn limit reached)"), + WorkerRunResult::Interrupted { message, .. } => println!("(interrupted: {message})"), WorkerRunResult::RolledBack => println!("(empty turn rolled back)"), } diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index aeb139b4..638adadc 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1650,13 +1650,13 @@ where WorkerRunResult::Paused => (WorkerStatus::Paused, RunResult::Paused), WorkerRunResult::LimitReached => (WorkerStatus::Idle, RunResult::LimitReached), WorkerRunResult::RolledBack => (WorkerStatus::Idle, RunResult::RolledBack), - WorkerRunResult::Interrupted(_message) if pause_requested => { + WorkerRunResult::Interrupted { .. } if pause_requested => { let _ = event_tx.send(Event::RunEnd { result: RunResult::Paused }); return (WorkerStatus::Paused, shutdown_requested); } - WorkerRunResult::Interrupted(message) => { + WorkerRunResult::Interrupted { code, message } => { let _ = event_tx.send(Event::Error { - code: ErrorCode::Internal, + code, message: message.clone(), }); if parent_originated { diff --git a/crates/worker/src/internal_worker.rs b/crates/worker/src/internal_worker.rs index de03898b..e0464198 100644 --- a/crates/worker/src/internal_worker.rs +++ b/crates/worker/src/internal_worker.rs @@ -199,10 +199,12 @@ where on_cancel_sender(worker.engine_mut().cancel_sender()); match worker.run_text(&input).await { - Ok(WorkerRunResult::Interrupted(message)) => Err(InternalWorkerError { - source: WorkerError::Engine(EngineError::Aborted(message)), + Ok(lifecycle @ WorkerRunResult::Finished) + | Ok(lifecycle @ WorkerRunResult::Paused) + | Ok(lifecycle @ WorkerRunResult::RolledBack) => Ok(InternalWorkerResult { usage: last_usage.lock().ok().and_then(|slot| slot.clone()), identity, + lifecycle, history_entries: store.entries_count(session_id, segment_id), }), Ok(WorkerRunResult::LimitReached) => Err(InternalWorkerError { @@ -213,10 +215,10 @@ where identity, history_entries: store.entries_count(session_id, segment_id), }), - Ok(lifecycle) => Ok(InternalWorkerResult { + Ok(WorkerRunResult::Interrupted { message, .. }) => Err(InternalWorkerError { + source: WorkerError::Engine(EngineError::Aborted(message)), usage: last_usage.lock().ok().and_then(|slot| slot.clone()), identity, - lifecycle, history_entries: store.entries_count(session_id, segment_id), }), Err(source) => Err(InternalWorkerError { @@ -246,6 +248,7 @@ impl Default for InternalWorkerVisibility { pub(crate) enum InternalWorkerSessionStatus { Idle, Running, + Paused, Stopping, Stopped, Failed, @@ -256,9 +259,10 @@ impl InternalWorkerSessionStatus { match self { Self::Idle => 0, Self::Running => 1, - Self::Stopping => 2, - Self::Stopped => 3, - Self::Failed => 4, + Self::Paused => 2, + Self::Stopping => 3, + Self::Stopped => 4, + Self::Failed => 5, } } @@ -266,13 +270,35 @@ impl InternalWorkerSessionStatus { match value { 0 => Self::Idle, 1 => Self::Running, - 2 => Self::Stopping, - 3 => Self::Stopped, + 2 => Self::Paused, + 3 => Self::Stopping, + 4 => Self::Stopped, _ => Self::Failed, } } } +fn classify_internal_turn_result( + result: Result, +) -> (InternalWorkerSessionStatus, Option) { + match result { + Ok(WorkerRunResult::Finished) => (InternalWorkerSessionStatus::Idle, None), + Ok(WorkerRunResult::Paused) => (InternalWorkerSessionStatus::Paused, None), + Ok(WorkerRunResult::LimitReached) => ( + InternalWorkerSessionStatus::Stopped, + Some("internal Worker reached its turn limit".to_string()), + ), + Ok(WorkerRunResult::Interrupted { message, .. }) => { + (InternalWorkerSessionStatus::Stopped, Some(message)) + } + Ok(WorkerRunResult::RolledBack) => ( + InternalWorkerSessionStatus::Stopped, + Some("internal Worker run was cancelled before AI output".to_string()), + ), + Err(error) => (InternalWorkerSessionStatus::Failed, Some(error.to_string())), + } +} + #[derive(Debug, thiserror::Error)] pub(crate) enum InternalWorkerSessionError { #[error("failed to build internal Worker session: {message}")] @@ -367,6 +393,7 @@ impl InternalWorkerSessionHandle { entries, status: match self.status() { InternalWorkerSessionStatus::Running => WorkerStatus::Running, + InternalWorkerSessionStatus::Paused => WorkerStatus::Paused, InternalWorkerSessionStatus::Idle => WorkerStatus::Idle, InternalWorkerSessionStatus::Stopping | InternalWorkerSessionStatus::Stopped @@ -402,6 +429,7 @@ impl InternalWorkerSessionHandle { .map_err( |current| match InternalWorkerSessionStatus::decode(current) { InternalWorkerSessionStatus::Running + | InternalWorkerSessionStatus::Paused | InternalWorkerSessionStatus::Stopping => InternalWorkerSessionError::Busy, InternalWorkerSessionStatus::Stopped | InternalWorkerSessionStatus::Failed => { InternalWorkerSessionError::Stopped @@ -747,21 +775,7 @@ pub(crate) async fn prepare_internal_worker_session( loop { tokio::select! { result = &mut run => { - let (turn_status, error) = match result { - Ok(WorkerRunResult::Interrupted(message)) => ( - InternalWorkerSessionStatus::Stopped, - Some(message), - ), - Ok(WorkerRunResult::LimitReached) => ( - InternalWorkerSessionStatus::Stopped, - Some("internal Worker reached its turn limit".to_string()), - ), - Ok(_) => (InternalWorkerSessionStatus::Idle, None), - Err(error) => ( - InternalWorkerSessionStatus::Failed, - Some(error.to_string()), - ), - }; + let (turn_status, error) = classify_internal_turn_result(result); actor_in_flight.clear(); status.store(turn_status.encode(), std::sync::atomic::Ordering::Release); if let Some(message) = error { @@ -771,10 +785,15 @@ pub(crate) async fn prepare_internal_worker_session( message, }); } - let protocol_status = if turn_status == InternalWorkerSessionStatus::Idle { - WorkerStatus::Idle - } else { - WorkerStatus::Stopped + let protocol_status = match turn_status { + InternalWorkerSessionStatus::Idle => WorkerStatus::Idle, + InternalWorkerSessionStatus::Paused => WorkerStatus::Paused, + InternalWorkerSessionStatus::Stopped + | InternalWorkerSessionStatus::Failed => WorkerStatus::Stopped, + InternalWorkerSessionStatus::Running + | InternalWorkerSessionStatus::Stopping => { + unreachable!("run completion cannot remain active") + } }; let _ = event_tx.send(Event::Status { status: protocol_status, @@ -1261,6 +1280,52 @@ permission = "write" assert_eq!(result.identity.kind, "test"); } + #[test] + fn internal_turn_result_mapping_is_exhaustive() { + let cases = [ + ( + WorkerRunResult::Finished, + InternalWorkerSessionStatus::Idle, + false, + ), + ( + WorkerRunResult::Paused, + InternalWorkerSessionStatus::Paused, + false, + ), + ( + WorkerRunResult::LimitReached, + InternalWorkerSessionStatus::Stopped, + true, + ), + ( + WorkerRunResult::Interrupted { + code: protocol::ErrorCode::Internal, + message: "cancelled".to_string(), + }, + InternalWorkerSessionStatus::Stopped, + true, + ), + ( + WorkerRunResult::RolledBack, + InternalWorkerSessionStatus::Stopped, + true, + ), + ]; + + for (result, expected_status, expects_error) in cases { + let (status, error) = classify_internal_turn_result(Ok(result)); + assert_eq!(status, expected_status); + assert_eq!(error.is_some(), expects_error); + } + + let (status, error) = classify_internal_turn_result(Err(WorkerError::Engine( + EngineError::Aborted("fatal".to_string()), + ))); + assert_eq!(status, InternalWorkerSessionStatus::Failed); + assert!(error.is_some_and(|message| message.contains("fatal"))); + } + #[tokio::test] async fn fatal_internal_run_transitions_to_stopped_protocol_status() { let calls = Arc::new(AtomicUsize::new(0)); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index c9f06a91..9207a232 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -77,8 +77,8 @@ use crate::skill::{SkillActivationResponse, SkillClientError}; #[cfg(test)] use async_trait::async_trait; use protocol::{ - AlertLevel, AlertSource, CompactionLifecycle, CompactionLifecycleState, Event, RewindSummary, - RewindTarget, RewindTargetId, Segment, + AlertLevel, AlertSource, CompactionLifecycle, CompactionLifecycleState, ErrorCode, Event, + RewindSummary, RewindTarget, RewindTargetId, Segment, }; use tokio::net::UnixStream; use tokio::sync::broadcast; @@ -2964,7 +2964,10 @@ impl Worker { } EngineRunExit::Interrupted(reason) => { self.last_run_interrupted = true; - Ok(WorkerRunResult::Interrupted(stop_reason_message(&reason))) + Ok(WorkerRunResult::Interrupted { + code: stop_reason_error_code(&reason), + message: stop_reason_message(&reason), + }) } EngineRunExit::Yielded => unreachable!("yielded handled above"), } @@ -4512,7 +4515,7 @@ impl Worker { fn extract_internal_worker_lifecycle_error(lifecycle: &WorkerRunResult) -> Option { match lifecycle { WorkerRunResult::RolledBack => Some(WorkerError::Engine(EngineError::Cancelled)), - WorkerRunResult::Interrupted(message) => { + WorkerRunResult::Interrupted { message, .. } => { Some(WorkerError::Engine(EngineError::Aborted(message.clone()))) } WorkerRunResult::Finished | WorkerRunResult::Paused | WorkerRunResult::LimitReached => None, @@ -5521,6 +5524,23 @@ fn restore_manifest_from_worker_metadata_snapshot( } } +fn stop_reason_error_code(reason: &StopReason) -> ErrorCode { + match reason { + StopReason::ContextWindowExceeded | StopReason::Unexpected(EngineError::Client(_)) => { + ErrorCode::ProviderError + } + StopReason::Unexpected(EngineError::Tool(_)) => ErrorCode::ToolError, + StopReason::LimitReached + | StopReason::Cancelled + | StopReason::Unexpected( + EngineError::Aborted(_) + | EngineError::Cancelled + | EngineError::ConfigWarnings(_) + | EngineError::HistoryAppend(_), + ) => ErrorCode::Internal, + } +} + fn stop_reason_message(reason: &StopReason) -> String { match reason { StopReason::LimitReached => "engine turn limit reached".to_string(), @@ -5540,7 +5560,7 @@ pub enum WorkerRunResult { /// The worker reached its configured max_turns limit. LimitReached, /// The run was interrupted by a known or unexpected terminal cause. - Interrupted(String), + Interrupted { code: ErrorCode, message: String }, /// The submit-time user turn was rolled back because no AI output was materialized. RolledBack, } diff --git a/web/workspace/src/lib/workspace/sidebar/WorkersNavSection.svelte b/web/workspace/src/lib/workspace/sidebar/WorkersNavSection.svelte index b2fffcd6..bb6a0743 100644 --- a/web/workspace/src/lib/workspace/sidebar/WorkersNavSection.svelte +++ b/web/workspace/src/lib/workspace/sidebar/WorkersNavSection.svelte @@ -5,7 +5,7 @@ workspaceWorkersStore, type SidebarWorker, } from './worker-subscription'; - import { canShowWorkerInSidebar } from './workers'; + import { canShowWorkerInSidebar, sidebarWorkerActivity } from './workers'; const COLLAPSED_WORKER_COUNT = 6; @@ -69,6 +69,7 @@