diff --git a/Cargo.lock b/Cargo.lock index 961baa889..5156ef43c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3706,6 +3706,19 @@ dependencies = [ "zeroize", ] +[[package]] +name = "runtime-api" +version = "0.1.0" +dependencies = [ + "api-macros", + "manifest", + "protocol", + "serde", + "tokio", + "workdir", + "workspace-api", +] + [[package]] name = "rusqlite" version = "0.37.0" @@ -6126,6 +6139,7 @@ dependencies = [ "protocol", "reqwest", "ring", + "runtime-api", "serde", "serde_json", "serial_test", @@ -6272,6 +6286,7 @@ dependencies = [ "protocol", "reqwest", "ring", + "runtime-api", "rusqlite", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index c7bd87a64..49e82d425 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ members = [ "crates/mcp", "crates/worker", "crates/worker-runtime", + "crates/runtime-api", "crates/plugin-pdk", "crates/yoi", "crates/protocol", @@ -47,6 +48,7 @@ default-members = [ "crates/mcp", "crates/worker", "crates/worker-runtime", + "crates/runtime-api", "crates/plugin-pdk", "crates/yoi", "crates/protocol", @@ -101,6 +103,7 @@ session-metrics = { path = "crates/session-metrics" } session-analytics = { path = "crates/session-analytics" } session-store = { path = "crates/session-store" } standalone = { path = "crates/standalone" } +runtime-api = { path = "crates/runtime-api" } secrets = { path = "crates/secrets" } tools = { path = "crates/tools" } config-source = { path = "crates/config-source" } diff --git a/crates/api-macros-impl/src/lib.rs b/crates/api-macros-impl/src/lib.rs index de30af0ae..061c690a8 100644 --- a/crates/api-macros-impl/src/lib.rs +++ b/crates/api-macros-impl/src/lib.rs @@ -949,18 +949,23 @@ fn reqwest_adapter_tokens( .map_err(#api_crate::reqwest::ClientError::from); } }; - let public_error = match (operation.error_status, operation.error_body.as_ref()) { - (Some(status), Some(error)) => quote! { - if __actual == #status { + let public_error = match operation.error_body.as_ref() { + Some(error) => quote! { + if (400..=599).contains(&__actual) { + let __status = __response.status(); let __error = __response .decode_json::<#error>(#api_crate::reqwest::DecodeKind::PublicError) - .map_err(#api_crate::reqwest::ClientError::from)?; + .map_err(|_| #api_crate::reqwest::ClientError::failure( + #api_crate::reqwest::ClientFailure::ErrorResponseDecode { + status: __status.as_u16(), + }, + ))?; return ::core::result::Result::Err( - #api_crate::reqwest::ClientError::public(__response.status(), __error), + #api_crate::reqwest::ClientError::public(__status, __error), ); } }, - _ => quote!(), + None => quote!(), }; quote! { pub async fn #method_ident( @@ -1192,7 +1197,10 @@ fn axum_adapter_tokens( let invocation = quote!(#trait_ident::#method_ident(&*__service, #(#call_arguments),*).await); let mapped = if operation.fallible { let error = match operation.error_status { - Some(status) => quote!(#api_crate::axum::json_response(#api_crate::axum::status(#status), error)), + Some(_) => quote!({ + let status = #api_crate::HttpError::status_code(&error); + #api_crate::axum::json_response(#api_crate::axum::status(status), error) + }), None => quote!({ let _ = error; #api_crate::axum::empty_response(#api_crate::axum::framework::StatusCode::INTERNAL_SERVER_ERROR) }), }; quote! { diff --git a/crates/api-macros/src/lib.rs b/crates/api-macros/src/lib.rs index 2826fbf51..79ef65b92 100644 --- a/crates/api-macros/src/lib.rs +++ b/crates/api-macros/src/lib.rs @@ -171,6 +171,15 @@ pub trait Operation { const METADATA: OperationMetadata; } +/// Public JSON error that can select its concrete HTTP status at the server boundary. +/// +/// `error_status` remains the contract's default/documented status. Implementations may return a +/// different 4xx/5xx status for one named error shape, allowing a generated client to preserve a +/// service's typed 401/403/404/409 responses without introducing duplicate route handlers. +pub trait HttpError { + fn status_code(&self) -> u16; +} + /// Marker used when an operation has no request, response, or public error body. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum NoBody {} diff --git a/crates/api-macros/src/reqwest.rs b/crates/api-macros/src/reqwest.rs index b45406c9c..dd54b58d6 100644 --- a/crates/api-macros/src/reqwest.rs +++ b/crates/api-macros/src/reqwest.rs @@ -352,6 +352,8 @@ impl ClientCore { let response = request.send().await.map_err(|error| { if error.is_timeout() { ClientFailure::Timeout + } else if error.is_connect() { + ClientFailure::Connect } else { ClientFailure::Transport } @@ -461,10 +463,12 @@ pub enum DecodeKind { pub enum ClientFailure { Authorization, RequestEncoding, + Connect, Transport, Timeout, ResponseTooLarge { limit: usize }, UnexpectedStatus { expected: u16, actual: u16 }, + ErrorResponseDecode { status: u16 }, Decode { kind: DecodeKind }, UnexpectedBody, } @@ -480,6 +484,7 @@ impl fmt::Display for ClientFailure { match self { Self::Authorization => formatter.write_str("request authorization failed"), Self::RequestEncoding => formatter.write_str("request encoding failed"), + Self::Connect => formatter.write_str("HTTP connection failed"), Self::Transport => formatter.write_str("HTTP transport failed"), Self::Timeout => formatter.write_str("HTTP request timed out"), Self::ResponseTooLarge { limit } => { @@ -491,6 +496,9 @@ impl fmt::Display for ClientFailure { "expected HTTP status {expected}, received {actual}" ) } + Self::ErrorResponseDecode { status } => { + write!(formatter, "HTTP {status} error response JSON was malformed") + } Self::Decode { kind: DecodeKind::Success, } => formatter.write_str("success response JSON was malformed"), diff --git a/crates/api-macros/tests/http_adapters.rs b/crates/api-macros/tests/http_adapters.rs index 542d2122d..efc9d3a68 100644 --- a/crates/api-macros/tests/http_adapters.rs +++ b/crates/api-macros/tests/http_adapters.rs @@ -28,6 +28,14 @@ pub struct Widget { #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct PublicError { message: String, + #[serde(skip)] + status: u16, +} + +impl api_macros::HttpError for PublicError { + fn status_code(&self) -> u16 { + self.status + } } #[api(reqwest, axum)] @@ -65,6 +73,7 @@ impl WidgetApi for WidgetService { if request.name == "bad" { return Err(PublicError { message: "sensitive-public-detail".to_owned(), + status: 409, }); } Ok(Widget { @@ -210,10 +219,16 @@ async fn generated_client_and_router_follow_the_normalized_contract() { .await .expect_err("typed public error"); assert!(!format!("{public_error:?}").contains("sensitive-public-detail")); + assert!(matches!( + &public_error, + api_reqwest::ClientError::Public { status, .. } + if *status == api_reqwest::framework::StatusCode::CONFLICT + )); assert_eq!( public_error.into_public_error(), Some(PublicError { message: "sensitive-public-detail".to_owned(), + status: 0, }) ); diff --git a/crates/runtime-api/Cargo.toml b/crates/runtime-api/Cargo.toml new file mode 100644 index 000000000..404355a17 --- /dev/null +++ b/crates/runtime-api/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "runtime-api" +description = "Typed JSON REST contract for Yoi Runtime Worker management" +version = "0.1.0" +edition.workspace = true +license.workspace = true + +[dependencies] +api-macros = { workspace = true, features = ["reqwest", "axum"] } +manifest.workspace = true +protocol.workspace = true +serde = { workspace = true, features = ["derive"] } +workspace-api.workspace = true +workdir.workspace = true + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread"] } diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs new file mode 100644 index 000000000..9bcac8a4d --- /dev/null +++ b/crates/runtime-api/src/lib.rs @@ -0,0 +1,960 @@ +//! Typed Worker Runtime management HTTP contract. +//! +//! This crate owns the public REST DTOs and generated client/server adapters only. Runtime domain +//! state, authorization, body limits, and operation permission policy remain in their existing +//! authorities. + +use std::collections::BTreeMap; + +use api_macros::api; +pub use api_macros::axum as server_support; +pub use api_macros::reqwest as client_support; +pub use api_macros::{ApiContract, HttpMethod}; +use protocol::{CompletionEntry, CompletionKind, Segment, WorkerId, WorkerStateSnapshot}; +use serde::{Deserialize, Serialize}; +use workdir::workspace::{MaterializerKind, RuntimeWorkerRef, RuntimeWorkingDirectorySummary}; +use workspace_api::{RepositoryAccessMode, RepositorySource}; + +pub const RUNTIME_API_VERSION: &str = "v1"; +pub const RUNTIME_API_BASE_PATH: &str = "/v1"; + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct RuntimeApiError { + pub error: RuntimeApiErrorDetail, + #[serde(skip)] + status: u16, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct RuntimeApiErrorDetail { + pub code: String, + pub message: String, +} + +impl RuntimeApiError { + pub fn new(status: u16, code: impl Into, message: impl Into) -> Self { + Self { + error: RuntimeApiErrorDetail { + code: code.into(), + message: message.into(), + }, + status, + } + } + + pub fn status(&self) -> u16 { + self.status + } +} + +impl api_macros::HttpError for RuntimeApiError { + fn status_code(&self) -> u16 { + self.status + } +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct RuntimePingResponse { + pub runtime_id: String, + pub protocol_version: u32, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeBackendKind { + Memory, + FsStore, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeStatus { + Running, + Stopped, +} + +fn unknown_platform_component() -> String { + "unknown".to_string() +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct RuntimeSummary { + pub display_name: Option, + pub backend: RuntimeBackendKind, + pub status: RuntimeStatus, + pub worker_count: usize, + pub active_worker_count: usize, + pub stopped_worker_count: usize, + pub diagnostic_count: usize, + #[serde(default = "unknown_platform_component")] + pub os: String, + #[serde(default = "unknown_platform_component")] + pub arch: String, + #[serde(default)] + pub worker_creation_available: bool, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct RuntimeSummaryResponse { + pub runtime: RuntimeSummary, +} + +#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)] +pub struct WorkerRef { + pub worker_id: WorkerId, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkspaceApiRef { + pub workspace_id: String, + pub base_url: String, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerWorkspaceApiRequest { + pub workspace_api: WorkspaceApiRef, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkerStatus { + Idle, + Running, + Paused, + Stopped, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(tag = "kind", content = "value", rename_all = "snake_case")] +pub enum ProfileSelector { + Builtin(String), + Named(String), +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ProfileSourceGraphSummary { + pub source_count: usize, + pub total_source_bytes: u64, + pub entrypoints: BTreeMap, + pub import_count: usize, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ProfileSourceArchiveRef { + pub id: String, + pub digest: String, + pub size_bytes: u64, + pub source_graph: ProfileSourceGraphSummary, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ProfileSourceArchive { + pub reference: ProfileSourceArchiveRef, + pub content: Vec, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ProfileSourceArchiveSource { + Embedded { archive: ProfileSourceArchive }, + WorkspaceConfig { archive: ProfileSourceArchiveRef }, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ConfigBundleRef { + pub id: String, + pub digest: String, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct RepositorySelector(pub String); + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkingDirectoryRepository { + pub id: String, + pub provider: String, + pub source: RepositorySource, + pub source_revision: u64, + pub source_fingerprint: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub selector: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum BackendResourceKind { + ProfileSourceArchive, + RepositorySshAccess, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum BackendResourceOperation { + FetchArchive, + FetchOnce, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ResourceRedactionPolicy { + RuntimeInternalOnly, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct BackendResourceHandle { + pub kind: BackendResourceKind, + pub workspace_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scope_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub runtime_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_id: Option, + pub resource_id: String, + pub digest: String, + pub operation: BackendResourceOperation, + pub expires_at_unix_seconds: i64, + pub nonce: String, + pub revision: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub generation: Option, + pub max_bytes: u64, + pub content_type: String, + pub redaction: ResourceRedactionPolicy, + pub audit_correlation_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub profile_source_graph: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct RepositorySshCredentialCandidate { + pub credential_id: String, + pub credential_revision: u64, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct RepositorySshMaterializationAccess { + pub credential_candidates: Vec, + pub host_trust_id: String, + pub host_trust_revision: u64, + pub access: RepositoryAccessMode, + pub expires_at_epoch_seconds: u64, + pub repository_id: String, + pub repository_source_fingerprint: String, + pub repository_uri: String, + pub secret_resource: BackendResourceHandle, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct RepositoryMaterializationContext { + pub workspace_id: String, + pub runtime_id: String, + pub operation_id: String, + pub config_revision: u64, + pub config_projection_digest: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ssh: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkingDirectoryRequest { + pub repository: WorkingDirectoryRepository, + #[serde(default)] + pub materializer: MaterializerKind, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub backend_workdir_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub materialization: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkingDirectoryClaim { + pub working_directory_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub relative_cwd: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkingDirectoryStatus { + pub summary: RuntimeWorkingDirectorySummary, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkerInputKind { + User, + Notify, + Compact, + ListRewindTargets, + RegisterPeer, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerInput { + pub kind: WorkerInputKind, + pub content: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub submission_request_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub segments: Option>, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct CreateWorkerRequest { + pub worker_id: WorkerId, + pub create_fingerprint: String, + pub profile: ProfileSelector, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub display_name: Option, + pub profile_source: ProfileSourceArchiveSource, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub config_bundle: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub initial_input: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub working_directory_request: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub working_directory: Option, + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub worker_observation_enabled: bool, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub worker_observation_grants: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspace_api: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub memory_settings: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerSummary { + pub worker_ref: WorkerRef, + pub worker_id: WorkerId, + pub status: WorkerStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub created_at_ms: Option, + pub execution_metadata_available: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_state: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspace_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub working_directory: Option, + pub profile: ProfileSelector, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub display_name: Option, + pub profile_source: ProfileSourceArchiveRef, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub config_bundle: Option, +} + +pub type WorkerDetail = WorkerSummary; + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkersResponse { + pub workers: Vec, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerResponse { + pub worker: WorkerDetail, +} + +/// Explicit restore outcome returned for every reachable Runtime restore attempt, +/// including preflight rejection and uncertain post-side-effect reconciliation. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerRestoreResponse { + pub state: workspace_api::WorkerRestoreState, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reason_code: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub message: Option, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkerStatusFilter { + Stopped, +} + +#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerListQuery { + #[serde(skip_serializing_if = "Option::is_none")] + pub status: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerDeleteResult { + pub worker_id: WorkerId, + pub deleted: bool, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerDeleteResponse { + pub worker: WorkerDeleteResult, +} + +#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerLifecycleRequest { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reason: Option, +} + +#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] +pub struct EmptyObjectRequest {} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerLifecycleAck { + pub worker_ref: WorkerRef, + pub status: WorkerStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_state: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerLifecycleResponse { + pub ack: WorkerLifecycleAck, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerSubmissionAck { + pub submission_request_id: String, + pub submission_id: String, + pub disposition: protocol::SubmissionDisposition, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerInteractionAck { + pub worker_ref: WorkerRef, + pub status: WorkerStatus, + #[serde(skip_serializing_if = "Option::is_none")] + pub submission: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerInputResponse { + pub ack: WorkerInteractionAck, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct CompletionRequest { + pub kind: CompletionKind, + #[serde(default)] + pub prefix: String, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct CompletionResponse { + pub kind: CompletionKind, + pub prefix: String, + pub entries: Vec, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum SessionDisposition { + Archive, + Purge, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum DiagnosticsDisposition { + Purge, + Retain, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerRetentionInventory { + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: WorkerId, + pub session_id: Option, + pub segment_ids: Vec, + pub session_bytes: u64, + pub diagnostics_bytes: u64, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerRetentionExecutionRequest { + pub operation_id: String, + pub input_fingerprint: String, + pub archive_id: Option, + pub workspace_id: String, + pub source_runtime_id: String, + pub worker_id: WorkerId, + pub expected_worker_revision: String, + pub source_created_at: String, + pub removed_at: String, + pub effective_profile: Option, + pub retention_class: Option, + pub policy_id: String, + pub policy_revision: u64, + pub session_disposition: SessionDisposition, + pub diagnostics_disposition: DiagnosticsDisposition, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerSessionArchiveManifest { + pub schema_version: u32, + pub archive_id: String, + pub workspace_id: String, + pub source_runtime_id: String, + pub source_worker_id: WorkerId, + pub source_session_id: String, + pub segment_ids: Vec, + pub source_created_at: String, + pub removed_at: String, + pub archived_at_unix_seconds: u64, + pub effective_profile: Option, + pub retention_class: Option, + pub content_checksum_sha256: String, + pub content_bytes: u64, + pub content_file_count: u64, + pub policy_id: String, + pub policy_revision: u64, + pub operation_id: String, + pub input_fingerprint: String, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct WorkerRetentionExecutionResult { + pub operation_id: String, + pub input_fingerprint: String, + pub expected_worker_revision: String, + pub worker_id: WorkerId, + pub session_disposition: SessionDisposition, + pub diagnostics_disposition: DiagnosticsDisposition, + pub archive: Option, + pub source_removed: bool, + pub diagnostics_retained: bool, +} + +#[api(reqwest, axum)] +pub trait RuntimeApi { + #[get("/v1/ping", status = 200, error_status = 400)] + async fn ping( + &self, + #[header("x-yoi-workspace-id")] workspace_id: String, + ) -> Result; + + #[get("/v1/runtime", status = 200, error_status = 400)] + async fn runtime_summary(&self) -> Result; + + #[get("/v1/workers", status = 200, error_status = 400)] + async fn list_workers( + &self, + #[query] query: WorkerListQuery, + ) -> Result; + + #[get("/v1/workers/{worker_id}", status = 200, error_status = 400)] + async fn get_worker( + &self, + #[path] worker_id: String, + ) -> Result; + + #[post("/v1/workers", status = 200, error_status = 400)] + async fn create_worker( + &self, + #[body] request: CreateWorkerRequest, + ) -> Result; + + #[delete("/v1/workers/{worker_id}", status = 200, error_status = 400)] + async fn delete_worker( + &self, + #[path] worker_id: String, + ) -> Result; + + #[post("/v1/workers/{worker_id}/input", status = 200, error_status = 400)] + async fn send_worker_input( + &self, + #[path] worker_id: String, + #[body] input: WorkerInput, + ) -> Result; + + #[post("/v1/workers/{worker_id}/stop", status = 200, error_status = 400)] + async fn stop_worker( + &self, + #[path] worker_id: String, + #[body] request: WorkerLifecycleRequest, + ) -> Result; + + #[post("/v1/workers/{worker_id}/cancel", status = 200, error_status = 400)] + async fn cancel_worker( + &self, + #[path] worker_id: String, + #[body] request: WorkerLifecycleRequest, + ) -> Result; + + #[post("/v1/workers/{worker_id}/restore", status = 200, error_status = 400)] + async fn restore_worker( + &self, + #[path] worker_id: String, + #[body] request: EmptyObjectRequest, + ) -> Result; + + #[post( + "/v1/workers/{worker_id}/workspace-api", + status = 200, + error_status = 400 + )] + async fn replace_worker_workspace_api( + &self, + #[path] worker_id: String, + #[body] request: WorkerWorkspaceApiRequest, + ) -> Result; + + #[post( + "/v1/workers/{worker_id}/completions", + status = 200, + error_status = 400 + )] + async fn complete_worker_arguments( + &self, + #[path] worker_id: String, + #[body] request: CompletionRequest, + ) -> Result; + + #[get( + "/v1/workers/{worker_id}/retention/inventory", + status = 200, + error_status = 400 + )] + async fn retention_inventory( + &self, + #[path] worker_id: String, + ) -> Result; + + #[post( + "/v1/workers/{worker_id}/retention/execute", + status = 200, + error_status = 400 + )] + async fn execute_retention( + &self, + #[path] worker_id: String, + #[body] request: WorkerRetentionExecutionRequest, + ) -> Result; +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct RemainingRuntimeRoute { + pub method: &'static str, + pub path: &'static str, + pub reason: &'static str, +} + +pub const RUNTIME_ROUTE_VERIFICATION_CHALLENGE: &str = + "/v1/workspace-runtime-verification/challenge"; +pub const RUNTIME_ROUTE_VERIFICATION_ACK: &str = + "/v1/workspace-runtime-verification/acknowledgement"; +pub const RUNTIME_ROUTE_CONFIG_BUNDLES: &str = "/v1/config-bundles"; +pub const RUNTIME_ROUTE_CONFIG_BUNDLE_AVAILABILITY: &str = + "/v1/config-bundles/{bundle_id}/availability"; +pub const RUNTIME_ROUTE_WORKSPACE_PROMPT_PROJECTIONS: &str = "/v1/workspace-prompt-projections"; +pub const RUNTIME_ROUTE_WORKING_DIRECTORIES: &str = "/v1/working-directories"; +pub const RUNTIME_ROUTE_WORKING_DIRECTORY_REPOSITORY_ACCESS: &str = + "/v1/working-directories/repository-access"; +pub const RUNTIME_ROUTE_SSH_PROBE: &str = "/v1/repositories/ssh/probe"; +pub const RUNTIME_ROUTE_REPOSITORY_REFS_OBSERVE: &str = "/v1/repository-refs/observe"; +pub const RUNTIME_ROUTE_WORKDIR_SESSIONS: &str = + "/v1/working-directories/{working_directory_id}/sessions"; +pub const RUNTIME_ROUTE_WORKDIR_SESSION_OPERATIONS: &str = + "/v1/workdir-sessions/{session_id}/operations"; +pub const RUNTIME_ROUTE_WORKDIR_SESSION: &str = "/v1/workdir-sessions/{session_id}"; +pub const RUNTIME_ROUTE_WORKING_DIRECTORY: &str = "/v1/working-directories/{working_directory_id}"; +pub const RUNTIME_ROUTE_PROTOCOL_WS: &str = "/v1/protocol/ws"; +pub const RUNTIME_ROUTE_WORKER_PROTOCOL_WS: &str = "/v1/workers/{worker_id}/protocol/ws"; +pub const RUNTIME_ROUTE_WORKER_ATTACHMENTS: &str = "/v1/workers/{worker_id}/attachments"; +pub const RUNTIME_ROUTE_WORKER_ATTACHMENT: &str = + "/v1/workers/{worker_id}/attachments/{artifact_id}"; + +/// Intentionally out-of-contract Runtime routes. This inventory keeps manual paths visible rather +/// than letting them be mistaken for management-contract omissions. +pub const REMAINING_RUNTIME_ROUTES: &[RemainingRuntimeRoute] = &[ + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_VERIFICATION_CHALLENGE, + reason: "Workspace verification handshake", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_VERIFICATION_ACK, + reason: "Workspace verification handshake", + }, + RemainingRuntimeRoute { + method: "GET", + path: RUNTIME_ROUTE_CONFIG_BUNDLES, + reason: "Workspace Config transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_CONFIG_BUNDLES, + reason: "Workspace Config transport", + }, + RemainingRuntimeRoute { + method: "GET", + path: RUNTIME_ROUTE_CONFIG_BUNDLE_AVAILABILITY, + reason: "Workspace Config transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_WORKSPACE_PROMPT_PROJECTIONS, + reason: "Workspace Config transport", + }, + RemainingRuntimeRoute { + method: "GET", + path: RUNTIME_ROUTE_WORKING_DIRECTORIES, + reason: "Workdir transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_WORKING_DIRECTORIES, + reason: "Workdir transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_WORKING_DIRECTORY_REPOSITORY_ACCESS, + reason: "Workdir transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_SSH_PROBE, + reason: "Repository SSH transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_REPOSITORY_REFS_OBSERVE, + reason: "Repository observation transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_WORKDIR_SESSIONS, + reason: "Workdir session transport", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_WORKDIR_SESSION_OPERATIONS, + reason: "Workdir session transport", + }, + RemainingRuntimeRoute { + method: "DELETE", + path: RUNTIME_ROUTE_WORKDIR_SESSION, + reason: "Workdir session transport", + }, + RemainingRuntimeRoute { + method: "GET", + path: RUNTIME_ROUTE_WORKING_DIRECTORY, + reason: "Workdir transport", + }, + RemainingRuntimeRoute { + method: "DELETE", + path: RUNTIME_ROUTE_WORKING_DIRECTORY, + reason: "Workdir transport", + }, + RemainingRuntimeRoute { + method: "GET", + path: RUNTIME_ROUTE_PROTOCOL_WS, + reason: "WebSocket protocol transport (ws-server feature)", + }, + RemainingRuntimeRoute { + method: "GET", + path: RUNTIME_ROUTE_WORKER_PROTOCOL_WS, + reason: "WebSocket protocol transport (ws-server feature)", + }, + RemainingRuntimeRoute { + method: "POST", + path: RUNTIME_ROUTE_WORKER_ATTACHMENTS, + reason: "Binary attachment transport", + }, + RemainingRuntimeRoute { + method: "DELETE", + path: RUNTIME_ROUTE_WORKER_ATTACHMENT, + reason: "Binary attachment transport", + }, +]; + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Clone)] + struct RoundTripService; + + fn test_error(status: u16) -> RuntimeApiError { + RuntimeApiError::new(status, "test_error", "test error") + } + + impl RuntimeApi for RoundTripService { + async fn ping( + &self, + _workspace_id: String, + ) -> Result { + Err(test_error(501)) + } + + async fn runtime_summary(&self) -> Result { + Ok(RuntimeSummaryResponse { + runtime: RuntimeSummary { + display_name: Some("round-trip".to_string()), + backend: RuntimeBackendKind::Memory, + status: RuntimeStatus::Running, + worker_count: 0, + active_worker_count: 0, + stopped_worker_count: 0, + diagnostic_count: 0, + os: "test".to_string(), + arch: "test".to_string(), + worker_creation_available: true, + }, + }) + } + + async fn list_workers( + &self, + query: WorkerListQuery, + ) -> Result { + assert_eq!(query.status, Some(WorkerStatusFilter::Stopped)); + Ok(WorkersResponse { + workers: Vec::new(), + }) + } + + async fn get_worker(&self, worker_id: String) -> Result { + assert_eq!(worker_id, "missing worker"); + Err(test_error(404)) + } + + async fn create_worker( + &self, + _request: CreateWorkerRequest, + ) -> Result { + Err(test_error(501)) + } + async fn delete_worker( + &self, + _worker_id: String, + ) -> Result { + Err(test_error(501)) + } + async fn send_worker_input( + &self, + _worker_id: String, + _input: WorkerInput, + ) -> Result { + Err(test_error(501)) + } + async fn stop_worker( + &self, + _worker_id: String, + _request: WorkerLifecycleRequest, + ) -> Result { + Err(test_error(501)) + } + async fn cancel_worker( + &self, + _worker_id: String, + _request: WorkerLifecycleRequest, + ) -> Result { + Err(test_error(501)) + } + async fn restore_worker( + &self, + _worker_id: String, + _request: EmptyObjectRequest, + ) -> Result { + Err(test_error(501)) + } + async fn replace_worker_workspace_api( + &self, + _worker_id: String, + _request: WorkerWorkspaceApiRequest, + ) -> Result { + Err(test_error(501)) + } + async fn complete_worker_arguments( + &self, + _worker_id: String, + _request: CompletionRequest, + ) -> Result { + Err(test_error(501)) + } + async fn retention_inventory( + &self, + _worker_id: String, + ) -> Result { + Err(test_error(501)) + } + async fn execute_retention( + &self, + _worker_id: String, + _request: WorkerRetentionExecutionRequest, + ) -> Result { + Err(test_error(501)) + } + } + + #[tokio::test] + async fn generated_client_and_router_round_trip_paths_queries_and_errors() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + server_support::framework::serve(listener, RuntimeApiAxum::router(RoundTripService)) + .await + .unwrap(); + }); + let client = RuntimeApiClient::builder(&format!("http://{address}")) + .unwrap() + .build() + .unwrap(); + + let summary = client.runtime_summary().await.unwrap(); + assert_eq!(summary.runtime.display_name.as_deref(), Some("round-trip")); + let workers = client + .list_workers(WorkerListQuery { + status: Some(WorkerStatusFilter::Stopped), + }) + .await + .unwrap(); + assert!(workers.workers.is_empty()); + let error = client + .get_worker("missing worker".to_string()) + .await + .unwrap_err(); + assert!( + matches!(error, client_support::ClientError::Public { status, .. } if status.as_u16() == 404) + ); + server.abort(); + } + + #[test] + fn contract_inventory_is_complete_and_unique() { + let operations = RuntimeApiMetadata::OPERATIONS; + assert_eq!(operations.len(), 14); + let mut routes = operations + .iter() + .map(|operation| (format!("{:?}", operation.method), operation.path)) + .collect::>(); + routes.sort_unstable(); + routes.dedup(); + assert_eq!(routes.len(), operations.len()); + assert!( + operations + .iter() + .all(|operation| operation.path.starts_with(RUNTIME_API_BASE_PATH)) + ); + } + + #[test] + fn remaining_route_inventory_does_not_overlap_contract() { + assert_eq!(REMAINING_RUNTIME_ROUTES.len(), 20); + for remaining in REMAINING_RUNTIME_ROUTES { + assert!(!RuntimeApiMetadata::OPERATIONS.iter().any(|operation| { + format!("{:?}", operation.method).eq_ignore_ascii_case(remaining.method) + && operation.path == remaining.path + })); + } + } +} diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index dabdf1409..68eb0cd97 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -32,6 +32,7 @@ flow = { path = "../flow" } fs4 = { workspace = true, optional = true } manifest.workspace = true protocol.workspace = true +runtime-api.workspace = true serde = { workspace = true, features = ["derive"] } session-store.workspace = true sha2.workspace = true diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index fadffb45c..a42ab6fa3 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -6,6 +6,8 @@ //! Runtime process directly; a backend is expected to own any browser-facing //! credentials, registration, and policy. +mod runtime_management_api; + use crate::auth::{RuntimeAuthContext, new_token_id, unix_now_seconds}; use crate::catalog::{ ConfigBundleRef, CreateWorkerRequest, RepositoryRefObservationRequest, WorkerDetail, @@ -25,13 +27,11 @@ use crate::retention::{ #[cfg(feature = "ws-server")] use crate::runtime::RuntimeSubscriptionRecvError; use crate::ssh_host_key_probe::{ - SSH_HOST_KEY_PROBE_OPERATION, SSH_HOST_KEY_PROBE_PATH, SSH_KEYSCAN_TIMEOUT, - SshHostKeyProbeError, SshHostKeyProbeRequest, SshHostKeyProbeResponse, - probe_ssh_host_keys_with_program, + SSH_HOST_KEY_PROBE_OPERATION, SSH_KEYSCAN_TIMEOUT, SshHostKeyProbeError, + SshHostKeyProbeRequest, SshHostKeyProbeResponse, probe_ssh_host_keys_with_program, }; use crate::workspace_issuer::{ - RuntimeVerificationSigner, VerifiedWorkspaceCapability, WORKSPACE_VERIFICATION_ACK_PATH, - WORKSPACE_VERIFICATION_CHALLENGE_PATH, WORKSPACE_VERIFICATION_OPERATION, + RuntimeVerificationSigner, VerifiedWorkspaceCapability, WORKSPACE_VERIFICATION_OPERATION, WorkspaceCapabilityExpectation, WorkspaceCapabilityVerifier, WorkspaceRuntimeVerificationAcknowledgement, WorkspaceRuntimeVerificationAuthority, WorkspaceRuntimeVerificationChallenge, WorkspaceRuntimeVerificationReceipt, @@ -74,7 +74,6 @@ use workdir::{ WorkdirTransportError, WorkdirTransportErrorCode, }, }; -use workspace_api::WorkerRestoreState; const DEFAULT_RUNTIME_HTTP_PORT: u16 = 38800; pub const RUNTIME_HTTP_PROTOCOL_MIN_VERSION: u32 = 1; @@ -165,6 +164,18 @@ pub async fn serve_runtime_http_with_workspace_auth( Ok(()) } +fn remaining_route(path: &'static str, methods: &[&str]) -> &'static str { + for method in methods { + assert!( + runtime_api::REMAINING_RUNTIME_ROUTES + .iter() + .any(|route| route.method == *method && route.path == path), + "handwritten Runtime route {method} {path} is missing from runtime-api inventory", + ); + } + path +} + /// Build the REST router for an existing Runtime. /// /// Handlers delegate to [`Runtime`] methods and keep Worker authority Runtime-local. @@ -210,101 +221,102 @@ fn runtime_http_router_with_auth_and_ssh_keyscan_program( }; let router = Router::new() - .route("/v1/ping", get(get_runtime_ping)) .route( - WORKSPACE_VERIFICATION_CHALLENGE_PATH, + remaining_route(runtime_api::RUNTIME_ROUTE_VERIFICATION_CHALLENGE, &["POST"]), post(post_workspace_verification_challenge), ) .route( - WORKSPACE_VERIFICATION_ACK_PATH, + remaining_route(runtime_api::RUNTIME_ROUTE_VERIFICATION_ACK, &["POST"]), post(post_workspace_verification_acknowledgement), ) - .route("/v1/runtime", get(get_runtime)) .route( - "/v1/config-bundles", + remaining_route(runtime_api::RUNTIME_ROUTE_CONFIG_BUNDLES, &["GET", "POST"]), get(list_config_bundles).post(store_config_bundle), ) .route( - "/v1/config-bundles/{bundle_id}/availability", + remaining_route( + runtime_api::RUNTIME_ROUTE_CONFIG_BUNDLE_AVAILABILITY, + &["GET"], + ), get(check_config_bundle), ) .route( - "/v1/workspace-prompt-projections", + remaining_route( + runtime_api::RUNTIME_ROUTE_WORKSPACE_PROMPT_PROJECTIONS, + &["POST"], + ), post(observe_workspace_prompt_projection), ) .route( - "/v1/working-directories", + remaining_route( + runtime_api::RUNTIME_ROUTE_WORKING_DIRECTORIES, + &["GET", "POST"], + ), get(list_working_directories).post(create_working_directory), ) .route( - "/v1/working-directories/repository-access", + remaining_route( + runtime_api::RUNTIME_ROUTE_WORKING_DIRECTORY_REPOSITORY_ACCESS, + &["POST"], + ), post(authorize_working_directory_repository_access), ) .route( - SSH_HOST_KEY_PROBE_PATH, + remaining_route(runtime_api::RUNTIME_ROUTE_SSH_PROBE, &["POST"]), post(probe_repository_ssh_host_keys), ) - .route("/v1/repository-refs/observe", post(observe_repository_ref)) .route( - "/v1/working-directories/{working_directory_id}/sessions", + remaining_route( + runtime_api::RUNTIME_ROUTE_REPOSITORY_REFS_OBSERVE, + &["POST"], + ), + post(observe_repository_ref), + ) + .route( + remaining_route(runtime_api::RUNTIME_ROUTE_WORKDIR_SESSIONS, &["POST"]), post(open_workdir_session), ) .route( - "/v1/workdir-sessions/{session_id}/operations", + remaining_route( + runtime_api::RUNTIME_ROUTE_WORKDIR_SESSION_OPERATIONS, + &["POST"], + ), post(run_workdir_session_operation), ) .route( - "/v1/workdir-sessions/{session_id}", + remaining_route(runtime_api::RUNTIME_ROUTE_WORKDIR_SESSION, &["DELETE"]), delete(close_workdir_session), ) .route( - "/v1/working-directories/{working_directory_id}", + remaining_route( + runtime_api::RUNTIME_ROUTE_WORKING_DIRECTORY, + &["GET", "DELETE"], + ), get(get_working_directory).delete(cleanup_working_directory), ) - .route("/v1/workers", get(list_workers).post(create_worker)) .route( - "/v1/workers/{worker_id}", - get(get_worker).delete(delete_worker), - ) - .route( - "/v1/workers/{worker_id}/retention/inventory", - get(worker_retention_inventory), - ) - .route( - "/v1/workers/{worker_id}/retention/execute", - post(execute_worker_retention), - ) - .route("/v1/workers/{worker_id}/input", post(send_worker_input)) - .route( - "/v1/workers/{worker_id}/attachments", + remaining_route(runtime_api::RUNTIME_ROUTE_WORKER_ATTACHMENTS, &["POST"]), post(upload_worker_file).layer(DefaultBodyLimit::max(MAX_WORKER_FILE_UPLOAD_BYTES)), ) .route( - "/v1/workers/{worker_id}/attachments/{artifact_id}", + remaining_route(runtime_api::RUNTIME_ROUTE_WORKER_ATTACHMENT, &["DELETE"]), delete(delete_worker_uploaded_file), - ) - .route("/v1/workers/{worker_id}/restore", post(restore_worker)) - .route( - "/v1/workers/{worker_id}/workspace-api", - post(replace_worker_workspace_api), - ) - .route( - "/v1/workers/{worker_id}/completions", - post(worker_completions), - ) - .route("/v1/workers/{worker_id}/stop", post(stop_worker)) - .route("/v1/workers/{worker_id}/cancel", post(cancel_worker)); + ); #[cfg(feature = "ws-server")] let router = router - .route("/v1/protocol/ws", get(runtime_protocol_ws)) .route( - "/v1/workers/{worker_id}/protocol/ws", + remaining_route(runtime_api::RUNTIME_ROUTE_PROTOCOL_WS, &["GET"]), + get(runtime_protocol_ws), + ) + .route( + remaining_route(runtime_api::RUNTIME_ROUTE_WORKER_PROTOCOL_WS, &["GET"]), get(worker_protocol_ws), ); router .with_state(state.clone()) + .merge(runtime_management_api::router(state.clone())) .layer(middleware::from_fn_with_state(state, require_runtime_auth)) } @@ -373,11 +385,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 { @@ -421,20 +435,6 @@ pub struct RuntimeHttpWorkerResponse { pub worker: WorkerDetail, } -/// Explicit restore outcome. This envelope is returned for every reachable -/// Runtime restore attempt, including preflight rejection and uncertain commit. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct RuntimeHttpWorkerRestoreResponse { - pub state: WorkerRestoreState, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub worker: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub reason_code: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub message: Option, -} - /// Replace the Workspace API binding for an existing Worker. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RuntimeHttpWorkerWorkspaceApiRequest { @@ -685,6 +685,7 @@ async fn post_workspace_verification_acknowledgement( })) } +#[allow(dead_code)] async fn get_runtime_ping( State(state): State, Extension(auth): Extension, @@ -727,6 +728,7 @@ async fn get_runtime_ping( })) } +#[allow(dead_code)] async fn get_runtime( State(state): State, ) -> RestResult { @@ -806,6 +808,7 @@ async fn check_config_bundle( })) } +#[allow(dead_code)] async fn list_workers( State(state): State, auth: Option>, @@ -1146,6 +1149,7 @@ async fn cleanup_working_directory( })) } +#[allow(dead_code)] async fn get_worker( State(state): State, auth: Option>, @@ -1160,6 +1164,7 @@ async fn get_worker( Ok(Json(RuntimeHttpWorkerResponse { worker })) } +#[allow(dead_code)] async fn delete_worker( State(state): State, auth: Option>, @@ -1174,6 +1179,7 @@ async fn delete_worker( Ok(Json(RuntimeHttpWorkerDeleteResponse { worker })) } +#[allow(dead_code)] async fn create_worker( State(state): State, auth: Option>, @@ -1188,6 +1194,7 @@ async fn create_worker( Ok(Json(RuntimeHttpWorkerResponse { worker })) } +#[allow(dead_code)] async fn replace_worker_workspace_api( State(state): State, auth: Option>, @@ -1210,27 +1217,6 @@ async fn replace_worker_workspace_api( Ok(Json(RuntimeHttpWorkerResponse { worker })) } -async fn restore_worker( - State(state): State, - auth: Option>, - Path(worker_id): Path, -) -> RestResult { - let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let result = match auth_workspace_scope(&state, auth.as_ref())? { - Some(scope) => state - .runtime - .restore_worker_operation_scoped(&scope, &worker_ref), - None => state.runtime.restore_worker_operation(&worker_ref), - } - .map_err(RuntimeHttpRestError::runtime)?; - Ok(Json(RuntimeHttpWorkerRestoreResponse { - state: result.state, - worker: result.worker, - reason_code: result.reason_code, - message: result.message, - })) -} - #[cfg(feature = "ws-server")] const RUNTIME_PROTOCOL_OUTBOUND_CAPACITY: usize = 256; @@ -1780,6 +1766,7 @@ fn protocol_error_event(message: impl Into) -> protocol::Event { } } +#[allow(dead_code)] async fn worker_retention_inventory( State(state): State, auth: Option>, @@ -1800,6 +1787,7 @@ async fn worker_retention_inventory( .map_err(RuntimeHttpRestError::runtime) } +#[allow(dead_code)] async fn execute_worker_retention( State(state): State, auth: Option>, @@ -1835,6 +1823,7 @@ async fn execute_worker_retention( .map_err(RuntimeHttpRestError::runtime) } +#[allow(dead_code)] async fn send_worker_input( State(state): State, auth: Option>, @@ -1851,6 +1840,7 @@ async fn send_worker_input( Ok(Json(RuntimeHttpWorkerInputResponse { ack })) } +#[allow(dead_code)] async fn worker_completions( State(state): State, auth: Option>, @@ -1987,6 +1977,7 @@ async fn delete_worker_uploaded_file( })) } +#[allow(dead_code)] async fn stop_worker( State(state): State, auth: Option>, @@ -2005,6 +1996,7 @@ async fn stop_worker( Ok(Json(RuntimeHttpWorkerLifecycleResponse { ack })) } +#[allow(dead_code)] async fn cancel_worker( State(state): State, auth: Option>, @@ -2102,8 +2094,9 @@ async fn require_runtime_auth( }; match workspace_auth.verifier.verify(token, &expected) { Ok(verified) => { - let is_verification = path_and_query == WORKSPACE_VERIFICATION_CHALLENGE_PATH - || path_and_query == WORKSPACE_VERIFICATION_ACK_PATH; + let is_verification = path_and_query + == runtime_api::RUNTIME_ROUTE_VERIFICATION_CHALLENGE + || path_and_query == runtime_api::RUNTIME_ROUTE_VERIFICATION_ACK; if !is_verification { let record = match workspace_auth .verifications @@ -2152,7 +2145,8 @@ async fn require_runtime_auth( token_id: verified.token_id, expires_at: u64::try_from(verified.expires_at).unwrap_or(0), }); - return next.run(request).await; + let context = request.extensions().get::().cloned(); + return runtime_management_api::scope_auth(context, next.run(request)).await; } Err(error) => { return RuntimeHttpRestError::new( @@ -2168,7 +2162,8 @@ async fn require_runtime_auth( let workspace_bootstrap_request = request.method() == Method::POST && matches!( request.uri().path(), - WORKSPACE_VERIFICATION_CHALLENGE_PATH | WORKSPACE_VERIFICATION_ACK_PATH + runtime_api::RUNTIME_ROUTE_VERIFICATION_CHALLENGE + | runtime_api::RUNTIME_ROUTE_VERIFICATION_ACK ); if state.workspace_auth.is_some() && !workspace_bootstrap_request { let local_token_matches = state @@ -2202,7 +2197,8 @@ async fn require_runtime_auth( expires_at: 0, }); } - next.run(request).await + let context = request.extensions().get::().cloned(); + runtime_management_api::scope_auth(context, next.run(request)).await } fn auth_workspace_scope( @@ -2239,7 +2235,8 @@ fn auth_workspace_scope( } fn workspace_runtime_operation(method: &Method, path: &str) -> &'static str { - if (path == WORKSPACE_VERIFICATION_CHALLENGE_PATH || path == WORKSPACE_VERIFICATION_ACK_PATH) + if (path == runtime_api::RUNTIME_ROUTE_VERIFICATION_CHALLENGE + || path == runtime_api::RUNTIME_ROUTE_VERIFICATION_ACK) && *method == Method::POST { return WORKSPACE_VERIFICATION_OPERATION; @@ -2266,9 +2263,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s if path == "/v1/workers" && *method == Method::POST { return Some("workers:create"); } - if (path == "/v1/working-directories/repository-access" - || path == "/v1/repository-refs/observe" - || path == SSH_HOST_KEY_PROBE_PATH) + if (path == runtime_api::RUNTIME_ROUTE_WORKING_DIRECTORY_REPOSITORY_ACCESS + || path == runtime_api::RUNTIME_ROUTE_REPOSITORY_REFS_OBSERVE + || path == runtime_api::RUNTIME_ROUTE_SSH_PROBE) && *method == Method::POST { return Some(SSH_HOST_KEY_PROBE_OPERATION); @@ -2278,9 +2275,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s { return Some("workdirs:operate"); } - if path.starts_with("/v1/config-bundles") - || path.starts_with("/v1/workspace-prompt-projections") - || path.starts_with("/v1/working-directories") + if path.starts_with(runtime_api::RUNTIME_ROUTE_CONFIG_BUNDLES) + || path.starts_with(runtime_api::RUNTIME_ROUTE_WORKSPACE_PROMPT_PROJECTIONS) + || path.starts_with(runtime_api::RUNTIME_ROUTE_WORKING_DIRECTORIES) { return Some("workers:create"); } @@ -2293,7 +2290,7 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s if path.ends_with("/stop") || path.ends_with("/cancel") { return Some("workers:stop"); } - if path == "/v1/protocol/ws" { + if path == runtime_api::RUNTIME_ROUTE_PROTOCOL_WS { return Some("workers:list"); } if path.ends_with("/protocol") || path.ends_with("/protocol/ws") { @@ -2488,6 +2485,11 @@ fn status_for_runtime_error(error: &RuntimeError) -> StatusCode { { StatusCode::NOT_FOUND } + RuntimeError::InvalidRequest(message) + if message.contains("already created with a different fingerprint") => + { + StatusCode::CONFLICT + } RuntimeError::RuntimeStopped | RuntimeError::RuntimeStoreAlreadyOpen { .. } | RuntimeError::WorkerExecutionUnavailable { .. } @@ -2526,6 +2528,11 @@ fn code_for_runtime_error(error: &RuntimeError) -> String { RuntimeError::WorkerExecutionRejected { .. } => "worker_execution_rejected".to_string(), RuntimeError::WorkspaceOwnerMismatch { .. } => "workspace_owner_mismatch".to_string(), RuntimeError::LimitTooLarge { .. } => "limit_too_large".to_string(), + RuntimeError::InvalidRequest(message) + if message.contains("already created with a different fingerprint") => + { + "worker_create_conflict".to_string() + } RuntimeError::InvalidRequest(_) => "invalid_request".to_string(), RuntimeError::WorkingDirectory(diagnostic) => diagnostic.code.clone(), RuntimeError::InvalidInitialInputKind { .. } => "invalid_initial_input_kind".to_string(), @@ -2569,6 +2576,7 @@ mod tests { WorkerExecutionSpawnResult, }; use crate::management::RuntimeOptions; + use crate::retention::{DiagnosticsDisposition, SessionDisposition}; use crate::workspace_issuer::{ InMemoryWorkspaceClaimReplayProtection, InMemoryWorkspaceRuntimeVerificationAuthority, WorkspaceCapabilityClaims, WorkspaceCapabilityVerifier, WorkspaceIssuerTrustRecord, @@ -2580,14 +2588,108 @@ mod tests { use manifest::{Scope, SharedScope}; use sha2::Digest as _; use tower::ServiceExt; + + #[derive(Clone)] + struct TestRuntimeAuthorizer(String); + + impl runtime_api::client_support::RequestAuthorizer for TestRuntimeAuthorizer { + fn authorize( + &self, + _request: runtime_api::client_support::AuthorizerRequest<'_>, + ) -> Result< + runtime_api::client_support::framework::header::HeaderMap, + runtime_api::client_support::AuthorizationError, + > { + let mut headers = runtime_api::client_support::framework::header::HeaderMap::new(); + headers.insert( + runtime_api::client_support::framework::header::AUTHORIZATION, + format!("Bearer {}", self.0) + .parse() + .map_err(|_| runtime_api::client_support::AuthorizationError::new())?, + ); + Ok(headers) + } + } + use workdir::{ GrepOutputMode, GrepRequest, LocalWorkdirSession, StatRequest, Workdir, WorkdirPath, WorkdirSessionCapabilities, }; + #[test] + fn generated_runtime_routes_are_classified_by_permission_authority() { + for operation in ::OPERATIONS { + let method = match operation.method { + runtime_api::HttpMethod::Get => Method::GET, + runtime_api::HttpMethod::Post => Method::POST, + runtime_api::HttpMethod::Put => Method::PUT, + runtime_api::HttpMethod::Patch => Method::PATCH, + runtime_api::HttpMethod::Delete => Method::DELETE, + runtime_api::HttpMethod::Head | runtime_api::HttpMethod::Options => { + panic!("unexpected management contract method") + } + }; + let path = operation.path.replace("{worker_id}", "worker-1"); + let permission = required_runtime_permission(&method, &path); + if operation.operation_id == "runtime_summary" { + assert_eq!( + permission, None, + "{:?} {}", + operation.method, operation.path + ); + } else { + assert!( + permission.is_some(), + "unclassified generated route: {:?} {}", + operation.method, + operation.path + ); + } + } + } + + #[test] + fn remaining_route_inventory_covers_every_handwritten_router_path() { + let inventoried = runtime_api::REMAINING_RUNTIME_ROUTES + .iter() + .map(|route| route.path) + .collect::>(); + for path in [ + runtime_api::RUNTIME_ROUTE_VERIFICATION_CHALLENGE, + runtime_api::RUNTIME_ROUTE_VERIFICATION_ACK, + runtime_api::RUNTIME_ROUTE_CONFIG_BUNDLES, + runtime_api::RUNTIME_ROUTE_CONFIG_BUNDLE_AVAILABILITY, + runtime_api::RUNTIME_ROUTE_WORKSPACE_PROMPT_PROJECTIONS, + runtime_api::RUNTIME_ROUTE_WORKING_DIRECTORIES, + runtime_api::RUNTIME_ROUTE_WORKING_DIRECTORY_REPOSITORY_ACCESS, + runtime_api::RUNTIME_ROUTE_SSH_PROBE, + runtime_api::RUNTIME_ROUTE_REPOSITORY_REFS_OBSERVE, + runtime_api::RUNTIME_ROUTE_WORKDIR_SESSIONS, + runtime_api::RUNTIME_ROUTE_WORKDIR_SESSION_OPERATIONS, + runtime_api::RUNTIME_ROUTE_WORKDIR_SESSION, + runtime_api::RUNTIME_ROUTE_WORKING_DIRECTORY, + runtime_api::RUNTIME_ROUTE_WORKER_ATTACHMENTS, + runtime_api::RUNTIME_ROUTE_WORKER_ATTACHMENT, + runtime_api::RUNTIME_ROUTE_PROTOCOL_WS, + runtime_api::RUNTIME_ROUTE_WORKER_PROTOCOL_WS, + ] { + assert!( + inventoried.contains(path), + "missing handwritten route inventory for {path}" + ); + } + } + #[tokio::test] async fn workspace_signed_verification_requires_exact_request_and_acknowledges_response() { - let runtime = Runtime::new_memory(); + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(AcceptingBackend)) + .unwrap(); + runtime + .store_config_bundle(test_bundle(ProfileSelector::Builtin( + "builtin:coder".to_string(), + ))) + .unwrap(); let workspace_identity = RuntimeIdentityMaterial::generate("workspace-key").unwrap(); let runtime_identity = RuntimeIdentityMaterial::generate("runtime-test").unwrap(); let workspace_public_key = @@ -2620,7 +2722,7 @@ mod tests { ) .unwrap(); let app = runtime_http_router_with_workspace_auth( - runtime, + runtime.clone(), None, WorkspaceRuntimeHttpAuth { verifier, @@ -2653,7 +2755,7 @@ mod tests { worker_id: None, operation: WORKSPACE_VERIFICATION_OPERATION.to_string(), method: "POST".to_string(), - path_and_query: WORKSPACE_VERIFICATION_CHALLENGE_PATH.to_string(), + path_and_query: runtime_api::RUNTIME_ROUTE_VERIFICATION_CHALLENGE.to_string(), body_digest: workspace_request_body_digest(&body), iat: unix_now_i64(), exp: challenge.expires_at, @@ -2667,7 +2769,7 @@ mod tests { .oneshot( Request::builder() .method(Method::POST) - .uri(WORKSPACE_VERIFICATION_CHALLENGE_PATH) + .uri(runtime_api::RUNTIME_ROUTE_VERIFICATION_CHALLENGE) .header(header::AUTHORIZATION, format!("Bearer {token}")) .header(header::CONTENT_TYPE, "application/json") .body(Body::from(body)) @@ -2713,7 +2815,7 @@ mod tests { }; let acknowledgement_body = serde_json::to_vec(&acknowledgement).unwrap(); let acknowledgement_claims = WorkspaceCapabilityClaims { - path_and_query: WORKSPACE_VERIFICATION_ACK_PATH.to_string(), + path_and_query: runtime_api::RUNTIME_ROUTE_VERIFICATION_ACK.to_string(), body_digest: workspace_request_body_digest(&acknowledgement_body), jti: "ack-token".to_string(), ..claims @@ -2728,7 +2830,7 @@ mod tests { .oneshot( Request::builder() .method(Method::POST) - .uri(WORKSPACE_VERIFICATION_ACK_PATH) + .uri(runtime_api::RUNTIME_ROUTE_VERIFICATION_ACK) .header( header::AUTHORIZATION, format!("Bearer {acknowledgement_token}"), @@ -2746,7 +2848,7 @@ mod tests { for authorization in [None, Some("Bearer malformed")] { let mut request = Request::builder() .method(Method::POST) - .uri("/v1/config-bundles"); + .uri(runtime_api::RUNTIME_ROUTE_CONFIG_BUNDLES); if let Some(authorization) = authorization { request = request.header(header::AUTHORIZATION, authorization); } @@ -2772,6 +2874,7 @@ mod tests { ) .unwrap(); let response = app + .clone() .oneshot( Request::builder() .method(Method::GET) @@ -2784,6 +2887,64 @@ mod tests { .await .unwrap(); assert_eq!(response.status(), StatusCode::OK); + + let mut retention_create = task_request("retention-scope"); + retention_create.workspace_api = Some(WorkspaceApiRef { + workspace_id: "workspace-a".to_string(), + base_url: "https://backend.test".to_string(), + }); + let retention_worker_id = retention_create.worker_id.clone(); + runtime.create_worker(retention_create).unwrap(); + let retention_path = format!("/v1/workers/{retention_worker_id}/retention/execute"); + let retention_request = WorkerRetentionExecutionRequest { + operation_id: "retention-op".to_string(), + input_fingerprint: "fingerprint".to_string(), + archive_id: None, + workspace_id: "workspace-b".to_string(), + source_runtime_id: "runtime-a".to_string(), + worker_id: retention_worker_id, + expected_worker_revision: "revision".to_string(), + source_created_at: "2025-01-01T00:00:00Z".to_string(), + removed_at: "2025-01-02T00:00:00Z".to_string(), + effective_profile: None, + retention_class: None, + policy_id: "policy".to_string(), + policy_revision: 1, + session_disposition: SessionDisposition::Purge, + diagnostics_disposition: DiagnosticsDisposition::Retain, + }; + let retention_body = serde_json::to_vec(&retention_request).unwrap(); + let retention_claims = WorkspaceCapabilityClaims { + operation: required_runtime_permission(&Method::POST, &retention_path) + .unwrap() + .to_string(), + method: "POST".to_string(), + path_and_query: retention_path.clone(), + worker_id: Some(retention_worker_id.to_string()), + body_digest: workspace_request_body_digest(&retention_body), + jti: "retention-scope-token".to_string(), + ..ping_claims + }; + let retention_token = issue_workspace_capability_token( + &workspace_identity.signing_key().unwrap(), + &retention_claims, + ) + .unwrap(); + let response = app + .oneshot( + Request::builder() + .method(Method::POST) + .uri(retention_path) + .header(header::AUTHORIZATION, format!("Bearer {retention_token}")) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(retention_body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + let error: RuntimeHttpErrorResponse = read_json(response).await; + assert_eq!(error.error.code, "worker_not_found"); } #[test] @@ -2941,15 +3102,21 @@ mod tests { #[test] fn workdir_routes_require_dedicated_operation_permission() { assert_eq!( - required_runtime_permission(&Method::POST, "/v1/working-directories/repository-access",), + required_runtime_permission( + &Method::POST, + runtime_api::RUNTIME_ROUTE_WORKING_DIRECTORY_REPOSITORY_ACCESS, + ), Some("workdirs:operate") ); assert_eq!( - required_runtime_permission(&Method::POST, "/v1/repository-refs/observe"), + required_runtime_permission( + &Method::POST, + runtime_api::RUNTIME_ROUTE_REPOSITORY_REFS_OBSERVE + ), Some("workdirs:operate") ); assert_eq!( - required_runtime_permission(&Method::POST, SSH_HOST_KEY_PROBE_PATH), + required_runtime_permission(&Method::POST, runtime_api::RUNTIME_ROUTE_SSH_PROBE), Some(SSH_HOST_KEY_PROBE_OPERATION) ); assert_eq!( @@ -3264,6 +3431,132 @@ 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 listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server_app = app.clone(); + let server = tokio::spawn(async move { + axum::serve(listener, server_app).await.unwrap(); + }); + let client = runtime_api::RuntimeApiClient::builder(&format!("http://{address}")) + .unwrap() + .authorizer(TestRuntimeAuthorizer(token.to_string())) + .build() + .unwrap(); + let generated_request: runtime_api::CreateWorkerRequest = serde_json::from_value( + serde_json::to_value(task_request("generated-roundtrip")).unwrap(), + ) + .unwrap(); + let generated_worker_id = generated_request.worker_id.to_string(); + let mut conflicting_request = generated_request.clone(); + conflicting_request.create_fingerprint = "different-fingerprint".to_string(); + let generated = client.create_worker(generated_request).await.unwrap(); + assert_eq!(generated.worker.worker_id.to_string(), generated_worker_id); + let generated_input: runtime_api::WorkerInput = serde_json::from_value( + serde_json::to_value(WorkerInput::user("generated client input")).unwrap(), + ) + .unwrap(); + client + .send_worker_input(generated_worker_id.clone(), generated_input) + .await + .unwrap(); + client + .stop_worker( + generated_worker_id.clone(), + runtime_api::WorkerLifecycleRequest::default(), + ) + .await + .unwrap(); + client + .restore_worker( + generated_worker_id.clone(), + runtime_api::EmptyObjectRequest::default(), + ) + .await + .unwrap(); + let conflict = client.create_worker(conflicting_request).await.unwrap_err(); + assert!( + matches!( + &conflict, + runtime_api::client_support::ClientError::Public { status, .. } + if *status == StatusCode::CONFLICT + ), + "{conflict:?}" + ); + client + .stop_worker( + generated_worker_id.clone(), + runtime_api::WorkerLifecycleRequest::default(), + ) + .await + .unwrap(); + client + .delete_worker(generated_worker_id.clone()) + .await + .unwrap(); + let missing = client + .get_worker(generated_worker_id.clone()) + .await + .unwrap_err(); + assert!(matches!( + missing, + runtime_api::client_support::ClientError::Public { status, .. } + if status == StatusCode::NOT_FOUND + )); + let unauthorized = runtime_api::RuntimeApiClient::builder(&format!("http://{address}")) + .unwrap() + .build() + .unwrap() + .list_workers(runtime_api::WorkerListQuery::default()) + .await + .unwrap_err(); + assert!(matches!( + unauthorized, + runtime_api::client_support::ClientError::Public { status, .. } + if status == StatusCode::UNAUTHORIZED + )); + let forbidden_response = app + .clone() + .oneshot( + Request::builder() + .method(Method::GET) + .uri("/v1/ping") + .header(header::AUTHORIZATION, format!("Bearer {token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(forbidden_response.status(), StatusCode::FORBIDDEN); + let bounded_client = runtime_api::RuntimeApiClient::builder(&format!("http://{address}")) + .unwrap() + .authorizer(TestRuntimeAuthorizer(token.to_string())) + .response_body_limit(1) + .build() + .unwrap(); + assert!(matches!( + bounded_client.runtime_summary().await.unwrap_err(), + runtime_api::client_support::ClientError::Failure( + runtime_api::client_support::ClientFailure::ResponseTooLarge { .. } + ) + )); + server.abort(); + let response = authed_json_request( app.clone(), Method::POST, @@ -3344,9 +3637,12 @@ mod tests { ) .await; assert_eq!(response.status(), StatusCode::OK); - let restored: RuntimeHttpWorkerRestoreResponse = read_json(response).await; + let restored: runtime_api::WorkerRestoreResponse = read_json(response).await; assert_eq!(restored.state, WorkerRestoreState::Accepted); - assert_eq!(restored.worker.unwrap().status, WorkerStatus::Idle); + assert_eq!( + restored.worker.unwrap().status, + runtime_api::WorkerStatus::Idle + ); let response = authed_empty_request( app.clone(), @@ -3406,11 +3702,11 @@ mod tests { #[test] fn ssh_probe_uses_workdir_operation_capability() { assert_eq!( - required_runtime_permission(&Method::POST, SSH_HOST_KEY_PROBE_PATH), + required_runtime_permission(&Method::POST, runtime_api::RUNTIME_ROUTE_SSH_PROBE), Some(SSH_HOST_KEY_PROBE_OPERATION) ); assert_eq!( - workspace_runtime_operation(&Method::POST, SSH_HOST_KEY_PROBE_PATH), + workspace_runtime_operation(&Method::POST, runtime_api::RUNTIME_ROUTE_SSH_PROBE), SSH_HOST_KEY_PROBE_OPERATION ); } @@ -3427,7 +3723,7 @@ mod tests { let response = authed_json_request( app, Method::POST, - SSH_HOST_KEY_PROBE_PATH, + runtime_api::RUNTIME_ROUTE_SSH_PROBE, token, &SshHostKeyProbeRequest { hostname: "-oProxyCommand=malicious".to_string(), @@ -3492,8 +3788,14 @@ mod tests { .unwrap(); assert_eq!(unauthenticated.status(), StatusCode::UNAUTHORIZED); - let response = - authed_json_request(app, Method::POST, SSH_HOST_KEY_PROBE_PATH, token, &body).await; + let response = authed_json_request( + app, + Method::POST, + runtime_api::RUNTIME_ROUTE_SSH_PROBE, + token, + &body, + ) + .await; assert_eq!(response.status(), StatusCode::OK); assert_eq!( std::fs::read_to_string(recorded_arguments).unwrap(), diff --git a/crates/worker-runtime/src/http_server/runtime_management_api.rs b/crates/worker-runtime/src/http_server/runtime_management_api.rs new file mode 100644 index 000000000..6547ecb43 --- /dev/null +++ b/crates/worker-runtime/src/http_server/runtime_management_api.rs @@ -0,0 +1,476 @@ +use super::*; + +use std::future::Future; + +const MANAGEMENT_BODY_LIMIT: usize = 8 * 1024 * 1024; + +// Axum request extensions remain the authorization authority. The generated adapter deliberately +// has no authentication model, so the verified extension is scoped across its service call. +tokio::task_local! { + static AUTH_CONTEXT: Option; +} + +pub(super) async fn scope_auth(context: Option, future: F) -> F::Output +where + F: Future, +{ + AUTH_CONTEXT.scope(context, future).await +} + +fn auth_extension() -> Option> { + AUTH_CONTEXT + .try_with(Clone::clone) + .ok() + .flatten() + .map(Extension) +} + +fn required_auth() -> Result, runtime_api::RuntimeApiError> { + auth_extension().ok_or_else(|| { + runtime_api::RuntimeApiError::new( + StatusCode::FORBIDDEN.as_u16(), + "workspace_scope_required", + "Runtime operation requires an authorization context", + ) + }) +} + +#[derive(Clone)] +struct RuntimeManagementApi { + state: RuntimeHttpState, +} + +pub(super) fn router(state: RuntimeHttpState) -> Router { + runtime_api::RuntimeApiAxum::router(RuntimeManagementApi { state }) + .layer(axum::extract::DefaultBodyLimit::max(MANAGEMENT_BODY_LIMIT)) + .layer(middleware::from_fn(runtime_api_wire_compatibility)) +} + +async fn runtime_api_wire_compatibility( + mut request: axum::extract::Request, + next: Next, +) -> Response { + 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, + axum::http::HeaderValue::from_static("application/json"), + ); + } + let (parts, body) = request.into_parts(); + let body = match axum::body::to_bytes(body, MANAGEMENT_BODY_LIMIT).await { + Ok(body) if body.is_empty() => Bytes::from_static(b"{}"), + Ok(body) => body, + Err(_) => { + return RuntimeHttpRestError::new( + StatusCode::PAYLOAD_TOO_LARGE, + "request_body_too_large", + "request body exceeds Runtime management limit", + ) + .into_response(); + } + }; + request = axum::extract::Request::from_parts(parts, Body::from(body)); + } + let response = next.run(request).await; + 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, + 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(); + } + response +} + +fn request(value: T) -> Result +where + T: Serialize, + U: serde::de::DeserializeOwned, +{ + serde_json::to_value(value) + .and_then(serde_json::from_value) + .map_err(|error| { + runtime_api::RuntimeApiError::new( + StatusCode::BAD_REQUEST.as_u16(), + "invalid_request", + error.to_string(), + ) + }) +} + +fn response(value: T) -> Result +where + T: Serialize, + U: serde::de::DeserializeOwned, +{ + serde_json::to_value(value) + .and_then(serde_json::from_value) + .map_err(|error| { + runtime_api::RuntimeApiError::new( + StatusCode::INTERNAL_SERVER_ERROR.as_u16(), + "runtime_response_conversion_failed", + error.to_string(), + ) + }) +} + +fn api_error(error: RuntimeHttpRestError) -> runtime_api::RuntimeApiError { + runtime_api::RuntimeApiError::new(error.status.as_u16(), error.code, error.message) +} + +impl runtime_api::RuntimeApi for RuntimeManagementApi { + async fn ping( + &self, + workspace_id: String, + ) -> Result { + 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::SERVICE_UNAVAILABLE.as_u16(), + "runtime_ping_identity_unavailable", + "Runtime ping identity is not configured", + ) + })?; + Ok(runtime_api::RuntimePingResponse { + runtime_id: runtime_id.to_string(), + protocol_version: RUNTIME_HTTP_PROTOCOL_VERSION, + }) + } + + async fn runtime_summary( + &self, + ) -> Result { + let runtime = self + .state + .runtime + .summary() + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error)?; + let runtime = response(runtime)?; + Ok(runtime_api::RuntimeSummaryResponse { runtime }) + } + + async fn list_workers( + &self, + query: runtime_api::WorkerListQuery, + ) -> Result { + 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)?; + let workers = response(workers)?; + Ok(runtime_api::WorkersResponse { workers }) + } + + async fn get_worker( + &self, + worker_id: String, + ) -> Result { + 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 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)?; + let worker = response(worker)?; + Ok(runtime_api::WorkerResponse { worker }) + } + + async fn delete_worker( + &self, + worker_id: String, + ) -> Result { + 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( + &self, + worker_id: String, + input: runtime_api::WorkerInput, + ) -> Result { + 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)?; + let ack = response(ack)?; + Ok(runtime_api::WorkerInputResponse { ack }) + } + + async fn stop_worker( + &self, + worker_id: String, + value: runtime_api::WorkerLifecycleRequest, + ) -> Result { + 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)?; + let ack = response(ack)?; + Ok(runtime_api::WorkerLifecycleResponse { ack }) + } + + async fn cancel_worker( + &self, + worker_id: String, + value: runtime_api::WorkerLifecycleRequest, + ) -> Result { + 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)?; + let ack = response(ack)?; + Ok(runtime_api::WorkerLifecycleResponse { ack }) + } + + async fn restore_worker( + &self, + worker_id: String, + _request: runtime_api::EmptyObjectRequest, + ) -> Result { + let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?; + let result = match auth_workspace_scope(&self.state, auth_extension().as_ref()) + .map_err(api_error)? + { + Some(scope) => self + .state + .runtime + .restore_worker_operation_scoped(&scope, &worker_ref), + None => self.state.runtime.restore_worker_operation(&worker_ref), + } + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error)?; + Ok(runtime_api::WorkerRestoreResponse { + state: result.state, + worker: result.worker.map(response).transpose()?, + reason_code: result.reason_code, + message: result.message, + }) + } + + async fn replace_worker_workspace_api( + &self, + worker_id: String, + value: runtime_api::WorkerWorkspaceApiRequest, + ) -> Result { + 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)?; + let worker = response(worker)?; + Ok(runtime_api::WorkerResponse { worker }) + } + + async fn complete_worker_arguments( + &self, + worker_id: String, + value: runtime_api::CompletionRequest, + ) -> Result { + 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)?; + let entries = response(entries)?; + Ok(runtime_api::CompletionResponse { + kind, + prefix, + entries, + }) + } + + async fn retention_inventory( + &self, + worker_id: String, + ) -> Result { + 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( + &self, + worker_id: String, + value: runtime_api::WorkerRetentionExecutionRequest, + ) -> Result { + 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::NOT_FOUND.as_u16(), + "worker_not_found", + "worker does not belong to the authenticated Workspace", + )); + } + 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/Cargo.toml b/crates/workspace-server/Cargo.toml index e4e589409..f0069c05f 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -23,6 +23,7 @@ config-source.workspace = true flow = { path = "../flow" } manifest.workspace = true protocol = { workspace = true } +runtime-api.workspace = true project-record.workspace = true reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "native-tls"] } ring.workspace = true diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 1e71e2ab2..a3b13e52a 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -15,7 +15,6 @@ use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::{ - error::Error as _, future::Future, io::Read as _, net::{IpAddr, SocketAddr, ToSocketAddrs}, @@ -47,15 +46,9 @@ use worker_runtime::http_server::{ RUNTIME_PING_PERMISSION, RUNTIME_WORKSPACE_SCOPE_HEADER, RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest, RuntimeHttpErrorResponse, RuntimeHttpPingResponse, RuntimeHttpRepositoryAccessResponse, - RuntimeHttpSummaryResponse, RuntimeHttpUploadedFileDeleteResponse, - RuntimeHttpUploadedFileResponse, RuntimeHttpWorkerCompletionsRequest, - RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse, - RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest, - RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, - RuntimeHttpWorkerRestoreResponse, RuntimeHttpWorkerWorkspaceApiRequest, - RuntimeHttpWorkersResponse, RuntimeHttpWorkingDirectoriesResponse, - RuntimeHttpWorkingDirectoryResponse, RuntimeHttpWorkspacePromptProjectionRequest, - RuntimeHttpWorkspacePromptProjectionResponse, + RuntimeHttpUploadedFileDeleteResponse, RuntimeHttpUploadedFileResponse, + RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse, + RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse, }; use worker_runtime::identity::{ RuntimeWorkerRef, WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef, @@ -3411,49 +3404,7 @@ pub struct RemoteWorkerRuntime { resource_broker: BackendResourceBroker, http: BlockingHttpClient, async_http: AsyncHttpClient, -} - -fn remote_runtime_ping_transport_failure(error: reqwest::Error) -> RuntimePingFailure { - if error.is_timeout() { - return RuntimePingFailure::new( - RuntimePingFailureKind::Timeout, - "runtime_ping_timeout", - "Runtime ping timed out", - ); - } - let mut source = error.source(); - let mut tls_error = false; - while let Some(current) = source { - let message = current.to_string().to_ascii_lowercase(); - if message.contains("tls") - || message.contains("certificate") - || message.contains("unknownissuer") - || message.contains("handshake") - { - tls_error = true; - break; - } - source = current.source(); - } - if tls_error { - return RuntimePingFailure::new( - RuntimePingFailureKind::TlsOrTransport, - "runtime_ping_tls_failed", - "Runtime TLS connection failed", - ); - } - if error.is_connect() { - return RuntimePingFailure::new( - RuntimePingFailureKind::NetworkUnreachable, - "runtime_ping_network_unreachable", - "Runtime could not be reached", - ); - } - RuntimePingFailure::new( - RuntimePingFailureKind::TlsOrTransport, - "runtime_ping_transport_failed", - "Runtime ping transport failed", - ) + request_timeout: Duration, } fn workspace_runtime_operation(method: &str, path_and_query: &str) -> &'static str { @@ -3519,6 +3470,54 @@ fn worker_id_from_remote_path(path_and_query: &str) -> Option { (!worker_id.is_empty()).then(|| worker_id.to_string()) } +#[derive(Clone)] +struct RemoteRuntimeApiAuthorizer { + workspace_id: String, + bearer_token: Option, + workspace_authorization: Option, +} + +impl runtime_api::client_support::RequestAuthorizer for RemoteRuntimeApiAuthorizer { + fn authorize( + &self, + request: runtime_api::client_support::AuthorizerRequest<'_>, + ) -> Result { + let mut headers = reqwest::header::HeaderMap::new(); + headers.insert( + "x-yoi-workspace-id", + self.workspace_id + .parse() + .map_err(|_| runtime_api::client_support::AuthorizationError::new())?, + ); + let authorization = if let Some(authorizer) = &self.workspace_authorization { + let operation = + workspace_runtime_operation(request.method.as_str(), request.path_and_query); + let worker_id = worker_id_from_remote_path(request.path_and_query); + let token = authorizer + .issue( + request.method.as_str(), + request.path_and_query, + operation, + worker_id.as_deref(), + request.body, + ) + .map_err(|_| runtime_api::client_support::AuthorizationError::new())?; + format!("Bearer {token}") + } else if let Some(token) = self.bearer_token.as_deref() { + format!("Bearer {token}") + } else { + return Ok(headers); + }; + headers.insert( + AUTHORIZATION, + authorization + .parse() + .map_err(|_| runtime_api::client_support::AuthorizationError::new())?, + ); + Ok(headers) + } +} + impl RemoteWorkerRuntime { pub fn new( config: RemoteRuntimeConfig, @@ -3590,6 +3589,7 @@ impl RemoteWorkerRuntime { resource_broker: BackendResourceBroker::default(), http, async_http, + request_timeout: timeout, }) } @@ -3623,6 +3623,72 @@ impl RemoteWorkerRuntime { .await } + fn runtime_api_client( + &self, + timeout: Duration, + response_body_limit: usize, + ) -> Result, RuntimeDiagnostic> { + runtime_api::RuntimeApiClient::builder(&self.base_url) + .map_err(|error| { + diagnostic( + "remote_runtime_client_build_failed", + DiagnosticSeverity::Error, + error.to_string(), + ) + })? + .client(self.async_http.clone()) + .authorizer(RemoteRuntimeApiAuthorizer { + workspace_id: self.workspace_id.clone(), + bearer_token: self.bearer_token.clone(), + workspace_authorization: self.workspace_authorization.clone(), + }) + .request_timeout(timeout) + .response_body_limit(response_body_limit) + .build() + .map_err(|error| { + diagnostic( + "remote_runtime_client_build_failed", + DiagnosticSeverity::Error, + error.to_string(), + ) + }) + } + + fn run_runtime_api( + &self, + timeout: Duration, + response_body_limit: usize, + operation: F, + ) -> Result + where + T: Send + 'static, + F: FnOnce(runtime_api::RuntimeApiClient) -> Fut + + Send + + 'static, + Fut: std::future::Future< + Output = Result< + T, + runtime_api::client_support::ClientError, + >, + > + Send + + 'static, + { + let client = self.runtime_api_client(timeout, response_body_limit)?; + let runtime_id = self.runtime_id.clone(); + run_blocking_http(move || { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|_| { + runtime_api::client_support::ClientError::failure( + runtime_api::client_support::ClientFailure::Transport, + ) + })? + .block_on(operation(client)) + }) + .map_err(|error| runtime_api_diagnostic(&runtime_id, error)) + } + fn endpoint(&self, path: &str) -> String { format!("{}{}", self.base_url, path) } @@ -3753,88 +3819,6 @@ impl RemoteWorkerRuntime { self.send_json(path, "DELETE", &[], self.http.delete(self.endpoint(path))) } - fn ping_http(&self) -> Result { - const PATH: &str = "/v1/ping"; - let workspace_id = self.workspace_id.clone(); - let bearer_token = self.bearer_token.clone(); - let capability_token = match &self.workspace_authorization { - Some(authorization) => Some( - authorization - .issue("GET", PATH, RUNTIME_PING_PERMISSION, None, &[]) - .map_err(|diagnostic| { - RuntimePingFailure::new( - RuntimePingFailureKind::Authentication, - diagnostic.code, - diagnostic.message, - ) - })?, - ), - None => None, - }; - let request = self - .http - .get(self.endpoint(PATH)) - .header(RUNTIME_WORKSPACE_SCOPE_HEADER, &workspace_id); - run_blocking_http(move || { - let request = match capability_token.as_deref().or(bearer_token.as_deref()) { - Some(token) => request.header(AUTHORIZATION, format!("Bearer {token}")), - None => request, - }; - let response = request - .send() - .map_err(remote_runtime_ping_transport_failure)?; - match response.status() { - StatusCode::UNAUTHORIZED => { - return Err(RuntimePingFailure::new( - RuntimePingFailureKind::Authentication, - "runtime_ping_authentication_failed", - "Runtime rejected the connection-test credential", - )); - } - StatusCode::FORBIDDEN => { - return Err(RuntimePingFailure::new( - RuntimePingFailureKind::Authorization, - "runtime_ping_authorization_failed", - "Runtime rejected the connection-test scope or permission", - )); - } - status if !status.is_success() => { - return Err(RuntimePingFailure::new( - RuntimePingFailureKind::TlsOrTransport, - "runtime_ping_http_failed", - "Runtime ping returned an unsuccessful HTTP response", - )); - } - _ => {} - } - let mut body = Vec::new(); - response - .take((MAX_RUNTIME_PING_RESPONSE_BYTES + 1) as u64) - .read_to_end(&mut body) - .map_err(|_| { - RuntimePingFailure::new( - RuntimePingFailureKind::TlsOrTransport, - "runtime_ping_response_read_failed", - "Runtime ping response could not be read", - ) - })?; - if body.len() > MAX_RUNTIME_PING_RESPONSE_BYTES { - return Err(RuntimePingFailure::new( - RuntimePingFailureKind::MalformedResponse, - "runtime_ping_response_too_large", - "Runtime ping response exceeded the allowed size", - )); - } - serde_json::from_slice::(&body).map_err(|_| { - RuntimePingFailure::new( - RuntimePingFailureKind::MalformedResponse, - "runtime_ping_malformed_response", - "Runtime ping returned an unrecognized response", - ) - }) - }) - } - fn send_json( &self, path: &str, @@ -4003,7 +3987,7 @@ impl RemoteWorkerRuntime { fn lifecycle_result_from_response( &self, worker_id: &str, - response: RuntimeHttpWorkerLifecycleResponse, + response: worker_runtime::catalog::WorkerLifecycleAck, ) -> WorkerLifecycleResult { WorkerLifecycleResult { state: WorkerOperationState::Accepted, @@ -4013,7 +3997,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) ), )], } @@ -4026,24 +4010,33 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } fn runtime_summary(&self, limit: usize) -> RuntimeSummary { - match self.get_json::("/v1/runtime") { + match self + .run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + |client| async move { client.runtime_summary().await }, + ) + .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 { @@ -4066,7 +4059,50 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } fn ping(&self) -> Result { - self.ping_http() + let workspace_id = self.workspace_id.clone(); + self.run_runtime_api( + self.request_timeout, + MAX_RUNTIME_PING_RESPONSE_BYTES, + move |client| async move { client.ping(workspace_id).await }, + ) + .and_then(runtime_contract_convert) + .map_err(|diagnostic| { + let kind = if diagnostic.code.contains("auth") { + RuntimePingFailureKind::Authentication + } else if diagnostic.code.contains("network_unreachable") { + RuntimePingFailureKind::NetworkUnreachable + } else if diagnostic.code.contains("timeout") { + RuntimePingFailureKind::Timeout + } else if diagnostic.code.contains("invalid_response") + || diagnostic.code.contains("response_too_large") + { + RuntimePingFailureKind::MalformedResponse + } else { + RuntimePingFailureKind::TlsOrTransport + }; + let (code, message) = match kind { + RuntimePingFailureKind::Authentication => ( + "runtime_ping_authentication_failed", + "Runtime rejected ping authentication", + ), + RuntimePingFailureKind::NetworkUnreachable => ( + "runtime_ping_network_unreachable", + "Runtime could not be reached", + ), + RuntimePingFailureKind::Timeout => { + ("runtime_ping_timeout", "Runtime ping timed out") + } + RuntimePingFailureKind::MalformedResponse => ( + "runtime_ping_malformed_response", + "Runtime ping returned a malformed response", + ), + _ => ( + "runtime_ping_transport_failed", + "Runtime ping transport failed", + ), + }; + RuntimePingFailure::new(kind, code, message) + }) } fn activate_workspace_authorization(&self, binding: crate::store::WorkspaceRuntimeBinding) { @@ -4124,10 +4160,23 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { if limit == 0 { return RuntimeList::new(Vec::new(), Vec::new()); } - match self.get_json::("/v1/workers") { + match self + .run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + |client| async move { + client + .list_workers(runtime_api::WorkerListQuery::default()) + .await + }, + ) + .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)) @@ -4142,10 +4191,25 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { if limit == 0 { return RuntimeList::new(Vec::new(), Vec::new()); } - match self.get_json::("/v1/workers?status=stopped") { + match self + .run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + |client| async move { + client + .list_workers(runtime_api::WorkerListQuery { + status: Some(runtime_api::WorkerStatusFilter::Stopped), + }) + .await + }, + ) + .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)) @@ -4157,9 +4221,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } fn worker(&self, worker_id: &str) -> WorkerLookupResult { - match self.get_json::(&format!("/v1/workers/{worker_id}")) { + 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.get_worker(worker_id_owned).await }, + ) + .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 { @@ -4174,20 +4247,34 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult { - match self.post_json::<_, RuntimeHttpWorkerRestoreResponse>( - &format!("/v1/workers/{worker_id}/restore"), - &serde_json::json!({}), - ) { - Ok(response) => { - let diagnostics = match (response.reason_code, response.message) { + 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 + .restore_worker(worker_id_owned, runtime_api::EmptyObjectRequest::default()) + .await + }, + ) + .and_then(|value| { + let worker = value + .worker + .map(runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>) + .transpose()?; + Ok((value.state, worker, value.reason_code, value.message)) + }) { + Ok((state, worker, reason_code, message)) => { + let diagnostics = match (reason_code, message) { (Some(code), Some(message)) => { vec![diagnostic(code, DiagnosticSeverity::Warning, message)] } _ => Vec::new(), }; WorkerRestoreResult { - state: response.state, - worker: response.worker.map(|worker| self.map_worker_detail(worker)), + state, + worker: worker.map(|worker| self.map_worker_detail(worker)), diagnostics, } } @@ -4206,13 +4293,33 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { worker_id: &str, workspace_api: WorkspaceApiRef, ) -> WorkerWorkspaceApiResult { - match self.post_json::<_, RuntimeHttpWorkerResponse>( - &format!("/v1/workers/{worker_id}/workspace-api"), - &RuntimeHttpWorkerWorkspaceApiRequest { workspace_api }, - ) { + let request = match runtime_contract_convert(workspace_api) { + Ok(workspace_api) => runtime_api::WorkerWorkspaceApiRequest { workspace_api }, + Err(diagnostic) => { + return WorkerWorkspaceApiResult { + state: WorkerOperationState::Rejected, + worker: None, + diagnostics: vec![diagnostic], + }; + } + }; + let worker_id = worker_id.to_string(); + match self + .run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { + client + .replace_worker_workspace_api(worker_id, request) + .await + }, + ) + .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 { @@ -4409,34 +4516,29 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { workspace_api: Some(workspace_api), memory_settings: request.resolved_memory_settings.clone(), }; - let create_body = match serde_json::to_vec(&create) { - Ok(body) => body, - Err(error) => { + let create = match runtime_contract_convert(create) { + Ok(create) => create, + Err(diagnostic) => { return WorkerSpawnResult { state: WorkerOperationState::Rejected, worker: None, acceptance_evidence: Vec::new(), - diagnostics: vec![diagnostic( - "remote_runtime_request_encode_failed", - DiagnosticSeverity::Error, - error.to_string(), - )], + diagnostics: vec![diagnostic], }; } }; - match self.send_json::( - "/v1/workers", - "POST", - &create_body, - self.http - .post(self.endpoint("/v1/workers")) - .timeout(REMOTE_WORKER_CREATE_TIMEOUT) - .header(CONTENT_TYPE, "application/json") - .body(create_body.clone()), - ) { + match self + .run_runtime_api( + REMOTE_WORKER_CREATE_TIMEOUT, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { client.create_worker(create).await }, + ) + .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(), @@ -4504,13 +4606,21 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { worker_id: &str, request: WorkerLifecycleRequest, ) -> WorkerLifecycleResult { - let body = RuntimeHttpWorkerLifecycleRequest { + let body = runtime_api::WorkerLifecycleRequest { reason: request.reason, }; - match self.post_json::<_, RuntimeHttpWorkerLifecycleResponse>( - &format!("/v1/workers/{worker_id}/stop"), - &body, - ) { + 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.stop_worker(worker_id_owned, body).await }, + ) + .and_then(|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), } @@ -4521,29 +4631,46 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { worker_id: &str, request: WorkerLifecycleRequest, ) -> WorkerLifecycleResult { - let body = RuntimeHttpWorkerLifecycleRequest { + let body = runtime_api::WorkerLifecycleRequest { reason: request.reason, }; - match self.post_json::<_, RuntimeHttpWorkerLifecycleResponse>( - &format!("/v1/workers/{worker_id}/cancel"), - &body, - ) { + 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.cancel_worker(worker_id_owned, body).await }, + ) + .and_then(|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), } } fn delete_worker(&self, worker_id: &str) -> WorkerDeleteResult { + let worker_id_owned = worker_id.to_string(); match self - .delete_json::(&format!("/v1/workers/{worker_id}")) - { + .run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { client.delete_worker(worker_id_owned).await }, + ) + .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 { @@ -4559,9 +4686,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { &self, worker_id: &str, ) -> Result { - self.get_json::(&format!( - "/v1/workers/{worker_id}/retention/inventory" - )) + let worker_id = worker_id.to_string(); + self.run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { client.retention_inventory(worker_id).await }, + ) + .and_then(runtime_contract_convert) .map_err(|diagnostic| diagnostic.message) } @@ -4570,10 +4701,14 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { request: WorkerRetentionExecutionRequest, ) -> Result { let worker_id = request.worker_id.to_string(); - self.post_json::<_, WorkerRetentionExecutionResult>( - &format!("/v1/workers/{worker_id}/retention/execute"), - &request, + let request = runtime_contract_convert(request).map_err(|diagnostic| diagnostic.message)?; + let worker_id = worker_id.to_string(); + self.run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { client.execute_retention(worker_id, request).await }, ) + .and_then(runtime_contract_convert) .map_err(|diagnostic| diagnostic.message) } @@ -4610,9 +4745,21 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { submission_request_id: None, segments: request.segments, }; - match self.post_json::<_, RuntimeHttpWorkerInputResponse>( - &format!("/v1/workers/{worker_id}/input"), - &input, + let input = match runtime_contract_convert(input) { + Ok(input) => input, + Err(diagnostic) => { + return WorkerInputResult { + state: WorkerOperationState::Rejected, + worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), + diagnostics: vec![diagnostic], + }; + } + }; + 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.send_worker_input(worker_id_owned, input).await }, ) { Ok(_) => WorkerInputResult { state: WorkerOperationState::Accepted, @@ -4680,13 +4827,21 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { worker_id: &str, request: WorkerCompletionsRequest, ) -> WorkerCompletionsResult { - let request = RuntimeHttpWorkerCompletionsRequest { + let request = runtime_api::CompletionRequest { kind: request.kind, prefix: request.prefix, }; - match self.post_json::<_, RuntimeHttpWorkerCompletionsResponse>( - &format!("/v1/workers/{worker_id}/completions"), - &request, + 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 + }, ) { Ok(response) => WorkerCompletionsResult { worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), @@ -4697,8 +4852,8 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { }, Err(diagnostic) => WorkerCompletionsResult { worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()), - kind: request.kind, - prefix: request.prefix, + kind: failure_kind, + prefix: failure_prefix, entries: Vec::new(), diagnostics: vec![diagnostic], }, @@ -5211,6 +5366,84 @@ fn remote_reqwest_diagnostic(runtime_id: &str, err: reqwest::Error) -> RuntimeDi } } +fn runtime_contract_convert(value: T) -> Result +where + T: Serialize, + U: DeserializeOwned, +{ + serde_json::to_value(value) + .and_then(serde_json::from_value) + .map_err(|error| { + diagnostic( + "remote_runtime_contract_conversion_failed", + DiagnosticSeverity::Error, + error.to_string(), + ) + }) +} + +fn runtime_api_diagnostic( + runtime_id: &str, + error: runtime_api::client_support::ClientError, +) -> RuntimeDiagnostic { + match error { + runtime_api::client_support::ClientError::Public { status, error } => { + let remote_code = error.error.code; + let message = sanitize_remote_runtime_message(&remote_code, &error.error.message); + let (code, severity) = match status { + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => ( + "remote_runtime_auth_failed".to_string(), + DiagnosticSeverity::Error, + ), + _ => ( + remote_code, + if status.is_server_error() { + DiagnosticSeverity::Error + } else { + DiagnosticSeverity::Warning + }, + ), + }; + diagnostic( + code, + severity, + format!( + "Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {message}" + ), + ) + } + runtime_api::client_support::ClientError::Failure(failure) => { + let code = match failure { + runtime_api::client_support::ClientFailure::Timeout => "remote_runtime_timeout", + runtime_api::client_support::ClientFailure::Connect => { + "remote_runtime_network_unreachable" + } + runtime_api::client_support::ClientFailure::Authorization => { + "remote_runtime_authorization_failed" + } + runtime_api::client_support::ClientFailure::ErrorResponseDecode { status } + if status == StatusCode::UNAUTHORIZED.as_u16() + || status == StatusCode::FORBIDDEN.as_u16() => + { + "remote_runtime_auth_failed" + } + runtime_api::client_support::ClientFailure::ResponseTooLarge { .. } => { + "remote_runtime_response_too_large" + } + runtime_api::client_support::ClientFailure::Decode { .. } => { + "remote_runtime_invalid_response" + } + _ => "remote_runtime_client_error", + }; + diagnostic( + code, + DiagnosticSeverity::Error, + format!("Remote Runtime client error for '{runtime_id}': {failure}"), + ) + } + } +} + fn sanitize_remote_runtime_message(code: &str, message: &str) -> String { if message.contains('/') || message.contains('\\') { format!("remote Runtime returned {code}; backend-private details were omitted")