feat: live reload runtime connections
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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<String>,
|
||||
listener: TcpListener,
|
||||
}
|
||||
|
||||
impl RuntimeHttpServer {
|
||||
/// Build a Runtime from config and bind the configured address.
|
||||
pub async fn bind(config: RuntimeHttpServerConfig) -> Result<Self, RuntimeHttpServerError> {
|
||||
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<SocketAddr, RuntimeHttpServerError> {
|
||||
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<String>) -> Rou
|
||||
.layer(middleware::from_fn_with_state(state, require_local_token))
|
||||
}
|
||||
|
||||
fn runtime_from_config(
|
||||
config: &RuntimeHttpServerConfig,
|
||||
) -> Result<Runtime, RuntimeHttpServerError> {
|
||||
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;
|
||||
|
||||
@@ -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};
|
||||
|
||||
+221
-104
@@ -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<I, S>(args: I) -> Result<Option<RuntimeHttpServerConfig>, ProcessError>
|
||||
fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
|
||||
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<I, S>(args: I) -> Result<Option<ProcessConfig>, ProcessError>
|
||||
where
|
||||
I: IntoIterator<Item = S>,
|
||||
S: Into<String>,
|
||||
{
|
||||
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::<VecDeque<_>>();
|
||||
|
||||
@@ -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::<SocketAddr>().map_err(|error| {
|
||||
config.http.bind_addr = value.parse::<SocketAddr>().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 <PATH>".to_string())
|
||||
})?;
|
||||
config.store = RuntimeHttpStoreSelection::Fs { root };
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "fs-store")]
|
||||
fn apply_fs_store_selection(
|
||||
config: &mut RuntimeHttpServerConfig,
|
||||
root: Option<PathBuf>,
|
||||
) -> Result<(), ProcessError> {
|
||||
let root = root
|
||||
.ok_or_else(|| ProcessError::usage("--store fs requires --fs-root <PATH>".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<PathBuf>,
|
||||
worker_metadata_dir: Option<PathBuf>,
|
||||
worker_runtime_base_dir: Option<PathBuf>,
|
||||
profile: Option<String>,
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "fs-store"))]
|
||||
fn apply_fs_store_selection(
|
||||
_config: &mut RuntimeHttpServerConfig,
|
||||
root: Option<PathBuf>,
|
||||
) -> 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<Self, ProcessError> {
|
||||
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<std::io::Error> 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 <ADDR> Bind socket address (default: 127.0.0.1:0)\n\
|
||||
--bind <ADDR> Bind socket address (default: 127.0.0.1:38800)\n\
|
||||
--runtime-id <ID> Runtime authority id (default: generated)\n\
|
||||
--display-name <NAME> Runtime display name\n\
|
||||
--store <memory|fs> Store selection (default: memory)\n\
|
||||
--fs-root <PATH> Filesystem store root; requires fs-store feature\n\
|
||||
--workspace <PATH> Workspace root used for spawned Workers (default: cwd)\n\
|
||||
--cwd <PATH> Process cwd used for spawned Workers (default: workspace)\n\
|
||||
--profile <SELECTOR> Force spawned Workers to use a Profile selector\n\
|
||||
--worker-store-dir <PATH> Worker session store directory\n\
|
||||
--worker-metadata-dir <PATH> Worker metadata directory\n\
|
||||
--worker-runtime-base-dir <PATH> Worker controller runtime directory\n\
|
||||
--store <memory|fs> Runtime catalog store selection (default: memory)\n\
|
||||
--fs-root <PATH> Runtime catalog filesystem store root\n\
|
||||
--local-token <TOKEN> Minimal local bearer token placeholder\n\
|
||||
--local-token-env <ENV> Read local bearer token placeholder from env\n\
|
||||
--max-transcript-projection-items <N> 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::<SocketAddr>().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::<SocketAddr>().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")
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,733 @@
|
||||
//! Adapter from `worker-runtime` execution backend boundary to the real
|
||||
//! `worker` crate controller/run lifecycle.
|
||||
//!
|
||||
//! The adapter intentionally owns real `WorkerHandle`s internally and exposes
|
||||
//! only the opaque `worker-runtime` execution handle to callers. Browser/API
|
||||
//! projections therefore keep the existing runtime redaction boundary: no raw
|
||||
//! socket paths, session paths, manifests, credentials, or handles leave this
|
||||
//! module.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::future::Future;
|
||||
use std::path::PathBuf;
|
||||
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};
|
||||
use session_store::FsStore;
|
||||
use session_store::{CombinedStore, FsWorkerStore};
|
||||
use tokio::runtime::Runtime;
|
||||
use tokio::sync::broadcast;
|
||||
|
||||
use worker::{Worker, WorkerController, WorkerHandle};
|
||||
|
||||
const DEFAULT_BACKEND_ID: &str = "worker-crate";
|
||||
const RUNTIME_TASK_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
|
||||
/// Factory seam used by [`WorkerRuntimeExecutionBackend`] to construct a real
|
||||
/// controller-backed Worker for a Runtime catalog entry.
|
||||
#[async_trait]
|
||||
pub trait RuntimeWorkerFactory: Send + Sync + 'static {
|
||||
async fn spawn_controller(
|
||||
&self,
|
||||
request: WorkerExecutionSpawnRequest,
|
||||
) -> Result<WorkerHandle, String>;
|
||||
}
|
||||
|
||||
/// Production factory that resolves a normal Worker profile and spawns it under
|
||||
/// `WorkerController`.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ProfileRuntimeWorkerFactory {
|
||||
workspace_root: PathBuf,
|
||||
cwd: PathBuf,
|
||||
store_dir: Option<PathBuf>,
|
||||
worker_metadata_dir: Option<PathBuf>,
|
||||
profile: Option<String>,
|
||||
runtime_base_dir: Option<PathBuf>,
|
||||
}
|
||||
|
||||
impl ProfileRuntimeWorkerFactory {
|
||||
pub fn new(workspace_root: impl Into<PathBuf>) -> Self {
|
||||
let workspace_root = workspace_root.into();
|
||||
Self {
|
||||
cwd: workspace_root.clone(),
|
||||
workspace_root,
|
||||
store_dir: None,
|
||||
worker_metadata_dir: None,
|
||||
profile: None,
|
||||
runtime_base_dir: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
|
||||
self.cwd = cwd.into();
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_store_dir(mut self, store_dir: impl Into<PathBuf>) -> Self {
|
||||
self.store_dir = Some(store_dir.into());
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_worker_metadata_dir(mut self, worker_metadata_dir: impl Into<PathBuf>) -> Self {
|
||||
self.worker_metadata_dir = Some(worker_metadata_dir.into());
|
||||
self
|
||||
}
|
||||
|
||||
/// Set the profile selector used for Runtime-created Workers. When unset,
|
||||
/// normal default profile discovery is used.
|
||||
pub fn with_profile(mut self, profile: impl Into<String>) -> Self {
|
||||
self.profile = Some(profile.into());
|
||||
self
|
||||
}
|
||||
|
||||
pub fn with_runtime_base_dir(mut self, runtime_base_dir: impl Into<PathBuf>) -> Self {
|
||||
self.runtime_base_dir = Some(runtime_base_dir.into());
|
||||
self
|
||||
}
|
||||
|
||||
fn store_dir(&self) -> Result<PathBuf, String> {
|
||||
self.store_dir
|
||||
.clone()
|
||||
.or_else(paths::sessions_dir)
|
||||
.ok_or_else(|| {
|
||||
"could not resolve sessions directory (set YOI_HOME, YOI_DATA_DIR, or HOME)"
|
||||
.to_string()
|
||||
})
|
||||
}
|
||||
|
||||
fn worker_metadata_dir(&self, store_dir: &std::path::Path) -> PathBuf {
|
||||
self.worker_metadata_dir
|
||||
.clone()
|
||||
.or_else(|| paths::data_dir().map(|data_dir| data_dir.join("workers")))
|
||||
.or_else(|| store_dir.parent().map(|parent| parent.join("workers")))
|
||||
.unwrap_or_else(|| PathBuf::from("workers"))
|
||||
}
|
||||
|
||||
fn runtime_base_dir(&self) -> Result<PathBuf, String> {
|
||||
self.runtime_base_dir
|
||||
.clone()
|
||||
.or_else(|| worker::runtime::dir::default_base().ok())
|
||||
.ok_or_else(|| "could not resolve worker runtime directory".to_string())
|
||||
}
|
||||
|
||||
fn runtime_worker_name(request: &WorkerExecutionSpawnRequest) -> String {
|
||||
request.worker_ref.worker_id.to_string()
|
||||
}
|
||||
|
||||
fn runtime_profile_value(
|
||||
profile: &crate::catalog::ProfileSelector,
|
||||
) -> Option<std::borrow::Cow<'_, str>> {
|
||||
match profile {
|
||||
crate::catalog::ProfileSelector::RuntimeDefault => None,
|
||||
crate::catalog::ProfileSelector::Named(name) => {
|
||||
Some(std::borrow::Cow::Borrowed(name.as_str()))
|
||||
}
|
||||
crate::catalog::ProfileSelector::Builtin(name) => {
|
||||
if name.starts_with("builtin:") {
|
||||
Some(std::borrow::Cow::Borrowed(name.as_str()))
|
||||
} else {
|
||||
Some(std::borrow::Cow::Owned(format!("builtin:{name}")))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn runtime_profile<'a>(
|
||||
&'a self,
|
||||
request: &'a WorkerExecutionSpawnRequest,
|
||||
) -> Option<std::borrow::Cow<'a, str>> {
|
||||
if let Some(profile) = self.profile.as_deref() {
|
||||
return Some(std::borrow::Cow::Borrowed(profile));
|
||||
}
|
||||
Self::runtime_profile_value(&request.request.profile)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
|
||||
async fn spawn_controller(
|
||||
&self,
|
||||
request: WorkerExecutionSpawnRequest,
|
||||
) -> Result<WorkerHandle, String> {
|
||||
let worker_name = Self::runtime_worker_name(&request);
|
||||
let profile = self.runtime_profile(&request);
|
||||
let (mut manifest, loader) = worker::entrypoint::resolve_runtime_profile_manifest(
|
||||
profile.as_deref(),
|
||||
&self.workspace_root,
|
||||
&worker_name,
|
||||
)?;
|
||||
manifest.worker.name = worker_name;
|
||||
|
||||
let store_dir = self.store_dir()?;
|
||||
let session_store = FsStore::new(&store_dir).map_err(|err| {
|
||||
format!(
|
||||
"failed to initialize session store at {}: {err}",
|
||||
store_dir.display()
|
||||
)
|
||||
})?;
|
||||
let worker_metadata_dir = self.worker_metadata_dir(&store_dir);
|
||||
let worker_metadata_store = FsWorkerStore::new(&worker_metadata_dir).map_err(|err| {
|
||||
format!(
|
||||
"failed to initialize worker metadata store at {}: {err}",
|
||||
worker_metadata_dir.display()
|
||||
)
|
||||
})?;
|
||||
let store = CombinedStore::new(session_store, worker_metadata_store);
|
||||
|
||||
let worker = Worker::from_manifest_with_context(
|
||||
manifest,
|
||||
store,
|
||||
loader,
|
||||
self.workspace_root.clone(),
|
||||
self.cwd.clone(),
|
||||
)
|
||||
.await
|
||||
.map_err(|err| format!("failed to create Worker from profile: {err}"))?;
|
||||
|
||||
let runtime_base = self.runtime_base_dir()?;
|
||||
let (handle, _shutdown_rx) = WorkerController::spawn(worker, &runtime_base)
|
||||
.await
|
||||
.map_err(|err| format!("failed to spawn Worker controller: {err}"))?;
|
||||
Ok(handle)
|
||||
}
|
||||
}
|
||||
|
||||
struct RuntimeWorkerExecution {
|
||||
handle: WorkerHandle,
|
||||
busy: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
/// `worker-runtime` execution backend backed by real `worker` crate Workers.
|
||||
pub struct WorkerRuntimeExecutionBackend<F = ProfileRuntimeWorkerFactory> {
|
||||
backend_id: String,
|
||||
factory: Arc<F>,
|
||||
runtime: Mutex<Option<Runtime>>,
|
||||
workers: Mutex<HashMap<crate::identity::WorkerRef, RuntimeWorkerExecution>>,
|
||||
}
|
||||
|
||||
impl WorkerRuntimeExecutionBackend<ProfileRuntimeWorkerFactory> {
|
||||
pub fn from_workspace(workspace_root: impl Into<PathBuf>) -> Result<Self, String> {
|
||||
Self::new(ProfileRuntimeWorkerFactory::new(workspace_root))
|
||||
}
|
||||
}
|
||||
|
||||
impl<F> WorkerRuntimeExecutionBackend<F>
|
||||
where
|
||||
F: RuntimeWorkerFactory,
|
||||
{
|
||||
pub fn new(factory: F) -> Result<Self, String> {
|
||||
let runtime = tokio::runtime::Builder::new_multi_thread()
|
||||
.thread_name("yoi-runtime-worker-adapter")
|
||||
.enable_all()
|
||||
.build()
|
||||
.map_err(|err| format!("failed to build worker adapter runtime: {err}"))?;
|
||||
Ok(Self {
|
||||
backend_id: DEFAULT_BACKEND_ID.to_string(),
|
||||
factory: Arc::new(factory),
|
||||
runtime: Mutex::new(Some(runtime)),
|
||||
workers: Mutex::new(HashMap::new()),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn with_backend_id(mut self, backend_id: impl Into<String>) -> Self {
|
||||
self.backend_id = backend_id.into();
|
||||
self
|
||||
}
|
||||
|
||||
fn wait_for_runtime_task<T>(receiver: mpsc::Receiver<Result<T, String>>) -> Result<T, String> {
|
||||
receiver
|
||||
.recv_timeout(RUNTIME_TASK_TIMEOUT)
|
||||
.map_err(|err| format!("worker adapter task did not complete: {err}"))?
|
||||
}
|
||||
|
||||
fn spawn_on_adapter_runtime<Fut>(&self, task: Fut) -> Result<(), String>
|
||||
where
|
||||
Fut: Future<Output = ()> + Send + 'static,
|
||||
{
|
||||
let runtime = self
|
||||
.runtime
|
||||
.lock()
|
||||
.map_err(|_| "worker adapter runtime lock is poisoned".to_string())?;
|
||||
let runtime = runtime
|
||||
.as_ref()
|
||||
.ok_or_else(|| "worker adapter runtime is shutting down".to_string())?;
|
||||
runtime.spawn(task);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn run_on_adapter_runtime<T, Fut>(&self, task: Fut) -> Result<T, String>
|
||||
where
|
||||
T: Send + 'static,
|
||||
Fut: Future<Output = Result<T, String>> + Send + 'static,
|
||||
{
|
||||
let (tx, rx) = mpsc::sync_channel(1);
|
||||
self.spawn_on_adapter_runtime(async move {
|
||||
let _ = tx.send(task.await);
|
||||
})?;
|
||||
Self::wait_for_runtime_task(rx)
|
||||
}
|
||||
|
||||
fn get_execution(
|
||||
&self,
|
||||
handle: &WorkerExecutionHandle,
|
||||
) -> Result<(WorkerHandle, Arc<AtomicBool>), WorkerExecutionResult> {
|
||||
if handle.backend_id() != self.backend_id() {
|
||||
return Err(WorkerExecutionResult::rejected(
|
||||
WorkerExecutionOperation::Input,
|
||||
format!(
|
||||
"execution handle belongs to backend {}, not {}",
|
||||
handle.backend_id(),
|
||||
self.backend_id()
|
||||
),
|
||||
));
|
||||
}
|
||||
let workers = self.workers.lock().map_err(|_| {
|
||||
WorkerExecutionResult::errored(
|
||||
WorkerExecutionOperation::Input,
|
||||
"worker adapter registry lock is poisoned",
|
||||
)
|
||||
})?;
|
||||
workers
|
||||
.get(handle.worker_ref())
|
||||
.map(|execution| (execution.handle.clone(), execution.busy.clone()))
|
||||
.ok_or_else(|| {
|
||||
WorkerExecutionResult::rejected(
|
||||
WorkerExecutionOperation::Input,
|
||||
"execution handle does not reference a live Worker",
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn send_method(
|
||||
&self,
|
||||
operation: WorkerExecutionOperation,
|
||||
worker: WorkerHandle,
|
||||
method: Method,
|
||||
accepted_run_state: WorkerExecutionRunState,
|
||||
) -> WorkerExecutionResult {
|
||||
self.run_on_adapter_runtime(async move {
|
||||
worker
|
||||
.send(method)
|
||||
.await
|
||||
.map_err(|err| format!("failed to send Worker method: {err}"))
|
||||
})
|
||||
.map(|_| WorkerExecutionResult::accepted(operation, accepted_run_state))
|
||||
.unwrap_or_else(|message| WorkerExecutionResult::errored(operation, message))
|
||||
}
|
||||
}
|
||||
|
||||
impl<F> Drop for WorkerRuntimeExecutionBackend<F> {
|
||||
fn drop(&mut self) {
|
||||
if let Ok(mut runtime) = self.runtime.lock()
|
||||
&& let Some(runtime) = runtime.take()
|
||||
{
|
||||
let _ = std::thread::spawn(move || drop(runtime)).join();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<F> WorkerExecutionBackend for WorkerRuntimeExecutionBackend<F>
|
||||
where
|
||||
F: RuntimeWorkerFactory,
|
||||
{
|
||||
fn backend_id(&self) -> &str {
|
||||
&self.backend_id
|
||||
}
|
||||
|
||||
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
|
||||
if self
|
||||
.workers
|
||||
.lock()
|
||||
.map(|workers| workers.contains_key(&request.worker_ref))
|
||||
.unwrap_or(false)
|
||||
{
|
||||
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::busy(
|
||||
WorkerExecutionOperation::Spawn,
|
||||
"Worker is already connected to execution backend",
|
||||
));
|
||||
}
|
||||
|
||||
let factory = self.factory.clone();
|
||||
let bridge_context = request.context.clone();
|
||||
let worker_ref = request.worker_ref.clone();
|
||||
let spawn_result =
|
||||
self.run_on_adapter_runtime(async move { factory.spawn_controller(request).await });
|
||||
|
||||
let handle = match spawn_result {
|
||||
Ok(handle) => handle,
|
||||
Err(message) => {
|
||||
return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored(
|
||||
WorkerExecutionOperation::Spawn,
|
||||
message,
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
let mut events = handle.subscribe();
|
||||
let bridge_handle = handle.clone();
|
||||
let busy = Arc::new(AtomicBool::new(false));
|
||||
let bridge_busy = busy.clone();
|
||||
if let Err(message) = self.spawn_on_adapter_runtime(async move {
|
||||
loop {
|
||||
match events.recv().await {
|
||||
Ok(event) => {
|
||||
let _ = bridge_context.publish_protocol_event(event);
|
||||
if bridge_handle.shared_state.get_status() == WorkerStatus::Idle {
|
||||
bridge_busy.store(false, Ordering::SeqCst);
|
||||
}
|
||||
}
|
||||
Err(broadcast::error::RecvError::Lagged(_)) => continue,
|
||||
Err(broadcast::error::RecvError::Closed) => break,
|
||||
}
|
||||
}
|
||||
}) {
|
||||
return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored(
|
||||
WorkerExecutionOperation::Spawn,
|
||||
message,
|
||||
));
|
||||
}
|
||||
|
||||
let mut workers = match self.workers.lock() {
|
||||
Ok(workers) => workers,
|
||||
Err(_) => {
|
||||
return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored(
|
||||
WorkerExecutionOperation::Spawn,
|
||||
"worker adapter registry lock is poisoned",
|
||||
));
|
||||
}
|
||||
};
|
||||
workers.insert(worker_ref.clone(), RuntimeWorkerExecution { handle, busy });
|
||||
|
||||
WorkerExecutionSpawnResult::Connected {
|
||||
handle: WorkerExecutionHandle::new(worker_ref, self.backend_id()),
|
||||
run_state: WorkerExecutionRunState::Idle,
|
||||
}
|
||||
}
|
||||
|
||||
fn dispatch_input(
|
||||
&self,
|
||||
handle: &WorkerExecutionHandle,
|
||||
input: WorkerInput,
|
||||
) -> WorkerExecutionResult {
|
||||
let (worker, busy) = match self.get_execution(handle) {
|
||||
Ok(execution) => execution,
|
||||
Err(mut result) => {
|
||||
result.operation = WorkerExecutionOperation::Input;
|
||||
return result;
|
||||
}
|
||||
};
|
||||
|
||||
if worker.shared_state.get_status() != WorkerStatus::Idle
|
||||
|| busy
|
||||
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
|
||||
.is_err()
|
||||
{
|
||||
return WorkerExecutionResult::busy(
|
||||
WorkerExecutionOperation::Input,
|
||||
"Worker is already running; runtime adapter v0 does not queue input",
|
||||
);
|
||||
}
|
||||
|
||||
let WorkerInputKind::User = input.kind else {
|
||||
busy.store(false, Ordering::SeqCst);
|
||||
return WorkerExecutionResult::unsupported(
|
||||
WorkerExecutionOperation::Input,
|
||||
"runtime adapter currently dispatches user input only",
|
||||
);
|
||||
};
|
||||
let content = input.content.trim().to_string();
|
||||
if content.is_empty() {
|
||||
busy.store(false, Ordering::SeqCst);
|
||||
return WorkerExecutionResult::rejected(
|
||||
WorkerExecutionOperation::Input,
|
||||
"runtime adapter rejects empty user input",
|
||||
);
|
||||
}
|
||||
|
||||
let result = self.send_method(
|
||||
WorkerExecutionOperation::Input,
|
||||
worker,
|
||||
Method::Run {
|
||||
input: vec![Segment::text(content)],
|
||||
},
|
||||
WorkerExecutionRunState::Busy,
|
||||
);
|
||||
if result.outcome != crate::execution::WorkerExecutionOutcome::Accepted {
|
||||
busy.store(false, Ordering::SeqCst);
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
|
||||
let (worker, _busy) = match self.get_execution(handle) {
|
||||
Ok(execution) => execution,
|
||||
Err(mut result) => {
|
||||
result.operation = WorkerExecutionOperation::Stop;
|
||||
return result;
|
||||
}
|
||||
};
|
||||
self.send_method(
|
||||
WorkerExecutionOperation::Stop,
|
||||
worker,
|
||||
Method::Shutdown,
|
||||
WorkerExecutionRunState::Stopped,
|
||||
)
|
||||
}
|
||||
|
||||
fn cancel_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
|
||||
let (worker, _busy) = match self.get_execution(handle) {
|
||||
Ok(execution) => execution,
|
||||
Err(mut result) => {
|
||||
result.operation = WorkerExecutionOperation::Cancel;
|
||||
return result;
|
||||
}
|
||||
};
|
||||
self.send_method(
|
||||
WorkerExecutionOperation::Cancel,
|
||||
worker,
|
||||
Method::Cancel,
|
||||
WorkerExecutionRunState::Idle,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
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};
|
||||
|
||||
#[derive(Clone)]
|
||||
struct MockClient {
|
||||
responses: Arc<Vec<Vec<LlmEvent>>>,
|
||||
call_count: Arc<AtomicUsize>,
|
||||
captured: Arc<Mutex<Vec<Request>>>,
|
||||
}
|
||||
|
||||
impl MockClient {
|
||||
fn new(events: Vec<LlmEvent>) -> Self {
|
||||
Self {
|
||||
responses: Arc::new(vec![events]),
|
||||
call_count: Arc::new(AtomicUsize::new(0)),
|
||||
captured: Arc::new(Mutex::new(Vec::new())),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl LlmClient for MockClient {
|
||||
fn clone_boxed(&self) -> Box<dyn LlmClient> {
|
||||
Box::new(self.clone())
|
||||
}
|
||||
|
||||
async fn stream(
|
||||
&self,
|
||||
request: Request,
|
||||
) -> Result<Pin<Box<dyn Stream<Item = Result<LlmEvent, ClientError>> + Send>>, ClientError>
|
||||
{
|
||||
self.captured.lock().unwrap().push(request);
|
||||
let idx = self.call_count.fetch_add(1, Ordering::SeqCst);
|
||||
let events = self.responses.get(idx).cloned().unwrap_or_default();
|
||||
Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok))))
|
||||
}
|
||||
}
|
||||
|
||||
struct MockFactory {
|
||||
client: MockClient,
|
||||
runtime_base: PathBuf,
|
||||
cwd: PathBuf,
|
||||
store_dir: PathBuf,
|
||||
worker_metadata_dir: PathBuf,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl RuntimeWorkerFactory for MockFactory {
|
||||
async fn spawn_controller(
|
||||
&self,
|
||||
_request: WorkerExecutionSpawnRequest,
|
||||
) -> Result<WorkerHandle, String> {
|
||||
let manifest = WorkerManifest::from_toml(
|
||||
r#"
|
||||
[worker]
|
||||
name = "runtime-adapter-test"
|
||||
pwd = "./"
|
||||
|
||||
[model]
|
||||
scheme = "anthropic"
|
||||
model_id = "test-model"
|
||||
|
||||
[engine]
|
||||
max_tokens = 100
|
||||
|
||||
[[scope.allow]]
|
||||
target = "./"
|
||||
permission = "write"
|
||||
"#,
|
||||
)
|
||||
.map_err(|err| err.to_string())?;
|
||||
let store = CombinedStore::new(
|
||||
FsStore::new(&self.store_dir).map_err(|err| err.to_string())?,
|
||||
FsWorkerStore::new(&self.worker_metadata_dir).map_err(|err| err.to_string())?,
|
||||
);
|
||||
let scope = Scope::writable(&self.cwd).map_err(|err| err.to_string())?;
|
||||
let worker = Worker::new(
|
||||
manifest,
|
||||
Engine::new(self.client.clone()),
|
||||
store,
|
||||
self.cwd.clone(),
|
||||
scope,
|
||||
)
|
||||
.await
|
||||
.map_err(|err| err.to_string())?;
|
||||
let (handle, _shutdown_rx) = WorkerController::spawn(worker, &self.runtime_base)
|
||||
.await
|
||||
.map_err(|err| err.to_string())?;
|
||||
Ok(handle)
|
||||
}
|
||||
}
|
||||
|
||||
fn simple_text_events() -> Vec<LlmEvent> {
|
||||
vec![
|
||||
LlmEvent::text_block_start(0),
|
||||
LlmEvent::text_delta(0, "hello"),
|
||||
LlmEvent::text_delta(0, " from worker"),
|
||||
LlmEvent::text_block_stop(0, None),
|
||||
LlmEvent::Status(StatusEvent {
|
||||
status: ResponseStatus::Completed,
|
||||
}),
|
||||
]
|
||||
}
|
||||
|
||||
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: ConfigBundleRef {
|
||||
id: bundle.metadata.id,
|
||||
digest: bundle.metadata.digest,
|
||||
},
|
||||
initial_input: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builtin_profile_selector_is_not_double_prefixed() {
|
||||
assert_eq!(
|
||||
ProfileRuntimeWorkerFactory::runtime_profile_value(
|
||||
&crate::catalog::ProfileSelector::Builtin("coder".to_string())
|
||||
)
|
||||
.as_deref(),
|
||||
Some("builtin:coder")
|
||||
);
|
||||
assert_eq!(
|
||||
ProfileRuntimeWorkerFactory::runtime_profile_value(
|
||||
&crate::catalog::ProfileSelector::Builtin("builtin:coder".to_string())
|
||||
)
|
||||
.as_deref(),
|
||||
Some("builtin:coder")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn adapter_dispatches_user_input_through_worker_run_lifecycle() {
|
||||
let client = MockClient::new(simple_text_events());
|
||||
let runtime_base = tempfile::tempdir().unwrap();
|
||||
let cwd = tempfile::tempdir().unwrap();
|
||||
let store = tempfile::tempdir().unwrap();
|
||||
let factory = MockFactory {
|
||||
client: client.clone(),
|
||||
runtime_base: runtime_base.path().to_path_buf(),
|
||||
cwd: cwd.path().to_path_buf(),
|
||||
store_dir: store.path().join("sessions"),
|
||||
worker_metadata_dir: store.path().join("workers"),
|
||||
};
|
||||
let backend = WorkerRuntimeExecutionBackend::new(factory).unwrap();
|
||||
let runtime = EmbeddedRuntime::with_execution_backend(
|
||||
RuntimeOptions {
|
||||
runtime_id: RuntimeId::new("embedded"),
|
||||
..RuntimeOptions::default()
|
||||
},
|
||||
Arc::new(backend),
|
||||
)
|
||||
.unwrap();
|
||||
runtime.store_config_bundle(test_bundle()).unwrap();
|
||||
let detail = runtime.create_worker(create_request("chat")).unwrap();
|
||||
|
||||
runtime
|
||||
.send_input(&detail.worker_ref, WorkerInput::user("say hello"))
|
||||
.unwrap();
|
||||
|
||||
let deadline = std::time::Instant::now() + Duration::from_secs(5);
|
||||
loop {
|
||||
let projection = runtime
|
||||
.transcript_projection(&detail.worker_ref, TranscriptQuery::new(0, 10))
|
||||
.unwrap();
|
||||
if projection.items.iter().any(|item| {
|
||||
item.role == TranscriptRole::Assistant && item.content == "hello from worker"
|
||||
}) {
|
||||
break;
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"timed out waiting for assistant transcript projection"
|
||||
);
|
||||
std::thread::sleep(Duration::from_millis(20));
|
||||
}
|
||||
|
||||
assert_eq!(client.captured.lock().unwrap().len(), 1);
|
||||
let observations = runtime
|
||||
.read_worker_observation_events(
|
||||
&detail.worker_ref,
|
||||
crate::observation::WorkerObservationCursor::zero(),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(
|
||||
observations
|
||||
.iter()
|
||||
.any(|event| matches!(event.payload, protocol::Event::TextDone { .. }))
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user