diff --git a/Cargo.lock b/Cargo.lock index 4b2f4b62..d87247cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3508,6 +3508,7 @@ dependencies = [ "schemars", "serde", "serde_json", + "sha2 0.11.0", "tokio", "ts-rs", "uuid", diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 6aae3f6a..b3cea8de 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -35,9 +35,9 @@ pub use backend_workspace::{ }; pub use client::{Client, ClientError}; pub use target::{ - BackendTarget, Dashboard, ResolvedTarget, StandaloneSessionListIntent, - StandaloneSessionResumeIntent, StandaloneTarget, Target, TargetError, TargetKind, - WorkerConnection, WorkerConnectionSelector, WorkerList, WorkerListRequest, WorkerSpawn, + BackendTarget, Dashboard, ResolvedTarget, StandaloneTarget, StandaloneWorkerListIntent, + StandaloneWorkerResumeIntent, Target, TargetError, TargetKind, WorkerConnection, + WorkerConnectionSelector, WorkerList, WorkerListRequest, WorkerSpawn, }; pub use workspace_api::{ObjectiveDetail, ObjectiveSummary}; pub use workspace_product::BackendWorkspaceProductClient; diff --git a/crates/client/src/target.rs b/crates/client/src/target.rs index 42695859..6b5fc783 100644 --- a/crates/client/src/target.rs +++ b/crates/client/src/target.rs @@ -105,16 +105,16 @@ pub struct WorkerSpawn { } #[derive(Debug, Clone, PartialEq, Eq)] -pub struct StandaloneSessionListIntent { +pub struct StandaloneWorkerListIntent { pub state_dir: PathBuf, pub cwd: PathBuf, pub include_all: bool, } #[derive(Debug, Clone, PartialEq, Eq)] -pub struct StandaloneSessionResumeIntent { +pub struct StandaloneWorkerResumeIntent { pub state_dir: PathBuf, - pub session_id: String, + pub worker_id: String, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -175,22 +175,22 @@ pub trait Target: fmt::Debug + Send + Sync { Err(TargetError::unsupported("Worker spawn", self.kind())) } - fn standalone_session_list( + fn standalone_worker_list( &self, _include_all: bool, - ) -> Result { + ) -> Result { Err(TargetError::unsupported( - "standalone session listing", + "standalone Worker listing", self.kind(), )) } - fn standalone_session_resume( + fn standalone_worker_resume( &self, - _session_id: String, - ) -> Result { + _worker_id: String, + ) -> Result { Err(TargetError::unsupported( - "standalone session restore", + "standalone Worker restore", self.kind(), )) } @@ -243,26 +243,26 @@ impl Target for StandaloneTarget { }) } - fn standalone_session_list( + fn standalone_worker_list( &self, include_all: bool, - ) -> Result { + ) -> Result { let cwd = std::env::current_dir() .map_err(|error| TargetError::invalid(self.kind(), error.to_string()))?; - Ok(StandaloneSessionListIntent { + Ok(StandaloneWorkerListIntent { state_dir: self.state_dir.clone(), cwd, include_all, }) } - fn standalone_session_resume( + fn standalone_worker_resume( &self, - session_id: String, - ) -> Result { - Ok(StandaloneSessionResumeIntent { + worker_id: String, + ) -> Result { + Ok(StandaloneWorkerResumeIntent { state_dir: self.state_dir.clone(), - session_id, + worker_id, }) } } @@ -437,17 +437,17 @@ mod tests { } #[test] - fn standalone_target_builds_explicit_session_intents() { - let target = StandaloneTarget::new("/tmp/yoi-client-sessions"); - let list = target.standalone_session_list(true).unwrap(); - assert_eq!(list.state_dir, PathBuf::from("/tmp/yoi-client-sessions")); + fn standalone_target_builds_explicit_worker_intents() { + let target = StandaloneTarget::new("/tmp/yoi-client-workers"); + let list = target.standalone_worker_list(true).unwrap(); + assert_eq!(list.state_dir, PathBuf::from("/tmp/yoi-client-workers")); assert!(list.include_all); assert!(list.cwd.is_absolute()); let resume = target - .standalone_session_resume("019d1234-0000-7000-8000-000000000000".to_string()) + .standalone_worker_resume("019d1234-0000-7000-8000-000000000000".to_string()) .unwrap(); assert_eq!(resume.state_dir, list.state_dir); - assert_eq!(resume.session_id, "019d1234-0000-7000-8000-000000000000"); + assert_eq!(resume.worker_id, "019d1234-0000-7000-8000-000000000000"); } } diff --git a/crates/protocol/Cargo.toml b/crates/protocol/Cargo.toml index 7d6032ab..8a48a041 100644 --- a/crates/protocol/Cargo.toml +++ b/crates/protocol/Cargo.toml @@ -14,6 +14,7 @@ json-schema = ["dep:schemars"] schemars = { workspace = true, optional = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } +sha2.workspace = true tokio = { workspace = true, features = ["io-util"], optional = true } ts-rs = { version = "12.0.1", optional = true } -uuid = { workspace = true, features = ["serde"] } +uuid = { workspace = true, features = ["serde", "v7"] } diff --git a/crates/protocol/src/identity.rs b/crates/protocol/src/identity.rs new file mode 100644 index 00000000..48c420ab --- /dev/null +++ b/crates/protocol/src/identity.rs @@ -0,0 +1,132 @@ +use std::{fmt, str::FromStr}; + +use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; +use sha2::{Digest, Sha256}; +use uuid::{Uuid, Version}; + +/// Stable Worker identity independent of its current Runtime placement or +/// conversation Session. +/// +/// Workspace authority allocates this ID for managed Workers. A standalone +/// Worker store allocates it locally when no Workspace authority is present. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct WorkerId(Uuid); + +impl WorkerId { + pub fn now_v7() -> Self { + Self(Uuid::now_v7()) + } + + /// Converts a legacy Runtime-local numeric id into a syntactically valid + /// migration-only UUIDv7 value. New Worker allocation must use `now_v7`. + pub fn from_legacy_u64(value: u64) -> Self { + let mut bytes = [0_u8; 16]; + bytes[8..].copy_from_slice(&value.to_be_bytes()); + bytes[6] = 0x70; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + Self(Uuid::from_bytes(bytes)) + } + + pub fn from_legacy_binding(workspace_id: &str, runtime_id: &str, value: u64) -> Self { + let mut hasher = Sha256::new(); + hasher.update(b"yoi.workspace-worker-id.v1\0"); + hasher.update(workspace_id.as_bytes()); + hasher.update([0]); + hasher.update(runtime_id.as_bytes()); + hasher.update([0]); + hasher.update(value.to_be_bytes()); + let digest = hasher.finalize(); + let mut bytes = [0_u8; 16]; + bytes.copy_from_slice(&digest[..16]); + // Migrated ids sort before normally allocated UUIDv7 values while retaining + // deterministic collision-resistant payload bits. + bytes[..6].fill(0); + bytes[6] = (bytes[6] & 0x0f) | 0x70; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + Self(Uuid::from_bytes(bytes)) + } + + pub fn parse(value: &str) -> Option { + let value = Uuid::parse_str(value).ok()?; + (value.get_version() == Some(Version::SortRand)).then_some(Self(value)) + } + + pub const fn as_uuid(self) -> Uuid { + self.0 + } + + #[must_use] + pub fn short(self) -> String { + let simple = self.0.simple().to_string(); + simple[simple.len() - 12..].to_string() + } +} + +impl fmt::Display for WorkerId { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(formatter) + } +} + +impl FromStr for WorkerId { + type Err = WorkerIdParseError; + + fn from_str(value: &str) -> Result { + Self::parse(value).ok_or(WorkerIdParseError) + } +} + +impl Serialize for WorkerId { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(&self.to_string()) + } +} + +impl<'de> Deserialize<'de> for WorkerId { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = String::deserialize(deserializer)?; + Self::parse(&value).ok_or_else(|| de::Error::custom("Worker id must be a UUIDv7")) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct WorkerIdParseError; + +impl fmt::Display for WorkerIdParseError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("Worker id must be a UUIDv7") + } +} + +impl std::error::Error for WorkerIdParseError {} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn worker_id_accepts_only_uuid_v7() { + let worker_id = WorkerId::now_v7(); + assert_eq!(WorkerId::parse(&worker_id.to_string()), Some(worker_id)); + assert!(WorkerId::parse("30").is_none()); + assert!(WorkerId::parse(&Uuid::nil().to_string()).is_none()); + } + + #[test] + fn legacy_worker_id_mapping_is_stable() { + assert_eq!( + WorkerId::from_legacy_binding("workspace", "runtime", 42), + WorkerId::from_legacy_binding("workspace", "runtime", 42) + ); + assert_ne!( + WorkerId::from_legacy_binding("workspace", "runtime", 42), + WorkerId::from_legacy_binding("workspace", "runtime", 43) + ); + } +} diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 8c498f09..9d494b5d 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -1,3 +1,4 @@ +pub mod identity; #[cfg(feature = "stream")] pub mod stream; pub mod subscription; @@ -8,6 +9,8 @@ use std::path::PathBuf; use serde::{Deserialize, Serialize}; +pub use identity::{WorkerId, WorkerIdParseError}; + fn default_true() -> bool { true } diff --git a/crates/standalone/src/host.rs b/crates/standalone/src/host.rs index b6e2f7ba..7bfc1e51 100644 --- a/crates/standalone/src/host.rs +++ b/crates/standalone/src/host.rs @@ -5,7 +5,7 @@ use agen::llm_client::client::LlmClient; use client::Client; use client::transport::in_process::{Peer as InProcessPeer, Socket as InProcessSocket}; use protocol::stream::{decode_method, encode_event}; -use protocol::{Event, Method}; +use protocol::{Event, Method, WorkerId}; use session_store::{ CombinedStore, FsStore, FsWorkerStore, WorkerActiveSegmentRef, WorkerMetadataStore, }; @@ -20,14 +20,14 @@ use worker::{BootstrappedWorker, WorkerError, WorkerFilesystemAuthority, WorkerW use crate::launch::ResolvedStandaloneLaunch; use crate::store::{ - StaleLeasePolicy, StandaloneSessionId, StandaloneSessionLease, StandaloneSessionRecord, - StandaloneSessionStore, StandaloneShutdownReason, StandaloneStoreError, + StaleLeasePolicy, StandaloneShutdownReason, StandaloneStoreError, StandaloneWorkerLease, + StandaloneWorkerRecord, StandaloneWorkerStore, }; const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10); type StandaloneBackingStore = CombinedStore; -/// One client-owned top-level Worker and its standalone session authority. +/// One client-owned top-level Worker and its standalone Worker authority. /// /// The host deliberately exposes the existing typed Worker protocol rather than owning an /// HTTP/WebSocket server or creating Runtime/Workspace/Ticket/Workdir domain records. @@ -35,21 +35,21 @@ pub struct StandaloneHost { handle: worker::WorkerHandle, shutdown: Option, shutdown_timeout: Duration, - store: StandaloneSessionStore, + store: StandaloneWorkerStore, worker_store: FsWorkerStore, - record: StandaloneSessionRecord, - lease: Option, + record: StandaloneWorkerRecord, + lease: Option, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Error)] pub enum StandaloneStartupError { #[error("the standalone state store could not be opened or validated")] StateStore, - #[error("the standalone session is already active")] - SessionActive, - #[error("the standalone session lease cannot be observed safely; recovery is rejected")] + #[error("the standalone Worker is already active")] + WorkerActive, + #[error("the standalone Worker lease cannot be observed safely; recovery is rejected")] LeaseLivenessUnknown, - #[error("the standalone session working directory is unavailable or changed")] + #[error("the standalone Worker working directory is unavailable or changed")] WorkingDirectoryUnavailable, #[error("the resolved Worker configuration or persisted history is invalid")] WorkerConfiguration, @@ -67,7 +67,7 @@ pub enum StandaloneShutdownError { DeadlineExceeded, #[error("the standalone Worker shutdown confirmation was lost")] ConfirmationLost, - #[error("the standalone session final state could not be committed")] + #[error("the standalone Worker final state could not be committed")] StateStore, } @@ -87,22 +87,24 @@ impl StandaloneHost { } async fn start_with_optional_model_client( - mut launch: ResolvedStandaloneLaunch, + launch: ResolvedStandaloneLaunch, model_client: Option>, ) -> Result { - let store = StandaloneSessionStore::open(&launch.state_dir) - .map_err(classify_store_startup_error)?; + let store = + StandaloneWorkerStore::open(&launch.state_dir).map_err(classify_store_startup_error)?; let allocation = store .allocate(&launch.cwd, StaleLeasePolicy::Reject) .map_err(classify_store_startup_error)?; - let id = allocation.id(); + let worker_id = allocation.worker_id(); - // The standalone session ID is the local identity. A unique internal Worker name avoids - // process-global allocation collisions without creating a Runtime/Workspace Worker ID. - launch.profile.manifest.worker.name = format!("standalone-{id}"); + // WorkerId is the stable identity. The current Worker store remains + // name-keyed, so keep its derived storage key separate from the + // user-facing profile name. let manifest = launch.profile.manifest.clone(); - let worker_name = manifest.worker.name.clone(); - let (backing_store, worker_store) = match backing_store(&store, id) { + let storage_key = format!("standalone-{worker_id}"); + let mut bootstrap_manifest = manifest.clone(); + bootstrap_manifest.worker.name = storage_key.clone(); + let (backing_store, worker_store) = match backing_store(&store, worker_id) { Ok(stores) => stores, Err(error) => { let _ = store.abandon_allocation(allocation); @@ -112,10 +114,10 @@ impl StandaloneHost { let filesystem_authority = WorkerFilesystemAuthority::local(launch.cwd.clone(), launch.cwd.clone()); let workspace_context = WorkerWorkspaceContext::local_filesystem(None); - let runtime_base = store.runtime_dir(id); + let runtime_base = store.runtime_dir(worker_id); let mut bootstrap = WorkerBootstrap::new( - manifest.clone(), + bootstrap_manifest, backing_store, launch.prompt_catalog, workspace_context, @@ -133,7 +135,7 @@ impl StandaloneHost { return Err(classify_startup_error(error)); } }; - let active = match active_pointer(&worker_store, &worker_name) { + let active = match active_pointer(&worker_store, &storage_key) { Ok(active) => active, Err(error) => { stop_started_worker(started).await; @@ -141,16 +143,20 @@ impl StandaloneHost { return Err(error); } }; - let record = - match store.commit_created(&allocation, manifest, active.session_id, active.segment_id) - { - Ok(record) => record, - Err(_) => { - stop_started_worker(started).await; - let _ = store.abandon_allocation(allocation); - return Err(StandaloneStartupError::StateStore); - } - }; + let record = match store.commit_created( + &allocation, + manifest, + storage_key, + active.session_id, + active.segment_id, + ) { + Ok(record) => record, + Err(_) => { + stop_started_worker(started).await; + let _ = store.abandon_allocation(allocation); + return Err(StandaloneStartupError::StateStore); + } + }; Ok(Self::from_started( started, store, @@ -162,50 +168,46 @@ impl StandaloneHost { pub async fn restore( state_dir: PathBuf, - session_id: StandaloneSessionId, + worker_id: WorkerId, ) -> Result { - Self::restore_with_optional_model_client(state_dir, session_id, None).await + Self::restore_with_optional_model_client(state_dir, worker_id, None).await } pub async fn restore_with_model_client( state_dir: PathBuf, - session_id: StandaloneSessionId, + worker_id: WorkerId, model_client: C, ) -> Result where C: LlmClient + 'static, { - Self::restore_with_optional_model_client( - state_dir, - session_id, - Some(Box::new(model_client)), - ) - .await + Self::restore_with_optional_model_client(state_dir, worker_id, Some(Box::new(model_client))) + .await } async fn restore_with_optional_model_client( state_dir: PathBuf, - session_id: StandaloneSessionId, + worker_id: WorkerId, model_client: Option>, ) -> Result { - let store = - StandaloneSessionStore::open(state_dir).map_err(classify_store_startup_error)?; + let store = StandaloneWorkerStore::open(state_dir).map_err(classify_store_startup_error)?; let record = store - .load(session_id) + .load(worker_id) .map_err(classify_store_startup_error)?; record.cwd.verify().map_err(classify_store_startup_error)?; let lease = store - .acquire_lease(session_id, StaleLeasePolicy::Recover) + .acquire_lease(worker_id, StaleLeasePolicy::Recover) .map_err(classify_store_startup_error)?; - let (backing_store, worker_store) = backing_store(&store, session_id)?; - let worker_name = record.worker_name.clone(); - let manifest = record.manifest.clone(); + let (backing_store, worker_store) = backing_store(&store, worker_id)?; + let storage_key = record.storage_key.clone(); + let mut manifest = record.manifest.clone(); + manifest.worker.name = storage_key.clone(); let filesystem_authority = WorkerFilesystemAuthority::local( record.cwd.canonical_path.clone(), record.cwd.canonical_path.clone(), ); let workspace_context = WorkerWorkspaceContext::local_filesystem(None); - let runtime_base = store.runtime_dir(session_id); + let runtime_base = store.runtime_dir(worker_id); let mut bootstrap = WorkerBootstrap::new( manifest, @@ -220,11 +222,11 @@ impl StandaloneHost { bootstrap = bootstrap.with_model_client(model_client); } let prepared = bootstrap - .prepare_restored(&worker_name) + .prepare_restored(&storage_key) .await .map_err(classify_startup_error)?; let started = prepared.start().await.map_err(classify_startup_error)?; - let active = match active_pointer(&worker_store, &worker_name) { + let active = match active_pointer(&worker_store, &storage_key) { Ok(active) => active, Err(error) => { stop_started_worker(started).await; @@ -251,10 +253,10 @@ impl StandaloneHost { fn from_started( started: BootstrappedWorker, - store: StandaloneSessionStore, + store: StandaloneWorkerStore, worker_store: FsWorkerStore, - record: StandaloneSessionRecord, - lease: StandaloneSessionLease, + record: StandaloneWorkerRecord, + lease: StandaloneWorkerLease, ) -> Self { Self { handle: started.handle, @@ -268,12 +270,12 @@ impl StandaloneHost { } #[must_use] - pub fn session_id(&self) -> StandaloneSessionId { - self.record.session_id + pub fn worker_id(&self) -> WorkerId { + self.record.worker_id } #[must_use] - pub fn record(&self) -> &StandaloneSessionRecord { + pub fn record(&self) -> &StandaloneWorkerRecord { &self.record } @@ -310,7 +312,7 @@ impl StandaloneHost { return Err(StandaloneShutdownError::DeadlineExceeded); } } - let active = match active_pointer(&self.worker_store, &self.record.worker_name) { + let active = match active_pointer(&self.worker_store, &self.record.storage_key) { Ok(active) => active, Err(_) => { self.retain_lease(); @@ -451,12 +453,12 @@ async fn send_protocol_event(peer: &InProcessPeer, event: Event) -> bool { } fn backing_store( - store: &StandaloneSessionStore, - id: StandaloneSessionId, + store: &StandaloneWorkerStore, + worker_id: WorkerId, ) -> Result<(StandaloneBackingStore, FsWorkerStore), StandaloneStartupError> { - let session_store = - FsStore::new(store.session_log_dir(id)).map_err(|_| StandaloneStartupError::StateStore)?; - let worker_store = FsWorkerStore::new(store.worker_metadata_dir(id)) + let session_store = FsStore::new(store.sessions_dir(worker_id)) + .map_err(|_| StandaloneStartupError::StateStore)?; + let worker_store = FsWorkerStore::new(store.worker_metadata_dir(worker_id)) .map_err(|_| StandaloneStartupError::StateStore)?; Ok(( CombinedStore::new(session_store, worker_store.clone()), @@ -466,10 +468,10 @@ fn backing_store( fn active_pointer( worker_store: &FsWorkerStore, - worker_name: &str, + storage_key: &str, ) -> Result { worker_store - .read_by_name(worker_name) + .read_by_name(storage_key) .map_err(|_| StandaloneStartupError::StateStore)? .and_then(|metadata| metadata.active) .ok_or(StandaloneStartupError::StateStore) @@ -482,7 +484,7 @@ async fn stop_started_worker(started: BootstrappedWorker) { fn classify_store_startup_error(error: StandaloneStoreError) -> StandaloneStartupError { match error { - StandaloneStoreError::SessionLeased(_) => StandaloneStartupError::SessionActive, + StandaloneStoreError::WorkerLeased(_) => StandaloneStartupError::WorkerActive, StandaloneStoreError::LeaseLivenessUnknown(_) => { StandaloneStartupError::LeaseLivenessUnknown } diff --git a/crates/standalone/src/lib.rs b/crates/standalone/src/lib.rs index 8c04f863..100537a9 100644 --- a/crates/standalone/src/lib.rs +++ b/crates/standalone/src/lib.rs @@ -10,8 +10,8 @@ pub mod store; pub use host::{StandaloneHost, StandaloneShutdownError, StandaloneStartupError}; pub use launch::{ResolvedStandaloneLaunch, StandaloneLaunchConfig, StandaloneLaunchError}; +pub use protocol::WorkerId; pub use store::{ - StaleLeasePolicy, StandaloneCwdIdentity, StandaloneListScope, StandaloneSessionId, - StandaloneSessionRecord, StandaloneSessionStatus, StandaloneSessionStore, - StandaloneShutdownReason, StandaloneStoreError, + StaleLeasePolicy, StandaloneCwdIdentity, StandaloneListScope, StandaloneShutdownReason, + StandaloneStoreError, StandaloneWorkerRecord, StandaloneWorkerStatus, StandaloneWorkerStore, }; diff --git a/crates/standalone/src/store.rs b/crates/standalone/src/store.rs index e1dfea0c..78923f3a 100644 --- a/crates/standalone/src/store.rs +++ b/crates/standalone/src/store.rs @@ -1,12 +1,11 @@ -use std::fmt; use std::fs::{self, File, OpenOptions}; use std::io::{self, Write}; use std::path::{Path, PathBuf}; -use std::str::FromStr; use std::time::{SystemTime, UNIX_EPOCH}; use fs4::fs_std::FileExt; use manifest::WorkerManifest; +use protocol::WorkerId; use serde::{Deserialize, Serialize}; use session_store::{SegmentId, SessionId}; use thiserror::Error; @@ -16,47 +15,10 @@ const RECORD_FILE: &str = "record.json"; const COMMIT_MARKER: &str = "commit.pending"; const LEASE_FILE: &str = "lease.json"; const LEASE_LOCK_FILE: &str = "lease.lock"; -const SESSION_DIR: &str = "session"; +const SESSIONS_DIR: &str = "sessions"; const WORKER_DIR: &str = "worker"; const SCHEMA_VERSION: u32 = 1; -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] -#[serde(transparent)] -pub struct StandaloneSessionId(Uuid); - -impl StandaloneSessionId { - #[must_use] - pub fn new() -> Self { - Self(Uuid::now_v7()) - } - - #[must_use] - pub fn short(self) -> String { - let simple = self.0.simple().to_string(); - simple[simple.len() - 12..].to_string() - } -} - -impl Default for StandaloneSessionId { - fn default() -> Self { - Self::new() - } -} - -impl fmt::Display for StandaloneSessionId { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - self.0.fmt(formatter) - } -} - -impl FromStr for StandaloneSessionId { - type Err = uuid::Error; - - fn from_str(value: &str) -> Result { - Uuid::parse_str(value).map(Self) - } -} - #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct StandaloneCwdIdentity { pub canonical_path: PathBuf, @@ -100,7 +62,7 @@ impl StandaloneCwdIdentity { #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] -pub enum StandaloneSessionStatus { +pub enum StandaloneWorkerStatus { Active, Stopped, } @@ -115,17 +77,20 @@ pub enum StandaloneShutdownReason { } #[derive(Debug, Clone, Serialize, Deserialize)] -pub struct StandaloneSessionRecord { +pub struct StandaloneWorkerRecord { pub schema_version: u32, pub revision: u64, - pub session_id: StandaloneSessionId, + pub worker_id: WorkerId, + /// User-facing Worker name resolved from the profile. pub worker_name: String, + /// Internal key used by the current name-keyed Worker store. + pub storage_key: String, pub cwd: StandaloneCwdIdentity, pub manifest: WorkerManifest, pub active_session_id: SessionId, #[serde(default, skip_serializing_if = "Option::is_none")] pub active_segment_id: Option, - pub status: StandaloneSessionStatus, + pub status: StandaloneWorkerStatus, pub created_at_unix_ms: u64, pub updated_at_unix_ms: u64, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -145,11 +110,11 @@ pub enum StaleLeasePolicy { } #[derive(Debug, Clone)] -pub struct StandaloneSessionStore { +pub struct StandaloneWorkerStore { root: PathBuf, } -impl StandaloneSessionStore { +impl StandaloneWorkerStore { pub fn open(root: impl Into) -> Result { let root = root.into(); fs::create_dir_all(&root).map_err(StandaloneStoreError::Io)?; @@ -171,35 +136,41 @@ impl StandaloneSessionStore { &self, cwd: impl AsRef, policy: StaleLeasePolicy, - ) -> Result { - let id = StandaloneSessionId::new(); + ) -> Result { + let worker_id = WorkerId::now_v7(); let cwd = StandaloneCwdIdentity::capture(cwd)?; - let dir = self.session_dir(id); + let dir = self.worker_dir(worker_id); fs::create_dir(&dir).map_err(StandaloneStoreError::Io)?; - fs::create_dir(dir.join(SESSION_DIR)).map_err(StandaloneStoreError::Io)?; + fs::create_dir(dir.join(SESSIONS_DIR)).map_err(StandaloneStoreError::Io)?; fs::create_dir(dir.join(WORKER_DIR)).map_err(StandaloneStoreError::Io)?; - let lease = self.acquire_lease(id, policy)?; - Ok(StandaloneSessionAllocation { id, cwd, lease }) + let lease = self.acquire_lease(worker_id, policy)?; + Ok(StandaloneWorkerAllocation { + worker_id, + cwd, + lease, + }) } pub fn commit_created( &self, - allocation: &StandaloneSessionAllocation, + allocation: &StandaloneWorkerAllocation, manifest: WorkerManifest, + storage_key: String, active_session_id: SessionId, active_segment_id: Option, - ) -> Result { + ) -> Result { let now = now_unix_ms()?; - let record = StandaloneSessionRecord { + let record = StandaloneWorkerRecord { schema_version: SCHEMA_VERSION, revision: 1, - session_id: allocation.id, + worker_id: allocation.worker_id, worker_name: manifest.worker.name.clone(), + storage_key, cwd: allocation.cwd.clone(), manifest, active_session_id, active_segment_id, - status: StandaloneSessionStatus::Active, + status: StandaloneWorkerStatus::Active, created_at_unix_ms: now, updated_at_unix_ms: now, shutdown_reason: None, @@ -208,22 +179,19 @@ impl StandaloneSessionStore { Ok(record) } - pub fn load( - &self, - id: StandaloneSessionId, - ) -> Result { - let dir = self.session_dir(id); + pub fn load(&self, id: WorkerId) -> Result { + let dir = self.worker_dir(id); if dir.join(COMMIT_MARKER).exists() { return Err(StandaloneStoreError::IncompleteCommit(id)); } let bytes = fs::read(dir.join(RECORD_FILE)).map_err(|error| { if error.kind() == io::ErrorKind::NotFound { - StandaloneStoreError::SessionNotFound(id) + StandaloneStoreError::WorkerNotFound(id) } else { StandaloneStoreError::Io(error) } })?; - let record: StandaloneSessionRecord = serde_json::from_slice(&bytes) + let record: StandaloneWorkerRecord = serde_json::from_slice(&bytes) .map_err(|source| StandaloneStoreError::CorruptRecord { id, source })?; if record.schema_version > SCHEMA_VERSION { return Err(StandaloneStoreError::NewerSchema { @@ -232,7 +200,7 @@ impl StandaloneSessionStore { supported: SCHEMA_VERSION, }); } - if record.schema_version != SCHEMA_VERSION || record.session_id != id { + if record.schema_version != SCHEMA_VERSION || record.worker_id != id { return Err(StandaloneStoreError::InvalidRecord(id)); } Ok(record) @@ -243,7 +211,7 @@ impl StandaloneSessionStore { cwd: impl AsRef, scope: StandaloneListScope, limit: usize, - ) -> Result, StandaloneStoreError> { + ) -> Result, StandaloneStoreError> { let current_cwd = (scope == StandaloneListScope::CurrentCwd) .then(|| StandaloneCwdIdentity::capture(cwd)) .transpose()?; @@ -269,12 +237,7 @@ impl StandaloneSessionStore { right .updated_at_unix_ms .cmp(&left.updated_at_unix_ms) - .then_with(|| { - right - .session_id - .to_string() - .cmp(&left.session_id.to_string()) - }) + .then_with(|| right.worker_id.to_string().cmp(&left.worker_id.to_string())) }); records.truncate(limit); Ok(records) @@ -282,10 +245,10 @@ impl StandaloneSessionStore { pub fn acquire_lease( &self, - id: StandaloneSessionId, + id: WorkerId, policy: StaleLeasePolicy, - ) -> Result { - let dir = self.session_dir(id); + ) -> Result { + let dir = self.worker_dir(id); let path = dir.join(LEASE_FILE); let _guard = LeaseMutationGuard::acquire(&dir)?; let lease = LeaseRecord::current()?; @@ -296,7 +259,7 @@ impl StandaloneSessionStore { file.write_all(b"\n").map_err(StandaloneStoreError::Io)?; file.sync_all().map_err(StandaloneStoreError::Io)?; sync_directory(&dir)?; - return Ok(StandaloneSessionLease { + return Ok(StandaloneWorkerLease { path, lease_id: lease.lease_id, released: false, @@ -306,7 +269,7 @@ impl StandaloneSessionStore { let existing = read_lease(&path, id)?; match existing.liveness() { LeaseLiveness::Live => { - return Err(StandaloneStoreError::SessionLeased(id)); + return Err(StandaloneStoreError::WorkerLeased(id)); } LeaseLiveness::Unknown => { return Err(StandaloneStoreError::LeaseLivenessUnknown(id)); @@ -326,16 +289,16 @@ impl StandaloneSessionStore { pub fn update_active_pointer( &self, - record: &StandaloneSessionRecord, + record: &StandaloneWorkerRecord, active_session_id: SessionId, active_segment_id: Option, - ) -> Result { + ) -> Result { let mut next = record.clone(); next.revision = next.revision.saturating_add(1); next.updated_at_unix_ms = now_unix_ms()?; next.active_session_id = active_session_id; next.active_segment_id = active_segment_id; - next.status = StandaloneSessionStatus::Active; + next.status = StandaloneWorkerStatus::Active; next.shutdown_reason = None; self.commit_record(Some(record.revision), &next)?; Ok(next) @@ -343,73 +306,73 @@ impl StandaloneSessionStore { pub fn mark_stopped( &self, - record: &StandaloneSessionRecord, + record: &StandaloneWorkerRecord, active_session_id: SessionId, active_segment_id: Option, reason: StandaloneShutdownReason, - ) -> Result { + ) -> Result { let mut next = record.clone(); next.revision = next.revision.saturating_add(1); next.updated_at_unix_ms = now_unix_ms()?; next.active_session_id = active_session_id; next.active_segment_id = active_segment_id; - next.status = StandaloneSessionStatus::Stopped; + next.status = StandaloneWorkerStatus::Stopped; next.shutdown_reason = Some(reason); self.commit_record(Some(record.revision), &next)?; Ok(next) } - pub fn delete(&self, id: StandaloneSessionId) -> Result<(), StandaloneStoreError> { + pub fn delete(&self, id: WorkerId) -> Result<(), StandaloneStoreError> { let record = self.load(id)?; - if record.status != StandaloneSessionStatus::Stopped { + if record.status != StandaloneWorkerStatus::Stopped { return Err(StandaloneStoreError::DeleteActive(id)); } - let session_dir = self.session_dir(id); - let _guard = LeaseMutationGuard::acquire(&session_dir)?; - let lease_path = session_dir.join(LEASE_FILE); + let worker_dir = self.worker_dir(id); + let _guard = LeaseMutationGuard::acquire(&worker_dir)?; + let lease_path = worker_dir.join(LEASE_FILE); if lease_path.exists() { let lease = read_lease(&lease_path, id)?; return Err(match lease.liveness() { - LeaseLiveness::Live => StandaloneStoreError::SessionLeased(id), + LeaseLiveness::Live => StandaloneStoreError::WorkerLeased(id), LeaseLiveness::Stale => StandaloneStoreError::StaleLease(id), LeaseLiveness::Unknown => StandaloneStoreError::LeaseLivenessUnknown(id), }); } - fs::remove_dir_all(self.session_dir(id)).map_err(StandaloneStoreError::Io)?; + fs::remove_dir_all(self.worker_dir(id)).map_err(StandaloneStoreError::Io)?; sync_directory(&self.root) } #[must_use] - pub fn session_log_dir(&self, id: StandaloneSessionId) -> PathBuf { - self.session_dir(id).join(SESSION_DIR) + pub fn sessions_dir(&self, id: WorkerId) -> PathBuf { + self.worker_dir(id).join(SESSIONS_DIR) } #[must_use] - pub fn worker_metadata_dir(&self, id: StandaloneSessionId) -> PathBuf { - self.session_dir(id).join(WORKER_DIR) + pub fn worker_metadata_dir(&self, id: WorkerId) -> PathBuf { + self.worker_dir(id).join(WORKER_DIR) } #[must_use] - pub(crate) fn runtime_dir(&self, id: StandaloneSessionId) -> PathBuf { - self.session_dir(id).join("runtime") + pub(crate) fn runtime_dir(&self, id: WorkerId) -> PathBuf { + self.worker_dir(id).join("runtime") } pub(crate) fn abandon_allocation( &self, - allocation: StandaloneSessionAllocation, + allocation: StandaloneWorkerAllocation, ) -> Result<(), StandaloneStoreError> { - let id = allocation.id; + let worker_id = allocation.worker_id; allocation.lease.release()?; - fs::remove_dir_all(self.session_dir(id)).map_err(StandaloneStoreError::Io)?; + fs::remove_dir_all(self.worker_dir(worker_id)).map_err(StandaloneStoreError::Io)?; sync_directory(&self.root) } fn commit_record( &self, expected_revision: Option, - next: &StandaloneSessionRecord, + next: &StandaloneWorkerRecord, ) -> Result<(), StandaloneStoreError> { - let dir = self.session_dir(next.session_id); + let dir = self.worker_dir(next.worker_id); let marker = dir.join(COMMIT_MARKER); let mut marker_file = OpenOptions::new() .write(true) @@ -417,7 +380,7 @@ impl StandaloneSessionStore { .open(&marker) .map_err(|error| { if error.kind() == io::ErrorKind::AlreadyExists { - StandaloneStoreError::IncompleteCommit(next.session_id) + StandaloneStoreError::IncompleteCommit(next.worker_id) } else { StandaloneStoreError::Io(error) } @@ -427,11 +390,11 @@ impl StandaloneSessionStore { sync_directory(&dir)?; if let Some(expected) = expected_revision { - let current = self.load_record_while_committing(next.session_id)?; + let current = self.load_record_while_committing(next.worker_id)?; if current.revision != expected { let _ = fs::remove_file(&marker); return Err(StandaloneStoreError::RevisionConflict { - id: next.session_id, + id: next.worker_id, expected, found: current.revision, }); @@ -461,30 +424,30 @@ impl StandaloneSessionStore { fn load_record_while_committing( &self, - id: StandaloneSessionId, - ) -> Result { + id: WorkerId, + ) -> Result { let bytes = - fs::read(self.session_dir(id).join(RECORD_FILE)).map_err(StandaloneStoreError::Io)?; + fs::read(self.worker_dir(id).join(RECORD_FILE)).map_err(StandaloneStoreError::Io)?; serde_json::from_slice(&bytes) .map_err(|source| StandaloneStoreError::CorruptRecord { id, source }) } - fn session_dir(&self, id: StandaloneSessionId) -> PathBuf { + fn worker_dir(&self, id: WorkerId) -> PathBuf { self.root.join(id.to_string()) } } #[derive(Debug)] -pub struct StandaloneSessionAllocation { - id: StandaloneSessionId, +pub struct StandaloneWorkerAllocation { + worker_id: WorkerId, cwd: StandaloneCwdIdentity, - lease: StandaloneSessionLease, + lease: StandaloneWorkerLease, } -impl StandaloneSessionAllocation { +impl StandaloneWorkerAllocation { #[must_use] - pub fn id(&self) -> StandaloneSessionId { - self.id + pub fn worker_id(&self) -> WorkerId { + self.worker_id } #[must_use] @@ -492,19 +455,19 @@ impl StandaloneSessionAllocation { &self.cwd } - pub fn into_lease(self) -> StandaloneSessionLease { + pub fn into_lease(self) -> StandaloneWorkerLease { self.lease } } #[derive(Debug)] -pub struct StandaloneSessionLease { +pub struct StandaloneWorkerLease { path: PathBuf, lease_id: Uuid, released: bool, } -impl StandaloneSessionLease { +impl StandaloneWorkerLease { pub fn release(mut self) -> Result<(), StandaloneStoreError> { self.release_inner() } @@ -534,7 +497,7 @@ impl StandaloneSessionLease { } } -impl Drop for StandaloneSessionLease { +impl Drop for StandaloneWorkerLease { fn drop(&mut self) { let _ = self.release_inner(); } @@ -624,7 +587,7 @@ fn classify_lease_liveness( } } -fn read_lease(path: &Path, id: StandaloneSessionId) -> Result { +fn read_lease(path: &Path, id: WorkerId) -> Result { let bytes = fs::read(path).map_err(StandaloneStoreError::Io)?; serde_json::from_slice(&bytes) .map_err(|source| StandaloneStoreError::CorruptLease { id, source }) @@ -692,47 +655,47 @@ pub enum StandaloneStoreError { CwdUnavailable(#[source] io::Error), #[error("standalone cwd is not a directory")] CwdNotDirectory, - #[error("standalone cwd identity no longer matches the persisted session")] + #[error("standalone cwd identity no longer matches the persisted Worker")] CwdIdentityMismatch, - #[error("standalone session {0} was not found")] - SessionNotFound(StandaloneSessionId), - #[error("standalone session {0} has an incomplete metadata commit")] - IncompleteCommit(StandaloneSessionId), - #[error("standalone session {0} has invalid metadata")] - InvalidRecord(StandaloneSessionId), - #[error("standalone session {id} metadata is corrupt")] + #[error("standalone Worker {0} was not found")] + WorkerNotFound(WorkerId), + #[error("standalone Worker {0} has an incomplete metadata commit")] + IncompleteCommit(WorkerId), + #[error("standalone Worker {0} has invalid metadata")] + InvalidRecord(WorkerId), + #[error("standalone Worker {id} metadata is corrupt")] CorruptRecord { - id: StandaloneSessionId, + id: WorkerId, #[source] source: serde_json::Error, }, - #[error("standalone session {id} lease is corrupt")] + #[error("standalone Worker {id} lease is corrupt")] CorruptLease { - id: StandaloneSessionId, + id: WorkerId, #[source] source: serde_json::Error, }, - #[error("standalone session {id} uses schema {found}, newer than supported schema {supported}")] + #[error("standalone Worker {id} uses schema {found}, newer than supported schema {supported}")] NewerSchema { - id: StandaloneSessionId, + id: WorkerId, found: u32, supported: u32, }, - #[error("standalone session {0} is already active")] - SessionLeased(StandaloneSessionId), - #[error("standalone session {0} lease liveness cannot be proven; recovery is rejected")] - LeaseLivenessUnknown(StandaloneSessionId), - #[error("standalone session {0} has a stale lease; explicit recovery is required")] - StaleLease(StandaloneSessionId), - #[error("standalone session lease ownership changed")] + #[error("standalone Worker {0} is already active")] + WorkerLeased(WorkerId), + #[error("standalone Worker {0} lease liveness cannot be proven; recovery is rejected")] + LeaseLivenessUnknown(WorkerId), + #[error("standalone Worker {0} has a stale lease; explicit recovery is required")] + StaleLease(WorkerId), + #[error("standalone Worker lease ownership changed")] LeaseOwnershipLost, - #[error("standalone session {0} must be stopped before deletion")] - DeleteActive(StandaloneSessionId), + #[error("standalone Worker {0} must be stopped before deletion")] + DeleteActive(WorkerId), #[error( - "standalone session {id} metadata revision changed (expected {expected}, found {found})" + "standalone Worker {id} metadata revision changed (expected {expected}, found {found})" )] RevisionConflict { - id: StandaloneSessionId, + id: WorkerId, expected: u64, found: u64, }, diff --git a/crates/standalone/tests/host.rs b/crates/standalone/tests/host.rs index 8198f3ac..755f5718 100644 --- a/crates/standalone/tests/host.rs +++ b/crates/standalone/tests/host.rs @@ -14,7 +14,7 @@ use futures::{Stream, stream}; use protocol::{Event, Method}; use standalone::{ StaleLeasePolicy, StandaloneHost, StandaloneLaunchConfig, StandaloneListScope, - StandaloneSessionStatus, StandaloneSessionStore, StandaloneStartupError, StandaloneStoreError, + StandaloneStartupError, StandaloneStoreError, StandaloneWorkerStatus, StandaloneWorkerStore, }; use uuid::Uuid; @@ -90,6 +90,12 @@ async fn in_process_host_runs_text_and_read_tool_then_shuts_down() { let host = StandaloneHost::start_with_model_client(launch, client) .await .expect("start in-process host"); + assert_eq!(host.record().worker_name, worker_name); + assert_eq!(host.record().manifest.worker.name, worker_name); + assert_eq!( + host.record().storage_key, + format!("standalone-{}", host.worker_id()) + ); let mut protocol_client = host.connect(); protocol_client @@ -238,7 +244,7 @@ type TestResult = Result<(), Box>; async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope() -> TestResult { let temp = tempfile::tempdir()?; let cwd = temp.path().join("project"); - let state_dir = temp.path().join("client").join("standalone-sessions"); + let state_dir = temp.path().join("client").join("standalone-workers"); std::fs::create_dir_all(&cwd)?; let launch = StandaloneLaunchConfig::new( &cwd, @@ -268,7 +274,7 @@ async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope( ], ]); let host = StandaloneHost::start_with_model_client(launch, first_client).await?; - let session_id = host.session_id(); + let worker_id = host.worker_id(); let mut protocol_client = host.connect(); protocol_client .send(&Method::run_text("first request")) @@ -283,11 +289,11 @@ async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope( wait_for_run_end(&mut protocol_client).await?; host.shutdown().await?; - let store = StandaloneSessionStore::open(&state_dir)?; + let store = StandaloneWorkerStore::open(&state_dir)?; let current = store.list(&cwd, StandaloneListScope::CurrentCwd, 100)?; assert_eq!(current.len(), 1); - assert_eq!(current[0].session_id, session_id); - assert_eq!(current[0].status, StandaloneSessionStatus::Stopped); + assert_eq!(current[0].worker_id, worker_id); + assert_eq!(current[0].status, StandaloneWorkerStatus::Stopped); let other_cwd = temp.path().join("other"); std::fs::create_dir(&other_cwd)?; assert!( @@ -307,8 +313,13 @@ async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope( ]]); let second_inspection = second_client.clone(); let host = - StandaloneHost::restore_with_model_client(state_dir.clone(), session_id, second_client) + StandaloneHost::restore_with_model_client(state_dir.clone(), worker_id, second_client) .await?; + assert_eq!( + host.record().worker_name, + "display-name-is-not-session-identity" + ); + assert_eq!(host.record().storage_key, format!("standalone-{worker_id}")); let mut protocol_client = host.connect(); let snapshot = format!( "{:?}", @@ -338,11 +349,11 @@ async fn standalone_restore_preserves_history_tasks_notifications_and_cwd_scope( assert!(projected.contains("persisted task"), "{projected}"); host.shutdown().await?; - store.delete(session_id)?; + store.delete(worker_id)?; assert!(cwd.exists(), "deleting session state must not mutate cwd"); assert!(matches!( - store.load(session_id), - Err(StandaloneStoreError::SessionNotFound(_)) + store.load(worker_id), + Err(StandaloneStoreError::WorkerNotFound(_)) )); Ok(()) } @@ -363,28 +374,25 @@ async fn standalone_restore_rejects_concurrent_lease_and_missing_cwd() -> TestRe .resolve()?; let host = StandaloneHost::start_with_model_client(launch, ScriptedClient::new(Vec::new())).await?; - let session_id = host.session_id(); - let store = StandaloneSessionStore::open(&state_dir)?; + let worker_id = host.worker_id(); + let store = StandaloneWorkerStore::open(&state_dir)?; assert!(matches!( - store.acquire_lease(session_id, StaleLeasePolicy::Recover), - Err(StandaloneStoreError::SessionLeased(id)) if id == session_id + store.acquire_lease(worker_id, StaleLeasePolicy::Recover), + Err(StandaloneStoreError::WorkerLeased(id)) if id == worker_id )); let restore = StandaloneHost::restore_with_model_client( state_dir.clone(), - session_id, + worker_id, ScriptedClient::new(Vec::new()), ) .await; - assert!(matches!( - restore, - Err(StandaloneStartupError::SessionActive) - )); + assert!(matches!(restore, Err(StandaloneStartupError::WorkerActive))); host.shutdown().await?; std::fs::rename(&cwd, &moved)?; let restore = StandaloneHost::restore_with_model_client( state_dir, - session_id, + worker_id, ScriptedClient::new(Vec::new()), ) .await; @@ -421,11 +429,11 @@ async fn standalone_restore_recovers_only_a_proven_stale_lease() -> TestResult { }); let host = StandaloneHost::start_with_model_client(launch, ScriptedClient::new(Vec::new())).await?; - let session_id = host.session_id(); + let worker_id = host.worker_id(); host.shutdown().await?; - let store = StandaloneSessionStore::open(&state_dir)?; + let store = StandaloneWorkerStore::open(&state_dir)?; assert!(matches!( - store.load(session_id)?.manifest.profile, + store.load(worker_id)?.manifest.profile, Some(manifest::ProfileManifestSnapshot { source: manifest::ProfileSource::Registry { source: manifest::ProfileRegistrySource::User, @@ -434,9 +442,9 @@ async fn standalone_restore_recovers_only_a_proven_stale_lease() -> TestResult { .. }) )); - let session_dir = state_dir.join(session_id.to_string()); + let worker_dir = state_dir.join(worker_id.to_string()); std::fs::write( - session_dir.join("lease.json"), + worker_dir.join("lease.json"), serde_json::to_vec(&serde_json::json!({ "lease_id": uuid::Uuid::now_v7(), "pid": u32::MAX, @@ -447,7 +455,7 @@ async fn standalone_restore_recovers_only_a_proven_stale_lease() -> TestResult { let host = StandaloneHost::restore_with_model_client( state_dir, - session_id, + worker_id, ScriptedClient::new(Vec::new()), ) .await?; @@ -468,11 +476,11 @@ async fn standalone_restore_rejects_lease_with_missing_start_marker() -> TestRes .resolve()?; let host = StandaloneHost::start_with_model_client(launch, ScriptedClient::new(Vec::new())).await?; - let session_id = host.session_id(); + let worker_id = host.worker_id(); host.shutdown().await?; - let session_dir = state_dir.join(session_id.to_string()); + let worker_dir = state_dir.join(worker_id.to_string()); std::fs::write( - session_dir.join("lease.json"), + worker_dir.join("lease.json"), serde_json::to_vec(&serde_json::json!({ "lease_id": uuid::Uuid::now_v7(), "pid": std::process::id(), @@ -480,14 +488,14 @@ async fn standalone_restore_rejects_lease_with_missing_start_marker() -> TestRes }))?, )?; - let store = StandaloneSessionStore::open(&state_dir)?; + let store = StandaloneWorkerStore::open(&state_dir)?; assert!(matches!( - store.acquire_lease(session_id, StaleLeasePolicy::Recover), - Err(StandaloneStoreError::LeaseLivenessUnknown(id)) if id == session_id + store.acquire_lease(worker_id, StaleLeasePolicy::Recover), + Err(StandaloneStoreError::LeaseLivenessUnknown(id)) if id == worker_id )); let restore = StandaloneHost::restore_with_model_client( state_dir, - session_id, + worker_id, ScriptedClient::new(Vec::new()), ) .await; @@ -511,23 +519,23 @@ async fn standalone_metadata_fails_closed_on_incomplete_or_newer_records() -> Te .resolve()?; let host = StandaloneHost::start_with_model_client(launch, ScriptedClient::new(Vec::new())).await?; - let session_id = host.session_id(); + let worker_id = host.worker_id(); host.shutdown().await?; - let store = StandaloneSessionStore::open(&state_dir)?; - let session_dir = state_dir.join(session_id.to_string()); - std::fs::write(session_dir.join("commit.pending"), b"interrupted\n")?; + let store = StandaloneWorkerStore::open(&state_dir)?; + let worker_dir = state_dir.join(worker_id.to_string()); + std::fs::write(worker_dir.join("commit.pending"), b"interrupted\n")?; assert!(matches!( - store.load(session_id), - Err(StandaloneStoreError::IncompleteCommit(id)) if id == session_id + store.load(worker_id), + Err(StandaloneStoreError::IncompleteCommit(id)) if id == worker_id )); - std::fs::remove_file(session_dir.join("commit.pending"))?; - let record_path = session_dir.join("record.json"); + std::fs::remove_file(worker_dir.join("commit.pending"))?; + let record_path = worker_dir.join("record.json"); let mut record: serde_json::Value = serde_json::from_slice(&std::fs::read(&record_path)?)?; record["schema_version"] = serde_json::json!(u32::MAX); std::fs::write(&record_path, serde_json::to_vec_pretty(&record)?)?; assert!(matches!( - store.load(session_id), - Err(StandaloneStoreError::NewerSchema { id, .. }) if id == session_id + store.load(worker_id), + Err(StandaloneStoreError::NewerSchema { id, .. }) if id == worker_id )); Ok(()) } diff --git a/crates/tui/src/console/mod.rs b/crates/tui/src/console/mod.rs index ec0454ff..6a63dff0 100644 --- a/crates/tui/src/console/mod.rs +++ b/crates/tui/src/console/mod.rs @@ -25,9 +25,7 @@ use tokio::sync::mpsc; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD}; use client::transport::Socket; -use client::{ - BackendRuntimeTarget, Client, StandaloneSessionResumeIntent, connect_backend_runtime, -}; +use client::{BackendRuntimeTarget, Client, StandaloneWorkerResumeIntent, connect_backend_runtime}; use crate::app::{ActionbarNoticeLevel, ActionbarNoticeSource, App}; use crate::composer_keys::{ComposerEditAction, composer_edit_action}; @@ -193,18 +191,18 @@ pub(crate) async fn run_standalone( } pub(crate) async fn run_standalone_restore( - intent: StandaloneSessionResumeIntent, + intent: StandaloneWorkerResumeIntent, ) -> Result<(), Box> { - let session_id = intent.session_id.parse().map_err(|error| { + let worker_id = intent.worker_id.parse().map_err(|error| { io::Error::new( io::ErrorKind::InvalidInput, - format!("Invalid standalone session ID: {error}"), + format!("Invalid standalone Worker ID: {error}"), ) })?; - let host = StandaloneHost::restore(intent.state_dir, session_id) + let host = StandaloneHost::restore(intent.state_dir, worker_id) .await .map_err(|error| io::Error::other(format!("Standalone restore failed: {error}")))?; - let worker_label = format!("standalone-{}", session_id.short()); + let worker_label = host.record().worker_name.clone(); let history_root = host.record().cwd.canonical_path.clone(); run_standalone_host(host, worker_label, history_root).await } diff --git a/crates/tui/src/lib.rs b/crates/tui/src/lib.rs index 724b9fc0..71eacd14 100644 --- a/crates/tui/src/lib.rs +++ b/crates/tui/src/lib.rs @@ -46,8 +46,8 @@ pub enum LaunchMode { worker_name: Option, profile: Option, }, - /// Restore one client-owned standalone session. The current cwd is the default scope; - /// `include_all` opts into all standalone sessions under the same client data root. + /// Restore one client-owned standalone Worker. The current cwd is the default scope; + /// `include_all` opts into all standalone Workers under the same client data root. StandaloneResume { include_all: bool }, /// List Backend Workers and attach to the selected Worker. Workers { diff --git a/crates/tui/src/standalone_picker.rs b/crates/tui/src/standalone_picker.rs index 44581429..35ccbb5f 100644 --- a/crates/tui/src/standalone_picker.rs +++ b/crates/tui/src/standalone_picker.rs @@ -1,7 +1,7 @@ use std::io; use std::time::Duration; -use client::{StandaloneSessionListIntent, StandaloneSessionResumeIntent, Target}; +use client::{StandaloneWorkerListIntent, StandaloneWorkerResumeIntent, Target}; use crossterm::event::{self, Event as TermEvent, KeyCode, KeyEventKind, KeyModifiers}; use ratatui::Terminal; use ratatui::backend::CrosstermBackend; @@ -9,7 +9,7 @@ use ratatui::layout::{Constraint, Layout}; use ratatui::prelude::{Color, Line, Modifier, Span, Style}; use ratatui::widgets::Paragraph; use ratatui::{TerminalOptions, Viewport}; -use standalone::{StandaloneListScope, StandaloneSessionRecord, StandaloneSessionStore}; +use standalone::{StandaloneListScope, StandaloneWorkerRecord, StandaloneWorkerStore}; use thiserror::Error; const LIMIT: usize = 100; @@ -17,28 +17,28 @@ const LIMIT: usize = 100; pub(crate) fn pick( target: &dyn Target, include_all: bool, -) -> Result, StandalonePickerError> { +) -> Result, StandalonePickerError> { let intent = target - .standalone_session_list(include_all) + .standalone_worker_list(include_all) .map_err(StandalonePickerError::Target)?; let records = load_records(&intent)?; if records.is_empty() { - return Err(StandalonePickerError::NoSessions { include_all }); + return Err(StandalonePickerError::NoWorkers { include_all }); } let selected = run_picker(records)?; selected .map(|record| { target - .standalone_session_resume(record.session_id.to_string()) + .standalone_worker_resume(record.worker_id.to_string()) .map_err(StandalonePickerError::Target) }) .transpose() } fn load_records( - intent: &StandaloneSessionListIntent, -) -> Result, StandalonePickerError> { - let store = StandaloneSessionStore::open(&intent.state_dir) + intent: &StandaloneWorkerListIntent, +) -> Result, StandalonePickerError> { + let store = StandaloneWorkerStore::open(&intent.state_dir) .map_err(StandalonePickerError::StateStore)?; store .list( @@ -54,8 +54,8 @@ fn load_records( } fn run_picker( - records: Vec, -) -> Result, StandalonePickerError> { + records: Vec, +) -> Result, StandalonePickerError> { let height = u16::try_from(records.len().saturating_add(3).min(20)).unwrap_or(20); let mut terminal = Terminal::with_options( CrosstermBackend::new(io::stdout()), @@ -94,14 +94,14 @@ fn run_picker( } } -fn draw(frame: &mut ratatui::Frame<'_>, records: &[StandaloneSessionRecord], selected: usize) { +fn draw(frame: &mut ratatui::Frame<'_>, records: &[StandaloneWorkerRecord], selected: usize) { let mut constraints = vec![Constraint::Length(1)]; constraints.extend(records.iter().map(|_| Constraint::Length(1))); constraints.push(Constraint::Length(1)); let rows = Layout::vertical(constraints).split(frame.area()); frame.render_widget( Paragraph::new(Line::from(Span::styled( - "resume standalone session", + "resume standalone Worker", Style::default().add_modifier(Modifier::BOLD), ))), rows[0], @@ -120,7 +120,10 @@ fn draw(frame: &mut ratatui::Frame<'_>, records: &[StandaloneSessionRecord], sel frame.render_widget( Paragraph::new(Line::from(vec![ Span::raw(marker), - Span::styled(record.session_id.short(), style), + Span::styled( + format!("{} ({})", record.worker_name, record.worker_id.short()), + style, + ), Span::raw(format!( " [{:?}] updated:{} {}", record.status, record.updated_at_unix_ms, cwd @@ -139,13 +142,13 @@ fn draw(frame: &mut ratatui::Frame<'_>, records: &[StandaloneSessionRecord], sel pub(crate) enum StandalonePickerError { #[error("standalone target error: {0}")] Target(#[source] client::TargetError), - #[error("standalone session state is unavailable: {0}")] + #[error("standalone Worker state is unavailable: {0}")] StateStore(#[source] standalone::StandaloneStoreError), #[error( - "no standalone sessions found for this cwd; use `yoi --local --resume --all` to include all cwd identities" + "no standalone Workers found for this cwd; use `yoi --local --resume --all` to include all cwd identities" )] - NoSessions { include_all: bool }, - #[error("standalone session picker I/O failed: {0}")] + NoWorkers { include_all: bool }, + #[error("standalone Worker picker I/O failed: {0}")] Io(#[source] io::Error), } diff --git a/crates/worker-runtime/src/identity.rs b/crates/worker-runtime/src/identity.rs index c16fa7c3..3b1b85dd 100644 --- a/crates/worker-runtime/src/identity.rs +++ b/crates/worker-runtime/src/identity.rs @@ -1,105 +1,9 @@ -use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; +use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; -use std::{fmt, str::FromStr}; -use uuid::{Uuid, Version}; +pub use protocol::{WorkerId, WorkerIdParseError}; pub use workdir::workspace::RuntimeWorkerRef; -/// Stable Workspace-owned Worker identity. -/// -/// Runtime placement is deliberately not part of this value. New identities are -/// allocated by Workspace authority before a Runtime create request is sent. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] -pub struct WorkerId(Uuid); - -impl WorkerId { - pub fn now_v7() -> Self { - Self(Uuid::now_v7()) - } - - /// Converts a legacy Runtime-local numeric id into a syntactically valid - /// migration-only UUIDv7 value. New Worker allocation must use `now_v7`. - pub fn from_legacy_u64(value: u64) -> Self { - let mut bytes = [0_u8; 16]; - bytes[8..].copy_from_slice(&value.to_be_bytes()); - bytes[6] = 0x70; - bytes[8] = (bytes[8] & 0x3f) | 0x80; - Self(Uuid::from_bytes(bytes)) - } - - pub fn from_legacy_binding(workspace_id: &str, runtime_id: &str, value: u64) -> Self { - let mut hasher = Sha256::new(); - hasher.update(b"yoi.workspace-worker-id.v1\0"); - hasher.update(workspace_id.as_bytes()); - hasher.update([0]); - hasher.update(runtime_id.as_bytes()); - hasher.update([0]); - hasher.update(value.to_be_bytes()); - let digest = hasher.finalize(); - let mut bytes = [0_u8; 16]; - bytes.copy_from_slice(&digest[..16]); - // Migrated ids sort before normally allocated UUIDv7 values while retaining - // deterministic collision-resistant payload bits. - bytes[..6].fill(0); - bytes[6] = (bytes[6] & 0x0f) | 0x70; - bytes[8] = (bytes[8] & 0x3f) | 0x80; - Self(Uuid::from_bytes(bytes)) - } - - pub fn parse(value: &str) -> Option { - let value = Uuid::parse_str(value).ok()?; - (value.get_version() == Some(Version::SortRand)).then_some(Self(value)) - } - - pub const fn as_uuid(self) -> Uuid { - self.0 - } -} - -impl fmt::Display for WorkerId { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - self.0.fmt(formatter) - } -} - -impl FromStr for WorkerId { - type Err = WorkerIdParseError; - - fn from_str(value: &str) -> Result { - Self::parse(value).ok_or(WorkerIdParseError) - } -} - -impl Serialize for WorkerId { - fn serialize(&self, serializer: S) -> Result - where - S: Serializer, - { - serializer.serialize_str(&self.to_string()) - } -} - -impl<'de> Deserialize<'de> for WorkerId { - fn deserialize(deserializer: D) -> Result - where - D: Deserializer<'de>, - { - let value = String::deserialize(deserializer)?; - Self::parse(&value).ok_or_else(|| de::Error::custom("Worker id must be a UUIDv7")) - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct WorkerIdParseError; - -impl fmt::Display for WorkerIdParseError { - fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter.write_str("Worker id must be a UUIDv7") - } -} - -impl std::error::Error for WorkerIdParseError {} - #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct LegacyWorkerIdentityMapping { pub workspace_id: String, @@ -140,7 +44,7 @@ pub fn legacy_worker_identity_mapping_digest(mappings: &[LegacyWorkerIdentityMap } /// Runtime-local authority reference for Worker operations. The contained id is -/// nevertheless the Workspace-owned stable identity; the Runtime does not mint it. +/// nevertheless the stable Worker identity; the Runtime does not mint it. #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] pub struct WorkerRef { pub worker_id: WorkerId, @@ -164,14 +68,6 @@ impl TryFrom<&RuntimeWorkerRef> for WorkerRef { mod tests { use super::*; - #[test] - fn worker_id_accepts_only_uuid_v7() { - let worker_id = WorkerId::now_v7(); - assert_eq!(WorkerId::parse(&worker_id.to_string()), Some(worker_id)); - assert!(WorkerId::parse("30").is_none()); - assert!(WorkerId::parse(&Uuid::nil().to_string()).is_none()); - } - #[test] fn runtime_worker_ref_preserves_stable_worker_identity() { let worker_id = WorkerId::now_v7(); diff --git a/crates/yoi/src/cli_connection.rs b/crates/yoi/src/cli_connection.rs index a9670187..7a669ab7 100644 --- a/crates/yoi/src/cli_connection.rs +++ b/crates/yoi/src/cli_connection.rs @@ -118,7 +118,7 @@ fn standalone_target() -> Result, ParseError> { })? .join("client") .join("standalone") - .join("sessions"); + .join("workers"); Ok(Box::new(StandaloneTarget::new(state_dir))) } diff --git a/crates/yoi/src/main.rs b/crates/yoi/src/main.rs index 1c97da5f..8700b670 100644 --- a/crates/yoi/src/main.rs +++ b/crates/yoi/src/main.rs @@ -798,7 +798,7 @@ fn parse_console_options( } if session.is_some() { return Err(ParseError( - "--local does not accept legacy --session; use --local --resume for Standalone session restore" + "--local does not accept legacy --session; use --local --resume for Standalone Worker restore" .to_string(), )); } @@ -834,7 +834,7 @@ fn parse_console_options( if target.kind() == TargetKind::Standalone && (session.is_some() || socket_override.is_some()) { return Err(ParseError( - "Standalone does not accept legacy Worker session or socket selectors; use --resume for the standalone session store" + "Standalone does not accept legacy Worker session or socket selectors; use --resume for the standalone Worker store" .to_string(), )); } @@ -951,7 +951,7 @@ fn parse_workers_args( )?; if target.kind() != TargetKind::Backend { return Err(ParseError( - "yoi workers requires a Backend connection target; use yoi --local --resume for Standalone sessions" + "yoi workers requires a Backend connection target; use yoi --local --resume for Standalone Workers" .to_string(), )); } @@ -1750,8 +1750,8 @@ Usage: Target selection: --local Use the client-owned one-process Standalone host - --resume With --local, restore from the Standalone session store - --all With Standalone restore, include sessions from every cwd identity + --resume With --local, restore from the Standalone Worker store + --all With Standalone restore, include Workers from every cwd identity --backend Use a Workspace Backend explicitly --workspace-id Scope Backend routes to a Workspace id @@ -1761,7 +1761,7 @@ Target selection: Connection-aware commands: yoi Standalone: new Console. Backend: Worker picker. - yoi resume Standalone session picker or stopped Backend Worker picker. + yoi resume Standalone Worker picker or stopped Backend Worker picker. yoi workers Backend Workspace Worker picker. yoi panel Backend Workspace dashboard. @@ -1800,7 +1800,7 @@ Usage: yoi --backend [--workspace-id ] workers [-r|--stopped] [--workspace ] [--runtime-id ] Authority: - Lists Workers from the selected Backend Workspace. Standalone sessions are restored with + Lists Workers from the selected Backend Workspace. Standalone Workers are restored with `yoi --local --resume` and are not part of the Workspace Worker catalog. Options: @@ -1822,13 +1822,13 @@ Usage: yoi [TARGET] resume [--workspace |--all] [--runtime-id ] Target options: - --local Restore from the client-owned Standalone session store + --local Restore from the client-owned Standalone Worker store --backend Restore a stopped Backend Workspace Worker --workspace-id Scope Backend routes to a Workspace id Options: - --workspace Scope Standalone sessions to this cwd identity (defaults to cwd) - --all Include Standalone sessions from every cwd identity + --workspace Scope Standalone Workers to this cwd identity (defaults to cwd) + --all Include Standalone Workers from every cwd identity --runtime-id Restrict the Backend stopped-Worker picker to a Runtime id -h, --help Print help "#; @@ -2222,8 +2222,8 @@ backend = "shared" mode, LaunchMode::StandaloneResume { include_all: false } )); - let intent = target.standalone_session_list(false).unwrap(); - assert!(intent.state_dir.ends_with("client/standalone/sessions")); + let intent = target.standalone_worker_list(false).unwrap(); + assert!(intent.state_dir.ends_with("client/standalone/workers")); assert!(!intent.include_all); let mode = parse_args_from(["--local", "--resume", "--all"]).unwrap(); @@ -2264,7 +2264,7 @@ backend = "shared" let err = parse_args_slice_with_connection_resolver(&session_args, &resolver).unwrap_err(); assert_eq!( err.0, - "--local does not accept legacy --session; use --local --resume for Standalone session restore" + "--local does not accept legacy --session; use --local --resume for Standalone Worker restore" ); let socket_args = [ @@ -2903,12 +2903,12 @@ backend = "shared" } #[test] - fn parse_resume_help_uses_standalone_session_store_terminology() { + fn parse_resume_help_uses_standalone_worker_store_terminology() { match parse_args_from(["resume", "--help"]).unwrap() { Mode::ResumeHelp => {} _ => panic!("expected ResumeHelp mode"), } - assert!(RESUME_HELP.contains("Standalone session store")); + assert!(RESUME_HELP.contains("Standalone Worker store")); assert!(RESUME_HELP.contains("Backend stopped-Worker picker")); assert!(!RESUME_HELP.contains("local Worker records")); assert!(!RESUME_HELP.contains("local workspace"));