diff --git a/.yoi/tickets/00001KWMBAA6V/item.md b/.yoi/tickets/00001KWMBAA6V/item.md index c1229288..89b13978 100644 --- a/.yoi/tickets/00001KWMBAA6V/item.md +++ b/.yoi/tickets/00001KWMBAA6V/item.md @@ -1,9 +1,11 @@ --- title: 'Live-reload Runtime connection registry changes' -state: 'planning' +state: 'done' created_at: '2026-07-03T15:59:48Z' -updated_at: '2026-07-03T15:59:48Z' +updated_at: '2026-07-03T19:47:23Z' assignee: null +queued_by: 'yoi ticket' +queued_at: '2026-07-03T16:52:35Z' --- ## 背景 diff --git a/.yoi/tickets/00001KWMBAA6V/thread.md b/.yoi/tickets/00001KWMBAA6V/thread.md index fd280f90..7caab4e5 100644 --- a/.yoi/tickets/00001KWMBAA6V/thread.md +++ b/.yoi/tickets/00001KWMBAA6V/thread.md @@ -4,4 +4,237 @@ LocalTicketBackend によって作成されました。 +--- + + + +## Intake summary + +Marked ready by `yoi ticket state`. + +--- + + + +## State changed + +Marked ready by `yoi ticket state`. + + +--- + + + +## State changed + +Ticket を `yoi ticket` が queued にしました。 + + +--- + + + +## State changed + +State changed to `inprogress`. + + +--- + + + +## Implementation report + +実装報告。 + +- `RuntimeRegistry` を live mutation 可能な共有 registry にし、Settings の remote Runtime add/delete から `register_or_replace` / `unregister_if_idle` を呼ぶようにした。 +- add は persisted config 書き込み前に remote runtime config を解決・構築し、成功後に active registry へ反映する。成功時は `restart_required = false` と `runtime_registry_applied` diagnostic を返す。 +- delete は active registry 側に worker が残っている場合 `remote_runtime_delete_blocked` で拒否し、config は削除しない。worker が無ければ config と active registry の両方から削除する。 +- Settings Test 表示は connected / verified / unchecked warning を分け、接続できた範囲が分かるようにした。 +- `worker-runtime-rest-server` の default bind port を `127.0.0.1:38800` に固定し、default config コメントも同じ port に合わせた。 +- remote runtime blocking HTTP client/request は async context 直下で Tokio runtime を drop しない boundary 経由にした。 + +Validation: +- `cargo fmt --check` +- `git diff --check` +- `cargo test -p yoi-workspace-server` +- `cargo test -p worker-runtime --features ws-server` +- `cd web/workspace && deno task check && deno task test` + +Skipped: +- `nix build .#yoi --no-link` は重いため、ユーザー指示により未実行。 + + +--- + + + +## State changed + +State changed to `done`. + + +--- + + + +## Implementation report + +Follow-up after manual Browser test. + +Observed failure: +- Browser could connect/test `arcadia-test`, but worker creation failed with `workspace_worker_create_failed`: Runtime did not accept worker creation. + +Root cause: +- The standalone `worker-runtime-rest-server` started from `worker-runtime` has no `WorkerExecutionBackend`; it can serve control/catalog endpoints but cannot spawn Workers. +- The connection test previously reported spawn as unknown, while runtime launch candidates projected remote worker creation as available. + +Fix: +- Runtime summary now exposes `worker_creation_available` based on whether an execution backend is attached. +- Remote Runtime capability projection uses that flag for `can_spawn_worker` / `can_accept_input`. +- Settings connection test now marks `workers.spawn` incompatible with `remote_runtime_worker_creation_unavailable` when the Runtime has no execution backend. +- Added backend test covering a control-only Runtime with no execution backend. + +Validation: +- `cargo fmt --check` +- `git diff --check` +- `cargo test -p worker-runtime --features ws-server` +- `cargo test -p yoi-workspace-server` +- `cd web/workspace && deno task check && deno task test` + +Skipped: +- `nix build .#yoi --no-link` per earlier user instruction to skip heavy build. + + +--- + + + +## Implementation report + +Follow-up after `can_spawn_worker=false` review. + +User feedback: +- A registered Runtime that cannot spawn workers should not exist as a usable Runtime. + +Fix: +- Removed the control-only `worker-runtime` package binary surface by disabling Cargo autobin discovery and deleting `crates/worker-runtime/src/main.rs`. +- Added `worker-runtime-rest-server` under the `worker` package, where it can attach `WorkerRuntimeExecutionBackend` to the Runtime before serving REST/WebSocket endpoints. +- The server binary now starts a worker-backed Runtime, so `/v1/runtime` reports `worker_creation_available=true` for the intended standalone process. +- Retained fail-closed capability projection for old/control-only Runtime summaries so a process without execution backend is rejected as incompatible rather than exposed as a usable candidate. + +New local command: +- `cargo run -p worker --features runtime-adapter --bin worker-runtime-rest-server -- --workspace .` + +Validation: +- `cargo fmt --check` +- `git diff --check` +- `cargo check -p yoi` +- `cargo test -p worker-runtime --features ws-server` +- `cargo test -p worker --features runtime-adapter --bin worker-runtime-rest-server` +- `cargo test -p yoi-workspace-server` +- `cd web/workspace && deno task check && deno task test` +- Smoke: started worker-backed Runtime on `127.0.0.1:38801` and verified `/v1/runtime` contains `"worker_creation_available":true`. + +Skipped: +- `nix build .#yoi --no-link` per earlier user instruction to skip heavy build. + + +--- + + + +## Implementation report + +Follow-up after dependency-boundary review. + +User feedback: +- Putting the standalone Runtime server binary in `worker` was also wrong; `worker` should not depend on Runtime just to host the adapter. + +Correction: +- Split the Runtime/Worker bridge into a dedicated `worker-runtime-adapter` crate. +- Moved `WorkerRuntimeExecutionBackend` / `ProfileRuntimeWorkerFactory` out of `worker` and into `worker-runtime-adapter`. +- Moved the `worker-runtime-rest-server` binary into `worker-runtime-adapter`. +- Removed the `worker` -> `worker-runtime` dependency and removed the `runtime-adapter` feature from `worker`. +- `worker-runtime` remains the Runtime domain/API crate; `worker` remains Worker implementation; `worker-runtime-adapter` is the only crate depending on both. + +New local command: +- `cargo run -p worker-runtime-adapter --bin worker-runtime-rest-server -- --workspace .` + +Validation: +- `cargo fmt --check` +- `git diff --check` +- `cargo check -p worker` +- `cargo check -p yoi` +- `cargo test -p worker-runtime --features ws-server` +- `cargo test -p worker-runtime-adapter` +- `cargo test -p yoi-workspace-server` +- `cd web/workspace && deno task check && deno task test` + +Skipped: +- `nix build .#yoi --no-link` per earlier user instruction to skip heavy build. + + +--- + + + +## Implementation report + +Correction after dependency-boundary review. + +User feedback: +- A separate bridge crate is unnecessary; the Runtime should bundle/manage Workers directly. + +Correction: +- Removed `worker-runtime-adapter` crate entirely. +- Moved `WorkerRuntimeExecutionBackend` / `ProfileRuntimeWorkerFactory` into `worker-runtime::worker_backend`. +- Moved the `worker-runtime-rest-server` binary back into `worker-runtime`, but kept it worker-backed rather than the old control-only process. +- `worker` no longer depends on `worker-runtime`; `worker-runtime` now depends on `worker` and owns the Runtime process that bundles Workers. +- `workspace-server` now imports `worker_runtime::worker_backend::WorkerRuntimeExecutionBackend`. + +Correct local command: +- `cargo run -p worker-runtime --features ws-server,fs-store --bin worker-runtime-rest-server -- --workspace .` + +Validation: +- `cargo fmt --check` +- `git diff --check` +- `cargo check -p worker` +- `cargo check -p yoi` +- `cargo check -p worker-runtime --features ws-server,fs-store` +- `cargo test -p worker-runtime --features ws-server,fs-store` +- `cargo test -p yoi-workspace-server` +- `cd web/workspace && deno task check && deno task test` + +Skipped: +- `nix build .#yoi --no-link` per earlier user instruction to skip heavy build. + + +--- + + + +## Implementation report + +Follow-up after Runtime generation path review. + +User feedback: +- Runtime generation paths should not be duplicated; a Runtime that can be served over HTTP must be constructed explicitly with the intended Worker backend rather than implicitly from HTTP server config. + +Fix: +- Removed `RuntimeHttpServer::bind`. +- Removed `serve_configured_runtime_http`. +- Removed the internal `runtime_from_config` path that built backend-less Runtime instances from `RuntimeHttpServerConfig`. +- `http_server` now serves only an already-constructed `Runtime` via `serve_runtime_http(runtime, listener, token)`. + +Validation: +- `cargo fmt --check` +- `git diff --check` +- `cargo test -p worker-runtime --features ws-server,fs-store` +- `cargo test -p yoi-workspace-server` + +Skipped: +- `nix build .#yoi --no-link` per earlier user instruction to skip heavy build. + + --- diff --git a/Cargo.lock b/Cargo.lock index 203bf133..30b91e9f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5904,7 +5904,6 @@ dependencies = [ "uuid", "wasmtime", "wat", - "worker-runtime", "workflow", "yoi-plugin-pdk", ] @@ -5913,16 +5912,22 @@ dependencies = [ name = "worker-runtime" version = "0.1.0" dependencies = [ + "async-trait", "axum", "futures", + "llm-engine", + "manifest", "protocol", "serde", "serde_json", + "session-store", "sha2 0.11.0", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", "tower", + "worker", ] [[package]] diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index 3553f0bb..8ceab741 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -4,29 +4,37 @@ description = "Embedded memory-backed Runtime API for Worker management" version = "0.1.0" edition.workspace = true license.workspace = true +autobins = false [[bin]] name = "worker-runtime-rest-server" path = "src/main.rs" -required-features = ["http-server"] +required-features = ["ws-server", "fs-store"] [features] default = [] fs-store = ["dep:serde_json"] -http-server = ["dep:axum", "dep:serde_json", "dep:tokio", "dep:tower"] -ws-server = ["http-server", "axum/ws", "dep:futures", "dep:protocol", "tokio/sync"] +http-server = ["dep:axum", "dep:serde_json", "dep:tower"] +ws-server = ["http-server", "axum/ws", "dep:futures", "tokio/sync"] [dependencies] +async-trait.workspace = true axum = { workspace = true, optional = true } futures = { workspace = true, optional = true } -protocol = { workspace = true, optional = true } +manifest.workspace = true +protocol.workspace = true serde = { workspace = true, features = ["derive"] } +session-store.workspace = true sha2.workspace = true serde_json = { workspace = true, optional = true } thiserror = { workspace = true } -tokio = { workspace = true, features = ["net", "rt"], optional = true } +tokio = { workspace = true, features = ["net", "rt", "sync", "time"] } tower = { workspace = true, features = ["util"], optional = true } +worker.workspace = true [dev-dependencies] +futures.workspace = true +llm-engine.workspace = true +tempfile.workspace = true tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } tokio-tungstenite.workspace = true diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index a69f0ed9..48b2a185 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -12,11 +12,9 @@ use crate::catalog::{ }; use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary}; use crate::error::RuntimeError; -#[cfg(feature = "fs-store")] -use crate::fs_store::FsRuntimeStoreOptions; use crate::identity::{RuntimeId, WorkerId, WorkerRef}; use crate::interaction::{WorkerInput, WorkerInteractionAck}; -use crate::management::{RuntimeLimits, RuntimeOptions, RuntimeSummary}; +use crate::management::{RuntimeLimits, RuntimeSummary}; #[cfg(feature = "ws-server")] use crate::observation::WorkerObservationCursor; use crate::observation::{TranscriptProjection, TranscriptQuery}; @@ -40,6 +38,12 @@ use std::path::PathBuf; use std::sync::Arc; use tokio::net::TcpListener; +const DEFAULT_RUNTIME_HTTP_PORT: u16 = 38800; + +fn default_runtime_http_bind_addr() -> SocketAddr { + SocketAddr::from(([127, 0, 0, 1], DEFAULT_RUNTIME_HTTP_PORT)) +} + /// v0 Runtime REST server configuration. #[derive(Clone, PartialEq, Eq)] pub struct RuntimeHttpServerConfig { @@ -63,7 +67,7 @@ pub struct RuntimeHttpServerConfig { impl Default for RuntimeHttpServerConfig { fn default() -> Self { Self { - bind_addr: SocketAddr::from(([127, 0, 0, 1], 0)), + bind_addr: default_runtime_http_bind_addr(), runtime_id: None, display_name: None, limits: RuntimeLimits::default(), @@ -102,48 +106,6 @@ pub enum RuntimeHttpStoreSelection { }, } -/// Bound REST server instance. -pub struct RuntimeHttpServer { - runtime: Runtime, - local_token: Option, - listener: TcpListener, -} - -impl RuntimeHttpServer { - /// Build a Runtime from config and bind the configured address. - pub async fn bind(config: RuntimeHttpServerConfig) -> Result { - let runtime = runtime_from_config(&config)?; - let listener = TcpListener::bind(config.bind_addr).await?; - Ok(Self { - runtime, - local_token: config.local_token, - listener, - }) - } - - /// Address actually bound by the server. - pub fn local_addr(&self) -> Result { - Ok(self.listener.local_addr()?) - } - - /// Runtime owned by this server. - pub fn runtime(&self) -> Runtime { - self.runtime.clone() - } - - /// Serve requests until the axum server is stopped or returns an error. - pub async fn serve(self) -> Result<(), RuntimeHttpServerError> { - serve_runtime_http(self.runtime, self.listener, self.local_token).await - } -} - -/// Convenience entry point: bind and serve a configured Runtime REST process API. -pub async fn serve_configured_runtime_http( - config: RuntimeHttpServerConfig, -) -> Result<(), RuntimeHttpServerError> { - RuntimeHttpServer::bind(config).await?.serve().await -} - /// Serve an existing Runtime on a pre-bound listener. pub async fn serve_runtime_http( runtime: Runtime, @@ -194,27 +156,6 @@ pub fn runtime_http_router(runtime: Runtime, local_token: Option) -> Rou .layer(middleware::from_fn_with_state(state, require_local_token)) } -fn runtime_from_config( - config: &RuntimeHttpServerConfig, -) -> Result { - match &config.store { - RuntimeHttpStoreSelection::Memory => Ok(Runtime::with_options(RuntimeOptions { - runtime_id: config.runtime_id.clone(), - display_name: config.display_name.clone(), - limits: config.limits.clone(), - })), - #[cfg(feature = "fs-store")] - RuntimeHttpStoreSelection::Fs { root } => { - Ok(Runtime::with_fs_store(FsRuntimeStoreOptions { - root: root.clone(), - runtime_id: config.runtime_id.clone(), - display_name: config.display_name.clone(), - limits: config.limits.clone(), - })?) - } - } -} - #[derive(Clone)] struct RuntimeHttpState { runtime: Runtime, @@ -1134,7 +1075,7 @@ mod ws_tests { WorkerExecutionResult, WorkerExecutionRunState, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, }; - use crate::interaction::WorkerInput; + use crate::management::RuntimeOptions; use futures::{SinkExt, StreamExt}; use std::sync::Arc; use tokio_tungstenite::connect_async; diff --git a/crates/worker-runtime/src/lib.rs b/crates/worker-runtime/src/lib.rs index 52df2450..d300d499 100644 --- a/crates/worker-runtime/src/lib.rs +++ b/crates/worker-runtime/src/lib.rs @@ -1,11 +1,10 @@ //! Embedded Runtime domain API for Worker management. //! -//! `worker-runtime` keeps its core independent from HTTP/WebSocket servers, -//! provider execution, and the existing Worker host. Filesystem persistence is -//! available only through the optional `fs-store` feature, and the minimal REST -//! process adapter is available only through the optional `http-server` feature. -//! The crate defines the in-process Runtime authority surface that higher layers -//! can later adapt into registries or backend APIs. +//! `worker-runtime` owns the Runtime authority surface and, for the standalone +//! process, wires that Runtime to the real Worker host. Filesystem persistence is +//! available only through the optional `fs-store` feature, and the REST/WebSocket +//! server is available only through the optional `http-server` / `ws-server` +//! features. pub mod catalog; pub mod config_bundle; @@ -21,6 +20,7 @@ pub mod interaction; pub mod management; pub mod observation; mod runtime; +pub mod worker_backend; #[cfg(feature = "fs-store")] pub use fs_store::{FsRuntimeStore, FsRuntimeStoreOptions}; diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index 8b511d58..4b1e49e6 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -1,8 +1,8 @@ -//! Minimal Runtime REST process wrapper. +//! Worker-backed Runtime REST process wrapper. //! -//! This binary is available only when the `http-server` feature is enabled. It -//! starts a Runtime-local command API intended for a trusted backend/proxy; -//! browsers must not connect to this Runtime process directly. +//! This binary starts a Runtime command API with a real worker execution backend. +//! A REST Runtime process that cannot spawn Workers is not a valid Runtime for the +//! Workspace Browser. use std::collections::VecDeque; use std::env; @@ -11,11 +11,16 @@ use std::fmt; use std::net::SocketAddr; use std::path::PathBuf; use std::process::ExitCode; +use std::sync::Arc; +use worker_runtime::error::RuntimeError; +use worker_runtime::fs_store::FsRuntimeStoreOptions; use worker_runtime::http_server::{ - RuntimeHttpServer, RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection, + RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection, }; use worker_runtime::identity::RuntimeId; +use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend}; +use worker_runtime::{Runtime, RuntimeOptions}; fn main() -> ExitCode { match run() { @@ -40,25 +45,77 @@ fn run() -> Result<(), ProcessError> { }; let runtime = tokio::runtime::Builder::new_current_thread() - .enable_io() + .enable_all() .build()?; runtime.block_on(async move { - let server = RuntimeHttpServer::bind(config).await?; - let local_addr = server.local_addr()?; + let listener = tokio::net::TcpListener::bind(config.http.bind_addr).await?; + let local_addr = listener.local_addr()?; + let worker_runtime = build_runtime(&config)?; eprintln!( "worker-runtime REST server listening on {local_addr}; intended client is a trusted backend/proxy, not a browser" ); - server.serve().await + worker_runtime::http_server::serve_runtime_http( + worker_runtime, + listener, + config.http.local_token, + ) + .await + .map_err(ProcessError::from) })?; Ok(()) } -fn parse_args(args: I) -> Result, ProcessError> +fn build_runtime(config: &ProcessConfig) -> Result { + let mut factory = ProfileRuntimeWorkerFactory::new(config.workspace_root.clone()) + .with_cwd(config.cwd.clone()); + if let Some(store_dir) = config.worker_store_dir.clone() { + factory = factory.with_store_dir(store_dir); + } + if let Some(metadata_dir) = config.worker_metadata_dir.clone() { + factory = factory.with_worker_metadata_dir(metadata_dir); + } + if let Some(runtime_base_dir) = config.worker_runtime_base_dir.clone() { + factory = factory.with_runtime_base_dir(runtime_base_dir); + } + if let Some(profile) = config.profile.clone() { + factory = factory.with_profile(profile); + } + let backend = + Arc::new(WorkerRuntimeExecutionBackend::new(factory).map_err(ProcessError::WorkerAdapter)?); + + match &config.http.store { + RuntimeHttpStoreSelection::Memory => { + Runtime::with_execution_backend(runtime_options_from_http(&config.http), backend) + .map_err(ProcessError::Runtime) + } + RuntimeHttpStoreSelection::Fs { root } => { + let mut options = FsRuntimeStoreOptions::new(root.clone()); + options.runtime_id = config.http.runtime_id.clone(); + options.display_name = config.http.display_name.clone(); + options.limits = config.http.limits.clone(); + Runtime::with_fs_store_and_execution_backend(options, backend) + .map_err(ProcessError::Runtime) + } + _ => Err(ProcessError::usage( + "unsupported Runtime catalog store selection".to_string(), + )), + } +} + +fn runtime_options_from_http(config: &RuntimeHttpServerConfig) -> RuntimeOptions { + RuntimeOptions { + runtime_id: config.runtime_id.clone(), + display_name: config.display_name.clone(), + limits: config.limits.clone(), + } +} + +fn parse_args(args: I) -> Result, ProcessError> where I: IntoIterator, S: Into, { - let mut config = RuntimeHttpServerConfig::default(); + let mut config = ProcessConfig::default()?; let mut store = StoreArg::Memory; let mut args = args.into_iter().map(Into::into).collect::>(); @@ -68,18 +125,39 @@ where "--help" | "-h" => return Ok(None), "--bind" => { let value = take_value(&flag, inline_value, &mut args)?; - config.bind_addr = value.parse::().map_err(|error| { + config.http.bind_addr = value.parse::().map_err(|error| { ProcessError::usage(format!("invalid --bind socket address `{value}`: {error}")) })?; } "--runtime-id" => { let value = take_value(&flag, inline_value, &mut args)?; - config.runtime_id = Some(RuntimeId::new(value).ok_or_else(|| { + config.http.runtime_id = Some(RuntimeId::new(value).ok_or_else(|| { ProcessError::usage("--runtime-id must not be empty".to_string()) })?); } "--display-name" => { - config.display_name = Some(take_value(&flag, inline_value, &mut args)?); + config.http.display_name = Some(take_value(&flag, inline_value, &mut args)?); + } + "--workspace" => { + config.workspace_root = PathBuf::from(take_value(&flag, inline_value, &mut args)?); + } + "--cwd" => { + config.cwd = PathBuf::from(take_value(&flag, inline_value, &mut args)?); + } + "--worker-store-dir" => { + config.worker_store_dir = + Some(PathBuf::from(take_value(&flag, inline_value, &mut args)?)); + } + "--worker-metadata-dir" => { + config.worker_metadata_dir = + Some(PathBuf::from(take_value(&flag, inline_value, &mut args)?)); + } + "--worker-runtime-base-dir" => { + config.worker_runtime_base_dir = + Some(PathBuf::from(take_value(&flag, inline_value, &mut args)?)); + } + "--profile" => { + config.profile = Some(take_value(&flag, inline_value, &mut args)?); } "--store" => { let value = take_value(&flag, inline_value, &mut args)?; @@ -106,7 +184,7 @@ where "--local-token must not be empty when provided".to_string(), )); } - config.local_token = Some(value); + config.http.local_token = Some(value); } "--local-token-env" => { let name = take_value(&flag, inline_value, &mut args)?; @@ -120,14 +198,14 @@ where "--local-token-env `{name}` resolved to an empty value" ))); } - config.local_token = Some(value); + config.http.local_token = Some(value); } "--max-transcript-projection-items" => { - config.limits.max_transcript_projection_items = + config.http.limits.max_transcript_projection_items = parse_usize_flag(&flag, take_value(&flag, inline_value, &mut args)?)?; } "--max-event-batch-items" => { - config.limits.max_event_batch_items = + config.http.limits.max_event_batch_items = parse_usize_flag(&flag, take_value(&flag, inline_value, &mut args)?)?; } _ => { @@ -136,7 +214,10 @@ where } } - apply_store_selection(&mut config, store)?; + apply_store_selection(&mut config.http, store)?; + if config.cwd.as_os_str().is_empty() { + config.cwd = config.workspace_root.clone(); + } Ok(Some(config)) } @@ -180,31 +261,40 @@ fn apply_store_selection( config.store = RuntimeHttpStoreSelection::Memory; Ok(()) } - StoreArg::Fs { root } => apply_fs_store_selection(config, root), + StoreArg::Fs { root } => { + let root = root.ok_or_else(|| { + ProcessError::usage("--store fs requires --fs-root ".to_string()) + })?; + config.store = RuntimeHttpStoreSelection::Fs { root }; + Ok(()) + } } } -#[cfg(feature = "fs-store")] -fn apply_fs_store_selection( - config: &mut RuntimeHttpServerConfig, - root: Option, -) -> Result<(), ProcessError> { - let root = root - .ok_or_else(|| ProcessError::usage("--store fs requires --fs-root ".to_string()))?; - config.store = RuntimeHttpStoreSelection::Fs { root }; - Ok(()) +#[derive(Clone, Debug, PartialEq, Eq)] +struct ProcessConfig { + http: RuntimeHttpServerConfig, + workspace_root: PathBuf, + cwd: PathBuf, + worker_store_dir: Option, + worker_metadata_dir: Option, + worker_runtime_base_dir: Option, + profile: Option, } -#[cfg(not(feature = "fs-store"))] -fn apply_fs_store_selection( - _config: &mut RuntimeHttpServerConfig, - root: Option, -) -> Result<(), ProcessError> { - let _ = root; - Err(ProcessError::usage( - "fs store selection requires building worker-runtime with features `http-server,fs-store`" - .to_string(), - )) +impl ProcessConfig { + fn default() -> Result { + let workspace_root = env::current_dir()?; + Ok(Self { + http: RuntimeHttpServerConfig::default(), + cwd: workspace_root.clone(), + workspace_root, + worker_store_dir: None, + worker_metadata_dir: None, + worker_runtime_base_dir: None, + profile: None, + }) + } } #[derive(Clone, Debug, PartialEq, Eq)] @@ -217,6 +307,8 @@ enum StoreArg { enum ProcessError { Usage(String), Server(RuntimeHttpServerError), + Runtime(RuntimeError), + WorkerAdapter(String), Io(std::io::Error), } @@ -231,6 +323,8 @@ impl fmt::Display for ProcessError { match self { Self::Usage(message) => message.fmt(f), Self::Server(error) => error.fmt(f), + Self::Runtime(error) => error.fmt(f), + Self::WorkerAdapter(error) => error.fmt(f), Self::Io(error) => error.fmt(f), } } @@ -239,8 +333,9 @@ impl fmt::Display for ProcessError { impl Error for ProcessError { fn source(&self) -> Option<&(dyn Error + 'static)> { match self { - Self::Usage(_) => None, + Self::Usage(_) | Self::WorkerAdapter(_) => None, Self::Server(error) => Some(error), + Self::Runtime(error) => Some(error), Self::Io(error) => Some(error), } } @@ -260,14 +355,20 @@ impl From for ProcessError { fn usage() -> &'static str { "Usage: worker-runtime-rest-server [OPTIONS]\n\n\ -Starts the worker-runtime REST command API for a trusted backend/proxy.\n\ +Starts a worker-backed Runtime REST command API for a trusted backend/proxy.\n\ Browsers must not connect to this Runtime process directly.\n\n\ Options:\n\ - --bind Bind socket address (default: 127.0.0.1:0)\n\ + --bind Bind socket address (default: 127.0.0.1:38800)\n\ --runtime-id Runtime authority id (default: generated)\n\ --display-name Runtime display name\n\ - --store Store selection (default: memory)\n\ - --fs-root Filesystem store root; requires fs-store feature\n\ + --workspace Workspace root used for spawned Workers (default: cwd)\n\ + --cwd Process cwd used for spawned Workers (default: workspace)\n\ + --profile Force spawned Workers to use a Profile selector\n\ + --worker-store-dir Worker session store directory\n\ + --worker-metadata-dir Worker metadata directory\n\ + --worker-runtime-base-dir Worker controller runtime directory\n\ + --store Runtime catalog store selection (default: memory)\n\ + --fs-root Runtime catalog filesystem store root\n\ --local-token Minimal local bearer token placeholder\n\ --local-token-env Read local bearer token placeholder from env\n\ --max-transcript-projection-items Override transcript projection limit\n\ @@ -279,68 +380,84 @@ Options:\n\ mod tests { use super::*; - #[test] - fn parses_memory_runtime_process_config() { - let config = parse_args([ - "--bind", - "127.0.0.1:48181", - "--runtime-id=runtime-review", - "--display-name", - "review runtime", - "--store", - "memory", - "--local-token", - "local-placeholder", - "--max-transcript-projection-items", - "32", - "--max-event-batch-items=16", - ]) - .unwrap() - .unwrap(); - - assert_eq!( - config.bind_addr, - "127.0.0.1:48181".parse::().unwrap() - ); - assert_eq!( - config.runtime_id.as_ref().map(RuntimeId::as_str), - Some("runtime-review") - ); - assert_eq!(config.display_name.as_deref(), Some("review runtime")); - assert!(matches!(config.store, RuntimeHttpStoreSelection::Memory)); - assert_eq!(config.local_token.as_deref(), Some("local-placeholder")); - assert_eq!(config.limits.max_transcript_projection_items, 32); - assert_eq!(config.limits.max_event_batch_items, 16); - } - - #[cfg(feature = "fs-store")] - #[test] - fn parses_fs_store_runtime_process_config_when_feature_enabled() { - let config = parse_args(["--fs-root", "/tmp/yoi-worker-runtime-store"]) - .unwrap() - .unwrap(); - - assert!(matches!( - config.store, - RuntimeHttpStoreSelection::Fs { ref root } - if root == &PathBuf::from("/tmp/yoi-worker-runtime-store") - )); - } - - #[cfg(not(feature = "fs-store"))] - #[test] - fn rejects_fs_store_runtime_process_config_without_feature() { - let error = parse_args(["--store", "fs", "--fs-root", "/tmp/store"]).unwrap_err(); - - assert!( - error - .to_string() - .contains("requires building worker-runtime with features") - ); - } - #[test] fn help_does_not_start_server() { assert!(parse_args(["--help"]).unwrap().is_none()); } + + #[test] + fn default_bind_addr_is_stable_for_local_dev() { + let config = parse_args([] as [&str; 0]).unwrap().unwrap(); + + assert_eq!( + config.http.bind_addr, + "127.0.0.1:38800".parse::().unwrap() + ); + } + + #[test] + fn parses_memory_runtime_process_config() { + let config = parse_args([ + "--bind", + "127.0.0.1:0", + "--runtime-id=runtime-alpha", + "--display-name", + "Runtime Alpha", + "--workspace", + "/tmp/workspace", + "--cwd", + "/tmp/workspace/subdir", + "--profile", + "builtin:coder", + "--local-token-env", + "TEST_RUNTIME_TOKEN", + ]); + assert!(config.is_err()); + + unsafe { + env::set_var("TEST_RUNTIME_TOKEN", "secret-token"); + } + let config = parse_args([ + "--bind", + "127.0.0.1:0", + "--runtime-id=runtime-alpha", + "--display-name", + "Runtime Alpha", + "--workspace", + "/tmp/workspace", + "--cwd", + "/tmp/workspace/subdir", + "--profile", + "builtin:coder", + "--local-token-env", + "TEST_RUNTIME_TOKEN", + ]) + .unwrap() + .unwrap(); + unsafe { + env::remove_var("TEST_RUNTIME_TOKEN"); + } + + assert_eq!(config.http.bind_addr, "127.0.0.1:0".parse().unwrap()); + assert_eq!( + config.http.runtime_id.as_ref().map(ToString::to_string), + Some("runtime-alpha".to_string()) + ); + assert_eq!(config.http.display_name.as_deref(), Some("Runtime Alpha")); + assert_eq!(config.workspace_root, PathBuf::from("/tmp/workspace")); + assert_eq!(config.cwd, PathBuf::from("/tmp/workspace/subdir")); + assert_eq!(config.profile.as_deref(), Some("builtin:coder")); + assert_eq!(config.http.local_token.as_deref(), Some("secret-token")); + } + + #[test] + fn parses_fs_store_runtime_process_config() { + let config = parse_args(["--store", "fs", "--fs-root", "/tmp/runtime-store"]) + .unwrap() + .unwrap(); + assert!(matches!( + config.http.store, + RuntimeHttpStoreSelection::Fs { ref root } if root == &PathBuf::from("/tmp/runtime-store") + )); + } } diff --git a/crates/worker-runtime/src/management.rs b/crates/worker-runtime/src/management.rs index e9a926f0..008355e5 100644 --- a/crates/worker-runtime/src/management.rs +++ b/crates/worker-runtime/src/management.rs @@ -65,4 +65,6 @@ pub struct RuntimeSummary { pub cancelled_worker_count: usize, pub diagnostic_count: usize, pub limits: RuntimeLimits, + #[serde(default)] + pub worker_creation_available: bool, } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 73e736d3..6d6373a1 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -161,6 +161,7 @@ impl Runtime { cancelled_worker_count, diagnostic_count: state.diagnostics.len(), limits: state.limits.clone(), + worker_creation_available: state.execution_backend.is_some(), }) } diff --git a/crates/worker/src/runtime_adapter.rs b/crates/worker-runtime/src/worker_backend.rs similarity index 90% rename from crates/worker/src/runtime_adapter.rs rename to crates/worker-runtime/src/worker_backend.rs index 93581d52..2aef5af9 100644 --- a/crates/worker/src/runtime_adapter.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -14,6 +14,11 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; +use crate::execution::{ + WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult, + WorkerExecutionRunState, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, +}; +use crate::interaction::{WorkerInput, WorkerInputKind}; use async_trait::async_trait; use manifest::paths; use protocol::{Method, Segment, WorkerStatus}; @@ -21,13 +26,8 @@ use session_store::FsStore; use session_store::{CombinedStore, FsWorkerStore}; use tokio::runtime::Runtime; use tokio::sync::broadcast; -use worker_runtime::execution::{ - WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult, - WorkerExecutionRunState, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, -}; -use worker_runtime::interaction::{WorkerInput, WorkerInputKind}; -use crate::{Worker, WorkerController, WorkerHandle}; +use worker::{Worker, WorkerController, WorkerHandle}; const DEFAULT_BACKEND_ID: &str = "worker-crate"; const RUNTIME_TASK_TIMEOUT: Duration = Duration::from_secs(10); @@ -115,7 +115,7 @@ impl ProfileRuntimeWorkerFactory { fn runtime_base_dir(&self) -> Result { self.runtime_base_dir .clone() - .or_else(|| crate::runtime::dir::default_base().ok()) + .or_else(|| worker::runtime::dir::default_base().ok()) .ok_or_else(|| "could not resolve worker runtime directory".to_string()) } @@ -124,14 +124,14 @@ impl ProfileRuntimeWorkerFactory { } fn runtime_profile_value( - profile: &worker_runtime::catalog::ProfileSelector, + profile: &crate::catalog::ProfileSelector, ) -> Option> { match profile { - worker_runtime::catalog::ProfileSelector::RuntimeDefault => None, - worker_runtime::catalog::ProfileSelector::Named(name) => { + crate::catalog::ProfileSelector::RuntimeDefault => None, + crate::catalog::ProfileSelector::Named(name) => { Some(std::borrow::Cow::Borrowed(name.as_str())) } - worker_runtime::catalog::ProfileSelector::Builtin(name) => { + crate::catalog::ProfileSelector::Builtin(name) => { if name.starts_with("builtin:") { Some(std::borrow::Cow::Borrowed(name.as_str())) } else { @@ -160,7 +160,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { ) -> Result { let worker_name = Self::runtime_worker_name(&request); let profile = self.runtime_profile(&request); - let (mut manifest, loader) = crate::entrypoint::resolve_runtime_profile_manifest( + let (mut manifest, loader) = worker::entrypoint::resolve_runtime_profile_manifest( profile.as_deref(), &self.workspace_root, &worker_name, @@ -211,7 +211,7 @@ pub struct WorkerRuntimeExecutionBackend { backend_id: String, factory: Arc, runtime: Mutex>, - workers: Mutex>, + workers: Mutex>, } impl WorkerRuntimeExecutionBackend { @@ -461,7 +461,7 @@ where }, WorkerExecutionRunState::Busy, ); - if result.outcome != worker_runtime::execution::WorkerExecutionOutcome::Accepted { + if result.outcome != crate::execution::WorkerExecutionOutcome::Accepted { busy.store(false, Ordering::SeqCst); } result @@ -506,17 +506,17 @@ mod tests { use std::pin::Pin; use std::sync::atomic::{AtomicUsize, Ordering}; + use crate::Runtime as EmbeddedRuntime; + use crate::catalog::{ConfigBundleRef, CreateWorkerRequest, ProfileSelector}; + use crate::identity::RuntimeId; + use crate::management::RuntimeOptions; + use crate::observation::{TranscriptQuery, TranscriptRole}; use async_trait::async_trait; use futures::Stream; use llm_engine::Engine; use llm_engine::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent}; use llm_engine::llm_client::{ClientError, LlmClient, Request}; use manifest::{Scope, WorkerManifest}; - use worker_runtime::Runtime as EmbeddedRuntime; - use worker_runtime::catalog::{CreateWorkerRequest, ProfileSelector, WorkerIntent}; - use worker_runtime::identity::RuntimeId; - use worker_runtime::management::RuntimeOptions; - use worker_runtime::observation::{TranscriptQuery, TranscriptRole}; #[derive(Clone)] struct MockClient { @@ -619,17 +619,37 @@ mod tests { ] } - fn create_request(name: &str) -> CreateWorkerRequest { - CreateWorkerRequest { - intent: WorkerIntent::Role { - role: name.to_string(), - purpose: None, + fn test_bundle() -> crate::config_bundle::ConfigBundle { + crate::config_bundle::ConfigBundle { + metadata: crate::config_bundle::ConfigBundleMetadata { + id: "adapter-test-bundle".to_string(), + digest: String::new(), + revision: "test".to_string(), + workspace_id: "adapter-test".to_string(), + created_at: "test".to_string(), + provenance: crate::config_bundle::ConfigBundleProvenance { + source: "test".to_string(), + detail: None, + }, }, + profiles: vec![crate::config_bundle::ConfigProfileDescriptor { + selector: ProfileSelector::RuntimeDefault, + label: Some("adapter-test".to_string()), + }], + declarations: Vec::new(), + } + .with_computed_digest() + } + + fn create_request(_name: &str) -> CreateWorkerRequest { + let bundle = test_bundle(); + CreateWorkerRequest { profile: ProfileSelector::RuntimeDefault, - config_bundle: None, - requested_capabilities: Vec::new(), - workspace_refs: Vec::new(), - mount_refs: Vec::new(), + config_bundle: ConfigBundleRef { + id: bundle.metadata.id, + digest: bundle.metadata.digest, + }, + initial_input: None, } } @@ -637,14 +657,14 @@ mod tests { fn builtin_profile_selector_is_not_double_prefixed() { assert_eq!( ProfileRuntimeWorkerFactory::runtime_profile_value( - &worker_runtime::catalog::ProfileSelector::Builtin("coder".to_string()) + &crate::catalog::ProfileSelector::Builtin("coder".to_string()) ) .as_deref(), Some("builtin:coder") ); assert_eq!( ProfileRuntimeWorkerFactory::runtime_profile_value( - &worker_runtime::catalog::ProfileSelector::Builtin("builtin:coder".to_string()) + &crate::catalog::ProfileSelector::Builtin("builtin:coder".to_string()) ) .as_deref(), Some("builtin:coder") @@ -673,6 +693,7 @@ mod tests { Arc::new(backend), ) .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); let detail = runtime.create_worker(create_request("chat")).unwrap(); runtime @@ -700,7 +721,7 @@ mod tests { let observations = runtime .read_worker_observation_events( &detail.worker_ref, - worker_runtime::observation::WorkerObservationCursor::zero(), + crate::observation::WorkerObservationCursor::zero(), ) .unwrap(); assert!( diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index d7a11e5f..22a8e9b1 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -7,7 +7,6 @@ autobins = false [features] default = [] -runtime-adapter = ["dep:worker-runtime"] [dependencies] async-trait = { workspace = true } @@ -19,7 +18,6 @@ mcp = { workspace = true } protocol = { workspace = true } provider = { workspace = true } client = { workspace = true } -worker-runtime = { workspace = true, features = ["ws-server"], optional = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } reqwest = { version = "0.13", default-features = false, features = ["blocking", "native-tls"] } diff --git a/crates/worker/src/entrypoint.rs b/crates/worker/src/entrypoint.rs index 981726cd..4e41bb81 100644 --- a/crates/worker/src/entrypoint.rs +++ b/crates/worker/src/entrypoint.rs @@ -225,8 +225,7 @@ fn load_profile( Ok((resolved.manifest, PromptLoader::builtins_only())) } -#[cfg(feature = "runtime-adapter")] -pub(crate) fn resolve_runtime_profile_manifest( +pub fn resolve_runtime_profile_manifest( profile: Option<&str>, workspace_root: &Path, worker_name: &str, diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 1f89097d..399dfc4c 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -10,8 +10,6 @@ pub(crate) mod in_flight; pub mod ipc; pub mod prompt; pub mod runtime; -#[cfg(feature = "runtime-adapter")] -pub mod runtime_adapter; pub mod segment_log_sink; pub mod shared_state; mod shutdown_after_idle; diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index 11182650..31a4083a 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -23,7 +23,7 @@ thiserror.workspace = true ticket.workspace = true tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread", "sync"] } tokio-tungstenite.workspace = true -worker = { workspace = true, features = ["runtime-adapter"] } +worker.workspace = true worker-runtime = { workspace = true, features = ["ws-server", "fs-store"] } toml.workspace = true tracing.workspace = true diff --git a/crates/workspace-server/src/config.rs b/crates/workspace-server/src/config.rs index 5ba59434..fac1ec5f 100644 --- a/crates/workspace-server/src/config.rs +++ b/crates/workspace-server/src/config.rs @@ -299,7 +299,9 @@ impl ResolvedWorkspaceBackendConfig { } } -fn resolve_remote_runtime(config: &RemoteRuntimeConfigFile) -> Result { +pub(crate) fn resolve_remote_runtime( + config: &RemoteRuntimeConfigFile, +) -> Result { if let Some(token_ref) = config.token_ref.as_deref() { return Err(Error::Config(format!( "remote runtime `{}` uses token_ref `{token_ref}`, but secret ref resolution is not implemented for workspace backend config yet", diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 103a7bc0..4107aa59 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -6,7 +6,11 @@ use reqwest::header::{AUTHORIZATION, CONTENT_TYPE}; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; -use std::{path::PathBuf, sync::Arc, time::Duration}; +use std::{ + path::PathBuf, + sync::{Arc, RwLock}, + time::Duration, +}; use worker_runtime::catalog::{ ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail as EmbeddedWorkerDetail, WorkerStatus as EmbeddedWorkerStatus, @@ -45,6 +49,22 @@ const MAX_HOST_SCAN: usize = 256; const MAX_IDENTIFIER_LEN: usize = 120; const ID_DIGEST_HEX_LEN: usize = 16; +fn run_blocking_http(operation: F) -> T +where + T: Send + 'static, + F: FnOnce() -> T + Send + 'static, +{ + match tokio::runtime::Handle::try_current() { + Ok(handle) if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread => { + tokio::task::block_in_place(operation) + } + Ok(_) => std::thread::spawn(operation) + .join() + .expect("blocking HTTP thread panicked"), + Err(_) => operation(), + } +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct RuntimeDiagnostic { pub code: String, @@ -643,31 +663,95 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { } } +#[derive(Debug, Clone)] +pub enum RuntimeRegistryUnregisterResult { + Removed, + NotFound, + BlockedByWorkers { + worker_count: usize, + diagnostics: Vec, + }, +} + #[derive(Clone)] pub struct RuntimeRegistry { - runtimes: Vec>, + runtimes: Arc>>>, } impl RuntimeRegistry { pub fn new(runtimes: Vec>) -> Self { - Self { runtimes } + Self { + runtimes: Arc::new(RwLock::new(runtimes)), + } } pub fn for_workspace(embedded_runtime: EmbeddedWorkerRuntime) -> Self { Self::new(vec![Arc::new(embedded_runtime)]) } - pub fn register(&mut self, runtime: R) + pub fn register(&self, runtime: R) where R: WorkspaceWorkerRuntime + 'static, { - self.runtimes.push(Arc::new(runtime)); + self.runtimes + .write() + .expect("runtime registry lock poisoned") + .push(Arc::new(runtime)); + } + + pub fn register_or_replace(&self, runtime: R) + where + R: WorkspaceWorkerRuntime + 'static, + { + let runtime = Arc::new(runtime); + let runtime_id = runtime.runtime_id().to_string(); + let mut runtimes = self + .runtimes + .write() + .expect("runtime registry lock poisoned"); + runtimes.retain(|existing| existing.runtime_id() != runtime_id); + runtimes.push(runtime); + } + + pub fn unregister_if_idle( + &self, + runtime_id: &str, + worker_scan_limit: usize, + ) -> Result { + validate_backend_identifier("runtime_id", runtime_id)?; + let runtime = self + .runtimes_snapshot() + .into_iter() + .find(|runtime| runtime.runtime_id() == runtime_id); + if let Some(runtime) = runtime { + let worker_list = runtime.list_workers(worker_scan_limit); + if !worker_list.items.is_empty() { + return Ok(RuntimeRegistryUnregisterResult::BlockedByWorkers { + worker_count: worker_list.items.len(), + diagnostics: worker_list.diagnostics, + }); + } + } else { + return Ok(RuntimeRegistryUnregisterResult::NotFound); + } + + let mut runtimes = self + .runtimes + .write() + .expect("runtime registry lock poisoned"); + let before = runtimes.len(); + runtimes.retain(|runtime| runtime.runtime_id() != runtime_id); + if runtimes.len() == before { + Ok(RuntimeRegistryUnregisterResult::NotFound) + } else { + Ok(RuntimeRegistryUnregisterResult::Removed) + } } pub fn list_runtimes(&self, limit: usize) -> RuntimeList { let mut diagnostics = Vec::new(); let mut items = Vec::new(); - for runtime in self.runtimes.iter().take(limit) { + for runtime in self.runtimes_snapshot().iter().take(limit) { let summary = runtime.runtime_summary(limit); diagnostics.extend(summary.diagnostics.iter().cloned()); items.push(summary); @@ -679,7 +763,7 @@ impl RuntimeRegistry { pub fn list_hosts(&self, limit: usize) -> RuntimeList { let mut items = Vec::new(); let mut diagnostics = Vec::new(); - for runtime in &self.runtimes { + for runtime in self.runtimes_snapshot() { if items.len() >= limit { break; } @@ -694,7 +778,7 @@ impl RuntimeRegistry { pub fn list_workers(&self, limit: usize) -> RuntimeList { let mut items = Vec::new(); let mut diagnostics = Vec::new(); - for runtime in &self.runtimes { + for runtime in self.runtimes_snapshot() { if items.len() >= limit { break; } @@ -716,7 +800,7 @@ impl RuntimeRegistry { let mut host_found = false; let mut diagnostics = Vec::new(); let mut items = Vec::new(); - for runtime in &self.runtimes { + for runtime in self.runtimes_snapshot() { let host_list = runtime.list_hosts(MAX_HOST_SCAN); diagnostics.extend(host_list.diagnostics); if !host_list.items.iter().any(|host| host.host_id == host_id) { @@ -894,13 +978,23 @@ impl RuntimeRegistry { }) } + fn runtimes_snapshot(&self) -> Vec> { + self.runtimes + .read() + .expect("runtime registry lock poisoned") + .clone() + } + fn runtime( &self, runtime_id: &str, - ) -> Result<&Arc, RuntimeRegistryError> { + ) -> Result, RuntimeRegistryError> { self.runtimes + .read() + .expect("runtime registry lock poisoned") .iter() .find(|runtime| runtime.runtime_id() == runtime_id) + .cloned() .ok_or_else(|| RuntimeRegistryError::UnknownRuntime(runtime_id.to_string())) } } @@ -1548,7 +1642,7 @@ impl RemoteRuntimeConfig { display_name: display_name.into(), base_url: base_url.into(), bearer_token, - cached_capabilities: remote_runtime_capabilities(200, false), + cached_capabilities: remote_runtime_capabilities(200, false, false), cached_status: "configured".to_string(), timeout: Duration::from_secs(10), } @@ -1586,14 +1680,14 @@ impl RemoteWorkerRuntime { pub fn new(config: RemoteRuntimeConfig) -> Result { validate_backend_identifier("runtime_id", &config.runtime_id)?; let base_url = config.base_url.trim_end_matches('/').to_string(); - let http = BlockingHttpClient::builder() - .timeout(config.timeout) - .build() - .map_err(|err| RuntimeRegistryError::RuntimeOperationFailed { - runtime_id: config.runtime_id.clone(), - code: "remote_runtime_client_build_failed".to_string(), - message: err.to_string(), - })?; + let timeout = config.timeout; + let http = + run_blocking_http(move || BlockingHttpClient::builder().timeout(timeout).build()) + .map_err(|err| RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: config.runtime_id.clone(), + code: "remote_runtime_client_build_failed".to_string(), + message: err.to_string(), + })?; Ok(Self { host_id: host_id_for_remote_runtime(&config.runtime_id), runtime_id: config.runtime_id, @@ -1628,18 +1722,9 @@ impl RemoteWorkerRuntime { format!("{base}/v1/workers/{worker_id}/events/ws") } - fn authorize(&self, request: RequestBuilder) -> RequestBuilder { - let request = request.header(CONTENT_TYPE, "application/json"); - if let Some(token) = self.bearer_token.as_deref() { - request.header(AUTHORIZATION, format!("Bearer {token}")) - } else { - request - } - } - fn get_json(&self, path: &str) -> Result where - T: DeserializeOwned, + T: DeserializeOwned + Send + 'static, { self.send_json(self.http.get(self.endpoint(path))) } @@ -1647,38 +1732,43 @@ impl RemoteWorkerRuntime { fn post_json(&self, path: &str, body: &B) -> Result where B: Serialize + ?Sized, - T: DeserializeOwned, + T: DeserializeOwned + Send + 'static, { self.send_json(self.http.post(self.endpoint(path)).json(body)) } fn send_json(&self, request: RequestBuilder) -> Result where - T: DeserializeOwned, + T: DeserializeOwned + Send + 'static, { - let response = self - .authorize(request) - .send() - .map_err(|err| remote_reqwest_diagnostic(&self.runtime_id, err))?; - let status = response.status(); - if status.is_success() { - response.json::().map_err(|err| { - diagnostic( - "remote_runtime_malformed_response", - DiagnosticSeverity::Error, - format!( - "Remote Runtime returned malformed JSON for '{}': {err}", - self.runtime_id - ), - ) - }) - } else { - Err(remote_http_status_diagnostic( - &self.runtime_id, - status, - response, - )) - } + let runtime_id = self.runtime_id.clone(); + let bearer_token = self.bearer_token.clone(); + run_blocking_http(move || { + let request = request.header(CONTENT_TYPE, "application/json"); + let request = if let Some(token) = bearer_token.as_deref() { + request.header(AUTHORIZATION, format!("Bearer {token}")) + } else { + request + }; + let response = request + .send() + .map_err(|err| remote_reqwest_diagnostic(&runtime_id, err))?; + let status = response.status(); + if status.is_success() { + response.json::().map_err(|err| { + diagnostic( + "remote_runtime_malformed_response", + DiagnosticSeverity::Error, + format!( + "Remote Runtime returned malformed JSON for '{}': {err}", + runtime_id + ), + ) + }) + } else { + Err(remote_http_status_diagnostic(&runtime_id, status, response)) + } + }) } fn map_worker_summary(&self, summary: worker_runtime::catalog::WorkerSummary) -> WorkerSummary { @@ -1790,7 +1880,11 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } else { vec![self.host_id.clone()] }, - capabilities: remote_runtime_capabilities(limit, true), + capabilities: remote_runtime_capabilities( + limit, + true, + response.runtime.worker_creation_available, + ), diagnostics: vec![diagnostic( "remote_runtime_backend_proxy", DiagnosticSeverity::Info, @@ -1827,7 +1921,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { status: "configured".to_string(), observed_at: Utc::now().to_rfc3339(), last_seen_at: None, - capabilities: remote_runtime_capabilities(limit, true), + capabilities: remote_runtime_capabilities(limit, true, false), diagnostics: vec![diagnostic( "remote_runtime_backend_proxy", DiagnosticSeverity::Info, @@ -2489,14 +2583,18 @@ fn percent_encode(input: &str, keep: impl Fn(u8) -> bool) -> String { encoded } -fn remote_runtime_capabilities(limit: usize, available: bool) -> RuntimeCapabilitySummary { +fn remote_runtime_capabilities( + limit: usize, + available: bool, + worker_creation_available: bool, +) -> RuntimeCapabilitySummary { RuntimeCapabilitySummary { can_list_hosts: true, can_list_workers: available, can_get_worker: available, - can_spawn_worker: available, + can_spawn_worker: available && worker_creation_available, can_stop_worker: available, - can_accept_input: available, + can_accept_input: available && worker_creation_available, has_workspace_fs: false, has_shell: false, has_git: false, @@ -3327,6 +3425,19 @@ mod tests { ); } + #[tokio::test] + async fn remote_runtime_client_can_initialize_inside_tokio_context() { + let runtime = RemoteWorkerRuntime::new(RemoteRuntimeConfig::new( + "remote:async-init", + "Remote Async Init", + "http://127.0.0.1:9", + None, + )) + .unwrap(); + + assert_eq!(runtime.runtime_id(), "remote:async-init"); + } + #[test] fn remote_runtime_registry_routes_commands_without_browser_secret_leaks() { let worker_json = worker_json("remote:primary", "worker-remote-1"); @@ -3362,7 +3473,7 @@ mod tests { ), ]); let secret = "secret-token-do-not-leak".to_string(); - let mut registry = RuntimeRegistry::new(Vec::new()); + let registry = RuntimeRegistry::new(Vec::new()); registry.register( RemoteWorkerRuntime::new(RemoteRuntimeConfig::new( "remote:primary", @@ -3559,7 +3670,7 @@ mod tests { .to_string(), ), ]); - let mut registry = RuntimeRegistry::new(Vec::new()); + let registry = RuntimeRegistry::new(Vec::new()); registry.register( RemoteWorkerRuntime::new(RemoteRuntimeConfig::new( "remote:primary", @@ -3605,7 +3716,7 @@ mod tests { 401, json!({ "error": { "code": "unauthorized", "message": "bad token" } }).to_string(), )]); - let mut registry = RuntimeRegistry::new(Vec::new()); + let registry = RuntimeRegistry::new(Vec::new()); registry.register( RemoteWorkerRuntime::new(RemoteRuntimeConfig::new( "remote:primary", diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 73257b8e..8dec228e 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -12,20 +12,20 @@ use chrono::{SecondsFormat, Utc}; use futures::StreamExt; use serde::{Deserialize, Serialize}; use tokio::net::TcpListener; -use worker::runtime_adapter::WorkerRuntimeExecutionBackend; +use worker_runtime::worker_backend::WorkerRuntimeExecutionBackend; use crate::companion::{ CompanionCancelRequest, CompanionConsole, CompanionMessageRequest, CompanionMessageResponse, CompanionStatusResponse, CompanionTranscriptProjection, }; -use crate::config::{RemoteRuntimeConfigFile, WorkspaceBackendConfigFile}; +use crate::config::{RemoteRuntimeConfigFile, WorkspaceBackendConfigFile, resolve_remote_runtime}; use crate::hosts::{ ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime, RuntimeDiagnostic, RuntimeRegistry, - RuntimeSummary, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, - WorkerLifecycleResult, WorkerOperationState, WorkerSpawnAcceptanceRequirement, - WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, WorkerSummary, - WorkerTranscriptProjection, + RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerInputRequest, WorkerInputResult, + WorkerLifecycleRequest, WorkerLifecycleResult, WorkerOperationState, + WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, + WorkerSummary, WorkerTranscriptProjection, }; use crate::identity::WorkspaceIdentity; use crate::observation::{ @@ -171,7 +171,7 @@ impl WorkspaceApi { updated_at: config.workspace_created_at.clone(), }) .await?; - let mut runtime = RuntimeRegistry::for_workspace( + let runtime = RuntimeRegistry::for_workspace( EmbeddedWorkerRuntime::new_fs_store_with_execution_backend( config.workspace_id.clone(), config.embedded_runtime_store_root.clone(), @@ -782,7 +782,7 @@ async fn add_remote_runtime_connection( "a remote Runtime connection with that id is already configured", )); } - local_config.runtimes.remote.push(RemoteRuntimeConfigFile { + let remote_config = RemoteRuntimeConfigFile { id, endpoint: request.endpoint.trim().to_string(), display_name: request @@ -792,9 +792,30 @@ async fn add_remote_runtime_connection( .filter(|value| !value.is_empty()) .map(ToOwned::to_owned), token_ref: None, - }); + }; + let active_config = remote_runtime_config_from_file(&remote_config).map_err(|diagnostic| { + ApiError::with_diagnostics( + Error::RuntimeOperationFailed { + runtime_id: remote_config.id.clone(), + code: diagnostic.code.clone(), + message: diagnostic.message.clone(), + }, + vec![diagnostic], + ) + })?; + let active_runtime = RemoteWorkerRuntime::new(active_config).map_err(|err| err.into_error())?; + local_config.runtimes.remote.push(remote_config); write_workspace_backend_config_for_settings(&api, &local_config)?; - let mut response = runtime_connection_mutation_response(&api, &local_config); + api.runtime.register_or_replace(active_runtime); + let mut response = runtime_connection_mutation_response( + &api, + &local_config, + vec![settings_diagnostic( + "runtime_registry_applied", + DiagnosticSeverity::Info, + "Remote Runtime config was persisted and applied to the active Runtime registry without restarting the Workspace backend.", + )], + ); response.diagnostics.push(settings_diagnostic( "workspace_backend_config_rewritten", DiagnosticSeverity::Info, @@ -822,8 +843,44 @@ async fn delete_remote_runtime_connection( if before == local_config.runtimes.remote.len() { return Err(Error::UnknownRuntime(runtime_id).into()); } + match api + .runtime + .unregister_if_idle(&runtime_id, api.config.max_records.min(200)) + .map_err(|err| err.into_error())? + { + RuntimeRegistryUnregisterResult::Removed | RuntimeRegistryUnregisterResult::NotFound => {} + RuntimeRegistryUnregisterResult::BlockedByWorkers { + worker_count, + diagnostics, + } => { + let mut diagnostics = diagnostics; + diagnostics.push(settings_diagnostic( + "remote_runtime_delete_blocked", + DiagnosticSeverity::Error, + format!( + "Remote Runtime '{runtime_id}' has {worker_count} active worker(s); stop or move them before deleting the connection." + ), + )); + return Err(ApiError::with_diagnostics( + Error::RuntimeOperationFailed { + runtime_id, + code: "remote_runtime_delete_blocked".to_string(), + message: "Remote Runtime connection has active workers".to_string(), + }, + diagnostics, + )); + } + } write_workspace_backend_config_for_settings(&api, &local_config)?; - let mut response = runtime_connection_mutation_response(&api, &local_config); + let mut response = runtime_connection_mutation_response( + &api, + &local_config, + vec![settings_diagnostic( + "runtime_registry_applied", + DiagnosticSeverity::Info, + "Remote Runtime config was removed from persisted config and the active Runtime registry without restarting the Workspace backend.", + )], + ); response.diagnostics.push(settings_diagnostic( "workspace_backend_config_rewritten", DiagnosticSeverity::Info, @@ -1277,16 +1334,13 @@ fn runtime_connection_settings_response( fn runtime_connection_mutation_response( api: &WorkspaceApi, local_config: &WorkspaceBackendConfigFile, + diagnostics: Vec, ) -> RuntimeConnectionMutationResponse { RuntimeConnectionMutationResponse { workspace_id: api.config.workspace_id.clone(), - restart_required: true, - remotes: remote_runtime_connection_summaries(api, local_config, true), - diagnostics: vec![settings_diagnostic( - "runtime_registry_restart_required", - DiagnosticSeverity::Warning, - "Runtime connection config changed; restart the Workspace backend for the live Runtime registry to use the new config.", - )], + restart_required: false, + remotes: remote_runtime_connection_summaries(api, local_config, false), + diagnostics, } } @@ -1443,6 +1497,18 @@ fn validate_public_runtime_id(runtime_id: &str) -> ApiResult<()> { Ok(()) } +fn remote_runtime_config_from_file( + remote: &RemoteRuntimeConfigFile, +) -> std::result::Result { + resolve_remote_runtime(remote).map_err(|err| { + settings_diagnostic( + "remote_runtime_apply_failed", + DiagnosticSeverity::Error, + err.to_string(), + ) + }) +} + async fn test_remote_runtime_config( api: &WorkspaceApi, remote: &RemoteRuntimeConfigFile, @@ -1529,7 +1595,10 @@ async fn test_remote_runtime_config( ); } }; - observation.available("runtime.summary", "Runtime summary was readable."); + 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, @@ -1544,7 +1613,10 @@ async fn test_remote_runtime_config( match probe_remote_json(&client, workers_url, "workers.list", "Worker list").await { Ok(payload) => match serde_json::from_value::(payload) { Ok(workers) => { - observation.available("workers.list", "Worker list was readable."); + observation.available( + "workers.list", + "Verified: /v1/workers responded with a recognized worker list.", + ); Some(workers) } Err(_) => { @@ -1574,7 +1646,10 @@ async fn test_remote_runtime_config( 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::(payload) { - Ok(_) => observation.available("workers.detail", "Worker detail was readable."), + 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( @@ -1591,13 +1666,13 @@ async fn test_remote_runtime_config( } else { observation.unknown( "workers.detail", - "Worker detail compatibility could not be proven because the Runtime reported no workers during the lightweight probe.", + "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", - "Worker event websocket URL can be constructed from the configured HTTP(S) Runtime endpoint, but no websocket connection was opened during this lightweight test.", + "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") { @@ -1621,8 +1696,10 @@ async fn test_remote_runtime_config( Ok(payload) => { match serde_json::from_value::(payload) { Ok(bundles) => { - observation - .available("config_bundles.list", "Config-bundle list was readable."); + observation.available( + "config_bundles.list", + "Verified: /v1/config-bundles responded with a recognized config-bundle list.", + ); Some(bundles) } Err(_) => { @@ -1665,7 +1742,7 @@ async fn test_remote_runtime_config( { Ok(_) => observation.available( "config_bundles.availability", - "Config-bundle availability was readable for an advertised bundle.", + "Verified: config-bundle availability was confirmed for an advertised bundle.", ), Err(_) => observation.incompatible( "config_bundles.availability", @@ -1686,21 +1763,32 @@ async fn test_remote_runtime_config( } else { observation.unknown( "config_bundles.availability", - "Config-bundle availability compatibility could not be proven because the Runtime advertised no bundles during the lightweight probe.", + "No connection problem found. Config-bundle availability was not checked because the remote Runtime advertised no bundles during the lightweight probe.", ); } - observation.unknown( - "workers.spawn", - "Worker spawn compatibility was not proven because the lightweight test does not create remote workers as a side effect.", - ); + 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", - "Worker input dispatch compatibility was not proven because the lightweight test does not send model-visible input as a side effect.", + "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", - "Config-bundle sync compatibility was not proven because the lightweight test does not upload bundles as a side effect.", + "No connection problem found. Config-bundle sync was not checked because this lightweight test does not upload bundles as a side effect.", ); RemoteRuntimeTestResponse { @@ -1709,11 +1797,14 @@ async fn test_remote_runtime_config( checked_at, state: observation.state().to_string(), protocol_version, - compatibility_basis: "Observed worker-runtime HTTP endpoints for summary, worker listing/detail when available, event websocket URL construction, and config-bundle listing/availability when available; side-effecting spawn/input/sync operations are reported unknown unless the Runtime API proves them.".to_string(), + 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!( - "runtime_status={:?}; incompatible={}; unknown={}", - summary.runtime.status, observation.incompatible_count, observation.unknown_count + "connected=true; runtime_status={:?}; available={}; incompatible={}; warnings={}", + summary.runtime.status, + observation.available_count, + observation.incompatible_count, + observation.unknown_count ), diagnostics: observation.diagnostics, } @@ -1747,13 +1838,20 @@ fn remote_runtime_test_failed( struct RuntimeCompatibilityObservation { capabilities: Vec, diagnostics: Vec, + available_count: usize, incompatible_count: usize, unknown_count: usize, } impl RuntimeCompatibilityObservation { - fn available(&mut self, operation: &str, _message: &str) { + fn available(&mut self, operation: &str, message: impl Into) { + 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) { @@ -1775,8 +1873,6 @@ impl RuntimeCompatibilityObservation { fn state(&self) -> &'static str { if self.incompatible_count > 0 { "incompatible" - } else if self.unknown_count > 0 { - "unknown" } else { "compatible" } @@ -2144,6 +2240,9 @@ impl IntoResponse for ApiError { Error::RuntimeOperationFailed { code, .. } if code == "remote_runtime_unsupported" => { StatusCode::NOT_IMPLEMENTED } + Error::RuntimeOperationFailed { code, .. } if code.ends_with("_blocked") => { + StatusCode::CONFLICT + } Error::RuntimeOperationFailed { code, .. } if code.starts_with("workspace_settings_") || code.starts_with("invalid_") @@ -2365,7 +2464,7 @@ mod tests { } #[tokio::test] - async fn runtime_connection_settings_add_delete_persist_restart_required() { + async fn runtime_connection_settings_add_delete_apply_live_registry() { let dir = tempfile::tempdir().unwrap(); let app = test_app(dir.path()).await; @@ -2383,9 +2482,16 @@ mod tests { }), ) .await; - assert_eq!(added["restart_required"], true); + assert_eq!(added["restart_required"], false); assert_eq!(added["remotes"][0]["runtime_id"], "team-runtime"); assert_eq!(added["remotes"][0]["endpoint_configured"], true); + assert!( + added["diagnostics"] + .as_array() + .unwrap() + .iter() + .any(|diagnostic| diagnostic["code"] == "runtime_registry_applied") + ); let projected = serde_json::to_string(&added).unwrap(); assert!(!projected.contains("runtime.example.invalid")); @@ -2397,22 +2503,113 @@ mod tests { "https://runtime.example.invalid" ); + let launch_options = get_json(app.clone(), "/api/workers/launch-options").await; + assert!( + launch_options["runtimes"] + .as_array() + .unwrap() + .iter() + .any(|runtime| runtime["runtime_id"] == "team-runtime") + ); + let deleted = request_json( - app, + app.clone(), "DELETE", "/api/settings/runtime-connections/remotes/team-runtime", None, StatusCode::OK, ) .await; - assert_eq!(deleted["restart_required"], true); + assert_eq!(deleted["restart_required"], false); assert_eq!(deleted["remotes"].as_array().unwrap().len(), 0); + let launch_options = get_json(app.clone(), "/api/workers/launch-options").await; + assert!( + !launch_options["runtimes"] + .as_array() + .unwrap() + .iter() + .any(|runtime| runtime["runtime_id"] == "team-runtime") + ); let persisted = WorkspaceBackendConfigFile::load_for_workspace(dir.path()).unwrap(); assert!(persisted.runtimes.remote.is_empty()); } - #[tokio::test] - async fn runtime_connection_test_reports_observed_and_unknown_capabilities_without_endpoint_leak() + #[tokio::test(flavor = "multi_thread")] + async fn runtime_connection_delete_rejects_active_remote_workers() { + 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 { + worker_runtime::http_server::serve_runtime_http(runtime, runtime_listener, None) + .await + .unwrap() + } + }); + + let dir = tempfile::tempdir().unwrap(); + let app = test_app(dir.path()).await; + let added = post_json( + app.clone(), + "/api/settings/runtime-connections/remotes", + serde_json::json!({ + "runtime_id": "busy-runtime", + "display_name": "Busy Runtime", + "endpoint": format!("http://{runtime_addr}") + }), + ) + .await; + assert_eq!(added["restart_required"], false); + let created = post_json( + app.clone(), + "/api/workers", + serde_json::json!({ + "runtime_id": "busy-runtime", + "display_name": "Remote Test Worker", + "profile": "runtime_default", + "initial_text": "" + }), + ) + .await; + assert_eq!(created["runtime_id"], "busy-runtime"); + let workers = get_json(app.clone(), "/api/workers").await; + assert!( + workers["items"] + .as_array() + .unwrap() + .iter() + .any(|worker| worker["runtime_id"] == "busy-runtime"), + "expected remote runtime to report at least one worker, got {workers}" + ); + + let response = request_json( + app, + "DELETE", + "/api/settings/runtime-connections/remotes/busy-runtime", + None, + StatusCode::CONFLICT, + ) + .await; + assert!( + response["message"] + .as_str() + .unwrap() + .contains("remote_runtime_delete_blocked") + ); + assert!( + response["diagnostics"] + .as_array() + .unwrap() + .iter() + .any(|diagnostic| { diagnostic["code"] == "remote_runtime_delete_blocked" }) + ); + let persisted = WorkspaceBackendConfigFile::load_for_workspace(dir.path()).unwrap(); + assert_eq!(persisted.runtimes.remote.len(), 1); + } + + #[tokio::test(flavor = "multi_thread")] + async fn runtime_connection_test_reports_compatible_with_unknown_warnings_without_endpoint_leak() { let (runtime, _worker_ref) = runtime_with_worker(); let runtime_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -2449,7 +2646,7 @@ mod tests { serde_json::json!({}), ) .await; - assert_eq!(response["state"], "unknown"); + assert_eq!(response["state"], "compatible"); let capabilities = response["capabilities"].as_array().unwrap(); assert!( capabilities @@ -2464,14 +2661,14 @@ mod tests { assert!( capabilities .iter() - .any(|value| value == "workers.spawn:unknown") + .any(|value| value == "workers.spawn:available") ); assert!( response["diagnostics"] .as_array() .unwrap() .iter() - .any(|diagnostic| { diagnostic["code"] == "workers.spawn.unknown" }) + .any(|diagnostic| { diagnostic["code"] == "workers.spawn.available" }) ); let projected = serde_json::to_string(&response).unwrap(); assert!(!projected.contains(&endpoint)); @@ -2479,6 +2676,60 @@ mod tests { assert_eq!(response["protocol_version"], serde_json::Value::Null); } + #[tokio::test(flavor = "multi_thread")] + async fn runtime_connection_test_marks_missing_execution_backend_incompatible() { + let runtime = + worker_runtime::Runtime::with_options(worker_runtime::RuntimeOptions::default()); + let runtime_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let runtime_addr = runtime_listener.local_addr().unwrap(); + tokio::spawn(async move { + worker_runtime::http_server::serve_runtime_http(runtime, runtime_listener, None) + .await + .unwrap() + }); + + let dir = tempfile::tempdir().unwrap(); + let endpoint = format!("http://{runtime_addr}"); + WorkspaceBackendConfigFile { + runtimes: crate::config::WorkspaceBackendRuntimesConfig { + remote: vec![RemoteRuntimeConfigFile { + id: "control-only-runtime".to_string(), + display_name: Some("Control-only Runtime".to_string()), + endpoint, + token_ref: None, + }], + }, + ..WorkspaceBackendConfigFile::default() + } + .write_for_workspace(dir.path()) + .unwrap(); + let app = test_app(dir.path()).await; + + let response = post_json( + app, + "/api/settings/runtime-connections/remotes/control-only-runtime/test", + serde_json::json!({}), + ) + .await; + assert_eq!(response["state"], "incompatible"); + assert!( + response["capabilities"] + .as_array() + .unwrap() + .iter() + .any(|value| { value == "workers.spawn:incompatible" }) + ); + assert!( + response["diagnostics"] + .as_array() + .unwrap() + .iter() + .any(|diagnostic| { + diagnostic["code"] == "remote_runtime_worker_creation_unavailable" + }) + ); + } + #[tokio::test] async fn browser_worker_create_succeeds_and_preserves_unsupported_diagnostics() { let dir = tempfile::tempdir().unwrap(); diff --git a/resources/workspace-backend.default.toml b/resources/workspace-backend.default.toml index 96cfedc2..8c75c4ed 100644 --- a/resources/workspace-backend.default.toml +++ b/resources/workspace-backend.default.toml @@ -46,6 +46,6 @@ max_records = 200 # # [[runtimes.remote]] # id = "example" -# endpoint = "http://127.0.0.1:8790" +# endpoint = "http://127.0.0.1:38800" # display_name = "Example Runtime" # token_ref = "local:example-runtime-token" diff --git a/web/workspace/src/lib/workspace-settings/SettingsPage.svelte b/web/workspace/src/lib/workspace-settings/SettingsPage.svelte index b2241cac..3c77b464 100644 --- a/web/workspace/src/lib/workspace-settings/SettingsPage.svelte +++ b/web/workspace/src/lib/workspace-settings/SettingsPage.svelte @@ -184,6 +184,13 @@ } } + function capabilityOperations(test: RemoteRuntimeTestResponse, state: "available" | "unknown" | "incompatible"): string[] { + const suffix = `:${state}`; + return test.capabilities + .filter((capability) => capability.endsWith(suffix)) + .map((capability) => capability.slice(0, -suffix.length)); + } + function applyRuntimeMutation(data: RuntimeConnectionMutationResponse) { runtimeSettings = runtimeSettings ? { ...runtimeSettings, remotes: data.remotes, diagnostics: data.diagnostics } @@ -335,10 +342,18 @@ {#if tests[remote.runtime_id]} {@const test = tests[remote.runtime_id]} + {@const available = capabilityOperations(test, "available")} + {@const unchecked = capabilityOperations(test, "unknown")}
Test: {test.state} {test.health_result} · {test.checked_at}

{test.compatibility_basis}

+ {#if available.length > 0} +

Verified areas: {available.join(', ')}

+ {/if} + {#if unchecked.length > 0} +

Unchecked warning areas: {unchecked.join(', ')}

+ {/if} {@render DiagnosticsList({ diagnostics: test.diagnostics })}
{/if}