From 027a346e9dfd9025d49a7063d48f42b518e07493 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 1 Oct 2026 11:58:49 +0900 Subject: [PATCH] fix tracked job dispatch and delivery recovery --- crates/workspace-server/src/hosts.rs | 55 +++- crates/workspace-server/src/server.rs | 344 ++++++++++++++++++++++---- crates/workspace-server/src/store.rs | 94 ++++--- docs/design/durable-operations.md | 48 ++-- 4 files changed, 423 insertions(+), 118 deletions(-) diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 394d3316b..01f956d74 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -710,6 +710,8 @@ pub struct WorkerInputRequest { pub kind: WorkerInputKind, pub content: String, #[serde(default, skip_serializing_if = "Option::is_none")] + pub submission_request_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub segments: Option>, } @@ -730,9 +732,18 @@ pub struct WorkerCompletionsResult { pub diagnostics: Vec, } +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum WorkerInputDisposition { + Accepted, + Rejected, + Unknown, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkerInputResult { pub state: InternalWorkerOperationState, + pub disposition: WorkerInputDisposition, #[serde(flatten)] pub worker: RuntimeWorkerRef, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -1207,6 +1218,7 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { fn send_input(&self, worker_id: &str, _request: WorkerInputRequest) -> WorkerInputResult { WorkerInputResult { state: InternalWorkerOperationState::Unsupported, + disposition: WorkerInputDisposition::Rejected, worker: RuntimeWorkerRef::new(self.runtime_id().to_string(), worker_id.to_string()), runtime_run_id: None, diagnostics: vec![diagnostic( @@ -2993,6 +3005,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { return embedded_input_rejected( &self.runtime_id, worker_id, + WorkerInputDisposition::Rejected, diagnostic( "embedded_worker_execution_unavailable", HostDiagnosticSeverity::Info, @@ -3006,6 +3019,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { return embedded_input_rejected( &self.runtime_id, worker_id, + WorkerInputDisposition::Rejected, diagnostic( "embedded_worker_id_invalid", HostDiagnosticSeverity::Warning, @@ -3022,21 +3036,34 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer, }, content: request.content, - submission_request_id: None, + submission_request_id: request.submission_request_id, segments: request.segments, }; match self.runtime.send_input(&worker_ref, input) { Ok(ack) => WorkerInputResult { state: InternalWorkerOperationState::Accepted, + disposition: WorkerInputDisposition::Accepted, worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), - runtime_run_id: ack.submission.map(|submission| submission.submission_id), + runtime_run_id: ack + .submission + .map(|submission| submission.submission_request_id), diagnostics: Vec::new(), }, - Err(error) => embedded_input_rejected( - &self.runtime_id, - worker_id, - embedded_runtime_diagnostic(&error), - ), + Err(error) => { + let disposition = match &error { + EmbeddedRuntimeError::WorkerExecutionRejected { + outcome: worker_runtime::execution::WorkerExecutionOutcome::Errored, + .. + } => WorkerInputDisposition::Unknown, + _ => WorkerInputDisposition::Rejected, + }; + embedded_input_rejected( + &self.runtime_id, + worker_id, + disposition, + embedded_runtime_diagnostic(&error), + ) + } } } @@ -5000,7 +5027,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer, }, content: request.content, - submission_request_id: None, + submission_request_id: request.submission_request_id, segments: request.segments, }; let input = match runtime_contract_convert(input) { @@ -5008,6 +5035,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { Err(diagnostic) => { return WorkerInputResult { state: InternalWorkerOperationState::Rejected, + disposition: WorkerInputDisposition::Rejected, worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), runtime_run_id: None, diagnostics: vec![diagnostic], @@ -5022,11 +5050,12 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { ) { Ok(response) => WorkerInputResult { state: InternalWorkerOperationState::Accepted, + disposition: WorkerInputDisposition::Accepted, worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), runtime_run_id: response .ack .submission - .map(|submission| submission.submission_id), + .map(|submission| submission.submission_request_id), diagnostics: Vec::new(), }, Err(diagnostic) => remote_input_rejected(&self.runtime_id, worker_id, diagnostic), @@ -5410,10 +5439,12 @@ fn profile_display_name(profile_label: &str) -> String { fn embedded_input_rejected( runtime_id: &str, worker_id: &str, + disposition: WorkerInputDisposition, diagnostic: RuntimeDiagnostic, ) -> WorkerInputResult { WorkerInputResult { state: InternalWorkerOperationState::Rejected, + disposition, worker: RuntimeWorkerRef::new(runtime_id.to_string(), worker_id.to_string()), runtime_run_id: None, diagnostics: vec![diagnostic], @@ -5427,6 +5458,7 @@ fn remote_input_rejected( ) -> WorkerInputResult { WorkerInputResult { state: InternalWorkerOperationState::Rejected, + disposition: WorkerInputDisposition::Unknown, worker: RuntimeWorkerRef::new(runtime_id.to_string(), worker_id.to_string()), runtime_run_id: None, diagnostics: vec![diagnostic], @@ -7105,10 +7137,13 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "hello".to_string(), + submission_request_id: Some("tracked-input-1".to_string()), segments: None, }, ); assert_eq!(input.state, InternalWorkerOperationState::Accepted); + assert_eq!(input.disposition, WorkerInputDisposition::Accepted); + assert_eq!(input.runtime_run_id.as_deref(), Some("tracked-input-1")); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); loop { @@ -7200,6 +7235,7 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "hello embedded runtime".to_string(), + submission_request_id: None, segments: None, }, ) @@ -7484,6 +7520,7 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "hello remote".to_string(), + submission_request_id: None, segments: None, }, ) diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 2f408ead5..dd6f8aa06 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -160,11 +160,12 @@ use crate::hosts::{ RemoteWorkerRuntime, RuntimeDiagnostic, RuntimePingFailureKind, RuntimeRegistry, RuntimeRegistryError, RuntimeRegistryUnregisterResult, TicketWorkerRole, WorkerCompletionsRequest, WorkerCompletionsResult, WorkerControlOperation, WorkerCreateBinding, - WorkerInputKind, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, - WorkerLifecycleResult, WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, - WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, WorkerTicketAssignmentRequest, - WorkspaceRuntimeAuthorization, is_disallowed_remote_runtime_address, - is_loopback_runtime_origin, worker_spawn_create_fingerprint, workspace_worker_summary, + WorkerInputDisposition, WorkerInputKind, WorkerInputRequest, WorkerInputResult, + WorkerLifecycleRequest, WorkerLifecycleResult, WorkerSpawnAcceptanceRequirement, + WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, + WorkerTicketAssignmentRequest, WorkspaceRuntimeAuthorization, + is_disallowed_remote_runtime_address, is_loopback_runtime_origin, + worker_spawn_create_fingerprint, workspace_worker_summary, }; use crate::memory_backend::execute_memory_backend_operation_with_authority; use crate::memory_staging::{ @@ -3245,8 +3246,37 @@ impl WorkspaceApi { recover_workdir_removals(&api)?; recover_runtime_removals(&api).await; let recovery_now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); - api.store + let deliveries = api + .store .recover_backend_job_deliveries(&api.config.workspace_id, &recovery_now)?; + for delivery in deliveries { + let Some(job) = api + .store + .get_backend_job(&api.config.workspace_id, &delivery.job_id)? + else { + continue; + }; + let Some(attempt) = api.store.get_backend_job_attempt( + &api.config.workspace_id, + &delivery.job_id, + &delivery.attempt_id, + )? + else { + continue; + }; + if job.state == BackendJobState::Completed + && attempt.state == BackendJobAttemptState::Completed + { + api.deliver_backend_job_result( + &BackendJobResultAcceptance { + job, + attempt, + replayed: true, + }, + &delivery.target, + ); + } + } api.recover_backend_jobs()?; Ok(api) } @@ -3269,8 +3299,11 @@ impl WorkspaceApi { self.dispatch_backend_job_reservation(reservation) } - /// Explicit re-evaluation creates a new bounded attempt. Transport replay - /// must call `dispatch_backend_job` with the original immutable request. + /// Explicit re-evaluation creates a new bounded attempt only after a + /// definitive failure. Unknown execution outcomes require manual attention + /// and cannot be replayed because the original tracked run may still execute. + /// Transport replay must call `dispatch_backend_job` with the original + /// immutable request. pub fn retry_backend_job(&self, job_id: &str) -> Result { let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); let reservation = @@ -3361,12 +3394,13 @@ impl WorkspaceApi { } }; let bound_at = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); + let runtime_run_id = format!("backend-job-run:{}", reservation.attempt.attempt_id); let (commit, newly_bound) = self.store.bind_backend_job_attempt_worker( &self.config.workspace_id, &request.job_id, &reservation.attempt.attempt_id, &worker.worker, - None, + Some(&runtime_run_id), &bound_at, )?; if newly_bound { @@ -3402,6 +3436,7 @@ impl WorkspaceApi { WorkerInputRequest { kind: WorkerInputKind::User, content: worker_input, + submission_request_id: Some(runtime_run_id.clone()), segments: None, }, ) { @@ -3426,57 +3461,45 @@ impl WorkspaceApi { .first() .map(|diagnostic| diagnostic.message.as_str()) .unwrap_or("embedded Runtime rejected Backend Job input"); + let (state, category) = match submitted.disposition { + WorkerInputDisposition::Unknown => { + (BackendJobAttemptState::Unknown, "input_outcome_unknown") + } + WorkerInputDisposition::Accepted | WorkerInputDisposition::Rejected => { + (BackendJobAttemptState::Failed, "input_rejected") + } + }; self.store.finish_backend_job_attempt( &self.config.workspace_id, &request.job_id, &reservation.attempt.attempt_id, - BackendJobAttemptState::Failed, - "input_rejected", + state, + category, detail, &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; return Err(Error::RuntimeOperationFailed { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), - code: "backend_job_input_rejected".to_string(), + code: format!("backend_job_{category}"), message: detail.to_string(), }); } - let Some(runtime_run_id) = submitted.runtime_run_id.as_deref() else { + if submitted.runtime_run_id.as_deref() != Some(runtime_run_id.as_str()) { self.store.finish_backend_job_attempt( &self.config.workspace_id, &request.job_id, &reservation.attempt.attempt_id, BackendJobAttemptState::Unknown, - "runtime_run_missing", - "embedded Runtime accepted Backend Job input without run evidence", + "runtime_run_mismatch", + "embedded Runtime accepted Backend Job input without matching tracked run evidence", &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; return Err(Error::RuntimeOperationFailed { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), - code: "backend_job_runtime_run_missing".to_string(), - message: "embedded Runtime accepted Backend Job input without run evidence" + code: "backend_job_runtime_run_mismatch".to_string(), + message: "embedded Runtime accepted Backend Job input without matching tracked run evidence" .to_string(), }); - }; - if let Err(error) = self.store.bind_backend_job_attempt_runtime_run( - &self.config.workspace_id, - &request.job_id, - &reservation.attempt.attempt_id, - &worker.worker, - runtime_run_id, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ) { - let detail = error.to_string(); - self.store.finish_backend_job_attempt( - &self.config.workspace_id, - &request.job_id, - &reservation.attempt.attempt_id, - BackendJobAttemptState::Unknown, - "runtime_run_persistence_failed", - &detail, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - )?; - return Err(error); } let job = self .store @@ -3681,6 +3704,7 @@ impl WorkspaceApi { WorkerInputRequest { kind: WorkerInputKind::Notify, content, + submission_request_id: Some(delivery.delivery_id.clone()), segments: None, }, ); @@ -3688,14 +3712,19 @@ impl WorkspaceApi { Ok(result) if result.state == InternalWorkerOperationState::Accepted => { (BackendJobDeliveryState::Completed, None, None) } + Ok(result) if result.disposition == WorkerInputDisposition::Unknown => ( + BackendJobDeliveryState::Unknown, + Some("notification_outcome_unknown".to_string()), + result.diagnostics.first().map(|item| item.message.clone()), + ), Ok(result) => ( BackendJobDeliveryState::Failed, Some("notification_rejected".to_string()), result.diagnostics.first().map(|item| item.message.clone()), ), Err(error) => ( - BackendJobDeliveryState::Failed, - Some("notification_unavailable".to_string()), + BackendJobDeliveryState::Unknown, + Some("notification_outcome_unknown".to_string()), Some(error.message()), ), }; @@ -5582,6 +5611,7 @@ fn worker_input_request_from_api( Ok(WorkerInputRequest { kind, content: request.content, + submission_request_id: None, segments, }) } @@ -14583,6 +14613,7 @@ fn notify_ticket_recipients( WorkerInputRequest { kind: WorkerInputKind::Notify, content: content.clone(), + submission_request_id: None, segments: None, }, ); @@ -15683,6 +15714,7 @@ fn dispatch_orchestrator_queue_attention(api: &WorkspaceApi) { WorkerInputRequest { kind: WorkerInputKind::Notify, content, + submission_request_id: None, segments: None, }, ) @@ -15963,6 +15995,7 @@ fn try_reuse_memory_consolidation_worker( WorkerInputRequest { kind: WorkerInputKind::User, content: input_content.to_string(), + submission_request_id: None, segments: None, }, ) @@ -33590,6 +33623,10 @@ mod tests { materializer: worker_runtime::working_directory::RuntimeGitMaterializer, spawn_failure: std::sync::Mutex>, input_failure: std::sync::Mutex>, + input_hook: std::sync::Mutex< + Option>, + >, + input_request_ids: std::sync::Mutex>>, inputs: std::sync::Mutex>, protocol_methods: std::sync::Mutex>, @@ -33612,6 +33649,8 @@ mod tests { ), spawn_failure: std::sync::Mutex::new(None), input_failure: std::sync::Mutex::new(None), + input_hook: std::sync::Mutex::new(None), + input_request_ids: std::sync::Mutex::new(Vec::new()), inputs: std::sync::Mutex::new(Vec::new()), protocol_methods: std::sync::Mutex::new(Vec::new()), } @@ -33623,6 +33662,22 @@ mod tests { *self.input_failure.lock().unwrap() = Some(message.into()); } + fn before_input_returns( + &self, + hook: impl FnOnce(&worker_runtime::identity::WorkerRef) + Send + 'static, + ) { + *self.input_hook.lock().unwrap() = Some(Box::new(hook)); + } + + fn take_input_request_ids(&self) -> Vec> { + std::mem::take( + &mut *self + .input_request_ids + .lock() + .expect("input request ids lock"), + ) + } + fn take_inputs(&self) -> Vec<(worker_runtime::identity::WorkerRef, String)> { std::mem::take(&mut *self.inputs.lock().expect("inputs lock")) } @@ -33783,10 +33838,17 @@ mod tests { handle: &worker_runtime::execution::WorkerExecutionHandle, input: worker_runtime::interaction::WorkerInput, ) -> worker_runtime::execution::WorkerExecutionResult { + self.input_request_ids + .lock() + .expect("input request ids lock") + .push(input.submission_request_id.clone()); self.inputs .lock() .expect("inputs lock") .push((handle.worker_ref().clone(), input.content.clone())); + if let Some(hook) = self.input_hook.lock().unwrap().take() { + hook(handle.worker_ref()); + } if let Some(message) = self.input_failure.lock().unwrap().clone() { return worker_runtime::execution::WorkerExecutionResult::errored( worker_runtime::execution::WorkerExecutionOperation::Input, @@ -34054,6 +34116,173 @@ mod tests { assert_eq!(accepted.job.state, BackendJobState::Completed); } + #[tokio::test] + async fn backend_job_fast_result_is_accepted_after_prebound_tracked_run() { + let temp = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(temp.path()).await; + let request = BackendJobRequest { + job_id: "runner-fast-result-1".to_string(), + purpose: "fast_result_contract_check".to_string(), + input_revision: "revision-1".to_string(), + input_ref: "test://runner/fast-result/1".to_string(), + input: serde_json::json!({"value": "immutable"}), + instruction: "Return a structured check result.".to_string(), + profile: "builtin:backend-job".to_string(), + source_worker: None, + notification_target: None, + limits: crate::backend_job::BackendJobLimits::default(), + }; + let acceptance_observed = Arc::new(std::sync::Mutex::new(None)); + let hook_api = api.clone(); + let hook_request = request.clone(); + let hook_observed = acceptance_observed.clone(); + execution.before_input_returns(move |worker| { + let source = + RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, worker.worker_id.to_string()); + let accepted = hook_api + .accept_backend_job_result( + &source, + &BackendJobResultSubmission { + job_id: hook_request.job_id.clone(), + attempt_id: crate::backend_job::attempt_id(&hook_request.job_id, 1), + input_revision: hook_request.input_revision.clone(), + result: serde_json::json!({"valid": true}), + }, + ) + .is_ok(); + *hook_observed.lock().unwrap() = Some(accepted); + }); + + let dispatched = api.dispatch_backend_job(&request).unwrap(); + assert_eq!(*acceptance_observed.lock().unwrap(), Some(true)); + assert_eq!(dispatched.job.state, BackendJobState::Completed); + let expected_run_id = format!( + "backend-job-run:{}", + crate::backend_job::attempt_id(&request.job_id, 1) + ); + assert_eq!( + dispatched.attempt.runtime_run_id.as_deref(), + Some(expected_run_id.as_str()) + ); + } + + #[tokio::test] + async fn backend_job_delivery_recovery_replays_same_tracked_notification_id() { + let temp = tempfile::tempdir().unwrap(); + let config = test_server_config(temp.path()); + let store = SqliteWorkspaceStore::open(config.database_path.clone()).unwrap(); + let execution = Arc::new(DeterministicExecutionBackend::default()); + let api = WorkspaceApi::new_with_execution_backend( + config.clone(), + Arc::new(store), + execution.clone(), + ) + .await + .unwrap(); + let target = api + .spawn_workspace_worker( + EMBEDDED_WORKER_RUNTIME_ID, + WorkerSpawnRequest { + requested_worker_name: Some("job-notification-target".to_string()), + intent: WorkerSpawnIntent::WorkspaceCompanion, + acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: 0, + }, + profile: ProfileSelector::Builtin("builtin:companion".to_string()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachment_requests: Vec::new(), + resolved_workdir_attachment_requests: Vec::new(), + resolved_workdir_attachments: Vec::new(), + resolved_config_bundle: Some(runtime_test_bundle()), + resolved_worker_observation_enabled: false, + resolved_worker_observation_grants: Vec::new(), + resolved_workspace_api: None, + resolved_memory_settings: None, + resolved_control_operation: None, + }, + ) + .unwrap() + .worker + .unwrap() + .worker; + let request = BackendJobRequest { + job_id: "runner-delivery-recovery-1".to_string(), + purpose: "delivery_recovery_contract_check".to_string(), + input_revision: "revision-1".to_string(), + input_ref: "test://runner/delivery-recovery/1".to_string(), + input: serde_json::json!({"value": "immutable"}), + instruction: "Return a structured check result.".to_string(), + profile: "builtin:backend-job".to_string(), + source_worker: None, + notification_target: Some(target.clone()), + limits: crate::backend_job::BackendJobLimits { + timeout_seconds: 1, + ..Default::default() + }, + }; + let dispatched = api.dispatch_backend_job(&request).unwrap(); + let worker = dispatched.attempt.worker.clone().unwrap(); + let acceptance = api + .store + .accept_backend_job_result( + TEST_WORKSPACE_ID, + &worker, + &BackendJobResultSubmission { + job_id: request.job_id.clone(), + attempt_id: dispatched.attempt.attempt_id.clone(), + input_revision: request.input_revision.clone(), + result: serde_json::json!({"valid": true}), + }, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + let delivery_id = format!( + "{}:{}:notification", + request.job_id, acceptance.attempt.attempt_id + ); + let (_, claimed) = api + .store + .reserve_backend_job_delivery( + TEST_WORKSPACE_ID, + &delivery_id, + &request.job_id, + &target, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + assert!(claimed); + let recoverable = api + .store + .recover_backend_job_deliveries( + TEST_WORKSPACE_ID, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + assert_eq!(recoverable.len(), 1); + execution.take_input_request_ids(); + api.deliver_backend_job_result(&acceptance, &target); + let request_ids = execution.take_input_request_ids(); + assert!( + request_ids + .iter() + .any(|request_id| request_id.as_deref() == Some(delivery_id.as_str())), + "recovery must replay the durable delivery id as the tracked notification id: {request_ids:?}" + ); + let (delivery, claimed) = api + .store + .reserve_backend_job_delivery( + TEST_WORKSPACE_ID, + &delivery_id, + &request.job_id, + &target, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + assert!(!claimed); + assert_eq!(delivery.state, BackendJobDeliveryState::Completed); + } + #[tokio::test] async fn backend_job_spawn_transport_error_terminalizes_reserved_attempt() { let temp = tempfile::tempdir().unwrap(); @@ -34089,7 +34318,7 @@ mod tests { } #[tokio::test] - async fn backend_job_input_failure_is_durable_and_retryable_without_reusing_attempt() { + async fn backend_job_input_error_is_durable_unknown_and_not_automatically_retried() { let temp = tempfile::tempdir().unwrap(); let (api, execution) = test_api_with_recording_backend(temp.path()).await; execution.reject_inputs("model transport unavailable"); @@ -34116,20 +34345,25 @@ mod tests { ) .unwrap() .unwrap(); - assert_eq!(failed.state, BackendJobAttemptState::Failed); - assert_eq!(failed.failure_category.as_deref(), Some("input_rejected")); - let _retry_error = api.retry_backend_job(&request.job_id).unwrap_err(); - let second = api - .store - .get_backend_job_attempt( - TEST_WORKSPACE_ID, - &request.job_id, - &crate::backend_job::attempt_id(&request.job_id, 2), - ) - .unwrap() - .unwrap(); - assert_eq!(second.attempt, 2); - assert_eq!(second.state, BackendJobAttemptState::Failed); + assert_eq!(failed.state, BackendJobAttemptState::Unknown); + assert_eq!( + failed.failure_category.as_deref(), + Some("input_outcome_unknown") + ); + assert!( + api.retry_backend_job(&request.job_id).is_err(), + "an ambiguous input outcome must not be retried automatically" + ); + assert!( + api.store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &request.job_id, + &crate::backend_job::attempt_id(&request.job_id, 2), + ) + .unwrap() + .is_none() + ); } #[tokio::test] @@ -44454,6 +44688,7 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "persist me".to_string(), + submission_request_id: None, segments: None, }, ) @@ -44514,6 +44749,7 @@ mod tests { WorkerInputRequest { kind: WorkerInputKind::User, content: "should not be routed to corrupted handle".to_string(), + submission_request_id: None, segments: None, }, ) diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index e021a02c1..dd6b3d720 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -1532,7 +1532,11 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { target: &RuntimeWorkerRef, now: &str, ) -> Result<(BackendJobDeliveryRecord, bool)>; - fn recover_backend_job_deliveries(&self, workspace_id: &str, now: &str) -> Result; + fn recover_backend_job_deliveries( + &self, + workspace_id: &str, + now: &str, + ) -> Result>; fn finish_backend_job_delivery( &self, workspace_id: &str, @@ -5915,9 +5919,10 @@ impl ControlPlaneStore for SqliteWorkspaceStore { let previous_id = attempt_id(job_id, job.current_attempt); let previous = read_backend_job_attempt(&tx, workspace_id, job_id, &previous_id)? .ok_or_else(|| Error::Store("Backend Job current attempt is missing".to_string()))?; - if !matches!(previous.state, BackendJobAttemptState::Failed | BackendJobAttemptState::Unknown) { + if previous.state != BackendJobAttemptState::Failed { return Err(Error::InvalidInput( - "Backend Job re-evaluation requires a failed or unknown current attempt".to_string(), + "Backend Job re-evaluation requires a definitively failed current attempt; unknown execution outcomes cannot be retried" + .to_string(), )); } if job.current_attempt >= job.request.limits.max_attempts { @@ -6285,17 +6290,39 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } - fn recover_backend_job_deliveries(&self, workspace_id: &str, now: &str) -> Result { + fn recover_backend_job_deliveries( + &self, + workspace_id: &str, + now: &str, + ) -> Result> { self.with_conn_mut(|conn| { - conn.execute( + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + tx.execute( "UPDATE backend_job_deliveries - SET state = 'unknown', failure_category = 'restart_ambiguous', - failure_detail = 'delivery was in progress when the Backend restarted; notification will not be replayed', + SET state = 'pending', failure_category = NULL, failure_detail = NULL, updated_at = ?2 - WHERE workspace_id = ?1 AND state = 'sending'", + WHERE workspace_id = ?1 AND state IN ('sending', 'unknown')", params![workspace_id, now], - ) - .map_err(Into::into) + )?; + let mut statement = tx.prepare( + "SELECT delivery_id FROM backend_job_deliveries + WHERE workspace_id = ?1 AND state = 'pending' + ORDER BY created_at, delivery_id", + )?; + let delivery_ids = statement + .query_map(params![workspace_id], |row| row.get::<_, String>(0))? + .collect::>>()?; + drop(statement); + let deliveries = delivery_ids + .into_iter() + .map(|delivery_id| { + read_backend_job_delivery(&tx, workspace_id, &delivery_id)?.ok_or_else(|| { + Error::Store("recoverable Backend Job delivery disappeared".to_string()) + }) + }) + .collect::>>()?; + tx.commit()?; + Ok(deliveries) }) } @@ -19669,8 +19696,9 @@ INSERT INTO worker_registry ( .unwrap(); assert!(!claimed); - // A restart cannot know whether an in-flight transport completed, so - // it terminalizes the claim as unknown instead of sending twice. + // Restart recovery returns ambiguous in-flight claims to pending. The + // caller retries the same durable delivery id, which is also the + // Runtime notification request id and is deduplicated by the Worker. let (_, claimed) = store .reserve_backend_job_delivery( workspace_id, @@ -19681,12 +19709,12 @@ INSERT INTO worker_registry ( ) .unwrap(); assert!(claimed); - assert_eq!( - store - .recover_backend_job_deliveries(workspace_id, "2026-09-01T00:00:09Z") - .unwrap(), - 1 - ); + let recoverable = store + .recover_backend_job_deliveries(workspace_id, "2026-09-01T00:00:09Z") + .unwrap(); + assert_eq!(recoverable.len(), 1); + assert_eq!(recoverable[0].delivery_id, "delivery-2"); + assert_eq!(recoverable[0].state, BackendJobDeliveryState::Pending); let (delivery, claimed) = store .reserve_backend_job_delivery( workspace_id, @@ -19696,24 +19724,18 @@ INSERT INTO worker_registry ( "2026-09-01T00:00:10Z", ) .unwrap(); - assert!(!claimed); - assert_eq!(delivery.state, BackendJobDeliveryState::Unknown); - assert_eq!( - delivery.failure_category.as_deref(), - Some("restart_ambiguous") - ); - assert!( - store - .finish_backend_job_delivery( - workspace_id, - "delivery-2", - BackendJobDeliveryState::Completed, - None, - None, - "2026-09-01T00:00:11Z", - ) - .is_err() - ); + assert!(claimed); + assert_eq!(delivery.state, BackendJobDeliveryState::Sending); + store + .finish_backend_job_delivery( + workspace_id, + "delivery-2", + BackendJobDeliveryState::Completed, + None, + None, + "2026-09-01T00:00:11Z", + ) + .unwrap(); assert_eq!( store .get_backend_job(workspace_id, &request.job_id) diff --git a/docs/design/durable-operations.md b/docs/design/durable-operations.md index 39269d87f..0fd602e4b 100644 --- a/docs/design/durable-operations.md +++ b/docs/design/durable-operations.md @@ -316,39 +316,49 @@ The durable split is deliberate: - `backend_jobs` stores immutable bounded input, its fingerprint, the current attempt number, and the final structured result or terminal failure; - `backend_job_attempts` binds one attempt and input revision to one dedicated - `(Workspace, Runtime, Worker)` identity, optional Runtime run evidence, a - deadline, and its bounded outcome; + `(Workspace, Runtime, Worker)` identity, a precommitted tracked Runtime + submission request identity, a deadline, and its bounded outcome; - `backend_job_deliveries` records best-effort notification separately from the - result, so a missing or stopped recipient never discards a result or reruns - model work. + result. Its durable delivery ID is also the Runtime's tracked notification + request ID, so restart recovery can replay an ambiguous `sending` claim + without duplicating Worker context. A missing or stopped recipient never + discards a result or reruns model work. Exact dispatch retry reuses the deterministic attempt and Worker allocation key. -Only explicit re-evaluation increments the bounded attempt number. The runner +Only explicit re-evaluation of a definitively failed attempt increments the +bounded attempt number; unknown outcomes are not replayable. The runner uses the existing embedded Runtime spawn, Worker registry, Session input, observation, usage, and lifecycle paths with no Workdir attachment. It uses the dedicated `builtin:backend-job` profile, whose tools cannot mutate Workspace state, recursively create Workers or Jobs, or inherit caller Workdir authority. -Success is accepted only by the structured result boundary. The submission is -fenced by Workspace, Job ID, attempt ID, input revision, and the bound Runtime -Worker identity, and its JSON bytes are bounded before persistence. Worker final -prose, Idle, Stopped, or a successful model call are not success evidence. -Identical structured submission is idempotent; conflicting replay, stale or late -attempts, wrong revisions, and other Workers fail closed. +Success is accepted only by the structured result boundary. Before Runtime +input can execute, the attempt is fenced by Workspace, Job ID, attempt ID, input +revision, bound Runtime Worker identity, and the deterministic tracked submission +request identity. Ordinary Worker input and Console mutation are rejected for +Job-owned Workers, so no later run in the same Session can exercise the result +capability. Result JSON bytes are bounded before persistence. Worker final prose, +Idle, Stopped, or a successful model call are not success evidence. Identical +structured submission is idempotent; conflicting replay, stale or late attempts, +wrong revisions, and other Workers fail closed. On restart, an unbound reservation is redispatched with its stable allocation. -A dispatched attempt continues only when its bound Worker remains observable; -otherwise it becomes `unknown`. Durable deadlines fence acceptance before the -runner requests Worker stop. The design does not claim general exactly-once -model execution: a crash between Runtime creation and durable binding converges -through stable allocation where available, and unrecoverable external evidence -is surfaced as `unknown` rather than guessed. Historical attempts preserve the -reason for retry without duplicating the Runtime Worker lifecycle state machine. +A dispatched attempt continues only when its bound Worker and precommitted +tracked submission identity remain observable; otherwise it becomes `unknown`. +Ambiguous dispatch outcomes are also `unknown` and cannot be retried as a new +attempt. Durable deadlines fence acceptance before the runner requests Worker +stop. Notification recovery is independent: `sending` or `unknown` delivery +claims return to `pending` and replay the same tracked notification request ID. +The design does not claim general exactly-once model execution: a crash between +Runtime creation and durable binding converges through stable allocation where +available, and unrecoverable external evidence is surfaced as `unknown` rather +than guessed. Historical attempts preserve the reason for retry without +duplicating the Runtime Worker lifecycle state machine. Workspace Worker subscriptions expose an explicit Job binding (`job_id`, `attempt_id`, and `purpose`) in snapshot and live projections. Sidebar hiding is a Web View policy over that canonical metadata; the Worker remains registered, -observable, and eligible for its canonical Console route. +observable, and eligible for its canonical read-only Console route. ## Diagnostics and audit