Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

Merge commit '5e9f7a7dc3d1169118412376f16b72850aab63f2' into work/T-593-workspace-runtime-bindings

Browse Source 
# Conflicts:
#	crates/workspace-server/src/server.rs
This commit is contained in:
Hare
2026-09-06 03:46:09 +09:00
parent 51c6d7f835 5e9f7a7dc3
commit 88fad3893e
11 changed files with 1067 additions and 589 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/http_server.rs
+136 -1
View File
@@ -33,7 +33,7 @@ use axum::extract::rejection::{JsonRejection, QueryRejection};
#[cfg(feature = "ws-server")]
use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade};
use axum::extract::{DefaultBodyLimit, Extension, Path, Query, State};
use axum::http::{Method, Request, StatusCode, header};
use axum::http::{HeaderMap, Method, Request, StatusCode, header};
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Response};
use axum::routing::{delete, get, post};
@@ -66,6 +66,11 @@ use workdir::{
};
const DEFAULT_RUNTIME_HTTP_PORT: u16 = 38800;
pub const RUNTIME_HTTP_PROTOCOL_MIN_VERSION: u32 = 1;
pub const RUNTIME_HTTP_PROTOCOL_MAX_VERSION: u32 = 1;
pub const RUNTIME_HTTP_PROTOCOL_VERSION: u32 = RUNTIME_HTTP_PROTOCOL_MAX_VERSION;
pub const RUNTIME_PING_PERMISSION: &str = "runtime:ping";
pub const RUNTIME_WORKSPACE_SCOPE_HEADER: &str = "x-yoi-workspace-id";
fn default_runtime_http_bind_addr() -> SocketAddr {
SocketAddr::from(([127, 0, 0, 1], DEFAULT_RUNTIME_HTTP_PORT))
@@ -187,6 +192,7 @@ fn runtime_http_router_with_optional_auth(
};
let router = Router::new()
.route("/v1/ping", get(get_runtime_ping))
.route("/v1/runtime", get(get_runtime))
.route(
"/v1/config-bundles",
@@ -340,6 +346,14 @@ enum RuntimeHttpWorkerStatusFilter {
Stopped,
}
/// `GET /v1/ping` response.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RuntimeHttpPingResponse {
pub runtime_id: String,
pub protocol_version: u32,
}
/// `GET /v1/workers` response.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkersResponse {
@@ -461,6 +475,48 @@ struct RuntimeWorkerEventsWsQuery {
type RestResult<T> = Result<Json<T>, RuntimeHttpRestError>;
async fn get_runtime_ping(
State(state): State<RuntimeHttpState>,
Extension(auth): Extension<RuntimeAuthContext>,
headers: HeaderMap,
) -> RestResult<RuntimeHttpPingResponse> {
let requested_workspace_id = headers
.get(RUNTIME_WORKSPACE_SCOPE_HEADER)
.and_then(|value| value.to_str().ok())
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| {
RuntimeHttpRestError::new(
StatusCode::FORBIDDEN,
"runtime_ping_workspace_scope_required",
"Runtime ping requires the target Workspace scope",
)
})?;
if requested_workspace_id != auth.workspace_id {
return Err(RuntimeHttpRestError::new(
StatusCode::FORBIDDEN,
"runtime_ping_workspace_scope_mismatch",
"Runtime ping Workspace scope does not match the authenticated capability",
));
}
let runtime_id = state
.auth
.as_ref()
.map(|config| config.runtime_id.trim())
.filter(|runtime_id| !runtime_id.is_empty())
.ok_or_else(|| {
RuntimeHttpRestError::new(
StatusCode::SERVICE_UNAVAILABLE,
"runtime_ping_identity_unavailable",
"Runtime ping identity is not configured",
)
})?;
Ok(Json(RuntimeHttpPingResponse {
runtime_id: runtime_id.to_string(),
protocol_version: RUNTIME_HTTP_PROTOCOL_VERSION,
}))
}
async fn get_runtime(
State(state): State<RuntimeHttpState>,
) -> RestResult<RuntimeHttpSummaryResponse> {
@@ -1767,6 +1823,9 @@ fn auth_workspace_scope(
}
fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static str> {
if path == "/v1/ping" && *method == Method::GET {
return Some(RUNTIME_PING_PERMISSION);
}
if path == "/v1/runtime" {
return None;
}
@@ -2084,6 +2143,82 @@ mod tests {
WorkdirPath, WorkdirSessionCapabilities,
};
#[tokio::test]
async fn ping_requires_scoped_permission_and_returns_versioned_identity() {
let runtime = Runtime::new_memory();
let (auth, signer) = auth_config_and_signer();
let app = runtime_http_router_with_auth(runtime, None, auth);
let token =
token_for_workspace_with_permissions(&signer, "workspace-a", [RUNTIME_PING_PERMISSION]);
let request = Request::builder()
.method(Method::GET)
.uri("/v1/ping")
.header(header::AUTHORIZATION, format!("Bearer {token}"))
.header(RUNTIME_WORKSPACE_SCOPE_HEADER, "workspace-a")
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert_eq!(
serde_json::from_slice::<RuntimeHttpPingResponse>(&body).unwrap(),
RuntimeHttpPingResponse {
runtime_id: "runtime-test".to_string(),
protocol_version: RUNTIME_HTTP_PROTOCOL_VERSION,
}
);
let wrong_scope_token =
token_for_workspace_with_permissions(&signer, "workspace-a", [RUNTIME_PING_PERMISSION]);
let wrong_scope_request = Request::builder()
.method(Method::GET)
.uri("/v1/ping")
.header(header::AUTHORIZATION, format!("Bearer {wrong_scope_token}"))
.header(RUNTIME_WORKSPACE_SCOPE_HEADER, "workspace-b")
.body(Body::empty())
.unwrap();
assert_eq!(
app.oneshot(wrong_scope_request).await.unwrap().status(),
StatusCode::FORBIDDEN
);
}
#[tokio::test]
async fn ping_rejects_token_without_ping_permission() {
let runtime = Runtime::new_memory();
let (auth, signer) = auth_config_and_signer();
let app = runtime_http_router_with_auth(runtime, None, auth);
let missing_credential = Request::builder()
.method(Method::GET)
.uri("/v1/ping")
.header(RUNTIME_WORKSPACE_SCOPE_HEADER, "workspace-a")
.body(Body::empty())
.unwrap();
assert_eq!(
app.clone()
.oneshot(missing_credential)
.await
.unwrap()
.status(),
StatusCode::UNAUTHORIZED
);
let token = token_for_workspace_with_permissions(&signer, "workspace-a", ["workers:read"]);
let request = Request::builder()
.method(Method::GET)
.uri("/v1/ping")
.header(header::AUTHORIZATION, format!("Bearer {token}"))
.header(RUNTIME_WORKSPACE_SCOPE_HEADER, "workspace-a")
.body(Body::empty())
.unwrap();
assert_eq!(
app.oneshot(request).await.unwrap().status(),
StatusCode::FORBIDDEN
);
}
#[test]
fn attachment_routes_require_worker_input_permission() {
assert_eq!(
crates/workspace-api/src/lib.rs
+56 -6
View File
@@ -1205,16 +1205,39 @@ pub struct CreateRemoteRuntimeRequest {
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(rename_all = "snake_case")]
pub enum RuntimeConnectionTestStatus {
Compatible,
Failed,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(rename_all = "snake_case")]
pub enum RuntimeConnectionTestFailureKind {
Authentication,
Authorization,
NetworkUnreachable,
Timeout,
TlsOrTransport,
MalformedResponse,
ProtocolVersionMismatch,
RuntimeIdentityMismatch,
Configuration,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct RuntimeConnectionTestResponse {
pub workspace_id: String,
pub runtime_id: String,
pub checked_at: String,
pub state: String,
pub protocol_version: Option<String>,
pub compatibility_basis: String,
#[serde(default)]
pub capabilities: Vec<String>,
pub health_result: String,
pub status: RuntimeConnectionTestStatus,
pub failure_kind: Option<RuntimeConnectionTestFailureKind>,
pub expected_protocol_version: u32,
pub actual_protocol_version: Option<u32>,
#[serde(default)]
pub diagnostics: Vec<Diagnostic>,
}
@@ -2371,6 +2394,9 @@ pub fn catalog_typescript() -> String {
RepositoryListResponse::decl(&config),
RepositoryDetailResponse::decl(&config),
RepositoryLogResponse::decl(&config),
RuntimeConnectionTestStatus::decl(&config),
RuntimeConnectionTestFailureKind::decl(&config),
RuntimeConnectionTestResponse::decl(&config),
]
.map(|declaration| format!("export {declaration}"));
@@ -3060,6 +3086,27 @@ mod tests {
assert!(serde_json::from_value::<RepositoryListResponse>(stale).is_err());
}
#[test]
fn runtime_connection_test_response_is_closed_and_typed() {
let compatible = serde_json::json!({
"workspace_id": "workspace-test",
"runtime_id": "runtime-test",
"checked_at": "2026-09-01T12:00:00Z",
"status": "compatible",
"failure_kind": null,
"expected_protocol_version": 1,
"actual_protocol_version": 1,
"diagnostics": []
});
let parsed: RuntimeConnectionTestResponse =
serde_json::from_value(compatible.clone()).unwrap();
assert_eq!(serde_json::to_value(parsed).unwrap(), compatible);
let mut unknown = compatible;
unknown["capabilities"] = serde_json::json!(["shell"]);
assert!(serde_json::from_value::<RuntimeConnectionTestResponse>(unknown).is_err());
}
#[cfg(feature = "typescript")]
#[test]
fn generated_catalog_typescript_keeps_public_wrappers_and_nullability() {
@@ -3080,6 +3127,9 @@ mod tests {
assert!(output.contains(
"export type WorkspaceProfileSourceProvenance = \"project_profile_source_tree\""
));
assert!(output.contains("export type RuntimeConnectionTestResponse ="));
assert!(output.contains("status: RuntimeConnectionTestStatus"));
assert!(output.contains("failure_kind: RuntimeConnectionTestFailureKind | null"));
assert!(!output.contains("repository_key: string, display_name"));
}
crates/workspace-server/src/hosts.rs
+215 -9
View File
@@ -9,7 +9,9 @@ use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::{
error::Error as _,
future::Future,
io::Read as _,
path::PathBuf,
pin::Pin,
sync::{Arc, RwLock},
@@ -38,13 +40,15 @@ use worker_runtime::error::RuntimeError as EmbeddedRuntimeError;
use worker_runtime::execution::WorkerExecutionRunState;
use worker_runtime::fs_store::FsRuntimeStoreOptions;
use worker_runtime::http_server::{
RUNTIME_PING_PERMISSION, RUNTIME_WORKSPACE_SCOPE_HEADER,
RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest,
RuntimeHttpErrorResponse, RuntimeHttpRepositoryAccessResponse, RuntimeHttpSummaryResponse,
RuntimeHttpUploadedFileDeleteResponse, RuntimeHttpUploadedFileResponse,
RuntimeHttpWorkerCompletionsRequest, RuntimeHttpWorkerCompletionsResponse,
RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerInputResponse,
RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse,
RuntimeHttpWorkerResponse, RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
RuntimeHttpErrorResponse, RuntimeHttpPingResponse, RuntimeHttpRepositoryAccessResponse,
RuntimeHttpSummaryResponse, RuntimeHttpUploadedFileDeleteResponse,
RuntimeHttpUploadedFileResponse, RuntimeHttpWorkerCompletionsRequest,
RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse,
RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest,
RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse,
RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse,
};
@@ -64,6 +68,7 @@ pub const EMBEDDED_RUNTIME_ID: &str = "embedded-worker-runtime";
const EMBEDDED_HOST_KIND: &str = "embedded-worker-runtime-host";
const REMOTE_HOST_KIND: &str = "remote-worker-runtime-host";
const MAX_DIAGNOSTICS: usize = 16;
const MAX_RUNTIME_PING_RESPONSE_BYTES: usize = 8 * 1024;
const MAX_HOST_SCAN: usize = 256;
const MAX_IDENTIFIER_LEN: usize = 120;
const ID_DIGEST_HEX_LEN: usize = 16;
@@ -760,11 +765,50 @@ fn default_worker_input_kind() -> WorkerInputKind {
WorkerInputKind::User
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimePingFailureKind {
Authentication,
Authorization,
NetworkUnreachable,
Timeout,
TlsOrTransport,
MalformedResponse,
Configuration,
Unsupported,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimePingFailure {
pub kind: RuntimePingFailureKind,
pub diagnostic: RuntimeDiagnostic,
}
impl RuntimePingFailure {
fn new(
kind: RuntimePingFailureKind,
code: impl Into<String>,
message: impl Into<String>,
) -> Self {
Self {
kind,
diagnostic: diagnostic(code, DiagnosticSeverity::Error, message.into()),
}
}
}
pub trait WorkspaceWorkerRuntime: Send + Sync {
fn runtime_id(&self) -> &str;
fn runtime_summary(&self, limit: usize) -> RuntimeSummary;
fn ping(&self) -> Result<RuntimeHttpPingResponse, RuntimePingFailure> {
Err(RuntimePingFailure::new(
RuntimePingFailureKind::Unsupported,
"runtime_ping_unsupported",
"Runtime connection testing is unavailable for this Runtime provider",
))
}
fn list_hosts(&self, limit: usize) -> RuntimeList<HostSummary>;
fn list_workers(&self, limit: usize) -> RuntimeList<WorkerSummary>;
@@ -1811,6 +1855,17 @@ impl RuntimeRegistry {
})
}
pub fn ping(&self, runtime_id: &str) -> Result<RuntimeHttpPingResponse, RuntimePingFailure> {
let runtime = self.runtime(runtime_id).map_err(|_| {
RuntimePingFailure::new(
RuntimePingFailureKind::Configuration,
"runtime_ping_registration_unavailable",
"Registered Runtime binding is unavailable",
)
})?;
runtime.ping()
}
fn runtimes_snapshot(&self) -> Vec<Arc<dyn WorkspaceWorkerRuntime>> {
self.runtimes
.read()
@@ -2927,6 +2982,49 @@ pub struct RemoteWorkerRuntime {
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",
)
}
fn all_remote_runtime_permissions() -> Vec<String> {
[
"workers:list",
@@ -3075,14 +3173,18 @@ impl RemoteWorkerRuntime {
self.send_json(path, self.http.delete(self.endpoint(path)))
}
fn runtime_capability_token(&self, path: &str) -> Option<String> {
fn runtime_capability_token_with_permissions(
&self,
path: &str,
permissions: Vec<String>,
) -> Option<String> {
let auth = self.auth.as_ref()?;
let signer = CapabilityTokenSigner::new(&auth.server_id, &auth.server_private_key);
let claims = capability_claims(
&auth.server_id,
&self.runtime_id,
&self.workspace_id,
all_remote_runtime_permissions(),
permissions,
300,
)
.map_err(|error| {
@@ -3105,6 +3207,82 @@ impl RemoteWorkerRuntime {
.ok()
}
fn runtime_capability_token(&self, path: &str) -> Option<String> {
self.runtime_capability_token_with_permissions(path, all_remote_runtime_permissions())
}
fn ping_http(&self) -> Result<RuntimeHttpPingResponse, RuntimePingFailure> {
const PATH: &str = "/v1/ping";
let workspace_id = self.workspace_id.clone();
let bearer_token = self.bearer_token.clone();
let capability_token = self.runtime_capability_token_with_permissions(
PATH,
vec![RUNTIME_PING_PERMISSION.to_string()],
);
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::<RuntimeHttpPingResponse>(&body).map_err(|_| {
RuntimePingFailure::new(
RuntimePingFailureKind::MalformedResponse,
"runtime_ping_malformed_response",
"Runtime ping returned an unrecognized response",
)
})
})
}
fn send_json<T>(&self, path: &str, request: RequestBuilder) -> Result<T, RuntimeDiagnostic>
where
T: DeserializeOwned + Send + 'static,
@@ -3292,6 +3470,10 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
}
}
fn ping(&self) -> Result<RuntimeHttpPingResponse, RuntimePingFailure> {
self.ping_http()
}
fn list_hosts(&self, limit: usize) -> RuntimeList<HostSummary> {
if limit == 0 {
return RuntimeList::new(Vec::new(), Vec::new());
@@ -4588,7 +4770,7 @@ mod tests {
use super::*;
use serde_json::json;
use std::collections::HashMap;
use std::io::{Read as _, Write as _};
use std::io::Write as _;
use std::net::TcpListener;
use std::sync::{Arc, Mutex};
use std::thread;
@@ -5791,6 +5973,30 @@ mod tests {
assert_eq!(runtime.runtime_id(), "remote:async-init");
}
#[test]
fn remote_runtime_ping_classifies_unreachable_without_endpoint_leak() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let endpoint = format!("http://{}", listener.local_addr().unwrap());
drop(listener);
let runtime = RemoteWorkerRuntime::new(
RemoteRuntimeConfig::new(
"remote:unreachable",
"Remote Unreachable",
endpoint.clone(),
Some("secret-token".to_string()),
),
"workspace-test".to_string(),
"http://127.0.0.1:8787".to_string(),
)
.unwrap();
let failure = runtime.ping().unwrap_err();
assert_eq!(failure.kind, RuntimePingFailureKind::NetworkUnreachable);
assert_eq!(failure.diagnostic.code, "runtime_ping_network_unreachable");
assert!(!failure.diagnostic.message.contains(&endpoint));
assert!(!format!("{failure:?}").contains("secret-token"));
}
#[test]
fn remote_runtime_registry_routes_commands_without_browser_secret_leaks() {
let worker_id = EmbeddedWorkerId::from_legacy_u64(1).to_string();
crates/workspace-server/src/server.rs
+349 -543
View File
@@ -53,6 +53,10 @@ use workdir::workspace::{
};
use workdir::{CommandHandle, WorkdirSessionHandle};
use worker::feature::builtin::{WorkerObservationSubject, WorkerObservationSubjectRef};
use worker_runtime::http_server::{
RUNTIME_HTTP_PROTOCOL_MAX_VERSION, RUNTIME_HTTP_PROTOCOL_MIN_VERSION,
RUNTIME_HTTP_PROTOCOL_VERSION,
};
use worker_runtime::resource::{BackendResourceError, BackendResourceFetchRequest};
use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend};
use workspace_api::{
@@ -72,7 +76,8 @@ use workspace_api::{
PasskeyRegistrationOptionsResponse, ProfileSettingsResponse, PutRepositorySshHostTrustRequest,
RepositoryAccessProjection, RepositoryDetailResponse, RepositoryListResponse,
RepositoryLogResponse, RepositorySshCredential, RepositorySshHostTrust, RequestActor,
RotateRepositorySshCredentialRequest, RuntimeConnectionTestResponse, RuntimeManagementSummary,
RotateRepositorySshCredentialRequest, RuntimeConnectionTestFailureKind,
RuntimeConnectionTestResponse, RuntimeConnectionTestStatus, RuntimeManagementSummary,
TICKET_ORCHESTRATION_PLANS_QUERY_PATH, TICKET_RELATIONS_QUERY_PATH,
UpdateWorkspaceMetadataRequest, WhoamiResponse, WorkerLaunchOptionsResponse,
WorkerLaunchProfileCandidate, WorkerLaunchRuntimeOption, WorkerLaunchWorkerSummary,
@@ -106,14 +111,14 @@ use crate::config_source::ConfigCommitRequest;
use crate::hosts::{
ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EMBEDDED_RUNTIME_ID,
EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime,
RuntimeDiagnostic, RuntimeRegistry, RuntimeRegistryError, RuntimeRegistryUnregisterResult,
TicketWorkerRole, WorkerCapabilitySummary, WorkerCompletionsRequest, WorkerCompletionsResult,
WorkerControlOperation, WorkerCreateBinding, WorkerImplementationSummary, WorkerInputKind,
WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, WorkerLifecycleResult,
WorkerOperationState, WorkerRestoreResult, WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent,
WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, WorkerSummary,
WorkerTicketAssignmentRequest, WorkerWorkspaceSummary, worker_spawn_create_fingerprint,
workspace_worker_summary,
RuntimeDiagnostic, RuntimePingFailureKind, RuntimeRegistry, RuntimeRegistryError,
RuntimeRegistryUnregisterResult, TicketWorkerRole, WorkerCapabilitySummary,
WorkerCompletionsRequest, WorkerCompletionsResult, WorkerControlOperation, WorkerCreateBinding,
WorkerImplementationSummary, WorkerInputKind, WorkerInputRequest, WorkerInputResult,
WorkerLifecycleRequest, WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult,
WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult,
WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerTicketAssignmentRequest,
WorkerWorkspaceSummary, worker_spawn_create_fingerprint, workspace_worker_summary,
};
use crate::identity::WorkspaceIdentity;
use crate::memory_backend::execute_memory_backend_operation_with_authority;
@@ -148,7 +153,7 @@ use crate::store::{
RepositoryRecord, TicketAssignmentPrincipal, TicketAssignmentRole, TicketCoderAssignmentRecord,
TicketRoleAssignmentRecord, UserRecord, WorkdirCreateOperationRecord, WorkdirRegistryRecord,
WorkerControlGrantRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord, WorkspaceRecord,
WorkspaceResourceKind, WorkspaceRuntimeBinding,
WorkspaceResourceKind,
};
use crate::workdir_removal::{
WorkdirRemovalAttemptOwner, WorkdirRemovalDisposition, WorkdirRemovalOperation,
@@ -163,11 +168,7 @@ use worker_runtime::catalog::{
WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
};
use worker_runtime::config_bundle::ConfigBundle;
use worker_runtime::http_server::{
MAX_WORKER_FILE_UPLOAD_BYTES, RuntimeHttpConfigBundleAvailabilityResponse,
RuntimeHttpConfigBundlesResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerResponse,
RuntimeHttpWorkersResponse,
};
use worker_runtime::http_server::MAX_WORKER_FILE_UPLOAD_BYTES;
use worker_runtime::identity::{RuntimeWorkerRef, WorkerId};
const EMBEDDED_WORKER_RUNTIME_ID: &str = "embedded-worker-runtime";
@@ -12513,13 +12514,37 @@ async fn test_runtime_connection(
State(api): State<WorkspaceApi>,
AxumPath(runtime_id): AxumPath<String>,
) -> ApiResult<Json<RuntimeConnectionTestResponse>> {
let binding = api
.store
.get_workspace_runtime_binding(&api.config.workspace_id, &runtime_id)
let runtime_id = runtime_id.trim().to_string();
if runtime_id.is_empty() {
return Err(Error::InvalidRuntimeIdentifier {
kind: "runtime".to_string(),
value: runtime_id,
}
.into());
}
api.store
.get_workspace_runtime_binding(api.workspace_id(), &runtime_id)
.await?
.filter(|binding| binding.revoked_at.is_none())
.ok_or_else(|| Error::UnknownRuntime(runtime_id.clone()))?;
Ok(Json(test_remote_runtime_binding(&api, &binding).await))
let checked_at = Utc::now().to_rfc3339();
let runtime = api.runtime.clone();
let ping_runtime_id = runtime_id.clone();
let ping = tokio::task::spawn_blocking(move || runtime.ping(&ping_runtime_id))
.await
.map_err(|_| Error::RuntimeOperationFailed {
runtime_id: runtime_id.clone(),
code: "runtime_connection_test_unavailable".to_string(),
message: "Runtime connection test could not be completed".to_string(),
})?;
Ok(Json(runtime_connection_test_response(
api.workspace_id(),
&runtime_id,
checked_at,
ping,
)))
}
async fn get_worker_launch_options(
@@ -14520,415 +14545,111 @@ fn remote_runtime_config_from_binding(
Ok(remote)
}
async fn test_remote_runtime_binding(
api: &WorkspaceApi,
remote: &WorkspaceRuntimeBinding,
) -> RuntimeConnectionTestResponse {
let checked_at = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
let client = match reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(5))
.build()
{
Ok(client) => client,
Err(_) => {
return remote_runtime_test_failed(
api,
remote,
checked_at,
"remote_runtime_test_client_unavailable",
"Remote Runtime test client could not be initialized.",
);
}
};
let mut observation = RuntimeCompatibilityObservation::default();
let summary_url = match remote_probe_url(remote, "/v1/runtime") {
Ok(url) => url,
Err(diagnostic) => {
return remote_runtime_test_failed(
api,
remote,
checked_at,
diagnostic.code,
diagnostic.message,
);
}
};
let summary_payload =
match probe_remote_json(&client, summary_url, "runtime.summary", "Runtime summary").await {
Ok(payload) => payload,
Err(diagnostic) => {
return remote_runtime_test_failed(
api,
remote,
checked_at,
diagnostic.code,
diagnostic.message,
);
}
};
let protocol_version = summary_payload
.get("protocol_version")
.and_then(|value| value.as_str())
.map(ToOwned::to_owned);
let summary = match serde_json::from_value::<RuntimeHttpSummaryResponse>(summary_payload) {
Ok(summary) => summary,
Err(_) => {
return remote_runtime_test_failed(
api,
remote,
checked_at,
"remote_runtime_malformed_summary",
"Remote Runtime summary responded, but the payload was not recognized.",
);
}
};
observation.available(
"runtime.summary",
"Connected: /v1/runtime responded with a recognized worker-runtime summary.",
);
let workers_url = match remote_probe_url(remote, "/v1/workers") {
Ok(url) => url,
Err(diagnostic) => {
observation.incompatible("workers.list", diagnostic);
String::new()
}
};
let workers = if workers_url.is_empty() {
None
} else {
match probe_remote_json(&client, workers_url, "workers.list", "Worker list").await {
Ok(payload) => match serde_json::from_value::<RuntimeHttpWorkersResponse>(payload) {
Ok(workers) => {
observation.available(
"workers.list",
"Verified: /v1/workers responded with a recognized worker list.",
);
Some(workers)
}
Err(_) => {
observation.incompatible(
"workers.list",
settings_diagnostic(
"remote_runtime_workers_malformed",
DiagnosticSeverity::Error,
"Remote Runtime worker list responded, but the payload was not recognized.",
),
);
None
}
},
Err(diagnostic) => {
observation.incompatible("workers.list", diagnostic);
None
}
}
};
if let Some(worker) = workers.as_ref().and_then(|workers| workers.workers.first()) {
let path = format!(
"/v1/workers/{}",
encode_path_segment(&worker.worker_id.to_string())
);
match remote_probe_url(remote, &path) {
Ok(url) => match probe_remote_json(&client, url, "workers.detail", "Worker detail").await {
Ok(payload) => match serde_json::from_value::<RuntimeHttpWorkerResponse>(payload) {
Ok(_) => observation.available(
"workers.detail",
"Verified: worker detail responded for an existing worker reported by the remote Runtime.",
),
Err(_) => observation.incompatible(
"workers.detail",
settings_diagnostic(
"remote_runtime_worker_detail_malformed",
DiagnosticSeverity::Error,
"Remote Runtime worker detail responded, but the payload was not recognized.",
),
),
},
Err(diagnostic) => observation.incompatible("workers.detail", diagnostic),
},
Err(diagnostic) => observation.incompatible("workers.detail", diagnostic),
}
} else {
observation.unknown(
"workers.detail",
"No connection problem found. Worker detail was not checked because the remote Runtime reported no workers during the lightweight probe.",
);
}
observation.available(
"workers.events_ws.construct",
"Verified: worker event websocket URL can be constructed from the configured HTTP(S) Runtime endpoint. The lightweight test does not open a websocket stream.",
);
let bundles_url = match remote_probe_url(remote, "/v1/config-bundles") {
Ok(url) => url,
Err(diagnostic) => {
observation.incompatible("config_bundles.list", diagnostic);
String::new()
}
};
let bundles = if bundles_url.is_empty() {
None
} else {
match probe_remote_json(
&client,
bundles_url,
"config_bundles.list",
"Config-bundle list",
)
.await
{
Ok(payload) => {
match serde_json::from_value::<RuntimeHttpConfigBundlesResponse>(payload) {
Ok(bundles) => {
observation.available(
"config_bundles.list",
"Verified: /v1/config-bundles responded with a recognized config-bundle list.",
);
Some(bundles)
}
Err(_) => {
observation.incompatible(
"config_bundles.list",
settings_diagnostic(
"remote_runtime_config_bundles_malformed",
DiagnosticSeverity::Error,
"Remote Runtime config-bundle list responded, but the payload was not recognized.",
),
);
None
}
}
}
Err(diagnostic) => {
observation.incompatible("config_bundles.list", diagnostic);
None
}
}
};
if let Some(bundle) = bundles.as_ref().and_then(|bundles| bundles.bundles.first()) {
let path = format!(
"/v1/config-bundles/{}/availability?digest={}",
encode_path_segment(&bundle.id),
encode_path_segment(&bundle.digest)
);
match remote_probe_url(remote, &path) {
Ok(url) => match probe_remote_json(
&client,
url,
"config_bundles.availability",
"Config-bundle availability",
)
.await
{
Ok(payload) => {
match serde_json::from_value::<RuntimeHttpConfigBundleAvailabilityResponse>(payload)
{
Ok(_) => observation.available(
"config_bundles.availability",
"Verified: config-bundle availability was confirmed for an advertised bundle.",
),
Err(_) => observation.incompatible(
"config_bundles.availability",
settings_diagnostic(
"remote_runtime_config_bundle_availability_malformed",
DiagnosticSeverity::Error,
"Remote Runtime config-bundle availability responded, but the payload was not recognized.",
),
),
}
}
Err(diagnostic) => {
observation.incompatible("config_bundles.availability", diagnostic)
}
},
Err(diagnostic) => observation.incompatible("config_bundles.availability", diagnostic),
}
} else {
observation.unknown(
"config_bundles.availability",
"No connection problem found. Config-bundle availability was not checked because the remote Runtime advertised no bundles during the lightweight probe.",
);
}
if summary.runtime.worker_creation_available {
observation.available(
"workers.spawn",
"Verified: /v1/runtime reports worker creation is enabled by a Runtime execution backend. The lightweight test does not create a worker.",
);
} else {
observation.incompatible(
"workers.spawn",
settings_diagnostic(
"remote_runtime_worker_creation_unavailable",
DiagnosticSeverity::Error,
"Connected to the Runtime, but worker creation is unavailable because this Runtime process has no execution backend attached.",
),
);
}
observation.unknown(
"workers.input_dispatch",
"No connection problem found. Worker input dispatch was not checked because this lightweight test does not send model-visible input as a side effect.",
);
observation.unknown(
"config_bundles.sync",
"No connection problem found. Config-bundle sync was not checked because this lightweight test does not upload bundles as a side effect.",
);
RuntimeConnectionTestResponse {
workspace_id: api.config.workspace_id.clone(),
runtime_id: remote.runtime_id.clone(),
checked_at,
state: observation.state().to_string(),
protocol_version,
compatibility_basis: "Connected to /v1/runtime and verified non-side-effecting worker-runtime HTTP endpoints. No incompatible operation was found; warning items below are unproven optional or side-effecting checks, not connection failures.".to_string(),
capabilities: observation.capabilities,
health_result: format!(
"connected=true; runtime_status={:?}; available={}; incompatible={}; warnings={}",
summary.runtime.status,
observation.available_count,
observation.incompatible_count,
observation.unknown_count
),
diagnostics: observation
.diagnostics
.into_iter()
.map(Into::into)
.collect(),
}
}
fn remote_runtime_test_failed(
api: &WorkspaceApi,
remote: &WorkspaceRuntimeBinding,
fn runtime_connection_test_response(
workspace_id: &str,
runtime_id: &str,
checked_at: String,
code: impl Into<String>,
message: impl Into<String>,
ping: std::result::Result<
worker_runtime::http_server::RuntimeHttpPingResponse,
crate::hosts::RuntimePingFailure,
>,
) -> RuntimeConnectionTestResponse {
match ping {
Ok(ping) if ping.runtime_id != runtime_id => runtime_connection_test_failure(
workspace_id,
runtime_id,
checked_at,
RuntimeConnectionTestFailureKind::RuntimeIdentityMismatch,
None,
RuntimeDiagnostic::new(
"runtime_ping_identity_mismatch",
"error",
"Runtime ping identity does not match the registered Runtime",
),
),
Ok(ping)
if !(RUNTIME_HTTP_PROTOCOL_MIN_VERSION..=RUNTIME_HTTP_PROTOCOL_MAX_VERSION)
.contains(&ping.protocol_version) =>
{
let code = if ping.protocol_version > RUNTIME_HTTP_PROTOCOL_MAX_VERSION {
"runtime_ping_protocol_newer"
} else {
"runtime_ping_protocol_older"
};
runtime_connection_test_failure(
workspace_id,
runtime_id,
checked_at,
RuntimeConnectionTestFailureKind::ProtocolVersionMismatch,
Some(ping.protocol_version),
RuntimeDiagnostic::new(
code,
"error",
"Runtime protocol version is incompatible with this Server",
),
)
}
Ok(ping) => RuntimeConnectionTestResponse {
workspace_id: workspace_id.to_string(),
runtime_id: runtime_id.to_string(),
checked_at,
status: RuntimeConnectionTestStatus::Compatible,
failure_kind: None,
expected_protocol_version: RUNTIME_HTTP_PROTOCOL_VERSION,
actual_protocol_version: Some(ping.protocol_version),
diagnostics: Vec::new(),
},
Err(failure) => runtime_connection_test_failure(
workspace_id,
runtime_id,
checked_at,
match failure.kind {
RuntimePingFailureKind::Authentication => {
RuntimeConnectionTestFailureKind::Authentication
}
RuntimePingFailureKind::Authorization => {
RuntimeConnectionTestFailureKind::Authorization
}
RuntimePingFailureKind::NetworkUnreachable => {
RuntimeConnectionTestFailureKind::NetworkUnreachable
}
RuntimePingFailureKind::Timeout => RuntimeConnectionTestFailureKind::Timeout,
RuntimePingFailureKind::TlsOrTransport => {
RuntimeConnectionTestFailureKind::TlsOrTransport
}
RuntimePingFailureKind::MalformedResponse => {
RuntimeConnectionTestFailureKind::MalformedResponse
}
RuntimePingFailureKind::Configuration | RuntimePingFailureKind::Unsupported => {
RuntimeConnectionTestFailureKind::Configuration
}
},
None,
failure.diagnostic,
),
}
}
fn runtime_connection_test_failure(
workspace_id: &str,
runtime_id: &str,
checked_at: String,
failure_kind: RuntimeConnectionTestFailureKind,
actual_protocol_version: Option<u32>,
diagnostic: RuntimeDiagnostic,
) -> RuntimeConnectionTestResponse {
RuntimeConnectionTestResponse {
workspace_id: api.config.workspace_id.clone(),
runtime_id: remote.runtime_id.clone(),
workspace_id: workspace_id.to_string(),
runtime_id: runtime_id.to_string(),
checked_at,
state: "failed".to_string(),
protocol_version: None,
compatibility_basis: "worker-runtime lightweight HTTP compatibility probes".to_string(),
capabilities: Vec::new(),
health_result: "failed".to_string(),
diagnostics: vec![settings_diagnostic(code, DiagnosticSeverity::Error, message).into()],
status: RuntimeConnectionTestStatus::Failed,
failure_kind: Some(failure_kind),
expected_protocol_version: RUNTIME_HTTP_PROTOCOL_VERSION,
actual_protocol_version,
diagnostics: vec![diagnostic.into()],
}
}
#[derive(Default)]
struct RuntimeCompatibilityObservation {
capabilities: Vec<String>,
diagnostics: Vec<RuntimeDiagnostic>,
available_count: usize,
incompatible_count: usize,
unknown_count: usize,
}
impl RuntimeCompatibilityObservation {
fn available(&mut self, operation: &str, message: impl Into<String>) {
self.available_count += 1;
self.capabilities.push(format!("{operation}:available"));
self.diagnostics.push(settings_diagnostic(
format!("{operation}.available"),
DiagnosticSeverity::Info,
message,
));
}
fn unknown(&mut self, operation: &str, message: impl Into<String>) {
self.unknown_count += 1;
self.capabilities.push(format!("{operation}:unknown"));
self.diagnostics.push(settings_diagnostic(
format!("{operation}.unknown"),
DiagnosticSeverity::Warning,
message,
));
}
fn incompatible(&mut self, operation: &str, diagnostic: RuntimeDiagnostic) {
self.incompatible_count += 1;
self.capabilities.push(format!("{operation}:incompatible"));
self.diagnostics.push(diagnostic);
}
fn state(&self) -> &'static str {
if self.incompatible_count > 0 {
"incompatible"
} else {
"compatible"
}
}
}
fn remote_probe_url(
remote: &WorkspaceRuntimeBinding,
path: &str,
) -> std::result::Result<String, RuntimeDiagnostic> {
let endpoint = remote.base_url.trim();
if !(endpoint.starts_with("http://") || endpoint.starts_with("https://")) {
return Err(settings_diagnostic(
"remote_runtime_endpoint_invalid",
DiagnosticSeverity::Error,
"Configured remote Runtime endpoint is not an absolute HTTP(S) URL.",
));
}
Ok(format!("{}{}", endpoint.trim_end_matches('/'), path))
}
async fn probe_remote_json(
client: &reqwest::Client,
url: String,
operation: &'static str,
label: &'static str,
) -> std::result::Result<serde_json::Value, RuntimeDiagnostic> {
let response = client.get(url).send().await.map_err(|error| {
let (code, message) = if error.is_timeout() {
(
format!("{operation}.timeout"),
format!("Remote Runtime probe for {label} timed out."),
)
} else if error.is_connect() {
(
format!("{operation}.connect_failed"),
format!("Remote Runtime probe for {label} could not connect."),
)
} else {
(
format!("{operation}.request_failed"),
format!("Remote Runtime probe for {label} failed before a response was received."),
)
};
settings_diagnostic(code, DiagnosticSeverity::Error, message)
})?;
if !response.status().is_success() {
return Err(settings_diagnostic(
format!("{operation}.http_status"),
DiagnosticSeverity::Error,
format!(
"Remote Runtime probe for {label} returned HTTP status {}.",
response.status().as_u16()
),
));
}
response.json::<serde_json::Value>().await.map_err(|_| {
settings_diagnostic(
format!("{operation}.malformed_json"),
DiagnosticSeverity::Error,
format!("Remote Runtime probe for {label} returned an unrecognized JSON payload."),
)
})
}
fn worker_launch_options_response(api: &WorkspaceApi) -> ApiResult<WorkerLaunchOptionsResponse> {
let runtimes = api
.runtime
@@ -24314,6 +24035,90 @@ mod tests {
axum::serve(listener, proxy).await
}
async fn runtime_ping_stub(
status: StatusCode,
body: serde_json::Value,
) -> (String, tokio::task::JoinHandle<()>) {
async fn ping(
State((status, body)): State<(StatusCode, serde_json::Value)>,
headers: HeaderMap,
) -> (StatusCode, Json<serde_json::Value>) {
assert_eq!(
headers
.get(worker_runtime::http_server::RUNTIME_WORKSPACE_SCOPE_HEADER)
.and_then(|value| value.to_str().ok()),
Some(TEST_WORKSPACE_ID)
);
assert!(
headers
.get(axum::http::header::AUTHORIZATION)
.and_then(|value| value.to_str().ok())
.is_some_and(|value| value.starts_with("Bearer "))
);
(status, Json(body))
}
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind ping stub");
let base_url = format!("http://{}", listener.local_addr().expect("ping stub addr"));
let app = Router::new()
.route("/v1/ping", axum::routing::get(ping))
.with_state((status, body));
let server = tokio::spawn(async move {
axum::serve(listener, app).await.expect("serve ping stub");
});
(base_url, server)
}
async fn test_app_with_remote_runtime(
workspace_root: impl Into<PathBuf>,
runtime_id: &str,
endpoint: String,
) -> Router {
let api = test_api(workspace_root).await;
api.store
.upsert_workspace_runtime_binding_record(
WorkspaceRuntimeBinding {
workspace_id: TEST_WORKSPACE_ID.to_string(),
runtime_id: runtime_id.to_string(),
display_name: "Probe Runtime".to_string(),
base_url: endpoint.clone(),
public_key: RuntimeIdentityMaterial::generate(runtime_id)
.unwrap()
.public_key,
public_key_fingerprint: String::new(),
created_at: "1".to_string(),
updated_at: "1".to_string(),
revoked_at: None,
},
false,
)
.await
.unwrap();
api.runtime.register_or_replace(
RemoteWorkerRuntime::new(
RemoteRuntimeConfig {
runtime_id: runtime_id.to_string(),
workspace_id: Some(TEST_WORKSPACE_ID.to_string()),
display_name: "Probe Runtime".to_string(),
base_url: endpoint,
bearer_token: Some("test-connection-token".to_string()),
auth: None,
cached_worker_creation_available: true,
cached_os: "linux".to_string(),
cached_arch: "x86_64".to_string(),
cached_status: "active".to_string(),
timeout: std::time::Duration::from_secs(2),
},
TEST_WORKSPACE_ID.to_string(),
"http://127.0.0.1:1".to_string(),
)
.unwrap(),
);
build_inner_router(api)
}
async fn test_app(workspace_root: impl Into<PathBuf>) -> Router {
build_inner_router(test_api(workspace_root).await)
}
@@ -25599,140 +25404,141 @@ mod tests {
);
}
#[tokio::test(flavor = "multi_thread")]
async fn runtime_connection_test_reports_compatible_with_unknown_warnings_without_endpoint_leak()
{
let (runtime, _worker_ref) = runtime_with_worker();
let runtime_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let runtime_addr = runtime_listener.local_addr().unwrap();
tokio::spawn({
let runtime = runtime.clone();
async move {
serve_runtime_http_with_injected_test_auth(runtime, runtime_listener)
.await
.unwrap()
}
});
async fn run_runtime_connection_test(
body: serde_json::Value,
status: StatusCode,
) -> serde_json::Value {
let (endpoint, _server) = runtime_ping_stub(status, body).await;
let dir = tempfile::tempdir().unwrap();
let endpoint = format!("http://{runtime_addr}");
let api = test_api(dir.path()).await;
api.store
.upsert_workspace_runtime_binding_record(
WorkspaceRuntimeBinding {
workspace_id: TEST_WORKSPACE_ID.to_string(),
runtime_id: "probe-runtime".to_string(),
display_name: "Probe Runtime".to_string(),
base_url: endpoint.clone(),
public_key: RuntimeIdentityMaterial::generate("probe-runtime")
.unwrap()
.public_key,
public_key_fingerprint: String::new(),
created_at: "2026-01-01T00:00:00Z".to_string(),
updated_at: "2026-01-01T00:00:00Z".to_string(),
revoked_at: None,
},
false,
)
.await
.unwrap();
let app = build_inner_router(api);
let response = post_json(
let app = test_app_with_remote_runtime(dir.path(), "probe-runtime", endpoint).await;
post_json(
app,
&format!("/api/w/{TEST_WORKSPACE_ID}/runtimes/probe-runtime/connection-tests"),
serde_json::json!({}),
)
.await;
assert_eq!(response["state"], "compatible");
let capabilities = response["capabilities"].as_array().unwrap();
assert!(
capabilities
.iter()
.any(|value| value == "runtime.summary:available")
);
assert!(
capabilities
.iter()
.any(|value| value == "workers.list:available")
);
assert!(
capabilities
.iter()
.any(|value| value == "workers.spawn:available")
);
assert!(
response["diagnostics"]
.as_array()
.unwrap()
.iter()
.any(|diagnostic| { diagnostic["code"] == "workers.spawn.available" })
);
let projected = serde_json::to_string(&response).unwrap();
assert!(!projected.contains(&endpoint));
assert!(!projected.contains(&runtime_addr.to_string()));
assert_eq!(response["protocol_version"], serde_json::Value::Null);
.await
}
#[tokio::test(flavor = "multi_thread")]
async fn runtime_connection_test_marks_missing_execution_backend_incompatible() {
let runtime =
worker_runtime::Runtime::with_options(worker_runtime::RuntimeOptions::default());
let runtime_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let runtime_addr = runtime_listener.local_addr().unwrap();
tokio::spawn(async move {
serve_runtime_http_with_injected_test_auth(runtime, runtime_listener)
.await
.unwrap()
});
let dir = tempfile::tempdir().unwrap();
let endpoint = format!("http://{runtime_addr}");
let api = test_api(dir.path()).await;
api.store
.upsert_workspace_runtime_binding_record(
WorkspaceRuntimeBinding {
workspace_id: TEST_WORKSPACE_ID.to_string(),
runtime_id: "control-only-runtime".to_string(),
display_name: "Control-only Runtime".to_string(),
base_url: endpoint,
public_key: RuntimeIdentityMaterial::generate("control-only-runtime")
.unwrap()
.public_key,
public_key_fingerprint: String::new(),
created_at: "2026-01-01T00:00:00Z".to_string(),
updated_at: "2026-01-01T00:00:00Z".to_string(),
revoked_at: None,
},
false,
)
.await
.unwrap();
let app = build_inner_router(api);
let response = post_json(
app,
&format!("/api/w/{TEST_WORKSPACE_ID}/runtimes/control-only-runtime/connection-tests"),
serde_json::json!({}),
async fn runtime_connection_test_reports_exact_compatible_protocol() {
let response = run_runtime_connection_test(
serde_json::json!({
"runtime_id": "probe-runtime",
"protocol_version": RUNTIME_HTTP_PROTOCOL_VERSION,
}),
StatusCode::OK,
)
.await;
assert_eq!(response["state"], "incompatible");
assert!(
response["capabilities"]
.as_array()
.unwrap()
.iter()
.any(|value| { value == "workers.spawn:incompatible" })
assert_eq!(response["status"], "compatible");
assert_eq!(response["failure_kind"], serde_json::Value::Null);
assert_eq!(
response["expected_protocol_version"],
RUNTIME_HTTP_PROTOCOL_VERSION
);
assert!(
response["diagnostics"]
.as_array()
.unwrap()
.iter()
.any(|diagnostic| {
diagnostic["code"] == "remote_runtime_worker_creation_unavailable"
})
assert_eq!(
response["actual_protocol_version"],
RUNTIME_HTTP_PROTOCOL_VERSION
);
assert_eq!(response["diagnostics"], serde_json::json!([]));
let projected = serde_json::to_string(&response).unwrap();
assert!(!projected.contains("Bearer"));
assert!(!projected.contains("public_key"));
}
#[tokio::test(flavor = "multi_thread")]
async fn runtime_connection_test_rejects_newer_protocol() {
let newer = RUNTIME_HTTP_PROTOCOL_MAX_VERSION + 1;
let response = run_runtime_connection_test(
serde_json::json!({
"runtime_id": "probe-runtime",
"protocol_version": newer,
}),
StatusCode::OK,
)
.await;
assert_eq!(response["status"], "failed");
assert_eq!(response["failure_kind"], "protocol_version_mismatch");
assert_eq!(response["actual_protocol_version"], newer);
assert_eq!(
response["diagnostics"][0]["code"],
"runtime_ping_protocol_newer"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn runtime_connection_test_rejects_older_protocol() {
let older = RUNTIME_HTTP_PROTOCOL_MIN_VERSION.saturating_sub(1);
let response = run_runtime_connection_test(
serde_json::json!({
"runtime_id": "probe-runtime",
"protocol_version": older,
}),
StatusCode::OK,
)
.await;
assert_eq!(response["status"], "failed");
assert_eq!(response["failure_kind"], "protocol_version_mismatch");
assert_eq!(response["actual_protocol_version"], older);
assert_eq!(
response["diagnostics"][0]["code"],
"runtime_ping_protocol_older"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn runtime_connection_test_classifies_authentication_failure() {
let response = run_runtime_connection_test(
serde_json::json!({"error": "credential details must not escape"}),
StatusCode::UNAUTHORIZED,
)
.await;
assert_eq!(response["status"], "failed");
assert_eq!(response["failure_kind"], "authentication");
assert_eq!(response["actual_protocol_version"], serde_json::Value::Null);
let projected = serde_json::to_string(&response).unwrap();
assert!(!projected.contains("credential details"));
}
#[tokio::test(flavor = "multi_thread")]
async fn runtime_connection_test_rejects_malformed_ping_response() {
let response = run_runtime_connection_test(
serde_json::json!({
"runtime_id": "probe-runtime",
"protocol_version": "not-a-number",
"unexpected": true,
}),
StatusCode::OK,
)
.await;
assert_eq!(response["status"], "failed");
assert_eq!(response["failure_kind"], "malformed_response");
assert_eq!(
response["diagnostics"][0]["code"],
"runtime_ping_malformed_response"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn runtime_connection_test_rejects_runtime_identity_mismatch() {
let response = run_runtime_connection_test(
serde_json::json!({
"runtime_id": "different-runtime",
"protocol_version": RUNTIME_HTTP_PROTOCOL_VERSION,
}),
StatusCode::OK,
)
.await;
assert_eq!(response["status"], "failed");
assert_eq!(response["failure_kind"], "runtime_identity_mismatch");
assert_eq!(response["actual_protocol_version"], serde_json::Value::Null);
let projected = serde_json::to_string(&response).unwrap();
assert!(!projected.contains("different-runtime"));
}
#[tokio::test]