1mod admin;
36mod anthropic;
37mod attribution;
38mod budget;
39mod cache_admin;
40mod cancel;
41mod chat_params;
42mod chat_template;
43mod cli;
44mod completion;
45mod continuation;
46mod conversations;
47mod decode_task;
48mod embeddings;
49mod generate;
50mod grammar_request;
51mod health;
52mod journal;
53mod json_mode;
54mod limits;
55mod loaded;
56mod mcp;
57mod model;
58mod openai_extra;
59mod output;
60mod policy;
61mod prefill_batch;
62mod reasoning_budget;
63mod reasoning_tokens;
64mod rerank;
65mod response_cache;
66pub(crate) mod responses;
67mod resume;
68mod sample_step;
69mod sampling_knobs;
70mod security;
71mod serving;
72mod session;
73mod slots;
74mod sse;
75mod stats;
76mod stop;
77mod stream_events;
78mod tasks;
79mod tool_grammar;
80mod unsupported_sampling;
81mod utf8_stream;
82
83use std::cell::RefCell;
84use std::convert::Infallible;
85use std::net::SocketAddr;
86use std::path::PathBuf;
87use std::rc::Rc;
88use std::sync::{Arc, Mutex, MutexGuard};
89use std::time::Duration;
90
91use axum::{
92 extract::State,
93 http::StatusCode,
94 response::sse::{Event, Sse},
95 response::{IntoResponse, Response},
96 routing::{get, post},
97 Json, Router,
98};
99use serde::{Deserialize, Serialize};
100
101use cli::{apply_cli_overrides, print_available_devices};
102pub use cli::{ServerArgs, BUILT_WITH_CUDA, BUILT_WITH_METAL};
103
104use ferrox_core::cache::KvBlockPool;
105use ferrox_models::kimi_tokenizer::KimiTokenizer;
106use ferrox_models::sampling::SamplingParams;
107use ferrox_models::tokenizer::{SpecialTokens, StopTokens};
108use ferrox_models::{Decoder, Gemma4Engine, KimiEngine, MlaEngine, PrefixCache};
109use generate::{FinishReason, GenerationParams};
110pub(crate) use loaded::{ActiveModel, Loaded};
111use model::ServerTokenizer;
112use rerank::encoder_endpoints;
113use response_cache::ResponseCache;
114use sampling_knobs::SamplingKnobs;
115
116#[allow(clippy::large_enum_variant)] pub(crate) enum Model {
125 Gguf(GgufModel),
126 Kimi(KimiModel),
127 Mla(MlaModel),
128 Gemma4(Gemma4Model),
129 Glm52(Glm52Model),
130}
131
132pub(crate) struct GgufModel {
133 decoder: Arc<Decoder>,
134 tokenizer: Arc<ServerTokenizer>,
135 stop_tokens: StopTokens,
136 bos_id: Option<usize>,
137 is_synthetic: bool,
138 chat_template: chat_template::PromptTemplate,
139}
140
141pub(crate) struct KimiModel {
142 engine: KimiEngine,
143 tokenizer: KimiTokenizer,
144 stop_tokens: StopTokens,
145 chat_template: chat_template::PromptTemplate,
146}
147
148pub(crate) struct MlaModel {
149 engine: MlaEngine,
150 tokenizer: ServerTokenizer,
151 stop_tokens: StopTokens,
152 bos_id: Option<usize>,
153 name: String,
154 chat_template: chat_template::PromptTemplate,
155}
156
157pub(crate) struct Gemma4Model {
158 engine: Gemma4Engine,
159 tokenizer: ServerTokenizer,
160 stop_tokens: StopTokens,
161 bos_id: Option<usize>,
162 name: String,
163 chat_template: chat_template::PromptTemplate,
164}
165
166pub(crate) struct Glm52Model {
167 engine: ferrox_models::Glm52Engine,
168 tokenizer: ServerTokenizer,
169 stop_tokens: StopTokens,
170 bos_id: Option<usize>,
171 name: String,
172 chat_template: chat_template::PromptTemplate,
173}
174
175impl Model {
176 pub(crate) fn chat_template(&self) -> chat_template::PromptTemplate {
177 match self {
178 Model::Gguf(m) => m.chat_template.clone(),
179 Model::Kimi(m) => m.chat_template.clone(),
180 Model::Mla(m) => m.chat_template.clone(),
181 Model::Gemma4(m) => m.chat_template.clone(),
182 Model::Glm52(m) => m.chat_template.clone(),
183 }
184 }
185
186 fn is_synthetic(&self) -> bool {
191 match self {
192 Model::Gguf(m) => m.is_synthetic,
193 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => false,
194 }
195 }
196
197 fn tokenizer_kind(&self) -> &'static str {
198 match self {
199 Model::Gguf(m) => m.tokenizer.kind(),
200 Model::Kimi(_) => "kimi-tiktoken-bpe",
201 Model::Mla(m) => m.tokenizer.kind(),
202 Model::Gemma4(m) => m.tokenizer.kind(),
203 Model::Glm52(m) => m.tokenizer.kind(),
204 }
205 }
206
207 fn expert_store_stats(&self) -> Option<ferrox_core::expert_store::ExpertStoreStats> {
211 match self {
212 Model::Gguf(m) => m.decoder.expert_store_stats(),
213 Model::Kimi(m) => m.engine.weights.expert_store_stats(),
214 Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
215 }
216 }
217
218 pub(crate) fn name(&self) -> &str {
219 match self {
220 Model::Gguf(m) => m.decoder.config.name,
221 Model::Kimi(_) => "kimi-k3",
222 Model::Mla(m) => m.name.as_str(),
223 Model::Gemma4(m) => m.name.as_str(),
224 Model::Glm52(m) => m.name.as_str(),
225 }
226 }
227
228 pub(crate) fn encode(&self, text: &str, specials: SpecialTokens) -> Vec<usize> {
255 match self {
256 Model::Gguf(m) => m.tokenizer.encode(text, specials),
257 Model::Kimi(m) => m
258 .tokenizer
259 .encode(text, specials)
260 .into_iter()
261 .map(|id| id as usize)
262 .collect(),
263 Model::Mla(m) => m.tokenizer.encode(text, specials),
264 Model::Gemma4(m) => m.tokenizer.encode(text, specials),
265 Model::Glm52(m) => m.tokenizer.encode(text, specials),
266 }
267 }
268
269 pub(crate) fn bos_id(&self) -> Option<usize> {
278 match self {
279 Model::Gguf(m) => m.bos_id,
280 Model::Kimi(_) => None,
281 Model::Mla(m) => m.bos_id,
282 Model::Gemma4(m) => m.bos_id,
283 Model::Glm52(m) => m.bos_id,
284 }
285 }
286
287 pub(crate) fn decode(&self, ids: &[usize]) -> String {
288 match self {
289 Model::Gguf(m) => m.tokenizer.decode(ids),
290 Model::Kimi(m) => {
291 let ids32: Vec<u32> = ids.iter().map(|&id| id as u32).collect();
292 m.tokenizer.decode(&ids32)
293 }
294 Model::Mla(m) => m.tokenizer.decode(ids),
295 Model::Gemma4(m) => m.tokenizer.decode(ids),
296 Model::Glm52(m) => m.tokenizer.decode(ids),
297 }
298 }
299
300 pub(crate) fn embed_tokens(&self, tokens: &[usize]) -> Option<Vec<Vec<f32>>> {
303 match self {
304 Model::Gguf(m) => {
305 let mut caches: Vec<_> = m.decoder.config.new_kv_caches();
306 Some(m.decoder.forward_hidden_batch(tokens, 0, &mut caches))
307 }
308 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
309 }
310 }
311
312 pub(crate) fn gguf_decoder(&self) -> Option<&Arc<Decoder>> {
318 match self {
319 Model::Gguf(m) => Some(&m.decoder),
320 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
321 }
322 }
323
324 pub(crate) fn vocab_size(&self) -> Option<usize> {
325 match self {
326 Model::Gguf(m) => Some(m.decoder.config.vocab_size),
327 Model::Kimi(m) => Some(m.tokenizer.vocab_size()),
328 Model::Mla(m) => Some(ferrox_models::Engine::vocab_size(&m.engine)),
329 Model::Gemma4(m) => Some(ferrox_models::Engine::vocab_size(&m.engine)),
330 Model::Glm52(m) => Some(ferrox_models::Engine::vocab_size(&m.engine)),
331 }
332 }
333
334 fn has_real_vocabulary(&self) -> bool {
341 match self {
342 Model::Gguf(m) => !matches!(*m.tokenizer, model::ServerTokenizer::Byte),
343 Model::Kimi(_) => true,
344 Model::Mla(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
345 Model::Gemma4(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
346 Model::Glm52(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
347 }
348 }
349}
350
351impl ferrox_models::dry::DryVocab for Model {
357 fn n_tokens(&self) -> usize {
358 self.vocab_size().unwrap_or(0)
359 }
360
361 fn detokenize(&self, token: usize) -> String {
362 self.decode(&[token])
363 }
364
365 fn tokenize(&self, text: &str) -> Vec<usize> {
366 self.encode(text, SpecialTokens::AsText)
367 }
368}
369
370pub(crate) struct AppState {
371 pub(crate) embedding: Option<Arc<ferrox_models::EmbeddingModel>>,
385 active: std::sync::RwLock<Option<Arc<ActiveModel>>>,
404 pub(crate) load_in_progress: std::sync::atomic::AtomicBool,
408 pub(crate) tasks: Arc<tasks::TaskRegistry>,
410 pub(crate) cancels: Arc<cancel::CancelRegistry>,
414 pub(crate) stats: stats::Stats,
417 pub(crate) streams: resume::StreamRegistry,
420 pub(crate) model_dir: Option<PathBuf>,
422 response_cache: Mutex<ResponseCache>,
426 pub(crate) kv_pool: Option<generate::KvPoolConfig>,
438 pub(crate) paged_kv: Option<generate::PagedKvConfig>,
451 pub(crate) prefix_cache: Option<Arc<Mutex<PrefixCache>>>,
461 sessions: session::SessionStore,
467 requests_total: std::sync::atomic::AtomicU64,
468 request_errors_total: std::sync::atomic::AtomicU64,
469 started_at: std::time::Instant,
470 last_request_ms: std::sync::atomic::AtomicU64,
475 detection: Arc<health::Detection>,
477 mcp: Option<mcp::LoadedMcpConfig>,
479 pub(crate) continuous_batching_enabled: bool,
483 pub(crate) metal_private_decode_gate: Option<Arc<std::sync::Mutex<()>>>,
487 loading_model: Mutex<Option<String>>,
491 last_load_error: Mutex<Option<(String, String)>>,
495 pub(crate) serving: Mutex<crate::stats::ServingStats>,
500 pub(crate) maintenance: Mutex<crate::policy::maintenance::MaintenanceGate>,
505 pub(crate) footprint:
511 Mutex<crate::policy::footprint::ProbeCache<crate::policy::footprint::Footprint>>,
512 pub(crate) started_unix: u64,
522}
523
524pub(crate) const FOOTPRINT_TTL_MS: u64 = 2_000;
530
531impl AppState {
532 pub(crate) fn active(&self) -> Option<Arc<ActiveModel>> {
536 self.active
537 .read()
538 .unwrap_or_else(|p| p.into_inner())
539 .clone()
540 }
541
542 pub(crate) fn require_active(&self) -> Result<Arc<ActiveModel>, ApiError> {
548 self.active().ok_or_else(|| {
549 (
550 StatusCode::SERVICE_UNAVAILABLE,
551 Json(serde_json::json!({"error": {
552 "message": "no model is loaded; POST /admin/models/load with an id from \
553 GET /admin/models",
554 "type": "model_not_loaded"
555 }})),
556 )
557 })
558 }
559
560 pub(crate) fn require_model(&self) -> Result<Arc<Model>, ApiError> {
568 Ok(Arc::clone(self.require_active()?.generative()?))
569 }
570
571 pub(crate) fn swap_active(&self, next: Option<Arc<ActiveModel>>) -> Option<Arc<ActiveModel>> {
579 let mut guard = self.active.write().unwrap_or_else(|p| p.into_inner());
580 std::mem::replace(&mut *guard, next)
581 }
582
583 fn mark_request_finished(&self) {
587 let ms = self.started_at.elapsed().as_millis().min(u64::MAX as u128) as u64;
588 self.last_request_ms
589 .store(ms, std::sync::atomic::Ordering::Relaxed);
590 }
591
592 pub(crate) fn uptime(&self) -> Duration {
593 self.started_at.elapsed()
594 }
595
596 pub(crate) fn requests_total(&self) -> u64 {
597 self.requests_total
598 .load(std::sync::atomic::Ordering::Relaxed)
599 }
600
601 pub(crate) fn errors_total(&self) -> u64 {
602 self.request_errors_total
603 .load(std::sync::atomic::Ordering::Relaxed)
604 }
605
606 pub(crate) fn cache_stats(&self) -> response_cache::CacheStats {
607 lock_cache(&self.response_cache).stats()
608 }
609
610 pub(crate) fn last_request_age_seconds(&self) -> Option<f64> {
613 let last = self
614 .last_request_ms
615 .load(std::sync::atomic::Ordering::Relaxed);
616 (last > 0)
617 .then(|| self.uptime().as_secs_f64() - (last as f64 / 1000.0))
618 .map(|age| age.max(0.0))
619 }
620
621 pub(crate) fn loading_model_id(&self) -> Option<String> {
622 self.loading_model
623 .lock()
624 .unwrap_or_else(|p| p.into_inner())
625 .clone()
626 }
627
628 pub(crate) fn set_loading_model(&self, id: Option<String>) {
629 *self.loading_model.lock().unwrap_or_else(|p| p.into_inner()) = id;
630 }
631
632 pub(crate) fn last_load_error(&self) -> Option<(String, String)> {
633 self.last_load_error
634 .lock()
635 .unwrap_or_else(|p| p.into_inner())
636 .clone()
637 }
638
639 pub(crate) fn set_last_load_error(&self, error: Option<(String, String)>) {
640 *self
641 .last_load_error
642 .lock()
643 .unwrap_or_else(|p| p.into_inner()) = error;
644 }
645
646 pub(crate) fn active_model_name(&self) -> Option<String> {
655 self.active().map(|a| a.name().to_string())
656 }
657
658 pub(crate) fn embedding_model(&self) -> Option<Arc<ferrox_models::EmbeddingModel>> {
668 self.embedding
669 .clone()
670 .or_else(|| self.active().and_then(|a| a.encoder().map(Arc::clone)))
671 }
672
673 pub(crate) fn embedding_model_name(&self) -> Option<String> {
677 match self.embedding_model() {
678 Some(e) => Some(e.name().to_string()),
679 None => self.active_model_name(),
680 }
681 }
682
683 pub(crate) fn record_request(&self, record: stats::Record<'_>) {
684 self.stats.record(stats::entry(record));
685 }
686}
687
688fn lock_cache(cache: &Mutex<ResponseCache>) -> MutexGuard<'_, ResponseCache> {
693 cache
694 .lock()
695 .unwrap_or_else(|poisoned| poisoned.into_inner())
696}
697
698#[derive(Debug, Clone, Deserialize)]
699#[serde(untagged)]
700pub(crate) enum MessageContent {
701 Text(String),
702 Parts(Vec<ContentPart>),
703}
704
705#[derive(Debug, Clone, Deserialize)]
706struct ContentPart {
707 #[serde(rename = "type")]
708 kind: String,
709 #[serde(default)]
710 text: Option<String>,
711 #[serde(default)]
712 image_url: Option<serde_json::Value>,
713}
714
715impl MessageContent {
716 fn as_text(&self) -> String {
717 match self {
718 Self::Text(s) => s.clone(),
719 Self::Parts(parts) => parts
720 .iter()
721 .filter_map(|p| p.text.as_deref())
722 .collect::<Vec<_>>()
723 .join(""),
724 }
725 }
726
727 fn has_image(&self) -> bool {
728 match self {
729 Self::Text(_) => false,
730 Self::Parts(parts) => parts
731 .iter()
732 .any(|p| p.kind == "image_url" || p.image_url.is_some()),
733 }
734 }
735}
736
737#[derive(Debug, Clone, Deserialize)]
738pub(crate) struct ChatMessage {
739 pub(crate) role: String,
740 #[serde(default)]
744 pub(crate) content: Option<MessageContent>,
745 #[serde(default)]
749 pub(crate) tool_calls: Option<Vec<ToolCallIn>>,
750 #[serde(default)]
755 #[allow(dead_code)]
756 pub(crate) tool_call_id: Option<String>,
757 #[serde(default, alias = "reasoning")]
773 pub(crate) reasoning_content: Option<String>,
774}
775
776impl ChatMessage {
777 fn rendered_content(&self) -> String {
784 let mut out = self
785 .content
786 .as_ref()
787 .map(MessageContent::as_text)
788 .unwrap_or_default();
789 if let Some(calls) = &self.tool_calls {
790 for call in calls {
791 out.push_str(&format!(
792 "<tool_call>{{\"name\": \"{}\", \"arguments\": {}}}</tool_call>",
793 call.function.name, call.function.arguments
794 ));
795 }
796 }
797 out
798 }
799}
800
801#[derive(Debug, Clone, Deserialize)]
802pub(crate) struct ToolCallIn {
803 #[serde(default)]
804 #[allow(dead_code)]
805 id: String,
806 #[serde(rename = "type", default)]
807 #[allow(dead_code)]
808 kind: String,
809 function: ToolCallFunctionIn,
810}
811
812#[derive(Debug, Clone, Deserialize)]
813struct ToolCallFunctionIn {
814 name: String,
815 arguments: String,
820}
821
822#[derive(Debug, Clone, Deserialize)]
825struct ToolDef {
826 #[serde(rename = "type", default)]
827 #[allow(dead_code)]
828 kind: String,
829 function: ToolFunctionDef,
830}
831
832#[derive(Debug, Clone, Deserialize)]
833struct ToolFunctionDef {
834 name: String,
835 #[serde(default)]
836 description: Option<String>,
837 #[serde(default)]
838 parameters: Option<serde_json::Value>,
839}
840
841#[derive(Debug, Clone, Deserialize)]
852#[serde(untagged)]
853enum ToolChoice {
854 Mode(String),
855 Specific(serde_json::Value),
856}
857
858#[derive(Deserialize)]
861#[serde(untagged)]
862enum StopParam {
863 One(String),
864 Many(Vec<String>),
865}
866
867#[derive(Deserialize)]
868struct ChatCompletionRequest {
869 model: String,
870 messages: Vec<ChatMessage>,
871 #[serde(default = "default_max_tokens")]
872 max_tokens: usize,
873 #[serde(default)]
874 temperature: Option<f32>,
875 #[serde(default)]
876 top_p: Option<f32>,
877 #[serde(default)]
882 min_p: Option<f32>,
883 #[serde(default)]
884 top_k: Option<usize>,
885 #[serde(default)]
886 repetition_penalty: Option<f32>,
887 #[serde(flatten)]
891 extra_samplers: crate::sampling_knobs::ExtraSamplerFields,
892 #[serde(default)]
893 seed: Option<u64>,
894 #[serde(default)]
895 stop: Option<StopParam>,
896 #[serde(default)]
897 stream: Option<bool>,
898 #[serde(default)]
911 stream_resumable: Option<bool>,
912 #[serde(default)]
922 ignore_eos: Option<bool>,
923 #[serde(default)]
924 tools: Vec<ToolDef>,
925 #[serde(default)]
926 tool_choice: Option<ToolChoice>,
927 #[serde(default)]
935 chat_template_kwargs: Option<serde_json::Map<String, serde_json::Value>>,
936 #[serde(default)]
944 reasoning_effort: Option<String>,
945 #[serde(default)]
949 thinking: Option<ThinkingSwitch>,
950 #[serde(default)]
955 session_id: Option<String>,
956 #[serde(default, deserialize_with = "continuation::deserialize")]
964 continue_final_message: continuation::ContinueFinalMessage,
965 #[serde(default, alias = "thinking_budget_tokens")]
973 reasoning_budget_tokens: Option<reasoning_budget::BudgetTokens>,
974 #[serde(default)]
976 logprobs: Option<bool>,
977 #[serde(default)]
978 top_logprobs: Option<u32>,
979 #[serde(default)]
980 n: Option<u32>,
981 #[serde(default)]
982 presence_penalty: Option<f32>,
983 #[serde(default)]
984 frequency_penalty: Option<f32>,
985 #[serde(default)]
986 response_format: Option<serde_json::Value>,
987 #[serde(default)]
994 logit_bias: Option<serde_json::Value>,
995 #[serde(default)]
1005 samplers: Option<serde_json::Value>,
1006 #[serde(default)]
1016 grammar: Option<String>,
1017}
1018
1019const DEFAULT_CHAT_MAX_TOKENS: usize = 32_768;
1031
1032#[derive(Debug, Clone, Deserialize)]
1034pub(crate) struct ThinkingSwitch {
1035 #[serde(rename = "type")]
1036 pub(crate) kind: String,
1037}
1038
1039const THINKING_KWARG_KEYS: [&str; 4] = [
1043 "enable_thinking",
1044 "thinking",
1045 "thinking_mode",
1046 "reasoning_effort",
1047];
1048
1049const DISABLE_EFFORTS: [&str; 2] = ["none", "off"];
1053
1054fn default_max_tokens() -> usize {
1055 DEFAULT_CHAT_MAX_TOKENS
1056}
1057
1058impl ChatCompletionRequest {
1059 fn sampling_knobs(&self) -> Result<SamplingKnobs, ApiError> {
1068 let mut knobs = SamplingKnobs {
1069 temperature: self.temperature,
1070 top_p: self.top_p,
1071 min_p: self.min_p,
1072 top_k: self.top_k,
1073 repetition_penalty: self.repetition_penalty,
1074 presence_penalty: self.presence_penalty,
1075 frequency_penalty: self.frequency_penalty,
1076 penalty_last_n: None,
1080 sampler_order: unsupported_sampling::parse_sampler_order(
1081 self.samplers.as_ref(),
1082 "/v1/chat/completions",
1083 )?,
1084 ..SamplingKnobs::default()
1085 };
1086 self.extra_samplers.apply(&mut knobs);
1087 Ok(knobs)
1088 }
1089
1090 fn sampling_params(
1091 &self,
1092 model: crate::sampling_knobs::SamplerModel<'_>,
1093 ) -> Result<SamplingParams, ApiError> {
1094 self.sampling_knobs()?.resolve(model).map_err(|e| {
1095 unsupported_feature(&format!("`dry_multiplier` on /v1/chat/completions: {e}"))
1096 })
1097 }
1098
1099 fn stop_sequences(&self) -> Vec<String> {
1100 self.stop
1101 .as_ref()
1102 .map(|s| match s {
1103 StopParam::One(v) => vec![v.clone()],
1104 StopParam::Many(v) => v.clone(),
1105 })
1106 .unwrap_or_default()
1107 }
1108
1109 fn tools_active(&self) -> bool {
1114 !self.tools.is_empty()
1115 && !matches!(&self.tool_choice, Some(ToolChoice::Mode(m)) if m == "none")
1116 }
1117
1118 fn forced_tool_choice(&self) -> Result<Option<tool_grammar::Forced<'_>>, ApiError> {
1130 match &self.tool_choice {
1131 Some(ToolChoice::Mode(m)) if m == "required" => Ok(Some(tool_grammar::Forced::Any)),
1132 Some(ToolChoice::Specific(value)) => {
1133 let name = value
1137 .get("function")
1138 .and_then(|f| f.get("name"))
1139 .or_else(|| value.get("name"))
1140 .and_then(|n| n.as_str());
1141 match name {
1142 Some(name) => Ok(Some(tool_grammar::Forced::Named(name))),
1143 None => Err(invalid_request(
1144 "tool_choice must be \"auto\", \"none\", \"required\", or an object with \
1145 function.name",
1146 "tool_choice",
1147 )),
1148 }
1149 }
1150 _ => Ok(None),
1151 }
1152 }
1153
1154 fn tool_specs(&self) -> Vec<tool_grammar::ToolSpec<'_>> {
1156 self.tools
1157 .iter()
1158 .map(|t| tool_grammar::ToolSpec {
1159 name: &t.function.name,
1160 parameters: t.function.parameters.as_ref(),
1161 })
1162 .collect()
1163 }
1164
1165 fn thinking_direction(&self) -> Option<bool> {
1208 if let Some(switch) = &self.thinking {
1209 return match switch.kind.trim().to_ascii_lowercase().as_str() {
1210 "disabled" => Some(false),
1211 "enabled" => Some(true),
1212 _ => None,
1215 };
1216 }
1217 let effort = self.reasoning_effort.as_ref()?;
1218 DISABLE_EFFORTS
1219 .contains(&effort.trim().to_ascii_lowercase().as_str())
1220 .then_some(false)
1221 }
1222
1223 fn resolve_template_kwargs(
1224 &self,
1225 template: &chat_template::PromptTemplate,
1226 ) -> serde_json::Map<String, serde_json::Value> {
1227 let mut kwargs = self.chat_template_kwargs.clone().unwrap_or_default();
1228 let caller_steered = THINKING_KWARG_KEYS.iter().any(|k| kwargs.contains_key(*k));
1232
1233 if !caller_steered {
1234 match self.thinking_direction() {
1235 Some(false) => {
1236 for (k, v) in crate::policy::effort::thinking_off_kwargs() {
1237 kwargs.insert(k, v);
1238 }
1239 return kwargs;
1242 }
1243 Some(true) => {
1244 for (k, v) in crate::policy::effort::thinking_on_kwargs() {
1245 kwargs.insert(k, v);
1246 }
1247 }
1248 None => {}
1249 }
1250 if let Some(effort) = &self.reasoning_effort {
1251 kwargs
1252 .entry("reasoning_effort".to_string())
1253 .or_insert_with(|| serde_json::json!(effort));
1254 }
1255 }
1256
1257 let offered: Vec<serde_json::Value> = if self.tools_active() {
1258 self.tools.iter().map(chat_template::tool_json).collect()
1259 } else {
1260 Vec::new()
1261 };
1262 let thinking = crate::policy::effort::resolve_thinking_mode(Some(&kwargs), Some(&offered));
1263 if thinking == crate::policy::effort::ThinkingMode::Thinking {
1264 for (k, v) in crate::policy::effort::thinking_on_kwargs() {
1265 kwargs.entry(k).or_insert(v);
1266 }
1267 }
1268 match crate::policy::effort::sanitize_effort(&mut kwargs, template.efforts()) {
1269 crate::policy::effort::EffortMapping::Mapped(to) => {
1270 tracing::debug!("reasoning_effort quantized to {}", to.as_str());
1271 }
1272 crate::policy::effort::EffortMapping::Dropped => {
1273 tracing::debug!(
1274 "reasoning_effort dropped: this checkpoint's template grades no gear close \
1275 enough, so its own default applies"
1276 );
1277 }
1278 crate::policy::effort::EffortMapping::Unchanged => {}
1279 }
1280 crate::policy::effort::broadcast_effort_spellings(&mut kwargs);
1281 kwargs
1282 }
1283
1284 fn validate_supported_fields(&self) -> Result<(), ApiError> {
1287 if self.max_tokens == 0 {
1294 return Err(invalid_request(
1295 "max_tokens must be at least 1",
1296 "max_tokens",
1297 ));
1298 }
1299 if let Some(switch) = &self.thinking {
1304 let kind = switch.kind.trim().to_ascii_lowercase();
1305 if kind != "enabled" && kind != "disabled" {
1306 return Err(invalid_request(
1307 "thinking.type must be \"enabled\" or \"disabled\"",
1308 "thinking.type",
1309 ));
1310 }
1311 }
1312 for msg in &self.messages {
1313 if msg.content.as_ref().is_some_and(MessageContent::has_image) {
1314 return Err(unsupported_feature(
1315 "image_url content parts are not implemented (multimodal/VL deferred, see docs/API.md)",
1316 ));
1317 }
1318 }
1319 if self.logprobs == Some(true) || self.top_logprobs.is_some() {
1320 return Err(unsupported_feature(
1321 "logprobs / top_logprobs are not implemented yet (see docs/API.md)",
1322 ));
1323 }
1324 if self.n.is_some_and(|n| n > 1) {
1325 return Err(unsupported_feature(
1326 "n > 1 is not implemented (single completion only)",
1327 ));
1328 }
1329 unsupported_sampling::refuse_logit_bias(self.logit_bias.as_ref(), "/v1/chat/completions")?;
1330 unsupported_sampling::parse_sampler_order(self.samplers.as_ref(), "/v1/chat/completions")?;
1334 let stated_grammar =
1349 grammar_request::for_request(self.grammar.as_deref(), self.response_format.as_ref())?;
1350 if let Some(forced) = self.forced_tool_choice()? {
1358 if self.tools.is_empty() {
1359 return Err(invalid_request(
1360 "tool_choice forces a tool call, but no tools were offered",
1361 "tool_choice",
1362 ));
1363 }
1364 if let tool_grammar::Forced::Named(name) = forced {
1365 if !self.tools.iter().any(|t| t.function.name == name) {
1366 return Err(invalid_request(
1367 &format!(
1368 "tool_choice names {name:?}, which is not one of the tools offered"
1369 ),
1370 "tool_choice",
1371 ));
1372 }
1373 }
1374 if stated_grammar.is_some() {
1384 return Err(invalid_request(
1385 "a forced tool_choice and a \"grammar\" or response_format \"json_schema\" \
1386 are two different constraints on the same generation; send one",
1387 "tool_choice",
1388 ));
1389 }
1390 if self.json_object_mode() {
1391 return Err(invalid_request(
1392 "a forced tool_choice cannot be combined with response_format json_object: \
1393 the tool-call markers are not JSON",
1394 "tool_choice",
1395 ));
1396 }
1397 }
1398 Ok(())
1399 }
1400
1401 fn effective_stop_sequences(&self) -> Vec<String> {
1408 let mut stop = self.stop_sequences();
1409 if self.tools_active() {
1410 stop.push("</tool_call>".to_string());
1411 }
1412 stop
1413 }
1414
1415 fn json_object_mode(&self) -> bool {
1416 self.response_format
1417 .as_ref()
1418 .and_then(|v| v.get("type"))
1419 .and_then(|v| v.as_str())
1420 == Some("json_object")
1421 }
1422}
1423
1424#[derive(Serialize)]
1425struct ChatCompletionChoice {
1426 index: usize,
1427 message: ChatCompletionResponseMessage,
1428 finish_reason: &'static str,
1429}
1430
1431#[derive(Serialize)]
1432struct ChatCompletionResponseMessage {
1433 role: &'static str,
1434 #[serde(skip_serializing_if = "Option::is_none")]
1435 content: Option<String>,
1436 #[serde(skip_serializing_if = "Option::is_none")]
1440 reasoning_content: Option<String>,
1441 #[serde(skip_serializing_if = "Option::is_none")]
1442 tool_calls: Option<Vec<ToolCallOut>>,
1443}
1444
1445#[derive(Serialize, Clone)]
1446struct ToolCallOut {
1447 id: String,
1448 #[serde(rename = "type")]
1449 kind: &'static str,
1450 function: ToolCallFunctionOut,
1451}
1452
1453#[derive(Serialize, Clone)]
1461struct ToolCallDelta {
1462 index: usize,
1463 #[serde(skip_serializing_if = "Option::is_none")]
1464 id: Option<String>,
1465 #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
1466 kind: Option<&'static str>,
1467 function: ToolCallFunctionDelta,
1468}
1469
1470#[derive(Serialize, Clone, Default)]
1471struct ToolCallFunctionDelta {
1472 #[serde(skip_serializing_if = "Option::is_none")]
1473 name: Option<String>,
1474 #[serde(skip_serializing_if = "Option::is_none")]
1477 arguments: Option<String>,
1478}
1479
1480impl ToolCallDelta {
1481 fn whole(index: usize, name: String, arguments: String) -> Self {
1483 ToolCallDelta {
1484 index,
1485 id: Some(format!("call_{index}")),
1486 kind: Some("function"),
1487 function: ToolCallFunctionDelta {
1488 name: Some(name),
1489 arguments: Some(arguments),
1490 },
1491 }
1492 }
1493
1494 fn opening(index: usize, name: String) -> Self {
1496 ToolCallDelta {
1497 index,
1498 id: Some(format!("call_{index}")),
1499 kind: Some("function"),
1500 function: ToolCallFunctionDelta {
1501 name: Some(name),
1502 arguments: Some(String::new()),
1503 },
1504 }
1505 }
1506
1507 fn arguments(index: usize, fragment: String) -> Self {
1509 ToolCallDelta {
1510 index,
1511 id: None,
1512 kind: None,
1513 function: ToolCallFunctionDelta {
1514 name: None,
1515 arguments: Some(fragment),
1516 },
1517 }
1518 }
1519}
1520
1521#[derive(Serialize, Clone)]
1522struct ToolCallFunctionOut {
1523 name: String,
1524 arguments: String,
1528}
1529
1530#[derive(Serialize)]
1531struct ChatCompletionResponse {
1532 id: String,
1533 request_id: String,
1539 object: &'static str,
1540 model: String,
1541 choices: Vec<ChatCompletionChoice>,
1542 usage: generate::Usage,
1547 ferrox_cache: &'static str,
1558}
1559
1560#[derive(Serialize)]
1561struct ChatCompletionChunkDelta {
1562 #[serde(skip_serializing_if = "Option::is_none")]
1563 role: Option<&'static str>,
1564 #[serde(skip_serializing_if = "Option::is_none")]
1565 content: Option<String>,
1566 #[serde(skip_serializing_if = "Option::is_none")]
1568 reasoning_content: Option<String>,
1569 #[serde(skip_serializing_if = "Option::is_none")]
1570 tool_calls: Option<Vec<ToolCallDelta>>,
1571}
1572
1573#[derive(Serialize)]
1574struct ChatCompletionChunkChoice {
1575 index: usize,
1576 delta: ChatCompletionChunkDelta,
1577 finish_reason: Option<&'static str>,
1578}
1579
1580#[derive(Serialize)]
1581struct ChatCompletionChunk {
1582 id: String,
1583 #[serde(skip_serializing_if = "Option::is_none")]
1590 request_id: Option<String>,
1591 object: &'static str,
1592 model: String,
1593 choices: Vec<ChatCompletionChunkChoice>,
1594 #[serde(skip_serializing_if = "Option::is_none")]
1597 usage: Option<generate::Usage>,
1598}
1599
1600async fn health(State(state): State<Arc<AppState>>) -> Response {
1606 let snapshot = state.detection.snapshot();
1607 let mut capabilities = snapshot.capabilities;
1608 let active = state.active();
1609
1610 capabilities.push(match active.as_deref() {
1613 None => ferrox_api::Capability::unavailable(
1618 ferrox_api::health::capability::REAL_WEIGHTS,
1619 ferrox_api::health::reason::MODEL_NOT_LOADED,
1620 "No model is loaded. POST /admin/models/load with an id from GET /admin/models.",
1621 ),
1622 Some(active) if active.is_synthetic() => ferrox_api::Capability::unavailable(
1623 ferrox_api::health::capability::REAL_WEIGHTS,
1624 ferrox_api::health::reason::MODEL_NOT_LOADED,
1625 "Serving synthetic random weights: set FERROX_MODEL_PATH (or -m) to a real \
1626 checkpoint. Output from this model is noise.",
1627 ),
1628 Some(active) if active.encoder().is_some() => {
1638 let endpoints = active
1639 .encoder()
1640 .map(|e| encoder_endpoints(e))
1641 .unwrap_or_default();
1642 let served_by = match endpoints.is_empty() {
1643 true => "no endpoint in this build serves it".to_string(),
1644 false => format!("served by {}", endpoints.join(" and ")),
1645 };
1646 ferrox_api::Capability::available(
1647 ferrox_api::health::capability::REAL_WEIGHTS,
1648 format!(
1649 "Serving the real embedding checkpoint '{}'. This is an ENCODER, \
1650 {served_by}; generation endpoints refuse it.",
1651 active.name(),
1652 ),
1653 )
1654 }
1655 Some(active) => ferrox_api::Capability::available(
1656 ferrox_api::health::capability::REAL_WEIGHTS,
1657 format!("Serving the real checkpoint '{}'.", active.name()),
1658 ),
1659 });
1660 capabilities.push(if active.as_ref().is_some_and(|a| a.batcher.is_some()) {
1661 ferrox_api::Capability::available(
1662 ferrox_api::health::capability::CONTINUOUS_BATCHING,
1663 if state.continuous_batching_enabled && continuous_batching_env().is_none() {
1664 "On by default on Metal. Concurrent requests share one batched decode worker."
1665 } else {
1666 "Concurrent requests share one batched decode step."
1667 },
1668 )
1669 } else if state.metal_private_decode_gate.is_some() {
1670 ferrox_api::Capability::unavailable(
1671 ferrox_api::health::capability::CONTINUOUS_BATCHING,
1672 ferrox_api::health::reason::DISABLED,
1673 "Off; private Metal decodes serialize (one at a time). Set FERROX_CONTINUOUS_BATCHING=1 or --cont-batching for parallel serving.",
1674 )
1675 } else {
1676 ferrox_api::Capability::unavailable(
1677 ferrox_api::health::capability::CONTINUOUS_BATCHING,
1678 ferrox_api::health::reason::DISABLED,
1679 "Off; set FERROX_CONTINUOUS_BATCHING=1 (incompatible with a KV pool or prefix cache).",
1680 )
1681 });
1682
1683 let last_request_ms = state
1684 .last_request_ms
1685 .load(std::sync::atomic::Ordering::Relaxed);
1686 let uptime = state.started_at.elapsed();
1687 let health_state = if active.is_none() {
1695 ferrox_api::HealthState::Unavailable
1696 } else {
1697 snapshot.state
1698 };
1699 let body = ferrox_api::HealthResponse {
1700 state: health_state,
1701 reason: match health_state {
1702 ferrox_api::HealthState::Ready => None,
1703 ferrox_api::HealthState::Unavailable => {
1704 Some(ferrox_api::health::reason::MODEL_NOT_LOADED.to_string())
1705 }
1706 ferrox_api::HealthState::Detecting => {
1707 Some(ferrox_api::health::reason::DETECTING.to_string())
1708 }
1709 },
1710 detail: match health_state {
1711 ferrox_api::HealthState::Ready => None,
1712 ferrox_api::HealthState::Unavailable => Some(
1713 "No model is loaded. POST /admin/models/load with an id from GET /admin/models."
1714 .to_string(),
1715 ),
1716 ferrox_api::HealthState::Detecting => {
1717 Some("Probing available compute backends.".to_string())
1718 }
1719 },
1720 model: active
1721 .as_deref()
1722 .map(|active| ferrox_api::health::ModelSummary {
1723 id: active.name().to_string(),
1724 tokenizer: active.tokenizer_kind().to_string(),
1725 synthetic_weights: active.is_synthetic(),
1726 }),
1727 capabilities,
1728 version: env!("CARGO_PKG_VERSION").to_string(),
1729 pid: std::process::id(),
1730 uptime_seconds: uptime.as_secs_f64(),
1731 server_time_unix_ms: std::time::SystemTime::now()
1732 .duration_since(std::time::UNIX_EPOCH)
1733 .map(|d| d.as_millis().min(u64::MAX as u128) as u64)
1734 .unwrap_or(0),
1735 last_request_age_seconds: (last_request_ms > 0)
1736 .then(|| uptime.as_secs_f64() - (last_request_ms as f64 / 1000.0))
1737 .map(|age| age.max(0.0)),
1738 };
1739
1740 let status =
1741 StatusCode::from_u16(body.state.http_status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
1742 (status, Json(body)).into_response()
1743}
1744
1745async fn list_models(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
1746 let Some(active) = state.active() else {
1750 return Json(serde_json::json!({ "object": "list", "data": [] }));
1751 };
1752 let mut model_entry = serde_json::json!({
1753 "id": active.name(),
1754 "object": "model",
1755 "ferrox_synthetic_weights": active.is_synthetic(),
1756 "ferrox_tokenizer": active.tokenizer_kind(),
1757 });
1758 if let Some(encoder) = active.encoder() {
1764 model_entry["ferrox_model_kind"] = serde_json::json!("embedding");
1765 model_entry["ferrox_endpoints"] = serde_json::json!(encoder_endpoints(encoder));
1766 model_entry["ferrox_n_embd"] = serde_json::json!(encoder.n_embd());
1767 model_entry["ferrox_pooling"] = serde_json::json!(encoder.pooling_type().name());
1768 model_entry["ferrox_context_length"] = serde_json::json!(encoder.n_ctx_train());
1769 }
1770 if let Some(model) = active.generative_opt() {
1778 let parser_configured =
1779 crate::policy::parser::ReasoningFormat::infer(active.name()).is_some();
1780 let gears = model.chat_template().think_gears(parser_configured);
1781 if !gears.is_empty() {
1782 model_entry["supported_reasoning_efforts"] = serde_json::json!(gears.supported);
1783 if let Some(default) = &gears.default {
1784 model_entry["default_reasoning_effort"] = serde_json::json!(default);
1785 }
1786 model_entry["reasoning_effort_kwargs"] = serde_json::json!(gears.kwargs);
1789 }
1790 }
1791 if let Some(mcp) = &state.mcp {
1792 model_entry["ferrox_mcp"] = mcp.models_metadata();
1793 }
1794 Json(serde_json::json!({
1795 "object": "list",
1796 "data": [model_entry]
1797 }))
1798}
1799
1800async fn serving_stats(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
1814 let now_ms = state.uptime().as_millis().min(u64::MAX as u128) as u64;
1815 let mut serving = state.serving.lock().unwrap_or_else(|p| p.into_inner());
1816 let active = state.active();
1817 Json(serde_json::json!({
1818 "model": active.as_ref().map(|a| a.name()),
1819 "state": state
1820 .maintenance
1821 .lock()
1822 .unwrap_or_else(|p| p.into_inner())
1823 .state()
1824 .as_str(),
1825 "uptime_s": state.uptime().as_secs(),
1826 "throughput": {
1827 "decode_tps": (serving.decode_tokens_per_second(now_ms) * 10.0).round() / 10.0,
1828 "prefill_tps": (serving.prefill_tokens_per_second(now_ms) * 10.0).round() / 10.0,
1829 },
1830 "requests": {
1831 "active": state.cancels.live_count(),
1832 "completed": state.stats.recorded_total(),
1833 "p95_ms": state.stats.p95_duration_ms(),
1834 "ttft_mean_ms": state.stats.ttft_mean_ms(),
1835 "prompt_tokens_total": state.stats.tokens_prompt_total(),
1836 "completion_tokens_total": state.stats.tokens_generated_total(),
1837 },
1838 "pools": cache_admin::pool_gauges(&state),
1844 "memory": cache_admin::footprint_json(&state),
1847 }))
1848}
1849
1850#[derive(Deserialize)]
1851struct RequestsQuery {
1852 #[serde(default)]
1853 since: u64,
1854 #[serde(default = "default_requests_limit")]
1855 limit: usize,
1856}
1857
1858fn default_requests_limit() -> usize {
1859 stats::MAX_PAGE
1860}
1861
1862async fn recent_requests(
1870 State(state): State<Arc<AppState>>,
1871 axum::extract::Query(q): axum::extract::Query<RequestsQuery>,
1872) -> Json<serde_json::Value> {
1873 let (rows, cursor, missed) = state.stats.page(q.since, q.limit);
1874 Json(serde_json::json!({
1875 "requests": rows,
1876 "next_cursor": cursor,
1877 "missed": missed,
1878 "total": state.stats.recorded_total(),
1879 }))
1880}
1881
1882#[derive(Serialize)]
1883struct CombinedCacheStats {
1884 response_cache: response_cache::CacheStats,
1885 prefix_cache: Option<ferrox_models::PrefixCacheStats>,
1887}
1888
1889async fn cache_stats(State(state): State<Arc<AppState>>) -> Json<CombinedCacheStats> {
1890 Json(CombinedCacheStats {
1891 response_cache: lock_cache(&state.response_cache).stats(),
1892 prefix_cache: state
1893 .prefix_cache
1894 .as_ref()
1895 .map(|pc| pc.lock().unwrap_or_else(|p| p.into_inner()).stats()),
1896 })
1897}
1898
1899async fn metrics(State(state): State<Arc<AppState>>) -> Response {
1904 use std::sync::atomic::Ordering;
1905
1906 let cache_stats = lock_cache(&state.response_cache).stats();
1907 let active = state.active();
1908 let requests_total = state.requests_total.load(Ordering::Relaxed);
1909 let errors_total = state.request_errors_total.load(Ordering::Relaxed);
1910 let uptime = state.started_at.elapsed().as_secs_f64();
1911
1912 let body = format!(
1913 "# HELP ferrox_requests_total Total chat completion requests received.\n\
1914 # TYPE ferrox_requests_total counter\n\
1915 ferrox_requests_total {requests_total}\n\
1916 # HELP ferrox_request_errors_total Total chat completion requests that returned an error.\n\
1917 # TYPE ferrox_request_errors_total counter\n\
1918 ferrox_request_errors_total {errors_total}\n\
1919 # HELP ferrox_cache_hits_total Whole-response cache hits.\n\
1920 # TYPE ferrox_cache_hits_total counter\n\
1921 ferrox_cache_hits_total {}\n\
1922 # HELP ferrox_cache_misses_total Whole-response cache misses.\n\
1923 # TYPE ferrox_cache_misses_total counter\n\
1924 ferrox_cache_misses_total {}\n\
1925 # HELP ferrox_cache_entries Current whole-response cache entry count.\n\
1926 # TYPE ferrox_cache_entries gauge\n\
1927 ferrox_cache_entries {}\n\
1928 # HELP ferrox_synthetic_weights 1 if serving synthetic random weights instead of a real checkpoint.\n\
1929 # TYPE ferrox_synthetic_weights gauge\n\
1930 ferrox_synthetic_weights {}\n\
1931 # HELP ferrox_uptime_seconds Seconds since this server process started.\n\
1932 # TYPE ferrox_uptime_seconds gauge\n\
1933 ferrox_uptime_seconds {uptime}\n",
1934 cache_stats.hits,
1935 cache_stats.misses,
1936 cache_stats.entries,
1937 active
1941 .as_ref()
1942 .map(|a| a.is_synthetic() as u8)
1943 .unwrap_or(0),
1944 );
1945
1946 let body = match active
1950 .as_ref()
1951 .and_then(|a| a.expert_store_stats())
1952 {
1953 Some(es) => format!(
1954 "{body}\
1955 # HELP ferrox_expert_cache_hits_total Expert-store cache hits.\n\
1956 # TYPE ferrox_expert_cache_hits_total counter\n\
1957 ferrox_expert_cache_hits_total {}\n\
1958 # HELP ferrox_expert_cache_misses_total Expert-store cache misses (source reads).\n\
1959 # TYPE ferrox_expert_cache_misses_total counter\n\
1960 ferrox_expert_cache_misses_total {}\n\
1961 # HELP ferrox_expert_cache_evictions_total Expert-store LRU evictions.\n\
1962 # TYPE ferrox_expert_cache_evictions_total counter\n\
1963 ferrox_expert_cache_evictions_total {}\n\
1964 # HELP ferrox_expert_cache_pass_throughs_total Acquires served uncached (entry could not fit the budget).\n\
1965 # TYPE ferrox_expert_cache_pass_throughs_total counter\n\
1966 ferrox_expert_cache_pass_throughs_total {}\n\
1967 # HELP ferrox_expert_cache_bytes_read_total Bytes read from the checkpoint for expert misses.\n\
1968 # TYPE ferrox_expert_cache_bytes_read_total counter\n\
1969 ferrox_expert_cache_bytes_read_total {}\n\
1970 # HELP ferrox_expert_cache_resident_bytes Current expert-cache footprint in bytes.\n\
1971 # TYPE ferrox_expert_cache_resident_bytes gauge\n\
1972 ferrox_expert_cache_resident_bytes {}\n",
1973 es.hits, es.misses, es.evictions, es.pass_throughs, es.bytes_read, es.resident_bytes,
1974 ),
1975 None => body,
1976 };
1977
1978 let body = match active.as_ref().and_then(|a| a.batcher.as_ref()) {
1983 Some(batcher) => {
1984 let sched = batcher.stats();
1985 format!(
1986 "{body}\
1987 # HELP ferrox_prefill_chunks_total Bounded prefill chunks the batch scheduler has run.\n\
1988 # TYPE ferrox_prefill_chunks_total counter\n\
1989 ferrox_prefill_chunks_total {}\n\
1990 # HELP ferrox_prefill_tokens_total Prompt tokens run through chunked prefill.\n\
1991 # TYPE ferrox_prefill_tokens_total counter\n\
1992 ferrox_prefill_tokens_total {}\n\
1993 # HELP ferrox_decode_steps_total Batched decode steps the batch scheduler has run.\n\
1994 # TYPE ferrox_decode_steps_total counter\n\
1995 ferrox_decode_steps_total {}\n\
1996 # HELP ferrox_scheduler_queue_depth Requests waiting for admission to the batch scheduler.\n\
1997 # TYPE ferrox_scheduler_queue_depth gauge\n\
1998 ferrox_scheduler_queue_depth {}\n\
1999 # HELP ferrox_scheduler_queue_rejected_total Requests refused with 503 because the admission queue was full.\n\
2000 # TYPE ferrox_scheduler_queue_rejected_total counter\n\
2001 ferrox_scheduler_queue_rejected_total {}\n\
2002 # HELP ferrox_kv_blocks_total KV blocks in the scheduler's admission budget (0 when unconfigured).\n\
2003 # TYPE ferrox_kv_blocks_total gauge\n\
2004 ferrox_kv_blocks_total {}\n\
2005 # HELP ferrox_kv_blocks_free KV blocks not reserved by an in-flight request.\n\
2006 # TYPE ferrox_kv_blocks_free gauge\n\
2007 ferrox_kv_blocks_free {}\n\
2008 # HELP ferrox_kv_block_size Token positions per KV block.\n\
2009 # TYPE ferrox_kv_block_size gauge\n\
2010 ferrox_kv_block_size {}\n\
2011 # HELP ferrox_kv_rejected_too_large_total Requests refused with 400 because they exceed the whole KV block budget.\n\
2012 # TYPE ferrox_kv_rejected_too_large_total counter\n\
2013 ferrox_kv_rejected_too_large_total {}\n\
2014 # HELP ferrox_kv_rejected_context_length_total Requests refused with 400 for exceeding the per-request context ceiling.\n\
2015 # TYPE ferrox_kv_rejected_context_length_total counter\n\
2016 ferrox_kv_rejected_context_length_total {}\n\
2017 # HELP ferrox_scheduler_aborted_total Requests the batch scheduler stopped because they were cancelled.\n\
2018 # TYPE ferrox_scheduler_aborted_total counter\n\
2019 ferrox_scheduler_aborted_total {}\n\
2020 # HELP ferrox_scheduler_max_seqs Cap on in-flight sequences (-np / FERROX_CB_MAX_SEQS); 0 when unlimited.\n\
2021 # TYPE ferrox_scheduler_max_seqs gauge\n\
2022 ferrox_scheduler_max_seqs {}\n\
2023 # HELP ferrox_scheduler_prefill_chunk Prompt tokens per prefill chunk (-b / -ub / FERROX_CB_PREFILL_CHUNK).\n\
2024 # TYPE ferrox_scheduler_prefill_chunk gauge\n\
2025 ferrox_scheduler_prefill_chunk {}\n",
2026 sched.prefill_chunks,
2027 sched.prefill_tokens,
2028 sched.decode_steps,
2029 sched.queue_depth,
2030 sched.queue_rejected,
2031 sched.kv_blocks_total,
2032 sched.kv_blocks_free,
2033 sched.kv_block_size,
2034 sched.kv_rejected_too_large,
2035 sched.kv_rejected_context_length,
2036 sched.aborted,
2037 sched.max_seqs,
2038 sched.prefill_chunk,
2039 )
2040 }
2041 None => body,
2042 };
2043
2044 (
2045 [(
2046 axum::http::header::CONTENT_TYPE,
2047 "text/plain; version=0.0.4",
2048 )],
2049 body,
2050 )
2051 .into_response()
2052}
2053
2054pub(crate) type ApiError = (StatusCode, Json<serde_json::Value>);
2055
2056pub(crate) fn invalid_request(message: &str, param: &str) -> ApiError {
2062 (
2063 StatusCode::BAD_REQUEST,
2064 Json(serde_json::json!({"error": {
2065 "message": message,
2066 "type": "invalid_request_error",
2067 "param": param,
2068 "code": null,
2069 }})),
2070 )
2071}
2072
2073pub(crate) fn unsupported_feature(message: &str) -> ApiError {
2074 (
2075 StatusCode::NOT_IMPLEMENTED,
2076 Json(serde_json::json!({"error": {"message": message, "type": "unsupported"}})),
2077 )
2078}
2079
2080pub(crate) fn decode_error_response(e: generate::DecodeError) -> ApiError {
2081 let status = match e {
2082 generate::DecodeError::TokenOutOfVocab { .. } => StatusCode::BAD_REQUEST,
2083 generate::DecodeError::KvBudgetExceeded { .. } => StatusCode::BAD_REQUEST,
2088 generate::DecodeError::KvPoolExhausted | generate::DecodeError::QueueFull { .. } => {
2094 StatusCode::SERVICE_UNAVAILABLE
2095 }
2096 generate::DecodeError::GrammarConstraint { .. } => StatusCode::BAD_REQUEST,
2100 generate::DecodeError::ReasoningBudget { .. } => StatusCode::INTERNAL_SERVER_ERROR,
2104 };
2105 tracing::warn!("decode error: {e}");
2106 let mut body = serde_json::json!({"error": {"message": e.to_string()}});
2107 if let generate::DecodeError::KvBudgetExceeded {
2113 binding,
2114 estimated_bytes,
2115 limit_bytes,
2116 positions,
2117 positions_limit,
2118 ..
2119 } = &e
2120 {
2121 body["error"]["type"] = serde_json::json!("invalid_request_error");
2122 body["error"]["code"] = serde_json::json!(binding);
2123 body["error"]["binding"] = serde_json::json!(binding);
2124 body["error"]["estimated_bytes"] = serde_json::json!(estimated_bytes);
2125 body["error"]["limit_bytes"] = serde_json::json!(limit_bytes);
2126 body["error"]["positions"] = serde_json::json!(positions);
2127 body["error"]["positions_limit"] = serde_json::json!(positions_limit);
2128 }
2129 if let Some(secs) = e.retry_after_secs() {
2133 body["error"]["retry_after_seconds"] = serde_json::json!(secs);
2134 }
2135 (status, Json(body))
2136}
2137
2138pub(crate) fn join_error_response(e: tokio::task::JoinError) -> ApiError {
2139 tracing::error!("generation task panicked: {e}");
2140 (
2141 StatusCode::INTERNAL_SERVER_ERROR,
2142 Json(serde_json::json!({"error": {"message": "internal error during generation"}})),
2143 )
2144}
2145
2146#[allow(clippy::too_many_arguments)] fn run_generation_emit(
2156 model: &Model,
2157 prompt: &str,
2158 params: &GenerationParams,
2159 kv_pool: Option<&generate::KvPoolConfig>,
2160 paged_kv: Option<&generate::PagedKvConfig>,
2161 prefix_cache: Option<&Mutex<PrefixCache>>,
2162 continuous_batcher: Option<&serving::batch::ContinuousBatcher>,
2163 ceiling: Option<&budget::ContextCeiling>,
2164 metal_private_decode_gate: Option<&std::sync::Mutex<()>>,
2165 mut emit: impl FnMut(&str),
2166) -> Result<(FinishReason, generate::Usage, String), generate::DecodeError> {
2167 let synthetic = model.is_synthetic();
2168 let mut chunks = Vec::new();
2169 let params = &{
2175 let mut resolved = params.clone();
2176 resolved.stop_token_ids = crate::stop::resolve_stop_tokens(&resolved.stop, |text| {
2177 model.encode(text, SpecialTokens::Parse)
2178 });
2179 resolved.reasoning_budget = resolved
2185 .reasoning_budget
2186 .armed(resolved.reasoning, prompt, |text| {
2187 model.encode(text, SpecialTokens::Parse)
2188 })
2189 .map_err(|detail| generate::DecodeError::ReasoningBudget { detail })?;
2190 resolved
2191 };
2192 let used_batcher = matches!((model, continuous_batcher), (Model::Gguf(_), Some(_)));
2193 let _metal_private_guard =
2194 acquire_metal_private_decode_gate(metal_private_decode_gate, used_batcher);
2195 let (finish, usage) = match model {
2196 Model::Gguf(m) => {
2197 if let Some(batcher) = continuous_batcher {
2198 let mut tokens = m.tokenizer.encode(prompt, SpecialTokens::Parse);
2199 ferrox_models::tokenizer::prepend_bos(&mut tokens, m.bos_id);
2200 let (finish, _generated_ids, text, usage) = if synthetic {
2201 batcher.generate(tokens, params.clone(), m.stop_tokens.clone())?
2202 } else {
2203 batcher.generate_streaming(
2204 tokens,
2205 params.clone(),
2206 m.stop_tokens.clone(),
2207 Some(|chunk: &str| {
2208 if !chunk.is_empty() {
2209 chunks.push(chunk.to_string());
2210 emit(chunk);
2211 }
2212 }),
2213 )?
2214 };
2215 if !text.is_empty() && chunks.is_empty() {
2216 chunks.push(text);
2217 }
2218 (finish, usage)
2219 } else {
2220 generate::generate(
2221 &m.decoder,
2222 m.tokenizer.as_ref(),
2223 &m.stop_tokens,
2224 m.bos_id,
2225 prompt,
2226 params,
2227 kv_pool,
2228 paged_kv,
2229 prefix_cache,
2230 ceiling,
2231 |chunk| {
2232 chunks.push(chunk.to_string());
2233 if !synthetic {
2234 emit(chunk);
2235 }
2236 },
2237 )?
2238 }
2239 }
2240 Model::Kimi(m) => generate::generate_engine(
2241 &m.engine,
2242 &m.tokenizer,
2243 &m.stop_tokens,
2244 None,
2245 prompt,
2246 params,
2247 |chunk| {
2248 chunks.push(chunk.to_string());
2249 if !synthetic {
2250 emit(chunk);
2251 }
2252 },
2253 )?,
2254 Model::Mla(m) => generate::generate_engine(
2255 &m.engine,
2256 &m.tokenizer,
2257 &m.stop_tokens,
2258 m.bos_id,
2259 prompt,
2260 params,
2261 |chunk| {
2262 chunks.push(chunk.to_string());
2263 if !synthetic {
2264 emit(chunk);
2265 }
2266 },
2267 )?,
2268 Model::Gemma4(m) => generate::generate_engine(
2269 &m.engine,
2270 &m.tokenizer,
2271 &m.stop_tokens,
2272 m.bos_id,
2273 prompt,
2274 params,
2275 |chunk| {
2276 chunks.push(chunk.to_string());
2277 if !synthetic {
2278 emit(chunk);
2279 }
2280 },
2281 )?,
2282 Model::Glm52(m) => generate::generate_engine(
2283 &m.engine,
2284 &m.tokenizer,
2285 &m.stop_tokens,
2286 m.bos_id,
2287 prompt,
2288 params,
2289 |chunk| {
2290 chunks.push(chunk.to_string());
2291 if !synthetic {
2292 emit(chunk);
2293 }
2294 },
2295 )?,
2296 };
2297
2298 let mut full = chunks.concat();
2299 if synthetic {
2300 full = format!(
2301 "[ferrox synthetic-weight demo: no real checkpoint loaded -- set FERROX_MODEL_PATH \
2302 to serve a real model. Decoded ids -> {full:?}]"
2303 );
2304 emit(&full);
2305 } else if used_batcher && !full.is_empty() && chunks.is_empty() {
2306 emit(&full);
2307 }
2308
2309 Ok((finish, usage, full))
2310}
2311
2312#[allow(clippy::too_many_arguments)] pub(crate) fn run_generation(
2317 model: &Model,
2318 prompt: &str,
2319 params: &GenerationParams,
2320 kv_pool: Option<&generate::KvPoolConfig>,
2321 paged_kv: Option<&generate::PagedKvConfig>,
2322 prefix_cache: Option<&Mutex<PrefixCache>>,
2323 continuous_batcher: Option<&serving::batch::ContinuousBatcher>,
2324 ceiling: Option<&budget::ContextCeiling>,
2325 metal_private_decode_gate: Option<&std::sync::Mutex<()>>,
2326) -> Result<(Vec<String>, FinishReason, generate::Usage), generate::DecodeError> {
2327 let (finish, usage, full) = run_generation_emit(
2328 model,
2329 prompt,
2330 params,
2331 kv_pool,
2332 paged_kv,
2333 prefix_cache,
2334 continuous_batcher,
2335 ceiling,
2336 metal_private_decode_gate,
2337 |_| {},
2338 )?;
2339 Ok((
2340 if full.is_empty() {
2341 Vec::new()
2342 } else {
2343 vec![full]
2344 },
2345 finish,
2346 usage,
2347 ))
2348}
2349
2350pub(crate) fn prompt_from_messages(
2361 messages: &[ChatMessage],
2362 template: &chat_template::PromptTemplate,
2363 tools: &[ToolDef],
2364 extra: serde_json::Map<String, serde_json::Value>,
2365) -> Result<String, ApiError> {
2366 let rendered = if tools.is_empty() || template.handles_tools() {
2367 template.render(messages, tools, extra)
2368 } else {
2369 let mut with_preamble = Vec::with_capacity(messages.len() + 1);
2370 with_preamble.push(ChatMessage {
2371 role: "system".to_string(),
2372 content: Some(MessageContent::Text(tool_preamble(tools))),
2373 tool_calls: None,
2374 tool_call_id: None,
2375 reasoning_content: None,
2376 });
2377 with_preamble.extend_from_slice(messages);
2378 template.render(&with_preamble, &[], extra)
2379 };
2380 rendered.map_err(template_error_response)
2381}
2382
2383fn template_error_response(err: ferrox_models::chat_template::TemplateError) -> ApiError {
2388 (
2389 StatusCode::BAD_REQUEST,
2390 Json(serde_json::json!({
2391 "error": {
2392 "message": format!("chat template failed to render: {err}"),
2393 "type": "invalid_request_error",
2394 "param": "messages",
2395 "code": null,
2396 }
2397 })),
2398 )
2399}
2400
2401fn tool_preamble(tools: &[ToolDef]) -> String {
2414 let mut out = String::from(
2415 "You can call tools to help answer the user. To call a tool, respond with \
2416 EXACTLY one line in this format and nothing else:\n\
2417 <tool_call>{\"name\": \"<tool name>\", \"arguments\": {<arguments as a JSON \
2418 object matching that tool's parameters>}}</tool_call>\n\n\
2419 Available tools:\n",
2420 );
2421 for t in tools {
2422 out.push_str(&format!(
2423 "- {}: {}\n parameters (JSON schema): {}\n",
2424 t.function.name,
2425 t.function.description.as_deref().unwrap_or(""),
2426 t.function
2427 .parameters
2428 .as_ref()
2429 .map(|v| v.to_string())
2430 .unwrap_or_else(|| "{}".to_string()),
2431 ));
2432 }
2433 out
2434}
2435
2436fn tool_call_deltas(
2446 events: Vec<crate::policy::parser::ToolCallEvent>,
2447 opened: &std::cell::Cell<usize>,
2448) -> (String, Vec<ToolCallDelta>) {
2449 let mut text = String::new();
2450 let mut deltas = Vec::new();
2451 for event in events {
2452 match event {
2453 crate::policy::parser::ToolCallEvent::Text(chunk) => text.push_str(&chunk),
2454 crate::policy::parser::ToolCallEvent::CallStart { index, name } => {
2455 opened.set(opened.get().max(index + 1));
2456 deltas.push(ToolCallDelta::opening(index, name));
2457 }
2458 crate::policy::parser::ToolCallEvent::CallArguments { index, fragment } => {
2459 if !fragment.is_empty() {
2460 deltas.push(ToolCallDelta::arguments(index, fragment));
2461 }
2462 }
2463 crate::policy::parser::ToolCallEvent::CallEnd { .. } => {}
2464 }
2465 }
2466 (text, deltas)
2467}
2468
2469fn build_response_message(
2480 text: String,
2481 tools: &[ToolDef],
2482 posture: output::OutputPosture,
2483 base_finish: &'static str,
2484) -> (ChatCompletionResponseMessage, &'static str) {
2485 let parsed = output::parse_output(&text, tools, posture);
2486 let calls: Vec<ToolCallOut> = parsed
2487 .calls
2488 .into_iter()
2489 .enumerate()
2490 .map(|(index, call)| ToolCallOut {
2491 id: format!("call_{index}"),
2492 kind: "function",
2493 function: ToolCallFunctionOut {
2494 name: call.name,
2495 arguments: call.arguments,
2496 },
2497 })
2498 .collect();
2499 if !calls.is_empty() {
2500 return (
2501 ChatCompletionResponseMessage {
2502 role: "assistant",
2503 content: None,
2504 reasoning_content: parsed.reasoning,
2505 tool_calls: Some(calls),
2506 },
2507 "tool_calls",
2508 );
2509 }
2510 (
2511 ChatCompletionResponseMessage {
2512 role: "assistant",
2513 content: Some(parsed.content),
2514 reasoning_content: parsed.reasoning,
2515 tool_calls: None,
2516 },
2517 base_finish,
2518 )
2519}
2520
2521fn resolve_history(state: &AppState, req: &ChatCompletionRequest) -> Vec<ChatMessage> {
2526 let mut history = match &req.session_id {
2527 Some(id) => state.sessions.extend_and_get(id, &req.messages),
2528 None => req.messages.clone(),
2529 };
2530 if req.json_object_mode() {
2531 inject_json_object_system_hint(&mut history);
2532 }
2533 history
2534}
2535
2536fn inject_json_object_system_hint(messages: &mut Vec<ChatMessage>) {
2537 const HINT: &str =
2538 "You must respond with valid JSON only (a single JSON object, no markdown fences).";
2539 if let Some(sys) = messages.iter_mut().find(|m| m.role == "system") {
2540 match &mut sys.content {
2541 Some(MessageContent::Text(s)) if !s.contains("JSON") => {
2542 s.push_str("\n\n");
2543 s.push_str(HINT);
2544 }
2545 None => {
2546 sys.content = Some(MessageContent::Text(HINT.to_string()));
2547 }
2548 _ => {}
2549 }
2550 } else {
2551 messages.insert(
2552 0,
2553 ChatMessage {
2554 role: "system".to_string(),
2555 content: Some(MessageContent::Text(HINT.to_string())),
2556 tool_calls: None,
2557 tool_call_id: None,
2558 reasoning_content: None,
2559 },
2560 );
2561 }
2562}
2563
2564async fn chat_completions(
2565 State(state): State<Arc<AppState>>,
2566 headers: axum::http::HeaderMap,
2567 Json(req): Json<ChatCompletionRequest>,
2568) -> Response {
2569 let attribution = attribution::Attribution::from_headers(&headers);
2570 state
2571 .requests_total
2572 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2573 let started = std::time::Instant::now();
2574
2575 let request_id = ferrox_api::next_request_id();
2579 let stream = req.stream.unwrap_or(false);
2580
2581 let refusal = cache_admin::check_admission(&state)
2587 .err()
2588 .or_else(|| req.validate_supported_fields().err());
2589 if let Some(err) = refusal {
2590 state
2591 .request_errors_total
2592 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2593 let response = err.into_response();
2594 state.record_request(stats::Record {
2595 request_id: &request_id,
2596 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
2597 model: state.active_model_name(),
2598 status: response.status().as_u16(),
2599 stream,
2600 duration_ms: started.elapsed().as_millis() as u64,
2601 usage: None,
2602 attribution: &attribution,
2603 });
2604 return response;
2605 }
2606
2607 let response = if stream {
2608 chat_completions_stream(
2609 Arc::clone(&state),
2610 req,
2611 request_id.clone(),
2612 started,
2613 attribution.clone(),
2614 )
2615 .await
2616 .into_response()
2617 } else {
2618 chat_completions_full(
2619 Arc::clone(&state),
2620 req,
2621 request_id.clone(),
2622 started,
2623 attribution.clone(),
2624 )
2625 .await
2626 .into_response()
2627 };
2628
2629 if response.status().is_client_error() || response.status().is_server_error() {
2630 state
2631 .request_errors_total
2632 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2633 state.record_request(stats::Record {
2637 request_id: &request_id,
2638 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
2639 model: state.active_model_name(),
2642 status: response.status().as_u16(),
2643 stream,
2644 duration_ms: started.elapsed().as_millis() as u64,
2645 usage: None,
2646 attribution: &attribution,
2647 });
2648 }
2649 state.mark_request_finished();
2650
2651 response
2652}
2653
2654async fn chat_completions_full(
2655 state: Arc<AppState>,
2656 req: ChatCompletionRequest,
2657 request_id: String,
2658 started: std::time::Instant,
2659 attribution: attribution::Attribution,
2660) -> Result<Json<ChatCompletionResponse>, ApiError> {
2661 let tools_active = req.tools_active();
2662 let active = state.require_active()?;
2666 let history = resolve_history(&state, &req);
2667 let template = active.generative()?.chat_template();
2668 let kwargs = req.resolve_template_kwargs(&template);
2669 let prompt = req.render_prompt(&history, &template, &req.tools, kwargs, active.name())?;
2670 let params =
2678 req.generation_params_for_template(&template, active.name(), active.sampler_model())?;
2679 let key = req.is_cacheable().then(|| req.cache_key(&prompt, ¶ms));
2680
2681 let (completion, cache_status) = if let Some(cached) = key
2682 .as_ref()
2683 .and_then(|key| lock_cache(&state.response_cache).get(key))
2684 {
2685 tracing::debug!("cache hit for key {}", key.as_ref().unwrap().digest());
2686 (cached, "hit")
2687 } else {
2688 let (chunks, finish, usage) = decode_task::buffered(
2689 decode_task::DecodeHandles::take(&state, &active)?,
2690 prompt.clone(),
2691 params,
2692 )
2693 .await?;
2694
2695 let completion = response_cache::CachedCompletion {
2696 content: chunks.concat(),
2697 finish,
2698 usage,
2699 };
2700 let cache_status = match key {
2706 Some(key) => match completion.clone().cacheable() {
2710 Some(cacheable) => {
2711 tracing::debug!("cache miss for key {}", key.digest());
2712 lock_cache(&state.response_cache).put(key, cacheable);
2713 "miss"
2714 }
2715 None => "skip",
2716 },
2717 None => "skip",
2718 };
2719 (completion, cache_status)
2720 };
2721 let content = completion.content;
2722
2723 if req.json_object_mode() {
2724 json_mode::validate_json_object_output(&content)?;
2725 }
2726
2727 if let Some(id) = &req.session_id {
2731 state.sessions.store_reply(
2732 id,
2733 ChatMessage {
2734 role: "assistant".to_string(),
2735 content: Some(MessageContent::Text(content.clone())),
2736 tool_calls: None,
2737 tool_call_id: None,
2738 reasoning_content: None,
2739 },
2740 );
2741 }
2742
2743 let (message, finish_reason) = build_response_message(
2744 content,
2745 if tools_active { &req.tools } else { &[] },
2746 output::OutputPosture::resolve(active.name(), &prompt),
2747 completion.finish.as_str(),
2748 );
2749
2750 state.record_request(stats::Record {
2751 request_id: &request_id,
2752 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
2753 model: Some(active.name().to_string()),
2756 status: 200,
2757 stream: false,
2758 duration_ms: started.elapsed().as_millis() as u64,
2759 usage: Some(&completion.usage),
2760 attribution: &attribution,
2761 });
2762
2763 Ok(Json(ChatCompletionResponse {
2764 id: request_id.clone(),
2765 request_id,
2766 object: "chat.completion",
2767 model: req.model,
2768 choices: vec![ChatCompletionChoice {
2769 index: 0,
2770 message,
2771 finish_reason,
2772 }],
2773 usage: completion.usage,
2774 ferrox_cache: cache_status,
2775 }))
2776}
2777
2778async fn chat_completions_stream(
2779 state: Arc<AppState>,
2780 req: ChatCompletionRequest,
2781 request_id: String,
2782 started: std::time::Instant,
2783 attribution: attribution::Attribution,
2784) -> Result<Response, ApiError> {
2785 let tools_active = req.tools_active();
2787 let active = state.require_active()?;
2791 let history = resolve_history(&state, &req);
2792 let template = active.generative()?.chat_template();
2793 let kwargs = req.resolve_template_kwargs(&template);
2794 let prompt = req.render_prompt(&history, &template, &req.tools, kwargs, active.name())?;
2795 let model_name = req.model.clone();
2796 let session_id = req.session_id.clone();
2797 let sessions = state.sessions.clone();
2798
2799 let model = Arc::clone(active.generative()?);
2800 let kv_pool = state.kv_pool.clone();
2801 let paged_kv = state.paged_kv.clone();
2802 let prefix_cache = state.prefix_cache.clone();
2803 let batcher = active.batcher.clone();
2804 let ceiling = active.ceiling.clone();
2805 let metal_private_decode_gate = state.metal_private_decode_gate.clone();
2806 let mut params =
2807 req.generation_params_for_template(&template, active.name(), active.sampler_model())?;
2808 let stats_state = Arc::clone(&state);
2809 let served_model = active.name().to_string();
2812 let posture = output::OutputPosture::resolve(&served_model, &prompt);
2816 let offered_tools: Vec<ToolDef> = if tools_active {
2819 req.tools.clone()
2820 } else {
2821 Vec::new()
2822 };
2823
2824 let (cancel_token, cancel_guard) = state.cancels.register(&request_id);
2829 params.cancel = Some(cancel_token.clone());
2830
2831 let overlap = true;
2840
2841 let slot = req
2845 .stream_resumable
2846 .unwrap_or(false)
2847 .then(|| state.streams.register(&request_id));
2848 let emitter = resume::Emitter::new(slot);
2849
2850 let (tx, rx) = tokio::sync::mpsc::channel::<Result<Event, Infallible>>(64);
2851 let keepalive = sse::keepalive_event(&ChatCompletionChunk {
2855 id: request_id.clone(),
2856 request_id: None,
2857 object: "chat.completion.chunk",
2858 model: model_name.clone(),
2859 choices: vec![ChatCompletionChunkChoice {
2860 index: 0,
2861 delta: ChatCompletionChunkDelta {
2862 role: None,
2863 content: None,
2864 reasoning_content: None,
2865 tool_calls: None,
2866 },
2867 finish_reason: None,
2868 }],
2869 usage: None,
2870 });
2871
2872 tokio::task::spawn_blocking(move || {
2873 let _cancel_guard = cancel_guard;
2876 let tx_chunks = tx.clone();
2877 let orphan_timeout = sse::orphan_timeout_from_env();
2882 let mut first = true;
2883 let head_request_id = request_id.clone();
2884 let stream_reasoning: Rc<RefCell<Option<crate::policy::parser::ReasoningParser>>> =
2894 Rc::new(RefCell::new(posture.reasoning_parser()));
2895 let emit_reasoning = Rc::clone(&stream_reasoning);
2896 let stream_tools: Rc<RefCell<Option<crate::policy::parser::ToolCallParser>>> = Rc::new(
2900 RefCell::new(tools_active.then(|| posture.tool_call_parser(&offered_tools))),
2901 );
2902 let emit_tools = Rc::clone(&stream_tools);
2903 let streamed_calls = Rc::new(std::cell::Cell::new(0usize));
2907 let emit_streamed_calls = Rc::clone(&streamed_calls);
2908 let result = run_generation_emit(
2909 &model,
2910 &prompt,
2911 ¶ms,
2912 kv_pool.as_ref(),
2913 paged_kv.as_ref(),
2914 prefix_cache.as_deref(),
2915 batcher.as_ref(),
2916 ceiling.as_deref(),
2917 metal_private_decode_gate.as_deref(),
2918 |chunk| {
2919 if !overlap || chunk.is_empty() {
2920 return;
2921 }
2922 let (reasoning, content) = match emit_reasoning.borrow_mut().as_mut() {
2923 Some(parser) => {
2924 let delta = parser.push(chunk);
2925 (delta.reasoning, delta.content)
2926 }
2927 None => (String::new(), chunk.to_string()),
2928 };
2929 let (content, tool_calls) = match emit_tools.borrow_mut().as_mut() {
2933 Some(parser) => {
2934 let (text, calls) =
2935 tool_call_deltas(parser.push(&content), &emit_streamed_calls);
2936 (text, calls)
2937 }
2938 None => (content, Vec::new()),
2939 };
2940 if reasoning.is_empty() && content.is_empty() && tool_calls.is_empty() {
2943 return;
2944 }
2945 let role = if first { Some("assistant") } else { None };
2946 let request_id = first.then(|| head_request_id.clone());
2947 first = false;
2948 let payload = ChatCompletionChunk {
2949 id: head_request_id.clone(),
2950 request_id,
2951 object: "chat.completion.chunk",
2952 model: model_name.clone(),
2953 choices: vec![ChatCompletionChunkChoice {
2954 index: 0,
2955 delta: ChatCompletionChunkDelta {
2956 role,
2957 content: (!content.is_empty()).then_some(content),
2958 reasoning_content: (!reasoning.is_empty()).then_some(reasoning),
2959 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
2960 },
2961 finish_reason: None,
2962 }],
2963 usage: None,
2964 };
2965 if let Err(why) =
2974 sse::send_or_orphan(&tx_chunks, Ok(emitter.event(&payload)), orphan_timeout)
2975 {
2976 if why == sse::SendFailure::Orphaned {
2977 tracing::warn!(
2978 "SSE stream {head_request_id} accepted nothing for the orphan \
2979 deadline; treating it as abandoned"
2980 );
2981 }
2982 if !emitter.is_resumable() {
2994 cancel_token.cancel();
2995 }
2996 }
2997 },
2998 );
2999
3000 let mut pending_request_id = first.then(|| request_id.clone());
3006
3007 match result {
3008 Ok((finish, usage, full_text)) => {
3009 if let Some(id) = &session_id {
3010 sessions.store_reply(
3011 id,
3012 ChatMessage {
3013 role: "assistant".to_string(),
3014 content: Some(MessageContent::Text(full_text.clone())),
3015 tool_calls: None,
3016 tool_call_id: None,
3017 reasoning_content: None,
3018 },
3019 );
3020 }
3021 let mut streamed_finish: Option<&'static str> = None;
3027 if overlap {
3028 let tail = stream_reasoning
3029 .borrow_mut()
3030 .as_mut()
3031 .map(|parser| parser.flush())
3032 .unwrap_or_default();
3033 let (mut content, mut tool_calls) = (tail.content, Vec::new());
3034 if let Some(parser) = stream_tools.borrow_mut().as_mut() {
3035 let mut events = parser.push(&content);
3036 events.extend(parser.finish());
3037 let (text, calls) = tool_call_deltas(events, &streamed_calls);
3038 content = text;
3039 tool_calls = calls;
3040 }
3041 if !content.is_empty() || !tail.reasoning.is_empty() || !tool_calls.is_empty() {
3042 let payload = ChatCompletionChunk {
3043 id: request_id.clone(),
3044 request_id: pending_request_id.take(),
3045 object: "chat.completion.chunk",
3046 model: model_name.clone(),
3047 choices: vec![ChatCompletionChunkChoice {
3048 index: 0,
3049 delta: ChatCompletionChunkDelta {
3050 role: None,
3051 content: (!content.is_empty()).then_some(content),
3052 reasoning_content: (!tail.reasoning.is_empty())
3053 .then_some(tail.reasoning),
3054 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
3055 },
3056 finish_reason: None,
3057 }],
3058 usage: None,
3059 };
3060 let _ =
3061 sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3062 }
3063 if streamed_calls.get() > 0 {
3064 streamed_finish = Some("tool_calls");
3065 }
3066 } else {
3067 let parsed = output::parse_output(&full_text, &offered_tools, posture);
3070 let tool_calls: Vec<ToolCallDelta> = parsed
3071 .calls
3072 .iter()
3073 .enumerate()
3074 .map(|(index, call)| {
3075 ToolCallDelta::whole(index, call.name.clone(), call.arguments.clone())
3076 })
3077 .collect();
3078 if !tool_calls.is_empty() {
3079 streamed_finish = Some("tool_calls");
3080 }
3081 if !tool_calls.is_empty()
3082 || !parsed.content.is_empty()
3083 || parsed.reasoning.is_some()
3084 {
3085 let payload = ChatCompletionChunk {
3086 id: request_id.clone(),
3087 request_id: pending_request_id.take(),
3088 object: "chat.completion.chunk",
3089 model: model_name.clone(),
3090 choices: vec![ChatCompletionChunkChoice {
3091 index: 0,
3092 delta: ChatCompletionChunkDelta {
3093 role: Some("assistant"),
3094 content: (!parsed.content.is_empty() && tool_calls.is_empty())
3095 .then(|| parsed.content.clone()),
3096 reasoning_content: parsed.reasoning.clone(),
3097 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
3098 },
3099 finish_reason: None,
3100 }],
3101 usage: None,
3102 };
3103 let _ =
3104 sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3105 }
3106 }
3107 let final_finish_reason = match streamed_finish {
3111 Some(reason) if finish.as_str() != "length" => reason,
3112 _ => finish.as_str(),
3113 };
3114 let final_payload = ChatCompletionChunk {
3115 id: request_id.clone(),
3116 request_id: pending_request_id.take(),
3117 object: "chat.completion.chunk",
3118 model: model_name,
3119 choices: vec![ChatCompletionChunkChoice {
3120 index: 0,
3121 delta: ChatCompletionChunkDelta {
3122 role: None,
3123 content: None,
3124 reasoning_content: None,
3125 tool_calls: None,
3126 },
3127 finish_reason: Some(final_finish_reason),
3128 }],
3129 usage: Some(usage.clone()),
3130 };
3131 let _ = sse::send_or_orphan(&tx, Ok(emitter.event(&final_payload)), orphan_timeout);
3132 let _ = sse::send_or_orphan(&tx, Ok(emitter.done()), orphan_timeout);
3133 stats_state.record_request(stats::Record {
3138 request_id: &request_id,
3139 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
3140 model: Some(served_model.clone()),
3141 status: 200,
3142 stream: true,
3143 duration_ms: started.elapsed().as_millis() as u64,
3144 usage: Some(&usage),
3145 attribution: &attribution,
3146 });
3147 }
3148 Err(e) => {
3149 tracing::warn!("decode error on streamed request {request_id}: {e}");
3150 stats_state.record_request(stats::Record {
3156 request_id: &request_id,
3157 route: ferrox_api::routes::V1_CHAT_COMPLETIONS,
3158 model: Some(served_model.clone()),
3159 status: 500,
3160 stream: true,
3161 duration_ms: started.elapsed().as_millis() as u64,
3162 usage: None,
3163 attribution: &attribution,
3164 });
3165 let payload = ChatCompletionChunk {
3166 id: request_id.clone(),
3167 request_id: pending_request_id.take(),
3168 object: "chat.completion.chunk",
3169 model: model_name,
3170 choices: vec![ChatCompletionChunkChoice {
3171 index: 0,
3172 delta: ChatCompletionChunkDelta {
3173 role: Some("assistant"),
3174 content: Some(format!("[error: {e}]")),
3175 reasoning_content: None,
3176 tool_calls: None,
3177 },
3178 finish_reason: Some("stop"),
3179 }],
3180 usage: None,
3181 };
3182 let _ = sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3183 let _ = sse::send_or_orphan(&tx, Ok(emitter.done()), orphan_timeout);
3184 }
3185 }
3186 drop(emitter);
3190 });
3191
3192 let stream = sse::with_keepalive(rx, keepalive, sse::KEEPALIVE_INTERVAL);
3193 Ok((
3216 [(
3217 axum::http::HeaderName::from_static("x-accel-buffering"),
3218 axum::http::HeaderValue::from_static("no"),
3219 )],
3220 Sse::new(stream),
3221 )
3222 .into_response())
3223}
3224
3225fn protected_routes() -> Router<Arc<AppState>> {
3256 use ferrox_api::routes;
3257
3258 Router::new()
3259 .route(routes::V1_MODELS, get(list_models))
3260 .route(routes::V1_RESPONSES, post(responses::responses))
3264 .route(
3265 &axum_path(routes::V1_RESPONSE),
3266 get(responses::responses_get),
3267 )
3268 .route(
3269 &axum_path(routes::V1_RESPONSE_CANCEL),
3270 post(responses::responses_cancel),
3271 )
3272 .route(&axum_path(routes::SLOTS_ID), post(slots::post_slot))
3273 .route(routes::V1_STATS, get(serving_stats))
3274 .route(routes::V1_REQUESTS, get(recent_requests))
3275 .route(routes::V1_CACHE_STATUS, get(cache_admin::cache_status))
3276 .route(routes::V1_CACHE_REBUILD, post(cache_admin::cache_rebuild))
3277 .route(routes::ADMIN_PREPARE_STOP, post(cache_admin::prepare_stop))
3278 .route(routes::V1_CHAT_COMPLETIONS, post(chat_completions))
3279 .route(routes::V1_CANCEL, post(cancel_generation))
3283 .route(&axum_path(routes::V1_STREAM), get(resume::resume))
3288 .route(&axum_path(routes::V1_STREAM_POLL), get(resume::poll))
3289 .route(routes::V1_MESSAGES, post(anthropic::messages))
3290 .route(
3291 routes::V1_MESSAGES_COUNT_TOKENS,
3292 post(anthropic::count_tokens),
3293 )
3294 .route(routes::V1_COMPLETIONS, post(openai_extra::completions))
3295 .route(routes::COMPLETION, post(completion::completion))
3300 .route(routes::COMPLETIONS, post(completion::completion))
3301 .route(routes::V1_TOKENIZE, post(openai_extra::tokenize))
3302 .route(routes::V1_DETOKENIZE, post(openai_extra::detokenize))
3303 .route(routes::TOKENIZE, post(openai_extra::tokenize))
3309 .route(routes::DETOKENIZE, post(openai_extra::detokenize))
3310 .route(routes::V1_EMBEDDINGS, post(embeddings::embeddings))
3311 .route(routes::V1_RERANK, post(rerank::rerank))
3315 .route(routes::RERANK, post(rerank::rerank))
3316 .route(routes::CACHE_STATS, get(cache_stats))
3317 .route(routes::METRICS, get(metrics))
3318 .route(routes::ADMIN_MODELS, get(admin::models))
3323 .route(routes::ADMIN_MODELS_LOAD, post(admin::load_model))
3324 .route(routes::ADMIN_MODELS_UNLOAD, post(admin::unload_model))
3325 .route(routes::ADMIN_DOWNLOAD, post(admin::download))
3326 .route(routes::ADMIN_TASKS, get(admin::tasks))
3327 .route(&admin::cancel_route(), post(admin::cancel_task))
3328 .route(routes::ADMIN_STATS, get(admin::stats))
3329 .merge(conversations::router())
3333}
3334
3335fn axum_path(template: &str) -> String {
3336 let mut out = String::with_capacity(template.len());
3337 let mut rest = template;
3338 while let Some(open) = rest.find('{') {
3339 let Some(close) = rest[open..].find('}').map(|c| open + c) else {
3340 break;
3341 };
3342 out.push_str(&rest[..open]);
3343 out.push(':');
3344 out.push_str(&rest[open + 1..close]);
3345 rest = &rest[close + 1..];
3346 }
3347 out.push_str(rest);
3348 out
3349}
3350
3351async fn cancel_generation(
3360 State(state): State<Arc<AppState>>,
3361 Json(req): Json<ferrox_api::CancelGenerationRequest>,
3362) -> Response {
3363 let cancelled = state.cancels.cancel(&req.request_id);
3364 let status = if cancelled {
3365 StatusCode::OK
3366 } else {
3367 StatusCode::NOT_FOUND
3368 };
3369 let detail = if cancelled {
3370 "the generation was asked to stop; it ends at its next token".to_string()
3371 } else {
3372 "no generation with that request_id is running -- it has already \
3373 finished, was never issued, or was served by a path that does \
3374 not register for cancellation"
3375 .to_string()
3376 };
3377 (
3378 status,
3379 Json(ferrox_api::CancelGenerationResponse {
3380 request_id: req.request_id,
3381 cancelled,
3382 detail,
3383 }),
3384 )
3385 .into_response()
3386}
3387
3388type Activated = (
3392 Loaded,
3393 Option<serving::batch::ContinuousBatcher>,
3394 Option<Arc<budget::ContextCeiling>>,
3395);
3396
3397fn price_batcher_config(path: Option<&str>) -> serving::batch::BatcherConfig {
3415 let mut batcher = serving::batch::BatcherConfig::from_env();
3416 if batcher.max_context.is_some() && batcher.kv_blocks.is_some() {
3417 return batcher;
3420 }
3421 let Some(path) = path else {
3422 return batcher;
3423 };
3424 let priced = budget::price_gguf(path, ferrox_models::KvElem::F32, 1);
3430 let Some((priced, gguf_ctx, source)) = priced else {
3431 return batcher;
3432 };
3433 let Some(derived) = budget::derive_limits(&priced, gguf_ctx, batcher.kv_block_size) else {
3434 tracing::warn!(
3438 "this checkpoint's weights leave no room for KV inside the {source}: {} weight \
3439 bytes against a {} byte budget. Serving with no derived context ceiling -- set \
3440 FERROX_DEVICE_BUDGET_BYTES if the probe is wrong, or FERROX_CB_MAX_CONTEXT to \
3441 admit on a number you choose.",
3442 priced.weights_bytes,
3443 priced.device_budget_bytes,
3444 );
3445 return batcher;
3446 };
3447 tracing::info!("{source}");
3448 tracing::info!("{}", derived.fit);
3449 let adopted = budget::apply_derived(&mut batcher, &derived);
3450 if adopted.max_context {
3451 tracing::info!(
3452 "derived per-request context ceiling: {} token positions (prompt + max_tokens); \
3453 override with FERROX_CB_MAX_CONTEXT",
3454 derived.max_context
3455 );
3456 }
3457 if adopted.kv_blocks {
3458 tracing::info!(
3459 "derived KV block budget: {} blocks x {} positions; override with FERROX_CB_KV_BLOCKS",
3460 derived.kv_blocks,
3461 batcher.kv_block_size
3462 );
3463 }
3464 batcher
3465}
3466
3467pub(crate) fn activate_loaded_model(
3476 loaded: model::LoadedModel,
3477 enable_continuous_batching: bool,
3478 path: Option<&str>,
3479 paged_kv: Option<&generate::PagedKvConfig>,
3480) -> Activated {
3481 match loaded {
3482 model::LoadedModel::Gguf(g) => {
3483 let decoder = Arc::new(g.decoder);
3484 let tokenizer = Arc::new(g.tokenizer);
3485 let config = price_batcher_config(path);
3486 let shape =
3492 ferrox_models::KvShape::from_config(&decoder.config, ferrox_models::KvElem::F32);
3493 let ceiling = Arc::new(budget::ContextCeiling::new(config.max_context, shape));
3494 let batcher = if enable_continuous_batching {
3495 tracing::info!(
3496 "continuous batching enabled: decode steps share Decoder::forward_multi_seq \
3497 (stop sequences use the same pending-buffer trim as the private generate loop)"
3498 );
3499 let tok = Arc::clone(&tokenizer);
3500 let decode = Arc::new(move |ids: &[usize]| tok.decode_bytes(ids));
3501 Some(serving::batch::ContinuousBatcher::spawn_with_ceiling(
3502 Arc::clone(&decoder),
3503 decode,
3504 config,
3505 Arc::clone(&ceiling),
3506 paged_kv.cloned(),
3507 ))
3508 } else {
3509 None
3510 };
3511 (
3512 Loaded::Generative(Arc::new(Model::Gguf(GgufModel {
3513 decoder,
3514 tokenizer,
3515 stop_tokens: g.stop_tokens,
3516 bos_id: g.bos_id,
3517 is_synthetic: g.is_synthetic,
3518 chat_template: g.chat_template,
3519 }))),
3520 batcher,
3521 Some(ceiling),
3522 )
3523 }
3524 model::LoadedModel::Kimi(k) => (
3525 Loaded::Generative(Arc::new(Model::Kimi(KimiModel {
3526 engine: k.engine,
3527 tokenizer: k.tokenizer,
3528 stop_tokens: k.stop_tokens,
3529 chat_template: k.chat_template,
3530 }))),
3531 None,
3532 None,
3533 ),
3534 model::LoadedModel::Mla(m) => (
3535 Loaded::Generative(Arc::new(Model::Mla(MlaModel {
3536 engine: m.engine,
3537 tokenizer: m.tokenizer,
3538 stop_tokens: m.stop_tokens,
3539 bos_id: m.bos_id,
3540 name: m.name,
3541 chat_template: m.chat_template,
3542 }))),
3543 None,
3544 None,
3545 ),
3546 model::LoadedModel::Gemma4(m) => (
3547 Loaded::Generative(Arc::new(Model::Gemma4(Gemma4Model {
3548 engine: m.engine,
3549 tokenizer: m.tokenizer,
3550 stop_tokens: m.stop_tokens,
3551 bos_id: m.bos_id,
3552 name: m.name,
3553 chat_template: m.chat_template,
3554 }))),
3555 None,
3556 None,
3557 ),
3558 model::LoadedModel::Glm52(g) => (
3559 Loaded::Generative(Arc::new(Model::Glm52(Glm52Model {
3560 engine: g.engine,
3561 tokenizer: g.tokenizer,
3562 stop_tokens: g.stop_tokens,
3563 bos_id: g.bos_id,
3564 name: g.name,
3565 chat_template: g.chat_template,
3566 }))),
3567 None,
3568 None,
3569 ),
3570 model::LoadedModel::Encoder(e) => (Loaded::Encoder(e), None, None),
3575 }
3576}
3577
3578struct StartupModels {
3585 loaded: model::LoadedModel,
3586 embedding: Option<Arc<ferrox_models::EmbeddingModel>>,
3587}
3588
3589fn continuous_batching_env() -> Option<bool> {
3590 match std::env::var("FERROX_CONTINUOUS_BATCHING")
3591 .ok()
3592 .map(|v| v.trim().to_ascii_lowercase())
3593 .as_deref()
3594 {
3595 None => None,
3596 Some("1" | "true" | "yes" | "on") => Some(true),
3597 Some("0" | "false" | "no" | "off") => Some(false),
3598 _ => None,
3599 }
3600}
3601
3602fn metal_private_decode_active() -> bool {
3603 #[cfg(feature = "metal")]
3604 {
3605 BUILT_WITH_METAL
3606 && ferrox_metal::attn::metal_attn_enabled()
3607 && std::env::var("FERROX_METAL").ok().as_deref() != Some("0")
3608 }
3609 #[cfg(not(feature = "metal"))]
3610 {
3611 false
3612 }
3613}
3614
3615fn continuous_batching_compatible(
3616 loaded: &model::LoadedModel,
3617 kv_pool: &Option<generate::KvPoolConfig>,
3618 prefix_cache: &Option<Arc<Mutex<PrefixCache>>>,
3619 paged_kv: &Option<generate::PagedKvConfig>,
3620) -> bool {
3621 matches!(loaded, model::LoadedModel::Gguf(_))
3622 && (paged_kv.is_some() || (kv_pool.is_none() && prefix_cache.is_none()))
3623}
3624
3625fn resolve_continuous_batching_enabled(
3626 loaded: &model::LoadedModel,
3627 kv_pool: &Option<generate::KvPoolConfig>,
3628 prefix_cache: &Option<Arc<Mutex<PrefixCache>>>,
3629 paged_kv: &Option<generate::PagedKvConfig>,
3630) -> bool {
3631 if !continuous_batching_compatible(loaded, kv_pool, prefix_cache, paged_kv) {
3632 return false;
3633 }
3634 match continuous_batching_env() {
3635 Some(true) => true,
3636 Some(false) => false,
3637 None => metal_private_decode_active(),
3638 }
3639}
3640
3641fn acquire_metal_private_decode_gate(
3642 gate: Option<&std::sync::Mutex<()>>,
3643 used_batcher: bool,
3644) -> Option<std::sync::MutexGuard<'_, ()>> {
3645 if used_batcher {
3646 None
3647 } else {
3648 gate.map(|g| g.lock().unwrap_or_else(|p| p.into_inner()))
3649 }
3650}
3651
3652fn build_app_state(
3653 models: StartupModels,
3654 kv_pool: Option<generate::KvPoolConfig>,
3655 paged_kv: Option<generate::PagedKvConfig>,
3656 prefix_cache: Option<Arc<Mutex<PrefixCache>>>,
3657 enable_continuous_batching: bool,
3658 mcp: Option<mcp::LoadedMcpConfig>,
3659 detection: Arc<health::Detection>,
3660) -> AppState {
3661 let StartupModels { loaded, embedding } = models;
3662 let configured_path = std::env::var("FERROX_MODEL_PATH").ok();
3663 let (loaded, batcher, ceiling) = activate_loaded_model(
3664 loaded,
3665 enable_continuous_batching,
3666 configured_path.as_deref(),
3667 paged_kv.as_ref(),
3668 );
3669 let id = startup_model_id();
3675 let metal_private_decode_gate = if enable_continuous_batching || !metal_private_decode_active()
3676 {
3677 None
3678 } else {
3679 tracing::info!(
3680 "Metal private-loop decode will serialize concurrent requests until \
3681 continuous batching is enabled (FERROX_CONTINUOUS_BATCHING=1 or --cont-batching)"
3682 );
3683 Some(Arc::new(std::sync::Mutex::new(())))
3684 };
3685 AppState {
3686 embedding,
3687 active: std::sync::RwLock::new(Some(Arc::new(ActiveModel {
3688 id,
3689 loaded,
3690 batcher,
3691 ceiling,
3692 checkpoint_path: configured_path.as_deref().map(PathBuf::from),
3693 }))),
3694 paged_kv,
3695 load_in_progress: std::sync::atomic::AtomicBool::new(false),
3696 tasks: Arc::new(tasks::TaskRegistry::new()),
3697 cancels: Arc::new(cancel::CancelRegistry::new()),
3698 stats: stats::Stats::new(),
3699 streams: resume::StreamRegistry::new(),
3700 model_dir: admin::model_dirs().into_iter().next(),
3701 response_cache: Mutex::new(ResponseCache::new(1000, Duration::from_secs(3600))),
3702 kv_pool,
3703 prefix_cache,
3704 sessions: session::SessionStore::new(),
3705 requests_total: std::sync::atomic::AtomicU64::new(0),
3706 request_errors_total: std::sync::atomic::AtomicU64::new(0),
3707 started_at: std::time::Instant::now(),
3708 last_request_ms: std::sync::atomic::AtomicU64::new(0),
3709 detection,
3710 mcp,
3711 continuous_batching_enabled: enable_continuous_batching,
3712 metal_private_decode_gate,
3713 loading_model: Mutex::new(None),
3714 last_load_error: Mutex::new(None),
3715 serving: Mutex::new(crate::stats::ServingStats::default()),
3716 maintenance: Mutex::new(crate::policy::maintenance::MaintenanceGate::serving()),
3717 footprint: Mutex::new(crate::policy::footprint::ProbeCache::new(FOOTPRINT_TTL_MS)),
3718 started_unix: unix_now(),
3719 }
3720}
3721
3722fn load_embedding_model() -> anyhow::Result<Option<Arc<ferrox_models::EmbeddingModel>>> {
3730 let Ok(path) = std::env::var("FERROX_EMBEDDING_MODEL_PATH") else {
3731 return Ok(None);
3732 };
3733 let model = ferrox_models::EmbeddingModel::from_gguf_path(&path)
3734 .map_err(|e| anyhow::anyhow!("FERROX_EMBEDDING_MODEL_PATH={path}: {e}"))?;
3735 tracing::info!(
3736 "loaded embedding model '{}' ({}, {} dims, pooling {}, max {} tokens)",
3737 model.name(),
3738 model.architecture(),
3739 model.n_embd(),
3740 model.pooling_type().name(),
3741 model.n_ctx_train(),
3742 );
3743 Ok(Some(Arc::new(model)))
3744}
3745
3746fn unix_now() -> u64 {
3751 std::time::SystemTime::now()
3752 .duration_since(std::time::UNIX_EPOCH)
3753 .map(|d| d.as_secs())
3754 .unwrap_or(0)
3755}
3756
3757fn startup_model_id() -> Option<String> {
3762 let configured = std::env::var("FERROX_MODEL_PATH").ok()?;
3763 let configured = std::fs::canonicalize(&configured).ok()?;
3764 admin::discover(&admin::model_dirs())
3765 .into_iter()
3766 .find(|d| {
3767 std::fs::canonicalize(&d.path)
3768 .map(|p| p == configured)
3769 .unwrap_or(false)
3770 })
3771 .map(|d| d.id)
3772}
3773
3774fn init_cpu_pool() {
3783 match ferrox_core::threads::init_cpu_pool() {
3784 Some(n) => eprintln!(
3785 "ferrox-server: rayon pool {n} threads (perf cores {}; override with FERROX_CPU_THREADS)",
3786 ferrox_core::threads::perf_core_count()
3787 ),
3788 None => eprintln!("ferrox-server: global rayon pool already built; leaving it alone"),
3789 }
3790}
3791
3792fn announce_ready(addr: SocketAddr, scheme: &str) {
3805 use std::io::Write;
3806 let ready =
3807 ferrox_api::ServerReady::new(addr, scheme, env!("CARGO_PKG_VERSION"), std::process::id());
3808 let mut stdout = std::io::stdout().lock();
3809 let _ = writeln!(stdout, "{}", ready.to_line());
3810 let _ = stdout.flush();
3811}
3812
3813async fn shutdown_signal(exit_on_stdin_close: bool) {
3825 if !exit_on_stdin_close {
3826 std::future::pending::<()>().await;
3827 return;
3828 }
3829 let _ = tokio::task::spawn_blocking(|| {
3830 use std::io::Read;
3831 let mut sink = [0u8; 256];
3832 let mut stdin = std::io::stdin().lock();
3833 loop {
3834 match stdin.read(&mut sink) {
3835 Ok(0) => break,
3837 Ok(_) => continue,
3839 Err(e) => {
3840 tracing::warn!("stdin read failed ({e}); treating it as closed");
3841 break;
3842 }
3843 }
3844 }
3845 })
3846 .await;
3847 tracing::info!("stdin closed; shutting down");
3848}
3849
3850fn tokio_worker_threads() -> usize {
3855 std::env::var("FERROX_TOKIO_WORKERS")
3856 .ok()
3857 .and_then(|v| v.trim().parse::<usize>().ok())
3858 .filter(|n| *n > 0)
3859 .unwrap_or(2)
3860}
3861
3862fn install_ring_crypto_provider() {
3879 let _ = rustls::crypto::ring::default_provider().install_default();
3880}
3881
3882pub fn run_server(args: ServerArgs) -> anyhow::Result<()> {
3888 if args.list_devices {
3889 print_available_devices();
3890 return Ok(());
3891 }
3892 apply_cli_overrides(&args)?;
3893
3894 let _instance = {
3898 use ferrox_core::instance::{register, InstancePolicy};
3899 let policy = if args.allow_multiple_instances {
3900 InstancePolicy::Multi
3901 } else {
3902 InstancePolicy::from_env_or(InstancePolicy::Single)
3903 };
3904 let model = std::env::var("FERROX_MODEL_PATH").ok();
3905 register(
3906 "server",
3907 model.as_deref(),
3908 ferrox_core::instance::current_backend(),
3909 policy,
3910 )
3911 .map_err(|conflict| anyhow::anyhow!("{conflict}"))?
3912 };
3913
3914 let journal = journal::Journal::from_env();
3915 eprintln!(
3916 "ferrox-server: process lifecycle journal at {:?} (override with FERROX_JOURNAL_PATH)",
3917 journal.path()
3918 );
3919 journal.append(&journal::Record::session_start(
3920 env!("CARGO_PKG_VERSION"),
3921 std::process::id(),
3922 ));
3923 journal::install_panic_hook(journal.clone());
3924
3925 let mcp_config_path = args.mcp_config.clone();
3926 let exit_on_stdin_close = args.exit_on_stdin_close
3927 || std::env::var("FERROX_EXIT_ON_STDIN_CLOSE")
3928 .map(|v| v == "1")
3929 .unwrap_or(false);
3930
3931 unsafe { ferrox_core::weight_matrix::default_cpu_int_dot_on() };
3935 init_cpu_pool();
3936
3937 let runtime = tokio::runtime::Builder::new_multi_thread()
3938 .worker_threads(tokio_worker_threads())
3939 .enable_all()
3940 .build()?;
3941 let result = runtime.block_on(run(mcp_config_path, exit_on_stdin_close));
3942
3943 let reason = match &result {
3944 Ok(()) => "normal".to_string(),
3945 Err(e) => e.to_string(),
3946 };
3947 journal.append(&journal::Record::session_exit(reason));
3948
3949 runtime.shutdown_background();
3954
3955 result
3956}
3957
3958async fn run(mcp_config_path: Option<PathBuf>, exit_on_stdin_close: bool) -> anyhow::Result<()> {
3959 let _ = tracing_subscriber::fmt::try_init();
3965
3966 let addr = std::env::var("FERROX_ADDR").unwrap_or_else(|_| "127.0.0.1:8383".to_string());
3976 let api_key_configured = std::env::var("FERROX_API_KEY").is_ok();
3977 let allow_unauthenticated_remote = std::env::var("FERROX_ALLOW_UNAUTHENTICATED_REMOTE")
3978 .map(|v| v == "1")
3979 .unwrap_or(false);
3980 if let Err(msg) =
3981 security::check_bind_authorization(&addr, api_key_configured, allow_unauthenticated_remote)
3982 {
3983 anyhow::bail!(msg);
3984 }
3985
3986 let embedding_model = load_embedding_model()?;
3993
3994 let mut loaded = model::load()?;
3995 match &loaded {
3996 model::LoadedModel::Gguf(g) => tracing::info!(
3997 "loaded GGUF model '{}' (synthetic={}, tokenizer={})",
3998 g.decoder.config.name,
3999 g.is_synthetic,
4000 g.tokenizer.kind()
4001 ),
4002 model::LoadedModel::Kimi(k) => tracing::info!(
4003 "loaded Kimi K3 checkpoint (tokenizer={} base tokens)",
4004 k.tokenizer.vocab_size()
4005 ),
4006 model::LoadedModel::Mla(m) => tracing::info!(
4007 "loaded MLA GGUF '{}' (tokenizer={})",
4008 m.name,
4009 m.tokenizer.kind()
4010 ),
4011 model::LoadedModel::Gemma4(m) => tracing::info!(
4012 "loaded Gemma4 GGUF '{}' (tokenizer={})",
4013 m.name,
4014 m.tokenizer.kind()
4015 ),
4016 model::LoadedModel::Glm52(g) => tracing::info!(
4017 "loaded GLM-5.2 GGUF '{}' (tokenizer={})",
4018 g.name,
4019 g.tokenizer.kind()
4020 ),
4021 model::LoadedModel::Encoder(_) => {}
4024 }
4025 let metal_default_moe_budget = {
4033 #[cfg(feature = "metal")]
4034 {
4035 ferrox_core::metal_dense_enabled()
4036 && std::env::var("FERROX_GPU_VRAM_BUDGET_BYTES").is_err()
4037 }
4038 #[cfg(not(feature = "metal"))]
4039 {
4040 false
4041 }
4042 };
4043 if let Ok(budget_str) = std::env::var("FERROX_GPU_VRAM_BUDGET_BYTES") {
4044 let budget: u64 = budget_str
4045 .parse()
4046 .expect("FERROX_GPU_VRAM_BUDGET_BYTES must be a non-negative integer");
4047 match &mut loaded {
4048 model::LoadedModel::Gguf(g) => {
4049 tracing::info!(
4050 "GPU expert placement enabled: {budget} byte VRAM budget for routed experts \
4051 (CUDA and/or Metal matvecs when built with the matching feature)"
4052 );
4053 g.decoder.gpu_vram_budget_bytes = Some(budget);
4054 }
4055 model::LoadedModel::Kimi(_) => {
4056 tracing::warn!(
4057 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is Kimi K3 -- not \
4058 supported yet (its MoE stack isn't wired to PlacementPlan), ignoring"
4059 );
4060 }
4061 model::LoadedModel::Mla(_) => {
4062 tracing::warn!(
4063 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is MLA -- dense \
4064 FFN path only today; ignoring expert VRAM budget"
4065 );
4066 }
4067 model::LoadedModel::Gemma4(_) => {
4068 tracing::warn!(
4069 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is Gemma4 -- \
4070 ignoring expert VRAM budget"
4071 );
4072 }
4073 model::LoadedModel::Glm52(_) => {
4074 tracing::warn!(
4075 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is GLM-5.2 DSA -- \
4076 GPU expert placement not wired yet; ignoring"
4077 );
4078 }
4079 model::LoadedModel::Encoder(_) => {
4080 tracing::warn!(
4081 "FERROX_GPU_VRAM_BUDGET_BYTES is set but the loaded model is an encoder -- \
4082 it has no routed experts to place; ignoring"
4083 );
4084 }
4085 }
4086 } else if metal_default_moe_budget {
4087 const METAL_DEFAULT_MOE_BUDGET: u64 = 64 * 1024 * 1024 * 1024;
4091 if let model::LoadedModel::Gguf(g) = &mut loaded {
4092 tracing::info!(
4093 "Metal MoE expert placement default-on ({METAL_DEFAULT_MOE_BUDGET} byte budget); \
4094 set FERROX_GPU_VRAM_BUDGET_BYTES=0 to force CPU experts"
4095 );
4096 g.decoder.gpu_vram_budget_bytes = Some(METAL_DEFAULT_MOE_BUDGET);
4097 }
4098 }
4099 #[cfg(feature = "cuda")]
4100 {
4101 if ferrox_core::cuda_dense_enabled() {
4102 tracing::info!(
4103 "CUDA dense matvec enabled for WeightMatrix::apply \
4104 (FERROX_CUDA=0|cpu forces CPU; weight buffers stay resident after first upload)"
4105 );
4106 } else {
4107 tracing::info!(
4108 "CUDA dense matvec disabled (FERROX_CUDA); dense decode uses CPU or Metal"
4109 );
4110 }
4111 }
4112 #[cfg(feature = "metal")]
4113 {
4114 if ferrox_core::metal_dense_enabled() {
4115 tracing::info!(
4116 "Metal dense matvec enabled for WeightMatrix::apply \
4117 (FERROX_METAL=0|cpu forces CPU; weight buffers stay resident after first upload)"
4118 );
4119 match std::env::var("FERROX_METAL_ATTN").ok().as_deref() {
4120 Some("1") | Some("true") | Some("on") | Some("attn") => {
4121 tracing::info!(
4122 "Metal fused attention requested (FERROX_METAL_ATTN): \
4123 QKV→RoPE→GQA→O on-GPU for Norm/NeoX decode without QKV bias/QK-norm"
4124 );
4125 }
4126 _ => {}
4127 }
4128 tracing::info!(
4129 "Metal greedy GPU argmax: temperature<=0 folds \
4130 final_norm+lm_head+argmax into the dense stack"
4131 );
4132 } else {
4133 tracing::info!("Metal dense matvec disabled (FERROX_METAL); dense decode uses CPU");
4134 }
4135 }
4136 let blocks_env = std::env::var("FERROX_KV_POOL_BLOCKS");
4147 let block_size_env = std::env::var("FERROX_KV_POOL_BLOCK_SIZE");
4148 let byte_budget_env = std::env::var("FERROX_KV_BYTE_BUDGET");
4149 if blocks_env.is_ok() && byte_budget_env.is_ok() {
4150 panic!(
4151 "FERROX_KV_POOL_BLOCKS and FERROX_KV_BYTE_BUDGET are mutually exclusive \
4152 (set one block-count source plus FERROX_KV_POOL_BLOCK_SIZE, or neither to disable)"
4153 );
4154 }
4155 let kv_pool = match (blocks_env, block_size_env, byte_budget_env) {
4156 (Ok(blocks), Ok(block_size), Err(_)) => {
4157 let total_blocks: usize = blocks
4158 .parse()
4159 .expect("FERROX_KV_POOL_BLOCKS must be a positive integer");
4160 let block_size: usize = block_size
4161 .parse()
4162 .expect("FERROX_KV_POOL_BLOCK_SIZE must be a positive integer");
4163 let queue_wait_ms: u64 = std::env::var("FERROX_KV_POOL_QUEUE_TIMEOUT_MS")
4169 .ok()
4170 .map(|v| {
4171 v.parse()
4172 .expect("FERROX_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4173 })
4174 .unwrap_or(0);
4175 tracing::info!(
4176 "KV cache block pool enabled: {total_blocks} blocks x {block_size} positions \
4177 each, shared across all concurrent requests, {queue_wait_ms}ms admission queue wait"
4178 );
4179 Some(generate::KvPoolConfig {
4180 pool: Arc::new(Mutex::new(KvBlockPool::new(block_size, total_blocks))),
4181 queue_wait: Duration::from_millis(queue_wait_ms),
4182 })
4183 }
4184 (Err(_), Ok(block_size), Ok(byte_budget)) => {
4185 let block_size: usize = block_size
4186 .parse()
4187 .expect("FERROX_KV_POOL_BLOCK_SIZE must be a positive integer");
4188 let budget: u64 = byte_budget
4189 .parse()
4190 .expect("FERROX_KV_BYTE_BUDGET must be a positive integer");
4191 let cfg = match &loaded {
4192 model::LoadedModel::Gguf(g) => &g.decoder.config,
4193 model::LoadedModel::Kimi(_)
4194 | model::LoadedModel::Mla(_)
4195 | model::LoadedModel::Gemma4(_)
4196 | model::LoadedModel::Glm52(_)
4197 | model::LoadedModel::Encoder(_) => {
4198 panic!(
4199 "FERROX_KV_BYTE_BUDGET requires a GGUF decoder model \
4200 (set FERROX_MODEL_PATH to a generic-decoder .gguf file)"
4201 );
4202 }
4203 };
4204 let bytes_per_block = block_size
4205 * cfg.kv_heads_all_layers()
4206 * cfg.head_dim
4207 * 2
4208 * std::mem::size_of::<f32>();
4209 assert!(
4210 bytes_per_block > 0,
4211 "derived KV block byte size must be positive (check model config and block size)"
4212 );
4213 let total_blocks = (budget as usize / bytes_per_block).max(1);
4214 let queue_wait_ms: u64 = std::env::var("FERROX_KV_POOL_QUEUE_TIMEOUT_MS")
4215 .ok()
4216 .map(|v| {
4217 v.parse()
4218 .expect("FERROX_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4219 })
4220 .unwrap_or(0);
4221 tracing::info!(
4222 "KV cache block pool enabled from byte budget: {budget} bytes / \
4223 {bytes_per_block} bytes per block ({block_size} positions x {} layers) -> \
4224 {total_blocks} blocks, {queue_wait_ms}ms admission queue wait",
4225 cfg.n_layers
4226 );
4227 Some(generate::KvPoolConfig {
4228 pool: Arc::new(Mutex::new(KvBlockPool::new(block_size, total_blocks))),
4229 queue_wait: Duration::from_millis(queue_wait_ms),
4230 })
4231 }
4232 (Err(_), Err(_), Err(_)) => None,
4233 (Err(_), Ok(_), Err(_)) => panic!(
4234 "FERROX_KV_POOL_BLOCK_SIZE requires FERROX_KV_POOL_BLOCKS or FERROX_KV_BYTE_BUDGET \
4235 (or unset all three to disable KV cache pooling)"
4236 ),
4237 (Ok(_), Ok(_), Ok(_)) => {
4238 unreachable!("FERROX_KV_POOL_BLOCKS and FERROX_KV_BYTE_BUDGET are mutually exclusive")
4239 }
4240 (Ok(_), Err(_), _) | (Err(_), Err(_), Ok(_)) => panic!(
4241 "FERROX_KV_POOL_BLOCKS/FERROX_KV_BYTE_BUDGET and FERROX_KV_POOL_BLOCK_SIZE must be \
4242 set together (or neither, to disable KV cache pooling)"
4243 ),
4244 };
4245 let paged_kv = match (
4251 std::env::var("FERROX_PAGED_KV_BLOCKS"),
4252 std::env::var("FERROX_PAGED_KV_BLOCK_SIZE"),
4253 ) {
4254 (Ok(blocks), Ok(block_size)) => {
4255 assert!(
4256 kv_pool.is_none(),
4257 "FERROX_PAGED_KV_BLOCKS and FERROX_KV_POOL_BLOCKS/FERROX_KV_BYTE_BUDGET are \
4258 mutually exclusive: both bound the same KV memory, by different means. \
4259 Set one."
4260 );
4261 let blocks_per_layer: usize = blocks
4275 .parse()
4276 .expect("FERROX_PAGED_KV_BLOCKS must be a positive integer");
4277 let block_size: usize = block_size
4278 .parse()
4279 .expect("FERROX_PAGED_KV_BLOCK_SIZE must be a positive integer");
4280 let gguf = match &loaded {
4281 model::LoadedModel::Gguf(g) => g,
4282 _ => panic!(
4283 "FERROX_PAGED_KV_BLOCKS requires a GGUF decoder model \
4284 (set FERROX_MODEL_PATH to a generic-decoder .gguf file)"
4285 ),
4286 };
4287 let cfg = &gguf.decoder.config;
4288 let queue_wait_ms: u64 = std::env::var("FERROX_KV_POOL_QUEUE_TIMEOUT_MS")
4289 .ok()
4290 .map(|v| {
4291 v.parse()
4292 .expect("FERROX_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4293 })
4294 .unwrap_or(0);
4295 tracing::info!(
4296 "Paged KV enabled: {blocks_per_layer} blocks x {block_size} positions per \
4297 layer across {} layers, shared by all concurrent requests, \
4298 {queue_wait_ms}ms admission queue wait",
4299 cfg.n_layers
4300 );
4301 let radix = Some(Arc::new(Mutex::new(crate::policy::radix::RadixCache::new(
4306 block_size,
4307 ))));
4308 let anchor_token = crate::policy::anchor::resolve_anchor_token(
4317 crate::policy::parser::ToolCallFormat::infer(
4318 &std::env::var("FERROX_MODEL_PATH").unwrap_or_default(),
4319 )
4320 .opener(),
4321 |text| {
4322 gguf.tokenizer
4323 .encode(text, SpecialTokens::Parse)
4324 .into_iter()
4325 .map(|t| t as u32)
4326 .collect()
4327 },
4328 );
4329 if let Some(id) = anchor_token {
4330 tracing::info!(
4331 "Paged KV window slide: tool-call anchor is token {id}, so a turn's \
4332 window stops short of where its next turn rejoins"
4333 );
4334 }
4335 let slide_interval: usize = std::env::var("FERROX_PAGED_KV_SLIDE_INTERVAL")
4336 .ok()
4337 .map(|v| {
4338 v.parse()
4339 .expect("FERROX_PAGED_KV_SLIDE_INTERVAL must be a positive integer")
4340 })
4341 .unwrap_or(crate::policy::pool_budget::DEFAULT_SWA_EVICTION_INTERVAL);
4342 if let Some(window) = cfg.uniform_sliding_window() {
4343 tracing::info!(
4344 "Paged KV window slide enabled: every layer slides by {window} every \
4345 {slide_interval} decode steps, so a request holds its prompt and a \
4346 window rather than its whole context"
4347 );
4348 } else if cfg.kv_block_window().is_some() {
4349 tracing::info!(
4350 "Paged KV window slide NOT enabled: this model has full-attention layers, \
4351 and a page group holds one block in every layer"
4352 );
4353 }
4354 Some(generate::PagedKvConfig {
4355 store: Arc::new(cfg.new_paged_kv(block_size, blocks_per_layer)),
4358 queue_wait: Duration::from_millis(queue_wait_ms),
4359 radix,
4360 anchor_token,
4361 slide_interval,
4362 })
4363 }
4364 (Err(_), Err(_)) => None,
4365 _ => panic!(
4366 "FERROX_PAGED_KV_BLOCKS and FERROX_PAGED_KV_BLOCK_SIZE must be set together \
4367 (or neither, to disable paged KV)"
4368 ),
4369 };
4370 let prefix_cache = std::env::var("FERROX_PREFIX_CACHE_ENTRIES").ok().map(|v| {
4377 let max_entries: usize = v
4378 .parse()
4379 .expect("FERROX_PREFIX_CACHE_ENTRIES must be a positive integer");
4380 if kv_pool.is_some() {
4381 tracing::warn!(
4382 "FERROX_PREFIX_CACHE_ENTRIES is set but so is the KV pool -- prefix \
4383 caching will never be consulted while a KV pool is configured"
4384 );
4385 }
4386 assert!(
4393 paged_kv.is_none(),
4394 "FERROX_PREFIX_CACHE_ENTRIES and FERROX_PAGED_KV_BLOCKS are mutually exclusive: \
4395 the prefix cache stores contiguous KV snapshots, which a paged request does not \
4396 produce, so the cache could never hit. Set one."
4397 );
4398 tracing::info!(
4399 "KV-prefix cache enabled: up to {max_entries} stored prefixes, shared across \
4400 all requests"
4401 );
4402 Arc::new(Mutex::new(PrefixCache::new(max_entries)))
4403 });
4404 if matches!(
4405 loaded,
4406 model::LoadedModel::Kimi(_) | model::LoadedModel::Mla(_) | model::LoadedModel::Glm52(_)
4407 ) && (kv_pool.is_some() || prefix_cache.is_some())
4408 {
4409 tracing::warn!(
4410 "KV pool / prefix cache are configured but the loaded model is Kimi, MLA, or GLM-5.2 -- \
4411 neither is consulted for those engines (state shapes differ from Decoder KV); see \
4412 ferrox_models::engine's module docs"
4413 );
4414 }
4415 let enable_cb =
4416 resolve_continuous_batching_enabled(&loaded, &kv_pool, &prefix_cache, &paged_kv);
4417 if enable_cb && continuous_batching_env().is_none() && metal_private_decode_active() {
4418 tracing::info!(
4419 "continuous batching enabled by default on Metal for safe parallel serving \
4420 (set FERROX_CONTINUOUS_BATCHING=0 or --no-cont-batching to use the private path)"
4421 );
4422 }
4423 if continuous_batching_env() == Some(true)
4424 && !continuous_batching_compatible(&loaded, &kv_pool, &prefix_cache, &paged_kv)
4425 && (kv_pool.is_some() || prefix_cache.is_some())
4426 {
4427 tracing::warn!(
4428 "FERROX_CONTINUOUS_BATCHING=1 ignored while KV pool or prefix cache is configured \
4429 (those modes keep the private generate path)"
4430 );
4431 }
4432 if let Ok(n) = std::env::var("FERROX_CHUNKED_PREFILL") {
4433 if let Ok(chunk) = n.parse::<usize>() {
4434 if chunk > 0 {
4435 tracing::info!("chunked prefill enabled: {chunk} tokens per forward_batch chunk");
4436 }
4437 }
4438 }
4439 if matches!(
4440 std::env::var("FERROX_CPU_KV_OFFLOAD").ok().as_deref(),
4441 Some("1")
4442 ) {
4443 tracing::warn!(
4444 "FERROX_CPU_KV_OFFLOAD=1: syncing Metal KV to host after each decode step \
4445 (minimal spill; full layer offload still planned)"
4446 );
4447 }
4448
4449 let mcp = match mcp_config_path {
4450 Some(path) => {
4451 let loaded = mcp::load_mcp_config(&path)?;
4452 tracing::info!(
4453 "MCP config loaded from {} ({} server(s); invocation not wired yet)",
4454 loaded.path,
4455 loaded.servers.len()
4456 );
4457 Some(loaded)
4458 }
4459 None => None,
4460 };
4461
4462 let detection = health::Detection::spawn();
4466
4467 let state = Arc::new(build_app_state(
4468 StartupModels {
4469 loaded,
4470 embedding: embedding_model,
4471 },
4472 kv_pool,
4473 paged_kv,
4474 prefix_cache,
4475 enable_cb,
4476 mcp,
4477 detection,
4478 ));
4479
4480 use ferrox_api::routes;
4484
4485 let public = Router::new().route(routes::HEALTH, get(health));
4490
4491 let mut protected = protected_routes();
4492
4493 if let Ok(key) = std::env::var("FERROX_API_KEY") {
4498 tracing::info!("API key auth enabled");
4499 let auth = limits::AuthConfig {
4500 api_key: Arc::new(key),
4501 };
4502 protected = protected.route_layer(axum::middleware::from_fn_with_state(
4503 auth,
4504 limits::require_api_key,
4505 ));
4506 }
4507 if let Ok(rpm) = std::env::var("FERROX_RATE_LIMIT_PER_MINUTE") {
4508 let rpm: u32 = rpm
4509 .parse()
4510 .expect("FERROX_RATE_LIMIT_PER_MINUTE must be a positive integer");
4511 tracing::info!("rate limiting enabled: {rpm} requests/minute (global)");
4512 let limiter = Arc::new(limits::RateLimiter::per_minute(rpm));
4513 protected = protected.route_layer(axum::middleware::from_fn_with_state(
4514 limiter,
4515 limits::rate_limit,
4516 ));
4517 }
4518 if let Ok(spec) = std::env::var("FERROX_CORS_ORIGINS") {
4526 let origins = security::parse_cors_origins(&spec)
4527 .unwrap_or_else(|e| panic!("FERROX_CORS_ORIGINS: {e}"));
4528 tracing::info!(
4529 "CORS enabled: {} allow-listed origin(s) ({})",
4530 origins.len(),
4531 spec
4532 );
4533 let cors = tower_http::cors::CorsLayer::new()
4534 .allow_origin(tower_http::cors::AllowOrigin::list(origins))
4535 .allow_methods([axum::http::Method::GET, axum::http::Method::POST])
4536 .allow_headers([
4537 axum::http::header::CONTENT_TYPE,
4538 axum::http::header::AUTHORIZATION,
4539 axum::http::HeaderName::from_static(attribution::CLIENT_HEADER),
4545 axum::http::HeaderName::from_static("last-event-id"),
4549 ]);
4550 protected = protected.route_layer(cors);
4551 }
4552
4553 let app = public
4557 .merge(protected)
4558 .layer(axum::middleware::from_fn(limits::retry_after))
4559 .with_state(state);
4560
4561 let tls_paths = security::tls_paths_from_env().unwrap_or_else(|e| panic!("{e}"));
4567 install_ring_crypto_provider();
4568 match tls_paths {
4573 Some(paths) => {
4574 let config =
4575 axum_server::tls_rustls::RustlsConfig::from_pem_file(&paths.cert, &paths.key)
4576 .await
4577 .map_err(|e| {
4578 anyhow::anyhow!(
4579 "failed to load TLS cert/key ({:?}, {:?}): {e}",
4580 paths.cert,
4581 paths.key
4582 )
4583 })?;
4584 let socket_addr: std::net::SocketAddr = addr
4585 .parse()
4586 .map_err(|e| anyhow::anyhow!("invalid FERROX_ADDR {addr:?} for TLS: {e}"))?;
4587 let listener = std::net::TcpListener::bind(socket_addr)?;
4588 listener.set_nonblocking(true)?;
4596 let bound = listener.local_addr()?;
4597 tracing::info!("TLS enabled: ferrox-server listening on https://{bound}");
4598 announce_ready(bound, "https");
4599
4600 let handle = axum_server::Handle::new();
4601 let shutdown_handle = handle.clone();
4602 tokio::spawn(async move {
4603 shutdown_signal(exit_on_stdin_close).await;
4604 shutdown_handle.graceful_shutdown(Some(Duration::from_secs(5)));
4605 });
4606 axum_server::from_tcp_rustls(listener, config)?
4607 .handle(handle)
4608 .serve(app.into_make_service())
4609 .await?;
4610 }
4611 None => {
4612 let listener = tokio::net::TcpListener::bind(&addr).await?;
4613 let bound = listener.local_addr()?;
4614 tracing::info!("ferrox-server listening on {bound}");
4615 announce_ready(bound, "http");
4616 axum::serve(listener, app)
4617 .with_graceful_shutdown(shutdown_signal(exit_on_stdin_close))
4618 .await?;
4619 }
4620 }
4621 Ok(())
4622}
4623
4624#[cfg(test)]
4625mod tests {
4626 use super::*;
4627 use ferrox_models::config::test_dense_fixture;
4628
4629 #[test]
4630 fn the_ready_line_round_trips_through_a_parent_reading_stdout() {
4631 let addr: SocketAddr = "127.0.0.1:51999".parse().unwrap();
4632 let ready = ferrox_api::ServerReady::new(addr, "http", "0.5.0", std::process::id());
4633 let parsed = ferrox_api::ServerReady::from_line(&ready.to_line()).unwrap();
4634 assert_eq!(parsed.port, 51999);
4635 assert_eq!(parsed.base_url(), "http://127.0.0.1:51999");
4636 assert!(ferrox_api::ServerReady::from_line("INFO ferrox-server listening").is_none());
4638 }
4639
4640 fn test_model() -> Model {
4641 let cfg = test_dense_fixture();
4645 Model::Gguf(GgufModel {
4646 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 32)),
4647 tokenizer: Arc::new(ServerTokenizer::Byte),
4648 stop_tokens: StopTokens::default(),
4649 bos_id: None,
4650 is_synthetic: true,
4651 chat_template: chat_template::PromptTemplate::plain(),
4652 })
4653 }
4654
4655 fn greedy_params(max_tokens: usize) -> GenerationParams {
4656 GenerationParams {
4657 reasoning: None,
4658 max_tokens,
4659 sampling: SamplingParams::default(),
4660 seed: 1,
4661 stop: Vec::new(),
4662 stop_token_ids: Vec::new(),
4663 json_object: false,
4664 grammar: None,
4665 cancel: None,
4666 ignore_eos: false,
4667 reasoning_budget: crate::reasoning_budget::ReasoningBudget::Unrestricted,
4668 }
4669 }
4670
4671 fn test_model_full_byte_vocab() -> Model {
4675 test_model_full_byte_vocab_with_eos(None)
4676 }
4677
4678 fn test_model_full_byte_vocab_with_eos(eos: Option<usize>) -> Model {
4685 let mut cfg = test_dense_fixture();
4686 cfg.vocab_size = 256;
4687 Model::Gguf(GgufModel {
4688 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
4689 tokenizer: Arc::new(ServerTokenizer::Byte),
4690 stop_tokens: StopTokens::from_eos(eos),
4691 bos_id: None,
4692 is_synthetic: true,
4693 chat_template: chat_template::PromptTemplate::plain(),
4694 })
4695 }
4696
4697 fn test_state(model: Model, response_cache: ResponseCache) -> AppState {
4701 AppState {
4702 embedding: None,
4703 paged_kv: None,
4704 active: std::sync::RwLock::new(Some(Arc::new(ActiveModel {
4705 id: None,
4706 loaded: Loaded::Generative(Arc::new(model)),
4707 batcher: None,
4708 ceiling: None,
4709 checkpoint_path: None,
4710 }))),
4711 load_in_progress: std::sync::atomic::AtomicBool::new(false),
4712 tasks: Arc::new(tasks::TaskRegistry::new()),
4713 cancels: Arc::new(cancel::CancelRegistry::new()),
4714 stats: stats::Stats::new(),
4715 streams: resume::StreamRegistry::new(),
4716 model_dir: None,
4717 response_cache: Mutex::new(response_cache),
4718 kv_pool: None,
4719 prefix_cache: None,
4720 sessions: session::SessionStore::new(),
4721 requests_total: std::sync::atomic::AtomicU64::new(0),
4722 request_errors_total: std::sync::atomic::AtomicU64::new(0),
4723 started_at: std::time::Instant::now(),
4724 last_request_ms: std::sync::atomic::AtomicU64::new(0),
4725 detection: Arc::new(health::Detection::ready(health::probe_backends())),
4726 mcp: None,
4727 continuous_batching_enabled: false,
4728 metal_private_decode_gate: None,
4729 loading_model: Mutex::new(None),
4730 last_load_error: Mutex::new(None),
4731 serving: Mutex::new(crate::stats::ServingStats::default()),
4732 maintenance: Mutex::new(crate::policy::maintenance::MaintenanceGate::serving()),
4733 footprint: Mutex::new(crate::policy::footprint::ProbeCache::new(FOOTPRINT_TTL_MS)),
4734 started_unix: unix_now(),
4735 }
4736 }
4737
4738 fn test_app() -> Router {
4747 test_app_with_state(Arc::new(test_state(
4748 test_model_full_byte_vocab(),
4749 ResponseCache::new(1000, Duration::from_secs(3600)),
4750 )))
4751 }
4752
4753 fn test_app_with_state(state: Arc<AppState>) -> Router {
4756 Router::new()
4766 .route(ferrox_api::routes::HEALTH, get(health))
4767 .merge(protected_routes())
4768 .with_state(state)
4769 }
4770
4771 fn named_test_model(name: &'static str, vocab_size: usize) -> Model {
4772 let mut cfg = test_dense_fixture();
4773 cfg.name = name;
4774 cfg.vocab_size = vocab_size;
4775 Model::Gguf(GgufModel {
4776 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
4777 tokenizer: Arc::new(ServerTokenizer::Byte),
4778 stop_tokens: StopTokens::default(),
4779 bos_id: None,
4780 is_synthetic: true,
4781 chat_template: chat_template::PromptTemplate::plain(),
4782 })
4783 }
4784
4785 fn model_with_template(name: &'static str, source: &str) -> Model {
4789 let mut cfg = test_dense_fixture();
4790 cfg.name = name;
4791 cfg.vocab_size = 256;
4792 Model::Gguf(GgufModel {
4793 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
4794 tokenizer: Arc::new(ServerTokenizer::Byte),
4795 stop_tokens: StopTokens::default(),
4796 bos_id: None,
4797 is_synthetic: true,
4798 chat_template: chat_template::PromptTemplate::from_gguf_metadata(
4799 Some(source),
4800 Some("qwen3"),
4801 false,
4802 true,
4803 None,
4804 None,
4805 ),
4806 })
4807 }
4808
4809 #[tokio::test]
4818 async fn a_template_that_rejects_the_conversation_is_a_400_on_the_streaming_path() {
4819 let strict = "{% if messages | length > 1 %}\
4822 {{ raise_exception('this template takes one turn') }}\
4823 {% endif %}{{ messages[0].content }}";
4824 let state = Arc::new(test_state(
4825 model_with_template("strict", strict),
4826 ResponseCache::new(4, Duration::from_secs(60)),
4827 ));
4828 let app = test_app_with_state(state);
4829
4830 let (status, body) = post_json_uri(
4831 &app,
4832 "/v1/chat/completions",
4833 serde_json::json!({
4834 "model": "strict",
4835 "stream": true,
4836 "messages": [
4837 {"role": "user", "content": "one"},
4838 {"role": "user", "content": "two"},
4839 ],
4840 }),
4841 )
4842 .await;
4843 assert_eq!(status, StatusCode::BAD_REQUEST);
4844 assert_eq!(body["error"]["param"], serde_json::json!("messages"));
4845 assert!(
4846 body["error"]["message"]
4847 .as_str()
4848 .unwrap()
4849 .contains("one turn"),
4850 "the template's own message must reach the caller: {body}"
4851 );
4852
4853 let (status, _) = post_json_uri(
4855 &app,
4856 "/v1/chat/completions",
4857 serde_json::json!({
4858 "model": "strict",
4859 "stream": true,
4860 "max_tokens": 1,
4861 "messages": [{"role": "user", "content": "one"}],
4862 }),
4863 )
4864 .await;
4865 assert_eq!(status, StatusCode::OK);
4866 }
4867
4868 #[tokio::test]
4870 async fn models_advertises_the_gears_this_checkpoint_actually_has() {
4871 let reasoning = "{% if enable_thinking %}<think>{% endif %}\
4872 {% if reasoning_effort %}\
4873 {% if reasoning_effort not in ['low','medium','high'] %}\
4874 {{ raise_exception('bad effort') }}\
4875 {% endif %}[{{ reasoning_effort }}]\
4876 {% endif %}{{ messages[0].content }}";
4877 let state = Arc::new(test_state(
4878 model_with_template("thinker", reasoning),
4879 ResponseCache::new(4, Duration::from_secs(60)),
4880 ));
4881 let app = test_app_with_state(state);
4882 let (status, models) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
4883 assert_eq!(status, StatusCode::OK);
4884 let entry = &models["data"][0];
4885 assert_eq!(
4886 entry["supported_reasoning_efforts"],
4887 serde_json::json!(["off", "low", "medium", "high"])
4888 );
4889 assert_eq!(entry["default_reasoning_effort"], serde_json::json!("off"));
4890 }
4891
4892 #[tokio::test]
4897 async fn a_checkpoint_with_no_thinking_controls_advertises_neither_field() {
4898 let app = test_app();
4899 let (_, models) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
4900 let entry = &models["data"][0];
4901 assert!(entry.get("supported_reasoning_efforts").is_none());
4902 assert!(entry.get("default_reasoning_effort").is_none());
4903 }
4904
4905 fn active_model(state: &AppState, name: &'static str) -> Arc<ActiveModel> {
4906 Arc::new(ActiveModel {
4907 id: Some(name.to_string()),
4908 loaded: Loaded::Generative(Arc::new(named_test_model(name, 256))),
4909 batcher: None,
4910 ceiling: None,
4911 checkpoint_path: None,
4912 })
4913 .tap_into(state)
4914 }
4915
4916 trait TapInto {
4918 fn tap_into(self, state: &AppState) -> Self;
4919 }
4920 impl TapInto for Arc<ActiveModel> {
4921 fn tap_into(self, state: &AppState) -> Self {
4922 state.swap_active(Some(Arc::clone(&self)));
4923 self
4924 }
4925 }
4926
4927 #[test]
4933 fn an_in_flight_request_keeps_the_model_it_started_on() {
4934 let state = test_state(
4935 named_test_model("model-a", 256),
4936 ResponseCache::new(4, Duration::from_secs(60)),
4937 );
4938
4939 let in_flight = state.active().expect("a model is loaded");
4942 assert_eq!(in_flight.name(), "model-a");
4943
4944 active_model(&state, "model-b");
4945
4946 assert_eq!(state.active().unwrap().name(), "model-b");
4948 assert_eq!(in_flight.name(), "model-a");
4950 let (_chunks, finish, _usage) = run_generation(
4951 in_flight.generative().unwrap(),
4952 "hi",
4953 &greedy_params(3),
4954 None,
4955 None,
4956 None,
4957 None,
4958 None,
4959 None,
4960 )
4961 .expect("the old model must still decode after being swapped out");
4962 assert!(matches!(finish, FinishReason::Length | FinishReason::Stop));
4963 }
4964
4965 #[test]
4970 fn a_swapped_out_model_lives_until_its_last_holder_releases_it() {
4971 let state = test_state(
4972 named_test_model("model-a", 256),
4973 ResponseCache::new(4, Duration::from_secs(60)),
4974 );
4975 let in_flight = state.active().expect("a model is loaded");
4976 let weights = Arc::clone(in_flight.generative().unwrap());
4977 assert!(Arc::strong_count(&weights) >= 2);
4978
4979 let previous = state.swap_active(Some(Arc::new(ActiveModel {
4980 id: Some("model-b".to_string()),
4981 loaded: Loaded::Generative(Arc::new(named_test_model("model-b", 256))),
4982 batcher: None,
4983 ceiling: None,
4984 checkpoint_path: None,
4985 })));
4986 drop(previous);
4987 assert!(Arc::strong_count(&weights) >= 2);
4989 drop(in_flight);
4990 assert_eq!(Arc::strong_count(&weights), 1);
4991 }
4992
4993 #[tokio::test]
4997 async fn unloading_answers_503_instead_of_serving_the_dropped_model() {
4998 let state = Arc::new(test_state(
4999 named_test_model("model-a", 256),
5000 ResponseCache::new(4, Duration::from_secs(60)),
5001 ));
5002 let app = test_app_with_state(Arc::clone(&state));
5003
5004 let (status, body) = post_json_uri(
5005 &app,
5006 ferrox_api::routes::ADMIN_MODELS_UNLOAD,
5007 serde_json::json!({}),
5008 )
5009 .await;
5010 assert_eq!(status, StatusCode::OK);
5011 assert_eq!(body["ok"], true);
5012 assert!(body["active"].is_null());
5013 assert!(state.active().is_none());
5014
5015 let (status, _) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
5016 assert_eq!(status, StatusCode::OK);
5017 let (_, models) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
5018 assert_eq!(models["data"].as_array().unwrap().len(), 0);
5019
5020 let (status, body) = post_json_uri(
5021 &app,
5022 "/v1/chat/completions",
5023 serde_json::json!({
5024 "model": "x",
5025 "messages": [{"role": "user", "content": "hi"}]
5026 }),
5027 )
5028 .await;
5029 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5030 assert_eq!(body["error"]["type"], "model_not_loaded");
5031 }
5032
5033 #[tokio::test]
5037 async fn health_reports_the_unloaded_state_rather_than_going_silent() {
5038 let state = Arc::new(test_state(
5039 named_test_model("model-a", 256),
5040 ResponseCache::new(4, Duration::from_secs(60)),
5041 ));
5042 let app = test_app_with_state(Arc::clone(&state));
5043 state.swap_active(None);
5044
5045 let (status, body) = get_json(&app, ferrox_api::routes::HEALTH).await;
5046 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5049 assert_eq!(body["state"], "unavailable");
5050 assert_eq!(body["reason"], "model_not_loaded");
5051 assert!(body["model"].is_null());
5052 let real_weights = body["capabilities"]
5053 .as_array()
5054 .unwrap()
5055 .iter()
5056 .find(|c| c["id"] == "real_weights")
5057 .cloned()
5058 .expect("real_weights is always reported");
5059 assert_eq!(real_weights["available"], false);
5060 assert_eq!(real_weights["reason"], "model_not_loaded");
5061 }
5062
5063 #[tokio::test]
5067 async fn a_finished_request_lands_in_the_stats_ring_with_both_durations() {
5068 let app = test_app();
5069
5070 let (status, completion) = post_json_uri(
5071 &app,
5072 "/v1/chat/completions",
5073 serde_json::json!({
5074 "model": "x",
5075 "messages": [{"role": "user", "content": "hi"}],
5076 "max_tokens": 4
5077 }),
5078 )
5079 .await;
5080 assert_eq!(status, StatusCode::OK);
5081 let request_id = completion["request_id"].as_str().unwrap().to_string();
5082
5083 let (status, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5084 assert_eq!(status, StatusCode::OK);
5085 let recent = stats["recent"].as_array().unwrap();
5086 assert_eq!(recent.len(), 1);
5087 let row = &recent[0];
5088 assert_eq!(row["request_id"], request_id);
5089 assert_eq!(row["route"], ferrox_api::routes::V1_CHAT_COMPLETIONS);
5090 assert_eq!(row["status"], 200);
5091 assert_eq!(row["stream"], false);
5092 assert!(row["duration_ms"].is_number());
5095 assert!(row["decode_ms"].is_number());
5096 assert!(stats["tokens_generated_total"].as_u64().unwrap() > 0);
5097 assert_eq!(
5098 stats["tokens_prompt_total"].as_u64().unwrap(),
5099 row["prompt_tokens"].as_u64().unwrap()
5100 );
5101 }
5102
5103 #[tokio::test]
5107 async fn a_rejected_request_is_recorded_too() {
5108 let state = Arc::new(test_state(
5109 named_test_model("model-a", 256),
5110 ResponseCache::new(4, Duration::from_secs(60)),
5111 ));
5112 let app = test_app_with_state(Arc::clone(&state));
5113 state.swap_active(None);
5114
5115 let (status, _) = post_json_uri(
5116 &app,
5117 "/v1/chat/completions",
5118 serde_json::json!({"model": "x", "messages": [{"role": "user", "content": "hi"}]}),
5119 )
5120 .await;
5121 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5122
5123 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5124 let recent = stats["recent"].as_array().unwrap();
5125 assert_eq!(recent.len(), 1);
5126 assert_eq!(recent[0]["status"], 503);
5127 assert_eq!(recent[0]["completion_tokens"], 0);
5128 assert!(recent[0]["decode_ms"].is_null());
5129 assert_eq!(stats["errors_total"], 1);
5130 }
5131
5132 async fn post_json_with_headers(
5136 app: &Router,
5137 uri: &str,
5138 body: serde_json::Value,
5139 headers: &[(&str, &str)],
5140 ) -> (StatusCode, serde_json::Value) {
5141 use http_body_util::BodyExt;
5142 use tower::ServiceExt;
5143
5144 let mut builder = axum::http::Request::builder()
5145 .method("POST")
5146 .uri(uri)
5147 .header("content-type", "application/json");
5148 for (name, value) in headers {
5149 builder = builder.header(*name, *value);
5150 }
5151 let response = app
5152 .clone()
5153 .oneshot(
5154 builder
5155 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
5156 .unwrap(),
5157 )
5158 .await
5159 .unwrap();
5160 let status = response.status();
5161 let bytes = response.into_body().collect().await.unwrap().to_bytes();
5162 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
5163 (status, json)
5164 }
5165
5166 #[tokio::test]
5170 async fn tokenize_detokenize_and_embeddings_all_land_in_the_ring() {
5171 let app = test_app();
5172
5173 let (status, _) = post_json_uri(
5174 &app,
5175 ferrox_api::routes::V1_TOKENIZE,
5176 serde_json::json!({"prompt": "hello"}),
5177 )
5178 .await;
5179 assert_eq!(status, StatusCode::OK);
5180 let (status, _) = post_json_uri(
5181 &app,
5182 ferrox_api::routes::V1_DETOKENIZE,
5183 serde_json::json!({"tokens": [104, 105]}),
5184 )
5185 .await;
5186 assert_eq!(status, StatusCode::OK);
5187 let (status, _) = post_json_uri(
5188 &app,
5189 ferrox_api::routes::V1_EMBEDDINGS,
5190 serde_json::json!({"input": "hello"}),
5191 )
5192 .await;
5193 assert_eq!(status, StatusCode::OK);
5194
5195 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5196 let routes: Vec<&str> = stats["recent"]
5197 .as_array()
5198 .unwrap()
5199 .iter()
5200 .map(|row| row["route"].as_str().unwrap())
5201 .collect();
5202 for expected in [
5203 ferrox_api::routes::V1_TOKENIZE,
5204 ferrox_api::routes::V1_DETOKENIZE,
5205 ferrox_api::routes::V1_EMBEDDINGS,
5206 ] {
5207 assert!(
5208 routes.contains(&expected),
5209 "{expected} is missing: {routes:?}"
5210 );
5211 }
5212
5213 let row = |route: &str| {
5214 stats["recent"]
5215 .as_array()
5216 .unwrap()
5217 .iter()
5218 .find(|r| r["route"] == route)
5219 .cloned()
5220 .unwrap()
5221 };
5222 let embed = row(ferrox_api::routes::V1_EMBEDDINGS);
5226 assert!(embed["prompt_tokens"].as_u64().unwrap() > 0);
5227 assert!(embed["decode_ms"].is_null());
5228 assert_eq!(embed["completion_tokens"], 0);
5229 assert_eq!(row(ferrox_api::routes::V1_TOKENIZE)["prompt_tokens"], 0);
5233 assert_eq!(
5234 stats["tokens_prompt_total"].as_u64().unwrap(),
5235 embed["prompt_tokens"].as_u64().unwrap(),
5236 "only the forward pass counted"
5237 );
5238 }
5239
5240 fn streaming_test_app() -> Router {
5245 let mut cfg = test_dense_fixture();
5246 cfg.vocab_size = 256;
5247 let model = Model::Gguf(GgufModel {
5248 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5249 tokenizer: Arc::new(ServerTokenizer::Byte),
5250 stop_tokens: StopTokens::default(),
5251 bos_id: None,
5252 is_synthetic: false,
5253 chat_template: chat_template::PromptTemplate::plain(),
5254 });
5255 test_app_with_state(Arc::new(test_state(
5256 model,
5257 ResponseCache::new(1000, Duration::from_secs(3600)),
5258 )))
5259 }
5260
5261 #[tokio::test]
5265 async fn the_native_completion_wire_is_not_the_openai_one() {
5266 let app = test_app();
5267
5268 let (status, native) = post_json_uri(
5269 &app,
5270 ferrox_api::routes::COMPLETION,
5271 serde_json::json!({"prompt": "hi", "n_predict": 4}),
5272 )
5273 .await;
5274 assert_eq!(status, StatusCode::OK, "{native}");
5275 assert!(native["content"].is_string(), "{native}");
5276 assert_eq!(native["stop"], true);
5277 assert_eq!(native["stop_type"], "limit");
5278 assert_eq!(native["stopping_word"], "");
5279 assert_eq!(native["truncated"], false);
5280 assert_eq!(native["id_slot"], -1);
5281 assert!(native["timings"]["prompt_n"].is_number(), "{native}");
5282 assert!(native["generation_settings"]["n_predict"] == 4, "{native}");
5283 assert!(
5284 native.get("choices").is_none(),
5285 "the native shape has no `choices`: {native}"
5286 );
5287
5288 let (status, openai) = post_json_uri(
5289 &app,
5290 ferrox_api::routes::V1_COMPLETIONS,
5291 serde_json::json!({"prompt": "hi", "max_tokens": 4}),
5292 )
5293 .await;
5294 assert_eq!(status, StatusCode::OK);
5295 assert!(openai["choices"][0]["text"].is_string(), "{openai}");
5296 assert!(
5297 openai.get("content").is_none(),
5298 "the OpenAI shape has no top-level `content`: {openai}"
5299 );
5300 }
5301
5302 #[tokio::test]
5306 async fn both_native_spellings_reach_the_same_handler() {
5307 let app = test_app();
5308 for route in [
5309 ferrox_api::routes::COMPLETION,
5310 ferrox_api::routes::COMPLETIONS,
5311 ] {
5312 let (status, body) = post_json_uri(
5313 &app,
5314 route,
5315 serde_json::json!({"prompt": "hi", "n_predict": 2, "seed": 1}),
5316 )
5317 .await;
5318 assert_eq!(status, StatusCode::OK, "{route}: {body}");
5319 assert_eq!(body["stop"], true, "{route}");
5320 assert!(body["content"].is_string(), "{route}");
5321 }
5322
5323 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5326 let routes: Vec<&str> = stats["recent"]
5327 .as_array()
5328 .unwrap()
5329 .iter()
5330 .map(|row| row["route"].as_str().unwrap())
5331 .collect();
5332 assert!(
5333 routes.contains(&ferrox_api::routes::COMPLETION),
5334 "{routes:?}"
5335 );
5336 assert!(
5337 routes.contains(&ferrox_api::routes::COMPLETIONS),
5338 "{routes:?}"
5339 );
5340 }
5341
5342 #[tokio::test]
5348 async fn a_native_stream_ends_on_a_stop_frame_with_no_done_sentinel() {
5349 let app = streaming_test_app();
5350 let raw = post_sse_raw_uri(
5351 &app,
5352 ferrox_api::routes::COMPLETION,
5353 serde_json::json!({"prompt": "hi", "n_predict": 6, "stream": true, "seed": 7}),
5354 )
5355 .await;
5356
5357 assert!(
5358 !raw.contains("[DONE]"),
5359 "llama.cpp's native stream has no sentinel: {raw}"
5360 );
5361 let frames: Vec<serde_json::Value> = raw
5362 .lines()
5363 .filter_map(|line| line.strip_prefix("data: "))
5364 .map(|json| serde_json::from_str(json).expect("every frame is one JSON object"))
5365 .collect();
5366 assert!(frames.len() >= 2, "expected partials then a final: {raw}");
5367
5368 let (last, partials) = frames.split_last().unwrap();
5369 assert_eq!(last["stop"], true, "the last frame closes the stream");
5370 assert!(last["timings"].is_object(), "{last}");
5371 assert!(last["stop_type"].is_string(), "{last}");
5372 for partial in partials {
5373 assert_eq!(partial["stop"], false, "{partial}");
5374 assert!(partial["content"].is_string(), "{partial}");
5375 assert!(partial.get("timings").is_none(), "{partial}");
5379 assert!(partial.get("generation_settings").is_none(), "{partial}");
5380 }
5381 let streamed: String = partials
5384 .iter()
5385 .filter_map(|p| p["content"].as_str())
5386 .collect();
5387 assert_eq!(last["content"].as_str().unwrap(), streamed);
5388 }
5389
5390 #[tokio::test]
5395 async fn an_unbounded_n_predict_is_refused_rather_than_quietly_shrunk() {
5396 let app = test_app();
5397 for body in [
5398 serde_json::json!({"prompt": "hi"}),
5399 serde_json::json!({"prompt": "hi", "n_predict": -1}),
5400 ] {
5401 let (status, refusal) =
5402 post_json_uri(&app, ferrox_api::routes::COMPLETION, body.clone()).await;
5403 assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{body}: {refusal}");
5404 assert!(
5405 refusal["error"]["message"]
5406 .as_str()
5407 .unwrap()
5408 .contains("n_predict"),
5409 "{refusal}"
5410 );
5411 }
5412 let (status, _) = post_json_uri(
5415 &app,
5416 ferrox_api::routes::COMPLETION,
5417 serde_json::json!({"prompt": "hi", "n_predict": 2}),
5418 )
5419 .await;
5420 assert_eq!(status, StatusCode::OK);
5421 }
5422
5423 #[tokio::test]
5432 async fn a_stop_string_halts_the_answer_and_is_named_back() {
5433 let app = streaming_test_app();
5434 let ask = |stop: serde_json::Value| {
5435 let app = app.clone();
5436 async move {
5437 post_json_uri(
5438 &app,
5439 ferrox_api::routes::COMPLETION,
5440 serde_json::json!({
5441 "prompt": "hi",
5442 "n_predict": 64,
5443 "ignore_eos": true,
5444 "stop": stop,
5445 }),
5446 )
5447 .await
5448 .1
5449 }
5450 };
5451
5452 let baseline = ask(serde_json::json!([])).await;
5453 assert_eq!(baseline["stop_type"], "limit");
5454 assert_eq!(baseline["stopping_word"], "");
5455 let text = baseline["content"].as_str().unwrap().to_string();
5456 let sentinel: String = text.chars().skip(1).take(2).collect();
5462 assert_eq!(
5463 sentinel.chars().count(),
5464 2,
5465 "the fixture must produce enough output to cut: {text:?}"
5466 );
5467 let cut = text.find(&sentinel).expect("it came out of this text");
5468
5469 let stopped = ask(serde_json::json!([sentinel])).await;
5470 assert_eq!(stopped["stop_type"], "word", "{stopped}");
5471 assert_eq!(stopped["stopping_word"], sentinel);
5472 assert_eq!(
5473 stopped["content"].as_str().unwrap(),
5474 &text[..cut],
5475 "the answer must be cut at the sentinel, not run past it"
5476 );
5477 }
5478
5479 #[tokio::test]
5485 async fn the_llama_cpp_spelling_of_tokenize_reaches_the_same_handler() {
5486 let app = test_app();
5487
5488 let (v1_status, v1) = post_json_uri(
5489 &app,
5490 ferrox_api::routes::V1_TOKENIZE,
5491 serde_json::json!({"prompt": "hello"}),
5492 )
5493 .await;
5494 let (alias_status, alias) = post_json_uri(
5495 &app,
5496 ferrox_api::routes::TOKENIZE,
5497 serde_json::json!({"content": "hello"}),
5498 )
5499 .await;
5500 assert_eq!(v1_status, StatusCode::OK);
5501 assert_eq!(alias_status, StatusCode::OK, "{alias}");
5502 assert_eq!(v1["tokens"], alias["tokens"]);
5503 assert!(!alias["tokens"].as_array().unwrap().is_empty());
5504
5505 let (status, both_ways) = post_json_uri(
5508 &app,
5509 ferrox_api::routes::TOKENIZE,
5510 serde_json::json!({"prompt": "hello"}),
5511 )
5512 .await;
5513 assert_eq!(status, StatusCode::OK);
5514 assert_eq!(both_ways["tokens"], v1["tokens"]);
5515 }
5516
5517 #[tokio::test]
5522 async fn detokenize_answers_under_both_dialects_keys() {
5523 let app = test_app();
5524 for route in [
5525 ferrox_api::routes::DETOKENIZE,
5526 ferrox_api::routes::V1_DETOKENIZE,
5527 ] {
5528 let (status, body) =
5529 post_json_uri(&app, route, serde_json::json!({"tokens": [104, 105]})).await;
5530 assert_eq!(status, StatusCode::OK, "{route}");
5531 assert_eq!(body["text"], "hi", "{route}");
5532 assert_eq!(body["content"], body["text"], "{route}");
5533 }
5534 }
5535
5536 #[tokio::test]
5540 async fn the_alias_is_recorded_under_the_path_the_client_called() {
5541 let app = test_app();
5542 let (status, _) = post_json_uri(
5543 &app,
5544 ferrox_api::routes::TOKENIZE,
5545 serde_json::json!({"content": "hello"}),
5546 )
5547 .await;
5548 assert_eq!(status, StatusCode::OK);
5549
5550 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5551 let routes: Vec<&str> = stats["recent"]
5552 .as_array()
5553 .unwrap()
5554 .iter()
5555 .map(|row| row["route"].as_str().unwrap())
5556 .collect();
5557 assert!(
5558 routes.contains(&ferrox_api::routes::TOKENIZE),
5559 "the alias must be its own row: {routes:?}"
5560 );
5561 assert!(
5562 !routes.contains(&ferrox_api::routes::V1_TOKENIZE),
5563 "nothing called /v1/tokenize: {routes:?}"
5564 );
5565 }
5566
5567 #[tokio::test]
5572 async fn add_special_prepends_the_same_bos_the_decoder_would() {
5573 let mut cfg = test_dense_fixture();
5574 cfg.vocab_size = 256;
5575 let model = Model::Gguf(GgufModel {
5576 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5577 tokenizer: Arc::new(ServerTokenizer::Byte),
5578 stop_tokens: StopTokens::default(),
5579 bos_id: Some(7),
5580 is_synthetic: true,
5581 chat_template: chat_template::PromptTemplate::plain(),
5582 });
5583 let app = test_app_with_state(Arc::new(test_state(
5584 model,
5585 ResponseCache::new(1000, Duration::from_secs(3600)),
5586 )));
5587
5588 let (_, plain) = post_json_uri(
5589 &app,
5590 ferrox_api::routes::TOKENIZE,
5591 serde_json::json!({"content": "hi"}),
5592 )
5593 .await;
5594 let (_, special) = post_json_uri(
5595 &app,
5596 ferrox_api::routes::TOKENIZE,
5597 serde_json::json!({"content": "hi", "add_special": true}),
5598 )
5599 .await;
5600
5601 assert_eq!(plain["tokens"], serde_json::json!([104, 105]));
5602 assert_eq!(special["tokens"], serde_json::json!([7, 104, 105]));
5603 assert_eq!(special["count"], 3);
5604 }
5605
5606 #[tokio::test]
5609 async fn a_rejected_embeddings_request_is_recorded_with_its_status() {
5610 let app = test_app();
5611 let (status, _) = post_json_uri(
5612 &app,
5613 ferrox_api::routes::V1_EMBEDDINGS,
5614 serde_json::json!({"input": "hi", "encoding_format": "base64"}),
5615 )
5616 .await;
5617 assert_eq!(status, StatusCode::BAD_REQUEST);
5618
5619 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5620 let recent = stats["recent"].as_array().unwrap();
5621 assert_eq!(recent.len(), 1);
5622 assert_eq!(recent[0]["route"], ferrox_api::routes::V1_EMBEDDINGS);
5623 assert_eq!(recent[0]["status"], 400);
5624 assert_eq!(
5625 recent[0]["prompt_tokens"], 0,
5626 "a rejected call embedded nothing"
5627 );
5628 }
5629
5630 #[tokio::test]
5633 async fn a_row_names_the_key_that_served_it_without_carrying_the_key() {
5634 let app = test_app();
5635 let key = "sk-monitor-secret";
5636 let (status, _) = post_json_with_headers(
5637 &app,
5638 "/v1/chat/completions",
5639 serde_json::json!({
5640 "model": "x",
5641 "messages": [{"role": "user", "content": "hi"}],
5642 "max_tokens": 2
5643 }),
5644 &[
5645 ("authorization", &format!("Bearer {key}")),
5646 ("x-ferrox-client", "ferrox-studio"),
5647 ],
5648 )
5649 .await;
5650 assert_eq!(status, StatusCode::OK);
5651
5652 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5653 let row = stats["recent"].as_array().unwrap()[0].clone();
5654 let fingerprint = row["via_api_key"]
5655 .as_str()
5656 .expect("the row names the key that served it")
5657 .to_string();
5658 assert_eq!(fingerprint, attribution::key_fingerprint(key));
5659 assert!(!fingerprint.contains(key));
5660 assert!(
5661 !serde_json::to_string(&stats).unwrap().contains(key),
5662 "the stats payload must not carry the key in any form"
5663 );
5664 assert_eq!(row["client"], "ferrox-studio");
5665 }
5666
5667 #[tokio::test]
5670 async fn different_keys_are_different_callers_and_no_key_is_null() {
5671 let app = test_app();
5672 let body = serde_json::json!({
5673 "model": "x",
5674 "messages": [{"role": "user", "content": "hi"}],
5675 "max_tokens": 1
5676 });
5677 for headers in [
5678 vec![("authorization", "Bearer key-one")],
5679 vec![("authorization", "Bearer key-two")],
5680 vec![],
5681 ] {
5682 let (status, _) =
5683 post_json_with_headers(&app, "/v1/chat/completions", body.clone(), &headers).await;
5684 assert_eq!(status, StatusCode::OK);
5685 }
5686
5687 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5688 let recent = stats["recent"].as_array().unwrap();
5689 assert_eq!(recent.len(), 3);
5690 let one = recent[0]["via_api_key"].as_str().unwrap();
5691 let two = recent[1]["via_api_key"].as_str().unwrap();
5692 assert_ne!(one, two, "two keys must not collapse into one caller");
5693 assert!(
5694 recent[2]["via_api_key"].is_null(),
5695 "an unauthenticated call is null, not a fingerprint of nothing"
5696 );
5697 assert!(recent[2]["client"].is_null());
5698 }
5699
5700 #[tokio::test]
5705 async fn a_row_names_the_model_that_served_it_not_the_one_requested() {
5706 let state = Arc::new(test_state(
5707 named_test_model("really-loaded", 256),
5708 ResponseCache::new(4, Duration::from_secs(60)),
5709 ));
5710 let app = test_app_with_state(Arc::clone(&state));
5711
5712 let (status, _) = post_json_uri(
5713 &app,
5714 "/v1/chat/completions",
5715 serde_json::json!({
5716 "model": "gpt-4-turbo-that-is-not-here",
5717 "messages": [{"role": "user", "content": "hi"}],
5718 "max_tokens": 2
5719 }),
5720 )
5721 .await;
5722 assert_eq!(status, StatusCode::OK);
5723
5724 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5725 assert_eq!(stats["recent"][0]["model"], "really-loaded");
5726
5727 state.swap_active(None);
5730 let (status, _) = post_json_uri(
5731 &app,
5732 "/v1/chat/completions",
5733 serde_json::json!({
5734 "model": "gpt-4-turbo-that-is-not-here",
5735 "messages": [{"role": "user", "content": "hi"}]
5736 }),
5737 )
5738 .await;
5739 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5740 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5741 let recent = stats["recent"].as_array().unwrap();
5742 assert!(recent[recent.len() - 1]["model"].is_null());
5743 }
5744
5745 #[tokio::test]
5749 async fn a_streamed_row_names_the_model_it_decoded_against() {
5750 let state = Arc::new(test_state(
5751 named_test_model("model-before", 256),
5752 ResponseCache::new(4, Duration::from_secs(60)),
5753 ));
5754 let app = test_app_with_state(Arc::clone(&state));
5755 let _ = post_sse_raw(&app, resumable_request()).await;
5756 active_model(&state, "model-after");
5758
5759 let (_, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5760 assert_eq!(stats["recent"][0]["model"], "model-before");
5761 }
5762
5763 #[tokio::test]
5766 async fn the_queue_gauge_is_null_when_nothing_can_queue() {
5767 let app = test_app();
5768 let (status, stats) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
5769 assert_eq!(status, StatusCode::OK);
5770 assert!(
5771 stats["queue_depth"].is_null(),
5772 "without continuous batching nothing queues, so there is nothing to measure"
5773 );
5774 assert!(stats["queue_rejected_total"].is_null());
5775 assert_eq!(
5776 stats["generating_now"], 0,
5777 "work in progress is measured and really is zero here"
5778 );
5779 }
5780
5781 async fn post_sse_raw(app: &Router, body: serde_json::Value) -> String {
5786 post_sse_raw_uri(app, ferrox_api::routes::V1_CHAT_COMPLETIONS, body).await
5787 }
5788
5789 async fn post_sse_raw_uri(app: &Router, uri: &str, body: serde_json::Value) -> String {
5793 use http_body_util::BodyExt;
5794 use tower::ServiceExt;
5795
5796 let response = app
5797 .clone()
5798 .oneshot(
5799 axum::http::Request::builder()
5800 .method("POST")
5801 .uri(uri)
5802 .header("content-type", "application/json")
5803 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
5804 .unwrap(),
5805 )
5806 .await
5807 .unwrap();
5808 let bytes = response.into_body().collect().await.unwrap().to_bytes();
5809 String::from_utf8(bytes.to_vec()).unwrap()
5810 }
5811
5812 async fn get_json_with_headers(
5813 app: &Router,
5814 uri: &str,
5815 headers: &[(&str, &str)],
5816 ) -> (StatusCode, serde_json::Value) {
5817 use http_body_util::BodyExt;
5818 use tower::ServiceExt;
5819
5820 let mut builder = axum::http::Request::builder().method("GET").uri(uri);
5821 for (name, value) in headers {
5822 builder = builder.header(*name, *value);
5823 }
5824 let response = app
5825 .clone()
5826 .oneshot(builder.body(axum::body::Body::empty()).unwrap())
5827 .await
5828 .unwrap();
5829 let status = response.status();
5830 let bytes = response.into_body().collect().await.unwrap().to_bytes();
5831 (
5832 status,
5833 serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({})),
5834 )
5835 }
5836
5837 fn sse_field<'a>(body: &'a str, field: &str) -> Vec<&'a str> {
5838 body.lines()
5839 .filter_map(|line| line.strip_prefix(field))
5840 .map(str::trim)
5841 .collect()
5842 }
5843
5844 fn resumable_request() -> serde_json::Value {
5845 serde_json::json!({
5846 "model": "m",
5847 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
5848 "max_tokens": 4,
5849 "temperature": 0,
5850 "stream": true,
5851 "stream_resumable": true,
5852 })
5853 }
5854
5855 #[tokio::test]
5860 async fn a_resumable_stream_numbers_every_event_and_states_retry_once() {
5861 let app = test_app();
5862 let body = post_sse_raw(&app, resumable_request()).await;
5863
5864 let request_id = body
5865 .lines()
5866 .find_map(|l| l.strip_prefix("data: "))
5867 .and_then(|d| serde_json::from_str::<serde_json::Value>(d).ok())
5868 .and_then(|v| v["request_id"].as_str().map(str::to_string))
5869 .expect("the first chunk names the request");
5870
5871 let ids = sse_field(&body, "id:");
5872 let datas = sse_field(&body, "data:");
5873 assert_eq!(
5874 ids.len(),
5875 datas.len(),
5876 "every event carries an id, or a reconnect cannot name where it stopped"
5877 );
5878 for (i, id) in ids.iter().enumerate() {
5879 assert_eq!(*id, format!("{request_id}:{i}"));
5880 }
5881 let retries = sse_field(&body, "retry:");
5882 assert_eq!(
5883 retries.len(),
5884 1,
5885 "the reconnect delay is stated once, not on every event"
5886 );
5887 assert_eq!(retries[0], "1500");
5888 assert!(
5889 body.contains("data: [DONE]"),
5890 "the end of stream is still stated"
5891 );
5892 }
5893
5894 #[tokio::test]
5898 async fn a_plain_stream_carries_no_id_because_nothing_could_replay_it() {
5899 let app = test_app();
5900 let mut request = resumable_request();
5901 request["stream_resumable"] = serde_json::json!(false);
5902 let body = post_sse_raw(&app, request).await;
5903 assert!(!sse_field(&body, "data:").is_empty(), "it still streams");
5904 assert!(
5905 sse_field(&body, "id:").is_empty(),
5906 "an id promises a replay this stream cannot serve"
5907 );
5908 assert!(sse_field(&body, "retry:").is_empty());
5909 }
5910
5911 #[tokio::test]
5915 async fn the_polling_fallback_serves_exactly_what_the_stream_delivered() {
5916 let app = test_app();
5917 let body = post_sse_raw(&app, resumable_request()).await;
5918 let request_id = sse_field(&body, "id:")[0]
5919 .rsplit_once(':')
5920 .unwrap()
5921 .0
5922 .to_string();
5923 let streamed: Vec<String> = sse_field(&body, "data:")
5924 .iter()
5925 .map(|d| d.to_string())
5926 .collect();
5927
5928 let (status, polled) = get_json(
5929 &app,
5930 &format!("{}?from=0", ferrox_api::routes::v1_stream_poll(&request_id)),
5931 )
5932 .await;
5933 assert_eq!(status, StatusCode::OK);
5934 let events: Vec<String> = polled["events"]
5935 .as_array()
5936 .unwrap()
5937 .iter()
5938 .map(|e| e["data"].as_str().unwrap().to_string())
5939 .collect();
5940 assert_eq!(
5941 events, streamed,
5942 "the fallback must deliver the same answer, not a re-run of it"
5943 );
5944 assert_eq!(polled["request_id"], request_id);
5945 assert_eq!(
5946 polled["done"], false,
5947 "events were still being handed out, so the client must ask again"
5948 );
5949
5950 let next = polled["next_index"].as_u64().unwrap();
5953 let (_, drained) = get_json(
5954 &app,
5955 &format!(
5956 "{}?from={next}",
5957 ferrox_api::routes::v1_stream_poll(&request_id)
5958 ),
5959 )
5960 .await;
5961 assert_eq!(drained["done"], true);
5962 assert_eq!(drained["events"].as_array().unwrap().len(), 0);
5963 }
5964
5965 #[tokio::test]
5969 async fn a_resume_continues_after_the_last_event_id_rather_than_repeating() {
5970 let app = test_app();
5971 let body = post_sse_raw(&app, resumable_request()).await;
5972 let ids = sse_field(&body, "id:");
5973 let datas: Vec<String> = sse_field(&body, "data:")
5974 .iter()
5975 .map(|d| d.to_string())
5976 .collect();
5977 assert!(
5978 ids.len() >= 3,
5979 "need a few events to resume into the middle"
5980 );
5981 let request_id = ids[0].rsplit_once(':').unwrap().0.to_string();
5982
5983 let (status, resumed) = get_json_with_headers(
5984 &app,
5985 &format!("{}/poll", ferrox_api::routes::v1_stream(&request_id)),
5986 &[],
5987 )
5988 .await;
5989 assert_eq!(status, StatusCode::OK);
5990 assert_eq!(resumed["events"].as_array().unwrap().len(), datas.len());
5991
5992 let (_, tail) = get_json(
5994 &app,
5995 &format!("{}?from=2", ferrox_api::routes::v1_stream_poll(&request_id)),
5996 )
5997 .await;
5998 let tail_events: Vec<String> = tail["events"]
5999 .as_array()
6000 .unwrap()
6001 .iter()
6002 .map(|e| e["data"].as_str().unwrap().to_string())
6003 .collect();
6004 assert_eq!(tail_events, datas[2..].to_vec());
6005 }
6006
6007 #[tokio::test]
6010 async fn an_sse_reconnect_resumes_from_the_last_event_id() {
6011 use http_body_util::BodyExt;
6012 use tower::ServiceExt;
6013
6014 let app = test_app();
6015 let body = post_sse_raw(&app, resumable_request()).await;
6016 let ids = sse_field(&body, "id:");
6017 let datas: Vec<String> = sse_field(&body, "data:")
6018 .iter()
6019 .map(|d| d.to_string())
6020 .collect();
6021 let request_id = ids[0].rsplit_once(':').unwrap().0.to_string();
6022
6023 let response = app
6024 .clone()
6025 .oneshot(
6026 axum::http::Request::builder()
6027 .method("GET")
6028 .uri(ferrox_api::routes::v1_stream(&request_id))
6029 .header("last-event-id", format!("{request_id}:0"))
6030 .body(axum::body::Body::empty())
6031 .unwrap(),
6032 )
6033 .await
6034 .unwrap();
6035 assert_eq!(response.status(), StatusCode::OK);
6036 assert_eq!(
6037 response
6038 .headers()
6039 .get("x-accel-buffering")
6040 .and_then(|v| v.to_str().ok()),
6041 Some("no"),
6042 "the reconnect needs the same anti-buffering header as the stream"
6043 );
6044 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6045 let resumed = String::from_utf8(bytes.to_vec()).unwrap();
6046 assert_eq!(
6047 sse_field(&resumed, "data:")
6048 .iter()
6049 .map(|d| d.to_string())
6050 .collect::<Vec<_>>(),
6051 datas[1..].to_vec()
6052 );
6053 assert_eq!(sse_field(&resumed, "id:")[0], format!("{request_id}:1"));
6054 }
6055
6056 #[tokio::test]
6060 async fn a_last_event_id_from_another_stream_is_refused() {
6061 let app = test_app();
6062 let body = post_sse_raw(&app, resumable_request()).await;
6063 let request_id = sse_field(&body, "id:")[0]
6064 .rsplit_once(':')
6065 .unwrap()
6066 .0
6067 .to_string();
6068
6069 let (status, err) = get_json_with_headers(
6070 &app,
6071 &ferrox_api::routes::v1_stream(&request_id),
6072 &[("last-event-id", "chatcmpl-someone-else:3")],
6073 )
6074 .await;
6075 assert_eq!(status, StatusCode::BAD_REQUEST);
6076 assert_eq!(err["error"]["code"], "bad_last_event_id");
6077 }
6078
6079 #[tokio::test]
6083 async fn resuming_a_stream_that_was_never_resumable_is_a_404_that_says_why() {
6084 let app = test_app();
6085 let mut request = resumable_request();
6086 request["stream_resumable"] = serde_json::json!(false);
6087 let body = post_sse_raw(&app, request).await;
6088 let request_id = body
6089 .lines()
6090 .find_map(|l| l.strip_prefix("data: "))
6091 .and_then(|d| serde_json::from_str::<serde_json::Value>(d).ok())
6092 .and_then(|v| v["request_id"].as_str().map(str::to_string))
6093 .unwrap();
6094
6095 let (status, err) = get_json(&app, &ferrox_api::routes::v1_stream_poll(&request_id)).await;
6096 assert_eq!(status, StatusCode::NOT_FOUND);
6097 assert_eq!(err["error"]["code"], "stream_not_found");
6098 assert!(err["error"]["message"]
6099 .as_str()
6100 .unwrap()
6101 .contains("stream_resumable"));
6102 }
6103
6104 #[test]
6108 fn the_axum_stream_patterns_match_the_published_templates() {
6109 assert_eq!(
6110 axum_path(ferrox_api::routes::V1_STREAM),
6111 "/v1/stream/:request_id"
6112 );
6113 assert_eq!(
6114 axum_path(ferrox_api::routes::V1_STREAM_POLL),
6115 "/v1/stream/:request_id/poll"
6116 );
6117 assert_eq!(
6118 ferrox_api::routes::v1_stream("abc"),
6119 axum_path(ferrox_api::routes::V1_STREAM).replace(":request_id", "abc")
6120 );
6121 }
6122
6123 #[test]
6133 fn no_published_template_reaches_the_router_with_its_braces() {
6134 for template in [
6135 ferrox_api::routes::V1_STREAM,
6136 ferrox_api::routes::V1_STREAM_POLL,
6137 ferrox_api::routes::V1_RESPONSE,
6138 ferrox_api::routes::V1_RESPONSE_CANCEL,
6139 ferrox_api::routes::ADMIN_TASK_CANCEL,
6140 ] {
6141 assert!(
6142 template.contains('{'),
6143 "{template} is in the template list but has no placeholder"
6144 );
6145 let mounted = axum_path(template);
6146 assert!(
6147 !mounted.contains('{') && !mounted.contains('}'),
6148 "{template} would be mounted as {mounted}, whose braces axum reads as a literal segment"
6149 );
6150 assert!(
6151 mounted.contains(':'),
6152 "{template} lost its placeholder entirely and would match one path only"
6153 );
6154 }
6155 }
6156
6157 #[tokio::test]
6165 async fn an_unknown_response_id_gets_the_handler_not_a_bare_404() {
6166 let app = test_app();
6167 let (status, body) = get_json(&app, "/v1/responses/resp_nonexistent").await;
6168 assert_eq!(status, StatusCode::NOT_FOUND);
6169 assert!(
6170 !body.is_null(),
6171 "empty body means axum never matched the route, so the id was read as a literal segment"
6172 );
6173 }
6174
6175 #[tokio::test]
6178 async fn the_task_list_starts_empty_rather_than_absent() {
6179 let app = test_app();
6180 let (status, body) = get_json(&app, ferrox_api::routes::ADMIN_TASKS).await;
6181 assert_eq!(status, StatusCode::OK);
6182 assert_eq!(body["tasks"].as_array().unwrap().len(), 0);
6183 }
6184
6185 #[tokio::test]
6194 async fn the_slots_route_is_registered_and_refuses_by_naming_slot_save_path() {
6195 assert!(
6196 std::env::var("FERROX_SLOT_SAVE_PATH").is_err(),
6197 "this test asserts the unconfigured behaviour"
6198 );
6199 let app = test_app();
6200 let (status, body) = post_json_uri(
6201 &app,
6202 &format!("{}?action=save", ferrox_api::routes::slots_id(0)),
6203 serde_json::json!({"filename": "sys.fslot", "prompt": "hi"}),
6204 )
6205 .await;
6206 assert_eq!(status, StatusCode::NOT_IMPLEMENTED);
6207 assert!(
6208 body["error"]["message"]
6209 .as_str()
6210 .unwrap()
6211 .contains("--slot-save-path"),
6212 "{body}"
6213 );
6214 }
6215
6216 async fn post_json_uri(
6217 app: &Router,
6218 uri: &str,
6219 body: serde_json::Value,
6220 ) -> (StatusCode, serde_json::Value) {
6221 use http_body_util::BodyExt;
6222 use tower::ServiceExt;
6223
6224 let response = app
6225 .clone()
6226 .oneshot(
6227 axum::http::Request::builder()
6228 .method("POST")
6229 .uri(uri)
6230 .header("content-type", "application/json")
6231 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
6232 .unwrap(),
6233 )
6234 .await
6235 .unwrap();
6236 let status = response.status();
6237 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6238 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
6239 (status, json)
6240 }
6241
6242 async fn post_json(app: &Router, body: serde_json::Value) -> serde_json::Value {
6243 post_json_uri(app, "/v1/chat/completions", body).await.1
6244 }
6245
6246 #[tokio::test]
6255 async fn stats_says_what_the_engine_is_using_and_which_quantity_that_is() {
6256 let app = test_app();
6257 let (status, body) = get_json(&app, ferrox_api::routes::V1_STATS).await;
6258 assert_eq!(status, StatusCode::OK);
6259
6260 let memory = &body["memory"];
6261 if memory.is_null() {
6262 return;
6265 }
6266 assert!(
6267 memory["bytes"].as_u64().is_some_and(|b| b > 0),
6268 "a read that produced a zero is a broken read, not an idle \
6269 engine: {memory}"
6270 );
6271 assert!(
6272 ["pss", "rss"].contains(&memory["kind"].as_str().unwrap_or("")),
6273 "the quantity must travel with the number: {memory}"
6274 );
6275 }
6276
6277 #[tokio::test]
6283 async fn stats_reports_a_pool_it_does_not_have_as_absent_and_not_as_zero() {
6284 let app = test_app();
6285 let (status, body) = get_json(&app, ferrox_api::routes::V1_STATS).await;
6286 assert_eq!(status, StatusCode::OK);
6287 for pool in ["kv_pages", "window_slots", "state_slots"] {
6288 assert!(
6289 body["pools"][pool].is_null(),
6290 "{pool} must be null rather than a zero row: {}",
6291 body["pools"]
6292 );
6293 }
6294 }
6295
6296 #[tokio::test]
6304 async fn a_messages_response_states_the_id_that_v1_cancel_takes() {
6305 use http_body_util::BodyExt;
6306 use tower::ServiceExt;
6307
6308 let app = test_app();
6309 let send = |body: serde_json::Value| {
6310 let app = app.clone();
6311 async move {
6312 app.oneshot(
6313 axum::http::Request::builder()
6314 .method("POST")
6315 .uri(ferrox_api::routes::V1_MESSAGES)
6316 .header("content-type", "application/json")
6317 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
6318 .unwrap(),
6319 )
6320 .await
6321 .unwrap()
6322 }
6323 };
6324
6325 let ok = send(serde_json::json!({
6326 "model": "test",
6327 "max_tokens": 1,
6328 "messages": [{"role": "user", "content": "hi"}],
6329 }))
6330 .await;
6331 assert_eq!(ok.status(), StatusCode::OK);
6332 let id = ok
6333 .headers()
6334 .get("request-id")
6335 .expect("a served message names its id")
6336 .to_str()
6337 .unwrap()
6338 .to_string();
6339 assert!(!id.is_empty());
6340
6341 let bad = send(serde_json::json!({"model": "test"})).await;
6344 assert!(bad.status().is_client_error());
6345 let other = bad.headers().get("request-id").expect("errors too");
6346 assert_ne!(other.to_str().unwrap(), id);
6347 let _ = bad.into_body().collect().await.unwrap();
6348 }
6349
6350 #[tokio::test]
6358 async fn a_request_that_arrives_mid_rebuild_is_refused_and_admitted_again_after() {
6359 let state = Arc::new(test_state(
6360 test_model_full_byte_vocab(),
6361 ResponseCache::new(1000, Duration::from_secs(3600)),
6362 ));
6363 let app = test_app_with_state(Arc::clone(&state));
6364 let body = serde_json::json!({
6365 "model": "test",
6366 "messages": [{"role": "user", "content": "hi"}],
6367 "max_tokens": 1,
6368 });
6369
6370 state
6371 .maintenance
6372 .lock()
6373 .unwrap()
6374 .begin_rebuild()
6375 .expect("a fresh server is serving, so the rebuild starts");
6376 let (status, refused) = post_json_uri(&app, "/v1/chat/completions", body.clone()).await;
6377 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
6378 assert_eq!(refused["error"]["type"], "cache_rebuilding");
6379
6380 state.maintenance.lock().unwrap().finish_rebuild(true);
6381 let (status, _) = post_json_uri(&app, "/v1/chat/completions", body).await;
6382 assert_eq!(
6383 status,
6384 StatusCode::OK,
6385 "the gate reopens; a rebuild is not a latch"
6386 );
6387 }
6388
6389 #[tokio::test]
6394 async fn cancelling_an_id_that_is_not_generating_is_a_404_that_says_so() {
6395 let app = test_app();
6396 let (status, body) = post_json_uri(
6397 &app,
6398 ferrox_api::routes::V1_CANCEL,
6399 serde_json::json!({ "request_id": "chatcmpl-never-issued" }),
6400 )
6401 .await;
6402 assert_eq!(status, StatusCode::NOT_FOUND);
6403 assert_eq!(body["cancelled"], serde_json::json!(false));
6404 assert_eq!(body["request_id"], "chatcmpl-never-issued");
6405 assert!(
6406 body["detail"].as_str().is_some_and(|d| !d.is_empty()),
6407 "the verdict must carry a human reason: {body}"
6408 );
6409 }
6410
6411 #[tokio::test]
6415 async fn cancelling_a_live_generation_signals_its_token_and_answers_200() {
6416 let state = Arc::new(test_state(
6417 test_model_full_byte_vocab(),
6418 ResponseCache::new(1000, Duration::from_secs(3600)),
6419 ));
6420 let app = test_app_with_state(Arc::clone(&state));
6421 let (token, _guard) = state.cancels.register("chatcmpl-live");
6422
6423 let (status, before) = get_json(&app, ferrox_api::routes::ADMIN_STATS).await;
6424 assert_eq!(status, StatusCode::OK);
6425 assert_eq!(before["generating_now"], serde_json::json!(1));
6426
6427 let (status, body) = post_json_uri(
6428 &app,
6429 ferrox_api::routes::V1_CANCEL,
6430 serde_json::json!({ "request_id": "chatcmpl-live" }),
6431 )
6432 .await;
6433 assert_eq!(status, StatusCode::OK);
6434 assert_eq!(body["cancelled"], serde_json::json!(true));
6435 assert!(
6436 token.is_cancelled(),
6437 "the endpoint answered ok without setting the flag the decode loop reads"
6438 );
6439 }
6440
6441 #[tokio::test]
6442 async fn tokenize_detokenize_roundtrip_and_embeddings_mean() {
6443 let app = test_app();
6444 let (status, tok) =
6445 post_json_uri(&app, "/v1/tokenize", serde_json::json!({ "prompt": "Hi" })).await;
6446 assert_eq!(status, StatusCode::OK);
6447 let tokens = tok["tokens"].as_array().unwrap();
6448 assert_eq!(tok["count"], tokens.len());
6449 assert!(!tokens.is_empty());
6450
6451 let (status, detok) = post_json_uri(
6452 &app,
6453 "/v1/detokenize",
6454 serde_json::json!({ "tokens": tokens }),
6455 )
6456 .await;
6457 assert_eq!(status, StatusCode::OK);
6458 assert_eq!(detok["text"], "Hi");
6459
6460 let (status, emb) = post_json_uri(
6461 &app,
6462 "/v1/embeddings",
6463 serde_json::json!({
6464 "input": "Hi",
6465 "embedding_type": "mean"
6466 }),
6467 )
6468 .await;
6469 assert_eq!(status, StatusCode::OK);
6470 let vec = emb["data"][0]["embedding"].as_array().unwrap();
6471 assert!(!vec.is_empty());
6472 assert!(vec.iter().all(|v| v.as_f64().is_some()));
6473 }
6474
6475 #[tokio::test]
6481 async fn the_decoder_path_still_refuses_a_pooling_it_cannot_mean() {
6482 let app = test_app();
6483 let (status, body) = post_json_uri(
6484 &app,
6485 "/v1/embeddings",
6486 serde_json::json!({ "input": "Hi", "embedding_type": "cls" }),
6487 )
6488 .await;
6489 assert_eq!(status, StatusCode::BAD_REQUEST);
6490 let msg = body["error"]["message"].as_str().unwrap();
6491 assert!(msg.contains("mean") && msg.contains("last"), "{msg}");
6492 }
6493
6494 #[tokio::test]
6499 #[ignore = "needs models/bge-small-en-v1.5-q8_0.gguf"]
6500 async fn a_real_embedding_model_serves_v1_embeddings() {
6501 let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
6502 .join("../../models/bge-small-en-v1.5-q8_0.gguf");
6503 if !path.exists() {
6504 eprintln!("SKIP: {} not present", path.display());
6505 return;
6506 }
6507 let encoder = ferrox_models::EmbeddingModel::from_gguf_path(&path).expect("load bge");
6508 let mut state = test_state(
6509 test_model_full_byte_vocab(),
6510 ResponseCache::new(1000, Duration::from_secs(3600)),
6511 );
6512 state.embedding = Some(Arc::new(encoder));
6513 let app = test_app_with_state(Arc::new(state));
6514
6515 let (status, body) = post_json_uri(
6516 &app,
6517 "/v1/embeddings",
6518 serde_json::json!({ "input": ["Hello world", "a second input"] }),
6519 )
6520 .await;
6521 assert_eq!(status, StatusCode::OK, "{body}");
6522 assert_eq!(body["model"], "bge-small-en-v1.5");
6523 let data = body["data"].as_array().unwrap();
6524 assert_eq!(data.len(), 2);
6525 for (i, row) in data.iter().enumerate() {
6526 assert_eq!(row["index"], i);
6527 let v: Vec<f64> = row["embedding"]
6528 .as_array()
6529 .unwrap()
6530 .iter()
6531 .map(|x| x.as_f64().unwrap())
6532 .collect();
6533 assert_eq!(v.len(), 384, "the encoder\'s width, not the decoder\'s");
6534 let norm = v.iter().map(|x| x * x).sum::<f64>().sqrt();
6535 assert!((norm - 1.0).abs() < 1e-4, "not L2-normalized: {norm}");
6536 }
6537 assert!(body["usage"]["prompt_tokens"].as_u64().unwrap() >= 4 + 2);
6540
6541 let (status, mean) = post_json_uri(
6545 &app,
6546 "/v1/embeddings",
6547 serde_json::json!({ "input": "Hello world", "embedding_type": "mean" }),
6548 )
6549 .await;
6550 assert_eq!(status, StatusCode::OK);
6551 assert_ne!(mean["data"][0]["embedding"], data[0]["embedding"]);
6552 }
6553
6554 #[tokio::test]
6566 #[ignore = "needs models/bge-small-en-v1.5-q8_0.gguf"]
6567 async fn an_encoder_can_be_the_loaded_model() {
6568 let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
6569 .join("../../models/bge-small-en-v1.5-q8_0.gguf");
6570 if !path.exists() {
6571 eprintln!("SKIP: {} not present", path.display());
6572 return;
6573 }
6574
6575 let loaded = model::load_from_path(path.to_str().unwrap()).expect("load bge as the model");
6579 assert!(
6580 matches!(loaded, model::LoadedModel::Encoder(_)),
6581 "an encoder-only GGUF reached a decoder loader"
6582 );
6583 let (loaded, batcher, ceiling) = activate_loaded_model(loaded, true, None, None);
6584 assert!(
6585 matches!(loaded, Loaded::Encoder(_)),
6586 "the encoder did not stay an encoder through activation"
6587 );
6588 assert!(
6589 batcher.is_none() && ceiling.is_none(),
6590 "an encoder was given a decode batcher or a KV ceiling it has no use for"
6591 );
6592
6593 let state = test_state(
6594 test_model_full_byte_vocab(),
6595 ResponseCache::new(1000, Duration::from_secs(3600)),
6596 );
6597 state.swap_active(Some(Arc::new(ActiveModel {
6598 id: None,
6599 loaded,
6600 batcher,
6601 ceiling,
6602 checkpoint_path: None,
6603 })));
6604 let app = test_app_with_state(Arc::new(state));
6605
6606 let (status, body) = post_json_uri(
6608 &app,
6609 "/v1/embeddings",
6610 serde_json::json!({ "input": "Hello world" }),
6611 )
6612 .await;
6613 assert_eq!(status, StatusCode::OK, "{body}");
6614 assert_eq!(body["model"], "bge-small-en-v1.5");
6615 let v = body["data"][0]["embedding"].as_array().unwrap();
6616 assert_eq!(v.len(), 384, "the encoder's width, not the decoder's");
6617
6618 let (status, body) = post_json_uri(
6620 &app,
6621 "/v1/chat/completions",
6622 serde_json::json!({
6623 "model": "bge-small-en-v1.5",
6624 "messages": [{"role": "user", "content": "hi"}],
6625 }),
6626 )
6627 .await;
6628 assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{body}");
6629 let msg = body["error"]["message"].as_str().unwrap();
6630 for fact in [
6631 "bge-small-en-v1.5",
6632 "bert",
6633 "embedding model",
6634 "/v1/embeddings",
6635 ] {
6636 assert!(msg.contains(fact), "the refusal does not say {fact}: {msg}");
6637 }
6638
6639 let (status, models) = get_json(&app, ferrox_api::routes::V1_MODELS).await;
6641 assert_eq!(status, StatusCode::OK);
6642 let entry = &models["data"][0];
6643 assert_eq!(entry["id"], "bge-small-en-v1.5");
6644 assert_eq!(entry["ferrox_model_kind"], "embedding");
6645 assert_eq!(entry["ferrox_tokenizer"], "gguf-wordpiece");
6646 assert_eq!(entry["ferrox_n_embd"], 384);
6647 assert_eq!(entry["ferrox_pooling"], "CLS");
6648 assert_eq!(
6649 entry["ferrox_endpoints"],
6650 serde_json::json!(["/v1/embeddings"])
6651 );
6652 assert!(entry.get("supported_reasoning_efforts").is_none());
6655
6656 let (status, health) = get_json(&app, ferrox_api::routes::HEALTH).await;
6658 assert_eq!(status, StatusCode::OK, "an encoder is a loaded model");
6659 assert_eq!(health["model"]["id"], "bge-small-en-v1.5");
6660 assert_eq!(health["model"]["synthetic_weights"], false);
6661 let weights = health["capabilities"]
6662 .as_array()
6663 .unwrap()
6664 .iter()
6665 .find(|c| c["id"] == ferrox_api::health::capability::REAL_WEIGHTS)
6666 .expect("a real-weights capability row");
6667 let detail = weights["detail"].as_str().unwrap_or_default();
6668 assert!(detail.contains("ENCODER"), "{detail}");
6669 let (status, body) = post_json_uri(
6677 &app,
6678 ferrox_api::routes::V1_TOKENIZE,
6679 serde_json::json!({ "content": "hello world" }),
6680 )
6681 .await;
6682 assert_eq!(
6683 status,
6684 StatusCode::OK,
6685 "an encoder has a real tokenizer: {body}"
6686 );
6687 let tokens = body["tokens"].as_array().expect("tokens array").clone();
6688 assert!(!tokens.is_empty(), "WordPiece produced nothing: {body}");
6689
6690 let (status, body) = post_json_uri(
6691 &app,
6692 ferrox_api::routes::V1_DETOKENIZE,
6693 serde_json::json!({ "tokens": tokens }),
6694 )
6695 .await;
6696 assert_eq!(status, StatusCode::OK, "{body}");
6697 let round_tripped = body["content"].as_str().expect("content").to_string();
6698 assert!(
6699 round_tripped.contains("hello") && round_tripped.contains("world"),
6700 "the ids did not decode back through the encoder's own vocabulary: {round_tripped}"
6701 );
6702
6703 let (status, _) = post_json_uri(
6706 &app,
6707 "/v1/completions",
6708 serde_json::json!({ "model": "m", "prompt": "hi", "max_tokens": 1 }),
6709 )
6710 .await;
6711 assert_eq!(
6712 status,
6713 StatusCode::NOT_IMPLEMENTED,
6714 "tokenizing an encoder must not have opened a path to generating with one"
6715 );
6716 }
6717
6718 #[tokio::test]
6723 async fn metrics_exposes_expert_store_counters_when_streaming_is_active() {
6724 use http_body_util::BodyExt;
6725 use tower::ServiceExt;
6726
6727 let fixture = concat!(
6728 "../ferrox-models/tests/fixtures/",
6729 "ferrox_real_moe_test.gguf"
6730 );
6731 let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join(fixture);
6732 let decoder = Decoder::from_gguf_with_expert_cache(
6733 &fixture,
6734 ferrox_models::config::test_moe_fixture(),
6735 Some(1024 * 1024),
6736 )
6737 .expect("MoE fixture must load store-backed");
6738
6739 let mut caches: Vec<ferrox_core::cache::KvCache> = decoder.config.new_kv_caches();
6743 decoder.forward_token(1, 0, &mut caches);
6744
6745 let model = Model::Gguf(GgufModel {
6746 decoder: Arc::new(decoder),
6747 tokenizer: Arc::new(ServerTokenizer::Byte),
6748 stop_tokens: StopTokens::default(),
6749 bos_id: None,
6750 is_synthetic: false,
6751 chat_template: chat_template::PromptTemplate::plain(),
6752 });
6753 let state = Arc::new(test_state(
6754 model,
6755 ResponseCache::new(16, Duration::from_secs(60)),
6756 ));
6757 let app = Router::new()
6758 .route("/metrics", axum::routing::get(metrics))
6759 .route("/v1/chat/completions", post(chat_completions))
6760 .with_state(state);
6761
6762 let fetch_metrics = |app: Router| async move {
6763 let resp = app
6764 .oneshot(
6765 axum::http::Request::builder()
6766 .method("GET")
6767 .uri("/metrics")
6768 .body(axum::body::Body::empty())
6769 .unwrap(),
6770 )
6771 .await
6772 .unwrap();
6773 let bytes = resp.into_body().collect().await.unwrap().to_bytes();
6774 String::from_utf8(bytes.to_vec()).unwrap()
6775 };
6776
6777 let after = fetch_metrics(app.clone()).await;
6778 assert!(
6779 after.contains("ferrox_expert_cache_misses_total"),
6780 "streaming model must expose expert-cache metrics: {after}"
6781 );
6782 let misses: u64 = after
6783 .lines()
6784 .find(|l| l.starts_with("ferrox_expert_cache_misses_total"))
6785 .and_then(|l| l.split_whitespace().nth(1))
6786 .and_then(|v| v.parse().ok())
6787 .expect("misses metric line must parse");
6788 assert!(
6789 misses > 0,
6790 "decode must have read experts through the store: {after}"
6791 );
6792 }
6793
6794 fn weather_tool() -> serde_json::Value {
6795 serde_json::json!({
6796 "type": "function",
6797 "function": {
6798 "name": "get_weather",
6799 "description": "Get the current weather for a location.",
6800 "parameters": {
6801 "type": "object",
6802 "properties": {"location": {"type": "string"}},
6803 "required": ["location"]
6804 }
6805 }
6806 })
6807 }
6808
6809 fn weather_tool_def() -> ToolDef {
6810 ToolDef {
6811 kind: "function".to_string(),
6812 function: ToolFunctionDef {
6813 name: "get_weather".to_string(),
6814 description: Some("Get the current weather for a location.".to_string()),
6815 parameters: Some(serde_json::json!({
6816 "type": "object",
6817 "properties": {"location": {"type": "string"}},
6818 "required": ["location"]
6819 })),
6820 },
6821 }
6822 }
6823
6824 #[test]
6825 fn tool_preamble_mentions_every_tool_name_and_description() {
6826 let preamble = tool_preamble(&[weather_tool_def()]);
6827 assert!(preamble.contains("get_weather"));
6828 assert!(preamble.contains("Get the current weather for a location."));
6829 assert!(preamble.contains("<tool_call>"));
6830 assert!(preamble.contains("</tool_call>"));
6831 }
6832
6833 #[test]
6834 fn a_real_marker_becomes_a_structured_tool_call() {
6835 let text = "sure, let me check.<tool_call>{\"name\": \"get_weather\", \"arguments\": {\"location\": \"Paris\"}}</tool_call>";
6836 let (message, finish) = build_response_message(
6837 text.to_string(),
6838 &[weather_tool_def()],
6839 output::OutputPosture::for_model("test-model"),
6840 "stop",
6841 );
6842 assert_eq!(finish, "tool_calls");
6843 let calls = message.tool_calls.expect("must carry a tool call");
6844 assert_eq!(calls[0].function.name, "get_weather");
6845 let parsed: serde_json::Value = serde_json::from_str(&calls[0].function.arguments).unwrap();
6846 assert_eq!(parsed["location"], "Paris");
6847 }
6848
6849 #[test]
6850 fn a_plain_answer_is_not_promoted_to_a_tool_call() {
6851 let (message, finish) = build_response_message(
6852 "just an answer".to_string(),
6853 &[weather_tool_def()],
6854 output::OutputPosture::for_model("test-model"),
6855 "stop",
6856 );
6857 assert_eq!(finish, "stop");
6858 assert!(message.tool_calls.is_none());
6859 assert_eq!(message.content.as_deref(), Some("just an answer"));
6860 }
6861
6862 #[test]
6865 fn a_malformed_payload_is_not_a_tool_call() {
6866 let (message, finish) = build_response_message(
6867 "<tool_call>not valid json at all</tool_call>".to_string(),
6868 &[weather_tool_def()],
6869 output::OutputPosture::for_model("test-model"),
6870 "stop",
6871 );
6872 assert_eq!(finish, "stop");
6873 assert!(message.tool_calls.is_none());
6874 }
6875
6876 #[test]
6879 fn a_tool_that_was_never_offered_is_not_returned() {
6880 let (message, finish) = build_response_message(
6881 "<tool_call>{\"name\": \"ping\", \"arguments\": {}}</tool_call>".to_string(),
6882 &[weather_tool_def()],
6883 output::OutputPosture::for_model("test-model"),
6884 "stop",
6885 );
6886 assert_eq!(finish, "stop");
6887 assert!(message.tool_calls.is_none());
6888 }
6889
6890 #[test]
6892 fn marker_text_with_no_tools_offered_stays_content() {
6893 let (message, finish) = build_response_message(
6894 "<tool_call>{\"name\": \"get_weather\", \"arguments\": {}}</tool_call>".to_string(),
6895 &[],
6896 output::OutputPosture::for_model("test-model"),
6897 "stop",
6898 );
6899 assert_eq!(finish, "stop");
6900 assert!(message.tool_calls.is_none());
6901 assert!(message.content.is_some());
6902 }
6903
6904 #[test]
6908 fn a_streamed_call_opens_then_delivers_its_arguments_in_pieces() {
6909 let opened = std::cell::Cell::new(0usize);
6910 let mut parser = crate::policy::parser::ToolCallParser::new(
6911 crate::policy::parser::ToolCallFormat::Qwen3Coder,
6912 vec![
6913 crate::policy::parser::tool_call::ToolSchema::with_parameters(
6914 "write_file",
6915 serde_json::json!({"type": "object", "properties": {
6916 "path": {"type": "string"},
6917 "contents": {"type": "string"}
6918 }}),
6919 ),
6920 ],
6921 );
6922 let wire = "<tool_call><function=write_file>\
6923 <parameter=path>\n/tmp/x\n</parameter>\
6924 <parameter=contents>\nhello world\n</parameter>\
6925 </function></tool_call>";
6926
6927 let mut deltas = Vec::new();
6928 let mut text = String::new();
6929 for piece in wire.as_bytes().chunks(7) {
6930 let chunk = String::from_utf8_lossy(piece).into_owned();
6931 let (more_text, more) = tool_call_deltas(parser.push(&chunk), &opened);
6932 text.push_str(&more_text);
6933 deltas.extend(more);
6934 }
6935 let (more_text, more) = tool_call_deltas(parser.finish(), &opened);
6936 text.push_str(&more_text);
6937 deltas.extend(more);
6938
6939 assert_eq!(opened.get(), 1, "one call opened");
6940 assert!(text.is_empty(), "the markers are not content: {text:?}");
6941
6942 let first = &deltas[0];
6943 assert_eq!(first.index, 0);
6944 assert_eq!(first.id.as_deref(), Some("call_0"));
6945 assert_eq!(first.kind, Some("function"));
6946 assert_eq!(first.function.name.as_deref(), Some("write_file"));
6947
6948 let joined: String = deltas
6951 .iter()
6952 .filter_map(|d| d.function.arguments.clone())
6953 .collect();
6954 let parsed: serde_json::Value =
6955 serde_json::from_str(&joined).expect("the fragments concatenate to valid JSON");
6956 assert_eq!(parsed["path"], serde_json::json!("/tmp/x"));
6957 assert_eq!(parsed["contents"], serde_json::json!("hello world"));
6958 assert!(
6959 deltas.len() >= 3,
6960 "the arguments arrived in pieces, not whole: {}",
6961 deltas.len()
6962 );
6963 assert!(
6964 deltas[1..].iter().all(|d| d.function.name.is_none()),
6965 "only the opening delta carries identity"
6966 );
6967 }
6968
6969 #[test]
6971 fn text_around_a_streamed_call_is_still_content() {
6972 let opened = std::cell::Cell::new(0usize);
6973 let mut parser = crate::policy::parser::ToolCallParser::new(
6974 crate::policy::parser::ToolCallFormat::Qwen25,
6975 vec![crate::policy::parser::tool_call::ToolSchema::new(
6976 "get_weather",
6977 )],
6978 );
6979 let wire = "let me check. <tool_call>{\"name\": \"get_weather\", \
6980 \"arguments\": {}}</tool_call> done";
6981 let mut text = String::new();
6982 for piece in wire.as_bytes().chunks(5) {
6983 let chunk = String::from_utf8_lossy(piece).into_owned();
6984 let (more, _) = tool_call_deltas(parser.push(&chunk), &opened);
6985 text.push_str(&more);
6986 }
6987 let (more, _) = tool_call_deltas(parser.finish(), &opened);
6988 text.push_str(&more);
6989
6990 assert_eq!(opened.get(), 1);
6991 assert!(text.starts_with("let me check. "), "{text:?}");
6992 assert!(text.ends_with(" done"), "{text:?}");
6993 assert!(!text.contains("<tool_call>"), "markers leaked: {text:?}");
6994 }
6995
6996 #[test]
6999 fn a_reasoning_block_is_split_out_of_the_answer() {
7000 let (message, finish) = build_response_message(
7001 "<think>weighing it up</think>The answer is 4.".to_string(),
7002 &[],
7003 output::OutputPosture::for_model("Qwen3-8B"),
7004 "stop",
7005 );
7006 assert_eq!(finish, "stop");
7007 assert_eq!(message.content.as_deref(), Some("The answer is 4."));
7008 assert_eq!(message.reasoning_content.as_deref(), Some("weighing it up"));
7009 }
7010
7011 #[test]
7014 fn a_non_reasoning_model_keeps_a_literal_marker_in_its_answer() {
7015 let (message, _) = build_response_message(
7016 "Use the <think> tag like this.".to_string(),
7017 &[],
7018 output::OutputPosture::for_model("llama-3.1-8b"),
7019 "stop",
7020 );
7021 assert_eq!(
7022 message.content.as_deref(),
7023 Some("Use the <think> tag like this.")
7024 );
7025 assert!(message.reasoning_content.is_none());
7026 }
7027
7028 #[tokio::test]
7036 async fn a_request_with_no_tools_or_session_behaves_exactly_as_before() {
7037 let app = test_app();
7038 let body = serde_json::json!({
7039 "model": "m",
7040 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7041 "max_tokens": 4,
7042 "temperature": 0,
7043 });
7044 let resp = post_json(&app, body).await;
7045 let message = &resp["choices"][0]["message"];
7046 assert!(message["content"].is_string());
7047 assert!(message.get("tool_calls").is_none());
7048 assert_eq!(resp["choices"][0]["finish_reason"], "length");
7049 assert_eq!(resp["usage"]["completion_tokens"], 4);
7050 assert_eq!(
7051 resp["usage"]["total_tokens"],
7052 resp["usage"]["prompt_tokens"].as_u64().unwrap() + 4
7053 );
7054 }
7055
7056 async fn get_json(app: &Router, uri: &str) -> (StatusCode, serde_json::Value) {
7057 use http_body_util::BodyExt;
7058 use tower::ServiceExt;
7059
7060 let response = app
7061 .clone()
7062 .oneshot(
7063 axum::http::Request::builder()
7064 .method("GET")
7065 .uri(uri)
7066 .body(axum::body::Body::empty())
7067 .unwrap(),
7068 )
7069 .await
7070 .unwrap();
7071 let status = response.status();
7072 let bytes = response.into_body().collect().await.unwrap().to_bytes();
7073 (status, serde_json::from_slice(&bytes).unwrap())
7074 }
7075
7076 #[tokio::test]
7077 async fn health_answers_a_capability_handshake_not_a_boolean() {
7078 let app = test_app();
7079 let (status, body) = get_json(&app, ferrox_api::routes::HEALTH).await;
7080 assert_eq!(status, StatusCode::OK);
7081
7082 let health: ferrox_api::HealthResponse = serde_json::from_value(body).unwrap();
7083 assert_eq!(health.state, ferrox_api::HealthState::Ready);
7084 assert!(health.pid > 0);
7085 assert!(health.server_time_unix_ms > 0);
7086 assert_eq!(health.last_request_age_seconds, None);
7089
7090 for id in [
7093 ferrox_api::health::capability::CPU,
7094 ferrox_api::health::capability::METAL,
7095 ferrox_api::health::capability::CUDA,
7096 ferrox_api::health::capability::REAL_WEIGHTS,
7097 ferrox_api::health::capability::CONTINUOUS_BATCHING,
7098 ] {
7099 let cap = health
7100 .capability(id)
7101 .unwrap_or_else(|| panic!("{id} missing"));
7102 assert!(!cap.reason.is_empty(), "{cap:?}");
7103 assert!(!cap.detail.is_empty(), "{cap:?}");
7104 }
7105 let weights = health
7109 .capability(ferrox_api::health::capability::REAL_WEIGHTS)
7110 .unwrap();
7111 assert!(!weights.available);
7112 assert_eq!(weights.reason, ferrox_api::health::reason::MODEL_NOT_LOADED);
7113 assert!(health.model.as_ref().unwrap().synthetic_weights);
7114 }
7115
7116 #[tokio::test]
7117 async fn health_vouches_for_liveness_after_a_request_has_been_served() {
7118 let app = test_app();
7119 let _ = post_json(
7120 &app,
7121 serde_json::json!({
7122 "model": "m",
7123 "messages": [{"role": "user", "content": "\u{1}"}],
7124 "max_tokens": 1,
7125 "temperature": 0,
7126 }),
7127 )
7128 .await;
7129 let (_status, body) = get_json(&app, ferrox_api::routes::HEALTH).await;
7130 let health: ferrox_api::HealthResponse = serde_json::from_value(body).unwrap();
7131 let age = health
7132 .last_request_age_seconds
7133 .expect("a served request is evidence of liveness");
7134 assert!((0.0..5.0).contains(&age), "implausible age {age}");
7135 }
7136
7137 async fn post_sse_chunks(app: &Router, body: serde_json::Value) -> Vec<serde_json::Value> {
7139 use http_body_util::BodyExt;
7140 use tower::ServiceExt;
7141
7142 let response = app
7143 .clone()
7144 .oneshot(
7145 axum::http::Request::builder()
7146 .method("POST")
7147 .uri("/v1/chat/completions")
7148 .header("content-type", "application/json")
7149 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
7150 .unwrap(),
7151 )
7152 .await
7153 .unwrap();
7154 let bytes = response.into_body().collect().await.unwrap().to_bytes();
7155 String::from_utf8(bytes.to_vec())
7156 .unwrap()
7157 .lines()
7158 .filter_map(|line| line.strip_prefix("data: "))
7159 .filter(|payload| *payload != "[DONE]")
7160 .map(|payload| serde_json::from_str(payload).unwrap())
7161 .collect()
7162 }
7163
7164 #[tokio::test]
7165 async fn a_stream_states_its_request_id_once_in_the_first_chunk() {
7166 let app = test_app();
7167 let chunks = post_sse_chunks(
7168 &app,
7169 serde_json::json!({
7170 "model": "m",
7171 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7172 "max_tokens": 4,
7173 "temperature": 0,
7174 "stream": true,
7175 }),
7176 )
7177 .await;
7178
7179 assert!(!chunks.is_empty());
7180 let request_id = chunks[0]["request_id"]
7181 .as_str()
7182 .expect("the first chunk names the request")
7183 .to_string();
7184 assert!(request_id.starts_with("chatcmpl-"), "{request_id}");
7185 for (i, chunk) in chunks.iter().enumerate().skip(1) {
7188 assert!(
7189 chunk.get("request_id").is_none(),
7190 "chunk {i} repeats request_id"
7191 );
7192 }
7193 for chunk in &chunks {
7196 assert_eq!(chunk["id"], serde_json::json!(request_id));
7197 }
7198
7199 let other = post_sse_chunks(
7200 &app,
7201 serde_json::json!({
7202 "model": "m",
7203 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7204 "max_tokens": 4,
7205 "temperature": 0,
7206 "stream": true,
7207 }),
7208 )
7209 .await;
7210 assert_ne!(
7211 other[0]["request_id"].as_str().unwrap(),
7212 request_id,
7213 "two concurrent chats must not share an id"
7214 );
7215 }
7216
7217 #[tokio::test]
7218 async fn a_non_streamed_response_names_the_same_request_id_as_its_completion_id() {
7219 let app = test_app();
7220 let resp = post_json(
7221 &app,
7222 serde_json::json!({
7223 "model": "m",
7224 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7225 "max_tokens": 2,
7226 "temperature": 0,
7227 }),
7228 )
7229 .await;
7230 assert_eq!(resp["id"], resp["request_id"]);
7231 assert!(resp["request_id"]
7232 .as_str()
7233 .unwrap()
7234 .starts_with("chatcmpl-"));
7235 }
7236
7237 #[tokio::test]
7240 async fn usage_carries_separate_prefill_and_decode_timings() {
7241 let app = test_app();
7242 let resp = post_json(
7243 &app,
7244 serde_json::json!({
7245 "model": "m",
7246 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7247 "max_tokens": 4,
7248 "temperature": 0,
7249 }),
7250 )
7251 .await;
7252 let usage = &resp["usage"];
7253 assert!(usage["prompt_eval_duration_ms"].is_number(), "{usage}");
7254 assert!(usage["generation_duration_ms"].is_number(), "{usage}");
7255 assert!(usage["time_to_first_token_ms"].is_number(), "{usage}");
7256 assert!(usage["predicted_per_second"].is_number(), "{usage}");
7257 assert!(usage.get("cached_tokens").is_none(), "{usage}");
7259 }
7260
7261 #[tokio::test]
7269 async fn a_tools_request_with_no_marker_in_the_output_falls_back_to_plain_content() {
7270 let app = test_app();
7271 let body = serde_json::json!({
7272 "model": "m",
7273 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7274 "max_tokens": 4,
7275 "temperature": 0,
7276 "tools": [weather_tool()],
7277 });
7278 let resp = post_json(&app, body).await;
7279 let message = &resp["choices"][0]["message"];
7280 assert!(
7281 message["content"].is_string(),
7282 "must fall back to plain content when no real tool-call marker is present: {resp:?}"
7283 );
7284 assert!(message.get("tool_calls").is_none());
7285 assert_eq!(resp["choices"][0]["finish_reason"], "length");
7289 }
7290
7291 #[tokio::test]
7295 async fn a_cache_hit_reports_the_original_finish_reason_and_usage() {
7296 let app = test_app();
7297 let body = serde_json::json!({
7298 "model": "m",
7299 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
7300 "max_tokens": 3,
7301 "temperature": 0,
7302 });
7303 let first = post_json(&app, body.clone()).await;
7304 assert_eq!(first["ferrox_cache"], "miss");
7305 let second = post_json(&app, body).await;
7306 assert_eq!(second["ferrox_cache"], "hit");
7307 assert_eq!(
7308 first["choices"][0]["message"]["content"],
7309 second["choices"][0]["message"]["content"]
7310 );
7311 assert_eq!(
7312 first["choices"][0]["finish_reason"],
7313 second["choices"][0]["finish_reason"]
7314 );
7315 assert_eq!(first["usage"], second["usage"]);
7316 assert_eq!(second["usage"]["completion_tokens"], 3);
7317 }
7318
7319 #[tokio::test]
7330 async fn a_grammar_request_is_not_answered_from_an_unconstrained_cache_entry() {
7331 let app = test_app();
7332 let plain = serde_json::json!({
7333 "model": "m",
7334 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
7335 "max_tokens": 3,
7336 "temperature": 0,
7337 });
7338
7339 let first = post_json(&app, plain.clone()).await;
7340 assert_eq!(first["ferrox_cache"], "miss");
7341 let unconstrained = first["choices"][0]["message"]["content"]
7342 .as_str()
7343 .expect("content")
7344 .to_string();
7345
7346 let mut constrained = plain.clone();
7347 constrained["grammar"] = serde_json::json!("root ::= \"yes\"");
7348 let second = post_json(&app, constrained).await;
7349 assert_eq!(
7350 second["ferrox_cache"], "miss",
7351 "a grammar is part of the key, so this body has never been answered"
7352 );
7353 let constrained_answer = second["choices"][0]["message"]["content"]
7357 .as_str()
7358 .expect("content")
7359 .to_string();
7360 assert!(
7361 constrained_answer.contains("-> \"yes\"]"),
7362 "the grammar must have been compiled AND applied, not skipped \
7363 by a cache hit: {constrained_answer}"
7364 );
7365 assert_ne!(
7366 constrained_answer, unconstrained,
7367 "the constrained request was served the unconstrained answer"
7368 );
7369
7370 let third = post_json(&app, plain).await;
7374 assert_eq!(third["ferrox_cache"], "hit");
7375 assert_eq!(third["choices"][0]["message"]["content"], unconstrained);
7376 }
7377
7378 #[tokio::test]
7400 async fn a_json_object_request_does_not_reuse_the_unconstrained_cache_entry() {
7401 let state = Arc::new(test_state(
7402 test_model_full_byte_vocab(),
7403 ResponseCache::new(1000, Duration::from_secs(3600)),
7404 ));
7405 let app = test_app_with_state(state.clone());
7406 let plain = serde_json::json!({
7407 "model": "m",
7408 "messages": [
7409 {"role": "system", "content": "Answer in JSON when it helps."},
7410 {"role": "user", "content": "\u{1}\u{2}"},
7411 ],
7412 "max_tokens": 3,
7413 "temperature": 0,
7414 });
7415
7416 let first = post_json(&app, plain.clone()).await;
7417 assert_eq!(first["ferrox_cache"], "miss");
7418 assert_eq!(state.cache_stats().entries, 1);
7419
7420 let mut as_json = plain.clone();
7421 as_json["response_format"] = serde_json::json!({"type": "json_object"});
7422 let (status, _) = post_json_uri(&app, "/v1/chat/completions", as_json).await;
7423 assert_eq!(
7424 status,
7425 StatusCode::BAD_REQUEST,
7426 "the demo banner is not a JSON object, whoever generated it"
7427 );
7428 assert_eq!(
7429 state.cache_stats().hits,
7430 0,
7431 "a json_object request must not be answered from an entry the \
7432 JSON mask never produced"
7433 );
7434 assert_eq!(
7435 state.cache_stats().entries,
7436 2,
7437 "json_object must key its own entry, not reuse the unconstrained \
7438 one it happens to render the same prompt as"
7439 );
7440 }
7441
7442 #[tokio::test]
7454 async fn an_ignore_eos_request_is_not_answered_from_a_cache_entry_that_stopped_at_eos() {
7455 let app = test_app_with_state(Arc::new(test_state(
7456 test_model_full_byte_vocab_with_eos(Some(0x77)),
7457 ResponseCache::new(1000, Duration::from_secs(3600)),
7458 )));
7459 let body = serde_json::json!({
7460 "model": "m",
7461 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
7462 "max_tokens": 6,
7463 "temperature": 0,
7464 });
7465
7466 let stopped = post_json(&app, body.clone()).await;
7467 assert_eq!(stopped["ferrox_cache"], "miss");
7468 assert_eq!(
7469 stopped["choices"][0]["finish_reason"], "stop",
7470 "the fixture is only meaningful if the model's EOS really fires here"
7471 );
7472 assert_eq!(stopped["usage"]["completion_tokens"], 1);
7473
7474 let mut ignoring = body.clone();
7475 ignoring["ignore_eos"] = serde_json::json!(true);
7476 let ran_on = post_json(&app, ignoring).await;
7477 assert_eq!(
7478 ran_on["ferrox_cache"], "miss",
7479 "ignore_eos is part of the key, so this body has never been answered"
7480 );
7481 assert_eq!(
7482 ran_on["usage"]["completion_tokens"], 6,
7483 "ignore_eos must run the full budget, not replay the EOS-terminated answer"
7484 );
7485 assert_eq!(ran_on["choices"][0]["finish_reason"], "length");
7486 assert_ne!(
7487 ran_on["choices"][0]["message"]["content"],
7488 stopped["choices"][0]["message"]["content"]
7489 );
7490 }
7491
7492 #[tokio::test]
7498 async fn session_reuse_produces_the_same_output_as_manually_resending_full_history() {
7499 let session_app = test_app();
7500 let manual_app = test_app();
7501
7502 let turn1 = post_json(
7504 &session_app,
7505 serde_json::json!({
7506 "model": "m",
7507 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7508 "session_id": "s1",
7509 "max_tokens": 5,
7510 "temperature": 0,
7511 }),
7512 )
7513 .await;
7514 let reply1 = turn1["choices"][0]["message"]["content"]
7515 .as_str()
7516 .unwrap()
7517 .to_string();
7518
7519 let manual_turn1 = post_json(
7524 &manual_app,
7525 serde_json::json!({
7526 "model": "m",
7527 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7528 "max_tokens": 5,
7529 "temperature": 0,
7530 }),
7531 )
7532 .await;
7533 assert_eq!(
7534 manual_turn1["choices"][0]["message"]["content"]
7535 .as_str()
7536 .unwrap(),
7537 reply1
7538 );
7539
7540 let turn2 = post_json(
7542 &session_app,
7543 serde_json::json!({
7544 "model": "m",
7545 "messages": [{"role": "user", "content": "\u{4}\u{5}"}],
7546 "session_id": "s1",
7547 "max_tokens": 5,
7548 "temperature": 0,
7549 }),
7550 )
7551 .await;
7552 let reply2 = turn2["choices"][0]["message"]["content"]
7553 .as_str()
7554 .unwrap()
7555 .to_string();
7556
7557 let manual_turn2 = post_json(
7561 &manual_app,
7562 serde_json::json!({
7563 "model": "m",
7564 "messages": [
7565 {"role": "user", "content": "\u{1}\u{2}\u{3}"},
7566 {"role": "assistant", "content": reply1},
7567 {"role": "user", "content": "\u{4}\u{5}"},
7568 ],
7569 "max_tokens": 5,
7570 "temperature": 0,
7571 }),
7572 )
7573 .await;
7574 assert_eq!(
7575 manual_turn2["choices"][0]["message"]["content"]
7576 .as_str()
7577 .unwrap(),
7578 reply2,
7579 "resuming a session must produce identical output to manually resending the full history"
7580 );
7581 }
7582
7583 #[test]
7586 fn lock_cache_recovers_from_a_poisoned_mutex() {
7587 let cache = Arc::new(Mutex::new(ResponseCache::new(10, Duration::from_secs(60))));
7588
7589 let poison_cache = Arc::clone(&cache);
7590 let _ = std::thread::spawn(move || {
7591 let _guard = poison_cache.lock().unwrap();
7592 panic!("simulated panic while holding the lock");
7593 })
7594 .join();
7595
7596 let recovered = lock_cache(&cache);
7598 assert_eq!(recovered.stats().entries, 0);
7599 }
7600
7601 #[test]
7602 fn is_cacheable_true_for_greedy_or_seeded_requests() {
7603 let mut req_body = serde_json::json!({
7604 "model": "m",
7605 "messages": [{"role": "user", "content": "hi"}],
7606 });
7607 let req: ChatCompletionRequest = serde_json::from_value(req_body.clone()).unwrap();
7608 assert!(
7609 req.is_cacheable(),
7610 "default (temperature 0) must be cacheable"
7611 );
7612
7613 req_body["temperature"] = serde_json::json!(0.8);
7614 let req: ChatCompletionRequest = serde_json::from_value(req_body.clone()).unwrap();
7615 assert!(
7616 !req.is_cacheable(),
7617 "unseeded sampling must never be cacheable"
7618 );
7619
7620 req_body["seed"] = serde_json::json!(42);
7621 let req: ChatCompletionRequest = serde_json::from_value(req_body).unwrap();
7622 assert!(
7623 req.is_cacheable(),
7624 "sampling with an explicit seed is deterministic and must be cacheable"
7625 );
7626 }
7627
7628 const GRADED: &str = "{% if reasoning_effort %}\
7632 {% if reasoning_effort not in ['low','medium','high'] %}\
7633 {{ raise_exception('unsupported effort') }}\
7634 {% endif %}E:{{ reasoning_effort }}|{% endif %}\
7635 {% if enable_thinking %}THINK|{% endif %}{{ messages[0].content }}";
7636
7637 fn graded_template() -> chat_template::PromptTemplate {
7638 chat_template::PromptTemplate::from_gguf_metadata(
7639 Some(GRADED),
7640 Some("qwen3"),
7641 false,
7642 true,
7643 None,
7644 None,
7645 )
7646 }
7647
7648 fn chat_request(value: serde_json::Value) -> ChatCompletionRequest {
7649 serde_json::from_value(value).expect("request")
7650 }
7651
7652 #[test]
7659 fn a_grammar_on_the_chat_wire_reaches_the_generation_params() {
7660 let req = chat_request(serde_json::json!({
7661 "model": "m",
7662 "messages": [{"role": "user", "content": "hi"}],
7663 "grammar": "root ::= \"a\"+",
7664 }));
7665 req.validate_supported_fields()
7666 .expect("a valid grammar is a valid request");
7667 let params = req
7668 .generation_params(crate::sampling_knobs::SamplerModel::absent())
7669 .expect("a valid grammar compiles at params time too");
7670 assert!(
7671 params.grammar.is_some(),
7672 "the grammar was dropped between the wire and the sampler"
7673 );
7674 assert!(
7675 params.needs_vocab_logits(),
7676 "a grammar request that may fold lm_head into a GPU argmax is \
7677 a grammar request served unconstrained"
7678 );
7679
7680 let plain = chat_request(serde_json::json!({
7681 "model": "m",
7682 "messages": [{"role": "user", "content": "hi"}],
7683 }));
7684 assert!(plain
7685 .generation_params(crate::sampling_knobs::SamplerModel::absent())
7686 .unwrap()
7687 .grammar
7688 .is_none());
7689 }
7690
7691 fn tool_request(tool_choice: serde_json::Value) -> ChatCompletionRequest {
7692 chat_request(serde_json::json!({
7693 "model": "m",
7694 "messages": [{"role": "user", "content": "weather in Rome?"}],
7695 "tools": [weather_tool()],
7696 "tool_choice": tool_choice,
7697 }))
7698 }
7699
7700 #[test]
7704 fn a_forced_tool_choice_puts_a_grammar_on_the_generation_params() {
7705 for choice in [
7706 serde_json::json!("required"),
7707 serde_json::json!({"type": "function", "function": {"name": "get_weather"}}),
7708 ] {
7709 let req = tool_request(choice.clone());
7710 req.validate_supported_fields()
7711 .unwrap_or_else(|e| panic!("{choice} is a valid request: {e:?}"));
7712 let params = req
7713 .generation_params_for_template(
7714 &graded_template(),
7715 "Qwen3-8B",
7716 crate::sampling_knobs::SamplerModel::absent(),
7717 )
7718 .unwrap_or_else(|e| panic!("{choice} compiles: {e:?}"));
7719 let grammar = params
7720 .grammar
7721 .as_ref()
7722 .unwrap_or_else(|| panic!("{choice} was accepted and then not enforced"));
7723 assert!(
7724 grammar.is_awaiting_trigger(),
7725 "the model must be free to think before it calls"
7726 );
7727 assert!(
7728 !grammar.allows_eog(),
7729 "{choice} must not be able to end the turn without a call"
7730 );
7731 assert!(
7737 params.needs_vocab_logits(),
7738 "{choice} would let a backend return a token id instead of logits"
7739 );
7740 assert!(
7741 !generate::greedy_gpu_fold_allowed(¶ms),
7742 "{choice} at temperature 0 must still refuse the greedy GPU fold"
7743 );
7744 }
7745 }
7746
7747 #[test]
7749 fn an_unforced_tool_choice_leaves_the_generation_unconstrained() {
7750 for choice in [serde_json::json!("auto"), serde_json::json!("none")] {
7751 let req = tool_request(choice.clone());
7752 req.validate_supported_fields().expect("still supported");
7753 let params = match req.generation_params_for_template(
7754 &graded_template(),
7755 "Qwen3-8B",
7756 crate::sampling_knobs::SamplerModel::absent(),
7757 ) {
7758 Ok(p) => p,
7759 Err((status, _)) => panic!("{choice} has no constraint to compile: {status}"),
7760 };
7761 assert!(
7762 params.grammar.is_none(),
7763 "{choice} does not force a call and must not be constrained"
7764 );
7765 }
7766 }
7767
7768 #[test]
7771 fn a_forced_tool_choice_refuses_rather_than_quietly_not_forcing() {
7772 let req = chat_request(serde_json::json!({
7774 "model": "m",
7775 "messages": [{"role": "user", "content": "hi"}],
7776 "tool_choice": "required",
7777 }));
7778 let (status, _) = req
7779 .validate_supported_fields()
7780 .expect_err("nothing to call");
7781 assert_eq!(status, StatusCode::BAD_REQUEST);
7782
7783 let req =
7785 tool_request(serde_json::json!({"type": "function", "function": {"name": "nope"}}));
7786 let (status, Json(body)) = req.validate_supported_fields().expect_err("no such tool");
7787 assert_eq!(status, StatusCode::BAD_REQUEST);
7788 assert_eq!(body["error"]["param"], "tool_choice");
7789
7790 let req = tool_request(serde_json::json!({"type": "function"}));
7792 let (status, _) = req.validate_supported_fields().expect_err("names nothing");
7793 assert_eq!(status, StatusCode::BAD_REQUEST);
7794
7795 let req = chat_request(serde_json::json!({
7797 "model": "m",
7798 "messages": [{"role": "user", "content": "hi"}],
7799 "tools": [weather_tool()],
7800 "tool_choice": "required",
7801 "grammar": "root ::= \"a\"+",
7802 }));
7803 let (status, _) = req
7804 .validate_supported_fields()
7805 .expect_err("a grammar and a forced call are two constraints");
7806 assert_eq!(status, StatusCode::BAD_REQUEST);
7807
7808 let req = tool_request(serde_json::json!("required"));
7814 let (status, Json(body)) = match req.generation_params_for_template(
7815 &graded_template(),
7816 "muse-glimmer-8b",
7817 crate::sampling_knobs::SamplerModel::absent(),
7818 ) {
7819 Err(e) => e,
7820 Ok(_) => panic!("a muse_glimmer call's boundary is a channel, not a marker"),
7821 };
7822 assert_eq!(status, StatusCode::NOT_IMPLEMENTED);
7823 assert!(
7824 body["error"]["message"]
7825 .as_str()
7826 .unwrap()
7827 .contains("muse_glimmer"),
7828 "{body}"
7829 );
7830
7831 let req = tool_request(serde_json::json!("required"));
7837 let params = req
7838 .generation_params_for_template(
7839 &graded_template(),
7840 "gemma-4-E2B-it",
7841 crate::sampling_knobs::SamplerModel::absent(),
7842 )
7843 .expect("a gemma4 forced tool_choice is served");
7844 assert!(
7845 params.grammar.is_some(),
7846 "a forced tool_choice must arrive as the generation's grammar"
7847 );
7848 }
7849
7850 #[test]
7853 fn an_unparseable_grammar_on_the_chat_wire_is_a_400() {
7854 let req = chat_request(serde_json::json!({
7855 "model": "m",
7856 "messages": [{"role": "user", "content": "hi"}],
7857 "grammar": "root ::= \"a",
7858 }));
7859 let (status, Json(body)) = req
7860 .validate_supported_fields()
7861 .expect_err("this does not parse");
7862 assert_eq!(status, StatusCode::BAD_REQUEST);
7863 assert_eq!(body["error"]["param"], "grammar");
7864 assert!(
7865 req.generation_params(crate::sampling_knobs::SamplerModel::absent())
7866 .is_err(),
7867 "and again at params time"
7868 );
7869 }
7870
7871 #[test]
7878 fn response_format_json_schema_becomes_the_requests_grammar() {
7879 let req = chat_request(serde_json::json!({
7880 "model": "m",
7881 "messages": [{"role": "user", "content": "hi"}],
7882 "response_format": {
7883 "type": "json_schema",
7884 "json_schema": {"name": "x", "schema": {"type": "boolean"}},
7885 },
7886 }));
7887 req.validate_supported_fields()
7888 .expect("a boolean schema converts");
7889 let params = req
7890 .generation_params(crate::sampling_knobs::SamplerModel::absent())
7891 .expect("and compiles");
7892 let grammar = params.grammar.expect("the schema is the grammar");
7893 let mut g = (*grammar).clone();
7894 g.accept_token(0, b"true").expect("a boolean is accepted");
7895 assert!(g.allows_eog(), "and completes the parse");
7896 assert!(
7897 !params.json_object,
7898 "a schema is not the json_object character-class mask"
7899 );
7900 }
7901
7902 #[test]
7906 fn an_unconvertible_response_format_schema_is_a_400_naming_the_keyword() {
7907 let req = chat_request(serde_json::json!({
7908 "model": "m",
7909 "messages": [{"role": "user", "content": "hi"}],
7910 "response_format": {
7911 "type": "json_schema",
7912 "json_schema": {"name": "x", "schema": {"type": "integer", "minimum": 3}},
7913 },
7914 }));
7915 let (status, Json(body)) = req
7916 .validate_supported_fields()
7917 .expect_err("minimum has no grammar in this port");
7918 assert_eq!(status, StatusCode::BAD_REQUEST);
7919 assert!(
7920 body["error"]["message"]
7921 .as_str()
7922 .expect("a message")
7923 .contains("minimum"),
7924 "the refusal must name the keyword: {body}"
7925 );
7926 assert!(
7927 req.generation_params(crate::sampling_knobs::SamplerModel::absent())
7928 .is_err(),
7929 "and again at params time"
7930 );
7931 }
7932
7933 #[test]
7939 fn a_forced_tool_choice_and_a_schema_are_two_constraints() {
7940 let req = chat_request(serde_json::json!({
7941 "model": "m",
7942 "messages": [{"role": "user", "content": "hi"}],
7943 "tool_choice": "required",
7944 "tools": [{
7945 "type": "function",
7946 "function": {"name": "f", "parameters": {"type": "object"}},
7947 }],
7948 "response_format": {
7949 "type": "json_schema",
7950 "json_schema": {"name": "x", "schema": {"type": "boolean"}},
7951 },
7952 }));
7953 let (status, Json(body)) = req
7954 .validate_supported_fields()
7955 .expect_err("two constraints, one generation");
7956 assert_eq!(status, StatusCode::BAD_REQUEST);
7957 assert_eq!(body["error"]["param"], "tool_choice");
7958 }
7959
7960 #[test]
7963 fn an_omitted_output_budget_is_a_whole_answer_not_sixteen_tokens() {
7964 let req = chat_request(serde_json::json!({
7965 "model": "m",
7966 "messages": [{"role": "user", "content": "hi"}],
7967 }));
7968 assert_eq!(req.max_tokens, DEFAULT_CHAT_MAX_TOKENS);
7969 }
7970
7971 #[test]
7976 fn min_p_reaches_the_sampler_from_the_chat_wire() {
7977 let asked = chat_request(serde_json::json!({
7978 "model": "m",
7979 "messages": [{"role": "user", "content": "hi"}],
7980 "min_p": 0.07,
7981 }));
7982 assert_eq!(
7983 asked
7984 .sampling_params(crate::sampling_knobs::SamplerModel::absent())
7985 .expect("knobs")
7986 .min_p,
7987 0.07
7988 );
7989
7990 let silent = chat_request(serde_json::json!({
7991 "model": "m",
7992 "messages": [{"role": "user", "content": "hi"}],
7993 }));
7994 assert_eq!(
7995 silent
7996 .sampling_params(crate::sampling_knobs::SamplerModel::absent())
7997 .expect("knobs")
7998 .min_p,
7999 0.0,
8000 "an unset min_p must be off, not llama.cpp's CLI default"
8001 );
8002 }
8003
8004 #[test]
8013 fn no_sampler_knob_is_missing_from_the_cache_key() {
8014 let base = serde_json::json!({
8015 "model": "m",
8016 "messages": [{"role": "user", "content": "hi"}],
8017 "seed": 1,
8018 });
8019 let key_for = |body: serde_json::Value| {
8020 let req = chat_request(body);
8021 let params = req
8022 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8023 .expect("params");
8024 req.cache_key("prompt", ¶ms)
8025 };
8026 let baseline = key_for(base.clone());
8027 for (knob, value) in [
8028 ("temperature", serde_json::json!(0.5)),
8029 ("top_p", serde_json::json!(0.9)),
8030 ("min_p", serde_json::json!(0.05)),
8031 ("top_k", serde_json::json!(40)),
8032 ("repetition_penalty", serde_json::json!(1.1)),
8033 ("presence_penalty", serde_json::json!(0.3)),
8034 ("frequency_penalty", serde_json::json!(0.3)),
8035 (
8036 "samplers",
8037 serde_json::json!(["penalties", "top_p", "top_k", "min_p", "temperature"]),
8038 ),
8039 ] {
8040 let mut body = base.clone();
8041 body[knob] = value;
8042 assert_ne!(
8043 key_for(body),
8044 baseline,
8045 "`{knob}` is not in the cache key: two requests differing \
8046 only in it would share one cached answer"
8047 );
8048 }
8049 }
8050
8051 #[test]
8061 fn no_constraint_is_missing_from_the_cache_key() {
8062 let base = serde_json::json!({
8063 "model": "m",
8064 "messages": [{"role": "user", "content": "pick one"}],
8065 });
8066 let key_for = |body: serde_json::Value| {
8067 let req = chat_request(body);
8068 let params = req
8069 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8070 .expect("params");
8071 req.cache_key("prompt", ¶ms)
8072 };
8073 let baseline = key_for(base.clone());
8074 for (field, value) in [
8075 ("grammar", serde_json::json!("root ::= \"yes\" | \"no\"")),
8076 (
8077 "response_format",
8078 serde_json::json!({"type": "json_object"}),
8079 ),
8080 (
8081 "response_format",
8082 serde_json::json!({"type": "json_schema", "json_schema": {
8083 "name": "answer",
8084 "schema": {"type": "object", "properties": {"a": {"type": "string"}}}
8085 }}),
8086 ),
8087 ("ignore_eos", serde_json::json!(true)),
8088 ("stop", serde_json::json!(["\n"])),
8089 ("max_tokens", serde_json::json!(7)),
8090 ] {
8091 let mut body = base.clone();
8092 body[field] = value.clone();
8093 assert_ne!(
8094 key_for(body),
8095 baseline,
8096 "`{field}: {value}` is not in the cache key: two requests \
8097 differing only in it would share one cached answer"
8098 );
8099 }
8100 }
8101
8102 #[test]
8106 fn an_explicit_zero_output_budget_is_a_client_error() {
8107 let req = chat_request(serde_json::json!({
8108 "model": "m",
8109 "messages": [{"role": "user", "content": "hi"}],
8110 "max_tokens": 0,
8111 }));
8112 let (status, body) = req.validate_supported_fields().expect_err("rejected");
8113 assert_eq!(status, StatusCode::BAD_REQUEST);
8114 assert_eq!(body["error"]["param"], serde_json::json!("max_tokens"));
8115 }
8116
8117 #[test]
8120 fn a_request_can_turn_thinking_off() {
8121 let template = graded_template();
8122 for body in [
8123 serde_json::json!({
8124 "model": "m",
8125 "messages": [{"role": "user", "content": "hi"}],
8126 "reasoning_effort": "none",
8127 }),
8128 serde_json::json!({
8129 "model": "m",
8130 "messages": [{"role": "user", "content": "hi"}],
8131 "thinking": {"type": "disabled"},
8132 }),
8133 ] {
8134 let kwargs = chat_request(body).resolve_template_kwargs(&template);
8135 assert_eq!(kwargs["enable_thinking"], serde_json::json!(false));
8136 assert_eq!(kwargs["thinking_mode"], serde_json::json!("disabled"));
8137 assert!(!kwargs.contains_key("reasoning_effort"));
8140 }
8141 }
8142
8143 #[test]
8146 fn a_disabled_switch_beats_an_effort_in_the_same_request() {
8147 let template = graded_template();
8148 let kwargs = chat_request(serde_json::json!({
8149 "model": "m",
8150 "messages": [{"role": "user", "content": "hi"}],
8151 "reasoning_effort": "high",
8152 "thinking": {"type": "disabled"},
8153 }))
8154 .resolve_template_kwargs(&template);
8155 assert_eq!(kwargs["enable_thinking"], serde_json::json!(false));
8156 assert!(!kwargs.contains_key("reasoning_effort"));
8157 }
8158
8159 #[test]
8162 fn an_unrecognized_thinking_switch_is_refused_rather_than_read_as_on() {
8163 let req = chat_request(serde_json::json!({
8164 "model": "m",
8165 "messages": [{"role": "user", "content": "hi"}],
8166 "thinking": {"type": "disable"},
8167 }));
8168 let (status, _) = req.validate_supported_fields().expect_err("rejected");
8169 assert_eq!(status, StatusCode::BAD_REQUEST);
8170 }
8171
8172 #[test]
8175 fn an_explicit_template_kwarg_stands_the_protocol_knobs_down() {
8176 let template = graded_template();
8177 let kwargs = chat_request(serde_json::json!({
8178 "model": "m",
8179 "messages": [{"role": "user", "content": "hi"}],
8180 "reasoning_effort": "none",
8181 "chat_template_kwargs": {"enable_thinking": true},
8182 }))
8183 .resolve_template_kwargs(&template);
8184 assert_eq!(kwargs["enable_thinking"], serde_json::json!(true));
8185 }
8186
8187 #[test]
8191 fn an_off_vocabulary_reasoning_effort_is_quantized_rather_than_interpolated() {
8192 let template = graded_template();
8193 let req = chat_request(serde_json::json!({
8194 "model": "m",
8195 "messages": [{"role": "user", "content": "hi"}],
8196 "reasoning_effort": "minimal",
8197 }));
8198 let kwargs = req.resolve_template_kwargs(&template);
8199 assert_eq!(kwargs["reasoning_effort"], serde_json::json!("low"));
8200 let prompt = prompt_from_messages(&req.messages, &template, &[], kwargs).expect("renders");
8201 assert!(prompt.starts_with("E:low|"), "{prompt}");
8202 }
8203
8204 #[test]
8208 fn an_effort_with_no_near_gear_is_dropped_so_the_template_default_applies() {
8209 let template = graded_template();
8210 let req = chat_request(serde_json::json!({
8211 "model": "m",
8212 "messages": [{"role": "user", "content": "hi"}],
8213 "chat_template_kwargs": {"reasoning_effort": "none"},
8214 }));
8215 let kwargs = req.resolve_template_kwargs(&template);
8216 assert!(!kwargs.contains_key("reasoning_effort"));
8217 let prompt = prompt_from_messages(&req.messages, &template, &[], kwargs).expect("renders");
8218 assert_eq!(prompt, "hi");
8219 }
8220
8221 #[test]
8224 fn chat_template_kwargs_wins_over_the_top_level_reasoning_effort() {
8225 let template = graded_template();
8226 let req = chat_request(serde_json::json!({
8227 "model": "m",
8228 "messages": [{"role": "user", "content": "hi"}],
8229 "reasoning_effort": "low",
8230 "chat_template_kwargs": {"reasoning_effort": "high"},
8231 }));
8232 assert_eq!(
8233 req.resolve_template_kwargs(&template)["reasoning_effort"],
8234 serde_json::json!("high")
8235 );
8236 }
8237
8238 #[test]
8242 fn offering_tools_turns_thinking_on_by_itself() {
8243 let template = graded_template();
8244 let quiet = chat_request(serde_json::json!({
8245 "model": "m",
8246 "messages": [{"role": "user", "content": "hi"}],
8247 }));
8248 assert!(!quiet
8249 .resolve_template_kwargs(&template)
8250 .contains_key("enable_thinking"));
8251
8252 let with_tools = chat_request(serde_json::json!({
8253 "model": "m",
8254 "messages": [{"role": "user", "content": "hi"}],
8255 "tools": [{"type": "function", "function": {"name": "get_weather"}}],
8256 }));
8257 let kwargs = with_tools.resolve_template_kwargs(&template);
8258 assert_eq!(kwargs["enable_thinking"], serde_json::json!(true));
8259 let prompt =
8260 prompt_from_messages(&with_tools.messages, &template, &[], kwargs).expect("renders");
8261 assert!(prompt.starts_with("THINK|"), "{prompt}");
8262 }
8263
8264 #[test]
8269 fn a_prompt_that_opens_the_reasoning_block_makes_the_first_token_reasoning() {
8270 let opener = chat_template::PromptTemplate::from_gguf_metadata(
8271 Some("{{ messages[0].content }}{% if enable_thinking %}<think>{% endif %}"),
8272 Some("qwen3"),
8273 false,
8274 true,
8275 None,
8276 None,
8277 );
8278 let req = chat_request(serde_json::json!({
8279 "model": "m",
8280 "messages": [{"role": "user", "content": "hi"}],
8281 "chat_template_kwargs": {"enable_thinking": true},
8282 }));
8283 let kwargs = req.resolve_template_kwargs(&opener);
8284 let prompt = prompt_from_messages(&req.messages, &opener, &[], kwargs).expect("renders");
8285 assert!(prompt.ends_with("<think>"), "{prompt}");
8286
8287 let posture = output::OutputPosture::resolve("Qwen3-8B", &prompt);
8290 let (message, _) = build_response_message(
8291 "weighing it up</think>Paris.".to_string(),
8292 &[],
8293 posture,
8294 "stop",
8295 );
8296 assert_eq!(message.reasoning_content.as_deref(), Some("weighing it up"));
8297 assert_eq!(message.content.as_deref(), Some("Paris."));
8298
8299 let closed = output::OutputPosture::resolve("Qwen3-8B", "<|im_start|>assistant\n");
8302 let (message, _) = build_response_message(
8303 "weighing it up</think>Paris.".to_string(),
8304 &[],
8305 closed,
8306 "stop",
8307 );
8308 assert_eq!(message.reasoning_content, None);
8309 }
8310
8311 #[test]
8312 fn stop_param_accepts_both_single_string_and_array() {
8313 let req: ChatCompletionRequest = serde_json::from_value(serde_json::json!({
8314 "model": "m",
8315 "messages": [{"role": "user", "content": "hi"}],
8316 "stop": "END",
8317 }))
8318 .unwrap();
8319 assert_eq!(req.stop_sequences(), vec!["END".to_string()]);
8320
8321 let req: ChatCompletionRequest = serde_json::from_value(serde_json::json!({
8322 "model": "m",
8323 "messages": [{"role": "user", "content": "hi"}],
8324 "stop": ["A", "B"],
8325 }))
8326 .unwrap();
8327 assert_eq!(req.stop_sequences(), vec!["A".to_string(), "B".to_string()]);
8328 }
8329
8330 #[test]
8331 fn run_generation_rejects_out_of_vocab_tokens_instead_of_panicking() {
8332 let model = test_model();
8333 let result = run_generation(
8334 &model,
8335 "hello",
8336 &greedy_params(4),
8337 None,
8338 None,
8339 None,
8340 None,
8341 None,
8342 None,
8343 );
8344 assert!(matches!(
8345 result,
8346 Err(generate::DecodeError::TokenOutOfVocab { .. })
8347 ));
8348 }
8349
8350 #[test]
8354 fn run_generation_honors_an_exhausted_kv_pool_and_maps_it_to_a_503() {
8355 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
8357 let pool = Arc::new(Mutex::new(ferrox_core::cache::KvBlockPool::new(64, 2)));
8358
8359 let holder_pool = Arc::clone(&pool);
8360 let holder = std::thread::spawn(move || {
8361 let mut held = ferrox_core::cache::KvCache::with_pool(1, 1, holder_pool, 0).unwrap();
8362 held.push(&[0.0], &[0.0]).unwrap(); std::thread::sleep(Duration::from_millis(200));
8364 drop(held);
8365 });
8366 std::thread::sleep(Duration::from_millis(15));
8367
8368 let config = generate::KvPoolConfig {
8369 pool,
8370 queue_wait: Duration::ZERO,
8371 };
8372 let result = run_generation(
8373 &model,
8374 &prompt,
8375 &greedy_params(4),
8376 Some(&config),
8377 None,
8378 None,
8379 None,
8380 None,
8381 None,
8382 );
8383 assert!(matches!(
8384 result,
8385 Err(generate::DecodeError::KvPoolExhausted)
8386 ));
8387
8388 let (status, _body) = decode_error_response(result.unwrap_err());
8389 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
8390 holder.join().unwrap();
8391 }
8392
8393 #[test]
8401 fn a_request_too_big_for_the_whole_pool_is_a_400_not_a_retryable_503() {
8402 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
8404 let pool = Arc::new(Mutex::new(ferrox_core::cache::KvBlockPool::new(64, 1)));
8406 let config = generate::KvPoolConfig {
8407 pool,
8408 queue_wait: Duration::ZERO,
8409 };
8410
8411 let result = run_generation(
8412 &model,
8413 &prompt,
8414 &greedy_params(4),
8415 Some(&config),
8416 None,
8417 None,
8418 None,
8419 None,
8420 None,
8421 );
8422 let err = result.expect_err("one block cannot hold two layers' caches");
8423 assert!(
8424 matches!(
8425 &err,
8426 generate::DecodeError::KvBudgetExceeded { binding, .. }
8427 if *binding == ferrox_models::Ceiling::DeviceMemory.code()
8428 ),
8429 "expected an immovable device-memory refusal, got {err:?}"
8430 );
8431 let (status, _body) = decode_error_response(err);
8432 assert_eq!(status, StatusCode::BAD_REQUEST);
8433 }
8434
8435 #[test]
8441 fn decode_error_response_maps_a_full_queue_to_a_retryable_503() {
8442 let (status, Json(body)) = decode_error_response(generate::DecodeError::QueueFull {
8443 queued: 512,
8444 cap: 512,
8445 });
8446 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
8447 assert_eq!(body["error"]["retry_after_seconds"], 1);
8448 let message = body["error"]["message"].as_str().expect("message");
8449 assert!(message.contains("512"), "{message}");
8450 }
8451
8452 #[test]
8453 fn decode_error_response_omits_a_retry_hint_for_an_unretryable_error() {
8454 let (_status, Json(body)) = decode_error_response(generate::DecodeError::TokenOutOfVocab {
8455 token: 99,
8456 vocab_size: 32,
8457 });
8458 assert!(
8459 body["error"]["retry_after_seconds"].is_null(),
8460 "retrying a prompt this model cannot tokenize never helps"
8461 );
8462 }
8463
8464 #[test]
8465 fn decode_error_response_maps_token_out_of_vocab_to_bad_request() {
8466 let (status, _body) = decode_error_response(generate::DecodeError::TokenOutOfVocab {
8467 token: 99,
8468 vocab_size: 32,
8469 });
8470 assert_eq!(status, StatusCode::BAD_REQUEST);
8471 }
8472
8473 #[test]
8474 fn run_generation_succeeds_and_releases_blocks_when_the_pool_has_room() {
8475 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
8477 let pool = Arc::new(Mutex::new(ferrox_core::cache::KvBlockPool::new(64, 2)));
8478 let config = generate::KvPoolConfig {
8479 pool: pool.clone(),
8480 queue_wait: Duration::ZERO,
8481 };
8482
8483 let (_, finish, _usage) = run_generation(
8484 &model,
8485 &prompt,
8486 &greedy_params(4),
8487 Some(&config),
8488 None,
8489 None,
8490 None,
8491 None,
8492 None,
8493 )
8494 .unwrap();
8495 assert_eq!(finish, FinishReason::Length);
8496 assert_eq!(
8497 pool.lock().unwrap().free_blocks(),
8498 2,
8499 "a completed request must return its blocks to the pool"
8500 );
8501 }
8502
8503 #[tokio::test]
8508 async fn concurrent_requests_against_the_same_model_do_not_interfere() {
8509 let model = Arc::new(test_model());
8510 let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
8511
8512 let mut handles = Vec::new();
8513 for _ in 0..8 {
8514 let model = Arc::clone(&model);
8515 let prompt = prompt.clone();
8516 handles.push(tokio::task::spawn_blocking(move || {
8517 run_generation(
8518 &model,
8519 &prompt,
8520 &greedy_params(6),
8521 None,
8522 None,
8523 None,
8524 None,
8525 None,
8526 None,
8527 )
8528 .unwrap()
8529 }));
8530 }
8531
8532 let mut results = Vec::new();
8533 for h in handles {
8534 results.push(h.await.unwrap());
8535 }
8536 for r in &results[1..] {
8541 assert_eq!(r.0, results[0].0, "decoded chunks must match");
8542 assert_eq!(r.1, results[0].1, "finish reason must match");
8543 assert_eq!(
8544 r.2.prompt_tokens, results[0].2.prompt_tokens,
8545 "prompt token count must match"
8546 );
8547 assert_eq!(
8548 r.2.completion_tokens, results[0].2.completion_tokens,
8549 "completion token count must match"
8550 );
8551 }
8552 }
8553
8554 fn write_safetensors_shard(tensors: &[(String, Vec<usize>, Vec<f32>)]) -> Vec<u8> {
8561 let mut header_entries = Vec::new();
8562 let mut data = Vec::new();
8563 for (name, shape, values) in tensors {
8564 let start = data.len();
8565 for v in values {
8566 data.extend_from_slice(&v.to_le_bytes());
8567 }
8568 let end = data.len();
8569 let shape_str = shape
8570 .iter()
8571 .map(|d| d.to_string())
8572 .collect::<Vec<_>>()
8573 .join(",");
8574 header_entries.push(format!(
8575 "\"{name}\":{{\"dtype\":\"F32\",\"shape\":[{shape_str}],\"data_offsets\":[{start},{end}]}}"
8576 ));
8577 }
8578 let header = format!("{{{}}}", header_entries.join(","));
8579 let header_bytes = header.as_bytes();
8580 let mut out = Vec::with_capacity(8 + header_bytes.len() + data.len());
8581 out.extend_from_slice(&(header_bytes.len() as u64).to_le_bytes());
8582 out.extend_from_slice(header_bytes);
8583 out.extend_from_slice(&data);
8584 out
8585 }
8586
8587 fn build_synthetic_kimi_loaded() -> model::KimiLoaded {
8598 use ferrox_models::config::{AttentionKind, KdaConfig, KimiHybridAttention, MlaConfig};
8599 use ferrox_models::kimi_loader::KimiRealHparams;
8600 use ferrox_moe::{GatingFunction, MoeLayerConfig};
8601
8602 let hidden_dim = 8;
8603 let kda_num_heads = 2;
8604 let kda_head_dim = 3;
8605 let kda_proj = kda_num_heads * kda_head_dim;
8606 let conv_kernel = 4;
8607 let dense_intermediate = 5;
8608 let vocab_size = 256;
8612 let mla_num_heads = 1;
8613 let mla_q_lora_rank = 2;
8614 let mla_kv_lora_rank = 2;
8615 let mla_qk_nope_head_dim = 2;
8616 let mla_qk_rope_head_dim = 2;
8617 let mla_v_head_dim = 2;
8618
8619 let model_cfg = ferrox_models::ModelConfig {
8620 rope_layers: ferrox_models::rope_layers::RopeLayers::All,
8621 layer_shapes: ferrox_models::layer_shapes::LayerShapes::Uniform,
8622 name: "synthetic-kimi-server-test",
8623 n_layers: 1,
8624 n_mtp_blocks: 0,
8625 hidden_dim,
8626 n_heads: 1,
8627 n_kv_heads: 1,
8628 head_dim: 4,
8629 vocab_size,
8630 rope_theta: 10000.0,
8631 rms_norm_eps: 1e-5,
8632 sliding_window: None,
8633 moe: MoeLayerConfig {
8634 expert_weights_scale: 1.0,
8635 n_experts: 1,
8636 n_experts_active: 1,
8637 n_shared_experts: 0,
8638 hidden_dim,
8639 expert_ffn_dim: 4,
8640 gating: GatingFunction::Sigmoid,
8641 norm_topk_prob: true,
8642 expert_group_count: None,
8643 expert_group_used_count: None,
8644 },
8645 n_dense_leading_layers: 1,
8650 attention: AttentionKind::KimiHybrid(KimiHybridAttention {
8651 kda_layers: vec![1],
8652 full_attn_layers: vec![],
8653 mla: MlaConfig {
8654 num_heads: mla_num_heads,
8655 q_lora_rank: mla_q_lora_rank,
8656 kv_lora_rank: mla_kv_lora_rank,
8657 qk_nope_head_dim: mla_qk_nope_head_dim,
8658 qk_rope_head_dim: mla_qk_rope_head_dim,
8659 v_head_dim: mla_v_head_dim,
8660 use_output_gate: true,
8661 rope: None,
8662 },
8663 kda: KdaConfig {
8664 num_heads: kda_num_heads,
8665 head_dim: kda_head_dim,
8666 short_conv_kernel_size: conv_kernel,
8667 gate_lower_bound: -5.0,
8668 use_full_rank_gate: true,
8669 },
8670 }),
8671 rope_freqs: None,
8672 rope_attn_factor: 1.0,
8673 rope_dim: None,
8674 rope_dim_swa: None,
8675 rope_freqs_long: None,
8676 rope_freqs_short: None,
8677 rope_orig_ctx: None,
8678 rope_layout: ferrox_models::config::RopeLayout::Neox,
8679 qk_norm_style: ferrox_models::capability::QkNormStyle::WholeVector,
8680 swa_layers: ferrox_models::swa_layers::SwaLayers::All,
8681 attn_logit_softcap: None,
8682 final_logit_softcap: None,
8683 embedding_scale: None,
8684 residual_scale: None,
8685 clamp_kqv: None,
8686 logit_multiplier: None,
8687 attention_scale: None,
8688 rope_theta_swa: None,
8689 ffn_activation: ferrox_models::config::FfnActivation::Swiglu,
8690 best_effort_fields: &["synthetic test config, not a real preset"],
8691 };
8692 let hp = KimiRealHparams {
8693 hidden_dim,
8694 kda_num_heads,
8695 kda_head_dim,
8696 mla_num_heads,
8697 mla_q_lora_rank,
8698 mla_kv_lora_rank,
8699 mla_qk_nope_head_dim,
8700 mla_qk_rope_head_dim,
8701 mla_v_head_dim,
8702 dense_intermediate_dim: dense_intermediate,
8703 moe_hidden_dim: hidden_dim,
8704 moe_intermediate_dim: 4,
8705 n_experts: 1,
8706 num_shared_experts: 0,
8707 };
8708
8709 let prefix = "language_model.model.layers.0";
8713 let mut tensors: Vec<(String, Vec<usize>, Vec<f32>)> = Vec::new();
8714 let push = |tensors: &mut Vec<(String, Vec<usize>, Vec<f32>)>,
8715 name: String,
8716 shape: Vec<usize>,
8717 n: usize| {
8718 tensors.push((name, shape, vec![0.01f32; n]));
8719 };
8720 push(
8721 &mut tensors,
8722 format!("{prefix}.input_layernorm.weight"),
8723 vec![hidden_dim],
8724 hidden_dim,
8725 );
8726 push(
8727 &mut tensors,
8728 format!("{prefix}.post_attention_layernorm.weight"),
8729 vec![hidden_dim],
8730 hidden_dim,
8731 );
8732 push(
8733 &mut tensors,
8734 format!("{prefix}.self_attention_res_norm.weight"),
8735 vec![hidden_dim],
8736 hidden_dim,
8737 );
8738 push(
8739 &mut tensors,
8740 format!("{prefix}.self_attention_res_proj.weight"),
8741 vec![1, hidden_dim],
8742 hidden_dim,
8743 );
8744 push(
8745 &mut tensors,
8746 format!("{prefix}.mlp_res_norm.weight"),
8747 vec![hidden_dim],
8748 hidden_dim,
8749 );
8750 push(
8751 &mut tensors,
8752 format!("{prefix}.mlp_res_proj.weight"),
8753 vec![1, hidden_dim],
8754 hidden_dim,
8755 );
8756 push(
8757 &mut tensors,
8758 format!("{prefix}.self_attn.q_proj.weight"),
8759 vec![kda_proj, hidden_dim],
8760 kda_proj * hidden_dim,
8761 );
8762 push(
8763 &mut tensors,
8764 format!("{prefix}.self_attn.k_proj.weight"),
8765 vec![kda_proj, hidden_dim],
8766 kda_proj * hidden_dim,
8767 );
8768 push(
8769 &mut tensors,
8770 format!("{prefix}.self_attn.v_proj.weight"),
8771 vec![kda_proj, hidden_dim],
8772 kda_proj * hidden_dim,
8773 );
8774 push(
8775 &mut tensors,
8776 format!("{prefix}.self_attn.q_conv1d.weight"),
8777 vec![kda_proj, 1, conv_kernel],
8778 kda_proj * conv_kernel,
8779 );
8780 push(
8781 &mut tensors,
8782 format!("{prefix}.self_attn.k_conv1d.weight"),
8783 vec![kda_proj, 1, conv_kernel],
8784 kda_proj * conv_kernel,
8785 );
8786 push(
8787 &mut tensors,
8788 format!("{prefix}.self_attn.v_conv1d.weight"),
8789 vec![kda_proj, 1, conv_kernel],
8790 kda_proj * conv_kernel,
8791 );
8792 push(
8793 &mut tensors,
8794 format!("{prefix}.self_attn.A_log"),
8795 vec![kda_num_heads],
8796 kda_num_heads,
8797 );
8798 push(
8799 &mut tensors,
8800 format!("{prefix}.self_attn.f_a_proj.weight"),
8801 vec![kda_head_dim, hidden_dim],
8802 kda_head_dim * hidden_dim,
8803 );
8804 push(
8805 &mut tensors,
8806 format!("{prefix}.self_attn.f_b_proj.weight"),
8807 vec![kda_proj, kda_head_dim],
8808 kda_proj * kda_head_dim,
8809 );
8810 push(
8811 &mut tensors,
8812 format!("{prefix}.self_attn.dt_bias"),
8813 vec![kda_proj],
8814 kda_proj,
8815 );
8816 push(
8817 &mut tensors,
8818 format!("{prefix}.self_attn.b_proj.weight"),
8819 vec![kda_num_heads, hidden_dim],
8820 kda_num_heads * hidden_dim,
8821 );
8822 push(
8823 &mut tensors,
8824 format!("{prefix}.self_attn.g_proj.weight"),
8825 vec![kda_proj, hidden_dim],
8826 kda_proj * hidden_dim,
8827 );
8828 push(
8829 &mut tensors,
8830 format!("{prefix}.self_attn.o_norm.weight"),
8831 vec![kda_head_dim],
8832 kda_head_dim,
8833 );
8834 push(
8835 &mut tensors,
8836 format!("{prefix}.self_attn.o_proj.weight"),
8837 vec![hidden_dim, kda_proj],
8838 hidden_dim * kda_proj,
8839 );
8840 push(
8841 &mut tensors,
8842 format!("{prefix}.mlp.gate_proj.weight"),
8843 vec![dense_intermediate, hidden_dim],
8844 dense_intermediate * hidden_dim,
8845 );
8846 push(
8847 &mut tensors,
8848 format!("{prefix}.mlp.up_proj.weight"),
8849 vec![dense_intermediate, hidden_dim],
8850 dense_intermediate * hidden_dim,
8851 );
8852 push(
8853 &mut tensors,
8854 format!("{prefix}.mlp.down_proj.weight"),
8855 vec![hidden_dim, dense_intermediate],
8856 hidden_dim * dense_intermediate,
8857 );
8858 push(
8859 &mut tensors,
8860 "language_model.model.embed_tokens.weight".to_string(),
8861 vec![vocab_size, hidden_dim],
8862 vocab_size * hidden_dim,
8863 );
8864 push(
8865 &mut tensors,
8866 "language_model.lm_head.weight".to_string(),
8867 vec![vocab_size, hidden_dim],
8868 vocab_size * hidden_dim,
8869 );
8870 push(
8871 &mut tensors,
8872 "language_model.model.norm.weight".to_string(),
8873 vec![hidden_dim],
8874 hidden_dim,
8875 );
8876 push(
8877 &mut tensors,
8878 "language_model.model.output_attn_res_norm.weight".to_string(),
8879 vec![hidden_dim],
8880 hidden_dim,
8881 );
8882 push(
8883 &mut tensors,
8884 "language_model.model.output_attn_res_proj.weight".to_string(),
8885 vec![1, hidden_dim],
8886 hidden_dim,
8887 );
8888
8889 static FIXTURE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
8902 let dir = std::env::temp_dir().join(format!(
8903 "ferrox_server_kimi_e2e_test_{}_{}",
8904 std::process::id(),
8905 FIXTURE.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
8906 ));
8907 std::fs::create_dir_all(&dir).unwrap();
8908 let shard_bytes = write_safetensors_shard(&tensors);
8909 std::fs::write(dir.join("shard0.safetensors"), &shard_bytes).unwrap();
8910 let map_entries: Vec<String> = tensors
8911 .iter()
8912 .map(|(name, ..)| format!("\"{name}\":\"shard0.safetensors\""))
8913 .collect();
8914 let index = format!("{{\"weight_map\":{{{}}}}}", map_entries.join(","));
8915 std::fs::write(dir.join("model.safetensors.index.json"), &index).unwrap();
8916
8917 use base64::Engine;
8921 let vocab_lines: Vec<String> = (0..vocab_size as u32)
8922 .map(|b| {
8923 let b64 = base64::engine::general_purpose::STANDARD.encode([b as u8]);
8924 format!("{b64} {b}")
8925 })
8926 .collect();
8927 std::fs::write(dir.join("tiktoken.model"), vocab_lines.join("\n")).unwrap();
8928
8929 let loaded = model::load_kimi_checkpoint_with_config(dir.to_str().unwrap(), model_cfg, hp)
8930 .expect("must load the synthetic Kimi checkpoint end to end");
8931 std::fs::remove_dir_all(&dir).ok();
8932 loaded
8933 }
8934
8935 #[test]
8943 fn kimi_model_serves_real_text_end_to_end_via_run_generation() {
8944 let loaded = build_synthetic_kimi_loaded();
8945 let state = build_app_state(
8946 StartupModels {
8947 loaded: model::LoadedModel::Kimi(loaded),
8948 embedding: None,
8949 },
8950 None,
8951 None,
8952 None,
8953 false,
8954 None,
8955 Arc::new(health::Detection::ready(health::probe_backends())),
8956 );
8957 let active = state.active().expect("a freshly built state has a model");
8958 assert_eq!(active.tokenizer_kind(), "kimi-tiktoken-bpe");
8959 assert!(!active.is_synthetic());
8960
8961 let (_chunks, finish, _usage) = run_generation(
8962 active.generative().unwrap(),
8963 "hi",
8964 &greedy_params(5),
8965 None,
8966 None,
8967 None,
8968 None,
8969 None,
8970 None,
8971 )
8972 .expect("a real Kimi checkpoint must generate without error");
8973 assert!(matches!(finish, FinishReason::Length | FinishReason::Stop));
8974 }
8975
8976 #[test]
8989 fn a_grammar_constrains_the_engine_decode_path() {
8990 let loaded = build_synthetic_kimi_loaded();
8991 let state = build_app_state(
8992 StartupModels {
8993 loaded: model::LoadedModel::Kimi(loaded),
8994 embedding: None,
8995 },
8996 None,
8997 None,
8998 None,
8999 false,
9000 None,
9001 Arc::new(health::Detection::ready(health::probe_backends())),
9002 );
9003 let active = state.active().expect("a freshly built state has a model");
9004
9005 let run = |grammar: Option<&str>| {
9006 let mut params = greedy_params(6);
9007 params.grammar = grammar.map(|src| {
9008 Arc::new(
9009 ferrox_models::grammar::Grammar::from_str_with_root(src, "root")
9010 .expect("test grammar parses"),
9011 )
9012 });
9013 run_generation(
9014 active.generative().unwrap(),
9015 "hi",
9016 ¶ms,
9017 None,
9018 None,
9019 None,
9020 None,
9021 None,
9022 None,
9023 )
9024 };
9025
9026 let (chunks, _, _) = run(None).expect("the unconstrained run must serve");
9027 let unconstrained = chunks.concat();
9028 assert!(
9029 unconstrained.chars().any(|c| c != 'a'),
9030 "the unconstrained run produced only `a` ({unconstrained:?}), so the \
9031 constrained run below would prove nothing"
9032 );
9033
9034 let (chunks, finish, _) =
9035 run(Some(r#"root ::= "a"+"#)).expect("a grammar this vocabulary can spell must serve");
9036 let constrained = chunks.concat();
9037 assert!(
9038 !constrained.is_empty() && constrained.chars().all(|c| c == 'a'),
9039 "the engine decode path served text its grammar forbids ({constrained:?}): \
9040 the constraint was dropped between `generate_engine` and the sampler"
9041 );
9042 assert!(matches!(finish, FinishReason::Length | FinishReason::Stop));
9043 }
9044
9045 #[test]
9055 fn kv_pool_and_prefix_cache_are_never_consulted_for_a_kimi_model() {
9056 let loaded = build_synthetic_kimi_loaded();
9057 let state = build_app_state(
9058 StartupModels {
9059 loaded: model::LoadedModel::Kimi(loaded),
9060 embedding: None,
9061 },
9062 None,
9063 None,
9064 None,
9065 false,
9066 None,
9067 Arc::new(health::Detection::ready(health::probe_backends())),
9068 );
9069
9070 let pool = Arc::new(Mutex::new(ferrox_core::cache::KvBlockPool::new(64, 4)));
9071 let kv_pool_config = generate::KvPoolConfig {
9072 pool: pool.clone(),
9073 queue_wait: Duration::ZERO,
9074 };
9075 let pc = Mutex::new(PrefixCache::new(4));
9076
9077 run_generation(
9078 state
9079 .active()
9080 .expect("a freshly built state has a model")
9081 .generative()
9082 .unwrap(),
9083 "hi",
9084 &greedy_params(5),
9085 Some(&kv_pool_config),
9086 None,
9087 Some(&pc),
9088 None,
9089 None,
9090 None,
9091 )
9092 .expect("a real Kimi checkpoint must generate without error");
9093
9094 assert_eq!(
9095 pool.lock().unwrap().free_blocks(),
9096 4,
9097 "the KV pool must be completely untouched by a Kimi request"
9098 );
9099 let stats = pc.lock().unwrap().stats();
9100 assert_eq!(
9101 stats.hits + stats.misses,
9102 0,
9103 "the prefix cache must never be consulted for a Kimi request"
9104 );
9105 }
9106}