feat: enforce arbitrary worker singleton ownership

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<String>,
+    /// 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<String>,
     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,

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 (

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])?;

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

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",