Merge remote-tracking branch 'origin/work/T-674-backend-job-runner' into develop

crates/manifest/src/builtin_profile.rs

@@ -94,6 +94,13 @@ pub const BUILTIN_PROFILE_RESOURCES: &[BuiltinProfileResource] = &[
         description: "Workspace orchestration and Worker control.",
         imports: BASE_IMPORT,
     },
+    BuiltinProfileResource {
+        selector: Some("builtin:backend-job"),
+        path: "profiles/backend-job.dcdl",
+        source: include_str!("../../../resources/profiles/backend-job.dcdl"),
+        description: "Backend-owned bounded Job Worker.",
+        imports: BASE_IMPORT,
+    },
     BuiltinProfileResource {
         selector: Some("builtin:memory-consolidation"),
         path: "profiles/memory-consolidation.dcdl",

crates/protocol/src/lib.rs

@@ -1005,6 +1005,18 @@ pub enum Event {
         submission_request_id: String,
         message: String,
     },
+    /// Durable notification acceptance. The request identity is persisted with
+    /// the notification receipt before this event is emitted. Repeating the
+    /// same request id and exact payload returns the same acknowledgement
+    /// without appending the notification twice.
+    NotificationAccepted {
+        notification_request_id: String,
+    },
+    /// Correlated notification rejection before durable acceptance.
+    NotificationRejected {
+        notification_request_id: String,
+        message: String,
+    },
     /// Revisioned FIFO replacement following enqueue, activation, cancel, or clear.
     PendingSubmissionsChanged {
         pending: PendingSubmissionsSnapshot,
@@ -2809,6 +2821,32 @@ mod tests {
         }
     }
 
+    #[test]
+    fn notification_receipt_events_roundtrip_with_request_identity() {
+        for event in [
+            Event::NotificationAccepted {
+                notification_request_id: "notification-1".into(),
+            },
+            Event::NotificationRejected {
+                notification_request_id: "notification-1".into(),
+                message: "invalid notification".into(),
+            },
+        ] {
+            let json = serde_json::to_string(&event).unwrap();
+            let decoded: Event = serde_json::from_str(&json).unwrap();
+            match decoded {
+                Event::NotificationAccepted {
+                    notification_request_id,
+                }
+                | Event::NotificationRejected {
+                    notification_request_id,
+                    ..
+                } => assert_eq!(notification_request_id, "notification-1"),
+                other => panic!("expected notification receipt, got {other:?}"),
+            }
+        }
+    }
+
     #[test]
     fn event_error_format() {
         let event = Event::Error {

crates/protocol/src/subscription.rs

@@ -631,6 +631,24 @@ impl SubscriptionWorkerWorkdirAttachment {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
+#[serde(deny_unknown_fields)]
+pub struct SubscriptionWorkerJob {
+    pub job_id: String,
+    pub attempt_id: String,
+    pub purpose: String,
+}
+
+impl SubscriptionWorkerJob {
+    pub fn validate(&self) -> Result<(), SubscriptionValidationError> {
+        validate_identifier("job_id", &self.job_id, MAX_RESOURCE_ID_BYTES)?;
+        validate_identifier("attempt_id", &self.attempt_id, MAX_RESOURCE_ID_BYTES)?;
+        validate_identifier("job_purpose", &self.purpose, MAX_RESOURCE_ID_BYTES)
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
 #[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
@@ -664,6 +682,10 @@ pub struct SubscriptionWorker {
     pub display_name: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub profile: Option<String>,
+    /// Backend-owned Job binding. Runtime producers leave this unset; the
+    /// Workspace projection enriches both snapshots and live updates.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub job: Option<SubscriptionWorkerJob>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub workdir_attachments: Vec<SubscriptionWorkerWorkdirAttachment>,
 }
@@ -677,6 +699,9 @@ impl SubscriptionWorker {
         if let Some(resource_key) = &self.resource_key {
             validate_identifier("resource_key", resource_key, MAX_RESOURCE_ID_BYTES)?;
         }
+        if let Some(job) = &self.job {
+            job.validate()?;
+        }
         let mut aliases = HashSet::new();
         let mut workdir_ids = HashSet::new();
         for attachment in &self.workdir_attachments {
@@ -952,6 +977,7 @@ mod tests {
             workspace_id: Some("workspace-1".to_string()),
             display_name: Some(format!("Worker {value}")),
             profile: Some("builtin:coder".to_string()),
+            job: None,
             workdir_attachments: Vec::new(),
         }
     }

crates/protocol/src/typescript.rs

@@ -28,7 +28,7 @@ use crate::{
         SubscriptionRequest, SubscriptionRequestId, SubscriptionResponse, SubscriptionSnapshot,
         SubscriptionTerminationCode, SubscriptionWorkdirId, SubscriptionWorker,
         SubscriptionWorkerAvailability, SubscriptionWorkerId, SubscriptionWorkerIds,
-        SubscriptionWorkerProtocolMethod, SubscriptionWorkerState,
+        SubscriptionWorkerJob, SubscriptionWorkerProtocolMethod, SubscriptionWorkerState,
         SubscriptionWorkerWorkdirAttachment, WorkspaceSubscriptionWorkdir,
     },
 };
@@ -132,6 +132,7 @@ pub fn generated_protocol_types() -> String {
     push_decl::<SubscriptionWorkerAvailability>(&cfg, &mut output);
     push_decl::<EventSubscriptionSelector>(&cfg, &mut output);
     push_decl::<SubscriptionWorkerWorkdirAttachment>(&cfg, &mut output);
+    push_decl::<SubscriptionWorkerJob>(&cfg, &mut output);
     push_decl::<SubscriptionWorker>(&cfg, &mut output);
     push_decl::<WorkspaceSubscriptionWorkdir>(&cfg, &mut output);
     push_decl::<SubscriptionSnapshot>(&cfg, &mut output);
@@ -232,6 +233,7 @@ fn rust_serialized_subscription_frame_fixtures() -> Vec<SubscriptionFrame> {
         workspace_id: Some("fixture-workspace".to_string()),
         display_name: None,
         profile: None,
+        job: None,
         workdir_attachments: vec![SubscriptionWorkerWorkdirAttachment {
             alias: "external".to_string(),
             repository_id: None,

crates/tui/src/app.rs

@@ -1143,8 +1143,9 @@ impl App {
         }
 
         match event {
-            Event::SubmissionAccepted { .. } => {}
-            Event::SubmissionRejected { message, .. } => self.push_error(message),
+            Event::SubmissionAccepted { .. } | Event::NotificationAccepted { .. } => {}
+            Event::SubmissionRejected { message, .. }
+            | Event::NotificationRejected { message, .. } => self.push_error(message),
             Event::PendingSubmissionsChanged { pending } => {
                 self.pending_submissions = pending;
             }

crates/worker-runtime/src/execution.rs

@@ -66,6 +66,12 @@ pub struct WorkerSubmissionAck {
     pub disposition: protocol::SubmissionDisposition,
 }
 
+/// Evidence that a notification request reached the durable Worker session boundary.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct WorkerNotificationAck {
+    pub notification_request_id: String,
+}
+
 /// Typed execution result class. Results are transient operation outcomes and
 /// are not persisted as Worker lifecycle authority.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -78,6 +84,8 @@ pub struct WorkerExecutionResult {
     pub message: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub submission: Option<WorkerSubmissionAck>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub notification: Option<WorkerNotificationAck>,
 }
 
 /// Backend result class for a Worker execution operation.
@@ -99,6 +107,7 @@ impl WorkerExecutionResult {
             worker_state: None,
             message: None,
             submission: None,
+            notification: None,
         }
     }
 
@@ -123,6 +132,23 @@ impl WorkerExecutionResult {
                 submission_id: submission_id.into(),
                 disposition,
             }),
+            notification: None,
+        }
+    }
+
+    pub fn accepted_notification(
+        operation: WorkerExecutionOperation,
+        notification_request_id: impl Into<String>,
+    ) -> Self {
+        Self {
+            operation,
+            outcome: WorkerExecutionOutcome::Accepted,
+            worker_state: None,
+            message: None,
+            submission: None,
+            notification: Some(WorkerNotificationAck {
+                notification_request_id: notification_request_id.into(),
+            }),
         }
     }
 
@@ -133,6 +159,7 @@ impl WorkerExecutionResult {
             worker_state: None,
             message: Some(message.into()),
             submission: None,
+            notification: None,
         }
     }
 
@@ -143,6 +170,7 @@ impl WorkerExecutionResult {
             worker_state: None,
             message: Some(message.into()),
             submission: None,
+            notification: None,
         }
     }
 
@@ -153,6 +181,7 @@ impl WorkerExecutionResult {
             worker_state: None,
             message: Some(message.into()),
             submission: None,
+            notification: None,
         }
     }
 
@@ -163,6 +192,7 @@ impl WorkerExecutionResult {
             worker_state: None,
             message: Some(message.into()),
             submission: None,
+            notification: None,
         }
     }
 

crates/worker-runtime/src/interaction.rs

@@ -94,6 +94,9 @@ pub struct WorkerInteractionAck {
     pub worker_ref: WorkerRef,
     pub status: WorkerStatus,
     /// Present for User Submit and absent for non-Submit interactions.
-    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub submission: Option<crate::execution::WorkerSubmissionAck>,
+    /// Present only after a Notify request reaches the Worker's durable receipt boundary.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub notification: Option<crate::execution::WorkerNotificationAck>,
 }

crates/worker-runtime/src/runtime.rs

@@ -2104,6 +2104,12 @@ impl Runtime {
                 .clone()
                 .expect("User input request id must be assigned before dispatch")
         });
