chore: merge updated develop into T-656

crates/tui/src/backend_worker_picker.rs

@@ -153,7 +153,7 @@ async fn prepare_selected_worker(
     }
 
     let observation = lifecycle.observe(&target).await?;
-    apply_worker_session_observation(&mut target, observation.observation);
+    apply_worker_session_observation(&mut target, observation.observation)?;
     Ok(target)
 }
 
@@ -217,21 +217,19 @@ enum WorkerPickerResult {
 fn apply_worker_session_observation(
     target: &mut client::BackendRuntimeTarget,
     observation: WorkerSessionAvailability,
-) {
+) -> Result<(), io::Error> {
     match observation {
-        WorkerSessionAvailability::LiveProtocol => {}
+        WorkerSessionAvailability::LiveProtocol => Ok(()),
         WorkerSessionAvailability::RetainedSnapshot { snapshot, .. } => {
             target.initial_snapshot = Some(snapshot);
             target.initial_notice = Some("read-only retained Session snapshot".to_string());
+            Ok(())
         }
-        WorkerSessionAvailability::Unavailable { message, .. } => {
+        WorkerSessionAvailability::Unavailable { reason, message } => {
-            let bounded_message: String = message.chars().take(512).collect();
+            let message = message.chars().take(512).collect::<String>();
-            target.initial_snapshot = Some(protocol::SessionSnapshot {
+            Err(io::Error::other(format!(
-                pending_submissions: protocol::PendingSubmissionsSnapshot::default(),
+                "Backend Worker Session is unavailable: {reason:?}: {message}"
-                entries: Vec::new(),
+            )))
-            });
-            target.initial_notice =
-                Some(format!("retained Session unavailable: {bounded_message}"));
         }
     }
 }
@@ -770,14 +768,15 @@ mod tests {
     }
 
     #[test]
-    fn authoritative_observation_selects_live_retained_and_unavailable_detail_modes() {
+    fn authoritative_observation_opens_only_live_or_retained_sessions() {
         let mut live = client::BackendRuntimeTarget::new(
             "http://127.0.0.1:3000",
             "workspace-a",
             "runtime-a",
             "worker-a",
         );
-        apply_worker_session_observation(&mut live, WorkerSessionAvailability::LiveProtocol);
+        apply_worker_session_observation(&mut live, WorkerSessionAvailability::LiveProtocol)
+            .unwrap();
         assert!(live.initial_snapshot.is_none());
 
         let snapshot = protocol::SessionSnapshot {
@@ -795,25 +794,25 @@ mod tests {
                 },
                 snapshot: snapshot.clone(),
             },
-        );
+        )
+        .unwrap();
         assert_eq!(retained.initial_snapshot, Some(snapshot));
         assert!(retained.initial_notice.unwrap().contains("read-only"));
 
         let mut unavailable = live;
-        apply_worker_session_observation(
+        let error = apply_worker_session_observation(
             &mut unavailable,
             WorkerSessionAvailability::Unavailable {
                 reason: client::WorkerSessionUnavailableReason::RetentionMissing,
                 message: "retention expired".to_string(),
             },
-        );
+        )
-        assert!(unavailable.initial_snapshot.is_some());
+        .unwrap_err()
-        assert!(
+        .to_string();
-            unavailable
+        assert!(unavailable.initial_snapshot.is_none());
-                .initial_notice
+        assert!(unavailable.initial_notice.is_none());
-                .unwrap()
+        assert!(error.contains("RetentionMissing"));
-                .contains("retention expired")
+        assert!(error.contains("retention expired"));
-        );
     }
 
     #[test]

crates/workspace-server/src/server.rs

@@ -124,6 +124,8 @@ use worker_runtime::http_server::{
     RUNTIME_HTTP_PROTOCOL_VERSION,
 };
 #[cfg(test)]
+use worker_runtime::catalog::WorkingDirectoryAttachmentStatus;
+#[cfg(test)]
 use worker_runtime::resource::BackendResourceError;
 use worker_runtime::resource::BackendResourceFetchRequest;
 use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend};
@@ -217,8 +219,8 @@ use worker_runtime::catalog::{
     RepositoryRefObservation, RepositoryRefObservationRequest,
     RepositorySelector as RuntimeRepositorySelector, RepositorySshCredentialCandidate,
     RepositorySshMaterializationAccess, SensitiveString, WorkingDirectoryAttachmentClaim,
-    WorkingDirectoryAttachmentRequest, WorkingDirectoryAttachmentStatus,
+    WorkingDirectoryAttachmentRequest, WorkingDirectoryRepository, WorkingDirectoryRequest,
-    WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
+    WorkspaceApiRef,
 };
 use worker_runtime::config_bundle::ConfigBundle;
 use worker_runtime::http_server::MAX_WORKER_FILE_UPLOAD_BYTES;
