1mod admin;
36mod anthropic;
37mod attribution;
38mod budget;
39mod cache_admin;
40mod cancel;
41mod chat_template;
42mod completion;
43mod conversations;
44mod decode_task;
45mod embeddings;
46mod generate;
47mod grammar_request;
48mod health;
49mod journal;
50mod json_mode;
51mod limits;
52mod loaded;
53mod mcp;
54mod model;
55mod openai_extra;
56mod output;
57mod policy;
58mod reasoning_tokens;
59mod rerank;
60mod response_cache;
61pub(crate) mod responses;
62mod resume;
63mod sample_step;
64mod sampling_knobs;
65mod security;
66mod serving;
67mod session;
68mod sse;
69mod stats;
70mod stop;
71mod stream_events;
72mod tasks;
73mod tool_grammar;
74mod unsupported_sampling;
75mod utf8_stream;
76
77use std::cell::RefCell;
78use std::convert::Infallible;
79use std::fmt;
80use std::net::{IpAddr, Ipv4Addr, SocketAddr};
81use std::path::PathBuf;
82use std::rc::Rc;
83use std::str::FromStr;
84use std::sync::{Arc, Mutex, MutexGuard};
85use std::time::Duration;
86
87use axum::{
88 extract::State,
89 http::StatusCode,
90 response::sse::{Event, Sse},
91 response::{IntoResponse, Response},
92 routing::{get, post},
93 Json, Router,
94};
95use clap::{Parser, ValueEnum};
96use serde::{Deserialize, Serialize};
97
98use ferrox_core::cache::KvBlockPool;
99use ferrox_models::kimi_tokenizer::KimiTokenizer;
100use ferrox_models::sampling::SamplingParams;
101use ferrox_models::tokenizer::StopTokens;
102use ferrox_models::{Decoder, Gemma4Engine, KimiEngine, MlaEngine, PrefixCache};
103use generate::{FinishReason, GenerationParams};
104pub(crate) use loaded::{ActiveModel, Loaded};
105use model::ServerTokenizer;
106use rerank::encoder_endpoints;
107use response_cache::{CacheKey, ResponseCache};
108use sampling_knobs::SamplingKnobs;
109
110#[derive(Parser, Debug, PartialEq)]
114#[command(
121 name = "ferrox-server",
122 about = "OpenAI-compatible Ferrox inference server"
123)]
124pub struct ServerArgs {
125 #[arg(short = 'm', long = "model", value_name = "FILE")]
127 model: Option<String>,
128
129 #[arg(
137 long = "hf-repo",
138 visible_alias = "hf",
139 value_name = "REPO[:QUANT]",
140 conflicts_with = "model"
141 )]
142 hf_repo: Option<String>,
143
144 #[arg(long = "hf-file", value_name = "FILE", requires = "hf_repo")]
149 hf_file: Option<String>,
150
151 #[arg(short = 'c', long = "ctx-size", value_name = "N")]
157 ctx_size: Option<usize>,
158
159 #[arg(long = "api-key", value_name = "KEY")]
162 api_key: Option<String>,
163
164 #[arg(long = "api-key-file", value_name = "PATH", conflicts_with = "api_key")]
169 api_key_file: Option<std::path::PathBuf>,
170
171 #[arg(long = "alias", visible_alias = "model-alias", value_name = "NAME")]
174 alias: Option<String>,
175
176 #[arg(long = "ctk", visible_alias = "cache-type-k", value_name = "TYPE")]
179 ctk: Option<String>,
180
181 #[arg(long = "jinja", default_value_t = false)]
187 jinja: bool,
188
189 #[arg(long = "no-jinja", default_value_t = false)]
192 no_jinja: bool,
193
194 #[arg(long = "no-warmup", default_value_t = false)]
196 no_warmup: bool,
197
198 #[arg(long = "flash-attn", visible_alias = "fa", value_name = "MODE", num_args = 0..=1, default_missing_value = "auto")]
202 flash_attn: Option<String>,
203
204 #[arg(long, value_name = "HOST")]
206 host: Option<IpAddr>,
207
208 #[arg(long, value_name = "PORT")]
213 port: Option<u16>,
214
215 #[arg(short = 't', long = "threads", value_name = "N")]
217 threads: Option<usize>,
218
219 #[arg(
221 long = "device",
222 visible_alias = "dev",
223 value_name = "DEVICE",
224 ignore_case = true
225 )]
226 device: Option<OffloadDevice>,
227
228 #[arg(long = "list-devices", default_value_t = false)]
230 list_devices: bool,
231
232 #[arg(
237 long = "n-gpu-layers",
238 visible_aliases = ["gpu-layers", "ngl"],
239 value_name = "N"
240 )]
241 n_gpu_layers: Option<GpuLayers>,
242
243 #[arg(long = "mcp-config", value_name = "PATH")]
245 mcp_config: Option<PathBuf>,
246
247 #[arg(long = "exit-on-stdin-close", default_value_t = false)]
255 exit_on_stdin_close: bool,
256
257 #[arg(
260 long = "cont-batching",
261 visible_aliases = ["continuous-batching", "cb"],
262 default_value_t = false
263 )]
264 cont_batching: bool,
265
266 #[arg(
269 long = "no-cont-batching",
270 default_value_t = false,
271 conflicts_with = "cont_batching"
272 )]
273 no_cont_batching: bool,
274
275 #[arg(long = "parallel", visible_alias = "np", value_name = "N")]
279 parallel: Option<usize>,
280
281 #[arg(long = "allow-multiple-instances", default_value_t = false)]
286 allow_multiple_instances: bool,
287}
288
289impl ServerArgs {
290 pub fn parse_llama_style<I>(argv: I) -> Self
297 where
298 I: IntoIterator<Item = String>,
299 {
300 Self::parse_from(rewrite_llama_style_argv(argv.into_iter().collect()))
301 }
302}
303
304#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
305enum OffloadDevice {
306 Auto,
307 None,
308 Cpu,
309 Metal,
310 Cuda,
311}
312
313#[derive(Debug, Clone, Copy, PartialEq, Eq)]
314enum GpuLayers {
315 Auto,
316 All,
317 Count(u32),
318}
319
320impl GpuLayers {
321 fn offload_enabled(self) -> bool {
322 !matches!(self, Self::Count(0))
323 }
324}
325
326impl FromStr for GpuLayers {
327 type Err = String;
328
329 fn from_str(value: &str) -> Result<Self, Self::Err> {
330 match value {
331 "auto" => Ok(Self::Auto),
332 "all" => Ok(Self::All),
333 _ => value
334 .parse::<u32>()
335 .map(Self::Count)
336 .map_err(|_| "expected 0, a positive integer, 'auto', or 'all'".into()),
337 }
338 }
339}
340
341impl fmt::Display for GpuLayers {
342 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
343 match self {
344 Self::Auto => f.write_str("auto"),
345 Self::All => f.write_str("all"),
346 Self::Count(value) => value.fmt(f),
347 }
348 }
349}
350
351pub const BUILT_WITH_METAL: bool = cfg!(feature = "metal");
360
361pub const BUILT_WITH_CUDA: bool = cfg!(feature = "cuda");
364
365fn rewrite_llama_style_argv(args: Vec<String>) -> Vec<String> {
366 args.into_iter()
367 .map(|arg| match arg.as_str() {
368 "-ngl" => "--n-gpu-layers".into(),
369 "-dev" => "--device".into(),
370 "-cb" => "--cont-batching".into(),
371 "-np" => "--parallel".into(),
372 "-hf" => "--hf-repo".into(),
377 "-hff" => "--hf-file".into(),
378 _ => arg,
379 })
380 .collect()
381}
382
383fn print_available_devices() {
384 println!("Available devices:");
385 println!(" CPU");
386
387 let metal = ferrox_metal::MetalProfile::detect();
388 if let Some(name) = metal.device_name {
389 println!(" Metal: {name}");
390 }
391
392 let cuda = ferrox_cuda::HardwareProfile::detect();
393 if cuda.cuda_available {
394 let name = cuda.cuda_device_name.as_deref().unwrap_or("unknown device");
395 println!(" CUDA: {name}");
396 if cuda.cuda_device_count > 1 {
397 println!(" ({} devices detected)", cuda.cuda_device_count);
398 }
399 }
400}
401
402fn cli_bind_addr(args: &ServerArgs, env_addr: Option<&str>) -> Option<String> {
403 if args.host.is_none() && args.port.is_none() {
404 return None;
405 }
406
407 let existing = env_addr.and_then(|value| value.parse::<SocketAddr>().ok());
408 let host = args
409 .host
410 .or_else(|| existing.map(|addr| addr.ip()))
411 .unwrap_or(IpAddr::V4(Ipv4Addr::LOCALHOST));
412 let port = args
413 .port
414 .or_else(|| existing.map(|addr| addr.port()))
415 .unwrap_or(8383);
416 Some(SocketAddr::new(host, port).to_string())
417}
418
419fn resolve_hf_repo(spec: &str, file: Option<&str>) -> anyhow::Result<String> {
425 let mut hf = ferrox_models::hub::HfRef::parse(spec);
426 if let Some(f) = file {
427 hf.file = Some(f.to_string());
428 }
429 eprintln!(
430 "ferrox: resolving {} on the Hub{}",
431 hf.repo,
432 hf.quant
433 .as_deref()
434 .map(|q| format!(" ({q})"))
435 .unwrap_or_default()
436 );
437
438 let mut last = std::time::Instant::now();
439 let mut draw = move |done: u64, total: Option<u64>| {
440 if last.elapsed() < std::time::Duration::from_millis(200) {
441 return;
442 }
443 last = std::time::Instant::now();
444 let mib = done as f64 / 1024.0 / 1024.0;
445 match total {
446 Some(t) if t > 0 => {
447 eprint!(
448 "\r {mib:>9.1} MiB {:5.1}%",
449 (done as f64 / t as f64) * 100.0
450 )
451 }
452 _ => eprint!("\r {mib:>9.1} MiB"),
453 }
454 };
455
456 let (path, downloaded) = hf
457 .ensure_local(&mut draw)
458 .map_err(|e| anyhow::anyhow!("{e}"))?;
459 if downloaded {
460 eprintln!();
461 eprintln!("ferrox: downloaded {}", path.display());
462 } else {
463 eprintln!("ferrox: using cached {}", path.display());
464 }
465 Ok(path.to_string_lossy().into_owned())
466}
467
468fn apply_cli_overrides(args: &ServerArgs) -> anyhow::Result<()> {
469 if let Some(model) = &args.model {
470 unsafe { std::env::set_var("FERROX_MODEL_PATH", model) };
472 }
473 if let Some(spec) = &args.hf_repo {
474 let path = resolve_hf_repo(spec, args.hf_file.as_deref())?;
475 unsafe { std::env::set_var("FERROX_MODEL_PATH", &path) };
477 }
478 if let Some(n) = args.ctx_size {
479 if n == 0 {
480 anyhow::bail!("--ctx-size must be greater than zero");
481 }
482 unsafe { std::env::set_var("FERROX_CB_MAX_CONTEXT", n.to_string()) };
484 }
485 if let Some(key) = &args.api_key {
486 unsafe { std::env::set_var("FERROX_API_KEY", key) };
488 }
489 if let Some(path) = &args.api_key_file {
490 let key = std::fs::read_to_string(path)
491 .map_err(|e| anyhow::anyhow!("reading --api-key-file {}: {e}", path.display()))?;
492 let key = key.trim();
493 if key.is_empty() {
494 anyhow::bail!(
495 "--api-key-file {} is empty: an empty key would leave every route open, \
496 which is the opposite of what passing the flag asked for",
497 path.display()
498 );
499 }
500 unsafe { std::env::set_var("FERROX_API_KEY", key) };
502 }
503 if let Some(alias) = &args.alias {
504 unsafe { std::env::set_var("FERROX_MODEL_NAME", alias) };
506 }
507 if let Some(ctk) = &args.ctk {
508 unsafe { std::env::set_var("FERROX_CTK", ctk.trim()) };
510 }
511 if args.no_jinja {
516 anyhow::bail!(
517 "--no-jinja: ferrox has no template-free mode. It compiles and evaluates the GGUF's \
518 own tokenizer.chat_template, which is what llama.cpp's --jinja turns on, and there \
519 is no sniffing fallback to switch to. Use --no-cnv on `ferrox run` for a raw \
520 completion"
521 );
522 }
523 if let Some(mode) = &args.flash_attn {
524 let mode = mode.trim().to_ascii_lowercase();
525 if mode == "off" || mode == "disabled" || mode == "0" {
526 anyhow::bail!(
527 "--flash-attn off: fused attention is a backend property here, not a per-run \
528 switch. Set FERROX_METAL_ATTN=0 to take the unfused Metal path, or --device cpu"
529 );
530 }
531 }
532
533 if let Some(addr) = cli_bind_addr(args, std::env::var("FERROX_ADDR").ok().as_deref()) {
534 unsafe { std::env::set_var("FERROX_ADDR", addr) };
536 }
537
538 if let Some(threads) = args.threads {
539 if threads == 0 {
540 anyhow::bail!("--threads must be greater than zero");
541 }
542 unsafe {
544 std::env::set_var("FERROX_CPU_THREADS", threads.to_string());
545 std::env::set_var("RAYON_NUM_THREADS", threads.to_string());
546 }
547 }
548
549 if args.device.is_none() && args.n_gpu_layers.is_none() {
550 } else {
552 let layers = args.n_gpu_layers.unwrap_or(GpuLayers::Auto);
553 let device = if layers.offload_enabled() {
554 args.device.unwrap_or(OffloadDevice::Auto)
555 } else {
556 OffloadDevice::None
557 };
558
559 match device {
560 OffloadDevice::None | OffloadDevice::Cpu => unsafe {
561 std::env::set_var("FERROX_METAL", "0");
562 std::env::set_var("FERROX_METAL_ATTN", "0");
563 std::env::set_var("FERROX_CUDA", "0");
564 },
565 OffloadDevice::Auto => unsafe {
566 std::env::set_var("FERROX_METAL", "auto");
567 std::env::set_var("FERROX_CUDA", "auto");
568 if std::env::var_os("FERROX_METAL_ATTN").is_none() {
569 std::env::set_var("FERROX_METAL_ATTN", "1");
570 }
571 },
572 OffloadDevice::Metal => {
573 #[cfg(not(feature = "metal"))]
574 {
575 anyhow::bail!(
576 "Metal requested but this binary was built without --features metal"
577 );
578 }
579 #[cfg(feature = "metal")]
580 {
581 if !ferrox_metal::MetalProfile::detect().available {
582 anyhow::bail!("Metal requested but no Metal device is available");
583 }
584 unsafe {
585 std::env::set_var("FERROX_METAL", "1");
586 if std::env::var_os("FERROX_METAL_ATTN").is_none() {
587 std::env::set_var("FERROX_METAL_ATTN", "1");
588 }
589 std::env::set_var("FERROX_CUDA", "0");
590 }
591 }
592 }
593 OffloadDevice::Cuda => {
594 #[cfg(not(feature = "cuda"))]
595 {
596 anyhow::bail!(
597 "CUDA requested but this binary was built without --features cuda"
598 );
599 }
600 #[cfg(feature = "cuda")]
601 {
602 if !ferrox_cuda::HardwareProfile::detect().cuda_available {
603 anyhow::bail!("CUDA requested but no CUDA device is available");
604 }
605 unsafe {
606 std::env::set_var("FERROX_CUDA", "1");
607 std::env::set_var("FERROX_METAL", "0");
608 std::env::set_var("FERROX_METAL_ATTN", "0");
609 }
610 }
611 }
612 }
613 }
614
615 if let Some(n) = args.parallel {
616 if n == 0 {
617 anyhow::bail!("--parallel must be greater than zero");
618 }
619 unsafe { std::env::set_var("FERROX_CB_MAX_SEQS", n.to_string()) };
621 }
622
623 if args.cont_batching {
624 unsafe { std::env::set_var("FERROX_CONTINUOUS_BATCHING", "1") };
626 } else if args.no_cont_batching {
627 unsafe { std::env::set_var("FERROX_CONTINUOUS_BATCHING", "0") };
629 } else if args.parallel.is_some() {
630 unsafe { std::env::set_var("FERROX_CONTINUOUS_BATCHING", "1") };
633 }
634
635 Ok(())
636}
637
638#[allow(clippy::large_enum_variant)] pub(crate) enum Model {
647 Gguf(GgufModel),
648 Kimi(KimiModel),
649 Mla(MlaModel),
650 Gemma4(Gemma4Model),
651 Glm52(Glm52Model),
652}
653
654pub(crate) struct GgufModel {
655 decoder: Arc<Decoder>,
656 tokenizer: Arc<ServerTokenizer>,
657 stop_tokens: StopTokens,
658 bos_id: Option<usize>,
659 is_synthetic: bool,
660 chat_template: chat_template::PromptTemplate,
661}
662
663pub(crate) struct KimiModel {
664 engine: KimiEngine,
665 tokenizer: KimiTokenizer,
666 stop_tokens: StopTokens,
667 chat_template: chat_template::PromptTemplate,
668}
669
670pub(crate) struct MlaModel {
671 engine: MlaEngine,
672 tokenizer: ServerTokenizer,
673 stop_tokens: StopTokens,
674 bos_id: Option<usize>,
675 name: String,
676 chat_template: chat_template::PromptTemplate,
677}
678
679pub(crate) struct Gemma4Model {
680 engine: Gemma4Engine,
681 tokenizer: ServerTokenizer,
682 stop_tokens: StopTokens,
683 bos_id: Option<usize>,
684 name: String,
685 chat_template: chat_template::PromptTemplate,
686}
687
688pub(crate) struct Glm52Model {
689 engine: ferrox_models::Glm52Engine,
690 tokenizer: ServerTokenizer,
691 stop_tokens: StopTokens,
692 bos_id: Option<usize>,
693 name: String,
694 chat_template: chat_template::PromptTemplate,
695}
696
697impl Model {
698 pub(crate) fn chat_template(&self) -> chat_template::PromptTemplate {
699 match self {
700 Model::Gguf(m) => m.chat_template.clone(),
701 Model::Kimi(m) => m.chat_template.clone(),
702 Model::Mla(m) => m.chat_template.clone(),
703 Model::Gemma4(m) => m.chat_template.clone(),
704 Model::Glm52(m) => m.chat_template.clone(),
705 }
706 }
707
708 fn is_synthetic(&self) -> bool {
713 match self {
714 Model::Gguf(m) => m.is_synthetic,
715 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => false,
716 }
717 }
718
719 fn tokenizer_kind(&self) -> &'static str {
720 match self {
721 Model::Gguf(m) => m.tokenizer.kind(),
722 Model::Kimi(_) => "kimi-tiktoken-bpe",
723 Model::Mla(m) => m.tokenizer.kind(),
724 Model::Gemma4(m) => m.tokenizer.kind(),
725 Model::Glm52(m) => m.tokenizer.kind(),
726 }
727 }
728
729 fn expert_store_stats(&self) -> Option<ferrox_core::expert_store::ExpertStoreStats> {
733 match self {
734 Model::Gguf(m) => m.decoder.expert_store_stats(),
735 Model::Kimi(m) => m.engine.weights.expert_store_stats(),
736 Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
737 }
738 }
739
740 pub(crate) fn name(&self) -> &str {
741 match self {
742 Model::Gguf(m) => m.decoder.config.name,
743 Model::Kimi(_) => "kimi-k3",
744 Model::Mla(m) => m.name.as_str(),
745 Model::Gemma4(m) => m.name.as_str(),
746 Model::Glm52(m) => m.name.as_str(),
747 }
748 }
749
750 pub(crate) fn encode(&self, text: &str) -> Vec<usize> {
751 match self {
752 Model::Gguf(m) => m.tokenizer.encode(text),
753 Model::Kimi(m) => m
754 .tokenizer
755 .encode(text)
756 .into_iter()
757 .map(|id| id as usize)
758 .collect(),
759 Model::Mla(m) => m.tokenizer.encode(text),
760 Model::Gemma4(m) => m.tokenizer.encode(text),
761 Model::Glm52(m) => m.tokenizer.encode(text),
762 }
763 }
764
765 pub(crate) fn bos_id(&self) -> Option<usize> {
774 match self {
775 Model::Gguf(m) => m.bos_id,
776 Model::Kimi(_) => None,
777 Model::Mla(m) => m.bos_id,
778 Model::Gemma4(m) => m.bos_id,
779 Model::Glm52(m) => m.bos_id,
780 }
781 }
782
783 pub(crate) fn decode(&self, ids: &[usize]) -> String {
784 match self {
785 Model::Gguf(m) => m.tokenizer.decode(ids),
786 Model::Kimi(m) => {
787 let ids32: Vec<u32> = ids.iter().map(|&id| id as u32).collect();
788 m.tokenizer.decode(&ids32)
789 }
790 Model::Mla(m) => m.tokenizer.decode(ids),
791 Model::Gemma4(m) => m.tokenizer.decode(ids),
792 Model::Glm52(m) => m.tokenizer.decode(ids),
793 }
794 }
795
796 pub(crate) fn embed_tokens(&self, tokens: &[usize]) -> Option<Vec<Vec<f32>>> {
799 match self {
800 Model::Gguf(m) => {
801 let mut caches: Vec<_> = (0..m.decoder.layers.len())
802 .map(|_| {
803 ferrox_core::cache::KvCache::new(
804 m.decoder.config.n_kv_heads,
805 m.decoder.config.head_dim,
806 )
807 })
808 .collect();
809 Some(m.decoder.forward_hidden_batch(tokens, 0, &mut caches))
810 }
811 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
812 }
813 }
814
815 pub(crate) fn vocab_size(&self) -> Option<usize> {
816 match self {
817 Model::Gguf(m) => Some(m.decoder.config.vocab_size),
818 Model::Kimi(m) => Some(m.tokenizer.vocab_size()),
819 Model::Mla(m) => Some(ferrox_models::Engine::vocab_size(&m.engine)),
820 Model::Gemma4(m) => Some(ferrox_models::Engine::vocab_size(&m.engine)),
821 Model::Glm52(m) => Some(ferrox_models::Engine::vocab_size(&m.engine)),
822 }
823 }
824
825 fn has_real_vocabulary(&self) -> bool {
832 match self {
833 Model::Gguf(m) => !matches!(*m.tokenizer, model::ServerTokenizer::Byte),
834 Model::Kimi(_) => true,
835 Model::Mla(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
836 Model::Gemma4(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
837 Model::Glm52(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
838 }
839 }
840}
841
842impl ferrox_models::dry::DryVocab for Model {
848 fn n_tokens(&self) -> usize {
849 self.vocab_size().unwrap_or(0)
850 }
851
852 fn detokenize(&self, token: usize) -> String {
853 self.decode(&[token])
854 }
855
856 fn tokenize(&self, text: &str) -> Vec<usize> {
857 self.encode(text)
858 }
859}
860
861pub(crate) struct AppState {
862 pub(crate) embedding: Option<Arc<ferrox_models::EmbeddingModel>>,
876 active: std::sync::RwLock<Option<Arc<ActiveModel>>>,
895 pub(crate) load_in_progress: std::sync::atomic::AtomicBool,
899 pub(crate) tasks: Arc<tasks::TaskRegistry>,
901 pub(crate) cancels: Arc<cancel::CancelRegistry>,
905 pub(crate) stats: stats::Stats,
908 pub(crate) streams: resume::StreamRegistry,
911 pub(crate) model_dir: Option<PathBuf>,
913 response_cache: Mutex<ResponseCache>,
917 pub(crate) kv_pool: Option<generate::KvPoolConfig>,
929 pub(crate) paged_kv: Option<generate::PagedKvConfig>,
942 pub(crate) prefix_cache: Option<Arc<Mutex<PrefixCache>>>,
952 sessions: session::SessionStore,
958 requests_total: std::sync::atomic::AtomicU64,
959 request_errors_total: std::sync::atomic::AtomicU64,
960 started_at: std::time::Instant,
961 last_request_ms: std::sync::atomic::AtomicU64,
966 detection: Arc<health::Detection>,
968 mcp: Option<mcp::LoadedMcpConfig>,
970 pub(crate) continuous_batching_enabled: bool,
974 pub(crate) metal_private_decode_gate: Option<Arc<std::sync::Mutex<()>>>,
978 loading_model: Mutex<Option<String>>,
982 last_load_error: Mutex<Option<(String, String)>>,
986 pub(crate) serving: Mutex<crate::stats::ServingStats>,
991 pub(crate) maintenance: Mutex<crate::policy::maintenance::MaintenanceGate>,
996 pub(crate) footprint:
1002 Mutex<crate::policy::footprint::ProbeCache<crate::policy::footprint::Footprint>>,
1003 pub(crate) started_unix: u64,
1013}
1014
1015pub(crate) const FOOTPRINT_TTL_MS: u64 = 2_000;
1021
1022impl AppState {
1023 pub(crate) fn active(&self) -> Option<Arc<ActiveModel>> {
1027 self.active
1028 .read()
1029 .unwrap_or_else(|p| p.into_inner())
1030 .clone()
1031 }
1032
1033 pub(crate) fn require_active(&self) -> Result<Arc<ActiveModel>, ApiError> {
1039 self.active().ok_or_else(|| {
1040 (
1041 StatusCode::SERVICE_UNAVAILABLE,
1042 Json(serde_json::json!({"error": {
1043 "message": "no model is loaded; POST /admin/models/load with an id from \
1044 GET /admin/models",
1045 "type": "model_not_loaded"
1046 }})),
1047 )
1048 })
1049 }
1050
1051 pub(crate) fn require_model(&self) -> Result<Arc<Model>, ApiError> {
1059 Ok(Arc::clone(self.require_active()?.generative()?))
1060 }
1061
1062 pub(crate) fn swap_active(&self, next: Option<Arc<ActiveModel>>) -> Option<Arc<ActiveModel>> {
1070 let mut guard = self.active.write().unwrap_or_else(|p| p.into_inner());
1071 std::mem::replace(&mut *guard, next)
1072 }
1073
1074 fn mark_request_finished(&self) {
1078 let ms = self.started_at.elapsed().as_millis().min(u64::MAX as u128) as u64;
1079 self.last_request_ms
1080 .store(ms, std::sync::atomic::Ordering::Relaxed);
1081 }
1082
1083 pub(crate) fn uptime(&self) -> Duration {
1084 self.started_at.elapsed()
1085 }
1086
1087 pub(crate) fn requests_total(&self) -> u64 {
1088 self.requests_total
1089 .load(std::sync::atomic::Ordering::Relaxed)
1090 }
1091
1092 pub(crate) fn errors_total(&self) -> u64 {
1093 self.request_errors_total
1094 .load(std::sync::atomic::Ordering::Relaxed)
1095 }
1096
1097 pub(crate) fn cache_stats(&self) -> response_cache::CacheStats {
1098 lock_cache(&self.response_cache).stats()
1099 }
1100
1101 pub(crate) fn last_request_age_seconds(&self) -> Option<f64> {
1104 let last = self
1105 .last_request_ms
1106 .load(std::sync::atomic::Ordering::Relaxed);
1107 (last > 0)
1108 .then(|| self.uptime().as_secs_f64() - (last as f64 / 1000.0))
1109 .map(|age| age.max(0.0))
1110 }
1111
1112 pub(crate) fn loading_model_id(&self) -> Option<String> {
1113 self.loading_model
1114 .lock()
1115 .unwrap_or_else(|p| p.into_inner())
1116 .clone()
1117 }
1118
1119 pub(crate) fn set_loading_model(&self, id: Option<String>) {
1120 *self.loading_model.lock().unwrap_or_else(|p| p.into_inner()) = id;
1121 }
1122
1123 pub(crate) fn last_load_error(&self) -> Option<(String, String)> {
1124 self.last_load_error
1125 .lock()
1126 .unwrap_or_else(|p| p.into_inner())
1127 .clone()
1128 }
1129
1130 pub(crate) fn set_last_load_error(&self, error: Option<(String, String)>) {
1131 *self
1132 .last_load_error
1133 .lock()
1134 .unwrap_or_else(|p| p.into_inner()) = error;
1135 }
1136
1137 pub(crate) fn active_model_name(&self) -> Option<String> {
1146 self.active().map(|a| a.name().to_string())
1147 }
1148
1149 pub(crate) fn embedding_model(&self) -> Option<Arc<ferrox_models::EmbeddingModel>> {
1159 self.embedding
1160 .clone()
1161 .or_else(|| self.active().and_then(|a| a.encoder().map(Arc::clone)))
1162 }
1163
1164 pub(crate) fn embedding_model_name(&self) -> Option<String> {
1168 match self.embedding_model() {
1169 Some(e) => Some(e.name().to_string()),
1170 None => self.active_model_name(),
1171 }
1172 }
1173
1174 pub(crate) fn record_request(&self, record: stats::Record<'_>) {
1175 self.stats.record(stats::entry(record));
1176 }
1177}
1178
1179fn lock_cache(cache: &Mutex<ResponseCache>) -> MutexGuard<'_, ResponseCache> {
1184 cache
1185 .lock()
1186 .unwrap_or_else(|poisoned| poisoned.into_inner())
1187}
1188
1189#[derive(Debug, Clone, Deserialize)]
1190#[serde(untagged)]
1191pub(crate) enum MessageContent {
1192 Text(String),
1193 Parts(Vec<ContentPart>),
1194}
1195
1196#[derive(Debug, Clone, Deserialize)]
1197struct ContentPart {
1198 #[serde(rename = "type")]
1199 kind: String,
1200 #[serde(default)]
1201 text: Option<String>,
1202 #[serde(default)]
1203 image_url: Option<serde_json::Value>,
1204}
1205
1206impl MessageContent {
1207 fn as_text(&self) -> String {
1208 match self {
1209 Self::Text(s) => s.clone(),
1210 Self::Parts(parts) => parts
1211 .iter()
1212 .filter_map(|p| p.text.as_deref())
1213 .collect::<Vec<_>>()
1214 .join(""),
1215 }
1216 }
1217
1218 fn has_image(&self) -> bool {
1219 match self {
1220 Self::Text(_) => false,
1221 Self::Parts(parts) => parts
1222 .iter()
1223 .any(|p| p.kind == "image_url" || p.image_url.is_some()),
1224 }
1225 }
1226}
1227
1228#[derive(Debug, Clone, Deserialize)]
1229pub(crate) struct ChatMessage {
1230 pub(crate) role: String,
1231 #[serde(default)]
1235 pub(crate) content: Option<MessageContent>,
1236 #[serde(default)]
1240 pub(crate) tool_calls: Option<Vec<ToolCallIn>>,
1241 #[serde(default)]
1246 #[allow(dead_code)]
1247 pub(crate) tool_call_id: Option<String>,
1248 #[serde(default, alias = "reasoning")]
1264 pub(crate) reasoning_content: Option<String>,
1265}
1266
1267impl ChatMessage {
1268 fn rendered_content(&self) -> String {
1275 let mut out = self
1276 .content
1277 .as_ref()
1278 .map(MessageContent::as_text)
1279 .unwrap_or_default();
1280 if let Some(calls) = &self.tool_calls {
1281 for call in calls {
1282 out.push_str(&format!(
1283 "<tool_call>{{\"name\": \"{}\", \"arguments\": {}}}</tool_call>",
1284 call.function.name, call.function.arguments
1285 ));
1286 }
1287 }
1288 out
1289 }
1290}
1291
1292#[derive(Debug, Clone, Deserialize)]
1293pub(crate) struct ToolCallIn {
1294 #[serde(default)]
1295 #[allow(dead_code)]
1296 id: String,
1297 #[serde(rename = "type", default)]
1298 #[allow(dead_code)]
1299 kind: String,
1300 function: ToolCallFunctionIn,
1301}
1302
1303#[derive(Debug, Clone, Deserialize)]
1304struct ToolCallFunctionIn {
1305 name: String,
1306 arguments: String,
1311}
1312
1313#[derive(Debug, Clone, Deserialize)]
1316struct ToolDef {
1317 #[serde(rename = "type", default)]
1318 #[allow(dead_code)]
1319 kind: String,
1320 function: ToolFunctionDef,
1321}
1322
1323#[derive(Debug, Clone, Deserialize)]
1324struct ToolFunctionDef {
1325 name: String,
1326 #[serde(default)]
1327 description: Option<String>,
1328 #[serde(default)]
1329 parameters: Option<serde_json::Value>,
1330}
1331
1332#[derive(Debug, Clone, Deserialize)]
1343#[serde(untagged)]
1344enum ToolChoice {
1345 Mode(String),
1346 Specific(serde_json::Value),
1347}
1348
1349#[derive(Deserialize)]
1352#[serde(untagged)]
1353enum StopParam {
1354 One(String),
1355 Many(Vec<String>),
1356}
1357
1358#[derive(Deserialize)]
1359struct ChatCompletionRequest {
1360 model: String,
1361 messages: Vec<ChatMessage>,
1362 #[serde(default = "default_max_tokens")]
1363 max_tokens: usize,
1364 #[serde(default)]
1365 temperature: Option<f32>,
1366 #[serde(default)]
1367 top_p: Option<f32>,
1368 #[serde(default)]
1373 min_p: Option<f32>,
1374 #[serde(default)]
1375 top_k: Option<usize>,
1376 #[serde(default)]
1377 repetition_penalty: Option<f32>,
1378 #[serde(flatten)]
1382 extra_samplers: crate::sampling_knobs::ExtraSamplerFields,
1383 #[serde(default)]
1384 seed: Option<u64>,
1385 #[serde(default)]
1386 stop: Option<StopParam>,
1387 #[serde(default)]
1388 stream: Option<bool>,
1389 #[serde(default)]
1402 stream_resumable: Option<bool>,
1403 #[serde(default)]
1413 ignore_eos: Option<bool>,
1414 #[serde(default)]
1415 tools: Vec<ToolDef>,
1416 #[serde(default)]
1417 tool_choice: Option<ToolChoice>,
1418 #[serde(default)]
1426 chat_template_kwargs: Option<serde_json::Map<String, serde_json::Value>>,
1427 #[serde(default)]
1435 reasoning_effort: Option<String>,
1436 #[serde(default)]
1440 thinking: Option<ThinkingSwitch>,
1441 #[serde(default)]
1446 session_id: Option<String>,
1447 #[serde(default)]
1449 logprobs: Option<bool>,
1450 #[serde(default)]
1451 top_logprobs: Option<u32>,
1452 #[serde(default)]
1453 n: Option<u32>,
1454 #[serde(default)]
1455 presence_penalty: Option<f32>,
1456 #[serde(default)]
1457 frequency_penalty: Option<f32>,
1458 #[serde(default)]
1459 response_format: Option<serde_json::Value>,
1460 #[serde(default)]
1467 logit_bias: Option<serde_json::Value>,
1468 #[serde(default)]
1478 samplers: Option<serde_json::Value>,
1479 #[serde(default)]
1489 grammar: Option<String>,
1490}
1491
1492const DEFAULT_CHAT_MAX_TOKENS: usize = 32_768;
1504
1505#[derive(Debug, Clone, Deserialize)]
1507pub(crate) struct ThinkingSwitch {
1508 #[serde(rename = "type")]
1509 pub(crate) kind: String,
1510}
1511
1512const THINKING_KWARG_KEYS: [&str; 4] = [
1516 "enable_thinking",
1517 "thinking",
1518 "thinking_mode",
1519 "reasoning_effort",
1520];
1521
1522const DISABLE_EFFORTS: [&str; 2] = ["none", "off"];
1526
1527fn default_max_tokens() -> usize {
1528 DEFAULT_CHAT_MAX_TOKENS
1529}
1530
1531impl ChatCompletionRequest {
1532 fn sampling_knobs(&self) -> Result<SamplingKnobs, ApiError> {
1541 let mut knobs = SamplingKnobs {
1542 temperature: self.temperature,
1543 top_p: self.top_p,
1544 min_p: self.min_p,
1545 top_k: self.top_k,
1546 repetition_penalty: self.repetition_penalty,
1547 presence_penalty: self.presence_penalty,
1548 frequency_penalty: self.frequency_penalty,
1549 penalty_last_n: None,
1553 sampler_order: unsupported_sampling::parse_sampler_order(
1554 self.samplers.as_ref(),
1555 "/v1/chat/completions",
1556 )?,
1557 ..SamplingKnobs::default()
1558 };
1559 self.extra_samplers.apply(&mut knobs);
1560 Ok(knobs)
1561 }
1562
1563 fn sampling_params(
1564 &self,
1565 model: crate::sampling_knobs::SamplerModel<'_>,
1566 ) -> Result<SamplingParams, ApiError> {
1567 self.sampling_knobs()?.resolve(model).map_err(|e| {
1568 unsupported_feature(&format!("`dry_multiplier` on /v1/chat/completions: {e}"))
1569 })
1570 }
1571
1572 fn stop_sequences(&self) -> Vec<String> {
1573 self.stop
1574 .as_ref()
1575 .map(|s| match s {
1576 StopParam::One(v) => vec![v.clone()],
1577 StopParam::Many(v) => v.clone(),
1578 })
1579 .unwrap_or_default()
1580 }
1581
1582 fn tools_active(&self) -> bool {
1587 !self.tools.is_empty()
1588 && !matches!(&self.tool_choice, Some(ToolChoice::Mode(m)) if m == "none")
1589 }
1590
1591 fn forced_tool_choice(&self) -> Result<Option<tool_grammar::Forced<'_>>, ApiError> {
1603 match &self.tool_choice {
1604 Some(ToolChoice::Mode(m)) if m == "required" => Ok(Some(tool_grammar::Forced::Any)),
1605 Some(ToolChoice::Specific(value)) => {
1606 let name = value
1610 .get("function")
1611 .and_then(|f| f.get("name"))
1612 .or_else(|| value.get("name"))
1613 .and_then(|n| n.as_str());
1614 match name {
1615 Some(name) => Ok(Some(tool_grammar::Forced::Named(name))),
1616 None => Err(invalid_request(
1617 "tool_choice must be \"auto\", \"none\", \"required\", or an object with \
1618 function.name",
1619 "tool_choice",
1620 )),
1621 }
1622 }
1623 _ => Ok(None),
1624 }
1625 }
1626
1627 fn tool_specs(&self) -> Vec<tool_grammar::ToolSpec<'_>> {
1629 self.tools
1630 .iter()
1631 .map(|t| tool_grammar::ToolSpec {
1632 name: &t.function.name,
1633 parameters: t.function.parameters.as_ref(),
1634 })
1635 .collect()
1636 }
1637
1638 fn thinking_direction(&self) -> Option<bool> {
1681 if let Some(switch) = &self.thinking {
1682 return match switch.kind.trim().to_ascii_lowercase().as_str() {
1683 "disabled" => Some(false),
1684 "enabled" => Some(true),
1685 _ => None,
1688 };
1689 }
1690 let effort = self.reasoning_effort.as_ref()?;
1691 DISABLE_EFFORTS
1692 .contains(&effort.trim().to_ascii_lowercase().as_str())
1693 .then_some(false)
1694 }
1695
1696 fn resolve_template_kwargs(
1697 &self,
1698 template: &chat_template::PromptTemplate,
1699 ) -> serde_json::Map<String, serde_json::Value> {
1700 let mut kwargs = self.chat_template_kwargs.clone().unwrap_or_default();
1701 let caller_steered = THINKING_KWARG_KEYS.iter().any(|k| kwargs.contains_key(*k));
1705
1706 if !caller_steered {
1707 match self.thinking_direction() {
1708 Some(false) => {
1709 for (k, v) in crate::policy::effort::thinking_off_kwargs() {
1710 kwargs.insert(k, v);
1711 }
1712 return kwargs;
1715 }
1716 Some(true) => {
1717 for (k, v) in crate::policy::effort::thinking_on_kwargs() {
1718 kwargs.insert(k, v);
1719 }
1720 }
1721 None => {}
1722 }
1723 if let Some(effort) = &self.reasoning_effort {
1724 kwargs
1725 .entry("reasoning_effort".to_string())
1726 .or_insert_with(|| serde_json::json!(effort));
1727 }
1728 }
1729
1730 let offered: Vec<serde_json::Value> = if self.tools_active() {
1731 self.tools.iter().map(chat_template::tool_json).collect()
1732 } else {
1733 Vec::new()
1734 };
1735 let thinking = crate::policy::effort::resolve_thinking_mode(Some(&kwargs), Some(&offered));
1736 if thinking == crate::policy::effort::ThinkingMode::Thinking {
1737 for (k, v) in crate::policy::effort::thinking_on_kwargs() {
1738 kwargs.entry(k).or_insert(v);
1739 }
1740 }
1741 match crate::policy::effort::sanitize_effort(&mut kwargs, template.efforts()) {
1742 crate::policy::effort::EffortMapping::Mapped(to) => {
1743 tracing::debug!("reasoning_effort quantized to {}", to.as_str());
1744 }
1745 crate::policy::effort::EffortMapping::Dropped => {
1746 tracing::debug!(
1747 "reasoning_effort dropped: this checkpoint's template grades no gear close \
1748 enough, so its own default applies"
1749 );
1750 }
1751 crate::policy::effort::EffortMapping::Unchanged => {}
1752 }
1753 crate::policy::effort::broadcast_effort_spellings(&mut kwargs);
1754 kwargs
1755 }
1756
1757 fn validate_supported_fields(&self) -> Result<(), ApiError> {
1760 if self.max_tokens == 0 {
1767 return Err(invalid_request(
1768 "max_tokens must be at least 1",
1769 "max_tokens",
1770 ));
1771 }
1772 if let Some(switch) = &self.thinking {
1777 let kind = switch.kind.trim().to_ascii_lowercase();
1778 if kind != "enabled" && kind != "disabled" {
1779 return Err(invalid_request(
1780 "thinking.type must be \"enabled\" or \"disabled\"",
1781 "thinking.type",
1782 ));
1783 }
1784 }
1785 for msg in &self.messages {
1786 if msg.content.as_ref().is_some_and(MessageContent::has_image) {
1787 return Err(unsupported_feature(
1788 "image_url content parts are not implemented (multimodal/VL deferred, see docs/API.md)",
1789 ));
1790 }
1791 }
1792 if self.logprobs == Some(true) || self.top_logprobs.is_some() {
1793 return Err(unsupported_feature(
1794 "logprobs / top_logprobs are not implemented yet (see docs/API.md)",
1795 ));
1796 }
1797 if self.n.is_some_and(|n| n > 1) {
1798 return Err(unsupported_feature(
1799 "n > 1 is not implemented (single completion only)",
1800 ));
1801 }
1802 unsupported_sampling::refuse_logit_bias(self.logit_bias.as_ref(), "/v1/chat/completions")?;
1803 unsupported_sampling::parse_sampler_order(self.samplers.as_ref(), "/v1/chat/completions")?;
1807 let stated_grammar =
1822 grammar_request::for_request(self.grammar.as_deref(), self.response_format.as_ref())?;
1823 if let Some(forced) = self.forced_tool_choice()? {
1831 if self.tools.is_empty() {
1832 return Err(invalid_request(
1833 "tool_choice forces a tool call, but no tools were offered",
1834 "tool_choice",
1835 ));
1836 }
1837 if let tool_grammar::Forced::Named(name) = forced {
1838 if !self.tools.iter().any(|t| t.function.name == name) {
1839 return Err(invalid_request(
1840 &format!(
1841 "tool_choice names {name:?}, which is not one of the tools offered"
1842 ),
1843 "tool_choice",
1844 ));
1845 }
1846 }
1847 if stated_grammar.is_some() {
1857 return Err(invalid_request(
1858 "a forced tool_choice and a \"grammar\" or response_format \"json_schema\" \
1859 are two different constraints on the same generation; send one",
1860 "tool_choice",
1861 ));
1862 }
1863 if self.json_object_mode() {
1864 return Err(invalid_request(
1865 "a forced tool_choice cannot be combined with response_format json_object: \
1866 the tool-call markers are not JSON",
1867 "tool_choice",
1868 ));
1869 }
1870 }
1871 Ok(())
1872 }
1873
1874 fn effective_stop_sequences(&self) -> Vec<String> {
1881 let mut stop = self.stop_sequences();
1882 if self.tools_active() {
1883 stop.push("</tool_call>".to_string());
1884 }
1885 stop
1886 }
1887
1888 fn json_object_mode(&self) -> bool {
1889 self.response_format
1890 .as_ref()
1891 .and_then(|v| v.get("type"))
1892 .and_then(|v| v.as_str())
1893 == Some("json_object")
1894 }
1895
1896 fn generation_params(
1901 &self,
1902 model: crate::sampling_knobs::SamplerModel<'_>,
1903 ) -> Result<GenerationParams, ApiError> {
1904 Ok(GenerationParams {
1905 reasoning: None,
1910 max_tokens: self.max_tokens,
1911 sampling: self.sampling_params(model)?,
1912 seed: self.resolved_seed(),
1913 stop: self.effective_stop_sequences(),
1914 stop_token_ids: Vec::new(),
1918 json_object: self.json_object_mode(),
1919 grammar: grammar_request::for_request(
1920 self.grammar.as_deref(),
1921 self.response_format.as_ref(),
1922 )?,
1923 cancel: None,
1926 ignore_eos: self.ignore_eos.unwrap_or(false),
1927 })
1928 }
1929
1930 fn generation_params_for_template(
1941 &self,
1942 template: &chat_template::PromptTemplate,
1943 served_model: &str,
1944 model: crate::sampling_knobs::SamplerModel<'_>,
1945 ) -> Result<GenerationParams, ApiError> {
1946 let mut params = self.generation_params(model)?;
1947 params.reasoning = crate::policy::parser::ReasoningFormat::infer(served_model);
1952 if let Some(stop) = template.end_of_turn() {
1953 if !params.stop.iter().any(|s| s == stop) {
1954 params.stop.push(stop.to_string());
1955 }
1956 }
1957 if let Some(forced) = self.forced_tool_choice()? {
1958 params.grammar = Some(tool_grammar::build(
1962 forced,
1963 &self.tool_specs(),
1964 policy::parser::ToolCallFormat::infer(served_model),
1965 )?);
1966 }
1967 Ok(params)
1968 }
1969
1970 fn is_cacheable(&self) -> bool {
1979 self.temperature.unwrap_or(0.0) <= 0.0 || self.seed.is_some()
1980 }
1981
1982 fn cache_key(&self, prompt: &str, params: &GenerationParams) -> CacheKey {
1998 CacheKey {
1999 model: self.model.clone(),
2000 prompt: prompt.to_string(),
2001 generation: response_cache::generation_key(params),
2002 seed: self.seed,
2003 }
2004 }
2005
2006 fn resolved_seed(&self) -> u64 {
2007 self.seed.unwrap_or_else(|| {
2008 std::time::SystemTime::now()
2009 .duration_since(std::time::UNIX_EPOCH)
2010 .map(|d| d.as_nanos() as u64)
2011 .unwrap_or(0xDEFA017)
2012 })
2013 }
2014}
2015
2016#[derive(Serialize)]
2017struct ChatCompletionChoice {
2018 index: usize,
2019 message: ChatCompletionResponseMessage,
2020 finish_reason: &'static str,
2021}
2022
2023#[derive(Serialize)]
2024struct ChatCompletionResponseMessage {
2025 role: &'static str,
2026 #[serde(skip_serializing_if = "Option::is_none")]
2027 content: Option<String>,
2028 #[serde(skip_serializing_if = "Option::is_none")]
2032 reasoning_content: Option<String>,
2033 #[serde(skip_serializing_if = "Option::is_none")]
2034 tool_calls: Option<Vec<ToolCallOut>>,
2035}
2036
2037#[derive(Serialize, Clone)]
2038struct ToolCallOut {
2039 id: String,
2040 #[serde(rename = "type")]
2041 kind: &'static str,
2042 function: ToolCallFunctionOut,
2043}
2044
2045#[derive(Serialize, Clone)]
2053struct ToolCallDelta {
2054 index: usize,
2055 #[serde(skip_serializing_if = "Option::is_none")]
2056 id: Option<String>,
2057 #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
2058 kind: Option<&'static str>,
2059 function: ToolCallFunctionDelta,
2060}
2061
2062#[derive(Serialize, Clone, Default)]
2063struct ToolCallFunctionDelta {
2064 #[serde(skip_serializing_if = "Option::is_none")]
2065 name: Option<String>,
2066 #[serde(skip_serializing_if = "Option::is_none")]
2069 arguments: Option<String>,
2070}
2071
2072impl ToolCallDelta {
2073 fn whole(index: usize, name: String, arguments: String) -> Self {
2075 ToolCallDelta {
2076 index,
2077 id: Some(format!("call_{index}")),
2078 kind: Some("function"),
2079 function: ToolCallFunctionDelta {
2080 name: Some(name),
2081 arguments: Some(arguments),
2082 },
2083 }
2084 }
2085
2086 fn opening(index: usize, name: String) -> Self {
2088 ToolCallDelta {
2089 index,
2090 id: Some(format!("call_{index}")),
2091 kind: Some("function"),
2092 function: ToolCallFunctionDelta {
2093 name: Some(name),
2094 arguments: Some(String::new()),
2095 },
2096 }
2097 }
2098
2099 fn arguments(index: usize, fragment: String) -> Self {
2101 ToolCallDelta {
2102 index,
2103 id: None,
2104 kind: None,
2105 function: ToolCallFunctionDelta {
2106 name: None,
2107 arguments: Some(fragment),
2108 },
2109 }
2110 }
2111}
2112
2113#[derive(Serialize, Clone)]
2114struct ToolCallFunctionOut {
2115 name: String,
2116 arguments: String,
2120}
2121
2122#[derive(Serialize)]
2123struct ChatCompletionResponse {
2124 id: String,
2125 request_id: String,
2131 object: &'static str,
2132 model: String,
2133 choices: Vec<ChatCompletionChoice>,
2134 usage: generate::Usage,
2139 ferrox_cache: &'static str,
2150}
2151
2152#[derive(Serialize)]
2153struct ChatCompletionChunkDelta {
2154 #[serde(skip_serializing_if = "Option::is_none")]
2155 role: Option<&'static str>,
2156 #[serde(skip_serializing_if = "Option::is_none")]
2157 content: Option<String>,
2158 #[serde(skip_serializing_if = "Option::is_none")]
2160 reasoning_content: Option<String>,
2161 #[serde(skip_serializing_if = "Option::is_none")]
2162 tool_calls: Option<Vec<ToolCallDelta>>,
2163}
2164
2165#[derive(Serialize)]
2166struct ChatCompletionChunkChoice {
2167 index: usize,
2168 delta: ChatCompletionChunkDelta,
2169 finish_reason: Option<&'static str>,
2170}
2171
2172#[derive(Serialize)]
2173struct ChatCompletionChunk {
2174 id: String,
2175 #[serde(skip_serializing_if = "Option::is_none")]
2182 request_id: Option<String>,
2183 object: &'static str,
2184 model: String,
2185 choices: Vec<ChatCompletionChunkChoice>,
2186 #[serde(skip_serializing_if = "Option::is_none")]
2189 usage: Option<generate::Usage>,
2190}
2191
2192async fn health(State(state): State<Arc<AppState>>) -> Response {
2198 let snapshot = state.detection.snapshot();
2199 let mut capabilities = snapshot.capabilities;
2200 let active = state.active();
2201
2202 capabilities.push(match active.as_deref() {
2205 None => ferrox_api::Capability::unavailable(
2210 ferrox_api::health::capability::REAL_WEIGHTS,
2211 ferrox_api::health::reason::MODEL_NOT_LOADED,
2212 "No model is loaded. POST /admin/models/load with an id from GET /admin/models.",
2213 ),
2214 Some(active) if active.is_synthetic() => ferrox_api::Capability::unavailable(
2215 ferrox_api::health::capability::REAL_WEIGHTS,
2216 ferrox_api::health::reason::MODEL_NOT_LOADED,
2217 "Serving synthetic random weights: set FERROX_MODEL_PATH (or -m) to a real \
2218 checkpoint. Output from this model is noise.",
2219 ),
2220 Some(active) if active.encoder().is_some() => {
2230 let endpoints = active
2231 .encoder()
2232 .map(|e| encoder_endpoints(e))
2233 .unwrap_or_default();
2234 let served_by = match endpoints.is_empty() {
2235 true => "no endpoint in this build serves it".to_string(),
2236 false => format!("served by {}", endpoints.join(" and ")),
2237 };
2238 ferrox_api::Capability::available(
2239 ferrox_api::health::capability::REAL_WEIGHTS,
2240 format!(
2241 "Serving the real embedding checkpoint '{}'. This is an ENCODER, \
2242 {served_by}; generation endpoints refuse it.",
2243 active.name(),
2244 ),
2245 )
2246 }
2247 Some(active) => ferrox_api::Capability::available(
2248 ferrox_api::health::capability::REAL_WEIGHTS,
2249 format!("Serving the real checkpoint '{}'.", active.name()),
2250 ),
2251 });
2252 capabilities.push(if active.as_ref().is_some_and(|a| a.batcher.is_some()) {
2253 ferrox_api::Capability::available(
2254 ferrox_api::health::capability::CONTINUOUS_BATCHING,
2255 if state.continuous_batching_enabled && continuous_batching_env().is_none() {
2256 "On by default on Metal. Concurrent requests share one batched decode worker."
2257 } else {
2258 "Concurrent requests share one batched decode step."
2259 },
2260 )
2261 } else if state.metal_private_decode_gate.is_some() {
2262 ferrox_api::Capability::unavailable(
2263 ferrox_api::health::capability::CONTINUOUS_BATCHING,
2264 ferrox_api::health::reason::DISABLED,
2265 "Off; private Metal decodes serialize (one at a time). Set FERROX_CONTINUOUS_BATCHING=1 or --cont-batching for parallel serving.",
2266 )
2267 } else {
2268 ferrox_api::Capability::unavailable(
2269 ferrox_api::health::capability::CONTINUOUS_BATCHING,
2270 ferrox_api::health::reason::DISABLED,
2271 "Off; set FERROX_CONTINUOUS_BATCHING=1 (incompatible with a KV pool or prefix cache).",
2272 )
2273 });
2274
2275 let last_request_ms = state
2276 .last_request_ms
2277 .load(std::sync::atomic::Ordering::Relaxed);
2278 let uptime = state.started_at.elapsed();
2279 let health_state = if active.is_none() {
2287 ferrox_api::HealthState::Unavailable
2288 } else {
2289 snapshot.state
2290 };
2291 let body = ferrox_api::HealthResponse {
2292 state: health_state,
2293 reason: match health_state {
2294 ferrox_api::HealthState::Ready => None,
2295 ferrox_api::HealthState::Unavailable => {
2296 Some(ferrox_api::health::reason::MODEL_NOT_LOADED.to_string())
2297 }
2298 ferrox_api::HealthState::Detecting => {
2299 Some(ferrox_api::health::reason::DETECTING.to_string())
2300 }
2301 },
2302 detail: match health_state {
2303 ferrox_api::HealthState::Ready => None,
2304 ferrox_api::HealthState::Unavailable => Some(
2305 "No model is loaded. POST /admin/models/load with an id from GET /admin/models."
2306 .to_string(),
2307 ),
2308 ferrox_api::HealthState::Detecting => {
2309 Some("Probing available compute backends.".to_string())
2310 }
2311 },
2312 model: active
2313 .as_deref()
2314 .map(|active| ferrox_api::health::ModelSummary {
2315 id: active.name().to_string(),
2316 tokenizer: active.tokenizer_kind().to_string(),
2317 synthetic_weights: active.is_synthetic(),
2318 }),
2319 capabilities,
2320 version: env!("CARGO_PKG_VERSION").to_string(),
2321 pid: std::process::id(),
2322 uptime_seconds: uptime.as_secs_f64(),
2323 server_time_unix_ms: std::time::SystemTime::now()
2324 .duration_since(std::time::UNIX_EPOCH)
2325 .map(|d| d.as_millis().min(u64::MAX as u128) as u64)
2326 .unwrap_or(0),
2327 last_request_age_seconds: (last_request_ms > 0)
2328 .then(|| uptime.as_secs_f64() - (last_request_ms as f64 / 1000.0))
2329 .map(|age| age.max(0.0)),
2330 };
2331
2332 let status =
2333 StatusCode::from_u16(body.state.http_status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
2334 (status, Json(body)).into_response()
2335}
2336
2337async fn list_models(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
2338 let Some(active) = state.active() else {
2342 return Json(serde_json::json!({ "object": "list", "data": [] }));
2343 };
2344 let mut model_entry = serde_json::json!({
2345 "id": active.name(),
2346 "object": "model",
2347 "ferrox_synthetic_weights": active.is_synthetic(),
2348 "ferrox_tokenizer": active.tokenizer_kind(),
2349 });
2350 if let Some(encoder) = active.encoder() {
2356 model_entry["ferrox_model_kind"] = serde_json::json!("embedding");
2357 model_entry["ferrox_endpoints"] = serde_json::json!(encoder_endpoints(encoder));
2358 model_entry["ferrox_n_embd"] = serde_json::json!(encoder.n_embd());
2359 model_entry["ferrox_pooling"] = serde_json::json!(encoder.pooling_type().name());
2360 model_entry["ferrox_context_length"] = serde_json::json!(encoder.n_ctx_train());
2361 }
2362 if let Some(model) = active.generative_opt() {
2370 let parser_configured =
2371 crate::policy::parser::ReasoningFormat::infer(active.name()).is_some();
2372 let gears = model.chat_template().think_gears(parser_configured);
2373 if !gears.is_empty() {
2374 model_entry["supported_reasoning_efforts"] = serde_json::json!(gears.supported);
2375 if let Some(default) = &gears.default {
2376 model_entry["default_reasoning_effort"] = serde_json::json!(default);
2377 }
2378 model_entry["reasoning_effort_kwargs"] = serde_json::json!(gears.kwargs);
2381 }
2382 }
2383 if let Some(mcp) = &state.mcp {
2384 model_entry["ferrox_mcp"] = mcp.models_metadata();
2385 }
2386 Json(serde_json::json!({
2387 "object": "list",
2388 "data": [model_entry]
2389 }))
2390}
2391
2392async fn serving_stats(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
2406 let now_ms = state.uptime().as_millis().min(u64::MAX as u128) as u64;
2407 let mut serving = state.serving.lock().unwrap_or_else(|p| p.into_inner());
2408 let active = state.active();
2409 Json(serde_json::json!({
2410 "model": active.as_ref().map(|a| a.name()),
2411 "state": state
2412 .maintenance
2413 .lock()
2414 .unwrap_or_else(|p| p.into_inner())
2415 .state()
2416 .as_str(),
2417 "uptime_s": state.uptime().as_secs(),
2418 "throughput": {
2419 "decode_tps": (serving.decode_tokens_per_second(now_ms) * 10.0).round() / 10.0,
2420 "prefill_tps": (serving.prefill_tokens_per_second(now_ms) * 10.0).round() / 10.0,
2421 },
2422 "requests": {
2423 "active": state.cancels.live_count(),
2424 "completed": state.stats.recorded_total(),
2425 "p95_ms": state.stats.p95_duration_ms(),
2426 "ttft_mean_ms": state.stats.ttft_mean_ms(),
2427 "prompt_tokens_total": state.stats.tokens_prompt_total(),
2428 "completion_tokens_total": state.stats.tokens_generated_total(),
2429 },
2430 "pools": cache_admin::pool_gauges(&state),
2436 "memory": cache_admin::footprint_json(&state),
2439 }))
2440}
2441
2442#[derive(Deserialize)]
2443struct RequestsQuery {
2444 #[serde(default)]
2445 since: u64,
2446 #[serde(default = "default_requests_limit")]
2447 limit: usize,
2448}
2449
2450fn default_requests_limit() -> usize {
2451 stats::MAX_PAGE
2452}
2453
2454async fn recent_requests(
2462 State(state): State<Arc<AppState>>,
2463 axum::extract::Query(q): axum::extract::Query<RequestsQuery>,
2464) -> Json<serde_json::Value> {
2465 let (rows, cursor, missed) = state.stats.page(q.since, q.limit);
2466 Json(serde_json::json!({
2467 "requests": rows,
2468 "next_cursor": cursor,
2469 "missed": missed,
2470 "total": state.stats.recorded_total(),
2471 }))
2472}
2473
2474#[derive(Serialize)]
2475struct CombinedCacheStats {
2476 response_cache: response_cache::CacheStats,
2477 prefix_cache: Option<ferrox_models::PrefixCacheStats>,
2479}
2480
2481async fn cache_stats(State(state): State<Arc<AppState>>) -> Json<CombinedCacheStats> {
2482 Json(CombinedCacheStats {
2483 response_cache: lock_cache(&state.response_cache).stats(),
2484 prefix_cache: state
2485 .prefix_cache
2486 .as_ref()
2487 .map(|pc| pc.lock().unwrap_or_else(|p| p.into_inner()).stats()),
2488 })
2489}
2490
2491async fn metrics(State(state): State<Arc<AppState>>) -> Response {
2496 use std::sync::atomic::Ordering;
2497
2498 let cache_stats = lock_cache(&state.response_cache).stats();
2499 let active = state.active();
2500 let requests_total = state.requests_total.load(Ordering::Relaxed);
2501 let errors_total = state.request_errors_total.load(Ordering::Relaxed);
2502 let uptime = state.started_at.elapsed().as_secs_f64();
2503
2504 let body = format!(
2505 "# HELP ferrox_requests_total Total chat completion requests received.\n\
2506 # TYPE ferrox_requests_total counter\n\
2507 ferrox_requests_total {requests_total}\n\
2508 # HELP ferrox_request_errors_total Total chat completion requests that returned an error.\n\
2509 # TYPE ferrox_request_errors_total counter\n\
2510 ferrox_request_errors_total {errors_total}\n\
2511 # HELP ferrox_cache_hits_total Whole-response cache hits.\n\
2512 # TYPE ferrox_cache_hits_total counter\n\
2513 ferrox_cache_hits_total {}\n\
2514 # HELP ferrox_cache_misses_total Whole-response cache misses.\n\
2515 # TYPE ferrox_cache_misses_total counter\n\
2516 ferrox_cache_misses_total {}\n\
2517 # HELP ferrox_cache_entries Current whole-response cache entry count.\n\
2518 # TYPE ferrox_cache_entries gauge\n\
2519 ferrox_cache_entries {}\n\
2520 # HELP ferrox_synthetic_weights 1 if serving synthetic random weights instead of a real checkpoint.\n\
2521 # TYPE ferrox_synthetic_weights gauge\n\
2522 ferrox_synthetic_weights {}\n\
2523 # HELP ferrox_uptime_seconds Seconds since this server process started.\n\
2524 # TYPE ferrox_uptime_seconds gauge\n\
2525 ferrox_uptime_seconds {uptime}\n",
2526 cache_stats.hits,
2527 cache_stats.misses,
2528 cache_stats.entries,
2529 active
2533 .as_ref()
2534 .map(|a| a.is_synthetic() as u8)
2535 .unwrap_or(0),
2536 );
2537
2538 let body = match active
2542 .as_ref()
2543 .and_then(|a| a.expert_store_stats())
2544 {
2545 Some(es) => format!(
2546 "{body}\
2547 # HELP ferrox_expert_cache_hits_total Expert-store cache hits.\n\
2548 # TYPE ferrox_expert_cache_hits_total counter\n\
2549 ferrox_expert_cache_hits_total {}\n\
2550 # HELP ferrox_expert_cache_misses_total Expert-store cache misses (source reads).\n\
2551 # TYPE ferrox_expert_cache_misses_total counter\n\
2552 ferrox_expert_cache_misses_total {}\n\
2553 # HELP ferrox_expert_cache_evictions_total Expert-store LRU evictions.\n\
2554 # TYPE ferrox_expert_cache_evictions_total counter\n\
2555 ferrox_expert_cache_evictions_total {}\n\
2556 # HELP ferrox_expert_cache_pass_throughs_total Acquires served uncached (entry could not fit the budget).\n\
2557 # TYPE ferrox_expert_cache_pass_throughs_total counter\n\
2558 ferrox_expert_cache_pass_throughs_total {}\n\
2559 # HELP ferrox_expert_cache_bytes_read_total Bytes read from the checkpoint for expert misses.\n\
2560 # TYPE ferrox_expert_cache_bytes_read_total counter\n\
2561 ferrox_expert_cache_bytes_read_total {}\n\
2562 # HELP ferrox_expert_cache_resident_bytes Current expert-cache footprint in bytes.\n\
2563 # TYPE ferrox_expert_cache_resident_bytes gauge\n\
2564 ferrox_expert_cache_resident_bytes {}\n",
2565 es.hits, es.misses, es.evictions, es.pass_throughs, es.bytes_read, es.resident_bytes,
2566 ),
2567 None => body,
2568 };
2569
2570 let body = match active.as_ref().and_then(|a| a.batcher.as_ref()) {
2575 Some(batcher) => {
2576 let sched = batcher.stats();
2577 format!(
2578 "{body}\
2579 # HELP ferrox_prefill_chunks_total Bounded prefill chunks the batch scheduler has run.\n\
2580 # TYPE ferrox_prefill_chunks_total counter\n\
2581 ferrox_prefill_chunks_total {}\n\
2582 # HELP ferrox_prefill_tokens_total Prompt tokens run through chunked prefill.\n\
2583 # TYPE ferrox_prefill_tokens_total counter\n\
2584 ferrox_prefill_tokens_total {}\n\
2585 # HELP ferrox_decode_steps_total Batched decode steps the batch scheduler has run.\n\
2586 # TYPE ferrox_decode_steps_total counter\n\
2587 ferrox_decode_steps_total {}\n\
2588 # HELP ferrox_scheduler_queue_depth Requests waiting for admission to the batch scheduler.\n\
2589 # TYPE ferrox_scheduler_queue_depth gauge\n\
2590 ferrox_scheduler_queue_depth {}\n\
2591 # HELP ferrox_scheduler_queue_rejected_total Requests refused with 503 because the admission queue was full.\n\
2592 # TYPE ferrox_scheduler_queue_rejected_total counter\n\
2593 ferrox_scheduler_queue_rejected_total {}\n\
2594 # HELP ferrox_kv_blocks_total KV blocks in the scheduler's admission budget (0 when unconfigured).\n\
2595 # TYPE ferrox_kv_blocks_total gauge\n\
2596 ferrox_kv_blocks_total {}\n\
2597 # HELP ferrox_kv_blocks_free KV blocks not reserved by an in-flight request.\n\
2598 # TYPE ferrox_kv_blocks_free gauge\n\
2599 ferrox_kv_blocks_free {}\n\
2600 # HELP ferrox_kv_block_size Token positions per KV block.\n\
2601 # TYPE ferrox_kv_block_size gauge\n\
2602 ferrox_kv_block_size {}\n\
2603 # HELP ferrox_kv_rejected_too_large_total Requests refused with 400 because they exceed the whole KV block budget.\n\
2604 # TYPE ferrox_kv_rejected_too_large_total counter\n\
2605 ferrox_kv_rejected_too_large_total {}\n\
2606 # HELP ferrox_kv_rejected_context_length_total Requests refused with 400 for exceeding the per-request context ceiling.\n\
2607 # TYPE ferrox_kv_rejected_context_length_total counter\n\
2608 ferrox_kv_rejected_context_length_total {}\n\
2609 # HELP ferrox_scheduler_aborted_total Requests the batch scheduler stopped because they were cancelled.\n\
2610 # TYPE ferrox_scheduler_aborted_total counter\n\
2611 ferrox_scheduler_aborted_total {}\n",
2612 sched.prefill_chunks,
2613 sched.prefill_tokens,
2614 sched.decode_steps,
2615 sched.queue_depth,
2616 sched.queue_rejected,
2617 sched.kv_blocks_total,
2618 sched.kv_blocks_free,
2619 sched.kv_block_size,
2620 sched.kv_rejected_too_large,
2621 sched.kv_rejected_context_length,
2622 sched.aborted,
2623 )
2624 }
2625 None => body,
2626 };
2627
2628 (
2629 [(
2630 axum::http::header::CONTENT_TYPE,
2631 "text/plain; version=0.0.4",
2632 )],
2633 body,
2634 )
2635 .into_response()
2636}
2637
2638pub(crate) type ApiError = (StatusCode, Json<serde_json::Value>);
2639
2640pub(crate) fn invalid_request(message: &str, param: &str) -> ApiError {
2646 (
2647 StatusCode::BAD_REQUEST,
2648 Json(serde_json::json!({"error": {
2649 "message": message,
2650 "type": "invalid_request_error",
2651 "param": param,
2652 "code": null,
2653 }})),
2654 )
2655}
2656
2657pub(crate) fn unsupported_feature(message: &str) -> ApiError {
2658 (
2659 StatusCode::NOT_IMPLEMENTED,
2660 Json(serde_json::json!({"error": {"message": message, "type": "unsupported"}})),
2661 )
2662}
2663
2664pub(crate) fn decode_error_response(e: generate::DecodeError) -> ApiError {
2665 let status = match e {
2666 generate::DecodeError::TokenOutOfVocab { .. } => StatusCode::BAD_REQUEST,
2667 generate::DecodeError::KvBudgetExceeded { .. } => StatusCode::BAD_REQUEST,
2672 generate::DecodeError::KvPoolExhausted | generate::DecodeError::QueueFull { .. } => {
2678 StatusCode::SERVICE_UNAVAILABLE
2679 }
2680 generate::DecodeError::GrammarConstraint { .. } => StatusCode::BAD_REQUEST,
2684 };
2685 tracing::warn!("decode error: {e}");
2686 let mut body = serde_json::json!({"error": {"message": e.to_string()}});
2687 if let generate::DecodeError::KvBudgetExceeded {
2693 binding,
2694 estimated_bytes,
2695 limit_bytes,
2696 positions,
2697 positions_limit,
2698 ..
2699 } = &e
2700 {
2701 body["error"]["type"] = serde_json::json!("invalid_request_error");
2702 body["error"]["code"] = serde_json::json!(binding);
2703 body["error"]["binding"] = serde_json::json!(binding);
2704 body["error"]["estimated_bytes"] = serde_json::json!(estimated_bytes);
2705 body["error"]["limit_bytes"] = serde_json::json!(limit_bytes);
2706 body["error"]["positions"] = serde_json::json!(positions);
2707 body["error"]["positions_limit"] = serde_json::json!(positions_limit);
2708 }
2709 if let Some(secs) = e.retry_after_secs() {
2713 body["error"]["retry_after_seconds"] = serde_json::json!(secs);
2714 }
2715 (status, Json(body))
2716}
2717
2718pub(crate) fn join_error_response(e: tokio::task::JoinError) -> ApiError {
2719 tracing::error!("generation task panicked: {e}");
2720 (
2721 StatusCode::INTERNAL_SERVER_ERROR,
2722 Json(serde_json::json!({"error": {"message": "internal error during generation"}})),
2723 )
2724}
2725
2726#[allow(clippy::too_many_arguments)] fn run_generation_emit(
2736 model: &Model,
2737 prompt: &str,
2738 params: &GenerationParams,
2739 kv_pool: Option<&generate::KvPoolConfig>,
2740 paged_kv: Option<&generate::PagedKvConfig>,
2741 prefix_cache: Option<&Mutex<PrefixCache>>,
2742 continuous_batcher: Option<&serving::batch::ContinuousBatcher>,
2743 ceiling: Option<&budget::ContextCeiling>,
2744 metal_private_decode_gate: Option<&std::sync::Mutex<()>>,
2745 mut emit: impl FnMut(&str),
2746) -> Result<(FinishReason, generate::Usage, String), generate::DecodeError> {
2747 let synthetic = model.is_synthetic();
2748 let mut chunks = Vec::new();
2749 let params = &{
2755 let mut resolved = params.clone();
2756 resolved.stop_token_ids =
2757 crate::stop::resolve_stop_tokens(&resolved.stop, |text| model.encode(text));
2758 resolved
2759 };
2760 let used_batcher = matches!((model, continuous_batcher), (Model::Gguf(_), Some(_)));
2761 let _metal_private_guard =
2762 acquire_metal_private_decode_gate(metal_private_decode_gate, used_batcher);
2763 let (finish, usage) = match model {
2764 Model::Gguf(m) => {
2765 if let Some(batcher) = continuous_batcher {
2766 let mut tokens = m.tokenizer.encode(prompt);
2767 ferrox_models::tokenizer::prepend_bos(&mut tokens, m.bos_id);
2768 let (finish, _generated_ids, text, usage) = if synthetic {
2769 batcher.generate(tokens, params.clone(), m.stop_tokens.clone())?
2770 } else {
2771 batcher.generate_streaming(
2772 tokens,
2773 params.clone(),
2774 m.stop_tokens.clone(),
2775 Some(|chunk: &str| {
2776 if !chunk.is_empty() {
2777 chunks.push(chunk.to_string());
2778 emit(chunk);
2779 }
2780 }),
2781 )?
2782 };
2783 if !text.is_empty() && chunks.is_empty() {
2784 chunks.push(text);
2785 }
2786 (finish, usage)
2787 } else {
2788 generate::generate(
2789 &m.decoder,
2790 m.tokenizer.as_ref(),
2791 &m.stop_tokens,
2792 m.bos_id,
2793 prompt,
2794 params,
2795 kv_pool,
2796 paged_kv,
2797 prefix_cache,
2798 ceiling,
2799 |chunk| {
2800 chunks.push(chunk.to_string());
2801 if !synthetic {
2802 emit(chunk);
2803 }
2804 },
2805 )?
2806 }
2807 }
2808 Model::Kimi(m) => generate::generate_engine(
2809 &m.engine,
2810 &m.tokenizer,
2811 &m.stop_tokens,
2812 None,
2813 prompt,
2814 params,
2815 |chunk| {
2816 chunks.push(chunk.to_string());
2817 if !synthetic {
2818 emit(chunk);
2819 }
2820 },
2821 )?,
2822 Model::Mla(m) => generate::generate_engine(
2823 &m.engine,
2824 &m.tokenizer,
2825 &m.stop_tokens,
2826 m.bos_id,
2827 prompt,
2828 params,
2829 |chunk| {
2830 chunks.push(chunk.to_string());
2831 if !synthetic {
2832 emit(chunk);
2833 }
2834 },
2835 )?,
2836 Model::Gemma4(m) => generate::generate_engine(
2837 &m.engine,
2838 &m.tokenizer,
2839 &m.stop_tokens,
2840 m.bos_id,
2841 prompt,
2842 params,
2843 |chunk| {
2844 chunks.push(chunk.to_string());
2845 if !synthetic {
2846 emit(chunk);
2847 }
2848 },
2849 )?,
2850 Model::Glm52(m) => generate::generate_engine(
2851 &m.engine,
2852 &m.tokenizer,
2853 &m.stop_tokens,
2854 m.bos_id,
2855 prompt,
2856 params,
2857 |chunk| {
2858 chunks.push(chunk.to_string());
2859 if !synthetic {
2860 emit(chunk);
2861 }
2862 },
2863 )?,
2864 };
2865
2866 let mut full = chunks.concat();
2867 if synthetic {
2868 full = format!(
2869 "[ferrox synthetic-weight demo: no real checkpoint loaded -- set FERROX_MODEL_PATH \
2870 to serve a real model. Decoded ids -> {full:?}]"
2871 );
2872 emit(&full);
2873 } else if used_batcher && !full.is_empty() && chunks.is_empty() {
2874 emit(&full);
2875 }
2876
2877 Ok((finish, usage, full))
2878}
2879
2880#[allow(clippy::too_many_arguments)] pub(crate) fn run_generation(
2885 model: &Model,
2886 prompt: &str,
2887 params: &GenerationParams,
2888 kv_pool: Option<&generate::KvPoolConfig>,
2889 paged_kv: Option<&generate::PagedKvConfig>,
2890 prefix_cache: Option<&Mutex<PrefixCache>>,
2891 continuous_batcher: Option<&serving::batch::ContinuousBatcher>,
2892 ceiling: Option<&budget::ContextCeiling>,
2893 metal_private_decode_gate: Option<&std::sync::Mutex<()>>,
2894) -> Result<(Vec<String>, FinishReason, generate::Usage), generate::DecodeError> {
2895 let (finish, usage, full) = run_generation_emit(
2896 model,
2897 prompt,
2898 params,
2899 kv_pool,
2900 paged_kv,
2901 prefix_cache,
2902 continuous_batcher,
2903 ceiling,
2904 metal_private_decode_gate,
2905 |_| {},
2906 )?;
2907 Ok((
2908 if full.is_empty() {
2909 Vec::new()
2910 } else {
2911 vec![full]
2912 },
2913 finish,
2914 usage,
2915 ))
2916}
2917
2918pub(crate) fn prompt_from_messages(
2929 messages: &[ChatMessage],
2930 template: &chat_template::PromptTemplate,
2931 tools: &[ToolDef],
2932 extra: serde_json::Map<String, serde_json::Value>,
2933) -> Result<String, ApiError> {
2934 let rendered = if tools.is_empty() || template.handles_tools() {
2935 template.render(messages, tools, extra)
2936 } else {
2937 let mut with_preamble = Vec::with_capacity(messages.len() + 1);
2938 with_preamble.push(ChatMessage {
2939 role: "system".to_string(),
2940 content: Some(MessageContent::Text(tool_preamble(tools))),
2941 tool_calls: None,
2942 tool_call_id: None,
2943 reasoning_content: None,
2944 });
2945 with_preamble.extend_from_slice(messages);
2946 template.render(&with_preamble, &[], extra)
2947 };
2948 rendered.map_err(template_error_response)
2949}
2950
2951fn template_error_response(err: ferrox_models::chat_template::TemplateError) -> ApiError {
2956 (
2957 StatusCode::BAD_REQUEST,
2958 Json(serde_json::json!({
2959 "error": {
2960 "message": format!("chat template failed to render: {err}"),
2961 "type": "invalid_request_error",
2962 "param": "messages",
2963 "code": null,
2964 }
2965 })),
2966 )
2967}
2968
2969fn tool_preamble(tools: &[ToolDef]) -> String {
2982 let mut out = String::from(
2983 "You can call tools to help answer the user. To call a tool, respond with \
2984 EXACTLY one line in this format and nothing else:\n\
2985 <tool_call>{\"name\": \"<tool name>\", \"arguments\": {<arguments as a JSON \
2986 object matching that tool's parameters>}}</tool_call>\n\n\
2987 Available tools:\n",
2988 );
2989 for t in tools {
2990 out.push_str(&format!(
2991 "- {}: {}\n parameters (JSON schema): {}\n",
2992 t.function.name,
2993 t.function.description.as_deref().unwrap_or(""),
2994 t.function
2995 .parameters
2996 .as_ref()
2997 .map(|v| v.to_string())
2998 .unwrap_or_else(|| "{}".to_string()),
2999 ));
3000 }
3001 out
3002}
3003
3004fn tool_call_deltas(
3014 events: Vec<crate::policy::parser::ToolCallEvent>,
3015 opened: &std::cell::Cell<usize>,
3016) -> (String, Vec<ToolCallDelta>) {
3017 let mut text = String::new();
3018 let mut deltas = Vec::new();
3019 for event in events {
3020 match event {
3021 crate::policy::parser::ToolCallEvent::Text(chunk) => text.push_str(&chunk),
3022 crate::policy::parser::ToolCallEvent::CallStart { index, name } => {
3023 opened.set(opened.get().max(index + 1));
3024 deltas.push(ToolCallDelta::opening(index, name));
3025 }
3026 crate::policy::parser::ToolCallEvent::CallArguments { index, fragment } => {
3027 if !fragment.is_empty() {
3028 deltas.push(ToolCallDelta::arguments(index, fragment));
3029 }
3030 }
3031 crate::policy::parser::ToolCallEvent::CallEnd { .. } => {}
3032 }
3033 }
3034 (text, deltas)
3035}
3036
3037fn build_response_message(
3048 text: String,
3049 tools: &[ToolDef],
3050 posture: output::OutputPosture,
3051 base_finish: &'static str,
3052) -> (ChatCompletionResponseMessage, &'static str) {
3053 let parsed = output::parse_output(&text, tools, posture);
3054 let calls: Vec<ToolCallOut> = parsed
3055 .calls
3056 .into_iter()
3057 .enumerate()
3058 .map(|(index, call)| ToolCallOut {
3059 id: format!("call_{index}"),
3060 kind: "function",
3061 function: ToolCallFunctionOut {
3062 name: call.name,
3063 arguments: call.arguments,
3064 },
3065 })
3066 .collect();
3067 if !calls.is_empty() {
3068 return (
3069 ChatCompletionResponseMessage {
3070 role: "assistant",
3071 content: None,
3072 reasoning_content: parsed.reasoning,
3073 tool_calls: Some(calls),
3074 },
3075 "tool_calls",
3076 );
3077 }
3078 (
3079 ChatCompletionResponseMessage {
3080 role: "assistant",
3081 content: Some(parsed.content),
3082 reasoning_content: parsed.reasoning,
3083 tool_calls: None,
3084 },
3085 base_finish,
3086 )
3087}
3088
3089fn resolve_history(state: &AppState, req: &ChatCompletionRequest) -> Vec<ChatMessage> {
3094 let mut history = match &req.session_id {
3095 Some(id) => state.sessions.extend_and_get(id, &req.messages),
3096 None => req.messages.clone(),
3097 };
3098 if req.json_object_mode() {
3099 inject_json_object_system_hint(&mut history);
3100 }
3101 history
3102}
3103
3104fn inject_json_object_system_hint(messages: &mut Vec<ChatMessage>) {
3105 const HINT: &str =
3106 "You must respond with valid JSON only (a single JSON object, no markdown fences).";
3107 if let Some(sys) = messages.iter_mut().find(|m| m.role == "system") {
3108 match &mut sys.content {
3109 Some(MessageContent::Text(s)) if !s.contains("JSON") => {
3110 s.push_str("\n\n");
3111 s.push_str(HINT);
3112 }
3113 None => {
3114 sys.content = Some(MessageContent::Text(HINT.to_string()));
3115 }
3116 _ => {}
3117 }
3118 } else {
3119 messages.insert(
3120 0,
3121 ChatMessage {
3122 role: "system".to_string(),
3123 content: Some(MessageContent::Text(HINT.to_string())),
3124 tool_calls: None,
3125 tool_call_id: None,
3126 reasoning_content: None,
3127 },
3128 );
3129 }
3130}
3131
3132async fn chat_completions(
3133 State(state): State<Arc<AppState>>,
3134 headers: axum::http::HeaderMap,
3135 Json(req): Json<ChatCompletionRequest>,
3136) -> Response {
3137 let attribution = attribution::Attribution::from_headers(&headers);
3138 state
3139 .requests_total
3140 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3141 let started = std::time::Instant::now();
3142
3143 let request_id = ferrox_api::next_request_id();
3147 let stream = req.stream.unwrap_or(false);
3148
3149 let refusal = cache_admin::check_admission(&state)
3155 .err()
3156 .or_else(|| req.validate_supported_fields().err());
3157 if let Some(err) = refusal {
3158 state
3159 .request_errors_total
3160 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3161 let response = err.into_response();
3162 state.record_request(stats::Record {
3163 request_id: &request_id,
3164 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
3165 model: state.active_model_name(),
3166 status: response.status().as_u16(),
3167 stream,
3168 duration_ms: started.elapsed().as_millis() as u64,
3169 usage: None,
3170 attribution: &attribution,
3171 });
3172 return response;
3173 }
3174
3175 let response = if stream {
3176 chat_completions_stream(
3177 Arc::clone(&state),
3178 req,
3179 request_id.clone(),
3180 started,
3181 attribution.clone(),
3182 )
3183 .await
3184 .into_response()
3185 } else {
3186 chat_completions_full(
3187 Arc::clone(&state),
3188 req,
3189 request_id.clone(),
3190 started,
3191 attribution.clone(),
3192 )
3193 .await
3194 .into_response()
3195 };
3196
3197 if response.status().is_client_error() || response.status().is_server_error() {
3198 state
3199 .request_errors_total
3200 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3201 state.record_request(stats::Record {
3205 request_id: &request_id,
3206 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
3207 model: state.active_model_name(),
3210 status: response.status().as_u16(),
3211 stream,
3212 duration_ms: started.elapsed().as_millis() as u64,
3213 usage: None,
3214 attribution: &attribution,
3215 });
3216 }
3217 state.mark_request_finished();
3218
3219 response
3220}
3221
3222async fn chat_completions_full(
3223 state: Arc<AppState>,
3224 req: ChatCompletionRequest,
3225 request_id: String,
3226 started: std::time::Instant,
3227 attribution: attribution::Attribution,
3228) -> Result<Json<ChatCompletionResponse>, ApiError> {
3229 let tools_active = req.tools_active();
3230 let active = state.require_active()?;
3234 let history = resolve_history(&state, &req);
3235 let template = active.generative()?.chat_template();
3236 let kwargs = req.resolve_template_kwargs(&template);
3237 let prompt = prompt_from_messages(&history, &template, &req.tools, kwargs)?;
3238 let params =
3246 req.generation_params_for_template(&template, active.name(), active.sampler_model())?;
3247 let key = req.is_cacheable().then(|| req.cache_key(&prompt, ¶ms));
3248
3249 let (completion, cache_status) = if let Some(cached) = key
3250 .as_ref()
3251 .and_then(|key| lock_cache(&state.response_cache).get(key))
3252 {
3253 tracing::debug!("cache hit for key {}", key.as_ref().unwrap().digest());
3254 (cached, "hit")
3255 } else {
3256 let (chunks, finish, usage) = decode_task::buffered(
3257 decode_task::DecodeHandles::take(&state, &active)?,
3258 prompt.clone(),
3259 params,
3260 )
3261 .await?;
3262
3263 let completion = response_cache::CachedCompletion {
3264 content: chunks.concat(),
3265 finish,
3266 usage,
3267 };
3268 let cache_status = match key {
3274 Some(key) => match completion.clone().cacheable() {
3278 Some(cacheable) => {
3279 tracing::debug!("cache miss for key {}", key.digest());
3280 lock_cache(&state.response_cache).put(key, cacheable);
3281 "miss"
3282 }
3283 None => "skip",
3284 },
3285 None => "skip",
3286 };
3287 (completion, cache_status)
3288 };
3289 let content = completion.content;
3290
3291 if req.json_object_mode() {
3292 json_mode::validate_json_object_output(&content)?;
3293 }
3294
3295 if let Some(id) = &req.session_id {
3299 state.sessions.store_reply(
3300 id,
3301 ChatMessage {
3302 role: "assistant".to_string(),
3303 content: Some(MessageContent::Text(content.clone())),
3304 tool_calls: None,
3305 tool_call_id: None,
3306 reasoning_content: None,
3307 },
3308 );
3309 }
3310
3311 let (message, finish_reason) = build_response_message(
3312 content,
3313 if tools_active { &req.tools } else { &[] },
3314 output::OutputPosture::resolve(active.name(), &prompt),
3315 completion.finish.as_str(),
3316 );
3317
3318 state.record_request(stats::Record {
3319 request_id: &request_id,
3320 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
3321 model: Some(active.name().to_string()),
3324 status: 200,
3325 stream: false,
3326 duration_ms: started.elapsed().as_millis() as u64,
3327 usage: Some(&completion.usage),
3328 attribution: &attribution,
3329 });
3330
3331 Ok(Json(ChatCompletionResponse {
3332 id: request_id.clone(),
3333 request_id,
3334 object: "chat.completion",
3335 model: req.model,
3336 choices: vec![ChatCompletionChoice {
3337 index: 0,
3338 message,
3339 finish_reason,
3340 }],
3341 usage: completion.usage,
3342 ferrox_cache: cache_status,
3343 }))
3344}
3345
3346async fn chat_completions_stream(
3347 state: Arc<AppState>,
3348 req: ChatCompletionRequest,
3349 request_id: String,
3350 started: std::time::Instant,
3351 attribution: attribution::Attribution,
3352) -> Result<Response, ApiError> {
3353 let tools_active = req.tools_active();
3355 let active = state.require_active()?;
3359 let history = resolve_history(&state, &req);
3360 let template = active.generative()?.chat_template();
3361 let kwargs = req.resolve_template_kwargs(&template);
3362 let prompt = prompt_from_messages(&history, &template, &req.tools, kwargs)?;
3363 let model_name = req.model.clone();
3364 let session_id = req.session_id.clone();
3365 let sessions = state.sessions.clone();
3366
3367 let model = Arc::clone(active.generative()?);
3368 let kv_pool = state.kv_pool.clone();
3369 let paged_kv = state.paged_kv.clone();
3370 let prefix_cache = state.prefix_cache.clone();
3371 let batcher = active.batcher.clone();
3372 let ceiling = active.ceiling.clone();
3373 let metal_private_decode_gate = state.metal_private_decode_gate.clone();
3374 let mut params =
3375 req.generation_params_for_template(&template, active.name(), active.sampler_model())?;
3376 let stats_state = Arc::clone(&state);
3377 let served_model = active.name().to_string();
3380 let posture = output::OutputPosture::resolve(&served_model, &prompt);
3384 let offered_tools: Vec<ToolDef> = if tools_active {
3387 req.tools.clone()
3388 } else {
3389 Vec::new()
3390 };
3391
3392 let (cancel_token, cancel_guard) = state.cancels.register(&request_id);
3397 params.cancel = Some(cancel_token.clone());
3398
3399 let overlap = true;
3408
3409 let slot = req
3413 .stream_resumable
3414 .unwrap_or(false)
3415 .then(|| state.streams.register(&request_id));
3416 let emitter = resume::Emitter::new(slot);
3417
3418 let (tx, rx) = tokio::sync::mpsc::channel::<Result<Event, Infallible>>(64);
3419 let keepalive = sse::keepalive_event(&ChatCompletionChunk {
3423 id: request_id.clone(),
3424 request_id: None,
3425 object: "chat.completion.chunk",
3426 model: model_name.clone(),
3427 choices: vec![ChatCompletionChunkChoice {
3428 index: 0,
3429 delta: ChatCompletionChunkDelta {
3430 role: None,
3431 content: None,
3432 reasoning_content: None,
3433 tool_calls: None,
3434 },
3435 finish_reason: None,
3436 }],
3437 usage: None,
3438 });
3439
3440 tokio::task::spawn_blocking(move || {
3441 let _cancel_guard = cancel_guard;
3444 let tx_chunks = tx.clone();
3445 let orphan_timeout = sse::orphan_timeout_from_env();
3450 let mut first = true;
3451 let head_request_id = request_id.clone();
3452 let stream_reasoning: Rc<RefCell<Option<crate::policy::parser::ReasoningParser>>> =
3462 Rc::new(RefCell::new(posture.reasoning_parser()));
3463 let emit_reasoning = Rc::clone(&stream_reasoning);
3464 let stream_tools: Rc<RefCell<Option<crate::policy::parser::ToolCallParser>>> = Rc::new(
3468 RefCell::new(tools_active.then(|| posture.tool_call_parser(&offered_tools))),
3469 );
3470 let emit_tools = Rc::clone(&stream_tools);
3471 let streamed_calls = Rc::new(std::cell::Cell::new(0usize));
3475 let emit_streamed_calls = Rc::clone(&streamed_calls);
3476 let result = run_generation_emit(
3477 &model,
3478 &prompt,
3479 ¶ms,
3480 kv_pool.as_ref(),
3481 paged_kv.as_ref(),
3482 prefix_cache.as_deref(),
3483 batcher.as_ref(),
3484 ceiling.as_deref(),
3485 metal_private_decode_gate.as_deref(),
3486 |chunk| {
3487 if !overlap || chunk.is_empty() {
3488 return;
3489 }
3490 let (reasoning, content) = match emit_reasoning.borrow_mut().as_mut() {
3491 Some(parser) => {
3492 let delta = parser.push(chunk);
3493 (delta.reasoning, delta.content)
3494 }
3495 None => (String::new(), chunk.to_string()),
3496 };
3497 let (content, tool_calls) = match emit_tools.borrow_mut().as_mut() {
3501 Some(parser) => {
3502 let (text, calls) =
3503 tool_call_deltas(parser.push(&content), &emit_streamed_calls);
3504 (text, calls)
3505 }
3506 None => (content, Vec::new()),
3507 };
3508 if reasoning.is_empty() && content.is_empty() && tool_calls.is_empty() {
3511 return;
3512 }
3513 let role = if first { Some("assistant") } else { None };
3514 let request_id = first.then(|| head_request_id.clone());
3515 first = false;
3516 let payload = ChatCompletionChunk {
3517 id: head_request_id.clone(),
3518 request_id,
3519 object: "chat.completion.chunk",
3520 model: model_name.clone(),
3521 choices: vec![ChatCompletionChunkChoice {
3522 index: 0,
3523 delta: ChatCompletionChunkDelta {
3524 role,
3525 content: (!content.is_empty()).then_some(content),
3526 reasoning_content: (!reasoning.is_empty()).then_some(reasoning),
3527 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
3528 },
3529 finish_reason: None,
3530 }],
3531 usage: None,
3532 };
3533 if let Err(why) =
3542 sse::send_or_orphan(&tx_chunks, Ok(emitter.event(&payload)), orphan_timeout)
3543 {
3544 if why == sse::SendFailure::Orphaned {
3545 tracing::warn!(
3546 "SSE stream {head_request_id} accepted nothing for the orphan \
3547 deadline; treating it as abandoned"
3548 );
3549 }
3550 if !emitter.is_resumable() {
3562 cancel_token.cancel();
3563 }
3564 }
3565 },
3566 );
3567
3568 let mut pending_request_id = first.then(|| request_id.clone());
3574
3575 match result {
3576 Ok((finish, usage, full_text)) => {
3577 if let Some(id) = &session_id {
3578 sessions.store_reply(
3579 id,
3580 ChatMessage {
3581 role: "assistant".to_string(),
3582 content: Some(MessageContent::Text(full_text.clone())),
3583 tool_calls: None,
3584 tool_call_id: None,
3585 reasoning_content: None,
3586 },
3587 );
3588 }
3589 let mut streamed_finish: Option<&'static str> = None;
3595 if overlap {
3596 let tail = stream_reasoning
3597 .borrow_mut()
3598 .as_mut()
3599 .map(|parser| parser.flush())
3600 .unwrap_or_default();
3601 let (mut content, mut tool_calls) = (tail.content, Vec::new());
3602 if let Some(parser) = stream_tools.borrow_mut().as_mut() {
3603 let mut events = parser.push(&content);
3604 events.extend(parser.finish());
3605 let (text, calls) = tool_call_deltas(events, &streamed_calls);
3606 content = text;
3607 tool_calls = calls;
3608 }
3609 if !content.is_empty() || !tail.reasoning.is_empty() || !tool_calls.is_empty() {
3610 let payload = ChatCompletionChunk {
3611 id: request_id.clone(),
3612 request_id: pending_request_id.take(),
3613 object: "chat.completion.chunk",
3614 model: model_name.clone(),
3615 choices: vec![ChatCompletionChunkChoice {
3616 index: 0,
3617 delta: ChatCompletionChunkDelta {
3618 role: None,
3619 content: (!content.is_empty()).then_some(content),
3620 reasoning_content: (!tail.reasoning.is_empty())
3621 .then_some(tail.reasoning),
3622 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
3623 },
3624 finish_reason: None,
3625 }],
3626 usage: None,
3627 };
3628 let _ =
3629 sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3630 }
3631 if streamed_calls.get() > 0 {
3632 streamed_finish = Some("tool_calls");
3633 }
3634 } else {
3635 let parsed = output::parse_output(&full_text, &offered_tools, posture);
3638 let tool_calls: Vec<ToolCallDelta> = parsed
3639 .calls
3640 .iter()
3641 .enumerate()
3642 .map(|(index, call)| {
3643 ToolCallDelta::whole(index, call.name.clone(), call.arguments.clone())
3644 })
3645 .collect();
3646 if !tool_calls.is_empty() {
3647 streamed_finish = Some("tool_calls");
3648 }
3649 if !tool_calls.is_empty()
3650 || !parsed.content.is_empty()
3651 || parsed.reasoning.is_some()
3652 {
3653 let payload = ChatCompletionChunk {
3654 id: request_id.clone(),
3655 request_id: pending_request_id.take(),
3656 object: "chat.completion.chunk",
3657 model: model_name.clone(),
3658 choices: vec![ChatCompletionChunkChoice {
3659 index: 0,
3660 delta: ChatCompletionChunkDelta {
3661 role: Some("assistant"),
3662 content: (!parsed.content.is_empty() && tool_calls.is_empty())
3663 .then(|| parsed.content.clone()),
3664 reasoning_content: parsed.reasoning.clone(),
3665 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
3666 },
3667 finish_reason: None,
3668 }],
3669 usage: None,
3670 };
3671 let _ =
3672 sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3673 }
3674 }
3675 let final_finish_reason = match streamed_finish {
3679 Some(reason) if finish.as_str() != "length" => reason,
3680 _ => finish.as_str(),
3681 };
3682 let final_payload = ChatCompletionChunk {
3683 id: request_id.clone(),
3684 request_id: pending_request_id.take(),
3685 object: "chat.completion.chunk",
3686 model: model_name,
3687 choices: vec![ChatCompletionChunkChoice {
3688 index: 0,
3689 delta: ChatCompletionChunkDelta {
3690 role: None,
3691 content: None,
3692 reasoning_content: None,
3693 tool_calls: None,
3694 },
3695 finish_reason: Some(final_finish_reason),
3696 }],
3697 usage: Some(usage.clone()),
3698 };
3699 let _ = sse::send_or_orphan(&tx, Ok(emitter.event(&final_payload)), orphan_timeout);
3700 let _ = sse::send_or_orphan(&tx, Ok(emitter.done()), orphan_timeout);
3701 stats_state.record_request(stats::Record {
3706 request_id: &request_id,
3707 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
3708 model: Some(served_model.clone()),
3709 status: 200,
3710 stream: true,
3711 duration_ms: started.elapsed().as_millis() as u64,
3712 usage: Some(&usage),
3713 attribution: &attribution,
3714 });
3715 }
3716 Err(e) => {
3717 tracing::warn!("decode error on streamed request {request_id}: {e}");
3718 stats_state.record_request(stats::Record {
3724 request_id: &request_id,
3725 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
3726 model: Some(served_model.clone()),
3727 status: 500,
3728 stream: true,
3729 duration_ms: started.elapsed().as_millis() as u64,
3730 usage: None,
3731 attribution: &attribution,
3732 });
3733 let payload = ChatCompletionChunk {
3734 id: request_id.clone(),
3735 request_id: pending_request_id.take(),
3736 object: "chat.completion.chunk",
3737 model: model_name,
3738 choices: vec![ChatCompletionChunkChoice {
3739 index: 0,
3740 delta: ChatCompletionChunkDelta {
3741 role: Some("assistant"),
3742 content: Some(format!("[error: {e}]")),
3743 reasoning_content: None,
3744 tool_calls: None,
3745 },
3746 finish_reason: Some("stop"),
3747 }],
3748 usage: None,
3749 };
3750 let _ = sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3751 let _ = sse::send_or_orphan(&tx, Ok(emitter.done()), orphan_timeout);
3752 }
3753 }
3754 drop(emitter);
3758 });
3759
3760 let stream = sse::with_keepalive(rx, keepalive, sse::KEEPALIVE_INTERVAL);
3761 Ok((
3784 [(
3785 axum::http::HeaderName::from_static("x-accel-buffering"),
3786 axum::http::HeaderValue::from_static("no"),
3787 )],
3788 Sse::new(stream),
3789 )
3790 .into_response())
3791}
3792
3793fn axum_path(template: &str) -> String {
3810 let mut out = String::with_capacity(template.len());
3811 let mut rest = template;
3812 while let Some(open) = rest.find('{') {
3813 let Some(close) = rest[open..].find('}').map(|c| open + c) else {
3814 break;
3815 };
3816 out.push_str(&rest[..open]);
3817 out.push(':');
3818 out.push_str(&rest[open + 1..close]);
3819 rest = &rest[close + 1..];
3820 }
3821 out.push_str(rest);
3822 out
3823}
3824
3825async fn cancel_generation(
3834 State(state): State<Arc<AppState>>,
3835 Json(req): Json<ferrox_api::CancelGenerationRequest>,
3836) -> Response {
3837 let cancelled = state.cancels.cancel(&req.request_id);
3838 let status = if cancelled {
3839 StatusCode::OK
3840 } else {
3841 StatusCode::NOT_FOUND
3842 };
3843 let detail = if cancelled {
3844 "the generation was asked to stop; it ends at its next token".to_string()
3845 } else {
3846 "no generation with that request_id is running -- it has already \
3847 finished, was never issued, or was served by a path that does \
3848 not register for cancellation"
3849 .to_string()
3850 };
3851 (
3852 status,
3853 Json(ferrox_api::CancelGenerationResponse {
3854 request_id: req.request_id,
3855 cancelled,
3856 detail,
3857 }),
3858 )
3859 .into_response()
3860}
3861
3862type Activated = (
3866 Loaded,
3867 Option<serving::batch::ContinuousBatcher>,
3868 Option<Arc<budget::ContextCeiling>>,
3869);
3870
3871fn price_batcher_config(path: Option<&str>) -> serving::batch::BatcherConfig {
3889 let mut batcher = serving::batch::BatcherConfig::from_env();
3890 if batcher.max_context.is_some() && batcher.kv_blocks.is_some() {
3891 return batcher;
3894 }
3895 let Some(path) = path else {
3896 return batcher;
3897 };
3898 let priced = budget::price_gguf(path, ferrox_models::KvElem::F32, 1);
3904 let Some((priced, gguf_ctx, source)) = priced else {
3905 return batcher;
3906 };
3907 let Some(derived) = budget::derive_limits(&priced, gguf_ctx, batcher.kv_block_size) else {
3908 tracing::warn!(
3912 "this checkpoint's weights leave no room for KV inside the {source}: {} weight \
3913 bytes against a {} byte budget. Serving with no derived context ceiling -- set \
3914 FERROX_DEVICE_BUDGET_BYTES if the probe is wrong, or FERROX_CB_MAX_CONTEXT to \
3915 admit on a number you choose.",
3916 priced.weights_bytes,
3917 priced.device_budget_bytes,
3918 );
3919 return batcher;
3920 };
3921 tracing::info!("{source}");
3922 tracing::info!("{}", derived.fit);
3923 let adopted = budget::apply_derived(&mut batcher, &derived);
3924 if adopted.max_context {
3925 tracing::info!(
3926 "derived per-request context ceiling: {} token positions (prompt + max_tokens); \
3927 override with FERROX_CB_MAX_CONTEXT",
3928 derived.max_context
3929 );
3930 }
3931 if adopted.kv_blocks {
3932 tracing::info!(
3933 "derived KV block budget: {} blocks x {} positions; override with FERROX_CB_KV_BLOCKS",
3934 derived.kv_blocks,
3935 batcher.kv_block_size
3936 );
3937 }
3938 batcher
3939}
3940
3941pub(crate) fn activate_loaded_model(
3950 loaded: model::LoadedModel,
3951 enable_continuous_batching: bool,
3952 path: Option<&str>,
3953 paged_kv: Option<&generate::PagedKvConfig>,
3954) -> Activated {
3955 match loaded {
3956 model::LoadedModel::Gguf(g) => {
3957 let decoder = Arc::new(g.decoder);
3958 let tokenizer = Arc::new(g.tokenizer);
3959 let config = price_batcher_config(path);
3960 let shape =
3966 ferrox_models::KvShape::from_config(&decoder.config, ferrox_models::KvElem::F32);
3967 let ceiling = Arc::new(budget::ContextCeiling::new(config.max_context, shape));
3968 let batcher = if enable_continuous_batching {
3969 tracing::info!(
3970 "continuous batching enabled: decode steps share Decoder::forward_multi_seq \
3971 (stop sequences use the same pending-buffer trim as the private generate loop)"
3972 );
3973 let tok = Arc::clone(&tokenizer);
3974 let decode = Arc::new(move |ids: &[usize]| tok.decode_bytes(ids));
3975 Some(serving::batch::ContinuousBatcher::spawn_with_ceiling(
3976 Arc::clone(&decoder),
3977 decode,
3978 config,
3979 Arc::clone(&ceiling),
3980 paged_kv.cloned(),
3981 ))
3982 } else {
3983 None
3984 };
3985 (
3986 Loaded::Generative(Arc::new(Model::Gguf(GgufModel {
3987 decoder,
3988 tokenizer,
3989 stop_tokens: g.stop_tokens,
3990 bos_id: g.bos_id,
3991 is_synthetic: g.is_synthetic,
3992 chat_template: g.chat_template,
3993 }))),
3994 batcher,
3995 Some(ceiling),
3996 )
3997 }
3998 model::LoadedModel::Kimi(k) => (
3999 Loaded::Generative(Arc::new(Model::Kimi(KimiModel {
4000 engine: k.engine,
4001 tokenizer: k.tokenizer,
4002 stop_tokens: k.stop_tokens,
4003 chat_template: k.chat_template,
4004 }))),
4005 None,
4006 None,
4007 ),
4008 model::LoadedModel::Mla(m) => (
4009 Loaded::Generative(Arc::new(Model::Mla(MlaModel {
4010 engine: m.engine,
4011 tokenizer: m.tokenizer,
4012 stop_tokens: m.stop_tokens,
4013 bos_id: m.bos_id,
4014 name: m.name,
4015 chat_template: m.chat_template,
4016 }))),
4017 None,
4018 None,
4019 ),
4020 model::LoadedModel::Gemma4(m) => (
4021 Loaded::Generative(Arc::new(Model::Gemma4(Gemma4Model {
4022 engine: m.engine,
4023 tokenizer: m.tokenizer,
4024 stop_tokens: m.stop_tokens,
4025 bos_id: m.bos_id,
4026 name: m.name,
4027 chat_template: m.chat_template,
4028 }))),
4029 None,
4030 None,
4031 ),
4032 model::LoadedModel::Glm52(g) => (
4033 Loaded::Generative(Arc::new(Model::Glm52(Glm52Model {
4034 engine: g.engine,
4035 tokenizer: g.tokenizer,
4036 stop_tokens: g.stop_tokens,
4037 bos_id: g.bos_id,
4038 name: g.name,
4039 chat_template: g.chat_template,
4040 }))),
4041 None,
4042 None,
4043 ),
4044 model::LoadedModel::Encoder(e) => (Loaded::Encoder(e), None, None),
4049 }
4050}
4051
4052struct StartupModels {
4059 loaded: model::LoadedModel,
4060 embedding: Option<Arc<ferrox_models::EmbeddingModel>>,
4061}
4062
4063fn continuous_batching_env() -> Option<bool> {
4064 match std::env::var("FERROX_CONTINUOUS_BATCHING")
4065 .ok()
4066 .map(|v| v.trim().to_ascii_lowercase())
4067 .as_deref()
4068 {
4069 None => None,
4070 Some("1" | "true" | "yes" | "on") => Some(true),
4071 Some("0" | "false" | "no" | "off") => Some(false),
4072 _ => None,
4073 }
4074}
4075
4076fn metal_private_decode_active() -> bool {
4077 #[cfg(feature = "metal")]
4078 {
4079 BUILT_WITH_METAL
4080 && ferrox_metal::attn::metal_attn_enabled()
4081 && std::env::var("FERROX_METAL").ok().as_deref() != Some("0")
4082 }
4083 #[cfg(not(feature = "metal"))]
4084 {
4085 false
4086 }
4087}
4088
4089fn continuous_batching_compatible(
4090 loaded: &model::LoadedModel,
4091 kv_pool: &Option<generate::KvPoolConfig>,
4092 prefix_cache: &Option<Arc<Mutex<PrefixCache>>>,
4093 paged_kv: &Option<generate::PagedKvConfig>,
4094) -> bool {
4095 matches!(loaded, model::LoadedModel::Gguf(_))
4096 && (paged_kv.is_some() || (kv_pool.is_none() && prefix_cache.is_none()))
4097}
4098
4099fn resolve_continuous_batching_enabled(
4100 loaded: &model::LoadedModel,
4101 kv_pool: &Option<generate::KvPoolConfig>,
4102 prefix_cache: &Option<Arc<Mutex<PrefixCache>>>,
4103 paged_kv: &Option<generate::PagedKvConfig>,
4104) -> bool {
4105 if !continuous_batching_compatible(loaded, kv_pool, prefix_cache, paged_kv) {
4106 return false;
4107 }
4108 match continuous_batching_env() {
4109 Some(true) => true,
4110 Some(false) => false,
4111 None => metal_private_decode_active(),
4112 }
4113}
4114
4115fn acquire_metal_private_decode_gate(
4116 gate: Option<&std::sync::Mutex<()>>,
4117 used_batcher: bool,
4118) -> Option<std::sync::MutexGuard<'_, ()>> {
4119 if used_batcher {
4120 None
4121 } else {
4122 gate.map(|g| g.lock().unwrap_or_else(|p| p.into_inner()))
4123 }
4124}
4125
4126fn build_app_state(
4127 models: StartupModels,
4128 kv_pool: Option<generate::KvPoolConfig>,
4129 paged_kv: Option<generate::PagedKvConfig>,
4130 prefix_cache: Option<Arc<Mutex<PrefixCache>>>,
4131 enable_continuous_batching: bool,
4132 mcp: Option<mcp::LoadedMcpConfig>,
4133 detection: Arc<health::Detection>,
4134) -> AppState {
4135 let StartupModels { loaded, embedding } = models;
4136 let (loaded, batcher, ceiling) = activate_loaded_model(
4137 loaded,
4138 enable_continuous_batching,
4139 std::env::var("FERROX_MODEL_PATH").ok().as_deref(),
4140 paged_kv.as_ref(),
4141 );
4142 let id = startup_model_id();
4148 let metal_private_decode_gate = if enable_continuous_batching || !metal_private_decode_active()
4149 {
4150 None
4151 } else {
4152 tracing::info!(
4153 "Metal private-loop decode will serialize concurrent requests until \
4154 continuous batching is enabled (FERROX_CONTINUOUS_BATCHING=1 or --cont-batching)"
4155 );
4156 Some(Arc::new(std::sync::Mutex::new(())))
4157 };
4158 AppState {
4159 embedding,
4160 active: std::sync::RwLock::new(Some(Arc::new(ActiveModel {
4161 id,
4162 loaded,
4163 batcher,
4164 ceiling,
4165 }))),
4166 paged_kv,
4167 load_in_progress: std::sync::atomic::AtomicBool::new(false),
4168 tasks: Arc::new(tasks::TaskRegistry::new()),
4169 cancels: Arc::new(cancel::CancelRegistry::new()),
4170 stats: stats::Stats::new(),
4171 streams: resume::StreamRegistry::new(),
4172 model_dir: admin::model_dirs().into_iter().next(),
4173 response_cache: Mutex::new(ResponseCache::new(1000, Duration::from_secs(3600))),
4174 kv_pool,
4175 prefix_cache,
4176 sessions: session::SessionStore::new(),
4177 requests_total: std::sync::atomic::AtomicU64::new(0),
4178 request_errors_total: std::sync::atomic::AtomicU64::new(0),
4179 started_at: std::time::Instant::now(),
4180 last_request_ms: std::sync::atomic::AtomicU64::new(0),
4181 detection,
4182 mcp,
4183 continuous_batching_enabled: enable_continuous_batching,
4184 metal_private_decode_gate,
4185 loading_model: Mutex::new(None),
4186 last_load_error: Mutex::new(None),
4187 serving: Mutex::new(crate::stats::ServingStats::default()),
4188 maintenance: Mutex::new(crate::policy::maintenance::MaintenanceGate::serving()),
4189 footprint: Mutex::new(crate::policy::footprint::ProbeCache::new(FOOTPRINT_TTL_MS)),
4190 started_unix: unix_now(),
4191 }
4192}
4193
4194fn load_embedding_model() -> anyhow::Result<Option<Arc<ferrox_models::EmbeddingModel>>> {
4202 let Ok(path) = std::env::var("FERROX_EMBEDDING_MODEL_PATH") else {
4203 return Ok(None);
4204 };
4205 let model = ferrox_models::EmbeddingModel::from_gguf_path(&path)
4206 .map_err(|e| anyhow::anyhow!("FERROX_EMBEDDING_MODEL_PATH={path}: {e}"))?;
4207 tracing::info!(
4208 "loaded embedding model '{}' ({}, {} dims, pooling {}, max {} tokens)",
4209 model.name(),
4210 model.architecture(),
4211 model.n_embd(),
4212 model.pooling_type().name(),
4213 model.n_ctx_train(),
4214 );
4215 Ok(Some(Arc::new(model)))
4216}
4217
4218fn unix_now() -> u64 {
4223 std::time::SystemTime::now()
4224 .duration_since(std::time::UNIX_EPOCH)
4225 .map(|d| d.as_secs())
4226 .unwrap_or(0)
4227}
4228
4229fn startup_model_id() -> Option<String> {
4234 let configured = std::env::var("FERROX_MODEL_PATH").ok()?;
4235 let configured = std::fs::canonicalize(&configured).ok()?;
4236 admin::discover(&admin::model_dirs())
4237 .into_iter()
4238 .find(|d| {
4239 std::fs::canonicalize(&d.path)
4240 .map(|p| p == configured)
4241 .unwrap_or(false)
4242 })
4243 .map(|d| d.id)
4244}
4245
4246fn init_cpu_pool() {
4255 match ferrox_core::threads::init_cpu_pool() {
4256 Some(n) => eprintln!(
4257 "ferrox-server: rayon pool {n} threads (perf cores {}; override with FERROX_CPU_THREADS)",
4258 ferrox_core::threads::perf_core_count()
4259 ),
4260 None => eprintln!("ferrox-server: global rayon pool already built; leaving it alone"),
4261 }
4262}
4263
4264fn announce_ready(addr: SocketAddr, scheme: &str) {
4277 use std::io::Write;
4278 let ready =
4279 ferrox_api::ServerReady::new(addr, scheme, env!("CARGO_PKG_VERSION"), std::process::id());
4280 let mut stdout = std::io::stdout().lock();
4281 let _ = writeln!(stdout, "{}", ready.to_line());
4282 let _ = stdout.flush();
4283}
4284
4285async fn shutdown_signal(exit_on_stdin_close: bool) {
4297 if !exit_on_stdin_close {
4298 std::future::pending::<()>().await;
4299 return;
4300 }
4301 let _ = tokio::task::spawn_blocking(|| {
4302 use std::io::Read;
4303 let mut sink = [0u8; 256];
4304 let mut stdin = std::io::stdin().lock();
4305 loop {
4306 match stdin.read(&mut sink) {
4307 Ok(0) => break,
4309 Ok(_) => continue,
4311 Err(e) => {
4312 tracing::warn!("stdin read failed ({e}); treating it as closed");
4313 break;
4314 }
4315 }
4316 }
4317 })
4318 .await;
4319 tracing::info!("stdin closed; shutting down");
4320}
4321
4322fn tokio_worker_threads() -> usize {
4327 std::env::var("FERROX_TOKIO_WORKERS")
4328 .ok()
4329 .and_then(|v| v.trim().parse::<usize>().ok())
4330 .filter(|n| *n > 0)
4331 .unwrap_or(2)
4332}
4333
4334fn install_ring_crypto_provider() {
4351 let _ = rustls::crypto::ring::default_provider().install_default();
4352}
4353
4354pub fn run_server(args: ServerArgs) -> anyhow::Result<()> {
4360 if args.list_devices {
4361 print_available_devices();
4362 return Ok(());
4363 }
4364 apply_cli_overrides(&args)?;
4365
4366 let _instance = {
4370 use ferrox_core::instance::{register, InstancePolicy};
4371 let policy = if args.allow_multiple_instances {
4372 InstancePolicy::Multi
4373 } else {
4374 InstancePolicy::from_env_or(InstancePolicy::Single)
4375 };
4376 let model = std::env::var("FERROX_MODEL_PATH").ok();
4377 register(
4378 "server",
4379 model.as_deref(),
4380 ferrox_core::instance::current_backend(),
4381 policy,
4382 )
4383 .map_err(|conflict| anyhow::anyhow!("{conflict}"))?
4384 };
4385
4386 let journal = journal::Journal::from_env();
4387 eprintln!(
4388 "ferrox-server: process lifecycle journal at {:?} (override with FERROX_JOURNAL_PATH)",
4389 journal.path()
4390 );
4391 journal.append(&journal::Record::session_start(
4392 env!("CARGO_PKG_VERSION"),
4393 std::process::id(),
4394 ));
4395 journal::install_panic_hook(journal.clone());
4396
4397 let mcp_config_path = args.mcp_config.clone();
4398 let exit_on_stdin_close = args.exit_on_stdin_close
4399 || std::env::var("FERROX_EXIT_ON_STDIN_CLOSE")
4400 .map(|v| v == "1")
4401 .unwrap_or(false);
4402
4403 unsafe { ferrox_core::weight_matrix::default_cpu_int_dot_on() };
4407 init_cpu_pool();
4408
4409 let runtime = tokio::runtime::Builder::new_multi_thread()
4410 .worker_threads(tokio_worker_threads())
4411 .enable_all()
4412 .build()?;
4413 let result = runtime.block_on(run(mcp_config_path, exit_on_stdin_close));
4414
4415 let reason = match &result {
4416 Ok(()) => "normal".to_string(),
4417 Err(e) => e.to_string(),
4418 };
4419 journal.append(&journal::Record::session_exit(reason));
4420
4421 runtime.shutdown_background();
4426
4427 result
4428}
4429
4430async fn run(mcp_config_path: Option<PathBuf>, exit_on_stdin_close: bool) -> anyhow::Result<()> {
4431 let _ = tracing_subscriber::fmt::try_init();
4437
4438 let addr = std::env::var("FERROX_ADDR").unwrap_or_else(|_| "127.0.0.1:8383".to_string());
4448 let api_key_configured = std::env::var("FERROX_API_KEY").is_ok();
4449 let allow_unauthenticated_remote = std::env::var("FERROX_ALLOW_UNAUTHENTICATED_REMOTE")
4450 .map(|v| v == "1")
4451 .unwrap_or(false);
4452 if let Err(msg) =
4453 security::check_bind_authorization(&addr, api_key_configured, allow_unauthenticated_remote)
4454 {
4455 anyhow::bail!(msg);
4456 }
4457
4458 let embedding_model = load_embedding_model()?;
4465
4466 let mut loaded = model::load()?;
4467 match &loaded {
4468 model::LoadedModel::Gguf(g) => tracing::info!(
4469 "loaded GGUF model '{}' (synthetic={}, tokenizer={})",
4470 g.decoder.config.name,
4471 g.is_synthetic,
4472 g.tokenizer.kind()
4473 ),
4474 model::LoadedModel::Kimi(k) => tracing::info!(
4475 "loaded Kimi K3 checkpoint (tokenizer={} base tokens)",
4476 k.tokenizer.vocab_size()
4477 ),
4478 model::LoadedModel::Mla(m) => tracing::info!(
4479 "loaded MLA GGUF '{}' (tokenizer={})",
4480 m.name,
4481 m.tokenizer.kind()
4482 ),
4483 model::LoadedModel::Gemma4(m) => tracing::info!(
4484 "loaded Gemma4 GGUF '{}' (tokenizer={})",
4485 m.name,
4486 m.tokenizer.kind()
4487 ),
4488 model::LoadedModel::Glm52(g) => tracing::info!(
4489 "loaded GLM-5.2 GGUF '{}' (tokenizer={})",
4490 g.name,
4491 g.tokenizer.kind()
4492 ),
4493 model::LoadedModel::Encoder(_) => {}
4496 }
4497 let metal_default_moe_budget = {
4505 #[cfg(feature = "metal")]
4506 {
4507 ferrox_core::metal_dense_enabled()
4508 && std::env::var("FERROX_GPU_VRAM_BUDGET_BYTES").is_err()
4509 }
4510 #[cfg(not(feature = "metal"))]
4511 {
4512 false
4513 }
4514 };
4515 if let Ok(budget_str) = std::env::var("FERROX_GPU_VRAM_BUDGET_BYTES") {
4516 let budget: u64 = budget_str
4517 .parse()
4518 .expect("FERROX_GPU_VRAM_BUDGET_BYTES must be a non-negative integer");
4519 match &mut loaded {
4520 model::LoadedModel::Gguf(g) => {
4521 tracing::info!(
4522 "GPU expert placement enabled: {budget} byte VRAM budget for routed experts \
4523 (CUDA and/or Metal matvecs when built with the matching feature)"
4524 );
4525 g.decoder.gpu_vram_budget_bytes = Some(budget);
4526 }
4527 model::LoadedModel::Kimi(_) => {
4528 tracing::warn!(
4529 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is Kimi K3 -- not \
4530 supported yet (its MoE stack isn't wired to PlacementPlan), ignoring"
4531 );
4532 }
4533 model::LoadedModel::Mla(_) => {
4534 tracing::warn!(
4535 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is MLA -- dense \
4536 FFN path only today; ignoring expert VRAM budget"
4537 );
4538 }
4539 model::LoadedModel::Gemma4(_) => {
4540 tracing::warn!(
4541 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is Gemma4 -- \
4542 ignoring expert VRAM budget"
4543 );
4544 }
4545 model::LoadedModel::Glm52(_) => {
4546 tracing::warn!(
4547 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is GLM-5.2 DSA -- \
4548 GPU expert placement not wired yet; ignoring"
4549 );
4550 }
4551 model::LoadedModel::Encoder(_) => {
4552 tracing::warn!(
4553 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is an encoder -- \
4554 it has no routed experts to place; ignoring"
4555 );
4556 }
4557 }
4558 } else if metal_default_moe_budget {
4559 const METAL_DEFAULT_MOE_BUDGET: u64 = 64 * 1024 * 1024 * 1024;
4563 if let model::LoadedModel::Gguf(g) = &mut loaded {
4564 tracing::info!(
4565 "Metal MoE expert placement default-on ({METAL_DEFAULT_MOE_BUDGET} byte budget); \
4566 set FERROX_GPU_VRAM_BUDGET_BYTES=0 to force CPU experts"
4567 );
4568 g.decoder.gpu_vram_budget_bytes = Some(METAL_DEFAULT_MOE_BUDGET);
4569 }
4570 }
4571 #[cfg(feature = "cuda")]
4572 {
4573 if ferrox_core::cuda_dense_enabled() {
4574 tracing::info!(
4575 "CUDA dense matvec enabled for WeightMatrix::apply \
4576 (FERROX_CUDA=0|cpu forces CPU; weight buffers stay resident after first upload)"
4577 );
4578 } else {
4579 tracing::info!(
4580 "CUDA dense matvec disabled (FERROX_CUDA); dense decode uses CPU or Metal"
4581 );
4582 }
4583 }
4584 #[cfg(feature = "metal")]
4585 {
4586 if ferrox_core::metal_dense_enabled() {
4587 tracing::info!(
4588 "Metal dense matvec enabled for WeightMatrix::apply \
4589 (FERROX_METAL=0|cpu forces CPU; weight buffers stay resident after first upload)"
4590 );
4591 match std::env::var("FERROX_METAL_ATTN").ok().as_deref() {
4592 Some("1") | Some("true") | Some("on") | Some("attn") => {
4593 tracing::info!(
4594 "Metal fused attention requested (FERROX_METAL_ATTN): \
4595 QKV→RoPE→GQA→O on-GPU for Norm/NeoX decode without QKV bias/QK-norm"
4596 );
4597 }
4598 _ => {}
4599 }
4600 tracing::info!(
4601 "Metal greedy GPU argmax: temperature<=0 folds \
4602 final_norm+lm_head+argmax into the dense stack"
4603 );
4604 } else {
4605 tracing::info!("Metal dense matvec disabled (FERROX_METAL); dense decode uses CPU");
4606 }
4607 }
4608 let blocks_env = std::env::var("FERROX_KV_POOL_BLOCKS");
4619 let block_size_env = std::env::var("FERROX_KV_POOL_BLOCK_SIZE");
4620 let byte_budget_env = std::env::var("FERROX_KV_BYTE_BUDGET");
4621 if blocks_env.is_ok() && byte_budget_env.is_ok() {
4622 panic!(
4623 "FERROX_KV_POOL_BLOCKS and FERROX_KV_BYTE_BUDGET are mutually exclusive \
4624 (set one block-count source plus FERROX_KV_POOL_BLOCK_SIZE, or neither to disable)"
4625 );
4626 }
4627 let kv_pool = match (blocks_env, block_size_env, byte_budget_env) {
4628 (Ok(blocks), Ok(block_size), Err(_)) => {
4629 let total_blocks: usize = blocks
4630 .parse()
4631 .expect("FERROX_KV_POOL_BLOCKS must be a positive integer");
4632 let block_size: usize = block_size
4633 .parse()
4634 .expect("FERROX_KV_POOL_BLOCK_SIZE must be a positive integer");
4635 let queue_wait_ms: u64 = std::env::var("FERROX_KV_POOL_QUEUE_TIMEOUT_MS")
4641 .ok()
4642 .map(|v| {
4643 v.parse()
4644 .expect("FERROX_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4645 })
4646 .unwrap_or(0);
4647 tracing::info!(
4648 "KV cache block pool enabled: {total_blocks} blocks x {block_size} positions \
4649 each, shared across all concurrent requests, {queue_wait_ms}ms admission queue wait"
4650 );
4651 Some(generate::KvPoolConfig {
4652 pool: Arc::new(Mutex::new(KvBlockPool::new(block_size, total_blocks))),
4653 queue_wait: Duration::from_millis(queue_wait_ms),
4654 })
4655 }
4656 (Err(_), Ok(block_size), Ok(byte_budget)) => {
4657 let block_size: usize = block_size
4658 .parse()
4659 .expect("FERROX_KV_POOL_BLOCK_SIZE must be a positive integer");
4660 let budget: u64 = byte_budget
4661 .parse()
4662 .expect("FERROX_KV_BYTE_BUDGET must be a positive integer");
4663 let cfg = match &loaded {
4664 model::LoadedModel::Gguf(g) => &g.decoder.config,
4665 model::LoadedModel::Kimi(_)
4666 | model::LoadedModel::Mla(_)
4667 | model::LoadedModel::Gemma4(_)
4668 | model::LoadedModel::Glm52(_)
4669 | model::LoadedModel::Encoder(_) => {
4670 panic!(
4671 "FERROX_KV_BYTE_BUDGET requires a GGUF decoder model \
4672 (set FERROX_MODEL_PATH to a generic-decoder .gguf file)"
4673 );
4674 }
4675 };
4676 let bytes_per_block = block_size
4677 * cfg.n_layers
4678 * cfg.n_kv_heads
4679 * cfg.head_dim
4680 * 2
4681 * std::mem::size_of::<f32>();
4682 assert!(
4683 bytes_per_block > 0,
4684 "derived KV block byte size must be positive (check model config and block size)"
4685 );
4686 let total_blocks = (budget as usize / bytes_per_block).max(1);
4687 let queue_wait_ms: u64 = std::env::var("FERROX_KV_POOL_QUEUE_TIMEOUT_MS")
4688 .ok()
4689 .map(|v| {
4690 v.parse()
4691 .expect("FERROX_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4692 })
4693 .unwrap_or(0);
4694 tracing::info!(
4695 "KV cache block pool enabled from byte budget: {budget} bytes / \
4696 {bytes_per_block} bytes per block ({block_size} positions x {} layers) -> \
4697 {total_blocks} blocks, {queue_wait_ms}ms admission queue wait",
4698 cfg.n_layers
4699 );
4700 Some(generate::KvPoolConfig {
4701 pool: Arc::new(Mutex::new(KvBlockPool::new(block_size, total_blocks))),
4702 queue_wait: Duration::from_millis(queue_wait_ms),
4703 })
4704 }
4705 (Err(_), Err(_), Err(_)) => None,
4706 (Err(_), Ok(_), Err(_)) => panic!(
4707 "FERROX_KV_POOL_BLOCK_SIZE requires FERROX_KV_POOL_BLOCKS or FERROX_KV_BYTE_BUDGET \
4708 (or unset all three to disable KV cache pooling)"
4709 ),
4710 (Ok(_), Ok(_), Ok(_)) => {
4711 unreachable!("FERROX_KV_POOL_BLOCKS and FERROX_KV_BYTE_BUDGET are mutually exclusive")
4712 }
4713 (Ok(_), Err(_), _) | (Err(_), Err(_), Ok(_)) => panic!(
4714 "FERROX_KV_POOL_BLOCKS/FERROX_KV_BYTE_BUDGET and FERROX_KV_POOL_BLOCK_SIZE must be \
4715 set together (or neither, to disable KV cache pooling)"
4716 ),
4717 };
4718 let paged_kv = match (
4724 std::env::var("FERROX_PAGED_KV_BLOCKS"),
4725 std::env::var("FERROX_PAGED_KV_BLOCK_SIZE"),
4726 ) {
4727 (Ok(blocks), Ok(block_size)) => {
4728 assert!(
4729 kv_pool.is_none(),
4730 "FERROX_PAGED_KV_BLOCKS and FERROX_KV_POOL_BLOCKS/FERROX_KV_BYTE_BUDGET are \
4731 mutually exclusive: both bound the same KV memory, by different means. \
4732 Set one."
4733 );
4734 let blocks_per_layer: usize = blocks
4748 .parse()
4749 .expect("FERROX_PAGED_KV_BLOCKS must be a positive integer");
4750 let block_size: usize = block_size
4751 .parse()
4752 .expect("FERROX_PAGED_KV_BLOCK_SIZE must be a positive integer");
4753 let gguf = match &loaded {
4754 model::LoadedModel::Gguf(g) => g,
4755 _ => panic!(
4756 "FERROX_PAGED_KV_BLOCKS requires a GGUF decoder model \
4757 (set FERROX_MODEL_PATH to a generic-decoder .gguf file)"
4758 ),
4759 };
4760 let cfg = &gguf.decoder.config;
4761 let queue_wait_ms: u64 = std::env::var("FERROX_KV_POOL_QUEUE_TIMEOUT_MS")
4762 .ok()
4763 .map(|v| {
4764 v.parse()
4765 .expect("FERROX_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4766 })
4767 .unwrap_or(0);
4768 tracing::info!(
4769 "Paged KV enabled: {blocks_per_layer} blocks x {block_size} positions per \
4770 layer across {} layers, shared by all concurrent requests, \
4771 {queue_wait_ms}ms admission queue wait",
4772 cfg.n_layers
4773 );
4774 let radix = Some(Arc::new(Mutex::new(crate::policy::radix::RadixCache::new(
4779 block_size,
4780 ))));
4781 let anchor_token = crate::policy::anchor::resolve_anchor_token(
4790 crate::policy::parser::ToolCallFormat::infer(
4791 &std::env::var("FERROX_MODEL_PATH").unwrap_or_default(),
4792 )
4793 .opener(),
4794 |text| {
4795 gguf.tokenizer
4796 .encode(text)
4797 .into_iter()
4798 .map(|t| t as u32)
4799 .collect()
4800 },
4801 );
4802 if let Some(id) = anchor_token {
4803 tracing::info!(
4804 "Paged KV window slide: tool-call anchor is token {id}, so a turn's \
4805 window stops short of where its next turn rejoins"
4806 );
4807 }
4808 let slide_interval: usize = std::env::var("FERROX_PAGED_KV_SLIDE_INTERVAL")
4809 .ok()
4810 .map(|v| {
4811 v.parse()
4812 .expect("FERROX_PAGED_KV_SLIDE_INTERVAL must be a positive integer")
4813 })
4814 .unwrap_or(crate::policy::pool_budget::DEFAULT_SWA_EVICTION_INTERVAL);
4815 if let Some(window) = cfg.uniform_sliding_window() {
4816 tracing::info!(
4817 "Paged KV window slide enabled: every layer slides by {window} every \
4818 {slide_interval} decode steps, so a request holds its prompt and a \
4819 window rather than its whole context"
4820 );
4821 } else if cfg.kv_block_window().is_some() {
4822 tracing::info!(
4823 "Paged KV window slide NOT enabled: this model has full-attention layers, \
4824 and a page group holds one block in every layer"
4825 );
4826 }
4827 Some(generate::PagedKvConfig {
4828 store: Arc::new(ferrox_core::cache::SharedPagedKv::new(
4829 cfg.n_layers,
4830 block_size,
4831 blocks_per_layer,
4832 cfg.n_kv_heads,
4833 cfg.head_dim,
4834 )),
4835 queue_wait: Duration::from_millis(queue_wait_ms),
4836 radix,
4837 anchor_token,
4838 slide_interval,
4839 })
4840 }
4841 (Err(_), Err(_)) => None,
4842 _ => panic!(
4843 "FERROX_PAGED_KV_BLOCKS and FERROX_PAGED_KV_BLOCK_SIZE must be set together \
4844 (or neither, to disable paged KV)"
4845 ),
4846 };
4847 let prefix_cache = std::env::var("FERROX_PREFIX_CACHE_ENTRIES").ok().map(|v| {
4854 let max_entries: usize = v
4855 .parse()
4856 .expect("FERROX_PREFIX_CACHE_ENTRIES must be a positive integer");
4857 if kv_pool.is_some() {
4858 tracing::warn!(
4859 "FERROX_PREFIX_CACHE_ENTRIES is set but so is the KV pool -- prefix \
4860 caching will never be consulted while a KV pool is configured"
4861 );
4862 }
4863 assert!(
4870 paged_kv.is_none(),
4871 "FERROX_PREFIX_CACHE_ENTRIES and FERROX_PAGED_KV_BLOCKS are mutually exclusive: \
4872 the prefix cache stores contiguous KV snapshots, which a paged request does not \
4873 produce, so the cache could never hit. Set one."
4874 );
4875 tracing::info!(
4876 "KV-prefix cache enabled: up to {max_entries} stored prefixes, shared across \
4877 all requests"
4878 );
4879 Arc::new(Mutex::new(PrefixCache::new(max_entries)))
4880 });
4881 if matches!(
4882 loaded,
4883 model::LoadedModel::Kimi(_) | model::LoadedModel::Mla(_) | model::LoadedModel::Glm52(_)
4884 ) && (kv_pool.is_some() || prefix_cache.is_some())
4885 {
4886 tracing::warn!(
4887 "KV pool / prefix cache are configured but the loaded model is Kimi, MLA, or GLM-5.2 -- \
4888 neither is consulted for those engines (state shapes differ from Decoder KV); see \
4889 ferrox_models::engine's module docs"
4890 );
4891 }
4892 let enable_cb =
4893 resolve_continuous_batching_enabled(&loaded, &kv_pool, &prefix_cache, &paged_kv);
4894 if enable_cb && continuous_batching_env().is_none() && metal_private_decode_active() {
4895 tracing::info!(
4896 "continuous batching enabled by default on Metal for safe parallel serving \
4897 (set FERROX_CONTINUOUS_BATCHING=0 or --no-cont-batching to use the private path)"
4898 );
4899 }
4900 if continuous_batching_env() == Some(true)
4901 && !continuous_batching_compatible(&loaded, &kv_pool, &prefix_cache, &paged_kv)
4902 && (kv_pool.is_some() || prefix_cache.is_some())
4903 {
4904 tracing::warn!(
4905 "FERROX_CONTINUOUS_BATCHING=1 ignored while KV pool or prefix cache is configured \
4906 (those modes keep the private generate path)"
4907 );
4908 }
4909 if let Ok(n) = std::env::var("FERROX_CHUNKED_PREFILL") {
4910 if let Ok(chunk) = n.parse::<usize>() {
4911 if chunk > 0 {
4912 tracing::info!("chunked prefill enabled: {chunk} tokens per forward_batch chunk");
4913 }
4914 }
4915 }
4916 if matches!(
4917 std::env::var("FERROX_CPU_KV_OFFLOAD").ok().as_deref(),
4918 Some("1")
4919 ) {
4920 tracing::warn!(
4921 "FERROX_CPU_KV_OFFLOAD=1: syncing Metal KV to host after each decode step \
4922 (minimal spill; full layer offload still planned)"
4923 );
4924 }
4925
4926 let mcp = match mcp_config_path {
4927 Some(path) => {
4928 let loaded = mcp::load_mcp_config(&path)?;
4929 tracing::info!(
4930 "MCP config loaded from {} ({} server(s); invocation not wired yet)",
4931 loaded.path,
4932 loaded.servers.len()
4933 );
4934 Some(loaded)
4935 }
4936 None => None,
4937 };
4938
4939 let detection = health::Detection::spawn();
4943
4944 let state = Arc::new(build_app_state(
4945 StartupModels {
4946 loaded,
4947 embedding: embedding_model,
4948 },
4949 kv_pool,
4950 paged_kv,
4951 prefix_cache,
4952 enable_cb,
4953 mcp,
4954 detection,
4955 ));
4956
4957 use ferrox_api::routes;
4961
4962 let public = Router::new().route(routes::HEALTH, get(health));
4967
4968 let mut protected = Router::new()
4969 .route(routes::V1_MODELS, get(list_models))
4970 .route(routes::V1_RESPONSES, post(responses::responses))
4974 .route(
4975 &axum_path(routes::V1_RESPONSE),
4976 get(responses::responses_get),
4977 )
4978 .route(
4979 &axum_path(routes::V1_RESPONSE_CANCEL),
4980 post(responses::responses_cancel),
4981 )
4982 .route(routes::V1_STATS, get(serving_stats))
4983 .route(routes::V1_REQUESTS, get(recent_requests))
4984 .route(routes::V1_CACHE_STATUS, get(cache_admin::cache_status))
4985 .route(routes::V1_CACHE_REBUILD, post(cache_admin::cache_rebuild))
4986 .route(routes::ADMIN_PREPARE_STOP, post(cache_admin::prepare_stop))
4987 .route(routes::V1_CHAT_COMPLETIONS, post(chat_completions))
4988 .route(routes::V1_CANCEL, post(cancel_generation))
4992 .route(&axum_path(routes::V1_STREAM), get(resume::resume))
4997 .route(&axum_path(routes::V1_STREAM_POLL), get(resume::poll))
4998 .route(routes::V1_MESSAGES, post(anthropic::messages))
4999 .route(
5000 routes::V1_MESSAGES_COUNT_TOKENS,
5001 post(anthropic::count_tokens),
5002 )
5003 .route(routes::V1_COMPLETIONS, post(openai_extra::completions))
5004 .route(routes::COMPLETION, post(completion::completion))
5009 .route(routes::COMPLETIONS, post(completion::completion))
5010 .route(routes::V1_TOKENIZE, post(openai_extra::tokenize))
5011 .route(routes::V1_DETOKENIZE, post(openai_extra::detokenize))
5012 .route(routes::TOKENIZE, post(openai_extra::tokenize))
5018 .route(routes::DETOKENIZE, post(openai_extra::detokenize))
5019 .route(routes::V1_EMBEDDINGS, post(embeddings::embeddings))
5020 .route(routes::V1_RERANK, post(rerank::rerank))
5024 .route(routes::RERANK, post(rerank::rerank))
5025 .route(routes::CACHE_STATS, get(cache_stats))
5026 .route(routes::METRICS, get(metrics))
5027 .route(routes::ADMIN_MODELS, get(admin::models))
5032 .route(routes::ADMIN_MODELS_LOAD, post(admin::load_model))
5033 .route(routes::ADMIN_MODELS_UNLOAD, post(admin::unload_model))
5034 .route(routes::ADMIN_DOWNLOAD, post(admin::download))
5035 .route(routes::ADMIN_TASKS, get(admin::tasks))
5036 .route(&admin::cancel_route(), post(admin::cancel_task))
5037 .route(routes::ADMIN_STATS, get(admin::stats))
5038 .merge(conversations::router());
5042
5043 if let Ok(key) = std::env::var("FERROX_API_KEY") {
5048 tracing::info!("API key auth enabled");
5049 let auth = limits::AuthConfig {
5050 api_key: Arc::new(key),
5051 };
5052 protected = protected.route_layer(axum::middleware::from_fn_with_state(
5053 auth,
5054 limits::require_api_key,
5055 ));
5056 }
5057 if let Ok(rpm) = std::env::var("FERROX_RATE_LIMIT_PER_MINUTE") {
5058 let rpm: u32 = rpm
5059 .parse()
5060 .expect("FERROX_RATE_LIMIT_PER_MINUTE must be a positive integer");
5061 tracing::info!("rate limiting enabled: {rpm} requests/minute (global)");
5062 let limiter = Arc::new(limits::RateLimiter::per_minute(rpm));
5063 protected = protected.route_layer(axum::middleware::from_fn_with_state(
5064 limiter,
5065 limits::rate_limit,
5066 ));
5067 }
5068 if let Ok(spec) = std::env::var("FERROX_CORS_ORIGINS") {
5076 let origins = security::parse_cors_origins(&spec)
5077 .unwrap_or_else(|e| panic!("FERROX_CORS_ORIGINS: {e}"));
5078 tracing::info!(
5079 "CORS enabled: {} allow-listed origin(s) ({})",
5080 origins.len(),
5081 spec
5082 );
5083 let cors = tower_http::cors::CorsLayer::new()
5084 .allow_origin(tower_http::cors::AllowOrigin::list(origins))
5085 .allow_methods([axum::http::Method::GET, axum::http::Method::POST])
5086 .allow_headers([
5087 axum::http::header::CONTENT_TYPE,
5088 axum::http::header::AUTHORIZATION,
5089 axum::http::HeaderName::from_static(attribution::CLIENT_HEADER),
5095 axum::http::HeaderName::from_static("last-event-id"),
5099 ]);
5100 protected = protected.route_layer(cors);
5101 }
5102
5103 let app = public
5107 .merge(protected)
5108 .layer(axum::middleware::from_fn(limits::retry_after))
5109 .with_state(state);
5110
5111 let tls_paths = security::tls_paths_from_env().unwrap_or_else(|e| panic!("{e}"));
5117 install_ring_crypto_provider();
5118 match tls_paths {
5123 Some(paths) => {
5124 let config =
5125 axum_server::tls_rustls::RustlsConfig::from_pem_file(&paths.cert, &paths.key)
5126 .await
5127 .map_err(|e| {
5128 anyhow::anyhow!(
5129 "failed to load TLS cert/key ({:?}, {:?}): {e}",
5130 paths.cert,
5131 paths.key
5132 )
5133 })?;
5134 let socket_addr: std::net::SocketAddr = addr
5135 .parse()
5136 .map_err(|e| anyhow::anyhow!("invalid FERROX_ADDR {addr:?} for TLS: {e}"))?;
5137 let listener = std::net::TcpListener::bind(socket_addr)?;
5138 listener.set_nonblocking(true)?;
5146 let bound = listener.local_addr()?;
5147 tracing::info!("TLS enabled: ferrox-server listening on https://{bound}");
5148 announce_ready(bound, "https");
5149
5150 let handle = axum_server::Handle::new();
5151 let shutdown_handle = handle.clone();
5152 tokio::spawn(async move {
5153 shutdown_signal(exit_on_stdin_close).await;
5154 shutdown_handle.graceful_shutdown(Some(Duration::from_secs(5)));
5155 });
5156 axum_server::from_tcp_rustls(listener, config)?
5157 .handle(handle)
5158 .serve(app.into_make_service())
5159 .await?;
5160 }
5161 None => {
5162 let listener = tokio::net::TcpListener::bind(&addr).await?;
5163 let bound = listener.local_addr()?;
5164 tracing::info!("ferrox-server listening on {bound}");
5165 announce_ready(bound, "http");
5166 axum::serve(listener, app)
5167 .with_graceful_shutdown(shutdown_signal(exit_on_stdin_close))
5168 .await?;
5169 }
5170 }
5171 Ok(())
5172}
5173
5174#[cfg(test)]
5175mod tests {
5176 use super::*;
5177 use ferrox_models::config::test_dense_fixture;
5178
5179 #[test]
5180 fn parses_llama_server_style_options() {
5181 let argv = [
5182 "ferrox-server",
5183 "-m",
5184 "model.gguf",
5185 "--host",
5186 "::1",
5187 "--port",
5188 "9000",
5189 "-t",
5190 "4",
5191 "-dev",
5192 "Metal",
5193 "-ngl",
5194 "all",
5195 ]
5196 .into_iter()
5197 .map(String::from)
5198 .collect();
5199 let args = ServerArgs::try_parse_from(rewrite_llama_style_argv(argv)).unwrap();
5200
5201 assert_eq!(args.model.as_deref(), Some("model.gguf"));
5202 assert_eq!(args.host, Some(IpAddr::V6(std::net::Ipv6Addr::LOCALHOST)));
5203 assert_eq!(args.port, Some(9000));
5204 assert_eq!(args.threads, Some(4));
5205 assert_eq!(args.device, Some(OffloadDevice::Metal));
5206 assert_eq!(args.n_gpu_layers, Some(GpuLayers::All));
5207 assert_eq!(
5208 cli_bind_addr(&args, Some("127.0.0.1:8383")).as_deref(),
5209 Some("[::1]:9000")
5210 );
5211 }
5212
5213 #[test]
5214 fn port_zero_survives_argument_parsing_as_a_real_request() {
5215 let argv = ["ferrox-server", "--port", "0"]
5220 .into_iter()
5221 .map(String::from)
5222 .collect();
5223 let args = ServerArgs::try_parse_from(rewrite_llama_style_argv(argv)).unwrap();
5224 assert_eq!(args.port, Some(0));
5225 assert_eq!(
5226 cli_bind_addr(&args, Some("127.0.0.1:8383")).as_deref(),
5227 Some("127.0.0.1:0")
5228 );
5229 }
5230
5231 #[test]
5232 fn parallel_flag_parses_and_rewrites_np() {
5233 let argv = ["ferrox-server", "-np", "4"]
5234 .into_iter()
5235 .map(String::from)
5236 .collect();
5237 let args = ServerArgs::try_parse_from(rewrite_llama_style_argv(argv)).unwrap();
5238 assert_eq!(args.parallel, Some(4));
5239 }
5240
5241 #[test]
5242 fn stdin_close_exit_is_opt_in() {
5243 let args =
5246 ServerArgs::try_parse_from(["ferrox-server"].into_iter().map(String::from)).unwrap();
5247 assert!(!args.exit_on_stdin_close);
5248 let args = ServerArgs::try_parse_from(
5249 ["ferrox-server", "--exit-on-stdin-close"]
5250 .into_iter()
5251 .map(String::from),
5252 )
5253 .unwrap();
5254 assert!(args.exit_on_stdin_close);
5255 }
5256
5257 #[test]
5258 fn the_ready_line_round_trips_through_a_parent_reading_stdout() {
5259 let addr: SocketAddr = "127.0.0.1:51999".parse().unwrap();
5260 let ready = ferrox_api::ServerReady::new(addr, "http", "0.5.0", std::process::id());
5261 let parsed = ferrox_api::ServerReady::from_line(&ready.to_line()).unwrap();
5262 assert_eq!(parsed.port, 51999);
5263 assert_eq!(parsed.base_url(), "http://127.0.0.1:51999");
5264 assert!(ferrox_api::ServerReady::from_line("INFO ferrox-server listening").is_none());
5266 }
5267
5268 fn test_model() -> Model {
5269 let cfg = test_dense_fixture();
5273 Model::Gguf(GgufModel {
5274 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 32)),
5275 tokenizer: Arc::new(ServerTokenizer::Byte),
5276 stop_tokens: StopTokens::default(),
5277 bos_id: None,
5278 is_synthetic: true,
5279 chat_template: chat_template::PromptTemplate::plain(),
5280 })
5281 }
5282
5283 fn greedy_params(max_tokens: usize) -> GenerationParams {
5284 GenerationParams {
5285 reasoning: None,
5286 max_tokens,
5287 sampling: SamplingParams::default(),
5288 seed: 1,
5289 stop: Vec::new(),
5290 stop_token_ids: Vec::new(),
5291 json_object: false,
5292 grammar: None,
5293 cancel: None,
5294 ignore_eos: false,
5295 }
5296 }
5297
5298 fn test_model_full_byte_vocab() -> Model {
5302 test_model_full_byte_vocab_with_eos(None)
5303 }
5304
5305 fn test_model_full_byte_vocab_with_eos(eos: Option<usize>) -> Model {
5312 let mut cfg = test_dense_fixture();
5313 cfg.vocab_size = 256;
5314 Model::Gguf(GgufModel {
5315 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5316 tokenizer: Arc::new(ServerTokenizer::Byte),
5317 stop_tokens: StopTokens::from_eos(eos),
5318 bos_id: None,
5319 is_synthetic: true,
5320 chat_template: chat_template::PromptTemplate::plain(),
5321 })
5322 }
5323
5324 fn test_state(model: Model, response_cache: ResponseCache) -> AppState {
5328 AppState {
5329 embedding: None,
5330 paged_kv: None,
5331 active: std::sync::RwLock::new(Some(Arc::new(ActiveModel {
5332 id: None,
5333 loaded: Loaded::Generative(Arc::new(model)),
5334 batcher: None,
5335 ceiling: None,
5336 }))),
5337 load_in_progress: std::sync::atomic::AtomicBool::new(false),
5338 tasks: Arc::new(tasks::TaskRegistry::new()),
5339 cancels: Arc::new(cancel::CancelRegistry::new()),
5340 stats: stats::Stats::new(),
5341 streams: resume::StreamRegistry::new(),
5342 model_dir: None,
5343 response_cache: Mutex::new(response_cache),
5344 kv_pool: None,
5345 prefix_cache: None,
5346 sessions: session::SessionStore::new(),
5347 requests_total: std::sync::atomic::AtomicU64::new(0),
5348 request_errors_total: std::sync::atomic::AtomicU64::new(0),
5349 started_at: std::time::Instant::now(),
5350 last_request_ms: std::sync::atomic::AtomicU64::new(0),
5351 detection: Arc::new(health::Detection::ready(health::probe_backends())),
5352 mcp: None,
5353 continuous_batching_enabled: false,
5354 metal_private_decode_gate: None,
5355 loading_model: Mutex::new(None),
5356 last_load_error: Mutex::new(None),
5357 serving: Mutex::new(crate::stats::ServingStats::default()),
5358 maintenance: Mutex::new(crate::policy::maintenance::MaintenanceGate::serving()),
5359 footprint: Mutex::new(crate::policy::footprint::ProbeCache::new(FOOTPRINT_TTL_MS)),
5360 started_unix: unix_now(),
5361 }
5362 }
5363
5364 fn test_app() -> Router {
5373 test_app_with_state(Arc::new(test_state(
5374 test_model_full_byte_vocab(),
5375 ResponseCache::new(1000, Duration::from_secs(3600)),
5376 )))
5377 }
5378
5379 fn test_app_with_state(state: Arc<AppState>) -> Router {
5382 Router::new()
5383 .route(ferrox_api::routes::HEALTH, get(health))
5384 .route(ferrox_api::routes::V1_MODELS, get(list_models))
5385 .route(ferrox_api::routes::V1_RESPONSES, post(responses::responses))
5386 .route(
5387 &axum_path(ferrox_api::routes::V1_RESPONSE),
5388 get(responses::responses_get),
5389 )
5390 .route(
5391 &axum_path(ferrox_api::routes::V1_RESPONSE_CANCEL),
5392 post(responses::responses_cancel),
5393 )
5394 .route(ferrox_api::routes::V1_STATS, get(serving_stats))
5395 .route(ferrox_api::routes::V1_REQUESTS, get(recent_requests))
5396 .route(
5397 ferrox_api::routes::V1_CACHE_STATUS,
5398 get(cache_admin::cache_status),
5399 )
5400 .route(
5401 ferrox_api::routes::V1_CACHE_REBUILD,
5402 post(cache_admin::cache_rebuild),
5403 )
5404 .route(
5405 ferrox_api::routes::ADMIN_PREPARE_STOP,
5406 post(cache_admin::prepare_stop),
5407 )
5408 .route("/v1/chat/completions", post(chat_completions))
5409 .route(ferrox_api::routes::V1_MESSAGES, post(anthropic::messages))
5410 .route(
5411 ferrox_api::routes::V1_MESSAGES_COUNT_TOKENS,
5412 post(anthropic::count_tokens),
5413 )
5414 .route("/v1/tokenize", post(openai_extra::tokenize))
5415 .route("/v1/detokenize", post(openai_extra::detokenize))
5416 .route(ferrox_api::routes::TOKENIZE, post(openai_extra::tokenize))
5420 .route(
5421 ferrox_api::routes::DETOKENIZE,
5422 post(openai_extra::detokenize),
5423 )
5424 .route("/v1/embeddings", post(embeddings::embeddings))
5425 .route("/v1/completions", post(openai_extra::completions))
5426 .route(ferrox_api::routes::COMPLETION, post(completion::completion))
5428 .route(
5429 ferrox_api::routes::COMPLETIONS,
5430 post(completion::completion),
5431 )
5432 .route(
5433 ferrox_api::routes::ADMIN_MODELS_UNLOAD,
5434 post(admin::unload_model),
5435 )
5436 .route(ferrox_api::routes::ADMIN_TASKS, get(admin::tasks))
5437 .route(ferrox_api::routes::ADMIN_STATS, get(admin::stats))
5438 .route(ferrox_api::routes::V1_CANCEL, post(cancel_generation))
5439 .route(
5440 &axum_path(ferrox_api::routes::V1_STREAM),
5441 get(resume::resume),
5442 )
5443 .route(
5444 &axum_path(ferrox_api::routes::V1_STREAM_POLL),
5445 get(resume::poll),
5446 )
5447 .with_state(state)
5448 }
5449
5450 fn named_test_model(name: &'static str, vocab_size: usize) -> Model {
5451 let mut cfg = test_dense_fixture();
5452 cfg.name = name;
5453 cfg.vocab_size = vocab_size;
5454 Model::Gguf(GgufModel {
5455 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5456 tokenizer: Arc::new(ServerTokenizer::Byte),
5457 stop_tokens: StopTokens::default(),
5458 bos_id: None,
5459 is_synthetic: true,
5460 chat_template: chat_template::PromptTemplate::plain(),
5461 })
5462 }
5463
5464 fn model_with_template(name: &'static str, source: &str) -> Model {
5468 let mut cfg = test_dense_fixture();
5469 cfg.name = name;
5470 cfg.vocab_size = 256;
5471 Model::Gguf(GgufModel {
5472 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5473 tokenizer: Arc::new(ServerTokenizer::Byte),
5474 stop_tokens: StopTokens::default(),
5475 bos_id: None,
5476 is_synthetic: true,
5477 chat_template: chat_template::PromptTemplate::from_gguf_metadata(
5478 Some(source),
5479 Some("qwen3"),
5480 false,
5481 true,
5482 None,
5483 None,
5484 ),
5485 })
5486 }
5487
5488 #[tokio::test]
5497 async fn a_template_that_rejects_the_conversation_is_a_400_on_the_streaming_path() {
5498 let strict = "{% if messages | length > 1 %}\
5501 {{ raise_exception('this template takes one turn') }}\
5502 {% endif %}{{ messages[0].content }}";
5503 let state = Arc::new(test_state(
5504 model_with_template("strict", strict),
5505 ResponseCache::new(4, Duration::from_secs(60)),
5506 ));
5507 let app = test_app_with_state(state);
5508
5509 let (status, body) = post_json_uri(
5510 &app,
5511 "/v1/chat/completions",
5512 serde_json::json!({
5513 "model": "strict",
5514 "stream": true,
5515 "messages": [
5516 {"role": "user", "content": "one"},
5517 {"role": "user", "content": "two"},
5518 ],
5519 }),
5520 )
5521 .await;
5522 assert_eq!(status, StatusCode::BAD_REQUEST);
5523 assert_eq!(body["error"]["param"], serde_json::json!("messages"));
5524 assert!(
5525 body["error"]["message"]
5526 .as_str()
5527 .unwrap()
5528 .contains("one turn"),
5529 "the template's own message must reach the caller: {body}"
5530 );
5531
5532 let (status, _) = post_json_uri(
5534 &app,
5535 "/v1/chat/completions",
5536 serde_json::json!({
5537 "model": "strict",
5538 "stream": true,
5539 "max_tokens": 1,
5540 "messages": [{"role": "user", "content": "one"}],
5541 }),
5542 )
5543 .await;
5544 assert_eq!(status, StatusCode::OK);
5545 }
5546
5547 #[tokio::test]
5549 async fn models_advertises_the_gears_this_checkpoint_actually_has() {
5550 let reasoning = "{% if enable_thinking %}<think>{% endif %}\
5551 {% if reasoning_effort %}\
5552 {% if reasoning_effort not in ['low','medium','high'] %}\
5553 {{ raise_exception('bad effort') }}\
5554 {% endif %}[{{ reasoning_effort }}]\
5555 {% endif %}{{ messages[0].content }}";
5556 let state = Arc::new(test_state(
5557 model_with_template("thinker", reasoning),
5558 ResponseCache::new(4, Duration::from_secs(60)),
5559 ));
5560 let app = test_app_with_state(state);
5561 let (status, models) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
5562 assert_eq!(status, StatusCode::OK);
5563 let entry = &models["data"][0];
5564 assert_eq!(
5565 entry["supported_reasoning_efforts"],
5566 serde_json::json!(["off", "low", "medium", "high"])
5567 );
5568 assert_eq!(entry["default_reasoning_effort"], serde_json::json!("off"));
5569 }
5570
5571 #[tokio::test]
5576 async fn a_checkpoint_with_no_thinking_controls_advertises_neither_field() {
5577 let app = test_app();
5578 let (_, models) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
5579 let entry = &models["data"][0];
5580 assert!(entry.get("supported_reasoning_efforts").is_none());
5581 assert!(entry.get("default_reasoning_effort").is_none());
5582 }
5583
5584 fn active_model(state: &AppState, name: &'static str) -> Arc<ActiveModel> {
5585 Arc::new(ActiveModel {
5586 id: Some(name.to_string()),
5587 loaded: Loaded::Generative(Arc::new(named_test_model(name, 256))),
5588 batcher: None,
5589 ceiling: None,
5590 })
5591 .tap_into(state)
5592 }
5593
5594 trait TapInto {
5596 fn tap_into(self, state: &AppState) -> Self;
5597 }
5598 impl TapInto for Arc<ActiveModel> {
5599 fn tap_into(self, state: &AppState) -> Self {
5600 state.swap_active(Some(Arc::clone(&self)));
5601 self
5602 }
5603 }
5604
5605 #[test]
5611 fn an_in_flight_request_keeps_the_model_it_started_on() {
5612 let state = test_state(
5613 named_test_model("model-a", 256),
5614 ResponseCache::new(4, Duration::from_secs(60)),
5615 );
5616
5617 let in_flight = state.active().expect("a model is loaded");
5620 assert_eq!(in_flight.name(), "model-a");
5621
5622 active_model(&state, "model-b");
5623
5624 assert_eq!(state.active().unwrap().name(), "model-b");
5626 assert_eq!(in_flight.name(), "model-a");
5628 let (_chunks, finish, _usage) = run_generation(
5629 in_flight.generative().unwrap(),
5630 "hi",
5631 &greedy_params(3),
5632 None,
5633 None,
5634 None,
5635 None,
5636 None,
5637 None,
5638 )
5639 .expect("the old model must still decode after being swapped out");
5640 assert!(matches!(finish, FinishReason::Length | FinishReason::Stop));
5641 }
5642
5643 #[test]
5648 fn a_swapped_out_model_lives_until_its_last_holder_releases_it() {
5649 let state = test_state(
5650 named_test_model("model-a", 256),
5651 ResponseCache::new(4, Duration::from_secs(60)),
5652 );
5653 let in_flight = state.active().expect("a model is loaded");
5654 let weights = Arc::clone(in_flight.generative().unwrap());
5655 assert!(Arc::strong_count(&weights) >= 2);
5656
5657 let previous = state.swap_active(Some(Arc::new(ActiveModel {
5658 id: Some("model-b".to_string()),
5659 loaded: Loaded::Generative(Arc::new(named_test_model("model-b", 256))),
5660 batcher: None,
5661 ceiling: None,
5662 })));
5663 drop(previous);
5664 assert!(Arc::strong_count(&weights) >= 2);
5666 drop(in_flight);
5667 assert_eq!(Arc::strong_count(&weights), 1);
5668 }
5669
5670 #[tokio::test]
5674 async fn unloading_answers_503_instead_of_serving_the_dropped_model() {
5675 let state = Arc::new(test_state(
5676 named_test_model("model-a", 256),
5677 ResponseCache::new(4, Duration::from_secs(60)),
5678 ));
5679 let app = test_app_with_state(Arc::clone(&state));
5680
5681 let (status, body) = post_json_uri(
5682 &app,
5683 ferrox_api::routes::ADMIN_MODELS_UNLOAD,
5684 serde_json::json!({}),
5685 )
5686 .await;
5687 assert_eq!(status, StatusCode::OK);
5688 assert_eq!(body["ok"], true);
5689 assert!(body["active"].is_null());
5690 assert!(state.active().is_none());
5691
5692 let (status, _) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
5693 assert_eq!(status, StatusCode::OK);
5694 let (_, models) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
5695 assert_eq!(models["data"].as_array().unwrap().len(), 0);
5696
5697 let (status, body) = post_json_uri(
5698 &app,
5699 "/v1/chat/completions",
5700 serde_json::json!({
5701 "model": "x",
5702 "messages": [{"role": "user", "content": "hi"}]
5703 }),
5704 )
5705 .await;
5706 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5707 assert_eq!(body["error"]["type"], "model_not_loaded");
5708 }
5709
5710 #[tokio::test]
5714 async fn health_reports_the_unloaded_state_rather_than_going_silent() {
5715 let state = Arc::new(test_state(
5716 named_test_model("model-a", 256),
5717 ResponseCache::new(4, Duration::from_secs(60)),
5718 ));
5719 let app = test_app_with_state(Arc::clone(&state));
5720 state.swap_active(None);
5721
5722 let (status, body) = get_json(&app, ferrox_api::routes::HEALTH).await;
5723 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5726 assert_eq!(body["state"], "unavailable");
5727 assert_eq!(body["reason"], "model_not_loaded");
5728 assert!(body["model"].is_null());
5729 let real_weights = body["capabilities"]
5730 .as_array()
5731 .unwrap()
5732 .iter()
5733 .find(|c| c["id"] == "real_weights")
5734 .cloned()
5735 .expect("real_weights is always reported");
5736 assert_eq!(real_weights["available"], false);
5737 assert_eq!(real_weights["reason"], "model_not_loaded");
5738 }
5739
5740 #[tokio::test]
5744 async fn a_finished_request_lands_in_the_stats_ring_with_both_durations() {
5745 let app = test_app();
5746
5747 let (status, completion) = post_json_uri(
5748 &app,
5749 "/v1/chat/completions",
5750 serde_json::json!({
5751 "model": "x",
5752 "messages": [{"role": "user", "content": "hi"}],
5753 "max_tokens": 4
5754 }),
5755 )
5756 .await;
5757 assert_eq!(status, StatusCode::OK);
5758 let request_id = completion["request_id"].as_str().unwrap().to_string();
5759
5760 let (status, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5761 assert_eq!(status, StatusCode::OK);
5762 let recent = stats["recent"].as_array().unwrap();
5763 assert_eq!(recent.len(), 1);
5764 let row = &recent[0];
5765 assert_eq!(row["request_id"], request_id);
5766 assert_eq!(row["route"], ferrox_api::routes::V1_CHAT_COMPLETIONS);
5767 assert_eq!(row["status"], 200);
5768 assert_eq!(row["stream"], false);
5769 assert!(row["duration_ms"].is_number());
5772 assert!(row["decode_ms"].is_number());
5773 assert!(stats["tokens_generated_total"].as_u64().unwrap() > 0);
5774 assert_eq!(
5775 stats["tokens_prompt_total"].as_u64().unwrap(),
5776 row["prompt_tokens"].as_u64().unwrap()
5777 );
5778 }
5779
5780 #[tokio::test]
5784 async fn a_rejected_request_is_recorded_too() {
5785 let state = Arc::new(test_state(
5786 named_test_model("model-a", 256),
5787 ResponseCache::new(4, Duration::from_secs(60)),
5788 ));
5789 let app = test_app_with_state(Arc::clone(&state));
5790 state.swap_active(None);
5791
5792 let (status, _) = post_json_uri(
5793 &app,
5794 "/v1/chat/completions",
5795 serde_json::json!({"model": "x", "messages": [{"role": "user", "content": "hi"}]}),
5796 )
5797 .await;
5798 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5799
5800 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5801 let recent = stats["recent"].as_array().unwrap();
5802 assert_eq!(recent.len(), 1);
5803 assert_eq!(recent[0]["status"], 503);
5804 assert_eq!(recent[0]["completion_tokens"], 0);
5805 assert!(recent[0]["decode_ms"].is_null());
5806 assert_eq!(stats["errors_total"], 1);
5807 }
5808
5809 async fn post_json_with_headers(
5813 app: &Router,
5814 uri: &str,
5815 body: serde_json::Value,
5816 headers: &[(&str, &str)],
5817 ) -> (StatusCode, serde_json::Value) {
5818 use http_body_util::BodyExt;
5819 use tower::ServiceExt;
5820
5821 let mut builder = axum::http::Request::builder()
5822 .method("POST")
5823 .uri(uri)
5824 .header("content-type", "application/json");
5825 for (name, value) in headers {
5826 builder = builder.header(*name, *value);
5827 }
5828 let response = app
5829 .clone()
5830 .oneshot(
5831 builder
5832 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
5833 .unwrap(),
5834 )
5835 .await
5836 .unwrap();
5837 let status = response.status();
5838 let bytes = response.into_body().collect().await.unwrap().to_bytes();
5839 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
5840 (status, json)
5841 }
5842
5843 #[tokio::test]
5847 async fn tokenize_detokenize_and_embeddings_all_land_in_the_ring() {
5848 let app = test_app();
5849
5850 let (status, _) = post_json_uri(
5851 &app,
5852 ferrox_api::routes::V1_TOKENIZE,
5853 serde_json::json!({"prompt": "hello"}),
5854 )
5855 .await;
5856 assert_eq!(status, StatusCode::OK);
5857 let (status, _) = post_json_uri(
5858 &app,
5859 ferrox_api::routes::V1_DETOKENIZE,
5860 serde_json::json!({"tokens": [104, 105]}),
5861 )
5862 .await;
5863 assert_eq!(status, StatusCode::OK);
5864 let (status, _) = post_json_uri(
5865 &app,
5866 ferrox_api::routes::V1_EMBEDDINGS,
5867 serde_json::json!({"input": "hello"}),
5868 )
5869 .await;
5870 assert_eq!(status, StatusCode::OK);
5871
5872 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5873 let routes: Vec<&str> = stats["recent"]
5874 .as_array()
5875 .unwrap()
5876 .iter()
5877 .map(|row| row["route"].as_str().unwrap())
5878 .collect();
5879 for expected in [
5880 ferrox_api::routes::V1_TOKENIZE,
5881 ferrox_api::routes::V1_DETOKENIZE,
5882 ferrox_api::routes::V1_EMBEDDINGS,
5883 ] {
5884 assert!(
5885 routes.contains(&expected),
5886 "{expected} is missing: {routes:?}"
5887 );
5888 }
5889
5890 let row = |route: &str| {
5891 stats["recent"]
5892 .as_array()
5893 .unwrap()
5894 .iter()
5895 .find(|r| r["route"] == route)
5896 .cloned()
5897 .unwrap()
5898 };
5899 let embed = row(ferrox_api::routes::V1_EMBEDDINGS);
5903 assert!(embed["prompt_tokens"].as_u64().unwrap() > 0);
5904 assert!(embed["decode_ms"].is_null());
5905 assert_eq!(embed["completion_tokens"], 0);
5906 assert_eq!(row(ferrox_api::routes::V1_TOKENIZE)["prompt_tokens"], 0);
5910 assert_eq!(
5911 stats["tokens_prompt_total"].as_u64().unwrap(),
5912 embed["prompt_tokens"].as_u64().unwrap(),
5913 "only the forward pass counted"
5914 );
5915 }
5916
5917 fn streaming_test_app() -> Router {
5922 let mut cfg = test_dense_fixture();
5923 cfg.vocab_size = 256;
5924 let model = Model::Gguf(GgufModel {
5925 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5926 tokenizer: Arc::new(ServerTokenizer::Byte),
5927 stop_tokens: StopTokens::default(),
5928 bos_id: None,
5929 is_synthetic: false,
5930 chat_template: chat_template::PromptTemplate::plain(),
5931 });
5932 test_app_with_state(Arc::new(test_state(
5933 model,
5934 ResponseCache::new(1000, Duration::from_secs(3600)),
5935 )))
5936 }
5937
5938 #[tokio::test]
5942 async fn the_native_completion_wire_is_not_the_openai_one() {
5943 let app = test_app();
5944
5945 let (status, native) = post_json_uri(
5946 &app,
5947 ferrox_api::routes::COMPLETION,
5948 serde_json::json!({"prompt": "hi", "n_predict": 4}),
5949 )
5950 .await;
5951 assert_eq!(status, StatusCode::OK, "{native}");
5952 assert!(native["content"].is_string(), "{native}");
5953 assert_eq!(native["stop"], true);
5954 assert_eq!(native["stop_type"], "limit");
5955 assert_eq!(native["stopping_word"], "");
5956 assert_eq!(native["truncated"], false);
5957 assert_eq!(native["id_slot"], -1);
5958 assert!(native["timings"]["prompt_n"].is_number(), "{native}");
5959 assert!(native["generation_settings"]["n_predict"] == 4, "{native}");
5960 assert!(
5961 native.get("choices").is_none(),
5962 "the native shape has no `choices`: {native}"
5963 );
5964
5965 let (status, openai) = post_json_uri(
5966 &app,
5967 ferrox_api::routes::V1_COMPLETIONS,
5968 serde_json::json!({"prompt": "hi", "max_tokens": 4}),
5969 )
5970 .await;
5971 assert_eq!(status, StatusCode::OK);
5972 assert!(openai["choices"][0]["text"].is_string(), "{openai}");
5973 assert!(
5974 openai.get("content").is_none(),
5975 "the OpenAI shape has no top-level `content`: {openai}"
5976 );
5977 }
5978
5979 #[tokio::test]
5983 async fn both_native_spellings_reach_the_same_handler() {
5984 let app = test_app();
5985 for route in [
5986 ferrox_api::routes::COMPLETION,
5987 ferrox_api::routes::COMPLETIONS,
5988 ] {
5989 let (status, body) = post_json_uri(
5990 &app,
5991 route,
5992 serde_json::json!({"prompt": "hi", "n_predict": 2, "seed": 1}),
5993 )
5994 .await;
5995 assert_eq!(status, StatusCode::OK, "{route}: {body}");
5996 assert_eq!(body["stop"], true, "{route}");
5997 assert!(body["content"].is_string(), "{route}");
5998 }
5999
6000 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6003 let routes: Vec<&str> = stats["recent"]
6004 .as_array()
6005 .unwrap()
6006 .iter()
6007 .map(|row| row["route"].as_str().unwrap())
6008 .collect();
6009 assert!(
6010 routes.contains(&ferrox_api::routes::COMPLETION),
6011 "{routes:?}"
6012 );
6013 assert!(
6014 routes.contains(&ferrox_api::routes::COMPLETIONS),
6015 "{routes:?}"
6016 );
6017 }
6018
6019 #[tokio::test]
6025 async fn a_native_stream_ends_on_a_stop_frame_with_no_done_sentinel() {
6026 let app = streaming_test_app();
6027 let raw = post_sse_raw_uri(
6028 &app,
6029 ferrox_api::routes::COMPLETION,
6030 serde_json::json!({"prompt": "hi", "n_predict": 6, "stream": true, "seed": 7}),
6031 )
6032 .await;
6033
6034 assert!(
6035 !raw.contains("[DONE]"),
6036 "llama.cpp's native stream has no sentinel: {raw}"
6037 );
6038 let frames: Vec<serde_json::Value> = raw
6039 .lines()
6040 .filter_map(|line| line.strip_prefix("data: "))
6041 .map(|json| serde_json::from_str(json).expect("every frame is one JSON object"))
6042 .collect();
6043 assert!(frames.len() >= 2, "expected partials then a final: {raw}");
6044
6045 let (last, partials) = frames.split_last().unwrap();
6046 assert_eq!(last["stop"], true, "the last frame closes the stream");
6047 assert!(last["timings"].is_object(), "{last}");
6048 assert!(last["stop_type"].is_string(), "{last}");
6049 for partial in partials {
6050 assert_eq!(partial["stop"], false, "{partial}");
6051 assert!(partial["content"].is_string(), "{partial}");
6052 assert!(partial.get("timings").is_none(), "{partial}");
6056 assert!(partial.get("generation_settings").is_none(), "{partial}");
6057 }
6058 let streamed: String = partials
6061 .iter()
6062 .filter_map(|p| p["content"].as_str())
6063 .collect();
6064 assert_eq!(last["content"].as_str().unwrap(), streamed);
6065 }
6066
6067 #[tokio::test]
6072 async fn an_unbounded_n_predict_is_refused_rather_than_quietly_shrunk() {
6073 let app = test_app();
6074 for body in [
6075 serde_json::json!({"prompt": "hi"}),
6076 serde_json::json!({"prompt": "hi", "n_predict": -1}),
6077 ] {
6078 let (status, refusal) =
6079 post_json_uri(&app, ferrox_api::routes::COMPLETION, body.clone()).await;
6080 assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{body}: {refusal}");
6081 assert!(
6082 refusal["error"]["message"]
6083 .as_str()
6084 .unwrap()
6085 .contains("n_predict"),
6086 "{refusal}"
6087 );
6088 }
6089 let (status, _) = post_json_uri(
6092 &app,
6093 ferrox_api::routes::COMPLETION,
6094 serde_json::json!({"prompt": "hi", "n_predict": 2}),
6095 )
6096 .await;
6097 assert_eq!(status, StatusCode::OK);
6098 }
6099
6100 #[tokio::test]
6109 async fn a_stop_string_halts_the_answer_and_is_named_back() {
6110 let app = streaming_test_app();
6111 let ask = |stop: serde_json::Value| {
6112 let app = app.clone();
6113 async move {
6114 post_json_uri(
6115 &app,
6116 ferrox_api::routes::COMPLETION,
6117 serde_json::json!({
6118 "prompt": "hi",
6119 "n_predict": 64,
6120 "ignore_eos": true,
6121 "stop": stop,
6122 }),
6123 )
6124 .await
6125 .1
6126 }
6127 };
6128
6129 let baseline = ask(serde_json::json!([])).await;
6130 assert_eq!(baseline["stop_type"], "limit");
6131 assert_eq!(baseline["stopping_word"], "");
6132 let text = baseline["content"].as_str().unwrap().to_string();
6133 let sentinel: String = text.chars().skip(1).take(2).collect();
6139 assert_eq!(
6140 sentinel.chars().count(),
6141 2,
6142 "the fixture must produce enough output to cut: {text:?}"
6143 );
6144 let cut = text.find(&sentinel).expect("it came out of this text");
6145
6146 let stopped = ask(serde_json::json!([sentinel])).await;
6147 assert_eq!(stopped["stop_type"], "word", "{stopped}");
6148 assert_eq!(stopped["stopping_word"], sentinel);
6149 assert_eq!(
6150 stopped["content"].as_str().unwrap(),
6151 &text[..cut],
6152 "the answer must be cut at the sentinel, not run past it"
6153 );
6154 }
6155
6156 #[tokio::test]
6162 async fn the_llama_cpp_spelling_of_tokenize_reaches_the_same_handler() {
6163 let app = test_app();
6164
6165 let (v1_status, v1) = post_json_uri(
6166 &app,
6167 ferrox_api::routes::V1_TOKENIZE,
6168 serde_json::json!({"prompt": "hello"}),
6169 )
6170 .await;
6171 let (alias_status, alias) = post_json_uri(
6172 &app,
6173 ferrox_api::routes::TOKENIZE,
6174 serde_json::json!({"content": "hello"}),
6175 )
6176 .await;
6177 assert_eq!(v1_status, StatusCode::OK);
6178 assert_eq!(alias_status, StatusCode::OK, "{alias}");
6179 assert_eq!(v1["tokens"], alias["tokens"]);
6180 assert!(!alias["tokens"].as_array().unwrap().is_empty());
6181
6182 let (status, both_ways) = post_json_uri(
6185 &app,
6186 ferrox_api::routes::TOKENIZE,
6187 serde_json::json!({"prompt": "hello"}),
6188 )
6189 .await;
6190 assert_eq!(status, StatusCode::OK);
6191 assert_eq!(both_ways["tokens"], v1["tokens"]);
6192 }
6193
6194 #[tokio::test]
6199 async fn detokenize_answers_under_both_dialects_keys() {
6200 let app = test_app();
6201 for route in [
6202 ferrox_api::routes::DETOKENIZE,
6203 ferrox_api::routes::V1_DETOKENIZE,
6204 ] {
6205 let (status, body) =
6206 post_json_uri(&app, route, serde_json::json!({"tokens": [104, 105]})).await;
6207 assert_eq!(status, StatusCode::OK, "{route}");
6208 assert_eq!(body["text"], "hi", "{route}");
6209 assert_eq!(body["content"], body["text"], "{route}");
6210 }
6211 }
6212
6213 #[tokio::test]
6217 async fn the_alias_is_recorded_under_the_path_the_client_called() {
6218 let app = test_app();
6219 let (status, _) = post_json_uri(
6220 &app,
6221 ferrox_api::routes::TOKENIZE,
6222 serde_json::json!({"content": "hello"}),
6223 )
6224 .await;
6225 assert_eq!(status, StatusCode::OK);
6226
6227 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6228 let routes: Vec<&str> = stats["recent"]
6229 .as_array()
6230 .unwrap()
6231 .iter()
6232 .map(|row| row["route"].as_str().unwrap())
6233 .collect();
6234 assert!(
6235 routes.contains(&ferrox_api::routes::TOKENIZE),
6236 "the alias must be its own row: {routes:?}"
6237 );
6238 assert!(
6239 !routes.contains(&ferrox_api::routes::V1_TOKENIZE),
6240 "nothing called /v1/tokenize: {routes:?}"
6241 );
6242 }
6243
6244 #[tokio::test]
6249 async fn add_special_prepends_the_same_bos_the_decoder_would() {
6250 let mut cfg = test_dense_fixture();
6251 cfg.vocab_size = 256;
6252 let model = Model::Gguf(GgufModel {
6253 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
6254 tokenizer: Arc::new(ServerTokenizer::Byte),
6255 stop_tokens: StopTokens::default(),
6256 bos_id: Some(7),
6257 is_synthetic: true,
6258 chat_template: chat_template::PromptTemplate::plain(),
6259 });
6260 let app = test_app_with_state(Arc::new(test_state(
6261 model,
6262 ResponseCache::new(1000, Duration::from_secs(3600)),
6263 )));
6264
6265 let (_, plain) = post_json_uri(
6266 &app,
6267 ferrox_api::routes::TOKENIZE,
6268 serde_json::json!({"content": "hi"}),
6269 )
6270 .await;
6271 let (_, special) = post_json_uri(
6272 &app,
6273 ferrox_api::routes::TOKENIZE,
6274 serde_json::json!({"content": "hi", "add_special": true}),
6275 )
6276 .await;
6277
6278 assert_eq!(plain["tokens"], serde_json::json!([104, 105]));
6279 assert_eq!(special["tokens"], serde_json::json!([7, 104, 105]));
6280 assert_eq!(special["count"], 3);
6281 }
6282
6283 #[tokio::test]
6286 async fn a_rejected_embeddings_request_is_recorded_with_its_status() {
6287 let app = test_app();
6288 let (status, _) = post_json_uri(
6289 &app,
6290 ferrox_api::routes::V1_EMBEDDINGS,
6291 serde_json::json!({"input": "hi", "encoding_format": "base64"}),
6292 )
6293 .await;
6294 assert_eq!(status, StatusCode::BAD_REQUEST);
6295
6296 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6297 let recent = stats["recent"].as_array().unwrap();
6298 assert_eq!(recent.len(), 1);
6299 assert_eq!(recent[0]["route"], ferrox_api::routes::V1_EMBEDDINGS);
6300 assert_eq!(recent[0]["status"], 400);
6301 assert_eq!(
6302 recent[0]["prompt_tokens"], 0,
6303 "a rejected call embedded nothing"
6304 );
6305 }
6306
6307 #[tokio::test]
6310 async fn a_row_names_the_key_that_served_it_without_carrying_the_key() {
6311 let app = test_app();
6312 let key = "sk-monitor-secret";
6313 let (status, _) = post_json_with_headers(
6314 &app,
6315 "/v1/chat/completions",
6316 serde_json::json!({
6317 "model": "x",
6318 "messages": [{"role": "user", "content": "hi"}],
6319 "max_tokens": 2
6320 }),
6321 &[
6322 ("authorization", &format!("Bearer {key}")),
6323 ("x-ferrox-client", "ferrox-studio"),
6324 ],
6325 )
6326 .await;
6327 assert_eq!(status, StatusCode::OK);
6328
6329 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6330 let row = stats["recent"].as_array().unwrap()[0].clone();
6331 let fingerprint = row["via_api_key"]
6332 .as_str()
6333 .expect("the row names the key that served it")
6334 .to_string();
6335 assert_eq!(fingerprint, attribution::key_fingerprint(key));
6336 assert!(!fingerprint.contains(key));
6337 assert!(
6338 !serde_json::to_string(&stats).unwrap().contains(key),
6339 "the stats payload must not carry the key in any form"
6340 );
6341 assert_eq!(row["client"], "ferrox-studio");
6342 }
6343
6344 #[tokio::test]
6347 async fn different_keys_are_different_callers_and_no_key_is_null() {
6348 let app = test_app();
6349 let body = serde_json::json!({
6350 "model": "x",
6351 "messages": [{"role": "user", "content": "hi"}],
6352 "max_tokens": 1
6353 });
6354 for headers in [
6355 vec![("authorization", "Bearer key-one")],
6356 vec![("authorization", "Bearer key-two")],
6357 vec![],
6358 ] {
6359 let (status, _) =
6360 post_json_with_headers(&app, "/v1/chat/completions", body.clone(), &headers).await;
6361 assert_eq!(status, StatusCode::OK);
6362 }
6363
6364 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6365 let recent = stats["recent"].as_array().unwrap();
6366 assert_eq!(recent.len(), 3);
6367 let one = recent[0]["via_api_key"].as_str().unwrap();
6368 let two = recent[1]["via_api_key"].as_str().unwrap();
6369 assert_ne!(one, two, "two keys must not collapse into one caller");
6370 assert!(
6371 recent[2]["via_api_key"].is_null(),
6372 "an unauthenticated call is null, not a fingerprint of nothing"
6373 );
6374 assert!(recent[2]["client"].is_null());
6375 }
6376
6377 #[tokio::test]
6382 async fn a_row_names_the_model_that_served_it_not_the_one_requested() {
6383 let state = Arc::new(test_state(
6384 named_test_model("really-loaded", 256),
6385 ResponseCache::new(4, Duration::from_secs(60)),
6386 ));
6387 let app = test_app_with_state(Arc::clone(&state));
6388
6389 let (status, _) = post_json_uri(
6390 &app,
6391 "/v1/chat/completions",
6392 serde_json::json!({
6393 "model": "gpt-4-turbo-that-is-not-here",
6394 "messages": [{"role": "user", "content": "hi"}],
6395 "max_tokens": 2
6396 }),
6397 )
6398 .await;
6399 assert_eq!(status, StatusCode::OK);
6400
6401 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6402 assert_eq!(stats["recent"][0]["model"], "really-loaded");
6403
6404 state.swap_active(None);
6407 let (status, _) = post_json_uri(
6408 &app,
6409 "/v1/chat/completions",
6410 serde_json::json!({
6411 "model": "gpt-4-turbo-that-is-not-here",
6412 "messages": [{"role": "user", "content": "hi"}]
6413 }),
6414 )
6415 .await;
6416 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
6417 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6418 let recent = stats["recent"].as_array().unwrap();
6419 assert!(recent[recent.len() - 1]["model"].is_null());
6420 }
6421
6422 #[tokio::test]
6426 async fn a_streamed_row_names_the_model_it_decoded_against() {
6427 let state = Arc::new(test_state(
6428 named_test_model("model-before", 256),
6429 ResponseCache::new(4, Duration::from_secs(60)),
6430 ));
6431 let app = test_app_with_state(Arc::clone(&state));
6432 let _ = post_sse_raw(&app, resumable_request()).await;
6433 active_model(&state, "model-after");
6435
6436 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6437 assert_eq!(stats["recent"][0]["model"], "model-before");
6438 }
6439
6440 #[tokio::test]
6443 async fn the_queue_gauge_is_null_when_nothing_can_queue() {
6444 let app = test_app();
6445 let (status, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6446 assert_eq!(status, StatusCode::OK);
6447 assert!(
6448 stats["queue_depth"].is_null(),
6449 "without continuous batching nothing queues, so there is nothing to measure"
6450 );
6451 assert!(stats["queue_rejected_total"].is_null());
6452 assert_eq!(
6453 stats["generating_now"], 0,
6454 "work in progress is measured and really is zero here"
6455 );
6456 }
6457
6458 async fn post_sse_raw(app: &Router, body: serde_json::Value) -> String {
6463 post_sse_raw_uri(app, ferrox_api::routes::V1_CHAT_COMPLETIONS, body).await
6464 }
6465
6466 async fn post_sse_raw_uri(app: &Router, uri: &str, body: serde_json::Value) -> String {
6470 use http_body_util::BodyExt;
6471 use tower::ServiceExt;
6472
6473 let response = app
6474 .clone()
6475 .oneshot(
6476 axum::http::Request::builder()
6477 .method("POST")
6478 .uri(uri)
6479 .header("content-type", "application/json")
6480 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
6481 .unwrap(),
6482 )
6483 .await
6484 .unwrap();
6485 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6486 String::from_utf8(bytes.to_vec()).unwrap()
6487 }
6488
6489 async fn get_json_with_headers(
6490 app: &Router,
6491 uri: &str,
6492 headers: &[(&str, &str)],
6493 ) -> (StatusCode, serde_json::Value) {
6494 use http_body_util::BodyExt;
6495 use tower::ServiceExt;
6496
6497 let mut builder = axum::http::Request::builder().method("GET").uri(uri);
6498 for (name, value) in headers {
6499 builder = builder.header(*name, *value);
6500 }
6501 let response = app
6502 .clone()
6503 .oneshot(builder.body(axum::body::Body::empty()).unwrap())
6504 .await
6505 .unwrap();
6506 let status = response.status();
6507 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6508 (
6509 status,
6510 serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({})),
6511 )
6512 }
6513
6514 fn sse_field<'a>(body: &'a str, field: &str) -> Vec<&'a str> {
6515 body.lines()
6516 .filter_map(|line| line.strip_prefix(field))
6517 .map(str::trim)
6518 .collect()
6519 }
6520
6521 fn resumable_request() -> serde_json::Value {
6522 serde_json::json!({
6523 "model": "m",
6524 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
6525 "max_tokens": 4,
6526 "temperature": 0,
6527 "stream": true,
6528 "stream_resumable": true,
6529 })
6530 }
6531
6532 #[tokio::test]
6537 async fn a_resumable_stream_numbers_every_event_and_states_retry_once() {
6538 let app = test_app();
6539 let body = post_sse_raw(&app, resumable_request()).await;
6540
6541 let request_id = body
6542 .lines()
6543 .find_map(|l| l.strip_prefix("data: "))
6544 .and_then(|d| serde_json::from_str::<serde_json::Value>(d).ok())
6545 .and_then(|v| v["request_id"].as_str().map(str::to_string))
6546 .expect("the first chunk names the request");
6547
6548 let ids = sse_field(&body, "id:");
6549 let datas = sse_field(&body, "data:");
6550 assert_eq!(
6551 ids.len(),
6552 datas.len(),
6553 "every event carries an id, or a reconnect cannot name where it stopped"
6554 );
6555 for (i, id) in ids.iter().enumerate() {
6556 assert_eq!(*id, format!("{request_id}:{i}"));
6557 }
6558 let retries = sse_field(&body, "retry:");
6559 assert_eq!(
6560 retries.len(),
6561 1,
6562 "the reconnect delay is stated once, not on every event"
6563 );
6564 assert_eq!(retries[0], "1500");
6565 assert!(
6566 body.contains("data: [DONE]"),
6567 "the end of stream is still stated"
6568 );
6569 }
6570
6571 #[tokio::test]
6575 async fn a_plain_stream_carries_no_id_because_nothing_could_replay_it() {
6576 let app = test_app();
6577 let mut request = resumable_request();
6578 request["stream_resumable"] = serde_json::json!(false);
6579 let body = post_sse_raw(&app, request).await;
6580 assert!(!sse_field(&body, "data:").is_empty(), "it still streams");
6581 assert!(
6582 sse_field(&body, "id:").is_empty(),
6583 "an id promises a replay this stream cannot serve"
6584 );
6585 assert!(sse_field(&body, "retry:").is_empty());
6586 }
6587
6588 #[tokio::test]
6592 async fn the_polling_fallback_serves_exactly_what_the_stream_delivered() {
6593 let app = test_app();
6594 let body = post_sse_raw(&app, resumable_request()).await;
6595 let request_id = sse_field(&body, "id:")[0]
6596 .rsplit_once(':')
6597 .unwrap()
6598 .0
6599 .to_string();
6600 let streamed: Vec<String> = sse_field(&body, "data:")
6601 .iter()
6602 .map(|d| d.to_string())
6603 .collect();
6604
6605 let (status, polled) = get_json(
6606 &app,
6607 &format!("{}?from=0", ferrox_api::routes::v1_stream_poll(&request_id)),
6608 )
6609 .await;
6610 assert_eq!(status, StatusCode::OK);
6611 let events: Vec<String> = polled["events"]
6612 .as_array()
6613 .unwrap()
6614 .iter()
6615 .map(|e| e["data"].as_str().unwrap().to_string())
6616 .collect();
6617 assert_eq!(
6618 events, streamed,
6619 "the fallback must deliver the same answer, not a re-run of it"
6620 );
6621 assert_eq!(polled["request_id"], request_id);
6622 assert_eq!(
6623 polled["done"], false,
6624 "events were still being handed out, so the client must ask again"
6625 );
6626
6627 let next = polled["next_index"].as_u64().unwrap();
6630 let (_, drained) = get_json(
6631 &app,
6632 &format!(
6633 "{}?from={next}",
6634 ferrox_api::routes::v1_stream_poll(&request_id)
6635 ),
6636 )
6637 .await;
6638 assert_eq!(drained["done"], true);
6639 assert_eq!(drained["events"].as_array().unwrap().len(), 0);
6640 }
6641
6642 #[tokio::test]
6646 async fn a_resume_continues_after_the_last_event_id_rather_than_repeating() {
6647 let app = test_app();
6648 let body = post_sse_raw(&app, resumable_request()).await;
6649 let ids = sse_field(&body, "id:");
6650 let datas: Vec<String> = sse_field(&body, "data:")
6651 .iter()
6652 .map(|d| d.to_string())
6653 .collect();
6654 assert!(
6655 ids.len() >= 3,
6656 "need a few events to resume into the middle"
6657 );
6658 let request_id = ids[0].rsplit_once(':').unwrap().0.to_string();
6659
6660 let (status, resumed) = get_json_with_headers(
6661 &app,
6662 &format!("{}/poll", ferrox_api::routes::v1_stream(&request_id)),
6663 &[],
6664 )
6665 .await;
6666 assert_eq!(status, StatusCode::OK);
6667 assert_eq!(resumed["events"].as_array().unwrap().len(), datas.len());
6668
6669 let (_, tail) = get_json(
6671 &app,
6672 &format!("{}?from=2", ferrox_api::routes::v1_stream_poll(&request_id)),
6673 )
6674 .await;
6675 let tail_events: Vec<String> = tail["events"]
6676 .as_array()
6677 .unwrap()
6678 .iter()
6679 .map(|e| e["data"].as_str().unwrap().to_string())
6680 .collect();
6681 assert_eq!(tail_events, datas[2..].to_vec());
6682 }
6683
6684 #[tokio::test]
6687 async fn an_sse_reconnect_resumes_from_the_last_event_id() {
6688 use http_body_util::BodyExt;
6689 use tower::ServiceExt;
6690
6691 let app = test_app();
6692 let body = post_sse_raw(&app, resumable_request()).await;
6693 let ids = sse_field(&body, "id:");
6694 let datas: Vec<String> = sse_field(&body, "data:")
6695 .iter()
6696 .map(|d| d.to_string())
6697 .collect();
6698 let request_id = ids[0].rsplit_once(':').unwrap().0.to_string();
6699
6700 let response = app
6701 .clone()
6702 .oneshot(
6703 axum::http::Request::builder()
6704 .method("GET")
6705 .uri(ferrox_api::routes::v1_stream(&request_id))
6706 .header("last-event-id", format!("{request_id}:0"))
6707 .body(axum::body::Body::empty())
6708 .unwrap(),
6709 )
6710 .await
6711 .unwrap();
6712 assert_eq!(response.status(), StatusCode::OK);
6713 assert_eq!(
6714 response
6715 .headers()
6716 .get("x-accel-buffering")
6717 .and_then(|v| v.to_str().ok()),
6718 Some("no"),
6719 "the reconnect needs the same anti-buffering header as the stream"
6720 );
6721 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6722 let resumed = String::from_utf8(bytes.to_vec()).unwrap();
6723 assert_eq!(
6724 sse_field(&resumed, "data:")
6725 .iter()
6726 .map(|d| d.to_string())
6727 .collect::<Vec<_>>(),
6728 datas[1..].to_vec()
6729 );
6730 assert_eq!(sse_field(&resumed, "id:")[0], format!("{request_id}:1"));
6731 }
6732
6733 #[tokio::test]
6737 async fn a_last_event_id_from_another_stream_is_refused() {
6738 let app = test_app();
6739 let body = post_sse_raw(&app, resumable_request()).await;
6740 let request_id = sse_field(&body, "id:")[0]
6741 .rsplit_once(':')
6742 .unwrap()
6743 .0
6744 .to_string();
6745
6746 let (status, err) = get_json_with_headers(
6747 &app,
6748 &ferrox_api::routes::v1_stream(&request_id),
6749 &[("last-event-id", "chatcmpl-someone-else:3")],
6750 )
6751 .await;
6752 assert_eq!(status, StatusCode::BAD_REQUEST);
6753 assert_eq!(err["error"]["code"], "bad_last_event_id");
6754 }
6755
6756 #[tokio::test]
6760 async fn resuming_a_stream_that_was_never_resumable_is_a_404_that_says_why() {
6761 let app = test_app();
6762 let mut request = resumable_request();
6763 request["stream_resumable"] = serde_json::json!(false);
6764 let body = post_sse_raw(&app, request).await;
6765 let request_id = body
6766 .lines()
6767 .find_map(|l| l.strip_prefix("data: "))
6768 .and_then(|d| serde_json::from_str::<serde_json::Value>(d).ok())
6769 .and_then(|v| v["request_id"].as_str().map(str::to_string))
6770 .unwrap();
6771
6772 let (status, err) = get_json(&app, &ferrox_api::routes::v1_stream_poll(&request_id)).await;
6773 assert_eq!(status, StatusCode::NOT_FOUND);
6774 assert_eq!(err["error"]["code"], "stream_not_found");
6775 assert!(err["error"]["message"]
6776 .as_str()
6777 .unwrap()
6778 .contains("stream_resumable"));
6779 }
6780
6781 #[test]
6785 fn the_axum_stream_patterns_match_the_published_templates() {
6786 assert_eq!(
6787 axum_path(ferrox_api::routes::V1_STREAM),
6788 "/v1/stream/:request_id"
6789 );
6790 assert_eq!(
6791 axum_path(ferrox_api::routes::V1_STREAM_POLL),
6792 "/v1/stream/:request_id/poll"
6793 );
6794 assert_eq!(
6795 ferrox_api::routes::v1_stream("abc"),
6796 axum_path(ferrox_api::routes::V1_STREAM).replace(":request_id", "abc")
6797 );
6798 }
6799
6800 #[test]
6810 fn no_published_template_reaches_the_router_with_its_braces() {
6811 for template in [
6812 ferrox_api::routes::V1_STREAM,
6813 ferrox_api::routes::V1_STREAM_POLL,
6814 ferrox_api::routes::V1_RESPONSE,
6815 ferrox_api::routes::V1_RESPONSE_CANCEL,
6816 ferrox_api::routes::ADMIN_TASK_CANCEL,
6817 ] {
6818 assert!(
6819 template.contains('{'),
6820 "{template} is in the template list but has no placeholder"
6821 );
6822 let mounted = axum_path(template);
6823 assert!(
6824 !mounted.contains('{') && !mounted.contains('}'),
6825 "{template} would be mounted as {mounted}, whose braces axum reads as a literal segment"
6826 );
6827 assert!(
6828 mounted.contains(':'),
6829 "{template} lost its placeholder entirely and would match one path only"
6830 );
6831 }
6832 }
6833
6834 #[tokio::test]
6842 async fn an_unknown_response_id_gets_the_handler_not_a_bare_404() {
6843 let app = test_app();
6844 let (status, body) = get_json(&app, "/v1/responses/resp_nonexistent").await;
6845 assert_eq!(status, StatusCode::NOT_FOUND);
6846 assert!(
6847 !body.is_null(),
6848 "empty body means axum never matched the route, so the id was read as a literal segment"
6849 );
6850 }
6851
6852 #[tokio::test]
6855 async fn the_task_list_starts_empty_rather_than_absent() {
6856 let app = test_app();
6857 let (status, body) = get_json(&app, ferrox_api::routes::ADMIN_TASKS).await;
6858 assert_eq!(status, StatusCode::OK);
6859 assert_eq!(body["tasks"].as_array().unwrap().len(), 0);
6860 }
6861
6862 async fn post_json_uri(
6863 app: &Router,
6864 uri: &str,
6865 body: serde_json::Value,
6866 ) -> (StatusCode, serde_json::Value) {
6867 use http_body_util::BodyExt;
6868 use tower::ServiceExt;
6869
6870 let response = app
6871 .clone()
6872 .oneshot(
6873 axum::http::Request::builder()
6874 .method("POST")
6875 .uri(uri)
6876 .header("content-type", "application/json")
6877 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
6878 .unwrap(),
6879 )
6880 .await
6881 .unwrap();
6882 let status = response.status();
6883 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6884 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
6885 (status, json)
6886 }
6887
6888 async fn post_json(app: &Router, body: serde_json::Value) -> serde_json::Value {
6889 post_json_uri(app, "/v1/chat/completions", body).await.1
6890 }
6891
6892 #[tokio::test]
6901 async fn stats_says_what_the_engine_is_using_and_which_quantity_that_is() {
6902 let app = test_app();
6903 let (status, body) = get_json(&app, ferrox_api::routes::V1_STATS).await;
6904 assert_eq!(status, StatusCode::OK);
6905
6906 let memory = &body["memory"];
6907 if memory.is_null() {
6908 return;
6911 }
6912 assert!(
6913 memory["bytes"].as_u64().is_some_and(|b| b > 0),
6914 "a read that produced a zero is a broken read, not an idle \
6915 engine: {memory}"
6916 );
6917 assert!(
6918 ["pss", "rss"].contains(&memory["kind"].as_str().unwrap_or("")),
6919 "the quantity must travel with the number: {memory}"
6920 );
6921 }
6922
6923 #[tokio::test]
6929 async fn stats_reports_a_pool_it_does_not_have_as_absent_and_not_as_zero() {
6930 let app = test_app();
6931 let (status, body) = get_json(&app, ferrox_api::routes::V1_STATS).await;
6932 assert_eq!(status, StatusCode::OK);
6933 for pool in ["kv_pages", "window_slots", "state_slots"] {
6934 assert!(
6935 body["pools"][pool].is_null(),
6936 "{pool} must be null rather than a zero row: {}",
6937 body["pools"]
6938 );
6939 }
6940 }
6941
6942 #[tokio::test]
6950 async fn a_messages_response_states_the_id_that_v1_cancel_takes() {
6951 use http_body_util::BodyExt;
6952 use tower::ServiceExt;
6953
6954 let app = test_app();
6955 let send = |body: serde_json::Value| {
6956 let app = app.clone();
6957 async move {
6958 app.oneshot(
6959 axum::http::Request::builder()
6960 .method("POST")
6961 .uri(ferrox_api::routes::V1_MESSAGES)
6962 .header("content-type", "application/json")
6963 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
6964 .unwrap(),
6965 )
6966 .await
6967 .unwrap()
6968 }
6969 };
6970
6971 let ok = send(serde_json::json!({
6972 "model": "test",
6973 "max_tokens": 1,
6974 "messages": [{"role": "user", "content": "hi"}],
6975 }))
6976 .await;
6977 assert_eq!(ok.status(), StatusCode::OK);
6978 let id = ok
6979 .headers()
6980 .get("request-id")
6981 .expect("a served message names its id")
6982 .to_str()
6983 .unwrap()
6984 .to_string();
6985 assert!(!id.is_empty());
6986
6987 let bad = send(serde_json::json!({"model": "test"})).await;
6990 assert!(bad.status().is_client_error());
6991 let other = bad.headers().get("request-id").expect("errors too");
6992 assert_ne!(other.to_str().unwrap(), id);
6993 let _ = bad.into_body().collect().await.unwrap();
6994 }
6995
6996 #[tokio::test]
7004 async fn a_request_that_arrives_mid_rebuild_is_refused_and_admitted_again_after() {
7005 let state = Arc::new(test_state(
7006 test_model_full_byte_vocab(),
7007 ResponseCache::new(1000, Duration::from_secs(3600)),
7008 ));
7009 let app = test_app_with_state(Arc::clone(&state));
7010 let body = serde_json::json!({
7011 "model": "test",
7012 "messages": [{"role": "user", "content": "hi"}],
7013 "max_tokens": 1,
7014 });
7015
7016 state
7017 .maintenance
7018 .lock()
7019 .unwrap()
7020 .begin_rebuild()
7021 .expect("a fresh server is serving, so the rebuild starts");
7022 let (status, refused) = post_json_uri(&app, "/v1/chat/completions", body.clone()).await;
7023 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
7024 assert_eq!(refused["error"]["type"], "cache_rebuilding");
7025
7026 state.maintenance.lock().unwrap().finish_rebuild(true);
7027 let (status, _) = post_json_uri(&app, "/v1/chat/completions", body).await;
7028 assert_eq!(
7029 status,
7030 StatusCode::OK,
7031 "the gate reopens; a rebuild is not a latch"
7032 );
7033 }
7034
7035 #[tokio::test]
7040 async fn cancelling_an_id_that_is_not_generating_is_a_404_that_says_so() {
7041 let app = test_app();
7042 let (status, body) = post_json_uri(
7043 &app,
7044 ferrox_api::routes::V1_CANCEL,
7045 serde_json::json!({ "request_id": "chatcmpl-never-issued" }),
7046 )
7047 .await;
7048 assert_eq!(status, StatusCode::NOT_FOUND);
7049 assert_eq!(body["cancelled"], serde_json::json!(false));
7050 assert_eq!(body["request_id"], "chatcmpl-never-issued");
7051 assert!(
7052 body["detail"].as_str().is_some_and(|d| !d.is_empty()),
7053 "the verdict must carry a human reason: {body}"
7054 );
7055 }
7056
7057 #[tokio::test]
7061 async fn cancelling_a_live_generation_signals_its_token_and_answers_200() {
7062 let state = Arc::new(test_state(
7063 test_model_full_byte_vocab(),
7064 ResponseCache::new(1000, Duration::from_secs(3600)),
7065 ));
7066 let app = test_app_with_state(Arc::clone(&state));
7067 let (token, _guard) = state.cancels.register("chatcmpl-live");
7068
7069 let (status, before) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
7070 assert_eq!(status, StatusCode::OK);
7071 assert_eq!(before["generating_now"], serde_json::json!(1));
7072
7073 let (status, body) = post_json_uri(
7074 &app,
7075 ferrox_api::routes::V1_CANCEL,
7076 serde_json::json!({ "request_id": "chatcmpl-live" }),
7077 )
7078 .await;
7079 assert_eq!(status, StatusCode::OK);
7080 assert_eq!(body["cancelled"], serde_json::json!(true));
7081 assert!(
7082 token.is_cancelled(),
7083 "the endpoint answered ok without setting the flag the decode loop reads"
7084 );
7085 }
7086
7087 #[tokio::test]
7088 async fn tokenize_detokenize_roundtrip_and_embeddings_mean() {
7089 let app = test_app();
7090 let (status, tok) =
7091 post_json_uri(&app, "/v1/tokenize", serde_json::json!({ "prompt": "Hi" })).await;
7092 assert_eq!(status, StatusCode::OK);
7093 let tokens = tok["tokens"].as_array().unwrap();
7094 assert_eq!(tok["count"], tokens.len());
7095 assert!(!tokens.is_empty());
7096
7097 let (status, detok) = post_json_uri(
7098 &app,
7099 "/v1/detokenize",
7100 serde_json::json!({ "tokens": tokens }),
7101 )
7102 .await;
7103 assert_eq!(status, StatusCode::OK);
7104 assert_eq!(detok["text"], "Hi");
7105
7106 let (status, emb) = post_json_uri(
7107 &app,
7108 "/v1/embeddings",
7109 serde_json::json!({
7110 "input": "Hi",
7111 "embedding_type": "mean"
7112 }),
7113 )
7114 .await;
7115 assert_eq!(status, StatusCode::OK);
7116 let vec = emb["data"][0]["embedding"].as_array().unwrap();
7117 assert!(!vec.is_empty());
7118 assert!(vec.iter().all(|v| v.as_f64().is_some()));
7119 }
7120
7121 #[tokio::test]
7127 async fn the_decoder_path_still_refuses_a_pooling_it_cannot_mean() {
7128 let app = test_app();
7129 let (status, body) = post_json_uri(
7130 &app,
7131 "/v1/embeddings",
7132 serde_json::json!({ "input": "Hi", "embedding_type": "cls" }),
7133 )
7134 .await;
7135 assert_eq!(status, StatusCode::BAD_REQUEST);
7136 let msg = body["error"]["message"].as_str().unwrap();
7137 assert!(msg.contains("mean") && msg.contains("last"), "{msg}");
7138 }
7139
7140 #[tokio::test]
7145 #[ignore = "needs models/bge-small-en-v1.5-q8_0.gguf"]
7146 async fn a_real_embedding_model_serves_v1_embeddings() {
7147 let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
7148 .join("../../models/bge-small-en-v1.5-q8_0.gguf");
7149 if !path.exists() {
7150 eprintln!("SKIP: {} not present", path.display());
7151 return;
7152 }
7153 let encoder = ferrox_models::EmbeddingModel::from_gguf_path(&path).expect("load bge");
7154 let mut state = test_state(
7155 test_model_full_byte_vocab(),
7156 ResponseCache::new(1000, Duration::from_secs(3600)),
7157 );
7158 state.embedding = Some(Arc::new(encoder));
7159 let app = test_app_with_state(Arc::new(state));
7160
7161 let (status, body) = post_json_uri(
7162 &app,
7163 "/v1/embeddings",
7164 serde_json::json!({ "input": ["Hello world", "a second input"] }),
7165 )
7166 .await;
7167 assert_eq!(status, StatusCode::OK, "{body}");
7168 assert_eq!(body["model"], "bge-small-en-v1.5");
7169 let data = body["data"].as_array().unwrap();
7170 assert_eq!(data.len(), 2);
7171 for (i, row) in data.iter().enumerate() {
7172 assert_eq!(row["index"], i);
7173 let v: Vec<f64> = row["embedding"]
7174 .as_array()
7175 .unwrap()
7176 .iter()
7177 .map(|x| x.as_f64().unwrap())
7178 .collect();
7179 assert_eq!(v.len(), 384, "the encoder\'s width, not the decoder\'s");
7180 let norm = v.iter().map(|x| x * x).sum::<f64>().sqrt();
7181 assert!((norm - 1.0).abs() < 1e-4, "not L2-normalized: {norm}");
7182 }
7183 assert!(body["usage"]["prompt_tokens"].as_u64().unwrap() >= 4 + 2);
7186
7187 let (status, mean) = post_json_uri(
7191 &app,
7192 "/v1/embeddings",
7193 serde_json::json!({ "input": "Hello world", "embedding_type": "mean" }),
7194 )
7195 .await;
7196 assert_eq!(status, StatusCode::OK);
7197 assert_ne!(mean["data"][0]["embedding"], data[0]["embedding"]);
7198 }
7199
7200 #[tokio::test]
7212 #[ignore = "needs models/bge-small-en-v1.5-q8_0.gguf"]
7213 async fn an_encoder_can_be_the_loaded_model() {
7214 let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
7215 .join("../../models/bge-small-en-v1.5-q8_0.gguf");
7216 if !path.exists() {
7217 eprintln!("SKIP: {} not present", path.display());
7218 return;
7219 }
7220
7221 let loaded = model::load_from_path(path.to_str().unwrap()).expect("load bge as the model");
7225 assert!(
7226 matches!(loaded, model::LoadedModel::Encoder(_)),
7227 "an encoder-only GGUF reached a decoder loader"
7228 );
7229 let (loaded, batcher, ceiling) = activate_loaded_model(loaded, true, None, None);
7230 assert!(
7231 matches!(loaded, Loaded::Encoder(_)),
7232 "the encoder did not stay an encoder through activation"
7233 );
7234 assert!(
7235 batcher.is_none() && ceiling.is_none(),
7236 "an encoder was given a decode batcher or a KV ceiling it has no use for"
7237 );
7238
7239 let state = test_state(
7240 test_model_full_byte_vocab(),
7241 ResponseCache::new(1000, Duration::from_secs(3600)),
7242 );
7243 state.swap_active(Some(Arc::new(ActiveModel {
7244 id: None,
7245 loaded,
7246 batcher,
7247 ceiling,
7248 })));
7249 let app = test_app_with_state(Arc::new(state));
7250
7251 let (status, body) = post_json_uri(
7253 &app,
7254 "/v1/embeddings",
7255 serde_json::json!({ "input": "Hello world" }),
7256 )
7257 .await;
7258 assert_eq!(status, StatusCode::OK, "{body}");
7259 assert_eq!(body["model"], "bge-small-en-v1.5");
7260 let v = body["data"][0]["embedding"].as_array().unwrap();
7261 assert_eq!(v.len(), 384, "the encoder's width, not the decoder's");
7262
7263 let (status, body) = post_json_uri(
7265 &app,
7266 "/v1/chat/completions",
7267 serde_json::json!({
7268 "model": "bge-small-en-v1.5",
7269 "messages": [{"role": "user", "content": "hi"}],
7270 }),
7271 )
7272 .await;
7273 assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{body}");
7274 let msg = body["error"]["message"].as_str().unwrap();
7275 for fact in [
7276 "bge-small-en-v1.5",
7277 "bert",
7278 "embedding model",
7279 "/v1/embeddings",
7280 ] {
7281 assert!(msg.contains(fact), "the refusal does not say {fact}: {msg}");
7282 }
7283
7284 let (status, models) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
7286 assert_eq!(status, StatusCode::OK);
7287 let entry = &models["data"][0];
7288 assert_eq!(entry["id"], "bge-small-en-v1.5");
7289 assert_eq!(entry["ferrox_model_kind"], "embedding");
7290 assert_eq!(entry["ferrox_tokenizer"], "gguf-wordpiece");
7291 assert_eq!(entry["ferrox_n_embd"], 384);
7292 assert_eq!(entry["ferrox_pooling"], "CLS");
7293 assert_eq!(
7294 entry["ferrox_endpoints"],
7295 serde_json::json!(["/v1/embeddings"])
7296 );
7297 assert!(entry.get("supported_reasoning_efforts").is_none());
7300
7301 let (status, health) = get_json(&app, ferrox_api::routes::HEALTH).await;
7303 assert_eq!(status, StatusCode::OK, "an encoder is a loaded model");
7304 assert_eq!(health["model"]["id"], "bge-small-en-v1.5");
7305 assert_eq!(health["model"]["synthetic_weights"], false);
7306 let weights = health["capabilities"]
7307 .as_array()
7308 .unwrap()
7309 .iter()
7310 .find(|c| c["id"] == ferrox_api::health::capability::REAL_WEIGHTS)
7311 .expect("a real-weights capability row");
7312 let detail = weights["detail"].as_str().unwrap_or_default();
7313 assert!(detail.contains("ENCODER"), "{detail}");
7314 let (status, body) = post_json_uri(
7322 &app,
7323 ferrox_api::routes::V1_TOKENIZE,
7324 serde_json::json!({ "content": "hello world" }),
7325 )
7326 .await;
7327 assert_eq!(
7328 status,
7329 StatusCode::OK,
7330 "an encoder has a real tokenizer: {body}"
7331 );
7332 let tokens = body["tokens"].as_array().expect("tokens array").clone();
7333 assert!(!tokens.is_empty(), "WordPiece produced nothing: {body}");
7334
7335 let (status, body) = post_json_uri(
7336 &app,
7337 ferrox_api::routes::V1_DETOKENIZE,
7338 serde_json::json!({ "tokens": tokens }),
7339 )
7340 .await;
7341 assert_eq!(status, StatusCode::OK, "{body}");
7342 let round_tripped = body["content"].as_str().expect("content").to_string();
7343 assert!(
7344 round_tripped.contains("hello") && round_tripped.contains("world"),
7345 "the ids did not decode back through the encoder's own vocabulary: {round_tripped}"
7346 );
7347
7348 let (status, _) = post_json_uri(
7351 &app,
7352 "/v1/completions",
7353 serde_json::json!({ "model": "m", "prompt": "hi", "max_tokens": 1 }),
7354 )
7355 .await;
7356 assert_eq!(
7357 status,
7358 StatusCode::NOT_IMPLEMENTED,
7359 "tokenizing an encoder must not have opened a path to generating with one"
7360 );
7361 }
7362
7363 #[tokio::test]
7368 async fn metrics_exposes_expert_store_counters_when_streaming_is_active() {
7369 use http_body_util::BodyExt;
7370 use tower::ServiceExt;
7371
7372 let fixture = concat!(
7373 "../ferrox-models/tests/fixtures/",
7374 "ferrox_real_moe_test.gguf"
7375 );
7376 let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join(fixture);
7377 let decoder = Decoder::from_gguf_with_expert_cache(
7378 &fixture,
7379 ferrox_models::config::test_moe_fixture(),
7380 Some(1024 * 1024),
7381 )
7382 .expect("MoE fixture must load store-backed");
7383
7384 let mut caches: Vec<ferrox_core::cache::KvCache> = decoder
7388 .layers
7389 .iter()
7390 .map(|_| {
7391 ferrox_core::cache::KvCache::new(decoder.config.n_kv_heads, decoder.config.head_dim)
7392 })
7393 .collect();
7394 decoder.forward_token(1, 0, &mut caches);
7395
7396 let model = Model::Gguf(GgufModel {
7397 decoder: Arc::new(decoder),
7398 tokenizer: Arc::new(ServerTokenizer::Byte),
7399 stop_tokens: StopTokens::default(),
7400 bos_id: None,
7401 is_synthetic: false,
7402 chat_template: chat_template::PromptTemplate::plain(),
7403 });
7404 let state = Arc::new(test_state(
7405 model,
7406 ResponseCache::new(16, Duration::from_secs(60)),
7407 ));
7408 let app = Router::new()
7409 .route("/metrics", axum::routing::get(metrics))
7410 .route("/v1/chat/completions", post(chat_completions))
7411 .with_state(state);
7412
7413 let fetch_metrics = |app: Router| async move {
7414 let resp = app
7415 .oneshot(
7416 axum::http::Request::builder()
7417 .method("GET")
7418 .uri("/metrics")
7419 .body(axum::body::Body::empty())
7420 .unwrap(),
7421 )
7422 .await
7423 .unwrap();
7424 let bytes = resp.into_body().collect().await.unwrap().to_bytes();
7425 String::from_utf8(bytes.to_vec()).unwrap()
7426 };
7427
7428 let after = fetch_metrics(app.clone()).await;
7429 assert!(
7430 after.contains("ferrox_expert_cache_misses_total"),
7431 "streaming model must expose expert-cache metrics: {after}"
7432 );
7433 let misses: u64 = after
7434 .lines()
7435 .find(|l| l.starts_with("ferrox_expert_cache_misses_total"))
7436 .and_then(|l| l.split_whitespace().nth(1))
7437 .and_then(|v| v.parse().ok())
7438 .expect("misses metric line must parse");
7439 assert!(
7440 misses > 0,
7441 "decode must have read experts through the store: {after}"
7442 );
7443 }
7444
7445 fn weather_tool() -> serde_json::Value {
7446 serde_json::json!({
7447 "type": "function",
7448 "function": {
7449 "name": "get_weather",
7450 "description": "Get the current weather for a location.",
7451 "parameters": {
7452 "type": "object",
7453 "properties": {"location": {"type": "string"}},
7454 "required": ["location"]
7455 }
7456 }
7457 })
7458 }
7459
7460 fn weather_tool_def() -> ToolDef {
7461 ToolDef {
7462 kind: "function".to_string(),
7463 function: ToolFunctionDef {
7464 name: "get_weather".to_string(),
7465 description: Some("Get the current weather for a location.".to_string()),
7466 parameters: Some(serde_json::json!({
7467 "type": "object",
7468 "properties": {"location": {"type": "string"}},
7469 "required": ["location"]
7470 })),
7471 },
7472 }
7473 }
7474
7475 #[test]
7476 fn tool_preamble_mentions_every_tool_name_and_description() {
7477 let preamble = tool_preamble(&[weather_tool_def()]);
7478 assert!(preamble.contains("get_weather"));
7479 assert!(preamble.contains("Get the current weather for a location."));
7480 assert!(preamble.contains("<tool_call>"));
7481 assert!(preamble.contains("</tool_call>"));
7482 }
7483
7484 #[test]
7485 fn a_real_marker_becomes_a_structured_tool_call() {
7486 let text = "sure, let me check.<tool_call>{\"name\": \"get_weather\", \"arguments\": {\"location\": \"Paris\"}}</tool_call>";
7487 let (message, finish) = build_response_message(
7488 text.to_string(),
7489 &[weather_tool_def()],
7490 output::OutputPosture::for_model("test-model"),
7491 "stop",
7492 );
7493 assert_eq!(finish, "tool_calls");
7494 let calls = message.tool_calls.expect("must carry a tool call");
7495 assert_eq!(calls[0].function.name, "get_weather");
7496 let parsed: serde_json::Value = serde_json::from_str(&calls[0].function.arguments).unwrap();
7497 assert_eq!(parsed["location"], "Paris");
7498 }
7499
7500 #[test]
7501 fn a_plain_answer_is_not_promoted_to_a_tool_call() {
7502 let (message, finish) = build_response_message(
7503 "just an answer".to_string(),
7504 &[weather_tool_def()],
7505 output::OutputPosture::for_model("test-model"),
7506 "stop",
7507 );
7508 assert_eq!(finish, "stop");
7509 assert!(message.tool_calls.is_none());
7510 assert_eq!(message.content.as_deref(), Some("just an answer"));
7511 }
7512
7513 #[test]
7516 fn a_malformed_payload_is_not_a_tool_call() {
7517 let (message, finish) = build_response_message(
7518 "<tool_call>not valid json at all</tool_call>".to_string(),
7519 &[weather_tool_def()],
7520 output::OutputPosture::for_model("test-model"),
7521 "stop",
7522 );
7523 assert_eq!(finish, "stop");
7524 assert!(message.tool_calls.is_none());
7525 }
7526
7527 #[test]
7530 fn a_tool_that_was_never_offered_is_not_returned() {
7531 let (message, finish) = build_response_message(
7532 "<tool_call>{\"name\": \"ping\", \"arguments\": {}}</tool_call>".to_string(),
7533 &[weather_tool_def()],
7534 output::OutputPosture::for_model("test-model"),
7535 "stop",
7536 );
7537 assert_eq!(finish, "stop");
7538 assert!(message.tool_calls.is_none());
7539 }
7540
7541 #[test]
7543 fn marker_text_with_no_tools_offered_stays_content() {
7544 let (message, finish) = build_response_message(
7545 "<tool_call>{\"name\": \"get_weather\", \"arguments\": {}}</tool_call>".to_string(),
7546 &[],
7547 output::OutputPosture::for_model("test-model"),
7548 "stop",
7549 );
7550 assert_eq!(finish, "stop");
7551 assert!(message.tool_calls.is_none());
7552 assert!(message.content.is_some());
7553 }
7554
7555 #[test]
7559 fn a_streamed_call_opens_then_delivers_its_arguments_in_pieces() {
7560 let opened = std::cell::Cell::new(0usize);
7561 let mut parser = crate::policy::parser::ToolCallParser::new(
7562 crate::policy::parser::ToolCallFormat::Qwen3Coder,
7563 vec![
7564 crate::policy::parser::tool_call::ToolSchema::with_parameters(
7565 "write_file",
7566 serde_json::json!({"type": "object", "properties": {
7567 "path": {"type": "string"},
7568 "contents": {"type": "string"}
7569 }}),
7570 ),
7571 ],
7572 );
7573 let wire = "<tool_call><function=write_file>\
7574 <parameter=path>\n/tmp/x\n</parameter>\
7575 <parameter=contents>\nhello world\n</parameter>\
7576 </function></tool_call>";
7577
7578 let mut deltas = Vec::new();
7579 let mut text = String::new();
7580 for piece in wire.as_bytes().chunks(7) {
7581 let chunk = String::from_utf8_lossy(piece).into_owned();
7582 let (more_text, more) = tool_call_deltas(parser.push(&chunk), &opened);
7583 text.push_str(&more_text);
7584 deltas.extend(more);
7585 }
7586 let (more_text, more) = tool_call_deltas(parser.finish(), &opened);
7587 text.push_str(&more_text);
7588 deltas.extend(more);
7589
7590 assert_eq!(opened.get(), 1, "one call opened");
7591 assert!(text.is_empty(), "the markers are not content: {text:?}");
7592
7593 let first = &deltas[0];
7594 assert_eq!(first.index, 0);
7595 assert_eq!(first.id.as_deref(), Some("call_0"));
7596 assert_eq!(first.kind, Some("function"));
7597 assert_eq!(first.function.name.as_deref(), Some("write_file"));
7598
7599 let joined: String = deltas
7602 .iter()
7603 .filter_map(|d| d.function.arguments.clone())
7604 .collect();
7605 let parsed: serde_json::Value =
7606 serde_json::from_str(&joined).expect("the fragments concatenate to valid JSON");
7607 assert_eq!(parsed["path"], serde_json::json!("/tmp/x"));
7608 assert_eq!(parsed["contents"], serde_json::json!("hello world"));
7609 assert!(
7610 deltas.len() >= 3,
7611 "the arguments arrived in pieces, not whole: {}",
7612 deltas.len()
7613 );
7614 assert!(
7615 deltas[1..].iter().all(|d| d.function.name.is_none()),
7616 "only the opening delta carries identity"
7617 );
7618 }
7619
7620 #[test]
7622 fn text_around_a_streamed_call_is_still_content() {
7623 let opened = std::cell::Cell::new(0usize);
7624 let mut parser = crate::policy::parser::ToolCallParser::new(
7625 crate::policy::parser::ToolCallFormat::Qwen25,
7626 vec![crate::policy::parser::tool_call::ToolSchema::new(
7627 "get_weather",
7628 )],
7629 );
7630 let wire = "let me check. <tool_call>{\"name\": \"get_weather\", \
7631 \"arguments\": {}}</tool_call> done";
7632 let mut text = String::new();
7633 for piece in wire.as_bytes().chunks(5) {
7634 let chunk = String::from_utf8_lossy(piece).into_owned();
7635 let (more, _) = tool_call_deltas(parser.push(&chunk), &opened);
7636 text.push_str(&more);
7637 }
7638 let (more, _) = tool_call_deltas(parser.finish(), &opened);
7639 text.push_str(&more);
7640
7641 assert_eq!(opened.get(), 1);
7642 assert!(text.starts_with("let me check. "), "{text:?}");
7643 assert!(text.ends_with(" done"), "{text:?}");
7644 assert!(!text.contains("<tool_call>"), "markers leaked: {text:?}");
7645 }
7646
7647 #[test]
7650 fn a_reasoning_block_is_split_out_of_the_answer() {
7651 let (message, finish) = build_response_message(
7652 "<think>weighing it up</think>The answer is 4.".to_string(),
7653 &[],
7654 output::OutputPosture::for_model("Qwen3-8B"),
7655 "stop",
7656 );
7657 assert_eq!(finish, "stop");
7658 assert_eq!(message.content.as_deref(), Some("The answer is 4."));
7659 assert_eq!(message.reasoning_content.as_deref(), Some("weighing it up"));
7660 }
7661
7662 #[test]
7665 fn a_non_reasoning_model_keeps_a_literal_marker_in_its_answer() {
7666 let (message, _) = build_response_message(
7667 "Use the <think> tag like this.".to_string(),
7668 &[],
7669 output::OutputPosture::for_model("llama-3.1-8b"),
7670 "stop",
7671 );
7672 assert_eq!(
7673 message.content.as_deref(),
7674 Some("Use the <think> tag like this.")
7675 );
7676 assert!(message.reasoning_content.is_none());
7677 }
7678
7679 #[tokio::test]
7687 async fn a_request_with_no_tools_or_session_behaves_exactly_as_before() {
7688 let app = test_app();
7689 let body = serde_json::json!({
7690 "model": "m",
7691 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7692 "max_tokens": 4,
7693 "temperature": 0,
7694 });
7695 let resp = post_json(&app, body).await;
7696 let message = &resp["choices"][0]["message"];
7697 assert!(message["content"].is_string());
7698 assert!(message.get("tool_calls").is_none());
7699 assert_eq!(resp["choices"][0]["finish_reason"], "length");
7700 assert_eq!(resp["usage"]["completion_tokens"], 4);
7701 assert_eq!(
7702 resp["usage"]["total_tokens"],
7703 resp["usage"]["prompt_tokens"].as_u64().unwrap() + 4
7704 );
7705 }
7706
7707 async fn get_json(app: &Router, uri: &str) -> (StatusCode, serde_json::Value) {
7708 use http_body_util::BodyExt;
7709 use tower::ServiceExt;
7710
7711 let response = app
7712 .clone()
7713 .oneshot(
7714 axum::http::Request::builder()
7715 .method("GET")
7716 .uri(uri)
7717 .body(axum::body::Body::empty())
7718 .unwrap(),
7719 )
7720 .await
7721 .unwrap();
7722 let status = response.status();
7723 let bytes = response.into_body().collect().await.unwrap().to_bytes();
7724 (status, serde_json::from_slice(&bytes).unwrap())
7725 }
7726
7727 #[tokio::test]
7728 async fn health_answers_a_capability_handshake_not_a_boolean() {
7729 let app = test_app();
7730 let (status, body) = get_json(&app, ferrox_api::routes::HEALTH).await;
7731 assert_eq!(status, StatusCode::OK);
7732
7733 let health: ferrox_api::HealthResponse = serde_json::from_value(body).unwrap();
7734 assert_eq!(health.state, ferrox_api::HealthState::Ready);
7735 assert!(health.pid > 0);
7736 assert!(health.server_time_unix_ms > 0);
7737 assert_eq!(health.last_request_age_seconds, None);
7740
7741 for id in [
7744 ferrox_api::health::capability::CPU,
7745 ferrox_api::health::capability::METAL,
7746 ferrox_api::health::capability::CUDA,
7747 ferrox_api::health::capability::REAL_WEIGHTS,
7748 ferrox_api::health::capability::CONTINUOUS_BATCHING,
7749 ] {
7750 let cap = health
7751 .capability(id)
7752 .unwrap_or_else(|| panic!("{id} missing"));
7753 assert!(!cap.reason.is_empty(), "{cap:?}");
7754 assert!(!cap.detail.is_empty(), "{cap:?}");
7755 }
7756 let weights = health
7760 .capability(ferrox_api::health::capability::REAL_WEIGHTS)
7761 .unwrap();
7762 assert!(!weights.available);
7763 assert_eq!(weights.reason, ferrox_api::health::reason::MODEL_NOT_LOADED);
7764 assert!(health.model.as_ref().unwrap().synthetic_weights);
7765 }
7766
7767 #[tokio::test]
7768 async fn health_vouches_for_liveness_after_a_request_has_been_served() {
7769 let app = test_app();
7770 let _ = post_json(
7771 &app,
7772 serde_json::json!({
7773 "model": "m",
7774 "messages": [{"role": "user", "content": "\u{1}"}],
7775 "max_tokens": 1,
7776 "temperature": 0,
7777 }),
7778 )
7779 .await;
7780 let (_status, body) = get_json(&app, ferrox_api::routes::HEALTH).await;
7781 let health: ferrox_api::HealthResponse = serde_json::from_value(body).unwrap();
7782 let age = health
7783 .last_request_age_seconds
7784 .expect("a served request is evidence of liveness");
7785 assert!((0.0..5.0).contains(&age), "implausible age {age}");
7786 }
7787
7788 async fn post_sse_chunks(app: &Router, body: serde_json::Value) -> Vec<serde_json::Value> {
7790 use http_body_util::BodyExt;
7791 use tower::ServiceExt;
7792
7793 let response = app
7794 .clone()
7795 .oneshot(
7796 axum::http::Request::builder()
7797 .method("POST")
7798 .uri("/v1/chat/completions")
7799 .header("content-type", "application/json")
7800 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
7801 .unwrap(),
7802 )
7803 .await
7804 .unwrap();
7805 let bytes = response.into_body().collect().await.unwrap().to_bytes();
7806 String::from_utf8(bytes.to_vec())
7807 .unwrap()
7808 .lines()
7809 .filter_map(|line| line.strip_prefix("data: "))
7810 .filter(|payload| *payload != "[DONE]")
7811 .map(|payload| serde_json::from_str(payload).unwrap())
7812 .collect()
7813 }
7814
7815 #[tokio::test]
7816 async fn a_stream_states_its_request_id_once_in_the_first_chunk() {
7817 let app = test_app();
7818 let chunks = post_sse_chunks(
7819 &app,
7820 serde_json::json!({
7821 "model": "m",
7822 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7823 "max_tokens": 4,
7824 "temperature": 0,
7825 "stream": true,
7826 }),
7827 )
7828 .await;
7829
7830 assert!(!chunks.is_empty());
7831 let request_id = chunks[0]["request_id"]
7832 .as_str()
7833 .expect("the first chunk names the request")
7834 .to_string();
7835 assert!(request_id.starts_with("chatcmpl-"), "{request_id}");
7836 for (i, chunk) in chunks.iter().enumerate().skip(1) {
7839 assert!(
7840 chunk.get("request_id").is_none(),
7841 "chunk {i} repeats request_id"
7842 );
7843 }
7844 for chunk in &chunks {
7847 assert_eq!(chunk["id"], serde_json::json!(request_id));
7848 }
7849
7850 let other = post_sse_chunks(
7851 &app,
7852 serde_json::json!({
7853 "model": "m",
7854 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7855 "max_tokens": 4,
7856 "temperature": 0,
7857 "stream": true,
7858 }),
7859 )
7860 .await;
7861 assert_ne!(
7862 other[0]["request_id"].as_str().unwrap(),
7863 request_id,
7864 "two concurrent chats must not share an id"
7865 );
7866 }
7867
7868 #[tokio::test]
7869 async fn a_non_streamed_response_names_the_same_request_id_as_its_completion_id() {
7870 let app = test_app();
7871 let resp = post_json(
7872 &app,
7873 serde_json::json!({
7874 "model": "m",
7875 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7876 "max_tokens": 2,
7877 "temperature": 0,
7878 }),
7879 )
7880 .await;
7881 assert_eq!(resp["id"], resp["request_id"]);
7882 assert!(resp["request_id"]
7883 .as_str()
7884 .unwrap()
7885 .starts_with("chatcmpl-"));
7886 }
7887
7888 #[tokio::test]
7891 async fn usage_carries_separate_prefill_and_decode_timings() {
7892 let app = test_app();
7893 let resp = post_json(
7894 &app,
7895 serde_json::json!({
7896 "model": "m",
7897 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7898 "max_tokens": 4,
7899 "temperature": 0,
7900 }),
7901 )
7902 .await;
7903 let usage = &resp["usage"];
7904 assert!(usage["prompt_eval_duration_ms"].is_number(), "{usage}");
7905 assert!(usage["generation_duration_ms"].is_number(), "{usage}");
7906 assert!(usage["time_to_first_token_ms"].is_number(), "{usage}");
7907 assert!(usage["predicted_per_second"].is_number(), "{usage}");
7908 assert!(usage.get("cached_tokens").is_none(), "{usage}");
7910 }
7911
7912 #[tokio::test]
7920 async fn a_tools_request_with_no_marker_in_the_output_falls_back_to_plain_content() {
7921 let app = test_app();
7922 let body = serde_json::json!({
7923 "model": "m",
7924 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7925 "max_tokens": 4,
7926 "temperature": 0,
7927 "tools": [weather_tool()],
7928 });
7929 let resp = post_json(&app, body).await;
7930 let message = &resp["choices"][0]["message"];
7931 assert!(
7932 message["content"].is_string(),
7933 "must fall back to plain content when no real tool-call marker is present: {resp:?}"
7934 );
7935 assert!(message.get("tool_calls").is_none());
7936 assert_eq!(resp["choices"][0]["finish_reason"], "length");
7940 }
7941
7942 #[tokio::test]
7946 async fn a_cache_hit_reports_the_original_finish_reason_and_usage() {
7947 let app = test_app();
7948 let body = serde_json::json!({
7949 "model": "m",
7950 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
7951 "max_tokens": 3,
7952 "temperature": 0,
7953 });
7954 let first = post_json(&app, body.clone()).await;
7955 assert_eq!(first["ferrox_cache"], "miss");
7956 let second = post_json(&app, body).await;
7957 assert_eq!(second["ferrox_cache"], "hit");
7958 assert_eq!(
7959 first["choices"][0]["message"]["content"],
7960 second["choices"][0]["message"]["content"]
7961 );
7962 assert_eq!(
7963 first["choices"][0]["finish_reason"],
7964 second["choices"][0]["finish_reason"]
7965 );
7966 assert_eq!(first["usage"], second["usage"]);
7967 assert_eq!(second["usage"]["completion_tokens"], 3);
7968 }
7969
7970 #[tokio::test]
7981 async fn a_grammar_request_is_not_answered_from_an_unconstrained_cache_entry() {
7982 let app = test_app();
7983 let plain = serde_json::json!({
7984 "model": "m",
7985 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
7986 "max_tokens": 3,
7987 "temperature": 0,
7988 });
7989
7990 let first = post_json(&app, plain.clone()).await;
7991 assert_eq!(first["ferrox_cache"], "miss");
7992 let unconstrained = first["choices"][0]["message"]["content"]
7993 .as_str()
7994 .expect("content")
7995 .to_string();
7996
7997 let mut constrained = plain.clone();
7998 constrained["grammar"] = serde_json::json!("root ::= \"yes\"");
7999 let second = post_json(&app, constrained).await;
8000 assert_eq!(
8001 second["ferrox_cache"], "miss",
8002 "a grammar is part of the key, so this body has never been answered"
8003 );
8004 let constrained_answer = second["choices"][0]["message"]["content"]
8008 .as_str()
8009 .expect("content")
8010 .to_string();
8011 assert!(
8012 constrained_answer.contains("-> \"yes\"]"),
8013 "the grammar must have been compiled AND applied, not skipped \
8014 by a cache hit: {constrained_answer}"
8015 );
8016 assert_ne!(
8017 constrained_answer, unconstrained,
8018 "the constrained request was served the unconstrained answer"
8019 );
8020
8021 let third = post_json(&app, plain).await;
8025 assert_eq!(third["ferrox_cache"], "hit");
8026 assert_eq!(third["choices"][0]["message"]["content"], unconstrained);
8027 }
8028
8029 #[tokio::test]
8051 async fn a_json_object_request_does_not_reuse_the_unconstrained_cache_entry() {
8052 let state = Arc::new(test_state(
8053 test_model_full_byte_vocab(),
8054 ResponseCache::new(1000, Duration::from_secs(3600)),
8055 ));
8056 let app = test_app_with_state(state.clone());
8057 let plain = serde_json::json!({
8058 "model": "m",
8059 "messages": [
8060 {"role": "system", "content": "Answer in JSON when it helps."},
8061 {"role": "user", "content": "\u{1}\u{2}"},
8062 ],
8063 "max_tokens": 3,
8064 "temperature": 0,
8065 });
8066
8067 let first = post_json(&app, plain.clone()).await;
8068 assert_eq!(first["ferrox_cache"], "miss");
8069 assert_eq!(state.cache_stats().entries, 1);
8070
8071 let mut as_json = plain.clone();
8072 as_json["response_format"] = serde_json::json!({"type": "json_object"});
8073 let (status, _) = post_json_uri(&app, "/v1/chat/completions", as_json).await;
8074 assert_eq!(
8075 status,
8076 StatusCode::BAD_REQUEST,
8077 "the demo banner is not a JSON object, whoever generated it"
8078 );
8079 assert_eq!(
8080 state.cache_stats().hits,
8081 0,
8082 "a json_object request must not be answered from an entry the \
8083 JSON mask never produced"
8084 );
8085 assert_eq!(
8086 state.cache_stats().entries,
8087 2,
8088 "json_object must key its own entry, not reuse the unconstrained \
8089 one it happens to render the same prompt as"
8090 );
8091 }
8092
8093 #[tokio::test]
8105 async fn an_ignore_eos_request_is_not_answered_from_a_cache_entry_that_stopped_at_eos() {
8106 let app = test_app_with_state(Arc::new(test_state(
8107 test_model_full_byte_vocab_with_eos(Some(0x77)),
8108 ResponseCache::new(1000, Duration::from_secs(3600)),
8109 )));
8110 let body = serde_json::json!({
8111 "model": "m",
8112 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
8113 "max_tokens": 6,
8114 "temperature": 0,
8115 });
8116
8117 let stopped = post_json(&app, body.clone()).await;
8118 assert_eq!(stopped["ferrox_cache"], "miss");
8119 assert_eq!(
8120 stopped["choices"][0]["finish_reason"], "stop",
8121 "the fixture is only meaningful if the model's EOS really fires here"
8122 );
8123 assert_eq!(stopped["usage"]["completion_tokens"], 1);
8124
8125 let mut ignoring = body.clone();
8126 ignoring["ignore_eos"] = serde_json::json!(true);
8127 let ran_on = post_json(&app, ignoring).await;
8128 assert_eq!(
8129 ran_on["ferrox_cache"], "miss",
8130 "ignore_eos is part of the key, so this body has never been answered"
8131 );
8132 assert_eq!(
8133 ran_on["usage"]["completion_tokens"], 6,
8134 "ignore_eos must run the full budget, not replay the EOS-terminated answer"
8135 );
8136 assert_eq!(ran_on["choices"][0]["finish_reason"], "length");
8137 assert_ne!(
8138 ran_on["choices"][0]["message"]["content"],
8139 stopped["choices"][0]["message"]["content"]
8140 );
8141 }
8142
8143 #[tokio::test]
8149 async fn session_reuse_produces_the_same_output_as_manually_resending_full_history() {
8150 let session_app = test_app();
8151 let manual_app = test_app();
8152
8153 let turn1 = post_json(
8155 &session_app,
8156 serde_json::json!({
8157 "model": "m",
8158 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
8159 "session_id": "s1",
8160 "max_tokens": 5,
8161 "temperature": 0,
8162 }),
8163 )
8164 .await;
8165 let reply1 = turn1["choices"][0]["message"]["content"]
8166 .as_str()
8167 .unwrap()
8168 .to_string();
8169
8170 let manual_turn1 = post_json(
8175 &manual_app,
8176 serde_json::json!({
8177 "model": "m",
8178 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
8179 "max_tokens": 5,
8180 "temperature": 0,
8181 }),
8182 )
8183 .await;
8184 assert_eq!(
8185 manual_turn1["choices"][0]["message"]["content"]
8186 .as_str()
8187 .unwrap(),
8188 reply1
8189 );
8190
8191 let turn2 = post_json(
8193 &session_app,
8194 serde_json::json!({
8195 "model": "m",
8196 "messages": [{"role": "user", "content": "\u{4}\u{5}"}],
8197 "session_id": "s1",
8198 "max_tokens": 5,
8199 "temperature": 0,
8200 }),
8201 )
8202 .await;
8203 let reply2 = turn2["choices"][0]["message"]["content"]
8204 .as_str()
8205 .unwrap()
8206 .to_string();
8207
8208 let manual_turn2 = post_json(
8212 &manual_app,
8213 serde_json::json!({
8214 "model": "m",
8215 "messages": [
8216 {"role": "user", "content": "\u{1}\u{2}\u{3}"},
8217 {"role": "assistant", "content": reply1},
8218 {"role": "user", "content": "\u{4}\u{5}"},
8219 ],
8220 "max_tokens": 5,
8221 "temperature": 0,
8222 }),
8223 )
8224 .await;
8225 assert_eq!(
8226 manual_turn2["choices"][0]["message"]["content"]
8227 .as_str()
8228 .unwrap(),
8229 reply2,
8230 "resuming a session must produce identical output to manually resending the full history"
8231 );
8232 }
8233
8234 #[test]
8237 fn lock_cache_recovers_from_a_poisoned_mutex() {
8238 let cache = Arc::new(Mutex::new(ResponseCache::new(10, Duration::from_secs(60))));
8239
8240 let poison_cache = Arc::clone(&cache);
8241 let _ = std::thread::spawn(move || {
8242 let _guard = poison_cache.lock().unwrap();
8243 panic!("simulated panic while holding the lock");
8244 })
8245 .join();
8246
8247 let recovered = lock_cache(&cache);
8249 assert_eq!(recovered.stats().entries, 0);
8250 }
8251
8252 #[test]
8253 fn is_cacheable_true_for_greedy_or_seeded_requests() {
8254 let mut req_body = serde_json::json!({
8255 "model": "m",
8256 "messages": [{"role": "user", "content": "hi"}],
8257 });
8258 let req: ChatCompletionRequest = serde_json::from_value(req_body.clone()).unwrap();
8259 assert!(
8260 req.is_cacheable(),
8261 "default (temperature 0) must be cacheable"
8262 );
8263
8264 req_body["temperature"] = serde_json::json!(0.8);
8265 let req: ChatCompletionRequest = serde_json::from_value(req_body.clone()).unwrap();
8266 assert!(
8267 !req.is_cacheable(),
8268 "unseeded sampling must never be cacheable"
8269 );
8270
8271 req_body["seed"] = serde_json::json!(42);
8272 let req: ChatCompletionRequest = serde_json::from_value(req_body).unwrap();
8273 assert!(
8274 req.is_cacheable(),
8275 "sampling with an explicit seed is deterministic and must be cacheable"
8276 );
8277 }
8278
8279 const GRADED: &str = "{% if reasoning_effort %}\
8283 {% if reasoning_effort not in ['low','medium','high'] %}\
8284 {{ raise_exception('unsupported effort') }}\
8285 {% endif %}E:{{ reasoning_effort }}|{% endif %}\
8286 {% if enable_thinking %}THINK|{% endif %}{{ messages[0].content }}";
8287
8288 fn graded_template() -> chat_template::PromptTemplate {
8289 chat_template::PromptTemplate::from_gguf_metadata(
8290 Some(GRADED),
8291 Some("qwen3"),
8292 false,
8293 true,
8294 None,
8295 None,
8296 )
8297 }
8298
8299 fn chat_request(value: serde_json::Value) -> ChatCompletionRequest {
8300 serde_json::from_value(value).expect("request")
8301 }
8302
8303 #[test]
8310 fn the_reasoning_split_is_resolved_from_the_served_model_not_the_request() {
8311 let req = chat_request(serde_json::json!({
8312 "model": "some-alias",
8315 "messages": [{"role": "user", "content": "hi"}],
8316 }));
8317 let template = chat_template::PromptTemplate::plain();
8318
8319 let thinks = req
8320 .generation_params_for_template(
8321 &template,
8322 "Qwen3-8B",
8323 crate::sampling_knobs::SamplerModel::absent(),
8324 )
8325 .expect("params");
8326 assert!(
8327 thinks.reasoning.is_some(),
8328 "a thinking checkpoint must carry its format into generation"
8329 );
8330
8331 let plain = req
8332 .generation_params_for_template(
8333 &template,
8334 "Llama-3.2-1B-Instruct",
8335 crate::sampling_knobs::SamplerModel::absent(),
8336 )
8337 .expect("params");
8338 assert!(
8339 plain.reasoning.is_none(),
8340 "a checkpoint with no reasoning format must carry none, so the \
8341 usage field stays absent rather than becoming a zero"
8342 );
8343 }
8344
8345 #[test]
8352 fn a_grammar_on_the_chat_wire_reaches_the_generation_params() {
8353 let req = chat_request(serde_json::json!({
8354 "model": "m",
8355 "messages": [{"role": "user", "content": "hi"}],
8356 "grammar": "root ::= \"a\"+",
8357 }));
8358 req.validate_supported_fields()
8359 .expect("a valid grammar is a valid request");
8360 let params = req
8361 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8362 .expect("a valid grammar compiles at params time too");
8363 assert!(
8364 params.grammar.is_some(),
8365 "the grammar was dropped between the wire and the sampler"
8366 );
8367 assert!(
8368 params.needs_vocab_logits(),
8369 "a grammar request that may fold lm_head into a GPU argmax is \
8370 a grammar request served unconstrained"
8371 );
8372
8373 let plain = chat_request(serde_json::json!({
8374 "model": "m",
8375 "messages": [{"role": "user", "content": "hi"}],
8376 }));
8377 assert!(plain
8378 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8379 .unwrap()
8380 .grammar
8381 .is_none());
8382 }
8383
8384 fn tool_request(tool_choice: serde_json::Value) -> ChatCompletionRequest {
8385 chat_request(serde_json::json!({
8386 "model": "m",
8387 "messages": [{"role": "user", "content": "weather in Rome?"}],
8388 "tools": [weather_tool()],
8389 "tool_choice": tool_choice,
8390 }))
8391 }
8392
8393 #[test]
8397 fn a_forced_tool_choice_puts_a_grammar_on_the_generation_params() {
8398 for choice in [
8399 serde_json::json!("required"),
8400 serde_json::json!({"type": "function", "function": {"name": "get_weather"}}),
8401 ] {
8402 let req = tool_request(choice.clone());
8403 req.validate_supported_fields()
8404 .unwrap_or_else(|e| panic!("{choice} is a valid request: {e:?}"));
8405 let params = req
8406 .generation_params_for_template(
8407 &graded_template(),
8408 "Qwen3-8B",
8409 crate::sampling_knobs::SamplerModel::absent(),
8410 )
8411 .unwrap_or_else(|e| panic!("{choice} compiles: {e:?}"));
8412 let grammar = params
8413 .grammar
8414 .as_ref()
8415 .unwrap_or_else(|| panic!("{choice} was accepted and then not enforced"));
8416 assert!(
8417 grammar.is_awaiting_trigger(),
8418 "the model must be free to think before it calls"
8419 );
8420 assert!(
8421 !grammar.allows_eog(),
8422 "{choice} must not be able to end the turn without a call"
8423 );
8424 assert!(
8430 params.needs_vocab_logits(),
8431 "{choice} would let a backend return a token id instead of logits"
8432 );
8433 assert!(
8434 !generate::greedy_gpu_fold_allowed(¶ms),
8435 "{choice} at temperature 0 must still refuse the greedy GPU fold"
8436 );
8437 }
8438 }
8439
8440 #[test]
8442 fn an_unforced_tool_choice_leaves_the_generation_unconstrained() {
8443 for choice in [serde_json::json!("auto"), serde_json::json!("none")] {
8444 let req = tool_request(choice.clone());
8445 req.validate_supported_fields().expect("still supported");
8446 let params = match req.generation_params_for_template(
8447 &graded_template(),
8448 "Qwen3-8B",
8449 crate::sampling_knobs::SamplerModel::absent(),
8450 ) {
8451 Ok(p) => p,
8452 Err((status, _)) => panic!("{choice} has no constraint to compile: {status}"),
8453 };
8454 assert!(
8455 params.grammar.is_none(),
8456 "{choice} does not force a call and must not be constrained"
8457 );
8458 }
8459 }
8460
8461 #[test]
8464 fn a_forced_tool_choice_refuses_rather_than_quietly_not_forcing() {
8465 let req = chat_request(serde_json::json!({
8467 "model": "m",
8468 "messages": [{"role": "user", "content": "hi"}],
8469 "tool_choice": "required",
8470 }));
8471 let (status, _) = req
8472 .validate_supported_fields()
8473 .expect_err("nothing to call");
8474 assert_eq!(status, StatusCode::BAD_REQUEST);
8475
8476 let req =
8478 tool_request(serde_json::json!({"type": "function", "function": {"name": "nope"}}));
8479 let (status, Json(body)) = req.validate_supported_fields().expect_err("no such tool");
8480 assert_eq!(status, StatusCode::BAD_REQUEST);
8481 assert_eq!(body["error"]["param"], "tool_choice");
8482
8483 let req = tool_request(serde_json::json!({"type": "function"}));
8485 let (status, _) = req.validate_supported_fields().expect_err("names nothing");
8486 assert_eq!(status, StatusCode::BAD_REQUEST);
8487
8488 let req = chat_request(serde_json::json!({
8490 "model": "m",
8491 "messages": [{"role": "user", "content": "hi"}],
8492 "tools": [weather_tool()],
8493 "tool_choice": "required",
8494 "grammar": "root ::= \"a\"+",
8495 }));
8496 let (status, _) = req
8497 .validate_supported_fields()
8498 .expect_err("a grammar and a forced call are two constraints");
8499 assert_eq!(status, StatusCode::BAD_REQUEST);
8500
8501 let req = tool_request(serde_json::json!("required"));
8507 let (status, Json(body)) = match req.generation_params_for_template(
8508 &graded_template(),
8509 "Gemma4-27B",
8510 crate::sampling_knobs::SamplerModel::absent(),
8511 ) {
8512 Err(e) => e,
8513 Ok(_) => panic!("a gemma4 call's arguments are not an object rule"),
8514 };
8515 assert_eq!(status, StatusCode::NOT_IMPLEMENTED);
8516 assert!(
8517 body["error"]["message"]
8518 .as_str()
8519 .unwrap()
8520 .contains("gemma4"),
8521 "{body}"
8522 );
8523 }
8524
8525 #[test]
8528 fn an_unparseable_grammar_on_the_chat_wire_is_a_400() {
8529 let req = chat_request(serde_json::json!({
8530 "model": "m",
8531 "messages": [{"role": "user", "content": "hi"}],
8532 "grammar": "root ::= \"a",
8533 }));
8534 let (status, Json(body)) = req
8535 .validate_supported_fields()
8536 .expect_err("this does not parse");
8537 assert_eq!(status, StatusCode::BAD_REQUEST);
8538 assert_eq!(body["error"]["param"], "grammar");
8539 assert!(
8540 req.generation_params(crate::sampling_knobs::SamplerModel::absent())
8541 .is_err(),
8542 "and again at params time"
8543 );
8544 }
8545
8546 #[test]
8553 fn response_format_json_schema_becomes_the_requests_grammar() {
8554 let req = chat_request(serde_json::json!({
8555 "model": "m",
8556 "messages": [{"role": "user", "content": "hi"}],
8557 "response_format": {
8558 "type": "json_schema",
8559 "json_schema": {"name": "x", "schema": {"type": "boolean"}},
8560 },
8561 }));
8562 req.validate_supported_fields()
8563 .expect("a boolean schema converts");
8564 let params = req
8565 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8566 .expect("and compiles");
8567 let grammar = params.grammar.expect("the schema is the grammar");
8568 let mut g = (*grammar).clone();
8569 g.accept_token(0, b"true").expect("a boolean is accepted");
8570 assert!(g.allows_eog(), "and completes the parse");
8571 assert!(
8572 !params.json_object,
8573 "a schema is not the json_object character-class mask"
8574 );
8575 }
8576
8577 #[test]
8581 fn an_unconvertible_response_format_schema_is_a_400_naming_the_keyword() {
8582 let req = chat_request(serde_json::json!({
8583 "model": "m",
8584 "messages": [{"role": "user", "content": "hi"}],
8585 "response_format": {
8586 "type": "json_schema",
8587 "json_schema": {"name": "x", "schema": {"type": "integer", "minimum": 3}},
8588 },
8589 }));
8590 let (status, Json(body)) = req
8591 .validate_supported_fields()
8592 .expect_err("minimum has no grammar in this port");
8593 assert_eq!(status, StatusCode::BAD_REQUEST);
8594 assert!(
8595 body["error"]["message"]
8596 .as_str()
8597 .expect("a message")
8598 .contains("minimum"),
8599 "the refusal must name the keyword: {body}"
8600 );
8601 assert!(
8602 req.generation_params(crate::sampling_knobs::SamplerModel::absent())
8603 .is_err(),
8604 "and again at params time"
8605 );
8606 }
8607
8608 #[test]
8614 fn a_forced_tool_choice_and_a_schema_are_two_constraints() {
8615 let req = chat_request(serde_json::json!({
8616 "model": "m",
8617 "messages": [{"role": "user", "content": "hi"}],
8618 "tool_choice": "required",
8619 "tools": [{
8620 "type": "function",
8621 "function": {"name": "f", "parameters": {"type": "object"}},
8622 }],
8623 "response_format": {
8624 "type": "json_schema",
8625 "json_schema": {"name": "x", "schema": {"type": "boolean"}},
8626 },
8627 }));
8628 let (status, Json(body)) = req
8629 .validate_supported_fields()
8630 .expect_err("two constraints, one generation");
8631 assert_eq!(status, StatusCode::BAD_REQUEST);
8632 assert_eq!(body["error"]["param"], "tool_choice");
8633 }
8634
8635 #[test]
8638 fn an_omitted_output_budget_is_a_whole_answer_not_sixteen_tokens() {
8639 let req = chat_request(serde_json::json!({
8640 "model": "m",
8641 "messages": [{"role": "user", "content": "hi"}],
8642 }));
8643 assert_eq!(req.max_tokens, DEFAULT_CHAT_MAX_TOKENS);
8644 }
8645
8646 #[test]
8651 fn min_p_reaches_the_sampler_from_the_chat_wire() {
8652 let asked = chat_request(serde_json::json!({
8653 "model": "m",
8654 "messages": [{"role": "user", "content": "hi"}],
8655 "min_p": 0.07,
8656 }));
8657 assert_eq!(
8658 asked
8659 .sampling_params(crate::sampling_knobs::SamplerModel::absent())
8660 .expect("knobs")
8661 .min_p,
8662 0.07
8663 );
8664
8665 let silent = chat_request(serde_json::json!({
8666 "model": "m",
8667 "messages": [{"role": "user", "content": "hi"}],
8668 }));
8669 assert_eq!(
8670 silent
8671 .sampling_params(crate::sampling_knobs::SamplerModel::absent())
8672 .expect("knobs")
8673 .min_p,
8674 0.0,
8675 "an unset min_p must be off, not llama.cpp's CLI default"
8676 );
8677 }
8678
8679 #[test]
8688 fn no_sampler_knob_is_missing_from_the_cache_key() {
8689 let base = serde_json::json!({
8690 "model": "m",
8691 "messages": [{"role": "user", "content": "hi"}],
8692 "seed": 1,
8693 });
8694 let key_for = |body: serde_json::Value| {
8695 let req = chat_request(body);
8696 let params = req
8697 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8698 .expect("params");
8699 req.cache_key("prompt", ¶ms)
8700 };
8701 let baseline = key_for(base.clone());
8702 for (knob, value) in [
8703 ("temperature", serde_json::json!(0.5)),
8704 ("top_p", serde_json::json!(0.9)),
8705 ("min_p", serde_json::json!(0.05)),
8706 ("top_k", serde_json::json!(40)),
8707 ("repetition_penalty", serde_json::json!(1.1)),
8708 ("presence_penalty", serde_json::json!(0.3)),
8709 ("frequency_penalty", serde_json::json!(0.3)),
8710 (
8711 "samplers",
8712 serde_json::json!(["penalties", "top_p", "top_k", "min_p", "temperature"]),
8713 ),
8714 ] {
8715 let mut body = base.clone();
8716 body[knob] = value;
8717 assert_ne!(
8718 key_for(body),
8719 baseline,
8720 "`{knob}` is not in the cache key: two requests differing \
8721 only in it would share one cached answer"
8722 );
8723 }
8724 }
8725
8726 #[test]
8736 fn no_constraint_is_missing_from_the_cache_key() {
8737 let base = serde_json::json!({
8738 "model": "m",
8739 "messages": [{"role": "user", "content": "pick one"}],
8740 });
8741 let key_for = |body: serde_json::Value| {
8742 let req = chat_request(body);
8743 let params = req
8744 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8745 .expect("params");
8746 req.cache_key("prompt", ¶ms)
8747 };
8748 let baseline = key_for(base.clone());
8749 for (field, value) in [
8750 ("grammar", serde_json::json!("root ::= \"yes\" | \"no\"")),
8751 (
8752 "response_format",
8753 serde_json::json!({"type": "json_object"}),
8754 ),
8755 (
8756 "response_format",
8757 serde_json::json!({"type": "json_schema", "json_schema": {
8758 "name": "answer",
8759 "schema": {"type": "object", "properties": {"a": {"type": "string"}}}
8760 }}),
8761 ),
8762 ("ignore_eos", serde_json::json!(true)),
8763 ("stop", serde_json::json!(["\n"])),
8764 ("max_tokens", serde_json::json!(7)),
8765 ] {
8766 let mut body = base.clone();
8767 body[field] = value.clone();
8768 assert_ne!(
8769 key_for(body),
8770 baseline,
8771 "`{field}: {value}` is not in the cache key: two requests \
8772 differing only in it would share one cached answer"
8773 );
8774 }
8775 }
8776
8777 #[test]
8781 fn an_explicit_zero_output_budget_is_a_client_error() {
8782 let req = chat_request(serde_json::json!({
8783 "model": "m",
8784 "messages": [{"role": "user", "content": "hi"}],
8785 "max_tokens": 0,
8786 }));
8787 let (status, body) = req.validate_supported_fields().expect_err("rejected");
8788 assert_eq!(status, StatusCode::BAD_REQUEST);
8789 assert_eq!(body["error"]["param"], serde_json::json!("max_tokens"));
8790 }
8791
8792 #[test]
8795 fn a_request_can_turn_thinking_off() {
8796 let template = graded_template();
8797 for body in [
8798 serde_json::json!({
8799 "model": "m",
8800 "messages": [{"role": "user", "content": "hi"}],
8801 "reasoning_effort": "none",
8802 }),
8803 serde_json::json!({
8804 "model": "m",
8805 "messages": [{"role": "user", "content": "hi"}],
8806 "thinking": {"type": "disabled"},
8807 }),
8808 ] {
8809 let kwargs = chat_request(body).resolve_template_kwargs(&template);
8810 assert_eq!(kwargs["enable_thinking"], serde_json::json!(false));
8811 assert_eq!(kwargs["thinking_mode"], serde_json::json!("disabled"));
8812 assert!(!kwargs.contains_key("reasoning_effort"));
8815 }
8816 }
8817
8818 #[test]
8821 fn a_disabled_switch_beats_an_effort_in_the_same_request() {
8822 let template = graded_template();
8823 let kwargs = chat_request(serde_json::json!({
8824 "model": "m",
8825 "messages": [{"role": "user", "content": "hi"}],
8826 "reasoning_effort": "high",
8827 "thinking": {"type": "disabled"},
8828 }))
8829 .resolve_template_kwargs(&template);
8830 assert_eq!(kwargs["enable_thinking"], serde_json::json!(false));
8831 assert!(!kwargs.contains_key("reasoning_effort"));
8832 }
8833
8834 #[test]
8837 fn an_unrecognized_thinking_switch_is_refused_rather_than_read_as_on() {
8838 let req = chat_request(serde_json::json!({
8839 "model": "m",
8840 "messages": [{"role": "user", "content": "hi"}],
8841 "thinking": {"type": "disable"},
8842 }));
8843 let (status, _) = req.validate_supported_fields().expect_err("rejected");
8844 assert_eq!(status, StatusCode::BAD_REQUEST);
8845 }
8846
8847 #[test]
8850 fn an_explicit_template_kwarg_stands_the_protocol_knobs_down() {
8851 let template = graded_template();
8852 let kwargs = chat_request(serde_json::json!({
8853 "model": "m",
8854 "messages": [{"role": "user", "content": "hi"}],
8855 "reasoning_effort": "none",
8856 "chat_template_kwargs": {"enable_thinking": true},
8857 }))
8858 .resolve_template_kwargs(&template);
8859 assert_eq!(kwargs["enable_thinking"], serde_json::json!(true));
8860 }
8861
8862 #[test]
8866 fn an_off_vocabulary_reasoning_effort_is_quantized_rather_than_interpolated() {
8867 let template = graded_template();
8868 let req = chat_request(serde_json::json!({
8869 "model": "m",
8870 "messages": [{"role": "user", "content": "hi"}],
8871 "reasoning_effort": "minimal",
8872 }));
8873 let kwargs = req.resolve_template_kwargs(&template);
8874 assert_eq!(kwargs["reasoning_effort"], serde_json::json!("low"));
8875 let prompt = prompt_from_messages(&req.messages, &template, &[], kwargs).expect("renders");
8876 assert!(prompt.starts_with("E:low|"), "{prompt}");
8877 }
8878
8879 #[test]
8883 fn an_effort_with_no_near_gear_is_dropped_so_the_template_default_applies() {
8884 let template = graded_template();
8885 let req = chat_request(serde_json::json!({
8886 "model": "m",
8887 "messages": [{"role": "user", "content": "hi"}],
8888 "chat_template_kwargs": {"reasoning_effort": "none"},
8889 }));
8890 let kwargs = req.resolve_template_kwargs(&template);
8891 assert!(!kwargs.contains_key("reasoning_effort"));
8892 let prompt = prompt_from_messages(&req.messages, &template, &[], kwargs).expect("renders");
8893 assert_eq!(prompt, "hi");
8894 }
8895
8896 #[test]
8899 fn chat_template_kwargs_wins_over_the_top_level_reasoning_effort() {
8900 let template = graded_template();
8901 let req = chat_request(serde_json::json!({
8902 "model": "m",
8903 "messages": [{"role": "user", "content": "hi"}],
8904 "reasoning_effort": "low",
8905 "chat_template_kwargs": {"reasoning_effort": "high"},
8906 }));
8907 assert_eq!(
8908 req.resolve_template_kwargs(&template)["reasoning_effort"],
8909 serde_json::json!("high")
8910 );
8911 }
8912
8913 #[test]
8917 fn offering_tools_turns_thinking_on_by_itself() {
8918 let template = graded_template();
8919 let quiet = chat_request(serde_json::json!({
8920 "model": "m",
8921 "messages": [{"role": "user", "content": "hi"}],
8922 }));
8923 assert!(!quiet
8924 .resolve_template_kwargs(&template)
8925 .contains_key("enable_thinking"));
8926
8927 let with_tools = chat_request(serde_json::json!({
8928 "model": "m",
8929 "messages": [{"role": "user", "content": "hi"}],
8930 "tools": [{"type": "function", "function": {"name": "get_weather"}}],
8931 }));
8932 let kwargs = with_tools.resolve_template_kwargs(&template);
8933 assert_eq!(kwargs["enable_thinking"], serde_json::json!(true));
8934 let prompt =
8935 prompt_from_messages(&with_tools.messages, &template, &[], kwargs).expect("renders");
8936 assert!(prompt.starts_with("THINK|"), "{prompt}");
8937 }
8938
8939 #[test]
8944 fn a_prompt_that_opens_the_reasoning_block_makes_the_first_token_reasoning() {
8945 let opener = chat_template::PromptTemplate::from_gguf_metadata(
8946 Some("{{ messages[0].content }}{% if enable_thinking %}<think>{% endif %}"),
8947 Some("qwen3"),
8948 false,
8949 true,
8950 None,
8951 None,
8952 );
8953 let req = chat_request(serde_json::json!({
8954 "model": "m",
8955 "messages": [{"role": "user", "content": "hi"}],
8956 "chat_template_kwargs": {"enable_thinking": true},
8957 }));
8958 let kwargs = req.resolve_template_kwargs(&opener);
8959 let prompt = prompt_from_messages(&req.messages, &opener, &[], kwargs).expect("renders");
8960 assert!(prompt.ends_with("<think>"), "{prompt}");
8961
8962 let posture = output::OutputPosture::resolve("Qwen3-8B", &prompt);
8965 let (message, _) = build_response_message(
8966 "weighing it up</think>Paris.".to_string(),
8967 &[],
8968 posture,
8969 "stop",
8970 );
8971 assert_eq!(message.reasoning_content.as_deref(), Some("weighing it up"));
8972 assert_eq!(message.content.as_deref(), Some("Paris."));
8973
8974 let closed = output::OutputPosture::resolve("Qwen3-8B", "<|im_start|>assistant\n");
8977 let (message, _) = build_response_message(
8978 "weighing it up</think>Paris.".to_string(),
8979 &[],
8980 closed,
8981 "stop",
8982 );
8983 assert_eq!(message.reasoning_content, None);
8984 }
8985
8986 #[test]
8987 fn stop_param_accepts_both_single_string_and_array() {
8988 let req: ChatCompletionRequest = serde_json::from_value(serde_json::json!({
8989 "model": "m",
8990 "messages": [{"role": "user", "content": "hi"}],
8991 "stop": "END",
8992 }))
8993 .unwrap();
8994 assert_eq!(req.stop_sequences(), vec!["END".to_string()]);
8995
8996 let req: ChatCompletionRequest = serde_json::from_value(serde_json::json!({
8997 "model": "m",
8998 "messages": [{"role": "user", "content": "hi"}],
8999 "stop": ["A", "B"],
9000 }))
9001 .unwrap();
9002 assert_eq!(req.stop_sequences(), vec!["A".to_string(), "B".to_string()]);
9003 }
9004
9005 #[test]
9006 fn run_generation_rejects_out_of_vocab_tokens_instead_of_panicking() {
9007 let model = test_model();
9008 let result = run_generation(
9009 &model,
9010 "hello",
9011 &greedy_params(4),
9012 None,
9013 None,
9014 None,
9015 None,
9016 None,
9017 None,
9018 );
9019 assert!(matches!(
9020 result,
9021 Err(generate::DecodeError::TokenOutOfVocab { .. })
9022 ));
9023 }
9024
9025 #[test]
9029 fn run_generation_honors_an_exhausted_kv_pool_and_maps_it_to_a_503() {
9030 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9032 let pool = Arc::new(Mutex::new(ferrox_core::cache::KvBlockPool::new(64, 2)));
9033
9034 let holder_pool = Arc::clone(&pool);
9035 let holder = std::thread::spawn(move || {
9036 let mut held = ferrox_core::cache::KvCache::with_pool(1, 1, holder_pool, 0).unwrap();
9037 held.push(&[0.0], &[0.0]).unwrap(); std::thread::sleep(Duration::from_millis(200));
9039 drop(held);
9040 });
9041 std::thread::sleep(Duration::from_millis(15));
9042
9043 let config = generate::KvPoolConfig {
9044 pool,
9045 queue_wait: Duration::ZERO,
9046 };
9047 let result = run_generation(
9048 &model,
9049 &prompt,
9050 &greedy_params(4),
9051 Some(&config),
9052 None,
9053 None,
9054 None,
9055 None,
9056 None,
9057 );
9058 assert!(matches!(
9059 result,
9060 Err(generate::DecodeError::KvPoolExhausted)
9061 ));
9062
9063 let (status, _body) = decode_error_response(result.unwrap_err());
9064 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
9065 holder.join().unwrap();
9066 }
9067
9068 #[test]
9076 fn a_request_too_big_for_the_whole_pool_is_a_400_not_a_retryable_503() {
9077 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9079 let pool = Arc::new(Mutex::new(ferrox_core::cache::KvBlockPool::new(64, 1)));
9081 let config = generate::KvPoolConfig {
9082 pool,
9083 queue_wait: Duration::ZERO,
9084 };
9085
9086 let result = run_generation(
9087 &model,
9088 &prompt,
9089 &greedy_params(4),
9090 Some(&config),
9091 None,
9092 None,
9093 None,
9094 None,
9095 None,
9096 );
9097 let err = result.expect_err("one block cannot hold two layers' caches");
9098 assert!(
9099 matches!(
9100 &err,
9101 generate::DecodeError::KvBudgetExceeded { binding, .. }
9102 if *binding == ferrox_models::Ceiling::DeviceMemory.code()
9103 ),
9104 "expected an immovable device-memory refusal, got {err:?}"
9105 );
9106 let (status, _body) = decode_error_response(err);
9107 assert_eq!(status, StatusCode::BAD_REQUEST);
9108 }
9109
9110 #[test]
9116 fn decode_error_response_maps_a_full_queue_to_a_retryable_503() {
9117 let (status, Json(body)) = decode_error_response(generate::DecodeError::QueueFull {
9118 queued: 512,
9119 cap: 512,
9120 });
9121 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
9122 assert_eq!(body["error"]["retry_after_seconds"], 1);
9123 let message = body["error"]["message"].as_str().expect("message");
9124 assert!(message.contains("512"), "{message}");
9125 }
9126
9127 #[test]
9128 fn decode_error_response_omits_a_retry_hint_for_an_unretryable_error() {
9129 let (_status, Json(body)) = decode_error_response(generate::DecodeError::TokenOutOfVocab {
9130 token: 99,
9131 vocab_size: 32,
9132 });
9133 assert!(
9134 body["error"]["retry_after_seconds"].is_null(),
9135 "retrying a prompt this model cannot tokenize never helps"
9136 );
9137 }
9138
9139 #[test]
9140 fn decode_error_response_maps_token_out_of_vocab_to_bad_request() {
9141 let (status, _body) = decode_error_response(generate::DecodeError::TokenOutOfVocab {
9142 token: 99,
9143 vocab_size: 32,
9144 });
9145 assert_eq!(status, StatusCode::BAD_REQUEST);
9146 }
9147
9148 #[test]
9149 fn run_generation_succeeds_and_releases_blocks_when_the_pool_has_room() {
9150 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9152 let pool = Arc::new(Mutex::new(ferrox_core::cache::KvBlockPool::new(64, 2)));
9153 let config = generate::KvPoolConfig {
9154 pool: pool.clone(),
9155 queue_wait: Duration::ZERO,
9156 };
9157
9158 let (_, finish, _usage) = run_generation(
9159 &model,
9160 &prompt,
9161 &greedy_params(4),
9162 Some(&config),
9163 None,
9164 None,
9165 None,
9166 None,
9167 None,
9168 )
9169 .unwrap();
9170 assert_eq!(finish, FinishReason::Length);
9171 assert_eq!(
9172 pool.lock().unwrap().free_blocks(),
9173 2,
9174 "a completed request must return its blocks to the pool"
9175 );
9176 }
9177
9178 #[tokio::test]
9183 async fn concurrent_requests_against_the_same_model_do_not_interfere() {
9184 let model = Arc::new(test_model());
9185 let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9186
9187 let mut handles = Vec::new();
9188 for _ in 0..8 {
9189 let model = Arc::clone(&model);
9190 let prompt = prompt.clone();
9191 handles.push(tokio::task::spawn_blocking(move || {
9192 run_generation(
9193 &model,
9194 &prompt,
9195 &greedy_params(6),
9196 None,
9197 None,
9198 None,
9199 None,
9200 None,
9201 None,
9202 )
9203 .unwrap()
9204 }));
9205 }
9206
9207 let mut results = Vec::new();
9208 for h in handles {
9209 results.push(h.await.unwrap());
9210 }
9211 for r in &results[1..] {
9216 assert_eq!(r.0, results[0].0, "decoded chunks must match");
9217 assert_eq!(r.1, results[0].1, "finish reason must match");
9218 assert_eq!(
9219 r.2.prompt_tokens, results[0].2.prompt_tokens,
9220 "prompt token count must match"
9221 );
9222 assert_eq!(
9223 r.2.completion_tokens, results[0].2.completion_tokens,
9224 "completion token count must match"
9225 );
9226 }
9227 }
9228
9229 fn write_safetensors_shard(tensors: &[(String, Vec<usize>, Vec<f32>)]) -> Vec<u8> {
9236 let mut header_entries = Vec::new();
9237 let mut data = Vec::new();
9238 for (name, shape, values) in tensors {
9239 let start = data.len();
9240 for v in values {
9241 data.extend_from_slice(&v.to_le_bytes());
9242 }
9243 let end = data.len();
9244 let shape_str = shape
9245 .iter()
9246 .map(|d| d.to_string())
9247 .collect::<Vec<_>>()
9248 .join(",");
9249 header_entries.push(format!(
9250 "\"{name}\":{{\"dtype\":\"F32\",\"shape\":[{shape_str}],\"data_offsets\":[{start},{end}]}}"
9251 ));
9252 }
9253 let header = format!("{{{}}}", header_entries.join(","));
9254 let header_bytes = header.as_bytes();
9255 let mut out = Vec::with_capacity(8 + header_bytes.len() + data.len());
9256 out.extend_from_slice(&(header_bytes.len() as u64).to_le_bytes());
9257 out.extend_from_slice(header_bytes);
9258 out.extend_from_slice(&data);
9259 out
9260 }
9261
9262 fn build_synthetic_kimi_loaded() -> model::KimiLoaded {
9273 use ferrox_models::config::{AttentionKind, KdaConfig, KimiHybridAttention, MlaConfig};
9274 use ferrox_models::kimi_loader::KimiRealHparams;
9275 use ferrox_moe::{GatingFunction, MoeLayerConfig};
9276
9277 let hidden_dim = 8;
9278 let kda_num_heads = 2;
9279 let kda_head_dim = 3;
9280 let kda_proj = kda_num_heads * kda_head_dim;
9281 let conv_kernel = 4;
9282 let dense_intermediate = 5;
9283 let vocab_size = 256;
9287 let mla_num_heads = 1;
9288 let mla_q_lora_rank = 2;
9289 let mla_kv_lora_rank = 2;
9290 let mla_qk_nope_head_dim = 2;
9291 let mla_qk_rope_head_dim = 2;
9292 let mla_v_head_dim = 2;
9293
9294 let model_cfg = ferrox_models::ModelConfig {
9295 name: "synthetic-kimi-server-test",
9296 n_layers: 1,
9297 hidden_dim,
9298 n_heads: 1,
9299 n_kv_heads: 1,
9300 head_dim: 4,
9301 vocab_size,
9302 rope_theta: 10000.0,
9303 rms_norm_eps: 1e-5,
9304 sliding_window: None,
9305 moe: MoeLayerConfig {
9306 expert_weights_scale: 1.0,
9307 n_experts: 1,
9308 n_experts_active: 1,
9309 n_shared_experts: 0,
9310 hidden_dim,
9311 expert_ffn_dim: 4,
9312 gating: GatingFunction::Sigmoid,
9313 norm_topk_prob: true,
9314 expert_group_count: None,
9315 expert_group_used_count: None,
9316 },
9317 n_dense_leading_layers: 1,
9322 attention: AttentionKind::KimiHybrid(KimiHybridAttention {
9323 kda_layers: vec![1],
9324 full_attn_layers: vec![],
9325 mla: MlaConfig {
9326 num_heads: mla_num_heads,
9327 q_lora_rank: mla_q_lora_rank,
9328 kv_lora_rank: mla_kv_lora_rank,
9329 qk_nope_head_dim: mla_qk_nope_head_dim,
9330 qk_rope_head_dim: mla_qk_rope_head_dim,
9331 v_head_dim: mla_v_head_dim,
9332 use_output_gate: true,
9333 rope: None,
9334 },
9335 kda: KdaConfig {
9336 num_heads: kda_num_heads,
9337 head_dim: kda_head_dim,
9338 short_conv_kernel_size: conv_kernel,
9339 gate_lower_bound: -5.0,
9340 use_full_rank_gate: true,
9341 },
9342 }),
9343 rope_freqs: None,
9344 rope_attn_factor: 1.0,
9345 rope_dim: None,
9346 rope_freqs_long: None,
9347 rope_freqs_short: None,
9348 rope_orig_ctx: None,
9349 rope_layout: ferrox_models::config::RopeLayout::Neox,
9350 qk_norm_style: ferrox_models::capability::QkNormStyle::WholeVector,
9351 swa_pattern: None,
9352 swa_dense_first: false,
9353 attn_logit_softcap: None,
9354 final_logit_softcap: None,
9355 embedding_scale: None,
9356 attention_scale: None,
9357 rope_theta_swa: None,
9358 ffn_activation: ferrox_models::config::FfnActivation::Swiglu,
9359 best_effort_fields: &["synthetic test config, not a real preset"],
9360 };
9361 let hp = KimiRealHparams {
9362 hidden_dim,
9363 kda_num_heads,
9364 kda_head_dim,
9365 mla_num_heads,
9366 mla_q_lora_rank,
9367 mla_kv_lora_rank,
9368 mla_qk_nope_head_dim,
9369 mla_qk_rope_head_dim,
9370 mla_v_head_dim,
9371 dense_intermediate_dim: dense_intermediate,
9372 moe_hidden_dim: hidden_dim,
9373 moe_intermediate_dim: 4,
9374 n_experts: 1,
9375 num_shared_experts: 0,
9376 };
9377
9378 let prefix = "language_model.model.layers.0";
9382 let mut tensors: Vec<(String, Vec<usize>, Vec<f32>)> = Vec::new();
9383 let push = |tensors: &mut Vec<(String, Vec<usize>, Vec<f32>)>,
9384 name: String,
9385 shape: Vec<usize>,
9386 n: usize| {
9387 tensors.push((name, shape, vec![0.01f32; n]));
9388 };
9389 push(
9390 &mut tensors,
9391 format!("{prefix}.input_layernorm.weight"),
9392 vec![hidden_dim],
9393 hidden_dim,
9394 );
9395 push(
9396 &mut tensors,
9397 format!("{prefix}.post_attention_layernorm.weight"),
9398 vec![hidden_dim],
9399 hidden_dim,
9400 );
9401 push(
9402 &mut tensors,
9403 format!("{prefix}.self_attention_res_norm.weight"),
9404 vec![hidden_dim],
9405 hidden_dim,
9406 );
9407 push(
9408 &mut tensors,
9409 format!("{prefix}.self_attention_res_proj.weight"),
9410 vec![1, hidden_dim],
9411 hidden_dim,
9412 );
9413 push(
9414 &mut tensors,
9415 format!("{prefix}.mlp_res_norm.weight"),
9416 vec![hidden_dim],
9417 hidden_dim,
9418 );
9419 push(
9420 &mut tensors,
9421 format!("{prefix}.mlp_res_proj.weight"),
9422 vec![1, hidden_dim],
9423 hidden_dim,
9424 );
9425 push(
9426 &mut tensors,
9427 format!("{prefix}.self_attn.q_proj.weight"),
9428 vec![kda_proj, hidden_dim],
9429 kda_proj * hidden_dim,
9430 );
9431 push(
9432 &mut tensors,
9433 format!("{prefix}.self_attn.k_proj.weight"),
9434 vec![kda_proj, hidden_dim],
9435 kda_proj * hidden_dim,
9436 );
9437 push(
9438 &mut tensors,
9439 format!("{prefix}.self_attn.v_proj.weight"),
9440 vec![kda_proj, hidden_dim],
9441 kda_proj * hidden_dim,
9442 );
9443 push(
9444 &mut tensors,
9445 format!("{prefix}.self_attn.q_conv1d.weight"),
9446 vec![kda_proj, 1, conv_kernel],
9447 kda_proj * conv_kernel,
9448 );
9449 push(
9450 &mut tensors,
9451 format!("{prefix}.self_attn.k_conv1d.weight"),
9452 vec![kda_proj, 1, conv_kernel],
9453 kda_proj * conv_kernel,
9454 );
9455 push(
9456 &mut tensors,
9457 format!("{prefix}.self_attn.v_conv1d.weight"),
9458 vec![kda_proj, 1, conv_kernel],
9459 kda_proj * conv_kernel,
9460 );
9461 push(
9462 &mut tensors,
9463 format!("{prefix}.self_attn.A_log"),
9464 vec![kda_num_heads],
9465 kda_num_heads,
9466 );
9467 push(
9468 &mut tensors,
9469 format!("{prefix}.self_attn.f_a_proj.weight"),
9470 vec![kda_head_dim, hidden_dim],
9471 kda_head_dim * hidden_dim,
9472 );
9473 push(
9474 &mut tensors,
9475 format!("{prefix}.self_attn.f_b_proj.weight"),
9476 vec![kda_proj, kda_head_dim],
9477 kda_proj * kda_head_dim,
9478 );
9479 push(
9480 &mut tensors,
9481 format!("{prefix}.self_attn.dt_bias"),
9482 vec![kda_proj],
9483 kda_proj,
9484 );
9485 push(
9486 &mut tensors,
9487 format!("{prefix}.self_attn.b_proj.weight"),
9488 vec![kda_num_heads, hidden_dim],
9489 kda_num_heads * hidden_dim,
9490 );
9491 push(
9492 &mut tensors,
9493 format!("{prefix}.self_attn.g_proj.weight"),
9494 vec![kda_proj, hidden_dim],
9495 kda_proj * hidden_dim,
9496 );
9497 push(
9498 &mut tensors,
9499 format!("{prefix}.self_attn.o_norm.weight"),
9500 vec![kda_head_dim],
9501 kda_head_dim,
9502 );
9503 push(
9504 &mut tensors,
9505 format!("{prefix}.self_attn.o_proj.weight"),
9506 vec![hidden_dim, kda_proj],
9507 hidden_dim * kda_proj,
9508 );
9509 push(
9510 &mut tensors,
9511 format!("{prefix}.mlp.gate_proj.weight"),
9512 vec![dense_intermediate, hidden_dim],
9513 dense_intermediate * hidden_dim,
9514 );
9515 push(
9516 &mut tensors,
9517 format!("{prefix}.mlp.up_proj.weight"),
9518 vec![dense_intermediate, hidden_dim],
9519 dense_intermediate * hidden_dim,
9520 );
9521 push(
9522 &mut tensors,
9523 format!("{prefix}.mlp.down_proj.weight"),
9524 vec![hidden_dim, dense_intermediate],
9525 hidden_dim * dense_intermediate,
9526 );
9527 push(
9528 &mut tensors,
9529 "language_model.model.embed_tokens.weight".to_string(),
9530 vec![vocab_size, hidden_dim],
9531 vocab_size * hidden_dim,
9532 );
9533 push(
9534 &mut tensors,
9535 "language_model.lm_head.weight".to_string(),
9536 vec![vocab_size, hidden_dim],
9537 vocab_size * hidden_dim,
9538 );
9539 push(
9540 &mut tensors,
9541 "language_model.model.norm.weight".to_string(),
9542 vec![hidden_dim],
9543 hidden_dim,
9544 );
9545 push(
9546 &mut tensors,
9547 "language_model.model.output_attn_res_norm.weight".to_string(),
9548 vec![hidden_dim],
9549 hidden_dim,
9550 );
9551 push(
9552 &mut tensors,
9553 "language_model.model.output_attn_res_proj.weight".to_string(),
9554 vec![1, hidden_dim],
9555 hidden_dim,
9556 );
9557
9558 static FIXTURE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
9571 let dir = std::env::temp_dir().join(format!(
9572 "ferrox_server_kimi_e2e_test_{}_{}",
9573 std::process::id(),
9574 FIXTURE.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
9575 ));
9576 std::fs::create_dir_all(&dir).unwrap();
9577 let shard_bytes = write_safetensors_shard(&tensors);
9578 std::fs::write(dir.join("shard0.safetensors"), &shard_bytes).unwrap();
9579 let map_entries: Vec<String> = tensors
9580 .iter()
9581 .map(|(name, ..)| format!("\"{name}\":\"shard0.safetensors\""))
9582 .collect();
9583 let index = format!("{{\"weight_map\":{{{}}}}}", map_entries.join(","));
9584 std::fs::write(dir.join("model.safetensors.index.json"), &index).unwrap();
9585
9586 use base64::Engine;
9590 let vocab_lines: Vec<String> = (0..vocab_size as u32)
9591 .map(|b| {
9592 let b64 = base64::engine::general_purpose::STANDARD.encode([b as u8]);
9593 format!("{b64} {b}")
9594 })
9595 .collect();
9596 std::fs::write(dir.join("tiktoken.model"), vocab_lines.join("\n")).unwrap();
9597
9598 let loaded = model::load_kimi_checkpoint_with_config(dir.to_str().unwrap(), model_cfg, hp)
9599 .expect("must load the synthetic Kimi checkpoint end to end");
9600 std::fs::remove_dir_all(&dir).ok();
9601 loaded
9602 }
9603
9604 #[test]
9612 fn kimi_model_serves_real_text_end_to_end_via_run_generation() {
9613 let loaded = build_synthetic_kimi_loaded();
9614 let state = build_app_state(
9615 StartupModels {
9616 loaded: model::LoadedModel::Kimi(loaded),
9617 embedding: None,
9618 },
9619 None,
9620 None,
9621 None,
9622 false,
9623 None,
9624 Arc::new(health::Detection::ready(health::probe_backends())),
9625 );
9626 let active = state.active().expect("a freshly built state has a model");
9627 assert_eq!(active.tokenizer_kind(), "kimi-tiktoken-bpe");
9628 assert!(!active.is_synthetic());
9629
9630 let (_chunks, finish, _usage) = run_generation(
9631 active.generative().unwrap(),
9632 "hi",
9633 &greedy_params(5),
9634 None,
9635 None,
9636 None,
9637 None,
9638 None,
9639 None,
9640 )
9641 .expect("a real Kimi checkpoint must generate without error");
9642 assert!(matches!(finish, FinishReason::Length | FinishReason::Stop));
9643 }
9644
9645 #[test]
9658 fn a_grammar_constrains_the_engine_decode_path() {
9659 let loaded = build_synthetic_kimi_loaded();
9660 let state = build_app_state(
9661 StartupModels {
9662 loaded: model::LoadedModel::Kimi(loaded),
9663 embedding: None,
9664 },
9665 None,
9666 None,
9667 None,
9668 false,
9669 None,
9670 Arc::new(health::Detection::ready(health::probe_backends())),
9671 );
9672 let active = state.active().expect("a freshly built state has a model");
9673
9674 let run = |grammar: Option<&str>| {
9675 let mut params = greedy_params(6);
9676 params.grammar = grammar.map(|src| {
9677 Arc::new(
9678 ferrox_models::grammar::Grammar::from_str_with_root(src, "root")
9679 .expect("test grammar parses"),
9680 )
9681 });
9682 run_generation(
9683 active.generative().unwrap(),
9684 "hi",
9685 ¶ms,
9686 None,
9687 None,
9688 None,
9689 None,
9690 None,
9691 None,
9692 )
9693 };
9694
9695 let (chunks, _, _) = run(None).expect("the unconstrained run must serve");
9696 let unconstrained = chunks.concat();
9697 assert!(
9698 unconstrained.chars().any(|c| c != 'a'),
9699 "the unconstrained run produced only `a` ({unconstrained:?}), so the \
9700 constrained run below would prove nothing"
9701 );
9702
9703 let (chunks, finish, _) =
9704 run(Some(r#"root ::= "a"+"#)).expect("a grammar this vocabulary can spell must serve");
9705 let constrained = chunks.concat();
9706 assert!(
9707 !constrained.is_empty() && constrained.chars().all(|c| c == 'a'),
9708 "the engine decode path served text its grammar forbids ({constrained:?}): \
9709 the constraint was dropped between `generate_engine` and the sampler"
9710 );
9711 assert!(matches!(finish, FinishReason::Length | FinishReason::Stop));
9712 }
9713
9714 #[test]
9724 fn kv_pool_and_prefix_cache_are_never_consulted_for_a_kimi_model() {
9725 let loaded = build_synthetic_kimi_loaded();
9726 let state = build_app_state(
9727 StartupModels {
9728 loaded: model::LoadedModel::Kimi(loaded),
9729 embedding: None,
9730 },
9731 None,
9732 None,
9733 None,
9734 false,
9735 None,
9736 Arc::new(health::Detection::ready(health::probe_backends())),
9737 );
9738
9739 let pool = Arc::new(Mutex::new(ferrox_core::cache::KvBlockPool::new(64, 4)));
9740 let kv_pool_config = generate::KvPoolConfig {
9741 pool: pool.clone(),
9742 queue_wait: Duration::ZERO,
9743 };
9744 let pc = Mutex::new(PrefixCache::new(4));
9745
9746 run_generation(
9747 state
9748 .active()
9749 .expect("a freshly built state has a model")
9750 .generative()
9751 .unwrap(),
9752 "hi",
9753 &greedy_params(5),
9754 Some(&kv_pool_config),
9755 None,
9756 Some(&pc),
9757 None,
9758 None,
9759 None,
9760 )
9761 .expect("a real Kimi checkpoint must generate without error");
9762
9763 assert_eq!(
9764 pool.lock().unwrap().free_blocks(),
9765 4,
9766 "the KV pool must be completely untouched by a Kimi request"
9767 );
9768 let stats = pc.lock().unwrap().stats();
9769 assert_eq!(
9770 stats.hits + stats.misses,
9771 0,
9772 "the prefix cache must never be consulted for a Kimi request"
9773 );
9774 }
9775}