diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index c1c6e9e1b..471f2ff06 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -2077,6 +2077,7 @@ pub trait ServerApi { )] async fn runtime_worker_restore_alias( &self, + #[extension] context: ServerRequestContext, #[path] runtime_id: String, #[path] worker_id: String, ) -> Result; @@ -2091,6 +2092,7 @@ pub trait ServerApi { )] async fn runtime_worker_restore( &self, + #[extension] context: ServerRequestContext, #[path] workspace_id: String, #[path] runtime_id: String, #[path] worker_id: String, @@ -2107,6 +2109,7 @@ pub trait ServerApi { )] async fn runtime_worker_pin( &self, + #[extension] context: ServerRequestContext, #[path] workspace_id: String, #[path] runtime_id: String, #[path] worker_id: String, @@ -2122,6 +2125,7 @@ pub trait ServerApi { )] async fn runtime_worker_unpin( &self, + #[extension] context: ServerRequestContext, #[path] workspace_id: String, #[path] runtime_id: String, #[path] worker_id: String, @@ -2151,6 +2155,7 @@ pub trait ServerApi { )] async fn runtime_cleanup_execute( &self, + #[extension] context: ServerRequestContext, #[path] workspace_id: String, #[path] runtime_id: String, #[body] request: ExecuteRuntimeCleanupRequest, @@ -2166,6 +2171,7 @@ pub trait ServerApi { )] async fn runtime_worker_input_alias( &self, + #[extension] context: ServerRequestContext, #[path] runtime_id: String, #[path] worker_id: String, #[body] request: RuntimeWorkerInputRequest, @@ -2181,6 +2187,7 @@ pub trait ServerApi { )] async fn runtime_worker_input( &self, + #[extension] context: ServerRequestContext, #[path] workspace_id: String, #[path] runtime_id: String, #[path] worker_id: String, @@ -2297,6 +2304,7 @@ pub trait ServerApi { )] async fn runtime_worker_stop_alias( &self, + #[extension] context: ServerRequestContext, #[path] runtime_id: String, #[path] worker_id: String, #[body] request: RuntimeWorkerLifecycleRequest, @@ -2312,6 +2320,7 @@ pub trait ServerApi { )] async fn runtime_worker_stop( &self, + #[extension] context: ServerRequestContext, #[path] workspace_id: String, #[path] runtime_id: String, #[path] worker_id: String, @@ -2328,6 +2337,7 @@ pub trait ServerApi { )] async fn runtime_worker_cancel_alias( &self, + #[extension] context: ServerRequestContext, #[path] runtime_id: String, #[path] worker_id: String, #[body] request: RuntimeWorkerLifecycleRequest, @@ -2343,6 +2353,7 @@ pub trait ServerApi { )] async fn runtime_worker_cancel( &self, + #[extension] context: ServerRequestContext, #[path] workspace_id: String, #[path] runtime_id: String, #[path] worker_id: String, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index ed7afcead..fac133171 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1,3 +1,8 @@ +mod worker_operations; +pub(crate) use worker_operations::{ + WorkerOperationContext, WorkspaceWorker, WorkspaceWorkerMethodSender, +}; + #[path = "server_workspace_config.rs"] mod workspace_config; use std::collections::{BTreeMap, HashMap, HashSet}; @@ -2130,35 +2135,6 @@ impl WorkerRemovalService { } } -impl crate::worker_source::VerifiedWorkerRemoveExecutor for WorkerRemovalService { - fn execute( - &self, - source: crate::worker_source::VerifiedWorkerMutationSource, - target_runtime_id: &str, - target_worker_id: &str, - reason: &str, - ) -> std::result::Result { - let executor = self.clone(); - let target_runtime_id = target_runtime_id.to_string(); - let target_worker_id = target_worker_id.to_string(); - let reason = reason.to_string(); - std::thread::spawn(move || { - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .map_err(|error| error.to_string())? - .block_on(executor.execute_async( - source, - &target_runtime_id, - &target_worker_id, - &reason, - )) - }) - .join() - .map_err(|_| "embedded WorkerRemove executor thread panicked".to_string())? - } -} - #[derive(Clone)] pub struct WorkspaceServerApi { template: Arc, @@ -3453,7 +3429,9 @@ impl WorkspaceApi { }; if let Some(dispatcher) = worker_remove_dispatcher { dispatcher - .install_executor(Arc::new(WorkerRemovalService::new(&api))) + .install_executor(Arc::new( + worker_operations::EmbeddedWorkspaceWorkerRemoveExecutor::new(&api), + )) .map_err(|message| Error::Config(message.to_string()))?; } let expired_external_workdirs = api.store.reconcile_external_workdir_grants_after_restart( @@ -6674,7 +6652,18 @@ fn worker_spawn_result_to_api( }) } -fn worker_input_result_to_api(result: WorkerInputResult) -> server_api::RuntimeWorkerInputResult { +fn worker_input_result_to_api( + mut result: WorkerInputResult, +) -> server_api::RuntimeWorkerInputResult { + // The existing wire result has no disposition field. Keep uncertain delivery + // distinguishable without collapsing it into a definite refusal or replaying input. + if result.disposition == WorkerInputDisposition::Unknown { + result.diagnostics.push(RuntimeDiagnostic { + code: "worker_input_outcome_unknown".into(), + severity: HostDiagnosticSeverity::Warning, + message: "Worker input delivery outcome is unknown; inspect the current session before retrying".into(), + }); + } server_api::RuntimeWorkerInputResult { state: result.state.into(), runtime_id: result.worker.runtime_id, @@ -8127,7 +8116,7 @@ impl server_api::ServerApi for ServerApiContractService { let Json(response) = list_known_workers( State(self.workspace_api()?.clone()), AxumPath(ScopedWorkspacePath { workspace_id }), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, ) .await .map_err(ApiError::into_repository_api_error)?; @@ -8144,7 +8133,7 @@ impl server_api::ServerApi for ServerApiContractService { spawn_known_worker( State(self.workspace_api()?.clone()), AxumPath(ScopedWorkspacePath { workspace_id }), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, Json(request), ) .await @@ -8167,7 +8156,7 @@ impl server_api::ServerApi for ServerApiContractService { workspace_id, worker: RuntimeWorkerRef::new(runtime_id, worker_id), }), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, Json(request), ) .await @@ -8190,7 +8179,7 @@ impl server_api::ServerApi for ServerApiContractService { workspace_id, worker: RuntimeWorkerRef::new(runtime_id, worker_id), }), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, Json(request), ) .await @@ -8213,7 +8202,7 @@ impl server_api::ServerApi for ServerApiContractService { workspace_id, worker: RuntimeWorkerRef::new(runtime_id, worker_id), }), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, Json(request), ) .await @@ -8235,7 +8224,7 @@ impl server_api::ServerApi for ServerApiContractService { workspace_id, worker: RuntimeWorkerRef::new(runtime_id, worker_id), }), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, ) .await .map(|Json(response)| response) @@ -8253,7 +8242,7 @@ impl server_api::ServerApi for ServerApiContractService { let Json(response) = scoped_list_worker_observation_sessions( State(self.workspace_api()?.clone()), AxumPath(ScopedWorkspacePath { workspace_id }), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, ) .await .map_err(ApiError::into_repository_api_error)?; @@ -8272,7 +8261,7 @@ impl server_api::ServerApi for ServerApiContractService { let Json(response) = scoped_capture_worker_observation_session( State(self.workspace_api()?.clone()), AxumPath(ScopedWorkspacePath { workspace_id }), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, Json(request), ) .await @@ -8947,34 +8936,40 @@ impl server_api::ServerApi for ServerApiContractService { async fn runtime_worker_restore_alias( &self, + context: server_api::ServerRequestContext, runtime_id: String, worker_id: String, ) -> std::result::Result { - restore_runtime_worker( - State(self.workspace_api()?.clone()), - AxumPath((runtime_id, worker_id)), - ) - .await - .map(|Json(response)| response) - .map_err(ApiError::into_repository_api_error) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + restore_runtime_worker_with_context(api, runtime_id, worker_id, operation_context) + .await + .map(|Json(response)| response) + .map_err(ApiError::into_repository_api_error) } async fn runtime_worker_restore( &self, + context: server_api::ServerRequestContext, workspace_id: String, runtime_id: String, worker_id: String, query: server_api::RestoreTicketAssignmentQuery, ) -> std::result::Result { - scoped_restore_runtime_worker( - State(self.workspace_api()?.clone()), - AxumPath(ScopedRuntimeWorkerPath { + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + validate_workspace_scope(&api, &workspace_id) + .map_err(ApiError::into_repository_api_error)?; + scoped_restore_runtime_worker_with_context( + api, + ScopedRuntimeWorkerPath { workspace_id, worker: RuntimeWorkerRef::new(runtime_id, worker_id), - }), - Query(query), + }, + query, + operation_context, ) .await .map(|Json(response)| response) @@ -8983,40 +8978,42 @@ impl server_api::ServerApi for ServerApiContractService { async fn runtime_worker_pin( &self, + context: server_api::ServerRequestContext, workspace_id: String, runtime_id: String, worker_id: String, ) -> std::result::Result { - let Json(response) = scoped_pin_runtime_worker( - State(self.workspace_api()?.clone()), - AxumPath(ScopedRuntimeWorkerPath { - workspace_id, - worker: RuntimeWorkerRef::new(runtime_id, worker_id), - }), - ) - .await - .map_err(ApiError::into_repository_api_error)?; - Ok(response) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + validate_workspace_scope(&api, &workspace_id) + .map_err(ApiError::into_repository_api_error)?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + worker + .set_pinned(&operation_context, true) + .await + .map_err(ApiError::into_repository_api_error) } async fn runtime_worker_unpin( &self, + context: server_api::ServerRequestContext, workspace_id: String, runtime_id: String, worker_id: String, ) -> std::result::Result { - let Json(response) = scoped_unpin_runtime_worker( - State(self.workspace_api()?.clone()), - AxumPath(ScopedRuntimeWorkerPath { - workspace_id, - worker: RuntimeWorkerRef::new(runtime_id, worker_id), - }), - ) - .await - .map_err(ApiError::into_repository_api_error)?; - Ok(response) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + validate_workspace_scope(&api, &workspace_id) + .map_err(ApiError::into_repository_api_error)?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + worker + .set_pinned(&operation_context, false) + .await + .map_err(ApiError::into_repository_api_error) } async fn runtime_cleanup_plan( @@ -9039,6 +9036,7 @@ impl server_api::ServerApi for ServerApiContractService { async fn runtime_cleanup_execute( &self, + context: server_api::ServerRequestContext, workspace_id: String, runtime_id: String, request: server_api::ExecuteRuntimeCleanupRequest, @@ -9046,55 +9044,58 @@ impl server_api::ServerApi for ServerApiContractService { server_api::RuntimeCleanupExecutionResponse, server_api::RepositoryApiError, > { - let Json(response) = scoped_execute_runtime_cleanup( - State(self.workspace_api()?.clone()), - AxumPath(ScopedRuntimePath { - workspace_id, - runtime_id, - }), - Json(request), - ) - .await - .map_err(ApiError::into_repository_api_error)?; - Ok(response) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + validate_workspace_scope(&api, &workspace_id) + .map_err(ApiError::into_repository_api_error)?; + execute_runtime_cleanup_with_context(&api, &runtime_id, request, &operation_context) + .await + .map_err(ApiError::into_repository_api_error) } async fn runtime_worker_input_alias( &self, + context: server_api::ServerRequestContext, runtime_id: String, worker_id: String, request: server_api::RuntimeWorkerInputRequest, ) -> std::result::Result { - let Json(response) = send_runtime_worker_input( - State(self.workspace_api()?.clone()), - AxumPath((runtime_id, worker_id)), - Json(request), - ) - .await - .map_err(ApiError::into_repository_api_error)?; - Ok(response) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + let request = worker_input_request_from_api(request) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + worker + .input(&operation_context, request) + .await + .map(worker_input_result_to_api) + .map_err(ApiError::into_repository_api_error) } async fn runtime_worker_input( &self, + context: server_api::ServerRequestContext, workspace_id: String, runtime_id: String, worker_id: String, request: server_api::RuntimeWorkerInputRequest, ) -> std::result::Result { - let Json(response) = scoped_send_runtime_worker_input( - State(self.workspace_api()?.clone()), - AxumPath(ScopedRuntimeWorkerPath { - workspace_id, - worker: RuntimeWorkerRef::new(runtime_id, worker_id), - }), - Json(request), - ) - .await - .map_err(ApiError::into_repository_api_error)?; - Ok(response) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + validate_workspace_scope(&api, &workspace_id) + .map_err(ApiError::into_repository_api_error)?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + let request = worker_input_request_from_api(request) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + worker + .input(&operation_context, request) + .await + .map(worker_input_result_to_api) + .map_err(ApiError::into_repository_api_error) } async fn runtime_worker_attachment_upload_grant( @@ -9239,78 +9240,96 @@ impl server_api::ServerApi for ServerApiContractService { async fn runtime_worker_stop_alias( &self, + context: server_api::ServerRequestContext, runtime_id: String, worker_id: String, request: server_api::RuntimeWorkerLifecycleRequest, ) -> std::result::Result { - let Json(response) = stop_runtime_worker( - State(self.workspace_api()?.clone()), - AxumPath((runtime_id, worker_id)), - Json(request), - ) - .await - .map_err(ApiError::into_repository_api_error)?; - Ok(response) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + worker + .stop( + &operation_context, + worker_lifecycle_request_from_api(request), + ) + .await + .map(worker_lifecycle_result_to_api) + .map_err(ApiError::into_repository_api_error) } async fn runtime_worker_stop( &self, + context: server_api::ServerRequestContext, workspace_id: String, runtime_id: String, worker_id: String, request: server_api::RuntimeWorkerLifecycleRequest, ) -> std::result::Result { - let Json(response) = scoped_stop_runtime_worker( - State(self.workspace_api()?.clone()), - AxumPath(ScopedRuntimeWorkerPath { - workspace_id, - worker: RuntimeWorkerRef::new(runtime_id, worker_id), - }), - Json(request), - ) - .await - .map_err(ApiError::into_repository_api_error)?; - Ok(response) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + validate_workspace_scope(&api, &workspace_id) + .map_err(ApiError::into_repository_api_error)?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + worker + .stop( + &operation_context, + worker_lifecycle_request_from_api(request), + ) + .await + .map(worker_lifecycle_result_to_api) + .map_err(ApiError::into_repository_api_error) } async fn runtime_worker_cancel_alias( &self, + context: server_api::ServerRequestContext, runtime_id: String, worker_id: String, request: server_api::RuntimeWorkerLifecycleRequest, ) -> std::result::Result { - let Json(response) = cancel_runtime_worker( - State(self.workspace_api()?.clone()), - AxumPath((runtime_id, worker_id)), - Json(request), - ) - .await - .map_err(ApiError::into_repository_api_error)?; - Ok(response) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + worker + .cancel( + &operation_context, + worker_lifecycle_request_from_api(request), + ) + .await + .map(worker_lifecycle_result_to_api) + .map_err(ApiError::into_repository_api_error) } async fn runtime_worker_cancel( &self, + context: server_api::ServerRequestContext, workspace_id: String, runtime_id: String, worker_id: String, request: server_api::RuntimeWorkerLifecycleRequest, ) -> std::result::Result { - let Json(response) = scoped_cancel_runtime_worker( - State(self.workspace_api()?.clone()), - AxumPath(ScopedRuntimeWorkerPath { - workspace_id, - worker: RuntimeWorkerRef::new(runtime_id, worker_id), - }), - Json(request), - ) - .await - .map_err(ApiError::into_repository_api_error)?; - Ok(response) + let api = self.workspace_api()?.clone(); + let operation_context = WorkerOperationContext::from_request(&api, &context)?; + validate_workspace_scope(&api, &workspace_id) + .map_err(ApiError::into_repository_api_error)?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id) + .map_err(|error| ApiError::from(error).into_repository_api_error())?; + worker + .cancel( + &operation_context, + worker_lifecycle_request_from_api(request), + ) + .await + .map(worker_lifecycle_result_to_api) + .map_err(ApiError::into_repository_api_error) } async fn repository_list( @@ -22229,13 +22248,17 @@ async fn scoped_workspace_protocol_ws( State(api): State, Extension(actor): Extension, AxumPath(workspace_id): AxumPath, + headers: HeaderMap, ws: axum::extract::ws::WebSocketUpgrade, ) -> std::result::Result { validate_workspace_scope(&api, &workspace_id).map_err(|error| error.into_response())?; - let input_source = authenticated_browser_input_source(&actor); + let context = WorkerOperationContext::Browser { + headers, + source: authenticated_browser_input_source(&actor), + }; Ok(ws .on_upgrade(move |socket| { - crate::workspace_subscription::serve_workspace_subscription(api, socket, input_source) + crate::workspace_subscription::serve_workspace_subscription(api, socket, context) }) .into_response()) } @@ -22380,13 +22403,22 @@ async fn worker_remove_contract( None, ) })?; - let response = WorkerRemovalService::new(api) - .execute_async( - source, - &request.target_runtime_id, - &request.target_worker_id, - &request.reason, - ) + let worker = + WorkspaceWorker::resolve(api, &request.target_runtime_id, &request.target_worker_id) + .map_err(|error| match error { + Error::UnknownWorker { .. } => server_api::WorkerRemoveApiError::new( + StatusCode::NOT_FOUND.as_u16(), + None, + Some("unknown_worker".into()), + Some("The target Worker is not known in this Workspace".into()), + None, + ), + error => worker_remove_repository_error( + ApiError::from(error).into_repository_api_error(), + ), + })?; + let response = worker + .remove(source, &request.reason) .await .unwrap_or_else(|_| { worker_remove_error_response( @@ -22736,36 +22768,16 @@ async fn send_known_worker_input( ) -> 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 request.kind.as_deref() == Some("notify") { - "notify" - } else { - "send_input" + let context = WorkerOperationContext::WorkerControl { + controller: RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id), }; - let grant = authorize_known_worker_permission( - &api, - &path.workspace_id, - &controller, - &path.worker, - permission, - )?; - let lock = worker_control_lock(&api, &grant.grant_id); - let _guard = lock.lock().await; - authorize_known_worker_permission( - &api, - &path.workspace_id, - &controller, - &path.worker, - permission, - )?; - let mut request = worker_input_request_from_api(request)?; + let worker = WorkspaceWorker::resolve(&api, &path.worker.runtime_id, &path.worker.worker_id)?; + let request = worker_input_request_from_api(request)?; let idle_only = request.kind == WorkerInputKind::User; - if idle_only { - request.kind = WorkerInputKind::UserIfIdle; - } - let Json(result) = - dispatch_runtime_worker_input(api, path.worker.runtime_id, path.worker.worker_id, request)?; - if result.state != server_api::WorkerOperationState::Accepted { + let result = worker.input(&context, request).await?; + let outcome_unknown = result.disposition == WorkerInputDisposition::Unknown; + let result = worker_input_result_to_api(result); + if result.state != server_api::WorkerOperationState::Accepted && !outcome_unknown { let detail = result .diagnostics .iter() @@ -22792,24 +22804,15 @@ async fn cancel_known_worker( ) -> 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 = authorize_known_worker_permission( - &api, - &path.workspace_id, - &controller, - &path.worker, - "cancel", - )?; - let lock = worker_control_lock(&api, &grant.grant_id); - let _guard = lock.lock().await; - authorize_known_worker_permission( - &api, - &path.workspace_id, - &controller, - &path.worker, - "cancel", - )?; - scoped_cancel_runtime_worker(State(api), AxumPath(path), Json(request)).await + let context = WorkerOperationContext::WorkerControl { + controller: RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id), + }; + let worker = WorkspaceWorker::resolve(&api, &path.worker.runtime_id, &path.worker.worker_id)?; + Ok(Json(worker_lifecycle_result_to_api( + worker + .cancel(&context, worker_lifecycle_request_from_api(request)) + .await?, + ))) } async fn stop_known_worker( @@ -22820,14 +22823,15 @@ async fn stop_known_worker( ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; - let subject = path.worker.clone(); - let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id); - let grant = - authorize_known_worker_permission(&api, &path.workspace_id, &controller, &subject, "stop")?; - let lock = worker_control_lock(&api, &grant.grant_id); - let _guard = lock.lock().await; - authorize_known_worker_permission(&api, &path.workspace_id, &controller, &subject, "stop")?; - scoped_stop_runtime_worker(State(api), AxumPath(path), Json(request)).await + let context = WorkerOperationContext::WorkerControl { + controller: RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id), + }; + let worker = WorkspaceWorker::resolve(&api, &path.worker.runtime_id, &path.worker.worker_id)?; + Ok(Json(worker_lifecycle_result_to_api( + worker + .stop(&context, worker_lifecycle_request_from_api(request)) + .await?, + ))) } async fn restore_known_worker( @@ -22837,27 +22841,11 @@ async fn restore_known_worker( ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; - let subject = path.worker.clone(); - let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id); - let grant = authorize_known_worker_permission( - &api, - &path.workspace_id, - &controller, - &subject, - "restore", - )?; - let lock = worker_control_lock(&api, &grant.grant_id); - let _guard = lock.lock().await; - authorize_known_worker_permission(&api, &path.workspace_id, &controller, &subject, "restore")?; - scoped_restore_runtime_worker( - State(api), - AxumPath(path), - Query(RestoreTicketAssignmentQuery { - ticket_id: None, - assignment_operation_id: None, - }), - ) - .await + let context = WorkerOperationContext::WorkerControl { + controller: RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id), + }; + restore_runtime_worker_with_context(api, path.worker.runtime_id, path.worker.worker_id, context) + .await } async fn scoped_list_worker_observation_sessions( @@ -25388,48 +25376,19 @@ fn recover_workdir_removals(api: &WorkspaceApi) -> Result<()> { Ok(()) } +#[cfg(test)] async fn set_worker_retention( api: WorkspaceApi, runtime_id: String, runtime_worker_id: String, pinned: bool, ) -> ApiResult> { - parse_runtime_worker_id_for_registry(&runtime_worker_id)?; - let worker_ref = RuntimeWorkerRef::new(runtime_id.clone(), runtime_worker_id.clone()); - if api - .store - .get_worker_registry(&api.config.workspace_id, &worker_ref)? - .is_none() - { - if let Ok(worker) = api.runtime.worker(&worker_ref) { - let _ = sync_worker_observation(&api, &worker); - } - } - let retention_state = if pinned { "pinned" } else { "normal" }; - let changed = api.worker_projection.publish_ordered(|| { - api.store - .update_worker_retention( - &api.config.workspace_id, - &worker_ref, - retention_state, - now_registry_timestamp().as_str(), - ) - .map(|commit| (!commit.changes.is_empty(), commit)) - })?; - if !changed { - return Err(cleanup_api_error( - runtime_id.as_str(), - "workspace_worker_retention_unknown_worker", - "Worker is not known to the Backend registry", - )); - } - Ok(Json(WorkerRetentionResponse { - workspace_id: api.config.workspace_id, - runtime_id: worker_ref.runtime_id, - worker_id: worker_ref.worker_id, - pinned, - retention_state: retention_state.to_string(), - })) + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &runtime_worker_id)?; + Ok(Json( + worker + .set_pinned(&WorkerOperationContext::Backend, pinned) + .await?, + )) } fn build_runtime_cleanup_plan( @@ -25656,11 +25615,30 @@ fn cleanup_plan_digest( Ok(format!("sha256:{digest}")) } +#[cfg(test)] async fn execute_runtime_cleanup( api: &WorkspaceApi, runtime_id: &str, request: ExecuteRuntimeCleanupRequest, ) -> ApiResult { + execute_runtime_cleanup_with_context(api, runtime_id, request, &WorkerOperationContext::Backend) + .await +} + +async fn execute_runtime_cleanup_with_context( + api: &WorkspaceApi, + runtime_id: &str, + request: ExecuteRuntimeCleanupRequest, + context: &WorkerOperationContext, +) -> ApiResult { + if matches!(context, WorkerOperationContext::WorkerControl { .. }) { + return Err(Error::WorkspacePermissionDenied( + "Worker control uses target-bound WorkerRemove authority, not aggregate cleanup".into(), + ) + .into()); + } + + context.authorize_workspace(api).await?; let plan = build_runtime_cleanup_plan(api, runtime_id)?; if request.expected_plan_revision != plan.revision || request.expected_plan_digest != plan.digest @@ -25713,8 +25691,8 @@ async fn execute_runtime_cleanup( )); } parse_runtime_worker_id_for_registry(&candidate.runtime_worker_id)?; - WorkerRemovalService::new(api) - .execute_cleanup_removal(candidate) + WorkspaceWorker::resolve(api, &worker.runtime_id, &worker.worker_id)? + .delete_from_plan(context, candidate) .await?; api.workdir_session_locks .lock() @@ -26063,15 +26041,26 @@ async fn scoped_get_runtime_worker( .await } -async fn scoped_restore_runtime_worker( - State(api): State, - AxumPath(path): AxumPath, - Query(query): Query, +async fn scoped_restore_runtime_worker_with_context( + api: WorkspaceApi, + path: ScopedRuntimeWorkerPath, + query: RestoreTicketAssignmentQuery, + context: WorkerOperationContext, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; + let resolved = WorkspaceWorker::resolve(&api, &path.worker.runtime_id, &path.worker.worker_id)?; + let _authorization = resolved.authorize_operation(&context, "restore").await?; let workspace_id = path.workspace_id.clone(); - let runtime_id = path.worker.runtime_id.clone(); - let worker_id = path.worker.worker_id.clone(); + let runtime_id = resolved.identity().runtime_id.clone(); + let worker_id = resolved.identity().worker_id.clone(); + if matches!(context, WorkerOperationContext::WorkerControl { .. }) + && (query.ticket_id.is_some() || query.assignment_operation_id.is_some()) + { + return Err(Error::WorkspacePermissionDenied( + "Worker control grants do not authorize Ticket assignment".into(), + ) + .into()); + } let assignment_request = match ( query.ticket_id.clone(), query.assignment_operation_id.clone(), @@ -26134,9 +26123,12 @@ async fn scoped_restore_runtime_worker( })); } } - let response = restore_runtime_worker( - State(api.clone()), - AxumPath((runtime_id.clone(), worker_id.clone())), + drop(_authorization); + let response = restore_runtime_worker_with_context( + api.clone(), + runtime_id.clone(), + worker_id.clone(), + context, ) .await?; if let Some(assignment) = assignment_request.as_ref() { @@ -26146,22 +26138,6 @@ async fn scoped_restore_runtime_worker( Ok(response) } -async fn scoped_pin_runtime_worker( - State(api): State, - AxumPath(path): AxumPath, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - set_worker_retention(api, path.worker.runtime_id, path.worker.worker_id, true).await -} - -async fn scoped_unpin_runtime_worker( - State(api): State, - AxumPath(path): AxumPath, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - set_worker_retention(api, path.worker.runtime_id, path.worker.worker_id, false).await -} - async fn scoped_runtime_cleanup_plan( State(api): State, AxumPath(path): AxumPath, @@ -26171,6 +26147,7 @@ async fn scoped_runtime_cleanup_plan( Ok(Json(plan)) } +#[cfg(test)] async fn scoped_execute_runtime_cleanup( State(api): State, AxumPath(path): AxumPath, @@ -26403,6 +26380,7 @@ async fn scoped_delete_runtime_worker_uploaded_file( .await } +#[cfg(test)] async fn scoped_send_runtime_worker_input( State(api): State, AxumPath(path): AxumPath, @@ -26431,36 +26409,9 @@ async fn scoped_runtime_worker_completions( .await } -async fn scoped_stop_runtime_worker( - State(api): State, - AxumPath(path): AxumPath, - Json(request): Json, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - stop_runtime_worker( - State(api), - AxumPath((path.worker.runtime_id, path.worker.worker_id)), - Json(request), - ) - .await -} - -async fn scoped_cancel_runtime_worker( - State(api): State, - AxumPath(path): AxumPath, - Json(request): Json, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - cancel_runtime_worker( - State(api), - AxumPath((path.worker.runtime_id, path.worker.worker_id)), - Json(request), - ) - .await -} - async fn scoped_worker_protocol_ws( ws: WebSocketUpgrade, + headers: HeaderMap, State(api): State, Extension(actor): Extension, AxumPath(path): AxumPath, @@ -26472,6 +26423,7 @@ async fn scoped_worker_protocol_ws( State(api), Extension(actor), AxumPath((path.worker.runtime_id, path.worker.worker_id)), + headers, ws, ) .await @@ -28936,27 +28888,9 @@ fn resolve_workspace_worker_reference( runtime_id: &str, reference: &str, ) -> ApiResult { - let direct = RuntimeWorkerRef::new(runtime_id, reference); - if let Some(record) = api - .store - .get_worker_registry(&api.config.workspace_id, &direct)? - { - return Ok(record.worker); - } - let worker = api - .store - .resolve_worker_resource_reference(&api.config.workspace_id, reference)? - .filter(|worker| worker.runtime_id == runtime_id) - .ok_or_else(|| Error::UnknownWorker { - worker: RuntimeWorkerRef::new(runtime_id, reference), - })?; - let record = api - .store - .get_worker_registry(&api.config.workspace_id, &worker)? - .ok_or_else(|| Error::UnknownWorker { - worker: RuntimeWorkerRef::new(runtime_id, reference), - })?; - Ok(record.worker) + Ok(WorkspaceWorker::resolve(api, runtime_id, reference)? + .identity() + .clone()) } async fn get_runtime_worker( @@ -28984,12 +28918,14 @@ async fn get_runtime_worker( })) } -async fn restore_runtime_worker( - State(api): State, - AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, +async fn restore_runtime_worker_with_context( + api: WorkspaceApi, + runtime_id: String, + worker_id: String, + context: WorkerOperationContext, ) -> ApiResult> { - let worker = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; - let result = api.restore_workspace_worker(&worker)?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id)?; + let result = worker.restore(&context).await?; let projected_worker = if let Some(worker) = result.worker.as_ref() { let record = sync_worker_observation(&api, worker)?; let links = api @@ -29665,37 +29601,27 @@ async fn check_runtime_config_bundle( )?)) } +#[cfg(test)] async fn send_runtime_worker_input( State(api): State, AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, Json(request): Json, ) -> ApiResult> { let request = worker_input_request_from_api(request)?; - dispatch_runtime_worker_input(api, runtime_id, worker_id, request) + dispatch_runtime_worker_input(api, runtime_id, worker_id, request).await } -fn dispatch_runtime_worker_input( +#[cfg(test)] +async fn dispatch_runtime_worker_input( api: WorkspaceApi, runtime_id: String, worker_id: String, request: WorkerInputRequest, ) -> ApiResult> { - let worker = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; - if api - .store - .worker_registry_projection(&api.config.workspace_id, &worker)? - .and_then(|projection| projection.job) - .is_some() - { - return Err(Error::InvalidInput( - "Backend Job Workers accept only their Backend-owned immutable input".to_string(), - ) - .into()); - } - let result = api - .runtime - .send_input(&worker, request) - .map_err(|err| err.into_error())?; + let worker = WorkspaceWorker::resolve(&api, &runtime_id, &worker_id)?; + let result = worker + .input(&WorkerOperationContext::Backend, request) + .await?; Ok(Json(worker_input_result_to_api(result))) } @@ -29724,44 +29650,6 @@ async fn runtime_worker_completions( Ok(Json(worker_completions_result_to_api(result)?)) } -async fn stop_runtime_worker( - State(api): State, - AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, - Json(request): Json, -) -> ApiResult> { - let request = worker_lifecycle_request_from_api(request); - let worker = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; - let result = api - .runtime - .stop_worker(&worker, request) - .map_err(|err| err.into_error())?; - parse_runtime_worker_id_for_registry(&worker.worker_id)?; - let session_lock = current_worker_session_lock(&api, &worker); - let _session_guard = session_lock.lock().await; - close_current_worker_session_locked(&api, &worker).await?; - if let Some(record) = api - .store - .get_worker_registry(&api.config.workspace_id, &worker)? - { - sync_linked_workdir_after_worker_stop(&api, &worker.runtime_id, &record)?; - } - Ok(Json(worker_lifecycle_result_to_api(result))) -} - -async fn cancel_runtime_worker( - State(api): State, - AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, - Json(request): Json, -) -> ApiResult> { - let request = worker_lifecycle_request_from_api(request); - let worker = resolve_workspace_worker_reference(&api, &runtime_id, &worker_id)?; - let result = api - .runtime - .cancel_worker(&worker, request) - .map_err(|err| err.into_error())?; - Ok(Json(worker_lifecycle_result_to_api(result))) -} - fn authenticated_browser_input_source(actor: &RequestActor) -> protocol::AuthenticatedInputSource { protocol::AuthenticatedInputSource::Account { account_id: actor.account_id.clone(), @@ -29791,10 +29679,25 @@ pub(crate) fn authorize_browser_worker_method( }), protocol::Method::SubmitTracked { .. } | protocol::Method::SubmitIfIdle { .. } - | protocol::Method::NotifyTracked { .. } => { + | protocol::Method::NotifyTracked { .. } + | protocol::Method::WorkerEvent(_) => { Err("authenticated Worker input source is server-owned") } - other => Ok(other), + other @ (protocol::Method::ListPendingSubmissions + | protocol::Method::CancelPendingSubmission { .. } + | protocol::Method::ClearPendingSubmissions { .. } + | protocol::Method::ContinuePending { .. } + | protocol::Method::Resume { .. } + | protocol::Method::Cancel { .. } + | protocol::Method::Pause { .. } + | protocol::Method::Compact { .. } + | protocol::Method::ListRewindTargets + | protocol::Method::RewindTo { .. } + | protocol::Method::Shutdown { .. } + | protocol::Method::ListCompletions { .. } + | protocol::Method::ListWorkers + | protocol::Method::RestoreWorker { .. } + | protocol::Method::RegisterPeer { .. }) => Ok(other), } } @@ -29802,40 +29705,59 @@ async fn worker_protocol_ws( State(api): State, Extension(actor): Extension, AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, + headers: HeaderMap, ws: WebSocketUpgrade, -) -> impl IntoResponse { - let worker = RuntimeWorkerRef::new(&runtime_id, &worker_id); - let job_read_only = match api - .store - .worker_registry_projection(&api.config.workspace_id, &worker) - { - Ok(projection) => projection.and_then(|projection| projection.job).is_some(), +) -> Response { + let worker = match WorkspaceWorker::resolve(&api, &runtime_id, &worker_id) { + Ok(worker) => worker, Err(error) => return ApiError::from(error).into_response(), }; - let source = match api.observation_proxy.source(&worker) { - Ok(source) => source, - Err(ObservationProxyError::WorkerNotFound(_)) => { - match api.runtime.observation_source(&worker) { - Ok(source) => source, - Err(error) => return ApiError::from(error.into_error()).into_response(), + let context = WorkerOperationContext::Browser { + headers, + source: authenticated_browser_input_source(&actor), + }; + ws.on_upgrade(move |mut socket| async move { + match worker.connect_protocol(&context).await { + Ok(connection) => workspace_worker_protocol_session(connection, socket, context).await, + Err(error) => { + let _ = send_protocol_event(&mut socket, &protocol_error_event(error.to_string())) + .await; + let _ = socket.send(WsMessage::Close(None)).await; } } - Err(error) => { - return ( - StatusCode::BAD_REQUEST, - Json(serde_json::json!({ - "error": error.code(), - "message": error.message(), - })), - ) - .into_response(); - } - }; - let input_source = authenticated_browser_input_source(&actor); - let scope = worker_runtime::RuntimeWorkspaceScope::new(api.workspace_id(), "embedded-backend"); - ws.on_upgrade(move |socket| { - worker_protocol_ws_session(source, socket, input_source, job_read_only, scope) }) + .into_response() +} + +async fn workspace_worker_protocol_session( + mut connection: worker_operations::WorkspaceWorkerConnection, + mut socket: WebSocket, + context: WorkerOperationContext, +) { + loop { + tokio::select! { + inbound = socket.next() => match inbound { + Some(Ok(WsMessage::Text(text))) => { + let result = match decode_method(&text) { + Ok(method) => connection.sender.send(&context, method).await.map_err(|error| error.to_string()), + Err(error) => Err(error.to_string()), + }; + if let Err(error) = result { + if !send_protocol_event(&mut socket, &protocol_error_event(error)).await { break; } + } + } + Some(Ok(WsMessage::Ping(value))) => { if socket.send(WsMessage::Pong(value)).await.is_err() { break; } } + Some(Ok(WsMessage::Pong(_))) => {} + _ => break, + }, + event = connection.events.recv() => { + let Some(event) = event else { break; }; + let shutdown = matches!(event, protocol::Event::Shutdown); + if !send_protocol_event(&mut socket, &event).await || shutdown { break; } + } + } + } + let _ = socket.send(WsMessage::Close(None)).await; } pub(crate) struct WorkspaceWorkerProtocolConnection { @@ -30024,6 +29946,7 @@ fn embedded_protocol_dispatch_is_terminal( ) || transport.validate().is_err() } +#[cfg(test)] async fn worker_protocol_ws_session( source: RuntimeObservationSource, socket: WebSocket, @@ -30042,6 +29965,7 @@ async fn worker_protocol_ws_session( } } +#[cfg(test)] async fn remote_worker_protocol_ws_session( config: RuntimeObservationSourceConfig, socket: WebSocket, @@ -30192,6 +30116,7 @@ async fn remote_worker_protocol_ws_session( } } +#[cfg(test)] async fn embedded_worker_protocol_ws_session( source: crate::observation::EmbeddedRuntimeObservationSource, mut socket: WebSocket, @@ -33172,6 +33097,7 @@ impl IntoResponse for ApiError { #[cfg(test)] mod tests { + mod worker_operations_tests; include!("server_workspace_config_tests.rs"); use super::*; use axum::body::{Body, to_bytes}; @@ -38521,6 +38447,11 @@ mod tests { Request::builder() .method("POST") .uri(format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers")) + // Unit fixture stands below proof middleware, so inject its verified subject. + .extension(crate::worker_source::VerifiedRuntimeRequestSource { + runtime_id: fixture.controller.runtime_id.clone(), + worker_id: Some(fixture.controller.worker_id.clone()), + }) .header("content-type", "application/json") .header("x-yoi-runtime-id", &fixture.controller.runtime_id) .header("x-yoi-worker-id", &fixture.controller.worker_id) @@ -38836,6 +38767,10 @@ mod tests { Request::builder() .method("POST") .uri(format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers")) + .extension(crate::worker_source::VerifiedRuntimeRequestSource { + runtime_id: controller.runtime_id.clone(), + worker_id: Some(controller.worker_id.clone()), + }) .header("content-type", "application/json") .header("x-yoi-runtime-id", &controller.runtime_id) .header("x-yoi-worker-id", &controller.worker_id) @@ -51511,7 +51446,27 @@ mod tests { #[tokio::test] async fn embedded_worker_remove_executes_retention_and_returns_bounded_result() { let temp = tempfile::tempdir().unwrap(); - let api = test_api(temp.path()).await; + let config = test_server_config(temp.path()); + let store = Arc::new(test_control_store(&config)); + seed_test_registered_workspace(store.as_ref(), &config) + .await + .unwrap(); + let dispatcher = Arc::new( + crate::worker_source::EmbeddedServerWorkerMutationDispatcher::new( + config.clone(), + store.clone(), + ), + ); + // Use the same constructor-installed executor as the production embedded factory. + let api = WorkspaceApi::new_with_execution_backend_and_broker( + config, + store, + Arc::new(DeterministicExecutionBackend::default()), + BackendResourceBroker::default(), + Some(dispatcher.clone()), + ) + .await + .unwrap(); let Json(orchestrator) = scoped_start_workspace_orchestrator( State(api.clone()), AxumPath(ScopedWorkspacePath { @@ -51552,6 +51507,27 @@ mod tests { ) .unwrap(); let target = spawned.worker.unwrap().worker; + seed_worker_control_grant(&api, &source, &target, "embedded-valid-proof"); + let target_key = api + .store + .worker_resource_key(TEST_WORKSPACE_ID, &target) + .unwrap() + .unwrap(); + let forwarder = worker_runtime::worker_source::RuntimeWorkerMutationForwarder::embedded( + &source.runtime_id, + worker_runtime::RuntimeWorkspaceScope::new(TEST_WORKSPACE_ID, "embedded-backend"), + &source.worker_id, + dispatcher, + ); + let running = forwarder + .execute_worker_remove( + &target.runtime_id, + &target_key, + "must retain running Worker", + ) + .unwrap(); + assert_eq!(running.status, StatusCode::CONFLICT.as_u16()); + assert!(running.body.contains("worker_not_stopped")); let stopped = api .runtime .stop_worker( @@ -51584,25 +51560,30 @@ mod tests { .unwrap(); let summary = api.runtime.worker(&target).unwrap(); sync_worker_observation(&api, &summary).unwrap(); - seed_worker_control_grant(&api, &source, &target, "embedded-valid-proof"); + let mut subscriber = api.worker_projection.subscribe_ordered(10).unwrap(); let _ = subscriber.take_snapshot(); - let response = WorkerRemovalService::new(&api) - .execute_async( - crate::worker_source::VerifiedWorkerMutationSource { - runtime_id: source.runtime_id, - worker_id: source.worker_id, - actor_kind: worker_runtime::auth::WorkerMutationActorKind::Worker, - permission: worker_runtime::auth::WORKER_REMOVE_PERMISSION.to_string(), - jti: "embedded-valid-proof".to_string(), - }, - &target.runtime_id, - &target.worker_id, - "retire completed Worker", - ) + let handle = WorkspaceWorker::resolve(&api, &target.runtime_id, &target_key).unwrap(); + handle + .set_pinned(&WorkerOperationContext::Backend, true) .await .unwrap(); + let pinned = forwarder + .execute_worker_remove(&target.runtime_id, &target_key, "must retain pinned Worker") + .unwrap(); + assert_eq!(pinned.status, StatusCode::CONFLICT.as_u16()); + assert!(pinned.body.contains("worker_removal_blocked")); + handle + .set_pinned(&WorkerOperationContext::Backend, false) + .await + .unwrap(); + + // A resource key must be normalized by common reception after its exact + // target-bound proof is verified; the old raw service executor cannot do this. + let response = forwarder + .execute_worker_remove(&target.runtime_id, &target_key, "retire completed Worker") + .unwrap(); assert_eq!( response.status, StatusCode::OK.as_u16(), @@ -57854,7 +57835,7 @@ mod tests { ) .await .unwrap(); - let app = build_inner_router(api.clone()); + let app = authenticated_protocol_fixture_router(api.clone()); let runtimes = get_json(app.clone(), "/api/runtimes").await; let embedded_summary = runtimes["items"] @@ -58813,6 +58794,25 @@ mod tests { } } + // Transport fixtures below the authentication router still need real credentials + // for operation-time reauthorization. Public-router tests use real proofs/tokens. + fn authenticated_protocol_fixture_router(api: WorkspaceApi) -> Router { + let token = seed_test_api_token(api.store.as_ref(), TEST_WORKSPACE_ID); + build_inner_router(api) + .layer(Extension(test_browser_request_actor())) + .layer(axum::middleware::from_fn( + move |mut request: axum::extract::Request, next: Next| { + let value = format!("Bearer {token}"); + async move { + request + .headers_mut() + .insert(axum::http::header::AUTHORIZATION, value.parse().unwrap()); + next.run(request).await + } + }, + )) + } + async fn spawn_workspace_proxy( source: RuntimeObservationSourceConfig, ) -> (String, tempfile::TempDir) { @@ -58831,7 +58831,8 @@ mod tests { .unwrap(); let app_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let app_addr = app_listener.local_addr().unwrap(); - let app = build_inner_router(api).layer(Extension(test_browser_request_actor())); + seed_worker_source_member(&api, &runtime_id, &worker_id); + let app = authenticated_protocol_fixture_router(api); tokio::spawn(async move { axum::serve(app_listener, app).await.unwrap() }); ( format!("ws://{app_addr}/api/runtimes/{runtime_id}/workers/{worker_id}/protocol/ws"), @@ -58844,8 +58845,7 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); - let app = build_inner_router(test_api(dir.path()).await) - .layer(Extension(test_browser_request_actor())); + let app = authenticated_protocol_fixture_router(test_api(dir.path()).await); let server = tokio::spawn(async move { let _ = axum::serve(listener, app).await; }); @@ -58942,7 +58942,7 @@ mod tests { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let projection_api = api.clone(); - let app = build_inner_router(api).layer(Extension(test_browser_request_actor())); + let app = authenticated_protocol_fixture_router(api); let server = tokio::spawn(async move { let _ = axum::serve(listener, app).await; }); diff --git a/crates/workspace-server/src/server/tests/worker_operations_tests.rs b/crates/workspace-server/src/server/tests/worker_operations_tests.rs new file mode 100644 index 000000000..d6083877f --- /dev/null +++ b/crates/workspace-server/src/server/tests/worker_operations_tests.rs @@ -0,0 +1,864 @@ +use super::*; + +fn operation_input(kind: WorkerInputKind) -> WorkerInputRequest { + WorkerInputRequest { + kind, + content: "operation-boundary-test".into(), + submission_request_id: None, + segments: None, + } +} + +fn operation_lifecycle() -> WorkerLifecycleRequest { + WorkerLifecycleRequest { + reason: Some("operation-boundary-test".into()), + ticket_assignment: None, + } +} + +fn operation_cleanup_candidate(identity: &RuntimeWorkerRef) -> CleanupWorkerCandidate { + CleanupWorkerCandidate { + target_id: format!("worker:{}:{}", identity.runtime_id, identity.worker_id), + action: server_api::CleanupTargetKind::WorkerDelete, + worker_id: identity.worker_id.clone(), + runtime_worker_id: identity.worker_id.clone(), + runtime_id: identity.runtime_id.clone(), + reason: "operation-boundary-test".into(), + blocking_reason: None, + pinned: false, + retention_state: "normal".into(), + linked_workdir_ids: Vec::new(), + running_linked: false, + estimated_reclaim_bytes: None, + } +} + +async fn browser_operation_context(api: &WorkspaceApi) -> (WorkerOperationContext, String) { + seed_test_api_token(api.store.as_ref(), "worker-operation-browser"); + let token = "worker-operation-browser-session"; + api.store + .create_browser_session(&BrowserSessionRecord { + token_hash: crate::auth::token_hash(token), + session_id: "worker-operation-session".into(), + user_id: "user-worker-operation-browser".into(), + created_at: "2026-01-01T00:00:00Z".into(), + expires_at: "2099-01-01T00:00:00Z".into(), + revoked_at: None, + }) + .unwrap(); + let AuthConfig::Passkey { + cookie_name, + origin, + .. + } = &api.config.auth; + let mut headers = HeaderMap::new(); + headers.insert(ORIGIN, origin.parse().unwrap()); + headers.insert( + axum::http::header::COOKIE, + format!("{cookie_name}={token}").parse().unwrap(), + ); + let actor = resolve_actor(&ServerAuthApi::from(api), &headers) + .await + .unwrap() + .expect("valid browser session"); + ( + WorkerOperationContext::Browser { + headers, + source: authenticated_browser_input_source(&actor), + }, + crate::auth::token_hash(token), + ) +} + +fn browser_operation_methods() -> Vec { + vec![ + protocol::Method::Pause { + command: protocol::WorkerCommandEnvelope::new(701), + }, + protocol::Method::Resume { + command: protocol::WorkerCommandEnvelope::new(702), + }, + protocol::Method::Cancel { + command: protocol::WorkerCommandEnvelope::new(703), + }, + protocol::Method::Compact { + command: protocol::WorkerCommandEnvelope::new(704), + }, + protocol::Method::Submit { + submission_request_id: "worker-operation-submit".into(), + input: vec![protocol::Segment::Text { + content: "must not reach execution".into(), + }], + }, + protocol::Method::Notify { + notification_request_id: "worker-operation-notify".into(), + message: "must not reach execution".into(), + }, + protocol_test_completion("worker-operation-completion"), + ] +} + +#[tokio::test] +async fn handle_resource_keys_preserve_identity_for_equal_local_ids_across_runtimes() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + let a = RuntimeWorkerRef::new("offline-runtime-a", "shared-local-id"); + let b = RuntimeWorkerRef::new("offline-runtime-b", "shared-local-id"); + for identity in [&a, &b] { + seed_worker_source_member(&api, &identity.runtime_id, &identity.worker_id); + } + let key = |identity: &RuntimeWorkerRef| { + api.store + .worker_resource_key(TEST_WORKSPACE_ID, identity) + .unwrap() + .unwrap() + }; + assert_ne!(key(&a), key(&b)); + + for identity in [&a, &b] { + let direct = + WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap(); + let canonical = + WorkspaceWorker::resolve(&api, &identity.runtime_id, &key(identity)).unwrap(); + let pinned = canonical + .set_pinned(&WorkerOperationContext::Backend, true) + .await + .unwrap(); + assert_eq!(pinned.runtime_id, identity.runtime_id); + assert_eq!(pinned.worker_id, identity.worker_id); + let unpinned = direct + .set_pinned(&WorkerOperationContext::Backend, false) + .await + .unwrap(); + assert_eq!(unpinned.runtime_id, pinned.runtime_id); + assert_eq!(unpinned.worker_id, pinned.worker_id); + } + assert!(matches!( + WorkspaceWorker::resolve(&api, &b.runtime_id, &key(&a)), + Err(Error::UnknownWorker { .. }) + )); +} + +#[tokio::test] +async fn handle_resolution_rejects_unknown_foreign_and_internal_only_workers() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + let known = spawn_ticket_check_source(&api, "operation-resolution-parent"); + let mut foreign = api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &known) + .unwrap() + .unwrap(); + let mut other_workspace = api + .store + .get_workspace(TEST_WORKSPACE_ID) + .await + .unwrap() + .unwrap(); + other_workspace.workspace_id = "other-workspace".into(); + api.store.upsert_workspace(&other_workspace).await.unwrap(); + foreign.workspace_id = other_workspace.workspace_id; + foreign.worker.worker_id = "foreign-only-worker".into(); + api.store.upsert_worker_registry(&foreign).unwrap(); + assert!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &foreign.worker) + .unwrap() + .is_none() + ); + + let RuntimeObservationSource::Embedded(source) = + api.runtime.observation_source(&known).unwrap() + else { + panic!("expected embedded fixture"); + }; + let internal = protocol::InternalWorkerRef { + session_id: "internal-only-session".into(), + parent_session_id: None, + name: "internal-only-child".into(), + kind: protocol::InternalWorkerKind::SubWorker, + }; + source + .runtime + .observe_worker_event( + &source.worker_ref, + protocol::Event::InternalWorker { + worker: internal.clone(), + revision: 1, + event: Box::new(protocol::Event::WorkerState { + snapshot: protocol::WorkerStateSnapshot::initial(), + }), + }, + ) + .unwrap(); + + for reference in [ + "unknown-worker", + &foreign.worker.worker_id, + &internal.session_id, + &internal.name, + ] { + assert!( + matches!( + WorkspaceWorker::resolve(&api, &known.runtime_id, reference), + Err(Error::UnknownWorker { .. }) + ), + "unexpected registry handle for {reference}" + ); + } +} + +#[tokio::test] +async fn offline_handle_can_pin_but_deleted_handle_cannot_mutate_or_connect() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + let identity = RuntimeWorkerRef::new("unregistered-offline-runtime", "42"); + seed_worker_source_member(&api, &identity.runtime_id, &identity.worker_id); + assert!( + api.runtime.worker(&identity).is_err(), + "fixture must have no live Runtime" + ); + let handle = WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap(); + let retained = handle + .set_pinned(&WorkerOperationContext::Backend, true) + .await + .unwrap(); + assert!(retained.pinned); + assert_eq!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &identity) + .unwrap() + .unwrap() + .retention_state, + "pinned" + ); + + let mut wrong_candidate = operation_cleanup_candidate(&identity); + wrong_candidate.runtime_worker_id = "different-worker".into(); + assert!(matches!( + handle.delete_from_plan(&WorkerOperationContext::Backend, &wrong_candidate) + .await.unwrap_err().error, + Error::InvalidInput(message) if message.contains("resolved Worker") + )); + + api.store + .delete_worker_registry(TEST_WORKSPACE_ID, &identity) + .unwrap(); + assert!( + matches!(handle.set_pinned(&WorkerOperationContext::Backend, false) + .await.unwrap_err().error, Error::UnknownWorker { worker } if worker == identity) + ); + assert!(matches!( + handle + .restore(&WorkerOperationContext::Backend) + .await + .unwrap_err() + .error, + Error::UnknownWorker { .. } + )); + assert!(matches!( + handle + .stop(&WorkerOperationContext::Backend, operation_lifecycle()) + .await + .unwrap_err() + .error, + Error::UnknownWorker { .. } + )); + assert!(matches!( + handle + .cancel(&WorkerOperationContext::Backend, operation_lifecycle()) + .await + .unwrap_err() + .error, + Error::UnknownWorker { .. } + )); + for kind in [WorkerInputKind::User, WorkerInputKind::Notify] { + assert!(matches!( + handle + .input(&WorkerOperationContext::Backend, operation_input(kind)) + .await + .unwrap_err() + .error, + Error::UnknownWorker { .. } + )); + } + assert!(matches!( + handle + .delete_from_plan( + &WorkerOperationContext::Backend, + &operation_cleanup_candidate(&identity) + ) + .await + .unwrap_err() + .error, + Error::UnknownWorker { .. } + )); + assert!(matches!( + handle + .connect_protocol(&WorkerOperationContext::Backend) + .await, + Err(Error::UnknownWorker { .. }) + )); + assert!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &identity) + .unwrap() + .is_none(), + "a stale handle must not recreate its registry row" + ); +} + +#[tokio::test] +async fn worker_control_handle_rechecks_revoked_grants_for_each_operation() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + let controller = RuntimeWorkerRef::new("offline-runtime", "controller"); + let subject = RuntimeWorkerRef::new("offline-runtime", "subject"); + for identity in [&controller, &subject] { + seed_worker_source_member(&api, &identity.runtime_id, &identity.worker_id); + } + api.store + .create_worker_control_grant(&WorkerControlGrantRecord { + workspace_id: TEST_WORKSPACE_ID.into(), + grant_id: "operation-grant".into(), + controller: controller.clone(), + subject: subject.clone(), + relation: "spawned".into(), + origin: "test".into(), + permissions: [ + "restore", + "stop", + "cancel", + "send_input", + "notify", + "pin", + "remove", + ] + .into_iter() + .map(str::to_owned) + .collect(), + operation_id: "operation-grant-test".into(), + created_at: now_registry_timestamp(), + revoked_at: None, + }) + .unwrap(); + let handle = WorkspaceWorker::resolve(&api, &subject.runtime_id, &subject.worker_id).unwrap(); + let context = WorkerOperationContext::WorkerControl { controller }; + handle.set_pinned(&context, true).await.unwrap(); + api.store + .revoke_worker_control_grant( + TEST_WORKSPACE_ID, + "operation-grant", + &now_registry_timestamp(), + ) + .unwrap(); + + for permission in [ + "restore", + "stop", + "cancel", + "send_input", + "notify", + "pin", + "remove", + ] { + assert!( + matches!( + handle.authorize_operation(&context, permission).await, + Err(ApiError { + error: Error::UnknownWorker { .. }, + .. + }) + ), + "revoked grant authorized {permission}" + ); + } + assert!(matches!( + handle.restore(&context).await.unwrap_err().error, + Error::UnknownWorker { .. } + )); + assert!(matches!( + handle + .stop(&context, operation_lifecycle()) + .await + .unwrap_err() + .error, + Error::UnknownWorker { .. } + )); + assert!(matches!( + handle + .cancel(&context, operation_lifecycle()) + .await + .unwrap_err() + .error, + Error::UnknownWorker { .. } + )); + for kind in [WorkerInputKind::User, WorkerInputKind::Notify] { + assert!(matches!( + handle + .input(&context, operation_input(kind)) + .await + .unwrap_err() + .error, + Error::UnknownWorker { .. } + )); + } + assert!(matches!( + handle.set_pinned(&context, false).await.unwrap_err().error, + Error::UnknownWorker { .. } + )); + assert!(matches!( + handle + .delete_from_plan(&context, &operation_cleanup_candidate(&subject)) + .await + .unwrap_err() + .error, + Error::UnknownWorker { .. } + )); + assert_eq!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &subject) + .unwrap() + .unwrap() + .retention_state, + "pinned" + ); +} + +#[tokio::test] +async fn generic_and_scoped_input_aliases_match_handle_idle_only_and_notify_semantics() { + let fixture = guarded_spawn_fixture().await; + let response = post_guarded_spawn( + &fixture, + guarded_spawn_payload(&fixture, "operation-alias-parity"), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let created: BrowserCreateWorkerResponse = serde_json::from_slice(&bytes).unwrap(); + let api = fixture.api.clone(); + let runtime = &fixture.runtime; + let subject = RuntimeWorkerRef::new(created.runtime_id, created.worker_id); + let controller = fixture.controller.clone(); + let key = api + .store + .worker_resource_key(TEST_WORKSPACE_ID, &subject) + .unwrap() + .unwrap(); + let handle = WorkspaceWorker::resolve(&api, &subject.runtime_id, &key).unwrap(); + let context = WorkerOperationContext::WorkerControl { controller }; + let idle_only = handle + .input(&context, operation_input(WorkerInputKind::User)) + .await + .unwrap(); + assert_eq!(idle_only.state, InternalWorkerOperationState::Rejected); + let notify = handle + .input(&context, operation_input(WorkerInputKind::Notify)) + .await + .unwrap(); + assert_eq!(notify.state, InternalWorkerOperationState::Accepted); + + let request = || server_api::RuntimeWorkerInputRequest { + kind: Some("user".into()), + content: "human input".into(), + segments: None, + }; + let Json(generic) = send_runtime_worker_input( + State(api.clone()), + AxumPath((subject.runtime_id.clone(), key.clone())), + Json(request()), + ) + .await + .unwrap(); + let Json(scoped) = scoped_send_runtime_worker_input( + State(api.clone()), + AxumPath(ScopedRuntimeWorkerPath { + workspace_id: TEST_WORKSPACE_ID.into(), + worker: RuntimeWorkerRef::new(&subject.runtime_id, &key), + }), + Json(request()), + ) + .await + .unwrap(); + assert_eq!(generic.state, scoped.state); + assert_eq!(generic.state, server_api::WorkerOperationState::Accepted); + assert_eq!( + runtime + .input_requests + .lock() + .unwrap() + .iter() + .map(|request| request.kind.clone()) + .collect::>(), + vec![ + WorkerInputKind::UserIfIdle, + WorkerInputKind::Notify, + WorkerInputKind::User, + WorkerInputKind::User + ] + ); +} + +#[tokio::test] +async fn browser_protocol_sender_rejects_job_input_and_all_live_methods() { + let workspace = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(workspace.path()).await; + let request = BackendJobRequest { + job_id: "worker-operation-read-only-job".into(), + purpose: "operation_boundary_test".into(), + input_revision: "revision-1".into(), + input_ref: "test://worker-operation/input".into(), + input: serde_json::json!({"immutable": true}), + instruction: "Return a structured result.".into(), + profile: "builtin:backend-job".into(), + source_worker: None, + notification_target: None, + limits: crate::backend_job::BackendJobLimits::default(), + }; + let dispatched = api.dispatch_backend_job(&request).unwrap(); + let identity = dispatched.attempt.worker.unwrap(); + assert_eq!(execution.take_inputs().len(), 1); + let handle = WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap(); + let (context, _) = browser_operation_context(&api).await; + let mut connection = handle.connect_protocol(&context).await.unwrap(); + assert!(matches!( + tokio::time::timeout(std::time::Duration::from_secs(2), connection.events.recv()) + .await + .unwrap() + .unwrap(), + protocol::Event::Snapshot { .. } + )); + for method in browser_operation_methods() { + assert!(matches!(connection.sender.send(&context, method).await, + Err(Error::InvalidInput(message)) if message.contains("read-only"))); + } + for kind in [WorkerInputKind::User, WorkerInputKind::Notify] { + assert!( + matches!(handle.input(&context, operation_input(kind)).await.unwrap_err().error, + Error::InvalidInput(message) if message.contains("read-only")) + ); + } + assert!(execution.protocol_methods().is_empty()); + assert!(execution.take_inputs().is_empty()); +} + +#[tokio::test] +async fn connected_browser_sender_rechecks_revoked_credentials_before_each_method() { + let workspace = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(workspace.path()).await; + let identity = spawn_ticket_check_source(&api, "operation-browser-revocation"); + let handle = WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap(); + let (context, token_hash) = browser_operation_context(&api).await; + let mut connection = handle.connect_protocol(&context).await.unwrap(); + assert!(matches!( + tokio::time::timeout(std::time::Duration::from_secs(2), connection.events.recv()) + .await + .unwrap() + .unwrap(), + protocol::Event::Snapshot { .. } + )); + // Prove these credentials were usable before revocation, then keep the same sender. + handle.set_pinned(&context, true).await.unwrap(); + assert!( + api.store + .revoke_browser_session(&token_hash, &now_registry_timestamp()) + .unwrap() + ); + assert!(matches!( + handle.connect_protocol(&context).await, + Err(Error::WorkspacePermissionDenied(_)) + )); + for method in browser_operation_methods() { + assert!(matches!( + connection.sender.send(&context, method).await, + Err(Error::WorkspacePermissionDenied(_)) + )); + } + assert!(matches!( + handle.set_pinned(&context, false).await.unwrap_err().error, + Error::WorkspacePermissionDenied(_) + )); + assert!( + execution.protocol_methods().is_empty(), + "revoked credentials must not reach the execution channel" + ); + assert_eq!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &identity) + .unwrap() + .unwrap() + .retention_state, + "pinned" + ); +} + +fn operation_request_context( + source: Option, +) -> server_api::ServerRequestContext { + server_api::ServerRequestContext { + actor: None, + worker_source: None, + runtime_source: source, + origin: None, + transport_headers: Vec::new(), + } +} + +#[tokio::test] +async fn request_context_uses_verified_subject_not_headers_or_account_fallback() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + let mut headers_only = operation_request_context(None); + headers_only.transport_headers = vec![ + ("x-yoi-runtime-id".into(), b"runtime-a".to_vec()), + ("x-yoi-worker-id".into(), b"controller-a".to_vec()), + ]; + assert!(WorkerOperationContext::from_request(&api, &headers_only).is_err()); + let mut runtime_only = operation_request_context(Some(server_api::ServerRuntimeSource { + runtime_id: "runtime-a".into(), + worker_id: None, + })); + runtime_only.actor = Some(test_browser_request_actor()); + assert!(WorkerOperationContext::from_request(&api, &runtime_only).is_err()); + let mut mismatched = operation_request_context(Some(server_api::ServerRuntimeSource { + runtime_id: "runtime-a".into(), + worker_id: Some("controller-a".into()), + })); + mismatched.actor = Some(test_browser_request_actor()); + mismatched.transport_headers = vec![("x-yoi-worker-id".into(), b"controller-b".to_vec())]; + assert!(WorkerOperationContext::from_request(&api, &mismatched).is_err()); + seed_worker_source_member(&api, "runtime-a", "controller-a"); + let context = operation_request_context(Some(server_api::ServerRuntimeSource { + runtime_id: "runtime-a".into(), + worker_id: Some("controller-a".into()), + })); + assert!( + matches!(WorkerOperationContext::from_request(&api, &context).unwrap(), + WorkerOperationContext::WorkerControl { controller } if controller == RuntimeWorkerRef::new("runtime-a", "controller-a")) + ); +} + +#[tokio::test] +async fn generic_management_aliases_cannot_bypass_worker_grants_or_target_bound_remove() { + use server_api::ServerApi; + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + seed_worker_source_member(&api, "offline-runtime", "controller"); + seed_worker_source_member(&api, "offline-runtime", "subject"); + let service = ServerApiContractService::Workspace(api.clone()); + let context = || { + operation_request_context(Some(server_api::ServerRuntimeSource { + runtime_id: "offline-runtime".into(), + worker_id: Some("controller".into()), + })) + }; + let input = || server_api::RuntimeWorkerInputRequest { + kind: Some("user".into()), + content: "must not execute".into(), + segments: None, + }; + assert!( + service + .runtime_worker_input_alias( + context(), + "offline-runtime".into(), + "subject".into(), + input() + ) + .await + .is_err() + ); + assert!( + service + .runtime_worker_input( + context(), + TEST_WORKSPACE_ID.into(), + "offline-runtime".into(), + "subject".into(), + input() + ) + .await + .is_err() + ); + assert!( + service + .runtime_worker_restore_alias(context(), "offline-runtime".into(), "subject".into()) + .await + .is_err() + ); + let lifecycle = || server_api::RuntimeWorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }; + assert!( + service + .runtime_worker_stop_alias( + context(), + "offline-runtime".into(), + "subject".into(), + lifecycle() + ) + .await + .is_err() + ); + assert!( + service + .runtime_worker_cancel_alias( + context(), + "offline-runtime".into(), + "subject".into(), + lifecycle() + ) + .await + .is_err() + ); + assert!( + service + .runtime_worker_pin( + context(), + TEST_WORKSPACE_ID.into(), + "offline-runtime".into(), + "subject".into() + ) + .await + .is_err() + ); + let cleanup = service + .runtime_cleanup_execute( + context(), + TEST_WORKSPACE_ID.into(), + "offline-runtime".into(), + ExecuteRuntimeCleanupRequest { + expected_plan_revision: "irrelevant".into(), + expected_plan_digest: "irrelevant".into(), + worker_target_ids: Vec::new(), + workdir_target_ids: Vec::new(), + confirm_dirty_discard_target_ids: Vec::new(), + }, + ) + .await + .unwrap_err(); + assert!(cleanup.message.contains("target-bound WorkerRemove")); + assert_eq!( + api.store + .get_worker_registry( + TEST_WORKSPACE_ID, + &RuntimeWorkerRef::new("offline-runtime", "subject") + ) + .unwrap() + .unwrap() + .retention_state, + "normal" + ); +} + +#[test] +fn uncertain_unary_input_keeps_a_distinct_diagnostic_without_claiming_acceptance() { + let result = worker_input_result_to_api(WorkerInputResult { + state: InternalWorkerOperationState::Rejected, + disposition: WorkerInputDisposition::Unknown, + worker: RuntimeWorkerRef::new("runtime-a", "worker-a"), + runtime_run_id: None, + diagnostics: Vec::new(), + }); + assert_eq!(result.state, server_api::WorkerOperationState::Rejected); + assert!( + result + .diagnostics + .iter() + .any(|diagnostic| diagnostic.code == "worker_input_outcome_unknown") + ); +} + +#[tokio::test] +async fn grant_replacement_while_admission_waits_cannot_use_another_grants_lock() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + let controller = RuntimeWorkerRef::new("offline-runtime", "controller"); + let subject = RuntimeWorkerRef::new("offline-runtime", "subject"); + for identity in [&controller, &subject] { + seed_worker_source_member(&api, &identity.runtime_id, &identity.worker_id); + } + let grant = |id: &str| WorkerControlGrantRecord { + workspace_id: TEST_WORKSPACE_ID.into(), + grant_id: id.into(), + controller: controller.clone(), + subject: subject.clone(), + relation: "spawned".into(), + origin: "test".into(), + permissions: vec!["pin".into()], + operation_id: id.into(), + created_at: now_registry_timestamp(), + revoked_at: None, + }; + api.store + .create_worker_control_grant(&grant("original-grant")) + .unwrap(); + let guard = worker_control_lock(&api, "original-grant") + .lock_owned() + .await; + let handle = WorkspaceWorker::resolve(&api, &subject.runtime_id, &subject.worker_id).unwrap(); + let context = WorkerOperationContext::WorkerControl { + controller: controller.clone(), + }; + let admission = handle.set_pinned(&context, true); + tokio::pin!(admission); + // Poll deterministically to the held lock before replacing grant authority. + assert!(futures::poll!(admission.as_mut()).is_pending()); + api.store + .revoke_worker_control_grant( + TEST_WORKSPACE_ID, + "original-grant", + &now_registry_timestamp(), + ) + .unwrap(); + api.store + .create_worker_control_grant(&grant("replacement-grant")) + .unwrap(); + drop(guard); + assert!(matches!( + admission.await.unwrap_err().error, + Error::RepositoryConflict(_) + )); + assert_eq!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &subject) + .unwrap() + .unwrap() + .retention_state, + "normal" + ); +} + +#[tokio::test] +async fn browser_protocol_connection_and_cleanup_reject_cookie_origin_mismatch() { + let workspace = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(workspace.path()).await; + let identity = spawn_ticket_check_source(&api, "operation-origin"); + let handle = WorkspaceWorker::resolve(&api, &identity.runtime_id, &identity.worker_id).unwrap(); + let (mut context, _) = browser_operation_context(&api).await; + let WorkerOperationContext::Browser { headers, .. } = &mut context else { + unreachable!() + }; + headers.insert(ORIGIN, "https://untrusted.example".parse().unwrap()); + assert!(matches!( + handle.connect_protocol(&context).await, + Err(Error::WorkspacePermissionDenied(_)) + )); + let result = execute_runtime_cleanup_with_context( + &api, + EMBEDDED_WORKER_RUNTIME_ID, + ExecuteRuntimeCleanupRequest { + expected_plan_revision: "irrelevant".into(), + expected_plan_digest: "irrelevant".into(), + worker_target_ids: Vec::new(), + workdir_target_ids: Vec::new(), + confirm_dirty_discard_target_ids: Vec::new(), + }, + &context, + ) + .await + .unwrap_err(); + assert!(matches!(result.error, Error::WorkspacePermissionDenied(_))); + assert!(execution.protocol_methods().is_empty()); +} diff --git a/crates/workspace-server/src/server/worker_operations.rs b/crates/workspace-server/src/server/worker_operations.rs new file mode 100644 index 000000000..d8ada5fec --- /dev/null +++ b/crates/workspace-server/src/server/worker_operations.rs @@ -0,0 +1,591 @@ +//! Workspace identity-bound operation reception. Runtime/controller remain execution authority. +use super::*; + +/// Caller authority supplied by authenticated Server adapters, never public JSON/identity hints. +/// The unchecked Backend variant exists only for lower-level regression fixtures. +#[derive(Clone)] +pub(crate) enum WorkerOperationContext { + #[cfg(test)] + Backend, + WorkerControl { + controller: RuntimeWorkerRef, + }, + Browser { + headers: HeaderMap, + source: protocol::AuthenticatedInputSource, + }, +} + +impl WorkerOperationContext { + pub(super) async fn authorize_workspace(&self, api: &WorkspaceApi) -> ApiResult<()> { + if let Self::Browser { headers, source } = self { + let actor = resolve_actor(&ServerAuthApi::from(api), headers) + .await? + .ok_or_else(|| { + Error::WorkspacePermissionDenied( + "Worker connection credentials were revoked".into(), + ) + })?; + if matches!(actor.auth_method, ActorAuthMethod::BrowserSession) { + let AuthConfig::Passkey { origin, .. } = &api.config.auth; + if headers.get(ORIGIN).and_then(|value| value.to_str().ok()) + != Some(origin.as_str()) + { + return Err(Error::WorkspacePermissionDenied( + "cross-origin Worker mutation denied".into(), + ) + .into()); + } + } + if authenticated_browser_input_source(&actor) != *source { + return Err(Error::WorkspacePermissionDenied( + "Worker connection Account changed".into(), + ) + .into()); + } + if !api + .store + .get_workspace(api.workspace_id()) + .await? + .is_some_and(|workspace| workspace.state == "active") + { + return Err(Error::WorkspacePermissionDenied( + "Workspace is not accepting Worker mutations".into(), + ) + .into()); + } + } + Ok(()) + } + pub(super) fn from_request( + api: &WorkspaceApi, + context: &server_api::ServerRequestContext, + ) -> std::result::Result { + if let Some(source) = context.runtime_source.as_ref() { + // Check hints against middleware's verified subject without replaying its proof + // or requiring the target Runtime to be online just to represent identity. + current_worker_contract_headers(context)?; + let controller = RuntimeWorkerRef::new( + &source.runtime_id, + source + .worker_id + .as_deref() + .expect("source binding required a Worker identity"), + ); + if api + .store + .get_worker_registry(api.workspace_id(), &controller) + .map_err(|error| ApiError::from(error).into_repository_api_error())? + .is_none() + { + return Err(ApiError::from(Error::UnknownWorker { worker: controller }) + .into_repository_api_error()); + } + return Ok(Self::WorkerControl { controller }); + } + let actor = context.actor.as_ref().ok_or_else(|| { + ApiError::from(Error::WorkspacePermissionDenied( + "authenticated Worker operation source required".into(), + )) + .into_repository_api_error() + })?; + let headers = contract_request_headers(context)?; + Ok(Self::Browser { + headers, + source: authenticated_browser_input_source(actor), + }) + } +} + +/// A registry identity, not an authorization lease or a snapshot of live execution. +#[derive(Clone)] +pub(crate) struct WorkspaceWorker { + api: Services, + removal: WorkerRemovalService, + identity: RuntimeWorkerRef, +} + +// The removal-only specialization keeps the installed embedded dispatcher from +// retaining WorkspaceApi -> Runtime -> dispatcher -> WorkspaceApi as a strong cycle. +// Both specializations resolve the same registry identity and call the same remove method. +impl WorkspaceWorker { + fn resolve_with_services( + api: Services, + removal: WorkerRemovalService, + runtime_id: &str, + reference: &str, + ) -> Result { + let direct = RuntimeWorkerRef::new(runtime_id, reference); + let identity = if let Some(record) = removal + .store + .get_worker_registry(&removal.workspace_id, &direct)? + { + record.worker + } else { + let worker = removal + .store + .resolve_worker_resource_reference(&removal.workspace_id, reference)? + .filter(|worker| worker.runtime_id == runtime_id) + .ok_or_else(|| Error::UnknownWorker { + worker: direct.clone(), + })?; + removal + .store + .get_worker_registry(&removal.workspace_id, &worker)? + .ok_or_else(|| Error::UnknownWorker { worker: direct })? + .worker + }; + Ok(Self { + api, + removal, + identity, + }) + } + pub(super) fn identity(&self) -> &RuntimeWorkerRef { + &self.identity + } + + fn current_record(&self) -> ApiResult { + self.removal + .store + .get_worker_registry(&self.removal.workspace_id, &self.identity)? + .ok_or_else(|| { + Error::UnknownWorker { + worker: self.identity.clone(), + } + .into() + }) + } + + pub(super) async fn remove( + &self, + source: crate::worker_source::VerifiedWorkerMutationSource, + reason: &str, + ) -> std::result::Result { + self.current_record() + .map_err(|error| error.error.to_string())?; + self.removal + .execute_async( + source, + &self.identity.runtime_id, + &self.identity.worker_id, + reason, + ) + .await + } +} + +/// Server-owned adapter for Runtime's independent embedded Tool dispatcher. +/// The proof was target-bound and verified before this synchronous transport seam. +pub(super) struct EmbeddedWorkspaceWorkerRemoveExecutor { + removal: WorkerRemovalService, +} +impl EmbeddedWorkspaceWorkerRemoveExecutor { + pub(super) fn new(api: &WorkspaceApi) -> Self { + Self { + removal: WorkerRemovalService::new(api), + } + } +} +impl crate::worker_source::VerifiedWorkerRemoveExecutor for EmbeddedWorkspaceWorkerRemoveExecutor { + fn execute( + &self, + source: crate::worker_source::VerifiedWorkerMutationSource, + target_runtime_id: &str, + target_worker_id: &str, + reason: &str, + ) -> std::result::Result { + let worker = match WorkspaceWorker::resolve_with_services( + (), + self.removal.clone(), + target_runtime_id, + target_worker_id, + ) { + Ok(worker) => worker, + Err(Error::UnknownWorker { .. }) => { + return Ok(worker_remove_error_response( + StatusCode::NOT_FOUND, + "unknown_worker", + "The target Worker is not known in this Workspace", + )); + } + Err(error) => return Err(error.to_string()), + }; + let reason = reason.to_owned(); + std::thread::spawn(move || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|error| error.to_string())? + .block_on(worker.remove(source, &reason)) + }) + .join() + .map_err(|_| "embedded WorkerRemove executor thread panicked".to_owned())? + } +} + +impl WorkspaceWorker { + pub(crate) fn resolve(api: &WorkspaceApi, runtime_id: &str, reference: &str) -> Result { + Self::resolve_with_services( + api.clone(), + WorkerRemovalService::new(api), + runtime_id, + reference, + ) + } + + pub(super) async fn authorize_operation( + &self, + context: &WorkerOperationContext, + permission: &str, + ) -> ApiResult>> { + self.current_record()?; + match context { + #[cfg(test)] + WorkerOperationContext::Backend => Ok(None), + WorkerOperationContext::Browser { .. } => { + context.authorize_workspace(&self.api).await?; + Ok(None) + } + WorkerOperationContext::WorkerControl { controller } => { + if self + .api + .store + .get_worker_registry(self.api.workspace_id(), controller)? + .is_none() + { + return Err(Error::UnknownWorker { + worker: controller.clone(), + } + .into()); + } + let grant = authorize_known_worker_permission( + &self.api, + self.api.workspace_id(), + controller, + &self.identity, + permission, + )?; + let guard = worker_control_lock(&self.api, &grant.grant_id) + .lock_owned() + .await; + self.current_record()?; + let current_grant = authorize_known_worker_permission( + &self.api, + self.api.workspace_id(), + controller, + &self.identity, + permission, + )?; + if current_grant.grant_id != grant.grant_id { + return Err(Error::RepositoryConflict( + "Worker control grant changed while waiting for admission".into(), + ) + .into()); + } + Ok(Some(guard)) + } + } + } + + pub(super) async fn restore( + &self, + context: &WorkerOperationContext, + ) -> ApiResult { + let _guard = self.authorize_operation(context, "restore").await?; + // Preserve singleton, Workdir, credentials, bindings and pending operation reconciliation. + self.api.restore_workspace_worker(&self.identity) + } + + pub(super) async fn stop( + &self, + context: &WorkerOperationContext, + request: WorkerLifecycleRequest, + ) -> ApiResult { + let _guard = self.authorize_operation(context, "stop").await?; + let result = self + .api + .runtime + .stop_worker(&self.identity, request) + .map_err(|e| e.into_error())?; + if result.state != InternalWorkerOperationState::Accepted { + return Ok(result); + } + parse_runtime_worker_id_for_registry(&self.identity.worker_id)?; + let session_lock = current_worker_session_lock(&self.api, &self.identity); + let _session_guard = session_lock.lock().await; + close_current_worker_session_locked(&self.api, &self.identity).await?; + if let Some(record) = self + .api + .store + .get_worker_registry(self.api.workspace_id(), &self.identity)? + { + sync_linked_workdir_after_worker_stop(&self.api, &self.identity.runtime_id, &record)?; + } + Ok(result) + } + + pub(super) async fn cancel( + &self, + context: &WorkerOperationContext, + request: WorkerLifecycleRequest, + ) -> ApiResult { + let _guard = self.authorize_operation(context, "cancel").await?; + Ok(self + .api + .runtime + .cancel_worker(&self.identity, request) + .map_err(|e| e.into_error())?) + } + + pub(super) async fn input( + &self, + context: &WorkerOperationContext, + mut request: WorkerInputRequest, + ) -> ApiResult { + let permission = if request.kind == WorkerInputKind::Notify { + "notify" + } else { + "send_input" + }; + let _guard = self.authorize_operation(context, permission).await?; + self.ensure_interactive()?; + if matches!(context, WorkerOperationContext::WorkerControl { .. }) + && request.kind == WorkerInputKind::User + { + request.kind = WorkerInputKind::UserIfIdle; + } + Ok(self + .api + .runtime + .send_input(&self.identity, request) + .map_err(|e| e.into_error())?) + } + + fn ensure_interactive(&self) -> ApiResult<()> { + if self + .api + .store + .worker_registry_projection(self.api.workspace_id(), &self.identity)? + .and_then(|projection| projection.job) + .is_some() + { + return Err(Error::InvalidInput("Backend Job Workers accept only their Backend-owned immutable input; Console is read-only".into()).into()); + } + Ok(()) + } + + pub(super) async fn set_pinned( + &self, + context: &WorkerOperationContext, + pinned: bool, + ) -> ApiResult { + let _guard = self.authorize_operation(context, "pin").await?; + let retention_state = if pinned { "pinned" } else { "normal" }; + let changed = self.api.worker_projection.publish_ordered(|| { + self.api + .store + .update_worker_retention( + self.api.workspace_id(), + &self.identity, + retention_state, + now_registry_timestamp().as_str(), + ) + .map(|commit| (!commit.changes.is_empty(), commit)) + })?; + if !changed { + return Err(cleanup_api_error( + &self.identity.runtime_id, + "workspace_worker_retention_unknown_worker", + "Worker is not known to the Backend registry", + )); + } + Ok(WorkerRetentionResponse { + workspace_id: self.api.workspace_id().to_owned(), + runtime_id: self.identity.runtime_id.clone(), + worker_id: self.identity.worker_id.clone(), + pinned, + retention_state: retention_state.into(), + }) + } + + /// The caller owns the full cleanup-plan revision/digest check. The removal service + /// rechecks live assignment/retention/use at its existing commit boundary. + pub(super) async fn delete_from_plan( + &self, + context: &WorkerOperationContext, + candidate: &CleanupWorkerCandidate, + ) -> ApiResult<()> { + let _guard = self.authorize_operation(context, "remove").await?; + if candidate.runtime_id != self.identity.runtime_id + || candidate.runtime_worker_id != self.identity.worker_id + { + return Err(Error::InvalidInput( + "cleanup candidate does not address the resolved Worker".into(), + ) + .into()); + } + WorkerRemovalService::new(&self.api) + .execute_cleanup_removal(candidate) + .await + } + + pub(crate) async fn connect_protocol( + &self, + context: &WorkerOperationContext, + ) -> Result { + let _guard = self + .authorize_operation(context, "observe") + .await + .map_err(|error| error.error)?; + let source = match context { + WorkerOperationContext::Browser { source, .. } => Some(source), + _ => None, + }; + let connection = + connect_workspace_worker_protocol(&self.api, &self.identity, source).await?; + Ok(WorkspaceWorkerConnection { + sender: WorkspaceWorkerMethodSender { + worker: self.clone(), + methods: connection.methods, + }, + events: connection.events, + }) + } +} + +/// An execution connection is bound to exactly the identity which created it. No caller +/// can pair a different target handle with a subscription's raw sending capability. +#[derive(Clone)] +pub(crate) struct WorkspaceWorkerMethodSender { + worker: WorkspaceWorker, + methods: tokio::sync::mpsc::Sender, +} +pub(crate) struct WorkspaceWorkerConnection { + pub(crate) sender: WorkspaceWorkerMethodSender, + pub(crate) events: tokio::sync::mpsc::Receiver, +} +impl WorkspaceWorkerMethodSender { + pub(crate) async fn send( + &self, + context: &WorkerOperationContext, + method: protocol::Method, + ) -> Result<()> { + match method { + protocol::Method::Pause { command } => self.pause(context, command).await, + protocol::Method::Resume { command } => self.resume(context, command).await, + protocol::Method::Cancel { command } => self.cancel(context, command).await, + protocol::Method::Compact { command } => self.compact(context, command).await, + protocol::Method::Submit { + submission_request_id, + input, + } => self.submit(context, submission_request_id, input).await, + protocol::Method::Notify { + notification_request_id, + message, + } => self.notify(context, notification_request_id, message).await, + other => self.dispatch(context, other).await, + } + } + async fn dispatch( + &self, + context: &WorkerOperationContext, + method: protocol::Method, + ) -> Result<()> { + self.send_checked(context, method) + .await + .map_err(|error| error.error) + } + pub(crate) async fn pause( + &self, + context: &WorkerOperationContext, + command: protocol::WorkerCommandEnvelope, + ) -> Result<()> { + self.dispatch(context, protocol::Method::Pause { command }) + .await + } + pub(crate) async fn resume( + &self, + context: &WorkerOperationContext, + command: protocol::WorkerCommandEnvelope, + ) -> Result<()> { + self.dispatch(context, protocol::Method::Resume { command }) + .await + } + pub(crate) async fn cancel( + &self, + context: &WorkerOperationContext, + command: protocol::WorkerCommandEnvelope, + ) -> Result<()> { + self.dispatch(context, protocol::Method::Cancel { command }) + .await + } + pub(crate) async fn compact( + &self, + context: &WorkerOperationContext, + command: protocol::WorkerCommandEnvelope, + ) -> Result<()> { + self.dispatch(context, protocol::Method::Compact { command }) + .await + } + pub(crate) async fn submit( + &self, + context: &WorkerOperationContext, + submission_request_id: String, + input: Vec, + ) -> Result<()> { + self.dispatch( + context, + protocol::Method::Submit { + submission_request_id, + input, + }, + ) + .await + } + pub(crate) async fn notify( + &self, + context: &WorkerOperationContext, + notification_request_id: String, + message: String, + ) -> Result<()> { + self.dispatch( + context, + protocol::Method::Notify { + notification_request_id, + message, + }, + ) + .await + } + + async fn send_checked( + &self, + context: &WorkerOperationContext, + method: protocol::Method, + ) -> ApiResult<()> { + let _guard = self + .worker + .authorize_operation(context, "send_input") + .await?; + self.worker.ensure_interactive()?; + let WorkerOperationContext::Browser { source, .. } = context else { + return Err(Error::WorkspacePermissionDenied( + "Browser Worker protocol context required".into(), + ) + .into()); + }; + let method = authorize_browser_worker_method(method, source) + .map_err(|message| Error::InvalidInput(message.into()))?; + self.methods.send(method).await.map_err(|_| { + Error::RuntimeOperationFailed { + runtime_id: self.worker.identity.runtime_id.clone(), + code: "worker_protocol_closed".into(), + message: + "Worker execution connection closed; no implicit Restore or retry was performed" + .into(), + } + .into() + }) + } +} diff --git a/crates/workspace-server/src/workspace_subscription.rs b/crates/workspace-server/src/workspace_subscription.rs index 54f3182ae..058d8bc99 100644 --- a/crates/workspace-server/src/workspace_subscription.rs +++ b/crates/workspace-server/src/workspace_subscription.rs @@ -12,7 +12,7 @@ use worker_runtime::identity::RuntimeWorkerRef; use crate::runtime_subscription::RuntimeSubscriptionBroker; use crate::server::{ - WorkspaceApi, authorize_browser_worker_method, connect_workspace_worker_protocol, + WorkerOperationContext, WorkspaceApi, WorkspaceWorker, WorkspaceWorkerMethodSender, }; const OUTBOUND_CAPACITY: usize = 256; @@ -20,13 +20,13 @@ const WORKER_PROTOCOL_SNAPSHOT_TIMEOUT: std::time::Duration = std::time::Duratio struct ActiveSubscription { task: tokio::task::JoinHandle<()>, - methods: Option>, + methods: Option, } pub(crate) async fn serve_workspace_subscription( api: WorkspaceApi, socket: WebSocket, - input_source: protocol::AuthenticatedInputSource, + context: WorkerOperationContext, ) { let broker = api.runtime_subscription_broker().clone(); let (mut socket_sender, mut socket_receiver) = socket.split(); @@ -90,16 +90,20 @@ pub(crate) async fn serve_workspace_subscription( worker_id, runtime_id: Some(runtime_id), } => { - let worker = RuntimeWorkerRef::new(&runtime_id, worker_id.as_str()); - match connect_workspace_worker_protocol( + let connection = match WorkspaceWorker::resolve( &api, - &worker, - Some(&input_source), - ) - .await - { + &runtime_id, + worker_id.as_str(), + ) { + Ok(worker) => worker + .connect_protocol(&context) + .await + .map_err(|error| error.to_string()), + Err(error) => Err(error.to_string()), + }; + match connection { Ok(connection) => { - let methods = connection.methods.clone(); + let methods = connection.sender.clone(); let task = tokio::spawn(run_worker_protocol( request_id, subscription_id.clone(), @@ -165,12 +169,7 @@ pub(crate) async fn serve_workspace_subscription( else { break; }; - let Ok(method) = - authorize_browser_worker_method(message.method, &input_source) - else { - break; - }; - if methods.send(method).await.is_err() { + if methods.send(&context, message.method).await.is_err() { break; } } diff --git a/docs/architecture/workspace-worker-operations.md b/docs/architecture/workspace-worker-operations.md new file mode 100644 index 000000000..ea247b845 --- /dev/null +++ b/docs/architecture/workspace-worker-operations.md @@ -0,0 +1,140 @@ +# Workspace Worker operations + +T-702 documents the implemented Workspace operation-reception boundary. It does not move execution authority into the Workspace registry or make all Worker-related routes use one API. The common implementation is [`worker_operations.rs`](../../crates/workspace-server/src/server/worker_operations.rs); production adapters and service orchestration are in [`server.rs`](../../crates/workspace-server/src/server.rs). + +## Identity reception is not execution authority + +`WorkspaceWorker::resolve` resolves a Workspace registry reference to a `RuntimeWorkerRef` (Runtime ID and Worker ID). Resolution first tries that exact pair, then a Workspace resource reference constrained to the requested Runtime. It returns an identity-bound handle, not a live execution snapshot, authorization lease, or promise that the Worker is running. Each operation rereads the target registry record and authorizes its own context. + +The owners have different responsibilities: + +- **Workspace registry and domain services:** durable Workspace identity, retention/pinning, Workdir links and sessions, Ticket assignments, credentials/bindings, control grants, restore coordination, and removal plans/catalog publication. Registry projections and observation streams describe Workers; they do not admit controller commands. +- **`RuntimeRegistry` and Runtime adapters:** select the Runtime owning the resolved pair and delegate execution. Remote and embedded adapters preserve their existing transport/error behavior. Runtime owns lifecycle operation settlement and checks Workspace scope at the scoped embedded execution boundary. +- **Execution backend and Worker controller/session:** own the actual execution endpoint, command admission, authoritative Idle/Busy state, in-flight run transitions, Submit FIFO and durable receipts, Notify receipts, and compaction. A catalog status is not sufficient evidence for command admission. + +The term *controller* in a `WorkerControl` context means the calling Runtime Worker holding a grant over a target. It is not the target's in-process execution controller. + +## Production ingress → common method → execution owner + +For the HTTP rows below, `R` abbreviates `/api/runtimes/{runtime_id}/workers/{worker_id}` and `W` abbreviates `/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}`. Aliases operate in the server's configured Workspace; scoped routes additionally validate the Workspace path. The declarations live in [`server-api`](../../crates/server-api/src/lib.rs), and the generated contract service implementations in `server.rs` are the production unary adapters. + +| Ingress | Common reception / coordination | Execution or persistence owner | +| --- | --- | --- | +| `POST R/restore` and `POST W/restore` | `WorkerOperationContext::from_request` → `restore_runtime_worker_with_context` → `WorkspaceWorker::restore`; scoped route first uses `scoped_restore_runtime_worker_with_context` | `WorkspaceApi::restore_workspace_worker` retains the existing singleton, Workdir, credential/binding and pending-operation reconciliation flow; Runtime/backend owns restored execution | +| `POST R/input` and `POST W/input` | Resolve → convert request → `WorkspaceWorker::input` | `RuntimeRegistry::send_input` → Runtime adapter → execution backend/controller/session | +| `POST R/stop` and `POST W/stop` | `WorkspaceWorker::stop` | Runtime owns Stop; on an **Accepted** result Workspace closes the current Workdir session and synchronizes linked Workdir state. Rejected/Unsupported results do not perform that finalization | +| `POST R/cancel` and `POST W/cancel` | `WorkspaceWorker::cancel` | `RuntimeRegistry::cancel_worker` → target execution controller; not registry removal | +| `PUT W/pin` / `DELETE W/pin` | `WorkspaceWorker::set_pinned(true/false)` | Workspace store retention update, ordered through `worker_projection`; no execution command | +| `GET /api/w/{workspace_id}/runtimes/{runtime_id}/cleanup-plan` | `build_runtime_cleanup_plan` (preview, not a Worker mutation) | Workspace cleanup planner/projections | +| `POST /api/w/{workspace_id}/runtimes/{runtime_id}/cleanup-executions` | `execute_runtime_cleanup_with_context` validates plan revision/digest and selected candidates → each selected Worker's `delete_from_plan` | `WorkerRemovalService::execute_cleanup_removal` → existing retention planning, Runtime retention execution, and Workspace catalog commit. Workdir-only actions remain in the cleanup service | +| `POST /api/w/{workspace_id}/workers/remove` (`WorkerRemove`, not a browser removal alias) | `worker_remove_contract` verifies its source proof against the requested target → `WorkspaceWorker::remove` | `WorkerRemovalService::execute_async` checks the `remove` grant and coordinates retention/removal. It rejects self-removal and requires a stopped target for a new removal; existing durable removal operations have their own recovery path | +| Embedded `WorkerRemove` Tool | Runtime-owned in-process target-bound proof → `EmbeddedServerWorkerMutationDispatcher` verifies/consumes proof → installed `EmbeddedWorkspaceWorkerRemoveExecutor` → `WorkspaceWorker::resolve_with_services` → the same `WorkspaceWorker::remove` | Same `WorkerRemovalService::execute_async`, grant-lock/recheck, self/live/retention guards, durable plan/recovery and catalog commit as HTTP; no Runtime-to-Server dependency | +| Dedicated WS `R/protocol/ws` and `W/protocol/ws` | `worker_protocol_ws` (scoped wrapper for `W`) → `WorkspaceWorker::connect_protocol` → `workspace_worker_protocol_session` → bound sender | One remote WS or embedded execution transport; controller/session owns protocol operation results | +| Multiplexed WS `/api/w/{workspace_id}/protocol/ws`, runtime-scoped `WorkerProtocol` selector | `scoped_workspace_protocol_ws` → `serve_workspace_subscription` → resolve + `connect_protocol`; subscription stores the bound sender | Same connected execution transport as dedicated WS; the subscription layer wraps its snapshot/events, not a different execution implementation | +| Worker-control `POST /api/w/{workspace_id}/worker-control/workers/{runtime_id}/{worker_id}/{input,cancel,stop,restore}` | `send_known_worker_input`, `cancel_known_worker`, `stop_known_worker`, `restore_known_worker` authenticate the Worker source, construct `WorkerControl`, and call the same identity methods/restore adapter | Same Runtime execution owners, under the applicable active target grant; known-worker input adds its existing non-Accepted-result error translation | + +Both WS adapters call `WorkspaceWorkerMethodSender::send`, which selects the following **named methods** rather than implementing separate command semantics in each transport: + +| Public protocol method | Named common sender method | Distinct owner/receipt | +| --- | --- | --- | +| `Pause { command }` | `pause` | Controller command admission/acknowledgement; pause is not Stop | +| `Resume { command }` | `resume` | Controller resumes the paused run; not Restore | +| `Cancel { command }` | `cancel` | Controller command admission for cancelling the current run; distinct from HTTP lifecycle request/result | +| `Compact { command }` | `compact` | Controller maintenance/compaction, with command acknowledgement and subsequent lifecycle events | +| `Submit { submission_request_id, input }` | `submit` | Session durable Submit receipt/FIFO and later run execution | +| `Notify { notification_request_id, message }` | `notify` | Session durable Notify receipt and advisory delivery, not the Submit FIFO | + +All six methods funnel through `dispatch` / `send_checked`. Other browser-allowed protocol methods also use that checked dispatch; the six named methods are not an exhaustive allowlist. For example pending-submission operations, rewind, Shutdown, completions, ListWorkers, RestoreWorker and RegisterPeer are allowed by `authorize_browser_worker_method`. This forwarding does not replace their lower-layer semantics with Workspace registry operations. + +`build_inner_router` installs the three production WS routes using typed `websocket_route` registrations and merges the generated Workspace contract router for unary routes. [`workspace_subscription.rs`](../../crates/workspace-server/src/workspace_subscription.rs) handles multiplexing. Its `WorkspaceWorkers` subscription is a separate catalog stream with no method sender; only a runtime-scoped `WorkerProtocol` subscription carries an operational sender. + +## Context and authorization restrictions + +`WorkerOperationContext` is crate-private. Clients cannot submit this enum to select their authority. + +- **Request-derived Worker control:** `from_request` uses the middleware-verified Runtime Worker subject and current Workspace registry membership. The shared source-binding helper rejects Runtime-only proofs and conflicting identity hints; no hint-only request or Account fallback can impersonate a Worker. Known-worker adapters bind their source through the same verified request extension before constructing the operation context. +- **Account/browser context:** despite the `Browser` name, this also represents account-authenticated unary requests. It retains request headers and the server-derived Account input source. On each common authorization call, the server re-resolves credentials, rejects revoked credentials or an Account change, requires the configured Origin for `BrowserSession` actors, and requires an active Workspace. It does not interpret the operation's permission string as a Worker-control grant check for an Account actor; request middleware remains part of the route's authorization boundary. +- **Worker-control context:** the target and calling Worker must be registered. The server requires an active grant for this controller/target pair and the operation's permission, locks by grant ID, then rereads the target and rechecks permission and the exact grant ID after acquiring the lock. A replacement grant requires new admission under its own lock. A missing active grant is reported as UnknownWorker; a present grant lacking the permission is InvalidInput. Merely knowing an ID does not grant control. +- **Test Backend context:** `WorkerOperationContext::Backend` exists only under `#[cfg(test)]`. It bypasses context authorization after the target-record check. There is no production Backend enum variant for a client, Worker Tool, or generic internal caller to select. Production Backend services continue to own their explicit domain flows. + +Worker-control grants are Workspace-owned records, not inferred parentage or universal access to every Runtime Worker. The production spawn flow persists a controller/subject grant with `send_input`, `notify`, `cancel`, `stop`, `restore`, `remove`, and `observe`; that default does not include `pin`. Known-worker discovery follows active grants, while Internal children use their separate parent-owned registry. Other grant origins/permission sets must be read from their own authority rather than assumed from this spawn default. + +Permissions are `restore`, `stop`, `cancel`, `pin`, and `remove` for those identity operations. HTTP input uses `notify` for Notify and `send_input` for all other accepted kinds. `connect_protocol` checks `observe`; every connected method dispatch checks `send_input`, requires an interactive target, then explicitly requires a **Browser** context. Thus an observe grant is not a mutation grant, and a Worker-control grant does not enable browser WS method forwarding. WS Notify also uses this browser-only `send_input` check, not HTTP input's `notify` grant mapping. + +`input` and all connected protocol method sends reject Workers with a Backend Job projection: their immutable input is Backend-owned and their Console is read-only. Connecting for observation is not itself rejected by that interactive check. + +For Worker-control user input, `input` converts `User` to `UserIfIdle`; admission belongs to the target controller and never queues behind an active run. Browser Submit may durably queue while Running/Paused. Notify is advisory input, not a queued Submit; it can be incorporated at a between-turn boundary or drive the notification flow while Idle. The known-worker input adapter turns definite non-Accepted input results into errors, but keeps uncertain outcomes as the existing typed result with a `worker_input_outcome_unknown` diagnostic, and supplies the Idle-only guidance for user input. + +Scoped Restore may coordinate Ticket assignment only when both `ticket_id` and `assignment_operation_id` are supplied. A Worker-control context cannot use either assignment parameter. Aggregate cleanup rejects all Worker-control requests: Worker callers must retain the stronger target-bound WorkerRemove proof/retention path. A control grant is not authority over Ticket assignments or Workdir cleanup. The cleanup coordinator revalidates Account credentials, Origin, and active Workspace even for a Workdir-only request. + +The HTTP and embedded `WorkerRemove` entry points share the registry-bound removal method. `WorkerRemove` intentionally has a distinct authorization contract from the other operations. It accepts a verified mutation source rather than `WorkerOperationContext`, verifies its target-bound proof before calling the identity method, and keeps source-specific and grant revalidation in the removal service. It is not permission-free just because it does not call `authorize_operation`. + +The embedded executor retains only the removal services with a **weak** RuntimeRegistry reference. Its removal-only `WorkspaceWorker<()>` specialization shares resolution, current-record checks and the removal method with the full `WorkspaceWorker` handle. This service projection avoids a strong `WorkspaceApi → Runtime → dispatcher → WorkspaceApi` ownership cycle; it is not another Worker ledger or an authorization fallback. It cannot call the full handle's restore/input/protocol methods. The existing independent Runtime-owned proof forwarder and verification/replay seam stay unchanged. + +## Tokens and outcomes are operation-specific + +There is no universal Worker command token or universal success result. + +- `RuntimeWorkerRef` identifies the registry target; a Workspace resource reference is an ingress lookup form. Neither identifies a particular run or proves authority. +- `WorkerCommandEnvelope.command_id` orders controller lifecycle commands. `WorkerCommandAcknowledgement` contains command kind, disposition (`Accepted`, `StaleCommandId`, `Conflict`, `InvalidState`) and authoritative state. These IDs belong to controller command admission, not durable Submit identity or Workspace removal planning. +- `submission_request_id` is the caller's Submit retry key in an authenticated-source namespace. A durable `SubmissionAccepted` receipt also contains the Worker-allocated `submission_id` and `Started`/`Queued` disposition. `SubmissionRejected` is correlated by request ID. Acceptance is not a successful completed run; `last_finished_submission_request_id` is a teardown fence, not a result-success claim. +- `notification_request_id` correlates `NotificationAccepted` / `NotificationRejected`. Notify has its own durable receipt and exact-payload retry behavior; it does not allocate a Submit FIFO entry. Browser Submit/Notify are rewritten to `SubmitTracked`/`NotifyTracked` with the server's Account source. Browser-supplied tracked forms, `SubmitIfIdle`, and `WorkerEvent` are rejected. +- Runtime `WorkerLifecycleOperationId` is a durable lifecycle/reconciliation identity, separate from a controller command ID. Ticket `assignment_operation_id`, cleanup plan revision/digest, and retention plan/operation IDs and fingerprints likewise belong to their own domain checks. A multiplexed `subscription_id` only routes frames to an already-bound sender. + +The common sender returns `Result<()>` after authorization and enqueueing on its transport channel. This is **not** a controller acknowledgement, durable receipt, or proof of upstream delivery. Actual outcomes arrive on the event stream. The dedicated WS reports dispatch errors as protocol error events; the multiplexed adapter currently terminates the connection when sending a method fails. It also requires an initial session Snapshot before accepting the WorkerProtocol subscription. + +Unary input/lifecycle responses preserve their existing API projections: `WorkerOperationState` is `Accepted`, `Unsupported`, or `Rejected`, with Runtime/Worker IDs and diagnostics. Restore instead has `Accepted`, `Rejected`, `RolledBack`, and `ReconciliationRequired`. Pin returns retention state; cleanup returns per-target actions/statuses; WorkerRemove has its own response/error body. These are not interchangeable acknowledgements. + +A transport or store failure can leave an **unknown outcome** after effects have started. Lower-layer input results have `WorkerInputDisposition::{Accepted,Rejected,Unknown}` and an optional `runtime_run_id`, but `worker_input_result_to_api` projects only state, target IDs and diagnostics: the public unary input result does not expose those two fields. Unknown delivery is explicitly preserved as a `worker_input_outcome_unknown` diagnostic; neither HTTP adapter changes it into definite rejection or retries it. Runtime execution results also distinguish Busy/Rejected/Errored/Unsupported; error handling and durable restore/removal recovery remain lower-layer responsibilities. Do not infer that a rejected-looking projected response or a closed WS proves that nothing was accepted. The new reception layer does not add automatic replay or an exactly-once guarantee across every ingress. In particular, `worker_input_request_from_api` sets `submission_request_id` to `None`; the unary Workspace input contract does not provide the browser WS Submit retry-key contract. + +## Connected execution lifetime + +`WorkspaceWorkerMethodSender` stores both the resolved Worker handle and the sending channel privately. A WS caller cannot pair a different Worker handle with a subscription's raw sending capability. Registry identity can outlive an execution, but an operational connection does not follow it to a replacement execution. + +`connect_workspace_worker_protocol` obtains an observation source from the Workspace proxy (falling back to the Runtime only for WorkerNotFound). It then connects to one remote execution WS or attaches one embedded `WorkerProtocolTransport`. The embedded path captures the real endpoint and its initial snapshot, not the WorkerRef-wide catalog observation bus. + +Embedded dispatch uses `send_connected_protocol_method_scoped`. Runtime verifies that the transport target equals the authorized WorkerRef, checks scope and pending-operation settlement, and validates the captured endpoint before any durable mutation. Even a delayed terminal-close observation cannot make the old endpoint Stop or send to a new execution under the same identity. + +Shutdown and a closed execution relay terminate the operational stream. Embedded lifecycle/store/transport uncertainty is terminal; a settled Busy/Rejected/Unsupported rejection may leave it usable only if the endpoint still validates. Remote bridge failure closes its channels. `send_checked` reports `worker_protocol_closed` when its channel is closed and explicitly states that no implicit Restore or retry occurred. No bridge in this path retargets, restores, or reconnects to the next execution. The caller must explicitly Restore when authorized and establish a new connection; this is different from catalog observation continuing to describe the same registry identity. + +## Internal Workers are parent-owned + +Internal SubWorkers and service Internal Workers are not Workspace registry targets or independently addressable Runtime protocol subjects. They remain separate, parent-owned sessions. Their visible identity is `InternalWorkerRef.session_id` (`name` is display-only), and snapshots/events are presented through the parent's protocol stream; service-private Internal Workers are excluded from the public child inventory. + +Direct-child control/observation stays in the parent-owned Worker machinery. A Workspace Runtime Worker grant is not a grant to an arbitrary Internal Worker, and this registry handle does not create an Internal Worker Runtime record, independent transport, or implicit Restore path. The Workspace observation capture adapter explicitly rejects a non-RuntimeWorker subject; it does not resolve a SubWorker as a registry Worker. + +## Regression helpers are not production routes + +Lower-level test functions in `server.rs` must not be mistaken for installed unauthenticated endpoints: + +- `set_worker_retention`, `send_runtime_worker_input` / `dispatch_runtime_worker_input`, and `execute_runtime_cleanup` / `scoped_execute_runtime_cleanup` are `#[cfg(test)]` wrappers using the test-only Backend context. +- `worker_protocol_ws_session`, `remote_worker_protocol_ws_session`, and `embedded_worker_protocol_ws_session` are `#[cfg(test)]` transport regression helpers. Production dedicated WS uses `workspace_worker_protocol_session` with the common bound sender; production multiplexed WS uses `serve_workspace_subscription` with that same sender type. +- Production `connect_workspace_worker_protocol`, `connect_remote_worker_protocol`, and `connect_embedded_worker_protocol` are real shared connection helpers, not those test-only sessions. The grant-checked observation capture service also calls the connection helper to read a committed Snapshot; that read-only capture is outside the six operational WS methods. + +Source presence alone is not route wiring. Follow `build_inner_router`, generated contract implementations, and the typed route declarations when auditing production ingress. + +## Minimum scope, limitations, and future guard + +The boundary above covers the T-702 minimum operation set. It is not a claim that every session/file/completion API has been migrated: + +- Session snapshot/history and retained session attachment reads remain their existing observation/retention paths. They are not controller command admission. +- Attachment upload grants and upload cancellation keep their Account/target-bound grant checks; file upload and uploaded-file deletion still resolve the target and delegate through `api.runtime`, rather than calling a new `WorkspaceWorker` file method. +- Unary completions still resolve the registry reference and call `api.runtime.worker_completions`. Protocol `ListCompletions` is also browser-allowed through checked WS dispatch. Those are different ingress paths. +- Worker creation, Backend Job execution, rollback compensation, Workspace deletion, and other domain-owned orchestration are not universally rewritten to this handle. Legitimate lower-level Runtime calls still exist. + +Known limitations worth follow-up, rather than undocumented guarantees: + +1. Browser credentials/Account are rechecked for each common operation or WS method send, but not continuously for every outgoing observation event. This is not a promise of immediate revocation-driven stream teardown. +2. Cleanup is sequential rather than an all-target atomic transaction. The existing cleanup plan revision/digest and each removal service's live revalidation remain distinct protections. +3. Unary input has its existing narrower wire contract, rather than the WS authenticated receipt envelope. Identity unification does not create a new retry contract or manufacture caller request IDs across transports. + +**T-701's stale-observation Restore guard is not implemented here.** T-701 will connect the caller's observed revision/generation to Runtime's existing operation lock, distinguish a new Restore intent from pending reconciliation, and handle UI conflicts. This boundary leaves those Restore inputs/results and the pending-operation state machine intact rather than creating a duplicate guard. + +### Implementation references + +- [`server/worker_operations.rs`](../../crates/workspace-server/src/server/worker_operations.rs): identity handle, contexts, authorization, named methods and private bound sender. +- [`server.rs`](../../crates/workspace-server/src/server.rs): contract adapters, `build_inner_router`, scoped Restore/cleanup, WorkerRemovalService, dedicated WS and shared transport bridges. +- [`workspace_subscription.rs`](../../crates/workspace-server/src/workspace_subscription.rs): multiplexed WorkerProtocol binding and catalog subscriptions. +- [`hosts.rs`](../../crates/workspace-server/src/hosts.rs): Runtime registry/adapters and distinct input/lifecycle result projections. +- [`protocol`](../../crates/protocol/src/lib.rs): command envelopes/acknowledgements, Submit/Notify receipts and Internal Worker presentation. +- [`worker-runtime/runtime.rs`](../../crates/worker-runtime/src/runtime.rs) and [`execution.rs`](../../crates/worker-runtime/src/execution.rs): scoped connected endpoint validation, lifecycle identities and execution results. diff --git a/docs/report/2026-10-06-t702-command-output-readback.md b/docs/report/2026-10-06-t702-command-output-readback.md new file mode 100644 index 000000000..a9ef6bd48 --- /dev/null +++ b/docs/report/2026-10-06-t702-command-output-readback.md @@ -0,0 +1,7 @@ +# Command-output readback during T-702 + +During a cold `cargo test -p yoi-workspace-server` run, Bash bounded its inline output and reported a spill file under the Worker's declared readable `bash-output` boundary. A subsequent `Grep` of that exact returned absolute path was rejected as outside allowed scope. I did not attempt another path to bypass the rejection; instead I reran the relevant tests with bounded output filtering. + +This made a timed-out crate test's failure details unavailable through the advertised readback path. Suggested improvement: issue a readback capability/reference for Bash spills usable by Read/Grep, or ensure the Workdir attachment routes the declared readable spill directory consistently. Include the permitted tool/attachment in the spill message if it differs from the repository attachment. + +The first whole-crate test also reached the provider's 600-second timeout while an outdated WebSocket fixture awaited a subscription that was rejected by the new operation-time authentication. After providing genuine fixture credentials, the targeted subscription test completed; subsequent whole-crate runs use bounded test concurrency. A timeout was treated as incomplete validation, not a passing suite. diff --git a/docs/report/2026-10-06-t702-stale-checker-validation.md b/docs/report/2026-10-06-t702-stale-checker-validation.md new file mode 100644 index 000000000..c70e53c12 --- /dev/null +++ b/docs/report/2026-10-06-t702-stale-checker-validation.md @@ -0,0 +1,28 @@ +# T-702 validation: intermittent stale Ticket checker assertion + +The full `yoi-workspace-server` library suite intermittently fails +`server::tests::ticket_checker_suppresses_stale_and_clean_results_and_diagnoses_invalid_output` +at its post-stale-result `assert!(execution.take_inputs().is_empty())`. The same unchanged +assertion passed in some isolated/full runs and failed in others during T-702 validation. + +To separate this from Worker operation reception, I archived original base +`b96bd110c623fafdd7e8df1b153e8c640c793abe` under the ignored `target/t702-baseline` +directory without moving/replacing refs or discarding the Ticket Workdir. A first +shared-target build failed with stale dependency-interface artifacts and was **not** +baseline evidence. I then used an isolated `CARGO_TARGET_DIR=target/t702-baseline-build` +with `CARGO_BUILD_JOBS=3`. The baseline compiled successfully; its isolated checker test +passed. Its full library run with `--test-threads=4` reproduced the same assertion: +**599 passed, 1 failed, 600 total** (79.19s), at baseline `server.rs:44331`. +The post-review T-702 fix's full library run reported **611 passed, 1 failed, 612 total** +(79.01s), at the corresponding `server.rs:44266` assertion. This demonstrates an inherited +intermittent validation failure, not a green suite or proof of its root cause. +A subsequent post-review-fix whole-crate run, with the same implementation and no skips, +passed **612 library + 11 binary tests and doc tests** (library 81.24s). That successful +run does not erase the failed runs or resolve the underlying checker race. + +The Worker reception fix's 12 operation-boundary tests, 5 WorkerRemove tests (including +constructor-installed embedded dispatcher, target-bound proof, resource-key normalization, +running/pinned refusals, retention and catalog removal), root `cargo check`, formatting and +diff checks passed. The checker implementation/test was not changed or skipped to conceal +the failure. Follow-up should instrument competing checker-job dispatch and deterministic +execution-input consumption so this assertion has an explicit synchronization boundary.