fix: harden retained session observation

Cargo.lock

@@ -4170,8 +4170,10 @@ dependencies = [
  "agen",
  "async-trait",
  "base64 0.22.1",
+ "filetime",
  "fs4",
  "futures",
+ "libc",
  "protocol",
  "serde",
  "serde_json",

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<protocol::SessionSnapshot>,
+    pub initial_notice: Option<String>,
 }
 
 impl BackendRuntimeTarget {
@@ -55,6 +56,7 @@ impl BackendRuntimeTarget {
             runtime_id: runtime_id.into(),
             worker_id: worker_id.into(),
             initial_snapshot: None,
+            initial_notice: None,
         }
     }
 

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,

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 }

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<Vec<u8>> {
+    #[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;

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,
+    LoggedSessionHistoryOrigin, SessionId, StoreError, SystemItem, WorkerAggregateStore,
-    WorkerMetadataStore, WorkerSessionStore, WorkerStoreError,
+    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 { .. } => {
+        StoreError::Io(_) => RetainedSnapshotReadError::StorageUnavailable,
-            RetainedSnapshotReadError::CorruptLog
+        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 metadata_path = aggregate_root.join("metadata.json");
-        let manifest_before = std::fs::read(aggregate_root.join("session/session.json")).unwrap();
+        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!(
+        for (index, path) in [&metadata_path, &manifest_path, &segment_path]
-            std::fs::read(aggregate_root.join("metadata.json")).unwrap(),
+            .into_iter()
-            metadata_before
+            .enumerate()
-        );
+        {
-        assert_eq!(
+            let metadata = std::fs::metadata(path).unwrap();
-            std::fs::read(aggregate_root.join("session/session.json")).unwrap(),
+            assert_eq!(
-            manifest_before
+                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)

crates/session-store/src/worker_metadata.rs

@@ -404,6 +404,22 @@ impl WorkerAggregateStore {
         })
     }
 
+    pub fn read_read_only(&self) -> Result<Option<WorkerMetadata>, 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 {

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<PathBuf>) -> Result<Self, StoreError> {
         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<Vec<LogEntry>, 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<u64, StoreError> {
         Ok(fs::metadata(self.log_path(segment_id))?.len())
     }

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)
-            let api = ServerApiClient::builder(&target.base_url)
+            .map_err(|error| io::Error::other(error.to_string()))?
-                .map_err(|error| io::Error::other(error.to_string()))?
+            .build()
-                .build()
+            .map_err(|error| io::Error::other(error.to_string()))?;
-                .map_err(|error| io::Error::other(error.to_string()))?;
+        let response = api
-            let response = api
+            .worker_session(
-                .worker_session(
+                attach_target.workspace_id.clone(),
-                    attach_target.workspace_id.clone(),
+                selected.runtime_id.clone(),
-                    selected.runtime_id.clone(),
+                selected.worker_id.clone(),
-                    selected.worker_id.clone(),
+            )
-                )
+            .await
-                .await
+            .map_err(|error| {
-                .map_err(|error| {
+                io::Error::other(format!(
-                    io::Error::other(format!(
+                    "failed to observe Worker Session {}/{}: {error}",
-                        "failed to observe retained Worker Session {}/{}: {error}",
+                    selected.runtime_id, selected.worker_id
-                        selected.runtime_id, selected.worker_id
+                ))
-                    ))
+            })?;
-                })?;
+        apply_worker_session_observation(&mut attach_target, response.observation);
-            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());
-                }
-            };
-        }
         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<BackendWorkerSummary>,
@@ -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"));

crates/tui/src/console/mod.rs

@@ -471,7 +471,10 @@ async fn run_backend_console<T: Socket>(
     client: Client<T>,
     connected: bool,
 ) -> Result<(), Box<dyn std::error::Error>> {
-    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);

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
                     }

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): 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",
+  );
+});

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
+  );
+}

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";
-            protocolState = "connecting";
+        const controller = new AbortController();
-            const controller = new AbortController();
+        let closeLiveSubscription: (() => void) | undefined;
-            void fetch(
+        const requestIdentity: WorkerSessionRequestIdentity = {
-                workerApiPath(
+            token,
-                    `/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent(target.workerId)}/session`,
+            workspaceId: target.workspaceId,
-                ),
+            runtimeId: target.runtimeId,
-                { signal: controller.signal, headers: { accept: "application/json" } },
+            workerId: target.workerId,
-            )
+        };
-                .then(async (response) => {
+        void fetch(
-                    if (!response.ok) {
+            workerApiPath(
-                        throw new Error(`failed to load retained Worker Session (${response.status})`);
+                `/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent(target.workerId)}/session`,
-                    }
+            ),
-                    return (await response.json()) as
+            { signal: controller.signal, headers: { accept: "application/json" } },
-                        | { availability: "live_protocol" }
+        )
-                        | { availability: "retained_snapshot"; snapshot: SessionSnapshot }
+            .then(async (response) => {
-                        | { availability: "unavailable"; message: string };
+                if (!response.ok) {
-                })
+                    throw new Error(`failed to observe Worker Session (${response.status})`);
-                .then((observation) => {
+                }
-                    if (token !== reloadToken || controller.signal.aborted) return;
+                return (await response.json()) as WorkerSessionObservation;
-                    if (observation.availability === "unavailable") {
+            })
-                        throw new Error(observation.message);
+            .then((observation) => {
-                    }
+                const currentIdentity: WorkerSessionRequestIdentity = {
-                    if (observation.availability === "live_protocol") {
+                    token: reloadToken,
-                        throw new Error("Worker became live; reload to connect to its protocol stream.");
+                    workspaceId,
-                    }
+                    runtimeId,
-                    handleIncomingProtocolEvent({
+                    workerId,
-                        event: "snapshot",
+                };
-                        data: {
+                if (
-                            session: observation.snapshot,
+                    !isCurrentWorkerSessionRequest(requestIdentity, currentIdentity) ||
-                            greeting: {
+                    controller.signal.aborted
-                                worker_name: "retained worker",
+                )
-                                cwd: "",
+                    return;
-                                provider: "retained session",
+                const action = workerSessionAction(observation);
-                                model: "",
+                if (action.kind === "show_unavailable") {
-                                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: "retained_session_unavailable",
                             severity: "warning",
-                            message: error instanceof Error ? error.message : String(error),
+                            message: action.message,
                         },
                     ];
-                });
+                    return;
-            return () => controller.abort();
+                }
-        }
+                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(
             {