diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index fb13d4fd8..91c905ab6 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -638,28 +638,13 @@ pub struct RemainingRuntimeRoute { pub const REMAINING_RUNTIME_ROUTES: &[RemainingRuntimeRoute] = &[ RemainingRuntimeRoute { method: "POST", - path: "/v1/bootstrap", - reason: "bootstrap lifecycle", - }, - RemainingRuntimeRoute { - method: "GET", - path: "/v1/protocol/ws", - reason: "WebSocket protocol transport", - }, - RemainingRuntimeRoute { - method: "GET", - path: "/v1/workers/{worker_id}/protocol/ws", - reason: "WebSocket protocol transport", + path: "/v1/workspace-runtime-verification/challenge", + reason: "Workspace verification handshake", }, RemainingRuntimeRoute { method: "POST", - path: "/v1/workers/{worker_id}/attachments", - reason: "binary attachment transport", - }, - RemainingRuntimeRoute { - method: "DELETE", - path: "/v1/workers/{worker_id}/attachments/{artifact_id}", - reason: "binary attachment transport", + path: "/v1/workspace-runtime-verification/acknowledgement", + reason: "Workspace verification handshake", }, RemainingRuntimeRoute { method: "GET", @@ -691,6 +676,56 @@ pub const REMAINING_RUNTIME_ROUTES: &[RemainingRuntimeRoute] = &[ path: "/v1/working-directories", reason: "Workdir transport", }, + RemainingRuntimeRoute { + method: "POST", + path: "/v1/working-directories/repository-access", + reason: "Workdir transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: "/v1/repositories/ssh/probe", + reason: "Repository SSH transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: "/v1/repository-refs/observe", + reason: "Repository observation transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: "/v1/working-directories/{working_directory_id}/sessions", + reason: "Workdir session transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: "/v1/workdir-sessions/{session_id}/operations", + reason: "Workdir session transport", + }, + RemainingRuntimeRoute { + method: "DELETE", + path: "/v1/workdir-sessions/{session_id}", + reason: "Workdir session transport", + }, + RemainingRuntimeRoute { + method: "GET", + path: "/v1/working-directories/{working_directory_id}", + reason: "Workdir transport", + }, + RemainingRuntimeRoute { + method: "DELETE", + path: "/v1/working-directories/{working_directory_id}", + reason: "Workdir transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: "/v1/workers/{worker_id}/attachments", + reason: "Binary attachment transport", + }, + RemainingRuntimeRoute { + method: "DELETE", + path: "/v1/workers/{worker_id}/attachments/{artifact_id}", + reason: "Binary attachment transport", + }, ]; #[cfg(test)] @@ -866,6 +901,7 @@ mod tests { #[test] fn remaining_route_inventory_does_not_overlap_contract() { + assert_eq!(REMAINING_RUNTIME_ROUTES.len(), 18); for remaining in REMAINING_RUNTIME_ROUTES { assert!(!RuntimeApiMetadata::OPERATIONS.iter().any(|operation| { format!("{:?}", operation.method).eq_ignore_ascii_case(remaining.method) diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index c40f01a18..c6f9d8af0 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -348,11 +348,13 @@ struct RuntimeHttpConfigBundleAvailabilityQuery { digest: String, } +#[allow(dead_code)] #[derive(Clone, Debug, Default, Deserialize)] struct RuntimeHttpWorkersQuery { status: Option, } +#[allow(dead_code)] #[derive(Clone, Copy, Debug, Deserialize)] #[serde(rename_all = "snake_case")] enum RuntimeHttpWorkerStatusFilter { @@ -646,6 +648,7 @@ async fn post_workspace_verification_acknowledgement( })) } +#[allow(dead_code)] async fn get_runtime_ping( State(state): State, Extension(auth): Extension, @@ -688,6 +691,7 @@ async fn get_runtime_ping( })) } +#[allow(dead_code)] async fn get_runtime( State(state): State, ) -> RestResult { @@ -767,6 +771,7 @@ async fn check_config_bundle( })) } +#[allow(dead_code)] async fn list_workers( State(state): State, auth: Option>, @@ -1107,6 +1112,7 @@ async fn cleanup_working_directory( })) } +#[allow(dead_code)] async fn get_worker( State(state): State, auth: Option>, @@ -1121,6 +1127,7 @@ async fn get_worker( Ok(Json(RuntimeHttpWorkerResponse { worker })) } +#[allow(dead_code)] async fn delete_worker( State(state): State, auth: Option>, @@ -1135,6 +1142,7 @@ async fn delete_worker( Ok(Json(RuntimeHttpWorkerDeleteResponse { worker })) } +#[allow(dead_code)] async fn create_worker( State(state): State, auth: Option>, @@ -1149,6 +1157,7 @@ async fn create_worker( Ok(Json(RuntimeHttpWorkerResponse { worker })) } +#[allow(dead_code)] async fn replace_worker_workspace_api( State(state): State, auth: Option>, @@ -1171,6 +1180,7 @@ async fn replace_worker_workspace_api( Ok(Json(RuntimeHttpWorkerResponse { worker })) } +#[allow(dead_code)] async fn restore_worker( State(state): State, auth: Option>, @@ -1734,6 +1744,7 @@ fn protocol_error_event(message: impl Into) -> protocol::Event { } } +#[allow(dead_code)] async fn worker_retention_inventory( State(state): State, auth: Option>, @@ -1754,6 +1765,7 @@ async fn worker_retention_inventory( .map_err(RuntimeHttpRestError::runtime) } +#[allow(dead_code)] async fn execute_worker_retention( State(state): State, auth: Option>, @@ -1789,6 +1801,7 @@ async fn execute_worker_retention( .map_err(RuntimeHttpRestError::runtime) } +#[allow(dead_code)] async fn send_worker_input( State(state): State, auth: Option>, @@ -1805,6 +1818,7 @@ async fn send_worker_input( Ok(Json(RuntimeHttpWorkerInputResponse { ack })) } +#[allow(dead_code)] async fn worker_completions( State(state): State, auth: Option>, @@ -1941,6 +1955,7 @@ async fn delete_worker_uploaded_file( })) } +#[allow(dead_code)] async fn stop_worker( State(state): State, auth: Option>, @@ -1959,6 +1974,7 @@ async fn stop_worker( Ok(Json(RuntimeHttpWorkerLifecycleResponse { ack })) } +#[allow(dead_code)] async fn cancel_worker( State(state): State, auth: Option>, @@ -3252,6 +3268,22 @@ mod tests { let token = "local-token"; let app = runtime_http_router(runtime.clone(), token.to_string()); + let malformed_response = app + .clone() + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/v1/workers") + .header(header::AUTHORIZATION, format!("Bearer {token}")) + .body(Body::from("{")) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(malformed_response.status(), StatusCode::BAD_REQUEST); + let malformed: RuntimeHttpErrorResponse = read_json(malformed_response).await; + assert_eq!(malformed.error.code, "invalid_json"); + let response = authed_json_request( app.clone(), Method::POST, diff --git a/crates/worker-runtime/src/http_server/runtime_management_api.rs b/crates/worker-runtime/src/http_server/runtime_management_api.rs index 542e3700f..969a2119f 100644 --- a/crates/worker-runtime/src/http_server/runtime_management_api.rs +++ b/crates/worker-runtime/src/http_server/runtime_management_api.rs @@ -50,7 +50,17 @@ async fn runtime_api_wire_compatibility( mut request: axum::extract::Request, next: Next, ) -> Response { - if !matches!(*request.method(), Method::GET | Method::HEAD) { + let is_json_request = !matches!(*request.method(), Method::GET | Method::HEAD); + let is_query_request = request.method() == Method::GET; + if request.uri().path() == "/v1/ping" && !request.headers().contains_key("x-yoi-workspace-id") { + return RuntimeHttpRestError::new( + StatusCode::FORBIDDEN, + "runtime_ping_workspace_scope_required", + "Runtime ping requires an authenticated Workspace scope", + ) + .into_response(); + } + if is_json_request { if !request.headers().contains_key(header::CONTENT_TYPE) { request.headers_mut().insert( header::CONTENT_TYPE, @@ -73,11 +83,28 @@ async fn runtime_api_wire_compatibility( request = axum::extract::Request::from_parts(parts, Body::from(body)); } let response = next.run(request).await; - if response.status() == StatusCode::UNPROCESSABLE_ENTITY { + let is_json_error = response + .headers() + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value.starts_with("application/json")); + if matches!( + response.status(), + StatusCode::BAD_REQUEST | StatusCode::UNPROCESSABLE_ENTITY + ) && !is_json_error + { return RuntimeHttpRestError::new( StatusCode::BAD_REQUEST, - "invalid_json", - "request body must be valid JSON", + if is_query_request { + "invalid_query" + } else { + "invalid_json" + }, + if is_query_request { + "request query parameters are invalid" + } else { + "request body must be valid JSON" + }, ) .into_response(); } @@ -125,79 +152,117 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { &self, workspace_id: String, ) -> Result { - let mut headers = HeaderMap::new(); - headers.insert( - "x-yoi-workspace-id", - workspace_id.parse().map_err(|error| { + let Extension(auth) = required_auth()?; + if workspace_id != auth.workspace_id { + return Err(runtime_api::RuntimeApiError::new( + StatusCode::FORBIDDEN.as_u16(), + "runtime_ping_workspace_scope_mismatch", + "Runtime ping Workspace scope does not match the authenticated capability", + )); + } + let runtime_id = self + .state + .workspace_auth + .as_ref() + .map(|auth| auth.signer.runtime_id().trim()) + .filter(|runtime_id| !runtime_id.is_empty()) + .ok_or_else(|| { runtime_api::RuntimeApiError::new( - StatusCode::BAD_REQUEST.as_u16(), - "invalid_workspace_header", - format!("invalid Workspace ping header: {error}"), + StatusCode::SERVICE_UNAVAILABLE.as_u16(), + "runtime_ping_identity_unavailable", + "Runtime ping identity is not configured", ) - })?, - ); - let Json(value) = get_runtime_ping(State(self.state.clone()), required_auth()?, headers) - .await - .map_err(api_error)?; - response(value) + })?; + Ok(runtime_api::RuntimePingResponse { + runtime_id: runtime_id.to_string(), + protocol_version: RUNTIME_HTTP_PROTOCOL_VERSION, + }) } async fn runtime_summary( &self, ) -> Result { - let Json(value) = get_runtime(State(self.state.clone())) - .await + let runtime = self + .state + .runtime + .summary() + .map_err(RuntimeHttpRestError::runtime) .map_err(api_error)?; - response(value) + let runtime = response(runtime)?; + Ok(runtime_api::RuntimeSummaryResponse { runtime }) } async fn list_workers( &self, query: runtime_api::WorkerListQuery, ) -> Result { - let Json(value) = list_workers( - State(self.state.clone()), - auth_extension(), - Ok(Query(request(query)?)), - ) - .await + let scope = + auth_workspace_scope(&self.state, auth_extension().as_ref()).map_err(api_error)?; + let workers = match (query.status, scope.as_ref()) { + (Some(runtime_api::WorkerStatusFilter::Stopped), Some(scope)) => { + self.state.runtime.list_stopped_workers_scoped(scope) + } + (Some(runtime_api::WorkerStatusFilter::Stopped), None) => { + self.state.runtime.list_stopped_workers() + } + (None, Some(scope)) => self.state.runtime.list_workers_scoped(scope), + (None, None) => self.state.runtime.list_workers(), + } + .map_err(RuntimeHttpRestError::runtime) .map_err(api_error)?; - response(value) + let workers = response(workers)?; + Ok(runtime_api::WorkersResponse { workers }) } async fn get_worker( &self, worker_id: String, ) -> Result { - let Json(value) = get_worker(State(self.state.clone()), auth_extension(), Path(worker_id)) - .await - .map_err(api_error)?; - response(value) + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => self.state.runtime.worker_detail_scoped(&scope, &worker_ref), + None => self.state.runtime.worker_detail(&worker_ref), + } + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error)?; + let worker = response(worker)?; + Ok(runtime_api::WorkerResponse { worker }) } async fn create_worker( &self, value: runtime_api::CreateWorkerRequest, ) -> Result { - let Json(value) = create_worker( - State(self.state.clone()), - auth_extension(), - Ok(Json(request(value)?)), - ) - .await + let request: CreateWorkerRequest = request(value)?; + let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => self.state.runtime.create_worker_scoped(&scope, request), + None => self.state.runtime.create_worker(request), + } + .map_err(RuntimeHttpRestError::runtime) .map_err(api_error)?; - response(value) + let worker = response(worker)?; + Ok(runtime_api::WorkerResponse { worker }) } async fn delete_worker( &self, worker_id: String, ) -> Result { - let Json(value) = - delete_worker(State(self.state.clone()), auth_extension(), Path(worker_id)) - .await - .map_err(api_error)?; - response(value) + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => self.state.runtime.delete_worker_scoped(&scope, &worker_ref), + None => self.state.runtime.delete_worker(&worker_ref), + } + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error)?; + let worker = response(worker)?; + Ok(runtime_api::WorkerDeleteResponse { worker }) } async fn send_worker_input( @@ -205,15 +270,21 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { worker_id: String, input: runtime_api::WorkerInput, ) -> Result { - let Json(value) = send_worker_input( - State(self.state.clone()), - auth_extension(), - Path(worker_id), - Ok(Json(request(input)?)), - ) - .await + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let input: WorkerInput = request(input)?; + let ack = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => self + .state + .runtime + .send_input_scoped(&scope, &worker_ref, input), + None => self.state.runtime.send_input(&worker_ref, input), + } + .map_err(RuntimeHttpRestError::runtime) .map_err(api_error)?; - response(value) + let ack = response(ack)?; + Ok(runtime_api::WorkerInputResponse { ack }) } async fn stop_worker( @@ -221,23 +292,20 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { worker_id: String, value: runtime_api::WorkerLifecycleRequest, ) -> Result { - let body = serde_json::to_vec(&request::<_, RuntimeHttpWorkerLifecycleRequest>(value)?) - .map_err(|error| { - runtime_api::RuntimeApiError::new( - 500, - "runtime_request_conversion_failed", - error.to_string(), - ) - })?; - let Json(value) = stop_worker( - State(self.state.clone()), - auth_extension(), - Path(worker_id), - Bytes::from(body), - ) - .await + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let ack = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => self + .state + .runtime + .stop_worker_scoped(&scope, &worker_ref, value.reason), + None => self.state.runtime.stop_worker(&worker_ref, value.reason), + } + .map_err(RuntimeHttpRestError::runtime) .map_err(api_error)?; - response(value) + let ack = response(ack)?; + Ok(runtime_api::WorkerLifecycleResponse { ack }) } async fn cancel_worker( @@ -245,23 +313,21 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { worker_id: String, value: runtime_api::WorkerLifecycleRequest, ) -> Result { - let body = serde_json::to_vec(&request::<_, RuntimeHttpWorkerLifecycleRequest>(value)?) - .map_err(|error| { - runtime_api::RuntimeApiError::new( - 500, - "runtime_request_conversion_failed", - error.to_string(), - ) - })?; - let Json(value) = cancel_worker( - State(self.state.clone()), - auth_extension(), - Path(worker_id), - Bytes::from(body), - ) - .await + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let ack = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => { + self.state + .runtime + .cancel_worker_scoped(&scope, &worker_ref, value.reason) + } + None => self.state.runtime.cancel_worker(&worker_ref, value.reason), + } + .map_err(RuntimeHttpRestError::runtime) .map_err(api_error)?; - response(value) + let ack = response(ack)?; + Ok(runtime_api::WorkerLifecycleResponse { ack }) } async fn restore_worker( @@ -269,11 +335,20 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { worker_id: String, _request: runtime_api::EmptyObjectRequest, ) -> Result { - let Json(value) = - restore_worker(State(self.state.clone()), auth_extension(), Path(worker_id)) - .await - .map_err(api_error)?; - response(value) + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => self + .state + .runtime + .restore_worker_scoped(&scope, &worker_ref), + None => self.state.runtime.restore_worker(&worker_ref), + } + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error)?; + let worker = response(worker)?; + Ok(runtime_api::WorkerResponse { worker }) } async fn replace_worker_workspace_api( @@ -281,15 +356,25 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { worker_id: String, value: runtime_api::WorkerWorkspaceApiRequest, ) -> Result { - let Json(value) = replace_worker_workspace_api( - State(self.state.clone()), - auth_extension(), - Path(worker_id), - Ok(Json(request(value)?)), - ) - .await + let workspace_api: WorkspaceApiRef = request(value.workspace_api)?; + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => self.state.runtime.replace_worker_workspace_api_scoped( + &scope, + &worker_ref, + workspace_api, + ), + None => self + .state + .runtime + .replace_worker_workspace_api(&worker_ref, workspace_api), + } + .map_err(RuntimeHttpRestError::runtime) .map_err(api_error)?; - response(value) + let worker = response(worker)?; + Ok(runtime_api::WorkerResponse { worker }) } async fn complete_worker_arguments( @@ -297,29 +382,53 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { worker_id: String, value: runtime_api::CompletionRequest, ) -> Result { - let Json(value) = worker_completions( - State(self.state.clone()), - auth_extension(), - Path(worker_id), - Ok(Json(request(value)?)), - ) - .await + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let kind = value.kind; + let prefix = value.prefix; + let entries = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => { + self.state + .runtime + .worker_completions_scoped(&scope, &worker_ref, kind, &prefix) + } + None => self + .state + .runtime + .worker_completions(&worker_ref, kind, &prefix), + } + .map_err(RuntimeHttpRestError::runtime) .map_err(api_error)?; - response(value) + let entries = response(entries)?; + Ok(runtime_api::CompletionResponse { + kind, + prefix, + entries, + }) } async fn retention_inventory( &self, worker_id: String, ) -> Result { - let Json(value) = worker_retention_inventory( - State(self.state.clone()), - auth_extension(), - Path(worker_id), - ) - .await - .map_err(api_error)?; - response(value) + let scope = auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + .ok_or_else(|| { + runtime_api::RuntimeApiError::new( + StatusCode::FORBIDDEN.as_u16(), + "workspace_scope_required", + "Worker retention inventory requires workspace-scoped authorization", + ) + })?; + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let inventory = self + .state + .runtime + .worker_retention_inventory(&scope.workspace_id, &worker_ref) + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error)?; + response(inventory) } async fn execute_retention( @@ -327,14 +436,37 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { worker_id: String, value: runtime_api::WorkerRetentionExecutionRequest, ) -> Result { - let Json(value) = execute_worker_retention( - State(self.state.clone()), - auth_extension(), - Path(worker_id), - Ok(Json(request(value)?)), - ) - .await - .map_err(api_error)?; - response(value) + let scope = auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + .ok_or_else(|| { + runtime_api::RuntimeApiError::new( + StatusCode::FORBIDDEN.as_u16(), + "workspace_scope_required", + "Worker retention execution requires workspace-scoped authorization", + ) + })?; + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let request: WorkerRetentionExecutionRequest = request(value)?; + if request.workspace_id != scope.workspace_id { + return Err(runtime_api::RuntimeApiError::new( + StatusCode::FORBIDDEN.as_u16(), + "workspace_scope_mismatch", + "Worker retention request workspace does not match the authenticated workspace scope", + )); + } + if request.worker_id != worker_ref.worker_id { + return Err(runtime_api::RuntimeApiError::new( + StatusCode::BAD_REQUEST.as_u16(), + "worker_id_mismatch", + "Worker retention request worker_id does not match route worker_id", + )); + } + let result = self + .state + .runtime + .execute_worker_retention(&request) + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error)?; + response(result) } } diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index a509e5a59..de36c1606 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -46,10 +46,7 @@ use worker_runtime::http_server::{ RUNTIME_PING_PERMISSION, RUNTIME_WORKSPACE_SCOPE_HEADER, RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest, RuntimeHttpErrorResponse, RuntimeHttpPingResponse, RuntimeHttpRepositoryAccessResponse, - RuntimeHttpSummaryResponse, RuntimeHttpUploadedFileDeleteResponse, - RuntimeHttpUploadedFileResponse, RuntimeHttpWorkerCompletionsResponse, - RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, - RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse, + RuntimeHttpUploadedFileDeleteResponse, RuntimeHttpUploadedFileResponse, RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse, RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse, }; @@ -3981,7 +3978,7 @@ impl RemoteWorkerRuntime { fn lifecycle_result_from_response( &self, worker_id: &str, - response: RuntimeHttpWorkerLifecycleResponse, + response: worker_runtime::catalog::WorkerLifecycleAck, ) -> WorkerLifecycleResult { WorkerLifecycleResult { state: WorkerOperationState::Accepted, @@ -3991,7 +3988,7 @@ impl RemoteWorkerRuntime { DiagnosticSeverity::Info, format!( "Remote Runtime acknowledged lifecycle operation for '{worker_id}' with status {}", - embedded_worker_status_label(response.ack.status) + embedded_worker_status_label(response.status) ), )], } @@ -4010,25 +4007,27 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { MAX_REMOTE_RUNTIME_RESPONSE_BYTES, |client| async move { client.runtime_summary().await }, ) - .and_then(|value| runtime_contract_convert::<_, RuntimeHttpSummaryResponse>(value)) - { + .and_then(|value| { + runtime_contract_convert::<_, worker_runtime::management::RuntimeSummary>( + value.runtime, + ) + }) { Ok(response) => RuntimeSummary { runtime_id: self.runtime_id.clone(), label: response - .runtime .display_name .unwrap_or_else(|| self.display_name.clone()), kind: "remote_worker_runtime".to_string(), - status: embedded_runtime_status_label(response.runtime.status).to_string(), + status: embedded_runtime_status_label(response.status).to_string(), source: RuntimeSourceSummary::remote_http(), host_ids: if limit == 0 { Vec::new() } else { vec![self.host_id.clone()] }, - worker_creation_available: response.runtime.worker_creation_available, - os: response.runtime.os, - arch: response.runtime.arch, + worker_creation_available: response.worker_creation_available, + os: response.os, + arch: response.arch, diagnostics: Vec::new(), }, Err(diagnostic) => RuntimeSummary { @@ -4162,11 +4161,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { .await }, ) - .and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkersResponse>(value)) - { + .and_then(|value| { + runtime_contract_convert::<_, Vec>( + value.workers, + ) + }) { Ok(response) => RuntimeList::new( response - .workers .into_iter() .take(limit) .map(|worker| self.map_worker_summary(worker)) @@ -4193,11 +4194,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { .await }, ) - .and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkersResponse>(value)) - { + .and_then(|value| { + runtime_contract_convert::<_, Vec>( + value.workers, + ) + }) { Ok(response) => RuntimeList::new( response - .workers .into_iter() .take(limit) .map(|worker| self.map_worker_summary(worker)) @@ -4216,10 +4219,11 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { MAX_REMOTE_RUNTIME_RESPONSE_BYTES, move |client| async move { client.get_worker(worker_id_owned).await }, ) - .and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerResponse>(value)) - { + .and_then(|value| { + runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker) + }) { Ok(response) => WorkerLookupResult { - worker: Some(self.map_worker_detail(response.worker)), + worker: Some(self.map_worker_detail(response)), diagnostics: Vec::new(), }, Err(diagnostic) if diagnostic.code == "worker_not_found" => WorkerLookupResult { @@ -4245,11 +4249,12 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { .await }, ) - .and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerResponse>(value)) - { + .and_then(|value| { + runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker) + }) { Ok(response) => WorkerRestoreResult { state: WorkerOperationState::Accepted, - worker: Some(self.map_worker_detail(response.worker)), + worker: Some(self.map_worker_detail(response)), diagnostics: Vec::new(), }, Err(diagnostic) => WorkerRestoreResult { @@ -4265,10 +4270,8 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { worker_id: &str, workspace_api: WorkspaceApiRef, ) -> WorkerWorkspaceApiResult { - let request = match runtime_contract_convert(RuntimeHttpWorkerWorkspaceApiRequest { - workspace_api, - }) { - Ok(request) => request, + let request = match runtime_contract_convert(workspace_api) { + Ok(workspace_api) => runtime_api::WorkerWorkspaceApiRequest { workspace_api }, Err(diagnostic) => { return WorkerWorkspaceApiResult { state: WorkerOperationState::Rejected, @@ -4288,11 +4291,12 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { .await }, ) - .and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerResponse>(value)) - { + .and_then(|value| { + runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker) + }) { Ok(response) => WorkerWorkspaceApiResult { state: WorkerOperationState::Accepted, - worker: Some(self.map_worker_detail(response.worker)), + worker: Some(self.map_worker_detail(response)), diagnostics: Vec::new(), }, Err(diagnostic) => WorkerWorkspaceApiResult { @@ -4506,11 +4510,12 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { MAX_REMOTE_RUNTIME_RESPONSE_BYTES, move |client| async move { client.create_worker(create).await }, ) - .and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerResponse>(value)) - { + .and_then(|value| { + runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker) + }) { Ok(response) => WorkerSpawnResult { state: WorkerOperationState::Accepted, - worker: Some(self.map_worker_detail(response.worker)), + worker: Some(self.map_worker_detail(response)), acceptance_evidence: vec![WorkerSpawnAcceptanceEvidence { kind: "remote_runtime_worker_created".to_string(), detail: "worker-runtime REST create endpoint accepted the Worker".to_string(), @@ -4589,7 +4594,9 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { move |client| async move { client.stop_worker(worker_id_owned, body).await }, ) .and_then(|value| { - runtime_contract_convert::<_, RuntimeHttpWorkerLifecycleResponse>(value) + runtime_contract_convert::<_, worker_runtime::catalog::WorkerLifecycleAck>( + value.ack, + ) }) { Ok(response) => self.lifecycle_result_from_response(worker_id, response), Err(diagnostic) => remote_lifecycle_rejected(&self.runtime_id, worker_id, diagnostic), @@ -4612,7 +4619,9 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { move |client| async move { client.cancel_worker(worker_id_owned, body).await }, ) .and_then(|value| { - runtime_contract_convert::<_, RuntimeHttpWorkerLifecycleResponse>(value) + runtime_contract_convert::<_, worker_runtime::catalog::WorkerLifecycleAck>( + value.ack, + ) }) { Ok(response) => self.lifecycle_result_from_response(worker_id, response), Err(diagnostic) => remote_lifecycle_rejected(&self.runtime_id, worker_id, diagnostic), @@ -4627,15 +4636,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { MAX_REMOTE_RUNTIME_RESPONSE_BYTES, move |client| async move { client.delete_worker(worker_id_owned).await }, ) - .and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerDeleteResponse>(value)) - { + .and_then(|value| { + runtime_contract_convert::<_, worker_runtime::management::WorkerDeleteResult>( + value.worker, + ) + }) { Ok(response) => WorkerDeleteResult { state: WorkerOperationState::Accepted, worker: RuntimeWorkerRef::new( self.runtime_id.clone(), - response.worker.worker_id.to_string(), + response.worker_id.to_string(), ), - deleted: response.worker.deleted, + deleted: response.deleted, diagnostics: Vec::new(), }, Err(diagnostic) => WorkerDeleteResult { @@ -4799,19 +4811,15 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { let failure_kind = request.kind; let failure_prefix = request.prefix.clone(); let worker_id_owned = worker_id.to_string(); - match self - .run_runtime_api( - self.request_timeout, - MAX_REMOTE_RUNTIME_RESPONSE_BYTES, - move |client| async move { - client - .complete_worker_arguments(worker_id_owned, request) - .await - }, - ) - .and_then(|value| { - runtime_contract_convert::<_, RuntimeHttpWorkerCompletionsResponse>(value) - }) { + match self.run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { + client + .complete_worker_arguments(worker_id_owned, request) + .await + }, + ) { Ok(response) => WorkerCompletionsResult { worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), kind: response.kind,