Merge remote-tracking branch 'origin/develop' into work/T-689-web-subject-create

crates/client/src/backend_runtime.rs

@@ -819,7 +819,8 @@ mod tests {
             "profiles": [{
                 "id": "builtin:default",
                 "label": "Default",
-                "description": ""
+                "description": "",
+                "feature_connections": { "subjektiv": false }
             }],
             "repositories": [],
             "working_directories": [],
@@ -883,6 +884,7 @@ mod tests {
             ticket_assignment: None,
             initial_submit: Vec::new(),
             workdir_attachments: Vec::new(),
+            feature_connections: Default::default(),
             control_operation_id: None,
         };
 

crates/client/src/workspace_product.rs

@@ -532,6 +532,7 @@ impl BackendWorkspaceProductClient {
                 content: format!("Please handle intake for Ticket {ticket_id}."),
             }],
             workdir_attachments: Vec::new(),
+            feature_connections: Default::default(),
             control_operation_id: None,
         };
         let workspace_id = self.workspace_id.clone();

crates/protocol/src/lib.rs

@@ -175,6 +175,11 @@ pub struct WorkerStateSnapshot {
         schemars(range(min = 0, max = 9_007_199_254_740_991_u64))
     )]
     pub last_command_id: u64,
+    /// Most recent Submit request whose controller Run has returned and whose
+    /// post-Run state transition has been published. This is a teardown fence,
+    /// not a successful-result claim; cancelled and failed Runs may also finish.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub last_finished_submission_request_id: Option<String>,
     pub state: WorkerState,
 }
 
@@ -182,6 +187,7 @@ impl WorkerStateSnapshot {
     pub fn initial() -> Self {
         Self {
             last_command_id: 0,
+            last_finished_submission_request_id: None,
             state: WorkerState::Idle,
         }
     }
@@ -209,6 +215,7 @@ impl From<WorkerStatus> for WorkerStateSnapshot {
         };
         Self {
             last_command_id: 0,
+            last_finished_submission_request_id: None,
             state,
         }
     }
@@ -1037,10 +1044,11 @@ pub enum SessionContentPart {
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
 #[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SessionToolAttachment {
+    /// Stable, bounded identity derived from the owning Session entry, attachment
+    /// position, and immutable image bytes. It is a locator, not a bearer secret.
+    pub attachment_id: String,
     pub media_type: String,
-    /// Base64-encoded durable attachment body. Public snapshots preserve the
-    /// committed multimodal value instead of replacing it with placeholder text.
-    pub data_base64: String,
+    pub byte_len: u64,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -1925,6 +1933,7 @@ mod tests {
     fn worker_state_snapshot_wire_shape_has_one_authoritative_state() {
         let snapshot = WorkerStateSnapshot {
             last_command_id: 7,
+            last_finished_submission_request_id: Some("request-7".to_string()),
             state: WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running)),
         };
         let value = serde_json::to_value(&snapshot).unwrap();
@@ -1932,6 +1941,7 @@ mod tests {
             value,
             serde_json::json!({
                 "last_command_id": 7,
+                "last_finished_submission_request_id": "request-7",
                 "state": {
                     "kind": "busy",
                     "state": { "kind": "run", "state": "running" }
@@ -2648,6 +2658,7 @@ mod tests {
         let event = Event::WorkerState {
             snapshot: WorkerStateSnapshot {
                 last_command_id: 9,
+                last_finished_submission_request_id: None,
                 state: WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running)),
             },
         };
@@ -2669,6 +2680,7 @@ mod tests {
                 snapshot: WorkerStateSnapshot {
                     last_command_id: 9,
                     state: WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running)),
+                    ..
                 }
             }
         ));

crates/runtime-api/src/lib.rs

@@ -1085,6 +1085,8 @@ pub const RUNTIME_ROUTE_WORKER_PROTOCOL_WS: &str = "/v1/workers/{worker_id}/prot
 pub const RUNTIME_ROUTE_WORKER_ATTACHMENTS: &str = "/v1/workers/{worker_id}/attachments";
 pub const RUNTIME_ROUTE_WORKER_ATTACHMENT: &str =
     "/v1/workers/{worker_id}/attachments/{artifact_id}";
+pub const RUNTIME_ROUTE_WORKER_SESSION_ATTACHMENT: &str =
+    "/v1/workers/{worker_id}/sessions/{session_id}/attachments/{attachment_id}";
 
 /// Intentionally out-of-contract Runtime routes. This inventory keeps manual paths visible rather
 /// than letting them be mistaken for management-contract omissions.
@@ -1189,6 +1191,11 @@ pub const REMAINING_RUNTIME_ROUTES: &[RemainingRuntimeRoute] = &[
         path: RUNTIME_ROUTE_WORKER_ATTACHMENT,
         reason: "Binary attachment transport",
     },
+    RemainingRuntimeRoute {
+        method: "GET",
+        path: RUNTIME_ROUTE_WORKER_SESSION_ATTACHMENT,
+        reason: "Session image attachment transport",
+    },
 ];
 
 #[cfg(test)]
