From 6383b20501914bbf24a78438855669e29b7466dc Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 5 Nov 2026 04:11:08 +0900 Subject: [PATCH] fix(runtime): reconcile cold Stop owners before Subject cleanup Retain exact canonical termination for pending Stop recovery, separate controller join from factory cleanup, and confirm removed originals only from an existing Subject receipt. Preserve cancellation and missing-proof refusal boundaries. Bind HTTP/WS fixtures to real scoped Controller ownership. Checkpoint validation: normal Runtime lib build and 38 focused cases passed; normal affected suite previously 619/627, not rerun as all-green. Root check, format, names and diff pass. Migration focused runner 24 pass. Incomplete and unapproved: ordinary removal-before-Subject history consumer, one connected Server restored-controller teardown, final affected suites and fresh whole-source review remain. --- crates/worker-runtime/src/execution.rs | 77 +- crates/worker-runtime/src/http_server.rs | 483 ++++++++--- .../src/http_server/controller_fixture.rs | 515 +++++++++++ .../src/http_server/controller_ws_tests.rs | 170 ++++ .../tests/session_material_tests.rs | 30 +- ...ubject_job_removal_purge_recovery_tests.rs | 28 +- crates/worker-runtime/src/runtime.rs | 87 +- .../src/runtime/subject_delete.rs | 159 +++- .../src/runtime/subject_delete_tests.rs | 103 ++- .../src/subject_delete_backend_tests.rs | 815 +++++++++++++++++- .../src/subject_delete_create_tests.rs | 78 +- crates/worker-runtime/src/worker_backend.rs | 365 ++++++-- .../worker_backend_restore_failure_tests.rs | 259 +++--- ...orker_backend_restore_publication_tests.rs | 16 +- 14 files changed, 2760 insertions(+), 425 deletions(-) create mode 100644 crates/worker-runtime/src/http_server/controller_fixture.rs create mode 100644 crates/worker-runtime/src/http_server/controller_ws_tests.rs 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();