fix backend job execution fencing and delivery

crates/worker/src/controller.rs

@@ -1178,7 +1178,10 @@ where
         ),
         None => durable_parent_notifications,
     };
+    let backend_job_profile = worker.manifest().engine.instruction == "internal.backend_job_system";
     let prompts = worker.prompts().clone();
+    let tracker = tools::Tracker::new();
+    if !backend_job_profile {
         let paste_store = worker.store().clone();
         let paste_session_id = worker.session_id();
         worker
@@ -1196,7 +1199,6 @@ where
         // 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(),
@@ -1211,6 +1213,7 @@ where
             )
         };
         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(

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#"

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<Self> {
         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}`"
             ))),

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<String>,
     pub diagnostics: Vec<RuntimeDiagnostic>,
 }
 
@@ -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],
     }
 }

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,

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,8 +3295,7 @@ impl WorkspaceApi {
             operation_id: allocation_key(&reservation.attempt.attempt_id),
             input_fingerprint: reservation.job.intent_fingerprint.clone(),
         };
-        let spawned = self
+        let spawned = match self.spawn_workspace_worker(
-            .spawn_workspace_worker(
             EMBEDDED_WORKER_RUNTIME_ID,
             WorkerSpawnRequest {
                 requested_worker_name: Some(format!("job:{}", request.purpose)),
@@ -3318,8 +3320,22 @@ impl WorkspaceApi {
                 resolved_memory_settings: None,
                 resolved_control_operation: Some(control_operation),
             },
-            )
+        ) {
-            .map_err(|error| error.error)?;
+            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
+        self.schedule_backend_job_timeout(
-            .runtime
+            &request.job_id,
-            .send_input(
+            &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,
             },
-            )
+        ) {
-            .map_err(RuntimeRegistryError::into_error)?;
+            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,
-            &reservation.attempt.deadline_at,
+                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,
+            &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<Json<server_api::RuntimeWorkerInputResult>> {
     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();

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<BackendJobAttemptRecord>;
     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<BackendJobDeliveryRecord>;
+    ) -> Result<(BackendJobDeliveryRecord, bool)>;
+    fn recover_backend_job_deliveries(&self, workspace_id: &str, now: &str) -> Result<usize>;
     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<BackendJobAttemptRecord> {
+        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<BackendJobDeliveryRecord> {
+    ) -> 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);
+                    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((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<usize> {
+        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<BackendJobDeliveryRecord> {
-        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,15 +6321,17 @@ 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 current.state != BackendJobDeliveryState::Sending {
-                if state != BackendJobDeliveryState::Completed {
+                if current.state == state {
-                    return Err(Error::InvalidInput(
-                        "completed Backend Job delivery cannot be replaced".to_string(),
-                    ));
-                }
                     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",
                 params![workspace_id, delivery_id, state.as_str(), failure_category, detail, now],
@@ -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<_>>(),
             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
+        // Delivery is claimed atomically before transport I/O. Concurrent
-        // not alter the already-committed Job result.
+        // 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::<Vec<_>>();
+            handles
+                .into_iter()
+                .map(|handle| handle.join().unwrap())
+                .collect::<Vec<_>>()
+        });
+        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)