+        let expected_notification_id = (input.kind == WorkerInputKind::Notify).then(|| {
+            input
+                .submission_request_id
+                .clone()
+                .expect("Notify request id must be assigned before dispatch")
+        });
         self.ensure_worker_execution(worker_ref)?;
         let (backend, handle) = {
             let state = self.lock()?;
@@ -2160,7 +2166,28 @@ impl Runtime {
             });
         }
 
+        if let Some(expected_notification_id) = expected_notification_id
+            && dispatch_result
+                .notification
+                .as_ref()
+                .is_none_or(|ack| ack.notification_request_id != expected_notification_id)
+        {
+            let result = WorkerExecutionResult::rejected(
+                WorkerExecutionOperation::Input,
+                "execution backend did not acknowledge the durable Runtime notification request id",
+            );
+            self.record_execution_result(worker_ref, result.clone())?;
+            return Err(RuntimeError::WorkerExecutionRejected {
+                worker_id: worker_ref.worker_id.clone(),
+                operation: result.operation,
+                outcome: result.outcome,
+                message: result.message_or_default(),
+                result,
+            });
+        }
+
         let submission = dispatch_result.submission.clone();
+        let notification = dispatch_result.notification.clone();
         let mut state = self.lock()?;
         state.ensure_running()?;
         let worker = state.worker_mut(worker_ref)?;
@@ -2180,6 +2207,7 @@ impl Runtime {
             worker_ref: worker_ref.clone(),
             status,
             submission,
+            notification,
         })
     }
 
@@ -4180,6 +4208,7 @@ impl RuntimeState {
             workspace_id: worker.workspace_id.clone(),
             display_name: worker.display_name.clone(),
             profile,
+            job: None,
             workdir_attachments,
         })
     }
@@ -6450,7 +6479,8 @@ mod tests {
             _handle: &WorkerExecutionHandle,
             input: WorkerInput,
         ) -> WorkerExecutionResult {
-            let submission_id = input.submission_request_id.clone();
+            let request_id = input.submission_request_id.clone();
+            let input_kind = input.kind.clone();
             self.dispatched_inputs.lock().unwrap().push(input);
             let mut result = self
                 .dispatch_result
@@ -6458,20 +6488,28 @@ mod tests {
                 .unwrap()
                 .clone()
                 .unwrap_or_else(|| {
-                    WorkerExecutionResult::accepted_submission(
-                        WorkerExecutionOperation::Input,
-                        "request-test",
-                        "test-submission",
-                        protocol::SubmissionDisposition::Started,
-                    )
+                    if input_kind == WorkerInputKind::Notify {
+                        WorkerExecutionResult::accepted_notification(
+                            WorkerExecutionOperation::Input,
+                            request_id
+                                .clone()
+                                .expect("Runtime assigns Notify request identity before dispatch"),
+                        )
+                    } else {
+                        WorkerExecutionResult::accepted_submission(
+                            WorkerExecutionOperation::Input,
+                            "request-test",
+                            "test-submission",
+                            protocol::SubmissionDisposition::Started,
+                        )
+                    }
                 });
             if !self
                 .preserve_submission_acknowledgement_id
                 .load(Ordering::SeqCst)
-                && let (Some(ack), Some(submission_id)) =
-                    (result.submission.as_mut(), submission_id)
+                && let (Some(ack), Some(request_id)) = (result.submission.as_mut(), request_id)
             {
-                ack.submission_request_id = submission_id;
+                ack.submission_request_id = request_id;
             }
             result
         }
@@ -7910,6 +7948,29 @@ mod tests {
         assert!(runtime.list_workers().unwrap().is_empty());
     }
 
+    #[test]
+    fn notify_rejects_backend_enqueue_without_durable_worker_receipt() {
+        let (runtime, backend) = runtime_and_backend();
+        backend.set_dispatch_result(WorkerExecutionResult::accepted(
+            WorkerExecutionOperation::Input,
+        ));
+        let detail = runtime
+            .create_worker(task_request("notify receipt"))
+            .unwrap();
+        let mut input = WorkerInput::notify("do not acknowledge enqueue alone");
+        input.submission_request_id = Some("notification-request".to_string());
+
+        let error = runtime.send_input(&detail.worker_ref, input).unwrap_err();
+
+        assert!(matches!(
+            error,
+            RuntimeError::WorkerExecutionRejected {
+                outcome: crate::execution::WorkerExecutionOutcome::Rejected,
+                ..
+            }
+        ));
+    }
+
     #[test]
     fn connected_backend_busy_dispatch_is_typed_and_not_transcribed() {
         let (runtime, backend) = runtime_and_backend();

crates/worker-runtime/src/worker_backend.rs

@@ -45,6 +45,11 @@ use protocol::{Event, Method, Segment, WorkerCommandEnvelope};
 
 static NEXT_INTERNAL_COMMAND_ID: AtomicU64 = AtomicU64::new(1);
 
+enum NotificationAcceptance {
+    Accepted,
+    Rejected(String),
+}
+
 fn rollback_materialized_spawn_workdirs(
     materializer: &dyn WorkingDirectoryMaterializer,
     bindings: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
@@ -1883,6 +1888,75 @@ where
         .unwrap_or_else(|message| WorkerExecutionResult::errored(operation, message))
     }
 
+    fn send_notification_and_wait_for_acceptance(
+        &self,
+        operation: WorkerExecutionOperation,
+        worker: WorkerHandle,
+        method: Method,
+        notification_request_id: String,
+    ) -> WorkerExecutionResult {
+        let request_id = notification_request_id.clone();
+        let result = self.run_cancellable_on_adapter_runtime(USER_INPUT_TASK_TIMEOUT, async move {
+            // Subscribe before enqueueing so a fast durable acceptance cannot
+            // race the Runtime acknowledgement.
+            let mut events = worker.subscribe();
+            worker
+                .send(method)
+                .await
+                .map_err(|err| format!("failed to send Worker method: {err}"))?;
+
+            tokio::time::timeout(USER_INPUT_COMMIT_TIMEOUT, async move {
+                loop {
+                    match events.recv().await {
+                        Ok(Event::NotificationAccepted {
+                            notification_request_id,
+                        }) if notification_request_id == request_id => {
+                            return Ok(NotificationAcceptance::Accepted);
+                        }
+                        Ok(Event::NotificationRejected {
+                            notification_request_id,
+                            message,
+                        }) if notification_request_id == request_id => {
+                            return Ok(NotificationAcceptance::Rejected(message));
+                        }
+                        Ok(Event::Shutdown) => {
+                            return Err(
+                                "worker shut down before notification was durably accepted"
+                                    .to_string(),
+                            );
+                        }
+                        Ok(_) => {}
+                        Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
+                            return Err(format!(
+                                "worker notification acknowledgement lagged by {skipped} protocol event(s)"
+                            ));
+                        }
+                        Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                            return Err(
+                                "worker event stream closed before notification was durably accepted"
+                                    .to_string(),
+                            );
+                        }
+                    }
+                }
+            })
+            .await
+            .map_err(|_| {
+                "timed out waiting for durable Worker notification acceptance".to_string()
+            })?
+        });
+
+        match result {
+            Ok(NotificationAcceptance::Accepted) => {
+                WorkerExecutionResult::accepted_notification(operation, notification_request_id)
+            }
+            Ok(NotificationAcceptance::Rejected(message)) => {
+                WorkerExecutionResult::rejected(operation, message)
+            }
+            Err(message) => WorkerExecutionResult::errored(operation, message),
+        }
+    }
+
     fn retain_uncertain_unconnected_controller(
         &self,
         worker_ref: &crate::identity::WorkerRef,
@@ -2879,16 +2953,17 @@ where
             let notification_request_id = input
                 .submission_request_id
                 .unwrap_or_else(protocol::new_submission_request_id);
-            return self.send_method(
+            return self.send_notification_and_wait_for_acceptance(
                 WorkerExecutionOperation::Input,
                 worker,
                 Method::NotifyTracked {
                     notification_request_id: notification_request_id.clone(),
                     message: input.content,
                     source: protocol::AuthenticatedInputSource::Backend {
-                        operation_id: notification_request_id,
+                        operation_id: notification_request_id.clone(),
                     },
                 },
+                notification_request_id,
             );
         }
 
@@ -5108,7 +5183,7 @@ mod tests {
     }
 
     #[test]
-    fn running_worker_accepts_notify_without_a_submit_receipt() {
+    fn running_worker_waits_for_durable_notify_receipt() {
         let client = MockClient::sequential(vec![MockResponse::Hang(vec![])]);
         let runtime_base = tempfile::tempdir().unwrap();
         let cwd = tempfile::tempdir().unwrap();
@@ -5143,6 +5218,13 @@ mod tests {
             .expect("Running Worker must accept Notify without a Submit receipt");
 
         assert!(acknowledgement.submission.is_none());
+        assert_eq!(
+            acknowledgement
+                .notification
+                .as_ref()
+                .map(|ack| ack.notification_request_id.as_str()),
+            Some("notification-request")
+        );
         wait_for_adapter_state(&backend, &detail.worker_ref, WorkerStatus::Running);
     }
 

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,6 +1228,20 @@ where
     let worker_enabled = feature_config.worker.enabled;
     let sub_worker_enabled = feature_config.sub_worker.enabled;
     let mut feature_registry = FeatureRegistryBuilder::new();
+    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(
+                std::io::ErrorKind::InvalidInput,
+                "Backend Job result capability requires Backend Workspace API authority",
+            ));
+        }
+        feature_registry.add_module(
+            crate::feature::builtin::backend_job_result::BackendJobResultFeature::new(
+                workspace_client,
+            ),
+        );
+    }
     let memory_install_plan = crate::feature::builtin::memory::MemoryFeatureInstallPlan::prepare(
         worker.manifest(),
         worker.workspace_client_handle(),
@@ -1472,6 +1489,40 @@ where
     Ok(workdir_for_view)
 }
 
