use std::ffi::OsStr;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant};
use anyhow::{Context, Result, bail};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, Command};
use tokio_util::sync::CancellationToken;
use crate::shared::api::OpenAiClient;
use crate::shared::api::llama_args::{self, ManagedRole};
use crate::shared::i18n::Locale;
#[derive(Debug, Clone)]
pub struct ManagedConfig {
pub binary: PathBuf,
pub model_path: Option<String>,
pub mmproj: Option<String>,
pub gpu_layers: i32,
pub context_size: u32,
pub batch_size: Option<u32>,
pub parallel: u32,
pub jinja: bool,
pub reasoning_format: Option<String>,
pub embeddings: bool,
pub no_mmap: bool,
pub flash_attn: Option<String>,
pub spec_type: Option<String>,
pub draft_model: Option<String>,
pub draft_gpu_layers: Option<i32>,
pub draft_n_max: Option<u32>,
pub draft_n_min: Option<u32>,
pub host: String,
pub port: u16,
pub role: ManagedRole,
pub extra_args: Vec<String>,
pub kernel_cache: Option<PathBuf>,
}
impl ManagedConfig {
pub fn base_url(&self) -> String {
format!("http://127.0.0.1:{}/v1", self.port)
}
pub fn is_runnable(&self) -> bool {
!self.binary.as_os_str().is_empty()
&& self
.model_path
.as_deref()
.is_some_and(|m| !m.trim().is_empty())
}
}
pub const CPU_BATCH: u32 = 256;
const METAL_HOST: bool = cfg!(target_os = "macos");
pub const SERVER_UBATCH: u32 = 512;
pub const LLAMA_DEFAULT_BATCH: u32 = 2048;
pub const PREFILL_HOLD_LIMIT_SECS: u32 = 5;
pub const PREFILL_SAMPLE_MIN: u32 = 256;
fn auto_batch(gpu_layers: i32, metal: bool) -> Option<u32> {
(gpu_layers == 0 || metal).then_some(CPU_BATCH)
}
pub fn launched_batch(batch_size: Option<u32>, gpu_layers: i32) -> u32 {
batch_size
.or(auto_batch(gpu_layers, METAL_HOST))
.unwrap_or(LLAMA_DEFAULT_BATCH)
}
pub fn prefill_hold(batch: u32, prefill: crate::shared::api::contract::Prefill) -> Option<u32> {
if prefill.tokens < PREFILL_SAMPLE_MIN || batch <= CPU_BATCH {
return None;
}
let tps = prefill.tokens_per_second()?;
let hold = f64::from(batch) / tps;
(hold > f64::from(PREFILL_HOLD_LIMIT_SECS)).then(|| hold.round() as u32)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NoMmapSpelling {
LoadMode,
NoMmap,
}
impl NoMmapSpelling {
fn args(self) -> Vec<String> {
match self {
NoMmapSpelling::LoadMode => vec!["--load-mode".into(), "none".into()],
NoMmapSpelling::NoMmap => vec!["--no-mmap".into()],
}
}
}
pub fn spelling_from_help(help: &str) -> NoMmapSpelling {
if help.contains("--load-mode") {
NoMmapSpelling::LoadMode
} else {
NoMmapSpelling::NoMmap
}
}
pub fn no_mmap_spelling(binary: &std::path::Path) -> NoMmapSpelling {
match std::process::Command::new(binary).arg("--help").output() {
Ok(out) => {
let help = String::from_utf8_lossy(&out.stdout).into_owned()
+ &String::from_utf8_lossy(&out.stderr);
let spelling = spelling_from_help(&help);
tracing::debug!(?spelling, "the binary was asked how it spells no-mmap");
spelling
}
Err(err) => {
tracing::debug!(%err, "the binary did not answer --help; assuming the current spelling");
NoMmapSpelling::LoadMode
}
}
}
pub fn build_args(cfg: &ManagedConfig, no_mmap: NoMmapSpelling) -> Vec<String> {
let mut args = vec![
"--host".to_string(),
cfg.host.clone(),
"--port".to_string(),
cfg.port.to_string(),
"-ngl".to_string(),
cfg.gpu_layers.to_string(),
"-c".to_string(),
cfg.context_size.to_string(),
];
if cfg.parallel > 1 {
args.push("-np".into());
args.push(cfg.parallel.to_string());
args.push("--kv-unified".into());
}
if !cfg.embeddings
&& let Some(batch) = cfg.batch_size.or(auto_batch(cfg.gpu_layers, METAL_HOST))
{
args.push("-b".into());
args.push(batch.to_string());
args.push("-ub".into());
args.push(batch.min(SERVER_UBATCH).to_string());
}
if let Some(m) = &cfg.model_path {
args.push("-m".into());
args.push(m.clone());
}
if let Some(p) = &cfg.mmproj {
args.push("--mmproj".into());
args.push(p.clone());
}
if cfg.jinja {
args.push("--jinja".into());
}
if let Some(rf) = &cfg.reasoning_format {
args.push("--reasoning-format".into());
args.push(rf.clone());
}
if cfg.embeddings {
args.push("--embeddings".into());
args.push("-ub".into());
args.push(cfg.context_size.to_string());
args.push("-b".into());
args.push(cfg.context_size.to_string());
}
if cfg.no_mmap {
args.extend(no_mmap.args());
}
if let Some(fa) = &cfg.flash_attn {
args.push("--flash-attn".into());
args.push(fa.clone());
}
if let Some(st) = &cfg.spec_type {
args.push("--spec-type".into());
args.push(st.clone());
}
if let Some(md) = &cfg.draft_model {
args.push("-md".into());
args.push(md.clone());
}
if let Some(ngld) = cfg.draft_gpu_layers {
args.push("-ngld".into());
args.push(ngld.to_string());
}
if let Some(n) = cfg.draft_n_max {
args.push("--spec-draft-n-max".into());
args.push(n.to_string());
}
if let Some(n) = cfg.draft_n_min {
args.push("--spec-draft-n-min".into());
args.push(n.to_string());
}
args.extend(cfg.extra_args.iter().cloned());
args
}
pub struct ServerHandle {
kill: CancellationToken,
exited: ChildExit,
base_url: String,
}
#[derive(Debug, Clone, Default)]
pub struct ChildExit {
exited: CancellationToken,
reaped: CancellationToken,
said: Arc<Mutex<Said>>,
}
const BURST_GAP: Duration = Duration::from_secs(2);
const CAUSE_WINDOW: Duration = Duration::from_secs(10);
#[derive(Debug, Default)]
struct Said {
refusal: Option<String>,
missing: Option<String>,
burst: Option<Burst>,
ended: Option<Instant>,
}
#[derive(Debug)]
struct Burst {
first: String,
last: Instant,
}
impl Said {
fn record(&mut self, line: &str, at: Instant) {
if let Some(refusal) = refusal_line(line) {
self.refusal.get_or_insert(refusal);
} else if let Some(lib) = missing_library(line) {
self.missing.get_or_insert(lib);
} else if let Some(record) = error_record(line) {
match &mut self.burst {
Some(burst) if at.saturating_duration_since(burst.last) < BURST_GAP => {
burst.last = at;
}
_ => {
self.burst = Some(Burst {
first: record,
last: at,
})
}
}
}
}
fn logged_cause(&self) -> Option<&str> {
let burst = self.burst.as_ref()?;
match self.ended {
Some(end) if end.saturating_duration_since(burst.last) > CAUSE_WINDOW => None,
_ => Some(&burst.first),
}
}
}
impl ChildExit {
pub fn is_exited(&self) -> bool {
self.exited.is_cancelled()
}
pub async fn wait(&self) {
self.exited.cancelled().await
}
pub fn message(&self, loc: &Locale) -> String {
let said = self.said.lock().unwrap_or_else(PoisonError::into_inner);
if let Some(refusal) = &said.refusal {
return loc.tf("ui.err.managed.early_exit_said", &[("said", refusal)]);
}
if let Some(lib) = &said.missing {
return missing_library_message(lib, loc);
}
match said.logged_cause() {
Some(cause) if ptx_newer_than_driver(cause) => {
loc.tf("ui.err.managed.cuda_too_new", &[("said", cause)])
}
Some(cause) => loc.tf("ui.err.managed.exit_logged", &[("said", cause)]),
None => loc.t("ui.err.managed.early_exit").to_string(),
}
}
#[cfg(test)]
pub(crate) fn exited_saying(said: Option<&str>) -> Self {
let exit = Self::default();
exit.said.lock().unwrap().refusal = said.map(str::to_string);
exit.exited.cancel();
exit.reaped.cancel();
exit
}
}
fn ptx_newer_than_driver(record: &str) -> bool {
record.contains("PTX") && record.contains("unsupported toolchain")
}
pub fn missing_library(line: &str) -> Option<String> {
let clean = strip_ansi(line);
let lib = if let Some((_, rest)) = clean.split_once(": error while loading shared libraries: ")
{
rest.split(": ").next()?.trim()
} else {
let (head, rest) = clean.split_once(": Library not loaded: ")?;
if !head.trim_start().starts_with("dyld") {
return None;
}
rest.trim().rsplit('/').next()?
};
(!lib.is_empty()).then(|| lib.to_string())
}
pub fn missing_library_message(lib: &str, loc: &Locale) -> String {
let key = if lib.starts_with("libgomp.so") {
"ui.err.managed.missing_openmp"
} else {
"ui.err.managed.missing_library"
};
loc.tf(key, &[("lib", lib)])
}
impl Drop for ServerHandle {
fn drop(&mut self) {
self.kill.cancel();
}
}
impl ServerHandle {
pub fn base_url(&self) -> &str {
&self.base_url
}
pub fn exited(&self) -> ChildExit {
self.exited.clone()
}
pub fn launch(cfg: &ManagedConfig, loc: &'static Locale) -> Result<Self> {
Self::preflight(cfg, loc)?;
Self::spawn(cfg, loc)
}
fn preflight(cfg: &ManagedConfig, loc: &'static Locale) -> Result<()> {
if let Err(refused) = llama_args::check(&cfg.extra_args, cfg.role) {
bail!("{}", refused.launch_message(loc));
}
if let Some(model) = &cfg.model_path {
preflight_model(model, loc)?;
}
if let Some(draft) = &cfg.draft_model
&& !std::path::Path::new(draft).is_file()
{
bail!(
"{}",
loc.tf("ui.err.managed.draft_not_found", &[("path", draft)])
);
}
if let Some(mmproj) = &cfg.mmproj
&& !std::path::Path::new(mmproj).is_file()
{
bail!(
"{}",
loc.tf("ui.err.managed.mmproj_not_found", &[("path", mmproj)])
);
}
Ok(())
}
fn spawn(cfg: &ManagedConfig, loc: &'static Locale) -> Result<Self> {
let no_mmap = if cfg.no_mmap {
no_mmap_spelling(&cfg.binary)
} else {
NoMmapSpelling::LoadMode
};
let args = build_args(cfg, no_mmap);
tracing::info!(binary = %cfg.binary.display(), ?args, "launching managed llama-server");
let child = server_command(cfg, &args).spawn().with_context(|| {
loc.tf(
"ui.err.managed.spawn",
&[("path", &cfg.binary.display().to_string())],
)
})?;
let (kill, exited) = supervise(child);
Ok(Self {
kill,
exited,
base_url: cfg.base_url(),
})
}
}
pub fn port_taken(cfg: &ManagedConfig) -> bool {
std::net::TcpListener::bind(("127.0.0.1", cfg.port)).is_err()
}
#[derive(Default)]
pub struct PortLedger {
claims: Mutex<Vec<Claim>>,
}
struct Claim {
port: u16,
released: CancellationToken,
exit: ChildExit,
}
impl PortLedger {
pub fn launch(&self, cfg: &ManagedConfig, loc: &'static Locale) -> Result<ServerHandle> {
ServerHandle::preflight(cfg, loc)?;
let mut claims = self.claims.lock().unwrap_or_else(PoisonError::into_inner);
claims.retain(|c| !c.exit.reaped.is_cancelled());
let ours_leaving = claims
.iter()
.any(|c| c.port == cfg.port && c.released.is_cancelled());
if !ours_leaving && port_taken(cfg) {
bail!("{}", port_busy_message(cfg, loc));
}
let handle = ServerHandle::spawn(cfg, loc)?;
claims.push(Claim {
port: cfg.port,
released: handle.kill.clone(),
exit: handle.exited.clone(),
});
Ok(handle)
}
}
fn port_busy_message(cfg: &ManagedConfig, loc: &Locale) -> String {
let tab = match cfg.role {
ManagedRole::Assistant => "ui.settings.tab.assistant",
ManagedRole::Impersonation => "ui.settings.tab.impersonation",
ManagedRole::Embedder => "ui.settings.tab.embeddings",
};
loc.tf(
"ui.err.managed.port_busy",
&[
("port", &cfg.port.to_string()),
("section", loc.t("ui.settings.section.model")),
("tab", loc.t(tab)),
("field", loc.t("ui.settings.field.port")),
],
)
}
fn server_command(cfg: &ManagedConfig, args: &[String]) -> Command {
let mut cmd = Command::new(&cfg.binary);
cmd.args(args)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
for var in llama_args::scrubbed_env(cfg.role) {
cmd.env_remove(var);
}
let inherited = std::env::var_os(CUDA_CACHE_PATH);
if let Some(dir) = kernel_cache_for(cfg.kernel_cache.as_deref(), inherited.as_deref()) {
if let Err(err) = std::fs::create_dir_all(dir) {
tracing::warn!(dir = %dir.display(), %err, "no CUDA kernel cache directory");
}
cmd.env(CUDA_CACHE_PATH, dir);
}
cmd
}
const CUDA_CACHE_PATH: &str = "CUDA_CACHE_PATH";
fn kernel_cache_for<'a>(ours: Option<&'a Path>, inherited: Option<&OsStr>) -> Option<&'a Path> {
match inherited {
Some(value) if !value.is_empty() => None,
_ => ours,
}
}
fn supervise(mut child: Child) -> (CancellationToken, ChildExit) {
let exited = ChildExit::default();
let mut readers = Vec::new();
if let Some(out) = child.stdout.take() {
readers.push(tokio::spawn(forward_lines(out, false, exited.said.clone())));
}
if let Some(err) = child.stderr.take() {
readers.push(tokio::spawn(forward_lines(err, true, exited.said.clone())));
}
let kill = CancellationToken::new();
spawn_monitor(child, kill.clone(), exited.clone(), readers);
(kill, exited)
}
pub fn preflight_model(model: &str, loc: &'static Locale) -> Result<()> {
if !std::path::Path::new(model).is_file() {
bail!(
"{}",
loc.tf("ui.err.managed.model_not_found", &[("path", model)])
);
}
check_split_model(model, loc)
}
fn check_split_model(model: &str, loc: &'static Locale) -> Result<()> {
let Some(shard) = crate::shared::gguf::parse_shard(model) else {
return Ok(());
};
if shard.index != 1 {
bail!(
"{}",
loc.tf(
"ui.err.managed.shard_not_first",
&[("path", &shard.first())]
)
);
}
for part in shard.all().iter().skip(1) {
if !std::path::Path::new(part).is_file() {
bail!(
"{}",
loc.tf("ui.err.managed.shard_missing", &[("path", part)])
);
}
}
Ok(())
}
const OUTPUT_DRAIN: Duration = Duration::from_secs(1);
fn spawn_monitor(
mut child: Child,
kill: CancellationToken,
exited: ChildExit,
readers: Vec<tokio::task::JoinHandle<()>>,
) {
tokio::spawn(async move {
tokio::select! {
status = child.wait() => {
exited
.said
.lock()
.unwrap_or_else(PoisonError::into_inner)
.ended = Some(Instant::now());
exited.reaped.cancel();
match status {
Ok(s) => tracing::warn!(status = ?s, "managed llama-server exited on its own"),
Err(e) => tracing::warn!(error = %e, "error waiting for the child llama-server"),
}
let drained = async {
for reader in readers {
let _ = reader.await;
}
};
let _ = tokio::time::timeout(OUTPUT_DRAIN, drained).await;
exited.exited.cancel();
}
_ = kill.cancelled() => {
let _ = child.start_kill();
let _ = child.wait().await;
exited.reaped.cancel();
}
}
});
}
pub async fn wait_until_ready(
client: &OpenAiClient,
timeout: Duration,
exited: Option<ChildExit>,
loc: &'static Locale,
) -> Result<()> {
let deadline = tokio::time::Instant::now() + timeout;
loop {
if client.probe().await.is_ok() {
return Ok(());
}
if let Some(exit) = exited.as_ref().filter(|e| e.is_exited()) {
bail!("{}", exit.message(loc));
}
if tokio::time::Instant::now() >= deadline {
bail!(
"{}",
loc.tf(
"ui.err.managed.timeout",
&[("timeout", &format!("{timeout:?}"))]
)
);
}
let sleep = tokio::time::sleep(Duration::from_millis(500));
match &exited {
Some(ex) => {
tokio::select! {
_ = sleep => {}
_ = ex.wait() => {}
}
}
None => sleep.await,
}
}
}
async fn forward_lines<R>(reader: R, is_err: bool, said: Arc<Mutex<Said>>)
where
R: tokio::io::AsyncRead + Unpin,
{
let mut lines = BufReader::new(reader).lines();
while let Ok(Some(line)) = lines.next_line().await {
said.lock()
.unwrap_or_else(PoisonError::into_inner)
.record(&line, Instant::now());
if is_err {
tracing::warn!(target: "llama-server", "{line}");
} else {
tracing::info!(target: "llama-server", "{line}");
}
}
}
fn refusal_line(line: &str) -> Option<String> {
let clean = strip_ansi(line);
let clean = clean.trim();
clean.starts_with("error").then(|| clean.to_string())
}
fn error_record(line: &str) -> Option<String> {
let clean = strip_ansi(line);
let mut rest = clean.trim();
if let Some((stamp, after)) = rest.split_once(' ')
&& stamp.bytes().all(|b| b.is_ascii_digit() || b == b'.')
&& stamp.bytes().any(|b| b.is_ascii_digit())
{
rest = after.trim_start();
}
let message = rest.strip_prefix("E ")?;
let message = message.split_whitespace().collect::<Vec<_>>().join(" ");
(!message.is_empty()).then_some(message)
}
fn strip_ansi(line: &str) -> String {
let mut out = String::with_capacity(line.len());
let mut chars = line.chars();
while let Some(c) = chars.next() {
if c != '\u{1b}' {
out.push(c);
continue;
}
if chars.clone().next() == Some('[') {
chars.next();
for c in chars.by_ref() {
if ('@'..='~').contains(&c) {
break;
}
}
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
use crate::shared::i18n::{Lang, locale};
fn ru() -> &'static Locale {
locale(Lang::Ru)
}
fn base_cfg() -> ManagedConfig {
ManagedConfig {
binary: PathBuf::from("llama-server"),
model_path: None,
mmproj: None,
gpu_layers: 99,
context_size: 8192,
batch_size: None,
parallel: 1,
jinja: true,
reasoning_format: None,
embeddings: false,
no_mmap: false,
flash_attn: None,
spec_type: None,
draft_model: None,
draft_gpu_layers: None,
draft_n_max: None,
draft_n_min: None,
host: "127.0.0.1".into(),
port: 8000,
role: ManagedRole::Assistant,
extra_args: vec![],
kernel_cache: None,
}
}
#[test]
fn a_cpu_only_line_gets_the_measured_batch() {
let cfg = ManagedConfig {
gpu_layers: 0,
..base_cfg()
};
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
let at = |flag: &str| {
args.iter()
.position(|a| a == flag)
.map(|i| args[i + 1].clone())
};
assert_eq!(at("-b").as_deref(), Some("256"));
assert_eq!(at("-ub").as_deref(), Some("256"));
}
#[test]
fn a_gpu_line_is_byte_for_byte_what_it_was() {
let args = build_args(&base_cfg(), NoMmapSpelling::LoadMode);
let has_batch = args.iter().any(|a| a == "-b" || a == "-ub");
assert_eq!(has_batch, METAL_HOST, "{args:?}");
let partial = ManagedConfig {
gpu_layers: 20,
..base_cfg()
};
let partial_batch = build_args(&partial, NoMmapSpelling::LoadMode)
.iter()
.any(|a| a == "-b");
assert_eq!(partial_batch, METAL_HOST);
}
#[test]
fn auto_takes_the_knee_on_the_cpu_and_on_metal() {
assert_eq!(auto_batch(0, false), Some(CPU_BATCH), "the CPU");
assert_eq!(auto_batch(99, true), Some(CPU_BATCH), "Metal");
assert_eq!(auto_batch(0, true), Some(CPU_BATCH), "a Mac on its CPU");
assert_eq!(auto_batch(99, false), None, "a GPU elsewhere");
assert_eq!(auto_batch(20, false), None, "a partial offload");
}
#[test]
fn a_typed_batch_is_passed_as_is_with_the_micro_batch_clamped() {
let at = |cfg: &ManagedConfig, flag: &str| {
let args = build_args(cfg, NoMmapSpelling::LoadMode);
args.iter()
.position(|a| a == flag)
.map(|i| args[i + 1].clone())
};
let big = ManagedConfig {
batch_size: Some(1024),
..base_cfg()
};
assert_eq!(at(&big, "-b").as_deref(), Some("1024"));
assert_eq!(at(&big, "-ub").as_deref(), Some("512"));
let small = ManagedConfig {
batch_size: Some(128),
gpu_layers: 0,
..base_cfg()
};
assert_eq!(at(&small, "-b").as_deref(), Some("128"));
assert_eq!(at(&small, "-ub").as_deref(), Some("128"));
let cpu_default = ManagedConfig {
batch_size: Some(2048),
gpu_layers: 0,
..base_cfg()
};
assert_eq!(at(&cpu_default, "-b").as_deref(), Some("2048"));
assert_eq!(at(&cpu_default, "-ub").as_deref(), Some("512"));
}
#[test]
fn the_embedding_server_keeps_its_own_batch() {
let cfg = ManagedConfig {
embeddings: true,
gpu_layers: 0,
batch_size: Some(128),
context_size: 8192,
..base_cfg()
};
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
let values: Vec<&String> = args
.iter()
.enumerate()
.filter(|(i, a)| (*a == "-b" || *a == "-ub") && *i + 1 < args.len())
.map(|(i, _)| &args[i + 1])
.collect();
assert_eq!(values, vec!["8192", "8192"], "{args:?}");
}
#[test]
fn sessions_above_one_add_np_with_a_unified_pool() {
let one = build_args(&base_cfg(), NoMmapSpelling::LoadMode);
assert!(
!one.iter().any(|a| a == "-np" || a == "--kv-unified"),
"{one:?}"
);
let cfg = ManagedConfig {
parallel: 3,
..base_cfg()
};
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
let np = args.iter().position(|a| a == "-np").expect("-np present");
assert_eq!(args[np + 1], "3");
assert!(args.contains(&"--kv-unified".to_string()), "{args:?}");
let c = args.iter().position(|a| a == "-c").unwrap();
assert_eq!(args[c + 1], "8192");
}
#[test]
fn args_include_host_port_ngl_ctx_jinja() {
let args = build_args(&base_cfg(), NoMmapSpelling::LoadMode);
let host = args.iter().position(|a| a == "--host").unwrap();
assert_eq!(args[host + 1], "127.0.0.1");
let p = args.iter().position(|a| a == "--port").unwrap();
assert_eq!(args[p + 1], "8000");
let ngl = args.iter().position(|a| a == "-ngl").unwrap();
assert_eq!(args[ngl + 1], "99");
let c = args.iter().position(|a| a == "-c").unwrap();
assert_eq!(args[c + 1], "8192");
assert!(args.contains(&"--jinja".to_string()));
}
#[test]
fn args_for_model_with_reasoning() {
let cfg = ManagedConfig {
model_path: Some("gemma.gguf".into()),
reasoning_format: Some("auto".into()),
..base_cfg()
};
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
let m = args.iter().position(|a| a == "-m").unwrap();
assert_eq!(args[m + 1], "gemma.gguf");
let rf = args.iter().position(|a| a == "--reasoning-format").unwrap();
assert_eq!(args[rf + 1], "auto");
assert!(!args.contains(&"--embeddings".to_string()));
}
#[test]
fn embeddings_flag_and_no_jinja() {
let cfg = ManagedConfig {
embeddings: true,
jinja: false,
..base_cfg()
};
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
assert!(args.contains(&"--embeddings".to_string()));
assert!(!args.contains(&"--jinja".to_string()));
let ub = args.iter().position(|a| a == "-ub").expect("-ub present");
assert_eq!(args[ub + 1], cfg.context_size.to_string());
let b = args.iter().position(|a| a == "-b").expect("-b present");
assert_eq!(args[b + 1], cfg.context_size.to_string());
}
#[test]
fn no_batch_flags_for_non_embedding_server() {
let cfg = base_cfg();
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
let ub = args
.iter()
.position(|a| a == "-ub")
.map(|i| args[i + 1].clone());
let expected = METAL_HOST.then(|| CPU_BATCH.to_string());
assert_eq!(ub, expected, "{args:?}");
assert_ne!(ub, Some(cfg.context_size.to_string()));
}
#[test]
fn flash_attn_flag_present_only_when_set() {
assert!(
!build_args(&base_cfg(), NoMmapSpelling::LoadMode)
.contains(&"--flash-attn".to_string())
);
let cfg = ManagedConfig {
flash_attn: Some("on".into()),
..base_cfg()
};
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
let fa = args.iter().position(|a| a == "--flash-attn").unwrap();
assert_eq!(args[fa + 1], "on");
}
#[test]
fn spec_decoding_args_for_mtp_draft() {
let cfg = ManagedConfig {
spec_type: Some("draft-mtp".into()),
draft_model: Some("mtp.gguf".into()),
draft_gpu_layers: Some(99),
draft_n_max: Some(5),
draft_n_min: Some(1),
..base_cfg()
};
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
let st = args.iter().position(|a| a == "--spec-type").unwrap();
assert_eq!(args[st + 1], "draft-mtp");
let md = args.iter().position(|a| a == "-md").unwrap();
assert_eq!(args[md + 1], "mtp.gguf");
let ngld = args.iter().position(|a| a == "-ngld").unwrap();
assert_eq!(args[ngld + 1], "99");
let nmax = args.iter().position(|a| a == "--spec-draft-n-max").unwrap();
assert_eq!(args[nmax + 1], "5");
let nmin = args.iter().position(|a| a == "--spec-draft-n-min").unwrap();
assert_eq!(args[nmin + 1], "1");
}
#[test]
fn no_spec_args_by_default() {
let args = build_args(&base_cfg(), NoMmapSpelling::LoadMode);
assert!(!args.contains(&"--spec-type".to_string()));
assert!(!args.contains(&"-md".to_string()));
assert!(!args.contains(&"-ngld".to_string()));
}
#[test]
fn launch_missing_draft_model_file_errors_before_spawn() {
let cfg = ManagedConfig {
model_path: None,
draft_model: Some("definitely/missing/draft-xyz.gguf".into()),
..base_cfg()
};
let err = match ServerHandle::launch(&cfg, ru()) {
Err(e) => e,
Ok(_) => panic!("expected an error about a missing draft model file"),
};
assert!(err.to_string().contains("черновой модели"), "{err}");
}
#[test]
fn mmproj_flag_present_only_when_set() {
assert!(
!build_args(&base_cfg(), NoMmapSpelling::LoadMode).contains(&"--mmproj".to_string())
);
let cfg = ManagedConfig {
model_path: Some("gemma.gguf".into()),
mmproj: Some("mmproj-gemma.gguf".into()),
..base_cfg()
};
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
let p = args
.iter()
.position(|a| a == "--mmproj")
.expect("--mmproj present");
assert_eq!(args[p + 1], "mmproj-gemma.gguf");
}
#[test]
fn launch_missing_mmproj_file_errors_before_spawn() {
const MISSING: &str = "definitely/missing/mmproj-xyz.gguf";
let cfg = ManagedConfig {
model_path: None,
mmproj: Some(MISSING.into()),
..base_cfg()
};
let err = match ServerHandle::launch(&cfg, ru()) {
Err(e) => e,
Ok(_) => panic!("expected an error about a missing projector file"),
};
let expected = ru().tf("ui.err.managed.mmproj_not_found", &[("path", MISSING)]);
assert!(err.to_string().contains(&expected), "{err}");
}
#[test]
fn no_mmap_flag_present_only_when_enabled() {
for spelling in [NoMmapSpelling::LoadMode, NoMmapSpelling::NoMmap] {
let off = build_args(&base_cfg(), spelling);
assert!(!off.iter().any(|a| a == "--no-mmap" || a == "--load-mode"));
}
let cfg = ManagedConfig {
no_mmap: true,
..base_cfg()
};
let load_mode = build_args(&cfg, NoMmapSpelling::LoadMode);
let at = load_mode
.iter()
.position(|a| a == "--load-mode")
.expect("the current spelling");
assert_eq!(
load_mode[at + 1],
"none",
"`none` is llama.cpp's own word for no special loading mode"
);
assert!(
!load_mode.iter().any(|a| a == "--no-mmap"),
"a build that takes --load-mode is refused by the old flag, not helped by it"
);
assert!(
build_args(&cfg, NoMmapSpelling::NoMmap).contains(&"--no-mmap".to_string()),
"and a binary from before the rename takes only the old one"
);
}
#[test]
fn the_spelling_is_read_out_of_the_binarys_own_help() {
let current = "-lm, --load-mode MODE model loading mode (default: auto)\n\
- auto: mmap, unless a device does not support it\n\
- none: no special loading mode\n";
assert_eq!(spelling_from_help(current), NoMmapSpelling::LoadMode);
let older = " --no-mmap do not memory-map model (slower load but may reduce pageouts)\n\
--mlock force system to keep model in RAM\n";
assert_eq!(spelling_from_help(older), NoMmapSpelling::NoMmap);
assert_eq!(
spelling_from_help(""),
NoMmapSpelling::NoMmap,
"a help text that mentions neither predates the rename"
);
}
#[test]
fn a_binary_that_cannot_be_run_answers_the_current_spelling() {
assert_eq!(
no_mmap_spelling(std::path::Path::new("definitely/missing/llama-server-xyz")),
NoMmapSpelling::LoadMode
);
}
#[test]
fn base_url_uses_port() {
assert_eq!(base_cfg().base_url(), "http://127.0.0.1:8000/v1");
}
#[test]
fn launch_missing_model_file_errors_before_spawn() {
let cfg = ManagedConfig {
model_path: Some("definitely/missing/model-xyz.gguf".into()),
..base_cfg()
};
let err = match ServerHandle::launch(&cfg, ru()) {
Err(e) => e,
Ok(_) => panic!("expected an error about a missing model file"),
};
assert!(err.to_string().contains("файл модели"), "{err}");
}
fn preflight_error(dir: &std::path::Path, parts: &[&str], model: &str) -> String {
for part in parts {
std::fs::write(dir.join(part), b"gguf").expect("write a part");
}
let cfg = ManagedConfig {
binary: dir.join("no-such-llama-server"),
model_path: Some(dir.join(model).display().to_string()),
..base_cfg()
};
match ServerHandle::launch(&cfg, ru()) {
Err(e) => e.to_string(),
Ok(_) => panic!("launch cannot succeed: the binary does not exist"),
}
}
fn expect_about(dir: &std::path::Path, key: &str, file: &str) -> String {
ru().tf(key, &[("path", &dir.join(file).display().to_string())])
}
fn expect_spawn(dir: &std::path::Path) -> String {
expect_about(dir, "ui.err.managed.spawn", "no-such-llama-server")
}
#[test]
fn launch_missing_split_part_errors_before_spawn() {
let dir = tempfile::tempdir().expect("tempdir");
let (d, first) = (dir.path(), "model-00001-of-00003.gguf");
let err = preflight_error(d, &[first], first);
let missing = expect_about(
d,
"ui.err.managed.shard_missing",
"model-00002-of-00003.gguf",
);
assert!(err.contains(&missing), "{err}");
let all = ["model-00002-of-00003.gguf", "model-00003-of-00003.gguf"];
let err = preflight_error(d, &all, first);
assert!(err.contains(&expect_spawn(d)), "{err}");
}
#[test]
fn launch_from_a_later_split_part_points_at_the_first() {
let dir = tempfile::tempdir().expect("tempdir");
let d = dir.path();
let err = preflight_error(
d,
&["model-00002-of-00003.gguf"],
"model-00002-of-00003.gguf",
);
let first = expect_about(
d,
"ui.err.managed.shard_not_first",
"model-00001-of-00003.gguf",
);
assert!(err.contains(&first), "{err}");
}
#[test]
fn a_single_file_model_is_not_split_checked() {
let dir = tempfile::tempdir().expect("tempdir");
let err = preflight_error(dir.path(), &["gemma-4-it.gguf"], "gemma-4-it.gguf");
assert!(err.contains(&expect_spawn(dir.path())), "{err}");
}
#[test]
fn launch_error_is_localized() {
let cfg = ManagedConfig {
model_path: Some("definitely/missing/model-xyz.gguf".into()),
..base_cfg()
};
let en = match ServerHandle::launch(&cfg, locale(Lang::En)) {
Err(e) => e.to_string(),
Ok(_) => panic!("expected an error about a missing model file"),
};
assert!(en.contains("model file not found"), "{en}");
assert!(!en.chars().any(|c| ('а'..='я').contains(&c)), "{en}");
}
#[tokio::test]
async fn wait_until_ready_bails_on_early_exit() {
let client = OpenAiClient::new("http://127.0.0.1:1/v1");
let exited = ChildExit::exited_saying(Some("error: invalid argument: --x"));
let err = wait_until_ready(&client, Duration::from_secs(600), Some(exited), ru())
.await
.expect_err("expected an early-exit error");
assert!(
err.to_string()
.contains("отказался запускаться: error: invalid argument: --x"),
"{err}"
);
}
#[tokio::test]
async fn monitor_cancels_exited_when_child_dies() {
let child = exiting_child();
let kill = CancellationToken::new();
let exited = ChildExit::default();
spawn_monitor(child, kill, exited.clone(), vec![]);
tokio::time::timeout(Duration::from_secs(5), exited.wait())
.await
.expect("the monitor must arm exited on process exit");
assert!(exited.is_exited());
}
#[tokio::test]
#[ignore = "requires a local llama-server binary + model (MINDFORK_LLAMA_BIN, MINDFORK_MODEL)"]
async fn managed_cpu_line_launches_with_the_measured_batch_live() {
use crate::shared::api::EngineBackend;
let (Ok(bin), Ok(model)) = (
std::env::var("MINDFORK_LLAMA_BIN"),
std::env::var("MINDFORK_MODEL"),
) else {
eprintln!("skip: MINDFORK_LLAMA_BIN / MINDFORK_MODEL not set");
return;
};
let cfg = ManagedConfig {
binary: bin.into(),
model_path: Some(model),
gpu_layers: 0,
context_size: 2048,
port: 18124,
..base_cfg()
};
let args = build_args(&cfg, NoMmapSpelling::LoadMode);
eprintln!("the line: {}", args.join(" "));
let at = |flag: &str| {
args.iter()
.position(|a| a == flag)
.map(|i| args[i + 1].clone())
};
assert_eq!(at("-b").as_deref(), Some("256"));
assert_eq!(at("-ub").as_deref(), Some("256"));
let handle = ServerHandle::launch(&cfg, locale(Lang::En)).expect("launch");
let client = OpenAiClient::new(handle.base_url());
wait_until_ready(
&client,
Duration::from_secs(600),
Some(handle.exited()),
locale(Lang::En),
)
.await
.expect("the server should come up on the CPU line");
let slots = client.parallel_slots().await;
let window = client.context_budget().await;
eprintln!("live CPU launch: slots {slots:?}, window {window:?}");
assert_eq!(
slots,
Some(4),
"the server's own default: four unified slots"
);
assert_eq!(window, Some(2048));
}
#[tokio::test]
#[ignore = "requires a local llama-server binary + model (MINDFORK_LLAMA_BIN, MINDFORK_MODEL)"]
async fn managed_sessions_launch_three_slots_over_one_pool_live() {
use crate::shared::api::EngineBackend;
let (Ok(bin), Ok(model)) = (
std::env::var("MINDFORK_LLAMA_BIN"),
std::env::var("MINDFORK_MODEL"),
) else {
eprintln!("skip: MINDFORK_LLAMA_BIN / MINDFORK_MODEL not set");
return;
};
let cfg = ManagedConfig {
binary: bin.into(),
model_path: Some(model),
parallel: 3,
context_size: 4096,
port: 18123,
..base_cfg()
};
let handle = ServerHandle::launch(&cfg, locale(Lang::En)).expect("launch");
let client = OpenAiClient::new(handle.base_url());
wait_until_ready(
&client,
Duration::from_secs(600),
Some(handle.exited()),
locale(Lang::En),
)
.await
.expect("the server should come up");
let slots = client.parallel_slots().await;
let window = client.context_budget().await;
eprintln!("live managed launch: slots {slots:?}, per-slot window {window:?}");
assert_eq!(slots, Some(3));
assert_eq!(
window,
Some(4096),
"unified pool: every slot may use the whole -c"
);
}
#[tokio::test]
#[ignore = "requires a local llama-server binary + model (MINDFORK_LLAMA_BIN, MINDFORK_MODEL)"]
async fn managed_extra_args_reach_a_real_server_live() {
let (Ok(bin), Ok(model)) = (
std::env::var("MINDFORK_LLAMA_BIN"),
std::env::var("MINDFORK_MODEL"),
) else {
eprintln!("skip: MINDFORK_LLAMA_BIN / MINDFORK_MODEL not set");
return;
};
let en = locale(Lang::En);
let cfg = |extra: &[&str]| ManagedConfig {
binary: bin.clone().into(),
model_path: Some(model.clone()),
context_size: 2048,
port: 18125,
extra_args: extra.iter().map(|a| a.to_string()).collect(),
..base_cfg()
};
let started = std::time::Instant::now();
let handle = ServerHandle::launch(&cfg(&["--n-cpu-mo", "4"]), en).expect("launch");
let client = OpenAiClient::new(handle.base_url());
let err = wait_until_ready(&client, Duration::from_secs(600), Some(handle.exited()), en)
.await
.expect_err("a misspelled flag cannot come up");
eprintln!("control arm ({:?}): {err}", started.elapsed());
assert!(
err.to_string()
.contains("refused to start: error: invalid argument: --n-cpu-mo"),
"{err}"
);
drop(handle);
struct Unset(&'static str);
impl Drop for Unset {
fn drop(&mut self) {
unsafe { std::env::remove_var(self.0) };
}
}
unsafe { std::env::set_var("LLAMA_API_KEY", "set-for-some-other-tool") };
let _unset = Unset("LLAMA_API_KEY");
let line = ["--alias", "mindfork-extra-args", "--n-cpu-moe", "4"];
let handle = ServerHandle::launch(&cfg(&line), en).expect("launch");
let client = OpenAiClient::new(handle.base_url());
wait_until_ready(&client, Duration::from_secs(600), Some(handle.exited()), en)
.await
.expect("the server should come up with the raw arguments");
let models = reqwest::get(format!("{}/models", handle.base_url()))
.await
.expect("GET /v1/models");
let status = models.status();
let body = models.text().await.unwrap_or_default();
eprintln!("live: /v1/models {status}: {body}");
assert!(
status.is_success(),
"{status}: the inherited key reached the child"
);
assert!(
body.contains("\"mindfork-extra-args\""),
"the alias is the catalogue's id"
);
}
#[test]
fn the_launched_batch_follows_the_launch_line() {
assert_eq!(launched_batch(Some(1024), 99), 1024, "typed wins");
assert_eq!(launched_batch(Some(1024), 0), 1024);
assert_eq!(launched_batch(None, 0), CPU_BATCH, "the CPU auto");
let gpu = if METAL_HOST {
CPU_BATCH
} else {
LLAMA_DEFAULT_BATCH
};
assert_eq!(
launched_batch(None, 99),
gpu,
"a GPU host at its defaults runs the server's, Metal the knee"
);
}
#[test]
fn the_prefill_hold_is_said_only_when_worth_it() {
use crate::shared::api::contract::Prefill;
let slow = Prefill {
tokens: 1800,
ms: 20_000,
};
assert_eq!(prefill_hold(LLAMA_DEFAULT_BATCH, slow), Some(23));
assert_eq!(prefill_hold(CPU_BATCH, slow), None);
let fast = Prefill {
tokens: 1800,
ms: 900,
};
assert_eq!(prefill_hold(LLAMA_DEFAULT_BATCH, fast), None);
let short = Prefill {
tokens: 100,
ms: 5_000,
};
assert_eq!(prefill_hold(LLAMA_DEFAULT_BATCH, short), None);
let mid = Prefill {
tokens: 1500,
ms: 10_000,
};
assert_eq!(prefill_hold(1024, mid), Some(7));
assert_eq!(
prefill_hold(LLAMA_DEFAULT_BATCH, Prefill { tokens: 0, ms: 0 }),
None
);
}
#[test]
fn a_refused_extra_argument_stops_the_launch_before_the_spawn() {
let dir = tempfile::tempdir().expect("tempdir");
let launch = |extra: &[&str]| {
let cfg = ManagedConfig {
binary: dir.path().join("no-such-llama-server"),
extra_args: extra.iter().map(|a| a.to_string()).collect(),
..base_cfg()
};
match ServerHandle::launch(&cfg, locale(Lang::En)) {
Err(e) => e.to_string(),
Ok(_) => panic!("launch cannot succeed: the binary does not exist"),
}
};
let err = launch(&["--n-cpu-moe", "20", "--tools", "all"]);
assert!(
err.starts_with("Extra arguments: --tools") && err.contains("External"),
"{err}"
);
let err = launch(&["--n-cpu-moe", "20"]);
assert!(err.contains("spawning llama-server"), "{err}");
}
#[test]
fn the_launch_judges_the_line_by_its_role() {
let dir = tempfile::tempdir().expect("tempdir");
let launch = |role: ManagedRole| {
let cfg = ManagedConfig {
binary: dir.path().join("no-such-llama-server"),
role,
extra_args: vec!["-c".into(), "4096".into()],
..base_cfg()
};
ServerHandle::launch(&cfg, locale(Lang::En))
.err()
.map(|e| e.to_string())
.unwrap_or_default()
};
assert!(launch(ManagedRole::Assistant).contains("Context (-c)"));
assert!(launch(ManagedRole::Embedder).contains("spawning llama-server"));
}
#[test]
fn the_command_removes_the_refused_flags_environment() {
let removed = |role: ManagedRole| -> Vec<String> {
let cfg = ManagedConfig { role, ..base_cfg() };
server_command(&cfg, &[])
.as_std()
.get_envs()
.filter(|(_, v)| v.is_none())
.map(|(k, _)| k.to_string_lossy().into_owned())
.collect()
};
let chat = removed(ManagedRole::Assistant);
for var in [
"LLAMA_API_KEY",
"LLAMA_ARG_TOOLS",
"LLAMA_ARG_HF_REPO",
"HF_TOKEN",
] {
assert!(chat.iter().any(|v| v == var), "{var} not removed: {chat:?}");
}
assert!(!chat.iter().any(|v| v == "LLAMA_ARG_N_CPU_MOE"));
assert!(
!removed(ManagedRole::Embedder)
.iter()
.any(|v| v == "LLAMA_ARG_EMBEDDINGS")
);
}
#[test]
fn the_refusal_line_is_the_argument_parsers_error() {
for line in [
"error: invalid argument: --bogus-flag",
"error while handling argument \"-c\": invalid stoi argument",
"\u{1b}[31merror: invalid argument: --x\u{1b}[0m\r",
] {
let got = refusal_line(line).expect(line);
assert!(
got.starts_with("error") && !got.contains('\u{1b}'),
"{got:?}"
);
assert!(!got.ends_with('\r'), "{got:?}");
}
for line in [
"0.00.047.856 E srv load_model: failed to load model, 'fake.gguf'",
"-c, --ctx-size N size of the prompt context",
"to show complete usage, run with -h",
"",
] {
assert_eq!(refusal_line(line), None, "{line}");
}
assert_eq!(
strip_ansi("\u{1b}[34m0.00.000.787\u{1b}[0m I srv"),
"0.00.000.787 I srv"
);
}
#[tokio::test]
async fn an_exit_is_announced_with_the_line_it_printed() {
let child = shell(
"echo error: invalid argument: --n-cpu-mo 1>&2 & exit 1",
"echo 'error: invalid argument: --n-cpu-mo' >&2; exit 1",
)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true)
.spawn()
.expect("spawn of a short-lived process");
let (_kill, exited) = supervise(child);
tokio::time::timeout(Duration::from_secs(10), exited.wait())
.await
.expect("the monitor must arm exited on process exit");
let msg = exited.message(locale(Lang::En));
assert!(
msg.starts_with("llama-server refused to start: error: invalid argument: --n-cpu-mo"),
"{msg}"
);
}
#[tokio::test]
async fn the_exit_waits_for_the_output_to_be_read() {
let exited = ChildExit::default();
let said = exited.said.clone();
let reader = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(300)).await;
said.lock().unwrap().refusal = Some("error: late".into());
});
spawn_monitor(
exiting_child(),
CancellationToken::new(),
exited.clone(),
vec![reader],
);
tokio::time::timeout(Duration::from_secs(5), exited.wait())
.await
.expect("the monitor must arm exited");
assert_eq!(
exited.message(locale(Lang::En)),
"llama-server refused to start: error: late"
);
}
#[tokio::test]
async fn an_endless_reader_does_not_hold_the_exit() {
let exited = ChildExit::default();
let reader = tokio::spawn(std::future::pending::<()>());
spawn_monitor(
exiting_child(),
CancellationToken::new(),
exited.clone(),
vec![reader],
);
tokio::time::timeout(OUTPUT_DRAIN * 5, exited.wait())
.await
.expect("the drain must be bounded");
}
fn shell(windows: &str, unix: &str) -> Command {
let mut c;
if cfg!(windows) {
c = Command::new("cmd");
c.args(["/C", windows]);
} else {
c = Command::new("sh");
c.args(["-c", unix]);
}
c
}
fn exiting_child() -> Child {
shell("exit", "exit 0")
.stdout(Stdio::null())
.stderr(Stdio::null())
.kill_on_drop(true)
.spawn()
.expect("spawn of a short-lived process")
}
#[tokio::test]
async fn the_reader_keeps_the_first_refusal() {
let said = Arc::new(Mutex::new(Said::default()));
let output = b"0.00.000.8 I srv init\nerror: invalid argument: --a\nusage:\nerror: b\n";
forward_lines(&output[..], true, said.clone()).await;
assert_eq!(
said.lock().unwrap().refusal.as_deref(),
Some("error: invalid argument: --a")
);
}
#[test]
fn a_silent_exit_keeps_the_generic_message() {
let en = locale(Lang::En);
assert_eq!(
ChildExit::exited_saying(None).message(en),
en.t("ui.err.managed.early_exit")
);
}
const CUDA_ABORT: &[&str] = &[
"0.43.034.880 E CUDA error: the provided PTX was compiled with an unsupported toolchain.",
"0.43.034.891 E current device: 0, in function ggml_cuda_kernel_can_use_pdl at /__w/llama.cpp/llama.cpp/ggml/src/ggml-cuda/common.cuh:1670",
"0.43.034.892 E cudaFuncGetAttributes(&attr, kernel)",
"/workspace/mindfork/data/llama/cuda-13.4-b11327/libggml-base.so.0(ggml_abort+0x15b)[0x7324f5147b3b]",
];
const BIND_FAILURE: &[&str] = &[
"0.01.423.360 W srv llama_server: security: no API key is set and CORS allows all origins",
"0.01.423.595 E srv start: couldn't bind HTTP server socket, hostname: 127.0.0.1, port: 8001",
"0.01.423.602 I srv operator(): operator(): cleaning up before exit...",
"0.01.424.063 E srv llama_server: exiting due to HTTP server error",
];
fn said_after(lines: &[&str], start: Instant, reaped_after: Duration) -> Said {
let mut said = Said::default();
let mut at = start;
for line in lines {
said.record(line, at);
at += Duration::from_millis(1);
}
said.ended = Some(at + reaped_after);
said
}
fn exit_with(said: Said) -> ChildExit {
let exit = ChildExit::default();
*exit.said.lock().unwrap() = said;
exit.exited.cancel();
exit
}
#[test]
fn an_error_record_is_its_message() {
assert_eq!(
error_record(CUDA_ABORT[0]).as_deref(),
Some("CUDA error: the provided PTX was compiled with an unsupported toolchain.")
);
assert_eq!(
error_record(BIND_FAILURE[1]).as_deref(),
Some("srv start: couldn't bind HTTP server socket, hostname: 127.0.0.1, port: 8001")
);
assert_eq!(
error_record("\u{1b}[34m0.00.047.856\u{1b}[0m E srv load_model: failed\r").as_deref(),
Some("srv load_model: failed")
);
assert_eq!(
error_record("E llama_model_load: error").as_deref(),
Some("llama_model_load: error")
);
for line in [
BIND_FAILURE[0],
BIND_FAILURE[2],
CUDA_ABORT[3],
"/__w/llama.cpp/llama.cpp/ggml/src/ggml-cuda/ggml-cuda.cu:109: CUDA error",
"error: invalid argument: --bogus",
"0.00.000.001 E ",
"Every line that starts with an E word",
"",
] {
assert_eq!(error_record(line), None, "{line}");
}
}
#[test]
fn the_cause_is_the_first_record_of_the_burst() {
let t = Instant::now();
let cuda = said_after(CUDA_ABORT, t, Duration::from_millis(900));
assert_eq!(
cuda.logged_cause(),
Some("CUDA error: the provided PTX was compiled with an unsupported toolchain.")
);
let bind = said_after(BIND_FAILURE, t, Duration::from_millis(80));
assert_eq!(
bind.logged_cause(),
Some("srv start: couldn't bind HTTP server socket, hostname: 127.0.0.1, port: 8001")
);
}
#[test]
fn an_old_burst_does_not_explain_a_later_exit() {
let t = Instant::now();
let request = "1.00.000.000 E srv send_error: task id = 5, error: the request exceeds the available context size";
let mut said = Said::default();
said.record(request, t);
let crash = t + Duration::from_secs(3600);
said.record(CUDA_ABORT[0], crash);
said.record(CUDA_ABORT[1], crash + Duration::from_millis(1));
said.ended = Some(crash + Duration::from_secs(1));
assert_eq!(
said.logged_cause(),
Some("CUDA error: the provided PTX was compiled with an unsupported toolchain.")
);
let mut silent = Said::default();
silent.record(request, t);
silent.ended = Some(t + CAUSE_WINDOW + Duration::from_secs(1));
assert_eq!(
silent.logged_cause(),
None,
"a silent crash keeps the generic message"
);
silent.ended = Some(t + CAUSE_WINDOW - Duration::from_secs(1));
assert!(silent.logged_cause().is_some());
}
#[test]
fn the_burst_gap_splits_bursts() {
let t = Instant::now();
let mut said = Said::default();
said.record("0.1 E first", t);
said.record("0.2 E second", t + BURST_GAP - Duration::from_millis(1));
assert_eq!(said.logged_cause(), Some("first"));
said.record("0.3 E third", t + BURST_GAP * 3);
assert_eq!(said.logged_cause(), Some("third"));
}
#[test]
fn the_exit_message_names_the_logged_cause() {
let en = locale(Lang::En);
let t = Instant::now();
let msg = exit_with(said_after(CUDA_ABORT, t, Duration::from_millis(900))).message(en);
assert!(
msg.starts_with("llama-server stopped with an error: CUDA error: the provided PTX"),
"{msg}"
);
assert!(
msg.contains("mindfork llama setup --backend cuda-12") && msg.contains("nvidia-smi"),
"the CUDA case names the way out: {msg}"
);
let ru_msg = exit_with(said_after(CUDA_ABORT, t, Duration::ZERO)).message(ru());
assert!(ru_msg.contains("драйвер"), "{ru_msg}");
let msg = exit_with(said_after(BIND_FAILURE, t, Duration::ZERO)).message(en);
assert_eq!(
msg,
"llama-server stopped with an error: srv start: couldn't bind HTTP server socket, hostname: 127.0.0.1, port: 8001"
);
let mut both = said_after(BIND_FAILURE, t, Duration::ZERO);
both.record("error: invalid argument: --x", t);
assert_eq!(
exit_with(both).message(en),
"llama-server refused to start: error: invalid argument: --x",
"the parser's refusal comes first"
);
}
#[tokio::test]
async fn an_exit_is_announced_with_the_record_it_logged() {
let child = shell(
"echo 0.43.034.880 E CUDA error: the provided PTX was compiled with an unsupported toolchain. 1>&2 & exit 1",
"echo '0.43.034.880 E CUDA error: the provided PTX was compiled with an unsupported toolchain.' >&2; exit 1",
)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true)
.spawn()
.expect("spawn of a short-lived process");
let (_kill, exited) = supervise(child);
tokio::time::timeout(Duration::from_secs(10), exited.wait())
.await
.expect("the monitor must arm exited on process exit");
let msg = exited.message(locale(Lang::En));
assert!(
msg.starts_with("llama-server stopped with an error: CUDA error: the provided PTX"),
"{msg}"
);
}
fn on_a_taken_port(dir: &std::path::Path) -> (ManagedConfig, std::net::TcpListener) {
let held = std::net::TcpListener::bind("127.0.0.1:0").expect("bind a stranger");
let cfg = ManagedConfig {
binary: dir.join("no-such-llama-server"),
port: held.local_addr().unwrap().port(),
role: ManagedRole::Embedder,
..base_cfg()
};
(cfg, held)
}
#[test]
fn a_port_someone_listens_on_is_taken() {
let dir = tempfile::tempdir().expect("tempdir");
let (cfg, held) = on_a_taken_port(dir.path());
assert!(port_taken(&cfg));
drop(held);
assert!(!port_taken(&cfg), "the control: a free port is not taken");
}
#[test]
fn the_ledger_refuses_a_port_a_stranger_holds() {
let dir = tempfile::tempdir().expect("tempdir");
let (cfg, held) = on_a_taken_port(dir.path());
let ledger = PortLedger::default();
let err = match ledger.launch(&cfg, ru()) {
Err(e) => e.to_string(),
Ok(_) => panic!("a taken port must be refused"),
};
assert_eq!(err, port_busy_message(&cfg, ru()));
assert!(
err.contains(&cfg.port.to_string())
&& err.contains("Эмбеддинги")
&& err.contains("Порт"),
"{err}"
);
drop(held);
let err = match ledger.launch(&cfg, ru()) {
Err(e) => e.to_string(),
Ok(_) => panic!("the binary does not exist"),
};
assert_eq!(err, expect_spawn(dir.path()));
}
#[test]
fn the_ledger_names_a_missing_model_before_a_taken_port() {
let dir = tempfile::tempdir().expect("tempdir");
let (taken, _held) = on_a_taken_port(dir.path());
let cfg = ManagedConfig {
model_path: Some(dir.path().join("absent.gguf").display().to_string()),
..taken
};
let err = match PortLedger::default().launch(&cfg, ru()) {
Err(e) => e.to_string(),
Ok(_) => panic!("the model is absent"),
};
assert_eq!(
err,
expect_about(dir.path(), "ui.err.managed.model_not_found", "absent.gguf")
);
}
#[test]
fn the_ledger_launches_onto_its_own_leaving_server() {
let dir = tempfile::tempdir().expect("tempdir");
let (cfg, _held) = on_a_taken_port(dir.path());
let claim = |released: bool, reaped: bool| {
let exit = ChildExit::default();
let token = CancellationToken::new();
if released {
token.cancel();
}
if reaped {
exit.reaped.cancel();
}
Claim {
port: cfg.port,
released: token,
exit,
}
};
let launched = |claim: Claim| {
let ledger = PortLedger::default();
ledger.claims.lock().unwrap().push(claim);
match ledger.launch(&cfg, ru()) {
Err(e) => e.to_string(),
Ok(_) => panic!("the binary does not exist"),
}
};
assert_eq!(
launched(claim(true, false)),
expect_spawn(dir.path()),
"leaving: ours"
);
let busy = port_busy_message(&cfg, ru());
assert_eq!(
launched(claim(true, true)),
busy,
"reaped: the holder is someone else"
);
assert_eq!(
launched(claim(false, false)),
busy,
"a live server of ours is no exemption"
);
}
#[tokio::test]
async fn reaped_is_armed_on_exit_and_on_kill() {
let exited = ChildExit::default();
spawn_monitor(
exiting_child(),
CancellationToken::new(),
exited.clone(),
vec![],
);
tokio::time::timeout(Duration::from_secs(5), exited.reaped.cancelled())
.await
.expect("an exit arms reaped");
assert!(
exited.said.lock().unwrap().ended.is_some(),
"an exit is stamped, or every old record would read as its cause"
);
let long = shell("ping -n 30 127.0.0.1 >NUL", "sleep 30")
.stdout(Stdio::null())
.stderr(Stdio::null())
.kill_on_drop(true)
.spawn()
.expect("spawn of a long-lived process");
let (kill, exited) = (CancellationToken::new(), ChildExit::default());
spawn_monitor(long, kill.clone(), exited.clone(), vec![]);
assert!(!exited.reaped.is_cancelled());
kill.cancel();
tokio::time::timeout(Duration::from_secs(5), exited.reaped.cancelled())
.await
.expect("a kill arms reaped");
assert!(!exited.is_exited(), "a kill is not an exit on its own");
}
const NO_LIBGOMP: &str = "/workspace/mindfork/data/llama/cuda-12.8-b11332//llama-server: error while loading shared libraries: libgomp.so.1: cannot open shared object file: No such file or directory";
const NO_LIBGGML_BASE: &str = "dyld[2653]: Library not loaded: @rpath/libggml-base.0.dylib";
#[test]
fn the_loader_line_names_the_missing_library() {
assert_eq!(missing_library(NO_LIBGOMP).as_deref(), Some("libgomp.so.1"));
assert_eq!(
missing_library("./x: error while loading shared libraries: libcudart.so.12: cannot open shared object file: No such file or directory").as_deref(),
Some("libcudart.so.12")
);
assert_eq!(
missing_library(NO_LIBGGML_BASE).as_deref(),
Some("libggml-base.0.dylib")
);
assert_eq!(
missing_library("dyld: Library not loaded: @rpath/libllama.dylib").as_deref(),
Some("libllama.dylib")
);
for line in [
"error: invalid argument: --x",
CUDA_ABORT[0],
"0.00.000.001 I srv load_model: loading model",
"error while loading shared libraries",
"srv: Library not loaded: @rpath/libggml.dylib",
"dyld[1]: Library not loaded: ",
"",
] {
assert_eq!(missing_library(line), None, "{line}");
}
}
#[test]
fn a_missing_library_is_said_with_the_way_out() {
let en = locale(Lang::En);
let omp = missing_library_message("libgomp.so.1", en);
assert!(
omp.contains("libgomp.so.1") && omp.contains("apt-get install libgomp1"),
"{omp}"
);
let other = missing_library_message("libfoo.so.3", en);
assert!(
other.contains("libfoo.so.3") && !other.contains("libgomp1"),
"{other}"
);
assert!(missing_library_message("libgomp.so.1", ru()).contains("libgomp1"));
}
#[test]
fn the_exit_message_names_the_missing_library() {
let en = locale(Lang::En);
let t = Instant::now();
let mut said = said_after(BIND_FAILURE, t, Duration::ZERO);
said.record(NO_LIBGOMP, t);
assert_eq!(
exit_with(said).message(en),
missing_library_message("libgomp.so.1", en)
);
}
#[tokio::test]
async fn an_exit_is_announced_with_the_library_the_loader_missed() {
let child = shell(
"echo ./llama-server: error while loading shared libraries: libgomp.so.1: cannot open shared object file: No such file or directory 1>&2 & exit 127",
"echo './llama-server: error while loading shared libraries: libgomp.so.1: cannot open shared object file: No such file or directory' >&2; exit 127",
)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true)
.spawn()
.expect("spawn of a short-lived process");
let (_kill, exited) = supervise(child);
tokio::time::timeout(Duration::from_secs(10), exited.wait())
.await
.expect("the monitor must arm exited on process exit");
let en = locale(Lang::En);
assert_eq!(
exited.message(en),
missing_library_message("libgomp.so.1", en)
);
}
#[test]
fn the_kernel_cache_yields_to_one_the_user_named() {
let ours = Path::new("data/cuda-cache");
assert_eq!(kernel_cache_for(Some(ours), None), Some(ours));
assert_eq!(
kernel_cache_for(Some(ours), Some(OsStr::new(""))),
Some(ours)
);
assert_eq!(
kernel_cache_for(Some(ours), Some(OsStr::new("/elsewhere"))),
None
);
assert_eq!(kernel_cache_for(None, None), None);
}
#[test]
fn the_command_carries_the_kernel_cache() {
let given = |cmd: &Command| {
cmd.as_std()
.get_envs()
.find(|(k, _)| *k == CUDA_CACHE_PATH)
.and_then(|(_, v)| v.map(PathBuf::from))
};
let dir = tempfile::tempdir().expect("tempdir");
let cache = dir.path().join("data").join("cuda-cache");
let cfg = ManagedConfig {
kernel_cache: Some(cache.clone()),
..base_cfg()
};
let cmd = server_command(&cfg, &[]);
if std::env::var_os(CUDA_CACHE_PATH).is_some_and(|v| !v.is_empty()) {
assert_eq!(given(&cmd), None);
return;
}
assert_eq!(given(&cmd), Some(cache.clone()));
assert!(cache.is_dir(), "created for the driver");
assert_eq!(given(&server_command(&base_cfg(), &[])), None);
}
}