diff --git a/crates/workdir/src/workspace.rs b/crates/workdir/src/workspace.rs index d8e4aa28..1ea2cbb1 100644 --- a/crates/workdir/src/workspace.rs +++ b/crates/workdir/src/workspace.rs @@ -126,7 +126,7 @@ pub struct WorkingDirectoryCurrentObservation { #[serde(default, skip_serializing_if = "Option::is_none")] pub cleanliness: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub primary_worker_id: Option, + pub primary_worker_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub occupied_by: Option, } @@ -151,7 +151,7 @@ pub struct WorkingDirectorySummary { #[serde(default, skip_serializing_if = "Option::is_none")] pub cleanliness: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub primary_worker_id: Option, + pub primary_worker_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub occupied_by: Option, } @@ -177,7 +177,7 @@ impl WorkingDirectorySummary { current_ref: self.current_ref.clone(), status: self.status.clone(), cleanliness: self.cleanliness.clone(), - primary_worker_id: self.primary_worker_id, + primary_worker_id: self.primary_worker_id.clone(), occupied_by: self.occupied_by.clone(), } } diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index e9263a76..703cae62 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -148,10 +148,10 @@ impl std::fmt::Debug for WorkspaceApiRef { /// summarized without exposing raw host paths. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct CreateWorkerRequest { - #[serde(default, skip_serializing_if = "Option::is_none")] - pub idempotency_key: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub idempotency_fingerprint: Option, + /// Workspace-owned stable identity reserved before this request reaches a Runtime. + pub worker_id: WorkerId, + /// Canonical create-intent fingerprint bound to `worker_id` for retry recovery. + pub create_fingerprint: String, pub profile: ProfileSelector, #[serde(default, skip_serializing_if = "Option::is_none")] pub display_name: Option, diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 61b75a10..fc676607 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -11,7 +11,7 @@ use std::io::{BufReader, Write}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; -const SCHEMA_VERSION: u32 = 1; +const SCHEMA_VERSION: u32 = 2; const RUNTIME_FILE: &str = "runtime.json"; const WORKERS_DIR: &str = "workers"; const WORKER_FILE: &str = "worker.json"; @@ -24,6 +24,7 @@ static NEXT_TMP_SEQUENCE: AtomicU64 = AtomicU64::new(1); pub struct FsRuntimeStoreOptions { /// Root directory containing this Runtime's store data. pub root: PathBuf, + pub runtime_id: String, pub display_name: Option, } @@ -31,14 +32,19 @@ impl FsRuntimeStoreOptions { pub fn new(root: impl Into) -> Self { Self { root: root.into(), + runtime_id: "local".to_string(), display_name: None, } } + pub fn with_runtime_id(mut self, runtime_id: impl Into) -> Self { + self.runtime_id = runtime_id.into(); + self + } } /// Filesystem persistence boundary for one Worker Runtime state. /// -/// Authority is Runtime-local typed Worker identity. Legacy pod paths, socket +/// Authority is the Workspace-owned typed Worker identity. Legacy pod paths, socket /// paths, and session paths are deliberately not part of the layout or lookup API. #[derive(Clone, Debug, PartialEq, Eq)] pub struct FsRuntimeStore { @@ -54,7 +60,10 @@ impl FsRuntimeStore { &self.root } - pub(crate) fn open_or_create(root: PathBuf) -> Result { + pub(crate) fn open_or_create( + root: PathBuf, + runtime_id: &str, + ) -> Result { let existed = root.exists(); if existed && !root.is_dir() { return Err(RuntimeError::StoreCorrupt { @@ -82,6 +91,9 @@ impl FsRuntimeStore { } } + if existed { + migrate_v1_worker_identity(&root, runtime_id)?; + } let store = Self { root }; let state = if existed { Some(store.load_runtime_state()?) @@ -241,7 +253,6 @@ pub(crate) struct OpenedFsRuntimeStore { pub(crate) struct PersistedRuntimeState { pub(crate) display_name: Option, pub(crate) status: RuntimeStatus, - pub(crate) next_worker_sequence: u64, pub(crate) next_diagnostic_id: u64, pub(crate) workers: BTreeMap, pub(crate) workspace_owners: BTreeMap, @@ -259,13 +270,151 @@ pub(crate) struct PersistedWorkerRecord { pub(crate) working_directory: Option, } +fn runtime_io_error(operation: &'static str, path: &Path, source: std::io::Error) -> RuntimeError { + RuntimeError::StoreIo { + operation, + path: path.to_path_buf(), + source, + } +} + +fn runtime_store_corrupt(path: &Path, message: String) -> RuntimeError { + RuntimeError::StoreCorrupt { + operation: "migrate Worker identity", + path: path.to_path_buf(), + message, + } +} + +fn migrate_v1_worker_identity(root: &Path, runtime_id: &str) -> Result<(), RuntimeError> { + let runtime_path = root.join(RUNTIME_FILE); + let bytes = + fs::read(&runtime_path).map_err(|error| runtime_io_error("read", &runtime_path, error))?; + let mut document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| { + runtime_store_corrupt( + &runtime_path, + format!("decode Runtime state {}: {error}", runtime_path.display()), + ) + })?; + let schema_version = document + .get("schema_version") + .and_then(serde_json::Value::as_u64) + .ok_or_else(|| { + runtime_store_corrupt( + &runtime_path, + "Runtime state is missing schema_version".to_string(), + ) + })?; + if schema_version == u64::from(SCHEMA_VERSION) { + return Ok(()); + } + if schema_version != 1 { + return Err(runtime_store_corrupt( + &runtime_path, + format!( + "unsupported Runtime store schema version {schema_version}; expected 1 or {SCHEMA_VERSION}" + ), + )); + } + + let legacy_ids = document + .get("workers") + .and_then(serde_json::Value::as_array) + .ok_or_else(|| { + runtime_store_corrupt( + &runtime_path, + "Runtime state workers must be an array".to_string(), + ) + })? + .iter() + .map(|value| { + value.as_u64().ok_or_else(|| { + runtime_store_corrupt( + &runtime_path, + "legacy Worker id must be unsigned".to_string(), + ) + }) + }) + .collect::, _>>()?; + + let mut migrated_ids = Vec::with_capacity(legacy_ids.len()); + for legacy_id in legacy_ids { + let legacy_dir = root.join("workers").join(legacy_id.to_string()); + let legacy_snapshot_path = legacy_dir.join(WORKER_FILE); + let bytes = fs::read(&legacy_snapshot_path) + .map_err(|error| runtime_io_error("read", &legacy_snapshot_path, error))?; + let mut snapshot: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| { + runtime_store_corrupt( + &runtime_path, + format!( + "decode Worker snapshot {}: {error}", + legacy_snapshot_path.display() + ), + ) + })?; + let workspace_id = snapshot + .get("workspace_id") + .and_then(serde_json::Value::as_str) + .unwrap_or("local") + .to_string(); + let worker_id = WorkerId::from_legacy_binding(&workspace_id, runtime_id, legacy_id); + let worker_id_text = worker_id.to_string(); + snapshot["schema_version"] = serde_json::Value::from(SCHEMA_VERSION); + snapshot["worker_id"] = serde_json::Value::String(worker_id_text.clone()); + snapshot["worker_ref"]["worker_id"] = serde_json::Value::String(worker_id_text.clone()); + let request = snapshot + .get_mut("request") + .and_then(serde_json::Value::as_object_mut) + .ok_or_else(|| { + runtime_store_corrupt( + &runtime_path, + "Worker snapshot request must be an object".to_string(), + ) + })?; + let fingerprint = request + .remove("idempotency_fingerprint") + .and_then(|value| value.as_str().map(ToOwned::to_owned)) + .unwrap_or_else(|| format!("legacy:{workspace_id}:{runtime_id}:{legacy_id}")); + request.remove("idempotency_key"); + request.insert( + "worker_id".to_string(), + serde_json::Value::String(worker_id_text.clone()), + ); + request.insert( + "create_fingerprint".to_string(), + serde_json::Value::String(fingerprint), + ); + + let migrated_dir = root.join("workers").join(&worker_id_text); + fs::rename(&legacy_dir, &migrated_dir) + .map_err(|error| runtime_io_error("rename", &legacy_dir, error))?; + let migrated_snapshot_path = migrated_dir.join(WORKER_FILE); + atomic_write_json( + &migrated_snapshot_path, + &snapshot, + "migrate Worker identity", + )?; + migrated_ids.push(serde_json::Value::String(worker_id_text)); + } + + document["schema_version"] = serde_json::Value::from(SCHEMA_VERSION); + document["workers"] = serde_json::Value::Array(migrated_ids); + if let Some(object) = document.as_object_mut() { + object.remove("next_worker_sequence"); + } + atomic_write_json( + &runtime_path, + &document, + "migrate Runtime Worker identities", + ) +} + #[derive(Clone, Debug, Serialize, Deserialize)] struct RuntimeSnapshot { schema_version: u32, display_name: Option, backend: RuntimeBackendKind, status: RuntimeStatus, - next_worker_sequence: u64, next_diagnostic_id: u64, #[serde(default)] config_bundles: BTreeMap, @@ -297,7 +446,6 @@ impl RuntimeSnapshot { display_name: state.display_name.clone(), backend: RuntimeBackendKind::FsStore, status: state.status, - next_worker_sequence: state.next_worker_sequence, next_diagnostic_id: state.next_diagnostic_id, config_bundles: BTreeMap::new(), workspace_owners: state.workspace_owners.clone(), @@ -333,7 +481,6 @@ impl RuntimeSnapshot { PersistedRuntimeState { display_name: self.display_name, status: self.status, - next_worker_sequence: self.next_worker_sequence, next_diagnostic_id: self.next_diagnostic_id, workers, workspace_owners: self.workspace_owners, diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index d04e9d42..0c0e0dcf 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2162,8 +2162,8 @@ mod tests { let profile = ProfileSelector::Builtin("builtin:coder".to_string()); let bundle = test_bundle(profile.clone()); CreateWorkerRequest { - idempotency_key: None, - idempotency_fingerprint: None, + worker_id: WorkerId::now_v7(), + create_fingerprint: "test-create".to_string(), profile, display_name: None, profile_source: crate::catalog::ProfileSourceArchiveSource::Http { @@ -2659,12 +2659,14 @@ mod tests { async fn runtime_errors_use_typed_rest_error_shape() { let token = "local-token"; let app = runtime_http_router(Runtime::new_memory(), token.to_string()); - let response = authed_empty_request(app, Method::GET, "/v1/workers/999", token).await; + let missing = crate::identity::WorkerId::from_legacy_u64(999); + let response = + authed_empty_request(app, Method::GET, &format!("/v1/workers/{missing}"), token).await; assert_eq!(response.status(), StatusCode::NOT_FOUND); let error: RuntimeHttpErrorResponse = read_json(response).await; assert_eq!(error.error.code, "worker_not_found"); - assert!(error.error.message.contains("999")); + assert!(error.error.message.contains(&missing.to_string())); } #[tokio::test] @@ -2795,8 +2797,8 @@ mod ws_tests { fn ws_create_request() -> CreateWorkerRequest { let bundle = ws_test_bundle(ProfileSelector::Builtin("builtin:companion".to_string())); CreateWorkerRequest { - idempotency_key: None, - idempotency_fingerprint: None, + worker_id: WorkerId::now_v7(), + create_fingerprint: "test-create".to_string(), profile: ProfileSelector::Builtin("builtin:companion".to_string()), display_name: None, profile_source: crate::catalog::ProfileSourceArchiveSource::Http { diff --git a/crates/worker-runtime/src/identity.rs b/crates/worker-runtime/src/identity.rs index 65366995..78133811 100644 --- a/crates/worker-runtime/src/identity.rs +++ b/crates/worker-runtime/src/identity.rs @@ -1,50 +1,108 @@ -use serde::{Deserialize, Serialize}; -use std::fmt; +use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; +use std::{fmt, str::FromStr}; +use uuid::{Uuid, Version}; + pub use workdir::workspace::RuntimeWorkerRef; -/// Runtime-local Worker identity. -#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] -#[serde(transparent)] -pub struct WorkerId(u64); +/// 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 new(value: u64) -> Self { - Self(value) + 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 { + use sha2::{Digest, Sha256}; + + 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 { - value.parse::().ok().map(Self) + let value = Uuid::parse_str(value).ok()?; + (value.get_version() == Some(Version::SortRand)).then_some(Self(value)) } - pub(crate) fn generated(sequence: u64) -> Self { - Self(sequence) - } - - pub fn as_u64(&self) -> u64 { + pub const fn as_uuid(self) -> Uuid { self.0 } } impl fmt::Display for WorkerId { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.0.fmt(f) + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(formatter) } } -/// Convert an opaque Workspace Worker reference only at the Runtime-local boundary. -impl TryFrom<&RuntimeWorkerRef> for WorkerRef { - type Error = std::num::ParseIntError; +impl FromStr for WorkerId { + type Err = WorkerIdParseError; - fn try_from(value: &RuntimeWorkerRef) -> Result { - value - .worker_id - .parse::() - .map(WorkerId::new) - .map(Self::new) + fn from_str(value: &str) -> Result { + Self::parse(value).ok_or(WorkerIdParseError) } } -/// Runtime-local authority reference for Worker operations. +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 {} + +/// Runtime-local authority reference for Worker operations. The contained id is +/// nevertheless the Workspace-owned stable identity; the Runtime does not mint it. #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] pub struct WorkerRef { pub worker_id: WorkerId, @@ -56,28 +114,43 @@ impl WorkerRef { } } +impl TryFrom<&RuntimeWorkerRef> for WorkerRef { + type Error = WorkerIdParseError; + + fn try_from(value: &RuntimeWorkerRef) -> Result { + value.worker_id.parse().map(Self::new) + } +} + #[cfg(test)] mod tests { use super::*; #[test] - fn runtime_worker_ref_preserves_structured_identity_and_json_fields() { - let worker = RuntimeWorkerRef::new("arcadia", "30"); - assert_eq!(worker.runtime_id, "arcadia"); - assert_eq!(worker.worker_id, "30"); + 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(); + let worker = RuntimeWorkerRef::new("arcadia", worker_id.to_string()); assert_eq!( WorkerRef::try_from(&worker).unwrap(), - WorkerRef::new(WorkerId::new(30)) + WorkerRef::new(worker_id) ); assert_eq!( serde_json::to_value(&worker).unwrap(), - serde_json::json!({"runtime_id": "arcadia", "worker_id": "30"}) + serde_json::json!({"runtime_id": "arcadia", "worker_id": worker_id.to_string()}) ); } #[test] - fn runtime_worker_ref_does_not_treat_composite_text_as_local_worker_id() { - let worker = RuntimeWorkerRef::new("arcadia", "embedded-worker-runtime-5"); + fn runtime_worker_ref_rejects_legacy_numeric_identity() { + let worker = RuntimeWorkerRef::new("arcadia", "30"); assert!(WorkerRef::try_from(&worker).is_err()); } } diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index 84aed40a..117f0987 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -112,7 +112,14 @@ fn build_runtime(config: &ProcessConfig) -> Result { .map_err(ProcessError::Runtime) } RuntimeHttpStoreSelection::Fs { root } => { - let mut options = FsRuntimeStoreOptions::new(root.clone()); + let mut options = FsRuntimeStoreOptions::new(root.clone()).with_runtime_id( + config + .http + .auth + .as_ref() + .map(|auth| auth.runtime_id.as_str()) + .unwrap_or("local"), + ); options.display_name = config.http.display_name.clone(); Runtime::with_fs_store_and_execution_backend(options, backend) .map_err(ProcessError::Runtime) diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index a81f37cd..99967c3b 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -275,14 +275,13 @@ impl FsWorkerRetentionProvider { )); continue; } - let Ok(worker_number) = raw_id.parse::() else { + let Ok(worker_id) = raw_id.parse::() else { diagnostics.push(runtime_aggregate_diagnostic( &bounded_id, "aggregate_worker_id_invalid", )); continue; }; - let worker_id = WorkerId::new(worker_number); let worker_dir = self.worker_dir(worker_id); let snapshot: WorkerGenerationSnapshot = match read_json( &worker_dir.join("worker.json"), @@ -1315,7 +1314,7 @@ mod tests { #[test] fn archive_is_verified_before_source_removal_and_retry_converges() { let temp = tempfile::tempdir().unwrap(); - let worker_id = WorkerId::new(7); + let worker_id = WorkerId::from_legacy_u64(7); source(temp.path(), worker_id, 4); let provider = FsWorkerRetentionProvider::new(temp.path()); let request = request(worker_id, 4, SessionDisposition::Archive); @@ -1325,7 +1324,7 @@ mod tests { let archive = first.archive.as_ref().unwrap(); assert_eq!(archive.source_session_id, "session-a"); assert_eq!(archive.segment_ids, vec!["segment-a"]); - assert!(!temp.path().join("workers/7").exists()); + assert!(!temp.path().join(format!("workers/{worker_id}")).exists()); assert!( temp.path() .join("archives/workers/archive-a/session/segments/segment-a.jsonl") @@ -1346,7 +1345,7 @@ mod tests { #[test] fn archive_failure_keeps_live_source_for_retry() { let temp = tempfile::tempdir().unwrap(); - let worker_id = WorkerId::new(8); + let worker_id = WorkerId::from_legacy_u64(8); source(temp.path(), worker_id, 2); let collision = temp.path().join("archives/workers/archive-a"); fs::create_dir_all(&collision).unwrap(); @@ -1358,7 +1357,11 @@ mod tests { .execute(&request(worker_id, 2, SessionDisposition::Archive)) .is_err() ); - assert!(temp.path().join("workers/8/session").is_dir()); + assert!( + temp.path() + .join(format!("workers/{worker_id}/session")) + .is_dir() + ); assert!( !temp .path() @@ -1370,7 +1373,7 @@ mod tests { #[test] fn target_inventory_and_execute_reject_cross_workspace_aggregate() { let temp = tempfile::tempdir().unwrap(); - let worker_id = WorkerId::new(16); + let worker_id = WorkerId::from_legacy_u64(16); source(temp.path(), worker_id, 3); let provider = FsWorkerRetentionProvider::new(temp.path()); assert!(matches!( @@ -1383,7 +1386,11 @@ mod tests { provider.execute(&request), Err(RuntimeError::WorkerNotFound { .. }) )); - assert!(temp.path().join("workers/16/session").is_dir()); + assert!( + temp.path() + .join(format!("workers/{worker_id}/session")) + .is_dir() + ); assert!( !temp .path() @@ -1401,18 +1408,22 @@ mod tests { fn purge_removes_aggregate_and_rejects_stale_generation() { let temp = tempfile::tempdir().unwrap(); let provider = FsWorkerRetentionProvider::new(temp.path()); - let worker_id = WorkerId::new(9); + let worker_id = WorkerId::from_legacy_u64(9); source(temp.path(), worker_id, 5); let stale = request(worker_id, 4, SessionDisposition::Purge); assert!(provider.execute(&stale).is_err()); - assert!(temp.path().join("workers/9/session").is_dir()); + assert!( + temp.path() + .join(format!("workers/{worker_id}/session")) + .is_dir() + ); let mut current = request(worker_id, 5, SessionDisposition::Purge); current.operation_id = "operation-current".to_string(); current.input_fingerprint = "fingerprint-current".to_string(); let result = provider.execute(¤t).unwrap(); assert!(result.archive.is_none()); - assert!(!temp.path().join("workers/9").exists()); + assert!(!temp.path().join(format!("workers/{worker_id}")).exists()); assert!( temp.path() .join("retention/operations/operation-current.json") @@ -1423,7 +1434,7 @@ mod tests { #[test] fn pending_receipt_recovers_delete_to_receipt_crash_window() { let temp = tempfile::tempdir().unwrap(); - let worker_id = WorkerId::new(11); + let worker_id = WorkerId::from_legacy_u64(11); source(temp.path(), worker_id, 1); let provider = FsWorkerRetentionProvider::new(temp.path()); let request = request(worker_id, 1, SessionDisposition::Archive); @@ -1442,10 +1453,15 @@ mod tests { #[test] fn provider_snapshot_scans_aggregate_storage_independent_of_runtime_catalog() { let temp = tempfile::tempdir().unwrap(); - source(temp.path(), WorkerId::new(13), 2); - source(temp.path(), WorkerId::new(14), 1); + source(temp.path(), WorkerId::from_legacy_u64(13), 2); + let other_worker = WorkerId::from_legacy_u64(14); + source(temp.path(), other_worker, 1); write_json( - &temp.path().join("workers/14/worker.json"), + &temp + .path() + .join("workers") + .join(other_worker.to_string()) + .join("worker.json"), &serde_json::json!({"workspace_id": "other-workspace", "run_generation": 1}), ); fs::create_dir_all(temp.path().join("workers/not-a-worker")).unwrap(); @@ -1454,15 +1470,20 @@ mod tests { b"not-json", ) .unwrap(); - fs::create_dir_all(temp.path().join("workers/15")).unwrap(); - fs::write(temp.path().join("workers/15/worker.json"), b"not-json").unwrap(); + let corrupt_worker = WorkerId::from_legacy_u64(15); + let corrupt_worker_dir = temp.path().join("workers").join(corrupt_worker.to_string()); + fs::create_dir_all(&corrupt_worker_dir).unwrap(); + fs::write(corrupt_worker_dir.join("worker.json"), b"not-json").unwrap(); let provider = FsWorkerRetentionProvider::new(temp.path()); let snapshot = provider.snapshot("workspace-a", "runtime-a").unwrap(); assert_eq!(snapshot.workers().len(), 1); - assert_eq!(snapshot.workers()[0].worker_id, WorkerId::new(13)); + assert_eq!( + snapshot.workers()[0].worker_id, + WorkerId::from_legacy_u64(13) + ); assert!(snapshot.diagnostics().iter().any(|diagnostic| { - diagnostic.worker_id() == "14" + diagnostic.worker_id() == other_worker.to_string() && diagnostic.category() == "aggregate_workspace_mismatch" })); assert!(snapshot.diagnostics().iter().any(|diagnostic| { @@ -1470,7 +1491,7 @@ mod tests { && diagnostic.category() == "aggregate_worker_id_invalid" })); assert!(snapshot.diagnostics().iter().any(|diagnostic| { - diagnostic.worker_id() == "15" + diagnostic.worker_id() == corrupt_worker.to_string() && diagnostic.category() == "aggregate_worker_record_corrupt" })); } @@ -1478,7 +1499,7 @@ mod tests { #[test] fn diagnostics_retry_rejects_corrupt_existing_archive_before_source_delete() { let temp = tempfile::tempdir().unwrap(); - let worker_id = WorkerId::new(12); + let worker_id = WorkerId::from_legacy_u64(12); source(temp.path(), worker_id, 1); let provider = FsWorkerRetentionProvider::new(temp.path()); let mut request = request(worker_id, 1, SessionDisposition::Archive); @@ -1499,14 +1520,18 @@ mod tests { .unwrap(); assert!(provider.execute(&request).is_err()); - assert!(temp.path().join("workers/12/session").is_dir()); + assert!( + temp.path() + .join(format!("workers/{worker_id}/session")) + .is_dir() + ); assert!(provider.completed_for(&request).unwrap().is_none()); } #[test] fn concurrent_retry_produces_one_archive() { let temp = tempfile::tempdir().unwrap(); - let worker_id = WorkerId::new(10); + let worker_id = WorkerId::from_legacy_u64(10); source(temp.path(), worker_id, 1); let provider = Arc::new(FsWorkerRetentionProvider::new(temp.path())); let request = Arc::new(request(worker_id, 1, SessionDisposition::Archive)); diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 75b1e19c..b45abbe7 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -196,7 +196,7 @@ impl Runtime { options: FsRuntimeStoreOptions, execution_backend: Option, ) -> Result { - let opened = FsRuntimeStore::open_or_create(options.root)?; + let opened = FsRuntimeStore::open_or_create(options.root, &options.runtime_id)?; let mut state = if let Some(persisted) = opened.state { RuntimeState::from_persisted(persisted, opened.store)? } else { @@ -492,7 +492,7 @@ impl Runtime { let state = self.lock()?; status.summary.primary_worker_id = state .primary_worker_id_for_workdir(status.summary.working_directory_id.as_str()) - .map(|worker_id| worker_id.as_u64()); + .map(|worker_id| worker_id.to_string()); Ok(status) } @@ -518,11 +518,6 @@ impl Runtime { request: CreateWorkerRequest, scope: Option<&RuntimeWorkspaceScope>, ) -> Result { - if request.idempotency_key.is_some() != request.idempotency_fingerprint.is_some() { - return Err(RuntimeError::InvalidRequest( - "idempotency_key and idempotency_fingerprint must be provided together".to_string(), - )); - } let (backend, worker_ref, spawn_request) = { let mut state = self.lock()?; state.ensure_running()?; @@ -534,19 +529,21 @@ impl Runtime { if let Some(scope) = scope { state.ensure_workspace_owner(scope, true)?; }; - if let Some(idempotency_key) = request.idempotency_key.as_deref() { - let workspace_id = scope.map(|scope| scope.workspace_id.as_str()); - if let Some(existing) = state.workers.values().find(|record| { - record.workspace_id.as_deref() == workspace_id - && record.request.idempotency_key.as_deref() == Some(idempotency_key) - }) { - if existing.request.idempotency_fingerprint != request.idempotency_fingerprint { - return Err(RuntimeError::InvalidRequest(format!( - "worker creation idempotency key {idempotency_key} was already used with different input" - ))); - } - return Ok(existing.detail()); + let workspace_id = scope.map(|scope| scope.workspace_id.as_str()); + if let Some(existing) = state.workers.get(&request.worker_id) { + if existing.workspace_id.as_deref() != workspace_id { + return Err(RuntimeError::InvalidRequest(format!( + "worker {} already belongs to another Workspace scope", + request.worker_id + ))); } + if existing.request.create_fingerprint != request.create_fingerprint { + return Err(RuntimeError::InvalidRequest(format!( + "worker {} was already created with a different fingerprint", + request.worker_id + ))); + } + return Ok(existing.detail()); } state.validate_worker_config_boundary(&request)?; if let Some(working_directory_id) = requested_primary_workdir_id(&request) { @@ -565,9 +562,8 @@ impl Runtime { })?; let config_bundle = state.resolve_config_bundle_ref(request.config_bundle.as_ref())?; - let worker_id = WorkerId::generated(state.next_worker_sequence); - state.next_worker_sequence += 1; - let worker_ref = WorkerRef::new(worker_id.clone()); + let worker_id = request.worker_id; + let worker_ref = WorkerRef::new(worker_id); let record = WorkerRecord { worker_ref: worker_ref.clone(), @@ -1850,7 +1846,6 @@ struct RuntimeState { persistence: RuntimePersistence, status: RuntimeStatus, execution_backend: Option, - next_worker_sequence: u64, #[cfg(feature = "fs-store")] next_diagnostic_id: u64, workers: BTreeMap, @@ -1878,7 +1873,6 @@ impl RuntimeState { persistence: RuntimePersistence::Memory, status: RuntimeStatus::Running, execution_backend: None, - next_worker_sequence: 1, #[cfg(feature = "fs-store")] next_diagnostic_id: 1, workers: BTreeMap::new(), @@ -1907,7 +1901,6 @@ impl RuntimeState { persistence: RuntimePersistence::Fs(store), status: RuntimeStatus::Running, execution_backend: None, - next_worker_sequence: 1, #[cfg(feature = "fs-store")] next_diagnostic_id: 1, workers: BTreeMap::new(), @@ -1958,7 +1951,6 @@ impl RuntimeState { persistence: RuntimePersistence::Fs(store), status: persisted.status, execution_backend: None, - next_worker_sequence: persisted.next_worker_sequence, next_diagnostic_id, workers, config_bundles: BTreeMap::new(), @@ -1982,7 +1974,6 @@ impl RuntimeState { PersistedRuntimeState { display_name: self.display_name.clone(), status: self.status, - next_worker_sequence: self.next_worker_sequence, next_diagnostic_id: self.next_diagnostic_id, workers: self .workers @@ -2589,6 +2580,11 @@ fn requested_primary_workdir_id(request: &CreateWorkerRequest) -> Option<&str> { } fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), RuntimeError> { + if request.create_fingerprint.trim().is_empty() { + return Err(RuntimeError::InvalidRequest( + "create_fingerprint must not be empty".to_string(), + )); + } match &request.profile_source { crate::catalog::ProfileSourceArchiveSource::Embedded { archive } => { archive.verify().map_err(|err| { @@ -2791,8 +2787,8 @@ mod tests { let profile = ProfileSelector::Builtin("builtin:coder".to_string()); let bundle = test_bundle_for_profile(profile.clone()); CreateWorkerRequest { - idempotency_key: None, - idempotency_fingerprint: None, + worker_id: WorkerId::now_v7(), + create_fingerprint: "test-create".to_string(), profile, display_name: None, profile_source: crate::catalog::ProfileSourceArchiveSource::Http { @@ -3570,8 +3566,7 @@ mod tests { fn create_worker_idempotency_reuses_worker_and_rejects_different_input() { let runtime = runtime_with_backend(); let mut request = task_request("idempotent"); - request.idempotency_key = Some("operation-1".to_string()); - request.idempotency_fingerprint = Some("sha256:input-1".to_string()); + request.create_fingerprint = "sha256:input-1".to_string(); request.working_directory = Some(WorkingDirectoryClaim { working_directory_id: "workdir-idempotent".to_string(), relative_cwd: None, @@ -3587,7 +3582,7 @@ mod tests { workdir_count_after_first ); - request.idempotency_fingerprint = Some("sha256:different".to_string()); + request.create_fingerprint = "sha256:different".to_string(); let error = runtime.create_worker(request).unwrap_err(); assert!(matches!(error, RuntimeError::InvalidRequest(_))); assert_eq!(runtime.list_workers().unwrap().len(), 1); @@ -4141,6 +4136,85 @@ mod tests { } } + #[cfg(feature = "fs-store")] + #[test] + fn fs_store_migrates_legacy_numeric_worker_identity_to_workspace_uuid() { + let root = fs_store_root("worker-id-v1"); + let runtime_id = "arcadia"; + let runtime = Runtime::with_fs_store_and_execution_backend( + crate::fs_store::FsRuntimeStoreOptions { + root: root.clone(), + runtime_id: runtime_id.to_string(), + display_name: None, + }, + Arc::new(TestExecutionBackend::default()), + ) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker_scoped( + &RuntimeWorkspaceScope::new("workspace-a", "server"), + task_request("legacy"), + ) + .unwrap(); + drop(runtime); + + let current_dir = root.join("workers").join(worker.worker_id.to_string()); + let legacy_dir = root.join("workers").join("7"); + std::fs::rename(¤t_dir, &legacy_dir).unwrap(); + let worker_path = legacy_dir.join("worker.json"); + let mut worker_json: serde_json::Value = + serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); + worker_json["schema_version"] = serde_json::json!(1); + worker_json["worker_id"] = serde_json::json!(7); + worker_json["worker_ref"]["worker_id"] = serde_json::json!(7); + let request = worker_json["request"].as_object_mut().unwrap(); + request.remove("worker_id"); + request.remove("create_fingerprint"); + request.insert("idempotency_key".to_string(), serde_json::Value::Null); + request.insert( + "idempotency_fingerprint".to_string(), + serde_json::Value::Null, + ); + std::fs::write( + &worker_path, + serde_json::to_vec_pretty(&worker_json).unwrap(), + ) + .unwrap(); + + let runtime_path = root.join("runtime.json"); + let mut runtime_json: serde_json::Value = + serde_json::from_slice(&std::fs::read(&runtime_path).unwrap()).unwrap(); + runtime_json["schema_version"] = serde_json::json!(1); + runtime_json["workers"] = serde_json::json!([7]); + runtime_json["next_worker_sequence"] = serde_json::json!(8); + std::fs::write( + &runtime_path, + serde_json::to_vec_pretty(&runtime_json).unwrap(), + ) + .unwrap(); + + let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { + root: root.clone(), + runtime_id: runtime_id.to_string(), + display_name: None, + }) + .unwrap(); + let expected = WorkerId::from_legacy_binding("workspace-a", runtime_id, 7); + let detail = restored.worker_detail(&WorkerRef::new(expected)).unwrap(); + assert_eq!(detail.worker_id, expected); + assert_eq!(detail.worker_ref.worker_id, expected); + assert!(root.join("workers").join(expected.to_string()).exists()); + assert!(!legacy_dir.exists()); + let migrated_runtime: serde_json::Value = + serde_json::from_slice(&std::fs::read(runtime_path).unwrap()).unwrap(); + assert_eq!(migrated_runtime["schema_version"], serde_json::json!(2)); + assert!(migrated_runtime.get("next_worker_sequence").is_none()); + + drop(restored); + let _ = std::fs::remove_dir_all(root); + } + #[cfg(feature = "fs-store")] #[test] fn fs_store_restores_workers_without_legacy_event_or_protocol_observation_logs() { @@ -4148,6 +4222,7 @@ mod tests { let runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), + runtime_id: "test-runtime".to_string(), display_name: Some("filesystem runtime".to_string()), }, Arc::new(TestExecutionBackend::default()), @@ -4189,6 +4264,7 @@ mod tests { let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }) .unwrap(); @@ -4239,6 +4315,7 @@ mod tests { let runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }, Arc::new(TestExecutionBackend::default()), @@ -4264,6 +4341,7 @@ mod tests { let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }) .unwrap(); @@ -4313,6 +4391,7 @@ mod tests { let runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }, Arc::new(TestExecutionBackend::default()), @@ -4326,6 +4405,7 @@ mod tests { let backendless = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }) .unwrap(); @@ -4337,6 +4417,7 @@ mod tests { let restored = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }, restoring_backend.clone(), @@ -4360,6 +4441,7 @@ mod tests { let runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }, Arc::new(TestExecutionBackend::default()), @@ -4379,6 +4461,7 @@ mod tests { let restored = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }, restoring_backend.clone(), @@ -4415,6 +4498,7 @@ mod tests { let corrupt_root = fs_store_root("corrupt"); let corrupt_runtime = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { root: corrupt_root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }) .unwrap(); @@ -4427,6 +4511,7 @@ mod tests { drop(corrupt_runtime); let err = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { root: corrupt_root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }) .unwrap_err(); @@ -4437,6 +4522,7 @@ mod tests { let missing_runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { root: missing_root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }, Arc::new(TestExecutionBackend::default()), @@ -4456,6 +4542,7 @@ mod tests { drop(missing_runtime); let loaded = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { root: missing_root.clone(), + runtime_id: "test-runtime".to_string(), display_name: None, }) .expect("invalid worker snapshot should not make runtime store unreadable"); diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 804076e0..7efbc92e 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -1998,6 +1998,7 @@ mod tests { WorkingDirectoryRequest, }; use crate::execution::WorkerExecutionContext; + use crate::identity::WorkerId; use crate::identity::WorkerRef; use crate::management::RuntimeOptions; use crate::observation::WorkerObservationCursor; @@ -2119,7 +2120,7 @@ mod tests { #[test] fn restart_restore_reconstructs_runtime_owned_worker_mutation_client() { let identity = RuntimeIdentityMaterial::generate("runtime-source").unwrap(); - let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(17)); + let worker_ref = WorkerRef::new(crate::identity::WorkerId::from_legacy_u64(17)); let backend = RuntimeWorkspaceBackendRef::Http { workspace_id: "workspace-a".to_string(), base_url: "https://server.invalid".to_string(), @@ -2486,8 +2487,8 @@ mod tests { fn create_request(_name: &str) -> CreateWorkerRequest { let bundle = test_bundle(); CreateWorkerRequest { - idempotency_key: None, - idempotency_fingerprint: None, + worker_id: WorkerId::now_v7(), + create_fingerprint: "test-create".to_string(), profile: ProfileSelector::Builtin("builtin:companion".to_string()), display_name: None, profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded { @@ -2590,7 +2591,8 @@ mod tests { #[tokio::test] async fn runtime_provider_projects_only_explicit_live_canonical_grants() { let hub = Arc::new(RuntimeWorkerObservationHub::default()); - let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(7)); + let worker_id = crate::identity::WorkerId::from_legacy_u64(7); + let worker_ref = WorkerRef::new(worker_id); let shared_state = Arc::new(WorkerSharedState::new( "peer-worker".to_string(), session_store::new_segment_id(), @@ -2614,7 +2616,7 @@ mod tests { sink: SegmentLogSink::new(), }, ); - let grant = crate::identity::RuntimeWorkerRef::new("runtime-1", "7"); + let grant = crate::identity::RuntimeWorkerRef::new("runtime-1", worker_id.to_string()); let provider = RuntimeGrantedWorkerObservationProvider { runtime_id: "runtime-1".to_string(), workspace_id: "workspace-1".to_string(), @@ -2628,7 +2630,7 @@ mod tests { listed[0].subject, WorkerObservationSubjectRef::RuntimeWorker { runtime_id: "runtime-1".to_string(), - worker_id: "7".to_string(), + worker_id: worker_id.to_string(), } ); provider @@ -2668,8 +2670,9 @@ mod tests { } #[test] - fn runtime_worker_name_is_runtime_local() { - let worker_ref = crate::identity::WorkerRef::new(crate::identity::WorkerId::new(1)); + fn runtime_worker_name_uses_workspace_worker_identity() { + let worker_ref = + crate::identity::WorkerRef::new(crate::identity::WorkerId::from_legacy_u64(1)); let request = WorkerExecutionSpawnRequest { worker_ref: worker_ref.clone(), run_generation: 1, @@ -2682,7 +2685,7 @@ mod tests { assert_eq!( ProfileRuntimeWorkerFactory::runtime_worker_name(&request), - "worker-runtime-1" + format!("worker-runtime-{}", request.worker_ref.worker_id) ); assert_ne!( ProfileRuntimeWorkerFactory::runtime_worker_name(&request), @@ -2765,8 +2768,10 @@ mod tests { async fn restore_pending_workspace_worker_without_system_prompt_fails_closed() { let root = tempfile::tempdir().unwrap(); let runtime_store_dir = root.path().join("runtime"); - let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(1)); - let worker_aggregate_dir = runtime_store_dir.join("workers/1"); + let worker_ref = WorkerRef::new(crate::identity::WorkerId::from_legacy_u64(1)); + let worker_aggregate_dir = runtime_store_dir + .join("workers") + .join(worker_ref.worker_id.to_string()); let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref); let session_id = session_store::new_session_id(); let manifest = manifest::WorkerManifest::from_toml(&format!( @@ -2842,8 +2847,10 @@ mod tests { let root = tempfile::tempdir().unwrap(); let long_component = "embedded-workspace-store-segment".repeat(4); let runtime_store_dir = root.path().join(long_component); - let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(1)); - let worker_aggregate_dir = runtime_store_dir.join("workers/1"); + let worker_ref = WorkerRef::new(crate::identity::WorkerId::from_legacy_u64(1)); + let worker_aggregate_dir = runtime_store_dir + .join("workers") + .join(worker_ref.worker_id.to_string()); let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref); let session_id = session_store::new_session_id(); let manifest = manifest::WorkerManifest::from_toml(&format!( @@ -2880,7 +2887,10 @@ mod tests { ) .unwrap(); - let run_dir = runtime_store_dir.join("workers/1/runs/2"); + let run_dir = runtime_store_dir + .join("workers") + .join(worker_ref.worker_id.to_string()) + .join("runs/2"); let socket_path = run_dir.join("worker.sock"); assert!( socket_path.as_os_str().as_encoded_bytes().len() > 107, @@ -2926,6 +2936,7 @@ mod tests { let runtime_store_dir = root.path().join(long_component); let runtime_options = crate::fs_store::FsRuntimeStoreOptions { root: runtime_store_dir.clone(), + runtime_id: "test-runtime".to_string(), display_name: Some("embedded".to_string()), }; @@ -2946,7 +2957,10 @@ mod tests { let mut request = create_request("embedded singleton"); request.profile = ProfileSelector::Builtin("default".to_string()); let worker = runtime.create_worker(request).unwrap(); - let first_run_socket = runtime_store_dir.join("workers/1/runs/1/worker.sock"); + let first_run_socket = runtime_store_dir + .join("workers") + .join(worker.worker_id.to_string()) + .join("runs/1/worker.sock"); assert!( first_run_socket.as_os_str().as_encoded_bytes().len() > 107, "test path must exceed Linux sockaddr_un.sun_path capacity: {}", @@ -2998,7 +3012,10 @@ mod tests { diagnostic.code == "worker_execution_restore_failed" && diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref) })); - let restored_run = runtime_store_dir.join("workers/1/runs/2"); + let restored_run = runtime_store_dir + .join("workers") + .join(worker.worker_id.to_string()) + .join("runs/2"); assert!(!restored_run.join("worker.sock").exists()); assert!(restored_run.join("worker.out.log").is_file()); assert!(restored_run.join("worker.err.log").is_file()); diff --git a/crates/worker-runtime/src/working_directory.rs b/crates/worker-runtime/src/working_directory.rs index 56f367d6..c675281b 100644 --- a/crates/worker-runtime/src/working_directory.rs +++ b/crates/worker-runtime/src/working_directory.rs @@ -803,7 +803,7 @@ mod tests { } fn worker_ref(sequence: u64) -> WorkerRef { - WorkerRef::new(WorkerId::generated(sequence)) + WorkerRef::new(WorkerId::from_legacy_u64(sequence)) } #[test] diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 6087d38d..c43d3b22 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -340,6 +340,14 @@ pub struct WorkerTicketAssignmentRequest { pub operation_id: String, } +pub(crate) fn worker_spawn_create_fingerprint( + request: &WorkerSpawnRequest, +) -> Result { + let encoded = serde_json::to_vec(request) + .map_err(|error| format!("serialize Worker create input: {error}"))?; + Ok(format!("sha256:{}", digest_hex(&encoded, 64))) +} + pub(crate) fn worker_spawn_idempotency( request: &WorkerSpawnRequest, ) -> Result, String> { @@ -366,6 +374,12 @@ pub struct WorkerControlOperation { pub input_fingerprint: String, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerCreateBinding { + pub worker_id: EmbeddedWorkerId, + pub create_fingerprint: String, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct WorkerSpawnRequest { @@ -763,7 +777,11 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { } } - fn spawn_worker(&self, request: WorkerSpawnRequest) -> WorkerSpawnResult { + fn spawn_worker( + &self, + _binding: WorkerCreateBinding, + request: WorkerSpawnRequest, + ) -> WorkerSpawnResult { WorkerSpawnResult { state: WorkerOperationState::Unsupported, worker: None, @@ -1226,6 +1244,7 @@ impl RuntimeRegistry { pub fn spawn_worker( &self, runtime_id: &str, + binding: WorkerCreateBinding, request: WorkerSpawnRequest, ) -> Result { validate_backend_identifier("runtime_id", runtime_id)?; @@ -1269,7 +1288,7 @@ impl RuntimeRegistry { }); } } - Ok(runtime.spawn_worker(request)) + Ok(runtime.spawn_worker(binding, request)) } pub fn create_working_directory( @@ -1606,6 +1625,7 @@ impl EmbeddedWorkerRuntime { let runtime = worker_runtime::Runtime::with_fs_store_and_execution_backend( FsRuntimeStoreOptions { root: store_root.into(), + runtime_id: EMBEDDED_RUNTIME_ID.to_string(), display_name: Some("embedded".to_string()), }, backend, @@ -1948,7 +1968,11 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { } } - fn spawn_worker(&self, request: WorkerSpawnRequest) -> WorkerSpawnResult { + fn spawn_worker( + &self, + binding: WorkerCreateBinding, + request: WorkerSpawnRequest, + ) -> WorkerSpawnResult { let mut diagnostics = Vec::new(); if request.resolved_working_directory_request.is_some() || request.resolved_working_directory.is_some() @@ -1981,7 +2005,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { diagnostics.push(diagnostic( "embedded_worker_name_display_only", DiagnosticSeverity::Info, - "requested_worker_name is used only as display_name; embedded Runtime allocates opaque runtime-local worker ids".to_string(), + "requested_worker_name is used only as display_name; Worker identity is allocated by Workspace authority".to_string(), )); } if matches!(request.acceptance, WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments } if expected_segments > 0) @@ -2010,11 +2034,6 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { }; } }; - let (idempotency_key, idempotency_fingerprint) = worker_spawn_idempotency(&request) - .expect("WorkerSpawnRequest serialization is infallible") - .map_or((None, None), |(key, fingerprint)| { - (Some(key), Some(fingerprint)) - }); let workspace_api = match required_worker_workspace_api(&request) { Ok(workspace_api) => workspace_api, Err(diagnostic) => { @@ -2030,8 +2049,8 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { let workspace_id = workspace_api.workspace_id.clone(); let config_bundle = spawn_config_bundle_ref(&request); let create_request = CreateWorkerRequest { - idempotency_key, - idempotency_fingerprint, + worker_id: binding.worker_id, + create_fingerprint: binding.create_fingerprint, profile, display_name: request.requested_worker_name.clone(), config_bundle, @@ -3113,7 +3132,11 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } } - fn spawn_worker(&self, request: WorkerSpawnRequest) -> WorkerSpawnResult { + fn spawn_worker( + &self, + binding: WorkerCreateBinding, + request: WorkerSpawnRequest, + ) -> WorkerSpawnResult { if matches!( request.acceptance, WorkerSpawnAcceptanceRequirement::SocketReady @@ -3152,11 +3175,6 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { }; } }; - let (idempotency_key, idempotency_fingerprint) = worker_spawn_idempotency(&request) - .expect("WorkerSpawnRequest serialization is infallible") - .map_or((None, None), |(key, fingerprint)| { - (Some(key), Some(fingerprint)) - }); let workspace_api = match required_worker_workspace_api(&request) { Ok(workspace_api) => workspace_api, Err(diagnostic) => { @@ -3170,8 +3188,8 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { }; let config_bundle = spawn_config_bundle_ref(&request); let create = CreateWorkerRequest { - idempotency_key, - idempotency_fingerprint, + worker_id: binding.worker_id, + create_fingerprint: binding.create_fingerprint, profile, display_name: request.requested_worker_name.clone(), config_bundle, @@ -4245,6 +4263,13 @@ mod tests { use std::sync::{Arc, Mutex}; use std::thread; + fn test_create_binding() -> WorkerCreateBinding { + WorkerCreateBinding { + worker_id: EmbeddedWorkerId::now_v7(), + create_fingerprint: "sha256:test-create".to_string(), + } + } + fn test_workspace_api() -> WorkspaceApiRef { WorkspaceApiRef { workspace_id: "workspace-test".to_string(), @@ -4899,11 +4924,16 @@ mod tests { digest: bundle.metadata.digest.clone(), }; request.resolved_config_bundle = Some(bundle); + let binding = test_create_binding(); let result = registry - .spawn_worker("embedded-worker-runtime", request) + .spawn_worker("embedded-worker-runtime", binding.clone(), request) .expect("spawn request"); assert_eq!(result.state, WorkerOperationState::Accepted); + assert_eq!( + result.worker.as_ref().unwrap().worker.worker_id, + binding.worker_id.to_string() + ); let check = registry .check_config_bundle("embedded-worker-runtime", bundle_ref) .expect("bundle check"); @@ -4921,7 +4951,7 @@ mod tests { let mut request = embedded_spawn_request(); request.resolved_workspace_api = None; - let spawned = runtime.spawn_worker(request); + let spawned = runtime.spawn_worker(test_create_binding(), request); assert_eq!(spawned.state, WorkerOperationState::Rejected); assert!( @@ -4939,7 +4969,7 @@ mod tests { Arc::new(FailingSpawnBackend), ) .expect("test backend should connect"); - let spawned = runtime.spawn_worker(embedded_spawn_request()); + let spawned = runtime.spawn_worker(test_create_binding(), embedded_spawn_request()); assert_eq!(spawned.state, WorkerOperationState::Rejected); assert!(spawned.acceptance_evidence.is_empty()); assert!(spawned.diagnostics.iter().any(|diagnostic| { @@ -5006,7 +5036,7 @@ mod tests { Arc::new(AcceptingExecutionBackend::default()), ) .expect("test backend should connect"); - let spawned = runtime.spawn_worker(embedded_spawn_request()); + let spawned = runtime.spawn_worker(test_create_binding(), embedded_spawn_request()); assert_eq!(spawned.state, WorkerOperationState::Accepted); let worker = spawned.worker.expect("created embedded worker"); assert!(worker.capabilities.can_stop); @@ -5064,6 +5094,7 @@ mod tests { let spawned = registry .spawn_worker( EMBEDDED_RUNTIME_ID, + test_create_binding(), WorkerSpawnRequest { intent: WorkerSpawnIntent::TicketRole { ticket_id: "00001KVZSGT0Q".to_string(), @@ -5162,6 +5193,7 @@ mod tests { let spawned = registry .spawn_worker( EMBEDDED_RUNTIME_ID, + test_create_binding(), WorkerSpawnRequest { intent: WorkerSpawnIntent::TicketRole { ticket_id: "00001KVZSGT0Q".to_string(), @@ -5204,6 +5236,7 @@ mod tests { let result = registry .spawn_worker( EMBEDDED_RUNTIME_ID, + test_create_binding(), WorkerSpawnRequest { intent: WorkerSpawnIntent::WorkspaceCompanion, requested_worker_name: None, @@ -5251,7 +5284,8 @@ mod tests { #[test] fn remote_runtime_registry_routes_commands_without_browser_secret_leaks() { - let worker_json = worker_json("remote:primary", "1"); + let worker_id = EmbeddedWorkerId::from_legacy_u64(1).to_string(); + let worker_json = worker_json("remote:primary", &worker_id); let (base_url, server) = serve_mock_http(vec![ mock_response( "GET", @@ -5262,19 +5296,19 @@ mod tests { ), mock_response( "GET", - "/v1/workers/1", + format!("/v1/workers/{worker_id}"), true, 200, json!({ "worker": worker_json.clone() }).to_string(), ), mock_response( "POST", - "/v1/workers/1/input", + format!("/v1/workers/{worker_id}/input"), true, 200, json!({ "ack": { - "worker_ref": { "runtime_id": "remote:primary", "worker_id": 1 }, + "worker_ref": { "runtime_id": "remote:primary", "worker_id": worker_id.clone() }, "status": "running" } }) @@ -5298,20 +5332,24 @@ mod tests { ); let observation = registry - .observation_source(&RuntimeWorkerRef::new("remote:primary", "1")) + .observation_source(&RuntimeWorkerRef::new("remote:primary", &worker_id)) .expect("remote runtime exposes backend-owned WS observation source"); let crate::observation::RuntimeObservationSource::RemoteWs(observation) = observation else { panic!("remote runtime should expose a remote WS observation source"); }; assert!(observation.endpoint.starts_with("ws://127.0.0.1:")); - assert!(observation.endpoint.ends_with("/v1/workers/1/protocol/ws")); + assert!( + observation + .endpoint + .ends_with(&format!("/v1/workers/{worker_id}/protocol/ws")) + ); assert_eq!(observation.bearer_token.as_deref(), Some(secret.as_str())); let workers = registry.list_workers(10); assert_eq!(workers.items.len(), 1); assert_eq!(workers.items[0].worker.runtime_id, "remote:primary"); - assert_eq!(workers.items[0].worker.worker_id, "1"); + assert_eq!(workers.items[0].worker.worker_id, worker_id.as_str()); assert_eq!( workers.items[0].implementation.kind, "remote_worker_runtime" @@ -5324,7 +5362,7 @@ mod tests { let input = registry .send_input( - &RuntimeWorkerRef::new("remote:primary", "1"), + &RuntimeWorkerRef::new("remote:primary", &worker_id), WorkerInputRequest { kind: WorkerInputKind::User, content: "hello remote".to_string(), @@ -5350,6 +5388,10 @@ mod tests { #[test] fn remote_runtime_projection_uses_canonical_worker_status_for_stop_capability() { + let worker_ids = (1..=4) + .map(|value| EmbeddedWorkerId::from_legacy_u64(value).to_string()) + .collect::>(); + let worker_id = worker_ids[0].clone(); let (base_url, server) = serve_mock_http(vec![ mock_response( "GET", @@ -5358,21 +5400,26 @@ mod tests { 200, json!({ "workers": [ - worker_json_with_status("remote:primary", "1", "stopped"), - worker_json_with_status("remote:primary", "2", "cancelled"), - worker_json_with_status("remote:primary", "3", "paused"), - worker_json_with_status("remote:primary", "4", "idle") + worker_json_with_status("remote:primary", &worker_ids[0], "stopped"), + worker_json_with_status("remote:primary", &worker_ids[1], "cancelled"), + worker_json_with_status("remote:primary", &worker_ids[2], "paused"), + worker_json_with_status("remote:primary", &worker_ids[3], "idle") ] }) .to_string(), ), mock_response( "GET", - "/v1/workers/1", + format!("/v1/workers/{worker_id}"), true, 200, json!({ - "worker": worker_json_with_status("remote:primary", "1", "stopped")}) + "worker": worker_json_with_status( + "remote:primary", + &worker_ids[0], + "stopped" + ) + }) .to_string(), ), ]); @@ -5402,7 +5449,7 @@ mod tests { assert_eq!(workers.items[3].state, "idle"); let stopped_detail = registry - .worker(&RuntimeWorkerRef::new("remote:primary", "1")) + .worker(&RuntimeWorkerRef::new("remote:primary", &worker_id)) .unwrap(); assert!(!stopped_detail.capabilities.can_stop); assert_eq!(stopped_detail.state, "stopped"); @@ -5595,7 +5642,7 @@ mod tests { #[derive(Clone)] struct MockResponse { method: &'static str, - path: &'static str, + path: String, require_auth: bool, status: u16, body: String, @@ -5603,14 +5650,14 @@ mod tests { fn mock_response( method: &'static str, - path: &'static str, + path: impl Into, require_auth: bool, status: u16, body: String, ) -> MockResponse { MockResponse { method, - path, + path: path.into(), require_auth, status, body, @@ -5667,7 +5714,6 @@ mod tests { worker_id: &str, status: &str, ) -> serde_json::Value { - let worker_id = worker_id.parse::().unwrap(); json!({ "worker_ref": { "runtime_id": runtime_id, "worker_id": worker_id }, "runtime_id": runtime_id, diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index 90b89935..f57787dd 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -285,7 +285,7 @@ impl SqliteWorkspaceStore { let worker=match load_worker(&tx,&req.workspace_id,&req.worker)? { Some(v)=>v, None=>{ - let other:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM worker_registry WHERE runtime_id=?1 AND runtime_worker_id=?2 AND workspace_id!=?3)",params![req.worker.runtime_id,req.worker.worker_id,req.workspace_id],|r|r.get(0))?; + let other:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM worker_registry WHERE runtime_id=?1 AND worker_id=?2 AND workspace_id!=?3)",params![req.worker.runtime_id,req.worker.worker_id,req.workspace_id],|r|r.get(0))?; return Err(StoreError::InvalidInput(if other{"cross-workspace".into()}else{"worker-missing".into()})); } }; @@ -366,11 +366,13 @@ impl SqliteWorkspaceStore { plan_id: plan.plan_id.clone(), reason: "Worker disappeared after execution fence".to_string(), })?; - let worker_number = plan.worker.worker_id.parse::().map_err(|_| { - WorkerRetentionError::Invalid( - "Runtime Worker id is not a canonical unsigned integer".to_string(), - ) - })?; + let worker_id = plan + .worker + .worker_id + .parse::() + .map_err(|_| { + WorkerRetentionError::Invalid("Worker id must be a canonical UUIDv7".to_string()) + })?; let removed_at = plan.created_at.clone(); let prior_failure_category = plan.failure_category.clone(); Ok(PreparedWorkerRemoval { @@ -380,7 +382,7 @@ impl SqliteWorkspaceStore { archive_id: plan.archive_id.clone(), workspace_id: plan.workspace_id.clone(), source_runtime_id: plan.worker.runtime_id.clone(), - worker_id: worker_runtime::identity::WorkerId::new(worker_number), + worker_id: worker_id, expected_worker_revision: plan.worker_revision.clone(), expected_run_generation: plan.run_generation, source_created_at: worker.created_at, @@ -435,11 +437,13 @@ impl SqliteWorkspaceStore { return Ok(None); }; let prior_failure_category = plan.failure_category.clone(); - let worker_number = plan.worker.worker_id.parse::().map_err(|_| { - WorkerRetentionError::Invalid( - "Runtime Worker id is not a canonical unsigned integer".to_string(), - ) - })?; + let worker_id = plan + .worker + .worker_id + .parse::() + .map_err(|_| { + WorkerRetentionError::Invalid("Worker id must be a canonical UUIDv7".to_string()) + })?; let worker = if plan.state == WorkerRemovalPlanState::Succeeded { None } else { @@ -455,7 +459,7 @@ impl SqliteWorkspaceStore { archive_id: plan.archive_id.clone(), workspace_id: plan.workspace_id.clone(), source_runtime_id: plan.worker.runtime_id.clone(), - worker_id: worker_runtime::identity::WorkerId::new(worker_number), + worker_id: worker_id, expected_worker_revision: plan.worker_revision.clone(), expected_run_generation: plan.run_generation, source_created_at: worker @@ -544,7 +548,7 @@ impl SqliteWorkspaceStore { if plan.metadata_disposition==MetadataDisposition::Tombstone{ tx.execute("INSERT OR IGNORE INTO worker_tombstones(workspace_id,runtime_id,worker_id,display_name,profile,worker_created_at,removed_at,archive_id,policy_id,policy_revision,operation_id) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,worker.display_name,worker.profile,worker.created_at,now,plan.archive_id,plan.policy_id,plan.policy_revision,operation_id])?; } - let deleted=tx.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND runtime_worker_id=?3 AND updated_at=?4",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.worker_revision])?; + let deleted=tx.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND updated_at=?4",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.worker_revision])?; if deleted!=1{return Err(StoreError::InvalidInput(format!("stale:{}:removal fence changed",plan.plan_id)));} tx.execute("UPDATE worker_removal_operations SET state='succeeded',failure_category=NULL,updated_at=?1 WHERE operation_id=?2",params![now,operation_id])?; tx.execute("INSERT OR IGNORE INTO worker_retention_audit_events(event_id,operation_id,workspace_id,event_kind,detail,created_at) VALUES(?1,?2,?3,'worker_removed',?4,?5)",params![stable("wre",operation_id),operation_id,workspace_id,format!("runtime_id={} worker_id={} session={} metadata={} diagnostics={}",plan.worker.runtime_id,plan.worker.worker_id,sess(plan.session_disposition),meta(plan.metadata_disposition),diag(plan.diagnostics_disposition)),now])?; @@ -593,7 +597,7 @@ impl SqliteWorkspaceStore { let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; let policy_configured = load_policy(&tx, workspace_id)?.is_some(); let mut statement = tx.prepare( - "SELECT CAST(runtime_worker_id AS TEXT), retention_state + "SELECT CAST(worker_id AS TEXT), retention_state FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2", )?; let registry = statement @@ -718,7 +722,7 @@ struct WorkerRow { updated_at: String, } fn load_worker(c: &Connection, w: &str, r: &RuntimeWorkerRef) -> crate::Result> { - c.query_row("SELECT display_name,profile,retention_state,created_at,updated_at FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND runtime_worker_id=?3",params![w,r.runtime_id,r.worker_id],|x|Ok(WorkerRow{display_name:x.get(0)?,profile:x.get(1)?,retention_state:x.get(2)?,created_at:x.get(3)?,updated_at:x.get(4)?})).optional().map_err(StoreError::from) + c.query_row("SELECT display_name,profile,retention_state,created_at,updated_at FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![w,r.runtime_id,r.worker_id],|x|Ok(WorkerRow{display_name:x.get(0)?,profile:x.get(1)?,retention_state:x.get(2)?,created_at:x.get(3)?,updated_at:x.get(4)?})).optional().map_err(StoreError::from) } fn load_policy(c: &Connection, w: &str) -> crate::Result> { c.query_row( @@ -1042,16 +1046,33 @@ mod tests { use super::*; use crate::store::{ControlPlaneStore, TicketWorkerAssignmentRecord, WorkerRegistryRecord}; use worker_runtime::identity::WorkerId; + fn worker_id() -> WorkerId { + WorkerId::from_legacy_u64(1) + } fn setup() -> SqliteWorkspaceStore { let s = SqliteWorkspaceStore::in_memory().unwrap(); - s.with_conn(|c|{c.execute("INSERT INTO workspaces(workspace_id,display_name,state,created_at,updated_at)VALUES('w','W','active','t','t')",[])?;c.execute("INSERT INTO worker_registry(workspace_id,runtime_id,runtime_worker_id,display_name,profile,retention_state,created_at,updated_at)VALUES('w','r',1,'one','builtin:coder','normal','created','rev1')",[])?;Ok(())}).unwrap(); + s.with_conn(|c| { + c.execute( + "INSERT INTO workspaces(workspace_id,display_name,state,created_at,updated_at) \ + VALUES('w','W','active','t','t')", + [], + )?; + c.execute( + "INSERT INTO worker_registry(\ + workspace_id,worker_id,runtime_id,display_name,profile,retention_state,created_at,updated_at\ + ) VALUES('w',?1,'r','one','builtin:coder','normal','created','rev1')", + [worker_id().to_string()], + )?; + Ok(()) + }) + .unwrap(); s } fn inv() -> WorkerRetentionInventory { WorkerRetentionInventory { workspace_id: "w".into(), runtime_id: "r".into(), - worker_id: WorkerId::new(1), + worker_id: worker_id(), run_generation: 2, session_id: Some("s".into()), segment_ids: vec!["a".into()], @@ -1064,7 +1085,7 @@ mod tests { workspace_id: "w".into(), worker: RuntimeWorkerRef { runtime_id: "r".into(), - worker_id: "1".into(), + worker_id: worker_id().to_string(), }, expected_worker_revision: "rev1".into(), reason: "cleanup".into(), @@ -1213,7 +1234,10 @@ mod tests { SessionDisposition::Archive ); assert_eq!(prepared.runtime_request.policy_revision, 1); - assert_eq!(prepared.runtime_request.worker_id, WorkerId::new(1)); + assert_eq!( + prepared.runtime_request.worker_id, + WorkerId::from_legacy_u64(1) + ); let retry = s .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) .unwrap(); @@ -1234,7 +1258,7 @@ mod tests { operation_id: p.operation_id.clone(), input_fingerprint: p.input_fingerprint.clone(), expected_worker_revision: p.worker_revision.clone(), - worker_id: WorkerId::new(1), + worker_id: worker_id(), session_disposition: p.session_disposition, diagnostics_disposition: p.diagnostics_disposition, archive: Some(worker_runtime::retention::WorkerSessionArchiveManifest { @@ -1242,7 +1266,7 @@ mod tests { archive_id: p.archive_id.clone().unwrap(), workspace_id: "w".into(), source_runtime_id: "r".into(), - source_worker_id: WorkerId::new(1), + source_worker_id: worker_id(), source_session_id: "s".into(), segment_ids: vec!["a".into()], source_created_at: "created".into(), @@ -1284,7 +1308,23 @@ mod tests { #[test] fn assignment_and_orphan_are_authoritative() { let s = setup(); - s.with_conn(|c|{c.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,runtime_id,worker_id,assigned_by,assigned_at)VALUES('w','ticket','assignment','r','1','test','t')",[])?;c.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,runtime_id,worker_id,updated_at)VALUES('w','ticket','assignment','r','1','t')",[])?;Ok(())}).unwrap(); + s.with_conn(|c| { + let stable_worker_id = worker_id().to_string(); + c.execute( + "INSERT INTO ticket_worker_assignments(\ + workspace_id,ticket_id,assignment_id,runtime_id,worker_id,assigned_by,assigned_at\ + ) VALUES('w','ticket','assignment','r',?1,'test','t')", + [&stable_worker_id], + )?; + c.execute( + "INSERT INTO ticket_current_worker_assignments(\ + workspace_id,ticket_id,assignment_id,runtime_id,worker_id,updated_at\ + ) VALUES('w','ticket','assignment','r',?1,'t')", + [&stable_worker_id], + )?; + Ok(()) + }) + .unwrap(); let p = s.plan_worker_removal(&req(), &inv()).unwrap(); assert!( matches!(&p.blockers[..],[WorkerRemovalBlocker::CurrentAssignment{assignment_id,ticket_id}] if assignment_id=="assignment"&&ticket_id=="ticket") @@ -1292,7 +1332,7 @@ mod tests { let runtime_only = WorkerRetentionInventory { workspace_id: "w".into(), runtime_id: "r".into(), - worker_id: WorkerId::new(2), + worker_id: WorkerId::from_legacy_u64(2), run_generation: 1, session_id: Some("orphan-session".into()), segment_ids: vec![], @@ -1304,10 +1344,12 @@ mod tests { .unwrap(); assert_eq!(diagnostics.len(), 2); assert!(diagnostics.iter().any(|item| { - item.worker_id == "2" && item.category == "runtime_aggregate_without_backend_registry" + item.worker_id == WorkerId::from_legacy_u64(2).to_string() + && item.category == "runtime_aggregate_without_backend_registry" })); assert!(diagnostics.iter().any(|item| { - item.worker_id == "1" && item.category == "backend_registry_without_runtime_aggregate" + item.worker_id == worker_id().to_string() + && item.category == "backend_registry_without_runtime_aggregate" })); let count: i64 = s .with_conn(|conn| { @@ -1375,7 +1417,7 @@ mod tests { operation_id: p.operation_id.clone(), input_fingerprint: p.input_fingerprint.clone(), expected_worker_revision: p.worker_revision.clone(), - worker_id: WorkerId::new(1), + worker_id: worker_id(), session_disposition: SessionDisposition::Purge, diagnostics_disposition: DiagnosticsDisposition::Purge, archive: None, @@ -1394,7 +1436,7 @@ mod tests { operation_id: plan.operation_id.clone(), input_fingerprint: plan.input_fingerprint.clone(), expected_worker_revision: plan.worker_revision.clone(), - worker_id: WorkerId::new(1), + worker_id: worker_id(), session_disposition: plan.session_disposition, diagnostics_disposition: plan.diagnostics_disposition, archive: None, @@ -1408,15 +1450,15 @@ mod tests { store .begin_worker_removal("w", &plan.plan_id, &plan.input_fingerprint) .unwrap(); - result.worker_id = WorkerId::new(2); + result.worker_id = WorkerId::from_legacy_u64(2); assert!( store .commit_worker_removal("w", &plan.operation_id, &plan.input_fingerprint, &result) .is_err() ); let count: i64 = store.with_conn(|conn| conn.query_row( - "SELECT COUNT(*) FROM worker_registry WHERE workspace_id='w' AND runtime_id='r' AND runtime_worker_id=1", - [], + "SELECT COUNT(*) FROM worker_registry WHERE workspace_id='w' AND runtime_id='r' AND worker_id=?1", + [worker_id().to_string()], |row| row.get(0), ).map_err(StoreError::from)).unwrap(); assert_eq!(count, 1); @@ -1433,7 +1475,7 @@ mod tests { workspace_id: "w".into(), worker: RuntimeWorkerRef { runtime_id: "r".into(), - worker_id: "1".into(), + worker_id: worker_id().to_string(), }, display_name: "stale".into(), profile: None, @@ -1447,8 +1489,8 @@ mod tests { }; store.upsert_worker_registry(&stale).unwrap(); let revision: String = store.with_conn(|conn| conn.query_row( - "SELECT updated_at FROM worker_registry WHERE workspace_id='w' AND runtime_id='r' AND runtime_worker_id=1", - [], + "SELECT updated_at FROM worker_registry WHERE workspace_id='w' AND runtime_id='r' AND worker_id=?1", + [worker_id().to_string()], |row| row.get(0), ).map_err(StoreError::from)).unwrap(); assert_eq!(revision, "rev1"); @@ -1459,7 +1501,7 @@ mod tests { assignment_id: "new-assignment".into(), worker: RuntimeWorkerRef { runtime_id: "r".into(), - worker_id: "1".into(), + worker_id: worker_id().to_string(), }, assigned_by: "test".into(), assigned_at: "t".into(), @@ -1488,7 +1530,7 @@ mod tests { operation_id: prepared.plan.operation_id.clone(), input_fingerprint: prepared.plan.input_fingerprint.clone(), expected_worker_revision: prepared.plan.worker_revision.clone(), - worker_id: WorkerId::new(1), + worker_id: worker_id(), session_disposition: prepared.plan.session_disposition, diagnostics_disposition: prepared.plan.diagnostics_disposition, archive: None, @@ -1522,7 +1564,7 @@ mod tests { operation_id: prepared.plan.operation_id.clone(), input_fingerprint: prepared.plan.input_fingerprint.clone(), expected_worker_revision: prepared.plan.worker_revision.clone(), - worker_id: WorkerId::new(1), + worker_id: worker_id(), session_disposition: prepared.plan.session_disposition, diagnostics_disposition: prepared.plan.diagnostics_disposition, archive: Some(worker_runtime::retention::WorkerSessionArchiveManifest { @@ -1530,7 +1572,7 @@ mod tests { archive_id: prepared.plan.archive_id.clone().unwrap(), workspace_id: "w".into(), source_runtime_id: "r".into(), - source_worker_id: WorkerId::new(1), + source_worker_id: worker_id(), source_session_id: "s".into(), segment_ids: vec!["a".into()], source_created_at: "created".into(), diff --git a/crates/workspace-server/src/runtime_subscription_tests.rs b/crates/workspace-server/src/runtime_subscription_tests.rs index 08a93fd7..60b587a4 100644 --- a/crates/workspace-server/src/runtime_subscription_tests.rs +++ b/crates/workspace-server/src/runtime_subscription_tests.rs @@ -8,6 +8,7 @@ use worker_runtime::execution::{ WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult, WorkerExecutionRunState, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, }; +use worker_runtime::identity::WorkerId; use worker_runtime::profile_archive::{ProfileSourceArchiveRef, ProfileSourceGraphSummary}; #[derive(Debug)] @@ -57,8 +58,8 @@ const TOKEN: &str = "runtime-subscription-test-token"; fn create_request(name: &str) -> CreateWorkerRequest { CreateWorkerRequest { - idempotency_key: None, - idempotency_fingerprint: None, + worker_id: WorkerId::now_v7(), + create_fingerprint: "test-create".to_string(), profile: ProfileSelector::Builtin("builtin:companion".to_string()), display_name: Some(name.to_string()), config_bundle: None, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index e6b445b7..2b65afed 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -76,11 +76,12 @@ use crate::hosts::{ EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime, RuntimeDiagnostic, RuntimeRegistry, RuntimeRegistryError, RuntimeRegistryUnregisterResult, RuntimeSummary, TicketWorkerRole, WorkerCapabilitySummary, WorkerCompletionsRequest, - WorkerCompletionsResult, WorkerControlOperation, WorkerImplementationSummary, WorkerInputKind, - WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, WorkerLifecycleResult, - WorkerOperationState, WorkerRestoreResult, WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, - WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, WorkerSummary, - WorkerTicketAssignmentRequest, WorkerWorkspaceSummary, + WorkerCompletionsResult, WorkerControlOperation, WorkerCreateBinding, + WorkerImplementationSummary, WorkerInputKind, WorkerInputRequest, WorkerInputResult, + WorkerLifecycleRequest, WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult, + WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, + WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerTicketAssignmentRequest, + WorkerWorkspaceSummary, worker_spawn_create_fingerprint, }; use crate::identity::WorkspaceIdentity; use crate::memory_backend::execute_memory_backend_operation_with_authority; @@ -120,7 +121,7 @@ use worker_runtime::http_server::{ RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundlesResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse, }; -use worker_runtime::identity::RuntimeWorkerRef; +use worker_runtime::identity::{RuntimeWorkerRef, WorkerId}; const EMBEDDED_WORKER_RUNTIME_ID: &str = "embedded-worker-runtime"; @@ -887,7 +888,40 @@ impl WorkspaceApi { &now_registry_timestamp(), )?; } - let result = match self.runtime.spawn_worker(runtime_id, request) { + let create_fingerprint = worker_spawn_create_fingerprint(&request) + .map_err(|message| Error::Config(message.to_string()))?; + let allocation_key = request + .resolved_control_operation + .as_ref() + .map(|operation| operation.operation_id.clone()) + .or_else(|| { + request + .ticket_assignment + .as_ref() + .map(|assignment| assignment.operation_id.clone()) + }) + .unwrap_or_else(|| format!("manual:{}", WorkerId::now_v7())); + let worker_id = self + .config_store + .reserve_worker_create( + &self.config.workspace_id, + runtime_id, + &allocation_key, + &create_fingerprint, + ) + .map_err(|error| Error::RuntimeOperationFailed { + runtime_id: runtime_id.to_string(), + code: "workspace_worker_allocation_conflict".to_string(), + message: error.to_string(), + })?; + let create_binding = WorkerCreateBinding { + worker_id, + create_fingerprint, + }; + let result = match self + .runtime + .spawn_worker(runtime_id, create_binding, request) + { Ok(result) => result, Err(error) => { if let Some((workdir_id, reservation_id)) = attachment_reservation.as_ref() { @@ -911,6 +945,24 @@ impl WorkspaceApi { return Ok(result); }; let worker_ref = worker.worker.clone(); + if worker_ref.worker_id != worker_id.to_string() { + if let Some((workdir_id, reservation_id)) = attachment_reservation.as_ref() { + let _ = self.store.release_worker_workdir_attachment_reservation( + &self.config.workspace_id, + workdir_id, + reservation_id, + ); + } + return Err(Error::RuntimeOperationFailed { + runtime_id: runtime_id.to_string(), + code: "workspace_worker_identity_mismatch".to_string(), + message: format!( + "Runtime returned Worker {} for reserved Workspace Worker {}", + worker_ref.worker_id, worker_id + ), + } + .into()); + } let replacement = match self .runtime .replace_worker_workspace_api(&worker_ref, workspace_api) @@ -1017,6 +1069,9 @@ impl WorkspaceApi { return Err(error); } } + self.config_store + .complete_worker_create_reservation(&self.config.workspace_id, worker_id) + .map_err(|error| Error::Config(error.to_string()))?; Ok(result) } @@ -11889,11 +11944,11 @@ fn working_directory_request_for_browser( }) } -fn parse_runtime_worker_id_for_registry(worker_id: &str) -> ApiResult { - worker_id.parse::().map_err(|_| { +fn parse_runtime_worker_id_for_registry(worker_id: &str) -> ApiResult { + worker_id.parse::().map_err(|_| { settings_bad_request( "workspace_worker_id_invalid", - "Runtime Worker id must be an unsigned integer", + "Workspace Worker id must be a UUIDv7", ) }) } @@ -12420,6 +12475,13 @@ mod tests { ObjectiveTicketLinkRecord, SqliteWorkspaceStore, WorkspaceRecord, }; + fn test_create_binding() -> WorkerCreateBinding { + WorkerCreateBinding { + worker_id: WorkerId::now_v7(), + create_fingerprint: "sha256:test-create".to_string(), + } + } + #[test] fn reopen_confirmation_rejects_api_token_actor_before_session_resolution() { let mut headers = HeaderMap::new(); @@ -14084,6 +14146,7 @@ mod tests { .runtime .spawn_worker( EMBEDDED_WORKER_RUNTIME_ID, + test_create_binding(), WorkerSpawnRequest { requested_worker_name: Some(MEMORY_CONSOLIDATION_PROFILE.to_string()), intent: WorkerSpawnIntent::WorkspaceOrchestrator, @@ -14313,6 +14376,7 @@ mod tests { .runtime .spawn_worker( EMBEDDED_WORKER_RUNTIME_ID, + test_create_binding(), WorkerSpawnRequest { requested_worker_name: Some("notification-source".to_string()), intent: WorkerSpawnIntent::TicketRole { @@ -14533,13 +14597,21 @@ mod tests { }; let source_worker = api .runtime - .spawn_worker(EMBEDDED_WORKER_RUNTIME_ID, spawn("source-worker")) + .spawn_worker( + EMBEDDED_WORKER_RUNTIME_ID, + test_create_binding(), + spawn("source-worker"), + ) .unwrap() .worker .unwrap(); let recipient_worker = api .runtime - .spawn_worker(EMBEDDED_WORKER_RUNTIME_ID, spawn("recipient-worker")) + .spawn_worker( + EMBEDDED_WORKER_RUNTIME_ID, + test_create_binding(), + spawn("recipient-worker"), + ) .unwrap() .worker .unwrap(); @@ -14727,6 +14799,7 @@ mod tests { .runtime .spawn_worker( EMBEDDED_WORKER_RUNTIME_ID, + test_create_binding(), WorkerSpawnRequest { requested_worker_name: Some("orchestrator-source".to_string()), intent: WorkerSpawnIntent::TicketRole { @@ -15037,9 +15110,25 @@ mod tests { TEST_CREATED_AT, ) .unwrap(); + let reserved_worker_id = api + .config_store + .reserve_worker_create( + TEST_WORKSPACE_ID, + EMBEDDED_WORKER_RUNTIME_ID, + "pending-spawn-operation", + &pending_fingerprint, + ) + .unwrap(); let spawned_before_backend_failure = api .runtime - .spawn_worker(EMBEDDED_WORKER_RUNTIME_ID, pending_request.clone()) + .spawn_worker( + EMBEDDED_WORKER_RUNTIME_ID, + WorkerCreateBinding { + worker_id: reserved_worker_id, + create_fingerprint: pending_fingerprint.clone(), + }, + pending_request.clone(), + ) .unwrap() .worker .unwrap(); @@ -16632,8 +16721,8 @@ mod tests { fn runtime_create_request() -> worker_runtime::catalog::CreateWorkerRequest { let bundle = runtime_test_bundle(); worker_runtime::catalog::CreateWorkerRequest { - idempotency_key: None, - idempotency_fingerprint: None, + worker_id: WorkerId::now_v7(), + create_fingerprint: "test-create".to_string(), profile: worker_runtime::catalog::ProfileSelector::Builtin( "builtin:companion".to_string(), ), @@ -17847,6 +17936,7 @@ mod tests { .runtime .spawn_worker( "embedded-worker-runtime", + test_create_binding(), WorkerSpawnRequest { intent: WorkerSpawnIntent::TicketRole { ticket_id: "00001KVZSGT0Q".to_string(), diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index f1d59410..4b39ad2a 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -8,7 +8,7 @@ use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params}; use serde::{Deserialize, Serialize}; use uuid::Uuid; -use worker_runtime::identity::RuntimeWorkerRef; +use worker_runtime::identity::{RuntimeWorkerRef, WorkerId}; use crate::{Error, Result}; @@ -201,6 +201,11 @@ const MIGRATIONS: &[Migration] = &[ name: "add Objective query indexes", apply: add_objective_query_indexes, }, + Migration { + version: 37, + name: "promote Workspace Worker UUIDv7 identity", + apply: promote_workspace_worker_uuid_identity, + }, ]; struct Migration { @@ -968,6 +973,95 @@ impl SqliteWorkspaceStore { f(&mut conn) } + pub(crate) fn reserve_worker_create( + &self, + workspace_id: &str, + runtime_id: &str, + allocation_key: &str, + create_fingerprint: &str, + ) -> Result { + if allocation_key.trim().is_empty() || create_fingerprint.trim().is_empty() { + return Err(Error::InvalidInput( + "Worker create allocation key and fingerprint must be non-empty".to_string(), + )); + } + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let existing = tx + .query_row( + "SELECT worker_id, runtime_id, create_fingerprint \ + FROM worker_create_reservations \ + WHERE workspace_id = ?1 AND allocation_key = ?2", + params![workspace_id, allocation_key], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + )) + }, + ) + .optional()?; + if let Some((worker_id, reserved_runtime_id, reserved_fingerprint)) = existing { + if reserved_runtime_id != runtime_id || reserved_fingerprint != create_fingerprint { + return Err(Error::InvalidInput(format!( + "Worker create allocation `{allocation_key}` was already used with different input" + ))); + } + return worker_id.parse::().map_err(|_| { + Error::Store(format!( + "Worker create allocation `{allocation_key}` has a non-UUIDv7 worker id" + )) + }); + } + + let worker_id = WorkerId::now_v7(); + let now = chrono::Utc::now().to_rfc3339(); + tx.execute( + "INSERT INTO worker_create_reservations(\ + workspace_id, allocation_key, worker_id, runtime_id, create_fingerprint,\ + state, created_at, updated_at\ + ) VALUES (?1, ?2, ?3, ?4, ?5, 'reserved', ?6, ?6)", + params![ + workspace_id, + allocation_key, + worker_id.to_string(), + runtime_id, + create_fingerprint, + now + ], + )?; + tx.commit()?; + Ok(worker_id) + }) + } + + pub(crate) fn complete_worker_create_reservation( + &self, + workspace_id: &str, + worker_id: WorkerId, + ) -> Result<()> { + self.with_conn(|conn| { + let changed = conn.execute( + "UPDATE worker_create_reservations \ + SET state = 'created', updated_at = ?3 \ + WHERE workspace_id = ?1 AND worker_id = ?2", + params![ + workspace_id, + worker_id.to_string(), + chrono::Utc::now().to_rfc3339() + ], + )?; + if changed != 1 { + return Err(Error::Store(format!( + "Worker create reservation {} was not found", + worker_id + ))); + } + Ok(()) + }) + } + fn materialize_workspace_config(&self, workspace_id: &str, created_at: &str) -> Result<()> { self.with_conn_mut(|conn| { let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; @@ -2276,7 +2370,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { "SELECT EXISTS( SELECT 1 FROM worker_removal_operations WHERE workspace_id = ?1 AND runtime_id = ?2 - AND CAST(worker_id AS INTEGER) = ?3 + AND worker_id = ?3 AND state IN ('executing', 'failed', 'succeeded') )", params![ @@ -2291,11 +2385,12 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } conn.execute( r#"INSERT INTO worker_registry ( - workspace_id, runtime_id, runtime_worker_id, display_name, profile, + workspace_id, runtime_id, worker_id, display_name, profile, retention_state, transcript_ref, session_ref, summary_ref, diagnostics_ref, created_at, updated_at ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) - ON CONFLICT(workspace_id, runtime_id, runtime_worker_id) DO UPDATE SET + ON CONFLICT(workspace_id, worker_id) DO UPDATE SET + runtime_id = excluded.runtime_id, display_name = excluded.display_name, profile = excluded.profile, retention_state = CASE @@ -2312,7 +2407,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { SELECT 1 FROM worker_removal_operations retention WHERE retention.workspace_id = excluded.workspace_id AND retention.runtime_id = excluded.runtime_id - AND CAST(retention.worker_id AS INTEGER) = excluded.runtime_worker_id + AND retention.worker_id = excluded.worker_id AND retention.state IN ('executing', 'failed', 'succeeded') )"#, params![ @@ -2342,7 +2437,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { self.with_conn(|conn| { conn.query_row( worker_registry_select_sql( - "WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3", + "WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", ) .as_str(), params![workspace_id, worker.runtime_id, worker.worker_id], @@ -2383,11 +2478,11 @@ impl ControlPlaneStore for SqliteWorkspaceStore { let changed = conn.execute( r#"UPDATE worker_registry SET retention_state = ?4, updated_at = ?5 - WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3 + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND NOT EXISTS ( SELECT 1 FROM worker_removal_operations retention WHERE retention.workspace_id = ?1 AND retention.runtime_id = ?2 - AND CAST(retention.worker_id AS INTEGER) = ?3 + AND retention.worker_id = ?3 AND retention.state IN ('executing', 'failed') )"#, params![ @@ -2412,11 +2507,11 @@ impl ControlPlaneStore for SqliteWorkspaceStore { tx.execute( r#"UPDATE worker_workdir_links SET unlinked_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') - WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3 AND unlinked_at IS NULL"#, + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL"#, params![workspace_id, worker.runtime_id, worker.worker_id], )?; let changed = tx.execute( - "DELETE FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3", + "DELETE FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", params![workspace_id, worker.runtime_id, worker.worker_id], )?; tx.commit()?; @@ -2440,7 +2535,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) ON CONFLICT ( workspace_id, - controller_runtime_id, controller_worker_id, operation_id ) DO NOTHING"#, @@ -3281,9 +3375,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } tx.execute( r#"INSERT INTO worker_workdir_links ( - workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at + workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL) - ON CONFLICT(workspace_id, runtime_id, runtime_worker_id, workdir_id, role) DO UPDATE SET + ON CONFLICT(workspace_id, worker_id, workdir_id, role) DO UPDATE SET linked_at = excluded.linked_at, unlinked_at = NULL"#, params![ @@ -3365,9 +3459,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } let active_for_worker = tx .query_row( - r#"SELECT workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at FROM worker_workdir_links - WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3 AND unlinked_at IS NULL"#, + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL"#, params![ record.workspace_id, record.worker.runtime_id, @@ -3388,7 +3482,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } let active_for_workdir = tx .query_row( - r#"SELECT workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at FROM worker_workdir_links WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL"#, params![record.workspace_id, record.workdir_id], @@ -3403,9 +3497,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } let write = tx.execute( r#"INSERT INTO worker_workdir_links ( - workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at + workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL) - ON CONFLICT(workspace_id, runtime_id, runtime_worker_id, workdir_id, role) DO UPDATE SET + ON CONFLICT(workspace_id, worker_id, workdir_id, role) DO UPDATE SET linked_at = excluded.linked_at, unlinked_at = NULL"#, params![ @@ -3445,9 +3539,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore { )?; let active = tx .query_row( - r#"SELECT workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at FROM worker_workdir_links - WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3 AND unlinked_at IS NULL"#, + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL"#, params![workspace_id, worker.runtime_id, worker.worker_id], read_worker_workdir_link_record, ) @@ -3467,7 +3561,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { let changed = tx.execute( r#"UPDATE worker_workdir_links SET unlinked_at = ?4 - WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3 AND unlinked_at IS NULL"#, + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL"#, params![workspace_id, worker.runtime_id, worker.worker_id, unlinked_at], )?; if changed != 1 { @@ -3493,7 +3587,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { let exists = conn.query_row( r#"SELECT EXISTS( SELECT 1 FROM worker_workdir_links - WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3 + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 )"#, params![workspace_id, worker.runtime_id, worker.worker_id], |row| row.get(0), @@ -3509,9 +3603,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ) -> Result> { self.with_conn(|conn| { let mut stmt = conn.prepare( - r#"SELECT workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at FROM worker_workdir_links - WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3 AND unlinked_at IS NULL + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL ORDER BY linked_at DESC"#, )?; let rows = stmt.query_map( @@ -3530,7 +3624,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ) -> Result> { self.with_conn(|conn| { let mut stmt = conn.prepare( - r#"SELECT workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at FROM worker_workdir_links WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL ORDER BY linked_at DESC"#, @@ -3797,7 +3891,7 @@ fn read_worker_workdir_link_record( ) -> rusqlite::Result { Ok(WorkerWorkdirLinkRecord { workspace_id: row.get(0)?, - worker: RuntimeWorkerRef::new(row.get::<_, String>(1)?, row.get::<_, u64>(2)?.to_string()), + worker: RuntimeWorkerRef::new(row.get::<_, String>(1)?, row.get::<_, String>(2)?), workdir_id: row.get(3)?, role: row.get(4)?, linked_at: row.get(5)?, @@ -3880,7 +3974,7 @@ fn read_memory_staging_resolution_record( fn worker_registry_select_sql(where_clause: &str) -> String { format!( - "SELECT workspace_id, runtime_id, runtime_worker_id, display_name, profile, \ + "SELECT workspace_id, runtime_id, worker_id, display_name, profile, \ retention_state, transcript_ref, session_ref, summary_ref, diagnostics_ref, \ created_at, updated_at FROM worker_registry {where_clause}" ) @@ -3889,7 +3983,7 @@ fn worker_registry_select_sql(where_clause: &str) -> String { fn read_worker_registry_record(row: &rusqlite::Row<'_>) -> rusqlite::Result { Ok(WorkerRegistryRecord { workspace_id: row.get(0)?, - worker: RuntimeWorkerRef::new(row.get::<_, String>(1)?, row.get::<_, u64>(2)?.to_string()), + worker: RuntimeWorkerRef::new(row.get::<_, String>(1)?, row.get::<_, String>(2)?), display_name: row.get(3)?, profile: row.get(4)?, retention_state: row.get(5)?, @@ -3912,11 +4006,8 @@ fn read_worker_control_grant_record( Ok(WorkerControlGrantRecord { workspace_id: row.get(0)?, grant_id: row.get(1)?, - controller: RuntimeWorkerRef::new( - row.get::<_, String>(2)?, - row.get::<_, u64>(3)?.to_string(), - ), - subject: RuntimeWorkerRef::new(row.get::<_, String>(4)?, row.get::<_, u64>(5)?.to_string()), + controller: RuntimeWorkerRef::new(row.get::<_, String>(2)?, row.get::<_, String>(3)?), + subject: RuntimeWorkerRef::new(row.get::<_, String>(4)?, row.get::<_, String>(5)?), relation: row.get(6)?, origin: row.get(7)?, permissions, @@ -5069,6 +5160,201 @@ fn add_objective_query_indexes(conn: &Connection) -> Result<()> { Ok(()) } +fn promote_workspace_worker_uuid_identity(conn: &Connection) -> Result<()> { + conn.execute_batch( + r#" + PRAGMA defer_foreign_keys = ON; + CREATE TEMP TABLE worker_identity_v37 ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + runtime_worker_id INTEGER NOT NULL, + worker_id TEXT NOT NULL UNIQUE, + PRIMARY KEY (workspace_id, runtime_id, runtime_worker_id) + ); + "#, + )?; + + let legacy_workers = { + let mut statement = conn.prepare( + "SELECT workspace_id, runtime_id, runtime_worker_id FROM worker_registry \ + ORDER BY workspace_id, runtime_id, runtime_worker_id", + )?; + statement + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, i64>(2)?, + )) + })? + .collect::, _>>()? + }; + for (workspace_id, runtime_id, runtime_worker_id) in legacy_workers { + conn.execute( + "INSERT INTO worker_identity_v37(\ + workspace_id, runtime_id, runtime_worker_id, worker_id\ + ) VALUES (?1, ?2, ?3, ?4)", + params![ + workspace_id, + runtime_id, + runtime_worker_id, + WorkerId::from_legacy_binding( + &workspace_id, + &runtime_id, + u64::try_from(runtime_worker_id).map_err(|_| { + Error::InvalidInput(format!( + "legacy Runtime Worker id {runtime_worker_id} is negative" + )) + })?, + ) + .to_string() + ], + )?; + } + + conn.execute_batch( + r#" + CREATE TABLE worker_registry_v37 ( + workspace_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + display_name TEXT NOT NULL, + profile TEXT, + retention_state TEXT NOT NULL CHECK (retention_state IN ('normal', 'pinned')), + transcript_ref TEXT, + session_ref TEXT, + summary_ref TEXT, + diagnostics_ref TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, worker_id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE + ); + INSERT INTO worker_registry_v37( + workspace_id, worker_id, runtime_id, display_name, profile, + retention_state, transcript_ref, session_ref, summary_ref, + diagnostics_ref, created_at, updated_at + ) + SELECT r.workspace_id, m.worker_id, r.runtime_id, r.display_name, r.profile, + r.retention_state, r.transcript_ref, r.session_ref, r.summary_ref, + r.diagnostics_ref, r.created_at, r.updated_at + FROM worker_registry r + JOIN worker_identity_v37 m + ON m.workspace_id = r.workspace_id + AND m.runtime_id = r.runtime_id + AND m.runtime_worker_id = r.runtime_worker_id; + + CREATE TABLE worker_workdir_links_v37 ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + workdir_id TEXT NOT NULL, + role TEXT NOT NULL, + linked_at TEXT NOT NULL, + unlinked_at TEXT, + PRIMARY KEY (workspace_id, worker_id, workdir_id, role), + FOREIGN KEY (workspace_id, worker_id) + REFERENCES worker_registry_v37(workspace_id, worker_id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, workdir_id) + REFERENCES workdir_registry(workspace_id, workdir_id) ON DELETE CASCADE + ); + INSERT INTO worker_workdir_links_v37( + workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at + ) + SELECT l.workspace_id, l.runtime_id, m.worker_id, l.workdir_id, l.role, l.linked_at, l.unlinked_at + FROM worker_workdir_links l + JOIN worker_identity_v37 m + ON m.workspace_id = l.workspace_id + AND m.runtime_id = l.runtime_id + AND m.runtime_worker_id = l.runtime_worker_id; + + CREATE TABLE worker_control_grants_v37 ( + grant_id TEXT NOT NULL, + workspace_id TEXT NOT NULL, + controller_runtime_id TEXT NOT NULL, + controller_worker_id TEXT NOT NULL, + subject_runtime_id TEXT NOT NULL, + subject_worker_id TEXT NOT NULL, + relation TEXT NOT NULL, + origin TEXT NOT NULL, + permissions_json TEXT NOT NULL, + operation_id TEXT NOT NULL, + created_at TEXT NOT NULL, + revoked_at TEXT, + PRIMARY KEY (workspace_id, grant_id), + UNIQUE (workspace_id, controller_worker_id, operation_id), + FOREIGN KEY (workspace_id, controller_worker_id) + REFERENCES worker_registry_v37(workspace_id, worker_id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, subject_worker_id) + REFERENCES worker_registry_v37(workspace_id, worker_id) ON DELETE CASCADE + ); + INSERT INTO worker_control_grants_v37( + grant_id, workspace_id, + controller_runtime_id, controller_worker_id, + subject_runtime_id, subject_worker_id, + relation, origin, permissions_json, operation_id, + created_at, revoked_at + ) + SELECT g.grant_id, g.workspace_id, + g.controller_runtime_id, controller.worker_id, + g.subject_runtime_id, subject.worker_id, + g.relation, g.origin, g.permissions_json, g.operation_id, + g.created_at, g.revoked_at + FROM worker_control_grants g + JOIN worker_identity_v37 controller + ON controller.workspace_id = g.workspace_id + AND controller.runtime_id = g.controller_runtime_id + AND controller.runtime_worker_id = g.controller_worker_id + JOIN worker_identity_v37 subject + ON subject.workspace_id = g.workspace_id + AND subject.runtime_id = g.subject_runtime_id + AND subject.runtime_worker_id = g.subject_worker_id; + + DROP TABLE worker_control_grants; + DROP TABLE worker_workdir_links; + DROP TABLE worker_registry; + ALTER TABLE worker_registry_v37 RENAME TO worker_registry; + ALTER TABLE worker_workdir_links_v37 RENAME TO worker_workdir_links; + ALTER TABLE worker_control_grants_v37 RENAME TO worker_control_grants; + CREATE INDEX worker_registry_runtime + ON worker_registry(workspace_id, runtime_id, worker_id); + CREATE INDEX worker_workdir_links_workdir + ON worker_workdir_links(workspace_id, workdir_id); + CREATE UNIQUE INDEX worker_workdir_links_active_worker_unique + ON worker_workdir_links(workspace_id, worker_id) + WHERE unlinked_at IS NULL; + CREATE UNIQUE INDEX worker_workdir_links_active_workdir_unique + ON worker_workdir_links(workspace_id, workdir_id) + WHERE unlinked_at IS NULL; + CREATE INDEX worker_control_grants_controller + ON worker_control_grants( + workspace_id, controller_worker_id, revoked_at + ); + CREATE INDEX worker_control_grants_subject + ON worker_control_grants( + workspace_id, subject_worker_id, revoked_at + ); + CREATE TABLE worker_create_reservations ( + workspace_id TEXT NOT NULL, + allocation_key TEXT NOT NULL, + worker_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + create_fingerprint TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('reserved', 'created')), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, allocation_key), + UNIQUE (workspace_id, worker_id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE + ); + CREATE INDEX worker_create_reservations_worker + ON worker_create_reservations(workspace_id, worker_id); + DROP TABLE worker_identity_v37; + "#, + )?; + Ok(()) +} + fn remove_worker_control_delegation_authority(conn: &Connection) -> Result<()> { let mut statement = conn.prepare("SELECT workspace_id, grant_id, permissions_json FROM worker_control_grants")?; @@ -5822,8 +6108,53 @@ INSERT INTO worker_control_grants ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 36); + assert_eq!(current_schema_version(&conn).unwrap(), 37); assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap()); + let controller_worker_id: String = conn + .query_row( + "SELECT worker_id FROM worker_registry WHERE display_name = 'Controller'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!( + controller_worker_id, + WorkerId::from_legacy_binding("workspace-a", "runtime-a", 1).to_string() + ); + let (grant_controller, grant_subject): (String, String) = conn + .query_row( + "SELECT controller_worker_id, subject_worker_id \ + FROM worker_control_grants WHERE grant_id = 'spawned'", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!(grant_controller, controller_worker_id); + assert_eq!( + grant_subject, + WorkerId::from_legacy_binding("workspace-a", "runtime-a", 2).to_string() + ); + let worker_registry_pk = { + let mut statement = conn.prepare("PRAGMA table_info(worker_registry)").unwrap(); + statement + .query_map([], |row| { + Ok((row.get::<_, String>(1)?, row.get::<_, i64>(5)?)) + }) + .unwrap() + .filter_map(|row| { + let (name, position) = row.unwrap(); + (position > 0).then_some((position, name)) + }) + .collect::>() + }; + assert_eq!( + worker_registry_pk, + vec![ + (1, "workspace_id".to_string()), + (2, "worker_id".to_string()) + ] + ); + let (permissions_json, revoked_at): (String, Option) = conn .query_row( "SELECT permissions_json, revoked_at FROM worker_control_grants WHERE grant_id = 'spawned'", @@ -5870,7 +6201,7 @@ INSERT INTO worker_control_grants ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 36); + assert_eq!(current_schema_version(&conn).unwrap(), 37); assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap()); } @@ -5903,7 +6234,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 36); + assert_eq!(current_schema_version(&conn).unwrap(), 37); assert!(table_exists(&conn, "flow_sources").unwrap()); assert!(table_exists(&conn, "flow_source_revisions").unwrap()); assert!(!table_exists(&conn, "flow_instances").unwrap()); @@ -5970,7 +6301,7 @@ INSERT INTO worker_workdir_attachment_reservations ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 36); + assert_eq!(current_schema_version(&conn).unwrap(), 37); let repositories_sql: String = conn .query_row( "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'", @@ -6150,7 +6481,7 @@ INSERT INTO workdir_registry ( let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 36); + assert_eq!(store.schema_version().await.unwrap(), 37); assert!( !store .with_conn(|conn| table_exists(conn, "worker_workspace_credentials")) @@ -6167,13 +6498,64 @@ INSERT INTO workdir_registry ( store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 36); + assert_eq!(reopened.schema_version().await.unwrap(), 37); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) ); } + #[tokio::test] + async fn worker_create_reservation_allocates_uuid_before_runtime_and_replays_exact_input() { + let dir = tempfile::tempdir().unwrap(); + let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap(); + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: "workspace-a".to_string(), + owner_account_id: None, + display_name: "Workspace A".to_string(), + state: "active".to_string(), + created_at: "2026-08-06T00:00:00Z".to_string(), + updated_at: "2026-08-06T00:00:00Z".to_string(), + }) + .await + .unwrap(); + + let reserved = store + .reserve_worker_create("workspace-a", "arcadia", "operation-1", "sha256:one") + .unwrap(); + assert_eq!( + reserved.as_uuid().get_version(), + Some(uuid::Version::SortRand) + ); + assert_eq!( + store + .reserve_worker_create("workspace-a", "arcadia", "operation-1", "sha256:one") + .unwrap(), + reserved + ); + assert!( + store + .reserve_worker_create("workspace-a", "arcadia", "operation-1", "sha256:different") + .is_err() + ); + store + .complete_worker_create_reservation("workspace-a", reserved) + .unwrap(); + let state: String = store + .with_conn(|conn| { + conn.query_row( + "SELECT state FROM worker_create_reservations \ + WHERE workspace_id = 'workspace-a' AND worker_id = ?1", + [reserved.to_string()], + |row| row.get(0), + ) + .map_err(Error::from) + }) + .unwrap(); + assert_eq!(state, "created"); + } + #[tokio::test] async fn workspace_flow_sources_keep_revisions_and_builtins_stay_resources() { let dir = tempfile::tempdir().unwrap(); @@ -6528,6 +6910,7 @@ INSERT INTO workdir_registry ( "artifacts", "audit_events", "worker_registry", + "worker_create_reservations", "ticket_worker_assignments", "ticket_current_worker_assignments", "ticket_worker_assignment_events", @@ -6607,8 +6990,8 @@ INSERT INTO workdir_registry ( "worker_registry", [ "workspace_id", + "worker_id", "runtime_id", - "runtime_worker_id", "display_name", "profile", "retention_state", @@ -6714,7 +7097,7 @@ INSERT INTO workdir_registry ( .unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 36); + assert_eq!(store.schema_version().await.unwrap(), 37); store .with_conn(|conn| { @@ -6903,7 +7286,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn repository_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 36); + assert_eq!(store.schema_version().await.unwrap(), 37); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -6969,7 +7352,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn memory_authority_records_round_trip_and_close_staging() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 36); + assert_eq!(store.schema_version().await.unwrap(), 37); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -7360,7 +7743,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn account_and_login_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 36); + assert_eq!(store.schema_version().await.unwrap(), 37); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { account_id: "acct-user-alice".to_string(),