From 0f9f06048af00335d7f590582436e29c722a50ee Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 1 Aug 2026 22:55:58 +0900 Subject: [PATCH] runtime: remove legacy event polling authority --- crates/worker-runtime/src/catalog.rs | 3 - crates/worker-runtime/src/fs_store.rs | 165 +------- crates/worker-runtime/src/http_server.rs | 6 +- crates/worker-runtime/src/interaction.rs | 1 - crates/worker-runtime/src/lib.rs | 2 +- crates/worker-runtime/src/main.rs | 13 - crates/worker-runtime/src/management.rs | 27 +- crates/worker-runtime/src/observation.rs | 50 --- crates/worker-runtime/src/runtime.rs | 381 ++---------------- crates/workspace-server/src/hosts.rs | 32 +- crates/workspace-server/src/store.rs | 6 +- .../src/lib/workspace/sidebar/types.ts | 1 - 12 files changed, 72 insertions(+), 615 deletions(-) diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 4e7fb3ca..035e6b28 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -261,7 +261,6 @@ pub struct WorkerSummary { pub profile_source: ProfileSourceArchiveRef, #[serde(default, skip_serializing_if = "Option::is_none")] pub config_bundle: Option, - pub last_event_id: u64, } /// Full Worker catalog/lifecycle detail. @@ -280,7 +279,6 @@ pub struct WorkerDetail { pub profile_source: ProfileSourceArchiveRef, #[serde(default, skip_serializing_if = "Option::is_none")] pub config_bundle: Option, - pub last_event_id: u64, } /// Acknowledgement returned by stop/cancel lifecycle operations. @@ -288,5 +286,4 @@ pub struct WorkerDetail { pub struct WorkerLifecycleAck { pub worker_ref: WorkerRef, pub status: WorkerStatus, - pub event_id: u64, } diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 6e67181c..608436c4 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -3,18 +3,16 @@ use crate::config_bundle::ConfigBundle; use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic}; use crate::error::RuntimeError; use crate::identity::{WorkerId, WorkerRef}; -use crate::management::{RuntimeBackendKind, RuntimeLimits, RuntimeStatus}; -use crate::observation::{EventCursor, RuntimeEvent, RuntimeEventBatch}; +use crate::management::{RuntimeBackendKind, RuntimeStatus}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::fs::{self, File, OpenOptions}; -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufReader, Write}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; const SCHEMA_VERSION: u32 = 1; const RUNTIME_FILE: &str = "runtime.json"; -const EVENTS_FILE: &str = "events.jsonl"; const WORKERS_DIR: &str = "workers"; const LEGACY_RUNTIMES_DIR: &str = "runtimes"; const WORKER_FILE: &str = "worker.json"; @@ -28,7 +26,6 @@ pub struct FsRuntimeStoreOptions { /// Root directory containing this Runtime's store data. pub root: PathBuf, pub display_name: Option, - pub limits: RuntimeLimits, } impl FsRuntimeStoreOptions { @@ -36,7 +33,6 @@ impl FsRuntimeStoreOptions { Self { root: root.into(), display_name: None, - limits: RuntimeLimits::default(), } } } @@ -59,43 +55,6 @@ impl FsRuntimeStore { &self.root } - /// Read persisted Runtime events directly from the event log with the same - /// bounded cursor semantics as [`crate::Runtime::read_events`]. - pub fn read_events( - &self, - cursor: &EventCursor, - limit: usize, - max_limit: usize, - ) -> Result { - if limit > max_limit { - return Err(RuntimeError::LimitTooLarge { - requested: limit, - max: max_limit, - }); - } - - let events = read_json_lines::(&self.events_path(), "read events")?; - let mut selected = Vec::new(); - for event in events - .iter() - .filter(|event| event.id >= cursor.next_event_id) - .take(limit) - { - selected.push(event.clone()); - } - let next_event_id = selected - .last() - .map(|event| event.id + 1) - .unwrap_or(cursor.next_event_id); - let has_more = events.iter().any(|event| event.id >= next_event_id); - - Ok(RuntimeEventBatch { - cursor: EventCursor { next_event_id }, - events: selected, - has_more, - }) - } - pub(crate) fn open_or_create(root: PathBuf) -> Result { let existed = root.exists(); if existed && !root.is_dir() { @@ -115,6 +74,18 @@ impl FsRuntimeStore { path: root.join(WORKERS_DIR), source, })?; + let legacy_events = root.join("events.jsonl"); + match fs::remove_file(&legacy_events) { + Ok(()) => {} + Err(source) if source.kind() == std::io::ErrorKind::NotFound => {} + Err(source) => { + return Err(RuntimeError::StoreIo { + operation: "remove legacy runtime events", + path: legacy_events, + source, + }); + } + } let store = Self { root }; let state = if existed { @@ -165,19 +136,11 @@ impl FsRuntimeStore { }) } - pub(crate) fn append_event(&self, event: &RuntimeEvent) -> Result<(), RuntimeError> { - if let Some(worker_ref) = &event.worker_ref { - self.ensure_worker_ref(worker_ref)?; - } - append_json_line(&self.events_path(), event, "append event") - } - pub(crate) fn load_runtime_state(&self) -> Result { let runtime_path = self.runtime_path(); let mut snapshot: RuntimeSnapshot = read_json(&runtime_path, "read runtime snapshot")?; snapshot.validate(&runtime_path)?; - let events = read_json_lines::(&self.events_path(), "read events")?; let workers_dir = self.root.join(WORKERS_DIR); if !workers_dir.exists() { return Err(RuntimeError::StoreMissing { @@ -250,7 +213,7 @@ impl FsRuntimeStore { } } - Ok(snapshot.into_persisted(events, workers)) + Ok(snapshot.into_persisted(workers)) } fn ensure_worker_ref(&self, _worker_ref: &WorkerRef) -> Result<(), RuntimeError> { @@ -261,10 +224,6 @@ impl FsRuntimeStore { self.root.join(RUNTIME_FILE) } - fn events_path(&self) -> PathBuf { - self.root.join(EVENTS_FILE) - } - fn worker_dir(&self, worker_id: &WorkerId) -> PathBuf { self.root.join(WORKERS_DIR).join(worker_id.to_string()) } @@ -287,14 +246,11 @@ pub(crate) struct OpenedFsRuntimeStore { pub(crate) struct PersistedRuntimeState { pub(crate) display_name: Option, pub(crate) status: RuntimeStatus, - pub(crate) limits: RuntimeLimits, pub(crate) next_worker_sequence: u64, - pub(crate) next_event_id: u64, pub(crate) next_diagnostic_id: u64, pub(crate) workers: BTreeMap, pub(crate) workspace_owners: BTreeMap, pub(crate) config_bundles: BTreeMap, - pub(crate) events: Vec, pub(crate) diagnostics: Vec, } @@ -305,7 +261,6 @@ pub(crate) struct PersistedWorkerRecord { pub(crate) request: CreateWorkerRequest, pub(crate) workspace_id: Option, pub(crate) working_directory: Option, - pub(crate) last_event_id: u64, } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -314,9 +269,7 @@ struct RuntimeSnapshot { display_name: Option, backend: RuntimeBackendKind, status: RuntimeStatus, - limits: RuntimeLimits, next_worker_sequence: u64, - next_event_id: u64, next_diagnostic_id: u64, #[serde(default)] config_bundles: BTreeMap, @@ -348,9 +301,7 @@ impl RuntimeSnapshot { display_name: state.display_name.clone(), backend: RuntimeBackendKind::FsStore, status: state.status, - limits: state.limits.clone(), next_worker_sequence: state.next_worker_sequence, - next_event_id: state.next_event_id, next_diagnostic_id: state.next_diagnostic_id, config_bundles: state.config_bundles.clone(), workspace_owners: state.workspace_owners.clone(), @@ -381,20 +332,16 @@ impl RuntimeSnapshot { fn into_persisted( self, - events: Vec, workers: BTreeMap, ) -> PersistedRuntimeState { PersistedRuntimeState { display_name: self.display_name, status: self.status, - limits: self.limits, next_worker_sequence: self.next_worker_sequence, - next_event_id: self.next_event_id, next_diagnostic_id: self.next_diagnostic_id, workers, config_bundles: self.config_bundles, workspace_owners: self.workspace_owners, - events, diagnostics: self.diagnostics, } } @@ -414,7 +361,6 @@ struct WorkerSnapshot { /// write the removed execution projection. #[serde(default, rename = "execution", skip_serializing)] legacy_execution: Option, - last_event_id: u64, } #[derive(Clone, Debug, Deserialize)] @@ -433,7 +379,6 @@ impl WorkerSnapshot { workspace_id: worker.workspace_id.clone(), working_directory: worker.working_directory.clone(), legacy_execution: None, - last_event_id: worker.last_event_id, } } @@ -477,7 +422,6 @@ impl WorkerSnapshot { self.legacy_execution .and_then(|execution| execution.working_directory) }), - last_event_id: self.last_event_id, } } } @@ -525,11 +469,6 @@ fn migrate_legacy_single_runtime_layout(root: &Path) -> Result<(), RuntimeError> &root.join(RUNTIME_FILE), "migrate legacy runtime snapshot", )?; - rename_if_exists( - &legacy_dir.join(EVENTS_FILE), - &root.join(EVENTS_FILE), - "migrate legacy runtime events", - )?; rename_if_exists( &legacy_dir.join(WORKERS_DIR), &root.join(WORKERS_DIR), @@ -581,42 +520,6 @@ where }) } -fn read_json_lines(path: &Path, operation: &'static str) -> Result, RuntimeError> -where - T: for<'de> Deserialize<'de>, -{ - let file = File::open(path).map_err(|source| match source.kind() { - std::io::ErrorKind::NotFound => RuntimeError::StoreMissing { - operation, - path: path.to_path_buf(), - }, - _ => RuntimeError::StoreIo { - operation, - path: path.to_path_buf(), - source, - }, - })?; - let reader = BufReader::new(file); - let mut items = Vec::new(); - for (index, line) in reader.lines().enumerate() { - let line = line.map_err(|source| RuntimeError::StoreIo { - operation, - path: path.to_path_buf(), - source, - })?; - if line.trim().is_empty() { - continue; - } - let item = serde_json::from_str(&line).map_err(|source| RuntimeError::StoreCorrupt { - operation, - path: path.to_path_buf(), - message: format!("line {}: {source}", index + 1), - })?; - items.push(item); - } - Ok(items) -} - fn atomic_write_json(path: &Path, value: &T, operation: &'static str) -> Result<(), RuntimeError> where T: Serialize, @@ -676,44 +579,6 @@ where write_result } -fn append_json_line(path: &Path, value: &T, operation: &'static str) -> Result<(), RuntimeError> -where - T: Serialize, -{ - let parent = path.parent().ok_or_else(|| RuntimeError::StoreCorrupt { - operation, - path: path.to_path_buf(), - message: "path has no parent directory".to_string(), - })?; - fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo { - operation, - path: parent.to_path_buf(), - source, - })?; - let mut file = OpenOptions::new() - .create(true) - .append(true) - .open(path) - .map_err(|source| RuntimeError::StoreIo { - operation, - path: path.to_path_buf(), - source, - })?; - serde_json::to_writer(&mut file, value).map_err(|source| RuntimeError::StoreCorrupt { - operation, - path: path.to_path_buf(), - message: format!("serialize json: {source}"), - })?; - file.write_all(b"\n") - .and_then(|()| file.flush()) - .and_then(|()| file.sync_all()) - .map_err(|source| RuntimeError::StoreIo { - operation, - path: path.to_path_buf(), - source, - }) -} - fn tmp_path_for(path: &Path) -> PathBuf { let sequence = NEXT_TMP_SEQUENCE.fetch_add(1, Ordering::Relaxed); let file_name = path diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 8812390c..20391dea 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -18,7 +18,7 @@ use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleS use crate::error::RuntimeError; use crate::identity::{WorkerId, WorkerRef}; use crate::interaction::{WorkerInput, WorkerInteractionAck}; -use crate::management::{RuntimeLimits, RuntimeSummary, WorkerDeleteResult}; +use crate::management::{RuntimeSummary, WorkerDeleteResult}; #[cfg(feature = "ws-server")] use crate::observation::WorkerObservationCursor; #[cfg(feature = "ws-server")] @@ -68,8 +68,6 @@ pub struct RuntimeHttpServerConfig { pub bind_addr: SocketAddr, /// Optional display label surfaced by `GET /v1/runtime`. pub display_name: Option, - /// Bounded Runtime API limits. - pub limits: RuntimeLimits, /// v0 store selection for the Runtime process. pub store: RuntimeHttpStoreSelection, /// Minimal local bearer token placeholder for backend-to-Runtime calls. @@ -84,7 +82,6 @@ impl Default for RuntimeHttpServerConfig { Self { bind_addr: default_runtime_http_bind_addr(), display_name: None, - limits: RuntimeLimits::default(), store: RuntimeHttpStoreSelection::Memory, local_token: None, auth: None, @@ -97,7 +94,6 @@ impl fmt::Debug for RuntimeHttpServerConfig { f.debug_struct("RuntimeHttpServerConfig") .field("bind_addr", &self.bind_addr) .field("display_name", &self.display_name) - .field("limits", &self.limits) .field("store", &self.store) .field( "local_token", diff --git a/crates/worker-runtime/src/interaction.rs b/crates/worker-runtime/src/interaction.rs index 225544c7..3a912fa0 100644 --- a/crates/worker-runtime/src/interaction.rs +++ b/crates/worker-runtime/src/interaction.rs @@ -52,5 +52,4 @@ impl WorkerInput { pub struct WorkerInteractionAck { pub worker_ref: WorkerRef, pub status: WorkerStatus, - pub event_id: u64, } diff --git a/crates/worker-runtime/src/lib.rs b/crates/worker-runtime/src/lib.rs index e6d3adec..43d226bf 100644 --- a/crates/worker-runtime/src/lib.rs +++ b/crates/worker-runtime/src/lib.rs @@ -28,5 +28,5 @@ pub mod working_directory; #[cfg(feature = "fs-store")] pub use fs_store::{FsRuntimeStore, FsRuntimeStoreOptions}; -pub use management::{RuntimeLimits, RuntimeOptions}; +pub use management::RuntimeOptions; pub use runtime::{Runtime, RuntimeWorkspaceScope}; diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index 19d942c4..1ecc3695 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -107,7 +107,6 @@ fn build_runtime(config: &ProcessConfig) -> Result { RuntimeHttpStoreSelection::Fs { root } => { let mut options = FsRuntimeStoreOptions::new(root.clone()); options.display_name = config.http.display_name.clone(); - options.limits = config.http.limits.clone(); Runtime::with_fs_store_and_execution_backend(options, backend) .map_err(ProcessError::Runtime) } @@ -120,7 +119,6 @@ fn build_runtime(config: &ProcessConfig) -> Result { fn runtime_options_from_http(config: &RuntimeHttpServerConfig) -> RuntimeOptions { RuntimeOptions { display_name: config.display_name.clone(), - limits: config.limits.clone(), } } @@ -207,10 +205,6 @@ where } config.http.local_token = Some(value); } - "--max-event-batch-items" => { - config.http.limits.max_event_batch_items = - parse_usize_flag(&flag, take_value(&flag, inline_value, &mut args)?)?; - } _ => { return Err(ProcessError::usage(format!("unknown argument `{flag}`"))); } @@ -255,12 +249,6 @@ fn ensure_no_inline_value(flag: &str, inline_value: Option<&str>) -> Result<(), Ok(()) } -fn parse_usize_flag(flag: &str, value: String) -> Result { - value - .parse::() - .map_err(|error| ProcessError::usage(format!("invalid {flag} value `{value}`: {error}"))) -} - fn apply_store_selection(config: &mut ProcessConfig) { if config.no_store { config.http.store = RuntimeHttpStoreSelection::Memory; @@ -806,7 +794,6 @@ Options: --no-store Disable Runtime catalog persistence for ephemeral runs --local-token Minimal local bearer token placeholder --local-token-env Read local bearer token placeholder from env - --max-event-batch-items Override event batch limit -h, --help Show this help Auth commands: diff --git a/crates/worker-runtime/src/management.rs b/crates/worker-runtime/src/management.rs index 3b6110c2..16fec9c0 100644 --- a/crates/worker-runtime/src/management.rs +++ b/crates/worker-runtime/src/management.rs @@ -18,34 +18,10 @@ pub enum RuntimeStatus { Stopped, } -/// Guardrails for bounded Runtime APIs. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct RuntimeLimits { - pub max_event_batch_items: usize, -} - -impl Default for RuntimeLimits { - fn default() -> Self { - Self { - max_event_batch_items: 256, - } - } -} - /// Options used to construct an embedded memory Runtime. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RuntimeOptions { pub display_name: Option, - pub limits: RuntimeLimits, -} - -impl Default for RuntimeOptions { - fn default() -> Self { - Self { - display_name: None, - limits: RuntimeLimits::default(), - } - } } fn unknown_platform_component() -> String { @@ -69,7 +45,6 @@ pub struct RuntimeSummary { pub stopped_worker_count: usize, pub cancelled_worker_count: usize, pub diagnostic_count: usize, - pub limits: RuntimeLimits, #[serde(default = "unknown_platform_component")] pub os: String, #[serde(default = "unknown_platform_component")] diff --git a/crates/worker-runtime/src/observation.rs b/crates/worker-runtime/src/observation.rs index 87f5edb6..7282981f 100644 --- a/crates/worker-runtime/src/observation.rs +++ b/crates/worker-runtime/src/observation.rs @@ -1,56 +1,6 @@ use crate::identity::WorkerRef; use serde::{Deserialize, Serialize}; -/// Event cursor. `next_event_id` is the first event id that should be returned -/// by the next poll. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct EventCursor { - pub next_event_id: u64, -} - -/// Placeholder subscription handle for future streaming APIs. v0 is explicit -/// poll-only so HTTP/WS/SSE dependencies are not pulled into this crate. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct EventSubscription { - pub cursor: EventCursor, - pub mode: EventSubscriptionMode, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum EventSubscriptionMode { - PollOnly, -} - -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct RuntimeEvent { - pub id: u64, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub worker_ref: Option, - pub kind: RuntimeEventKind, - pub message: String, -} - -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -#[serde(tag = "kind", rename_all = "snake_case")] -pub enum RuntimeEventKind { - RuntimeStarted, - RuntimeStopped, - WorkerCreated, - WorkerExecutionRestored, - WorkerInputAccepted, - WorkerStopped, - WorkerCancelled, - WorkerDeleted, -} - -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct RuntimeEventBatch { - pub cursor: EventCursor, - pub events: Vec, - pub has_more: bool, -} - /// Runtime-local cursor for worker-scoped WebSocket observation. #[cfg(feature = "ws-server")] #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index eafbbeb5..52ee3e40 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -22,12 +22,7 @@ use crate::fs_store::{ 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, + RuntimeBackendKind, RuntimeOptions, RuntimeStatus, RuntimeSummary, WorkerDeleteResult, }; #[cfg(feature = "ws-server")] use crate::observation::{WorkerObservationCursor, WorkerObservationEvent}; @@ -62,7 +57,7 @@ impl RuntimeWorkspaceScope { } } -const RUNTIME_EVENT_SUBSCRIPTION_QUEUE_CAPACITY: usize = 256; +const SUBSCRIPTION_QUEUE_CAPACITY: usize = 256; #[derive(Clone, Debug)] pub struct RuntimeSubscriptionUpdate { @@ -130,7 +125,7 @@ impl Drop for RuntimeEventSelectorSubscription { return; }; if let Ok(mut state) = runtime.lock() { - state.event_subscriptions.remove(&self.subscription_id); + state.subscriptions.remove(&self.subscription_id); } } } @@ -148,15 +143,14 @@ pub struct Runtime { } impl Runtime { - /// Create a memory-backed Runtime with generated identity and default limits. + /// Create a memory-backed Runtime with generated identity. 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"); + let state = RuntimeState::new(options.display_name); Self { inner: Arc::new(Mutex::new(state)), } @@ -200,12 +194,8 @@ impl Runtime { 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"); + let state = RuntimeState::new_fs_backed(options.display_name, opened.store); state.persist_runtime_snapshot()?; - state.persist_event_by_id(event_id)?; state }; state.execution_backend = execution_backend; @@ -241,7 +231,6 @@ impl Runtime { 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(), @@ -294,22 +283,25 @@ impl Runtime { /// Stop the Runtime. v0 keeps data readable after stop, but rejects new /// create/send/worker lifecycle mutations. - pub fn stop_runtime(&self) -> Result { + pub fn stop_runtime(&self) -> Result<(), RuntimeError> { let mut state = self.lock()?; if state.status == RuntimeStatus::Stopped { - return Ok(state.last_event_id()); + return Ok(()); } state.status = RuntimeStatus::Stopped; - for worker in state.workers.values_mut() { + let mut stopped = Vec::new(); + for (worker_id, worker) in &mut state.workers { if worker.status.is_active() { worker.status = WorkerStatus::Stopped; + stopped.push(*worker_id); } } - let event_id = state.push_event(None, RuntimeEventKind::RuntimeStopped, "runtime stopped"); + for worker_id in stopped { + state.publish_worker_upsert(worker_id)?; + } state.persist_runtime_snapshot()?; state.persist_workers()?; - state.persist_event_by_id(event_id)?; - Ok(event_id) + Ok(()) } /// Create a Runtime-owned working directory through the attached execution backend. @@ -483,11 +475,6 @@ impl Runtime { 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(), @@ -497,7 +484,6 @@ impl Runtime { request: request.clone(), working_directory: None, execution_handle: None, - last_event_id: event_id, }; state.workers.insert(worker_id, record); let spawn_request = WorkerExecutionSpawnRequest { @@ -617,11 +603,11 @@ impl Runtime { let snapshot_revision = state.subscription_revision; let subscription_id = state.next_event_subscription_id; state.next_event_subscription_id = state.next_event_subscription_id.saturating_add(1); - let (sender, receiver) = mpsc::channel(RUNTIME_EVENT_SUBSCRIPTION_QUEUE_CAPACITY); + let (sender, receiver) = mpsc::channel(SUBSCRIPTION_QUEUE_CAPACITY); let lagged = Arc::new(AtomicBool::new(false)); - state.event_subscriptions.insert( + state.subscriptions.insert( subscription_id, - RuntimeEventSubscriptionSink { + SubscriptionSink { selector: selector.clone(), workspace_id: scope.map(|scope| scope.workspace_id.clone()), sender, @@ -977,13 +963,7 @@ impl Runtime { 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.status = worker_status_from_run_state(dispatch_result.run_state); let status = worker.status; #[cfg(feature = "ws-server")] @@ -994,12 +974,10 @@ impl Runtime { state.publish_worker_upsert(worker_ref.worker_id)?; state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; - state.persist_event_by_id(event_id)?; Ok(WorkerInteractionAck { worker_ref: worker_ref.clone(), status, - event_id, }) } @@ -1128,7 +1106,6 @@ impl Runtime { state.publish_worker_upsert(worker_ref.worker_id)?; state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; - state.persist_event_by_id(detail.last_event_id)?; Ok(detail) } @@ -1139,9 +1116,6 @@ impl Runtime { if let Some(workspace_id) = workspace_id.as_deref() { state.forget_workspace_owner_if_unused(workspace_id); } - state.events.retain(|event| { - event.id != record.last_event_id || event.worker_ref.as_ref() != Some(worker_ref) - }); state.publish_worker_removed(worker_ref.worker_id, workspace_id.as_deref())?; } Ok(()) @@ -1215,19 +1189,15 @@ impl Runtime { self.stop_worker(worker_ref, reason) } - /// Stop a Worker. Repeated stops are idempotent and return the last event id. + /// Stop a Worker. Repeated stops are idempotent. pub fn stop_worker( &self, worker_ref: &WorkerRef, reason: Option, ) -> Result { 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()), - ) + let _ = reason; + self.transition_worker(worker_ref, WorkerStatus::Stopped) } /// Cancel a Worker through a workspace-scoped Runtime authorization context. @@ -1241,19 +1211,15 @@ impl Runtime { self.cancel_worker(worker_ref, reason) } - /// Cancel a Worker. Repeated cancels are idempotent and return the last event id. + /// Cancel a Worker. Repeated cancels are idempotent. pub fn cancel_worker( &self, worker_ref: &WorkerRef, reason: Option, ) -> Result { 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()), - ) + let _ = reason; + self.transition_worker(worker_ref, WorkerStatus::Cancelled) } /// Delete a non-running Worker through a workspace-scoped Runtime authorization context. @@ -1294,77 +1260,15 @@ impl Runtime { 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.publish_worker_removed(worker_ref.worker_id, removed_workspace_id.as_deref())?; 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 { - Ok(EventCursor { next_event_id: 1 }) - } - - /// Cursor pointing after the current last event. - pub fn event_cursor_now(&self) -> Result { - 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 { - 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 { - 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( @@ -1492,8 +1396,6 @@ impl Runtime { &self, worker_ref: &WorkerRef, status: WorkerStatus, - event_kind: RuntimeEventKind, - reason: String, ) -> Result { let mut state = self.lock()?; state.ensure_running()?; @@ -1505,25 +1407,20 @@ impl Runtime { 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.execution_handle = None; - worker.last_event_id = event_id; let status = worker.status; state.publish_worker_upsert(worker_ref.worker_id)?; 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, }) } @@ -1633,22 +1530,15 @@ impl Runtime { ) -> 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); worker.status = worker_status_from_run_state(run_state); worker.working_directory = working_directory; - worker.last_event_id = event_id; } state.publish_worker_upsert(worker_ref.worker_id)?; state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; - state.persist_event_by_id(event_id)?; Ok(()) } @@ -1666,7 +1556,7 @@ enum RuntimePersistence { } #[derive(Debug)] -struct RuntimeEventSubscriptionSink { +struct SubscriptionSink { selector: EventSubscriptionSelector, workspace_id: Option, sender: mpsc::Sender, @@ -1680,21 +1570,18 @@ struct RuntimeState { #[cfg_attr(not(feature = "fs-store"), allow(dead_code))] persistence: RuntimePersistence, status: RuntimeStatus, - limits: RuntimeLimits, execution_backend: Option, next_worker_sequence: u64, - next_event_id: u64, #[cfg(feature = "fs-store")] next_diagnostic_id: u64, workers: BTreeMap, workspace_owners: BTreeMap, config_bundles: BTreeMap, - events: Vec, diagnostics: Vec, subscription_revision: u64, worker_subject_revisions: BTreeMap, next_event_subscription_id: u64, - event_subscriptions: BTreeMap, + subscriptions: BTreeMap, #[cfg(feature = "ws-server")] next_observation_sequence: u64, #[cfg(feature = "ws-server")] @@ -1704,27 +1591,24 @@ struct RuntimeState { } impl RuntimeState { - fn new(display_name: Option, limits: RuntimeLimits) -> Self { + fn new(display_name: Option) -> 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(), workspace_owners: BTreeMap::new(), config_bundles: BTreeMap::new(), - events: Vec::new(), diagnostics: Vec::new(), subscription_revision: 0, worker_subject_revisions: BTreeMap::new(), next_event_subscription_id: 1, - event_subscriptions: BTreeMap::new(), + subscriptions: BTreeMap::new(), #[cfg(feature = "ws-server")] next_observation_sequence: 1, #[cfg(feature = "ws-server")] @@ -1735,31 +1619,24 @@ impl RuntimeState { } #[cfg(feature = "fs-store")] - fn new_fs_backed( - display_name: Option, - limits: RuntimeLimits, - store: FsRuntimeStore, - ) -> Self { + fn new_fs_backed(display_name: Option, 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(), workspace_owners: BTreeMap::new(), config_bundles: BTreeMap::new(), - events: Vec::new(), diagnostics: Vec::new(), subscription_revision: 0, worker_subject_revisions: BTreeMap::new(), next_event_subscription_id: 1, - event_subscriptions: BTreeMap::new(), + subscriptions: BTreeMap::new(), #[cfg(feature = "ws-server")] next_observation_sequence: 1, #[cfg(feature = "ws-server")] @@ -1788,7 +1665,6 @@ impl RuntimeState { request: worker.request, working_directory: worker.working_directory, execution_handle: None, - last_event_id: worker.last_event_id, }, ); } @@ -1798,20 +1674,17 @@ impl RuntimeState { 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, workspace_owners: persisted.workspace_owners, - events: persisted.events, diagnostics, subscription_revision: 0, worker_subject_revisions: BTreeMap::new(), next_event_subscription_id: 1, - event_subscriptions: BTreeMap::new(), + subscriptions: BTreeMap::new(), #[cfg(feature = "ws-server")] next_observation_sequence: 1, #[cfg(feature = "ws-server")] @@ -1826,9 +1699,7 @@ impl RuntimeState { 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 @@ -1837,7 +1708,6 @@ impl RuntimeState { .collect(), config_bundles: self.config_bundles.clone(), workspace_owners: self.workspace_owners.clone(), - events: self.events.clone(), diagnostics: self.diagnostics.clone(), } } @@ -1880,23 +1750,6 @@ impl RuntimeState { 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(feature = "fs-store")] fn persist_workers(&self) -> Result<(), RuntimeError> { if self.fs_store().is_some() { @@ -1922,11 +1775,6 @@ impl RuntimeState { 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(()) @@ -2191,7 +2039,7 @@ impl RuntimeState { update: RuntimeSubscriptionUpdate, ) { let mut closed = Vec::new(); - for (subscription_id, sink) in &self.event_subscriptions { + for (subscription_id, sink) in &self.subscriptions { if sink .workspace_id .as_deref() @@ -2221,7 +2069,7 @@ impl RuntimeState { } } for subscription_id in closed { - self.event_subscriptions.remove(&subscription_id); + self.subscriptions.remove(&subscription_id); } } @@ -2270,27 +2118,6 @@ impl RuntimeState { Ok(()) } - fn push_event( - &mut self, - worker_ref: Option, - kind: RuntimeEventKind, - message: impl Into, - ) -> 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, @@ -2391,7 +2218,6 @@ struct WorkerRecord { request: CreateWorkerRequest, working_directory: Option, execution_handle: Option, - last_event_id: u64, } impl WorkerRecord { @@ -2410,7 +2236,6 @@ impl WorkerRecord { display_name: self.request.display_name.clone(), profile_source: self.request.profile_source.reference(), config_bundle: self.request.config_bundle.clone(), - last_event_id: self.last_event_id, } } @@ -2425,7 +2250,6 @@ impl WorkerRecord { display_name: self.request.display_name.clone(), profile_source: self.request.profile_source.reference(), config_bundle: self.request.config_bundle.clone(), - last_event_id: self.last_event_id, } } @@ -2437,7 +2261,6 @@ impl WorkerRecord { request: self.request.clone(), workspace_id: self.workspace_id.clone(), working_directory: self.working_directory.clone(), - last_event_id: self.last_event_id, } } } @@ -2882,7 +2705,7 @@ mod tests { payload => panic!("unexpected subscription payload: {payload:?}"), } - runtime.stop_worker(&created.worker_ref, None).unwrap(); + runtime.stop_runtime().unwrap(); let update = receive_subscription_update(&mut subscription).unwrap(); assert_eq!(update.subject_revision, 2); match update.payload { @@ -2977,14 +2800,11 @@ mod tests { .unwrap(); { let mut state = runtime.lock().unwrap(); - for _ in 0..=RUNTIME_EVENT_SUBSCRIPTION_QUEUE_CAPACITY { + for _ in 0..=SUBSCRIPTION_QUEUE_CAPACITY { state.publish_worker_upsert(created.worker_id).unwrap(); } } - assert_eq!( - subscription.receiver.len(), - RUNTIME_EVENT_SUBSCRIPTION_QUEUE_CAPACITY - ); + assert_eq!(subscription.receiver.len(), SUBSCRIPTION_QUEUE_CAPACITY); assert!(matches!( receive_subscription_update(&mut subscription), Err(RuntimeSubscriptionRecvError::Lagged) @@ -2997,9 +2817,9 @@ mod tests { let subscription = runtime .subscribe_event_selector(EventSubscriptionSelector::RuntimeWorkers) .unwrap(); - assert_eq!(runtime.lock().unwrap().event_subscriptions.len(), 1); + assert_eq!(runtime.lock().unwrap().subscriptions.len(), 1); drop(subscription); - assert!(runtime.lock().unwrap().event_subscriptions.is_empty()); + assert!(runtime.lock().unwrap().subscriptions.is_empty()); } #[test] @@ -3347,15 +3167,6 @@ mod tests { 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] @@ -3550,14 +3361,6 @@ mod tests { runtime.worker_detail(&detail.worker_ref).unwrap().status, WorkerStatus::Stopped ); - let events = runtime - .read_events(&runtime.event_cursor_from_start().unwrap(), 16) - .unwrap() - .events; - assert!(events.iter().any(|event| { - event.kind == RuntimeEventKind::WorkerStopped - && event.worker_ref.as_ref() == Some(&detail.worker_ref) - })); } #[test] @@ -3586,9 +3389,6 @@ mod tests { 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()), @@ -3597,7 +3397,7 @@ mod tests { runtime.store_config_bundle(test_bundle()).unwrap(); let detail = runtime.create_worker(task_request("chat")).unwrap(); - let first = runtime + runtime .send_input(&detail.worker_ref, WorkerInput::user("hello")) .unwrap(); runtime @@ -3607,7 +3407,6 @@ mod tests { 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, @@ -3665,7 +3464,6 @@ mod tests { #[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(); @@ -3679,7 +3477,6 @@ mod tests { 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 @@ -3689,28 +3486,11 @@ mod tests { 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(); @@ -3724,7 +3504,6 @@ mod tests { 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 @@ -3734,46 +3513,6 @@ mod tests { 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")] @@ -3801,15 +3540,12 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_restores_workers_and_events_but_not_protocol_observations() { + fn fs_store_restores_workers_without_legacy_event_or_protocol_observation_logs() { 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()), ) @@ -3836,22 +3572,23 @@ mod tests { .unwrap(); assert!(worker_snapshot.get("status").is_none()); assert!(worker_snapshot.get("execution").is_none()); + assert!(!root.join("events.jsonl").exists()); std::fs::write( worker_store_dir.join("observations.jsonl"), b"{\"legacy\":true}\n", ) .unwrap(); - let store = runtime_store(&runtime); + std::fs::write(root.join("events.jsonl"), b"obsolete runtime event\n").unwrap(); 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!(!root.join("events.jsonl").exists()); assert!(!worker_store_dir.join("observations.jsonl").exists()); #[cfg(feature = "ws-server")] { @@ -3861,15 +3598,6 @@ mod tests { assert!(observations.is_empty()); } - 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 @@ -3899,7 +3627,6 @@ mod tests { crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), display_name: None, - limits: RuntimeLimits::default(), }, Arc::new(TestExecutionBackend::default()), ) @@ -3925,7 +3652,6 @@ mod tests { let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), display_name: None, - limits: RuntimeLimits::default(), }) .unwrap(); let restored_scoped = restored.worker_detail(&scoped.worker_ref).unwrap(); @@ -3975,7 +3701,6 @@ mod tests { crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), display_name: None, - limits: RuntimeLimits::default(), }, Arc::new(TestExecutionBackend::default()), ) @@ -3989,7 +3714,6 @@ mod tests { let backendless = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), display_name: None, - limits: RuntimeLimits::default(), }) .unwrap(); let stopped_worker = backendless.worker_detail(&worker.worker_ref).unwrap(); @@ -4001,7 +3725,6 @@ mod tests { crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), display_name: None, - limits: RuntimeLimits::default(), }, restoring_backend.clone(), ) @@ -4013,18 +3736,6 @@ mod tests { 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); } @@ -4037,7 +3748,6 @@ mod tests { crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), display_name: None, - limits: RuntimeLimits::default(), }, Arc::new(TestExecutionBackend::default()), ) @@ -4057,7 +3767,6 @@ mod tests { crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), display_name: None, - limits: RuntimeLimits::default(), }, restoring_backend.clone(), ) @@ -4094,7 +3803,6 @@ mod tests { 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); @@ -4107,7 +3815,6 @@ mod tests { 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 { .. })); @@ -4118,7 +3825,6 @@ mod tests { crate::fs_store::FsRuntimeStoreOptions { root: missing_root.clone(), display_name: None, - limits: RuntimeLimits::default(), }, Arc::new(TestExecutionBackend::default()), ) @@ -4138,7 +3844,6 @@ mod tests { 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()); diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index dbbe3ba3..a7c1951e 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -494,8 +494,6 @@ pub struct WorkerLifecycleResult { pub state: WorkerOperationState, pub runtime_id: String, pub worker_id: String, - #[serde(skip_serializing_if = "Option::is_none")] - pub event_id: Option, pub diagnostics: Vec, } @@ -549,8 +547,6 @@ pub struct WorkerInputResult { pub state: WorkerOperationState, pub runtime_id: String, pub worker_id: String, - #[serde(skip_serializing_if = "Option::is_none")] - pub event_id: Option, pub diagnostics: Vec, } @@ -795,7 +791,6 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { state: WorkerOperationState::Unsupported, runtime_id: self.runtime_id().to_string(), worker_id: worker_id.to_string(), - event_id: None, diagnostics: vec![diagnostic( "worker_stop_pending", DiagnosticSeverity::Info, @@ -815,7 +810,6 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { state: WorkerOperationState::Unsupported, runtime_id: self.runtime_id().to_string(), worker_id: worker_id.to_string(), - event_id: None, diagnostics: vec![diagnostic( "worker_cancel_pending", DiagnosticSeverity::Info, @@ -852,7 +846,6 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { state: WorkerOperationState::Unsupported, runtime_id: self.runtime_id().to_string(), worker_id: worker_id.to_string(), - event_id: None, diagnostics: vec![diagnostic( "worker_input_pending", DiagnosticSeverity::Info, @@ -1413,7 +1406,6 @@ impl EmbeddedWorkerRuntime { FsRuntimeStoreOptions { root: store_root.into(), display_name: Some("embedded".to_string()), - limits: EmbeddedRuntimeOptions::default().limits, }, backend, )?; @@ -1944,11 +1936,10 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { ); }; match self.runtime.stop_worker(&worker_ref, request.reason) { - Ok(ack) => WorkerLifecycleResult { + Ok(_) => WorkerLifecycleResult { state: WorkerOperationState::Accepted, runtime_id: self.runtime_id.clone(), worker_id: worker_id.to_string(), - event_id: Some(ack.event_id), diagnostics: Vec::new(), }, Err(error) => embedded_lifecycle_rejected( @@ -1989,11 +1980,10 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { ); }; match self.runtime.cancel_worker(&worker_ref, request.reason) { - Ok(ack) => WorkerLifecycleResult { + Ok(_) => WorkerLifecycleResult { state: WorkerOperationState::Accepted, runtime_id: self.runtime_id.clone(), worker_id: worker_id.to_string(), - event_id: Some(ack.event_id), diagnostics: Vec::new(), }, Err(error) => embedded_lifecycle_rejected( @@ -2120,11 +2110,10 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { segments: request.segments, }; match self.runtime.send_input(&worker_ref, input) { - Ok(ack) => WorkerInputResult { + Ok(_) => WorkerInputResult { state: WorkerOperationState::Accepted, runtime_id: self.runtime_id.clone(), worker_id: worker_id.to_string(), - event_id: Some(ack.event_id), diagnostics: Vec::new(), }, Err(error) => embedded_input_rejected( @@ -2545,7 +2534,6 @@ impl RemoteWorkerRuntime { state: WorkerOperationState::Accepted, runtime_id: self.runtime_id.clone(), worker_id: worker_id.to_string(), - event_id: Some(response.ack.event_id), diagnostics: vec![diagnostic( "remote_runtime_lifecycle_accepted", DiagnosticSeverity::Info, @@ -2991,11 +2979,10 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { &format!("/v1/workers/{worker_id}/input"), &input, ) { - Ok(response) => WorkerInputResult { + Ok(_) => WorkerInputResult { state: WorkerOperationState::Accepted, runtime_id: self.runtime_id.clone(), worker_id: worker_id.to_string(), - event_id: Some(response.ack.event_id), diagnostics: Vec::new(), }, Err(diagnostic) => remote_input_rejected(&self.runtime_id, worker_id, diagnostic), @@ -3390,7 +3377,6 @@ fn embedded_input_rejected( state: WorkerOperationState::Rejected, runtime_id: runtime_id.to_string(), worker_id: worker_id.to_string(), - event_id: None, diagnostics: vec![diagnostic], } } @@ -3404,7 +3390,6 @@ fn remote_input_rejected( state: WorkerOperationState::Rejected, runtime_id: runtime_id.to_string(), worker_id: worker_id.to_string(), - event_id: None, diagnostics: vec![diagnostic], } } @@ -3418,7 +3403,6 @@ fn embedded_lifecycle_rejected( state: WorkerOperationState::Rejected, runtime_id: runtime_id.to_string(), worker_id: worker_id.to_string(), - event_id: None, diagnostics: vec![diagnostic], } } @@ -3432,7 +3416,6 @@ fn remote_lifecycle_rejected( state: WorkerOperationState::Rejected, runtime_id: runtime_id.to_string(), worker_id: worker_id.to_string(), - event_id: None, diagnostics: vec![diagnostic], } } @@ -4689,8 +4672,7 @@ mod tests { json!({ "ack": { "worker_ref": { "runtime_id": "remote:primary", "worker_id": 1 }, - "status": "running", - "event_id": 8 + "status": "running" } }) .to_string(), @@ -4749,7 +4731,6 @@ mod tests { ) .unwrap(); assert_eq!(input.state, WorkerOperationState::Accepted); - assert_eq!(input.event_id, Some(8)); server.join().expect("mock remote server finished"); let browser_payload = serde_json::to_string(&(workers, input)).unwrap(); @@ -5023,8 +5004,7 @@ mod tests { "size_bytes": 0, "source_graph": { "source_count": 0, "total_source_bytes": 0, "entrypoints": {}, "import_count": 0 } }, - "config_bundle": { "id": "remote-bundle", "digest": "remote-digest" }, - "last_event_id": 0 + "config_bundle": { "id": "remote-bundle", "digest": "remote-digest" } }) } } diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 77287faa..aaa9bc9b 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -4487,7 +4487,8 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); #[tokio::test] async fn worker_credential_binds_once_and_notification_outbox_is_durable() { let dir = tempfile::tempdir().unwrap(); - let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap(); + let db = dir.path().join("server.db"); + let store = SqliteWorkspaceStore::open(&db).unwrap(); store .upsert_workspace(&WorkspaceRecord { workspace_id: "workspace-a".to_string(), @@ -4592,6 +4593,9 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); }], ) .unwrap(); + + drop(store); + let store = SqliteWorkspaceStore::open(&db).unwrap(); let pending = store .list_pending_ticket_notification_deliveries("workspace-a", 10) .unwrap(); diff --git a/web/workspace/src/lib/workspace/sidebar/types.ts b/web/workspace/src/lib/workspace/sidebar/types.ts index 0779f88f..47b1c102 100644 --- a/web/workspace/src/lib/workspace/sidebar/types.ts +++ b/web/workspace/src/lib/workspace/sidebar/types.ts @@ -253,7 +253,6 @@ export type WorkerInputResult = { state: WorkerOperationState; runtime_id: string; worker_id: string; - event_id?: number | null; diagnostics: Diagnostic[]; };