fix: recover job cleanup across runtime restart

crates/worker-runtime/src/catalog.rs

@@ -367,6 +367,10 @@ pub struct WorkerDetail {
     pub created_at_ms: Option<u64>,
     /// Whether the persisted execution metadata was valid when this identity was loaded.
     pub execution_metadata_available: bool,
+    /// Whether this Runtime process reconstructed the Worker from durable execution state.
+    /// This proves that work from the prior process incarnation is no longer running.
+    #[serde(default)]
+    pub execution_reconstructed: bool,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub worker_state: Option<protocol::WorkerStateSnapshot>,
     #[serde(default, skip_serializing_if = "Option::is_none")]

crates/worker-runtime/src/runtime.rs

@@ -1096,6 +1096,7 @@ impl Runtime {
                 status: WorkerStatus::Stopped,
                 worker_state: None,
                 last_finished_submission_request_id: None,
+                execution_reconstructed: false,
                 workspace_id: scope.map(|scope| scope.workspace_id.clone()),
                 profile: durable_request.profile.clone(),
                 display_name: durable_request.display_name.clone(),
@@ -4041,6 +4042,7 @@ impl RuntimeState {
                     status: worker.status,
                     worker_state: None,
                     last_finished_submission_request_id: worker.last_finished_submission_request_id,
+                    execution_reconstructed: true,
                     workspace_id: worker.workspace_id,
                     profile: worker.profile,
                     display_name: worker.display_name,
@@ -4939,6 +4941,7 @@ struct WorkerRecord {
     status: WorkerStatus,
     worker_state: Option<protocol::WorkerStateSnapshot>,
     last_finished_submission_request_id: Option<String>,
+    execution_reconstructed: bool,
     workspace_id: Option<String>,
     profile: ProfileSelector,
     display_name: Option<String>,
@@ -5008,6 +5011,7 @@ impl WorkerRecord {
             status: self.status,
             created_at_ms: self.created_at_ms,
             execution_metadata_available: self.execution_metadata_available,
+            execution_reconstructed: self.execution_reconstructed,
             worker_state: self.worker_state.clone(),
             workspace_id: self.workspace_id.clone(),
             workdir_attachments: self.workdir_attachments.clone(),
@@ -9283,6 +9287,12 @@ mod tests {
         let mut input = WorkerInput::user("run durable Backend Job attempt");
         input.submission_request_id = Some("job-attempt-request".to_string());
         runtime.send_input(&worker.worker_ref, input).unwrap();
+        assert!(
+            !runtime
+                .worker_detail(&worker.worker_ref)
+                .unwrap()
+                .execution_reconstructed
+        );
 
         let aggregate_path = root
             .join("workers")
@@ -9301,9 +9311,9 @@ mod tests {
             Arc::new(TestExecutionBackend::default()),
         )
         .unwrap();
-        let restored_state = restored
+        let restored_detail = restored.worker_detail(&worker.worker_ref).unwrap();
-            .worker_detail(&worker.worker_ref)
+        assert!(restored_detail.execution_reconstructed);
-            .unwrap()
+        let restored_state = restored_detail
             .worker_state
             .expect("automatic restore publishes fresh Worker state");
         assert_eq!(restored_state.last_command_id, 0);

crates/workspace-server/src/hosts.rs

@@ -259,6 +259,9 @@ pub struct InternalWorkerSummary {
     pub availability: protocol::subscription::SubscriptionWorkerAvailability,
     /// Runtime catalog lifecycle compatibility state.
     pub state: String,
+    /// Runtime-process reconstruction evidence used only for lifecycle coordination.
+    #[serde(skip)]
+    pub execution_reconstructed: bool,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub worker_state: Option<protocol::WorkerStateSnapshot>,
     pub last_seen_at: Option<String>,
@@ -2377,6 +2380,7 @@ impl EmbeddedWorkerRuntime {
                 summary.execution_metadata_available,
             )
             .to_string(),
+            execution_reconstructed: false,
             worker_state: summary.worker_state.clone(),
             last_seen_at: None,
             pinned: false,
@@ -2417,6 +2421,7 @@ impl EmbeddedWorkerRuntime {
             availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
             state: embedded_worker_state_label(detail.status, detail.execution_metadata_available)
                 .to_string(),
+            execution_reconstructed: detail.execution_reconstructed,
             worker_state: detail.worker_state.clone(),
             last_seen_at: None,
             pinned: false,
@@ -4368,6 +4373,7 @@ impl RemoteWorkerRuntime {
                 summary.execution_metadata_available,
             )
             .to_string(),
+            execution_reconstructed: false,
             worker_state: summary.worker_state.clone(),
             last_seen_at: None,
             pinned: false,
@@ -4406,6 +4412,7 @@ impl RemoteWorkerRuntime {
             availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
             state: embedded_worker_state_label(detail.status, detail.execution_metadata_available)
                 .to_string(),
+            execution_reconstructed: detail.execution_reconstructed,
             worker_state: detail.worker_state.clone(),
             last_seen_at: None,
             pinned: false,
@@ -6271,6 +6278,7 @@ pub fn placeholder_worker(host_id: impl Into<String>) -> InternalWorkerSummary {
         },
         availability: protocol::subscription::SubscriptionWorkerAvailability::Unavailable,
         state: "unsupported".to_string(),
+        execution_reconstructed: false,
         worker_state: None,
         last_seen_at: None,
         pinned: false,
@@ -6999,6 +7007,7 @@ mod tests {
                     },
                     availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
                     state: "available".to_string(),
+                    execution_reconstructed: false,
                     worker_state: None,
                     last_seen_at: None,
                     pinned: false,

crates/workspace-server/src/server.rs

@@ -3785,6 +3785,7 @@ impl WorkspaceApi {
             match self.runtime.worker(worker) {
                 Ok(detail)
                     if detail.state.eq_ignore_ascii_case("stopped")
+                        || detail.execution_reconstructed
                         || detail.worker_state.as_ref().is_some_and(|snapshot| {
                             snapshot.state == protocol::WorkerState::Idle
                                 && snapshot.last_finished_submission_request_id.as_deref()
@@ -3794,10 +3795,13 @@ impl WorkspaceApi {
                     return Ok(());
                 }
                 Ok(_) => {
-                    // The result capability runs inside this tracked Submit. The
+                    // In this process incarnation the result capability runs inside
-                    // Worker controller publishes this exact request id together
+                    // this tracked Submit. The controller publishes the exact request
-                    // with its post-Run state only after the invocation returns,
+                    // id only after the invocation returns. A Worker reconstructed
-                    // so an unrelated initial Idle snapshot cannot release cleanup.
+                    // from durable state is handled above instead: opening the
+                    // Worker store exclusively proves its prior process and any
+                    // pre-marker Run are gone, even if execution restoration has
+                    // not yet produced a fresh live snapshot.
                     tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                 }
                 Err(RuntimeRegistryError::UnknownWorker { .. }) => return Ok(()),
@@ -30374,6 +30378,7 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> InternalWorker
         singleton_key: None,
         tags: Vec::new(),
         state: "missing".to_string(),
+        execution_reconstructed: false,
         worker_state: None,
         last_seen_at: Some(record.updated_at.clone()),
         pinned: record.retention_state == "pinned",
@@ -34429,6 +34434,7 @@ mod tests {
                 },
                 availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
                 state: "idle".to_string(),
+                execution_reconstructed: false,
                 worker_state: None,
                 last_seen_at: None,
                 pinned: false,
@@ -41602,6 +41608,7 @@ mod tests {
             Option<Box<dyn FnOnce(&worker_runtime::identity::WorkerRef) + Send + 'static>>,
         >,
         input_request_ids: std::sync::Mutex<Vec<Option<String>>>,
+        accept_restores: std::sync::atomic::AtomicBool,
         inputs: std::sync::Mutex<Vec<(worker_runtime::identity::WorkerRef, String)>>,
         protocol_methods:
             std::sync::Mutex<Vec<(worker_runtime::identity::WorkerRef, protocol::Method)>>,
@@ -41626,6 +41633,7 @@ mod tests {
                 input_failure: std::sync::Mutex::new(None),
                 input_hook: std::sync::Mutex::new(None),
                 input_request_ids: std::sync::Mutex::new(Vec::new()),
+                accept_restores: std::sync::atomic::AtomicBool::new(false),
                 inputs: std::sync::Mutex::new(Vec::new()),
                 protocol_methods: std::sync::Mutex::new(Vec::new()),
             }
@@ -41637,6 +41645,11 @@ mod tests {
             *self.input_failure.lock().unwrap() = Some(message.into());
         }
 
+        fn accept_restores(&self) {
+            self.accept_restores
+                .store(true, std::sync::atomic::Ordering::SeqCst);
+        }
+
         fn before_input_returns(
             &self,
             hook: impl FnOnce(&worker_runtime::identity::WorkerRef) + Send + 'static,
@@ -41765,6 +41778,43 @@ mod tests {
             }
         }
 
+        fn restore_worker(
+            &self,
+            request: worker_runtime::execution::WorkerExecutionRestoreRequest,
+        ) -> worker_runtime::execution::WorkerExecutionSpawnResult {
+            if !self
+                .accept_restores
+                .load(std::sync::atomic::Ordering::SeqCst)
+            {
+                return worker_runtime::execution::WorkerExecutionSpawnResult::Rejected(
+                    worker_runtime::execution::WorkerExecutionResult::rejected(
+                        worker_runtime::execution::WorkerExecutionOperation::Restore,
+                        "deterministic test backend restore disabled",
+                    ),
+                );
+            }
+            let workdir_attachments = request
+                .workdir_attachments
+                .iter()
+                .map(|(alias, binding)| WorkingDirectoryAttachmentStatus {
+                    alias: alias.clone(),
+                    working_directory: binding.status(),
+                })
+                .collect();
+            self.contexts
+                .lock()
+                .unwrap()
+                .insert(request.worker_ref.clone(), request.context);
+            worker_runtime::execution::WorkerExecutionSpawnResult::Connected {
+                handle: worker_runtime::execution::WorkerExecutionHandle::new(
+                    request.worker_ref,
+                    self.backend_id(),
+                ),
+                worker_state: protocol::WorkerStateSnapshot::initial(),
+                workdir_attachments,
+            }
+        }
+
         fn dispatch_method(
             &self,
             handle: &worker_runtime::execution::WorkerExecutionHandle,
@@ -43122,14 +43172,14 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn backend_job_cleanup_restart_recovery_converges_interrupted_and_removed_workers() {
+    async fn backend_job_cleanup_restart_before_run_finish_marker_converges() {
         let temp = tempfile::tempdir().unwrap();
-        let (api, _execution) = test_api_with_recording_backend(temp.path()).await;
+        let config = test_server_config(temp.path());
         let request = BackendJobRequest {
-            job_id: "runner-cleanup-restart-1".to_string(),
+            job_id: "runner-cleanup-pre-marker-restart-1".to_string(),
-            purpose: "cleanup_restart_contract_check".to_string(),
+            purpose: "cleanup_pre_marker_restart_contract_check".to_string(),
             input_revision: "revision-1".to_string(),
-            input_ref: "test://runner/cleanup-restart/1".to_string(),
+            input_ref: "test://runner/cleanup-pre-marker-restart/1".to_string(),
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
@@ -43137,59 +43187,165 @@ mod tests {
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
         };
-        let dispatched = api.dispatch_backend_job(&request).unwrap();
+        let store = Arc::new(SqliteWorkspaceStore::open(config.database_path.clone()).unwrap());
-        let worker = dispatched.attempt.worker.clone().unwrap();
+        seed_test_registered_workspace(store.as_ref(), &config)
+            .await
+            .unwrap();
+        let execution = Arc::new(DeterministicExecutionBackend::default());
+        let embedded = EmbeddedWorkerRuntime::new_fs_store_with_execution_backend(
+            TEST_WORKSPACE_ID,
+            config.embedded_runtime_store_root.clone(),
+            execution.clone(),
+        )
+        .unwrap();
+        let binding = test_create_binding();
+        let runtime = embedded.subscription_runtime();
+        let runtime_worker_ref = worker_runtime::identity::WorkerRef::new(binding.worker_id);
+        let detail = runtime
+            .create_worker_scoped(
+                &worker_runtime::RuntimeWorkspaceScope::new(TEST_WORKSPACE_ID, "embedded-backend"),
+                worker_runtime::catalog::CreateWorkerRequest {
+                    worker_id: binding.worker_id,
+                    create_fingerprint: binding.create_fingerprint,
+                    profile: ProfileSelector::Builtin(request.profile.clone()),
+                    display_name: Some("job:cleanup_pre_marker_restart_contract_check".to_string()),
+                    config_bundle: None,
+                    profile_source: worker_runtime::catalog::ProfileSourceArchiveSource::Embedded {
+                        archive: crate::profile_settings::builtin_profile_source_archive(
+                            &ProfileSelector::Builtin(request.profile.clone()),
+                        )
+                        .unwrap(),
+                    },
+                    initial_input: None,
+                    workdir_attachment_requests: Vec::new(),
+                    workdir_attachments: Vec::new(),
+                    worker_observation_enabled: false,
+                    worker_observation_grants: Vec::new(),
+                    workspace_api: Some(test_worker_workspace_api(EMBEDDED_WORKER_RUNTIME_ID)),
+                    memory_settings: Some(test_worker_memory_settings()),
+                    subjektiv_attached: false,
+                },
+            )
+            .unwrap();
+        let worker =
+            RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, detail.worker_id.to_string());
+        runtime
+            .observe_worker_event(
+                &runtime_worker_ref,
+                protocol::Event::WorkerState {
+                    snapshot: protocol::WorkerStateSnapshot {
+                        last_command_id: 0,
+                        last_finished_submission_request_id: None,
+                        state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(
+                            protocol::WorkerRunState::Running,
+                        )),
+                    },
+                },
+            )
+            .unwrap();
+        let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
+        let reservation = store
+            .reserve_backend_job(TEST_WORKSPACE_ID, &request, &now)
+            .unwrap();
+        let (_, claimed) = store
+            .claim_backend_job_attempt_dispatch(
+                TEST_WORKSPACE_ID,
+                &request.job_id,
+                &reservation.attempt.attempt_id,
+                &now,
+            )
+            .unwrap();
+        assert!(claimed);
+        let runtime_run_id = crate::backend_job::runtime_run_id(&reservation.attempt.attempt_id);
+        store
+            .upsert_worker_registry(&WorkerRegistryRecord {
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+                worker: worker.clone(),
+                display_name: "job:cleanup_pre_marker_restart_contract_check".to_string(),
+                profile: Some(request.profile.clone()),
+                retention_state: "normal".to_string(),
+                transcript_ref: None,
+                session_ref: None,
+                summary_ref: None,
+                diagnostics_ref: None,
+                created_at: now.clone(),
+                updated_at: now.clone(),
+            })
+            .unwrap();
+        store
+            .bind_backend_job_attempt_worker(
+                TEST_WORKSPACE_ID,
+                &request.job_id,
+                &reservation.attempt.attempt_id,
+                &worker,
+                Some(&runtime_run_id),
+                &now,
+            )
+            .unwrap();
         seed_worker_session_for_cleanup(temp.path(), &worker);
-        api.store
+        let before_restart = runtime.worker_detail(&runtime_worker_ref).unwrap();
+        assert!(!before_restart.execution_reconstructed);
+        assert!(before_restart.worker_state.is_some_and(|snapshot| {
+            snapshot.last_command_id == 0
+                && snapshot.last_finished_submission_request_id.is_none()
+                && snapshot.state
+                    == protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(
+                        protocol::WorkerRunState::Running,
+                    ))
+        }));
+        store
             .accept_backend_job_result(
                 TEST_WORKSPACE_ID,
                 &worker,
                 &BackendJobResultSubmission {
                     job_id: request.job_id.clone(),
-                    attempt_id: dispatched.attempt.attempt_id.clone(),
+                    attempt_id: reservation.attempt.attempt_id.clone(),
                     input_revision: request.input_revision.clone(),
                     result: serde_json::json!({"persisted": true}),
                 },
-                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+                &now,
             )
             .unwrap();
-        let (claimed, did_claim) = api
-            .store
-            .claim_backend_job_worker_cleanup(
-                TEST_WORKSPACE_ID,
-                &request.job_id,
-                &dispatched.attempt.attempt_id,
-                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-            )
-            .unwrap();
-        assert!(did_claim);
         assert_eq!(
-            claimed.worker_cleanup_state,
+            store
-            Some(BackendJobWorkerCleanupState::Executing)
-        );
-
-        api.recover_backend_jobs().unwrap();
-        let cleaned = wait_for_backend_job_worker_cleanup(
-            &api,
-            &request.job_id,
-            &dispatched.attempt.attempt_id,
-            BackendJobWorkerCleanupState::Completed,
-        )
-        .await;
-        assert_eq!(cleaned.result, Some(serde_json::json!({"persisted": true})));
-        api.recover_backend_jobs().unwrap();
-        assert_eq!(
-            api.store
                 .get_backend_job_attempt(
                     TEST_WORKSPACE_ID,
                     &request.job_id,
-                    &dispatched.attempt.attempt_id,
+                    &reservation.attempt.attempt_id,
                 )
                 .unwrap()
                 .unwrap()
                 .worker_cleanup_state,
-            Some(BackendJobWorkerCleanupState::Completed)
+            Some(BackendJobWorkerCleanupState::Pending)
         );
+        let attempt_id = reservation.attempt.attempt_id;
+        drop(runtime);
+        drop(embedded);
+        drop(store);
+        drop(execution);
+
+        let restarted_store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap();
+        let restarted_execution = Arc::new(DeterministicExecutionBackend::default());
+        restarted_execution.accept_restores();
+        let restarted = WorkspaceApi::new_with_execution_backend(
+            config,
+            Arc::new(restarted_store),
+            restarted_execution,
+        )
+        .await
+        .unwrap();
+        let cleaned = wait_for_backend_job_worker_cleanup(
+            &restarted,
+            &request.job_id,
+            &attempt_id,
+            BackendJobWorkerCleanupState::Completed,
+        )
+        .await;
+        assert_eq!(cleaned.result, Some(serde_json::json!({"persisted": true})));
+        assert!(matches!(
+            restarted.runtime.worker(&worker),
+            Err(RuntimeRegistryError::UnknownWorker { .. })
+        ));
     }
 
     #[tokio::test]