diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index 3acf5a36f..d3a0fed54 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -878,6 +878,7 @@ mod tests { let create = BackendCreateWorkerRequest { runtime_id: "embedded".to_string(), display_name: "Coder one".to_string(), + singleton_key: None, profile: Some("builtin:coder".to_string()), ticket_assignment: None, initial_submit: Vec::new(), diff --git a/crates/client/src/workspace_product.rs b/crates/client/src/workspace_product.rs index 6426efafd..0cacb0273 100644 --- a/crates/client/src/workspace_product.rs +++ b/crates/client/src/workspace_product.rs @@ -431,6 +431,7 @@ impl BackendWorkspaceProductClient { let request = CreateWorkspaceWorkerRequest { runtime_id: runtime.runtime_id.clone(), display_name: format!("intake-{ticket_id}"), + singleton_key: None, profile: Some("builtin:intake".to_string()), ticket_assignment: None, initial_submit: vec![protocol::Segment::Text { diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index bff7dbe1e..7bed30902 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -6859,6 +6859,9 @@ pub struct RuntimeWorkerSpawnRequest { pub intent: RuntimeWorkerSpawnIntent, #[serde(default, skip_serializing_if = "Option::is_none")] pub requested_worker_name: Option, + /// Opaque Workspace-scoped singleton identity. The Backend, not the Runtime, owns this lease. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub singleton_key: Option, pub acceptance: RuntimeWorkerSpawnAcceptanceRequirement, pub profile: RuntimeProfileSelector, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -8230,6 +8233,9 @@ pub struct CreateWorkspaceWorkerTicketAssignmentRequest { pub struct CreateWorkspaceWorkerRequest { pub runtime_id: String, pub display_name: String, + /// Opaque Workspace-scoped singleton identity. Possession does not grant Worker access. + #[serde(default)] + pub singleton_key: Option, #[serde(default)] pub profile: Option, #[serde(default)] @@ -10372,6 +10378,7 @@ mod tests { let request = serde_json::to_value(CreateWorkspaceWorkerRequest { runtime_id: "runtime-a".to_string(), display_name: "Worker A".to_string(), + singleton_key: Some("subjektiv:subject-a".to_string()), profile: None, ticket_assignment: None, initial_submit: Vec::new(), @@ -10384,6 +10391,7 @@ mod tests { serde_json::json!({ "runtime_id": "runtime-a", "display_name": "Worker A", + "singleton_key": "subjektiv:subject-a", "profile": null, "ticket_assignment": null, "initial_submit": [], diff --git a/crates/tui/src/backend_spawn.rs b/crates/tui/src/backend_spawn.rs index f6d440d62..40d4da58c 100644 --- a/crates/tui/src/backend_spawn.rs +++ b/crates/tui/src/backend_spawn.rs @@ -162,6 +162,7 @@ fn request_from_selection(selection: Selection) -> BackendCreateWorkerRequest { BackendCreateWorkerRequest { runtime_id: selection.runtime_id, display_name: selection.display_name, + singleton_key: None, profile: Some(selection.profile), initial_submit: Vec::new(), workdir_attachments: Vec::new(), diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index eab20bc41..1e2b4d964 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -474,6 +474,7 @@ pub struct WorkerLifecycleSpawnRequest { /// Ordered alias-keyed selections forwarded as canonical Workdir attachments. pub workdir_attachments: Vec, pub profile: String, + pub singleton_key: Option, pub ticket_id: Option, pub operation_id: Option, pub display_name: String, @@ -508,6 +509,7 @@ fn workspace_worker_create_request( Ok(CreateWorkspaceWorkerRequest { runtime_id: request.runtime_id, display_name: request.display_name, + singleton_key: request.singleton_key, profile: Some(request.profile), ticket_assignment, initial_submit: request.initial_submit, @@ -716,6 +718,9 @@ struct WorkerSpawnInput { /// under the stable `workdir` alias. working_directory_id: String, profile: String, + /// Optional opaque Workspace singleton key. Prefixes are caller-owned and uninterpreted. + #[serde(default)] + singleton_key: Option, /// Optional inprogress Ticket already accepted by the Orchestrator. Set /// this with one Flow segment to assign the new Coder atomically. #[serde(default)] @@ -910,6 +915,7 @@ impl Tool for WorkspaceWorkerTool { .transpose()?, }], profile: non_empty(input.profile, "profile")?, + singleton_key: input.singleton_key, ticket_id, operation_id, display_name: input @@ -1751,6 +1757,7 @@ mod tests { }, ], profile: "builtin:coder".to_string(), + singleton_key: Some("subjektiv:subject-1".to_string()), ticket_id: None, operation_id: None, display_name: "Coder".to_string(), diff --git a/crates/worker/src/feature/builtin/orchestration.rs b/crates/worker/src/feature/builtin/orchestration.rs index 953284271..515764bb2 100644 --- a/crates/worker/src/feature/builtin/orchestration.rs +++ b/crates/worker/src/feature/builtin/orchestration.rs @@ -127,6 +127,7 @@ impl Tool for SpawnTicketCoderTool { runtime_id, workdir_attachments, profile: CODER_PROFILE.to_string(), + singleton_key: None, ticket_id: Some(ticket.id.clone()), operation_id: Some(format!("spawn-ticket-coder:{}:{call_id}", ticket.id)), display_name: format!("Coder · {}", ticket.resource_key), diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 01f956d74..0aeb1d9c6 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -507,6 +507,9 @@ pub struct WorkerSpawnRequest { pub intent: WorkerSpawnIntent, #[serde(skip_serializing_if = "Option::is_none")] pub requested_worker_name: Option, + /// Opaque Backend-owned singleton key. Runtime providers must not infer semantics from it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub singleton_key: Option, pub acceptance: WorkerSpawnAcceptanceRequirement, pub profile: ProfileSelector, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -6911,6 +6914,7 @@ mod tests { role: TicketWorkerRole::Coder, }, requested_worker_name: None, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -7198,6 +7202,7 @@ mod tests { role: TicketWorkerRole::Coder, }, requested_worker_name: Some("friendly-name-is-not-authority".to_string()), + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -7299,6 +7304,7 @@ mod tests { role: TicketWorkerRole::Coder, }, requested_worker_name: None, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -7340,6 +7346,7 @@ mod tests { WorkerSpawnRequest { intent: WorkerSpawnIntent::WorkspaceCompanion, requested_worker_name: None, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::SocketReady, profile: ProfileSelector::Builtin("builtin:companion".to_string()), ticket_assignment: None, diff --git a/crates/workspace-server/src/latest_schema.sql b/crates/workspace-server/src/latest_schema.sql index faa3d7d30..f5f021c53 100644 --- a/crates/workspace-server/src/latest_schema.sql +++ b/crates/workspace-server/src/latest_schema.sql @@ -768,8 +768,37 @@ CREATE TABLE worker_create_reservations ( state TEXT NOT NULL CHECK (state IN ('reserved', 'created', 'removed')), created_at TEXT NOT NULL, updated_at TEXT NOT NULL, request_fingerprint TEXT, memory_settings_revision INTEGER, memory_language TEXT, + singleton_key TEXT, + singleton_generation INTEGER CHECK (singleton_generation IS NULL OR singleton_generation > 0), PRIMARY KEY (workspace_id, allocation_key), UNIQUE (workspace_id, worker_id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE, + CHECK ((singleton_key IS NULL) = (singleton_generation IS NULL)) + ); +CREATE TRIGGER worker_create_reservation_singleton_lease_insert +BEFORE INSERT ON worker_create_reservations +WHEN (NEW.singleton_key IS NULL AND NEW.singleton_generation IS NOT NULL) + OR (NEW.singleton_key IS NOT NULL AND NEW.singleton_generation IS NULL) +BEGIN + SELECT RAISE(ABORT, 'worker_singleton_lease_incomplete'); +END; +CREATE TRIGGER worker_create_reservation_singleton_lease_update +BEFORE UPDATE OF singleton_key, singleton_generation ON worker_create_reservations +WHEN (NEW.singleton_key IS NULL AND NEW.singleton_generation IS NOT NULL) + OR (NEW.singleton_key IS NOT NULL AND NEW.singleton_generation IS NULL) +BEGIN + SELECT RAISE(ABORT, 'worker_singleton_lease_incomplete'); +END; +CREATE TABLE worker_singleton_owners ( + workspace_id TEXT NOT NULL, + singleton_key TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + generation INTEGER NOT NULL CHECK (generation > 0), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, singleton_key), + UNIQUE (workspace_id, worker_id), FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE ); CREATE TABLE worker_diagnostics_archives ( diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index 7b4cc6869..9847d8b5d 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -471,7 +471,7 @@ impl SqliteWorkspaceStore { } tx.execute("UPDATE worker_workdir_links SET unlinked_at=?4 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND unlinked_at IS NULL",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,now])?; 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])?; - tx.execute("UPDATE worker_create_reservations SET state='removed',updated_at=?4 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state='created'",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,now])?; + tx.execute("UPDATE worker_create_reservations SET state='removed',updated_at=?4 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state IN ('reserved','created')",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,now])?; if deleted!=1{return Err(StoreError::InvalidInput(format!("stale:{}:removal fence changed",plan.plan_id)));} let catalog=commit_worker_catalog_removal(&tx,workspace_id,&plan.worker)?; tx.execute("UPDATE worker_removal_operations SET state='succeeded',failure_category=NULL,updated_at=?1 WHERE operation_id=?2",params![now,operation_id])?; diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index c064f5a29..f1a2bcbe1 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -3352,6 +3352,7 @@ impl WorkspaceApi { EMBEDDED_WORKER_RUNTIME_ID, WorkerSpawnRequest { requested_worker_name: Some(format!("job:{}", request.purpose)), + singleton_key: None, intent: WorkerSpawnIntent::BackendJob { job_id: request.job_id.clone(), attempt_id: reservation.attempt.attempt_id.clone(), @@ -4060,6 +4061,9 @@ impl WorkspaceApi { runtime_id: &str, mut request: WorkerSpawnRequest, ) -> ApiResult { + if let Some(singleton_key) = request.singleton_key.as_deref() { + crate::store::validate_worker_singleton_key(singleton_key)?; + } self.validate_worker_spawn_repository_scope(runtime_id, &mut request)?; let workspace_api = self.workspace_api_ref(runtime_id); request.resolved_workspace_api = Some(workspace_api.clone()); @@ -4099,6 +4103,7 @@ impl WorkspaceApi { .collect::>(); let request_fingerprint = worker_spawn_create_fingerprint(&request) .map_err(|message| Error::Config(message.to_string()))?; + let singleton_key = request.singleton_key.clone(); let current_memory_settings = self .config_store .get_workspace_memory_settings(&self.config.workspace_id)?; @@ -4113,20 +4118,28 @@ impl WorkspaceApi { .map(|assignment| assignment.operation_id.clone()) }) .unwrap_or_else(|| format!("manual:{}", WorkerId::now_v7())); - let reservation = self - .config_store - .reserve_worker_create( - &self.config.workspace_id, - runtime_id, - &allocation_key, - &request_fingerprint, - ¤t_memory_settings, - ) - .map_err(|error| Error::RuntimeOperationFailed { - runtime_id: runtime_id.to_string(), - code: "workspace_worker_allocation_conflict".to_string(), - message: error.to_string(), - })?; + let reservation = match self.config_store.reserve_worker_create( + &self.config.workspace_id, + runtime_id, + &allocation_key, + &request_fingerprint, + singleton_key.as_deref(), + ¤t_memory_settings, + ) { + Ok(reservation) => reservation, + Err(error) => { + let error = match error { + Error::RepositoryConflict(message) => Error::RepositoryConflict(message), + error => Error::RuntimeOperationFailed { + runtime_id: runtime_id.to_string(), + code: "workspace_worker_allocation_conflict".to_string(), + message: error.to_string(), + }, + }; + let diagnostics = cleanup_spawn_created_workdirs(self, &spawned_workdir_ids); + return Err(ApiError::with_diagnostics(error, diagnostics)); + } + }; let worker_id = reservation.worker_id; request.resolved_memory_settings = Some(reservation.memory_settings); let reservation_fingerprint = reservation.create_fingerprint.clone(); @@ -4189,6 +4202,35 @@ impl WorkspaceApi { { Ok(result) => result, Err(error) => { + if matches!(&error, RuntimeRegistryError::UnknownRuntime(_)) { + // Registry lookup failed before dispatch, so no Runtime could have created this + // Worker. Terminalize the reservation immediately instead of retaining an + // uncertain singleton generation forever. + let mut diagnostics = + release_worker_workdir_attachment_reservations(self, &reserved_attachments); + if let Err(cleanup_error) = self.config_store.fail_worker_create_reservation( + &self.config.workspace_id, + runtime_id, + worker_id, + &reservation_fingerprint, + ) { + diagnostics.push(spawn_compensation_diagnostic( + "worker_spawn_compensation_create_reservation_release_failed", + format!( + "Failed to terminalize undispatched Worker create reservation {}: {}", + worker_id, + sanitize_backend_error(&cleanup_error.to_string()) + ), + )); + } + write_workspace_worker_create_failure( + runtime_id, + worker_id, + "runtime_spawn_undispatched", + &diagnostics, + ); + return Err(ApiError::with_diagnostics(error.into_error(), diagnostics)); + } let diagnostics = compensate_failed_workspace_worker_create( self, runtime_id, @@ -4202,7 +4244,7 @@ impl WorkspaceApi { return Err(ApiError::with_diagnostics(error.into_error(), diagnostics)); } }; - let Some(worker) = result.worker.as_ref() else { + let Some(worker) = result.worker.as_mut() else { result .diagnostics .extend(compensate_failed_workspace_worker_create( @@ -4217,6 +4259,12 @@ impl WorkspaceApi { )); return Ok(result); }; + // Runtime metadata is display-only. The lease returned by the same DB transaction that + // reserved creation is the sole authority for the projected singleton key. + worker.singleton_key = reservation + .singleton + .as_ref() + .map(|lease| lease.key.clone()); let worker_ref = worker.worker.clone(); if worker_ref.worker_id != worker_id.to_string() { let diagnostics = compensate_failed_workspace_worker_create( @@ -4401,6 +4449,11 @@ impl WorkspaceApi { &self, worker: &RuntimeWorkerRef, ) -> ApiResult { + // Fence displaced singleton generations before renewing credentials, replacing bindings, + // or asking any Runtime to launch execution. + let singleton_lease = self + .store + .require_current_worker_singleton_owner(&self.config.workspace_id, worker)?; for link in self .store .list_worker_workdir_links(&self.config.workspace_id, worker)? @@ -4464,10 +4517,13 @@ impl WorkspaceApi { } } } - let binding = self + let mut binding = self .runtime .replace_worker_workspace_api(worker, self.workspace_api_ref(&worker.runtime_id)) .map_err(|error| error.into_error())?; + if let Some(summary) = binding.worker.as_mut() { + summary.singleton_key = singleton_lease.as_ref().map(|lease| lease.key.clone()); + } if binding.state != InternalWorkerOperationState::Accepted { return Ok(InternalWorkerRestoreResult { state: server_api::WorkerRestoreState::Rejected, @@ -4476,10 +4532,14 @@ impl WorkspaceApi { }); } sync_runtime_worker_workdir_attachments(self, worker)?; - Ok(self + let mut restored = self .runtime .restore_worker(worker) - .map_err(|error| error.into_error())?) + .map_err(|error| error.into_error())?; + if let Some(summary) = restored.worker.as_mut() { + summary.singleton_key = singleton_lease.as_ref().map(|lease| lease.key.clone()); + } + Ok(restored) } fn repository_reader(&self) -> RepositoryRegistryReader { @@ -5819,6 +5879,7 @@ fn worker_spawn_request_from_api( Ok(WorkerSpawnRequest { intent, requested_worker_name: request.requested_worker_name, + singleton_key: request.singleton_key, acceptance, profile, ticket_assignment: request @@ -16162,37 +16223,33 @@ fn require_online_workspace_orchestrator_source( } fn find_online_workspace_orchestrator(api: &WorkspaceApi) -> Option { - api.runtime - .list_workers(1000) - .items - .into_iter() - .find(|worker| { - worker.singleton_key.as_deref() - == Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY) - && worker.workspace.workspace_id.as_deref() - == Some(api.config.workspace_id.as_str()) - && matches!(worker.state.as_str(), "idle" | "running" | "paused") - }) + let owner = api + .store + .current_worker_singleton_owner( + &api.config.workspace_id, + crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY, + ) + .ok() + .flatten()?; + let mut worker = api.runtime.worker(&owner.worker).ok()?; + worker.singleton_key = Some(owner.key); + (worker.workspace.workspace_id.as_deref() == Some(api.config.workspace_id.as_str()) + && matches!(worker.state.as_str(), "idle" | "running" | "paused")) + .then_some(worker) } fn find_workspace_orchestrator(api: &WorkspaceApi) -> Option { - let is_orchestrator = |worker: &InternalWorkerSummary| { - worker.singleton_key.as_deref() == Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY) - }; - if let Some(worker) = find_online_workspace_orchestrator(api) { - return Some(worker); - } - for runtime in api.runtime.list_runtimes(1000).items { - if let Ok(stopped) = api - .runtime - .list_stopped_workers_for_runtime(&runtime.runtime_id, 1000) - { - if let Some(worker) = stopped.items.into_iter().find(is_orchestrator) { - return Some(worker); - } - } - } - None + let owner = api + .store + .current_worker_singleton_owner( + &api.config.workspace_id, + crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY, + ) + .ok() + .flatten()?; + let mut worker = api.runtime.worker(&owner.worker).ok()?; + worker.singleton_key = Some(owner.key); + Some(worker) } async fn scoped_get_memory_document( @@ -16242,7 +16299,6 @@ async fn scoped_memory_backend_operation( const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation"; const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation"; -const MEMORY_CONSOLIDATION_WORKER_SCAN_LIMIT: usize = 100; async fn scoped_memory_consolidation( State(api): State, @@ -16302,6 +16358,7 @@ fn start_memory_staging_consolidation( &runtime_id, WorkerSpawnRequest { requested_worker_name: Some(MEMORY_CONSOLIDATION_PROFILE.to_string()), + singleton_key: Some(MEMORY_CONSOLIDATION_SINGLETON_KEY.to_string()), intent: WorkerSpawnIntent::WorkspaceOrchestrator, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 1, @@ -16351,25 +16408,24 @@ fn start_memory_staging_consolidation( fn try_reuse_memory_consolidation_worker( api: &WorkspaceApi, - runtime_id: &str, + _runtime_id: &str, input_content: &str, candidate_count: usize, total_bytes: u64, ) -> ApiResult> { - let workers = api - .runtime - .list_workers_for_runtime(runtime_id, MEMORY_CONSOLIDATION_WORKER_SCAN_LIMIT) - .map_err(|err| err.into_error())?; - let mut consolidaters = workers - .items - .into_iter() - .filter(is_memory_consolidation_worker) - .collect::>(); - if consolidaters.is_empty() { + let Some(owner) = api.store.current_worker_singleton_owner( + &api.config.workspace_id, + MEMORY_CONSOLIDATION_SINGLETON_KEY, + )? + else { return Ok(None); - } - consolidaters.sort_by(|a, b| a.worker.worker_id.cmp(&b.worker.worker_id)); - if let Some(worker) = consolidaters.iter().find(|worker| worker.state != "idle") { + }; + let mut worker = api + .runtime + .worker(&owner.worker) + .map_err(|error| error.into_error())?; + worker.singleton_key = Some(owner.key); + if worker.state != "idle" { return Ok(Some(MemoryConsolidationOutput { status: "skipped_existing_not_idle".to_string(), summary: format!( @@ -16380,10 +16436,6 @@ fn try_reuse_memory_consolidation_worker( total_bytes, })); } - let worker = consolidaters - .first() - .expect("non-empty consolidater list") - .clone(); let input = api .runtime .send_input( @@ -16418,11 +16470,6 @@ fn try_reuse_memory_consolidation_worker( })) } -fn is_memory_consolidation_worker(worker: &InternalWorkerSummary) -> bool { - worker.singleton_key.as_deref() == Some(MEMORY_CONSOLIDATION_SINGLETON_KEY) - || worker.profile.as_deref() == Some(MEMORY_CONSOLIDATION_PROFILE) -} - fn memory_consolidation_input_content(candidate_count: usize, total_bytes: u64) -> String { format!( "Process pending Memory staging candidates through MemoryStagingList, MemoryStagingRead, Memory tools, and MemoryStagingClose. Current backlog: {candidate_count} candidate(s), {total_bytes} byte(s)." @@ -17780,10 +17827,7 @@ async fn scoped_start_workspace_orchestrator( if workspace_orchestrator_is_online(&existing) { return Ok(Json(workspace_orchestrator_response(&api, "existing"))); } - let restored = api - .runtime - .restore_worker(&existing.worker) - .map_err(|error| error.into_error())?; + let restored = api.restore_workspace_worker(&existing.worker)?; if restored.state == server_api::WorkerRestoreState::Accepted { *api.orchestrator_attention_fingerprint .lock() @@ -17805,7 +17849,7 @@ async fn scoped_start_workspace_orchestrator( .runtime .delete_worker(&existing.worker) .map_err(|error| error.into_error())?; - if deleted.state != InternalWorkerOperationState::Accepted { + if deleted.state != InternalWorkerOperationState::Accepted || !deleted.deleted { return Err(ApiError::with_diagnostics( Error::RuntimeOperationFailed { runtime_id: existing.worker.runtime_id.clone(), @@ -17815,6 +17859,10 @@ async fn scoped_start_workspace_orchestrator( deleted.diagnostics, )); } + let removal = api + .store + .delete_worker_registry(&api.config.workspace_id, &existing.worker)?; + api.worker_projection.publish_commit(removal)?; disposition = "recreated"; } @@ -17824,6 +17872,7 @@ async fn scoped_start_workspace_orchestrator( requested_worker_name: Some( crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY.to_string(), ), + singleton_key: Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY.to_string()), intent: WorkerSpawnIntent::WorkspaceOrchestrator, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, @@ -22296,6 +22345,22 @@ async fn execute_runtime_removal( .await; return Err(error.into()); } + if api.store.has_reserved_worker_create_for_runtime( + &operation.workspace_id, + &operation.runtime_id, + )? { + let _ = api + .store + .mark_runtime_removal_failed( + &operation.operation_id, + "runtime_removal_reserved_worker_create_blocked", + ) + .await; + return Err(Error::RuntimeBindingConflict( + "runtime_removal_reserved_worker_create_blocked".to_string(), + ) + .into()); + } let binding = api .store .get_workspace_runtime_binding(&operation.workspace_id, &operation.runtime_id) @@ -22984,6 +23049,7 @@ async fn create_workspace_worker_inner( let CreateWorkspaceWorkerRequest { runtime_id, display_name, + singleton_key, profile, ticket_assignment, initial_submit, @@ -23069,6 +23135,7 @@ async fn create_workspace_worker_inner( browser_worker_spawn_policy(ticket_assignment, &initial_submit)?; let request = WorkerSpawnRequest { requested_worker_name: Some(display_name.clone()), + singleton_key, intent, acceptance, profile: profile_selector, @@ -24009,6 +24076,38 @@ fn finalize_spawn_compensation_after_worker_delete( diagnostics } +fn cleanup_spawn_created_workdirs( + api: &WorkspaceApi, + workdir_ids: &[String], +) -> Vec { + let mut diagnostics = Vec::new(); + for workdir_id in workdir_ids { + match execute_workdir_removal( + api, + workdir_id, + "backend:worker_spawn_compensation", + "remove Workdir created before rejected Worker spawn", + ) { + Ok(result) if result.disposition == WorkingDirectoryRemovalDisposition::Removed => {} + Ok(result) => diagnostics.push(spawn_compensation_diagnostic( + "worker_spawn_compensation_workdir_cleanup_failed", + format!( + "Durable removal retained pre-spawn Workdir `{workdir_id}`: disposition={:?}, retryable={}", + result.disposition, result.retryable, + ), + )), + Err(error) => diagnostics.push(spawn_compensation_diagnostic( + "worker_spawn_compensation_workdir_cleanup_failed", + format!( + "Failed to reserve durable removal for pre-spawn Workdir `{workdir_id}`: {}", + sanitize_backend_error(&error.to_string()) + ), + )), + } + } + diagnostics +} + fn lifecycle_failure_detail( action: &str, result: &std::result::Result, @@ -24049,6 +24148,9 @@ async fn create_runtime_worker( Json(request): Json, ) -> ApiResult> { let mut request = worker_spawn_request_from_api(request)?; + if let Some(singleton_key) = request.singleton_key.as_deref() { + crate::store::validate_worker_singleton_key(singleton_key)?; + } validate_worker_initial_submit(&request.initial_submit)?; if let Some(assignment) = request.ticket_assignment.as_ref() && let Some(worker) = existing_lifecycle_assignment_worker(&api, assignment, &runtime_id)? @@ -24098,18 +24200,24 @@ async fn create_runtime_worker( .as_ref() .map(|assignment| assignment.operation_id.clone()) }); + let mut created_workdir_ids = Vec::new(); for attachment in &mut request.resolved_workdir_attachment_requests { let workdir_id = upsert_pending_backend_workdir(&api, &runtime_id, &mut attachment.working_directory)?; + created_workdir_ids.push(workdir_id.clone()); let operation_id = repository_operation_id .clone() .unwrap_or_else(|| format!("worker-spawn-workdir:{workdir_id}")); - authorize_worker_spawn_workdir_materialization( + if let Err(error) = authorize_worker_spawn_workdir_materialization( &api, &runtime_id, &operation_id, &mut attachment.working_directory, - )?; + ) { + let mut diagnostics = error.diagnostics; + diagnostics.extend(cleanup_spawn_created_workdirs(&api, &created_workdir_ids)); + return Err(ApiError::with_diagnostics(error.error, diagnostics)); + } } let requested_worker_name = request.requested_worker_name.clone(); let spawn_idempotency = @@ -24117,7 +24225,7 @@ async fn create_runtime_worker( if let (Some(assignment), Some((_, fingerprint))) = (lifecycle_assignment.as_ref(), spawn_idempotency.as_ref()) { - api.store.reserve_ticket_assignment_operation( + if let Err(error) = api.store.reserve_ticket_assignment_operation( &api.config.workspace_id, &assignment.operation_id, &assignment.ticket_id, @@ -24125,13 +24233,11 @@ async fn create_runtime_worker( None, fingerprint, &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), - )?; + ) { + let diagnostics = cleanup_spawn_created_workdirs(&api, &created_workdir_ids); + return Err(ApiError::with_diagnostics(error, diagnostics)); + } } - let created_workdir_ids = request - .resolved_workdir_attachment_requests - .iter() - .filter_map(|attachment| attachment.working_directory.backend_workdir_id.clone()) - .collect::>(); let result = api.spawn_workspace_worker(&runtime_id, request)?; if let Some(worker) = result.worker.as_ref() { let compensation = WorkerSpawnCompensationContext { @@ -24909,9 +25015,31 @@ fn project_workspace_worker( fn project_workspace_worker_with_attachments( api: &WorkspaceApi, - summary: InternalWorkerSummary, + mut summary: InternalWorkerSummary, workdir_attachments: Vec, ) -> ApiResult { + match api + .store + .require_current_worker_singleton_owner(&api.config.workspace_id, &summary.worker) + { + Ok(lease) => summary.singleton_key = lease.map(|lease| lease.key), + Err(Error::RepositoryConflict(message)) + if message.starts_with("worker_singleton_fenced:") => + { + summary.singleton_key = None; + summary.availability = + protocol::subscription::SubscriptionWorkerAvailability::Unavailable; + summary.state = "unavailable".to_string(); + summary.worker_state = None; + summary.diagnostics.push(RuntimeDiagnostic { + code: "worker_singleton_fenced".to_string(), + severity: HostDiagnosticSeverity::Warning, + message: "This historical singleton generation is fenced by Backend ownership." + .to_string(), + }); + } + Err(error) => return Err(error.into()), + } let resource_key = api .store .resource_key( @@ -28427,6 +28555,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "External expiry Worker".to_string(), + singleton_key: None, profile: Some("builtin:coder".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -28992,6 +29121,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.to_string(), display_name: "External multi-attachment Worker".to_string(), + singleton_key: None, profile: Some("builtin:coder".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -29239,6 +29369,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "External analysis Worker".to_string(), + singleton_key: None, profile: Some("builtin:coder".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -31654,6 +31785,7 @@ mod tests { foreign_repository.id = "foreign".to_string(); let mut workdir_flow_launch = WorkerSpawnRequest { requested_worker_name: Some("cross-workspace-workdir".to_string()), + singleton_key: None, intent: WorkerSpawnIntent::WorkspaceCoding, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 1, @@ -31913,6 +32045,7 @@ mod tests { ticket_id: ticket.id.clone(), role: TicketWorkerRole::Coder, }, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 1, }, @@ -32013,6 +32146,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: "missing-runtime".to_string(), display_name: "Rejected Coder".to_string(), + singleton_key: None, profile: Some("builtin:coder".to_string()), ticket_assignment: Some(CreateWorkspaceWorkerTicketAssignmentRequest { ticket_id: ticket.id.clone(), @@ -32051,6 +32185,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Scoped Worker".to_string(), + singleton_key: None, profile: Some("builtin:coder".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -32085,6 +32220,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Cross-runtime Worker".to_string(), + singleton_key: None, profile: Some("builtin:coder".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -32173,6 +32309,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Generic Worker".to_string(), + singleton_key: None, profile: Some("builtin:coder".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -32430,6 +32567,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Guarded spawn controller".to_string(), + singleton_key: None, profile: Some("builtin:orchestrator".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -32911,6 +33049,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Strict remote spawn controller".to_string(), + singleton_key: None, profile: Some("builtin:orchestrator".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -33133,6 +33272,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Control caller".to_string(), + singleton_key: None, profile: None, ticket_assignment: None, initial_submit: Vec::new(), @@ -33164,6 +33304,7 @@ mod tests { let request = || CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Idempotent controlled child".to_string(), + singleton_key: None, profile: None, ticket_assignment: None, initial_submit: Vec::new(), @@ -33230,6 +33371,267 @@ mod tests { assert_eq!(grants[0].operation_id, "control-spawn-retry"); } + #[tokio::test] + async fn arbitrary_singleton_create_conflicts_while_idle_or_stopped_and_restore_keeps_key() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + let request = || CreateWorkspaceWorkerRequest { + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), + display_name: "Subject Worker".to_string(), + singleton_key: Some("subjektiv:subject-42".to_string()), + profile: Some("builtin:companion".to_string()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachments: Vec::new(), + control_operation_id: None, + }; + + let Json(created) = + create_workspace_worker(State(api.clone()), HeaderMap::new(), Json(request())) + .await + .unwrap(); + assert_eq!( + created.worker.singleton_key.as_deref(), + Some("subjektiv:subject-42") + ); + let duplicate = + create_workspace_worker(State(api.clone()), HeaderMap::new(), Json(request())) + .await + .unwrap_err(); + assert!( + matches!(duplicate.error, Error::RepositoryConflict(message) if message.contains("worker_singleton_owned")) + ); + + let worker = RuntimeWorkerRef::new(&created.runtime_id, &created.worker_id); + let stopped = api + .runtime + .stop_worker( + &worker, + WorkerLifecycleRequest { + reason: Some("singleton restore test".to_string()), + ticket_assignment: None, + }, + ) + .unwrap(); + assert_eq!(stopped.state, InternalWorkerOperationState::Accepted); + let stopped_duplicate = + create_workspace_worker(State(api.clone()), HeaderMap::new(), Json(request())) + .await + .unwrap_err(); + assert!(matches!( + stopped_duplicate.error, + Error::RepositoryConflict(_) + )); + + let restored = api.restore_workspace_worker(&worker).unwrap(); + assert_eq!(restored.state, server_api::WorkerRestoreState::Rejected); + assert_eq!( + api.store + .current_worker_singleton_owner(&api.config.workspace_id, "subjektiv:subject-42",) + .unwrap() + .unwrap() + .worker, + worker + ); + } + + #[tokio::test] + async fn singleton_retry_disconnect_replacement_and_restore_are_generation_fenced() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await; + let runtime = WorkdirlessFixtureRuntime::default(); + api.runtime.register_or_replace(runtime.clone()); + let request = |operation_id: &str| WorkerSpawnRequest { + requested_worker_name: Some("Subject Worker".to_string()), + singleton_key: Some("subjektiv:subject-retry".to_string()), + intent: WorkerSpawnIntent::WorkspaceCompanion, + acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: 0, + }, + profile: ProfileSelector::Builtin("builtin:companion".to_string()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachment_requests: Vec::new(), + resolved_workdir_attachment_requests: Vec::new(), + resolved_workdir_attachments: Vec::new(), + resolved_config_bundle: None, + resolved_worker_observation_enabled: false, + resolved_worker_observation_grants: Vec::new(), + resolved_workspace_api: None, + resolved_memory_settings: None, + resolved_control_operation: Some(WorkerControlOperation { + operation_id: operation_id.to_string(), + input_fingerprint: format!("sha256:{operation_id}"), + }), + }; + + let mut invalid_request = request("singleton-create-invalid"); + invalid_request.singleton_key = Some(String::new()); + assert!(matches!( + api.spawn_workspace_worker(WorkdirlessFixtureRuntime::RUNTIME_ID, invalid_request) + .unwrap_err() + .error, + Error::InvalidInput(_) + )); + assert!(runtime.spawn_requests().is_empty()); + + assert!(matches!( + api.spawn_workspace_worker("missing-runtime", request("singleton-create-undispatched")) + .unwrap_err() + .error, + Error::UnknownRuntime(_) + )); + + // A rejected attempt is terminalized only after the Runtime confirms that its reserved + // Worker is absent. A fresh retry then takes the next generation without leaving the key + // permanently wedged. + runtime.reject_next_spawn(); + let rejected = api + .spawn_workspace_worker( + WorkdirlessFixtureRuntime::RUNTIME_ID, + request("singleton-create-1"), + ) + .unwrap(); + assert_eq!(rejected.state, InternalWorkerOperationState::Rejected); + assert!( + api.store + .current_worker_singleton_owner(&api.config.workspace_id, "subjektiv:subject-retry") + .unwrap() + .is_none() + ); + let created = api + .spawn_workspace_worker( + WorkdirlessFixtureRuntime::RUNTIME_ID, + request("singleton-create-retry"), + ) + .unwrap(); + assert_eq!(created.state, InternalWorkerOperationState::Accepted); + let worker = created.worker.unwrap(); + let current_owner = api + .store + .current_worker_singleton_owner(&api.config.workspace_id, "subjektiv:subject-retry") + .unwrap() + .unwrap(); + assert_eq!(current_owner.worker, worker.worker); + assert_eq!(current_owner.generation, 3); + assert_eq!( + worker.singleton_key, + Some("subjektiv:subject-retry".to_string()) + ); + + // Idle and paused Runtime metadata never releases Backend ownership, and a conflict is + // decided before the Runtime sees another spawn request. + runtime + .workers + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner)[0] + .state = "paused".to_string(); + let spawn_count = runtime.spawn_requests().len(); + assert!(matches!( + api.spawn_workspace_worker( + WorkdirlessFixtureRuntime::RUNTIME_ID, + request("singleton-create-conflict"), + ) + .unwrap_err() + .error, + Error::RepositoryConflict(message) if message.contains("worker_singleton_owned") + )); + assert_eq!(runtime.spawn_requests().len(), spawn_count); + let restored = api.restore_workspace_worker(&worker.worker).unwrap(); + assert_eq!(restored.state, server_api::WorkerRestoreState::Accepted); + assert_eq!( + restored.worker.unwrap().singleton_key.as_deref(), + Some("subjektiv:subject-retry") + ); + + // Replacing the connected Runtime with an empty instance models an unavailable/restarted + // Runtime. Ownership remains durable while restore cannot observe the Worker. + let disconnected_runtime = WorkdirlessFixtureRuntime::default(); + api.runtime + .register_or_replace(disconnected_runtime.clone()); + let disconnected_restore = api.restore_workspace_worker(&worker.worker).unwrap(); + assert_eq!( + disconnected_restore.state, + server_api::WorkerRestoreState::Rejected + ); + assert_eq!( + api.store + .current_worker_singleton_owner( + &api.config.workspace_id, + "subjektiv:subject-retry", + ) + .unwrap() + .unwrap() + .worker, + worker.worker + ); + assert!( + api.spawn_workspace_worker( + WorkdirlessFixtureRuntime::RUNTIME_ID, + request("singleton-create-disconnected"), + ) + .is_err() + ); + + // Reconnection restores the current generation. Only confirmed terminal deletion permits + // a replacement, after which even a stale Runtime copy of the old Worker is fenced before + // any Runtime restore side effect. + api.runtime.register_or_replace(runtime.clone()); + assert_eq!( + api.restore_workspace_worker(&worker.worker).unwrap().state, + server_api::WorkerRestoreState::Accepted + ); + let stale_worker = runtime + .workers + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .iter() + .find(|candidate| candidate.worker == worker.worker) + .cloned() + .unwrap(); + let deleted = api.runtime.delete_worker(&worker.worker).unwrap(); + assert!(deleted.deleted); + api.store + .delete_worker_registry(&api.config.workspace_id, &worker.worker) + .unwrap(); + runtime + .workers + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(stale_worker); + assert!(matches!( + api.restore_workspace_worker(&worker.worker).unwrap_err().error, + Error::RepositoryConflict(message) if message.contains("worker_singleton_fenced") + )); + let replacement = api + .spawn_workspace_worker( + WorkdirlessFixtureRuntime::RUNTIME_ID, + request("singleton-create-2"), + ) + .unwrap() + .worker + .unwrap(); + assert_ne!(replacement.worker.worker_id, worker.worker.worker_id); + assert!(matches!( + api.restore_workspace_worker(&worker.worker).unwrap_err().error, + Error::RepositoryConflict(message) if message.contains("worker_singleton_fenced") + )); + assert_eq!( + api.store + .current_worker_singleton_owner( + &api.config.workspace_id, + "subjektiv:subject-retry", + ) + .unwrap() + .unwrap() + .worker, + replacement.worker + ); + } + #[tokio::test] async fn explicit_orchestrator_launch_marks_only_the_dedicated_worker() { let workspace = tempfile::tempdir().unwrap(); @@ -33243,6 +33645,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: "Generic Orchestrator Profile Worker".to_string(), + singleton_key: None, profile: Some("builtin:orchestrator".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -33260,6 +33663,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), display_name: crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY.to_string(), + singleton_key: None, profile: Some("builtin:orchestrator".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -34425,6 +34829,7 @@ mod tests { WorkerSpawnRequest { requested_worker_name: Some(name.to_string()), intent: WorkerSpawnIntent::WorkspaceCompanion, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -35329,6 +35734,7 @@ mod tests { WorkerSpawnRequest { requested_worker_name: Some("job-notification-target".to_string()), intent: WorkerSpawnIntent::WorkspaceCompanion, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -37564,13 +37970,12 @@ mod tests { let resolved_config_bundle = None; let existing = api - .runtime - .spawn_worker( + .spawn_workspace_worker( EMBEDDED_WORKER_RUNTIME_ID, - test_create_binding(), WorkerSpawnRequest { requested_worker_name: Some(MEMORY_CONSOLIDATION_PROFILE.to_string()), intent: WorkerSpawnIntent::WorkspaceOrchestrator, + singleton_key: Some(MEMORY_CONSOLIDATION_SINGLETON_KEY.to_string()), acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -37630,10 +38035,48 @@ mod tests { let consolidaters_after_second = workers_after_second .items .iter() - .filter(|worker| is_memory_consolidation_worker(worker)) + .filter(|worker| { + worker.singleton_key.as_deref() == Some(MEMORY_CONSOLIDATION_SINGLETON_KEY) + }) .collect::>(); assert_eq!(consolidaters_after_second.len(), 1); assert_eq!(consolidaters_after_second[0].worker.worker_id, worker_id); + + let singleton_worker = RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, &worker_id); + let stopped = api + .runtime + .stop_worker( + &singleton_worker, + WorkerLifecycleRequest { + reason: Some("replace completed Memory singleton".to_string()), + ticket_assignment: None, + }, + ) + .unwrap(); + assert_eq!(stopped.state, InternalWorkerOperationState::Accepted); + let removed = api.runtime.delete_worker(&singleton_worker).unwrap(); + assert!(removed.deleted); + api.store + .delete_worker_registry(&api.config.workspace_id, &singleton_worker) + .unwrap(); + let replacement = start_memory_staging_consolidation( + api.clone(), + MemoryConsolidateStagingOperation { force: true }, + ) + .unwrap(); + assert_eq!(replacement.status, "started"); + let replacement_owner = api + .store + .current_worker_singleton_owner( + &api.config.workspace_id, + MEMORY_CONSOLIDATION_SINGLETON_KEY, + ) + .unwrap() + .unwrap(); + assert_ne!( + replacement_owner.worker.worker_id, worker_id, + "removed singleton generation must be replaceable" + ); } fn init_clean_git_workspace(path: &std::path::Path) { @@ -37836,6 +38279,7 @@ mod tests { ticket_id: "notification-source".to_string(), role: TicketWorkerRole::Coder, }, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -38019,6 +38463,7 @@ mod tests { Json(CreateWorkspaceWorkerRequest { runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.to_string(), display_name: "Manual Coder candidate".to_string(), + singleton_key: None, profile: Some("builtin:coder".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -38577,6 +39022,7 @@ mod tests { ticket_id: "implementation-cancellation".to_string(), role: TicketWorkerRole::Coder, }, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -38680,6 +39126,7 @@ mod tests { ticket_id: name.to_string(), role: TicketWorkerRole::Coder, }, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -39295,6 +39742,7 @@ mod tests { ticket_id: "source-ticket".to_string(), role: TicketWorkerRole::Coder, }, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -39361,6 +39809,7 @@ mod tests { ticket_id: "source-ticket".to_string(), role: TicketWorkerRole::Coder, }, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -39705,6 +40154,7 @@ mod tests { ticket_id: ticket.id.clone(), role: server_api::RuntimeTicketWorkerRole::Coder, }, + singleton_key: None, acceptance: server_api::RuntimeWorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 1, }, @@ -40675,6 +41125,7 @@ mod tests { WorkerSpawnRequest { intent: WorkerSpawnIntent::WorkspaceCompanion, requested_worker_name: Some("guard-target".to_string()), + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -40733,6 +41184,7 @@ mod tests { WorkerSpawnRequest { intent: WorkerSpawnIntent::WorkspaceCompanion, requested_worker_name: Some("remove-target".to_string()), + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -40845,6 +41297,7 @@ mod tests { WorkerSpawnRequest { intent: WorkerSpawnIntent::WorkspaceCompanion, requested_worker_name: Some("compensation-target".to_string()), + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -42781,6 +43234,7 @@ mod tests { WorkerSpawnRequest { intent: WorkerSpawnIntent::WorkspaceCompanion, requested_worker_name: Some("cleanup-projection-target".to_string()), + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -43320,6 +43774,7 @@ mod tests { "runtime-test", "spawn-flow-race", &"f".repeat(64), + None, &memory_settings, ) .unwrap(); @@ -45057,6 +45512,7 @@ mod tests { Json(server_api::RuntimeWorkerSpawnRequest { intent: server_api::RuntimeWorkerSpawnIntent::WorkspaceCompanion, requested_worker_name: Some("Runtime Worker".to_string()), + singleton_key: None, acceptance: server_api::RuntimeWorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -45085,6 +45541,7 @@ mod tests { CreateWorkspaceWorkerRequest { runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.to_string(), display_name: "Remote Worker".to_string(), + singleton_key: None, profile: Some("builtin:companion".to_string()), ticket_assignment: None, initial_submit: Vec::new(), @@ -45860,6 +46317,7 @@ mod tests { role: TicketWorkerRole::Coder, }, requested_worker_name: None, + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, @@ -46516,6 +46974,7 @@ mod tests { let spawn_request = WorkerSpawnRequest { intent: WorkerSpawnIntent::WorkspaceCompanion, requested_worker_name: Some("multiplexed-console".to_string()), + singleton_key: None, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: 0, }, diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index cea5ba56d..c97acaecd 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -27,7 +27,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore; use crate::{Error, Result}; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 72; +const LATEST_SCHEMA_VERSION: i64 = 73; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings"; const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit"; @@ -66,6 +66,8 @@ const BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME: &str = "atomic Backend Job notification delivery claims"; const BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME: &str = "durable bounded Backend Job dispatch queue"; +const WORKER_SINGLETON_OWNERSHIP_MIGRATION_NAME: &str = + "durable Workspace Worker singleton ownership"; const TICKET_SCHEMA_VERSION_WITH_TARGETS: i64 = 7; const TICKET_SCHEMA_BASELINE_NAME: &str = "ticket schema baseline"; const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#" @@ -223,6 +225,11 @@ const MIGRATIONS: &[Migration] = &[ name: BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME, apply: migrate_backend_job_dispatch_queue_v71_to_v72, }, + Migration { + version: 73, + name: WORKER_SINGLETON_OWNERSHIP_MIGRATION_NAME, + apply: migrate_worker_singleton_ownership_v72_to_v73, + }, ]; #[derive(Clone, Copy)] @@ -277,6 +284,18 @@ pub struct WorkerCreateReservation { pub worker_id: WorkerId, pub create_fingerprint: String, pub memory_settings: manifest::WorkspaceMemorySettingsSnapshot, + pub singleton: Option, +} + +/// Durable lease proving which Worker generation owns one opaque Workspace singleton key. +/// +/// The current owner lives in `worker_singleton_owners`; matching lease data remains on every +/// create reservation so a displaced generation can be fenced from later restore attempts. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerSingletonLease { + pub key: String, + pub generation: u64, + pub worker: RuntimeWorkerRef, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -1566,6 +1585,16 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { now: &str, ) -> Result; + fn current_worker_singleton_owner( + &self, + workspace_id: &str, + singleton_key: &str, + ) -> Result>; + fn require_current_worker_singleton_owner( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result>; fn upsert_worker_registry( &self, record: &WorkerRegistryRecord, @@ -1580,6 +1609,11 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { workspace_id: &str, worker: &RuntimeWorkerRef, ) -> Result; + fn has_reserved_worker_create_for_runtime( + &self, + workspace_id: &str, + runtime_id: &str, + ) -> Result; fn list_worker_registry( &self, workspace_id: &str, @@ -2095,6 +2129,7 @@ impl SqliteWorkspaceStore { runtime_id: &str, allocation_key: &str, request_fingerprint: &str, + singleton_key: Option<&str>, current_memory_settings: &WorkspaceMemorySettingsRecord, ) -> Result { if allocation_key.trim().is_empty() || request_fingerprint.trim().is_empty() { @@ -2102,13 +2137,17 @@ impl SqliteWorkspaceStore { "Worker create allocation key and fingerprint must be non-empty".to_string(), )); } + if let Some(key) = singleton_key { + validate_worker_singleton_key(key)?; + } validate_workspace_memory_settings_record(current_memory_settings, workspace_id)?; self.with_conn_mut(|conn| { let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; let existing = tx .query_row( "SELECT worker_id, runtime_id, request_fingerprint, create_fingerprint, \ - memory_settings_revision, memory_language, state \ + memory_settings_revision, memory_language, state, singleton_key, \ + singleton_generation \ FROM worker_create_reservations \ WHERE workspace_id = ?1 AND allocation_key = ?2", params![workspace_id, allocation_key], @@ -2121,11 +2160,13 @@ impl SqliteWorkspaceStore { row.get::<_, Option>(4)?, row.get::<_, Option>(5)?, row.get::<_, String>(6)?, + row.get::<_, Option>(7)?, + row.get::<_, Option>(8)?, )) }, ) .optional()?; - if let Some((worker_id, reserved_runtime_id, stored_request_fingerprint, create_fingerprint, revision, language, state)) = existing { + if let Some((worker_id, reserved_runtime_id, stored_request_fingerprint, create_fingerprint, revision, language, state, stored_singleton_key, singleton_generation)) = existing { if state == "removed" { return Err(Error::InvalidInput(format!( "Worker create allocation {allocation_key} was terminally removed" @@ -2133,6 +2174,7 @@ impl SqliteWorkspaceStore { } if reserved_runtime_id != runtime_id || stored_request_fingerprint.as_deref() != Some(request_fingerprint) + || stored_singleton_key.as_deref() != singleton_key { return Err(Error::InvalidInput(format!( "Worker create allocation {allocation_key} was already used with different input" @@ -2161,10 +2203,31 @@ impl SqliteWorkspaceStore { language, }; validate_workspace_memory_settings_snapshot(&snapshot, workspace_id)?; + let singleton = match (stored_singleton_key, singleton_generation) { + (Some(key), Some(generation)) => Some(WorkerSingletonLease { + key, + generation: generation.try_into().map_err(|_| { + Error::Store(format!( + "Worker create allocation {allocation_key} has an invalid singleton generation" + )) + })?, + worker: RuntimeWorkerRef::new(reserved_runtime_id, worker_id.to_string()), + }), + (None, None) => None, + _ => { + return Err(Error::Store(format!( + "Worker create allocation {allocation_key} has an incomplete singleton lease" + ))); + } + }; + if let Some(lease) = singleton.as_ref() { + require_current_worker_singleton_lease(&tx, workspace_id, lease)?; + } return Ok(WorkerCreateReservation { worker_id, create_fingerprint, memory_settings: snapshot, + singleton, }); } @@ -2197,41 +2260,178 @@ impl SqliteWorkspaceStore { let create_fingerprint = bound_worker_create_fingerprint(request_fingerprint, &snapshot); let worker_id = WorkerId::now_v7(); + let worker_id_text = worker_id.to_string(); let now = chrono::Utc::now().to_rfc3339(); + let singleton_owner = singleton_key + .map(|key| current_worker_singleton_owner(&tx, workspace_id, key)) + .transpose()? + .flatten(); + let singleton_generation = match singleton_owner.as_ref() { + None if singleton_key.is_some() => Some(1_u64), + None => None, + Some((owner, state)) if state.as_deref() == Some("removed") => { + Some(owner.generation.saturating_add(1)) + } + Some((owner, _)) => { + return Err(Error::RepositoryConflict(format!( + "worker_singleton_owned: singleton key is already owned by Worker {}:{}", + owner.worker.runtime_id, owner.worker.worker_id + ))); + } + }; allocate_resource_key( &tx, workspace_id, WorkspaceResourceKind::Worker, - &worker_id.to_string(), + &worker_id_text, &now, )?; tx.execute( "INSERT INTO worker_create_reservations(\ workspace_id, allocation_key, worker_id, runtime_id, create_fingerprint,\ state, created_at, updated_at, request_fingerprint,\ - memory_settings_revision, memory_language\ - ) VALUES (?1, ?2, ?3, ?4, ?5, 'reserved', ?6, ?6, ?7, ?8, ?9)", + memory_settings_revision, memory_language, singleton_key, singleton_generation\ + ) VALUES (?1, ?2, ?3, ?4, ?5, 'reserved', ?6, ?6, ?7, ?8, ?9, ?10, ?11)", params![ workspace_id, allocation_key, - worker_id.to_string(), + worker_id_text, runtime_id, create_fingerprint, now, request_fingerprint, snapshot.settings_revision as i64, snapshot.language, + singleton_key, + singleton_generation.map(|generation| generation as i64), ], )?; + let singleton = singleton_key.zip(singleton_generation).map(|(key, generation)| { + WorkerSingletonLease { + key: key.to_string(), + generation, + worker: RuntimeWorkerRef::new(runtime_id, worker_id_text.clone()), + } + }); + if let Some(lease) = singleton.as_ref() { + let changed = if let Some((previous, _)) = singleton_owner { + tx.execute( + "UPDATE worker_singleton_owners \ + SET runtime_id = ?4, worker_id = ?5, generation = ?6, updated_at = ?7 \ + WHERE workspace_id = ?1 AND singleton_key = ?2 AND worker_id = ?3", + params![ + workspace_id, + lease.key, + previous.worker.worker_id, + runtime_id, + worker_id_text, + lease.generation as i64, + now, + ], + )? + } else { + tx.execute( + "INSERT INTO worker_singleton_owners(\ + workspace_id, singleton_key, runtime_id, worker_id, generation, created_at, updated_at\ + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6)", + params![ + workspace_id, + lease.key, + runtime_id, + worker_id_text, + lease.generation as i64, + now, + ], + )? + }; + if changed != 1 { + return Err(Error::RepositoryConflict( + "worker_singleton_raced: singleton ownership changed during create reservation" + .to_string(), + )); + } + } tx.commit()?; Ok(WorkerCreateReservation { worker_id, create_fingerprint, memory_settings: snapshot, + singleton, }) }) } + pub(crate) fn current_worker_singleton_owner( + &self, + workspace_id: &str, + singleton_key: &str, + ) -> Result> { + validate_worker_singleton_key(singleton_key)?; + self.with_conn(|conn| { + Ok( + current_worker_singleton_owner(conn, workspace_id, singleton_key)?.and_then( + |(lease, state)| (state.as_deref() != Some("removed")).then_some(lease), + ), + ) + }) + } + + /// Returns the Worker's historical singleton lease and rejects it when another generation is + /// current. Workers without a singleton lease continue through the ordinary lifecycle path. + pub(crate) fn require_current_worker_singleton_owner( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result> { + self.with_conn(|conn| { + let stored = conn + .query_row( + "SELECT singleton_key, singleton_generation, state \ + FROM worker_create_reservations \ + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", + params![workspace_id, worker.runtime_id, worker.worker_id], + |row| { + Ok(( + row.get::<_, Option>(0)?, + row.get::<_, Option>(1)?, + row.get::<_, String>(2)?, + )) + }, + ) + .optional()?; + let Some((key, generation, state)) = stored else { + return Ok(None); + }; + let lease = match (key, generation) { + (None, None) => return Ok(None), + (Some(key), Some(generation)) => WorkerSingletonLease { + key, + generation: generation.try_into().map_err(|_| { + Error::Store(format!( + "Worker {}:{} has an invalid singleton generation", + worker.runtime_id, worker.worker_id + )) + })?, + worker: worker.clone(), + }, + _ => { + return Err(Error::Store(format!( + "Worker {}:{} has an incomplete singleton lease", + worker.runtime_id, worker.worker_id + ))); + } + }; + if state == "removed" { + return Err(Error::RepositoryConflict(format!( + "worker_singleton_fenced: Worker {}:{} generation {} has been terminally removed from singleton key {:?}", + worker.runtime_id, worker.worker_id, lease.generation, lease.key + ))); + } + require_current_worker_singleton_lease(conn, workspace_id, &lease)?; + Ok(Some(lease)) + }) + } + pub(crate) fn complete_worker_create_reservation( &self, workspace_id: &str, @@ -6458,6 +6658,22 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn current_worker_singleton_owner( + &self, + workspace_id: &str, + singleton_key: &str, + ) -> Result> { + SqliteWorkspaceStore::current_worker_singleton_owner(self, workspace_id, singleton_key) + } + + fn require_current_worker_singleton_owner( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result> { + SqliteWorkspaceStore::require_current_worker_singleton_owner(self, workspace_id, worker) + } + fn upsert_worker_registry( &self, record: &WorkerRegistryRecord, @@ -6594,6 +6810,27 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn has_reserved_worker_create_for_runtime( + &self, + workspace_id: &str, + runtime_id: &str, + ) -> Result { + self.with_conn(|conn| { + conn.query_row( + r#"SELECT EXISTS( + SELECT 1 + FROM worker_create_reservations + WHERE workspace_id = ?1 + AND runtime_id = ?2 + AND state = 'reserved' + )"#, + params![workspace_id, runtime_id], + |row| row.get::<_, bool>(0), + ) + .map_err(Error::from) + }) + } + fn list_worker_registry( &self, workspace_id: &str, @@ -6913,7 +7150,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { "UPDATE worker_create_reservations SET state = 'removed', updated_at = ?4 WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 - AND state = 'created'", + AND state IN ('reserved', 'created')", params![ workspace_id, worker.runtime_id, @@ -13444,6 +13681,160 @@ fn migrate_backend_job_dispatch_queue_v71_to_v72(conn: &Connection) -> Result<() Ok(()) } +fn migrate_worker_singleton_ownership_v72_to_v73(conn: &Connection) -> Result<()> { + let current = current_schema_version(conn)?; + if current != 72 { + return Err(Error::Store(format!( + "expected schema version 72 before {WORKER_SINGLETON_OWNERSHIP_MIGRATION_NAME} migration, found {current}" + ))); + } + let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; + if !column_exists(&tx, "worker_create_reservations", "singleton_key")? { + tx.execute( + "ALTER TABLE worker_create_reservations ADD COLUMN singleton_key TEXT", + [], + )?; + } + if !column_exists(&tx, "worker_create_reservations", "singleton_generation")? { + tx.execute( + "ALTER TABLE worker_create_reservations ADD COLUMN singleton_generation INTEGER \ + CHECK (singleton_generation IS NULL OR singleton_generation > 0)", + [], + )?; + } + tx.execute_batch( + r#" + CREATE TABLE IF NOT EXISTS worker_singleton_owners ( + workspace_id TEXT NOT NULL, + singleton_key TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + generation INTEGER NOT NULL CHECK (generation > 0), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, singleton_key), + UNIQUE (workspace_id, worker_id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE + ); + CREATE TRIGGER IF NOT EXISTS worker_create_reservation_singleton_lease_insert + BEFORE INSERT ON worker_create_reservations + WHEN (NEW.singleton_key IS NULL AND NEW.singleton_generation IS NOT NULL) + OR (NEW.singleton_key IS NOT NULL AND NEW.singleton_generation IS NULL) + BEGIN + SELECT RAISE(ABORT, 'worker_singleton_lease_incomplete'); + END; + CREATE TRIGGER IF NOT EXISTS worker_create_reservation_singleton_lease_update + BEFORE UPDATE OF singleton_key, singleton_generation ON worker_create_reservations + WHEN (NEW.singleton_key IS NULL AND NEW.singleton_generation IS NOT NULL) + OR (NEW.singleton_key IS NOT NULL AND NEW.singleton_generation IS NULL) + BEGIN + SELECT RAISE(ABORT, 'worker_singleton_lease_incomplete'); + END; + + -- Preserve the two historical singleton conventions as Backend ownership. If an old + -- database contains duplicates, deterministically fence every Worker except the oldest. + INSERT INTO worker_singleton_owners ( + workspace_id, singleton_key, runtime_id, worker_id, generation, created_at, updated_at + ) + SELECT worker.workspace_id, 'workspace-orchestrator', worker.runtime_id, worker.worker_id, + 1, worker.created_at, worker.updated_at + FROM worker_registry worker + WHERE worker.profile IN ('orchestrator', 'builtin:orchestrator') + AND worker.display_name = 'Workspace Orchestrator' + AND NOT EXISTS ( + SELECT 1 FROM worker_registry older + WHERE older.workspace_id = worker.workspace_id + AND older.profile IN ('orchestrator', 'builtin:orchestrator') + AND older.display_name = 'Workspace Orchestrator' + AND (older.created_at < worker.created_at + OR (older.created_at = worker.created_at AND older.worker_id < worker.worker_id)) + ); + INSERT INTO worker_singleton_owners ( + workspace_id, singleton_key, runtime_id, worker_id, generation, created_at, updated_at + ) + SELECT worker.workspace_id, 'workspace-memory-consolidation', worker.runtime_id, + worker.worker_id, 1, worker.created_at, worker.updated_at + FROM worker_registry worker + WHERE worker.profile IN ('memory-consolidation', 'builtin:memory-consolidation') + AND NOT EXISTS ( + SELECT 1 FROM worker_registry older + WHERE older.workspace_id = worker.workspace_id + AND older.profile IN ('memory-consolidation', 'builtin:memory-consolidation') + AND (older.created_at < worker.created_at + OR (older.created_at = worker.created_at AND older.worker_id < worker.worker_id)) + ); + INSERT INTO worker_create_reservations ( + workspace_id, allocation_key, worker_id, runtime_id, create_fingerprint, + state, created_at, updated_at, request_fingerprint, + memory_settings_revision, memory_language, singleton_key, singleton_generation + ) + SELECT worker.workspace_id, 'migration-v73-singleton:' || worker.worker_id, + worker.worker_id, worker.runtime_id, + 'migration-v73:' || CASE + WHEN worker.profile IN ('orchestrator', 'builtin:orchestrator') + AND worker.display_name = 'Workspace Orchestrator' + THEN 'workspace-orchestrator' + ELSE 'workspace-memory-consolidation' + END, + 'created', worker.created_at, worker.updated_at, NULL, NULL, NULL, + CASE + WHEN worker.profile IN ('orchestrator', 'builtin:orchestrator') + AND worker.display_name = 'Workspace Orchestrator' + THEN 'workspace-orchestrator' + ELSE 'workspace-memory-consolidation' + END, + 1 + FROM worker_registry worker + WHERE ( + ( + worker.profile IN ('orchestrator', 'builtin:orchestrator') + AND worker.display_name = 'Workspace Orchestrator' + ) + OR worker.profile IN ('memory-consolidation', 'builtin:memory-consolidation') + ) + AND NOT EXISTS ( + SELECT 1 FROM worker_create_reservations reservation + WHERE reservation.workspace_id = worker.workspace_id + AND reservation.worker_id = worker.worker_id + ); + UPDATE worker_create_reservations + SET singleton_key = ( + SELECT CASE + WHEN worker.profile IN ('orchestrator', 'builtin:orchestrator') + AND worker.display_name = 'Workspace Orchestrator' + THEN 'workspace-orchestrator' + ELSE 'workspace-memory-consolidation' + END + FROM worker_registry worker + WHERE worker.workspace_id = worker_create_reservations.workspace_id + AND worker.worker_id = worker_create_reservations.worker_id + AND ( + (worker.profile IN ('orchestrator', 'builtin:orchestrator') + AND worker.display_name = 'Workspace Orchestrator') + OR worker.profile IN ('memory-consolidation', 'builtin:memory-consolidation') + ) + ), + singleton_generation = 1 + WHERE EXISTS ( + SELECT 1 FROM worker_registry worker + WHERE worker.workspace_id = worker_create_reservations.workspace_id + AND worker.worker_id = worker_create_reservations.worker_id + AND ( + (worker.profile IN ('orchestrator', 'builtin:orchestrator') + AND worker.display_name = 'Workspace Orchestrator') + OR worker.profile IN ('memory-consolidation', 'builtin:memory-consolidation') + ) + ); + "#, + )?; + tx.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", + params![73_i64, WORKER_SINGLETON_OWNERSHIP_MIGRATION_NAME], + )?; + tx.commit()?; + Ok(()) +} + fn create_latest_workspace_schema(conn: &Connection) -> Result<()> { conn.execute_batch("DROP TABLE IF EXISTS typed_ticket_targets;")?; conn.execute_batch(include_str!("latest_schema.sql"))?; @@ -13896,6 +14287,90 @@ fn normalize_workspace_memory_language(language: &str) -> Result { Ok(language.to_string()) } +pub(crate) fn validate_worker_singleton_key(key: &str) -> Result<()> { + if key.is_empty() { + return Err(Error::InvalidInput( + "Worker singleton key must not be empty".to_string(), + )); + } + if key.len() > 512 { + return Err(Error::InvalidInput( + "Worker singleton key must not exceed 512 UTF-8 bytes".to_string(), + )); + } + Ok(()) +} + +fn current_worker_singleton_owner( + conn: &Connection, + workspace_id: &str, + singleton_key: &str, +) -> Result)>> { + let owner = conn + .query_row( + "SELECT owner.runtime_id, owner.worker_id, owner.generation, reservation.state \ + FROM worker_singleton_owners owner \ + LEFT JOIN worker_create_reservations reservation \ + ON reservation.workspace_id = owner.workspace_id \ + AND reservation.worker_id = owner.worker_id \ + WHERE owner.workspace_id = ?1 AND owner.singleton_key = ?2", + params![workspace_id, singleton_key], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, i64>(2)?, + row.get::<_, Option>(3)?, + )) + }, + ) + .optional()?; + owner + .map(|(runtime_id, worker_id, generation, state)| { + let generation = generation.try_into().map_err(|_| { + Error::Store(format!( + "Worker singleton key {singleton_key:?} has an invalid generation" + )) + })?; + Ok(( + WorkerSingletonLease { + key: singleton_key.to_string(), + generation, + worker: RuntimeWorkerRef::new(runtime_id, worker_id), + }, + state, + )) + }) + .transpose() +} + +fn require_current_worker_singleton_lease( + conn: &Connection, + workspace_id: &str, + expected: &WorkerSingletonLease, +) -> Result<()> { + let current = + current_worker_singleton_owner(conn, workspace_id, &expected.key)?.map(|(lease, _)| lease); + if current.as_ref() != Some(expected) { + let current_worker = current + .map(|lease| { + format!( + "{}:{} generation {}", + lease.worker.runtime_id, lease.worker.worker_id, lease.generation + ) + }) + .unwrap_or_else(|| "no current owner".to_string()); + return Err(Error::RepositoryConflict(format!( + "worker_singleton_fenced: Worker {}:{} generation {} no longer owns singleton key {:?}; current owner is {current_worker}", + expected.worker.runtime_id, + expected.worker.worker_id, + expected.generation, + expected.key + ))); + } + Ok(()) +} + fn bound_worker_create_fingerprint( request_fingerprint: &str, snapshot: &manifest::WorkspaceMemorySettingsSnapshot, @@ -15627,6 +16102,10 @@ mod tests { version: 72, name: BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME.to_string(), }, + WorkspaceSchemaMigrationStep { + version: 73, + name: WORKER_SINGLETON_OWNERSHIP_MIGRATION_NAME.to_string(), + }, ] ); @@ -15703,6 +16182,7 @@ mod tests { BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME.to_string(), ), (72, BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME.to_string()), + (73, WORKER_SINGLETON_OWNERSHIP_MIGRATION_NAME.to_string()), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -15988,7 +16468,8 @@ mod tests { .map(|migration| migration.version) .collect::>(), vec![ - 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72 + 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, + 73 ] ); SqliteWorkspaceStore::migrate_database(&path).unwrap(); @@ -15997,12 +16478,160 @@ mod tests { current_schema_version(&conn).unwrap(), LATEST_SCHEMA_VERSION ); - assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 23); + assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 24); assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap()); assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap()); assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap()); } + #[test] + fn singleton_migration_backfills_a_fenceable_reservation_for_legacy_workers() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("server.db"); + prepare_schema_v50(&path, Some("workspace-a")); + let conn = Connection::open(&path).unwrap(); + configure_sqlite(&conn).unwrap(); + for migration in MIGRATIONS + .iter() + .filter(|migration| migration.version > 50 && migration.version <= 72) + { + (migration.apply)(&conn).unwrap(); + } + conn.execute( + "INSERT INTO workspace_memory_settings(\ + workspace_id, settings_revision, language, created_at, updated_at\ + ) VALUES ('workspace-a', 1, 'English', '1', '1')", + [], + ) + .unwrap(); + conn.execute( + r#"INSERT INTO worker_registry( + workspace_id, worker_id, runtime_id, display_name, profile, retention_state, + created_at, updated_at + ) VALUES ( + 'workspace-a', 'legacy-orchestrator', 'legacy-runtime', + 'Workspace Orchestrator', 'builtin:orchestrator', 'normal', '1', '2' + )"#, + [], + ) + .unwrap(); + conn.execute( + r#"INSERT INTO worker_registry( + workspace_id, worker_id, runtime_id, display_name, profile, retention_state, + created_at, updated_at + ) VALUES ( + 'workspace-a', 'legacy-orchestrator-duplicate', 'other-runtime', + 'Workspace Orchestrator', 'builtin:orchestrator', 'normal', '3', '4' + )"#, + [], + ) + .unwrap(); + conn.execute_batch( + r#" + INSERT INTO worker_registry( + workspace_id, worker_id, runtime_id, display_name, profile, retention_state, + created_at, updated_at + ) VALUES ( + 'workspace-a', 'legacy-memory', 'memory-runtime', 'Memory Consolidation', + 'builtin:memory-consolidation', 'normal', '1', '2' + ); + INSERT INTO worker_registry( + workspace_id, worker_id, runtime_id, display_name, profile, retention_state, + created_at, updated_at + ) VALUES ( + 'workspace-a', 'legacy-memory-duplicate', 'other-memory-runtime', + 'Memory Consolidation duplicate', 'builtin:memory-consolidation', 'normal', '3', '4' + ); + INSERT INTO worker_create_reservations( + workspace_id, allocation_key, worker_id, runtime_id, create_fingerprint, + state, created_at, updated_at + ) VALUES ( + 'workspace-a', 'existing-memory-reservation', 'legacy-memory-duplicate', + 'other-memory-runtime', 'existing-memory', 'created', '3', '4' + ); + "#, + ) + .unwrap(); + assert_eq!( + conn.query_row( + "SELECT COUNT(*) FROM worker_create_reservations WHERE workspace_id = 'workspace-a'", + [], + |row| row.get::<_, i64>(0), + ) + .unwrap(), + 1 + ); + + migrate_worker_singleton_ownership_v72_to_v73(&conn).unwrap(); + assert!( + conn.execute( + "UPDATE worker_create_reservations SET singleton_generation = NULL \ + WHERE workspace_id = 'workspace-a' AND worker_id = 'legacy-orchestrator'", + [], + ) + .is_err(), + "migrated databases must reject incomplete singleton leases" + ); + drop(conn); + let store = SqliteWorkspaceStore::open(&path).unwrap(); + let legacy = RuntimeWorkerRef::new("legacy-runtime", "legacy-orchestrator"); + let lease = store + .require_current_worker_singleton_owner("workspace-a", &legacy) + .unwrap() + .expect("legacy singleton lease"); + assert_eq!(lease.key, "workspace-orchestrator"); + assert_eq!(lease.generation, 1); + let duplicate = RuntimeWorkerRef::new("other-runtime", "legacy-orchestrator-duplicate"); + assert!( + store + .require_current_worker_singleton_owner("workspace-a", &duplicate) + .unwrap_err() + .to_string() + .contains("worker_singleton_fenced") + ); + let memory_owner = RuntimeWorkerRef::new("memory-runtime", "legacy-memory"); + assert_eq!( + store + .require_current_worker_singleton_owner("workspace-a", &memory_owner) + .unwrap() + .unwrap() + .key, + "workspace-memory-consolidation" + ); + let memory_duplicate = + RuntimeWorkerRef::new("other-memory-runtime", "legacy-memory-duplicate"); + assert!( + store + .require_current_worker_singleton_owner("workspace-a", &memory_duplicate) + .unwrap_err() + .to_string() + .contains("worker_singleton_fenced") + ); + + store + .delete_worker_registry("workspace-a", &legacy) + .unwrap(); + let memory = store.get_workspace_memory_settings("workspace-a").unwrap(); + let replacement = store + .reserve_worker_create( + "workspace-a", + "replacement-runtime", + "replacement-operation", + "sha256:replacement", + Some("workspace-orchestrator"), + &memory, + ) + .unwrap(); + assert_eq!(replacement.singleton.unwrap().generation, 2); + assert!( + store + .require_current_worker_singleton_owner("workspace-a", &legacy) + .unwrap_err() + .to_string() + .contains("worker_singleton_fenced") + ); + } + #[test] fn schema_v50_embedded_runtime_fans_out_to_every_workspace() { let temp = tempfile::tempdir().unwrap(); @@ -17310,6 +17939,7 @@ mod tests { "arcadia", "operation-1", "sha256:one", + None, &memory_settings, ) .unwrap(); @@ -17341,6 +17971,7 @@ mod tests { "arcadia", "operation-1", "sha256:one", + None, &updated_memory_settings, ) .unwrap(), @@ -17353,6 +17984,7 @@ mod tests { "arcadia", "operation-1", "sha256:different", + None, &updated_memory_settings, ) .is_err() @@ -17380,6 +18012,7 @@ mod tests { "arcadia", "operation-2", "sha256:two", + None, &updated_memory_settings, ) .unwrap(); @@ -17538,6 +18171,7 @@ mod tests { "arcadia", "operation-1", "sha256:one", + None, &updated_memory_settings, ) .is_err() @@ -17568,6 +18202,7 @@ mod tests { "arcadia", "operation-corrupt", "sha256:corrupt", + None, &corrupt, ) .is_err() @@ -17585,6 +18220,189 @@ mod tests { assert!(store.get_workspace_memory_settings("workspace-a").is_err()); } + #[tokio::test] + async fn worker_singleton_ownership_is_workspace_scoped_durable_and_fences_old_generations() { + let dir = tempfile::tempdir().unwrap(); + let database = dir.path().join("server.db"); + let store = SqliteWorkspaceStore::open(&database).unwrap(); + for workspace_id in ["workspace-a", "workspace-b"] { + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: workspace_id.to_string(), + owner_account_id: "owner-account".to_string(), + display_name: workspace_id.to_string(), + state: "active".to_string(), + created_at: "2026-10-01T00:00:00Z".to_string(), + updated_at: "2026-10-01T00:00:00Z".to_string(), + }) + .await + .unwrap(); + } + let key = "subjektiv:主体 id/?!"; + let memory_a = store.get_workspace_memory_settings("workspace-a").unwrap(); + let first = store + .reserve_worker_create( + "workspace-a", + "runtime-a", + "operation-a1", + "sha256:a1", + Some(key), + &memory_a, + ) + .unwrap(); + let first_lease = first.singleton.clone().unwrap(); + assert_eq!(first_lease.key, key); + assert_eq!(first_lease.generation, 1); + assert_eq!( + store + .reserve_worker_create( + "workspace-a", + "runtime-a", + "operation-a1", + "sha256:a1", + Some(key), + &memory_a, + ) + .unwrap(), + first + ); + let conflict = store + .reserve_worker_create( + "workspace-a", + "runtime-b", + "operation-a2", + "sha256:a2", + Some(key), + &memory_a, + ) + .unwrap_err(); + assert!( + matches!(conflict, Error::RepositoryConflict(message) if message.contains("worker_singleton_owned")) + ); + + // The same opaque key is independent in another Workspace. + let memory_b = store.get_workspace_memory_settings("workspace-b").unwrap(); + let other_workspace = store + .reserve_worker_create( + "workspace-b", + "runtime-b", + "operation-b1", + "sha256:b1", + Some(key), + &memory_b, + ) + .unwrap(); + assert_eq!(other_workspace.singleton.unwrap().generation, 1); + + // A failed launch keeps the key pointing at the failed generation until Runtime absence is + // confirmed and the reservation becomes terminal. A fresh operation can then replace it. + store + .fail_worker_create_reservation( + "workspace-a", + "runtime-a", + first.worker_id, + &first.create_fingerprint, + ) + .unwrap(); + assert!( + store + .current_worker_singleton_owner("workspace-a", key) + .unwrap() + .is_none() + ); + let replacement = store + .reserve_worker_create( + "workspace-a", + "runtime-b", + "operation-a3", + "sha256:a3", + Some(key), + &memory_a, + ) + .unwrap(); + let replacement_lease = replacement.singleton.clone().unwrap(); + assert_eq!(replacement_lease.generation, 2); + assert!( + store + .require_current_worker_singleton_owner("workspace-a", &first_lease.worker) + .unwrap_err() + .to_string() + .contains("worker_singleton_fenced") + ); + assert_eq!( + store + .require_current_worker_singleton_owner("workspace-a", &replacement_lease.worker,) + .unwrap(), + Some(replacement_lease.clone()) + ); + drop(store); + + let reopened = SqliteWorkspaceStore::open(database).unwrap(); + assert_eq!( + reopened + .current_worker_singleton_owner("workspace-a", key) + .unwrap(), + Some(replacement_lease) + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn concurrent_worker_singleton_reservations_have_one_winner() { + let dir = tempfile::tempdir().unwrap(); + let database = dir.path().join("server.db"); + let store = SqliteWorkspaceStore::open(&database).unwrap(); + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: "workspace-a".to_string(), + owner_account_id: "owner-account".to_string(), + display_name: "Workspace A".to_string(), + state: "active".to_string(), + created_at: "2026-10-01T00:00:00Z".to_string(), + updated_at: "2026-10-01T00:00:00Z".to_string(), + }) + .await + .unwrap(); + drop(store); + + let barrier = Arc::new(std::sync::Barrier::new(2)); + let handles = (0..2) + .map(|index| { + let database = database.clone(); + let barrier = barrier.clone(); + std::thread::spawn(move || { + let store = SqliteWorkspaceStore::open(database).unwrap(); + let memory = store.get_workspace_memory_settings("workspace-a").unwrap(); + barrier.wait(); + store.reserve_worker_create( + "workspace-a", + if index == 0 { "runtime-a" } else { "runtime-b" }, + &format!("operation-{index}"), + &format!("sha256:{index}"), + Some("shared-key"), + &memory, + ) + }) + }) + .collect::>(); + let results = handles + .into_iter() + .map(|handle| handle.join().unwrap()) + .collect::>(); + assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); + assert_eq!(results.iter().filter(|result| result.is_err()).count(), 1); + let store = SqliteWorkspaceStore::open(database).unwrap(); + let owner = store + .current_worker_singleton_owner("workspace-a", "shared-key") + .unwrap() + .unwrap(); + assert!( + results + .into_iter() + .flatten() + .any(|reservation| { reservation.worker_id.to_string() == owner.worker.worker_id }) + ); + } + #[tokio::test] async fn workspace_flow_sources_keep_revisions_and_builtins_stay_resources() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/workspace-server/src/worker_source.rs b/crates/workspace-server/src/worker_source.rs index 94c2a4e2d..5dcd62d50 100644 --- a/crates/workspace-server/src/worker_source.rs +++ b/crates/workspace-server/src/worker_source.rs @@ -108,6 +108,9 @@ pub async fn verify_runtime_request_source_proof_with_store( if member.is_none() && !reserved { return Err(WorkerMutationSourceProofError::WorkerCatalogMembership); } + store + .require_current_worker_singleton_owner(workspace_id, &worker) + .map_err(|error| WorkerMutationSourceProofError::Authority(error.to_string()))?; } Ok(VerifiedRuntimeRequestSource { runtime_id: claims.iss, @@ -254,6 +257,9 @@ async fn verify_worker_remove_source_with( if member.is_none() { return Err(WorkerMutationSourceProofError::WorkerCatalogMembership); } + store + .require_current_worker_singleton_owner(&config.workspace_id, &worker) + .map_err(|error| WorkerMutationSourceProofError::Authority(error.to_string()))?; let consumed_at = chrono::Utc::now().to_rfc3339(); let consumed = store diff --git a/crates/workspace-server/src/workspace_deletion.rs b/crates/workspace-server/src/workspace_deletion.rs index 797fc7a69..fb5d1a6a0 100644 --- a/crates/workspace-server/src/workspace_deletion.rs +++ b/crates/workspace-server/src/workspace_deletion.rs @@ -80,6 +80,7 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[ "worker_removal_operations", "worker_retention_audit_events", "worker_session_archives", + "worker_singleton_owners", "worker_tombstones", "worker_workdir_attachment_reservations", "worker_workdir_links", diff --git a/openapi/server-api.json b/openapi/server-api.json index 197279460..d7745ca2b 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -1545,6 +1545,14 @@ "runtime_id": { "type": "string" }, + "singleton_key": { + "default": null, + "description": "Opaque Workspace-scoped singleton identity. Possession does not grant Worker access.", + "type": [ + "string", + "null" + ] + }, "ticket_assignment": { "anyOf": [ { @@ -7690,6 +7698,13 @@ "null" ] }, + "singleton_key": { + "description": "Opaque Workspace-scoped singleton identity. The Backend, not the Runtime, owns this lease.", + "type": [ + "string", + "null" + ] + }, "ticket_assignment": { "anyOf": [ { @@ -13989,7 +14004,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:ae559492220ec118f04c2c0c9593e8b3cb44c3cb857be3ae85715ace26da8043" + "x-yoi-source-digest": "sha256:a0956070b29f49aafac3b19a0fa54626f285bdf01337f26afc4389ff6d23495a" }, "openapi": "3.1.0", "paths": { diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index 5bf2a6295..4a43a0719 100644 --- a/web/workspace/src/lib/generated/companion-api.ts +++ b/web/workspace/src/lib/generated/companion-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-companion-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:ae559492220ec118f04c2c0c9593e8b3cb44c3cb857be3ae85715ace26da8043) +// Input contract: 0.1.0 (sha256:a0956070b29f49aafac3b19a0fa54626f285bdf01337f26afc4389ff6d23495a) // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/companion-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check diff --git a/web/workspace/src/lib/generated/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index 06819d7c5..ed25bd296 100644 --- a/web/workspace/src/lib/generated/repository-api.ts +++ b/web/workspace/src/lib/generated/repository-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-repository-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:ae559492220ec118f04c2c0c9593e8b3cb44c3cb857be3ae85715ace26da8043) +// Input contract: 0.1.0 (sha256:a0956070b29f49aafac3b19a0fa54626f285bdf01337f26afc4389ff6d23495a) // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output // Output: web/workspace/src/lib/generated/repository-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check diff --git a/web/workspace/src/lib/generated/runtime-api.ts b/web/workspace/src/lib/generated/runtime-api.ts index c18141c5c..090217650 100644 --- a/web/workspace/src/lib/generated/runtime-api.ts +++ b/web/workspace/src/lib/generated/runtime-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-runtime-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:ae559492220ec118f04c2c0c9593e8b3cb44c3cb857be3ae85715ace26da8043) +// Input contract: 0.1.0 (sha256:a0956070b29f49aafac3b19a0fa54626f285bdf01337f26afc4389ff6d23495a) // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/runtime-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check @@ -115,7 +115,7 @@ export type RuntimeWorkerSpawnAcceptanceRequirement = { kind: "socket_ready"; } export type RuntimeWorkerSpawnIntent = { kind: "workspace_companion"; } | { kind: "workspace_orchestrator"; } | { kind: "workspace_coding"; } | { kind: "ticket_role"; role: RuntimeTicketWorkerRole; ticket_id: string; }; -export type RuntimeWorkerSpawnRequest = { acceptance: RuntimeWorkerSpawnAcceptanceRequirement; initial_submit?: Array; intent: RuntimeWorkerSpawnIntent; profile: RuntimeProfileSelector; requested_worker_name?: string | null; ticket_assignment?: RuntimeWorkerTicketAssignmentRequest | null; workdir_attachment_requests?: Array; }; +export type RuntimeWorkerSpawnRequest = { acceptance: RuntimeWorkerSpawnAcceptanceRequirement; initial_submit?: Array; intent: RuntimeWorkerSpawnIntent; profile: RuntimeProfileSelector; requested_worker_name?: string | null; singleton_key?: string | null; ticket_assignment?: RuntimeWorkerTicketAssignmentRequest | null; workdir_attachment_requests?: Array; }; export type RuntimeWorkerSpawnResponse = { acceptance_evidence?: Array; diagnostics?: Array; state: WorkerOperationState; worker?: RuntimeWorkerSummary | null; }; diff --git a/web/workspace/src/lib/generated/ticket-api.ts b/web/workspace/src/lib/generated/ticket-api.ts index 5b2262e05..014210126 100644 --- a/web/workspace/src/lib/generated/ticket-api.ts +++ b/web/workspace/src/lib/generated/ticket-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:ae559492220ec118f04c2c0c9593e8b3cb44c3cb857be3ae85715ace26da8043) +// Input contract: 0.1.0 (sha256:a0956070b29f49aafac3b19a0fa54626f285bdf01337f26afc4389ff6d23495a) // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/ticket-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check diff --git a/web/workspace/src/lib/generated/workdir-api.ts b/web/workspace/src/lib/generated/workdir-api.ts index ce1b2a795..db5aedb99 100644 --- a/web/workspace/src/lib/generated/workdir-api.ts +++ b/web/workspace/src/lib/generated/workdir-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-workdir-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:7b06d005d3436758c96fc4cfd462df76460f89620fd1b58fdf05009d96bb9052) +// Input contract: 0.1.0 (sha256:a0956070b29f49aafac3b19a0fa54626f285bdf01337f26afc4389ff6d23495a) // Options: workdir-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/workdir-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_workdir_api_types -- --check diff --git a/web/workspace/src/lib/generated/worker-launch-api.ts b/web/workspace/src/lib/generated/worker-launch-api.ts index 61770c180..7085c833c 100644 --- a/web/workspace/src/lib/generated/worker-launch-api.ts +++ b/web/workspace/src/lib/generated/worker-launch-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-worker-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:ae559492220ec118f04c2c0c9593e8b3cb44c3cb857be3ae85715ace26da8043) +// Input contract: 0.1.0 (sha256:a0956070b29f49aafac3b19a0fa54626f285bdf01337f26afc4389ff6d23495a) // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/worker-launch-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check @@ -11,7 +11,7 @@ export type BrowserWorkerWorkingDirectorySelection = { alias: string; relative_c export type BrowserWorkspaceOrchestratorResponse = { diagnostics: Array; disposition: string; online: boolean; worker?: WorkerLaunchWorkerSummary | null; workspace_id: string; }; -export type CreateWorkspaceWorkerRequest = { control_operation_id?: string | null; display_name: string; initial_submit?: Array; profile?: string | null; runtime_id: string; ticket_assignment?: CreateWorkspaceWorkerTicketAssignmentRequest | null; workdir_attachments?: Array; }; +export type CreateWorkspaceWorkerRequest = { control_operation_id?: string | null; display_name: string; initial_submit?: Array; profile?: string | null; runtime_id: string; singleton_key?: string | null; ticket_assignment?: CreateWorkspaceWorkerTicketAssignmentRequest | null; workdir_attachments?: Array; }; export type CreateWorkspaceWorkerTicketAssignmentRequest = { operation_id: string; ticket_id: string; };