diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index f90490ce..6f04d0f6 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -85,24 +85,19 @@ impl AuthenticatedInputSource { } } -/// Immutable identity and revision fence for one state-changing Worker command. +/// Caller-owned identity for one state-changing Worker command. +/// +/// A controller accepts command ids in strictly increasing order. Exact retries +/// of an accepted id must retain the same command kind. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] pub struct WorkerCommandEnvelope { - /// Caller-owned sequence. A controller accepts command ids in strictly - /// increasing order for one execution generation. pub command_id: u64, - pub expected_execution_generation: u64, - pub expected_worker_state_revision: u64, } impl WorkerCommandEnvelope { - pub fn for_snapshot(command_id: u64, snapshot: &WorkerStateSnapshot) -> Self { - Self { - command_id, - expected_execution_generation: snapshot.execution_generation, - expected_worker_state_revision: snapshot.revision, - } + pub fn new(command_id: u64) -> Self { + Self { command_id } } } @@ -122,8 +117,6 @@ pub enum WorkerCommandKind { #[serde(rename_all = "snake_case")] pub enum WorkerCommandDisposition { Accepted, - StaleExecutionGeneration, - StaleWorkerStateRevision, StaleCommandId, Conflict, InvalidState, @@ -175,18 +168,14 @@ pub enum WorkerMaintenanceState { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] pub struct WorkerStateSnapshot { - pub execution_generation: u64, - pub revision: u64, - /// Highest lifecycle command id observed by this controller generation. + /// Highest lifecycle command id observed by this controller instance. pub last_command_id: u64, pub state: WorkerState, } impl WorkerStateSnapshot { - pub fn initial(execution_generation: u64) -> Self { + pub fn initial() -> Self { Self { - execution_generation, - revision: 0, last_command_id: 0, state: WorkerState::Idle, } @@ -204,53 +193,6 @@ impl WorkerStateSnapshot { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum WorkerStateSnapshotApply { - Applied, - Duplicate, - Stale, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct WorkerStateSnapshotConflict { - pub execution_generation: u64, - pub revision: u64, -} - -impl std::fmt::Display for WorkerStateSnapshotConflict { - fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!( - formatter, - "conflicting worker state snapshots at generation {} revision {}", - self.execution_generation, self.revision - ) - } -} - -impl std::error::Error for WorkerStateSnapshotConflict {} - -pub fn apply_worker_state_snapshot( - current: &mut WorkerStateSnapshot, - incoming: &WorkerStateSnapshot, -) -> Result { - use std::cmp::Ordering; - - let ordering = (incoming.execution_generation, incoming.revision) - .cmp(&(current.execution_generation, current.revision)); - match ordering { - Ordering::Greater => { - *current = incoming.clone(); - Ok(WorkerStateSnapshotApply::Applied) - } - Ordering::Less => Ok(WorkerStateSnapshotApply::Stale), - Ordering::Equal if incoming == current => Ok(WorkerStateSnapshotApply::Duplicate), - Ordering::Equal => Err(WorkerStateSnapshotConflict { - execution_generation: incoming.execution_generation, - revision: incoming.revision, - }), - } -} - impl From for WorkerStateSnapshot { fn from(status: WorkerStatus) -> Self { let state = match status { @@ -261,8 +203,6 @@ impl From for WorkerStateSnapshot { WorkerStatus::Paused => WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Paused)), }; Self { - execution_generation: 1, - revision: 0, last_command_id: 0, state, } @@ -1697,55 +1637,24 @@ mod tests { } #[test] - fn worker_state_snapshot_apply_is_monotonic_and_detects_conflicts() { - let mut current = WorkerStateSnapshot::initial(4); - let mut newer = current.clone(); - newer.revision = 1; - newer.state = WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running)); - - assert_eq!( - apply_worker_state_snapshot(&mut current, &newer), - Ok(WorkerStateSnapshotApply::Applied) - ); - assert_eq!( - apply_worker_state_snapshot(&mut current, &newer), - Ok(WorkerStateSnapshotApply::Duplicate) - ); - - let stale_revision = WorkerStateSnapshot::initial(4); - assert_eq!( - apply_worker_state_snapshot(&mut current, &stale_revision), - Ok(WorkerStateSnapshotApply::Stale) - ); - let stale_generation = WorkerStateSnapshot { - execution_generation: 3, - revision: u64::MAX, - ..newer.clone() + fn worker_state_snapshot_wire_shape_has_one_authoritative_state() { + let snapshot = WorkerStateSnapshot { + last_command_id: 7, + state: WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running)), }; + let value = serde_json::to_value(&snapshot).unwrap(); assert_eq!( - apply_worker_state_snapshot(&mut current, &stale_generation), - Ok(WorkerStateSnapshotApply::Stale) - ); - - let conflicting = WorkerStateSnapshot { - state: WorkerState::Idle, - ..newer.clone() - }; - assert_eq!( - apply_worker_state_snapshot(&mut current, &conflicting), - Err(WorkerStateSnapshotConflict { - execution_generation: 4, - revision: 1, + value, + serde_json::json!({ + "last_command_id": 7, + "state": { + "kind": "busy", + "state": { "kind": "run", "state": "running" } + } }) ); - assert_eq!(current, newer); - - let next_generation = WorkerStateSnapshot::initial(5); - assert_eq!( - apply_worker_state_snapshot(&mut current, &next_generation), - Ok(WorkerStateSnapshotApply::Applied) - ); - assert_eq!(current, next_generation); + assert!(value.get("execution_generation").is_none()); + assert!(value.get("revision").is_none()); } #[test] @@ -1942,28 +1851,21 @@ mod tests { } #[test] - fn lifecycle_methods_roundtrip_with_fences() { + fn lifecycle_methods_roundtrip_with_command_identity() { for method in [ Method::Pause { - command: WorkerCommandEnvelope { - command_id: 11, - expected_execution_generation: 4, - expected_worker_state_revision: 8, - }, + command: WorkerCommandEnvelope { command_id: 11 }, }, Method::Compact { - command: WorkerCommandEnvelope { - command_id: 12, - expected_execution_generation: 4, - expected_worker_state_revision: 9, - }, + command: WorkerCommandEnvelope { command_id: 12 }, }, ] { let json = serde_json::to_string(&method).unwrap(); + assert!(!json.contains("expected_execution_generation")); + assert!(!json.contains("expected_worker_state_revision")); let decoded: Method = serde_json::from_str(&json).unwrap(); match decoded { Method::Pause { command } | Method::Compact { command } => { - assert_eq!(command.expected_execution_generation, 4); assert!(command.command_id >= 11); } other => panic!("unexpected lifecycle method: {other:?}"), @@ -2260,7 +2162,7 @@ mod tests { #[test] fn event_snapshot_in_flight_roundtrip_and_default() { - let inbound = r#"{"event":"snapshot","data":{"session":{"entries":[]},"greeting":{"worker_name":"test","cwd":"/tmp","provider":"p","model":"m","scope_summary":"s","tools":[]},"state":{"execution_generation":1,"revision":1,"last_command_id":0,"state":{"kind":"busy","state":{"kind":"run","state":"running"}}}}}"#; + let inbound = r#"{"event":"snapshot","data":{"session":{"entries":[]},"greeting":{"worker_name":"test","cwd":"/tmp","provider":"p","model":"m","scope_summary":"s","tools":[]},"state":{"last_command_id":0,"state":{"kind":"busy","state":{"kind":"run","state":"running"}}}}}"#; let decoded: Event = serde_json::from_str(inbound).unwrap(); match decoded { Event::Snapshot { in_flight, .. } => assert!(in_flight.is_empty()), @@ -2404,8 +2306,6 @@ mod tests { fn event_worker_state_format() { let event = Event::WorkerState { snapshot: WorkerStateSnapshot { - execution_generation: 7, - revision: 3, last_command_id: 9, state: WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running)), }, @@ -2413,8 +2313,12 @@ mod tests { let json = serde_json::to_string(&event).unwrap(); let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); assert_eq!(parsed["event"], "worker_state"); - assert_eq!(parsed["data"]["snapshot"]["execution_generation"], 7); - assert_eq!(parsed["data"]["snapshot"]["revision"], 3); + assert!( + parsed["data"]["snapshot"] + .get("execution_generation") + .is_none() + ); + assert!(parsed["data"]["snapshot"].get("revision").is_none()); assert_eq!(parsed["data"]["snapshot"]["state"]["kind"], "busy"); let decoded: Event = serde_json::from_str(&json).unwrap(); @@ -2422,10 +2326,8 @@ mod tests { decoded, Event::WorkerState { snapshot: WorkerStateSnapshot { - execution_generation: 7, - revision: 3, + last_command_id: 9, state: WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running)), - .. } } )); @@ -2885,8 +2787,6 @@ mod tests { "tools": [] }, "state": { - "execution_generation": 1, - "revision": 0, "last_command_id": 0, "state": { "kind": "idle" } } diff --git a/crates/standalone/src/host.rs b/crates/standalone/src/host.rs index 1131bccd..c7e6e754 100644 --- a/crates/standalone/src/host.rs +++ b/crates/standalone/src/host.rs @@ -318,10 +318,7 @@ impl StandaloneHost { } pub async fn shutdown(mut self) -> Result<(), StandaloneShutdownError> { - let command = protocol::WorkerCommandEnvelope::for_snapshot( - u64::MAX, - &self.handle.shared_state.snapshot(), - ); + let command = protocol::WorkerCommandEnvelope::new(u64::MAX); let _ = self.handle.send(Method::Shutdown { command }).await; let Some(shutdown) = self.shutdown.take() else { self.retain_lease(); @@ -504,10 +501,7 @@ fn active_pointer( } async fn stop_started_worker(started: BootstrappedWorker) { - let command = protocol::WorkerCommandEnvelope::for_snapshot( - u64::MAX, - &started.handle.shared_state.snapshot(), - ); + let command = protocol::WorkerCommandEnvelope::new(u64::MAX); let _ = started.handle.send(Method::Shutdown { command }).await; let _ = tokio::time::timeout(Duration::from_secs(2), started.shutdown).await; } diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index b7900da5..f9d4ba23 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -342,7 +342,7 @@ impl App { Self { worker_name, connected: false, - worker_state: WorkerStateSnapshot::initial(1), + worker_state: WorkerStateSnapshot::initial(), next_command_id: 1, worker_status: WorkerStatus::Idle, running: false, @@ -1128,25 +1128,14 @@ impl App { let command_id = self .next_command_id .max(self.worker_state.last_command_id.saturating_add(1)); - let command = WorkerCommandEnvelope::for_snapshot(command_id, &self.worker_state); + let command = WorkerCommandEnvelope::new(command_id); self.next_command_id = command_id.saturating_add(1); command } fn apply_worker_state_snapshot(&mut self, snapshot: &WorkerStateSnapshot) { - match protocol::apply_worker_state_snapshot(&mut self.worker_state, snapshot) { - Ok(protocol::WorkerStateSnapshotApply::Applied) => { - self.set_worker_status(self.worker_state.catalog_status()); - } - Ok( - protocol::WorkerStateSnapshotApply::Duplicate - | protocol::WorkerStateSnapshotApply::Stale, - ) => {} - Err(error) => self.handle_error( - ErrorCode::Internal, - format!("worker state stream rejected: {error}"), - ), - } + self.worker_state = snapshot.clone(); + self.set_worker_status(self.worker_state.catalog_status()); } pub fn handle_worker_event(&mut self, event: Event) -> Option { @@ -3651,8 +3640,6 @@ mod completion_flow_tests { assert_eq!(app.worker_status, WorkerStatus::Idle); let running = WorkerStateSnapshot { - execution_generation: 1, - revision: 1, state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Running, )), @@ -3669,11 +3656,9 @@ mod completion_flow_tests { } #[test] - fn worker_state_events_and_acknowledgements_share_monotonic_application() { + fn worker_state_events_and_acknowledgements_replace_full_state() { let mut app = App::new("test".into()); let running = WorkerStateSnapshot { - execution_generation: 4, - revision: 3, state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Running, )), @@ -3682,22 +3667,22 @@ mod completion_flow_tests { app.handle_worker_event(Event::WorkerState { snapshot: running.clone(), }); - app.handle_worker_event(Event::WorkerState { - snapshot: WorkerStateSnapshot { - revision: 2, - state: protocol::WorkerState::Idle, - ..running.clone() - }, - }); assert_eq!(app.worker_state, running); + let fresh_idle = WorkerStateSnapshot { + state: protocol::WorkerState::Idle, + last_command_id: 0, + }; + app.handle_worker_event(Event::WorkerState { + snapshot: fresh_idle.clone(), + }); + assert_eq!(app.worker_state, fresh_idle); + let paused = WorkerStateSnapshot { - revision: 4, state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Paused, )), last_command_id: 3, - ..running.clone() }; app.handle_worker_event(Event::CommandAcknowledged { acknowledgement: protocol::WorkerCommandAcknowledgement { @@ -3708,17 +3693,6 @@ mod completion_flow_tests { }, }); assert_eq!(app.worker_state, paused); - - app.handle_worker_event(Event::WorkerState { - snapshot: WorkerStateSnapshot { - state: protocol::WorkerState::Idle, - ..paused.clone() - }, - }); - assert_eq!(app.worker_state, paused); - assert!(app.run_error_messages.iter().any(|message| { - message.contains("conflicting worker state snapshots at generation 4 revision 4") - })); } #[test] @@ -4340,7 +4314,7 @@ mod completion_flow_tests { fn snapshot_restores_and_runtime_clear_removes_compaction_progress() { let mut app = App::new("test".into()); assert_eq!(app.worker_state.state, protocol::WorkerState::Idle); - let mut state = protocol::WorkerStateSnapshot::initial(2); + let mut state = protocol::WorkerStateSnapshot::initial(); state.state = protocol::WorkerState::Busy(protocol::WorkerBusyState::Maintenance( protocol::WorkerMaintenanceState::Compacting, )); @@ -4395,10 +4369,7 @@ mod completion_flow_tests { } fn test_worker_state(status: WorkerStatus) -> WorkerStateSnapshot { - let mut snapshot = WorkerStateSnapshot::from(status); - snapshot.execution_generation = 1; - snapshot.revision = 1; - snapshot + WorkerStateSnapshot::from(status) } fn test_greeting() -> protocol::Greeting { diff --git a/crates/tui/src/backend_worker_picker.rs b/crates/tui/src/backend_worker_picker.rs index 41302821..ac7b2a7b 100644 --- a/crates/tui/src/backend_worker_picker.rs +++ b/crates/tui/src/backend_worker_picker.rs @@ -459,8 +459,6 @@ mod tests { }, state: "idle".to_string(), worker_state: Some(protocol::WorkerStateSnapshot { - execution_generation: 1, - revision: 1, state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Running, )), @@ -593,8 +591,6 @@ mod tests { short.display_name = short.label.clone(); short.state = "idle".to_string(); short.worker_state = Some(protocol::WorkerStateSnapshot { - execution_generation: 1, - revision: 2, state: protocol::WorkerState::Idle, last_command_id: 0, }); diff --git a/crates/tui/src/command.rs b/crates/tui/src/command.rs index ba24299e..26e6a40d 100644 --- a/crates/tui/src/command.rs +++ b/crates/tui/src/command.rs @@ -410,10 +410,7 @@ fn compact_command(invocation: CommandInvocation<'_>) -> CommandExecution { let _ = invocation.args.raw(); CommandExecution { method: Some(Method::Compact { - command: protocol::WorkerCommandEnvelope::for_snapshot( - 0, - &protocol::WorkerStateSnapshot::initial(1), - ), + command: protocol::WorkerCommandEnvelope::new(0), }), diagnostics: vec![CommandDiagnostic::new("compact requested")], exit_command_mode: true, diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 91dda346..0b0aaae3 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -243,8 +243,6 @@ impl fmt::Debug for WorkerExecutionContext { #[derive(Clone, Debug)] pub struct WorkerExecutionSpawnRequest { pub worker_ref: WorkerRef, - /// Monotonic execution generation reserved durably before launch. - pub run_generation: u64, pub request: crate::catalog::CreateWorkerRequest, pub workspace_scope: Option, pub context: WorkerExecutionContext, @@ -256,8 +254,6 @@ pub struct WorkerExecutionSpawnRequest { #[derive(Clone, Debug)] pub struct WorkerExecutionRestoreRequest { pub worker_ref: WorkerRef, - /// Monotonic execution generation reserved durably before restore. - pub run_generation: u64, pub request: crate::catalog::CreateWorkerRequest, pub workspace_scope: Option, pub context: WorkerExecutionContext, diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index a2706ba4..dd929a31 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -17,10 +17,8 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; -const SCHEMA_VERSION: u32 = 6; -const PREVIOUS_SCHEMA_VERSION: u32 = 5; -const EXECUTION_SCHEMA_VERSION: u32 = 4; -const PRE_EXECUTION_SCHEMA_VERSION: u32 = 3; +const SCHEMA_VERSION: u32 = 7; +const PREVIOUS_SCHEMA_VERSION: u32 = 6; const RUNTIME_FILE: &str = "runtime.json"; const WORKERS_DIR: &str = "workers"; const WORKER_FILE: &str = "worker.json"; @@ -371,13 +369,12 @@ pub(crate) struct PersistedRuntimeState { } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub(crate) struct PersistedWorkerExecutionBinding { - pub(crate) run_generation: u64, -} +#[serde(deny_unknown_fields)] +pub(crate) struct PersistedWorkerExecutionBinding {} #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub(crate) struct PersistedWorkerExecution { - pub(crate) last_run_generation: u64, pub(crate) binding: Option, pub(crate) restore_intent: WorkerRestoreIntent, } @@ -465,8 +462,8 @@ fn plan_runtime_store_migration( format!("Runtime store schema version {schema_version} is out of range"), ) })?; - let staging = migration_sibling(root, "schema-v6-staging")?; - let backup = migration_sibling(root, "pre-schema-v6-backup")?; + let staging = migration_sibling(root, "schema-v7-staging")?; + let backup = migration_sibling(root, "pre-schema-v7-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -492,14 +489,11 @@ fn plan_runtime_store_migration( }; return Ok((plan, Vec::new())); } - if !matches!( - current_schema_version, - PRE_EXECUTION_SCHEMA_VERSION | EXECUTION_SCHEMA_VERSION | PREVIOUS_SCHEMA_VERSION - ) { + if current_schema_version != PREVIOUS_SCHEMA_VERSION { return Err(runtime_store_corrupt( &runtime_path, format!( - "unsupported Runtime store schema version {schema_version}; expected {PRE_EXECUTION_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, or {SCHEMA_VERSION}" + "unsupported Runtime store schema version {schema_version}; expected {PREVIOUS_SCHEMA_VERSION} or {SCHEMA_VERSION}" ), )); } @@ -661,118 +655,19 @@ struct DiagnosticWorkerRefMigrationCounts { cleared: usize, } -fn migrate_v1_worker_document( - mut snapshot: serde_json::Value, - mapping: &LegacyWorkerIdentityMapping, - snapshot_path: &Path, -) -> Result { - let worker_id_text = mapping.worker_id.to_string(); - let snapshot_object = snapshot.as_object_mut().ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "Worker snapshot must be an object".to_string(), - ) - })?; - snapshot_object.insert( - "schema_version".to_string(), - serde_json::Value::from(SCHEMA_VERSION), - ); - snapshot_object.insert( - "worker_id".to_string(), - serde_json::Value::String(worker_id_text.clone()), - ); - snapshot_object - .get_mut("worker_ref") - .and_then(serde_json::Value::as_object_mut) - .ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "Worker snapshot worker_ref must be an object".to_string(), - ) - })? - .insert( - "worker_id".to_string(), - serde_json::Value::String(worker_id_text.clone()), - ); - let request = snapshot_object - .get_mut("request") - .and_then(serde_json::Value::as_object_mut) - .ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "Worker snapshot request must be an object".to_string(), - ) - })?; - let fingerprint = request - .remove("idempotency_fingerprint") - .and_then(|value| value.as_str().map(ToOwned::to_owned)) - .unwrap_or_else(|| { - format!( - "legacy:{}:{}:{}", - mapping.workspace_id, mapping.runtime_id, mapping.legacy_worker_id - ) - }); - request.remove("idempotency_key"); - request.insert( - "worker_id".to_string(), - serde_json::Value::String(worker_id_text), - ); - request.insert( - "create_fingerprint".to_string(), - serde_json::Value::String(fingerprint), - ); - Ok(snapshot) -} - -fn max_persisted_run_generation(snapshot_path: &Path) -> Result { - let worker_dir = snapshot_path.parent().ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "Worker snapshot path is missing its aggregate directory".to_string(), - ) - })?; - let runs_dir = worker_dir.join("runs"); - if !runs_dir - .try_exists() - .map_err(|source| runtime_io_error("inspect Worker runs", &runs_dir, source))? - { - return Ok(0); - } - let entries = fs::read_dir(&runs_dir) - .map_err(|source| runtime_io_error("read Worker runs", &runs_dir, source))?; - let mut max_generation = 0; - for entry in entries { - let entry = - entry.map_err(|source| runtime_io_error("read Worker runs", &runs_dir, source))?; - let Some(generation) = entry - .file_name() - .to_str() - .and_then(|name| name.parse::().ok()) - else { - continue; - }; - max_generation = max_generation.max(generation); - } - Ok(max_generation) -} - fn migrate_worker_document( mut document: serde_json::Value, source_schema_version: u32, - mapping: Option<&LegacyWorkerIdentityMapping>, + _mapping: Option<&LegacyWorkerIdentityMapping>, snapshot_path: &Path, ) -> Result { - if source_schema_version == 1 { - document = migrate_v1_worker_document( - document, - mapping.ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "schema-v1 Worker migration is missing its identity mapping".to_string(), - ) - })?, + if source_schema_version != PREVIOUS_SCHEMA_VERSION { + return Err(runtime_store_corrupt( snapshot_path, - )?; + format!( + "unsupported Worker snapshot schema {source_schema_version}; expected {PREVIOUS_SCHEMA_VERSION}" + ), + )); } let object = document.as_object_mut().ok_or_else(|| { runtime_store_corrupt( @@ -780,117 +675,80 @@ fn migrate_worker_document( "Worker snapshot must be an object".to_string(), ) })?; - let declared_run_generation = object - .remove("run_generation") - .map(|value| { - value.as_u64().ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "Worker snapshot run_generation must be an unsigned integer".to_string(), - ) - }) - }) - .transpose()?; - let legacy_execution = object.remove("execution"); - let execution = legacy_execution - .as_ref() - .and_then(serde_json::Value::as_object); - let persisted_last_run_generation = execution - .and_then(|execution| execution.get("last_run_generation")) - .map(|value| { - value.as_u64().ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "Worker execution last_run_generation must be an unsigned integer".to_string(), - ) - }) - }) - .transpose()?; - let binding_run_generation = execution - .and_then(|execution| execution.get("binding")) - .and_then(serde_json::Value::as_object) - .and_then(|binding| binding.get("run_generation")) - .map(|value| { - value.as_u64().ok_or_else(|| { + let execution = object + .get_mut("execution") + .and_then(serde_json::Value::as_object_mut) + .ok_or_else(|| { + runtime_store_corrupt( + snapshot_path, + "Worker snapshot execution must be an object".to_string(), + ) + })?; + let last_run_generation = execution + .remove("last_run_generation") + .and_then(|value| value.as_u64()) + .ok_or_else(|| { + runtime_store_corrupt( + snapshot_path, + "Worker execution last_run_generation must be an unsigned integer".to_string(), + ) + })?; + let binding = execution.get_mut("binding").ok_or_else(|| { + runtime_store_corrupt( + snapshot_path, + "Worker execution is missing binding".to_string(), + ) + })?; + if let Some(binding_object) = binding.as_object_mut() { + let binding_run_generation = binding_object + .remove("run_generation") + .and_then(|value| value.as_u64()) + .ok_or_else(|| { runtime_store_corrupt( snapshot_path, "Worker execution binding run_generation must be an unsigned integer" .to_string(), ) - }) - }) - .transpose()?; - let run_generation = declared_run_generation - .into_iter() - .chain(persisted_last_run_generation) - .chain(binding_run_generation) - .chain(std::iter::once(max_persisted_run_generation( - snapshot_path, - )?)) - .max() - .unwrap_or(0); - if !object.contains_key("working_directory") { - if let Some(working_directory) = legacy_execution - .as_ref() - .and_then(serde_json::Value::as_object) - .and_then(|execution| execution.get("working_directory")) - .cloned() - { - object.insert("working_directory".to_string(), working_directory); - } - } - let legacy_materialization = object - .get("working_directory") - .and_then(|working_directory| working_directory.get("summary")) - .and_then(|summary| summary.get("materializer_kind")) - .and_then(serde_json::Value::as_str) - .is_some_and(|kind| matches!(kind, "runtime_git_cache" | "local_git_worktree")); - if legacy_materialization { - object.insert("working_directory".to_string(), serde_json::Value::Null); - } - if let Some(profile_source) = object - .get_mut("request") - .and_then(serde_json::Value::as_object_mut) - .and_then(|request| request.get_mut("profile_source")) - .and_then(serde_json::Value::as_object_mut) - && profile_source - .get("kind") - .and_then(serde_json::Value::as_str) - == Some("http") - { - let archive = profile_source - .get_mut("location") - .and_then(serde_json::Value::as_object_mut) - .and_then(|location| location.remove("archive")) - .ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "legacy HTTP profile source is missing its archive".to_string(), - ) })?; - profile_source.clear(); - profile_source.insert( - "kind".to_string(), - serde_json::Value::String("workspace_config".to_string()), - ); - profile_source.insert("archive".to_string(), archive); + if binding_run_generation != last_run_generation { + return Err(runtime_store_corrupt( + snapshot_path, + format!( + "execution binding run_generation {binding_run_generation} does not match last_run_generation {last_run_generation}" + ), + )); + } + if !binding_object.is_empty() { + return Err(runtime_store_corrupt( + snapshot_path, + "Worker execution binding contains unsupported fields".to_string(), + )); + } + } else if !binding.is_null() { + return Err(runtime_store_corrupt( + snapshot_path, + "Worker execution binding must be an object or null".to_string(), + )); + } + if !execution.contains_key("restore_intent") { + return Err(runtime_store_corrupt( + snapshot_path, + "Worker execution is missing restore_intent".to_string(), + )); + } + if execution + .keys() + .any(|key| key != "binding" && key != "restore_intent") + { + return Err(runtime_store_corrupt( + snapshot_path, + "Worker execution contains unsupported fields".to_string(), + )); } object.insert( "schema_version".to_string(), serde_json::Value::from(SCHEMA_VERSION), ); - object.insert( - "status".to_string(), - serde_json::Value::String("stopped".to_string()), - ); - object.insert( - "execution".to_string(), - serde_json::json!({ - "last_run_generation": run_generation, - "binding": null, - "restore_intent": "explicit", - }), - ); Ok(document) } @@ -1259,8 +1117,8 @@ fn migrate_runtime_store( if !plan.migration_required { return Ok(plan); } - let staging = migration_sibling(root, "schema-v6-staging")?; - let backup = migration_sibling(root, "pre-schema-v6-backup")?; + let staging = migration_sibling(root, "schema-v7-staging")?; + let backup = migration_sibling(root, "pre-schema-v7-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -1534,52 +1392,18 @@ impl WorkerSnapshot { ), }); } - if let Some(binding) = self.execution.binding.as_ref() - && binding.run_generation != self.execution.last_run_generation - { - return Err(RuntimeError::StoreCorrupt { - operation: "read worker snapshot", - path: path.to_path_buf(), - message: format!( - "execution binding run_generation {} does not match last_run_generation {}", - binding.run_generation, self.execution.last_run_generation - ), - }); - } match (self.status, self.execution.restore_intent) { (status, WorkerRestoreIntent::Automatic) if status.is_active() => { - let Some(binding) = self.execution.binding.as_ref() else { + if self.execution.binding.is_none() { return Err(RuntimeError::StoreCorrupt { operation: "read worker snapshot", path: path.to_path_buf(), message: "automatic restore intent requires an execution binding" .to_string(), }); - }; - if binding.run_generation == 0 { - return Err(RuntimeError::StoreCorrupt { - operation: "read worker snapshot", - path: path.to_path_buf(), - message: "execution binding run_generation must be greater than zero" - .to_string(), - }); - } - } - (WorkerStatus::Stopped, WorkerRestoreIntent::Explicit) => { - if self - .execution - .binding - .as_ref() - .is_some_and(|binding| binding.run_generation == 0) - { - return Err(RuntimeError::StoreCorrupt { - operation: "read worker snapshot", - path: path.to_path_buf(), - message: "execution binding run_generation must be greater than zero" - .to_string(), - }); } } + (WorkerStatus::Stopped, WorkerRestoreIntent::Explicit) => {} _ => { return Err(RuntimeError::StoreCorrupt { operation: "read worker snapshot", @@ -1837,96 +1661,47 @@ mod tests { } #[test] - fn schema_v5_worker_migration_recovers_last_generation_from_run_aggregates() { - let root = tempfile::tempdir().unwrap(); - let worker_dir = root.path().join("worker-a"); - fs::create_dir_all(worker_dir.join("runs/1")).unwrap(); - fs::create_dir_all(worker_dir.join("runs/7")).unwrap(); - fs::create_dir_all(worker_dir.join("runs/incomplete")).unwrap(); - let path = worker_dir.join(WORKER_FILE); - let source = serde_json::json!({ - "schema_version": 5, - "execution": { - "binding": null, - "restore_intent": "explicit" - } - }); - - let migrated = - migrate_worker_document(source, PREVIOUS_SCHEMA_VERSION, None, &path).unwrap(); - - assert_eq!( - migrated["execution"]["last_run_generation"], - serde_json::json!(7) - ); - assert_eq!(migrated["execution"]["binding"], serde_json::Value::Null); - } - - #[test] - fn schema_v4_worker_migration_discards_unsupported_linked_worktree_binding() { - let source = serde_json::json!({ - "schema_version": 4, - "request": { - "profile_source": { - "kind": "http", - "location": { - "url": "https://workspace.example.test/archive", - "etag": "profile-source:test", - "archive": { - "id": "profiles-v1", - "digest": "sha256:test", - "size_bytes": 1, - "source_graph": { - "source_count": 1, - "total_source_bytes": 1, - "entrypoints": {}, - "import_count": 0 - } - } - } - } - }, - "working_directory": { - "summary": { - "materializer_kind": "runtime_git_cache" - } - } - }); + fn schema_v6_worker_migration_removes_generation_and_preserves_active_restore() { let path = Path::new("worker.json"); + let source = serde_json::json!({ + "schema_version": PREVIOUS_SCHEMA_VERSION, + "status": "running", + "execution": { + "last_run_generation": 7, + "binding": { "run_generation": 7 }, + "restore_intent": "automatic" + } + }); let migrated = - migrate_worker_document(source, EXECUTION_SCHEMA_VERSION, None, path).unwrap(); + migrate_worker_document(source, PREVIOUS_SCHEMA_VERSION, None, path).unwrap(); assert_eq!(migrated["schema_version"], SCHEMA_VERSION); - assert_eq!(migrated["status"], "stopped"); - assert_eq!(migrated["working_directory"], serde_json::Value::Null); - assert_eq!( - migrated["request"]["profile_source"]["kind"], - "workspace_config" - ); - assert_eq!( - migrated["request"]["profile_source"]["archive"]["id"], - "profiles-v1" - ); - assert_eq!(migrated["execution"]["restore_intent"], "explicit"); + assert_eq!(migrated["status"], "running"); + assert_eq!(migrated["execution"]["binding"], serde_json::json!({})); + assert_eq!(migrated["execution"]["restore_intent"], "automatic"); + assert!(migrated["execution"].get("last_run_generation").is_none()); } #[test] - fn schema_v4_worker_migration_preserves_runtime_clone_observation() { + fn schema_v6_worker_migration_rejects_mismatched_generation_state() { + let path = Path::new("worker.json"); let source = serde_json::json!({ - "schema_version": 4, - "working_directory": { - "summary": { - "materializer_kind": "runtime_git_clone" - } + "schema_version": PREVIOUS_SCHEMA_VERSION, + "execution": { + "last_run_generation": 7, + "binding": { "run_generation": 6 }, + "restore_intent": "automatic" } }); - let expected = source["working_directory"].clone(); - let path = Path::new("worker.json"); - let migrated = - migrate_worker_document(source, EXECUTION_SCHEMA_VERSION, None, path).unwrap(); + let error = + migrate_worker_document(source, PREVIOUS_SCHEMA_VERSION, None, path).unwrap_err(); - assert_eq!(migrated["working_directory"], expected); + assert!( + error + .to_string() + .contains("does not match last_run_generation") + ); } } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index b067bd68..e1be998c 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -3126,7 +3126,6 @@ mod tests { WorkerExecutionSpawnResult::Connected { handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()), worker_state: protocol::WorkerStateSnapshot { - execution_generation: request.run_generation, ..protocol::WorkerStatus::Idle.into() }, working_directory: request @@ -3143,7 +3142,6 @@ mod tests { WorkerExecutionSpawnResult::Connected { handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()), worker_state: protocol::WorkerStateSnapshot { - execution_generation: request.run_generation, ..protocol::WorkerStatus::Idle.into() }, working_directory: request.previous_working_directory, @@ -3561,7 +3559,6 @@ mod ws_tests { WorkerExecutionSpawnResult::Connected { handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()), worker_state: protocol::WorkerStateSnapshot { - execution_generation: request.run_generation, ..protocol::WorkerStatus::Idle.into() }, working_directory: request @@ -3604,7 +3601,7 @@ mod ws_tests { context_window: 0, context_tokens: 0, }, - state: protocol::WorkerStateSnapshot::initial(1), + state: protocol::WorkerStateSnapshot::initial(), in_flight: protocol::InFlightSnapshot { blocks: Vec::new(), commands: Vec::new(), @@ -3833,7 +3830,6 @@ mod ws_tests { .unwrap() .worker_state .expect("connected test Worker must expose its initial state"); - snapshot.revision += 1; snapshot.state = protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Running, )); diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index 95cd115f..1c498d02 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -1273,14 +1273,14 @@ mod tests { } #[test] - fn migration_dry_run_accepts_supported_schema_v3_without_workers_field() { + fn migration_dry_run_accepts_previous_schema_without_workers_field() { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime"); std::fs::create_dir_all(root.join("workers")).unwrap(); std::fs::write( root.join("runtime.json"), serde_json::to_vec_pretty(&serde_json::json!({ - "schema_version": 3, + "schema_version": 6, "display_name": "local", "backend": "fs_store", "status": "running", @@ -1312,14 +1312,14 @@ mod tests { } #[test] - fn migration_dry_run_rejects_schema_v3_document_that_cannot_decode_as_v6() { + fn migration_dry_run_rejects_previous_schema_that_cannot_decode_as_current() { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime"); std::fs::create_dir_all(root.join("workers")).unwrap(); std::fs::write( root.join("runtime.json"), serde_json::to_vec_pretty(&serde_json::json!({ - "schema_version": 3, + "schema_version": 6, "display_name": "local", "backend": "fs_store", "status": 3, diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index 91ecd710..0d02fb3e 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -39,7 +39,6 @@ pub struct WorkerRetentionInventory { pub workspace_id: String, pub runtime_id: String, pub worker_id: WorkerId, - pub run_generation: u64, pub session_id: Option, pub segment_ids: Vec, pub session_bytes: u64, @@ -118,7 +117,6 @@ pub struct WorkerRetentionExecutionRequest { pub source_runtime_id: String, pub worker_id: WorkerId, pub expected_worker_revision: String, - pub expected_run_generation: u64, pub source_created_at: String, pub removed_at: String, pub effective_profile: Option, @@ -171,7 +169,6 @@ pub(crate) trait WorkerRetentionProvider: Send + Sync { workspace_id: &str, runtime_id: &str, worker_id: WorkerId, - run_generation: u64, ) -> Result; fn execute( @@ -283,7 +280,7 @@ impl FsWorkerRetentionProvider { continue; }; let worker_dir = self.worker_dir(worker_id); - let snapshot: WorkerGenerationSnapshot = match read_json( + let snapshot: WorkerAggregateSnapshot = match read_json( &worker_dir.join("worker.json"), "scan Worker retention inventory", ) { @@ -303,12 +300,7 @@ impl FsWorkerRetentionProvider { )); continue; } - match self.inventory( - workspace_id, - runtime_id, - worker_id, - snapshot.run_generation(), - ) { + match self.inventory(workspace_id, runtime_id, worker_id) { Ok(item) => workers.push(item), Err(_) => diagnostics.push(runtime_aggregate_diagnostic( &bounded_id, @@ -380,26 +372,18 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { workspace_id: &str, runtime_id: &str, worker_id: WorkerId, - run_generation: u64, ) -> Result { let worker_dir = self.worker_dir(worker_id); if !worker_dir.is_dir() { return Err(RuntimeError::WorkerNotFound { worker_id }); } - let worker: WorkerGenerationSnapshot = read_json( + let worker: WorkerAggregateSnapshot = read_json( &worker_dir.join("worker.json"), "inventory Worker retention", )?; if worker.workspace_id.as_deref() != Some(workspace_id) { return Err(RuntimeError::WorkerNotFound { worker_id }); } - let current_run_generation = worker.run_generation(); - if current_run_generation != run_generation { - return Err(RuntimeError::InvalidRequest(format!( - "Worker retention inventory expected generation {run_generation}, current generation is {}", - current_run_generation - ))); - } let session_dir = worker_dir.join("session"); let (session_id, segment_ids, session_bytes) = if session_dir.is_dir() { let manifest: CanonicalSessionManifest = read_json( @@ -437,7 +421,6 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { workspace_id: workspace_id.to_string(), runtime_id: runtime_id.to_string(), worker_id, - run_generation, session_id, segment_ids, session_bytes, @@ -497,21 +480,13 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { worker_id: request.worker_id, }); } - let snapshot: WorkerGenerationSnapshot = + let snapshot: WorkerAggregateSnapshot = read_json(&worker_dir.join("worker.json"), "execute Worker retention")?; if snapshot.workspace_id.as_deref() != Some(request.workspace_id.as_str()) { return Err(RuntimeError::WorkerNotFound { worker_id: request.worker_id, }); } - let run_generation = snapshot.run_generation(); - if run_generation != request.expected_run_generation { - return Err(RuntimeError::InvalidRequest(format!( - "Worker retention plan expected generation {}, current generation is {}", - request.expected_run_generation, run_generation - ))); - } - let archive = match request.session_disposition { SessionDisposition::Archive => { Some(commit_session_archive(self, request, &worker_dir)?) @@ -572,30 +547,9 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { } #[derive(Deserialize)] -struct WorkerGenerationSnapshot { +struct WorkerAggregateSnapshot { #[serde(default)] workspace_id: Option, - execution: WorkerGenerationExecution, -} - -#[derive(Deserialize)] -struct WorkerGenerationExecution { - binding: Option, -} - -#[derive(Deserialize)] -struct WorkerGenerationBinding { - run_generation: u64, -} - -impl WorkerGenerationSnapshot { - fn run_generation(&self) -> u64 { - self.execution - .binding - .as_ref() - .map(|binding| binding.run_generation) - .unwrap_or(0) - } } #[derive(Deserialize)] @@ -1286,13 +1240,12 @@ mod tests { fs::write(path, serde_json::to_vec_pretty(value).unwrap()).unwrap(); } - fn source(root: &Path, worker_id: WorkerId, generation: u64) { + fn source(root: &Path, worker_id: WorkerId) { let worker = root.join("workers").join(worker_id.to_string()); write_json( &worker.join("worker.json"), &serde_json::json!({ - "workspace_id": "workspace-a", - "execution": {"binding": {"run_generation": generation}} + "workspace_id": "workspace-a" }), ); write_json( @@ -1301,22 +1254,17 @@ mod tests { ); fs::create_dir_all(worker.join("session/segments")).unwrap(); fs::write(worker.join("session/segments/segment-a.jsonl"), b"one\n").unwrap(); - fs::create_dir_all(worker.join(format!("runs/{generation}"))).unwrap(); + fs::create_dir_all(worker.join("runs/attempt-a")).unwrap(); fs::write( - worker.join(format!("runs/{generation}/worker.out.log")), + worker.join("runs/attempt-a/worker.out.log"), b"diagnostic\n", ) .unwrap(); - fs::write( - worker.join(format!("runs/{generation}/worker.sock")), - b"not retained", - ) - .unwrap(); + fs::write(worker.join("runs/attempt-a/worker.sock"), b"not retained").unwrap(); } fn request( worker_id: WorkerId, - generation: u64, disposition: SessionDisposition, ) -> WorkerRetentionExecutionRequest { WorkerRetentionExecutionRequest { @@ -1328,7 +1276,6 @@ mod tests { workspace_id: "workspace-a".to_string(), source_runtime_id: "runtime-a".to_string(), worker_id, - expected_run_generation: generation, source_created_at: "2026-01-01T00:00:00Z".to_string(), removed_at: "2026-01-02T00:00:00Z".to_string(), effective_profile: Some("builtin:coder".to_string()), @@ -1344,9 +1291,9 @@ mod tests { fn archive_is_verified_before_source_removal_and_retry_converges() { let temp = tempfile::tempdir().unwrap(); let worker_id = WorkerId::from_legacy_u64(7); - source(temp.path(), worker_id, 4); + source(temp.path(), worker_id); let provider = FsWorkerRetentionProvider::new(temp.path()); - let request = request(worker_id, 4, SessionDisposition::Archive); + let request = request(worker_id, SessionDisposition::Archive); let first = provider.execute(&request).unwrap(); assert!(first.source_removed); @@ -1375,7 +1322,7 @@ mod tests { fn archive_failure_keeps_live_source_for_retry() { let temp = tempfile::tempdir().unwrap(); let worker_id = WorkerId::from_legacy_u64(8); - source(temp.path(), worker_id, 2); + source(temp.path(), worker_id); let collision = temp.path().join("archives/workers/archive-a"); fs::create_dir_all(&collision).unwrap(); fs::write(collision.join("manifest.json"), b"not-json").unwrap(); @@ -1383,7 +1330,7 @@ mod tests { assert!( provider - .execute(&request(worker_id, 2, SessionDisposition::Archive)) + .execute(&request(worker_id, SessionDisposition::Archive)) .is_err() ); assert!( @@ -1403,13 +1350,13 @@ mod tests { fn target_inventory_and_execute_reject_cross_workspace_aggregate() { let temp = tempfile::tempdir().unwrap(); let worker_id = WorkerId::from_legacy_u64(16); - source(temp.path(), worker_id, 3); + source(temp.path(), worker_id); let provider = FsWorkerRetentionProvider::new(temp.path()); assert!(matches!( - provider.inventory("other-workspace", "runtime-a", worker_id, 3), + provider.inventory("other-workspace", "runtime-a", worker_id), Err(RuntimeError::WorkerNotFound { .. }) )); - let mut request = request(worker_id, 3, SessionDisposition::Purge); + let mut request = request(worker_id, SessionDisposition::Purge); request.workspace_id = "other-workspace".to_string(); assert!(matches!( provider.execute(&request), @@ -1434,20 +1381,12 @@ mod tests { } #[test] - fn purge_removes_aggregate_and_rejects_stale_generation() { + fn purge_removes_worker_aggregate() { let temp = tempfile::tempdir().unwrap(); let provider = FsWorkerRetentionProvider::new(temp.path()); let worker_id = WorkerId::from_legacy_u64(9); - source(temp.path(), worker_id, 5); - let stale = request(worker_id, 4, SessionDisposition::Purge); - assert!(provider.execute(&stale).is_err()); - assert!( - temp.path() - .join(format!("workers/{worker_id}/session")) - .is_dir() - ); - - let mut current = request(worker_id, 5, SessionDisposition::Purge); + source(temp.path(), worker_id); + let mut current = request(worker_id, SessionDisposition::Purge); current.operation_id = "operation-current".to_string(); current.input_fingerprint = "fingerprint-current".to_string(); let result = provider.execute(¤t).unwrap(); @@ -1464,9 +1403,9 @@ mod tests { fn pending_receipt_recovers_delete_to_receipt_crash_window() { let temp = tempfile::tempdir().unwrap(); let worker_id = WorkerId::from_legacy_u64(11); - source(temp.path(), worker_id, 1); + source(temp.path(), worker_id); let provider = FsWorkerRetentionProvider::new(temp.path()); - let request = request(worker_id, 1, SessionDisposition::Archive); + let request = request(worker_id, SessionDisposition::Archive); let completed = provider.execute(&request).unwrap(); let receipt_path = temp.path().join("retention/operations/operation-a.json"); let mut receipt: RetentionOperationReceipt = @@ -1482,9 +1421,9 @@ mod tests { #[test] fn provider_snapshot_scans_aggregate_storage_independent_of_runtime_catalog() { let temp = tempfile::tempdir().unwrap(); - source(temp.path(), WorkerId::from_legacy_u64(13), 2); + source(temp.path(), WorkerId::from_legacy_u64(13)); let other_worker = WorkerId::from_legacy_u64(14); - source(temp.path(), other_worker, 1); + source(temp.path(), other_worker); write_json( &temp .path() @@ -1492,8 +1431,7 @@ mod tests { .join(other_worker.to_string()) .join("worker.json"), &serde_json::json!({ - "workspace_id": "other-workspace", - "execution": {"binding": {"run_generation": 1}} + "workspace_id": "other-workspace" }), ); fs::create_dir_all(temp.path().join("workers/not-a-worker")).unwrap(); @@ -1532,9 +1470,9 @@ mod tests { fn diagnostics_retry_rejects_corrupt_existing_archive_before_source_delete() { let temp = tempfile::tempdir().unwrap(); let worker_id = WorkerId::from_legacy_u64(12); - source(temp.path(), worker_id, 1); + source(temp.path(), worker_id); let provider = FsWorkerRetentionProvider::new(temp.path()); - let mut request = request(worker_id, 1, SessionDisposition::Archive); + let mut request = request(worker_id, SessionDisposition::Archive); request.diagnostics_disposition = DiagnosticsDisposition::Retain; provider.execute(&request).unwrap(); @@ -1543,10 +1481,10 @@ mod tests { serde_json::from_slice(&fs::read(&receipt_path).unwrap()).unwrap(); receipt.result.source_removed = false; fs::write(&receipt_path, serde_json::to_vec_pretty(&receipt).unwrap()).unwrap(); - source(temp.path(), worker_id, 1); + source(temp.path(), worker_id); fs::write( temp.path() - .join("archives/diagnostics/operation-a/runs/1/worker.out.log"), + .join("archives/diagnostics/operation-a/runs/attempt-a/worker.out.log"), b"corrupt\n", ) .unwrap(); @@ -1564,9 +1502,9 @@ mod tests { fn concurrent_retry_produces_one_archive() { let temp = tempfile::tempdir().unwrap(); let worker_id = WorkerId::from_legacy_u64(10); - source(temp.path(), worker_id, 1); + source(temp.path(), worker_id); let provider = Arc::new(FsWorkerRetentionProvider::new(temp.path())); - let request = Arc::new(request(worker_id, 1, SessionDisposition::Archive)); + let request = Arc::new(request(worker_id, SessionDisposition::Archive)); let barrier = Arc::new(Barrier::new(3)); let handles = (0..2) .map(|_| { diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index a8f32b28..428c2c37 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -987,7 +987,6 @@ impl Runtime { worker_state: None, workspace_id: scope.map(|scope| scope.workspace_id.clone()), request: durable_request, - run_generation: 1, execution_bound: true, restore_intent: WorkerRestoreIntent::Explicit, working_directory: None, @@ -999,7 +998,6 @@ impl Runtime { state.persist_worker(&worker_ref.worker_id)?; let spawn_request = WorkerExecutionSpawnRequest { worker_ref: worker_ref.clone(), - run_generation: 1, request, workspace_scope: scope.cloned(), context: self.execution_context(worker_ref.clone()), @@ -1361,7 +1359,7 @@ impl Runtime { let (backend, request) = { let mut state = self.lock()?; state.ensure_running()?; - let (worker_request, previous_working_directory, run_generation) = { + let (worker_request, previous_working_directory) = { let worker = state.worker(worker_ref)?; if worker.execution_handle.is_some() { if worker.status.is_active() { @@ -1387,11 +1385,7 @@ impl Runtime { } _ => {} } - ( - worker.request.clone(), - worker.working_directory.clone(), - worker.run_generation.saturating_add(1).max(1), - ) + (worker.request.clone(), worker.working_directory.clone()) }; let backend = state.execution_backend.clone().ok_or_else(|| { RuntimeError::WorkerExecutionUnavailable { @@ -1401,7 +1395,6 @@ impl Runtime { })?; { let worker = state.worker_mut(worker_ref)?; - worker.run_generation = run_generation; worker.execution_bound = true; } state.persist_worker(&worker_ref.worker_id)?; @@ -1413,7 +1406,6 @@ impl Runtime { }); let request = WorkerExecutionRestoreRequest { worker_ref: worker_ref.clone(), - run_generation, request: worker_request, workspace_scope, context: self.execution_context(worker_ref.clone()), @@ -1910,12 +1902,7 @@ impl Runtime { }; let mut state = self.lock()?; let worker = state.worker_mut(worker_ref)?; - let applied = worker - .apply_worker_state(&snapshot) - .is_ok_and(|result| matches!(result, protocol::WorkerStateSnapshotApply::Applied)); - if !applied { - return Ok(()); - } + worker.apply_worker_state(&snapshot); state.publish_worker_upsert(worker_ref.worker_id)?; state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; @@ -2517,7 +2504,6 @@ impl Runtime { workspace_id, runtime_id, worker.worker_id, - worker.run_generation, ) } @@ -2589,12 +2575,6 @@ impl Runtime { "Worker retention requires a stopped Worker".to_string(), )); } - if worker.run_generation != request.expected_run_generation { - return Err(RuntimeError::InvalidRequest(format!( - "Worker retention plan expected generation {}, current generation is {}", - request.expected_run_generation, worker.run_generation - ))); - } let result = provider.execute(request)?; state.workers.remove(&request.worker_id); state.persist_runtime_snapshot()?; @@ -2758,7 +2738,6 @@ impl RuntimeState { let diagnostics = persisted.diagnostics; let next_diagnostic_id = persisted.next_diagnostic_id; for (worker_id, worker) in persisted.workers { - let run_generation = worker.execution.last_run_generation; workers.insert( worker_id, WorkerRecord { @@ -2768,7 +2747,6 @@ impl RuntimeState { worker_state: None, workspace_id: worker.workspace_id, request: worker.request, - run_generation, execution_bound: worker.execution.binding.is_some(), restore_intent: worker.execution.restore_intent, working_directory: worker.working_directory, @@ -3563,14 +3541,8 @@ impl RuntimeState { } => snapshot, _ => return false, }; - match worker.apply_worker_state(incoming) { - Ok(protocol::WorkerStateSnapshotApply::Applied) => true, - Ok( - protocol::WorkerStateSnapshotApply::Duplicate - | protocol::WorkerStateSnapshotApply::Stale, - ) - | Err(_) => false, - } + worker.apply_worker_state(incoming); + true } } @@ -3614,7 +3586,6 @@ struct WorkerRecord { worker_state: Option, workspace_id: Option, request: CreateWorkerRequest, - run_generation: u64, execution_bound: bool, restore_intent: WorkerRestoreIntent, working_directory: Option, @@ -3623,17 +3594,8 @@ struct WorkerRecord { } impl WorkerRecord { - fn apply_worker_state( - &mut self, - incoming: &protocol::WorkerStateSnapshot, - ) -> Result { - match self.worker_state.as_mut() { - Some(current) => protocol::apply_worker_state_snapshot(current, incoming), - None => { - self.worker_state = Some(incoming.clone()); - Ok(protocol::WorkerStateSnapshotApply::Applied) - } - } + fn apply_worker_state(&mut self, incoming: &protocol::WorkerStateSnapshot) { + self.worker_state = Some(incoming.clone()); } fn belongs_to_workspace(&self, workspace_id: &str) -> bool { @@ -3678,12 +3640,9 @@ impl WorkerRecord { request: self.request.clone(), status: self.status, execution: PersistedWorkerExecution { - last_run_generation: self.run_generation, binding: self .execution_bound - .then_some(PersistedWorkerExecutionBinding { - run_generation: self.run_generation, - }), + .then_some(PersistedWorkerExecutionBinding {}), restore_intent: self.restore_intent, }, workspace_id: self.workspace_id.clone(), @@ -4028,11 +3987,7 @@ mod tests { } fn test_command() -> protocol::WorkerCommandEnvelope { - protocol::WorkerCommandEnvelope { - command_id: 1, - expected_execution_generation: 1, - expected_worker_state_revision: 0, - } + protocol::WorkerCommandEnvelope { command_id: 1 } } #[test] @@ -4891,7 +4846,6 @@ mod tests { restore_result: Mutex>, restore_gate: Mutex>>, restore_count: Mutex, - run_generations: Mutex>, config_bundles: Mutex>>, workspace_config_fetches: Mutex>, workspace_config_results: Mutex>, @@ -4989,10 +4943,6 @@ mod tests { } fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { - self.run_generations - .lock() - .unwrap() - .push(request.run_generation); self.config_bundles .lock() .unwrap() @@ -5004,7 +4954,6 @@ mod tests { WorkerExecutionSpawnResult::Connected { handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()), worker_state: protocol::WorkerStateSnapshot { - execution_generation: request.run_generation, ..protocol::WorkerStatus::Idle.into() }, working_directory: request @@ -5023,10 +4972,6 @@ mod tests { if let Some(gate) = restore_gate { gate.enter_and_wait(); } - self.run_generations - .lock() - .unwrap() - .push(request.run_generation); self.config_bundles .lock() .unwrap() @@ -5041,7 +4986,6 @@ mod tests { WorkerExecutionSpawnResult::Connected { handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()), worker_state: protocol::WorkerStateSnapshot { - execution_generation: request.run_generation, ..protocol::WorkerStatus::Idle.into() }, working_directory: request @@ -5962,11 +5906,8 @@ mod tests { assert_eq!(*backend.restore_count.lock().unwrap(), 1); assert_eq!(restored.status, WorkerStatus::Idle); assert_eq!( - restored - .worker_state - .as_ref() - .map(|state| state.execution_generation), - Some(2) + restored.worker_state.as_ref().map(|state| &state.state), + Some(&protocol::WorkerState::Idle) ); } @@ -6010,16 +5951,7 @@ mod tests { ); assert_eq!(*backend.restore_count.lock().unwrap(), 1); assert_eq!(restored[0].worker_ref, restored[1].worker_ref); - assert_eq!( - restored[0] - .worker_state - .as_ref() - .map(|state| state.execution_generation), - restored[1] - .worker_state - .as_ref() - .map(|state| state.execution_generation) - ); + assert_eq!(restored[0].worker_state, restored[1].worker_state); assert!(runtime.worker_operations.lock().unwrap().is_empty()); } @@ -6087,69 +6019,55 @@ mod tests { let worker_state = restored .worker_state .expect("restored Worker must expose its initial state"); - assert_eq!(worker_state.execution_generation, 2); + assert_eq!(worker_state.last_command_id, 0); assert_eq!(worker_state.state, protocol::WorkerState::Idle); } #[test] - fn runtime_applies_only_newer_worker_state_snapshots() { + fn runtime_replaces_worker_state_with_each_full_snapshot() { let (runtime, _) = runtime_and_backend(); let detail = runtime - .create_worker(task_request("state ordering")) + .create_worker(task_request("state replacement")) .unwrap(); let running = protocol::WorkerStateSnapshot { - execution_generation: 7, - revision: 3, state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Running, )), last_command_id: 2, }; - assert!({ + { let mut state = runtime.lock().unwrap(); - state.project_protocol_event_to_worker_state( + assert!(state.project_protocol_event_to_worker_state( &detail.worker_ref, &protocol::Event::WorkerState { snapshot: running.clone(), }, - ) - }); + )); + } assert_eq!( runtime .worker_detail(&detail.worker_ref) .unwrap() .worker_state, - Some(running.clone()) + Some(running) ); - assert!({ + let fresh = protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Idle, + last_command_id: 0, + }; + { let mut state = runtime.lock().unwrap(); - !state.project_protocol_event_to_worker_state( + assert!(state.project_protocol_event_to_worker_state( &detail.worker_ref, &protocol::Event::WorkerState { - snapshot: protocol::WorkerStateSnapshot { - revision: 2, - state: protocol::WorkerState::Idle, - ..running.clone() - }, + snapshot: fresh.clone(), }, - ) - }); - assert!({ - let mut state = runtime.lock().unwrap(); - !state.project_protocol_event_to_worker_state( - &detail.worker_ref, - &protocol::Event::WorkerState { - snapshot: protocol::WorkerStateSnapshot { - state: protocol::WorkerState::Idle, - ..running.clone() - }, - }, - ) - }); + )); + } let after = runtime.worker_detail(&detail.worker_ref).unwrap(); assert_eq!(after.status, WorkerStatus::Idle); - assert_eq!(after.worker_state, Some(running)); + assert_eq!(after.worker_state, Some(fresh)); } #[test] @@ -6369,7 +6287,6 @@ mod tests { WorkerExecutionSpawnResult::Connected { handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()), worker_state: protocol::WorkerStateSnapshot { - execution_generation: request.run_generation, ..protocol::WorkerStatus::Idle.into() }, working_directory: request @@ -6468,15 +6385,14 @@ mod tests { .unwrap(); assert_eq!(*backend.restore_count.lock().unwrap(), 1); - assert_eq!(*backend.run_generations.lock().unwrap(), vec![1, 2]); let restored = runtime.worker_detail(&detail.worker_ref).unwrap(); assert_eq!(restored.status, WorkerStatus::Idle); assert_eq!( restored .worker_state .as_ref() - .map(|snapshot| (snapshot.execution_generation, &snapshot.state)), - Some((2, &protocol::WorkerState::Idle)) + .map(|snapshot| &snapshot.state), + Some(&protocol::WorkerState::Idle) ); } @@ -6828,7 +6744,7 @@ mod tests { assert!( error .to_string() - .contains("unsupported Runtime store schema version 2; expected 3, 4, 5, or 6") + .contains("unsupported Runtime store schema version 2; expected 6 or 7") ); let _ = std::fs::remove_dir_all(root); @@ -6870,15 +6786,16 @@ mod tests { let worker_snapshot: serde_json::Value = serde_json::from_slice(&std::fs::read(worker_store_dir.join("worker.json")).unwrap()) .unwrap(); - assert_eq!(worker_snapshot["schema_version"], serde_json::json!(6)); + assert_eq!(worker_snapshot["schema_version"], serde_json::json!(7)); assert_eq!(worker_snapshot["status"], serde_json::json!("stopped")); - assert_eq!( - worker_snapshot["execution"]["last_run_generation"], - serde_json::json!(1) + assert!( + worker_snapshot["execution"] + .get("last_run_generation") + .is_none() ); assert_eq!( - worker_snapshot["execution"]["binding"]["run_generation"], - serde_json::json!(1) + worker_snapshot["execution"]["binding"], + serde_json::json!({}) ); assert_eq!( worker_snapshot["execution"]["restore_intent"], @@ -7256,8 +7173,8 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_migrates_schema_v3_workers_to_stopped_explicit_restore() { - let root = fs_store_root("schema-v3-restore-intent"); + fn fs_store_migrates_schema_v6_workers_without_losing_automatic_restore() { + let root = fs_store_root("schema-v6-no-generation"); let options = crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), runtime_id: "test-runtime".to_string(), @@ -7270,7 +7187,7 @@ mod tests { .unwrap(); runtime.store_config_bundle(test_bundle()).unwrap(); let worker = runtime - .create_worker(task_request("schema v3 worker")) + .create_worker(task_request("schema v6 worker")) .unwrap(); drop(runtime); @@ -7281,7 +7198,7 @@ mod tests { .join("worker.json"); let mut runtime_json: serde_json::Value = serde_json::from_slice(&std::fs::read(&runtime_path).unwrap()).unwrap(); - runtime_json["schema_version"] = serde_json::json!(3); + runtime_json["schema_version"] = serde_json::json!(6); std::fs::write( &runtime_path, serde_json::to_vec_pretty(&runtime_json).unwrap(), @@ -7289,10 +7206,9 @@ mod tests { .unwrap(); let mut worker_json: serde_json::Value = serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); - worker_json["schema_version"] = serde_json::json!(3); - worker_json.as_object_mut().unwrap().remove("status"); - worker_json.as_object_mut().unwrap().remove("execution"); - worker_json["run_generation"] = serde_json::json!(7); + worker_json["schema_version"] = serde_json::json!(6); + worker_json["execution"]["last_run_generation"] = serde_json::json!(1); + worker_json["execution"]["binding"] = serde_json::json!({"run_generation": 1}); std::fs::write( &worker_path, serde_json::to_vec_pretty(&worker_json).unwrap(), @@ -7302,35 +7218,24 @@ mod tests { let backend = Arc::new(TestExecutionBackend::default()); let migrated = Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap(); - assert_eq!(*backend.restore_count.lock().unwrap(), 0); + assert_eq!(*backend.restore_count.lock().unwrap(), 1); assert_eq!( migrated.worker_detail(&worker.worker_ref).unwrap().status, - WorkerStatus::Stopped + WorkerStatus::Idle ); let migrated_json: serde_json::Value = serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); - assert_eq!(migrated_json["schema_version"], serde_json::json!(6)); - assert_eq!(migrated_json["status"], serde_json::json!("stopped")); - assert_eq!( - migrated_json["execution"]["last_run_generation"], - serde_json::json!(7) - ); - assert_eq!( - migrated_json["execution"]["binding"], - serde_json::Value::Null + assert_eq!(migrated_json["schema_version"], serde_json::json!(7)); + assert_eq!(migrated_json["execution"]["binding"], serde_json::json!({})); + assert!( + migrated_json["execution"] + .get("last_run_generation") + .is_none() ); assert_eq!( migrated_json["execution"]["restore_intent"], - serde_json::json!("explicit") + serde_json::json!("automatic") ); - assert!(matches!( - migrated.send_input(&worker.worker_ref, WorkerInput::notify("do not restore")), - Err(RuntimeError::WorkerExecutionUnavailable { .. }) - )); - - migrated.restore_worker(&worker.worker_ref).unwrap(); - assert_eq!(*backend.restore_count.lock().unwrap(), 1); - assert_eq!(backend.run_generations.lock().unwrap().as_slice(), &[8]); let _ = std::fs::remove_dir_all(root); } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index cc405c64..e17794e8 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -57,7 +57,7 @@ fn next_internal_command( }) .unwrap_or(floor) .max(floor); - Ok(WorkerCommandEnvelope::for_snapshot(command_id, &snapshot)) + Ok(WorkerCommandEnvelope::new(command_id)) } use session_store::{CombinedStore, WorkerAggregateStore, WorkerSessionStore}; #[cfg(test)] @@ -403,6 +403,13 @@ impl ProfileRuntimeWorkerFactory { }) } + fn worker_run_dir(&self, worker_ref: &WorkerRef) -> Result { + Ok(self + .worker_aggregate_dir(worker_ref)? + .join("runs") + .join(uuid::Uuid::now_v7().to_string())) + } + fn runtime_worker_name_for_ref(worker_ref: &crate::identity::WorkerRef) -> String { format!("worker-runtime-{}", worker_ref.worker_id) } @@ -898,9 +905,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { )?; let store = CombinedStore::new(session_store, worker_metadata_store); - let run_dir = worker_aggregate_dir - .join("runs") - .join(request.run_generation.to_string()); + let run_dir = self.worker_run_dir(&request.worker_ref)?; let bash_output_dir = bash_output_dir_for_worker_id(&request.worker_ref.worker_id); let mut prepared = WorkerBootstrap::new( manifest, @@ -1152,9 +1157,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { } let workspace_client = worker.workspace_client_handle(); - let run_dir = worker_aggregate_dir - .join("runs") - .join(request.run_generation.to_string()); + let run_dir = self.worker_run_dir(&request.worker_ref)?; let bash_output_dir = bash_output_dir_for_worker_id(&request.worker_ref.worker_id); let started = PreparedWorker::new( worker, @@ -1745,25 +1748,17 @@ fn apply_protocol_worker_state( current: &Arc>, event: &mut Event, ) -> Result { - let (incoming, replace_stale) = match event { - Event::WorkerState { snapshot } => (snapshot, false), - Event::Snapshot { state, .. } => (state, true), - Event::CommandAcknowledged { acknowledgement } => (&mut acknowledgement.state, true), + let incoming = match event { + Event::WorkerState { snapshot } => snapshot, + Event::Snapshot { state, .. } => state, + Event::CommandAcknowledged { acknowledgement } => &mut acknowledgement.state, _ => return Ok(true), }; let mut current = current .write() .map_err(|_| "worker state projection lock is poisoned".to_string())?; - match protocol::apply_worker_state_snapshot(&mut current, incoming) { - Ok(protocol::WorkerStateSnapshotApply::Applied) - | Ok(protocol::WorkerStateSnapshotApply::Duplicate) => Ok(true), - Ok(protocol::WorkerStateSnapshotApply::Stale) if replace_stale => { - *incoming = current.clone(); - Ok(true) - } - Ok(protocol::WorkerStateSnapshotApply::Stale) => Ok(false), - Err(error) => Err(error.to_string()), - } + *current = incoming.clone(); + Ok(true) } impl WorkerExecutionBackend for WorkerRuntimeExecutionBackend @@ -2572,37 +2567,33 @@ mod tests { .read() .unwrap() .clone(); - WorkerCommandEnvelope::for_snapshot(state.last_command_id.saturating_add(1), &state) + WorkerCommandEnvelope::new(state.last_command_id.saturating_add(1)) } #[test] - fn protocol_bridge_applies_state_and_acknowledgement_monotonically() { + fn protocol_bridge_replaces_every_full_state_snapshot() { let running = protocol::WorkerStateSnapshot { - execution_generation: 4, - revision: 3, state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Running, )), last_command_id: 2, }; - let current = Arc::new(RwLock::new(running.clone())); - let mut stale = Event::WorkerState { - snapshot: protocol::WorkerStateSnapshot { - revision: 2, - state: protocol::WorkerState::Idle, - ..running.clone() - }, + let current = Arc::new(RwLock::new(running)); + let idle = protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Idle, + last_command_id: 0, }; - assert!(!apply_protocol_worker_state(¤t, &mut stale).unwrap()); - assert_eq!(*current.read().unwrap(), running); + let mut replacement = Event::WorkerState { + snapshot: idle.clone(), + }; + assert!(apply_protocol_worker_state(¤t, &mut replacement).unwrap()); + assert_eq!(*current.read().unwrap(), idle); let paused = protocol::WorkerStateSnapshot { - revision: 4, state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Paused, )), last_command_id: 3, - ..running.clone() }; let mut acknowledgement = Event::CommandAcknowledged { acknowledgement: protocol::WorkerCommandAcknowledgement { @@ -2614,15 +2605,6 @@ mod tests { }; assert!(apply_protocol_worker_state(¤t, &mut acknowledgement).unwrap()); assert_eq!(*current.read().unwrap(), paused); - - let mut conflict = Event::WorkerState { - snapshot: protocol::WorkerStateSnapshot { - state: protocol::WorkerState::Idle, - ..paused.clone() - }, - }; - assert!(apply_protocol_worker_state(¤t, &mut conflict).is_err()); - assert_eq!(*current.read().unwrap(), paused); } #[test] @@ -3037,7 +3019,6 @@ mod tests { ) -> Result { let request = WorkerExecutionSpawnRequest { worker_ref: request.worker_ref, - run_generation: request.run_generation, request: request.request, workspace_scope: request.workspace_scope, context: request.context, @@ -3388,7 +3369,6 @@ mod tests { crate::identity::WorkerRef::new(crate::identity::WorkerId::from_legacy_u64(1)); let request = WorkerExecutionSpawnRequest { worker_ref: worker_ref.clone(), - run_generation: 1, request: create_request("1"), workspace_scope: None, context: test_execution_context(worker_ref), @@ -3518,7 +3498,6 @@ mod tests { .with_remote_worker_mutation_identity(identity) .restore_controller(WorkerExecutionRestoreRequest { worker_ref: worker_ref.clone(), - run_generation: 1, request, workspace_scope: Some(crate::runtime::RuntimeWorkspaceScope::new( "workspace-restore", @@ -3583,11 +3562,13 @@ mod tests { ) .unwrap(); - let run_dir = runtime_store_dir + let runs_dir = runtime_store_dir .join("workers") .join(worker_ref.worker_id.to_string()) - .join("runs/2"); - let socket_path = run_dir.join("worker.sock"); + .join("runs"); + let socket_path = runs_dir + .join(uuid::Uuid::nil().to_string()) + .join("worker.sock"); assert!( socket_path.as_os_str().as_encoded_bytes().len() > 107, "test path must exceed Linux sockaddr_un.sun_path capacity: {}", @@ -3599,7 +3580,6 @@ mod tests { .with_controller_transport(WorkerControllerTransport::InProcess) .restore_controller(WorkerExecutionRestoreRequest { worker_ref: worker_ref.clone(), - run_generation: 2, request: create_request("embedded restore"), workspace_scope: None, context: test_execution_context(worker_ref), @@ -3614,7 +3594,12 @@ mod tests { controller.handle.shared_state.catalog_status(), WorkerStatus::Idle ); - assert!(!socket_path.exists()); + let run_dir = std::fs::read_dir(&runs_dir) + .unwrap() + .map(|entry| entry.unwrap().path()) + .next() + .expect("restore must create one run artifact directory"); + assert!(!run_dir.join("worker.sock").exists()); assert!(run_dir.join("worker.out.log").is_file()); assert!(run_dir.join("worker.err.log").is_file()); let worker_state = Arc::new(RwLock::new(controller.handle.shared_state.snapshot())); @@ -3726,10 +3711,16 @@ mod tests { let mut request = create_request("embedded singleton"); request.profile = ProfileSelector::Builtin("default".to_string()); let worker = runtime.create_worker(request).unwrap(); - let first_run_socket = runtime_store_dir + let runs_dir = runtime_store_dir .join("workers") .join(worker.worker_id.to_string()) - .join("runs/1/worker.sock"); + .join("runs"); + let first_run = std::fs::read_dir(&runs_dir) + .unwrap() + .map(|entry| entry.unwrap().path()) + .next() + .expect("fresh launch must create one run artifact directory"); + let first_run_socket = first_run.join("worker.sock"); assert!( first_run_socket.as_os_str().as_encoded_bytes().len() > 107, "test path must exceed Linux sockaddr_un.sun_path capacity: {}", @@ -3766,10 +3757,11 @@ mod tests { diagnostic.code == "worker_execution_restore_failed" && diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref) })); - let restored_run = runtime_store_dir - .join("workers") - .join(worker.worker_id.to_string()) - .join("runs/2"); + let restored_run = std::fs::read_dir(&runs_dir) + .unwrap() + .map(|entry| entry.unwrap().path()) + .find(|path| path != &first_run) + .expect("restore must create a distinct run artifact directory"); assert!(!restored_run.join("worker.sock").exists()); assert!(restored_run.join("worker.out.log").is_file()); assert!(restored_run.join("worker.err.log").is_file()); diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index b22bd998..ebcb5c62 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -190,12 +190,6 @@ fn command_admission_disposition( Err(WorkerCommandDisposition::StaleCommandId) } WorkerCommandAdmission::Conflict => Err(WorkerCommandDisposition::Conflict), - WorkerCommandAdmission::ExecutionGenerationMismatch => { - Err(WorkerCommandDisposition::StaleExecutionGeneration) - } - WorkerCommandAdmission::StateRevisionMismatch => { - Err(WorkerCommandDisposition::StaleWorkerStateRevision) - } } } @@ -204,14 +198,14 @@ fn validate_command( kind: WorkerCommandKind, shared_state: &WorkerSharedState, ) -> Result<(), WorkerCommandDisposition> { - command_admission_disposition(shared_state.admit_command(envelope, kind, true)) + command_admission_disposition(shared_state.admit_command(envelope, kind)) } fn validate_shutdown_command( envelope: WorkerCommandEnvelope, shared_state: &WorkerSharedState, ) -> Result<(), WorkerCommandDisposition> { - match shared_state.admit_command(envelope, WorkerCommandKind::Shutdown, false) { + match shared_state.admit_command(envelope, WorkerCommandKind::Shutdown) { WorkerCommandAdmission::Accepted | WorkerCommandAdmission::Retry => Ok(()), admission => command_admission_disposition(admission), } @@ -793,19 +787,11 @@ impl WorkerController { .await .map_err(|error| std::io::Error::other(error.to_string()))?; let greeting = build_greeting(&worker); - let execution_generation = runtime_dir - .path() - .file_name() - .and_then(|name| name.to_str()) - .and_then(|name| name.parse::().ok()) - .filter(|generation| *generation > 0) - .unwrap_or(1); - let shared_state = Arc::new(WorkerSharedState::new_with_generation( + let shared_state = Arc::new(WorkerSharedState::new( worker.manifest().worker.name.clone(), worker.segment_id(), manifest_toml.clone(), greeting, - execution_generation, )); if let Some(fs_for_view) = fs_for_view { shared_state.set_fs_view(crate::fs_view::WorkerFsView::new(fs_for_view)); @@ -3436,7 +3422,7 @@ mod tests { .transition(WorkerState::Busy(WorkerBusyState::Run( WorkerRunState::Running, ))); - let command = WorkerCommandEnvelope::for_snapshot(1, &env.shared_state.snapshot()); + let command = WorkerCommandEnvelope::new(1); tokio::spawn(async move { tokio::time::sleep(Duration::from_millis(10)).await; method_tx @@ -3719,7 +3705,7 @@ mod tests { .transition(WorkerState::Busy(WorkerBusyState::Run( WorkerRunState::Running, ))); - let command = WorkerCommandEnvelope::for_snapshot(1, &env.shared_state.snapshot()); + let command = WorkerCommandEnvelope::new(1); env._method_tx .send(Method::Compact { command }) .await @@ -3770,8 +3756,8 @@ mod tests { } #[test] - fn command_admission_rejects_stale_generation_revision_and_order() { - let shared = WorkerSharedState::new_with_generation( + fn command_admission_rejects_stale_ids_and_reuse_conflicts() { + let shared = WorkerSharedState::new( "worker".into(), session_store::new_segment_id(), String::new(), @@ -3785,39 +3771,10 @@ mod tests { context_window: 1, context_tokens: 0, }, - 9, - ); - assert_eq!( - validate_command( - WorkerCommandEnvelope { - command_id: 1, - expected_execution_generation: 8, - expected_worker_state_revision: 0, - }, - WorkerCommandKind::Pause, - &shared, - ), - Err(WorkerCommandDisposition::StaleExecutionGeneration) - ); - assert_eq!( - validate_command( - WorkerCommandEnvelope { - command_id: 2, - expected_execution_generation: 9, - expected_worker_state_revision: 1, - }, - WorkerCommandKind::Pause, - &shared, - ), - Err(WorkerCommandDisposition::StaleWorkerStateRevision) ); assert!( validate_command( - WorkerCommandEnvelope { - command_id: 1, - expected_execution_generation: 9, - expected_worker_state_revision: 0, - }, + WorkerCommandEnvelope { command_id: 1 }, WorkerCommandKind::Pause, &shared, ) @@ -3825,11 +3782,7 @@ mod tests { ); assert_eq!( validate_command( - WorkerCommandEnvelope { - command_id: 1, - expected_execution_generation: 9, - expected_worker_state_revision: 0, - }, + WorkerCommandEnvelope { command_id: 1 }, WorkerCommandKind::Pause, &shared, ), @@ -3837,11 +3790,7 @@ mod tests { ); assert_eq!( validate_command( - WorkerCommandEnvelope { - command_id: 1, - expected_execution_generation: 9, - expected_worker_state_revision: 0, - }, + WorkerCommandEnvelope { command_id: 1 }, WorkerCommandKind::Cancel, &shared, ), @@ -3849,11 +3798,7 @@ mod tests { ); assert!( validate_command( - WorkerCommandEnvelope { - command_id: 2, - expected_execution_generation: 9, - expected_worker_state_revision: 1, - }, + WorkerCommandEnvelope { command_id: 2 }, WorkerCommandKind::Pause, &shared, ) diff --git a/crates/worker/src/internal_worker.rs b/crates/worker/src/internal_worker.rs index 25488987..d5185833 100644 --- a/crates/worker/src/internal_worker.rs +++ b/crates/worker/src/internal_worker.rs @@ -296,7 +296,6 @@ impl InternalWorkerSessionStatus { fn send_internal_worker_state( event_tx: &broadcast::Sender, - state_revision: &std::sync::atomic::AtomicU64, status: InternalWorkerSessionStatus, ) { let state = match status { @@ -313,13 +312,8 @@ fn send_internal_worker_state( protocol::WorkerBusyState::Run(protocol::WorkerRunState::Cancelling), ), }; - let revision = state_revision - .fetch_add(1, std::sync::atomic::Ordering::AcqRel) - .saturating_add(1); let _ = event_tx.send(Event::WorkerState { snapshot: protocol::WorkerStateSnapshot { - execution_generation: 1, - revision, last_command_id: 0, state, }, @@ -382,7 +376,6 @@ pub(crate) struct InternalWorkerSessionSnapshot { pub(crate) struct InternalWorkerSessionHandle { command_tx: tokio::sync::mpsc::Sender, status: Arc, - state_revision: Arc, store: EphemeralSessionStore, session_id: SessionId, segment_id: SegmentId, @@ -433,7 +426,7 @@ impl InternalWorkerSessionHandle { } fn emit_worker_state(&self, status: InternalWorkerSessionStatus) { - send_internal_worker_state(&self.event_tx, &self.state_revision, status); + send_internal_worker_state(&self.event_tx, status); } pub(crate) fn protocol_snapshot(&self) -> InternalWorkerSessionSnapshot { @@ -803,13 +796,11 @@ pub(crate) async fn prepare_internal_worker_session( let status = Arc::new(std::sync::atomic::AtomicU8::new( InternalWorkerSessionStatus::Idle.encode(), )); - let state_revision = Arc::new(std::sync::atomic::AtomicU64::new(0)); let state_changed = Arc::new(tokio::sync::Notify::new()); let last_error = Arc::new(Mutex::new(None)); let handle = InternalWorkerSessionHandle { command_tx, status: status.clone(), - state_revision: state_revision.clone(), store, session_id, segment_id, @@ -845,11 +836,7 @@ pub(crate) async fn prepare_internal_worker_session( message, }); } - send_internal_worker_state( - &event_tx, - &state_revision, - turn_status, - ); + send_internal_worker_state(&event_tx, turn_status); if let Some(callback) = &on_turn_end { callback(turn_status); } @@ -891,11 +878,7 @@ pub(crate) async fn prepare_internal_worker_session( InternalWorkerSessionStatus::Stopped.encode(), std::sync::atomic::Ordering::Release, ); - send_internal_worker_state( - &event_tx, - &state_revision, - InternalWorkerSessionStatus::Stopped, - ); + send_internal_worker_state(&event_tx, InternalWorkerSessionStatus::Stopped); let _ = event_tx.send(Event::Shutdown); state_changed.notify_waiters(); if let Some(done) = stop_done { @@ -1147,7 +1130,6 @@ pub(crate) fn test_internal_worker_session( status: Arc::new(std::sync::atomic::AtomicU8::new( InternalWorkerSessionStatus::Idle.encode(), )), - state_revision: Arc::new(std::sync::atomic::AtomicU64::new(0)), store, session_id, segment_id, diff --git a/crates/worker/src/shared_state.rs b/crates/worker/src/shared_state.rs index 810b59f4..73fe014e 100644 --- a/crates/worker/src/shared_state.rs +++ b/crates/worker/src/shared_state.rs @@ -28,8 +28,6 @@ pub(crate) enum WorkerCommandAdmission { Retry, Conflict, StaleCommandId, - ExecutionGenerationMismatch, - StateRevisionMismatch, } /// Shared state between WorkerController and runtime directory. @@ -59,23 +57,13 @@ impl WorkerSharedState { segment_id: SegmentId, manifest_toml: String, greeting: protocol::Greeting, - ) -> Self { - Self::new_with_generation(worker_name, segment_id, manifest_toml, greeting, 1) - } - - pub fn new_with_generation( - worker_name: String, - segment_id: SegmentId, - manifest_toml: String, - greeting: protocol::Greeting, - execution_generation: u64, ) -> Self { Self { worker_name, segment_id, manifest_toml, greeting, - state: RwLock::new(WorkerStateSnapshot::initial(execution_generation)), + state: RwLock::new(WorkerStateSnapshot::initial()), accepted_commands: RwLock::new(VecDeque::new()), fs_view: OnceLock::new(), flow_transition_enabled: AtomicBool::new(false), @@ -108,7 +96,6 @@ impl WorkerSharedState { .write() .expect("worker state lock poisoned; refusing an inferred fallback state"); if snapshot.state != state { - snapshot.revision = snapshot.revision.saturating_add(1); snapshot.state = state; } snapshot.clone() @@ -118,7 +105,6 @@ impl WorkerSharedState { &self, envelope: WorkerCommandEnvelope, kind: WorkerCommandKind, - require_state_revision: bool, ) -> WorkerCommandAdmission { let mut snapshot = self .state @@ -138,18 +124,11 @@ impl WorkerSharedState { WorkerCommandAdmission::Conflict }; } - if envelope.expected_execution_generation != snapshot.execution_generation { - return WorkerCommandAdmission::ExecutionGenerationMismatch; - } - if require_state_revision && envelope.expected_worker_state_revision != snapshot.revision { - return WorkerCommandAdmission::StateRevisionMismatch; - } if envelope.command_id <= snapshot.last_command_id { return WorkerCommandAdmission::StaleCommandId; } snapshot.last_command_id = envelope.command_id; - snapshot.revision = snapshot.revision.saturating_add(1); accepted.push_back(AcceptedWorkerCommand { envelope, kind, @@ -248,12 +227,11 @@ mod tests { use super::*; fn test_state() -> WorkerSharedState { - WorkerSharedState::new_with_generation( + WorkerSharedState::new( "test-worker".into(), session_store::new_segment_id(), "[engine]\nname = \"test-worker\"".into(), test_greeting(), - 7, ) } @@ -273,43 +251,34 @@ mod tests { #[test] fn initial_snapshot_is_idle() { let state = test_state(); - assert_eq!(state.snapshot(), WorkerStateSnapshot::initial(7)); + assert_eq!(state.snapshot(), WorkerStateSnapshot::initial()); assert_eq!(state.catalog_status(), WorkerStatus::Idle); } #[test] - fn transitions_increment_revision_only_when_state_changes() { + fn transitions_publish_full_state() { let state = test_state(); let running = WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running)); - let snapshot = state.transition(running.clone()); - assert_eq!(snapshot.revision, 1); - assert_eq!(snapshot.state, running); - assert_eq!(state.transition(running).revision, 1); + assert_eq!(state.transition(running.clone()).state, running); let paused = WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Paused)); let snapshot = state.transition(paused.clone()); - assert_eq!(snapshot.revision, 2); assert_eq!(snapshot.state, paused); + assert_eq!(snapshot.last_command_id, 0); assert_eq!(state.catalog_status(), WorkerStatus::Paused); } #[test] - fn accepted_command_identity_advances_revision_and_detects_reuse_conflicts() { + fn accepted_command_identity_advances_last_id_and_detects_reuse_conflicts() { let state = test_state(); - let envelope = WorkerCommandEnvelope { - command_id: 9, - expected_execution_generation: 7, - expected_worker_state_revision: 0, - }; + let envelope = WorkerCommandEnvelope { command_id: 9 }; assert_eq!( - state.admit_command(envelope, WorkerCommandKind::Pause, true), + state.admit_command(envelope, WorkerCommandKind::Pause), WorkerCommandAdmission::Accepted ); assert_eq!( state.snapshot(), WorkerStateSnapshot { - execution_generation: 7, - revision: 1, last_command_id: 9, state: WorkerState::Idle, } @@ -325,14 +294,14 @@ mod tests { Some(Some(WorkerCommandDisposition::Accepted)) ); assert_eq!( - state.admit_command(envelope, WorkerCommandKind::Pause, true), + state.admit_command(envelope, WorkerCommandKind::Pause), WorkerCommandAdmission::Retry ); assert_eq!( - state.admit_command(envelope, WorkerCommandKind::Cancel, true), + state.admit_command(envelope, WorkerCommandKind::Cancel), WorkerCommandAdmission::Conflict ); - assert_eq!(state.snapshot().revision, 1); + assert_eq!(state.snapshot().last_command_id, 9); } #[test] @@ -343,8 +312,8 @@ mod tests { ))); let parsed: serde_json::Value = serde_json::from_str(&state.status_json()).unwrap(); assert_eq!(parsed["state"], "running"); - assert_eq!(parsed["worker_state"]["execution_generation"], 7); - assert_eq!(parsed["worker_state"]["revision"], 1); + assert!(parsed["worker_state"].get("execution_generation").is_none()); + assert!(parsed["worker_state"].get("revision").is_none()); assert_eq!(parsed["worker_state"]["state"]["kind"], "busy"); assert_eq!(parsed["worker_name"], "test-worker"); assert!(parsed["segment_id"].is_string()); diff --git a/crates/worker/src/spawn/comm_tools.rs b/crates/worker/src/spawn/comm_tools.rs index 28efb9a6..11ad8e14 100644 --- a/crates/worker/src/spawn/comm_tools.rs +++ b/crates/worker/src/spawn/comm_tools.rs @@ -138,10 +138,7 @@ mod tests { ); let method = Method::Shutdown { - command: protocol::WorkerCommandEnvelope::for_snapshot( - 1, - &protocol::WorkerStateSnapshot::initial(1), - ), + command: protocol::WorkerCommandEnvelope::new(1), }; connect_and_send(&socket, &method).await.unwrap(); diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index 555a9044..41643793 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -1135,7 +1135,7 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() { } } - let compact = protocol::WorkerCommandEnvelope::for_snapshot(1, &handle.shared_state.snapshot()); + let compact = protocol::WorkerCommandEnvelope::new(1); handle .send(Method::Compact { command: compact }) .await @@ -1154,7 +1154,7 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() { } } - let cancel = protocol::WorkerCommandEnvelope::for_snapshot(2, &handle.shared_state.snapshot()); + let cancel = protocol::WorkerCommandEnvelope::new(2); handle .send(Method::Cancel { command: cancel }) .await @@ -1183,7 +1183,7 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() { } } - let compact = protocol::WorkerCommandEnvelope::for_snapshot(3, &handle.shared_state.snapshot()); + let compact = protocol::WorkerCommandEnvelope::new(3); handle .send(Method::Compact { command: compact }) .await @@ -1201,8 +1201,7 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() { break; } } - let shutdown = - protocol::WorkerCommandEnvelope::for_snapshot(4, &handle.shared_state.snapshot()); + let shutdown = protocol::WorkerCommandEnvelope::new(4); handle .send(Method::Shutdown { command: shutdown }) .await @@ -1269,7 +1268,7 @@ async fn controller_compact_method_publishes_progress_and_clear() { } } - let command = protocol::WorkerCommandEnvelope::for_snapshot(1, &handle.shared_state.snapshot()); + let command = protocol::WorkerCommandEnvelope::new(1); handle .send(Method::Compact { command }) .await @@ -1316,6 +1315,6 @@ async fn controller_compact_method_publishes_progress_and_clear() { protocol::WorkerStatus::Idle, "successful manual compaction must release the execution fence" ); - let command = protocol::WorkerCommandEnvelope::for_snapshot(2, &handle.shared_state.snapshot()); + let command = protocol::WorkerCommandEnvelope::new(2); let _ = handle.send(Method::Shutdown { command }).await; } diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index cba7156e..72e587fe 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -27,11 +27,8 @@ type TestStore = CombinedStore; static NEXT_COMMAND_ID: AtomicU64 = AtomicU64::new(1); -fn worker_command(handle: &WorkerHandle) -> protocol::WorkerCommandEnvelope { - protocol::WorkerCommandEnvelope::for_snapshot( - NEXT_COMMAND_ID.fetch_add(1, Ordering::Relaxed), - &handle.shared_state.snapshot(), - ) +fn worker_command(_handle: &WorkerHandle) -> protocol::WorkerCommandEnvelope { + protocol::WorkerCommandEnvelope::new(NEXT_COMMAND_ID.fetch_add(1, Ordering::Relaxed)) } /// Reconstruct a worker-history-like `Vec` from the live session diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index f15845c1..28f7249f 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -5836,7 +5836,6 @@ mod tests { self.backend_id(), ), worker_state: protocol::WorkerStateSnapshot { - execution_generation: request.run_generation, ..protocol::WorkerStatus::Idle.into() }, working_directory: request diff --git a/crates/workspace-server/src/latest_schema.sql b/crates/workspace-server/src/latest_schema.sql index b2d8c083..50bb77b8 100644 --- a/crates/workspace-server/src/latest_schema.sql +++ b/crates/workspace-server/src/latest_schema.sql @@ -775,7 +775,7 @@ CREATE TABLE "worker_registry" ( CREATE TABLE worker_removal_operations ( operation_id TEXT PRIMARY KEY, plan_id TEXT NOT NULL UNIQUE, input_fingerprint TEXT NOT NULL, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, - worker_revision TEXT NOT NULL, run_generation INTEGER NOT NULL CHECK(run_generation>=0), + worker_revision TEXT NOT NULL, policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL, session_disposition TEXT NOT NULL, metadata_disposition TEXT NOT NULL, archive_retention_kind TEXT NOT NULL, archive_retention_seconds INTEGER, diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index 19c22ccb..6aa4925c 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -91,7 +91,6 @@ pub struct WorkerRemovalPlan { pub workspace_id: String, pub worker: RuntimeWorkerRef, pub worker_revision: String, - pub run_generation: u64, pub policy_id: String, pub policy_revision: u64, pub session_disposition: SessionDisposition, @@ -219,7 +218,7 @@ impl SqliteWorkspaceStore { let plan_id=stable("wrp",&fp); let operation_id=stable("wro",&fp); let archive_id=(policy.session_disposition==SessionDisposition::Archive).then(||stable("wra",&fp)); let state=if blockers.is_empty(){WorkerRemovalPlanState::Planned}else{WorkerRemovalPlanState::Blocked}; - tx.execute("INSERT OR IGNORE INTO worker_removal_operations(operation_id,plan_id,input_fingerprint,workspace_id,runtime_id,worker_id,worker_revision,run_generation,policy_id,policy_revision,session_disposition,metadata_disposition,archive_retention_kind,archive_retention_seconds,diagnostics_disposition,diagnostics_retention_seconds,archive_id,blockers_json,state,reason,created_at,updated_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?21)",params![operation_id,plan_id,fp,req.workspace_id,req.worker.runtime_id,req.worker.worker_id,worker.updated_at,inv.run_generation,policy.policy_id,policy.revision,sess(policy.session_disposition),meta(policy.metadata_disposition),archive_kind(policy.archive_retention),archive_seconds(policy.archive_retention),diag(policy.diagnostics_disposition),policy.diagnostics_retention_seconds,archive_id,serde_json::to_string(&blockers).map_err(|e|StoreError::InvalidInput(e.to_string()))?,state_s(state),req.reason,now])?; + tx.execute("INSERT OR IGNORE INTO worker_removal_operations(operation_id,plan_id,input_fingerprint,workspace_id,runtime_id,worker_id,worker_revision,policy_id,policy_revision,session_disposition,metadata_disposition,archive_retention_kind,archive_retention_seconds,diagnostics_disposition,diagnostics_retention_seconds,archive_id,blockers_json,state,reason,created_at,updated_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?20)",params![operation_id,plan_id,fp,req.workspace_id,req.worker.runtime_id,req.worker.worker_id,worker.updated_at,policy.policy_id,policy.revision,sess(policy.session_disposition),meta(policy.metadata_disposition),archive_kind(policy.archive_retention),archive_seconds(policy.archive_retention),diag(policy.diagnostics_disposition),policy.diagnostics_retention_seconds,archive_id,serde_json::to_string(&blockers).map_err(|e|StoreError::InvalidInput(e.to_string()))?,state_s(state),req.reason,now])?; let plan=load_plan(&tx,&plan_id)?.ok_or_else(||StoreError::InvalidInput("plan missing".into()))?; if plan.input_fingerprint!=fp{return Err(StoreError::InvalidInput(format!("fingerprint:{}",plan.operation_id)));} tx.commit()?; Ok(plan) @@ -304,7 +303,6 @@ impl SqliteWorkspaceStore { source_runtime_id: plan.worker.runtime_id.clone(), worker_id: worker_id, expected_worker_revision: plan.worker_revision.clone(), - expected_run_generation: plan.run_generation, source_created_at: worker.created_at, removed_at, effective_profile: worker.profile, @@ -376,7 +374,6 @@ impl SqliteWorkspaceStore { source_runtime_id: plan.worker.runtime_id.clone(), worker_id: worker_id, expected_worker_revision: plan.worker_revision.clone(), - expected_run_generation: plan.run_generation, source_created_at: worker .as_ref() .map(|worker| worker.created_at.clone()) @@ -682,20 +679,20 @@ fn load_plan_op(c: &Connection, id: &str) -> crate::Result crate::Result> { let query = format!( "SELECT plan_id,operation_id,input_fingerprint,workspace_id,runtime_id,worker_id, - worker_revision,run_generation,policy_id,policy_revision,session_disposition, + worker_revision,policy_id,policy_revision,session_disposition, metadata_disposition,archive_retention_kind,archive_retention_seconds, diagnostics_disposition,diagnostics_retention_seconds,archive_id,blockers_json, state,reason,created_at,updated_at,failure_category FROM worker_removal_operations WHERE {key}=?1" ); c.query_row(&query, params![id], |row| { - let session: String = row.get(10)?; - let metadata: String = row.get(11)?; - let archive_kind: String = row.get(12)?; - let archive_seconds: Option = row.get(13)?; - let diagnostics: String = row.get(14)?; - let blockers: String = row.get(17)?; - let state: String = row.get(18)?; + let session: String = row.get(9)?; + let metadata: String = row.get(10)?; + let archive_kind: String = row.get(11)?; + let archive_seconds: Option = row.get(12)?; + let diagnostics: String = row.get(13)?; + let blockers: String = row.get(16)?; + let state: String = row.get(17)?; Ok(WorkerRemovalPlan { plan_id: row.get(0)?, operation_id: row.get(1)?, @@ -706,27 +703,26 @@ fn load_plan_q(c: &Connection, key: &str, id: &str) -> crate::Result(7)? as u64, - policy_id: row.get(8)?, - policy_revision: row.get::<_, i64>(9)? as u64, + policy_id: row.get(7)?, + policy_revision: row.get::<_, i64>(8)? as u64, session_disposition: parse_s(&session)?, metadata_disposition: parse_m(&metadata)?, archive_retention: parse_archive(&archive_kind, archive_seconds)?, diagnostics_disposition: parse_d(&diagnostics)?, - diagnostics_retention_seconds: row.get::<_, Option>(15)?.map(|v| v as u64), - archive_id: row.get(16)?, + diagnostics_retention_seconds: row.get::<_, Option>(14)?.map(|v| v as u64), + archive_id: row.get(15)?, blockers: serde_json::from_str(&blockers).map_err(|error| { rusqlite::Error::FromSqlConversionFailure( - 17, + 16, rusqlite::types::Type::Text, Box::new(error), ) })?, state: parse_state(&state)?, - reason: row.get(19)?, - created_at: row.get(20)?, - updated_at: row.get(21)?, - failure_category: row.get(22)?, + reason: row.get(18)?, + created_at: row.get(19)?, + updated_at: row.get(20)?, + failure_category: row.get(21)?, }) }) .optional() @@ -755,7 +751,6 @@ fn fingerprint( r.worker.runtime_id, r.worker.worker_id, worker_revision, - i.run_generation, i.session_id, i.segment_ids, p.policy_id, @@ -992,7 +987,6 @@ mod tests { workspace_id: "w".into(), runtime_id: "r".into(), worker_id: worker_id(), - run_generation: 2, session_id: Some("s".into()), segment_ids: vec!["a".into()], session_bytes: 1, @@ -1139,13 +1133,12 @@ mod tests { } #[test] - fn prepared_execution_is_derived_from_pinned_plan_generation() { + fn prepared_execution_is_derived_from_pinned_plan() { let s = setup(); let plan = s.plan_worker_removal(&req(), &inv()).unwrap(); let prepared = s .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) .unwrap(); - assert_eq!(prepared.runtime_request.expected_run_generation, 2); assert_eq!( prepared.runtime_request.session_disposition, SessionDisposition::Archive @@ -1264,7 +1257,6 @@ mod tests { workspace_id: "w".into(), runtime_id: "r".into(), worker_id: WorkerId::from_legacy_u64(2), - run_generation: 1, session_id: Some("orphan-session".into()), segment_ids: vec![], session_bytes: 10, diff --git a/crates/workspace-server/src/runtime_subscription_tests.rs b/crates/workspace-server/src/runtime_subscription_tests.rs index ee58f3d0..9d618c19 100644 --- a/crates/workspace-server/src/runtime_subscription_tests.rs +++ b/crates/workspace-server/src/runtime_subscription_tests.rs @@ -20,7 +20,6 @@ impl WorkerExecutionBackend for TestExecutionBackend { WorkerExecutionSpawnResult::connected( WorkerExecutionHandle::new(request.worker_ref, self.backend_id()), protocol::WorkerStateSnapshot { - execution_generation: request.run_generation, ..protocol::WorkerStatus::Idle.into() }, None, @@ -182,7 +181,6 @@ async fn equal_downstream_selectors_share_one_upstream_subscription() { .worker_state .clone() .expect("connected test Worker must expose its initial state"); - running.revision += 1; running.state = protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Running, )); @@ -348,7 +346,6 @@ async fn embedded_runtime_uses_in_process_subscription_source() { .worker_state .clone() .expect("connected test Worker must expose its initial state"); - running.revision += 1; running.state = protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Running, )); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 57e79c15..c3f56388 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -21861,7 +21861,6 @@ mod tests { self.backend_id(), ), worker_state: protocol::WorkerStateSnapshot { - execution_generation: request.run_generation, ..protocol::WorkerStatus::Idle.into() }, working_directory, @@ -21901,10 +21900,11 @@ mod tests { context_window: 0, context_tokens: 0, }, - state: protocol::WorkerStateSnapshot::initial(1), + state: protocol::WorkerStateSnapshot::initial(), in_flight: protocol::InFlightSnapshot { blocks: Vec::new(), commands: Vec::new(), + compaction: None, }, internal_workers: Vec::new(), }) @@ -21965,12 +21965,12 @@ mod tests { uuid::Uuid::now_v7().to_string(), protocol::SubmissionDisposition::Started, ) - .with_worker_state(protocol::WorkerStateSnapshot::initial(1)) + .with_worker_state(protocol::WorkerStateSnapshot::initial()) } else { worker_runtime::execution::WorkerExecutionResult::accepted( worker_runtime::execution::WorkerExecutionOperation::Input, ) - .with_worker_state(protocol::WorkerStateSnapshot::initial(1)) + .with_worker_state(protocol::WorkerStateSnapshot::initial()) } } } @@ -31548,11 +31548,7 @@ mod tests { protocol::subscription::SubscriptionWorkerProtocolMethod { subscription_id: second_protocol_subscription_id, method: protocol::Method::Resume { - command: protocol::WorkerCommandEnvelope { - command_id: 1, - expected_execution_generation: 1, - expected_worker_state_revision: 0, - }, + command: protocol::WorkerCommandEnvelope { command_id: 1 }, }, }, ), diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index ad44b18e..b574ce0a 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -18,7 +18,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore; use crate::{Error, Result}; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 60; +const LATEST_SCHEMA_VERSION: i64 = 61; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings"; const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit"; @@ -35,6 +35,7 @@ const REMOVE_WORKDIR_CACHE_GENERATION_MIGRATION_NAME: &str = const WORKDIR_CREDENTIAL_CANDIDATE_SNAPSHOT_MIGRATION_NAME: &str = "Workdir create credential candidate snapshots"; const RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME: &str = "guarded Runtime removal operations"; +const REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME: &str = "remove obsolete Worker run generation"; const MIGRATIONS: &[Migration] = &[ Migration { @@ -87,6 +88,11 @@ const MIGRATIONS: &[Migration] = &[ name: RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME, apply: migrate_runtime_removal_operations_v59_to_v60, }, + Migration { + version: 61, + name: REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME, + apply: migrate_worker_run_generation_v60_to_v61, + }, ]; #[derive(Clone, Copy)] @@ -9767,11 +9773,22 @@ fn migrate_runtime_removal_operations_v59_to_v60(conn: &Connection) -> Result<() BEGIN SELECT RAISE(ABORT, 'runtime_removal_in_progress'); END; "#, )?; + tx.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", + params![60, RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME], + )?; + tx.commit()?; + Ok(()) +} + +fn migrate_worker_run_generation_v60_to_v61(conn: &Connection) -> Result<()> { + let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; + tx.execute_batch("ALTER TABLE worker_removal_operations DROP COLUMN run_generation;")?; tx.execute( "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", params![ LATEST_SCHEMA_VERSION, - RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME + REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME ], )?; tx.commit()?; @@ -10779,6 +10796,8 @@ mod tests { DROP TRIGGER workdir_removal_insert_blocked_by_runtime_removal; \ DROP TRIGGER workdir_removal_update_blocked_by_runtime_removal; \ DROP TABLE runtime_removal_operations; \ + ALTER TABLE worker_removal_operations \ + ADD COLUMN run_generation INTEGER NOT NULL DEFAULT 0; \ DELETE FROM __yoi_schema_migrations; \ INSERT INTO __yoi_schema_migrations(version, name) \ VALUES (59, 'workspace schema baseline');", @@ -10818,6 +10837,50 @@ mod tests { .unwrap(); } + #[test] + fn schema_v60_upgrade_removes_worker_run_generation_column() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("server.db"); + drop(SqliteWorkspaceStore::open(&path).unwrap()); + { + let conn = Connection::open(&path).unwrap(); + conn.execute_batch( + "ALTER TABLE worker_removal_operations \ + ADD COLUMN run_generation INTEGER NOT NULL DEFAULT 0; \ + DELETE FROM __yoi_schema_migrations; \ + INSERT INTO __yoi_schema_migrations(version,name) \ + VALUES (60,'workspace schema baseline');", + ) + .unwrap(); + } + + drop(SqliteWorkspaceStore::open(&path).unwrap()); + let conn = Connection::open(&path).unwrap(); + assert_eq!( + current_schema_version(&conn).unwrap(), + LATEST_SCHEMA_VERSION + ); + let columns = conn + .prepare("PRAGMA table_info(worker_removal_operations)") + .unwrap() + .query_map([], |row| row.get::<_, String>(1)) + .unwrap() + .collect::>>() + .unwrap(); + assert!(!columns.iter().any(|column| column == "run_generation")); + assert_eq!( + conn.query_row("PRAGMA quick_check", [], |row| row.get::<_, String>(0)) + .unwrap(), + "ok" + ); + let foreign_key_failures: i64 = conn + .query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(foreign_key_failures, 0); + } + #[test] fn schema_v59_runtime_removal_migration_rolls_back_partial_ddl_and_marker() { let temp = tempfile::tempdir().unwrap(); @@ -10948,6 +11011,8 @@ mod tests { DROP TRIGGER workdir_removal_insert_blocked_by_runtime_removal; DROP TRIGGER workdir_removal_update_blocked_by_runtime_removal; DROP TABLE runtime_removal_operations; + ALTER TABLE worker_removal_operations + ADD COLUMN run_generation INTEGER NOT NULL DEFAULT 0; DROP INDEX workspace_signing_identity_audit_workspace_idx; DROP TABLE workspace_signing_identity_audit; DROP TABLE workspace_signing_identity_provisioning_operations; @@ -11097,6 +11162,10 @@ mod tests { version: 60, name: RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME.to_string(), }, + WorkspaceSchemaMigrationStep { + version: 61, + name: REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME.to_string(), + }, ] ); @@ -11143,6 +11212,10 @@ mod tests { 60, RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME.to_string(), ), + ( + 61, + REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME.to_string(), + ), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -11213,7 +11286,7 @@ mod tests { .iter() .map(|migration| migration.version) .collect::>(), - vec![52, 53, 54, 55, 56, 57, 58, 59, 60] + vec![52, 53, 54, 55, 56, 57, 58, 59, 60, 61] ); SqliteWorkspaceStore::migrate_database(&path).unwrap(); let conn = Connection::open(&path).unwrap(); @@ -11221,7 +11294,7 @@ mod tests { current_schema_version(&conn).unwrap(), LATEST_SCHEMA_VERSION ); - assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 11); + assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 12); } #[test] diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 621fa566..8e1f8ed7 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -10,16 +10,11 @@ export type CompletionKind = "file"; export type WorkerStatus = "idle" | "running" | "paused" | "stopped"; -export type WorkerCommandEnvelope = { -/** - * Caller-owned sequence. A controller accepts command ids in strictly - * increasing order for one execution generation. - */ -command_id: number, expected_execution_generation: number, expected_worker_state_revision: number, }; +export type WorkerCommandEnvelope = { command_id: number, }; export type WorkerCommandKind = "resume" | "cancel" | "pause" | "compact" | "shutdown"; -export type WorkerCommandDisposition = "accepted" | "stale_execution_generation" | "stale_worker_state_revision" | "stale_command_id" | "conflict" | "invalid_state"; +export type WorkerCommandDisposition = "accepted" | "stale_command_id" | "conflict" | "invalid_state"; export type WorkerCommandAcknowledgement = { command_id: number, command: WorkerCommandKind, disposition: WorkerCommandDisposition, /** @@ -35,9 +30,9 @@ export type WorkerBusyState = { "kind": "run", "state": WorkerRunState } | { "ki export type WorkerState = { "kind": "idle" } | { "kind": "busy", "state": WorkerBusyState }; -export type WorkerStateSnapshot = { execution_generation: number, revision: number, +export type WorkerStateSnapshot = { /** - * Highest lifecycle command id observed by this controller generation. + * Highest lifecycle command id observed by this controller instance. */ last_command_id: number, state: WorkerState, }; diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index d8d27253..665427f8 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -21,8 +21,6 @@ declare const Deno: { function workerState(status: WorkerStatus): WorkerStateSnapshot { return { - execution_generation: 1, - revision: status === "idle" ? 0 : 1, last_command_id: 0, state: status === "idle" ? { kind: "idle" } @@ -218,17 +216,17 @@ Deno.test("console routing projects live errors but not completion replies", () ); }); -Deno.test("Worker state events and acknowledgements apply monotonically", () => { +Deno.test("Worker state events and acknowledgements replace the full state", () => { const projector = createConsoleProjector(); const running: WorkerStateSnapshot = { - execution_generation: 4, - revision: 3, last_command_id: 2, state: { kind: "busy", state: { kind: "run", state: "running" } }, }; + const freshIdle: WorkerStateSnapshot = { + last_command_id: 0, + state: { kind: "idle" }, + }; const paused: WorkerStateSnapshot = { - ...running, - revision: 4, last_command_id: 3, state: { kind: "busy", state: { kind: "run", state: "paused" } }, }; @@ -238,11 +236,8 @@ Deno.test("Worker state events and acknowledgements apply monotonically", () => event: { event: "worker_state", data: { snapshot: running } }, }, { - eventId: "stale", - event: { - event: "worker_state", - data: { snapshot: { ...running, revision: 2, state: { kind: "idle" } } }, - }, + eventId: "fresh-idle", + event: { event: "worker_state", data: { snapshot: freshIdle } }, }, { eventId: "pause-ack", @@ -263,18 +258,17 @@ Deno.test("Worker state events and acknowledgements apply monotonically", () => assertEquals(projection.status, "paused"); projection = projector.append([{ - eventId: "conflict", + eventId: "replacement", event: { event: "worker_state", - data: { snapshot: { ...paused, state: { kind: "idle" } } }, + data: { snapshot: freshIdle }, }, }]); - assertEquals(projection.workerState, paused); + assertEquals(projection.workerState, freshIdle); + assertEquals(projection.status, "idle"); assert( - projection.lines.some((line) => - line.eventId === "conflict:worker-state-conflict" && line.error - ), - "conflicting equal-version snapshots must fail closed", + !projection.lines.some((line) => line.eventId?.includes("worker-state-conflict")), + "full snapshots must not be rejected by a client-side version comparison", ); }); diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index 7d4e0b55..e8e99008 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -784,58 +784,12 @@ function refreshCompactionActivity( return changed ? { ...projection, lines } : projection; } -function workerStateEqual(left: WorkerState, right: WorkerState): boolean { - if (left.kind !== right.kind) return false; - if (left.kind === "idle" || right.kind === "idle") return true; - return left.state.kind === right.state.kind && - left.state.state === right.state.state; -} - -function workerStateSnapshotEqual( - left: WorkerStateSnapshot, - right: WorkerStateSnapshot, -): boolean { - return left.execution_generation === right.execution_generation && - left.revision === right.revision && - left.last_command_id === right.last_command_id && - workerStateEqual(left.state, right.state); -} - function applyWorkerStateSnapshot( projection: ConsoleProjection, incoming: WorkerStateSnapshot, - eventId: string, ): void { - const current = projection.workerState; - if (!current) { - projection.workerState = incoming; - projection.status = workerStatusFromState(incoming); - return; - } - const generationOrder = incoming.execution_generation - - current.execution_generation; - const revisionOrder = incoming.revision - current.revision; - if (generationOrder > 0 || (generationOrder === 0 && revisionOrder > 0)) { - projection.workerState = incoming; - projection.status = workerStatusFromState(incoming); - return; - } - if (generationOrder < 0 || (generationOrder === 0 && revisionOrder < 0)) { - return; - } - if (!workerStateSnapshotEqual(current, incoming)) { - projection.lines.push( - line( - `${eventId}:worker-state-conflict`, - "error", - "error ยท internal", - `worker state stream rejected: conflicting snapshots at generation ${incoming.execution_generation} revision ${incoming.revision}`, - undefined, - false, - true, - ), - ); - } + projection.workerState = incoming; + projection.status = workerStatusFromState(incoming); } export function applyProtocolEvent( @@ -1004,7 +958,7 @@ export function applyProtocolEvent( }; } } - applyWorkerStateSnapshot(next, event.data.state, envelope.eventId); + applyWorkerStateSnapshot(next, event.data.state); break; } case "internal_worker": { @@ -1053,14 +1007,10 @@ export function applyProtocolEvent( break; } case "worker_state": - applyWorkerStateSnapshot(next, event.data.snapshot, envelope.eventId); + applyWorkerStateSnapshot(next, event.data.snapshot); break; case "command_acknowledged": - applyWorkerStateSnapshot( - next, - event.data.acknowledgement.state, - envelope.eventId, - ); + applyWorkerStateSnapshot(next, event.data.acknowledgement.state); break; case "command": applyCommandEvent(next, envelope.eventId, event.data.event); diff --git a/web/workspace/src/lib/workspace/console/run-status.test.ts b/web/workspace/src/lib/workspace/console/run-status.test.ts index 0063a242..e0d99047 100644 --- a/web/workspace/src/lib/workspace/console/run-status.test.ts +++ b/web/workspace/src/lib/workspace/console/run-status.test.ts @@ -76,7 +76,6 @@ Deno.test("new invoke and running snapshot reset run activity", () => { entries: [], greeting: { text: "", profile: "" }, state: { - execution_generation: 1, revision: 0, last_command_id: 0, state: { kind: "idle" }, diff --git a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts index 94f43ba6..3b6629ae 100644 --- a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts +++ b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts @@ -794,8 +794,9 @@ Deno.test("Worker Console route resolves logical Worker authority before Runtime ) && consolePage.includes('sendWorkerControl("cancel")') && consolePage.includes("lifecycleMethod(command)") && - consolePage.includes("expected_worker_state_revision") && - consolePage.includes("expected_execution_generation") && + consolePage.includes("command_id: commandId") && + !consolePage.includes("expected_worker_state_revision") && + !consolePage.includes("expected_execution_generation") && consolePage.includes("onsubmit={handleComposerSubmit}") && consolePage.includes("disabled={!composerEditable}") && consolePage.includes("class:stop={workerRunning}") && diff --git a/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts b/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts index 1c43e43e..3aa55fce 100644 --- a/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts +++ b/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts @@ -37,8 +37,6 @@ function worker( Deno.test('Worker list state uses the authoritative live snapshot separately from lifecycle', () => { const active = worker('runtime-a', 'worker-1', 1); active.worker_state = { - execution_generation: 4, - revision: 2, last_command_id: 1, state: { kind: 'busy', state: { kind: 'run', state: 'paused' } }, }; diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte index 5b5ff42d..28f3cbd3 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte @@ -562,8 +562,6 @@ nextWorkerCommandId = commandId + 1; const envelope = { command_id: commandId, - expected_execution_generation: state.execution_generation, - expected_worker_state_revision: state.revision, }; switch (command) { case "pause":