1mod admin;
36mod anthropic;
37mod attribution;
38mod best_of;
39mod budget;
40mod cache_admin;
41mod cache_salt;
42mod cancel;
43mod chat_params;
44mod chat_template;
45mod cli;
46mod completion;
47mod continuation;
48mod conversations;
49mod decode_task;
50mod embeddings;
51mod generate;
52mod grammar_request;
53mod health;
54mod journal;
55mod json_mode;
56mod limits;
57mod loaded;
58mod logprobs;
59mod lora;
60mod mcp;
61mod model;
62mod openai_extra;
63mod output;
64mod policy;
65mod prefill_batch;
66mod reasoning_budget;
67mod reasoning_tokens;
68mod request_tail;
69mod rerank;
70mod response_cache;
71pub(crate) mod responses;
72mod resume;
73mod sample_step;
74mod sampling_knobs;
75mod sampling_loop;
76mod security;
77mod serving;
78mod session;
79mod slots;
80mod sse;
81mod stats;
82mod stop;
83mod stream_events;
84mod tasks;
85mod tool_grammar;
86mod unimplemented_fields;
87mod unsupported_sampling;
88mod utf8_stream;
89
90use std::cell::RefCell;
91use std::convert::Infallible;
92use std::net::SocketAddr;
93use std::path::PathBuf;
94use std::rc::Rc;
95use std::sync::{Arc, Mutex, MutexGuard};
96use std::time::Duration;
97
98use axum::{
99 extract::State,
100 http::StatusCode,
101 response::sse::{Event, Sse},
102 response::{IntoResponse, Response},
103 routing::{get, post},
104 Json, Router,
105};
106use serde::{Deserialize, Serialize};
107
108use cli::apply_cli_overrides;
109pub use cli::{ServerArgs, BUILT_WITH_CUDA, BUILT_WITH_METAL};
110
111use frink_core::cache::KvBlockPool;
112use frink_models::kimi_tokenizer::KimiTokenizer;
113use frink_models::sampling::SamplingParams;
114use frink_models::tokenizer::{SpecialTokens, StopTokens};
115use frink_models::{Decoder, Gemma4Engine, KimiEngine, MlaEngine, PrefixCache};
116#[cfg(test)]
117use generate::FinishReason;
118use generate::GenerationParams;
119pub(crate) use loaded::{ActiveModel, Loaded, SleptModel};
120use model::ServerTokenizer;
121use rerank::encoder_endpoints;
122use response_cache::ResponseCache;
123use sampling_knobs::SamplingKnobs;
124
125#[allow(clippy::large_enum_variant)] pub(crate) enum Model {
134 Gguf(GgufModel),
135 Kimi(KimiModel),
136 Mla(MlaModel),
137 Gemma4(Gemma4Model),
138 Glm52(Glm52Model),
139}
140
141pub(crate) struct GgufModel {
142 decoder: Arc<Decoder>,
143 tokenizer: Arc<ServerTokenizer>,
144 stop_tokens: StopTokens,
145 bos_id: Option<usize>,
146 is_synthetic: bool,
147 chat_template: chat_template::PromptTemplate,
148}
149
150pub(crate) struct KimiModel {
151 engine: KimiEngine,
152 tokenizer: KimiTokenizer,
153 stop_tokens: StopTokens,
154 chat_template: chat_template::PromptTemplate,
155}
156
157pub(crate) struct MlaModel {
158 engine: MlaEngine,
159 tokenizer: ServerTokenizer,
160 stop_tokens: StopTokens,
161 bos_id: Option<usize>,
162 name: String,
163 chat_template: chat_template::PromptTemplate,
164}
165
166pub(crate) struct Gemma4Model {
167 engine: Gemma4Engine,
168 tokenizer: ServerTokenizer,
169 stop_tokens: StopTokens,
170 bos_id: Option<usize>,
171 name: String,
172 chat_template: chat_template::PromptTemplate,
173}
174
175pub(crate) struct Glm52Model {
176 engine: frink_models::Glm52Engine,
177 tokenizer: ServerTokenizer,
178 stop_tokens: StopTokens,
179 bos_id: Option<usize>,
180 name: String,
181 chat_template: chat_template::PromptTemplate,
182}
183
184impl Model {
185 pub(crate) fn chat_template(&self) -> chat_template::PromptTemplate {
186 match self {
187 Model::Gguf(m) => m.chat_template.clone(),
188 Model::Kimi(m) => m.chat_template.clone(),
189 Model::Mla(m) => m.chat_template.clone(),
190 Model::Gemma4(m) => m.chat_template.clone(),
191 Model::Glm52(m) => m.chat_template.clone(),
192 }
193 }
194
195 fn is_synthetic(&self) -> bool {
200 match self {
201 Model::Gguf(m) => m.is_synthetic,
202 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => false,
203 }
204 }
205
206 fn tokenizer_kind(&self) -> &'static str {
207 match self {
208 Model::Gguf(m) => m.tokenizer.kind(),
209 Model::Kimi(_) => "kimi-tiktoken-bpe",
210 Model::Mla(m) => m.tokenizer.kind(),
211 Model::Gemma4(m) => m.tokenizer.kind(),
212 Model::Glm52(m) => m.tokenizer.kind(),
213 }
214 }
215
216 fn expert_store_stats(&self) -> Option<frink_core::expert_store::ExpertStoreStats> {
220 match self {
221 Model::Gguf(m) => m.decoder.expert_store_stats(),
222 Model::Kimi(m) => m.engine.weights.expert_store_stats(),
223 Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
224 }
225 }
226
227 pub(crate) fn name(&self) -> &str {
228 match self {
229 Model::Gguf(m) => m.decoder.config.name,
230 Model::Kimi(_) => "kimi-k3",
231 Model::Mla(m) => m.name.as_str(),
232 Model::Gemma4(m) => m.name.as_str(),
233 Model::Glm52(m) => m.name.as_str(),
234 }
235 }
236
237 pub(crate) fn encode(&self, text: &str, specials: SpecialTokens) -> Vec<usize> {
264 match self {
265 Model::Gguf(m) => m.tokenizer.encode(text, specials),
266 Model::Kimi(m) => m
267 .tokenizer
268 .encode(text, specials)
269 .into_iter()
270 .map(|id| id as usize)
271 .collect(),
272 Model::Mla(m) => m.tokenizer.encode(text, specials),
273 Model::Gemma4(m) => m.tokenizer.encode(text, specials),
274 Model::Glm52(m) => m.tokenizer.encode(text, specials),
275 }
276 }
277
278 pub(crate) fn bos_id(&self) -> Option<usize> {
287 match self {
288 Model::Gguf(m) => m.bos_id,
289 Model::Kimi(_) => None,
290 Model::Mla(m) => m.bos_id,
291 Model::Gemma4(m) => m.bos_id,
292 Model::Glm52(m) => m.bos_id,
293 }
294 }
295
296 pub(crate) fn decode(&self, ids: &[usize]) -> String {
297 match self {
298 Model::Gguf(m) => m.tokenizer.decode(ids),
299 Model::Kimi(m) => {
300 let ids32: Vec<u32> = ids.iter().map(|&id| id as u32).collect();
301 m.tokenizer.decode(&ids32)
302 }
303 Model::Mla(m) => m.tokenizer.decode(ids),
304 Model::Gemma4(m) => m.tokenizer.decode(ids),
305 Model::Glm52(m) => m.tokenizer.decode(ids),
306 }
307 }
308
309 pub(crate) fn embed_tokens(&self, tokens: &[usize]) -> Option<Vec<Vec<f32>>> {
312 match self {
313 Model::Gguf(m) => {
314 let mut caches: Vec<_> = m.decoder.config.new_kv_caches();
315 Some(m.decoder.forward_hidden_batch(tokens, 0, &mut caches))
316 }
317 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
318 }
319 }
320
321 pub(crate) fn gguf_decoder(&self) -> Option<&Arc<Decoder>> {
327 match self {
328 Model::Gguf(m) => Some(&m.decoder),
329 Model::Kimi(_) | Model::Mla(_) | Model::Gemma4(_) | Model::Glm52(_) => None,
330 }
331 }
332
333 pub(crate) fn vocab_size(&self) -> Option<usize> {
334 match self {
335 Model::Gguf(m) => Some(m.decoder.config.vocab_size),
336 Model::Kimi(m) => Some(m.tokenizer.vocab_size()),
337 Model::Mla(m) => Some(frink_models::Engine::vocab_size(&m.engine)),
338 Model::Gemma4(m) => Some(frink_models::Engine::vocab_size(&m.engine)),
339 Model::Glm52(m) => Some(frink_models::Engine::vocab_size(&m.engine)),
340 }
341 }
342
343 fn has_real_vocabulary(&self) -> bool {
350 match self {
351 Model::Gguf(m) => !matches!(*m.tokenizer, model::ServerTokenizer::Byte),
352 Model::Kimi(_) => true,
353 Model::Mla(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
354 Model::Gemma4(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
355 Model::Glm52(m) => !matches!(m.tokenizer, model::ServerTokenizer::Byte),
356 }
357 }
358}
359
360impl frink_models::dry::DryVocab for Model {
366 fn n_tokens(&self) -> usize {
367 self.vocab_size().unwrap_or(0)
368 }
369
370 fn detokenize(&self, token: usize) -> String {
371 self.decode(&[token])
372 }
373
374 fn tokenize(&self, text: &str) -> Vec<usize> {
375 self.encode(text, SpecialTokens::AsText)
376 }
377}
378
379pub(crate) struct AppState {
380 pub(crate) embedding: Option<Arc<frink_models::EmbeddingModel>>,
394 active: std::sync::RwLock<Option<Arc<ActiveModel>>>,
413 pub(crate) load_in_progress: std::sync::atomic::AtomicBool,
417 pub(crate) slept: Mutex<Option<SleptModel>>,
427 pub(crate) tasks: Arc<tasks::TaskRegistry>,
429 pub(crate) cancels: Arc<cancel::CancelRegistry>,
433 pub(crate) stats: stats::Stats,
436 pub(crate) streams: resume::StreamRegistry,
439 pub(crate) model_dir: Option<PathBuf>,
441 response_cache: Mutex<ResponseCache>,
445 pub(crate) kv_pool: Option<generate::KvPoolConfig>,
457 pub(crate) paged_kv: Option<generate::PagedKvConfig>,
470 pub(crate) prefix_cache: Option<Arc<Mutex<PrefixCache>>>,
480 sessions: session::SessionStore,
486 requests_total: std::sync::atomic::AtomicU64,
487 request_errors_total: std::sync::atomic::AtomicU64,
488 started_at: std::time::Instant,
489 last_request_ms: std::sync::atomic::AtomicU64,
494 detection: Arc<health::Detection>,
496 mcp: Option<mcp::LoadedMcpConfig>,
498 pub(crate) continuous_batching_enabled: bool,
502 pub(crate) metal_private_decode_gate: Option<Arc<std::sync::Mutex<()>>>,
506 loading_model: Mutex<Option<String>>,
510 last_load_error: Mutex<Option<(String, String)>>,
514 pub(crate) serving: Mutex<crate::stats::ServingStats>,
519 pub(crate) maintenance: Mutex<crate::policy::maintenance::MaintenanceGate>,
524 pub(crate) footprint:
530 Mutex<crate::policy::footprint::ProbeCache<crate::policy::footprint::Footprint>>,
531 pub(crate) started_unix: u64,
541}
542
543pub(crate) const FOOTPRINT_TTL_MS: u64 = 2_000;
549
550impl AppState {
551 pub(crate) fn active(&self) -> Option<Arc<ActiveModel>> {
555 self.active
556 .read()
557 .unwrap_or_else(|p| p.into_inner())
558 .clone()
559 }
560
561 pub(crate) fn is_sleeping(&self) -> bool {
568 self.slept
569 .lock()
570 .unwrap_or_else(|p| p.into_inner())
571 .is_some()
572 }
573
574 pub(crate) fn require_active(&self) -> Result<Arc<ActiveModel>, ApiError> {
575 if let Some(active) = self.active() {
576 return Ok(active);
577 }
578 if self.is_sleeping() {
582 return Err((
583 StatusCode::SERVICE_UNAVAILABLE,
584 Json(serde_json::json!({"error": {
585 "message": "this server is asleep; POST /wake_up to reload the model it put \
586 away",
587 "type": "server_sleeping"
588 }})),
589 ));
590 }
591 Err((
592 StatusCode::SERVICE_UNAVAILABLE,
593 Json(serde_json::json!({"error": {
594 "message": "no model is loaded; POST /admin/models/load with an id from \
595 GET /admin/models",
596 "type": "model_not_loaded"
597 }})),
598 ))
599 }
600
601 pub(crate) fn require_model(&self) -> Result<Arc<Model>, ApiError> {
609 Ok(Arc::clone(self.require_active()?.generative()?))
610 }
611
612 pub(crate) fn swap_active(&self, next: Option<Arc<ActiveModel>>) -> Option<Arc<ActiveModel>> {
620 let mut guard = self.active.write().unwrap_or_else(|p| p.into_inner());
621 std::mem::replace(&mut *guard, next)
622 }
623
624 fn mark_request_finished(&self) {
628 let ms = self.started_at.elapsed().as_millis().min(u64::MAX as u128) as u64;
629 self.last_request_ms
630 .store(ms, std::sync::atomic::Ordering::Relaxed);
631 }
632
633 pub(crate) fn uptime(&self) -> Duration {
634 self.started_at.elapsed()
635 }
636
637 pub(crate) fn requests_total(&self) -> u64 {
638 self.requests_total
639 .load(std::sync::atomic::Ordering::Relaxed)
640 }
641
642 pub(crate) fn errors_total(&self) -> u64 {
643 self.request_errors_total
644 .load(std::sync::atomic::Ordering::Relaxed)
645 }
646
647 pub(crate) fn cache_stats(&self) -> response_cache::CacheStats {
648 lock_cache(&self.response_cache).stats()
649 }
650
651 pub(crate) fn last_request_age_seconds(&self) -> Option<f64> {
654 let last = self
655 .last_request_ms
656 .load(std::sync::atomic::Ordering::Relaxed);
657 (last > 0)
658 .then(|| self.uptime().as_secs_f64() - (last as f64 / 1000.0))
659 .map(|age| age.max(0.0))
660 }
661
662 pub(crate) fn loading_model_id(&self) -> Option<String> {
663 self.loading_model
664 .lock()
665 .unwrap_or_else(|p| p.into_inner())
666 .clone()
667 }
668
669 pub(crate) fn set_loading_model(&self, id: Option<String>) {
670 *self.loading_model.lock().unwrap_or_else(|p| p.into_inner()) = id;
671 }
672
673 pub(crate) fn last_load_error(&self) -> Option<(String, String)> {
674 self.last_load_error
675 .lock()
676 .unwrap_or_else(|p| p.into_inner())
677 .clone()
678 }
679
680 pub(crate) fn set_last_load_error(&self, error: Option<(String, String)>) {
681 *self
682 .last_load_error
683 .lock()
684 .unwrap_or_else(|p| p.into_inner()) = error;
685 }
686
687 pub(crate) fn active_model_name(&self) -> Option<String> {
696 self.active().map(|a| a.name().to_string())
697 }
698
699 pub(crate) fn embedding_model(&self) -> Option<Arc<frink_models::EmbeddingModel>> {
709 self.embedding
710 .clone()
711 .or_else(|| self.active().and_then(|a| a.encoder().map(Arc::clone)))
712 }
713
714 pub(crate) fn embedding_model_name(&self) -> Option<String> {
718 match self.embedding_model() {
719 Some(e) => Some(e.name().to_string()),
720 None => self.active_model_name(),
721 }
722 }
723
724 pub(crate) fn record_request(&self, record: stats::Record<'_>) {
725 self.stats.record(stats::entry(record));
726 }
727}
728
729fn lock_cache(cache: &Mutex<ResponseCache>) -> MutexGuard<'_, ResponseCache> {
734 cache
735 .lock()
736 .unwrap_or_else(|poisoned| poisoned.into_inner())
737}
738
739#[derive(Debug, Clone, Deserialize)]
740#[serde(untagged)]
741pub(crate) enum MessageContent {
742 Text(String),
743 Parts(Vec<ContentPart>),
744}
745
746#[derive(Debug, Clone, Deserialize)]
747struct ContentPart {
748 #[serde(rename = "type")]
749 kind: String,
750 #[serde(default)]
751 text: Option<String>,
752 #[serde(default)]
753 image_url: Option<serde_json::Value>,
754}
755
756impl MessageContent {
757 fn as_text(&self) -> String {
758 match self {
759 Self::Text(s) => s.clone(),
760 Self::Parts(parts) => parts
761 .iter()
762 .filter_map(|p| p.text.as_deref())
763 .collect::<Vec<_>>()
764 .join(""),
765 }
766 }
767
768 fn has_image(&self) -> bool {
769 match self {
770 Self::Text(_) => false,
771 Self::Parts(parts) => parts
772 .iter()
773 .any(|p| p.kind == "image_url" || p.image_url.is_some()),
774 }
775 }
776}
777
778#[derive(Debug, Clone, Deserialize)]
779pub(crate) struct ChatMessage {
780 pub(crate) role: String,
781 #[serde(default)]
785 pub(crate) content: Option<MessageContent>,
786 #[serde(default)]
790 pub(crate) tool_calls: Option<Vec<ToolCallIn>>,
791 #[serde(default)]
796 #[allow(dead_code)]
797 pub(crate) tool_call_id: Option<String>,
798 #[serde(default, alias = "reasoning")]
814 pub(crate) reasoning_content: Option<String>,
815}
816
817impl ChatMessage {
818 fn rendered_content(&self) -> String {
825 let mut out = self
826 .content
827 .as_ref()
828 .map(MessageContent::as_text)
829 .unwrap_or_default();
830 if let Some(calls) = &self.tool_calls {
831 for call in calls {
832 out.push_str(&format!(
833 "<tool_call>{{\"name\": \"{}\", \"arguments\": {}}}</tool_call>",
834 call.function.name, call.function.arguments
835 ));
836 }
837 }
838 out
839 }
840}
841
842#[derive(Debug, Clone, Deserialize)]
843pub(crate) struct ToolCallIn {
844 #[serde(default)]
845 #[allow(dead_code)]
846 id: String,
847 #[serde(rename = "type", default)]
848 #[allow(dead_code)]
849 kind: String,
850 function: ToolCallFunctionIn,
851}
852
853#[derive(Debug, Clone, Deserialize)]
854struct ToolCallFunctionIn {
855 name: String,
856 arguments: String,
861}
862
863#[derive(Debug, Clone, Deserialize)]
866struct ToolDef {
867 #[serde(rename = "type", default)]
868 #[allow(dead_code)]
869 kind: String,
870 function: ToolFunctionDef,
871}
872
873#[derive(Debug, Clone, Deserialize)]
874struct ToolFunctionDef {
875 name: String,
876 #[serde(default)]
877 description: Option<String>,
878 #[serde(default)]
879 parameters: Option<serde_json::Value>,
880}
881
882#[derive(Debug, Clone, Deserialize)]
893#[serde(untagged)]
894enum ToolChoice {
895 Mode(String),
896 Specific(serde_json::Value),
897}
898
899#[derive(Deserialize)]
902#[serde(untagged)]
903enum StopParam {
904 One(String),
905 Many(Vec<String>),
906}
907
908#[derive(Deserialize)]
909struct ChatCompletionRequest {
910 model: String,
911 messages: Vec<ChatMessage>,
912 #[serde(default = "default_max_tokens")]
913 max_tokens: usize,
914 #[serde(default)]
915 temperature: Option<f32>,
916 #[serde(default)]
917 top_p: Option<f32>,
918 #[serde(default)]
923 min_p: Option<f32>,
924 #[serde(default)]
925 top_k: Option<usize>,
926 #[serde(default)]
927 repetition_penalty: Option<f32>,
928 #[serde(flatten)]
932 extra_samplers: crate::sampling_knobs::ExtraSamplerFields,
933 #[serde(flatten)]
937 unimplemented: crate::unimplemented_fields::UnimplementedFields,
938 #[serde(default)]
939 seed: Option<u64>,
940 #[serde(default)]
941 stop: Option<StopParam>,
942 #[serde(default)]
943 stream: Option<bool>,
944 #[serde(default)]
957 stream_resumable: Option<bool>,
958 #[serde(default)]
969 ignore_eos: Option<bool>,
970 #[serde(default)]
971 tools: Vec<ToolDef>,
972 #[serde(default)]
973 tool_choice: Option<ToolChoice>,
974 #[serde(default)]
982 chat_template_kwargs: Option<serde_json::Map<String, serde_json::Value>>,
983 #[serde(default)]
991 reasoning_effort: Option<String>,
992 #[serde(default)]
996 thinking: Option<ThinkingSwitch>,
997 #[serde(default)]
1002 session_id: Option<String>,
1003 #[serde(default, deserialize_with = "continuation::deserialize")]
1011 continue_final_message: continuation::ContinueFinalMessage,
1012 #[serde(default, alias = "thinking_budget_tokens")]
1020 reasoning_budget_tokens: Option<reasoning_budget::BudgetTokens>,
1021 #[serde(default)]
1023 logprobs: Option<bool>,
1024 #[serde(default)]
1025 top_logprobs: Option<u32>,
1026 #[serde(default)]
1027 presence_penalty: Option<f32>,
1028 #[serde(default)]
1029 frequency_penalty: Option<f32>,
1030 #[serde(default)]
1031 response_format: Option<serde_json::Value>,
1032 #[serde(default)]
1039 logit_bias: Option<serde_json::Value>,
1040 #[serde(default)]
1044 lora: Option<Vec<frink_api::LoraScaleRequest>>,
1045 #[serde(default)]
1055 samplers: Option<serde_json::Value>,
1056 #[serde(default)]
1066 grammar: Option<String>,
1067}
1068
1069const DEFAULT_CHAT_MAX_TOKENS: usize = 32_768;
1081
1082#[derive(Debug, Clone, Deserialize)]
1084pub(crate) struct ThinkingSwitch {
1085 #[serde(rename = "type")]
1086 pub(crate) kind: String,
1087}
1088
1089const THINKING_KWARG_KEYS: [&str; 4] = [
1093 "enable_thinking",
1094 "thinking",
1095 "thinking_mode",
1096 "reasoning_effort",
1097];
1098
1099const DISABLE_EFFORTS: [&str; 2] = ["none", "off"];
1103
1104fn default_max_tokens() -> usize {
1105 DEFAULT_CHAT_MAX_TOKENS
1106}
1107
1108impl ChatCompletionRequest {
1109 fn sampling_knobs(&self) -> Result<SamplingKnobs, ApiError> {
1118 let mut knobs = SamplingKnobs {
1119 temperature: self.temperature,
1120 top_p: self.top_p,
1121 min_p: self.min_p,
1122 top_k: self.top_k,
1123 repetition_penalty: self.repetition_penalty,
1124 presence_penalty: self.presence_penalty,
1125 frequency_penalty: self.frequency_penalty,
1126 penalty_last_n: None,
1130 sampler_order: unsupported_sampling::parse_sampler_order(
1131 self.samplers.as_ref(),
1132 "/v1/chat/completions",
1133 )?,
1134 ..SamplingKnobs::default()
1135 };
1136 self.extra_samplers.apply(&mut knobs);
1137 Ok(knobs)
1138 }
1139
1140 fn sampling_params(
1141 &self,
1142 model: crate::sampling_knobs::SamplerModel<'_>,
1143 ) -> Result<SamplingParams, ApiError> {
1144 self.sampling_knobs()?.resolve(model).map_err(|e| {
1145 unsupported_feature(&format!("`dry_multiplier` on /v1/chat/completions: {e}"))
1146 })
1147 }
1148
1149 fn stop_sequences(&self) -> Vec<String> {
1150 self.stop
1151 .as_ref()
1152 .map(|s| match s {
1153 StopParam::One(v) => vec![v.clone()],
1154 StopParam::Many(v) => v.clone(),
1155 })
1156 .unwrap_or_default()
1157 }
1158
1159 fn n_logprobs(&self) -> Result<Option<usize>, ApiError> {
1173 const MAX: u32 = 20;
1174 match (self.logprobs, self.top_logprobs) {
1175 (Some(true), Some(n)) if n > MAX => Err(invalid_request(
1176 &format!(
1177 "`top_logprobs` is {n}; this server reports at most {MAX} alternatives per \
1178 position, as upstream does"
1179 ),
1180 "top_logprobs",
1181 )),
1182 (Some(true), Some(n)) => Ok(Some(n as usize)),
1183 (Some(true), None) => Ok(Some(0)),
1187 (_, Some(_)) => Err(invalid_request(
1188 "`top_logprobs` requires `logprobs: true`",
1189 "top_logprobs",
1190 )),
1191 _ => Ok(None),
1192 }
1193 }
1194
1195 fn several_choices(&self) -> bool {
1200 self.unimplemented.n.is_some_and(|n| n > 1)
1201 }
1202
1203 fn tools_active(&self) -> bool {
1204 !self.tools.is_empty()
1205 && !matches!(&self.tool_choice, Some(ToolChoice::Mode(m)) if m == "none")
1206 }
1207
1208 fn forced_tool_choice(&self) -> Result<Option<tool_grammar::Forced<'_>>, ApiError> {
1220 match &self.tool_choice {
1221 Some(ToolChoice::Mode(m)) if m == "required" => Ok(Some(tool_grammar::Forced::Any)),
1222 Some(ToolChoice::Specific(value)) => {
1223 let name = value
1227 .get("function")
1228 .and_then(|f| f.get("name"))
1229 .or_else(|| value.get("name"))
1230 .and_then(|n| n.as_str());
1231 match name {
1232 Some(name) => Ok(Some(tool_grammar::Forced::Named(name))),
1233 None => Err(invalid_request(
1234 "tool_choice must be \"auto\", \"none\", \"required\", or an object with \
1235 function.name",
1236 "tool_choice",
1237 )),
1238 }
1239 }
1240 _ => Ok(None),
1241 }
1242 }
1243
1244 fn tool_specs(&self) -> Vec<tool_grammar::ToolSpec<'_>> {
1246 self.tools
1247 .iter()
1248 .map(|t| tool_grammar::ToolSpec {
1249 name: &t.function.name,
1250 parameters: t.function.parameters.as_ref(),
1251 })
1252 .collect()
1253 }
1254
1255 fn thinking_direction(&self) -> Option<bool> {
1298 if let Some(switch) = &self.thinking {
1299 return match switch.kind.trim().to_ascii_lowercase().as_str() {
1300 "disabled" => Some(false),
1301 "enabled" => Some(true),
1302 _ => None,
1305 };
1306 }
1307 let effort = self.reasoning_effort.as_ref()?;
1308 DISABLE_EFFORTS
1309 .contains(&effort.trim().to_ascii_lowercase().as_str())
1310 .then_some(false)
1311 }
1312
1313 fn resolve_template_kwargs(
1314 &self,
1315 template: &chat_template::PromptTemplate,
1316 ) -> serde_json::Map<String, serde_json::Value> {
1317 let mut kwargs = self.chat_template_kwargs.clone().unwrap_or_default();
1318 let caller_steered = THINKING_KWARG_KEYS.iter().any(|k| kwargs.contains_key(*k));
1322
1323 if !caller_steered {
1324 match self.thinking_direction() {
1325 Some(false) => {
1326 for (k, v) in crate::policy::effort::thinking_off_kwargs() {
1327 kwargs.insert(k, v);
1328 }
1329 return kwargs;
1332 }
1333 Some(true) => {
1334 for (k, v) in crate::policy::effort::thinking_on_kwargs() {
1335 kwargs.insert(k, v);
1336 }
1337 }
1338 None => {}
1339 }
1340 if let Some(effort) = &self.reasoning_effort {
1341 kwargs
1342 .entry("reasoning_effort".to_string())
1343 .or_insert_with(|| serde_json::json!(effort));
1344 }
1345 }
1346
1347 let offered: Vec<serde_json::Value> = if self.tools_active() {
1348 self.tools.iter().map(chat_template::tool_json).collect()
1349 } else {
1350 Vec::new()
1351 };
1352 let thinking = crate::policy::effort::resolve_thinking_mode(Some(&kwargs), Some(&offered));
1353 if thinking == crate::policy::effort::ThinkingMode::Thinking {
1354 for (k, v) in crate::policy::effort::thinking_on_kwargs() {
1355 kwargs.entry(k).or_insert(v);
1356 }
1357 }
1358 match crate::policy::effort::sanitize_effort(&mut kwargs, template.efforts()) {
1359 crate::policy::effort::EffortMapping::Mapped(to) => {
1360 tracing::debug!("reasoning_effort quantized to {}", to.as_str());
1361 }
1362 crate::policy::effort::EffortMapping::Dropped => {
1363 tracing::debug!(
1364 "reasoning_effort dropped: this checkpoint's template grades no gear close \
1365 enough, so its own default applies"
1366 );
1367 }
1368 crate::policy::effort::EffortMapping::Unchanged => {}
1369 }
1370 crate::policy::effort::broadcast_effort_spellings(&mut kwargs);
1371 kwargs
1372 }
1373
1374 fn validate_supported_fields(&self) -> Result<(), ApiError> {
1377 if self.max_tokens == 0 {
1384 return Err(invalid_request(
1385 "max_tokens must be at least 1",
1386 "max_tokens",
1387 ));
1388 }
1389 if let Some(switch) = &self.thinking {
1394 let kind = switch.kind.trim().to_ascii_lowercase();
1395 if kind != "enabled" && kind != "disabled" {
1396 return Err(invalid_request(
1397 "thinking.type must be \"enabled\" or \"disabled\"",
1398 "thinking.type",
1399 ));
1400 }
1401 }
1402 for msg in &self.messages {
1403 if msg.content.as_ref().is_some_and(MessageContent::has_image) {
1404 return Err(unsupported_feature(
1405 "image_url content parts are not implemented (multimodal/VL deferred, see docs/API.md)",
1406 ));
1407 }
1408 }
1409 self.n_logprobs()?;
1413 self.unimplemented.refuse("/v1/chat/completions")?;
1417 unsupported_sampling::refuse_logit_bias(self.logit_bias.as_ref(), "/v1/chat/completions")?;
1418 unsupported_sampling::parse_sampler_order(self.samplers.as_ref(), "/v1/chat/completions")?;
1422 let stated_grammar =
1437 grammar_request::for_request(self.grammar.as_deref(), self.response_format.as_ref())?;
1438 if let Some(forced) = self.forced_tool_choice()? {
1446 if self.tools.is_empty() {
1447 return Err(invalid_request(
1448 "tool_choice forces a tool call, but no tools were offered",
1449 "tool_choice",
1450 ));
1451 }
1452 if let tool_grammar::Forced::Named(name) = forced {
1453 if !self.tools.iter().any(|t| t.function.name == name) {
1454 return Err(invalid_request(
1455 &format!(
1456 "tool_choice names {name:?}, which is not one of the tools offered"
1457 ),
1458 "tool_choice",
1459 ));
1460 }
1461 }
1462 if stated_grammar.is_some() {
1472 return Err(invalid_request(
1473 "a forced tool_choice and a \"grammar\" or response_format \"json_schema\" \
1474 are two different constraints on the same generation; send one",
1475 "tool_choice",
1476 ));
1477 }
1478 if self.json_object_mode() {
1479 return Err(invalid_request(
1480 "a forced tool_choice cannot be combined with response_format json_object: \
1481 the tool-call markers are not JSON",
1482 "tool_choice",
1483 ));
1484 }
1485 }
1486 Ok(())
1487 }
1488
1489 fn effective_stop_sequences(&self) -> Vec<String> {
1496 let mut stop = self.stop_sequences();
1497 if self.tools_active() {
1498 stop.push("</tool_call>".to_string());
1499 }
1500 stop
1501 }
1502
1503 fn json_object_mode(&self) -> bool {
1504 self.response_format
1505 .as_ref()
1506 .and_then(|v| v.get("type"))
1507 .and_then(|v| v.as_str())
1508 == Some("json_object")
1509 }
1510}
1511
1512#[derive(Serialize)]
1513struct ChatCompletionChoice {
1514 index: usize,
1515 message: ChatCompletionResponseMessage,
1516 finish_reason: &'static str,
1517 #[serde(skip_serializing_if = "Option::is_none")]
1522 logprobs: Option<serde_json::Value>,
1523}
1524
1525#[derive(Serialize)]
1526struct ChatCompletionResponseMessage {
1527 role: &'static str,
1528 #[serde(skip_serializing_if = "Option::is_none")]
1529 content: Option<String>,
1530 #[serde(skip_serializing_if = "Option::is_none")]
1534 reasoning_content: Option<String>,
1535 #[serde(skip_serializing_if = "Option::is_none")]
1536 tool_calls: Option<Vec<ToolCallOut>>,
1537}
1538
1539#[derive(Serialize, Clone)]
1540struct ToolCallOut {
1541 id: String,
1542 #[serde(rename = "type")]
1543 kind: &'static str,
1544 function: ToolCallFunctionOut,
1545}
1546
1547#[derive(Serialize, Clone)]
1555struct ToolCallDelta {
1556 index: usize,
1557 #[serde(skip_serializing_if = "Option::is_none")]
1558 id: Option<String>,
1559 #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
1560 kind: Option<&'static str>,
1561 function: ToolCallFunctionDelta,
1562}
1563
1564#[derive(Serialize, Clone, Default)]
1565struct ToolCallFunctionDelta {
1566 #[serde(skip_serializing_if = "Option::is_none")]
1567 name: Option<String>,
1568 #[serde(skip_serializing_if = "Option::is_none")]
1571 arguments: Option<String>,
1572}
1573
1574impl ToolCallDelta {
1575 fn whole(index: usize, name: String, arguments: String) -> Self {
1577 ToolCallDelta {
1578 index,
1579 id: Some(format!("call_{index}")),
1580 kind: Some("function"),
1581 function: ToolCallFunctionDelta {
1582 name: Some(name),
1583 arguments: Some(arguments),
1584 },
1585 }
1586 }
1587
1588 fn opening(index: usize, name: String) -> Self {
1590 ToolCallDelta {
1591 index,
1592 id: Some(format!("call_{index}")),
1593 kind: Some("function"),
1594 function: ToolCallFunctionDelta {
1595 name: Some(name),
1596 arguments: Some(String::new()),
1597 },
1598 }
1599 }
1600
1601 fn arguments(index: usize, fragment: String) -> Self {
1603 ToolCallDelta {
1604 index,
1605 id: None,
1606 kind: None,
1607 function: ToolCallFunctionDelta {
1608 name: None,
1609 arguments: Some(fragment),
1610 },
1611 }
1612 }
1613}
1614
1615#[derive(Serialize, Clone)]
1616struct ToolCallFunctionOut {
1617 name: String,
1618 arguments: String,
1622}
1623
1624#[derive(Serialize)]
1625struct ChatCompletionResponse {
1626 id: String,
1627 request_id: String,
1633 object: &'static str,
1634 model: String,
1635 choices: Vec<ChatCompletionChoice>,
1636 usage: generate::Usage,
1641 frink_cache: &'static str,
1652}
1653
1654#[derive(Serialize)]
1655struct ChatCompletionChunkDelta {
1656 #[serde(skip_serializing_if = "Option::is_none")]
1657 role: Option<&'static str>,
1658 #[serde(skip_serializing_if = "Option::is_none")]
1659 content: Option<String>,
1660 #[serde(skip_serializing_if = "Option::is_none")]
1662 reasoning_content: Option<String>,
1663 #[serde(skip_serializing_if = "Option::is_none")]
1664 tool_calls: Option<Vec<ToolCallDelta>>,
1665}
1666
1667#[derive(Serialize)]
1668struct ChatCompletionChunkChoice {
1669 index: usize,
1670 delta: ChatCompletionChunkDelta,
1671 finish_reason: Option<&'static str>,
1672}
1673
1674#[derive(Serialize)]
1675struct ChatCompletionChunk {
1676 id: String,
1677 #[serde(skip_serializing_if = "Option::is_none")]
1684 request_id: Option<String>,
1685 object: &'static str,
1686 model: String,
1687 choices: Vec<ChatCompletionChunkChoice>,
1688 #[serde(skip_serializing_if = "Option::is_none")]
1691 usage: Option<generate::Usage>,
1692}
1693
1694async fn health(State(state): State<Arc<AppState>>) -> Response {
1700 let snapshot = state.detection.snapshot();
1701 let mut capabilities = snapshot.capabilities;
1702 let active = state.active();
1703
1704 capabilities.push(match active.as_deref() {
1707 None => frink_api::Capability::unavailable(
1712 frink_api::health::capability::REAL_WEIGHTS,
1713 frink_api::health::reason::MODEL_NOT_LOADED,
1714 "No model is loaded. POST /admin/models/load with an id from GET /admin/models.",
1715 ),
1716 Some(active) if active.is_synthetic() => frink_api::Capability::unavailable(
1717 frink_api::health::capability::REAL_WEIGHTS,
1718 frink_api::health::reason::MODEL_NOT_LOADED,
1719 "Serving synthetic random weights: set FRINK_MODEL_PATH (or -m) to a real \
1720 checkpoint. Output from this model is noise.",
1721 ),
1722 Some(active) if active.encoder().is_some() => {
1732 let endpoints = active
1733 .encoder()
1734 .map(|e| encoder_endpoints(e))
1735 .unwrap_or_default();
1736 let served_by = match endpoints.is_empty() {
1737 true => "no endpoint in this build serves it".to_string(),
1738 false => format!("served by {}", endpoints.join(" and ")),
1739 };
1740 frink_api::Capability::available(
1741 frink_api::health::capability::REAL_WEIGHTS,
1742 format!(
1743 "Serving the real embedding checkpoint '{}'. This is an ENCODER, \
1744 {served_by}; generation endpoints refuse it.",
1745 active.name(),
1746 ),
1747 )
1748 }
1749 Some(active) => frink_api::Capability::available(
1750 frink_api::health::capability::REAL_WEIGHTS,
1751 format!("Serving the real checkpoint '{}'.", active.name()),
1752 ),
1753 });
1754 capabilities.push(if active.as_ref().is_some_and(|a| a.batcher.is_some()) {
1755 frink_api::Capability::available(
1756 frink_api::health::capability::CONTINUOUS_BATCHING,
1757 if state.continuous_batching_enabled && continuous_batching_env().is_none() {
1758 "On by default on Metal. Concurrent requests share one batched decode worker."
1759 } else {
1760 "Concurrent requests share one batched decode step."
1761 },
1762 )
1763 } else if state.metal_private_decode_gate.is_some() {
1764 frink_api::Capability::unavailable(
1765 frink_api::health::capability::CONTINUOUS_BATCHING,
1766 frink_api::health::reason::DISABLED,
1767 "Off; private Metal decodes serialize (one at a time). Set FRINK_CONTINUOUS_BATCHING=1 or --cont-batching for parallel serving.",
1768 )
1769 } else {
1770 frink_api::Capability::unavailable(
1771 frink_api::health::capability::CONTINUOUS_BATCHING,
1772 frink_api::health::reason::DISABLED,
1773 "Off; set FRINK_CONTINUOUS_BATCHING=1 (incompatible with a KV pool or prefix cache).",
1774 )
1775 });
1776
1777 let last_request_ms = state
1778 .last_request_ms
1779 .load(std::sync::atomic::Ordering::Relaxed);
1780 let uptime = state.started_at.elapsed();
1781 let health_state = if active.is_none() {
1789 frink_api::HealthState::Unavailable
1790 } else {
1791 snapshot.state
1792 };
1793 let body = frink_api::HealthResponse {
1794 state: health_state,
1795 reason: match health_state {
1796 frink_api::HealthState::Ready => None,
1797 frink_api::HealthState::Unavailable => {
1798 Some(frink_api::health::reason::MODEL_NOT_LOADED.to_string())
1799 }
1800 frink_api::HealthState::Detecting => {
1801 Some(frink_api::health::reason::DETECTING.to_string())
1802 }
1803 },
1804 detail: match health_state {
1805 frink_api::HealthState::Ready => None,
1806 frink_api::HealthState::Unavailable => Some(
1807 "No model is loaded. POST /admin/models/load with an id from GET /admin/models."
1808 .to_string(),
1809 ),
1810 frink_api::HealthState::Detecting => {
1811 Some("Probing available compute backends.".to_string())
1812 }
1813 },
1814 model: active
1815 .as_deref()
1816 .map(|active| frink_api::health::ModelSummary {
1817 id: active.name().to_string(),
1818 tokenizer: active.tokenizer_kind().to_string(),
1819 synthetic_weights: active.is_synthetic(),
1820 }),
1821 capabilities,
1822 version: env!("CARGO_PKG_VERSION").to_string(),
1823 pid: std::process::id(),
1824 uptime_seconds: uptime.as_secs_f64(),
1825 server_time_unix_ms: std::time::SystemTime::now()
1826 .duration_since(std::time::UNIX_EPOCH)
1827 .map(|d| d.as_millis().min(u64::MAX as u128) as u64)
1828 .unwrap_or(0),
1829 last_request_age_seconds: (last_request_ms > 0)
1830 .then(|| uptime.as_secs_f64() - (last_request_ms as f64 / 1000.0))
1831 .map(|age| age.max(0.0)),
1832 };
1833
1834 let status =
1835 StatusCode::from_u16(body.state.http_status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
1836 (status, Json(body)).into_response()
1837}
1838
1839async fn list_models(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
1840 let Some(active) = state.active() else {
1844 return Json(serde_json::json!({ "object": "list", "data": [] }));
1845 };
1846 let mut model_entry = serde_json::json!({
1847 "id": active.name(),
1848 "object": "model",
1849 "frink_synthetic_weights": active.is_synthetic(),
1850 "frink_tokenizer": active.tokenizer_kind(),
1851 });
1852 if let Some(encoder) = active.encoder() {
1858 model_entry["frink_model_kind"] = serde_json::json!("embedding");
1859 model_entry["frink_endpoints"] = serde_json::json!(encoder_endpoints(encoder));
1860 model_entry["frink_n_embd"] = serde_json::json!(encoder.n_embd());
1861 model_entry["frink_pooling"] = serde_json::json!(encoder.pooling_type().name());
1862 model_entry["frink_context_length"] = serde_json::json!(encoder.n_ctx_train());
1863 }
1864 if let Some(model) = active.generative_opt() {
1872 let parser_configured = active.reasoning_format().is_some();
1873 let gears = model.chat_template().think_gears(parser_configured);
1874 if !gears.is_empty() {
1875 model_entry["supported_reasoning_efforts"] = serde_json::json!(gears.supported);
1876 if let Some(default) = &gears.default {
1877 model_entry["default_reasoning_effort"] = serde_json::json!(default);
1878 }
1879 model_entry["reasoning_effort_kwargs"] = serde_json::json!(gears.kwargs);
1882 }
1883 }
1884 if let Some(mcp) = &state.mcp {
1885 model_entry["frink_mcp"] = mcp.models_metadata();
1886 }
1887 Json(serde_json::json!({
1888 "object": "list",
1889 "data": [model_entry]
1890 }))
1891}
1892
1893async fn serving_stats(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
1907 let now_ms = state.uptime().as_millis().min(u64::MAX as u128) as u64;
1908 let mut serving = state.serving.lock().unwrap_or_else(|p| p.into_inner());
1909 let active = state.active();
1910 Json(serde_json::json!({
1911 "model": active.as_ref().map(|a| a.name()),
1912 "state": state
1913 .maintenance
1914 .lock()
1915 .unwrap_or_else(|p| p.into_inner())
1916 .state()
1917 .as_str(),
1918 "uptime_s": state.uptime().as_secs(),
1919 "throughput": {
1920 "decode_tps": (serving.decode_tokens_per_second(now_ms) * 10.0).round() / 10.0,
1921 "prefill_tps": (serving.prefill_tokens_per_second(now_ms) * 10.0).round() / 10.0,
1922 },
1923 "requests": {
1924 "active": state.cancels.live_count(),
1925 "completed": state.stats.recorded_total(),
1926 "p95_ms": state.stats.p95_duration_ms(),
1927 "ttft_mean_ms": state.stats.ttft_mean_ms(),
1928 "prompt_tokens_total": state.stats.tokens_prompt_total(),
1929 "completion_tokens_total": state.stats.tokens_generated_total(),
1930 },
1931 "pools": cache_admin::pool_gauges(&state),
1937 "memory": cache_admin::footprint_json(&state),
1940 }))
1941}
1942
1943#[derive(Deserialize)]
1944struct RequestsQuery {
1945 #[serde(default)]
1946 since: u64,
1947 #[serde(default = "default_requests_limit")]
1948 limit: usize,
1949}
1950
1951fn default_requests_limit() -> usize {
1952 stats::MAX_PAGE
1953}
1954
1955async fn recent_requests(
1963 State(state): State<Arc<AppState>>,
1964 axum::extract::Query(q): axum::extract::Query<RequestsQuery>,
1965) -> Json<serde_json::Value> {
1966 let (rows, cursor, missed) = state.stats.page(q.since, q.limit);
1967 Json(serde_json::json!({
1968 "requests": rows,
1969 "next_cursor": cursor,
1970 "missed": missed,
1971 "total": state.stats.recorded_total(),
1972 }))
1973}
1974
1975#[derive(Serialize)]
1976struct CombinedCacheStats {
1977 response_cache: response_cache::CacheStats,
1978 prefix_cache: Option<frink_models::PrefixCacheStats>,
1980}
1981
1982async fn cache_stats(State(state): State<Arc<AppState>>) -> Json<CombinedCacheStats> {
1983 Json(CombinedCacheStats {
1984 response_cache: lock_cache(&state.response_cache).stats(),
1985 prefix_cache: state
1986 .prefix_cache
1987 .as_ref()
1988 .map(|pc| pc.lock().unwrap_or_else(|p| p.into_inner()).stats()),
1989 })
1990}
1991
1992async fn metrics(State(state): State<Arc<AppState>>) -> Response {
1997 use std::sync::atomic::Ordering;
1998
1999 let cache_stats = lock_cache(&state.response_cache).stats();
2000 let active = state.active();
2001 let requests_total = state.requests_total.load(Ordering::Relaxed);
2002 let errors_total = state.request_errors_total.load(Ordering::Relaxed);
2003 let uptime = state.started_at.elapsed().as_secs_f64();
2004
2005 let body = format!(
2006 "# HELP frink_requests_total Total chat completion requests received.\n\
2007 # TYPE frink_requests_total counter\n\
2008 frink_requests_total {requests_total}\n\
2009 # HELP frink_request_errors_total Total chat completion requests that returned an error.\n\
2010 # TYPE frink_request_errors_total counter\n\
2011 frink_request_errors_total {errors_total}\n\
2012 # HELP frink_cache_hits_total Whole-response cache hits.\n\
2013 # TYPE frink_cache_hits_total counter\n\
2014 frink_cache_hits_total {}\n\
2015 # HELP frink_cache_misses_total Whole-response cache misses.\n\
2016 # TYPE frink_cache_misses_total counter\n\
2017 frink_cache_misses_total {}\n\
2018 # HELP frink_cache_entries Current whole-response cache entry count.\n\
2019 # TYPE frink_cache_entries gauge\n\
2020 frink_cache_entries {}\n\
2021 # HELP frink_synthetic_weights 1 if serving synthetic random weights instead of a real checkpoint.\n\
2022 # TYPE frink_synthetic_weights gauge\n\
2023 frink_synthetic_weights {}\n\
2024 # HELP frink_uptime_seconds Seconds since this server process started.\n\
2025 # TYPE frink_uptime_seconds gauge\n\
2026 frink_uptime_seconds {uptime}\n",
2027 cache_stats.hits,
2028 cache_stats.misses,
2029 cache_stats.entries,
2030 active
2034 .as_ref()
2035 .map(|a| a.is_synthetic() as u8)
2036 .unwrap_or(0),
2037 );
2038
2039 let body = match active
2043 .as_ref()
2044 .and_then(|a| a.expert_store_stats())
2045 {
2046 Some(es) => format!(
2047 "{body}\
2048 # HELP frink_expert_cache_hits_total Expert-store cache hits.\n\
2049 # TYPE frink_expert_cache_hits_total counter\n\
2050 frink_expert_cache_hits_total {}\n\
2051 # HELP frink_expert_cache_misses_total Expert-store cache misses (source reads).\n\
2052 # TYPE frink_expert_cache_misses_total counter\n\
2053 frink_expert_cache_misses_total {}\n\
2054 # HELP frink_expert_cache_evictions_total Expert-store LRU evictions.\n\
2055 # TYPE frink_expert_cache_evictions_total counter\n\
2056 frink_expert_cache_evictions_total {}\n\
2057 # HELP frink_expert_cache_pass_throughs_total Acquires served uncached (entry could not fit the budget).\n\
2058 # TYPE frink_expert_cache_pass_throughs_total counter\n\
2059 frink_expert_cache_pass_throughs_total {}\n\
2060 # HELP frink_expert_cache_bytes_read_total Bytes read from the checkpoint for expert misses.\n\
2061 # TYPE frink_expert_cache_bytes_read_total counter\n\
2062 frink_expert_cache_bytes_read_total {}\n\
2063 # HELP frink_expert_cache_resident_bytes Current expert-cache footprint in bytes.\n\
2064 # TYPE frink_expert_cache_resident_bytes gauge\n\
2065 frink_expert_cache_resident_bytes {}\n",
2066 es.hits, es.misses, es.evictions, es.pass_throughs, es.bytes_read, es.resident_bytes,
2067 ),
2068 None => body,
2069 };
2070
2071 let body = match active.as_ref().and_then(|a| a.batcher.as_ref()) {
2076 Some(batcher) => {
2077 let sched = batcher.stats();
2078 format!(
2079 "{body}\
2080 # HELP frink_prefill_chunks_total Bounded prefill chunks the batch scheduler has run.\n\
2081 # TYPE frink_prefill_chunks_total counter\n\
2082 frink_prefill_chunks_total {}\n\
2083 # HELP frink_prefill_tokens_total Prompt tokens run through chunked prefill.\n\
2084 # TYPE frink_prefill_tokens_total counter\n\
2085 frink_prefill_tokens_total {}\n\
2086 # HELP frink_decode_steps_total Batched decode steps the batch scheduler has run.\n\
2087 # TYPE frink_decode_steps_total counter\n\
2088 frink_decode_steps_total {}\n\
2089 # HELP frink_scheduler_queue_depth Requests waiting for admission to the batch scheduler.\n\
2090 # TYPE frink_scheduler_queue_depth gauge\n\
2091 frink_scheduler_queue_depth {}\n\
2092 # HELP frink_scheduler_queue_rejected_total Requests refused with 503 because the admission queue was full.\n\
2093 # TYPE frink_scheduler_queue_rejected_total counter\n\
2094 frink_scheduler_queue_rejected_total {}\n\
2095 # HELP frink_kv_blocks_total KV blocks in the scheduler's admission budget (0 when unconfigured).\n\
2096 # TYPE frink_kv_blocks_total gauge\n\
2097 frink_kv_blocks_total {}\n\
2098 # HELP frink_kv_blocks_free KV blocks not reserved by an in-flight request.\n\
2099 # TYPE frink_kv_blocks_free gauge\n\
2100 frink_kv_blocks_free {}\n\
2101 # HELP frink_kv_block_size Token positions per KV block.\n\
2102 # TYPE frink_kv_block_size gauge\n\
2103 frink_kv_block_size {}\n\
2104 # HELP frink_kv_rejected_too_large_total Requests refused with 400 because they exceed the whole KV block budget.\n\
2105 # TYPE frink_kv_rejected_too_large_total counter\n\
2106 frink_kv_rejected_too_large_total {}\n\
2107 # HELP frink_kv_rejected_context_length_total Requests refused with 400 for exceeding the per-request context ceiling.\n\
2108 # TYPE frink_kv_rejected_context_length_total counter\n\
2109 frink_kv_rejected_context_length_total {}\n\
2110 # HELP frink_scheduler_aborted_total Requests the batch scheduler stopped because they were cancelled.\n\
2111 # TYPE frink_scheduler_aborted_total counter\n\
2112 frink_scheduler_aborted_total {}\n\
2113 # HELP frink_scheduler_max_seqs Cap on in-flight sequences (-np / FRINK_CB_MAX_SEQS); 0 when unlimited.\n\
2114 # TYPE frink_scheduler_max_seqs gauge\n\
2115 frink_scheduler_max_seqs {}\n\
2116 # HELP frink_scheduler_prefill_chunk Prompt tokens per prefill chunk (-b / -ub / FRINK_CB_PREFILL_CHUNK).\n\
2117 # TYPE frink_scheduler_prefill_chunk gauge\n\
2118 frink_scheduler_prefill_chunk {}\n",
2119 sched.prefill_chunks,
2120 sched.prefill_tokens,
2121 sched.decode_steps,
2122 sched.queue_depth,
2123 sched.queue_rejected,
2124 sched.kv_blocks_total,
2125 sched.kv_blocks_free,
2126 sched.kv_block_size,
2127 sched.kv_rejected_too_large,
2128 sched.kv_rejected_context_length,
2129 sched.aborted,
2130 sched.max_seqs,
2131 sched.prefill_chunk,
2132 )
2133 }
2134 None => body,
2135 };
2136
2137 (
2138 [(
2139 axum::http::header::CONTENT_TYPE,
2140 "text/plain; version=0.0.4",
2141 )],
2142 body,
2143 )
2144 .into_response()
2145}
2146
2147pub(crate) type ApiError = (StatusCode, Json<serde_json::Value>);
2148
2149pub(crate) fn invalid_request(message: &str, param: &str) -> ApiError {
2155 (
2156 StatusCode::BAD_REQUEST,
2157 Json(serde_json::json!({"error": {
2158 "message": message,
2159 "type": "invalid_request_error",
2160 "param": param,
2161 "code": null,
2162 }})),
2163 )
2164}
2165
2166pub(crate) fn unsupported_feature(message: &str) -> ApiError {
2167 (
2168 StatusCode::NOT_IMPLEMENTED,
2169 Json(serde_json::json!({"error": {"message": message, "type": "unsupported"}})),
2170 )
2171}
2172
2173pub(crate) fn decode_error_response(e: generate::DecodeError) -> ApiError {
2174 let status = match e {
2175 generate::DecodeError::TokenOutOfVocab { .. } => StatusCode::BAD_REQUEST,
2176 generate::DecodeError::Unsupported(_) => StatusCode::NOT_IMPLEMENTED,
2180 generate::DecodeError::KvBudgetExceeded { .. } => StatusCode::BAD_REQUEST,
2185 generate::DecodeError::KvPoolExhausted | generate::DecodeError::QueueFull { .. } => {
2191 StatusCode::SERVICE_UNAVAILABLE
2192 }
2193 generate::DecodeError::GrammarConstraint { .. } => StatusCode::BAD_REQUEST,
2197 generate::DecodeError::ReasoningBudget { .. } => StatusCode::INTERNAL_SERVER_ERROR,
2201 };
2202 tracing::warn!("decode error: {e}");
2203 let mut body = serde_json::json!({"error": {"message": e.to_string()}});
2204 if let generate::DecodeError::KvBudgetExceeded {
2210 binding,
2211 estimated_bytes,
2212 limit_bytes,
2213 positions,
2214 positions_limit,
2215 ..
2216 } = &e
2217 {
2218 body["error"]["type"] = serde_json::json!("invalid_request_error");
2219 body["error"]["code"] = serde_json::json!(binding);
2220 body["error"]["binding"] = serde_json::json!(binding);
2221 body["error"]["estimated_bytes"] = serde_json::json!(estimated_bytes);
2222 body["error"]["limit_bytes"] = serde_json::json!(limit_bytes);
2223 body["error"]["positions"] = serde_json::json!(positions);
2224 body["error"]["positions_limit"] = serde_json::json!(positions_limit);
2225 }
2226 if let Some(secs) = e.retry_after_secs() {
2230 body["error"]["retry_after_seconds"] = serde_json::json!(secs);
2231 }
2232 (status, Json(body))
2233}
2234
2235pub(crate) fn join_error_response(e: tokio::task::JoinError) -> ApiError {
2236 tracing::error!("generation task panicked: {e}");
2237 (
2238 StatusCode::INTERNAL_SERVER_ERROR,
2239 Json(serde_json::json!({"error": {"message": "internal error during generation"}})),
2240 )
2241}
2242
2243#[allow(clippy::too_many_arguments)] fn run_generation_emit(
2253 model: &Model,
2254 prompt: &str,
2255 params: &GenerationParams,
2256 kv_pool: Option<&generate::KvPoolConfig>,
2257 paged_kv: Option<&generate::PagedKvConfig>,
2258 prefix_cache: Option<&Mutex<PrefixCache>>,
2259 continuous_batcher: Option<&serving::batch::ContinuousBatcher>,
2260 ceiling: Option<&budget::ContextCeiling>,
2261 metal_private_decode_gate: Option<&std::sync::Mutex<()>>,
2262 mut emit: impl FnMut(&str),
2263 ) -> Result<generate::Generated, generate::DecodeError> {
2269 let synthetic = model.is_synthetic();
2270 let _lora_lease = lora::lease(model, params.lora.as_deref());
2275 let mut chunks: Vec<Vec<String>> = vec![Vec::new(); params.n.max(1)];
2276 let params = &{
2282 let mut resolved = params.clone();
2283 resolved.stop_token_ids = crate::stop::resolve_stop_tokens(&resolved.stop, |text| {
2284 model.encode(text, SpecialTokens::Parse)
2285 });
2286 resolved.reasoning_budget = resolved
2292 .reasoning_budget
2293 .armed(resolved.reasoning, prompt, |text| {
2294 model.encode(text, SpecialTokens::Parse)
2295 })
2296 .map_err(|detail| generate::DecodeError::ReasoningBudget { detail })?;
2297 resolved
2298 };
2299 let used_batcher = matches!((model, continuous_batcher), (Model::Gguf(_), Some(_)));
2300 let _metal_private_guard =
2301 acquire_metal_private_decode_gate(metal_private_decode_gate, used_batcher);
2302 let (finishes, prompt_rows, prompt_ids, usage) = match model {
2303 Model::Gguf(m) => {
2304 if let Some(batcher) = continuous_batcher {
2305 let mut tokens = m.tokenizer.encode(prompt, SpecialTokens::Parse);
2306 frink_models::tokenizer::prepend_bos(&mut tokens, m.bos_id);
2307 let (finish, _generated_ids, text, usage) = if synthetic {
2308 batcher.generate(tokens, params.clone(), m.stop_tokens.clone())?
2309 } else {
2310 batcher.generate_streaming(
2311 tokens,
2312 params.clone(),
2313 m.stop_tokens.clone(),
2314 Some(|chunk: &str| {
2315 if !chunk.is_empty() {
2316 chunks[0].push(chunk.to_string());
2317 emit(chunk);
2318 }
2319 }),
2320 )?
2321 };
2322 if !text.is_empty() && chunks[0].is_empty() {
2323 chunks[0].push(text);
2324 }
2325 (vec![(finish, Vec::new())], Vec::new(), Vec::new(), usage)
2335 } else {
2336 generate::generate(
2337 &m.decoder,
2338 m.tokenizer.as_ref(),
2339 &m.stop_tokens,
2340 m.bos_id,
2341 prompt,
2342 params,
2343 kv_pool,
2344 paged_kv,
2345 prefix_cache,
2346 ceiling,
2347 |choice, chunk| {
2348 chunks[choice].push(chunk.to_string());
2349 if !synthetic && choice == 0 {
2353 emit(chunk);
2354 }
2355 },
2356 )?
2357 }
2358 }
2359 Model::Kimi(m) => generate::generate_engine(
2360 &m.engine,
2361 &m.tokenizer,
2362 &m.stop_tokens,
2363 None,
2364 prompt,
2365 params,
2366 |chunk| {
2367 chunks[0].push(chunk.to_string());
2368 if !synthetic {
2369 emit(chunk);
2370 }
2371 },
2372 )?,
2373 Model::Mla(m) => generate::generate_engine(
2374 &m.engine,
2375 &m.tokenizer,
2376 &m.stop_tokens,
2377 m.bos_id,
2378 prompt,
2379 params,
2380 |chunk| {
2381 chunks[0].push(chunk.to_string());
2382 if !synthetic {
2383 emit(chunk);
2384 }
2385 },
2386 )?,
2387 Model::Gemma4(m) => generate::generate_engine(
2388 &m.engine,
2389 &m.tokenizer,
2390 &m.stop_tokens,
2391 m.bos_id,
2392 prompt,
2393 params,
2394 |chunk| {
2395 chunks[0].push(chunk.to_string());
2396 if !synthetic {
2397 emit(chunk);
2398 }
2399 },
2400 )?,
2401 Model::Glm52(m) => generate::generate_engine(
2402 &m.engine,
2403 &m.tokenizer,
2404 &m.stop_tokens,
2405 m.bos_id,
2406 prompt,
2407 params,
2408 |chunk| {
2409 chunks[0].push(chunk.to_string());
2410 if !synthetic {
2411 emit(chunk);
2412 }
2413 },
2414 )?,
2415 };
2416
2417 let mut full = chunks[0].concat();
2418 if synthetic {
2419 full = format!(
2420 "[frink synthetic-weight demo: no real checkpoint loaded -- set FRINK_MODEL_PATH \
2421 to serve a real model. Decoded ids -> {full:?}]"
2422 );
2423 emit(&full);
2424 } else if used_batcher && !full.is_empty() && chunks[0].is_empty() {
2425 emit(&full);
2426 }
2427
2428 debug_assert_eq!(finishes.len(), chunks.len(), "one finish reason per choice");
2434 let mut out: Vec<generate::GeneratedChoice> = finishes
2435 .into_iter()
2436 .zip(chunks.into_iter().map(|c| c.concat()))
2437 .map(|((finish, logprobs), text)| generate::GeneratedChoice {
2438 finish,
2439 text,
2440 logprobs,
2441 })
2442 .collect();
2443 if let Some(first) = out.first_mut() {
2444 if synthetic {
2452 first.logprobs.clear();
2453 }
2454 first.text = full;
2455 }
2456 Ok(generate::Generated {
2457 choices: out,
2458 prompt_rows,
2459 prompt_ids,
2460 usage,
2461 })
2462}
2463
2464#[allow(clippy::too_many_arguments)] pub(crate) fn run_generation(
2469 model: &Model,
2470 prompt: &str,
2471 params: &GenerationParams,
2472 kv_pool: Option<&generate::KvPoolConfig>,
2473 paged_kv: Option<&generate::PagedKvConfig>,
2474 prefix_cache: Option<&Mutex<PrefixCache>>,
2475 continuous_batcher: Option<&serving::batch::ContinuousBatcher>,
2476 ceiling: Option<&budget::ContextCeiling>,
2477 metal_private_decode_gate: Option<&std::sync::Mutex<()>>,
2478 ) -> Result<generate::Generated, generate::DecodeError> {
2481 run_generation_emit(
2482 model,
2483 prompt,
2484 params,
2485 kv_pool,
2486 paged_kv,
2487 prefix_cache,
2488 continuous_batcher,
2489 ceiling,
2490 metal_private_decode_gate,
2491 |_| {},
2492 )
2493}
2494
2495pub(crate) fn prompt_from_messages(
2506 messages: &[ChatMessage],
2507 template: &chat_template::PromptTemplate,
2508 tools: &[ToolDef],
2509 extra: serde_json::Map<String, serde_json::Value>,
2510) -> Result<String, ApiError> {
2511 let rendered = if tools.is_empty() || template.handles_tools() {
2512 template.render(messages, tools, extra)
2513 } else {
2514 let mut with_preamble = Vec::with_capacity(messages.len() + 1);
2515 with_preamble.push(ChatMessage {
2516 role: "system".to_string(),
2517 content: Some(MessageContent::Text(tool_preamble(tools))),
2518 tool_calls: None,
2519 tool_call_id: None,
2520 reasoning_content: None,
2521 });
2522 with_preamble.extend_from_slice(messages);
2523 template.render(&with_preamble, &[], extra)
2524 };
2525 rendered.map_err(template_error_response)
2526}
2527
2528fn template_error_response(err: frink_models::chat_template::TemplateError) -> ApiError {
2533 (
2534 StatusCode::BAD_REQUEST,
2535 Json(serde_json::json!({
2536 "error": {
2537 "message": format!("chat template failed to render: {err}"),
2538 "type": "invalid_request_error",
2539 "param": "messages",
2540 "code": null,
2541 }
2542 })),
2543 )
2544}
2545
2546fn tool_preamble(tools: &[ToolDef]) -> String {
2559 let mut out = String::from(
2560 "You can call tools to help answer the user. To call a tool, respond with \
2561 EXACTLY one line in this format and nothing else:\n\
2562 <tool_call>{\"name\": \"<tool name>\", \"arguments\": {<arguments as a JSON \
2563 object matching that tool's parameters>}}</tool_call>\n\n\
2564 Available tools:\n",
2565 );
2566 for t in tools {
2567 out.push_str(&format!(
2568 "- {}: {}\n parameters (JSON schema): {}\n",
2569 t.function.name,
2570 t.function.description.as_deref().unwrap_or(""),
2571 t.function
2572 .parameters
2573 .as_ref()
2574 .map(|v| v.to_string())
2575 .unwrap_or_else(|| "{}".to_string()),
2576 ));
2577 }
2578 out
2579}
2580
2581fn tool_call_deltas(
2591 events: Vec<crate::policy::parser::ToolCallEvent>,
2592 opened: &std::cell::Cell<usize>,
2593) -> (String, Vec<ToolCallDelta>) {
2594 let mut text = String::new();
2595 let mut deltas = Vec::new();
2596 for event in events {
2597 match event {
2598 crate::policy::parser::ToolCallEvent::Text(chunk) => text.push_str(&chunk),
2599 crate::policy::parser::ToolCallEvent::CallStart { index, name } => {
2600 opened.set(opened.get().max(index + 1));
2601 deltas.push(ToolCallDelta::opening(index, name));
2602 }
2603 crate::policy::parser::ToolCallEvent::CallArguments { index, fragment } => {
2604 if !fragment.is_empty() {
2605 deltas.push(ToolCallDelta::arguments(index, fragment));
2606 }
2607 }
2608 crate::policy::parser::ToolCallEvent::CallEnd { .. } => {}
2609 }
2610 }
2611 (text, deltas)
2612}
2613
2614fn build_response_message(
2625 text: String,
2626 tools: &[ToolDef],
2627 posture: output::OutputPosture,
2628 base_finish: &'static str,
2629) -> (ChatCompletionResponseMessage, &'static str) {
2630 let parsed = output::parse_output(&text, tools, posture);
2631 let calls: Vec<ToolCallOut> = parsed
2632 .calls
2633 .into_iter()
2634 .enumerate()
2635 .map(|(index, call)| ToolCallOut {
2636 id: format!("call_{index}"),
2637 kind: "function",
2638 function: ToolCallFunctionOut {
2639 name: call.name,
2640 arguments: call.arguments,
2641 },
2642 })
2643 .collect();
2644 if !calls.is_empty() {
2645 return (
2646 ChatCompletionResponseMessage {
2647 role: "assistant",
2648 content: None,
2649 reasoning_content: parsed.reasoning,
2650 tool_calls: Some(calls),
2651 },
2652 "tool_calls",
2653 );
2654 }
2655 (
2656 ChatCompletionResponseMessage {
2657 role: "assistant",
2658 content: Some(parsed.content),
2659 reasoning_content: parsed.reasoning,
2660 tool_calls: None,
2661 },
2662 base_finish,
2663 )
2664}
2665
2666fn resolve_history(state: &AppState, req: &ChatCompletionRequest) -> Vec<ChatMessage> {
2671 let mut history = match &req.session_id {
2672 Some(id) => state.sessions.extend_and_get(id, &req.messages),
2673 None => req.messages.clone(),
2674 };
2675 if req.json_object_mode() {
2676 inject_json_object_system_hint(&mut history);
2677 }
2678 history
2679}
2680
2681fn inject_json_object_system_hint(messages: &mut Vec<ChatMessage>) {
2682 const HINT: &str =
2683 "You must respond with valid JSON only (a single JSON object, no markdown fences).";
2684 if let Some(sys) = messages.iter_mut().find(|m| m.role == "system") {
2685 match &mut sys.content {
2686 Some(MessageContent::Text(s)) if !s.contains("JSON") => {
2687 s.push_str("\n\n");
2688 s.push_str(HINT);
2689 }
2690 None => {
2691 sys.content = Some(MessageContent::Text(HINT.to_string()));
2692 }
2693 _ => {}
2694 }
2695 } else {
2696 messages.insert(
2697 0,
2698 ChatMessage {
2699 role: "system".to_string(),
2700 content: Some(MessageContent::Text(HINT.to_string())),
2701 tool_calls: None,
2702 tool_call_id: None,
2703 reasoning_content: None,
2704 },
2705 );
2706 }
2707}
2708
2709async fn chat_completions(
2710 State(state): State<Arc<AppState>>,
2711 headers: axum::http::HeaderMap,
2712 Json(req): Json<ChatCompletionRequest>,
2713) -> Response {
2714 let attribution = attribution::Attribution::from_headers(&headers);
2715 state
2716 .requests_total
2717 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2718 let started = std::time::Instant::now();
2719
2720 let request_id = frink_api::next_request_id();
2724 let stream = req.stream.unwrap_or(false);
2725
2726 let refusal = cache_admin::check_admission(&state)
2732 .err()
2733 .or_else(|| req.validate_supported_fields().err());
2734 if let Some(err) = refusal {
2735 state
2736 .request_errors_total
2737 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2738 let response = err.into_response();
2739 state.record_request(stats::Record {
2740 request_id: &request_id,
2741 route: frink_api::routes::V1_CHAT_COMPLETIONS,
2742 model: state.active_model_name(),
2743 status: response.status().as_u16(),
2744 stream,
2745 duration_ms: started.elapsed().as_millis() as u64,
2746 usage: None,
2747 attribution: &attribution,
2748 });
2749 return response;
2750 }
2751
2752 let response = if stream {
2753 chat_completions_stream(
2754 Arc::clone(&state),
2755 req,
2756 request_id.clone(),
2757 started,
2758 attribution.clone(),
2759 )
2760 .await
2761 .into_response()
2762 } else {
2763 chat_completions_full(
2764 Arc::clone(&state),
2765 req,
2766 request_id.clone(),
2767 started,
2768 attribution.clone(),
2769 )
2770 .await
2771 .into_response()
2772 };
2773
2774 if response.status().is_client_error() || response.status().is_server_error() {
2775 state
2776 .request_errors_total
2777 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2778 state.record_request(stats::Record {
2782 request_id: &request_id,
2783 route: frink_api::routes::V1_CHAT_COMPLETIONS,
2784 model: state.active_model_name(),
2787 status: response.status().as_u16(),
2788 stream,
2789 duration_ms: started.elapsed().as_millis() as u64,
2790 usage: None,
2791 attribution: &attribution,
2792 });
2793 }
2794 state.mark_request_finished();
2795
2796 response
2797}
2798
2799async fn chat_completions_full(
2800 state: Arc<AppState>,
2801 req: ChatCompletionRequest,
2802 request_id: String,
2803 started: std::time::Instant,
2804 attribution: attribution::Attribution,
2805) -> Result<Json<ChatCompletionResponse>, ApiError> {
2806 let tools_active = req.tools_active();
2807 let active = state.require_active()?;
2811 let history = resolve_history(&state, &req);
2812 let template = active.generative()?.chat_template();
2813 let kwargs = req.resolve_template_kwargs(&template);
2814 let prompt = req.render_prompt(&history, &template, &req.tools, kwargs, active.name())?;
2815 let mut params =
2823 req.generation_params_for_template(&template, active.name(), active.sampler_model())?;
2824 params.lora = lora::resolve_request(active.generative()?, req.lora.as_deref())?;
2825 let key = req.is_cacheable().then(|| req.cache_key(&prompt, ¶ms));
2826
2827 let mut generated_logprobs: Vec<crate::sampling_loop::PerTokenProbs> = Vec::new();
2831 let n_logprobs = req.n_logprobs()?;
2834 let decode_piece = |id: usize| active.decode_any(&[id]);
2836 let (completion, cache_status) = if let Some(cached) = key
2837 .as_ref()
2838 .and_then(|key| lock_cache(&state.response_cache).get(key))
2839 {
2840 tracing::debug!("cache hit for key {}", key.as_ref().unwrap().digest());
2841 (cached, "hit")
2842 } else {
2843 let produced = decode_task::buffered(
2844 decode_task::DecodeHandles::take(&state, &active)?,
2845 prompt.clone(),
2846 params,
2847 )
2848 .await?;
2849 let usage = produced.usage;
2850 let choices = produced.choices;
2851
2852 generated_logprobs = choices.iter().map(|c| c.logprobs.clone()).collect();
2857 let completion = response_cache::CachedCompletion {
2858 choices: choices.into_iter().map(|c| (c.finish, c.text)).collect(),
2859 usage,
2860 };
2861 let cache_status = match key {
2867 Some(key) => match completion.clone().cacheable() {
2871 Some(cacheable) => {
2872 tracing::debug!("cache miss for key {}", key.digest());
2873 lock_cache(&state.response_cache).put(key, cacheable);
2874 "miss"
2875 }
2876 None => "skip",
2877 },
2878 None => "skip",
2879 };
2880 (completion, cache_status)
2881 };
2882 let content = completion.first_text().to_string();
2885
2886 if req.json_object_mode() {
2887 json_mode::validate_json_object_output(&content)?;
2888 }
2889
2890 if let Some(id) = &req.session_id {
2894 state.sessions.store_reply(
2895 id,
2896 ChatMessage {
2897 role: "assistant".to_string(),
2898 content: Some(MessageContent::Text(content.clone())),
2899 tool_calls: None,
2900 tool_call_id: None,
2901 reasoning_content: None,
2902 },
2903 );
2904 }
2905
2906 let posture = output::OutputPosture::resolve_full(
2911 active.reasoning_format(),
2912 active.tool_call_format(),
2913 &prompt,
2914 );
2915 let tools: &[_] = if tools_active { &req.tools } else { &[] };
2916 let wanted = req.unimplemented.n.unwrap_or(1).max(1) as usize;
2920 let ranked: Vec<(generate::FinishReason, String)> = if completion.choices.len() > wanted {
2921 let scored: Vec<crate::generate::GeneratedChoice> = completion
2922 .choices
2923 .into_iter()
2924 .zip(
2925 generated_logprobs
2926 .iter()
2927 .cloned()
2928 .chain(std::iter::repeat(Vec::new())),
2929 )
2930 .map(
2931 |((finish, text), logprobs)| crate::generate::GeneratedChoice {
2932 finish,
2933 text,
2934 logprobs,
2935 },
2936 )
2937 .collect();
2938 let best = crate::best_of::take_best(scored, wanted);
2939 generated_logprobs = best.iter().map(|c| c.logprobs.clone()).collect();
2940 best.into_iter().map(|c| (c.finish, c.text)).collect()
2941 } else {
2942 completion.choices
2943 };
2944 let rendered: Vec<ChatCompletionChoice> = ranked
2945 .into_iter()
2946 .enumerate()
2947 .map(|(index, (finish, text))| {
2948 let (message, finish_reason) =
2949 build_response_message(text, tools, posture, finish.as_str());
2950 ChatCompletionChoice {
2951 index,
2952 message,
2953 finish_reason,
2954 logprobs: n_logprobs.map(|k| {
2955 crate::logprobs::render_chat(
2956 generated_logprobs.get(index).unwrap_or(&Vec::new()),
2957 Some(k),
2958 &decode_piece,
2959 )
2960 }),
2961 }
2962 })
2963 .collect();
2964
2965 state.record_request(stats::Record {
2966 request_id: &request_id,
2967 route: frink_api::routes::V1_CHAT_COMPLETIONS,
2968 model: Some(active.name().to_string()),
2971 status: 200,
2972 stream: false,
2973 duration_ms: started.elapsed().as_millis() as u64,
2974 usage: Some(&completion.usage),
2975 attribution: &attribution,
2976 });
2977
2978 Ok(Json(ChatCompletionResponse {
2979 id: request_id.clone(),
2980 request_id,
2981 object: "chat.completion",
2982 model: req.model,
2983 choices: rendered,
2984 usage: completion.usage,
2985 frink_cache: cache_status,
2986 }))
2987}
2988
2989async fn chat_completions_stream(
2990 state: Arc<AppState>,
2991 req: ChatCompletionRequest,
2992 request_id: String,
2993 started: std::time::Instant,
2994 attribution: attribution::Attribution,
2995) -> Result<Response, ApiError> {
2996 if req.several_choices() {
3006 return Err(unsupported_feature(
3007 "`n` > 1 with `stream` is not implemented: the choices would arrive one after \
3008 another rather than interleaved by `choices[].index`. Send the request without \
3009 `stream`, which serves `n` on this route.",
3010 ));
3011 }
3012 let tools_active = req.tools_active();
3013 let active = state.require_active()?;
3017 let history = resolve_history(&state, &req);
3018 let template = active.generative()?.chat_template();
3019 let kwargs = req.resolve_template_kwargs(&template);
3020 let prompt = req.render_prompt(&history, &template, &req.tools, kwargs, active.name())?;
3021 let model_name = req.model.clone();
3022 let session_id = req.session_id.clone();
3023 let sessions = state.sessions.clone();
3024
3025 let model = Arc::clone(active.generative()?);
3026 let kv_pool = state.kv_pool.clone();
3027 let paged_kv = state.paged_kv.clone();
3028 let prefix_cache = state.prefix_cache.clone();
3029 let batcher = active.batcher.clone();
3030 let ceiling = active.ceiling.clone();
3031 let metal_private_decode_gate = state.metal_private_decode_gate.clone();
3032 let mut params =
3033 req.generation_params_for_template(&template, active.name(), active.sampler_model())?;
3034 params.lora = lora::resolve_request(active.generative()?, req.lora.as_deref())?;
3035 let stats_state = Arc::clone(&state);
3036 let served_model = active.name().to_string();
3039 let posture = output::OutputPosture::resolve_full(
3043 active.reasoning_format(),
3044 active.tool_call_format(),
3045 &prompt,
3046 );
3047 let offered_tools: Vec<ToolDef> = if tools_active {
3050 req.tools.clone()
3051 } else {
3052 Vec::new()
3053 };
3054
3055 let (cancel_token, cancel_guard) = state.cancels.register(&request_id);
3060 params.cancel = Some(cancel_token.clone());
3061
3062 let overlap = true;
3071
3072 let slot = req
3076 .stream_resumable
3077 .unwrap_or(false)
3078 .then(|| state.streams.register(&request_id));
3079 let emitter = resume::Emitter::new(slot);
3080
3081 let (tx, rx) = tokio::sync::mpsc::channel::<Result<Event, Infallible>>(64);
3082 let keepalive = sse::keepalive_event(&ChatCompletionChunk {
3086 id: request_id.clone(),
3087 request_id: None,
3088 object: "chat.completion.chunk",
3089 model: model_name.clone(),
3090 choices: vec![ChatCompletionChunkChoice {
3091 index: 0,
3092 delta: ChatCompletionChunkDelta {
3093 role: None,
3094 content: None,
3095 reasoning_content: None,
3096 tool_calls: None,
3097 },
3098 finish_reason: None,
3099 }],
3100 usage: None,
3101 });
3102
3103 tokio::task::spawn_blocking(move || {
3104 let _cancel_guard = cancel_guard;
3107 let tx_chunks = tx.clone();
3108 let orphan_timeout = sse::orphan_timeout_from_env();
3113 let mut first = true;
3114 let head_request_id = request_id.clone();
3115 let stream_reasoning: Rc<RefCell<Option<crate::policy::parser::ReasoningParser>>> =
3125 Rc::new(RefCell::new(posture.reasoning_parser()));
3126 let emit_reasoning = Rc::clone(&stream_reasoning);
3127 let stream_tools: Rc<RefCell<Option<crate::policy::parser::ToolCallParser>>> = Rc::new(
3131 RefCell::new(tools_active.then(|| posture.tool_call_parser(&offered_tools))),
3132 );
3133 let emit_tools = Rc::clone(&stream_tools);
3134 let streamed_calls = Rc::new(std::cell::Cell::new(0usize));
3138 let emit_streamed_calls = Rc::clone(&streamed_calls);
3139 let result = run_generation_emit(
3140 &model,
3141 &prompt,
3142 ¶ms,
3143 kv_pool.as_ref(),
3144 paged_kv.as_ref(),
3145 prefix_cache.as_deref(),
3146 batcher.as_ref(),
3147 ceiling.as_deref(),
3148 metal_private_decode_gate.as_deref(),
3149 |chunk| {
3150 if !overlap || chunk.is_empty() {
3151 return;
3152 }
3153 let (reasoning, content) = match emit_reasoning.borrow_mut().as_mut() {
3154 Some(parser) => {
3155 let delta = parser.push(chunk);
3156 (delta.reasoning, delta.content)
3157 }
3158 None => (String::new(), chunk.to_string()),
3159 };
3160 let (content, tool_calls) = match emit_tools.borrow_mut().as_mut() {
3164 Some(parser) => {
3165 let (text, calls) =
3166 tool_call_deltas(parser.push(&content), &emit_streamed_calls);
3167 (text, calls)
3168 }
3169 None => (content, Vec::new()),
3170 };
3171 if reasoning.is_empty() && content.is_empty() && tool_calls.is_empty() {
3174 return;
3175 }
3176 let role = if first { Some("assistant") } else { None };
3177 let request_id = first.then(|| head_request_id.clone());
3178 first = false;
3179 let payload = ChatCompletionChunk {
3180 id: head_request_id.clone(),
3181 request_id,
3182 object: "chat.completion.chunk",
3183 model: model_name.clone(),
3184 choices: vec![ChatCompletionChunkChoice {
3185 index: 0,
3186 delta: ChatCompletionChunkDelta {
3187 role,
3188 content: (!content.is_empty()).then_some(content),
3189 reasoning_content: (!reasoning.is_empty()).then_some(reasoning),
3190 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
3191 },
3192 finish_reason: None,
3193 }],
3194 usage: None,
3195 };
3196 if let Err(why) =
3205 sse::send_or_orphan(&tx_chunks, Ok(emitter.event(&payload)), orphan_timeout)
3206 {
3207 if why == sse::SendFailure::Orphaned {
3208 tracing::warn!(
3209 "SSE stream {head_request_id} accepted nothing for the orphan \
3210 deadline; treating it as abandoned"
3211 );
3212 }
3213 if !emitter.is_resumable() {
3225 cancel_token.cancel();
3226 }
3227 }
3228 },
3229 );
3230
3231 let mut pending_request_id = first.then(|| request_id.clone());
3237
3238 match result {
3239 Ok(generated) => {
3242 let usage = generated.usage;
3243 let one = generated
3244 .choices
3245 .into_iter()
3246 .next()
3247 .expect("a generation produces at least one choice");
3248 let (finish, full_text) = (one.finish, one.text);
3249 if let Some(id) = &session_id {
3250 sessions.store_reply(
3251 id,
3252 ChatMessage {
3253 role: "assistant".to_string(),
3254 content: Some(MessageContent::Text(full_text.clone())),
3255 tool_calls: None,
3256 tool_call_id: None,
3257 reasoning_content: None,
3258 },
3259 );
3260 }
3261 let mut streamed_finish: Option<&'static str> = None;
3267 if overlap {
3268 let tail = stream_reasoning
3269 .borrow_mut()
3270 .as_mut()
3271 .map(|parser| parser.flush())
3272 .unwrap_or_default();
3273 let (mut content, mut tool_calls) = (tail.content, Vec::new());
3274 if let Some(parser) = stream_tools.borrow_mut().as_mut() {
3275 let mut events = parser.push(&content);
3276 events.extend(parser.finish());
3277 let (text, calls) = tool_call_deltas(events, &streamed_calls);
3278 content = text;
3279 tool_calls = calls;
3280 }
3281 if !content.is_empty() || !tail.reasoning.is_empty() || !tool_calls.is_empty() {
3282 let payload = ChatCompletionChunk {
3283 id: request_id.clone(),
3284 request_id: pending_request_id.take(),
3285 object: "chat.completion.chunk",
3286 model: model_name.clone(),
3287 choices: vec![ChatCompletionChunkChoice {
3288 index: 0,
3289 delta: ChatCompletionChunkDelta {
3290 role: None,
3291 content: (!content.is_empty()).then_some(content),
3292 reasoning_content: (!tail.reasoning.is_empty())
3293 .then_some(tail.reasoning),
3294 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
3295 },
3296 finish_reason: None,
3297 }],
3298 usage: None,
3299 };
3300 let _ =
3301 sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3302 }
3303 if streamed_calls.get() > 0 {
3304 streamed_finish = Some("tool_calls");
3305 }
3306 } else {
3307 let parsed = output::parse_output(&full_text, &offered_tools, posture);
3310 let tool_calls: Vec<ToolCallDelta> = parsed
3311 .calls
3312 .iter()
3313 .enumerate()
3314 .map(|(index, call)| {
3315 ToolCallDelta::whole(index, call.name.clone(), call.arguments.clone())
3316 })
3317 .collect();
3318 if !tool_calls.is_empty() {
3319 streamed_finish = Some("tool_calls");
3320 }
3321 if !tool_calls.is_empty()
3322 || !parsed.content.is_empty()
3323 || parsed.reasoning.is_some()
3324 {
3325 let payload = ChatCompletionChunk {
3326 id: request_id.clone(),
3327 request_id: pending_request_id.take(),
3328 object: "chat.completion.chunk",
3329 model: model_name.clone(),
3330 choices: vec![ChatCompletionChunkChoice {
3331 index: 0,
3332 delta: ChatCompletionChunkDelta {
3333 role: Some("assistant"),
3334 content: (!parsed.content.is_empty() && tool_calls.is_empty())
3335 .then(|| parsed.content.clone()),
3336 reasoning_content: parsed.reasoning.clone(),
3337 tool_calls: (!tool_calls.is_empty()).then_some(tool_calls),
3338 },
3339 finish_reason: None,
3340 }],
3341 usage: None,
3342 };
3343 let _ =
3344 sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3345 }
3346 }
3347 let final_finish_reason = match streamed_finish {
3351 Some(reason) if finish.as_str() != "length" => reason,
3352 _ => finish.as_str(),
3353 };
3354 let final_payload = ChatCompletionChunk {
3355 id: request_id.clone(),
3356 request_id: pending_request_id.take(),
3357 object: "chat.completion.chunk",
3358 model: model_name,
3359 choices: vec![ChatCompletionChunkChoice {
3360 index: 0,
3361 delta: ChatCompletionChunkDelta {
3362 role: None,
3363 content: None,
3364 reasoning_content: None,
3365 tool_calls: None,
3366 },
3367 finish_reason: Some(final_finish_reason),
3368 }],
3369 usage: Some(usage.clone()),
3370 };
3371 let _ = sse::send_or_orphan(&tx, Ok(emitter.event(&final_payload)), orphan_timeout);
3372 let _ = sse::send_or_orphan(&tx, Ok(emitter.done()), orphan_timeout);
3373 stats_state.record_request(stats::Record {
3378 request_id: &request_id,
3379 route: frink_api::routes::V1_CHAT_COMPLETIONS,
3380 model: Some(served_model.clone()),
3381 status: 200,
3382 stream: true,
3383 duration_ms: started.elapsed().as_millis() as u64,
3384 usage: Some(&usage),
3385 attribution: &attribution,
3386 });
3387 }
3388 Err(e) => {
3389 tracing::warn!("decode error on streamed request {request_id}: {e}");
3390 stats_state.record_request(stats::Record {
3396 request_id: &request_id,
3397 route: frink_api::routes::V1_CHAT_COMPLETIONS,
3398 model: Some(served_model.clone()),
3399 status: 500,
3400 stream: true,
3401 duration_ms: started.elapsed().as_millis() as u64,
3402 usage: None,
3403 attribution: &attribution,
3404 });
3405 let payload = ChatCompletionChunk {
3406 id: request_id.clone(),
3407 request_id: pending_request_id.take(),
3408 object: "chat.completion.chunk",
3409 model: model_name,
3410 choices: vec![ChatCompletionChunkChoice {
3411 index: 0,
3412 delta: ChatCompletionChunkDelta {
3413 role: Some("assistant"),
3414 content: Some(format!("[error: {e}]")),
3415 reasoning_content: None,
3416 tool_calls: None,
3417 },
3418 finish_reason: Some("stop"),
3419 }],
3420 usage: None,
3421 };
3422 let _ = sse::send_or_orphan(&tx, Ok(emitter.event(&payload)), orphan_timeout);
3423 let _ = sse::send_or_orphan(&tx, Ok(emitter.done()), orphan_timeout);
3424 }
3425 }
3426 drop(emitter);
3430 });
3431
3432 let stream = sse::with_keepalive(rx, keepalive, sse::KEEPALIVE_INTERVAL);
3433 Ok((
3456 [(
3457 axum::http::HeaderName::from_static("x-accel-buffering"),
3458 axum::http::HeaderValue::from_static("no"),
3459 )],
3460 Sse::new(stream),
3461 )
3462 .into_response())
3463}
3464
3465fn protected_routes() -> Router<Arc<AppState>> {
3496 use frink_api::routes;
3497
3498 Router::new()
3499 .route(routes::V1_MODELS, get(list_models))
3500 .route(routes::V1_RESPONSES, post(responses::responses))
3504 .route(
3505 &axum_path(routes::V1_RESPONSE),
3506 get(responses::responses_get),
3507 )
3508 .route(
3509 &axum_path(routes::V1_RESPONSE_CANCEL),
3510 post(responses::responses_cancel),
3511 )
3512 .route(&axum_path(routes::SLOTS_ID), post(slots::post_slot))
3513 .route(routes::V1_STATS, get(serving_stats))
3514 .route(routes::V1_REQUESTS, get(recent_requests))
3515 .route(routes::V1_CACHE_STATUS, get(cache_admin::cache_status))
3516 .route(routes::V1_CACHE_REBUILD, post(cache_admin::cache_rebuild))
3517 .route(routes::ADMIN_PREPARE_STOP, post(cache_admin::prepare_stop))
3518 .route(
3519 routes::LORA_ADAPTERS,
3520 get(lora::get_lora_adapters).post(lora::post_lora_adapters),
3521 )
3522 .route(routes::V1_CHAT_COMPLETIONS, post(chat_completions))
3523 .route(routes::V1_CANCEL, post(cancel_generation))
3527 .route(&axum_path(routes::V1_STREAM), get(resume::resume))
3532 .route(&axum_path(routes::V1_STREAM_POLL), get(resume::poll))
3533 .route(routes::V1_MESSAGES, post(anthropic::messages))
3534 .route(
3535 routes::V1_MESSAGES_COUNT_TOKENS,
3536 post(anthropic::count_tokens),
3537 )
3538 .route(routes::V1_COMPLETIONS, post(openai_extra::completions))
3539 .route(routes::COMPLETION, post(completion::completion))
3544 .route(routes::COMPLETIONS, post(completion::completion))
3545 .route(routes::V1_TOKENIZE, post(openai_extra::tokenize))
3546 .route(routes::V1_DETOKENIZE, post(openai_extra::detokenize))
3547 .route(routes::TOKENIZE, post(openai_extra::tokenize))
3553 .route(routes::DETOKENIZE, post(openai_extra::detokenize))
3554 .route(routes::V1_EMBEDDINGS, post(embeddings::embeddings))
3555 .route(routes::V1_RERANK, post(rerank::rerank))
3559 .route(routes::RERANK, post(rerank::rerank))
3560 .route(routes::CACHE_STATS, get(cache_stats))
3561 .route(routes::METRICS, get(metrics))
3562 .route(routes::ADMIN_MODELS, get(admin::models))
3567 .route(routes::ADMIN_MODELS_LOAD, post(admin::load_model))
3568 .route(routes::ADMIN_MODELS_UNLOAD, post(admin::unload_model))
3569 .route(routes::SLEEP, post(admin::sleep))
3573 .route(routes::WAKE_UP, post(admin::wake_up))
3574 .route(routes::IS_SLEEPING, get(admin::is_sleeping))
3575 .route(routes::ADMIN_DOWNLOAD, post(admin::download))
3576 .route(routes::ADMIN_TASKS, get(admin::tasks))
3577 .route(&admin::cancel_route(), post(admin::cancel_task))
3578 .route(routes::ADMIN_STATS, get(admin::stats))
3579 .merge(conversations::router())
3583}
3584
3585fn axum_path(template: &str) -> String {
3586 let mut out = String::with_capacity(template.len());
3587 let mut rest = template;
3588 while let Some(open) = rest.find('{') {
3589 let Some(close) = rest[open..].find('}').map(|c| open + c) else {
3590 break;
3591 };
3592 out.push_str(&rest[..open]);
3593 out.push(':');
3594 out.push_str(&rest[open + 1..close]);
3595 rest = &rest[close + 1..];
3596 }
3597 out.push_str(rest);
3598 out
3599}
3600
3601async fn cancel_generation(
3610 State(state): State<Arc<AppState>>,
3611 Json(req): Json<frink_api::CancelGenerationRequest>,
3612) -> Response {
3613 let cancelled = state.cancels.cancel(&req.request_id);
3614 let status = if cancelled {
3615 StatusCode::OK
3616 } else {
3617 StatusCode::NOT_FOUND
3618 };
3619 let detail = if cancelled {
3620 "the generation was asked to stop; it ends at its next token".to_string()
3621 } else {
3622 "no generation with that request_id is running -- it has already \
3623 finished, was never issued, or was served by a path that does \
3624 not register for cancellation"
3625 .to_string()
3626 };
3627 (
3628 status,
3629 Json(frink_api::CancelGenerationResponse {
3630 request_id: req.request_id,
3631 cancelled,
3632 detail,
3633 }),
3634 )
3635 .into_response()
3636}
3637
3638type Activated = (
3642 Loaded,
3643 Option<serving::batch::ContinuousBatcher>,
3644 Option<Arc<budget::ContextCeiling>>,
3645);
3646
3647fn price_batcher_config(path: Option<&str>) -> serving::batch::BatcherConfig {
3665 let mut batcher = serving::batch::BatcherConfig::from_env();
3666 if batcher.max_context.is_some() && batcher.kv_blocks.is_some() {
3667 return batcher;
3670 }
3671 let Some(path) = path else {
3672 return batcher;
3673 };
3674 let priced = budget::price_gguf(path, frink_models::KvElem::F32, 1);
3680 let Some((priced, gguf_ctx, source)) = priced else {
3681 return batcher;
3682 };
3683 let Some(derived) = budget::derive_limits(&priced, gguf_ctx, batcher.kv_block_size) else {
3684 tracing::warn!(
3688 "this checkpoint's weights leave no room for KV inside the {source}: {} weight \
3689 bytes against a {} byte budget. Serving with no derived context ceiling -- set \
3690 FRINK_DEVICE_BUDGET_BYTES if the probe is wrong, or FRINK_CB_MAX_CONTEXT to \
3691 admit on a number you choose.",
3692 priced.weights_bytes,
3693 priced.device_budget_bytes,
3694 );
3695 return batcher;
3696 };
3697 tracing::info!("{source}");
3698 tracing::info!("{}", derived.fit);
3699 let adopted = budget::apply_derived(&mut batcher, &derived);
3700 if adopted.max_context {
3701 tracing::info!(
3702 "derived per-request context ceiling: {} token positions (prompt + max_tokens); \
3703 override with FRINK_CB_MAX_CONTEXT",
3704 derived.max_context
3705 );
3706 }
3707 if adopted.kv_blocks {
3708 tracing::info!(
3709 "derived KV block budget: {} blocks x {} positions; override with FRINK_CB_KV_BLOCKS",
3710 derived.kv_blocks,
3711 batcher.kv_block_size
3712 );
3713 }
3714 if let Some(narrowed) = adopted.max_context_narrowed {
3715 tracing::info!(
3716 "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",
3717 batcher.kv_blocks.unwrap_or_default(),
3718 batcher.kv_block_size
3719 );
3720 }
3721 batcher
3722}
3723
3724pub(crate) fn activate_loaded_model(
3733 loaded: model::LoadedModel,
3734 enable_continuous_batching: bool,
3735 path: Option<&str>,
3736 paged_kv: Option<&generate::PagedKvConfig>,
3737) -> Activated {
3738 match loaded {
3739 model::LoadedModel::Gguf(g) => {
3740 let decoder = Arc::new(g.decoder);
3741 let tokenizer = Arc::new(g.tokenizer);
3742 let config = price_batcher_config(path);
3743 let shape =
3749 frink_models::KvShape::from_config(&decoder.config, frink_models::KvElem::F32);
3750 let ceiling = Arc::new(budget::ContextCeiling::new(config.max_context, shape));
3751 let batcher = if enable_continuous_batching {
3752 tracing::info!(
3753 "continuous batching enabled: decode steps share Decoder::forward_multi_seq \
3754 (stop sequences use the same pending-buffer trim as the private generate loop)"
3755 );
3756 let tok = Arc::clone(&tokenizer);
3757 let decode = Arc::new(move |ids: &[usize]| tok.decode_bytes(ids));
3758 Some(serving::batch::ContinuousBatcher::spawn_with_ceiling(
3759 Arc::clone(&decoder),
3760 decode,
3761 config,
3762 Arc::clone(&ceiling),
3763 paged_kv.cloned(),
3764 ))
3765 } else {
3766 None
3767 };
3768 (
3769 Loaded::Generative(Arc::new(Model::Gguf(GgufModel {
3770 decoder,
3771 tokenizer,
3772 stop_tokens: g.stop_tokens,
3773 bos_id: g.bos_id,
3774 is_synthetic: g.is_synthetic,
3775 chat_template: g.chat_template,
3776 }))),
3777 batcher,
3778 Some(ceiling),
3779 )
3780 }
3781 model::LoadedModel::Kimi(k) => (
3782 Loaded::Generative(Arc::new(Model::Kimi(KimiModel {
3783 engine: k.engine,
3784 tokenizer: k.tokenizer,
3785 stop_tokens: k.stop_tokens,
3786 chat_template: k.chat_template,
3787 }))),
3788 None,
3789 None,
3790 ),
3791 model::LoadedModel::Mla(m) => (
3792 Loaded::Generative(Arc::new(Model::Mla(MlaModel {
3793 engine: m.engine,
3794 tokenizer: m.tokenizer,
3795 stop_tokens: m.stop_tokens,
3796 bos_id: m.bos_id,
3797 name: m.name,
3798 chat_template: m.chat_template,
3799 }))),
3800 None,
3801 None,
3802 ),
3803 model::LoadedModel::Gemma4(m) => (
3804 Loaded::Generative(Arc::new(Model::Gemma4(Gemma4Model {
3805 engine: m.engine,
3806 tokenizer: m.tokenizer,
3807 stop_tokens: m.stop_tokens,
3808 bos_id: m.bos_id,
3809 name: m.name,
3810 chat_template: m.chat_template,
3811 }))),
3812 None,
3813 None,
3814 ),
3815 model::LoadedModel::Glm52(g) => (
3816 Loaded::Generative(Arc::new(Model::Glm52(Glm52Model {
3817 engine: g.engine,
3818 tokenizer: g.tokenizer,
3819 stop_tokens: g.stop_tokens,
3820 bos_id: g.bos_id,
3821 name: g.name,
3822 chat_template: g.chat_template,
3823 }))),
3824 None,
3825 None,
3826 ),
3827 model::LoadedModel::Encoder(e) => (Loaded::Encoder(e), None, None),
3832 }
3833}
3834
3835struct StartupModels {
3842 loaded: model::LoadedModel,
3843 embedding: Option<Arc<frink_models::EmbeddingModel>>,
3844}
3845
3846fn continuous_batching_env() -> Option<bool> {
3847 match std::env::var("FRINK_CONTINUOUS_BATCHING")
3848 .ok()
3849 .map(|v| v.trim().to_ascii_lowercase())
3850 .as_deref()
3851 {
3852 None => None,
3853 Some("1" | "true" | "yes" | "on") => Some(true),
3854 Some("0" | "false" | "no" | "off") => Some(false),
3855 _ => None,
3856 }
3857}
3858
3859fn metal_private_decode_active() -> bool {
3860 #[cfg(feature = "metal")]
3861 {
3862 BUILT_WITH_METAL
3863 && frink_metal::attn::metal_attn_enabled()
3864 && std::env::var("FRINK_METAL").ok().as_deref() != Some("0")
3865 }
3866 #[cfg(not(feature = "metal"))]
3867 {
3868 false
3869 }
3870}
3871
3872fn continuous_batching_compatible(
3873 loaded: &model::LoadedModel,
3874 kv_pool: &Option<generate::KvPoolConfig>,
3875 prefix_cache: &Option<Arc<Mutex<PrefixCache>>>,
3876 paged_kv: &Option<generate::PagedKvConfig>,
3877) -> bool {
3878 matches!(loaded, model::LoadedModel::Gguf(_))
3879 && (paged_kv.is_some() || (kv_pool.is_none() && prefix_cache.is_none()))
3880}
3881
3882fn resolve_continuous_batching_enabled(
3883 loaded: &model::LoadedModel,
3884 kv_pool: &Option<generate::KvPoolConfig>,
3885 prefix_cache: &Option<Arc<Mutex<PrefixCache>>>,
3886 paged_kv: &Option<generate::PagedKvConfig>,
3887) -> bool {
3888 if !continuous_batching_compatible(loaded, kv_pool, prefix_cache, paged_kv) {
3889 return false;
3890 }
3891 match continuous_batching_env() {
3892 Some(true) => true,
3893 Some(false) => false,
3894 None => metal_private_decode_active(),
3895 }
3896}
3897
3898fn acquire_metal_private_decode_gate(
3899 gate: Option<&std::sync::Mutex<()>>,
3900 used_batcher: bool,
3901) -> Option<std::sync::MutexGuard<'_, ()>> {
3902 if used_batcher {
3903 None
3904 } else {
3905 gate.map(|g| g.lock().unwrap_or_else(|p| p.into_inner()))
3906 }
3907}
3908
3909fn build_app_state(
3910 models: StartupModels,
3911 kv_pool: Option<generate::KvPoolConfig>,
3912 paged_kv: Option<generate::PagedKvConfig>,
3913 prefix_cache: Option<Arc<Mutex<PrefixCache>>>,
3914 enable_continuous_batching: bool,
3915 mcp: Option<mcp::LoadedMcpConfig>,
3916 detection: Arc<health::Detection>,
3917) -> AppState {
3918 let StartupModels { loaded, embedding } = models;
3919 let configured_path = std::env::var("FRINK_MODEL_PATH").ok();
3920 let (loaded, batcher, ceiling) = activate_loaded_model(
3921 loaded,
3922 enable_continuous_batching,
3923 configured_path.as_deref(),
3924 paged_kv.as_ref(),
3925 );
3926 let id = startup_model_id();
3932 let metal_private_decode_gate = if enable_continuous_batching || !metal_private_decode_active()
3933 {
3934 None
3935 } else {
3936 tracing::info!(
3937 "Metal private-loop decode will serialize concurrent requests until \
3938 continuous batching is enabled (FRINK_CONTINUOUS_BATCHING=1 or --cont-batching)"
3939 );
3940 Some(Arc::new(std::sync::Mutex::new(())))
3941 };
3942 AppState {
3943 slept: Mutex::new(None),
3944 embedding,
3945 active: std::sync::RwLock::new(Some(Arc::new(ActiveModel {
3946 id,
3947 loaded,
3948 batcher,
3949 ceiling,
3950 checkpoint_path: configured_path.as_deref().map(PathBuf::from),
3951 }))),
3952 paged_kv,
3953 load_in_progress: std::sync::atomic::AtomicBool::new(false),
3954 tasks: Arc::new(tasks::TaskRegistry::new()),
3955 cancels: Arc::new(cancel::CancelRegistry::new()),
3956 stats: stats::Stats::new(),
3957 streams: resume::StreamRegistry::new(),
3958 model_dir: admin::model_dirs().into_iter().next(),
3959 response_cache: Mutex::new(ResponseCache::new(1000, Duration::from_secs(3600))),
3960 kv_pool,
3961 prefix_cache,
3962 sessions: session::SessionStore::new(),
3963 requests_total: std::sync::atomic::AtomicU64::new(0),
3964 request_errors_total: std::sync::atomic::AtomicU64::new(0),
3965 started_at: std::time::Instant::now(),
3966 last_request_ms: std::sync::atomic::AtomicU64::new(0),
3967 detection,
3968 mcp,
3969 continuous_batching_enabled: enable_continuous_batching,
3970 metal_private_decode_gate,
3971 loading_model: Mutex::new(None),
3972 last_load_error: Mutex::new(None),
3973 serving: Mutex::new(crate::stats::ServingStats::default()),
3974 maintenance: Mutex::new(crate::policy::maintenance::MaintenanceGate::serving()),
3975 footprint: Mutex::new(crate::policy::footprint::ProbeCache::new(FOOTPRINT_TTL_MS)),
3976 started_unix: unix_now(),
3977 }
3978}
3979
3980fn load_embedding_model() -> anyhow::Result<Option<Arc<frink_models::EmbeddingModel>>> {
3988 let Ok(path) = std::env::var("FRINK_EMBEDDING_MODEL_PATH") else {
3989 return Ok(None);
3990 };
3991 let model = frink_models::EmbeddingModel::from_gguf_path(&path)
3992 .map_err(|e| anyhow::anyhow!("FRINK_EMBEDDING_MODEL_PATH={path}: {e}"))?;
3993 tracing::info!(
3994 "loaded embedding model '{}' ({}, {} dims, pooling {}, max {} tokens)",
3995 model.name(),
3996 model.architecture(),
3997 model.n_embd(),
3998 model.pooling_type().name(),
3999 model.n_ctx_train(),
4000 );
4001 Ok(Some(Arc::new(model)))
4002}
4003
4004fn unix_now() -> u64 {
4009 std::time::SystemTime::now()
4010 .duration_since(std::time::UNIX_EPOCH)
4011 .map(|d| d.as_secs())
4012 .unwrap_or(0)
4013}
4014
4015fn startup_model_id() -> Option<String> {
4020 let configured = std::env::var("FRINK_MODEL_PATH").ok()?;
4021 let configured = std::fs::canonicalize(&configured).ok()?;
4022 admin::discover(&admin::model_dirs())
4023 .into_iter()
4024 .find(|d| {
4025 std::fs::canonicalize(&d.path)
4026 .map(|p| p == configured)
4027 .unwrap_or(false)
4028 })
4029 .map(|d| d.id)
4030}
4031
4032fn init_cpu_pool() {
4041 match frink_core::threads::init_cpu_pool() {
4042 Some(n) => eprintln!(
4043 "frink-server: rayon pool {n} threads (perf cores {}; override with FRINK_CPU_THREADS)",
4044 frink_core::threads::perf_core_count()
4045 ),
4046 None => eprintln!("frink-server: global rayon pool already built; leaving it alone"),
4047 }
4048}
4049
4050fn announce_ready(addr: SocketAddr, scheme: &str) {
4063 use std::io::Write;
4064 let ready =
4065 frink_api::ServerReady::new(addr, scheme, env!("CARGO_PKG_VERSION"), std::process::id());
4066 let mut stdout = std::io::stdout().lock();
4067 let _ = writeln!(stdout, "{}", ready.to_line());
4068 let _ = stdout.flush();
4069}
4070
4071async fn shutdown_signal(exit_on_stdin_close: bool) {
4083 if !exit_on_stdin_close {
4084 std::future::pending::<()>().await;
4085 return;
4086 }
4087 let _ = tokio::task::spawn_blocking(|| {
4088 use std::io::Read;
4089 let mut sink = [0u8; 256];
4090 let mut stdin = std::io::stdin().lock();
4091 loop {
4092 match stdin.read(&mut sink) {
4093 Ok(0) => break,
4095 Ok(_) => continue,
4097 Err(e) => {
4098 tracing::warn!("stdin read failed ({e}); treating it as closed");
4099 break;
4100 }
4101 }
4102 }
4103 })
4104 .await;
4105 tracing::info!("stdin closed; shutting down");
4106}
4107
4108fn tokio_worker_threads() -> usize {
4113 std::env::var("FRINK_TOKIO_WORKERS")
4114 .ok()
4115 .and_then(|v| v.trim().parse::<usize>().ok())
4116 .filter(|n| *n > 0)
4117 .unwrap_or(2)
4118}
4119
4120fn install_ring_crypto_provider() {
4137 let _ = rustls::crypto::ring::default_provider().install_default();
4138}
4139
4140pub fn run_server(args: ServerArgs) -> anyhow::Result<()> {
4146 if args.list_devices {
4147 frink_models::devices::print_available_devices();
4148 return Ok(());
4149 }
4150 apply_cli_overrides(&args)?;
4151
4152 let _instance = {
4156 use frink_core::instance::{register, InstancePolicy};
4157 let policy = if args.allow_multiple_instances {
4158 InstancePolicy::Multi
4159 } else {
4160 InstancePolicy::from_env_or(InstancePolicy::Single)
4161 };
4162 let model = std::env::var("FRINK_MODEL_PATH").ok();
4163 register(
4164 "server",
4165 model.as_deref(),
4166 frink_core::instance::current_backend(),
4167 policy,
4168 )
4169 .map_err(|conflict| anyhow::anyhow!("{conflict}"))?
4170 };
4171
4172 let journal = journal::Journal::from_env();
4173 eprintln!(
4174 "frink-server: process lifecycle journal at {:?} (override with FRINK_JOURNAL_PATH)",
4175 journal.path()
4176 );
4177 journal.append(&journal::Record::session_start(
4178 env!("CARGO_PKG_VERSION"),
4179 std::process::id(),
4180 ));
4181 journal::install_panic_hook(journal.clone());
4182
4183 let mcp_config_path = args.mcp_config.clone();
4184 let exit_on_stdin_close = args.exit_on_stdin_close
4185 || std::env::var("FRINK_EXIT_ON_STDIN_CLOSE")
4186 .map(|v| v == "1")
4187 .unwrap_or(false);
4188
4189 unsafe { frink_core::weight_matrix::default_cpu_int_dot_on() };
4193 init_cpu_pool();
4194
4195 let runtime = tokio::runtime::Builder::new_multi_thread()
4196 .worker_threads(tokio_worker_threads())
4197 .enable_all()
4198 .build()?;
4199 let result = runtime.block_on(run(mcp_config_path, exit_on_stdin_close));
4200
4201 let reason = match &result {
4202 Ok(()) => "normal".to_string(),
4203 Err(e) => e.to_string(),
4204 };
4205 journal.append(&journal::Record::session_exit(reason));
4206
4207 runtime.shutdown_background();
4212
4213 result
4214}
4215
4216async fn run(mcp_config_path: Option<PathBuf>, exit_on_stdin_close: bool) -> anyhow::Result<()> {
4217 let _ = tracing_subscriber::fmt::try_init();
4223
4224 let addr = std::env::var("FRINK_ADDR").unwrap_or_else(|_| "127.0.0.1:8383".to_string());
4234 let api_key_configured = std::env::var("FRINK_API_KEY").is_ok();
4235 let allow_unauthenticated_remote = std::env::var("FRINK_ALLOW_UNAUTHENTICATED_REMOTE")
4236 .map(|v| v == "1")
4237 .unwrap_or(false);
4238 if let Err(msg) =
4239 security::check_bind_authorization(&addr, api_key_configured, allow_unauthenticated_remote)
4240 {
4241 anyhow::bail!(msg);
4242 }
4243
4244 let embedding_model = load_embedding_model()?;
4251
4252 let mut loaded = model::load()?;
4253 match &loaded {
4254 model::LoadedModel::Gguf(g) => tracing::info!(
4255 "loaded GGUF model '{}' (synthetic={}, tokenizer={})",
4256 g.decoder.config.name,
4257 g.is_synthetic,
4258 g.tokenizer.kind()
4259 ),
4260 model::LoadedModel::Kimi(k) => tracing::info!(
4261 "loaded Kimi K3 checkpoint (tokenizer={} base tokens)",
4262 k.tokenizer.vocab_size()
4263 ),
4264 model::LoadedModel::Mla(m) => tracing::info!(
4265 "loaded MLA GGUF '{}' (tokenizer={})",
4266 m.name,
4267 m.tokenizer.kind()
4268 ),
4269 model::LoadedModel::Gemma4(m) => tracing::info!(
4270 "loaded Gemma4 GGUF '{}' (tokenizer={})",
4271 m.name,
4272 m.tokenizer.kind()
4273 ),
4274 model::LoadedModel::Glm52(g) => tracing::info!(
4275 "loaded GLM-5.2 GGUF '{}' (tokenizer={})",
4276 g.name,
4277 g.tokenizer.kind()
4278 ),
4279 model::LoadedModel::Encoder(_) => {}
4282 }
4283 let metal_default_moe_budget = {
4291 #[cfg(feature = "metal")]
4292 {
4293 frink_core::metal_dense_enabled()
4294 && std::env::var("FRINK_GPU_VRAM_BUDGET_BYTES").is_err()
4295 }
4296 #[cfg(not(feature = "metal"))]
4297 {
4298 false
4299 }
4300 };
4301 if let Ok(budget_str) = std::env::var("FRINK_GPU_VRAM_BUDGET_BYTES") {
4302 let budget: u64 = budget_str
4303 .parse()
4304 .expect("FRINK_GPU_VRAM_BUDGET_BYTES must be a non-negative integer");
4305 match &mut loaded {
4306 model::LoadedModel::Gguf(g) => {
4307 tracing::info!(
4308 "GPU expert placement enabled: {budget} byte VRAM budget for routed experts \
4309 (CUDA and/or Metal matvecs when built with the matching feature)"
4310 );
4311 g.decoder.gpu_vram_budget_bytes = Some(budget);
4312 }
4313 model::LoadedModel::Kimi(_) => {
4314 tracing::warn!(
4315 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is Kimi K3 -- not \
4316 supported yet (its MoE stack isn't wired to PlacementPlan), ignoring"
4317 );
4318 }
4319 model::LoadedModel::Mla(_) => {
4320 tracing::warn!(
4321 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is MLA -- dense \
4322 FFN path only today; ignoring expert VRAM budget"
4323 );
4324 }
4325 model::LoadedModel::Gemma4(_) => {
4326 tracing::warn!(
4327 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is Gemma4 -- \
4328 ignoring expert VRAM budget"
4329 );
4330 }
4331 model::LoadedModel::Glm52(_) => {
4332 tracing::warn!(
4333 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is GLM-5.2 DSA -- \
4334 GPU expert placement not wired yet; ignoring"
4335 );
4336 }
4337 model::LoadedModel::Encoder(_) => {
4338 tracing::warn!(
4339 "FRINK_GPU_VRAM_BUDGET_BYTES is set but the loaded model is an encoder -- \
4340 it has no routed experts to place; ignoring"
4341 );
4342 }
4343 }
4344 } else if metal_default_moe_budget {
4345 const METAL_DEFAULT_MOE_BUDGET: u64 = 64 * 1024 * 1024 * 1024;
4349 if let model::LoadedModel::Gguf(g) = &mut loaded {
4350 tracing::info!(
4351 "Metal MoE expert placement default-on ({METAL_DEFAULT_MOE_BUDGET} byte budget); \
4352 set FRINK_GPU_VRAM_BUDGET_BYTES=0 to force CPU experts"
4353 );
4354 g.decoder.gpu_vram_budget_bytes = Some(METAL_DEFAULT_MOE_BUDGET);
4355 }
4356 }
4357 #[cfg(feature = "cuda")]
4358 {
4359 if frink_core::cuda_dense_enabled() {
4360 tracing::info!(
4361 "CUDA dense matvec enabled for WeightMatrix::apply \
4362 (FRINK_CUDA=0|cpu forces CPU; weight buffers stay resident after first upload)"
4363 );
4364 } else {
4365 tracing::info!(
4366 "CUDA dense matvec disabled (FRINK_CUDA); dense decode uses CPU or Metal"
4367 );
4368 }
4369 }
4370 #[cfg(feature = "metal")]
4371 {
4372 if frink_core::metal_dense_enabled() {
4373 tracing::info!(
4374 "Metal dense matvec enabled for WeightMatrix::apply \
4375 (FRINK_METAL=0|cpu forces CPU; weight buffers stay resident after first upload)"
4376 );
4377 match std::env::var("FRINK_METAL_ATTN").ok().as_deref() {
4378 Some("1") | Some("true") | Some("on") | Some("attn") => {
4379 tracing::info!(
4380 "Metal fused attention requested (FRINK_METAL_ATTN): \
4381 QKV→RoPE→GQA→O on-GPU for Norm/NeoX decode without QKV bias/QK-norm"
4382 );
4383 }
4384 _ => {}
4385 }
4386 tracing::info!(
4387 "Metal greedy GPU argmax: temperature<=0 folds \
4388 final_norm+lm_head+argmax into the dense stack"
4389 );
4390 } else {
4391 tracing::info!("Metal dense matvec disabled (FRINK_METAL); dense decode uses CPU");
4392 }
4393 }
4394 let blocks_env = std::env::var("FRINK_KV_POOL_BLOCKS");
4405 let block_size_env = std::env::var("FRINK_KV_POOL_BLOCK_SIZE");
4406 let byte_budget_env = std::env::var("FRINK_KV_BYTE_BUDGET");
4407 if blocks_env.is_ok() && byte_budget_env.is_ok() {
4408 panic!(
4409 "FRINK_KV_POOL_BLOCKS and FRINK_KV_BYTE_BUDGET are mutually exclusive \
4410 (set one block-count source plus FRINK_KV_POOL_BLOCK_SIZE, or neither to disable)"
4411 );
4412 }
4413 let kv_pool = match (blocks_env, block_size_env, byte_budget_env) {
4414 (Ok(blocks), Ok(block_size), Err(_)) => {
4415 let total_blocks: usize = blocks
4416 .parse()
4417 .expect("FRINK_KV_POOL_BLOCKS must be a positive integer");
4418 let block_size: usize = block_size
4419 .parse()
4420 .expect("FRINK_KV_POOL_BLOCK_SIZE must be a positive integer");
4421 let queue_wait_ms: u64 = std::env::var("FRINK_KV_POOL_QUEUE_TIMEOUT_MS")
4427 .ok()
4428 .map(|v| {
4429 v.parse()
4430 .expect("FRINK_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4431 })
4432 .unwrap_or(0);
4433 tracing::info!(
4434 "KV cache block pool enabled: {total_blocks} blocks x {block_size} positions \
4435 each, shared across all concurrent requests, {queue_wait_ms}ms admission queue wait"
4436 );
4437 Some(generate::KvPoolConfig {
4438 pool: Arc::new(Mutex::new(KvBlockPool::new(block_size, total_blocks))),
4439 queue_wait: Duration::from_millis(queue_wait_ms),
4440 })
4441 }
4442 (Err(_), Ok(block_size), Ok(byte_budget)) => {
4443 let block_size: usize = block_size
4444 .parse()
4445 .expect("FRINK_KV_POOL_BLOCK_SIZE must be a positive integer");
4446 let budget: u64 = byte_budget
4447 .parse()
4448 .expect("FRINK_KV_BYTE_BUDGET must be a positive integer");
4449 let cfg = match &loaded {
4450 model::LoadedModel::Gguf(g) => &g.decoder.config,
4451 model::LoadedModel::Kimi(_)
4452 | model::LoadedModel::Mla(_)
4453 | model::LoadedModel::Gemma4(_)
4454 | model::LoadedModel::Glm52(_)
4455 | model::LoadedModel::Encoder(_) => {
4456 panic!(
4457 "FRINK_KV_BYTE_BUDGET requires a GGUF decoder model \
4458 (set FRINK_MODEL_PATH to a generic-decoder .gguf file)"
4459 );
4460 }
4461 };
4462 let bytes_per_block = block_size
4463 * cfg.kv_heads_all_layers()
4464 * (cfg.head_dim + cfg.v_head_dim())
4465 * std::mem::size_of::<f32>();
4466 assert!(
4467 bytes_per_block > 0,
4468 "derived KV block byte size must be positive (check model config and block size)"
4469 );
4470 let total_blocks = (budget as usize / bytes_per_block).max(1);
4471 let queue_wait_ms: u64 = std::env::var("FRINK_KV_POOL_QUEUE_TIMEOUT_MS")
4472 .ok()
4473 .map(|v| {
4474 v.parse()
4475 .expect("FRINK_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4476 })
4477 .unwrap_or(0);
4478 tracing::info!(
4479 "KV cache block pool enabled from byte budget: {budget} bytes / \
4480 {bytes_per_block} bytes per block ({block_size} positions x {} layers) -> \
4481 {total_blocks} blocks, {queue_wait_ms}ms admission queue wait",
4482 cfg.n_layers
4483 );
4484 Some(generate::KvPoolConfig {
4485 pool: Arc::new(Mutex::new(KvBlockPool::new(block_size, total_blocks))),
4486 queue_wait: Duration::from_millis(queue_wait_ms),
4487 })
4488 }
4489 (Err(_), Err(_), Err(_)) => None,
4490 (Err(_), Ok(_), Err(_)) => panic!(
4491 "FRINK_KV_POOL_BLOCK_SIZE requires FRINK_KV_POOL_BLOCKS or FRINK_KV_BYTE_BUDGET \
4492 (or unset all three to disable KV cache pooling)"
4493 ),
4494 (Ok(_), Ok(_), Ok(_)) => {
4495 unreachable!("FRINK_KV_POOL_BLOCKS and FRINK_KV_BYTE_BUDGET are mutually exclusive")
4496 }
4497 (Ok(_), Err(_), _) | (Err(_), Err(_), Ok(_)) => panic!(
4498 "FRINK_KV_POOL_BLOCKS/FRINK_KV_BYTE_BUDGET and FRINK_KV_POOL_BLOCK_SIZE must be \
4499 set together (or neither, to disable KV cache pooling)"
4500 ),
4501 };
4502 let paged_kv = match (
4508 std::env::var("FRINK_PAGED_KV_BLOCKS"),
4509 std::env::var("FRINK_PAGED_KV_BLOCK_SIZE"),
4510 ) {
4511 (Ok(blocks), Ok(block_size)) => {
4512 assert!(
4513 kv_pool.is_none(),
4514 "FRINK_PAGED_KV_BLOCKS and FRINK_KV_POOL_BLOCKS/FRINK_KV_BYTE_BUDGET are \
4515 mutually exclusive: both bound the same KV memory, by different means. \
4516 Set one."
4517 );
4518 let blocks_per_layer: usize = blocks
4532 .parse()
4533 .expect("FRINK_PAGED_KV_BLOCKS must be a positive integer");
4534 let block_size: usize = block_size
4535 .parse()
4536 .expect("FRINK_PAGED_KV_BLOCK_SIZE must be a positive integer");
4537 let gguf = match &loaded {
4538 model::LoadedModel::Gguf(g) => g,
4539 _ => panic!(
4540 "FRINK_PAGED_KV_BLOCKS requires a GGUF decoder model \
4541 (set FRINK_MODEL_PATH to a generic-decoder .gguf file)"
4542 ),
4543 };
4544 let cfg = &gguf.decoder.config;
4545 let queue_wait_ms: u64 = std::env::var("FRINK_KV_POOL_QUEUE_TIMEOUT_MS")
4546 .ok()
4547 .map(|v| {
4548 v.parse()
4549 .expect("FRINK_KV_POOL_QUEUE_TIMEOUT_MS must be a non-negative integer")
4550 })
4551 .unwrap_or(0);
4552 tracing::info!(
4553 "Paged KV enabled: {blocks_per_layer} blocks x {block_size} positions per \
4554 layer across {} layers, shared by all concurrent requests, \
4555 {queue_wait_ms}ms admission queue wait",
4556 cfg.n_layers
4557 );
4558 let radix = Some(Arc::new(Mutex::new(crate::policy::radix::RadixCache::new(
4563 block_size,
4564 ))));
4565 let anchor_token = crate::policy::anchor::resolve_anchor_token(
4574 crate::policy::parser::ToolCallFormat::infer(
4575 &std::env::var("FRINK_MODEL_PATH").unwrap_or_default(),
4576 )
4577 .opener(),
4578 |text| {
4579 gguf.tokenizer
4580 .encode(text, SpecialTokens::Parse)
4581 .into_iter()
4582 .map(|t| t as u32)
4583 .collect()
4584 },
4585 );
4586 if let Some(id) = anchor_token {
4587 tracing::info!(
4588 "Paged KV window slide: tool-call anchor is token {id}, so a turn's \
4589 window stops short of where its next turn rejoins"
4590 );
4591 }
4592 let slide_interval: usize = std::env::var("FRINK_PAGED_KV_SLIDE_INTERVAL")
4593 .ok()
4594 .map(|v| {
4595 v.parse()
4596 .expect("FRINK_PAGED_KV_SLIDE_INTERVAL must be a positive integer")
4597 })
4598 .unwrap_or(crate::policy::pool_budget::DEFAULT_SWA_EVICTION_INTERVAL);
4599 if let Some(window) = cfg.uniform_sliding_window() {
4600 tracing::info!(
4601 "Paged KV window slide enabled: every layer slides by {window} every \
4602 {slide_interval} decode steps, so a request holds its prompt and a \
4603 window rather than its whole context"
4604 );
4605 } else if cfg.kv_block_window().is_some() {
4606 tracing::info!(
4607 "Paged KV window slide NOT enabled: this model has full-attention layers, \
4608 and a page group holds one block in every layer"
4609 );
4610 }
4611 Some(generate::PagedKvConfig {
4612 store: Arc::new(cfg.new_paged_kv(block_size, blocks_per_layer)),
4615 queue_wait: Duration::from_millis(queue_wait_ms),
4616 radix,
4617 anchor_token,
4618 slide_interval,
4619 })
4620 }
4621 (Err(_), Err(_)) => None,
4622 _ => panic!(
4623 "FRINK_PAGED_KV_BLOCKS and FRINK_PAGED_KV_BLOCK_SIZE must be set together \
4624 (or neither, to disable paged KV)"
4625 ),
4626 };
4627 let prefix_cache = std::env::var("FRINK_PREFIX_CACHE_ENTRIES").ok().map(|v| {
4634 let max_entries: usize = v
4635 .parse()
4636 .expect("FRINK_PREFIX_CACHE_ENTRIES must be a positive integer");
4637 if kv_pool.is_some() {
4638 tracing::warn!(
4639 "FRINK_PREFIX_CACHE_ENTRIES is set but so is the KV pool -- prefix \
4640 caching will never be consulted while a KV pool is configured"
4641 );
4642 }
4643 assert!(
4650 paged_kv.is_none(),
4651 "FRINK_PREFIX_CACHE_ENTRIES and FRINK_PAGED_KV_BLOCKS are mutually exclusive: \
4652 the prefix cache stores contiguous KV snapshots, which a paged request does not \
4653 produce, so the cache could never hit. Set one."
4654 );
4655 tracing::info!(
4656 "KV-prefix cache enabled: up to {max_entries} stored prefixes, shared across \
4657 all requests"
4658 );
4659 Arc::new(Mutex::new(PrefixCache::new(max_entries)))
4660 });
4661 if matches!(
4662 loaded,
4663 model::LoadedModel::Kimi(_) | model::LoadedModel::Mla(_) | model::LoadedModel::Glm52(_)
4664 ) && (kv_pool.is_some() || prefix_cache.is_some())
4665 {
4666 tracing::warn!(
4667 "KV pool / prefix cache are configured but the loaded model is Kimi, MLA, or GLM-5.2 -- \
4668 neither is consulted for those engines (state shapes differ from Decoder KV); see \
4669 frink_models::engine's module docs"
4670 );
4671 }
4672 let enable_cb =
4673 resolve_continuous_batching_enabled(&loaded, &kv_pool, &prefix_cache, &paged_kv);
4674 if enable_cb && continuous_batching_env().is_none() && metal_private_decode_active() {
4675 tracing::info!(
4676 "continuous batching enabled by default on Metal for safe parallel serving \
4677 (set FRINK_CONTINUOUS_BATCHING=0 or --no-cont-batching to use the private path)"
4678 );
4679 }
4680 if continuous_batching_env() == Some(true)
4681 && !continuous_batching_compatible(&loaded, &kv_pool, &prefix_cache, &paged_kv)
4682 && (kv_pool.is_some() || prefix_cache.is_some())
4683 {
4684 tracing::warn!(
4685 "FRINK_CONTINUOUS_BATCHING=1 ignored while KV pool or prefix cache is configured \
4686 (those modes keep the private generate path)"
4687 );
4688 }
4689 if let Ok(n) = std::env::var("FRINK_CHUNKED_PREFILL") {
4690 if let Ok(chunk) = n.parse::<usize>() {
4691 if chunk > 0 {
4692 tracing::info!("chunked prefill enabled: {chunk} tokens per forward_batch chunk");
4693 }
4694 }
4695 }
4696 if matches!(
4697 std::env::var("FRINK_CPU_KV_OFFLOAD").ok().as_deref(),
4698 Some("1")
4699 ) {
4700 tracing::warn!(
4701 "FRINK_CPU_KV_OFFLOAD=1: syncing Metal KV to host after each decode step \
4702 (minimal spill; full layer offload still planned)"
4703 );
4704 }
4705
4706 let mcp = match mcp_config_path {
4707 Some(path) => {
4708 let loaded = mcp::load_mcp_config(&path)?;
4709 tracing::info!(
4710 "MCP config loaded from {} ({} server(s); invocation not wired yet)",
4711 loaded.path,
4712 loaded.servers.len()
4713 );
4714 Some(loaded)
4715 }
4716 None => None,
4717 };
4718
4719 let detection = health::Detection::spawn();
4723
4724 let state = Arc::new(build_app_state(
4725 StartupModels {
4726 loaded,
4727 embedding: embedding_model,
4728 },
4729 kv_pool,
4730 paged_kv,
4731 prefix_cache,
4732 enable_cb,
4733 mcp,
4734 detection,
4735 ));
4736
4737 use frink_api::routes;
4741
4742 let public = Router::new().route(routes::HEALTH, get(health));
4747
4748 let mut protected = protected_routes();
4749
4750 if let Ok(key) = std::env::var("FRINK_API_KEY") {
4755 tracing::info!("API key auth enabled");
4756 let auth = limits::AuthConfig {
4757 api_key: Arc::new(key),
4758 };
4759 protected = protected.route_layer(axum::middleware::from_fn_with_state(
4760 auth,
4761 limits::require_api_key,
4762 ));
4763 }
4764 if let Ok(rpm) = std::env::var("FRINK_RATE_LIMIT_PER_MINUTE") {
4765 let rpm: u32 = rpm
4766 .parse()
4767 .expect("FRINK_RATE_LIMIT_PER_MINUTE must be a positive integer");
4768 tracing::info!("rate limiting enabled: {rpm} requests/minute (global)");
4769 let limiter = Arc::new(limits::RateLimiter::per_minute(rpm));
4770 protected = protected.route_layer(axum::middleware::from_fn_with_state(
4771 limiter,
4772 limits::rate_limit,
4773 ));
4774 }
4775 if let Ok(spec) = std::env::var("FRINK_CORS_ORIGINS") {
4783 let origins = security::parse_cors_origins(&spec)
4784 .unwrap_or_else(|e| panic!("FRINK_CORS_ORIGINS: {e}"));
4785 tracing::info!(
4786 "CORS enabled: {} allow-listed origin(s) ({})",
4787 origins.len(),
4788 spec
4789 );
4790 let cors = tower_http::cors::CorsLayer::new()
4791 .allow_origin(tower_http::cors::AllowOrigin::list(origins))
4792 .allow_methods([axum::http::Method::GET, axum::http::Method::POST])
4793 .allow_headers([
4794 axum::http::header::CONTENT_TYPE,
4795 axum::http::header::AUTHORIZATION,
4796 axum::http::HeaderName::from_static(attribution::CLIENT_HEADER),
4802 axum::http::HeaderName::from_static("last-event-id"),
4806 ]);
4807 protected = protected.route_layer(cors);
4808 }
4809
4810 let app = public
4814 .merge(protected)
4815 .layer(axum::middleware::from_fn(limits::retry_after))
4816 .with_state(state);
4817
4818 let tls_paths = security::tls_paths_from_env().unwrap_or_else(|e| panic!("{e}"));
4824 install_ring_crypto_provider();
4825 match tls_paths {
4830 Some(paths) => {
4831 let config =
4832 axum_server::tls_rustls::RustlsConfig::from_pem_file(&paths.cert, &paths.key)
4833 .await
4834 .map_err(|e| {
4835 anyhow::anyhow!(
4836 "failed to load TLS cert/key ({:?}, {:?}): {e}",
4837 paths.cert,
4838 paths.key
4839 )
4840 })?;
4841 let socket_addr: std::net::SocketAddr = addr
4842 .parse()
4843 .map_err(|e| anyhow::anyhow!("invalid FRINK_ADDR {addr:?} for TLS: {e}"))?;
4844 let listener = std::net::TcpListener::bind(socket_addr)?;
4845 listener.set_nonblocking(true)?;
4853 let bound = listener.local_addr()?;
4854 tracing::info!("TLS enabled: frink-server listening on https://{bound}");
4855 announce_ready(bound, "https");
4856
4857 let handle = axum_server::Handle::new();
4858 let shutdown_handle = handle.clone();
4859 tokio::spawn(async move {
4860 shutdown_signal(exit_on_stdin_close).await;
4861 shutdown_handle.graceful_shutdown(Some(Duration::from_secs(5)));
4862 });
4863 axum_server::from_tcp_rustls(listener, config)?
4864 .handle(handle)
4865 .serve(app.into_make_service())
4866 .await?;
4867 }
4868 None => {
4869 let listener = tokio::net::TcpListener::bind(&addr).await?;
4870 let bound = listener.local_addr()?;
4871 tracing::info!("frink-server listening on {bound}");
4872 announce_ready(bound, "http");
4873 axum::serve(listener, app)
4874 .with_graceful_shutdown(shutdown_signal(exit_on_stdin_close))
4875 .await?;
4876 }
4877 }
4878 Ok(())
4879}
4880
4881#[cfg(test)]
4882pub(crate) mod tests {
4883 use super::*;
4884 use frink_models::config::test_dense_fixture;
4885
4886 #[test]
4887 fn the_ready_line_round_trips_through_a_parent_reading_stdout() {
4888 let addr: SocketAddr = "127.0.0.1:51999".parse().unwrap();
4889 let ready = frink_api::ServerReady::new(addr, "http", "0.5.0", std::process::id());
4890 let parsed = frink_api::ServerReady::from_line(&ready.to_line()).unwrap();
4891 assert_eq!(parsed.port, 51999);
4892 assert_eq!(parsed.base_url(), "http://127.0.0.1:51999");
4893 assert!(frink_api::ServerReady::from_line("INFO frink-server listening").is_none());
4895 }
4896
4897 fn test_model() -> Model {
4898 let cfg = test_dense_fixture();
4902 Model::Gguf(GgufModel {
4903 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 32)),
4904 tokenizer: Arc::new(ServerTokenizer::Byte),
4905 stop_tokens: StopTokens::default(),
4906 bos_id: None,
4907 is_synthetic: true,
4908 chat_template: chat_template::PromptTemplate::plain(),
4909 })
4910 }
4911
4912 fn greedy_params(max_tokens: usize) -> GenerationParams {
4913 GenerationParams {
4914 cache_salt: None,
4915 prompt_logprobs: None,
4916 wants_logprobs: false,
4917 n: 1,
4918 reasoning: None,
4919 max_tokens,
4920 sampling: SamplingParams::default(),
4921 seed: 1,
4922 stop: Vec::new(),
4923 stop_token_ids: Vec::new(),
4924 json_object: false,
4925 grammar: None,
4926 cancel: None,
4927 ignore_eos: false,
4928 reasoning_budget: crate::reasoning_budget::ReasoningBudget::Unrestricted,
4929 lora: None,
4930 }
4931 }
4932
4933 fn test_model_full_byte_vocab() -> Model {
4937 test_model_full_byte_vocab_with_eos(None)
4938 }
4939
4940 fn test_model_full_byte_vocab_with_eos(eos: Option<usize>) -> Model {
4947 let mut cfg = test_dense_fixture();
4948 cfg.vocab_size = 256;
4949 Model::Gguf(GgufModel {
4950 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
4951 tokenizer: Arc::new(ServerTokenizer::Byte),
4952 stop_tokens: StopTokens::from_eos(eos),
4953 bos_id: None,
4954 is_synthetic: true,
4955 chat_template: chat_template::PromptTemplate::plain(),
4956 })
4957 }
4958
4959 pub(crate) fn test_state(model: Model, response_cache: ResponseCache) -> AppState {
4963 test_state_at(model, response_cache, None)
4964 }
4965
4966 pub(crate) fn test_state_at(
4970 model: Model,
4971 response_cache: ResponseCache,
4972 checkpoint_path: Option<std::path::PathBuf>,
4973 ) -> AppState {
4974 AppState {
4975 slept: Mutex::new(None),
4976 embedding: None,
4977 paged_kv: None,
4978 active: std::sync::RwLock::new(Some(Arc::new(ActiveModel {
4979 id: None,
4980 loaded: Loaded::Generative(Arc::new(model)),
4981 batcher: None,
4982 ceiling: None,
4983 checkpoint_path,
4984 }))),
4985 load_in_progress: std::sync::atomic::AtomicBool::new(false),
4986 tasks: Arc::new(tasks::TaskRegistry::new()),
4987 cancels: Arc::new(cancel::CancelRegistry::new()),
4988 stats: stats::Stats::new(),
4989 streams: resume::StreamRegistry::new(),
4990 model_dir: None,
4991 response_cache: Mutex::new(response_cache),
4992 kv_pool: None,
4993 prefix_cache: None,
4994 sessions: session::SessionStore::new(),
4995 requests_total: std::sync::atomic::AtomicU64::new(0),
4996 request_errors_total: std::sync::atomic::AtomicU64::new(0),
4997 started_at: std::time::Instant::now(),
4998 last_request_ms: std::sync::atomic::AtomicU64::new(0),
4999 detection: Arc::new(health::Detection::ready(health::probe_backends())),
5000 mcp: None,
5001 continuous_batching_enabled: false,
5002 metal_private_decode_gate: None,
5003 loading_model: Mutex::new(None),
5004 last_load_error: Mutex::new(None),
5005 serving: Mutex::new(crate::stats::ServingStats::default()),
5006 maintenance: Mutex::new(crate::policy::maintenance::MaintenanceGate::serving()),
5007 footprint: Mutex::new(crate::policy::footprint::ProbeCache::new(FOOTPRINT_TTL_MS)),
5008 started_unix: unix_now(),
5009 }
5010 }
5011
5012 pub(crate) fn test_app() -> Router {
5021 test_app_with_state(Arc::new(test_state(
5022 test_model_full_byte_vocab(),
5023 ResponseCache::new(1000, Duration::from_secs(3600)),
5024 )))
5025 }
5026
5027 pub(crate) fn test_app_with_state(state: Arc<AppState>) -> Router {
5030 Router::new()
5040 .route(frink_api::routes::HEALTH, get(health))
5041 .merge(protected_routes())
5042 .with_state(state)
5043 }
5044
5045 fn named_test_model(name: &'static str, vocab_size: usize) -> Model {
5046 let mut cfg = test_dense_fixture();
5047 cfg.name = name;
5048 cfg.vocab_size = vocab_size;
5049 Model::Gguf(GgufModel {
5050 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5051 tokenizer: Arc::new(ServerTokenizer::Byte),
5052 stop_tokens: StopTokens::default(),
5053 bos_id: None,
5054 is_synthetic: true,
5055 chat_template: chat_template::PromptTemplate::plain(),
5056 })
5057 }
5058
5059 fn model_with_template(name: &'static str, source: &str) -> Model {
5063 let mut cfg = test_dense_fixture();
5064 cfg.name = name;
5065 cfg.vocab_size = 256;
5066 Model::Gguf(GgufModel {
5067 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5068 tokenizer: Arc::new(ServerTokenizer::Byte),
5069 stop_tokens: StopTokens::default(),
5070 bos_id: None,
5071 is_synthetic: true,
5072 chat_template: chat_template::PromptTemplate::from_gguf_metadata(
5073 Some(source),
5074 Some("qwen3"),
5075 false,
5076 true,
5077 None,
5078 None,
5079 ),
5080 })
5081 }
5082
5083 #[tokio::test]
5092 async fn a_template_that_rejects_the_conversation_is_a_400_on_the_streaming_path() {
5093 let strict = "{% if messages | length > 1 %}\
5096 {{ raise_exception('this template takes one turn') }}\
5097 {% endif %}{{ messages[0].content }}";
5098 let state = Arc::new(test_state(
5099 model_with_template("strict", strict),
5100 ResponseCache::new(4, Duration::from_secs(60)),
5101 ));
5102 let app = test_app_with_state(state);
5103
5104 let (status, body) = post_json_uri(
5105 &app,
5106 "/v1/chat/completions",
5107 serde_json::json!({
5108 "model": "strict",
5109 "stream": true,
5110 "messages": [
5111 {"role": "user", "content": "one"},
5112 {"role": "user", "content": "two"},
5113 ],
5114 }),
5115 )
5116 .await;
5117 assert_eq!(status, StatusCode::BAD_REQUEST);
5118 assert_eq!(body["error"]["param"], serde_json::json!("messages"));
5119 assert!(
5120 body["error"]["message"]
5121 .as_str()
5122 .unwrap()
5123 .contains("one turn"),
5124 "the template's own message must reach the caller: {body}"
5125 );
5126
5127 let (status, _) = post_json_uri(
5129 &app,
5130 "/v1/chat/completions",
5131 serde_json::json!({
5132 "model": "strict",
5133 "stream": true,
5134 "max_tokens": 1,
5135 "messages": [{"role": "user", "content": "one"}],
5136 }),
5137 )
5138 .await;
5139 assert_eq!(status, StatusCode::OK);
5140 }
5141
5142 #[tokio::test]
5144 async fn models_advertises_the_gears_this_checkpoint_actually_has() {
5145 let reasoning = "{% if enable_thinking %}<think>{% endif %}\
5146 {% if reasoning_effort %}\
5147 {% if reasoning_effort not in ['low','medium','high'] %}\
5148 {{ raise_exception('bad effort') }}\
5149 {% endif %}[{{ reasoning_effort }}]\
5150 {% endif %}{{ messages[0].content }}";
5151 let state = Arc::new(test_state(
5152 model_with_template("thinker", reasoning),
5153 ResponseCache::new(4, Duration::from_secs(60)),
5154 ));
5155 let app = test_app_with_state(state);
5156 let (status, models) = get_json(&app, frink_api::routes::V1_MODELS).await;
5157 assert_eq!(status, StatusCode::OK);
5158 let entry = &models["data"][0];
5159 assert_eq!(
5160 entry["supported_reasoning_efforts"],
5161 serde_json::json!(["off", "low", "medium", "high"])
5162 );
5163 assert_eq!(entry["default_reasoning_effort"], serde_json::json!("off"));
5164 }
5165
5166 #[tokio::test]
5171 async fn a_checkpoint_with_no_thinking_controls_advertises_neither_field() {
5172 let app = test_app();
5173 let (_, models) = get_json(&app, frink_api::routes::V1_MODELS).await;
5174 let entry = &models["data"][0];
5175 assert!(entry.get("supported_reasoning_efforts").is_none());
5176 assert!(entry.get("default_reasoning_effort").is_none());
5177 }
5178
5179 fn active_model(state: &AppState, name: &'static str) -> Arc<ActiveModel> {
5180 Arc::new(ActiveModel {
5181 id: Some(name.to_string()),
5182 loaded: Loaded::Generative(Arc::new(named_test_model(name, 256))),
5183 batcher: None,
5184 ceiling: None,
5185 checkpoint_path: None,
5186 })
5187 .tap_into(state)
5188 }
5189
5190 trait TapInto {
5192 fn tap_into(self, state: &AppState) -> Self;
5193 }
5194 impl TapInto for Arc<ActiveModel> {
5195 fn tap_into(self, state: &AppState) -> Self {
5196 state.swap_active(Some(Arc::clone(&self)));
5197 self
5198 }
5199 }
5200
5201 #[test]
5207 fn an_in_flight_request_keeps_the_model_it_started_on() {
5208 let state = test_state(
5209 named_test_model("model-a", 256),
5210 ResponseCache::new(4, Duration::from_secs(60)),
5211 );
5212
5213 let in_flight = state.active().expect("a model is loaded");
5216 assert_eq!(in_flight.name(), "model-a");
5217
5218 active_model(&state, "model-b");
5219
5220 assert_eq!(state.active().unwrap().name(), "model-b");
5222 assert_eq!(in_flight.name(), "model-a");
5224 let produced = run_generation(
5225 in_flight.generative().unwrap(),
5226 "hi",
5227 &greedy_params(3),
5228 None,
5229 None,
5230 None,
5231 None,
5232 None,
5233 None,
5234 )
5235 .expect("the old model must still decode after being swapped out");
5236 assert!(matches!(
5237 produced.choices[0].finish,
5238 FinishReason::Length | FinishReason::Stop
5239 ));
5240 }
5241
5242 #[test]
5247 fn a_swapped_out_model_lives_until_its_last_holder_releases_it() {
5248 let state = test_state(
5249 named_test_model("model-a", 256),
5250 ResponseCache::new(4, Duration::from_secs(60)),
5251 );
5252 let in_flight = state.active().expect("a model is loaded");
5253 let weights = Arc::clone(in_flight.generative().unwrap());
5254 assert!(Arc::strong_count(&weights) >= 2);
5255
5256 let previous = state.swap_active(Some(Arc::new(ActiveModel {
5257 id: Some("model-b".to_string()),
5258 loaded: Loaded::Generative(Arc::new(named_test_model("model-b", 256))),
5259 batcher: None,
5260 ceiling: None,
5261 checkpoint_path: None,
5262 })));
5263 drop(previous);
5264 assert!(Arc::strong_count(&weights) >= 2);
5266 drop(in_flight);
5267 assert_eq!(Arc::strong_count(&weights), 1);
5268 }
5269
5270 #[tokio::test]
5274 async fn unloading_answers_503_instead_of_serving_the_dropped_model() {
5275 let state = Arc::new(test_state(
5276 named_test_model("model-a", 256),
5277 ResponseCache::new(4, Duration::from_secs(60)),
5278 ));
5279 let app = test_app_with_state(Arc::clone(&state));
5280
5281 let (status, body) = post_json_uri(
5282 &app,
5283 frink_api::routes::ADMIN_MODELS_UNLOAD,
5284 serde_json::json!({}),
5285 )
5286 .await;
5287 assert_eq!(status, StatusCode::OK);
5288 assert_eq!(body["ok"], true);
5289 assert!(body["active"].is_null());
5290 assert!(state.active().is_none());
5291
5292 let (status, _) = get_json(&app, frink_api::routes::V1_MODELS).await;
5293 assert_eq!(status, StatusCode::OK);
5294 let (_, models) = get_json(&app, frink_api::routes::V1_MODELS).await;
5295 assert_eq!(models["data"].as_array().unwrap().len(), 0);
5296
5297 let (status, body) = post_json_uri(
5298 &app,
5299 "/v1/chat/completions",
5300 serde_json::json!({
5301 "model": "x",
5302 "messages": [{"role": "user", "content": "hi"}]
5303 }),
5304 )
5305 .await;
5306 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5307 assert_eq!(body["error"]["type"], "model_not_loaded");
5308 }
5309
5310 #[tokio::test]
5314 async fn health_reports_the_unloaded_state_rather_than_going_silent() {
5315 let state = Arc::new(test_state(
5316 named_test_model("model-a", 256),
5317 ResponseCache::new(4, Duration::from_secs(60)),
5318 ));
5319 let app = test_app_with_state(Arc::clone(&state));
5320 state.swap_active(None);
5321
5322 let (status, body) = get_json(&app, frink_api::routes::HEALTH).await;
5323 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5326 assert_eq!(body["state"], "unavailable");
5327 assert_eq!(body["reason"], "model_not_loaded");
5328 assert!(body["model"].is_null());
5329 let real_weights = body["capabilities"]
5330 .as_array()
5331 .unwrap()
5332 .iter()
5333 .find(|c| c["id"] == "real_weights")
5334 .cloned()
5335 .expect("real_weights is always reported");
5336 assert_eq!(real_weights["available"], false);
5337 assert_eq!(real_weights["reason"], "model_not_loaded");
5338 }
5339
5340 #[tokio::test]
5344 async fn a_finished_request_lands_in_the_stats_ring_with_both_durations() {
5345 let app = test_app();
5346
5347 let (status, completion) = post_json_uri(
5348 &app,
5349 "/v1/chat/completions",
5350 serde_json::json!({
5351 "model": "x",
5352 "messages": [{"role": "user", "content": "hi"}],
5353 "max_tokens": 4
5354 }),
5355 )
5356 .await;
5357 assert_eq!(status, StatusCode::OK);
5358 let request_id = completion["request_id"].as_str().unwrap().to_string();
5359
5360 let (status, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
5361 assert_eq!(status, StatusCode::OK);
5362 let recent = stats["recent"].as_array().unwrap();
5363 assert_eq!(recent.len(), 1);
5364 let row = &recent[0];
5365 assert_eq!(row["request_id"], request_id);
5366 assert_eq!(row["route"], frink_api::routes::V1_CHAT_COMPLETIONS);
5367 assert_eq!(row["status"], 200);
5368 assert_eq!(row["stream"], false);
5369 assert!(row["duration_ms"].is_number());
5372 assert!(row["decode_ms"].is_number());
5373 assert!(stats["tokens_generated_total"].as_u64().unwrap() > 0);
5374 assert_eq!(
5375 stats["tokens_prompt_total"].as_u64().unwrap(),
5376 row["prompt_tokens"].as_u64().unwrap()
5377 );
5378 }
5379
5380 #[tokio::test]
5384 async fn a_rejected_request_is_recorded_too() {
5385 let state = Arc::new(test_state(
5386 named_test_model("model-a", 256),
5387 ResponseCache::new(4, Duration::from_secs(60)),
5388 ));
5389 let app = test_app_with_state(Arc::clone(&state));
5390 state.swap_active(None);
5391
5392 let (status, _) = post_json_uri(
5393 &app,
5394 "/v1/chat/completions",
5395 serde_json::json!({"model": "x", "messages": [{"role": "user", "content": "hi"}]}),
5396 )
5397 .await;
5398 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
5399
5400 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
5401 let recent = stats["recent"].as_array().unwrap();
5402 assert_eq!(recent.len(), 1);
5403 assert_eq!(recent[0]["status"], 503);
5404 assert_eq!(recent[0]["completion_tokens"], 0);
5405 assert!(recent[0]["decode_ms"].is_null());
5406 assert_eq!(stats["errors_total"], 1);
5407 }
5408
5409 async fn post_json_with_headers(
5413 app: &Router,
5414 uri: &str,
5415 body: serde_json::Value,
5416 headers: &[(&str, &str)],
5417 ) -> (StatusCode, serde_json::Value) {
5418 use http_body_util::BodyExt;
5419 use tower::ServiceExt;
5420
5421 let mut builder = axum::http::Request::builder()
5422 .method("POST")
5423 .uri(uri)
5424 .header("content-type", "application/json");
5425 for (name, value) in headers {
5426 builder = builder.header(*name, *value);
5427 }
5428 let response = app
5429 .clone()
5430 .oneshot(
5431 builder
5432 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
5433 .unwrap(),
5434 )
5435 .await
5436 .unwrap();
5437 let status = response.status();
5438 let bytes = response.into_body().collect().await.unwrap().to_bytes();
5439 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
5440 (status, json)
5441 }
5442
5443 #[tokio::test]
5447 async fn tokenize_detokenize_and_embeddings_all_land_in_the_ring() {
5448 let app = test_app();
5449
5450 let (status, _) = post_json_uri(
5451 &app,
5452 frink_api::routes::V1_TOKENIZE,
5453 serde_json::json!({"prompt": "hello"}),
5454 )
5455 .await;
5456 assert_eq!(status, StatusCode::OK);
5457 let (status, _) = post_json_uri(
5458 &app,
5459 frink_api::routes::V1_DETOKENIZE,
5460 serde_json::json!({"tokens": [104, 105]}),
5461 )
5462 .await;
5463 assert_eq!(status, StatusCode::OK);
5464 let (status, _) = post_json_uri(
5465 &app,
5466 frink_api::routes::V1_EMBEDDINGS,
5467 serde_json::json!({"input": "hello"}),
5468 )
5469 .await;
5470 assert_eq!(status, StatusCode::OK);
5471
5472 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
5473 let routes: Vec<&str> = stats["recent"]
5474 .as_array()
5475 .unwrap()
5476 .iter()
5477 .map(|row| row["route"].as_str().unwrap())
5478 .collect();
5479 for expected in [
5480 frink_api::routes::V1_TOKENIZE,
5481 frink_api::routes::V1_DETOKENIZE,
5482 frink_api::routes::V1_EMBEDDINGS,
5483 ] {
5484 assert!(
5485 routes.contains(&expected),
5486 "{expected} is missing: {routes:?}"
5487 );
5488 }
5489
5490 let row = |route: &str| {
5491 stats["recent"]
5492 .as_array()
5493 .unwrap()
5494 .iter()
5495 .find(|r| r["route"] == route)
5496 .cloned()
5497 .unwrap()
5498 };
5499 let embed = row(frink_api::routes::V1_EMBEDDINGS);
5503 assert!(embed["prompt_tokens"].as_u64().unwrap() > 0);
5504 assert!(embed["decode_ms"].is_null());
5505 assert_eq!(embed["completion_tokens"], 0);
5506 assert_eq!(row(frink_api::routes::V1_TOKENIZE)["prompt_tokens"], 0);
5510 assert_eq!(
5511 stats["tokens_prompt_total"].as_u64().unwrap(),
5512 embed["prompt_tokens"].as_u64().unwrap(),
5513 "only the forward pass counted"
5514 );
5515 }
5516
5517 fn streaming_test_app() -> Router {
5522 let mut cfg = test_dense_fixture();
5523 cfg.vocab_size = 256;
5524 let model = Model::Gguf(GgufModel {
5525 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
5526 tokenizer: Arc::new(ServerTokenizer::Byte),
5527 stop_tokens: StopTokens::default(),
5528 bos_id: None,
5529 is_synthetic: false,
5530 chat_template: chat_template::PromptTemplate::plain(),
5531 });
5532 test_app_with_state(Arc::new(test_state(
5533 model,
5534 ResponseCache::new(1000, Duration::from_secs(3600)),
5535 )))
5536 }
5537
5538 #[tokio::test]
5547 async fn chat_logprobs_are_rendered_and_are_never_served_from_cache() {
5548 let app = test_app();
5549 let body = |logprobs: Option<(bool, Option<u32>)>| {
5550 let mut b = serde_json::json!({
5551 "model": "x",
5552 "messages": [{"role": "user", "content": "hi"}],
5553 "max_tokens": 4
5554 });
5555 if let Some((on, top)) = logprobs {
5556 b["logprobs"] = serde_json::json!(on);
5557 if let Some(n) = top {
5558 b["top_logprobs"] = serde_json::json!(n);
5559 }
5560 }
5561 b
5562 };
5563
5564 let (status, plain) =
5566 post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, body(None)).await;
5567 assert_eq!(status, StatusCode::OK, "{plain}");
5568 assert!(plain["choices"][0]["logprobs"].is_null(), "{plain}");
5569
5570 for attempt in 0..2 {
5574 let (status, with) = post_json_uri(
5575 &app,
5576 frink_api::routes::V1_CHAT_COMPLETIONS,
5577 body(Some((true, Some(2)))),
5578 )
5579 .await;
5580 assert_eq!(status, StatusCode::OK, "{with}");
5581 assert_ne!(
5582 with["frink_cache"], "hit",
5583 "attempt {attempt} replayed a cached answer for a logprobs request: {with}"
5584 );
5585 let lp = &with["choices"][0]["logprobs"];
5586 assert!(lp.is_object(), "attempt {attempt}: {with}");
5587 let content = lp["content"].as_array().expect("content");
5588 assert!(lp["tokens"].is_null(), "completions shape leaked: {lp}");
5590 for entry in content {
5591 assert!(entry["token"].is_string(), "{entry}");
5592 assert!(entry["bytes"].is_array(), "{entry}");
5593 let v = entry["logprob"].as_f64().expect("a real number");
5594 assert!(v <= 0.0 && v.is_finite(), "{entry}");
5595 let top = entry["top_logprobs"].as_array().expect("top_logprobs");
5596 assert!(top.len() <= 2, "asked for 2, got {}", top.len());
5597 }
5598 }
5599 }
5600
5601 #[tokio::test]
5607 async fn the_chat_logprobs_pair_is_validated() {
5608 let app = test_app();
5609 for (extra, why) in [
5610 (serde_json::json!({"top_logprobs": 3}), "without logprobs"),
5611 (
5612 serde_json::json!({"logprobs": true, "top_logprobs": 21}),
5613 "above the cap",
5614 ),
5615 ] {
5616 let mut body = serde_json::json!({
5617 "model": "x",
5618 "messages": [{"role": "user", "content": "hi"}],
5619 "max_tokens": 2
5620 });
5621 for (k, v) in extra.as_object().unwrap() {
5622 body[k] = v.clone();
5623 }
5624 let (status, answer) =
5625 post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, body).await;
5626 assert_eq!(status, StatusCode::BAD_REQUEST, "{why}: {answer}");
5627 assert!(
5628 answer["error"]["message"]
5629 .as_str()
5630 .is_some_and(|m| m.contains("top_logprobs")),
5631 "{why}: {answer}"
5632 );
5633 }
5634 }
5635
5636 #[tokio::test]
5645 async fn sleep_refuses_a_model_it_could_not_bring_back() {
5646 let app = test_app();
5647 let (status, answer) =
5648 post_json_uri(&app, frink_api::routes::SLEEP, serde_json::json!({})).await;
5649 assert_eq!(status, StatusCode::CONFLICT, "{answer}");
5650 assert_eq!(answer["error"]["type"], "not_reloadable", "{answer}");
5651 let (_, still) = get_json_uri(&app, frink_api::routes::IS_SLEEPING).await;
5654 assert_eq!(still["is_sleeping"], false, "{still}");
5655 let (status, _) = post_json_uri(
5656 &app,
5657 frink_api::routes::V1_CHAT_COMPLETIONS,
5658 serde_json::json!({
5659 "model": "x",
5660 "messages": [{"role": "user", "content": "hi"}],
5661 "max_tokens": 2
5662 }),
5663 )
5664 .await;
5665 assert_eq!(status, StatusCode::OK, "a refused sleep unloaded the model");
5666 }
5667
5668 #[tokio::test]
5677 async fn sleep_remembers_what_unload_forgets() {
5678 let state = Arc::new(test_state_at(
5681 test_model_full_byte_vocab(),
5682 ResponseCache::new(1000, Duration::from_secs(3600)),
5683 Some(std::path::PathBuf::from("/nonexistent/fixture.gguf")),
5684 ));
5685 let app = test_app_with_state(Arc::clone(&state));
5686 let ask = || {
5687 let app = app.clone();
5688 async move {
5689 post_json_uri(
5690 &app,
5691 frink_api::routes::V1_CHAT_COMPLETIONS,
5692 serde_json::json!({
5693 "model": "x",
5694 "messages": [{"role": "user", "content": "hi"}],
5695 "max_tokens": 2
5696 }),
5697 )
5698 .await
5699 }
5700 };
5701
5702 let (status, _) = ask().await;
5703 assert_eq!(status, StatusCode::OK, "the fixture server serves");
5704 let (_, awake) = get_json_uri(&app, frink_api::routes::IS_SLEEPING).await;
5705 assert_eq!(awake["is_sleeping"], false, "{awake}");
5706
5707 let (status, slept) =
5708 post_json_uri(&app, frink_api::routes::SLEEP, serde_json::json!({})).await;
5709 assert_eq!(status, StatusCode::OK, "{slept}");
5710 assert_eq!(slept["is_sleeping"], true, "{slept}");
5711 let (_, now) = get_json_uri(&app, frink_api::routes::IS_SLEEPING).await;
5712 assert_eq!(now["is_sleeping"], true, "{now}");
5713
5714 let (status, refused) = ask().await;
5717 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{refused}");
5718 assert_eq!(
5719 refused["error"]["type"], "server_sleeping",
5720 "an asleep server reported itself as empty: {refused}"
5721 );
5722
5723 let (status, again) =
5725 post_json_uri(&app, frink_api::routes::SLEEP, serde_json::json!({})).await;
5726 assert_eq!(status, StatusCode::OK, "{again}");
5727 assert_eq!(again["is_sleeping"], true, "{again}");
5728 }
5729
5730 #[tokio::test]
5734 async fn waking_a_server_that_is_awake_is_refused() {
5735 let app = test_app();
5736 let (status, answer) =
5737 post_json_uri(&app, frink_api::routes::WAKE_UP, serde_json::json!({})).await;
5738 assert_eq!(status, StatusCode::CONFLICT, "{answer}");
5739 assert_eq!(answer["error"]["type"], "not_sleeping", "{answer}");
5740 }
5741
5742 #[tokio::test]
5749 async fn a_salt_keeps_one_callers_cached_answer_from_another() {
5750 let app = test_app();
5751 let body = |salt: Option<&str>| {
5752 let mut b = serde_json::json!({
5753 "model": "x",
5754 "messages": [{"role": "user", "content": "the same prompt"}],
5755 "max_tokens": 4,
5756 "seed": 1
5757 });
5758 if let Some(s) = salt {
5759 b["cache_salt"] = serde_json::json!(s);
5760 }
5761 b
5762 };
5763 let post = |b: serde_json::Value| {
5764 let app = app.clone();
5765 async move { post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, b).await }
5766 };
5767
5768 let (status, _) = post(body(Some("tenant-a"))).await;
5770 assert_eq!(status, StatusCode::OK);
5771 let (_, again) = post(body(Some("tenant-a"))).await;
5772 assert_eq!(
5773 again["frink_cache"], "hit",
5774 "the owner did not get its own entry back: {again}"
5775 );
5776
5777 let (_, other) = post(body(Some("tenant-b"))).await;
5779 assert_ne!(
5780 other["frink_cache"], "hit",
5781 "a different caller was served tenant-a's answer: {other}"
5782 );
5783
5784 let (_, shared) = post(body(None)).await;
5786 assert_ne!(
5787 shared["frink_cache"], "hit",
5788 "an unsalted request was served a salted answer: {shared}"
5789 );
5790 }
5791
5792 #[tokio::test]
5797 async fn chat_serves_several_choices_and_refuses_the_streaming_pair() {
5798 let app = test_app();
5799 let body = |n: u32, stream: bool| {
5800 serde_json::json!({
5801 "model": "x",
5802 "messages": [{"role": "user", "content": "hi"}],
5803 "max_tokens": 4,
5804 "temperature": 1.0,
5805 "n": n,
5806 "stream": stream
5807 })
5808 };
5809
5810 let (status, one) =
5811 post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, body(1, false)).await;
5812 assert_eq!(status, StatusCode::OK, "{one}");
5813
5814 let (status, three) =
5815 post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, body(3, false)).await;
5816 assert_eq!(status, StatusCode::OK, "{three}");
5817 let choices = three["choices"].as_array().expect("an array");
5818 assert_eq!(choices.len(), 3, "{three}");
5819 for (i, c) in choices.iter().enumerate() {
5820 assert_eq!(c["index"], i);
5821 assert!(c["message"]["role"].is_string(), "{c}");
5822 assert!(c["finish_reason"].is_string(), "{c}");
5823 }
5824 assert_eq!(
5826 three["usage"]["prompt_tokens"], one["usage"]["prompt_tokens"],
5827 "n = 3 billed the prompt more than once"
5828 );
5829
5830 let (status, refused) =
5833 post_json_uri(&app, frink_api::routes::V1_CHAT_COMPLETIONS, body(3, true)).await;
5834 assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{refused}");
5835 let message = refused["error"]["message"].as_str().unwrap_or_default();
5836 assert!(
5837 message.contains('n') && message.contains("stream"),
5838 "{refused}"
5839 );
5840 }
5841
5842 #[tokio::test]
5852 async fn every_route_refuses_the_same_unimplemented_fields() {
5853 let app = test_app();
5854 let fields = [
5855 ("n", serde_json::json!(3)),
5856 ("best_of", serde_json::json!(2)),
5857 ("prompt_logprobs", serde_json::json!(1)),
5858 ("echo", serde_json::json!(true)),
5859 ("use_beam_search", serde_json::json!(true)),
5860 ("truncate_prompt_tokens", serde_json::json!(8)),
5861 ("prompt_embeds", serde_json::json!("AA==")),
5862 ("allowed_token_ids", serde_json::json!([1, 2])),
5863 ("bad_words", serde_json::json!(["x"])),
5864 ("skip_special_tokens", serde_json::json!(false)),
5865 ("return_tokens_as_token_ids", serde_json::json!(true)),
5866 ];
5867 for (field, value) in fields {
5868 for (uri, base) in [
5869 (
5870 frink_api::routes::V1_CHAT_COMPLETIONS,
5871 serde_json::json!({
5872 "model": "x",
5873 "messages": [{"role": "user", "content": "hi"}],
5874 "max_tokens": 2
5875 }),
5876 ),
5877 (
5878 frink_api::routes::V1_COMPLETIONS,
5879 serde_json::json!({"prompt": "hi", "max_tokens": 2}),
5880 ),
5881 (
5882 frink_api::routes::COMPLETION,
5883 serde_json::json!({"prompt": "hi", "n_predict": 2}),
5884 ),
5885 ] {
5886 let mut body = base;
5887 body[field] = value.clone();
5888 if field == "prompt_logprobs" && uri == frink_api::routes::V1_COMPLETIONS {
5895 let (status, answer) = post_json_uri(&app, uri, body).await;
5896 assert_eq!(status, StatusCode::OK, "{uri} refused it: {answer}");
5897 assert!(
5898 answer["prompt_logprobs"].is_array(),
5899 "served without the field: {answer}"
5900 );
5901 continue;
5902 }
5903 if (field == "n" || field == "best_of")
5904 && (uri == frink_api::routes::V1_COMPLETIONS
5905 || uri == frink_api::routes::V1_CHAT_COMPLETIONS)
5906 {
5907 let (status, answer) = post_json_uri(&app, uri, body).await;
5908 assert_eq!(
5909 status,
5910 StatusCode::OK,
5911 "{uri} refused a served `{field}`: {answer}"
5912 );
5913 let want = if field == "n" { 3 } else { 1 };
5917 assert_eq!(
5918 answer["choices"].as_array().map(Vec::len),
5919 Some(want),
5920 "{field}: {answer}"
5921 );
5922 continue;
5923 }
5924 let (status, answer) = post_json_uri(&app, uri, body).await;
5925 assert_eq!(
5926 status,
5927 StatusCode::NOT_IMPLEMENTED,
5928 "{uri} served `{field}` instead of refusing it: {answer}"
5929 );
5930 assert!(
5931 answer["error"]["message"]
5932 .as_str()
5933 .is_some_and(|m| m.contains(field)),
5934 "{uri} refused `{field}` without naming it: {answer}"
5935 );
5936 }
5937 }
5938 }
5939
5940 #[tokio::test]
5941 async fn the_native_completion_wire_is_not_the_openai_one() {
5942 let app = test_app();
5943
5944 let (status, native) = post_json_uri(
5945 &app,
5946 frink_api::routes::COMPLETION,
5947 serde_json::json!({"prompt": "hi", "n_predict": 4}),
5948 )
5949 .await;
5950 assert_eq!(status, StatusCode::OK, "{native}");
5951 assert!(native["content"].is_string(), "{native}");
5952 assert_eq!(native["stop"], true);
5953 assert_eq!(native["stop_type"], "limit");
5954 assert_eq!(native["stopping_word"], "");
5955 assert_eq!(native["truncated"], false);
5956 assert_eq!(native["id_slot"], -1);
5957 assert!(native["timings"]["prompt_n"].is_number(), "{native}");
5958 assert!(native["generation_settings"]["n_predict"] == 4, "{native}");
5959 assert!(
5960 native.get("choices").is_none(),
5961 "the native shape has no `choices`: {native}"
5962 );
5963
5964 let (status, openai) = post_json_uri(
5965 &app,
5966 frink_api::routes::V1_COMPLETIONS,
5967 serde_json::json!({"prompt": "hi", "max_tokens": 4}),
5968 )
5969 .await;
5970 assert_eq!(status, StatusCode::OK);
5971 assert!(openai["choices"][0]["text"].is_string(), "{openai}");
5972 assert!(
5973 openai.get("content").is_none(),
5974 "the OpenAI shape has no top-level `content`: {openai}"
5975 );
5976 }
5977
5978 #[tokio::test]
5982 async fn both_native_spellings_reach_the_same_handler() {
5983 let app = test_app();
5984 for route in [
5985 frink_api::routes::COMPLETION,
5986 frink_api::routes::COMPLETIONS,
5987 ] {
5988 let (status, body) = post_json_uri(
5989 &app,
5990 route,
5991 serde_json::json!({"prompt": "hi", "n_predict": 2, "seed": 1}),
5992 )
5993 .await;
5994 assert_eq!(status, StatusCode::OK, "{route}: {body}");
5995 assert_eq!(body["stop"], true, "{route}");
5996 assert!(body["content"].is_string(), "{route}");
5997 }
5998
5999 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6002 let routes: Vec<&str> = stats["recent"]
6003 .as_array()
6004 .unwrap()
6005 .iter()
6006 .map(|row| row["route"].as_str().unwrap())
6007 .collect();
6008 assert!(
6009 routes.contains(&frink_api::routes::COMPLETION),
6010 "{routes:?}"
6011 );
6012 assert!(
6013 routes.contains(&frink_api::routes::COMPLETIONS),
6014 "{routes:?}"
6015 );
6016 }
6017
6018 #[tokio::test]
6024 async fn a_native_stream_ends_on_a_stop_frame_with_no_done_sentinel() {
6025 let app = streaming_test_app();
6026 let raw = post_sse_raw_uri(
6027 &app,
6028 frink_api::routes::COMPLETION,
6029 serde_json::json!({"prompt": "hi", "n_predict": 6, "stream": true, "seed": 7}),
6030 )
6031 .await;
6032
6033 assert!(
6034 !raw.contains("[DONE]"),
6035 "llama.cpp's native stream has no sentinel: {raw}"
6036 );
6037 let frames: Vec<serde_json::Value> = raw
6038 .lines()
6039 .filter_map(|line| line.strip_prefix("data: "))
6040 .map(|json| serde_json::from_str(json).expect("every frame is one JSON object"))
6041 .collect();
6042 assert!(frames.len() >= 2, "expected partials then a final: {raw}");
6043
6044 let (last, partials) = frames.split_last().unwrap();
6045 assert_eq!(last["stop"], true, "the last frame closes the stream");
6046 assert!(last["timings"].is_object(), "{last}");
6047 assert!(last["stop_type"].is_string(), "{last}");
6048 for partial in partials {
6049 assert_eq!(partial["stop"], false, "{partial}");
6050 assert!(partial["content"].is_string(), "{partial}");
6051 assert!(partial.get("timings").is_none(), "{partial}");
6055 assert!(partial.get("generation_settings").is_none(), "{partial}");
6056 }
6057 let streamed: String = partials
6060 .iter()
6061 .filter_map(|p| p["content"].as_str())
6062 .collect();
6063 assert_eq!(last["content"].as_str().unwrap(), streamed);
6064 }
6065
6066 #[tokio::test]
6071 async fn an_unbounded_n_predict_is_refused_rather_than_quietly_shrunk() {
6072 let app = test_app();
6073 for body in [
6074 serde_json::json!({"prompt": "hi"}),
6075 serde_json::json!({"prompt": "hi", "n_predict": -1}),
6076 ] {
6077 let (status, refusal) =
6078 post_json_uri(&app, frink_api::routes::COMPLETION, body.clone()).await;
6079 assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{body}: {refusal}");
6080 assert!(
6081 refusal["error"]["message"]
6082 .as_str()
6083 .unwrap()
6084 .contains("n_predict"),
6085 "{refusal}"
6086 );
6087 }
6088 let (status, _) = post_json_uri(
6091 &app,
6092 frink_api::routes::COMPLETION,
6093 serde_json::json!({"prompt": "hi", "n_predict": 2}),
6094 )
6095 .await;
6096 assert_eq!(status, StatusCode::OK);
6097 }
6098
6099 #[tokio::test]
6108 async fn a_stop_string_halts_the_answer_and_is_named_back() {
6109 let app = streaming_test_app();
6110 let ask = |stop: serde_json::Value| {
6111 let app = app.clone();
6112 async move {
6113 post_json_uri(
6114 &app,
6115 frink_api::routes::COMPLETION,
6116 serde_json::json!({
6117 "prompt": "hi",
6118 "n_predict": 64,
6119 "ignore_eos": true,
6120 "stop": stop,
6121 }),
6122 )
6123 .await
6124 .1
6125 }
6126 };
6127
6128 let baseline = ask(serde_json::json!([])).await;
6129 assert_eq!(baseline["stop_type"], "limit");
6130 assert_eq!(baseline["stopping_word"], "");
6131 let text = baseline["content"].as_str().unwrap().to_string();
6132 let sentinel: String = text.chars().skip(1).take(2).collect();
6138 assert_eq!(
6139 sentinel.chars().count(),
6140 2,
6141 "the fixture must produce enough output to cut: {text:?}"
6142 );
6143 let cut = text.find(&sentinel).expect("it came out of this text");
6144
6145 let stopped = ask(serde_json::json!([sentinel])).await;
6146 assert_eq!(stopped["stop_type"], "word", "{stopped}");
6147 assert_eq!(stopped["stopping_word"], sentinel);
6148 assert_eq!(
6149 stopped["content"].as_str().unwrap(),
6150 &text[..cut],
6151 "the answer must be cut at the sentinel, not run past it"
6152 );
6153 }
6154
6155 #[tokio::test]
6161 async fn the_llama_cpp_spelling_of_tokenize_reaches_the_same_handler() {
6162 let app = test_app();
6163
6164 let (v1_status, v1) = post_json_uri(
6165 &app,
6166 frink_api::routes::V1_TOKENIZE,
6167 serde_json::json!({"prompt": "hello"}),
6168 )
6169 .await;
6170 let (alias_status, alias) = post_json_uri(
6171 &app,
6172 frink_api::routes::TOKENIZE,
6173 serde_json::json!({"content": "hello"}),
6174 )
6175 .await;
6176 assert_eq!(v1_status, StatusCode::OK);
6177 assert_eq!(alias_status, StatusCode::OK, "{alias}");
6178 assert_eq!(v1["tokens"], alias["tokens"]);
6179 assert!(!alias["tokens"].as_array().unwrap().is_empty());
6180
6181 let (status, both_ways) = post_json_uri(
6184 &app,
6185 frink_api::routes::TOKENIZE,
6186 serde_json::json!({"prompt": "hello"}),
6187 )
6188 .await;
6189 assert_eq!(status, StatusCode::OK);
6190 assert_eq!(both_ways["tokens"], v1["tokens"]);
6191 }
6192
6193 #[tokio::test]
6198 async fn detokenize_answers_under_both_dialects_keys() {
6199 let app = test_app();
6200 for route in [
6201 frink_api::routes::DETOKENIZE,
6202 frink_api::routes::V1_DETOKENIZE,
6203 ] {
6204 let (status, body) =
6205 post_json_uri(&app, route, serde_json::json!({"tokens": [104, 105]})).await;
6206 assert_eq!(status, StatusCode::OK, "{route}");
6207 assert_eq!(body["text"], "hi", "{route}");
6208 assert_eq!(body["content"], body["text"], "{route}");
6209 }
6210 }
6211
6212 #[tokio::test]
6216 async fn the_alias_is_recorded_under_the_path_the_client_called() {
6217 let app = test_app();
6218 let (status, _) = post_json_uri(
6219 &app,
6220 frink_api::routes::TOKENIZE,
6221 serde_json::json!({"content": "hello"}),
6222 )
6223 .await;
6224 assert_eq!(status, StatusCode::OK);
6225
6226 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6227 let routes: Vec<&str> = stats["recent"]
6228 .as_array()
6229 .unwrap()
6230 .iter()
6231 .map(|row| row["route"].as_str().unwrap())
6232 .collect();
6233 assert!(
6234 routes.contains(&frink_api::routes::TOKENIZE),
6235 "the alias must be its own row: {routes:?}"
6236 );
6237 assert!(
6238 !routes.contains(&frink_api::routes::V1_TOKENIZE),
6239 "nothing called /v1/tokenize: {routes:?}"
6240 );
6241 }
6242
6243 #[tokio::test]
6248 async fn add_special_prepends_the_same_bos_the_decoder_would() {
6249 let mut cfg = test_dense_fixture();
6250 cfg.vocab_size = 256;
6251 let model = Model::Gguf(GgufModel {
6252 decoder: Arc::new(Decoder::new_random_small(cfg, 2, 256)),
6253 tokenizer: Arc::new(ServerTokenizer::Byte),
6254 stop_tokens: StopTokens::default(),
6255 bos_id: Some(7),
6256 is_synthetic: true,
6257 chat_template: chat_template::PromptTemplate::plain(),
6258 });
6259 let app = test_app_with_state(Arc::new(test_state(
6260 model,
6261 ResponseCache::new(1000, Duration::from_secs(3600)),
6262 )));
6263
6264 let (_, plain) = post_json_uri(
6265 &app,
6266 frink_api::routes::TOKENIZE,
6267 serde_json::json!({"content": "hi"}),
6268 )
6269 .await;
6270 let (_, special) = post_json_uri(
6271 &app,
6272 frink_api::routes::TOKENIZE,
6273 serde_json::json!({"content": "hi", "add_special": true}),
6274 )
6275 .await;
6276
6277 assert_eq!(plain["tokens"], serde_json::json!([104, 105]));
6278 assert_eq!(special["tokens"], serde_json::json!([7, 104, 105]));
6279 assert_eq!(special["count"], 3);
6280 }
6281
6282 #[tokio::test]
6285 async fn a_rejected_embeddings_request_is_recorded_with_its_status() {
6286 let app = test_app();
6287 let (status, _) = post_json_uri(
6288 &app,
6289 frink_api::routes::V1_EMBEDDINGS,
6290 serde_json::json!({"input": "hi", "encoding_format": "base64"}),
6291 )
6292 .await;
6293 assert_eq!(status, StatusCode::BAD_REQUEST);
6294
6295 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6296 let recent = stats["recent"].as_array().unwrap();
6297 assert_eq!(recent.len(), 1);
6298 assert_eq!(recent[0]["route"], frink_api::routes::V1_EMBEDDINGS);
6299 assert_eq!(recent[0]["status"], 400);
6300 assert_eq!(
6301 recent[0]["prompt_tokens"], 0,
6302 "a rejected call embedded nothing"
6303 );
6304 }
6305
6306 #[tokio::test]
6309 async fn a_row_names_the_key_that_served_it_without_carrying_the_key() {
6310 let app = test_app();
6311 let key = "sk-monitor-secret";
6312 let (status, _) = post_json_with_headers(
6313 &app,
6314 "/v1/chat/completions",
6315 serde_json::json!({
6316 "model": "x",
6317 "messages": [{"role": "user", "content": "hi"}],
6318 "max_tokens": 2
6319 }),
6320 &[
6321 ("authorization", &format!("Bearer {key}")),
6322 ("x-frink-client", "frink-studio"),
6323 ],
6324 )
6325 .await;
6326 assert_eq!(status, StatusCode::OK);
6327
6328 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6329 let row = stats["recent"].as_array().unwrap()[0].clone();
6330 let fingerprint = row["via_api_key"]
6331 .as_str()
6332 .expect("the row names the key that served it")
6333 .to_string();
6334 assert_eq!(fingerprint, attribution::key_fingerprint(key));
6335 assert!(!fingerprint.contains(key));
6336 assert!(
6337 !serde_json::to_string(&stats).unwrap().contains(key),
6338 "the stats payload must not carry the key in any form"
6339 );
6340 assert_eq!(row["client"], "frink-studio");
6341 }
6342
6343 #[tokio::test]
6346 async fn different_keys_are_different_callers_and_no_key_is_null() {
6347 let app = test_app();
6348 let body = serde_json::json!({
6349 "model": "x",
6350 "messages": [{"role": "user", "content": "hi"}],
6351 "max_tokens": 1
6352 });
6353 for headers in [
6354 vec![("authorization", "Bearer key-one")],
6355 vec![("authorization", "Bearer key-two")],
6356 vec![],
6357 ] {
6358 let (status, _) =
6359 post_json_with_headers(&app, "/v1/chat/completions", body.clone(), &headers).await;
6360 assert_eq!(status, StatusCode::OK);
6361 }
6362
6363 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6364 let recent = stats["recent"].as_array().unwrap();
6365 assert_eq!(recent.len(), 3);
6366 let one = recent[0]["via_api_key"].as_str().unwrap();
6367 let two = recent[1]["via_api_key"].as_str().unwrap();
6368 assert_ne!(one, two, "two keys must not collapse into one caller");
6369 assert!(
6370 recent[2]["via_api_key"].is_null(),
6371 "an unauthenticated call is null, not a fingerprint of nothing"
6372 );
6373 assert!(recent[2]["client"].is_null());
6374 }
6375
6376 #[tokio::test]
6381 async fn a_row_names_the_model_that_served_it_not_the_one_requested() {
6382 let state = Arc::new(test_state(
6383 named_test_model("really-loaded", 256),
6384 ResponseCache::new(4, Duration::from_secs(60)),
6385 ));
6386 let app = test_app_with_state(Arc::clone(&state));
6387
6388 let (status, _) = post_json_uri(
6389 &app,
6390 "/v1/chat/completions",
6391 serde_json::json!({
6392 "model": "gpt-4-turbo-that-is-not-here",
6393 "messages": [{"role": "user", "content": "hi"}],
6394 "max_tokens": 2
6395 }),
6396 )
6397 .await;
6398 assert_eq!(status, StatusCode::OK);
6399
6400 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6401 assert_eq!(stats["recent"][0]["model"], "really-loaded");
6402
6403 state.swap_active(None);
6406 let (status, _) = post_json_uri(
6407 &app,
6408 "/v1/chat/completions",
6409 serde_json::json!({
6410 "model": "gpt-4-turbo-that-is-not-here",
6411 "messages": [{"role": "user", "content": "hi"}]
6412 }),
6413 )
6414 .await;
6415 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
6416 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6417 let recent = stats["recent"].as_array().unwrap();
6418 assert!(recent[recent.len() - 1]["model"].is_null());
6419 }
6420
6421 #[tokio::test]
6425 async fn a_streamed_row_names_the_model_it_decoded_against() {
6426 let state = Arc::new(test_state(
6427 named_test_model("model-before", 256),
6428 ResponseCache::new(4, Duration::from_secs(60)),
6429 ));
6430 let app = test_app_with_state(Arc::clone(&state));
6431 let _ = post_sse_raw(&app, resumable_request()).await;
6432 active_model(&state, "model-after");
6434
6435 let (_, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6436 assert_eq!(stats["recent"][0]["model"], "model-before");
6437 }
6438
6439 #[tokio::test]
6442 async fn the_queue_gauge_is_null_when_nothing_can_queue() {
6443 let app = test_app();
6444 let (status, stats) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
6445 assert_eq!(status, StatusCode::OK);
6446 assert!(
6447 stats["queue_depth"].is_null(),
6448 "without continuous batching nothing queues, so there is nothing to measure"
6449 );
6450 assert!(stats["queue_rejected_total"].is_null());
6451 assert_eq!(
6452 stats["generating_now"], 0,
6453 "work in progress is measured and really is zero here"
6454 );
6455 }
6456
6457 async fn post_sse_raw(app: &Router, body: serde_json::Value) -> String {
6462 post_sse_raw_uri(app, frink_api::routes::V1_CHAT_COMPLETIONS, body).await
6463 }
6464
6465 async fn post_sse_raw_uri(app: &Router, uri: &str, body: serde_json::Value) -> String {
6469 use http_body_util::BodyExt;
6470 use tower::ServiceExt;
6471
6472 let response = app
6473 .clone()
6474 .oneshot(
6475 axum::http::Request::builder()
6476 .method("POST")
6477 .uri(uri)
6478 .header("content-type", "application/json")
6479 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
6480 .unwrap(),
6481 )
6482 .await
6483 .unwrap();
6484 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6485 String::from_utf8(bytes.to_vec()).unwrap()
6486 }
6487
6488 async fn get_json_with_headers(
6489 app: &Router,
6490 uri: &str,
6491 headers: &[(&str, &str)],
6492 ) -> (StatusCode, serde_json::Value) {
6493 use http_body_util::BodyExt;
6494 use tower::ServiceExt;
6495
6496 let mut builder = axum::http::Request::builder().method("GET").uri(uri);
6497 for (name, value) in headers {
6498 builder = builder.header(*name, *value);
6499 }
6500 let response = app
6501 .clone()
6502 .oneshot(builder.body(axum::body::Body::empty()).unwrap())
6503 .await
6504 .unwrap();
6505 let status = response.status();
6506 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6507 (
6508 status,
6509 serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({})),
6510 )
6511 }
6512
6513 fn sse_field<'a>(body: &'a str, field: &str) -> Vec<&'a str> {
6514 body.lines()
6515 .filter_map(|line| line.strip_prefix(field))
6516 .map(str::trim)
6517 .collect()
6518 }
6519
6520 fn resumable_request() -> serde_json::Value {
6521 serde_json::json!({
6522 "model": "m",
6523 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
6524 "max_tokens": 4,
6525 "temperature": 0,
6526 "stream": true,
6527 "stream_resumable": true,
6528 })
6529 }
6530
6531 #[tokio::test]
6536 async fn a_resumable_stream_numbers_every_event_and_states_retry_once() {
6537 let app = test_app();
6538 let body = post_sse_raw(&app, resumable_request()).await;
6539
6540 let request_id = body
6541 .lines()
6542 .find_map(|l| l.strip_prefix("data: "))
6543 .and_then(|d| serde_json::from_str::<serde_json::Value>(d).ok())
6544 .and_then(|v| v["request_id"].as_str().map(str::to_string))
6545 .expect("the first chunk names the request");
6546
6547 let ids = sse_field(&body, "id:");
6548 let datas = sse_field(&body, "data:");
6549 assert_eq!(
6550 ids.len(),
6551 datas.len(),
6552 "every event carries an id, or a reconnect cannot name where it stopped"
6553 );
6554 for (i, id) in ids.iter().enumerate() {
6555 assert_eq!(*id, format!("{request_id}:{i}"));
6556 }
6557 let retries = sse_field(&body, "retry:");
6558 assert_eq!(
6559 retries.len(),
6560 1,
6561 "the reconnect delay is stated once, not on every event"
6562 );
6563 assert_eq!(retries[0], "1500");
6564 assert!(
6565 body.contains("data: [DONE]"),
6566 "the end of stream is still stated"
6567 );
6568 }
6569
6570 #[tokio::test]
6574 async fn a_plain_stream_carries_no_id_because_nothing_could_replay_it() {
6575 let app = test_app();
6576 let mut request = resumable_request();
6577 request["stream_resumable"] = serde_json::json!(false);
6578 let body = post_sse_raw(&app, request).await;
6579 assert!(!sse_field(&body, "data:").is_empty(), "it still streams");
6580 assert!(
6581 sse_field(&body, "id:").is_empty(),
6582 "an id promises a replay this stream cannot serve"
6583 );
6584 assert!(sse_field(&body, "retry:").is_empty());
6585 }
6586
6587 #[tokio::test]
6591 async fn the_polling_fallback_serves_exactly_what_the_stream_delivered() {
6592 let app = test_app();
6593 let body = post_sse_raw(&app, resumable_request()).await;
6594 let request_id = sse_field(&body, "id:")[0]
6595 .rsplit_once(':')
6596 .unwrap()
6597 .0
6598 .to_string();
6599 let streamed: Vec<String> = sse_field(&body, "data:")
6600 .iter()
6601 .map(|d| d.to_string())
6602 .collect();
6603
6604 let (status, polled) = get_json(
6605 &app,
6606 &format!("{}?from=0", frink_api::routes::v1_stream_poll(&request_id)),
6607 )
6608 .await;
6609 assert_eq!(status, StatusCode::OK);
6610 let events: Vec<String> = polled["events"]
6611 .as_array()
6612 .unwrap()
6613 .iter()
6614 .map(|e| e["data"].as_str().unwrap().to_string())
6615 .collect();
6616 assert_eq!(
6617 events, streamed,
6618 "the fallback must deliver the same answer, not a re-run of it"
6619 );
6620 assert_eq!(polled["request_id"], request_id);
6621 assert_eq!(
6622 polled["done"], false,
6623 "events were still being handed out, so the client must ask again"
6624 );
6625
6626 let next = polled["next_index"].as_u64().unwrap();
6629 let (_, drained) = get_json(
6630 &app,
6631 &format!(
6632 "{}?from={next}",
6633 frink_api::routes::v1_stream_poll(&request_id)
6634 ),
6635 )
6636 .await;
6637 assert_eq!(drained["done"], true);
6638 assert_eq!(drained["events"].as_array().unwrap().len(), 0);
6639 }
6640
6641 #[tokio::test]
6645 async fn a_resume_continues_after_the_last_event_id_rather_than_repeating() {
6646 let app = test_app();
6647 let body = post_sse_raw(&app, resumable_request()).await;
6648 let ids = sse_field(&body, "id:");
6649 let datas: Vec<String> = sse_field(&body, "data:")
6650 .iter()
6651 .map(|d| d.to_string())
6652 .collect();
6653 assert!(
6654 ids.len() >= 3,
6655 "need a few events to resume into the middle"
6656 );
6657 let request_id = ids[0].rsplit_once(':').unwrap().0.to_string();
6658
6659 let (status, resumed) = get_json_with_headers(
6660 &app,
6661 &format!("{}/poll", frink_api::routes::v1_stream(&request_id)),
6662 &[],
6663 )
6664 .await;
6665 assert_eq!(status, StatusCode::OK);
6666 assert_eq!(resumed["events"].as_array().unwrap().len(), datas.len());
6667
6668 let (_, tail) = get_json(
6670 &app,
6671 &format!("{}?from=2", frink_api::routes::v1_stream_poll(&request_id)),
6672 )
6673 .await;
6674 let tail_events: Vec<String> = tail["events"]
6675 .as_array()
6676 .unwrap()
6677 .iter()
6678 .map(|e| e["data"].as_str().unwrap().to_string())
6679 .collect();
6680 assert_eq!(tail_events, datas[2..].to_vec());
6681 }
6682
6683 #[tokio::test]
6686 async fn an_sse_reconnect_resumes_from_the_last_event_id() {
6687 use http_body_util::BodyExt;
6688 use tower::ServiceExt;
6689
6690 let app = test_app();
6691 let body = post_sse_raw(&app, resumable_request()).await;
6692 let ids = sse_field(&body, "id:");
6693 let datas: Vec<String> = sse_field(&body, "data:")
6694 .iter()
6695 .map(|d| d.to_string())
6696 .collect();
6697 let request_id = ids[0].rsplit_once(':').unwrap().0.to_string();
6698
6699 let response = app
6700 .clone()
6701 .oneshot(
6702 axum::http::Request::builder()
6703 .method("GET")
6704 .uri(frink_api::routes::v1_stream(&request_id))
6705 .header("last-event-id", format!("{request_id}:0"))
6706 .body(axum::body::Body::empty())
6707 .unwrap(),
6708 )
6709 .await
6710 .unwrap();
6711 assert_eq!(response.status(), StatusCode::OK);
6712 assert_eq!(
6713 response
6714 .headers()
6715 .get("x-accel-buffering")
6716 .and_then(|v| v.to_str().ok()),
6717 Some("no"),
6718 "the reconnect needs the same anti-buffering header as the stream"
6719 );
6720 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6721 let resumed = String::from_utf8(bytes.to_vec()).unwrap();
6722 assert_eq!(
6723 sse_field(&resumed, "data:")
6724 .iter()
6725 .map(|d| d.to_string())
6726 .collect::<Vec<_>>(),
6727 datas[1..].to_vec()
6728 );
6729 assert_eq!(sse_field(&resumed, "id:")[0], format!("{request_id}:1"));
6730 }
6731
6732 #[tokio::test]
6736 async fn a_last_event_id_from_another_stream_is_refused() {
6737 let app = test_app();
6738 let body = post_sse_raw(&app, resumable_request()).await;
6739 let request_id = sse_field(&body, "id:")[0]
6740 .rsplit_once(':')
6741 .unwrap()
6742 .0
6743 .to_string();
6744
6745 let (status, err) = get_json_with_headers(
6746 &app,
6747 &frink_api::routes::v1_stream(&request_id),
6748 &[("last-event-id", "chatcmpl-someone-else:3")],
6749 )
6750 .await;
6751 assert_eq!(status, StatusCode::BAD_REQUEST);
6752 assert_eq!(err["error"]["code"], "bad_last_event_id");
6753 }
6754
6755 #[tokio::test]
6759 async fn resuming_a_stream_that_was_never_resumable_is_a_404_that_says_why() {
6760 let app = test_app();
6761 let mut request = resumable_request();
6762 request["stream_resumable"] = serde_json::json!(false);
6763 let body = post_sse_raw(&app, request).await;
6764 let request_id = body
6765 .lines()
6766 .find_map(|l| l.strip_prefix("data: "))
6767 .and_then(|d| serde_json::from_str::<serde_json::Value>(d).ok())
6768 .and_then(|v| v["request_id"].as_str().map(str::to_string))
6769 .unwrap();
6770
6771 let (status, err) = get_json(&app, &frink_api::routes::v1_stream_poll(&request_id)).await;
6772 assert_eq!(status, StatusCode::NOT_FOUND);
6773 assert_eq!(err["error"]["code"], "stream_not_found");
6774 assert!(err["error"]["message"]
6775 .as_str()
6776 .unwrap()
6777 .contains("stream_resumable"));
6778 }
6779
6780 #[test]
6784 fn the_axum_stream_patterns_match_the_published_templates() {
6785 assert_eq!(
6786 axum_path(frink_api::routes::V1_STREAM),
6787 "/v1/stream/:request_id"
6788 );
6789 assert_eq!(
6790 axum_path(frink_api::routes::V1_STREAM_POLL),
6791 "/v1/stream/:request_id/poll"
6792 );
6793 assert_eq!(
6794 frink_api::routes::v1_stream("abc"),
6795 axum_path(frink_api::routes::V1_STREAM).replace(":request_id", "abc")
6796 );
6797 }
6798
6799 #[test]
6809 fn no_published_template_reaches_the_router_with_its_braces() {
6810 for template in [
6811 frink_api::routes::V1_STREAM,
6812 frink_api::routes::V1_STREAM_POLL,
6813 frink_api::routes::V1_RESPONSE,
6814 frink_api::routes::V1_RESPONSE_CANCEL,
6815 frink_api::routes::ADMIN_TASK_CANCEL,
6816 ] {
6817 assert!(
6818 template.contains('{'),
6819 "{template} is in the template list but has no placeholder"
6820 );
6821 let mounted = axum_path(template);
6822 assert!(
6823 !mounted.contains('{') && !mounted.contains('}'),
6824 "{template} would be mounted as {mounted}, whose braces axum reads as a literal segment"
6825 );
6826 assert!(
6827 mounted.contains(':'),
6828 "{template} lost its placeholder entirely and would match one path only"
6829 );
6830 }
6831 }
6832
6833 #[tokio::test]
6841 async fn an_unknown_response_id_gets_the_handler_not_a_bare_404() {
6842 let app = test_app();
6843 let (status, body) = get_json(&app, "/v1/responses/resp_nonexistent").await;
6844 assert_eq!(status, StatusCode::NOT_FOUND);
6845 assert!(
6846 !body.is_null(),
6847 "empty body means axum never matched the route, so the id was read as a literal segment"
6848 );
6849 }
6850
6851 #[tokio::test]
6854 async fn the_task_list_starts_empty_rather_than_absent() {
6855 let app = test_app();
6856 let (status, body) = get_json(&app, frink_api::routes::ADMIN_TASKS).await;
6857 assert_eq!(status, StatusCode::OK);
6858 assert_eq!(body["tasks"].as_array().unwrap().len(), 0);
6859 }
6860
6861 #[tokio::test]
6870 async fn the_slots_route_is_registered_and_refuses_by_naming_slot_save_path() {
6871 assert!(
6872 std::env::var("FRINK_SLOT_SAVE_PATH").is_err(),
6873 "this test asserts the unconfigured behaviour"
6874 );
6875 let app = test_app();
6876 let (status, body) = post_json_uri(
6877 &app,
6878 &format!("{}?action=save", frink_api::routes::slots_id(0)),
6879 serde_json::json!({"filename": "sys.fslot", "prompt": "hi"}),
6880 )
6881 .await;
6882 assert_eq!(status, StatusCode::NOT_IMPLEMENTED);
6883 assert!(
6884 body["error"]["message"]
6885 .as_str()
6886 .unwrap()
6887 .contains("--slot-save-path"),
6888 "{body}"
6889 );
6890 }
6891
6892 pub(crate) async fn post_json_uri(
6893 app: &Router,
6894 uri: &str,
6895 body: serde_json::Value,
6896 ) -> (StatusCode, serde_json::Value) {
6897 use http_body_util::BodyExt;
6898 use tower::ServiceExt;
6899
6900 let response = app
6901 .clone()
6902 .oneshot(
6903 axum::http::Request::builder()
6904 .method("POST")
6905 .uri(uri)
6906 .header("content-type", "application/json")
6907 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
6908 .unwrap(),
6909 )
6910 .await
6911 .unwrap();
6912 let status = response.status();
6913 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6914 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
6915 (status, json)
6916 }
6917
6918 pub(crate) async fn get_json_uri(app: &Router, uri: &str) -> (StatusCode, serde_json::Value) {
6921 use http_body_util::BodyExt;
6922 use tower::ServiceExt;
6923
6924 let response = app
6925 .clone()
6926 .oneshot(
6927 axum::http::Request::builder()
6928 .method("GET")
6929 .uri(uri)
6930 .body(axum::body::Body::empty())
6931 .unwrap(),
6932 )
6933 .await
6934 .unwrap();
6935 let status = response.status();
6936 let bytes = response.into_body().collect().await.unwrap().to_bytes();
6937 let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::json!({}));
6938 (status, json)
6939 }
6940
6941 async fn post_json(app: &Router, body: serde_json::Value) -> serde_json::Value {
6942 post_json_uri(app, "/v1/chat/completions", body).await.1
6943 }
6944
6945 #[tokio::test]
6954 async fn stats_says_what_the_engine_is_using_and_which_quantity_that_is() {
6955 let app = test_app();
6956 let (status, body) = get_json(&app, frink_api::routes::V1_STATS).await;
6957 assert_eq!(status, StatusCode::OK);
6958
6959 let memory = &body["memory"];
6960 if memory.is_null() {
6961 return;
6964 }
6965 assert!(
6966 memory["bytes"].as_u64().is_some_and(|b| b > 0),
6967 "a read that produced a zero is a broken read, not an idle \
6968 engine: {memory}"
6969 );
6970 assert!(
6971 ["pss", "rss"].contains(&memory["kind"].as_str().unwrap_or("")),
6972 "the quantity must travel with the number: {memory}"
6973 );
6974 }
6975
6976 #[tokio::test]
6982 async fn stats_reports_a_pool_it_does_not_have_as_absent_and_not_as_zero() {
6983 let app = test_app();
6984 let (status, body) = get_json(&app, frink_api::routes::V1_STATS).await;
6985 assert_eq!(status, StatusCode::OK);
6986 for pool in ["kv_pages", "window_slots", "state_slots"] {
6987 assert!(
6988 body["pools"][pool].is_null(),
6989 "{pool} must be null rather than a zero row: {}",
6990 body["pools"]
6991 );
6992 }
6993 }
6994
6995 #[tokio::test]
7003 async fn a_messages_response_states_the_id_that_v1_cancel_takes() {
7004 use http_body_util::BodyExt;
7005 use tower::ServiceExt;
7006
7007 let app = test_app();
7008 let send = |body: serde_json::Value| {
7009 let app = app.clone();
7010 async move {
7011 app.oneshot(
7012 axum::http::Request::builder()
7013 .method("POST")
7014 .uri(frink_api::routes::V1_MESSAGES)
7015 .header("content-type", "application/json")
7016 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
7017 .unwrap(),
7018 )
7019 .await
7020 .unwrap()
7021 }
7022 };
7023
7024 let ok = send(serde_json::json!({
7025 "model": "test",
7026 "max_tokens": 1,
7027 "messages": [{"role": "user", "content": "hi"}],
7028 }))
7029 .await;
7030 assert_eq!(ok.status(), StatusCode::OK);
7031 let id = ok
7032 .headers()
7033 .get("request-id")
7034 .expect("a served message names its id")
7035 .to_str()
7036 .unwrap()
7037 .to_string();
7038 assert!(!id.is_empty());
7039
7040 let bad = send(serde_json::json!({"model": "test"})).await;
7043 assert!(bad.status().is_client_error());
7044 let other = bad.headers().get("request-id").expect("errors too");
7045 assert_ne!(other.to_str().unwrap(), id);
7046 let _ = bad.into_body().collect().await.unwrap();
7047 }
7048
7049 #[tokio::test]
7057 async fn a_request_that_arrives_mid_rebuild_is_refused_and_admitted_again_after() {
7058 let state = Arc::new(test_state(
7059 test_model_full_byte_vocab(),
7060 ResponseCache::new(1000, Duration::from_secs(3600)),
7061 ));
7062 let app = test_app_with_state(Arc::clone(&state));
7063 let body = serde_json::json!({
7064 "model": "test",
7065 "messages": [{"role": "user", "content": "hi"}],
7066 "max_tokens": 1,
7067 });
7068
7069 state
7070 .maintenance
7071 .lock()
7072 .unwrap()
7073 .begin_rebuild()
7074 .expect("a fresh server is serving, so the rebuild starts");
7075 let (status, refused) = post_json_uri(&app, "/v1/chat/completions", body.clone()).await;
7076 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
7077 assert_eq!(refused["error"]["type"], "cache_rebuilding");
7078
7079 state.maintenance.lock().unwrap().finish_rebuild(true);
7080 let (status, _) = post_json_uri(&app, "/v1/chat/completions", body).await;
7081 assert_eq!(
7082 status,
7083 StatusCode::OK,
7084 "the gate reopens; a rebuild is not a latch"
7085 );
7086 }
7087
7088 #[tokio::test]
7093 async fn cancelling_an_id_that_is_not_generating_is_a_404_that_says_so() {
7094 let app = test_app();
7095 let (status, body) = post_json_uri(
7096 &app,
7097 frink_api::routes::V1_CANCEL,
7098 serde_json::json!({ "request_id": "chatcmpl-never-issued" }),
7099 )
7100 .await;
7101 assert_eq!(status, StatusCode::NOT_FOUND);
7102 assert_eq!(body["cancelled"], serde_json::json!(false));
7103 assert_eq!(body["request_id"], "chatcmpl-never-issued");
7104 assert!(
7105 body["detail"].as_str().is_some_and(|d| !d.is_empty()),
7106 "the verdict must carry a human reason: {body}"
7107 );
7108 }
7109
7110 #[tokio::test]
7114 async fn cancelling_a_live_generation_signals_its_token_and_answers_200() {
7115 let state = Arc::new(test_state(
7116 test_model_full_byte_vocab(),
7117 ResponseCache::new(1000, Duration::from_secs(3600)),
7118 ));
7119 let app = test_app_with_state(Arc::clone(&state));
7120 let (token, _guard) = state.cancels.register("chatcmpl-live");
7121
7122 let (status, before) = get_json(&app, frink_api::routes::ADMIN_STATS).await;
7123 assert_eq!(status, StatusCode::OK);
7124 assert_eq!(before["generating_now"], serde_json::json!(1));
7125
7126 let (status, body) = post_json_uri(
7127 &app,
7128 frink_api::routes::V1_CANCEL,
7129 serde_json::json!({ "request_id": "chatcmpl-live" }),
7130 )
7131 .await;
7132 assert_eq!(status, StatusCode::OK);
7133 assert_eq!(body["cancelled"], serde_json::json!(true));
7134 assert!(
7135 token.is_cancelled(),
7136 "the endpoint answered ok without setting the flag the decode loop reads"
7137 );
7138 }
7139
7140 #[tokio::test]
7141 async fn tokenize_detokenize_roundtrip_and_embeddings_mean() {
7142 let app = test_app();
7143 let (status, tok) =
7144 post_json_uri(&app, "/v1/tokenize", serde_json::json!({ "prompt": "Hi" })).await;
7145 assert_eq!(status, StatusCode::OK);
7146 let tokens = tok["tokens"].as_array().unwrap();
7147 assert_eq!(tok["count"], tokens.len());
7148 assert!(!tokens.is_empty());
7149
7150 let (status, detok) = post_json_uri(
7151 &app,
7152 "/v1/detokenize",
7153 serde_json::json!({ "tokens": tokens }),
7154 )
7155 .await;
7156 assert_eq!(status, StatusCode::OK);
7157 assert_eq!(detok["text"], "Hi");
7158
7159 let (status, emb) = post_json_uri(
7160 &app,
7161 "/v1/embeddings",
7162 serde_json::json!({
7163 "input": "Hi",
7164 "embedding_type": "mean"
7165 }),
7166 )
7167 .await;
7168 assert_eq!(status, StatusCode::OK);
7169 let vec = emb["data"][0]["embedding"].as_array().unwrap();
7170 assert!(!vec.is_empty());
7171 assert!(vec.iter().all(|v| v.as_f64().is_some()));
7172 }
7173
7174 #[tokio::test]
7180 async fn the_decoder_path_still_refuses_a_pooling_it_cannot_mean() {
7181 let app = test_app();
7182 let (status, body) = post_json_uri(
7183 &app,
7184 "/v1/embeddings",
7185 serde_json::json!({ "input": "Hi", "embedding_type": "cls" }),
7186 )
7187 .await;
7188 assert_eq!(status, StatusCode::BAD_REQUEST);
7189 let msg = body["error"]["message"].as_str().unwrap();
7190 assert!(msg.contains("mean") && msg.contains("last"), "{msg}");
7191 }
7192
7193 #[tokio::test]
7198 #[ignore = "needs models/bge-small-en-v1.5-q8_0.gguf"]
7199 async fn a_real_embedding_model_serves_v1_embeddings() {
7200 let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
7201 .join("../../models/bge-small-en-v1.5-q8_0.gguf");
7202 if !path.exists() {
7203 eprintln!("SKIP: {} not present", path.display());
7204 return;
7205 }
7206 let encoder = frink_models::EmbeddingModel::from_gguf_path(&path).expect("load bge");
7207 let mut state = test_state(
7208 test_model_full_byte_vocab(),
7209 ResponseCache::new(1000, Duration::from_secs(3600)),
7210 );
7211 state.embedding = Some(Arc::new(encoder));
7212 let app = test_app_with_state(Arc::new(state));
7213
7214 let (status, body) = post_json_uri(
7215 &app,
7216 "/v1/embeddings",
7217 serde_json::json!({ "input": ["Hello world", "a second input"] }),
7218 )
7219 .await;
7220 assert_eq!(status, StatusCode::OK, "{body}");
7221 assert_eq!(body["model"], "bge-small-en-v1.5");
7222 let data = body["data"].as_array().unwrap();
7223 assert_eq!(data.len(), 2);
7224 for (i, row) in data.iter().enumerate() {
7225 assert_eq!(row["index"], i);
7226 let v: Vec<f64> = row["embedding"]
7227 .as_array()
7228 .unwrap()
7229 .iter()
7230 .map(|x| x.as_f64().unwrap())
7231 .collect();
7232 assert_eq!(v.len(), 384, "the encoder\'s width, not the decoder\'s");
7233 let norm = v.iter().map(|x| x * x).sum::<f64>().sqrt();
7234 assert!((norm - 1.0).abs() < 1e-4, "not L2-normalized: {norm}");
7235 }
7236 assert!(body["usage"]["prompt_tokens"].as_u64().unwrap() >= 4 + 2);
7239
7240 let (status, mean) = post_json_uri(
7244 &app,
7245 "/v1/embeddings",
7246 serde_json::json!({ "input": "Hello world", "embedding_type": "mean" }),
7247 )
7248 .await;
7249 assert_eq!(status, StatusCode::OK);
7250 assert_ne!(mean["data"][0]["embedding"], data[0]["embedding"]);
7251 }
7252
7253 #[tokio::test]
7265 #[ignore = "needs models/bge-small-en-v1.5-q8_0.gguf"]
7266 async fn an_encoder_can_be_the_loaded_model() {
7267 let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
7268 .join("../../models/bge-small-en-v1.5-q8_0.gguf");
7269 if !path.exists() {
7270 eprintln!("SKIP: {} not present", path.display());
7271 return;
7272 }
7273
7274 let loaded = model::load_from_path(path.to_str().unwrap()).expect("load bge as the model");
7278 assert!(
7279 matches!(loaded, model::LoadedModel::Encoder(_)),
7280 "an encoder-only GGUF reached a decoder loader"
7281 );
7282 let (loaded, batcher, ceiling) = activate_loaded_model(loaded, true, None, None);
7283 assert!(
7284 matches!(loaded, Loaded::Encoder(_)),
7285 "the encoder did not stay an encoder through activation"
7286 );
7287 assert!(
7288 batcher.is_none() && ceiling.is_none(),
7289 "an encoder was given a decode batcher or a KV ceiling it has no use for"
7290 );
7291
7292 let state = test_state(
7293 test_model_full_byte_vocab(),
7294 ResponseCache::new(1000, Duration::from_secs(3600)),
7295 );
7296 state.swap_active(Some(Arc::new(ActiveModel {
7297 id: None,
7298 loaded,
7299 batcher,
7300 ceiling,
7301 checkpoint_path: None,
7302 })));
7303 let app = test_app_with_state(Arc::new(state));
7304
7305 let (status, body) = post_json_uri(
7307 &app,
7308 "/v1/embeddings",
7309 serde_json::json!({ "input": "Hello world" }),
7310 )
7311 .await;
7312 assert_eq!(status, StatusCode::OK, "{body}");
7313 assert_eq!(body["model"], "bge-small-en-v1.5");
7314 let v = body["data"][0]["embedding"].as_array().unwrap();
7315 assert_eq!(v.len(), 384, "the encoder's width, not the decoder's");
7316
7317 let (status, body) = post_json_uri(
7319 &app,
7320 "/v1/chat/completions",
7321 serde_json::json!({
7322 "model": "bge-small-en-v1.5",
7323 "messages": [{"role": "user", "content": "hi"}],
7324 }),
7325 )
7326 .await;
7327 assert_eq!(status, StatusCode::NOT_IMPLEMENTED, "{body}");
7328 let msg = body["error"]["message"].as_str().unwrap();
7329 for fact in [
7330 "bge-small-en-v1.5",
7331 "bert",
7332 "embedding model",
7333 "/v1/embeddings",
7334 ] {
7335 assert!(msg.contains(fact), "the refusal does not say {fact}: {msg}");
7336 }
7337
7338 let (status, models) = get_json(&app, frink_api::routes::V1_MODELS).await;
7340 assert_eq!(status, StatusCode::OK);
7341 let entry = &models["data"][0];
7342 assert_eq!(entry["id"], "bge-small-en-v1.5");
7343 assert_eq!(entry["frink_model_kind"], "embedding");
7344 assert_eq!(entry["frink_tokenizer"], "gguf-wordpiece");
7345 assert_eq!(entry["frink_n_embd"], 384);
7346 assert_eq!(entry["frink_pooling"], "CLS");
7347 assert_eq!(
7348 entry["frink_endpoints"],
7349 serde_json::json!(["/v1/embeddings"])
7350 );
7351 assert!(entry.get("supported_reasoning_efforts").is_none());
7354
7355 let (status, health) = get_json(&app, frink_api::routes::HEALTH).await;
7357 assert_eq!(status, StatusCode::OK, "an encoder is a loaded model");
7358 assert_eq!(health["model"]["id"], "bge-small-en-v1.5");
7359 assert_eq!(health["model"]["synthetic_weights"], false);
7360 let weights = health["capabilities"]
7361 .as_array()
7362 .unwrap()
7363 .iter()
7364 .find(|c| c["id"] == frink_api::health::capability::REAL_WEIGHTS)
7365 .expect("a real-weights capability row");
7366 let detail = weights["detail"].as_str().unwrap_or_default();
7367 assert!(detail.contains("ENCODER"), "{detail}");
7368 let (status, body) = post_json_uri(
7376 &app,
7377 frink_api::routes::V1_TOKENIZE,
7378 serde_json::json!({ "content": "hello world" }),
7379 )
7380 .await;
7381 assert_eq!(
7382 status,
7383 StatusCode::OK,
7384 "an encoder has a real tokenizer: {body}"
7385 );
7386 let tokens = body["tokens"].as_array().expect("tokens array").clone();
7387 assert!(!tokens.is_empty(), "WordPiece produced nothing: {body}");
7388
7389 let (status, body) = post_json_uri(
7390 &app,
7391 frink_api::routes::V1_DETOKENIZE,
7392 serde_json::json!({ "tokens": tokens }),
7393 )
7394 .await;
7395 assert_eq!(status, StatusCode::OK, "{body}");
7396 let round_tripped = body["content"].as_str().expect("content").to_string();
7397 assert!(
7398 round_tripped.contains("hello") && round_tripped.contains("world"),
7399 "the ids did not decode back through the encoder's own vocabulary: {round_tripped}"
7400 );
7401
7402 let (status, _) = post_json_uri(
7405 &app,
7406 "/v1/completions",
7407 serde_json::json!({ "model": "m", "prompt": "hi", "max_tokens": 1 }),
7408 )
7409 .await;
7410 assert_eq!(
7411 status,
7412 StatusCode::NOT_IMPLEMENTED,
7413 "tokenizing an encoder must not have opened a path to generating with one"
7414 );
7415 }
7416
7417 #[tokio::test]
7422 async fn metrics_exposes_expert_store_counters_when_streaming_is_active() {
7423 use http_body_util::BodyExt;
7424 use tower::ServiceExt;
7425
7426 let fixture = concat!(
7427 "../frink-models/tests/fixtures/",
7428 "frink_real_moe_test.gguf"
7429 );
7430 let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join(fixture);
7431 let decoder = Decoder::from_gguf_with_expert_cache(
7432 &fixture,
7433 frink_models::config::test_moe_fixture(),
7434 Some(1024 * 1024),
7435 )
7436 .expect("MoE fixture must load store-backed");
7437
7438 let mut caches: Vec<frink_core::cache::KvCache> = decoder.config.new_kv_caches();
7442 decoder.forward_token(1, 0, &mut caches);
7443
7444 let model = Model::Gguf(GgufModel {
7445 decoder: Arc::new(decoder),
7446 tokenizer: Arc::new(ServerTokenizer::Byte),
7447 stop_tokens: StopTokens::default(),
7448 bos_id: None,
7449 is_synthetic: false,
7450 chat_template: chat_template::PromptTemplate::plain(),
7451 });
7452 let state = Arc::new(test_state(
7453 model,
7454 ResponseCache::new(16, Duration::from_secs(60)),
7455 ));
7456 let app = Router::new()
7457 .route("/metrics", axum::routing::get(metrics))
7458 .route("/v1/chat/completions", post(chat_completions))
7459 .with_state(state);
7460
7461 let fetch_metrics = |app: Router| async move {
7462 let resp = app
7463 .oneshot(
7464 axum::http::Request::builder()
7465 .method("GET")
7466 .uri("/metrics")
7467 .body(axum::body::Body::empty())
7468 .unwrap(),
7469 )
7470 .await
7471 .unwrap();
7472 let bytes = resp.into_body().collect().await.unwrap().to_bytes();
7473 String::from_utf8(bytes.to_vec()).unwrap()
7474 };
7475
7476 let after = fetch_metrics(app.clone()).await;
7477 assert!(
7478 after.contains("frink_expert_cache_misses_total"),
7479 "streaming model must expose expert-cache metrics: {after}"
7480 );
7481 let misses: u64 = after
7482 .lines()
7483 .find(|l| l.starts_with("frink_expert_cache_misses_total"))
7484 .and_then(|l| l.split_whitespace().nth(1))
7485 .and_then(|v| v.parse().ok())
7486 .expect("misses metric line must parse");
7487 assert!(
7488 misses > 0,
7489 "decode must have read experts through the store: {after}"
7490 );
7491 }
7492
7493 fn weather_tool() -> serde_json::Value {
7494 serde_json::json!({
7495 "type": "function",
7496 "function": {
7497 "name": "get_weather",
7498 "description": "Get the current weather for a location.",
7499 "parameters": {
7500 "type": "object",
7501 "properties": {"location": {"type": "string"}},
7502 "required": ["location"]
7503 }
7504 }
7505 })
7506 }
7507
7508 fn weather_tool_def() -> ToolDef {
7509 ToolDef {
7510 kind: "function".to_string(),
7511 function: ToolFunctionDef {
7512 name: "get_weather".to_string(),
7513 description: Some("Get the current weather for a location.".to_string()),
7514 parameters: Some(serde_json::json!({
7515 "type": "object",
7516 "properties": {"location": {"type": "string"}},
7517 "required": ["location"]
7518 })),
7519 },
7520 }
7521 }
7522
7523 #[test]
7524 fn tool_preamble_mentions_every_tool_name_and_description() {
7525 let preamble = tool_preamble(&[weather_tool_def()]);
7526 assert!(preamble.contains("get_weather"));
7527 assert!(preamble.contains("Get the current weather for a location."));
7528 assert!(preamble.contains("<tool_call>"));
7529 assert!(preamble.contains("</tool_call>"));
7530 }
7531
7532 #[test]
7533 fn a_real_marker_becomes_a_structured_tool_call() {
7534 let text = "sure, let me check.<tool_call>{\"name\": \"get_weather\", \"arguments\": {\"location\": \"Paris\"}}</tool_call>";
7535 let (message, finish) = build_response_message(
7536 text.to_string(),
7537 &[weather_tool_def()],
7538 output::OutputPosture::for_model("test-model"),
7539 "stop",
7540 );
7541 assert_eq!(finish, "tool_calls");
7542 let calls = message.tool_calls.expect("must carry a tool call");
7543 assert_eq!(calls[0].function.name, "get_weather");
7544 let parsed: serde_json::Value = serde_json::from_str(&calls[0].function.arguments).unwrap();
7545 assert_eq!(parsed["location"], "Paris");
7546 }
7547
7548 #[test]
7549 fn a_plain_answer_is_not_promoted_to_a_tool_call() {
7550 let (message, finish) = build_response_message(
7551 "just an answer".to_string(),
7552 &[weather_tool_def()],
7553 output::OutputPosture::for_model("test-model"),
7554 "stop",
7555 );
7556 assert_eq!(finish, "stop");
7557 assert!(message.tool_calls.is_none());
7558 assert_eq!(message.content.as_deref(), Some("just an answer"));
7559 }
7560
7561 #[test]
7564 fn a_malformed_payload_is_not_a_tool_call() {
7565 let (message, finish) = build_response_message(
7566 "<tool_call>not valid json at all</tool_call>".to_string(),
7567 &[weather_tool_def()],
7568 output::OutputPosture::for_model("test-model"),
7569 "stop",
7570 );
7571 assert_eq!(finish, "stop");
7572 assert!(message.tool_calls.is_none());
7573 }
7574
7575 #[test]
7578 fn a_tool_that_was_never_offered_is_not_returned() {
7579 let (message, finish) = build_response_message(
7580 "<tool_call>{\"name\": \"ping\", \"arguments\": {}}</tool_call>".to_string(),
7581 &[weather_tool_def()],
7582 output::OutputPosture::for_model("test-model"),
7583 "stop",
7584 );
7585 assert_eq!(finish, "stop");
7586 assert!(message.tool_calls.is_none());
7587 }
7588
7589 #[test]
7591 fn marker_text_with_no_tools_offered_stays_content() {
7592 let (message, finish) = build_response_message(
7593 "<tool_call>{\"name\": \"get_weather\", \"arguments\": {}}</tool_call>".to_string(),
7594 &[],
7595 output::OutputPosture::for_model("test-model"),
7596 "stop",
7597 );
7598 assert_eq!(finish, "stop");
7599 assert!(message.tool_calls.is_none());
7600 assert!(message.content.is_some());
7601 }
7602
7603 #[test]
7607 fn a_streamed_call_opens_then_delivers_its_arguments_in_pieces() {
7608 let opened = std::cell::Cell::new(0usize);
7609 let mut parser = crate::policy::parser::ToolCallParser::new(
7610 crate::policy::parser::ToolCallFormat::Qwen3Coder,
7611 vec![
7612 crate::policy::parser::tool_call::ToolSchema::with_parameters(
7613 "write_file",
7614 serde_json::json!({"type": "object", "properties": {
7615 "path": {"type": "string"},
7616 "contents": {"type": "string"}
7617 }}),
7618 ),
7619 ],
7620 );
7621 let wire = "<tool_call><function=write_file>\
7622 <parameter=path>\n/tmp/x\n</parameter>\
7623 <parameter=contents>\nhello world\n</parameter>\
7624 </function></tool_call>";
7625
7626 let mut deltas = Vec::new();
7627 let mut text = String::new();
7628 for piece in wire.as_bytes().chunks(7) {
7629 let chunk = String::from_utf8_lossy(piece).into_owned();
7630 let (more_text, more) = tool_call_deltas(parser.push(&chunk), &opened);
7631 text.push_str(&more_text);
7632 deltas.extend(more);
7633 }
7634 let (more_text, more) = tool_call_deltas(parser.finish(), &opened);
7635 text.push_str(&more_text);
7636 deltas.extend(more);
7637
7638 assert_eq!(opened.get(), 1, "one call opened");
7639 assert!(text.is_empty(), "the markers are not content: {text:?}");
7640
7641 let first = &deltas[0];
7642 assert_eq!(first.index, 0);
7643 assert_eq!(first.id.as_deref(), Some("call_0"));
7644 assert_eq!(first.kind, Some("function"));
7645 assert_eq!(first.function.name.as_deref(), Some("write_file"));
7646
7647 let joined: String = deltas
7650 .iter()
7651 .filter_map(|d| d.function.arguments.clone())
7652 .collect();
7653 let parsed: serde_json::Value =
7654 serde_json::from_str(&joined).expect("the fragments concatenate to valid JSON");
7655 assert_eq!(parsed["path"], serde_json::json!("/tmp/x"));
7656 assert_eq!(parsed["contents"], serde_json::json!("hello world"));
7657 assert!(
7658 deltas.len() >= 3,
7659 "the arguments arrived in pieces, not whole: {}",
7660 deltas.len()
7661 );
7662 assert!(
7663 deltas[1..].iter().all(|d| d.function.name.is_none()),
7664 "only the opening delta carries identity"
7665 );
7666 }
7667
7668 #[test]
7670 fn text_around_a_streamed_call_is_still_content() {
7671 let opened = std::cell::Cell::new(0usize);
7672 let mut parser = crate::policy::parser::ToolCallParser::new(
7673 crate::policy::parser::ToolCallFormat::Qwen25,
7674 vec![crate::policy::parser::tool_call::ToolSchema::new(
7675 "get_weather",
7676 )],
7677 );
7678 let wire = "let me check. <tool_call>{\"name\": \"get_weather\", \
7679 \"arguments\": {}}</tool_call> done";
7680 let mut text = String::new();
7681 for piece in wire.as_bytes().chunks(5) {
7682 let chunk = String::from_utf8_lossy(piece).into_owned();
7683 let (more, _) = tool_call_deltas(parser.push(&chunk), &opened);
7684 text.push_str(&more);
7685 }
7686 let (more, _) = tool_call_deltas(parser.finish(), &opened);
7687 text.push_str(&more);
7688
7689 assert_eq!(opened.get(), 1);
7690 assert!(text.starts_with("let me check. "), "{text:?}");
7691 assert!(text.ends_with(" done"), "{text:?}");
7692 assert!(!text.contains("<tool_call>"), "markers leaked: {text:?}");
7693 }
7694
7695 #[test]
7698 fn a_reasoning_block_is_split_out_of_the_answer() {
7699 let (message, finish) = build_response_message(
7700 "<think>weighing it up</think>The answer is 4.".to_string(),
7701 &[],
7702 output::OutputPosture::for_model("Qwen3-8B"),
7703 "stop",
7704 );
7705 assert_eq!(finish, "stop");
7706 assert_eq!(message.content.as_deref(), Some("The answer is 4."));
7707 assert_eq!(message.reasoning_content.as_deref(), Some("weighing it up"));
7708 }
7709
7710 #[test]
7713 fn a_non_reasoning_model_keeps_a_literal_marker_in_its_answer() {
7714 let (message, _) = build_response_message(
7715 "Use the <think> tag like this.".to_string(),
7716 &[],
7717 output::OutputPosture::for_model("llama-3.1-8b"),
7718 "stop",
7719 );
7720 assert_eq!(
7721 message.content.as_deref(),
7722 Some("Use the <think> tag like this.")
7723 );
7724 assert!(message.reasoning_content.is_none());
7725 }
7726
7727 #[tokio::test]
7735 async fn a_request_with_no_tools_or_session_behaves_exactly_as_before() {
7736 let app = test_app();
7737 let body = serde_json::json!({
7738 "model": "m",
7739 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7740 "max_tokens": 4,
7741 "temperature": 0,
7742 });
7743 let resp = post_json(&app, body).await;
7744 let message = &resp["choices"][0]["message"];
7745 assert!(message["content"].is_string());
7746 assert!(message.get("tool_calls").is_none());
7747 assert_eq!(resp["choices"][0]["finish_reason"], "length");
7748 assert_eq!(resp["usage"]["completion_tokens"], 4);
7749 assert_eq!(
7750 resp["usage"]["total_tokens"],
7751 resp["usage"]["prompt_tokens"].as_u64().unwrap() + 4
7752 );
7753 }
7754
7755 pub(crate) async fn get_json(app: &Router, uri: &str) -> (StatusCode, serde_json::Value) {
7756 use http_body_util::BodyExt;
7757 use tower::ServiceExt;
7758
7759 let response = app
7760 .clone()
7761 .oneshot(
7762 axum::http::Request::builder()
7763 .method("GET")
7764 .uri(uri)
7765 .body(axum::body::Body::empty())
7766 .unwrap(),
7767 )
7768 .await
7769 .unwrap();
7770 let status = response.status();
7771 let bytes = response.into_body().collect().await.unwrap().to_bytes();
7772 (status, serde_json::from_slice(&bytes).unwrap())
7773 }
7774
7775 #[tokio::test]
7776 async fn health_answers_a_capability_handshake_not_a_boolean() {
7777 let app = test_app();
7778 let (status, body) = get_json(&app, frink_api::routes::HEALTH).await;
7779 assert_eq!(status, StatusCode::OK);
7780
7781 let health: frink_api::HealthResponse = serde_json::from_value(body).unwrap();
7782 assert_eq!(health.state, frink_api::HealthState::Ready);
7783 assert!(health.pid > 0);
7784 assert!(health.server_time_unix_ms > 0);
7785 assert_eq!(health.last_request_age_seconds, None);
7788
7789 for id in [
7792 frink_api::health::capability::CPU,
7793 frink_api::health::capability::METAL,
7794 frink_api::health::capability::CUDA,
7795 frink_api::health::capability::REAL_WEIGHTS,
7796 frink_api::health::capability::CONTINUOUS_BATCHING,
7797 ] {
7798 let cap = health
7799 .capability(id)
7800 .unwrap_or_else(|| panic!("{id} missing"));
7801 assert!(!cap.reason.is_empty(), "{cap:?}");
7802 assert!(!cap.detail.is_empty(), "{cap:?}");
7803 }
7804 let weights = health
7808 .capability(frink_api::health::capability::REAL_WEIGHTS)
7809 .unwrap();
7810 assert!(!weights.available);
7811 assert_eq!(weights.reason, frink_api::health::reason::MODEL_NOT_LOADED);
7812 assert!(health.model.as_ref().unwrap().synthetic_weights);
7813 }
7814
7815 #[tokio::test]
7816 async fn health_vouches_for_liveness_after_a_request_has_been_served() {
7817 let app = test_app();
7818 let _ = post_json(
7819 &app,
7820 serde_json::json!({
7821 "model": "m",
7822 "messages": [{"role": "user", "content": "\u{1}"}],
7823 "max_tokens": 1,
7824 "temperature": 0,
7825 }),
7826 )
7827 .await;
7828 let (_status, body) = get_json(&app, frink_api::routes::HEALTH).await;
7829 let health: frink_api::HealthResponse = serde_json::from_value(body).unwrap();
7830 let age = health
7831 .last_request_age_seconds
7832 .expect("a served request is evidence of liveness");
7833 assert!((0.0..5.0).contains(&age), "implausible age {age}");
7834 }
7835
7836 async fn post_sse_chunks(app: &Router, body: serde_json::Value) -> Vec<serde_json::Value> {
7838 use http_body_util::BodyExt;
7839 use tower::ServiceExt;
7840
7841 let response = app
7842 .clone()
7843 .oneshot(
7844 axum::http::Request::builder()
7845 .method("POST")
7846 .uri("/v1/chat/completions")
7847 .header("content-type", "application/json")
7848 .body(axum::body::Body::from(serde_json::to_vec(&body).unwrap()))
7849 .unwrap(),
7850 )
7851 .await
7852 .unwrap();
7853 let bytes = response.into_body().collect().await.unwrap().to_bytes();
7854 String::from_utf8(bytes.to_vec())
7855 .unwrap()
7856 .lines()
7857 .filter_map(|line| line.strip_prefix("data: "))
7858 .filter(|payload| *payload != "[DONE]")
7859 .map(|payload| serde_json::from_str(payload).unwrap())
7860 .collect()
7861 }
7862
7863 #[tokio::test]
7864 async fn a_stream_states_its_request_id_once_in_the_first_chunk() {
7865 let app = test_app();
7866 let chunks = post_sse_chunks(
7867 &app,
7868 serde_json::json!({
7869 "model": "m",
7870 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7871 "max_tokens": 4,
7872 "temperature": 0,
7873 "stream": true,
7874 }),
7875 )
7876 .await;
7877
7878 assert!(!chunks.is_empty());
7879 let request_id = chunks[0]["request_id"]
7880 .as_str()
7881 .expect("the first chunk names the request")
7882 .to_string();
7883 assert!(request_id.starts_with("chatcmpl-"), "{request_id}");
7884 for (i, chunk) in chunks.iter().enumerate().skip(1) {
7887 assert!(
7888 chunk.get("request_id").is_none(),
7889 "chunk {i} repeats request_id"
7890 );
7891 }
7892 for chunk in &chunks {
7895 assert_eq!(chunk["id"], serde_json::json!(request_id));
7896 }
7897
7898 let other = post_sse_chunks(
7899 &app,
7900 serde_json::json!({
7901 "model": "m",
7902 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7903 "max_tokens": 4,
7904 "temperature": 0,
7905 "stream": true,
7906 }),
7907 )
7908 .await;
7909 assert_ne!(
7910 other[0]["request_id"].as_str().unwrap(),
7911 request_id,
7912 "two concurrent chats must not share an id"
7913 );
7914 }
7915
7916 #[tokio::test]
7917 async fn a_non_streamed_response_names_the_same_request_id_as_its_completion_id() {
7918 let app = test_app();
7919 let resp = post_json(
7920 &app,
7921 serde_json::json!({
7922 "model": "m",
7923 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7924 "max_tokens": 2,
7925 "temperature": 0,
7926 }),
7927 )
7928 .await;
7929 assert_eq!(resp["id"], resp["request_id"]);
7930 assert!(resp["request_id"]
7931 .as_str()
7932 .unwrap()
7933 .starts_with("chatcmpl-"));
7934 }
7935
7936 #[tokio::test]
7939 async fn usage_carries_separate_prefill_and_decode_timings() {
7940 let app = test_app();
7941 let resp = post_json(
7942 &app,
7943 serde_json::json!({
7944 "model": "m",
7945 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7946 "max_tokens": 4,
7947 "temperature": 0,
7948 }),
7949 )
7950 .await;
7951 let usage = &resp["usage"];
7952 assert!(usage["prompt_eval_duration_ms"].is_number(), "{usage}");
7953 assert!(usage["generation_duration_ms"].is_number(), "{usage}");
7954 assert!(usage["time_to_first_token_ms"].is_number(), "{usage}");
7955 assert!(usage["predicted_per_second"].is_number(), "{usage}");
7956 assert!(usage.get("cached_tokens").is_none(), "{usage}");
7958 }
7959
7960 #[tokio::test]
7968 async fn a_tools_request_with_no_marker_in_the_output_falls_back_to_plain_content() {
7969 let app = test_app();
7970 let body = serde_json::json!({
7971 "model": "m",
7972 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
7973 "max_tokens": 4,
7974 "temperature": 0,
7975 "tools": [weather_tool()],
7976 });
7977 let resp = post_json(&app, body).await;
7978 let message = &resp["choices"][0]["message"];
7979 assert!(
7980 message["content"].is_string(),
7981 "must fall back to plain content when no real tool-call marker is present: {resp:?}"
7982 );
7983 assert!(message.get("tool_calls").is_none());
7984 assert_eq!(resp["choices"][0]["finish_reason"], "length");
7988 }
7989
7990 #[tokio::test]
7994 async fn a_cache_hit_reports_the_original_finish_reason_and_usage() {
7995 let app = test_app();
7996 let body = serde_json::json!({
7997 "model": "m",
7998 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
7999 "max_tokens": 3,
8000 "temperature": 0,
8001 });
8002 let first = post_json(&app, body.clone()).await;
8003 assert_eq!(first["frink_cache"], "miss");
8004 let second = post_json(&app, body).await;
8005 assert_eq!(second["frink_cache"], "hit");
8006 assert_eq!(
8007 first["choices"][0]["message"]["content"],
8008 second["choices"][0]["message"]["content"]
8009 );
8010 assert_eq!(
8011 first["choices"][0]["finish_reason"],
8012 second["choices"][0]["finish_reason"]
8013 );
8014 assert_eq!(first["usage"], second["usage"]);
8015 assert_eq!(second["usage"]["completion_tokens"], 3);
8016 }
8017
8018 #[tokio::test]
8029 async fn a_grammar_request_is_not_answered_from_an_unconstrained_cache_entry() {
8030 let app = test_app();
8031 let plain = serde_json::json!({
8032 "model": "m",
8033 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
8034 "max_tokens": 3,
8035 "temperature": 0,
8036 });
8037
8038 let first = post_json(&app, plain.clone()).await;
8039 assert_eq!(first["frink_cache"], "miss");
8040 let unconstrained = first["choices"][0]["message"]["content"]
8041 .as_str()
8042 .expect("content")
8043 .to_string();
8044
8045 let mut constrained = plain.clone();
8046 constrained["grammar"] = serde_json::json!("root ::= \"yes\"");
8047 let second = post_json(&app, constrained).await;
8048 assert_eq!(
8049 second["frink_cache"], "miss",
8050 "a grammar is part of the key, so this body has never been answered"
8051 );
8052 let constrained_answer = second["choices"][0]["message"]["content"]
8056 .as_str()
8057 .expect("content")
8058 .to_string();
8059 assert!(
8060 constrained_answer.contains("-> \"yes\"]"),
8061 "the grammar must have been compiled AND applied, not skipped \
8062 by a cache hit: {constrained_answer}"
8063 );
8064 assert_ne!(
8065 constrained_answer, unconstrained,
8066 "the constrained request was served the unconstrained answer"
8067 );
8068
8069 let third = post_json(&app, plain).await;
8073 assert_eq!(third["frink_cache"], "hit");
8074 assert_eq!(third["choices"][0]["message"]["content"], unconstrained);
8075 }
8076
8077 #[tokio::test]
8099 async fn a_json_object_request_does_not_reuse_the_unconstrained_cache_entry() {
8100 let state = Arc::new(test_state(
8101 test_model_full_byte_vocab(),
8102 ResponseCache::new(1000, Duration::from_secs(3600)),
8103 ));
8104 let app = test_app_with_state(state.clone());
8105 let plain = serde_json::json!({
8106 "model": "m",
8107 "messages": [
8108 {"role": "system", "content": "Answer in JSON when it helps."},
8109 {"role": "user", "content": "\u{1}\u{2}"},
8110 ],
8111 "max_tokens": 3,
8112 "temperature": 0,
8113 });
8114
8115 let first = post_json(&app, plain.clone()).await;
8116 assert_eq!(first["frink_cache"], "miss");
8117 assert_eq!(state.cache_stats().entries, 1);
8118
8119 let mut as_json = plain.clone();
8120 as_json["response_format"] = serde_json::json!({"type": "json_object"});
8121 let (status, _) = post_json_uri(&app, "/v1/chat/completions", as_json).await;
8122 assert_eq!(
8123 status,
8124 StatusCode::BAD_REQUEST,
8125 "the demo banner is not a JSON object, whoever generated it"
8126 );
8127 assert_eq!(
8128 state.cache_stats().hits,
8129 0,
8130 "a json_object request must not be answered from an entry the \
8131 JSON mask never produced"
8132 );
8133 assert_eq!(
8134 state.cache_stats().entries,
8135 2,
8136 "json_object must key its own entry, not reuse the unconstrained \
8137 one it happens to render the same prompt as"
8138 );
8139 }
8140
8141 #[tokio::test]
8153 async fn an_ignore_eos_request_is_not_answered_from_a_cache_entry_that_stopped_at_eos() {
8154 let app = test_app_with_state(Arc::new(test_state(
8155 test_model_full_byte_vocab_with_eos(Some(0x77)),
8156 ResponseCache::new(1000, Duration::from_secs(3600)),
8157 )));
8158 let body = serde_json::json!({
8159 "model": "m",
8160 "messages": [{"role": "user", "content": "\u{1}\u{2}"}],
8161 "max_tokens": 6,
8162 "temperature": 0,
8163 });
8164
8165 let stopped = post_json(&app, body.clone()).await;
8166 assert_eq!(stopped["frink_cache"], "miss");
8167 assert_eq!(
8168 stopped["choices"][0]["finish_reason"], "stop",
8169 "the fixture is only meaningful if the model's EOS really fires here"
8170 );
8171 assert_eq!(stopped["usage"]["completion_tokens"], 1);
8172
8173 let mut ignoring = body.clone();
8174 ignoring["ignore_eos"] = serde_json::json!(true);
8175 let ran_on = post_json(&app, ignoring).await;
8176 assert_eq!(
8177 ran_on["frink_cache"], "miss",
8178 "ignore_eos is part of the key, so this body has never been answered"
8179 );
8180 assert_eq!(
8181 ran_on["usage"]["completion_tokens"], 6,
8182 "ignore_eos must run the full budget, not replay the EOS-terminated answer"
8183 );
8184 assert_eq!(ran_on["choices"][0]["finish_reason"], "length");
8185 assert_ne!(
8186 ran_on["choices"][0]["message"]["content"],
8187 stopped["choices"][0]["message"]["content"]
8188 );
8189 }
8190
8191 #[tokio::test]
8197 async fn session_reuse_produces_the_same_output_as_manually_resending_full_history() {
8198 let session_app = test_app();
8199 let manual_app = test_app();
8200
8201 let turn1 = post_json(
8203 &session_app,
8204 serde_json::json!({
8205 "model": "m",
8206 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
8207 "session_id": "s1",
8208 "max_tokens": 5,
8209 "temperature": 0,
8210 }),
8211 )
8212 .await;
8213 let reply1 = turn1["choices"][0]["message"]["content"]
8214 .as_str()
8215 .unwrap()
8216 .to_string();
8217
8218 let manual_turn1 = post_json(
8223 &manual_app,
8224 serde_json::json!({
8225 "model": "m",
8226 "messages": [{"role": "user", "content": "\u{1}\u{2}\u{3}"}],
8227 "max_tokens": 5,
8228 "temperature": 0,
8229 }),
8230 )
8231 .await;
8232 assert_eq!(
8233 manual_turn1["choices"][0]["message"]["content"]
8234 .as_str()
8235 .unwrap(),
8236 reply1
8237 );
8238
8239 let turn2 = post_json(
8241 &session_app,
8242 serde_json::json!({
8243 "model": "m",
8244 "messages": [{"role": "user", "content": "\u{4}\u{5}"}],
8245 "session_id": "s1",
8246 "max_tokens": 5,
8247 "temperature": 0,
8248 }),
8249 )
8250 .await;
8251 let reply2 = turn2["choices"][0]["message"]["content"]
8252 .as_str()
8253 .unwrap()
8254 .to_string();
8255
8256 let manual_turn2 = post_json(
8260 &manual_app,
8261 serde_json::json!({
8262 "model": "m",
8263 "messages": [
8264 {"role": "user", "content": "\u{1}\u{2}\u{3}"},
8265 {"role": "assistant", "content": reply1},
8266 {"role": "user", "content": "\u{4}\u{5}"},
8267 ],
8268 "max_tokens": 5,
8269 "temperature": 0,
8270 }),
8271 )
8272 .await;
8273 assert_eq!(
8274 manual_turn2["choices"][0]["message"]["content"]
8275 .as_str()
8276 .unwrap(),
8277 reply2,
8278 "resuming a session must produce identical output to manually resending the full history"
8279 );
8280 }
8281
8282 #[test]
8285 fn lock_cache_recovers_from_a_poisoned_mutex() {
8286 let cache = Arc::new(Mutex::new(ResponseCache::new(10, Duration::from_secs(60))));
8287
8288 let poison_cache = Arc::clone(&cache);
8289 let _ = std::thread::spawn(move || {
8290 let _guard = poison_cache.lock().unwrap();
8291 panic!("simulated panic while holding the lock");
8292 })
8293 .join();
8294
8295 let recovered = lock_cache(&cache);
8297 assert_eq!(recovered.stats().entries, 0);
8298 }
8299
8300 #[test]
8301 fn is_cacheable_true_for_greedy_or_seeded_requests() {
8302 let mut req_body = serde_json::json!({
8303 "model": "m",
8304 "messages": [{"role": "user", "content": "hi"}],
8305 });
8306 let req: ChatCompletionRequest = serde_json::from_value(req_body.clone()).unwrap();
8307 assert!(
8308 req.is_cacheable(),
8309 "default (temperature 0) must be cacheable"
8310 );
8311
8312 req_body["temperature"] = serde_json::json!(0.8);
8313 let req: ChatCompletionRequest = serde_json::from_value(req_body.clone()).unwrap();
8314 assert!(
8315 !req.is_cacheable(),
8316 "unseeded sampling must never be cacheable"
8317 );
8318
8319 req_body["seed"] = serde_json::json!(42);
8320 let req: ChatCompletionRequest = serde_json::from_value(req_body).unwrap();
8321 assert!(
8322 req.is_cacheable(),
8323 "sampling with an explicit seed is deterministic and must be cacheable"
8324 );
8325 }
8326
8327 const GRADED: &str = "{% if reasoning_effort %}\
8331 {% if reasoning_effort not in ['low','medium','high'] %}\
8332 {{ raise_exception('unsupported effort') }}\
8333 {% endif %}E:{{ reasoning_effort }}|{% endif %}\
8334 {% if enable_thinking %}THINK|{% endif %}{{ messages[0].content }}";
8335
8336 fn graded_template() -> chat_template::PromptTemplate {
8337 chat_template::PromptTemplate::from_gguf_metadata(
8338 Some(GRADED),
8339 Some("qwen3"),
8340 false,
8341 true,
8342 None,
8343 None,
8344 )
8345 }
8346
8347 fn chat_request(value: serde_json::Value) -> ChatCompletionRequest {
8348 serde_json::from_value(value).expect("request")
8349 }
8350
8351 #[test]
8358 fn a_grammar_on_the_chat_wire_reaches_the_generation_params() {
8359 let req = chat_request(serde_json::json!({
8360 "model": "m",
8361 "messages": [{"role": "user", "content": "hi"}],
8362 "grammar": "root ::= \"a\"+",
8363 }));
8364 req.validate_supported_fields()
8365 .expect("a valid grammar is a valid request");
8366 let params = req
8367 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8368 .expect("a valid grammar compiles at params time too");
8369 assert!(
8370 params.grammar.is_some(),
8371 "the grammar was dropped between the wire and the sampler"
8372 );
8373 assert!(
8374 params.needs_vocab_logits(),
8375 "a grammar request that may fold lm_head into a GPU argmax is \
8376 a grammar request served unconstrained"
8377 );
8378
8379 let plain = chat_request(serde_json::json!({
8380 "model": "m",
8381 "messages": [{"role": "user", "content": "hi"}],
8382 }));
8383 assert!(plain
8384 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8385 .unwrap()
8386 .grammar
8387 .is_none());
8388 }
8389
8390 fn tool_request(tool_choice: serde_json::Value) -> ChatCompletionRequest {
8391 chat_request(serde_json::json!({
8392 "model": "m",
8393 "messages": [{"role": "user", "content": "weather in Rome?"}],
8394 "tools": [weather_tool()],
8395 "tool_choice": tool_choice,
8396 }))
8397 }
8398
8399 #[test]
8403 fn a_forced_tool_choice_puts_a_grammar_on_the_generation_params() {
8404 for choice in [
8405 serde_json::json!("required"),
8406 serde_json::json!({"type": "function", "function": {"name": "get_weather"}}),
8407 ] {
8408 let req = tool_request(choice.clone());
8409 req.validate_supported_fields()
8410 .unwrap_or_else(|e| panic!("{choice} is a valid request: {e:?}"));
8411 let params = req
8412 .generation_params_for_template(
8413 &graded_template(),
8414 "Qwen3-8B",
8415 crate::sampling_knobs::SamplerModel::absent(),
8416 )
8417 .unwrap_or_else(|e| panic!("{choice} compiles: {e:?}"));
8418 let grammar = params
8419 .grammar
8420 .as_ref()
8421 .unwrap_or_else(|| panic!("{choice} was accepted and then not enforced"));
8422 assert!(
8423 grammar.is_awaiting_trigger(),
8424 "the model must be free to think before it calls"
8425 );
8426 assert!(
8427 !grammar.allows_eog(),
8428 "{choice} must not be able to end the turn without a call"
8429 );
8430 assert!(
8436 params.needs_vocab_logits(),
8437 "{choice} would let a backend return a token id instead of logits"
8438 );
8439 assert!(
8440 !generate::greedy_gpu_fold_allowed(¶ms),
8441 "{choice} at temperature 0 must still refuse the greedy GPU fold"
8442 );
8443 }
8444 }
8445
8446 #[test]
8448 fn an_unforced_tool_choice_leaves_the_generation_unconstrained() {
8449 for choice in [serde_json::json!("auto"), serde_json::json!("none")] {
8450 let req = tool_request(choice.clone());
8451 req.validate_supported_fields().expect("still supported");
8452 let params = match req.generation_params_for_template(
8453 &graded_template(),
8454 "Qwen3-8B",
8455 crate::sampling_knobs::SamplerModel::absent(),
8456 ) {
8457 Ok(p) => p,
8458 Err((status, _)) => panic!("{choice} has no constraint to compile: {status}"),
8459 };
8460 assert!(
8461 params.grammar.is_none(),
8462 "{choice} does not force a call and must not be constrained"
8463 );
8464 }
8465 }
8466
8467 #[test]
8470 fn a_forced_tool_choice_refuses_rather_than_quietly_not_forcing() {
8471 let req = chat_request(serde_json::json!({
8473 "model": "m",
8474 "messages": [{"role": "user", "content": "hi"}],
8475 "tool_choice": "required",
8476 }));
8477 let (status, _) = req
8478 .validate_supported_fields()
8479 .expect_err("nothing to call");
8480 assert_eq!(status, StatusCode::BAD_REQUEST);
8481
8482 let req =
8484 tool_request(serde_json::json!({"type": "function", "function": {"name": "nope"}}));
8485 let (status, Json(body)) = req.validate_supported_fields().expect_err("no such tool");
8486 assert_eq!(status, StatusCode::BAD_REQUEST);
8487 assert_eq!(body["error"]["param"], "tool_choice");
8488
8489 let req = tool_request(serde_json::json!({"type": "function"}));
8491 let (status, _) = req.validate_supported_fields().expect_err("names nothing");
8492 assert_eq!(status, StatusCode::BAD_REQUEST);
8493
8494 let req = chat_request(serde_json::json!({
8496 "model": "m",
8497 "messages": [{"role": "user", "content": "hi"}],
8498 "tools": [weather_tool()],
8499 "tool_choice": "required",
8500 "grammar": "root ::= \"a\"+",
8501 }));
8502 let (status, _) = req
8503 .validate_supported_fields()
8504 .expect_err("a grammar and a forced call are two constraints");
8505 assert_eq!(status, StatusCode::BAD_REQUEST);
8506
8507 let req = tool_request(serde_json::json!("required"));
8513 let (status, Json(body)) = match req.generation_params_for_template(
8514 &graded_template(),
8515 "muse-glimmer-8b",
8516 crate::sampling_knobs::SamplerModel::absent(),
8517 ) {
8518 Err(e) => e,
8519 Ok(_) => panic!("a muse_glimmer call's boundary is a channel, not a marker"),
8520 };
8521 assert_eq!(status, StatusCode::NOT_IMPLEMENTED);
8522 assert!(
8523 body["error"]["message"]
8524 .as_str()
8525 .unwrap()
8526 .contains("muse_glimmer"),
8527 "{body}"
8528 );
8529
8530 let req = tool_request(serde_json::json!("required"));
8536 let params = req
8537 .generation_params_for_template(
8538 &graded_template(),
8539 "gemma-4-E2B-it",
8540 crate::sampling_knobs::SamplerModel::absent(),
8541 )
8542 .expect("a gemma4 forced tool_choice is served");
8543 assert!(
8544 params.grammar.is_some(),
8545 "a forced tool_choice must arrive as the generation's grammar"
8546 );
8547 }
8548
8549 #[test]
8552 fn an_unparseable_grammar_on_the_chat_wire_is_a_400() {
8553 let req = chat_request(serde_json::json!({
8554 "model": "m",
8555 "messages": [{"role": "user", "content": "hi"}],
8556 "grammar": "root ::= \"a",
8557 }));
8558 let (status, Json(body)) = req
8559 .validate_supported_fields()
8560 .expect_err("this does not parse");
8561 assert_eq!(status, StatusCode::BAD_REQUEST);
8562 assert_eq!(body["error"]["param"], "grammar");
8563 assert!(
8564 req.generation_params(crate::sampling_knobs::SamplerModel::absent())
8565 .is_err(),
8566 "and again at params time"
8567 );
8568 }
8569
8570 #[test]
8577 fn response_format_json_schema_becomes_the_requests_grammar() {
8578 let req = chat_request(serde_json::json!({
8579 "model": "m",
8580 "messages": [{"role": "user", "content": "hi"}],
8581 "response_format": {
8582 "type": "json_schema",
8583 "json_schema": {"name": "x", "schema": {"type": "boolean"}},
8584 },
8585 }));
8586 req.validate_supported_fields()
8587 .expect("a boolean schema converts");
8588 let params = req
8589 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8590 .expect("and compiles");
8591 let grammar = params.grammar.expect("the schema is the grammar");
8592 let mut g = (*grammar).clone();
8593 g.accept_token(0, b"true").expect("a boolean is accepted");
8594 assert!(g.allows_eog(), "and completes the parse");
8595 assert!(
8596 !params.json_object,
8597 "a schema is not the json_object character-class mask"
8598 );
8599 }
8600
8601 #[test]
8605 fn an_unconvertible_response_format_schema_is_a_400_naming_the_keyword() {
8606 let req = chat_request(serde_json::json!({
8607 "model": "m",
8608 "messages": [{"role": "user", "content": "hi"}],
8609 "response_format": {
8610 "type": "json_schema",
8611 "json_schema": {"name": "x", "schema": {"type": "integer", "minimum": 3}},
8612 },
8613 }));
8614 let (status, Json(body)) = req
8615 .validate_supported_fields()
8616 .expect_err("minimum has no grammar in this port");
8617 assert_eq!(status, StatusCode::BAD_REQUEST);
8618 assert!(
8619 body["error"]["message"]
8620 .as_str()
8621 .expect("a message")
8622 .contains("minimum"),
8623 "the refusal must name the keyword: {body}"
8624 );
8625 assert!(
8626 req.generation_params(crate::sampling_knobs::SamplerModel::absent())
8627 .is_err(),
8628 "and again at params time"
8629 );
8630 }
8631
8632 #[test]
8638 fn a_forced_tool_choice_and_a_schema_are_two_constraints() {
8639 let req = chat_request(serde_json::json!({
8640 "model": "m",
8641 "messages": [{"role": "user", "content": "hi"}],
8642 "tool_choice": "required",
8643 "tools": [{
8644 "type": "function",
8645 "function": {"name": "f", "parameters": {"type": "object"}},
8646 }],
8647 "response_format": {
8648 "type": "json_schema",
8649 "json_schema": {"name": "x", "schema": {"type": "boolean"}},
8650 },
8651 }));
8652 let (status, Json(body)) = req
8653 .validate_supported_fields()
8654 .expect_err("two constraints, one generation");
8655 assert_eq!(status, StatusCode::BAD_REQUEST);
8656 assert_eq!(body["error"]["param"], "tool_choice");
8657 }
8658
8659 #[test]
8662 fn an_omitted_output_budget_is_a_whole_answer_not_sixteen_tokens() {
8663 let req = chat_request(serde_json::json!({
8664 "model": "m",
8665 "messages": [{"role": "user", "content": "hi"}],
8666 }));
8667 assert_eq!(req.max_tokens, DEFAULT_CHAT_MAX_TOKENS);
8668 }
8669
8670 #[test]
8675 fn min_p_reaches_the_sampler_from_the_chat_wire() {
8676 let asked = chat_request(serde_json::json!({
8677 "model": "m",
8678 "messages": [{"role": "user", "content": "hi"}],
8679 "min_p": 0.07,
8680 }));
8681 assert_eq!(
8682 asked
8683 .sampling_params(crate::sampling_knobs::SamplerModel::absent())
8684 .expect("knobs")
8685 .min_p,
8686 0.07
8687 );
8688
8689 let silent = chat_request(serde_json::json!({
8690 "model": "m",
8691 "messages": [{"role": "user", "content": "hi"}],
8692 }));
8693 assert_eq!(
8694 silent
8695 .sampling_params(crate::sampling_knobs::SamplerModel::absent())
8696 .expect("knobs")
8697 .min_p,
8698 0.0,
8699 "an unset min_p must be off, not llama.cpp's CLI default"
8700 );
8701 }
8702
8703 #[test]
8712 fn no_sampler_knob_is_missing_from_the_cache_key() {
8713 let base = serde_json::json!({
8714 "model": "m",
8715 "messages": [{"role": "user", "content": "hi"}],
8716 "seed": 1,
8717 });
8718 let key_for = |body: serde_json::Value| {
8719 let req = chat_request(body);
8720 let params = req
8721 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8722 .expect("params");
8723 req.cache_key("prompt", ¶ms)
8724 };
8725 let baseline = key_for(base.clone());
8726 for (knob, value) in [
8727 ("temperature", serde_json::json!(0.5)),
8728 ("top_p", serde_json::json!(0.9)),
8729 ("min_p", serde_json::json!(0.05)),
8730 ("top_k", serde_json::json!(40)),
8731 ("repetition_penalty", serde_json::json!(1.1)),
8732 ("presence_penalty", serde_json::json!(0.3)),
8733 ("frequency_penalty", serde_json::json!(0.3)),
8734 (
8735 "samplers",
8736 serde_json::json!(["penalties", "top_p", "top_k", "min_p", "temperature"]),
8737 ),
8738 ] {
8739 let mut body = base.clone();
8740 body[knob] = value;
8741 assert_ne!(
8742 key_for(body),
8743 baseline,
8744 "`{knob}` is not in the cache key: two requests differing \
8745 only in it would share one cached answer"
8746 );
8747 }
8748 }
8749
8750 #[test]
8760 fn no_constraint_is_missing_from_the_cache_key() {
8761 let base = serde_json::json!({
8762 "model": "m",
8763 "messages": [{"role": "user", "content": "pick one"}],
8764 });
8765 let key_for = |body: serde_json::Value| {
8766 let req = chat_request(body);
8767 let params = req
8768 .generation_params(crate::sampling_knobs::SamplerModel::absent())
8769 .expect("params");
8770 req.cache_key("prompt", ¶ms)
8771 };
8772 let baseline = key_for(base.clone());
8773 for (field, value) in [
8774 ("grammar", serde_json::json!("root ::= \"yes\" | \"no\"")),
8775 (
8776 "response_format",
8777 serde_json::json!({"type": "json_object"}),
8778 ),
8779 (
8780 "response_format",
8781 serde_json::json!({"type": "json_schema", "json_schema": {
8782 "name": "answer",
8783 "schema": {"type": "object", "properties": {"a": {"type": "string"}}}
8784 }}),
8785 ),
8786 ("ignore_eos", serde_json::json!(true)),
8787 ("stop", serde_json::json!(["\n"])),
8788 ("max_tokens", serde_json::json!(7)),
8789 ] {
8790 let mut body = base.clone();
8791 body[field] = value.clone();
8792 assert_ne!(
8793 key_for(body),
8794 baseline,
8795 "`{field}: {value}` is not in the cache key: two requests \
8796 differing only in it would share one cached answer"
8797 );
8798 }
8799 }
8800
8801 #[test]
8805 fn an_explicit_zero_output_budget_is_a_client_error() {
8806 let req = chat_request(serde_json::json!({
8807 "model": "m",
8808 "messages": [{"role": "user", "content": "hi"}],
8809 "max_tokens": 0,
8810 }));
8811 let (status, body) = req.validate_supported_fields().expect_err("rejected");
8812 assert_eq!(status, StatusCode::BAD_REQUEST);
8813 assert_eq!(body["error"]["param"], serde_json::json!("max_tokens"));
8814 }
8815
8816 #[test]
8819 fn a_request_can_turn_thinking_off() {
8820 let template = graded_template();
8821 for body in [
8822 serde_json::json!({
8823 "model": "m",
8824 "messages": [{"role": "user", "content": "hi"}],
8825 "reasoning_effort": "none",
8826 }),
8827 serde_json::json!({
8828 "model": "m",
8829 "messages": [{"role": "user", "content": "hi"}],
8830 "thinking": {"type": "disabled"},
8831 }),
8832 ] {
8833 let kwargs = chat_request(body).resolve_template_kwargs(&template);
8834 assert_eq!(kwargs["enable_thinking"], serde_json::json!(false));
8835 assert_eq!(kwargs["thinking_mode"], serde_json::json!("disabled"));
8836 assert!(!kwargs.contains_key("reasoning_effort"));
8839 }
8840 }
8841
8842 #[test]
8845 fn a_disabled_switch_beats_an_effort_in_the_same_request() {
8846 let template = graded_template();
8847 let kwargs = chat_request(serde_json::json!({
8848 "model": "m",
8849 "messages": [{"role": "user", "content": "hi"}],
8850 "reasoning_effort": "high",
8851 "thinking": {"type": "disabled"},
8852 }))
8853 .resolve_template_kwargs(&template);
8854 assert_eq!(kwargs["enable_thinking"], serde_json::json!(false));
8855 assert!(!kwargs.contains_key("reasoning_effort"));
8856 }
8857
8858 #[test]
8861 fn an_unrecognized_thinking_switch_is_refused_rather_than_read_as_on() {
8862 let req = chat_request(serde_json::json!({
8863 "model": "m",
8864 "messages": [{"role": "user", "content": "hi"}],
8865 "thinking": {"type": "disable"},
8866 }));
8867 let (status, _) = req.validate_supported_fields().expect_err("rejected");
8868 assert_eq!(status, StatusCode::BAD_REQUEST);
8869 }
8870
8871 #[test]
8874 fn an_explicit_template_kwarg_stands_the_protocol_knobs_down() {
8875 let template = graded_template();
8876 let kwargs = chat_request(serde_json::json!({
8877 "model": "m",
8878 "messages": [{"role": "user", "content": "hi"}],
8879 "reasoning_effort": "none",
8880 "chat_template_kwargs": {"enable_thinking": true},
8881 }))
8882 .resolve_template_kwargs(&template);
8883 assert_eq!(kwargs["enable_thinking"], serde_json::json!(true));
8884 }
8885
8886 #[test]
8890 fn an_off_vocabulary_reasoning_effort_is_quantized_rather_than_interpolated() {
8891 let template = graded_template();
8892 let req = chat_request(serde_json::json!({
8893 "model": "m",
8894 "messages": [{"role": "user", "content": "hi"}],
8895 "reasoning_effort": "minimal",
8896 }));
8897 let kwargs = req.resolve_template_kwargs(&template);
8898 assert_eq!(kwargs["reasoning_effort"], serde_json::json!("low"));
8899 let prompt = prompt_from_messages(&req.messages, &template, &[], kwargs).expect("renders");
8900 assert!(prompt.starts_with("E:low|"), "{prompt}");
8901 }
8902
8903 #[test]
8907 fn an_effort_with_no_near_gear_is_dropped_so_the_template_default_applies() {
8908 let template = graded_template();
8909 let req = chat_request(serde_json::json!({
8910 "model": "m",
8911 "messages": [{"role": "user", "content": "hi"}],
8912 "chat_template_kwargs": {"reasoning_effort": "none"},
8913 }));
8914 let kwargs = req.resolve_template_kwargs(&template);
8915 assert!(!kwargs.contains_key("reasoning_effort"));
8916 let prompt = prompt_from_messages(&req.messages, &template, &[], kwargs).expect("renders");
8917 assert_eq!(prompt, "hi");
8918 }
8919
8920 #[test]
8923 fn chat_template_kwargs_wins_over_the_top_level_reasoning_effort() {
8924 let template = graded_template();
8925 let req = chat_request(serde_json::json!({
8926 "model": "m",
8927 "messages": [{"role": "user", "content": "hi"}],
8928 "reasoning_effort": "low",
8929 "chat_template_kwargs": {"reasoning_effort": "high"},
8930 }));
8931 assert_eq!(
8932 req.resolve_template_kwargs(&template)["reasoning_effort"],
8933 serde_json::json!("high")
8934 );
8935 }
8936
8937 #[test]
8941 fn offering_tools_turns_thinking_on_by_itself() {
8942 let template = graded_template();
8943 let quiet = chat_request(serde_json::json!({
8944 "model": "m",
8945 "messages": [{"role": "user", "content": "hi"}],
8946 }));
8947 assert!(!quiet
8948 .resolve_template_kwargs(&template)
8949 .contains_key("enable_thinking"));
8950
8951 let with_tools = chat_request(serde_json::json!({
8952 "model": "m",
8953 "messages": [{"role": "user", "content": "hi"}],
8954 "tools": [{"type": "function", "function": {"name": "get_weather"}}],
8955 }));
8956 let kwargs = with_tools.resolve_template_kwargs(&template);
8957 assert_eq!(kwargs["enable_thinking"], serde_json::json!(true));
8958 let prompt =
8959 prompt_from_messages(&with_tools.messages, &template, &[], kwargs).expect("renders");
8960 assert!(prompt.starts_with("THINK|"), "{prompt}");
8961 }
8962
8963 #[test]
8968 fn a_prompt_that_opens_the_reasoning_block_makes_the_first_token_reasoning() {
8969 let opener = chat_template::PromptTemplate::from_gguf_metadata(
8970 Some("{{ messages[0].content }}{% if enable_thinking %}<think>{% endif %}"),
8971 Some("qwen3"),
8972 false,
8973 true,
8974 None,
8975 None,
8976 );
8977 let req = chat_request(serde_json::json!({
8978 "model": "m",
8979 "messages": [{"role": "user", "content": "hi"}],
8980 "chat_template_kwargs": {"enable_thinking": true},
8981 }));
8982 let kwargs = req.resolve_template_kwargs(&opener);
8983 let prompt = prompt_from_messages(&req.messages, &opener, &[], kwargs).expect("renders");
8984 assert!(prompt.ends_with("<think>"), "{prompt}");
8985
8986 let posture = output::OutputPosture::resolve("Qwen3-8B", &prompt);
8989 let (message, _) = build_response_message(
8990 "weighing it up</think>Paris.".to_string(),
8991 &[],
8992 posture,
8993 "stop",
8994 );
8995 assert_eq!(message.reasoning_content.as_deref(), Some("weighing it up"));
8996 assert_eq!(message.content.as_deref(), Some("Paris."));
8997
8998 let closed = output::OutputPosture::resolve("Qwen3-8B", "<|im_start|>assistant\n");
9001 let (message, _) = build_response_message(
9002 "weighing it up</think>Paris.".to_string(),
9003 &[],
9004 closed,
9005 "stop",
9006 );
9007 assert_eq!(message.reasoning_content, None);
9008 }
9009
9010 #[test]
9011 fn stop_param_accepts_both_single_string_and_array() {
9012 let req: ChatCompletionRequest = serde_json::from_value(serde_json::json!({
9013 "model": "m",
9014 "messages": [{"role": "user", "content": "hi"}],
9015 "stop": "END",
9016 }))
9017 .unwrap();
9018 assert_eq!(req.stop_sequences(), vec!["END".to_string()]);
9019
9020 let req: ChatCompletionRequest = serde_json::from_value(serde_json::json!({
9021 "model": "m",
9022 "messages": [{"role": "user", "content": "hi"}],
9023 "stop": ["A", "B"],
9024 }))
9025 .unwrap();
9026 assert_eq!(req.stop_sequences(), vec!["A".to_string(), "B".to_string()]);
9027 }
9028
9029 #[test]
9030 fn run_generation_rejects_out_of_vocab_tokens_instead_of_panicking() {
9031 let model = test_model();
9032 let result = run_generation(
9033 &model,
9034 "hello",
9035 &greedy_params(4),
9036 None,
9037 None,
9038 None,
9039 None,
9040 None,
9041 None,
9042 );
9043 assert!(matches!(
9044 result,
9045 Err(generate::DecodeError::TokenOutOfVocab { .. })
9046 ));
9047 }
9048
9049 #[test]
9053 fn run_generation_honors_an_exhausted_kv_pool_and_maps_it_to_a_503() {
9054 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9056 let pool = Arc::new(Mutex::new(frink_core::cache::KvBlockPool::new(64, 2)));
9057
9058 let holder_pool = Arc::clone(&pool);
9059 let holder = std::thread::spawn(move || {
9060 let mut held = frink_core::cache::KvCache::with_pool(1, 1, holder_pool, 0).unwrap();
9061 held.push(&[0.0], &[0.0]).unwrap(); std::thread::sleep(Duration::from_millis(200));
9063 drop(held);
9064 });
9065 std::thread::sleep(Duration::from_millis(15));
9066
9067 let config = generate::KvPoolConfig {
9068 pool,
9069 queue_wait: Duration::ZERO,
9070 };
9071 let result = run_generation(
9072 &model,
9073 &prompt,
9074 &greedy_params(4),
9075 Some(&config),
9076 None,
9077 None,
9078 None,
9079 None,
9080 None,
9081 );
9082 assert!(matches!(
9083 result,
9084 Err(generate::DecodeError::KvPoolExhausted)
9085 ));
9086
9087 let (status, _body) = decode_error_response(result.unwrap_err());
9088 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
9089 holder.join().unwrap();
9090 }
9091
9092 #[test]
9100 fn a_request_too_big_for_the_whole_pool_is_a_400_not_a_retryable_503() {
9101 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9103 let pool = Arc::new(Mutex::new(frink_core::cache::KvBlockPool::new(64, 1)));
9105 let config = generate::KvPoolConfig {
9106 pool,
9107 queue_wait: Duration::ZERO,
9108 };
9109
9110 let result = run_generation(
9111 &model,
9112 &prompt,
9113 &greedy_params(4),
9114 Some(&config),
9115 None,
9116 None,
9117 None,
9118 None,
9119 None,
9120 );
9121 let err = result.expect_err("one block cannot hold two layers' caches");
9122 assert!(
9123 matches!(
9124 &err,
9125 generate::DecodeError::KvBudgetExceeded { binding, .. }
9126 if *binding == frink_models::Ceiling::DeviceMemory.code()
9127 ),
9128 "expected an immovable device-memory refusal, got {err:?}"
9129 );
9130 let (status, _body) = decode_error_response(err);
9131 assert_eq!(status, StatusCode::BAD_REQUEST);
9132 }
9133
9134 #[test]
9140 fn decode_error_response_maps_a_full_queue_to_a_retryable_503() {
9141 let (status, Json(body)) = decode_error_response(generate::DecodeError::QueueFull {
9142 queued: 512,
9143 cap: 512,
9144 });
9145 assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
9146 assert_eq!(body["error"]["retry_after_seconds"], 1);
9147 let message = body["error"]["message"].as_str().expect("message");
9148 assert!(message.contains("512"), "{message}");
9149 }
9150
9151 #[test]
9152 fn decode_error_response_omits_a_retry_hint_for_an_unretryable_error() {
9153 let (_status, Json(body)) = decode_error_response(generate::DecodeError::TokenOutOfVocab {
9154 token: 99,
9155 vocab_size: 32,
9156 });
9157 assert!(
9158 body["error"]["retry_after_seconds"].is_null(),
9159 "retrying a prompt this model cannot tokenize never helps"
9160 );
9161 }
9162
9163 #[test]
9164 fn decode_error_response_maps_token_out_of_vocab_to_bad_request() {
9165 let (status, _body) = decode_error_response(generate::DecodeError::TokenOutOfVocab {
9166 token: 99,
9167 vocab_size: 32,
9168 });
9169 assert_eq!(status, StatusCode::BAD_REQUEST);
9170 }
9171
9172 #[test]
9173 fn run_generation_succeeds_and_releases_blocks_when_the_pool_has_room() {
9174 let model = test_model(); let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9176 let pool = Arc::new(Mutex::new(frink_core::cache::KvBlockPool::new(64, 2)));
9177 let config = generate::KvPoolConfig {
9178 pool: pool.clone(),
9179 queue_wait: Duration::ZERO,
9180 };
9181
9182 let produced = run_generation(
9183 &model,
9184 &prompt,
9185 &greedy_params(4),
9186 Some(&config),
9187 None,
9188 None,
9189 None,
9190 None,
9191 None,
9192 )
9193 .unwrap();
9194 assert_eq!(produced.choices[0].finish, FinishReason::Length);
9195 assert_eq!(
9196 pool.lock().unwrap().free_blocks(),
9197 2,
9198 "a completed request must return its blocks to the pool"
9199 );
9200 }
9201
9202 #[tokio::test]
9207 async fn concurrent_requests_against_the_same_model_do_not_interfere() {
9208 let model = Arc::new(test_model());
9209 let prompt = String::from_utf8(vec![1u8, 2]).unwrap();
9210
9211 let mut handles = Vec::new();
9212 for _ in 0..8 {
9213 let model = Arc::clone(&model);
9214 let prompt = prompt.clone();
9215 handles.push(tokio::task::spawn_blocking(move || {
9216 run_generation(
9217 &model,
9218 &prompt,
9219 &greedy_params(6),
9220 None,
9221 None,
9222 None,
9223 None,
9224 None,
9225 None,
9226 )
9227 .unwrap()
9228 }));
9229 }
9230
9231 let mut results = Vec::new();
9232 for h in handles {
9233 results.push(h.await.unwrap());
9234 }
9235 for r in &results[1..] {
9240 assert_eq!(r.choices, results[0].choices, "choices must match");
9244 assert_eq!(
9245 r.usage.prompt_tokens, results[0].usage.prompt_tokens,
9246 "prompt token count must match"
9247 );
9248 assert_eq!(
9249 r.usage.completion_tokens, results[0].usage.completion_tokens,
9250 "completion token count must match"
9251 );
9252 }
9253 }
9254
9255 fn write_safetensors_shard(tensors: &[(String, Vec<usize>, Vec<f32>)]) -> Vec<u8> {
9262 let mut header_entries = Vec::new();
9263 let mut data = Vec::new();
9264 for (name, shape, values) in tensors {
9265 let start = data.len();
9266 for v in values {
9267 data.extend_from_slice(&v.to_le_bytes());
9268 }
9269 let end = data.len();
9270 let shape_str = shape
9271 .iter()
9272 .map(|d| d.to_string())
9273 .collect::<Vec<_>>()
9274 .join(",");
9275 header_entries.push(format!(
9276 "\"{name}\":{{\"dtype\":\"F32\",\"shape\":[{shape_str}],\"data_offsets\":[{start},{end}]}}"
9277 ));
9278 }
9279 let header = format!("{{{}}}", header_entries.join(","));
9280 let header_bytes = header.as_bytes();
9281 let mut out = Vec::with_capacity(8 + header_bytes.len() + data.len());
9282 out.extend_from_slice(&(header_bytes.len() as u64).to_le_bytes());
9283 out.extend_from_slice(header_bytes);
9284 out.extend_from_slice(&data);
9285 out
9286 }
9287
9288 fn build_synthetic_kimi_loaded() -> model::KimiLoaded {
9299 use frink_models::config::{AttentionKind, KdaConfig, KimiHybridAttention, MlaConfig};
9300 use frink_models::kimi_loader::KimiRealHparams;
9301 use frink_moe::{GatingFunction, MoeLayerConfig};
9302
9303 let hidden_dim = 8;
9304 let kda_num_heads = 2;
9305 let kda_head_dim = 3;
9306 let kda_proj = kda_num_heads * kda_head_dim;
9307 let conv_kernel = 4;
9308 let dense_intermediate = 5;
9309 let vocab_size = 256;
9313 let mla_num_heads = 1;
9314 let mla_q_lora_rank = 2;
9315 let mla_kv_lora_rank = 2;
9316 let mla_qk_nope_head_dim = 2;
9317 let mla_qk_rope_head_dim = 2;
9318 let mla_v_head_dim = 2;
9319
9320 let model_cfg = frink_models::ModelConfig {
9321 rope_layers: frink_models::rope_layers::RopeLayers::All,
9322 layer_shapes: frink_models::layer_shapes::LayerShapes::Uniform,
9323 name: "synthetic-kimi-server-test",
9324 n_layers: 1,
9325 n_mtp_blocks: 0,
9326 hidden_dim,
9327 n_heads: 1,
9328 n_kv_heads: 1,
9329 head_dim: 4,
9330 v_head_dim: None,
9331 vocab_size,
9332 rope_theta: 10000.0,
9333 rms_norm_eps: 1e-5,
9334 post_norm_eps: 1e-5,
9335 sliding_window: None,
9336 moe: MoeLayerConfig {
9337 expert_weights_scale: 1.0,
9338 routed_weight_before_ffn: false,
9339 n_experts: 1,
9340 n_experts_active: 1,
9341 n_shared_experts: 0,
9342 hidden_dim,
9343 expert_ffn_dim: 4,
9344 gating: GatingFunction::Sigmoid,
9345 norm_topk_prob: true,
9346 expert_group_count: None,
9347 expert_group_used_count: None,
9348 },
9349 n_dense_leading_layers: 1,
9354 moe_interleave_step: None,
9355 norm_function: frink_models::norm::NormFunction::Rms,
9356 attention: AttentionKind::KimiHybrid(KimiHybridAttention {
9357 kda_layers: vec![1],
9358 full_attn_layers: vec![],
9359 mla: MlaConfig {
9360 num_heads: mla_num_heads,
9361 q_lora_rank: mla_q_lora_rank,
9362 kv_lora_rank: mla_kv_lora_rank,
9363 qk_nope_head_dim: mla_qk_nope_head_dim,
9364 qk_rope_head_dim: mla_qk_rope_head_dim,
9365 v_head_dim: mla_v_head_dim,
9366 use_output_gate: true,
9367 rope: None,
9368 },
9369 kda: KdaConfig {
9370 num_heads: kda_num_heads,
9371 head_dim: kda_head_dim,
9372 short_conv_kernel_size: conv_kernel,
9373 gate_lower_bound: -5.0,
9374 use_full_rank_gate: true,
9375 },
9376 }),
9377 rope_freqs: None,
9378 rope_attn_factor: 1.0,
9379 rope_dim: None,
9380 rope_dim_swa: None,
9381 rope_freqs_long: None,
9382 rope_freqs_short: None,
9383 rope_orig_ctx: None,
9384 rope_layout: frink_models::config::RopeLayout::Neox,
9385 qk_norm_style: frink_models::capability::QkNormStyle::WholeVector,
9386 swa_layers: frink_models::swa_layers::SwaLayers::All,
9387 attn_logit_softcap: None,
9388 final_logit_softcap: None,
9389 embedding_scale: None,
9390 residual_scale: None,
9391 normed_residual_scale: None,
9392 clamp_kqv: None,
9393 attn_temperature: None,
9394 router_input: frink_models::router_input::RouterInput::NormedFfnInput,
9395 block_sub_norms: false,
9396 parallel_residual: false,
9397 learned_positions: false,
9398 attn_value_scale: None,
9399 alibi_max_bias: None,
9400 layer_loops: None,
9401 skip_stream: false,
9402 parallel_ssm: false,
9403 swa_chunked: false,
9404 weightless_qk_norm: false,
9405 logit_multiplier: None,
9406 attention_scale: None,
9407 rope_theta_swa: None,
9408 ffn_activation: frink_models::config::FfnActivation::Swiglu,
9409 best_effort_fields: &["synthetic test config, not a real preset"],
9410 };
9411 let hp = KimiRealHparams {
9412 hidden_dim,
9413 kda_num_heads,
9414 kda_head_dim,
9415 mla_num_heads,
9416 mla_q_lora_rank,
9417 mla_kv_lora_rank,
9418 mla_qk_nope_head_dim,
9419 mla_qk_rope_head_dim,
9420 mla_v_head_dim,
9421 dense_intermediate_dim: dense_intermediate,
9422 moe_hidden_dim: hidden_dim,
9423 moe_intermediate_dim: 4,
9424 n_experts: 1,
9425 num_shared_experts: 0,
9426 };
9427
9428 let prefix = "language_model.model.layers.0";
9432 let mut tensors: Vec<(String, Vec<usize>, Vec<f32>)> = Vec::new();
9433 let push = |tensors: &mut Vec<(String, Vec<usize>, Vec<f32>)>,
9434 name: String,
9435 shape: Vec<usize>,
9436 n: usize| {
9437 tensors.push((name, shape, vec![0.01f32; n]));
9438 };
9439 push(
9440 &mut tensors,
9441 format!("{prefix}.input_layernorm.weight"),
9442 vec![hidden_dim],
9443 hidden_dim,
9444 );
9445 push(
9446 &mut tensors,
9447 format!("{prefix}.post_attention_layernorm.weight"),
9448 vec![hidden_dim],
9449 hidden_dim,
9450 );
9451 push(
9452 &mut tensors,
9453 format!("{prefix}.self_attention_res_norm.weight"),
9454 vec![hidden_dim],
9455 hidden_dim,
9456 );
9457 push(
9458 &mut tensors,
9459 format!("{prefix}.self_attention_res_proj.weight"),
9460 vec![1, hidden_dim],
9461 hidden_dim,
9462 );
9463 push(
9464 &mut tensors,
9465 format!("{prefix}.mlp_res_norm.weight"),
9466 vec![hidden_dim],
9467 hidden_dim,
9468 );
9469 push(
9470 &mut tensors,
9471 format!("{prefix}.mlp_res_proj.weight"),
9472 vec![1, hidden_dim],
9473 hidden_dim,
9474 );
9475 push(
9476 &mut tensors,
9477 format!("{prefix}.self_attn.q_proj.weight"),
9478 vec![kda_proj, hidden_dim],
9479 kda_proj * hidden_dim,
9480 );
9481 push(
9482 &mut tensors,
9483 format!("{prefix}.self_attn.k_proj.weight"),
9484 vec![kda_proj, hidden_dim],
9485 kda_proj * hidden_dim,
9486 );
9487 push(
9488 &mut tensors,
9489 format!("{prefix}.self_attn.v_proj.weight"),
9490 vec![kda_proj, hidden_dim],
9491 kda_proj * hidden_dim,
9492 );
9493 push(
9494 &mut tensors,
9495 format!("{prefix}.self_attn.q_conv1d.weight"),
9496 vec![kda_proj, 1, conv_kernel],
9497 kda_proj * conv_kernel,
9498 );
9499 push(
9500 &mut tensors,
9501 format!("{prefix}.self_attn.k_conv1d.weight"),
9502 vec![kda_proj, 1, conv_kernel],
9503 kda_proj * conv_kernel,
9504 );
9505 push(
9506 &mut tensors,
9507 format!("{prefix}.self_attn.v_conv1d.weight"),
9508 vec![kda_proj, 1, conv_kernel],
9509 kda_proj * conv_kernel,
9510 );
9511 push(
9512 &mut tensors,
9513 format!("{prefix}.self_attn.A_log"),
9514 vec![kda_num_heads],
9515 kda_num_heads,
9516 );
9517 push(
9518 &mut tensors,
9519 format!("{prefix}.self_attn.f_a_proj.weight"),
9520 vec![kda_head_dim, hidden_dim],
9521 kda_head_dim * hidden_dim,
9522 );
9523 push(
9524 &mut tensors,
9525 format!("{prefix}.self_attn.f_b_proj.weight"),
9526 vec![kda_proj, kda_head_dim],
9527 kda_proj * kda_head_dim,
9528 );
9529 push(
9530 &mut tensors,
9531 format!("{prefix}.self_attn.dt_bias"),
9532 vec![kda_proj],
9533 kda_proj,
9534 );
9535 push(
9536 &mut tensors,
9537 format!("{prefix}.self_attn.b_proj.weight"),
9538 vec![kda_num_heads, hidden_dim],
9539 kda_num_heads * hidden_dim,
9540 );
9541 push(
9542 &mut tensors,
9543 format!("{prefix}.self_attn.g_proj.weight"),
9544 vec![kda_proj, hidden_dim],
9545 kda_proj * hidden_dim,
9546 );
9547 push(
9548 &mut tensors,
9549 format!("{prefix}.self_attn.o_norm.weight"),
9550 vec![kda_head_dim],
9551 kda_head_dim,
9552 );
9553 push(
9554 &mut tensors,
9555 format!("{prefix}.self_attn.o_proj.weight"),
9556 vec![hidden_dim, kda_proj],
9557 hidden_dim * kda_proj,
9558 );
9559 push(
9560 &mut tensors,
9561 format!("{prefix}.mlp.gate_proj.weight"),
9562 vec![dense_intermediate, hidden_dim],
9563 dense_intermediate * hidden_dim,
9564 );
9565 push(
9566 &mut tensors,
9567 format!("{prefix}.mlp.up_proj.weight"),
9568 vec![dense_intermediate, hidden_dim],
9569 dense_intermediate * hidden_dim,
9570 );
9571 push(
9572 &mut tensors,
9573 format!("{prefix}.mlp.down_proj.weight"),
9574 vec![hidden_dim, dense_intermediate],
9575 hidden_dim * dense_intermediate,
9576 );
9577 push(
9578 &mut tensors,
9579 "language_model.model.embed_tokens.weight".to_string(),
9580 vec![vocab_size, hidden_dim],
9581 vocab_size * hidden_dim,
9582 );
9583 push(
9584 &mut tensors,
9585 "language_model.lm_head.weight".to_string(),
9586 vec![vocab_size, hidden_dim],
9587 vocab_size * hidden_dim,
9588 );
9589 push(
9590 &mut tensors,
9591 "language_model.model.norm.weight".to_string(),
9592 vec![hidden_dim],
9593 hidden_dim,
9594 );
9595 push(
9596 &mut tensors,
9597 "language_model.model.output_attn_res_norm.weight".to_string(),
9598 vec![hidden_dim],
9599 hidden_dim,
9600 );
9601 push(
9602 &mut tensors,
9603 "language_model.model.output_attn_res_proj.weight".to_string(),
9604 vec![1, hidden_dim],
9605 hidden_dim,
9606 );
9607
9608 static FIXTURE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
9621 let dir = std::env::temp_dir().join(format!(
9622 "frink_server_kimi_e2e_test_{}_{}",
9623 std::process::id(),
9624 FIXTURE.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
9625 ));
9626 std::fs::create_dir_all(&dir).unwrap();
9627 let shard_bytes = write_safetensors_shard(&tensors);
9628 std::fs::write(dir.join("shard0.safetensors"), &shard_bytes).unwrap();
9629 let map_entries: Vec<String> = tensors
9630 .iter()
9631 .map(|(name, ..)| format!("\"{name}\":\"shard0.safetensors\""))
9632 .collect();
9633 let index = format!("{{\"weight_map\":{{{}}}}}", map_entries.join(","));
9634 std::fs::write(dir.join("model.safetensors.index.json"), &index).unwrap();
9635
9636 use base64::Engine;
9640 let vocab_lines: Vec<String> = (0..vocab_size as u32)
9641 .map(|b| {
9642 let b64 = base64::engine::general_purpose::STANDARD.encode([b as u8]);
9643 format!("{b64} {b}")
9644 })
9645 .collect();
9646 std::fs::write(dir.join("tiktoken.model"), vocab_lines.join("\n")).unwrap();
9647
9648 let loaded = model::load_kimi_checkpoint_with_config(dir.to_str().unwrap(), model_cfg, hp)
9649 .expect("must load the synthetic Kimi checkpoint end to end");
9650 std::fs::remove_dir_all(&dir).ok();
9651 loaded
9652 }
9653
9654 #[test]
9662 fn kimi_model_serves_real_text_end_to_end_via_run_generation() {
9663 let loaded = build_synthetic_kimi_loaded();
9664 let state = build_app_state(
9665 StartupModels {
9666 loaded: model::LoadedModel::Kimi(loaded),
9667 embedding: None,
9668 },
9669 None,
9670 None,
9671 None,
9672 false,
9673 None,
9674 Arc::new(health::Detection::ready(health::probe_backends())),
9675 );
9676 let active = state.active().expect("a freshly built state has a model");
9677 assert_eq!(active.tokenizer_kind(), "kimi-tiktoken-bpe");
9678 assert!(!active.is_synthetic());
9679
9680 let produced = run_generation(
9681 active.generative().unwrap(),
9682 "hi",
9683 &greedy_params(5),
9684 None,
9685 None,
9686 None,
9687 None,
9688 None,
9689 None,
9690 )
9691 .expect("a real Kimi checkpoint must generate without error");
9692 assert!(matches!(
9693 produced.choices[0].finish,
9694 FinishReason::Length | FinishReason::Stop
9695 ));
9696 }
9697
9698 #[test]
9711 fn a_grammar_constrains_the_engine_decode_path() {
9712 let loaded = build_synthetic_kimi_loaded();
9713 let state = build_app_state(
9714 StartupModels {
9715 loaded: model::LoadedModel::Kimi(loaded),
9716 embedding: None,
9717 },
9718 None,
9719 None,
9720 None,
9721 false,
9722 None,
9723 Arc::new(health::Detection::ready(health::probe_backends())),
9724 );
9725 let active = state.active().expect("a freshly built state has a model");
9726
9727 let run = |grammar: Option<&str>| {
9728 let mut params = greedy_params(6);
9729 params.grammar = grammar.map(|src| {
9730 Arc::new(
9731 frink_models::grammar::Grammar::from_str_with_root(src, "root")
9732 .expect("test grammar parses"),
9733 )
9734 });
9735 run_generation(
9736 active.generative().unwrap(),
9737 "hi",
9738 ¶ms,
9739 None,
9740 None,
9741 None,
9742 None,
9743 None,
9744 None,
9745 )
9746 };
9747
9748 let produced = run(None).expect("the unconstrained run must serve");
9749 let unconstrained = produced.choices[0].text.clone();
9750 assert!(
9751 unconstrained.chars().any(|c| c != 'a'),
9752 "the unconstrained run produced only `a` ({unconstrained:?}), so the \
9753 constrained run below would prove nothing"
9754 );
9755
9756 let produced =
9757 run(Some(r#"root ::= "a"+"#)).expect("a grammar this vocabulary can spell must serve");
9758 let one = produced.choices.into_iter().next().unwrap();
9759 let (finish, constrained) = (one.finish, one.text);
9760 assert!(
9761 !constrained.is_empty() && constrained.chars().all(|c| c == 'a'),
9762 "the engine decode path served text its grammar forbids ({constrained:?}): \
9763 the constraint was dropped between `generate_engine` and the sampler"
9764 );
9765 assert!(matches!(finish, FinishReason::Length | FinishReason::Stop));
9766 }
9767
9768 #[test]
9778 fn kv_pool_and_prefix_cache_are_never_consulted_for_a_kimi_model() {
9779 let loaded = build_synthetic_kimi_loaded();
9780 let state = build_app_state(
9781 StartupModels {
9782 loaded: model::LoadedModel::Kimi(loaded),
9783 embedding: None,
9784 },
9785 None,
9786 None,
9787 None,
9788 false,
9789 None,
9790 Arc::new(health::Detection::ready(health::probe_backends())),
9791 );
9792
9793 let pool = Arc::new(Mutex::new(frink_core::cache::KvBlockPool::new(64, 4)));
9794 let kv_pool_config = generate::KvPoolConfig {
9795 pool: pool.clone(),
9796 queue_wait: Duration::ZERO,
9797 };
9798 let pc = Mutex::new(PrefixCache::new(4));
9799
9800 run_generation(
9801 state
9802 .active()
9803 .expect("a freshly built state has a model")
9804 .generative()
9805 .unwrap(),
9806 "hi",
9807 &greedy_params(5),
9808 Some(&kv_pool_config),
9809 None,
9810 Some(&pc),
9811 None,
9812 None,
9813 None,
9814 )
9815 .expect("a real Kimi checkpoint must generate without error");
9816
9817 assert_eq!(
9818 pool.lock().unwrap().free_blocks(),
9819 4,
9820 "the KV pool must be completely untouched by a Kimi request"
9821 );
9822 let stats = pc.lock().unwrap().stats();
9823 assert_eq!(
9824 stats.hits + stats.misses,
9825 0,
9826 "the prefix cache must never be consulted for a Kimi request"
9827 );
9828 }
9829}