@@ -1478,7 +1485,7 @@ mod tests {
 
     #[test]
     fn remaining_route_inventory_does_not_overlap_contract() {
-        assert_eq!(REMAINING_RUNTIME_ROUTES.len(), 20);
+        assert_eq!(REMAINING_RUNTIME_ROUTES.len(), 21);
         for remaining in REMAINING_RUNTIME_ROUTES {
             assert!(!RuntimeApiMetadata::OPERATIONS.iter().any(|operation| {
                 format!("{:?}", operation.method).eq_ignore_ascii_case(remaining.method)

crates/server-api/src/lib.rs

@@ -249,7 +249,6 @@ impl_openapi_schema!(
     SubjektivMemoryQueryResponse,
     SubjektivMemoryReadResponse,
     SubjektivMemoryListRevisionsResponse,
-    SubjektivSubjectWorkerStartRequest,
     SubjektivStageCandidateRequest,
     SubjektivStageCandidateResponse,
     SubjektivRecordSessionRequest,
@@ -1234,21 +1233,6 @@ pub trait ServerApi {
         #[path] workspace_id: String,
         #[path] subject_id: String,
     ) -> Result<SubjektivSubjectResponse, RepositoryApiError>;
-    #[post(
-        "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/worker",
-        status = 200,
-        error_status = 400,
-        additional_error_statuses = [401, 403, 404, 409, 500, 502, 503],
-        bearer_auth = true,
-        browser_auth = true,
-    )]
-    async fn subjektiv_subject_worker_start(
-        &self,
-        #[extension] context: ServerRequestContext,
-        #[path] workspace_id: String,
-        #[path] subject_id: String,
-        #[body] request: SubjektivSubjectWorkerStartRequest,
-    ) -> Result<BrowserCreateWorkerResponse, RepositoryApiError>;
     #[post(
         "/api/w/{workspace_id}/subjektiv/staging",
         status = 200,
@@ -4780,12 +4764,6 @@ pub struct SubjektivMemoryRevisionsQuery {
     pub cursor: Option<String>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
-#[serde(deny_unknown_fields)]
-pub struct SubjektivSubjectWorkerStartRequest {
-    pub worker: CreateWorkspaceWorkerRequest,
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
 #[serde(deny_unknown_fields)]
 pub struct SubjektivStageCandidateRequest {
@@ -9489,6 +9467,14 @@ pub struct WorkerLaunchProfileCandidate {
     pub id: String,
     pub label: String,
     pub description: String,
+    pub feature_connections: WorkerLaunchFeatureConnectionOptions,
+}
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(deny_unknown_fields)]
+pub struct WorkerLaunchFeatureConnectionOptions {
+    pub subjektiv: bool,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
@@ -9520,6 +9506,22 @@ pub struct CreateWorkspaceWorkerTicketAssignmentRequest {
     pub operation_id: String,
 }
 
+#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(deny_unknown_fields)]
+pub struct WorkspaceWorkerFeatureConnectionsRequest {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    #[cfg_attr(feature = "typescript", ts(optional = nullable))]
+    pub subjektiv: Option<SubjektivWorkerConnectionRequest>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(deny_unknown_fields)]
+pub struct SubjektivWorkerConnectionRequest {
+    pub subject_id: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
 #[serde(deny_unknown_fields)]
@@ -9537,6 +9539,9 @@ pub struct CreateWorkspaceWorkerRequest {
     pub initial_submit: Vec<protocol::Segment>,
     #[serde(default)]
     pub workdir_attachments: Vec<BrowserWorkerWorkingDirectorySelection>,
+    /// Explicit product Feature connections. Backend validation turns these into
+    /// trusted host attachments; Runtime and Profile inputs cannot self-attest them.
+    pub feature_connections: WorkspaceWorkerFeatureConnectionsRequest,
     /// Backend idempotency key used only for authenticated Worker-owned spawn/control.
     #[serde(default)]
     pub control_operation_id: Option<String>,
@@ -10705,11 +10710,14 @@ pub fn worker_launch_api_typescript() -> String {
         RuntimeWorkerWorkdirAttachmentSummary::decl(&config),
         WorkerLaunchWorkerSummary::decl(&config),
         WorkerLaunchRuntimeOption::decl(&config),
+        WorkerLaunchFeatureConnectionOptions::decl(&config),
         WorkerLaunchProfileCandidate::decl(&config),
         WorkingDirectoryRepositoryOption::decl(&config),
         WorkerLaunchOptionsResponse::decl(&config),
         BrowserWorkerWorkingDirectorySelection::decl(&config),
         CreateWorkspaceWorkerTicketAssignmentRequest::decl(&config),
+        SubjektivWorkerConnectionRequest::decl(&config),
+        WorkspaceWorkerFeatureConnectionsRequest::decl(&config),
         CreateWorkspaceWorkerRequest::decl(&config),
         BrowserCreateWorkerResponse::decl(&config),
         BrowserWorkspaceOrchestratorResponse::decl(&config),
@@ -11750,6 +11758,7 @@ mod tests {
             ticket_assignment: None,
             initial_submit: Vec::new(),
             workdir_attachments: Vec::new(),
+            feature_connections: Default::default(),
             control_operation_id: None,
         })
         .unwrap();
@@ -11763,6 +11772,7 @@ mod tests {
                 "ticket_assignment": null,
                 "initial_submit": [],
                 "workdir_attachments": [],
+                "feature_connections": {},
                 "control_operation_id": null,
             })
         );

crates/session-store/src/lib.rs

@@ -110,9 +110,12 @@ pub use public_snapshot::{
     DEFAULT_RETAINED_HISTORY_MAX_ENTRIES, DEFAULT_RETAINED_HISTORY_MAX_RESPONSE_BYTES,
     DEFAULT_RETAINED_HISTORY_MAX_SCAN_BYTES, DEFAULT_RETAINED_HISTORY_MAX_SEGMENTS,
     DEFAULT_RETAINED_HISTORY_PAGE_TURNS, DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
-    MAX_RETAINED_HISTORY_PAGE_TURNS, RetainedHistoryReadError, RetainedHistoryReadLimits,
-    RetainedSessionIdentity, RetainedSessionSnapshot, RetainedSnapshotReadError,
-    project_session_snapshot, read_retained_session_history_page, read_retained_session_snapshot,
+    MAX_RETAINED_HISTORY_PAGE_TURNS, RetainedAttachmentReadError, RetainedHistoryReadError,
+    RetainedHistoryReadLimits, RetainedSessionAttachment, RetainedSessionIdentity,
+    RetainedSessionSnapshot, RetainedSnapshotReadError, project_session_snapshot,
+    read_retained_session_attachment, read_retained_session_history_page,
+    read_retained_session_snapshot, read_session_attachment_from_entries,
+    session_tool_attachment_id,
 };
 pub use segment::{
     SegmentStartState, append_entry, append_system_item, classify_logged_history_entry,

crates/session-store/src/public_snapshot.rs

@@ -1,7 +1,4 @@
-use base64::{
-    Engine as _,
-    engine::general_purpose::{STANDARD as BASE64, URL_SAFE_NO_PAD},
-};
+use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use protocol::{
     Segment, SessionContentPart, SessionConversationTurn, SessionEntryProvenance,
     SessionHistoryLineageBoundary, SessionHistoryPage, SessionMessageRole, SessionSnapshot,
@@ -54,6 +51,32 @@ pub enum RetainedSnapshotReadError {
     SnapshotTooLarge,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RetainedSessionAttachment {
+    pub media_type: String,
+    pub data: Vec<u8>,
+}
+
+#[derive(Debug, Error, PartialEq, Eq)]
+pub enum RetainedAttachmentReadError {
+    #[error("retained worker state is unavailable")]
+    RetentionMissing,
+    #[error("retained worker state has no active session pointer")]
+    ActivePointerMissing,
+    #[error("retained session identity does not match")]
+    SessionMismatch,
+    #[error("session attachment was not found")]
+    NotFound,
+    #[error("retained session requires migration")]
+    MigrationRequired,
+    #[error("retained session log is corrupt")]
+    CorruptLog,
+    #[error("retained session storage is unavailable")]
+    StorageUnavailable,
+    #[error("retained session attachment exceeds a resource limit")]
+    ResourceLimit,
+}
+
 pub const DEFAULT_RETAINED_HISTORY_PAGE_TURNS: usize = 5;
 pub const MAX_RETAINED_HISTORY_PAGE_TURNS: usize = 5;
 pub const DEFAULT_RETAINED_HISTORY_MAX_SCAN_BYTES: u64 = 64 * 1024 * 1024;
@@ -175,7 +198,9 @@ pub fn read_retained_session_snapshot(
         .ok_or(RetainedSnapshotReadError::ActivePointerMissing)?;
     let store = WorkerSessionStore::open_read_only(aggregate_root.join("session"))
         .map_err(map_store_error)?;
-    if store.segment_log_len(segment_id).map_err(map_store_error)? > max_bytes {
+    if store.segment_log_len(segment_id).map_err(map_store_error)?
+        > DEFAULT_RETAINED_HISTORY_MAX_SCAN_BYTES
+    {
         return Err(RetainedSnapshotReadError::SnapshotTooLarge);
     }
     let entries = store
@@ -199,6 +224,153 @@ pub fn read_retained_session_snapshot(
     })
 }
 
+/// Resolve one immutable image body from the active retained Session. The
+/// caller supplies only public identities; storage paths never cross this API.
+pub fn read_retained_session_attachment(
+    aggregate_root: &Path,
+    worker_name: &str,
+    expected_session_id: &str,
+    attachment_id: &str,
+    max_scan_bytes: u64,
+    max_attachment_bytes: u64,
+) -> Result<RetainedSessionAttachment, RetainedAttachmentReadError> {
+    let aggregate = WorkerAggregateStore::open_read_only(aggregate_root, worker_name)
+        .map_err(map_attachment_worker_store_error)?;
+    let metadata = aggregate
+        .read_read_only()
+        .map_err(map_attachment_worker_store_error)?
+        .ok_or(RetainedAttachmentReadError::RetentionMissing)?;
+    let active = metadata
+        .active
+        .ok_or(RetainedAttachmentReadError::ActivePointerMissing)?;
+    if active.session_id.to_string() != expected_session_id {
+        return Err(RetainedAttachmentReadError::SessionMismatch);
+    }
+    let segment_id = active
+        .segment_id
+        .ok_or(RetainedAttachmentReadError::ActivePointerMissing)?;
+    let store = WorkerSessionStore::open_read_only(aggregate_root.join("session"))
+        .map_err(map_attachment_store_error)?;
+    let limits = RetainedHistoryReadLimits {
+        max_scan_bytes,
+        ..RetainedHistoryReadLimits::default()
+    };
+    let (lineage, _, mut scanned_bytes, mut scanned_entries) =
+        load_adopted_lineage(&store, active.session_id, segment_id, limits)
+            .map_err(map_history_attachment_error)?;
+    for segment in &lineage {
+        let adopted_end = locate_adopted_segment_end(
+            &store,
+            active.session_id,
+            segment,
+            limits,
+            &mut scanned_bytes,
+            &mut scanned_entries,
+        )
+        .map_err(map_history_attachment_error)?;
+        let mut reader = store
+            .open_retained_segment_reader(active.session_id, segment.segment_id, adopted_end)
+            .map_err(map_attachment_store_error)?;
+        loop {
+            let remaining = limits.max_scan_bytes.saturating_sub(scanned_bytes);
+            let (record, bytes) = reader
+                .previous_record(remaining)
+                .map_err(map_attachment_store_error)?;
+            scanned_bytes = scanned_bytes
+                .checked_add(bytes)
+                .ok_or(RetainedAttachmentReadError::ResourceLimit)?;
+            let Some(record) = record else {
+                break;
+            };
+            scanned_entries = scanned_entries
+                .checked_add(persisted_entry_units(std::slice::from_ref(&record.entry)))
+                .ok_or(RetainedAttachmentReadError::ResourceLimit)?;
+            if scanned_entries > limits.max_entries {
+                return Err(RetainedAttachmentReadError::ResourceLimit);
+            }
+            if let Some(result) =
+                session_attachment_from_record(&record.entry, attachment_id, max_attachment_bytes)
+            {
+                return result;
+            }
+        }
+    }
+    Err(RetainedAttachmentReadError::NotFound)
+}
+
+pub fn read_session_attachment_from_entries(
+    entries: &[LogEntry],
+    attachment_id: &str,
+    max_attachment_bytes: u64,
+) -> Result<RetainedSessionAttachment, RetainedAttachmentReadError> {
+    for record in entries {
+        if let Some(result) =
+            session_attachment_from_record(record, attachment_id, max_attachment_bytes)
+        {
+            return result;
+        }
+    }
+    Err(RetainedAttachmentReadError::NotFound)
+}
+
+fn session_attachment_from_record(
+    record: &LogEntry,
+    attachment_id: &str,
+    max_attachment_bytes: u64,
+) -> Option<Result<RetainedSessionAttachment, RetainedAttachmentReadError>> {
+    let mut found = None;
+    visit_record_history(record, &mut |entry| {
+        if found.is_none() {
+            found = attachment_from_entry(entry, attachment_id, max_attachment_bytes);
+        }
+    });
+    found
+}
+
+fn visit_record_history(record: &LogEntry, visit: &mut impl FnMut(&LoggedHistoryEntry)) {
+    match record {
+        LogEntry::AnnotatedSegmentStart { history, .. }
+        | LogEntry::AnnotatedUserInput { history, .. } => {
+            for entry in history {
+                visit(entry);
+            }
+        }
+        LogEntry::AnnotatedAssistantItem { entry, .. }
+        | LogEntry::AnnotatedToolResult { entry, .. } => visit(entry),
+        _ => {}
+    }
+}
+
+fn attachment_from_entry(
+    entry: &LoggedHistoryEntry,
+    requested_id: &str,
+    max_attachment_bytes: u64,
+) -> Option<Result<RetainedSessionAttachment, RetainedAttachmentReadError>> {
+    let LoggedItem::ToolResult { attachments, .. } = &entry.item else {
+        return None;
+    };
+    for (attachment_index, attachment) in attachments.iter().enumerate() {
+        let crate::logged_item::LoggedAttachment::Image { mime_type, data } = attachment;
+        if session_tool_attachment_id(
+            &entry.metadata.entry_id.0,
+            attachment_index,
+            mime_type,
+            data,
+        ) != requested_id
+        {
+            continue;
+        }
+        if data.len() as u64 > max_attachment_bytes {
+            return Some(Err(RetainedAttachmentReadError::ResourceLimit));
+        }
+        return Some(Ok(RetainedSessionAttachment {
+            media_type: mime_type.clone(),
+            data: data.clone(),
+        }));
+    }
+    None
+}
+
 /// Read one bounded backward page from the retained active Segment's adopted
 /// lineage without creating, migrating, restoring, or rewriting persisted state.
 ///
@@ -935,6 +1107,54 @@ fn map_history_store_error(error: StoreError) -> RetainedHistoryReadError {
     }
 }
 
+fn map_history_attachment_error(error: RetainedHistoryReadError) -> RetainedAttachmentReadError {
+    match error {
+        RetainedHistoryReadError::RetentionMissing => RetainedAttachmentReadError::RetentionMissing,
+        RetainedHistoryReadError::ActivePointerMissing => {
+            RetainedAttachmentReadError::ActivePointerMissing
+        }
+        RetainedHistoryReadError::MigrationRequired => {
+            RetainedAttachmentReadError::MigrationRequired
+        }
+        RetainedHistoryReadError::CorruptLog | RetainedHistoryReadError::InvalidCursor => {
+            RetainedAttachmentReadError::CorruptLog
+        }
+        RetainedHistoryReadError::StorageUnavailable => {
+            RetainedAttachmentReadError::StorageUnavailable
+        }
+        RetainedHistoryReadError::ResourceLimit => RetainedAttachmentReadError::ResourceLimit,
+    }
+}
+
+fn map_attachment_worker_store_error(error: WorkerStoreError) -> RetainedAttachmentReadError {
+    match error {
+        WorkerStoreError::Io(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            RetainedAttachmentReadError::RetentionMissing
+        }
+        WorkerStoreError::Serde(_) | WorkerStoreError::InvalidWorkerName(_) => {
+            RetainedAttachmentReadError::CorruptLog
+        }
+        WorkerStoreError::Io(_) => RetainedAttachmentReadError::StorageUnavailable,
+    }
+}
+
+fn map_attachment_store_error(error: StoreError) -> RetainedAttachmentReadError {
+    match error {
+        StoreError::Io(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            RetainedAttachmentReadError::RetentionMissing
+        }
+        StoreError::Corrupt { message, .. } if message.contains("requires migration") => {
+            RetainedAttachmentReadError::MigrationRequired
+        }
+        StoreError::ReadLimitExceeded => RetainedAttachmentReadError::ResourceLimit,
+        StoreError::Io(_) => RetainedAttachmentReadError::StorageUnavailable,
+        StoreError::Serde(_) | StoreError::Corrupt { .. } | StoreError::NotFound(_) => {
+            RetainedAttachmentReadError::CorruptLog
+        }
+        _ => RetainedAttachmentReadError::CorruptLog,
+    }
+}
+
 fn map_worker_store_error(error: WorkerStoreError) -> RetainedSnapshotReadError {
     match error {
         WorkerStoreError::Io(error) if error.kind() == std::io::ErrorKind::NotFound => {
@@ -1028,7 +1248,7 @@ pub(crate) fn project_session_snapshot_for_segment(
             } => extend_history(&mut entries, history, Some(segments), *ts),
             LogEntry::AnnotatedAssistantItem { ts, entry }
             | LogEntry::AnnotatedToolResult { ts, entry } => {
-                if let Some(data) = project_item(&entry.item) {
+                if let Some(data) = project_item(&entry.metadata.entry_id.0, &entry.item) {
                     entries.push(history_entry(entry, *ts, data));
                 }
             }
@@ -1188,7 +1408,7 @@ fn extend_history(
                 segments: input_segments.cloned().unwrap_or_default(),
             }
         } else {
-            let Some(data) = project_item(&entry.item) else {
+            let Some(data) = project_item(&entry.metadata.entry_id.0, &entry.item) else {
                 continue;
             };
             data
@@ -1306,7 +1526,25 @@ fn provenance(origin: &LoggedSessionHistoryOrigin) -> SessionEntryProvenance {
     }
 }
 
-fn project_item(item: &LoggedItem) -> Option<SessionSnapshotEntryData> {
+pub fn session_tool_attachment_id(
+    entry_id: &str,
+    attachment_index: usize,
+    media_type: &str,
+    data: &[u8],
+) -> String {
+    let mut hash = Sha256::new();
+    hash.update(b"yoi-session-tool-attachment-v1\0");
+    hash.update((entry_id.len() as u64).to_be_bytes());
+    hash.update(entry_id.as_bytes());
+    hash.update((attachment_index as u64).to_be_bytes());
+    hash.update((media_type.len() as u64).to_be_bytes());
+    hash.update(media_type.as_bytes());
+    hash.update((data.len() as u64).to_be_bytes());
+    hash.update(data);
+    URL_SAFE_NO_PAD.encode(hash.finalize())
+}
+
+fn project_item(entry_id: &str, item: &LoggedItem) -> Option<SessionSnapshotEntryData> {
     match item {
         LoggedItem::Message { role, content } => {
             let role = match role {
@@ -1355,11 +1593,18 @@ fn project_item(item: &LoggedItem) -> Option<SessionSnapshotEntryData> {
             is_error: *is_error,
             attachments: attachments
                 .iter()
-                .map(|attachment| match attachment {
+                .enumerate()
+                .map(|(attachment_index, attachment)| match attachment {
                     crate::logged_item::LoggedAttachment::Image { mime_type, data } => {
                         SessionToolAttachment {
+                            attachment_id: session_tool_attachment_id(
+                                entry_id,
+                                attachment_index,
+                                mime_type,
+                                data,
+                            ),
                             media_type: mime_type.clone(),
-                            data_base64: BASE64.encode(data),
+                            byte_len: data.len() as u64,
                         }
                     }
                 })
@@ -3307,6 +3552,206 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn retained_attachment_resolves_from_an_adopted_compaction_ancestor() {
+        let worker_name = "worker-ancestor-image";
+        let session_id = crate::new_session_id();
+        let source_segment = crate::new_segment_id();
+        let active_segment = crate::new_segment_id();
+        let image_entry = LoggedHistoryEntry {
+            item: LoggedItem::ToolResult {
+                call_id: "call-ancestor-image".into(),
+                summary: "ancestor screenshot".into(),
+                content: None,
+                attachments: vec![crate::logged_item::LoggedAttachment::Image {
+                    mime_type: "image/png".into(),
+                    data: vec![7, 8, 9],
+                }],
+                disposition: agen::tool::ToolResultDisposition::Success,
+                is_error: false,
+            },
+            metadata: LoggedSessionHistoryMetadata {
+                entry_id: LoggedSessionHistoryEntryId("ancestor-image".into()),
+                origin: LoggedSessionHistoryOrigin::LegacyUnknown,
+                derivation: None,
+            },
+        };
+        let mut source_log = vec![segment_start(session_id, Vec::new(), None, None)];
+        source_log.push(LogEntry::AnnotatedUserInput {
+            ts: 2,
+            segments: vec![Segment::Text {
+                content: "show image".into(),
+            }],
+            history: vec![user_message(1)],
+            extensions: Vec::new(),
+        });
+        source_log.push(LogEntry::AnnotatedToolResult {
+            ts: 3,
+            entry: image_entry,
+        });
+        source_log.push(LogEntry::TurnEnd {
+            ts: 4,
+            turn_count: 1,
+        });
+        let active_log = vec![segment_start(
+            session_id,
+            Vec::new(),
+            None,
+            Some(SegmentOrigin {
+                segment_id: source_segment,
+                at_turn_index: 1,
+            }),
+        )];
+        let (_root, aggregate_root) = persist_history_fixture(
+            worker_name,
+            session_id,
+            active_segment,
+            vec![(source_segment, source_log), (active_segment, active_log)],
+        );
+
+        let page = read_retained_session_history_page(
+            &aggregate_root,
+            worker_name,
+            None,
+            None,
+            RetainedHistoryReadLimits::default(),
+        )
+        .unwrap();
+        let attachment = page.turns[0]
+            .entries
+            .iter()
+            .find_map(|entry| match &entry.data {
+                SessionSnapshotEntryData::ToolResult { attachments, .. } => attachments.first(),
+                _ => None,
+            })
+            .expect("ancestor attachment reference");
+        let body = read_retained_session_attachment(
+            &aggregate_root,
+            worker_name,
+            &session_id.to_string(),
+            &attachment.attachment_id,
+            DEFAULT_RETAINED_HISTORY_MAX_SCAN_BYTES,
+            10 * 1024 * 1024,
+        )
+        .unwrap();
+        assert_eq!(body.media_type, "image/png");
+        assert_eq!(body.data, vec![7, 8, 9]);
+    }
+
+    #[test]
+    fn large_legacy_inline_images_project_as_bounded_refs_and_fetch_from_retained_storage() {
+        let worker_name = "worker-images";
+        let session_id = crate::new_session_id();
+        let segment_id = crate::new_segment_id();
+        // Approximate the decoded byte sizes of the four W-296 screenshots.
+        let sizes = [3_475_194_usize, 3_459_051, 5_735_340, 5_703_924];
+        let attachments = sizes
+            .iter()
+            .enumerate()
+            .map(
+                |(index, size)| crate::logged_item::LoggedAttachment::Image {
+                    mime_type: "image/png".into(),
+                    data: vec![index as u8 + 1; *size],
+                },
+            )
+            .collect();
+        let tool_result = LoggedHistoryEntry {
+            item: LoggedItem::ToolResult {
+                call_id: "call-images".into(),
+                summary: "four screenshots".into(),
+                content: None,
+                attachments,
+                disposition: agen::tool::ToolResultDisposition::Success,
+                is_error: false,
+            },
+            metadata: LoggedSessionHistoryMetadata {
+                entry_id: LoggedSessionHistoryEntryId("tool-images".into()),
+                origin: LoggedSessionHistoryOrigin::LegacyUnknown,
+                derivation: None,
+            },
+        };
+        let live = project_current_session_snapshot(&[LogEntry::AnnotatedToolResult {
+            ts: 2,
+            entry: tool_result.clone(),
+        }]);
+        let log = vec![LogEntry::AnnotatedSegmentStart {
+            ts: 1,
+            session_id,
+            system_prompt: None,
+            config: RequestConfig::default(),
+            history: vec![tool_result],
+            forked_from: None,
+            compacted_from: None,
+        }];
+        let (_root, aggregate_root) =
+            persist_history_fixture(worker_name, session_id, segment_id, vec![(segment_id, log)]);
+
+        let retained = read_retained_session_snapshot(
+            &aggregate_root,
+            worker_name,
+            DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
+        )
+        .unwrap();
+        let SessionSnapshotEntryData::ToolResult { attachments, .. } =
+            &retained.snapshot.entries[0].data
+        else {
+            panic!("expected tool result");
+        };
+        assert_eq!(attachments.len(), 4);
+        let SessionSnapshotEntryData::ToolResult {
+            attachments: live_attachments,
+            ..
+        } = &live.entries[0].data
+        else {
+            panic!("expected live tool result");
+        };
+        assert_eq!(live_attachments, attachments);
+        let snapshot_json = serde_json::to_vec(&retained.snapshot).unwrap();
+        assert!(snapshot_json.len() < 16 * 1024);
+        assert!(!String::from_utf8_lossy(&snapshot_json).contains("data_base64"));
+
+        for (index, attachment) in attachments.iter().enumerate() {
+            assert_eq!(attachment.byte_len, sizes[index] as u64);
+            assert_eq!(attachment.attachment_id.len(), 43);
+            let body = read_retained_session_attachment(
+                &aggregate_root,
+                worker_name,
+                &session_id.to_string(),
+                &attachment.attachment_id,
+                DEFAULT_RETAINED_HISTORY_MAX_SCAN_BYTES,
+                10 * 1024 * 1024,
+            )
+            .unwrap();
+            assert_eq!(body.media_type, "image/png");
+            assert_eq!(body.data.len(), sizes[index]);
+            assert!(body.data.iter().all(|byte| *byte == index as u8 + 1));
+        }
+        assert_eq!(
+            read_retained_session_attachment(
+                &aggregate_root,
+                worker_name,
+                &crate::new_session_id().to_string(),
+                &attachments[0].attachment_id,
+                DEFAULT_RETAINED_HISTORY_MAX_SCAN_BYTES,
+                10 * 1024 * 1024,
+            )
+            .unwrap_err(),
+            RetainedAttachmentReadError::SessionMismatch
+        );
+        assert_eq!(
+            read_retained_session_attachment(
+                &aggregate_root,
+                worker_name,
+                &session_id.to_string(),
+                "../../metadata.json",
+                DEFAULT_RETAINED_HISTORY_MAX_SCAN_BYTES,
+                10 * 1024 * 1024,
+            )
+            .unwrap_err(),
+            RetainedAttachmentReadError::NotFound
+        );
+    }
+
     #[test]
     fn retained_reads_do_not_create_missing_aggregate() {
         let root = tempfile::tempdir().unwrap();

crates/tui/src/app.rs

@@ -3761,6 +3761,7 @@ mod completion_flow_tests {
                 protocol::WorkerRunState::Running,
             )),
             last_command_id: 0,
+            last_finished_submission_request_id: None,
         };
         app.handle_worker_event(Event::WorkerState {
             snapshot: running.clone(),
@@ -3780,6 +3781,7 @@ mod completion_flow_tests {
                 protocol::WorkerRunState::Running,
             )),
             last_command_id: 2,
+            last_finished_submission_request_id: Some("request-old".to_string()),
         };
         app.handle_worker_event(Event::WorkerState {
             snapshot: running.clone(),
@@ -3789,6 +3791,7 @@ mod completion_flow_tests {
         let fresh_idle = WorkerStateSnapshot {
             state: protocol::WorkerState::Idle,
             last_command_id: 0,
+            last_finished_submission_request_id: None,
         };
         app.handle_worker_event(Event::WorkerState {
             snapshot: fresh_idle.clone(),
@@ -3800,6 +3803,7 @@ mod completion_flow_tests {
                 protocol::WorkerRunState::Paused,
             )),
             last_command_id: 3,
+            last_finished_submission_request_id: None,
         };
         app.handle_worker_event(Event::CommandAcknowledged {
             acknowledgement: protocol::WorkerCommandAcknowledgement {

crates/tui/src/backend_spawn.rs

@@ -167,6 +167,7 @@ fn request_from_selection(selection: Selection) -> BackendCreateWorkerRequest {
         initial_submit: Vec::new(),
         workdir_attachments: Vec::new(),
         ticket_assignment: None,
+        feature_connections: Default::default(),
         control_operation_id: None,
     }
 }
@@ -431,11 +432,13 @@ mod tests {
                     id: "builtin:default".to_string(),
                     label: "Default".to_string(),
                     description: String::new(),
+                    feature_connections: Default::default(),
                 },
                 BackendWorkerLaunchProfileCandidate {
                     id: "builtin:coder".to_string(),
                     label: "Coder".to_string(),
                     description: "Ticket implementation".to_string(),
+                    feature_connections: Default::default(),
                 },
             ],
             default_profile: Some("builtin:coder".to_string()),

crates/tui/src/backend_worker_picker.rs

@@ -579,6 +579,7 @@ mod tests {
                     protocol::WorkerRunState::Running,
                 )),
                 last_command_id: 0,
+                last_finished_submission_request_id: None,
             }),
             last_seen_at: None,
             pinned: false,
@@ -869,6 +870,7 @@ mod tests {
         short.worker_state = Some(protocol::WorkerStateSnapshot {
             state: protocol::WorkerState::Idle,
             last_command_id: 0,
+            last_finished_submission_request_id: None,
         });
 
         let mut long = worker("runtime-a", "worker-b", None);

crates/tui/src/console/mod.rs

@@ -508,6 +508,7 @@ impl RetainedSnapshotSocket {
                 },
                 state: protocol::WorkerStateSnapshot {
                     last_command_id: 0,
+                    last_finished_submission_request_id: None,
                     state: protocol::WorkerState::Idle,
                 },
                 in_flight: protocol::InFlightSnapshot::default(),

crates/worker-runtime/src/catalog.rs

@@ -367,6 +367,10 @@ pub struct WorkerDetail {
     pub created_at_ms: Option<u64>,
     /// Whether the persisted execution metadata was valid when this identity was loaded.
     pub execution_metadata_available: bool,
+    /// Whether this Runtime process reconstructed the Worker from durable execution state.
+    /// This proves that work from the prior process incarnation is no longer running.
+    #[serde(default)]
+    pub execution_reconstructed: bool,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub worker_state: Option<protocol::WorkerStateSnapshot>,
     #[serde(default, skip_serializing_if = "Option::is_none")]

crates/worker-runtime/src/execution.rs

@@ -514,6 +514,13 @@ pub struct WorkerSessionHistoryRequest {
     pub limit: Option<usize>,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct WorkerSessionAttachmentRequest {
+    pub worker_ref: WorkerRef,
+    pub session_id: String,
+    pub attachment_id: String,
+}
+
 pub trait WorkerExecutionBackend: Send + Sync + 'static {
     fn backend_id(&self) -> &str;
 
@@ -544,6 +551,14 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
         }
     }
 
+    fn worker_session_attachment(
+        &self,
+        _request: WorkerSessionAttachmentRequest,
+    ) -> Result<session_store::RetainedSessionAttachment, session_store::RetainedAttachmentReadError>
+    {
+        Err(session_store::RetainedAttachmentReadError::RetentionMissing)
+    }
+
     fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult;
 
     /// Read-only restore validation performed before any live controller,
@@ -780,6 +795,14 @@ impl WorkerExecutionBackendRef {
         self.backend.worker_session_history(request)
     }
 
+    pub(crate) fn worker_session_attachment(
+        &self,
+        request: WorkerSessionAttachmentRequest,
+    ) -> Result<session_store::RetainedSessionAttachment, session_store::RetainedAttachmentReadError>
+    {
+        self.backend.worker_session_attachment(request)
+    }
+
     pub(crate) fn spawn_worker(
         &self,
         request: WorkerExecutionSpawnRequest,

crates/worker-runtime/src/fs_store.rs

@@ -523,6 +523,7 @@ pub(crate) struct PersistedWorkerRecord {
     pub(crate) config_bundle: Option<ConfigBundleRef>,
     pub(crate) created_at_ms: Option<u64>,
     pub(crate) status: WorkerStatus,
+    pub(crate) last_finished_submission_request_id: Option<String>,
     pub(crate) execution_state: PersistedWorkerExecutionState,
     pub(crate) workspace_id: Option<String>,
     pub(crate) workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
@@ -1824,6 +1825,8 @@ struct WorkerAggregateRecord {
     created_at_ms: Option<u64>,
     status: WorkerStatus,
     #[serde(default, skip_serializing_if = "Option::is_none")]
+    last_finished_submission_request_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     workspace_id: Option<String>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
@@ -1874,6 +1877,7 @@ impl WorkerAggregateRecord {
             config_bundle: worker.config_bundle.clone(),
             created_at_ms: worker.created_at_ms,
             status: worker.status,
+            last_finished_submission_request_id: worker.last_finished_submission_request_id.clone(),
             workspace_id: worker.workspace_id.clone(),
             workdir_attachments: worker.workdir_attachments.clone(),
             logical_workdir_attachments: worker.logical_workdir_attachments.clone(),
@@ -1893,6 +1897,7 @@ impl WorkerAggregateRecord {
     }
 
     fn validate(self, path: &Path) -> Result<PersistedWorkerRecord, RuntimeError> {
+        let last_finished_submission_request_id = self.last_finished_submission_request_id;
         let identity = WorkerIdentityRecord {
             schema_version: self.schema_version,
             worker_ref: self.worker_ref,
@@ -1984,7 +1989,9 @@ impl WorkerAggregateRecord {
             }
             WorkerExecutionStateRecord::Unavailable => PersistedWorkerExecutionState::Unavailable,
         };
-        Ok(identity.into_persisted(execution_state))
+        let mut persisted = identity.into_persisted(execution_state);
+        persisted.last_finished_submission_request_id = last_finished_submission_request_id;
+        Ok(persisted)
     }
 }
 
@@ -2164,6 +2171,7 @@ impl WorkerIdentityRecord {
             config_bundle: self.config_bundle,
             created_at_ms: self.created_at_ms,
             status: self.status,
+            last_finished_submission_request_id: None,
             execution_state,
             workspace_id: self.workspace_id,
             workdir_attachments,

crates/worker-runtime/src/http_server.rs

@@ -336,6 +336,13 @@ fn runtime_http_router_with_auth_and_ssh_keyscan_program(
         .route(
             remaining_route(runtime_api::RUNTIME_ROUTE_WORKER_ATTACHMENT, &["DELETE"]),
             delete(delete_worker_uploaded_file),
+        )
+        .route(
+            remaining_route(
+                runtime_api::RUNTIME_ROUTE_WORKER_SESSION_ATTACHMENT,
+                &["GET"],
+            ),
+            get(get_worker_session_attachment),
         );
 
     #[cfg(feature = "ws-server")]
@@ -1980,6 +1987,60 @@ async fn delete_worker_uploaded_file(
     }))
 }
 
+async fn get_worker_session_attachment(
+    State(state): State<RuntimeHttpState>,
+    auth: Option<Extension<RuntimeAuthContext>>,
+    Path((worker_id, session_id, attachment_id)): Path<(String, String, String)>,
+    Query(request): Query<runtime_api::WorkerSessionRequest>,
+) -> Result<Response, RuntimeHttpRestError> {
+    let worker_ref = worker_ref_for(&state.runtime, worker_id)?;
+    let scope = auth_workspace_scope(&state, auth.as_ref())?.ok_or_else(|| {
+        RuntimeHttpRestError::new(
+            StatusCode::FORBIDDEN,
+            "runtime_worker_session_scope_required",
+            "Worker Session attachment reads require a Workspace-scoped capability",
+        )
+    })?;
+    if scope.workspace_id != request.workspace_id {
+        return Err(RuntimeHttpRestError::new(
+            StatusCode::FORBIDDEN,
+            "runtime_worker_session_workspace_scope_mismatch",
+            "Worker Session Workspace scope does not match the authenticated capability",
+        ));
+    }
+    let attachment = state
+        .runtime
+        .worker_session_attachment_scoped(&scope, &worker_ref, session_id, attachment_id)
+        .map_err(|error| {
+            use session_store::RetainedAttachmentReadError as Error;
+            let (status, code) = match error {
+                Error::RetentionMissing | Error::ActivePointerMissing => {
+                    (StatusCode::GONE, "session_attachment_expired")
+                }
+                Error::SessionMismatch | Error::NotFound => {
+                    (StatusCode::NOT_FOUND, "session_attachment_not_found")
+                }
+                Error::ResourceLimit => (
+                    StatusCode::PAYLOAD_TOO_LARGE,
+                    "session_attachment_too_large",
+                ),
+                Error::MigrationRequired | Error::CorruptLog | Error::StorageUnavailable => (
+                    StatusCode::SERVICE_UNAVAILABLE,
+                    "session_attachment_unavailable",
+                ),
+            };
+            RuntimeHttpRestError::new(status, code, error.to_string())
+        })?;
+    let content_type = axum::http::HeaderValue::from_str(&attachment.media_type).map_err(|_| {
+        RuntimeHttpRestError::new(
+            StatusCode::SERVICE_UNAVAILABLE,
+            "session_attachment_media_type_invalid",
+            "Session attachment media type is invalid",
+        )
+    })?;
+    Ok(([(header::CONTENT_TYPE, content_type)], attachment.data).into_response())
+}
+
 #[allow(dead_code)]
 async fn stop_worker(
     State(state): State<RuntimeHttpState>,
@@ -2289,7 +2350,7 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
     }
     if path.ends_with("/input")
         || path.ends_with("/restore")
-        || path.contains("/attachments")
+        || (path.contains("/attachments") && *method != Method::GET)
         || path.contains("/workdir-attachments")
     {
         return Some("workers:input");
@@ -3031,6 +3092,13 @@ mod tests {
             ),
             Some("workers:input")
         );
+        assert_eq!(
+            required_runtime_permission(
+                &Method::GET,
+                "/v1/workers/7/sessions/session-1/attachments/attachment-1"
+            ),
+            Some("workers:read")
+        );
         assert_eq!(
             required_runtime_permission(&Method::POST, "/v1/workers/7/workdir-attachments"),
             Some("workers:input")

crates/worker-runtime/src/runtime.rs

@@ -1095,6 +1095,8 @@ impl Runtime {
                 worker_id: worker_id.clone(),
                 status: WorkerStatus::Stopped,
                 worker_state: None,
+                last_finished_submission_request_id: None,
+                execution_reconstructed: false,
                 workspace_id: scope.map(|scope| scope.workspace_id.clone()),
                 profile: durable_request.profile.clone(),
                 display_name: durable_request.display_name.clone(),
@@ -1673,6 +1675,39 @@ impl Runtime {
         })
     }
 
+    pub fn worker_session_attachment_scoped(
+        &self,
+        scope: &RuntimeWorkspaceScope,
+        worker_ref: &WorkerRef,
+        session_id: String,
+        attachment_id: String,
+    ) -> Result<session_store::RetainedSessionAttachment, session_store::RetainedAttachmentReadError>
+    {
+        let operation_lock = self
+            .worker_operation_lock(worker_ref.worker_id)
+            .map_err(|_| session_store::RetainedAttachmentReadError::RetentionMissing)?;
+        let _operation_guard = operation_lock
+            .lock()
+            .map_err(|_| session_store::RetainedAttachmentReadError::StorageUnavailable)?;
+        self.ensure_worker_in_workspace(scope, worker_ref)
+            .map_err(|_| session_store::RetainedAttachmentReadError::NotFound)?;
+        let backend = {
+            let state = self
+                .lock()
+                .map_err(|_| session_store::RetainedAttachmentReadError::StorageUnavailable)?;
+            state
+                .worker(worker_ref)
+                .map_err(|_| session_store::RetainedAttachmentReadError::NotFound)?;
+            state.execution_backend.clone()
+        }
+        .ok_or(session_store::RetainedAttachmentReadError::StorageUnavailable)?;
+        backend.worker_session_attachment(crate::execution::WorkerSessionAttachmentRequest {
+            worker_ref: worker_ref.clone(),
+            session_id,
+            attachment_id,
+        })
+    }
+
     /// Attach a live execution through a workspace-scoped Runtime authorization context.
     pub fn restore_worker_scoped(
         &self,
@@ -3386,6 +3421,16 @@ impl Runtime {
         candidate.pending_stop = None;
         candidate.restore_candidate_context = None;
         candidate.status = status;
+        let mut worker_state = worker_state;
+        if worker_state.last_finished_submission_request_id.is_none() {
+            // The restored controller starts with fresh live state, but this
+            // historical fence proves that a previously accepted Submit Run
+            // already returned. Preserve it across process reconstruction so
+            // durable consumers can finish post-restart cleanup without
+            // waiting for an old request to run again.
+            worker_state.last_finished_submission_request_id =
+                candidate.last_finished_submission_request_id.clone();
+        }
         let _ = candidate.apply_worker_state(&worker_state);
         candidate.restore_intent = restore_intent_for_status(candidate.status);
         candidate.workdir_attachments = workdir_attachments;
@@ -3996,6 +4041,8 @@ impl RuntimeState {
                     worker_id: worker.worker_id,
                     status: worker.status,
                     worker_state: None,
+                    last_finished_submission_request_id: worker.last_finished_submission_request_id,
+                    execution_reconstructed: true,
                     workspace_id: worker.workspace_id,
                     profile: worker.profile,
                     display_name: worker.display_name,
@@ -4893,6 +4940,8 @@ struct WorkerRecord {
     worker_id: WorkerId,
     status: WorkerStatus,
     worker_state: Option<protocol::WorkerStateSnapshot>,
+    last_finished_submission_request_id: Option<String>,
+    execution_reconstructed: bool,
     workspace_id: Option<String>,
     profile: ProfileSelector,
     display_name: Option<String>,
@@ -4914,6 +4963,9 @@ struct WorkerRecord {
 
 impl WorkerRecord {
     fn apply_worker_state(&mut self, incoming: &protocol::WorkerStateSnapshot) {
+        if let Some(request_id) = &incoming.last_finished_submission_request_id {
+            self.last_finished_submission_request_id = Some(request_id.clone());
+        }
         self.worker_state = Some(incoming.clone());
     }
 
@@ -4959,6 +5011,7 @@ impl WorkerRecord {
             status: self.status,
             created_at_ms: self.created_at_ms,
             execution_metadata_available: self.execution_metadata_available,
+            execution_reconstructed: self.execution_reconstructed,
             worker_state: self.worker_state.clone(),
             workspace_id: self.workspace_id.clone(),
             workdir_attachments: self.workdir_attachments.clone(),
@@ -5022,6 +5075,7 @@ impl WorkerRecord {
             config_bundle: self.config_bundle.clone(),
             created_at_ms: self.created_at_ms,
             status: self.status,
+            last_finished_submission_request_id: self.last_finished_submission_request_id.clone(),
             execution_state,
             workspace_id: self.workspace_id.clone(),
             workdir_attachments: self.workdir_attachments.clone(),
@@ -8028,6 +8082,7 @@ mod tests {
                 protocol::WorkerRunState::Running,
             )),
             last_command_id: 2,
+            last_finished_submission_request_id: Some("request-old".to_string()),
         };
         {
             let mut state = runtime.lock().unwrap();
@@ -8049,6 +8104,7 @@ mod tests {
         let fresh = protocol::WorkerStateSnapshot {
             state: protocol::WorkerState::Idle,
             last_command_id: 0,
+            last_finished_submission_request_id: None,
         };
         {
             let mut state = runtime.lock().unwrap();
@@ -9196,6 +9252,81 @@ mod tests {
         let _ = std::fs::remove_dir_all(root);
     }
 
+    #[cfg(feature = "fs-store")]
+    #[test]
+    fn fs_store_preserves_finished_submission_fence_across_active_restore() {
+        let root = fs_store_root("finished-submission-restore");
+        let options = crate::fs_store::FsRuntimeStoreOptions {
+            root: root.clone(),
+            runtime_id: "test-runtime".to_string(),
+            display_name: None,
+        };
+        let execution = Arc::new(TestExecutionBackend::default());
+        let runtime =
+            Runtime::with_fs_store_and_execution_backend(options.clone(), execution.clone())
+                .unwrap();
+        runtime.store_config_bundle(test_bundle()).unwrap();
+        let worker = runtime
+            .create_worker(task_request("persist finished submission fence"))
+            .unwrap();
+        execution.set_dispatch_result(
+            WorkerExecutionResult::accepted_submission(
+                WorkerExecutionOperation::Input,
+                "job-attempt-request",
+                "submission-id",
+                protocol::SubmissionDisposition::Started,
+            )
+            .with_worker_state(protocol::WorkerStateSnapshot {
+                state: protocol::WorkerState::Idle,
+                // Production Submit Run does not consume a lifecycle command
+                // id, so this intentionally remains zero.
+                last_command_id: 0,
+                last_finished_submission_request_id: Some("job-attempt-request".to_string()),
+            }),
+        );
+        let mut input = WorkerInput::user("run durable Backend Job attempt");
+        input.submission_request_id = Some("job-attempt-request".to_string());
+        runtime.send_input(&worker.worker_ref, input).unwrap();
+        assert!(
+            !runtime
+                .worker_detail(&worker.worker_ref)
+                .unwrap()
+                .execution_reconstructed
+        );
+
+        let aggregate_path = root
+            .join("workers")
+            .join(worker.worker_id.to_string())
+            .join("worker.json");
+        let aggregate: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&aggregate_path).unwrap()).unwrap();
+        assert_eq!(
+            aggregate["last_finished_submission_request_id"],
+            serde_json::json!("job-attempt-request")
+        );
+        drop(runtime);
+
+        let restored = Runtime::with_fs_store_and_execution_backend(
+            options,
+            Arc::new(TestExecutionBackend::default()),
+        )
+        .unwrap();
+        let restored_detail = restored.worker_detail(&worker.worker_ref).unwrap();
+        assert!(restored_detail.execution_reconstructed);
+        let restored_state = restored_detail
+            .worker_state
+            .expect("automatic restore publishes fresh Worker state");
+        assert_eq!(restored_state.last_command_id, 0);
+        assert_eq!(
+            restored_state
+                .last_finished_submission_request_id
+                .as_deref(),
+            Some("job-attempt-request")
+        );
+
+        let _ = std::fs::remove_dir_all(root);
+    }
+
     #[cfg(feature = "fs-store")]
     #[test]
     fn fs_store_explicitly_retries_rejected_automatic_restore() {

crates/worker-runtime/src/worker_backend.rs

@@ -186,6 +186,16 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static {
         Err(session_store::RetainedHistoryReadError::RetentionMissing)
     }
 
+    fn retained_session_attachment(
+        &self,
+        _worker_ref: &WorkerRef,
+        _session_id: &str,
+        _attachment_id: &str,
+    ) -> Result<session_store::RetainedSessionAttachment, session_store::RetainedAttachmentReadError>
+    {
+        Err(session_store::RetainedAttachmentReadError::RetentionMissing)
+    }
+
     async fn spawn_controller(
         &self,
         request: WorkerExecutionSpawnRequest,
@@ -1116,6 +1126,27 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
         )
     }
 
+    fn retained_session_attachment(
+        &self,
+        worker_ref: &WorkerRef,
+        session_id: &str,
+        attachment_id: &str,
+    ) -> Result<session_store::RetainedSessionAttachment, session_store::RetainedAttachmentReadError>
+    {
+        let aggregate_dir = self
+            .worker_aggregate_dir(worker_ref)
+            .map_err(|_| session_store::RetainedAttachmentReadError::RetentionMissing)?;
+        let worker_name = Self::runtime_worker_name_for_ref(worker_ref);
+        session_store::read_retained_session_attachment(
+            &aggregate_dir,
+            &worker_name,
+            session_id,
+            attachment_id,
+            session_store::DEFAULT_RETAINED_HISTORY_MAX_SCAN_BYTES,
+            10 * 1024 * 1024,
+        )
+    }
+
     fn observe_workspace_prompt_projection(
         &self,
         projection: worker::WorkspacePromptProjection,
@@ -2412,6 +2443,36 @@ where
         }
     }
 
+    fn worker_session_attachment(
+        &self,
+        request: crate::execution::WorkerSessionAttachmentRequest,
+    ) -> Result<session_store::RetainedSessionAttachment, session_store::RetainedAttachmentReadError>
+    {
+        let live = self
+            .workers
+            .lock()
+            .map_err(|_| session_store::RetainedAttachmentReadError::StorageUnavailable)?
+            .get(&request.worker_ref)
+            .map(|execution| execution.handle.clone());
+        if let Some(handle) = live {
+            match handle.session_attachment(
+                &request.session_id,
+                &request.attachment_id,
+                10 * 1024 * 1024,
+            ) {
+                Ok(attachment) => return Ok(attachment),
+                Err(session_store::RetainedAttachmentReadError::SessionMismatch)
+                | Err(session_store::RetainedAttachmentReadError::NotFound) => {}
+                Err(error) => return Err(error),
+            }
+        }
+        self.factory.retained_session_attachment(
+            &request.worker_ref,
+            &request.session_id,
+            &request.attachment_id,
+        )
+    }
+
     fn observe_workspace_prompt_projection(
         &self,
         projection: worker::WorkspacePromptProjection,
@@ -3824,11 +3885,13 @@ mod tests {
                 protocol::WorkerRunState::Running,
             )),
             last_command_id: 2,
+            last_finished_submission_request_id: Some("request-old".to_string()),
         };
         let current = Arc::new(RwLock::new(running));
         let idle = protocol::WorkerStateSnapshot {
             state: protocol::WorkerState::Idle,
             last_command_id: 0,
+            last_finished_submission_request_id: None,
         };
         let mut replacement = Event::WorkerState {
             snapshot: idle.clone(),
@@ -3841,6 +3904,7 @@ mod tests {
                 protocol::WorkerRunState::Paused,
             )),
             last_command_id: 3,
+            last_finished_submission_request_id: None,
         };
         let mut acknowledgement = Event::CommandAcknowledged {
             acknowledgement: protocol::WorkerCommandAcknowledgement {

crates/worker/src/controller.rs

@@ -129,6 +129,27 @@ impl WorkerHandle {
         self.sink.subscribe_with_snapshot().0
     }
 
+    pub fn session_attachment(
+        &self,
+        session_id: &str,
+        attachment_id: &str,
+        max_attachment_bytes: u64,
+    ) -> Result<session_store::RetainedSessionAttachment, session_store::RetainedAttachmentReadError>
+    {
+        if session_id == self.session_id.to_string() {
+            return session_store::read_session_attachment_from_entries(
+                &self.committed_entries(),
+                attachment_id,
+                max_attachment_bytes,
+            );
+        }
+        self.spawned_registry
+            .session_attachment(session_id, attachment_id, max_attachment_bytes)
+            .unwrap_or(Err(
+                session_store::RetainedAttachmentReadError::SessionMismatch,
+            ))
+    }
+
     pub fn snapshot_event(&self) -> Event {
         self.snapshot_event_with_entry_subscription().0
     }
@@ -300,6 +321,7 @@ async fn finish_controller_run<C, St>(
     runtime_dir: &RuntimeDir,
     working_event_tx: &broadcast::Sender<Event>,
     new_status: WorkerStatus,
+    finished_submission_request_id: Option<String>,
 ) where
     C: LlmClient + Clone + 'static,
     St: Store + WorkerMetadataStore + Clone + 'static,
@@ -320,7 +342,20 @@ async fn finish_controller_run<C, St>(
         Some(worker.total_tokens().tokens),
         protocol::ContextTokenSource::Estimated,
     );
-    set_controller_status(shared_state, runtime_dir, working_event_tx, new_status).await;
+    if let Some(submission_request_id) = finished_submission_request_id {
+        let state = match new_status {
+            WorkerStatus::Idle | WorkerStatus::Stopped => WorkerState::Idle,
+            WorkerStatus::Running => {
+                WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Running))
+            }
+            WorkerStatus::Paused => WorkerState::Busy(WorkerBusyState::Run(WorkerRunState::Paused)),
+        };
+        let snapshot = shared_state.finish_submission_run(state, submission_request_id);
+        let _ = runtime_dir.write_status(shared_state).await;
+        let _ = working_event_tx.send(Event::WorkerState { snapshot });
+    } else {
+        set_controller_status(shared_state, runtime_dir, working_event_tx, new_status).await;
+    }
 }
 
 /// Pending turn launch staged by an event handler for the next outer-loop
@@ -2023,6 +2058,10 @@ async fn controller_loop<C, St>(
             // interrupted/error turn from being carried into the next snapshot.
             worker.clear_in_flight_events();
             let parent_originated = run.is_parent_originated();
+            let finished_submission_request_id = match &run {
+                PendingRun::Submit(submission) => Some(submission.submission_request_id.clone()),
+                PendingRun::RunForNotification { .. } | PendingRun::Resume => None,
+            };
             let user_input_submit = matches!(&run, PendingRun::Submit(_));
             if !user_input_submit {
                 set_controller_status(
@@ -2121,12 +2160,16 @@ async fn controller_loop<C, St>(
                     }
                 }
             }
+            let finished_submission_request_id = (new_status != WorkerStatus::Paused)
+                .then_some(finished_submission_request_id)
+                .flatten();
             finish_controller_run(
                 &mut worker,
                 &shared_state,
                 &runtime_dir,
                 &working_event_tx,
                 new_status,
+                finished_submission_request_id,
             )
             .await;
             if shutdown {

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

@@ -522,6 +522,7 @@ fn workspace_worker_create_request(
                 relative_cwd: attachment.relative_cwd,
             })
             .collect(),
+        feature_connections: Default::default(),
         control_operation_id: Some(control_operation_id),
     })
 }

crates/worker/src/internal_worker.rs

@@ -335,6 +335,7 @@ fn send_internal_worker_state(
     let _ = event_tx.send(Event::WorkerState {
         snapshot: protocol::WorkerStateSnapshot {
             last_command_id: 0,
+            last_finished_submission_request_id: None,
             state,
         },
     });
@@ -521,6 +522,19 @@ impl InternalWorkerSessionHandle {
             .unwrap_or_default()
     }
 
+    pub(crate) fn session_attachment(
+        &self,
+        attachment_id: &str,
+        max_attachment_bytes: u64,
+    ) -> Result<session_store::RetainedSessionAttachment, session_store::RetainedAttachmentReadError>
+    {
+        session_store::read_session_attachment_from_entries(
+            &self.entries(),
+            attachment_id,
+            max_attachment_bytes,
+        )
+    }
+
     pub(crate) async fn send(
         &self,
         input: impl Into<String>,
@@ -1447,6 +1461,50 @@ permission = "write"
         assert_eq!(result.identity.kind, "test");
     }
 
+    #[tokio::test]
+    async fn visible_internal_session_resolves_its_projected_image_attachment() {
+        let (handle, _events) =
+            test_internal_worker_session(InternalWorkerVisibility::ParentClient);
+        let image = vec![1, 3, 5, 7];
+        let entry_id = "internal-image-entry";
+        let attachment_id =
+            session_store::session_tool_attachment_id(entry_id, 0, "image/png", &image);
+        handle.publish_test_entry(LogEntry::AnnotatedToolResult {
+            ts: 1,
+            entry: session_store::LoggedHistoryEntry {
+                item: session_store::LoggedItem::ToolResult {
+                    call_id: "internal-image-call".into(),
+                    summary: "internal screenshot".into(),
+                    content: None,
+                    attachments: vec![session_store::logged_item::LoggedAttachment::Image {
+                        mime_type: "image/png".into(),
+                        data: image.clone(),
+                    }],
+                    disposition: agen::tool::ToolResultDisposition::Success,
+                    is_error: false,
+                },
+                metadata: session_store::LoggedSessionHistoryMetadata {
+                    entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()),
+                    origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
+                    derivation: None,
+                },
+            },
+        });
+
+        let projected = handle.protocol_snapshot();
+        let protocol::SessionSnapshotEntryData::ToolResult { attachments, .. } =
+            &projected.session.entries[0].data
+        else {
+            panic!("expected projected tool result");
+        };
+        assert_eq!(attachments[0].attachment_id, attachment_id);
+        let attachment = handle
+            .session_attachment(&attachment_id, 1024)
+            .expect("internal image body");
+        assert_eq!(attachment.media_type, "image/png");
+        assert_eq!(attachment.data, image);
+    }
+
     #[test]
     fn internal_turn_result_mapping_is_exhaustive() {
         let cases = [

crates/worker/src/shared_state.rs

@@ -117,6 +117,23 @@ impl WorkerSharedState {
         snapshot.clone()
     }
 
+    /// Publish the post-Run state and the exact Submit request whose controller
+    /// invocation has returned. Updating both in one snapshot prevents observers
+    /// from treating an earlier Idle state as evidence for a later Run.
+    pub(crate) fn finish_submission_run(
+        &self,
+        state: WorkerState,
+        submission_request_id: String,
+    ) -> WorkerStateSnapshot {
+        let mut snapshot = self
+            .state
+            .write()
+            .expect("worker state lock poisoned; refusing an inferred fallback state");
+        snapshot.state = state;
+        snapshot.last_finished_submission_request_id = Some(submission_request_id);
+        snapshot.clone()
+    }
+
     pub(crate) fn admit_command(
         &self,
         envelope: WorkerCommandEnvelope,
@@ -305,6 +322,14 @@ mod tests {
         assert_eq!(snapshot.state, paused);
         assert_eq!(snapshot.last_command_id, 0);
         assert_eq!(state.catalog_status(), WorkerStatus::Paused);
+
+        let finished = state.finish_submission_run(WorkerState::Idle, "request-1".to_string());
+        assert_eq!(finished.state, WorkerState::Idle);
+        assert_eq!(
+            finished.last_finished_submission_request_id.as_deref(),
+            Some("request-1")
+        );
+        assert_eq!(finished.last_command_id, 0);
     }
 
     #[test]
@@ -319,6 +344,7 @@ mod tests {
             state.snapshot(),
             WorkerStateSnapshot {
                 last_command_id: 9,
+                last_finished_submission_request_id: None,
                 state: WorkerState::Idle,
             }
         );

crates/worker/src/spawn/registry.rs

@@ -774,6 +774,41 @@ impl SpawnedWorkerRegistry {
         });
     }
 
+    pub(crate) fn session_attachment(
+        &self,
+        session_id: &str,
+        attachment_id: &str,
+        max_attachment_bytes: u64,
+    ) -> Option<
+        Result<
+            session_store::RetainedSessionAttachment,
+            session_store::RetainedAttachmentReadError,
+        >,
+    > {
+        let internal_session = self
+            .internal_records
+            .lock()
+            .ok()?
+            .iter()
+            .find(|record| {
+                record.session.visibility() == InternalWorkerVisibility::ParentClient
+                    && record.session.session_id_string() == session_id
+            })
+            .map(|record| record.session.clone());
+        let session = internal_session.or_else(|| {
+            self.service_records
+                .lock()
+                .ok()?
+                .iter()
+                .find(|record| {
+                    record.session.visibility() == InternalWorkerVisibility::ParentClient
+                        && record.session.session_id_string() == session_id
+                })
+                .map(|record| record.session.clone())
+        })?;
+        Some(session.session_attachment(attachment_id, max_attachment_bytes))
+    }
+
     pub(crate) fn internal_worker_snapshots(&self) -> Vec<InternalWorkerSnapshot> {
         let parent_session_id = self
             .parent_protocol
@@ -1258,6 +1293,59 @@ mod tests {
         )
     }
 
+    #[tokio::test]
+    async fn parent_visible_internal_attachment_resolves_by_session_identity() {
+        let registry = registry();
+        let (visible, _visible_tx) =
+            record("visible", InternalWorkerVisibility::ParentClient).await;
+        let session_id = visible.session.session_id_string();
+        let image = vec![2, 4, 6, 8];
+        let entry_id = "visible-image-entry";
+        let attachment_id =
+            session_store::session_tool_attachment_id(entry_id, 0, "image/png", &image);
+        visible
+            .session
+            .publish_test_entry(LogEntry::AnnotatedToolResult {
+                ts: 1,
+                entry: session_store::LoggedHistoryEntry {
+                    item: session_store::LoggedItem::ToolResult {
+                        call_id: "visible-image-call".into(),
+                        summary: "visible image".into(),
+                        content: None,
+                        attachments: vec![session_store::logged_item::LoggedAttachment::Image {
+                            mime_type: "image/png".into(),
+                            data: image.clone(),
+                        }],
+                        disposition: agen::tool::ToolResultDisposition::Success,
+                        is_error: false,
+                    },
+                    metadata: session_store::LoggedSessionHistoryMetadata {
+                        entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()),
+                        origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
+                        derivation: None,
+                    },
+                },
+            });
+        registry.internal_records.lock().unwrap().push(visible);
+
+        let attachment = registry
+            .session_attachment(&session_id, &attachment_id, 1024)
+            .expect("visible session")
+            .expect("visible attachment");
+        assert_eq!(attachment.data, image);
+
+        let (private, _private_tx) =
+            record("private", InternalWorkerVisibility::ServicePrivate).await;
+        let private_session_id = private.session.session_id_string();
+        registry.internal_records.lock().unwrap().push(private);
+        assert!(
+            registry
+                .session_attachment(&private_session_id, &attachment_id, 1024)
+                .is_none(),
+            "service-private sessions must not become parent-fetchable"
+        );
+    }
+
     #[tokio::test]
     async fn visible_internal_worker_starts_with_own_live_snapshot_metadata() {
         let registry = registry();

crates/worker/tests/controller_test.rs

@@ -1299,12 +1299,10 @@ async fn run_end_returns_to_idle_without_busy_status() {
     let worker = make_worker(client).await;
     let handle = spawn_controller(worker).await;
     let mut rx = handle.subscribe();
+    let submission_request_id = protocol::new_submission_request_id();
 
     handle
-        .send(Method::submit_text(
-            protocol::new_submission_request_id(),
-            "Hello",
-        ))
+        .send(Method::submit_text(submission_request_id.clone(), "Hello"))
         .await
         .unwrap();
 
@@ -1319,7 +1317,11 @@ async fn run_end_returns_to_idle_without_busy_status() {
                         saw_run_end = true;
                     }
                     Ok(Event::WorkerState { snapshot })
-                        if saw_run_end && snapshot.catalog_status() == WorkerStatus::Idle => {
+                        if saw_run_end
+                            && snapshot.catalog_status() == WorkerStatus::Idle
+                            && snapshot.last_finished_submission_request_id.as_deref()
+                                == Some(submission_request_id.as_str()) =>
+                    {
                         saw_idle_status = true;
                         break;
                     }
@@ -1334,7 +1336,7 @@ async fn run_end_returns_to_idle_without_busy_status() {
     assert!(saw_run_end, "expected RunEnd::Finished");
     assert!(
         saw_idle_status,
-        "expected idle status immediately after RunEnd"
+        "expected exact Submit teardown fence in idle status immediately after RunEnd"
     );
     assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Idle);
 }

crates/workspace-server/Cargo.toml

@@ -40,6 +40,7 @@ tokio-tungstenite = { workspace = true, features = ["rustls-tls-webpki-roots"] }
 worker.workspace = true
 workdir = { workspace = true, features = ["http-client"] }
 worker-runtime.workspace = true
+session-store.workspace = true
 toml.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
@@ -51,7 +52,6 @@ webauthn-rs = { workspace = true }
 
 [dev-dependencies]
 agen.workspace = true
-session-store.workspace = true
 tempfile.workspace = true
 tools.workspace = true
 yoi = { path = "../yoi" }

crates/workspace-server/src/backend_job.rs

@@ -284,6 +284,38 @@ pub struct BackendJobRecord {
     pub completed_at: Option<String>,
 }
 
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum BackendJobWorkerCleanupState {
+    Pending,
+    Executing,
+    Completed,
+    Failed,
+}
+
+impl BackendJobWorkerCleanupState {
+    pub(crate) fn as_str(self) -> &'static str {
+        match self {
+            Self::Pending => "pending",
+            Self::Executing => "executing",
+            Self::Completed => "completed",
+            Self::Failed => "failed",
+        }
+    }
+
+    pub(crate) fn parse(value: &str) -> Result<Self> {
+        match value {
+            "pending" => Ok(Self::Pending),
+            "executing" => Ok(Self::Executing),
+            "completed" => Ok(Self::Completed),
+            "failed" => Ok(Self::Failed),
+            other => Err(Error::Store(format!(
+                "unknown Backend Job Worker cleanup state `{other}`"
+            ))),
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct BackendJobAttemptRecord {
     pub workspace_id: String,
@@ -300,6 +332,11 @@ pub struct BackendJobAttemptRecord {
     pub result_digest: Option<String>,
     pub failure_category: Option<String>,
     pub failure_detail: Option<String>,
+    pub worker_cleanup_state: Option<BackendJobWorkerCleanupState>,
+    pub worker_cleanup_failure_category: Option<String>,
+    pub worker_cleanup_failure_detail: Option<String>,
+    pub worker_cleanup_updated_at: Option<String>,
+    pub worker_cleanup_completed_at: Option<String>,
     pub created_at: String,
     pub updated_at: String,
     pub completed_at: Option<String>,

crates/workspace-server/src/hosts.rs

@@ -259,6 +259,9 @@ pub struct InternalWorkerSummary {
     pub availability: protocol::subscription::SubscriptionWorkerAvailability,
     /// Runtime catalog lifecycle compatibility state.
     pub state: String,
+    /// Runtime-process reconstruction evidence used only for lifecycle coordination.
+    #[serde(skip)]
+    pub execution_reconstructed: bool,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub worker_state: Option<protocol::WorkerStateSnapshot>,
     pub last_seen_at: Option<String>,
@@ -770,6 +773,12 @@ pub struct WorkerProxyConnectPoint {
     pub diagnostics: Vec<RuntimeDiagnostic>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RuntimeSessionAttachment {
+    pub media_type: String,
+    pub data: Vec<u8>,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum RuntimeRegistryError {
     InvalidIdentifier {
@@ -1220,6 +1229,23 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
         })
     }
 
+    fn worker_session_attachment(
+        &self,
+        _workspace_id: &str,
+        worker_ref: &EmbeddedWorkerRef,
+        _session_id: &str,
+        _attachment_id: &str,
+    ) -> Result<RuntimeSessionAttachment, RuntimeRegistryError> {
+        Err(RuntimeRegistryError::RuntimeOperationFailed {
+            runtime_id: self.runtime_id().to_string(),
+            code: "session_attachment_unsupported".to_string(),
+            message: format!(
+                "Session attachment reads are not supported for Worker `{}`",
+                worker_ref.worker_id
+            ),
+        })
+    }
+
     fn session_public_search(
         &self,
         _request: runtime_api::SessionPublicSearchRequest,
@@ -2096,6 +2122,27 @@ impl RuntimeRegistry {
             })
     }
 
+    pub fn worker_session_attachment(
+        &self,
+        workspace_id: &str,
+        worker: &RuntimeWorkerRef,
+        session_id: &str,
+        attachment_id: &str,
+    ) -> Result<RuntimeSessionAttachment, RuntimeRegistryError> {
+        validate_backend_identifier("runtime_id", &worker.runtime_id)?;
+        validate_backend_identifier("worker_id", &worker.worker_id)?;
+        validate_backend_identifier("session_id", session_id)?;
+        validate_backend_identifier("attachment_id", attachment_id)?;
+        let runtime = self.runtime(&worker.runtime_id)?;
+        let worker_ref =
+            EmbeddedWorkerRef::new(EmbeddedWorkerId::parse(&worker.worker_id).ok_or_else(
+                || RuntimeRegistryError::UnknownWorker {
+                    worker: worker.clone(),
+                },
+            )?);
+        runtime.worker_session_attachment(workspace_id, &worker_ref, session_id, attachment_id)
+    }
+
     pub fn session_public_search(
         &self,
         runtime_id: &str,
@@ -2333,6 +2380,7 @@ impl EmbeddedWorkerRuntime {
                 summary.execution_metadata_available,
             )
             .to_string(),
+            execution_reconstructed: false,
             worker_state: summary.worker_state.clone(),
             last_seen_at: None,
             pinned: false,
@@ -2373,6 +2421,7 @@ impl EmbeddedWorkerRuntime {
             availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
             state: embedded_worker_state_label(detail.status, detail.execution_metadata_available)
                 .to_string(),
+            execution_reconstructed: detail.execution_reconstructed,
             worker_state: detail.worker_state.clone(),
             last_seen_at: None,
             pinned: false,
@@ -3017,6 +3066,46 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
             .map_err(|error| error.to_string())
     }
 
+    fn worker_session_attachment(
+        &self,
+        workspace_id: &str,
+        worker_ref: &EmbeddedWorkerRef,
+        session_id: &str,
+        attachment_id: &str,
+    ) -> Result<RuntimeSessionAttachment, RuntimeRegistryError> {
+        let scope = RuntimeWorkspaceScope::new(workspace_id, "embedded-backend");
+        self.runtime
+            .worker_session_attachment_scoped(
+                &scope,
+                worker_ref,
+                session_id.to_string(),
+                attachment_id.to_string(),
+            )
+            .map(|attachment| RuntimeSessionAttachment {
+                media_type: attachment.media_type,
+                data: attachment.data,
+            })
+            .map_err(|error| RuntimeRegistryError::RuntimeOperationFailed {
+                runtime_id: self.runtime_id.clone(),
+                code: match error {
+                    session_store::RetainedAttachmentReadError::RetentionMissing
+                    | session_store::RetainedAttachmentReadError::ActivePointerMissing => {
+                        "session_attachment_expired"
+                    }
+                    session_store::RetainedAttachmentReadError::SessionMismatch
+                    | session_store::RetainedAttachmentReadError::NotFound => {
+                        "session_attachment_not_found"
+                    }
+                    session_store::RetainedAttachmentReadError::ResourceLimit => {
+                        "session_attachment_too_large"
+                    }
+                    _ => "session_attachment_unavailable",
+                }
+                .to_string(),
+                message: error.to_string(),
+            })
+    }
+
     fn session_public_search(
         &self,
         request: runtime_api::SessionPublicSearchRequest,
@@ -3732,7 +3821,7 @@ fn workspace_runtime_operation(method: &str, path_and_query: &str) -> &'static s
     }
     if path.ends_with("/input")
         || path.ends_with("/restore")
-        || path.contains("/attachments")
+        || (path.contains("/attachments") && method != "GET")
         || path.contains("/workdir-attachments")
     {
         return "workers:input";
@@ -4059,6 +4148,64 @@ impl RemoteWorkerRuntime {
         self.send_json(path, "GET", &[], self.http.get(self.endpoint(path)))
     }
 
+    fn get_session_attachment(
+        &self,
+        path: &str,
+    ) -> Result<RuntimeSessionAttachment, RuntimeDiagnostic> {
+        let runtime_id = self.runtime_id.clone();
+        let workspace_id = self.workspace_id.clone();
+        let bearer_token = self.bearer_token.clone();
+        let capability_token = match &self.workspace_authorization {
+            Some(authorization) => Some(authorization.issue(
+                "GET",
+                path,
+                workspace_runtime_operation("GET", path),
+                worker_id_from_remote_path(path).as_deref(),
+                &[],
+            )?),
+            None => None,
+        };
+        let request = self.http.get(self.endpoint(path));
+        run_blocking_http(move || {
+            let mut request = request.header(RUNTIME_WORKSPACE_SCOPE_HEADER, &workspace_id);
+            if let Some(token) = capability_token.as_deref().or(bearer_token.as_deref()) {
+                request = request.header(AUTHORIZATION, format!("Bearer {token}"));
+            }
+            let response = request
+                .send()
+                .map_err(|error| remote_reqwest_diagnostic(&runtime_id, error))?;
+            let status = response.status();
+            if !status.is_success() {
+                return Err(remote_http_status_diagnostic(&runtime_id, status, response));
+            }
+            let media_type = response
+                .headers()
+                .get(CONTENT_TYPE)
+                .and_then(|value| value.to_str().ok())
+                .unwrap_or("application/octet-stream")
+                .to_string();
+            let mut data = Vec::new();
+            response
+                .take((MAX_REMOTE_RUNTIME_RESPONSE_BYTES + 1) as u64)
+                .read_to_end(&mut data)
+                .map_err(|_| {
+                    diagnostic(
+                        "remote_runtime_response_read_failed",
+                        HostDiagnosticSeverity::Error,
+                        "Remote Runtime attachment response could not be read".to_string(),
+                    )
+                })?;
+            if data.len() > MAX_REMOTE_RUNTIME_RESPONSE_BYTES {
+                return Err(diagnostic(
+                    "remote_runtime_response_too_large",
+                    HostDiagnosticSeverity::Error,
+                    "Remote Runtime attachment response exceeded the allowed size".to_string(),
+                ));
+            }
+            Ok(RuntimeSessionAttachment { media_type, data })
+        })
+    }
+
     fn post_json<B, T>(&self, path: &str, body: &B) -> Result<T, RuntimeDiagnostic>
     where
         B: Serialize + ?Sized,
@@ -4226,6 +4373,7 @@ impl RemoteWorkerRuntime {
                 summary.execution_metadata_available,
             )
             .to_string(),
+            execution_reconstructed: false,
             worker_state: summary.worker_state.clone(),
             last_seen_at: None,
             pinned: false,
@@ -4264,6 +4412,7 @@ impl RemoteWorkerRuntime {
             availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
             state: embedded_worker_state_label(detail.status, detail.execution_metadata_available)
                 .to_string(),
+            execution_reconstructed: detail.execution_reconstructed,
             worker_state: detail.worker_state.clone(),
             last_seen_at: None,
             pinned: false,
@@ -5093,6 +5242,29 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
         .map_err(|diagnostic| diagnostic.message)
     }
 
+    fn worker_session_attachment(
+        &self,
+        workspace_id: &str,
+        worker_ref: &EmbeddedWorkerRef,
+        session_id: &str,
+        attachment_id: &str,
+    ) -> Result<RuntimeSessionAttachment, RuntimeRegistryError> {
+        let path = format!(
+            "/v1/workers/{}/sessions/{}/attachments/{}?workspace_id={}",
+            url_path_segment_encode(&worker_ref.worker_id.to_string()),
+            url_path_segment_encode(session_id),
+            url_path_segment_encode(attachment_id),
+            url_query_value_encode(workspace_id),
+        );
+        self.get_session_attachment(&path).map_err(|diagnostic| {
+            RuntimeRegistryError::RuntimeOperationFailed {
+                runtime_id: self.runtime_id.clone(),
+                code: diagnostic.code,
+                message: diagnostic.message,
+            }
+        })
+    }
+
     fn session_public_search(
         &self,
         request: runtime_api::SessionPublicSearchRequest,
@@ -6106,6 +6278,7 @@ pub fn placeholder_worker(host_id: impl Into<String>) -> InternalWorkerSummary {
         },
         availability: protocol::subscription::SubscriptionWorkerAvailability::Unavailable,
         state: "unsupported".to_string(),
+        execution_reconstructed: false,
         worker_state: None,
         last_seen_at: None,
         pinned: false,
@@ -6834,6 +7007,7 @@ mod tests {
                     },
                     availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
                     state: "available".to_string(),
+                    execution_reconstructed: false,
                     worker_state: None,
                     last_seen_at: None,
                     pinned: false,
@@ -7648,6 +7822,13 @@ mod tests {
             ),
             "workers:input"
         );
+        assert_eq!(
+            workspace_runtime_operation(
+                "GET",
+                &format!("/v1/workers/{worker_id}/sessions/session-1/attachments/attachment-1")
+            ),
+            "workers:read"
+        );
         assert_eq!(
             workspace_runtime_operation("POST", SSH_HOST_KEY_PROBE_PATH),
             SSH_HOST_KEY_PROBE_OPERATION

crates/workspace-server/src/latest_schema.sql

@@ -875,6 +875,13 @@ CREATE TABLE backend_job_attempts (
             result_digest TEXT,
             failure_category TEXT,
             failure_detail TEXT,
+            worker_cleanup_state TEXT CHECK (
+                worker_cleanup_state IN ('pending', 'executing', 'completed', 'failed')
+            ),
+            worker_cleanup_failure_category TEXT,
+            worker_cleanup_failure_detail TEXT,
+            worker_cleanup_updated_at TEXT,
+            worker_cleanup_completed_at TEXT,
             created_at TEXT NOT NULL,
             updated_at TEXT NOT NULL,
             completed_at TEXT,
@@ -889,6 +896,8 @@ CREATE TABLE backend_job_attempts (
         );
 CREATE INDEX backend_job_attempts_recovery
         ON backend_job_attempts(workspace_id, state, deadline_at);
+CREATE INDEX backend_job_attempts_worker_cleanup
+        ON backend_job_attempts(workspace_id, worker_cleanup_state, updated_at);
 CREATE TABLE backend_job_deliveries (
             workspace_id TEXT NOT NULL,
             delivery_id TEXT NOT NULL,

crates/workspace-server/src/profile_settings.rs

@@ -447,6 +447,23 @@ pub fn build_virtual_profile_config_bundle_with_prompt_projection(
     Ok(Some(bundle))
 }
 
+pub(crate) fn resolve_profile_manifest_from_config_bundle(
+    bundle: &ConfigBundle,
+    selector: &str,
+) -> Result<manifest::WorkerManifest> {
+    let archive = bundle.profile_source_archive.as_ref().ok_or_else(|| {
+        Error::Config(format!(
+            "resolved Profile {selector:?} is missing its source archive"
+        ))
+    })?;
+    let verified = archive.verify().map_err(|error| {
+        Error::Config(format!("failed to verify Profile {selector:?}: {error}"))
+    })?;
+    verified
+        .resolve_profile(selector, Path::new("/"), "worker-launch-validation")
+        .map_err(|error| Error::Config(format!("failed to resolve Profile {selector:?}: {error}")))
+}
+
 fn build_virtual_profile_archive(
     selector: &str,
     entry: &VirtualProfileEntry,

crates/workspace-server/src/store.rs

@@ -21,13 +21,14 @@ use crate::backend_job::{
     BackendJobAttemptRecord, BackendJobAttemptState, BackendJobDeliveryRecord,
     BackendJobDeliveryState, BackendJobRecord, BackendJobRequest, BackendJobReservation,
     BackendJobResultAcceptance, BackendJobResultSubmission, BackendJobState,
-    BackendJobWorkerBinding, attempt_id, bounded_failure_detail, result_digest,
+    BackendJobWorkerBinding, BackendJobWorkerCleanupState, attempt_id, bounded_failure_detail,
+    result_digest,
 };
 use crate::workspace_deletion::WorkspaceDeletionStore;
 use crate::{Error, Result};
 
 const OLDEST_SCHEMA_VERSION: i64 = 50;
-const LATEST_SCHEMA_VERSION: i64 = 76;
+const LATEST_SCHEMA_VERSION: i64 = 77;
 const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
 const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings";
 const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit";
@@ -74,6 +75,8 @@ const RUNTIME_SCOPED_WORKER_IDENTITY_MIGRATION_NAME: &str =
     "Runtime-scoped durable Worker catalog identity";
 const ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME: &str =
     "preserve explicit observe grants for committed Worker Session archives";
+const BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME: &str =
+    "durable terminal Backend Job Worker cleanup";
 const TICKET_SCHEMA_VERSION_WITH_TARGETS: i64 = 7;
 const TICKET_SCHEMA_BASELINE_NAME: &str = "ticket schema baseline";
 const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#"
@@ -251,6 +254,11 @@ const MIGRATIONS: &[Migration] = &[
         name: ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME,
         apply: migrate_archive_observe_grants_v75_to_v76,
     },
+    Migration {
+        version: 77,
+        name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME,
+        apply: migrate_backend_job_worker_cleanup_v76_to_v77,
+    },
 ];
 
 #[derive(Clone, Copy)]
@@ -1592,6 +1600,34 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
         failure_detail: &str,
         now: &str,
     ) -> Result<BackendJobAttemptRecord>;
+    fn recover_backend_job_worker_cleanups(
+        &self,
+        workspace_id: &str,
+        now: &str,
+        limit: usize,
+    ) -> Result<Vec<BackendJobAttemptRecord>>;
+    fn list_backend_job_worker_cleanups(
+        &self,
+        workspace_id: &str,
+        limit: usize,
+    ) -> Result<Vec<BackendJobAttemptRecord>>;
+    fn claim_backend_job_worker_cleanup(
+        &self,
+        workspace_id: &str,
+        job_id: &str,
+        attempt_id: &str,
+        now: &str,
+    ) -> Result<(BackendJobAttemptRecord, bool)>;
+    fn finish_backend_job_worker_cleanup(
+        &self,
+        workspace_id: &str,
+        job_id: &str,
+        attempt_id: &str,
+        state: BackendJobWorkerCleanupState,
+        failure_category: Option<&str>,
+        failure_detail: Option<&str>,
+        now: &str,
+    ) -> Result<BackendJobAttemptRecord>;
     fn reserve_backend_job_delivery(
         &self,
         workspace_id: &str,
@@ -6648,7 +6684,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                 ));
             }
             let attempt_changed = tx.execute(
-                "UPDATE backend_job_attempts SET state = 'completed', result_json = ?4, result_digest = ?5, failure_category = NULL, failure_detail = NULL, completed_at = ?6, updated_at = ?6 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state = 'dispatched'",
+                "UPDATE backend_job_attempts SET state = 'completed', result_json = ?4, result_digest = ?5, failure_category = NULL, failure_detail = NULL, worker_cleanup_state = 'pending', worker_cleanup_failure_category = NULL, worker_cleanup_failure_detail = NULL, worker_cleanup_updated_at = ?6, worker_cleanup_completed_at = NULL, completed_at = ?6, updated_at = ?6 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state = 'dispatched'",
                 params![workspace_id, submission.job_id, submission.attempt_id, result_json, digest, now],
             )?;
             let job_changed = tx.execute(
@@ -6707,7 +6743,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                 return Ok(attempt);
             }
             tx.execute(
-                "UPDATE backend_job_attempts SET state = ?4, failure_category = ?5, failure_detail = ?6, completed_at = ?7, updated_at = ?7 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state IN ('reserved', 'dispatching', 'dispatched')",
+                "UPDATE backend_job_attempts SET state = ?4, failure_category = ?5, failure_detail = ?6, worker_cleanup_state = CASE WHEN worker_id IS NULL THEN 'completed' ELSE 'pending' END, worker_cleanup_failure_category = NULL, worker_cleanup_failure_detail = NULL, worker_cleanup_updated_at = ?7, worker_cleanup_completed_at = CASE WHEN worker_id IS NULL THEN ?7 ELSE NULL END, completed_at = ?7, updated_at = ?7 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state IN ('reserved', 'dispatching', 'dispatched')",
                 params![workspace_id, job_id, attempt_id, state.as_str(), failure_category, detail, now],
             )?;
             if attempt.attempt == job.current_attempt && job.state == BackendJobState::Pending {
@@ -6728,6 +6764,154 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         })
     }
 
+    fn recover_backend_job_worker_cleanups(
+        &self,
+        workspace_id: &str,
+        now: &str,
+        limit: usize,
+    ) -> Result<Vec<BackendJobAttemptRecord>> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            tx.execute(
+                "UPDATE backend_job_attempts SET worker_cleanup_state = 'pending', worker_cleanup_failure_category = 'cleanup_interrupted', worker_cleanup_failure_detail = 'Backend restart interrupted Worker cleanup; retrying canonical removal', worker_cleanup_updated_at = ?2 WHERE workspace_id = ?1 AND worker_cleanup_state = 'executing'",
+                params![workspace_id, now],
+            )?;
+            let mut statement = tx.prepare(
+                "SELECT job_id, attempt_id FROM backend_job_attempts WHERE workspace_id = ?1 AND state IN ('completed', 'failed', 'unknown') AND runtime_id IS NOT NULL AND worker_cleanup_state IN ('pending', 'failed') ORDER BY worker_cleanup_updated_at, created_at LIMIT ?2",
+            )?;
+            let identities = statement
+                .query_map(params![workspace_id, limit as i64], |row| {
+                    Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
+                })?
+                .collect::<rusqlite::Result<Vec<_>>>()?;
+            drop(statement);
+            let attempts = identities
+                .into_iter()
+                .map(|(job_id, attempt_id)| {
+                    read_backend_job_attempt(&tx, workspace_id, &job_id, &attempt_id)?.ok_or_else(
+                        || Error::Store("recoverable Backend Job Worker cleanup disappeared".to_string()),
+                    )
+                })
+                .collect::<Result<Vec<_>>>()?;
+            tx.commit()?;
+            Ok(attempts)
+        })
+    }
+
+    fn list_backend_job_worker_cleanups(
+        &self,
+        workspace_id: &str,
+        limit: usize,
+    ) -> Result<Vec<BackendJobAttemptRecord>> {
+        self.with_conn(|conn| {
+            let mut statement = conn.prepare(
+                "SELECT job_id, attempt_id FROM backend_job_attempts WHERE workspace_id = ?1 AND state IN ('completed', 'failed', 'unknown') AND runtime_id IS NOT NULL AND worker_cleanup_state IN ('pending', 'failed') ORDER BY worker_cleanup_updated_at, created_at LIMIT ?2",
+            )?;
+            let identities = statement
+                .query_map(params![workspace_id, limit as i64], |row| {
+                    Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
+                })?
+                .collect::<rusqlite::Result<Vec<_>>>()?;
+            identities
+                .into_iter()
+                .map(|(job_id, attempt_id)| {
+                    read_backend_job_attempt(conn, workspace_id, &job_id, &attempt_id)?.ok_or_else(
+                        || Error::Store("recoverable Backend Job Worker cleanup disappeared".to_string()),
+                    )
+                })
+                .collect()
+        })
+    }
+
+    fn claim_backend_job_worker_cleanup(
+        &self,
+        workspace_id: &str,
+        job_id: &str,
+        attempt_id: &str,
+        now: &str,
+    ) -> Result<(BackendJobAttemptRecord, bool)> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)?
+                .ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job attempt `{attempt_id}`")))?;
+            if !attempt.state.terminal() || attempt.worker.is_none() {
+                return Err(Error::InvalidInput(
+                    "Backend Job Worker cleanup requires a terminal bound attempt".to_string(),
+                ));
+            }
+            if !matches!(
+                attempt.worker_cleanup_state,
+                Some(BackendJobWorkerCleanupState::Pending | BackendJobWorkerCleanupState::Failed)
+            ) {
+                tx.commit()?;
+                return Ok((attempt, false));
+            }
+            let changed = tx.execute(
+                "UPDATE backend_job_attempts SET worker_cleanup_state = 'executing', worker_cleanup_failure_category = NULL, worker_cleanup_failure_detail = NULL, worker_cleanup_updated_at = ?4 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND worker_cleanup_state IN ('pending', 'failed')",
+                params![workspace_id, job_id, attempt_id, now],
+            )?;
+            if changed != 1 {
+                return Err(Error::Store(
+                    "Backend Job Worker cleanup claim compare-and-set failed".to_string(),
+                ));
+            }
+            let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)?
+                .ok_or_else(|| Error::Store("claimed Backend Job Worker cleanup disappeared".to_string()))?;
+            tx.commit()?;
+            Ok((attempt, true))
+        })
+    }
+
+    fn finish_backend_job_worker_cleanup(
+        &self,
+        workspace_id: &str,
+        job_id: &str,
+        attempt_id: &str,
+        state: BackendJobWorkerCleanupState,
+        failure_category: Option<&str>,
+        failure_detail: Option<&str>,
+        now: &str,
+    ) -> Result<BackendJobAttemptRecord> {
+        if !matches!(
+            state,
+            BackendJobWorkerCleanupState::Completed | BackendJobWorkerCleanupState::Failed
+        ) {
+            return Err(Error::InvalidInput(
+                "Backend Job Worker cleanup completion requires completed or failed state"
+                    .to_string(),
+            ));
+        }
+        let failure_detail = failure_detail.map(bounded_failure_detail);
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let current = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)?
+                .ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job attempt `{attempt_id}`")))?;
+            if current.worker_cleanup_state == Some(state) {
+                tx.commit()?;
+                return Ok(current);
+            }
+            if current.worker_cleanup_state != Some(BackendJobWorkerCleanupState::Executing) {
+                return Err(Error::InvalidInput(
+                    "Backend Job Worker cleanup is not executing".to_string(),
+                ));
+            }
+            let completed_at = (state == BackendJobWorkerCleanupState::Completed).then_some(now);
+            let changed = tx.execute(
+                "UPDATE backend_job_attempts SET worker_cleanup_state = ?4, worker_cleanup_failure_category = ?5, worker_cleanup_failure_detail = ?6, worker_cleanup_updated_at = ?7, worker_cleanup_completed_at = ?8 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND worker_cleanup_state = 'executing'",
+                params![workspace_id, job_id, attempt_id, state.as_str(), failure_category, failure_detail, now, completed_at],
+            )?;
+            if changed != 1 {
+                return Err(Error::Store(
+                    "Backend Job Worker cleanup completion compare-and-set failed".to_string(),
+                ));
+            }
+            let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)?
+                .ok_or_else(|| Error::Store("finished Backend Job Worker cleanup disappeared".to_string()))?;
+            tx.commit()?;
+            Ok(attempt)
+        })
+    }
+
     fn reserve_backend_job_delivery(
         &self,
         workspace_id: &str,
@@ -10457,14 +10641,16 @@ fn read_backend_job_attempt(
     attempt_id: &str,
 ) -> Result<Option<BackendJobAttemptRecord>> {
     let row = conn.query_row(
-        "SELECT attempt, input_revision, state, runtime_id, worker_id, runtime_run_id, dispatched_at, deadline_at, result_json, result_digest, failure_category, failure_detail, created_at, updated_at, completed_at FROM backend_job_attempts WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3",
+        "SELECT attempt, input_revision, state, runtime_id, worker_id, runtime_run_id, dispatched_at, deadline_at, result_json, result_digest, failure_category, failure_detail, worker_cleanup_state, worker_cleanup_failure_category, worker_cleanup_failure_detail, worker_cleanup_updated_at, worker_cleanup_completed_at, created_at, updated_at, completed_at FROM backend_job_attempts WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3",
         params![workspace_id, job_id, attempt_id],
         |row| Ok((
             row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?,
             row.get::<_, Option<String>>(3)?, row.get::<_, Option<String>>(4)?, row.get::<_, Option<String>>(5)?,
             row.get::<_, Option<String>>(6)?, row.get::<_, String>(7)?, row.get::<_, Option<String>>(8)?,
             row.get::<_, Option<String>>(9)?, row.get::<_, Option<String>>(10)?, row.get::<_, Option<String>>(11)?,
-            row.get::<_, String>(12)?, row.get::<_, String>(13)?, row.get::<_, Option<String>>(14)?,
+            row.get::<_, Option<String>>(12)?, row.get::<_, Option<String>>(13)?, row.get::<_, Option<String>>(14)?,
+            row.get::<_, Option<String>>(15)?, row.get::<_, Option<String>>(16)?, row.get::<_, String>(17)?,
+            row.get::<_, String>(18)?, row.get::<_, Option<String>>(19)?,
         )),
     ).optional()?;
     let Some((
@@ -10480,6 +10666,11 @@ fn read_backend_job_attempt(
         result_digest,
         failure_category,
         failure_detail,
+        worker_cleanup_state,
+        worker_cleanup_failure_category,
+        worker_cleanup_failure_detail,
+        worker_cleanup_updated_at,
+        worker_cleanup_completed_at,
         created_at,
         updated_at,
         completed_at,
@@ -10521,6 +10712,14 @@ fn read_backend_job_attempt(
         result_digest,
         failure_category,
         failure_detail,
+        worker_cleanup_state: worker_cleanup_state
+            .as_deref()
+            .map(BackendJobWorkerCleanupState::parse)
+            .transpose()?,
+        worker_cleanup_failure_category,
+        worker_cleanup_failure_detail,
+        worker_cleanup_updated_at,
+        worker_cleanup_completed_at,
         created_at,
         updated_at,
         completed_at,
@@ -14248,6 +14447,45 @@ fn migrate_archive_observe_grants_v75_to_v76(conn: &Connection) -> Result<()> {
     Ok(())
 }
 
+fn migrate_backend_job_worker_cleanup_v76_to_v77(conn: &Connection) -> Result<()> {
+    let current = current_schema_version(conn)?;
+    if current != 76 {
+        return Err(Error::Store(format!(
+            "expected schema version 76 before {BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME} migration, found {current}"
+        )));
+    }
+    let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
+    tx.execute_batch(
+        r#"
+        ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_state TEXT
+            CHECK (worker_cleanup_state IN ('pending', 'executing', 'completed', 'failed'));
+        ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_failure_category TEXT;
+        ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_failure_detail TEXT;
+        ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_updated_at TEXT;
+        ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_completed_at TEXT;
+        UPDATE backend_job_attempts
+        SET worker_cleanup_state = CASE
+                WHEN runtime_id IS NULL THEN 'completed'
+                ELSE 'pending'
+            END,
+            worker_cleanup_updated_at = updated_at,
+            worker_cleanup_completed_at = CASE
+                WHEN runtime_id IS NULL THEN updated_at
+                ELSE NULL
+            END
+        WHERE state IN ('completed', 'failed', 'unknown');
+        CREATE INDEX backend_job_attempts_worker_cleanup
+            ON backend_job_attempts(workspace_id, worker_cleanup_state, updated_at);
+        "#,
+    )?;
+    tx.execute(
+        "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
+        params![77_i64, BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME],
+    )?;
+    tx.commit()?;
+    Ok(())
+}
+
 fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
     conn.execute_batch("DROP TABLE IF EXISTS typed_ticket_targets;")?;
     conn.execute_batch(include_str!("latest_schema.sql"))?;
@@ -16810,6 +17048,10 @@ mod tests {
                     version: 76,
                     name: ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME.to_string(),
                 },
+                WorkspaceSchemaMigrationStep {
+                    version: 77,
+                    name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(),
+                },
             ]
         );
 
@@ -16893,6 +17135,10 @@ mod tests {
                             RUNTIME_SCOPED_WORKER_IDENTITY_MIGRATION_NAME.to_string(),
                         ),
                         (76, ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME.to_string()),
+                        (
+                            77,
+                            BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(),
+                        ),
                     ]
                 );
                 assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -17160,6 +17406,71 @@ mod tests {
         );
     }
 
+    #[test]
+    fn backend_job_worker_cleanup_migration_queues_legacy_terminal_workers() {
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute_batch(
+            r#"
+            CREATE TABLE __yoi_schema_migrations (
+                version INTEGER PRIMARY KEY,
+                name TEXT NOT NULL
+            );
+            INSERT INTO __yoi_schema_migrations(version, name)
+                VALUES (76, 'workspace schema baseline');
+            CREATE TABLE backend_job_attempts (
+                workspace_id TEXT NOT NULL,
+                job_id TEXT NOT NULL,
+                attempt_id TEXT NOT NULL,
+                state TEXT NOT NULL,
+                runtime_id TEXT,
+                worker_id TEXT,
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+            );
+            INSERT INTO backend_job_attempts VALUES
+                ('w', 'terminal-bound', 'a1', 'completed', 'runtime', 'worker', '1', '2'),
+                ('w', 'terminal-unbound', 'a2', 'failed', NULL, NULL, '1', '3'),
+                ('w', 'running', 'a3', 'dispatched', 'runtime', 'running-worker', '1', '4');
+            "#,
+        )
+        .unwrap();
+
+        migrate_backend_job_worker_cleanup_v76_to_v77(&conn).unwrap();
+
+        let states = conn
+            .prepare(
+                "SELECT job_id, worker_cleanup_state, worker_cleanup_completed_at FROM backend_job_attempts ORDER BY job_id",
+            )
+            .unwrap()
+            .query_map([], |row| {
+                Ok((
+                    row.get::<_, String>(0)?,
+                    row.get::<_, Option<String>>(1)?,
+                    row.get::<_, Option<String>>(2)?,
+                ))
+            })
+            .unwrap()
+            .collect::<rusqlite::Result<Vec<_>>>()
+            .unwrap();
+        assert_eq!(
+            states,
+            vec![
+                ("running".to_string(), None, None,),
+                (
+                    "terminal-bound".to_string(),
+                    Some("pending".to_string()),
+                    None,
+                ),
+                (
+                    "terminal-unbound".to_string(),
+                    Some("completed".to_string()),
+                    Some("3".to_string()),
+                ),
+            ]
+        );
+        assert_eq!(current_schema_version(&conn).unwrap(), 77);
+    }
+
     #[test]
     fn migration_resumes_from_a_valid_partially_applied_chain() {
         let temp = tempfile::tempdir().unwrap();
@@ -17179,7 +17490,7 @@ mod tests {
                 .collect::<Vec<_>>(),
             vec![
                 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72,
-                73, 74, 75, 76
+                73, 74, 75, 76, 77
             ]
         );
         SqliteWorkspaceStore::migrate_database(&path).unwrap();
@@ -17188,7 +17499,7 @@ mod tests {
             current_schema_version(&conn).unwrap(),
             LATEST_SCHEMA_VERSION
         );
-        assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 27);
+        assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 28);
         assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap());
         assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap());
         assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap());

crates/workspace-server/src/subjektiv.rs

@@ -22,6 +22,7 @@ use crate::feature_storage::{
 
 pub const SUBJEKTIV_SCHEMA_VERSION: u32 = 1;
 pub const SUBJEKTIV_FEATURE_ID: &str = "subjektiv";
+pub const SUBJECT_WORKER_SINGLETON_PREFIX: &str = "subjektiv:";
 pub const MAX_STAGING_ANCHORS: usize = 10;
 /// Deterministic surface-generation policy. Input estimates use the repository's
 /// provider-independent UTF-8 byte estimate (`ceil(bytes / 4)`). The material
@@ -45,6 +46,8 @@ pub enum SubjektivError {
     SubjectNotFound(String),
     #[error("subject `{0}` is retired")]
     SubjectRetired(String),
+    #[error("the selected Worker Profile does not enable subjektiv")]
+    ProfileDisabled,
     #[error("staging candidate `{0}` was not found for this subject")]
     CandidateNotFound(String),
     #[error("staging candidate `{0}` already exists with different content")]
@@ -147,6 +150,23 @@ pub struct SubjectPage {
     pub has_more: bool,
 }
 
+/// Backend-authorized launch material for one active subject. The singleton key
+/// is derived by subjektiv and then acquired through Yoi's generic singleton path.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AuthorizedSubjectWorkerAttachment {
+    pub singleton_key: String,
+}
+
+pub fn subject_worker_singleton_key(subject_id: &str) -> Result<String> {
+    validate_label("subject id", subject_id)?;
+    if subject_id.trim() != subject_id {
+        return Err(SubjektivError::InvalidRecord(
+            "subject id must not contain surrounding whitespace".to_string(),
+        ));
+    }
+    Ok(format!("{SUBJECT_WORKER_SINGLETON_PREFIX}{subject_id}"))
+}
+
 /// Immutable host-recorded history that attributes one durable Session to the
 /// subject and Worker that produced it. This is archival provenance only: it is
 /// deliberately not a current-Worker link or execution ownership record.
@@ -1091,6 +1111,28 @@ impl SubjektivStore {
         })
     }
 
+    /// Validates an explicit product launch connection against this Workspace's
+    /// subject store and the already-resolved Profile policy. Subject knowledge
+    /// alone is insufficient: only an active local record produces host launch material.
+    pub fn authorize_worker_attachment(
+        &self,
+        subject_id: &str,
+        profile: &manifest::WorkerManifest,
+    ) -> Result<AuthorizedSubjectWorkerAttachment> {
+        if !profile.feature.subjektiv.profile.enabled {
+            return Err(SubjektivError::ProfileDisabled);
+        }
+        let subject = self
+            .subject(subject_id)?
+            .ok_or_else(|| SubjektivError::SubjectNotFound(subject_id.to_string()))?;
+        if subject.state != SubjectState::Active {
+            return Err(SubjektivError::SubjectRetired(subject_id.to_string()));
+        }
+        Ok(AuthorizedSubjectWorkerAttachment {
+            singleton_key: subject_worker_singleton_key(&subject.id)?,
+        })
+    }
+
     /// Lists the first deterministic bounded page of subjects. Ordering uses
     /// immutable creation time with the host-issued subject id as a total-order
     /// tie breaker.