+fn durably_accept_notification<St>(
+    notification_request_id: String,
+    message: String,
+    source_namespace: String,
+    provenance: session_store::LoggedSessionHistoryOrigin,
+    pending_submissions: &PendingSubmissionHandle<St>,
+    working_event_tx: &broadcast::Sender<Event>,
+) -> bool
+where
+    St: session_store::Store + Clone,
+{
+    let request_id = notification_request_id.clone();
+    match pending_submissions.accept_notification_from_source(
+        notification_request_id,
+        message,
+        source_namespace,
+        provenance,
+    ) {
+        Ok(_) => {
+            let _ = working_event_tx.send(Event::NotificationAccepted {
+                notification_request_id: request_id,
+            });
+            true
+        }
+        Err(error) => {
+            let _ = working_event_tx.send(Event::NotificationRejected {
+                notification_request_id: request_id,
+                message: error.to_string(),
+            });
+            false
+        }
+    }
+}
+
 fn durably_accept_method_while_busy<St>(
     method: Method,
     pending_submissions: &PendingSubmissionHandle<St>,
@@ -1544,17 +1595,14 @@ where
             notification_request_id,
             message,
         } => {
-            if let Err(error) = pending_submissions.accept_notification_from_source(
+            durably_accept_notification(
                 notification_request_id,
                 message,
                 pending_submissions.direct_client_namespace(),
                 session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
-            ) {
-                let _ = working_event_tx.send(Event::Error {
-                    code: ErrorCode::InvalidRequest,
-                    message: error.to_string(),
-                });
-            }
+                pending_submissions,
+                working_event_tx,
+            );
             None
         }
         Method::NotifyTracked {
@@ -1564,17 +1612,14 @@ where
         } => {
             let (source_namespace, provenance) =
                 resolved_input_source(pending_submissions, &source);
-            if let Err(error) = pending_submissions.accept_notification_from_source(
+            durably_accept_notification(
                 notification_request_id,
                 message,
                 source_namespace,
                 provenance,
-            ) {
-                let _ = working_event_tx.send(Event::Error {
-                    code: ErrorCode::InvalidRequest,
-                    message: error.to_string(),
-                });
-            }
+                pending_submissions,
+                working_event_tx,
+            );
             None
         }
         method => Some(method),
@@ -1602,9 +1647,16 @@ fn reject_method_while_attention_locked(
             });
             true
         }
