fix: map internal worker terminal lifecycles
This commit is contained in:
@@ -102,7 +102,7 @@ async fn run_and_persist(
|
|||||||
session_id: session_store::SessionId,
|
session_id: session_store::SessionId,
|
||||||
segment_id: session_store::SegmentId,
|
segment_id: session_store::SegmentId,
|
||||||
input: &str,
|
input: &str,
|
||||||
) -> (Engine<MockLlmClient>, agen::EngineResult) {
|
) -> (Engine<MockLlmClient>, agen::EngineRunExit) {
|
||||||
// Mirror Worker's run-entry contract: log the user input as segments
|
// Mirror Worker's run-entry contract: log the user input as segments
|
||||||
// before the worker pushes its flattened user_message; save_delta
|
// before the worker pushes its flattened user_message; save_delta
|
||||||
// skips the resulting user_message item to avoid double-write.
|
// 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();
|
session_store::save_turn_end(store, session_id, segment_id, worker.turn_count()).unwrap();
|
||||||
|
|
||||||
match &result {
|
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(
|
session_store::save_run_completed(
|
||||||
store,
|
store,
|
||||||
session_id,
|
session_id,
|
||||||
segment_id,
|
segment_id,
|
||||||
r.clone(),
|
legacy_result,
|
||||||
worker.last_run_interrupted(),
|
interrupted,
|
||||||
worker.active_run_turn_count(),
|
worker.active_run_turn_count(),
|
||||||
)
|
)
|
||||||
.unwrap();
|
.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(
|
session_store::save_run_errored(
|
||||||
store,
|
store,
|
||||||
session_id,
|
session_id,
|
||||||
segment_id,
|
segment_id,
|
||||||
e.to_string(),
|
format!("{reason:?}"),
|
||||||
worker.last_run_interrupted(),
|
true,
|
||||||
)
|
)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
let r = result.unwrap();
|
(worker, result)
|
||||||
(worker, r)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// =============================================================================
|
// =============================================================================
|
||||||
@@ -292,7 +310,7 @@ async fn session_resume_after_pause() {
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
let (_worker, result) = run_and_persist(worker, &store, sid, segid, "Weather?").await;
|
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
|
// Check RunCompleted is Paused
|
||||||
let entries = store.read_all(sid, segid).unwrap();
|
let entries = store.read_all(sid, segid).unwrap();
|
||||||
|
|||||||
@@ -66,6 +66,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|||||||
WorkerRunResult::Finished => println!("(finished)"),
|
WorkerRunResult::Finished => println!("(finished)"),
|
||||||
WorkerRunResult::Paused => println!("(paused)"),
|
WorkerRunResult::Paused => println!("(paused)"),
|
||||||
WorkerRunResult::LimitReached => println!("(turn limit reached)"),
|
WorkerRunResult::LimitReached => println!("(turn limit reached)"),
|
||||||
|
WorkerRunResult::Interrupted { message, .. } => println!("(interrupted: {message})"),
|
||||||
WorkerRunResult::RolledBack => println!("(empty turn rolled back)"),
|
WorkerRunResult::RolledBack => println!("(empty turn rolled back)"),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1650,13 +1650,13 @@ where
|
|||||||
WorkerRunResult::Paused => (WorkerStatus::Paused, RunResult::Paused),
|
WorkerRunResult::Paused => (WorkerStatus::Paused, RunResult::Paused),
|
||||||
WorkerRunResult::LimitReached => (WorkerStatus::Idle, RunResult::LimitReached),
|
WorkerRunResult::LimitReached => (WorkerStatus::Idle, RunResult::LimitReached),
|
||||||
WorkerRunResult::RolledBack => (WorkerStatus::Idle, RunResult::RolledBack),
|
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 });
|
let _ = event_tx.send(Event::RunEnd { result: RunResult::Paused });
|
||||||
return (WorkerStatus::Paused, shutdown_requested);
|
return (WorkerStatus::Paused, shutdown_requested);
|
||||||
}
|
}
|
||||||
WorkerRunResult::Interrupted(message) => {
|
WorkerRunResult::Interrupted { code, message } => {
|
||||||
let _ = event_tx.send(Event::Error {
|
let _ = event_tx.send(Event::Error {
|
||||||
code: ErrorCode::Internal,
|
code,
|
||||||
message: message.clone(),
|
message: message.clone(),
|
||||||
});
|
});
|
||||||
if parent_originated {
|
if parent_originated {
|
||||||
|
|||||||
@@ -199,10 +199,12 @@ where
|
|||||||
on_cancel_sender(worker.engine_mut().cancel_sender());
|
on_cancel_sender(worker.engine_mut().cancel_sender());
|
||||||
|
|
||||||
match worker.run_text(&input).await {
|
match worker.run_text(&input).await {
|
||||||
Ok(WorkerRunResult::Interrupted(message)) => Err(InternalWorkerError {
|
Ok(lifecycle @ WorkerRunResult::Finished)
|
||||||
source: WorkerError::Engine(EngineError::Aborted(message)),
|
| Ok(lifecycle @ WorkerRunResult::Paused)
|
||||||
|
| Ok(lifecycle @ WorkerRunResult::RolledBack) => Ok(InternalWorkerResult {
|
||||||
usage: last_usage.lock().ok().and_then(|slot| slot.clone()),
|
usage: last_usage.lock().ok().and_then(|slot| slot.clone()),
|
||||||
identity,
|
identity,
|
||||||
|
lifecycle,
|
||||||
history_entries: store.entries_count(session_id, segment_id),
|
history_entries: store.entries_count(session_id, segment_id),
|
||||||
}),
|
}),
|
||||||
Ok(WorkerRunResult::LimitReached) => Err(InternalWorkerError {
|
Ok(WorkerRunResult::LimitReached) => Err(InternalWorkerError {
|
||||||
@@ -213,10 +215,10 @@ where
|
|||||||
identity,
|
identity,
|
||||||
history_entries: store.entries_count(session_id, segment_id),
|
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()),
|
usage: last_usage.lock().ok().and_then(|slot| slot.clone()),
|
||||||
identity,
|
identity,
|
||||||
lifecycle,
|
|
||||||
history_entries: store.entries_count(session_id, segment_id),
|
history_entries: store.entries_count(session_id, segment_id),
|
||||||
}),
|
}),
|
||||||
Err(source) => Err(InternalWorkerError {
|
Err(source) => Err(InternalWorkerError {
|
||||||
@@ -246,6 +248,7 @@ impl Default for InternalWorkerVisibility {
|
|||||||
pub(crate) enum InternalWorkerSessionStatus {
|
pub(crate) enum InternalWorkerSessionStatus {
|
||||||
Idle,
|
Idle,
|
||||||
Running,
|
Running,
|
||||||
|
Paused,
|
||||||
Stopping,
|
Stopping,
|
||||||
Stopped,
|
Stopped,
|
||||||
Failed,
|
Failed,
|
||||||
@@ -256,9 +259,10 @@ impl InternalWorkerSessionStatus {
|
|||||||
match self {
|
match self {
|
||||||
Self::Idle => 0,
|
Self::Idle => 0,
|
||||||
Self::Running => 1,
|
Self::Running => 1,
|
||||||
Self::Stopping => 2,
|
Self::Paused => 2,
|
||||||
Self::Stopped => 3,
|
Self::Stopping => 3,
|
||||||
Self::Failed => 4,
|
Self::Stopped => 4,
|
||||||
|
Self::Failed => 5,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -266,13 +270,35 @@ impl InternalWorkerSessionStatus {
|
|||||||
match value {
|
match value {
|
||||||
0 => Self::Idle,
|
0 => Self::Idle,
|
||||||
1 => Self::Running,
|
1 => Self::Running,
|
||||||
2 => Self::Stopping,
|
2 => Self::Paused,
|
||||||
3 => Self::Stopped,
|
3 => Self::Stopping,
|
||||||
|
4 => Self::Stopped,
|
||||||
_ => Self::Failed,
|
_ => Self::Failed,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn classify_internal_turn_result(
|
||||||
|
result: Result<WorkerRunResult, WorkerError>,
|
||||||
|
) -> (InternalWorkerSessionStatus, Option<String>) {
|
||||||
|
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)]
|
#[derive(Debug, thiserror::Error)]
|
||||||
pub(crate) enum InternalWorkerSessionError {
|
pub(crate) enum InternalWorkerSessionError {
|
||||||
#[error("failed to build internal Worker session: {message}")]
|
#[error("failed to build internal Worker session: {message}")]
|
||||||
@@ -367,6 +393,7 @@ impl InternalWorkerSessionHandle {
|
|||||||
entries,
|
entries,
|
||||||
status: match self.status() {
|
status: match self.status() {
|
||||||
InternalWorkerSessionStatus::Running => WorkerStatus::Running,
|
InternalWorkerSessionStatus::Running => WorkerStatus::Running,
|
||||||
|
InternalWorkerSessionStatus::Paused => WorkerStatus::Paused,
|
||||||
InternalWorkerSessionStatus::Idle => WorkerStatus::Idle,
|
InternalWorkerSessionStatus::Idle => WorkerStatus::Idle,
|
||||||
InternalWorkerSessionStatus::Stopping
|
InternalWorkerSessionStatus::Stopping
|
||||||
| InternalWorkerSessionStatus::Stopped
|
| InternalWorkerSessionStatus::Stopped
|
||||||
@@ -402,6 +429,7 @@ impl InternalWorkerSessionHandle {
|
|||||||
.map_err(
|
.map_err(
|
||||||
|current| match InternalWorkerSessionStatus::decode(current) {
|
|current| match InternalWorkerSessionStatus::decode(current) {
|
||||||
InternalWorkerSessionStatus::Running
|
InternalWorkerSessionStatus::Running
|
||||||
|
| InternalWorkerSessionStatus::Paused
|
||||||
| InternalWorkerSessionStatus::Stopping => InternalWorkerSessionError::Busy,
|
| InternalWorkerSessionStatus::Stopping => InternalWorkerSessionError::Busy,
|
||||||
InternalWorkerSessionStatus::Stopped | InternalWorkerSessionStatus::Failed => {
|
InternalWorkerSessionStatus::Stopped | InternalWorkerSessionStatus::Failed => {
|
||||||
InternalWorkerSessionError::Stopped
|
InternalWorkerSessionError::Stopped
|
||||||
@@ -747,21 +775,7 @@ pub(crate) async fn prepare_internal_worker_session(
|
|||||||
loop {
|
loop {
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
result = &mut run => {
|
result = &mut run => {
|
||||||
let (turn_status, error) = match result {
|
let (turn_status, error) = classify_internal_turn_result(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()),
|
|
||||||
),
|
|
||||||
};
|
|
||||||
actor_in_flight.clear();
|
actor_in_flight.clear();
|
||||||
status.store(turn_status.encode(), std::sync::atomic::Ordering::Release);
|
status.store(turn_status.encode(), std::sync::atomic::Ordering::Release);
|
||||||
if let Some(message) = error {
|
if let Some(message) = error {
|
||||||
@@ -771,10 +785,15 @@ pub(crate) async fn prepare_internal_worker_session(
|
|||||||
message,
|
message,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
let protocol_status = if turn_status == InternalWorkerSessionStatus::Idle {
|
let protocol_status = match turn_status {
|
||||||
WorkerStatus::Idle
|
InternalWorkerSessionStatus::Idle => WorkerStatus::Idle,
|
||||||
} else {
|
InternalWorkerSessionStatus::Paused => WorkerStatus::Paused,
|
||||||
WorkerStatus::Stopped
|
InternalWorkerSessionStatus::Stopped
|
||||||
|
| InternalWorkerSessionStatus::Failed => WorkerStatus::Stopped,
|
||||||
|
InternalWorkerSessionStatus::Running
|
||||||
|
| InternalWorkerSessionStatus::Stopping => {
|
||||||
|
unreachable!("run completion cannot remain active")
|
||||||
|
}
|
||||||
};
|
};
|
||||||
let _ = event_tx.send(Event::Status {
|
let _ = event_tx.send(Event::Status {
|
||||||
status: protocol_status,
|
status: protocol_status,
|
||||||
@@ -1261,6 +1280,52 @@ permission = "write"
|
|||||||
assert_eq!(result.identity.kind, "test");
|
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]
|
#[tokio::test]
|
||||||
async fn fatal_internal_run_transitions_to_stopped_protocol_status() {
|
async fn fatal_internal_run_transitions_to_stopped_protocol_status() {
|
||||||
let calls = Arc::new(AtomicUsize::new(0));
|
let calls = Arc::new(AtomicUsize::new(0));
|
||||||
|
|||||||
@@ -77,8 +77,8 @@ use crate::skill::{SkillActivationResponse, SkillClientError};
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
use protocol::{
|
use protocol::{
|
||||||
AlertLevel, AlertSource, CompactionLifecycle, CompactionLifecycleState, Event, RewindSummary,
|
AlertLevel, AlertSource, CompactionLifecycle, CompactionLifecycleState, ErrorCode, Event,
|
||||||
RewindTarget, RewindTargetId, Segment,
|
RewindSummary, RewindTarget, RewindTargetId, Segment,
|
||||||
};
|
};
|
||||||
use tokio::net::UnixStream;
|
use tokio::net::UnixStream;
|
||||||
use tokio::sync::broadcast;
|
use tokio::sync::broadcast;
|
||||||
@@ -2964,7 +2964,10 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
|
|||||||
}
|
}
|
||||||
EngineRunExit::Interrupted(reason) => {
|
EngineRunExit::Interrupted(reason) => {
|
||||||
self.last_run_interrupted = true;
|
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"),
|
EngineRunExit::Yielded => unreachable!("yielded handled above"),
|
||||||
}
|
}
|
||||||
@@ -4512,7 +4515,7 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
|
|||||||
fn extract_internal_worker_lifecycle_error(lifecycle: &WorkerRunResult) -> Option<WorkerError> {
|
fn extract_internal_worker_lifecycle_error(lifecycle: &WorkerRunResult) -> Option<WorkerError> {
|
||||||
match lifecycle {
|
match lifecycle {
|
||||||
WorkerRunResult::RolledBack => Some(WorkerError::Engine(EngineError::Cancelled)),
|
WorkerRunResult::RolledBack => Some(WorkerError::Engine(EngineError::Cancelled)),
|
||||||
WorkerRunResult::Interrupted(message) => {
|
WorkerRunResult::Interrupted { message, .. } => {
|
||||||
Some(WorkerError::Engine(EngineError::Aborted(message.clone())))
|
Some(WorkerError::Engine(EngineError::Aborted(message.clone())))
|
||||||
}
|
}
|
||||||
WorkerRunResult::Finished | WorkerRunResult::Paused | WorkerRunResult::LimitReached => None,
|
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 {
|
fn stop_reason_message(reason: &StopReason) -> String {
|
||||||
match reason {
|
match reason {
|
||||||
StopReason::LimitReached => "engine turn limit reached".to_string(),
|
StopReason::LimitReached => "engine turn limit reached".to_string(),
|
||||||
@@ -5540,7 +5560,7 @@ pub enum WorkerRunResult {
|
|||||||
/// The worker reached its configured max_turns limit.
|
/// The worker reached its configured max_turns limit.
|
||||||
LimitReached,
|
LimitReached,
|
||||||
/// The run was interrupted by a known or unexpected terminal cause.
|
/// 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.
|
/// The submit-time user turn was rolled back because no AI output was materialized.
|
||||||
RolledBack,
|
RolledBack,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,7 +5,7 @@
|
|||||||
workspaceWorkersStore,
|
workspaceWorkersStore,
|
||||||
type SidebarWorker,
|
type SidebarWorker,
|
||||||
} from './worker-subscription';
|
} from './worker-subscription';
|
||||||
import { canShowWorkerInSidebar } from './workers';
|
import { canShowWorkerInSidebar, sidebarWorkerActivity } from './workers';
|
||||||
|
|
||||||
const COLLAPSED_WORKER_COUNT = 6;
|
const COLLAPSED_WORKER_COUNT = 6;
|
||||||
|
|
||||||
@@ -69,6 +69,7 @@
|
|||||||
<ul class="nav-list" aria-label="Workers">
|
<ul class="nav-list" aria-label="Workers">
|
||||||
{#each visibleWorkers as worker (`${worker.runtime_id}:${worker.worker_id}`)}
|
{#each visibleWorkers as worker (`${worker.runtime_id}:${worker.worker_id}`)}
|
||||||
{@const href = workerConsoleHref(worker, workspaceId)}
|
{@const href = workerConsoleHref(worker, workspaceId)}
|
||||||
|
{@const activity = sidebarWorkerActivity(worker)}
|
||||||
<li>
|
<li>
|
||||||
<a
|
<a
|
||||||
href={href}
|
href={href}
|
||||||
@@ -77,11 +78,11 @@
|
|||||||
aria-current={currentPath === href ? 'page' : undefined}
|
aria-current={currentPath === href ? 'page' : undefined}
|
||||||
>
|
>
|
||||||
<span class="worker-status-indicator">
|
<span class="worker-status-indicator">
|
||||||
{#if worker.state === 'running'}
|
{#if activity === 'worker-running'}
|
||||||
<span class="worker-status-spinner"><Spinner label="Running" /></span>
|
<span class="worker-status-spinner"><Spinner label="Running" /></span>
|
||||||
{:else if worker.has_running_internal_workers}
|
{:else if activity === 'subworker-running'}
|
||||||
<span class="worker-status-spinner is-subworker"><Spinner label="SubWorker running" /></span>
|
<span class="worker-status-spinner is-subworker"><Spinner label="SubWorker running" /></span>
|
||||||
{:else if worker.state === 'idle'}
|
{:else if activity === 'idle'}
|
||||||
<span class="worker-status-dot" aria-label="Idle"></span>
|
<span class="worker-status-dot" aria-label="Idle"></span>
|
||||||
{/if}
|
{/if}
|
||||||
</span>
|
</span>
|
||||||
|
|||||||
@@ -14,13 +14,18 @@ declare const Deno: {
|
|||||||
test(name: string, fn: () => void | Promise<void>): void;
|
test(name: string, fn: () => void | Promise<void>): void;
|
||||||
};
|
};
|
||||||
|
|
||||||
function worker(runtimeId: string, workerId: string, revision: number): SubscriptionWorker {
|
function worker(
|
||||||
|
runtimeId: string,
|
||||||
|
workerId: string,
|
||||||
|
revision: number,
|
||||||
|
hasRunningInternalWorkers = false,
|
||||||
|
): SubscriptionWorker {
|
||||||
return {
|
return {
|
||||||
worker_id: workerId,
|
worker_id: workerId,
|
||||||
runtime_id: runtimeId,
|
runtime_id: runtimeId,
|
||||||
subject_revision: revision,
|
subject_revision: revision,
|
||||||
state: 'idle',
|
state: 'idle',
|
||||||
has_running_internal_workers: false,
|
has_running_internal_workers: hasRunningInternalWorkers,
|
||||||
workspace_id: 'workspace-test',
|
workspace_id: 'workspace-test',
|
||||||
display_name: null,
|
display_name: null,
|
||||||
profile: null,
|
profile: null,
|
||||||
@@ -88,3 +93,28 @@ Deno.test('workspace Worker reducer ignores stale events and removes composite s
|
|||||||
assertEquals(projection.workers.size, 0);
|
assertEquals(projection.workers.size, 0);
|
||||||
assertEquals(projection.revisions.get('runtime-a:1'), 4);
|
assertEquals(projection.revisions.get('runtime-a:1'), 4);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
Deno.test('fatal child stop replaces the running-child sidebar projection', () => {
|
||||||
|
const projection = createWorkspaceWorkersProjection();
|
||||||
|
projection.workers.set('runtime-a:1', worker('runtime-a', '1', 1, true));
|
||||||
|
projection.revisions.set('runtime-a:1', 1);
|
||||||
|
|
||||||
|
applyWorkspaceWorkersFrame(projection, {
|
||||||
|
protocol_version: 1,
|
||||||
|
frame: 'event',
|
||||||
|
message: {
|
||||||
|
event: 'event',
|
||||||
|
data: {
|
||||||
|
subscription_id: 'subscription-1',
|
||||||
|
subject_revision: 2,
|
||||||
|
payload: {
|
||||||
|
event: 'worker_upserted',
|
||||||
|
data: { worker: worker('runtime-a', '1', 2, false) },
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
assertEquals(projection.workers.get('runtime-a:1')?.has_running_internal_workers, false);
|
||||||
|
assertEquals(projection.revisions.get('runtime-a:1'), 2);
|
||||||
|
});
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ import {
|
|||||||
canOpenWorkerConsole,
|
canOpenWorkerConsole,
|
||||||
canShowWorkerInSidebar,
|
canShowWorkerInSidebar,
|
||||||
compareWorkersForSidebar,
|
compareWorkersForSidebar,
|
||||||
|
sidebarWorkerActivity,
|
||||||
} from "./workers.ts";
|
} from "./workers.ts";
|
||||||
import type { Worker } from "./types.ts";
|
import type { Worker } from "./types.ts";
|
||||||
|
|
||||||
@@ -77,3 +78,21 @@ Deno.test("sidebar workers sort running then idle then stopped", () => {
|
|||||||
workers.sort(compareWorkersForSidebar);
|
workers.sort(compareWorkersForSidebar);
|
||||||
assertEquals(workers.map((candidate) => candidate.worker_id).join(","), "2,1,4,3");
|
assertEquals(workers.map((candidate) => candidate.worker_id).join(","), "2,1,4,3");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
Deno.test("fatal child stop clears the sidebar SubWorker spinner activity", () => {
|
||||||
|
const parent = { state: "idle", has_running_internal_workers: true };
|
||||||
|
assertEquals(sidebarWorkerActivity(parent), "subworker-running");
|
||||||
|
|
||||||
|
parent.has_running_internal_workers = false;
|
||||||
|
assertEquals(sidebarWorkerActivity(parent), "idle");
|
||||||
|
});
|
||||||
|
|
||||||
|
Deno.test("stopped parents do not fall back to the idle indicator", () => {
|
||||||
|
assertEquals(
|
||||||
|
sidebarWorkerActivity({
|
||||||
|
state: "stopped",
|
||||||
|
has_running_internal_workers: false,
|
||||||
|
}),
|
||||||
|
"none",
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|||||||
@@ -1,5 +1,24 @@
|
|||||||
import type { Worker } from './types';
|
import type { Worker } from './types';
|
||||||
|
|
||||||
|
export type SidebarWorkerActivity =
|
||||||
|
| 'worker-running'
|
||||||
|
| 'subworker-running'
|
||||||
|
| 'idle'
|
||||||
|
| 'none';
|
||||||
|
|
||||||
|
type WorkerActivitySource = Pick<Worker, 'state'> & {
|
||||||
|
has_running_internal_workers: boolean;
|
||||||
|
};
|
||||||
|
|
||||||
|
export function sidebarWorkerActivity(
|
||||||
|
worker: WorkerActivitySource,
|
||||||
|
): SidebarWorkerActivity {
|
||||||
|
if (worker.state === 'running') return 'worker-running';
|
||||||
|
if (worker.has_running_internal_workers) return 'subworker-running';
|
||||||
|
if (worker.state === 'idle') return 'idle';
|
||||||
|
return 'none';
|
||||||
|
}
|
||||||
|
|
||||||
export function canShowWorkerInSidebar(worker: Worker): boolean {
|
export function canShowWorkerInSidebar(worker: Worker): boolean {
|
||||||
return worker.implementation.kind !== 'backend_worker_registry';
|
return worker.implementation.kind !== 'backend_worker_registry';
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user