feat: add durable backend job runner

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/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/worker-runtime/src/runtime.rs

@@ -4180,6 +4180,7 @@ impl RuntimeState {
             workspace_id: worker.workspace_id.clone(),
             display_name: worker.display_name.clone(),
             profile,
+            job: None,
             workdir_attachments,
         })
     }

crates/worker/src/controller.rs

@@ -1225,6 +1225,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 worker.manifest().engine.instruction == "internal.backend_job_system" {
+        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(),

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/workspace-server/src/backend_job.rs

@@ -0,0 +1,489 @@
+//! 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,
+    Completed,
+    Failed,
+}
+
+impl BackendJobDeliveryState {
+    pub(crate) fn as_str(self) -> &'static str {
+        match self {
+            Self::Pending => "pending",
+            Self::Completed => "completed",
+            Self::Failed => "failed",
+        }
+    }
+
+    pub(crate) fn parse(value: &str) -> Result<Self> {
+        match value {
+            "pending" => Ok(Self::Pending),
+            "completed" => Ok(Self::Completed),
+            "failed" => Ok(Self::Failed),
+            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}")
+}
+
+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 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,
@@ -5890,6 +5895,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",

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', 'completed', 'failed')),
+            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/server.rs

@@ -13,9 +13,7 @@ use axum::http::header::{CONTENT_TYPE, LOCATION, ORIGIN};
 use axum::http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri};
 use axum::middleware::{self, Next};
 use axum::response::{IntoResponse, Response};
-use axum::routing::get;
+use axum::routing::{get, post};
-#[cfg(test)]
-use axum::routing::post;
 use axum::{Json, Router};
 use chrono::{Duration, SecondsFormat, Utc};
 use config_source::ConfigTreeSnapshot;
@@ -145,6 +143,11 @@ use crate::authority::{
     MemoryAuthority, ObjectiveAuthority, ObjectiveCreateInput, ObjectiveEditInput,
     SqliteWorkspaceAuthority, TicketAuthority, TicketMergeRevisionSource, merge_request_summary,
 };
+use crate::backend_job::{
+    ABSOLUTE_MAX_CONCURRENT_JOBS, BackendJobAttemptState, BackendJobDeliveryState,
+    BackendJobRequest, BackendJobReservation, BackendJobResultAcceptance,
+    BackendJobResultSubmission, BackendJobState, allocation_key,
+};
 use crate::companion::{
     CompanionCancelRequest, CompanionConsole, CompanionMessageRequest, CompanionMessageResponse,
     CompanionStatusResponse, CompanionTranscriptProjection,
@@ -3241,6 +3244,7 @@ impl WorkspaceApi {
         }
         recover_workdir_removals(&api)?;
         recover_runtime_removals(&api).await;
+        api.recover_backend_jobs()?;
         Ok(api)
     }
 
@@ -3248,6 +3252,387 @@ impl WorkspaceApi {
         self.config.workspace_id.as_str()
     }
 
