diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index d53035cce..cd76170d3 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -913,6 +913,38 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { Err("execution backend cannot confirm Subject Worker shutdown".into()) } + /// Retain the durable Runtime exclusion in the adapter even when no Stop + /// can yet be admitted (for example an unresolved failed create). + fn fence_subject_delete_worker( + &self, + _request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + Err("execution backend cannot fence Subject Worker admission".into()) + } + + /// Check current resources after a previously confirmed Subject deletion. + /// Runtime separately verifies its durable fence and physical originals; + /// this does not issue a Stop or establish historical termination. + fn confirm_subject_delete_worker_absent( + &self, + _request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + Err("execution backend cannot confirm Subject Worker resource absence".into()) + } + + /// Internal Runtime consumer: the durable fence and original Stop child are + /// owned by Runtime. A saved canonical receipt (settled or pending cleanup) + /// is read-only execution evidence, not permission to stop a replacement or + /// mint another operation. Runtime alone settles a still-pending Stop. + fn subject_delete_worker_stop( + &self, + _request: &runtime_api::SubjectDeleteWorkerRequest, + _stop: WorkerExecutionStopRequest, + _canonical: Option<&WorkerExecutionTermination>, + ) -> Result { + Err("execution backend cannot reconcile Subject Worker Stop evidence".into()) + } + /// Must reserve the actual Controller, not infer Idle from a catalog snapshot. fn subject_release( &self, @@ -1217,6 +1249,22 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { Vec::new() } + /// Internal pending-Stop recovery. Saved scope termination does not prove + /// factory cleanup; backends must reconcile both under their Worker lock. + fn reconcile_stop_worker_operation( + &self, + request: WorkerExecutionStopRequest, + saved: Option<&WorkerExecutionTermination>, + ) -> WorkerExecutionResult { + if saved.is_some() { + return WorkerExecutionResult::errored( + WorkerExecutionOperation::Stop, + "execution backend cannot reconcile saved Stop termination and cleanup", + ); + } + self.stop_worker_operation(request) + } + fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult { self.stop_worker(&request.worker_ref) } @@ -1297,11 +1345,28 @@ impl WorkerExecutionBackendRef { self.backend.preflight_subject_delete_worker(request) } - pub(crate) fn subject_delete_worker( + pub(crate) fn fence_subject_delete_worker( &self, request: &runtime_api::SubjectDeleteWorkerRequest, ) -> Result<(), String> { - self.backend.subject_delete_worker(request) + self.backend.fence_subject_delete_worker(request) + } + + pub(crate) fn confirm_subject_delete_worker_absent( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + self.backend.confirm_subject_delete_worker_absent(request) + } + + pub(crate) fn subject_delete_worker_stop( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + stop: WorkerExecutionStopRequest, + canonical: Option<&WorkerExecutionTermination>, + ) -> Result { + self.backend + .subject_delete_worker_stop(request, stop, canonical) } pub(crate) fn subject_release( @@ -1539,6 +1604,14 @@ impl WorkerExecutionBackendRef { .worker_completions(worker_ref, kind, prefix, context) } + pub(crate) fn reconcile_stop_worker_operation( + &self, + request: WorkerExecutionStopRequest, + saved: Option<&WorkerExecutionTermination>, + ) -> WorkerExecutionResult { + self.backend.reconcile_stop_worker_operation(request, saved) + } + pub(crate) fn stop_worker_operation( &self, request: WorkerExecutionStopRequest, diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 9126f3c94..2d59cee2b 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -10,6 +10,8 @@ //! Removal reserves only its target under the Runtime state mutex, releasing it before //! filesystem/Git inspection and unlink; conflicting target use is rejected promptly. +#[cfg(test)] +mod controller_fixture; mod runtime_management_api; #[cfg(test)] mod workdir_close_tests; @@ -3485,6 +3487,30 @@ mod tests { struct AcceptingBackend; + impl AcceptingBackend { + fn fixture_stop_result( + request: crate::execution::WorkerExecutionStopRequest, + ) -> WorkerExecutionResult { + // Only execution is substituted in these HTTP/retention fixtures. + // Model its bound stop report explicitly; do not relax Runtime's + // production requirement or claim real Controller/Host coverage. + let execution_id = + uuid::Uuid::parse_str(&request.worker_ref.worker_id.to_string()).unwrap(); + let mut result = WorkerExecutionResult::accepted(WorkerExecutionOperation::Stop); + result.execution_id = Some(execution_id); + result.termination = Some(crate::execution::WorkerExecutionTermination { + scope: crate::execution::WorkerExecutionTerminationScope::Controller, + worker_ref: request.worker_ref, + stop_operation_id: request.operation_id, + execution_id, + shutdown_command_id: None, + submissions: Vec::new(), + feature_failures: Vec::new(), + }); + result + } + } + impl WorkerExecutionBackend for AcceptingBackend { fn preflight_subject_delete_worker( &self, @@ -3498,6 +3524,35 @@ mod tests { ) -> Result<(), String> { Ok(()) } + fn fence_subject_delete_worker( + &self, + _request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + // This replaces admission exclusion only, not scope termination. + Ok(()) + } + fn confirm_subject_delete_worker_absent( + &self, + _request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + // This stateless execution substitute owns no Controller or resources. + // Runtime separately checks its confirmed durable fence and originals; + // this boundary check supplies no historical termination evidence. + Ok(()) + } + fn subject_delete_worker_stop( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + stop: crate::execution::WorkerExecutionStopRequest, + canonical: Option<&crate::execution::WorkerExecutionTermination>, + ) -> Result { + if stop.worker_ref.worker_id != request.worker_id { + return Err("fixture Subject Stop Worker mismatch".into()); + } + // Runtime supplies the original Stop child; canonical evidence must + // match that child unchanged, not be retagged for another operation. + Ok(self.reconcile_stop_worker_operation(stop, canonical)) + } fn recover_subject_create( &self, _request: &runtime_api::SubjectCreateRecoveryRequest, @@ -3557,42 +3612,88 @@ mod tests { WorkerExecutionResult::accepted(WorkerExecutionOperation::Stop) } + fn termination_receipt( + &self, + _worker_ref: &WorkerRef, + ) -> Result, String> { + // No receipt is retained by this unit fixture. Without a supplied + // original, deterministic construction alone is not Stop evidence. + Ok(None) + } + + fn reconcile_termination_receipt( + &self, + worker_ref: &WorkerRef, + saved: &crate::execution::WorkerExecutionTermination, + ) -> Result, String> { + let expected = + Self::fixture_stop_result(crate::execution::WorkerExecutionStopRequest { + worker_ref: worker_ref.clone(), + operation_id: saved.stop_operation_id, + }); + // Compare the whole bound fixture receipt, including scope, execution, + // shutdown command, submissions and Feature failures, not just IDs. + Ok(expected + .termination + .filter(|receipt| receipt == saved) + .map(|_| saved.clone())) + } + + fn reconcile_stop_worker_operation( + &self, + request: crate::execution::WorkerExecutionStopRequest, + saved: Option<&crate::execution::WorkerExecutionTermination>, + ) -> WorkerExecutionResult { + let Some(saved) = saved else { + return self.stop_worker_operation(request); + }; + let mut result = Self::fixture_stop_result(request); + if result.termination.as_ref() != Some(saved) { + return WorkerExecutionResult::errored( + WorkerExecutionOperation::Stop, + "canonical Stop does not match the whole bound fixture receipt", + ); + } + result.termination = Some(saved.clone()); + result + } + fn stop_worker_operation( &self, request: crate::execution::WorkerExecutionStopRequest, ) -> WorkerExecutionResult { - // Only execution is substituted in these HTTP/retention fixtures. - // Model its bound stop report explicitly; do not relax Runtime's - // production requirement or claim real Controller/Host coverage. - let execution_id = - uuid::Uuid::parse_str(&request.worker_ref.worker_id.to_string()).unwrap(); - let mut result = self.stop_worker(&request.worker_ref); - result.execution_id = Some(execution_id); - result.termination = Some(crate::execution::WorkerExecutionTermination { - scope: crate::execution::WorkerExecutionTerminationScope::Controller, - worker_ref: request.worker_ref, - stop_operation_id: request.operation_id, - execution_id, - shutdown_command_id: None, - submissions: Vec::new(), - feature_failures: Vec::new(), - }); - result + Self::fixture_stop_result(request) } } #[cfg(feature = "fs-store")] - #[derive(Default)] struct RejectingRestoreBackend { + inner: Arc, restores: Mutex>, inputs: Mutex>, - stops: Mutex>, + stops: Mutex>, + stop_results: Mutex>, + canonical_queries: Mutex>, + } + + #[cfg(feature = "fs-store")] + impl RejectingRestoreBackend { + fn new(inner: Arc) -> Self { + Self { + inner, + restores: Mutex::new(Vec::new()), + inputs: Mutex::new(Vec::new()), + stops: Mutex::new(Vec::new()), + stop_results: Mutex::new(Vec::new()), + canonical_queries: Mutex::new(Vec::new()), + } + } } #[cfg(feature = "fs-store")] impl WorkerExecutionBackend for RejectingRestoreBackend { fn backend_id(&self) -> &str { - "http-test" + self.inner.backend_id() } fn spawn_worker( @@ -3625,9 +3726,34 @@ mod tests { ) } - fn stop_worker(&self, worker_ref: &WorkerRef) -> WorkerExecutionResult { - self.stops.lock().unwrap().push(worker_ref.clone()); - WorkerExecutionResult::accepted(WorkerExecutionOperation::Stop) + fn stop_worker_operation( + &self, + request: crate::execution::WorkerExecutionStopRequest, + ) -> WorkerExecutionResult { + self.stops.lock().unwrap().push(request.clone()); + let result = self.inner.stop_worker_operation(request); + self.stop_results.lock().unwrap().push(result.clone()); + result + } + + fn stop_worker(&self, _worker_ref: &WorkerRef) -> WorkerExecutionResult { + panic!("HTTP lifecycle must retain its typed Stop operation"); + } + + fn termination_receipt( + &self, + worker_ref: &WorkerRef, + ) -> Result, String> { + self.inner.termination_receipt(worker_ref) + } + + fn reconcile_termination_receipt( + &self, + worker_ref: &WorkerRef, + saved: &crate::execution::WorkerExecutionTermination, + ) -> Result, String> { + self.canonical_queries.lock().unwrap().push(saved.clone()); + self.inner.reconcile_termination_receipt(worker_ref, saved) } } @@ -4240,34 +4366,52 @@ mod tests { #[cfg(feature = "fs-store")] #[tokio::test] - async fn rest_saved_active_worker_restore_rejection_is_observable_and_can_be_stopped_removed() { - let temp = tempfile::tempdir().unwrap(); - let options = crate::fs_store::FsRuntimeStoreOptions::new(temp.path().join("runtime")); - let runtime = Runtime::with_fs_store_and_execution_backend( - options.clone(), - Arc::new(AcceptingBackend), - ) - .unwrap(); - runtime - .store_config_bundle(test_bundle(ProfileSelector::Builtin( - "builtin:coder".to_string(), - ))) + async fn rest_saved_active_worker_restore_rejection_retains_unproven_stop_and_source() { + let mut fixture = super::controller_fixture::ControllerFixture::persisted(); + let created = fixture.worker.clone(); + let options = + crate::fs_store::FsRuntimeStoreOptions::new(fixture.root_path().join("runtime")) + .with_runtime_id("http-controller-fixture"); + let mut input = WorkerInput::user("materialize the original saved Session"); + input.submission_request_id = Some("http-cold-original".into()); + fixture + .runtime() + .send_input(&created.worker_ref, input) .unwrap(); - let mut request = task_request("saved active Worker"); - request.workspace_api = Some(WorkspaceApiRef { - workspace_id: "local".to_string(), - base_url: "http://127.0.0.1:8787".to_string(), - }); - let created = runtime - .create_worker_scoped(&RuntimeWorkspaceScope::new("local", "local-token"), request) - .unwrap(); - assert_eq!(created.status, crate::catalog::WorkerStatus::Idle); - assert!(created.worker_state.is_some()); - drop(runtime); + // Reclaim the actual seed Controller without saving a Runtime canonical + // Stop. The cold adapter must not infer this historical join from absence. + let raw_stop = + fixture + .backend() + .stop_worker_operation(crate::execution::WorkerExecutionStopRequest { + worker_ref: created.worker_ref.clone(), + operation_id: crate::execution::WorkerLifecycleOperationId::new(), + }); + assert!(raw_stop.is_accepted(), "{raw_stop:?}"); + super::controller_fixture::assert_real_termination( + &created.worker_ref, + raw_stop.termination.as_ref().unwrap(), + ); + let saved_session = fixture.saved_session(); + assert!( + fixture + .runtime() + .execution_termination(&created.worker_ref) + .unwrap() + .is_none() + ); + fixture.assert_persisted_request(fixture.runtime()); + fixture.close_catalog(); - let backend = Arc::new(RejectingRestoreBackend::default()); + let backend = Arc::new(RejectingRestoreBackend::new(fixture.empty_backend())); + assert!( + backend + .termination_receipt(&created.worker_ref) + .unwrap() + .is_none() + ); let runtime = - Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap(); + Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap(); // Startup attempted automatic restore without turning rejection into a // fabricated live Worker or silently changing its saved lifecycle status. assert_eq!( @@ -4302,7 +4446,7 @@ mod tests { Method::POST, &format!("{worker_uri}/restore"), token, - &runtime.test_restore_request(&created.worker_ref), + &fixture.restore_request(&runtime), ) .await; assert_eq!(response.status(), StatusCode::OK); @@ -4388,39 +4532,172 @@ mod tests { token, ) .await; - assert_eq!(response.status(), StatusCode::OK); - let stopped: RuntimeHttpWorkerLifecycleResponse = read_json(response).await; - assert_eq!(stopped.ack.worker_ref, created.worker_ref); - assert_eq!(stopped.ack.status, crate::catalog::WorkerStatus::Stopped); - assert_eq!( - *backend.stops.lock().unwrap(), - vec![created.worker_ref.clone()] + assert_eq!(response.status(), StatusCode::CONFLICT); + let error: RuntimeHttpErrorResponse = read_json(response).await; + assert_eq!(error.error.code, "worker_execution_rejected"); + let stops = backend.stops.lock().unwrap(); + assert_eq!(stops.len(), 1); + assert_eq!(stops[0].worker_ref, created.worker_ref); + let original_stop_id = stops[0].operation_id; + drop(stops); + assert!( + backend + .stop_results + .lock() + .unwrap() + .iter() + .all(|result| result.termination.is_none()) + ); + assert!( + runtime + .execution_termination(&created.worker_ref) + .unwrap() + .is_none() ); let response = authed_empty_request(app.clone(), Method::GET, &worker_uri, token).await; assert_eq!(response.status(), StatusCode::OK); let detail: RuntimeHttpWorkerResponse = read_json(response).await; - assert_eq!(detail.worker.status, crate::catalog::WorkerStatus::Stopped); + assert_ne!(detail.worker.status, crate::catalog::WorkerStatus::Stopped); + let response = authed_empty_request(app.clone(), Method::DELETE, &worker_uri, token).await; + assert_eq!( + response.status(), + StatusCode::BAD_REQUEST, + "pending unproven Stop must forbid removal" + ); + let error: RuntimeHttpErrorResponse = read_json(response).await; + assert_eq!(error.error.code, "invalid_request"); + assert_eq!( + backend.stops.lock().unwrap().len(), + 1, + "DELETE must not mint a replacement Stop" + ); + let record = runtime + .test_runtime_store() + .read_worker_record(&created.worker_ref) + .unwrap(); + let crate::fs_store::PersistedWorkerExecutionState::ReconciliationRequired( + crate::fs_store::PersistedWorkerLifecycleOperation::Stop(pending), + ) = record.execution_state + else { + panic!("unproven Stop must retain its durable owner"); + }; + assert_eq!(pending.operation_id, original_stop_id); + assert_eq!(pending.request.as_ref().unwrap(), fixture.create_request()); + let document: serde_json::Value = serde_json::from_slice( + &std::fs::read( + options + .root + .join("workers") + .join(created.worker_id.to_string()) + .join("worker.json"), + ) + .unwrap(), + ) + .unwrap(); + assert!(document["restore_guard"]["execution_termination"].is_null()); + assert_eq!( + fixture.saved_session(), + saved_session, + "cold refusal must preserve the original Session" + ); + let response = authed_empty_request(app, Method::GET, &worker_uri, token).await; + assert_eq!(response.status(), StatusCode::OK); + } + + #[cfg(feature = "fs-store")] + #[tokio::test] + async fn rest_cold_stopped_worker_deletion_reuses_real_canonical_stop_without_launch() { + let mut fixture = super::controller_fixture::ControllerFixture::persisted(); + let created = fixture.worker.clone(); + let mut input = WorkerInput::user("persist a real Session before the canonical Stop"); + input.submission_request_id = Some("http-canonical-original".into()); + fixture + .runtime() + .send_input(&created.worker_ref, input) + .unwrap(); + let token = "local-token"; + let worker_uri = format!("/v1/workers/{}", created.worker_id); + let app = runtime_http_router(fixture.runtime().clone(), token.into()); + let response = authed_empty_request( + app.clone(), + Method::POST, + &format!("{worker_uri}/stop"), + token, + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let stopped: RuntimeHttpWorkerLifecycleResponse = read_json(response).await; + assert_eq!(stopped.ack.status, crate::catalog::WorkerStatus::Stopped); + let original = fixture + .runtime() + .execution_termination(&created.worker_ref) + .unwrap() + .unwrap(); + super::controller_fixture::assert_real_termination(&created.worker_ref, &original); + assert_eq!( + fixture + .backend() + .termination_receipt(&created.worker_ref) + .unwrap(), + Some(original.clone()) + ); + let saved_session = fixture.saved_session(); + let options = + crate::fs_store::FsRuntimeStoreOptions::new(fixture.root_path().join("runtime")) + .with_runtime_id("http-controller-fixture"); + drop(app); + fixture.assert_persisted_request(fixture.runtime()); + fixture.close_catalog(); + let backend = Arc::new(RejectingRestoreBackend::new(fixture.empty_backend())); + assert!( + backend + .termination_receipt(&created.worker_ref) + .unwrap() + .is_none() + ); + let runtime = + Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap(); + let app = runtime_http_router(runtime.clone(), token.into()); + assert_eq!( + runtime.worker_detail(&created.worker_ref).unwrap().status, + crate::catalog::WorkerStatus::Stopped + ); + assert_eq!( + runtime.execution_termination(&created.worker_ref).unwrap(), + Some(original.clone()) + ); + fixture.assert_persisted_request(&runtime); + assert_eq!(fixture.saved_session(), saved_session); let response = authed_empty_request(app.clone(), Method::DELETE, &worker_uri, token).await; assert_eq!(response.status(), StatusCode::OK); let removed: RuntimeHttpWorkerDeleteResponse = read_json(response).await; assert_eq!(removed.worker.worker_id, created.worker_id); assert!(removed.worker.deleted); - // Removal asks the backend to reconfirm shutdown even for a saved - // Stopped Worker; it must use the same WorkerRef as the original stop. - assert_eq!( - *backend.stops.lock().unwrap(), - vec![created.worker_ref.clone(), created.worker_ref.clone()] + assert!( + backend.restores.lock().unwrap().is_empty(), + "Stopped catalog must not launch a restore" ); assert!( - !temp - .path() - .join("runtime/workers") + backend.stops.lock().unwrap().is_empty(), + "deletion must read the original Stop, not issue another" + ); + assert!( + backend + .canonical_queries + .lock() + .unwrap() + .iter() + .any(|saved| saved == &original) + ); + assert!( + !options + .root + .join("workers") .join(created.worker_id.to_string()) .exists() ); - let response = authed_empty_request(app, Method::GET, &worker_uri, token).await; assert_eq!(response.status(), StatusCode::NOT_FOUND); } @@ -5640,42 +5917,7 @@ mod ws_tests { )); } - #[tokio::test] - async fn protocol_ws_stop_closes_old_stream_and_restore_requires_new_connection() { - let (runtime, backend, worker_ref, url) = spawn_runtime_server().await; - let (mut old_stream, _) = connect_async(authed_ws_request(&url)).await.unwrap(); - let _ = next_frame(&mut old_stream).await; - runtime.stop_worker(&worker_ref, None).unwrap(); - runtime - .restore_worker(&worker_ref, runtime.test_restore_request(&worker_ref)) - .unwrap(); - backend.publish( - &worker_ref, - protocol::Event::TextDelta { - text: "new execution".into(), - }, - ); - let closed = tokio::time::timeout(std::time::Duration::from_secs(5), old_stream.next()) - .await - .expect("the old protocol connection must close on stop"); - assert!(matches!(closed, Some(Ok(Message::Close(_))) | None)); - - let (mut new_stream, _) = connect_async(authed_ws_request(&url)).await.unwrap(); - assert!(matches!( - next_frame(&mut new_stream).await, - protocol::Event::Snapshot { .. } - )); - backend.publish( - &worker_ref, - protocol::Event::TextDone { - text: "restored execution".into(), - }, - ); - assert!(matches!( - next_frame(&mut new_stream).await, - protocol::Event::TextDone { text } if text == "restored execution" - )); - } + include!("http_server/controller_ws_tests.rs"); #[tokio::test] async fn protocol_ws_cursor_resume_is_duplicate_safe_and_filters_workers() { @@ -5713,43 +5955,6 @@ mod ws_tests { )); } - #[tokio::test] - async fn protocol_ws_old_cursor_never_replays_stopped_execution_after_new_snapshot() { - let (runtime, backend, worker_ref, url) = spawn_runtime_server().await; - let old = backend.publish( - &worker_ref, - protocol::Event::TextDelta { - text: "old cursor".into(), - }, - ); - backend.publish( - &worker_ref, - protocol::Event::TextDone { - text: "old output".into(), - }, - ); - backend.publish(&worker_ref, protocol::Event::Shutdown); - runtime.stop_worker(&worker_ref, None).unwrap(); - runtime - .restore_worker(&worker_ref, runtime.test_restore_request(&worker_ref)) - .unwrap(); - let resume_url = format!("{url}?cursor={}", old.cursor); - let (mut stream, _) = connect_async(authed_ws_request(&resume_url)).await.unwrap(); - assert!(matches!( - next_frame(&mut stream).await, - protocol::Event::Snapshot { .. } - )); - backend.publish( - &worker_ref, - protocol::Event::TextDone { - text: "new output".into(), - }, - ); - assert!( - matches!(next_frame(&mut stream).await, protocol::Event::TextDone { text } if text == "new output") - ); - } - #[tokio::test] async fn protocol_ws_reports_malformed_cursor_and_method_frame() { let (_runtime, _backend, _worker_ref, url) = spawn_runtime_server().await; diff --git a/crates/worker-runtime/src/http_server/controller_fixture.rs b/crates/worker-runtime/src/http_server/controller_fixture.rs new file mode 100644 index 000000000..ee7937d92 --- /dev/null +++ b/crates/worker-runtime/src/http_server/controller_fixture.rs @@ -0,0 +1,515 @@ +//! HTTP/WS integration support: real Profile spawn, saved Session restore and +//! execution-owned paired protocol streams. Model I/O is scripted; WS cases also +//! substitute the external Workspace projection source with isolated loopback HTTP +//! through the production RuntimeWorkspaceRequestClient. Controller Submit/refresh, +//! restore, report/join and protocol streams remain production implementations. +//! Uses the ordinary lib suite and shared isolated restore-environment preflight. + +use super::*; +use crate::catalog::{ProfileSelector, WorkerDetail}; +use crate::config_bundle::{ConfigBundleMetadata, ConfigBundleProvenance, ConfigProfileDescriptor}; +use crate::execution::{WorkerExecutionBackend, WorkerExecutionTermination}; +use crate::management::RuntimeOptions; +use crate::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend}; +use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent}; +use agen::llm_client::{ClientError, LlmClient, Request as ModelRequest}; +use async_trait::async_trait; +use futures::Stream; +use session_store::{Store, WorkerAggregateStore, WorkerMetadataStore, WorkerSessionStore}; +use std::collections::BTreeMap; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; + +pub(super) const OLD_MARKER: &str = "controller-original-output"; +pub(super) const NEW_MARKER: &str = "controller-restored-output"; +pub(super) type Backend = WorkerRuntimeExecutionBackend; + +#[derive(Clone, Default)] +struct ScriptedModel { + calls: Arc, +} + +#[async_trait] +impl LlmClient for ScriptedModel { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + async fn stream( + &self, + _request: ModelRequest, + ) -> Result> + Send>>, ClientError> + { + let marker = match self.calls.fetch_add(1, Ordering::SeqCst) { + 0 => OLD_MARKER, + 1 => NEW_MARKER, + call => panic!("unexpected model call {call}; fixture permits exactly two Submits"), + }; + Ok(Box::pin(futures::stream::iter( + vec![ + LlmEvent::text_block_start(0), + LlmEvent::text_delta(0, marker), + LlmEvent::text_block_stop(0, None), + LlmEvent::Status(StatusEvent { + status: ResponseStatus::Completed, + }), + ] + .into_iter() + .map(Ok), + ))) + } +} + +// A narrow external source, not a refresh/cache/authorization bypass. Keep the +// original Worker WorkspaceApiRef (including its stored URL) untouched; only the +// explicitly injected request client's transport endpoint is substituted. This +// source does not claim to test Workspace Server authentication. +#[cfg(feature = "ws-server")] +struct ProjectionSource { + client: crate::workspace_request::RuntimeWorkspaceRequestClient, + server: tokio::task::JoinHandle<()>, + requests: Arc, +} + +#[cfg(feature = "ws-server")] +impl ProjectionSource { + fn new(root: &std::path::Path) -> Self { + const ROUTE: &str = "/api/w/local/config/projections/prompts"; + let catalog = worker::PromptCatalog::builtins_only() + .unwrap() + .projection() + .clone(); + let source_digest = if catalog.source_digest.is_empty() { + catalog.catalog_digest.clone() + } else { + catalog.source_digest.clone() + }; + let projection = worker::WorkspacePromptProjection::new( + "local", + source_digest, + catalog.catalog_digest.clone(), + catalog, + ) + .unwrap(); + let path = root.join("workspace-prompt-projection.json"); + std::fs::write(&path, serde_json::to_vec(&projection).unwrap()).unwrap(); + let body = std::fs::read_to_string(path).unwrap(); + let saved: worker::WorkspacePromptProjection = serde_json::from_str(&body).unwrap(); + saved.validate().unwrap(); + assert_eq!(saved, projection); + let _: server_api::WorkspacePromptProjection = serde_json::from_str(&body).unwrap(); + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let listener = tokio::net::TcpListener::from_std(listener).unwrap(); + let requests = Arc::new(AtomicUsize::new(0)); + let received = requests.clone(); + let app = axum::Router::new().fallback(move |request: axum::extract::Request| { + let body = body.clone(); + let received = received.clone(); + async move { + if request.method() != axum::http::Method::GET + || request.uri().to_string() != ROUTE + || request + .headers() + .get("x-yoi-runtime-id") + .and_then(|v| v.to_str().ok()) + != Some("http-controller-fixture") + { + return Err(axum::http::StatusCode::FORBIDDEN); + } + if axum::body::to_bytes(request.into_body(), 0).await.is_err() { + return Err(axum::http::StatusCode::BAD_REQUEST); + } + received.fetch_add(1, Ordering::SeqCst); + Ok(([("content-type", "application/json")], body)) + } + }); + Self { + client: crate::workspace_request::RuntimeWorkspaceRequestClient::new( + "local", + base_url, + "http-controller-fixture", + ), + server: tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }), + requests, + } + } +} + +#[cfg(feature = "ws-server")] +impl Drop for ProjectionSource { + fn drop(&mut self) { + self.server.abort(); + } +} + +// Own the catalog, adapter and loopback tasks until real Stop has joined the +// Controller and its relays. Unwind cleanup never creates termination evidence. +pub(super) struct ControllerFixture { + runtime: Option, + backend: Option>, + pub worker: WorkerDetail, + create_request: CreateWorkerRequest, + workspace_scope: RuntimeWorkspaceScope, + server: Option>, + #[cfg(feature = "ws-server")] + projection_source: Option, + model: ScriptedModel, + root: tempfile::TempDir, +} + +impl ControllerFixture { + fn root() -> tempfile::TempDir { + crate::restore_test_environment::require_restore_test_environment(); + let target = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .unwrap() + .parent() + .unwrap() + .join("target"); + tempfile::Builder::new() + .prefix("hc") + .rand_bytes(3) + .tempdir_in(target) + .unwrap() + } + + fn make_backend( + root: &std::path::Path, + model: ScriptedModel, + workspace_client: Option, + ) -> Arc { + let mut factory = ProfileRuntimeWorkerFactory::new(root) + .with_runtime_store_dir(root.join("runtime")) + .with_runtime_id("http-controller-fixture") + .with_spawn_model_client_factory(move || Box::new(model.clone())); + if let Some(client) = workspace_client { + factory = factory.with_workspace_request_client(client); + } + Arc::new(WorkerRuntimeExecutionBackend::new(factory).unwrap()) + } + + fn bundle() -> ConfigBundle { + let archive = crate::profile_archive::ProfileSourceArchive::build( + crate::profile_archive::ProfileSourceArchiveInput { + id: "http-controller-profile".into(), + entrypoints: BTreeMap::from([( + "builtin:companion".into(), + "profiles/controller.dcdl".into(), + )]), + imports: BTreeMap::new(), + sources: BTreeMap::from([( + "profiles/controller.dcdl".into(), + r#"{ + slug = "companion"; + description = "HTTP Controller fixture"; + scope = "workspace_read"; + model = { + scheme = "anthropic"; + model_id = "test-model"; + auth = { kind = "none"; }; + }; + engine = { max_tokens = 100; }; + }"# + .into(), + )]), + }, + ) + .unwrap(); + ConfigBundle { + metadata: ConfigBundleMetadata { + id: "http-controller-bundle".into(), + digest: String::new(), + workspace_id: "local".into(), + created_at: "test".into(), + provenance: ConfigBundleProvenance { + source: "test".into(), + detail: None, + }, + }, + profiles: vec![ConfigProfileDescriptor { + selector: ProfileSelector::Builtin("builtin:companion".into()), + label: None, + }], + declarations: Vec::new(), + prompt_catalog: None, + profile_source_archive: Some(archive), + profile_source_archive_handle: None, + } + .with_computed_digest() + } + + fn create( + runtime: Runtime, + backend: Arc, + root: tempfile::TempDir, + model: ScriptedModel, + ) -> Self { + let bundle = Self::bundle(); + let request = CreateWorkerRequest { + plugin_execution: None, + worker_id: WorkerId::now_v7(), + create_fingerprint: "http-controller-create".into(), + profile: ProfileSelector::Builtin("builtin:companion".into()), + display_name: None, + profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded { + archive: bundle.profile_source_archive.clone().unwrap(), + }, + config_bundle: Some(ConfigBundleRef { + id: bundle.metadata.id.clone(), + digest: bundle.metadata.digest.clone(), + }), + initial_input: None, + workdir_attachment_requests: Vec::new(), + workdir_attachments: Vec::new(), + worker_observation_enabled: false, + worker_observation_grants: Vec::new(), + workspace_api: Some(crate::catalog::WorkspaceApiRef { + workspace_id: "local".into(), + base_url: "http://127.0.0.1:1".into(), + }), + memory_settings: Some(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: "local".into(), + language: "English".into(), + }), + subjektiv_attached: false, + backend_job: None, + }; + runtime.store_config_bundle(bundle).unwrap(); + let workspace_scope = RuntimeWorkspaceScope::new("local", "local-token"); + let worker = runtime + .create_worker_scoped(&workspace_scope, request.clone()) + .unwrap(); + Self { + runtime: Some(runtime), + backend: Some(backend), + worker, + create_request: request, + workspace_scope, + server: None, + #[cfg(feature = "ws-server")] + projection_source: None, + model, + root, + } + } + + #[cfg(feature = "ws-server")] + pub fn memory() -> Self { + let root = Self::root(); + let model = ScriptedModel::default(); + let projection_source = ProjectionSource::new(root.path()); + let backend = Self::make_backend( + root.path(), + model.clone(), + Some(projection_source.client.clone()), + ); + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), backend.clone()).unwrap(); + let mut fixture = Self::create(runtime, backend, root, model); + fixture.projection_source = Some(projection_source); + fixture + } + + #[cfg(feature = "fs-store")] + pub fn persisted() -> Self { + let root = Self::root(); + let model = ScriptedModel::default(); + let backend = Self::make_backend(root.path(), model.clone(), None); + let runtime = Runtime::with_fs_store_and_execution_backend( + crate::fs_store::FsRuntimeStoreOptions::new(root.path().join("runtime")) + .with_runtime_id("http-controller-fixture"), + backend.clone(), + ) + .unwrap(); + Self::create(runtime, backend, root, model) + } + + pub fn runtime(&self) -> &Runtime { + self.runtime.as_ref().unwrap() + } + pub fn backend(&self) -> &Arc { + self.backend.as_ref().unwrap() + } + pub fn root_path(&self) -> &std::path::Path { + self.root.path() + } + pub fn create_request(&self) -> &CreateWorkerRequest { + &self.create_request + } + + pub fn restore_request(&self, runtime: &Runtime) -> runtime_api::WorkerRestoreRequest { + let detail = runtime.worker_detail(&self.worker.worker_ref).unwrap(); + let api = self.create_request.workspace_api.as_ref().unwrap(); + runtime_api::WorkerRestoreRequest { + expected_observation_token: detail.restore_observation_token.unwrap(), + request_id: uuid::Uuid::now_v7().to_string(), + preparation: Some(runtime_api::WorkerRestorePreparation { + workspace_api: Some(runtime_api::WorkspaceApiRef { + workspace_id: api.workspace_id.clone(), + base_url: api.base_url.clone(), + }), + ..Default::default() + }), + } + } + + #[cfg(feature = "ws-server")] + pub fn restore(&self) -> WorkerDetail { + self.runtime() + .restore_worker_scoped( + &self.workspace_scope, + &self.worker.worker_ref, + self.restore_request(self.runtime()), + ) + .unwrap() + } + + // Read the original aggregate through the production decoder and identity + // validator before discarding its catalog; never repair saved wire data. + #[cfg(feature = "fs-store")] + pub fn assert_persisted_request(&self, runtime: &Runtime) { + let record = runtime + .test_runtime_store() + .read_worker_record(&self.worker.worker_ref) + .unwrap(); + assert_eq!( + record.workspace_id.as_deref(), + Some(self.workspace_scope.workspace_id.as_str()) + ); + let crate::fs_store::PersistedWorkerExecutionState::Available(execution) = + record.execution_state + else { + panic!("seed catalog must retain its original available execution request"); + }; + assert_eq!(execution.request, self.create_request); + } + + // Only use after a real raw Stop or a settled Runtime Stop, never to imply + // that dropping the old adapter proves shutdown to its replacement. + #[cfg(feature = "fs-store")] + pub fn close_catalog(&mut self) { + self.runtime.take(); + self.backend.take(); + } + + #[cfg(feature = "fs-store")] + pub fn empty_backend(&self) -> Arc { + Self::make_backend(self.root.path(), self.model.clone(), None) + } + + pub fn saved_session(&self) -> (session_store::SessionId, session_store::SegmentId) { + let aggregate = self + .root + .path() + .join("runtime/workers") + .join(self.worker.worker_id.to_string()); + let name = + ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&self.worker.worker_ref); + let metadata = WorkerAggregateStore::open_read_only(&aggregate, name.clone()) + .unwrap() + .read_by_name(&name) + .unwrap() + .unwrap(); + let active = metadata + .active + .expect("real spawn must save Worker metadata"); + let segment = active + .segment_id + .expect("real Submit must materialize the saved Session"); + let entries = WorkerSessionStore::open_read_only(aggregate.join("session")) + .unwrap() + .read_all(active.session_id, segment) + .unwrap(); + assert!( + !entries.is_empty(), + "saved Session must contain actual Controller log entries" + ); + (active.session_id, segment) + } + + pub fn stop(&self) -> WorkerExecutionTermination { + self.runtime() + .stop_worker(&self.worker.worker_ref, None) + .unwrap(); + let receipt = self + .runtime() + .execution_termination(&self.worker.worker_ref) + .unwrap() + .unwrap(); + assert_real_termination(&self.worker.worker_ref, &receipt); + assert_eq!( + self.backend() + .termination_receipt(&self.worker.worker_ref) + .unwrap(), + Some(receipt.clone()) + ); + receipt + } + + pub fn assert_model_calls(&self, expected: usize) { + assert_eq!(self.model.calls.load(Ordering::SeqCst), expected); + } + + #[cfg(feature = "ws-server")] + pub async fn finish(&mut self) -> WorkerExecutionTermination { + let receipt = self.stop(); + if let Some(server) = self.server.take() { + server.abort(); + let _ = server.await; + } + if let Some(mut source) = self.projection_source.take() { + assert!( + source.requests.load(Ordering::SeqCst) > 0, + "real Submit must fetch the saved projection through the production client" + ); + source.server.abort(); + let _ = (&mut source.server).await; + } + receipt + } + + #[cfg(feature = "ws-server")] + pub async fn serve(&mut self) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let runtime = self.runtime().clone(); + self.server = Some(tokio::spawn(async move { + serve_runtime_http(runtime, listener, Some("local-token".into())) + .await + .unwrap(); + })); + format!( + "ws://{address}/v1/workers/{}/protocol/ws", + self.worker.worker_id + ) + } +} + +pub(super) fn assert_real_termination(worker: &WorkerRef, receipt: &WorkerExecutionTermination) { + assert_eq!( + receipt.scope, + crate::execution::WorkerExecutionTerminationScope::Controller + ); + assert!(receipt.confirms_scope_shutdown(worker, receipt.stop_operation_id)); + assert!( + receipt.shutdown_command_id.is_some(), + "real Controller must acknowledge Shutdown" + ); + assert!(receipt.feature_failures.is_empty(), "{receipt:?}"); +} + +impl Drop for ControllerFixture { + fn drop(&mut self) { + if let Some(runtime) = &self.runtime { + let _ = runtime.stop_worker(&self.worker.worker_ref, None); + } + if let Some(server) = self.server.take() { + server.abort(); + } + self.runtime.take(); + self.backend.take(); + } +} diff --git a/crates/worker-runtime/src/http_server/controller_ws_tests.rs b/crates/worker-runtime/src/http_server/controller_ws_tests.rs new file mode 100644 index 000000000..d2d1533eb --- /dev/null +++ b/crates/worker-runtime/src/http_server/controller_ws_tests.rs @@ -0,0 +1,170 @@ +// Included inside ws_tests. These two cases exercise the production Profile +// adapter, saved Session, paired protocol streams and Controller task/report/join; +// the other parser/filter unit cases intentionally keep their small WsBackend. + +use super::controller_fixture::{ControllerFixture, NEW_MARKER, OLD_MARKER}; + +type ControllerSocket = tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, +>; +const CONTROLLER_WAIT: std::time::Duration = std::time::Duration::from_secs(5); + +async fn controller_submit(stream: &mut ControllerSocket, request_id: &str) { + let method = protocol::Method::submit_text(request_id, "produce the scripted reply"); + stream.send(Message::Text(serde_json::to_string(&method).unwrap().into())).await.unwrap(); +} + +fn assert_no_old_operational_output(event: &protocol::Event) { + assert!(!matches!(event, + protocol::Event::TextDelta { text } | protocol::Event::TextDone { text } + if text.contains(OLD_MARKER) + ), "old operational output followed a replacement snapshot: {event:?}"); + assert!(!matches!(event, protocol::Event::Shutdown), "old Shutdown followed a replacement snapshot"); +} + +async fn controller_output( + stream: &mut ControllerSocket, + request_id: &str, + marker: &str, + reject_old: bool, +) { + let mut seen = Vec::new(); + tokio::time::timeout(CONTROLLER_WAIT, async { + let (mut accepted, mut output, mut finished) = (false, false, false); + while !(accepted && output && finished) { + let event = next_frame(stream).await; + seen.push(format!("{event:?}")); + if reject_old { assert_no_old_operational_output(&event); } + assert!(!matches!(&event, + protocol::Event::Error { .. } | protocol::Event::SubmissionRejected { .. } + ), "real Submit was rejected: {event:?}"); + accepted |= matches!(&event, + protocol::Event::SubmissionAccepted { submission_request_id, .. } + if submission_request_id == request_id + ); + output |= matches!(&event, protocol::Event::TextDone { text } if text == marker); + finished |= matches!(&event, + protocol::Event::WorkerState { snapshot } + if snapshot.last_finished_submission_request_id.as_deref() == Some(request_id) + && snapshot.state == protocol::WorkerState::Idle + ); + } + }).await.unwrap_or_else(|_| panic!("real Controller output deadline for {request_id}; seen {seen:?}")); +} + +async fn controller_snapshot(stream: &mut ControllerSocket, saved_output: bool) { + let snapshot = next_frame(stream).await; + let protocol::Event::Snapshot { session, .. } = snapshot else { + panic!("a Controller connection must start with its paired snapshot"); + }; + if saved_output { + assert!(serde_json::to_string(&session).unwrap().contains(OLD_MARKER), + "replacement snapshot must retain the original saved Session history"); + } +} + +async fn drain_original_connection(stream: &mut ControllerSocket) { + let mut seen = Vec::new(); + tokio::time::timeout(CONTROLLER_WAIT, async { + loop { + match stream.next().await { + None | Some(Ok(Message::Close(_))) => break, + Some(Ok(Message::Text(text))) => { + let event: protocol::Event = serde_json::from_str(&text).unwrap(); + assert!(!matches!(&event, + protocol::Event::TextDelta { text } | protocol::Event::TextDone { text } + if text.contains(NEW_MARKER) + ), "replacement output leaked onto the original connection: {event:?}"); + seen.push(format!("{event:?}")); + } + Some(Ok(Message::Ping(_) | Message::Pong(_))) => {} + other => panic!("original connection did not end with Close/EOF: {other:?}"), + } + } + }).await.unwrap_or_else(|_| panic!("original stream close deadline; buffered frames {seen:?}")); +} + +#[tokio::test] +async fn protocol_ws_stop_closes_old_stream_and_restore_requires_new_connection() { + let mut fixture = ControllerFixture::memory(); + let url = fixture.serve().await; + let worker_ref = fixture.worker.worker_ref.clone(); + let original_transport = fixture.backend().clone().attach_worker_protocol(&worker_ref).unwrap(); + assert!(original_transport.validate().is_ok()); + let (mut old_stream, _) = connect_async(authed_ws_request(&url)).await.unwrap(); + controller_snapshot(&mut old_stream, false).await; + controller_submit(&mut old_stream, "ws-original").await; + controller_output(&mut old_stream, "ws-original", OLD_MARKER, false).await; + let original_session = fixture.saved_session(); + let original = fixture.stop(); + assert!(original_transport.validate().is_err(), "original transport must be invalid after actual join"); + + let restored = fixture.restore(); + assert_eq!(restored.status, crate::catalog::WorkerStatus::Idle); + assert_eq!(fixture.saved_session().0, original_session.0); + let replacement_transport = fixture.backend().clone().attach_worker_protocol(&worker_ref).unwrap(); + assert!(replacement_transport.validate().is_ok()); + let stale_stop = original_transport.stop(crate::execution::WorkerExecutionStopRequest { + worker_ref: worker_ref.clone(), operation_id: original.stop_operation_id, + }); + assert!(!stale_stop.is_accepted(), "a captured old transport must not stop the replacement"); + assert!(stale_stop.termination.is_none()); + assert!(replacement_transport.validate().is_ok()); + + let (mut new_stream, _) = connect_async(authed_ws_request(&url)).await.unwrap(); + controller_snapshot(&mut new_stream, true).await; + controller_submit(&mut new_stream, "ws-replacement").await; + controller_output(&mut new_stream, "ws-replacement", NEW_MARKER, true).await; + // Real Stop may leave old state/log/Shutdown frames buffered; only the bound + // scope's report/join is termination evidence, not this socket draining. + drain_original_connection(&mut old_stream).await; + let replacement = fixture.finish().await; + assert_ne!(original.execution_id, replacement.execution_id); + assert_ne!(original.stop_operation_id, replacement.stop_operation_id); + assert!(replacement_transport.validate().is_err()); + fixture.assert_model_calls(2); +} + +#[tokio::test] +async fn protocol_ws_old_cursor_never_replays_stopped_execution_after_new_snapshot() { + let mut fixture = ControllerFixture::memory(); + let url = fixture.serve().await; + let worker_ref = fixture.worker.worker_ref.clone(); + let original_transport = fixture.backend().clone().attach_worker_protocol(&worker_ref).unwrap(); + let mut observation = fixture.runtime().subscribe_worker_observation().unwrap(); + let (mut old_stream, _) = connect_async(authed_ws_request(&url)).await.unwrap(); + controller_snapshot(&mut old_stream, false).await; + controller_submit(&mut old_stream, "cursor-original").await; + controller_output(&mut old_stream, "cursor-original", OLD_MARKER, false).await; + let old = tokio::time::timeout(CONTROLLER_WAIT, async { + loop { + let event = observation.recv().await.expect("real Runtime observation relay"); + if event.worker_ref == worker_ref + && matches!(&event.payload, protocol::Event::TextDone { text } if text == OLD_MARKER) + { + break event; + } + } + }).await.expect("old output must have entered real Runtime observation history"); + assert!(WorkerObservationCursor::decode(&old.cursor).is_some()); + let original_session = fixture.saved_session(); + let original = fixture.stop(); + assert!(original_transport.validate().is_err()); + let restored = fixture.restore(); + assert_eq!(restored.status, crate::catalog::WorkerStatus::Idle); + assert_eq!(fixture.saved_session().0, original_session.0); + let replacement_transport = fixture.backend().clone().attach_worker_protocol(&worker_ref).unwrap(); + assert!(replacement_transport.validate().is_ok()); + + let resume_url = format!("{url}?cursor={}", old.cursor); + let (mut stream, _) = connect_async(authed_ws_request(&resume_url)).await.unwrap(); + controller_snapshot(&mut stream, true).await; + controller_submit(&mut stream, "cursor-replacement").await; + controller_output(&mut stream, "cursor-replacement", NEW_MARKER, true).await; + drain_original_connection(&mut old_stream).await; + let replacement = fixture.finish().await; + assert_ne!(original.execution_id, replacement.execution_id); + assert_ne!(original.stop_operation_id, replacement.stop_operation_id); + assert!(replacement_transport.validate().is_err()); + fixture.assert_model_calls(2); +} diff --git a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs index 50949ff17..c0bebd3f9 100644 --- a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs @@ -1225,7 +1225,8 @@ async fn subject_create_recovery_signed_http_requires_delete_and_matching_worksp } #[tokio::test] -async fn subject_delete_worker_signed_http_requires_delete_and_matching_workspace() { +async fn subject_delete_worker_signed_http_requires_delete_scope_without_inventing_missing_originals() + { let f = Fixture::new(); let identity = RuntimeIdentityMaterial::generate("recovery-key").unwrap(); let trust = workspace_trust(&identity); @@ -1259,16 +1260,23 @@ async fn subject_delete_worker_signed_http_requires_delete_and_matching_workspac let claims = signed_claims(&trust, "POST", path, &body); assert_eq!(claims.operation, "workers:delete"); let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await; - assert_eq!(response.status(), StatusCode::OK); - let bytes = axum::body::to_bytes(response.into_body(), 16384) - .await - .unwrap(); - let receipt: runtime_api::SubjectDeleteWorkerResponse = - serde_json::from_slice(&bytes).unwrap(); - assert_eq!(receipt.worker_id, request.worker_id); - assert_eq!(receipt.workspace_id, request.workspace_id); - assert_eq!(receipt.create_fingerprint, request.create_fingerprint); - assert_eq!(receipt.operation_id, request.operation_id); + // Valid authorization reaches domain validation, but the arbitrary ID + // has neither an original Stop nor a confirmed historical fence. + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + let error: RuntimeHttpErrorResponse = read_json(response).await; + assert_eq!(error.error.code, "invalid_request"); + assert!( + error + .error + .message + .contains("Missing original confirmed Subject shutdown") + ); + assert!(f.runtime.worker_detail(&f.worker).is_ok()); + assert!( + f.runtime + .worker_detail(&WorkerRef::new(request.worker_id)) + .is_err() + ); } } diff --git a/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs index f757b690c..6d5b0e01f 100644 --- a/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs @@ -63,8 +63,7 @@ fn subject_job_removal_purge_partial_physical_failure_keeps_fence_and_retry_eras } #[test] -fn subject_job_removal_purge_unconfirmed_durable_deletion_fence_preserves_copies_until_reconciled() -{ +fn subject_job_removal_purge_unconfirmed_fence_never_invents_removed_execution_proof() { let f = Fixture::new(); let request = f.removal_purge_request(SessionDisposition::Archive); f.complete_removal(&request); @@ -80,8 +79,10 @@ fn subject_job_removal_purge_unconfirmed_durable_deletion_fence_preserves_copies } = f; drop(store); drop(runtime); - // Frozen durable crash-window fixture: permanent admission exclusion was - // saved, but no confirmed execution-shutdown response was committed. + // Retained incomplete-proof fixture: deliberately remove the confirmation + // after removal. Correct current ordering requires that confirmation before + // removal; this is not evidence that such a crash window can occur now. + // The old removal result cannot manufacture the missing Subject proof. let snapshot_path = temp.path().join("runtime.json"); let mut snapshot: serde_json::Value = serde_json::from_slice(&std::fs::read(&snapshot_path).unwrap()).unwrap(); @@ -101,17 +102,16 @@ fn subject_job_removal_purge_unconfirmed_durable_deletion_fence_preserves_copies ); assert_eq!(tree(&temp.path().join("archives")), copies); assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt); - runtime - .subject_delete_worker_scoped(&scope, request.deletion.clone()) - .unwrap(); - runtime - .purge_subject_job_removal_scoped(&scope, request) - .unwrap(); assert!( - !temp - .path() - .join("archives/workers/canonical-archive/session") - .exists() + runtime + .subject_delete_worker_scoped(&scope, request.deletion.clone()) + .is_err() ); + assert!( + runtime + .purge_subject_job_removal_scoped(&scope, request) + .is_err() + ); + assert_eq!(tree(&temp.path().join("archives")), copies); assert_eq!(std::fs::read(receipt_path).unwrap(), receipt); } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index d27d165f2..3b0cc0498 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -4375,16 +4375,31 @@ impl Runtime { StopAdmission::Stopped(ack) => return Ok(ack), StopAdmission::Pending(operation_id) => operation_id, }; - let backend = self.lock()?.execution_backend.clone().ok_or_else(|| { - RuntimeError::WorkerExecutionUnavailable { - worker_id: worker_ref.worker_id, - message: "runtime has no execution backend".into(), - } - })?; - let result = backend.stop_worker_operation(WorkerExecutionStopRequest { - operation_id, - worker_ref: worker_ref.clone(), - }); + let (backend, saved) = { + let mut state = self.lock()?; + state.refresh_uncertain_worker_write(worker_ref)?; + let backend = state.execution_backend.clone().ok_or_else(|| { + RuntimeError::WorkerExecutionUnavailable { + worker_id: worker_ref.worker_id, + message: "runtime has no execution backend".into(), + } + })?; + ( + backend, + state + .worker(worker_ref)? + .restore_guard + .execution_termination + .clone(), + ) + }; + let result = backend.reconcile_stop_worker_operation( + WorkerExecutionStopRequest { + operation_id, + worker_ref: worker_ref.clone(), + }, + saved.as_ref(), + ); self.commit_stopped_worker(worker_ref, operation_id, result) } @@ -5116,8 +5131,9 @@ impl Runtime { &self, worker_ref: &WorkerRef, ) -> Result<(), RuntimeError> { - let (backend, operation_id) = { - let state = self.lock()?; + let (backend, operation_id, saved) = { + let mut state = self.lock()?; + state.refresh_uncertain_worker_write(worker_ref)?; let worker = state.worker(worker_ref)?; let pending = worker.pending_stop.ok_or_else(|| { RuntimeError::InvalidRequest(format!( @@ -5131,12 +5147,19 @@ impl Runtime { message: "runtime has no execution backend".to_string(), } })?; - (backend, pending.operation_id) + ( + backend, + pending.operation_id, + worker.restore_guard.execution_termination.clone(), + ) }; - let result = backend.stop_worker_operation(WorkerExecutionStopRequest { - operation_id, - worker_ref: worker_ref.clone(), - }); + let result = backend.reconcile_stop_worker_operation( + WorkerExecutionStopRequest { + operation_id, + worker_ref: worker_ref.clone(), + }, + saved.as_ref(), + ); self.commit_stopped_worker(worker_ref, operation_id, result)?; Ok(()) } @@ -9980,6 +10003,36 @@ mod tests { Err("shutdown is unconfirmed".into()) } } + fn fence_subject_delete_worker( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + self.subject_delete_worker(request) + } + fn subject_delete_worker_stop( + &self, + _request: &runtime_api::SubjectDeleteWorkerRequest, + stop: WorkerExecutionStopRequest, + canonical: Option<&crate::execution::WorkerExecutionTermination>, + ) -> Result { + // Component substitution only. Real owner/report/join coverage lives + // in subject_delete_backend_tests, never in this response fixture. + if let Some(saved) = canonical { + let mut result = WorkerExecutionResult::accepted(WorkerExecutionOperation::Stop); + result.execution_id = Some(saved.execution_id); + result.termination = Some(saved.clone()); + Ok(result) + } else { + Ok(self.stop_worker_operation(stop)) + } + } + fn confirm_subject_delete_worker_absent( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + // Replaces only the adapter boundary in these persistence tests. + self.subject_delete_worker(request) + } fn recover_subject_create( &self, _request: &runtime_api::SubjectCreateRecoveryRequest, diff --git a/crates/worker-runtime/src/runtime/subject_delete.rs b/crates/worker-runtime/src/runtime/subject_delete.rs index 6dcecaf3d..a14df1e3e 100644 --- a/crates/worker-runtime/src/runtime/subject_delete.rs +++ b/crates/worker-runtime/src/runtime/subject_delete.rs @@ -1,5 +1,49 @@ use super::*; +impl RuntimeState { + // Absence is usable only after this exact Subject operation confirmed the + // original scope and cleanup. A new missing catalog ID is never proof. + fn confirmed_subject_delete_originals_absent( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result { + if self.workers.contains_key(&request.worker_id) { + return Ok(false); + } + if self.unreconciled_worker_writes.contains(&request.worker_id) { + return Err(RuntimeError::InvalidRequest( + "Subject Worker write remains uncertain".into(), + )); + } + self.subject_delete_worker_fences + .get(&request.worker_id) + .filter(|fence| fence.request == *request) + .and_then(|fence| fence.response.as_ref()) + .filter(|saved| { + saved.workspace_id == request.workspace_id + && saved.worker_id == request.worker_id + && saved.create_fingerprint == request.create_fingerprint + && saved.operation_id == request.operation_id + }) + .ok_or_else(|| { + RuntimeError::InvalidRequest("Missing original confirmed Subject shutdown".into()) + })?; + #[cfg(feature = "fs-store")] + if self + .fs_store() + .ok_or_else(|| { + RuntimeError::InvalidRequest("Subject original absence requires fs storage".into()) + })? + .worker_originals_absent(request.worker_id)? + { + return Ok(true); + } + Err(RuntimeError::InvalidRequest( + "Subject Worker original material remains or cannot be inspected".into(), + )) + } +} + /// Permanent exclusion survives WorkerRemoval. A receipt never substitutes for a /// fresh execution reconciliation on exact retry, including after Runtime reopen. #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] @@ -187,28 +231,105 @@ impl Runtime { ) })?; backend - .subject_delete_worker(&request) + .fence_subject_delete_worker(&request) .map_err(RuntimeError::InvalidRequest)?; - let mut state = self.lock()?; - if let Some(mut worker) = state.workers.get(&request.worker_id).cloned() { - let restore_receipt = - RuntimeWorkerRestoreResult::superseded_by_stop(worker.pending_restore); - worker.pending_stop = None; - worker.pending_restore = None; - if let Some(context) = worker.restore_candidate_context.take() { - context.discard_candidate(); + if self + .lock()? + .confirmed_subject_delete_originals_absent(&request)? + { + // The durable confirmed fence owns the historical shutdown proof; + // this separately checks current adapter resources after removal. + backend + .confirm_subject_delete_worker_absent(&request) + .map_err(RuntimeError::InvalidRequest)?; + let mut state = self.lock()?; + if !state.confirmed_subject_delete_originals_absent(&request)? { + return Err(RuntimeError::InvalidRequest( + "Subject originals changed during absence reconciliation".into(), + )); } - worker.restore_guard.rotate(); - worker.restore_guard.finish(restore_receipt); - worker.status = WorkerStatus::Stopped; - worker.worker_state = None; - worker.restore_intent = WorkerRestoreIntent::Explicit; - worker.internal_workers.clear(); - worker.execution_bound = false; - state.persist_worker_record(&worker)?; - state.workers.insert(request.worker_id, worker); - state.publish_worker_upsert(request.worker_id)?; + let response = runtime_api::SubjectDeleteWorkerResponse { + workspace_id: request.workspace_id.clone(), + worker_id: request.worker_id, + create_fingerprint: request.create_fingerprint.clone(), + operation_id: request.operation_id.clone(), + originals_absent: true, + }; + state + .subject_delete_worker_fences + .get_mut(&request.worker_id) + .expect("confirmed fence retained under Worker lock") + .response = Some(response.clone()); + state.persist_runtime_snapshot()?; + return Ok(response); } + let (stop, canonical, pending_stop) = { + let mut state = self.lock()?; + // A prior uncertain Stop write must be synchronized before it can + // supply evidence or choose the durable child for this retry. + state.refresh_uncertain_worker_write(&worker_ref)?; + let worker = state.worker(&worker_ref)?; + if let Some(pending) = worker.pending_stop { + ( + WorkerExecutionStopRequest { + worker_ref: worker_ref.clone(), + operation_id: pending.operation_id, + }, + worker.restore_guard.execution_termination.clone(), + true, + ) + } else if let Some(saved) = worker.restore_guard.execution_termination.clone() { + if worker.status != WorkerStatus::Stopped || worker.pending_restore.is_some() { + return Err(RuntimeError::InvalidRequest( + "Subject canonical Stop is not settled".into(), + )); + } + ( + WorkerExecutionStopRequest { + worker_ref: worker_ref.clone(), + operation_id: saved.stop_operation_id, + }, + Some(saved), + false, + ) + } else { + drop(state); + // Never call begin on an existing canonical receipt: it clears + // the old proof. No-record/never-started cases cannot get here + // through an empty backend map as evidence of historical shutdown. + let operation_id = WorkerLifecycleOperationId::new(); + self.begin_stop_operation(&worker_ref, operation_id)?; + ( + WorkerExecutionStopRequest { + worker_ref: worker_ref.clone(), + operation_id, + }, + None, + true, + ) + } + }; + let result = backend + .subject_delete_worker_stop(&request, stop.clone(), canonical.as_ref()) + .map_err(RuntimeError::InvalidRequest)?; + if let Some(saved) = canonical { + if !result.is_accepted() + || result.operation != WorkerExecutionOperation::Stop + || result.execution_id != Some(saved.execution_id) + || result.termination.as_ref() != Some(&saved) + || !saved.confirms_scope_shutdown(&worker_ref, stop.operation_id) + { + return Err(RuntimeError::InvalidRequest( + "Subject canonical Stop was not exactly reconciled".into(), + )); + } + } + if pending_stop { + // Settled A is read-only; pending B must settle through the ordinary + // validator only after cleanup reconciles its original whole receipt. + self.commit_stopped_worker(&worker_ref, stop.operation_id, result)?; + } + let mut state = self.lock()?; // This is storage-owner evidence under the same lock that permanently // excludes create/restore, not an inference from catalog projection. #[cfg(feature = "fs-store")] diff --git a/crates/worker-runtime/src/runtime/subject_delete_tests.rs b/crates/worker-runtime/src/runtime/subject_delete_tests.rs index 0a6feba55..2d8159a32 100644 --- a/crates/worker-runtime/src/runtime/subject_delete_tests.rs +++ b/crates/worker-runtime/src/runtime/subject_delete_tests.rs @@ -47,12 +47,18 @@ fn subject_delete_reserved_id_survives_unknown_reopen_and_rejects_identity_chang .is_err() ); backend.deletion_confirmed.store(true, Ordering::SeqCst); - let receipt = runtime - .subject_delete_worker_scoped(&scoped, request.clone()) - .unwrap(); - let mut expected = serde_json::to_value(&request).unwrap(); - expected["originals_absent"] = true.into(); - assert_eq!(serde_json::to_value(&receipt).unwrap(), expected); + // A generic adapter confirmation cannot replace the missing original + // Stop child/receipt, even for a never-started catalog ID. + assert!( + runtime + .subject_delete_worker_scoped(&scoped, request.clone()) + .is_err() + ); + assert!( + runtime.lock().unwrap().subject_delete_worker_fences[&create.worker_id] + .response + .is_none() + ); for index in 0..4 { let mut changed = request.clone(); match index { @@ -84,7 +90,7 @@ fn subject_delete_reserved_id_survives_unknown_reopen_and_rejects_identity_chang .forget_workspace_owner_if_unused("workspace-a") ); drop(runtime); - // Even a saved receipt must not manufacture fresh shutdown proof on reopen. + // A saved exclusion must not manufacture fresh shutdown proof on reopen. let backend = Arc::new(TestExecutionBackend::default()); let runtime = Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap(); assert_eq!(*backend.restore_count.lock().unwrap(), 0); @@ -94,11 +100,10 @@ fn subject_delete_reserved_id_survives_unknown_reopen_and_rejects_identity_chang .is_err() ); backend.deletion_confirmed.store(true, Ordering::SeqCst); - assert_eq!( + assert!( runtime .subject_delete_worker_scoped(&scoped, request) - .unwrap(), - receipt + .is_err() ); assert!(runtime.create_worker_scoped(&scoped, create).is_err()); } @@ -152,18 +157,30 @@ fn subject_delete_existing_stopped_or_active_worker_preserves_record_and_allows_ .contains_key(&create.worker_id) ); // A lost Worker settlement cannot be reported as completed deletion. - runtime - .test_runtime_store() - .fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename); - assert!( + if !stopped { runtime - .subject_delete_worker_scoped(&scoped, request.clone()) - .is_err() - ); + .test_runtime_store() + .fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename); + assert!( + runtime + .subject_delete_worker_scoped(&scoped, request.clone()) + .is_err() + ); + } + let original = runtime.execution_termination(&worker.worker_ref).unwrap(); let receipt = runtime .subject_delete_worker_scoped(&scoped, request.clone()) .unwrap(); - assert!(!receipt.originals_absent, "a stopped catalog record still has original material"); + assert!( + !receipt.originals_absent, + "a stopped catalog record still has original material" + ); + if let Some(original) = original { + assert_eq!( + runtime.execution_termination(&worker.worker_ref).unwrap(), + Some(original) + ); + } assert_eq!( runtime.worker_detail(&worker.worker_ref).unwrap().status, WorkerStatus::Stopped @@ -199,13 +216,24 @@ fn subject_delete_existing_stopped_or_active_worker_preserves_record_and_allows_ ); assert_eq!( runtime - .subject_delete_worker_scoped(&scoped, request) + .subject_delete_worker_scoped(&scoped, request.clone()) .unwrap(), receipt ); runtime .delete_worker_scoped(&scoped, &worker.worker_ref) .unwrap(); + // Canonical removal deletes the aggregate, not the permanent Subject + // fence. Its exact retry must finish from the confirmed owner history, + // without manufacturing another Stop from a missing catalog entry. + let removed = runtime + .subject_delete_worker_scoped(&scoped, request) + .unwrap(); + assert!(removed.originals_absent); + assert_eq!(removed.workspace_id, receipt.workspace_id); + assert_eq!(removed.worker_id, receipt.worker_id); + assert_eq!(removed.create_fingerprint, receipt.create_fingerprint); + assert_eq!(removed.operation_id, receipt.operation_id); assert!(runtime.create_worker_scoped(&scoped, create).is_err()); assert!( !runtime @@ -246,29 +274,44 @@ fn subject_delete_unavailable_backend_never_issues_receipt_but_preserves_exclusi backend.deletion_confirmed.store(true, Ordering::SeqCst); let runtime = Runtime::with_fs_store_and_execution_backend(options, backend).unwrap(); assert!(runtime.create_worker_scoped(&scoped, create).is_err()); - runtime - .subject_delete_worker_scoped(&scoped, request) - .unwrap(); + assert!( + runtime + .subject_delete_worker_scoped(&scoped, request) + .is_err() + ); } #[cfg(feature = "fs-store")] #[test] fn subject_delete_missing_catalog_does_not_claim_residual_originals_absent() { let root = tempfile::tempdir().unwrap(); - let options = crate::fs_store::FsRuntimeStoreOptions::new(root.path()).with_runtime_id("test-runtime"); + let options = + crate::fs_store::FsRuntimeStoreOptions::new(root.path()).with_runtime_id("test-runtime"); let backend = Arc::new(TestExecutionBackend::default()); backend.deletion_confirmed.store(true, Ordering::SeqCst); let runtime = Runtime::with_fs_store_and_execution_backend(options, backend).unwrap(); let scoped = scope("workspace-a", "server-a"); let create = scoped_task_request("unpublished original", "workspace-a"); - let original = root.path().join("workers").join(create.worker_id.to_string()).join("session/original.jsonl"); + let original = root + .path() + .join("workers") + .join(create.worker_id.to_string()) + .join("session/original.jsonl"); std::fs::create_dir_all(original.parent().unwrap()).unwrap(); std::fs::write(&original, "preserve original text").unwrap(); - let receipt = runtime.subject_delete_worker_scoped(&scoped, runtime_api::SubjectDeleteWorkerRequest { - workspace_id: scoped.workspace_id.clone(), worker_id: create.worker_id, - create_fingerprint: create.create_fingerprint.clone(), operation_id: "delete-unpublished".into(), - }).unwrap(); - assert!(!receipt.originals_absent); - assert_eq!(std::fs::read_to_string(original).unwrap(), "preserve original text"); + let result = runtime.subject_delete_worker_scoped( + &scoped, + runtime_api::SubjectDeleteWorkerRequest { + workspace_id: scoped.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint.clone(), + operation_id: "delete-unpublished".into(), + }, + ); + assert!(result.is_err()); + assert_eq!( + std::fs::read_to_string(original).unwrap(), + "preserve original text" + ); assert!(runtime.create_worker_scoped(&scoped, create).is_err()); } diff --git a/crates/worker-runtime/src/subject_delete_backend_tests.rs b/crates/worker-runtime/src/subject_delete_backend_tests.rs index 7ca9d05d8..dd2fd3f0c 100644 --- a/crates/worker-runtime/src/subject_delete_backend_tests.rs +++ b/crates/worker-runtime/src/subject_delete_backend_tests.rs @@ -343,7 +343,7 @@ fn subject_delete_without_original_owner_retains_child_identity_but_never_issues #[test] #[cfg(feature = "fs-store")] -fn subject_delete_consumes_prior_canonical_stop_without_retagging_original_scope() { +fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() { let temp = tempfile::tempdir_in(".").unwrap(); let root = fs::canonicalize(temp.path()).unwrap(); let backend = Arc::new( @@ -378,6 +378,39 @@ fn subject_delete_consumes_prior_canonical_stop_without_retagging_original_scope let worker = runtime .create_worker_scoped(&scope, create.clone()) .unwrap(); + let mut release = runtime_api::SubjectReleaseRequest { + workspace_id: scope.workspace_id.clone(), + worker_id: worker.worker_ref.worker_id, + binding_id: "cancelled-binding".into(), + operation_id: "release-before-stop".into(), + action: runtime_api::SubjectReleaseAction::Reserve, + }; + assert_eq!( + runtime + .subject_release_scoped(&scope, release.clone()) + .unwrap() + .state, + runtime_api::SubjectReleaseState::Reserved + ); + assert!( + runtime + .stop_worker_scoped(&scope, &worker.worker_ref, None) + .is_err() + ); + release.action = runtime_api::SubjectReleaseAction::Cancel; + assert_eq!( + runtime + .subject_release_scoped(&scope, release.clone()) + .unwrap() + .state, + runtime_api::SubjectReleaseState::Cancelled + ); + { + let reservations = backend.subject_reservations.lock().unwrap(); + let saved = &reservations[&(worker.worker_ref.clone(), release.operation_id.clone())]; + assert_eq!(saved.state, runtime_api::SubjectReleaseState::Cancelled); + assert!(saved.guard.is_none()); + } runtime .stop_worker_scoped(&scope, &worker.worker_ref, None) .unwrap(); @@ -391,6 +424,41 @@ fn subject_delete_consumes_prior_canonical_stop_without_retagging_original_scope create_fingerprint: create.create_fingerprint, operation_id: "delete-prior-stop".into(), }; + // Retained non-Cancelled state still blocks canonical consumption even if + // its local guard is unavailable; only the terminal no-guard row is harmless. + let key = (worker.worker_ref.clone(), release.operation_id); + backend + .subject_reservations + .lock() + .unwrap() + .get_mut(&key) + .unwrap() + .state = runtime_api::SubjectReleaseState::Reserved; + assert_eq!( + backend + .reconcile_termination_receipt(&worker.worker_ref, &original) + .unwrap(), + None + ); + assert!( + backend + .subject_delete_worker_stop( + &deletion, + WorkerExecutionStopRequest { + worker_ref: worker.worker_ref.clone(), + operation_id: original.stop_operation_id, + }, + Some(&original), + ) + .is_err() + ); + backend + .subject_reservations + .lock() + .unwrap() + .get_mut(&key) + .unwrap() + .state = runtime_api::SubjectReleaseState::Cancelled; runtime .subject_delete_worker_scoped(&scope, deletion.clone()) .unwrap(); @@ -403,4 +471,749 @@ fn subject_delete_consumes_prior_canonical_stop_without_retagging_original_scope backend.admissions.lock().unwrap()[&worker.worker_ref].deletion_stop_operation_id, Some(original.stop_operation_id) ); + assert_eq!( + runtime.execution_termination(&worker.worker_ref).unwrap(), + Some(original) + ); + assert!(backend.workers.lock().unwrap().is_empty()); + assert!(backend.pending_spawns.lock().unwrap().is_empty()); + assert!(backend.pending_restores.lock().unwrap().is_empty()); +} + +// Only the extra factory-owned cleanup is faulted. Controller Stop, scope join, +// Runtime persistence, cold loading, and reconciliation remain production code. +#[cfg(feature = "fs-store")] +struct SubjectCleanupFaultFactory { + inner: MockFactory, + spawns: Arc, + cleanup_allowed: Arc, + cleanup_calls: Arc, +} + +#[cfg(feature = "fs-store")] +#[async_trait] +impl RuntimeWorkerFactory for SubjectCleanupFaultFactory { + async fn spawn_controller( + &self, + request: WorkerExecutionSpawnRequest, + ) -> Result { + self.spawns.fetch_add(1, Ordering::SeqCst); + self.inner.spawn_controller(request).await + } + + async fn prepare_restore( + &self, + _: WorkerExecutionRestoreRequest, + ) -> Result, RuntimeRestoreFailure> { + panic!("saved Stop recovery must not prepare a Controller") + } + + async fn reconcile_stopped_worker(&self, _: &WorkerRef) -> Result<(), String> { + self.cleanup_calls.fetch_add(1, Ordering::SeqCst); + if self.cleanup_allowed.load(Ordering::SeqCst) { + Ok(()) + } else { + Err("injected extra factory cleanup failure after actual termination".into()) + } + } +} + +#[test] +#[cfg(feature = "fs-store")] +fn subject_delete_pending_actual_termination_survives_cold_cleanup_retry() { + use crate::fs_store::{ + PersistedWorkerExecutionState as Execution, PersistedWorkerLifecycleOperation as Operation, + }; + // Both direct Subject retry of pending B and ordinary startup reconciliation + // must consume the same saved whole receipt, without a warm successful Stop. + for startup_recovers in [false, true] { + let temp = tempfile::tempdir_in(".").unwrap(); + let root = fs::canonicalize(temp.path()).unwrap(); + let spawns = Arc::new(AtomicUsize::new(0)); + let cleanup_allowed = Arc::new(AtomicBool::new(false)); + let cleanup_calls = Arc::new(AtomicUsize::new(0)); + let observed_cwds = Arc::new(Mutex::new(Vec::new())); + let new_backend = || { + Arc::new( + WorkerRuntimeExecutionBackend::new(SubjectCleanupFaultFactory { + inner: MockFactory { + client: MockClient::new(vec![]), + runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"), + cwd: root.clone(), + store_dir: root.join("sessions"), + worker_metadata_dir: root.join("metadata"), + observed_cwds: observed_cwds.clone(), + observed_workspace_clients: Arc::default(), + }, + spawns: spawns.clone(), + cleanup_allowed: cleanup_allowed.clone(), + cleanup_calls: cleanup_calls.clone(), + }) + .unwrap(), + ) + }; + let backend = new_backend(); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog")) + .with_runtime_id("subject-pending-cleanup"); + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test"); + let mut create = create_request("Subject Stop pending extra cleanup"); + create.workspace_api = Some(WorkspaceApiRef { + workspace_id: scope.workspace_id.clone(), + base_url: "https://workspace.invalid".into(), + }); + create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: scope.workspace_id.clone(), + language: "English".into(), + }); + let worker = runtime + .create_worker_scoped(&scope, create.clone()) + .unwrap(); + let execution = backend.workers.lock().unwrap()[&worker.worker_ref].clone(); + let deletion = runtime_api::SubjectDeleteWorkerRequest { + workspace_id: scope.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint.clone(), + operation_id: "subject-pending-cleanup".into(), + }; + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + let original = runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .unwrap(); + assert_eq!(original.execution_id, execution.tasks.execution_id); + assert!(original.confirms_scope_shutdown(&worker.worker_ref, original.stop_operation_id)); + assert!(execution.handle.protocol_is_closed()); + assert!(execution.tasks.tasks.lock().unwrap().is_empty()); + assert!(backend.workers.lock().unwrap().is_empty()); + let before = runtime + .test_runtime_store() + .read_worker_record(&worker.worker_ref) + .unwrap(); + assert!( + matches!(before.execution_state, Execution::ReconciliationRequired(Operation::Stop(ref pending)) + if pending.operation_id == original.stop_operation_id) + ); + // Read the saved aggregate, not the adapter's still-warm receipt cache. + assert_eq!( + serde_json::to_value(&before.restore_guard).unwrap()["execution_termination"], + serde_json::to_value(&original).unwrap() + ); + assert_eq!(spawns.load(Ordering::SeqCst), 1); + drop(execution); + drop(runtime); + drop(backend); + + cleanup_allowed.store(startup_recovers, Ordering::SeqCst); + let cold = new_backend(); + assert!(cold.completed_stops.lock().unwrap().is_empty()); + let reopened = + EmbeddedRuntime::with_fs_store_and_execution_backend(options, cold.clone()).unwrap(); + let loaded = reopened + .test_runtime_store() + .read_worker_record(&worker.worker_ref) + .unwrap(); + if !startup_recovers { + assert!( + matches!(loaded.execution_state, Execution::ReconciliationRequired(Operation::Stop(ref pending)) + if pending.operation_id == original.stop_operation_id) + ); + assert_eq!( + serde_json::to_value(&loaded.restore_guard).unwrap()["execution_termination"], + serde_json::to_value(&original).unwrap() + ); + cleanup_allowed.store(true, Ordering::SeqCst); + } else { + assert!(!matches!( + loaded.execution_state, + Execution::ReconciliationRequired(_) + )); + assert_eq!(loaded.status, crate::catalog::WorkerStatus::Stopped); + } + let cleanup_before = cleanup_calls.load(Ordering::SeqCst); + reopened + .subject_delete_worker_scoped(&scope, deletion) + .unwrap(); + assert!(cleanup_calls.load(Ordering::SeqCst) > cleanup_before); + let settled = reopened + .test_runtime_store() + .read_worker_record(&worker.worker_ref) + .unwrap(); + assert!(!matches!( + settled.execution_state, + Execution::ReconciliationRequired(_) + )); + assert_eq!(settled.status, crate::catalog::WorkerStatus::Stopped); + assert_eq!( + serde_json::to_value(&settled.restore_guard).unwrap()["execution_termination"], + serde_json::to_value(&original).unwrap() + ); + assert_eq!( + reopened.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + assert_eq!( + cold.admissions.lock().unwrap()[&worker.worker_ref].deletion_stop_operation_id, + Some(original.stop_operation_id) + ); + assert!(cold.completed_stops.lock().unwrap().is_empty()); + assert!(cold.workers.lock().unwrap().is_empty()); + assert!(cold.pending_spawns.lock().unwrap().is_empty()); + assert!(cold.pending_restores.lock().unwrap().is_empty()); + assert_eq!( + spawns.load(Ordering::SeqCst), + 1, + "cold retry must not spawn a Controller" + ); + assert!(reopened.create_worker_scoped(&scope, create).is_err()); + } +} + +// This wrapper injects only the canonical write acknowledgement boundary after +// the real owner has stopped. It never supplies a termination receipt itself. +#[cfg(feature = "fs-store")] +struct SubjectStopSaveFaultBackend { + inner: Arc>, + store: Mutex>, + fault: Mutex>, +} +#[cfg(feature = "fs-store")] +impl WorkerExecutionBackend for SubjectStopSaveFaultBackend { + fn backend_id(&self) -> &str { + self.inner.backend_id() + } + fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { + self.inner.spawn_worker(request) + } + fn dispatch_input(&self, worker: &WorkerRef, input: WorkerInput) -> WorkerExecutionResult { + self.inner.dispatch_input(worker, input) + } + fn preflight_subject_delete_worker( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + self.inner.preflight_subject_delete_worker(request) + } + fn fence_subject_delete_worker( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + self.inner.fence_subject_delete_worker(request) + } + fn subject_delete_worker_stop( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + stop: WorkerExecutionStopRequest, + canonical: Option<&crate::execution::WorkerExecutionTermination>, + ) -> Result { + let result = self + .inner + .subject_delete_worker_stop(request, stop, canonical)?; + if result.is_accepted() + && canonical.is_none() + && let Some(fault) = self.fault.lock().unwrap().take() + { + self.store + .lock() + .unwrap() + .as_ref() + .unwrap() + .fail_next_worker_write(fault); + } + Ok(result) + } + fn reconcile_termination_receipt( + &self, + worker: &WorkerRef, + saved: &crate::execution::WorkerExecutionTermination, + ) -> Result, String> { + self.inner.reconcile_termination_receipt(worker, saved) + } +} + +#[test] +#[cfg(feature = "fs-store")] +fn subject_delete_first_stop_saves_original_child_and_reconciles_interrupted_commit() { + for fault in [ + None, + Some(crate::fs_store::AtomicWriteFault::BeforeRename), + Some(crate::fs_store::AtomicWriteFault::AfterRename), + ] { + let temp = tempfile::tempdir_in(".").unwrap(); + let root = fs::canonicalize(temp.path()).unwrap(); + let new_backend = || { + Arc::new( + WorkerRuntimeExecutionBackend::new(MockFactory { + client: MockClient::new(vec![]), + runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"), + cwd: root.clone(), + store_dir: root.join("sessions"), + worker_metadata_dir: root.join("metadata"), + observed_cwds: Arc::default(), + observed_workspace_clients: Arc::default(), + }) + .unwrap(), + ) + }; + let inner = new_backend(); + let backend = Arc::new(SubjectStopSaveFaultBackend { + inner: inner.clone(), + store: Mutex::new(None), + fault: Mutex::new(fault), + }); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog")) + .with_runtime_id("subject-first-stop"); + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + *backend.store.lock().unwrap() = Some(runtime.test_runtime_store()); + runtime.store_config_bundle(test_bundle()).unwrap(); + let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test"); + let mut create = create_request("Subject first Stop original proof"); + create.workspace_api = Some(WorkspaceApiRef { + workspace_id: scope.workspace_id.clone(), + base_url: "https://workspace.invalid".into(), + }); + create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: scope.workspace_id.clone(), + language: "English".into(), + }); + let worker = runtime + .create_worker_scoped(&scope, create.clone()) + .unwrap(); + let execution = inner.workers.lock().unwrap()[&worker.worker_ref].clone(); + let request = runtime_api::SubjectDeleteWorkerRequest { + workspace_id: scope.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint.clone(), + operation_id: "subject-first-stop".into(), + }; + let first = runtime.subject_delete_worker_scoped(&scope, request.clone()); + if fault.is_some() { + assert!(first.is_err()); + } else { + first.unwrap(); + } + let original = inner + .termination_receipt(&worker.worker_ref) + .unwrap() + .unwrap(); + assert_eq!(original.execution_id, execution.tasks.execution_id); + assert!(original.confirms_scope_shutdown(&worker.worker_ref, original.stop_operation_id)); + assert!(execution.handle.protocol_is_closed()); + assert!(execution.tasks.tasks.lock().unwrap().is_empty()); + assert_eq!( + inner.admissions.lock().unwrap()[&worker.worker_ref].deletion_stop_operation_id, + Some(original.stop_operation_id) + ); + if matches!(fault, Some(crate::fs_store::AtomicWriteFault::AfterRename)) { + runtime + .test_runtime_store() + .fail_next_worker_reconcile_sync(); + assert!( + runtime + .subject_delete_worker_scoped(&scope, request.clone()) + .is_err() + ); + } + // BeforeRename retains the same pending operation; AfterRename must + // reread/sync the committed original, never begin/clear/retag it. + runtime + .subject_delete_worker_scoped(&scope, request.clone()) + .unwrap(); + assert_eq!( + runtime.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + drop(runtime); + drop(backend); + drop(inner); + let cold = new_backend(); + let reopened = + EmbeddedRuntime::with_fs_store_and_execution_backend(options, cold.clone()).unwrap(); + assert!(cold.completed_stops.lock().unwrap().is_empty()); + reopened + .subject_delete_worker_scoped(&scope, request.clone()) + .unwrap(); + assert_eq!( + reopened.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + assert!( + cold.completed_stops.lock().unwrap().is_empty(), + "canonical read must not mint another Stop" + ); + assert_eq!( + cold.admissions.lock().unwrap()[&worker.worker_ref].deletion_stop_operation_id, + Some(original.stop_operation_id) + ); + assert!(cold.workers.lock().unwrap().is_empty()); + assert!(cold.factory.observed_cwds.lock().unwrap().is_empty()); + assert!(reopened.create_worker_scoped(&scope, create).is_err()); + } +} + +#[test] +#[cfg(feature = "fs-store")] +fn subject_delete_cold_prior_stop_rejects_replacement_and_mismatched_originals() { + let temp = tempfile::tempdir_in(".").unwrap(); + let root = fs::canonicalize(temp.path()).unwrap(); + let new_backend = || { + Arc::new( + WorkerRuntimeExecutionBackend::new(MockFactory { + client: MockClient::new(vec![]), + runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"), + cwd: root.clone(), + store_dir: root.join("sessions"), + worker_metadata_dir: root.join("metadata"), + observed_cwds: Arc::default(), + observed_workspace_clients: Arc::default(), + }) + .unwrap(), + ) + }; + let backend = new_backend(); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog")) + .with_runtime_id("subject-cold-prior"); + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test"); + let mut create = create_request("prior real Stop cold Subject"); + create.workspace_api = Some(WorkspaceApiRef { + workspace_id: scope.workspace_id.clone(), + base_url: "https://workspace.invalid".into(), + }); + create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: scope.workspace_id.clone(), + language: "English".into(), + }); + let worker = runtime + .create_worker_scoped(&scope, create.clone()) + .unwrap(); + runtime + .stop_worker_scoped(&scope, &worker.worker_ref, None) + .unwrap(); + let original = runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .unwrap(); + drop(runtime); + drop(backend); + let deletion = runtime_api::SubjectDeleteWorkerRequest { + workspace_id: scope.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint.clone(), + operation_id: "delete-cold-prior".into(), + }; + for invalid in [ + "worker", + "operation", + "execution", + "feature", + "whole-receipt", + "candidate", + "child", + "valid", + ] { + let backend = new_backend(); + if invalid == "candidate" { + let mut request = direct_restore_request(worker.worker_ref.clone()); + request.workspace_scope = Some(scope.clone()); + request.request = create.clone(); + assert!(matches!( + backend.restore_worker(request), + WorkerExecutionSpawnResult::Connected { .. } + )); + } else if invalid == "child" { + backend.admissions.lock().unwrap().insert( + worker.worker_ref.clone(), + WorkerAdmission { + workspace_id: Some(scope.workspace_id.clone()), + create_fingerprint: create.create_fingerprint.clone(), + deletion: Some(deletion.clone()), + deletion_stop_operation_id: Some( + crate::execution::WorkerLifecycleOperationId::new(), + ), + }, + ); + } else if invalid != "valid" { + let mut wrong = original.clone(); + match invalid { + "worker" => wrong.worker_ref = WorkerRef::new(WorkerId::now_v7()), + "operation" => { + wrong.stop_operation_id = crate::execution::WorkerLifecycleOperationId::new() + } + "execution" => wrong.execution_id = uuid::Uuid::now_v7(), + "feature" => { + wrong + .feature_failures + .push(crate::execution::WorkerFeatureShutdownFailure { + category: worker::hook::HookErrorCategory::Timeout, + diagnostic: "not joined".into(), + }) + } + "whole-receipt" => wrong + .submissions + .push(crate::execution::WorkerSubmissionAck { + submission_request_id: "other".into(), + submission_id: "other".into(), + disposition: protocol::SubmissionDisposition::Started, + }), + _ => unreachable!(), + } + backend + .completed_stops + .lock() + .unwrap() + .insert(worker.worker_ref.clone(), wrong); + } + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + let result = runtime.subject_delete_worker_scoped(&scope, deletion.clone()); + if invalid == "valid" { + result.unwrap(); + assert_eq!( + runtime.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + assert!(backend.completed_stops.lock().unwrap().is_empty()); + } else { + assert!(result.is_err(), "{invalid}"); + let record = runtime + .test_runtime_store() + .read_worker_record(&worker.worker_ref) + .unwrap(); + assert_eq!( + serde_json::to_value(&record.restore_guard).unwrap()["execution_termination"], + serde_json::to_value(&original).unwrap() + ); + if invalid == "candidate" { + assert!( + backend + .pending_restore(&worker.worker_ref) + .unwrap() + .is_some() + ); + let stopped = backend.stop_worker_operation(WorkerExecutionStopRequest { + worker_ref: worker.worker_ref.clone(), + operation_id: crate::execution::WorkerLifecycleOperationId::new(), + }); + assert!(stopped.is_accepted(), "{stopped:?}"); + } + } + } +} + +#[test] +#[cfg(feature = "fs-store")] +fn subject_delete_stopped_catalog_without_original_never_certifies_empty_backend() { + let root = tempfile::tempdir().unwrap(); + let backend = Arc::new( + WorkerRuntimeExecutionBackend::new(MockFactory { + client: MockClient::new(vec![]), + runtime_base: root.path().join("no-launch"), + cwd: root.path().to_path_buf(), + store_dir: root.path().join("sessions"), + worker_metadata_dir: root.path().join("metadata"), + observed_cwds: Arc::default(), + observed_workspace_clients: Arc::default(), + }) + .unwrap(), + ); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.path().join("catalog")) + .with_runtime_id("missing-original"); + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test"); + let mut create = create_request("never started"); + create.workspace_api = Some(WorkspaceApiRef { + workspace_id: scope.workspace_id.clone(), + base_url: "https://workspace.invalid".into(), + }); + let worker_ref = WorkerRef::new(create.worker_id); + runtime + .test_runtime_store() + .write_worker_record(&crate::fs_store::PersistedWorkerRecord { + restore_guard: crate::runtime::RestoreGuard::default(), + worker_ref: worker_ref.clone(), + worker_id: create.worker_id, + profile: create.profile.clone(), + display_name: None, + profile_source: create.profile_source.reference(), + config_bundle: create.config_bundle.clone(), + created_at_ms: None, + status: crate::catalog::WorkerStatus::Stopped, + last_finished_submission_request_id: None, + execution_state: crate::fs_store::PersistedWorkerExecutionState::Available( + crate::fs_store::PersistedWorkerExecution { + request: create.clone(), + binding: None, + restore_intent: crate::catalog::WorkerRestoreIntent::Explicit, + }, + ), + workspace_id: Some(scope.workspace_id.clone()), + workdir_attachments: Vec::new(), + logical_workdir_attachments: Vec::new(), + }) + .unwrap(); + drop(runtime); + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap(); + let deletion = runtime_api::SubjectDeleteWorkerRequest { + workspace_id: scope.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint, + operation_id: "missing-original".into(), + }; + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion) + .is_err() + ); + assert!( + runtime + .execution_termination(&worker_ref) + .unwrap() + .is_none() + ); + assert!(backend.completed_stops.lock().unwrap().is_empty()); + assert!(runtime.worker_detail(&worker_ref).is_ok()); + assert!(!root.path().join("no-launch").exists()); +} + +#[test] +#[cfg(feature = "fs-store")] +fn subject_delete_confirmed_removal_retries_warm_and_cold_without_new_stop() { + let temp = tempfile::tempdir_in(".").unwrap(); + let root = fs::canonicalize(temp.path()).unwrap(); + let spawns = Arc::new(AtomicUsize::new(0)); + let cleanup_allowed = Arc::new(AtomicBool::new(true)); + let cleanup_calls = Arc::new(AtomicUsize::new(0)); + let new_backend = || { + Arc::new( + WorkerRuntimeExecutionBackend::new(SubjectCleanupFaultFactory { + inner: MockFactory { + client: MockClient::new(vec![]), + runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"), + cwd: root.clone(), + store_dir: root.join("sessions"), + worker_metadata_dir: root.join("metadata"), + observed_cwds: Arc::default(), + observed_workspace_clients: Arc::default(), + }, + spawns: spawns.clone(), + cleanup_allowed: cleanup_allowed.clone(), + cleanup_calls: cleanup_calls.clone(), + }) + .unwrap(), + ) + }; + let backend = new_backend(); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog")) + .with_runtime_id("subject-after-removal"); + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test"); + let mut create = create_request("Subject confirmed removal retry"); + create.workspace_api = Some(WorkspaceApiRef { + workspace_id: scope.workspace_id.clone(), + base_url: "https://workspace.invalid".into(), + }); + create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: scope.workspace_id.clone(), + language: "English".into(), + }); + let worker = runtime + .create_worker_scoped(&scope, create.clone()) + .unwrap(); + let deletion = runtime_api::SubjectDeleteWorkerRequest { + workspace_id: scope.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint.clone(), + operation_id: "same-confirmed-delete".into(), + }; + let confirmed = runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .unwrap(); + assert!(!confirmed.originals_absent); + let original = runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .unwrap(); + assert!(original.confirms_scope_shutdown(&worker.worker_ref, original.stop_operation_id)); + runtime + .delete_worker_scoped(&scope, &worker.worker_ref) + .unwrap(); + assert!(runtime.worker_detail(&worker.worker_ref).is_err()); + let absent = runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .unwrap(); + let mut expected = confirmed; + expected.originals_absent = true; + assert_eq!(absent, expected); + assert_eq!( + backend.completed_stops.lock().unwrap()[&worker.worker_ref], + original + ); + drop(runtime); + drop(backend); + let cold = new_backend(); + let reopened = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), cold.clone()) + .unwrap(); + cleanup_allowed.store(false, Ordering::SeqCst); + assert!( + reopened + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err(), + "saved response cannot bypass current factory cleanup" + ); + cleanup_allowed.store(true, Ordering::SeqCst); + let material = options + .root + .join("workers") + .join(create.worker_id.to_string()); + fs::create_dir_all(&material).unwrap(); + fs::write(material.join("residual"), "unconfirmed material").unwrap(); + assert!( + reopened + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err(), + "catalog absence cannot hide residual original files" + ); + fs::remove_file(material.join("residual")).unwrap(); + fs::remove_dir(&material).unwrap(); + assert_eq!( + reopened + .subject_delete_worker_scoped(&scope, deletion) + .unwrap(), + expected + ); + assert_eq!(spawns.load(Ordering::SeqCst), 1); + assert!( + cold.completed_stops.lock().unwrap().is_empty(), + "absence confirmation never mints another termination" + ); + assert!(cold.workers.lock().unwrap().is_empty()); + assert!(cold.pending_spawns.lock().unwrap().is_empty()); + assert!(cold.pending_restores.lock().unwrap().is_empty()); + assert!(reopened.create_worker_scoped(&scope, create).is_err()); } diff --git a/crates/worker-runtime/src/subject_delete_create_tests.rs b/crates/worker-runtime/src/subject_delete_create_tests.rs index f92687a47..53f22293a 100644 --- a/crates/worker-runtime/src/subject_delete_create_tests.rs +++ b/crates/worker-runtime/src/subject_delete_create_tests.rs @@ -1,6 +1,8 @@ // Actual create factory future remains behind Runtime's Worker operation lock; // deletion must wait for that owner, then permanently reject the delayed retry. struct SubjectDeleteDeferredCreate { + inner: Option, + handles: Arc>>, entered: mpsc::Sender<()>, release: Arc, } @@ -14,21 +16,48 @@ impl RuntimeWorkerFactory for SubjectDeleteDeferredCreate { } async fn spawn_controller( &self, - _: WorkerExecutionSpawnRequest, + request: WorkerExecutionSpawnRequest, ) -> Result { self.entered.send(()).unwrap(); self.release.notified().await; - Err("create failed before any Controller".into()) + if let Some(inner) = &self.inner { + let controller = inner.spawn_controller(request).await?; + self.handles.lock().unwrap().push(controller.handle.clone()); + Ok(controller) + } else { + Err("create failed before any Controller".into()) + } } } #[test] fn subject_delete_serializes_inflight_create_and_rejects_all_late_admission() { + subject_delete_inflight_create(true); +} + +#[test] +fn subject_delete_unproven_create_failure_remains_fenced_without_termination() { + subject_delete_inflight_create(false); +} + +fn subject_delete_inflight_create(real_controller: bool) { let root = tempfile::tempdir().unwrap(); let (entered, entry) = mpsc::channel(); let release = Arc::new(tokio::sync::Notify::new()); + let handles = Arc::new(Mutex::new(Vec::new())); + let inner = real_controller.then(|| MockFactory { + client: MockClient::new(vec![]), + runtime_base: root.path().join("control"), + cwd: root.path().to_path_buf(), + store_dir: root.path().join("sessions"), + worker_metadata_dir: root.path().join("metadata"), + observed_cwds: Arc::default(), + observed_workspace_clients: Arc::default(), + }); let backend = Arc::new( WorkerRuntimeExecutionBackend::new(SubjectDeleteDeferredCreate { + inner, + handles: handles.clone(), entered, release: release.clone(), }) @@ -70,8 +99,49 @@ fn subject_delete_serializes_inflight_create_and_rejects_all_late_admission() { std::thread::spawn(move || runtime.subject_delete_worker_scoped(&scope, request)) }; release.notify_one(); - assert!(creating.join().unwrap().is_err()); - deleting.join().unwrap().unwrap(); + let creation = creating.join().unwrap(); + let deletion = deleting.join().unwrap(); + let worker_ref = WorkerRef::new(create.worker_id); + if real_controller { + creation.unwrap(); + deletion.unwrap(); + let original = runtime.execution_termination(&worker_ref).unwrap().unwrap(); + assert_eq!( + original.scope, + crate::execution::WorkerExecutionTerminationScope::Controller + ); + assert!(original.shutdown_command_id.is_some()); + assert!(original.feature_failures.is_empty()); + assert_eq!( + backend.admissions.lock().unwrap()[&worker_ref].deletion_stop_operation_id, + Some(original.stop_operation_id) + ); + assert_eq!( + backend.termination_receipt(&worker_ref).unwrap(), + Some(original) + ); + assert_eq!(handles.lock().unwrap().len(), 1); + assert!(handles.lock().unwrap()[0].protocol_is_closed()); + } else { + assert!(creation.is_err()); + assert!(deletion.is_err()); + assert!( + runtime + .execution_termination(&worker_ref) + .unwrap() + .is_none() + ); + assert!(backend.completed_stops.lock().unwrap().is_empty()); + assert!( + backend + .pending_spawns + .lock() + .unwrap() + .contains_key(&worker_ref) + ); + assert!(handles.lock().unwrap().is_empty()); + assert!(runtime.worker_detail(&worker_ref).is_ok()); + } assert!( runtime .create_worker_scoped(&scope, create.clone()) diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index e9f9c8a5c..dc80488e1 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -5266,6 +5266,147 @@ where Ok(()) } + // Caller holds the adapter Worker lock through validation, Stop and cleanup. + fn subject_delete_worker_under_lock( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + stop: Option, + canonical: Option<&crate::execution::WorkerExecutionTermination>, + ) -> Result { + let worker = WorkerRef::new(request.worker_id); + self.validate_subject_delete_identity(request)?; + if stop.as_ref().is_some_and(|stop| stop.worker_ref != worker) { + return Err("Subject Stop Worker identity mismatch".into()); + } + let canonical_result = if let Some(saved) = canonical { + let stop = stop + .as_ref() + .ok_or("canonical Stop requires original child")?; + Some(self.saved_stop_result_under_lock(stop, saved)?) + } else { + None + }; + let operation_id = { + let mut admissions = self + .admissions + .lock() + .map_err(|_| "worker admission lock poisoned")?; + let admission = admissions + .entry(worker.clone()) + .or_insert_with(|| WorkerAdmission { + workspace_id: Some(request.workspace_id.clone()), + create_fingerprint: request.create_fingerprint.clone(), + deletion: None, + deletion_stop_operation_id: None, + }); + admission.deletion = Some(request.clone()); + match admission.deletion_stop_operation_id { + Some(operation_id) => { + if stop + .as_ref() + .is_some_and(|stop| stop.operation_id != operation_id) + { + return Err("Subject Stop child identity changed".into()); + } + operation_id + } + None => { + // Reuse a completed original Stop, never retag its receipt + // as a newly generated deletion child. The Stop consumer + // below rejects replay if any replacement owner is present. + let retained = self + .completed_stops + .lock() + .map_err(|_| "stop completion receipts lock poisoned")? + .get(&worker) + .map(|receipt| receipt.stop_operation_id); + let operation_id = stop + .as_ref() + .map(|stop| stop.operation_id) + .or(retained) + .unwrap_or_default(); + if retained.is_some_and(|original| original != operation_id) { + return Err("Subject Stop differs from retained original operation".into()); + } + admission.deletion_stop_operation_id = Some(operation_id); + operation_id + } + } + }; + let result = if let Some(result) = canonical_result { + result + } else { + self.stop_worker_operation_under_lock(WorkerExecutionStopRequest { + worker_ref: worker.clone(), + operation_id, + }) + }; + let result = self.reconcile_stop_cleanup_under_lock(&worker, result); + if !result.is_accepted() { + return Ok(result); + } + self.subject_reservations + .lock() + .map_err(|_| "Subject reservations poisoned")? + .retain(|(saved, _), _| saved != &worker); + self.completed_restore_failures + .lock() + .map_err(|_| "completed restores poisoned")? + .remove(&worker); + Ok(result) + } + + // Caller holds the adapter Worker lock across receipt and cleanup checks. + fn saved_stop_result_under_lock( + &self, + stop: &WorkerExecutionStopRequest, + saved: &crate::execution::WorkerExecutionTermination, + ) -> Result { + if !saved.confirms_scope_shutdown(&stop.worker_ref, stop.operation_id) { + return Err("invalid canonical Stop receipt".into()); + } + // Even a same-execution live entry is not authority to reuse history. + let has_execution = self + .workers + .lock() + .map_err(|_| "worker registry poisoned")? + .contains_key(&stop.worker_ref); + if has_execution + || self + .read_termination_receipt_under_lock(&stop.worker_ref, Some(saved))? + .as_ref() + != Some(saved) + { + return Err("canonical Stop has a current execution or candidate".into()); + } + let mut result = WorkerExecutionResult::accepted(WorkerExecutionOperation::Stop); + result.execution_id = Some(saved.execution_id); + result.termination = Some(saved.clone()); + Ok(result) + } + + fn reconcile_stop_cleanup_under_lock( + &self, + worker: &WorkerRef, + mut result: WorkerExecutionResult, + ) -> WorkerExecutionResult { + if !result.is_accepted() { + return result; + } + // Scope join and factory-owned cache/artifact cleanup are independent. + let factory = self.factory.clone(); + let stopped = worker.clone(); + if let Err(error) = self + .run_cancellable_on_adapter_runtime(RUNTIME_TASK_TIMEOUT, async move { + factory.reconcile_stopped_worker(&stopped).await + }) + { + result.outcome = crate::execution::WorkerExecutionOutcome::Errored; + result.message = Some(error); + } + result + } + fn read_termination_receipt( &self, worker_ref: &WorkerRef, @@ -5275,6 +5416,14 @@ where let _guard = operation .lock() .map_err(|_| "Worker operation lock poisoned".to_string())?; + self.read_termination_receipt_under_lock(worker_ref, saved) + } + + fn read_termination_receipt_under_lock( + &self, + worker_ref: &WorkerRef, + saved: Option<&crate::execution::WorkerExecutionTermination>, + ) -> Result, String> { if self .pending_spawns .lock() @@ -5289,8 +5438,12 @@ where .subject_reservations .lock() .map_err(|_| "Subject reservation registry poisoned".to_string())? - .keys() - .any(|(worker, _)| worker == worker_ref) + .iter() + .any(|((worker, _), reservation)| { + worker == worker_ref + && (reservation.guard.is_some() + || reservation.state != runtime_api::SubjectReleaseState::Cancelled) + }) { return Ok(None); } @@ -5828,65 +5981,106 @@ where let worker = WorkerRef::new(request.worker_id); let lock = self.worker_lock(&worker)?; let _operation = lock.lock().map_err(|_| "worker operation lock poisoned")?; - self.validate_subject_delete_identity(request)?; - let operation_id = { - let mut admissions = self - .admissions - .lock() - .map_err(|_| "worker admission lock poisoned")?; - let admission = admissions - .entry(worker.clone()) - .or_insert_with(|| WorkerAdmission { - workspace_id: Some(request.workspace_id.clone()), - create_fingerprint: request.create_fingerprint.clone(), - deletion: None, - deletion_stop_operation_id: None, - }); - admission.deletion = Some(request.clone()); - match admission.deletion_stop_operation_id { - Some(operation_id) => operation_id, - None => { - // Reuse a completed original Stop, never retag its receipt - // as a newly generated deletion child. The Stop consumer - // below rejects replay if any replacement owner is present. - let operation_id = self - .completed_stops - .lock() - .map_err(|_| "stop completion receipts lock poisoned")? - .get(&worker) - .map(|receipt| receipt.stop_operation_id) - .unwrap_or_default(); - admission.deletion_stop_operation_id = Some(operation_id); - operation_id - } - } - }; - let result = self.stop_worker_operation_under_lock(WorkerExecutionStopRequest { - worker_ref: worker.clone(), - operation_id, - }); - if !result.is_accepted() { - return Err(result.message_or_default()); + let result = self.subject_delete_worker_under_lock(request, None, None)?; + if result.is_accepted() { + Ok(()) + } else { + Err(result.message_or_default()) } - // A live Controller and failed unpublished candidates can coexist. The - // ordinary stop's live branch joins the Controller; also reconcile any - // factory-owned cache/artifacts before issuing deletion proof. - let factory = self.factory.clone(); - let stopped = worker.clone(); - self.run_cancellable_on_adapter_runtime(RUNTIME_TASK_TIMEOUT, async move { - factory.reconcile_stopped_worker(&stopped).await - })?; - self.subject_reservations + } + + fn fence_subject_delete_worker( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + let worker = WorkerRef::new(request.worker_id); + let lock = self.worker_lock(&worker)?; + let _operation = lock.lock().map_err(|_| "worker operation lock poisoned")?; + self.validate_subject_delete_identity(request)?; + self.admissions .lock() - .map_err(|_| "Subject reservations poisoned")? - .retain(|(saved, _), _| saved != &worker); - self.completed_restore_failures - .lock() - .map_err(|_| "completed restores poisoned")? - .remove(&worker); + .map_err(|_| "worker admission lock poisoned")? + .entry(worker) + .or_insert_with(|| WorkerAdmission { + workspace_id: Some(request.workspace_id.clone()), + create_fingerprint: request.create_fingerprint.clone(), + deletion: None, + deletion_stop_operation_id: None, + }) + .deletion = Some(request.clone()); Ok(()) } + fn confirm_subject_delete_worker_absent( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + ) -> Result<(), String> { + let worker = WorkerRef::new(request.worker_id); + let lock = self.worker_lock(&worker)?; + let _operation = lock.lock().map_err(|_| "worker operation lock poisoned")?; + self.validate_subject_delete_identity(request)?; + if self + .runtime + .lock() + .map_err(|_| "adapter runtime poisoned")? + .is_none() + { + return Err("execution backend is unavailable".into()); + } + if self + .admissions + .lock() + .map_err(|_| "worker admission lock poisoned")? + .get(&worker) + .and_then(|admission| admission.deletion.as_ref()) + != Some(request) + { + return Err("exact Subject admission fence required".into()); + } + if self.pending_spawn_exists(&worker)? + || self + .pending_restores + .lock() + .map_err(|_| "restore registry poisoned")? + .contains_key(&worker) + || self + .workers + .lock() + .map_err(|_| "worker registry poisoned")? + .contains_key(&worker) + || self + .subject_reservations + .lock() + .map_err(|_| "Subject reservations poisoned")? + .iter() + .any(|((saved, _), reservation)| { + *saved == worker + && (reservation.guard.is_some() + || reservation.state != runtime_api::SubjectReleaseState::Cancelled) + }) + { + return Err("Subject Worker still has an execution resource owner".into()); + } + // The caller's confirmed fence proves prior shutdown. Factory artifacts + // are checked separately; neither an empty map nor cleanup mints proof. + let factory = self.factory.clone(); + self.run_cancellable_on_adapter_runtime(RUNTIME_TASK_TIMEOUT, async move { + factory.reconcile_stopped_worker(&worker).await + }) + } + + fn subject_delete_worker_stop( + &self, + request: &runtime_api::SubjectDeleteWorkerRequest, + stop: WorkerExecutionStopRequest, + canonical: Option<&crate::execution::WorkerExecutionTermination>, + ) -> Result { + let worker = WorkerRef::new(request.worker_id); + let lock = self.worker_lock(&worker)?; + let _operation = lock.lock().map_err(|_| "worker operation lock poisoned")?; + self.subject_delete_worker_under_lock(request, Some(stop), canonical) + } + fn recover_subject_create( &self, request: &runtime_api::SubjectCreateRecoveryRequest, @@ -7922,6 +8116,52 @@ where self.send_method(WorkerExecutionOperation::ProtocolMethod, worker, method) } + fn reconcile_stop_worker_operation( + &self, + request: WorkerExecutionStopRequest, + saved: Option<&crate::execution::WorkerExecutionTermination>, + ) -> WorkerExecutionResult { + let Some(saved) = saved else { + return self.stop_worker_operation(request); + }; + let reconcile = || -> Result { + let lock = self.worker_lock(&request.worker_ref)?; + let _guard = lock + .lock() + .map_err(|_| "worker operation lock is poisoned")?; + let execution = self + .workers + .lock() + .map_err(|_| "worker registry poisoned")? + .get(&request.worker_ref) + .cloned(); + let result = if let Some(execution) = execution { + // A warm original may still own resource cleanup after its join. + // Never send Stop to a replacement using the saved predecessor. + if !saved.confirms_scope_shutdown(&request.worker_ref, request.operation_id) + || execution + .tasks + .termination(&request.worker_ref, request.operation_id) + .as_ref() + != Some(saved) + || self + .read_termination_receipt_under_lock(&request.worker_ref, Some(saved))? + .as_ref() + != Some(saved) + { + return Err("saved Stop differs from current execution or candidate".into()); + } + self.stop_worker_operation_under_lock(request.clone()) + } else { + self.saved_stop_result_under_lock(&request, saved)? + }; + Ok(self.reconcile_stop_cleanup_under_lock(&request.worker_ref, result)) + }; + reconcile().unwrap_or_else(|message| { + WorkerExecutionResult::errored(WorkerExecutionOperation::Stop, message) + }) + } + fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult { let lock = match self.worker_lock(&request.worker_ref) { Ok(lock) => lock, @@ -8225,21 +8465,14 @@ mod tests { #[cfg(feature = "fs-store")] include!("subject_delete_create_tests.rs"); - // Both candidate and failure suites need the same guarded, checkout-owned - // temporary root, independently of Runtime's optional filesystem catalog. + // Candidate and failure suites keep the Yoi ownership preflight, while + // disposable source trees follow the runner's isolated temporary directory. fn restore_failure_tempdir() -> tempfile::TempDir { require_restore_test_environment(); - let target = Path::new(env!("CARGO_MANIFEST_DIR")) - .parent() - .unwrap() - .parent() - .unwrap() - .join("target"); - fs::create_dir_all(&target).unwrap(); tempfile::Builder::new() .prefix("r") .rand_bytes(3) - .tempdir_in(target) + .tempdir() .unwrap() } diff --git a/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs b/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs index a6827443d..7f56040d9 100644 --- a/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs @@ -87,18 +87,14 @@ struct PendingProfileSeedBackend { inner: Arc, } -impl WorkerExecutionBackend for PendingProfileSeedBackend { - fn backend_id(&self) -> &str { - self.inner.backend_id() - } - - fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { +impl PendingProfileSeedBackend { + fn seed_pending_head(&self, worker_ref: &WorkerRef) { let aggregate = self .inner .factory - .worker_aggregate_dir(&request.worker_ref) + .worker_aggregate_dir(&worker_ref) .unwrap(); - let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&request.worker_ref); + let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref); self.inner .run_on_adapter_runtime(async move { assert!( @@ -153,84 +149,6 @@ max_tokens = 100 Ok(()) }) .unwrap(); - WorkerExecutionSpawnResult::Connected { - worker_state: protocol::WorkerStateSnapshot::initial(), - workdir_attachments: Vec::new(), - } - } - - fn fetch_workspace_config( - &self, - request: WorkspaceConfigFetchRequest, - ) -> Result { - self.inner.fetch_workspace_config(request) - } - - fn observe_workspace_prompt_projection( - &self, - projection: worker::WorkspacePromptProjection, - ) -> Result<(), String> { - self.inner.observe_workspace_prompt_projection(projection) - } - - fn preflight_restore( - &self, - request: &WorkerExecutionRestoreRequest, - ) -> Result<(), WorkerExecutionResult> { - self.inner.preflight_restore(request) - } - - fn completed_restore_failure( - &self, - operation_id: crate::execution::WorkerLifecycleOperationId, - worker_ref: &WorkerRef, - ) -> Option { - self.inner - .completed_restore_failure(operation_id, worker_ref) - } - - fn restore_worker(&self, request: WorkerExecutionRestoreRequest) -> WorkerExecutionSpawnResult { - self.inner.restore_worker(request) - } - - fn restore_recovery_disposition( - &self, - operation_id: crate::execution::WorkerLifecycleOperationId, - worker_ref: &WorkerRef, - ) -> Option { - self.inner.restore_recovery_disposition(operation_id, worker_ref) - } - - fn reclaim_uncommitted_restore(&self, request: WorkerExecutionRestoreRequest) -> WorkerExecutionSpawnResult { - self.inner.reclaim_uncommitted_restore(request) - } - - fn reconcile_restore( - &self, - request: WorkerExecutionRestoreRequest, - ) -> WorkerExecutionSpawnResult { - self.inner.reconcile_restore(request) - } - - fn activate_restored_worker( - &self, - operation_id: crate::execution::WorkerLifecycleOperationId, - worker_ref: &WorkerRef, - ) -> Result<(), String> { - self.inner - .activate_restored_worker(operation_id, worker_ref) - } - - fn dispatch_input(&self, worker_ref: &WorkerRef, input: WorkerInput) -> WorkerExecutionResult { - self.inner.dispatch_input(worker_ref, input) - } - - fn dispatch_method(&self, worker_ref: &WorkerRef, method: Method) -> WorkerExecutionResult { - self.inner.dispatch_method(worker_ref, method) - } - - fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult { - self.inner.stop_worker_operation(request) } } @@ -283,20 +201,13 @@ impl ProfileFailureFixture { opens: provider_opens.clone(), }), ); - let execution: Arc = if pending_head { - Arc::new(PendingProfileSeedBackend { - inner: backend.clone(), - }) - } else { - backend.clone() - }; - let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend( + let mut runtime = EmbeddedRuntime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { - root: store, + root: store.clone(), runtime_id: "r".into(), display_name: None, }, - execution, + backend.clone(), ) .unwrap(); runtime.store_config_bundle(test_bundle()).unwrap(); @@ -308,7 +219,59 @@ impl ProfileFailureFixture { working_directory: working_directory_request(&repo), }]; } - let worker = runtime.create_worker(request).unwrap(); + let worker = if pending_head { + use crate::fs_store::{ + PersistedWorkerExecution, PersistedWorkerExecutionState, PersistedWorkerRecord, + }; + let worker_ref = WorkerRef::new(request.worker_id); + PendingProfileSeedBackend { + inner: backend.clone(), + } + .seed_pending_head(&worker_ref); + // This catalog has never launched a Controller. Stopped is a restore + // input fixture, not evidence of actual scope termination. + runtime + .test_runtime_store() + .write_worker_record(&PersistedWorkerRecord { + restore_guard: crate::runtime::RestoreGuard::default(), + worker_ref: worker_ref.clone(), + worker_id: request.worker_id, + profile: request.profile.clone(), + display_name: None, + profile_source: request.profile_source.reference(), + config_bundle: request.config_bundle.clone(), + created_at_ms: None, + status: crate::catalog::WorkerStatus::Stopped, + last_finished_submission_request_id: None, + execution_state: PersistedWorkerExecutionState::Available( + PersistedWorkerExecution { + request, + binding: None, + restore_intent: crate::catalog::WorkerRestoreIntent::Explicit, + }, + ), + workspace_id: None, + workdir_attachments: Vec::new(), + logical_workdir_attachments: Vec::new(), + }) + .unwrap(); + drop(runtime); + runtime = EmbeddedRuntime::with_fs_store_and_execution_backend( + crate::fs_store::FsRuntimeStoreOptions::new(store).with_runtime_id("r"), + backend.clone(), + ) + .unwrap(); + assert!( + runtime + .execution_termination(&worker_ref) + .unwrap() + .is_none() + ); + assert!(backend.workers.lock().unwrap().is_empty()); + runtime.worker_detail(&worker_ref).unwrap() + } else { + runtime.create_worker(request).unwrap() + }; let seed_handle = queue_inputs.then(|| { backend.workers.lock().unwrap()[&worker.worker_ref] .handle @@ -337,7 +300,9 @@ impl ProfileFailureFixture { Ok(()) }).unwrap(); } - runtime.stop_worker(&worker.worker_ref, None).unwrap(); + if !pending_head { + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + } // Count restore provider access, not the initial create/stop setup. provider_opens.store(0, Ordering::SeqCst); Self { @@ -697,10 +662,13 @@ impl ProfileFailureFixture { #[test] #[serial_test::serial(worker_allocation)] fn profile_restore_prior_effects_survive_read_only_reprepare_failure() { - for read_stage in [RuntimeRestoreStage::Metadata, RuntimeRestoreStage::Reconstruction] { + for read_stage in [ + RuntimeRestoreStage::Metadata, + RuntimeRestoreStage::Reconstruction, + ] { let activation_stage = RuntimeRestoreStage::ActivationEffects; let fixture = ProfileFailureFixture::new(Duration::from_secs(5)); - let _pause = fixture.runtime.test_pause_restore_continuation(); + let _pause = fixture.runtime.test_pause_restore_continuation(); let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload(); fixture.fail(activation_stage); let request = fixture @@ -754,6 +722,7 @@ fn profile_restore_prior_effects_survive_read_only_reprepare_failure() { #[serial_test::serial(worker_allocation)] fn profile_restore_prior_effects_from_pending_head_repair_survive_read_only_reprepare_failure() { let fixture = ProfileFailureFixture::with_pending_head(Duration::from_secs(5)); + let _pause = fixture.runtime.test_pause_restore_continuation(); let aggregate = fixture .backend .factory @@ -853,16 +822,21 @@ fn profile_restore_prior_effects_from_pending_head_repair_survive_read_only_repr let repaired_session = persisted_files(&aggregate.join("session")); for _ in 0..2 { + fixture + .backend + .wait_pending_restore(&pending, Duration::from_secs(5)) + .unwrap(); fixture.fail(RuntimeRestoreStage::Reconstruction); - let replay = fixture - .runtime - .restore_worker_operation(&fixture.worker.worker_ref, request.clone()) + // Drive one operation-owned reprepare before the public continuation + // can autonomously retry a transient failure again in the same call. + fixture + .backend + .retry_pending_restore_cleanup(&pending, true) + .unwrap(); + fixture + .backend + .wait_pending_restore(&pending, Duration::from_secs(5)) .unwrap(); - assert_eq!( - replay.state, - server_api::WorkerRestoreState::ReconciliationRequired, - "repair effect cannot become a rollback on the next read failure: {replay:?}" - ); fixture.assert_retained_restore_effects( &request, &pending, @@ -915,7 +889,11 @@ fn profile_restore_prior_effects_survive_rebuilt_candidate_pre_effect_activation .pending_restore(&fixture.worker.worker_ref) .unwrap() .unwrap(); - fixture.assert_retained_restore_effects(&request, &pending, RuntimeRestoreStage::ActivationEffects); + fixture.assert_retained_restore_effects( + &request, + &pending, + RuntimeRestoreStage::ActivationEffects, + ); let effected = persisted_files(&artifacts); assert_ne!(effected, pinned); @@ -1594,10 +1572,7 @@ fn profile_restore_publication_failure_retains_committed_candidate_until_cleanup "{result:?}" ); let diagnostic = serde_json::to_string(&result).unwrap(); - assert!( - diagnostic.contains("publication failed"), - "{diagnostic}" - ); + assert!(diagnostic.contains("publication failed"), "{diagnostic}"); assert!(diagnostic.contains("io=Some(Other)"), "{diagnostic}"); // Fail at the in-process publication boundary after activation effects. let pending = fixture @@ -1615,12 +1590,43 @@ fn profile_restore_publication_failure_retains_committed_candidate_until_cleanup assert!(!run_dir.join("worker.sock").exists()); assert!(fixture.backend.workers.lock().unwrap().is_empty()); assert!(pending.owns_prepared_candidate()); - assert!(pending.result.lock().unwrap().as_ref().unwrap().as_ref().unwrap().can_retry_publication()); - assert!(fixture.backend.factory.failed_restore_resources.lock().unwrap()[&fixture.worker.worker_ref].allocation.is_some()); - let durable: serde_json::Value = serde_json::from_slice(&fs::read( - fixture.root.path().join("s/workers").join(fixture.worker.worker_ref.worker_id.to_string()).join("worker.json") - ).unwrap()).unwrap(); - assert_eq!(durable["execution_state"]["execution"]["intent"]["commit"], "committed"); + assert!( + pending + .result + .lock() + .unwrap() + .as_ref() + .unwrap() + .as_ref() + .unwrap() + .can_retry_publication() + ); + assert!( + fixture + .backend + .factory + .failed_restore_resources + .lock() + .unwrap()[&fixture.worker.worker_ref] + .allocation + .is_some() + ); + let durable: serde_json::Value = serde_json::from_slice( + &fs::read( + fixture + .root + .path() + .join("s/workers") + .join(fixture.worker.worker_ref.worker_id.to_string()) + .join("worker.json"), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!( + durable["execution_state"]["execution"]["intent"]["commit"], + "committed" + ); assert!(durable["restore_guard"]["owners"][&request.request_id]["receipt"].is_null()); assert!( fixture @@ -1632,10 +1638,21 @@ fn profile_restore_publication_failure_retains_committed_candidate_until_cleanup ); // Explicit cleanup settles the retained owner without reconstructing it. drop(pending); - fixture.runtime.stop_worker(&fixture.worker.worker_ref, None).unwrap(); + fixture + .runtime + .stop_worker(&fixture.worker.worker_ref, None) + .unwrap(); fixture.assert_released(); assert!(!run_dir.exists()); - assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1); + assert_eq!( + fixture + .backend + .factory + .restore_faults + .executions + .load(Ordering::SeqCst), + 1 + ); } #[test] diff --git a/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs b/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs index fb46602c3..e9a7a7d2d 100644 --- a/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs @@ -878,8 +878,22 @@ mod publication { let second = EmbeddedRuntime::with_fs_store_and_execution_backend(options, fixture.backend.clone()).unwrap(); assert_eq!(second.restore_worker_operation_scoped(scope, &worker_ref, request).unwrap(), receipt); assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 0); - } else { + } else if commit == F::Committed { restarted.stop_worker_scoped(scope, &worker_ref, None).unwrap(); + assert!(restarted.execution_termination(&worker_ref).unwrap().is_some()); + } else { + // A cold cleanup-only owner can release resources but has no + // original Controller report/join. Preserve the pending history. + let error = restarted.stop_worker_scoped(scope, &worker_ref, None).unwrap_err(); + assert!(matches!(error, crate::error::RuntimeError::WorkerExecutionRejected { result, .. } if result.termination.is_none())); + assert!(restarted.execution_termination(&worker_ref).unwrap().is_none()); + assert!(restarted.worker_detail(&worker_ref).is_ok()); + let record = restarted.test_runtime_store().read_worker_record(&worker_ref).unwrap(); + assert!(matches!(record.execution_state, PersistedWorkerExecutionState::ReconciliationRequired(_))); + assert_eq!(fixture.canonical_files(), before); + let retry = restarted.restore_worker_operation(&worker_ref, request.clone()).unwrap(); + assert_eq!(retry.worker, receipt.worker); + assert_eq!(retry.state, server_api::WorkerRestoreState::ReconciliationRequired); } fixture.assert_released(); fixture.assert_quiet();