diff --git a/crates/tui/src/backend_worker_picker.rs b/crates/tui/src/backend_worker_picker.rs index df62a759b..4472ea407 100644 --- a/crates/tui/src/backend_worker_picker.rs +++ b/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, .. } => { - let bounded_message: String = message.chars().take(512).collect(); - target.initial_snapshot = Some(protocol::SessionSnapshot { - pending_submissions: protocol::PendingSubmissionsSnapshot::default(), - entries: Vec::new(), - }); - target.initial_notice = - Some(format!("retained Session unavailable: {bounded_message}")); + WorkerSessionAvailability::Unavailable { reason, message } => { + let message = message.chars().take(512).collect::(); + Err(io::Error::other(format!( + "Backend Worker Session is unavailable: {reason:?}: {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()); - assert!( - unavailable - .initial_notice - .unwrap() - .contains("retention expired") - ); + ) + .unwrap_err() + .to_string(); + assert!(unavailable.initial_snapshot.is_none()); + assert!(unavailable.initial_notice.is_none()); + assert!(error.contains("RetentionMissing")); + assert!(error.contains("retention expired")); } #[test] diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index f70330a97..ef2a1498d 100644 --- a/crates/workspace-server/src/server.rs +++ b/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, - WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef, + WorkingDirectoryAttachmentRequest, WorkingDirectoryRepository, WorkingDirectoryRequest, + 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_workspace_worker(&api, worker)?; + let worker = project_worker_registry_projection( + &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_workspace_worker(&api, summary)?) + Some(project_worker_registry_projection( + &api, + Some(worker), + &record, + links, + &workdirs, + )?) } else { None }; @@ -23745,19 +23757,6 @@ fn project_workspace_worker( api: &WorkspaceApi, summary: InternalWorkerSummary, ) -> ApiResult { - 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::>>()?; + project_workspace_worker_with_attachments(api, summary, workdir_attachments) +} + +fn project_workspace_worker_with_attachments( + api: &WorkspaceApi, + summary: InternalWorkerSummary, + workdir_attachments: Vec, +) -> ApiResult { + 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 = - merge_worker_registry_projection(Some(&worker), &record, links, &workdirs); - project_workspace_worker(api, summary) + project_worker_registry_projection(api, Some(&worker), &record, links, &workdirs) }) .collect() } @@ -23815,13 +23833,13 @@ fn workers_response(api: WorkspaceApi) -> ApiResult, record: &WorkerRegistryRecord, links: Vec, workdirs: &[WorkdirRegistryRecord], +) -> ApiResult { + 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 - .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.workdir_attachments.clear(); 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] - fn backend_worker_projection_preserves_missing_rows_links_and_redacts_paths() { + #[tokio::test] + 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(), - worker: RuntimeWorkerRef::new("embedded", "1"), + workspace_id: TEST_WORKSPACE_ID.to_string(), + 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(), - workdir_id: "0000019a00000000000".to_string(), + workspace_id: TEST_WORKSPACE_ID.to_string(), + workdir_id: "missing-workdir".to_string(), display_name: None, source: WorkdirRegistrySource::Repository { - runtime_id: "embedded".to_string(), - repository_id: "repo".to_string(), + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.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.worker_id, "1"); + assert_eq!(occupied_by.runtime_id, EMBEDDED_WORKER_RUNTIME_ID); + 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!(occupancy["worker_id"], "1"); + assert_eq!( + 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; } }; diff --git a/crates/workspace-server/src/workspace_subscription.rs b/crates/workspace-server/src/workspace_subscription.rs index f6c5d158f..5319e4a4d 100644 --- a/crates/workspace-server/src/workspace_subscription.rs +++ b/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, protocol_outbound: mpsc::Sender, ) { + 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) -> 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, + .. + } + )); + } +} diff --git a/web/workspace/src/lib/workspace/api/ticket-browser.test.ts b/web/workspace/src/lib/workspace/api/ticket-browser.test.ts index aa6ef8279..7b6c52d73 100644 --- a/web/workspace/src/lib/workspace/api/ticket-browser.test.ts +++ b/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", diff --git a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts index 1200041a7..245b47400 100644 --- a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts +++ b/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("workerId: result.data?.worker_id"), - "route load should resolve the logical Worker reference before exposing its execution target", + routeLoad.includes("runtimeId: result.data?.runtime_id ?? null") && + routeLoad.includes("workerId: result.data?.worker_id ?? null") && + !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", ); }); diff --git a/web/workspace/src/lib/workspace/session-observation.test.ts b/web/workspace/src/lib/workspace/session-observation.test.ts index 51e31cf02..27a7a2171 100644 --- a/web/workspace/src/lib/workspace/session-observation.test.ts +++ b/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); diff --git a/web/workspace/src/lib/workspace/session-observation.ts b/web/workspace/src/lib/workspace/session-observation.ts index 9c5894507..861b0e5d7 100644 --- a/web/workspace/src/lib/workspace/session-observation.ts +++ b/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 { - token: number; +export interface WorkerSessionTarget { 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, diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte index 2b5abd0ac..af1b0eb5c 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte @@ -1,4 +1,5 @@