+    /// Reserve durable Job intent and dispatch its current attempt through the
+    /// ordinary embedded Runtime Worker path. Replaying the same request
+    /// converges on the same attempt and Worker allocation.
+    pub fn dispatch_backend_job(
+        &self,
+        request: &BackendJobRequest,
+    ) -> Result<BackendJobReservation> {
+        let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
+        let reservation =
+            self.store
+                .reserve_backend_job(&self.config.workspace_id, request, &now)?;
+        self.dispatch_backend_job_reservation(reservation)
+    }
+
+    /// Explicit re-evaluation creates a new bounded attempt. Transport replay
+    /// must call `dispatch_backend_job` with the original immutable request.
+    pub fn retry_backend_job(&self, job_id: &str) -> Result<BackendJobReservation> {
+        let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
+        let reservation =
+            self.store
+                .reserve_backend_job_retry(&self.config.workspace_id, job_id, &now)?;
+        self.dispatch_backend_job_reservation(reservation)
+    }
+
+    fn dispatch_backend_job_reservation(
+        &self,
+        reservation: BackendJobReservation,
+    ) -> Result<BackendJobReservation> {
+        if reservation.job.state != BackendJobState::Pending
+            || reservation.attempt.state != BackendJobAttemptState::Reserved
+            || reservation.attempt.worker.is_some()
+        {
+            return Ok(reservation);
+        }
+        let request = &reservation.job.request;
+        let worker_input = request.worker_input(&reservation.attempt.attempt_id)?;
+        let control_operation = WorkerControlOperation {
+            operation_id: allocation_key(&reservation.attempt.attempt_id),
+            input_fingerprint: reservation.job.intent_fingerprint.clone(),
+        };
+        let spawned = self
+            .spawn_workspace_worker(
+                EMBEDDED_WORKER_RUNTIME_ID,
+                WorkerSpawnRequest {
+                    requested_worker_name: Some(format!("job:{}", request.purpose)),
+                    intent: WorkerSpawnIntent::BackendJob {
+                        job_id: request.job_id.clone(),
+                        attempt_id: reservation.attempt.attempt_id.clone(),
+                        purpose: request.purpose.clone(),
+                    },
+                    acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
+                        expected_segments: 0,
+                    },
+                    profile: ProfileSelector::Builtin(request.profile.clone()),
+                    ticket_assignment: None,
+                    initial_submit: Vec::new(),
+                    workdir_attachment_requests: Vec::new(),
+                    resolved_workdir_attachment_requests: Vec::new(),
+                    resolved_workdir_attachments: Vec::new(),
+                    resolved_config_bundle: None,
+                    resolved_worker_observation_enabled: false,
+                    resolved_worker_observation_grants: Vec::new(),
+                    resolved_workspace_api: None,
+                    resolved_memory_settings: None,
+                    resolved_control_operation: Some(control_operation),
+                },
+            )
+            .map_err(|error| error.error)?;
+        let worker = match (spawned.state, spawned.worker) {
+            (InternalWorkerOperationState::Accepted, Some(worker)) => worker,
+            _ => {
+                let detail = spawned
+                    .diagnostics
+                    .first()
+                    .map(|diagnostic| diagnostic.message.as_str())
+                    .unwrap_or("embedded Runtime rejected Backend Job Worker dispatch");
+                self.store.finish_backend_job_attempt(
+                    &self.config.workspace_id,
+                    &request.job_id,
+                    &reservation.attempt.attempt_id,
+                    BackendJobAttemptState::Failed,
+                    "dispatch_rejected",
+                    detail,
+                    &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+                )?;
+                return Err(Error::RuntimeOperationFailed {
+                    runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
+                    code: "backend_job_dispatch_rejected".to_string(),
+                    message: detail.to_string(),
+                });
+            }
+        };
+        let bound_at = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
+        let (commit, newly_bound) = self.store.bind_backend_job_attempt_worker(
+            &self.config.workspace_id,
+            &request.job_id,
+            &reservation.attempt.attempt_id,
+            &worker.worker,
+            None,
+            &bound_at,
+        )?;
+        if newly_bound {
+            self.worker_projection.publish_commit(commit)?;
+        } else {
+            let job = self
+                .store
+                .get_backend_job(&self.config.workspace_id, &request.job_id)?
+                .ok_or_else(|| Error::Store("bound Backend Job disappeared".to_string()))?;
+            let attempt = self
+                .store
+                .get_backend_job_attempt(
+                    &self.config.workspace_id,
+                    &request.job_id,
+                    &reservation.attempt.attempt_id,
+                )?
+                .ok_or_else(|| Error::Store("bound Backend Job attempt disappeared".to_string()))?;
+            return Ok(BackendJobReservation {
+                job,
+                attempt,
+                replayed: true,
+            });
+        }
+
+        let submitted = self
+            .runtime
+            .send_input(
+                &worker.worker,
+                WorkerInputRequest {
+                    kind: WorkerInputKind::User,
+                    content: worker_input,
+                    segments: None,
+                },
+            )
+            .map_err(RuntimeRegistryError::into_error)?;
+        if submitted.state != InternalWorkerOperationState::Accepted {
+            let detail = submitted
+                .diagnostics
+                .first()
+                .map(|diagnostic| diagnostic.message.as_str())
+                .unwrap_or("embedded Runtime rejected Backend Job input");
+            self.store.finish_backend_job_attempt(
+                &self.config.workspace_id,
+                &request.job_id,
+                &reservation.attempt.attempt_id,
+                BackendJobAttemptState::Failed,
+                "input_rejected",
+                detail,
+                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+            )?;
+            return Err(Error::RuntimeOperationFailed {
+                runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
+                code: "backend_job_input_rejected".to_string(),
+                message: detail.to_string(),
+            });
+        }
+        self.schedule_backend_job_timeout(
+            &request.job_id,
+            &reservation.attempt.attempt_id,
+            &reservation.attempt.deadline_at,
+        );
+        let job = self
+            .store
+            .get_backend_job(&self.config.workspace_id, &request.job_id)?
+            .ok_or_else(|| Error::Store("dispatched Backend Job disappeared".to_string()))?;
+        let attempt = self
+            .store
+            .get_backend_job_attempt(
+                &self.config.workspace_id,
+                &request.job_id,
+                &reservation.attempt.attempt_id,
+            )?
+            .ok_or_else(|| {
+                Error::Store("dispatched Backend Job attempt disappeared".to_string())
+            })?;
+        Ok(BackendJobReservation {
+            job,
+            attempt,
+            replayed: reservation.replayed,
+        })
+    }
+
+    /// Reconcile durable non-terminal attempts after Backend restart. Reserved
+    /// attempts are replay-dispatched through their deterministic allocation;
+    /// dispatched attempts keep running only while their bound Worker remains
+    /// known, and always receive a fresh deadline task.
+    pub fn recover_backend_jobs(&self) -> Result<()> {
+        let attempts = self.store.list_recoverable_backend_job_attempts(
+            &self.config.workspace_id,
+            usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS),
+        )?;
+        for attempt in attempts {
+            match attempt.state {
+                BackendJobAttemptState::Reserved => {
+                    let job = self
+                        .store
+                        .get_backend_job(&self.config.workspace_id, &attempt.job_id)?
+                        .ok_or_else(|| {
+                            Error::Store("recoverable Backend Job disappeared".to_string())
+                        })?;
+                    if let Err(error) =
+                        self.dispatch_backend_job_reservation(BackendJobReservation {
+                            job,
+                            attempt,
+                            replayed: true,
+                        })
+                    {
+                        tracing::warn!(%error, "Backend Job reserved-attempt recovery failed");
+                    }
+                }
+                BackendJobAttemptState::Dispatched => {
+                    let worker_known = attempt
+                        .worker
+                        .as_ref()
+                        .is_some_and(|worker| self.runtime.worker(worker).is_ok());
+                    if !worker_known {
+                        self.store.finish_backend_job_attempt(
+                            &self.config.workspace_id,
+                            &attempt.job_id,
+                            &attempt.attempt_id,
+                            BackendJobAttemptState::Unknown,
+                            "recovery_worker_unknown",
+                            "the bound Runtime Worker was not recoverable after Backend restart",
+                            &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+                        )?;
+                    } else {
+                        self.schedule_backend_job_timeout(
+                            &attempt.job_id,
+                            &attempt.attempt_id,
+                            &attempt.deadline_at,
+                        );
+                    }
+                }
+                BackendJobAttemptState::Completed
+                | BackendJobAttemptState::Failed
+                | BackendJobAttemptState::Unknown => {}
+            }
+        }
+        Ok(())
+    }
+
+    fn schedule_backend_job_timeout(&self, job_id: &str, attempt_id: &str, deadline_at: &str) {
+        let Ok(deadline) = chrono::DateTime::parse_from_rfc3339(deadline_at) else {
+            return;
+        };
+        let delay = (deadline.with_timezone(&Utc) - Utc::now())
+            .to_std()
+            .unwrap_or_default();
+        let Ok(handle) = tokio::runtime::Handle::try_current() else {
+            return;
+        };
+        let store = self.store.clone();
+        let runtime = self.runtime.clone();
+        let workspace_id = self.config.workspace_id.clone();
+        let job_id = job_id.to_string();
+        let attempt_id = attempt_id.to_string();
+        handle.spawn(async move {
+            tokio::time::sleep(delay).await;
+            let Ok(Some(attempt)) =
+                store.get_backend_job_attempt(&workspace_id, &job_id, &attempt_id)
+            else {
+                return;
+            };
+            if attempt.state != BackendJobAttemptState::Dispatched {
+                return;
+            }
+            // Fence result acceptance before stopping the Worker so a late
+            // completion cannot turn a timed-out attempt into success.
+            if store
+                .finish_backend_job_attempt(
+                    &workspace_id,
+                    &job_id,
+                    &attempt_id,
+                    BackendJobAttemptState::Failed,
+                    "timeout",
+                    "Backend Job attempt exceeded its durable deadline",
+                    &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+                )
+                .is_ok()
+            {
+                if let Some(worker) = attempt.worker {
+                    let _ = runtime.stop_worker(
+                        &worker,
+                        WorkerLifecycleRequest {
+                            reason: Some("Backend Job attempt timed out".to_string()),
+                            ticket_assignment: None,
+                        },
+                    );
+                }
+            }
+        });
+    }
+
+    /// Accept the only authoritative success signal for a Job attempt. The
+    /// source identity is fenced against the durable Worker binding. Result
+    /// persistence precedes best-effort notification delivery.
+    pub fn accept_backend_job_result(
+        &self,
+        source_worker: &RuntimeWorkerRef,
+        submission: &BackendJobResultSubmission,
+    ) -> Result<BackendJobResultAcceptance> {
+        let accepted_at = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
+        let acceptance = self.store.accept_backend_job_result(
+            &self.config.workspace_id,
+            source_worker,
+            submission,
+            &accepted_at,
+        )?;
+        if let Some(target) = acceptance.job.request.notification_target.as_ref() {
+            self.deliver_backend_job_result(&acceptance, target);
+        }
+        Ok(acceptance)
+    }
+
+    fn deliver_backend_job_result(
+        &self,
+        acceptance: &BackendJobResultAcceptance,
+        target: &RuntimeWorkerRef,
+    ) {
+        let delivery_id = format!(
+            "{}:{}:notification",
+            acceptance.job.request.job_id, acceptance.attempt.attempt_id
+        );
+        let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
+        let Ok(delivery) = self.store.reserve_backend_job_delivery(
+            &self.config.workspace_id,
+            &delivery_id,
+            &acceptance.job.request.job_id,
+            target,
+            &now,
+        ) else {
+            return;
+        };
+        if delivery.state == BackendJobDeliveryState::Completed {
+            return;
+        }
+        let content = format!(
+            "Backend Job `{}` completed attempt `{}` for `{}`. Durable result digest: {}.",
+            acceptance.job.request.job_id,
+            acceptance.attempt.attempt_id,
+            acceptance.job.request.purpose,
+            acceptance
+                .job
+                .result_digest
+                .as_deref()
+                .unwrap_or("unavailable")
+        );
+        let delivery_result = self.runtime.send_input(
+            target,
+            WorkerInputRequest {
+                kind: WorkerInputKind::Notify,
+                content,
+                segments: None,
+            },
+        );
+        let (state, category, detail): (_, Option<String>, Option<String>) = match delivery_result {
+            Ok(result) if result.state == InternalWorkerOperationState::Accepted => {
+                (BackendJobDeliveryState::Completed, None, None)
+            }
+            Ok(result) => (
+                BackendJobDeliveryState::Failed,
+                Some("notification_rejected".to_string()),
+                result.diagnostics.first().map(|item| item.message.clone()),
+            ),
+            Err(error) => (
+                BackendJobDeliveryState::Failed,
+                Some("notification_unavailable".to_string()),
+                Some(error.message()),
+            ),
+        };
+        // Delivery is intentionally separate from result authority. A delivery
+        // bookkeeping failure must never roll back or re-run completed work.
+        let _ = self.store.finish_backend_job_delivery(
+            &self.config.workspace_id,
+            &delivery_id,
+            state,
+            category.as_deref(),
+            detail.as_deref(),
+            &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+        );
+    }
+
     fn register_workspace_runtime_binding(
         &self,
         binding: WorkspaceRuntimeBinding,
@@ -9123,6 +9508,10 @@ fn build_inner_router(api: WorkspaceApi) -> Router {
         _,
         _,
     >(workspace, scoped_worker_protocol_ws)
+    .route(
+        "/api/w/{workspace_id}/workers/self/backend-job-result",
+        post(scoped_submit_backend_job_result),
+    )
     .fallback(get(static_or_spa_fallback))
     .with_state(api)
     .merge(generated_workspace_contract_router(contract_service));
@@ -14474,6 +14863,49 @@ async fn refresh_current_worker_session_locked(
     Ok(())
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SubmitBackendJobResultRequest {
+    result: serde_json::Value,
+}
+
+async fn scoped_submit_backend_job_result(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedWorkspacePath>,
+    headers: HeaderMap,
+    Json(request): Json<SubmitBackendJobResultRequest>,
+) -> ApiResult<Json<BackendJobResultAcceptance>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let worker = current_worker_identity(&api, &path.workspace_id, &headers)?;
+    let binding = api
+        .store
+        .worker_registry_projection(&api.config.workspace_id, &worker)?
+        .and_then(|projection| projection.job)
+        .ok_or_else(|| {
+            Error::InvalidInput(
+                "current Worker is not bound to an active Backend Job attempt".to_string(),
+            )
+        })?;
+    let attempt = api
+        .store
+        .get_backend_job_attempt(
+            &api.config.workspace_id,
+            &binding.job_id,
+            &binding.attempt_id,
+        )?
+        .ok_or_else(|| Error::Store("Backend Job attempt binding disappeared".to_string()))?;
+    let acceptance = api.accept_backend_job_result(
+        &worker,
+        &BackendJobResultSubmission {
+            job_id: binding.job_id,
+            attempt_id: binding.attempt_id,
+            input_revision: attempt.input_revision,
+            result: request.result,
+        },
+    )?;
+    Ok(Json(acceptance))
+}
+
 async fn scoped_attach_current_worker_workdir(
     State(api): State<WorkspaceApi>,
     AxumPath(path): AxumPath<ScopedWorkspacePath>,
@@ -24800,6 +25232,7 @@ fn record_worker_summary(
             workspace_id: worker.workspace.workspace_id.clone(),
             display_name: Some(worker.display_name.clone()),
             profile: worker.profile.clone(),
+            job: None,
             workdir_attachments: Vec::new(),
         };
         api.worker_projection
@@ -26714,6 +27147,7 @@ mod tests {
                     workspace_id: Some(TEST_WORKSPACE_ID.to_string()),
                     display_name: Some("Worker A".to_string()),
                     profile: Some("builtin:coder".to_string()),
+                    job: None,
                     workdir_attachments: Vec::new(),
                 },
                 availability,
@@ -26723,6 +27157,7 @@ mod tests {
                 projection_revision: 9,
                 observed_at: "2026-09-24T00:00:01Z".to_string(),
             }),
+            job: None,
         };
 
         let unavailable = worker_summary_from_projection(&projection).unwrap();
@@ -30275,6 +30710,7 @@ mod tests {
                     workspace_id: Some(TEST_WORKSPACE_ID.to_string()),
                     display_name: Some("Paused Worker".to_string()),
                     profile: Some("builtin:coder".to_string()),
+                    job: None,
                     workdir_attachments: Vec::new(),
                 },
                 "4",
@@ -33403,6 +33839,297 @@ mod tests {
         config
     }
 
+    #[tokio::test]
+    async fn backend_job_runner_uses_registered_embedded_worker_and_structured_result_authority() {
+        let temp = tempfile::tempdir().unwrap();
+        let (api, execution) = test_api_with_recording_backend(temp.path()).await;
+        let request = BackendJobRequest {
+            job_id: "runner-check-1".to_string(),
+            purpose: "runner_contract_check".to_string(),
+            input_revision: "revision-1".to_string(),
+            input_ref: "test://runner/input/1".to_string(),
+            input: serde_json::json!({"value": "immutable"}),
+            instruction: "Return a structured check result.".to_string(),
+            profile: "builtin:backend-job".to_string(),
+            source_worker: None,
+            notification_target: None,
+            limits: crate::backend_job::BackendJobLimits::default(),
+        };
+
+        let stale_reservation = api
+            .store
+            .reserve_backend_job(
+                TEST_WORKSPACE_ID,
+                &request,
+                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+            )
+            .unwrap();
+        let dispatched = api
+            .dispatch_backend_job_reservation(stale_reservation.clone())
+            .unwrap();
+        let concurrent_replay = api
+            .dispatch_backend_job_reservation(stale_reservation)
+            .unwrap();
+        assert!(concurrent_replay.replayed);
+        assert_eq!(
+            execution.take_inputs().len(),
+            1,
+            "a stale concurrent dispatch replay must not submit the attempt twice"
+        );
+        assert_eq!(dispatched.attempt.state, BackendJobAttemptState::Dispatched);
+        let worker = dispatched.attempt.worker.clone().expect("bound Job Worker");
+        assert_eq!(worker.runtime_id, EMBEDDED_WORKER_RUNTIME_ID);
+        assert!(
+            api.store
+                .get_worker_registry(TEST_WORKSPACE_ID, &worker)
+                .unwrap()
+                .is_some(),
+            "Job dispatch must use the ordinary Workspace Worker registry"
+        );
+        assert_eq!(
+            api.store
+                .worker_registry_projection(TEST_WORKSPACE_ID, &worker)
+                .unwrap()
+                .unwrap()
+                .job,
+            Some(crate::backend_job::BackendJobWorkerBinding {
+                job_id: request.job_id.clone(),
+                attempt_id: dispatched.attempt.attempt_id.clone(),
+                purpose: request.purpose.clone(),
+            })
+        );
+        let replay = api.dispatch_backend_job(&request).unwrap();
+        assert!(replay.replayed);
+        assert_eq!(replay.attempt.worker, Some(worker.clone()));
+
+        let accepted = api
+            .accept_backend_job_result(
+                &worker,
+                &BackendJobResultSubmission {
+                    job_id: request.job_id.clone(),
+                    attempt_id: dispatched.attempt.attempt_id,
+                    input_revision: request.input_revision,
+                    result: serde_json::json!({"valid": true}),
+                },
+            )
+            .unwrap();
+        assert_eq!(accepted.job.state, BackendJobState::Completed);
+    }
+
+    #[tokio::test]
+    async fn backend_job_input_failure_is_durable_and_retryable_without_reusing_attempt() {
+        let temp = tempfile::tempdir().unwrap();
+        let (api, execution) = test_api_with_recording_backend(temp.path()).await;
+        execution.reject_inputs("model transport unavailable");
+        let request = BackendJobRequest {
+            job_id: "runner-input-failure-1".to_string(),
+            purpose: "failure_contract_check".to_string(),
+            input_revision: "revision-1".to_string(),
+            input_ref: "test://runner/failure/1".to_string(),
+            input: serde_json::json!({"value": "immutable"}),
+            instruction: "Return a structured check result.".to_string(),
+            profile: "builtin:backend-job".to_string(),
+            source_worker: None,
+            notification_target: None,
+            limits: crate::backend_job::BackendJobLimits::default(),
+        };
+
+        assert!(api.dispatch_backend_job(&request).is_err());
+        let failed = api
+            .store
+            .get_backend_job_attempt(
+                TEST_WORKSPACE_ID,
+                &request.job_id,
+                &crate::backend_job::attempt_id(&request.job_id, 1),
+            )
+            .unwrap()
+            .unwrap();
+        assert_eq!(failed.state, BackendJobAttemptState::Failed);
+        assert_eq!(failed.failure_category.as_deref(), Some("input_rejected"));
+        let _retry_error = api.retry_backend_job(&request.job_id).unwrap_err();
+        let second = api
+            .store
+            .get_backend_job_attempt(
+                TEST_WORKSPACE_ID,
+                &request.job_id,
+                &crate::backend_job::attempt_id(&request.job_id, 2),
+            )
+            .unwrap()
+            .unwrap();
+        assert_eq!(second.attempt, 2);
+        assert_eq!(second.state, BackendJobAttemptState::Failed);
+    }
+
+    #[tokio::test]
+    async fn backend_job_recovery_dispatches_reserved_attempt_without_duplicate_worker() {
+        let temp = tempfile::tempdir().unwrap();
+        let api = test_api(temp.path()).await;
+        let request = BackendJobRequest {
+            job_id: "runner-restart-1".to_string(),
+            purpose: "restart_contract_check".to_string(),
+            input_revision: "revision-1".to_string(),
+            input_ref: "test://runner/restart/1".to_string(),
+            input: serde_json::json!({"value": "immutable"}),
+            instruction: "Return a structured check result.".to_string(),
+            profile: "builtin:backend-job".to_string(),
+            source_worker: None,
+            notification_target: None,
+            limits: crate::backend_job::BackendJobLimits::default(),
+        };
+        let reserved = api
+            .store
+            .reserve_backend_job(
+                TEST_WORKSPACE_ID,
+                &request,
+                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+            )
+            .unwrap();
+        assert_eq!(reserved.attempt.state, BackendJobAttemptState::Reserved);
+        api.recover_backend_jobs().unwrap();
+
+        let recovered = api
+            .store
+            .get_backend_job_attempt(
+                TEST_WORKSPACE_ID,
+                &request.job_id,
+                &reserved.attempt.attempt_id,
+            )
+            .unwrap()
+            .unwrap();
+        assert_eq!(recovered.state, BackendJobAttemptState::Dispatched);
+        let recovered_worker = recovered.worker.clone().expect("recovered Job Worker");
+        let replay = api.dispatch_backend_job(&request).unwrap();
+        assert!(replay.replayed);
+        assert_eq!(replay.attempt.worker, Some(recovered_worker));
+    }
+
+    #[tokio::test]
+    async fn backend_job_recovery_marks_unreconcilable_worker_execution_unknown() {
+        let temp = tempfile::tempdir().unwrap();
+        let api = test_api(temp.path()).await;
+        let request = BackendJobRequest {
+            job_id: "runner-unknown-1".to_string(),
+            purpose: "unknown_contract_check".to_string(),
+            input_revision: "revision-1".to_string(),
+            input_ref: "test://runner/unknown/1".to_string(),
+            input: serde_json::json!({"value": "immutable"}),
+            instruction: "Return a structured check result.".to_string(),
+            profile: "builtin:backend-job".to_string(),
+            source_worker: None,
+            notification_target: None,
+            limits: crate::backend_job::BackendJobLimits::default(),
+        };
+        let reserved = api
+            .store
+            .reserve_backend_job(
+                TEST_WORKSPACE_ID,
+                &request,
+                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+            )
+            .unwrap();
+        let missing_worker = RuntimeWorkerRef::new(
+            EMBEDDED_WORKER_RUNTIME_ID,
+            "0199a3ae-ecdf-7eb2-883e-a5b7b5f04a0d",
+        );
+        api.store
+            .upsert_worker_registry(&WorkerRegistryRecord {
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+                worker: missing_worker.clone(),
+                display_name: "Missing Job Worker".to_string(),
+                profile: Some("builtin:backend-job".to_string()),
+                retention_state: "normal".to_string(),
+                transcript_ref: None,
+                session_ref: None,
+                summary_ref: None,
+                diagnostics_ref: None,
+                created_at: "2026-09-01T00:00:00Z".to_string(),
+                updated_at: "2026-09-01T00:00:00Z".to_string(),
+            })
+            .unwrap();
+        api.store
+            .bind_backend_job_attempt_worker(
+                TEST_WORKSPACE_ID,
+                &request.job_id,
+                &reserved.attempt.attempt_id,
+                &missing_worker,
+                Some("missing-runtime-run"),
+                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+            )
+            .unwrap();
+
+        api.recover_backend_jobs().unwrap();
+        let attempt = api
+            .store
+            .get_backend_job_attempt(
+                TEST_WORKSPACE_ID,
+                &request.job_id,
+                &reserved.attempt.attempt_id,
+            )
+            .unwrap()
+            .unwrap();
+        assert_eq!(attempt.state, BackendJobAttemptState::Unknown);
+        assert_eq!(
+            attempt.failure_category.as_deref(),
+            Some("recovery_worker_unknown")
+        );
+        assert_eq!(
+            api.store
+                .get_backend_job(TEST_WORKSPACE_ID, &request.job_id)
+                .unwrap()
+                .unwrap()
+                .state,
+            BackendJobState::Unknown
+        );
+    }
+
+    #[tokio::test]
+    async fn backend_job_timeout_fences_late_structured_result() {
+        let temp = tempfile::tempdir().unwrap();
+        let api = test_api(temp.path()).await;
+        let request = BackendJobRequest {
+            job_id: "runner-timeout-1".to_string(),
+            purpose: "timeout_contract_check".to_string(),
+            input_revision: "revision-1".to_string(),
+            input_ref: "test://runner/timeout/1".to_string(),
+            input: serde_json::json!({"value": "immutable"}),
+            instruction: "Wait for a structured result.".to_string(),
+            profile: "builtin:backend-job".to_string(),
+            source_worker: None,
+            notification_target: None,
+            limits: crate::backend_job::BackendJobLimits {
+                timeout_seconds: 1,
+                max_attempts: 1,
+                ..Default::default()
+            },
+        };
+        let dispatched = api.dispatch_backend_job(&request).unwrap();
+        let worker = dispatched.attempt.worker.clone().unwrap();
+        tokio::time::sleep(std::time::Duration::from_millis(1_200)).await;
+        let attempt = api
+            .store
+            .get_backend_job_attempt(
+                TEST_WORKSPACE_ID,
+                &request.job_id,
+                &dispatched.attempt.attempt_id,
+            )
+            .unwrap()
+            .unwrap();
+        assert_eq!(attempt.state, BackendJobAttemptState::Failed);
+        assert_eq!(attempt.failure_category.as_deref(), Some("timeout"));
+        assert!(
+            api.accept_backend_job_result(
+                &worker,
+                &BackendJobResultSubmission {
+                    job_id: request.job_id,
+                    attempt_id: dispatched.attempt.attempt_id,
+                    input_revision: request.input_revision,
+                    result: serde_json::json!({"late": true}),
+                },
+            )
+            .is_err()
+        );
+    }
+
     #[tokio::test]
     async fn scoped_runtime_workdir_access_requires_workspace_binding() {
         let dir = tempfile::tempdir().unwrap();

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,55 @@ 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, optional Runtime run evidence, a
+  deadline, and its bounded outcome;
+- `backend_job_deliveries` records best-effort notification separately from the
+  result, so a missing or stopped recipient never discards a result or reruns
+  model work.
+
+Exact dispatch retry reuses the deterministic attempt and Worker allocation key.
+Only explicit re-evaluation increments the bounded attempt number. The runner
+uses the existing embedded Runtime spawn, Worker registry, Session input,
+observation, usage, and lifecycle paths with no Workdir attachment. It uses the
+dedicated `builtin:backend-job` profile, whose tools cannot mutate Workspace
+state, recursively create Workers or Jobs, or inherit caller Workdir authority.
+
+Success is accepted only by the structured result boundary. The submission is
+fenced by Workspace, Job ID, attempt ID, input revision, and the bound Runtime
+Worker identity, and its JSON bytes are bounded before persistence. Worker final
+prose, Idle, Stopped, or a successful model call are not success evidence.
+Identical structured submission is idempotent; conflicting replay, stale or late
+attempts, wrong revisions, and other Workers fail closed.
+
+On restart, an unbound reservation is redispatched with its stable allocation.
+A dispatched attempt continues only when its bound Worker remains observable;
+otherwise it becomes `unknown`. Durable deadlines fence acceptance before the
+runner requests Worker stop. The design does not claim general exactly-once
+model execution: a crash between Runtime creation and durable binding converges
+through stable allocation where available, and unrecoverable external evidence
+is surfaced as `unknown` rather than guessed. Historical attempts preserve the
+reason for retry without duplicating the Runtime Worker lifecycle state machine.
+
+Workspace Worker subscriptions expose an explicit Job binding (`job_id`,
+`attempt_id`, and `purpose`) in snapshot and live projections. Sidebar hiding is
+a Web View policy over that canonical metadata; the Worker remains registered,
+observable, and eligible for its canonical Console route.
+
 ## 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'>;