chore: merge current develop into T-588

crates/client/src/backend_runtime.rs

@@ -7,10 +7,15 @@ use tokio_tungstenite::tungstenite::client::IntoClientRequest;
 use tokio_tungstenite::tungstenite::http::HeaderValue;
 use tokio_tungstenite::tungstenite::http::header::AUTHORIZATION;
 pub use workspace_api::{
-    Diagnostic as BackendDiagnostic, DiagnosticSeverity as BackendDiagnosticSeverity,
+    BrowserCreateWorkerResponse as BackendCreateWorkerResponse,
-    ListResponse as BackendRuntimeListResponse, RuntimeSummary as BackendRuntimeSummary,
+    CreateWorkspaceWorkerRequest as BackendCreateWorkerRequest, Diagnostic as BackendDiagnostic,
+    DiagnosticSeverity as BackendDiagnosticSeverity, ListResponse as BackendRuntimeListResponse,
+    RuntimeSummary as BackendRuntimeSummary,
     WorkerCapabilitySummary as BackendWorkerCapabilitySummary,
     WorkerImplementationSummary as BackendWorkerImplementationSummary,
+    WorkerLaunchOptionsResponse as BackendWorkerLaunchOptions,
+    WorkerLaunchProfileCandidate as BackendWorkerLaunchProfileCandidate,
+    WorkerLaunchRuntimeOption as BackendWorkerLaunchRuntimeOption,
     WorkerRestoreResponse as BackendWorkerRestoreResponse,
     WorkerRestoreResult as BackendWorkerRestoreResult, WorkerSummary as BackendWorkerSummary,
     WorkerWorkspaceSummary as BackendWorkerWorkspaceSummary,
@@ -171,6 +176,47 @@ struct UploadedFileResponse {
     file: protocol::UploadedFileRef,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BackendWorkerLaunchTarget {
+    pub base_url: String,
+    pub workspace_id: Option<String>,
+}
+
+impl BackendWorkerLaunchTarget {
+    pub fn new(base_url: impl Into<String>, workspace_id: Option<String>) -> Self {
+        Self {
+            base_url: base_url.into(),
+            workspace_id,
+        }
+    }
+
+    pub fn select_workspace(&mut self, workspace_id: impl Into<String>) {
+        self.workspace_id = Some(workspace_id.into());
+    }
+
+    pub fn workspace_id(&self) -> Option<&str> {
+        self.workspace_id.as_deref()
+    }
+
+    pub fn runtime_target(
+        &self,
+        runtime_id: impl Into<String>,
+        worker_id: impl Into<String>,
+    ) -> Result<BackendRuntimeTarget, BackendRuntimeClientError> {
+        let workspace_id = self.workspace_id.clone().ok_or_else(|| {
+            BackendRuntimeClientError::InvalidTarget(
+                "workspace_id is required before creating a Backend worker".to_string(),
+            )
+        })?;
+        Ok(BackendRuntimeTarget::new(
+            self.base_url.clone(),
+            workspace_id,
+            runtime_id,
+            worker_id,
+        ))
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct BackendRuntimeListTarget {
     pub base_url: String,
@@ -255,6 +301,58 @@ impl From<reqwest::Error> for BackendRuntimeClientError {
     }
 }
 
+pub async fn get_backend_worker_launch_options(
+    target: &BackendWorkerLaunchTarget,
+) -> Result<BackendWorkerLaunchOptions, BackendRuntimeClientError> {
+    validate_launch_target(target)?;
+    let api = BackendApiClient::from_stored_token(&target.base_url)?;
+    get_backend_worker_launch_options_with_client(target, &api).await
+}
+
+async fn get_backend_worker_launch_options_with_client(
+    target: &BackendWorkerLaunchTarget,
+    api: &BackendApiClient,
+) -> Result<BackendWorkerLaunchOptions, BackendRuntimeClientError> {
+    let path = backend_workspace_workers_launch_options_path(
+        target
+            .workspace_id
+            .as_deref()
+            .expect("validated Backend Workspace scope"),
+    );
+    let response = api.request(HttpMethod::GET, &path)?.send().await?;
+    let response = api.require_success(response).await?;
+    Ok(response.json::<BackendWorkerLaunchOptions>().await?)
+}
+
+pub async fn create_backend_worker(
+    target: &BackendWorkerLaunchTarget,
+    request: &BackendCreateWorkerRequest,
+) -> Result<BackendCreateWorkerResponse, BackendRuntimeClientError> {
+    validate_launch_target(target)?;
+    let api = BackendApiClient::from_stored_token(&target.base_url)?;
+    create_backend_worker_with_client(target, request, &api).await
+}
+
+async fn create_backend_worker_with_client(
+    target: &BackendWorkerLaunchTarget,
+    request: &BackendCreateWorkerRequest,
+    api: &BackendApiClient,
+) -> Result<BackendCreateWorkerResponse, BackendRuntimeClientError> {
+    let path = backend_workspace_workers_path(
+        target
+            .workspace_id
+            .as_deref()
+            .expect("validated Backend Workspace scope"),
+    );
+    let response = api
+        .request(HttpMethod::POST, &path)?
+        .json(request)
+        .send()
+        .await?;
+    let response = api.require_success(response).await?;
+    Ok(response.json::<BackendCreateWorkerResponse>().await?)
+}
+
 pub async fn list_backend_workers(
     target: &BackendRuntimeListTarget,
 ) -> Result<BackendRuntimeListResponse<BackendWorkerSummary>, BackendRuntimeClientError> {
@@ -462,6 +560,30 @@ fn validate_target(target: &BackendRuntimeTarget) -> Result<(), BackendRuntimeCl
     Ok(())
 }
 
+fn validate_launch_target(
+    target: &BackendWorkerLaunchTarget,
+) -> Result<(), BackendRuntimeClientError> {
+    if target.base_url.trim().is_empty() {
+        return Err(BackendRuntimeClientError::InvalidTarget(
+            "Backend API base URL is required".to_string(),
+        ));
+    }
+    if !(target.base_url.starts_with("http://") || target.base_url.starts_with("https://")) {
+        return Err(BackendRuntimeClientError::InvalidTarget(
+            "Backend API base URL must start with http:// or https://".to_string(),
+        ));
+    }
+    match target.workspace_id.as_deref() {
+        Some("") => Err(BackendRuntimeClientError::InvalidTarget(
+            "workspace_id must not be empty".to_string(),
+        )),
+        None => Err(BackendRuntimeClientError::InvalidTarget(
+            "workspace selection is required before creating a Backend worker".to_string(),
+        )),
+        Some(_) => Ok(()),
+    }
+}
+
 fn validate_list_target(
     target: &BackendRuntimeListTarget,
 ) -> Result<(), BackendRuntimeClientError> {
@@ -496,6 +618,17 @@ fn validate_list_target(
     Ok(())
 }
 
+fn backend_workspace_workers_path(workspace_id: &str) -> String {
+    format!("/api/w/{}/workers", path_segment_encode(workspace_id))
+}
+
+fn backend_workspace_workers_launch_options_path(workspace_id: &str) -> String {
+    format!(
+        "{}/launch-options",
+        backend_workspace_workers_path(workspace_id)
+    )
+}
+
 fn backend_runtimes_path(workspace_id: &str) -> String {
     format!("/api/w/{}/runtimes", path_segment_encode(workspace_id))
 }
@@ -580,6 +713,155 @@ fn percent_encode(input: &str, keep: impl Fn(u8) -> bool) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
+
+    async fn serve_json_once(body: serde_json::Value) -> (String, tokio::task::JoinHandle<String>) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base_url = format!("http://{}", listener.local_addr().unwrap());
+        let task = tokio::spawn(async move {
+            let (mut socket, _) = listener.accept().await.unwrap();
+            let mut request = Vec::new();
+            let header_end = loop {
+                let mut buffer = [0_u8; 4096];
+                let read = socket.read(&mut buffer).await.unwrap();
+                assert!(read > 0, "client closed before sending HTTP headers");
+                request.extend_from_slice(&buffer[..read]);
+                if let Some(position) = request.windows(4).position(|part| part == b"\r\n\r\n") {
+                    break position + 4;
+                }
+            };
+            let headers = String::from_utf8_lossy(&request[..header_end]);
+            let content_length = headers
+                .lines()
+                .find_map(|line| {
+                    let (name, value) = line.split_once(':')?;
+                    name.eq_ignore_ascii_case("content-length")
+                        .then(|| value.trim().parse::<usize>().unwrap())
+                })
+                .unwrap_or(0);
+            while request.len() < header_end + content_length {
+                let mut buffer = [0_u8; 4096];
+                let read = socket.read(&mut buffer).await.unwrap();
+                assert!(read > 0, "client closed before sending HTTP body");
+                request.extend_from_slice(&buffer[..read]);
+            }
+
+            let body = serde_json::to_vec(&body).unwrap();
+            let response = format!(
+                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+                body.len()
+            );
+            socket.write_all(response.as_bytes()).await.unwrap();
+            socket.write_all(&body).await.unwrap();
+            String::from_utf8(request).unwrap()
+        });
+        (base_url, task)
+    }
+
+    #[tokio::test]
+    async fn launch_options_request_uses_workspace_path_and_bearer_auth() {
+        let (base_url, server) = serve_json_once(serde_json::json!({
+            "workspace_id": "team main",
+            "runtimes": [{
+                "runtime_id": "embedded",
+                "display_name": "Embedded",
+                "built_in": true,
+                "worker_creation_available": true,
+                "working_directory_required": false,
+                "status": "online",
+                "diagnostics": []
+            }],
+            "default_profile": "builtin:default",
+            "profiles": [{
+                "id": "builtin:default",
+                "label": "Default",
+                "description": ""
+            }],
+            "repositories": [],
+            "working_directories": [],
+            "diagnostics": []
+        }))
+        .await;
+        let target = BackendWorkerLaunchTarget::new(&base_url, Some("team main".to_string()));
+        let api = BackendApiClient::from_access_token_for_test(&base_url, "launch-secret").unwrap();
+
+        let response = get_backend_worker_launch_options_with_client(&target, &api)
+            .await
+            .unwrap();
+        assert_eq!(response.runtimes[0].runtime_id, "embedded");
+        let request = server.await.unwrap();
+        assert!(request.starts_with("GET /api/w/team%20main/workers/launch-options HTTP/1.1\r\n"));
+        assert!(
+            request
+                .to_ascii_lowercase()
+                .contains("authorization: bearer launch-secret\r\n")
+        );
+    }
+
+    #[tokio::test]
+    async fn create_worker_posts_frontend_contract_to_workspace_path() {
+        let (base_url, server) = serve_json_once(serde_json::json!({
+            "workspace_id": "workspace-1",
+            "runtime_id": "embedded",
+            "worker_id": "worker-1",
+            "console_href": "/w/workspace-1/workers/embedded/worker-1",
+            "worker": {
+                "runtime_id": "embedded",
+                "worker_id": "worker-1",
+                "host_id": "host-1",
+                "display_name": "Coder one",
+                "label": "Coder one",
+                "profile": "builtin:coder",
+                "singleton_key": null,
+                "tags": [],
+                "workspace": {
+                    "visibility": "workspace",
+                    "identity": "workspace",
+                    "workspace_id": "workspace-1"
+                },
+                "state": "idle",
+                "last_seen_at": null,
+                "pinned": false,
+                "retention_state": "resident",
+                "implementation": {"kind": "embedded", "display_hint": "Embedded"},
+                "capabilities": {"can_stop": true, "can_spawn_followup": false},
+                "diagnostics": []
+            },
+            "diagnostics": []
+        }))
+        .await;
+        let target = BackendWorkerLaunchTarget::new(&base_url, Some("workspace-1".to_string()));
+        let api = BackendApiClient::from_access_token_for_test(&base_url, "create-secret").unwrap();
+        let create = BackendCreateWorkerRequest {
+            runtime_id: "embedded".to_string(),
+            display_name: "Coder one".to_string(),
+            profile: Some("builtin:coder".to_string()),
+            ticket_assignment: None,
+            initial_submit: Vec::new(),
+            working_directory: None,
+            control_operation_id: None,
+        };
+
+        let response = create_backend_worker_with_client(&target, &create, &api)
+            .await
+            .unwrap();
+        assert_eq!(response.worker_id, "worker-1");
+        let request = server.await.unwrap();
+        assert!(request.starts_with("POST /api/w/workspace-1/workers HTTP/1.1\r\n"));
+        assert!(
+            request
+                .to_ascii_lowercase()
+                .contains("authorization: bearer create-secret\r\n")
+        );
+        let body = request.split_once("\r\n\r\n").unwrap().1;
+        let body: serde_json::Value = serde_json::from_str(body).unwrap();
+        assert_eq!(body["runtime_id"], "embedded");
+        assert_eq!(body["display_name"], "Coder one");
+        assert_eq!(body["profile"], "builtin:coder");
+        assert_eq!(body["initial_submit"], serde_json::json!([]));
+        assert_eq!(body["working_directory"], serde_json::Value::Null);
+    }
 
     #[test]
     fn protocol_url_uses_backend_runtime_worker_identity() {

crates/client/src/lib.rs

@@ -21,11 +21,14 @@ pub use backend_auth::{
     poll_device_login, start_device_login, wait_for_device_login,
 };
 pub use backend_runtime::{
-    BackendDiagnostic, BackendDiagnosticSeverity, BackendRuntimeClientError,
+    BackendCreateWorkerRequest, BackendCreateWorkerResponse, BackendDiagnostic,
-    BackendRuntimeListResponse, BackendRuntimeListTarget, BackendRuntimeSummary,
+    BackendDiagnosticSeverity, BackendRuntimeClientError, BackendRuntimeListResponse,
-    BackendRuntimeTarget, BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary,
+    BackendRuntimeListTarget, BackendRuntimeSummary, BackendRuntimeTarget,
-    BackendWorkerRestoreResponse, BackendWorkerRestoreResult, BackendWorkerSummary,
+    BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, BackendWorkerLaunchOptions,
-    BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, connect_backend_runtime,
+    BackendWorkerLaunchProfileCandidate, BackendWorkerLaunchRuntimeOption,
+    BackendWorkerLaunchTarget, BackendWorkerRestoreResponse, BackendWorkerRestoreResult,
+    BackendWorkerSummary, BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary,
+    connect_backend_runtime, create_backend_worker, get_backend_worker_launch_options,
     list_backend_stopped_workers, list_backend_workers, restore_backend_worker,
 };
 pub use backend_workspace::{
@@ -35,9 +38,9 @@ pub use backend_workspace::{
 };
 pub use client::{Client, ClientError};
 pub use target::{
-    BackendTarget, Dashboard, ResolvedTarget, StandaloneTarget, StandaloneWorkerListIntent,
+    BackendTarget, BackendWorkerLaunch, Dashboard, ResolvedTarget, StandaloneTarget,
-    StandaloneWorkerResumeIntent, Target, TargetError, TargetKind, WorkerConnection,
+    StandaloneWorkerListIntent, StandaloneWorkerResumeIntent, Target, TargetError, TargetKind,
-    WorkerConnectionSelector, WorkerList, WorkerListRequest, WorkerSpawn,
+    WorkerConnection, WorkerConnectionSelector, WorkerList, WorkerListRequest, WorkerSpawn,
 };
 pub use workspace_api::{
     CompanionCancelRequest, CompanionLifecycleState, CompanionMessageDisposition,

crates/client/src/target.rs

@@ -2,7 +2,7 @@ use std::{fmt, path::PathBuf};
 
 use crate::{
     BackendApiClient, BackendApiClientError, BackendOrigin, BackendRuntimeListTarget,
-    BackendRuntimeTarget,
+    BackendRuntimeTarget, BackendWorkerLaunchTarget,
 };
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -123,6 +123,11 @@ pub struct Dashboard {
     pub workspace_id: String,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BackendWorkerLaunch {
+    pub target: BackendWorkerLaunchTarget,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct WorkerList {
     pub backend_target: BackendRuntimeListTarget,
@@ -199,6 +204,13 @@ pub trait Target: fmt::Debug + Send + Sync {
         Err(TargetError::unsupported("Worker dashboard", self.kind()))
     }
 
+    fn launch_backend_worker(&self) -> Result<BackendWorkerLaunch, TargetError> {
+        Err(TargetError::unsupported(
+            "Backend Worker launch",
+            self.kind(),
+        ))
+    }
+
     fn list_workers(&self, _request: WorkerListRequest) -> Result<WorkerList, TargetError> {
         Err(TargetError::unsupported("Worker listing", self.kind()))
     }
@@ -299,6 +311,15 @@ impl Target for BackendTarget {
         })
     }
 
+    fn launch_backend_worker(&self) -> Result<BackendWorkerLaunch, TargetError> {
+        Ok(BackendWorkerLaunch {
+            target: BackendWorkerLaunchTarget::new(
+                self.base_url.clone(),
+                self.workspace_id.clone(),
+            ),
+        })
+    }
+
     fn list_workers(&self, request: WorkerListRequest) -> Result<WorkerList, TargetError> {
         Ok(WorkerList {
             backend_target: BackendRuntimeListTarget::new(

crates/tui/src/backend_spawn.rs

@@ -0,0 +1,483 @@
+use client::{
+    BackendCreateWorkerRequest, BackendWorkerLaunchOptions, BackendWorkerLaunchProfileCandidate,
+    BackendWorkerLaunchRuntimeOption, BackendWorkerLaunchTarget, create_backend_worker,
+    get_backend_worker_launch_options,
+};
+use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers};
+use ratatui::layout::{Constraint, Direction, Layout};
+use ratatui::style::{Color, Modifier, Style};
+use ratatui::text::{Line, Span};
+use ratatui::widgets::{Block, Borders, Paragraph, Wrap};
+
+use crate::backend_workspace_picker::select_backend_workspace;
+use crate::console;
+use crate::inline_terminal::{InlineTerminal, with_inline_terminal};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Field {
+    Name,
+    Runtime,
+    Profile,
+}
+
+impl Field {
+    fn next(self) -> Self {
+        match self {
+            Self::Name => Self::Runtime,
+            Self::Runtime => Self::Profile,
+            Self::Profile => Self::Name,
+        }
+    }
+
+    fn previous(self) -> Self {
+        match self {
+            Self::Name => Self::Profile,
+            Self::Runtime => Self::Name,
+            Self::Profile => Self::Runtime,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct Selection {
+    runtime_id: String,
+    display_name: String,
+    profile: String,
+}
+
+struct FormState {
+    field: Field,
+    display_name: String,
+    runtime_index: usize,
+    profile_index: usize,
+    status: String,
+}
+
+impl FormState {
+    fn new(options: &BackendWorkerLaunchOptions) -> Self {
+        let runtime_index = options
+            .runtimes
+            .iter()
+            .position(runtime_supports_workdirless_creation)
+            .unwrap_or(0);
+        let profile_index = options
+            .default_profile
+            .as_deref()
+            .and_then(|default| {
+                options
+                    .profiles
+                    .iter()
+                    .position(|candidate| candidate.id == default)
+            })
+            .unwrap_or(0);
+        Self {
+            field: Field::Name,
+            display_name: "Worker".to_string(),
+            runtime_index,
+            profile_index,
+            status: String::new(),
+        }
+    }
+
+    fn current_runtime<'a>(
+        &self,
+        options: &'a BackendWorkerLaunchOptions,
+    ) -> Option<&'a BackendWorkerLaunchRuntimeOption> {
+        options.runtimes.get(self.runtime_index)
+    }
+
+    fn current_profile<'a>(
+        &self,
+        options: &'a BackendWorkerLaunchOptions,
+    ) -> Option<&'a BackendWorkerLaunchProfileCandidate> {
+        options.profiles.get(self.profile_index)
+    }
+
+    fn cycle_runtime(&mut self, options: &BackendWorkerLaunchOptions, delta: isize) {
+        self.runtime_index = cycle_index(self.runtime_index, options.runtimes.len(), delta);
+        self.status.clear();
+    }
+
+    fn cycle_profile(&mut self, options: &BackendWorkerLaunchOptions, delta: isize) {
+        self.profile_index = cycle_index(self.profile_index, options.profiles.len(), delta);
+        self.status.clear();
+    }
+
+    fn submit(&mut self, options: &BackendWorkerLaunchOptions) -> Option<Selection> {
+        let display_name = self.display_name.trim();
+        if display_name.is_empty() {
+            self.status = "Worker name is required.".to_string();
+            self.field = Field::Name;
+            return None;
+        }
+        let Some(runtime) = self.current_runtime(options) else {
+            self.status = "No Runtime is available in this Workspace.".to_string();
+            self.field = Field::Runtime;
+            return None;
+        };
+        if !runtime.worker_creation_available {
+            self.status = "The selected Runtime cannot create Workers right now.".to_string();
+            self.field = Field::Runtime;
+            return None;
+        }
+        if runtime.working_directory_required {
+            self.status =
+                "The selected Runtime requires a workdir; this launch flow does not select one yet."
+                    .to_string();
+            self.field = Field::Runtime;
+            return None;
+        }
+        let Some(profile) = self.current_profile(options) else {
+            self.status = "No Worker profile is available.".to_string();
+            self.field = Field::Profile;
+            return None;
+        };
+        Some(Selection {
+            runtime_id: runtime.runtime_id.clone(),
+            display_name: display_name.to_string(),
+            profile: profile.id.clone(),
+        })
+    }
+}
+
+pub async fn run(mut target: BackendWorkerLaunchTarget) -> Result<(), Box<dyn std::error::Error>> {
+    if target.workspace_id().is_none() {
+        let Some(workspace) = select_backend_workspace(&target.base_url).await? else {
+            return Ok(());
+        };
+        target.select_workspace(workspace);
+    }
+
+    let options = get_backend_worker_launch_options(&target).await?;
+    let Some(selection) = select_worker(&options)? else {
+        return Ok(());
+    };
+    let request = request_from_selection(selection);
+    let created = create_backend_worker(&target, &request).await?;
+    let runtime_target = target.runtime_target(created.runtime_id, created.worker_id)?;
+    console::run_backend_runtime(runtime_target).await
+}
+
+fn request_from_selection(selection: Selection) -> BackendCreateWorkerRequest {
+    BackendCreateWorkerRequest {
+        runtime_id: selection.runtime_id,
+        display_name: selection.display_name,
+        profile: Some(selection.profile),
+        initial_submit: Vec::new(),
+        working_directory: None,
+        ticket_assignment: None,
+        control_operation_id: None,
+    }
+}
+
+const VIEWPORT_LINES: u16 = 14;
+
+fn select_worker(
+    options: &BackendWorkerLaunchOptions,
+) -> Result<Option<Selection>, Box<dyn std::error::Error>> {
+    with_inline_terminal(VIEWPORT_LINES, |terminal| run_form(terminal, options))
+}
+
+fn run_form(
+    terminal: &mut InlineTerminal,
+    options: &BackendWorkerLaunchOptions,
+) -> Result<Option<Selection>, Box<dyn std::error::Error>> {
+    let mut state = FormState::new(options);
+
+    loop {
+        terminal.draw(|frame| render(frame, &state, options))?;
+        let event = event::read()?;
+        let Event::Key(key) = event else {
+            continue;
+        };
+        if key.kind != KeyEventKind::Press {
+            continue;
+        }
+        if key.code == KeyCode::Char('c') && key.modifiers.contains(KeyModifiers::CONTROL) {
+            return Ok(None);
+        }
+
+        match key.code {
+            KeyCode::Esc => {
+                return Ok(None);
+            }
+            KeyCode::Tab | KeyCode::Down => {
+                state.field = state.field.next();
+                state.status.clear();
+            }
+            KeyCode::BackTab | KeyCode::Up => {
+                state.field = state.field.previous();
+                state.status.clear();
+            }
+            KeyCode::Left => match state.field {
+                Field::Runtime => state.cycle_runtime(options, -1),
+                Field::Profile => state.cycle_profile(options, -1),
+                Field::Name => {}
+            },
+            KeyCode::Right => match state.field {
+                Field::Runtime => state.cycle_runtime(options, 1),
+                Field::Profile => state.cycle_profile(options, 1),
+                Field::Name => {}
+            },
+            KeyCode::Enter => {
+                if let Some(selection) = state.submit(options) {
+                    return Ok(Some(selection));
+                }
+            }
+            KeyCode::Backspace if state.field == Field::Name => {
+                state.display_name.pop();
+                state.status.clear();
+            }
+            KeyCode::Char(character)
+                if state.field == Field::Name
+                    && !key.modifiers.contains(KeyModifiers::CONTROL)
+                    && !character.is_control() =>
+            {
+                state.display_name.push(character);
+                state.status.clear();
+            }
+            _ => {}
+        }
+    }
+}
+
+fn render(frame: &mut ratatui::Frame<'_>, state: &FormState, options: &BackendWorkerLaunchOptions) {
+    let area = frame.area();
+    let vertical = Layout::default()
+        .direction(Direction::Vertical)
+        .constraints([
+            Constraint::Length(1),
+            Constraint::Length(3),
+            Constraint::Length(3),
+            Constraint::Length(3),
+            Constraint::Length(3),
+            Constraint::Min(1),
+        ])
+        .split(area);
+
+    frame.render_widget(
+        Paragraph::new(Line::from(vec![
+            Span::styled(
+                "New Backend Worker",
+                Style::default().add_modifier(Modifier::BOLD),
+            ),
+            Span::raw(format!("  Workspace: {}", options.workspace_id)),
+        ])),
+        vertical[0],
+    );
+
+    let focused = Style::default().fg(Color::Cyan);
+    frame.render_widget(
+        Paragraph::new(state.display_name.as_str()).block(
+            Block::default()
+                .borders(Borders::ALL)
+                .title(" Name ")
+                .border_style(if state.field == Field::Name {
+                    focused
+                } else {
+                    Style::default()
+                }),
+        ),
+        vertical[1],
+    );
+
+    let runtime_text = state
+        .current_runtime(options)
+        .map(runtime_label)
+        .unwrap_or_else(|| "No Runtime available".to_string());
+    frame.render_widget(
+        Paragraph::new(runtime_text).block(
+            Block::default()
+                .borders(Borders::ALL)
+                .title(runtime_title(state, options))
+                .border_style(if state.field == Field::Runtime {
+                    focused
+                } else {
+                    Style::default()
+                }),
+        ),
+        vertical[2],
+    );
+
+    let profile_text = state
+        .current_profile(options)
+        .map(|profile| {
+            if profile.description.is_empty() {
+                profile.label.clone()
+            } else {
+                format!("{} — {}", profile.label, profile.description)
+            }
+        })
+        .unwrap_or_else(|| "No profile available".to_string());
+    frame.render_widget(
+        Paragraph::new(profile_text).block(
+            Block::default()
+                .borders(Borders::ALL)
+                .title(profile_title(state, options))
+                .border_style(if state.field == Field::Profile {
+                    focused
+                } else {
+                    Style::default()
+                }),
+        ),
+        vertical[3],
+    );
+
+    let status = if state.status.is_empty() {
+        "Tab/↑/↓: field  ←/→: choice  Enter: create  Esc/Ctrl-C: cancel"
+    } else {
+        state.status.as_str()
+    };
+    frame.render_widget(
+        Paragraph::new(status)
+            .style(if state.status.is_empty() {
+                Style::default().fg(Color::DarkGray)
+            } else {
+                Style::default().fg(Color::Yellow)
+            })
+            .wrap(Wrap { trim: true }),
+        vertical[4],
+    );
+
+    if state.field == Field::Name {
+        let max_cursor = vertical[1].width.saturating_sub(2) as usize;
+        frame.set_cursor_position((
+            vertical[1].x + 1 + state.display_name.chars().count().min(max_cursor) as u16,
+            vertical[1].y + 1,
+        ));
+    }
+}
+
+fn runtime_title(state: &FormState, options: &BackendWorkerLaunchOptions) -> String {
+    if options.runtimes.is_empty() {
+        " Runtime ".to_string()
+    } else {
+        format!(
+            " Runtime ({}/{}) ",
+            state.runtime_index + 1,
+            options.runtimes.len()
+        )
+    }
+}
+
+fn profile_title(state: &FormState, options: &BackendWorkerLaunchOptions) -> String {
+    if options.profiles.is_empty() {
+        " Profile ".to_string()
+    } else {
+        format!(
+            " Profile ({}/{}) ",
+            state.profile_index + 1,
+            options.profiles.len()
+        )
+    }
+}
+
+fn runtime_label(runtime: &BackendWorkerLaunchRuntimeOption) -> String {
+    let availability = if !runtime.worker_creation_available {
+        "unavailable"
+    } else if runtime.working_directory_required {
+        "workdir required"
+    } else {
+        "no workdir"
+    };
+    format!(
+        "{} [{}] — {availability}",
+        runtime.display_name, runtime.runtime_id
+    )
+}
+
+fn runtime_supports_workdirless_creation(runtime: &BackendWorkerLaunchRuntimeOption) -> bool {
+    runtime.worker_creation_available && !runtime.working_directory_required
+}
+
+fn cycle_index(current: usize, len: usize, delta: isize) -> usize {
+    if len == 0 {
+        return 0;
+    }
+    (current as isize + delta).rem_euclid(len as isize) as usize
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use client::{BackendDiagnostic, BackendWorkerLaunchOptions};
+
+    fn options() -> BackendWorkerLaunchOptions {
+        BackendWorkerLaunchOptions {
+            workspace_id: "workspace-1".to_string(),
+            runtimes: vec![
+                BackendWorkerLaunchRuntimeOption {
+                    runtime_id: "external".to_string(),
+                    display_name: "External".to_string(),
+                    built_in: false,
+                    worker_creation_available: true,
+                    working_directory_required: true,
+                    status: "online".to_string(),
+                    diagnostics: Vec::new(),
+                },
+                BackendWorkerLaunchRuntimeOption {
+                    runtime_id: "embedded".to_string(),
+                    display_name: "Embedded".to_string(),
+                    built_in: true,
+                    worker_creation_available: true,
+                    working_directory_required: false,
+                    status: "online".to_string(),
+                    diagnostics: Vec::new(),
+                },
+            ],
+            profiles: vec![
+                BackendWorkerLaunchProfileCandidate {
+                    id: "builtin:default".to_string(),
+                    label: "Default".to_string(),
+                    description: String::new(),
+                },
+                BackendWorkerLaunchProfileCandidate {
+                    id: "builtin:coder".to_string(),
+                    label: "Coder".to_string(),
+                    description: "Ticket implementation".to_string(),
+                },
+            ],
+            default_profile: Some("builtin:coder".to_string()),
+            repositories: Vec::new(),
+            working_directories: Vec::new(),
+            diagnostics: Vec::<BackendDiagnostic>::new(),
+        }
+    }
+
+    #[test]
+    fn defaults_to_workdirless_runtime_and_backend_default_profile() {
+        let options = options();
+        let state = FormState::new(&options);
+        assert_eq!(
+            state.current_runtime(&options).unwrap().runtime_id,
+            "embedded"
+        );
+        assert_eq!(state.current_profile(&options).unwrap().id, "builtin:coder");
+        assert_eq!(state.display_name, "Worker");
+    }
+
+    #[test]
+    fn workdir_required_runtime_cannot_be_submitted() {
+        let options = options();
+        let mut state = FormState::new(&options);
+        state.runtime_index = 0;
+        assert_eq!(state.submit(&options), None);
+        assert!(state.status.contains("requires a workdir"));
+        assert_eq!(state.field, Field::Runtime);
+    }
+
+    #[test]
+    fn selection_builds_workdirless_create_request() {
+        let request = request_from_selection(Selection {
+            runtime_id: "embedded".to_string(),
+            display_name: "Coder one".to_string(),
+            profile: "builtin:coder".to_string(),
+        });
+        assert_eq!(request.runtime_id, "embedded");
+        assert_eq!(request.display_name, "Coder one");
+        assert_eq!(request.profile.as_deref(), Some("builtin:coder"));
+        assert!(request.initial_submit.is_empty());
+        assert!(request.working_directory.is_none());
+        assert!(request.ticket_assignment.is_none());
+    }
+}

crates/tui/src/backend_worker_picker.rs

@@ -12,6 +12,7 @@ use ratatui::layout::{Constraint, Layout};
 use ratatui::style::{Color, Modifier, Style};
 use ratatui::text::{Line, Span};
 use ratatui::widgets::Paragraph;
+use unicode_width::UnicodeWidthStr;
 
 use crate::backend_workspace_picker::select_backend_workspace;
 use crate::console;
@@ -235,9 +236,10 @@ fn draw(frame: &mut Frame<'_>, state: &BackendWorkerPickerState) {
         layout[0],
     );
 
+    let column_widths = WorkerColumnWidths::from_workers(&state.workers);
     for (i, worker) in state.workers.iter().enumerate() {
         frame.render_widget(
-            Paragraph::new(row_line(worker, i == state.selected)),
+            Paragraph::new(row_line(worker, &column_widths, i == state.selected)),
             layout[i + 1],
         );
     }
@@ -272,7 +274,28 @@ fn picker_title(target: &BackendRuntimeListTarget) -> String {
     format!("backend workers   workspace: {workspace}   runtime: {runtime}")
 }
 
-fn row_line(worker: &BackendWorkerSummary, selected: bool) -> Line<'static> {
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+struct WorkerColumnWidths {
+    identity: usize,
+    name: usize,
+    state: usize,
+}
+
+impl WorkerColumnWidths {
+    fn from_workers(workers: &[BackendWorkerSummary]) -> Self {
+        workers.iter().fold(Self::default(), |widths, worker| Self {
+            identity: widths.identity.max(text_width(&short_worker_id(worker))),
+            name: widths.name.max(text_width(worker_name(worker))),
+            state: widths.state.max(text_width(&worker_state(worker))),
+        })
+    }
+}
+
+fn row_line(
+    worker: &BackendWorkerSummary,
+    widths: &WorkerColumnWidths,
+    selected: bool,
+) -> Line<'static> {
     let marker = if selected { "▶ " } else { "  " };
     let id_style = if selected {
         Style::default()
@@ -281,42 +304,58 @@ fn row_line(worker: &BackendWorkerSummary, selected: bool) -> Line<'static> {
     } else {
         Style::default().fg(Color::Cyan)
     };
-    let preview_style = if selected {
+    let name_style = if selected {
         Style::default().fg(Color::White)
     } else {
         Style::default().fg(Color::DarkGray)
     };
 
-    let label = if worker.label.is_empty() {
-        worker.worker_id.as_str()
-    } else {
-        worker.label.as_str()
-    };
-    let profile = worker.profile.as_deref().unwrap_or("-");
-
     Line::from(vec![
         Span::raw(marker),
-        Span::styled(short_worker_id(worker), id_style),
-        Span::raw("  "),
         Span::styled(
-            format!("[{}]", worker.state),
+            pad_column(&short_worker_id(worker), widths.identity),
-            state_style(worker.state.as_str()),
+            id_style,
         ),
         Span::raw("  "),
+        Span::styled(pad_column(worker_name(worker), widths.name), name_style),
+        Span::raw("  "),
         Span::styled(
-            format!("profile:{profile}"),
+            pad_column(&worker_state(worker), widths.state),
-            Style::default().fg(Color::DarkGray),
+            state_style(worker.state.as_str()),
         ),
         Span::raw("  "),
         Span::styled(
             working_directory_text(worker),
             Style::default().fg(Color::DarkGray),
         ),
-        Span::raw("  "),
-        Span::styled(label.to_string(), preview_style),
     ])
 }
 
+fn worker_name(worker: &BackendWorkerSummary) -> &str {
+    if !worker.label.is_empty() {
+        worker.label.as_str()
+    } else if !worker.display_name.is_empty() {
+        worker.display_name.as_str()
+    } else {
+        worker.worker_id.as_str()
+    }
+}
+
+fn worker_state(worker: &BackendWorkerSummary) -> String {
+    format!("[{}]", worker.state)
+}
+
+fn text_width(value: &str) -> usize {
+    UnicodeWidthStr::width(value)
+}
+
+fn pad_column(value: &str, width: usize) -> String {
+    format!(
+        "{value}{}",
+        " ".repeat(width.saturating_sub(text_width(value)))
+    )
+}
+
 fn state_style(state: &str) -> Style {
     match state {
         "running" | "idle" | "active" => Style::default()
@@ -347,11 +386,7 @@ fn working_directory_text(worker: &BackendWorkerSummary) -> String {
     let Some(wd) = worker.working_directory.as_ref() else {
         return "wd:—".to_string();
     };
-    let cleanliness = wd.cleanliness.as_deref().unwrap_or("unknown");
+    format!("wd:{}・{}", wd.repository_key, wd.working_directory_id)
-    format!(
-        "wd:{}:{} {} {}",
-        wd.repository_key, wd.working_directory_id, wd.status, cleanliness
-    )
 }
 
 #[cfg(test)]
@@ -395,18 +430,90 @@ mod tests {
         }
     }
 
-    #[test]
+    fn row_text(worker: &BackendWorkerSummary, widths: &WorkerColumnWidths) -> String {
-    fn worker_row_matches_inline_picker_shape() {
+        row_line(worker, widths, false)
-        let row = row_line(&worker("runtime-a", "worker-b", Some("default")), true);
-        let text = row
             .spans
             .into_iter()
             .map(|span| span.content)
-            .collect::<String>();
+            .collect()
-        assert!(text.starts_with("▶ W-1"));
+    }
-        assert!(text.contains("[running]"));
+
-        assert!(text.contains("profile:default"));
+    fn display_column(text: &str, value: &str) -> usize {
-        assert!(text.contains("wd:—"));
+        let byte_offset = text.find(value).expect("value in rendered row");
+        text_width(&text[..byte_offset])
+    }
+
+    #[test]
+    fn worker_row_orders_and_simplifies_columns() {
+        let mut worker = worker("runtime-a", "worker-b", Some("builtin:coder"));
+        worker.resource_key = "W-90".to_string();
+        worker.display_name = "Coder".to_string();
+        worker.label = "Coder · T-585".to_string();
+        worker.state = "stopped".to_string();
+        worker.working_directory = Some(
+            serde_json::from_value(serde_json::json!({
+                "working_directory_id": "001a06a9f0202000000",
+                "repository_key": "main",
+                "materializer_kind": "local_git_worktree",
+                "status": "active",
+                "cleanliness": "clean"
+            }))
+            .unwrap(),
+        );
+        let widths = WorkerColumnWidths::from_workers(std::slice::from_ref(&worker));
+        let text = row_text(&worker, &widths);
+
+        assert_eq!(
+            text,
+            "  W-90  Coder · T-585  [stopped]  wd:main・001a06a9f0202000000"
+        );
+        assert!(!text.contains("profile:"));
+        assert!(!text.contains("active clean"));
+    }
+
+    #[test]
+    fn worker_rows_align_identity_name_state_and_workdir_columns() {
+        let mut short = worker("runtime-a", "worker-a", None);
+        short.resource_key = "W-2".to_string();
+        short.label = "Coder".to_string();
+        short.display_name = short.label.clone();
+        short.state = "idle".to_string();
+
+        let mut long = worker("runtime-a", "worker-b", None);
+        long.resource_key = "W-100".to_string();
+        long.label = "Longer worker · T-9".to_string();
+        long.display_name = long.label.clone();
+        long.state = "stopped".to_string();
+
+        for worker in [&mut short, &mut long] {
+            worker.working_directory = Some(
+                serde_json::from_value(serde_json::json!({
+                    "working_directory_id": "workdir-1",
+                    "repository_key": "main",
+                    "materializer_kind": "local_git_worktree",
+                    "status": "active"
+                }))
+                .unwrap(),
+            );
+        }
+
+        let workers = vec![short, long];
+        let widths = WorkerColumnWidths::from_workers(&workers);
+        let first = row_text(&workers[0], &widths);
+        let second = row_text(&workers[1], &widths);
+
+        assert_eq!(
+            display_column(&first, "Coder"),
+            display_column(&second, "Longer")
+        );
+        assert_eq!(
+            display_column(&first, "[idle]"),
+            display_column(&second, "[stopped]")
+        );
+        assert_eq!(
+            display_column(&first, "wd:main"),
+            display_column(&second, "wd:main")
+        );
     }
 
     #[test]

crates/tui/src/lib.rs

@@ -1,5 +1,6 @@
 mod app;
 mod backend_dashboard;
+mod backend_spawn;
 mod backend_worker_picker;
 mod backend_workspace_picker;
 mod block;
@@ -51,6 +52,8 @@ pub enum LaunchMode {
     /// Restore one client-owned standalone Worker. The current cwd is the default scope;
     /// `include_all` opts into all standalone Workers under the same client data root.
     StandaloneResume { include_all: bool },
+    /// Create one Backend Worker and attach to it.
+    BackendSpawn,
     /// List Backend Workers and attach to the selected Worker.
     Workers {
         runtime_id: Option<String>,
@@ -161,6 +164,10 @@ pub async fn launch(options: LaunchOptions) -> ExitCode {
                 Err(error) => Err(Box::new(error) as Box<dyn std::error::Error>),
             }
         }
+        LaunchMode::BackendSpawn => match target.launch_backend_worker() {
+            Ok(launch) => backend_spawn::run(launch.target).await,
+            Err(e) => Err(Box::new(e) as Box<dyn std::error::Error>),
+        },
         LaunchMode::Workers {
             runtime_id,
             include_stopped,

crates/worker-runtime/src/http_server.rs

@@ -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> {
@@ -1843,6 +1899,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;
     }
@@ -2160,6 +2219,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 runtime_protocol_replaces_serialized_tracked_source() {
         let wire = serde_json::to_string(&protocol::Method::SubmitTracked {

crates/workspace-api/src/lib.rs

@@ -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 status: RuntimeConnectionTestStatus,
-    pub protocol_version: Option<String>,
+    pub failure_kind: Option<RuntimeConnectionTestFailureKind>,
-    pub compatibility_basis: String,
+    pub expected_protocol_version: u32,
-    #[serde(default)]
+    pub actual_protocol_version: Option<u32>,
-    pub capabilities: Vec<String>,
-    pub health_result: String,
     #[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

@@ -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,
+    RuntimeHttpErrorResponse, RuntimeHttpPingResponse, RuntimeHttpRepositoryAccessResponse,
-    RuntimeHttpUploadedFileDeleteResponse, RuntimeHttpUploadedFileResponse,
+    RuntimeHttpSummaryResponse, RuntimeHttpUploadedFileDeleteResponse,
-    RuntimeHttpWorkerCompletionsRequest, RuntimeHttpWorkerCompletionsResponse,
+    RuntimeHttpUploadedFileResponse, RuntimeHttpWorkerCompletionsRequest,
-    RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerInputResponse,
+    RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse,
-    RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse,
+    RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest,
-    RuntimeHttpWorkerResponse, RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
+    RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse,
+    RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
     RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
     RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse,
 };
@@ -64,6 +68,7 @@ pub(crate) 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>;
@@ -1791,6 +1835,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()
@@ -2901,6 +2956,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",
@@ -3049,14 +3147,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| {
@@ -3079,6 +3181,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,
@@ -3266,6 +3444,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());
@@ -4562,7 +4744,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;
@@ -5709,6 +5891,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

@@ -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,
@@ -107,14 +112,14 @@ use crate::config_source::ConfigCommitRequest;
 use crate::hosts::{
     ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EMBEDDED_RUNTIME_ID,
     EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime,
-    RuntimeDiagnostic, RuntimeRegistry, RuntimeRegistryError, RuntimeRegistryUnregisterResult,
+    RuntimeDiagnostic, RuntimePingFailureKind, RuntimeRegistry, RuntimeRegistryError,
-    TicketWorkerRole, WorkerCapabilitySummary, WorkerCompletionsRequest, WorkerCompletionsResult,
+    RuntimeRegistryUnregisterResult, TicketWorkerRole, WorkerCapabilitySummary,
-    WorkerControlOperation, WorkerCreateBinding, WorkerImplementationSummary, WorkerInputKind,
+    WorkerCompletionsRequest, WorkerCompletionsResult, WorkerControlOperation, WorkerCreateBinding,
-    WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, WorkerLifecycleResult,
+    WorkerImplementationSummary, WorkerInputKind, WorkerInputRequest, WorkerInputResult,
-    WorkerOperationState, WorkerRestoreResult, WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent,
+    WorkerLifecycleRequest, WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult,
-    WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, WorkerSummary,
+    WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult,
-    WorkerTicketAssignmentRequest, WorkerWorkspaceSummary, worker_spawn_create_fingerprint,
+    WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerTicketAssignmentRequest,
-    workspace_worker_summary,
+    WorkerWorkspaceSummary, worker_spawn_create_fingerprint, workspace_worker_summary,
 };
 use crate::identity::WorkspaceIdentity;
 use crate::memory_backend::execute_memory_backend_operation_with_authority;
@@ -164,11 +169,7 @@ use worker_runtime::catalog::{
     WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
 };
 use worker_runtime::config_bundle::ConfigBundle;
-use worker_runtime::http_server::{
+use worker_runtime::http_server::MAX_WORKER_FILE_UPLOAD_BYTES;
-    MAX_WORKER_FILE_UPLOAD_BYTES, RuntimeHttpConfigBundleAvailabilityResponse,
-    RuntimeHttpConfigBundlesResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerResponse,
-    RuntimeHttpWorkersResponse,
-};
 use worker_runtime::identity::{RuntimeWorkerRef, WorkerId};
 
 const EMBEDDED_WORKER_RUNTIME_ID: &str = "embedded-worker-runtime";
@@ -12522,14 +12523,39 @@ async fn test_runtime_connection(
     State(api): State<WorkspaceApi>,
     AxumPath(runtime_id): AxumPath<String>,
 ) -> ApiResult<Json<RuntimeConnectionTestResponse>> {
+    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());
+    }
     let runtime_config = load_backend_runtimes_config_for_settings(&api)?;
-    let remote = runtime_config
+    runtime_config
         .runtimes
         .remote
         .iter()
         .find(|remote| remote.id == runtime_id)
         .ok_or_else(|| Error::UnknownRuntime(runtime_id.clone()))?;
-    Ok(Json(test_remote_runtime_config(&api, remote).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(
@@ -14692,438 +14718,111 @@ fn remote_runtime_config_from_file(
     })
 }
 
-async fn test_remote_runtime_config(
+fn runtime_connection_test_response(
-    api: &WorkspaceApi,
+    workspace_id: &str,
-    remote: &RemoteRuntimeConfigFile,
+    runtime_id: &str,
-) -> RuntimeConnectionTestResponse {
-    let checked_at = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
-    if remote
-        .token_ref
-        .as_deref()
-        .is_some_and(|value| !value.trim().is_empty())
-    {
-        return RuntimeConnectionTestResponse {
-            workspace_id: api.config.workspace_id.clone(),
-            runtime_id: remote.id.clone(),
-            checked_at,
-            state: "rejected".to_string(),
-            protocol_version: None,
-            compatibility_basis: "not_checked_token_ref_unsupported".to_string(),
-            capabilities: Vec::new(),
-            health_result: "not_checked".to_string(),
-            diagnostics: vec![settings_diagnostic(
-                "remote_runtime_token_ref_unsupported",
-                DiagnosticSeverity::Error,
-                "Remote Runtime test cannot use token_ref in v0; no token or secret value was exposed to the Browser.",
-            )
-            .into()],
-        };
-    }
-
-    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.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: &RemoteRuntimeConfigFile,
     checked_at: String,
-    code: impl Into<String>,
+    ping: std::result::Result<
-    message: impl Into<String>,
+        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(),
+        workspace_id: workspace_id.to_string(),
-        runtime_id: remote.id.clone(),
+        runtime_id: runtime_id.to_string(),
         checked_at,
-        state: "failed".to_string(),
+        status: RuntimeConnectionTestStatus::Failed,
-        protocol_version: None,
+        failure_kind: Some(failure_kind),
-        compatibility_basis: "worker-runtime lightweight HTTP compatibility probes".to_string(),
+        expected_protocol_version: RUNTIME_HTTP_PROTOCOL_VERSION,
-        capabilities: Vec::new(),
+        actual_protocol_version,
-        health_result: "failed".to_string(),
+        diagnostics: vec![diagnostic.into()],
-        diagnostics: vec![settings_diagnostic(code, DiagnosticSeverity::Error, message).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: &RemoteRuntimeConfigFile,
-    path: &str,
-) -> std::result::Result<String, RuntimeDiagnostic> {
-    let endpoint = remote.endpoint.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
@@ -24492,6 +24191,71 @@ 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.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)
     }
@@ -25730,124 +25494,157 @@ mod tests {
         assert_eq!(persisted.runtimes.remote.len(), 1);
     }
 
-    #[tokio::test(flavor = "multi_thread")]
+    fn write_test_remote_runtime(root: &std::path::Path, runtime_id: &str, endpoint: String) {
-    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()
-            }
-        });
-
-        let dir = tempfile::tempdir().unwrap();
-        let endpoint = format!("http://{runtime_addr}");
         BackendRuntimesConfigFile {
             runtimes: WorkspaceBackendRuntimesConfig {
                 remote: vec![RemoteRuntimeConfigFile {
-                    id: "probe-runtime".to_string(),
+                    id: runtime_id.to_string(),
-                    endpoint: endpoint.clone(),
+                    endpoint,
                     display_name: Some("Probe Runtime".to_string()),
                     token_ref: None,
                 }],
             },
         }
-        .write_to_path(dir.path().join(".test-config/runtimes.toml"))
+        .write_to_path(root.join(".test-config/runtimes.toml"))
         .unwrap();
-        let app = test_app(dir.path()).await;
+    }
 
-        let response = post_json(
+    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();
+        write_test_remote_runtime(dir.path(), "probe-runtime", endpoint.clone());
+        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;
+        .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);
     }
 
     #[tokio::test(flavor = "multi_thread")]
-    async fn runtime_connection_test_marks_missing_execution_backend_incompatible() {
+    async fn runtime_connection_test_reports_exact_compatible_protocol() {
-        let runtime =
+        let response = run_runtime_connection_test(
-            worker_runtime::Runtime::with_options(worker_runtime::RuntimeOptions::default());
+            serde_json::json!({
-        let runtime_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+                "runtime_id": "probe-runtime",
-        let runtime_addr = runtime_listener.local_addr().unwrap();
+                "protocol_version": RUNTIME_HTTP_PROTOCOL_VERSION,
-        tokio::spawn(async move {
+            }),
-            serve_runtime_http_with_injected_test_auth(runtime, runtime_listener)
+            StatusCode::OK,
-                .await
-                .unwrap()
-        });
-
-        let dir = tempfile::tempdir().unwrap();
-        let endpoint = format!("http://{runtime_addr}");
-        BackendRuntimesConfigFile {
-            runtimes: WorkspaceBackendRuntimesConfig {
-                remote: vec![RemoteRuntimeConfigFile {
-                    id: "control-only-runtime".to_string(),
-                    display_name: Some("Control-only Runtime".to_string()),
-                    endpoint,
-                    token_ref: None,
-                }],
-            },
-        }
-        .write_to_path(dir.path().join(".test-config/runtimes.toml"))
-        .unwrap();
-        let app = test_app(dir.path()).await;
-
-        let response = post_json(
-            app,
-            &format!("/api/w/{TEST_WORKSPACE_ID}/runtimes/control-only-runtime/connection-tests"),
-            serde_json::json!({}),
         )
         .await;
-        assert_eq!(response["state"], "incompatible");
+
-        assert!(
+        assert_eq!(response["status"], "compatible");
-            response["capabilities"]
+        assert_eq!(response["failure_kind"], serde_json::Value::Null);
-                .as_array()
+        assert_eq!(
-                .unwrap()
+            response["expected_protocol_version"],
-                .iter()
+            RUNTIME_HTTP_PROTOCOL_VERSION
-                .any(|value| { value == "workers.spawn:incompatible" })
         );
-        assert!(
+        assert_eq!(
-            response["diagnostics"]
+            response["actual_protocol_version"],
-                .as_array()
+            RUNTIME_HTTP_PROTOCOL_VERSION
-                .unwrap()
-                .iter()
-                .any(|diagnostic| {
-                    diagnostic["code"] == "remote_runtime_worker_creation_unavailable"
-                })
         );
+        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]

crates/yoi/src/main.rs

@@ -363,10 +363,7 @@ fn parse_args_slice_with_connection_resolver<R: CliConnectionResolver + ?Sized>(
             &workspace_root,
         )?;
         let mode = if target.kind() == client::TargetKind::Backend {
-            LaunchMode::Workers {
+            LaunchMode::BackendSpawn
-                runtime_id: None,
-                include_stopped: false,
-            }
         } else {
             LaunchMode::Spawn {
                 worker_name: None,
@@ -822,10 +819,7 @@ fn parse_console_options<R: CliConnectionResolver + ?Sized>(
                     .to_string(),
             ));
         }
-        LaunchMode::Workers {
+        LaunchMode::BackendSpawn
-            runtime_id: None,
-            include_stopped: false,
-        }
     };
 
     Ok(Mode::Tui {
@@ -1720,7 +1714,7 @@ Target selection:
   Ticket, Objective, Worker catalog, PID, socket, or subprocess authority.
 
 Connection-aware commands:
-  yoi                         Standalone: new Console. Backend: Worker picker.
+  yoi                         Standalone: new Console. Backend: create and attach to a new Worker.
   yoi resume                  Standalone Worker picker or stopped Backend Worker picker.
   yoi workers                 Backend Workspace Worker picker.
   yoi panel                   Backend Workspace dashboard.
@@ -2115,6 +2109,46 @@ backend = "shared"
         }
     }
 
+    #[test]
+    fn parse_default_backend_creates_a_worker_before_attach() {
+        let resolver = DefaultBackendCliConnectionResolver {
+            backend_url: "http://default-backend.example",
+        };
+
+        match parse_args_slice_with_connection_resolver(&[], &resolver).unwrap() {
+            Mode::Tui {
+                target,
+                mode: LaunchMode::BackendSpawn,
+                ..
+            } => assert_eq!(target.kind(), TargetKind::Backend),
+            other => panic!("expected BackendSpawn mode, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn parse_bare_backend_creates_a_worker_before_attach() {
+        match parse_args_from([
+            "--backend",
+            "http://127.0.0.1:8787",
+            "--workspace-id",
+            "workspace-a",
+        ])
+        .unwrap()
+        {
+            Mode::Tui {
+                target,
+                mode: LaunchMode::BackendSpawn,
+                ..
+            } => {
+                assert_eq!(target.kind(), TargetKind::Backend);
+                let launch = target.launch_backend_worker().unwrap();
+                assert_eq!(launch.target.base_url, "http://127.0.0.1:8787");
+                assert_eq!(launch.target.workspace_id.as_deref(), Some("workspace-a"));
+            }
+            other => panic!("expected BackendSpawn mode, got {other:?}"),
+        }
+    }
+
     #[test]
     fn parse_workers_subcommand_uses_backend_runtime_picker() {
         match parse_args_from([

web/workspace/deno.json

@@ -6,7 +6,7 @@
     "dev": "deno run -A npm:vite@7.2.7 dev",
     "dev:backend": "cd ../.. && cargo run -p yoi-workspace-server --bin yoi-server -- serve --listen 127.0.0.1:8787",
     "check": "deno run -A npm:@sveltejs/kit@2.49.4 sync && deno run -A npm:svelte-check@4.3.4 --tsconfig ./tsconfig.json",
-    "test": "deno test --allow-read=src,test,tests --allow-env=LOG,VSCODE_TEXTMATE_DEBUG,NODE_ENV tests/workspace-model.test.ts tests/workspace-catalog.test.ts tests/profile-api.test.ts tests/skill-api.test.ts src/lib/workspace/auth/model.test.ts tests/auth-api.test.ts src/lib/workspace/api/http.test.ts src/lib/workspace/api/workers.test.ts src/lib/workspace/header/breadcrumb-model.test.ts src/lib/workspace/console/chat-submit.test.ts src/lib/workspace/console/composer-delivery.test.ts test/composer-history.test.ts tests/composer-paste.test.ts src/lib/workspace/console/composer-command.test.ts src/lib/workspace/console/composer-draft.test.ts src/lib/workspace/console/composer-completion.test.ts src/lib/workspace/console/markdown.test.ts test/console/ansi.test.ts src/lib/workspace/console/model.test.ts src/lib/workspace/companion/api.test.ts tests/workdir-api.test.ts src/lib/workspace/console/tasks.test.ts test/ticket-detail-route-reuse.test.ts test/repositories/ui.test.ts src/lib/workspace/console/worker-console.ui.test.ts src/lib/workspace/settings/model.test.ts src/lib/workspace/sidebar/override-stack.test.ts src/lib/workspace/sidebar/workers.test.ts src/lib/workspace/sidebar/workspace-switcher.test.ts src/lib/workspace/sidebar/worker-subscription.test.ts src/lib/workspace/sidebar/worker-launch.test.ts test/sidebar/worker-actions.test.ts src/lib/workspace/tickets/merge-request-resources.test.ts src/lib/workspace/tickets/ticket-panel.test.ts test/merge-request-status.test.ts test/config-source/decodal-grammar.test.ts test/config-source/editor-state.test.ts test/config-source/fixed-schema-wrapper.test.ts test/config-source/toolchain.test.ts test/config-source/wasm-parity.test.ts test/repository-access/api.test.ts test/repository-access/loader.test.ts test/repository-access/ui.test.ts",
+    "test": "deno test --allow-read=src,test,tests --allow-env=LOG,VSCODE_TEXTMATE_DEBUG,NODE_ENV tests/workspace-model.test.ts tests/workspace-catalog.test.ts tests/profile-api.test.ts tests/skill-api.test.ts src/lib/workspace/auth/model.test.ts tests/auth-api.test.ts src/lib/workspace/api/http.test.ts src/lib/workspace/api/workers.test.ts src/lib/workspace/header/breadcrumb-model.test.ts src/lib/workspace/console/chat-submit.test.ts src/lib/workspace/console/composer-delivery.test.ts test/composer-history.test.ts tests/composer-paste.test.ts src/lib/workspace/console/composer-command.test.ts src/lib/workspace/console/composer-draft.test.ts src/lib/workspace/console/composer-completion.test.ts src/lib/workspace/console/markdown.test.ts test/console/ansi.test.ts src/lib/workspace/console/model.test.ts src/lib/workspace/companion/api.test.ts tests/workdir-api.test.ts src/lib/workspace/console/tasks.test.ts test/ticket-detail-route-reuse.test.ts test/repositories/ui.test.ts src/lib/workspace/console/worker-console.ui.test.ts src/lib/workspace/settings/model.test.ts tests/runtime-connection.test.ts src/lib/workspace/sidebar/override-stack.test.ts src/lib/workspace/sidebar/workers.test.ts src/lib/workspace/sidebar/workspace-switcher.test.ts src/lib/workspace/sidebar/worker-subscription.test.ts src/lib/workspace/sidebar/worker-launch.test.ts test/sidebar/worker-actions.test.ts src/lib/workspace/tickets/merge-request-resources.test.ts src/lib/workspace/tickets/ticket-panel.test.ts test/merge-request-status.test.ts test/config-source/decodal-grammar.test.ts test/config-source/editor-state.test.ts test/config-source/fixed-schema-wrapper.test.ts test/config-source/toolchain.test.ts test/config-source/wasm-parity.test.ts test/repository-access/api.test.ts test/repository-access/loader.test.ts test/repository-access/ui.test.ts",
     "build": "deno run -A npm:vite@7.2.7 build",
     "preview": "deno run -A npm:vite@7.2.7 preview"
   },

web/workspace/src/lib/generated/workspace-api.ts

@@ -220,3 +220,27 @@ export type RepositoryLogResponse = {
   items: Array<GitCommitSummary>;
   diagnostics: Array<Diagnostic>;
 };
+
+export type RuntimeConnectionTestStatus = "compatible" | "failed";
+
+export type RuntimeConnectionTestFailureKind =
+  | "authentication"
+  | "authorization"
+  | "network_unreachable"
+  | "timeout"
+  | "tls_or_transport"
+  | "malformed_response"
+  | "protocol_version_mismatch"
+  | "runtime_identity_mismatch"
+  | "configuration";
+
+export type RuntimeConnectionTestResponse = {
+  workspace_id: string;
+  runtime_id: string;
+  checked_at: string;
+  status: RuntimeConnectionTestStatus;
+  failure_kind: RuntimeConnectionTestFailureKind | null;
+  expected_protocol_version: number;
+  actual_protocol_version: number | null;
+  diagnostics: Array<Diagnostic>;
+};

web/workspace/src/lib/workspace/api/runtime-connection.ts

@@ -0,0 +1,142 @@
+import type {
+  Diagnostic,
+  RuntimeConnectionTestFailureKind,
+  RuntimeConnectionTestResponse,
+} from "$lib/generated/workspace-api";
+
+const RESPONSE_KEYS = [
+  "workspace_id",
+  "runtime_id",
+  "checked_at",
+  "status",
+  "failure_kind",
+  "expected_protocol_version",
+  "actual_protocol_version",
+  "diagnostics",
+] as const;
+const DIAGNOSTIC_KEYS = ["code", "severity", "message"] as const;
+const FAILURE_KINDS = new Set<RuntimeConnectionTestFailureKind>([
+  "authentication",
+  "authorization",
+  "network_unreachable",
+  "timeout",
+  "tls_or_transport",
+  "malformed_response",
+  "protocol_version_mismatch",
+  "runtime_identity_mismatch",
+  "configuration",
+]);
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function hasExactKeys(
+  record: Record<string, unknown>,
+  expected: readonly string[],
+): boolean {
+  const actual = Object.keys(record).sort();
+  const wanted = [...expected].sort();
+  return actual.length === wanted.length &&
+    actual.every((key, index) => key === wanted[index]);
+}
+
+function isBoundedString(value: unknown, max = 1024): value is string {
+  return typeof value === "string" && value.length > 0 && value.length <= max;
+}
+
+function isProtocolVersion(value: unknown): value is number {
+  return Number.isSafeInteger(value) && (value as number) >= 0;
+}
+
+function parseDiagnostic(value: unknown): Diagnostic | null {
+  if (!isRecord(value) || !hasExactKeys(value, DIAGNOSTIC_KEYS)) return null;
+  if (!isBoundedString(value.code, 128) || !isBoundedString(value.message)) {
+    return null;
+  }
+  if (
+    value.severity !== "info" && value.severity !== "warning" &&
+    value.severity !== "error"
+  ) {
+    return null;
+  }
+  return {
+    code: value.code,
+    severity: value.severity,
+    message: value.message,
+  };
+}
+
+export function parseRuntimeConnectionTestResponse(
+  value: unknown,
+): RuntimeConnectionTestResponse | null {
+  if (!isRecord(value) || !hasExactKeys(value, RESPONSE_KEYS)) return null;
+  if (
+    !isBoundedString(value.workspace_id, 256) ||
+    !isBoundedString(value.runtime_id, 256) ||
+    !isBoundedString(value.checked_at, 128) ||
+    Number.isNaN(Date.parse(value.checked_at)) ||
+    (value.status !== "compatible" && value.status !== "failed") ||
+    !isProtocolVersion(value.expected_protocol_version) ||
+    (value.actual_protocol_version !== null &&
+      !isProtocolVersion(value.actual_protocol_version)) ||
+    !Array.isArray(value.diagnostics) ||
+    value.diagnostics.length > 16
+  ) {
+    return null;
+  }
+  const failureKind = value.failure_kind;
+  if (
+    failureKind !== null &&
+    !FAILURE_KINDS.has(failureKind as RuntimeConnectionTestFailureKind)
+  ) {
+    return null;
+  }
+  const diagnostics = value.diagnostics.map(parseDiagnostic);
+  if (diagnostics.some((diagnostic) => diagnostic === null)) return null;
+  if (
+    (value.status === "compatible" &&
+      (failureKind !== null ||
+        value.actual_protocol_version !== value.expected_protocol_version ||
+        diagnostics.length !== 0)) ||
+    (value.status === "failed" && failureKind === null)
+  ) {
+    return null;
+  }
+  return {
+    workspace_id: value.workspace_id,
+    runtime_id: value.runtime_id,
+    checked_at: value.checked_at,
+    status: value.status,
+    failure_kind: failureKind as RuntimeConnectionTestFailureKind | null,
+    expected_protocol_version: value.expected_protocol_version,
+    actual_protocol_version: value.actual_protocol_version,
+    diagnostics: diagnostics as Diagnostic[],
+  };
+}
+
+export async function testRuntimeConnection(
+  workspaceId: string,
+  runtimeId: string,
+  fetchImpl: typeof fetch = fetch,
+): Promise<RuntimeConnectionTestResponse> {
+  const response = await fetchImpl(
+    `/api/w/${encodeURIComponent(workspaceId)}/runtimes/${
+      encodeURIComponent(runtimeId)
+    }/connection-tests`,
+    { method: "POST" },
+  );
+  if (!response.ok) {
+    throw new Error(`Connection test failed (${response.status})`);
+  }
+  const parsed = parseRuntimeConnectionTestResponse(await response.json());
+  if (!parsed) {
+    throw new Error("Connection test returned an invalid response");
+  }
+  if (parsed.workspace_id !== workspaceId || parsed.runtime_id !== runtimeId) {
+    throw new Error(
+      "Connection test response did not match the selected Runtime",
+    );
+  }
+  return parsed;
+}

web/workspace/src/lib/workspace/console/worker-console.ui.test.ts

@@ -655,6 +655,9 @@ Deno.test("workspace Runtime inventory lives under Settings admin routes", async
       import.meta.url,
     ),
   );
+  const runtimeConnectionApi = await Deno.readTextFile(
+    new URL("../api/runtime-connection.ts", import.meta.url),
+  );
   const workdirsPage = await Deno.readTextFile(
     new URL(
       "./../../../routes/w/[workspaceId]/settings/runtimes/[runtimeId]/workdirs/+page.svelte",
@@ -678,9 +681,11 @@ Deno.test("workspace Runtime inventory lives under Settings admin routes", async
     runtimesPage.includes("Add remote Runtime") &&
       runtimesPage.includes("Open workdirs") &&
       runtimesPage.includes("settings-runtime-table") &&
-      runtimesPage.includes(
+      runtimesPage.includes("testRuntimeConnection") &&
-        "/runtimes/${encodeURIComponent(runtime.runtime_id)}/connection-tests",
+      runtimesPage.includes("data.workspaceId") &&
-      ) &&
+      runtimesPage.includes("runtime.runtime_id") &&
+      runtimeConnectionApi.includes("/runtimes/${") &&
+      runtimeConnectionApi.includes("}/connection-tests") &&
       runtimesPage.includes(
         "/settings/runtimes/${encodeURIComponent(runtime.runtime_id)}/workdirs",
       ),

web/workspace/src/lib/workspace/styles/settings.css

@@ -339,6 +339,9 @@
     background: rgba(255, 255, 255, 0.04);
     padding: 0.75rem;
   }
+  .settings-test-result.failed {
+    border-inline-start: 3px solid var(--danger);
+  }
   .settings-page {
     display: grid;
     gap: var(--space-5);

web/workspace/src/routes/w/[workspaceId]/settings/runtimes/+page.svelte

@@ -1,20 +1,11 @@
 <script lang="ts">
   import { invalidateAll } from '$app/navigation';
+  import type { RuntimeConnectionTestResponse } from '$lib/generated/workspace-api';
+  import { testRuntimeConnection } from '$lib/workspace/api/runtime-connection';
   import { workspaceApiPath } from '$lib/workspace/api/http';
-  import type { Diagnostic, Runtime } from '$lib/workspace/sidebar/types';
+  import type { Runtime } from '$lib/workspace/sidebar/types';
   import type { PageProps } from './$types';
 
-  type ConnectionTest = {
-    runtime_id: string;
-    checked_at: string;
-    state: string;
-    protocol_version?: string | null;
-    compatibility_basis: string;
-    capabilities: string[];
-    health_result: string;
-    diagnostics: Diagnostic[];
-  };
-
   let { data }: PageProps = $props();
   let runtimeId = $state('');
   let displayName = $state('');
@@ -22,12 +13,31 @@
   let showAddRuntime = $state(false);
   let busyRuntimeId = $state<string | null>(null);
   let requestError = $state<string | null>(null);
-  let testResults = $state<Record<string, ConnectionTest>>({});
+  let testResults = $state<Record<string, RuntimeConnectionTestResponse>>({});
 
   function runtimePlatform(runtime: Runtime): string {
     return runtime.os && runtime.arch ? `${runtime.os} / ${runtime.arch}` : 'Unknown';
   }
 
+  function connectionTestSummary(result: RuntimeConnectionTestResponse): string {
+    if (result.status === 'compatible') {
+      return `Compatible · protocol v${result.actual_protocol_version}`;
+    }
+    switch (result.failure_kind) {
+      case 'authentication': return 'Authentication failed';
+      case 'authorization': return 'Permission or Workspace scope rejected';
+      case 'network_unreachable': return 'Runtime unreachable';
+      case 'timeout': return 'Connection timed out';
+      case 'tls_or_transport': return 'TLS or transport failed';
+      case 'malformed_response': return 'Runtime returned an invalid ping response';
+      case 'protocol_version_mismatch':
+        return `Incompatible protocol · expected v${result.expected_protocol_version}, received v${result.actual_protocol_version ?? 'unknown'}`;
+      case 'runtime_identity_mismatch': return 'Runtime identity mismatch';
+      case 'configuration': return 'Runtime connection test is not configured';
+      default: return 'Connection test failed';
+    }
+  }
+
   function managementLabel(runtime: Runtime): string {
     if (runtime.management?.built_in) return 'Built-in';
     if (runtime.management?.config_managed) return 'Managed remote';
@@ -92,15 +102,7 @@
     requestError = null;
     busyRuntimeId = runtime.runtime_id;
     try {
-      const response = await fetch(
+      const result = await testRuntimeConnection(data.workspaceId, runtime.runtime_id);
-        workspaceApiPath(
-          data.workspaceId,
-          `/runtimes/${encodeURIComponent(runtime.runtime_id)}/connection-tests`,
-        ),
-        { method: 'POST' },
-      );
-      if (!response.ok) throw new Error(await responseError(response));
-      const result = await response.json() as ConnectionTest;
       testResults = { ...testResults, [runtime.runtime_id]: result };
     } catch (error) {
       requestError = error instanceof Error ? error.message : String(error);
@@ -229,10 +231,12 @@
                   {/if}
                   {#if testResults[runtime.runtime_id]}
                     {@const result = testResults[runtime.runtime_id]}
-                    <div class="settings-test-result">
+                    <div class:failed={result.status === 'failed'} class="settings-test-result">
-                      <strong>Connection test: {result.state}</strong>
+                      <strong>Connection test: {connectionTestSummary(result)}</strong>
-                      <span>{result.health_result}</span>
+                      {#if result.diagnostics[0]}
-                      <small>{result.compatibility_basis} · {result.checked_at}</small>
+                        <span>{result.diagnostics[0].message}</span>
+                      {/if}
+                      <small>Checked {new Date(result.checked_at).toLocaleString()}</small>
                     </div>
                   {/if}
                 </td>

web/workspace/tests/runtime-connection.test.ts

@@ -0,0 +1,103 @@
+declare const Deno: {
+  test(name: string, fn: () => void | Promise<void>): void;
+};
+
+import {
+  parseRuntimeConnectionTestResponse,
+  testRuntimeConnection,
+} from "../src/lib/workspace/api/runtime-connection.ts";
+
+function assertEquals(actual: unknown, expected: unknown): void {
+  if (JSON.stringify(actual) !== JSON.stringify(expected)) {
+    throw new Error(
+      `expected ${JSON.stringify(expected)}, got ${JSON.stringify(actual)}`,
+    );
+  }
+}
+
+function compatibleResponse(): Record<string, unknown> {
+  return {
+    workspace_id: "workspace-a",
+    runtime_id: "runtime-a",
+    checked_at: "2026-09-01T12:00:00Z",
+    status: "compatible",
+    failure_kind: null,
+    expected_protocol_version: 1,
+    actual_protocol_version: 1,
+    diagnostics: [],
+  };
+}
+
+Deno.test("runtime connection response accepts the exact compatible contract", () => {
+  assertEquals(
+    parseRuntimeConnectionTestResponse(compatibleResponse()),
+    compatibleResponse(),
+  );
+});
+
+Deno.test("runtime connection response rejects unknown fields and incoherent compatibility", () => {
+  assertEquals(
+    parseRuntimeConnectionTestResponse({
+      ...compatibleResponse(),
+      capabilities: ["shell"],
+    }),
+    null,
+  );
+  assertEquals(
+    parseRuntimeConnectionTestResponse({
+      ...compatibleResponse(),
+      actual_protocol_version: 2,
+    }),
+    null,
+  );
+  assertEquals(
+    parseRuntimeConnectionTestResponse({
+      ...compatibleResponse(),
+      failure_kind: "timeout",
+    }),
+    null,
+  );
+});
+
+Deno.test("runtime connection response rejects unknown failure kinds and unbounded diagnostics", () => {
+  const failed = {
+    ...compatibleResponse(),
+    status: "failed",
+    failure_kind: "future_failure",
+    actual_protocol_version: null,
+    diagnostics: [],
+  };
+  assertEquals(parseRuntimeConnectionTestResponse(failed), null);
+  assertEquals(
+    parseRuntimeConnectionTestResponse({
+      ...failed,
+      failure_kind: "timeout",
+      diagnostics: Array.from({ length: 17 }, () => ({
+        code: "timeout",
+        severity: "error",
+        message: "Timed out",
+      })),
+    }),
+    null,
+  );
+});
+
+Deno.test("runtime connection request rejects a mismatched response identity", async () => {
+  const fetchImpl = (() =>
+    Promise.resolve(
+      new Response(
+        JSON.stringify({ ...compatibleResponse(), runtime_id: "runtime-b" }),
+        { status: 200, headers: { "content-type": "application/json" } },
+      ),
+    )) as typeof fetch;
+  let message = "";
+  try {
+    await testRuntimeConnection("workspace-a", "runtime-a", fetchImpl);
+  } catch (error) {
+    message = error instanceof Error ? error.message : String(error);
+  }
+  assertEquals(
+    message,
+    "Connection test response did not match the selected Runtime",
+  );
+});