diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs
index 20ddab2d..f6f1db17 100644
--- a/crates/worker-runtime/src/catalog.rs
+++ b/crates/worker-runtime/src/catalog.rs
@@ -166,10 +166,29 @@ pub struct WorkingDirectoryStatus {
     pub summary: WorkingDirectorySummary,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct WorkspaceApiRef {
     pub workspace_id: String,
     pub base_url: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub runtime_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub access_token: Option<String>,
+}
+
+impl std::fmt::Debug for WorkspaceApiRef {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("WorkspaceApiRef")
+            .field("workspace_id", &self.workspace_id)
+            .field("base_url", &self.base_url)
+            .field("runtime_id", &self.runtime_id)
+            .field(
+                "access_token",
+                &self.access_token.as_ref().map(|_| "[redacted]"),
+            )
+            .finish()
+    }
 }
 
 /// Canonical Runtime Worker creation request.
diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs
index 1565bf90..47c13cd2 100644
--- a/crates/worker-runtime/src/http_server.rs
+++ b/crates/worker-runtime/src/http_server.rs
@@ -1153,6 +1153,8 @@ mod tests {
         request.workspace_api = Some(WorkspaceApiRef {
             workspace_id: workspace_id.to_string(),
             base_url: format!("https://workspace.example/{workspace_id}"),
+            runtime_id: None,
+            access_token: None,
         });
         request
     }
diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs
index 8bf620cb..e6a71785 100644
--- a/crates/worker-runtime/src/runtime.rs
+++ b/crates/worker-runtime/src/runtime.rs
@@ -2161,6 +2161,8 @@ mod tests {
         request.workspace_api = Some(WorkspaceApiRef {
             workspace_id: workspace_id.to_string(),
             base_url: format!("https://workspace.example/{workspace_id}"),
+            runtime_id: None,
+            access_token: None,
         });
         request
     }
diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs
index 113e40b8..4916ce62 100644
--- a/crates/worker-runtime/src/worker_backend.rs
+++ b/crates/worker-runtime/src/worker_backend.rs
@@ -23,6 +23,7 @@ use crate::execution::{
     WorkerExecutionRestoreRequest, WorkerExecutionResult, WorkerExecutionRunState,
     WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult,
 };
+use crate::identity::WorkerRef;
 use crate::interaction::{WorkerInput, WorkerInputKind};
 use crate::resource::{BackendResourceClient, ProfileSourceArchiveCache};
 use crate::working_directory::{
@@ -40,8 +41,8 @@ use tokio::sync::broadcast;
 #[cfg(feature = "ws-server")]
 use worker::ipc::protocol_session::{live_log_entry_event, subscribe_worker_protocol_session};
 use worker::{
-    PromptLoader, Worker, WorkerController, WorkerError, WorkerFilesystemAuthority, WorkerHandle,
-    WorkerWorkspaceContext, WorkspaceClient, WorkspaceId,
+    PromptLoader, RuntimeWorkspaceHttpClient, Worker, WorkerController, WorkerError,
+    WorkerFilesystemAuthority, WorkerHandle, WorkerWorkspaceContext, WorkspaceId,
 };
 
 const DEFAULT_BACKEND_ID: &str = "worker-crate";
@@ -259,6 +260,7 @@ enum RuntimeWorkspaceBackendRef {
     Http {
         workspace_id: String,
         base_url: String,
+        access_token: Option<String>,
     },
 }
 
@@ -268,20 +270,29 @@ impl RuntimeWorkspaceBackendRef {
             return Self::Http {
                 workspace_id: api.workspace_id.clone(),
                 base_url: api.base_url.clone(),
+                access_token: api.access_token.clone(),
             };
         }
         Self::None
     }
 
