diff --git a/Cargo.lock b/Cargo.lock index 5f3a0dc1a4..3a30cfa0c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -43,6 +43,19 @@ dependencies = [ "subtle", ] +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if 1.0.4", + "getrandom 0.3.4", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -249,6 +262,45 @@ dependencies = [ "winnow 0.7.15", ] +[[package]] +name = "asn1-rs" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56624a96882bb8c26d61312ae18cb45868e5a9992ea73c58e45c3101e56a1e60" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom 7.1.3", + "num-traits", + "rusticata-macros", + "thiserror 2.0.18", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", + "synstructure", +] + +[[package]] +name = "asn1-rs-impl" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "async-broadcast" version = "0.7.2" @@ -1491,6 +1543,20 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "dashmap" +version = "6.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf" +dependencies = [ + "cfg-if 1.0.4", + "crossbeam-utils", + "hashbrown 0.14.5", + "lock_api", + "once_cell", + "parking_lot_core", +] + [[package]] name = "data-encoding" version = "2.10.0" @@ -1564,6 +1630,20 @@ dependencies = [ "zeroize", ] +[[package]] +name = "der-parser" +version = "10.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom 7.1.3", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "deranged" version = "0.5.8" @@ -2399,6 +2479,12 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" + [[package]] name = "hashbrown" version = "0.15.5" @@ -3079,6 +3165,7 @@ version = "1.0.0-rc.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b98e206e3d3f2642f5c08c413755fc0ac19b54ae1a656af88be03454ce3ed2e6" dependencies = [ + "axum", "backon", "blake3", "bytes", @@ -3178,12 +3265,20 @@ version = "1.0.0-rc.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d102597d0ee523f17fdb672c532395e634dbe945429284c811430d63bacc0d8a" dependencies = [ + "http-body-util", + "hyper", + "hyper-util", "iroh-metrics-derive", "itoa", "n0-error", "portable-atomic", + "reqwest 0.13.2", + "rustls", + "rustls-platform-verifier 0.7.0", "ryu", "serde", + "tokio", + "tokio-util", "tracing", ] @@ -3205,9 +3300,12 @@ version = "1.0.0-rc.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "54f490405e42dd2ecf16be18a3587d2665401e94a498094f12322eaa6d5ebb2b" dependencies = [ + "ahash", "blake3", "bytes", "cfg_aliases", + "clap", + "dashmap", "data-encoding", "derive_more", "getrandom 0.4.2", @@ -3228,17 +3326,28 @@ dependencies = [ "pin-project", "postcard", "rand 0.10.1", + "rcgen", + "reloadable-state", "reqwest 0.13.2", "rustls", + "rustls-cert-file-reader", + "rustls-cert-reloadable-resolver", "rustls-pki-types", "serde", "serde_bytes", + "serde_json", + "sha1 0.11.0", + "simdutf8", "strum 0.28.0", + "time", "tokio", "tokio-rustls", + "tokio-rustls-acme", "tokio-util", "tokio-websockets", + "toml 1.1.2+spec-1.1.0", "tracing", + "tracing-subscriber", "url", "vergen-gitcl", "webpki-roots 1.0.6", @@ -3673,8 +3782,14 @@ name = "mesh-llm-api-server" version = "0.66.0" dependencies = [ "anyhow", + "base64", + "futures-util", + "iroh", + "iroh-relay", "mesh-llm-api-client", + "mesh-llm-host-runtime", "mesh-llm-node", + "serde_json", "tokio", ] @@ -3760,14 +3875,15 @@ dependencies = [ "crypto_box", "dirs", "ed25519-dalek", + "futures-util", "hex", "hf-hub", "httparse", "iroh", + "iroh-relay", "keyring", "libc", "mdns-sd", - "mesh-llm-api-server", "mesh-llm-client", "mesh-llm-guardrails", "mesh-llm-identity", @@ -3812,7 +3928,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tokio-stream", - "toml", + "toml 0.9.12+spec-1.1.0", "tracing", "tracing-subscriber", "url", @@ -4475,6 +4591,7 @@ dependencies = [ "rustc-hash", "rustls", "rustls-pki-types", + "rustls-platform-verifier 0.7.0", "slab", "sorted-index-buffer", "thiserror 2.0.18", @@ -4848,6 +4965,15 @@ dependencies = [ "objc2-security", ] +[[package]] +name = "oid-registry" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -5135,6 +5261,16 @@ dependencies = [ "hmac", ] +[[package]] +name = "pem" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" +dependencies = [ + "base64", + "serde_core", +] + [[package]] name = "pem-rfc7468" version = "1.0.0" @@ -5944,6 +6080,20 @@ dependencies = [ "unicode-width", ] +[[package]] +name = "rcgen" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10b99e0098aa4082912d4c649628623db6aba77335e4f4569ff5083a6448b32e" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "x509-parser", + "yasna", +] + [[package]] name = "redb" version = "3.1.3" @@ -6028,6 +6178,23 @@ version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" +[[package]] +name = "reloadable-core" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1dc20ac1418988b60072d783c9f68e28a173fb63493c127952f6face3b40c6e0" + +[[package]] +name = "reloadable-state" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3853ef78d45b50f8b989896304a85239539d39b7f866a000e8846b9b72d74ce8" +dependencies = [ + "arc-swap", + "reloadable-core", + "tokio", +] + [[package]] name = "reqwest" version = "0.12.28" @@ -6103,7 +6270,7 @@ dependencies = [ "quinn", "rustls", "rustls-pki-types", - "rustls-platform-verifier", + "rustls-platform-verifier 0.6.2", "serde", "serde_json", "serde_urlencoded", @@ -6240,6 +6407,15 @@ dependencies = [ "semver", ] +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom 7.1.3", +] + [[package]] name = "rustix" version = "0.38.44" @@ -6282,6 +6458,40 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-cert-file-reader" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8bb47c2a50fdfdaf95b0ac8b12620fc327da1fd4adbb30d0c56d866b005873ff" +dependencies = [ + "rustls-cert-read", + "rustls-pki-types", + "thiserror 2.0.18", + "tokio", +] + +[[package]] +name = "rustls-cert-read" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd46e8c5ae4de3345c4786a83f99ec7aff287209b9e26fa883c473aeb28f19d5" +dependencies = [ + "rustls-pki-types", +] + +[[package]] +name = "rustls-cert-reloadable-resolver" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe1baa8a3a1f05eaa9fc55aed4342867f70e5c170ea3bfed1b38c51a4857c0c8" +dependencies = [ + "futures-util", + "reloadable-state", + "rustls", + "rustls-cert-read", + "thiserror 2.0.18", +] + [[package]] name = "rustls-native-certs" version = "0.8.3" @@ -6325,6 +6535,27 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustls-platform-verifier" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0" +dependencies = [ + "core-foundation 0.10.1", + "core-foundation-sys", + "jni 0.22.4", + "log", + "once_cell", + "rustls", + "rustls-native-certs", + "rustls-platform-verifier-android", + "rustls-webpki", + "security-framework 3.7.0", + "security-framework-sys", + "webpki-root-certs", + "windows-sys 0.52.0", +] + [[package]] name = "rustls-platform-verifier-android" version = "0.1.1" @@ -6749,6 +6980,17 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if 1.0.4", + "cpufeatures 0.3.0", + "digest 0.11.2", +] + [[package]] name = "sha1_smol" version = "1.0.1" @@ -7566,6 +7808,34 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-rustls-acme" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1af8573b15fdad8d66da116198cd8fd8d87ff62a67c1c6c3df7f62da1170793f" +dependencies = [ + "async-trait", + "base64", + "chrono", + "futures", + "log", + "num-bigint", + "pem", + "proc-macro2", + "rcgen", + "reqwest 0.13.2", + "ring", + "rustls", + "serde", + "serde_json", + "thiserror 2.0.18", + "time", + "tokio", + "tokio-rustls", + "webpki-roots 1.0.6", + "x509-parser", +] + [[package]] name = "tokio-socks" version = "0.5.2" @@ -7658,6 +7928,21 @@ dependencies = [ "winnow 0.7.15", ] +[[package]] +name = "toml" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81f3d15e84cbcd896376e6730314d59fb5a87f31e4b038454184435cd57defee" +dependencies = [ + "indexmap", + "serde_core", + "serde_spanned", + "toml_datetime 1.1.1+spec-1.1.0", + "toml_parser", + "toml_writer", + "winnow 1.0.1", +] + [[package]] name = "toml_datetime" version = "0.7.5+spec-1.1.0" @@ -7899,7 +8184,7 @@ dependencies = [ "rand 0.9.2", "rustls", "rustls-pki-types", - "sha1", + "sha1 0.10.6", "thiserror 2.0.18", "utf-8", ] @@ -8029,7 +8314,7 @@ dependencies = [ "serde", "tempfile", "textwrap", - "toml", + "toml 0.9.12+spec-1.1.0", "uniffi_internal_macros", "uniffi_meta", "uniffi_pipeline", @@ -8085,7 +8370,7 @@ dependencies = [ "quote", "serde", "syn 2.0.117", - "toml", + "toml 0.9.12+spec-1.1.0", "uniffi_meta", ] @@ -9130,6 +9415,24 @@ version = "0.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ea6fc2961e4ef194dcbfe56bb845534d0dc8098940c7e5c012a258bfec6701bd" +[[package]] +name = "x509-parser" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom 7.1.3", + "oid-registry", + "ring", + "rusticata-macros", + "thiserror 2.0.18", + "time", +] + [[package]] name = "xdg-home" version = "1.3.0" @@ -9310,6 +9613,15 @@ dependencies = [ "serde_json", ] +[[package]] +name = "yasna" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd" +dependencies = [ + "time", +] + [[package]] name = "yoke" version = "0.8.2" @@ -9354,7 +9666,7 @@ dependencies = [ "rand 0.8.5", "serde", "serde_repr", - "sha1", + "sha1 0.10.6", "static_assertions", "tracing", "uds_windows", diff --git a/crates/mesh-llm-api-server/Cargo.toml b/crates/mesh-llm-api-server/Cargo.toml index 80a2659e9e..18c94b6dcd 100644 --- a/crates/mesh-llm-api-server/Cargo.toml +++ b/crates/mesh-llm-api-server/Cargo.toml @@ -13,11 +13,30 @@ categories = ["api-bindings", "network-programming"] [features] host-io = ["mesh-llm-api-client/host-io"] +# Run a real iroh-backed mesh node in-process (gossip, relay registration +# including --relay-auth, invite tokens, QUIC peer connections), instead of +# the default HTTP-shim client behaviour. Drags `mesh-llm-host-runtime` and +# its transitive deps (skippy, llama.cpp link path, etc.) into the build, +# so it is off by default. Consumers who want a Rust app to act as a real +# mesh peer should enable it. +host-runtime = ["dep:mesh-llm-host-runtime"] + [dependencies] anyhow.workspace = true mesh-llm-api-client = { path = "../mesh-llm-api-client", version = "0.66.0" } mesh-llm-node = { path = "../mesh-llm-node", version = "0.66.0" } +# Optional, gated by the `host-runtime` feature. +mesh-llm-host-runtime = { path = "../mesh-llm-host-runtime", optional = true, default-features = false } tokio = { version = "1", features = ["sync"] } [dev-dependencies] -tokio = { version = "1", features = ["macros", "rt"] } +tokio = { version = "1", features = ["macros", "rt", "rt-multi-thread", "time"] } +# In-process iroh-relay for testing the host-runtime feature: real +# AccessConfig::Restricted relay, real endpoint bind, real WebSocket +# upgrade. Same dev-dep iroh feature combo `mesh-llm-host-runtime` +# already uses for its own gated-relay tests. +iroh = { version = "1.0.0-rc.0", features = ["test-utils"] } +iroh-relay = { version = "1.0.0-rc.0", features = ["server", "test-utils"] } +futures-util = "0.3" +base64 = "0.22" +serde_json = "1" diff --git a/crates/mesh-llm-api-server/README.md b/crates/mesh-llm-api-server/README.md index 054c4db548..0655936d83 100644 --- a/crates/mesh-llm-api-server/README.md +++ b/crates/mesh-llm-api-server/README.md @@ -33,3 +33,51 @@ high-level serving errors. If an API is meant for client-only app integration, it belongs in `mesh-llm-api-client`. If it requires model management or local serving, it belongs in `mesh-llm-api-server`. + +## Running the full mesh-llm runtime in-process (`host-runtime` feature) + +For applications that want to run **exactly what `mesh-llm serve` / +`mesh-llm client` does** — not just consume mesh inference, but be the +running node — enable the `host-runtime` feature: + +```toml +mesh-llm-api-server = { version = "0.66.0", features = ["host-runtime"] } +``` + +Then call `run_serve(MeshServeSpec { ... })`: + +```rust +use mesh_llm_api_server::{run_serve, MeshServeSpec}; +use std::collections::HashMap; + +let mut relay_auths = HashMap::new(); +relay_auths.insert( + "https://gated.example/".to_string(), + "".to_string(), +); + +run_serve(MeshServeSpec { + client: true, + auto: true, + relays: vec!["https://gated.example/".into()], + relay_auths, + port: Some(9337), + console_port: Some(3131), + headless: true, + max_vram_gb: Some(0.0), + ..Default::default() +}) +.await?; +``` + +This drives the same `runtime::run_with_args` entry point the binary +uses. You get auto-discovery, election, tunnel manager, OpenAI HTTP +proxy on `--port`, management console on `--console`, local model +serving (when configured), plugin host — the entire mesh-llm runtime +inside your process. + +`MeshNode::builder()` (`host-runtime` feature also required for the +fine-grained options like `.relay(...)` and `.relay_auth(...)`) is the +composable alternative for apps that want to wire pieces themselves +rather than running the whole orchestration. See `docs/SDK.md` for the +full comparison. diff --git a/crates/mesh-llm-api-server/src/lib.rs b/crates/mesh-llm-api-server/src/lib.rs index cda685d3a3..a4db647718 100644 --- a/crates/mesh-llm-api-server/src/lib.rs +++ b/crates/mesh-llm-api-server/src/lib.rs @@ -13,11 +13,59 @@ pub use mesh_llm_api_client::{ MAX_RECONNECT_ATTEMPTS, }; pub use mesh_llm_node::serving::ServingController; + +/// Run the full mesh-llm runtime in-process — the same code path the +/// `mesh-llm` binary runs. Only available with the `host-runtime` feature. +/// +/// This is the SDK entry point for embedders who want their Rust app to +/// act exactly like running `mesh-llm serve` or `mesh-llm client` — +/// with auto-discovery, election, tunnel manager, OpenAI HTTP proxy, +/// management console, and local model serving (when configured) — +/// without spawning the binary as a subprocess. +/// +/// # Example +/// +/// ```no_run +/// # use std::collections::HashMap; +/// use mesh_llm_api_server::{run_serve, MeshServeSpec}; +/// +/// # async fn run() -> anyhow::Result<()> { +/// let mut relay_auths = HashMap::new(); +/// relay_auths.insert( +/// "https://gated.example/".to_string(), +/// "".to_string(), +/// ); +/// +/// run_serve(MeshServeSpec { +/// // Same flags `mesh-llm serve` / `mesh-llm client` accept. +/// client: true, // false (default) = serve role +/// auto: true, // == --auto +/// relays: vec!["https://gated.example/".into()], +/// relay_auths, // == --relay-auth URL=TOKEN +/// port: Some(9337), // OpenAI HTTP proxy port +/// console_port: Some(3131), // management API / web console +/// headless: true, // skip embedded web UI +/// max_vram_gb: Some(0.0), // client-only, no VRAM advert +/// ..MeshServeSpec::default() +/// }) +/// .await?; +/// # Ok(()) +/// # } +/// ``` +/// +/// The future blocks until the runtime exits. The runtime is not +/// currently `Send`-clean; if you need concurrent work, run on a +/// `tokio::task::LocalSet` rather than `tokio::spawn`. +/// +/// For finer-grained control — composing pieces without running the +/// whole orchestration — see [`MeshNodeBuilder`] instead. +#[cfg(feature = "host-runtime")] +pub use mesh_llm_host_runtime::host_node::{run_serve, MeshServeSpec}; pub use node::{ CapabilityLevel, CleanupPolicy, CleanupResult, DeleteModelOptions, DeleteModelResult, DevicePolicy, DownloadId, DownloadOptions, DownloadedModel, InstalledModel, LoadModelOptions, - MeshEvents, MeshInference, MeshModels, MeshNode, MeshNodeBuilder, MeshNodeConfig, MeshServing, - MeshStatusApi, ModelCacheStatus, ModelCapabilities, ModelDetails, ModelKind, ModelSearchQuery, - ModelSource, ModelSummary, PrunePolicy, PruneResult, ServedModel, ServingModelState, - ServingStatus, UnloadModelOptions, UnloadTarget, + MeshEvents, MeshInference, MeshModels, MeshNode, MeshNodeBuilder, MeshNodeConfig, MeshQuicBind, + MeshRole, MeshServing, MeshStatusApi, ModelCacheStatus, ModelCapabilities, ModelDetails, + ModelKind, ModelSearchQuery, ModelSource, ModelSummary, PrunePolicy, PruneResult, ServedModel, + ServingModelState, ServingStatus, UnloadModelOptions, UnloadTarget, }; diff --git a/crates/mesh-llm-api-server/src/node.rs b/crates/mesh-llm-api-server/src/node.rs index 5d906a6cb9..c592b26288 100644 --- a/crates/mesh-llm-api-server/src/node.rs +++ b/crates/mesh-llm-api-server/src/node.rs @@ -5,11 +5,43 @@ use crate::{ }; pub use mesh_llm_node::models::{CapabilityLevel, ModelCapabilities, ModelKind, ModelSource}; use mesh_llm_node::serving::ServingController; +use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use tokio::sync::Mutex; +#[cfg(feature = "host-runtime")] +use mesh_llm_host_runtime::host_node::{ + self, HostNode, HostNodeSpec, MeshNodeRole as HostNodeRole, + MeshQuicBindSelection as HostQuicBindSelection, +}; + +/// Mesh role for the SDK — mirrors `mesh-llm`'s `--client` flag. +/// +/// Default is `Serve` (the binary's default surface). Pick `Client` for +/// a no-GPU, no-model client-only node. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum MeshRole { + /// Serve a model (or join `--auto` as a serve candidate). + #[default] + Serve, + /// Client only — no GPU, no model. + Client, +} + +/// QUIC bind selection for the in-process mesh node. +/// +/// Defaults to ephemeral OS-chosen port on all interfaces, matching the +/// CLI's default behaviour. +#[derive(Debug, Clone, Copy, Default)] +pub struct MeshQuicBind { + /// Optional bind IP. + pub ip: Option, + /// Optional fixed UDP port (e.g. for NAT port forwarding). + pub port: Option, +} + #[derive(Clone, Debug, Default)] pub enum DevicePolicy { #[default] @@ -190,6 +222,18 @@ pub struct MeshNodeBuilder { serving_enabled: bool, device_policy: DevicePolicy, serving_controller: Option>, + // Real in-process mesh-node knobs (used under the `host-runtime` + // feature). Kept here even without the feature so the builder API + // is stable across feature combinations; without `host-runtime` they + // are stored but ignored. + role: MeshRole, + relays: Vec, + relay_auths: HashMap, + quic_bind: MeshQuicBind, + max_vram_gb: Option, + no_enumerate_host: bool, + openai_port: Option, + openai_listen_all: bool, } impl MeshNodeBuilder { @@ -239,6 +283,70 @@ impl MeshNodeBuilder { self } + /// Set the mesh role. Equivalent to the binary's `--client` flag + /// (passing [`MeshRole::Client`]) or the default `serve` surface. + pub fn role(mut self, role: MeshRole) -> Self { + self.role = role; + self + } + + /// Add an iroh relay URL. Equivalent to `mesh-llm … --relay `, + /// callable multiple times. Without any call, the bundled default + /// relays are used. + pub fn relay(mut self, url: impl Into) -> Self { + self.relays.push(url.into()); + self + } + + /// Per-relay bearer token for gated iroh relays. Equivalent to + /// `mesh-llm … --relay-auth URL=TOKEN`. The token is sent as + /// `Authorization: Bearer ` on the WebSocket upgrade to the + /// matching relay URL. Callable multiple times; only relays present + /// in the map use auth. + pub fn relay_auth(mut self, relay_url: impl Into, token: impl Into) -> Self { + self.relay_auths.insert(relay_url.into(), token.into()); + self + } + + /// QUIC bind selection (IP / fixed port). + pub fn quic_bind(mut self, bind: MeshQuicBind) -> Self { + self.quic_bind = bind; + self + } + + /// VRAM cap in GB used for planning and mesh advertisement. + /// `Some(0.0)` for client-only nodes. + pub fn max_vram_gb(mut self, gb: f64) -> Self { + self.max_vram_gb = Some(gb); + self + } + + /// Disable broadcasting GPU name, hostname, VRAM, and reserved + /// bytes to peers. Equivalent to `mesh-llm … --no-enumerate-host`. + pub fn no_enumerate_host(mut self, no_enumerate: bool) -> Self { + self.no_enumerate_host = no_enumerate; + self + } + + /// Bind an OpenAI-compatible HTTP proxy on this port when the node + /// starts. Equivalent to `mesh-llm … --port `. Use `0` for an + /// OS-assigned ephemeral port; read it back from + /// [`MeshNode::openai_base_url`] after `start()`. + /// + /// Only honoured under the `host-runtime` feature. Without the + /// feature this setter is a no-op. + pub fn openai_port(mut self, port: u16) -> Self { + self.openai_port = Some(port); + self + } + + /// Bind the OpenAI proxy on `0.0.0.0` instead of `127.0.0.1`. + /// Equivalent to `mesh-llm … --listen-all`. Default `false`. + pub fn openai_listen_all(mut self, listen_all: bool) -> Self { + self.openai_listen_all = listen_all; + self + } + pub fn build(self) -> Result { let owner_keypair = self.owner_keypair.ok_or(MeshApiError::InvalidConfig { message: "MeshNode identity is required", @@ -261,11 +369,27 @@ impl MeshNodeBuilder { device_policy: self.device_policy, }; + let host_node_spec = HostNodeSpecHolder { + role: self.role, + relays: self.relays, + relay_auths: self.relay_auths, + quic_bind: self.quic_bind, + max_vram_gb: self.max_vram_gb, + enumerate_host: !self.no_enumerate_host, + openai_port: self.openai_port, + openai_listen_all: self.openai_listen_all, + }; + Ok(MeshNode { inner: Arc::new(MeshNodeInner { client: Mutex::new(client), config, serving_controller: self.serving_controller, + host_node_spec, + #[cfg(feature = "host-runtime")] + host_node: Mutex::new(None), + #[cfg(feature = "host-runtime")] + openai_proxy: Mutex::new(None), }), }) } @@ -283,14 +407,42 @@ impl Default for MeshNodeBuilder { serving_enabled: false, device_policy: DevicePolicy::Auto, serving_controller: None, + role: MeshRole::default(), + relays: Vec::new(), + relay_auths: HashMap::new(), + quic_bind: MeshQuicBind::default(), + max_vram_gb: None, + no_enumerate_host: false, + openai_port: None, + openai_listen_all: false, } } } +/// Captured-from-builder spec used by `start()` under the +/// `host-runtime` feature. Stored unconditionally so the type layout +/// doesn't shift across feature combinations. +#[cfg_attr(not(feature = "host-runtime"), allow(dead_code))] +struct HostNodeSpecHolder { + role: MeshRole, + relays: Vec, + relay_auths: HashMap, + quic_bind: MeshQuicBind, + max_vram_gb: Option, + enumerate_host: bool, + openai_port: Option, + openai_listen_all: bool, +} + struct MeshNodeInner { client: Mutex, config: MeshNodeConfig, serving_controller: Option>, + host_node_spec: HostNodeSpecHolder, + #[cfg(feature = "host-runtime")] + host_node: Mutex>, + #[cfg(feature = "host-runtime")] + openai_proxy: Mutex>, } #[derive(Clone)] @@ -304,10 +456,84 @@ impl MeshNode { } pub async fn start(&self) -> Result<(), MeshApiError> { - self.inner.client.lock().await.join().await + #[cfg(feature = "host-runtime")] + { + // Idempotency guard: if the host node is already running, + // calling start() again would otherwise spawn a second iroh + // endpoint (and possibly a second OpenAI proxy bind on the + // same port, which would just fail) and orphan the first — + // stop() only tracks the most recent one. Cheap to make + // this a no-op. + if self.inner.host_node.lock().await.is_some() { + return Ok(()); + } + + let spec = HostNodeSpec { + role: match self.inner.host_node_spec.role { + MeshRole::Client => HostNodeRole::Client, + MeshRole::Serve => HostNodeRole::default(), + }, + relays: self.inner.host_node_spec.relays.clone(), + relay_auths: self.inner.host_node_spec.relay_auths.clone(), + quic_bind: HostQuicBindSelection { + ip: self.inner.host_node_spec.quic_bind.ip, + port: self.inner.host_node_spec.quic_bind.port, + }, + max_vram_gb: self.inner.host_node_spec.max_vram_gb, + enumerate_host: self.inner.host_node_spec.enumerate_host, + }; + let node = + host_node::start_host_node(spec) + .await + .map_err(|err| MeshApiError::Serving { + message: format!("host node start failed: {err}"), + })?; + if let Err(err) = node.join(self.inner.config.invite_token.as_str()).await { + node.shutdown().await; + return Err(MeshApiError::Serving { + message: format!("host node join failed: {err}"), + }); + } + node.start_accepting(); + + // Spin up the OpenAI HTTP proxy if the builder asked for one. + // Equivalent to `mesh-llm … --port `. Routes inference + // requests to mesh peers serving the requested model. + if let Some(port) = self.inner.host_node_spec.openai_port { + let listen_all = self.inner.host_node_spec.openai_listen_all; + let handle = + mesh_llm_host_runtime::host_node::start_openai_proxy(&node, port, listen_all) + .await + .map_err(|err| MeshApiError::Serving { + message: format!("openai proxy bind failed: {err}"), + })?; + *self.inner.openai_proxy.lock().await = Some(handle); + } + + *self.inner.host_node.lock().await = Some(node); + // Also flip the legacy HTTP-shim client's connected flag so + // status()/events() callers see a connected node. Harmless. + self.inner.client.lock().await.join().await + } + #[cfg(not(feature = "host-runtime"))] + { + // Touch the spec to silence dead-code on builds without the + // feature, and avoid surprising no-op stores. + let _ = &self.inner.host_node_spec; + self.inner.client.lock().await.join().await + } } pub async fn stop(&self) -> Result<(), MeshApiError> { + #[cfg(feature = "host-runtime")] + { + if let Some(proxy) = self.inner.openai_proxy.lock().await.take() { + proxy.shutdown(); + } + if let Some(node) = self.inner.host_node.lock().await.take() { + node.shutdown().await; + } + } self.inner.client.lock().await.disconnect().await; Ok(()) } @@ -316,6 +542,50 @@ impl MeshNode { self.inner.client.lock().await.reconnect().await } + /// Invite token other peers can use to join this node. + /// + /// Only meaningful when running under the `host-runtime` feature with + /// `start()` having been called. Without the feature, or before + /// `start()`, returns `None`. + #[cfg(feature = "host-runtime")] + pub async fn invite_token(&self) -> Option { + self.inner + .host_node + .lock() + .await + .as_ref() + .map(|n| n.invite_token()) + } + + /// Base URL of the in-process OpenAI HTTP proxy started via + /// [`MeshNodeBuilder::openai_port`]. + /// + /// Returns `None` if the builder didn't request a proxy or + /// `start()` has not been called. The returned URL is the value SDK + /// consumers feed to any OpenAI-compatible client library to route + /// inference requests through the mesh. + /// + /// Only meaningful under the `host-runtime` feature. + #[cfg(feature = "host-runtime")] + pub async fn openai_base_url(&self) -> Option { + self.inner + .openai_proxy + .lock() + .await + .as_ref() + .map(|p| p.base_url()) + } + + /// Set the display name advertised to peers. + /// + /// Only takes effect under the `host-runtime` feature after `start()`. + #[cfg(feature = "host-runtime")] + pub async fn set_display_name(&self, name: String) { + if let Some(node) = self.inner.host_node.lock().await.as_ref() { + node.set_display_name(name).await; + } + } + pub fn inference(&self) -> MeshInference { MeshInference { inner: self.inner.clone(), diff --git a/crates/mesh-llm-api-server/tests/host_node_gated_relay.rs b/crates/mesh-llm-api-server/tests/host_node_gated_relay.rs new file mode 100644 index 0000000000..e38415e015 --- /dev/null +++ b/crates/mesh-llm-api-server/tests/host_node_gated_relay.rs @@ -0,0 +1,209 @@ +//! End-to-end test: SDK consumer drives a real iroh-backed mesh node +//! through the published `MeshNodeBuilder` API, registers with a gated +//! iroh relay using `--relay-auth`-equivalent setters, and reaches +//! `online()`. +//! +//! Gated on the `host-runtime` feature — only meaningful when the SDK +//! actually starts a mesh node, not the HTTP-shim default. + +#![cfg(feature = "host-runtime")] + +use futures_util::StreamExt; +use iroh::endpoint::{presets, Endpoint, RelayMode}; +use iroh::test_utils::run_relay_server_with_access; +use iroh::{RelayConfig as IrohRelayConfig, RelayMap, SecretKey, Watcher}; +use iroh_relay::server::{Access, AccessConfig}; +use iroh_relay::tls::CaRootsConfig; +use mesh_llm_api_server::{InviteToken, MeshNode, MeshQuicBind, MeshRole, OwnerKeypair}; +use mesh_llm_host_runtime::host_node::{start_host_node, HostNodeSpec, MeshNodeRole}; +use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket}; +use std::time::Duration; + +/// Spawn an in-process iroh-relay that only admits `expected_token`. +async fn spawn_gated_relay(expected_token: &'static str) -> (String, iroh_relay::server::Server) { + let access = AccessConfig::Restricted(Box::new(move |request| { + Box::pin(async move { + if request.auth_token().as_deref() == Some(expected_token) { + Access::Allow + } else { + Access::Deny + } + }) + })); + let (_relay_map, relay_url, server) = run_relay_server_with_access(false, access) + .await + .expect("spawn gated relay"); + (relay_url.to_string(), server) +} + +/// Start an anchor `HostNode` pointed at the given relay so the test +/// stays offline / deterministic. Without an explicit relay, the anchor +/// would fall back to the bundled public relays and depend on the +/// internet being reachable. Optionally takes an auth token if the +/// target relay is gated (`AccessConfig::Restricted`). +async fn anchor_on_relay( + relay_url: &str, + relay_token: Option<&str>, +) -> (String, mesh_llm_host_runtime::host_node::HostNode) { + let mut relay_auths = std::collections::HashMap::new(); + if let Some(token) = relay_token { + relay_auths.insert(relay_url.to_string(), token.to_string()); + } + let anchor = start_host_node(HostNodeSpec { + role: MeshNodeRole::Client, + relays: vec![relay_url.to_string()], + relay_auths, + max_vram_gb: Some(0.0), + enumerate_host: false, + ..HostNodeSpec::default() + }) + .await + .expect("anchor host node should start"); + anchor.start_accepting(); + (anchor.invite_token(), anchor) +} + +fn free_local_udp_port() -> u16 { + let socket = + UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))).expect("allocate local UDP port"); + socket.local_addr().expect("read local UDP port").port() +} + +async fn probe_quic_port_released(port: u16) { + let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port); + let mut last_error = String::new(); + + for _ in 0..20 { + match Endpoint::builder(presets::Minimal) + .secret_key(SecretKey::generate()) + .relay_mode(RelayMode::Disabled) + .bind_addr(bind_addr) + .expect("configure probe bind address") + .bind() + .await + { + Ok(endpoint) => { + endpoint.close().await; + return; + } + Err(err) => { + last_error = format!("{err:#}"); + tokio::time::sleep(Duration::from_millis(25)).await; + } + } + } + + panic!("host-node cleanup should release UDP port {port}: {last_error}"); +} + +#[tokio::test] +async fn mesh_node_builder_threads_relay_auth_to_real_iroh_endpoint() { + const TOKEN: &str = "secret-bearer-token"; + let (gated_url, _server) = spawn_gated_relay(TOKEN).await; + let (invite, anchor) = anchor_on_relay(&gated_url, Some(TOKEN)).await; + let invite_token: InviteToken = invite.parse().expect("parse invite"); + + let node = MeshNode::builder() + .identity(OwnerKeypair::generate()) + .join(invite_token) + .role(MeshRole::Client) + .relay(&gated_url) + .relay_auth(&gated_url, TOKEN) + .max_vram_gb(0.0) + .build() + .expect("builder"); + + // start() should bring the underlying iroh endpoint online via the + // gated relay (admitted because we passed the matching token). We + // don't reach Ok unless the host-runtime path is actually used. + tokio::time::timeout(Duration::from_secs(10), node.start()) + .await + .expect("MeshNode.start() should resolve within 10s") + .expect("MeshNode.start() should succeed when relay_auth matches"); + + // Invite token from the running mesh node is non-empty. + let invite = node + .invite_token() + .await + .expect("invite_token should be populated after start"); + assert!(!invite.is_empty(), "invite token must not be empty"); + + node.stop().await.expect("stop"); + anchor.shutdown().await; +} + +#[tokio::test] +async fn mesh_node_start_fails_when_host_runtime_join_token_is_invalid() { + let quic_port = free_local_udp_port(); + let invite_token: InviteToken = "mesh-test:not-base64".parse().expect("parse invite"); + + let node = MeshNode::builder() + .identity(OwnerKeypair::generate()) + .join(invite_token) + .role(MeshRole::Client) + .quic_bind(MeshQuicBind { + ip: Some(IpAddr::V4(Ipv4Addr::LOCALHOST)), + port: Some(quic_port), + }) + .max_vram_gb(0.0) + .build() + .expect("builder"); + + let start = tokio::time::timeout(Duration::from_secs(10), node.start()) + .await + .expect("MeshNode.start() should resolve within 10s"); + + assert!( + start.is_err(), + "host-runtime start must attempt the configured join token" + ); + + assert!( + node.invite_token().await.is_none(), + "failed host-runtime starts must not store a node" + ); + probe_quic_port_released(quic_port).await; +} + +#[tokio::test] +async fn wrong_relay_token_is_rejected_by_gated_relay() { + // We can't observe relay-level denial through the SDK's surface + // yet, so verify the underlying iroh wire path directly: build an + // endpoint with the same relay map shape the SDK would build, + // but with the WRONG token, and assert iroh surfaces + // `not authorized` via `home_relay_status`. + // + // This is the property defended on the runtime side; it's what + // keeps the SDK's relay_auth setter honest end-to-end. + const TOKEN: &str = "secret-bearer-token"; + let (gated_url, _server) = spawn_gated_relay(TOKEN).await; + + let parsed: iroh::RelayUrl = gated_url.parse().expect("parse url"); + let cfg = IrohRelayConfig::new(parsed, None).with_auth_token("wrong-token"); + let map: RelayMap = RelayMap::from_iter([cfg]); + + let ep = Endpoint::builder(presets::Minimal) + .secret_key(SecretKey::generate()) + .relay_mode(RelayMode::Custom(map)) + .ca_roots_config(CaRootsConfig::insecure_skip_verify()) + .bind() + .await + .expect("endpoint bind"); + + let mut stream = ep.home_relay_status().stream(); + let auth_err = tokio::time::timeout(Duration::from_secs(5), async { + while let Some(status) = stream.next().await { + if let Some(err) = status.iter().filter_map(|s| s.last_error()).next() { + return Some(format!("{err:#}")); + } + } + None + }) + .await + .expect("home relay status within 5s") + .expect("home relay status should yield an error"); + assert!( + auth_err.contains("not authorized"), + "wrong token must be denied by gated relay, got: {auth_err}" + ); +} diff --git a/crates/mesh-llm-api-server/tests/openai_proxy.rs b/crates/mesh-llm-api-server/tests/openai_proxy.rs new file mode 100644 index 0000000000..53b2e77faa --- /dev/null +++ b/crates/mesh-llm-api-server/tests/openai_proxy.rs @@ -0,0 +1,175 @@ +//! End-to-end test: SDK consumer asks `MeshNodeBuilder` to spin up an +//! OpenAI HTTP proxy alongside the in-process mesh node, and we hit it +//! over real TCP/HTTP. +//! +//! Gated on the `host-runtime` feature. + +#![cfg(feature = "host-runtime")] + +use iroh::test_utils::run_relay_server_with_access; +use iroh_relay::server::AccessConfig; +use mesh_llm_api_server::{InviteToken, MeshNode, MeshRole, OwnerKeypair}; +use mesh_llm_host_runtime::host_node::{start_host_node, HostNode, HostNodeSpec, MeshNodeRole}; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; + +/// Spin up an in-process open relay so the test runs offline and +/// deterministic. Returns the relay URL string and the server (held by +/// the caller so it isn't dropped mid-test). +async fn spawn_open_relay() -> (String, iroh_relay::server::Server) { + let (_relay_map, relay_url, server) = + run_relay_server_with_access(false, AccessConfig::Everyone) + .await + .expect("spawn open relay"); + (relay_url.to_string(), server) +} + +/// Start an anchor `HostNode` that the SDK node can join. Pointed at +/// the in-process relay so we don't reach the bundled defaults. +async fn anchor_on_relay(relay_url: &str) -> (String, HostNode) { + let anchor = start_host_node(HostNodeSpec { + role: MeshNodeRole::Client, + relays: vec![relay_url.to_string()], + max_vram_gb: Some(0.0), + enumerate_host: false, + ..HostNodeSpec::default() + }) + .await + .expect("anchor host node should start"); + anchor.start_accepting(); + (anchor.invite_token(), anchor) +} + +/// Minimal HTTP GET against `host:port` returning the response status line. +/// Avoids pulling reqwest into dev-deps just for this smoke test. +async fn http_get(host_port: &str, path: &str) -> String { + let mut stream = TcpStream::connect(host_port) + .await + .expect("connect to proxy"); + let request = format!( + "GET {path} HTTP/1.1\r\nHost: {host_port}\r\nConnection: close\r\nAccept: application/json\r\n\r\n" + ); + stream + .write_all(request.as_bytes()) + .await + .expect("write request"); + let mut buf = Vec::with_capacity(1024); + stream.read_to_end(&mut buf).await.expect("read response"); + String::from_utf8_lossy(&buf).to_string() +} + +#[tokio::test] +async fn openai_proxy_binds_and_serves_v1_models_over_http() { + let (relay_url, _relay_server) = spawn_open_relay().await; + let (invite, anchor) = anchor_on_relay(&relay_url).await; + let invite_token: InviteToken = invite.parse().expect("parse invite"); + + let node = MeshNode::builder() + .identity(OwnerKeypair::generate()) + .join(invite_token) + .role(MeshRole::Client) + .relay(&relay_url) // keep the SDK node on our in-process relay too + .max_vram_gb(0.0) + // Port 0 → OS-assigned ephemeral. The handle reports the real one. + .openai_port(0) + .build() + .expect("builder"); + + tokio::time::timeout(Duration::from_secs(60), node.start()) + .await + .expect("MeshNode.start() should resolve within 60s") + .expect("MeshNode.start() should succeed"); + + let base = node + .openai_base_url() + .await + .expect("openai_base_url should be populated after start with openai_port"); + assert!(base.starts_with("http://127.0.0.1:"), "base url: {base}"); + + // host:port for our raw TCP probe. + let host_port = base + .strip_prefix("http://") + .expect("base url has http:// prefix"); + + // Hit /v1/models — should return 200 with a JSON body containing + // the OpenAI shape. With no peers serving anything, `data` should be + // an empty array but the endpoint itself must respond. + let response = http_get(host_port, "/v1/models").await; + let status_line = response.lines().next().unwrap_or_default(); + assert!( + status_line.starts_with("HTTP/1.1 200"), + "expected 200 OK from /v1/models, got status line {status_line:?}\n\nFull response:\n{response}" + ); + let body_start = response + .find("\r\n\r\n") + .expect("response has body separator") + + 4; + let body = &response[body_start..]; + assert!( + body.contains("\"data\""), + "expected JSON body containing `data` field, got: {body}" + ); + + node.stop().await.expect("stop"); + + // After stop(), the port should no longer answer. + let connect_after_stop = TcpStream::connect(host_port).await; + assert!( + connect_after_stop.is_err() + || tokio::time::timeout( + Duration::from_secs(1), + connect_after_stop.unwrap().read_u8(), + ) + .await + .is_ok(), // EOF on a half-shut connection is fine too. + "OpenAI proxy port {host_port} should be closed after MeshNode::stop()" + ); + + // Explicitly shut the anchor down so its iroh endpoint + accept + // loop don't leak into the next test. Dropping the handle alone is + // explicitly not the shutdown contract for `HostNode`. + anchor.shutdown().await; +} + +#[tokio::test] +async fn start_is_idempotent_on_repeat_calls() { + // Regression: a second .start() used to spawn a second iroh + // endpoint + OpenAI proxy and orphan the first (stop() only knew + // about the most recent). Now it's a no-op. + let (relay_url, _relay_server) = spawn_open_relay().await; + let (invite, anchor) = anchor_on_relay(&relay_url).await; + let invite_token: InviteToken = invite.parse().expect("parse invite"); + + let node = MeshNode::builder() + .identity(OwnerKeypair::generate()) + .join(invite_token) + .role(MeshRole::Client) + .relay(&relay_url) + .max_vram_gb(0.0) + .openai_port(0) + .build() + .expect("builder"); + + tokio::time::timeout(Duration::from_secs(60), node.start()) + .await + .expect("first start within 60s") + .expect("first start ok"); + + let base_before = node.openai_base_url().await.expect("first base url"); + + // Second start() must be a no-op: same proxy URL, no second bind. + tokio::time::timeout(Duration::from_secs(5), node.start()) + .await + .expect("second start should return immediately") + .expect("second start ok"); + + let base_after = node.openai_base_url().await.expect("second base url"); + assert_eq!( + base_before, base_after, + "second start() must not replace the running OpenAI proxy" + ); + + node.stop().await.expect("stop"); + anchor.shutdown().await; +} diff --git a/crates/mesh-llm-host-runtime/Cargo.toml b/crates/mesh-llm-host-runtime/Cargo.toml index 68772e4772..5e0ebf1194 100644 --- a/crates/mesh-llm-host-runtime/Cargo.toml +++ b/crates/mesh-llm-host-runtime/Cargo.toml @@ -28,7 +28,6 @@ mesh-llm-system = { path = "../mesh-llm-system", features = ["skippy-devices"] } mesh-llm-types = { path = "../mesh-llm-types" } mesh-llm-ui = { path = "../mesh-llm-ui", default-features = false } mesh-llm-node = { path = "../mesh-llm-node" } -mesh-llm-api-server = { path = "../mesh-llm-api-server" } mesh-client = { package = "mesh-llm-client", path = "../mesh-client", features = ["host-io"] } model-artifact = { path = "../model-artifact" } model-package = { path = "../model-package" } @@ -94,3 +93,10 @@ axum = "0.8" serial_test = "3" mesh-client = { package = "mesh-llm-client", path = "../mesh-client" } tempfile = "3" +# Used by the gated-relay regression test to spawn an in-process iroh-relay +# with AccessConfig::Restricted, then build a real iroh::Endpoint from our +# relay_map_from_urls output and verify --relay-auth tokens reach the relay +# on the WebSocket upgrade (and that the wrong token is rejected). +iroh = { version = "1.0.0-rc.0", features = ["test-utils"] } +iroh-relay = { version = "1.0.0-rc.0", features = ["server", "test-utils"] } +futures-util = "0.3" diff --git a/crates/mesh-llm-host-runtime/src/cli/mod.rs b/crates/mesh-llm-host-runtime/src/cli/mod.rs index 97a9328949..add2e8901c 100644 --- a/crates/mesh-llm-host-runtime/src/cli/mod.rs +++ b/crates/mesh-llm-host-runtime/src/cli/mod.rs @@ -8,6 +8,97 @@ use crate::cli::runtime::RuntimeCommand; use crate::crypto::TrustPolicy; use crate::network::discovery::MeshDiscoveryMode; +/// Parse a `URL=TOKEN` pair for `--relay-auth`. Splits on the first `=` only, +/// so tokens may contain `=` (base64 padding, JWTs). +/// +/// Error messages must never include the token portion of the input — +/// `--relay-auth` carries bearer credentials, and a parse failure could +/// otherwise leak them into terminal output, logs, and bug reports. The URL +/// is safe to echo back (it's the public identity of the relay). +fn parse_relay_auth_pair(s: &str) -> Result<(String, String), String> { + let Some((url, token)) = s.split_once('=') else { + return Err("expected URL=TOKEN, no '=' separator found (token redacted)".to_string()); + }; + if url.is_empty() { + return Err("expected URL=TOKEN, got empty URL (token redacted)".to_string()); + } + if token.is_empty() { + return Err(format!( + "expected URL=TOKEN, got empty token for URL {url:?}" + )); + } + Ok((url.to_string(), token.to_string())) +} + +#[cfg(test)] +mod relay_auth_parser_tests { + use super::parse_relay_auth_pair; + + #[test] + fn parses_simple_pair() { + let (url, token) = parse_relay_auth_pair("https://r.example/=abc123").unwrap(); + assert_eq!(url, "https://r.example/"); + assert_eq!(token, "abc123"); + } + + #[test] + fn preserves_equals_in_token() { + // Base64-padded tokens and NIP-98-style payloads often contain `=`. + let (_, token) = parse_relay_auth_pair("https://r/=eyJhbGciOiJFZERTQSJ9.payload==") + .expect("token with '=' must parse"); + assert_eq!(token, "eyJhbGciOiJFZERTQSJ9.payload=="); + } + + #[test] + fn rejects_missing_separator() { + assert!(parse_relay_auth_pair("no-separator").is_err()); + } + + #[test] + fn rejects_empty_url() { + assert!(parse_relay_auth_pair("=token").is_err()); + } + + #[test] + fn rejects_empty_token() { + assert!(parse_relay_auth_pair("https://r/=").is_err()); + } + + #[test] + fn parser_errors_never_leak_token_portion() { + // --relay-auth carries bearer credentials; if parsing fails, the + // token portion of the input must never appear in the error + // message (which lands in terminal output, logs, and bug reports). + // The URL is safe to echo back — it's the public identity of the + // relay — but everything after the first `=` is secret. + let secret_token = "super-secret-bearer-token-xyz-12345"; + + // Case 1: no `=` separator. Whole input is treated as a malformed + // URL-or-token blob; we cannot tell which it is, so redact both. + let err = parse_relay_auth_pair(secret_token).expect_err("should fail"); + assert!( + !err.contains(secret_token), + "missing-separator error must not echo the input: {err}" + ); + + // Case 2: empty URL (`=token`). URL is empty, the token portion is + // the secret — must not appear. + let err = parse_relay_auth_pair(&format!("={secret_token}")).expect_err("should fail"); + assert!( + !err.contains(secret_token), + "empty-URL error must not echo the token: {err}" + ); + + // Case 3: empty token (`URL=`). Token is empty, no secret to leak; + // the URL is fine to include and helps the user diagnose. + let err = parse_relay_auth_pair("https://r.example/=").expect_err("should fail"); + assert!( + err.contains("https://r.example/"), + "empty-token error should name the URL: {err}" + ); + } +} + #[derive(Subcommand, Debug)] pub(crate) enum TrustCommand { /// Add an owner to the local trust store allowlist. @@ -414,6 +505,23 @@ pub(crate) struct Cli { #[arg(long, hide = true)] pub(crate) relay: Vec, + /// Per-relay bearer token for gated iroh relays, formatted as + /// `URL=TOKEN`. Repeatable. The token is sent as + /// `Authorization: Bearer ` on the WebSocket upgrade to the + /// matching `--relay` URL. Relays not listed here register without + /// authentication (the correct behavior for public relays). + /// + /// Splits on the first `=` only, so tokens may contain `=` (base64 + /// padding, JWTs, etc.). + /// + /// Parsed opaquely by Clap; validation/splitting happens via + /// [`Cli::parse_relay_auths`] so we control the error output and + /// can guarantee bearer tokens never appear in failure messages + /// (Clap's default `invalid value '...' for '--relay-auth'` would + /// echo the full URL=TOKEN input on parse failure). + #[arg(long = "relay-auth", hide = true)] + pub(crate) relay_auth: Vec, + /// Bind QUIC to a fixed UDP port (for NAT port forwarding). #[arg(long, hide = true)] pub(crate) bind_port: Option, @@ -475,6 +583,24 @@ pub(crate) struct Cli { pub(crate) nostr_discovery: bool, } +impl Cli { + /// Split each `--relay-auth URL=TOKEN` entry into a `(url, token)` + /// pair, validating that both sides are present. + /// + /// On failure, the error message redacts the token portion of the + /// input. This is the contract the CLI surface promises: bearer + /// tokens never leak into terminal output, logs, or bug reports, + /// even on misconfiguration. (The opaque `Vec` Clap field + /// exists specifically so Clap's default `invalid value '...'` + /// error path can never echo the secret.) + pub(crate) fn parse_relay_auths(&self) -> anyhow::Result> { + self.relay_auth + .iter() + .map(|raw| parse_relay_auth_pair(raw).map_err(anyhow::Error::msg)) + .collect() + } +} + pub(crate) fn validate_discovery_mode_args(cli: &Cli) -> anyhow::Result<()> { if cli.mesh_discovery_mode != MeshDiscoveryMode::Mdns { return Ok(()); @@ -822,6 +948,7 @@ where "--draft", "--bin-dir", "--relay", + "--relay-auth", "--nostr-relay", "--config", "--owner-key", @@ -1083,6 +1210,82 @@ mod tests { ); } + #[test] + fn normalize_runtime_surface_args_treats_relay_auth_as_value_taking_before_serve() { + // Regression: --relay-auth carries a `URL=TOKEN` value, so the + // pseudo-subcommand scanner must skip the value and still discover + // `serve` (or `client`) as the runtime surface. If --relay-auth is not + // in the value-taking list the scanner stops at the token and Clap + // sees a malformed command. + let normalized = normalize_runtime_surface_args([ + "mesh-llm", + "--relay-auth", + "https://gated.example/=token", + "serve", + "--relay", + "https://gated.example/", + "--auto", + ]); + + assert_eq!(normalized.explicit_surface, Some(RuntimeSurface::Serve)); + assert_eq!( + normalized.normalized, + vec![ + "mesh-llm", + "--relay-auth", + "https://gated.example/=token", + "--relay", + "https://gated.example/", + "--auto", + ] + .into_iter() + .map(OsString::from) + .collect::>() + ); + + // And the resulting argv must actually parse cleanly through Clap so + // the relay-auth value reaches `Cli::relay_auth`, and the + // post-Clap split via parse_relay_auths() yields the expected + // (url, token) pair. + let cli = Cli::try_parse_from(&normalized.normalized).expect("clap parse"); + assert_eq!( + cli.relay_auth, + vec!["https://gated.example/=token".to_string()], + ); + assert_eq!( + cli.parse_relay_auths().expect("parse pair"), + vec![("https://gated.example/".to_string(), "token".to_string())], + ); + } + + #[test] + fn normalize_runtime_surface_args_relay_auth_before_client_invocation() { + // Same regression but for the `client` surface, including a token + // containing `=` (NIP-98-style base64 padding). + let normalized = normalize_runtime_surface_args([ + "mesh-llm", + "--relay-auth", + "https://gated.example/=eyJhbGciOiJFZERTQSJ9.payload==", + "client", + "--auto", + ]); + + assert_eq!(normalized.explicit_surface, Some(RuntimeSurface::Client)); + let cli = Cli::try_parse_from(&normalized.normalized).expect("clap parse"); + assert!(cli.client, "client surface flag should be set"); + assert_eq!( + cli.relay_auth, + vec!["https://gated.example/=eyJhbGciOiJFZERTQSJ9.payload==".to_string()], + ); + assert_eq!( + cli.parse_relay_auths().expect("parse pair"), + vec![( + "https://gated.example/".to_string(), + "eyJhbGciOiJFZERTQSJ9.payload==".to_string() + )], + ); + } + #[test] fn normalize_runtime_surface_args_keeps_non_runtime_subcommands() { let normalized = normalize_runtime_surface_args(["mesh-llm", "download", "foo"]); diff --git a/crates/mesh-llm-host-runtime/src/host_node.rs b/crates/mesh-llm-host-runtime/src/host_node.rs new file mode 100644 index 0000000000..5e908a7c41 --- /dev/null +++ b/crates/mesh-llm-host-runtime/src/host_node.rs @@ -0,0 +1,805 @@ +//! In-process mesh node entry point for the published SDK. +//! +//! This is the bridge between [`mesh-llm-api-server`][api-server] (the +//! public Rust SDK) and the real iroh-backed mesh node implementation in +//! `crate::mesh::Node`. Without this module the SDK can only run an +//! HTTP-shim "client" that flips `connected = true` and emits an event; +//! with it, a Rust application can `cargo add mesh-llm-api-server --features +//! host-runtime` and run an actual mesh peer that does gossip, relay +//! registration (including [`--relay-auth`][relay-auth]), invite tokens, +//! and QUIC peer connections — the same things the `mesh-llm` binary +//! does. +//! +//! [api-server]: https://docs.rs/mesh-llm-api-server +//! [relay-auth]: https://github.com/Mesh-LLM/mesh-llm/pull/641 +//! +//! ## Scope +//! +//! This module deliberately exposes only what the SDK needs: +//! +//! - [`HostNodeSpec`] — what the SDK passes in (role, relays, relay auths, +//! QUIC bind, VRAM cap, enumerate-host flag). +//! - [`HostNode`] — the handle the SDK gets back (`invite_token`, +//! `start_accepting`, `id`, `shutdown`). +//! - [`start_host_node`] — the entry point. +//! +//! It does not expose the full `mesh::Node` API. Internals stay +//! `pub(crate)` so we can keep refactoring without breaking SDK +//! consumers. +//! +//! ## What this does not do (yet) +//! +//! - It does not start local model serving. That requires plugging an +//! `EmbeddedServingController` from [`crate::sdk`] into the SDK's +//! `MeshNodeBuilder`. For client-only embedders (no GPU) this is +//! not needed — the OpenAI proxy still routes requests to remote +//! mesh peers serving the model. +//! - It does not start auto-discovery (`--auto`) of public meshes. +//! Consumers can call [`HostNode::join`] with an invite token they +//! obtained out of band (e.g. through Nostr). + +use crate::api; +use crate::inference::election; +use crate::mesh::{self, NodeRole, QuicBindSelection, RelayConfig}; +use crate::network::affinity::AffinityRouter; +use crate::network::openai::ingress::api_proxy; +use anyhow::{Context, Result}; +use std::collections::HashMap; +use std::net::SocketAddr; +use tokio::sync::{mpsc, watch}; +use tokio::task::JoinHandle; + +/// Configuration for [`start_host_node`]. +/// +/// Field shape mirrors the slice of `mesh-llm`'s CLI flags that +/// `crate::mesh::Node::start` consumes. New fields here track new CLI +/// flags as they get added. +/// +/// `Debug` is implemented manually and redacts every `relay_auths` token +/// (keys are still printed). This mirrors the token redaction in the +/// CLI parser so panics, log lines, and other `{:?}` outputs never +/// expose bearer credentials. +#[derive(Clone, Default)] +pub struct HostNodeSpec { + /// Mesh role. + pub role: NodeRole, + /// iroh relay URLs (empty = use bundled defaults). + pub relays: Vec, + /// Per-relay bearer tokens for gated iroh relays. Sent as + /// `Authorization: Bearer ` on the WebSocket upgrade to the + /// matching relay URL. + pub relay_auths: HashMap, + /// Local QUIC bind selection (IP and/or port). + pub quic_bind: QuicBindSelection, + /// VRAM cap in GB. `Some(0.0)` for client-only nodes that should not + /// advertise any VRAM. + pub max_vram_gb: Option, + /// Whether to publish a hardware survey to gossip. + pub enumerate_host: bool, +} + +impl std::fmt::Debug for HostNodeSpec { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("HostNodeSpec") + .field("role", &self.role) + .field("relays", &self.relays) + .field("relay_auths", &RedactedAuthMap(&self.relay_auths)) + .field("quic_bind", &self.quic_bind) + .field("max_vram_gb", &self.max_vram_gb) + .field("enumerate_host", &self.enumerate_host) + .finish() + } +} + +/// Helper that renders a `relay_auths` map with token values replaced +/// by `` and a length hint. Keys (relay URLs) are public. +struct RedactedAuthMap<'a>(&'a HashMap); + +impl std::fmt::Debug for RedactedAuthMap<'_> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let mut m = f.debug_map(); + for (url, token) in self.0 { + m.key(url); + m.value(&format_args!("", token.len())); + } + m.finish() + } +} + +/// A running mesh node started by [`start_host_node`]. +/// +/// Call [`HostNode::shutdown`] to stop the iroh endpoint and tear down +/// background tasks. Dropping this handle alone is not the shutdown +/// contract because spawned mesh tasks hold their own node clones. +#[derive(Clone)] +pub struct HostNode { + inner: mesh::Node, +} + +impl HostNode { + /// Start accepting incoming mesh connections. + /// + /// The iroh endpoint binds in [`start_host_node`], but the accept + /// loop waits for this call so the embedder can finish wiring (set a + /// display name, advertise models) before the node is reachable. + pub fn start_accepting(&self) { + self.inner.start_accepting(); + } + + /// Hex-formatted endpoint ID, suitable for logging. + pub fn id(&self) -> String { + self.inner.id().to_string() + } + + /// An invite token that other nodes can use to join this one. + pub fn invite_token(&self) -> String { + self.inner.invite_token() + } + + /// Join an existing mesh via an invite token produced elsewhere. + pub async fn join(&self, invite_token: &str) -> Result<()> { + self.inner.join(invite_token).await + } + + /// Set a human-readable display name advertised to peers. + pub async fn set_display_name(&self, name: String) { + self.inner.set_display_name(name).await; + } + + /// Replace the set of models this node advertises. + pub async fn set_models(&self, models: Vec) { + self.inner.set_models(models).await; + } + + /// Current set of advertised models. + pub async fn models(&self) -> Vec { + self.inner.models().await + } + + /// Shut the node down (best-effort). + pub async fn shutdown(&self) { + self.inner.shutdown().await; + } +} + +/// Bring an in-process mesh node online with the given spec. +/// +/// Equivalent to the iroh-endpoint slice of `mesh-llm serve` / `mesh-llm +/// client`: binds the iroh endpoint, attaches relay-auth tokens, waits +/// briefly for the home relay to come online, and returns a handle. The +/// caller is responsible for any further wiring (calling +/// [`HostNode::start_accepting`], setting models / display name, joining +/// other meshes via [`HostNode::join`]). +pub async fn start_host_node(spec: HostNodeSpec) -> Result { + let relay = RelayConfig { + urls: &spec.relays, + auths: &spec.relay_auths, + }; + let (node, _channels) = mesh::Node::start( + spec.role, + relay, + spec.quic_bind, + spec.max_vram_gb, + spec.enumerate_host, + None, // owner control config — not currently exposed to SDK + None, // config file — not relevant to SDK consumers + ) + .await?; + Ok(HostNode { inner: node }) +} + +// Re-export the types embedders need to express a spec. Hidden inside +// the curated `host_node` namespace, NOT at the crate root, so we can +// keep refactoring the underlying `mesh` module. +pub use mesh::{NodeRole as MeshNodeRole, QuicBindSelection as MeshQuicBindSelection}; + +/// Handle to a running in-process OpenAI HTTP proxy started by +/// [`start_openai_proxy`]. +/// +/// Drop the handle (or call [`OpenAiProxyHandle::shutdown`]) to stop the +/// proxy. While alive, requests against +/// `http://{bound_addr}/v1/{chat/completions,models,…}` route to mesh +/// peers via the underlying `HostNode`'s gossip + QUIC transport. +pub struct OpenAiProxyHandle { + addr: SocketAddr, + task: JoinHandle<()>, + /// Held so the no-op runtime-control receiver isn't dropped while the + /// proxy is alive (the proxy sends control requests on this channel; + /// dropping the receiver would close the sender and surface spurious + /// errors). Drained on a background task. + _control_drain: JoinHandle<()>, +} + +impl OpenAiProxyHandle { + /// The local address the proxy is bound to. When the embedder asks for + /// port 0 this is the OS-assigned ephemeral port. + pub fn local_addr(&self) -> SocketAddr { + self.addr + } + + /// Base URL suitable for OpenAI-compatible client libraries. + pub fn base_url(&self) -> String { + format!("http://{}", self.addr) + } + + /// Stop the proxy task. Idempotent; safe to call after drop(). + pub fn shutdown(&self) { + self.task.abort(); + self._control_drain.abort(); + } +} + +impl Drop for OpenAiProxyHandle { + fn drop(&mut self) { + self.task.abort(); + self._control_drain.abort(); + } +} + +/// Start an OpenAI-compatible HTTP proxy that fronts a [`HostNode`]. +/// +/// Equivalent to the `--port` slice of `mesh-llm serve` / `mesh-llm +/// client`: binds a TCP listener, accepts HTTP connections, parses +/// requests, and routes them to mesh peers that advertise the requested +/// model in gossip. Suitable for client-only embedders (no local +/// serving) and for embedders that have plugged a `ServingController` +/// into their `MeshNode` for local inference. +/// +/// Returns once the listener is bound; the proxy keeps running in a +/// background task until [`OpenAiProxyHandle::shutdown`] or drop. +/// +/// The `port = 0` case is supported and asks the OS for an ephemeral +/// port; read it from [`OpenAiProxyHandle::local_addr`] after this call +/// returns. +pub async fn start_openai_proxy( + node: &HostNode, + port: u16, + listen_all: bool, +) -> Result { + let bind_addr = if listen_all { + format!("0.0.0.0:{port}") + } else { + format!("127.0.0.1:{port}") + }; + let listener = tokio::net::TcpListener::bind(&bind_addr) + .await + .with_context(|| format!("binding OpenAI proxy to {bind_addr}"))?; + let local_addr = listener + .local_addr() + .context("reading OpenAI proxy local addr")?; + + // Targets watch channel: starts empty. Routing to remote peers does + // not read this — it reads `node.hosts_for_model()` at request time. + // The channel only matters if the embedder later wires local serving + // through `crate::sdk::EmbeddedServingController`. + let (_target_tx, target_rx) = watch::channel(election::ModelTargets::default()); + + // Runtime-control channel: the proxy sends model load/unload commands + // here. Embedders without local serving have no one to handle these, + // so spawn a background drain that just logs and discards. + let (control_tx, mut control_rx) = mpsc::unbounded_channel::(); + let control_drain = tokio::spawn(async move { + // Discard with a single line per request. We deliberately don't + // {:?} the request because RuntimeControlRequest doesn't impl + // Debug and is an internal type the SDK shouldn't widen for a + // background log line. + while control_rx.recv().await.is_some() { + tracing::debug!( + "SDK-mode OpenAI proxy received a runtime-control request with no handler attached; discarding" + ); + } + }); + + let affinity = AffinityRouter::new(); + let node_for_proxy = node.inner.clone(); + let task = tokio::spawn(async move { + api_proxy( + node_for_proxy, + local_addr.port(), + target_rx, + control_tx, + Some(listener), + listen_all, + affinity, + ) + .await; + }); + + Ok(OpenAiProxyHandle { + addr: local_addr, + task, + _control_drain: control_drain, + }) +} + +/// Full mesh-llm runtime configuration for [`run_serve`]. +/// +/// Every field maps to a `mesh-llm` CLI flag. Defaults match the +/// binary's defaults so the SDK consumer only sets what they want +/// different. +/// +/// Unlike [`HostNodeSpec`] (which only brings up the iroh endpoint), +/// a `MeshServeSpec` drives the **full** runtime path — the same code +/// path `mesh-llm serve` / `mesh-llm client` use. That means election, +/// tunnel manager, OpenAI proxy, management console, auto-discovery, +/// local model serving, plugin host — everything the binary does. +/// +/// `Debug` is implemented manually and redacts every `relay_auths` +/// token (keys are still printed) so bearer credentials never appear +/// in log/panic output. +#[derive(Clone, Default)] +pub struct MeshServeSpec { + /// Run as a client only (no GPU, no model). Maps to `--client`. + pub client: bool, + /// Auto-join the best discovered mesh. Maps to `--auto`. + pub auto: bool, + /// Publish this mesh for Nostr discovery. Maps to `--publish`. + pub publish: bool, + /// Human-readable mesh name. Maps to `--mesh-name`. + pub mesh_name: Option, + /// Region tag (e.g. "US"). Maps to `--region`. + pub region: Option, + /// Blackboard display name. Maps to `--name`. + pub display_name: Option, + /// Invite tokens to join. Maps to repeatable `--join `. + pub join: Vec, + /// Discovery filter (mesh name). Maps to `--discover [filter]`. + pub discover: Option, + + /// Models to serve. Path, catalog name, or HF ref. Maps to + /// repeatable `--model `. + pub models: Vec, + /// Raw local GGUF files. Maps to repeatable `--gguf `. + pub ggufs: Vec, + /// Explicit mmproj sidecar. Maps to `--mmproj`. + pub mmproj: Option, + + /// OpenAI API port. Default 9337. Maps to `--port`. + pub port: Option, + /// Console port. Default 3131. Maps to `--console`. + pub console_port: Option, + /// Disable the embedded web UI but keep the management API. Maps + /// to `--headless`. + pub headless: bool, + /// Enable blackboard on public meshes. Maps to `--blackboard`. + pub blackboard: bool, + + /// iroh relay URLs. Maps to repeatable `--relay `. + pub relays: Vec, + /// Per-relay bearer tokens for gated iroh relays. Maps to + /// repeatable `--relay-auth URL=TOKEN`. + pub relay_auths: HashMap, + /// Custom Nostr relay URLs. Maps to repeatable `--nostr-relay`. + pub nostr_relays: Vec, + /// Fixed QUIC bind port (NAT forwarding). Maps to `--bind-port`. + pub bind_port: Option, + /// Local QUIC bind IP. Maps to `--bind-ip`. + pub bind_ip: Option, + /// Bind to 0.0.0.0 instead of 127.0.0.1. Maps to `--listen-all`. + pub listen_all: bool, + + /// VRAM cap in GB. Maps to `--max-vram`. + pub max_vram_gb: Option, + /// Disable hardware survey gossip. Maps to `--no-enumerate-host`. + pub no_enumerate_host: bool, + + /// Config file path. Maps to `--config`. + pub config: Option, + /// Owner keystore path. Maps to `--owner-key`. + pub owner_key: Option, + /// Fail startup without owner attestation. Maps to `--owner-required`. + pub owner_required: bool, + /// Node certificate label. Maps to `--node-label`. + pub node_label: Option, + /// Add trusted owner IDs. Maps to repeatable `--trust-owner`. + pub trust_owners: Vec, + + /// Enable mesh runtime debug output. Maps to `--debug`. + pub debug: bool, + + /// Extra raw argv flags for anything this struct doesn't yet + /// expose typed. Inserted after the typed flags. Use sparingly. + pub extra_args: Vec, +} + +impl std::fmt::Debug for MeshServeSpec { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MeshServeSpec") + .field("client", &self.client) + .field("auto", &self.auto) + .field("publish", &self.publish) + .field("mesh_name", &self.mesh_name) + .field("region", &self.region) + .field("display_name", &self.display_name) + .field("join", &self.join) + .field("discover", &self.discover) + .field("models", &self.models) + .field("ggufs", &self.ggufs) + .field("mmproj", &self.mmproj) + .field("port", &self.port) + .field("console_port", &self.console_port) + .field("headless", &self.headless) + .field("blackboard", &self.blackboard) + .field("relays", &self.relays) + .field("relay_auths", &RedactedAuthMap(&self.relay_auths)) + .field("nostr_relays", &self.nostr_relays) + .field("bind_port", &self.bind_port) + .field("bind_ip", &self.bind_ip) + .field("listen_all", &self.listen_all) + .field("max_vram_gb", &self.max_vram_gb) + .field("no_enumerate_host", &self.no_enumerate_host) + .field("config", &self.config) + .field("owner_key", &self.owner_key) + .field("owner_required", &self.owner_required) + .field("node_label", &self.node_label) + .field("trust_owners", &self.trust_owners) + .field("debug", &self.debug) + .field("extra_args", &self.extra_args) + .finish() + } +} + +impl MeshServeSpec { + /// Serialise this spec into a CLI argv vector. Exposed primarily + /// for tests and embedders that want to see exactly what they're + /// about to run. + pub fn into_argv(self) -> Vec { + let mut argv: Vec = Vec::new(); + argv.push("mesh-llm".into()); + argv.push(if self.client { "client" } else { "serve" }.into()); + self.append_top_level(&mut argv); + self.append_model_args(&mut argv); + self.append_ports(&mut argv); + self.append_relay_args(&mut argv); + self.append_bind_args(&mut argv); + self.append_owner_args(&mut argv); + argv.extend(self.extra_args.into_iter().map(Into::into)); + argv + } + + fn append_top_level(&self, argv: &mut Vec) { + if self.debug { + argv.push("--debug".into()); + } + if self.auto { + argv.push("--auto".into()); + } + if self.publish { + argv.push("--publish".into()); + } + if let Some(name) = &self.mesh_name { + argv.push("--mesh-name".into()); + argv.push(name.into()); + } + if let Some(region) = &self.region { + argv.push("--region".into()); + argv.push(region.into()); + } + if let Some(display) = &self.display_name { + argv.push("--name".into()); + argv.push(display.into()); + } + for invite in &self.join { + argv.push("--join".into()); + argv.push(invite.into()); + } + if let Some(filter) = &self.discover { + argv.push("--discover".into()); + argv.push(filter.into()); + } + if self.headless { + argv.push("--headless".into()); + } + if self.blackboard { + argv.push("--blackboard".into()); + } + if let Some(gb) = self.max_vram_gb { + argv.push("--max-vram".into()); + argv.push(gb.to_string().into()); + } + if self.no_enumerate_host { + argv.push("--no-enumerate-host".into()); + } + } + + fn append_model_args(&self, argv: &mut Vec) { + for model in &self.models { + argv.push("--model".into()); + argv.push(model.into()); + } + for gguf in &self.ggufs { + argv.push("--gguf".into()); + argv.push(gguf.as_os_str().to_os_string()); + } + if let Some(mmproj) = &self.mmproj { + argv.push("--mmproj".into()); + argv.push(mmproj.as_os_str().to_os_string()); + } + } + + fn append_ports(&self, argv: &mut Vec) { + if let Some(port) = self.port { + argv.push("--port".into()); + argv.push(port.to_string().into()); + } + if let Some(console) = self.console_port { + argv.push("--console".into()); + argv.push(console.to_string().into()); + } + } + + fn append_relay_args(&self, argv: &mut Vec) { + for relay in &self.relays { + argv.push("--relay".into()); + argv.push(relay.into()); + } + for (url, token) in &self.relay_auths { + argv.push("--relay-auth".into()); + argv.push(format!("{url}={token}").into()); + } + for url in &self.nostr_relays { + argv.push("--nostr-relay".into()); + argv.push(url.into()); + } + } + + fn append_bind_args(&self, argv: &mut Vec) { + if let Some(port) = self.bind_port { + argv.push("--bind-port".into()); + argv.push(port.to_string().into()); + } + if let Some(ip) = self.bind_ip { + argv.push("--bind-ip".into()); + argv.push(ip.to_string().into()); + } + if self.listen_all { + argv.push("--listen-all".into()); + } + } + + fn append_owner_args(&self, argv: &mut Vec) { + if let Some(config) = &self.config { + argv.push("--config".into()); + argv.push(config.as_os_str().to_os_string()); + } + if let Some(owner_key) = &self.owner_key { + argv.push("--owner-key".into()); + argv.push(owner_key.as_os_str().to_os_string()); + } + if self.owner_required { + argv.push("--owner-required".into()); + } + if let Some(label) = &self.node_label { + argv.push("--node-label".into()); + argv.push(label.into()); + } + for owner in &self.trust_owners { + argv.push("--trust-owner".into()); + argv.push(owner.into()); + } + } +} + +/// Run the full mesh-llm runtime in-process. +/// +/// This is the in-process equivalent of running the `mesh-llm` binary. +/// Everything the CLI does happens here: auto-discovery, election, +/// tunnel manager, OpenAI HTTP proxy, management console, model load / +/// serving (when a serving controller / GGUF is configured), plugin +/// host — driven by the same `runtime::run_with_args` entry point +/// `mesh-llm serve` / `mesh-llm client` use. +/// +/// The future blocks until the runtime exits (signal, internal +/// shutdown, or fatal error). Embedders driving concurrent work should +/// run this on a `tokio::task::LocalSet` because the runtime is not +/// currently `Send`-clean. +/// +/// # Example +/// +/// ```no_run +/// # use std::collections::HashMap; +/// use mesh_llm_host_runtime::host_node::{run_serve, MeshServeSpec}; +/// +/// # async fn run() -> anyhow::Result<()> { +/// let mut auths = HashMap::new(); +/// auths.insert( +/// "https://gated.example/".to_string(), +/// "".to_string(), +/// ); +/// +/// run_serve(MeshServeSpec { +/// client: true, +/// auto: true, +/// relays: vec!["https://gated.example/".into()], +/// relay_auths: auths, +/// port: Some(9337), +/// console_port: Some(3131), +/// headless: true, +/// max_vram_gb: Some(0.0), +/// ..MeshServeSpec::default() +/// }) +/// .await?; +/// # Ok(()) +/// # } +/// ``` +pub async fn run_serve(spec: MeshServeSpec) -> Result<()> { + crate::run_with_args(spec.into_argv()).await +} + +#[cfg(test)] +mod tests { + use super::*; + use iroh::endpoint::{presets, Endpoint, RelayMode}; + use iroh::SecretKey; + use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket}; + use std::time::Duration; + + fn free_local_udp_port() -> u16 { + let socket = UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], 0))) + .expect("allocate local UDP port"); + socket.local_addr().expect("read local UDP port").port() + } + + async fn probe_quic_port_released(port: u16) -> anyhow::Result<()> { + let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port); + let mut last_error = None; + + for _ in 0..20 { + match Endpoint::builder(presets::Minimal) + .secret_key(SecretKey::generate()) + .relay_mode(RelayMode::Disabled) + .bind_addr(bind_addr)? + .bind() + .await + { + Ok(endpoint) => { + endpoint.close().await; + return Ok(()); + } + Err(err) => { + last_error = Some(err); + tokio::time::sleep(Duration::from_millis(25)).await; + } + } + } + + Err(anyhow::anyhow!( + "host-node shutdown should release UDP port {port}: {:?}", + last_error + )) + } + + #[tokio::test] + async fn id_returns_bare_hex_endpoint_id() -> anyhow::Result<()> { + let inner = mesh::Node::new_for_tests(mesh::NodeRole::Client).await?; + let expected = inner.id().to_string(); + let node = HostNode { inner }; + + assert_eq!(node.id(), expected); + assert!(!node.id().contains("PublicKey")); + + node.shutdown().await; + Ok(()) + } + + #[tokio::test] + async fn shutdown_closes_the_mesh_endpoint() -> anyhow::Result<()> { + let inner = mesh::Node::new_for_tests(mesh::NodeRole::Client).await?; + let node = HostNode { inner }; + + node.shutdown().await; + + assert!(node.inner.endpoint_is_closed_for_tests()); + Ok(()) + } + + #[tokio::test] + async fn shutdown_releases_fixed_quic_bind() -> anyhow::Result<()> { + let quic_port = free_local_udp_port(); + let node = start_host_node(HostNodeSpec { + role: MeshNodeRole::Client, + quic_bind: MeshQuicBindSelection { + ip: Some(IpAddr::V4(Ipv4Addr::LOCALHOST)), + port: Some(quic_port), + }, + max_vram_gb: Some(0.0), + enumerate_host: false, + ..HostNodeSpec::default() + }) + .await?; + + node.start_accepting(); + node.shutdown().await; + drop(node); + + probe_quic_port_released(quic_port).await + } + + #[test] + #[allow(clippy::cognitive_complexity)] + fn mesh_serve_spec_argv_parses_via_the_real_cli_parser() { + // The MeshServeSpec exists so SDK consumers can drive the same + // runtime the binary drives. If a flag we emit doesn't exist in + // the real Clap surface (typo, renamed, removed), this test + // fails immediately and points at the drifted field. + use clap::Parser; + + let mut auths = HashMap::new(); + auths.insert( + "https://gated.example/".to_string(), + "bearer-abc".to_string(), + ); + + let spec = MeshServeSpec { + client: true, + auto: true, + publish: false, + mesh_name: Some("my-mesh".into()), + region: Some("US".into()), + display_name: Some("sprout".into()), + join: vec!["invite-1".into(), "invite-2".into()], + discover: Some("public".into()), + models: vec!["Qwen3-8B-Q4_K_M".into()], + ggufs: vec!["/tmp/foo.gguf".into()], + mmproj: None, + port: Some(9337), + console_port: Some(3131), + headless: true, + blackboard: false, + relays: vec!["https://gated.example/".into()], + relay_auths: auths, + nostr_relays: vec![], + bind_port: Some(45000), + bind_ip: None, + listen_all: false, + max_vram_gb: Some(0.0), + no_enumerate_host: true, + config: None, + owner_key: None, + owner_required: false, + node_label: Some("sprout-app".into()), + trust_owners: vec!["owner-abc".into()], + debug: false, + extra_args: vec![], + }; + + let argv = spec.into_argv(); + let normalized = crate::cli::normalize_runtime_surface_args(argv); + let cli = crate::cli::Cli::try_parse_from(&normalized.normalized) + .expect("MeshServeSpec argv must parse via the real CLI"); + + assert!(cli.client); + assert!(cli.auto); + assert!(!cli.publish); + assert_eq!(cli.mesh_name.as_deref(), Some("my-mesh")); + assert_eq!(cli.region.as_deref(), Some("US")); + assert_eq!(cli.name.as_deref(), Some("sprout")); + assert_eq!( + cli.join, + vec!["invite-1".to_string(), "invite-2".to_string()] + ); + assert_eq!(cli.discover.as_deref(), Some("public")); + assert_eq!(cli.model, vec![std::path::PathBuf::from("Qwen3-8B-Q4_K_M")]); + assert_eq!(cli.gguf, vec![std::path::PathBuf::from("/tmp/foo.gguf")]); + assert_eq!(cli.port, 9337); + assert_eq!(cli.console, 3131); + assert!(cli.headless); + assert_eq!(cli.relay, vec!["https://gated.example/".to_string()]); + assert_eq!( + cli.relay_auth, + vec!["https://gated.example/=bearer-abc".to_string()], + ); + assert_eq!( + cli.parse_relay_auths().expect("parse pair"), + vec![( + "https://gated.example/".to_string(), + "bearer-abc".to_string() + )], + ); + assert_eq!(cli.bind_port, Some(45000)); + assert_eq!(cli.max_vram, Some(0.0)); + assert!(cli.no_enumerate_host); + assert_eq!(cli.node_label.as_deref(), Some("sprout-app")); + assert_eq!(cli.trust_owner, vec!["owner-abc".to_string()]); + } +} diff --git a/crates/mesh-llm-host-runtime/src/lib.rs b/crates/mesh-llm-host-runtime/src/lib.rs index 7b92912760..9511357495 100644 --- a/crates/mesh-llm-host-runtime/src/lib.rs +++ b/crates/mesh-llm-host-runtime/src/lib.rs @@ -15,6 +15,7 @@ mod runtime; mod runtime_data; mod system; +pub mod host_node; pub mod sdk; pub mod proto { @@ -32,6 +33,31 @@ pub async fn run() -> Result<()> { runtime::run().await } +/// Run the full mesh-llm runtime with a caller-supplied argv. +/// +/// Equivalent to `run()` except the argv comes from the caller instead +/// of `std::env::args_os()`. This is the SDK entry point for embedders +/// who want to run the same code path the binary runs — full +/// `mesh-llm serve` / `mesh-llm client` behaviour, including auto-discover, +/// local model serving (when configured), election, tunnel manager, +/// OpenAI proxy, and management console — from inside their own Rust +/// application. +/// +/// Build the argv from a typed config via +/// [`host_node::MeshServeSpec`][crate::host_node::MeshServeSpec] for +/// type-safety, or pass a `Vec<&str>` directly if you want raw control. +/// +/// The future returned blocks until the runtime exits. Embedders +/// driving concurrent work should use `tokio::task::LocalSet` (the +/// runtime is not currently `Send`-clean). +pub async fn run_with_args(args: I) -> Result<()> +where + I: IntoIterator, + S: Into, +{ + runtime::run_with_args(args).await +} + pub async fn run_main() -> i32 { match run().await { Ok(()) => 0, diff --git a/crates/mesh-llm-host-runtime/src/mesh/mod.rs b/crates/mesh-llm-host-runtime/src/mesh/mod.rs index 48414dcdbb..b46abcf7e2 100644 --- a/crates/mesh-llm-host-runtime/src/mesh/mod.rs +++ b/crates/mesh-llm-host-runtime/src/mesh/mod.rs @@ -255,6 +255,17 @@ pub struct QuicBindSelection { pub port: Option, } +/// Relay map plus per-relay bearer tokens for gated iroh-relays. +/// +/// `urls` is the relay map; `auths` is a sparse map of relay URL -> bearer +/// token used when registering with relays running `AccessConfig::Restricted`. +/// Public relays in the same map continue to register without auth. +#[derive(Clone, Copy, Debug)] +pub struct RelayConfig<'a> { + pub urls: &'a [String], + pub auths: &'a std::collections::HashMap, +} + fn quic_bind_addr(bind: QuicBindSelection) -> Option { if let Some(ip) = bind.ip { return Some(SocketAddr::new( @@ -409,11 +420,344 @@ fn effective_relay_urls(relay_urls: &[String]) -> Vec { } } -fn relay_map_from_urls(urls: &[String]) -> iroh::RelayMap { - let configs = urls - .iter() - .map(|url| iroh::RelayConfig::new(url.parse().expect("invalid relay URL"), None)); - iroh::RelayMap::from_iter(configs) +/// Build an [`iroh::RelayMap`] from URLs, attaching per-relay auth tokens +/// where configured. +/// +/// `auths` maps relay URLs (as they appear in `urls`) to bearer tokens. Tokens +/// are passed to `iroh::RelayConfig::with_auth_token` which sends them as +/// `Authorization: Bearer ` on the WebSocket upgrade. Relays not present +/// in the map register unauthenticated, which is the correct behavior for +/// public (`AccessConfig::Everyone`) relays. +/// +/// This is the wire-up that lets a gated iroh-relay (e.g. one running +/// `AccessConfig::Restricted` with NIP-98 admission) admit this node while +/// public relays in the same map continue to work normally. +fn relay_map_from_urls( + urls: &[String], + auths: &std::collections::HashMap, +) -> Result { + // Canonicalise auth-map keys via `iroh::RelayUrl` parse so logically + // equivalent URLs match (e.g. `https://x.example/` vs `https://x.example`). + // Without this, a user-supplied auth keyed by `https://x.example` would + // be silently dropped when `urls` contains `https://x.example/`, + // causing a gated relay to reject the registration. + // + // Invalid auth keys are surfaced as Err so misconfiguration is loud + // rather than silent. + let mut canonical_auths: std::collections::HashMap = + std::collections::HashMap::with_capacity(auths.len()); + for (raw_url, token) in auths { + let canonical: iroh::RelayUrl = raw_url + .parse() + .with_context(|| format!("invalid relay-auth URL `{raw_url}`"))?; + canonical_auths.insert(canonical, token); + } + + let configs = urls.iter().map(|url| { + let parsed: iroh::RelayUrl = url + .parse() + .with_context(|| format!("invalid relay URL `{url}`"))?; + let cfg = iroh::RelayConfig::new(parsed.clone(), None); + Ok(match canonical_auths.get(&parsed) { + Some(token) => cfg.with_auth_token((*token).clone()), + None => cfg, + }) + }); + Ok(iroh::RelayMap::from_iter( + configs.collect::>>()?, + )) +} + +#[cfg(test)] +mod relay_map_tests { + use super::relay_map_from_urls; + use std::collections::HashMap; + use std::sync::Arc; + + fn configs(map: &iroh::RelayMap) -> Vec> { + map.relays::>() + } + + #[test] + fn builds_map_without_auth_when_empty() { + let urls = vec!["https://r1.example/".to_string()]; + let map = relay_map_from_urls(&urls, &HashMap::new()).expect("relay map"); + let cfgs = configs(&map); + assert_eq!(cfgs.len(), 1); + assert!( + cfgs[0].auth_token.is_none(), + "no auth supplied → no auth_token set" + ); + } + + #[test] + fn attaches_auth_token_for_matching_url() { + let urls = vec!["https://gated.example/".to_string()]; + let mut auths = HashMap::new(); + auths.insert("https://gated.example/".to_string(), "nip98-bearer".into()); + let map = relay_map_from_urls(&urls, &auths).expect("relay map"); + let cfgs = configs(&map); + assert_eq!(cfgs.len(), 1); + assert_eq!(cfgs[0].auth_token.as_deref(), Some("nip98-bearer")); + } + + #[test] + fn auth_token_matches_canonicalised_url_with_or_without_trailing_slash() { + // Regression: relay_map_from_urls used to do a raw string lookup + // on the auths map, so `auths={"https://x/":token}` would silently + // miss `urls=["https://x"]`. A gated relay would then reject the + // registration with no clear signal. Canonicalising via + // iroh::RelayUrl normalises both sides. + let urls = vec!["https://gated.example".to_string()]; // no trailing / + let mut auths = HashMap::new(); + auths.insert("https://gated.example/".to_string(), "bearer".into()); // with / + + let map = relay_map_from_urls(&urls, &auths).expect("relay map"); + let cfgs = configs(&map); + assert_eq!(cfgs.len(), 1); + assert_eq!( + cfgs[0].auth_token.as_deref(), + Some("bearer"), + "trailing-slash variation must still match the auth token" + ); + } + + #[test] + fn malformed_relay_auth_key_surfaces_as_error() { + // Misconfigured auth keys should be loud, not silent. Without + // canonicalisation a bad key would just fail to match any URL + // and the gated relay would reject us without context. Now the + // build returns an Err naming the offending key. + let urls = vec!["https://gated.example/".to_string()]; + let mut auths = HashMap::new(); + auths.insert("not a url".to_string(), "bearer".into()); + + let err = + relay_map_from_urls(&urls, &auths).expect_err("malformed auth key must surface as Err"); + let msg = format!("{err:#}"); + assert!( + msg.contains("invalid relay-auth URL") && msg.contains("not a url"), + "error should name the offending auth key: {msg}" + ); + } + + #[test] + fn leaves_other_relays_unauthenticated_in_mixed_map() { + // The whole point: gated relay gets a token, public relays don't. + let urls = vec![ + "https://gated.example/".to_string(), + "https://public.iroh/".to_string(), + ]; + let mut auths = HashMap::new(); + auths.insert("https://gated.example/".to_string(), "bearer-xyz".into()); + + let map = relay_map_from_urls(&urls, &auths).expect("relay map"); + let by_url: HashMap> = configs(&map) + .into_iter() + .map(|cfg| (cfg.url.to_string(), cfg.auth_token.clone())) + .collect(); + + // Find the entries by matching on host substring, since iroh-relay may + // canonicalise the URL form (e.g. trailing dot on the host). + let gated = by_url + .iter() + .find(|(u, _)| u.contains("gated.example")) + .expect("gated relay should be in the map"); + let public = by_url + .iter() + .find(|(u, _)| u.contains("public.iroh")) + .expect("public relay should be in the map"); + + assert_eq!( + gated.1.as_deref(), + Some("bearer-xyz"), + "gated relay must carry its token" + ); + assert!( + public.1.is_none(), + "public relay must register without a token, got {:?}", + public.1 + ); + } + + #[test] + fn invalid_relay_url_does_not_panic() { + let urls = vec!["not a relay url".to_string()]; + + let result = std::panic::catch_unwind(|| relay_map_from_urls(&urls, &HashMap::new())); + + assert!( + result.is_ok(), + "invalid relay URLs should be returned as errors" + ); + assert!( + result.expect("relay URL parsing should not panic").is_err(), + "invalid relay URLs should return an error" + ); + } +} + +/// End-to-end regression tests for `--relay-auth` against a real in-process +/// iroh-relay running [`iroh_relay::server::AccessConfig::Restricted`]. +/// +/// These tests do not go through the full `Node::start` path — they exercise +/// `relay_map_from_urls` (the new wiring) plus the iroh `Endpoint` builder +/// the same way `bind_mesh_endpoint` does, with `ca_roots_config` overridden +/// for the relay's self-signed test cert. The contract being defended is: +/// +/// 1. A token configured for a gated relay URL reaches iroh as +/// `RelayConfig::with_auth_token`, gets sent as `Authorization: Bearer` +/// on the WebSocket upgrade, and the relay admits the endpoint. +/// 2. The wrong token (or no token) is rejected with `not authorized` and +/// the endpoint never reaches `online()`. +/// 3. Mixed maps work: a gated relay with the right token coexists with a +/// public relay (no token) in the same `RelayMap`. +#[cfg(test)] +mod gated_relay_e2e_tests { + use super::relay_map_from_urls; + use futures_util::StreamExt; + use iroh::endpoint::{presets, Endpoint, RelayMode}; + use iroh::test_utils::run_relay_server_with_access; + use iroh::SecretKey; + use iroh::Watcher; + use iroh_relay::server::{Access, AccessConfig}; + use iroh_relay::tls::CaRootsConfig; + use std::collections::HashMap; + use std::time::Duration; + + /// Spawn an in-process iroh-relay that only admits `expected_token`. + /// Returns (relay_url_string, drop-guard server). + async fn spawn_gated_relay( + expected_token: &'static str, + ) -> (String, iroh_relay::server::Server) { + let access = AccessConfig::Restricted(Box::new(move |request| { + Box::pin(async move { + if request.auth_token().as_deref() == Some(expected_token) { + Access::Allow + } else { + Access::Deny + } + }) + })); + let (_relay_map, relay_url, server) = run_relay_server_with_access(false, access) + .await + .expect("spawn gated relay"); + (relay_url.to_string(), server) + } + + /// Build an `Endpoint` configured the same way `bind_mesh_endpoint` does, + /// but using `relay_map_from_urls` for the relay map and accepting the + /// relay's self-signed test cert via `insecure_skip_verify`. + async fn build_endpoint( + relay_urls: &[String], + relay_auths: &HashMap, + ) -> Endpoint { + Endpoint::builder(presets::Minimal) + .secret_key(SecretKey::generate()) + .relay_mode(RelayMode::Custom( + relay_map_from_urls(relay_urls, relay_auths).expect("relay map"), + )) + .ca_roots_config(CaRootsConfig::insecure_skip_verify()) + .bind() + .await + .expect("endpoint bind") + } + + #[tokio::test] + async fn matching_token_admits_endpoint_to_gated_relay() { + const TOKEN: &str = "secret-token"; + let (relay_url, _server) = spawn_gated_relay(TOKEN).await; + + let urls = vec![relay_url.clone()]; + let mut auths = HashMap::new(); + auths.insert(relay_url, TOKEN.to_string()); + + let ep = build_endpoint(&urls, &auths).await; + tokio::time::timeout(Duration::from_secs(5), ep.online()) + .await + .expect("endpoint with matching token should come online"); + } + + #[tokio::test] + async fn wrong_token_is_rejected_by_gated_relay() { + const TOKEN: &str = "secret-token"; + let (relay_url, _server) = spawn_gated_relay(TOKEN).await; + + let urls = vec![relay_url.clone()]; + let mut auths = HashMap::new(); + auths.insert(relay_url, "wrong-token".to_string()); + + let ep = build_endpoint(&urls, &auths).await; + + // Observe the relay-side denial via home_relay_status before falling + // back to the timeout. We must see `not authorized` to prove the + // token actually reached the relay (rather than e.g. silently being + // dropped before the WebSocket upgrade). + let mut stream = ep.home_relay_status().stream(); + let auth_err = tokio::time::timeout(Duration::from_secs(5), async { + while let Some(status) = stream.next().await { + if let Some(err) = status.iter().filter_map(|s| s.last_error()).next() { + return Some(format!("{err:#}")); + } + } + None + }) + .await + .expect("home relay status should report an error within 5s") + .expect("home relay status should yield an error"); + assert!( + auth_err.contains("not authorized"), + "expected 'not authorized' in error, got: {auth_err}" + ); + + // And the endpoint must NOT come online. + let online = tokio::time::timeout(Duration::from_millis(500), ep.online()).await; + assert!( + online.is_err(), + "endpoint with wrong token must not reach online() within 500ms" + ); + } + + #[tokio::test] + async fn missing_token_for_gated_relay_is_rejected() { + const TOKEN: &str = "secret-token"; + let (relay_url, _server) = spawn_gated_relay(TOKEN).await; + + // No auth in the map at all → relay must deny. + let urls = vec![relay_url]; + let auths = HashMap::new(); + let ep = build_endpoint(&urls, &auths).await; + + let online = tokio::time::timeout(Duration::from_millis(500), ep.online()).await; + assert!( + online.is_err(), + "endpoint without a token must not be admitted by a gated relay" + ); + } + + #[tokio::test] + async fn mixed_map_authenticates_only_the_gated_relay() { + const TOKEN: &str = "secret-token"; + let (gated_url, _gated) = spawn_gated_relay(TOKEN).await; + + // Spin up a second, fully-open relay to stand in for a public iroh + // relay sharing the same map. + let (_public_map, public_url, _public) = + run_relay_server_with_access(false, AccessConfig::Everyone) + .await + .expect("spawn public relay"); + let public_url = public_url.to_string(); + + let urls = vec![gated_url.clone(), public_url.clone()]; + let mut auths = HashMap::new(); + auths.insert(gated_url, TOKEN.to_string()); + // Public relay intentionally absent from `auths`. + + let ep = build_endpoint(&urls, &auths).await; + tokio::time::timeout(Duration::from_secs(5), ep.online()) + .await + .expect("endpoint should come online via the mixed relay map"); + } } fn encode_endpoint_addr_token(addr: &EndpointAddr) -> String { @@ -1499,6 +1843,7 @@ fn startup_transport_config() -> iroh::endpoint::QuicTransportConfig { async fn bind_mesh_endpoint( secret_key: SecretKey, relay_urls: &[String], + relay_auths: &std::collections::HashMap, quic_bind: QuicBindSelection, ) -> Result { let mut builder = Endpoint::builder(iroh::endpoint::presets::Minimal) @@ -1513,7 +1858,8 @@ async fn bind_mesh_endpoint( tracing::info!("Relay: {:?}", urls); builder = builder.relay_mode(iroh::endpoint::RelayMode::Custom(relay_map_from_urls( &urls, - ))); + relay_auths, + )?)); if let Some(addr) = quic_bind_addr(quic_bind) { tracing::info!("Binding QUIC to {addr}"); @@ -1648,17 +1994,19 @@ fn init_owner_runtime( fn configure_control_relay( mut builder: iroh::endpoint::Builder, relay_urls: Option<&[String]>, -) -> iroh::endpoint::Builder { + relay_auths: &std::collections::HashMap, +) -> Result { if let Some(relay_urls) = relay_urls { let urls = effective_relay_urls(relay_urls); tracing::info!("Owner-control relay: {:?}", urls); builder = builder.relay_mode(iroh::endpoint::RelayMode::Custom(relay_map_from_urls( &urls, - ))); + relay_auths, + )?)); } else { builder = builder.relay_mode(iroh::endpoint::RelayMode::Disabled); } - builder + Ok(builder) } fn default_plugin_event_source(endpoint_id: EndpointId, source_peer_id: &mut String) { @@ -2979,17 +3327,29 @@ impl Node { } } + pub async fn shutdown(&self) { + self.accepting + .1 + .store(true, std::sync::atomic::Ordering::Release); + self.accepting.0.notify_waiters(); + self.shutdown_control_listener().await; + self.endpoint.close().await; + } + pub async fn start( role: NodeRole, - relay_urls: &[String], + relay: RelayConfig<'_>, quic_bind: QuicBindSelection, max_vram_gb: Option, enumerate_host: bool, owner_config: Option, config_path: Option<&std::path::Path>, ) -> Result<(Self, TunnelChannels)> { + let relay_urls = relay.urls; + let relay_auths = relay.auths; let secret_key = startup_secret_key(&role).await?; - let endpoint = bind_mesh_endpoint(secret_key.clone(), relay_urls, quic_bind).await?; + let endpoint = + bind_mesh_endpoint(secret_key.clone(), relay_urls, relay_auths, quic_bind).await?; // Wait briefly for relay connection so the invite token includes the relay URL. // On sinkholed networks this times out and we proceed without relay (direct UDP only). wait_for_endpoint_online( @@ -3119,6 +3479,7 @@ impl Node { .as_ref() .and_then(|config| config.control_advertise_addr), Some(relay_urls), + relay_auths, ) .await?; @@ -3257,12 +3618,18 @@ impl Node { } } + #[cfg(test)] + pub(crate) fn endpoint_is_closed_for_tests(&self) -> bool { + self.endpoint.is_closed() + } + async fn maybe_start_control_listener( &self, secret_key: SecretKey, bind_addr: Option, advertise_addr: Option, relay_urls: Option<&[String]>, + relay_auths: &std::collections::HashMap, ) -> Result<()> { if self.local_verified_owner_id().await.is_none() { return Ok(()); @@ -3272,7 +3639,7 @@ impl Node { .secret_key(secret_key) .alpns(vec![ALPN_CONTROL_V1.to_vec()]) .bind_addr(bind_addr.unwrap_or_else(default_control_bind_addr))?; - builder = configure_control_relay(builder, relay_urls); + builder = configure_control_relay(builder, relay_urls, relay_auths)?; let endpoint = builder.bind().await?; if relay_urls.is_some() { wait_for_endpoint_online( @@ -3425,7 +3792,7 @@ impl Node { } } - /// Enable accepting inbound connections. Call before join() or when ready to participate. + /// Enable accepting inbound connections once ready to participate in the mesh. /// Until this is called, the accept loop blocks waiting. pub fn start_accepting(&self) { self.accepting diff --git a/crates/mesh-llm-host-runtime/src/mesh/tests.rs b/crates/mesh-llm-host-runtime/src/mesh/tests.rs index 0b9e005edd..27953a3660 100644 --- a/crates/mesh-llm-host-runtime/src/mesh/tests.rs +++ b/crates/mesh-llm-host-runtime/src/mesh/tests.rs @@ -11,9 +11,19 @@ use crate::plugin; use crate::proto::node::{GossipFrame, NodeRole, PeerAnnouncement, RouteTableRequest}; use serial_test::serial; use skippy_protocol::proto::stage as skippy_stage_proto; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use tokio::sync::watch; +/// Empty per-relay auth map for tests that don't exercise gated relays. +/// +/// Bound to a local before being passed by reference so the borrow lives +/// for the full duration of any async call — cleaner than `&HashMap::new()` +/// at the call site, and future-proofs the call against the callee +/// holding the reference across an internal `.await`. +fn empty_relay_auths() -> HashMap { + HashMap::new() +} + #[test] fn quic_bind_addr_uses_explicit_port_on_all_platforms() { assert_eq!( @@ -1122,7 +1132,8 @@ async fn control_plane_listener_starts_with_owner() -> anyhow::Result<()> { let (node, secret_key) = Node::new_for_tests_with_secret(super::NodeRole::Worker).await?; *node.owner_summary.lock().await = verified_owner_summary("owner-a"); - node.maybe_start_control_listener(secret_key, None, None, None) + let auths = empty_relay_auths(); + node.maybe_start_control_listener(secret_key, None, None, None, &auths) .await?; let endpoint = node @@ -1147,7 +1158,8 @@ async fn control_plane_listener_uses_explicit_advertised_address() -> anyhow::Re *node.owner_summary.lock().await = verified_owner_summary("owner-a"); let advertised_addr = std::net::SocketAddr::from(([203, 0, 113, 10], 18443)); - node.maybe_start_control_listener(secret_key, None, Some(advertised_addr), None) + let auths = empty_relay_auths(); + node.maybe_start_control_listener(secret_key, None, Some(advertised_addr), None, &auths) .await?; let endpoint = node @@ -1169,11 +1181,13 @@ async fn control_plane_listener_uses_explicit_advertised_address() -> anyhow::Re async fn control_plane_listener_disabled_without_owner() -> anyhow::Result<()> { let (node, secret_key) = Node::new_for_tests_with_secret(super::NodeRole::Worker).await?; + let auths = empty_relay_auths(); node.maybe_start_control_listener( secret_key, Some("127.0.0.1:7447".parse().unwrap()), None, None, + &auths, ) .await?; @@ -1185,7 +1199,8 @@ async fn control_plane_listener_disabled_without_owner() -> anyhow::Result<()> { async fn control_plane_listener_accepts_only_control_alpn() -> anyhow::Result<()> { let (node, secret_key) = Node::new_for_tests_with_secret(super::NodeRole::Worker).await?; *node.owner_summary.lock().await = verified_owner_summary("owner-a"); - node.maybe_start_control_listener(secret_key, None, None, None) + let auths = empty_relay_auths(); + node.maybe_start_control_listener(secret_key, None, None, None, &auths) .await?; let endpoint = Node::decode_invite_token( &node @@ -1215,7 +1230,8 @@ async fn control_plane_listener_accepts_only_control_alpn() -> anyhow::Result<() async fn control_plane_endpoint_not_in_gossip_or_status() -> anyhow::Result<()> { let (node, secret_key) = Node::new_for_tests_with_secret(super::NodeRole::Worker).await?; *node.owner_summary.lock().await = verified_owner_summary("owner-a"); - node.maybe_start_control_listener(secret_key, None, None, None) + let auths = empty_relay_auths(); + node.maybe_start_control_listener(secret_key, None, None, None, &auths) .await?; let control_endpoint = node .control_endpoint() @@ -1249,7 +1265,8 @@ async fn control_plane_endpoint_not_in_gossip_or_status() -> anyhow::Result<()> async fn control_plane_listener_shutdown_stops_listener_task() -> anyhow::Result<()> { let (node, secret_key) = Node::new_for_tests_with_secret(super::NodeRole::Worker).await?; *node.owner_summary.lock().await = verified_owner_summary("owner-a"); - node.maybe_start_control_listener(secret_key, None, None, None) + let auths = empty_relay_auths(); + node.maybe_start_control_listener(secret_key, None, None, None, &auths) .await?; let endpoint = Node::decode_invite_token( &node @@ -5351,7 +5368,8 @@ async fn start_owner_control_test_server( *node.owner_attestation.lock().await = Some(ownership); *node.owner_summary.lock().await = owner_summary; *node.trust_store.lock().await = trust_store; - node.maybe_start_control_listener(secret_key.clone(), None, None, None) + let auths = empty_relay_auths(); + node.maybe_start_control_listener(secret_key.clone(), None, None, None, &auths) .await?; Ok((node, secret_key, config_path)) } diff --git a/crates/mesh-llm-host-runtime/src/runtime/mod.rs b/crates/mesh-llm-host-runtime/src/runtime/mod.rs index 803a5b1564..374488de11 100644 --- a/crates/mesh-llm-host-runtime/src/runtime/mod.rs +++ b/crates/mesh-llm-host-runtime/src/runtime/mod.rs @@ -3183,9 +3183,23 @@ async fn prepare_runtime_startup( } pub(crate) async fn run() -> Result<()> { + run_with_args(std::env::args_os()).await +} + +/// Same as [`run`] but with a caller-supplied argv instead of +/// `std::env::args_os()`. This is the SDK entry point for running a +/// full mesh-llm runtime in-process — the binary's `main()` ultimately +/// calls here with the real process argv, and SDK consumers can call it +/// with an argv they build themselves (e.g. from a typed +/// `crate::host_node::MeshServeSpec`). +pub(crate) async fn run_with_args(args: I) -> Result<()> +where + I: IntoIterator, + S: Into, +{ initialize_runtime_entrypoint()?; - let normalized_args = crate::cli::normalize_runtime_surface_args(std::env::args_os()); + let normalized_args = crate::cli::normalize_runtime_surface_args(args); let mut cli = Cli::parse_from(normalized_args.normalized.clone()); crate::cli::validate_discovery_mode_args(&cli)?; crate::cli::output::OutputManager::init_global( @@ -5199,9 +5213,14 @@ pub(crate) async fn run_plugin_mcp(cli: &Cli) -> Result<()> { let config = plugin::load_config(cli.config.as_deref())?; let owner_config = owner_runtime_config(cli, &config)?; let swarm_capture = configure_plugin_mcp_swarm_capture(cli)?; + let relay_auths: std::collections::HashMap = + cli.parse_relay_auths()?.into_iter().collect(); let (node, _channels) = mesh::Node::start( NodeRole::Client, - &cli.relay, + mesh::RelayConfig { + urls: &cli.relay, + auths: &relay_auths, + }, mesh::QuicBindSelection { ip: cli.bind_ip, port: cli.bind_port, @@ -5371,9 +5390,14 @@ async fn start_run_auto_node_and_plugins( emit_configuration_ui_read_only_hint(); } let max_vram = if cli.client { Some(0.0) } else { cli.max_vram }; + let relay_auths: std::collections::HashMap = + cli.parse_relay_auths()?.into_iter().collect(); let (node, channels) = mesh::Node::start( role, - &cli.relay, + mesh::RelayConfig { + urls: &cli.relay, + auths: &relay_auths, + }, mesh::QuicBindSelection { ip: cli.bind_ip, port: cli.bind_port, diff --git a/docs/SDK.md b/docs/SDK.md index b21f8733a2..4e73f87545 100644 --- a/docs/SDK.md +++ b/docs/SDK.md @@ -223,6 +223,74 @@ If no controller is attached, `serving.load()` returns an unsupported error. This is intentional: `mesh-llm-api-server` is platform-neutral and does not silently choose a native backend. +### Run the full mesh-llm runtime from Rust (`host-runtime` feature) + +`MeshNode::builder()` is the fine-grained surface: assemble a node piece +by piece (identity, invite, serving controller, OpenAI port, ...) and +drive it yourself. + +When you instead want to run **exactly what `mesh-llm serve` / +`mesh-llm client` does** — same code path, same defaults, same +behaviour — use `run_serve(spec)`. This is the in-process equivalent of +spawning the binary: auto-discovery, election, tunnel manager, OpenAI +HTTP proxy, management console, local model serving (when configured), +plugin host, all driven by the same `runtime::run_with_args` entry +point the binary calls. + +Enable the `host-runtime` feature on `mesh-llm-api-server`. This pulls +in `mesh-llm-host-runtime` and its transitive deps (skippy, llama.cpp +link path, ...); it is off by default to keep the SDK lean for +client-only consumers. + +```toml +[dependencies] +mesh-llm-api-server = { version = "0.66.0", features = ["host-runtime"] } +``` + +```rust +use mesh_llm_api_server::{run_serve, MeshServeSpec}; +use std::collections::HashMap; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let mut relay_auths = HashMap::new(); + relay_auths.insert( + "https://gated.example/".to_string(), + "".to_string(), + ); + + run_serve(MeshServeSpec { + // Same flags `mesh-llm serve` / `mesh-llm client` accept. + client: true, // false (default) = serve role + auto: true, // == --auto + relays: vec!["https://gated.example/".into()], + relay_auths, // == --relay-auth URL=TOKEN + port: Some(9337), // OpenAI HTTP proxy port + console_port: Some(3131), // management API / web console + headless: true, // skip embedded web UI + max_vram_gb: Some(0.0), // client-only, no VRAM advert + ..MeshServeSpec::default() + }) + .await?; + + Ok(()) +} +``` + +The future blocks until the runtime exits (signal, internal shutdown, +or fatal error). The runtime is not currently `Send`-clean; if you +need to drive concurrent work alongside it, run on a +`tokio::task::LocalSet` rather than `tokio::spawn`. + +Full `MeshServeSpec` covers every meaningful `mesh-llm` flag: +`client`, `auto`, `publish`, `mesh_name`, `region`, `display_name`, +`join`, `discover`, `models`, `ggufs`, `mmproj`, `port`, +`console_port`, `headless`, `blackboard`, `relays`, `relay_auths`, +`nostr_relays`, `bind_port`, `bind_ip`, `listen_all`, `max_vram_gb`, +`no_enumerate_host`, `config`, `owner_key`, `owner_required`, +`node_label`, `trust_owners`, `debug`, plus an `extra_args` escape +hatch for flags not yet typed. + ## Swift Usage Configure a native runtime before local serving: