From 78b2da3fc429f415cf231d826af682490029f9d1 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Wed, 8 Jul 2026 13:14:49 +0400 Subject: [PATCH] stash --- .../mvp-system/src/bin/mvp_one_node_chat.rs | 358 ++++-- crates/mvp-system/src/bin/orchestrator.rs | 1084 +++++++++++++---- crates/mvp-system/src/bin/worker_node.rs | 16 +- crates/mvp-system/src/config.rs | 133 +- crates/mvp-system/src/relay_provisioning.rs | 18 +- 5 files changed, 1271 insertions(+), 338 deletions(-) diff --git a/crates/mvp-system/src/bin/mvp_one_node_chat.rs b/crates/mvp-system/src/bin/mvp_one_node_chat.rs index de803d2..5b442e5 100644 --- a/crates/mvp-system/src/bin/mvp_one_node_chat.rs +++ b/crates/mvp-system/src/bin/mvp_one_node_chat.rs @@ -24,7 +24,6 @@ const DEFAULT_RPC_ADDR: &str = "127.0.0.1:19777"; const DEFAULT_NODE_IMAGE: &str = "swactor-mvp-node:latest"; const BASE_NODE_IMAGE: &str = "swactor-mvp-node-base:cuda12.6"; const MVP_RUNTIME_CONFIG_ENV: &str = "MVP_RUNTIME_CONFIG"; -const CACHED_MODEL_HOST_ENV: &str = "MVP_CACHED_MODEL_HOST_PATH"; const DEFAULT_CACHED_MODEL_FILE: &str = "Llama-3.2-1B-Instruct-Q4_K_M.gguf"; const REPO_MODEL_CACHE_DIR: &str = ".model-cache"; const DEFAULT_MAX_TOKENS: u32 = 64; @@ -138,8 +137,8 @@ impl Config { I: IntoIterator, { let args = ParsedArgs::parse(provided_args)?; - let loaded = chat_config::Config::load(args.config_path.as_deref())?; - let file = loaded.config; + let loaded = chat_config::TomlConfigOverlay::load(args.config_path.as_deref())?; + let toml = loaded.overlay; let profile = if args.vastai { RuntimeConfigProfile::Deploy } else { @@ -150,59 +149,59 @@ impl Config { } else { provider_from_env(profile)? }; - let relay_mode = relay_mode_from_sources(file.relay.mode.as_deref())?; + let relay_mode = relay_mode_from_sources(toml.relay.mode.as_deref())?; let relay_url = first_non_empty([ env_optional("MVP_IROH_RELAY_URL"), env_optional("SWACTOR_IROH_RELAY_URL"), - file.relay.url.clone(), + toml.relay.url.clone(), ]); let node_image = first_non_empty([ args.image, env_optional("MVP_NODE_IMAGE"), if args.vastai { - file.vastai.image.clone() + toml.vastai.image.clone() } else { None }, - file.image.node.clone(), + toml.image.node.clone(), Some(DEFAULT_NODE_IMAGE.to_owned()), ]) .expect("default image is non-empty"); let max_tokens = args .max_tokens .or(env_u32_optional("MVP_PROMPT_MAX_TOKENS")?) - .or(file.prompt.max_tokens) + .or(toml.prompt.max_tokens) .unwrap_or(DEFAULT_MAX_TOKENS); let dashboard = args .dashboard .or(env_bool_optional("MVP_DASHBOARD")?) - .or(file.prompt.dashboard) + .or(toml.prompt.dashboard) .unwrap_or(true); let build_image = args .build_image .or(env_bool_optional("MVP_BUILD_NODE_IMAGE")?) - .or(file.image.build) + .or(toml.image.build) .unwrap_or(true); let push_image = args .push_image .or(env_bool_optional("MVP_PUSH_NODE_IMAGE")?) - .or(file.image.push) + .or(toml.image.push) .unwrap_or(false); let force_image_refresh = args .force_image_refresh .or(env_bool_optional("MVP_FORCE_NODE_IMAGE_REFRESH")?) - .or(file.image.force_refresh) + .or(toml.image.force_refresh) .unwrap_or(false); let image_tag = first_non_empty([ args.image_tag, env_optional("MVP_NODE_IMAGE_TAG"), - file.image.tag.clone(), + toml.image.tag.clone(), ]); let rpc_addr = first_non_empty([ args.rpc_addr, env_optional("MVP_PROMPT_RPC_ADDR"), env_optional("MVP_PROMPT_RPC_BIND"), - file.prompt.rpc_addr.clone(), + toml.prompt.rpc_addr.clone(), Some(DEFAULT_RPC_ADDR.to_owned()), ]) .expect("default RPC address is non-empty"); @@ -211,23 +210,28 @@ impl Config { return Err("--dump-logs cannot be combined with --datastream-frame-log; use one datastream log destination".to_owned()); } (true, None) => Some(PathBuf::from("mvp-chat.log")), - (false, explicit) => { - explicit.or_else(|| env_optional("MVP_DATASTREAM_FRAME_LOG").map(PathBuf::from)) - } + (false, explicit) => explicit + .or_else(|| env_optional("MVP_DATASTREAM_FRAME_LOG").map(PathBuf::from)) + .or_else(|| toml.observability.datastream_frame_log.clone().map(PathBuf::from)), }; - let model_id = first_non_empty([env_optional("MVP_MODEL_ID"), file.model.id.clone()]); + let model_id = first_non_empty([env_optional("MVP_MODEL_ID"), toml.model.id.clone()]); let gguf_repo = - first_non_empty([env_optional("MVP_GGUF_REPO"), file.model.gguf_repo.clone()]); + first_non_empty([env_optional("MVP_GGUF_REPO"), toml.model.gguf_repo.clone()]); let gguf_file = - first_non_empty([env_optional("MVP_GGUF_FILE"), file.model.gguf_file.clone()]); + first_non_empty([env_optional("MVP_GGUF_FILE"), toml.model.gguf_file.clone()]); let gguf_revision = first_non_empty([ env_optional("MVP_GGUF_REVISION"), - file.model.gguf_revision.clone(), + toml.model.gguf_revision.clone(), ]); - let max_context = env_u32_optional("MVP_MAX_CONTEXT")?.or(file.model.max_context); + let max_context = env_u32_optional("MVP_MAX_CONTEXT")?.or(toml.model.max_context); + let cached_model = match (args.cached_model, toml.docker.cached_model_host_path.clone()) { + (Some(cached_model), _) => Some(cached_model), + (None, Some(path)) => Some(CachedModelConfig::from_arg(Some(path))?), + (None, None) => None, + }; let vastai = if args.vastai { Some(resolve_vastai_config( - &file.vastai, + &toml.vastai, &node_image, relay_url.clone(), )?) @@ -250,7 +254,7 @@ impl Config { image_tag, push_image, force_image_refresh, - cached_model: args.cached_model, + cached_model, datastream_frame_log, vastai_yes: args.vastai_yes, model_id, @@ -261,6 +265,107 @@ impl Config { vastai, }) } + + fn orchestrator_cli_args(&self, image_ref: &str) -> Vec { + let mut args = vec![ + "--runtime-config".to_owned(), + self.config_profile.as_str().to_owned(), + "--provider".to_owned(), + self.provider.as_str().to_owned(), + "--image".to_owned(), + image_ref.to_owned(), + "--rpc-bind".to_owned(), + self.rpc_addr.clone(), + "--max-tokens".to_owned(), + self.max_tokens.to_string(), + if self.dashboard { + "--dashboard".to_owned() + } else { + "--no-dashboard".to_owned() + }, + ]; + if let Some(relay_url) = &self.relay_url { + args.extend(["--relay-url".to_owned(), relay_url.clone()]); + } + args.extend([ + "--relay-mode".to_owned(), + relay_mode_env_value(&self.relay_mode).to_owned(), + ]); + if let Some(model_id) = &self.model_id { + args.extend(["--model-id".to_owned(), model_id.clone()]); + } + if let Some(repo) = &self.gguf_repo { + args.extend(["--gguf-repo".to_owned(), repo.clone()]); + } + if let Some(file) = &self.gguf_file { + args.extend(["--gguf-file".to_owned(), file.clone()]); + } + if let Some(revision) = &self.gguf_revision { + args.extend(["--gguf-revision".to_owned(), revision.clone()]); + } + if let Some(max_context) = self.max_context { + args.extend(["--max-context".to_owned(), max_context.to_string()]); + } + if let Some(cached_model) = &self.cached_model { + args.extend([ + "--cached-model-host-path".to_owned(), + cached_model.host_path.to_string_lossy().to_string(), + ]); + } + if let Some(path) = &self.datastream_frame_log { + args.extend([ + "--datastream-frame-log".to_owned(), + path.to_string_lossy().to_string(), + ]); + } + if let Some(vastai) = &self.vastai { + args.extend([ + "--vastai-api-key".to_owned(), + vastai.api_key.clone(), + "--vastai-bootstrap-command".to_owned(), + vastai.bootstrap_command.clone(), + "--no-vastai-confirm-lease".to_owned(), + ]); + if let Some(disk_gb) = vastai.disk_gb { + args.extend(["--vastai-disk-gb".to_owned(), disk_gb.to_string()]); + } + if let Some(gpu_name) = &vastai.gpu_name { + args.extend(["--vastai-gpu-name".to_owned(), gpu_name.clone()]); + } + if let Some(min_gpu_ram_mb) = vastai.min_gpu_ram_mb { + args.extend([ + "--vastai-min-gpu-ram-mb".to_owned(), + min_gpu_ram_mb.to_string(), + ]); + } + if let Some(min_down_mbps) = vastai.min_down_mbps { + args.extend(["--vastai-min-down-mbps".to_owned(), min_down_mbps.to_string()]); + } + if let Some(min_up_mbps) = vastai.min_up_mbps { + args.extend(["--vastai-min-up-mbps".to_owned(), min_up_mbps.to_string()]); + } + if let Some(min_reliability) = vastai.min_reliability { + args.extend([ + "--vastai-min-reliability".to_owned(), + min_reliability.to_string(), + ]); + } + if let Some(require_verified) = vastai.require_verified { + args.push(if require_verified { + "--vastai-require-verified".to_owned() + } else { + "--no-vastai-require-verified".to_owned() + }); + } + if let Some(onstart) = &vastai.onstart { + args.extend(["--vastai-onstart".to_owned(), onstart.clone()]); + } + if let Some(ssh_identity) = &vastai.ssh_identity { + args.extend(["--vastai-ssh-identity".to_owned(), ssh_identity.clone()]); + } + } + args + } } #[derive(Default)] @@ -660,82 +765,13 @@ struct OrchChild { impl OrchChild { fn spawn(config: &Config, image_ref: &str) -> Result { let mut command = Command::new(&config.orch_bin); + let mut orch_args = config.orchestrator_cli_args(image_ref); + orch_args.extend(config.orch_args.clone()); command - .args(&config.orch_args) - .env("MVP_NODE_PROVIDER", config.provider.as_str()) - .env(MVP_RUNTIME_CONFIG_ENV, config.config_profile.as_str()) - .env("MVP_NODE_IMAGE", image_ref) - .env( - "MVP_IROH_RELAY_MODE", - relay_mode_env_value(&config.relay_mode), - ) - .env("MVP_PROMPT_RPC_BIND", &config.rpc_addr) - .env("MVP_PROMPT_RPC_ADDR", &config.rpc_addr) - .env("MVP_PROMPT_MAX_TOKENS", config.max_tokens.to_string()) - .env("MVP_DASHBOARD", if config.dashboard { "1" } else { "0" }) + .args(&orch_args) .stdin(Stdio::piped()) .stdout(Stdio::null()) .stderr(Stdio::null()); - if let Some(relay_url) = &config.relay_url { - command.env("MVP_IROH_RELAY_URL", relay_url); - } - if let Some(vastai) = &config.vastai { - command - .env("MVP_VASTAI_API_KEY", &vastai.api_key) - .env("MVP_VASTAI_BOOTSTRAP_COMMAND", &vastai.bootstrap_command) - .env("MVP_VASTAI_CONFIRM_LEASE", "0"); - if let Some(disk_gb) = vastai.disk_gb { - command.env("MVP_VASTAI_DISK_GB", disk_gb.to_string()); - } - if let Some(gpu_name) = &vastai.gpu_name { - command.env("MVP_VASTAI_GPU_NAME", gpu_name); - } - if let Some(min_gpu_ram_mb) = vastai.min_gpu_ram_mb { - command.env("MVP_VASTAI_MIN_GPU_RAM_MB", min_gpu_ram_mb.to_string()); - } - if let Some(min_down_mbps) = vastai.min_down_mbps { - command.env("MVP_VASTAI_MIN_DOWN_MBPS", min_down_mbps.to_string()); - } - if let Some(min_up_mbps) = vastai.min_up_mbps { - command.env("MVP_VASTAI_MIN_UP_MBPS", min_up_mbps.to_string()); - } - if let Some(min_reliability) = vastai.min_reliability { - command.env("MVP_VASTAI_MIN_RELIABILITY", min_reliability.to_string()); - } - if let Some(require_verified) = vastai.require_verified { - command.env( - "MVP_VASTAI_REQUIRE_VERIFIED", - if require_verified { "1" } else { "0" }, - ); - } - if let Some(onstart) = &vastai.onstart { - command.env("MVP_VASTAI_ONSTART", onstart); - } - if let Some(ssh_identity) = &vastai.ssh_identity { - command.env("MVP_VASTAI_SSH_IDENTITY", ssh_identity); - } - } - if let Some(model_id) = &config.model_id { - command.env("MVP_MODEL_ID", model_id); - } - if let Some(repo) = &config.gguf_repo { - command.env("MVP_GGUF_REPO", repo); - } - if let Some(file) = &config.gguf_file { - command.env("MVP_GGUF_FILE", file); - } - if let Some(revision) = &config.gguf_revision { - command.env("MVP_GGUF_REVISION", revision); - } - if let Some(max_context) = config.max_context { - command.env("MVP_MAX_CONTEXT", max_context.to_string()); - } - if let Some(cached_model) = &config.cached_model { - command.env(CACHED_MODEL_HOST_ENV, &cached_model.host_path); - } - if let Some(path) = &config.datastream_frame_log { - command.env("MVP_DATASTREAM_FRAME_LOG", path); - } #[cfg(target_os = "linux")] unsafe { command.pre_exec(|| { @@ -1484,6 +1520,136 @@ mod tests { } } + + fn assert_arg_value(args: &[String], flag: &str, expected: &str) { + let flag_index = args + .iter() + .position(|arg| arg == flag) + .unwrap_or_else(|| panic!("missing CLI flag {flag}; args={args:?}")); + assert_eq!( + args.get(flag_index + 1).map(String::as_str), + Some(expected), + "unexpected value for CLI flag {flag}; args={args:?}" + ); + } + + fn assert_flag(args: &[String], flag: &str) { + assert!( + args.iter().any(|arg| arg == flag), + "missing CLI flag {flag}; args={args:?}" + ); + } + + #[test] + fn orchestrator_cli_args_cover_wrapper_launch_config() { + let config = Config { + orch_bin: PathBuf::from("mvp-orchestrator"), + orch_args: Vec::new(), + rpc_addr: "127.0.0.1:20123".to_owned(), + node_image: "docker.io/example/config-node:ignored".to_owned(), + config_profile: RuntimeConfigProfile::Local, + provider: ProviderKind::Docker, + relay_mode: iroh::RelayMode::Default, + relay_url: Some("https://relay.example.com".to_owned()), + max_tokens: 37, + dashboard: false, + build_image: false, + image_tag: None, + push_image: false, + force_image_refresh: false, + cached_model: Some(CachedModelConfig { + host_path: PathBuf::from("/var/cache/swactor/model.gguf"), + display_path: PathBuf::from("model.gguf"), + }), + datastream_frame_log: Some(PathBuf::from("/tmp/mvp-chat-frames.jsonl")), + vastai_yes: false, + vastai: None, + model_id: Some("wrapper-model".to_owned()), + gguf_repo: Some("example/wrapper-repo".to_owned()), + gguf_file: Some("wrapper-model.gguf".to_owned()), + gguf_revision: None, + max_context: Some(768), + }; + + let args = config.orchestrator_cli_args("docker.io/example/prepared-node:latest"); + + assert_arg_value(&args, "--runtime-config", "local"); + assert_arg_value(&args, "--provider", "docker"); + assert_arg_value(&args, "--image", "docker.io/example/prepared-node:latest"); + assert_arg_value(&args, "--rpc-bind", "127.0.0.1:20123"); + assert_arg_value(&args, "--max-tokens", "37"); + assert_flag(&args, "--no-dashboard"); + assert_arg_value(&args, "--relay-url", "https://relay.example.com"); + assert_arg_value(&args, "--model-id", "wrapper-model"); + assert_arg_value(&args, "--gguf-repo", "example/wrapper-repo"); + assert_arg_value(&args, "--gguf-file", "wrapper-model.gguf"); + assert_arg_value(&args, "--max-context", "768"); + assert_arg_value( + &args, + "--cached-model-host-path", + "/var/cache/swactor/model.gguf", + ); + assert_arg_value( + &args, + "--datastream-frame-log", + "/tmp/mvp-chat-frames.jsonl", + ); + } + + #[test] + fn orchestrator_cli_args_cover_vastai_config() { + let mut vastai = valid_vastai_config(); + vastai.onstart = Some("echo preparing vastai node".to_owned()); + let config = Config { + orch_bin: PathBuf::from("mvp-orchestrator"), + orch_args: Vec::new(), + rpc_addr: DEFAULT_RPC_ADDR.to_owned(), + node_image: "ghcr.io/swactor/mvp-node:latest".to_owned(), + config_profile: RuntimeConfigProfile::Deploy, + provider: ProviderKind::VastAi, + relay_mode: iroh::RelayMode::Default, + relay_url: Some(vastai.relay_url.clone()), + max_tokens: 128, + dashboard: false, + build_image: false, + image_tag: Some("trial".to_owned()), + push_image: true, + force_image_refresh: false, + cached_model: None, + datastream_frame_log: None, + vastai_yes: true, + vastai: Some(vastai), + model_id: None, + gguf_repo: None, + gguf_file: None, + gguf_revision: None, + max_context: None, + }; + + let args = config.orchestrator_cli_args("ghcr.io/swactor/mvp-node:latest"); + + assert_arg_value(&args, "--vastai-api-key", "vast-key"); + assert_arg_value( + &args, + "--vastai-bootstrap-command", + "/usr/local/bin/mvp-node", + ); + assert_flag(&args, "--no-vastai-confirm-lease"); + assert_arg_value(&args, "--vastai-disk-gb", "80"); + assert_arg_value(&args, "--vastai-gpu-name", "RTX 4090"); + assert_arg_value(&args, "--vastai-min-gpu-ram-mb", "16000"); + assert_arg_value(&args, "--vastai-min-down-mbps", "100"); + assert_arg_value(&args, "--vastai-min-up-mbps", "25"); + assert_arg_value(&args, "--vastai-min-reliability", "0.98"); + assert_flag(&args, "--vastai-require-verified"); + assert_arg_value(&args, "--vastai-onstart", "echo preparing vastai node"); + assert_arg_value( + &args, + "--vastai-ssh-identity", + "~/.ssh/swactor_vastai_ed25519", + ); + } + #[test] fn approval_parser_accepts_only_y_or_yes() { for (input, expected) in [ diff --git a/crates/mvp-system/src/bin/orchestrator.rs b/crates/mvp-system/src/bin/orchestrator.rs index be7baf2..1828f17 100644 --- a/crates/mvp-system/src/bin/orchestrator.rs +++ b/crates/mvp-system/src/bin/orchestrator.rs @@ -17,6 +17,7 @@ use mvp_system::actors::node_agent::{NodeAgentMsg, StageProvisionWire}; use mvp_system::actors::register_mvp_actor_codecs; #[cfg(feature = "local-e2e")] use mvp_system::dashboard_view::MvpClusterDashboardView; +use mvp_system::config::{DEFAULT_CONFIG_PATH, TomlConfigOverlay}; use mvp_system::distribution_stack::DistributionRuntimeStack; use mvp_system::node_provisioning::ProviderKind; use mvp_system::prompt_rpc::{PromptEvent, SubmitPrompt, read_submit_prompt, write_json_line}; @@ -26,9 +27,10 @@ use mvp_system::provisioning::{ ProvisionPlugin, }; #[cfg(test)] -use mvp_system::relay_provisioning::SWACTOR_IROH_RELAY_URL_ENV; +use mvp_system::relay_provisioning::relay_runtime_config_from_env; use mvp_system::relay_provisioning::{ - MVP_IROH_RELAY_URL_ENV, RelayRuntimeConfig, relay_mode_env_value, relay_runtime_config_from_env, + MVP_IROH_RELAY_URL_ENV, RelayRuntimeConfig, SWACTOR_IROH_RELAY_URL_ENV, relay_mode_env_value, + relay_runtime_config_from_settings, }; use mvp_system::run_plan::{GgufSource, TokenizerSource}; use mvp_system::telemetry::{ @@ -68,7 +70,7 @@ fn main() -> ExitCode { } fn run() -> Result<(), String> { - let mut config = Config::from_env_and_args()?; + let mut config = Config::from_defaults_toml_env_args(std::env::args().skip(1))?; config.prepare_vastai_ssh_key()?; let orch_stdio_rx = install_orch_stdio_capture()?; let mut orch_datastream = @@ -225,7 +227,7 @@ fn run() -> Result<(), String> { "ready", json!({"actor":datastream_sink,"channel":"local_mpsc"}), ); - let dashboard = DashboardSupport::start_from_env()?; + let dashboard = DashboardSupport::start(config.dashboard)?; orch_datastream.emit_bootstrap( dashboard.as_ref(), config.run_id, @@ -653,40 +655,95 @@ struct VastAiRuntimeConfig { } impl VastAiRuntimeConfig { - fn from_env() -> Result { + fn from_builder(builder: &ConfigBuilder) -> Result { let mut provisioning = VastAiProvisioningConfig::default(); - provisioning.disk_gb = env_u32("MVP_VASTAI_DISK_GB", provisioning.disk_gb)?; - provisioning.ssh_user = env_string("MVP_VASTAI_SSH_USER", &provisioning.ssh_user); - provisioning.confirm_lease = - env_bool("MVP_VASTAI_CONFIRM_LEASE", provisioning.confirm_lease)?; - provisioning.onstart = env_optional("MVP_VASTAI_ONSTART"); - provisioning.selection.gpu_name = env_optional("MVP_VASTAI_GPU_NAME"); - if let Some(min_gpu_ram_mb) = env_optional_u64("MVP_VASTAI_MIN_GPU_RAM_MB")? { + let disk_gb = builder + .vastai_disk_gb_raw + .as_ref() + .map(|value| ConfigBuilder::parse_value("MVP_VASTAI_DISK_GB", value)) + .transpose()? + .or(builder.vastai_disk_gb); + if let Some(disk_gb) = disk_gb { + provisioning.disk_gb = disk_gb; + } + if let Some(ssh_user) = &builder.vastai_ssh_user { + provisioning.ssh_user = ssh_user.clone(); + } + let confirm_lease = builder + .vastai_confirm_lease_raw + .as_ref() + .map(|value| ConfigBuilder::parse_bool("MVP_VASTAI_CONFIRM_LEASE", value)) + .transpose()? + .or(builder.vastai_confirm_lease); + if let Some(confirm_lease) = confirm_lease { + provisioning.confirm_lease = confirm_lease; + } + provisioning.onstart = builder.vastai_onstart.clone(); + provisioning.selection.gpu_name = builder.vastai_gpu_name.clone(); + let min_gpu_ram_mb = builder + .vastai_min_gpu_ram_mb_raw + .as_ref() + .map(|value| ConfigBuilder::parse_value("MVP_VASTAI_MIN_GPU_RAM_MB", value)) + .transpose()? + .or(builder.vastai_min_gpu_ram_mb); + if let Some(min_gpu_ram_mb) = min_gpu_ram_mb { provisioning.selection.min_gpu_ram_mb = Some(min_gpu_ram_mb); } - if let Some(min_down_mbps) = env_optional_f64("MVP_VASTAI_MIN_DOWN_MBPS")? { + let min_down_mbps = builder + .vastai_min_down_mbps_raw + .as_ref() + .map(|value| ConfigBuilder::parse_value("MVP_VASTAI_MIN_DOWN_MBPS", value)) + .transpose()? + .or(builder.vastai_min_down_mbps); + if let Some(min_down_mbps) = min_down_mbps { provisioning.selection.min_down_mbps = min_down_mbps; } - if let Some(min_up_mbps) = env_optional_f64("MVP_VASTAI_MIN_UP_MBPS")? { + let min_up_mbps = builder + .vastai_min_up_mbps_raw + .as_ref() + .map(|value| ConfigBuilder::parse_value("MVP_VASTAI_MIN_UP_MBPS", value)) + .transpose()? + .or(builder.vastai_min_up_mbps); + if let Some(min_up_mbps) = min_up_mbps { provisioning.selection.min_up_mbps = Some(min_up_mbps); } - if let Some(min_reliability) = env_optional_f64("MVP_VASTAI_MIN_RELIABILITY")? { + let min_reliability = builder + .vastai_min_reliability_raw + .as_ref() + .map(|value| ConfigBuilder::parse_value("MVP_VASTAI_MIN_RELIABILITY", value)) + .transpose()? + .or(builder.vastai_min_reliability); + if let Some(min_reliability) = min_reliability { provisioning.selection.min_reliability = min_reliability; } - provisioning.selection.require_verified = env_bool( - "MVP_VASTAI_REQUIRE_VERIFIED", - provisioning.selection.require_verified, - )?; - if let Some(poll_interval_secs) = env_optional_u64("MVP_VASTAI_POLL_INTERVAL_SECS")? { + let require_verified = builder + .vastai_require_verified_raw + .as_ref() + .map(|value| ConfigBuilder::parse_bool("MVP_VASTAI_REQUIRE_VERIFIED", value)) + .transpose()? + .or(builder.vastai_require_verified); + if let Some(require_verified) = require_verified { + provisioning.selection.require_verified = require_verified; + } + let poll_interval_secs = builder + .vastai_poll_interval_secs_raw + .as_ref() + .map(|value| ConfigBuilder::parse_value("MVP_VASTAI_POLL_INTERVAL_SECS", value)) + .transpose()? + .or(builder.vastai_poll_interval_secs); + if let Some(poll_interval_secs) = poll_interval_secs { provisioning.lifecycle.poll_interval = Duration::from_secs(poll_interval_secs); } + let ssh_identity = builder + .vastai_ssh_identity_raw + .as_ref() + .map(|value| expand_home_path(value)) + .transpose()?; Ok(Self { - api_key: env_optional("MVP_VASTAI_API_KEY").or_else(|| env_optional("VASTAI_API_KEY")), + api_key: builder.vastai_api_key.clone(), provisioning, - bootstrap_command: env_optional("MVP_VASTAI_BOOTSTRAP_COMMAND"), - ssh_identity: env_optional("MVP_VASTAI_SSH_IDENTITY") - .map(|value| expand_home_path(&value)) - .transpose()?, + bootstrap_command: builder.vastai_bootstrap_command.clone(), + ssh_identity, ssh_public_key: None, ssh_public_fingerprint: None, }) @@ -719,16 +776,17 @@ enum RuntimeConfigProfile { } impl RuntimeConfigProfile { - fn from_env() -> Result { - match env_optional(MVP_RUNTIME_CONFIG_ENV).as_deref() { - None | Some("local") => Ok(Self::Local), - Some("deploy") => Ok(Self::Deploy), - Some(other) => Err(format!( + fn parse(value: &str) -> Result { + match value.trim().to_ascii_lowercase().as_str() { + "local" => Ok(Self::Local), + "deploy" => Ok(Self::Deploy), + other => Err(format!( "unsupported {MVP_RUNTIME_CONFIG_ENV}={other:?}; use local or deploy" )), } } + fn as_str(self) -> &'static str { match self { Self::Local => "local", @@ -751,16 +809,12 @@ struct CachedModelConfig { } impl CachedModelConfig { - fn from_env(provider: ProviderKind) -> Result, String> { - let Some(raw_host_path) = env_optional(CACHED_MODEL_HOST_ENV) else { - return Ok(None); - }; + fn from_host_path(provider: ProviderKind, requested: PathBuf) -> Result { if provider != ProviderKind::Docker { return Err(format!( "{CACHED_MODEL_HOST_ENV} is a host-local cache path and requires provider=docker" )); } - let requested = PathBuf::from(&raw_host_path); let host_path = requested.canonicalize().map_err(|e| { format!( "resolve {CACHED_MODEL_HOST_ENV} path {}: {e}", @@ -774,10 +828,10 @@ impl CachedModelConfig { )); } let container_path = cached_model_container_path(&host_path)?; - Ok(Some(Self { + Ok(Self { host_path, container_path, - })) + }) } fn datastream_detail(&self) -> Value { @@ -818,118 +872,613 @@ struct Config { gguf_source: GgufSource, tokenizer: TokenizerSource, default_max_tokens: u32, + dashboard: bool, + max_context: Option, relay: RelayRuntimeConfig, vastai: Option, cached_model: Option, datastream_frame_log: Option, } -impl Config { - fn from_env_and_args() -> Result { - Self::from_env_and_args_iter(std::env::args().skip(1)) +#[derive(Clone)] +struct ConfigBuilder { + config_profile: RuntimeConfigProfile, + provider: Option, + image: String, + toml_vastai_image: Option, + image_overridden_after_toml: bool, + docker_gpus: String, + rpc_bind: String, + rpc_bind_label: &'static str, + run_id: u64, + node_id: u64, + stage_index: u32, + layer_end_exclusive: u32, + model_id: String, + gguf_source: GgufSource, + tokenizer: TokenizerSource, + default_max_tokens: u32, + dashboard: bool, + max_context: Option, + relay_mode: Option, + relay_url: Option, + vastai_api_key: Option, + vastai_bootstrap_command: Option, + vastai_disk_gb: Option, + vastai_disk_gb_raw: Option, + vastai_ssh_user: Option, + vastai_confirm_lease: Option, + vastai_confirm_lease_raw: Option, + vastai_onstart: Option, + vastai_ssh_identity_raw: Option, + vastai_gpu_name: Option, + vastai_min_gpu_ram_mb: Option, + vastai_min_gpu_ram_mb_raw: Option, + vastai_min_down_mbps: Option, + vastai_min_down_mbps_raw: Option, + vastai_min_up_mbps: Option, + vastai_min_up_mbps_raw: Option, + vastai_min_reliability: Option, + vastai_min_reliability_raw: Option, + vastai_require_verified: Option, + vastai_require_verified_raw: Option, + vastai_poll_interval_secs: Option, + vastai_poll_interval_secs_raw: Option, + cached_model_host_path: Option, + datastream_frame_log: Option, +} + +impl ConfigBuilder { + fn hardcoded_defaults() -> Self { + Self { + config_profile: RuntimeConfigProfile::Local, + provider: None, + image: DEFAULT_IMAGE.to_owned(), + toml_vastai_image: None, + image_overridden_after_toml: false, + docker_gpus: "all".to_owned(), + rpc_bind: DEFAULT_RPC_BIND.to_owned(), + rpc_bind_label: "MVP_PROMPT_RPC_BIND", + run_id: 1, + node_id: 1, + stage_index: 0, + layer_end_exclusive: 16, + model_id: DEFAULT_MODEL_ID.to_owned(), + gguf_source: GgufSource::HuggingFaceGguf { + repo: DEFAULT_HF_REPO.to_owned(), + file: DEFAULT_HF_FILE.to_owned(), + revision: None, + }, + tokenizer: TokenizerSource::EmbeddedGguf, + default_max_tokens: DEFAULT_MAX_TOKENS, + dashboard: false, + max_context: None, + relay_mode: None, + relay_url: None, + vastai_api_key: None, + vastai_bootstrap_command: None, + vastai_disk_gb: None, + vastai_disk_gb_raw: None, + vastai_ssh_user: None, + vastai_confirm_lease: None, + vastai_confirm_lease_raw: None, + vastai_onstart: None, + vastai_ssh_identity_raw: None, + vastai_gpu_name: None, + vastai_min_gpu_ram_mb: None, + vastai_min_gpu_ram_mb_raw: None, + vastai_min_down_mbps: None, + vastai_min_down_mbps_raw: None, + vastai_min_up_mbps: None, + vastai_min_up_mbps_raw: None, + vastai_min_reliability: None, + vastai_min_reliability_raw: None, + vastai_require_verified: None, + vastai_require_verified_raw: None, + vastai_poll_interval_secs: None, + vastai_poll_interval_secs_raw: None, + cached_model_host_path: None, + datastream_frame_log: None, + } } - fn from_env_and_args_iter(args: impl IntoIterator) -> Result { - let mut args = args.into_iter(); - let config_profile = RuntimeConfigProfile::from_env()?; - let provider = provider_from_env(config_profile)?; - let cached_model = CachedModelConfig::from_env(provider)?; - let vastai = if provider == ProviderKind::VastAi { - Some(VastAiRuntimeConfig::from_env()?) - } else { - None - }; - let gguf_source = cached_model - .as_ref() - .map(|cached_model| GgufSource::LocalPath(cached_model.container_path.clone())) - .unwrap_or_else(gguf_source_from_env); - let run_id = env_u64("MVP_RUN_ID", 1)?; - let relay = relay_runtime_config_from_env(run_id)?; - let mut config = Self { - config_profile, - image: env_string("MVP_NODE_IMAGE", DEFAULT_IMAGE), - docker_gpus: env_string("MVP_DOCKER_GPUS", "all"), - provider, - rpc_bind: env_string("MVP_PROMPT_RPC_BIND", DEFAULT_RPC_BIND) - .parse() - .map_err(|e| format!("invalid MVP_PROMPT_RPC_BIND: {e}"))?, - run_id, - node_id: env_u64("MVP_LOGICAL_NODE_ID", 1)?, - stage_index: env_u32("MVP_STAGE_INDEX", 0)?, - layer_end_exclusive: env_u32("MVP_LAYER_END_EXCLUSIVE", 16)?, - model_id: env_string("MVP_MODEL_ID", DEFAULT_MODEL_ID), - cached_model, - gguf_source, - tokenizer: tokenizer_from_env(), - default_max_tokens: env_u32("MVP_PROMPT_MAX_TOKENS", DEFAULT_MAX_TOKENS)?, - relay, - vastai, - datastream_frame_log: env_optional(DATASTREAM_FRAME_LOG_ENV).map(PathBuf::from), - }; + fn overlay_toml(mut self, overlay: TomlConfigOverlay) -> Result { + if let Some(profile) = overlay.runtime.profile { + self.config_profile = RuntimeConfigProfile::parse(&profile)?; + } + if let Some(run_id) = overlay.runtime.run_id { + self.run_id = run_id; + } + if let Some(node_id) = overlay.runtime.node_id { + self.node_id = node_id; + } + if let Some(stage_index) = overlay.runtime.stage_index { + self.stage_index = stage_index; + } + if let Some(layer_end_exclusive) = overlay.runtime.layer_end_exclusive { + self.layer_end_exclusive = layer_end_exclusive; + } + if let Some(provider) = overlay.provider.kind { + self.provider = Some(ProviderKind::parse_deploy(&provider)?); + } + if let Some(image) = overlay.image.node { + self.image = image; + } + if let Some(mode) = overlay.relay.mode { + self.relay_mode = Some(mode); + } + if let Some(url) = overlay.relay.url { + self.relay_url = Some(url); + } + if let Some(rpc_bind) = overlay.prompt.rpc_addr { + self.rpc_bind = rpc_bind; + self.rpc_bind_label = "[prompt].rpc_addr"; + } + if let Some(max_tokens) = overlay.prompt.max_tokens { + self.default_max_tokens = max_tokens; + } + if let Some(dashboard) = overlay.prompt.dashboard { + self.dashboard = dashboard; + } + if let Some(model_id) = overlay.model.id { + self.model_id = model_id; + } + if let Some(path) = overlay.model.gguf_local_path { + self.gguf_source = GgufSource::LocalPath(path); + } + if let Some(repo) = overlay.model.gguf_repo { + self.set_gguf_repo(repo); + } + if let Some(file) = overlay.model.gguf_file { + self.set_gguf_file(file); + } + if let Some(revision) = overlay.model.gguf_revision { + self.set_gguf_revision(Some(revision)); + } + if let Some(path) = overlay.model.tokenizer_local_path { + self.tokenizer = TokenizerSource::LocalPath(path); + } + if let Some(max_context) = overlay.model.max_context { + self.max_context = Some(max_context); + } + if let Some(gpus) = overlay.docker.gpus { + self.docker_gpus = gpus; + } + if let Some(path) = overlay.docker.cached_model_host_path { + self.cached_model_host_path = Some(PathBuf::from(path)); + } + if let Some(path) = overlay.observability.datastream_frame_log { + self.datastream_frame_log = Some(PathBuf::from(path)); + } + if let Some(image) = overlay.vastai.image { + self.toml_vastai_image = Some(image); + } + if let Some(api_key) = overlay.vastai.api_key { + self.vastai_api_key = Some(api_key); + } + if let Some(command) = overlay.vastai.bootstrap_command { + self.vastai_bootstrap_command = Some(command); + } + if let Some(disk_gb) = overlay.vastai.disk_gb { + self.vastai_disk_gb = Some(disk_gb); + } + if let Some(ssh_user) = overlay.vastai.ssh_user { + self.vastai_ssh_user = Some(ssh_user); + } + if let Some(confirm_lease) = overlay.vastai.confirm_lease { + self.vastai_confirm_lease = Some(confirm_lease); + } + if let Some(onstart) = overlay.vastai.onstart { + self.vastai_onstart = Some(onstart); + } + if let Some(identity) = overlay.vastai.ssh_identity { + self.vastai_ssh_identity_raw = Some(identity); + } + if let Some(gpu_name) = overlay.vastai.gpu_name { + self.vastai_gpu_name = Some(gpu_name); + } + if let Some(min_gpu_ram_mb) = overlay.vastai.min_gpu_ram_mb { + self.vastai_min_gpu_ram_mb = Some(min_gpu_ram_mb); + } + if let Some(min_down_mbps) = overlay.vastai.min_down_mbps { + self.vastai_min_down_mbps = Some(min_down_mbps); + } + if let Some(min_up_mbps) = overlay.vastai.min_up_mbps { + self.vastai_min_up_mbps = Some(min_up_mbps); + } + if let Some(min_reliability) = overlay.vastai.min_reliability { + self.vastai_min_reliability = Some(min_reliability); + } + if let Some(require_verified) = overlay.vastai.require_verified { + self.vastai_require_verified = Some(require_verified); + } + if let Some(poll_interval_secs) = overlay.vastai.poll_interval_secs { + self.vastai_poll_interval_secs = Some(poll_interval_secs); + } + Ok(self) + } + fn overlay_env(mut self) -> Result { + if let Some(profile) = env_optional(MVP_RUNTIME_CONFIG_ENV) { + self.config_profile = RuntimeConfigProfile::parse(&profile)?; + } + if let Some(run_id) = env_optional("MVP_RUN_ID") { + self.run_id = Self::parse_value("MVP_RUN_ID", &run_id)?; + } + if let Some(node_id) = env_optional("MVP_LOGICAL_NODE_ID") { + self.node_id = Self::parse_value("MVP_LOGICAL_NODE_ID", &node_id)?; + } + if let Some(stage_index) = env_optional("MVP_STAGE_INDEX") { + self.stage_index = Self::parse_value("MVP_STAGE_INDEX", &stage_index)?; + } + if let Some(layer_end_exclusive) = env_optional("MVP_LAYER_END_EXCLUSIVE") { + self.layer_end_exclusive = + Self::parse_value("MVP_LAYER_END_EXCLUSIVE", &layer_end_exclusive)?; + } + if let Some(provider) = env_optional("MVP_NODE_PROVIDER").or_else(|| env_optional("MVP_PROVIDER")) { + self.provider = Some(ProviderKind::parse_deploy(&provider)?); + } + if let Some(image) = env_optional("MVP_NODE_IMAGE") { + self.set_process_image(image); + } + if let Some(gpus) = env_optional("MVP_DOCKER_GPUS") { + self.docker_gpus = gpus; + } + if let Some(path) = env_optional(CACHED_MODEL_HOST_ENV) { + self.cached_model_host_path = Some(PathBuf::from(path)); + } + if let Some(rpc_bind) = env_optional("MVP_PROMPT_RPC_BIND") { + self.rpc_bind = rpc_bind; + self.rpc_bind_label = "MVP_PROMPT_RPC_BIND"; + } + if let Some(max_tokens) = env_optional("MVP_PROMPT_MAX_TOKENS") { + self.default_max_tokens = Self::parse_value("MVP_PROMPT_MAX_TOKENS", &max_tokens)?; + } + if let Some(dashboard) = env_optional("MVP_DASHBOARD") { + self.dashboard = Self::parse_bool("MVP_DASHBOARD", &dashboard)?; + } + if let Some(path) = env_optional(DATASTREAM_FRAME_LOG_ENV) { + self.datastream_frame_log = Some(PathBuf::from(path)); + } + if let Some(model_id) = env_optional("MVP_MODEL_ID") { + self.model_id = model_id; + } + if let Some(path) = env_optional("MVP_GGUF_LOCAL_PATH") { + self.gguf_source = GgufSource::LocalPath(path); + } + if let Some(repo) = env_optional("MVP_GGUF_REPO") { + self.set_gguf_repo(repo); + } + if let Some(file) = env_optional("MVP_GGUF_FILE") { + self.set_gguf_file(file); + } + if let Some(revision) = env_optional("MVP_GGUF_REVISION") { + self.set_gguf_revision(Some(revision)); + } + if let Some(path) = env_optional("MVP_TOKENIZER_LOCAL_PATH") { + self.tokenizer = TokenizerSource::LocalPath(path); + } + if let Some(max_context) = env_optional("MVP_MAX_CONTEXT") { + self.max_context = Some(Self::parse_value("MVP_MAX_CONTEXT", &max_context)?); + } + if let Some(mode) = env_optional("MVP_IROH_RELAY_MODE") { + self.relay_mode = Some(mode.to_ascii_lowercase()); + } + if let Some(url) = env_optional(MVP_IROH_RELAY_URL_ENV) + .or_else(|| env_optional(SWACTOR_IROH_RELAY_URL_ENV)) + { + self.relay_url = Some(url); + } + if let Some(api_key) = env_optional("MVP_VASTAI_API_KEY").or_else(|| env_optional("VASTAI_API_KEY")) { + self.vastai_api_key = Some(api_key); + } + if let Some(command) = env_optional("MVP_VASTAI_BOOTSTRAP_COMMAND") { + self.vastai_bootstrap_command = Some(command); + } + if let Some(identity) = env_optional("MVP_VASTAI_SSH_IDENTITY") { + self.vastai_ssh_identity_raw = Some(identity); + } + if let Some(disk_gb) = env_optional("MVP_VASTAI_DISK_GB") { + self.vastai_disk_gb_raw = Some(disk_gb); + } + if let Some(ssh_user) = env_optional("MVP_VASTAI_SSH_USER") { + self.vastai_ssh_user = Some(ssh_user); + } + if let Some(confirm_lease) = env_optional("MVP_VASTAI_CONFIRM_LEASE") { + self.vastai_confirm_lease_raw = Some(confirm_lease); + } + if let Some(onstart) = env_optional("MVP_VASTAI_ONSTART") { + self.vastai_onstart = Some(onstart); + } + if let Some(gpu_name) = env_optional("MVP_VASTAI_GPU_NAME") { + self.vastai_gpu_name = Some(gpu_name); + } + if let Some(min_gpu_ram_mb) = env_optional("MVP_VASTAI_MIN_GPU_RAM_MB") { + self.vastai_min_gpu_ram_mb_raw = Some(min_gpu_ram_mb); + } + if let Some(min_down_mbps) = env_optional("MVP_VASTAI_MIN_DOWN_MBPS") { + self.vastai_min_down_mbps_raw = Some(min_down_mbps); + } + if let Some(min_up_mbps) = env_optional("MVP_VASTAI_MIN_UP_MBPS") { + self.vastai_min_up_mbps_raw = Some(min_up_mbps); + } + if let Some(min_reliability) = env_optional("MVP_VASTAI_MIN_RELIABILITY") { + self.vastai_min_reliability_raw = Some(min_reliability); + } + if let Some(require_verified) = env_optional("MVP_VASTAI_REQUIRE_VERIFIED") { + self.vastai_require_verified_raw = Some(require_verified); + } + if let Some(poll_interval_secs) = env_optional("MVP_VASTAI_POLL_INTERVAL_SECS") { + self.vastai_poll_interval_secs_raw = Some(poll_interval_secs); + } + Ok(self) + } + + fn overlay_cli(mut self, args: impl IntoIterator) -> Result { + let mut args = args.into_iter(); while let Some(arg) = args.next() { match arg.as_str() { - "--image" => config.image = next_arg(&mut args, "--image")?, - "--gpus" => config.docker_gpus = next_arg(&mut args, "--gpus")?, + "--runtime-config" => { + self.config_profile = + RuntimeConfigProfile::parse(&next_arg(&mut args, "--runtime-config")?)? + } + "--provider" => { + self.provider = Some(ProviderKind::parse_deploy(&next_arg( + &mut args, + "--provider", + )?)?) + } + "--image" => self.set_process_image(next_arg(&mut args, "--image")?), + "--gpus" => self.docker_gpus = next_arg(&mut args, "--gpus")?, "--rpc-bind" => { - config.rpc_bind = next_arg(&mut args, "--rpc-bind")? - .parse() - .map_err(|e| format!("invalid --rpc-bind: {e}"))? + self.rpc_bind = next_arg(&mut args, "--rpc-bind")?; + self.rpc_bind_label = "--rpc-bind"; } - "--run-id" => config.run_id = parse_next(&mut args, "--run-id")?, - "--node-id" => config.node_id = parse_next(&mut args, "--node-id")?, - "--max-tokens" => { - config.default_max_tokens = parse_next(&mut args, "--max-tokens")? + "--run-id" => self.run_id = parse_next(&mut args, "--run-id")?, + "--node-id" => self.node_id = parse_next(&mut args, "--node-id")?, + "--stage-index" => self.stage_index = parse_next(&mut args, "--stage-index")?, + "--layer-end-exclusive" => { + self.layer_end_exclusive = parse_next(&mut args, "--layer-end-exclusive")? } + "--max-tokens" => self.default_max_tokens = parse_next(&mut args, "--max-tokens")?, + "--dashboard" => self.dashboard = true, + "--no-dashboard" => self.dashboard = false, "--datastream-frame-log" => { - config.datastream_frame_log = Some(PathBuf::from(next_arg( + self.datastream_frame_log = Some(PathBuf::from(next_arg( &mut args, "--datastream-frame-log", )?)); } - "--model-id" => config.model_id = next_arg(&mut args, "--model-id")?, + "--model-id" => self.model_id = next_arg(&mut args, "--model-id")?, "--gguf-local-path" => { - config.gguf_source = + self.gguf_source = GgufSource::LocalPath(next_arg(&mut args, "--gguf-local-path")?) } - "--gguf-repo" => { - let repo = next_arg(&mut args, "--gguf-repo")?; - config.gguf_source = match config.gguf_source { - GgufSource::HuggingFaceGguf { file, revision, .. } => { - GgufSource::HuggingFaceGguf { - repo, - file, - revision, - } - } - GgufSource::LocalPath(_) => GgufSource::HuggingFaceGguf { - repo, - file: env_string("MVP_GGUF_FILE", DEFAULT_HF_FILE), - revision: env_optional("MVP_GGUF_REVISION"), - }, - }; + "--gguf-repo" => self.set_gguf_repo(next_arg(&mut args, "--gguf-repo")?), + "--gguf-file" => self.set_gguf_file(next_arg(&mut args, "--gguf-file")?), + "--gguf-revision" => { + self.set_gguf_revision(Some(next_arg(&mut args, "--gguf-revision")?)) } - "--gguf-file" => { - let file = next_arg(&mut args, "--gguf-file")?; - config.gguf_source = match config.gguf_source { - GgufSource::HuggingFaceGguf { repo, revision, .. } => { - GgufSource::HuggingFaceGguf { - repo, - file, - revision, - } - } - GgufSource::LocalPath(_) => GgufSource::HuggingFaceGguf { - repo: env_string("MVP_GGUF_REPO", DEFAULT_HF_REPO), - file, - revision: env_optional("MVP_GGUF_REVISION"), - }, - }; + "--tokenizer-local-path" => { + self.tokenizer = + TokenizerSource::LocalPath(next_arg(&mut args, "--tokenizer-local-path")?) + } + "--max-context" => self.max_context = Some(parse_next(&mut args, "--max-context")?), + "--cached-model-host-path" => { + self.cached_model_host_path = Some(PathBuf::from(next_arg( + &mut args, + "--cached-model-host-path", + )?)); + } + "--relay-mode" => self.relay_mode = Some(next_arg(&mut args, "--relay-mode")?), + "--relay-url" => self.relay_url = Some(next_arg(&mut args, "--relay-url")?), + "--vastai-api-key" => { + self.vastai_api_key = Some(next_arg(&mut args, "--vastai-api-key")?) + } + "--vastai-bootstrap-command" => { + self.vastai_bootstrap_command = + Some(next_arg(&mut args, "--vastai-bootstrap-command")?) + } + "--vastai-ssh-identity" => { + self.vastai_ssh_identity_raw = Some(next_arg(&mut args, "--vastai-ssh-identity")?); + } + "--vastai-disk-gb" => { + self.vastai_disk_gb = Some(parse_next(&mut args, "--vastai-disk-gb")?); + self.vastai_disk_gb_raw = None; + } + "--vastai-ssh-user" => { + self.vastai_ssh_user = Some(next_arg(&mut args, "--vastai-ssh-user")?) + } + "--vastai-confirm-lease" => { + self.vastai_confirm_lease = Some(true); + self.vastai_confirm_lease_raw = None; + } + "--no-vastai-confirm-lease" => { + self.vastai_confirm_lease = Some(false); + self.vastai_confirm_lease_raw = None; + } + "--vastai-onstart" => { + self.vastai_onstart = Some(next_arg(&mut args, "--vastai-onstart")?) + } + "--vastai-gpu-name" => { + self.vastai_gpu_name = Some(next_arg(&mut args, "--vastai-gpu-name")?) + } + "--vastai-min-gpu-ram-mb" => { + self.vastai_min_gpu_ram_mb = + Some(parse_next(&mut args, "--vastai-min-gpu-ram-mb")?); + self.vastai_min_gpu_ram_mb_raw = None; + } + "--vastai-min-down-mbps" => { + self.vastai_min_down_mbps = + Some(parse_next(&mut args, "--vastai-min-down-mbps")?); + self.vastai_min_down_mbps_raw = None; + } + "--vastai-min-up-mbps" => { + self.vastai_min_up_mbps = + Some(parse_next(&mut args, "--vastai-min-up-mbps")?); + self.vastai_min_up_mbps_raw = None; + } + "--vastai-min-reliability" => { + self.vastai_min_reliability = + Some(parse_next(&mut args, "--vastai-min-reliability")?); + self.vastai_min_reliability_raw = None; + } + "--vastai-require-verified" => { + self.vastai_require_verified = Some(true); + self.vastai_require_verified_raw = None; + } + "--no-vastai-require-verified" => { + self.vastai_require_verified = Some(false); + self.vastai_require_verified_raw = None; + } + "--vastai-poll-interval-secs" => { + self.vastai_poll_interval_secs = + Some(parse_next(&mut args, "--vastai-poll-interval-secs")?); + self.vastai_poll_interval_secs_raw = None; } other => return Err(format!("unknown argument {other:?}")), } } - Ok(config) + Ok(self) + } + + fn finalize(self) -> Result { + let provider = self + .provider + .unwrap_or_else(|| self.config_profile.default_provider()); + let mut image = self.image.clone(); + if provider == ProviderKind::VastAi && !self.image_overridden_after_toml { + if let Some(vastai_image) = &self.toml_vastai_image { + image = vastai_image.clone(); + } + } + let cached_model = self + .cached_model_host_path + .clone() + .map(|path| CachedModelConfig::from_host_path(provider, path)) + .transpose()?; + let mut gguf_source = self.gguf_source.clone(); + if let Some(cached_model) = &cached_model { + gguf_source = GgufSource::LocalPath(cached_model.container_path.clone()); + } + let relay = relay_runtime_config_from_settings( + self.run_id, + self.relay_mode.as_deref(), + self.relay_url.as_deref(), + )?; + let vastai = if provider == ProviderKind::VastAi { + Some(VastAiRuntimeConfig::from_builder(&self)?) + } else { + None + }; + Ok(Config { + config_profile: self.config_profile, + image, + docker_gpus: self.docker_gpus, + provider, + rpc_bind: self + .rpc_bind + .parse() + .map_err(|e| format!("invalid {}: {e}", self.rpc_bind_label))?, + run_id: self.run_id, + node_id: self.node_id, + stage_index: self.stage_index, + layer_end_exclusive: self.layer_end_exclusive, + model_id: self.model_id, + gguf_source, + tokenizer: self.tokenizer, + default_max_tokens: self.default_max_tokens, + dashboard: self.dashboard, + max_context: self.max_context, + relay, + vastai, + cached_model, + datastream_frame_log: self.datastream_frame_log, + }) + } + + fn set_process_image(&mut self, image: String) { + self.image = image; + self.image_overridden_after_toml = true; + } + + fn set_gguf_repo(&mut self, repo: String) { + let (file, revision) = match &self.gguf_source { + GgufSource::HuggingFaceGguf { file, revision, .. } => (file.clone(), revision.clone()), + GgufSource::LocalPath(_) => (DEFAULT_HF_FILE.to_owned(), None), + }; + self.gguf_source = GgufSource::HuggingFaceGguf { + repo, + file, + revision, + }; + } + + fn set_gguf_file(&mut self, file: String) { + let (repo, revision) = match &self.gguf_source { + GgufSource::HuggingFaceGguf { repo, revision, .. } => (repo.clone(), revision.clone()), + GgufSource::LocalPath(_) => (DEFAULT_HF_REPO.to_owned(), None), + }; + self.gguf_source = GgufSource::HuggingFaceGguf { + repo, + file, + revision, + }; + } + + fn set_gguf_revision(&mut self, revision: Option) { + let (repo, file) = match &self.gguf_source { + GgufSource::HuggingFaceGguf { repo, file, .. } => (repo.clone(), file.clone()), + GgufSource::LocalPath(_) => (DEFAULT_HF_REPO.to_owned(), DEFAULT_HF_FILE.to_owned()), + }; + self.gguf_source = GgufSource::HuggingFaceGguf { + repo, + file, + revision, + }; + } + + fn parse_value(name: &str, value: &str) -> Result + where + T: std::str::FromStr, + T::Err: std::fmt::Display, + { + value + .parse::() + .map_err(|e| format!("invalid {name}={value:?}: {e}")) + } + + fn parse_bool(name: &str, value: &str) -> Result { + match value.to_ascii_lowercase().as_str() { + "1" | "true" | "yes" | "on" => Ok(true), + "0" | "false" | "no" | "off" => Ok(false), + _ => Err(format!( + "invalid {name}={value:?}; use 1/0, true/false, yes/no, or on/off" + )), + } + } +} + +impl Config { + fn from_defaults_toml_env_args(args: impl IntoIterator) -> Result { + Self::from_layers_with_path_and_args(Some(Path::new(DEFAULT_CONFIG_PATH)), args) + } + + fn from_layers_with_path_and_args( + path: Option<&Path>, + args: impl IntoIterator, + ) -> Result { + let mut builder = Self::hardcoded_defaults(); + if let Some(path) = path { + if let Some(overlay) = TomlConfigOverlay::load_optional(path)? { + builder = builder.overlay_toml(overlay)?; + } + } + builder.overlay_env()?.overlay_cli(args)?.finalize() + } + + fn hardcoded_defaults() -> ConfigBuilder { + ConfigBuilder::hardcoded_defaults() } fn provider_datastream_detail(&self) -> Value { @@ -1075,6 +1624,9 @@ impl Config { if matches!(self.tokenizer, TokenizerSource::LocalPath(_)) { keys.push("MVP_TOKENIZER_LOCAL_PATH"); } + if self.max_context.is_some() { + keys.push("MVP_MAX_CONTEXT"); + } keys } @@ -1139,6 +1691,9 @@ impl Config { if let TokenizerSource::LocalPath(path) = &self.tokenizer { env.push(("MVP_TOKENIZER_LOCAL_PATH".to_owned(), path.clone())); } + if let Some(max_context) = self.max_context { + env.push(("MVP_MAX_CONTEXT".to_owned(), max_context.to_string())); + } let args = match self.provider { ProviderKind::VastAi => self .vastai @@ -1528,8 +2083,8 @@ struct DashboardSupport { #[cfg(feature = "local-e2e")] impl DashboardSupport { - fn start_from_env() -> Result, String> { - if !env_bool("MVP_DASHBOARD", false)? { + fn start(enabled: bool) -> Result, String> { + if !enabled { return Ok(None); } let mut config = dashboard::DashboardConfig::default(); @@ -1554,8 +2109,8 @@ struct DashboardSupport; #[cfg(not(feature = "local-e2e"))] impl DashboardSupport { - fn start_from_env() -> Result, String> { - if env_bool("MVP_DASHBOARD", false)? { + fn start(enabled: bool) -> Result, String> { + if enabled { return Err( "MVP_DASHBOARD requires building mvp-system with feature local-e2e".to_owned(), ); @@ -2178,84 +2733,6 @@ fn optional_env(name: &str) -> Option<(String, String)> { env_optional(name).map(|value| (name.to_owned(), value)) } -fn env_string(name: &str, default: &str) -> String { - env_optional(name).unwrap_or_else(|| default.to_owned()) -} - -fn provider_from_env(config_profile: RuntimeConfigProfile) -> Result { - match env_optional("MVP_NODE_PROVIDER").or_else(|| env_optional("MVP_PROVIDER")) { - Some(value) => ProviderKind::parse_deploy(&value), - None => Ok(config_profile.default_provider()), - } -} - -fn env_bool(name: &str, default: bool) -> Result { - match env_optional(name) { - None => Ok(default), - Some(value) => match value.to_ascii_lowercase().as_str() { - "1" | "true" | "yes" | "on" => Ok(true), - "0" | "false" | "no" | "off" => Ok(false), - _ => Err(format!( - "invalid {name}={value:?}; use 1/0, true/false, yes/no, or on/off" - )), - }, - } -} - -fn env_u64(name: &str, default: u64) -> Result { - match env_optional(name) { - Some(value) => value - .parse::() - .map_err(|e| format!("invalid {name}={value:?}: {e}")), - None => Ok(default), - } -} - -fn env_u32(name: &str, default: u32) -> Result { - match env_optional(name) { - Some(value) => value - .parse::() - .map_err(|e| format!("invalid {name}={value:?}: {e}")), - None => Ok(default), - } -} - -fn env_optional_u64(name: &str) -> Result, String> { - env_optional(name) - .map(|value| { - value - .parse::() - .map_err(|e| format!("invalid {name}={value:?}: {e}")) - }) - .transpose() -} - -fn env_optional_f64(name: &str) -> Result, String> { - env_optional(name) - .map(|value| { - value - .parse::() - .map_err(|e| format!("invalid {name}={value:?}: {e}")) - }) - .transpose() -} - -fn gguf_source_from_env() -> GgufSource { - if let Some(path) = env_optional("MVP_GGUF_LOCAL_PATH") { - return GgufSource::LocalPath(path); - } - GgufSource::HuggingFaceGguf { - repo: env_string("MVP_GGUF_REPO", DEFAULT_HF_REPO), - file: env_string("MVP_GGUF_FILE", DEFAULT_HF_FILE), - revision: env_optional("MVP_GGUF_REVISION"), - } -} - -fn tokenizer_from_env() -> TokenizerSource { - env_optional("MVP_TOKENIZER_LOCAL_PATH") - .map(TokenizerSource::LocalPath) - .unwrap_or(TokenizerSource::EmbeddedGguf) -} fn resolve_vastai_ssh_identity(explicit: Option) -> Result { match explicit { @@ -2413,7 +2890,7 @@ fn command_output_failure_detail(output: &std::process::Output, secret: Option<& #[cfg(test)] fn relay_mode_from_env() -> Result { - relay_runtime_config_from_env(env_u64("MVP_RUN_ID", 1)?).map(|relay| relay.mode) + relay_runtime_config_from_env(1).map(|relay| relay.mode) } fn next_arg(args: &mut impl Iterator, name: &str) -> Result { @@ -2446,6 +2923,7 @@ mod tests { "MVP_CPU_LINE_PROFILE", "MVP_CPU_LINE_PROFILE_INTERVAL_MS", "CUDA_DEVICE_SCHEDULE", + "MVP_DASHBOARD", "MVP_DOCKER_GPUS", "MVP_GGUF_FILE", "MVP_GGUF_LOCAL_PATH", @@ -2456,6 +2934,7 @@ mod tests { "MVP_LAYER_END_EXCLUSIVE", "MVP_LOGICAL_NODE_ID", "MVP_MODEL_CACHE_DIR", + "MVP_MAX_CONTEXT", "MVP_MODEL_ID", "MVP_NODE_IMAGE", "MVP_NODE_PROVIDER", @@ -2533,8 +3012,9 @@ mod tests { fn selected_provider(settings: &[(&'static str, &'static str)]) -> ProviderKind { with_clean_env(settings, || { - let profile = RuntimeConfigProfile::from_env().expect("runtime config parses"); - provider_from_env(profile).expect("provider parses") + Config::from_layers_with_path_and_args(None, std::iter::empty::()) + .expect("config parses") + .provider }) } @@ -2546,7 +3026,7 @@ mod tests { fn node_spec_env(settings: &[(&'static str, &'static str)]) -> Vec<(String, String)> { with_clean_env(settings, || { - let config = Config::from_env_and_args_iter(std::iter::empty::()) + let config = Config::from_layers_with_path_and_args(None, std::iter::empty::()) .expect("config parses"); let coordinator = EndpointAddr::new(iroh::SecretKey::from_bytes(&[9; 32]).public()); let datastream_sink = ActorAddress([11; 32]); @@ -2605,7 +3085,7 @@ mod tests { ("MVP_VASTAI_SSH_IDENTITY", "/tmp/mvp-vastai-key"), ], || { - Config::from_env_and_args_iter(std::iter::empty::()) + Config::from_layers_with_path_and_args(None, std::iter::empty::()) .expect("vastai config parses without ssh-keygen or vastai CLI") }, ); @@ -2652,6 +3132,29 @@ mod tests { } } + struct TempTomlFile { + path: PathBuf, + } + + impl TempTomlFile { + fn new(file_name: &str, contents: &str) -> Self { + let path = std::env::temp_dir().join(format!( + "mvp-orchestrator-config-test-{}-{}-{file_name}", + std::process::id(), + std::thread::current().name().unwrap_or("unnamed") + )); + let _ = std::fs::remove_file(&path); + std::fs::write(&path, contents).expect("write temp TOML config"); + Self { path } + } + } + + impl Drop for TempTomlFile { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } + } + #[test] fn frame_archive_writes_jsonl_records_for_text_and_binary_payloads() { static NEXT_TEMP_FILE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); @@ -2777,6 +3280,163 @@ mod tests { assert_eq!(stderr_payload["line"]["line"], "lease chain detail"); } + #[test] + fn config_layers_defaults_toml_env_then_cli() { + let toml = TempTomlFile::new( + "layering.toml", + r#" +[runtime] +profile = "deploy" +run_id = 41 +node_id = 9 +stage_index = 3 +layer_end_exclusive = 24 + +[provider] +kind = "vastai" + +[image] +node = "docker.io/example/from-image-node:toml" + +[vastai] +image = "docker.io/example/from-vastai-image:toml" +api_key = "toml-key" +bootstrap_command = "/toml/bootstrap" +disk_gb = 60 +gpu_name = "RTX 4090" + +[prompt] +rpc_addr = "127.0.0.1:19999" +max_tokens = 17 +dashboard = true + +[model] +id = "toml-model" +gguf_repo = "toml/repo" +gguf_file = "toml.gguf" +gguf_revision = "toml-rev" +max_context = 384 + +[relay] +mode = "disabled" +"#, + ); + + let config = with_clean_env( + &[ + ("MVP_NODE_PROVIDER", "docker"), + ("MVP_NODE_IMAGE", "docker.io/example/from-env:latest"), + ("MVP_PROMPT_MAX_TOKENS", "23"), + ("MVP_MODEL_ID", "env-model"), + ("MVP_IROH_RELAY_MODE", "default"), + ], + || { + Config::from_layers_with_path_and_args( + Some(&toml.path), + [ + "--image", + "docker.io/example/from-cli:latest", + "--max-tokens", + "31", + "--model-id", + "cli-model", + "--max-context", + "768", + ] + .into_iter() + .map(str::to_owned), + ) + .expect("layered config parses") + }, + ); + + assert_eq!(config.provider, ProviderKind::Docker); + assert_eq!(config.image, "docker.io/example/from-cli:latest"); + assert_eq!(config.default_max_tokens, 31); + assert_eq!(config.model_id, "cli-model"); + assert_eq!(config.run_id, 41); + assert_eq!(config.node_id, 9); + assert_eq!(config.stage_index, 3); + assert_eq!(config.layer_end_exclusive, 24); + assert!(config.dashboard); + assert_eq!(config.max_context, Some(768)); + assert!(matches!(config.relay.mode, iroh::RelayMode::Default)); + assert!(config.vastai.is_none()); + } + + #[test] + fn toml_vastai_image_overrides_image_node_for_vastai_provider() { + let toml = TempTomlFile::new( + "vastai-image.toml", + r#" +[runtime] +profile = "deploy" + +[provider] +kind = "vastai" + +[image] +node = "docker.io/example/generic:toml" + +[vastai] +image = "docker.io/example/vastai:toml" +api_key = "k" +bootstrap_command = "/run" +"#, + ); + + let config = with_clean_env(&[], || { + Config::from_layers_with_path_and_args(Some(&toml.path), std::iter::empty::()) + .expect("VastAI TOML config parses") + }); + + assert_eq!(config.provider, ProviderKind::VastAi); + assert_eq!(config.image, "docker.io/example/vastai:toml"); + } + + #[test] + fn missing_toml_uses_hardcoded_defaults() { + let missing_path = std::env::temp_dir().join(format!( + "mvp-orchestrator-missing-config-{}-{}.toml", + std::process::id(), + std::thread::current().name().unwrap_or("unnamed") + )); + let _ = std::fs::remove_file(&missing_path); + + let config = with_clean_env(&[], || { + Config::from_layers_with_path_and_args( + Some(&missing_path), + std::iter::empty::(), + ) + .expect("missing optional TOML config uses defaults") + }); + + assert_eq!(config.provider, ProviderKind::Docker); + assert_eq!(config.image, DEFAULT_IMAGE); + assert_eq!(config.model_id, DEFAULT_MODEL_ID); + assert_eq!(config.default_max_tokens, DEFAULT_MAX_TOKENS); + assert!(!config.dashboard); + assert_eq!(config.max_context, None); + } + + #[test] + fn node_spec_propagates_max_context_when_configured() { + let config = with_clean_env(&[], || { + Config::from_layers_with_path_and_args( + None, + ["--max-context", "256"].into_iter().map(str::to_owned), + ) + .expect("CLI max context config parses") + }); + let coordinator = EndpointAddr::new(iroh::SecretKey::from_bytes(&[3; 32]).public()); + let datastream_sink = ActorAddress([17; 32]); + let spec = config + .node_spec(coordinator, datastream_sink) + .expect("node spec builds"); + + assert_eq!(env_value(&spec.env, "MVP_MAX_CONTEXT"), Some("256")); + } + #[test] fn runtime_profile_selects_provider_and_node_provider_takes_precedence() { assert_eq!( @@ -2841,7 +3501,7 @@ mod tests { ("MVP_VASTAI_MIN_DOWN_MBPS", "not-a-float"), ], || { - Config::from_env_and_args_iter(std::iter::empty::()) + Config::from_layers_with_path_and_args(None, std::iter::empty::()) .expect("docker config ignores VastAI-only env") }, ); @@ -2861,7 +3521,7 @@ mod tests { (CACHED_MODEL_HOST_ENV, model.raw_path.as_os_str().to_owned()), ], || { - Config::from_env_and_args_iter(std::iter::empty::()) + Config::from_layers_with_path_and_args(None, std::iter::empty::()) .expect("docker cached model config parses") }, ); @@ -2900,7 +3560,7 @@ mod tests { ), ("MVP_VASTAI_DISK_GB", OsString::from("not-a-u32")), ], - || match Config::from_env_and_args_iter(std::iter::empty::()) { + || match Config::from_layers_with_path_and_args(None, std::iter::empty::()) { Ok(_) => panic!("deploy cached model must be rejected"), Err(error) => error, }, diff --git a/crates/mvp-system/src/bin/worker_node.rs b/crates/mvp-system/src/bin/worker_node.rs index 14b07e1..bc235b9 100644 --- a/crates/mvp-system/src/bin/worker_node.rs +++ b/crates/mvp-system/src/bin/worker_node.rs @@ -417,18 +417,19 @@ fn run() -> Result<(), String> { "ready", json!({"actor":orchestrator,"source":orchestrator_source}), )?; - let node_actor = match stack.runtime.spawn(NodeAgentActor::new( + let node_agent = NodeAgentActor::new( stage::NodeId(config.logical_node_id), orchestrator, Some(*reports.addr()), - )) { + ); + let node_actor = match stack.runtime.spawn(node_agent) { Ok(actor) => { emit_stdio_node_event( &config, NODE_BOOTSTRAP_CHANNEL, "node_agent", "ready", - json!({"node_actor":actor}), + json!({"node_actor":actor,"source":"generated"}), )?; actor } @@ -438,7 +439,7 @@ fn run() -> Result<(), String> { NODE_BOOTSTRAP_CHANNEL, "node_agent", "failed", - json!({"error":error.to_string()}), + json!({"error":error.to_string(),"source":"generated"}), )?; return Err(format!("spawn node agent: {error}")); } @@ -569,7 +570,7 @@ fn run() -> Result<(), String> { .map_err(|e| format!("flush ready line: {e}"))?; if let Some(prompt) = &config.self_test_prompt { - run_self_test(&mut worker, &config, prompt, &mut datastream)?; + run_self_test(&mut worker, &config, prompt, &mut datastream, &mut driver, &stack)?; } let shutdown_rx = spawn_stdin_shutdown_listener(); @@ -1197,6 +1198,8 @@ fn run_self_test( config: &DeploymentConfig, prompt: &str, datastream: &mut DatastreamEmitter, + driver: &mut IrohDriver, + stack: &DistributionRuntimeStack, ) -> Result<(), String> { emit_node_event( datastream, @@ -1206,7 +1209,7 @@ fn run_self_test( "started", json!({"prompt_bytes":prompt.len()}), ); - let mut pump = || {}; + let mut pump = || pump_network(driver, stack); worker.configure_role( config.run_id, config.stage_index, @@ -1664,6 +1667,7 @@ impl Drop for TinygradWorker { } } + fn spawn_stdin_shutdown_listener() -> Receiver<()> { let (tx, rx) = mpsc::channel(); thread::spawn(move || { diff --git a/crates/mvp-system/src/config.rs b/crates/mvp-system/src/config.rs index 05067ff..0fbd995 100644 --- a/crates/mvp-system/src/config.rs +++ b/crates/mvp-system/src/config.rs @@ -4,18 +4,40 @@ use std::path::{Path, PathBuf}; use serde::Deserialize; use swactor_vastai::SelectionPolicy; -const DEFAULT_CONFIG_PATH: &str = ".config/config.toml"; +pub const DEFAULT_CONFIG_PATH: &str = ".config/config.toml"; +/// Transient serde target for `.config/config.toml`. +/// This is not runtime state; it exists only to interpret TOML fields and overlay them onto hardcoded defaults. #[derive(Clone, Debug, Default, Deserialize, PartialEq)] #[serde(default)] -pub struct Config { +pub struct TomlConfigOverlay { + pub runtime: RuntimeConfigOverlay, + pub provider: ProviderConfigOverlay, pub image: ImageConfig, pub relay: RelayConfig, pub prompt: PromptConfig, pub model: ModelConfig, + pub docker: DockerConfigOverlay, + pub observability: ObservabilityConfigOverlay, pub vastai: VastAiConfig, } +#[derive(Clone, Debug, Default, Deserialize, PartialEq)] +#[serde(default)] +pub struct RuntimeConfigOverlay { + pub profile: Option, + pub run_id: Option, + pub node_id: Option, + pub stage_index: Option, + pub layer_end_exclusive: Option, +} + +#[derive(Clone, Debug, Default, Deserialize, PartialEq)] +#[serde(default)] +pub struct ProviderConfigOverlay { + pub kind: Option, +} + #[derive(Clone, Debug, Default, Deserialize, PartialEq)] #[serde(default)] pub struct ImageConfig { @@ -45,12 +67,27 @@ pub struct PromptConfig { #[serde(default)] pub struct ModelConfig { pub id: Option, + pub gguf_local_path: Option, pub gguf_repo: Option, pub gguf_file: Option, pub gguf_revision: Option, + pub tokenizer_local_path: Option, pub max_context: Option, } +#[derive(Clone, Debug, Default, Deserialize, PartialEq)] +#[serde(default)] +pub struct DockerConfigOverlay { + pub gpus: Option, + pub cached_model_host_path: Option, +} + +#[derive(Clone, Debug, Default, Deserialize, PartialEq)] +#[serde(default)] +pub struct ObservabilityConfigOverlay { + pub datastream_frame_log: Option, +} + #[derive(Clone, Debug, Default, Deserialize, PartialEq)] #[serde(default)] pub struct VastAiConfig { @@ -58,20 +95,23 @@ pub struct VastAiConfig { pub image: Option, pub bootstrap_command: Option, pub disk_gb: Option, + pub ssh_user: Option, + pub confirm_lease: Option, pub gpu_name: Option, pub min_gpu_ram_mb: Option, pub min_down_mbps: Option, pub min_up_mbps: Option, pub min_reliability: Option, pub require_verified: Option, + pub poll_interval_secs: Option, pub onstart: Option, pub ssh_identity: Option, } #[derive(Clone, Debug, PartialEq)] -pub struct LoadedConfig { +pub struct LoadedTomlConfigOverlay { pub path: Option, - pub config: Config, + pub overlay: TomlConfigOverlay, } #[derive(Clone, Debug, PartialEq)] @@ -91,30 +131,31 @@ pub struct ResolvedVastAiConfig { pub ssh_identity: Option, } -impl Config { - pub fn load(path: Option<&Path>) -> Result { +impl TomlConfigOverlay { + pub fn load(path: Option<&Path>) -> Result { match path { - Some(path) => Self::load_required(path).map(|config| LoadedConfig { + Some(path) => Self::load_required(path).map(|overlay| LoadedTomlConfigOverlay { path: Some(path.to_path_buf()), - config, + overlay, }), None => { let default = Path::new(DEFAULT_CONFIG_PATH); - if default.is_file() { - Self::load_required(default).map(|config| LoadedConfig { - path: Some(default.to_path_buf()), - config, - }) - } else { - Ok(LoadedConfig { - path: None, - config: Self::default(), - }) - } + Self::load_optional(default).map(|overlay| LoadedTomlConfigOverlay { + path: overlay.as_ref().map(|_| default.to_path_buf()), + overlay: overlay.unwrap_or_default(), + }) } } } + pub fn load_optional(path: &Path) -> Result, String> { + if path.is_file() { + Self::load_required(path).map(Some) + } else { + Ok(None) + } + } + pub fn load_required(path: &Path) -> Result { let text = fs::read_to_string(path).map_err(|e| format!("read config {}: {e}", path.display()))?; @@ -222,8 +263,19 @@ mod tests { #[test] fn config_toml_parses_explicit_chat_and_vastai_fields() { - let config = Config::from_str( + let config = TomlConfigOverlay::from_str( r#" + +[runtime] +profile = "deploy" +run_id = 42 +node_id = 7 +stage_index = 2 +layer_end_exclusive = 24 + +[provider] +kind = "vastai" + [image] node = "ghcr.io/swactor/mvp-node:latest" tag = "trial" @@ -243,15 +295,27 @@ dashboard = false [model] id = "smollm2-135m-instruct-q4" +gguf_local_path = "/models/local.gguf" gguf_repo = "QuantFactory/SmolLM2-135M-Instruct-GGUF" gguf_file = "SmolLM2-135M-Instruct.Q4_K_M.gguf" gguf_revision = "main" +tokenizer_local_path = "/models/tokenizer.json" max_context = 512 + +[docker] +gpus = "all" +cached_model_host_path = "/cache/model.gguf" + +[observability] +datastream_frame_log = "/tmp/frames.jsonl" + [vastai] api_key = "vast-key" image = "registry.example.com/team/mvp-node:latest" bootstrap_command = "/opt/mvp/node --join" disk_gb = 80 +ssh_user = "ubuntu" +confirm_lease = true gpu_name = "RTX 4090" min_gpu_ram_mb = 24000 min_down_mbps = 250.5 @@ -260,10 +324,17 @@ min_reliability = 0.99 require_verified = true onstart = "echo preparing" ssh_identity = "~/.ssh/swactor_vastai_ed25519" +poll_interval_secs = 30 "#, ) .expect("explicit config TOML parses"); + assert_eq!(config.runtime.profile.as_deref(), Some("deploy")); + assert_eq!(config.runtime.run_id, Some(42)); + assert_eq!(config.runtime.node_id, Some(7)); + assert_eq!(config.runtime.stage_index, Some(2)); + assert_eq!(config.runtime.layer_end_exclusive, Some(24)); + assert_eq!(config.provider.kind.as_deref(), Some("vastai")); assert_eq!( config.image.node.as_deref(), Some("ghcr.io/swactor/mvp-node:latest") @@ -281,6 +352,10 @@ ssh_identity = "~/.ssh/swactor_vastai_ed25519" assert_eq!(config.prompt.max_tokens, Some(128)); assert_eq!(config.prompt.dashboard, Some(false)); assert_eq!(config.model.id.as_deref(), Some("smollm2-135m-instruct-q4")); + assert_eq!( + config.model.gguf_local_path.as_deref(), + Some("/models/local.gguf") + ); assert_eq!( config.model.gguf_repo.as_deref(), Some("QuantFactory/SmolLM2-135M-Instruct-GGUF") @@ -290,7 +365,20 @@ ssh_identity = "~/.ssh/swactor_vastai_ed25519" Some("SmolLM2-135M-Instruct.Q4_K_M.gguf") ); assert_eq!(config.model.gguf_revision.as_deref(), Some("main")); + assert_eq!( + config.model.tokenizer_local_path.as_deref(), + Some("/models/tokenizer.json") + ); assert_eq!(config.model.max_context, Some(512)); + assert_eq!(config.docker.gpus.as_deref(), Some("all")); + assert_eq!( + config.docker.cached_model_host_path.as_deref(), + Some("/cache/model.gguf") + ); + assert_eq!( + config.observability.datastream_frame_log.as_deref(), + Some("/tmp/frames.jsonl") + ); assert_eq!(config.vastai.api_key.as_deref(), Some("vast-key")); assert_eq!( config.vastai.image.as_deref(), @@ -301,12 +389,15 @@ ssh_identity = "~/.ssh/swactor_vastai_ed25519" Some("/opt/mvp/node --join") ); assert_eq!(config.vastai.disk_gb, Some(80)); + assert_eq!(config.vastai.ssh_user.as_deref(), Some("ubuntu")); + assert_eq!(config.vastai.confirm_lease, Some(true)); assert_eq!(config.vastai.gpu_name.as_deref(), Some("RTX 4090")); assert_eq!(config.vastai.min_gpu_ram_mb, Some(24_000)); assert_eq!(config.vastai.min_down_mbps, Some(250.5)); assert_eq!(config.vastai.min_up_mbps, Some(50.25)); assert_eq!(config.vastai.min_reliability, Some(0.99)); assert_eq!(config.vastai.require_verified, Some(true)); + assert_eq!(config.vastai.poll_interval_secs, Some(30)); assert_eq!(config.vastai.onstart.as_deref(), Some("echo preparing")); assert_eq!( config.vastai.ssh_identity.as_deref(), @@ -323,7 +414,7 @@ ssh_identity = "~/.ssh/swactor_vastai_ed25519" )); let _ = fs::remove_file(&path); - let error = Config::load(Some(&path)).expect_err("missing explicit config path errors"); + let error = TomlConfigOverlay::load(Some(&path)).expect_err("missing explicit config path errors"); assert!( error.contains(&path.display().to_string()), diff --git a/crates/mvp-system/src/relay_provisioning.rs b/crates/mvp-system/src/relay_provisioning.rs index 1a3842b..08588e9 100644 --- a/crates/mvp-system/src/relay_provisioning.rs +++ b/crates/mvp-system/src/relay_provisioning.rs @@ -126,12 +126,22 @@ impl RelayProvider for StaticRelayProvider { } pub fn relay_runtime_config_from_env(run_id: u64) -> Result { - match relay_mode_setting_from_env().as_deref() { + let mode = relay_mode_setting_from_env(); + let url = selected_relay_url_from_env(); + relay_runtime_config_from_settings(run_id, mode.as_deref(), url.as_deref()) +} + +pub fn relay_runtime_config_from_settings( + run_id: u64, + mode: Option<&str>, + url: Option<&str>, +) -> Result { + match mode { Some("disabled") => Ok(RelayRuntimeConfig { mode: RelayMode::Disabled, url: None, }), - None | Some("default") => relay_runtime_config_from_optional_static_provider(run_id), + None | Some("default") => relay_runtime_config_from_optional_static_provider(run_id, url), Some(other) => Err(format!( "unsupported {MVP_IROH_RELAY_MODE_ENV}={other:?}; use disabled or default" )), @@ -151,13 +161,15 @@ pub fn selected_relay_url_from_env() -> Option { fn relay_runtime_config_from_optional_static_provider( run_id: u64, + url: Option<&str>, ) -> Result { - let Some(mut provider) = StaticRelayProvider::from_env()? else { + let Some(raw_url) = url.map(str::trim).filter(|url| !url.is_empty()) else { return Ok(RelayRuntimeConfig { mode: RelayMode::Default, url: None, }); }; + let mut provider = StaticRelayProvider::from_url_str(raw_url)?; let lease = provider.provision_relay(RelayProvisionRequest { run_id, purpose: RelayPurpose::Combined,