1mod admin;
36mod anthropic;
37mod attribution;
38mod best_of;
39mod budget;
40mod cache_admin;
41mod cache_salt;
42mod cancel;
43mod chat_params;
44mod chat_stream_choice;
45mod chat_template;
46mod choice_stream;
47mod cli;
48mod completion;
49mod continuation;
50mod conversations;
51mod decode_task;
52mod embeddings;
53mod generate;
54mod grammar_request;
55mod health;
56mod journal;
57mod json_mode;
58mod limits;
59mod loaded;
60mod logprobs;
61mod lora;
62mod mcp;
63mod model;
64mod openai_extra;
65mod output;
66mod policy;
67mod prefill_batch;
68mod reasoning_budget;
69mod reasoning_tokens;
70mod request_tail;
71mod rerank;
72mod response_cache;
73pub(crate) mod responses;
74mod resume;
75mod round_robin;
76mod sample_step;
77mod sampling_knobs;
78mod sampling_loop;
79mod security;
80mod serving;
81mod session;
82mod slots;
83mod sse;
84mod stats;
85mod stop;
86mod stream_events;
87mod tasks;
88mod tool_grammar;
89mod unimplemented_fields;
90mod unsupported_sampling;
91mod utf8_stream;
92
93use std::cell::RefCell;
94use std::convert::Infallible;
95use std::net::SocketAddr;
96use std::path::PathBuf;
97use std::rc::Rc;
98use std::sync::{Arc, Mutex, MutexGuard};
99use std::time::Duration;
100
101use axum::{
102 extract::State,
103 http::StatusCode,
104 response::sse::{Event, Sse},
105 response::{IntoResponse, Response},
106 routing::{get, post},
107 Json, Router,
108};
109use serde::{Deserialize, Serialize};
110
111use cli::apply_cli_overrides;
112pub use cli::{ServerArgs, BUILT_WITH_CUDA, BUILT_WITH_METAL};
113
114use frink_core::cache::KvBlockPool;
115use frink_models::kimi_tokenizer::KimiTokenizer;
116use frink_models::sampling::SamplingParams;
117use frink_models::tokenizer::{SpecialTokens, StopTokens};
118use frink_models::{Decoder, Gemma4Engine, KimiEngine, MlaEngine, PrefixCache};
119#[cfg(test)]
120use generate::FinishReason;
121use generate::GenerationParams;
122pub(crate) use loaded::{ActiveModel, Loaded, SleptModel};
123use model::ServerTokenizer;
124use rerank::encoder_endpoints;
125use response_cache::ResponseCache;
126use sampling_knobs::SamplingKnobs;
127
128#[allow(clippy::large_enum_variant)] pub(crate) enum Model {
137 Gguf(GgufModel),
138 Kimi(KimiModel),
139 Mla(MlaModel),
140 Gemma4(Gemma4Model),
141 Glm52(Glm52Model),
142}
143
144pub(crate) struct GgufModel {
145 decoder: Arc<Decoder>,
146 tokenizer: Arc<ServerTokenizer>,
147 stop_tokens: StopTokens,
148 bos_id: Option<usize>,
149 is_synthetic: bool,
150 chat_template: chat_template::PromptTemplate,
151}
152
153pub(crate) struct KimiModel {
154 engine: KimiEngine,
155 tokenizer: KimiTokenizer,
156 stop_tokens: StopTokens,
157 chat_template: chat_template::PromptTemplate,
158}
159
160pub(crate) struct MlaModel {
161 engine: MlaEngine,
162 tokenizer: ServerTokenizer,
163 stop_tokens: StopTokens,
164 bos_id: Option<usize>,
165 name: String,
166 chat_template: chat_template::PromptTemplate,
167}
168
169pub(crate) struct Gemma4Model {
170 engine: Gemma4Engine,
171 tokenizer: ServerTokenizer,
172 stop_tokens: StopTokens,
173 bos_id: Option<usize>,
174 name: String,
175 chat_template: chat_template::PromptTemplate,
176}
177
178pub(crate) struct Glm52Model {
179 engine: frink_models::Glm52Engine,
180 tokenizer: ServerTokenizer,
181 stop_tokens: StopTokens,
182 bos_id: Option<usize>,
183 name: String,
184 chat_template: chat_template::PromptTemplate,
185}
186
187impl Model {
188 pub(crate) fn chat_template(&self) -> chat_template::PromptTemplate {
189 match self {
190 Model::Gguf(m) => m.chat_template.clone(),
191 Model::Kimi(m) => m.chat_template.clone(),
192 Model::Mla(m) => m.chat_template.clone(),
193 Model::Gemma4(m) => m.chat_template.clone(),
194 Model::Glm52(m) => m.chat_template.clone(),
195 }
196 }
197
198 fn is_synthetic(&self) -> bool {
203 match self {
204 Model::Gguf(m) => m.is_synthetic,
205 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => false,
206 }
207 }
208
209 fn tokenizer_kind(&self) -> &'static str {
210 match self {
211 Model::Gguf(m) => m.tokenizer.kind(),
212 Model::Kimi(_) => "kimi-tiktoken-bpe",
213 Model::Mla(m) => m.tokenizer.kind(),
214 Model::Gemma4(m) => m.tokenizer.kind(),
215 Model::Glm52(m) => m.tokenizer.kind(),
216 }
217 }
218
219 fn expert_store_stats(&self) -> Option<frink_core::expert_store::ExpertStoreStats> {
223 match self {
224 Model::Gguf(m) => m.decoder.expert_store_stats(),
225 Model::Kimi(m) => m.engine.weights.expert_store_stats(),
226 Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
227 }
228 }
229
230 pub(crate) fn name(&self) -> &str {
231 match self {
232 Model::Gguf(m) => m.decoder.config.name,
233 Model::Kimi(_) => "kimi-k3",
234 Model::Mla(m) => m.name.as_str(),
235 Model::Gemma4(m) => m.name.as_str(),
236 Model::Glm52(m) => m.name.as_str(),
237 }
238 }
239
240 pub(crate) fn encode(&self, text: &str, specials: SpecialTokens) -> Vec<usize> {
267 match self {
268 Model::Gguf(m) => m.tokenizer.encode(text, specials),
269 Model::Kimi(m) => m
270 .tokenizer
271 .encode(text, specials)
272 .into_iter()
273 .map(|id| id as usize)
274 .collect(),
275 Model::Mla(m) => m.tokenizer.encode(text, specials),
276 Model::Gemma4(m) => m.tokenizer.encode(text, specials),
277 Model::Glm52(m) => m.tokenizer.encode(text, specials),
278 }
279 }
280
281 pub(crate) fn bos_id(&self) -> Option<usize> {
290 match self {
291 Model::Gguf(m) => m.bos_id,
292 Model::Kimi(_) => None,
293 Model::Mla(m) => m.bos_id,
294 Model::Gemma4(m) => m.bos_id,
295 Model::Glm52(m) => m.bos_id,
296 }
297 }
298
299 pub(crate) fn decode(&self, ids: &[usize]) -> String {
300 match self {
301 Model::Gguf(m) => m.tokenizer.decode(ids),
302 Model::Kimi(m) => {
303 let ids32: Vec<u32> = ids.iter().map(|&id| id as u32).collect();
304 m.tokenizer.decode(&ids32)
305 }
306 Model::Mla(m) => m.tokenizer.decode(ids),
307 Model::Gemma4(m) => m.tokenizer.decode(ids),
308 Model::Glm52(m) => m.tokenizer.decode(ids),
309 }
310 }
311
312 pub(crate) fn embed_tokens(&self, tokens: &[usize]) -> Option<Vec<Vec<f32>>> {
315 match self {
316 Model::Gguf(m) => {
317 let mut caches: Vec<_> = m.decoder.config.new_kv_caches();
318 Some(m.decoder.forward_hidden_batch(tokens, 0, &mut caches))
319 }
320 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
321 }
322 }
323
324 pub(crate) fn gguf_decoder(&self) -> Option<&Arc<Decoder>> {
330 match self {
331 Model::Gguf(m) => Some(&m.decoder),
332 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
333 }
334 }
335
336 pub(crate) fn vocab_size(&self) -> Option<usize> {
337 match self {
338 Model::Gguf(m) => Some(m.decoder.config.vocab_size),
339 Model::Kimi(m) => Some(m.tokenizer.vocab_size()),
340 Model::Mla(m) => Some(frink_models::Engine::vocab_size(&m.engine)),
341 Model::Gemma4(m) => Some(frink_models::Engine::vocab_size(&m.engine)),
342 Model::Glm52(m) => Some(frink_models::Engine::vocab_size(&m.engine)),
343 }
344 }
345
346 fn has_real_vocabulary(&self) -> bool {
353 match self {
354 Model::Gguf(m) => !matches!(*m.tokenizer, model::ServerTokenizer::Byte),
355 Model::Kimi(_) => true,
356 Model::Mla(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
357 Model::Gemma4(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
358 Model::Glm52(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
359 }
360 }
361}
362
363impl frink_models::dry::DryVocab for Model {
369 fn n_tokens(&self) -> usize {
370 self.vocab_size().unwrap_or(0)
371 }
372
373 fn detokenize(&self, token: usize) -> String {
374 self.decode(&[token])
375 }
376
377 fn tokenize(&self, text: &str) -> Vec<usize> {
378 self.encode(text, SpecialTokens::AsText)
379 }
380}
381
382pub(crate) struct AppState {
383 pub(crate) embedding: Option<Arc<frink_models::EmbeddingModel>>,
397 active: std::sync::RwLock<Option<Arc<ActiveModel>>>,
416 pub(crate) load_in_progress: std::sync::atomic::AtomicBool,
420 pub(crate) slept: Mutex<Option<SleptModel>>,
430 pub(crate) tasks: Arc<tasks::TaskRegistry>,
432 pub(crate) cancels: Arc<cancel::CancelRegistry>,
436 pub(crate) stats: stats::Stats,
439 pub(crate) streams: resume::StreamRegistry,
442 pub(crate) model_dir: Option<PathBuf>,
444 response_cache: Mutex<ResponseCache>,
448 pub(crate) kv_pool: Option<generate::KvPoolConfig>,
460 pub(crate) paged_kv: Option<generate::PagedKvConfig>,
473 pub(crate) prefix_cache: Option<Arc<Mutex<PrefixCache>>>,
483 sessions: session::SessionStore,
489 requests_total: std::sync::atomic::AtomicU64,
490 request_errors_total: std::sync::atomic::AtomicU64,
491 started_at: std::time::Instant,
492 last_request_ms: std::sync::atomic::AtomicU64,
497 detection: Arc<health::Detection>,
499 mcp: Option<mcp::LoadedMcpConfig>,
501 pub(crate) continuous_batching_enabled: bool,
505 pub(crate) metal_private_decode_gate: Option<Arc<std::sync::Mutex<()>>>,
509 loading_model: Mutex<Option<String>>,
513 last_load_error: Mutex<Option<(String, String)>>,
517 pub(crate) serving: Mutex<crate::stats::ServingStats>,
522 pub(crate) maintenance: Mutex<crate::policy::maintenance::MaintenanceGate>,
527 pub(crate) footprint:
533 Mutex<crate::policy::footprint::ProbeCache<crate::policy::footprint::Footprint>>,
534 pub(crate) started_unix: u64,
544}
545
546pub(crate) const FOOTPRINT_TTL_MS: u64 = 2_000;
552
553impl AppState {
554 pub(crate) fn active(&self) -> Option<Arc<ActiveModel>> {
558 self.active
559 .read()
560 .unwrap_or_else(|p| p.into_inner())
561 .clone()
562 }
563
564 pub(crate) fn is_sleeping(&self) -> bool {
571 self.slept
572 .lock()
573 .unwrap_or_else(|p| p.into_inner())
574 .is_some()
575 }
576
577 pub(crate) fn require_active(&self) -> Result<Arc<ActiveModel>, ApiError> {
578 if let Some(active) = self.active() {
579 return Ok(active);
580 }
581 if self.is_sleeping() {
585 return Err((
586 StatusCode::SERVICE_UNAVAILABLE,
587 Json(serde_json::json!({"error": {
588 "message": "this server is asleep; POST /wake_up to reload the model it put \
589 away",
590 "type": "server_sleeping"
591 }})),
592 ));
593 }
594 Err((
595 StatusCode::SERVICE_UNAVAILABLE,
596 Json(serde_json::json!({"error": {
597 "message": "no model is loaded; POST /admin/models/load with an id from \
598 GET /admin/models",
599 "type": "model_not_loaded"
600 }})),
601 ))
602 }
603
604 pub(crate) fn require_model(&self) -> Result<Arc<Model>, ApiError> {
612 Ok(Arc::clone(self.require_active()?.generative()?))
613 }
614
615 pub(crate) fn swap_active(&self, next: Option<Arc<ActiveModel>>) -> Option<Arc<ActiveModel>> {
623 let mut guard = self.active.write().unwrap_or_else(|p| p.into_inner());
624 std::mem::replace(&mut *guard, next)
625 }
626
627 fn mark_request_finished(&self) {
631 let ms = self.started_at.elapsed().as_millis().min(u64::MAX as u128) as u64;
632 self.last_request_ms
633 .store(ms, std::sync::atomic::Ordering::Relaxed);
634 }
635
636 pub(crate) fn uptime(&self) -> Duration {
637 self.started_at.elapsed()
638 }
639
640 pub(crate) fn requests_total(&self) -> u64 {
641 self.requests_total
642 .load(std::sync::atomic::Ordering::Relaxed)
643 }
644
645 pub(crate) fn errors_total(&self) -> u64 {
646 self.request_errors_total
647 .load(std::sync::atomic::Ordering::Relaxed)
648 }
649
650 pub(crate) fn cache_stats(&self) -> response_cache::CacheStats {
651 lock_cache(&self.response_cache).stats()
652 }
653
654 pub(crate) fn last_request_age_seconds(&self) -> Option<f64> {
657 let last = self
658 .last_request_ms
659 .load(std::sync::atomic::Ordering::Relaxed);
660 (last > 0)
661 .then(|| self.uptime().as_secs_f64() - (last as f64 / 1000.0))
662 .map(|age| age.max(0.0))
663 }
664
665 pub(crate) fn loading_model_id(&self) -> Option<String> {
666 self.loading_model
667 .lock()
668 .unwrap_or_else(|p| p.into_inner())
669 .clone()
670 }
671
672 pub(crate) fn set_loading_model(&self, id: Option<String>) {
673 *self.loading_model.lock().unwrap_or_else(|p| p.into_inner()) = id;
674 }
675
676 pub(crate) fn last_load_error(&self) -> Option<(String, String)> {
677 self.last_load_error
678 .lock()
679 .unwrap_or_else(|p| p.into_inner())
680 .clone()
681 }
682
683 pub(crate) fn set_last_load_error(&self, error: Option<(String, String)>) {
684 *self
685 .last_load_error
686 .lock()
687 .unwrap_or_else(|p| p.into_inner()) = error;
688 }
689
690 pub(crate) fn active_model_name(&self) -> Option<String> {
699 self.active().map(|a| a.name().to_string())
700 }
701
702 pub(crate) fn embedding_model(&self) -> Option<Arc<frink_models::EmbeddingModel>> {
712 self.embedding
713 .clone()
714 .or_else(|| self.active().and_then(|a| a.encoder().map(Arc::clone)))
715 }
716
717 pub(crate) fn embedding_model_name(&self) -> Option<String> {
721 match self.embedding_model() {
722 Some(e) => Some(e.name().to_string()),
723 None => self.active_model_name(),
724 }
725 }
726
727 pub(crate) fn record_request(&self, record: stats::Record<'_>) {
728 self.stats.record(stats::entry(record));
729 }
730}
731
732fn lock_cache(cache: &Mutex<ResponseCache>) -> MutexGuard<'_, ResponseCache> {
737 cache
738 .lock()
739 .unwrap_or_else(|poisoned| poisoned.into_inner())
740}
741
742#[derive(Debug, Clone, Deserialize)]
743#[serde(untagged)]
744pub(crate) enum MessageContent {
745 Text(String),
746 Parts(Vec<ContentPart>),
747}
748
749#[derive(Debug, Clone, Deserialize)]
750struct ContentPart {
751 #[serde(rename = "type")]
752 kind: String,
753 #[serde(default)]
754 text: Option<String>,
755 #[serde(default)]
756 image_url: Option<serde_json::Value>,
757}
758
759impl MessageContent {
760 fn as_text(&self) -> String {
761 match self {
762 Self::Text(s) => s.clone(),
763 Self::Parts(parts) => parts
764 .iter()
765 .filter_map(|p| p.text.as_deref())
766 .collect::<Vec<_>>()
767 .join(""),
768 }
769 }
770
771 fn has_image(&self) -> bool {
772 match self {
773 Self::Text(_) => false,
774 Self::Parts(parts) => parts
775 .iter()
776 .any(|p| p.kind == "image_url" || p.image_url.is_some()),
777 }
778 }
779}
780
781#[derive(Debug, Clone, Deserialize)]
782pub(crate) struct ChatMessage {
783 pub(crate) role: String,
784 #[serde(default)]
788 pub(crate) content: Option<MessageContent>,
789 #[serde(default)]
793 pub(crate) tool_calls: Option<Vec<ToolCallIn>>,
794 #[serde(default)]
799 #[allow(dead_code)]
800 pub(crate) tool_call_id: Option<String>,
801 #[serde(default, alias = "reasoning")]
817 pub(crate) reasoning_content: Option<String>,
818}
819
820impl ChatMessage {
821 fn rendered_content(&self) -> String {
828 let mut out = self
829 .content
830 .as_ref()
831 .map(MessageContent::as_text)
832 .unwrap_or_default();
833 if let Some(calls) = &self.tool_calls {
834 for call in calls {
835 out.push_str(&format!(
836 "<tool_call>{{\"name\": \"{}\", \"arguments\": {}}}</tool_call>",
837 call.function.name, call.function.arguments
838 ));
839 }
840 }
841 out
842 }
843}
844
845#[derive(Debug, Clone, Deserialize)]
846pub(crate) struct ToolCallIn {
847 #[serde(default)]
848 #[allow(dead_code)]
849 id: String,
850 #[serde(rename = "type", default)]
851 #[allow(dead_code)]
852 kind: String,
853 function: ToolCallFunctionIn,
854}
855
856#[derive(Debug, Clone, Deserialize)]
857struct ToolCallFunctionIn {
858 name: String,
859 arguments: String,
864}
865
866#[derive(Debug, Clone, Deserialize)]
869struct ToolDef {
870 #[serde(rename = "type", default)]
871 #[allow(dead_code)]
872 kind: String,
873 function: ToolFunctionDef,
874}
875
876#[derive(Debug, Clone, Deserialize)]
877struct ToolFunctionDef {
878 name: String,
879 #[serde(default)]
880 description: Option<String>,
881 #[serde(default)]
882 parameters: Option<serde_json::Value>,
883}
884
885#[derive(Debug, Clone, Deserialize)]
896#[serde(untagged)]
897enum ToolChoice {
898 Mode(String),
899 Specific(serde_json::Value),
900}
901
902#[derive(Deserialize)]
905#[serde(untagged)]
906enum StopParam {
907 One(String),
908 Many(Vec<String>),
909}
910
911#[derive(Deserialize)]
912struct ChatCompletionRequest {
913 model: String,
914 messages: Vec<ChatMessage>,
915 #[serde(default = "default_max_tokens")]
916 max_tokens: usize,
917 #[serde(default)]
918 temperature: Option<f32>,
919 #[serde(default)]
920 top_p: Option<f32>,
921 #[serde(default)]
926 min_p: Option<f32>,
927 #[serde(default)]
928 top_k: Option<usize>,
929 #[serde(default)]
930 repetition_penalty: Option<f32>,
931 #[serde(flatten)]
935 extra_samplers: crate::sampling_knobs::ExtraSamplerFields,
936 #[serde(flatten)]
940 unimplemented: crate::unimplemented_fields::UnimplementedFields,
941 #[serde(default)]
942 seed: Option<u64>,
943 #[serde(default)]
944 stop: Option<StopParam>,
945 #[serde(default)]
946 stream: Option<bool>,
947 #[serde(default)]
960 stream_resumable: Option<bool>,
961 #[serde(default)]
972 ignore_eos: Option<bool>,
973 #[serde(default)]
974 tools: Vec<ToolDef>,
975 #[serde(default)]
976 tool_choice: Option<ToolChoice>,
977 #[serde(default)]
985 chat_template_kwargs: Option<serde_json::Map<String, serde_json::Value>>,
986 #[serde(default)]
994 reasoning_effort: Option<String>,
995 #[serde(default)]
999 thinking: Option<ThinkingSwitch>,
1000 #[serde(default)]
1005 session_id: Option<String>,
1006 #[serde(default, deserialize_with = "continuation::deserialize")]
1014 continue_final_message: continuation::ContinueFinalMessage,
1015 #[serde(default, alias = "thinking_budget_tokens")]
1023 reasoning_budget_tokens: Option<reasoning_budget::BudgetTokens>,
1024 #[serde(default)]
1026 logprobs: Option<bool>,
1027 #[serde(default)]
1028 top_logprobs: Option<u32>,
1029 #[serde(default)]
1030 presence_penalty: Option<f32>,
1031 #[serde(default)]
1032 frequency_penalty: Option<f32>,
1033 #[serde(default)]
1034 response_format: Option<serde_json::Value>,
1035 #[serde(default)]
1042 logit_bias: Option<serde_json::Value>,
1043 #[serde(default)]
1047 lora: Option<Vec<frink_api::LoraScaleRequest>>,
1048 #[serde(default)]
1058 samplers: Option<serde_json::Value>,
1059 #[serde(default)]
1069 grammar: Option<String>,
1070}
1071
1072const DEFAULT_CHAT_MAX_TOKENS: usize = 32_768;
1084
1085#[derive(Debug, Clone, Deserialize)]
1087pub(crate) struct ThinkingSwitch {
1088 #[serde(rename = "type")]
1089 pub(crate) kind: String,
1090}
1091
1092const THINKING_KWARG_KEYS: [&str; 4] = [
1096 "enable_thinking",
1097 "thinking",
1098 "thinking_mode",
1099 "reasoning_effort",
1100];
1101
1102const DISABLE_EFFORTS: [&str; 2] = ["none", "off"];
1106
1107fn default_max_tokens() -> usize {
1108 DEFAULT_CHAT_MAX_TOKENS
1109}
1110
1111impl ChatCompletionRequest {
1112 fn sampling_knobs(&self) -> Result<SamplingKnobs, ApiError> {
1121 let mut knobs = SamplingKnobs {
1122 temperature: self.temperature,
1123 top_p: self.top_p,
1124 min_p: self.min_p,
1125 top_k: self.top_k,
1126 repetition_penalty: self.repetition_penalty,
1127 presence_penalty: self.presence_penalty,
1128 frequency_penalty: self.frequency_penalty,
1129 penalty_last_n: None,
1133 sampler_order: unsupported_sampling::parse_sampler_order(
1134 self.samplers.as_ref(),
1135 "/v1/chat/completions",
1136 )?,
1137 ..SamplingKnobs::default()
1138 };
1139 self.extra_samplers.apply(&mut knobs);
1140 Ok(knobs)
1141 }
1142
1143 fn sampling_params(
1144 &self,
1145 model: crate::sampling_knobs::SamplerModel<'_>,
1146 ) -> Result<SamplingParams, ApiError> {
1147 self.sampling_knobs()?.resolve(model).map_err(|e| {
1148 unsupported_feature(&format!("`dry_multiplier` on /v1/chat/completions: {e}"))
1149 })
1150 }
1151
1152 fn stop_sequences(&self) -> Vec<String> {
1153 self.stop
1154 .as_ref()
1155 .map(|s| match s {
1156 StopParam::One(v) => vec![v.clone()],
1157 StopParam::Many(v) => v.clone(),
1158 })
1159 .unwrap_or_default()
1160 }
1161
1162 fn n_logprobs(&self) -> Result<Option<usize>, ApiError> {
1176 const MAX: u32 = 20;
1177 match (self.logprobs, self.top_logprobs) {
1178 (Some(true), Some(n)) if n > MAX => Err(invalid_request(
1179 &format!(
1180 "`top_logprobs` is {n}; this server reports at most {MAX} alternatives per \
1181 position, as upstream does"
1182 ),
1183 "top_logprobs",
1184 )),
1185 (Some(true), Some(n)) => Ok(Some(n as usize)),
1186 (Some(true), None) => Ok(Some(0)),
1190 (_, Some(_)) => Err(invalid_request(
1191 "`top_logprobs` requires `logprobs: true`",
1192 "top_logprobs",
1193 )),
1194 _ => Ok(None),
1195 }
1196 }
1197
1198 fn tools_active(&self) -> bool {
1199 !self.tools.is_empty()
1200 && !matches!(&self.tool_choice, Some(ToolChoice::Mode(m)) if m == "none")
1201 }
1202
1203 fn forced_tool_choice(&self) -> Result<Option<tool_grammar::Forced<'_>>, ApiError> {
1215 match &self.tool_choice {
1216 Some(ToolChoice::Mode(m)) if m == "required" => Ok(Some(tool_grammar::Forced::Any)),
1217 Some(ToolChoice::Specific(value)) => {
1218 let name = value
1222 .get("function")
1223 .and_then(|f| f.get("name"))
1224 .or_else(|| value.get("name"))
1225 .and_then(|n| n.as_str());
1226 match name {
1227 Some(name) => Ok(Some(tool_grammar::Forced::Named(name))),
1228 None => Err(invalid_request(
1229 "tool_choice must be \"auto\", \"none\", \"required\", or an object with \
1230 function.name",
1231 "tool_choice",
1232 )),
1233 }
1234 }
1235 _ => Ok(None),
1236 }
1237 }
1238
1239 fn tool_specs(&self) -> Vec<tool_grammar::ToolSpec<'_>> {
1241 self.tools
1242 .iter()
1243 .map(|t| tool_grammar::ToolSpec {
1244 name: &t.function.name,
1245 parameters: t.function.parameters.as_ref(),
1246 })
1247 .collect()
1248 }
1249
1250 fn thinking_direction(&self) -> Option<bool> {
1293 if let Some(switch) = &self.thinking {
1294 return match switch.kind.trim().to_ascii_lowercase().as_str() {
1295 "disabled" => Some(false),
1296 "enabled" => Some(true),
1297 _ => None,
1300 };
1301 }
1302 let effort = self.reasoning_effort.as_ref()?;
1303 DISABLE_EFFORTS
1304 .contains(&effort.trim().to_ascii_lowercase().as_str())
1305 .then_some(false)
1306 }
1307
1308 fn resolve_template_kwargs(
1309 &self,
1310 template: &chat_template::PromptTemplate,
1311 ) -> serde_json::Map<String, serde_json::Value> {
1312 let mut kwargs = self.chat_template_kwargs.clone().unwrap_or_default();
1313 let caller_steered = THINKING_KWARG_KEYS.iter().any(|k| kwargs.contains_key(*k));
1317
1318 if !caller_steered {
1319 match self.thinking_direction() {
1320 Some(false) => {
1321 for (k, v) in crate::policy::effort::thinking_off_kwargs() {
1322 kwargs.insert(k, v);
1323 }
1324 return kwargs;
1327 }
1328 Some(true) => {
1329 for (k, v) in crate::policy::effort::thinking_on_kwargs() {
1330 kwargs.insert(k, v);
1331 }
1332 }
1333 None => {}
1334 }
1335 if let Some(effort) = &self.reasoning_effort {
1336 kwargs
1337 .entry("reasoning_effort".to_string())
1338 .or_insert_with(|| serde_json::json!(effort));
1339 }
1340 }
1341
1342 let offered: Vec<serde_json::Value> = if self.tools_active() {
1343 self.tools.iter().map(chat_template::tool_json).collect()
1344 } else {
1345 Vec::new()
1346 };
1347 let thinking = crate::policy::effort::resolve_thinking_mode(Some(&kwargs), Some(&offered));
1348 if thinking == crate::policy::effort::ThinkingMode::Thinking {
1349 for (k, v) in crate::policy::effort::thinking_on_kwargs() {
1350 kwargs.entry(k).or_insert(v);
1351 }
1352 }
1353 match crate::policy::effort::sanitize_effort(&mut kwargs, template.efforts()) {
1354 crate::policy::effort::EffortMapping::Mapped(to) => {
1355 tracing::debug!("reasoning_effort quantized to {}", to.as_str());
1356 }
1357 crate::policy::effort::EffortMapping::Dropped => {
1358 tracing::debug!(
1359 "reasoning_effort dropped: this checkpoint's template grades no gear close \
1360 enough, so its own default applies"
1361 );
1362 }
1363 crate::policy::effort::EffortMapping::Unchanged => {}
1364 }
1365 crate::policy::effort::broadcast_effort_spellings(&mut kwargs);
1366 kwargs
1367 }
1368
1369 fn validate_supported_fields(&self) -> Result<(), ApiError> {
1372 if self.max_tokens == 0 {
1379 return Err(invalid_request(
1380 "max_tokens must be at least 1",
1381 "max_tokens",
1382 ));
1383 }
1384 if let Some(switch) = &self.thinking {
1389 let kind = switch.kind.trim().to_ascii_lowercase();
1390 if kind != "enabled" && kind != "disabled" {
1391 return Err(invalid_request(
1392 "thinking.type must be \"enabled\" or \"disabled\"",
1393 "thinking.type",
1394 ));
1395 }
1396 }
1397 for msg in &self.messages {
1398 if msg.content.as_ref().is_some_and(MessageContent::has_image) {
1399 return Err(unsupported_feature(
1400 "image_url content parts are not implemented (multimodal/VL deferred, see docs/API.md)",
1401 ));
1402 }
1403 }
1404 self.n_logprobs()?;
1408 self.unimplemented.refuse("/v1/chat/completions")?;
1412 unsupported_sampling::refuse_logit_bias(self.logit_bias.as_ref(), "/v1/chat/completions")?;
1413 unsupported_sampling::parse_sampler_order(self.samplers.as_ref(), "/v1/chat/completions")?;
1417 let stated_grammar =
1432 grammar_request::for_request(self.grammar.as_deref(), self.response_format.as_ref())?;
1433 if let Some(forced) = self.forced_tool_choice()? {
1441 if self.tools.is_empty() {
1442 return Err(invalid_request(
1443 "tool_choice forces a tool call, but no tools were offered",
1444 "tool_choice",
1445 ));
1446 }
1447 if let tool_grammar::Forced::Named(name) = forced {
1448 if !self.tools.iter().any(|t| t.function.name == name) {
1449 return Err(invalid_request(
1450 &format!(
1451 "tool_choice names {name:?}, which is not one of the tools offered"
1452 ),
1453 "tool_choice",
1454 ));
1455 }
1456 }
1457 if stated_grammar.is_some() {
1467 return Err(invalid_request(
1468 "a forced tool_choice and a \"grammar\" or response_format \"json_schema\" \
1469 are two different constraints on the same generation; send one",
1470 "tool_choice",
1471 ));
1472 }
1473 if self.json_object_mode() {
1474 return Err(invalid_request(
1475 "a forced tool_choice cannot be combined with response_format json_object: \
1476 the tool-call markers are not JSON",
1477 "tool_choice",
1478 ));
1479 }
1480 }
1481 Ok(())
1482 }
1483
1484 fn effective_stop_sequences(&self) -> Vec<String> {
1491 let mut stop = self.stop_sequences();
1492 if self.tools_active() {
1493 stop.push("</tool_call>".to_string());
1494 }
1495 stop
1496 }
1497
1498 fn json_object_mode(&self) -> bool {
1499 self.response_format
1500 .as_ref()
1501 .and_then(|v| v.get("type"))
1502 .and_then(|v| v.as_str())
1503 == Some("json_object")
1504 }
1505}
1506
1507#[derive(Serialize)]
1508struct ChatCompletionChoice {
1509 index: usize,
1510 message: ChatCompletionResponseMessage,
1511 finish_reason: &'static str,
1512 #[serde(skip_serializing_if = "Option::is_none")]
1517 logprobs: Option<serde_json::Value>,
1518}
1519
1520#[derive(Serialize)]
1521struct ChatCompletionResponseMessage {
1522 role: &'static str,
1523 #[serde(skip_serializing_if = "Option::is_none")]
1524 content: Option<String>,
1525 #[serde(skip_serializing_if = "Option::is_none")]
1529 reasoning_content: Option<String>,
1530 #[serde(skip_serializing_if = "Option::is_none")]
1531 tool_calls: Option<Vec<ToolCallOut>>,
1532}
1533
1534#[derive(Serialize, Clone)]
1535struct ToolCallOut {
1536 id: String,
1537 #[serde(rename = "type")]
1538 kind: &'static str,
1539 function: ToolCallFunctionOut,
1540}
1541
1542#[derive(Serialize, Clone)]
1550struct ToolCallDelta {
1551 index: usize,
1552 #[serde(skip_serializing_if = "Option::is_none")]
1553 id: Option<String>,
1554 #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
1555 kind: Option<&'static str>,
1556 function: ToolCallFunctionDelta,
1557}
1558
1559#[derive(Serialize, Clone, Default)]
1560struct ToolCallFunctionDelta {
1561 #[serde(skip_serializing_if = "Option::is_none")]
1562 name: Option<String>,
1563 #[serde(skip_serializing_if = "Option::is_none")]
1566 arguments: Option<String>,
1567}
1568
1569impl ToolCallDelta {
1570 fn whole(index: usize, name: String, arguments: String) -> Self {
1572 ToolCallDelta {
1573 index,
1574 id: Some(format!("call_{index}")),
1575 kind: Some("function"),
1576 function: ToolCallFunctionDelta {
1577 name: Some(name),
1578 arguments: Some(arguments),
1579 },
1580 }
1581 }
1582
1583 fn opening(index: usize, name: String) -> Self {
1585 ToolCallDelta {
1586 index,
1587 id: Some(format!("call_{index}")),
1588 kind: Some("function"),
1589 function: ToolCallFunctionDelta {
1590 name: Some(name),
1591 arguments: Some(String::new()),
1592 },
1593 }
1594 }
1595
1596 fn arguments(index: usize, fragment: String) -> Self {
1598 ToolCallDelta {
1599 index,
1600 id: None,
1601 kind: None,
1602 function: ToolCallFunctionDelta {
1603 name: None,
1604 arguments: Some(fragment),
1605 },
1606 }
1607 }
1608}
1609
1610#[derive(Serialize, Clone)]
1611struct ToolCallFunctionOut {
1612 name: String,
1613 arguments: String,
1617}
1618
1619#[derive(Serialize)]
1620struct ChatCompletionResponse {
1621 id: String,
1622 request_id: String,
1628 object: &'static str,
1629 model: String,
1630 choices: Vec<ChatCompletionChoice>,
1631 usage: generate::Usage,
1636 frink_cache: &'static str,
1647}
1648
1649#[derive(Serialize)]
1650struct ChatCompletionChunkDelta {
1651 #[serde(skip_serializing_if = "Option::is_none")]
1652 role: Option<&'static str>,
1653 #[serde(skip_serializing_if = "Option::is_none")]
1654 content: Option<String>,
1655 #[serde(skip_serializing_if = "Option::is_none")]
1657 reasoning_content: Option<String>,
1658 #[serde(skip_serializing_if = "Option::is_none")]
1659 tool_calls: Option<Vec<ToolCallDelta>>,
1660}
1661
1662#[derive(Serialize)]
1663struct ChatCompletionChunkChoice {
1664 index: usize,
1665 delta: ChatCompletionChunkDelta,
1666 finish_reason: Option<&'static str>,
1667}
1668
1669#[derive(Serialize)]
1670struct ChatCompletionChunk {
1671 id: String,
1672 #[serde(skip_serializing_if = "Option::is_none")]
1679 request_id: Option<String>,
1680 object: &'static str,
1681 model: String,
1682 choices: Vec<ChatCompletionChunkChoice>,
1683 #[serde(skip_serializing_if = "Option::is_none")]
1686 usage: Option<generate::Usage>,
1687}
1688
1689async fn health(State(state): State<Arc<AppState>>) -> Response {
1695 let snapshot = state.detection.snapshot();
1696 let mut capabilities = snapshot.capabilities;
1697 let active = state.active();
1698
1699 capabilities.push(match active.as_deref() {
1702 None => frink_api::Capability::unavailable(
1707 frink_api::health::capability::REAL_WEIGHTS,
1708 frink_api::health::reason::MODEL_NOT_LOADED,
1709 "No model is loaded. POST /admin/models/load with an id from GET /admin/models.",
1710 ),
1711 Some(active) if active.is_synthetic() => frink_api::Capability::unavailable(
1712 frink_api::health::capability::REAL_WEIGHTS,
1713 frink_api::health::reason::MODEL_NOT_LOADED,
1714 "Serving synthetic random weights: set FRINK_MODEL_PATH (or -m) to a real \
1715 checkpoint. Output from this model is noise.",
1716 ),
1717 Some(active) if active.encoder().is_some() => {
1727 let endpoints = active
1728 .encoder()
1729 .map(|e| encoder_endpoints(e))
1730 .unwrap_or_default();
1731 let served_by = match endpoints.is_empty() {
1732 true => "no endpoint in this build serves it".to_string(),
1733 false => format!("served by {}", endpoints.join(" and ")),
1734 };
1735 frink_api::Capability::available(
1736 frink_api::health::capability::REAL_WEIGHTS,
1737 format!(
1738 "Serving the real embedding checkpoint '{}'. This is an ENCODER, \
1739 {served_by}; generation endpoints refuse it.",
1740 active.name(),
1741 ),
1742 )
1743 }
1744 Some(active) => frink_api::Capability::available(
1745 frink_api::health::capability::REAL_WEIGHTS,
1746 format!("Serving the real checkpoint '{}'.", active.name()),
1747 ),
1748 });
1749 capabilities.push(if active.as_ref().is_some_and(|a| a.batcher.is_some()) {
1750 frink_api::Capability::available(
1751 frink_api::health::capability::CONTINUOUS_BATCHING,
1752 if state.continuous_batching_enabled && continuous_batching_env().is_none() {
1753 "On by default on Metal. Concurrent requests share one batched decode worker."
1754 } else {
1755 "Concurrent requests share one batched decode step."
1756 },
1757 )
1758 } else if state.metal_private_decode_gate.is_some() {
1759 frink_api::Capability::unavailable(
1760 frink_api::health::capability::CONTINUOUS_BATCHING,
1761 frink_api::health::reason::DISABLED,
1762 "Off; private Metal decodes serialize (one at a time). Set FRINK_CONTINUOUS_BATCHING=1 or --cont-batching for parallel serving.",
1763 )
1764 } else {
1765 frink_api::Capability::unavailable(
1766 frink_api::health::capability::CONTINUOUS_BATCHING,
1767 frink_api::health::reason::DISABLED,
1768 "Off; set FRINK_CONTINUOUS_BATCHING=1 (incompatible with a KV pool or prefix cache).",
1769 )
1770 });
1771
1772 let last_request_ms = state
1773 .last_request_ms
1774 .load(std::sync::atomic::Ordering::Relaxed);
1775 let uptime = state.started_at.elapsed();
1776 let health_state = if active.is_none() {
1784 frink_api::HealthState::Unavailable
1785 } else {
1786 snapshot.state
1787 };
1788 let body = frink_api::HealthResponse {
1789 state: health_state,
1790 reason: match health_state {
1791 frink_api::HealthState::Ready => None,
1792 frink_api::HealthState::Unavailable => {
1793 Some(frink_api::health::reason::MODEL_NOT_LOADED.to_string())
1794 }
1795 frink_api::HealthState::Detecting => {
1796 Some(frink_api::health::reason::DETECTING.to_string())
1797 }
1798 },
1799 detail: match health_state {
1800 frink_api::HealthState::Ready => None,
1801 frink_api::HealthState::Unavailable => Some(
1802 "No model is loaded. POST /admin/models/load with an id from GET /admin/models."
1803 .to_string(),
1804 ),
1805 frink_api::HealthState::Detecting => {
1806 Some("Probing available compute backends.".to_string())
1807 }
1808 },
1809 model: active
1810 .as_deref()
1811 .map(|active| frink_api::health::ModelSummary {
1812 id: active.name().to_string(),
1813 tokenizer: active.tokenizer_kind().to_string(),
1814 synthetic_weights: active.is_synthetic(),
1815 }),
1816 capabilities,
1817 version: env!("CARGO_PKG_VERSION").to_string(),
1818 pid: std::process::id(),
1819 uptime_seconds: uptime.as_secs_f64(),
1820 server_time_unix_ms: std::time::SystemTime::now()
1821 .duration_since(std::time::UNIX_EPOCH)
1822 .map(|d| d.as_millis().min(u64::MAX as u128) as u64)
1823 .unwrap_or(0),
1824 last_request_age_seconds: (last_request_ms > 0)
1825 .then(|| uptime.as_secs_f64() - (last_request_ms as f64 / 1000.0))
1826 .map(|age| age.max(0.0)),
1827 };
1828
1829 let status =
1830 StatusCode::from_u16(body.state.http_status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
1831 (status, Json(body)).into_response()
1832}
1833
1834async fn list_models(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
1835 let Some(active) = state.active() else {
1839 return Json(serde_json::json!({ "object": "list", "data": [] }));
1840 };
1841 let mut model_entry = serde_json::json!({
1842 "id": active.name(),
1843 "object": "model",
1844 "frink_synthetic_weights": active.is_synthetic(),
1845 "frink_tokenizer": active.tokenizer_kind(),
1846 });
1847 if let Some(encoder) = active.encoder() {
1853 model_entry["frink_model_kind"] = serde_json::json!("embedding");
1854 model_entry["frink_endpoints"] = serde_json::json!(encoder_endpoints(encoder));
1855 model_entry["frink_n_embd"] = serde_json::json!(encoder.n_embd());
1856 model_entry["frink_pooling"] = serde_json::json!(encoder.pooling_type().name());
1857 model_entry["frink_context_length"] = serde_json::json!(encoder.n_ctx_train());
1858 }
1859 if let Some(model) = active.generative_opt() {
1867 let parser_configured = active.reasoning_format().is_some();
1868 let gears = model.chat_template().think_gears(parser_configured);
1869 if !gears.is_empty() {
1870 model_entry["supported_reasoning_efforts"] = serde_json::json!(gears.supported);
1871 if let Some(default) = &gears.default {
1872 model_entry["default_reasoning_effort"] = serde_json::json!(default);
1873 }
1874 model_entry["reasoning_effort_kwargs"] = serde_json::json!(gears.kwargs);
1877 }
1878 }
1879 if let Some(mcp) = &state.mcp {
1880 model_entry["frink_mcp"] = mcp.models_metadata();
1881 }
1882 Json(serde_json::json!({
1883 "object": "list",
1884 "data": [model_entry]
1885 }))
1886}
1887
1888async fn serving_stats(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
1902 let now_ms = state.uptime().as_millis().min(u64::MAX as u128) as u64;
1903 let mut serving = state.serving.lock().unwrap_or_else(|p| p.into_inner());
1904 let active = state.active();
1905 Json(serde_json::json!({
1906 "model": active.as_ref().map(|a| a.name()),
1907 "state": state
1908 .maintenance
1909 .lock()
1910 .unwrap_or_else(|p| p.into_inner())
1911 .state()
1912 .as_str(),
1913 "uptime_s": state.uptime().as_secs(),
1914 "throughput": {
1915 "decode_tps": (serving.decode_tokens_per_second(now_ms) * 10.0).round() / 10.0,
1916 "prefill_tps": (serving.prefill_tokens_per_second(now_ms) * 10.0).round() / 10.0,
1917 },
1918 "requests": {
1919 "active": state.cancels.live_count(),
1920 "completed": state.stats.recorded_total(),
1921 "p95_ms": state.stats.p95_duration_ms(),
1922 "ttft_mean_ms": state.stats.ttft_mean_ms(),
1923 "prompt_tokens_total": state.stats.tokens_prompt_total(),
1924 "completion_tokens_total": state.stats.tokens_generated_total(),
1925 },
1926 "pools": cache_admin::pool_gauges(&state),
1932 "memory": cache_admin::footprint_json(&state),
1935 }))
1936}
1937
1938#[derive(Deserialize)]
1939struct RequestsQuery {
1940 #[serde(default)]
1941 since: u64,
1942 #[serde(default = "default_requests_limit")]
1943 limit: usize,
1944}
1945
1946fn default_requests_limit() -> usize {
1947 stats::MAX_PAGE
1948}
1949
1950async fn recent_requests(
1958 State(state): State<Arc<AppState>>,
1959 axum::extract::Query(q): axum::extract::Query<RequestsQuery>,
1960) -> Json<serde_json::Value> {
1961 let (rows, cursor, missed) = state.stats.page(q.since, q.limit);
1962 Json(serde_json::json!({
1963 "requests": rows,
1964 "next_cursor": cursor,
1965 "missed": missed,
1966 "total": state.stats.recorded_total(),
1967 }))
1968}
1969
1970#[derive(Serialize)]
1971struct CombinedCacheStats {
1972 response_cache: response_cache::CacheStats,
1973 prefix_cache: Option<frink_models::PrefixCacheStats>,
1975}
1976
1977async fn cache_stats(State(state): State<Arc<AppState>>) -> Json<CombinedCacheStats> {
1978 Json(CombinedCacheStats {
1979 response_cache: lock_cache(&state.response_cache).stats(),
1980 prefix_cache: state
1981 .prefix_cache
1982 .as_ref()
1983 .map(|pc| pc.lock().unwrap_or_else(|p| p.into_inner()).stats()),
1984 })
1985}
1986
1987async fn metrics(State(state): State<Arc<AppState>>) -> Response {
1992 use std::sync::atomic::Ordering;
1993
1994 let cache_stats = lock_cache(&state.response_cache).stats();
1995 let active = state.active();
1996 let requests_total = state.requests_total.load(Ordering::Relaxed);
1997 let errors_total = state.request_errors_total.load(Ordering::Relaxed);
1998 let uptime = state.started_at.elapsed().as_secs_f64();
1999
2000 let body = format!(
2001 "# HELP frink_requests_total Total chat completion requests received.\n\
2002 # TYPE frink_requests_total counter\n\
2003 frink_requests_total {requests_total}\n\
2004 # HELP frink_request_errors_total Total chat completion requests that returned an error.\n\
2005 # TYPE frink_request_errors_total counter\n\
2006 frink_request_errors_total {errors_total}\n\
2007 # HELP frink_cache_hits_total Whole-response cache hits.\n\
2008 # TYPE frink_cache_hits_total counter\n\
2009 frink_cache_hits_total {}\n\
2010 # HELP frink_cache_misses_total Whole-response cache misses.\n\
2011 # TYPE frink_cache_misses_total counter\n\
2012 frink_cache_misses_total {}\n\
2013 # HELP frink_cache_entries Current whole-response cache entry count.\n\
2014 # TYPE frink_cache_entries gauge\n\
2015 frink_cache_entries {}\n\
2016 # HELP frink_synthetic_weights 1 if serving synthetic random weights instead of a real checkpoint.\n\
2017 # TYPE frink_synthetic_weights gauge\n\
2018 frink_synthetic_weights {}\n\
2019 # HELP frink_uptime_seconds Seconds since this server process started.\n\
2020 # TYPE frink_uptime_seconds gauge\n\
2021 frink_uptime_seconds {uptime}\n",
2022 cache_stats.hits,
2023 cache_stats.misses,
2024 cache_stats.entries,
2025 active
2029 .as_ref()
2030 .map(|a| a.is_synthetic() as u8)
2031 .unwrap_or(0),
2032 );
2033
2034 let body = match active
2038 .as_ref()
2039 .and_then(|a| a.expert_store_stats())
2040 {
2041 Some(es) => format!(
2042 "{body}\
2043 # HELP frink_expert_cache_hits_total Expert-store cache hits.\n\
2044 # TYPE frink_expert_cache_hits_total counter\n\
2045 frink_expert_cache_hits_total {}\n\
2046 # HELP frink_expert_cache_misses_total Expert-store cache misses (source reads).\n\
2047 # TYPE frink_expert_cache_misses_total counter\n\
2048 frink_expert_cache_misses_total {}\n\
2049 # HELP frink_expert_cache_evictions_total Expert-store LRU evictions.\n\
2050 # TYPE frink_expert_cache_evictions_total counter\n\
2051 frink_expert_cache_evictions_total {}\n\
2052 # HELP frink_expert_cache_pass_throughs_total Acquires served uncached (entry could not fit the budget).\n\
2053 # TYPE frink_expert_cache_pass_throughs_total counter\n\
2054 frink_expert_cache_pass_throughs_total {}\n\
2055 # HELP frink_expert_cache_bytes_read_total Bytes read from the checkpoint for expert misses.\n\
2056 # TYPE frink_expert_cache_bytes_read_total counter\n\
2057 frink_expert_cache_bytes_read_total {}\n\
2058 # HELP frink_expert_cache_resident_bytes Current expert-cache footprint in bytes.\n\
2059 # TYPE frink_expert_cache_resident_bytes gauge\n\
2060 frink_expert_cache_resident_bytes {}\n",
2061 es.hits, es.misses, es.evictions, es.pass_throughs, es.bytes_read, es.resident_bytes,
2062 ),
2063 None => body,
2064 };
2065
2066 let body = match active.as_ref().and_then(|a| a.batcher.as_ref()) {
2071 Some(batcher) => {
2072 let sched = batcher.stats();
2073 format!(
2074 "{body}\
2075 # HELP frink_prefill_chunks_total Bounded prefill chunks the batch scheduler has run.\n\
2076 # TYPE frink_prefill_chunks_total counter\n\
2077 frink_prefill_chunks_total {}\n\
2078 # HELP frink_prefill_tokens_total Prompt tokens run through chunked prefill.\n\
2079 # TYPE frink_prefill_tokens_total counter\n\
2080 frink_prefill_tokens_total {}\n\
2081 # HELP frink_decode_steps_total Batched decode steps the batch scheduler has run.\n\
2082 # TYPE frink_decode_steps_total counter\n\
2083 frink_decode_steps_total {}\n\
2084 # HELP frink_scheduler_queue_depth Requests waiting for admission to the batch scheduler.\n\
2085 # TYPE frink_scheduler_queue_depth gauge\n\
2086 frink_scheduler_queue_depth {}\n\
2087 # HELP frink_scheduler_queue_rejected_total Requests refused with 503 because the admission queue was full.\n\
2088 # TYPE frink_scheduler_queue_rejected_total counter\n\
2089 frink_scheduler_queue_rejected_total {}\n\
2090 # HELP frink_kv_blocks_total KV blocks in the scheduler's admission budget (0 when unconfigured).\n\
2091 # TYPE frink_kv_blocks_total gauge\n\
2092 frink_kv_blocks_total {}\n\
2093 # HELP frink_kv_blocks_free KV blocks not reserved by an in-flight request.\n\
2094 # TYPE frink_kv_blocks_free gauge\n\
2095 frink_kv_blocks_free {}\n\
2096 # HELP frink_kv_block_size Token positions per KV block.\n\
2097 # TYPE frink_kv_block_size gauge\n\
2098 frink_kv_block_size {}\n\
2099 # HELP frink_kv_rejected_too_large_total Requests refused with 400 because they exceed the whole KV block budget.\n\
2100 # TYPE frink_kv_rejected_too_large_total counter\n\
2101 frink_kv_rejected_too_large_total {}\n\
2102 # HELP frink_kv_rejected_context_length_total Requests refused with 400 for exceeding the per-request context ceiling.\n\
2103 # TYPE frink_kv_rejected_context_length_total counter\n\
2104 frink_kv_rejected_context_length_total {}\n\
2105 # HELP frink_scheduler_aborted_total Requests the batch scheduler stopped because they were cancelled.\n\
2106 # TYPE frink_scheduler_aborted_total counter\n\
2107 frink_scheduler_aborted_total {}\n\
2108 # HELP frink_scheduler_max_seqs Cap on in-flight sequences (-np / FRINK_CB_MAX_SEQS); 0 when unlimited.\n\
2109 # TYPE frink_scheduler_max_seqs gauge\n\
2110 frink_scheduler_max_seqs {}\n\
2111 # HELP frink_scheduler_prefill_chunk Prompt tokens per prefill chunk (-b / -ub / FRINK_CB_PREFILL_CHUNK).\n\
2112 # TYPE frink_scheduler_prefill_chunk gauge\n\
2113 frink_scheduler_prefill_chunk {}\n",
2114 sched.prefill_chunks,
2115 sched.prefill_tokens,
2116 sched.decode_steps,
2117 sched.queue_depth,
2118 sched.queue_rejected,
2119 sched.kv_blocks_total,
2120 sched.kv_blocks_free,
2121 sched.kv_block_size,
2122 sched.kv_rejected_too_large,
2123 sched.kv_rejected_context_length,
2124 sched.aborted,
2125 sched.max_seqs,
2126 sched.prefill_chunk,
2127 )
2128 }
2129 None => body,
2130 };
2131
2132 (
2133 [(
2134 axum::http::header::CONTENT_TYPE,
2135 "text/plain; version=0.0.4",
2136 )],
2137 body,
2138 )
2139 .into_response()
2140}
2141
2142pub(crate) type ApiError = (StatusCode, Json<serde_json::Value>);
2143
2144pub(crate) fn invalid_request(message: &str, param: &str) -> ApiError {
2150 (
2151 StatusCode::BAD_REQUEST,
2152 Json(serde_json::json!({"error": {
2153 "message": message,
2154 "type": "invalid_request_error",
2155 "param": param,
2156 "code": null,
2157 }})),
2158 )
2159}
2160
2161pub(crate) fn unsupported_feature(message: &str) -> ApiError {
2162 (
2163 StatusCode::NOT_IMPLEMENTED,
2164 Json(serde_json::json!({"error": {"message": message, "type": "unsupported"}})),
2165 )
2166}
2167
2168pub(crate) fn decode_error_response(e: generate::DecodeError) -> ApiError {
2169 let status = match e {
2170 generate::DecodeError::TokenOutOfVocab { .. } => StatusCode::BAD_REQUEST,
2171 generate::DecodeError::Unsupported(_) => StatusCode::NOT_IMPLEMENTED,
2175 generate::DecodeError::KvBudgetExceeded { .. } => StatusCode::BAD_REQUEST,
2180 generate::DecodeError::KvPoolExhausted | generate::DecodeError::QueueFull { .. } => {
2186 StatusCode::SERVICE_UNAVAILABLE
2187 }
2188 generate::DecodeError::GrammarConstraint { .. } => StatusCode::BAD_REQUEST,
2192 generate::DecodeError::ReasoningBudget { .. } => StatusCode::INTERNAL_SERVER_ERROR,
2196 };
2197 tracing::warn!("decode error: {e}");
2198 let mut body = serde_json::json!({"error": {"message": e.to_string()}});
2199 if let generate::DecodeError::KvBudgetExceeded {
2205 binding,
2206 estimated_bytes,
2207 limit_bytes,
2208 positions,
2209 positions_limit,
2210 ..
2211 } = &e
2212 {
2213 body["error"]["type"] = serde_json::json!("invalid_request_error");
2214 body["error"]["code"] = serde_json::json!(binding);
2215 body["error"]["binding"] = serde_json::json!(binding);
2216 body["error"]["estimated_bytes"] = serde_json::json!(estimated_bytes);
2217 body["error"]["limit_bytes"] = serde_json::json!(limit_bytes);
2218 body["error"]["positions"] = serde_json::json!(positions);
2219 body["error"]["positions_limit"] = serde_json::json!(positions_limit);
2220 }
2221 if let Some(secs) = e.retry_after_secs() {
2225 body["error"]["retry_after_seconds"] = serde_json::json!(secs);
2226 }
2227 (status, Json(body))
2228}
2229
2230pub(crate) fn join_error_response(e: tokio::task::JoinError) -> ApiError {
2231 tracing::error!("generation task panicked: {e}");
2232 (
2233 StatusCode::INTERNAL_SERVER_ERROR,
2234 Json(serde_json::json!({"error": {"message": "internal error during generation"}})),
2235 )
2236}
2237
2238#[allow(clippy::too_many_arguments)] fn run_generation_emit(
2248 model: &Model,
2249 prompt: &str,
2250 params: &GenerationParams,
2251 kv_pool: Option<&generate::KvPoolConfig>,
2252 paged_kv: Option<&generate::PagedKvConfig>,
2253 prefix_cache: Option<&Mutex<PrefixCache>>,
2254 continuous_batcher: Option<&serving::batch::ContinuousBatcher>,
2255 ceiling: Option<&budget::ContextCeiling>,
2256 metal_private_decode_gate: Option<&std::sync::Mutex<()>>,
2257 mut emit: impl FnMut(usize, &str),
2262) -> Result<generate::Generated, generate::DecodeError> {
2263 let synthetic = model.is_synthetic();
2264 let _lora_lease = lora::lease(model, params.lora.as_deref());
2269 let mut chunks: Vec<Vec<String>> = vec![Vec::new(); params.n.max(1)];
2270 let params = &{
2276 let mut resolved = params.clone();
2277 resolved.stop_token_ids = crate::stop::resolve_stop_tokens(&resolved.stop, |text| {
2278 model.encode(text, SpecialTokens::Parse)
2279 });
2280 resolved.reasoning_budget = resolved
2286 .reasoning_budget
2287 .armed(resolved.reasoning, prompt, |text| {
2288 model.encode(text, SpecialTokens::Parse)
2289 })
2290 .map_err(|detail| generate::DecodeError::ReasoningBudget { detail })?;
2291 resolved
2292 };
2293 let used_batcher = matches!((model, continuous_batcher), (Model::Gguf(_), Some(_)));
2294 let _metal_private_guard =
2295 acquire_metal_private_decode_gate(metal_private_decode_gate, used_batcher);
2296 let (finishes, prompt_rows, prompt_ids, usage) = match model {
2297 Model::Gguf(m) => {
2298 if let Some(batcher) = continuous_batcher {
2299 let mut tokens = m.tokenizer.encode(prompt, SpecialTokens::Parse);
2300 frink_models::tokenizer::prepend_bos(&mut tokens, m.bos_id);
2301 let (finish, _generated_ids, text, usage) = if synthetic {
2302 batcher.generate(tokens, params.clone(), m.stop_tokens.clone())?
2303 } else {
2304 batcher.generate_streaming(
2305 tokens,
2306 params.clone(),
2307 m.stop_tokens.clone(),
2308 Some(|chunk: &str| {
2309 if !chunk.is_empty() {
2310 chunks[0].push(chunk.to_string());
2311 emit(0, chunk);
2312 }
2313 }),
2314 )?
2315 };
2316 if !text.is_empty() && chunks[0].is_empty() {
2317 chunks[0].push(text);
2318 }
2319 (vec![(finish, Vec::new())], Vec::new(), Vec::new(), usage)
2329 } else {
2330 generate::generate(
2331 &m.decoder,
2332 m.tokenizer.as_ref(),
2333 &m.stop_tokens,
2334 m.bos_id,
2335 prompt,
2336 params,
2337 kv_pool,
2338 paged_kv,
2339 prefix_cache,
2340 ceiling,
2341 |choice, chunk| {
2342 chunks[choice].push(chunk.to_string());
2343 if !synthetic {
2347 emit(choice, chunk);
2348 }
2349 },
2350 )?
2351 }
2352 }
2353 Model::Kimi(m) => generate::generate_engine(
2354 &m.engine,
2355 &m.tokenizer,
2356 &m.stop_tokens,
2357 None,
2358 prompt,
2359 params,
2360 |chunk| {
2361 chunks[0].push(chunk.to_string());
2362 if !synthetic {
2363 emit(0, chunk);
2364 }
2365 },
2366 )?,
2367 Model::Mla(m) => generate::generate_engine(
2368 &m.engine,
2369 &m.tokenizer,
2370 &m.stop_tokens,
2371 m.bos_id,
2372 prompt,
2373 params,
2374 |chunk| {
2375 chunks[0].push(chunk.to_string());
2376 if !synthetic {
2377 emit(0, chunk);
2378 }
2379 },
2380 )?,
2381 Model::Gemma4(m) => generate::generate_engine(
2382 &m.engine,
2383 &m.tokenizer,
2384 &m.stop_tokens,
2385 m.bos_id,
2386 prompt,
2387 params,
2388 |chunk| {
2389 chunks[0].push(chunk.to_string());
2390 if !synthetic {
2391 emit(0, chunk);
2392 }
2393 },
2394 )?,
2395 Model::Glm52(m) => generate::generate_engine(
2396 &m.engine,
2397 &m.tokenizer,
2398 &m.stop_tokens,
2399 m.bos_id,
2400 prompt,
2401 params,
2402 |chunk| {
2403 chunks[0].push(chunk.to_string());
2404 if !synthetic {
2405 emit(0, chunk);
2406 }
2407 },
2408 )?,
2409 };
2410
2411 let mut full = chunks[0].concat();
2412 if synthetic {
2413 full = format!(
2414 "[frink synthetic-weight demo: no real checkpoint loaded -- set FRINK_MODEL_PATH \
2415 to serve a real model. Decoded ids -> {full:?}]"
2416 );
2417 emit(0, &full);
2418 } else if used_batcher && !full.is_empty() && chunks[0].is_empty() {
2419 emit(0, &full);
2420 }
2421
2422 debug_assert_eq!(finishes.len(), chunks.len(), "one finish reason per choice");
2428 let mut out: Vec<generate::GeneratedChoice> = finishes
2429 .into_iter()
2430 .zip(chunks.into_iter().map(|c| c.concat()))
2431 .map(|((finish, logprobs), text)| generate::GeneratedChoice {
2432 finish,
2433 text,
2434 logprobs,
2435 })
2436 .collect();
2437 if let Some(first) = out.first_mut() {
2438 if synthetic {
2446 first.logprobs.clear();
2447 }
2448 first.text = full;
2449 }
2450 Ok(generate::Generated {
2451 choices: out,
2452 prompt_rows,
2453 prompt_ids,
2454 usage,
2455 })
2456}
2457
2458#[allow(clippy::too_many_arguments)] pub(crate) fn run_generation(
2463 model: &Model,
2464 prompt: &str,
2465 params: &GenerationParams,
2466 kv_pool: Option<&generate::KvPoolConfig>,
2467 paged_kv: Option<&generate::PagedKvConfig>,
2468 prefix_cache: Option<&Mutex<PrefixCache>>,
2469 continuous_batcher: Option<&serving::batch::ContinuousBatcher>,
2470 ceiling: Option<&budget::ContextCeiling>,
2471 metal_private_decode_gate: Option<&std::sync::Mutex<()>>,
2472 ) -> Result<generate::Generated, generate::DecodeError> {
2475 run_generation_emit(
2476 model,
2477 prompt,
2478 params,
2479 kv_pool,
2480 paged_kv,
2481 prefix_cache,
2482 continuous_batcher,
2483 ceiling,
2484 metal_private_decode_gate,
2485 |_, _| {},
2486 )
2487}
2488
2489pub(crate) fn prompt_from_messages(
2500 messages: &[ChatMessage],
2501 template: &chat_template::PromptTemplate,
2502 tools: &[ToolDef],
2503 extra: serde_json::Map<String, serde_json::Value>,
2504) -> Result<String, ApiError> {
2505 let rendered = if tools.is_empty() || template.handles_tools() {
2506 template.render(messages, tools, extra)
2507 } else {
2508 let mut with_preamble = Vec::with_capacity(messages.len() + 1);
2509 with_preamble.push(ChatMessage {
2510 role: "system".to_string(),
2511 content: Some(MessageContent::Text(tool_preamble(tools))),
2512 tool_calls: None,
2513 tool_call_id: None,
2514 reasoning_content: None,
2515 });
2516 with_preamble.extend_from_slice(messages);
2517 template.render(&with_preamble, &[], extra)
2518 };
2519 rendered.map_err(template_error_response)
2520}
2521
2522fn template_error_response(err: frink_models::chat_template::TemplateError) -> ApiError {
2527 (
2528 StatusCode::BAD_REQUEST,
2529 Json(serde_json::json!({
2530 "error": {
2531 "message": format!("chat template failed to render: {err}"),
2532 "type": "invalid_request_error",
2533 "param": "messages",
2534 "code": null,
2535 }
2536 })),
2537 )
2538}
2539
2540fn tool_preamble(tools: &[ToolDef]) -> String {
2553 let mut out = String::from(
2554 "You can call tools to help answer the user. To call a tool, respond with \
2555 EXACTLY one line in this format and nothing else:\n\
2556 <tool_call>{\"name\": \"<tool name>\", \"arguments\": {<arguments as a JSON \
2557 object matching that tool's parameters>}}</tool_call>\n\n\
2558 Available tools:\n",
2559 );
2560 for t in tools {
2561 out.push_str(&format!(
2562 "- {}: {}\n parameters (JSON schema): {}\n",
2563 t.function.name,
2564 t.function.description.as_deref().unwrap_or(""),
2565 t.function
2566 .parameters
2567 .as_ref()
2568 .map(|v| v.to_string())
2569 .unwrap_or_else(|| "{}".to_string()),
2570 ));
2571 }
2572 out
2573}
2574
2575fn tool_call_deltas(
2585 events: Vec<crate::policy::parser::ToolCallEvent>,
2586 opened: &std::cell::Cell<usize>,
2587) -> (String, Vec<ToolCallDelta>) {
2588 let mut text = String::new();
2589 let mut deltas = Vec::new();
2590 for event in events {
2591 match event {
2592 crate::policy::parser::ToolCallEvent::Text(chunk) => text.push_str(&chunk),
2593 crate::policy::parser::ToolCallEvent::CallStart { index, name } => {
2594 opened.set(opened.get().max(index + 1));
2595 deltas.push(ToolCallDelta::opening(index, name));
2596 }
2597 crate::policy::parser::ToolCallEvent::CallArguments { index, fragment } => {
2598 if !fragment.is_empty() {
2599 deltas.push(ToolCallDelta::arguments(index, fragment));
2600 }
2601 }
2602 crate::policy::parser::ToolCallEvent::CallEnd { .. } => {}
2603 }
2604 }
2605 (text, deltas)
2606}
2607
2608fn build_response_message(
2619 text: String,
2620 tools: &[ToolDef],
2621 posture: output::OutputPosture,
2622 base_finish: &'static str,
2623) -> (ChatCompletionResponseMessage, &'static str) {
2624 let parsed = output::parse_output(&text, tools, posture);
2625 let calls: Vec<ToolCallOut> = parsed
2626 .calls
2627 .into_iter()
2628 .enumerate()
2629 .map(|(index, call)| ToolCallOut {
2630 id: format!("call_{index}"),
2631 kind: "function",
2632 function: ToolCallFunctionOut {
2633 name: call.name,
2634 arguments: call.arguments,
2635 },
2636 })
2637 .collect();
2638 if !calls.is_empty() {
2639 return (
2640 ChatCompletionResponseMessage {
2641 role: "assistant",
2642 content: None,
2643 reasoning_content: parsed.reasoning,
2644 tool_calls: Some(calls),
2645 },
2646 "tool_calls",
2647 );
2648 }
2649 (
2650 ChatCompletionResponseMessage {
2651 role: "assistant",
2652 content: Some(parsed.content),
2653 reasoning_content: parsed.reasoning,
2654 tool_calls: None,
2655 },
2656 base_finish,
2657 )
2658}
2659
2660fn resolve_history(state: &AppState, req: &ChatCompletionRequest) -> Vec<ChatMessage> {
2665 let mut history = match &req.session_id {
2666 Some(id) => state.sessions.extend_and_get(id, &req.messages),
2667 None => req.messages.clone(),
2668 };
2669 if req.json_object_mode() {
2670 inject_json_object_system_hint(&mut history);
2671 }
2672 history
2673}
2674
2675fn inject_json_object_system_hint(messages: &mut Vec<ChatMessage>) {
2676 const HINT: &str =
2677 "You must respond with valid JSON only (a single JSON object, no markdown fences).";
2678 if let Some(sys) = messages.iter_mut().find(|m| m.role == "system") {
2679 match &mut sys.content {
2680 Some(MessageContent::Text(s)) if !s.contains("JSON") => {
2681 s.push_str("\n\n");
2682 s.push_str(HINT);
2683 }
2684 None => {
2685 sys.content = Some(MessageContent::Text(HINT.to_string()));
2686 }
2687 _ => {}
2688 }
2689 } else {
2690 messages.insert(
2691 0,
2692 ChatMessage {
2693 role: "system".to_string(),
2694 content: Some(MessageContent::Text(HINT.to_string())),
2695 tool_calls: None,
2696 tool_call_id: None,
2697 reasoning_content: None,
2698 },
2699 );
2700 }
2701}
2702
2703async fn chat_completions(
2704 State(state): State<Arc<AppState>>,
2705 headers: axum::http::HeaderMap,
2706 Json(req): Json<ChatCompletionRequest>,
2707) -> Response {
2708 let attribution = attribution::Attribution::from_headers(&headers);
2709 state
2710 .requests_total
2711 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2712 let started = std::time::Instant::now();
2713
2714 let request_id = frink_api::next_request_id();
2718 let stream = req.stream.unwrap_or(false);
2719
2720 let refusal = cache_admin::check_admission(&state)
2726 .err()
2727 .or_else(|| req.validate_supported_fields().err());
2728 if let Some(err) = refusal {
2729 state
2730 .request_errors_total
2731 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2732 let response = err.into_response();
2733 state.record_request(stats::Record {
2734 request_id: &request_id,
2735 route: frink_api::routes::V1_CHAT_COMPLETIONS,
2736 model: state.active_model_name(),
2737 status: response.status().as_u16(),
2738 stream,
2739 duration_ms: started.elapsed().as_millis() as u64,
2740 usage: None,
2741 attribution: &attribution,
2742 });
2743 return response;
2744 }
2745
2746 let response = if stream {
2747 chat_completions_stream(
2748 Arc::clone(&state),
2749 req,
2750 request_id.clone(),
2751 started,
2752 attribution.clone(),
2753 )
2754 .await
2755 .into_response()
2756 } else {
2757 chat_completions_full(
2758 Arc::clone(&state),
2759 req,
2760 request_id.clone(),
2761 started,
2762 attribution.clone(),
2763 )
2764 .await
2765 .into_response()
2766 };
2767
2768 if response.status().is_client_error() || response.status().is_server_error() {
2769 state
2770 .request_errors_total
2771 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2772 state.record_request(stats::Record {
2776 request_id: &request_id,
2777 route: frink_api::routes::V1_CHAT_COMPLETIONS,
2778 model: state.active_model_name(),
2781 status: response.status().as_u16(),
2782 stream,
2783 duration_ms: started.elapsed().as_millis() as u64,
2784 usage: None,
2785 attribution: &attribution,
2786 });
2787 }
2788 state.mark_request_finished();
2789
2790 response
2791}
2792
2793async fn chat_completions_full(
2794 state: Arc<AppState>,
2795 req: ChatCompletionRequest,
2796 request_id: String,
2797 started: std::time::Instant,
2798 attribution: attribution::Attribution,
2799) -> Result<Json<ChatCompletionResponse>, ApiError> {
2800 let tools_active = req.tools_active();
2801 let active = state.require_active()?;
2805 let history = resolve_history(&state, &req);
2806 let template = active.generative()?.chat_template();
2807 let kwargs = req.resolve_template_kwargs(&template);
2808 let prompt = req.render_prompt(&history, &template, &req.tools, kwargs, active.name())?;
2809 let mut params =
2817 req.generation_params_for_template(&template, active.name(), active.sampler_model())?;
2818 params.lora = lora::resolve_request(active.generative()?, req.lora.as_deref())?;
2819 let key = req.is_cacheable().then(|| req.cache_key(&prompt, ¶ms));
2820
2821 let mut generated_logprobs: Vec<crate::sampling_loop::PerTokenProbs> = Vec::new();
2825 let n_logprobs = req.n_logprobs()?;
2828 let decode_piece = |id: usize| active.decode_any(&[id]);
2830 let (completion, cache_status) = if let Some(cached) = key
2831 .as_ref()
2832 .and_then(|key| lock_cache(&state.response_cache).get(key))
2833 {
2834 tracing::debug!("cache hit for key {}", key.as_ref().unwrap().digest());
2835 (cached, "hit")
2836 } else {
2837 let produced = decode_task::buffered(
2838 decode_task::DecodeHandles::take(&state, &active)?,
2839 prompt.clone(),
2840 params,
2841 )
2842 .await?;
2843 let usage = produced.usage;
2844 let choices = produced.choices;
2845
2846 generated_logprobs = choices.iter().map(|c| c.logprobs.clone()).collect();
2851 let completion = response_cache::CachedCompletion {
2852 choices: choices.into_iter().map(|c| (c.finish, c.text)).collect(),
2853 usage,
2854 };
2855 let cache_status = match key {
2861 Some(key) => match completion.clone().cacheable() {
2865 Some(cacheable) => {
2866 tracing::debug!("cache miss for key {}", key.digest());
2867 lock_cache(&state.response_cache).put(key, cacheable);
2868 "miss"
2869 }
2870 None => "skip",
2871 },
2872 None => "skip",
2873 };
2874 (completion, cache_status)
2875 };
2876 let content = completion.first_text().to_string();
2879
2880 if req.json_object_mode() {
2881 json_mode::validate_json_object_output(&content)?;
2882 }
2883
2884 if let Some(id) = &req.session_id {
2888 state.sessions.store_reply(
2889 id,
2890 ChatMessage {
2891 role: "assistant".to_string(),
2892 content: Some(MessageContent::Text(content.clone())),
2893 tool_calls: None,
2894 tool_call_id: None,
2895 reasoning_content: None,
2896 },
2897 );
2898 }
2899
2900 let posture = output::OutputPosture::resolve_full(
2905 active.reasoning_format(),
2906 active.tool_call_format(),
2907 &prompt,
2908 );
2909 let tools: &[_] = if tools_active { &req.tools } else { &[] };
2910 let wanted = req.unimplemented.n.unwrap_or(1).max(1) as usize;
2914 let ranked: Vec<(generate::FinishReason, String)> = if completion.choices.len() > wanted {
2915 let scored: Vec<crate::generate::GeneratedChoice> = completion
2916 .choices
2917 .into_iter()
2918 .zip(
2919 generated_logprobs
2920 .iter()
2921 .cloned()
2922 .chain(std::iter::repeat(Vec::new())),
2923 )
2924 .map(
2925 |((finish, text), logprobs)| crate::generate::GeneratedChoice {
2926 finish,
2927 text,
2928 logprobs,
2929 },
2930 )
2931 .collect();
2932 let best = crate::best_of::take_best(scored, wanted);
2933 generated_logprobs = best.iter().map(|c| c.logprobs.clone()).collect();
2934 best.into_iter().map(|c| (c.finish, c.text)).collect()
2935 } else {
2936 completion.choices
2937 };
2938 let rendered: Vec<ChatCompletionChoice> = ranked
2939 .into_iter()
2940 .enumerate()
2941 .map(|(index, (finish, text))| {
2942 let (message, finish_reason) =
2943 build_response_message(text, tools, posture, finish.as_str());
2944 ChatCompletionChoice {
2945 index,
2946 message,
2947 finish_reason,
2948 logprobs: n_logprobs.map(|k| {
2949 crate::logprobs::render_chat(
2950 generated_logprobs.get(index).unwrap_or(&Vec::new()),
2951 Some(k),
2952 &decode_piece,
2953 )
2954 }),
2955 }
2956 })
2957 .collect();
2958
2959 state.record_request(stats::Record {
2960 request_id: &request_id,
2961 route: frink_api::routes::V1_CHAT_COMPLETIONS,
2962 model: Some(active.name().to_string()),
2965 status: 200,
2966 stream: false,
2967 duration_ms: started.elapsed().as_millis() as u64,
2968 usage: Some(&completion.usage),
2969 attribution: &attribution,
2970 });
2971
2972 Ok(Json(ChatCompletionResponse {
2973 id: request_id.clone(),
2974 request_id,
2975 object: "chat.completion",
2976 model: req.model,
2977 choices: rendered,
2978 usage: completion.usage,
2979 frink_cache: cache_status,
2980 }))
2981}
2982
2983async fn chat_completions_stream(
2984 state: Arc<AppState>,
2985 req: ChatCompletionRequest,
2986 request_id: String,
2987 started: std::time::Instant,
2988 attribution: attribution::Attribution,
2989) -> Result<Response, ApiError> {
2990 let tools_active = req.tools_active();
3000 let active = state.require_active()?;
3004 let history = resolve_history(&state, &req);
3005 let template = active.generative()?.chat_template();
3006 let kwargs = req.resolve_template_kwargs(&template);
3007 let prompt = req.render_prompt(&history, &template, &req.tools, kwargs, active.name())?;
3008 let model_name = req.model.clone();
3009 let session_id = req.session_id.clone();
3010 let sessions = state.sessions.clone();
3011
3012 let model = Arc::clone(active.generative()?);
3013 let kv_pool = state.kv_pool.clone();
3014 let paged_kv = state.paged_kv.clone();
3015 let prefix_cache = state.prefix_cache.clone();
3016 let batcher = active.batcher.clone();
3017 let ceiling = active.ceiling.clone();
3018 let metal_private_decode_gate = state.metal_private_decode_gate.clone();
3019 let mut params =
3020 req.generation_params_for_template(&template, active.name(), active.sampler_model())?;
3021 params.lora = lora::resolve_request(active.generative()?, req.lora.as_deref())?;
3022 params.interleave_choices = params.n > 1;
3028 let stats_state = Arc::clone(&state);
3029 let served_model = active.name().to_string();
3032 let posture = output::OutputPosture::resolve_full(
3036 active.reasoning_format(),
3037 active.tool_call_format(),
3038 &prompt,
3039 );
3040 let offered_tools: Vec<ToolDef> = if tools_active {
3043 req.tools.clone()
3044 } else {
3045 Vec::new()
3046 };
3047
3048 let (cancel_token, cancel_guard) = state.cancels.register(&request_id);
3053 params.cancel = Some(cancel_token.clone());
3054
3055 let overlap = true;
3064
3065 let slot = req
3069 .stream_resumable
3070 .unwrap_or(false)
3071 .then(|| state.streams.register(&request_id));
3072 let emitter = resume::Emitter::new(slot);
3073
3074 let (tx, rx) = tokio::sync::mpsc::channel::<Result<Event, Infallible>>(64);
3075 let keepalive = sse::keepalive_event(&ChatCompletionChunk {
3079 id: request_id.clone(),
3080 request_id: None,
3081 object: "chat.completion.chunk",
3082 model: model_name.clone(),
3083 choices: vec![ChatCompletionChunkChoice {
3084 index: 0,
3085 delta: ChatCompletionChunkDelta {
3086 role: None,
3087 content: None,
3088 reasoning_content: None,
3089 tool_calls: None,
3090 },
3091 finish_reason: None,
3092 }],
3093 usage: None,
3094 });
3095
3096 tokio::task::spawn_blocking(move || {
3097 let _cancel_guard = cancel_guard;
3100 let tx_chunks = tx.clone();
3101 let orphan_timeout = sse::orphan_timeout_from_env();
3106 let head_request_id = request_id.clone();
3107 let announced = std::cell::Cell::new(false);
3111 let emitters: Rc<RefCell<Vec<crate::chat_stream_choice::ChoiceEmitter>>> =
3117 Rc::new(RefCell::new(
3118 (0..params.n.max(1))
3119 .map(|_| {
3120 crate::chat_stream_choice::ChoiceEmitter::new(
3121 posture.reasoning_parser(),
3122 tools_active.then(|| posture.tool_call_parser(&offered_tools)),
3123 )
3124 })
3125 .collect(),
3126 ));
3127 let emit_choices = Rc::clone(&emitters);
3128 let result = run_generation_emit(
3129 &model,
3130 &prompt,
3131 ¶ms,
3132 kv_pool.as_ref(),
3133 paged_kv.as_ref(),
3134 prefix_cache.as_deref(),
3135 batcher.as_ref(),
3136 ceiling.as_deref(),
3137 metal_private_decode_gate.as_deref(),
3138 |choice, chunk| {
3139 if !overlap || chunk.is_empty() {
3140 return;
3141 }
3142 let mut held = emit_choices.borrow_mut();
3143 let Some(emitter_state) = held.get_mut(choice) else {
3144 return;
3145 };
3146 let delta = emitter_state.push(chunk);
3147 if delta.is_empty() {
3148 return;
3149 }
3150 let request_id = (!announced.get()).then(|| {
3153 announced.set(true);
3154 head_request_id.clone()
3155 });
3156 let wire = delta.into_choice(choice, emitter_state.start());
3157 drop(held);
3158 let payload = ChatCompletionChunk {
3159 id: head_request_id.clone(),
3160 request_id,
3161 object: "chat.completion.chunk",
3162 model: model_name.clone(),
3163 choices: vec![wire],
3164 usage: None,
3165 };
3166 if let Err(why) =
3175 sse::send_or_orphan(&tx_chunks, Ok(emitter.event(&payload)), orphan_timeout)
3176 {
3177 if why == sse::SendFailure::Orphaned {
3178 tracing::warn!(
3179 "SSE stream {head_request_id} accepted nothing for the orphan \
3180 deadline; treating it as abandoned"
3181 );
3182 }
3183 if !emitter.is_resumable() {
3195 cancel_token.cancel();
3196 }
3197 }
3198 },
3199 );
3200
3201 let mut pending_request_id = (!announced.get()).then(|| request_id.clone());
3207
3208 match result {
3209 Ok(generated) => {
3210 let usage = generated.usage;
3211 let produced: Vec<(generate::FinishReason, String)> = generated
3212 .choices
3213 .into_iter()
3214 .map(|c| (c.finish, c.text))
3215 .collect();
3216 assert!(
3217 !produced.is_empty(),
3218 "a generation produces at least one choice"
3219 );
3220 if let Some(id) = &session_id {
3225 sessions.store_reply(
3226 id,
3227 ChatMessage {
3228 role: "assistant".to_string(),
3229 content: Some(MessageContent::Text(produced[0].1.clone())),
3230 tool_calls: None,
3231 tool_call_id: None,
3232 reasoning_content: None,
3233 },
3234 );
3235 }
3236 for (index, (finish, full_text)) in produced.iter().enumerate() {
3237 let (finish, full_text) = (finish.clone(), full_text.as_str());
3238 let mut streamed_finish: Option<&'static str> = None;
3244 if overlap {
3245 let (tail, first, opened) = {
3246 let mut held = emitters.borrow_mut();
3247 let state = &mut held[index];
3248 let tail = state.flush();
3249 (tail, state.start(), state.opened_calls())
3250 };
3251 if !tail.is_empty() {
3252 let payload = ChatCompletionChunk {
3253 id: request_id.clone(),
3254 request_id: pending_request_id.take(),
3255 object: "chat.completion.chunk",
3256 model: model_name.clone(),
3257 choices: vec![tail.into_choice(index, first)],
3258 usage: None,
3259 };
3260 let _ = sse::send_or_orphan(
3261 &tx,
3262 Ok(emitter.event(&payload)),
3263 orphan_timeout,
3264 );
3265 }
3266 if opened > 0 {
3267 streamed_finish = Some("tool_calls");
3268 }
3269 } else {
3270 let parsed = output::parse_output(full_text, &offered_tools, posture);
3273 let tool_calls: Vec<ToolCallDelta> = parsed
3274 .calls
3275 .iter()
3276 .enumerate()
3277 .map(|(index, call)| {
3278 ToolCallDelta::whole(
3279 index,
3280 call.name.clone(),
3281 call.arguments.clone(),
3282 )
3283 })
3284 .collect();
3285 if !tool_calls.is_empty() {
3286 streamed_finish = Some("tool_calls");
3287 }
3288 if !tool_calls.is_empty()
3289 || !parsed.content.is_empty()
3290 || parsed.reasoning.is_some()
3291 {
3292 let payload = ChatCompletionChunk {
3293 id: request_id.clone(),
3294 request_id: pending_request_id.take(),
3295 object: "chat.completion.chunk",
3296 model: model_name.clone(),
3297 choices: vec![ChatCompletionChunkChoice {
3298 index,
3299 delta: ChatCompletionChunkDelta {
3300 role: Some("assistant"),
3301 content: (!parsed.content.is_empty()
3302 && tool_calls.is_empty())
3303 .then(|| parsed.content.clone()),
3304 reasoning_content: parsed.reasoning.clone(),
3305 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
3306 },
3307 finish_reason: None,
3308 }],
3309 usage: None,
3310 };
3311 let _ = sse::send_or_orphan(
3312 &tx,
3313 Ok(emitter.event(&payload)),
3314 orphan_timeout,
3315 );
3316 }
3317 }
3318 let final_finish_reason = match streamed_finish {
3322 Some(reason) if finish.as_str() != "length" => reason,
3323 _ => finish.as_str(),
3324 };
3325 let last = index + 1 == produced.len();
3329 let final_payload = ChatCompletionChunk {
3330 id: request_id.clone(),
3331 request_id: pending_request_id.take(),
3332 object: "chat.completion.chunk",
3333 model: model_name.clone(),
3334 choices: vec![ChatCompletionChunkChoice {
3335 index,
3336 delta: ChatCompletionChunkDelta {
3337 role: None,
3338 content: None,
3339 reasoning_content: None,
3340 tool_calls: None,
3341 },
3342 finish_reason: Some(final_finish_reason),
3343 }],
3344 usage: last.then(|| usage.clone()),
3345 };
3346 let _ =
3347 sse::send_or_orphan(&tx, Ok(emitter.event(&final_payload)), orphan_timeout);
3348 }
3349 let _ = sse::send_or_orphan(&tx, Ok(emitter.done()), orphan_timeout);
3350 stats_state.record_request(stats::Record {
3355 request_id: &request_id,
3356 route: frink_api::routes::V1_CHAT_COMPLETIONS,
3357 model: Some(served_model.clone()),
3358 status: 200,
3359 stream: true,
3360 duration_ms: started.elapsed().as_millis() as u64,
3361 usage: Some(&usage),
3362 attribution: &attribution,
3363 });
3364 }
3365 Err(e) => {
3366 tracing::warn!("decode error on streamed request {request_id}: {e}");
3367 stats_state.record_request(stats::Record {
3373 request_id: &request_id,
3374 route: frink_api::routes::V1_CHAT_COMPLETIONS,
3375 model: Some(served_model.clone()),
3376 status: 500,
3377 stream: true,
3378 duration_ms: started.elapsed().as_millis() as u64,
3379 usage: None,
3380 attribution: &attribution,
3381 });
3382 let payload = ChatCompletionChunk {
3383 id: request_id.clone(),
3384 request_id: pending_request_id.take(),
3385 object: "chat.completion.chunk",
3386 model: model_name,
3387 choices: vec![ChatCompletionChunkChoice {
3388 index: 0,
3389 delta: ChatCompletionChunkDelta {
3390 role: Some("assistant"),
3391 content: Some(format!("[error: {e}]")),
3392 reasoning_content: None,
3393 tool_calls: None,
3394 },
3395 finish_reason: Some("stop"),
3396 }],
3397 usage: None,
3398 };
3399 let _ = sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3400 let _ = sse::send_or_orphan(&tx, Ok(emitter.done()), orphan_timeout);
3401 }
3402 }
3403 drop(emitter);
3407 });
3408
3409 let stream = sse::with_keepalive(rx, keepalive, sse::KEEPALIVE_INTERVAL);
3410 Ok((
3433 [(
3434 axum::http::HeaderName::from_static("x-accel-buffering"),
3435 axum::http::HeaderValue::from_static("no"),
3436 )],
3437 Sse::new(stream),
3438 )
3439 .into_response())
3440}
3441
3442fn protected_routes() -> Router<Arc<AppState>> {
3473 use frink_api::routes;
3474
3475 Router::new()
3476 .route(routes::V1_MODELS, get(list_models))
3477 .route(routes::V1_RESPONSES, post(responses::responses))
3481 .route(
3482 &axum_path(routes::V1_RESPONSE),
3483 get(responses::responses_get),
3484 )
3485 .route(
3486 &axum_path(routes::V1_RESPONSE_CANCEL),
3487 post(responses::responses_cancel),
3488 )
3489 .route(&axum_path(routes::SLOTS_ID), post(slots::post_slot))
3490 .route(routes::V1_STATS, get(serving_stats))
3491 .route(routes::V1_REQUESTS, get(recent_requests))
3492 .route(routes::V1_CACHE_STATUS, get(cache_admin::cache_status))
3493 .route(routes::V1_CACHE_REBUILD, post(cache_admin::cache_rebuild))
3494 .route(routes::ADMIN_PREPARE_STOP, post(cache_admin::prepare_stop))
3495 .route(
3496 routes::LORA_ADAPTERS,
3497 get(lora::get_lora_adapters).post(lora::post_lora_adapters),
3498 )
3499 .route(routes::V1_CHAT_COMPLETIONS, post(chat_completions))
3500 .route(routes::V1_CANCEL, post(cancel_generation))
3504 .route(&axum_path(routes::V1_STREAM), get(resume::resume))
3509 .route(&axum_path(routes::V1_STREAM_POLL), get(resume::poll))
3510 .route(routes::V1_MESSAGES, post(anthropic::messages))
3511 .route(
3512 routes::V1_MESSAGES_COUNT_TOKENS,
3513 post(anthropic::count_tokens),
3514 )
3515 .route(routes::V1_COMPLETIONS, post(openai_extra::completions))
3516 .route(routes::COMPLETION, post(completion::completion))
3521 .route(routes::COMPLETIONS, post(completion::completion))
3522 .route(routes::V1_TOKENIZE, post(openai_extra::tokenize))
3523 .route(routes::V1_DETOKENIZE, post(openai_extra::detokenize))
3524 .route(routes::TOKENIZE, post(openai_extra::tokenize))
3530 .route(routes::DETOKENIZE, post(openai_extra::detokenize))
3531 .route(routes::V1_EMBEDDINGS, post(embeddings::embeddings))
3532 .route(routes::V1_RERANK, post(rerank::rerank))
3536 .route(routes::RERANK, post(rerank::rerank))
3537 .route(routes::CACHE_STATS, get(cache_stats))
3538 .route(routes::METRICS, get(metrics))
3539 .route(routes::ADMIN_MODELS, get(admin::models))
3544 .route(routes::ADMIN_MODELS_LOAD, post(admin::load_model))
3545 .route(routes::ADMIN_MODELS_UNLOAD, post(admin::unload_model))
3546 .route(routes::SLEEP, post(admin::sleep))
3550 .route(routes::WAKE_UP, post(admin::wake_up))
3551 .route(routes::IS_SLEEPING, get(admin::is_sleeping))
3552 .route(routes::ADMIN_DOWNLOAD, post(admin::download))
3553 .route(routes::ADMIN_TASKS, get(admin::tasks))
3554 .route(&admin::cancel_route(), post(admin::cancel_task))
3555 .route(routes::ADMIN_STATS, get(admin::stats))
3556 .merge(conversations::router())
3560}
3561
3562fn axum_path(template: &str) -> String {
3563 let mut out = String::with_capacity(template.len());
3564 let mut rest = template;
3565 while let Some(open) = rest.find('{') {
3566 let Some(close) = rest[open..].find('}').map(|c| open + c) else {
3567 break;
3568 };
3569 out.push_str(&rest[..open]);
3570 out.push(':');
3571 out.push_str(&rest[open + 1..close]);
3572 rest = &rest[close + 1..];
3573 }
3574 out.push_str(rest);
3575 out
3576}
3577
3578async fn cancel_generation(
3587 State(state): State<Arc<AppState>>,
3588 Json(req): Json<frink_api::CancelGenerationRequest>,
3589) -> Response {
3590 let cancelled = state.cancels.cancel(&req.request_id);
3591 let status = if cancelled {
3592 StatusCode::OK
3593 } else {
3594 StatusCode::NOT_FOUND
3595 };
3596 let detail = if cancelled {
3597 "the generation was asked to stop; it ends at its next token".to_string()
3598 } else {
3599 "no generation with that request_id is running -- it has already \
3600 finished, was never issued, or was served by a path that does \
3601 not register for cancellation"
3602 .to_string()
3603 };
3604 (
3605 status,
3606 Json(frink_api::CancelGenerationResponse {
3607 request_id: req.request_id,
3608 cancelled,
3609 detail,
3610 }),
3611 )
3612 .into_response()
3613}
3614
3615type Activated = (
3619 Loaded,
3620 Option<serving::batch::ContinuousBatcher>,
3621 Option<Arc<budget::ContextCeiling>>,
3622);
3623
3624fn price_batcher_config(path: Option<&str>) -> serving::batch::BatcherConfig {
3642 let mut batcher = serving::batch::BatcherConfig::from_env();
3643 if batcher.max_context.is_some() && batcher.kv_blocks.is_some() {
3644 return batcher;
3647 }
3648 let Some(path) = path else {
3649 return batcher;
3650 };
3651 let priced = budget::price_gguf(path, frink_models::KvElem::F32, 1);
3657 let Some((priced, gguf_ctx, source)) = priced else {
3658 return batcher;
3659 };
3660 let Some(derived) = budget::derive_limits(&priced, gguf_ctx, batcher.kv_block_size) else {
3661 tracing::warn!(
3665 "this checkpoint's weights leave no room for KV inside the {source}: {} weight \
3666 bytes against a {} byte budget. Serving with no derived context ceiling -- set \
3667 FRINK_DEVICE_BUDGET_BYTES if the probe is wrong, or FRINK_CB_MAX_CONTEXT to \
3668 admit on a number you choose.",
3669 priced.weights_bytes,
3670 priced.device_budget_bytes,
3671 );
3672 return batcher;
3673 };
3674 tracing::info!("{source}");
3675 tracing::info!("{}", derived.fit);
3676 let adopted = budget::apply_derived(&mut batcher, &derived);
3677 if adopted.max_context {
3678 tracing::info!(
3679 "derived per-request context ceiling: {} token positions (prompt + max_tokens); \
3680 override with FRINK_CB_MAX_CONTEXT",
3681 derived.max_context
3682 );
3683 }
3684 if adopted.kv_blocks {
3685 tracing::info!(
3686 "derived KV block budget: {} blocks x {} positions; override with FRINK_CB_KV_BLOCKS",
3687 derived.kv_blocks,
3688 batcher.kv_block_size
3689 );
3690 }
3691 if let Some(narrowed) = adopted.max_context_narrowed {
3692 tracing::info!(
3693 "per-request context ceiling narrowed to {narrowed} token positions: the whole KV ledger is {} blocks x {} positions, so a longer request could never be admitted",
3694 batcher.kv_blocks.unwrap_or_default(),
3695 batcher.kv_block_size
3696 );
3697 }
3698 batcher
3699}
3700
3701pub(crate) fn activate_loaded_model(
3710 loaded: model::LoadedModel,
3711 enable_continuous_batching: bool,
3712 path: Option<&str>,
3713 paged_kv: Option<&generate::PagedKvConfig>,
3714) -> Activated {
3715 match loaded {
3716 model::LoadedModel::Gguf(g) => {
3717 let decoder = Arc::new(g.decoder);
3718 let tokenizer = Arc::new(g.tokenizer);
3719 let config = price_batcher_config(path);
3720 let shape =
3726 frink_models::KvShape::from_config(&decoder.config, frink_models::KvElem::F32);
3727 let ceiling = Arc::new(budget::ContextCeiling::new(config.max_context, shape));
3728 let batcher = if enable_continuous_batching {
3729 tracing::info!(
3730 "continuous batching enabled: decode steps share Decoder::forward_multi_seq \
3731 (stop sequences use the same pending-buffer trim as the private generate loop)"
3732 );
3733 let tok = Arc::clone(&tokenizer);
3734 let decode = Arc::new(move |ids: &[usize]| tok.decode_bytes(ids));
3735 Some(serving::batch::ContinuousBatcher::spawn_with_ceiling(
3736 Arc::clone(&decoder),
3737 decode,
3738 config,
3739 Arc::clone(&ceiling),
3740 paged_kv.cloned(),
3741 ))
3742 } else {
3743 None
3744 };
3745 (
3746 Loaded::Generative(Arc::new(Model::Gguf(GgufModel {
3747 decoder,
3748 tokenizer,
3749 stop_tokens: g.stop_tokens,
3750 bos_id: g.bos_id,
3751 is_synthetic: g.is_synthetic,
3752 chat_template: g.chat_template,
3753 }))),
3754 batcher,
3755 Some(ceiling),
3756 )
3757 }
3758 model::LoadedModel::Kimi(k) => (
3759 Loaded::Generative(Arc::new(Model::Kimi(KimiModel {
3760 engine: k.engine,
3761 tokenizer: k.tokenizer,
3762 stop_tokens: k.stop_tokens,
3763 chat_template: k.chat_template,
3764 }))),
3765 None,
3766 None,
3767 ),
3768 model::LoadedModel::Mla(m) => (
3769 Loaded::Generative(Arc::new(Model::Mla(MlaModel {
3770 engine: m.engine,
3771 tokenizer: m.tokenizer,
3772 stop_tokens: m.stop_tokens,
3773 bos_id: m.bos_id,
3774 name: m.name,
3775 chat_template: m.chat_template,
3776 }))),
3777 None,
3778 None,
3779 ),
3780 model::LoadedModel::Gemma4(m) => (
3781 Loaded::Generative(Arc::new(Model::Gemma4(Gemma4Model {
3782 engine: m.engine,
3783 tokenizer: m.tokenizer,
3784 stop_tokens: m.stop_tokens,
3785 bos_id: m.bos_id,
3786 name: m.name,
3787 chat_template: m.chat_template,
3788 }))),
3789 None,
3790 None,
3791 ),
3792 model::LoadedModel::Glm52(g) => (
3793 Loaded::Generative(Arc::new(Model::Glm52(Glm52Model {
3794 engine: g.engine,
3795 tokenizer: g.tokenizer,
3796 stop_tokens: g.stop_tokens,
3797 bos_id: g.bos_id,
3798 name: g.name,
3799 chat_template: g.chat_template,
3800 }))),
3801 None,
3802 None,
3803 ),
3804 model::LoadedModel::Encoder(e) => (Loaded::Encoder(e), None, None),
3809 }
3810}
3811
3812struct StartupModels {
3819 loaded: model::LoadedModel,
3820 embedding: Option<Arc<frink_models::EmbeddingModel>>,
3821}
3822
3823fn continuous_batching_env() -> Option<bool> {
3824 match std::env::var("FRINK_CONTINUOUS_BATCHING")
3825 .ok()
3826 .map(|v| v.trim().to_ascii_lowercase())
3827 .as_deref()
3828 {
3829 None => None,
3830 Some("1" | "true" | "yes" | "on") => Some(true),
3831 Some("0" | "false" | "no" | "off") => Some(false),
3832 _ => None,
3833 }
3834}
3835
3836fn metal_private_decode_active() -> bool {
3837 #[cfg(feature = "metal")]
3838 {
3839 BUILT_WITH_METAL
3840 && frink_metal::attn::metal_attn_enabled()
3841 && std::env::var("FRINK_METAL").ok().as_deref() != Some("0")
3842 }
3843 #[cfg(not(feature = "metal"))]
3844 {
3845 false
3846 }
3847}
3848
3849fn continuous_batching_compatible(
3850 loaded: &model::LoadedModel,
3851 kv_pool: &Option<generate::KvPoolConfig>,
3852 prefix_cache: &Option<Arc<Mutex<PrefixCache>>>,
3853 paged_kv: &Option<generate::PagedKvConfig>,
3854) -> bool {
3855 matches!(loaded, model::LoadedModel::Gguf(_))
3856 && (paged_kv.is_some() || (kv_pool.is_none() && prefix_cache.is_none()))
3857}
3858
3859fn resolve_continuous_batching_enabled(
3860 loaded: &model::LoadedModel,
3861 kv_pool: &Option<generate::KvPoolConfig>,
3862 prefix_cache: &Option<Arc<Mutex<PrefixCache>>>,
3863 paged_kv: &Option<generate::PagedKvConfig>,
3864) -> bool {
3865 if !continuous_batching_compatible(loaded, kv_pool, prefix_cache, paged_kv) {
3866 return false;
3867 }
3868 match continuous_batching_env() {
3869 Some(true) => true,
3870 Some(false) => false,
3871 None => metal_private_decode_active(),
3872 }
3873}
3874
3875fn acquire_metal_private_decode_gate(
3876 gate: Option<&std::sync::Mutex<()>>,
3877 used_batcher: bool,
3878) -> Option<std::sync::MutexGuard<'_, ()>> {
3879 if used_batcher {
3880 None
3881 } else {
3882 gate.map(|g| g.lock().unwrap_or_else(|p| p.into_inner()))
3883 }
3884}
3885
3886fn build_app_state(
3887 models: StartupModels,
3888 kv_pool: Option<generate::KvPoolConfig>,
3889 paged_kv: Option<generate::PagedKvConfig>,
3890 prefix_cache: Option<Arc<Mutex<PrefixCache>>>,
3891 enable_continuous_batching: bool,
3892 mcp: Option<mcp::LoadedMcpConfig>,
3893 detection: Arc<health::Detection>,
3894) -> AppState {
3895 let StartupModels { loaded, embedding } = models;
3896 let configured_path = std::env::var("FRINK_MODEL_PATH").ok();
3897 let (loaded, batcher, ceiling) = activate_loaded_model(
3898 loaded,
3899 enable_continuous_batching,
3900 configured_path.as_deref(),
3901 paged_kv.as_ref(),
3902 );
3903 let id = startup_model_id();
3909 let metal_private_decode_gate = if enable_continuous_batching || !metal_private_decode_active()
3910 {
3911 None
3912 } else {
3913 tracing::info!(
3914 "Metal private-loop decode will serialize concurrent requests until \
3915 continuous batching is enabled (FRINK_CONTINUOUS_BATCHING=1 or --cont-batching)"
3916 );
3917 Some(Arc::new(std::sync::Mutex::new(())))
3918 };
3919 AppState {
3920 slept: Mutex::new(None),
3921 embedding,
3922 active: std::sync::RwLock::new(Some(Arc::new(ActiveModel {
3923 id,
3924 loaded,
3925 batcher,
3926 ceiling,
3927 checkpoint_path: configured_path.as_deref().map(PathBuf::from),
3928 }))),
3929 paged_kv,
3930 load_in_progress: std::sync::atomic::AtomicBool::new(false),
3931 tasks: Arc::new(tasks::TaskRegistry::new()),
3932 cancels: Arc::new(cancel::CancelRegistry::new()),
3933 stats: stats::Stats::new(),
3934 streams: resume::StreamRegistry::new(),
3935 model_dir: admin::model_dirs().into_iter().next(),
3936 response_cache: Mutex::new(ResponseCache::new(1000, Duration::from_secs(3600))),
3937 kv_pool,
3938 prefix_cache,
3939 sessions: session::SessionStore::new(),
3940 requests_total: std::sync::atomic::AtomicU64::new(0),
3941 request_errors_total: std::sync::atomic::AtomicU64::new(0),
3942 started_at: std::time::Instant::now(),
3943 last_request_ms: std::sync::atomic::AtomicU64::new(0),
3944 detection,
3945 mcp,
3946 continuous_batching_enabled: enable_continuous_batching,
3947 metal_private_decode_gate,
3948 loading_model: Mutex::new(None),
3949 last_load_error: Mutex::new(None),
3950 serving: Mutex::new(crate::stats::ServingStats::default()),
3951 maintenance: Mutex::new(crate::policy::maintenance::MaintenanceGate::serving()),
3952 footprint: Mutex::new(crate::policy::footprint::ProbeCache::new(FOOTPRINT_TTL_MS)),
3953 started_unix: unix_now(),
3954 }
3955}
3956
3957fn load_embedding_model() -> anyhow::Result<Option<Arc<frink_models::EmbeddingModel>>> {
3965 let Ok(path) = std::env::var("FRINK_EMBEDDING_MODEL_PATH") else {
3966 return Ok(None);
3967 };
3968 let model = frink_models::EmbeddingModel::from_gguf_path(&path)
3969 .map_err(|e| anyhow::anyhow!("FRINK_EMBEDDING_MODEL_PATH={path}: {e}"))?;
3970 tracing::info!(
3971 "loaded embedding model '{}' ({}, {} dims, pooling {}, max {} tokens)",
3972 model.name(),
3973 model.architecture(),
3974 model.n_embd(),
3975 model.pooling_type().name(),
3976 model.n_ctx_train(),
3977 );
3978 Ok(Some(Arc::new(model)))
3979}
3980
3981fn unix_now() -> u64 {
3986 std::time::SystemTime::now()
3987 .duration_since(std::time::UNIX_EPOCH)
3988 .map(|d| d.as_secs())
3989 .unwrap_or(0)
3990}
3991
3992fn startup_model_id() -> Option<String> {
3997 let configured = std::env::var("FRINK_MODEL_PATH").ok()?;
3998 let configured = std::fs::canonicalize(&configured).ok()?;
3999 admin::discover(&admin::model_dirs())
4000 .into_iter()
4001 .find(|d| {
4002 std::fs::canonicalize(&d.path)
4003 .map(|p| p == configured)
4004 .unwrap_or(false)
4005 })
4006 .map(|d| d.id)
4007}
4008
4009fn init_cpu_pool() {
4018 match frink_core::threads::init_cpu_pool() {
4019 Some(n) => eprintln!(
4020 "frink-server: rayon pool {n} threads (perf cores {}; override with FRINK_CPU_THREADS)",
4021 frink_core::threads::perf_core_count()
4022 ),
4023 None => eprintln!("frink-server: global rayon pool already built; leaving it alone"),
4024 }
4025}
4026
4027fn announce_ready(addr: SocketAddr, scheme: &str) {
4040 use std::io::Write;
4041 let ready =
4042 frink_api::ServerReady::new(addr, scheme, env!("CARGO_PKG_VERSION"), std::process::id());
4043 let mut stdout = std::io::stdout().lock();
4044 let _ = writeln!(stdout, "{}", ready.to_line());
4045 let _ = stdout.flush();
4046}
4047
4048async fn shutdown_signal(exit_on_stdin_close: bool) {
4060 if !exit_on_stdin_close {
4061 std::future::pending::<()>().await;
4062 return;
4063 }
4064 let _ = tokio::task::spawn_blocking(|| {
4065 use std::io::Read;
4066 let mut sink = [0u8; 256];
4067 let mut stdin = std::io::stdin().lock();
4068 loop {
4069 match stdin.read(&mut sink) {
4070 Ok(0) => break,
4072 Ok(_) => continue,
4074 Err(e) => {
4075 tracing::warn!("stdin read failed ({e}); treating it as closed");
4076 break;
4077 }
4078 }
4079 }
4080 })
4081 .await;
4082 tracing::info!("stdin closed; shutting down");
4083}
4084
4085fn tokio_worker_threads() -> usize {
4090 std::env::var("FRINK_TOKIO_WORKERS")
4091 .ok()
4092 .and_then(|v| v.trim().parse::<usize>().ok())
4093 .filter(|n| *n > 0)
4094 .unwrap_or(2)
4095}
4096
4097fn install_ring_crypto_provider() {
4114 let _ = rustls::crypto::ring::default_provider().install_default();
4115}
4116
4117pub fn run_server(args: ServerArgs) -> anyhow::Result<()> {
4123 if args.list_devices {
4124 frink_models::devices::print_available_devices();
4125 return Ok(());
4126 }
4127 apply_cli_overrides(&args)?;
4128
4129 let _instance = {
4133 use frink_core::instance::{register, InstancePolicy};
4134 let policy = if args.allow_multiple_instances {
4135 InstancePolicy::Multi
4136 } else {
4137 InstancePolicy::from_env_or(InstancePolicy::Single)
4138 };
4139 let model = std::env::var("FRINK_MODEL_PATH").ok();
4140 register(
4141 "server",
4142 model.as_deref(),
4143 frink_core::instance::current_backend(),
4144 policy,
4145 )
4146 .map_err(|conflict| anyhow::anyhow!("{conflict}"))?
4147 };
4148
4149 let journal = journal::Journal::from_env();
4150 eprintln!(
4151 "frink-server: process lifecycle journal at {:?} (override with FRINK_JOURNAL_PATH)",
4152 journal.path()
4153 );
4154 journal.append(&journal::Record::session_start(
4155 env!("CARGO_PKG_VERSION"),
4156 std::process::id(),
4157 ));
4158 journal::install_panic_hook(journal.clone());
4159
4160 let mcp_config_path = args.mcp_config.clone();
4161 let exit_on_stdin_close = args.exit_on_stdin_close
4162 || std::env::var("FRINK_EXIT_ON_STDIN_CLOSE")
4163 .map(|v| v == "1")
4164 .unwrap_or(false);
4165
4166 unsafe { frink_core::weight_matrix::default_cpu_int_dot_on() };
4170 init_cpu_pool();
4171
4172 let runtime = tokio::runtime::Builder::new_multi_thread()
4173 .worker_threads(tokio_worker_threads())
4174 .enable_all()
4175 .build()?;
4176 let result = runtime.block_on(run(mcp_config_path, exit_on_stdin_close));
4177
4178 let reason = match &result {
4179 Ok(()) => "normal".to_string(),
4180 Err(e) => e.to_string(),
4181 };
4182 journal.append(&journal::Record::session_exit(reason));
4183
4184 runtime.shutdown_background();
4189
4190 result
4191}
4192
4193async fn run(mcp_config_path: Option<PathBuf>, exit_on_stdin_close: bool) -> anyhow::Result<()> {
4194 let _ = tracing_subscriber::fmt::try_init();
4200
4201 let addr = std::env::var("FRINK_ADDR").unwrap_or_else(|_| "127.0.0.1:8383".to_string());
4211 let api_key_configured = std::env::var("FRINK_API_KEY").is_ok();
4212 let allow_unauthenticated_remote = std::env::var("FRINK_ALLOW_UNAUTHENTICATED_REMOTE")
4213 .map(|v| v == "1")
4214 .unwrap_or(false);
4215 if let Err(msg) =
4216 security::check_bind_authorization(&addr, api_key_configured, allow_unauthenticated_remote)
4217 {
4218 anyhow::bail!(msg);
4219 }
4220
4221 let embedding_model = load_embedding_model()?;
4228
4229 let mut loaded = model::load()?;
4230 match &loaded {
4231 model::LoadedModel::Gguf(g) => tracing::info!(
4232 "loaded GGUF model '{}' (synthetic={}, tokenizer={})",
4233 g.decoder.config.name,
4234 g.is_synthetic,
4235 g.tokenizer.kind()
4236 ),
4237 model::LoadedModel::Kimi(k) => tracing::info!(
4238 "loaded Kimi K3 checkpoint (tokenizer={} base tokens)",
4239 k.tokenizer.vocab_size()
4240 ),
4241 model::LoadedModel::Mla(m) => tracing::info!(
4242 "loaded MLA GGUF '{}' (tokenizer={})",
4243 m.name,
4244 m.tokenizer.kind()
4245 ),
4246 model::LoadedModel::Gemma4(m) => tracing::info!(
4247 "loaded Gemma4 GGUF '{}' (tokenizer={})",
4248 m.name,
4249 m.tokenizer.kind()
4250 ),
4251 model::LoadedModel::Glm52(g) => tracing::info!(
4252 "loaded GLM-5.2 GGUF '{}' (tokenizer={})",
4253 g.name,
4254 g.tokenizer.kind()
4255 ),
4256 model::LoadedModel::Encoder(_) => {}
4259 }
4260 let metal_default_moe_budget = {
4268 #[cfg(feature = "metal")]
4269 {
4270 frink_core::metal_dense_enabled()
4271 && std::env::var("FRINK_GPU_VRAM_BUDGET_BYTES").is_err()
4272 }
4273 #[cfg(not(feature = "metal"))]
4274 {
4275 false
4276 }
4277 };
4278 if let Ok(budget_str) = std::env::var("FRINK_GPU_VRAM_BUDGET_BYTES") {
4279 let budget: u64 = budget_str
4280 .parse()
4281 .expect("FRINK_GPU_VRAM_BUDGET_BYTES must be a non-negative integer");
4282 match &mut loaded {
4283 model::LoadedModel::Gguf(g) => {
4284 tracing::info!(
4285 "GPU expert placement enabled: {budget} byte VRAM budget for routed experts \
4286 (CUDA and/or Metal matvecs when built with the matching feature)"
4287 );
4288 g.decoder.gpu_vram_budget_bytes = Some(budget);
4289 }
4290 model::LoadedModel::Kimi(_) => {
4291 tracing::warn!(
4292 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is Kimi K3 -- not \
4293 supported yet (its MoE stack isn't wired to PlacementPlan), ignoring"
4294 );
4295 }
4296 model::LoadedModel::Mla(_) => {
4297 tracing::warn!(
4298 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is MLA -- dense \
4299 FFN path only today; ignoring expert VRAM budget"
4300 );
4301 }
4302 model::LoadedModel::Gemma4(_) => {
4303 tracing::warn!(
4304 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is Gemma4 -- \
4305 ignoring expert VRAM budget"
4306 );
4307 }
4308 model::LoadedModel::Glm52(_) => {
4309 tracing::warn!(
4310 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is GLM-5.2 DSA -- \
4311 GPU expert placement not wired yet; ignoring"
4312 );
4313 }
4314 model::LoadedModel::Encoder(_) => {
4315 tracing::warn!(
4316 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is an encoder -- \
4317 it has no routed experts to place; ignoring"
4318 );
4319 }
4320 }
4321 } else if metal_default_moe_budget {
4322 const METAL_DEFAULT_MOE_BUDGET: u64 = 64 * 1024 * 1024 * 1024;
4326 if let model::LoadedModel::Gguf(g) = &mut loaded {
4327 tracing::info!(
4328 "Metal MoE expert placement default-on ({METAL_DEFAULT_MOE_BUDGET} byte budget); \
4329 set FRINK_GPU_VRAM_BUDGET_BYTES=0 to force CPU experts"
4330 );
4331 g.decoder.gpu_vram_budget_bytes = Some(METAL_DEFAULT_MOE_BUDGET);
4332 }
4333 }
4334 #[cfg(feature = "cuda")]
4335 {
4336 if frink_core::cuda_dense_enabled() {
4337 tracing::info!(
4338 "CUDA dense matvec enabled for WeightMatrix::apply \
4339 (FRINK_CUDA=0|cpu forces CPU; weight buffers stay resident after first upload)"
4340 );
4341 } else {
4342 tracing::info!(
4343 "CUDA dense matvec disabled (FRINK_CUDA); dense decode uses CPU or Metal"
4344 );
4345 }
4346 }
4347 #[cfg(feature = "metal")]
4348 {
4349 if frink_core::metal_dense_enabled() {
4350 tracing::info!(
4351 "Metal dense matvec enabled for WeightMatrix::apply \
4352 (FRINK_METAL=0|cpu forces CPU; weight buffers stay resident after first upload)"
4353 );
4354 match std::env::var("FRINK_METAL_ATTN").ok().as_deref() {
4355 Some("1") | Some("true") | Some("on") | Some("attn") => {
4356 tracing::info!(
4357 "Metal fused attention requested (FRINK_METAL_ATTN): \
4358 QKV→RoPE→GQA→O on-GPU for Norm/NeoX decode without QKV bias/QK-norm"
4359 );
4360 }
4361 _ => {}
4362 }
4363 tracing::info!(
4364 "Metal greedy GPU argmax: temperature<=0 folds \
4365 final_norm+lm_head+argmax into the dense stack"
4366 );
4367 } else {
4368 tracing::info!("Metal dense matvec disabled (FRINK_METAL); dense decode uses CPU");
4369 }
4370 }
4371 let blocks_env = std::env::var("FRINK_KV_POOL_BLOCKS");
4382 let block_size_env = std::env::var("FRINK_KV_POOL_BLOCK_SIZE");
4383 let byte_budget_env = std::env::var("FRINK_KV_BYTE_BUDGET");
4384 if blocks_env.is_ok() && byte_budget_env.is_ok() {
4385 panic!(
4386 "FRINK_KV_POOL_BLOCKS and FRINK_KV_BYTE_BUDGET are mutually exclusive \
4387 (set one block-count source plus FRINK_KV_POOL_BLOCK_SIZE, or neither to disable)"
4388 );
4389 }
4390 let kv_pool = match (blocks_env, block_size_env, byte_budget_env) {
4391 (Ok(blocks), Ok(block_size), Err(_)) => {
4392 let total_blocks: usize = blocks
4393 .parse()
4394 .expect("FRINK_KV_POOL_BLOCKS must be a positive integer");
4395 let block_size: usize = block_size
4396 .parse()
4397 .expect("FRINK_KV_POOL_BLOCK_SIZE must be a positive integer");
4398 let queue_wait_ms: u64 = std::env::var("FRINK_KV_POOL_QUEUE_TIMEOUT_MS")
4404 .ok()
4405 .map(|v| {
4406 v.parse()
4407 .expect("FRINK_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4408 })
4409 .unwrap_or(0);
4410 tracing::info!(
4411 "KV cache block pool enabled: {total_blocks} blocks x {block_size} positions \
4412 each, shared across all concurrent requests, {queue_wait_ms}ms admission queue wait"
4413 );
4414 Some(generate::KvPoolConfig {
4415 pool: Arc::new(Mutex::new(KvBlockPool::new(block_size, total_blocks))),
4416 queue_wait: Duration::from_millis(queue_wait_ms),
4417 })
4418 }
4419 (Err(_), Ok(block_size), Ok(byte_budget)) => {
4420 let block_size: usize = block_size
4421 .parse()
4422 .expect("FRINK_KV_POOL_BLOCK_SIZE must be a positive integer");
4423 let budget: u64 = byte_budget
4424 .parse()
4425 .expect("FRINK_KV_BYTE_BUDGET must be a positive integer");
4426 let cfg = match &loaded {
4427 model::LoadedModel::Gguf(g) => &g.decoder.config,
4428 model::LoadedModel::Kimi(_)
4429 | model::LoadedModel::Mla(_)
4430 | model::LoadedModel::Gemma4(_)
4431 | model::LoadedModel::Glm52(_)
4432 | model::LoadedModel::Encoder(_) => {
4433 panic!(
4434 "FRINK_KV_BYTE_BUDGET requires a GGUF decoder model \
4435 (set FRINK_MODEL_PATH to a generic-decoder .gguf file)"
4436 );
4437 }
4438 };
4439 let bytes_per_block = block_size
4440 * cfg.kv_heads_all_layers()
4441 * (cfg.head_dim + cfg.v_head_dim())
4442 * std::mem::size_of::<f32>();
4443 assert!(
4444 bytes_per_block > 0,
4445 "derived KV block byte size must be positive (check model config and block size)"
4446 );
4447 let total_blocks = (budget as usize / bytes_per_block).max(1);
4448 let queue_wait_ms: u64 = std::env::var("FRINK_KV_POOL_QUEUE_TIMEOUT_MS")
4449 .ok()
4450 .map(|v| {
4451 v.parse()
4452 .expect("FRINK_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4453 })
4454 .unwrap_or(0);
4455 tracing::info!(
4456 "KV cache block pool enabled from byte budget: {budget} bytes / \
4457 {bytes_per_block} bytes per block ({block_size} positions x {} layers) -> \
4458 {total_blocks} blocks, {queue_wait_ms}ms admission queue wait",
4459 cfg.n_layers
4460 );
4461 Some(generate::KvPoolConfig {
4462 pool: Arc::new(Mutex::new(KvBlockPool::new(block_size, total_blocks))),
4463 queue_wait: Duration::from_millis(queue_wait_ms),
4464 })
4465 }
4466 (Err(_), Err(_), Err(_)) => None,
4467 (Err(_), Ok(_), Err(_)) => panic!(
4468 "FRINK_KV_POOL_BLOCK_SIZE requires FRINK_KV_POOL_BLOCKS or FRINK_KV_BYTE_BUDGET \
4469 (or unset all three to disable KV cache pooling)"
4470 ),
4471 (Ok(_), Ok(_), Ok(_)) => {
4472 unreachable!("FRINK_KV_POOL_BLOCKS and FRINK_KV_BYTE_BUDGET are mutually exclusive")
4473 }
4474 (Ok(_), Err(_), _) | (Err(_), Err(_), Ok(_)) => panic!(
4475 "FRINK_KV_POOL_BLOCKS/FRINK_KV_BYTE_BUDGET and FRINK_KV_POOL_BLOCK_SIZE must be \
4476 set together (or neither, to disable KV cache pooling)"
4477 ),
4478 };
4479 let paged_kv = match (
4485 std::env::var("FRINK_PAGED_KV_BLOCKS"),
4486 std::env::var("FRINK_PAGED_KV_BLOCK_SIZE"),
4487 ) {
4488 (Ok(blocks), Ok(block_size)) => {
4489 assert!(
4490 kv_pool.is_none(),
4491 "FRINK_PAGED_KV_BLOCKS and FRINK_KV_POOL_BLOCKS/FRINK_KV_BYTE_BUDGET are \
4492 mutually exclusive: both bound the same KV memory, by different means. \
4493 Set one."
4494 );
4495 let blocks_per_layer: usize = blocks
4509 .parse()
4510 .expect("FRINK_PAGED_KV_BLOCKS must be a positive integer");
4511 let block_size: usize = block_size
4512 .parse()
4513 .expect("FRINK_PAGED_KV_BLOCK_SIZE must be a positive integer");
4514 let gguf = match &loaded {
4515 model::LoadedModel::Gguf(g) => g,
4516 _ => panic!(
4517 "FRINK_PAGED_KV_BLOCKS requires a GGUF decoder model \
4518 (set FRINK_MODEL_PATH to a generic-decoder .gguf file)"
4519 ),
4520 };
4521 let cfg = &gguf.decoder.config;
4522 let queue_wait_ms: u64 = std::env::var("FRINK_KV_POOL_QUEUE_TIMEOUT_MS")
4523 .ok()
4524 .map(|v| {
4525 v.parse()
4526 .expect("FRINK_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4527 })
4528 .unwrap_or(0);
4529 tracing::info!(
4530 "Paged KV enabled: {blocks_per_layer} blocks x {block_size} positions per \
4531 layer across {} layers, shared by all concurrent requests, \
4532 {queue_wait_ms}ms admission queue wait",
4533 cfg.n_layers
4534 );
4535 let radix = Some(Arc::new(Mutex::new(crate::policy::radix::RadixCache::new(
4540 block_size,
4541 ))));
4542 let anchor_token = crate::policy::anchor::resolve_anchor_token(
4551 crate::policy::parser::ToolCallFormat::infer(
4552 &std::env::var("FRINK_MODEL_PATH").unwrap_or_default(),
4553 )
4554 .opener(),
4555 |text| {
4556 gguf.tokenizer
4557 .encode(text, SpecialTokens::Parse)
4558 .into_iter()
4559 .map(|t| t as u32)
4560 .collect()
4561 },
4562 );
4563 if let Some(id) = anchor_token {
4564 tracing::info!(
4565 "Paged KV window slide: tool-call anchor is token {id}, so a turn's \
4566 window stops short of where its next turn rejoins"
4567 );
4568 }
4569 let slide_interval: usize = std::env::var("FRINK_PAGED_KV_SLIDE_INTERVAL")
4570 .ok()
4571 .map(|v| {
4572 v.parse()
4573 .expect("FRINK_PAGED_KV_SLIDE_INTERVAL must be a positive integer")
4574 })
4575 .unwrap_or(crate::policy::pool_budget::DEFAULT_SWA_EVICTION_INTERVAL);
4576 if let Some(window) = cfg.uniform_sliding_window() {
4577 tracing::info!(
4578 "Paged KV window slide enabled: every layer slides by {window} every \
4579 {slide_interval} decode steps, so a request holds its prompt and a \
4580 window rather than its whole context"
4581 );
4582 } else if cfg.kv_block_window().is_some() {
4583 tracing::info!(
4584 "Paged KV window slide NOT enabled: this model has full-attention layers, \
4585 and a page group holds one block in every layer"
4586 );
4587 }
4588 Some(generate::PagedKvConfig {
4589 store: Arc::new(cfg.new_paged_kv(block_size, blocks_per_layer)),
4592 queue_wait: Duration::from_millis(queue_wait_ms),
4593 radix,
4594 anchor_token,
4595 slide_interval,
4596 })
4597 }
4598 (Err(_), Err(_)) => None,
4599 _ => panic!(
4600 "FRINK_PAGED_KV_BLOCKS and FRINK_PAGED_KV_BLOCK_SIZE must be set together \
4601 (or neither, to disable paged KV)"
4602 ),
4603 };
4604 let prefix_cache = std::env::var("FRINK_PREFIX_CACHE_ENTRIES").ok().map(|v| {
4611 let max_entries: usize = v
4612 .parse()
4613 .expect("FRINK_PREFIX_CACHE_ENTRIES must be a positive integer");
4614 if kv_pool.is_some() {
4615 tracing::warn!(
4616 "FRINK_PREFIX_CACHE_ENTRIES is set but so is the KV pool -- prefix \
4617 caching will never be consulted while a KV pool is configured"
4618 );
4619 }
4620 assert!(
4627 paged_kv.is_none(),
4628 "FRINK_PREFIX_CACHE_ENTRIES and FRINK_PAGED_KV_BLOCKS are mutually exclusive: \
4629 the prefix cache stores contiguous KV snapshots, which a paged request does not \
4630 produce, so the cache could never hit. Set one."
4631 );
4632 tracing::info!(
4633 "KV-prefix cache enabled: up to {max_entries} stored prefixes, shared across \
4634 all requests"
4635 );
4636 Arc::new(Mutex::new(PrefixCache::new(max_entries)))
4637 });
4638 if matches!(
4639 loaded,
4640 model::LoadedModel::Kimi(_) | model::LoadedModel::Mla(_) | model::LoadedModel::Glm52(_)
4641 ) && (kv_pool.is_some() || prefix_cache.is_some())
4642 {
4643 tracing::warn!(
4644 "KV pool / prefix cache are configured but the loaded model is Kimi, MLA, or GLM-5.2 -- \
4645 neither is consulted for those engines (state shapes differ from Decoder KV); see \
4646 frink_models::engine's module docs"
4647 );
4648 }
4649 let enable_cb =
4650 resolve_continuous_batching_enabled(&loaded, &kv_pool, &prefix_cache, &paged_kv);
4651 if enable_cb && continuous_batching_env().is_none() && metal_private_decode_active() {
4652 tracing::info!(
4653 "continuous batching enabled by default on Metal for safe parallel serving \
4654 (set FRINK_CONTINUOUS_BATCHING=0 or --no-cont-batching to use the private path)"
4655 );
4656 }
4657 if continuous_batching_env() == Some(true)
4658 && !continuous_batching_compatible(&loaded, &kv_pool, &prefix_cache, &paged_kv)
4659 && (kv_pool.is_some() || prefix_cache.is_some())
4660 {
4661 tracing::warn!(
4662 "FRINK_CONTINUOUS_BATCHING=1 ignored while KV pool or prefix cache is configured \
4663 (those modes keep the private generate path)"
4664 );
4665 }
4666 if let Ok(n) = std::env::var("FRINK_CHUNKED_PREFILL") {
4667 if let Ok(chunk) = n.parse::<usize>() {
4668 if chunk > 0 {
4669 tracing::info!("chunked prefill enabled: {chunk} tokens per forward_batch chunk");
4670 }
4671 }
4672 }
4673 if matches!(
4674 std::env::var("FRINK_CPU_KV_OFFLOAD").ok().as_deref(),
4675 Some("1")
4676 ) {
4677 tracing::warn!(
4678 "FRINK_CPU_KV_OFFLOAD=1: syncing Metal KV to host after each decode step \
4679 (minimal spill; full layer offload still planned)"
4680 );
4681 }
4682
4683 let mcp = match mcp_config_path {
4684 Some(path) => {
4685 let loaded = mcp::load_mcp_config(&path)?;
4686 tracing::info!(
4687 "MCP config loaded from {} ({} server(s); invocation not wired yet)",
4688 loaded.path,
4689 loaded.servers.len()
4690 );
4691 Some(loaded)
4692 }
4693 None => None,
4694 };
4695
4696 let detection = health::Detection::spawn();
4700
4701 let state = Arc::new(build_app_state(
4702 StartupModels {
4703 loaded,
4704 embedding: embedding_model,
4705 },
4706 kv_pool,
4707 paged_kv,
4708 prefix_cache,
4709 enable_cb,
4710 mcp,
4711 detection,
4712 ));
4713
4714 use frink_api::routes;
4718
4719 let public = Router::new().route(routes::HEALTH, get(health));
4724
4725 let mut protected = protected_routes();
4726
4727 if let Ok(key) = std::env::var("FRINK_API_KEY") {
4732 tracing::info!("API key auth enabled");
4733 let auth = limits::AuthConfig {
4734 api_key: Arc::new(key),
4735 };
4736 protected = protected.route_layer(axum::middleware::from_fn_with_state(
4737 auth,
4738 limits::require_api_key,
4739 ));
4740 }
4741 if let Ok(rpm) = std::env::var("FRINK_RATE_LIMIT_PER_MINUTE") {
4742 let rpm: u32 = rpm
4743 .parse()
4744 .expect("FRINK_RATE_LIMIT_PER_MINUTE must be a positive integer");
4745 tracing::info!("rate limiting enabled: {rpm} requests/minute (global)");
4746 let limiter = Arc::new(limits::RateLimiter::per_minute(rpm));
4747 protected = protected.route_layer(axum::middleware::from_fn_with_state(
4748 limiter,
4749 limits::rate_limit,
4750 ));
4751 }
4752 if let Ok(spec) = std::env::var("FRINK_CORS_ORIGINS") {
4760 let origins = security::parse_cors_origins(&spec)
4761 .unwrap_or_else(|e| panic!("FRINK_CORS_ORIGINS: {e}"));
4762 tracing::info!(
4763 "CORS enabled: {} allow-listed origin(s) ({})",
4764 origins.len(),
4765 spec
4766 );
4767 let cors = tower_http::cors::CorsLayer::new()
4768 .allow_origin(tower_http::cors::AllowOrigin::list(origins))
4769 .allow_methods([axum::http::Method::GET, axum::http::Method::POST])
4770 .allow_headers([
4771 axum::http::header::CONTENT_TYPE,
4772 axum::http::header::AUTHORIZATION,
4773 axum::http::HeaderName::from_static(attribution::CLIENT_HEADER),
4779 axum::http::HeaderName::from_static("last-event-id"),
4783 ]);
4784 protected = protected.route_layer(cors);
4785 }
4786
4787 let app = public
4791 .merge(protected)
4792 .layer(axum::middleware::from_fn(limits::retry_after))
4793 .with_state(state);
4794
4795 let tls_paths = security::tls_paths_from_env().unwrap_or_else(|e| panic!("{e}"));
4801 install_ring_crypto_provider();
4802 match tls_paths {
4807 Some(paths) => {
4808 let config =
4809 axum_server::tls_rustls::RustlsConfig::from_pem_file(&paths.cert, &paths.key)
4810 .await
4811 .map_err(|e| {
4812 anyhow::anyhow!(
4813 "failed to load TLS cert/key ({:?}, {:?}): {e}",
4814 paths.cert,
4815 paths.key
4816 )
4817 })?;
4818 let socket_addr: std::net::SocketAddr = addr
4819 .parse()
4820 .map_err(|e| anyhow::anyhow!("invalid FRINK_ADDR {addr:?} for TLS: {e}"))?;
4821 let listener = std::net::TcpListener::bind(socket_addr)?;
4822 listener.set_nonblocking(true)?;
4830 let bound = listener.local_addr()?;
4831 tracing::info!("TLS enabled: frink-server listening on https://{bound}");
4832 announce_ready(bound, "https");
4833
4834 let handle = axum_server::Handle::new();
4835 let shutdown_handle = handle.clone();
4836 tokio::spawn(async move {
4837 shutdown_signal(exit_on_stdin_close).await;
4838 shutdown_handle.graceful_shutdown(Some(Duration::from_secs(5)));
4839 });
4840 axum_server::from_tcp_rustls(listener, config)?
4841 .handle(handle)
4842 .serve(app.into_make_service())
4843 .await?;
4844 }
4845 None => {
4846 let listener = tokio::net::TcpListener::bind(&addr).await?;
4847 let bound = listener.local_addr()?;
4848 tracing::info!("frink-server listening on {bound}");
4849 announce_ready(bound, "http");
4850 axum::serve(listener, app)
4851 .with_graceful_shutdown(shutdown_signal(exit_on_stdin_close))
4852 .await?;
4853 }
4854 }
4855 Ok(())
4856}
4857
4858#[cfg(test)]
4859pub(crate) mod tests {
4860 use super::*;
4861 use frink_models::config::test_dense_fixture;
4862
4863 #[test]
4864 fn the_ready_line_round_trips_through_a_parent_reading_stdout() {
4865 let addr: SocketAddr = "127.0.0.1:51999".parse().unwrap();
4866 let ready = frink_api::ServerReady::new(addr, "http", "0.5.0", std::process::id());
4867 let parsed = frink_api::ServerReady::from_line(&ready.to_line()).unwrap();
4868 assert_eq!(parsed.port, 51999);
4869 assert_eq!(parsed.base_url(), "http://127.0.0.1:51999");
4870 assert!(frink_api::ServerReady::from_line("INFO frink-server listening").is_none());
4872 }
4873
4874 fn test_model() -> Model {
4875 let cfg = test_dense_fixture();
4879 Model::Gguf(GgufModel {
4880 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 32)),
4881 tokenizer: Arc::new(ServerTokenizer::Byte),
4882 stop_tokens: StopTokens::default(),
4883 bos_id: None,
4884 is_synthetic: true,
4885 chat_template: chat_template::PromptTemplate::plain(),
4886 })
4887 }
4888
4889 fn greedy_params(max_tokens: usize) -> GenerationParams {
4890 GenerationParams {
4891 cache_salt: None,
4892 prompt_logprobs: None,
4893 wants_logprobs: false,
4894 n: 1,
4895 interleave_choices: false,
4896 reasoning: None,
4897 max_tokens,
4898 sampling: SamplingParams::default(),
4899 seed: 1,
4900 stop: Vec::new(),
4901 stop_token_ids: Vec::new(),
4902 json_object: false,
4903 grammar: None,
4904 cancel: None,
4905 ignore_eos: false,
4906 reasoning_budget: crate::reasoning_budget::ReasoningBudget::Unrestricted,
4907 lora: None,
4908 }
4909 }
4910
4911 fn test_model_full_byte_vocab() -> Model {
4915 test_model_full_byte_vocab_with_eos(None)
4916 }
4917
4918 fn test_model_full_byte_vocab_with_eos(eos: Option<usize>) -> Model {
4925 let mut cfg = test_dense_fixture();
4926 cfg.vocab_size = 256;
4927 Model::Gguf(GgufModel {
4928 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
4929 tokenizer: Arc::new(ServerTokenizer::Byte),
4930 stop_tokens: StopTokens::from_eos(eos),
4931 bos_id: None,
4932 is_synthetic: true,
4933 chat_template: chat_template::PromptTemplate::plain(),
4934 })
4935 }
4936
4937 pub(crate) fn test_state(model: Model, response_cache: ResponseCache) -> AppState {
4941 test_state_at(model, response_cache, None)
4942 }
4943
4944 pub(crate) fn test_state_at(
4948 model: Model,
4949 response_cache: ResponseCache,
4950 checkpoint_path: Option<std::path::PathBuf>,
4951 ) -> AppState {
4952 AppState {
4953 slept: Mutex::new(None),
4954 embedding: None,
4955 paged_kv: None,
4956 active: std::sync::RwLock::new(Some(Arc::new(ActiveModel {
4957 id: None,
4958 loaded: Loaded::Generative(Arc::new(model)),
4959 batcher: None,
4960 ceiling: None,
4961 checkpoint_path,
4962 }))),
4963 load_in_progress: std::sync::atomic::AtomicBool::new(false),
4964 tasks: Arc::new(tasks::TaskRegistry::new()),
4965 cancels: Arc::new(cancel::CancelRegistry::new()),
4966 stats: stats::Stats::new(),
4967 streams: resume::StreamRegistry::new(),
4968 model_dir: None,
4969 response_cache: Mutex::new(response_cache),
4970 kv_pool: None,
4971 prefix_cache: None,
4972 sessions: session::SessionStore::new(),
4973 requests_total: std::sync::atomic::AtomicU64::new(0),
4974 request_errors_total: std::sync::atomic::AtomicU64::new(0),
4975 started_at: std::time::Instant::now(),
4976 last_request_ms: std::sync::atomic::AtomicU64::new(0),
4977 detection: Arc::new(health::Detection::ready(health::probe_backends())),
4978 mcp: None,
4979 continuous_batching_enabled: false,
4980 metal_private_decode_gate: None,
4981 loading_model: Mutex::new(None),
4982 last_load_error: Mutex::new(None),
4983 serving: Mutex::new(crate::stats::ServingStats::default()),
4984 maintenance: Mutex::new(crate::policy::maintenance::MaintenanceGate::serving()),
4985 footprint: Mutex::new(crate::policy::footprint::ProbeCache::new(FOOTPRINT_TTL_MS)),
4986 started_unix: unix_now(),
4987 }
4988 }
4989
4990 pub(crate) fn test_app() -> Router {
4999 test_app_with_state(Arc::new(test_state(
5000 test_model_full_byte_vocab(),
5001 ResponseCache::new(1000, Duration::from_secs(3600)),
5002 )))
5003 }
5004
5005 pub(crate) fn test_app_with_state(state: Arc<AppState>) -> Router {
5008 Router::new()
5018 .route(frink_api::routes::HEALTH, get(health))
5019 .merge(protected_routes())
5020 .with_state(state)
5021 }
5022
5023 fn named_test_model(name: &'static str, vocab_size: usize) -> Model {
5024 let mut cfg = test_dense_fixture();
5025 cfg.name = name;
5026 cfg.vocab_size = vocab_size;
5027 Model::Gguf(GgufModel {
5028 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5029 tokenizer: Arc::new(ServerTokenizer::Byte),
5030 stop_tokens: StopTokens::default(),
5031 bos_id: None,
5032 is_synthetic: true,
5033 chat_template: chat_template::PromptTemplate::plain(),
5034 })
5035 }
5036
5037 fn model_with_template(name: &'static str, source: &str) -> Model {
5041 let mut cfg = test_dense_fixture();
5042 cfg.name = name;
5043 cfg.vocab_size = 256;
5044 Model::Gguf(GgufModel {
5045 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5046 tokenizer: Arc::new(ServerTokenizer::Byte),
5047 stop_tokens: StopTokens::default(),
5048 bos_id: None,
5049 is_synthetic: true,
5050 chat_template: chat_template::PromptTemplate::from_gguf_metadata(
5051 Some(source),
5052 Some("qwen3"),
5053 false,
5054 true,
5055 None,
5056 None,
5057 ),
5058 })
5059 }
5060
5061 #[tokio::test]
5070 async fn a_template_that_rejects_the_conversation_is_a_400_on_the_streaming_path() {
5071 let strict = "{% if messages | length > 1 %}\
5074 {{ raise_exception('this template takes one turn') }}\
5075 {% endif %}{{ messages[0].content }}";
5076 let state = Arc::new(test_state(
5077 model_with_template("strict", strict),
5078 ResponseCache::new(4, Duration::from_secs(60)),
5079 ));
5080 let app = test_app_with_state(state);
5081
5082 let (status, body) = post_json_uri(
5083 &app,
5084 "/v1/chat/completions",
5085 serde_json::json!({
5086 "model": "strict",
5087 "stream": true,
5088 "messages": [
5089 {"role": "user", "content": "one"},
5090 {"role": "user", "content": "two"},
5091 ],
5092 }),
5093 )
5094 .await;
5095 assert_eq!(status, StatusCode::BAD_REQUEST);
5096 assert_eq!(body["error"]["param"], serde_json::json!("messages"));
5097 assert!(
5098 body["error"]["message"]
5099 .as_str()
5100 .unwrap()
5101 .contains("one turn"),
5102 "the template's own message must reach the caller: {body}"
5103 );
5104
5105 let (status, _) = post_json_uri(
5107 &app,
5108 "/v1/chat/completions",
5109 serde_json::json!({
5110 "model": "strict",
5111 "stream": true,
5112 "max_tokens": 1,
5113 "messages": [{"role": "user", "content": "one"}],
5114 }),
5115 )
5116 .await;
5117 assert_eq!(status, StatusCode::OK);
5118 }
5119
5120 #[tokio::test]
5122 async fn models_advertises_the_gears_this_checkpoint_actually_has() {
5123 let reasoning = "{% if enable_thinking %}<think>{% endif %}\
5124 {% if reasoning_effort %}\
5125 {% if reasoning_effort not in ['low','medium','high'] %}\
5126 {{ raise_exception('bad effort') }}\
5127 {% endif %}[{{ reasoning_effort }}]\
5128 {% endif %}{{ messages[0].content }}";
5129 let state = Arc::new(test_state(
5130 model_with_template("thinker", reasoning),
5131 ResponseCache::new(4, Duration::from_secs(60)),
5132 ));
5133 let app = test_app_with_state(state);
5134 let (status, models) = get_json(&app, frink_api::routes::V1_MODELS).await;
5135 assert_eq!(status, StatusCode::OK);
5136 let entry = &models["data"][0];
5137 assert_eq!(
5138 entry["supported_reasoning_efforts"],
5139 serde_json::json!(["off", "low", "medium", "high"])
5140 );
5141 assert_eq!(entry["default_reasoning_effort"], serde_json::json!("off"));
5142 }
5143
5144 #[tokio::test]
5149 async fn a_checkpoint_with_no_thinking_controls_advertises_neither_field() {
5150 let app = test_app();
5151 let (_, models) = get_json(&app, frink_api::routes::V1_MODELS).await;
5152 let entry = &models["data"][0];
5153 assert!(entry.get("supported_reasoning_efforts").is_none());
5154 assert!(entry.get("default_reasoning_effort").is_none());
5155 }
5156
5157 fn active_model(state: &AppState, name: &'static str) -> Arc<ActiveModel> {
5158 Arc::new(ActiveModel {
5159 id: Some(name.to_string()),
5160 loaded: Loaded::Generative(Arc::new(named_test_model(name, 256))),
5161 batcher: None,
5162 ceiling: None,
5163 checkpoint_path: None,
5164 })
5165 .tap_into(state)
5166 }
5167
5168 trait TapInto {
5170 fn tap_into(self, state: &AppState) -> Self;
5171 }
5172 impl TapInto for Arc<ActiveModel> {
5173 fn tap_into(self, state: &AppState) -> Self {
5174 state.swap_active(Some(Arc::clone(&self)));
5175 self
5176 }
5177 }
5178
5179 #[test]
5185 fn an_in_flight_request_keeps_the_model_it_started_on() {
5186 let state = test_state(
5187 named_test_model("model-a", 256),
5188 ResponseCache::new(4, Duration::from_secs(60)),
5189 );
5190
5191 let in_flight = state.active().expect("a model is loaded");
5194 assert_eq!(in_flight.name(), "model-a");
5195
5196 active_model(&state, "model-b");
5197
5198 assert_eq!(state.active().unwrap().name(), "model-b");
5200 assert_eq!(in_flight.name(), "model-a");
5202 let produced = run_generation(
5203 in_flight.generative().unwrap(),
5204 "hi",
5205 &greedy_params(3),
5206 None,
5207 None,
5208 None,
5209 None,
5210 None,
5211 None,
5212 )
5213 .expect("the old model must still decode after being swapped out");
5214 assert!(matches!(
5215 produced.choices[0].finish,
5216 FinishReason::Length | FinishReason::Stop
5217 ));
5218 }
5219
5220 #[test]
5225 fn a_swapped_out_model_lives_until_its_last_holder_releases_it() {
5226 let state = test_state(
5227 named_test_model("model-a", 256),
5228 ResponseCache::new(4, Duration::from_secs(60)),
5229 );
5230 let in_flight = state.active().expect("a model is loaded");
5231 let weights = Arc::clone(in_flight.generative().unwrap());
5232 assert!(Arc::strong_count(&weights) >= 2);
5233
5234 let previous = state.swap_active(Some(Arc::new(ActiveModel {
5235 id: Some("model-b".to_string()),
5236 loaded: Loaded::Generative(Arc::new(named_test_model("model-b", 256))),
5237 batcher: None,
5238 ceiling: None,
5239 checkpoint_path: None,
5240 })));
5241 drop(previous);
5242 assert!(Arc::strong_count(&weights) >= 2);
5244 drop(in_flight);
5245 assert_eq!(Arc::strong_count(&weights), 1);
5246 }
5247
5248 #[tokio::test]
5252 async fn unloading_answers_503_instead_of_serving_the_dropped_model() {
5253 let state = Arc::new(test_state(
5254 named_test_model("model-a", 256),
5255 ResponseCache::new(4, Duration::from_secs(60)),
5256 ));
5257 let app = test_app_with_state(Arc::clone(&state));
5258
5259 let (status, body) = post_json_uri(
5260 &app,
5261 frink_api::routes::ADMIN_MODELS_UNLOAD,
5262 serde_json::json!({}),
5263 )
5264 .await;
5265 assert_eq!(status, StatusCode::OK);
5266 assert_eq!(body["ok"], true);
5267 assert!(body["active"].is_null());
5268 assert!(state.active().is_none());
5269
5270 let (status, _) = get_json(&app, frink_api::routes::V1_MODELS).await;
5271 assert_eq!(status, StatusCode::OK);
5272 let (_, models) = get_json(&app, frink_api::routes::V1_MODELS).await;
5273 assert_eq!(models["data"].as_array().unwrap().len(), 0);
5274
5275 let (status, body) = post_json_uri(
5276 &app,
5277 "/v1/chat/completions",
5278 serde_json::json!({
5279 "model": "x",
5280 "messages": [{"role": "user", "content": "hi"}]
5281 }),
5282 )
5283 .await;
5284 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5285 assert_eq!(body["error"]["type"], "model_not_loaded");
5286 }
5287
5288 #[tokio::test]
5292 async fn health_reports_the_unloaded_state_rather_than_going_silent() {
5293 let state = Arc::new(test_state(
5294 named_test_model("model-a", 256),
5295 ResponseCache::new(4, Duration::from_secs(60)),
5296 ));
5297 let app = test_app_with_state(Arc::clone(&state));
5298 state.swap_active(None);
5299
5300 let (status, body) = get_json(&app, frink_api::routes::HEALTH).await;
5301 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5304 assert_eq!(body["state"], "unavailable");
5305 assert_eq!(body["reason"], "model_not_loaded");
5306 assert!(body["model"].is_null());
5307 let real_weights = body["capabilities"]
5308 .as_array()
5309 .unwrap()
5310 .iter()
5311 .find(|c| c["id"] == "real_weights")
5312 .cloned()
5313 .expect("real_weights is always reported");
5314 assert_eq!(real_weights["available"], false);
5315 assert_eq!(real_weights["reason"], "model_not_loaded");
5316 }
5317
5318 #[tokio::test]
5322 async fn a_finished_request_lands_in_the_stats_ring_with_both_durations() {
5323 let app = test_app();
5324
5325 let (status, completion) = post_json_uri(
5326 &app,
5327 "/v1/chat/completions",
5328 serde_json::json!({
5329 "model": "x",
5330 "messages": [{"role": "user", "content": "hi"}],
5331 "max_tokens": 4
5332 }),
5333 )
5334 .await;
5335 assert_eq!(status, StatusCode::OK);
5336 let request_id = completion["request_id"].as_str().unwrap().to_string();
5337
5338 let (status, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
5339 assert_eq!(status, StatusCode::OK);
5340 let recent = stats["recent"].as_array().unwrap();
5341 assert_eq!(recent.len(), 1);
5342 let row = &recent[0];
5343 assert_eq!(row["request_id"], request_id);
5344 assert_eq!(row["route"], frink_api::routes::V1_CHAT_COMPLETIONS);
5345 assert_eq!(row["status"], 200);
5346 assert_eq!(row["stream"], false);
5347 assert!(row["duration_ms"].is_number());
5350 assert!(row["decode_ms"].is_number());
5351 assert!(stats["tokens_generated_total"].as_u64().unwrap() > 0);
5352 assert_eq!(
5353 stats["tokens_prompt_total"].as_u64().unwrap(),
5354 row["prompt_tokens"].as_u64().unwrap()
5355 );
5356 }
5357
5358 #[tokio::test]
5362 async fn a_rejected_request_is_recorded_too() {
5363 let state = Arc::new(test_state(
5364 named_test_model("model-a", 256),
5365 ResponseCache::new(4, Duration::from_secs(60)),
5366 ));
5367 let app = test_app_with_state(Arc::clone(&state));
5368 state.swap_active(None);
5369
5370 let (status, _) = post_json_uri(
5371 &app,
5372 "/v1/chat/completions",
5373 serde_json::json!({"model": "x", "messages": [{"role": "user", "content": "hi"}]}),
5374 )
5375 .await;
5376 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5377
5378 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
5379 let recent = stats["recent"].as_array().unwrap();
5380 assert_eq!(recent.len(), 1);
5381 assert_eq!(recent[0]["status"], 503);
5382 assert_eq!(recent[0]["completion_tokens"], 0);
5383 assert!(recent[0]["decode_ms"].is_null());
5384 assert_eq!(stats["errors_total"], 1);
5385 }
5386
5387 async fn post_json_with_headers(
5391 app: &Router,
5392 uri: &str,
5393 body: serde_json::Value,
5394 headers: &[(&str, &str)],
5395 ) -> (StatusCode, serde_json::Value) {
5396 use http_body_util::BodyExt;
5397 use tower::ServiceExt;
5398
5399 let mut builder = axum::http::Request::builder()
5400 .method("POST")
5401 .uri(uri)
5402 .header("content-type", "application/json");
5403 for (name, value) in headers {
5404 builder = builder.header(*name, *value);
5405 }
5406 let response = app
5407 .clone()
5408 .oneshot(
5409 builder
5410 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
5411 .unwrap(),
5412 )
5413 .await
5414 .unwrap();
5415 let status = response.status();
5416 let bytes = response.into_body().collect().await.unwrap().to_bytes();
5417 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
5418 (status, json)
5419 }
5420
5421 #[tokio::test]
5425 async fn tokenize_detokenize_and_embeddings_all_land_in_the_ring() {
5426 let app = test_app();
5427
5428 let (status, _) = post_json_uri(
5429 &app,
5430 frink_api::routes::V1_TOKENIZE,
5431 serde_json::json!({"prompt": "hello"}),
5432 )
5433 .await;
5434 assert_eq!(status, StatusCode::OK);
5435 let (status, _) = post_json_uri(
5436 &app,
5437 frink_api::routes::V1_DETOKENIZE,
5438 serde_json::json!({"tokens": [104, 105]}),
5439 )
5440 .await;
5441 assert_eq!(status, StatusCode::OK);
5442 let (status, _) = post_json_uri(
5443 &app,
5444 frink_api::routes::V1_EMBEDDINGS,
5445 serde_json::json!({"input": "hello"}),
5446 )
5447 .await;
5448 assert_eq!(status, StatusCode::OK);
5449
5450 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
5451 let routes: Vec<&str> = stats["recent"]
5452 .as_array()
5453 .unwrap()
5454 .iter()
5455 .map(|row| row["route"].as_str().unwrap())
5456 .collect();
5457 for expected in [
5458 frink_api::routes::V1_TOKENIZE,
5459 frink_api::routes::V1_DETOKENIZE,
5460 frink_api::routes::V1_EMBEDDINGS,
5461 ] {
5462 assert!(
5463 routes.contains(&expected),
5464 "{expected} is missing: {routes:?}"
5465 );
5466 }
5467
5468 let row = |route: &str| {
5469 stats["recent"]
5470 .as_array()
5471 .unwrap()
5472 .iter()
5473 .find(|r| r["route"] == route)
5474 .cloned()
5475 .unwrap()
5476 };
5477 let embed = row(frink_api::routes::V1_EMBEDDINGS);
5481 assert!(embed["prompt_tokens"].as_u64().unwrap() > 0);
5482 assert!(embed["decode_ms"].is_null());
5483 assert_eq!(embed["completion_tokens"], 0);
5484 assert_eq!(row(frink_api::routes::V1_TOKENIZE)["prompt_tokens"], 0);
5488 assert_eq!(
5489 stats["tokens_prompt_total"].as_u64().unwrap(),
5490 embed["prompt_tokens"].as_u64().unwrap(),
5491 "only the forward pass counted"
5492 );
5493 }
5494
5495 fn streaming_test_app() -> Router {
5500 let mut cfg = test_dense_fixture();
5501 cfg.vocab_size = 256;
5502 let model = Model::Gguf(GgufModel {
5503 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5504 tokenizer: Arc::new(ServerTokenizer::Byte),
5505 stop_tokens: StopTokens::default(),
5506 bos_id: None,
5507 is_synthetic: false,
5508 chat_template: chat_template::PromptTemplate::plain(),
5509 });
5510 test_app_with_state(Arc::new(test_state(
5511 model,
5512 ResponseCache::new(1000, Duration::from_secs(3600)),
5513 )))
5514 }
5515
5516 #[tokio::test]
5525 async fn chat_logprobs_are_rendered_and_are_never_served_from_cache() {
5526 let app = test_app();
5527 let body = |logprobs: Option<(bool, Option<u32>)>| {
5528 let mut b = serde_json::json!({
5529 "model": "x",
5530 "messages": [{"role": "user", "content": "hi"}],
5531 "max_tokens": 4
5532 });
5533 if let Some((on, top)) = logprobs {
5534 b["logprobs"] = serde_json::json!(on);
5535 if let Some(n) = top {
5536 b["top_logprobs"] = serde_json::json!(n);
5537 }
5538 }
5539 b
5540 };
5541
5542 let (status, plain) =
5544 post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, body(None)).await;
5545 assert_eq!(status, StatusCode::OK, "{plain}");
5546 assert!(plain["choices"][0]["logprobs"].is_null(), "{plain}");
5547
5548 for attempt in 0..2 {
5552 let (status, with) = post_json_uri(
5553 &app,
5554 frink_api::routes::V1_CHAT_COMPLETIONS,
5555 body(Some((true, Some(2)))),
5556 )
5557 .await;
5558 assert_eq!(status, StatusCode::OK, "{with}");
5559 assert_ne!(
5560 with["frink_cache"], "hit",
5561 "attempt {attempt} replayed a cached answer for a logprobs request: {with}"
5562 );
5563 let lp = &with["choices"][0]["logprobs"];
5564 assert!(lp.is_object(), "attempt {attempt}: {with}");
5565 let content = lp["content"].as_array().expect("content");
5566 assert!(lp["tokens"].is_null(), "completions shape leaked: {lp}");
5568 for entry in content {
5569 assert!(entry["token"].is_string(), "{entry}");
5570 assert!(entry["bytes"].is_array(), "{entry}");
5571 let v = entry["logprob"].as_f64().expect("a real number");
5572 assert!(v <= 0.0 && v.is_finite(), "{entry}");
5573 let top = entry["top_logprobs"].as_array().expect("top_logprobs");
5574 assert!(top.len() <= 2, "asked for 2, got {}", top.len());
5575 }
5576 }
5577 }
5578
5579 #[tokio::test]
5585 async fn the_chat_logprobs_pair_is_validated() {
5586 let app = test_app();
5587 for (extra, why) in [
5588 (serde_json::json!({"top_logprobs": 3}), "without logprobs"),
5589 (
5590 serde_json::json!({"logprobs": true, "top_logprobs": 21}),
5591 "above the cap",
5592 ),
5593 ] {
5594 let mut body = serde_json::json!({
5595 "model": "x",
5596 "messages": [{"role": "user", "content": "hi"}],
5597 "max_tokens": 2
5598 });
5599 for (k, v) in extra.as_object().unwrap() {
5600 body[k] = v.clone();
5601 }
5602 let (status, answer) =
5603 post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, body).await;
5604 assert_eq!(status, StatusCode::BAD_REQUEST, "{why}: {answer}");
5605 assert!(
5606 answer["error"]["message"]
5607 .as_str()
5608 .is_some_and(|m| m.contains("top_logprobs")),
5609 "{why}: {answer}"
5610 );
5611 }
5612 }
5613
5614 #[tokio::test]
5623 async fn sleep_refuses_a_model_it_could_not_bring_back() {
5624 let app = test_app();
5625 let (status, answer) =
5626 post_json_uri(&app, frink_api::routes::SLEEP, serde_json::json!({})).await;
5627 assert_eq!(status, StatusCode::CONFLICT, "{answer}");
5628 assert_eq!(answer["error"]["type"], "not_reloadable", "{answer}");
5629 let (_, still) = get_json_uri(&app, frink_api::routes::IS_SLEEPING).await;
5632 assert_eq!(still["is_sleeping"], false, "{still}");
5633 let (status, _) = post_json_uri(
5634 &app,
5635 frink_api::routes::V1_CHAT_COMPLETIONS,
5636 serde_json::json!({
5637 "model": "x",
5638 "messages": [{"role": "user", "content": "hi"}],
5639 "max_tokens": 2
5640 }),
5641 )
5642 .await;
5643 assert_eq!(status, StatusCode::OK, "a refused sleep unloaded the model");
5644 }
5645
5646 #[tokio::test]
5655 async fn sleep_remembers_what_unload_forgets() {
5656 let state = Arc::new(test_state_at(
5659 test_model_full_byte_vocab(),
5660 ResponseCache::new(1000, Duration::from_secs(3600)),
5661 Some(std::path::PathBuf::from("/nonexistent/fixture.gguf")),
5662 ));
5663 let app = test_app_with_state(Arc::clone(&state));
5664 let ask = || {
5665 let app = app.clone();
5666 async move {
5667 post_json_uri(
5668 &app,
5669 frink_api::routes::V1_CHAT_COMPLETIONS,
5670 serde_json::json!({
5671 "model": "x",
5672 "messages": [{"role": "user", "content": "hi"}],
5673 "max_tokens": 2
5674 }),
5675 )
5676 .await
5677 }
5678 };
5679
5680 let (status, _) = ask().await;
5681 assert_eq!(status, StatusCode::OK, "the fixture server serves");
5682 let (_, awake) = get_json_uri(&app, frink_api::routes::IS_SLEEPING).await;
5683 assert_eq!(awake["is_sleeping"], false, "{awake}");
5684
5685 let (status, slept) =
5686 post_json_uri(&app, frink_api::routes::SLEEP, serde_json::json!({})).await;
5687 assert_eq!(status, StatusCode::OK, "{slept}");
5688 assert_eq!(slept["is_sleeping"], true, "{slept}");
5689 let (_, now) = get_json_uri(&app, frink_api::routes::IS_SLEEPING).await;
5690 assert_eq!(now["is_sleeping"], true, "{now}");
5691
5692 let (status, refused) = ask().await;
5695 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{refused}");
5696 assert_eq!(
5697 refused["error"]["type"], "server_sleeping",
5698 "an asleep server reported itself as empty: {refused}"
5699 );
5700
5701 let (status, again) =
5703 post_json_uri(&app, frink_api::routes::SLEEP, serde_json::json!({})).await;
5704 assert_eq!(status, StatusCode::OK, "{again}");
5705 assert_eq!(again["is_sleeping"], true, "{again}");
5706 }
5707
5708 #[tokio::test]
5712 async fn waking_a_server_that_is_awake_is_refused() {
5713 let app = test_app();
5714 let (status, answer) =
5715 post_json_uri(&app, frink_api::routes::WAKE_UP, serde_json::json!({})).await;
5716 assert_eq!(status, StatusCode::CONFLICT, "{answer}");
5717 assert_eq!(answer["error"]["type"], "not_sleeping", "{answer}");
5718 }
5719
5720 #[tokio::test]
5727 async fn a_salt_keeps_one_callers_cached_answer_from_another() {
5728 let app = test_app();
5729 let body = |salt: Option<&str>| {
5730 let mut b = serde_json::json!({
5731 "model": "x",
5732 "messages": [{"role": "user", "content": "the same prompt"}],
5733 "max_tokens": 4,
5734 "seed": 1
5735 });
5736 if let Some(s) = salt {
5737 b["cache_salt"] = serde_json::json!(s);
5738 }
5739 b
5740 };
5741 let post = |b: serde_json::Value| {
5742 let app = app.clone();
5743 async move { post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, b).await }
5744 };
5745
5746 let (status, _) = post(body(Some("tenant-a"))).await;
5748 assert_eq!(status, StatusCode::OK);
5749 let (_, again) = post(body(Some("tenant-a"))).await;
5750 assert_eq!(
5751 again["frink_cache"], "hit",
5752 "the owner did not get its own entry back: {again}"
5753 );
5754
5755 let (_, other) = post(body(Some("tenant-b"))).await;
5757 assert_ne!(
5758 other["frink_cache"], "hit",
5759 "a different caller was served tenant-a's answer: {other}"
5760 );
5761
5762 let (_, shared) = post(body(None)).await;
5764 assert_ne!(
5765 shared["frink_cache"], "hit",
5766 "an unsalted request was served a salted answer: {shared}"
5767 );
5768 }
5769
5770 #[tokio::test]
5773 async fn chat_serves_several_choices_from_one_prefill() {
5774 let app = test_app();
5775 let body = |n: u32, stream: bool| {
5776 serde_json::json!({
5777 "model": "x",
5778 "messages": [{"role": "user", "content": "hi"}],
5779 "max_tokens": 4,
5780 "temperature": 1.0,
5781 "n": n,
5782 "stream": stream
5783 })
5784 };
5785
5786 let (status, one) =
5787 post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, body(1, false)).await;
5788 assert_eq!(status, StatusCode::OK, "{one}");
5789
5790 let (status, three) =
5791 post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, body(3, false)).await;
5792 assert_eq!(status, StatusCode::OK, "{three}");
5793 let choices = three["choices"].as_array().expect("an array");
5794 assert_eq!(choices.len(), 3, "{three}");
5795 for (i, c) in choices.iter().enumerate() {
5796 assert_eq!(c["index"], i);
5797 assert!(c["message"]["role"].is_string(), "{c}");
5798 assert!(c["finish_reason"].is_string(), "{c}");
5799 }
5800 assert_eq!(
5802 three["usage"]["prompt_tokens"], one["usage"]["prompt_tokens"],
5803 "n = 3 billed the prompt more than once"
5804 );
5805 }
5806
5807 #[tokio::test]
5819 async fn a_streamed_n_interleaves_its_choices() {
5820 let app = streaming_test_app();
5821 let raw = post_sse_raw(
5822 &app,
5823 serde_json::json!({
5824 "model": "x",
5825 "messages": [{"role": "user", "content": "hi"}],
5826 "max_tokens": 6,
5827 "temperature": 1.0,
5828 "n": 3,
5829 "stream": true
5830 }),
5831 )
5832 .await;
5833
5834 let mut order: Vec<usize> = Vec::new();
5836 let mut finished: Vec<usize> = Vec::new();
5837 let mut usage_blocks = 0usize;
5838 for line in raw.lines() {
5839 let Some(rest) = line.strip_prefix("data: ") else {
5840 continue;
5841 };
5842 if rest.trim() == "[DONE]" {
5843 continue;
5844 }
5845 let v: serde_json::Value = serde_json::from_str(rest).expect(rest);
5846 if v.get("usage").is_some_and(|u| !u.is_null()) {
5847 usage_blocks += 1;
5848 }
5849 let Some(choice) = v["choices"].as_array().and_then(|c| c.first()) else {
5850 continue;
5851 };
5852 let index = choice["index"].as_u64().expect("an index") as usize;
5853 if choice["finish_reason"].is_string() {
5854 finished.push(index);
5855 continue;
5856 }
5857 order.push(index);
5858 }
5859
5860 assert_eq!(
5861 finished,
5862 vec![0, 1, 2],
5863 "one terminal chunk per choice, in index order: {raw}"
5864 );
5865 assert_eq!(usage_blocks, 1, "the usage block is the request's: {raw}");
5866 assert!(
5867 order.contains(&0) && order.contains(&2),
5868 "not every choice streamed: {order:?}"
5869 );
5870 let last_of_zero = order
5871 .iter()
5872 .rposition(|i| *i == 0)
5873 .expect("choice 0 streamed");
5874 let first_of_two = order
5875 .iter()
5876 .position(|i| *i == 2)
5877 .expect("choice 2 streamed");
5878 assert!(
5879 first_of_two < last_of_zero,
5880 "the choices arrived one after another rather than interleaved: {order:?}"
5881 );
5882 }
5883
5884 #[tokio::test]
5894 async fn every_route_refuses_the_same_unimplemented_fields() {
5895 let app = test_app();
5896 let fields = [
5897 ("n", serde_json::json!(3)),
5898 ("best_of", serde_json::json!(2)),
5899 ("prompt_logprobs", serde_json::json!(1)),
5900 ("echo", serde_json::json!(true)),
5901 ("use_beam_search", serde_json::json!(true)),
5902 ("truncate_prompt_tokens", serde_json::json!(8)),
5903 ("prompt_embeds", serde_json::json!("AA==")),
5904 ("allowed_token_ids", serde_json::json!([1, 2])),
5905 ("bad_words", serde_json::json!(["x"])),
5906 ("skip_special_tokens", serde_json::json!(false)),
5907 ("return_tokens_as_token_ids", serde_json::json!(true)),
5908 ];
5909 for (field, value) in fields {
5910 for (uri, base) in [
5911 (
5912 frink_api::routes::V1_CHAT_COMPLETIONS,
5913 serde_json::json!({
5914 "model": "x",
5915 "messages": [{"role": "user", "content": "hi"}],
5916 "max_tokens": 2
5917 }),
5918 ),
5919 (
5920 frink_api::routes::V1_COMPLETIONS,
5921 serde_json::json!({"prompt": "hi", "max_tokens": 2}),
5922 ),
5923 (
5924 frink_api::routes::COMPLETION,
5925 serde_json::json!({"prompt": "hi", "n_predict": 2}),
5926 ),
5927 ] {
5928 let mut body = base;
5929 body[field] = value.clone();
5930 if field == "prompt_logprobs" && uri == frink_api::routes::V1_COMPLETIONS {
5937 let (status, answer) = post_json_uri(&app, uri, body).await;
5938 assert_eq!(status, StatusCode::OK, "{uri} refused it: {answer}");
5939 assert!(
5940 answer["prompt_logprobs"].is_array(),
5941 "served without the field: {answer}"
5942 );
5943 continue;
5944 }
5945 if (field == "n" || field == "best_of")
5946 && (uri == frink_api::routes::V1_COMPLETIONS
5947 || uri == frink_api::routes::V1_CHAT_COMPLETIONS)
5948 {
5949 let (status, answer) = post_json_uri(&app, uri, body).await;
5950 assert_eq!(
5951 status,
5952 StatusCode::OK,
5953 "{uri} refused a served `{field}`: {answer}"
5954 );
5955 let want = if field == "n" { 3 } else { 1 };
5959 assert_eq!(
5960 answer["choices"].as_array().map(Vec::len),
5961 Some(want),
5962 "{field}: {answer}"
5963 );
5964 continue;
5965 }
5966 let (status, answer) = post_json_uri(&app, uri, body).await;
5967 assert_eq!(
5968 status,
5969 StatusCode::NOT_IMPLEMENTED,
5970 "{uri} served `{field}` instead of refusing it: {answer}"
5971 );
5972 assert!(
5973 answer["error"]["message"]
5974 .as_str()
5975 .is_some_and(|m| m.contains(field)),
5976 "{uri} refused `{field}` without naming it: {answer}"
5977 );
5978 }
5979 }
5980 }
5981
5982 #[tokio::test]
5983 async fn the_native_completion_wire_is_not_the_openai_one() {
5984 let app = test_app();
5985
5986 let (status, native) = post_json_uri(
5987 &app,
5988 frink_api::routes::COMPLETION,
5989 serde_json::json!({"prompt": "hi", "n_predict": 4}),
5990 )
5991 .await;
5992 assert_eq!(status, StatusCode::OK, "{native}");
5993 assert!(native["content"].is_string(), "{native}");
5994 assert_eq!(native["stop"], true);
5995 assert_eq!(native["stop_type"], "limit");
5996 assert_eq!(native["stopping_word"], "");
5997 assert_eq!(native["truncated"], false);
5998 assert_eq!(native["id_slot"], -1);
5999 assert!(native["timings"]["prompt_n"].is_number(), "{native}");
6000 assert!(native["generation_settings"]["n_predict"] == 4, "{native}");
6001 assert!(
6002 native.get("choices").is_none(),
6003 "the native shape has no `choices`: {native}"
6004 );
6005
6006 let (status, openai) = post_json_uri(
6007 &app,
6008 frink_api::routes::V1_COMPLETIONS,
6009 serde_json::json!({"prompt": "hi", "max_tokens": 4}),
6010 )
6011 .await;
6012 assert_eq!(status, StatusCode::OK);
6013 assert!(openai["choices"][0]["text"].is_string(), "{openai}");
6014 assert!(
6015 openai.get("content").is_none(),
6016 "the OpenAI shape has no top-level `content`: {openai}"
6017 );
6018 }
6019
6020 #[tokio::test]
6024 async fn both_native_spellings_reach_the_same_handler() {
6025 let app = test_app();
6026 for route in [
6027 frink_api::routes::COMPLETION,
6028 frink_api::routes::COMPLETIONS,
6029 ] {
6030 let (status, body) = post_json_uri(
6031 &app,
6032 route,
6033 serde_json::json!({"prompt": "hi", "n_predict": 2, "seed": 1}),
6034 )
6035 .await;
6036 assert_eq!(status, StatusCode::OK, "{route}: {body}");
6037 assert_eq!(body["stop"], true, "{route}");
6038 assert!(body["content"].is_string(), "{route}");
6039 }
6040
6041 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6044 let routes: Vec<&str> = stats["recent"]
6045 .as_array()
6046 .unwrap()
6047 .iter()
6048 .map(|row| row["route"].as_str().unwrap())
6049 .collect();
6050 assert!(
6051 routes.contains(&frink_api::routes::COMPLETION),
6052 "{routes:?}"
6053 );
6054 assert!(
6055 routes.contains(&frink_api::routes::COMPLETIONS),
6056 "{routes:?}"
6057 );
6058 }
6059
6060 #[tokio::test]
6066 async fn a_native_stream_ends_on_a_stop_frame_with_no_done_sentinel() {
6067 let app = streaming_test_app();
6068 let raw = post_sse_raw_uri(
6069 &app,
6070 frink_api::routes::COMPLETION,
6071 serde_json::json!({"prompt": "hi", "n_predict": 6, "stream": true, "seed": 7}),
6072 )
6073 .await;
6074
6075 assert!(
6076 !raw.contains("[DONE]"),
6077 "llama.cpp's native stream has no sentinel: {raw}"
6078 );
6079 let frames: Vec<serde_json::Value> = raw
6080 .lines()
6081 .filter_map(|line| line.strip_prefix("data: "))
6082 .map(|json| serde_json::from_str(json).expect("every frame is one JSON object"))
6083 .collect();
6084 assert!(frames.len() >= 2, "expected partials then a final: {raw}");
6085
6086 let (last, partials) = frames.split_last().unwrap();
6087 assert_eq!(last["stop"], true, "the last frame closes the stream");
6088 assert!(last["timings"].is_object(), "{last}");
6089 assert!(last["stop_type"].is_string(), "{last}");
6090 for partial in partials {
6091 assert_eq!(partial["stop"], false, "{partial}");
6092 assert!(partial["content"].is_string(), "{partial}");
6093 assert!(partial.get("timings").is_none(), "{partial}");
6097 assert!(partial.get("generation_settings").is_none(), "{partial}");
6098 }
6099 let streamed: String = partials
6102 .iter()
6103 .filter_map(|p| p["content"].as_str())
6104 .collect();
6105 assert_eq!(last["content"].as_str().unwrap(), streamed);
6106 }
6107
6108 #[tokio::test]
6113 async fn an_unbounded_n_predict_is_refused_rather_than_quietly_shrunk() {
6114 let app = test_app();
6115 for body in [
6116 serde_json::json!({"prompt": "hi"}),
6117 serde_json::json!({"prompt": "hi", "n_predict": -1}),
6118 ] {
6119 let (status, refusal) =
6120 post_json_uri(&app, frink_api::routes::COMPLETION, body.clone()).await;
6121 assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{body}: {refusal}");
6122 assert!(
6123 refusal["error"]["message"]
6124 .as_str()
6125 .unwrap()
6126 .contains("n_predict"),
6127 "{refusal}"
6128 );
6129 }
6130 let (status, _) = post_json_uri(
6133 &app,
6134 frink_api::routes::COMPLETION,
6135 serde_json::json!({"prompt": "hi", "n_predict": 2}),
6136 )
6137 .await;
6138 assert_eq!(status, StatusCode::OK);
6139 }
6140
6141 #[tokio::test]
6150 async fn a_stop_string_halts_the_answer_and_is_named_back() {
6151 let app = streaming_test_app();
6152 let ask = |stop: serde_json::Value| {
6153 let app = app.clone();
6154 async move {
6155 post_json_uri(
6156 &app,
6157 frink_api::routes::COMPLETION,
6158 serde_json::json!({
6159 "prompt": "hi",
6160 "n_predict": 64,
6161 "ignore_eos": true,
6162 "stop": stop,
6163 }),
6164 )
6165 .await
6166 .1
6167 }
6168 };
6169
6170 let baseline = ask(serde_json::json!([])).await;
6171 assert_eq!(baseline["stop_type"], "limit");
6172 assert_eq!(baseline["stopping_word"], "");
6173 let text = baseline["content"].as_str().unwrap().to_string();
6174 let sentinel: String = text.chars().skip(1).take(2).collect();
6180 assert_eq!(
6181 sentinel.chars().count(),
6182 2,
6183 "the fixture must produce enough output to cut: {text:?}"
6184 );
6185 let cut = text.find(&sentinel).expect("it came out of this text");
6186
6187 let stopped = ask(serde_json::json!([sentinel])).await;
6188 assert_eq!(stopped["stop_type"], "word", "{stopped}");
6189 assert_eq!(stopped["stopping_word"], sentinel);
6190 assert_eq!(
6191 stopped["content"].as_str().unwrap(),
6192 &text[..cut],
6193 "the answer must be cut at the sentinel, not run past it"
6194 );
6195 }
6196
6197 #[tokio::test]
6203 async fn the_llama_cpp_spelling_of_tokenize_reaches_the_same_handler() {
6204 let app = test_app();
6205
6206 let (v1_status, v1) = post_json_uri(
6207 &app,
6208 frink_api::routes::V1_TOKENIZE,
6209 serde_json::json!({"prompt": "hello"}),
6210 )
6211 .await;
6212 let (alias_status, alias) = post_json_uri(
6213 &app,
6214 frink_api::routes::TOKENIZE,
6215 serde_json::json!({"content": "hello"}),
6216 )
6217 .await;
6218 assert_eq!(v1_status, StatusCode::OK);
6219 assert_eq!(alias_status, StatusCode::OK, "{alias}");
6220 assert_eq!(v1["tokens"], alias["tokens"]);
6221 assert!(!alias["tokens"].as_array().unwrap().is_empty());
6222
6223 let (status, both_ways) = post_json_uri(
6226 &app,
6227 frink_api::routes::TOKENIZE,
6228 serde_json::json!({"prompt": "hello"}),
6229 )
6230 .await;
6231 assert_eq!(status, StatusCode::OK);
6232 assert_eq!(both_ways["tokens"], v1["tokens"]);
6233 }
6234
6235 #[tokio::test]
6240 async fn detokenize_answers_under_both_dialects_keys() {
6241 let app = test_app();
6242 for route in [
6243 frink_api::routes::DETOKENIZE,
6244 frink_api::routes::V1_DETOKENIZE,
6245 ] {
6246 let (status, body) =
6247 post_json_uri(&app, route, serde_json::json!({"tokens": [104, 105]})).await;
6248 assert_eq!(status, StatusCode::OK, "{route}");
6249 assert_eq!(body["text"], "hi", "{route}");
6250 assert_eq!(body["content"], body["text"], "{route}");
6251 }
6252 }
6253
6254 #[tokio::test]
6258 async fn the_alias_is_recorded_under_the_path_the_client_called() {
6259 let app = test_app();
6260 let (status, _) = post_json_uri(
6261 &app,
6262 frink_api::routes::TOKENIZE,
6263 serde_json::json!({"content": "hello"}),
6264 )
6265 .await;
6266 assert_eq!(status, StatusCode::OK);
6267
6268 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6269 let routes: Vec<&str> = stats["recent"]
6270 .as_array()
6271 .unwrap()
6272 .iter()
6273 .map(|row| row["route"].as_str().unwrap())
6274 .collect();
6275 assert!(
6276 routes.contains(&frink_api::routes::TOKENIZE),
6277 "the alias must be its own row: {routes:?}"
6278 );
6279 assert!(
6280 !routes.contains(&frink_api::routes::V1_TOKENIZE),
6281 "nothing called /v1/tokenize: {routes:?}"
6282 );
6283 }
6284
6285 #[tokio::test]
6290 async fn add_special_prepends_the_same_bos_the_decoder_would() {
6291 let mut cfg = test_dense_fixture();
6292 cfg.vocab_size = 256;
6293 let model = Model::Gguf(GgufModel {
6294 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
6295 tokenizer: Arc::new(ServerTokenizer::Byte),
6296 stop_tokens: StopTokens::default(),
6297 bos_id: Some(7),
6298 is_synthetic: true,
6299 chat_template: chat_template::PromptTemplate::plain(),
6300 });
6301 let app = test_app_with_state(Arc::new(test_state(
6302 model,
6303 ResponseCache::new(1000, Duration::from_secs(3600)),
6304 )));
6305
6306 let (_, plain) = post_json_uri(
6307 &app,
6308 frink_api::routes::TOKENIZE,
6309 serde_json::json!({"content": "hi"}),
6310 )
6311 .await;
6312 let (_, special) = post_json_uri(
6313 &app,
6314 frink_api::routes::TOKENIZE,
6315 serde_json::json!({"content": "hi", "add_special": true}),
6316 )
6317 .await;
6318
6319 assert_eq!(plain["tokens"], serde_json::json!([104, 105]));
6320 assert_eq!(special["tokens"], serde_json::json!([7, 104, 105]));
6321 assert_eq!(special["count"], 3);
6322 }
6323
6324 #[tokio::test]
6327 async fn a_rejected_embeddings_request_is_recorded_with_its_status() {
6328 let app = test_app();
6329 let (status, _) = post_json_uri(
6330 &app,
6331 frink_api::routes::V1_EMBEDDINGS,
6332 serde_json::json!({"input": "hi", "encoding_format": "base64"}),
6333 )
6334 .await;
6335 assert_eq!(status, StatusCode::BAD_REQUEST);
6336
6337 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6338 let recent = stats["recent"].as_array().unwrap();
6339 assert_eq!(recent.len(), 1);
6340 assert_eq!(recent[0]["route"], frink_api::routes::V1_EMBEDDINGS);
6341 assert_eq!(recent[0]["status"], 400);
6342 assert_eq!(
6343 recent[0]["prompt_tokens"], 0,
6344 "a rejected call embedded nothing"
6345 );
6346 }
6347
6348 #[tokio::test]
6351 async fn a_row_names_the_key_that_served_it_without_carrying_the_key() {
6352 let app = test_app();
6353 let key = "sk-monitor-secret";
6354 let (status, _) = post_json_with_headers(
6355 &app,
6356 "/v1/chat/completions",
6357 serde_json::json!({
6358 "model": "x",
6359 "messages": [{"role": "user", "content": "hi"}],
6360 "max_tokens": 2
6361 }),
6362 &[
6363 ("authorization", &format!("Bearer {key}")),
6364 ("x-frink-client", "frink-studio"),
6365 ],
6366 )
6367 .await;
6368 assert_eq!(status, StatusCode::OK);
6369
6370 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6371 let row = stats["recent"].as_array().unwrap()[0].clone();
6372 let fingerprint = row["via_api_key"]
6373 .as_str()
6374 .expect("the row names the key that served it")
6375 .to_string();
6376 assert_eq!(fingerprint, attribution::key_fingerprint(key));
6377 assert!(!fingerprint.contains(key));
6378 assert!(
6379 !serde_json::to_string(&stats).unwrap().contains(key),
6380 "the stats payload must not carry the key in any form"
6381 );
6382 assert_eq!(row["client"], "frink-studio");
6383 }
6384
6385 #[tokio::test]
6388 async fn different_keys_are_different_callers_and_no_key_is_null() {
6389 let app = test_app();
6390 let body = serde_json::json!({
6391 "model": "x",
6392 "messages": [{"role": "user", "content": "hi"}],
6393 "max_tokens": 1
6394 });
6395 for headers in [
6396 vec![("authorization", "Bearer key-one")],
6397 vec![("authorization", "Bearer key-two")],
6398 vec![],
6399 ] {
6400 let (status, _) =
6401 post_json_with_headers(&app, "/v1/chat/completions", body.clone(), &headers).await;
6402 assert_eq!(status, StatusCode::OK);
6403 }
6404
6405 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6406 let recent = stats["recent"].as_array().unwrap();
6407 assert_eq!(recent.len(), 3);
6408 let one = recent[0]["via_api_key"].as_str().unwrap();
6409 let two = recent[1]["via_api_key"].as_str().unwrap();
6410 assert_ne!(one, two, "two keys must not collapse into one caller");
6411 assert!(
6412 recent[2]["via_api_key"].is_null(),
6413 "an unauthenticated call is null, not a fingerprint of nothing"
6414 );
6415 assert!(recent[2]["client"].is_null());
6416 }
6417
6418 #[tokio::test]
6423 async fn a_row_names_the_model_that_served_it_not_the_one_requested() {
6424 let state = Arc::new(test_state(
6425 named_test_model("really-loaded", 256),
6426 ResponseCache::new(4, Duration::from_secs(60)),
6427 ));
6428 let app = test_app_with_state(Arc::clone(&state));
6429
6430 let (status, _) = post_json_uri(
6431 &app,
6432 "/v1/chat/completions",
6433 serde_json::json!({
6434 "model": "gpt-4-turbo-that-is-not-here",
6435 "messages": [{"role": "user", "content": "hi"}],
6436 "max_tokens": 2
6437 }),
6438 )
6439 .await;
6440 assert_eq!(status, StatusCode::OK);
6441
6442 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6443 assert_eq!(stats["recent"][0]["model"], "really-loaded");
6444
6445 state.swap_active(None);
6448 let (status, _) = post_json_uri(
6449 &app,
6450 "/v1/chat/completions",
6451 serde_json::json!({
6452 "model": "gpt-4-turbo-that-is-not-here",
6453 "messages": [{"role": "user", "content": "hi"}]
6454 }),
6455 )
6456 .await;
6457 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
6458 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6459 let recent = stats["recent"].as_array().unwrap();
6460 assert!(recent[recent.len() - 1]["model"].is_null());
6461 }
6462
6463 #[tokio::test]
6467 async fn a_streamed_row_names_the_model_it_decoded_against() {
6468 let state = Arc::new(test_state(
6469 named_test_model("model-before", 256),
6470 ResponseCache::new(4, Duration::from_secs(60)),
6471 ));
6472 let app = test_app_with_state(Arc::clone(&state));
6473 let _ = post_sse_raw(&app, resumable_request()).await;
6474 active_model(&state, "model-after");
6476
6477 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6478 assert_eq!(stats["recent"][0]["model"], "model-before");
6479 }
6480
6481 #[tokio::test]
6484 async fn the_queue_gauge_is_null_when_nothing_can_queue() {
6485 let app = test_app();
6486 let (status, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6487 assert_eq!(status, StatusCode::OK);
6488 assert!(
6489 stats["queue_depth"].is_null(),
6490 "without continuous batching nothing queues, so there is nothing to measure"
6491 );
6492 assert!(stats["queue_rejected_total"].is_null());
6493 assert_eq!(
6494 stats["generating_now"], 0,
6495 "work in progress is measured and really is zero here"
6496 );
6497 }
6498
6499 async fn post_sse_raw(app: &Router, body: serde_json::Value) -> String {
6504 post_sse_raw_uri(app, frink_api::routes::V1_CHAT_COMPLETIONS, body).await
6505 }
6506
6507 async fn post_sse_raw_uri(app: &Router, uri: &str, body: serde_json::Value) -> String {
6511 use http_body_util::BodyExt;
6512 use tower::ServiceExt;
6513
6514 let response = app
6515 .clone()
6516 .oneshot(
6517 axum::http::Request::builder()
6518 .method("POST")
6519 .uri(uri)
6520 .header("content-type", "application/json")
6521 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
6522 .unwrap(),
6523 )
6524 .await
6525 .unwrap();
6526 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6527 String::from_utf8(bytes.to_vec()).unwrap()
6528 }
6529
6530 async fn get_json_with_headers(
6531 app: &Router,
6532 uri: &str,
6533 headers: &[(&str, &str)],
6534 ) -> (StatusCode, serde_json::Value) {
6535 use http_body_util::BodyExt;
6536 use tower::ServiceExt;
6537
6538 let mut builder = axum::http::Request::builder().method("GET").uri(uri);
6539 for (name, value) in headers {
6540 builder = builder.header(*name, *value);
6541 }
6542 let response = app
6543 .clone()
6544 .oneshot(builder.body(axum::body::Body::empty()).unwrap())
6545 .await
6546 .unwrap();
6547 let status = response.status();
6548 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6549 (
6550 status,
6551 serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({})),
6552 )
6553 }
6554
6555 fn sse_field<'a>(body: &'a str, field: &str) -> Vec<&'a str> {
6556 body.lines()
6557 .filter_map(|line| line.strip_prefix(field))
6558 .map(str::trim)
6559 .collect()
6560 }
6561
6562 fn resumable_request() -> serde_json::Value {
6563 serde_json::json!({
6564 "model": "m",
6565 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
6566 "max_tokens": 4,
6567 "temperature": 0,
6568 "stream": true,
6569 "stream_resumable": true,
6570 })
6571 }
6572
6573 #[tokio::test]
6578 async fn a_resumable_stream_numbers_every_event_and_states_retry_once() {
6579 let app = test_app();
6580 let body = post_sse_raw(&app, resumable_request()).await;
6581
6582 let request_id = body
6583 .lines()
6584 .find_map(|l| l.strip_prefix("data: "))
6585 .and_then(|d| serde_json::from_str::<serde_json::Value>(d).ok())
6586 .and_then(|v| v["request_id"].as_str().map(str::to_string))
6587 .expect("the first chunk names the request");
6588
6589 let ids = sse_field(&body, "id:");
6590 let datas = sse_field(&body, "data:");
6591 assert_eq!(
6592 ids.len(),
6593 datas.len(),
6594 "every event carries an id, or a reconnect cannot name where it stopped"
6595 );
6596 for (i, id) in ids.iter().enumerate() {
6597 assert_eq!(*id, format!("{request_id}:{i}"));
6598 }
6599 let retries = sse_field(&body, "retry:");
6600 assert_eq!(
6601 retries.len(),
6602 1,
6603 "the reconnect delay is stated once, not on every event"
6604 );
6605 assert_eq!(retries[0], "1500");
6606 assert!(
6607 body.contains("data: [DONE]"),
6608 "the end of stream is still stated"
6609 );
6610 }
6611
6612 #[tokio::test]
6616 async fn a_plain_stream_carries_no_id_because_nothing_could_replay_it() {
6617 let app = test_app();
6618 let mut request = resumable_request();
6619 request["stream_resumable"] = serde_json::json!(false);
6620 let body = post_sse_raw(&app, request).await;
6621 assert!(!sse_field(&body, "data:").is_empty(), "it still streams");
6622 assert!(
6623 sse_field(&body, "id:").is_empty(),
6624 "an id promises a replay this stream cannot serve"
6625 );
6626 assert!(sse_field(&body, "retry:").is_empty());
6627 }
6628
6629 #[tokio::test]
6633 async fn the_polling_fallback_serves_exactly_what_the_stream_delivered() {
6634 let app = test_app();
6635 let body = post_sse_raw(&app, resumable_request()).await;
6636 let request_id = sse_field(&body, "id:")[0]
6637 .rsplit_once(':')
6638 .unwrap()
6639 .0
6640 .to_string();
6641 let streamed: Vec<String> = sse_field(&body, "data:")
6642 .iter()
6643 .map(|d| d.to_string())
6644 .collect();
6645
6646 let (status, polled) = get_json(
6647 &app,
6648 &format!("{}?from=0", frink_api::routes::v1_stream_poll(&request_id)),
6649 )
6650 .await;
6651 assert_eq!(status, StatusCode::OK);
6652 let events: Vec<String> = polled["events"]
6653 .as_array()
6654 .unwrap()
6655 .iter()
6656 .map(|e| e["data"].as_str().unwrap().to_string())
6657 .collect();
6658 assert_eq!(
6659 events, streamed,
6660 "the fallback must deliver the same answer, not a re-run of it"
6661 );
6662 assert_eq!(polled["request_id"], request_id);
6663 assert_eq!(
6664 polled["done"], false,
6665 "events were still being handed out, so the client must ask again"
6666 );
6667
6668 let next = polled["next_index"].as_u64().unwrap();
6671 let (_, drained) = get_json(
6672 &app,
6673 &format!(
6674 "{}?from={next}",
6675 frink_api::routes::v1_stream_poll(&request_id)
6676 ),
6677 )
6678 .await;
6679 assert_eq!(drained["done"], true);
6680 assert_eq!(drained["events"].as_array().unwrap().len(), 0);
6681 }
6682
6683 #[tokio::test]
6687 async fn a_resume_continues_after_the_last_event_id_rather_than_repeating() {
6688 let app = test_app();
6689 let body = post_sse_raw(&app, resumable_request()).await;
6690 let ids = sse_field(&body, "id:");
6691 let datas: Vec<String> = sse_field(&body, "data:")
6692 .iter()
6693 .map(|d| d.to_string())
6694 .collect();
6695 assert!(
6696 ids.len() >= 3,
6697 "need a few events to resume into the middle"
6698 );
6699 let request_id = ids[0].rsplit_once(':').unwrap().0.to_string();
6700
6701 let (status, resumed) = get_json_with_headers(
6702 &app,
6703 &format!("{}/poll", frink_api::routes::v1_stream(&request_id)),
6704 &[],
6705 )
6706 .await;
6707 assert_eq!(status, StatusCode::OK);
6708 assert_eq!(resumed["events"].as_array().unwrap().len(), datas.len());
6709
6710 let (_, tail) = get_json(
6712 &app,
6713 &format!("{}?from=2", frink_api::routes::v1_stream_poll(&request_id)),
6714 )
6715 .await;
6716 let tail_events: Vec<String> = tail["events"]
6717 .as_array()
6718 .unwrap()
6719 .iter()
6720 .map(|e| e["data"].as_str().unwrap().to_string())
6721 .collect();
6722 assert_eq!(tail_events, datas[2..].to_vec());
6723 }
6724
6725 #[tokio::test]
6728 async fn an_sse_reconnect_resumes_from_the_last_event_id() {
6729 use http_body_util::BodyExt;
6730 use tower::ServiceExt;
6731
6732 let app = test_app();
6733 let body = post_sse_raw(&app, resumable_request()).await;
6734 let ids = sse_field(&body, "id:");
6735 let datas: Vec<String> = sse_field(&body, "data:")
6736 .iter()
6737 .map(|d| d.to_string())
6738 .collect();
6739 let request_id = ids[0].rsplit_once(':').unwrap().0.to_string();
6740
6741 let response = app
6742 .clone()
6743 .oneshot(
6744 axum::http::Request::builder()
6745 .method("GET")
6746 .uri(frink_api::routes::v1_stream(&request_id))
6747 .header("last-event-id", format!("{request_id}:0"))
6748 .body(axum::body::Body::empty())
6749 .unwrap(),
6750 )
6751 .await
6752 .unwrap();
6753 assert_eq!(response.status(), StatusCode::OK);
6754 assert_eq!(
6755 response
6756 .headers()
6757 .get("x-accel-buffering")
6758 .and_then(|v| v.to_str().ok()),
6759 Some("no"),
6760 "the reconnect needs the same anti-buffering header as the stream"
6761 );
6762 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6763 let resumed = String::from_utf8(bytes.to_vec()).unwrap();
6764 assert_eq!(
6765 sse_field(&resumed, "data:")
6766 .iter()
6767 .map(|d| d.to_string())
6768 .collect::<Vec<_>>(),
6769 datas[1..].to_vec()
6770 );
6771 assert_eq!(sse_field(&resumed, "id:")[0], format!("{request_id}:1"));
6772 }
6773
6774 #[tokio::test]
6778 async fn a_last_event_id_from_another_stream_is_refused() {
6779 let app = test_app();
6780 let body = post_sse_raw(&app, resumable_request()).await;
6781 let request_id = sse_field(&body, "id:")[0]
6782 .rsplit_once(':')
6783 .unwrap()
6784 .0
6785 .to_string();
6786
6787 let (status, err) = get_json_with_headers(
6788 &app,
6789 &frink_api::routes::v1_stream(&request_id),
6790 &[("last-event-id", "chatcmpl-someone-else:3")],
6791 )
6792 .await;
6793 assert_eq!(status, StatusCode::BAD_REQUEST);
6794 assert_eq!(err["error"]["code"], "bad_last_event_id");
6795 }
6796
6797 #[tokio::test]
6801 async fn resuming_a_stream_that_was_never_resumable_is_a_404_that_says_why() {
6802 let app = test_app();
6803 let mut request = resumable_request();
6804 request["stream_resumable"] = serde_json::json!(false);
6805 let body = post_sse_raw(&app, request).await;
6806 let request_id = body
6807 .lines()
6808 .find_map(|l| l.strip_prefix("data: "))
6809 .and_then(|d| serde_json::from_str::<serde_json::Value>(d).ok())
6810 .and_then(|v| v["request_id"].as_str().map(str::to_string))
6811 .unwrap();
6812
6813 let (status, err) = get_json(&app, &frink_api::routes::v1_stream_poll(&request_id)).await;
6814 assert_eq!(status, StatusCode::NOT_FOUND);
6815 assert_eq!(err["error"]["code"], "stream_not_found");
6816 assert!(err["error"]["message"]
6817 .as_str()
6818 .unwrap()
6819 .contains("stream_resumable"));
6820 }
6821
6822 #[test]
6826 fn the_axum_stream_patterns_match_the_published_templates() {
6827 assert_eq!(
6828 axum_path(frink_api::routes::V1_STREAM),
6829 "/v1/stream/:request_id"
6830 );
6831 assert_eq!(
6832 axum_path(frink_api::routes::V1_STREAM_POLL),
6833 "/v1/stream/:request_id/poll"
6834 );
6835 assert_eq!(
6836 frink_api::routes::v1_stream("abc"),
6837 axum_path(frink_api::routes::V1_STREAM).replace(":request_id", "abc")
6838 );
6839 }
6840
6841 #[test]
6851 fn no_published_template_reaches_the_router_with_its_braces() {
6852 for template in [
6853 frink_api::routes::V1_STREAM,
6854 frink_api::routes::V1_STREAM_POLL,
6855 frink_api::routes::V1_RESPONSE,
6856 frink_api::routes::V1_RESPONSE_CANCEL,
6857 frink_api::routes::ADMIN_TASK_CANCEL,
6858 ] {
6859 assert!(
6860 template.contains('{'),
6861 "{template} is in the template list but has no placeholder"
6862 );
6863 let mounted = axum_path(template);
6864 assert!(
6865 !mounted.contains('{') && !mounted.contains('}'),
6866 "{template} would be mounted as {mounted}, whose braces axum reads as a literal segment"
6867 );
6868 assert!(
6869 mounted.contains(':'),
6870 "{template} lost its placeholder entirely and would match one path only"
6871 );
6872 }
6873 }
6874
6875 #[tokio::test]
6883 async fn an_unknown_response_id_gets_the_handler_not_a_bare_404() {
6884 let app = test_app();
6885 let (status, body) = get_json(&app, "/v1/responses/resp_nonexistent").await;
6886 assert_eq!(status, StatusCode::NOT_FOUND);
6887 assert!(
6888 !body.is_null(),
6889 "empty body means axum never matched the route, so the id was read as a literal segment"
6890 );
6891 }
6892
6893 #[tokio::test]
6896 async fn the_task_list_starts_empty_rather_than_absent() {
6897 let app = test_app();
6898 let (status, body) = get_json(&app, frink_api::routes::ADMIN_TASKS).await;
6899 assert_eq!(status, StatusCode::OK);
6900 assert_eq!(body["tasks"].as_array().unwrap().len(), 0);
6901 }
6902
6903 #[tokio::test]
6912 async fn the_slots_route_is_registered_and_refuses_by_naming_slot_save_path() {
6913 assert!(
6914 std::env::var("FRINK_SLOT_SAVE_PATH").is_err(),
6915 "this test asserts the unconfigured behaviour"
6916 );
6917 let app = test_app();
6918 let (status, body) = post_json_uri(
6919 &app,
6920 &format!("{}?action=save", frink_api::routes::slots_id(0)),
6921 serde_json::json!({"filename": "sys.fslot", "prompt": "hi"}),
6922 )
6923 .await;
6924 assert_eq!(status, StatusCode::NOT_IMPLEMENTED);
6925 assert!(
6926 body["error"]["message"]
6927 .as_str()
6928 .unwrap()
6929 .contains("--slot-save-path"),
6930 "{body}"
6931 );
6932 }
6933
6934 pub(crate) async fn post_json_uri(
6935 app: &Router,
6936 uri: &str,
6937 body: serde_json::Value,
6938 ) -> (StatusCode, serde_json::Value) {
6939 use http_body_util::BodyExt;
6940 use tower::ServiceExt;
6941
6942 let response = app
6943 .clone()
6944 .oneshot(
6945 axum::http::Request::builder()
6946 .method("POST")
6947 .uri(uri)
6948 .header("content-type", "application/json")
6949 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
6950 .unwrap(),
6951 )
6952 .await
6953 .unwrap();
6954 let status = response.status();
6955 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6956 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
6957 (status, json)
6958 }
6959
6960 pub(crate) async fn get_json_uri(app: &Router, uri: &str) -> (StatusCode, serde_json::Value) {
6963 use http_body_util::BodyExt;
6964 use tower::ServiceExt;
6965
6966 let response = app
6967 .clone()
6968 .oneshot(
6969 axum::http::Request::builder()
6970 .method("GET")
6971 .uri(uri)
6972 .body(axum::body::Body::empty())
6973 .unwrap(),
6974 )
6975 .await
6976 .unwrap();
6977 let status = response.status();
6978 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6979 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
6980 (status, json)
6981 }
6982
6983 async fn post_json(app: &Router, body: serde_json::Value) -> serde_json::Value {
6984 post_json_uri(app, "/v1/chat/completions", body).await.1
6985 }
6986
6987 #[tokio::test]
6996 async fn stats_says_what_the_engine_is_using_and_which_quantity_that_is() {
6997 let app = test_app();
6998 let (status, body) = get_json(&app, frink_api::routes::V1_STATS).await;
6999 assert_eq!(status, StatusCode::OK);
7000
7001 let memory = &body["memory"];
7002 if memory.is_null() {
7003 return;
7006 }
7007 assert!(
7008 memory["bytes"].as_u64().is_some_and(|b| b > 0),
7009 "a read that produced a zero is a broken read, not an idle \
7010 engine: {memory}"
7011 );
7012 assert!(
7013 ["pss", "rss"].contains(&memory["kind"].as_str().unwrap_or("")),
7014 "the quantity must travel with the number: {memory}"
7015 );
7016 }
7017
7018 #[tokio::test]
7024 async fn stats_reports_a_pool_it_does_not_have_as_absent_and_not_as_zero() {
7025 let app = test_app();
7026 let (status, body) = get_json(&app, frink_api::routes::V1_STATS).await;
7027 assert_eq!(status, StatusCode::OK);
7028 for pool in ["kv_pages", "window_slots", "state_slots"] {
7029 assert!(
7030 body["pools"][pool].is_null(),
7031 "{pool} must be null rather than a zero row: {}",
7032 body["pools"]
7033 );
7034 }
7035 }
7036
7037 #[tokio::test]
7045 async fn a_messages_response_states_the_id_that_v1_cancel_takes() {
7046 use http_body_util::BodyExt;
7047 use tower::ServiceExt;
7048
7049 let app = test_app();
7050 let send = |body: serde_json::Value| {
7051 let app = app.clone();
7052 async move {
7053 app.oneshot(
7054 axum::http::Request::builder()
7055 .method("POST")
7056 .uri(frink_api::routes::V1_MESSAGES)
7057 .header("content-type", "application/json")
7058 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
7059 .unwrap(),
7060 )
7061 .await
7062 .unwrap()
7063 }
7064 };
7065
7066 let ok = send(serde_json::json!({
7067 "model": "test",
7068 "max_tokens": 1,
7069 "messages": [{"role": "user", "content": "hi"}],
7070 }))
7071 .await;
7072 assert_eq!(ok.status(), StatusCode::OK);
7073 let id = ok
7074 .headers()
7075 .get("request-id")
7076 .expect("a served message names its id")
7077 .to_str()
7078 .unwrap()
7079 .to_string();
7080 assert!(!id.is_empty());
7081
7082 let bad = send(serde_json::json!({"model": "test"})).await;
7085 assert!(bad.status().is_client_error());
7086 let other = bad.headers().get("request-id").expect("errors too");
7087 assert_ne!(other.to_str().unwrap(), id);
7088 let _ = bad.into_body().collect().await.unwrap();
7089 }
7090
7091 #[tokio::test]
7099 async fn a_request_that_arrives_mid_rebuild_is_refused_and_admitted_again_after() {
7100 let state = Arc::new(test_state(
7101 test_model_full_byte_vocab(),
7102 ResponseCache::new(1000, Duration::from_secs(3600)),
7103 ));
7104 let app = test_app_with_state(Arc::clone(&state));
7105 let body = serde_json::json!({
7106 "model": "test",
7107 "messages": [{"role": "user", "content": "hi"}],
7108 "max_tokens": 1,
7109 });
7110
7111 state
7112 .maintenance
7113 .lock()
7114 .unwrap()
7115 .begin_rebuild()
7116 .expect("a fresh server is serving, so the rebuild starts");
7117 let (status, refused) = post_json_uri(&app, "/v1/chat/completions", body.clone()).await;
7118 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
7119 assert_eq!(refused["error"]["type"], "cache_rebuilding");
7120
7121 state.maintenance.lock().unwrap().finish_rebuild(true);
7122 let (status, _) = post_json_uri(&app, "/v1/chat/completions", body).await;
7123 assert_eq!(
7124 status,
7125 StatusCode::OK,
7126 "the gate reopens; a rebuild is not a latch"
7127 );
7128 }
7129
7130 #[tokio::test]
7135 async fn cancelling_an_id_that_is_not_generating_is_a_404_that_says_so() {
7136 let app = test_app();
7137 let (status, body) = post_json_uri(
7138 &app,
7139 frink_api::routes::V1_CANCEL,
7140 serde_json::json!({ "request_id": "chatcmpl-never-issued" }),
7141 )
7142 .await;
7143 assert_eq!(status, StatusCode::NOT_FOUND);
7144 assert_eq!(body["cancelled"], serde_json::json!(false));
7145 assert_eq!(body["request_id"], "chatcmpl-never-issued");
7146 assert!(
7147 body["detail"].as_str().is_some_and(|d| !d.is_empty()),
7148 "the verdict must carry a human reason: {body}"
7149 );
7150 }
7151
7152 #[tokio::test]
7156 async fn cancelling_a_live_generation_signals_its_token_and_answers_200() {
7157 let state = Arc::new(test_state(
7158 test_model_full_byte_vocab(),
7159 ResponseCache::new(1000, Duration::from_secs(3600)),
7160 ));
7161 let app = test_app_with_state(Arc::clone(&state));
7162 let (token, _guard) = state.cancels.register("chatcmpl-live");
7163
7164 let (status, before) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
7165 assert_eq!(status, StatusCode::OK);
7166 assert_eq!(before["generating_now"], serde_json::json!(1));
7167
7168 let (status, body) = post_json_uri(
7169 &app,
7170 frink_api::routes::V1_CANCEL,
7171 serde_json::json!({ "request_id": "chatcmpl-live" }),
7172 )
7173 .await;
7174 assert_eq!(status, StatusCode::OK);
7175 assert_eq!(body["cancelled"], serde_json::json!(true));
7176 assert!(
7177 token.is_cancelled(),
7178 "the endpoint answered ok without setting the flag the decode loop reads"
7179 );
7180 }
7181
7182 #[tokio::test]
7183 async fn tokenize_detokenize_roundtrip_and_embeddings_mean() {
7184 let app = test_app();
7185 let (status, tok) =
7186 post_json_uri(&app, "/v1/tokenize", serde_json::json!({ "prompt": "Hi" })).await;
7187 assert_eq!(status, StatusCode::OK);
7188 let tokens = tok["tokens"].as_array().unwrap();
7189 assert_eq!(tok["count"], tokens.len());
7190 assert!(!tokens.is_empty());
7191
7192 let (status, detok) = post_json_uri(
7193 &app,
7194 "/v1/detokenize",
7195 serde_json::json!({ "tokens": tokens }),
7196 )
7197 .await;
7198 assert_eq!(status, StatusCode::OK);
7199 assert_eq!(detok["text"], "Hi");
7200
7201 let (status, emb) = post_json_uri(
7202 &app,
7203 "/v1/embeddings",
7204 serde_json::json!({
7205 "input": "Hi",
7206 "embedding_type": "mean"
7207 }),
7208 )
7209 .await;
7210 assert_eq!(status, StatusCode::OK);
7211 let vec = emb["data"][0]["embedding"].as_array().unwrap();
7212 assert!(!vec.is_empty());
7213 assert!(vec.iter().all(|v| v.as_f64().is_some()));
7214 }
7215
7216 #[tokio::test]
7222 async fn the_decoder_path_still_refuses_a_pooling_it_cannot_mean() {
7223 let app = test_app();
7224 let (status, body) = post_json_uri(
7225 &app,
7226 "/v1/embeddings",
7227 serde_json::json!({ "input": "Hi", "embedding_type": "cls" }),
7228 )
7229 .await;
7230 assert_eq!(status, StatusCode::BAD_REQUEST);
7231 let msg = body["error"]["message"].as_str().unwrap();
7232 assert!(msg.contains("mean") && msg.contains("last"), "{msg}");
7233 }
7234
7235 #[tokio::test]
7240 #[ignore = "needs models/bge-small-en-v1.5-q8_0.gguf"]
7241 async fn a_real_embedding_model_serves_v1_embeddings() {
7242 let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
7243 .join("../../models/bge-small-en-v1.5-q8_0.gguf");
7244 if !path.exists() {
7245 eprintln!("SKIP: {} not present", path.display());
7246 return;
7247 }
7248 let encoder = frink_models::EmbeddingModel::from_gguf_path(&path).expect("load bge");
7249 let mut state = test_state(
7250 test_model_full_byte_vocab(),
7251 ResponseCache::new(1000, Duration::from_secs(3600)),
7252 );
7253 state.embedding = Some(Arc::new(encoder));
7254 let app = test_app_with_state(Arc::new(state));
7255
7256 let (status, body) = post_json_uri(
7257 &app,
7258 "/v1/embeddings",
7259 serde_json::json!({ "input": ["Hello world", "a second input"] }),
7260 )
7261 .await;
7262 assert_eq!(status, StatusCode::OK, "{body}");
7263 assert_eq!(body["model"], "bge-small-en-v1.5");
7264 let data = body["data"].as_array().unwrap();
7265 assert_eq!(data.len(), 2);
7266 for (i, row) in data.iter().enumerate() {
7267 assert_eq!(row["index"], i);
7268 let v: Vec<f64> = row["embedding"]
7269 .as_array()
7270 .unwrap()
7271 .iter()
7272 .map(|x| x.as_f64().unwrap())
7273 .collect();
7274 assert_eq!(v.len(), 384, "the encoder\'s width, not the decoder\'s");
7275 let norm = v.iter().map(|x| x * x).sum::<f64>().sqrt();
7276 assert!((norm - 1.0).abs() < 1e-4, "not L2-normalized: {norm}");
7277 }
7278 assert!(body["usage"]["prompt_tokens"].as_u64().unwrap() >= 4 + 2);
7281
7282 let (status, mean) = post_json_uri(
7286 &app,
7287 "/v1/embeddings",
7288 serde_json::json!({ "input": "Hello world", "embedding_type": "mean" }),
7289 )
7290 .await;
7291 assert_eq!(status, StatusCode::OK);
7292 assert_ne!(mean["data"][0]["embedding"], data[0]["embedding"]);
7293 }
7294
7295 #[tokio::test]
7307 #[ignore = "needs models/bge-small-en-v1.5-q8_0.gguf"]
7308 async fn an_encoder_can_be_the_loaded_model() {
7309 let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
7310 .join("../../models/bge-small-en-v1.5-q8_0.gguf");
7311 if !path.exists() {
7312 eprintln!("SKIP: {} not present", path.display());
7313 return;
7314 }
7315
7316 let loaded = model::load_from_path(path.to_str().unwrap()).expect("load bge as the model");
7320 assert!(
7321 matches!(loaded, model::LoadedModel::Encoder(_)),
7322 "an encoder-only GGUF reached a decoder loader"
7323 );
7324 let (loaded, batcher, ceiling) = activate_loaded_model(loaded, true, None, None);
7325 assert!(
7326 matches!(loaded, Loaded::Encoder(_)),
7327 "the encoder did not stay an encoder through activation"
7328 );
7329 assert!(
7330 batcher.is_none() && ceiling.is_none(),
7331 "an encoder was given a decode batcher or a KV ceiling it has no use for"
7332 );
7333
7334 let state = test_state(
7335 test_model_full_byte_vocab(),
7336 ResponseCache::new(1000, Duration::from_secs(3600)),
7337 );
7338 state.swap_active(Some(Arc::new(ActiveModel {
7339 id: None,
7340 loaded,
7341 batcher,
7342 ceiling,
7343 checkpoint_path: None,
7344 })));
7345 let app = test_app_with_state(Arc::new(state));
7346
7347 let (status, body) = post_json_uri(
7349 &app,
7350 "/v1/embeddings",
7351 serde_json::json!({ "input": "Hello world" }),
7352 )
7353 .await;
7354 assert_eq!(status, StatusCode::OK, "{body}");
7355 assert_eq!(body["model"], "bge-small-en-v1.5");
7356 let v = body["data"][0]["embedding"].as_array().unwrap();
7357 assert_eq!(v.len(), 384, "the encoder's width, not the decoder's");
7358
7359 let (status, body) = post_json_uri(
7361 &app,
7362 "/v1/chat/completions",
7363 serde_json::json!({
7364 "model": "bge-small-en-v1.5",
7365 "messages": [{"role": "user", "content": "hi"}],
7366 }),
7367 )
7368 .await;
7369 assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{body}");
7370 let msg = body["error"]["message"].as_str().unwrap();
7371 for fact in [
7372 "bge-small-en-v1.5",
7373 "bert",
7374 "embedding model",
7375 "/v1/embeddings",
7376 ] {
7377 assert!(msg.contains(fact), "the refusal does not say {fact}: {msg}");
7378 }
7379
7380 let (status, models) = get_json(&app, frink_api::routes::V1_MODELS).await;
7382 assert_eq!(status, StatusCode::OK);
7383 let entry = &models["data"][0];
7384 assert_eq!(entry["id"], "bge-small-en-v1.5");
7385 assert_eq!(entry["frink_model_kind"], "embedding");
7386 assert_eq!(entry["frink_tokenizer"], "gguf-wordpiece");
7387 assert_eq!(entry["frink_n_embd"], 384);
7388 assert_eq!(entry["frink_pooling"], "CLS");
7389 assert_eq!(
7390 entry["frink_endpoints"],
7391 serde_json::json!(["/v1/embeddings"])
7392 );
7393 assert!(entry.get("supported_reasoning_efforts").is_none());
7396
7397 let (status, health) = get_json(&app, frink_api::routes::HEALTH).await;
7399 assert_eq!(status, StatusCode::OK, "an encoder is a loaded model");
7400 assert_eq!(health["model"]["id"], "bge-small-en-v1.5");
7401 assert_eq!(health["model"]["synthetic_weights"], false);
7402 let weights = health["capabilities"]
7403 .as_array()
7404 .unwrap()
7405 .iter()
7406 .find(|c| c["id"] == frink_api::health::capability::REAL_WEIGHTS)
7407 .expect("a real-weights capability row");
7408 let detail = weights["detail"].as_str().unwrap_or_default();
7409 assert!(detail.contains("ENCODER"), "{detail}");
7410 let (status, body) = post_json_uri(
7418 &app,
7419 frink_api::routes::V1_TOKENIZE,
7420 serde_json::json!({ "content": "hello world" }),
7421 )
7422 .await;
7423 assert_eq!(
7424 status,
7425 StatusCode::OK,
7426 "an encoder has a real tokenizer: {body}"
7427 );
7428 let tokens = body["tokens"].as_array().expect("tokens array").clone();
7429 assert!(!tokens.is_empty(), "WordPiece produced nothing: {body}");
7430
7431 let (status, body) = post_json_uri(
7432 &app,
7433 frink_api::routes::V1_DETOKENIZE,
7434 serde_json::json!({ "tokens": tokens }),
7435 )
7436 .await;
7437 assert_eq!(status, StatusCode::OK, "{body}");
7438 let round_tripped = body["content"].as_str().expect("content").to_string();
7439 assert!(
7440 round_tripped.contains("hello") && round_tripped.contains("world"),
7441 "the ids did not decode back through the encoder's own vocabulary: {round_tripped}"
7442 );
7443
7444 let (status, _) = post_json_uri(
7447 &app,
7448 "/v1/completions",
7449 serde_json::json!({ "model": "m", "prompt": "hi", "max_tokens": 1 }),
7450 )
7451 .await;
7452 assert_eq!(
7453 status,
7454 StatusCode::NOT_IMPLEMENTED,
7455 "tokenizing an encoder must not have opened a path to generating with one"
7456 );
7457 }
7458
7459 #[tokio::test]
7464 async fn metrics_exposes_expert_store_counters_when_streaming_is_active() {
7465 use http_body_util::BodyExt;
7466 use tower::ServiceExt;
7467
7468 let fixture = concat!(
7469 "../frink-models/tests/fixtures/",
7470 "frink_real_moe_test.gguf"
7471 );
7472 let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join(fixture);
7473 let decoder = Decoder::from_gguf_with_expert_cache(
7474 &fixture,
7475 frink_models::config::test_moe_fixture(),
7476 Some(1024 * 1024),
7477 )
7478 .expect("MoE fixture must load store-backed");
7479
7480 let mut caches: Vec<frink_core::cache::KvCache> = decoder.config.new_kv_caches();
7484 decoder.forward_token(1, 0, &mut caches);
7485
7486 let model = Model::Gguf(GgufModel {
7487 decoder: Arc::new(decoder),
7488 tokenizer: Arc::new(ServerTokenizer::Byte),
7489 stop_tokens: StopTokens::default(),
7490 bos_id: None,
7491 is_synthetic: false,
7492 chat_template: chat_template::PromptTemplate::plain(),
7493 });
7494 let state = Arc::new(test_state(
7495 model,
7496 ResponseCache::new(16, Duration::from_secs(60)),
7497 ));
7498 let app = Router::new()
7499 .route("/metrics", axum::routing::get(metrics))
7500 .route("/v1/chat/completions", post(chat_completions))
7501 .with_state(state);
7502
7503 let fetch_metrics = |app: Router| async move {
7504 let resp = app
7505 .oneshot(
7506 axum::http::Request::builder()
7507 .method("GET")
7508 .uri("/metrics")
7509 .body(axum::body::Body::empty())
7510 .unwrap(),
7511 )
7512 .await
7513 .unwrap();
7514 let bytes = resp.into_body().collect().await.unwrap().to_bytes();
7515 String::from_utf8(bytes.to_vec()).unwrap()
7516 };
7517
7518 let after = fetch_metrics(app.clone()).await;
7519 assert!(
7520 after.contains("frink_expert_cache_misses_total"),
7521 "streaming model must expose expert-cache metrics: {after}"
7522 );
7523 let misses: u64 = after
7524 .lines()
7525 .find(|l| l.starts_with("frink_expert_cache_misses_total"))
7526 .and_then(|l| l.split_whitespace().nth(1))
7527 .and_then(|v| v.parse().ok())
7528 .expect("misses metric line must parse");
7529 assert!(
7530 misses > 0,
7531 "decode must have read experts through the store: {after}"
7532 );
7533 }
7534
7535 fn weather_tool() -> serde_json::Value {
7536 serde_json::json!({
7537 "type": "function",
7538 "function": {
7539 "name": "get_weather",
7540 "description": "Get the current weather for a location.",
7541 "parameters": {
7542 "type": "object",
7543 "properties": {"location": {"type": "string"}},
7544 "required": ["location"]
7545 }
7546 }
7547 })
7548 }
7549
7550 fn weather_tool_def() -> ToolDef {
7551 ToolDef {
7552 kind: "function".to_string(),
7553 function: ToolFunctionDef {
7554 name: "get_weather".to_string(),
7555 description: Some("Get the current weather for a location.".to_string()),
7556 parameters: Some(serde_json::json!({
7557 "type": "object",
7558 "properties": {"location": {"type": "string"}},
7559 "required": ["location"]
7560 })),
7561 },
7562 }
7563 }
7564
7565 #[test]
7566 fn tool_preamble_mentions_every_tool_name_and_description() {
7567 let preamble = tool_preamble(&[weather_tool_def()]);
7568 assert!(preamble.contains("get_weather"));
7569 assert!(preamble.contains("Get the current weather for a location."));
7570 assert!(preamble.contains("<tool_call>"));
7571 assert!(preamble.contains("</tool_call>"));
7572 }
7573
7574 #[test]
7575 fn a_real_marker_becomes_a_structured_tool_call() {
7576 let text = "sure, let me check.<tool_call>{\"name\": \"get_weather\", \"arguments\": {\"location\": \"Paris\"}}</tool_call>";
7577 let (message, finish) = build_response_message(
7578 text.to_string(),
7579 &[weather_tool_def()],
7580 output::OutputPosture::for_model("test-model"),
7581 "stop",
7582 );
7583 assert_eq!(finish, "tool_calls");
7584 let calls = message.tool_calls.expect("must carry a tool call");
7585 assert_eq!(calls[0].function.name, "get_weather");
7586 let parsed: serde_json::Value = serde_json::from_str(&calls[0].function.arguments).unwrap();
7587 assert_eq!(parsed["location"], "Paris");
7588 }
7589
7590 #[test]
7591 fn a_plain_answer_is_not_promoted_to_a_tool_call() {
7592 let (message, finish) = build_response_message(
7593 "just an answer".to_string(),
7594 &[weather_tool_def()],
7595 output::OutputPosture::for_model("test-model"),
7596 "stop",
7597 );
7598 assert_eq!(finish, "stop");
7599 assert!(message.tool_calls.is_none());
7600 assert_eq!(message.content.as_deref(), Some("just an answer"));
7601 }
7602
7603 #[test]
7606 fn a_malformed_payload_is_not_a_tool_call() {
7607 let (message, finish) = build_response_message(
7608 "<tool_call>not valid json at all</tool_call>".to_string(),
7609 &[weather_tool_def()],
7610 output::OutputPosture::for_model("test-model"),
7611 "stop",
7612 );
7613 assert_eq!(finish, "stop");
7614 assert!(message.tool_calls.is_none());
7615 }
7616
7617 #[test]
7620 fn a_tool_that_was_never_offered_is_not_returned() {
7621 let (message, finish) = build_response_message(
7622 "<tool_call>{\"name\": \"ping\", \"arguments\": {}}</tool_call>".to_string(),
7623 &[weather_tool_def()],
7624 output::OutputPosture::for_model("test-model"),
7625 "stop",
7626 );
7627 assert_eq!(finish, "stop");
7628 assert!(message.tool_calls.is_none());
7629 }
7630
7631 #[test]
7633 fn marker_text_with_no_tools_offered_stays_content() {
7634 let (message, finish) = build_response_message(
7635 "<tool_call>{\"name\": \"get_weather\", \"arguments\": {}}</tool_call>".to_string(),
7636 &[],
7637 output::OutputPosture::for_model("test-model"),
7638 "stop",
7639 );
7640 assert_eq!(finish, "stop");
7641 assert!(message.tool_calls.is_none());
7642 assert!(message.content.is_some());
7643 }
7644
7645 #[test]
7649 fn a_streamed_call_opens_then_delivers_its_arguments_in_pieces() {
7650 let opened = std::cell::Cell::new(0usize);
7651 let mut parser = crate::policy::parser::ToolCallParser::new(
7652 crate::policy::parser::ToolCallFormat::Qwen3Coder,
7653 vec![
7654 crate::policy::parser::tool_call::ToolSchema::with_parameters(
7655 "write_file",
7656 serde_json::json!({"type": "object", "properties": {
7657 "path": {"type": "string"},
7658 "contents": {"type": "string"}
7659 }}),
7660 ),
7661 ],
7662 );
7663 let wire = "<tool_call><function=write_file>\
7664 <parameter=path>\n/tmp/x\n</parameter>\
7665 <parameter=contents>\nhello world\n</parameter>\
7666 </function></tool_call>";
7667
7668 let mut deltas = Vec::new();
7669 let mut text = String::new();
7670 for piece in wire.as_bytes().chunks(7) {
7671 let chunk = String::from_utf8_lossy(piece).into_owned();
7672 let (more_text, more) = tool_call_deltas(parser.push(&chunk), &opened);
7673 text.push_str(&more_text);
7674 deltas.extend(more);
7675 }
7676 let (more_text, more) = tool_call_deltas(parser.finish(), &opened);
7677 text.push_str(&more_text);
7678 deltas.extend(more);
7679
7680 assert_eq!(opened.get(), 1, "one call opened");
7681 assert!(text.is_empty(), "the markers are not content: {text:?}");
7682
7683 let first = &deltas[0];
7684 assert_eq!(first.index, 0);
7685 assert_eq!(first.id.as_deref(), Some("call_0"));
7686 assert_eq!(first.kind, Some("function"));
7687 assert_eq!(first.function.name.as_deref(), Some("write_file"));
7688
7689 let joined: String = deltas
7692 .iter()
7693 .filter_map(|d| d.function.arguments.clone())
7694 .collect();
7695 let parsed: serde_json::Value =
7696 serde_json::from_str(&joined).expect("the fragments concatenate to valid JSON");
7697 assert_eq!(parsed["path"], serde_json::json!("/tmp/x"));
7698 assert_eq!(parsed["contents"], serde_json::json!("hello world"));
7699 assert!(
7700 deltas.len() >= 3,
7701 "the arguments arrived in pieces, not whole: {}",
7702 deltas.len()
7703 );
7704 assert!(
7705 deltas[1..].iter().all(|d| d.function.name.is_none()),
7706 "only the opening delta carries identity"
7707 );
7708 }
7709
7710 #[test]
7712 fn text_around_a_streamed_call_is_still_content() {
7713 let opened = std::cell::Cell::new(0usize);
7714 let mut parser = crate::policy::parser::ToolCallParser::new(
7715 crate::policy::parser::ToolCallFormat::Qwen25,
7716 vec![crate::policy::parser::tool_call::ToolSchema::new(
7717 "get_weather",
7718 )],
7719 );
7720 let wire = "let me check. <tool_call>{\"name\": \"get_weather\", \
7721 \"arguments\": {}}</tool_call> done";
7722 let mut text = String::new();
7723 for piece in wire.as_bytes().chunks(5) {
7724 let chunk = String::from_utf8_lossy(piece).into_owned();
7725 let (more, _) = tool_call_deltas(parser.push(&chunk), &opened);
7726 text.push_str(&more);
7727 }
7728 let (more, _) = tool_call_deltas(parser.finish(), &opened);
7729 text.push_str(&more);
7730
7731 assert_eq!(opened.get(), 1);
7732 assert!(text.starts_with("let me check. "), "{text:?}");
7733 assert!(text.ends_with(" done"), "{text:?}");
7734 assert!(!text.contains("<tool_call>"), "markers leaked: {text:?}");
7735 }
7736
7737 #[test]
7740 fn a_reasoning_block_is_split_out_of_the_answer() {
7741 let (message, finish) = build_response_message(
7742 "<think>weighing it up</think>The answer is 4.".to_string(),
7743 &[],
7744 output::OutputPosture::for_model("Qwen3-8B"),
7745 "stop",
7746 );
7747 assert_eq!(finish, "stop");
7748 assert_eq!(message.content.as_deref(), Some("The answer is 4."));
7749 assert_eq!(message.reasoning_content.as_deref(), Some("weighing it up"));
7750 }
7751
7752 #[test]
7755 fn a_non_reasoning_model_keeps_a_literal_marker_in_its_answer() {
7756 let (message, _) = build_response_message(
7757 "Use the <think> tag like this.".to_string(),
7758 &[],
7759 output::OutputPosture::for_model("llama-3.1-8b"),
7760 "stop",
7761 );
7762 assert_eq!(
7763 message.content.as_deref(),
7764 Some("Use the <think> tag like this.")
7765 );
7766 assert!(message.reasoning_content.is_none());
7767 }
7768
7769 #[tokio::test]
7777 async fn a_request_with_no_tools_or_session_behaves_exactly_as_before() {
7778 let app = test_app();
7779 let body = serde_json::json!({
7780 "model": "m",
7781 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7782 "max_tokens": 4,
7783 "temperature": 0,
7784 });
7785 let resp = post_json(&app, body).await;
7786 let message = &resp["choices"][0]["message"];
7787 assert!(message["content"].is_string());
7788 assert!(message.get("tool_calls").is_none());
7789 assert_eq!(resp["choices"][0]["finish_reason"], "length");
7790 assert_eq!(resp["usage"]["completion_tokens"], 4);
7791 assert_eq!(
7792 resp["usage"]["total_tokens"],
7793 resp["usage"]["prompt_tokens"].as_u64().unwrap() + 4
7794 );
7795 }
7796
7797 pub(crate) async fn get_json(app: &Router, uri: &str) -> (StatusCode, serde_json::Value) {
7798 use http_body_util::BodyExt;
7799 use tower::ServiceExt;
7800
7801 let response = app
7802 .clone()
7803 .oneshot(
7804 axum::http::Request::builder()
7805 .method("GET")
7806 .uri(uri)
7807 .body(axum::body::Body::empty())
7808 .unwrap(),
7809 )
7810 .await
7811 .unwrap();
7812 let status = response.status();
7813 let bytes = response.into_body().collect().await.unwrap().to_bytes();
7814 (status, serde_json::from_slice(&bytes).unwrap())
7815 }
7816
7817 #[tokio::test]
7818 async fn health_answers_a_capability_handshake_not_a_boolean() {
7819 let app = test_app();
7820 let (status, body) = get_json(&app, frink_api::routes::HEALTH).await;
7821 assert_eq!(status, StatusCode::OK);
7822
7823 let health: frink_api::HealthResponse = serde_json::from_value(body).unwrap();
7824 assert_eq!(health.state, frink_api::HealthState::Ready);
7825 assert!(health.pid > 0);
7826 assert!(health.server_time_unix_ms > 0);
7827 assert_eq!(health.last_request_age_seconds, None);
7830
7831 for id in [
7834 frink_api::health::capability::CPU,
7835 frink_api::health::capability::METAL,
7836 frink_api::health::capability::CUDA,
7837 frink_api::health::capability::REAL_WEIGHTS,
7838 frink_api::health::capability::CONTINUOUS_BATCHING,
7839 ] {
7840 let cap = health
7841 .capability(id)
7842 .unwrap_or_else(|| panic!("{id} missing"));
7843 assert!(!cap.reason.is_empty(), "{cap:?}");
7844 assert!(!cap.detail.is_empty(), "{cap:?}");
7845 }
7846 let weights = health
7850 .capability(frink_api::health::capability::REAL_WEIGHTS)
7851 .unwrap();
7852 assert!(!weights.available);
7853 assert_eq!(weights.reason, frink_api::health::reason::MODEL_NOT_LOADED);
7854 assert!(health.model.as_ref().unwrap().synthetic_weights);
7855 }
7856
7857 #[tokio::test]
7858 async fn health_vouches_for_liveness_after_a_request_has_been_served() {
7859 let app = test_app();
7860 let _ = post_json(
7861 &app,
7862 serde_json::json!({
7863 "model": "m",
7864 "messages": [{"role": "user", "content": "\u{1}"}],
7865 "max_tokens": 1,
7866 "temperature": 0,
7867 }),
7868 )
7869 .await;
7870 let (_status, body) = get_json(&app, frink_api::routes::HEALTH).await;
7871 let health: frink_api::HealthResponse = serde_json::from_value(body).unwrap();
7872 let age = health
7873 .last_request_age_seconds
7874 .expect("a served request is evidence of liveness");
7875 assert!((0.0..5.0).contains(&age), "implausible age {age}");
7876 }
7877
7878 async fn post_sse_chunks(app: &Router, body: serde_json::Value) -> Vec<serde_json::Value> {
7880 use http_body_util::BodyExt;
7881 use tower::ServiceExt;
7882
7883 let response = app
7884 .clone()
7885 .oneshot(
7886 axum::http::Request::builder()
7887 .method("POST")
7888 .uri("/v1/chat/completions")
7889 .header("content-type", "application/json")
7890 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
7891 .unwrap(),
7892 )
7893 .await
7894 .unwrap();
7895 let bytes = response.into_body().collect().await.unwrap().to_bytes();
7896 String::from_utf8(bytes.to_vec())
7897 .unwrap()
7898 .lines()
7899 .filter_map(|line| line.strip_prefix("data: "))
7900 .filter(|payload| *payload != "[DONE]")
7901 .map(|payload| serde_json::from_str(payload).unwrap())
7902 .collect()
7903 }
7904
7905 #[tokio::test]
7906 async fn a_stream_states_its_request_id_once_in_the_first_chunk() {
7907 let app = test_app();
7908 let chunks = post_sse_chunks(
7909 &app,
7910 serde_json::json!({
7911 "model": "m",
7912 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7913 "max_tokens": 4,
7914 "temperature": 0,
7915 "stream": true,
7916 }),
7917 )
7918 .await;
7919
7920 assert!(!chunks.is_empty());
7921 let request_id = chunks[0]["request_id"]
7922 .as_str()
7923 .expect("the first chunk names the request")
7924 .to_string();
7925 assert!(request_id.starts_with("chatcmpl-"), "{request_id}");
7926 for (i, chunk) in chunks.iter().enumerate().skip(1) {
7929 assert!(
7930 chunk.get("request_id").is_none(),
7931 "chunk {i} repeats request_id"
7932 );
7933 }
7934 for chunk in &chunks {
7937 assert_eq!(chunk["id"], serde_json::json!(request_id));
7938 }
7939
7940 let other = post_sse_chunks(
7941 &app,
7942 serde_json::json!({
7943 "model": "m",
7944 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7945 "max_tokens": 4,
7946 "temperature": 0,
7947 "stream": true,
7948 }),
7949 )
7950 .await;
7951 assert_ne!(
7952 other[0]["request_id"].as_str().unwrap(),
7953 request_id,
7954 "two concurrent chats must not share an id"
7955 );
7956 }
7957
7958 #[tokio::test]
7959 async fn a_non_streamed_response_names_the_same_request_id_as_its_completion_id() {
7960 let app = test_app();
7961 let resp = post_json(
7962 &app,
7963 serde_json::json!({
7964 "model": "m",
7965 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7966 "max_tokens": 2,
7967 "temperature": 0,
7968 }),
7969 )
7970 .await;
7971 assert_eq!(resp["id"], resp["request_id"]);
7972 assert!(resp["request_id"]
7973 .as_str()
7974 .unwrap()
7975 .starts_with("chatcmpl-"));
7976 }
7977
7978 #[tokio::test]
7981 async fn usage_carries_separate_prefill_and_decode_timings() {
7982 let app = test_app();
7983 let resp = post_json(
7984 &app,
7985 serde_json::json!({
7986 "model": "m",
7987 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7988 "max_tokens": 4,
7989 "temperature": 0,
7990 }),
7991 )
7992 .await;
7993 let usage = &resp["usage"];
7994 assert!(usage["prompt_eval_duration_ms"].is_number(), "{usage}");
7995 assert!(usage["generation_duration_ms"].is_number(), "{usage}");
7996 assert!(usage["time_to_first_token_ms"].is_number(), "{usage}");
7997 assert!(usage["predicted_per_second"].is_number(), "{usage}");
7998 assert!(usage.get("cached_tokens").is_none(), "{usage}");
8000 }
8001
8002 #[tokio::test]
8010 async fn a_tools_request_with_no_marker_in_the_output_falls_back_to_plain_content() {
8011 let app = test_app();
8012 let body = serde_json::json!({
8013 "model": "m",
8014 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
8015 "max_tokens": 4,
8016 "temperature": 0,
8017 "tools": [weather_tool()],
8018 });
8019 let resp = post_json(&app, body).await;
8020 let message = &resp["choices"][0]["message"];
8021 assert!(
8022 message["content"].is_string(),
8023 "must fall back to plain content when no real tool-call marker is present: {resp:?}"
8024 );
8025 assert!(message.get("tool_calls").is_none());
8026 assert_eq!(resp["choices"][0]["finish_reason"], "length");
8030 }
8031
8032 #[tokio::test]
8036 async fn a_cache_hit_reports_the_original_finish_reason_and_usage() {
8037 let app = test_app();
8038 let body = serde_json::json!({
8039 "model": "m",
8040 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
8041 "max_tokens": 3,
8042 "temperature": 0,
8043 });
8044 let first = post_json(&app, body.clone()).await;
8045 assert_eq!(first["frink_cache"], "miss");
8046 let second = post_json(&app, body).await;
8047 assert_eq!(second["frink_cache"], "hit");
8048 assert_eq!(
8049 first["choices"][0]["message"]["content"],
8050 second["choices"][0]["message"]["content"]
8051 );
8052 assert_eq!(
8053 first["choices"][0]["finish_reason"],
8054 second["choices"][0]["finish_reason"]
8055 );
8056 assert_eq!(first["usage"], second["usage"]);
8057 assert_eq!(second["usage"]["completion_tokens"], 3);
8058 }
8059
8060 #[tokio::test]
8071 async fn a_grammar_request_is_not_answered_from_an_unconstrained_cache_entry() {
8072 let app = test_app();
8073 let plain = serde_json::json!({
8074 "model": "m",
8075 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
8076 "max_tokens": 3,
8077 "temperature": 0,
8078 });
8079
8080 let first = post_json(&app, plain.clone()).await;
8081 assert_eq!(first["frink_cache"], "miss");
8082 let unconstrained = first["choices"][0]["message"]["content"]
8083 .as_str()
8084 .expect("content")
8085 .to_string();
8086
8087 let mut constrained = plain.clone();
8088 constrained["grammar"] = serde_json::json!("root ::= \"yes\"");
8089 let second = post_json(&app, constrained).await;
8090 assert_eq!(
8091 second["frink_cache"], "miss",
8092 "a grammar is part of the key, so this body has never been answered"
8093 );
8094 let constrained_answer = second["choices"][0]["message"]["content"]
8098 .as_str()
8099 .expect("content")
8100 .to_string();
8101 assert!(
8102 constrained_answer.contains("-> \"yes\"]"),
8103 "the grammar must have been compiled AND applied, not skipped \
8104 by a cache hit: {constrained_answer}"
8105 );
8106 assert_ne!(
8107 constrained_answer, unconstrained,
8108 "the constrained request was served the unconstrained answer"
8109 );
8110
8111 let third = post_json(&app, plain).await;
8115 assert_eq!(third["frink_cache"], "hit");
8116 assert_eq!(third["choices"][0]["message"]["content"], unconstrained);
8117 }
8118
8119 #[tokio::test]
8141 async fn a_json_object_request_does_not_reuse_the_unconstrained_cache_entry() {
8142 let state = Arc::new(test_state(
8143 test_model_full_byte_vocab(),
8144 ResponseCache::new(1000, Duration::from_secs(3600)),
8145 ));
8146 let app = test_app_with_state(state.clone());
8147 let plain = serde_json::json!({
8148 "model": "m",
8149 "messages": [
8150 {"role": "system", "content": "Answer in JSON when it helps."},
8151 {"role": "user", "content": "\u{1}\u{2}"},
8152 ],
8153 "max_tokens": 3,
8154 "temperature": 0,
8155 });
8156
8157 let first = post_json(&app, plain.clone()).await;
8158 assert_eq!(first["frink_cache"], "miss");
8159 assert_eq!(state.cache_stats().entries, 1);
8160
8161 let mut as_json = plain.clone();
8162 as_json["response_format"] = serde_json::json!({"type": "json_object"});
8163 let (status, _) = post_json_uri(&app, "/v1/chat/completions", as_json).await;
8164 assert_eq!(
8165 status,
8166 StatusCode::BAD_REQUEST,
8167 "the demo banner is not a JSON object, whoever generated it"
8168 );
8169 assert_eq!(
8170 state.cache_stats().hits,
8171 0,
8172 "a json_object request must not be answered from an entry the \
8173 JSON mask never produced"
8174 );
8175 assert_eq!(
8176 state.cache_stats().entries,
8177 2,
8178 "json_object must key its own entry, not reuse the unconstrained \
8179 one it happens to render the same prompt as"
8180 );
8181 }
8182
8183 #[tokio::test]
8195 async fn an_ignore_eos_request_is_not_answered_from_a_cache_entry_that_stopped_at_eos() {
8196 let app = test_app_with_state(Arc::new(test_state(
8197 test_model_full_byte_vocab_with_eos(Some(0x77)),
8198 ResponseCache::new(1000, Duration::from_secs(3600)),
8199 )));
8200 let body = serde_json::json!({
8201 "model": "m",
8202 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
8203 "max_tokens": 6,
8204 "temperature": 0,
8205 });
8206
8207 let stopped = post_json(&app, body.clone()).await;
8208 assert_eq!(stopped["frink_cache"], "miss");
8209 assert_eq!(
8210 stopped["choices"][0]["finish_reason"], "stop",
8211 "the fixture is only meaningful if the model's EOS really fires here"
8212 );
8213 assert_eq!(stopped["usage"]["completion_tokens"], 1);
8214
8215 let mut ignoring = body.clone();
8216 ignoring["ignore_eos"] = serde_json::json!(true);
8217 let ran_on = post_json(&app, ignoring).await;
8218 assert_eq!(
8219 ran_on["frink_cache"], "miss",
8220 "ignore_eos is part of the key, so this body has never been answered"
8221 );
8222 assert_eq!(
8223 ran_on["usage"]["completion_tokens"], 6,
8224 "ignore_eos must run the full budget, not replay the EOS-terminated answer"
8225 );
8226 assert_eq!(ran_on["choices"][0]["finish_reason"], "length");
8227 assert_ne!(
8228 ran_on["choices"][0]["message"]["content"],
8229 stopped["choices"][0]["message"]["content"]
8230 );
8231 }
8232
8233 #[tokio::test]
8239 async fn session_reuse_produces_the_same_output_as_manually_resending_full_history() {
8240 let session_app = test_app();
8241 let manual_app = test_app();
8242
8243 let turn1 = post_json(
8245 &session_app,
8246 serde_json::json!({
8247 "model": "m",
8248 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
8249 "session_id": "s1",
8250 "max_tokens": 5,
8251 "temperature": 0,
8252 }),
8253 )
8254 .await;
8255 let reply1 = turn1["choices"][0]["message"]["content"]
8256 .as_str()
8257 .unwrap()
8258 .to_string();
8259
8260 let manual_turn1 = post_json(
8265 &manual_app,
8266 serde_json::json!({
8267 "model": "m",
8268 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
8269 "max_tokens": 5,
8270 "temperature": 0,
8271 }),
8272 )
8273 .await;
8274 assert_eq!(
8275 manual_turn1["choices"][0]["message"]["content"]
8276 .as_str()
8277 .unwrap(),
8278 reply1
8279 );
8280
8281 let turn2 = post_json(
8283 &session_app,
8284 serde_json::json!({
8285 "model": "m",
8286 "messages": [{"role": "user", "content": "\u{4}\u{5}"}],
8287 "session_id": "s1",
8288 "max_tokens": 5,
8289 "temperature": 0,
8290 }),
8291 )
8292 .await;
8293 let reply2 = turn2["choices"][0]["message"]["content"]
8294 .as_str()
8295 .unwrap()
8296 .to_string();
8297
8298 let manual_turn2 = post_json(
8302 &manual_app,
8303 serde_json::json!({
8304 "model": "m",
8305 "messages": [
8306 {"role": "user", "content": "\u{1}\u{2}\u{3}"},
8307 {"role": "assistant", "content": reply1},
8308 {"role": "user", "content": "\u{4}\u{5}"},
8309 ],
8310 "max_tokens": 5,
8311 "temperature": 0,
8312 }),
8313 )
8314 .await;
8315 assert_eq!(
8316 manual_turn2["choices"][0]["message"]["content"]
8317 .as_str()
8318 .unwrap(),
8319 reply2,
8320 "resuming a session must produce identical output to manually resending the full history"
8321 );
8322 }
8323
8324 #[test]
8327 fn lock_cache_recovers_from_a_poisoned_mutex() {
8328 let cache = Arc::new(Mutex::new(ResponseCache::new(10, Duration::from_secs(60))));
8329
8330 let poison_cache = Arc::clone(&cache);
8331 let _ = std::thread::spawn(move || {
8332 let _guard = poison_cache.lock().unwrap();
8333 panic!("simulated panic while holding the lock");
8334 })
8335 .join();
8336
8337 let recovered = lock_cache(&cache);
8339 assert_eq!(recovered.stats().entries, 0);
8340 }
8341
8342 #[test]
8343 fn is_cacheable_true_for_greedy_or_seeded_requests() {
8344 let mut req_body = serde_json::json!({
8345 "model": "m",
8346 "messages": [{"role": "user", "content": "hi"}],
8347 });
8348 let req: ChatCompletionRequest = serde_json::from_value(req_body.clone()).unwrap();
8349 assert!(
8350 req.is_cacheable(),
8351 "default (temperature 0) must be cacheable"
8352 );
8353
8354 req_body["temperature"] = serde_json::json!(0.8);
8355 let req: ChatCompletionRequest = serde_json::from_value(req_body.clone()).unwrap();
8356 assert!(
8357 !req.is_cacheable(),
8358 "unseeded sampling must never be cacheable"
8359 );
8360
8361 req_body["seed"] = serde_json::json!(42);
8362 let req: ChatCompletionRequest = serde_json::from_value(req_body).unwrap();
8363 assert!(
8364 req.is_cacheable(),
8365 "sampling with an explicit seed is deterministic and must be cacheable"
8366 );
8367 }
8368
8369 const GRADED: &str = "{% if reasoning_effort %}\
8373 {% if reasoning_effort not in ['low','medium','high'] %}\
8374 {{ raise_exception('unsupported effort') }}\
8375 {% endif %}E:{{ reasoning_effort }}|{% endif %}\
8376 {% if enable_thinking %}THINK|{% endif %}{{ messages[0].content }}";
8377
8378 fn graded_template() -> chat_template::PromptTemplate {
8379 chat_template::PromptTemplate::from_gguf_metadata(
8380 Some(GRADED),
8381 Some("qwen3"),
8382 false,
8383 true,
8384 None,
8385 None,
8386 )
8387 }
8388
8389 fn chat_request(value: serde_json::Value) -> ChatCompletionRequest {
8390 serde_json::from_value(value).expect("request")
8391 }
8392
8393 #[test]
8400 fn a_grammar_on_the_chat_wire_reaches_the_generation_params() {
8401 let req = chat_request(serde_json::json!({
8402 "model": "m",
8403 "messages": [{"role": "user", "content": "hi"}],
8404 "grammar": "root ::= \"a\"+",
8405 }));
8406 req.validate_supported_fields()
8407 .expect("a valid grammar is a valid request");
8408 let params = req
8409 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8410 .expect("a valid grammar compiles at params time too");
8411 assert!(
8412 params.grammar.is_some(),
8413 "the grammar was dropped between the wire and the sampler"
8414 );
8415 assert!(
8416 params.needs_vocab_logits(),
8417 "a grammar request that may fold lm_head into a GPU argmax is \
8418 a grammar request served unconstrained"
8419 );
8420
8421 let plain = chat_request(serde_json::json!({
8422 "model": "m",
8423 "messages": [{"role": "user", "content": "hi"}],
8424 }));
8425 assert!(plain
8426 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8427 .unwrap()
8428 .grammar
8429 .is_none());
8430 }
8431
8432 fn tool_request(tool_choice: serde_json::Value) -> ChatCompletionRequest {
8433 chat_request(serde_json::json!({
8434 "model": "m",
8435 "messages": [{"role": "user", "content": "weather in Rome?"}],
8436 "tools": [weather_tool()],
8437 "tool_choice": tool_choice,
8438 }))
8439 }
8440
8441 #[test]
8445 fn a_forced_tool_choice_puts_a_grammar_on_the_generation_params() {
8446 for choice in [
8447 serde_json::json!("required"),
8448 serde_json::json!({"type": "function", "function": {"name": "get_weather"}}),
8449 ] {
8450 let req = tool_request(choice.clone());
8451 req.validate_supported_fields()
8452 .unwrap_or_else(|e| panic!("{choice} is a valid request: {e:?}"));
8453 let params = req
8454 .generation_params_for_template(
8455 &graded_template(),
8456 "Qwen3-8B",
8457 crate::sampling_knobs::SamplerModel::absent(),
8458 )
8459 .unwrap_or_else(|e| panic!("{choice} compiles: {e:?}"));
8460 let grammar = params
8461 .grammar
8462 .as_ref()
8463 .unwrap_or_else(|| panic!("{choice} was accepted and then not enforced"));
8464 assert!(
8465 grammar.is_awaiting_trigger(),
8466 "the model must be free to think before it calls"
8467 );
8468 assert!(
8469 !grammar.allows_eog(),
8470 "{choice} must not be able to end the turn without a call"
8471 );
8472 assert!(
8478 params.needs_vocab_logits(),
8479 "{choice} would let a backend return a token id instead of logits"
8480 );
8481 assert!(
8482 !generate::greedy_gpu_fold_allowed(¶ms),
8483 "{choice} at temperature 0 must still refuse the greedy GPU fold"
8484 );
8485 }
8486 }
8487
8488 #[test]
8490 fn an_unforced_tool_choice_leaves_the_generation_unconstrained() {
8491 for choice in [serde_json::json!("auto"), serde_json::json!("none")] {
8492 let req = tool_request(choice.clone());
8493 req.validate_supported_fields().expect("still supported");
8494 let params = match req.generation_params_for_template(
8495 &graded_template(),
8496 "Qwen3-8B",
8497 crate::sampling_knobs::SamplerModel::absent(),
8498 ) {
8499 Ok(p) => p,
8500 Err((status, _)) => panic!("{choice} has no constraint to compile: {status}"),
8501 };
8502 assert!(
8503 params.grammar.is_none(),
8504 "{choice} does not force a call and must not be constrained"
8505 );
8506 }
8507 }
8508
8509 #[test]
8512 fn a_forced_tool_choice_refuses_rather_than_quietly_not_forcing() {
8513 let req = chat_request(serde_json::json!({
8515 "model": "m",
8516 "messages": [{"role": "user", "content": "hi"}],
8517 "tool_choice": "required",
8518 }));
8519 let (status, _) = req
8520 .validate_supported_fields()
8521 .expect_err("nothing to call");
8522 assert_eq!(status, StatusCode::BAD_REQUEST);
8523
8524 let req =
8526 tool_request(serde_json::json!({"type": "function", "function": {"name": "nope"}}));
8527 let (status, Json(body)) = req.validate_supported_fields().expect_err("no such tool");
8528 assert_eq!(status, StatusCode::BAD_REQUEST);
8529 assert_eq!(body["error"]["param"], "tool_choice");
8530
8531 let req = tool_request(serde_json::json!({"type": "function"}));
8533 let (status, _) = req.validate_supported_fields().expect_err("names nothing");
8534 assert_eq!(status, StatusCode::BAD_REQUEST);
8535
8536 let req = chat_request(serde_json::json!({
8538 "model": "m",
8539 "messages": [{"role": "user", "content": "hi"}],
8540 "tools": [weather_tool()],
8541 "tool_choice": "required",
8542 "grammar": "root ::= \"a\"+",
8543 }));
8544 let (status, _) = req
8545 .validate_supported_fields()
8546 .expect_err("a grammar and a forced call are two constraints");
8547 assert_eq!(status, StatusCode::BAD_REQUEST);
8548
8549 let req = tool_request(serde_json::json!("required"));
8555 let (status, Json(body)) = match req.generation_params_for_template(
8556 &graded_template(),
8557 "muse-glimmer-8b",
8558 crate::sampling_knobs::SamplerModel::absent(),
8559 ) {
8560 Err(e) => e,
8561 Ok(_) => panic!("a muse_glimmer call's boundary is a channel, not a marker"),
8562 };
8563 assert_eq!(status, StatusCode::NOT_IMPLEMENTED);
8564 assert!(
8565 body["error"]["message"]
8566 .as_str()
8567 .unwrap()
8568 .contains("muse_glimmer"),
8569 "{body}"
8570 );
8571
8572 let req = tool_request(serde_json::json!("required"));
8578 let params = req
8579 .generation_params_for_template(
8580 &graded_template(),
8581 "gemma-4-E2B-it",
8582 crate::sampling_knobs::SamplerModel::absent(),
8583 )
8584 .expect("a gemma4 forced tool_choice is served");
8585 assert!(
8586 params.grammar.is_some(),
8587 "a forced tool_choice must arrive as the generation's grammar"
8588 );
8589 }
8590
8591 #[test]
8594 fn an_unparseable_grammar_on_the_chat_wire_is_a_400() {
8595 let req = chat_request(serde_json::json!({
8596 "model": "m",
8597 "messages": [{"role": "user", "content": "hi"}],
8598 "grammar": "root ::= \"a",
8599 }));
8600 let (status, Json(body)) = req
8601 .validate_supported_fields()
8602 .expect_err("this does not parse");
8603 assert_eq!(status, StatusCode::BAD_REQUEST);
8604 assert_eq!(body["error"]["param"], "grammar");
8605 assert!(
8606 req.generation_params(crate::sampling_knobs::SamplerModel::absent())
8607 .is_err(),
8608 "and again at params time"
8609 );
8610 }
8611
8612 #[test]
8619 fn response_format_json_schema_becomes_the_requests_grammar() {
8620 let req = chat_request(serde_json::json!({
8621 "model": "m",
8622 "messages": [{"role": "user", "content": "hi"}],
8623 "response_format": {
8624 "type": "json_schema",
8625 "json_schema": {"name": "x", "schema": {"type": "boolean"}},
8626 },
8627 }));
8628 req.validate_supported_fields()
8629 .expect("a boolean schema converts");
8630 let params = req
8631 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8632 .expect("and compiles");
8633 let grammar = params.grammar.expect("the schema is the grammar");
8634 let mut g = (*grammar).clone();
8635 g.accept_token(0, b"true").expect("a boolean is accepted");
8636 assert!(g.allows_eog(), "and completes the parse");
8637 assert!(
8638 !params.json_object,
8639 "a schema is not the json_object character-class mask"
8640 );
8641 }
8642
8643 #[test]
8647 fn an_unconvertible_response_format_schema_is_a_400_naming_the_keyword() {
8648 let req = chat_request(serde_json::json!({
8649 "model": "m",
8650 "messages": [{"role": "user", "content": "hi"}],
8651 "response_format": {
8652 "type": "json_schema",
8653 "json_schema": {"name": "x", "schema": {"type": "integer", "minimum": 3}},
8654 },
8655 }));
8656 let (status, Json(body)) = req
8657 .validate_supported_fields()
8658 .expect_err("minimum has no grammar in this port");
8659 assert_eq!(status, StatusCode::BAD_REQUEST);
8660 assert!(
8661 body["error"]["message"]
8662 .as_str()
8663 .expect("a message")
8664 .contains("minimum"),
8665 "the refusal must name the keyword: {body}"
8666 );
8667 assert!(
8668 req.generation_params(crate::sampling_knobs::SamplerModel::absent())
8669 .is_err(),
8670 "and again at params time"
8671 );
8672 }
8673
8674 #[test]
8680 fn a_forced_tool_choice_and_a_schema_are_two_constraints() {
8681 let req = chat_request(serde_json::json!({
8682 "model": "m",
8683 "messages": [{"role": "user", "content": "hi"}],
8684 "tool_choice": "required",
8685 "tools": [{
8686 "type": "function",
8687 "function": {"name": "f", "parameters": {"type": "object"}},
8688 }],
8689 "response_format": {
8690 "type": "json_schema",
8691 "json_schema": {"name": "x", "schema": {"type": "boolean"}},
8692 },
8693 }));
8694 let (status, Json(body)) = req
8695 .validate_supported_fields()
8696 .expect_err("two constraints, one generation");
8697 assert_eq!(status, StatusCode::BAD_REQUEST);
8698 assert_eq!(body["error"]["param"], "tool_choice");
8699 }
8700
8701 #[test]
8704 fn an_omitted_output_budget_is_a_whole_answer_not_sixteen_tokens() {
8705 let req = chat_request(serde_json::json!({
8706 "model": "m",
8707 "messages": [{"role": "user", "content": "hi"}],
8708 }));
8709 assert_eq!(req.max_tokens, DEFAULT_CHAT_MAX_TOKENS);
8710 }
8711
8712 #[test]
8717 fn min_p_reaches_the_sampler_from_the_chat_wire() {
8718 let asked = chat_request(serde_json::json!({
8719 "model": "m",
8720 "messages": [{"role": "user", "content": "hi"}],
8721 "min_p": 0.07,
8722 }));
8723 assert_eq!(
8724 asked
8725 .sampling_params(crate::sampling_knobs::SamplerModel::absent())
8726 .expect("knobs")
8727 .min_p,
8728 0.07
8729 );
8730
8731 let silent = chat_request(serde_json::json!({
8732 "model": "m",
8733 "messages": [{"role": "user", "content": "hi"}],
8734 }));
8735 assert_eq!(
8736 silent
8737 .sampling_params(crate::sampling_knobs::SamplerModel::absent())
8738 .expect("knobs")
8739 .min_p,
8740 0.0,
8741 "an unset min_p must be off, not llama.cpp's CLI default"
8742 );
8743 }
8744
8745 #[test]
8754 fn no_sampler_knob_is_missing_from_the_cache_key() {
8755 let base = serde_json::json!({
8756 "model": "m",
8757 "messages": [{"role": "user", "content": "hi"}],
8758 "seed": 1,
8759 });
8760 let key_for = |body: serde_json::Value| {
8761 let req = chat_request(body);
8762 let params = req
8763 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8764 .expect("params");
8765 req.cache_key("prompt", ¶ms)
8766 };
8767 let baseline = key_for(base.clone());
8768 for (knob, value) in [
8769 ("temperature", serde_json::json!(0.5)),
8770 ("top_p", serde_json::json!(0.9)),
8771 ("min_p", serde_json::json!(0.05)),
8772 ("top_k", serde_json::json!(40)),
8773 ("repetition_penalty", serde_json::json!(1.1)),
8774 ("presence_penalty", serde_json::json!(0.3)),
8775 ("frequency_penalty", serde_json::json!(0.3)),
8776 (
8777 "samplers",
8778 serde_json::json!(["penalties", "top_p", "top_k", "min_p", "temperature"]),
8779 ),
8780 ] {
8781 let mut body = base.clone();
8782 body[knob] = value;
8783 assert_ne!(
8784 key_for(body),
8785 baseline,
8786 "`{knob}` is not in the cache key: two requests differing \
8787 only in it would share one cached answer"
8788 );
8789 }
8790 }
8791
8792 #[test]
8802 fn no_constraint_is_missing_from_the_cache_key() {
8803 let base = serde_json::json!({
8804 "model": "m",
8805 "messages": [{"role": "user", "content": "pick one"}],
8806 });
8807 let key_for = |body: serde_json::Value| {
8808 let req = chat_request(body);
8809 let params = req
8810 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8811 .expect("params");
8812 req.cache_key("prompt", ¶ms)
8813 };
8814 let baseline = key_for(base.clone());
8815 for (field, value) in [
8816 ("grammar", serde_json::json!("root ::= \"yes\" | \"no\"")),
8817 (
8818 "response_format",
8819 serde_json::json!({"type": "json_object"}),
8820 ),
8821 (
8822 "response_format",
8823 serde_json::json!({"type": "json_schema", "json_schema": {
8824 "name": "answer",
8825 "schema": {"type": "object", "properties": {"a": {"type": "string"}}}
8826 }}),
8827 ),
8828 ("ignore_eos", serde_json::json!(true)),
8829 ("stop", serde_json::json!(["\n"])),
8830 ("max_tokens", serde_json::json!(7)),
8831 ] {
8832 let mut body = base.clone();
8833 body[field] = value.clone();
8834 assert_ne!(
8835 key_for(body),
8836 baseline,
8837 "`{field}: {value}` is not in the cache key: two requests \
8838 differing only in it would share one cached answer"
8839 );
8840 }
8841 }
8842
8843 #[test]
8847 fn an_explicit_zero_output_budget_is_a_client_error() {
8848 let req = chat_request(serde_json::json!({
8849 "model": "m",
8850 "messages": [{"role": "user", "content": "hi"}],
8851 "max_tokens": 0,
8852 }));
8853 let (status, body) = req.validate_supported_fields().expect_err("rejected");
8854 assert_eq!(status, StatusCode::BAD_REQUEST);
8855 assert_eq!(body["error"]["param"], serde_json::json!("max_tokens"));
8856 }
8857
8858 #[test]
8861 fn a_request_can_turn_thinking_off() {
8862 let template = graded_template();
8863 for body in [
8864 serde_json::json!({
8865 "model": "m",
8866 "messages": [{"role": "user", "content": "hi"}],
8867 "reasoning_effort": "none",
8868 }),
8869 serde_json::json!({
8870 "model": "m",
8871 "messages": [{"role": "user", "content": "hi"}],
8872 "thinking": {"type": "disabled"},
8873 }),
8874 ] {
8875 let kwargs = chat_request(body).resolve_template_kwargs(&template);
8876 assert_eq!(kwargs["enable_thinking"], serde_json::json!(false));
8877 assert_eq!(kwargs["thinking_mode"], serde_json::json!("disabled"));
8878 assert!(!kwargs.contains_key("reasoning_effort"));
8881 }
8882 }
8883
8884 #[test]
8887 fn a_disabled_switch_beats_an_effort_in_the_same_request() {
8888 let template = graded_template();
8889 let kwargs = chat_request(serde_json::json!({
8890 "model": "m",
8891 "messages": [{"role": "user", "content": "hi"}],
8892 "reasoning_effort": "high",
8893 "thinking": {"type": "disabled"},
8894 }))
8895 .resolve_template_kwargs(&template);
8896 assert_eq!(kwargs["enable_thinking"], serde_json::json!(false));
8897 assert!(!kwargs.contains_key("reasoning_effort"));
8898 }
8899
8900 #[test]
8903 fn an_unrecognized_thinking_switch_is_refused_rather_than_read_as_on() {
8904 let req = chat_request(serde_json::json!({
8905 "model": "m",
8906 "messages": [{"role": "user", "content": "hi"}],
8907 "thinking": {"type": "disable"},
8908 }));
8909 let (status, _) = req.validate_supported_fields().expect_err("rejected");
8910 assert_eq!(status, StatusCode::BAD_REQUEST);
8911 }
8912
8913 #[test]
8916 fn an_explicit_template_kwarg_stands_the_protocol_knobs_down() {
8917 let template = graded_template();
8918 let kwargs = chat_request(serde_json::json!({
8919 "model": "m",
8920 "messages": [{"role": "user", "content": "hi"}],
8921 "reasoning_effort": "none",
8922 "chat_template_kwargs": {"enable_thinking": true},
8923 }))
8924 .resolve_template_kwargs(&template);
8925 assert_eq!(kwargs["enable_thinking"], serde_json::json!(true));
8926 }
8927
8928 #[test]
8932 fn an_off_vocabulary_reasoning_effort_is_quantized_rather_than_interpolated() {
8933 let template = graded_template();
8934 let req = chat_request(serde_json::json!({
8935 "model": "m",
8936 "messages": [{"role": "user", "content": "hi"}],
8937 "reasoning_effort": "minimal",
8938 }));
8939 let kwargs = req.resolve_template_kwargs(&template);
8940 assert_eq!(kwargs["reasoning_effort"], serde_json::json!("low"));
8941 let prompt = prompt_from_messages(&req.messages, &template, &[], kwargs).expect("renders");
8942 assert!(prompt.starts_with("E:low|"), "{prompt}");
8943 }
8944
8945 #[test]
8949 fn an_effort_with_no_near_gear_is_dropped_so_the_template_default_applies() {
8950 let template = graded_template();
8951 let req = chat_request(serde_json::json!({
8952 "model": "m",
8953 "messages": [{"role": "user", "content": "hi"}],
8954 "chat_template_kwargs": {"reasoning_effort": "none"},
8955 }));
8956 let kwargs = req.resolve_template_kwargs(&template);
8957 assert!(!kwargs.contains_key("reasoning_effort"));
8958 let prompt = prompt_from_messages(&req.messages, &template, &[], kwargs).expect("renders");
8959 assert_eq!(prompt, "hi");
8960 }
8961
8962 #[test]
8965 fn chat_template_kwargs_wins_over_the_top_level_reasoning_effort() {
8966 let template = graded_template();
8967 let req = chat_request(serde_json::json!({
8968 "model": "m",
8969 "messages": [{"role": "user", "content": "hi"}],
8970 "reasoning_effort": "low",
8971 "chat_template_kwargs": {"reasoning_effort": "high"},
8972 }));
8973 assert_eq!(
8974 req.resolve_template_kwargs(&template)["reasoning_effort"],
8975 serde_json::json!("high")
8976 );
8977 }
8978
8979 #[test]
8983 fn offering_tools_turns_thinking_on_by_itself() {
8984 let template = graded_template();
8985 let quiet = chat_request(serde_json::json!({
8986 "model": "m",
8987 "messages": [{"role": "user", "content": "hi"}],
8988 }));
8989 assert!(!quiet
8990 .resolve_template_kwargs(&template)
8991 .contains_key("enable_thinking"));
8992
8993 let with_tools = chat_request(serde_json::json!({
8994 "model": "m",
8995 "messages": [{"role": "user", "content": "hi"}],
8996 "tools": [{"type": "function", "function": {"name": "get_weather"}}],
8997 }));
8998 let kwargs = with_tools.resolve_template_kwargs(&template);
8999 assert_eq!(kwargs["enable_thinking"], serde_json::json!(true));
9000 let prompt =
9001 prompt_from_messages(&with_tools.messages, &template, &[], kwargs).expect("renders");
9002 assert!(prompt.starts_with("THINK|"), "{prompt}");
9003 }
9004
9005 #[test]
9010 fn a_prompt_that_opens_the_reasoning_block_makes_the_first_token_reasoning() {
9011 let opener = chat_template::PromptTemplate::from_gguf_metadata(
9012 Some("{{ messages[0].content }}{% if enable_thinking %}<think>{% endif %}"),
9013 Some("qwen3"),
9014 false,
9015 true,
9016 None,
9017 None,
9018 );
9019 let req = chat_request(serde_json::json!({
9020 "model": "m",
9021 "messages": [{"role": "user", "content": "hi"}],
9022 "chat_template_kwargs": {"enable_thinking": true},
9023 }));
9024 let kwargs = req.resolve_template_kwargs(&opener);
9025 let prompt = prompt_from_messages(&req.messages, &opener, &[], kwargs).expect("renders");
9026 assert!(prompt.ends_with("<think>"), "{prompt}");
9027
9028 let posture = output::OutputPosture::resolve("Qwen3-8B", &prompt);
9031 let (message, _) = build_response_message(
9032 "weighing it up</think>Paris.".to_string(),
9033 &[],
9034 posture,
9035 "stop",
9036 );
9037 assert_eq!(message.reasoning_content.as_deref(), Some("weighing it up"));
9038 assert_eq!(message.content.as_deref(), Some("Paris."));
9039
9040 let closed = output::OutputPosture::resolve("Qwen3-8B", "<|im_start|>assistant\n");
9043 let (message, _) = build_response_message(
9044 "weighing it up</think>Paris.".to_string(),
9045 &[],
9046 closed,
9047 "stop",
9048 );
9049 assert_eq!(message.reasoning_content, None);
9050 }
9051
9052 #[test]
9053 fn stop_param_accepts_both_single_string_and_array() {
9054 let req: ChatCompletionRequest = serde_json::from_value(serde_json::json!({
9055 "model": "m",
9056 "messages": [{"role": "user", "content": "hi"}],
9057 "stop": "END",
9058 }))
9059 .unwrap();
9060 assert_eq!(req.stop_sequences(), vec!["END".to_string()]);
9061
9062 let req: ChatCompletionRequest = serde_json::from_value(serde_json::json!({
9063 "model": "m",
9064 "messages": [{"role": "user", "content": "hi"}],
9065 "stop": ["A", "B"],
9066 }))
9067 .unwrap();
9068 assert_eq!(req.stop_sequences(), vec!["A".to_string(), "B".to_string()]);
9069 }
9070
9071 #[test]
9072 fn run_generation_rejects_out_of_vocab_tokens_instead_of_panicking() {
9073 let model = test_model();
9074 let result = run_generation(
9075 &model,
9076 "hello",
9077 &greedy_params(4),
9078 None,
9079 None,
9080 None,
9081 None,
9082 None,
9083 None,
9084 );
9085 assert!(matches!(
9086 result,
9087 Err(generate::DecodeError::TokenOutOfVocab { .. })
9088 ));
9089 }
9090
9091 #[test]
9095 fn run_generation_honors_an_exhausted_kv_pool_and_maps_it_to_a_503() {
9096 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9098 let pool = Arc::new(Mutex::new(frink_core::cache::KvBlockPool::new(64, 2)));
9099
9100 let holder_pool = Arc::clone(&pool);
9101 let holder = std::thread::spawn(move || {
9102 let mut held = frink_core::cache::KvCache::with_pool(1, 1, holder_pool, 0).unwrap();
9103 held.push(&[0.0], &[0.0]).unwrap(); std::thread::sleep(Duration::from_millis(200));
9105 drop(held);
9106 });
9107 std::thread::sleep(Duration::from_millis(15));
9108
9109 let config = generate::KvPoolConfig {
9110 pool,
9111 queue_wait: Duration::ZERO,
9112 };
9113 let result = run_generation(
9114 &model,
9115 &prompt,
9116 &greedy_params(4),
9117 Some(&config),
9118 None,
9119 None,
9120 None,
9121 None,
9122 None,
9123 );
9124 assert!(matches!(
9125 result,
9126 Err(generate::DecodeError::KvPoolExhausted)
9127 ));
9128
9129 let (status, _body) = decode_error_response(result.unwrap_err());
9130 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
9131 holder.join().unwrap();
9132 }
9133
9134 #[test]
9142 fn a_request_too_big_for_the_whole_pool_is_a_400_not_a_retryable_503() {
9143 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9145 let pool = Arc::new(Mutex::new(frink_core::cache::KvBlockPool::new(64, 1)));
9147 let config = generate::KvPoolConfig {
9148 pool,
9149 queue_wait: Duration::ZERO,
9150 };
9151
9152 let result = run_generation(
9153 &model,
9154 &prompt,
9155 &greedy_params(4),
9156 Some(&config),
9157 None,
9158 None,
9159 None,
9160 None,
9161 None,
9162 );
9163 let err = result.expect_err("one block cannot hold two layers' caches");
9164 assert!(
9165 matches!(
9166 &err,
9167 generate::DecodeError::KvBudgetExceeded { binding, .. }
9168 if *binding == frink_models::Ceiling::DeviceMemory.code()
9169 ),
9170 "expected an immovable device-memory refusal, got {err:?}"
9171 );
9172 let (status, _body) = decode_error_response(err);
9173 assert_eq!(status, StatusCode::BAD_REQUEST);
9174 }
9175
9176 #[test]
9182 fn decode_error_response_maps_a_full_queue_to_a_retryable_503() {
9183 let (status, Json(body)) = decode_error_response(generate::DecodeError::QueueFull {
9184 queued: 512,
9185 cap: 512,
9186 });
9187 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
9188 assert_eq!(body["error"]["retry_after_seconds"], 1);
9189 let message = body["error"]["message"].as_str().expect("message");
9190 assert!(message.contains("512"), "{message}");
9191 }
9192
9193 #[test]
9194 fn decode_error_response_omits_a_retry_hint_for_an_unretryable_error() {
9195 let (_status, Json(body)) = decode_error_response(generate::DecodeError::TokenOutOfVocab {
9196 token: 99,
9197 vocab_size: 32,
9198 });
9199 assert!(
9200 body["error"]["retry_after_seconds"].is_null(),
9201 "retrying a prompt this model cannot tokenize never helps"
9202 );
9203 }
9204
9205 #[test]
9206 fn decode_error_response_maps_token_out_of_vocab_to_bad_request() {
9207 let (status, _body) = decode_error_response(generate::DecodeError::TokenOutOfVocab {
9208 token: 99,
9209 vocab_size: 32,
9210 });
9211 assert_eq!(status, StatusCode::BAD_REQUEST);
9212 }
9213
9214 #[test]
9215 fn run_generation_succeeds_and_releases_blocks_when_the_pool_has_room() {
9216 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9218 let pool = Arc::new(Mutex::new(frink_core::cache::KvBlockPool::new(64, 2)));
9219 let config = generate::KvPoolConfig {
9220 pool: pool.clone(),
9221 queue_wait: Duration::ZERO,
9222 };
9223
9224 let produced = run_generation(
9225 &model,
9226 &prompt,
9227 &greedy_params(4),
9228 Some(&config),
9229 None,
9230 None,
9231 None,
9232 None,
9233 None,
9234 )
9235 .unwrap();
9236 assert_eq!(produced.choices[0].finish, FinishReason::Length);
9237 assert_eq!(
9238 pool.lock().unwrap().free_blocks(),
9239 2,
9240 "a completed request must return its blocks to the pool"
9241 );
9242 }
9243
9244 #[tokio::test]
9249 async fn concurrent_requests_against_the_same_model_do_not_interfere() {
9250 let model = Arc::new(test_model());
9251 let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9252
9253 let mut handles = Vec::new();
9254 for _ in 0..8 {
9255 let model = Arc::clone(&model);
9256 let prompt = prompt.clone();
9257 handles.push(tokio::task::spawn_blocking(move || {
9258 run_generation(
9259 &model,
9260 &prompt,
9261 &greedy_params(6),
9262 None,
9263 None,
9264 None,
9265 None,
9266 None,
9267 None,
9268 )
9269 .unwrap()
9270 }));
9271 }
9272
9273 let mut results = Vec::new();
9274 for h in handles {
9275 results.push(h.await.unwrap());
9276 }
9277 for r in &results[1..] {
9282 assert_eq!(r.choices, results[0].choices, "choices must match");
9286 assert_eq!(
9287 r.usage.prompt_tokens, results[0].usage.prompt_tokens,
9288 "prompt token count must match"
9289 );
9290 assert_eq!(
9291 r.usage.completion_tokens, results[0].usage.completion_tokens,
9292 "completion token count must match"
9293 );
9294 }
9295 }
9296
9297 fn write_safetensors_shard(tensors: &[(String, Vec<usize>, Vec<f32>)]) -> Vec<u8> {
9304 let mut header_entries = Vec::new();
9305 let mut data = Vec::new();
9306 for (name, shape, values) in tensors {
9307 let start = data.len();
9308 for v in values {
9309 data.extend_from_slice(&v.to_le_bytes());
9310 }
9311 let end = data.len();
9312 let shape_str = shape
9313 .iter()
9314 .map(|d| d.to_string())
9315 .collect::<Vec<_>>()
9316 .join(",");
9317 header_entries.push(format!(
9318 "\"{name}\":{{\"dtype\":\"F32\",\"shape\":[{shape_str}],\"data_offsets\":[{start},{end}]}}"
9319 ));
9320 }
9321 let header = format!("{{{}}}", header_entries.join(","));
9322 let header_bytes = header.as_bytes();
9323 let mut out = Vec::with_capacity(8 + header_bytes.len() + data.len());
9324 out.extend_from_slice(&(header_bytes.len() as u64).to_le_bytes());
9325 out.extend_from_slice(header_bytes);
9326 out.extend_from_slice(&data);
9327 out
9328 }
9329
9330 fn build_synthetic_kimi_loaded() -> model::KimiLoaded {
9341 use frink_models::config::{AttentionKind, KdaConfig, KimiHybridAttention, MlaConfig};
9342 use frink_models::kimi_loader::KimiRealHparams;
9343 use frink_moe::{GatingFunction, MoeLayerConfig};
9344
9345 let hidden_dim = 8;
9346 let kda_num_heads = 2;
9347 let kda_head_dim = 3;
9348 let kda_proj = kda_num_heads * kda_head_dim;
9349 let conv_kernel = 4;
9350 let dense_intermediate = 5;
9351 let vocab_size = 256;
9355 let mla_num_heads = 1;
9356 let mla_q_lora_rank = 2;
9357 let mla_kv_lora_rank = 2;
9358 let mla_qk_nope_head_dim = 2;
9359 let mla_qk_rope_head_dim = 2;
9360 let mla_v_head_dim = 2;
9361
9362 let model_cfg = frink_models::ModelConfig {
9363 rope_layers: frink_models::rope_layers::RopeLayers::All,
9364 layer_shapes: frink_models::layer_shapes::LayerShapes::Uniform,
9365 name: "synthetic-kimi-server-test",
9366 n_layers: 1,
9367 n_mtp_blocks: 0,
9368 hidden_dim,
9369 n_heads: 1,
9370 n_kv_heads: 1,
9371 head_dim: 4,
9372 v_head_dim: None,
9373 vocab_size,
9374 rope_theta: 10000.0,
9375 rms_norm_eps: 1e-5,
9376 post_norm_eps: 1e-5,
9377 sliding_window: None,
9378 moe: MoeLayerConfig {
9379 expert_weights_scale: 1.0,
9380 routed_weight_before_ffn: false,
9381 n_experts: 1,
9382 n_experts_active: 1,
9383 n_shared_experts: 0,
9384 hidden_dim,
9385 expert_ffn_dim: 4,
9386 gating: GatingFunction::Sigmoid,
9387 norm_topk_prob: true,
9388 expert_group_count: None,
9389 expert_group_used_count: None,
9390 },
9391 n_dense_leading_layers: 1,
9396 moe_interleave_step: None,
9397 norm_function: frink_models::norm::NormFunction::Rms,
9398 attention: AttentionKind::KimiHybrid(KimiHybridAttention {
9399 kda_layers: vec![1],
9400 full_attn_layers: vec![],
9401 mla: MlaConfig {
9402 num_heads: mla_num_heads,
9403 q_lora_rank: mla_q_lora_rank,
9404 kv_lora_rank: mla_kv_lora_rank,
9405 qk_nope_head_dim: mla_qk_nope_head_dim,
9406 qk_rope_head_dim: mla_qk_rope_head_dim,
9407 v_head_dim: mla_v_head_dim,
9408 use_output_gate: true,
9409 rope: None,
9410 },
9411 kda: KdaConfig {
9412 num_heads: kda_num_heads,
9413 head_dim: kda_head_dim,
9414 short_conv_kernel_size: conv_kernel,
9415 gate_lower_bound: -5.0,
9416 use_full_rank_gate: true,
9417 },
9418 }),
9419 rope_freqs: None,
9420 rope_attn_factor: 1.0,
9421 rope_dim: None,
9422 rope_dim_swa: None,
9423 rope_freqs_long: None,
9424 rope_freqs_short: None,
9425 rope_orig_ctx: None,
9426 rope_layout: frink_models::config::RopeLayout::Neox,
9427 qk_norm_style: frink_models::capability::QkNormStyle::WholeVector,
9428 swa_layers: frink_models::swa_layers::SwaLayers::All,
9429 attn_logit_softcap: None,
9430 final_logit_softcap: None,
9431 embedding_scale: None,
9432 residual_scale: None,
9433 normed_residual_scale: None,
9434 clamp_kqv: None,
9435 attn_temperature: None,
9436 router_input: frink_models::router_input::RouterInput::NormedFfnInput,
9437 block_sub_norms: false,
9438 parallel_residual: false,
9439 learned_positions: false,
9440 attn_value_scale: None,
9441 alibi_max_bias: None,
9442 layer_loops: None,
9443 skip_stream: false,
9444 parallel_ssm: false,
9445 swa_chunked: false,
9446 weightless_qk_norm: false,
9447 logit_multiplier: None,
9448 attention_scale: None,
9449 rope_theta_swa: None,
9450 ffn_activation: frink_models::config::FfnActivation::Swiglu,
9451 best_effort_fields: &["synthetic test config, not a real preset"],
9452 };
9453 let hp = KimiRealHparams {
9454 hidden_dim,
9455 kda_num_heads,
9456 kda_head_dim,
9457 mla_num_heads,
9458 mla_q_lora_rank,
9459 mla_kv_lora_rank,
9460 mla_qk_nope_head_dim,
9461 mla_qk_rope_head_dim,
9462 mla_v_head_dim,
9463 dense_intermediate_dim: dense_intermediate,
9464 moe_hidden_dim: hidden_dim,
9465 moe_intermediate_dim: 4,
9466 n_experts: 1,
9467 num_shared_experts: 0,
9468 };
9469
9470 let prefix = "language_model.model.layers.0";
9474 let mut tensors: Vec<(String, Vec<usize>, Vec<f32>)> = Vec::new();
9475 let push = |tensors: &mut Vec<(String, Vec<usize>, Vec<f32>)>,
9476 name: String,
9477 shape: Vec<usize>,
9478 n: usize| {
9479 tensors.push((name, shape, vec![0.01f32; n]));
9480 };
9481 push(
9482 &mut tensors,
9483 format!("{prefix}.input_layernorm.weight"),
9484 vec![hidden_dim],
9485 hidden_dim,
9486 );
9487 push(
9488 &mut tensors,
9489 format!("{prefix}.post_attention_layernorm.weight"),
9490 vec![hidden_dim],
9491 hidden_dim,
9492 );
9493 push(
9494 &mut tensors,
9495 format!("{prefix}.self_attention_res_norm.weight"),
9496 vec![hidden_dim],
9497 hidden_dim,
9498 );
9499 push(
9500 &mut tensors,
9501 format!("{prefix}.self_attention_res_proj.weight"),
9502 vec![1, hidden_dim],
9503 hidden_dim,
9504 );
9505 push(
9506 &mut tensors,
9507 format!("{prefix}.mlp_res_norm.weight"),
9508 vec![hidden_dim],
9509 hidden_dim,
9510 );
9511 push(
9512 &mut tensors,
9513 format!("{prefix}.mlp_res_proj.weight"),
9514 vec![1, hidden_dim],
9515 hidden_dim,
9516 );
9517 push(
9518 &mut tensors,
9519 format!("{prefix}.self_attn.q_proj.weight"),
9520 vec![kda_proj, hidden_dim],
9521 kda_proj * hidden_dim,
9522 );
9523 push(
9524 &mut tensors,
9525 format!("{prefix}.self_attn.k_proj.weight"),
9526 vec![kda_proj, hidden_dim],
9527 kda_proj * hidden_dim,
9528 );
9529 push(
9530 &mut tensors,
9531 format!("{prefix}.self_attn.v_proj.weight"),
9532 vec![kda_proj, hidden_dim],
9533 kda_proj * hidden_dim,
9534 );
9535 push(
9536 &mut tensors,
9537 format!("{prefix}.self_attn.q_conv1d.weight"),
9538 vec![kda_proj, 1, conv_kernel],
9539 kda_proj * conv_kernel,
9540 );
9541 push(
9542 &mut tensors,
9543 format!("{prefix}.self_attn.k_conv1d.weight"),
9544 vec![kda_proj, 1, conv_kernel],
9545 kda_proj * conv_kernel,
9546 );
9547 push(
9548 &mut tensors,
9549 format!("{prefix}.self_attn.v_conv1d.weight"),
9550 vec![kda_proj, 1, conv_kernel],
9551 kda_proj * conv_kernel,
9552 );
9553 push(
9554 &mut tensors,
9555 format!("{prefix}.self_attn.A_log"),
9556 vec![kda_num_heads],
9557 kda_num_heads,
9558 );
9559 push(
9560 &mut tensors,
9561 format!("{prefix}.self_attn.f_a_proj.weight"),
9562 vec![kda_head_dim, hidden_dim],
9563 kda_head_dim * hidden_dim,
9564 );
9565 push(
9566 &mut tensors,
9567 format!("{prefix}.self_attn.f_b_proj.weight"),
9568 vec![kda_proj, kda_head_dim],
9569 kda_proj * kda_head_dim,
9570 );
9571 push(
9572 &mut tensors,
9573 format!("{prefix}.self_attn.dt_bias"),
9574 vec![kda_proj],
9575 kda_proj,
9576 );
9577 push(
9578 &mut tensors,
9579 format!("{prefix}.self_attn.b_proj.weight"),
9580 vec![kda_num_heads, hidden_dim],
9581 kda_num_heads * hidden_dim,
9582 );
9583 push(
9584 &mut tensors,
9585 format!("{prefix}.self_attn.g_proj.weight"),
9586 vec![kda_proj, hidden_dim],
9587 kda_proj * hidden_dim,
9588 );
9589 push(
9590 &mut tensors,
9591 format!("{prefix}.self_attn.o_norm.weight"),
9592 vec![kda_head_dim],
9593 kda_head_dim,
9594 );
9595 push(
9596 &mut tensors,
9597 format!("{prefix}.self_attn.o_proj.weight"),
9598 vec![hidden_dim, kda_proj],
9599 hidden_dim * kda_proj,
9600 );
9601 push(
9602 &mut tensors,
9603 format!("{prefix}.mlp.gate_proj.weight"),
9604 vec![dense_intermediate, hidden_dim],
9605 dense_intermediate * hidden_dim,
9606 );
9607 push(
9608 &mut tensors,
9609 format!("{prefix}.mlp.up_proj.weight"),
9610 vec![dense_intermediate, hidden_dim],
9611 dense_intermediate * hidden_dim,
9612 );
9613 push(
9614 &mut tensors,
9615 format!("{prefix}.mlp.down_proj.weight"),
9616 vec![hidden_dim, dense_intermediate],
9617 hidden_dim * dense_intermediate,
9618 );
9619 push(
9620 &mut tensors,
9621 "language_model.model.embed_tokens.weight".to_string(),
9622 vec![vocab_size, hidden_dim],
9623 vocab_size * hidden_dim,
9624 );
9625 push(
9626 &mut tensors,
9627 "language_model.lm_head.weight".to_string(),
9628 vec![vocab_size, hidden_dim],
9629 vocab_size * hidden_dim,
9630 );
9631 push(
9632 &mut tensors,
9633 "language_model.model.norm.weight".to_string(),
9634 vec![hidden_dim],
9635 hidden_dim,
9636 );
9637 push(
9638 &mut tensors,
9639 "language_model.model.output_attn_res_norm.weight".to_string(),
9640 vec![hidden_dim],
9641 hidden_dim,
9642 );
9643 push(
9644 &mut tensors,
9645 "language_model.model.output_attn_res_proj.weight".to_string(),
9646 vec![1, hidden_dim],
9647 hidden_dim,
9648 );
9649
9650 static FIXTURE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
9663 let dir = std::env::temp_dir().join(format!(
9664 "frink_server_kimi_e2e_test_{}_{}",
9665 std::process::id(),
9666 FIXTURE.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
9667 ));
9668 std::fs::create_dir_all(&dir).unwrap();
9669 let shard_bytes = write_safetensors_shard(&tensors);
9670 std::fs::write(dir.join("shard0.safetensors"), &shard_bytes).unwrap();
9671 let map_entries: Vec<String> = tensors
9672 .iter()
9673 .map(|(name, ..)| format!("\"{name}\":\"shard0.safetensors\""))
9674 .collect();
9675 let index = format!("{{\"weight_map\":{{{}}}}}", map_entries.join(","));
9676 std::fs::write(dir.join("model.safetensors.index.json"), &index).unwrap();
9677
9678 use base64::Engine;
9682 let vocab_lines: Vec<String> = (0..vocab_size as u32)
9683 .map(|b| {
9684 let b64 = base64::engine::general_purpose::STANDARD.encode([b as u8]);
9685 format!("{b64} {b}")
9686 })
9687 .collect();
9688 std::fs::write(dir.join("tiktoken.model"), vocab_lines.join("\n")).unwrap();
9689
9690 let loaded = model::load_kimi_checkpoint_with_config(dir.to_str().unwrap(), model_cfg, hp)
9691 .expect("must load the synthetic Kimi checkpoint end to end");
9692 std::fs::remove_dir_all(&dir).ok();
9693 loaded
9694 }
9695
9696 #[test]
9704 fn kimi_model_serves_real_text_end_to_end_via_run_generation() {
9705 let loaded = build_synthetic_kimi_loaded();
9706 let state = build_app_state(
9707 StartupModels {
9708 loaded: model::LoadedModel::Kimi(loaded),
9709 embedding: None,
9710 },
9711 None,
9712 None,
9713 None,
9714 false,
9715 None,
9716 Arc::new(health::Detection::ready(health::probe_backends())),
9717 );
9718 let active = state.active().expect("a freshly built state has a model");
9719 assert_eq!(active.tokenizer_kind(), "kimi-tiktoken-bpe");
9720 assert!(!active.is_synthetic());
9721
9722 let produced = run_generation(
9723 active.generative().unwrap(),
9724 "hi",
9725 &greedy_params(5),
9726 None,
9727 None,
9728 None,
9729 None,
9730 None,
9731 None,
9732 )
9733 .expect("a real Kimi checkpoint must generate without error");
9734 assert!(matches!(
9735 produced.choices[0].finish,
9736 FinishReason::Length | FinishReason::Stop
9737 ));
9738 }
9739
9740 #[test]
9753 fn a_grammar_constrains_the_engine_decode_path() {
9754 let loaded = build_synthetic_kimi_loaded();
9755 let state = build_app_state(
9756 StartupModels {
9757 loaded: model::LoadedModel::Kimi(loaded),
9758 embedding: None,
9759 },
9760 None,
9761 None,
9762 None,
9763 false,
9764 None,
9765 Arc::new(health::Detection::ready(health::probe_backends())),
9766 );
9767 let active = state.active().expect("a freshly built state has a model");
9768
9769 let run = |grammar: Option<&str>| {
9770 let mut params = greedy_params(6);
9771 params.grammar = grammar.map(|src| {
9772 Arc::new(
9773 frink_models::grammar::Grammar::from_str_with_root(src, "root")
9774 .expect("test grammar parses"),
9775 )
9776 });
9777 run_generation(
9778 active.generative().unwrap(),
9779 "hi",
9780 ¶ms,
9781 None,
9782 None,
9783 None,
9784 None,
9785 None,
9786 None,
9787 )
9788 };
9789
9790 let produced = run(None).expect("the unconstrained run must serve");
9791 let unconstrained = produced.choices[0].text.clone();
9792 assert!(
9793 unconstrained.chars().any(|c| c != 'a'),
9794 "the unconstrained run produced only `a` ({unconstrained:?}), so the \
9795 constrained run below would prove nothing"
9796 );
9797
9798 let produced =
9799 run(Some(r#"root ::= "a"+"#)).expect("a grammar this vocabulary can spell must serve");
9800 let one = produced.choices.into_iter().next().unwrap();
9801 let (finish, constrained) = (one.finish, one.text);
9802 assert!(
9803 !constrained.is_empty() && constrained.chars().all(|c| c == 'a'),
9804 "the engine decode path served text its grammar forbids ({constrained:?}): \
9805 the constraint was dropped between `generate_engine` and the sampler"
9806 );
9807 assert!(matches!(finish, FinishReason::Length | FinishReason::Stop));
9808 }
9809
9810 #[test]
9820 fn kv_pool_and_prefix_cache_are_never_consulted_for_a_kimi_model() {
9821 let loaded = build_synthetic_kimi_loaded();
9822 let state = build_app_state(
9823 StartupModels {
9824 loaded: model::LoadedModel::Kimi(loaded),
9825 embedding: None,
9826 },
9827 None,
9828 None,
9829 None,
9830 false,
9831 None,
9832 Arc::new(health::Detection::ready(health::probe_backends())),
9833 );
9834
9835 let pool = Arc::new(Mutex::new(frink_core::cache::KvBlockPool::new(64, 4)));
9836 let kv_pool_config = generate::KvPoolConfig {
9837 pool: pool.clone(),
9838 queue_wait: Duration::ZERO,
9839 };
9840 let pc = Mutex::new(PrefixCache::new(4));
9841
9842 run_generation(
9843 state
9844 .active()
9845 .expect("a freshly built state has a model")
9846 .generative()
9847 .unwrap(),
9848 "hi",
9849 &greedy_params(5),
9850 Some(&kv_pool_config),
9851 None,
9852 Some(&pc),
9853 None,
9854 None,
9855 None,
9856 )
9857 .expect("a real Kimi checkpoint must generate without error");
9858
9859 assert_eq!(
9860 pool.lock().unwrap().free_blocks(),
9861 4,
9862 "the KV pool must be completely untouched by a Kimi request"
9863 );
9864 let stats = pc.lock().unwrap().stats();
9865 assert_eq!(
9866 stats.hits + stats.misses,
9867 0,
9868 "the prefix cache must never be consulted for a Kimi request"
9869 );
9870 }
9871}