-    fn worker_context(&self) -> WorkerWorkspaceContext {
+    fn worker_context(&self, worker_ref: &WorkerRef) -> WorkerWorkspaceContext {
         match self {
             Self::None => WorkerWorkspaceContext::no_workspace(),
             Self::Http {
                 workspace_id,
                 base_url,
+                access_token,
             } => WorkerWorkspaceContext::with_client(
                 WorkspaceId::new(workspace_id.clone()).ok(),
-                WorkspaceClient::http(workspace_id.clone(), base_url.clone()),
+                Arc::new(
+                    RuntimeWorkspaceHttpClient::new(
+                        workspace_id.clone(),
+                        base_url.clone(),
+                        worker_ref.worker_id.to_string(),
+                    )
+                    .with_access_token(access_token.clone()),
+                ),
             ),
         }
     }
@@ -368,7 +379,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
             .unwrap_or(WorkerFilesystemAuthority::None);
         let workspace_backend_ref =
             RuntimeWorkspaceBackendRef::from_worker_request(&request.request);
-        let workspace_context = workspace_backend_ref.worker_context();
+        let workspace_context = workspace_backend_ref.worker_context(&request.worker_ref);
         let selector = profile.as_ref();
         let archive = self
             .resolve_profile_source_archive(&request.request.profile_source)
@@ -442,7 +453,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
             .unwrap_or(WorkerFilesystemAuthority::None);
         let workspace_backend_ref =
             RuntimeWorkspaceBackendRef::from_worker_request(&request.request);
-        let workspace_context = workspace_backend_ref.worker_context();
+        let workspace_context = workspace_backend_ref.worker_context(&request.worker_ref);
         let (manifest, loader) = Self::restore_fallback_manifest(&worker_name)?;
 
         let store_dir = self.store_dir()?;
@@ -1276,7 +1287,7 @@ mod tests {
         store_dir: PathBuf,
         worker_metadata_dir: PathBuf,
         observed_cwds: Arc<Mutex<Vec<PathBuf>>>,
-        observed_workspace_clients: Arc<Mutex<Vec<WorkspaceClient>>>,
+        observed_workspace_clients: Arc<Mutex<Vec<(String, Option<String>, bool)>>>,
     }
 
     #[async_trait]
@@ -1325,11 +1336,13 @@ mod tests {
                 .unwrap_or_else(|| self.cwd.clone());
             let workspace_backend_ref =
                 RuntimeWorkspaceBackendRef::from_worker_request(&request.request);
-            let workspace_context = workspace_backend_ref.worker_context();
-            self.observed_workspace_clients
-                .lock()
-                .unwrap()
-                .push(workspace_context.client().clone());
+            let workspace_context = workspace_backend_ref.worker_context(&request.worker_ref);
+            let workspace_client = workspace_context.client();
+            self.observed_workspace_clients.lock().unwrap().push((
+                workspace_client.kind().to_string(),
+                workspace_client.workspace_id().map(str::to_string),
+                workspace_client.is_available(),
+            ));
             let scope = Scope::writable(&scope_root).map_err(|err| err.to_string())?;
             let worker = Worker::new(
                 manifest,
@@ -1673,6 +1686,8 @@ mod tests {
         request.workspace_api = Some(crate::catalog::WorkspaceApiRef {
             workspace_id: "ws-test".to_string(),
             base_url: "http://127.0.0.1:3999".to_string(),
+            runtime_id: None,
+            access_token: None,
         });
         let detail = runtime.create_worker(request).unwrap();
 
@@ -1704,7 +1719,11 @@ mod tests {
         assert!(observed_cwds.lock().unwrap().is_empty());
         assert_eq!(
             observed_workspace_clients.lock().unwrap().as_slice(),
-            &[WorkspaceClient::http("ws-test", "http://127.0.0.1:3999")]
+            &[(
+                "runtime-http-proxy".to_string(),
+                Some("ws-test".to_string()),
+                true,
+            )]
         );
         let names = captured_tool_names(&client, 0);
         for forbidden in core_filesystem_tool_names() {
@@ -1781,9 +1800,7 @@ mod tests {
         assert!(cwd.join("README.md").exists());
         assert_eq!(
             observed_workspace_clients.lock().unwrap().as_slice(),
-            &[WorkspaceClient::Unavailable {
-                reason: "no workspace configured".to_string()
-            }]
+            &[("unavailable".to_string(), None, false)]
         );
     }
 
diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs
index b8c35fcd..48a43235 100644
--- a/crates/worker/src/controller.rs
+++ b/crates/worker/src/controller.rs
@@ -26,7 +26,7 @@ use crate::shutdown_after_idle::{
 use crate::spawn::comm_tools::{read_worker_output_tool, send_to_worker_tool, stop_worker_tool};
 use crate::spawn::registry::SpawnedWorkerRegistry;
 use crate::spawn::tool::spawn_worker_tool;
-use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult, WorkspaceClient};
+use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult};
 use protocol::{
     AlertLevel, AlertSource, ErrorCode, Event, Method, RewindTargetId, RunResult, Segment,
     TurnResult, WorkerStatus,
@@ -627,21 +627,16 @@ where
         // Ticket tools are typed operations over the current workspace Ticket backend.
         // Workspace access must be authority-bound to the Backend Workspace API; the
         // Worker must not fall back to a local `.yoi/tickets` store.
-        let ticket_backend = match worker.workspace_client() {
-            WorkspaceClient::Http {
-                workspace_id,
-                base_url,
-            } => crate::feature::builtin::ticket::TicketFeatureBackend::WorkspaceHttp {
-                workspace_id: workspace_id.clone(),
-                base_url: base_url.clone(),
-            },
-            _ => {
-                return Err(std::io::Error::new(
-                    std::io::ErrorKind::InvalidInput,
-                    "ticket tools require Backend Workspace API authority",
-                ));
-            }
-        };
+        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,
+                "ticket tools require Backend Workspace API authority",
+            ));
+        }
+        let ticket_backend = crate::feature::builtin::ticket::TicketFeatureBackend::WorkspaceClient(
+            workspace_client,
+        );
         feature_registry.add_module(
             crate::feature::builtin::ticket::ticket_tools_feature_with_backend(
                 ticket_backend,
@@ -668,21 +663,16 @@ where
     }
 
     {
-        let workspace_client = worker.workspace_client().clone();
+        let workspace_client = worker.workspace_client_handle();
         let engine = worker.engine_mut();
 
         // Objective tools expose read-only project Objective context through the
         // Backend Workspace API. Workers must not guess local `.yoi/objectives`
         // paths or read Objective files directly.
         if feature_config.objective.enabled {
-            if let WorkspaceClient::Http {
-                workspace_id,
-                base_url,
-            } = &workspace_client
-            {
+            if workspace_client.is_available() && workspace_client.workspace_id().is_some() {
                 for definition in crate::feature::builtin::objective::workspace_http_objective_tools(
-                    workspace_id.clone(),
-                    base_url.clone(),
+                    workspace_client.clone(),
                 ) {
                     engine.register_tool(definition);
                 }
@@ -705,20 +695,14 @@ where
                     "[feature.memory].enabled = true requires a [memory] configuration section",
                 )
             })?;
-            if let WorkspaceClient::Http {
-                workspace_id,
-                base_url,
-            } = workspace_client
-            {
+            if workspace_client.is_available() && workspace_client.workspace_id().is_some() {
                 let definitions = if feature_config.memory.staging {
                     crate::feature::builtin::memory::workspace_http_memory_consolidation_tools(
-                        workspace_id,
-                        base_url,
+                        workspace_client.clone(),
                     )
                 } else {
                     crate::feature::builtin::memory::workspace_http_memory_tools(
-                        workspace_id,
-                        base_url,
+                        workspace_client.clone(),
                     )
                 };
                 for definition in definitions {
diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs
index 4e034fa9..97fc334a 100644
--- a/crates/worker/src/feature/builtin/memory.rs
+++ b/crates/worker/src/feature/builtin/memory.rs
@@ -20,27 +20,25 @@ use schemars::JsonSchema;
 use serde::de::DeserializeOwned;
 use serde_json::json;
 
-use crate::worker::WorkspaceClient;
+use crate::worker::{
+    WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
+};
 
 #[derive(Clone, Debug)]
 pub struct WorkspaceHttpMemoryBackend {
-    workspace_id: String,
-    base_url: String,
+    client: Arc<dyn WorkspaceClient>,
 }
 
 impl WorkspaceHttpMemoryBackend {
-    pub fn new(workspace_id: impl Into<String>, base_url: impl Into<String>) -> Self {
-        Self {
-            workspace_id: workspace_id.into(),
-            base_url: base_url.into(),
-        }
+    pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
+        Self { client }
     }
 
     pub async fn execute_operation(
         &self,
         operation: MemoryBackendOperation,
     ) -> Result<MemoryBackendOperationResult, WorkspaceMemoryBackendError> {
-        execute_http_memory_backend(&self.workspace_id, &self.base_url, operation).await
+        execute_memory_backend(self.client.as_ref(), operation).await
     }
 
     async fn execute(&self, operation: MemoryBackendOperation) -> Result<ToolOutput, ToolError> {
@@ -59,7 +57,7 @@ pub enum WorkspaceMemoryBackendError {
     #[error("workspace memory backend is unavailable: {reason}")]
     Unavailable { reason: String },
     #[error("workspace memory backend request failed: {0}")]
-    Request(#[from] reqwest::Error),
+    Request(#[from] WorkspaceClientError),
     #[error("workspace memory backend returned HTTP {status}: {body}")]
     Http {
         status: reqwest::StatusCode,
@@ -71,73 +69,49 @@ pub enum WorkspaceMemoryBackendError {
     Backend(String),
 }
 
-impl WorkspaceClient {
+impl dyn WorkspaceClient + '_ {
     pub async fn execute_memory_backend_operation(
         &self,
         operation: MemoryBackendOperation,
     ) -> Result<MemoryBackendOperationResult, WorkspaceMemoryBackendError> {
-        match self {
-            WorkspaceClient::Http {
-                workspace_id,
-                base_url,
-            } => execute_http_memory_backend(workspace_id, base_url, operation).await,
-            WorkspaceClient::Available { kind } => Err(WorkspaceMemoryBackendError::Unavailable {
-                reason: format!(
-                    "workspace client kind `{kind}` does not expose the Backend Workspace API"
-                ),
-            }),
-            WorkspaceClient::Unavailable { reason } => {
-                Err(WorkspaceMemoryBackendError::Unavailable {
-                    reason: reason.clone(),
-                })
-            }
-        }
+        execute_memory_backend(self, operation).await
     }
 
     pub async fn request_memory_staging_consolidation(
         &self,
         operation: MemoryConsolidateStagingOperation,
     ) -> Result<MemoryConsolidationOutput, WorkspaceMemoryBackendError> {
-        match self {
-            WorkspaceClient::Http {
-                workspace_id,
-                base_url,
-            } => execute_http_memory_consolidation(workspace_id, base_url, operation).await,
-            WorkspaceClient::Available { kind } => Err(WorkspaceMemoryBackendError::Unavailable {
-                reason: format!(
-                    "workspace client kind `{kind}` does not expose the Backend Workspace API"
-                ),
-            }),
-            WorkspaceClient::Unavailable { reason } => {
-                Err(WorkspaceMemoryBackendError::Unavailable {
-                    reason: reason.clone(),
-                })
-            }
-        }
+        execute_memory_consolidation(self, operation).await
     }
 }
 
-async fn execute_http_memory_backend(
-    workspace_id: &str,
-    base_url: &str,
+async fn execute_memory_backend(
+    client: &dyn WorkspaceClient,
     operation: MemoryBackendOperation,
 ) -> Result<MemoryBackendOperationResult, WorkspaceMemoryBackendError> {
-    let url = format!(
-        "{}/api/w/{}/memory/backend",
-        base_url.trim_end_matches('/'),
-        workspace_id
-    );
-    let response = reqwest::Client::new()
-        .post(url)
-        .json(&operation)
-        .send()
-        .await?;
-    let status = response.status();
-    let body = response.text().await?;
-    if !status.is_success() {
-        return Err(WorkspaceMemoryBackendError::Http { status, body });
+    let workspace_id =
+        client
+            .workspace_id()
+            .ok_or_else(|| WorkspaceMemoryBackendError::Unavailable {
+                reason: format!(
+                    "workspace client kind `{}` has no workspace id",
+                    client.kind()
+                ),
+            })?;
+    let response = client.execute(WorkspaceRequest::json(
+        WorkspaceRequestMethod::Post,
+        format!("/api/w/{workspace_id}/memory/backend"),
+        serde_json::to_string(&operation)?,
+    ))?;
+    let status = reqwest::StatusCode::from_u16(response.status)
+        .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
+    if !response.is_success() {
+        return Err(WorkspaceMemoryBackendError::Http {
+            status,
+            body: response.body,
+        });
     }
-    match serde_json::from_str::<MemoryBackendHttpResponse>(&body)? {
+    match serde_json::from_str::<MemoryBackendHttpResponse>(&response.body)? {
         MemoryBackendHttpResponse::Ok { result } => Ok(result),
         MemoryBackendHttpResponse::Error { message } => {
             Err(WorkspaceMemoryBackendError::Backend(message))
@@ -145,34 +119,37 @@ async fn execute_http_memory_backend(
     }
 }
 
-async fn execute_http_memory_consolidation(
-    workspace_id: &str,
-    base_url: &str,
+async fn execute_memory_consolidation(
+    client: &dyn WorkspaceClient,
     operation: MemoryConsolidateStagingOperation,
 ) -> Result<MemoryConsolidationOutput, WorkspaceMemoryBackendError> {
-    let url = format!(
-        "{}/api/w/{}/memory/consolidation",
-        base_url.trim_end_matches('/'),
-        workspace_id
-    );
-    let response = reqwest::Client::new()
-        .post(url)
-        .json(&operation)
-        .send()
-        .await?;
-    let status = response.status();
-    let body = response.text().await?;
-    if !status.is_success() {
-        return Err(WorkspaceMemoryBackendError::Http { status, body });
+    let workspace_id =
+        client
+            .workspace_id()
+            .ok_or_else(|| WorkspaceMemoryBackendError::Unavailable {
+                reason: format!(
+                    "workspace client kind `{}` has no workspace id",
+                    client.kind()
+                ),
+            })?;
+    let response = client.execute(WorkspaceRequest::json(
+        WorkspaceRequestMethod::Post,
+        format!("/api/w/{workspace_id}/memory/consolidation"),
+        serde_json::to_string(&operation)?,
+    ))?;
+    let status = reqwest::StatusCode::from_u16(response.status)
+        .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
+    if !response.is_success() {
+        return Err(WorkspaceMemoryBackendError::Http {
+            status,
+            body: response.body,
+        });
     }
-    serde_json::from_str::<MemoryConsolidationOutput>(&body).map_err(Into::into)
+    serde_json::from_str::<MemoryConsolidationOutput>(&response.body).map_err(Into::into)
 }
 
-pub fn workspace_http_memory_tools(
-    workspace_id: impl Into<String>,
-    base_url: impl Into<String>,
-) -> Vec<ToolDefinition> {
-    let backend = WorkspaceHttpMemoryBackend::new(workspace_id, base_url);
+pub fn workspace_http_memory_tools(client: Arc<dyn WorkspaceClient>) -> Vec<ToolDefinition> {
+    let backend = WorkspaceHttpMemoryBackend::new(client);
     vec![
         memory_tool(
             "MemoryReadDocument",
@@ -215,13 +192,10 @@ pub fn workspace_http_memory_tools(
 }
 
 pub fn workspace_http_memory_consolidation_tools(
-    workspace_id: impl Into<String>,
-    base_url: impl Into<String>,
+    client: Arc<dyn WorkspaceClient>,
 ) -> Vec<ToolDefinition> {
-    let workspace_id = workspace_id.into();
-    let base_url = base_url.into();
-    let mut tools = workspace_http_memory_tools(workspace_id.clone(), base_url.clone());
-    let backend = WorkspaceHttpMemoryBackend::new(workspace_id, base_url);
+    let mut tools = workspace_http_memory_tools(client.clone());
+    let backend = WorkspaceHttpMemoryBackend::new(client);
     tools.extend([
         memory_tool(
             "MemoryStagingList",
@@ -370,6 +344,14 @@ mod tests {
     use super::*;
     use llm_engine::tool::ToolDefinition;
 
+    fn test_client() -> Arc<dyn WorkspaceClient> {
+        Arc::new(crate::worker::RuntimeWorkspaceHttpClient::new(
+            "workspace",
+            "http://backend",
+            "test-worker",
+        ))
+    }
+
     fn tool_names(definitions: Vec<ToolDefinition>) -> Vec<String> {
         let mut names = definitions
             .into_iter()
@@ -390,10 +372,7 @@ mod tests {
 
     #[test]
     fn normal_workspace_memory_tools_do_not_include_staging_tools() {
-        let names = tool_names(workspace_http_memory_tools(
-            "workspace".to_string(),
-            "http://backend".to_string(),
-        ));
+        let names = tool_names(workspace_http_memory_tools(test_client()));
 
         assert!(names.contains(&"MemoryQuery".to_string()));
         assert!(names.contains(&"MemoryReadDocument".to_string()));
@@ -410,7 +389,7 @@ mod tests {
     #[test]
     fn document_update_schema_is_edit_like_and_staging_close_has_no_legacy_kinds() {
         let update_schema = tool_meta(
-            workspace_http_memory_tools("workspace".to_string(), "http://backend".to_string()),
+            workspace_http_memory_tools(test_client()),
             "MemoryUpdateDocument",
         );
         assert_eq!(
@@ -423,10 +402,7 @@ mod tests {
         assert!(update_schema["properties"].get("body_md").is_none());
 
         let close_schema_text = tool_meta(
-            workspace_http_memory_consolidation_tools(
-                "workspace".to_string(),
-                "http://backend".to_string(),
-            ),
+            workspace_http_memory_consolidation_tools(test_client()),
             "MemoryStagingClose",
         )
         .to_string();
@@ -440,10 +416,7 @@ mod tests {
 
     #[test]
     fn consolidation_workspace_memory_tools_include_staging_tools() {
-        let names = tool_names(workspace_http_memory_consolidation_tools(
-            "workspace".to_string(),
-            "http://backend".to_string(),
-        ));
+        let names = tool_names(workspace_http_memory_consolidation_tools(test_client()));
 
         assert!(names.contains(&"MemoryQuery".to_string()));
         assert!(names.contains(&"MemoryReadDocument".to_string()));
diff --git a/crates/worker/src/feature/builtin/objective.rs b/crates/worker/src/feature/builtin/objective.rs
index 9c1cbc9e..49f46515 100644
--- a/crates/worker/src/feature/builtin/objective.rs
+++ b/crates/worker/src/feature/builtin/objective.rs
@@ -14,26 +14,27 @@ use llm_engine::tool::{
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 
+use crate::worker::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod};
+
 #[derive(Clone, Debug)]
 pub struct WorkspaceHttpObjectiveBackend {
-    workspace_id: String,
-    base_url: String,
+    client: Arc<dyn WorkspaceClient>,
 }
 
 impl WorkspaceHttpObjectiveBackend {
-    pub fn new(workspace_id: impl Into<String>, base_url: impl Into<String>) -> Self {
-        Self {
-            workspace_id: workspace_id.into(),
-            base_url: base_url.into().trim_end_matches('/').to_string(),
-        }
+    pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
+        Self { client }
     }
 
     async fn list(&self, input: ObjectiveListInput) -> Result<ToolOutput, ToolError> {
-        let mut url = format!("{}/api/w/{}/objectives", self.base_url, self.workspace_id);
+        let mut url = format!(
+            "/api/w/{}/objectives",
+            self.client.workspace_id().unwrap_or_default()
+        );
         if let Some(limit) = input.limit {
             url.push_str(&format!("?limit={}", limit.min(1000)));
         }
-        let response = get_json::<ObjectiveListResponse>(&url)
+        let response = get_json::<ObjectiveListResponse>(self.client.as_ref(), &url)
             .await
             .map_err(backend_error)?;
         let count = response.items.len();
@@ -46,7 +47,7 @@ impl WorkspaceHttpObjectiveBackend {
     async fn show(&self, input: ObjectiveShowInput) -> Result<ToolOutput, ToolError> {
         let id = validate_id(&input.id, "ObjectiveShow")?;
         let url = self.objective_url(id);
-        let response = get_json::<ObjectiveDetail>(&url)
+        let response = get_json::<ObjectiveDetail>(self.client.as_ref(), &url)
             .await
             .map_err(backend_error)?;
         Ok(objective_output(
@@ -61,11 +62,18 @@ impl WorkspaceHttpObjectiveBackend {
                 "ObjectiveCreate requires non-empty title".to_string(),
             ));
         }
-        let url = format!("{}/api/w/{}/objectives", self.base_url, self.workspace_id);
-        let response =
-            send_json::<ObjectiveCreateInput, ObjectiveDetail>(reqwest::Method::POST, &url, &input)
-                .await
-                .map_err(backend_error)?;
+        let url = format!(
+            "/api/w/{}/objectives",
+            self.client.workspace_id().unwrap_or_default()
+        );
+        let response = send_json::<ObjectiveCreateInput, ObjectiveDetail>(
+            self.client.as_ref(),
+            reqwest::Method::POST,
+            &url,
+            &input,
+        )
+        .await
+        .map_err(backend_error)?;
         Ok(objective_output(
             format!("Created objective {}", response.id),
             response,
@@ -86,10 +94,14 @@ impl WorkspaceHttpObjectiveBackend {
             new_string: input.new_string,
             replace_all: input.replace_all,
         };
-        let response =
-            send_json::<ObjectiveEditRequest, ObjectiveDetail>(reqwest::Method::PATCH, &url, &body)
-                .await
-                .map_err(backend_error)?;
+        let response = send_json::<ObjectiveEditRequest, ObjectiveDetail>(
+            self.client.as_ref(),
+            reqwest::Method::PATCH,
+            &url,
+            &body,
+        )
+        .await
+        .map_err(backend_error)?;
         Ok(objective_output(
             format!("Edited objective {}", response.id),
             response,
@@ -105,6 +117,7 @@ impl WorkspaceHttpObjectiveBackend {
         }
         let url = format!("{}/state", self.objective_url(id));
         let response = send_json::<ObjectiveSetStateRequest, ObjectiveDetail>(
+            self.client.as_ref(),
             reqwest::Method::POST,
             &url,
             &ObjectiveSetStateRequest { state: input.state },
@@ -122,6 +135,7 @@ impl WorkspaceHttpObjectiveBackend {
         let ticket_id = validate_id(&input.ticket_id, "ObjectiveLinkTicket")?;
         let url = format!("{}/ticket-links", self.objective_url(id));
         let response = send_json::<ObjectiveLinkTicketRequest, ObjectiveDetail>(
+            self.client.as_ref(),
             reqwest::Method::POST,
             &url,
             &ObjectiveLinkTicketRequest {
@@ -143,7 +157,7 @@ impl WorkspaceHttpObjectiveBackend {
         let id = validate_id(&input.id, "ObjectiveUnlinkTicket")?;
         let ticket_id = validate_id(&input.ticket_id, "ObjectiveUnlinkTicket")?;
         let url = format!("{}/ticket-links/{}", self.objective_url(id), ticket_id);
-        let response = delete_json::<ObjectiveDetail>(&url)
+        let response = delete_json::<ObjectiveDetail>(self.client.as_ref(), &url)
             .await
             .map_err(backend_error)?;
         Ok(objective_output(
@@ -153,17 +167,15 @@ impl WorkspaceHttpObjectiveBackend {
     }
 
     fn objective_url(&self, id: &str) -> String {
-        format!(
-            "{}/api/w/{}/objectives/{}",
-            self.base_url, self.workspace_id, id
-        )
+        let workspace_id = self.client.workspace_id().unwrap_or_default();
+        format!("/api/w/{workspace_id}/objectives/{id}")
     }
 }
 
 #[derive(Debug, thiserror::Error)]
 pub enum WorkspaceObjectiveBackendError {
     #[error("workspace objective backend request failed: {0}")]
-    Request(#[from] reqwest::Error),
+    Request(#[from] crate::worker::WorkspaceClientError),
     #[error("workspace objective backend returned HTTP {status}: {body}")]
     Http {
         status: reqwest::StatusCode,
@@ -182,41 +194,55 @@ fn backend_error(error: WorkspaceObjectiveBackendError) -> ToolError {
 }
 
 async fn get_json<T: for<'de> Deserialize<'de>>(
-    url: &str,
+    client: &dyn WorkspaceClient,
+    path: &str,
 ) -> Result<T, WorkspaceObjectiveBackendError> {
-    let response = reqwest::Client::new().get(url).send().await?;
-    decode_response(response).await
+    decode_response(client.execute(WorkspaceRequest::get(path))?)
 }
 
 async fn send_json<B: Serialize, T: for<'de> Deserialize<'de>>(
+    client: &dyn WorkspaceClient,
     method: reqwest::Method,
-    url: &str,
+    path: &str,
     body: &B,
 ) -> Result<T, WorkspaceObjectiveBackendError> {
-    let response = reqwest::Client::new()
-        .request(method, url)
-        .json(body)
-        .send()
-        .await?;
-    decode_response(response).await
+    let method = match method {
+        reqwest::Method::POST => WorkspaceRequestMethod::Post,
+        reqwest::Method::PUT => WorkspaceRequestMethod::Put,
+        reqwest::Method::PATCH => WorkspaceRequestMethod::Patch,
+        reqwest::Method::DELETE => WorkspaceRequestMethod::Delete,
+        _ => WorkspaceRequestMethod::Get,
+    };
+    decode_response(client.execute(WorkspaceRequest::json(
+        method,
+        path,
+        serde_json::to_string(body)?,
+    ))?)
 }
 
 async fn delete_json<T: for<'de> Deserialize<'de>>(
-    url: &str,
+    client: &dyn WorkspaceClient,
+    path: &str,
 ) -> Result<T, WorkspaceObjectiveBackendError> {
-    let response = reqwest::Client::new().delete(url).send().await?;
-    decode_response(response).await
+    decode_response(client.execute(WorkspaceRequest {
+        method: WorkspaceRequestMethod::Delete,
+        path: path.to_string(),
+        body: None,
+    })?)
 }
 
-async fn decode_response<T: for<'de> Deserialize<'de>>(
-    response: reqwest::Response,
+fn decode_response<T: for<'de> Deserialize<'de>>(
+    response: crate::worker::WorkspaceResponse,
 ) -> Result<T, WorkspaceObjectiveBackendError> {
-    let status = response.status();
-    let body = response.text().await?;
-    if !status.is_success() {
-        return Err(WorkspaceObjectiveBackendError::Http { status, body });
+    let status = reqwest::StatusCode::from_u16(response.status)
+        .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
+    if !response.is_success() {
+        return Err(WorkspaceObjectiveBackendError::Http {
+            status,
+            body: response.body,
+        });
     }
-    serde_json::from_str(&body).map_err(Into::into)
+    serde_json::from_str(&response.body).map_err(Into::into)
 }
 
 fn objective_output(summary: String, response: ObjectiveDetail) -> Result<ToolOutput, ToolError> {
@@ -236,11 +262,8 @@ fn validate_id<'a>(id: &'a str, tool_name: &str) -> Result<&'a str, ToolError> {
     Ok(id)
 }
 
-pub fn workspace_http_objective_tools(
-    workspace_id: impl Into<String>,
-    base_url: impl Into<String>,
-) -> Vec<ToolDefinition> {
-    let backend = WorkspaceHttpObjectiveBackend::new(workspace_id, base_url);
+pub fn workspace_http_objective_tools(client: Arc<dyn WorkspaceClient>) -> Vec<ToolDefinition> {
+    let backend = WorkspaceHttpObjectiveBackend::new(client);
     vec![
         objective_tool(
             "ObjectiveList",
@@ -600,10 +623,13 @@ mod tests {
 
     #[test]
     fn workspace_http_objective_tools_include_objective_crud_tools() {
-        let names = tool_names(workspace_http_objective_tools(
-            "workspace".to_string(),
-            "http://backend".to_string(),
-        ));
+        let names = tool_names(workspace_http_objective_tools(Arc::new(
+            crate::worker::RuntimeWorkspaceHttpClient::new(
+                "workspace",
+                "http://backend",
+                "test-worker",
+            ),
+        )));
 
         assert_eq!(
             names,
diff --git a/crates/worker/src/feature/builtin/session_explore.rs b/crates/worker/src/feature/builtin/session_explore.rs
index 5585a332..b71d48b7 100644
--- a/crates/worker/src/feature/builtin/session_explore.rs
+++ b/crates/worker/src/feature/builtin/session_explore.rs
@@ -29,7 +29,7 @@ const FINISH_EXTRACTION_DESCRIPTION: &str = "Finish the extract worker run after
 #[derive(Clone)]
 pub(crate) struct SessionExploreState {
     view: Arc<SessionReferenceView>,
-    workspace_client: WorkspaceClient,
+    workspace_client: Arc<dyn WorkspaceClient>,
     source: SourceRef,
     extract_run_id: String,
     staged: Arc<Mutex<Vec<String>>>,
@@ -39,7 +39,7 @@ pub(crate) struct SessionExploreState {
 impl SessionExploreState {
     pub(crate) fn new(
         view: SessionReferenceView,
-        workspace_client: WorkspaceClient,
+        workspace_client: Arc<dyn WorkspaceClient>,
         source: SourceRef,
     ) -> Self {
         Self {
@@ -615,7 +615,7 @@ mod tests {
 
     fn stub_memory_backend_response(
         body: &'static str,
-    ) -> (WorkspaceClient, mpsc::Receiver<String>) {
+    ) -> (Arc<dyn WorkspaceClient>, mpsc::Receiver<String>) {
         let listener = TcpListener::bind("127.0.0.1:0").unwrap();
         let addr = listener.local_addr().unwrap();
         let (tx, rx) = mpsc::channel();
@@ -659,7 +659,11 @@ mod tests {
             stream.write_all(response.as_bytes()).unwrap();
         });
         (
-            WorkspaceClient::http("test-workspace", format!("http://{addr}")),
+            Arc::new(crate::worker::RuntimeWorkspaceHttpClient::new(
+                "test-workspace",
+                format!("http://{addr}"),
+                "test-worker",
+            )),
             rx,
         )
     }
@@ -668,7 +672,7 @@ mod tests {
     fn descriptor_declares_session_explore_tools() {
         let state = SessionExploreState::new(
             SessionReferenceView::new("segment-1", vec![Item::user_message("remember this")]),
-            WorkspaceClient::available("test-backend"),
+            crate::worker::marker_workspace_client(None, "test-backend"),
             SourceRef {
                 segment_id: "segment-1".to_string(),
                 range: [0, 0],
diff --git a/crates/worker/src/feature/builtin/ticket.rs b/crates/worker/src/feature/builtin/ticket.rs
index 3b1f7e2f..b9e36510 100644
--- a/crates/worker/src/feature/builtin/ticket.rs
+++ b/crates/worker/src/feature/builtin/ticket.rs
@@ -4,7 +4,10 @@
 //! module only resolves the local backend root, declares the built-in feature,
 //! and contributes those tools through the normal feature registry path.
 
-use std::path::{Path, PathBuf};
+use std::{
+    path::{Path, PathBuf},
+    sync::Arc,
+};
 
 use ticket::{
     LocalTicketBackend, MarkdownText, NewOrchestrationPlanRecord, NewTicket, NewTicketEvent,
@@ -22,6 +25,7 @@ use crate::feature::{
     FeatureInstructionContribution, FeatureInstructionDeclaration, FeatureInstructionId,
     FeatureModule, ToolContribution, ToolDeclaration,
 };
+use crate::worker::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod};
 
 const FEATURE_ID: &str = "ticket";
 const FEATURE_NAME: &str = "Ticket tools";
@@ -183,13 +187,8 @@ const ORCHESTRATION_CONTROL_ADDITIONAL_TOOL_NAMES: &[&str] = &[
 
 #[derive(Clone, Debug)]
 pub enum TicketFeatureBackend {
-    Local {
-        root: PathBuf,
-    },
-    WorkspaceHttp {
-        workspace_id: String,
-        base_url: String,
-    },
+    Local { root: PathBuf },
+    WorkspaceClient(Arc<dyn WorkspaceClient>),
 }
 
 impl From<PathBuf> for TicketFeatureBackend {
@@ -274,7 +273,7 @@ impl TicketFeature {
     pub fn backend_root(&self) -> Option<&Path> {
         match &self.backend {
             TicketFeatureBackend::Local { root } => Some(root),
-            TicketFeatureBackend::WorkspaceHttp { .. } => None,
+            TicketFeatureBackend::WorkspaceClient(_) => None,
         }
     }
 
@@ -321,15 +320,9 @@ impl TicketFeature {
                         .into(),
                 )
             }
-            TicketFeatureBackend::WorkspaceHttp {
-                workspace_id,
-                base_url,
-            } => Some(
-                TicketToolBackend::new(WorkspaceHttpTicketBackend::new(
-                    workspace_id.clone(),
-                    base_url.clone(),
-                ))
-                .with_record_language(self.record_language.as_deref()),
+            TicketFeatureBackend::WorkspaceClient(client) => Some(
+                TicketToolBackend::new(WorkspaceHttpTicketBackend::new(client.clone()))
+                    .with_record_language(self.record_language.as_deref()),
             ),
         }
     }
@@ -386,22 +379,18 @@ impl FeatureModule for TicketFeature {
 
 #[derive(Clone, Debug)]
 struct WorkspaceHttpTicketBackend {
-    workspace_id: String,
-    base_url: String,
+    client: Arc<dyn WorkspaceClient>,
 }
 
 impl WorkspaceHttpTicketBackend {
-    fn new(workspace_id: String, base_url: String) -> Self {
-        Self {
-            workspace_id,
-            base_url: base_url.trim_end_matches('/').to_string(),
-        }
+    fn new(client: Arc<dyn WorkspaceClient>) -> Self {
+        Self { client }
     }
 
     fn endpoint(&self) -> String {
         format!(
-            "{}/api/w/{}/tickets/backend",
-            self.base_url, self.workspace_id
+            "/api/w/{}/tickets/backend",
+            self.client.workspace_id().unwrap_or_default()
         )
     }
 
@@ -409,44 +398,44 @@ impl WorkspaceHttpTicketBackend {
         &self,
         operation: TicketBackendOperation,
     ) -> TicketResult<TicketBackendOperationResult> {
+        let client = self.client.clone();
         let endpoint = self.endpoint();
         if tokio::runtime::Handle::try_current().is_ok() {
-            return std::thread::spawn(move || Self::invoke_http(endpoint, operation))
+            return std::thread::spawn(move || Self::invoke_client(client, endpoint, operation))
                 .join()
                 .map_err(|_| {
                     TicketError::Conflict("ticket backend request thread panicked".to_string())
                 })?;
         }
-        Self::invoke_http(endpoint, operation)
+        Self::invoke_client(client, endpoint, operation)
     }
 
-    fn invoke_http(
+    fn invoke_client(
+        client: Arc<dyn WorkspaceClient>,
         endpoint: String,
         operation: TicketBackendOperation,
     ) -> TicketResult<TicketBackendOperationResult> {
         let body = serde_json::to_string(&operation).map_err(|error| {
             TicketError::Conflict(format!("serialize ticket operation: {error}"))
         })?;
-        let response = reqwest::blocking::Client::new()
-            .post(endpoint)
-            .header(reqwest::header::CONTENT_TYPE, "application/json")
-            .body(body)
-            .send()
+        let response = client
+            .execute(WorkspaceRequest::json(
+                WorkspaceRequestMethod::Post,
+                endpoint,
+                body,
+            ))
             .map_err(|error| {
                 TicketError::Conflict(format!("ticket backend request failed: {error}"))
             })?;
-        let status = response.status();
-        let text = response.text().map_err(|error| {
-            TicketError::Conflict(format!("ticket backend response failed: {error}"))
-        })?;
-        if !status.is_success() {
+        if !response.is_success() {
             return Err(TicketError::Conflict(format!(
-                "ticket backend returned HTTP {status}: {text}"
+                "ticket backend returned HTTP {}: {}",
+                response.status, response.body
             )));
         }
-        match serde_json::from_str::<TicketBackendHttpResponse>(&text).map_err(|error| {
-            TicketError::Conflict(format!("decode ticket backend response: {error}"))
-        })? {
+        match serde_json::from_str::<TicketBackendHttpResponse>(&response.body).map_err(
+            |error| TicketError::Conflict(format!("decode ticket backend response: {error}")),
+        )? {
             TicketBackendHttpResponse::Ok { result } => Ok(result),
             TicketBackendHttpResponse::Error { message } => Err(TicketError::Conflict(message)),
         }
@@ -1126,8 +1115,13 @@ provider = "github"
 
     #[tokio::test(flavor = "multi_thread")]
     async fn workspace_http_backend_invoke_is_safe_inside_async_context() {
-        let backend =
-            WorkspaceHttpTicketBackend::new("workspace-a".to_string(), "not-a-url".to_string());
+        let backend = WorkspaceHttpTicketBackend::new(Arc::new(
+            crate::worker::RuntimeWorkspaceHttpClient::new(
+                "workspace-a",
+                "not-a-url",
+                "test-worker",
+            ),
+        ));
 
         let error = backend
             .invoke(TicketBackendOperation::DefaultIntakeReadyStateChangeBody {
@@ -1167,7 +1161,9 @@ provider = "github"
             .unwrap();
         });
 
-        let backend = WorkspaceHttpTicketBackend::new("workspace-a".to_string(), base_url);
+        let backend = WorkspaceHttpTicketBackend::new(Arc::new(
+            crate::worker::RuntimeWorkspaceHttpClient::new("workspace-a", base_url, "test-worker"),
+        ));
         let created = backend.create(NewTicket::new("HTTP ticket")).unwrap();
 
         server.join().unwrap();
diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs
index e792863e..66e74468 100644
--- a/crates/worker/src/lib.rs
+++ b/crates/worker/src/lib.rs
@@ -40,6 +40,9 @@ pub use runtime::dir::RuntimeDir;
 pub use segment_log_sink::SegmentLogSink;
 pub use shared_state::WorkerSharedState;
 pub use worker::{
-    LocalWorkingDirectory, Worker, WorkerError, WorkerFilesystemAuthority, WorkerRunResult,
-    WorkerWorkspaceContext, WorkspaceClient, WorkspaceId, WorkspaceIdError, apply_worker_manifest,
+    LocalWorkingDirectory, RuntimeWorkspaceHttpClient, Worker, WorkerError,
+    WorkerFilesystemAuthority, WorkerRunResult, WorkerWorkspaceContext, WorkspaceClient,
+    WorkspaceClientError, WorkspaceId, WorkspaceIdError, WorkspaceRequest, WorkspaceRequestMethod,
+    WorkspaceResponse, apply_worker_manifest, marker_workspace_client,
+    unavailable_workspace_client,
 };
diff --git a/crates/worker/src/skill.rs b/crates/worker/src/skill.rs
index 2c430786..2b1a1691 100644
--- a/crates/worker/src/skill.rs
+++ b/crates/worker/src/skill.rs
@@ -128,7 +128,7 @@ pub enum SkillClientError {
     #[error("workspace client kind `{0}` does not expose direct Skill HTTP operations")]
     UnsupportedClient(String),
     #[error("Skill request failed: {0}")]
-    Request(#[from] reqwest::Error),
+    Request(#[from] crate::worker::WorkspaceClientError),
     #[error("Skill API response JSON is invalid: {0}")]
     Json(#[from] serde_json::Error),
     #[error("Skill API returned HTTP {status}: {body}")]
@@ -140,7 +140,7 @@ pub enum SkillClientError {
     InvalidBaseUrl(String),
 }
 
-impl WorkspaceClient {
+impl dyn WorkspaceClient + '_ {
     pub fn list_skills(&self) -> Result<SkillCatalogResponse, SkillClientError> {
         self.get_skill_json("skills")
     }
@@ -157,29 +157,21 @@ impl WorkspaceClient {
         &self,
         path: &str,
     ) -> Result<T, SkillClientError> {
-        let Self::Http {
-            workspace_id,
-            base_url,
-        } = self
-        else {
-            return match self {
-                Self::Available { kind } => Err(SkillClientError::UnsupportedClient(kind.clone())),
-                Self::Unavailable { reason } => Err(SkillClientError::Unavailable(reason.clone())),
-                Self::Http { .. } => unreachable!(),
-            };
-        };
-        if base_url.trim().is_empty() {
-            return Err(SkillClientError::InvalidBaseUrl(base_url.clone()));
+        let workspace_id = self
+            .workspace_id()
+            .ok_or_else(|| SkillClientError::UnsupportedClient(self.kind().to_string()))?;
+        let response = self.execute(crate::worker::WorkspaceRequest::get(format!(
+            "/api/w/{workspace_id}/{path}"
+        )))?;
+        let status = reqwest::StatusCode::from_u16(response.status)
+            .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
+        if !response.is_success() {
+            return Err(SkillClientError::Http {
+                status,
+                body: response.body,
+            });
         }
-        let base = base_url.trim_end_matches('/');
-        let url = format!("{base}/api/w/{workspace_id}/{path}");
-        let response = reqwest::blocking::Client::new().get(url).send()?;
-        let status = response.status();
-        let body = response.text()?;
-        if !status.is_success() {
-            return Err(SkillClientError::Http { status, body });
-        }
-        Ok(serde_json::from_str(&body)?)
+        Ok(serde_json::from_str(&response.body)?)
     }
 }
 
@@ -201,13 +193,23 @@ mod tests {
             let mut request_line = String::new();
             reader.read_line(&mut request_line).unwrap();
             assert!(request_line.starts_with("GET /api/w/ws-1/skills HTTP/1.1"));
+            let mut worker_header = None;
+            let mut authorization = None;
             loop {
                 let mut line = String::new();
                 reader.read_line(&mut line).unwrap();
+                if let Some(value) = line.strip_prefix("x-yoi-worker-id: ") {
+                    worker_header = Some(value.trim().to_string());
+                }
+                if let Some(value) = line.strip_prefix("authorization: ") {
+                    authorization = Some(value.trim().to_string());
+                }
                 if line == "\r\n" || line.is_empty() {
                     break;
                 }
             }
+            assert_eq!(worker_header.as_deref(), Some("test-worker"));
+            assert_eq!(authorization.as_deref(), Some("Bearer test-credential"));
             let body = serde_json::json!({
                 "authority": "workspace-backend-skills-v0",
                 "entries": [{
@@ -229,8 +231,13 @@ mod tests {
             .unwrap();
         });
 
-        let client = WorkspaceClient::http("ws-1", format!("http://{addr}"));
-        let catalog = client.list_skills().unwrap();
+        let client = crate::worker::RuntimeWorkspaceHttpClient::new(
+            "ws-1",
+            format!("http://{addr}"),
+            "test-worker",
+        )
+        .with_access_token(Some("test-credential".to_string()));
+        let catalog = (&client as &dyn WorkspaceClient).list_skills().unwrap();
         assert_eq!(catalog.entries[0].name, "triage-errors");
         assert_eq!(catalog.entries[0].provenance.id, "workspace:triage-errors");
         handle.join().unwrap();
diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs
index 445b8512..20e05921 100644
--- a/crates/worker/src/worker.rs
+++ b/crates/worker/src/worker.rs
@@ -143,77 +143,296 @@ pub enum WorkspaceIdError {
     Empty,
 }
 
-/// Narrow path-free workspace API handle injected by Runtime/host code.
-///
-/// This is deliberately not a filesystem authority surface. A Worker may have a
-/// workspace client without local filesystem authority, or neither. Local
-/// path-backed implementations are represented only as a capability marker here;
-/// the actual paths remain under [`WorkerFilesystemAuthority::Local`] or in host
-/// adapter code.
+/// One authority-bound operation sent through the Runtime-supplied Workspace client.
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub enum WorkspaceClient {
-    /// Runtime/host supplied an HTTP workspace API endpoint.
-    Http {
-        workspace_id: String,
-        base_url: String,
-    },
-    /// Runtime/host supplied a workspace API handle. The string is an opaque
-    /// diagnostic/backend kind, not an endpoint, path, or secret-bearing value.
-    Available { kind: String },
-    /// Workspace-aware operations must fail closed or stay disabled.
-    Unavailable { reason: String },
+pub struct WorkspaceRequest {
+    pub method: WorkspaceRequestMethod,
+    pub path: String,
+    pub body: Option<String>,
 }
 
-impl WorkspaceClient {
-    pub fn available(kind: impl Into<String>) -> Self {
-        Self::Available { kind: kind.into() }
+impl WorkspaceRequest {
+    pub fn get(path: impl Into<String>) -> Self {
+        Self {
+            method: WorkspaceRequestMethod::Get,
+            path: path.into(),
+            body: None,
+        }
     }
 
-    pub fn http(workspace_id: impl Into<String>, base_url: impl Into<String>) -> Self {
-        Self::Http {
+    pub fn json(
+        method: WorkspaceRequestMethod,
+        path: impl Into<String>,
+        body: impl Into<String>,
+    ) -> Self {
+        Self {
+            method,
+            path: path.into(),
+            body: Some(body.into()),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum WorkspaceRequestMethod {
+    Get,
+    Post,
+    Put,
+    Patch,
+    Delete,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WorkspaceResponse {
+    pub status: u16,
+    pub body: String,
+}
+
+impl WorkspaceResponse {
+    pub fn is_success(&self) -> bool {
+        (200..300).contains(&self.status)
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum WorkspaceClientError {
+    #[error("workspace client is unavailable: {0}")]
+    Unavailable(String),
+    #[error("workspace request path must start with '/': {0}")]
+    InvalidPath(String),
+    #[error("workspace request failed: {0}")]
+    Request(String),
+}
+
+/// Path-free Workspace operation authority injected by Runtime/host code.
+///
+/// Workers receive this trait object rather than a Backend URL. The concrete
+/// implementation is responsible for binding Runtime/Worker identity and
+/// forwarding operations to the Workspace authority.
+pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
+    fn workspace_id(&self) -> Option<&str>;
+    fn kind(&self) -> &str;
+    fn is_available(&self) -> bool;
+    fn execute(&self, request: WorkspaceRequest)
+    -> Result<WorkspaceResponse, WorkspaceClientError>;
+}
+
+/// HTTP forwarding client created by Runtime for one concrete Worker execution.
+///
+/// The upstream endpoint and source headers are private implementation details;
+/// model-visible tools can only submit [`WorkspaceRequest`] values through the
+/// [`WorkspaceClient`] trait.
+pub struct RuntimeWorkspaceHttpClient {
+    workspace_id: String,
+    base_url: String,
+    worker_id: String,
+    access_token: Option<String>,
+}
+
+impl std::fmt::Debug for RuntimeWorkspaceHttpClient {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("RuntimeWorkspaceHttpClient")
+            .field("workspace_id", &self.workspace_id)
+            .field("base_url", &self.base_url)
+            .field("worker_id", &self.worker_id)
+            .field(
+                "access_token",
+                &self.access_token.as_ref().map(|_| "[redacted]"),
+            )
+            .finish()
+    }
+}
+
+impl RuntimeWorkspaceHttpClient {
+    pub fn new(
+        workspace_id: impl Into<String>,
+        base_url: impl Into<String>,
+        worker_id: impl Into<String>,
+    ) -> Self {
+        Self {
             workspace_id: workspace_id.into(),
-            base_url: base_url.into(),
+            base_url: base_url.into().trim_end_matches('/').to_string(),
+            worker_id: worker_id.into(),
+            access_token: None,
         }
     }
 
-    pub fn unavailable(reason: impl Into<String>) -> Self {
-        Self::Unavailable {
-            reason: reason.into(),
+    pub fn with_access_token(mut self, access_token: Option<String>) -> Self {
+        self.access_token = access_token;
+        self
+    }
+}
+
+impl WorkspaceClient for RuntimeWorkspaceHttpClient {
+    fn workspace_id(&self) -> Option<&str> {
+        Some(&self.workspace_id)
+    }
+
+    fn kind(&self) -> &str {
+        "runtime-http-proxy"
+    }
+
+    fn is_available(&self) -> bool {
+        true
+    }
+
+    fn execute(
+        &self,
+        request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        let base_url = self.base_url.clone();
+        let worker_id = self.worker_id.clone();
+        let access_token = self.access_token.clone();
+        if tokio::runtime::Handle::try_current().is_ok() {
+            return std::thread::spawn(move || {
+                execute_runtime_workspace_http(
+                    &base_url,
+                    &worker_id,
+                    access_token.as_deref(),
+                    request,
+                )
+            })
+            .join()
+            .map_err(|_| {
+                WorkspaceClientError::Request("workspace request thread panicked".to_string())
+            })?;
         }
+        execute_runtime_workspace_http(&base_url, &worker_id, access_token.as_deref(), request)
+    }
+}
+
+fn execute_runtime_workspace_http(
+    base_url: &str,
+    worker_id: &str,
+    access_token: Option<&str>,
+    request: WorkspaceRequest,
+) -> Result<WorkspaceResponse, WorkspaceClientError> {
+    if !request.path.starts_with('/') || request.path.starts_with("//") {
+        return Err(WorkspaceClientError::InvalidPath(request.path));
+    }
+    let url = format!("{base_url}{}", request.path);
+    let method = match request.method {
+        WorkspaceRequestMethod::Get => reqwest::Method::GET,
+        WorkspaceRequestMethod::Post => reqwest::Method::POST,
+        WorkspaceRequestMethod::Put => reqwest::Method::PUT,
+        WorkspaceRequestMethod::Patch => reqwest::Method::PATCH,
+        WorkspaceRequestMethod::Delete => reqwest::Method::DELETE,
+    };
+    let client = reqwest::blocking::Client::new();
+    let mut request_builder = client
+        .request(method, url)
+        .header("x-yoi-worker-id", worker_id);
+    if let Some(access_token) = access_token {
+        request_builder = request_builder.bearer_auth(access_token);
+    }
+    if let Some(body) = request.body {
+        request_builder = request_builder
+            .header(reqwest::header::CONTENT_TYPE, "application/json")
+            .body(body);
+    }
+    let response = request_builder
+        .send()
+        .map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
+    let status = response.status().as_u16();
+    let body = response
+        .text()
+        .map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
+    Ok(WorkspaceResponse { status, body })
+}
+
+#[derive(Debug)]
+struct MarkerWorkspaceClient {
+    workspace_id: Option<String>,
+    kind: String,
+    available: bool,
+    reason: String,
+}
+
+impl WorkspaceClient for MarkerWorkspaceClient {
+    fn workspace_id(&self) -> Option<&str> {
+        self.workspace_id.as_deref()
     }
 
-    pub fn local_filesystem() -> Self {
-        Self::available("local-filesystem")
+    fn kind(&self) -> &str {
+        &self.kind
     }
 
-    pub fn is_available(&self) -> bool {
-        matches!(self, Self::Available { .. } | Self::Http { .. })
+    fn is_available(&self) -> bool {
+        self.available
     }
+
+    fn execute(
+        &self,
+        _request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        Err(WorkspaceClientError::Unavailable(self.reason.clone()))
+    }
+}
+
+pub fn unavailable_workspace_client(
+    workspace_id: Option<&WorkspaceId>,
+    reason: impl Into<String>,
+) -> Arc<dyn WorkspaceClient> {
+    Arc::new(MarkerWorkspaceClient {
+        workspace_id: workspace_id.map(|id| id.as_str().to_string()),
+        kind: "unavailable".to_string(),
+        available: false,
+        reason: reason.into(),
+    })
+}
+
+pub fn marker_workspace_client(
+    workspace_id: Option<&WorkspaceId>,
+    kind: impl Into<String>,
+) -> Arc<dyn WorkspaceClient> {
+    let kind = kind.into();
+    Arc::new(MarkerWorkspaceClient {
+        workspace_id: workspace_id.map(|id| id.as_str().to_string()),
+        reason: format!("workspace client kind `{kind}` does not expose Workspace operations"),
+        kind,
+        available: true,
+    })
 }
 
 /// Workspace context supplied to a Worker separately from filesystem authority.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Clone)]
 pub struct WorkerWorkspaceContext {
     workspace_id: Option<WorkspaceId>,
-    client: WorkspaceClient,
+    client: Arc<dyn WorkspaceClient>,
+}
+
+impl std::fmt::Debug for WorkerWorkspaceContext {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("WorkerWorkspaceContext")
+            .field("workspace_id", &self.workspace_id)
+            .field("client_kind", &self.client.kind())
+            .field("client_available", &self.client.is_available())
+            .finish()
+    }
 }
 
 impl WorkerWorkspaceContext {
     pub fn no_workspace() -> Self {
         Self {
             workspace_id: None,
-            client: WorkspaceClient::unavailable("no workspace configured"),
+            client: unavailable_workspace_client(None, "no workspace configured"),
         }
     }
 
     pub fn unavailable(workspace_id: Option<WorkspaceId>, reason: impl Into<String>) -> Self {
+        let client = unavailable_workspace_client(workspace_id.as_ref(), reason);
         Self {
             workspace_id,
-            client: WorkspaceClient::unavailable(reason),
+            client,
         }
     }
 
-    pub fn with_client(workspace_id: Option<WorkspaceId>, client: WorkspaceClient) -> Self {
+    pub fn with_client(
+        workspace_id: Option<WorkspaceId>,
+        client: Arc<dyn WorkspaceClient>,
+    ) -> Self {
         Self {
             workspace_id,
             client,
@@ -221,15 +440,23 @@ impl WorkerWorkspaceContext {
     }
 
     pub fn local_filesystem(workspace_id: Option<WorkspaceId>) -> Self {
-        Self::with_client(workspace_id, WorkspaceClient::local_filesystem())
+        let client = marker_workspace_client(workspace_id.as_ref(), "local-filesystem");
+        Self {
+            workspace_id,
+            client,
+        }
     }
 
     pub fn workspace_id(&self) -> Option<&WorkspaceId> {
         self.workspace_id.as_ref()
     }
 
-    pub fn client(&self) -> &WorkspaceClient {
-        &self.client
+    pub fn client(&self) -> &dyn WorkspaceClient {
+        self.client.as_ref()
+    }
+
+    pub fn client_handle(&self) -> Arc<dyn WorkspaceClient> {
+        self.client.clone()
     }
 }
 
@@ -926,10 +1153,14 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
 
     /// Narrow workspace client/availability handle injected by Runtime/host.
     /// This never grants local filesystem authority.
-    pub fn workspace_client(&self) -> &WorkspaceClient {
+    pub fn workspace_client(&self) -> &dyn WorkspaceClient {
         self.workspace_context.client()
     }
 
+    pub fn workspace_client_handle(&self) -> Arc<dyn WorkspaceClient> {
+        self.workspace_context.client_handle()
+    }
+
     async fn resident_summary_from_workspace_authority(
         &self,
     ) -> Result<Option<String>, WorkerError> {
@@ -3197,7 +3428,7 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
             items_to_extract,
         );
         let session_explore_state =
-            SessionExploreState::new(session_view, self.workspace_client().clone(), source);
+            SessionExploreState::new(session_view, self.workspace_client_handle(), source);
         let input_text = render_extract_input(session_explore_state.view());
         let mut internal_tools = Vec::new();
         let mut internal_hook_builder = HookRegistryBuilder::new();
@@ -3464,7 +3695,7 @@ impl WorkerAuditBase {
 
     async fn emit(
         &self,
-        workspace_client: &WorkspaceClient,
+        workspace_client: &dyn WorkspaceClient,
         event_tx: Option<&broadcast::Sender<Event>>,
         status: memory::audit::WorkerLifecycleStatus,
         reason: impl Into<String>,
@@ -4936,7 +5167,7 @@ mod spawned_context_tests {
             false,
             WorkerWorkspaceContext::with_client(
                 Some(workspace_id.clone()),
-                WorkspaceClient::available("test-api"),
+                marker_workspace_client(Some(&workspace_id), "test-api"),
             ),
             WorkerFilesystemAuthority::None,
             manifest.scope.clone(),
@@ -5773,7 +6004,11 @@ mod build_summary_prompt_tests {
         });
         WorkerWorkspaceContext::with_client(
             Some(WorkspaceId::new("test-memory").unwrap()),
-            WorkspaceClient::http("test-memory", format!("http://{addr}")),
+            Arc::new(RuntimeWorkspaceHttpClient::new(
+                "test-memory",
+                format!("http://{addr}"),
+                "test-worker",
+            )),
         )
     }
 
@@ -5905,7 +6140,11 @@ mod build_summary_prompt_tests {
                 store,
                 WorkerWorkspaceContext::with_client(
                     Some(WorkspaceId::new("ws-skill").unwrap()),
-                    WorkspaceClient::http("ws-skill", format!("http://{addr}")),
+                    Arc::new(RuntimeWorkspaceHttpClient::new(
+                        "ws-skill",
+                        format!("http://{addr}"),
+                        "test-worker",
+                    )),
                 ),
                 authority,
                 scope,
diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs
index d860992b..31946869 100644
--- a/crates/workspace-server/src/hosts.rs
+++ b/crates/workspace-server/src/hosts.rs
@@ -329,6 +329,8 @@ pub struct WorkerSpawnRequest {
     pub resolved_working_directory: Option<WorkingDirectoryClaim>,
     #[serde(skip, default)]
     pub resolved_config_bundle: Option<ConfigBundle>,
+    #[serde(skip, default)]
+    pub resolved_workspace_api: Option<WorkspaceApiRef>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -1703,13 +1705,16 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
             initial_input: request.initial_input.clone(),
             working_directory_request: request.resolved_working_directory_request.clone(),
             working_directory: request.resolved_working_directory.clone(),
-            workspace_api: self
-                .backend_base_url
-                .as_ref()
-                .map(|base_url| WorkspaceApiRef {
-                    workspace_id: self.workspace_id.clone(),
-                    base_url: base_url.clone(),
-                }),
+            workspace_api: request.resolved_workspace_api.clone().or_else(|| {
+                self.backend_base_url
+                    .as_ref()
+                    .map(|base_url| WorkspaceApiRef {
+                        workspace_id: self.workspace_id.clone(),
+                        base_url: base_url.clone(),
+                        runtime_id: Some(self.runtime_id.clone()),
+                        access_token: None,
+                    })
+            }),
         };
         match self.runtime.create_worker(create_request) {
             Ok(detail) => WorkerSpawnResult {
@@ -2677,9 +2682,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
             initial_input: request.initial_input.clone(),
             working_directory_request: request.resolved_working_directory_request.clone(),
             working_directory: request.resolved_working_directory.clone(),
-            workspace_api: Some(WorkspaceApiRef {
-                workspace_id: self.workspace_id.clone(),
-                base_url: self.backend_base_url.clone(),
+            workspace_api: request.resolved_workspace_api.clone().or_else(|| {
+                Some(WorkspaceApiRef {
+                    workspace_id: self.workspace_id.clone(),
+                    base_url: self.backend_base_url.clone(),
+                    runtime_id: Some(self.runtime_id.clone()),
+                    access_token: None,
+                })
             }),
         };
         match self.post_json::<_, RuntimeHttpWorkerResponse>("/v1/workers", &create) {
@@ -3121,8 +3130,11 @@ fn embedded_profile_path(profile: &ProfileSelector) -> Result<String, String> {
 fn embedded_profile_label(profile: &ProfileSelector) -> Option<String> {
     Some(match profile {
         ProfileSelector::Builtin(name) | ProfileSelector::Named(name) => {
-            if name.strip_prefix("builtin:").unwrap_or(name) == MEMORY_CONSOLIDATION_PROFILE {
+            let builtin_name = name.strip_prefix("builtin:").unwrap_or(name);
+            if builtin_name == MEMORY_CONSOLIDATION_PROFILE {
                 MEMORY_CONSOLIDATION_PROFILE.to_string()
+            } else if builtin_name == WORKSPACE_ORCHESTRATOR_PROFILE {
+                WORKSPACE_ORCHESTRATOR_PROFILE.to_string()
             } else {
                 safe_display_hint(name)
             }
@@ -3132,6 +3144,8 @@ fn embedded_profile_label(profile: &ProfileSelector) -> Option<String> {
 
 const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation";
 const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation";
+const WORKSPACE_ORCHESTRATOR_PROFILE: &str = "orchestrator";
+pub(crate) const WORKSPACE_ORCHESTRATOR_SINGLETON_KEY: &str = "workspace-orchestrator";
 
 struct WorkerDisplayMetadata {
     display_name: String,
@@ -3160,6 +3174,20 @@ fn worker_display_metadata(
             tags,
         };
     }
+    if profile_label == Some(WORKSPACE_ORCHESTRATOR_PROFILE) {
+        let mut tags = vec!["orchestrator".to_string(), "singleton".to_string()];
+        if internal {
+            tags.insert(0, "internal".to_string());
+        }
+        return WorkerDisplayMetadata {
+            display_name: requested_display_name
+                .filter(|value| !value.trim().is_empty())
+                .map(safe_display_hint)
+                .unwrap_or_else(|| "Workspace Orchestrator".to_string()),
+            singleton_key: Some(WORKSPACE_ORCHESTRATOR_SINGLETON_KEY.to_string()),
+            tags,
+        };
+    }
     let display_name = requested_display_name
         .filter(|value| !value.trim().is_empty())
         .map(safe_display_hint)
@@ -4154,6 +4182,7 @@ mod tests {
             resolved_working_directory_request: None,
             resolved_working_directory: None,
             resolved_config_bundle: None,
+            resolved_workspace_api: None,
         }
     }
 
@@ -4280,6 +4309,7 @@ mod tests {
                     resolved_working_directory_request: None,
                     resolved_working_directory: None,
                     resolved_config_bundle: None,
+                    resolved_workspace_api: None,
                 },
             )
             .unwrap();
@@ -4376,6 +4406,7 @@ mod tests {
                     resolved_working_directory_request: None,
                     resolved_working_directory: None,
                     resolved_config_bundle: None,
+                    resolved_workspace_api: None,
                 },
             )
             .unwrap();
@@ -4408,6 +4439,7 @@ mod tests {
                     resolved_working_directory_request: None,
                     resolved_working_directory: None,
                     resolved_config_bundle: None,
+                    resolved_workspace_api: None,
                 },
             )
             .unwrap();
diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs
index 0747f467..6ced71e4 100644
--- a/crates/workspace-server/src/lib.rs
+++ b/crates/workspace-server/src/lib.rs
@@ -85,6 +85,10 @@ pub enum Error {
     UnknownRepository(String),
     #[error("workspace id does not match this Workspace backend")]
     WorkspaceIdMismatch,
+    #[error("Ticket assignment conflict: {0}")]
+    TicketAssignmentConflict(String),
+    #[error("Worker Workspace authentication failed: {0}")]
+    WorkerWorkspaceAuthentication(String),
     #[error("workspace identity error: {0}")]
     WorkspaceIdentity(String),
     #[error("store error: {0}")]
diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs
index 43a9b15c..3399aaab 100644
--- a/crates/workspace-server/src/server.rs
+++ b/crates/workspace-server/src/server.rs
@@ -91,8 +91,9 @@ use crate::resource_broker::BackendResourceBroker;
 use crate::skills;
 use crate::store::{
     AccountRecord, ApiTokenRecord, AuthChallengeRecord, BrowserSessionRecord, ControlPlaneStore,
-    DeviceLoginFlowRecord, PasskeyCredentialRecord, RepositoryRecord, UserRecord,
-    WorkdirRegistryRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord, WorkspaceRecord,
+    DeviceLoginFlowRecord, PasskeyCredentialRecord, RepositoryRecord, TicketNotificationRecipient,
+    TicketWorkerAssignmentRecord, UserRecord, WorkdirRegistryRecord, WorkerRegistryRecord,
+    WorkerWorkdirLinkRecord, WorkerWorkspaceCredentialRecord, WorkspaceRecord,
 };
 use crate::{Error, Result};
 use worker_runtime::catalog::{
@@ -520,6 +521,12 @@ pub fn build_router(api: WorkspaceApi) -> Router {
             "/api/w/{workspace_id}/tickets/{id}",
             get(scoped_get_ticket).patch(scoped_edit_ticket_item),
         )
+        .route(
+            "/api/w/{workspace_id}/tickets/{id}/assignment",
+            get(scoped_get_ticket_worker_assignment)
+                .put(scoped_set_ticket_worker_assignment)
+                .delete(scoped_clear_ticket_worker_assignment),
+        )
         .route(
             "/api/w/{workspace_id}/tickets/{id}/state",
             post(scoped_transition_ticket_state),
@@ -780,7 +787,22 @@ pub async fn serve(
     listener: TcpListener,
 ) -> Result<()> {
     let api = WorkspaceApi::new(config, store).await?;
-    axum::serve(listener, build_router(api)).await?;
+    let dispatcher_api = api.clone();
+    let dispatcher_workspace_id = dispatcher_api.config.workspace_id.clone();
+    let dispatcher = tokio::spawn(async move {
+        loop {
+            let api = dispatcher_api.clone();
+            let workspace_id = dispatcher_workspace_id.clone();
+            let _ = tokio::task::spawn_blocking(move || {
+                dispatch_pending_ticket_notifications(&api, &workspace_id)
+            })
+            .await;
+            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+        }
+    });
+    let result = axum::serve(listener, build_router(api)).await;
+    dispatcher.abort();
+    result?;
     Ok(())
 }
 
@@ -1530,6 +1552,142 @@ async fn scoped_get_ticket(
     get_ticket(State(api), AxumPath(path.id)).await
 }
 
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
+struct TicketWorkerAssignmentResponse {
+    workspace_id: String,
+    ticket_id: String,
+    assignment: Option<TicketWorkerAssignmentRecord>,
+}
+
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
+struct TicketWorkerAssignmentMutationResponse {
+    workspace_id: String,
+    ticket_id: String,
+    assignment: Option<TicketWorkerAssignmentRecord>,
+    previous_assignment_id: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SetTicketWorkerAssignmentRequest {
+    runtime_id: String,
+    worker_id: String,
+    expected_assignment_id: Option<String>,
+    assigned_by: Option<String>,
+}
+
+#[derive(Debug, Default, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ClearTicketWorkerAssignmentQuery {
+    expected_assignment_id: Option<String>,
+    actor: Option<String>,
+}
+
+async fn scoped_get_ticket_worker_assignment(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedRecordPath>,
+) -> ApiResult<Json<TicketWorkerAssignmentResponse>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let ticket = api.authority.ticket(&path.id)?;
+    let assignment = api
+        .store
+        .get_current_ticket_worker_assignment(&path.workspace_id, &ticket.id)?;
+    Ok(Json(TicketWorkerAssignmentResponse {
+        workspace_id: path.workspace_id,
+        ticket_id: ticket.id,
+        assignment,
+    }))
+}
+
+async fn scoped_set_ticket_worker_assignment(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedRecordPath>,
+    Json(request): Json<SetTicketWorkerAssignmentRequest>,
+) -> ApiResult<Json<TicketWorkerAssignmentMutationResponse>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let ticket = api.authority.ticket(&path.id)?;
+    let runtime_id = require_ticket_assignment_value("runtime_id", request.runtime_id)?;
+    let worker_id = require_ticket_assignment_value("worker_id", request.worker_id)?;
+    let expected_assignment_id = request
+        .expected_assignment_id
+        .map(|value| require_ticket_assignment_value("expected_assignment_id", value))
+        .transpose()?;
+    let assigned_by = request
+        .assigned_by
+        .map(|value| require_ticket_assignment_value("assigned_by", value))
+        .transpose()?
+        .unwrap_or_else(|| "workspace-api".to_string());
+    let worker = api
+        .runtime
+        .worker(&runtime_id, &worker_id)
+        .map_err(|err| err.into_error())?;
+    let assigned_at = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
+    let record = TicketWorkerAssignmentRecord {
+        workspace_id: path.workspace_id.clone(),
+        ticket_id: ticket.id.clone(),
+        assignment_id: new_id("tasg"),
+        runtime_id: worker.runtime_id,
+        worker_id: worker.worker_id,
+        assigned_by,
+        assigned_at,
+    };
+    let update = api.store.set_current_ticket_worker_assignment(
+        &record,
+        expected_assignment_id.as_deref(),
+        &new_id("tasev"),
+    )?;
+    Ok(Json(TicketWorkerAssignmentMutationResponse {
+        workspace_id: path.workspace_id,
+        ticket_id: ticket.id,
+        assignment: Some(update.current),
+        previous_assignment_id: update.previous.map(|assignment| assignment.assignment_id),
+    }))
+}
+
+async fn scoped_clear_ticket_worker_assignment(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedRecordPath>,
+    Query(query): Query<ClearTicketWorkerAssignmentQuery>,
+) -> ApiResult<Json<TicketWorkerAssignmentMutationResponse>> {
+    validate_workspace_scope(&api, &path.workspace_id)?;
+    let ticket = api.authority.ticket(&path.id)?;
+    let expected_assignment_id = query
+        .expected_assignment_id
+        .map(|value| require_ticket_assignment_value("expected_assignment_id", value))
+        .transpose()?;
+    let actor = query
+        .actor
+        .map(|value| require_ticket_assignment_value("actor", value))
+        .transpose()?
+        .unwrap_or_else(|| "workspace-api".to_string());
+    let previous = api.store.clear_current_ticket_worker_assignment(
+        &path.workspace_id,
+        &ticket.id,
+        expected_assignment_id.as_deref(),
+        &new_id("tasev"),
+        &actor,
+        &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
+    )?;
+    Ok(Json(TicketWorkerAssignmentMutationResponse {
+        workspace_id: path.workspace_id,
+        ticket_id: ticket.id,
+        assignment: None,
+        previous_assignment_id: previous.map(|assignment| assignment.assignment_id),
+    }))
+}
+
+fn require_ticket_assignment_value(field: &str, value: String) -> Result<String> {
+    let value = value.trim();
+    if value.is_empty() {
+        return Err(Error::RuntimeOperationFailed {
+            runtime_id: "workspace-server".to_string(),
+            code: "invalid_ticket_assignment".to_string(),
+            message: format!("{field} must not be empty"),
+        });
+    }
+    Ok(value.to_string())
+}
+
 #[derive(Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct BrowserEditTicketRequest {
@@ -1747,7 +1905,8 @@ async fn scoped_close_ticket(
 async fn scoped_ticket_backend_operation(
     State(api): State<WorkspaceApi>,
     AxumPath(path): AxumPath<ScopedWorkspacePath>,
-    Json(operation): Json<TicketBackendOperation>,
+    headers: HeaderMap,
+    Json(mut operation): Json<TicketBackendOperation>,
 ) -> ApiResult<Json<TicketBackendHttpResponse>> {
     validate_workspace_scope(&api, &path.workspace_id)?;
     let config = ticket::config::TicketConfig::load_workspace(&api.config.workspace_root)
@@ -1757,8 +1916,36 @@ async fn scoped_ticket_backend_operation(
         api.config.workspace_id.clone(),
     )
     .with_record_language(config.ticket_record_language());
+    let target = ticket_mutation_target(&operation).cloned();
+    let source = target
+        .as_ref()
+        .map(|_| authenticate_worker_mutation_source(&api, &path.workspace_id, &headers))
+        .transpose()?;
+    if let (Some(source), Some(target)) = (source.as_ref(), target.as_ref()) {
+        bind_worker_ticket_operation_source(
+            &api,
+            &path.workspace_id,
+            source,
+            target,
+            &mut operation,
+        )?;
+    }
+    let before = target.as_ref().and_then(|id| backend.show(id.clone()).ok());
     let response = match execute_ticket_backend_operation(&backend, operation) {
-        Ok(result) => TicketBackendHttpResponse::Ok { result },
+        Ok(result) => {
+            if let (Some(source), Some(target)) = (source, target) {
+                let after = backend.show(target).map_err(Error::from)?;
+                enqueue_worker_ticket_notification(
+                    &api,
+                    &path.workspace_id,
+                    &source,
+                    before.as_ref(),
+                    &after,
+                )?;
+                dispatch_pending_ticket_notifications(&api, &path.workspace_id);
+            }
+            TicketBackendHttpResponse::Ok { result }
+        }
         Err(error) => TicketBackendHttpResponse::Error {
             message: error.to_string(),
         },
@@ -1766,6 +1953,288 @@ async fn scoped_ticket_backend_operation(
     Ok(Json(response))
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct WorkerMutationSource {
+    runtime_id: String,
+    worker_id: String,
+}
+
+fn ticket_mutation_target(operation: &TicketBackendOperation) -> Option<&TicketIdOrSlug> {
+    match operation {
+        TicketBackendOperation::EditItem { id, .. }
+        | TicketBackendOperation::AddEvent { id, .. }
+        | TicketBackendOperation::AddStateChanged { id, .. }
+        | TicketBackendOperation::AddIntakeSummary { id, .. }
+        | TicketBackendOperation::SetStateField { id, .. }
+        | TicketBackendOperation::SetWorkflowState { id, .. }
+        | TicketBackendOperation::MarkIntakeReady { id, .. }
+        | TicketBackendOperation::QueueReady { id, .. }
+        | TicketBackendOperation::Review { id, .. }
+        | TicketBackendOperation::Close { id, .. }
+        | TicketBackendOperation::AddTicketRelation { id, .. }
+        | TicketBackendOperation::AddOrchestrationPlanRecord { id, .. } => Some(id),
+        _ => None,
+    }
+}
+
+fn bind_worker_ticket_operation_source(
+    api: &WorkspaceApi,
+    workspace_id: &str,
+    source: &WorkerMutationSource,
+    target: &TicketIdOrSlug,
+    operation: &mut TicketBackendOperation,
+) -> Result<()> {
+    let mut author = format!("worker:{}/{}", source.runtime_id, source.worker_id);
+    if let TicketBackendOperation::AddEvent { event, .. } = operation {
+        if event.kind == TicketEventKind::ImplementationReport {
+            let target_query = match target {
+                TicketIdOrSlug::Id(value)
+                | TicketIdOrSlug::Slug(value)
+                | TicketIdOrSlug::Query(value) => value.as_str(),
+            };
+            let ticket = api.authority.ticket(target_query)?;
+            let assignment = api
+                .store
+                .get_current_ticket_worker_assignment(workspace_id, &ticket.id)?
+                .ok_or_else(|| {
+                    Error::TicketAssignmentConflict(format!(
+                        "Ticket {} has no current Worker assignment",
+                        ticket.id
+                    ))
+                })?;
+            if assignment.runtime_id != source.runtime_id
+                || assignment.worker_id != source.worker_id
+            {
+                return Err(Error::TicketAssignmentConflict(format!(
+                    "Worker {}/{} does not hold current assignment {} for Ticket {}",
+                    source.runtime_id, source.worker_id, assignment.assignment_id, ticket.id
+                )));
+            }
+            author.push_str(&format!(" assignment:{}", assignment.assignment_id));
+        }
+    }
+    match operation {
+        TicketBackendOperation::EditItem { edit, .. } => edit.author = Some(author),
+        TicketBackendOperation::AddEvent { event, .. } => event.author = Some(author),
+        TicketBackendOperation::AddStateChanged { change, .. } => change.author = Some(author),
+        TicketBackendOperation::AddIntakeSummary { summary, .. } => summary.author = Some(author),
+        TicketBackendOperation::QueueReady { queued_by, .. } => *queued_by = author,
+        TicketBackendOperation::Review { review, .. } => review.author = Some(author),
+        TicketBackendOperation::AddTicketRelation { relation, .. } => {
+            relation.author = Some(author)
+        }
+        TicketBackendOperation::AddOrchestrationPlanRecord { record, .. } => {
+            record.author = Some(author)
+        }
+        _ => {}
+    }
+    Ok(())
+}
+
+fn authenticate_worker_mutation_source(
+    api: &WorkspaceApi,
+    workspace_id: &str,
+    headers: &HeaderMap,
+) -> Result<WorkerMutationSource> {
+    let token = headers
+        .get(axum::http::header::AUTHORIZATION)
+        .and_then(|value| value.to_str().ok())
+        .and_then(|value| value.strip_prefix("Bearer "))
+        .filter(|value| !value.trim().is_empty())
+        .ok_or_else(|| {
+            Error::WorkerWorkspaceAuthentication("missing Runtime Workspace credential".to_string())
+        })?;
+    let worker_id = headers
+        .get("x-yoi-worker-id")
+        .and_then(|value| value.to_str().ok())
+        .filter(|value| !value.trim().is_empty())
+        .ok_or_else(|| {
+            Error::WorkerWorkspaceAuthentication("missing Runtime-bound Worker id".to_string())
+        })?;
+    let credential = api
+        .store
+        .authenticate_worker_workspace_credential(token, workspace_id, worker_id)?
+        .ok_or_else(|| {
+            Error::WorkerWorkspaceAuthentication("invalid Runtime Workspace credential".to_string())
+        })?;
+    api.runtime
+        .worker(&credential.runtime_id, worker_id)
+        .map_err(|_| {
+            Error::WorkerWorkspaceAuthentication(
+                "credential does not identify a current Runtime Worker".to_string(),
+            )
+        })?;
+    Ok(WorkerMutationSource {
+        runtime_id: credential.runtime_id,
+        worker_id: worker_id.to_string(),
+    })
+}
+
+fn enqueue_worker_ticket_notification(
+    api: &WorkspaceApi,
+    workspace_id: &str,
+    source: &WorkerMutationSource,
+    before: Option<&ticket::Ticket>,
+    after: &ticket::Ticket,
+) -> Result<()> {
+    let mut recipients = Vec::new();
+    if let Some(assignment) = api
+        .store
+        .get_current_ticket_worker_assignment(workspace_id, &after.meta.id)?
+    {
+        if assignment.runtime_id != source.runtime_id || assignment.worker_id != source.worker_id {
+            recipients.push(TicketNotificationRecipient {
+                runtime_id: assignment.runtime_id,
+                worker_id: assignment.worker_id,
+                recipient_kind: "assigned".to_string(),
+            });
+        }
+    }
+    let previous_state = before
+        .map(|ticket| ticket.meta.workflow_state.as_str().to_string())
+        .unwrap_or_else(|| after.meta.workflow_state.as_str().to_string());
+    let current_state = after.meta.workflow_state.as_str().to_string();
+    if matches!(previous_state.as_str(), "queued" | "inprogress")
+        || matches!(current_state.as_str(), "queued" | "inprogress")
+    {
+        if let Some(orchestrator) = find_workspace_orchestrator(api) {
+            if orchestrator.runtime_id != source.runtime_id
+                || orchestrator.worker_id != source.worker_id
+            {
+                recipients.push(TicketNotificationRecipient {
+                    runtime_id: orchestrator.runtime_id,
+                    worker_id: orchestrator.worker_id,
+                    recipient_kind: "orchestrator".to_string(),
+                });
+            }
+        }
+    }
+    recipients.sort_by(|left, right| {
+        (&left.runtime_id, &left.worker_id).cmp(&(&right.runtime_id, &right.worker_id))
+    });
+    recipients.dedup_by(|left, right| {
+        left.runtime_id == right.runtime_id && left.worker_id == right.worker_id
+    });
+    api.store.enqueue_ticket_notification(
+        &new_id("tnfy"),
+        workspace_id,
+        &after.meta.id,
+        after.events.len().saturating_sub(1) as i64,
+        &source.runtime_id,
+        &source.worker_id,
+        &previous_state,
+        &current_state,
+        &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
+        &recipients,
+    )
+}
+
+fn dispatch_pending_ticket_notifications(api: &WorkspaceApi, workspace_id: &str) {
+    let Ok(deliveries) = api
+        .store
+        .list_pending_ticket_notification_deliveries(workspace_id, 100)
+    else {
+        return;
+    };
+    for delivery in deliveries {
+        if !ticket_notification_recipient_is_current(api, &delivery) {
+            let _ = api.store.mark_ticket_notification_delivered(
+                &delivery.notification_id,
+                &delivery.recipient_runtime_id,
+                &delivery.recipient_worker_id,
+                &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
+            );
+            continue;
+        }
+        let result = api.runtime.send_input(
+            &delivery.recipient_runtime_id,
+            &delivery.recipient_worker_id,
+            WorkerInputRequest {
+                kind: WorkerInputKind::System,
+                content: format!(
+                    "Ticket notification: workspace_id={} ticket_id={} event_sequence={}. Reread the Ticket before acting.",
+                    delivery.workspace_id, delivery.ticket_id, delivery.event_sequence
+                ),
+                segments: None,
+            },
+        );
+        match result {
+            Ok(result) if result.state == WorkerOperationState::Accepted => {
+                let _ = api.store.mark_ticket_notification_delivered(
+                    &delivery.notification_id,
+                    &delivery.recipient_runtime_id,
+                    &delivery.recipient_worker_id,
+                    &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
+                );
+            }
+            Ok(result) => {
+                let _ = api.store.mark_ticket_notification_failed(
+                    &delivery.notification_id,
+                    &delivery.recipient_runtime_id,
+                    &delivery.recipient_worker_id,
+                    &format!("Runtime rejected notification: {:?}", result.diagnostics),
+                );
+            }
+            Err(error) => {
+                let _ = api.store.mark_ticket_notification_failed(
+                    &delivery.notification_id,
+                    &delivery.recipient_runtime_id,
+                    &delivery.recipient_worker_id,
+                    &error.into_error().to_string(),
+                );
+            }
+        }
+    }
+}
+
+fn find_workspace_orchestrator(api: &WorkspaceApi) -> Option<WorkerSummary> {
+    let is_orchestrator = |worker: &WorkerSummary| {
+        worker.singleton_key.as_deref() == Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY)
+    };
+    if let Some(worker) = api
+        .runtime
+        .list_workers(1000)
+        .items
+        .into_iter()
+        .find(is_orchestrator)
+    {
+        return Some(worker);
+    }
+    for runtime in api.runtime.list_runtimes(1000).items {
+        if let Ok(stopped) = api
+            .runtime
+            .list_stopped_workers_for_runtime(&runtime.runtime_id, 1000)
+        {
+            if let Some(worker) = stopped.items.into_iter().find(is_orchestrator) {
+                return Some(worker);
+            }
+        }
+    }
+    None
+}
+
+fn ticket_notification_recipient_is_current(
+    api: &WorkspaceApi,
+    delivery: &crate::store::TicketNotificationDeliveryRecord,
+) -> bool {
+    match delivery.recipient_kind.as_str() {
+        "assigned" => api
+            .store
+            .get_current_ticket_worker_assignment(&delivery.workspace_id, &delivery.ticket_id)
+            .ok()
+            .flatten()
+            .is_some_and(|assignment| {
+                assignment.runtime_id == delivery.recipient_runtime_id
+                    && assignment.worker_id == delivery.recipient_worker_id
+            }),
+        "orchestrator" => find_workspace_orchestrator(api).is_some_and(|worker| {
+            worker.runtime_id == delivery.recipient_runtime_id
+                && worker.worker_id == delivery.recipient_worker_id
+        }),
+        _ => false,
+    }
+}
+
 #[derive(Debug, Clone, Serialize)]
 struct MemoryDocumentResponse {
     body_md: String,
@@ -1903,6 +2372,7 @@ fn start_memory_staging_consolidation(
                 resolved_working_directory_request: None,
                 resolved_working_directory: None,
                 resolved_config_bundle,
+                resolved_workspace_api: None,
             },
         )
         .map_err(|err| err.into_error())?;
@@ -3145,7 +3615,14 @@ async fn scoped_restore_runtime_worker(
     AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
 ) -> ApiResult<Json<WorkerRestoreResponse>> {
     validate_workspace_scope(&api, &path.workspace_id)?;
-    restore_runtime_worker(State(api), AxumPath((path.runtime_id, path.worker_id))).await
+    let workspace_id = path.workspace_id.clone();
+    let response = restore_runtime_worker(
+        State(api.clone()),
+        AxumPath((path.runtime_id, path.worker_id)),
+    )
+    .await?;
+    dispatch_pending_ticket_notifications(&api, &workspace_id);
+    Ok(response)
 }
 
 async fn scoped_pin_runtime_worker(
@@ -4523,6 +5000,7 @@ async fn create_workspace_worker(
                 resolved_working_directory_request: None,
                 resolved_working_directory,
                 resolved_config_bundle,
+                resolved_workspace_api: None,
             },
         )
         .map_err(|err| err.into_error())?;
@@ -4844,6 +5322,23 @@ async fn create_runtime_worker(
             .map(|claim| claim.working_directory_id.clone())
     };
     let requested_worker_name = request.requested_worker_name.clone();
+    if let Some(base_url) = api.config.backend_base_url.clone() {
+        let credential = WorkerWorkspaceCredentialRecord {
+            credential_id: new_id("wac"),
+            token: mint_secret("wac"),
+            workspace_id: api.config.workspace_id.clone(),
+            runtime_id: runtime_id.clone(),
+            worker_id: None,
+            created_at: Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
+        };
+        api.store.upsert_worker_workspace_credential(&credential)?;
+        request.resolved_workspace_api = Some(worker_runtime::catalog::WorkspaceApiRef {
+            workspace_id: api.config.workspace_id.clone(),
+            base_url,
+            runtime_id: Some(runtime_id.clone()),
+            access_token: Some(credential.token),
+        });
+    }
     let result = api
         .runtime
         .spawn_worker(&runtime_id, request)
@@ -6945,6 +7440,8 @@ impl ApiError {
 impl IntoResponse for ApiError {
     fn into_response(self) -> Response {
         let status = match &self.error {
+            Error::TicketAssignmentConflict(_) => StatusCode::CONFLICT,
+            Error::WorkerWorkspaceAuthentication(_) => StatusCode::UNAUTHORIZED,
             Error::InvalidRuntimeIdentifier { .. } => StatusCode::BAD_REQUEST,
             Error::Ticket(ticket::TicketError::NotFound(_)) => StatusCode::NOT_FOUND,
             Error::Ticket(
@@ -7862,6 +8359,7 @@ mod tests {
                     resolved_working_directory_request: None,
                     resolved_working_directory: None,
                     resolved_config_bundle,
+                    resolved_workspace_api: None,
                 },
             )
             .unwrap();
@@ -7940,6 +8438,363 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn ticket_assignment_endpoints_read_and_clear_current_assignment() {
+        let dir = tempfile::tempdir().unwrap();
+        let api = test_api(dir.path()).await;
+        let Json(created) = scoped_ticket_backend_operation(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+            }),
+            HeaderMap::new(),
+            Json(TicketBackendOperation::Create {
+                input: ticket::NewTicket::new("Assigned Ticket"),
+            }),
+        )
+        .await
+        .unwrap();
+        let ticket_id = match created {
+            TicketBackendHttpResponse::Ok {
+                result: ticket::TicketBackendOperationResult::TicketRef(ticket_ref),
+            } => ticket_ref.id,
+            other => panic!("unexpected create response: {other:?}"),
+        };
+        let assignment = TicketWorkerAssignmentRecord {
+            workspace_id: TEST_WORKSPACE_ID.to_string(),
+            ticket_id: ticket_id.clone(),
+            assignment_id: "assignment-api-1".to_string(),
+            runtime_id: "embedded".to_string(),
+            worker_id: "42".to_string(),
+            assigned_by: "test-user".to_string(),
+            assigned_at: TEST_CREATED_AT.to_string(),
+        };
+        api.store
+            .set_current_ticket_worker_assignment(&assignment, None, "event-api-1")
+            .unwrap();
+        let path = || ScopedRecordPath {
+            workspace_id: TEST_WORKSPACE_ID.to_string(),
+            id: ticket_id.clone(),
+        };
+
+        let Json(read) = scoped_get_ticket_worker_assignment(State(api.clone()), AxumPath(path()))
+            .await
+            .unwrap();
+        assert_eq!(read.assignment, Some(assignment));
+
+        let stale = scoped_clear_ticket_worker_assignment(
+            State(api.clone()),
+            AxumPath(path()),
+            Query(ClearTicketWorkerAssignmentQuery {
+                expected_assignment_id: Some("stale-assignment".to_string()),
+                actor: Some("test-user".to_string()),
+            }),
+        )
+        .await
+        .unwrap_err()
+        .into_response();
+        assert_eq!(stale.status(), StatusCode::CONFLICT);
+
+        let Json(cleared) = scoped_clear_ticket_worker_assignment(
+            State(api.clone()),
+            AxumPath(path()),
+            Query(ClearTicketWorkerAssignmentQuery {
+                expected_assignment_id: Some("assignment-api-1".to_string()),
+                actor: Some("test-user".to_string()),
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            cleared.previous_assignment_id.as_deref(),
+            Some("assignment-api-1")
+        );
+        assert_eq!(cleared.assignment, None);
+    }
+
+    #[tokio::test]
+    async fn authenticated_worker_ticket_mutation_routes_durable_assignment_notification() {
+        let dir = tempfile::tempdir().unwrap();
+        let api = test_api(dir.path()).await;
+        let spawn = |name: &str| WorkerSpawnRequest {
+            requested_worker_name: Some(name.to_string()),
+            intent: WorkerSpawnIntent::TicketRole {
+                ticket_id: name.to_string(),
+                role: TicketWorkerRole::Coder,
+            },
+            acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
+                expected_segments: 0,
+            },
+            profile: ProfileSelector::Builtin("builtin:coder".to_string()),
+            initial_input: None,
+            working_directory_request: None,
+            resolved_working_directory_request: None,
+            resolved_working_directory: None,
+            resolved_config_bundle: None,
+            resolved_workspace_api: None,
+        };
+        let source_worker = api
+            .runtime
+            .spawn_worker(EMBEDDED_WORKER_RUNTIME_ID, spawn("source-worker"))
+            .unwrap()
+            .worker
+            .unwrap();
+        let recipient_worker = api
+            .runtime
+            .spawn_worker(EMBEDDED_WORKER_RUNTIME_ID, spawn("recipient-worker"))
+            .unwrap()
+            .worker
+            .unwrap();
+        let backend = browser_ticket_backend(&api).unwrap();
+        let ticket_ref = backend
+            .create(ticket::NewTicket::new("Notify assigned Worker"))
+            .unwrap();
+        api.store
+            .set_current_ticket_worker_assignment(
+                &TicketWorkerAssignmentRecord {
+                    workspace_id: TEST_WORKSPACE_ID.to_string(),
+                    ticket_id: ticket_ref.id.clone(),
+                    assignment_id: "notify-assignment".to_string(),
+                    runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
+                    worker_id: recipient_worker.worker_id.clone(),
+                    assigned_by: "test-user".to_string(),
+                    assigned_at: TEST_CREATED_AT.to_string(),
+                },
+                None,
+                "notify-assignment-event",
+            )
+            .unwrap();
+        api.store
+            .upsert_worker_workspace_credential(&WorkerWorkspaceCredentialRecord {
+                credential_id: "source-credential".to_string(),
+                token: "source-secret".to_string(),
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+                runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
+                worker_id: Some(source_worker.worker_id.clone()),
+                created_at: TEST_CREATED_AT.to_string(),
+            })
+            .unwrap();
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            axum::http::header::AUTHORIZATION,
+            axum::http::HeaderValue::from_static("Bearer source-secret"),
+        );
+        headers.insert(
+            "x-yoi-worker-id",
+            axum::http::HeaderValue::from_str(&source_worker.worker_id).unwrap(),
+        );
+        let Json(response) = scoped_ticket_backend_operation(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+            }),
+            headers.clone(),
+            Json(TicketBackendOperation::AddEvent {
+                id: ticket_ref.id.clone().into(),
+                event: NewTicketEvent::new(TicketEventKind::Comment, "implementation update"),
+            }),
+        )
+        .await
+        .unwrap();
+        assert!(matches!(response, TicketBackendHttpResponse::Ok { .. }));
+        assert!(
+            api.store
+                .list_pending_ticket_notification_deliveries(TEST_WORKSPACE_ID, 10)
+                .unwrap()
+                .is_empty(),
+            "accepted Runtime system input must complete the outbox delivery"
+        );
+
+        let stale_report = scoped_ticket_backend_operation(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+            }),
+            headers.clone(),
+            Json(TicketBackendOperation::AddEvent {
+                id: ticket_ref.id.clone().into(),
+                event: NewTicketEvent::new(
+                    TicketEventKind::ImplementationReport,
+                    "stale assignment report",
+                ),
+            }),
+        )
+        .await
+        .unwrap_err()
+        .into_response();
+        assert_eq!(stale_report.status(), StatusCode::CONFLICT);
+
+        api.store
+            .set_current_ticket_worker_assignment(
+                &TicketWorkerAssignmentRecord {
+                    workspace_id: TEST_WORKSPACE_ID.to_string(),
+                    ticket_id: ticket_ref.id.clone(),
+                    assignment_id: "source-assignment".to_string(),
+                    runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
+                    worker_id: source_worker.worker_id.clone(),
+                    assigned_by: "test-user".to_string(),
+                    assigned_at: TEST_CREATED_AT.to_string(),
+                },
+                Some("notify-assignment"),
+                "source-assignment-event",
+            )
+            .unwrap();
+        let _ = scoped_ticket_backend_operation(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+            }),
+            headers.clone(),
+            Json(TicketBackendOperation::AddEvent {
+                id: ticket_ref.id.clone().into(),
+                event: NewTicketEvent::new(
+                    TicketEventKind::ImplementationReport,
+                    "current assignment report",
+                ),
+            }),
+        )
+        .await
+        .unwrap();
+        let reported = backend.show(ticket_ref.id.clone().into()).unwrap();
+        assert!(
+            reported
+                .events
+                .last()
+                .unwrap()
+                .author
+                .as_deref()
+                .is_some_and(|author| {
+                    author.contains(&source_worker.worker_id)
+                        && author.contains("assignment:source-assignment")
+                })
+        );
+
+        let unauthorized = scoped_ticket_backend_operation(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+            }),
+            HeaderMap::new(),
+            Json(TicketBackendOperation::AddEvent {
+                id: ticket_ref.id.clone().into(),
+                event: NewTicketEvent::new(TicketEventKind::Comment, "spoofed update"),
+            }),
+        )
+        .await
+        .unwrap_err()
+        .into_response();
+        assert_eq!(unauthorized.status(), StatusCode::UNAUTHORIZED);
+    }
+
+    #[tokio::test]
+    async fn queued_ticket_mutation_targets_current_orchestrator() {
+        let dir = tempfile::tempdir().unwrap();
+        let api = test_api(dir.path()).await;
+        let source = api
+            .runtime
+            .spawn_worker(
+                EMBEDDED_WORKER_RUNTIME_ID,
+                WorkerSpawnRequest {
+                    requested_worker_name: Some("orchestrator-source".to_string()),
+                    intent: WorkerSpawnIntent::TicketRole {
+                        ticket_id: "source-ticket".to_string(),
+                        role: TicketWorkerRole::Coder,
+                    },
+                    acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
+                        expected_segments: 0,
+                    },
+                    profile: ProfileSelector::Builtin("builtin:coder".to_string()),
+                    initial_input: None,
+                    working_directory_request: None,
+                    resolved_working_directory_request: None,
+                    resolved_working_directory: None,
+                    resolved_config_bundle: None,
+                    resolved_workspace_api: None,
+                },
+            )
+            .unwrap()
+            .worker
+            .unwrap();
+        let orchestrator = api
+            .runtime
+            .spawn_worker(
+                EMBEDDED_WORKER_RUNTIME_ID,
+                WorkerSpawnRequest {
+                    requested_worker_name: Some("workspace-orchestrator".to_string()),
+                    intent: WorkerSpawnIntent::WorkspaceOrchestrator,
+                    acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
+                        expected_segments: 0,
+                    },
+                    profile: ProfileSelector::Builtin("builtin:orchestrator".to_string()),
+                    initial_input: None,
+                    working_directory_request: None,
+                    resolved_working_directory_request: None,
+                    resolved_working_directory: None,
+                    resolved_config_bundle: None,
+                    resolved_workspace_api: None,
+                },
+            )
+            .unwrap()
+            .worker
+            .unwrap();
+        api.runtime
+            .stop_worker(
+                EMBEDDED_WORKER_RUNTIME_ID,
+                &orchestrator.worker_id,
+                WorkerLifecycleRequest {
+                    reason: Some("test pending delivery".to_string()),
+                },
+            )
+            .unwrap();
+        api.store
+            .upsert_worker_workspace_credential(&WorkerWorkspaceCredentialRecord {
+                credential_id: "orchestrator-source-credential".to_string(),
+                token: "orchestrator-source-secret".to_string(),
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+                runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
+                worker_id: Some(source.worker_id.clone()),
+                created_at: TEST_CREATED_AT.to_string(),
+            })
+            .unwrap();
+        let backend = browser_ticket_backend(&api).unwrap();
+        let mut input = ticket::NewTicket::new("Queued notification");
+        input.workflow_state = Some(TicketWorkflowState::Queued);
+        let ticket_ref = backend.create(input).unwrap();
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            axum::http::header::AUTHORIZATION,
+            axum::http::HeaderValue::from_static("Bearer orchestrator-source-secret"),
+        );
+        headers.insert(
+            "x-yoi-worker-id",
+            axum::http::HeaderValue::from_str(&source.worker_id).unwrap(),
+        );
+        let _ = scoped_ticket_backend_operation(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: TEST_WORKSPACE_ID.to_string(),
+            }),
+            headers,
+            Json(TicketBackendOperation::AddEvent {
+                id: ticket_ref.id.clone().into(),
+                event: NewTicketEvent::new(TicketEventKind::Comment, "queued update"),
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            api.store
+                .count_ticket_notification_deliveries_for_recipient(
+                    TEST_WORKSPACE_ID,
+                    &ticket_ref.id,
+                    EMBEDDED_WORKER_RUNTIME_ID,
+                    &orchestrator.worker_id,
+                )
+                .unwrap(),
+            1
+        );
+    }
+
     #[tokio::test]
     async fn ticket_browser_endpoints_mutate_typed_backend_and_return_thread() {
         let dir = tempfile::tempdir().unwrap();
@@ -7949,6 +8804,7 @@ mod tests {
             AxumPath(ScopedWorkspacePath {
                 workspace_id: TEST_WORKSPACE_ID.to_string(),
             }),
+            HeaderMap::new(),
             Json(TicketBackendOperation::Create {
                 input: ticket::NewTicket::new("Browser Ticket API"),
             }),
@@ -7970,6 +8826,7 @@ mod tests {
             AxumPath(ScopedWorkspacePath {
                 workspace_id: TEST_WORKSPACE_ID.to_string(),
             }),
+            HeaderMap::new(),
             Json(TicketBackendOperation::Create {
                 input: ticket::NewTicket::new("Related Browser Ticket"),
             }),
@@ -7982,23 +8839,18 @@ mod tests {
             } => ticket_ref.id,
             other => panic!("unexpected related create response: {other:?}"),
         };
-        let _ = scoped_ticket_backend_operation(
-            State(api.clone()),
-            AxumPath(ScopedWorkspacePath {
-                workspace_id: TEST_WORKSPACE_ID.to_string(),
-            }),
-            Json(TicketBackendOperation::AddTicketRelation {
-                id: ticket_id.clone().into(),
-                relation: ticket::NewTicketRelation {
+        browser_ticket_backend(&api)
+            .unwrap()
+            .add_ticket_relation(
+                ticket_id.clone().into(),
+                ticket::NewTicketRelation {
                     kind: ticket::TicketRelationKind::Related,
                     target: related_ticket_id.clone(),
                     note: Some("Browser relation".to_string()),
                     author: Some("browser-user".to_string()),
                 },
-            }),
-        )
-        .await
-        .unwrap();
+            )
+            .unwrap();
 
         let Json(edited) = scoped_edit_ticket_item(
             State(api.clone()),
@@ -8117,6 +8969,7 @@ mod tests {
             AxumPath(ScopedWorkspacePath {
                 workspace_id: TEST_WORKSPACE_ID.to_string(),
             }),
+            HeaderMap::new(),
             Json(TicketBackendOperation::Create {
                 input: ticket::NewTicket::new("Endpoint configured root"),
             }),
@@ -9712,6 +10565,7 @@ mod tests {
                     resolved_working_directory_request: None,
                     resolved_working_directory: None,
                     resolved_config_bundle: None,
+                    resolved_workspace_api: None,
                 },
             )
             .expect("spawn worker");
diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs
index 65ae6711..ea3ea892 100644
--- a/crates/workspace-server/src/store.rs
+++ b/crates/workspace-server/src/store.rs
@@ -87,6 +87,16 @@ const MIGRATIONS: &[Migration] = &[
         name: "remove unused control-plane Ticket tables",
         apply: remove_unused_control_plane_ticket_tables,
     },
+    Migration {
+        version: 15,
+        name: "ticket worker current assignment authority",
+        apply: create_ticket_worker_assignment_tables,
+    },
+    Migration {
+        version: 16,
+        name: "worker workspace credentials and Ticket notification outbox",
+        apply: create_ticket_notification_tables,
+    },
 ];
 
 struct Migration {
@@ -232,6 +242,66 @@ pub struct WorkerRegistryRecord {
     pub updated_at: String,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct TicketWorkerAssignmentRecord {
+    pub workspace_id: String,
+    pub ticket_id: String,
+    pub assignment_id: String,
+    pub runtime_id: String,
+    pub worker_id: String,
+    pub assigned_by: String,
+    pub assigned_at: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct TicketWorkerAssignmentEventRecord {
+    pub workspace_id: String,
+    pub ticket_id: String,
+    pub event_id: String,
+    pub action: String,
+    pub assignment_id: Option<String>,
+    pub previous_assignment_id: Option<String>,
+    pub actor: String,
+    pub created_at: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct TicketWorkerAssignmentUpdate {
+    pub current: TicketWorkerAssignmentRecord,
+    pub previous: Option<TicketWorkerAssignmentRecord>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct WorkerWorkspaceCredentialRecord {
+    pub credential_id: String,
+    pub token: String,
+    pub workspace_id: String,
+    pub runtime_id: String,
+    pub worker_id: Option<String>,
+    pub created_at: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct TicketNotificationRecipient {
+    pub runtime_id: String,
+    pub worker_id: String,
+    pub recipient_kind: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct TicketNotificationDeliveryRecord {
+    pub notification_id: String,
+    pub workspace_id: String,
+    pub ticket_id: String,
+    pub event_sequence: i64,
+    pub source_runtime_id: String,
+    pub source_worker_id: String,
+    pub recipient_runtime_id: String,
+    pub recipient_worker_id: String,
+    pub recipient_kind: String,
+    pub attempts: i64,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct WorkdirRegistryRecord {
     pub workspace_id: String,
@@ -483,6 +553,83 @@ pub trait ControlPlaneStore: Send + Sync {
         runtime_worker_id: u64,
     ) -> Result<bool>;
 
+    fn get_current_ticket_worker_assignment(
+        &self,
+        workspace_id: &str,
+        ticket_id: &str,
+    ) -> Result<Option<TicketWorkerAssignmentRecord>>;
+    fn set_current_ticket_worker_assignment(
+        &self,
+        record: &TicketWorkerAssignmentRecord,
+        expected_assignment_id: Option<&str>,
+        event_id: &str,
+    ) -> Result<TicketWorkerAssignmentUpdate>;
+    fn clear_current_ticket_worker_assignment(
+        &self,
+        workspace_id: &str,
+        ticket_id: &str,
+        expected_assignment_id: Option<&str>,
+        event_id: &str,
+        actor: &str,
+        created_at: &str,
+    ) -> Result<Option<TicketWorkerAssignmentRecord>>;
+    fn list_ticket_worker_assignment_events(
+        &self,
+        workspace_id: &str,
+        ticket_id: &str,
+        limit: usize,
+    ) -> Result<Vec<TicketWorkerAssignmentEventRecord>>;
+
+    fn upsert_worker_workspace_credential(
+        &self,
+        record: &WorkerWorkspaceCredentialRecord,
+    ) -> Result<()>;
+    fn authenticate_worker_workspace_credential(
+        &self,
+        token: &str,
+        workspace_id: &str,
+        worker_id: &str,
+    ) -> Result<Option<WorkerWorkspaceCredentialRecord>>;
+    fn enqueue_ticket_notification(
+        &self,
+        notification_id: &str,
+        workspace_id: &str,
+        ticket_id: &str,
+        event_sequence: i64,
+        source_runtime_id: &str,
+        source_worker_id: &str,
+        previous_state: &str,
+        current_state: &str,
+        created_at: &str,
+        recipients: &[TicketNotificationRecipient],
+    ) -> Result<()>;
+    fn list_pending_ticket_notification_deliveries(
+        &self,
+        workspace_id: &str,
+        limit: usize,
+    ) -> Result<Vec<TicketNotificationDeliveryRecord>>;
+    fn count_ticket_notification_deliveries_for_recipient(
+        &self,
+        workspace_id: &str,
+        ticket_id: &str,
+        runtime_id: &str,
+        worker_id: &str,
+    ) -> Result<usize>;
+    fn mark_ticket_notification_delivered(
+        &self,
+        notification_id: &str,
+        recipient_runtime_id: &str,
+        recipient_worker_id: &str,
+        delivered_at: &str,
+    ) -> Result<()>;
+    fn mark_ticket_notification_failed(
+        &self,
+        notification_id: &str,
+        recipient_runtime_id: &str,
+        recipient_worker_id: &str,
+        error: &str,
+    ) -> Result<()>;
+
     fn upsert_workdir_registry(&self, record: &WorkdirRegistryRecord) -> Result<()>;
     fn get_workdir_registry(
         &self,
@@ -1566,6 +1713,388 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         })
     }
 
+    fn get_current_ticket_worker_assignment(
+        &self,
+        workspace_id: &str,
+        ticket_id: &str,
+    ) -> Result<Option<TicketWorkerAssignmentRecord>> {
+        self.with_conn(|conn| {
+            conn.query_row(
+                current_ticket_worker_assignment_select_sql().as_str(),
+                params![workspace_id, ticket_id],
+                read_ticket_worker_assignment_record,
+            )
+            .optional()
+            .map_err(Error::from)
+        })
+    }
+
+    fn set_current_ticket_worker_assignment(
+        &self,
+        record: &TicketWorkerAssignmentRecord,
+        expected_assignment_id: Option<&str>,
+        event_id: &str,
+    ) -> Result<TicketWorkerAssignmentUpdate> {
+        self.with_conn(|conn| {
+            let tx = conn.unchecked_transaction()?;
+            let previous = tx
+                .query_row(
+                    current_ticket_worker_assignment_select_sql().as_str(),
+                    params![record.workspace_id, record.ticket_id],
+                    read_ticket_worker_assignment_record,
+                )
+                .optional()?;
+            require_expected_ticket_assignment(
+                record.ticket_id.as_str(),
+                previous.as_ref(),
+                expected_assignment_id,
+            )?;
+            tx.execute(
+                r#"INSERT INTO ticket_worker_assignments (
+                    workspace_id, ticket_id, assignment_id, runtime_id, worker_id,
+                    assigned_by, assigned_at
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)"#,
+                params![
+                    record.workspace_id,
+                    record.ticket_id,
+                    record.assignment_id,
+                    record.runtime_id,
+                    record.worker_id,
+                    record.assigned_by,
+                    record.assigned_at,
+                ],
+            )?;
+            tx.execute(
+                r#"INSERT INTO ticket_current_worker_assignments (
+                    workspace_id, ticket_id, assignment_id, updated_at
+                ) VALUES (?1, ?2, ?3, ?4)
+                ON CONFLICT(workspace_id, ticket_id) DO UPDATE SET
+                    assignment_id = excluded.assignment_id,
+                    updated_at = excluded.updated_at"#,
+                params![
+                    record.workspace_id,
+                    record.ticket_id,
+                    record.assignment_id,
+                    record.assigned_at,
+                ],
+            )?;
+            tx.execute(
+                r#"INSERT INTO ticket_worker_assignment_events (
+                    workspace_id, ticket_id, event_id, action, assignment_id,
+                    previous_assignment_id, actor, created_at
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)"#,
+                params![
+                    record.workspace_id,
+                    record.ticket_id,
+                    event_id,
+                    if previous.is_some() {
+                        "reassigned"
+                    } else {
+                        "assigned"
+                    },
+                    record.assignment_id,
+                    previous
+                        .as_ref()
+                        .map(|assignment| assignment.assignment_id.as_str()),
+                    record.assigned_by,
+                    record.assigned_at,
+                ],
+            )?;
+            tx.commit()?;
+            Ok(TicketWorkerAssignmentUpdate {
+                current: record.clone(),
+                previous,
+            })
+        })
+    }
+
+    fn clear_current_ticket_worker_assignment(
+        &self,
+        workspace_id: &str,
+        ticket_id: &str,
+        expected_assignment_id: Option<&str>,
+        event_id: &str,
+        actor: &str,
+        created_at: &str,
+    ) -> Result<Option<TicketWorkerAssignmentRecord>> {
+        self.with_conn(|conn| {
+            let tx = conn.unchecked_transaction()?;
+            let previous = tx
+                .query_row(
+                    current_ticket_worker_assignment_select_sql().as_str(),
+                    params![workspace_id, ticket_id],
+                    read_ticket_worker_assignment_record,
+                )
+                .optional()?;
+            require_expected_ticket_assignment(ticket_id, previous.as_ref(), expected_assignment_id)?;
+            let Some(previous) = previous else {
+                tx.commit()?;
+                return Ok(None);
+            };
+            tx.execute(
+                "DELETE FROM ticket_current_worker_assignments WHERE workspace_id = ?1 AND ticket_id = ?2",
+                params![workspace_id, ticket_id],
+            )?;
+            tx.execute(
+                r#"INSERT INTO ticket_worker_assignment_events (
+                    workspace_id, ticket_id, event_id, action, assignment_id,
+                    previous_assignment_id, actor, created_at
+                ) VALUES (?1, ?2, ?3, 'unassigned', NULL, ?4, ?5, ?6)"#,
+                params![
+                    workspace_id,
+                    ticket_id,
+                    event_id,
+                    previous.assignment_id,
+                    actor,
+                    created_at,
+                ],
+            )?;
+            tx.commit()?;
+            Ok(Some(previous))
+        })
+    }
+
+    fn list_ticket_worker_assignment_events(
+        &self,
+        workspace_id: &str,
+        ticket_id: &str,
+        limit: usize,
+    ) -> Result<Vec<TicketWorkerAssignmentEventRecord>> {
+        self.with_conn(|conn| {
+            let mut stmt = conn.prepare(
+                r#"SELECT workspace_id, ticket_id, event_id, action, assignment_id,
+                          previous_assignment_id, actor, created_at
+                   FROM ticket_worker_assignment_events
+                   WHERE workspace_id = ?1 AND ticket_id = ?2
+                   ORDER BY created_at DESC, event_id DESC
+                   LIMIT ?3"#,
+            )?;
+            let rows = stmt.query_map(
+                params![workspace_id, ticket_id, limit as i64],
+                read_ticket_worker_assignment_event_record,
+            )?;
+            rows.collect::<std::result::Result<Vec<_>, _>>()
+                .map_err(Error::from)
+        })
+    }
+
+    fn upsert_worker_workspace_credential(
+        &self,
+        record: &WorkerWorkspaceCredentialRecord,
+    ) -> Result<()> {
+        self.with_conn(|conn| {
+            conn.execute(
+                r#"INSERT INTO worker_workspace_credentials (
+                    credential_id, token, workspace_id, runtime_id, worker_id, created_at
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
+                ON CONFLICT(credential_id) DO UPDATE SET
+                    token = excluded.token,
+                    workspace_id = excluded.workspace_id,
+                    runtime_id = excluded.runtime_id,
+                    worker_id = excluded.worker_id,
+                    created_at = excluded.created_at"#,
+                params![
+                    record.credential_id,
+                    record.token,
+                    record.workspace_id,
+                    record.runtime_id,
+                    record.worker_id,
+                    record.created_at,
+                ],
+            )?;
+            Ok(())
+        })
+    }
+
+    fn authenticate_worker_workspace_credential(
+        &self,
+        token: &str,
+        workspace_id: &str,
+        worker_id: &str,
+    ) -> Result<Option<WorkerWorkspaceCredentialRecord>> {
+        self.with_conn(|conn| {
+            let tx = conn.unchecked_transaction()?;
+            let record = tx
+                .query_row(
+                    r#"SELECT credential_id, token, workspace_id, runtime_id, worker_id, created_at
+                       FROM worker_workspace_credentials
+                       WHERE token = ?1 AND workspace_id = ?2"#,
+                    params![token, workspace_id],
+                    |row| {
+                        Ok(WorkerWorkspaceCredentialRecord {
+                            credential_id: row.get(0)?,
+                            token: row.get(1)?,
+                            workspace_id: row.get(2)?,
+                            runtime_id: row.get(3)?,
+                            worker_id: row.get(4)?,
+                            created_at: row.get(5)?,
+                        })
+                    },
+                )
+                .optional()?;
+            let Some(mut record) = record else {
+                tx.commit()?;
+                return Ok(None);
+            };
+            if record.worker_id.as_deref().is_some_and(|bound| bound != worker_id) {
+                tx.commit()?;
+                return Ok(None);
+            }
+            if record.worker_id.is_none() {
+                tx.execute(
+                    "UPDATE worker_workspace_credentials SET worker_id = ?1 WHERE credential_id = ?2 AND worker_id IS NULL",
+                    params![worker_id, record.credential_id],
+                )?;
+                record.worker_id = Some(worker_id.to_string());
+            }
+            tx.commit()?;
+            Ok(Some(record))
+        })
+    }
+
+    fn enqueue_ticket_notification(
+        &self,
+        notification_id: &str,
+        workspace_id: &str,
+        ticket_id: &str,
+        event_sequence: i64,
+        source_runtime_id: &str,
+        source_worker_id: &str,
+        previous_state: &str,
+        current_state: &str,
+        created_at: &str,
+        recipients: &[TicketNotificationRecipient],
+    ) -> Result<()> {
+        self.with_conn(|conn| {
+            let tx = conn.unchecked_transaction()?;
+            tx.execute(
+                r#"INSERT INTO ticket_notification_outbox (
+                    notification_id, workspace_id, ticket_id, event_sequence,
+                    source_runtime_id, source_worker_id, previous_state, current_state, created_at
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)"#,
+                params![
+                    notification_id,
+                    workspace_id,
+                    ticket_id,
+                    event_sequence,
+                    source_runtime_id,
+                    source_worker_id,
+                    previous_state,
+                    current_state,
+                    created_at,
+                ],
+            )?;
+            for recipient in recipients {
+                tx.execute(
+                    r#"INSERT OR IGNORE INTO ticket_notification_deliveries (
+                        notification_id, recipient_runtime_id, recipient_worker_id,
+                        recipient_kind, attempts
+                    ) VALUES (?1, ?2, ?3, ?4, 0)"#,
+                    params![
+                        notification_id,
+                        recipient.runtime_id,
+                        recipient.worker_id,
+                        recipient.recipient_kind,
+                    ],
+                )?;
+            }
+            tx.commit()?;
+            Ok(())
+        })
+    }
+
+    fn list_pending_ticket_notification_deliveries(
+        &self,
+        workspace_id: &str,
+        limit: usize,
+    ) -> Result<Vec<TicketNotificationDeliveryRecord>> {
+        self.with_conn(|conn| {
+            let mut stmt = conn.prepare(
+                r#"SELECT o.notification_id, o.workspace_id, o.ticket_id, o.event_sequence,
+                          o.source_runtime_id, o.source_worker_id,
+                          d.recipient_runtime_id, d.recipient_worker_id, d.recipient_kind, d.attempts
+                   FROM ticket_notification_deliveries AS d
+                   JOIN ticket_notification_outbox AS o ON o.notification_id = d.notification_id
+                   WHERE o.workspace_id = ?1 AND d.delivered_at IS NULL
+                   ORDER BY o.created_at ASC, o.notification_id ASC
+                   LIMIT ?2"#,
+            )?;
+            let rows = stmt.query_map(params![workspace_id, limit as i64], |row| {
+                Ok(TicketNotificationDeliveryRecord {
+                    notification_id: row.get(0)?,
+                    workspace_id: row.get(1)?,
+                    ticket_id: row.get(2)?,
+                    event_sequence: row.get(3)?,
+                    source_runtime_id: row.get(4)?,
+                    source_worker_id: row.get(5)?,
+                    recipient_runtime_id: row.get(6)?,
+                    recipient_worker_id: row.get(7)?,
+                    recipient_kind: row.get(8)?,
+                    attempts: row.get(9)?,
+                })
+            })?;
+            rows.collect::<std::result::Result<Vec<_>, _>>()
+                .map_err(Error::from)
+        })
+    }
+
+    fn count_ticket_notification_deliveries_for_recipient(
+        &self,
+        workspace_id: &str,
+        ticket_id: &str,
+        runtime_id: &str,
+        worker_id: &str,
+    ) -> Result<usize> {
+        self.with_conn(|conn| {
+            let count = conn.query_row(
+                r#"SELECT COUNT(*)
+                   FROM ticket_notification_deliveries AS d
+                   JOIN ticket_notification_outbox AS o ON o.notification_id = d.notification_id
+                   WHERE o.workspace_id = ?1 AND o.ticket_id = ?2
+                     AND d.recipient_runtime_id = ?3 AND d.recipient_worker_id = ?4"#,
+                params![workspace_id, ticket_id, runtime_id, worker_id],
+                |row| row.get::<_, i64>(0),
+            )?;
+            Ok(count as usize)
+        })
+    }
+
+    fn mark_ticket_notification_delivered(
+        &self,
+        notification_id: &str,
+        recipient_runtime_id: &str,
+        recipient_worker_id: &str,
+        delivered_at: &str,
+    ) -> Result<()> {
+        self.with_conn(|conn| {
+            conn.execute(
+                r#"UPDATE ticket_notification_deliveries
+                   SET delivered_at = ?4, last_error = NULL, attempts = attempts + 1
+                   WHERE notification_id = ?1 AND recipient_runtime_id = ?2 AND recipient_worker_id = ?3"#,
+                params![notification_id, recipient_runtime_id, recipient_worker_id, delivered_at],
+            )?;
+            Ok(())
+        })
+    }
+
+    fn mark_ticket_notification_failed(
+        &self,
+        notification_id: &str,
+        recipient_runtime_id: &str,
+        recipient_worker_id: &str,
+        error: &str,
+    ) -> Result<()> {
+        self.with_conn(|conn| {
+            conn.execute(
+                r#"UPDATE ticket_notification_deliveries
+                   SET last_error = ?4, attempts = attempts + 1
+                   WHERE notification_id = ?1 AND recipient_runtime_id = ?2 AND recipient_worker_id = ?3"#,
+                params![notification_id, recipient_runtime_id, recipient_worker_id, error],
+            )?;
+            Ok(())
+        })
+    }
+
     fn upsert_workdir_registry(&self, record: &WorkdirRegistryRecord) -> Result<()> {
         self.with_conn(|conn| {
             conn.execute(
@@ -1996,6 +2525,63 @@ fn read_worker_registry_record(row: &rusqlite::Row<'_>) -> rusqlite::Result<Work
     })
 }
 
+fn current_ticket_worker_assignment_select_sql() -> String {
+    "SELECT a.workspace_id, a.ticket_id, a.assignment_id, a.runtime_id, a.worker_id, \
+            a.assigned_by, a.assigned_at \
+     FROM ticket_current_worker_assignments AS current \
+     JOIN ticket_worker_assignments AS a \
+       ON a.workspace_id = current.workspace_id \
+      AND a.ticket_id = current.ticket_id \
+      AND a.assignment_id = current.assignment_id \
+     WHERE current.workspace_id = ?1 AND current.ticket_id = ?2"
+        .to_owned()
+}
+
+fn read_ticket_worker_assignment_record(
+    row: &rusqlite::Row<'_>,
+) -> rusqlite::Result<TicketWorkerAssignmentRecord> {
+    Ok(TicketWorkerAssignmentRecord {
+        workspace_id: row.get(0)?,
+        ticket_id: row.get(1)?,
+        assignment_id: row.get(2)?,
+        runtime_id: row.get(3)?,
+        worker_id: row.get(4)?,
+        assigned_by: row.get(5)?,
+        assigned_at: row.get(6)?,
+    })
+}
+
+fn read_ticket_worker_assignment_event_record(
+    row: &rusqlite::Row<'_>,
+) -> rusqlite::Result<TicketWorkerAssignmentEventRecord> {
+    Ok(TicketWorkerAssignmentEventRecord {
+        workspace_id: row.get(0)?,
+        ticket_id: row.get(1)?,
+        event_id: row.get(2)?,
+        action: row.get(3)?,
+        assignment_id: row.get(4)?,
+        previous_assignment_id: row.get(5)?,
+        actor: row.get(6)?,
+        created_at: row.get(7)?,
+    })
+}
+
+fn require_expected_ticket_assignment(
+    ticket_id: &str,
+    current: Option<&TicketWorkerAssignmentRecord>,
+    expected_assignment_id: Option<&str>,
+) -> Result<()> {
+    let Some(expected_assignment_id) = expected_assignment_id else {
+        return Ok(());
+    };
+    if current.map(|assignment| assignment.assignment_id.as_str()) == Some(expected_assignment_id) {
+        return Ok(());
+    }
+    Err(Error::TicketAssignmentConflict(format!(
+        "Ticket {ticket_id} is no longer assigned to {expected_assignment_id}"
+    )))
+}
+
 fn workdir_registry_select_sql(where_clause: &str) -> String {
     format!(
         "SELECT workspace_id, workdir_id, runtime_id, repository_id, selector, resolved_commit, \
@@ -2175,6 +2761,95 @@ DROP TABLE IF EXISTS tickets;
     Ok(())
 }
 
+fn create_ticket_worker_assignment_tables(conn: &Connection) -> Result<()> {
+    conn.execute_batch(
+        r#"
+CREATE TABLE IF NOT EXISTS ticket_worker_assignments (
+    workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    ticket_id TEXT NOT NULL,
+    assignment_id TEXT NOT NULL,
+    runtime_id TEXT NOT NULL,
+    worker_id TEXT NOT NULL,
+    assigned_by TEXT NOT NULL,
+    assigned_at TEXT NOT NULL,
+    PRIMARY KEY (workspace_id, assignment_id),
+    UNIQUE (workspace_id, ticket_id, assignment_id)
+);
+
+CREATE TABLE IF NOT EXISTS ticket_current_worker_assignments (
+    workspace_id TEXT NOT NULL,
+    ticket_id TEXT NOT NULL,
+    assignment_id TEXT NOT NULL,
+    updated_at TEXT NOT NULL,
+    PRIMARY KEY (workspace_id, ticket_id),
+    FOREIGN KEY (workspace_id, ticket_id, assignment_id)
+        REFERENCES ticket_worker_assignments(workspace_id, ticket_id, assignment_id)
+        ON DELETE CASCADE
+);
+
+CREATE TABLE IF NOT EXISTS ticket_worker_assignment_events (
+    workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    ticket_id TEXT NOT NULL,
+    event_id TEXT NOT NULL,
+    action TEXT NOT NULL CHECK (action IN ('assigned', 'reassigned', 'unassigned')),
+    assignment_id TEXT,
+    previous_assignment_id TEXT,
+    actor TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    PRIMARY KEY (workspace_id, event_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_ticket_assignments_worker
+    ON ticket_worker_assignments(workspace_id, runtime_id, worker_id, assigned_at DESC);
+CREATE INDEX IF NOT EXISTS idx_ticket_assignment_events_ticket
+    ON ticket_worker_assignment_events(workspace_id, ticket_id, created_at DESC);
+"#,
+    )?;
+    Ok(())
+}
+
+fn create_ticket_notification_tables(conn: &Connection) -> Result<()> {
+    conn.execute_batch(
+        r#"
+CREATE TABLE IF NOT EXISTS worker_workspace_credentials (
+    credential_id TEXT PRIMARY KEY,
+    token TEXT NOT NULL UNIQUE,
+    workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    runtime_id TEXT NOT NULL,
+    worker_id TEXT,
+    created_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS ticket_notification_outbox (
+    notification_id TEXT PRIMARY KEY,
+    workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    ticket_id TEXT NOT NULL,
+    event_sequence INTEGER NOT NULL,
+    source_runtime_id TEXT NOT NULL,
+    source_worker_id TEXT NOT NULL,
+    previous_state TEXT NOT NULL,
+    current_state TEXT NOT NULL,
+    created_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS ticket_notification_deliveries (
+    notification_id TEXT NOT NULL REFERENCES ticket_notification_outbox(notification_id) ON DELETE CASCADE,
+    recipient_runtime_id TEXT NOT NULL,
+    recipient_worker_id TEXT NOT NULL,
+    recipient_kind TEXT NOT NULL CHECK (recipient_kind IN ('assigned', 'orchestrator')),
+    attempts INTEGER NOT NULL DEFAULT 0,
+    delivered_at TEXT,
+    last_error TEXT,
+    PRIMARY KEY (notification_id, recipient_runtime_id, recipient_worker_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_ticket_notification_pending
+    ON ticket_notification_deliveries(delivered_at, attempts);
+"#,
+    )?;
+    Ok(())
+}
+
 fn create_objective_event_tables(conn: &Connection) -> Result<()> {
     conn.execute_batch(
         r#"
@@ -2857,7 +3532,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
         let db = dir.path().join("control-plane.sqlite");
         let store = SqliteWorkspaceStore::open(&db).unwrap();
 
-        assert_eq!(store.schema_version().await.unwrap(), 14);
+        assert_eq!(store.schema_version().await.unwrap(), 16);
 
         let record = WorkspaceRecord {
             workspace_id: "local-dev".to_string(),
@@ -2870,13 +3545,191 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
         store.upsert_workspace(&record).await.unwrap();
 
         let reopened = SqliteWorkspaceStore::open(&db).unwrap();
-        assert_eq!(reopened.schema_version().await.unwrap(), 14);
+        assert_eq!(reopened.schema_version().await.unwrap(), 16);
         assert_eq!(
             reopened.get_workspace("local-dev").await.unwrap(),
             Some(record)
         );
     }
 
+    #[tokio::test]
+    async fn ticket_worker_assignment_replaces_current_and_preserves_audit_history() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap();
+        store
+            .upsert_workspace(&WorkspaceRecord {
+                workspace_id: "workspace-a".to_string(),
+                owner_account_id: None,
+                display_name: "Workspace A".to_string(),
+                state: "active".to_string(),
+                created_at: "2026-07-31T00:00:00Z".to_string(),
+                updated_at: "2026-07-31T00:00:00Z".to_string(),
+            })
+            .await
+            .unwrap();
+
+        let first = TicketWorkerAssignmentRecord {
+            workspace_id: "workspace-a".to_string(),
+            ticket_id: "ticket-1".to_string(),
+            assignment_id: "assignment-1".to_string(),
+            runtime_id: "runtime-1".to_string(),
+            worker_id: "worker-1".to_string(),
+            assigned_by: "user-1".to_string(),
+            assigned_at: "2026-07-31T00:00:01Z".to_string(),
+        };
+        let created = store
+            .set_current_ticket_worker_assignment(&first, None, "event-1")
+            .unwrap();
+        assert_eq!(created.current, first);
+        assert_eq!(created.previous, None);
+
+        let second = TicketWorkerAssignmentRecord {
+            assignment_id: "assignment-2".to_string(),
+            runtime_id: "runtime-2".to_string(),
+            worker_id: "worker-2".to_string(),
+            assigned_by: "user-2".to_string(),
+            assigned_at: "2026-07-31T00:00:02Z".to_string(),
+            ..first.clone()
+        };
+        let replaced = store
+            .set_current_ticket_worker_assignment(&second, Some("assignment-1"), "event-2")
+            .unwrap();
+        assert_eq!(replaced.current, second);
+        assert_eq!(replaced.previous, Some(first.clone()));
+        assert_eq!(
+            store
+                .get_current_ticket_worker_assignment("workspace-a", "ticket-1")
+                .unwrap(),
+            Some(second.clone())
+        );
+
+        let stale = store
+            .clear_current_ticket_worker_assignment(
+                "workspace-a",
+                "ticket-1",
+                Some("assignment-1"),
+                "event-stale",
+                "user-1",
+                "2026-07-31T00:00:03Z",
+            )
+            .unwrap_err();
+        assert!(matches!(stale, Error::TicketAssignmentConflict(_)));
+
+        let cleared = store
+            .clear_current_ticket_worker_assignment(
+                "workspace-a",
+                "ticket-1",
+                Some("assignment-2"),
+                "event-3",
+                "user-2",
+                "2026-07-31T00:00:03Z",
+            )
+            .unwrap();
+        assert_eq!(cleared, Some(second));
+        assert_eq!(
+            store
+                .get_current_ticket_worker_assignment("workspace-a", "ticket-1")
+                .unwrap(),
+            None
+        );
+
+        let events = store
+            .list_ticket_worker_assignment_events("workspace-a", "ticket-1", 10)
+            .unwrap();
+        assert_eq!(
+            events
+                .iter()
+                .map(|event| event.action.as_str())
+                .collect::<Vec<_>>(),
+            vec!["unassigned", "reassigned", "assigned"]
+        );
+        assert_eq!(events[1].assignment_id.as_deref(), Some("assignment-2"));
+        assert_eq!(
+            events[1].previous_assignment_id.as_deref(),
+            Some("assignment-1")
+        );
+    }
+
+    #[tokio::test]
+    async fn worker_credential_binds_once_and_notification_outbox_is_durable() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap();
+        store
+            .upsert_workspace(&WorkspaceRecord {
+                workspace_id: "workspace-a".to_string(),
+                owner_account_id: None,
+                display_name: "Workspace A".to_string(),
+                state: "active".to_string(),
+                created_at: "2026-07-31T00:00:00Z".to_string(),
+                updated_at: "2026-07-31T00:00:00Z".to_string(),
+            })
+            .await
+            .unwrap();
+        store
+            .upsert_worker_workspace_credential(&WorkerWorkspaceCredentialRecord {
+                credential_id: "credential-1".to_string(),
+                token: "secret-token".to_string(),
+                workspace_id: "workspace-a".to_string(),
+                runtime_id: "runtime-1".to_string(),
+                worker_id: None,
+                created_at: "2026-07-31T00:00:01Z".to_string(),
+            })
+            .unwrap();
+        let bound = store
+            .authenticate_worker_workspace_credential("secret-token", "workspace-a", "worker-1")
+            .unwrap()
+            .unwrap();
+        assert_eq!(bound.worker_id.as_deref(), Some("worker-1"));
+        assert!(
+            store
+                .authenticate_worker_workspace_credential(
+                    "secret-token",
+                    "workspace-a",
+                    "worker-2",
+                )
+                .unwrap()
+                .is_none()
+        );
+
+        store
+            .enqueue_ticket_notification(
+                "notification-1",
+                "workspace-a",
+                "ticket-1",
+                4,
+                "runtime-1",
+                "worker-1",
+                "queued",
+                "inprogress",
+                "2026-07-31T00:00:02Z",
+                &[TicketNotificationRecipient {
+                    runtime_id: "runtime-1".to_string(),
+                    worker_id: "worker-2".to_string(),
+                    recipient_kind: "assigned".to_string(),
+                }],
+            )
+            .unwrap();
+        let pending = store
+            .list_pending_ticket_notification_deliveries("workspace-a", 10)
+            .unwrap();
+        assert_eq!(pending.len(), 1);
+        assert_eq!(pending[0].event_sequence, 4);
+        store
+            .mark_ticket_notification_delivered(
+                "notification-1",
+                "runtime-1",
+                "worker-2",
+                "2026-07-31T00:00:03Z",
+            )
+            .unwrap();
+        assert!(
+            store
+                .list_pending_ticket_notification_deliveries("workspace-a", 10)
+                .unwrap()
+                .is_empty()
+        );
+    }
+
     #[test]
     fn fresh_schema_matches_workspace_db_v0_boundaries() {
         let conn = Connection::open_in_memory().unwrap();
@@ -2896,6 +3749,9 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
             "artifacts",
             "audit_events",
             "worker_registry",
+            "ticket_worker_assignments",
+            "ticket_current_worker_assignments",
+            "ticket_worker_assignment_events",
             "workdir_registry",
             "worker_workdir_links",
             "accounts",
@@ -3058,7 +3914,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
         .unwrap();
 
         let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 14);
+        assert_eq!(store.schema_version().await.unwrap(), 16);
 
         store
             .with_conn(|conn| {
@@ -3161,7 +4017,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
     #[tokio::test]
     async fn repository_records_round_trip() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 14);
+        assert_eq!(store.schema_version().await.unwrap(), 16);
         let workspace = WorkspaceRecord {
             workspace_id: "local-dev".to_string(),
             owner_account_id: None,
@@ -3199,7 +4055,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
     #[tokio::test]
     async fn memory_authority_records_round_trip_and_close_staging() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 14);
+        assert_eq!(store.schema_version().await.unwrap(), 16);
         let workspace = WorkspaceRecord {
             workspace_id: "local-dev".to_string(),
             owner_account_id: None,
@@ -3373,7 +4229,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
     #[tokio::test]
     async fn account_and_login_records_round_trip() {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        assert_eq!(store.schema_version().await.unwrap(), 14);
+        assert_eq!(store.schema_version().await.unwrap(), 16);
         let now = "2026-07-22T00:00:00Z".to_string();
         let account = AccountRecord {
             account_id: "acct-user-alice".to_string(),