-        Method::Notify { .. } | Method::NotifyTracked { .. } => {
-            let _ = working_event_tx.send(Event::Error {
-                code: ErrorCode::InvalidRequest,
+        Method::Notify {
+            notification_request_id,
+            ..
+        }
+        | Method::NotifyTracked {
+            notification_request_id,
+            ..
+        } => {
+            let _ = working_event_tx.send(Event::NotificationRejected {
+                notification_request_id: notification_request_id.clone(),
                 message: message.to_owned(),
             });
             true
@@ -2014,21 +2066,14 @@ async fn controller_loop<C, St>(
                 notification_request_id,
                 message,
             } => {
-                let source_namespace = pending_submissions.direct_client_namespace();
-                match pending_submissions.accept_notification_from_source(
+                durably_accept_notification(
                     notification_request_id,
                     message,
-                    source_namespace.clone(),
+                    pending_submissions.direct_client_namespace(),
                     session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
-                ) {
-                    Ok(_) => {}
-                    Err(error) => {
-                        let _ = working_event_tx.send(Event::Error {
-                            code: ErrorCode::InvalidRequest,
-                            message: error.to_string(),
-                        });
-                    }
-                }
+                    &pending_submissions,
+                    &working_event_tx,
+                );
             }
 
             Method::NotifyTracked {
@@ -2038,20 +2083,14 @@ async fn controller_loop<C, St>(
             } => {
                 let (source_namespace, provenance) =
                     resolved_input_source(&pending_submissions, &source);
-                match pending_submissions.accept_notification_from_source(
+                durably_accept_notification(
                     notification_request_id,
                     message,
-                    source_namespace.clone(),
+                    source_namespace,
                     provenance,
-                ) {
-                    Ok(_) => {}
-                    Err(error) => {
-                        let _ = working_event_tx.send(Event::Error {
-                            code: ErrorCode::InvalidRequest,
-                            message: error.to_string(),
-                        });
-                    }
-                }
+                    &pending_submissions,
+                    &working_event_tx,
+                );
             }
 
             Method::ListPendingSubmissions => {
@@ -3096,22 +3135,15 @@ where
                         notification_request_id,
                         message,
                     }) => {
-                        let source_namespace = pending_submissions.direct_client_namespace();
-                        match pending_submissions.accept_notification_from_source(
+                        if durably_accept_notification(
                             notification_request_id,
                             message,
-                            source_namespace.clone(),
+                            pending_submissions.direct_client_namespace(),
                             session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
+                            pending_submissions,
+                            working_event_tx,
                         ) {
-                            Ok(_) => {
-                                stage_pending_notifications(&pending_submissions, notify_buffer);
-                            }
-                            Err(error) => {
-                                let _ = working_event_tx.send(Event::Error {
-                                    code: ErrorCode::InvalidRequest,
-                                    message: error.to_string(),
-                                });
-                            }
+                            stage_pending_notifications(&pending_submissions, notify_buffer);
                         }
                         let _ = working_event_tx.send(Event::PendingSubmissionsChanged {
                             pending: pending_submissions.snapshot(),
@@ -3124,21 +3156,15 @@ where
                     }) => {
                         let (source_namespace, provenance) =
                             resolved_input_source(pending_submissions, &source);
-                        match pending_submissions.accept_notification_from_source(
+                        if durably_accept_notification(
                             notification_request_id,
                             message,
-                            source_namespace.clone(),
+                            source_namespace,
                             provenance,
+                            pending_submissions,
+                            working_event_tx,
                         ) {
-                            Ok(_) => {
-                                stage_pending_notifications(&pending_submissions, notify_buffer);
-                            }
-                            Err(error) => {
-                                let _ = working_event_tx.send(Event::Error {
-                                    code: ErrorCode::InvalidRequest,
-                                    message: error.to_string(),
-                                });
-                            }
+                            stage_pending_notifications(&pending_submissions, notify_buffer);
                         }
                         let _ = working_event_tx.send(Event::PendingSubmissionsChanged {
                             pending: pending_submissions.snapshot(),
@@ -3605,29 +3631,46 @@ mod tests {
     }
 
     #[test]
-    fn busy_notification_is_durable_before_shutdown_can_observe_it() {
+    fn busy_notification_receipt_follows_durable_idempotent_acceptance() {
         let dir = tempfile::tempdir().unwrap();
         let pending = PendingSubmissionHandle::for_test(dir.path());
-        let (event_tx, _event_rx) = broadcast::channel(4);
+        let (event_tx, mut event_rx) = broadcast::channel(4);
+        let method = Method::NotifyTracked {
+            notification_request_id: "completion-1".into(),
+            message: "subworker completed".into(),
+            source: protocol::AuthenticatedInputSource::UntrustedWire,
+        };
 
-        assert!(
-            durably_accept_method_while_busy(
-                Method::NotifyTracked {
-                    notification_request_id: "completion-1".into(),
-                    message: "subworker completed".into(),
-                    source: protocol::AuthenticatedInputSource::UntrustedWire,
-                },
-                &pending,
-                &event_tx,
-            )
-            .is_none()
+        for _ in 0..2 {
+            assert!(
+                durably_accept_method_while_busy(method.clone(), &pending, &event_tx).is_none()
+            );
+            let entries = pending.persisted_entries_for_test();
+            assert!(entries.iter().any(|entry| matches!(
+                entry,
+                LogEntry::Extension { domain, .. }
+                    if domain == "worker.pending_activations.v1"
+            )));
+            assert!(matches!(
+                event_rx.try_recv().unwrap(),
+                Event::NotificationAccepted {
+                    notification_request_id,
+                } if notification_request_id == "completion-1"
+            ));
+        }
+        assert_eq!(pending.snapshot().notification_count, 1);
+        assert_eq!(
+            pending
+                .persisted_entries_for_test()
+                .iter()
+                .filter(|entry| matches!(
+                    entry,
+                    LogEntry::Extension { domain, .. }
+                        if domain == "worker.pending_activations.v1"
+                ))
+                .count(),
+            1
         );
-        let entries = pending.persisted_entries_for_test();
-        assert!(entries.iter().any(|entry| matches!(
-            entry,
-            LogEntry::Extension { domain, .. }
-                if domain == "worker.pending_activations.v1"
-        )));
     }
 
     #[test]

crates/worker/src/feature/builtin.rs

@@ -4,6 +4,7 @@
 //! same descriptor-approved registry path used by feature modules. They are not
 //! an external plugin-loading surface.
 
+pub mod backend_job_result;
 pub mod flow_transition;
 pub mod manage_workdir;
 pub mod manage_worker;

crates/worker/src/feature/builtin/backend_job_result.rs

@@ -0,0 +1,175 @@
+//! Structured completion capability for Backend-owned Job Workers.
+//!
+//! The tool intentionally accepts only the result value. Workspace and Worker
+//! identity come from the authenticated Workspace client, while Job, attempt,
+//! and input revision are resolved from the Backend-owned binding.
+
+use std::sync::Arc;
+
+use agen::tool::{Tool, ToolError, ToolExecutionContext, ToolMeta, ToolOutput};
+use async_trait::async_trait;
+use schemars::JsonSchema;
+use serde::Deserialize;
+use serde_json::Value;
+
+use crate::feature::{
+    FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution,
+    ToolDeclaration, ToolDefinition,
+};
+use crate::worker::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod};
+
+const TOOL_NAME: &str = "SubmitBackendJobResult";
+const TOOL_DESCRIPTION: &str = "Submit the bounded structured result for this Backend-owned Job attempt. Workspace, Job, attempt, input revision, and source Worker identity are resolved and fenced by Backend authority; normal final prose is not success.";
+
+#[derive(Debug, Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct SubmitResultInput {
+    result: Value,
+}
+
+#[derive(Clone)]
+struct SubmitResultTool {
+    client: Arc<dyn WorkspaceClient>,
+}
+
+#[async_trait]
+impl Tool for SubmitResultTool {
+    async fn execute(
+        &self,
+        input_json: &str,
+        _context: ToolExecutionContext,
+    ) -> Result<ToolOutput, ToolError> {
+        let input: SubmitResultInput = serde_json::from_str(input_json)
+            .map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
+        let workspace_id = self.client.workspace_id().ok_or_else(|| {
+            ToolError::ExecutionFailed(
+                "Backend Job result submission requires Workspace identity".to_string(),
+            )
+        })?;
+        let response = self
+            .client
+            .execute(WorkspaceRequest::json(
+                WorkspaceRequestMethod::Post,
+                format!("/api/w/{workspace_id}/workers/self/backend-job-result"),
+                serde_json::to_string(&serde_json::json!({ "result": input.result }))
+                    .map_err(|error| ToolError::Internal(error.to_string()))?,
+            ))
+            .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
+        if !response.is_success() {
+            return Err(ToolError::ExecutionFailed(format!(
+                "Backend Job result submission was rejected with HTTP {}: {}",
+                response.status, response.body
+            )));
+        }
+        Ok(ToolOutput {
+            summary: "Backend Job result accepted".to_string(),
+            content: Some(response.body),
+            attachments: Vec::new(),
+        })
+    }
+}
+
+fn definition(client: Arc<dyn WorkspaceClient>) -> ToolDefinition {
+    let schema = serde_json::to_value(schemars::schema_for!(SubmitResultInput))
+        .expect("Backend Job result schema must serialize");
+    Arc::new(move || {
+        (
+            ToolMeta::new(TOOL_NAME)
+                .description(TOOL_DESCRIPTION)
+                .input_schema(schema.clone()),
+            Arc::new(SubmitResultTool {
+                client: client.clone(),
+            }) as Arc<dyn Tool>,
+        )
+    })
+}
+
+#[derive(Clone)]
+pub struct BackendJobResultFeature {
+    client: Arc<dyn WorkspaceClient>,
+}
+
+impl BackendJobResultFeature {
+    pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
+        Self { client }
+    }
+}
+
+impl FeatureModule for BackendJobResultFeature {
+    fn descriptor(&self) -> FeatureDescriptor {
+        FeatureDescriptor::builtin("backend-job-result", "Backend Job result")
+            .with_description("Single-purpose structured result capability for Backend-owned Jobs.")
+            .with_tool(ToolDeclaration::new(TOOL_NAME, TOOL_DESCRIPTION))
+    }
+
+    fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
+        context.tools().register(ToolContribution::new(
+            TOOL_NAME,
+            definition(self.client.clone()),
+        ))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::feature::FeatureRegistryBuilder;
+    use crate::hook::HookRegistryBuilder;
+    use crate::worker::{WorkspaceClientError, WorkspaceResponse};
+    use std::sync::Mutex;
+
+    #[derive(Debug, Default)]
+    struct RecordingClient {
+        requests: Mutex<Vec<WorkspaceRequest>>,
+    }
+
+    impl WorkspaceClient for RecordingClient {
+        fn workspace_id(&self) -> Option<&str> {
+            Some("workspace-test")
+        }
+        fn kind(&self) -> &str {
+            "test"
+        }
+        fn is_available(&self) -> bool {
+            true
+        }
+        fn execute(
+            &self,
+            request: WorkspaceRequest,
+        ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+            self.requests.lock().unwrap().push(request);
+            Ok(WorkspaceResponse {
+                status: 200,
+                body: "{\"replayed\":false}".to_string(),
+            })
+        }
+    }
+
+    #[tokio::test]
+    async fn feature_exposes_only_identity_free_structured_result_input() {
+        let client = Arc::new(RecordingClient::default());
+        let mut pending = Vec::new();
+        let mut hooks = HookRegistryBuilder::default();
+        let report = FeatureRegistryBuilder::new()
+            .with_module(BackendJobResultFeature::new(client.clone()))
+            .install_into_pending(&mut pending, &mut hooks);
+        assert_eq!(report.installed_tool_names(), [TOOL_NAME]);
+        let (meta, tool) = pending.remove(0)();
+        assert_eq!(meta.input_schema["required"], serde_json::json!(["result"]));
+        assert!(meta.input_schema["properties"].get("job_id").is_none());
+        tool.execute(
+            r#"{"result":{"valid":true}}"#,
+            ToolExecutionContext::direct(),
+        )
+        .await
+        .unwrap();
+        let requests = client.requests.lock().unwrap();
+        assert_eq!(requests.len(), 1);
+        assert_eq!(requests[0].method, WorkspaceRequestMethod::Post);
+        assert!(
+            requests[0]
+                .path
+                .ends_with("/workers/self/backend-job-result")
+        );
+    }
+}

crates/worker/src/prompt/catalog.rs

@@ -95,6 +95,7 @@ pub enum WorkerPrompt {
     MemoryExtractSystem,
     MemoryConsolidationSystem,
     FlowVerifierSystem,
+    BackendJobSystem,
     NotifyWrapper,
     InterruptToolResultSummary,
     InterruptSystemNote,
@@ -112,6 +113,7 @@ impl WorkerPrompt {
             Self::MemoryExtractSystem => "internal.memory_extract_system",
             Self::MemoryConsolidationSystem => "internal.memory_consolidation_system",
             Self::FlowVerifierSystem => "internal.flow_verifier_system",
+            Self::BackendJobSystem => "internal.backend_job_system",
             Self::NotifyWrapper => "internal.notify_wrapper",
             Self::InterruptToolResultSummary => "internal.interrupt_tool_result_summary",
             Self::InterruptSystemNote => "internal.interrupt_system_note",
@@ -130,6 +132,7 @@ impl WorkerPrompt {
         WorkerPrompt::MemoryExtractSystem,
         WorkerPrompt::MemoryConsolidationSystem,
         WorkerPrompt::FlowVerifierSystem,
+        WorkerPrompt::BackendJobSystem,
         WorkerPrompt::NotifyWrapper,
         WorkerPrompt::InterruptToolResultSummary,
         WorkerPrompt::InterruptSystemNote,

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

@@ -0,0 +1,538 @@
+//! Durable Backend-owned jobs executed by ordinary Runtime-managed Workers.
+//!
+//! A job is durable intent. An attempt binds that intent to one dedicated
+//! Worker, and a result is accepted only through the fenced structured-result
+//! API below. Runtime lifecycle and final prose are deliberately not success
+//! authority.
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use sha2::{Digest, Sha256};
+use worker_runtime::identity::RuntimeWorkerRef;
+
+use crate::{Error, Result};
+
+pub const BACKEND_JOB_RUNTIME_ID: &str = crate::hosts::EMBEDDED_RUNTIME_ID;
+pub const DEFAULT_MAX_CONCURRENT_JOBS: u16 = 8;
+pub const ABSOLUTE_MAX_CONCURRENT_JOBS: u16 = 64;
+pub const DEFAULT_JOB_TIMEOUT_SECONDS: u32 = 120;
+pub const MAX_JOB_TIMEOUT_SECONDS: u32 = 600;
+pub const DEFAULT_MAX_RESULT_BYTES: u32 = 16 * 1024;
+pub const ABSOLUTE_MAX_RESULT_BYTES: u32 = 64 * 1024;
+pub const DEFAULT_MAX_ATTEMPTS: u8 = 2;
+pub const ABSOLUTE_MAX_ATTEMPTS: u8 = 3;
+pub const MAX_JOB_INPUT_BYTES: usize = 128 * 1024;
+pub const MAX_JOB_INSTRUCTION_BYTES: usize = 16 * 1024;
+pub const MAX_JOB_PURPOSE_BYTES: usize = 120;
+pub const MAX_JOB_REFERENCE_BYTES: usize = 512;
+pub const MAX_JOB_FAILURE_BYTES: usize = 1024;
+pub const MAX_JOB_DELIVERY_BYTES: usize = 16 * 1024;
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum BackendJobState {
+    Pending,
+    Completed,
+    Failed,
+    Unknown,
+}
+
+impl BackendJobState {
+    pub(crate) fn as_str(self) -> &'static str {
+        match self {
+            Self::Pending => "pending",
+            Self::Completed => "completed",
+            Self::Failed => "failed",
+            Self::Unknown => "unknown",
+        }
+    }
+
+    pub(crate) fn parse(value: &str) -> Result<Self> {
+        match value {
+            "pending" => Ok(Self::Pending),
+            "completed" => Ok(Self::Completed),
+            "failed" => Ok(Self::Failed),
+            "unknown" => Ok(Self::Unknown),
+            other => Err(Error::Store(format!("unknown Backend Job state `{other}`"))),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum BackendJobAttemptState {
+    Reserved,
+    Dispatched,
+    Completed,
+    Failed,
+    Unknown,
+}
+
+impl BackendJobAttemptState {
+    pub(crate) fn as_str(self) -> &'static str {
+        match self {
+            Self::Reserved => "reserved",
+            Self::Dispatched => "dispatched",
+            Self::Completed => "completed",
+            Self::Failed => "failed",
+            Self::Unknown => "unknown",
+        }
+    }
+
+    pub(crate) fn parse(value: &str) -> Result<Self> {
+        match value {
+            "reserved" => Ok(Self::Reserved),
+            "dispatched" => Ok(Self::Dispatched),
+            "completed" => Ok(Self::Completed),
+            "failed" => Ok(Self::Failed),
+            "unknown" => Ok(Self::Unknown),
+            other => Err(Error::Store(format!(
+                "unknown Backend Job attempt state `{other}`"
+            ))),
+        }
+    }
+
+    pub fn terminal(self) -> bool {
+        matches!(self, Self::Completed | Self::Failed | Self::Unknown)
+    }
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[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}`"
+            ))),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct BackendJobLimits {
+    #[serde(default = "default_max_concurrent_jobs")]
+    pub max_concurrent_jobs: u16,
+    #[serde(default = "default_job_timeout_seconds")]
+    pub timeout_seconds: u32,
+    #[serde(default = "default_max_result_bytes")]
+    pub max_result_bytes: u32,
+    #[serde(default = "default_max_attempts")]
+    pub max_attempts: u8,
+}
+
+impl Default for BackendJobLimits {
+    fn default() -> Self {
+        Self {
+            max_concurrent_jobs: DEFAULT_MAX_CONCURRENT_JOBS,
+            timeout_seconds: DEFAULT_JOB_TIMEOUT_SECONDS,
+            max_result_bytes: DEFAULT_MAX_RESULT_BYTES,
+            max_attempts: DEFAULT_MAX_ATTEMPTS,
+        }
+    }
+}
+
+impl BackendJobLimits {
+    pub fn validate(&self) -> Result<()> {
+        if self.max_concurrent_jobs == 0 || self.max_concurrent_jobs > ABSOLUTE_MAX_CONCURRENT_JOBS
+        {
+            return Err(Error::InvalidInput(format!(
+                "Backend Job max_concurrent_jobs must be in 1..={ABSOLUTE_MAX_CONCURRENT_JOBS}"
+            )));
+        }
+        if self.timeout_seconds == 0 || self.timeout_seconds > MAX_JOB_TIMEOUT_SECONDS {
+            return Err(Error::InvalidInput(format!(
+                "Backend Job timeout_seconds must be in 1..={MAX_JOB_TIMEOUT_SECONDS}"
+            )));
+        }
+        if self.max_result_bytes == 0 || self.max_result_bytes > ABSOLUTE_MAX_RESULT_BYTES {
+            return Err(Error::InvalidInput(format!(
+                "Backend Job max_result_bytes must be in 1..={ABSOLUTE_MAX_RESULT_BYTES}"
+            )));
+        }
+        if self.max_attempts == 0 || self.max_attempts > ABSOLUTE_MAX_ATTEMPTS {
+            return Err(Error::InvalidInput(format!(
+                "Backend Job max_attempts must be in 1..={ABSOLUTE_MAX_ATTEMPTS}"
+            )));
+        }
+        Ok(())
+    }
+}
+
+const fn default_max_concurrent_jobs() -> u16 {
+    DEFAULT_MAX_CONCURRENT_JOBS
+}
+const fn default_job_timeout_seconds() -> u32 {
+    DEFAULT_JOB_TIMEOUT_SECONDS
+}
+const fn default_max_result_bytes() -> u32 {
+    DEFAULT_MAX_RESULT_BYTES
+}
+const fn default_max_attempts() -> u8 {
+    DEFAULT_MAX_ATTEMPTS
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
+pub struct BackendJobRequest {
+    /// Stable caller-selected idempotency key within one Workspace.
+    pub job_id: String,
+    /// Stable machine-readable use-site name, for example `ticket_item_check`.
+    pub purpose: String,
+    /// Revision fenced by the structured result.
+    pub input_revision: String,
+    /// Durable domain reference used to reload or explain the input.
+    pub input_ref: String,
+    /// Bounded immutable input snapshot. This must not contain credentials.
+    pub input: Value,
+    /// Bounded task instruction; the profile supplies the system instruction.
+    pub instruction: String,
+    /// Existing configured profile selector. Job dispatch never accepts raw profile source.
+    pub profile: String,
+    /// Writer provenance and optional advisory destination, not parent ownership.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub source_worker: Option<RuntimeWorkerRef>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub notification_target: Option<RuntimeWorkerRef>,
+    #[serde(default)]
+    pub limits: BackendJobLimits,
+}
+
+impl BackendJobRequest {
+    pub fn validate(&self) -> Result<()> {
+        validate_identifier("job_id", &self.job_id, 256)?;
+        validate_identifier("purpose", &self.purpose, MAX_JOB_PURPOSE_BYTES)?;
+        validate_identifier("input_revision", &self.input_revision, 256)?;
+        validate_bounded_text("input_ref", &self.input_ref, MAX_JOB_REFERENCE_BYTES)?;
+        validate_bounded_text("instruction", &self.instruction, MAX_JOB_INSTRUCTION_BYTES)?;
+        validate_identifier("profile", &self.profile, 256)?;
+        if self.profile != "builtin:backend-job" {
+            return Err(Error::InvalidInput(
+                "Backend Jobs must use the dedicated builtin:backend-job profile".to_string(),
+            ));
+        }
+        let input = serde_json::to_vec(&self.input).map_err(|error| {
+            Error::InvalidInput(format!("serialize Backend Job input: {error}"))
+        })?;
+        if input.len() > MAX_JOB_INPUT_BYTES {
+            return Err(Error::InvalidInput(format!(
+                "Backend Job input exceeds {MAX_JOB_INPUT_BYTES} bytes"
+            )));
+        }
+        self.limits.validate()
+    }
+
+    pub fn fingerprint(&self) -> Result<String> {
+        self.validate()?;
+        let encoded = serde_json::to_vec(self).map_err(|error| {
+            Error::InvalidInput(format!("serialize Backend Job intent: {error}"))
+        })?;
+        Ok(sha256(&encoded))
+    }
+
+    pub fn worker_input(&self, attempt_id: &str) -> Result<String> {
+        let input = serde_json::to_string(&self.input).map_err(|error| {
+            Error::InvalidInput(format!("serialize Backend Job input: {error}"))
+        })?;
+        Ok(format!(
+            "{}\n\nBackend Job envelope (immutable):\njob_id: {}\nattempt_id: {}\ninput_revision: {}\ninput_ref: {}\ninput_json: {}\n\nReturn success only through the structured Backend Job result capability. Final prose and Worker Idle/Stopped state are not result authority.",
+            self.instruction, self.job_id, attempt_id, self.input_revision, self.input_ref, input
+        ))
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct BackendJobRecord {
+    pub workspace_id: String,
+    pub request: BackendJobRequest,
+    pub intent_fingerprint: String,
+    pub state: BackendJobState,
+    pub current_attempt: u8,
+    pub result: Option<Value>,
+    pub result_digest: Option<String>,
+    pub failure_category: Option<String>,
+    pub failure_detail: Option<String>,
+    pub created_at: String,
+    pub updated_at: String,
+    pub completed_at: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct BackendJobAttemptRecord {
+    pub workspace_id: String,
+    pub job_id: String,
+    pub attempt_id: String,
+    pub attempt: u8,
+    pub input_revision: String,
+    pub state: BackendJobAttemptState,
+    pub worker: Option<RuntimeWorkerRef>,
+    pub runtime_run_id: Option<String>,
+    pub dispatched_at: Option<String>,
+    pub deadline_at: String,
+    pub result: Option<Value>,
+    pub result_digest: Option<String>,
+    pub failure_category: Option<String>,
+    pub failure_detail: Option<String>,
+    pub created_at: String,
+    pub updated_at: String,
+    pub completed_at: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct BackendJobReservation {
+    pub job: BackendJobRecord,
+    pub attempt: BackendJobAttemptRecord,
+    pub replayed: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
+pub struct BackendJobResultSubmission {
+    pub job_id: String,
+    pub attempt_id: String,
+    pub input_revision: String,
+    pub result: Value,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct BackendJobResultAcceptance {
+    pub job: BackendJobRecord,
+    pub attempt: BackendJobAttemptRecord,
+    pub replayed: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct BackendJobWorkerBinding {
+    pub job_id: String,
+    pub attempt_id: String,
+    pub purpose: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct BackendJobDeliveryRecord {
+    pub workspace_id: String,
+    pub delivery_id: String,
+    pub job_id: String,
+    pub attempt_id: String,
+    pub target: RuntimeWorkerRef,
+    pub state: BackendJobDeliveryState,
+    pub failure_category: Option<String>,
+    pub failure_detail: Option<String>,
+    pub created_at: String,
+    pub updated_at: String,
+    pub delivered_at: Option<String>,
+}
+
+pub fn attempt_id(job_id: &str, attempt: u8) -> String {
+    format!("{job_id}:attempt:{attempt}")
+}
+
+pub fn allocation_key(attempt_id: &str) -> String {
+    format!("backend-job:{attempt_id}")
+}
+
+/// Fixed-size Runtime request identity for the exact Job attempt. Caller-selected
+/// Job ids may be longer than the Worker protocol request-id bound, so never
+/// embed them directly in tracked Runtime identities.
+pub fn runtime_run_id(attempt_id: &str) -> String {
+    tracked_request_id("run", attempt_id.as_bytes())
+}
+
+/// Fixed-size durable delivery identity, also reused as the Worker's tracked
+/// notification request id so restart replay converges on one receipt.
+pub fn delivery_id(job_id: &str, attempt_id: &str) -> String {
+    let mut identity = Vec::with_capacity(job_id.len() + attempt_id.len() + 1);
+    identity.extend_from_slice(job_id.as_bytes());
+    identity.push(0);
+    identity.extend_from_slice(attempt_id.as_bytes());
+    tracked_request_id("notification", &identity)
+}
+
+fn tracked_request_id(kind: &str, identity: &[u8]) -> String {
+    format!("backend-job-{kind}:{}", sha256(identity))
+}
+
+pub fn result_digest(result: &Value, max_bytes: u32) -> Result<(String, String)> {
+    let encoded = serde_json::to_string(result)
+        .map_err(|error| Error::InvalidInput(format!("serialize Backend Job result: {error}")))?;
+    if encoded.len() > max_bytes as usize {
+        return Err(Error::InvalidInput(format!(
+            "Backend Job result exceeds {max_bytes} bytes"
+        )));
+    }
+    Ok((sha256(encoded.as_bytes()), encoded))
+}
+
+pub(crate) fn bounded_failure_detail(detail: &str) -> String {
+    if detail.len() <= MAX_JOB_FAILURE_BYTES {
+        return detail.to_string();
+    }
+    let mut end = MAX_JOB_FAILURE_BYTES;
+    while !detail.is_char_boundary(end) {
+        end -= 1;
+    }
+    detail[..end].to_string()
+}
+
+fn validate_identifier(field: &str, value: &str, max_bytes: usize) -> Result<()> {
+    validate_bounded_text(field, value, max_bytes)?;
+    if value.chars().any(char::is_control) {
+        return Err(Error::InvalidInput(format!(
+            "Backend Job {field} must not contain control characters"
+        )));
+    }
+    Ok(())
+}
+
+fn validate_bounded_text(field: &str, value: &str, max_bytes: usize) -> Result<()> {
+    if value.is_empty() || value.trim() != value {
+        return Err(Error::InvalidInput(format!(
+            "Backend Job {field} must be non-empty and trimmed"
+        )));
+    }
+    if value.len() > max_bytes {
+        return Err(Error::InvalidInput(format!(
+            "Backend Job {field} exceeds {max_bytes} bytes"
+        )));
+    }
+    Ok(())
+}
+
+fn sha256(bytes: &[u8]) -> String {
+    let digest = Sha256::digest(bytes);
+    let mut output = String::with_capacity(7 + digest.len() * 2);
+    output.push_str("sha256:");
+    for byte in digest {
+        use std::fmt::Write as _;
+        let _ = write!(output, "{byte:02x}");
+    }
+    output
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn request() -> BackendJobRequest {
+        BackendJobRequest {
+            job_id: "ticket-check:T-1:r7".to_string(),
+            purpose: "ticket_item_check".to_string(),
+            input_revision: "7".to_string(),
+            input_ref: "ticket://T-1/revisions/7".to_string(),
+            input: serde_json::json!({"title": "Check me"}),
+            instruction: "Check this immutable Ticket snapshot.".to_string(),
+            profile: "builtin:backend-job".to_string(),
+            source_worker: Some(RuntimeWorkerRef::new("runtime-a", "worker-a")),
+            notification_target: Some(RuntimeWorkerRef::new("runtime-a", "worker-a")),
+            limits: BackendJobLimits::default(),
+        }
+    }
+
+    #[test]
+    fn fingerprint_covers_revision_profile_limits_and_notification_provenance() {
+        let request = request();
+        let first = request.fingerprint().unwrap();
+        assert_eq!(first, request.fingerprint().unwrap());
+        for mutate in [
+            |request: &mut BackendJobRequest| request.input_revision = "8".to_string(),
+            |request: &mut BackendJobRequest| request.purpose = "alternate_check".to_string(),
+            |request: &mut BackendJobRequest| request.limits.max_attempts = 3,
+            |request: &mut BackendJobRequest| request.notification_target = None,
+        ] {
+            let mut changed = request.clone();
+            mutate(&mut changed);
+            assert_ne!(first, changed.fingerprint().unwrap());
+        }
+    }
+
+    #[test]
+    fn tracked_runtime_ids_are_fixed_size_for_maximum_job_ids() {
+        let job_id = "j".repeat(256);
+        let attempt_id = attempt_id(&job_id, ABSOLUTE_MAX_ATTEMPTS);
+        let run_id = runtime_run_id(&attempt_id);
+        let notification_id = delivery_id(&job_id, &attempt_id);
+
+        assert!(run_id.len() <= 128, "run id was {} bytes", run_id.len());
+        assert!(
+            notification_id.len() <= 128,
+            "notification id was {} bytes",
+            notification_id.len()
+        );
+        assert_eq!(run_id, runtime_run_id(&attempt_id));
+        assert_eq!(notification_id, delivery_id(&job_id, &attempt_id));
+        assert_ne!(run_id, runtime_run_id(&format!("{attempt_id}-other")));
+        assert_ne!(
+            notification_id,
+            delivery_id(&job_id, &format!("{attempt_id}-other"))
+        );
+    }
+
+    #[test]
+    fn result_is_bounded_before_digesting() {
+        let value = serde_json::json!({"ok": true});
+        let (digest, encoded) = result_digest(&value, 128).unwrap();
+        assert!(digest.starts_with("sha256:"));
+        assert_eq!(encoded, r#"{"ok":true}"#);
+        assert!(result_digest(&serde_json::json!({"body": "too long"}), 4).is_err());
+    }
+
+    #[test]
+    fn failure_detail_is_bounded_by_utf8_bytes_without_splitting_code_points() {
+        let detail = "界".repeat(MAX_JOB_FAILURE_BYTES);
+        let bounded = bounded_failure_detail(&detail);
+        assert!(bounded.len() <= MAX_JOB_FAILURE_BYTES);
+        assert!(detail.starts_with(&bounded));
+    }
+
+    #[test]
+    fn job_limits_reject_unbounded_execution_and_retry() {
+        for limits in [
+            BackendJobLimits {
+                timeout_seconds: 0,
+                ..Default::default()
+            },
+            BackendJobLimits {
+                timeout_seconds: MAX_JOB_TIMEOUT_SECONDS + 1,
+                ..Default::default()
+            },
+            BackendJobLimits {
+                max_attempts: 0,
+                ..Default::default()
+            },
+            BackendJobLimits {
+                max_attempts: ABSOLUTE_MAX_ATTEMPTS + 1,
+                ..Default::default()
+            },
+            BackendJobLimits {
+                max_concurrent_jobs: 0,
+                ..Default::default()
+            },
+        ] {
+            assert!(limits.validate().is_err());
+        }
+    }
+}

crates/workspace-server/src/hosts.rs

@@ -544,6 +544,11 @@ pub enum WorkerSpawnIntent {
     WorkspaceCompanion,
     WorkspaceOrchestrator,
     WorkspaceCoding,
+    BackendJob {
+        job_id: String,
+        attempt_id: String,
+        purpose: String,
+    },
     TicketRole {
         ticket_id: String,
         role: TicketWorkerRole,
@@ -705,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<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub segments: Option<Vec<protocol::Segment>>,
 }
 
@@ -725,11 +732,22 @@ pub struct WorkerCompletionsResult {
     pub diagnostics: Vec<RuntimeDiagnostic>,
 }
 
+#[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")]
+    pub runtime_run_id: Option<String>,
     pub diagnostics: Vec<RuntimeDiagnostic>,
 }
 
@@ -1200,7 +1218,9 @@ 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(
                 "worker_input_pending",
                 HostDiagnosticSeverity::Info,
@@ -2985,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,
@@ -2998,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,
@@ -3014,20 +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(_) => WorkerInputResult {
+            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_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),
+                )
+            }
         }
     }
 
@@ -4991,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) {
@@ -4999,7 +5035,9 @@ 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],
                 };
             }
@@ -5010,9 +5048,14 @@ 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,
+                disposition: WorkerInputDisposition::Accepted,
                 worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()),
+                runtime_run_id: response
+                    .ack
+                    .submission
+                    .map(|submission| submission.submission_request_id),
                 diagnostics: Vec::new(),
             },
             Err(diagnostic) => remote_input_rejected(&self.runtime_id, worker_id, diagnostic),
@@ -5396,11 +5439,14 @@ 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],
     }
 }
@@ -5412,7 +5458,9 @@ 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],
     }
 }
@@ -5890,6 +5938,7 @@ fn worker_spawn_intent_label(intent: &WorkerSpawnIntent) -> &'static str {
         WorkerSpawnIntent::WorkspaceCompanion => "workspace_companion",
         WorkerSpawnIntent::WorkspaceOrchestrator => "workspace_orchestrator",
         WorkerSpawnIntent::WorkspaceCoding => "workspace_coding",
+        WorkerSpawnIntent::BackendJob { .. } => "backend_job",
         WorkerSpawnIntent::TicketRole { role, .. } => match role {
             TicketWorkerRole::Intake => "ticket_intake",
             TicketWorkerRole::Orchestrator => "ticket_orchestrator",
@@ -7088,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 {
@@ -7183,6 +7235,7 @@ mod tests {
                 WorkerInputRequest {
                     kind: WorkerInputKind::User,
                     content: "hello embedded runtime".to_string(),
+                    submission_request_id: None,
                     segments: None,
                 },
             )
@@ -7467,6 +7520,7 @@ mod tests {
                 WorkerInputRequest {
                     kind: WorkerInputKind::User,
                     content: "hello remote".to_string(),
+                    submission_request_id: None,
                     segments: None,
                 },
             )

crates/workspace-server/src/latest_schema.sql

@@ -806,6 +806,79 @@ CREATE TABLE "worker_registry" (
             PRIMARY KEY (workspace_id, worker_id),
             FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
         );
+CREATE TABLE backend_jobs (
+            workspace_id TEXT NOT NULL,
+            job_id TEXT NOT NULL,
+            purpose TEXT NOT NULL,
+            input_revision TEXT NOT NULL,
+            input_ref TEXT NOT NULL,
+            request_json TEXT NOT NULL,
+            intent_fingerprint TEXT NOT NULL,
+            state TEXT NOT NULL CHECK (state IN ('pending', 'completed', 'failed', 'unknown')),
+            current_attempt INTEGER NOT NULL CHECK (current_attempt > 0 AND current_attempt <= 3),
+            result_json TEXT,
+            result_digest TEXT,
+            failure_category TEXT,
+            failure_detail TEXT,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL,
+            completed_at TEXT,
+            PRIMARY KEY (workspace_id, job_id),
+            FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+            CHECK ((result_json IS NULL) = (result_digest IS NULL)),
+            CHECK ((state = 'completed') = (result_json IS NOT NULL))
+        );
+CREATE INDEX backend_jobs_active
+        ON backend_jobs(workspace_id, state, updated_at);
+CREATE TABLE backend_job_attempts (
+            workspace_id TEXT NOT NULL,
+            job_id TEXT NOT NULL,
+            attempt_id TEXT NOT NULL,
+            attempt INTEGER NOT NULL CHECK (attempt > 0 AND attempt <= 3),
+            input_revision TEXT NOT NULL,
+            state TEXT NOT NULL CHECK (state IN ('reserved', 'dispatched', 'completed', 'failed', 'unknown')),
+            runtime_id TEXT,
+            worker_id TEXT,
+            runtime_run_id TEXT,
+            dispatched_at TEXT,
+            deadline_at TEXT NOT NULL,
+            result_json TEXT,
+            result_digest TEXT,
+            failure_category TEXT,
+            failure_detail TEXT,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL,
+            completed_at TEXT,
+            PRIMARY KEY (workspace_id, job_id, attempt_id),
+            UNIQUE (workspace_id, job_id, attempt),
+            UNIQUE (workspace_id, runtime_id, worker_id),
+            FOREIGN KEY (workspace_id, job_id)
+                REFERENCES backend_jobs(workspace_id, job_id) ON DELETE CASCADE,
+            CHECK ((runtime_id IS NULL) = (worker_id IS NULL)),
+            CHECK ((result_json IS NULL) = (result_digest IS NULL)),
+            CHECK ((state = 'completed') = (result_json IS NOT NULL))
+        );
+CREATE INDEX backend_job_attempts_recovery
+        ON backend_job_attempts(workspace_id, state, deadline_at);
+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
+        );
+CREATE INDEX backend_job_deliveries_pending
+        ON backend_job_deliveries(workspace_id, state, updated_at);
 CREATE TABLE worker_registry_observations (
     workspace_id TEXT NOT NULL,
     runtime_id TEXT NOT NULL,

crates/workspace-server/src/lib.rs

@@ -6,6 +6,7 @@
 
 pub mod auth;
 pub mod authority;
+pub mod backend_job;
 pub mod companion;
 pub mod config;
 pub mod config_source;

crates/workspace-server/src/workspace_deletion.rs

@@ -18,6 +18,9 @@ use crate::{Error, Result};
 const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
     "artifacts",
     "audit_events",
+    "backend_job_deliveries",
+    "backend_job_attempts",
+    "backend_jobs",
     "external_workdir_grants",
     "flow_source_revisions",
     "flow_sources",

crates/workspace-server/src/workspace_subscription.rs

@@ -449,6 +449,7 @@ fn project_registry_worker(
             workspace_id: Some(record.registry.workspace_id.clone()),
             display_name: Some(record.registry.display_name.clone()),
             profile: record.registry.profile.clone(),
+            job: None,
             workdir_attachments: Vec::new(),
         }
     };
@@ -457,6 +458,13 @@ fn project_registry_worker(
     worker.workspace_id = Some(record.registry.workspace_id);
     worker.display_name = Some(record.registry.display_name);
     worker.profile = record.registry.profile;
+    worker.job = record
+        .job
+        .map(|job| protocol::subscription::SubscriptionWorkerJob {
+            job_id: job.job_id,
+            attempt_id: job.attempt_id,
+            purpose: job.purpose,
+        });
     if !project_working_directory(api, &record.registry.worker, &mut worker) {
         return None;
     }
@@ -616,6 +624,7 @@ mod tests {
             workspace_id: Some("workspace-1".to_string()),
             display_name: Some("Companion".to_string()),
             profile: Some("builtin:companion".to_string()),
+            job: None,
             workdir_attachments: vec![
                 protocol::subscription::SubscriptionWorkerWorkdirAttachment {
                     alias: "external".to_string(),

docs/design/durable-operations.md

@@ -301,6 +301,68 @@ The public result contains only the Workdir ID,
 failure category. Internal operation identifiers, checkpoints, provider paths,
 and credentials are not public DTO fields.
 
+## Backend-owned bounded Jobs
+
+A Backend Job is durable side-effect intent whose model execution is delegated to
+one ordinary Runtime-managed Worker per attempt. The Workspace Server owns the
+Job; `source_worker` and `notification_target` are provenance and delivery
+addresses, not parent ownership. Stopping or losing either Worker does not revoke
+Job intent. Attempt records retain the bound Worker identity as historical
+evidence without foreign-keying Worker retention to Job retention; normal Worker
+stop, archive, pin/hold, and removal policy remains authoritative.
+
+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, a precommitted tracked Runtime
+  submission request identity, a deadline, and its bounded outcome;
+- `backend_job_deliveries` records best-effort notification separately from the
+  result. Its fixed-size collision-resistant durable delivery ID is also the
+  Runtime's tracked notification request ID. Runtime reports acceptance only
+  after the Worker has persisted the correlated notification receipt, 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 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. Before Runtime
+input can execute, the attempt is fenced by Workspace, Job ID, attempt ID, input
+revision, bound Runtime Worker identity, and a fixed-size collision-resistant
+tracked submission request identity derived from the attempt. 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 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 read-only Console route.
+
 ## Diagnostics and audit
 
 Persist bounded error categories and identifiers needed to investigate or retry.

resources/profiles/backend-job.dcdl

@@ -0,0 +1,33 @@
+import "./base.dcdl" // {
+    slug = "backend-job";
+    description = "Backend-owned bounded Job Worker profile.";
+    scope = "workspace_read";
+    model = { ref = "codex-oauth/gpt-5.6-luna"; };
+    engine = {
+        instruction = "internal.backend_job_system";
+        reasoning = "medium";
+        max_tokens = 8192;
+    };
+
+    feature = {
+        task = { enabled = false; };
+        memory = { enabled = false; };
+        web = { enabled = false; };
+        image = { enabled = false; };
+        sub_worker = { enabled = false; };
+        flow = { enabled = false; };
+        worker = { enabled = false; direct_spawn = false; };
+        workspace_worker_discovery = { enabled = false; };
+        objective = { enabled = false; };
+        manage_workdir = { enabled = false; };
+        ticket = { enabled = false; authoring = false; thread = false; intake = false; };
+        merge_request = {
+            show = false;
+            open = false;
+            review = false;
+            readiness_check = false;
+            complete = false;
+        };
+        orchestration = { enabled = false; };
+    };
+}

resources/prompts/catalog.dcdl

@@ -20,6 +20,7 @@ roleOrchestrator = import "./role/orchestrator.md";
 roleReviewer = import "./role/reviewer.md";
 internalCompactSystem = import "./internal/compact_system.md";
 internalFlowVerifierSystem = import "./internal/flow_verifier_system.md";
+internalBackendJobSystem = import "./internal/backend_job_system.md";
 internalMemoryConsolidationSystem = import "./internal/memory_consolidation_system.md";
 internalMemoryExtractSystem = import "./internal/memory_extract_system.md";
 internalWorkspaceOrchestratorQueueAttention = import "./internal/workspace_orchestrator_queue_attention.md";
@@ -55,6 +56,7 @@ in
     internal = {
         compact_system = internalCompactSystem.content;
         flow_verifier_system = internalFlowVerifierSystem.content;
+        backend_job_system = internalBackendJobSystem.content;
         memory_consolidation_system = internalMemoryConsolidationSystem.content;
         memory_extract_system = internalMemoryExtractSystem.content;
         workspace_orchestrator_queue_attention = internalWorkspaceOrchestratorQueueAttention.content;

resources/prompts/internal/backend_job_system.md

@@ -0,0 +1,5 @@
+You are a Backend-owned bounded Job Worker. Execute only the immutable Job envelope in the first user message.
+
+You are not a child of the Worker recorded as provenance or notification target. Do not create or mutate Tickets, Objectives, Workers, Workdirs, Merge Requests, Memory, or other Workspace state. Do not recursively create Jobs.
+
+A normal assistant response, Worker Idle state, or Worker stop is not Job success. Call `SubmitBackendJobResult` exactly once with the bounded structured result. If that capability rejects the result, explain the failure in final prose without claiming success.

web/workspace/src/lib/generated/protocol.ts

@@ -279,6 +279,8 @@ export type SubscriptionWorkerWorkdirAttachment = {
  */
 alias: string, repository_key?: string | null, working_directory_id: SubscriptionWorkdirId, };
 
+export type SubscriptionWorkerJob = { job_id: string, attempt_id: string, purpose: string, };
+
 export type SubscriptionWorker = { worker_id: SubscriptionWorkerId,
 /**
  * Set by the Workspace Server when projecting a Runtime-owned Worker to clients.
@@ -307,7 +309,12 @@ worker_state?: WorkerStateSnapshot | null,
 /**
  * Runtime catalog lifecycle compatibility projection; not foreground-state authority.
  */
-state: SubscriptionWorkerState, has_running_internal_workers: boolean, workspace_id?: string | null, display_name?: string | null, profile?: string | null, workdir_attachments?: Array<SubscriptionWorkerWorkdirAttachment>, };
+state: SubscriptionWorkerState, has_running_internal_workers: boolean, workspace_id?: string | null, display_name?: string | null, profile?: string | null,
+/**
+ * Backend-owned Job binding. Runtime producers leave this unset; the
+ * Workspace projection enriches both snapshots and live updates.
+ */
+job?: SubscriptionWorkerJob | null, workdir_attachments?: Array<SubscriptionWorkerWorkdirAttachment>, };
 
 export type WorkspaceSubscriptionWorkdir = { working_directory_id: SubscriptionWorkdirId, repository_key: string, state: string, };
 

web/workspace/src/lib/workspace/sidebar/types.ts

@@ -30,6 +30,7 @@ import type {
   Event as PodProtocolEvent,
   Method as PodProtocolMethod,
   Segment as PodProtocolSegment,
+  SubscriptionWorkerJob,
   WorkerStateSnapshot,
 } from "$lib/generated/protocol";
 import type {
@@ -88,6 +89,7 @@ export type Worker =
     display_name: string;
     tags: string[];
     worker_state?: WorkerStateSnapshot | null;
+    job?: SubscriptionWorkerJob | null;
     diagnostics: Diagnostic[];
   };
 

web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts

@@ -88,6 +88,47 @@ Deno.test('workspace Worker snapshot keeps equal local ids from different Runtim
   assertEquals([...projection.workers.keys()].sort(), ['runtime-a:1', 'runtime-b:1']);
 });
 
+Deno.test('workspace Worker snapshot and live update preserve canonical Job metadata', () => {
+  const projection = createWorkspaceWorkersProjection();
+  const snapshotWorker = worker('runtime-a', 'job-worker', 1);
+  snapshotWorker.job = {
+    job_id: 'check:T-1:r1',
+    attempt_id: 'check:T-1:r1:attempt:1',
+    purpose: 'ticket_item_check',
+  };
+  applyWorkspaceWorkersFrame(projection, {
+    protocol_version: 1,
+    frame: 'response',
+    message: {
+      result: 'subscribed',
+      payload: {
+        request_id: 'request-1',
+        subscription_id: 'subscription-1',
+        selector: { topic: 'workspace_workers' },
+        snapshot_revision: 1,
+        snapshot: { topic: 'workers', data: { workers: [snapshotWorker] } },
+      },
+    },
+  });
+  assertEquals(projection.workers.get('runtime-a:job-worker')?.job, snapshotWorker.job);
+
+  const updated = { ...snapshotWorker, subject_revision: 2 };
+  updated.job = { ...snapshotWorker.job, attempt_id: 'check:T-1:r1:attempt:2' };
+  applyWorkspaceWorkersFrame(projection, {
+    protocol_version: 1,
+    frame: 'event',
+    message: {
+      event: 'event',
+      data: {
+        subscription_id: 'subscription-1',
+        subject_revision: 2,
+        payload: { event: 'worker_upserted', data: { worker: updated } },
+      },
+    },
+  });
+  assertEquals(projection.workers.get('runtime-a:job-worker')?.job, updated.job);
+});
+
 Deno.test('workspace Worker reducer ignores stale events and removes composite subject', () => {
   const projection = createWorkspaceWorkersProjection();
   projection.workers.set('runtime-a:1', worker('runtime-a', '1', 3));

web/workspace/src/lib/workspace/sidebar/worker-subscription.ts

@@ -143,6 +143,7 @@ function projectWorker(
     display_name: displayName,
     label: displayName,
     profile: worker.profile ?? null,
+    job: worker.job ?? null,
     tags: [],
     workspace: {
       visibility: "workspace",

web/workspace/src/lib/workspace/sidebar/workers.test.ts

@@ -63,6 +63,19 @@ Deno.test("live runtime workers are sidebar targets and console targets", () =>
   assertEquals(canOpenWorkerConsole(liveWorker), true);
 });
 
+Deno.test("Backend Job Workers are hidden only from the sidebar View", () => {
+  const jobWorker = worker({
+    state: "running",
+    job: {
+      job_id: "check:T-1:r1",
+      attempt_id: "check:T-1:r1:attempt:1",
+      purpose: "ticket_item_check",
+    },
+  });
+  assertEquals(canShowWorkerInSidebar(jobWorker), false);
+  assertEquals(canOpenWorkerConsole(jobWorker), true);
+});
+
 Deno.test("sidebar workers sort running then idle then stopped", () => {
   const workers = [
     worker({ worker_id: "3", display_name: "Stopped", state: "stopped" }),

web/workspace/src/lib/workspace/sidebar/workers.ts

@@ -22,15 +22,15 @@ export function sidebarWorkerActivity(
 }
 
 export function canShowWorkerInSidebar(
-  worker: Pick<Worker, 'implementation'>,
+  worker: Pick<Worker, 'implementation' | 'job'>,
 ): boolean {
-  return worker.implementation.kind !== 'backend_worker_registry';
+  return !worker.job && worker.implementation.kind !== 'backend_worker_registry';
 }
 
 export function canOpenWorkerConsole(
   worker: Pick<Worker, 'implementation'>,
 ): boolean {
-  return canShowWorkerInSidebar(worker);
+  return worker.implementation.kind !== 'backend_worker_registry';
 }
 
 type SidebarWorkerLink = Pick<Worker, 'resource_key' | 'display_name'>;