feat: format incorporated runtime baseline

crates/worker-runtime/src/execution.rs

@@ -208,7 +208,9 @@ impl WorkerExecutionHandle {
 /// Runtime hooks available to an execution backend for one Worker.
 #[cfg(feature = "ws-server")]
 type WorkerObservationPublisher = Arc<
-    dyn Fn(WorkerRef, protocol::Event) -> Result<WorkerObservationEvent, RuntimeError> + Send + Sync,
+    dyn Fn(WorkerRef, protocol::Event) -> Result<WorkerObservationEvent, RuntimeError>
+        + Send
+        + Sync,
 >;
 
 #[cfg(feature = "ws-server")]
@@ -279,12 +281,10 @@ impl WorkerExecutionContext {
                 .map_err(|_| RuntimeError::StatePoisoned)?;
             match &*state {
                 WorkerExecutionPublicationState::Pending(_) => {
-                    let WorkerExecutionPublicationState::Pending(events) =
+                    let WorkerExecutionPublicationState::Pending(events) = std::mem::replace(
-                        std::mem::replace(
+                        &mut *state,
-                            &mut *state,
+                        WorkerExecutionPublicationState::Activating(VecDeque::new()),
-                            WorkerExecutionPublicationState::Activating(VecDeque::new()),
+                    ) else {
-                        )
-                    else {
                         unreachable!()
                     };
                     *state = WorkerExecutionPublicationState::Activating(events);
@@ -662,10 +662,7 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
         Vec::new()
     }
 
-    fn stop_worker_operation(
+    fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult {
-        &self,
-        request: WorkerExecutionStopRequest,
-    ) -> WorkerExecutionResult {
         let Some(handle) = request.handle.as_ref() else {
             return WorkerExecutionResult::unsupported(
                 WorkerExecutionOperation::Stop,

crates/worker-runtime/src/fs_store.rs

@@ -134,7 +134,8 @@ impl FsRuntimeStore {
 
     #[cfg(test)]
     pub(crate) fn fail_next_worker_write(&self, fault: AtomicWriteFault) {
-        self.worker_write_fault.store(fault as u8, Ordering::Release);
+        self.worker_write_fault
+            .store(fault as u8, Ordering::Release);
     }
 
     pub(crate) fn open_or_create(
@@ -967,12 +968,13 @@ fn migrate_worker_record(
     identity_path: &Path,
 ) -> Result<WorkerAggregateRecord, RuntimeError> {
     if source_schema_version == PREVIOUS_SCHEMA_VERSION {
-        let identity: WorkerIdentityRecord = serde_json::from_value(identity_document).map_err(|error| {
+        let identity: WorkerIdentityRecord =
-            runtime_store_corrupt(
+            serde_json::from_value(identity_document).map_err(|error| {
-                identity_path,
+                runtime_store_corrupt(
-                format!("decode schema-v8 Worker identity: {error}"),
+                    identity_path,
-            )
+                    format!("decode schema-v8 Worker identity: {error}"),
-        })?;
+                )
+            })?;
         identity.validate_for_schema(identity_path, PREVIOUS_SCHEMA_VERSION)?;
         let execution_path = identity_path.with_file_name(WORKER_EXECUTION_FILE);
         let execution: WorkerExecutionRecord =
@@ -987,12 +989,13 @@ fn migrate_worker_record(
         mapping,
         identity_path,
     )?;
-    let identity: WorkerIdentityRecord = serde_json::from_value(documents.identity).map_err(|error| {
+    let identity: WorkerIdentityRecord =
-        runtime_store_corrupt(
+        serde_json::from_value(documents.identity).map_err(|error| {
-            identity_path,
+            runtime_store_corrupt(
-            format!("decode validated schema-v8 Worker identity: {error}"),
+                identity_path,
-        )
+                format!("decode validated schema-v8 Worker identity: {error}"),
-    })?;
+            )
+        })?;
     let execution_path = identity_path.with_file_name(WORKER_EXECUTION_FILE);
     let execution: WorkerExecutionRecord =
         serde_json::from_value(documents.execution).map_err(|error| {
@@ -1965,8 +1968,9 @@ impl WorkerAggregateRecord {
                     return Err(RuntimeError::StoreCorrupt {
                         operation: "read Worker lifecycle reconciliation",
                         path: path.to_path_buf(),
-                        message: "Worker reconciliation status does not match the last settled status"
+                        message:
-                            .to_string(),
+                            "Worker reconciliation status does not match the last settled status"
+                                .to_string(),
                     });
                 }
                 WorkerExecutionRecord {
@@ -2971,7 +2975,11 @@ mod tests {
         let committed: WorkerAggregateRecord =
             read_bounded_json(&aggregate_path, "read test Worker aggregate").unwrap();
         assert_eq!(committed.created_at_ms, Some(2));
-        assert!(!aggregate_path.with_file_name(WORKER_EXECUTION_FILE).exists());
+        assert!(
+            !aggregate_path
+                .with_file_name(WORKER_EXECUTION_FILE)
+                .exists()
+        );
     }
 
     #[test]
@@ -3050,10 +3058,7 @@ mod tests {
         )
         .unwrap();
 
-        assert_eq!(
+        assert_eq!(migrated.identity["schema_version"], PREVIOUS_SCHEMA_VERSION);
-            migrated.identity["schema_version"],
-            PREVIOUS_SCHEMA_VERSION
-        );
         assert!(migrated.identity.get("request").is_none());
         assert!(migrated.identity.get("execution").is_none());
         assert_eq!(
@@ -3158,8 +3163,11 @@ mod tests {
                 .contains_key(&worker_id)
         );
         assert!(!worker_dir.join(WORKER_EXECUTION_FILE).exists());
-        let aggregate: serde_json::Value =
+        let aggregate: serde_json::Value = read_json(
-            read_json(&worker_dir.join(WORKER_FILE), "read migrated Worker aggregate").unwrap();
+            &worker_dir.join(WORKER_FILE),
+            "read migrated Worker aggregate",
+        )
+        .unwrap();
         assert_eq!(aggregate["schema_version"], SCHEMA_VERSION);
         assert_eq!(aggregate["execution_state"]["state"], "available");
         assert_eq!(
@@ -3192,10 +3200,7 @@ mod tests {
         )
         .unwrap();
 
-        assert_eq!(
+        assert_eq!(migrated.identity["schema_version"], PREVIOUS_SCHEMA_VERSION);
-            migrated.identity["schema_version"],
-            PREVIOUS_SCHEMA_VERSION
-        );
         assert_eq!(migrated.identity["status"], "running");
         assert!(migrated.identity.get("request").is_none());
         assert!(migrated.identity.get("execution").is_none());

crates/worker-runtime/src/runtime.rs

@@ -1775,13 +1775,12 @@ impl Runtime {
         }
 
         if !reconcile
-            && let Err(error) =
+            && let Err(error) = self.begin_restore_operation(
-                self.begin_restore_operation(
+                worker_ref,
-                    worker_ref,
+                operation_id,
-                    operation_id,
+                operation_mode,
-                    operation_mode,
+                candidate_context.clone(),
-                    candidate_context.clone(),
+            )
-                )
         {
             if matches!(error, RuntimeError::StoreCommitOutcomeUnknown { .. }) {
                 return Ok(RuntimeWorkerRestoreResult::failed(
@@ -1813,8 +1812,7 @@ impl Runtime {
                 );
                 match commit {
                     Ok(worker) => {
-                        if let Err(error) =
+                        if let Err(error) = backend.activate_restored_worker(operation_id, &handle)
-                            backend.activate_restored_worker(operation_id, &handle)
                         {
                             tracing::error!(
                                 worker_id = %worker_ref.worker_id,
@@ -1865,7 +1863,8 @@ impl Runtime {
                                     error = %error,
                                     "Worker restore commit failed and was rolled back"
                                 );
-                                if let Err(settle_error) = self.settle_restore_rollback(worker_ref) {
+                                if let Err(settle_error) = self.settle_restore_rollback(worker_ref)
+                                {
                                     tracing::error!(
                                         worker_id = %worker_ref.worker_id,
                                         error = %settle_error,
@@ -1983,11 +1982,8 @@ impl Runtime {
                             "failed to persist restore execution evidence; in-process reconciliation remains required"
                         );
                     }
-                } else if let Err(retain_error) =
+                } else if let Err(retain_error) = self
-                    self.retain_restore_reconciliation_pending(
+                    .retain_restore_reconciliation_pending(worker_ref, candidate_context.clone())
-                        worker_ref,
-                        candidate_context.clone(),
-                    )
                 {
                     tracing::error!(
                         worker_id = %worker_ref.worker_id,
@@ -2002,10 +1998,9 @@ impl Runtime {
                 ))
             }
             WorkerExecutionSpawnResult::Errored(_result) => {
-                if let Err(retain_error) = self.retain_restore_reconciliation_pending(
+                if let Err(retain_error) = self
-                    worker_ref,
+                    .retain_restore_reconciliation_pending(worker_ref, candidate_context.clone())
-                    candidate_context.clone(),
+                {
-                ) {
                     tracing::error!(
                         worker_id = %worker_ref.worker_id,
                         error = %retain_error,
@@ -3110,7 +3105,11 @@ impl Runtime {
                     message: "runtime has no execution backend".to_string(),
                 }
             })?;
-            (backend, worker.execution_handle.clone(), pending.operation_id)
+            (
+                backend,
+                worker.execution_handle.clone(),
+                pending.operation_id,
+            )
         };
         let result = backend.stop_worker_operation(WorkerExecutionStopRequest {
             operation_id,
@@ -3725,37 +3724,36 @@ impl RuntimeState {
                     None,
                     None,
                 ),
-                PersistedWorkerExecutionState::ReconciliationRequired(operation) => {
+                PersistedWorkerExecutionState::ReconciliationRequired(operation) => match operation
-                    match operation {
+                {
-                        PersistedWorkerLifecycleOperation::Restore(operation) => {
+                    PersistedWorkerLifecycleOperation::Restore(operation) => {
-                            let restore_intent =
+                        let restore_intent =
-                                restore_intent_for_status(operation.last_settled_status);
+                            restore_intent_for_status(operation.last_settled_status);
-                            (
+                        (
-                                Some(operation.request),
-                                true,
-                                operation.binding.is_some(),
-                                restore_intent,
-                                Some(PendingWorkerRestore {
-                                    operation_id: operation.operation_id,
-                                    mode: operation.mode.into(),
-                                    last_settled_status: operation.last_settled_status,
-                                }),
-                                None,
-                            )
-                        }
-                        PersistedWorkerLifecycleOperation::Stop(operation) => (
                             Some(operation.request),
                             true,
                             operation.binding.is_some(),
-                            operation.restore_intent,
+                            restore_intent,
-                            None,
+                            Some(PendingWorkerRestore {
-                            Some(PendingWorkerStop {
                                 operation_id: operation.operation_id,
+                                mode: operation.mode.into(),
                                 last_settled_status: operation.last_settled_status,
                             }),
-                        ),
+                            None,
+                        )
                     }
-                }
+                    PersistedWorkerLifecycleOperation::Stop(operation) => (
+                        Some(operation.request),
+                        true,
+                        operation.binding.is_some(),
+                        operation.restore_intent,
+                        None,
+                        Some(PendingWorkerStop {
+                            operation_id: operation.operation_id,
+                            last_settled_status: operation.last_settled_status,
+                        }),
+                    ),
+                },
                 PersistedWorkerExecutionState::Unavailable => (
                     None,
                     false,
@@ -4782,8 +4780,7 @@ impl WorkerRecord {
                     last_settled_status: pending.last_settled_status,
                 }),
             )
-        } else if let (Some(pending), Some(request)) =
+        } else if let (Some(pending), Some(request)) = (self.pending_restore, self.request.clone())
-            (self.pending_restore, self.request.clone())
         {
             PersistedWorkerExecutionState::ReconciliationRequired(
                 PersistedWorkerLifecycleOperation::Restore(PersistedWorkerRestoreOperation {
@@ -7397,11 +7394,8 @@ mod tests {
             display_name: None,
         };
         let backend = Arc::new(TestExecutionBackend::default());
-        let runtime = Runtime::with_fs_store_and_execution_backend(
+        let runtime =
-            options.clone(),
+            Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap();
-            backend.clone(),
-        )
-        .unwrap();
         runtime.store_config_bundle(test_bundle()).unwrap();
         let worker = runtime
             .create_worker(task_request("reconcile pending stop"))
@@ -7409,9 +7403,7 @@ mod tests {
         runtime_store(&runtime)
             .fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename);
 
-        let error = runtime
+        let error = runtime.stop_worker(&worker.worker_ref, None).unwrap_err();
-            .stop_worker(&worker.worker_ref, None)
-            .unwrap_err();
         assert!(matches!(
             error,
             RuntimeError::StoreCommitOutcomeUnknown { .. }
@@ -7480,11 +7472,9 @@ mod tests {
         drop(reconciled);
 
         let restarted_backend = Arc::new(TestExecutionBackend::default());
-        let restarted = Runtime::with_fs_store_and_execution_backend(
+        let restarted =
-            options,
+            Runtime::with_fs_store_and_execution_backend(options, restarted_backend.clone())
-            restarted_backend.clone(),
+                .unwrap();
-        )
-        .unwrap();
         assert_eq!(*restarted_backend.stop_count.lock().unwrap(), 0);
         assert_eq!(*restarted_backend.restore_count.lock().unwrap(), 0);
         assert_eq!(
@@ -7645,19 +7635,20 @@ mod tests {
         let cursor_before = runtime
             .worker_observation_cursor_now(&worker.worker_ref)
             .unwrap();
-        backend.restore_event.lock().unwrap().replace(
+        backend
-            protocol::Event::WorkerState {
+            .restore_event
+            .lock()
+            .unwrap()
+            .replace(protocol::Event::WorkerState {
                 snapshot: protocol::WorkerStatus::Running.into(),
-            },
+            });
-        );
         let gate = Arc::new(RestoreGate::default());
         *backend.restore_gate.lock().unwrap() = Some(gate.clone());
 
         let restoring_runtime = runtime.clone();
         let restoring_ref = worker.worker_ref.clone();
-        let restoring = std::thread::spawn(move || {
+        let restoring =
-            restoring_runtime.restore_worker_operation(&restoring_ref)
+            std::thread::spawn(move || restoring_runtime.restore_worker_operation(&restoring_ref));
-        });
         assert!(gate.wait_for_entered(1, std::time::Duration::from_secs(2)));
 
         let pending = runtime.worker_detail(&worker.worker_ref).unwrap();
@@ -8928,12 +8919,16 @@ mod tests {
             .create_worker(task_request("rolled back restore"))
             .unwrap();
         runtime.stop_worker(&worker.worker_ref, None).unwrap();
-        backend.restore_result.lock().unwrap().replace(
+        backend
-            WorkerExecutionSpawnResult::RolledBack(WorkerExecutionResult::errored(
+            .restore_result
-                WorkerExecutionOperation::Restore,
+            .lock()
-                "controller launch failed and cleanup completed",
+            .unwrap()
-            )),
+            .replace(WorkerExecutionSpawnResult::RolledBack(
-        );
+                WorkerExecutionResult::errored(
+                    WorkerExecutionOperation::Restore,
+                    "controller launch failed and cleanup completed",
+                ),
+            ));
 
         let result = runtime
             .restore_worker_operation(&worker.worker_ref)
@@ -9149,16 +9144,22 @@ mod tests {
         .unwrap();
         let workers = restored.list_workers().unwrap();
         assert_eq!(workers.len(), 1);
-        assert!(workers.iter().all(|summary| summary.worker_id != worker.worker_id));
+        assert!(
+            workers
+                .iter()
+                .all(|summary| summary.worker_id != worker.worker_id)
+        );
         assert!(matches!(
             restored.worker_detail(&worker.worker_ref),
             Err(RuntimeError::WorkerNotFound { .. })
         ));
-        assert!(restored
+        assert!(
-            .diagnostics()
+            restored
-            .unwrap()
+                .diagnostics()
-            .iter()
+                .unwrap()
-            .any(|diagnostic| diagnostic.code == "worker_record_unavailable"));
+                .iter()
+                .any(|diagnostic| diagnostic.code == "worker_record_unavailable")
+        );
 
         let restored_detail = restored
             .restore_worker(&restorable.worker_ref)

crates/worker-runtime/src/worker_backend.rs

@@ -1442,8 +1442,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
         }
 
         let workspace_client = worker.workspace_client_handle();
-        let run_dir =
+        let run_dir = self.worker_restore_run_dir(&request.worker_ref, request.operation_id)?;
-            self.worker_restore_run_dir(&request.worker_ref, request.operation_id)?;
         let bash_output_dir = bash_output_dir_for_worker_id(&request.worker_ref.worker_id);
         let started = PreparedWorker::new(
             worker,
@@ -1483,11 +1482,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
         workspace_id: Option<&str>,
         handle: &WorkerHandle,
     ) {
-        self.observation_hub.register(
+        self.observation_hub
-            worker_ref.clone(),
+            .register(worker_ref.clone(), workspace_id.map(str::to_string), handle);
-            workspace_id.map(str::to_string),
-            handle,
-        );
     }
 }
 
@@ -2953,13 +2949,10 @@ where
         self.send_method(WorkerExecutionOperation::ProtocolMethod, worker, method)
     }
 
-    fn stop_worker_operation(
+    fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult {
-        &self,
+        let handle = request
-        request: WorkerExecutionStopRequest,
+            .handle
-    ) -> WorkerExecutionResult {
+            .unwrap_or_else(|| WorkerExecutionHandle::new(request.worker_ref, self.backend_id()));
-        let handle = request.handle.unwrap_or_else(|| {
-            WorkerExecutionHandle::new(request.worker_ref, self.backend_id())
-        });
         self.stop_worker(&handle)
     }
 
@@ -3865,9 +3858,7 @@ mod tests {
             worker_ref,
             Arc::new(|worker_ref, payload| {
                 Ok(crate::observation::WorkerObservationEvent::new(
-                    1,
+                    1, worker_ref, payload,
-                    worker_ref,
-                    payload,
                 ))
             }),
         )
@@ -4894,11 +4885,9 @@ mod tests {
             observed_workspace_clients: observed_workspace_clients.clone(),
         };
         let backend = Arc::new(WorkerRuntimeExecutionBackend::new(factory).unwrap());
-        let runtime = EmbeddedRuntime::with_execution_backend(
+        let runtime =
-            RuntimeOptions::default(),
+            EmbeddedRuntime::with_execution_backend(RuntimeOptions::default(), backend.clone())
-            backend.clone(),
+                .unwrap();
-        )
-        .unwrap();
         runtime.store_config_bundle(test_bundle()).unwrap();
         let request = create_request("idempotent restore reconciliation");
         let worker = runtime.create_worker(request.clone()).unwrap();

crates/workspace-server/src/hosts.rs

@@ -2222,8 +2222,7 @@ impl EmbeddedWorkerRuntime {
                 identity: "runtime_registry_worker".to_string(),
                 workspace_id: summary.workspace_id.clone(),
             },
-            availability:
+            availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
-                protocol::subscription::SubscriptionWorkerAvailability::Observed,
             state: embedded_worker_state_label(
                 summary.status,
                 summary.execution_metadata_available,
@@ -2271,8 +2270,7 @@ impl EmbeddedWorkerRuntime {
                 identity: "runtime_registry_worker".to_string(),
                 workspace_id: detail.workspace_id.clone(),
             },
-            availability:
+            availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
-                protocol::subscription::SubscriptionWorkerAvailability::Observed,
             state: embedded_worker_state_label(detail.status, detail.execution_metadata_available)
                 .to_string(),
             worker_state: detail.worker_state.clone(),
@@ -4064,8 +4062,7 @@ impl RemoteWorkerRuntime {
                 identity: "runtime_registry_worker".to_string(),
                 workspace_id: summary.workspace_id.clone(),
             },
-            availability:
+            availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
-                protocol::subscription::SubscriptionWorkerAvailability::Observed,
             state: embedded_worker_state_label(
                 summary.status,
                 summary.execution_metadata_available,
@@ -4111,8 +4108,7 @@ impl RemoteWorkerRuntime {
                 identity: "runtime_registry_worker".to_string(),
                 workspace_id: detail.workspace_id.clone(),
             },
-            availability:
+            availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
-                protocol::subscription::SubscriptionWorkerAvailability::Observed,
             state: embedded_worker_state_label(detail.status, detail.execution_metadata_available)
                 .to_string(),
             worker_state: detail.worker_state.clone(),
@@ -5979,18 +5975,14 @@ mod tests {
 
     #[test]
     fn embedded_store_commit_unknown_diagnostic_is_path_free() {
-        let diagnostic = embedded_runtime_diagnostic(
+        let diagnostic =
-            &EmbeddedRuntimeError::StoreCommitOutcomeUnknown {
+            embedded_runtime_diagnostic(&EmbeddedRuntimeError::StoreCommitOutcomeUnknown {
                 operation: "sync Worker aggregate directory",
                 path: std::path::PathBuf::from("/private/runtime/workers/worker.json"),
                 source: std::io::Error::other("raw fsync failure"),
-            },
+            });
-        );
 
-        assert_eq!(
+        assert_eq!(diagnostic.code, "embedded_runtime_store_commit_unknown");
-            diagnostic.code,
-            "embedded_runtime_store_commit_unknown"
-        );
         assert_eq!(diagnostic.severity, HostDiagnosticSeverity::Error);
         assert!(!diagnostic.message.contains("/private/runtime"));
         assert!(!diagnostic.message.contains("raw fsync failure"));
@@ -6547,8 +6539,7 @@ mod tests {
                         identity: host_id.to_string(),
                         workspace_id: None,
                     },
-                    availability:
+                    availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
-                        protocol::subscription::SubscriptionWorkerAvailability::Observed,
                     state: "available".to_string(),
                     worker_state: None,
                     last_seen_at: None,