From 7a61a26670ed850cffd5ed96020570433a5975de Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 04:27:38 +0900 Subject: [PATCH 1/4] feat: observe retained worker sessions without restore --- Cargo.lock | 16 ++ Cargo.toml | 3 + crates/client/Cargo.toml | 2 + crates/client/src/backend_runtime.rs | 4 +- crates/client/src/lib.rs | 2 + crates/client/src/target.rs | 2 +- crates/runtime-api/Cargo.toml | 1 + crates/runtime-api/src/lib.rs | 67 ++++++- crates/server-api/Cargo.toml | 14 ++ crates/server-api/src/lib.rs | 59 ++++++ crates/session-store/src/lib.rs | 4 + crates/session-store/src/public_snapshot.rs | 171 +++++++++++++++++- crates/session-store/src/worker_metadata.rs | 21 +++ .../session-store/src/worker_session_store.rs | 26 +++ crates/tui/Cargo.toml | 1 + crates/tui/src/backend_worker_picker.rs | 136 +++----------- crates/tui/src/console/mod.rs | 69 ++++++- crates/worker-runtime/src/execution.rs | 22 +++ crates/worker-runtime/src/http_server.rs | 3 +- .../src/http_server/runtime_management_api.rs | 29 +++ crates/worker-runtime/src/runtime.rs | 68 ++++++- crates/worker-runtime/src/worker_backend.rs | 66 ++++++- crates/workspace-server/Cargo.toml | 1 + crates/workspace-server/src/hosts.rs | 62 +++++++ crates/workspace-server/src/server.rs | 39 ++++ .../workers/[workerRef]/console/+page.svelte | 70 ++++++- 26 files changed, 834 insertions(+), 124 deletions(-) create mode 100644 crates/server-api/Cargo.toml create mode 100644 crates/server-api/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 5156ef43c..2b3c02833 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -650,8 +650,10 @@ dependencies = [ "futures", "protocol", "reqwest", + "runtime-api", "serde", "serde_json", + "server-api", "tempfile", "thiserror 2.0.18", "ticket", @@ -3714,6 +3716,7 @@ dependencies = [ "manifest", "protocol", "serde", + "serde_json", "tokio", "workdir", "workspace-api", @@ -4128,6 +4131,18 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "server-api" +version = "0.1.0" +dependencies = [ + "api-macros", + "runtime-api", + "serde", + "serde_json", + "tokio", + "workspace-api", +] + [[package]] name = "session-analytics" version = "0.1.0" @@ -6291,6 +6306,7 @@ dependencies = [ "serde", "serde_json", "serde_yaml", + "server-api", "sha2 0.11.0", "ssh-key", "tempfile", diff --git a/Cargo.toml b/Cargo.toml index 49e82d425..bd739b6be 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ members = [ "crates/worker", "crates/worker-runtime", "crates/runtime-api", + "crates/server-api", "crates/plugin-pdk", "crates/yoi", "crates/protocol", @@ -49,6 +50,7 @@ default-members = [ "crates/worker", "crates/worker-runtime", "crates/runtime-api", + "crates/server-api", "crates/plugin-pdk", "crates/yoi", "crates/protocol", @@ -104,6 +106,7 @@ session-analytics = { path = "crates/session-analytics" } session-store = { path = "crates/session-store" } standalone = { path = "crates/standalone" } runtime-api = { path = "crates/runtime-api" } +server-api = { path = "crates/server-api" } secrets = { path = "crates/secrets" } tools = { path = "crates/tools" } config-source = { path = "crates/config-source" } diff --git a/crates/client/Cargo.toml b/crates/client/Cargo.toml index bc3aa6bfd..1e7411bab 100644 --- a/crates/client/Cargo.toml +++ b/crates/client/Cargo.toml @@ -8,11 +8,13 @@ license.workspace = true async-trait.workspace = true chrono = { version = "0.4", default-features = false, features = ["clock"] } protocol = { workspace = true } +runtime-api.workspace = true ticket = { workspace = true } futures = { workspace = true } reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "native-tls"] } serde = { workspace = true } serde_json = { workspace = true } +server-api.workspace = true thiserror = { workspace = true } tokio = { workspace = true, features = ["rt", "macros", "net", "io-util", "sync", "time"] } tokio-tungstenite = { workspace = true } diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index 6809c6409..a7039caf3 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -28,7 +28,7 @@ pub use workspace_api::{ WorkingDirectorySummary as BackendWorkingDirectorySummary, }; -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq)] pub struct BackendRuntimeTarget { /// Workspace Backend API root URL, for example `http://127.0.0.1:8787`. /// This is intentionally the Backend endpoint, not a Runtime endpoint. @@ -39,6 +39,7 @@ pub struct BackendRuntimeTarget { pub runtime_id: String, /// Backend-owned Worker identity used as path authority. pub worker_id: String, + pub initial_snapshot: Option, } impl BackendRuntimeTarget { @@ -53,6 +54,7 @@ impl BackendRuntimeTarget { workspace_id: workspace_id.into(), runtime_id: runtime_id.into(), worker_id: worker_id.into(), + initial_snapshot: None, } } diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 1bb4b8d74..76ae49e97 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -40,6 +40,8 @@ pub use backend_workspace::{ list_backend_workspaces_blocking, }; pub use client::{Client, ClientError}; +pub use runtime_api::{WorkerSessionAvailability, WorkerSessionUnavailableReason}; +pub use server_api::{ServerApi, ServerApiClient, WorkspaceWorkerSessionResponse}; pub use target::{ BackendTarget, BackendWorkerLaunch, Dashboard, ResolvedTarget, StandaloneTarget, StandaloneWorkerListIntent, StandaloneWorkerResumeIntent, Target, TargetError, TargetKind, diff --git a/crates/client/src/target.rs b/crates/client/src/target.rs index 1d2aea9bc..bc9a861aa 100644 --- a/crates/client/src/target.rs +++ b/crates/client/src/target.rs @@ -134,7 +134,7 @@ pub struct WorkerList { pub include_stopped: bool, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq)] pub struct WorkerConnection { pub target: BackendRuntimeTarget, } diff --git a/crates/runtime-api/Cargo.toml b/crates/runtime-api/Cargo.toml index 404355a17..d2fe97960 100644 --- a/crates/runtime-api/Cargo.toml +++ b/crates/runtime-api/Cargo.toml @@ -14,4 +14,5 @@ workspace-api.workspace = true workdir.workspace = true [dev-dependencies] +serde_json.workspace = true tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread"] } diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index 9bcac8a4d..a02e2488b 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -532,6 +532,44 @@ pub struct WorkerRetentionExecutionResult { pub diagnostics_retained: bool, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerSessionRequest { + pub workspace_id: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RetainedSessionIdentity { + pub session_id: String, + pub segment_id: String, + pub entry_count: u64, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkerSessionUnavailableReason { + RetentionMissing, + ActivePointerMissing, + CorruptLog, + MigrationRequired, + RetentionExpired, + SnapshotTooLarge, + StorageUnavailable, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(tag = "availability", rename_all = "snake_case")] +pub enum WorkerSessionAvailability { + LiveProtocol, + RetainedSnapshot { + identity: RetainedSessionIdentity, + snapshot: protocol::SessionSnapshot, + }, + Unavailable { + reason: WorkerSessionUnavailableReason, + message: String, + }, +} + #[api(reqwest, axum)] pub trait RuntimeApi { #[get("/v1/ping", status = 200, error_status = 400)] @@ -555,6 +593,13 @@ pub trait RuntimeApi { #[path] worker_id: String, ) -> Result; + #[get("/v1/workers/{worker_id}/session", status = 200, error_status = 400)] + async fn worker_session( + &self, + #[path] worker_id: String, + #[query] request: WorkerSessionRequest, + ) -> Result; + #[post("/v1/workers", status = 200, error_status = 400)] async fn create_worker( &self, @@ -827,6 +872,14 @@ mod tests { Err(test_error(404)) } + async fn worker_session( + &self, + _worker_id: String, + _request: WorkerSessionRequest, + ) -> Result { + Ok(WorkerSessionAvailability::LiveProtocol) + } + async fn create_worker( &self, _request: CreateWorkerRequest, @@ -929,10 +982,22 @@ mod tests { server.abort(); } + #[test] + fn worker_session_availability_has_closed_tagged_wire_shape() { + let value = serde_json::to_value(WorkerSessionAvailability::Unavailable { + reason: WorkerSessionUnavailableReason::SnapshotTooLarge, + message: "snapshot exceeds limit".to_string(), + }) + .unwrap(); + assert_eq!(value["availability"], "unavailable"); + assert_eq!(value["reason"], "snapshot_too_large"); + assert_eq!(value["message"], "snapshot exceeds limit"); + } + #[test] fn contract_inventory_is_complete_and_unique() { let operations = RuntimeApiMetadata::OPERATIONS; - assert_eq!(operations.len(), 14); + assert_eq!(operations.len(), 15); let mut routes = operations .iter() .map(|operation| (format!("{:?}", operation.method), operation.path)) diff --git a/crates/server-api/Cargo.toml b/crates/server-api/Cargo.toml new file mode 100644 index 000000000..609be216b --- /dev/null +++ b/crates/server-api/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "server-api" +version = "0.1.0" +edition = "2024" + +[dependencies] +api-macros = { workspace = true, features = ["reqwest", "axum"] } +runtime-api.workspace = true +serde = { workspace = true, features = ["derive"] } +workspace-api.workspace = true + +[dev-dependencies] +serde_json.workspace = true +tokio = { workspace = true } diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs new file mode 100644 index 000000000..88e782fd3 --- /dev/null +++ b/crates/server-api/src/lib.rs @@ -0,0 +1,59 @@ +//! Generated Workspace Server REST contract for public clients. + +use api_macros::api; +pub use api_macros::axum as server_support; +pub use api_macros::reqwest as client_support; +pub use api_macros::{ApiContract, HttpMethod}; +use serde::{Deserialize, Serialize}; +use workspace_api::WorkspaceWorkerSubject; + +pub type ServerApiError = runtime_api::RuntimeApiError; + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct WorkspaceWorkerSessionResponse { + pub subject: WorkspaceWorkerSubject, + #[serde(flatten)] + pub observation: runtime_api::WorkerSessionAvailability, +} + +#[api(reqwest, axum)] +pub trait ServerApi { + #[get( + "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/session", + status = 200, + error_status = 400 + )] + async fn worker_session( + &self, + #[path] workspace_id: String, + #[path] runtime_id: String, + #[path] worker_id: String, + ) -> Result; +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn worker_session_contract_and_flattened_availability_are_stable() { + let operations = ServerApiMetadata::OPERATIONS; + assert_eq!(operations.len(), 1); + assert_eq!(operations[0].method, HttpMethod::Get); + assert_eq!( + operations[0].path, + "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/session" + ); + + let response = WorkspaceWorkerSessionResponse { + subject: WorkspaceWorkerSubject::RuntimeWorker { + runtime_id: "runtime-a".to_string(), + worker_id: "worker-a".to_string(), + }, + observation: runtime_api::WorkerSessionAvailability::LiveProtocol, + }; + let value = serde_json::to_value(response).unwrap(); + assert_eq!(value["availability"], "live_protocol"); + assert_eq!(value["subject"]["kind"], "runtime_worker"); + } +} diff --git a/crates/session-store/src/lib.rs b/crates/session-store/src/lib.rs index bf89e149b..b2ac23c1e 100644 --- a/crates/session-store/src/lib.rs +++ b/crates/session-store/src/lib.rs @@ -57,6 +57,10 @@ pub use history::{ }; pub use logged_item::{LoggedContentPart, LoggedItem, LoggedRole, from_logged, to_logged}; pub use paste_artifact::PasteArtifactLimits; +pub use public_snapshot::{ + DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, RetainedSessionIdentity, RetainedSessionSnapshot, + RetainedSnapshotReadError, project_session_snapshot, read_retained_session_snapshot, +}; pub use segment::{ SegmentStartState, append_entry, append_system_item, classify_logged_history_entry, create_compacted_segment, create_segment, create_segment_with_ids, ensure_head_or_fork, fork, diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index b17935842..39a54a817 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -7,11 +7,119 @@ use protocol::{ SessionSnapshotEntry, SessionSnapshotEntryData, SessionToolAttachment, }; +use std::path::Path; + +use thiserror::Error; + use crate::{ LogEntry, LoggedContentPart, LoggedHistoryEntry, LoggedItem, LoggedRole, - LoggedSessionHistoryOrigin, SessionId, SystemItem, + LoggedSessionHistoryOrigin, SessionId, Store, StoreError, SystemItem, WorkerAggregateStore, + WorkerMetadataStore, WorkerSessionStore, WorkerStoreError, }; +/// Projection limit leaves one MiB for the typed API envelope so a public +/// retained-session response remains below the 16 MiB response ceiling. +pub const DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES: u64 = 15 * 1024 * 1024; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RetainedSessionIdentity { + pub session_id: String, + pub segment_id: String, + pub entry_count: u64, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct RetainedSessionSnapshot { + pub identity: RetainedSessionIdentity, + pub snapshot: SessionSnapshot, +} + +#[derive(Debug, Error)] +pub enum RetainedSnapshotReadError { + #[error("retained worker state is unavailable")] + RetentionMissing, + #[error("retained worker state has no active session pointer")] + ActivePointerMissing, + #[error("retained session requires migration")] + MigrationRequired, + #[error("retained session log is corrupt")] + CorruptLog, + #[error("retained session snapshot exceeds the observation limit")] + SnapshotTooLarge, +} + +/// Read the active retained Session without creating, migrating, restoring, or +/// rewriting any persisted state. +pub fn read_retained_session_snapshot( + aggregate_root: &Path, + worker_name: &str, + max_bytes: u64, +) -> Result { + let aggregate = WorkerAggregateStore::open_read_only(aggregate_root, worker_name) + .map_err(map_worker_store_error)?; + let metadata = aggregate + .read_by_name(worker_name) + .map_err(map_worker_store_error)? + .ok_or(RetainedSnapshotReadError::RetentionMissing)?; + let active = metadata + .active + .ok_or(RetainedSnapshotReadError::ActivePointerMissing)?; + let segment_id = active + .segment_id + .ok_or(RetainedSnapshotReadError::ActivePointerMissing)?; + let store = WorkerSessionStore::open_read_only(aggregate_root.join("session")) + .map_err(map_store_error)?; + if store.segment_log_len(segment_id).map_err(map_store_error)? > max_bytes { + return Err(RetainedSnapshotReadError::SnapshotTooLarge); + } + let entries = store + .read_all(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) + .map_err(|_| RetainedSnapshotReadError::CorruptLog)? + .len() as u64 + > max_bytes + { + return Err(RetainedSnapshotReadError::SnapshotTooLarge); + } + Ok(RetainedSessionSnapshot { + identity: RetainedSessionIdentity { + session_id: active.session_id.to_string(), + segment_id: segment_id.to_string(), + entry_count: entries.len() as u64, + }, + snapshot, + }) +} + +fn map_worker_store_error(error: WorkerStoreError) -> RetainedSnapshotReadError { + match error { + WorkerStoreError::Io(error) if error.kind() == std::io::ErrorKind::NotFound => { + RetainedSnapshotReadError::RetentionMissing + } + WorkerStoreError::Serde(_) | WorkerStoreError::InvalidWorkerName(_) => { + RetainedSnapshotReadError::CorruptLog + } + WorkerStoreError::Io(_) => RetainedSnapshotReadError::CorruptLog, + } +} + +fn map_store_error(error: StoreError) -> RetainedSnapshotReadError { + match error { + StoreError::Io(error) if error.kind() == std::io::ErrorKind::NotFound => { + RetainedSnapshotReadError::RetentionMissing + } + StoreError::Corrupt { message, .. } if message.contains("requires migration") => { + RetainedSnapshotReadError::MigrationRequired + } + StoreError::Io(_) | StoreError::Serde(_) | StoreError::Corrupt { .. } => { + RetainedSnapshotReadError::CorruptLog + } + _ => RetainedSnapshotReadError::CorruptLog, + } +} + /// Project a complete current-segment log. A valid segment starts with one /// canonical annotated SegmentStart record; malformed partial input uses the /// nil session only to keep the public failure projection deterministic. @@ -559,4 +667,65 @@ mod tests { SessionEntryProvenance::ModelOutput ); } + + #[test] + fn retained_snapshot_read_projects_active_segment_without_rewriting_files() { + let root = tempfile::tempdir().unwrap(); + let aggregate_root = root.path().join("worker-a"); + let aggregate = WorkerAggregateStore::new(&aggregate_root, "worker-a").unwrap(); + let session_id = crate::new_session_id(); + let segment_id = crate::new_segment_id(); + aggregate + .write(&crate::WorkerMetadata::new( + "worker-a", + Some(crate::WorkerActiveSegmentRef::active_segment( + session_id, segment_id, + )), + )) + .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 retained = read_retained_session_snapshot( + &aggregate_root, + "worker-a", + DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, + ) + .unwrap(); + + assert_eq!(retained.identity.session_id, session_id.to_string()); + 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 + ); + assert!(matches!( + read_retained_session_snapshot(&aggregate_root, "worker-a", 0), + Err(RetainedSnapshotReadError::SnapshotTooLarge) + )); + } + + #[test] + fn retained_snapshot_read_does_not_create_missing_aggregate() { + let root = tempfile::tempdir().unwrap(); + let aggregate = root.path().join("missing-worker"); + + let error = read_retained_session_snapshot( + &aggregate, + "missing-worker", + DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, + ) + .unwrap_err(); + + assert!(matches!(error, RetainedSnapshotReadError::RetentionMissing)); + assert!(!aggregate.exists()); + } } diff --git a/crates/session-store/src/worker_metadata.rs b/crates/session-store/src/worker_metadata.rs index 4491f393e..209cfd9d8 100644 --- a/crates/session-store/src/worker_metadata.rs +++ b/crates/session-store/src/worker_metadata.rs @@ -383,6 +383,27 @@ impl WorkerAggregateStore { }) } + /// Open an existing aggregate without creating or rewriting retained files. + pub fn open_read_only( + root: impl Into, + worker_name: impl Into, + ) -> Result { + let root = root.into(); + let worker_name = worker_name.into(); + validate_worker_name(&worker_name)?; + if !root.is_dir() { + return Err(WorkerStoreError::Io(std::io::Error::new( + std::io::ErrorKind::NotFound, + "worker aggregate is not retained", + ))); + } + Ok(Self { + update_lock: metadata_lock(&root), + root, + worker_name, + }) + } + 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 ee00d508b..c5c8e9acb 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -84,6 +84,32 @@ impl WorkerSessionStore { }) } + /// Open an already-retained Session without creating directories or migrating + /// 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 manifest: SessionManifest = serde_json::from_slice(&bytes)?; + if manifest.schema_version != SESSION_SCHEMA_VERSION { + return Err(StoreError::Corrupt { + line: 0, + message: format!( + "Worker Session schema version {} requires migration; expected {}", + manifest.schema_version, SESSION_SCHEMA_VERSION + ), + }); + } + Ok(Self { + root, + session_id: Arc::new(Mutex::new(Some(manifest.session_id))), + append_lock: Arc::new(Mutex::new(())), + }) + } + + pub fn segment_log_len(&self, segment_id: SegmentId) -> Result { + Ok(fs::metadata(self.log_path(segment_id))?.len()) + } + pub fn root_dir(&self) -> &Path { &self.root } diff --git a/crates/tui/Cargo.toml b/crates/tui/Cargo.toml index 18147a9b8..9a61c328b 100644 --- a/crates/tui/Cargo.toml +++ b/crates/tui/Cargo.toml @@ -27,6 +27,7 @@ session-store = { workspace = true } ticket = { workspace = true } serde = { workspace = true, features = ["derive"] } pulldown-cmark = { version = "0.13.3", default-features = false } +async-trait.workspace = true agen.workspace = true [dev-dependencies] diff --git a/crates/tui/src/backend_worker_picker.rs b/crates/tui/src/backend_worker_picker.rs index 34ce59560..f0731dbc7 100644 --- a/crates/tui/src/backend_worker_picker.rs +++ b/crates/tui/src/backend_worker_picker.rs @@ -3,9 +3,8 @@ use std::io; use std::time::Duration; use client::{ - BackendRuntimeListTarget, BackendWorkerRestoreResponse, BackendWorkerRestoreState, - BackendWorkerSummary, list_backend_stopped_workers, list_backend_workers, - restore_backend_worker, + BackendRuntimeListTarget, BackendWorkerSummary, ServerApiClient, WorkerSessionAvailability, + list_backend_stopped_workers, list_backend_workers, }; use crossterm::event::{self, Event as TermEvent, KeyCode, KeyEventKind, KeyModifiers}; use ratatui::Frame; @@ -81,62 +80,39 @@ pub(crate) async fn run( } WorkerPickerResult::Selected(selected) => selected, }; - let worker = if selected.state == "stopped" { - let restore_target = target - .runtime_target(selected.runtime_id.clone(), selected.worker_id.clone()) + 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 restore = restore_backend_worker(&restore_target) + 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 restore Backend worker {}/{}: {error}", + "failed to observe retained Worker Session {}/{}: {error}", selected.runtime_id, selected.worker_id )) })?; - restored_worker(restore).map_err(|error| { - io::Error::other(format!( - "failed to restore Backend worker {}/{}: {error}", - selected.runtime_id, selected.worker_id - )) - })? - } else { - selected - }; - let attach_target = target - .runtime_target(worker.runtime_id, worker.worker_id) - .map_err(|error| io::Error::other(error.to_string()))?; + 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; } } -fn restored_worker(response: BackendWorkerRestoreResponse) -> Result { - if response.result.state != BackendWorkerRestoreState::Accepted { - let diagnostics = response - .result - .diagnostics - .iter() - .map(|diagnostic| format!("{}: {}", diagnostic.code, diagnostic.message)) - .collect::>() - .join("; "); - let state = match response.result.state { - BackendWorkerRestoreState::Accepted => unreachable!(), - BackendWorkerRestoreState::Rejected => "rejected", - BackendWorkerRestoreState::RolledBack => "rolled back", - BackendWorkerRestoreState::ReconciliationRequired => "reconciliation required", - }; - return Err(if diagnostics.is_empty() { - format!("restore was {state} without a diagnostic") - } else { - format!("restore was {state}: {diagnostics}") - }); - } - - response - .result - .worker - .ok_or_else(|| "restore was accepted without a Worker snapshot".to_string()) -} - fn dedup_workers(workers: &mut Vec) { let mut seen = std::collections::HashSet::new(); workers.retain(|worker| seen.insert((worker.runtime_id.clone(), worker.worker_id.clone()))); @@ -437,8 +413,7 @@ fn working_directory_text(worker: &BackendWorkerSummary) -> String { mod tests { use super::*; use client::{ - BackendDiagnostic, BackendDiagnosticSeverity, BackendWorkerCapabilitySummary, - BackendWorkerImplementationSummary, BackendWorkerRestoreResult, + BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, BackendWorkerWorkspaceSummary, }; @@ -494,67 +469,6 @@ mod tests { text_width(&text[..byte_offset]) } - fn restore_response( - state: BackendWorkerRestoreState, - worker: Option, - diagnostics: Vec, - ) -> BackendWorkerRestoreResponse { - BackendWorkerRestoreResponse { - workspace_id: "workspace-a".to_string(), - runtime_id: "runtime-a".to_string(), - worker_id: "worker-a".to_string(), - result: BackendWorkerRestoreResult { - state, - worker, - diagnostics, - }, - } - } - - #[test] - fn rejected_restore_surfaces_diagnostic_instead_of_attaching_selected_worker() { - let error = restored_worker(restore_response( - BackendWorkerRestoreState::Rejected, - None, - vec![BackendDiagnostic { - code: "working_directory_not_found".to_string(), - severity: BackendDiagnosticSeverity::Error, - message: "working directory was not found".to_string(), - }], - )) - .expect_err("rejected restore must not produce a Worker to attach"); - - assert_eq!( - error, - "restore was rejected: working_directory_not_found: working directory was not found" - ); - } - - #[test] - fn accepted_restore_requires_returned_worker_snapshot() { - let error = restored_worker(restore_response( - BackendWorkerRestoreState::Accepted, - None, - Vec::new(), - )) - .expect_err("accepted restore without a Worker must not attach the stale selection"); - - assert_eq!(error, "restore was accepted without a Worker snapshot"); - } - - #[test] - fn accepted_restore_returns_authoritative_worker_snapshot() { - let worker = worker("runtime-a", "worker-a", Some("builtin:companion")); - let restored = restored_worker(restore_response( - BackendWorkerRestoreState::Accepted, - Some(worker.clone()), - Vec::new(), - )) - .expect("accepted restore should return its Worker snapshot"); - - assert_eq!(restored, worker); - } - #[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 af0410661..c831bfa4d 100644 --- a/crates/tui/src/console/mod.rs +++ b/crates/tui/src/console/mod.rs @@ -457,20 +457,81 @@ async fn run_standalone_host( pub(crate) async fn run_backend_runtime( target: BackendRuntimeTarget, +) -> Result<(), Box> { + if let Some(snapshot) = target.initial_snapshot.clone() { + let client = Client::new(RetainedSnapshotSocket::new(snapshot)?); + return run_backend_console(target, client, false).await; + } + let client = connect_backend_runtime(target.clone()).await?; + run_backend_console(target, client, true).await +} + +async fn run_backend_console( + target: BackendRuntimeTarget, + client: Client, + connected: bool, ) -> Result<(), Box> { let worker_label = target.display_label(); - let attachment_target = target.clone(); - let client = connect_backend_runtime(target).await?; 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); - app.connected = true; - let mut connection = ConsoleConnection::with_backend_target(client, attachment_target); + app.connected = connected; + let mut connection = ConsoleConnection::with_backend_target(client, target); let result = run_loop(&mut terminal, &mut app, &mut connection).await; let _ = leave_fullscreen(&mut terminal); result } +struct RetainedSnapshotSocket { + event: Option, +} + +impl RetainedSnapshotSocket { + fn new(snapshot: protocol::SessionSnapshot) -> Result { + Ok(Self { + event: Some(serde_json::to_string(&Event::Snapshot { + session: snapshot, + greeting: protocol::Greeting { + worker_name: "retained worker".to_string(), + cwd: String::new(), + provider: "retained session".to_string(), + model: String::new(), + scope_summary: "read-only retained session".to_string(), + tools: Vec::new(), + context_window: 0, + context_tokens: 0, + }, + state: protocol::WorkerStateSnapshot { + last_command_id: 0, + state: protocol::WorkerState::Idle, + }, + in_flight: protocol::InFlightSnapshot::default(), + internal_workers: Vec::new(), + })?), + }) + } +} + +#[async_trait::async_trait] +impl Socket for RetainedSnapshotSocket { + type Error = std::convert::Infallible; + + async fn send(&mut self, _message: String) -> Result<(), Self::Error> { + Ok(()) + } + + async fn next(&mut self) -> Result, Self::Error> { + if self.event.is_some() { + return Ok(self.event.take()); + } + std::future::pending::, Self::Error>>().await + } + + fn try_next(&mut self) -> Result, Self::Error> { + Ok(self.event.take()) + } +} + fn enter_fullscreen() -> Result> { let mut stdout = io::stdout(); // Enable button-event tracking so the transcript can own drag selection; diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index dab47f3f3..54934181a 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -318,6 +318,11 @@ impl WorkerExecutionSpawnResult { } } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct WorkerSessionObservationRequest { + pub worker_ref: WorkerRef, +} + pub trait WorkerExecutionBackend: Send + Sync + 'static { fn backend_id(&self) -> &str; @@ -328,6 +333,16 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { Err("execution backend does not support Workspace Config fetching".to_string()) } + fn worker_session( + &self, + _request: WorkerSessionObservationRequest, + ) -> runtime_api::WorkerSessionAvailability { + runtime_api::WorkerSessionAvailability::Unavailable { + reason: runtime_api::WorkerSessionUnavailableReason::StorageUnavailable, + message: "retained session storage is unavailable".to_string(), + } + } + fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult; /// Read-only restore validation performed before any live controller, @@ -525,6 +540,13 @@ impl WorkerExecutionBackendRef { self.backend.fetch_workspace_config(request) } + pub(crate) fn worker_session( + &self, + request: WorkerSessionObservationRequest, + ) -> runtime_api::WorkerSessionAvailability { + self.backend.worker_session(request) + } + pub(crate) fn spawn_worker( &self, request: WorkerExecutionSpawnRequest, diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index a42ab6fa3..9b9b18d89 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2566,7 +2566,7 @@ pub enum RuntimeHttpServerError { mod tests { use super::*; use crate::auth::RuntimeIdentityMaterial; - use crate::catalog::{ConfigBundleRef, ProfileSelector, WorkerStatus, WorkspaceApiRef}; + use crate::catalog::{ConfigBundleRef, ProfileSelector, WorkspaceApiRef}; use crate::config_bundle::{ ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigProfileDescriptor, }; @@ -2588,6 +2588,7 @@ mod tests { use manifest::{Scope, SharedScope}; use sha2::Digest as _; use tower::ServiceExt; + use workspace_api::WorkerRestoreState; #[derive(Clone)] struct TestRuntimeAuthorizer(String); diff --git a/crates/worker-runtime/src/http_server/runtime_management_api.rs b/crates/worker-runtime/src/http_server/runtime_management_api.rs index 6547ecb43..c650f1be5 100644 --- a/crates/worker-runtime/src/http_server/runtime_management_api.rs +++ b/crates/worker-runtime/src/http_server/runtime_management_api.rs @@ -231,6 +231,35 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { Ok(runtime_api::WorkerResponse { worker }) } + async fn worker_session( + &self, + worker_id: String, + request: runtime_api::WorkerSessionRequest, + ) -> Result { + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let scope = auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + .ok_or_else(|| { + runtime_api::RuntimeApiError::new( + StatusCode::FORBIDDEN.as_u16(), + "runtime_worker_session_scope_required", + "Worker Session observation requires a Workspace-scoped capability", + ) + })?; + if scope.workspace_id != request.workspace_id { + return Err(runtime_api::RuntimeApiError::new( + StatusCode::FORBIDDEN.as_u16(), + "runtime_worker_session_workspace_scope_mismatch", + "Worker Session Workspace scope does not match the authenticated capability", + )); + } + self.state + .runtime + .worker_session_scoped(&scope, &worker_ref) + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error) + } + async fn create_worker( &self, value: runtime_api::CreateWorkerRequest, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index b00ecde8a..6236dac05 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -15,7 +15,8 @@ use crate::execution::WorkerExecutionRestoreRequest; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionBackendRef, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult, WorkerExecutionSpawnRequest, - WorkerExecutionSpawnResult, WorkspaceConfigFetchRequest, WorkspaceConfigFetchResult, + WorkerExecutionSpawnResult, WorkerSessionObservationRequest, WorkspaceConfigFetchRequest, + WorkspaceConfigFetchResult, }; #[cfg(feature = "fs-store")] use crate::fs_store::{ @@ -1424,6 +1425,37 @@ impl Runtime { Ok(state.worker(worker_ref)?.detail()) } + /// Observe a Worker's live protocol availability or its retained Session. + /// The same lifecycle lock used by stop/restore fences the decision and read. + pub fn worker_session_scoped( + &self, + scope: &RuntimeWorkspaceScope, + worker_ref: &WorkerRef, + ) -> Result { + let operation_lock = self.worker_operation_lock(worker_ref.worker_id)?; + let _operation_guard = operation_lock + .lock() + .map_err(|_| RuntimeError::StatePoisoned)?; + self.ensure_worker_in_workspace(scope, worker_ref)?; + let backend = { + let state = self.lock()?; + let worker = state.worker(worker_ref)?; + if worker.execution_handle.is_some() && worker.status.is_active() { + return Ok(runtime_api::WorkerSessionAvailability::LiveProtocol); + } + state.execution_backend.clone() + }; + Ok(match backend { + Some(backend) => backend.worker_session(WorkerSessionObservationRequest { + worker_ref: worker_ref.clone(), + }), + None => runtime_api::WorkerSessionAvailability::Unavailable { + reason: runtime_api::WorkerSessionUnavailableReason::StorageUnavailable, + message: "retained session storage is unavailable".to_string(), + }, + }) + } + /// Attach a live execution through a workspace-scoped Runtime authorization context. pub fn restore_worker_scoped( &self, @@ -6743,6 +6775,40 @@ mod tests { ); } + #[test] + fn session_observation_distinguishes_live_and_retained_without_restore() { + let (runtime, backend) = runtime_and_backend(); + let scope = scope("local", "server-a"); + let detail = runtime + .create_worker_scoped(&scope, task_request("observe without restore")) + .unwrap(); + + assert!(matches!( + runtime + .worker_session_scoped(&scope, &detail.worker_ref) + .unwrap(), + runtime_api::WorkerSessionAvailability::LiveProtocol + )); + + runtime + .send_protocol_method( + &detail.worker_ref, + Method::Shutdown { + command: test_command(), + }, + ) + .unwrap(); + let observation = runtime + .worker_session_scoped(&scope, &detail.worker_ref) + .unwrap(); + + assert!(matches!( + observation, + runtime_api::WorkerSessionAvailability::Unavailable { .. } + )); + assert_eq!(*backend.restore_count.lock().unwrap(), 0); + } + #[test] fn stopped_worker_rejects_input_until_explicitly_restored() { let (runtime, backend) = runtime_and_backend(); diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index ccb564b3b..fc360f395 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -24,7 +24,8 @@ use crate::config_bundle::{ConfigBundle, workspace_config_etag}; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionRestoreRequest, WorkerExecutionResult, WorkerExecutionSpawnRequest, - WorkerExecutionSpawnResult, WorkspaceConfigFetchRequest, WorkspaceConfigFetchResult, + WorkerExecutionSpawnResult, WorkerSessionObservationRequest, WorkspaceConfigFetchRequest, + WorkspaceConfigFetchResult, }; use crate::identity::WorkerRef; use crate::interaction::{WorkerInput, WorkerInputKind}; @@ -119,6 +120,14 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static { Err("Runtime Worker factory does not support Workspace Config fetching".to_string()) } + fn retained_session_snapshot( + &self, + _worker_ref: &WorkerRef, + ) -> Result + { + Err(session_store::RetainedSnapshotReadError::RetentionMissing) + } + async fn spawn_controller( &self, request: WorkerExecutionSpawnRequest, @@ -807,6 +816,22 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { fetch_workspace_config_http(&request, client).await } + fn retained_session_snapshot( + &self, + worker_ref: &WorkerRef, + ) -> Result + { + let Some(root) = self.worker_aggregate_root.as_ref() else { + return Err(session_store::RetainedSnapshotReadError::RetentionMissing); + }; + let worker_name = Self::runtime_worker_name_for_ref(worker_ref); + session_store::read_retained_session_snapshot( + &root.join(&worker_name), + &worker_name, + session_store::DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, + ) + } + fn observe_workspace_prompt_projection( &self, projection: worker::WorkspacePromptProjection, @@ -1923,6 +1948,45 @@ where self.run_on_adapter_runtime(async move { factory.fetch_workspace_config(request).await }) } + fn worker_session( + &self, + request: WorkerSessionObservationRequest, + ) -> runtime_api::WorkerSessionAvailability { + match self.factory.retained_session_snapshot(&request.worker_ref) { + Ok(retained) => runtime_api::WorkerSessionAvailability::RetainedSnapshot { + identity: runtime_api::RetainedSessionIdentity { + session_id: retained.identity.session_id, + segment_id: retained.identity.segment_id, + entry_count: retained.identity.entry_count, + }, + snapshot: retained.snapshot, + }, + Err(error) => { + let reason = match error { + session_store::RetainedSnapshotReadError::RetentionMissing => { + runtime_api::WorkerSessionUnavailableReason::RetentionMissing + } + session_store::RetainedSnapshotReadError::ActivePointerMissing => { + runtime_api::WorkerSessionUnavailableReason::ActivePointerMissing + } + session_store::RetainedSnapshotReadError::MigrationRequired => { + runtime_api::WorkerSessionUnavailableReason::MigrationRequired + } + session_store::RetainedSnapshotReadError::CorruptLog => { + runtime_api::WorkerSessionUnavailableReason::CorruptLog + } + session_store::RetainedSnapshotReadError::SnapshotTooLarge => { + runtime_api::WorkerSessionUnavailableReason::SnapshotTooLarge + } + }; + runtime_api::WorkerSessionAvailability::Unavailable { + reason, + message: error.to_string(), + } + } + } + } + fn observe_workspace_prompt_projection( &self, projection: worker::WorkspacePromptProjection, diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index f0069c05f..4560ddd15 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -24,6 +24,7 @@ flow = { path = "../flow" } manifest.workspace = true protocol = { workspace = true } runtime-api.workspace = true +server-api.workspace = true project-record.workspace = true reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "native-tls"] } ring.workspace = true diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index a3b13e52a..cd278145f 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1132,6 +1132,17 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { )) } + fn worker_session( + &self, + _workspace_id: &str, + _worker_ref: &EmbeddedWorkerRef, + ) -> Result { + Ok(runtime_api::WorkerSessionAvailability::Unavailable { + reason: runtime_api::WorkerSessionUnavailableReason::StorageUnavailable, + message: "retained session storage is unavailable".to_string(), + }) + } + fn observation_source( &self, _worker_id: &str, @@ -1922,6 +1933,29 @@ impl RuntimeRegistry { }) } + pub fn worker_session( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result { + validate_backend_identifier("runtime_id", &worker.runtime_id)?; + validate_backend_identifier("worker_id", &worker.worker_id)?; + let runtime = self.runtime(&worker.runtime_id)?; + let worker_ref = + EmbeddedWorkerRef::new(EmbeddedWorkerId::parse(&worker.worker_id).ok_or_else( + || RuntimeRegistryError::UnknownWorker { + worker: worker.clone(), + }, + )?); + runtime + .worker_session(workspace_id, &worker_ref) + .map_err(|message| RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: worker.runtime_id.clone(), + code: "worker_session_observation_failed".to_string(), + message, + }) + } + pub fn observation_source( &self, worker: &RuntimeWorkerRef, @@ -2773,6 +2807,17 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { .map_err(|error| error.to_string()) } + fn worker_session( + &self, + workspace_id: &str, + worker_ref: &EmbeddedWorkerRef, + ) -> Result { + let scope = RuntimeWorkspaceScope::new(workspace_id, "embedded-backend"); + self.runtime + .worker_session_scoped(&scope, worker_ref) + .map_err(|error| error.to_string()) + } + fn observation_source( &self, worker_id: &str, @@ -4712,6 +4757,23 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { .map_err(|diagnostic| diagnostic.message) } + fn worker_session( + &self, + workspace_id: &str, + worker_ref: &EmbeddedWorkerRef, + ) -> Result { + let worker_id = worker_ref.worker_id.to_string(); + let request = runtime_api::WorkerSessionRequest { + workspace_id: workspace_id.to_string(), + }; + self.run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { client.worker_session(worker_id, request).await }, + ) + .map_err(|diagnostic| diagnostic.message) + } + fn observation_source( &self, worker_id: &str, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 721977812..25f318af6 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -3040,6 +3040,42 @@ fn build_server_auth_router(api: ServerAuthApi) -> Router { .with_state(api) } +impl server_api::ServerApi for WorkspaceApi { + async fn worker_session( + &self, + workspace_id: String, + runtime_id: String, + worker_id: String, + ) -> std::result::Result + { + if workspace_id != self.config.workspace_id { + return Err(runtime_api::RuntimeApiError::new( + StatusCode::NOT_FOUND.as_u16(), + "workspace_not_found", + "Workspace was not found", + )); + } + let worker = RuntimeWorkerRef::new(runtime_id.clone(), worker_id.clone()); + let observation = self + .runtime + .worker_session(&workspace_id, &worker) + .map_err(|_error| { + runtime_api::RuntimeApiError::new( + StatusCode::BAD_GATEWAY.as_u16(), + "worker_session_observation_failed", + "Worker Session observation failed", + ) + })?; + Ok(server_api::WorkspaceWorkerSessionResponse { + subject: workspace_api::WorkspaceWorkerSubject::RuntimeWorker { + runtime_id, + worker_id, + }, + observation, + }) + } +} + fn build_inner_router(api: WorkspaceApi) -> Router { let auth = build_server_auth_router(ServerAuthApi::from(&api)); let scoped_ticket_relations_query_path = @@ -3632,6 +3668,9 @@ fn build_inner_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/hosts/{host_id}/workers", get(scoped_list_host_workers), ) + .merge( + server_api::ServerApiAxum::router(api.clone()).with_state::(()), + ) .fallback(get(static_or_spa_fallback)) .with_state(api); auth.merge(workspace) 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 6ca0bd0f4..b85189a65 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,6 +42,7 @@ PendingSubmissionsSnapshot, RewindTarget, Segment, + SessionSnapshot, } from "$lib/generated/protocol"; import { MAX_FILES_PER_SUBMISSION, @@ -933,8 +934,73 @@ target: ConsoleTarget, ) { if (!targetWorker || targetWorker.state === "stopped") { - protocolState = "closed"; - return; + 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 = "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), + }, + ]; + }); + return () => controller.abort(); } protocolState = "connecting"; const subscription = workspaceMultiplexer(target.workspaceId).subscribe( From 74981f821726b5cfe02ceacec2981980d7e17b68 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 04:59:05 +0900 Subject: [PATCH 2/4] 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( { From 96e04bbc32e20230e3cb13470df31fba453d3cca Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 05:16:20 +0900 Subject: [PATCH 3/4] fix: preserve auth and response bounds for session reads --- crates/client/src/backend_runtime.rs | 98 +++++++++++++++++++ crates/client/src/lib.rs | 2 +- crates/server-api/src/lib.rs | 1 + crates/tui/src/backend_worker_picker.rs | 15 +-- .../lib/workspace/session-observation.test.ts | 14 +++ .../src/lib/workspace/session-observation.ts | 8 ++ .../workers/[workerRef]/console/+page.svelte | 3 +- 7 files changed, 127 insertions(+), 14 deletions(-) diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index ff6c0fb27..3d61f8966 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -278,6 +278,7 @@ impl BackendRuntimeListTarget { pub enum BackendRuntimeClientError { InvalidTarget(String), Api(BackendApiClientError), + SessionApi(server_api::ServerApiClientError), Http(reqwest::Error), Protocol(String), } @@ -287,6 +288,7 @@ impl fmt::Display for BackendRuntimeClientError { match self { Self::InvalidTarget(message) => f.write_str(message), Self::Api(error) => write!(f, "{error}"), + Self::SessionApi(error) => write!(f, "{error}"), Self::Http(error) => write!(f, "{error}"), Self::Protocol(message) => f.write_str(message), } @@ -295,6 +297,12 @@ impl fmt::Display for BackendRuntimeClientError { impl std::error::Error for BackendRuntimeClientError {} +impl From for BackendRuntimeClientError { + fn from(value: server_api::ServerApiClientError) -> Self { + Self::SessionApi(value) + } +} + impl From for BackendRuntimeClientError { fn from(error: BackendApiClientError) -> Self { Self::Api(error) @@ -307,6 +315,59 @@ impl From for BackendRuntimeClientError { } } +const WORKER_SESSION_RESPONSE_LIMIT: usize = 16 * 1024 * 1024; + +#[derive(Clone, Debug)] +struct StoredBearerAuthorizer { + authorization: String, +} + +impl server_api::client_support::RequestAuthorizer for StoredBearerAuthorizer { + fn authorize( + &self, + _request: server_api::client_support::AuthorizerRequest<'_>, + ) -> Result { + let mut headers = reqwest::header::HeaderMap::new(); + headers.insert( + reqwest::header::AUTHORIZATION, + self.authorization + .parse() + .map_err(|_| server_api::client_support::AuthorizationError::new())?, + ); + Ok(headers) + } +} + +pub async fn observe_backend_worker_session( + target: &BackendRuntimeTarget, +) -> Result { + validate_target(target)?; + let backend = BackendApiClient::from_stored_token(&target.base_url)?; + observe_backend_worker_session_with_client(target, &backend).await +} + +async fn observe_backend_worker_session_with_client( + target: &BackendRuntimeTarget, + backend: &BackendApiClient, +) -> Result { + let client = server_api::ServerApiClient::builder(&target.base_url) + .map_err(|error| BackendRuntimeClientError::Protocol(error.to_string()))? + .authorizer(StoredBearerAuthorizer { + authorization: backend.authorization_header_value(), + }) + .response_body_limit(WORKER_SESSION_RESPONSE_LIMIT) + .build() + .map_err(|error| BackendRuntimeClientError::Protocol(error.to_string()))?; + client + .worker_session( + target.workspace_id.clone(), + target.runtime_id.clone(), + target.worker_id.clone(), + ) + .await + .map_err(Into::into) +} + pub async fn get_backend_worker_launch_options( target: &BackendWorkerLaunchTarget, ) -> Result { @@ -765,6 +826,43 @@ mod tests { (base_url, task) } + #[tokio::test] + async fn worker_session_request_uses_bearer_and_sixteen_mib_response_limit() { + let large_message = "x".repeat(2 * 1024 * 1024); + let (base_url, server) = serve_json_once(serde_json::json!({ + "subject": { + "kind": "runtime_worker", + "runtime_id": "runtime-a", + "worker_id": "worker-a" + }, + "availability": "unavailable", + "reason": "storage_unavailable", + "message": large_message + })) + .await; + let target = BackendRuntimeTarget::new(&base_url, "workspace-a", "runtime-a", "worker-a"); + let api = + BackendApiClient::from_access_token_for_test(&base_url, "session-secret").unwrap(); + + let response = observe_backend_worker_session_with_client(&target, &api) + .await + .unwrap(); + assert!(matches!( + response.observation, + runtime_api::WorkerSessionAvailability::Unavailable { ref message, .. } + if message.len() == 2 * 1024 * 1024 + )); + let request = server.await.unwrap(); + assert!(request.starts_with( + "GET /api/w/workspace-a/runtimes/runtime-a/workers/worker-a/session HTTP/1.1\r\n" + )); + assert!( + request + .to_ascii_lowercase() + .contains("authorization: bearer session-secret\r\n") + ); + } + #[tokio::test] async fn launch_options_request_uses_workspace_path_and_bearer_auth() { let (base_url, server) = serve_json_once(serde_json::json!({ diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index ad44a2083..64d2b25f1 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -30,7 +30,7 @@ pub use backend_runtime::{ BackendWorkerRestoreResult, BackendWorkerRestoreState, BackendWorkerSummary, BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, connect_backend_runtime, create_backend_worker, get_backend_worker_launch_options, list_backend_stopped_workers, - list_backend_workers, restore_backend_worker, + list_backend_workers, observe_backend_worker_session, restore_backend_worker, }; pub use backend_workspace::{ BackendWorkspace, BackendWorkspaceCatalogTarget, BackendWorkspaceClientError, diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 88e782fd3..4e9bc6026 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -8,6 +8,7 @@ use serde::{Deserialize, Serialize}; use workspace_api::WorkspaceWorkerSubject; pub type ServerApiError = runtime_api::RuntimeApiError; +pub type ServerApiClientError = client_support::ClientError; #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct WorkspaceWorkerSessionResponse { diff --git a/crates/tui/src/backend_worker_picker.rs b/crates/tui/src/backend_worker_picker.rs index 1717184b0..1768334b2 100644 --- a/crates/tui/src/backend_worker_picker.rs +++ b/crates/tui/src/backend_worker_picker.rs @@ -3,8 +3,8 @@ use std::io; use std::time::Duration; use client::{ - BackendRuntimeListTarget, BackendWorkerSummary, ServerApiClient, WorkerSessionAvailability, - list_backend_stopped_workers, list_backend_workers, + BackendRuntimeListTarget, BackendWorkerSummary, WorkerSessionAvailability, + list_backend_stopped_workers, list_backend_workers, observe_backend_worker_session, }; use crossterm::event::{self, Event as TermEvent, KeyCode, KeyEventKind, KeyModifiers}; use ratatui::Frame; @@ -83,16 +83,7 @@ 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()))?; - 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(), - ) + let response = observe_backend_worker_session(&attach_target) .await .map_err(|error| { io::Error::other(format!( diff --git a/web/workspace/src/lib/workspace/session-observation.test.ts b/web/workspace/src/lib/workspace/session-observation.test.ts index c6450ec2d..51e31cf02 100644 --- a/web/workspace/src/lib/workspace/session-observation.test.ts +++ b/web/workspace/src/lib/workspace/session-observation.test.ts @@ -2,6 +2,7 @@ import { isCurrentWorkerSessionRequest, workerSessionAction, type WorkerSessionRequestIdentity, + workerSessionRequestInit, } from "./session-observation"; import type { SessionSnapshot } from "$lib/generated/protocol"; @@ -23,6 +24,19 @@ const snapshot: SessionSnapshot = { entries: [], }; +Deno.test("worker session request includes same-origin credentials", () => { + const controller = new AbortController(); + const init = workerSessionRequestInit(controller.signal); + assert( + init.credentials === "same-origin", + "session observation must use the Workspace auth cookie boundary", + ); + assert( + init.signal === controller.signal, + "request must retain its abort fence", + ); +}); + Deno.test("worker session observation selects live, retained, and unavailable behavior", () => { assert( workerSessionAction({ availability: "live_protocol" }).kind === diff --git a/web/workspace/src/lib/workspace/session-observation.ts b/web/workspace/src/lib/workspace/session-observation.ts index 4b5309a10..9c5894507 100644 --- a/web/workspace/src/lib/workspace/session-observation.ts +++ b/web/workspace/src/lib/workspace/session-observation.ts @@ -17,6 +17,14 @@ export interface WorkerSessionRequestIdentity { workerId: string; } +export function workerSessionRequestInit(signal: AbortSignal): RequestInit { + return { + signal, + credentials: "same-origin", + headers: { accept: "application/json" }, + }; +} + export function workerSessionAction( observation: WorkerSessionObservation, ): WorkerSessionAction { 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 d8dd3d5b7..6ff2e33d2 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 @@ -55,6 +55,7 @@ import { isCurrentWorkerSessionRequest, workerSessionAction, + workerSessionRequestInit, type WorkerSessionObservation, type WorkerSessionRequestIdentity, } from "$lib/workspace/session-observation"; @@ -951,7 +952,7 @@ workerApiPath( `/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent(target.workerId)}/session`, ), - { signal: controller.signal, headers: { accept: "application/json" } }, + workerSessionRequestInit(controller.signal), ) .then(async (response) => { if (!response.ok) { From 9d8f294d752b928e4aeb293e89e3037c4f4ee31f Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 05:41:12 +0900 Subject: [PATCH 4/4] fix: import generated worker summary type --- web/workspace/src/lib/workspace/sidebar/types.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/web/workspace/src/lib/workspace/sidebar/types.ts b/web/workspace/src/lib/workspace/sidebar/types.ts index a6642c2c3..4e5aad378 100644 --- a/web/workspace/src/lib/workspace/sidebar/types.ts +++ b/web/workspace/src/lib/workspace/sidebar/types.ts @@ -5,6 +5,7 @@ import type { WorkerLaunchOptionsResponse as SharedWorkerLaunchOptionsResponse, WorkerLaunchProfileCandidate as SharedWorkerLaunchProfileCandidate, WorkerLaunchRuntimeOption as SharedWorkerLaunchRuntimeOption, + WorkerLaunchWorkerSummary as WorkerSummary, WorkingDirectoryRepositoryOption as SharedWorkingDirectoryRepositoryOption, } from "$lib/generated/worker-launch-api"; import type {