From 19ade8d781cc03158aed119fe7314193ff92beaf Mon Sep 17 00:00:00 2001 From: Jimmy <1fe240cd1a8cf775f6f3060f115e5a303181f3abf28ad4cb0c2515f4a02b36a8@meshllm.communities.buzz.xyz> Date: Thu, 29 Oct 2026 16:07:34 +1100 Subject: [PATCH] fix: support native tool calls and immediate result replay --- README.md | 19 +++++- src/http.rs | 174 +++++++++++++++++++++++++++++++++++++++++++++++-- src/native.rs | 60 ++++++++++++++--- src/request.rs | 49 ++++++++++++++ src/worker.rs | 63 +++++++++++++++--- 5 files changed, 339 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index a27bfde..6f63788 100644 --- a/README.md +++ b/README.md @@ -4,8 +4,23 @@ Run a local MLX model on your Apple Silicon Mac with [Eredu](https://github.com/jbg/eredu), served through Mesh's normal OpenAI-compatible API. -**Early access:** Apple Silicon (macOS arm64) only. Text chat and streaming; -tool calls are rejected for now. +**Early access:** Apple Silicon (macOS arm64) only. Text chat, streaming and +checkpoint-native function calls through OpenAI JSON/SSE. Tools require an Eredu +registered native-tool template; unsupported templates fail closed. + +Tool requests support `tools`, `tool_choice` (`auto`, `none`, `required`), +`parallel_tool_calls` (default false), and `enable_thinking`. Named-function +`tool_choice` objects are not supported and are rejected, not silently ignored. +The plugin returns calls for your client to execute; it never executes tools. +Replay the assistant message and a `role: "tool"` message with its `tool_call_id` +to supply results. Function arguments on the wire are JSON strings. + +Validated locally with an existing `mlx-community/Qwen3-1.7B-4bit` checkpoint: +required tool call, result replay with `tool_choice: "none"` and a nonempty +follow-up, plus streaming tool-call output (`enable_thinking: false`). This is +single-node adapter evidence, not qualification of every model, parallel tool +execution, agent harness, or remote Mesh routing. Readiness does not advertise +blanket model tool capability. ## 1. Install diff --git a/src/http.rs b/src/http.rs index 6081390..13af6c5 100644 --- a/src/http.rs +++ b/src/http.rs @@ -71,6 +71,8 @@ async fn chat(State(state): State, Json(request): Json) - let mut terminal = false; while let Some(event) = generation.events.recv().await { let payload = match event { + Event::ToolStart { index, id: call_id, name } => chunk(&id, &model, json!({"tool_calls":[{"index":index,"id":call_id,"type":"function","function":{"name":name,"arguments":""}}]}), None), + Event::ToolArguments { index, fragment } => chunk(&id, &model, json!({"tool_calls":[{"index":index,"function":{"arguments":fragment}}]}), None), Event::Text(text) => chunk(&id, &model, json!({"content":text}), None), Event::Finished(reason) => { terminal = true; chunk(&id, &model, json!({}), Some(reason)) }, Event::Error(message) => { terminal = true; json!({"error":{"message":message}}) }, @@ -84,15 +86,49 @@ async fn chat(State(state): State, Json(request): Json) - return Sse::new(stream).into_response(); } let mut text = String::new(); + let mut calls: Vec = Vec::new(); + let mut output_bytes = 0usize; while let Some(event) = generation.events.recv().await { match event { + Event::ToolStart { index, id, name } => { + if index != calls.len() || calls.len() >= 128 { + return error(StatusCode::BAD_GATEWAY, "invalid tool call index".into()); + } + output_bytes += id.len() + name.len(); + calls.push( + json!({"id":id,"type":"function","function":{"name":name,"arguments":""}}), + ); + } + Event::ToolArguments { index, fragment } => { + let Some(call) = calls.get_mut(index) else { + return error( + StatusCode::BAD_GATEWAY, + "arguments without tool call".into(), + ); + }; + output_bytes += fragment.len(); + let args = call["function"]["arguments"].as_str().unwrap().to_owned() + &fragment; + call["function"]["arguments"] = Value::String(args); + } Event::Text(delta) => { - if text.len() + delta.len() > 1024 * 1024 { return error(StatusCode::BAD_GATEWAY, "output exceeds limit".into()); } + output_bytes += delta.len(); text.push_str(&delta); } - Event::Finished(reason) => return Json(json!({"id":id,"object":"chat.completion","model":model,"choices":[{"index":0,"message":{"role":"assistant","content":text},"finish_reason":reason}]})).into_response(), + Event::Finished(reason) => { + let mut message = json!({"role":"assistant","content":text}); + if !calls.is_empty() { + if text.is_empty() { + message["content"] = Value::Null; + } + message["tool_calls"] = json!(calls); + } + return Json(json!({"id":id,"object":"chat.completion","model":model,"choices":[{"index":0,"message":message,"finish_reason":reason}]})).into_response(); + } Event::Error(message) => return error(StatusCode::BAD_GATEWAY, message), } + if output_bytes > 1024 * 1024 { + return error(StatusCode::BAD_GATEWAY, "output exceeds limit".into()); + } } error( StatusCode::BAD_GATEWAY, @@ -123,9 +159,9 @@ mod tests { _: ChatRequest, _: &ActiveWork, output: &tokio::sync::mpsc::Sender, - ) -> Result<()> { + ) -> Result<&'static str> { publish(output, Event::Text("héllo\nworld".into()))?; - publish(output, Event::Finished("stop")) + Ok("stop") } fn settle(&mut self) -> Result<()> { Ok(()) @@ -163,4 +199,134 @@ mod tests { } assert!(frames[frames.len() - 2].contains("\"finish_reason\":\"stop\"")); } + struct ToolMock; + impl Backend for ToolMock { + fn generate( + &mut self, + request: ChatRequest, + _: &ActiveWork, + output: &tokio::sync::mpsc::Sender, + ) -> Result<&'static str> { + if request.messages.iter().any(|m| m["role"] == "tool") { + assert_eq!(request.messages[1]["tool_calls"][0]["id"], "call_7"); + assert_eq!( + request.messages[1]["tool_calls"][0]["function"]["arguments"], + json!({"city":"Bogotá"}) + ); + assert_eq!(request.messages[2]["tool_call_id"], "call_7"); + publish(output, Event::Text("Sunny in Bogotá".into()))?; + return Ok("stop"); + } + assert_eq!(request.tool_choice.as_deref(), Some("required")); + assert_eq!(request.tools[0]["function"]["name"], "weather"); + publish( + output, + Event::ToolStart { + index: 0, + id: "call_7".into(), + name: "weather".into(), + }, + )?; + publish( + output, + Event::ToolArguments { + index: 0, + fragment: "{\"city\":".into(), + }, + )?; + publish( + output, + Event::ToolArguments { + index: 0, + fragment: "\"Bogotá\"}".into(), + }, + )?; + Ok("tool_calls") + } + fn settle(&mut self) -> Result<()> { + Ok(()) + } + } + async fn tool_call(body: Value) -> (StatusCode, String) { + let worker = Worker::start(|| Ok(ToolMock)).await.unwrap(); + let app = router(Endpoint { + worker, + model: "local".into(), + }); + let response = app + .oneshot( + Request::post("/v1/chat/completions") + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), 65536).await.unwrap(); + (status, String::from_utf8(bytes.to_vec()).unwrap()) + } + fn tool_request(stream: bool) -> Value { + json!({"model":"local","stream":stream,"messages":[{"role":"user","content":"Weather?"}], + "tool_choice":"required","tools":[{"type":"function","function":{"name":"weather","parameters":{"type":"object"}}}]}) + } + #[tokio::test] + async fn tool_json_and_result_replay_round_trip() { + let (status, body) = tool_call(tool_request(false)).await; + assert_eq!(status, StatusCode::OK); + let value: Value = serde_json::from_str(&body).unwrap(); + let message = value["choices"][0]["message"].clone(); + assert!(message["content"].is_null()); + assert_eq!(value["choices"][0]["finish_reason"], "tool_calls"); + assert_eq!( + message["tool_calls"][0]["function"]["arguments"], + "{\"city\":\"Bogotá\"}" + ); + let (status, body) = tool_call(json!({"model":"local","messages":[ + {"role":"user","content":"Weather?"}, message, + {"role":"tool","tool_call_id":"call_7","content":"sunny"}]})) + .await; + assert_eq!(status, StatusCode::OK); + let answer: Value = serde_json::from_str(&body).unwrap(); + assert_eq!( + answer["choices"][0]["message"]["content"], + "Sunny in Bogotá" + ); + } + #[tokio::test] + async fn tool_sse_fragments_reconstruct_openai_call() { + let (status, body) = tool_call(tool_request(true)).await; + assert_eq!(status, StatusCode::OK); + let mut arguments = String::new(); + let mut id = String::new(); + let mut name = String::new(); + let mut finish = String::new(); + for frame in body.split("\n\n").filter_map(|f| f.strip_prefix("data: ")) { + if frame == "[DONE]" { + continue; + } + let v: Value = serde_json::from_str(frame).unwrap(); + let call = &v["choices"][0]["delta"]["tool_calls"][0]; + if !call.is_null() { + assert_eq!(call["index"], 0); + if let Some(s) = call["id"].as_str() { + id.push_str(s); + } + if let Some(s) = call["function"]["name"].as_str() { + name.push_str(s); + } + if let Some(s) = call["function"]["arguments"].as_str() { + arguments.push_str(s); + } + } + if let Some(s) = v["choices"][0]["finish_reason"].as_str() { + finish = s.into(); + } + } + assert_eq!(id, "call_7"); + assert_eq!(name, "weather"); + assert_eq!(arguments, "{\"city\":\"Bogotá\"}"); + assert_eq!(finish, "tool_calls"); + assert!(body.ends_with("data: [DONE]\n\n")); + } } diff --git a/src/native.rs b/src/native.rs index f5e8dba..e89a693 100644 --- a/src/native.rs +++ b/src/native.rs @@ -9,7 +9,7 @@ use eredu::{ LoadedModel, PreparedChatGenerationRequest, PreparedChatGenerationSettings, PreparedChatInput, default_local_device, local_device_plan, }, - runtime::chat::ChatTemplateRequest, + runtime::chat::{ChatTemplateRequest, NativeToolSupport, ParallelToolCallPolicy, ToolChoice}, }; use eredu_backend_mlx::MlxBackendFactory; use eredu_core::ExecutionPlanBackendFactory; @@ -37,21 +37,38 @@ impl Backend for Native { request: ChatRequest, work: &ActiveWork, output: &mpsc::Sender, - ) -> Result<()> { - if !request.tools.is_empty() + ) -> Result<&'static str> { + let needs_tools = !request.tools.is_empty() || request .messages .iter() - .any(|m| m.get("tool_calls").is_some() || m["role"] == "tool") - { - bail!("native tool calls are not yet validated by this adapter"); - } + .any(|m| m.get("tool_calls").is_some() || m["role"] == "tool"); self.0.reset()?; + let tool_choice = match request.tool_choice.as_deref() { + Some("none") => ToolChoice::None, + Some("required") => ToolChoice::Required, + _ => ToolChoice::Auto, + }; let prepared = self.0.prepare_chat(ChatTemplateRequest { messages: request.messages, + tools: request.tools, + tool_choice, + parallel_tool_calls: if request.parallel_tool_calls { + ParallelToolCallPolicy::Enabled { max_calls: None } + } else { + ParallelToolCallPolicy::Disabled + }, + enable_thinking: request.enable_thinking, add_generation_prompt: true, ..Default::default() })?; + if needs_tools + && let NativeToolSupport::Unsupported { reason } = prepared.native_tool_support() + { + bail!("native tools unavailable for this template: {reason}"); + } + let mut started_calls = 0usize; + let mut ended_calls = 0usize; let cancellation = GenerationCancellationToken::new(); let mut delivery_failed = false; let result = self @@ -74,8 +91,27 @@ impl Backend for Native { cancellation.cancel(); return; } - if let SemanticEvent::TextDelta(text) = event - && publish(output, Event::Text(text)).is_err() + let event = match event { + SemanticEvent::TextDelta(text) => Some(Event::Text(text)), + SemanticEvent::ToolCallStart { index, id, name } => { + started_calls += 1; + Some(Event::ToolStart { index, id, name }) + } + SemanticEvent::ToolArgumentsDelta { + index, + json_fragment, + } => Some(Event::ToolArguments { + index, + fragment: json_fragment, + }), + SemanticEvent::ToolCallEnd => { + ended_calls += 1; + None + } + _ => None, + }; + if let Some(event) = event + && publish(output, event).is_err() { delivery_failed = true; cancellation.cancel(); @@ -85,12 +121,16 @@ impl Backend for Native { if delivery_failed || work.is_cancelled() { bail!("generation cancelled by consumer"); } + if started_calls != ended_calls { + bail!("generation ended with incomplete tool calls"); + } let reason = match result.finish_reason { FinishReason::MaxTokens => "length", FinishReason::Cancelled => bail!("generation cancelled"), + _ if started_calls > 0 => "tool_calls", _ => "stop", }; - publish(output, Event::Finished(reason)) + Ok(reason) } fn settle(&mut self) -> Result<()> { diff --git a/src/request.rs b/src/request.rs index 2ee7896..c500c29 100644 --- a/src/request.rs +++ b/src/request.rs @@ -11,6 +11,11 @@ pub struct ChatRequest { pub stream: bool, #[serde(default)] pub tools: Vec, + #[serde(default)] + pub tool_choice: Option, + #[serde(default)] + pub parallel_tool_calls: bool, + pub enable_thinking: Option, pub max_tokens: Option, pub temperature: Option, } @@ -29,6 +34,30 @@ impl ChatRequest { if self.temperature.is_some_and(|t| !t.is_finite() || t < 0.0) { bail!("temperature must be finite and nonnegative"); } + if self.tools.len() > 128 { + bail!("at most 128 tools are supported"); + } + if let Some(choice) = self.tool_choice.as_deref() { + if !matches!(choice, "auto" | "none" | "required") { + bail!("tool_choice must be auto, none or required; named selection is unsupported"); + } + if choice == "required" && self.tools.is_empty() { + bail!("tool_choice required needs tools"); + } + } + for tool in &self.tools { + if tool["type"] != "function" + || tool + .pointer("/function/name") + .and_then(Value::as_str) + .is_none_or(str::is_empty) + || !tool + .pointer("/function/parameters") + .is_some_and(Value::is_object) + { + bail!("tools require function name and JSON schema parameters"); + } + } for message in &mut self.messages { normalize_message(message)?; } @@ -99,4 +128,24 @@ mod tests { let mut message = json!({"role":"tool", "content":"sunny"}); assert!(normalize_message(&mut message).is_err()); } + #[test] + fn validates_tool_policy_and_schemas() { + for policy in [ + json!("bogus"), + json!({"type":"function","function":{"name":"weather"}}), + json!("required"), + ] { + let request: Result = serde_json::from_value( + json!({"model":"local","messages":[{"role":"user","content":"hi"}],"tool_choice":policy}), + ); + assert!( + request + .map_err(anyhow::Error::from) + .and_then(|r| r.prepare("local")) + .is_err() + ); + } + let request: ChatRequest = serde_json::from_value(json!({"model":"local","messages":[{"role":"user","content":"hi"}],"tools":[{"type":"function","function":{"name":"x"}}]})).unwrap(); + assert!(request.prepare("local").is_err()); + } } diff --git a/src/worker.rs b/src/worker.rs index 8a947b5..bc31e27 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -8,6 +8,15 @@ use tokio::sync::mpsc; pub enum Event { Text(String), + ToolStart { + index: usize, + id: String, + name: String, + }, + ToolArguments { + index: usize, + fragment: String, + }, Finished(&'static str), Error(String), } @@ -18,7 +27,7 @@ pub trait Backend: 'static { request: ChatRequest, work: &ActiveWork, output: &mpsc::Sender, - ) -> Result<()>; + ) -> Result<&'static str>; /// Must establish device completion before the request lease can be released. fn settle(&mut self) -> Result<()>; } @@ -61,13 +70,21 @@ impl Worker { } while let Some(job) = input.blocking_recv() { let result = backend.generate(job.request, &job.work, &job.output); - // Settlement failure poisons the worker: never reuse native state. - if let Err(error) = backend.settle() { - let _ = job.output.try_send(Event::Error(error.to_string())); - break; - } - if let Err(error) = result { - let _ = job.output.try_send(Event::Error(error.to_string())); + // Hold the final event until settlement releases admission. A client may + // immediately submit its tool result after observing completion. + let settled = backend.settle(); + drop(job.work); + match (result, settled) { + (_, Err(error)) => { + let _ = job.output.try_send(Event::Error(error.to_string())); + break; + } + (Err(error), Ok(())) => { + let _ = job.output.try_send(Event::Error(error.to_string())); + } + (Ok(reason), Ok(())) => { + let _ = job.output.try_send(Event::Finished(reason)); + } } } })?; @@ -128,13 +145,13 @@ mod tests { _: ChatRequest, work: &ActiveWork, output: &mpsc::Sender, - ) -> Result<()> { + ) -> Result<&'static str> { while !work.is_cancelled() { if publish(output, Event::Text("hello".into())).is_err() { break; } } - Ok(()) + Ok("stop") } fn settle(&mut self) -> Result<()> { self.0.store(true, Ordering::Release); @@ -165,4 +182,30 @@ mod tests { .is_err() ); } + struct Immediate; + impl Backend for Immediate { + fn generate( + &mut self, + _: ChatRequest, + _: &ActiveWork, + _: &mpsc::Sender, + ) -> Result<&'static str> { + Ok("stop") + } + fn settle(&mut self) -> Result<()> { + Ok(()) + } + } + #[tokio::test] + async fn completion_releases_admission_before_immediate_followup() { + let worker = Worker::start(|| Ok(Immediate)).await.unwrap(); + for _ in 0..100 { + let mut generation = worker.submit(request()).unwrap(); + assert!(matches!( + generation.events.recv().await, + Some(Event::Finished("stop")) + )); + // Do not wait for channel closure before starting the next request. + } + } }