From 213ed760618cf1337f1350b710a3e87a137dc6c6 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 1 Oct 2026 11:18:14 +0900 Subject: [PATCH] fix backend job execution fencing and delivery --- crates/worker/src/controller.rs | 67 +-- crates/worker/tests/controller_test.rs | 33 ++ crates/workspace-server/src/backend_job.rs | 6 + crates/workspace-server/src/hosts.rs | 15 +- crates/workspace-server/src/latest_schema.sql | 2 +- crates/workspace-server/src/server.rs | 264 +++++++++-- crates/workspace-server/src/store.rs | 425 ++++++++++++++++-- 7 files changed, 703 insertions(+), 109 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index de5d836f5..42c4efc25 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1178,39 +1178,42 @@ where ), None => durable_parent_notifications, }; + let backend_job_profile = worker.manifest().engine.instruction == "internal.backend_job_system"; let prompts = worker.prompts().clone(); - let paste_store = worker.store().clone(); - let paste_session_id = worker.session_id(); - worker - .engine_mut() - .register_tool(crate::paste_artifact_tool::search_input_artifact_tool( - paste_store.clone(), - paste_session_id, - )); - worker - .engine_mut() - .register_tool(crate::paste_artifact_tool::read_input_artifact_tool( - paste_store, - paste_session_id, - )); - // Keep the alias-routed schemas installed for the Worker lifetime. The - // attachment set may be empty or mutate after registration, so target - // resolution and provider capability checks happen per invocation. let tracker = tools::Tracker::new(); - let workdir_tools = if restrict_workdir_tools_to_current_capabilities { - tools::routed_builtin_tools_for_current_capabilities( - workdir_sessions.clone(), - tracker.clone(), - bash_output_dir.clone(), - ) - } else { - tools::routed_builtin_tools( - workdir_sessions.clone(), - tracker.clone(), - bash_output_dir.clone(), - ) - }; - worker.engine_mut().register_tools(workdir_tools); + if !backend_job_profile { + let paste_store = worker.store().clone(); + let paste_session_id = worker.session_id(); + worker + .engine_mut() + .register_tool(crate::paste_artifact_tool::search_input_artifact_tool( + paste_store.clone(), + paste_session_id, + )); + worker + .engine_mut() + .register_tool(crate::paste_artifact_tool::read_input_artifact_tool( + paste_store, + paste_session_id, + )); + // Keep the alias-routed schemas installed for the Worker lifetime. The + // attachment set may be empty or mutate after registration, so target + // resolution and provider capability checks happen per invocation. + let workdir_tools = if restrict_workdir_tools_to_current_capabilities { + tools::routed_builtin_tools_for_current_capabilities( + workdir_sessions.clone(), + tracker.clone(), + bash_output_dir.clone(), + ) + } else { + tools::routed_builtin_tools( + workdir_sessions.clone(), + tracker.clone(), + bash_output_dir.clone(), + ) + }; + worker.engine_mut().register_tools(workdir_tools); + } if feature_config.image.enabled && model_supports_image_attachments(&spawner_manifest.model) { worker .engine_mut() @@ -1225,7 +1228,7 @@ where let worker_enabled = feature_config.worker.enabled; let sub_worker_enabled = feature_config.sub_worker.enabled; let mut feature_registry = FeatureRegistryBuilder::new(); - if worker.manifest().engine.instruction == "internal.backend_job_system" { + if backend_job_profile { let workspace_client = worker.workspace_client_handle(); if !workspace_client.is_available() || workspace_client.workspace_id().is_none() { return Err(std::io::Error::new( diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 1a3d5282e..69b30d11a 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -837,6 +837,39 @@ async fn feature_flags_default_to_core_tool_surface_only() { assert!(!names.iter().any(|name| name == "SubWorkerSpawn")); } +#[tokio::test] +async fn backend_job_profile_exposes_only_structured_result_capability() { + let workspace = tempfile::tempdir().unwrap(); + let resolved = ProfileResolver::new() + .with_workspace_base(workspace.path()) + .resolve( + &ProfileSelector::source_named(ProfileRegistrySource::Builtin, "backend-job"), + ProfileResolveOptions::with_worker_name("backend-job-worker"), + ) + .unwrap(); + let client = MockClient::new(simple_text_events()); + let client_for_assert = client.clone(); + let (worker, _pwd) = make_worker_with_manifest_and_workspace_context( + client, + resolved.manifest, + WorkerWorkspaceContext::with_client(None, Arc::new(AvailableWorkspaceClient)), + ) + .await; + let handle = spawn_controller(worker).await; + + handle + .send(Method::submit_text( + protocol::new_submission_request_id(), + "Run the bounded Backend Job.", + )) + .await + .unwrap(); + wait_for_status(&handle, WorkerStatus::Idle).await; + + let request = wait_for_captured_request(&client_for_assert).await; + assert_eq!(request_tool_names(&request), vec!["SubmitBackendJobResult"]); +} + #[tokio::test] async fn enabled_task_and_web_features_register_their_tools() { let manifest = r#" diff --git a/crates/workspace-server/src/backend_job.rs b/crates/workspace-server/src/backend_job.rs index 0788589e6..996f01780 100644 --- a/crates/workspace-server/src/backend_job.rs +++ b/crates/workspace-server/src/backend_job.rs @@ -101,24 +101,30 @@ impl BackendJobAttemptState { #[serde(rename_all = "snake_case")] pub enum BackendJobDeliveryState { Pending, + Sending, Completed, Failed, + Unknown, } impl BackendJobDeliveryState { pub(crate) fn as_str(self) -> &'static str { match self { Self::Pending => "pending", + Self::Sending => "sending", Self::Completed => "completed", Self::Failed => "failed", + Self::Unknown => "unknown", } } pub(crate) fn parse(value: &str) -> Result { match value { "pending" => Ok(Self::Pending), + "sending" => Ok(Self::Sending), "completed" => Ok(Self::Completed), "failed" => Ok(Self::Failed), + "unknown" => Ok(Self::Unknown), other => Err(Error::Store(format!( "unknown Backend Job delivery state `{other}`" ))), diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 91e640121..394d3316b 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -735,6 +735,8 @@ pub struct WorkerInputResult { pub state: InternalWorkerOperationState, #[serde(flatten)] pub worker: RuntimeWorkerRef, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub runtime_run_id: Option, pub diagnostics: Vec, } @@ -1206,6 +1208,7 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { WorkerInputResult { state: InternalWorkerOperationState::Unsupported, worker: RuntimeWorkerRef::new(self.runtime_id().to_string(), worker_id.to_string()), + runtime_run_id: None, diagnostics: vec![diagnostic( "worker_input_pending", HostDiagnosticSeverity::Info, @@ -3023,9 +3026,10 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { segments: request.segments, }; match self.runtime.send_input(&worker_ref, input) { - Ok(_) => WorkerInputResult { + Ok(ack) => WorkerInputResult { state: InternalWorkerOperationState::Accepted, worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), + runtime_run_id: ack.submission.map(|submission| submission.submission_id), diagnostics: Vec::new(), }, Err(error) => embedded_input_rejected( @@ -5005,6 +5009,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { return WorkerInputResult { state: InternalWorkerOperationState::Rejected, worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), + runtime_run_id: None, diagnostics: vec![diagnostic], }; } @@ -5015,9 +5020,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { MAX_REMOTE_RUNTIME_RESPONSE_BYTES, move |client| async move { client.send_worker_input(worker_id_owned, input).await }, ) { - Ok(_) => WorkerInputResult { + Ok(response) => WorkerInputResult { state: InternalWorkerOperationState::Accepted, worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), + runtime_run_id: response + .ack + .submission + .map(|submission| submission.submission_id), diagnostics: Vec::new(), }, Err(diagnostic) => remote_input_rejected(&self.runtime_id, worker_id, diagnostic), @@ -5406,6 +5415,7 @@ fn embedded_input_rejected( WorkerInputResult { state: InternalWorkerOperationState::Rejected, worker: RuntimeWorkerRef::new(runtime_id.to_string(), worker_id.to_string()), + runtime_run_id: None, diagnostics: vec![diagnostic], } } @@ -5418,6 +5428,7 @@ fn remote_input_rejected( WorkerInputResult { state: InternalWorkerOperationState::Rejected, worker: RuntimeWorkerRef::new(runtime_id.to_string(), worker_id.to_string()), + runtime_run_id: None, diagnostics: vec![diagnostic], } } diff --git a/crates/workspace-server/src/latest_schema.sql b/crates/workspace-server/src/latest_schema.sql index ecd814e24..11b68f143 100644 --- a/crates/workspace-server/src/latest_schema.sql +++ b/crates/workspace-server/src/latest_schema.sql @@ -867,7 +867,7 @@ CREATE TABLE backend_job_deliveries ( attempt_id TEXT NOT NULL, target_runtime_id TEXT NOT NULL, target_worker_id TEXT NOT NULL, - state TEXT NOT NULL CHECK (state IN ('pending', 'completed', 'failed')), + state TEXT NOT NULL CHECK (state IN ('pending', 'sending', 'completed', 'failed', 'unknown')), failure_category TEXT, failure_detail TEXT, created_at TEXT NOT NULL, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index dbeb59ad0..2f408ead5 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -3244,6 +3244,9 @@ impl WorkspaceApi { } recover_workdir_removals(&api)?; recover_runtime_removals(&api).await; + let recovery_now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); + api.store + .recover_backend_job_deliveries(&api.config.workspace_id, &recovery_now)?; api.recover_backend_jobs()?; Ok(api) } @@ -3292,34 +3295,47 @@ impl WorkspaceApi { operation_id: allocation_key(&reservation.attempt.attempt_id), input_fingerprint: reservation.job.intent_fingerprint.clone(), }; - let spawned = self - .spawn_workspace_worker( - EMBEDDED_WORKER_RUNTIME_ID, - WorkerSpawnRequest { - requested_worker_name: Some(format!("job:{}", request.purpose)), - intent: WorkerSpawnIntent::BackendJob { - job_id: request.job_id.clone(), - attempt_id: reservation.attempt.attempt_id.clone(), - purpose: request.purpose.clone(), - }, - acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { - expected_segments: 0, - }, - profile: ProfileSelector::Builtin(request.profile.clone()), - 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: None, - resolved_worker_observation_enabled: false, - resolved_worker_observation_grants: Vec::new(), - resolved_workspace_api: None, - resolved_memory_settings: None, - resolved_control_operation: Some(control_operation), + let spawned = match self.spawn_workspace_worker( + EMBEDDED_WORKER_RUNTIME_ID, + WorkerSpawnRequest { + requested_worker_name: Some(format!("job:{}", request.purpose)), + intent: WorkerSpawnIntent::BackendJob { + job_id: request.job_id.clone(), + attempt_id: reservation.attempt.attempt_id.clone(), + purpose: request.purpose.clone(), }, - ) - .map_err(|error| error.error)?; + acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: 0, + }, + profile: ProfileSelector::Builtin(request.profile.clone()), + 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: None, + resolved_worker_observation_enabled: false, + resolved_worker_observation_grants: Vec::new(), + resolved_workspace_api: None, + resolved_memory_settings: None, + resolved_control_operation: Some(control_operation), + }, + ) { + Ok(spawned) => spawned, + Err(error) => { + let detail = error.error.to_string(); + self.store.finish_backend_job_attempt( + &self.config.workspace_id, + &request.job_id, + &reservation.attempt.attempt_id, + BackendJobAttemptState::Failed, + "dispatch_error", + &detail, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + return Err(error.error); + } + }; let worker = match (spawned.state, spawned.worker) { (InternalWorkerOperationState::Accepted, Some(worker)) => worker, _ => { @@ -3375,17 +3391,35 @@ impl WorkspaceApi { }); } - let submitted = self - .runtime - .send_input( - &worker.worker, - WorkerInputRequest { - kind: WorkerInputKind::User, - content: worker_input, - segments: None, - }, - ) - .map_err(RuntimeRegistryError::into_error)?; + self.schedule_backend_job_timeout( + &request.job_id, + &reservation.attempt.attempt_id, + &reservation.attempt.deadline_at, + ); + + let submitted = match self.runtime.send_input( + &worker.worker, + WorkerInputRequest { + kind: WorkerInputKind::User, + content: worker_input, + segments: None, + }, + ) { + Ok(submitted) => submitted, + Err(error) => { + let detail = error.message(); + self.store.finish_backend_job_attempt( + &self.config.workspace_id, + &request.job_id, + &reservation.attempt.attempt_id, + BackendJobAttemptState::Unknown, + "input_outcome_unknown", + &detail, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + return Err(error.into_error()); + } + }; if submitted.state != InternalWorkerOperationState::Accepted { let detail = submitted .diagnostics @@ -3407,11 +3441,43 @@ impl WorkspaceApi { message: detail.to_string(), }); } - self.schedule_backend_job_timeout( + let Some(runtime_run_id) = submitted.runtime_run_id.as_deref() else { + 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", + &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" + .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, - &reservation.attempt.deadline_at, - ); + &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 .get_backend_job(&self.config.workspace_id, &request.job_id)? @@ -3462,6 +3528,18 @@ impl WorkspaceApi { } } BackendJobAttemptState::Dispatched => { + if attempt.runtime_run_id.is_none() { + self.store.finish_backend_job_attempt( + &self.config.workspace_id, + &attempt.job_id, + &attempt.attempt_id, + BackendJobAttemptState::Unknown, + "recovery_runtime_run_unknown", + "the Runtime accepted Worker dispatch without durable run evidence", + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + continue; + } let worker_known = attempt .worker .as_ref() @@ -3575,7 +3653,7 @@ impl WorkspaceApi { acceptance.job.request.job_id, acceptance.attempt.attempt_id ); let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); - let Ok(delivery) = self.store.reserve_backend_job_delivery( + let Ok((delivery, claimed)) = self.store.reserve_backend_job_delivery( &self.config.workspace_id, &delivery_id, &acceptance.job.request.job_id, @@ -3584,7 +3662,7 @@ impl WorkspaceApi { ) else { return; }; - if delivery.state == BackendJobDeliveryState::Completed { + if !claimed || delivery.state != BackendJobDeliveryState::Sending { return; } let content = format!( @@ -23747,6 +23825,17 @@ async fn send_runtime_worker_input( ) -> ApiResult> { let request = worker_input_request_from_api(request)?; let worker = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; + if api + .store + .worker_registry_projection(&api.config.workspace_id, &worker)? + .and_then(|projection| projection.job) + .is_some() + { + return Err(Error::InvalidInput( + "Backend Job Workers accept only their Backend-owned immutable input".to_string(), + ) + .into()); + } let result = api .runtime .send_input(&worker, request) @@ -23858,6 +23947,13 @@ async fn worker_protocol_ws( ws: WebSocketUpgrade, ) -> impl IntoResponse { let worker = RuntimeWorkerRef::new(&runtime_id, &worker_id); + let job_read_only = match api + .store + .worker_registry_projection(&api.config.workspace_id, &worker) + { + Ok(projection) => projection.and_then(|projection| projection.job).is_some(), + Err(error) => return ApiError::from(error).into_response(), + }; let source = match api.observation_proxy.source(&worker) { Ok(source) => source, Err(ObservationProxyError::WorkerNotFound(_)) => { @@ -23878,7 +23974,9 @@ async fn worker_protocol_ws( } }; let input_source = authenticated_browser_input_source(&actor); - ws.on_upgrade(move |socket| worker_protocol_ws_session(source, socket, input_source)) + ws.on_upgrade(move |socket| { + worker_protocol_ws_session(source, socket, input_source, job_read_only) + }) } pub(crate) struct WorkspaceWorkerProtocolConnection { @@ -24035,13 +24133,14 @@ async fn worker_protocol_ws_session( source: RuntimeObservationSource, socket: WebSocket, input_source: protocol::AuthenticatedInputSource, + read_only: bool, ) { match source { RuntimeObservationSource::RemoteWs(config) => { - remote_worker_protocol_ws_session(config, socket, input_source).await; + remote_worker_protocol_ws_session(config, socket, input_source, read_only).await; } RuntimeObservationSource::Embedded(source) => { - embedded_worker_protocol_ws_session(source, socket, input_source).await; + embedded_worker_protocol_ws_session(source, socket, input_source, read_only).await; } } } @@ -24050,6 +24149,7 @@ async fn remote_worker_protocol_ws_session( config: RuntimeObservationSourceConfig, socket: WebSocket, input_source: protocol::AuthenticatedInputSource, + read_only: bool, ) { let mut request = match config.endpoint.clone().into_client_request() { Ok(request) => request, @@ -24108,6 +24208,14 @@ async fn remote_worker_protocol_ws_session( inbound = client_stream.next() => { match inbound { Some(Ok(WsMessage::Text(text))) => { + if read_only { + if let Ok(event) = protocol::stream::encode_event(&protocol_error_event( + "Backend Job Worker Console is read-only", + )) { + let _ = client_sink.send(WsMessage::Text(event.into())).await; + } + continue; + } let method = match protocol::stream::decode_method(text.as_ref()) { Ok(method) => match authorize_browser_worker_method(method, &input_source) { Ok(method) => method, @@ -24191,6 +24299,7 @@ async fn embedded_worker_protocol_ws_session( source: crate::observation::EmbeddedRuntimeObservationSource, mut socket: WebSocket, input_source: protocol::AuthenticatedInputSource, + read_only: bool, ) { let mut upstream = match RuntimeObservationClient::connect(&RuntimeObservationSource::Embedded( source.clone(), @@ -24209,6 +24318,12 @@ async fn embedded_worker_protocol_ws_session( tokio::select! { inbound = socket.next() => { match inbound { + Some(Ok(WsMessage::Text(_))) if read_only => { + let event = protocol_error_event("Backend Job Worker Console is read-only"); + if !send_protocol_event(&mut socket, &event).await { + return; + } + } Some(Ok(WsMessage::Text(text))) => match decode_method(&text) { Ok(method) => match authorize_browser_worker_method(method, &input_source) { Ok(method) => match source.runtime.send_protocol_method(&source.worker_ref, method) { @@ -33877,6 +33992,10 @@ mod tests { "a stale concurrent dispatch replay must not submit the attempt twice" ); assert_eq!(dispatched.attempt.state, BackendJobAttemptState::Dispatched); + assert!( + dispatched.attempt.runtime_run_id.is_some(), + "accepted Job input must be fenced to durable Runtime run evidence" + ); let worker = dispatched.attempt.worker.clone().expect("bound Job Worker"); assert_eq!(worker.runtime_id, EMBEDDED_WORKER_RUNTIME_ID); assert!( @@ -33898,6 +34017,25 @@ mod tests { purpose: request.purpose.clone(), }) ); + let input_error = send_runtime_worker_input( + State(api.clone()), + AxumPath((worker.runtime_id.clone(), worker.worker_id.clone())), + Json(server_api::RuntimeWorkerInputRequest { + kind: Some("user".to_string()), + content: "attempt to rerun immutable Job input".to_string(), + segments: None, + }), + ) + .await + .unwrap_err(); + assert_eq!( + input_error.into_response().status(), + StatusCode::BAD_REQUEST + ); + assert!( + execution.take_inputs().is_empty(), + "Job-owned Workers must reject interactive reruns" + ); let replay = api.dispatch_backend_job(&request).unwrap(); assert!(replay.replayed); assert_eq!(replay.attempt.worker, Some(worker.clone())); @@ -33916,6 +34054,40 @@ mod tests { assert_eq!(accepted.job.state, BackendJobState::Completed); } + #[tokio::test] + async fn backend_job_spawn_transport_error_terminalizes_reserved_attempt() { + let temp = tempfile::tempdir().unwrap(); + let api = test_api(temp.path()).await; + api.runtime + .unregister_if_idle(EMBEDDED_WORKER_RUNTIME_ID, 10) + .unwrap(); + let request = BackendJobRequest { + job_id: "runner-spawn-error-1".to_string(), + purpose: "spawn_error_contract_check".to_string(), + input_revision: "revision-1".to_string(), + input_ref: "test://runner/spawn-error/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(), + }; + + assert!(api.dispatch_backend_job(&request).is_err()); + let attempt = api + .store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &request.job_id, + &crate::backend_job::attempt_id(&request.job_id, 1), + ) + .unwrap() + .unwrap(); + assert_eq!(attempt.state, BackendJobAttemptState::Failed); + assert_eq!(attempt.failure_category.as_deref(), Some("dispatch_error")); + } + #[tokio::test] async fn backend_job_input_failure_is_durable_and_retryable_without_reusing_attempt() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index f233b0c2c..e021a02c1 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -27,7 +27,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore; use crate::{Error, Result}; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 70; +const LATEST_SCHEMA_VERSION: i64 = 71; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings"; const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit"; @@ -62,6 +62,8 @@ const WORKDIR_COMMAND_ATTACHMENT_CAPABILITIES_MIGRATION_NAME: &str = "External Workdir command attachment capabilities"; const BACKEND_JOB_RUNNER_MIGRATION_NAME: &str = "Backend-owned Runtime Worker jobs and structured results"; +const BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME: &str = + "atomic Backend Job notification delivery claims"; const TICKET_SCHEMA_VERSION_WITH_TARGETS: i64 = 7; const TICKET_SCHEMA_BASELINE_NAME: &str = "ticket schema baseline"; const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#" @@ -209,6 +211,11 @@ const MIGRATIONS: &[Migration] = &[ name: BACKEND_JOB_RUNNER_MIGRATION_NAME, apply: migrate_backend_job_runner_v69_to_v70, }, + Migration { + version: 71, + name: BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME, + apply: migrate_backend_job_delivery_claims_v70_to_v71, + }, ]; #[derive(Clone, Copy)] @@ -1491,6 +1498,15 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { runtime_run_id: Option<&str>, now: &str, ) -> Result<(WorkerRegistryProjectionCommit, bool)>; + fn bind_backend_job_attempt_runtime_run( + &self, + workspace_id: &str, + job_id: &str, + attempt_id: &str, + worker: &RuntimeWorkerRef, + runtime_run_id: &str, + now: &str, + ) -> Result; fn accept_backend_job_result( &self, workspace_id: &str, @@ -1515,7 +1531,8 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { job_id: &str, target: &RuntimeWorkerRef, now: &str, - ) -> Result; + ) -> Result<(BackendJobDeliveryRecord, bool)>; + fn recover_backend_job_deliveries(&self, workspace_id: &str, now: &str) -> Result; fn finish_backend_job_delivery( &self, workspace_id: &str, @@ -5989,7 +6006,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore { )); } if let Some(bound) = &attempt.worker { - if bound != worker || attempt.runtime_run_id.as_deref() != runtime_run_id { + let conflicting_run = runtime_run_id + .is_some_and(|run_id| attempt.runtime_run_id.as_deref() != Some(run_id)); + if bound != worker || conflicting_run { return Err(Error::InvalidInput( "Backend Job attempt is already bound to different Runtime evidence".to_string(), )); @@ -6023,6 +6042,64 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn bind_backend_job_attempt_runtime_run( + &self, + workspace_id: &str, + job_id: &str, + attempt_id: &str, + worker: &RuntimeWorkerRef, + runtime_run_id: &str, + now: &str, + ) -> Result { + if runtime_run_id.is_empty() + || runtime_run_id.len() > 512 + || runtime_run_id.chars().any(char::is_control) + { + return Err(Error::InvalidInput( + "Backend Job Runtime run identity is invalid".to_string(), + )); + } + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)? + .ok_or_else(|| { + Error::InvalidInput(format!("unknown Backend Job attempt `{attempt_id}`")) + })?; + if attempt.state != BackendJobAttemptState::Dispatched + || attempt.worker.as_ref() != Some(worker) + { + return Err(Error::InvalidInput( + "Backend Job Runtime run does not match the dispatched attempt".to_string(), + )); + } + if let Some(existing) = attempt.runtime_run_id.as_deref() { + if existing != runtime_run_id { + return Err(Error::InvalidInput( + "Backend Job attempt is already bound to a different Runtime run" + .to_string(), + )); + } + tx.commit()?; + return Ok(attempt); + } + let changed = tx.execute( + "UPDATE backend_job_attempts SET runtime_run_id = ?4, updated_at = ?5 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state = 'dispatched' AND runtime_id = ?6 AND worker_id = ?7 AND runtime_run_id IS NULL", + params![workspace_id, job_id, attempt_id, runtime_run_id, now, worker.runtime_id, worker.worker_id], + )?; + if changed != 1 { + return Err(Error::Store( + "Backend Job Runtime run binding compare-and-set failed".to_string(), + )); + } + let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)? + .ok_or_else(|| { + Error::Store("bound Backend Job Runtime run disappeared".to_string()) + })?; + tx.commit()?; + Ok(attempt) + }) + } + fn accept_backend_job_result( &self, workspace_id: &str, @@ -6050,6 +6127,11 @@ impl ControlPlaneStore for SqliteWorkspaceStore { "Backend Job result source is not the attempt Worker".to_string(), )); } + if attempt.runtime_run_id.is_none() { + return Err(Error::InvalidInput( + "Backend Job result source has no fenced Runtime run".to_string(), + )); + } if attempt.state == BackendJobAttemptState::Completed { if attempt.result_digest.as_deref() != Some(digest.as_str()) { return Err(Error::InvalidInput( @@ -6152,7 +6234,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { job_id: &str, target: &RuntimeWorkerRef, now: &str, - ) -> Result { + ) -> Result<(BackendJobDeliveryRecord, bool)> { self.with_conn_mut(|conn| { let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; if let Some(existing) = read_backend_job_delivery(&tx, workspace_id, delivery_id)? { @@ -6161,8 +6243,23 @@ impl ControlPlaneStore for SqliteWorkspaceStore { "Backend Job delivery id was reused with different intent".to_string(), )); } + if existing.state != BackendJobDeliveryState::Pending { + tx.commit()?; + return Ok((existing, false)); + } + let claimed = tx.execute( + "UPDATE backend_job_deliveries SET state = 'sending', updated_at = ?3 WHERE workspace_id = ?1 AND delivery_id = ?2 AND state = 'pending'", + params![workspace_id, delivery_id, now], + )?; + if claimed != 1 { + return Err(Error::Store( + "Backend Job delivery claim lost inside immediate transaction".to_string(), + )); + } + let delivery = read_backend_job_delivery(&tx, workspace_id, delivery_id)? + .ok_or_else(|| Error::Store("claimed Backend Job delivery disappeared".to_string()))?; tx.commit()?; - return Ok(existing); + return Ok((delivery, true)); } let job = read_backend_job(&tx, workspace_id, job_id)? .ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job `{job_id}`")))?; @@ -6178,13 +6275,27 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } let current_attempt_id = attempt_id(job_id, job.current_attempt); tx.execute( - "INSERT INTO backend_job_deliveries (workspace_id, delivery_id, job_id, attempt_id, target_runtime_id, target_worker_id, state, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'pending', ?7, ?7)", + "INSERT INTO backend_job_deliveries (workspace_id, delivery_id, job_id, attempt_id, target_runtime_id, target_worker_id, state, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'sending', ?7, ?7)", params![workspace_id, delivery_id, job_id, current_attempt_id, target.runtime_id, target.worker_id, now], )?; let delivery = read_backend_job_delivery(&tx, workspace_id, delivery_id)? .ok_or_else(|| Error::Store("reserved Backend Job delivery disappeared".to_string()))?; tx.commit()?; - Ok(delivery) + Ok((delivery, true)) + }) + } + + fn recover_backend_job_deliveries(&self, workspace_id: &str, now: &str) -> Result { + self.with_conn_mut(|conn| { + conn.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', + updated_at = ?2 + WHERE workspace_id = ?1 AND state = 'sending'", + params![workspace_id, now], + ) + .map_err(Into::into) }) } @@ -6197,9 +6308,12 @@ impl ControlPlaneStore for SqliteWorkspaceStore { failure_detail: Option<&str>, now: &str, ) -> Result { - if state == BackendJobDeliveryState::Pending { + if matches!( + state, + BackendJobDeliveryState::Pending | BackendJobDeliveryState::Sending + ) { return Err(Error::InvalidInput( - "Backend Job delivery completion cannot remain pending".to_string(), + "Backend Job delivery completion must be terminal".to_string(), )); } let detail = failure_detail.map(bounded_failure_detail); @@ -6207,14 +6321,16 @@ impl ControlPlaneStore for SqliteWorkspaceStore { let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; let current = read_backend_job_delivery(&tx, workspace_id, delivery_id)? .ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job delivery `{delivery_id}`")))?; - if current.state == BackendJobDeliveryState::Completed { - if state != BackendJobDeliveryState::Completed { - return Err(Error::InvalidInput( - "completed Backend Job delivery cannot be replaced".to_string(), - )); + if current.state != BackendJobDeliveryState::Sending { + if current.state == state { + tx.commit()?; + return Ok(current); } - tx.commit()?; - return Ok(current); + return Err(Error::InvalidInput(format!( + "Backend Job delivery in state `{}` cannot be finished as `{}`", + current.state.as_str(), + state.as_str() + ))); } tx.execute( "UPDATE backend_job_deliveries SET state = ?3, failure_category = ?4, failure_detail = ?5, delivered_at = CASE WHEN ?3 = 'completed' THEN ?6 ELSE delivered_at END, updated_at = ?6 WHERE workspace_id = ?1 AND delivery_id = ?2", @@ -13103,6 +13219,57 @@ fn migrate_backend_job_runner_v69_to_v70(conn: &Connection) -> Result<()> { Ok(()) } +fn migrate_backend_job_delivery_claims_v70_to_v71(conn: &Connection) -> Result<()> { + let current = current_schema_version(conn)?; + if current != 70 { + return Err(Error::Store(format!( + "expected schema version 70 before {BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME} migration, found {current}" + ))); + } + let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; + tx.execute_batch( + r#" + DROP INDEX backend_job_deliveries_pending; + ALTER TABLE backend_job_deliveries RENAME TO backend_job_deliveries_v70; + CREATE TABLE backend_job_deliveries ( + workspace_id TEXT NOT NULL, + delivery_id TEXT NOT NULL, + job_id TEXT NOT NULL, + attempt_id TEXT NOT NULL, + target_runtime_id TEXT NOT NULL, + target_worker_id TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('pending', 'sending', 'completed', 'failed', 'unknown')), + failure_category TEXT, + failure_detail TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + delivered_at TEXT, + PRIMARY KEY (workspace_id, delivery_id), + FOREIGN KEY (workspace_id, job_id, attempt_id) + REFERENCES backend_job_attempts(workspace_id, job_id, attempt_id) ON DELETE CASCADE + ); + INSERT INTO backend_job_deliveries ( + workspace_id, delivery_id, job_id, attempt_id, target_runtime_id, + target_worker_id, state, failure_category, failure_detail, + created_at, updated_at, delivered_at + ) + SELECT workspace_id, delivery_id, job_id, attempt_id, target_runtime_id, + target_worker_id, state, failure_category, failure_detail, + created_at, updated_at, delivered_at + FROM backend_job_deliveries_v70; + DROP TABLE backend_job_deliveries_v70; + CREATE INDEX backend_job_deliveries_pending + ON backend_job_deliveries(workspace_id, state, updated_at); + "#, + )?; + tx.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", + params![71_i64, BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME], + )?; + tx.commit()?; + Ok(()) +} + fn create_latest_workspace_schema(conn: &Connection) -> Result<()> { conn.execute_batch("DROP TABLE IF EXISTS typed_ticket_targets;")?; conn.execute_batch(include_str!("latest_schema.sql"))?; @@ -15278,6 +15445,10 @@ mod tests { version: 70, name: BACKEND_JOB_RUNNER_MIGRATION_NAME.to_string(), }, + WorkspaceSchemaMigrationStep { + version: 71, + name: BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME.to_string(), + }, ] ); @@ -15349,6 +15520,10 @@ mod tests { WORKDIR_COMMAND_ATTACHMENT_CAPABILITIES_MIGRATION_NAME.to_string(), ), (70, BACKEND_JOB_RUNNER_MIGRATION_NAME.to_string()), + ( + 71, + BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME.to_string(), + ), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -15547,6 +15722,75 @@ mod tests { ); } + #[test] + fn schema_v71_adds_delivery_claim_states_without_losing_existing_intent() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("server.db"); + prepare_schema_v50(&path, Some("workspace-a")); + let conn = Connection::open(&path).unwrap(); + configure_sqlite(&conn).unwrap(); + for migration in MIGRATIONS + .iter() + .filter(|migration| migration.version <= 70) + { + (migration.apply)(&conn).unwrap(); + } + conn.execute_batch( + r#" + INSERT INTO backend_jobs ( + workspace_id, job_id, purpose, input_revision, input_ref, request_json, + intent_fingerprint, state, current_attempt, created_at, updated_at + ) VALUES ( + 'workspace-a', 'job-a', 'check', 'revision-a', 'test://job-a', '{}', + 'sha256:test', 'pending', 1, '1', '1' + ); + INSERT INTO backend_job_attempts ( + workspace_id, job_id, attempt_id, attempt, input_revision, state, + deadline_at, created_at, updated_at + ) VALUES ( + 'workspace-a', 'job-a', 'job-a:attempt:1', 1, 'revision-a', 'reserved', + '2', '1', '1' + ); + INSERT INTO backend_job_deliveries ( + workspace_id, delivery_id, job_id, attempt_id, target_runtime_id, + target_worker_id, state, created_at, updated_at + ) VALUES ( + 'workspace-a', 'delivery-a', 'job-a', 'job-a:attempt:1', 'runtime-a', + 'worker-a', 'pending', '1', '1' + ); + "#, + ) + .unwrap(); + + migrate_backend_job_delivery_claims_v70_to_v71(&conn).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 71); + assert_eq!( + conn.query_row( + "SELECT state FROM backend_job_deliveries WHERE delivery_id = 'delivery-a'", + [], + |row| row.get::<_, String>(0), + ) + .unwrap(), + "pending" + ); + assert_eq!( + conn.execute( + "UPDATE backend_job_deliveries SET state = 'sending' WHERE delivery_id = 'delivery-a'", + [], + ) + .unwrap(), + 1 + ); + assert_eq!( + conn.execute( + "UPDATE backend_job_deliveries SET state = 'unknown' WHERE delivery_id = 'delivery-a'", + [], + ) + .unwrap(), + 1 + ); + } + #[test] fn migration_resumes_from_a_valid_partially_applied_chain() { let temp = tempfile::tempdir().unwrap(); @@ -15565,7 +15809,7 @@ mod tests { .map(|migration| migration.version) .collect::>(), vec![ - 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70 + 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71 ] ); SqliteWorkspaceStore::migrate_database(&path).unwrap(); @@ -15574,7 +15818,7 @@ mod tests { current_schema_version(&conn).unwrap(), LATEST_SCHEMA_VERSION ); - assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 21); + assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 22); assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap()); assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap()); assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap()); @@ -19263,10 +19507,49 @@ INSERT INTO worker_registry ( &request.job_id, &first.attempt.attempt_id, &worker, - Some("runtime-run-1"), + None, "2026-09-01T00:00:02Z", ) .unwrap(); + let submission = BackendJobResultSubmission { + job_id: request.job_id.clone(), + attempt_id: first.attempt.attempt_id.clone(), + input_revision: request.input_revision.clone(), + result: serde_json::json!({"valid": true}), + }; + assert!( + store + .accept_backend_job_result( + workspace_id, + &worker, + &submission, + "2026-09-01T00:00:02Z", + ) + .is_err(), + "Worker identity without Runtime run evidence must not accept a result" + ); + store + .bind_backend_job_attempt_runtime_run( + workspace_id, + &request.job_id, + &first.attempt.attempt_id, + &worker, + "runtime-run-1", + "2026-09-01T00:00:02Z", + ) + .unwrap(); + assert!( + store + .bind_backend_job_attempt_runtime_run( + workspace_id, + &request.job_id, + &first.attempt.attempt_id, + &worker, + "runtime-run-2", + "2026-09-01T00:00:02Z", + ) + .is_err() + ); assert_eq!( store .worker_registry_projection(workspace_id, &worker) @@ -19279,12 +19562,6 @@ INSERT INTO worker_registry ( purpose: request.purpose.clone(), }) ); - let submission = BackendJobResultSubmission { - job_id: request.job_id.clone(), - attempt_id: first.attempt.attempt_id.clone(), - input_revision: request.input_revision.clone(), - result: serde_json::json!({"valid": true}), - }; assert!( store .accept_backend_job_result( @@ -19324,9 +19601,43 @@ INSERT INTO worker_registry ( .is_err() ); - // Missing notification targets are durable delivery failures and do - // not alter the already-committed Job result. - store + // Delivery is claimed atomically before transport I/O. Concurrent + // callers converge on one sender, and replays cannot send while that + // claim is active or after it reaches a terminal state. + let barrier = Arc::new(std::sync::Barrier::new(2)); + let claims = std::thread::scope(|scope| { + let handles = (0..2) + .map(|_| { + let store = store.clone(); + let source = source.clone(); + let job_id = request.job_id.clone(); + let barrier = barrier.clone(); + scope.spawn(move || { + barrier.wait(); + store + .reserve_backend_job_delivery( + workspace_id, + "delivery-1", + &job_id, + &source, + "2026-09-01T00:00:05Z", + ) + .unwrap() + }) + }) + .collect::>(); + handles + .into_iter() + .map(|handle| handle.join().unwrap()) + .collect::>() + }); + assert_eq!(claims.iter().filter(|(_, claimed)| *claimed).count(), 1); + assert!( + claims + .iter() + .all(|(delivery, _)| delivery.state == BackendJobDeliveryState::Sending) + ); + let (delivery, claimed) = store .reserve_backend_job_delivery( workspace_id, "delivery-1", @@ -19335,6 +19646,8 @@ INSERT INTO worker_registry ( "2026-09-01T00:00:05Z", ) .unwrap(); + assert!(!claimed); + assert_eq!(delivery.state, BackendJobDeliveryState::Sending); store .finish_backend_job_delivery( workspace_id, @@ -19345,6 +19658,62 @@ INSERT INTO worker_registry ( "2026-09-01T00:00:06Z", ) .unwrap(); + let (_, claimed) = store + .reserve_backend_job_delivery( + workspace_id, + "delivery-1", + &request.job_id, + &source, + "2026-09-01T00:00:07Z", + ) + .unwrap(); + assert!(!claimed); + + // A restart cannot know whether an in-flight transport completed, so + // it terminalizes the claim as unknown instead of sending twice. + let (_, claimed) = store + .reserve_backend_job_delivery( + workspace_id, + "delivery-2", + &request.job_id, + &source, + "2026-09-01T00:00:08Z", + ) + .unwrap(); + assert!(claimed); + assert_eq!( + store + .recover_backend_job_deliveries(workspace_id, "2026-09-01T00:00:09Z") + .unwrap(), + 1 + ); + let (delivery, claimed) = store + .reserve_backend_job_delivery( + workspace_id, + "delivery-2", + &request.job_id, + &source, + "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_eq!( store .get_backend_job(workspace_id, &request.job_id)