From e881c8fad8103e3c8edc98e08de5aa2793b3b4af Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 17 Aug 2026 02:51:18 +0900 Subject: [PATCH] fix: remove speculative worker control delegation --- .../src/feature/builtin/manage_worker.rs | 130 +--- crates/workspace-server/src/server.rs | 556 +----------------- crates/workspace-server/src/store.rs | 313 +++++----- 3 files changed, 155 insertions(+), 844 deletions(-) diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index a66a531c..54b97524 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -435,12 +435,6 @@ impl FeatureModule for ManageWorkerFeature { WorkerOperation::Remove => { definition::(operation, self.control.clone()) } - WorkerOperation::Share | WorkerOperation::Transfer => { - definition::(operation, self.control.clone()) - } - WorkerOperation::Revoke => { - definition::(operation, self.control.clone()) - } }; context .tools() @@ -540,19 +534,6 @@ struct WorkerRemoveInput { reason: String, } -#[derive(Debug, Deserialize, JsonSchema)] -#[serde(deny_unknown_fields)] -struct WorkerRevokeInput { - grant_id: String, -} - -#[derive(Debug, Deserialize, JsonSchema)] -#[serde(deny_unknown_fields)] -struct WorkerDelegateInput { - grant_id: String, - target_controller: WorkerSubjectInput, -} - struct WorkspaceWorkerTool { operation: WorkerOperation, control: Arc, @@ -568,13 +549,10 @@ enum WorkerOperation { Stop, Restore, Remove, - Share, - Transfer, - Revoke, } impl WorkerOperation { - const ALL: [Self; 11] = [ + const ALL: [Self; 8] = [ Self::List, Self::Spawn, Self::SendInput, @@ -583,9 +561,6 @@ impl WorkerOperation { Self::Stop, Self::Restore, Self::Remove, - Self::Share, - Self::Transfer, - Self::Revoke, ]; fn tool_name(self) -> &'static str { @@ -598,9 +573,6 @@ impl WorkerOperation { Self::Stop => "WorkerStop", Self::Restore => "WorkerRestore", Self::Remove => "WorkerRemove", - Self::Share => "WorkerShare", - Self::Transfer => "WorkerTransfer", - Self::Revoke => "WorkerRevoke", } } @@ -622,11 +594,6 @@ impl WorkerOperation { Self::Remove => { "Remove an eligible stopped, unassigned, non-internal Worker. Supply the current Worker revision and a bounded reason; Backend validation and retention are authoritative." } - Self::Share => "Share one controlled Runtime Worker with another known Runtime Worker.", - Self::Transfer => { - "Transfer one controlled Runtime Worker to another known Runtime Worker." - } - Self::Revoke => "Revoke one durable Runtime Worker control grant owned by this Worker.", } } } @@ -792,59 +759,6 @@ impl Tool for WorkspaceWorkerTool { ) .map_err(control_tool_error)? } - WorkerOperation::Share | WorkerOperation::Transfer => { - let input = parse::(input_json, self.operation.tool_name())?; - let grant_id = authority_id(&input.grant_id, "grant_id")?; - let (runtime_id, worker_id) = - runtime_subject_ids(&input.target_controller, self.operation)?; - let operation_id = format!( - "worker-control-{}:{}", - if self.operation == WorkerOperation::Transfer { - "transfer" - } else { - "share" - }, - non_empty(ctx.call_id.clone(), "tool call_id")? - ); - let action = if self.operation == WorkerOperation::Transfer { - "transfer" - } else { - "share" - }; - self.control - .execute_runtime(WorkspaceRequest::json( - WorkspaceRequestMethod::Post, - format!( - "/api/w/{}/worker-control/grants/{grant_id}/{action}", - self.control.workspace_id() - ), - serde_json::json!({ - "target_controller": { - "runtime_id": runtime_id, - "worker_id": worker_id, - }, - "operation_id": operation_id, - }) - .to_string(), - )) - .await - .map_err(control_tool_error)? - } - WorkerOperation::Revoke => { - let input = parse::(input_json, "WorkerRevoke")?; - let grant_id = authority_id(&input.grant_id, "grant_id")?; - self.control - .execute_runtime(WorkspaceRequest::json( - WorkspaceRequestMethod::Post, - format!( - "/api/w/{}/worker-control/grants/{grant_id}/revoke", - self.control.workspace_id() - ), - "{}", - )) - .await - .map_err(control_tool_error)? - } }; tool_output(self.operation, response) } @@ -1131,9 +1045,6 @@ mod tests { "WorkerStop", "WorkerRestore", "WorkerRemove", - "WorkerShare", - "WorkerTransfer", - "WorkerRevoke", ] ); } @@ -1248,45 +1159,6 @@ mod tests { } } - #[tokio::test] - async fn worker_share_and_transfer_use_typed_runtime_subjects_and_operation_ids() { - let client = Arc::new(RecordingWorkspaceClient::default()); - for (operation, action) in [ - (WorkerOperation::Share, "share"), - (WorkerOperation::Transfer, "transfer"), - ] { - WorkspaceWorkerTool { - operation, - control: test_control(client.clone()), - } - .execute( - &serde_json::json!({ - "grant_id": "grant-1", - "target_controller": { - "kind": "runtime_worker", - "runtime_id": "runtime-2", - "worker_id": "worker-9", - }, - }) - .to_string(), - ToolExecutionContext::new("call-delegate", "batch-delegate", 0), - ) - .await - .unwrap(); - let request = client.requests.lock().unwrap().last().cloned().unwrap(); - assert!(request.path.ends_with(&format!("/grant-1/{action}"))); - let body: serde_json::Value = - serde_json::from_str(request.body.as_deref().unwrap()).unwrap(); - assert_eq!(body["target_controller"]["runtime_id"], "runtime-2"); - assert!( - body["operation_id"] - .as_str() - .unwrap() - .contains("call-delegate") - ); - } - } - #[tokio::test] async fn worker_remove_forwards_only_target_revision_and_bounded_reason() { let client = Arc::new(RecordingWorkspaceClient::default()); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index efe0b2ae..c4807de7 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -103,9 +103,8 @@ use crate::skills; use crate::store::{ AccountRecord, ApiTokenRecord, AuthChallengeRecord, BrowserSessionRecord, ControlPlaneStore, DeviceLoginFlowRecord, FlowSourceRecord, PasskeyCredentialRecord, RepositoryRecord, - TicketWorkerAssignmentRecord, UserRecord, WorkdirRegistryRecord, - WorkerControlDelegationOperationRecord, WorkerControlGrantRecord, WorkerRegistryRecord, - WorkerWorkdirLinkRecord, WorkspaceRecord, + TicketWorkerAssignmentRecord, UserRecord, WorkdirRegistryRecord, WorkerControlGrantRecord, + WorkerRegistryRecord, WorkerWorkdirLinkRecord, WorkspaceRecord, }; use crate::{Error, Result}; use worker_runtime::catalog::{ @@ -1477,18 +1476,6 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/worker-control/workers", get(list_known_workers).post(spawn_known_worker), ) - .route( - "/api/w/{workspace_id}/worker-control/grants/{grant_id}/share", - post(share_worker_control_grant), - ) - .route( - "/api/w/{workspace_id}/worker-control/grants/{grant_id}/transfer", - post(transfer_worker_control_grant), - ) - .route( - "/api/w/{workspace_id}/worker-control/grants/{grant_id}/revoke", - post(revoke_worker_control_grant), - ) .route( "/api/w/{workspace_id}/worker-control/workers/{runtime_id}/{worker_id}/input", post(send_known_worker_input), @@ -2258,12 +2245,6 @@ struct ScopedWorkspacePath { workspace_id: String, } -#[derive(Debug, Deserialize)] -struct ScopedWorkerControlGrantPath { - workspace_id: String, - grant_id: String, -} - #[derive(Debug, Deserialize)] struct ScopedFlowPath { workspace_id: String, @@ -5917,7 +5898,6 @@ async fn scoped_workspace_orchestrator_status( #[derive(Debug, Serialize, Deserialize)] struct KnownWorkerRecord { - grant_id: String, subject: RuntimeWorkerRef, relation: String, origin: String, @@ -5952,7 +5932,6 @@ async fn list_known_workers( .worker(&grant.subject) .map_err(|error| error.into_error())?; items.push(KnownWorkerRecord { - grant_id: grant.grant_id, subject: grant.subject, relation: grant.relation, origin: grant.origin, @@ -6028,8 +6007,6 @@ async fn spawn_known_worker( "stop".to_string(), "restore".to_string(), "remove".to_string(), - "share".to_string(), - "transfer".to_string(), "observe".to_string(), ], operation_id, @@ -6083,250 +6060,6 @@ fn authorize_known_worker_permission( Ok(grant) } -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -struct DelegateWorkerControlGrantRequest { - target_controller: RuntimeWorkerRef, - operation_id: String, -} - -async fn share_worker_control_grant( - State(api): State, - AxumPath(path): AxumPath, - headers: HeaderMap, - Json(request): Json, -) -> ApiResult> { - delegate_worker_control_grant(api, path, headers, request, false).await -} - -async fn transfer_worker_control_grant( - State(api): State, - AxumPath(path): AxumPath, - headers: HeaderMap, - Json(request): Json, -) -> ApiResult> { - delegate_worker_control_grant(api, path, headers, request, true).await -} - -fn worker_control_delegation_input_fingerprint( - controller: &RuntimeWorkerRef, - grant: &WorkerControlGrantRecord, - action: &str, - target_controller: &RuntimeWorkerRef, -) -> Result { - let operation_input = serde_json::json!({ - "source_controller": controller, - "source_grant_id": &grant.grant_id, - "action": action, - "target_controller": target_controller, - "subject": &grant.subject, - "permissions": &grant.permissions, - }); - let operation_bytes = serde_json::to_vec(&operation_input).map_err(|error| { - Error::InvalidInput(format!("invalid Worker delegation input: {error}")) - })?; - Ok(format!( - "sha256:{}", - Sha256::digest(&operation_bytes) - .iter() - .map(|byte| format!("{byte:02x}")) - .collect::() - )) -} - -async fn delegate_worker_control_grant( - api: WorkspaceApi, - path: ScopedWorkerControlGrantPath, - headers: HeaderMap, - request: DelegateWorkerControlGrantRequest, - transfer: bool, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; - let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id); - let permission = if transfer { "transfer" } else { "share" }; - let grant = api - .store - .get_worker_control_grant(&path.workspace_id, &path.grant_id)? - .filter(|grant| { - grant.controller == controller - && grant.permissions.iter().any(|value| value == permission) - }) - .ok_or_else(|| Error::UnknownWorker { - worker: controller.clone(), - })?; - if request.target_controller == controller { - return Err(ApiError::from(Error::InvalidInput( - "target_controller must differ from the current Worker".to_string(), - ))); - } - let operation_id = request.operation_id.trim(); - if operation_id.is_empty() || operation_id.len() > 200 { - return Err(ApiError::from(Error::InvalidInput( - "operation_id must contain 1..=200 bytes".to_string(), - ))); - } - let input_fingerprint = worker_control_delegation_input_fingerprint( - &controller, - &grant, - permission, - &request.target_controller, - )?; - let now = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true); - let operation = api.store.reserve_worker_control_delegation_operation( - &WorkerControlDelegationOperationRecord { - workspace_id: path.workspace_id.clone(), - source_controller: controller.clone(), - source_grant_id: grant.grant_id.clone(), - operation_id: operation_id.to_string(), - input_fingerprint, - delegated_grant_id: None, - created_at: now.clone(), - completed_at: None, - }, - )?; - if let Some(delegated_grant_id) = operation.delegated_grant_id.as_deref() { - let delegated = api - .store - .get_worker_control_grant(&path.workspace_id, delegated_grant_id)? - .ok_or_else(|| { - Error::Store("completed Worker delegation references a missing grant".to_string()) - })?; - if transfer && grant.revoked_at.is_none() { - let lock = worker_control_lock(&api, &grant.grant_id); - let _guard = lock.lock().await; - if api - .store - .get_worker_control_grant(&path.workspace_id, &grant.grant_id)? - .is_some_and(|current| current.revoked_at.is_none()) - { - api.store - .revoke_worker_control_grant(&path.workspace_id, &grant.grant_id, &now)?; - } - } - return Ok(Json(delegated)); - } - if grant.revoked_at.is_some() { - return Err(ApiError::from(Error::UnknownWorker { - worker: grant.subject, - })); - } - api.store - .get_active_worker_control_grant( - &path.workspace_id, - &controller, - &request.target_controller, - )? - .ok_or_else(|| Error::UnknownWorker { - worker: request.target_controller.clone(), - })?; - api.store - .get_worker_registry(&path.workspace_id, &request.target_controller)? - .ok_or_else(|| Error::UnknownWorker { - worker: request.target_controller.clone(), - })?; - - let lock = worker_control_lock(&api, &grant.grant_id); - let _guard = lock.lock().await; - let current = api - .store - .get_worker_control_grant(&path.workspace_id, &path.grant_id)? - .filter(|candidate| { - candidate.controller == controller - && candidate.revoked_at.is_none() - && candidate - .permissions - .iter() - .any(|value| value == permission) - }) - .ok_or_else(|| Error::UnknownWorker { - worker: grant.subject.clone(), - })?; - api.store - .get_active_worker_control_grant( - &path.workspace_id, - &controller, - &request.target_controller, - )? - .ok_or_else(|| Error::UnknownWorker { - worker: request.target_controller.clone(), - })?; - let delegated_operation_id = format!( - "worker-control-delegate:{}:{}:{}:{}:{}", - controller.runtime_id, controller.worker_id, current.grant_id, permission, operation_id - ); - let expected_origin = format!("worker_control_{permission}:{}", current.grant_id); - let delegated = api - .store - .create_worker_control_grant(&WorkerControlGrantRecord { - workspace_id: path.workspace_id.clone(), - grant_id: new_id("wcg"), - controller: request.target_controller, - subject: current.subject.clone(), - relation: if transfer { "transferred" } else { "shared" }.to_string(), - origin: expected_origin, - permissions: current.permissions.clone(), - operation_id: delegated_operation_id, - created_at: now.clone(), - revoked_at: None, - })?; - api.store.complete_worker_control_delegation_operation( - &path.workspace_id, - &controller, - operation_id, - &delegated.grant_id, - &now, - )?; - if transfer - && !api - .store - .revoke_worker_control_grant(&path.workspace_id, ¤t.grant_id, &now)? - { - return Err(ApiError::from(Error::UnknownWorker { - worker: current.subject, - })); - } - Ok(Json(delegated)) -} - -async fn revoke_worker_control_grant( - State(api): State, - AxumPath(path): AxumPath, - headers: HeaderMap, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; - let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id); - let grant = api - .store - .get_worker_control_grant(&path.workspace_id, &path.grant_id)? - .filter(|grant| grant.controller == controller && grant.revoked_at.is_none()) - .ok_or_else(|| Error::UnknownWorker { - worker: controller.clone(), - })?; - let lock = worker_control_lock(&api, &grant.grant_id); - let _guard = lock.lock().await; - let current = api - .store - .get_worker_control_grant(&path.workspace_id, &path.grant_id)? - .filter(|candidate| candidate.controller == controller && candidate.revoked_at.is_none()) - .ok_or_else(|| Error::UnknownWorker { - worker: grant.subject.clone(), - })?; - let revoked_at = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true); - if !api - .store - .revoke_worker_control_grant(&path.workspace_id, &path.grant_id, &revoked_at)? - { - return Err(ApiError::from(Error::UnknownWorker { - worker: current.subject, - })); - } - let mut revoked = current; - revoked.revoked_at = Some(revoked_at); - Ok(Json(revoked)) -} - async fn send_known_worker_input( State(api): State, AxumPath(path): AxumPath, @@ -13532,7 +13265,6 @@ mod tests { .await .unwrap(); assert_eq!(known.items.len(), 1); - assert_eq!(known.items[0].grant_id, "orchestrator-controls-generic"); assert_eq!(known.items[0].subject, generic.worker_ref); assert_eq!(known.items[0].permissions, ["observe"]); @@ -13571,17 +13303,15 @@ mod tests { .unwrap(); assert!(capture["entries"].is_array()); - let Json(revoked) = revoke_worker_control_grant( - State(api.clone()), - AxumPath(ScopedWorkerControlGrantPath { - workspace_id: workspace_id.clone(), - grant_id: "orchestrator-controls-generic".to_string(), - }), - observation_headers.clone(), - ) - .await - .unwrap(); - assert!(revoked.revoked_at.is_some()); + let revoked = api + .store + .revoke_worker_control_grant( + &workspace_id, + "orchestrator-controls-generic", + "2026-07-27T00:00:02Z", + ) + .unwrap(); + assert!(revoked); let revoked_capture = scoped_capture_worker_observation_session( State(api.clone()), AxumPath(ScopedWorkspacePath { @@ -13620,270 +13350,6 @@ mod tests { .unwrap(); assert!(unauthorized["sessions"].as_array().unwrap().is_empty()); - for (grant_id, permission) in [ - ("orchestrator-share-source", "share"), - ("orchestrator-transfer-source", "transfer"), - ] { - api.store - .create_worker_control_grant(&WorkerControlGrantRecord { - workspace_id: workspace_id.clone(), - grant_id: grant_id.to_string(), - controller: dedicated.worker.clone(), - subject: generic.worker_ref.clone(), - relation: "spawned".to_string(), - origin: "test-delegation".to_string(), - permissions: vec!["observe".to_string(), permission.to_string()], - operation_id: format!("seed-{permission}"), - created_at: "2026-07-27T00:00:01Z".to_string(), - revoked_at: None, - }) - .unwrap(); - } - let target_controller = generic.worker_ref.clone(); - let Json(shared) = share_worker_control_grant( - State(api.clone()), - AxumPath(ScopedWorkerControlGrantPath { - workspace_id: workspace_id.clone(), - grant_id: "orchestrator-share-source".to_string(), - }), - observation_headers.clone(), - Json(DelegateWorkerControlGrantRequest { - target_controller: target_controller.clone(), - operation_id: "share-operation".to_string(), - }), - ) - .await - .unwrap(); - assert_eq!(shared.controller, target_controller); - assert_eq!(shared.relation, "shared"); - - let alternate_target = RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, "1000"); - let now = now_registry_timestamp(); - api.store - .upsert_worker_registry(&WorkerRegistryRecord { - workspace_id: workspace_id.clone(), - worker: alternate_target.clone(), - display_name: "Alternate known target".to_string(), - profile: None, - retention_state: "normal".to_string(), - transcript_ref: None, - session_ref: None, - summary_ref: None, - diagnostics_ref: None, - created_at: now.clone(), - updated_at: now.clone(), - }) - .unwrap(); - let registry_only_target = RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, "1001"); - api.store - .upsert_worker_registry(&WorkerRegistryRecord { - workspace_id: workspace_id.clone(), - worker: registry_only_target.clone(), - display_name: "Registry-only target".to_string(), - profile: None, - retention_state: "normal".to_string(), - transcript_ref: None, - session_ref: None, - summary_ref: None, - diagnostics_ref: None, - created_at: now.clone(), - updated_at: now.clone(), - }) - .unwrap(); - let unknown_target = share_worker_control_grant( - State(api.clone()), - AxumPath(ScopedWorkerControlGrantPath { - workspace_id: workspace_id.clone(), - grant_id: "orchestrator-share-source".to_string(), - }), - observation_headers.clone(), - Json(DelegateWorkerControlGrantRequest { - target_controller: registry_only_target, - operation_id: "share-registry-only".to_string(), - }), - ) - .await - .unwrap_err(); - assert_eq!( - unknown_target.into_response().status(), - StatusCode::NOT_FOUND - ); - for (grant_id, operation_id, subject) in [ - ( - "orchestrator-knows-alternate", - "seed-known-alternate", - alternate_target.clone(), - ), - ( - "orchestrator-second-share-source", - "seed-second-share", - generic.worker_ref.clone(), - ), - ] { - api.store - .create_worker_control_grant(&WorkerControlGrantRecord { - workspace_id: workspace_id.clone(), - grant_id: grant_id.to_string(), - controller: dedicated.worker.clone(), - subject, - relation: "spawned".to_string(), - origin: "test-delegation-conflict".to_string(), - permissions: vec!["share".to_string()], - operation_id: operation_id.to_string(), - created_at: now.clone(), - revoked_at: None, - }) - .unwrap(); - } - let changed_target = share_worker_control_grant( - State(api.clone()), - AxumPath(ScopedWorkerControlGrantPath { - workspace_id: workspace_id.clone(), - grant_id: "orchestrator-share-source".to_string(), - }), - observation_headers.clone(), - Json(DelegateWorkerControlGrantRequest { - target_controller: alternate_target, - operation_id: "share-operation".to_string(), - }), - ) - .await - .unwrap_err(); - assert_eq!( - changed_target.into_response().status(), - StatusCode::BAD_REQUEST - ); - let changed_source_grant = share_worker_control_grant( - State(api.clone()), - AxumPath(ScopedWorkerControlGrantPath { - workspace_id: workspace_id.clone(), - grant_id: "orchestrator-second-share-source".to_string(), - }), - observation_headers.clone(), - Json(DelegateWorkerControlGrantRequest { - target_controller: generic.worker_ref.clone(), - operation_id: "share-operation".to_string(), - }), - ) - .await - .unwrap_err(); - assert_eq!( - changed_source_grant.into_response().status(), - StatusCode::BAD_REQUEST - ); - - let recovery_source = api - .store - .get_worker_control_grant(&workspace_id, "orchestrator-transfer-source") - .unwrap() - .unwrap(); - let recovery_operation_id = "transfer-operation"; - let recovery_fingerprint = worker_control_delegation_input_fingerprint( - &dedicated.worker, - &recovery_source, - "transfer", - &generic.worker_ref, - ) - .unwrap(); - let recovery_now = now_registry_timestamp(); - api.store - .reserve_worker_control_delegation_operation(&WorkerControlDelegationOperationRecord { - workspace_id: workspace_id.clone(), - source_controller: dedicated.worker.clone(), - source_grant_id: recovery_source.grant_id.clone(), - operation_id: recovery_operation_id.to_string(), - input_fingerprint: recovery_fingerprint, - delegated_grant_id: None, - created_at: recovery_now.clone(), - completed_at: None, - }) - .unwrap(); - let precompleted_transfer = api - .store - .create_worker_control_grant(&WorkerControlGrantRecord { - workspace_id: workspace_id.clone(), - grant_id: "precompleted-transfer-grant".to_string(), - controller: generic.worker_ref.clone(), - subject: recovery_source.subject.clone(), - relation: "transferred".to_string(), - origin: format!("worker_control_transfer:{}", recovery_source.grant_id), - permissions: recovery_source.permissions.clone(), - operation_id: format!( - "worker-control-delegate:{}:{}:{}:transfer:{}", - dedicated.worker.runtime_id, - dedicated.worker.worker_id, - recovery_source.grant_id, - recovery_operation_id, - ), - created_at: recovery_now.clone(), - revoked_at: None, - }) - .unwrap(); - api.store - .complete_worker_control_delegation_operation( - &workspace_id, - &dedicated.worker, - recovery_operation_id, - &precompleted_transfer.grant_id, - &recovery_now, - ) - .unwrap(); - assert!( - api.store - .get_worker_control_grant(&workspace_id, &recovery_source.grant_id) - .unwrap() - .unwrap() - .revoked_at - .is_none() - ); - - let Json(transferred) = transfer_worker_control_grant( - State(api.clone()), - AxumPath(ScopedWorkerControlGrantPath { - workspace_id: workspace_id.clone(), - grant_id: "orchestrator-transfer-source".to_string(), - }), - observation_headers.clone(), - Json(DelegateWorkerControlGrantRequest { - target_controller: generic.worker_ref.clone(), - operation_id: "transfer-operation".to_string(), - }), - ) - .await - .unwrap(); - assert_eq!(transferred.grant_id, precompleted_transfer.grant_id); - assert!( - api.store - .get_worker_control_grant(&workspace_id, &recovery_source.grant_id) - .unwrap() - .unwrap() - .revoked_at - .is_some() - ); - let Json(transfer_replay) = transfer_worker_control_grant( - State(api.clone()), - AxumPath(ScopedWorkerControlGrantPath { - workspace_id: workspace_id.clone(), - grant_id: "orchestrator-transfer-source".to_string(), - }), - observation_headers, - Json(DelegateWorkerControlGrantRequest { - target_controller: generic.worker_ref.clone(), - operation_id: "transfer-operation".to_string(), - }), - ) - .await - .unwrap(); - assert_eq!(transfer_replay.grant_id, transferred.grant_id); - assert!( - api.store - .get_worker_control_grant(&workspace_id, "orchestrator-transfer-source") - .unwrap() - .unwrap() - .revoked_at - .is_some() - ); - let Json(existing) = scoped_start_workspace_orchestrator( State(api.clone()), AxumPath(ScopedWorkspacePath { diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index eb89c8f8..43228d9c 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -191,6 +191,11 @@ const MIGRATIONS: &[Migration] = &[ name: "create Worker control delegation operation authority", apply: create_worker_control_delegation_operation_authority, }, + Migration { + version: 35, + name: "remove Worker control delegation authority", + apply: remove_worker_control_delegation_authority, + }, ]; struct Migration { @@ -352,18 +357,6 @@ pub struct WorkerControlGrantRecord { pub revoked_at: Option, } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub struct WorkerControlDelegationOperationRecord { - pub workspace_id: String, - pub source_controller: RuntimeWorkerRef, - pub source_grant_id: String, - pub operation_id: String, - pub input_fingerprint: String, - pub delegated_grant_id: Option, - pub created_at: String, - pub completed_at: Option, -} - #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct TicketWorkerAssignmentRecord { pub workspace_id: String, @@ -804,19 +797,6 @@ pub trait ControlPlaneStore: Send + Sync { grant_id: &str, revoked_at: &str, ) -> Result; - fn reserve_worker_control_delegation_operation( - &self, - record: &WorkerControlDelegationOperationRecord, - ) -> Result; - fn complete_worker_control_delegation_operation( - &self, - workspace_id: &str, - source_controller: &RuntimeWorkerRef, - operation_id: &str, - delegated_grant_id: &str, - completed_at: &str, - ) -> Result; - fn get_ticket_assignment_operation( &self, workspace_id: &str, @@ -2578,95 +2558,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } - fn reserve_worker_control_delegation_operation( - &self, - record: &WorkerControlDelegationOperationRecord, - ) -> Result { - self.with_conn(|conn| { - conn.execute( - r#"INSERT INTO worker_control_delegation_operations ( - workspace_id, source_controller_runtime_id, source_controller_worker_id, - source_grant_id, operation_id, input_fingerprint, - delegated_grant_id, created_at, completed_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) - ON CONFLICT ( - workspace_id, source_controller_runtime_id, - source_controller_worker_id, operation_id - ) DO NOTHING"#, - params![ - record.workspace_id, - record.source_controller.runtime_id, - record.source_controller.worker_id, - record.source_grant_id, - record.operation_id, - record.input_fingerprint, - record.delegated_grant_id, - record.created_at, - record.completed_at, - ], - )?; - let persisted = read_worker_control_delegation_operation_by_key( - conn, - &record.workspace_id, - &record.source_controller, - &record.operation_id, - )? - .ok_or_else(|| { - Error::Store("worker control delegation operation was not persisted".to_string()) - })?; - if persisted.source_grant_id != record.source_grant_id - || persisted.input_fingerprint != record.input_fingerprint - { - return Err(Error::InvalidInput(format!( - "worker control delegation operation `{}` was already used with different input", - record.operation_id - ))); - } - Ok(persisted) - }) - } - - fn complete_worker_control_delegation_operation( - &self, - workspace_id: &str, - source_controller: &RuntimeWorkerRef, - operation_id: &str, - delegated_grant_id: &str, - completed_at: &str, - ) -> Result { - self.with_conn(|conn| { - conn.execute( - r#"UPDATE worker_control_delegation_operations - SET delegated_grant_id = ?5, completed_at = ?6 - WHERE workspace_id = ?1 - AND source_controller_runtime_id = ?2 - AND source_controller_worker_id = ?3 - AND operation_id = ?4 - AND (delegated_grant_id IS NULL OR delegated_grant_id = ?5)"#, - params![ - workspace_id, - source_controller.runtime_id, - source_controller.worker_id, - operation_id, - delegated_grant_id, - completed_at, - ], - )?; - read_worker_control_delegation_operation_by_key( - conn, - workspace_id, - source_controller, - operation_id, - )? - .filter(|record| record.delegated_grant_id.as_deref() == Some(delegated_grant_id)) - .ok_or_else(|| { - Error::InvalidInput(format!( - "worker control delegation operation `{operation_id}` completed with a different grant" - )) - }) - }) - } - fn get_ticket_assignment_operation( &self, workspace_id: &str, @@ -4024,52 +3915,6 @@ fn read_worker_control_grant_by_operation( .map_err(Error::from) } -fn read_worker_control_delegation_operation_record( - row: &rusqlite::Row<'_>, -) -> rusqlite::Result { - Ok(WorkerControlDelegationOperationRecord { - workspace_id: row.get(0)?, - source_controller: RuntimeWorkerRef::new( - row.get::<_, String>(1)?, - row.get::<_, u64>(2)?.to_string(), - ), - source_grant_id: row.get(3)?, - operation_id: row.get(4)?, - input_fingerprint: row.get(5)?, - delegated_grant_id: row.get(6)?, - created_at: row.get(7)?, - completed_at: row.get(8)?, - }) -} - -fn read_worker_control_delegation_operation_by_key( - conn: &Connection, - workspace_id: &str, - source_controller: &RuntimeWorkerRef, - operation_id: &str, -) -> Result> { - conn.query_row( - r#"SELECT workspace_id, - source_controller_runtime_id, source_controller_worker_id, - source_grant_id, operation_id, input_fingerprint, - delegated_grant_id, created_at, completed_at - FROM worker_control_delegation_operations - WHERE workspace_id = ?1 - AND source_controller_runtime_id = ?2 - AND source_controller_worker_id = ?3 - AND operation_id = ?4"#, - params![ - workspace_id, - source_controller.runtime_id, - source_controller.worker_id, - operation_id, - ], - read_worker_control_delegation_operation_record, - ) - .optional() - .map_err(Error::from) -} - fn current_ticket_worker_assignment_select_sql() -> String { "SELECT a.workspace_id, a.ticket_id, a.assignment_id, a.runtime_id, a.worker_id, \ a.assigned_by, a.assigned_at \ @@ -5168,6 +5013,58 @@ fn create_worker_control_delegation_operation_authority(conn: &Connection) -> Re Ok(()) } +fn remove_worker_control_delegation_authority(conn: &Connection) -> Result<()> { + let mut statement = + conn.prepare("SELECT workspace_id, grant_id, permissions_json FROM worker_control_grants")?; + let rows = statement.query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + )) + })?; + let mut permission_updates = Vec::new(); + for row in rows { + let (workspace_id, grant_id, permissions_json) = row?; + let mut permissions: Vec = + serde_json::from_str(&permissions_json).map_err(|error| { + Error::Store(format!( + "failed to decode Worker control grant `{grant_id}` permissions during delegation removal: {error}" + )) + })?; + let previous_len = permissions.len(); + permissions + .retain(|permission| !matches!(permission.as_str(), "share" | "transfer" | "revoke")); + if permissions.len() != previous_len { + permission_updates.push(( + workspace_id, + grant_id, + serde_json::to_string(&permissions).map_err(|error| { + Error::Store(format!( + "failed to encode Worker control grant permissions during delegation removal: {error}" + )) + })?, + )); + } + } + drop(statement); + + for (workspace_id, grant_id, permissions_json) in permission_updates { + conn.execute( + "UPDATE worker_control_grants SET permissions_json = ?3 WHERE workspace_id = ?1 AND grant_id = ?2", + params![workspace_id, grant_id, permissions_json], + )?; + } + conn.execute( + r#"UPDATE worker_control_grants + SET revoked_at = COALESCE(revoked_at, strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) + WHERE relation IN ('shared', 'transferred')"#, + [], + )?; + conn.execute_batch("DROP TABLE IF EXISTS worker_control_delegation_operations;")?; + Ok(()) +} + fn create_worker_mutation_source_proof_replay_guard(conn: &Connection) -> Result<()> { conn.execute_batch( r#" @@ -5819,6 +5716,82 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); } } + #[test] + fn schema_v35_removes_worker_control_delegation_authority() { + let conn = Connection::open_in_memory().unwrap(); + configure_sqlite(&conn).unwrap(); + for migration in MIGRATIONS + .iter() + .filter(|migration| migration.version <= 34) + { + let tx = conn.unchecked_transaction().unwrap(); + (migration.apply)(&tx).unwrap(); + tx.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", + params![migration.version, migration.name], + ) + .unwrap(); + tx.commit().unwrap(); + } + conn.execute_batch( + r#" +INSERT INTO workspaces ( + workspace_id, display_name, state, created_at, updated_at +) VALUES ('workspace-a', 'Workspace A', 'active', '1', '1'); +INSERT INTO worker_registry ( + workspace_id, runtime_id, runtime_worker_id, display_name, + retention_state, created_at, updated_at +) VALUES + ('workspace-a', 'runtime-a', 1, 'Controller', 'normal', '1', '1'), + ('workspace-a', 'runtime-a', 2, 'Spawned Worker', 'normal', '1', '1'), + ('workspace-a', 'runtime-a', 3, 'Shared Worker', 'normal', '1', '1'), + ('workspace-a', 'runtime-a', 4, 'Transferred Worker', 'normal', '1', '1'); +INSERT INTO worker_control_grants ( + workspace_id, grant_id, + controller_runtime_id, controller_worker_id, + subject_runtime_id, subject_worker_id, + relation, origin, permissions_json, operation_id, created_at, revoked_at +) VALUES + ('workspace-a', 'spawned', 'runtime-a', 1, 'runtime-a', 2, + 'spawned', 'spawn', '["observe","share","transfer","revoke","stop"]', 'spawn-op', '1', NULL), + ('workspace-a', 'shared', 'runtime-a', 1, 'runtime-a', 3, + 'shared', 'share', '["observe"]', 'share-op', '1', NULL), + ('workspace-a', 'transferred', 'runtime-a', 1, 'runtime-a', 4, + 'transferred', 'transfer', '["observe"]', 'transfer-op', '1', NULL); +"#, + ) + .unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 34); + assert!(table_exists(&conn, "worker_control_delegation_operations").unwrap()); + + apply_migrations(&conn).unwrap(); + + assert_eq!(current_schema_version(&conn).unwrap(), 35); + assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap()); + let (permissions_json, revoked_at): (String, Option) = conn + .query_row( + "SELECT permissions_json, revoked_at FROM worker_control_grants WHERE grant_id = 'spawned'", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!( + serde_json::from_str::>(&permissions_json).unwrap(), + vec!["observe", "stop"] + ); + assert!(revoked_at.is_none()); + for grant_id in ["shared", "transferred"] { + let revoked_at: Option = conn + .query_row( + "SELECT revoked_at FROM worker_control_grants WHERE grant_id = ?1", + [grant_id], + |row| row.get(0), + ) + .unwrap(); + assert!(revoked_at.is_some(), "{grant_id} grant remained active"); + } + } + #[test] fn schema_v24_adds_attachment_reservations_to_already_applied_v23() { let conn = Connection::open_in_memory().unwrap(); @@ -5841,7 +5814,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 34); + assert_eq!(current_schema_version(&conn).unwrap(), 35); assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap()); } @@ -5874,7 +5847,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 34); + assert_eq!(current_schema_version(&conn).unwrap(), 35); assert!(table_exists(&conn, "flow_sources").unwrap()); assert!(table_exists(&conn, "flow_source_revisions").unwrap()); assert!(!table_exists(&conn, "flow_instances").unwrap()); @@ -5941,7 +5914,7 @@ INSERT INTO worker_workdir_attachment_reservations ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 34); + assert_eq!(current_schema_version(&conn).unwrap(), 35); let repositories_sql: String = conn .query_row( "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'", @@ -6121,7 +6094,7 @@ INSERT INTO workdir_registry ( let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 34); + assert_eq!(store.schema_version().await.unwrap(), 35); assert!( !store .with_conn(|conn| table_exists(conn, "worker_workspace_credentials")) @@ -6138,7 +6111,7 @@ INSERT INTO workdir_registry ( store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 34); + assert_eq!(reopened.schema_version().await.unwrap(), 35); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) @@ -6685,7 +6658,7 @@ INSERT INTO workdir_registry ( .unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 34); + assert_eq!(store.schema_version().await.unwrap(), 35); store .with_conn(|conn| { @@ -6874,7 +6847,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn repository_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 34); + assert_eq!(store.schema_version().await.unwrap(), 35); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -6940,7 +6913,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn memory_authority_records_round_trip_and_close_staging() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 34); + assert_eq!(store.schema_version().await.unwrap(), 35); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -7201,7 +7174,7 @@ CREATE TABLE ticket_assignment_operations ( } #[tokio::test] - async fn worker_control_grants_are_idempotent_scoped_and_revocable() { + async fn worker_control_grants_are_idempotent_scoped_and_support_internal_invalidation() { let dir = tempfile::tempdir().unwrap(); let database = dir.path().join("control-grants.db"); let store = SqliteWorkspaceStore::open(&database).unwrap(); @@ -7331,7 +7304,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn account_and_login_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 34); + assert_eq!(store.schema_version().await.unwrap(), 35); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { account_id: "acct-user-alice".to_string(),