@@ -22517,8 +22519,13 @@ async fn get_runtime_worker(
         .store
         .list_workdir_registry(&api.config.workspace_id, 500)?;
     let updated_at = record.updated_at.clone();
-    let worker = merge_worker_registry_projection(Some(&worker), &record, links, &workdirs);
+    let worker = project_worker_registry_projection(
-    let worker = project_workspace_worker(&api, worker)?;
+        &api,
+        Some(&worker),
+        &record,
+        links,
+        &workdirs,
+    )?;
     Ok(Json(server_api::RuntimeWorkerShowResponse {
         worker,
         updated_at,
@@ -22539,8 +22546,13 @@ async fn restore_runtime_worker(
         let workdirs = api
             .store
             .list_workdir_registry(&api.config.workspace_id, 500)?;
-        let summary = merge_worker_registry_projection(Some(worker), &record, links, &workdirs);
+        Some(project_worker_registry_projection(
-        Some(project_workspace_worker(&api, summary)?)
+            &api,
+            Some(worker),
+            &record,
+            links,
+            &workdirs,
+        )?)
     } else {
         None
     };
@@ -23745,19 +23757,6 @@ fn project_workspace_worker(
     api: &WorkspaceApi,
     summary: InternalWorkerSummary,
 ) -> ApiResult<server_api::WorkerSummary> {
-    let resource_key = api
-        .store
-        .resource_key(
-            &api.config.workspace_id,
-            WorkspaceResourceKind::Worker,
-            &summary.worker.worker_id,
-        )?
-        .ok_or_else(|| {
-            Error::Store(format!(
-                "Workspace Worker `{}` has no resource key",
-                summary.worker.worker_id
-            ))
-        })?;
     let workdir_attachments = summary
         .workdir_attachments
         .iter()
@@ -23773,6 +23772,27 @@ fn project_workspace_worker(
             })
         })
         .collect::<ApiResult<Vec<_>>>()?;
+    project_workspace_worker_with_attachments(api, summary, workdir_attachments)
+}
+
+fn project_workspace_worker_with_attachments(
+    api: &WorkspaceApi,
+    summary: InternalWorkerSummary,
+    workdir_attachments: Vec<server_api::WorkerWorkdirAttachmentSummary>,
+) -> ApiResult<server_api::WorkerSummary> {
+    let resource_key = api
+        .store
+        .resource_key(
+            &api.config.workspace_id,
+            WorkspaceResourceKind::Worker,
+            &summary.worker.worker_id,
+        )?
+        .ok_or_else(|| {
+            Error::Store(format!(
+                "Workspace Worker `{}` has no resource key",
+                summary.worker.worker_id
+            ))
+        })?;
     Ok(workspace_worker_summary(
         summary,
         resource_key,
@@ -23794,9 +23814,7 @@ fn project_observed_workspace_workers(
             let links = api
                 .store
                 .list_worker_workdir_links(&api.config.workspace_id, &record.worker)?;
-            let summary =
+            project_worker_registry_projection(api, Some(&worker), &record, links, &workdirs)
-                merge_worker_registry_projection(Some(&worker), &record, links, &workdirs);
-            project_workspace_worker(api, summary)
         })
         .collect()
 }
@@ -23815,13 +23833,13 @@ fn workers_response(api: WorkspaceApi) -> ApiResult<server_api::WorkerListRespon
             .store
             .list_worker_workdir_links(&api.config.workspace_id, &projection.registry.worker)?;
         let observed = worker_summary_from_projection(&projection);
-        let summary = merge_worker_registry_projection(
+        items.push(project_worker_registry_projection(
+            &api,
             observed.as_ref(),
             &projection.registry,
             links,
             &workdir_records,
-        );
+        )?);
-        items.push(project_workspace_worker(&api, summary)?);
     }
     Ok(server_api::WorkerListResponse {
         workspace_id: api.config.workspace_id,
@@ -24691,11 +24709,35 @@ fn worker_summary_from_projection(
     Some(summary)
 }
 
-fn merge_worker_registry_projection(
+fn project_worker_registry_projection(
+    api: &WorkspaceApi,
     live: Option<&InternalWorkerSummary>,
     record: &WorkerRegistryRecord,
     links: Vec<WorkerWorkdirLinkRecord>,
     workdirs: &[WorkdirRegistryRecord],
+) -> ApiResult<server_api::WorkerSummary> {
+    let summary = merge_worker_registry_summary(live, record);
+    let mut workdir_attachments = Vec::with_capacity(links.len());
+    for link in links {
+        let Some(workdir) = workdirs
+            .iter()
+            .find(|workdir| workdir.workdir_id == link.workdir_id)
+        else {
+            continue;
+        };
+        workdir_attachments.push(server_api::WorkerWorkdirAttachmentSummary {
+            alias: workdir::WorkdirAttachmentAlias::new(link.alias)
+                .expect("persisted attachment aliases are validated on write")
+                .to_string(),
+            working_directory: projected_workdir_summary_from_record(api, workdir)?.into(),
+        });
+    }
+    project_workspace_worker_with_attachments(api, summary, workdir_attachments)
+}
+
+fn merge_worker_registry_summary(
+    live: Option<&InternalWorkerSummary>,
+    record: &WorkerRegistryRecord,
 ) -> InternalWorkerSummary {
     let mut summary = live
         .cloned()
@@ -24704,30 +24746,7 @@ fn merge_worker_registry_projection(
     summary.profile = record.profile.clone();
     summary.pinned = record.retention_state == "pinned";
     summary.retention_state = record.retention_state.clone();
-    summary.workdir_attachments = links
+    summary.workdir_attachments.clear();
-        .iter()
-        .filter_map(|link| {
-            workdirs
-                .iter()
-                .find(|workdir| workdir.workdir_id == link.workdir_id)
-                .map(|workdir| {
-                    let mut workdir_summary = runtime_workdir_summary_from_record(workdir);
-                    workdir_summary.occupied_by = Some(WorkingDirectoryOccupancy {
-                        runtime_id: record.worker.runtime_id.clone(),
-                        worker_id: record.worker.worker_id.clone(),
-                        display_name: record.display_name.clone(),
-                        linked_at: link.linked_at.clone(),
-                    });
-                    WorkingDirectoryAttachmentStatus {
-                        alias: workdir::WorkdirAttachmentAlias::new(link.alias.clone())
-                            .expect("persisted attachment aliases are validated on write"),
-                        working_directory: worker_runtime::catalog::WorkingDirectoryStatus {
-                            summary: workdir_summary,
-                        },
-                    }
-                })
-        })
-        .collect();
     summary
 }
 
@@ -25050,6 +25069,7 @@ fn preserve_workdir_identity_for_corrupted_summary(
     }
 }
 
+#[cfg(test)]
 fn runtime_workdir_summary_from_record(
     record: &WorkdirRegistryRecord,
 ) -> worker_runtime::catalog::WorkingDirectorySummary {
@@ -29546,28 +29566,33 @@ mod tests {
         assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
     }
 
-    #[test]
+    #[tokio::test]
-    fn backend_worker_projection_preserves_missing_rows_links_and_redacts_paths() {
+    async fn backend_worker_projection_preserves_missing_rows_links_and_redacts_paths() {
+        let dir = tempfile::tempdir().unwrap();
+        let api = test_api(dir.path()).await;
         let worker = WorkerRegistryRecord {
-            workspace_id: "workspace-1".to_string(),
+            workspace_id: TEST_WORKSPACE_ID.to_string(),
-            worker: RuntimeWorkerRef::new("embedded", "1"),
+            worker: RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, "missing-worker"),
             display_name: "Missing Worker".to_string(),
             profile: Some("builtin:coder".to_string()),
             retention_state: "pinned".to_string(),
-            transcript_ref: Some("runtime://embedded/workers/worker-1/transcript".to_string()),
+            transcript_ref: Some(
+                "runtime://embedded-worker-runtime/workers/missing-worker/transcript".to_string(),
+            ),
             session_ref: None,
             summary_ref: None,
             diagnostics_ref: None,
             created_at: "1".to_string(),
             updated_at: "2".to_string(),
         };
+        api.store.upsert_worker_registry(&worker).unwrap();
         let workdir = WorkdirRegistryRecord {
-            workspace_id: "workspace-1".to_string(),
+            workspace_id: TEST_WORKSPACE_ID.to_string(),
-            workdir_id: "0000019a00000000000".to_string(),
+            workdir_id: "missing-workdir".to_string(),
             display_name: None,
             source: WorkdirRegistrySource::Repository {
-                runtime_id: "embedded".to_string(),
+                runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
-                repository_id: "repo".to_string(),
+                repository_id: test_repository_id(&api),
             },
             creation_selector: Some("develop".to_string()),
             creation_ref: Some("abcdef".to_string()),
@@ -29576,13 +29601,14 @@ mod tests {
             current_ref: Some("fedcba".to_string()),
             current_tree: Some("tree-current".to_string()),
             observed_at_epoch_seconds: Some(3),
-            materialization_status: "missing".to_string(),
+            materialization_status: "not_found".to_string(),
             cleanliness: "clean".to_string(),
             created_at: "1".to_string(),
             updated_at: "3".to_string(),
         };
+        api.store.upsert_workdir_registry(&workdir).unwrap();
         let link = WorkerWorkdirLinkRecord {
-            workspace_id: "workspace-1".to_string(),
+            workspace_id: TEST_WORKSPACE_ID.to_string(),
             worker: worker.worker.clone(),
             workdir_id: workdir.workdir_id.clone(),
             alias: "attachment".to_string(),
@@ -29590,14 +29616,22 @@ mod tests {
             linked_at: "4".to_string(),
             unlinked_at: None,
         };
+        api.store.attach_worker_workdir(&link).unwrap();
 
-        let projected = merge_worker_registry_projection(None, &worker, vec![link], &[workdir]);
+        let projected = project_worker_registry_projection(
+            &api,
+            None,
+            &worker,
+            vec![link],
+            std::slice::from_ref(&workdir),
+        )
+        .unwrap();
 
         assert_eq!(projected.state, "missing");
-        let working_directory = &projected.workdir_attachments[0].working_directory.summary;
+        let working_directory = &projected.workdir_attachments[0].working_directory;
         assert_eq!(
             working_directory.status,
-            WorkingDirectoryStatusKind::NotFound
+            server_api::WorkingDirectoryStatusKind::NotFound
         );
         assert_eq!(
             working_directory.creation_selector.as_deref(),
@@ -29607,11 +29641,14 @@ mod tests {
         assert_eq!(working_directory.current_selector, None);
         assert_eq!(working_directory.current_ref.as_deref(), Some("fedcba"));
         let occupied_by = working_directory.occupied_by.as_ref().unwrap();
-        assert_eq!(occupied_by.runtime_id, "embedded");
+        assert_eq!(occupied_by.runtime_id, EMBEDDED_WORKER_RUNTIME_ID);
-        assert_eq!(occupied_by.worker_id, "1");
+        assert_eq!(occupied_by.worker_id, "missing-worker");
         let occupancy = serde_json::to_value(occupied_by).unwrap();
-        assert_eq!(occupancy["runtime_id"], "embedded");
+        assert_eq!(
-        assert_eq!(occupancy["worker_id"], "1");
+            occupancy["runtime_id"],
+            serde_json::Value::String(EMBEDDED_WORKER_RUNTIME_ID.to_string())
+        );
+        assert_eq!(occupancy["worker_id"], "missing-worker");
         assert!(occupancy.get("runtime_worker_id").is_none());
         assert_eq!(occupied_by.display_name, "Missing Worker");
         assert_eq!(occupied_by.linked_at, "4");
@@ -29620,6 +29657,116 @@ mod tests {
         assert!(!serialized.contains("materialized_path"));
     }
 
+    #[tokio::test]
+    async fn paused_worker_projection_preserves_external_workdir_source() {
+        let dir = tempfile::tempdir().unwrap();
+        let api = test_api(dir.path()).await;
+        let worker = WorkerRegistryRecord {
+            workspace_id: TEST_WORKSPACE_ID.to_string(),
+            worker: RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, "paused-worker"),
+            display_name: "Paused Worker".to_string(),
+            profile: Some("builtin:coder".to_string()),
+            retention_state: "normal".to_string(),
+            transcript_ref: None,
+            session_ref: None,
+            summary_ref: None,
+            diagnostics_ref: None,
+            created_at: "1".to_string(),
+            updated_at: "2".to_string(),
+        };
+        api.store.upsert_worker_registry(&worker).unwrap();
+        let grant = ExternalWorkdirGrantRecord {
+            grant_id: "external-grant".to_string(),
+            workspace_id: TEST_WORKSPACE_ID.to_string(),
+            workdir_id: "external-workdir".to_string(),
+            provider_instance_id: "external-provider".to_string(),
+            display_name: "External Workdir".to_string(),
+            permissions: "read_only".to_string(),
+            created_by: test_browser_request_actor().account_id,
+            created_at: "1".to_string(),
+            expires_at: None,
+            generation: 1,
+            status: "online".to_string(),
+            updated_at: "3".to_string(),
+        };
+        let workdir = WorkdirRegistryRecord {
+            workspace_id: TEST_WORKSPACE_ID.to_string(),
+            workdir_id: "external-workdir".to_string(),
+            display_name: Some("External Workdir".to_string()),
+            source: WorkdirRegistrySource::ExternalGrant {
+                grant_id: "external-grant".to_string(),
+            },
+            creation_selector: None,
+            creation_ref: None,
+            creation_tree: None,
+            current_selector: None,
+            current_ref: None,
+            current_tree: None,
+            observed_at_epoch_seconds: Some(3),
+            materialization_status: "present".to_string(),
+            cleanliness: "clean".to_string(),
+            created_at: "1".to_string(),
+            updated_at: "3".to_string(),
+        };
+        api.store
+            .create_external_workdir_grant(&grant, &workdir)
+            .unwrap();
+        let link = WorkerWorkdirLinkRecord {
+            workspace_id: TEST_WORKSPACE_ID.to_string(),
+            worker: worker.worker.clone(),
+            workdir_id: workdir.workdir_id.clone(),
+            alias: "attachment".to_string(),
+            capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
+            linked_at: "4".to_string(),
+            unlinked_at: None,
+        };
+        api.store.attach_worker_workdir(&link).unwrap();
+        api.worker_projection
+            .seed_observation(
+                EMBEDDED_WORKER_RUNTIME_ID,
+                &protocol::subscription::SubscriptionWorker {
+                    worker_id: protocol::subscription::SubscriptionWorkerId::new("paused-worker")
+                        .unwrap(),
+                    runtime_id: Some(EMBEDDED_WORKER_RUNTIME_ID.to_string()),
+                    resource_key: None,
+                    availability:
+                        protocol::subscription::SubscriptionWorkerAvailability::Observed,
+                    subject_revision: 1,
+                    worker_state: None,
+                    state: protocol::subscription::SubscriptionWorkerState::Paused,
+                    has_running_internal_workers: false,
+                    workspace_id: Some(TEST_WORKSPACE_ID.to_string()),
+                    display_name: Some("Paused Worker".to_string()),
+                    profile: Some("builtin:coder".to_string()),
+                    workdir_attachments: Vec::new(),
+                },
+                "4",
+            )
+            .unwrap();
+
+        let workers = workers_response(api.clone()).unwrap();
+        let projected = workers
+            .items
+            .iter()
+            .find(|worker| worker.worker_id == "paused-worker")
+            .unwrap();
+
+        assert_eq!(projected.state, "paused");
+        assert_eq!(projected.workdir_attachments.len(), 1);
+        let attachment = &projected.workdir_attachments[0];
+        assert_eq!(attachment.alias, "attachment");
+        assert!(matches!(
+            &attachment.working_directory.source,
+            server_api::WorkingDirectorySource::ExternalGrant { grant_id }
+                if grant_id == "external-grant"
+        ));
+        assert_eq!(
+            attachment.working_directory.materializer_kind,
+            server_api::WorkingDirectoryMaterializerKind::ClientHostedExternal
+        );
+        assert!(attachment.working_directory.cleanup_target.is_none());
+    }
+
     #[tokio::test]
     async fn flow_source_resolution_returns_immutable_workspace_and_builtin_snapshots() {
         let workspace = tempfile::tempdir().unwrap();
@@ -43306,10 +43453,19 @@ mod tests {
                     subscription_id,
                     selector:
                         protocol::subscription::EventSubscriptionSelector::WorkerProtocol { .. },
+                    snapshot:
+                        protocol::subscription::SubscriptionSnapshot::WorkerProtocol {
+                            events,
+                            ..
+                        },
                     ..
                 },
             ) = frame.payload
             {
+                assert!(
+                    matches!(events.as_slice(), [protocol::Event::Snapshot { .. }]),
+                    "Worker protocol subscription must atomically include its initial snapshot: {events:?}"
+                );
                 break subscription_id;
             }
         };

crates/workspace-server/src/workspace_subscription.rs

@@ -16,6 +16,7 @@ use crate::server::{
 };
 
 const OUTBOUND_CAPACITY: usize = 256;
+const WORKER_PROTOCOL_SNAPSHOT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
 
 struct ActiveSubscription {
     task: tokio::task::JoinHandle<()>,
@@ -225,6 +226,41 @@ async fn run_worker_protocol(
     control_outbound: mpsc::Sender<WsMessage>,
     protocol_outbound: mpsc::Sender<WsMessage>,
 ) {
+    let initial_event_result =
+        tokio::time::timeout(WORKER_PROTOCOL_SNAPSHOT_TIMEOUT, events.recv()).await;
+    let initial_event = match initial_event_result {
+        Ok(Some(event @ protocol::Event::Snapshot { .. })) => event,
+        Ok(Some(_)) => {
+            let _ = send_rejected(
+                &control_outbound,
+                request_id,
+                SubscriptionRejectionCode::Internal,
+                "Worker protocol did not begin with a session snapshot".to_string(),
+            )
+            .await;
+            return;
+        }
+        Ok(None) => {
+            let _ = send_rejected(
+                &control_outbound,
+                request_id,
+                SubscriptionRejectionCode::ResourceNotFound,
+                "Worker protocol closed before the session snapshot".to_string(),
+            )
+            .await;
+            return;
+        }
+        Err(_) => {
+            let _ = send_rejected(
+                &control_outbound,
+                request_id,
+                SubscriptionRejectionCode::Internal,
+                "Timed out waiting for the Worker protocol session snapshot".to_string(),
+            )
+            .await;
+            return;
+        }
+    };
     if send_frame(
         &control_outbound,
         SubscriptionFrame::new(SubscriptionFramePayload::Response(
@@ -238,7 +274,7 @@ async fn run_worker_protocol(
                 snapshot_revision: 0,
                 snapshot: SubscriptionSnapshot::WorkerProtocol {
                     worker_id: worker_id.clone(),
-                    events: Vec::new(),
+                    events: vec![initial_event],
                 },
             },
         )),
@@ -540,3 +576,77 @@ async fn send_frame(
         .await
         .map_err(|_| ())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn request_id() -> protocol::subscription::SubscriptionRequestId {
+        protocol::subscription::SubscriptionRequestId::new("request-1").unwrap()
+    }
+
+    fn subscription_id() -> SubscriptionId {
+        SubscriptionId::new("subscription-1").unwrap()
+    }
+
+    fn worker_id() -> protocol::subscription::SubscriptionWorkerId {
+        protocol::subscription::SubscriptionWorkerId::new("worker-1").unwrap()
+    }
+
+    async fn run_with_events(events: mpsc::Receiver<protocol::Event>) -> SubscriptionResponse {
+        let (control_outbound, mut control_receiver) = mpsc::channel(1);
+        let (protocol_outbound, _protocol_receiver) = mpsc::channel(1);
+        run_worker_protocol(
+            request_id(),
+            subscription_id(),
+            "runtime-1".to_string(),
+            worker_id(),
+            events,
+            control_outbound,
+            protocol_outbound,
+        )
+        .await;
+        let WsMessage::Text(text) = control_receiver.recv().await.unwrap() else {
+            panic!("expected text response");
+        };
+        let frame: SubscriptionFrame = serde_json::from_str(text.as_str()).unwrap();
+        let SubscriptionFramePayload::Response(response) = frame.payload else {
+            panic!("expected subscription response");
+        };
+        response
+    }
+
+    #[tokio::test]
+    async fn worker_protocol_rejects_a_non_snapshot_initial_event() {
+        let (event_sender, events) = mpsc::channel(1);
+        event_sender
+            .send(protocol::Event::Error {
+                code: protocol::ErrorCode::Internal,
+                message: "not a snapshot".to_string(),
+            })
+            .await
+            .unwrap();
+
+        assert!(matches!(
+            run_with_events(events).await,
+            SubscriptionResponse::SubscriptionRejected {
+                code: SubscriptionRejectionCode::Internal,
+                ..
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn worker_protocol_rejects_a_stream_closed_before_snapshot() {
+        let (event_sender, events) = mpsc::channel(1);
+        drop(event_sender);
+
+        assert!(matches!(
+            run_with_events(events).await,
+            SubscriptionResponse::SubscriptionRejected {
+                code: SubscriptionRejectionCode::ResourceNotFound,
+                ..
+            }
+        ));
+    }
+}

web/workspace/src/lib/workspace/api/ticket-browser.test.ts

@@ -117,6 +117,7 @@ Deno.test("Ticket Browser parser accepts target collections and rejects singular
     item_revision: "revision-1",
     linked_objectives: [],
     merge_request: null,
+    merge_requests: [],
     priority: "P2",
     queued_at: null,
     queued_by: null,
@@ -218,6 +219,10 @@ Deno.test("Ticket Browser parser validates the canonical Merge Request thread",
       event_id: "event-1",
       sequence: 1,
       subject_ref: "abc123",
+      ticket_item_revision: "ticket-revision",
+      ticket_merge_request_subjects: [
+        { merge_request_id: "mr-1", subject_ref: "abc123" },
+      ],
       requested_by: { runtime_id: "runtime-a", worker_id: "worker-a" },
       reviewer: { runtime_id: "runtime-a", worker_id: "reviewer-a" },
       created_at: "2026-09-22T00:00:00Z",

web/workspace/src/lib/workspace/console/worker-console.ui.test.ts

@@ -972,9 +972,14 @@ Deno.test("Worker Console route resolves logical Worker authority before Runtime
       routeLoad.includes("resourceKey(params.workerRef)") &&
       routeLoad.includes("/workers/${encodeURIComponent(reference)}") &&
       routeLoad.includes("canonicalResourceReference") &&
-      routeLoad.includes("runtimeId: result.data?.runtime_id") &&
+      routeLoad.includes("runtimeId: result.data?.runtime_id ?? null") &&
-      routeLoad.includes("workerId: result.data?.worker_id"),
+      routeLoad.includes("workerId: result.data?.worker_id ?? null") &&
-    "route load should resolve the logical Worker reference before exposing its execution target",
+      !routeLoad.includes('result.data?.runtime_id ?? ""') &&
+      consolePage.includes(
+        "resolveWorkerSessionTarget(workspaceId, runtimeId, workerId)",
+      ) &&
+      consolePage.includes("if (!target) return;"),
+    "route load should retain an unresolved Worker target as null and the Console must fail closed before Runtime APIs",
   );
   assert(
     consolePage.includes("workspaceApiPath(workspaceId, path)") &&
@@ -1228,9 +1233,9 @@ Deno.test("Workspace Worker list and Console share the multiplexed connection",
       consolePage.includes("const targetWorker = data.worker") &&
       consolePage.includes("worker = targetWorker") &&
       consolePage.includes(
-        "const consoleTarget = $derived({ workspaceId, runtimeId, workerId })",
+        "resolveWorkerSessionTarget(workspaceId, runtimeId, workerId)",
       ),
-    "A reused Console route should subscribe immediately on the live Workspace socket and install the new route Worker",
+    "A reused Console route should subscribe immediately only after resolving the new route Worker execution target",
   );
 });
 

web/workspace/src/lib/workspace/session-observation.test.ts

@@ -1,5 +1,6 @@
 import {
   isCurrentWorkerSessionRequest,
+  resolveWorkerSessionTarget,
   workerSessionAction,
   type WorkerSessionRequestIdentity,
   workerSessionRequestInit,
@@ -24,6 +25,30 @@ const snapshot: SessionSnapshot = {
   entries: [],
 };
 
+Deno.test("worker session target requires every routing identity", () => {
+  const target = resolveWorkerSessionTarget(
+    "workspace-a",
+    "runtime-a",
+    "worker-a",
+  );
+  assert(
+    target?.runtimeId === "runtime-a" && target.workerId === "worker-a",
+    "resolved identities must produce an API target",
+  );
+  for (
+    const unresolved of [
+      resolveWorkerSessionTarget("", "runtime-a", "worker-a"),
+      resolveWorkerSessionTarget("workspace-a", "", "worker-a"),
+      resolveWorkerSessionTarget("workspace-a", "runtime-a", ""),
+      resolveWorkerSessionTarget("workspace-a", null, "worker-a"),
+      resolveWorkerSessionTarget("workspace-a", "runtime-a", undefined),
+      resolveWorkerSessionTarget("workspace-a", "  ", "worker-a"),
+    ]
+  ) {
+    assert(unresolved === null, "incomplete identities must fail closed");
+  }
+});
+
 Deno.test("worker session request includes same-origin credentials", () => {
   const controller = new AbortController();
   const init = workerSessionRequestInit(controller.signal);

web/workspace/src/lib/workspace/session-observation.ts

@@ -10,13 +10,33 @@ export type WorkerSessionAction =
   | { kind: "apply_retained"; snapshot: SessionSnapshot }
   | { kind: "show_unavailable"; message: string };
 
-export interface WorkerSessionRequestIdentity {
+export interface WorkerSessionTarget {
-  token: number;
   workspaceId: string;
   runtimeId: string;
   workerId: string;
 }
 
+export interface WorkerSessionRequestIdentity extends WorkerSessionTarget {
+  token: number;
+}
+
+export function resolveWorkerSessionTarget(
+  workspaceId: string,
+  runtimeId: string | null | undefined,
+  workerId: string | null | undefined,
+): WorkerSessionTarget | null {
+  if (
+    workspaceId.trim().length === 0 ||
+    runtimeId?.trim().length === 0 ||
+    workerId?.trim().length === 0 ||
+    runtimeId == null ||
+    workerId == null
+  ) {
+    return null;
+  }
+  return { workspaceId, runtimeId, workerId };
+}
+
 export function workerSessionRequestInit(signal: AbortSignal): RequestInit {
   return {
     signal,

web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte

@@ -1,4 +1,5 @@
 <script lang="ts">
+    import { invalidateAll } from "$app/navigation";
     import { tick, untrack, type SvelteComponent } from "svelte";
     import ConsoleLineItem from "$lib/workspace/console/ConsoleLineItem.svelte";
     import ConsoleDisplayStateView from "$lib/workspace/console/ConsoleDisplayState.svelte";
@@ -61,18 +62,20 @@
     import { workspaceMultiplexer, type WorkspaceMultiplexerSubscription } from "$lib/workspace/multiplexer";
     import {
         isCurrentWorkerSessionRequest,
+        resolveWorkerSessionTarget,
         workerSessionAction,
         workerSessionRequestInit,
         type WorkerSessionObservation,
         type WorkerSessionRequestIdentity,
+        type WorkerSessionTarget,
     } from "$lib/workspace/session-observation";
     import type { Diagnostic, Worker } from "$lib/workspace/sidebar/types";
 
     type Props = {
         data: {
             workspaceId: string;
-            runtimeId: string;
+            runtimeId: string | null;
-            workerId: string;
+            workerId: string | null;
             worker: Worker | null;
             workerError: string | null;
         };
@@ -201,8 +204,8 @@
         (SvelteComponent & ComposerInputHandle) | null
     >(null);
     const composerDrafts = new Map<string, ComposerDraftCache>();
-    let activeComposerTargetKey = untrack(
+    let activeComposerTargetKey = untrack(() =>
-        () => `${workspaceId}:${runtimeId}:${workerId}`,
+        runtimeId && workerId ? `${workspaceId}:${runtimeId}:${workerId}` : "",
     );
     let timelineRailDragCleanup: (() => void) | null = null;
     let autoFollowConsole = $state(true);
@@ -231,15 +234,13 @@
     let nextReloadToken = 0;
     let reloadToken = $state(0);
 
-    type ConsoleTarget = {
+    type ConsoleTarget = WorkerSessionTarget;
-        workspaceId: string;
-        runtimeId: string;
-        workerId: string;
-    };
 
-    const consoleTarget = $derived({ workspaceId, runtimeId, workerId });
+    const consoleTarget = $derived(
+        resolveWorkerSessionTarget(workspaceId, runtimeId, workerId),
+    );
     const controlAlertId = $derived(
-        `worker-console-control:${runtimeId}:${workerId}`,
+        `worker-console-control:${runtimeId ?? "unresolved"}:${workerId ?? "unresolved"}`,
     );
 
     const workerViews = $derived(consoleWorkerViews(consoleProjection));
@@ -405,6 +406,10 @@
         const target = consoleTarget;
         beginConsoleLoad();
         const token = advanceReloadToken();
+        if (!target) {
+            void invalidateAll();
+            return;
+        }
         if (!worker) void loadWorker(target, token);
     }
 
@@ -837,7 +842,11 @@
     }
 
     function attachmentPath(): string {
-        return `/api/w/${encodeURIComponent(workspaceId)}/runtimes/${encodeURIComponent(runtimeId)}/workers/${encodeURIComponent(workerId)}`;
+        const target = consoleTarget;
+        if (!target) {
+            throw new Error("Worker execution target is unavailable.");
+        }
+        return `/api/w/${encodeURIComponent(target.workspaceId)}/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent(target.workerId)}`;
     }
 
     function updateAttachment(id: number, update: Partial<ComposerAttachment>): void {
@@ -1059,11 +1068,11 @@
                 return (await response.json()) as WorkerSessionObservation;
             })
             .then((observation) => {
+                const currentTarget = consoleTarget;
+                if (!currentTarget) return;
                 const currentIdentity: WorkerSessionRequestIdentity = {
                     token: reloadToken,
-                    workspaceId,
+                    ...currentTarget,
-                    runtimeId,
-                    workerId,
                 };
                 if (
                     !isCurrentWorkerSessionRequest(requestIdentity, currentIdentity) ||
@@ -1826,7 +1835,6 @@
 
     $effect(() => {
         const target = consoleTarget;
-        switchComposerTarget(target);
         const targetWorker = data.worker;
         const targetWorkerError = data.workerError;
         workerViewSelectionGeneration += 1;
@@ -1839,9 +1847,22 @@
         workerError = targetWorkerError;
         liveWorkerState = targetWorker?.state ?? null;
         streamDiagnostics = [];
-        protocolState = "connecting";
         consoleDisplayState = { kind: "loading", stage: "session" };
         const token = advanceReloadToken();
+        if (!target) {
+            untrack(discardAllAttachments);
+            protocolState = "error";
+            consoleDisplayState = {
+                kind: "unavailable",
+                reason: boundedConsoleReason(
+                    targetWorkerError,
+                    "Worker execution target is unavailable.",
+                ),
+            };
+            return;
+        }
+        switchComposerTarget(target);
+        protocolState = "connecting";
         if (!targetWorker) void loadWorker(target, token);
     });
 
@@ -1853,6 +1874,7 @@
         const target = consoleTarget;
         const targetWorker = worker;
         const token = reloadToken;
+        if (!target) return;
         if (
             targetWorker &&
             (targetWorker.runtime_id !== target.runtimeId ||

web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.ts

@@ -27,15 +27,17 @@ export const load = (async ({ fetch, params }) => {
     if (params.workerRef !== canonical) {
       redirect(
         308,
-        `/w/${encodeURIComponent(params.workspaceId)}/workers/${encodeURIComponent(canonical)}/console`,
+        `/w/${encodeURIComponent(params.workspaceId)}/workers/${
+          encodeURIComponent(canonical)
+        }/console`,
       );
     }
   }
 
   return {
     workspaceId: params.workspaceId,
-    runtimeId: result.data?.runtime_id ?? "",
+    runtimeId: result.data?.runtime_id ?? null,
-    workerId: result.data?.worker_id ?? "",
+    workerId: result.data?.worker_id ?? null,
     worker: result.data,
     workerError: result.error,
   };

web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts

@@ -167,6 +167,29 @@ afterEach(() => {
   vi.restoreAllMocks();
 });
 
+test("does not call Runtime APIs before the Worker execution target resolves", async () => {
+  const request = vi.fn();
+  vi.stubGlobal("fetch", request);
+  render(ConsolePage, {
+    data: {
+      workspaceId: "workspace-a",
+      runtimeId: null,
+      workerId: null,
+      worker: null,
+      workerError: "Worker API request failed with HTTP 404",
+    },
+  });
+
+  const unavailable = await screen.findByRole("alert");
+  expect(unavailable.textContent).toContain("Conversation unavailable");
+  expect(unavailable.textContent).toContain(
+    "Worker API request failed with HTTP 404",
+  );
+  await settleMicrotasks();
+  expect(request).not.toHaveBeenCalled();
+  expect(multiplexer.subscribe).not.toHaveBeenCalled();
+});
+
 test("keeps delayed Session and live initial snapshot work in loading state until projection applies", async () => {
   const sessionResponse = deferred<Response>();
   vi.stubGlobal("fetch", vi.fn(() => sessionResponse.promise));

web/workspace/test/merge-request-status.test.ts

@@ -40,6 +40,10 @@ function reviewRequested(
     event_id: eventId,
     sequence: 1,
     subject_ref: subjectRef,
+    ticket_item_revision: "ticket-revision",
+    ticket_merge_request_subjects: [
+      { merge_request_id: "MR-1", subject_ref: subjectRef },
+    ],
     requested_by: worker,
     reviewer: worker,
     created_at: createdAt,
@@ -57,6 +61,10 @@ function review(
     sequence: 1,
     request_event_id: requestEventId,
     subject_ref: subjectRef,
+    ticket_item_revision: "ticket-revision",
+    ticket_merge_request_subjects: [
+      { merge_request_id: "MR-1", subject_ref: subjectRef },
+    ],
     decision: "approve",
     body: "approved",
     findings: [],