From 19f666e7b1af32a45f1f7485a37c445b1024aef2 Mon Sep 17 00:00:00 2001 From: infosave2007 Date: Sat, 8 Aug 2026 15:52:20 +0300 Subject: [PATCH] feat(cmf): auto-detect tool support of managed local servers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cortiq serve >=0.5.62 advertises the loaded model's capabilities in /healthz ({"capabilities":{"tools":true}} when the chat template has a tools branch). The gateway now reads that handshake when the managed server becomes healthy and treats detected tool support as equivalent to declared caps in the candidate filter — so a low-tier chain made of a local model no longer widens agent (tools) traffic to cloud models once the local server can actually drive tools. Also adds an optional manual "caps" override to [[cmf.servers]] for older serve builds that do not advertise capabilities. --- src/admin/mod.rs | 1 + src/cmf_runtime.rs | 32 +++++++++++++++++++++++++++----- src/config.rs | 5 +++++ src/pipeline.rs | 20 +++++++++++++++----- src/registry.rs | 2 +- 5 files changed, 49 insertions(+), 11 deletions(-) diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 96bcb2d..a63fe8c 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -2197,6 +2197,7 @@ fn register_cmf_output(cfg: &mut crate::config::Config, output: &str) -> Option< o1_window: None, o1_sink: None, skip_mtp: false, + caps: Vec::new(), }); Some(id) } diff --git a/src/cmf_runtime.rs b/src/cmf_runtime.rs index 1e3911e..fa70555 100644 --- a/src/cmf_runtime.rs +++ b/src/cmf_runtime.rs @@ -56,6 +56,9 @@ struct Managed { pub struct CmfRuntime { status: Mutex, children: Mutex>, + /// Server ids whose healthz handshake advertised tool support — lets the + /// pipeline route tools traffic to local models with no manual caps. + pub detected_tools: Mutex>, } impl CmfRuntime { @@ -63,6 +66,7 @@ impl CmfRuntime { Arc::new(Self { status: Mutex::new(CmfStatus::default()), children: Mutex::new(Vec::new()), + detected_tools: Mutex::new(std::collections::HashSet::new()), }) } @@ -193,6 +197,7 @@ impl Default for CmfRuntime { Self { status: Mutex::new(CmfStatus::default()), children: Mutex::new(Vec::new()), + detected_tools: Mutex::new(std::collections::HashSet::new()), } } } @@ -332,8 +337,11 @@ fn spawn_serve(server: &CmfServer, cfg: &CmfCfg) -> Result { .map_err(|e| e.to_string()) } -/// Poll `/healthz` until the local server answers 200 (or the deadline passes). -async fn wait_healthy(host: &str, port: u16, tries: u32) -> bool { +/// Poll `/healthz` until the local server answers 200 (or the deadline +/// passes). Newer `cortiq serve` advertises the loaded model's capabilities +/// in the healthz body ({"capabilities":{"tools":true}}) — returned so the +/// caller can register them; older servers return `None` there. +async fn wait_healthy(host: &str, port: u16, tries: u32) -> (bool, Option) { let url = format!("http://{host}:{port}/healthz"); let client = reqwest::Client::new(); for _ in 0..tries { @@ -344,12 +352,17 @@ async fn wait_healthy(host: &str, port: u16, tries: u32) -> bool { .await { if r.status().is_success() { - return true; + let tools = r + .json::() + .await + .ok() + .and_then(|v| v["capabilities"]["tools"].as_bool()); + return (true, tools); } } tokio::time::sleep(Duration::from_millis(500)).await; } - false + (false, None) } /// Orchestrate the whole lifecycle: ensure installed → maybe update → spawn each @@ -432,8 +445,17 @@ pub async fn manage(rt: Arc, cfg: CmfCfg) { if !rt.is_running(&server.id) { continue; // failed to spawn / file missing } - if wait_healthy(&cfg.local_host, server.port, 240).await { + let (ok, tools) = wait_healthy(&cfg.local_host, server.port, 240).await; + if ok { rt.set_healthy(&server.id, true); + // capability handshake: the server said whether its model's chat + // template can drive tool calling — no manual caps needed + if tools == Some(true) { + rt.detected_tools.lock().unwrap().insert(server.id.clone()); + tracing::info!(id = %server.id, "local server advertises tool support"); + } else { + rt.detected_tools.lock().unwrap().remove(&server.id); + } rt.log(format!( "local CMF server ready — '{}' at http://{}:{}/v1", server.id, cfg.local_host, server.port diff --git a/src/config.rs b/src/config.rs index 9aff363..75cc5d0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -130,6 +130,10 @@ pub struct CmfServer { /// Disable Multi-Token Prediction speculative decoding (CMF_MTP=0). #[serde(default)] pub skip_mtp: bool, + /// Manual capability override (e.g. ["tools"]). Usually unnecessary: the + /// gateway auto-detects tool support from the server's healthz handshake. + #[serde(default)] + pub caps: Vec, } /// CMF-format model factory: import a HuggingFace model, convert it to a @@ -229,6 +233,7 @@ impl CmfCfg { o1_window: None, o1_sink: None, skip_mtp: false, + caps: Vec::new(), }]; } Vec::new() diff --git a/src/pipeline.rs b/src/pipeline.rs index e979372..6478e8a 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -183,6 +183,9 @@ impl Pipeline { &self, req: &ChatRequest, live: &crate::state::Live, + // Tool support detected via the local servers' healthz handshake — + // treated as equivalent to declared caps when filtering candidates. + detected_tools: &std::collections::HashSet, ) -> Result<(RouteDecision, String, Vec)> { // 1. Extract text for classification let text = self.extract_text( @@ -274,10 +277,13 @@ impl Pipeline { // Filter candidates by capabilities (e.g. tools support) let needs_tools = !req.tools.is_empty(); + let has_tools = |id: &str, prov: &std::sync::Arc| { + prov.caps().tools || detected_tools.contains(id) + }; let mut final_candidates = Vec::new(); for model_id in candidates { if let Some(prov) = live.registry.get(&model_id) { - if needs_tools && !prov.caps().tools { + if needs_tools && !has_tools(&model_id, &prov) { tracing::debug!( "Candidate '{}' skipped: lacks support for requested tools", model_id @@ -306,7 +312,7 @@ impl Pipeline { if final_candidates.is_empty() && live.cfg.cmf.manage_server { for id in live.local_candidates() { if let Some(prov) = live.registry.get(&id) { - if needs_tools && !prov.caps().tools { + if needs_tools && !has_tools(&id, &prov) { continue; } final_candidates.push(id); @@ -336,7 +342,7 @@ impl Pipeline { } for id in widened { if let Some(prov) = live.registry.get(&id) { - if needs_tools && !prov.caps().tools { + if needs_tools && !has_tools(&id, &prov) { continue; } final_candidates.push(id); @@ -360,7 +366,9 @@ impl Pipeline { async fn run_inner(&self, req: ChatRequest, state: &SharedState) -> Result { let live = state.live(); - let (decision, route_source, mut final_candidates) = self.resolve(&req, &live).await?; + let detected_tools = state.cmf.detected_tools.lock().unwrap().clone(); + let (decision, route_source, mut final_candidates) = + self.resolve(&req, &live, &detected_tools).await?; // Serving-when-promoted (gated): a promoted task-label is served by the // local model FIRST (failover to cloud on error). Per-request VETO uses @@ -507,7 +515,9 @@ impl Pipeline { .to_string(); let live = state.live(); - let (decision, route_source, final_candidates) = match self.resolve(&req, &live).await { + let detected_tools = state.cmf.detected_tools.lock().unwrap().clone(); + let (decision, route_source, final_candidates) = + match self.resolve(&req, &live, &detected_tools).await { Ok(t) => t, Err(e) => { record_error(state, &account, &protocol, &directive, started, &e); diff --git a/src/registry.rs b/src/registry.rs index 2c2689b..0d593b9 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -48,7 +48,7 @@ impl Registry { price_out: 0.0, kind: "chat".into(), api_key_env: None, - caps: Vec::new(), + caps: s.caps.clone(), temperature: s.temperature, top_p: s.top_p, max_tokens: s.max_tokens,