Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity
fdf1f43281
yoi/crates/worker-runtime/src/runtime.rs

2944 lines
107 KiB
Rust
Raw Blame History

use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerStatus,
WorkerSummary, WorkingDirectoryRequest,
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus,
};
use crate::config_bundle::{
ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle,
validate_config_bundle_ref,
};
#[cfg(feature = "fs-store")]
use crate::diagnostics::DiagnosticSeverity;
use crate::diagnostics::RuntimeDiagnostic;
use crate::error::RuntimeError;
#[cfg(feature = "fs-store")]
use crate::execution::WorkerExecutionRestoreRequest;
use crate::execution::{
WorkerExecutionBackend, WorkerExecutionBackendKind, WorkerExecutionBackendRef,
WorkerExecutionBindingIdentity, WorkerExecutionHandle, WorkerExecutionOperation,
WorkerExecutionResult, WorkerExecutionRunState, WorkerExecutionSpawnRequest,
WorkerExecutionSpawnResult, WorkerExecutionStatus,
};
#[cfg(feature = "fs-store")]
use crate::fs_store::{
FsRuntimeStore, FsRuntimeStoreOptions, PersistedRuntimeState, PersistedWorkerRecord,
};
use crate::identity::{WorkerId, WorkerRef};
use crate::interaction::{WorkerInput, WorkerInputKind, WorkerInteractionAck};
use crate::management::{
RuntimeBackendKind, RuntimeLimits, RuntimeOptions, RuntimeStatus, RuntimeSummary,
WorkerDeleteResult,
};
use crate::observation::{
EventCursor, EventSubscription, EventSubscriptionMode, RuntimeEvent, RuntimeEventBatch,
RuntimeEventKind,
};
#[cfg(feature = "ws-server")]
use crate::observation::{WorkerObservationCursor, WorkerObservationEvent};
use protocol::{Event, Method};
use std::collections::BTreeMap;
#[cfg(feature = "ws-server")]
use std::collections::VecDeque;
use std::sync::{Arc, Mutex, MutexGuard};
#[cfg(feature = "ws-server")]
use tokio::sync::broadcast;
/// Concrete embedded Runtime domain entity.
///
/// The default implementation is memory-backed and tools/provider-less by
/// design. An optional `fs-store` feature adds filesystem persistence while
/// preserving the same typed authority boundary. It can later be adapted by
/// backend registries or web servers without making sockets, sessions, or paths
/// public authority.
#[derive(Clone, Debug)]
pub struct Runtime {
inner: Arc<Mutex<RuntimeState>>,
}
impl Runtime {
/// Create a memory-backed Runtime with generated identity and default limits.
pub fn new_memory() -> Self {
Self::with_options(RuntimeOptions::default())
}
/// Create a memory-backed Runtime with explicit options.
pub fn with_options(options: RuntimeOptions) -> Self {
let mut state = RuntimeState::new(options.display_name, options.limits);
state.push_event(None, RuntimeEventKind::RuntimeStarted, "runtime started");
Self {
inner: Arc::new(Mutex::new(state)),
}
}
/// Create a memory-backed Runtime with an attached execution backend.
pub fn with_execution_backend(
options: RuntimeOptions,
backend: Arc<dyn WorkerExecutionBackend>,
) -> Result<Self, RuntimeError> {
let runtime = Self::with_options(options);
runtime.install_execution_backend(backend)?;
Ok(runtime)
}
/// Create or restore a filesystem-backed Runtime.
///
/// The store is scoped by `options.root`; if the directory already exists,
/// persisted state is loaded and validated. If it does not exist, a fresh
/// Runtime is initialized and durable files are created before return.
#[cfg(feature = "fs-store")]
pub fn with_fs_store(options: FsRuntimeStoreOptions) -> Result<Self, RuntimeError> {
Self::with_fs_store_inner(options, None)
}
/// Create or restore a filesystem-backed Runtime with an execution backend.
#[cfg(feature = "fs-store")]
pub fn with_fs_store_and_execution_backend(
options: FsRuntimeStoreOptions,
backend: Arc<dyn WorkerExecutionBackend>,
) -> Result<Self, RuntimeError> {
Self::with_fs_store_inner(options, Some(WorkerExecutionBackendRef::new(backend)?))
}
#[cfg(feature = "fs-store")]
fn with_fs_store_inner(
options: FsRuntimeStoreOptions,
execution_backend: Option<WorkerExecutionBackendRef>,
) -> Result<Self, RuntimeError> {
let opened = FsRuntimeStore::open_or_create(options.root)?;
let mut state = if let Some(persisted) = opened.state {
RuntimeState::from_persisted(persisted, opened.store)?
} else {
let mut state =
RuntimeState::new_fs_backed(options.display_name, options.limits, opened.store);
let event_id =
state.push_event(None, RuntimeEventKind::RuntimeStarted, "runtime started");
state.persist_runtime_snapshot()?;
state.persist_event_by_id(event_id)?;
state
};
state.execution_backend = execution_backend;
let runtime = Self {
inner: Arc::new(Mutex::new(state)),
};
runtime.restore_persisted_worker_executions()?;
Ok(runtime)
}
/// Management-plane summary.
pub fn summary(&self) -> Result<RuntimeSummary, RuntimeError> {
let state = self.lock()?;
let mut active_worker_count = 0;
let mut stopped_worker_count = 0;
let mut cancelled_worker_count = 0;
for worker in state.workers.values() {
match worker.status {
WorkerStatus::Running => active_worker_count += 1,
WorkerStatus::Stopped => stopped_worker_count += 1,
WorkerStatus::Cancelled => cancelled_worker_count += 1,
}
}
Ok(RuntimeSummary {
display_name: state.display_name.clone(),
backend: state.backend,
status: state.status,
worker_count: state.workers.len(),
active_worker_count,
stopped_worker_count,
cancelled_worker_count,
diagnostic_count: state.diagnostics.len(),
limits: state.limits.clone(),
os: std::env::consts::OS.to_string(),
arch: std::env::consts::ARCH.to_string(),
worker_creation_available: state.execution_backend.is_some(),
})
}
/// Current Runtime lifecycle state.
pub fn status(&self) -> Result<RuntimeStatus, RuntimeError> {
Ok(self.lock()?.status)
}
/// Store a backend-synced Profile/config bundle for later Worker creation.
pub fn store_config_bundle(
&self,
bundle: ConfigBundle,
) -> Result<ConfigBundleAvailability, RuntimeError> {
validate_config_bundle(&bundle)?;
let mut state = self.lock()?;
state.ensure_running()?;
let reference = ConfigBundleRef {
id: bundle.metadata.id.clone(),
digest: bundle.metadata.digest.clone(),
};
let summary = bundle.summary();
state
.config_bundles
.insert(bundle.metadata.id.clone(), bundle);
state.persist_runtime_snapshot()?;
Ok(ConfigBundleAvailability { reference, summary })
}
/// List synced config bundles known to this Runtime.
pub fn list_config_bundles(&self) -> Result<Vec<ConfigBundleSummary>, RuntimeError> {
Ok(self
.lock()?
.config_bundles
.values()
.map(ConfigBundle::summary)
.collect())
}
/// Validate that a config bundle reference is present and digest-matched.
pub fn check_config_bundle(
&self,
reference: &ConfigBundleRef,
) -> Result<ConfigBundleAvailability, RuntimeError> {
let state = self.lock()?;
state.check_config_bundle_ref(reference)
}
/// Stop the Runtime. v0 keeps data readable after stop, but rejects new
/// create/send/worker lifecycle mutations.
pub fn stop_runtime(&self) -> Result<u64, RuntimeError> {
let mut state = self.lock()?;
if state.status == RuntimeStatus::Stopped {
return Ok(state.last_event_id());
}
state.status = RuntimeStatus::Stopped;
for worker in state.workers.values_mut() {
if worker.status.is_active() {
worker.status = WorkerStatus::Stopped;
}
}
let event_id = state.push_event(None, RuntimeEventKind::RuntimeStopped, "runtime stopped");
state.persist_runtime_snapshot()?;
state.persist_workers()?;
state.persist_event_by_id(event_id)?;
Ok(event_id)
}
/// Create a Runtime-owned working directory through the attached execution backend.
pub fn create_working_directory(
&self,
request: WorkingDirectoryRequest,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory creation requires an execution backend".to_string(),
}
})?
};
backend
.create_working_directory(&request)
.map_err(RuntimeError::from)
}
/// List Runtime-owned working directories through the attached execution backend.
pub fn list_working_directories(
&self,
) -> Result<Vec<CatalogWorkingDirectoryStatus>, RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory listing requires an execution backend".to_string(),
}
})?
};
let statuses = backend.list_working_directories();
self.annotate_working_directory_statuses(statuses)
}
/// Get a Runtime-owned working directory status.
pub fn working_directory(
&self,
working_directory_id: &str,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory lookup requires an execution backend".to_string(),
}
})?
};
let status = backend
.working_directory(working_directory_id)
.map_err(RuntimeError::from)?;
self.annotate_working_directory_status(status)
}
/// Cleanup a Runtime-owned working directory.
pub fn cleanup_working_directory(
&self,
working_directory_id: &str,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
if let Some(worker_id) = state.primary_worker_id_for_workdir(working_directory_id) {
return Err(RuntimeError::InvalidRequest(format!(
"working directory {working_directory_id} is assigned to worker {worker_id}"
)));
}
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory cleanup requires an execution backend".to_string(),
}
})?
};
backend
.cleanup_working_directory(working_directory_id)
.map_err(RuntimeError::from)
}
fn annotate_working_directory_statuses(
&self,
statuses: Vec<CatalogWorkingDirectoryStatus>,
) -> Result<Vec<CatalogWorkingDirectoryStatus>, RuntimeError> {
statuses
.into_iter()
.map(|status| self.annotate_working_directory_status(status))
.collect()
}
fn annotate_working_directory_status(
&self,
mut status: CatalogWorkingDirectoryStatus,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
let state = self.lock()?;
status.summary.primary_worker_id =
state.primary_worker_id_for_workdir(status.summary.working_directory_id.as_str());
Ok(status)
}
/// Create a Worker through the canonical profile-source + execution backend path.
pub fn create_worker(
&self,
request: CreateWorkerRequest,
) -> Result<WorkerDetail, RuntimeError> {
let (backend, worker_ref, spawn_request) = {
let mut state = self.lock()?;
state.ensure_running()?;
validate_create_worker_request(&request)?;
state.validate_worker_config_boundary(&request)?;
if let Some(working_directory_id) = requested_primary_workdir_id(&request) {
if let Some(owner_worker_id) =
state.primary_worker_id_for_workdir(working_directory_id)
{
return Err(RuntimeError::InvalidRequest(format!(
"working directory {working_directory_id} is already assigned to worker {owner_worker_id}"
)));
}
}
let backend = state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "worker creation requires an execution backend".to_string(),
}
})?;
let worker_id = WorkerId::generated(state.next_worker_sequence);
state.next_worker_sequence += 1;
let worker_ref = WorkerRef::new(worker_id.clone());
let event_id = state.push_event(
Some(worker_ref.clone()),
RuntimeEventKind::WorkerCreated,
format!("worker {worker_id} created"),
);
let record = WorkerRecord {
worker_ref: worker_ref.clone(),
worker_id: worker_id.clone(),
status: WorkerStatus::Running,
request: request.clone(),
execution: WorkerExecutionStatus::unconnected(),
execution_handle: None,
last_event_id: event_id,
};
state.workers.insert(worker_id, record);
let spawn_request = WorkerExecutionSpawnRequest {
worker_ref: worker_ref.clone(),
request,
context: self.execution_context(worker_ref.clone()),
working_directory: None,
config_bundle: None,
};
(backend, worker_ref, spawn_request)
};
let spawn_result = backend.spawn_worker(spawn_request);
let (handle, run_state, working_directory) = match spawn_result {
WorkerExecutionSpawnResult::Connected {
handle,
run_state,
working_directory,
} => (handle, run_state, working_directory),
WorkerExecutionSpawnResult::Rejected(result)
| WorkerExecutionSpawnResult::Errored(result) => {
self.rollback_failed_create(&worker_ref)?;
return Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id.clone(),
operation: result.operation,
outcome: result.outcome,
message: result.message_or_default(),
result,
});
}
};
if let Some(initial_input) = {
let state = self.lock()?;
state.worker(&worker_ref)?.request.initial_input.clone()
} {
let dispatch_result = backend.dispatch_input(&handle, initial_input.clone());
if !dispatch_result.is_accepted() {
let _ = backend.stop_worker(&handle);
self.rollback_failed_create(&worker_ref)?;
return Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id.clone(),
operation: dispatch_result.operation,
outcome: dispatch_result.outcome,
message: dispatch_result.message_or_default(),
result: dispatch_result,
});
}
let detail = self.commit_created_worker(
&worker_ref,
handle,
WorkerExecutionRunState::Busy,
working_directory,
WorkerExecutionResult::accepted(
WorkerExecutionOperation::Input,
WorkerExecutionRunState::Busy,
),
)?;
self.record_input_observation(&worker_ref, initial_input)?;
Ok(detail)
} else {
self.commit_created_worker(
&worker_ref,
handle,
run_state,
working_directory,
WorkerExecutionResult::accepted(WorkerExecutionOperation::Spawn, run_state),
)
}
}
/// List Workers known to this Runtime.
pub fn list_workers(&self) -> Result<Vec<WorkerSummary>, RuntimeError> {
let state = self.lock()?;
Ok(state
.workers
.values()
.map(|worker| worker.summary())
.collect())
}
/// Fetch Worker detail. The supplied [`WorkerRef`] must match this Runtime.
pub fn worker_detail(&self, worker_ref: &WorkerRef) -> Result<WorkerDetail, RuntimeError> {
let state = self.lock()?;
let worker = state.worker(worker_ref)?;
Ok(worker.detail())
}
/// Accept input into a Worker.
pub fn send_input(
&self,
worker_ref: &WorkerRef,
input: WorkerInput,
) -> Result<WorkerInteractionAck, RuntimeError> {
let (backend, handle) = {
let mut state = self.lock()?;
state.ensure_running()?;
validate_worker_input(&input)?;
state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(worker_ref)?;
if !worker.status.is_active() {
return Err(RuntimeError::InvalidRequest(format!(
"worker {} is not running",
worker_ref.worker_id
)));
}
let backend = state.execution_backend.clone();
let handle = worker.execution_handle.clone();
match (backend, handle) {
(Some(backend), Some(handle)) => (backend, handle),
_ => {
let result = WorkerExecutionResult::rejected(
WorkerExecutionOperation::Input,
"worker has no execution backend",
);
let worker = state.worker_mut(worker_ref)?;
let mut execution = WorkerExecutionStatus::unconnected().with_result(result);
execution.binding = worker.execution.binding.clone();
worker.execution = execution;
state.persist_worker(&worker_ref.worker_id)?;
return Err(RuntimeError::WorkerExecutionUnavailable {
worker_id: worker_ref.worker_id.clone(),
message: "worker has no execution backend".to_string(),
});
}
}
};
let dispatch_result = backend.dispatch_input(&handle, input.clone());
if !dispatch_result.is_accepted() {
self.record_execution_result(worker_ref, dispatch_result.clone())?;
return Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id.clone(),
operation: dispatch_result.operation,
outcome: dispatch_result.outcome,
message: dispatch_result.message_or_default(),
result: dispatch_result,
});
}
let mut state = self.lock()?;
state.ensure_running()?;
let event_id = state.push_event(
Some(worker_ref.clone()),
RuntimeEventKind::WorkerInputAccepted,
"worker input accepted",
);
let worker = state.worker_mut(worker_ref)?;
worker.last_event_id = event_id;
worker.execution = WorkerExecutionStatus {
backend: WorkerExecutionBackendKind::Connected,
run_state: dispatch_result.run_state,
binding: worker.execution.binding.clone(),
working_directory: worker.execution.working_directory.clone(),
last_result: Some(dispatch_result),
};
let status = worker.status;
#[cfg(feature = "ws-server")]
let observation = {
let payload = input_protocol_event(&input);
Some(state.push_worker_observation_event(worker_ref.clone(), payload))
};
state.persist_runtime_snapshot()?;
state.persist_worker(&worker_ref.worker_id)?;
state.persist_event_by_id(event_id)?;
#[cfg(feature = "ws-server")]
if let Some(observation) = observation.as_ref() {
state.persist_worker_observation_event(observation)?;
}
Ok(WorkerInteractionAck {
worker_ref: worker_ref.clone(),
status,
event_id,
})
}
/// Return live completion entries for the Worker composer.
pub fn worker_completions(
&self,
worker_ref: &WorkerRef,
kind: protocol::CompletionKind,
prefix: &str,
) -> Result<Vec<protocol::CompletionEntry>, RuntimeError> {
let (backend, handle) = {
let state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(worker_ref)?;
(
state.execution_backend.clone(),
worker.execution_handle.clone(),
)
};
let Some((backend, handle)) = backend.zip(handle) else {
return Ok(Vec::new());
};
Ok(backend.worker_completions(&handle, kind, prefix))
}
/// Accept a protocol method for a Worker through a Backend/runtime transport.
///
/// Most methods are delivered to the execution backend unchanged. Methods with
/// direct same-connection replies in the local socket protocol return those
/// events from this function so WebSocket transports can write them back to the
/// requesting client without rebroadcasting them.
pub fn send_protocol_method(
&self,
worker_ref: &WorkerRef,
method: Method,
) -> Result<Vec<Event>, RuntimeError> {
if let Method::ListCompletions { kind, prefix } = method {
let entries = self.worker_completions(worker_ref, kind, &prefix)?;
return Ok(vec![Event::Completions { kind, entries }]);
}
let observation_payload = protocol_method_observation_event(&method);
let (backend, handle) = {
let mut state = self.lock()?;
state.ensure_running()?;
state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(worker_ref)?;
if !worker.status.is_active() {
return Err(RuntimeError::InvalidRequest(format!(
"worker {} is not running",
worker_ref.worker_id
)));
}
let backend = state.execution_backend.clone();
let handle = worker.execution_handle.clone();
match (backend, handle) {
(Some(backend), Some(handle)) => (backend, handle),
_ => {
let result = WorkerExecutionResult::rejected(
WorkerExecutionOperation::ProtocolMethod,
"worker has no execution backend",
);
let worker = state.worker_mut(worker_ref)?;
let mut execution = WorkerExecutionStatus::unconnected().with_result(result);
execution.binding = worker.execution.binding.clone();
worker.execution = execution;
state.persist_worker(&worker_ref.worker_id)?;
return Err(RuntimeError::WorkerExecutionUnavailable {
worker_id: worker_ref.worker_id.clone(),
message: "worker has no execution backend".to_string(),
});
}
}
};
let dispatch_result = backend.dispatch_method(&handle, method);
if !dispatch_result.is_accepted() {
self.record_execution_result(worker_ref, dispatch_result.clone())?;
return Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id.clone(),
operation: dispatch_result.operation,
outcome: dispatch_result.outcome,
message: dispatch_result.message_or_default(),
result: dispatch_result,
});
}
self.record_execution_result(worker_ref, dispatch_result)?;
if let Some(payload) = observation_payload {
self.record_protocol_method_observation(worker_ref, payload)?;
}
Ok(Vec::new())
}
fn commit_created_worker(
&self,
worker_ref: &WorkerRef,
handle: WorkerExecutionHandle,
run_state: WorkerExecutionRunState,
working_directory: Option<CatalogWorkingDirectoryStatus>,
result: WorkerExecutionResult,
) -> Result<WorkerDetail, RuntimeError> {
let mut state = self.lock()?;
let detail = {
let binding = WorkerExecutionBindingIdentity::from_handle(&handle);
let worker = state.worker_mut(worker_ref)?;
worker.execution_handle = Some(handle);
let mut execution = WorkerExecutionStatus::connected(run_state).with_binding(binding);
if let Some(status) = working_directory {
execution = execution.with_working_directory(status);
}
worker.execution = execution.with_result(result);
worker.detail()
};
state.persist_runtime_snapshot()?;
state.persist_worker(&worker_ref.worker_id)?;
state.persist_event_by_id(detail.last_event_id)?;
Ok(detail)
}
fn rollback_failed_create(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
if let Some(record) = state.workers.remove(&worker_ref.worker_id) {
state.events.retain(|event| {
event.id != record.last_event_id || event.worker_ref.as_ref() != Some(worker_ref)
});
}
Ok(())
}
fn record_execution_result(
&self,
worker_ref: &WorkerRef,
result: WorkerExecutionResult,
) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
let worker = state.worker_mut(worker_ref)?;
worker.execution = WorkerExecutionStatus {
backend: WorkerExecutionBackendKind::Connected,
run_state: result.run_state,
binding: worker.execution.binding.clone(),
working_directory: worker.execution.working_directory.clone(),
last_result: Some(result),
};
state.persist_worker(&worker_ref.worker_id)?;
Ok(())
}
fn dispatch_lifecycle_to_backend(
&self,
worker_ref: &WorkerRef,
operation: WorkerExecutionOperation,
) -> Result<(), RuntimeError> {
let Some((backend, handle)) = ({
let state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(worker_ref)?;
if !worker.status.is_active() {
return Ok(());
}
match (
state.execution_backend.clone(),
worker.execution_handle.clone(),
) {
(Some(backend), Some(handle)) => Some((backend, handle)),
_ => None,
}
}) else {
return Ok(());
};
let result = match operation {
WorkerExecutionOperation::Stop => backend.stop_worker(&handle),
WorkerExecutionOperation::Cancel => backend.cancel_worker(&handle),
WorkerExecutionOperation::Spawn
| WorkerExecutionOperation::Restore
| WorkerExecutionOperation::Input
| WorkerExecutionOperation::ProtocolMethod => return Ok(()),
};
if result.is_accepted() {
self.record_execution_result(worker_ref, result)?;
return Ok(());
}
self.record_execution_result(worker_ref, result.clone())?;
Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id.clone(),
operation: result.operation,
outcome: result.outcome,
message: result.message_or_default(),
result,
})
}
/// Stop a Worker. Repeated stops are idempotent and return the last event id.
pub fn stop_worker(
&self,
worker_ref: &WorkerRef,
reason: Option<String>,
) -> Result<WorkerLifecycleAck, RuntimeError> {
self.dispatch_lifecycle_to_backend(worker_ref, WorkerExecutionOperation::Stop)?;
self.transition_worker(
worker_ref,
WorkerStatus::Stopped,
RuntimeEventKind::WorkerStopped,
reason.unwrap_or_else(|| "worker stopped".to_string()),
)
}
/// Cancel a Worker. Repeated cancels are idempotent and return the last event id.
pub fn cancel_worker(
&self,
worker_ref: &WorkerRef,
reason: Option<String>,
) -> Result<WorkerLifecycleAck, RuntimeError> {
self.dispatch_lifecycle_to_backend(worker_ref, WorkerExecutionOperation::Cancel)?;
self.transition_worker(
worker_ref,
WorkerStatus::Cancelled,
RuntimeEventKind::WorkerCancelled,
reason.unwrap_or_else(|| "worker cancelled".to_string()),
)
}
/// Delete a non-running Worker from Runtime state and persisted Worker storage.
pub fn delete_worker(
&self,
worker_ref: &WorkerRef,
) -> Result<WorkerDeleteResult, RuntimeError> {
let mut state = self.lock()?;
state.ensure_running()?;
state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(worker_ref)?;
if worker.status.is_active() {
return Err(RuntimeError::InvalidRequest(format!(
"worker {} is running and must be stopped before deletion",
worker_ref.worker_id
)));
}
let removed = state.workers.remove(&worker_ref.worker_id).ok_or_else(|| {
RuntimeError::WorkerNotFound {
worker_id: worker_ref.worker_id,
}
})?;
#[cfg(feature = "ws-server")]
state
.observation_events
.retain(|event| event.worker_ref != *worker_ref);
let event_id = state.push_event(
Some(worker_ref.clone()),
RuntimeEventKind::WorkerDeleted,
"worker deleted",
);
state.persist_runtime_snapshot()?;
state.delete_worker_snapshot(&worker_ref.worker_id)?;
state.persist_event_by_id(event_id)?;
Ok(WorkerDeleteResult {
worker_id: removed.worker_id,
deleted: true,
})
}
/// Cursor pointing to the beginning of Runtime events.
pub fn event_cursor_from_start(&self) -> Result<EventCursor, RuntimeError> {
Ok(EventCursor { next_event_id: 1 })
}
/// Cursor pointing after the current last event.
pub fn event_cursor_now(&self) -> Result<EventCursor, RuntimeError> {
let state = self.lock()?;
Ok(EventCursor {
next_event_id: state.last_event_id() + 1,
})
}
/// Poll Runtime events from a cursor.
pub fn read_events(
&self,
cursor: &EventCursor,
limit: usize,
) -> Result<RuntimeEventBatch, RuntimeError> {
let state = self.lock()?;
if limit > state.limits.max_event_batch_items {
return Err(RuntimeError::LimitTooLarge {
requested: limit,
max: state.limits.max_event_batch_items,
});
}
let mut events = Vec::new();
for event in state
.events
.iter()
.filter(|event| event.id >= cursor.next_event_id)
.take(limit)
{
events.push(event.clone());
}
let next_event_id = events
.last()
.map(|event| event.id + 1)
.unwrap_or(cursor.next_event_id);
let has_more = state.events.iter().any(|event| event.id >= next_event_id);
Ok(RuntimeEventBatch {
cursor: EventCursor { next_event_id },
events,
has_more,
})
}
/// Create a poll-only placeholder subscription boundary for future streaming.
pub fn subscribe_events(&self, cursor: EventCursor) -> Result<EventSubscription, RuntimeError> {
Ok(EventSubscription {
cursor,
mode: EventSubscriptionMode::PollOnly,
})
}
/// Cursor pointing after the current worker-scoped protocol observation event.
#[cfg(feature = "ws-server")]
pub fn worker_observation_cursor_now(
&self,
worker_ref: &WorkerRef,
) -> Result<WorkerObservationCursor, RuntimeError> {
let state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
let sequence = state
.observation_events
.iter()
.rev()
.find(|event| &event.worker_ref == worker_ref)
.map(|event| event.sequence)
.unwrap_or(0);
Ok(WorkerObservationCursor::new(sequence))
}
/// Build the current Worker Snapshot event used as the first observation frame.
#[cfg(feature = "ws-server")]
pub fn worker_observation_snapshot(
&self,
worker_ref: &WorkerRef,
) -> Result<protocol::Event, RuntimeError> {
let (backend, handle) = {
let state = self.lock()?;
let worker = state.worker(worker_ref)?;
(
state.execution_backend.clone(),
worker.execution_handle.clone(),
)
};
if let (Some(backend), Some(handle)) = (backend, handle) {
if let Some(snapshot) = backend.worker_snapshot(&handle) {
return Ok(snapshot);
}
}
Ok(protocol::Event::Snapshot {
entries: Vec::new(),
greeting: protocol::Greeting {
worker_name: worker_ref.worker_id.to_string(),
cwd: String::new(),
provider: "worker-runtime".to_string(),
model: "worker-runtime".to_string(),
scope_summary: "runtime worker observation".to_string(),
tools: Vec::new(),
context_window: 0,
context_tokens: 0,
},
status: protocol::WorkerStatus::Idle,
in_flight: protocol::InFlightSnapshot { blocks: Vec::new() },
})
}
/// Replay retained worker-scoped protocol observation events after a cursor.
#[cfg(feature = "ws-server")]
pub fn read_worker_observation_events(
&self,
worker_ref: &WorkerRef,
cursor: WorkerObservationCursor,
) -> Result<Vec<WorkerObservationEvent>, RuntimeError> {
let state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
state.validate_worker_observation_cursor(worker_ref, cursor)?;
Ok(state
.observation_events
.iter()
.filter(|event| &event.worker_ref == worker_ref && event.sequence > cursor.sequence)
.cloned()
.collect())
}
/// Subscribe to live protocol observation events.
#[cfg(feature = "ws-server")]
pub fn subscribe_worker_observation(
&self,
) -> Result<broadcast::Receiver<WorkerObservationEvent>, RuntimeError> {
Ok(self.lock()?.observation_tx.subscribe())
}
/// Append a Worker protocol event to the observation bus.
#[cfg(feature = "ws-server")]
pub fn observe_worker_event(
&self,
worker_ref: &WorkerRef,
payload: protocol::Event,
) -> Result<WorkerObservationEvent, RuntimeError> {
let mut state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
let execution_state_changed =
state.project_protocol_event_to_execution(worker_ref, &payload);
let event = state.push_worker_observation_event(worker_ref.clone(), payload);
if execution_state_changed {
state.persist_worker(&worker_ref.worker_id)?;
}
state.persist_worker_observation_event(&event)?;
Ok(event)
}
/// Snapshot current diagnostics.
pub fn diagnostics(&self) -> Result<Vec<RuntimeDiagnostic>, RuntimeError> {
Ok(self.lock()?.diagnostics.clone())
}
#[cfg(feature = "ws-server")]
fn record_input_observation(
&self,
worker_ref: &WorkerRef,
input: WorkerInput,
) -> Result<(), RuntimeError> {
self.record_protocol_method_observation(worker_ref, input_protocol_event(&input))
}
#[cfg(not(feature = "ws-server"))]
fn record_input_observation(
&self,
_worker_ref: &WorkerRef,
_input: WorkerInput,
) -> Result<(), RuntimeError> {
Ok(())
}
#[cfg(feature = "ws-server")]
fn record_protocol_method_observation(
&self,
worker_ref: &WorkerRef,
payload: Event,
) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
let event = state.push_worker_observation_event(worker_ref.clone(), payload);
state.persist_worker_observation_event(&event)?;
Ok(())
}
#[cfg(not(feature = "ws-server"))]
fn record_protocol_method_observation(
&self,
_worker_ref: &WorkerRef,
_payload: Event,
) -> Result<(), RuntimeError> {
Ok(())
}
fn transition_worker(
&self,
worker_ref: &WorkerRef,
status: WorkerStatus,
event_kind: RuntimeEventKind,
reason: String,
) -> Result<WorkerLifecycleAck, RuntimeError> {
let mut state = self.lock()?;
state.ensure_running()?;
state.ensure_worker_ref(worker_ref)?;
{
let worker = state.worker(worker_ref)?;
if !worker.status.is_active() {
return Ok(WorkerLifecycleAck {
worker_ref: worker_ref.clone(),
status: worker.status,
event_id: worker.last_event_id,
});
}
}
let event_id = state.push_event(Some(worker_ref.clone()), event_kind, reason);
let worker = state.worker_mut(worker_ref)?;
worker.status = status;
worker.last_event_id = event_id;
let status = worker.status;
state.persist_runtime_snapshot()?;
state.persist_worker(&worker_ref.worker_id)?;
state.persist_event_by_id(event_id)?;
Ok(WorkerLifecycleAck {
worker_ref: worker_ref.clone(),
status,
event_id,
})
}
fn install_execution_backend(
&self,
backend: Arc<dyn WorkerExecutionBackend>,
) -> Result<(), RuntimeError> {
let backend = WorkerExecutionBackendRef::new(backend)?;
let mut state = self.lock()?;
state.execution_backend = Some(backend);
Ok(())
}
#[cfg(feature = "ws-server")]
fn execution_context(&self, worker_ref: WorkerRef) -> crate::execution::WorkerExecutionContext {
let runtime = self.clone();
crate::execution::WorkerExecutionContext::new(
worker_ref,
Arc::new(move |worker_ref, payload| runtime.observe_worker_event(&worker_ref, payload)),
)
}
#[cfg(not(feature = "ws-server"))]
fn execution_context(&self, worker_ref: WorkerRef) -> crate::execution::WorkerExecutionContext {
crate::execution::WorkerExecutionContext::new(worker_ref)
}
#[cfg(feature = "fs-store")]
fn restore_persisted_worker_executions(&self) -> Result<(), RuntimeError> {
#[derive(Clone)]
struct RestoreCandidate {
worker_ref: WorkerRef,
request: CreateWorkerRequest,
previous_execution: WorkerExecutionStatus,
}
let candidates = {
let mut state = self.lock()?;
let Some(backend) = state.execution_backend.clone() else {
return Ok(());
};
let backend_id = backend.backend_id().to_string();
let mut candidates = Vec::new();
let worker_ids: Vec<_> = state.workers.keys().cloned().collect();
for worker_id in worker_ids {
let Some(worker) = state.workers.get(&worker_id) else {
continue;
};
if !worker.status.is_active()
|| worker.execution_handle.is_some()
|| worker.execution.backend != WorkerExecutionBackendKind::Stale
|| worker
.execution
.binding
.as_ref()
.is_none_or(|binding| binding.backend_id != backend_id)
{
continue;
}
if let Some(working_directory) = worker.execution.working_directory.as_ref() {
if let Some(owner) = state.active_primary_worker_id_for_workdir_excluding(
&working_directory.summary.working_directory_id,
&worker.worker_id,
) {
let worker_ref = worker.worker_ref.clone();
let message = format!(
"worker {} cannot restore working directory {} because active worker {} is already the primary assignment",
worker.worker_id, working_directory.summary.working_directory_id, owner
);
state.record_restore_failure(
&worker_ref,
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
message,
),
)?;
continue;
}
}
candidates.push(RestoreCandidate {
worker_ref: worker.worker_ref.clone(),
request: worker.request.clone(),
previous_execution: worker.execution.clone(),
});
}
candidates
};
for candidate in candidates {
let backend = {
let state = self.lock()?;
state.execution_backend.clone()
};
let Some(backend) = backend else {
return Ok(());
};
let request = WorkerExecutionRestoreRequest {
worker_ref: candidate.worker_ref.clone(),
request: candidate.request,
context: self.execution_context(candidate.worker_ref.clone()),
previous_execution: candidate.previous_execution,
working_directory: None,
config_bundle: None,
};
match backend.restore_worker(request) {
WorkerExecutionSpawnResult::Connected {
handle,
run_state,
working_directory,
} => self.commit_restored_worker_execution(
&candidate.worker_ref,
handle,
run_state,
working_directory,
)?,
WorkerExecutionSpawnResult::Rejected(result)
| WorkerExecutionSpawnResult::Errored(result) => {
let mut state = self.lock()?;
state.record_restore_failure(&candidate.worker_ref, result)?;
}
}
}
Ok(())
}
#[cfg(feature = "fs-store")]
fn commit_restored_worker_execution(
&self,
worker_ref: &WorkerRef,
handle: WorkerExecutionHandle,
run_state: WorkerExecutionRunState,
working_directory: Option<CatalogWorkingDirectoryStatus>,
) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
let event_id = state.push_event(
Some(worker_ref.clone()),
RuntimeEventKind::WorkerExecutionRestored,
format!("worker {} execution restored", worker_ref.worker_id),
);
{
let worker = state.worker_mut(worker_ref)?;
worker.execution_handle = Some(handle.clone());
let mut execution = WorkerExecutionStatus::connected(run_state)
.with_binding(WorkerExecutionBindingIdentity::from_handle(&handle));
if let Some(status) = working_directory {
execution = execution.with_working_directory(status);
}
worker.execution = execution;
worker.last_event_id = event_id;
}
state.persist_runtime_snapshot()?;
state.persist_worker(&worker_ref.worker_id)?;
state.persist_event_by_id(event_id)?;
Ok(())
}
fn lock(&self) -> Result<MutexGuard<'_, RuntimeState>, RuntimeError> {
self.inner.lock().map_err(|_| RuntimeError::StatePoisoned)
}
}
#[cfg_attr(not(feature = "fs-store"), allow(dead_code))]
#[derive(Clone, Debug)]
enum RuntimePersistence {
Memory,
#[cfg(feature = "fs-store")]
Fs(FsRuntimeStore),
}
#[derive(Debug)]
struct RuntimeState {
display_name: Option<String>,
backend: RuntimeBackendKind,
#[cfg_attr(not(feature = "fs-store"), allow(dead_code))]
persistence: RuntimePersistence,
status: RuntimeStatus,
limits: RuntimeLimits,
execution_backend: Option<WorkerExecutionBackendRef>,
next_worker_sequence: u64,
next_event_id: u64,
#[cfg(feature = "fs-store")]
next_diagnostic_id: u64,
workers: BTreeMap<WorkerId, WorkerRecord>,
config_bundles: BTreeMap<String, ConfigBundle>,
events: Vec<RuntimeEvent>,
diagnostics: Vec<RuntimeDiagnostic>,
#[cfg(feature = "ws-server")]
next_observation_sequence: u64,
#[cfg(feature = "ws-server")]
observation_events: VecDeque<WorkerObservationEvent>,
#[cfg(feature = "ws-server")]
observation_tx: broadcast::Sender<WorkerObservationEvent>,
}
impl RuntimeState {
fn new(display_name: Option<String>, limits: RuntimeLimits) -> Self {
Self {
display_name,
backend: RuntimeBackendKind::Memory,
persistence: RuntimePersistence::Memory,
status: RuntimeStatus::Running,
limits,
execution_backend: None,
next_worker_sequence: 1,
next_event_id: 1,
#[cfg(feature = "fs-store")]
next_diagnostic_id: 1,
workers: BTreeMap::new(),
config_bundles: BTreeMap::new(),
events: Vec::new(),
diagnostics: Vec::new(),
#[cfg(feature = "ws-server")]
next_observation_sequence: 1,
#[cfg(feature = "ws-server")]
observation_events: VecDeque::new(),
#[cfg(feature = "ws-server")]
observation_tx: broadcast::channel(256).0,
}
}
#[cfg(feature = "fs-store")]
fn new_fs_backed(
display_name: Option<String>,
limits: RuntimeLimits,
store: FsRuntimeStore,
) -> Self {
Self {
display_name,
backend: RuntimeBackendKind::FsStore,
persistence: RuntimePersistence::Fs(store),
status: RuntimeStatus::Running,
limits,
execution_backend: None,
next_worker_sequence: 1,
next_event_id: 1,
#[cfg(feature = "fs-store")]
next_diagnostic_id: 1,
workers: BTreeMap::new(),
config_bundles: BTreeMap::new(),
events: Vec::new(),
diagnostics: Vec::new(),
#[cfg(feature = "ws-server")]
next_observation_sequence: 1,
#[cfg(feature = "ws-server")]
observation_events: VecDeque::new(),
#[cfg(feature = "ws-server")]
observation_tx: broadcast::channel(256).0,
}
}
#[cfg(feature = "fs-store")]
fn from_persisted(
persisted: PersistedRuntimeState,
store: FsRuntimeStore,
) -> Result<Self, RuntimeError> {
let mut workers = BTreeMap::new();
let mut diagnostics = persisted.diagnostics;
let mut next_diagnostic_id = persisted.next_diagnostic_id;
for (worker_id, worker) in persisted.workers {
let execution = if worker.execution.binding.is_some()
&& worker.execution.backend == WorkerExecutionBackendKind::Connected
{
let stale = WorkerExecutionStatus::stale(worker.execution);
diagnostics.push(RuntimeDiagnostic {
id: next_diagnostic_id,
severity: DiagnosticSeverity::Warning,
code: "worker_execution_mapping_stale".to_string(),
message: format!(
"worker {} has persisted execution binding identity but no live execution handle was restored",
worker.worker_id
),
worker_ref: Some(worker.worker_ref.clone()),
});
next_diagnostic_id += 1;
stale
} else {
worker.execution
};
workers.insert(
worker_id,
WorkerRecord {
worker_ref: worker.worker_ref,
worker_id: worker.worker_id,
status: worker.status,
request: worker.request,
execution,
execution_handle: None,
last_event_id: worker.last_event_id,
},
);
}
#[cfg(feature = "ws-server")]
let next_observation_sequence = persisted
.observation_events
.iter()
.map(|event| event.sequence)
.max()
.map(|sequence| sequence.saturating_add(1))
.unwrap_or(1);
#[cfg(feature = "ws-server")]
let observation_events = persisted.observation_events.into_iter().collect();
Ok(Self {
display_name: persisted.display_name,
backend: RuntimeBackendKind::FsStore,
persistence: RuntimePersistence::Fs(store),
status: persisted.status,
limits: persisted.limits,
execution_backend: None,
next_worker_sequence: persisted.next_worker_sequence,
next_event_id: persisted.next_event_id,
next_diagnostic_id,
workers,
config_bundles: persisted.config_bundles,
events: persisted.events,
diagnostics,
#[cfg(feature = "ws-server")]
next_observation_sequence,
#[cfg(feature = "ws-server")]
observation_events,
#[cfg(feature = "ws-server")]
observation_tx: broadcast::channel(256).0,
})
}
#[cfg(feature = "fs-store")]
fn persisted_state(&self) -> PersistedRuntimeState {
PersistedRuntimeState {
display_name: self.display_name.clone(),
status: self.status,
limits: self.limits.clone(),
next_worker_sequence: self.next_worker_sequence,
next_event_id: self.next_event_id,
next_diagnostic_id: self.next_diagnostic_id,
workers: self
.workers
.iter()
.map(|(worker_id, worker)| (worker_id.clone(), worker.persisted_record()))
.collect(),
config_bundles: self.config_bundles.clone(),
events: self.events.clone(),
#[cfg(feature = "ws-server")]
observation_events: Vec::new(),
diagnostics: self.diagnostics.clone(),
}
}
#[cfg(feature = "fs-store")]
fn fs_store(&self) -> Option<&FsRuntimeStore> {
match &self.persistence {
RuntimePersistence::Memory => None,
RuntimePersistence::Fs(store) => Some(store),
}
}
#[cfg(feature = "fs-store")]
fn persist_runtime_snapshot(&self) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
store.write_runtime_snapshot(&self.persisted_state())?;
}
Ok(())
}
#[cfg(feature = "fs-store")]
fn persist_worker(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
let worker =
self.workers
.get(worker_id)
.ok_or_else(|| RuntimeError::WorkerNotFound {
worker_id: *worker_id,
})?;
store.write_worker_snapshot(&worker.persisted_record())?;
}
Ok(())
}
#[cfg(feature = "fs-store")]
fn delete_worker_snapshot(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
store.delete_worker_snapshot(worker_id)?;
}
Ok(())
}
#[cfg(feature = "fs-store")]
fn persist_event_by_id(&self, event_id: u64) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
let event = self
.events
.iter()
.find(|event| event.id == event_id)
.ok_or_else(|| RuntimeError::StoreCorrupt {
operation: "persist event",
path: store.runtime_dir().to_path_buf(),
message: format!("event {event_id} is missing from runtime state"),
})?;
store.append_event(event)?;
}
Ok(())
}
#[cfg(all(feature = "fs-store", feature = "ws-server"))]
fn persist_worker_observation_event(
&self,
event: &WorkerObservationEvent,
) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
store.append_worker_observation_event(event)?;
}
Ok(())
}
#[cfg(all(not(feature = "fs-store"), feature = "ws-server"))]
fn persist_worker_observation_event(
&self,
_event: &WorkerObservationEvent,
) -> Result<(), RuntimeError> {
Ok(())
}
#[cfg(feature = "fs-store")]
fn persist_workers(&self) -> Result<(), RuntimeError> {
if self.fs_store().is_some() {
for worker_id in self.workers.keys() {
self.persist_worker(worker_id)?;
}
}
Ok(())
}
#[cfg(not(feature = "fs-store"))]
fn persist_runtime_snapshot(&self) -> Result<(), RuntimeError> {
Ok(())
}
#[cfg(not(feature = "fs-store"))]
fn persist_worker(&self, _worker_id: &WorkerId) -> Result<(), RuntimeError> {
Ok(())
}
#[cfg(not(feature = "fs-store"))]
fn delete_worker_snapshot(&self, _worker_id: &WorkerId) -> Result<(), RuntimeError> {
Ok(())
}
#[cfg(not(feature = "fs-store"))]
fn persist_event_by_id(&self, _event_id: u64) -> Result<(), RuntimeError> {
Ok(())
}
#[cfg(not(feature = "fs-store"))]
fn persist_workers(&self) -> Result<(), RuntimeError> {
Ok(())
}
fn ensure_running(&self) -> Result<(), RuntimeError> {
if self.status == RuntimeStatus::Stopped {
Err(RuntimeError::RuntimeStopped)
} else {
Ok(())
}
}
fn check_config_bundle_ref(
&self,
reference: &ConfigBundleRef,
) -> Result<ConfigBundleAvailability, RuntimeError> {
validate_config_bundle_ref(reference)?;
let bundle = self.config_bundles.get(&reference.id).ok_or_else(|| {
RuntimeError::ConfigBundleMissing {
bundle_id: reference.id.clone(),
}
})?;
if bundle.metadata.digest != reference.digest {
return Err(RuntimeError::ConfigBundleDigestMismatch {
bundle_id: reference.id.clone(),
expected_digest: reference.digest.clone(),
actual_digest: bundle.metadata.digest.clone(),
});
}
Ok(ConfigBundleAvailability {
reference: reference.clone(),
summary: bundle.summary(),
})
}
fn validate_worker_config_boundary(
&self,
_request: &CreateWorkerRequest,
) -> Result<(), RuntimeError> {
Ok(())
}
fn ensure_worker_ref(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> {
if !self.workers.contains_key(&worker_ref.worker_id) {
return Err(RuntimeError::WorkerNotFound {
worker_id: worker_ref.worker_id,
});
}
Ok(())
}
fn worker(&self, worker_ref: &WorkerRef) -> Result<&WorkerRecord, RuntimeError> {
self.ensure_worker_ref(worker_ref)?;
self.workers
.get(&worker_ref.worker_id)
.ok_or_else(|| RuntimeError::WorkerNotFound {
worker_id: worker_ref.worker_id,
})
}
fn worker_mut(&mut self, worker_ref: &WorkerRef) -> Result<&mut WorkerRecord, RuntimeError> {
self.ensure_worker_ref(worker_ref)?;
self.workers
.get_mut(&worker_ref.worker_id)
.ok_or_else(|| RuntimeError::WorkerNotFound {
worker_id: worker_ref.worker_id,
})
}
fn primary_worker_id_for_workdir(&self, working_directory_id: &str) -> Option<WorkerId> {
self.workers.values().find_map(|worker| {
if worker
.execution
.working_directory
.as_ref()
.is_some_and(|binding| binding.summary.working_directory_id == working_directory_id)
|| requested_primary_workdir_id(&worker.request) == Some(working_directory_id)
{
Some(worker.worker_id)
} else {
None
}
})
}
#[cfg(feature = "fs-store")]
fn active_primary_worker_id_for_workdir_excluding(
&self,
working_directory_id: &str,
excluded_worker_id: &WorkerId,
) -> Option<WorkerId> {
self.workers.values().find_map(|worker| {
if worker.worker_id == *excluded_worker_id || !worker.status.is_active() {
return None;
}
if worker
.execution
.working_directory
.as_ref()
.is_some_and(|binding| binding.summary.working_directory_id == working_directory_id)
|| requested_primary_workdir_id(&worker.request) == Some(working_directory_id)
{
Some(worker.worker_id)
} else {
None
}
})
}
#[cfg(feature = "fs-store")]
fn record_restore_failure(
&mut self,
worker_ref: &WorkerRef,
result: WorkerExecutionResult,
) -> Result<(), RuntimeError> {
let message = result
.message
.clone()
.unwrap_or_else(|| "worker execution restore failed".to_string());
let diagnostic_id = self.next_diagnostic_id;
self.next_diagnostic_id += 1;
self.diagnostics.push(RuntimeDiagnostic {
id: diagnostic_id,
severity: DiagnosticSeverity::Warning,
code: "worker_execution_restore_failed".to_string(),
message: format!(
"worker {} execution restore failed: {message}",
worker_ref.worker_id
),
worker_ref: Some(worker_ref.clone()),
});
let worker = self.worker_mut(worker_ref)?;
worker.execution_handle = None;
let mut execution = WorkerExecutionStatus::stale(worker.execution.clone());
execution.last_result = Some(result);
worker.execution = execution;
self.persist_runtime_snapshot()?;
self.persist_worker(&worker_ref.worker_id)?;
Ok(())
}
fn push_event(
&mut self,
worker_ref: Option<WorkerRef>,
kind: RuntimeEventKind,
message: impl Into<String>,
) -> u64 {
let id = self.next_event_id;
self.next_event_id += 1;
self.events.push(RuntimeEvent {
id,
worker_ref,
kind,
message: message.into(),
});
id
}
fn last_event_id(&self) -> u64 {
self.next_event_id.saturating_sub(1)
}
#[cfg(feature = "ws-server")]
fn validate_worker_observation_cursor(
&self,
worker_ref: &WorkerRef,
cursor: WorkerObservationCursor,
) -> Result<(), RuntimeError> {
if let Some(first) = self
.observation_events
.iter()
.find(|event| &event.worker_ref == worker_ref)
{
if cursor.sequence != 0 && cursor.sequence < first.sequence {
return Err(RuntimeError::InvalidRequest(format!(
"worker observation cursor {} is expired for worker {}",
cursor.encode(),
worker_ref.worker_id
)));
}
}
if cursor.sequence >= self.next_observation_sequence {
return Err(RuntimeError::InvalidRequest(format!(
"worker observation cursor {} is unknown for worker {}",
cursor.encode(),
worker_ref.worker_id
)));
}
Ok(())
}
#[cfg(feature = "ws-server")]
fn push_worker_observation_event(
&mut self,
worker_ref: WorkerRef,
payload: protocol::Event,
) -> WorkerObservationEvent {
const MAX_OBSERVATION_BACKLOG: usize = 1024;
let sequence = self.next_observation_sequence;
self.next_observation_sequence += 1;
let event = WorkerObservationEvent::new(sequence, worker_ref, payload);
self.observation_events.push_back(event.clone());
while self.observation_events.len() > MAX_OBSERVATION_BACKLOG {
self.observation_events.pop_front();
}
let _ = self.observation_tx.send(event.clone());
event
}
#[cfg(feature = "ws-server")]
fn project_protocol_event_to_execution(
&mut self,
worker_ref: &WorkerRef,
event: &protocol::Event,
) -> bool {
let Some(worker) = self.workers.get_mut(&worker_ref.worker_id) else {
return false;
};
let status = &mut worker.execution;
let next_run_state = match event {
protocol::Event::Status {
status: protocol::WorkerStatus::Running,
} => Some(WorkerExecutionRunState::Busy),
protocol::Event::Status {
status: protocol::WorkerStatus::Idle,
} => Some(WorkerExecutionRunState::Idle),
protocol::Event::Status {
status: protocol::WorkerStatus::Paused,
} => Some(WorkerExecutionRunState::Busy),
protocol::Event::Snapshot { status, .. } => match status {
protocol::WorkerStatus::Running => Some(WorkerExecutionRunState::Busy),
protocol::WorkerStatus::Idle => Some(WorkerExecutionRunState::Idle),
protocol::WorkerStatus::Paused => Some(WorkerExecutionRunState::Busy),
},
protocol::Event::RunEnd { result } => match result {
protocol::RunResult::Finished | protocol::RunResult::RolledBack => {
Some(WorkerExecutionRunState::Idle)
}
protocol::RunResult::Paused => Some(WorkerExecutionRunState::Busy),
protocol::RunResult::LimitReached => Some(WorkerExecutionRunState::Errored),
},
protocol::Event::Error { .. } => Some(WorkerExecutionRunState::Errored),
_ => None,
};
let Some(next_run_state) = next_run_state else {
return false;
};
if status.run_state == next_run_state {
return false;
}
status.run_state = next_run_state;
true
}
}
#[derive(Debug)]
struct WorkerRecord {
worker_ref: WorkerRef,
worker_id: WorkerId,
status: WorkerStatus,
request: CreateWorkerRequest,
execution: WorkerExecutionStatus,
execution_handle: Option<WorkerExecutionHandle>,
last_event_id: u64,
}
impl WorkerRecord {
fn summary(&self) -> WorkerSummary {
WorkerSummary {
worker_ref: self.worker_ref.clone(),
worker_id: self.worker_id,
status: self.status,
execution: self.execution.clone(),
profile: self.request.profile.clone(),
profile_source: self.request.profile_source.reference(),
config_bundle: self.request.config_bundle.clone(),
last_event_id: self.last_event_id,
}
}
fn detail(&self) -> WorkerDetail {
WorkerDetail {
worker_ref: self.worker_ref.clone(),
worker_id: self.worker_id,
status: self.status,
execution: self.execution.clone(),
profile: self.request.profile.clone(),
profile_source: self.request.profile_source.reference(),
config_bundle: self.request.config_bundle.clone(),
last_event_id: self.last_event_id,
}
}
#[cfg(feature = "fs-store")]
fn persisted_record(&self) -> PersistedWorkerRecord {
PersistedWorkerRecord {
worker_ref: self.worker_ref.clone(),
worker_id: self.worker_id.clone(),
status: self.status,
request: self.request.clone(),
execution: self.execution.clone(),
last_event_id: self.last_event_id,
}
}
}
fn requested_primary_workdir_id(request: &CreateWorkerRequest) -> Option<&str> {
request
.working_directory
.as_ref()
.map(|claim| claim.working_directory_id.as_str())
.or_else(|| {
request
.working_directory_request
.as_ref()
.and_then(|request| request.backend_workdir_id.as_deref())
})
}
fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), RuntimeError> {
match &request.profile_source {
crate::catalog::ProfileSourceArchiveSource::Embedded { archive } => {
archive.verify().map_err(|err| {
RuntimeError::InvalidRequest(format!("profile_source archive is invalid: {err}"))
})?;
}
crate::catalog::ProfileSourceArchiveSource::Http { location } => {
if location.url.trim().is_empty() {
return Err(RuntimeError::InvalidRequest(
"profile_source.location.url must not be empty".to_string(),
));
}
if location.archive.digest.trim().is_empty() {
return Err(RuntimeError::InvalidRequest(
"profile_source.location.archive.digest must not be empty".to_string(),
));
}
}
}
if let Some(input) = &request.initial_input {
if input.kind != WorkerInputKind::User {
return Err(RuntimeError::InvalidInitialInputKind {
kind: format!("{:?}", input.kind),
});
}
if input.content.trim().is_empty() {
return Err(RuntimeError::InvalidRequest(
"initial_input.content must not be empty".to_string(),
));
}
}
Ok(())
}
fn validate_worker_input(input: &WorkerInput) -> Result<(), RuntimeError> {
if !input.kind.is_empty_content_allowed() && input.content.trim().is_empty() {
return Err(RuntimeError::InvalidRequest(
"worker input content must not be empty".to_string(),
));
}
if input
.segments
.as_ref()
.is_some_and(|segments| segments.is_empty())
{
return Err(RuntimeError::InvalidRequest(
"worker input segments must not be empty".to_string(),
));
}
Ok(())
}
#[cfg(feature = "ws-server")]
fn protocol_method_observation_event(method: &Method) -> Option<Event> {
match method {
Method::Run { input } => Some(Event::UserMessage {
segments: input.clone(),
}),
Method::Notify { message, .. } => Some(Event::SystemItem {
item: serde_json::json!({
"kind": "embedded_worker_system_input",
"content": message,
}),
}),
Method::RegisterPeer { name } => Some(Event::SystemItem {
item: serde_json::json!({
"kind": "embedded_worker_command_input",
"command": "register_peer",
"content": name,
}),
}),
Method::Compact => Some(Event::SystemItem {
item: serde_json::json!({
"kind": "embedded_worker_command_input",
"command": "compact",
"content": "",
}),
}),
Method::ListRewindTargets => Some(Event::SystemItem {
item: serde_json::json!({
"kind": "embedded_worker_command_input",
"command": "list_rewind_targets",
"content": "",
}),
}),
_ => None,
}
}
#[cfg(not(feature = "ws-server"))]
fn protocol_method_observation_event(_method: &Method) -> Option<Event> {
None
}
#[cfg(feature = "ws-server")]
fn input_protocol_event(input: &WorkerInput) -> protocol::Event {
match input.kind {
WorkerInputKind::User => protocol::Event::UserMessage {
segments: input.segments.clone().unwrap_or_else(|| {
vec![protocol::Segment::Text {
content: input.content.clone(),
}]
}),
},
WorkerInputKind::System => protocol::Event::SystemItem {
item: serde_json::json!({
"kind": "embedded_worker_system_input",
"content": input.content.clone(),
}),
},
WorkerInputKind::Compact
| WorkerInputKind::ListRewindTargets
| WorkerInputKind::RegisterPeer => protocol::Event::SystemItem {
item: serde_json::json!({
"kind": "embedded_worker_command_input",
"command": input.kind,
"content": input.content.clone(),
}),
},
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::catalog::{ConfigBundleRef, ProfileSelector};
use crate::config_bundle::{
ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigDeclaration,
ConfigDeclarationKind, ConfigProfileDescriptor,
};
use crate::execution::{
WorkerExecutionBackend, WorkerExecutionContext, WorkerExecutionHandle,
WorkerExecutionRestoreRequest, WorkerExecutionRunState,
};
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
fn task_request(_objective: &str) -> CreateWorkerRequest {
let profile = ProfileSelector::Builtin("builtin:coder".to_string());
let bundle = test_bundle_for_profile(profile.clone());
CreateWorkerRequest {
profile,
profile_source: crate::catalog::ProfileSourceArchiveSource::Http {
location: crate::catalog::ProfileSourceArchiveHttpRef {
url: "http://127.0.0.1/profile-source.tar".to_string(),
etag: None,
archive: crate::profile_archive::ProfileSourceArchiveRef {
id: "test-profile-source".to_string(),
digest: "test-digest".to_string(),
size_bytes: 0,
source_graph: crate::profile_archive::ProfileSourceGraphSummary {
source_count: 0,
total_source_bytes: 0,
entrypoints: BTreeMap::new(),
import_count: 0,
},
},
},
},
config_bundle: Some(ConfigBundleRef {
id: bundle.metadata.id,
digest: bundle.metadata.digest,
}),
initial_input: None,
working_directory_request: None,
working_directory: None,
workspace_api: None,
}
}
fn test_bundle_for_profile(profile: ProfileSelector) -> ConfigBundle {
ConfigBundle {
metadata: ConfigBundleMetadata {
id: "bundle-1".to_string(),
digest: String::new(),
revision: "rev-1".to_string(),
workspace_id: "workspace-1".to_string(),
created_at: "2026-06-26T00:00:00Z".to_string(),
provenance: ConfigBundleProvenance {
source: "workspace-backend".to_string(),
detail: Some("profile-sync".to_string()),
},
},
profiles: vec![ConfigProfileDescriptor {
selector: profile,
label: Some("Coder".to_string()),
}],
declarations: vec![ConfigDeclaration {
kind: ConfigDeclarationKind::CapabilityGrant,
name: "read".to_string(),
reference: "capability:read".to_string(),
}],
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}
#[derive(Default)]
struct TestExecutionBackend {
dispatch_result: Mutex<Option<WorkerExecutionResult>>,
restore_result: Mutex<Option<WorkerExecutionSpawnResult>>,
restore_count: Mutex<u64>,
contexts: Mutex<BTreeMap<WorkerId, WorkerExecutionContext>>,
#[cfg(feature = "ws-server")]
snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
}
impl TestExecutionBackend {
fn set_dispatch_result(&self, result: WorkerExecutionResult) {
*self.dispatch_result.lock().unwrap() = Some(result);
}
#[cfg(feature = "ws-server")]
fn set_worker_snapshot(&self, worker_ref: &WorkerRef, snapshot: protocol::Event) {
self.snapshots
.lock()
.unwrap()
.insert(worker_ref.worker_id.clone(), snapshot);
}
#[cfg(feature = "ws-server")]
fn publish_text_delta(
&self,
worker_ref: &WorkerRef,
text: &str,
) -> Result<crate::observation::WorkerObservationEvent, RuntimeError> {
let contexts = self.contexts.lock().unwrap();
let context = contexts.get(&worker_ref.worker_id).expect("context stored");
context.publish_protocol_event(protocol::Event::TextDelta { text: text.into() })
}
}
impl WorkerExecutionBackend for TestExecutionBackend {
fn backend_id(&self) -> &str {
"test-execution-backend"
}
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
self.contexts
.lock()
.unwrap()
.insert(request.worker_ref.worker_id.clone(), request.context);
WorkerExecutionSpawnResult::Connected {
handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()),
run_state: WorkerExecutionRunState::Idle,
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
}
}
fn restore_worker(
&self,
request: WorkerExecutionRestoreRequest,
) -> WorkerExecutionSpawnResult {
*self.restore_count.lock().unwrap() += 1;
if let Some(result) = self.restore_result.lock().unwrap().clone() {
return result;
}
self.contexts
.lock()
.unwrap()
.insert(request.worker_ref.worker_id.clone(), request.context);
WorkerExecutionSpawnResult::Connected {
handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()),
run_state: WorkerExecutionRunState::Idle,
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
}
}
fn dispatch_input(
&self,
_handle: &WorkerExecutionHandle,
_input: WorkerInput,
) -> WorkerExecutionResult {
self.dispatch_result
.lock()
.unwrap()
.clone()
.unwrap_or_else(|| {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::Input,
WorkerExecutionRunState::Idle,
)
})
}
fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::Stop,
WorkerExecutionRunState::Stopped,
)
}
fn cancel_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::Cancel,
WorkerExecutionRunState::Stopped,
)
}
#[cfg(feature = "ws-server")]
fn worker_snapshot(&self, handle: &WorkerExecutionHandle) -> Option<protocol::Event> {
self.snapshots
.lock()
.unwrap()
.get(&handle.worker_ref().worker_id)
.cloned()
}
}
fn runtime_with_backend() -> Runtime {
let runtime = Runtime::with_execution_backend(
RuntimeOptions::default(),
Arc::new(TestExecutionBackend::default()),
)
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
runtime
}
fn runtime_and_backend() -> (Runtime, Arc<TestExecutionBackend>) {
let backend = Arc::new(TestExecutionBackend::default());
let runtime =
Runtime::with_execution_backend(RuntimeOptions::default(), backend.clone()).unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
(runtime, backend)
}
fn test_bundle() -> ConfigBundle {
test_bundle_for_profile(ProfileSelector::Builtin("builtin:coder".to_string()))
}
fn bundled_task_request(objective: &str, bundle: &ConfigBundle) -> CreateWorkerRequest {
let mut request = task_request(objective);
request.config_bundle = Some(ConfigBundleRef {
id: bundle.metadata.id.clone(),
digest: bundle.metadata.digest.clone(),
});
request
}
#[test]
fn create_list_and_detail_preserve_runtime_local_worker_authority() {
let runtime = runtime_with_backend();
let detail = runtime.create_worker(task_request("implement v0")).unwrap();
assert_eq!(detail.status, WorkerStatus::Running);
assert_eq!(detail.config_bundle.as_ref().unwrap().id, "bundle-1");
let list = runtime.list_workers().unwrap();
assert_eq!(list.len(), 1);
assert_eq!(list[0].worker_ref, detail.worker_ref);
assert_eq!(list[0].config_bundle, detail.config_bundle);
let fetched = runtime.worker_detail(&detail.worker_ref).unwrap();
assert_eq!(fetched.worker_id, detail.worker_id);
assert_eq!(fetched.profile, detail.profile);
}
#[test]
fn synced_config_bundle_is_stored_checked_and_used_for_worker_creation() {
let runtime = runtime_with_backend();
let bundle = test_bundle();
let availability = runtime.store_config_bundle(bundle.clone()).unwrap();
assert_eq!(availability.reference.id, "bundle-1");
assert_eq!(availability.reference.digest, bundle.metadata.digest);
let listed = runtime.list_config_bundles().unwrap();
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].id, "bundle-1");
let checked = runtime
.check_config_bundle(&availability.reference)
.unwrap();
assert_eq!(checked.summary.digest, availability.summary.digest);
let detail = runtime
.create_worker(bundled_task_request("synced", &bundle))
.unwrap();
assert_eq!(detail.config_bundle, Some(availability.reference));
}
#[test]
fn config_bundle_errors_are_typed() {
let runtime = Runtime::new_memory();
let bundle = test_bundle();
runtime.store_config_bundle(bundle.clone()).unwrap();
let mismatch = runtime
.check_config_bundle(&ConfigBundleRef {
id: bundle.metadata.id.clone(),
digest: "0".repeat(64),
})
.unwrap_err();
assert!(matches!(
mismatch,
RuntimeError::ConfigBundleDigestMismatch { .. }
));
let mut unsupported = test_bundle();
unsupported.declarations.push(ConfigDeclaration {
kind: ConfigDeclarationKind::Unsupported,
name: "plugin-registry".to_string(),
reference: "plugin-registry:v0".to_string(),
});
unsupported = unsupported.with_computed_digest();
let unsupported_err = runtime.store_config_bundle(unsupported).unwrap_err();
assert!(matches!(
unsupported_err,
RuntimeError::UnsupportedConfigDeclaration { .. }
));
}
#[test]
fn create_worker_rejects_system_initial_input_without_persisting_worker() {
let runtime = runtime_with_backend();
let mut request = task_request("system initial input");
request.initial_input = Some(WorkerInput::system("role/system belongs in config bundle"));
let error = runtime.create_worker(request).unwrap_err();
assert!(matches!(
error,
RuntimeError::InvalidInitialInputKind { .. }
));
assert!(runtime.list_workers().unwrap().is_empty());
let events = runtime
.read_events(&runtime.event_cursor_from_start().unwrap(), 16)
.unwrap();
assert!(
events
.events
.iter()
.all(|event| event.kind != RuntimeEventKind::WorkerCreated)
);
}
#[test]
fn create_worker_without_execution_backend_is_rejected_and_not_persisted() {
let runtime = Runtime::new_memory();
runtime.store_config_bundle(test_bundle()).unwrap();
let error = runtime
.create_worker(task_request("no backend"))
.unwrap_err();
assert!(matches!(
error,
RuntimeError::ExecutionBackendUnavailable { .. }
));
assert!(runtime.list_workers().unwrap().is_empty());
}
#[test]
fn create_worker_without_execution_backend_is_rejected_before_persisting_worker() {
let runtime = Runtime::new_memory();
let error = runtime
.create_worker(task_request("missing backend"))
.unwrap_err();
assert!(matches!(
error,
RuntimeError::ExecutionBackendUnavailable { .. }
));
assert!(runtime.list_workers().unwrap().is_empty());
}
#[test]
fn connected_backend_busy_dispatch_is_typed_and_not_transcribed() {
let (runtime, backend) = runtime_and_backend();
backend.set_dispatch_result(WorkerExecutionResult::busy(
WorkerExecutionOperation::Input,
"worker is already running",
));
let detail = runtime.create_worker(task_request("busy")).unwrap();
let err = runtime
.send_input(&detail.worker_ref, WorkerInput::user("wait"))
.unwrap_err();
assert!(matches!(
err,
RuntimeError::WorkerExecutionRejected {
outcome: crate::execution::WorkerExecutionOutcome::Busy,
..
}
));
let refreshed = runtime.worker_detail(&detail.worker_ref).unwrap();
assert_eq!(refreshed.execution.run_state, WorkerExecutionRunState::Busy);
#[cfg(feature = "ws-server")]
assert!(
runtime
.read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero())
.unwrap()
.is_empty()
);
}
#[cfg(feature = "ws-server")]
#[test]
fn backend_protocol_publish_hook_writes_observation_bus() {
let (runtime, backend) = runtime_and_backend();
let detail = runtime.create_worker(task_request("observe")).unwrap();
let observation = backend
.publish_text_delta(&detail.worker_ref, "from backend")
.unwrap();
assert_eq!(observation.worker_ref, detail.worker_ref);
let observations = runtime
.read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero())
.unwrap();
assert_eq!(observations.len(), 1);
assert!(matches!(
observations[0].payload,
protocol::Event::TextDelta { .. }
));
}
#[cfg(feature = "ws-server")]
#[test]
fn observation_snapshot_prefers_live_backend_snapshot() {
let (runtime, backend) = runtime_and_backend();
let detail = runtime
.create_worker(task_request("observe snapshot"))
.unwrap();
let expected_entry = serde_json::json!({"kind": "restored-log-entry"});
backend.set_worker_snapshot(
&detail.worker_ref,
protocol::Event::Snapshot {
entries: vec![expected_entry.clone()],
greeting: protocol::Greeting {
worker_name: "live-worker".to_string(),
cwd: "/tmp/live".to_string(),
provider: "test-provider".to_string(),
model: "test-model".to_string(),
scope_summary: "live snapshot".to_string(),
tools: Vec::new(),
context_window: 128,
context_tokens: 64,
},
status: protocol::WorkerStatus::Running,
in_flight: protocol::InFlightSnapshot { blocks: Vec::new() },
},
);
let snapshot = runtime
.worker_observation_snapshot(&detail.worker_ref)
.unwrap();
match snapshot {
protocol::Event::Snapshot {
entries,
greeting,
status,
..
} => {
assert_eq!(entries, vec![expected_entry]);
assert_eq!(greeting.worker_name, "live-worker");
assert_eq!(status, protocol::WorkerStatus::Running);
}
other => panic!("expected snapshot, got {other:?}"),
}
}
struct InputOnlyBackend;
impl WorkerExecutionBackend for InputOnlyBackend {
fn backend_id(&self) -> &str {
"input-only"
}
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
WorkerExecutionSpawnResult::Connected {
handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()),
run_state: WorkerExecutionRunState::Idle,
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
}
}
fn dispatch_input(
&self,
_handle: &WorkerExecutionHandle,
_input: WorkerInput,
) -> WorkerExecutionResult {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::Input,
WorkerExecutionRunState::Idle,
)
}
}
#[test]
fn connected_backend_stop_unsupported_is_typed_rejection() {
let runtime =
Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(InputOnlyBackend))
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let detail = runtime.create_worker(task_request("no stop")).unwrap();
let err = runtime
.stop_worker(&detail.worker_ref, Some("stop".to_string()))
.unwrap_err();
assert!(matches!(
err,
RuntimeError::WorkerExecutionRejected {
outcome: crate::execution::WorkerExecutionOutcome::Unsupported,
..
}
));
assert_eq!(
runtime.worker_detail(&detail.worker_ref).unwrap().status,
WorkerStatus::Running
);
}
#[cfg(feature = "ws-server")]
#[test]
fn send_input_records_protocol_observations() {
let runtime = Runtime::with_execution_backend(
RuntimeOptions {
limits: RuntimeLimits {
max_event_batch_items: 16,
},
..RuntimeOptions::default()
},
Arc::new(TestExecutionBackend::default()),
)
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let detail = runtime.create_worker(task_request("chat")).unwrap();
let first = runtime
.send_input(&detail.worker_ref, WorkerInput::user("hello"))
.unwrap();
runtime
.send_input(&detail.worker_ref, WorkerInput::system("note"))
.unwrap();
let observations = runtime
.read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero())
.unwrap();
assert_eq!(first.event_id, 3);
assert_eq!(observations.len(), 2);
assert!(matches!(
observations[0].payload,
protocol::Event::UserMessage { .. }
));
assert!(matches!(
observations[1].payload,
protocol::Event::SystemItem { .. }
));
}
#[test]
fn stop_and_cancel_workers_update_summary() {
let runtime = runtime_with_backend();
let stopped = runtime.create_worker(task_request("stop me")).unwrap();
let cancelled = runtime.create_worker(task_request("cancel me")).unwrap();
let stop_ack = runtime
.stop_worker(&stopped.worker_ref, Some("done".to_string()))
.unwrap();
assert_eq!(stop_ack.status, WorkerStatus::Stopped);
let cancel_ack = runtime
.cancel_worker(&cancelled.worker_ref, Some("abort".to_string()))
.unwrap();
assert_eq!(cancel_ack.status, WorkerStatus::Cancelled);
let summary = runtime.summary().unwrap();
assert_eq!(summary.worker_count, 2);
assert_eq!(summary.active_worker_count, 0);
assert_eq!(summary.stopped_worker_count, 1);
assert_eq!(summary.cancelled_worker_count, 1);
}
#[test]
fn delete_worker_removes_stopped_worker_from_runtime() {
let runtime = runtime_with_backend();
let worker = runtime.create_worker(task_request("delete me")).unwrap();
assert!(runtime.delete_worker(&worker.worker_ref).is_err());
runtime
.stop_worker(&worker.worker_ref, Some("done".to_string()))
.unwrap();
let result = runtime.delete_worker(&worker.worker_ref).unwrap();
assert!(result.deleted);
assert_eq!(result.worker_id, worker.worker_id);
assert!(matches!(
runtime.worker_detail(&worker.worker_ref),
Err(RuntimeError::WorkerNotFound { .. })
));
let summary = runtime.summary().unwrap();
assert_eq!(summary.worker_count, 0);
}
#[test]
fn stop_then_cancel_preserves_stopped_terminal_state() {
let runtime = runtime_with_backend();
let cursor = runtime.event_cursor_from_start().unwrap();
let worker = runtime
.create_worker(task_request("stable stopped"))
.unwrap();
let stop_ack = runtime
.stop_worker(&worker.worker_ref, Some("done".to_string()))
.unwrap();
let cancel_ack = runtime
.cancel_worker(&worker.worker_ref, Some("late cancel".to_string()))
.unwrap();
assert_eq!(stop_ack.status, WorkerStatus::Stopped);
assert_eq!(cancel_ack.status, WorkerStatus::Stopped);
assert_eq!(cancel_ack.event_id, stop_ack.event_id);
assert_eq!(
runtime.worker_detail(&worker.worker_ref).unwrap().status,
WorkerStatus::Stopped
);
let summary = runtime.summary().unwrap();
assert_eq!(summary.active_worker_count, 0);
assert_eq!(summary.stopped_worker_count, 1);
assert_eq!(summary.cancelled_worker_count, 0);
let events = runtime.read_events(&cursor, 10).unwrap().events;
assert_eq!(
events
.iter()
.filter(|event| event.kind == RuntimeEventKind::WorkerStopped)
.count(),
1
);
assert_eq!(
events
.iter()
.filter(|event| event.kind == RuntimeEventKind::WorkerCancelled)
.count(),
0
);
}
#[test]
fn cancel_then_stop_preserves_cancelled_terminal_state() {
let runtime = runtime_with_backend();
let cursor = runtime.event_cursor_from_start().unwrap();
let worker = runtime
.create_worker(task_request("stable cancelled"))
.unwrap();
let cancel_ack = runtime
.cancel_worker(&worker.worker_ref, Some("abort".to_string()))
.unwrap();
let stop_ack = runtime
.stop_worker(&worker.worker_ref, Some("late stop".to_string()))
.unwrap();
assert_eq!(cancel_ack.status, WorkerStatus::Cancelled);
assert_eq!(stop_ack.status, WorkerStatus::Cancelled);
assert_eq!(stop_ack.event_id, cancel_ack.event_id);
assert_eq!(
runtime.worker_detail(&worker.worker_ref).unwrap().status,
WorkerStatus::Cancelled
);
let summary = runtime.summary().unwrap();
assert_eq!(summary.active_worker_count, 0);
assert_eq!(summary.stopped_worker_count, 0);
assert_eq!(summary.cancelled_worker_count, 1);
let events = runtime.read_events(&cursor, 10).unwrap().events;
assert_eq!(
events
.iter()
.filter(|event| event.kind == RuntimeEventKind::WorkerCancelled)
.count(),
1
);
assert_eq!(
events
.iter()
.filter(|event| event.kind == RuntimeEventKind::WorkerStopped)
.count(),
0
);
}
#[test]
fn event_cursor_and_poll_only_subscription_are_bounded_placeholders() {
let runtime = runtime_with_backend();
let cursor = runtime.event_cursor_from_start().unwrap();
let subscription = runtime.subscribe_events(cursor.clone()).unwrap();
assert_eq!(subscription.mode, EventSubscriptionMode::PollOnly);
let worker = runtime.create_worker(task_request("events")).unwrap();
runtime
.send_input(&worker.worker_ref, WorkerInput::user("eventful"))
.unwrap();
let batch = runtime.read_events(&cursor, 2).unwrap();
assert_eq!(batch.events.len(), 2);
assert!(batch.has_more);
assert_eq!(batch.events[0].kind, RuntimeEventKind::RuntimeStarted);
assert_eq!(batch.events[1].kind, RuntimeEventKind::WorkerCreated);
let next = runtime.read_events(&batch.cursor, 2).unwrap();
assert_eq!(next.events.len(), 1);
assert_eq!(next.events[0].kind, RuntimeEventKind::WorkerInputAccepted);
assert!(!next.has_more);
}
#[cfg(feature = "fs-store")]
static NEXT_FS_TEST_ROOT: AtomicU64 = AtomicU64::new(1);
#[cfg(feature = "fs-store")]
fn fs_store_root(label: &str) -> std::path::PathBuf {
let sequence = NEXT_FS_TEST_ROOT.fetch_add(1, Ordering::Relaxed);
let root = std::env::temp_dir().join(format!(
"worker-runtime-fs-store-{label}-{}-{sequence}",
std::process::id()
));
let _ = std::fs::remove_dir_all(&root);
root
}
#[cfg(feature = "fs-store")]
fn runtime_store(runtime: &Runtime) -> FsRuntimeStore {
let state = runtime.lock().unwrap();
match &state.persistence {
RuntimePersistence::Fs(store) => store.clone(),
RuntimePersistence::Memory => panic!("expected fs-backed runtime"),
}
}
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_restores_workers_events_and_protocol_observations() {
let root = fs_store_root("restore");
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
display_name: Some("filesystem runtime".to_string()),
limits: RuntimeLimits {
max_event_batch_items: 2,
},
},
Arc::new(TestExecutionBackend::default()),
)
.unwrap();
assert_eq!(
runtime.summary().unwrap().backend,
RuntimeBackendKind::FsStore
);
runtime.store_config_bundle(test_bundle()).unwrap();
let worker = runtime.create_worker(task_request("persist me")).unwrap();
runtime
.send_input(&worker.worker_ref, WorkerInput::user("first"))
.unwrap();
runtime
.send_input(&worker.worker_ref, WorkerInput::system("second"))
.unwrap();
runtime
.stop_worker(&worker.worker_ref, Some("finished".to_string()))
.unwrap();
let store = runtime_store(&runtime);
drop(runtime);
let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
})
.unwrap();
let restored_worker = restored.worker_detail(&worker.worker_ref).unwrap();
assert_eq!(restored_worker.status, WorkerStatus::Stopped);
assert_eq!(
restored_worker.execution.backend,
WorkerExecutionBackendKind::Stale
);
assert_eq!(
restored_worker
.execution
.binding
.as_ref()
.map(|binding| binding.backend_id.as_str()),
Some("test-execution-backend")
);
assert!(
restored
.diagnostics()
.unwrap()
.iter()
.any(
|diagnostic| diagnostic.code == "worker_execution_mapping_stale"
&& diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref)
)
);
#[cfg(feature = "ws-server")]
{
let observations = restored
.read_worker_observation_events(&worker.worker_ref, WorkerObservationCursor::zero())
.unwrap();
assert_eq!(observations.len(), 2);
assert!(matches!(
observations[0].payload,
protocol::Event::UserMessage { .. }
));
assert!(matches!(
observations[1].payload,
protocol::Event::SystemItem { .. }
));
}
let cursor = restored.event_cursor_from_start().unwrap();
let batch = restored.read_events(&cursor, 2).unwrap();
assert_eq!(batch.events.len(), 2);
assert!(batch.has_more);
assert_eq!(batch.events[0].kind, RuntimeEventKind::RuntimeStarted);
assert_eq!(batch.events[1].kind, RuntimeEventKind::WorkerCreated);
let direct_events = store.read_events(&cursor, 2, 2).unwrap();
assert_eq!(direct_events.events, batch.events);
#[cfg(feature = "ws-server")]
{
let observation = restored
.observe_worker_event(
&worker.worker_ref,
protocol::Event::TextDelta {
text: "restored observation bus".to_string(),
},
)
.unwrap();
assert_eq!(observation.sequence, 3);
let observations = restored
.read_worker_observation_events(&worker.worker_ref, WorkerObservationCursor::zero())
.unwrap();
assert_eq!(observations.len(), 3);
assert_eq!(observations[2].cursor, observation.cursor);
}
let _ = std::fs::remove_dir_all(root);
}
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_restores_active_worker_execution_handles() {
let root = fs_store_root("execution-restore");
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
},
Arc::new(TestExecutionBackend::default()),
)
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let worker = runtime
.create_worker(task_request("restore active worker"))
.unwrap();
drop(runtime);
let restoring_backend = Arc::new(TestExecutionBackend::default());
let restored = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
},
restoring_backend.clone(),
)
.unwrap();
assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 1);
let restored_worker = restored.worker_detail(&worker.worker_ref).unwrap();
assert_eq!(restored_worker.status, WorkerStatus::Running);
assert_eq!(
restored_worker.execution.backend,
WorkerExecutionBackendKind::Connected
);
assert!(restored_worker.execution.binding.is_some());
restored
.send_input(&worker.worker_ref, WorkerInput::user("after restart"))
.unwrap();
let cursor = restored.event_cursor_from_start().unwrap();
assert!(
restored
.read_events(&cursor, 16)
.unwrap()
.events
.iter()
.any(
|event| event.kind == RuntimeEventKind::WorkerExecutionRestored
&& event.worker_ref.as_ref() == Some(&worker.worker_ref)
)
);
let _ = std::fs::remove_dir_all(root);
}
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_keeps_worker_stale_when_execution_restore_fails() {
let root = fs_store_root("execution-restore-failed");
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
},
Arc::new(TestExecutionBackend::default()),
)
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let worker = runtime
.create_worker(task_request("restore failure"))
.unwrap();
drop(runtime);
let restoring_backend = Arc::new(TestExecutionBackend::default());
*restoring_backend.restore_result.lock().unwrap() =
Some(WorkerExecutionSpawnResult::Errored(
WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, "restore boom"),
));
let restored = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
},
restoring_backend.clone(),
)
.unwrap();
assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 1);
let restored_worker = restored.worker_detail(&worker.worker_ref).unwrap();
assert_eq!(restored_worker.status, WorkerStatus::Running);
assert_eq!(
restored_worker.execution.backend,
WorkerExecutionBackendKind::Stale
);
assert!(
restored
.diagnostics()
.unwrap()
.iter()
.any(
|diagnostic| diagnostic.code == "worker_execution_restore_failed"
&& diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref)
)
);
let err = restored
.send_input(
&worker.worker_ref,
WorkerInput::user("after failed restore"),
)
.unwrap_err();
assert!(matches!(
err,
RuntimeError::WorkerExecutionUnavailable { .. }
));
let _ = std::fs::remove_dir_all(root);
}
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_reports_corrupt_and_missing_data() {
let corrupt_root = fs_store_root("corrupt");
let corrupt_runtime = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: corrupt_root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
})
.unwrap();
let corrupt_store = runtime_store(&corrupt_runtime);
std::fs::write(
corrupt_store.runtime_dir().join("runtime.json"),
b"not json",
)
.unwrap();
drop(corrupt_runtime);
let err = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: corrupt_root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
})
.unwrap_err();
assert!(matches!(err, RuntimeError::StoreCorrupt { .. }));
let _ = std::fs::remove_dir_all(corrupt_root);
let missing_root = fs_store_root("missing");
let missing_runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: missing_root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
},
Arc::new(TestExecutionBackend::default()),
)
.unwrap();
missing_runtime.store_config_bundle(test_bundle()).unwrap();
missing_runtime
.create_worker(task_request("missing worker snapshot"))
.unwrap();
let missing_store = runtime_store(&missing_runtime);
let mut worker_dirs = std::fs::read_dir(missing_store.runtime_dir().join("workers"))
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
worker_dirs.sort_by_key(|entry| entry.path());
std::fs::remove_file(worker_dirs[0].path().join("worker.json")).unwrap();
drop(missing_runtime);
let loaded = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: missing_root.clone(),
display_name: None,
limits: RuntimeLimits::default(),
})
.expect("invalid worker snapshot should not make runtime store unreadable");
assert!(loaded.list_workers().unwrap().is_empty());
assert!(
loaded
.diagnostics()
.unwrap()
.iter()
.any(|diagnostic| diagnostic.code == "worker_snapshot_ignored")
);
let _ = std::fs::remove_dir_all(missing_root);
}
}
View Git Blame Copy Permalink