From 74981f821726b5cfe02ceacec2981980d7e17b68 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 04:59:05 +0900 Subject: [PATCH] fix: harden retained session observation --- Cargo.lock | 2 + crates/client/src/backend_runtime.rs | 2 + crates/client/src/lib.rs | 4 +- crates/session-store/Cargo.toml | 4 + crates/session-store/src/lib.rs | 30 ++++ crates/session-store/src/public_snapshot.rs | 63 ++++--- crates/session-store/src/worker_metadata.rs | 16 ++ .../session-store/src/worker_session_store.rs | 29 +++- crates/tui/src/backend_worker_picker.rs | 113 ++++++++++--- crates/tui/src/console/mod.rs | 5 +- crates/worker-runtime/src/worker_backend.rs | 3 + .../lib/workspace/session-observation.test.ts | 70 ++++++++ .../src/lib/workspace/session-observation.ts | 46 +++++ .../workers/[workerRef]/console/+page.svelte | 160 +++++++++++------- 14 files changed, 436 insertions(+), 111 deletions(-) create mode 100644 web/workspace/src/lib/workspace/session-observation.test.ts create mode 100644 web/workspace/src/lib/workspace/session-observation.ts diff --git a/Cargo.lock b/Cargo.lock index 2b3c02833..21139f8db 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4170,8 +4170,10 @@ dependencies = [ "agen", "async-trait", "base64 0.22.1", + "filetime", "fs4", "futures", + "libc", "protocol", "serde", "serde_json", diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index a7039caf3..ff6c0fb27 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -40,6 +40,7 @@ pub struct BackendRuntimeTarget { /// Backend-owned Worker identity used as path authority. pub worker_id: String, pub initial_snapshot: Option, + pub initial_notice: Option, } impl BackendRuntimeTarget { @@ -55,6 +56,7 @@ impl BackendRuntimeTarget { runtime_id: runtime_id.into(), worker_id: worker_id.into(), initial_snapshot: None, + initial_notice: None, } } diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 76ae49e97..ad44a2083 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -40,7 +40,9 @@ pub use backend_workspace::{ list_backend_workspaces_blocking, }; pub use client::{Client, ClientError}; -pub use runtime_api::{WorkerSessionAvailability, WorkerSessionUnavailableReason}; +pub use runtime_api::{ + RetainedSessionIdentity, WorkerSessionAvailability, WorkerSessionUnavailableReason, +}; pub use server_api::{ServerApi, ServerApiClient, WorkspaceWorkerSessionResponse}; pub use target::{ BackendTarget, BackendWorkerLaunch, Dashboard, ResolvedTarget, StandaloneTarget, diff --git a/crates/session-store/Cargo.toml b/crates/session-store/Cargo.toml index 5dd8b6886..cc697cef3 100644 --- a/crates/session-store/Cargo.toml +++ b/crates/session-store/Cargo.toml @@ -20,8 +20,12 @@ unicode-normalization = "0.1.25" unicode-properties = { version = "0.1.4", features = ["general-category"] } unicode-security = "0.1.2" +[target.'cfg(any(target_os = "linux", target_os = "android"))'.dependencies] +libc = "0.2" + [dev-dependencies] async-trait = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } tempfile = { workspace = true } +filetime = "0.2" futures = { workspace = true } diff --git a/crates/session-store/src/lib.rs b/crates/session-store/src/lib.rs index b2ac23c1e..1fb147da2 100644 --- a/crates/session-store/src/lib.rs +++ b/crates/session-store/src/lib.rs @@ -31,6 +31,36 @@ //! })?; //! ``` +use std::io; +use std::path::Path; + +/// Read an existing retained file without updating its access timestamp. +/// Observation fails closed when the platform cannot provide that guarantee. +pub(crate) fn read_without_atime(path: &Path) -> io::Result> { + #[cfg(any(target_os = "linux", target_os = "android"))] + { + use std::fs::OpenOptions; + use std::io::Read; + use std::os::unix::fs::OpenOptionsExt; + + let mut file = OpenOptions::new() + .read(true) + .custom_flags(libc::O_NOATIME) + .open(path)?; + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes)?; + Ok(bytes) + } + #[cfg(not(any(target_os = "linux", target_os = "android")))] + { + let _ = path; + Err(io::Error::new( + io::ErrorKind::Unsupported, + "read-only retained observation requires no-atime file reads", + )) + } +} + pub mod event_trace; pub mod fs_store; pub mod history; diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index 39a54a817..752187064 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -13,8 +13,8 @@ use thiserror::Error; use crate::{ LogEntry, LoggedContentPart, LoggedHistoryEntry, LoggedItem, LoggedRole, - LoggedSessionHistoryOrigin, SessionId, Store, StoreError, SystemItem, WorkerAggregateStore, - WorkerMetadataStore, WorkerSessionStore, WorkerStoreError, + LoggedSessionHistoryOrigin, SessionId, StoreError, SystemItem, WorkerAggregateStore, + WorkerSessionStore, WorkerStoreError, }; /// Projection limit leaves one MiB for the typed API envelope so a public @@ -44,6 +44,8 @@ pub enum RetainedSnapshotReadError { MigrationRequired, #[error("retained session log is corrupt")] CorruptLog, + #[error("retained session storage is unavailable")] + StorageUnavailable, #[error("retained session snapshot exceeds the observation limit")] SnapshotTooLarge, } @@ -58,7 +60,7 @@ pub fn read_retained_session_snapshot( let aggregate = WorkerAggregateStore::open_read_only(aggregate_root, worker_name) .map_err(map_worker_store_error)?; let metadata = aggregate - .read_by_name(worker_name) + .read_read_only() .map_err(map_worker_store_error)? .ok_or(RetainedSnapshotReadError::RetentionMissing)?; let active = metadata @@ -73,7 +75,7 @@ pub fn read_retained_session_snapshot( return Err(RetainedSnapshotReadError::SnapshotTooLarge); } let entries = store - .read_all(active.session_id, segment_id) + .read_all_read_only(active.session_id, segment_id) .map_err(map_store_error)?; let snapshot = project_session_snapshot(active.session_id, &entries); if serde_json::to_vec(&snapshot) @@ -101,7 +103,7 @@ fn map_worker_store_error(error: WorkerStoreError) -> RetainedSnapshotReadError WorkerStoreError::Serde(_) | WorkerStoreError::InvalidWorkerName(_) => { RetainedSnapshotReadError::CorruptLog } - WorkerStoreError::Io(_) => RetainedSnapshotReadError::CorruptLog, + WorkerStoreError::Io(_) => RetainedSnapshotReadError::StorageUnavailable, } } @@ -113,9 +115,8 @@ fn map_store_error(error: StoreError) -> RetainedSnapshotReadError { StoreError::Corrupt { message, .. } if message.contains("requires migration") => { RetainedSnapshotReadError::MigrationRequired } - StoreError::Io(_) | StoreError::Serde(_) | StoreError::Corrupt { .. } => { - RetainedSnapshotReadError::CorruptLog - } + StoreError::Io(_) => RetainedSnapshotReadError::StorageUnavailable, + StoreError::Serde(_) | StoreError::Corrupt { .. } => RetainedSnapshotReadError::CorruptLog, _ => RetainedSnapshotReadError::CorruptLog, } } @@ -426,7 +427,7 @@ mod tests { use super::*; use crate::{ LoggedHistoryDerivation, LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, - LoggedWorkerSubject, + LoggedWorkerSubject, Store, WorkerMetadataStore, }; #[test] @@ -685,8 +686,20 @@ mod tests { .unwrap(); let session = WorkerSessionStore::new(aggregate_root.join("session")).unwrap(); session.create_segment(session_id, segment_id, &[]).unwrap(); - let metadata_before = std::fs::read(aggregate_root.join("metadata.json")).unwrap(); - let manifest_before = std::fs::read(aggregate_root.join("session/session.json")).unwrap(); + let metadata_path = aggregate_root.join("metadata.json"); + let manifest_path = aggregate_root.join("session/session.json"); + let segment_path = aggregate_root + .join("session/segments") + .join(format!("{segment_id}.jsonl")); + let metadata_before = std::fs::read(&metadata_path).unwrap(); + let manifest_before = std::fs::read(&manifest_path).unwrap(); + let old_atime = filetime::FileTime::from_unix_time(946_684_800, 0); + for path in [&metadata_path, &manifest_path, &segment_path] { + filetime::set_file_atime(path, old_atime).unwrap(); + } + let mtimes_before = [&metadata_path, &manifest_path, &segment_path].map(|path| { + filetime::FileTime::from_last_modification_time(&std::fs::metadata(path).unwrap()) + }); let retained = read_retained_session_snapshot( &aggregate_root, @@ -699,14 +712,26 @@ mod tests { assert_eq!(retained.identity.segment_id, segment_id.to_string()); assert_eq!(retained.identity.entry_count, 0); assert!(retained.snapshot.entries.is_empty()); - assert_eq!( - std::fs::read(aggregate_root.join("metadata.json")).unwrap(), - metadata_before - ); - assert_eq!( - std::fs::read(aggregate_root.join("session/session.json")).unwrap(), - manifest_before - ); + for (index, path) in [&metadata_path, &manifest_path, &segment_path] + .into_iter() + .enumerate() + { + let metadata = std::fs::metadata(path).unwrap(); + assert_eq!( + filetime::FileTime::from_last_access_time(&metadata), + old_atime, + "retained observation changed access time for {}", + path.display() + ); + assert_eq!( + filetime::FileTime::from_last_modification_time(&metadata), + mtimes_before[index], + "retained observation changed modification time for {}", + path.display() + ); + } + assert_eq!(std::fs::read(&metadata_path).unwrap(), metadata_before); + assert_eq!(std::fs::read(&manifest_path).unwrap(), manifest_before); assert!(matches!( read_retained_session_snapshot(&aggregate_root, "worker-a", 0), Err(RetainedSnapshotReadError::SnapshotTooLarge) diff --git a/crates/session-store/src/worker_metadata.rs b/crates/session-store/src/worker_metadata.rs index 209cfd9d8..f04a1245a 100644 --- a/crates/session-store/src/worker_metadata.rs +++ b/crates/session-store/src/worker_metadata.rs @@ -404,6 +404,22 @@ impl WorkerAggregateStore { }) } + pub fn read_read_only(&self) -> Result, WorkerStoreError> { + let content = match crate::read_without_atime(&self.metadata_path()) { + Ok(content) => content, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + let metadata: WorkerMetadata = serde_json::from_slice(&content)?; + if metadata.worker_name != self.worker_name { + return Err(WorkerStoreError::InvalidWorkerName(format!( + "aggregate identity mismatch: expected `{}`, found `{}`", + self.worker_name, metadata.worker_name + ))); + } + Ok(Some(metadata)) + } + fn validate_name(&self, worker_name: &str) -> Result<(), WorkerStoreError> { validate_worker_name(worker_name)?; if worker_name == self.worker_name { diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index c5c8e9acb..a30aedde2 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -88,7 +88,7 @@ impl WorkerSessionStore { /// persisted data. Observation paths must never mutate retained state. pub fn open_read_only(root: impl Into) -> Result { let root = root.into(); - let bytes = fs::read(root.join(SESSION_FILE))?; + let bytes = crate::read_without_atime(&root.join(SESSION_FILE))?; let manifest: SessionManifest = serde_json::from_slice(&bytes)?; if manifest.schema_version != SESSION_SCHEMA_VERSION { return Err(StoreError::Corrupt { @@ -106,6 +106,33 @@ impl WorkerSessionStore { }) } + pub fn read_all_read_only( + &self, + session_id: SessionId, + segment_id: SegmentId, + ) -> Result, StoreError> { + let retained_session_id = self.session_id.lock().map_err(|_| StoreError::Corrupt { + line: 0, + message: "Worker Session identity lock poisoned".to_string(), + })?; + if *retained_session_id != Some(session_id) { + return Err(StoreError::Corrupt { + line: 0, + message: "active Worker Session identity does not match retained manifest" + .to_string(), + }); + } + let path = self.log_path(segment_id); + let bytes = crate::read_without_atime(&path).map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + StoreError::NotFound(segment_id) + } else { + StoreError::Io(error) + } + })?; + parse_jsonl(&bytes) + } + pub fn segment_log_len(&self, segment_id: SegmentId) -> Result { Ok(fs::metadata(self.log_path(segment_id))?.len()) } diff --git a/crates/tui/src/backend_worker_picker.rs b/crates/tui/src/backend_worker_picker.rs index f0731dbc7..1717184b0 100644 --- a/crates/tui/src/backend_worker_picker.rs +++ b/crates/tui/src/backend_worker_picker.rs @@ -83,32 +83,24 @@ pub(crate) async fn run( let mut attach_target = target .runtime_target(selected.runtime_id.clone(), selected.worker_id.clone()) .map_err(|error| io::Error::other(error.to_string()))?; - if selected.state == "stopped" { - let api = ServerApiClient::builder(&target.base_url) - .map_err(|error| io::Error::other(error.to_string()))? - .build() - .map_err(|error| io::Error::other(error.to_string()))?; - let response = api - .worker_session( - attach_target.workspace_id.clone(), - selected.runtime_id.clone(), - selected.worker_id.clone(), - ) - .await - .map_err(|error| { - io::Error::other(format!( - "failed to observe retained Worker Session {}/{}: {error}", - selected.runtime_id, selected.worker_id - )) - })?; - attach_target.initial_snapshot = match response.observation { - WorkerSessionAvailability::RetainedSnapshot { snapshot, .. } => Some(snapshot), - WorkerSessionAvailability::LiveProtocol => None, - WorkerSessionAvailability::Unavailable { message, .. } => { - return Err(io::Error::other(message).into()); - } - }; - } + let api = ServerApiClient::builder(&target.base_url) + .map_err(|error| io::Error::other(error.to_string()))? + .build() + .map_err(|error| io::Error::other(error.to_string()))?; + let response = api + .worker_session( + attach_target.workspace_id.clone(), + selected.runtime_id.clone(), + selected.worker_id.clone(), + ) + .await + .map_err(|error| { + io::Error::other(format!( + "failed to observe Worker Session {}/{}: {error}", + selected.runtime_id, selected.worker_id + )) + })?; + apply_worker_session_observation(&mut attach_target, response.observation); return console::run_backend_runtime(attach_target).await; } } @@ -123,6 +115,28 @@ enum WorkerPickerResult { SwitchWorkspace, } +fn apply_worker_session_observation( + target: &mut client::BackendRuntimeTarget, + observation: WorkerSessionAvailability, +) { + match observation { + WorkerSessionAvailability::LiveProtocol => {} + WorkerSessionAvailability::RetainedSnapshot { snapshot, .. } => { + target.initial_snapshot = Some(snapshot); + target.initial_notice = Some("read-only retained Session snapshot".to_string()); + } + 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}")); + } + } +} + fn pick_worker( target: BackendRuntimeListTarget, mut workers: Vec, @@ -469,6 +483,53 @@ mod tests { text_width(&text[..byte_offset]) } + #[test] + fn authoritative_observation_selects_live_retained_and_unavailable_detail_modes() { + 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); + assert!(live.initial_snapshot.is_none()); + + let snapshot = protocol::SessionSnapshot { + pending_submissions: protocol::PendingSubmissionsSnapshot::default(), + entries: Vec::new(), + }; + let mut retained = live.clone(); + apply_worker_session_observation( + &mut retained, + WorkerSessionAvailability::RetainedSnapshot { + identity: client::RetainedSessionIdentity { + session_id: "session-a".to_string(), + segment_id: "segment-a".to_string(), + entry_count: 0, + }, + snapshot: snapshot.clone(), + }, + ); + assert_eq!(retained.initial_snapshot, Some(snapshot)); + assert!(retained.initial_notice.unwrap().contains("read-only")); + + let mut unavailable = live; + 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") + ); + } + #[test] fn worker_row_orders_and_simplifies_columns() { let mut worker = worker("runtime-a", "worker-b", Some("builtin:coder")); diff --git a/crates/tui/src/console/mod.rs b/crates/tui/src/console/mod.rs index c831bfa4d..a4e47896e 100644 --- a/crates/tui/src/console/mod.rs +++ b/crates/tui/src/console/mod.rs @@ -471,7 +471,10 @@ async fn run_backend_console( client: Client, connected: bool, ) -> Result<(), Box> { - let worker_label = target.display_label(); + let worker_label = match target.initial_notice.as_deref() { + Some(notice) => format!("{} — {notice}", target.display_label()), + None => target.display_label(), + }; let mut terminal = enter_fullscreen()?; let workspace_root = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")); let mut app = App::new_with_persistent_input_history(worker_label, &workspace_root); diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index fc360f395..e84846d8b 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -1975,6 +1975,9 @@ where session_store::RetainedSnapshotReadError::CorruptLog => { runtime_api::WorkerSessionUnavailableReason::CorruptLog } + session_store::RetainedSnapshotReadError::StorageUnavailable => { + runtime_api::WorkerSessionUnavailableReason::StorageUnavailable + } session_store::RetainedSnapshotReadError::SnapshotTooLarge => { runtime_api::WorkerSessionUnavailableReason::SnapshotTooLarge } diff --git a/web/workspace/src/lib/workspace/session-observation.test.ts b/web/workspace/src/lib/workspace/session-observation.test.ts new file mode 100644 index 000000000..c6450ec2d --- /dev/null +++ b/web/workspace/src/lib/workspace/session-observation.test.ts @@ -0,0 +1,70 @@ +import { + isCurrentWorkerSessionRequest, + workerSessionAction, + type WorkerSessionRequestIdentity, +} from "./session-observation"; +import type { SessionSnapshot } from "$lib/generated/protocol"; + +declare const Deno: { + test(name: string, fn: () => Promise | void): void; +}; + +function assert(condition: boolean, message: string): void { + if (!condition) throw new Error(message); +} + +const snapshot: SessionSnapshot = { + pending_submissions: { + revision: 0, + notification_count: 0, + head_id: null, + submissions: [], + }, + entries: [], +}; + +Deno.test("worker session observation selects live, retained, and unavailable behavior", () => { + assert( + workerSessionAction({ availability: "live_protocol" }).kind === + "subscribe_live", + "live observation must subscribe", + ); + const retained = workerSessionAction({ + availability: "retained_snapshot", + snapshot, + }); + assert( + retained.kind === "apply_retained" && retained.snapshot === snapshot, + "retained observation must apply its snapshot", + ); + const unavailable = workerSessionAction({ + availability: "unavailable", + message: "retention expired", + }); + assert( + unavailable.kind === "show_unavailable" && + unavailable.message === "retention expired", + "unavailable observation must stay recoverable and visible", + ); +}); + +Deno.test("late worker session response cannot overwrite a newer selection", () => { + const request: WorkerSessionRequestIdentity = { + token: 7, + workspaceId: "workspace-a", + runtimeId: "runtime-a", + workerId: "worker-a", + }; + assert( + isCurrentWorkerSessionRequest(request, request), + "matching request identity must remain current", + ); + assert( + !isCurrentWorkerSessionRequest(request, { + ...request, + token: 8, + workerId: "worker-b", + }), + "late response for an older token/worker must be fenced", + ); +}); diff --git a/web/workspace/src/lib/workspace/session-observation.ts b/web/workspace/src/lib/workspace/session-observation.ts new file mode 100644 index 000000000..4b5309a10 --- /dev/null +++ b/web/workspace/src/lib/workspace/session-observation.ts @@ -0,0 +1,46 @@ +import type { SessionSnapshot } from "$lib/generated/protocol"; + +export type WorkerSessionObservation = + | { availability: "live_protocol" } + | { availability: "retained_snapshot"; snapshot: SessionSnapshot } + | { availability: "unavailable"; message: string }; + +export type WorkerSessionAction = + | { kind: "subscribe_live" } + | { kind: "apply_retained"; snapshot: SessionSnapshot } + | { kind: "show_unavailable"; message: string }; + +export interface WorkerSessionRequestIdentity { + token: number; + workspaceId: string; + runtimeId: string; + workerId: string; +} + +export function workerSessionAction( + observation: WorkerSessionObservation, +): WorkerSessionAction { + switch (observation.availability) { + case "live_protocol": + return { kind: "subscribe_live" }; + case "retained_snapshot": + return { kind: "apply_retained", snapshot: observation.snapshot }; + case "unavailable": + return { + kind: "show_unavailable", + message: observation.message.slice(0, 512), + }; + } +} + +export function isCurrentWorkerSessionRequest( + request: WorkerSessionRequestIdentity, + current: WorkerSessionRequestIdentity, +): boolean { + return ( + request.token === current.token && + request.workspaceId === current.workspaceId && + request.runtimeId === current.runtimeId && + request.workerId === current.workerId + ); +} 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 b85189a65..d8dd3d5b7 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 @@ -42,7 +42,6 @@ PendingSubmissionsSnapshot, RewindTarget, Segment, - SessionSnapshot, } from "$lib/generated/protocol"; import { MAX_FILES_PER_SUBMISSION, @@ -53,6 +52,12 @@ import { pushWorkspaceAlert } from "$lib/workspace/alerts/store"; import { workspaceApiPath } from "$lib/workspace/api/http"; import { workspaceMultiplexer, type WorkspaceMultiplexerSubscription } from "$lib/workspace/multiplexer"; + import { + isCurrentWorkerSessionRequest, + workerSessionAction, + type WorkerSessionObservation, + type WorkerSessionRequestIdentity, + } from "$lib/workspace/session-observation"; import type { Diagnostic, Worker } from "$lib/workspace/sidebar/types"; type Props = { @@ -929,79 +934,108 @@ } function connectProtocolTransport( - targetWorker: Worker | null, + _targetWorker: Worker | null, token: number, target: ConsoleTarget, ) { - if (!targetWorker || targetWorker.state === "stopped") { - protocolState = "connecting"; - const controller = new AbortController(); - void fetch( - workerApiPath( - `/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent(target.workerId)}/session`, - ), - { signal: controller.signal, headers: { accept: "application/json" } }, - ) - .then(async (response) => { - if (!response.ok) { - throw new Error(`failed to load retained Worker Session (${response.status})`); - } - return (await response.json()) as - | { availability: "live_protocol" } - | { availability: "retained_snapshot"; snapshot: SessionSnapshot } - | { availability: "unavailable"; message: string }; - }) - .then((observation) => { - if (token !== reloadToken || controller.signal.aborted) return; - if (observation.availability === "unavailable") { - throw new Error(observation.message); - } - if (observation.availability === "live_protocol") { - throw new Error("Worker became live; reload to connect to its protocol stream."); - } - handleIncomingProtocolEvent({ - event: "snapshot", - data: { - session: observation.snapshot, - greeting: { - worker_name: "retained worker", - cwd: "", - provider: "retained session", - model: "", - scope_summary: "read-only retained session", - tools: [], - context_window: 0, - context_tokens: 0, - }, - state: { last_command_id: 0, state: { kind: "idle" } }, - in_flight: { responses: [], commands: [] }, - internal_workers: [], - }, - } as ProtocolEvent); + protocolState = "connecting"; + const controller = new AbortController(); + let closeLiveSubscription: (() => void) | undefined; + const requestIdentity: WorkerSessionRequestIdentity = { + token, + workspaceId: target.workspaceId, + runtimeId: target.runtimeId, + workerId: target.workerId, + }; + void fetch( + workerApiPath( + `/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent(target.workerId)}/session`, + ), + { signal: controller.signal, headers: { accept: "application/json" } }, + ) + .then(async (response) => { + if (!response.ok) { + throw new Error(`failed to observe Worker Session (${response.status})`); + } + return (await response.json()) as WorkerSessionObservation; + }) + .then((observation) => { + const currentIdentity: WorkerSessionRequestIdentity = { + token: reloadToken, + workspaceId, + runtimeId, + workerId, + }; + if ( + !isCurrentWorkerSessionRequest(requestIdentity, currentIdentity) || + controller.signal.aborted + ) + return; + const action = workerSessionAction(observation); + if (action.kind === "show_unavailable") { protocolState = "closed"; - streamDiagnostics = [ - ...streamDiagnostics, - { - code: "retained_session_read_only", - severity: "info", - message: "Showing a read-only retained Session snapshot.", - }, - ]; - }) - .catch((error) => { - if (token !== reloadToken || controller.signal.aborted) return; - protocolState = "error"; streamDiagnostics = [ ...streamDiagnostics, { code: "retained_session_unavailable", severity: "warning", - message: error instanceof Error ? error.message : String(error), + message: action.message, }, ]; - }); - return () => controller.abort(); - } + return; + } + if (action.kind === "subscribe_live") { + closeLiveSubscription = connectLiveProtocolTransport(token, target); + return; + } + handleIncomingProtocolEvent({ + event: "snapshot", + data: { + session: action.snapshot, + greeting: { + worker_name: "retained worker", + cwd: "", + provider: "retained session", + model: "", + scope_summary: "read-only retained session", + tools: [], + context_window: 0, + context_tokens: 0, + }, + state: { last_command_id: 0, state: { kind: "idle" } }, + in_flight: { responses: [], commands: [] }, + internal_workers: [], + }, + } as ProtocolEvent); + protocolState = "closed"; + streamDiagnostics = [ + ...streamDiagnostics, + { + code: "retained_session_read_only", + severity: "info", + message: "Showing a read-only retained Session snapshot.", + }, + ]; + }) + .catch((error) => { + if (token !== reloadToken || controller.signal.aborted) return; + protocolState = "error"; + streamDiagnostics = [ + ...streamDiagnostics, + { + code: "worker_session_observation_failed", + severity: "warning", + message: error instanceof Error ? error.message : String(error), + }, + ]; + }); + return () => { + controller.abort(); + closeLiveSubscription?.(); + }; + } + + function connectLiveProtocolTransport(token: number, target: ConsoleTarget) { protocolState = "connecting"; const subscription = workspaceMultiplexer(target.workspaceId).subscribe( {