hugo/DS4Server
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

Split Rust code into domain modules

Browse Source
This commit is contained in:
Georg Bauer
2026-07-25 11:22:59 +02:00
parent 7843592400
commit 4292e0d3f9
22 changed files with 7650 additions and 7536 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
src/server/http.rs Normal file
View File

@@ -0,0 +1,178 @@
use super::*;
pub(super) fn send_sse_headers(stream: &mut impl Write) -> Result<(), String> {
stream
.write_all(
b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: close\r\n\r\n",
)
.map_err(|error| error.to_string())
}
pub(super) fn send_sse(stream: &mut impl Write, value: &Value) -> Result<(), String> {
let body = serde_json::to_vec(value).map_err(|error| error.to_string())?;
stream
.write_all(b"data: ")
.map_err(|error| error.to_string())?;
stream.write_all(&body).map_err(|error| error.to_string())?;
stream.write_all(b"\n\n").map_err(|error| error.to_string())
}
pub(super) fn send_sse_error(stream: &mut impl Write, message: &str) -> Result<(), String> {
let body = serde_json::to_vec(&json!({"error": {"message": message, "type": "server_error"}}))
.map_err(|error| error.to_string())?;
stream
.write_all(b"event: error\ndata: ")
.map_err(|error| error.to_string())?;
stream.write_all(&body).map_err(|error| error.to_string())?;
stream.write_all(b"\n\n").map_err(|error| error.to_string())
}
pub(super) fn send_json(stream: &mut TcpStream, code: u16, value: &Value) -> Result<(), String> {
let mut body = serde_json::to_vec(value).map_err(|error| error.to_string())?;
body.push(b'\n');
send_response(stream, code, Some("application/json"), &body)
}
pub(super) fn send_error(stream: &mut TcpStream, code: u16, message: &str) -> Result<(), String> {
send_json(
stream,
code,
&json!({"error": {"message": message, "type": "invalid_request_error"}}),
)
}
pub(super) fn send_response(
stream: &mut TcpStream,
code: u16,
content_type: Option<&str>,
body: &[u8],
) -> Result<(), String> {
let reason = match code {
200 => "OK",
204 => "No Content",
400 => "Bad Request",
404 => "Not Found",
409 => "Conflict",
500 => "Internal Server Error",
_ => "Error",
};
let mut header = format!(
"HTTP/1.1 {code} {reason}\r\nContent-Length: {}\r\n",
body.len()
);
if let Some(content_type) = content_type {
header.push_str("Content-Type: ");
header.push_str(content_type);
header.push_str("\r\n");
}
header.push_str("Connection: close\r\n\r\n");
stream
.write_all(header.as_bytes())
.and_then(|()| stream.write_all(body))
.map_err(|error| error.to_string())
}
pub(super) fn read_request(stream: &mut TcpStream) -> Result<HttpRequest, String> {
let mut bytes = Vec::new();
let header_end = loop {
if bytes.len() >= MAX_HEADER_BYTES {
return Err("HTTP headers are too large".into());
}
let mut chunk = [0_u8; 4096];
let read = stream.read(&mut chunk).map_err(|error| error.to_string())?;
if read == 0 {
return Err("bad HTTP request".into());
}
bytes.extend_from_slice(&chunk[..read]);
if let Some(end) = find_header_end(&bytes) {
break end;
}
};
let header = std::str::from_utf8(&bytes[..header_end])
.map_err(|_| "HTTP headers are not UTF-8".to_owned())?;
let mut lines = header.lines();
let request_line = lines.next().ok_or_else(|| "bad HTTP request".to_owned())?;
let mut parts = request_line.split_whitespace();
let method = parts
.next()
.ok_or_else(|| "bad HTTP request".to_owned())?
.to_owned();
let path = parts
.next()
.ok_or_else(|| "bad HTTP request".to_owned())?
.to_owned();
let length = lines
.find_map(|line| {
line.split_once(':').and_then(|(name, value)| {
name.eq_ignore_ascii_case("content-length")
.then(|| value.trim().parse::<usize>().ok())
.flatten()
})
})
.unwrap_or(0);
if length > MAX_BODY_BYTES {
return Err("HTTP body is too large".into());
}
while bytes.len() < header_end + length {
let mut chunk = [0_u8; 8192];
let read = stream.read(&mut chunk).map_err(|error| error.to_string())?;
if read == 0 {
return Err("incomplete HTTP body".into());
}
bytes.extend_from_slice(&chunk[..read]);
}
Ok(HttpRequest {
method,
path: path.split('?').next().unwrap_or(&path).to_owned(),
body: bytes[header_end..header_end + length].to_vec(),
})
}
fn find_header_end(bytes: &[u8]) -> Option<usize> {
bytes
.windows(4)
.position(|window| window == b"\r\n\r\n")
.map(|position| position + 4)
.or_else(|| {
bytes
.windows(2)
.position(|window| window == b"\n\n")
.map(|position| position + 2)
})
}
pub(super) fn unix_time() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
pub(super) fn random_id(prefix: &str) -> String {
random_hex_id::<12>(prefix)
}
pub(super) fn random_tool_id(prefix: &str) -> String {
random_hex_id::<16>(prefix)
}
fn random_hex_id<const N: usize>(prefix: &str) -> String {
let mut bytes = [0_u8; N];
if File::open("/dev/urandom")
.and_then(|mut file| file.read_exact(&mut bytes))
.is_err()
{
let fallback = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
bytes.copy_from_slice(&fallback.to_le_bytes()[..N]);
}
let mut id = String::with_capacity(prefix.len() + N * 2);
id.push_str(prefix);
for byte in bytes {
use std::fmt::Write as _;
let _ = write!(id, "{byte:02x}");
}
id
}

595
src/server/request.rs Normal file
View File

@@ -0,0 +1,595 @@
use super::*;
pub(super) fn completion_request(mut value: Value) -> Result<ChatRequest, (u16, String)> {
let object = value
.as_object_mut()
.ok_or_else(|| (400, "invalid JSON request".to_owned()))?;
let prompt = object
.remove("prompt")
.ok_or_else(|| (400, "missing prompt".to_owned()))?;
let prompt = match prompt {
Value::String(text) => text,
Value::Array(values) => values
.into_iter()
.next()
.and_then(|value| value.as_str().map(str::to_owned))
.unwrap_or_default(),
_ => String::new(),
};
object.insert(
"messages".into(),
json!([
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": prompt}
]),
);
decode_chat_request(value)
}
pub(super) fn anthropic_request(value: Value) -> Result<ChatRequest, (u16, String)> {
let object = value
.as_object()
.ok_or_else(|| (400, "invalid JSON request".to_owned()))?;
let source = object
.get("messages")
.and_then(Value::as_array)
.ok_or_else(|| (400, "missing messages".to_owned()))?;
let mut messages = Vec::new();
let system = object.get("system").map(content_text).unwrap_or_default();
if !system.is_empty() {
messages.push(json!({"role": "system", "content": system}));
}
for message in source {
let role = message
.get("role")
.and_then(Value::as_str)
.unwrap_or("user");
let Some(blocks) = message.get("content").and_then(Value::as_array) else {
messages.push(json!({"role": role, "content": message.get("content").cloned().unwrap_or(Value::Null)}));
continue;
};
let mut text = String::new();
let mut reasoning = String::new();
let mut calls = Vec::new();
for block in blocks {
match block.get("type").and_then(Value::as_str).unwrap_or("text") {
"text" => text.push_str(block.get("text").and_then(Value::as_str).unwrap_or("")),
"thinking" | "redacted_thinking" => reasoning.push_str(
block
.get("thinking")
.or_else(|| block.get("data"))
.and_then(Value::as_str)
.unwrap_or(""),
),
"tool_use" => calls.push(json!({
"id": block.get("id").and_then(Value::as_str).unwrap_or(""),
"type": "function",
"function": {
"name": block.get("name").and_then(Value::as_str).unwrap_or(""),
"arguments": block.get("input").cloned().unwrap_or_else(|| json!({})).to_string()
}
})),
"tool_result" => {
if !text.is_empty() {
messages.push(json!({"role": role, "content": std::mem::take(&mut text)}));
}
messages.push(json!({
"role": "tool",
"content": block.get("content").map(content_text).unwrap_or_default(),
"tool_call_id": block.get("tool_use_id").and_then(Value::as_str).unwrap_or("")
}));
}
_ => {}
}
}
if !text.is_empty() || !reasoning.is_empty() || !calls.is_empty() {
messages.push(json!({
"role": role,
"content": text,
"reasoning_content": reasoning,
"tool_calls": calls
}));
}
}
let mut request = Map::new();
copy_request_fields(object, &mut request);
request.insert("messages".into(), Value::Array(messages));
if let Some(stops) = object.get("stop_sequences") {
request.insert("stop".into(), stops.clone());
}
if let Some(effort) = object
.get("output_config")
.and_then(|value| value.get("effort"))
{
request.insert("reasoning_effort".into(), effort.clone());
}
request.insert(
"tools".into(),
normalize_anthropic_tools(object.get("tools")),
);
decode_chat_request(Value::Object(request))
}
pub(super) fn responses_request(value: Value) -> Result<ChatRequest, (u16, String)> {
let object = value
.as_object()
.ok_or_else(|| (400, "invalid JSON request".to_owned()))?;
for key in ["previous_response_id", "conversation"] {
if object.get(key).is_some_and(|value| !value.is_null()) {
return Err((
400,
format!("{key} is not supported; replay full input instead"),
));
}
}
if let Some(choice) = object.get("tool_choice") {
match choice {
Value::String(choice) if choice == "none" || choice == "auto" => {}
Value::String(choice) => {
return Err((400, format!("tool_choice={choice} not supported")));
}
Value::Object(_) => return Err((400, "forced tool_choice not supported".into())),
_ => {}
}
}
let input = object
.get("input")
.ok_or_else(|| (400, "missing input".to_owned()))?;
let mut messages = Vec::new();
match input {
Value::String(text) => messages.push(json!({"role": "user", "content": text})),
Value::Array(items) => {
let mut pending_reasoning = String::new();
for item in items {
let item = item
.as_object()
.ok_or_else(|| (400, "invalid JSON request".to_owned()))?;
if item
.get("status")
.and_then(Value::as_str)
.is_some_and(|status| status != "completed")
{
return Err((400, "invalid JSON request".into()));
}
let kind = item
.get("type")
.and_then(Value::as_str)
.unwrap_or("message");
let role = item.get("role").and_then(Value::as_str).unwrap_or("user");
let consumes_reasoning = (kind == "message" && role == "assistant")
|| matches!(
kind,
"function_call"
| "custom_tool_call"
| "local_shell_call"
| "web_search_call"
| "tool_search_call"
| "image_generation_call"
);
let bookkeeping = matches!(kind, "compaction" | "context_compaction");
if !consumes_reasoning && !bookkeeping && !pending_reasoning.is_empty() {
messages.push(reasoning_message(std::mem::take(&mut pending_reasoning)));
}
match kind {
"message" => {
let mut message = json!({
"role": role,
"content": responses_content_text(item.get("content").unwrap_or(&Value::Null))?
});
if role == "assistant" && !pending_reasoning.is_empty() {
message["reasoning_content"] =
Value::String(std::mem::take(&mut pending_reasoning));
}
messages.push(message);
}
"function_call" | "custom_tool_call" => {
let name = if kind == "function_call" {
format!(
"{}{}",
item.get("namespace").and_then(Value::as_str).unwrap_or(""),
item.get("name").and_then(Value::as_str).unwrap_or("")
)
} else {
item.get("name")
.and_then(Value::as_str)
.unwrap_or("")
.to_owned()
};
push_responses_call(
&mut messages,
&mut pending_reasoning,
item,
&name,
item.get("arguments").or_else(|| item.get("input")),
);
}
"function_call_output" | "custom_tool_call_output" => messages.push(json!({
"role": "tool",
"content": responses_output_text(item.get("output"))?,
"tool_call_id": item.get("call_id").or_else(|| item.get("id")).and_then(Value::as_str).unwrap_or("")
})),
"reasoning" => {
for value in [item.get("summary"), item.get("content")]
.into_iter()
.flatten()
{
let text = responses_content_text(value)?;
if !pending_reasoning.is_empty() && !text.is_empty() {
pending_reasoning.push('\n');
}
pending_reasoning.push_str(&text);
}
}
"local_shell_call"
| "web_search_call"
| "tool_search_call"
| "image_generation_call" => {
let name = match kind {
"local_shell_call" => "local_shell",
"tool_search_call" => "tool_search",
other => other,
};
push_responses_call(
&mut messages,
&mut pending_reasoning,
item,
name,
item.get("action")
.or_else(|| item.get("arguments"))
.or_else(|| item.get("input")),
);
}
"local_shell_call_output"
| "web_search_call_output"
| "tool_search_output"
| "tool_search_call_output"
| "image_generation_call_output" => {
let content = if let Some(value) =
item.get("output").or_else(|| item.get("result"))
{
responses_output_text(Some(value))?
} else {
item.get("tools").map(Value::to_string).unwrap_or_default()
};
messages.push(json!({
"role": "tool",
"content": content,
"tool_call_id": item.get("call_id").or_else(|| item.get("id")).and_then(Value::as_str).unwrap_or("")
}));
}
"compaction" | "context_compaction" => {}
_ => return Err((400, "invalid JSON request".into())),
}
}
if !pending_reasoning.is_empty() {
messages.push(reasoning_message(pending_reasoning));
}
}
_ => return Err((400, "invalid JSON request".into())),
}
if let Some(instructions) = object.get("instructions") {
let instructions = match instructions {
Value::Null => String::new(),
Value::String(text) => text.clone(),
_ => return Err((400, "invalid JSON request".into())),
};
if !instructions.is_empty() {
messages.insert(0, json!({"role": "system", "content": instructions}));
}
}
let mut request = Map::new();
copy_request_fields(object, &mut request);
request.insert("messages".into(), Value::Array(messages));
if let Some(tokens) = object.get("max_output_tokens") {
request.insert("max_tokens".into(), tokens.clone());
}
if let Some(effort) = object
.get("reasoning")
.and_then(|value| value.get("effort"))
{
request.insert("reasoning_effort".into(), effort.clone());
}
request.insert(
"tools".into(),
normalize_responses_tools(object.get("tools")),
);
decode_chat_request(Value::Object(request))
}
fn reasoning_message(reasoning: String) -> Value {
json!({"role": "assistant", "content": "", "reasoning_content": reasoning})
}
fn push_responses_call(
messages: &mut Vec<Value>,
pending_reasoning: &mut String,
item: &Map<String, Value>,
name: &str,
arguments: Option<&Value>,
) {
let call = json!({
"id": item.get("call_id").or_else(|| item.get("id")).and_then(Value::as_str).unwrap_or(""),
"type": "function",
"function": {"name": name, "arguments": raw_json_text(arguments)}
});
if let Some(last) = messages.last_mut()
&& last.get("role").and_then(Value::as_str) == Some("assistant")
{
if !pending_reasoning.is_empty()
&& last
.get("reasoning_content")
.and_then(Value::as_str)
.is_none_or(str::is_empty)
{
last["reasoning_content"] = Value::String(std::mem::take(pending_reasoning));
}
if !last.get("tool_calls").is_some_and(Value::is_array) {
last["tool_calls"] = json!([]);
}
last["tool_calls"].as_array_mut().unwrap().push(call);
} else {
messages.push(json!({
"role": "assistant",
"content": "",
"reasoning_content": std::mem::take(pending_reasoning),
"tool_calls": [call]
}));
}
}
fn raw_json_text(value: Option<&Value>) -> String {
match value {
Some(Value::String(text)) => text.clone(),
Some(value) => value.to_string(),
None => "{}".into(),
}
}
fn responses_output_text(value: Option<&Value>) -> Result<String, (u16, String)> {
match value {
Some(Value::Array(_)) => responses_content_text(value.unwrap()),
Some(Value::String(text)) => Ok(text.clone()),
Some(value) => Ok(value.to_string()),
None => Ok(String::new()),
}
}
fn responses_content_text(value: &Value) -> Result<String, (u16, String)> {
match value {
Value::String(text) => Ok(text.clone()),
Value::Null => Ok(String::new()),
Value::Array(parts) => {
let mut text = String::new();
for part in parts {
match part {
Value::String(part) => text.push_str(part),
Value::Object(part)
if matches!(
part.get("type").and_then(Value::as_str),
Some(
"input_text"
| "output_text"
| "text"
| "summary_text"
| "reasoning_text"
)
) && matches!(
part.get("text"),
Some(Value::String(_) | Value::Null)
) =>
{
if let Some(value) = part.get("text").and_then(Value::as_str) {
text.push_str(value);
}
}
_ => return Err((400, "invalid JSON request".into())),
}
}
Ok(text)
}
_ => Err((400, "invalid JSON request".into())),
}
}
fn copy_request_fields(source: &Map<String, Value>, target: &mut Map<String, Value>) {
for key in [
"model",
"max_tokens",
"max_completion_tokens",
"temperature",
"top_p",
"min_p",
"top_k",
"seed",
"stream",
"stream_options",
"thinking",
"think",
"reasoning_effort",
"tool_choice",
"stop",
] {
if let Some(value) = source.get(key) {
target.insert(key.into(), value.clone());
}
}
}
fn normalize_anthropic_tools(tools: Option<&Value>) -> Value {
Value::Array(
tools
.and_then(Value::as_array)
.into_iter()
.flatten()
.map(|tool| {
json!({"type": "function", "function": {
"name": tool.get("name").and_then(Value::as_str).unwrap_or(""),
"description": tool.get("description").and_then(Value::as_str).unwrap_or(""),
"parameters": tool.get("input_schema").cloned().unwrap_or_else(|| json!({"type": "object"}))
}})
})
.collect(),
)
}
fn normalize_responses_tools(tools: Option<&Value>) -> Value {
Value::Array(
tools
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter(|tool| tool.get("type").and_then(Value::as_str) == Some("function"))
.map(|tool| {
json!({"type": "function", "function": {
"name": tool.get("name").and_then(Value::as_str).unwrap_or(""),
"description": tool.get("description").and_then(Value::as_str).unwrap_or(""),
"parameters": tool.get("parameters").cloned().unwrap_or_else(|| json!({"type": "object"}))
}})
})
.collect(),
)
}
pub(super) fn decode_chat_request(value: Value) -> Result<ChatRequest, (u16, String)> {
serde_json::from_value(value).map_err(|_| (400, "invalid JSON request".to_owned()))
}
pub(super) fn raw_tool_schemas(
body: &[u8],
protocol: Protocol,
) -> Result<Vec<String>, (u16, String)> {
let request: RawToolsRequest<'_> =
serde_json::from_slice(body).map_err(|_| (400, "invalid JSON request".to_owned()))?;
request
.tools
.into_iter()
.map(|tool| {
if protocol == Protocol::Chat {
let wrapper: RawOpenAiTool<'_> = serde_json::from_str(tool.get())
.map_err(|_| (400, "invalid JSON request".to_owned()))?;
Ok(wrapper.function.unwrap_or(tool).get().to_owned())
} else {
Ok(tool.get().to_owned())
}
})
.collect()
}
pub(super) fn parse_chat_request(
state: &State,
request: ChatRequest,
protocol: Protocol,
) -> Result<ParsedRequest, (u16, String)> {
if request.messages.is_empty() {
return Err((400, "missing messages".into()));
}
let requested_id = request.model.as_deref().unwrap_or_default();
let preferences = state
.preferences
.read()
.map_err(|_| (500, "preferences are unavailable".to_owned()))?
.clone();
let model = if requested_id.is_empty() {
ModelChoice::from_id(&preferences.selected_model)
} else {
model_alias(requested_id)
}
.ok_or_else(|| (400, format!("unknown model: {requested_id}")))?;
if !installed_endpoint_models(&state.models_path).contains(&model) {
return Err((400, format!("model is not installed and verified: {model}")));
}
let mut generation = preferences.generation().map_err(|error| (500, error))?;
generation.system_prompt.clear();
generation.max_generated_tokens = request
.max_completion_tokens
.or(request.max_tokens)
.unwrap_or(generation.max_generated_tokens);
if generation.max_generated_tokens <= 0 {
return Err((400, "max_tokens must be positive".into()));
}
generation.temperature = request.temperature.or(Some(1.0));
generation.top_p = request.top_p.or(Some(1.0));
generation.min_p = request.min_p.or(Some(0.05));
generation.seed = request.seed.filter(|seed| *seed > 0);
generation.reasoning_mode = request_reasoning(&request, requested_id)?;
let tools_enabled = (!request.tools.is_empty() || !request.tool_schemas.is_empty())
&& request.tool_choice.as_ref().and_then(Value::as_str) != Some("none");
let (system, messages) = render_messages(
state,
&request.messages,
&request.tools,
&request.tool_schemas,
tools_enabled,
protocol,
)?;
generation.system_prompt = system;
let runtime = preferences.runtime().map_err(|error| (500, error))?;
let mut effective = effective_settings(model, &generation, &runtime, &state.models_path)
.map_err(|error| (400, error))?;
effective.turn.top_k = request.top_k.unwrap_or(0);
if effective.turn.top_k < 0 {
return Err((400, "top_k must not be negative".into()));
}
effective.turn.stops = match request.stop {
Some(OneOrMany::One(stop)) => vec![stop],
Some(OneOrMany::Many(stops)) => stops,
None => Vec::new(),
};
effective.turn.stops.retain(|stop| !stop.is_empty());
Ok(ParsedRequest {
protocol,
model_id: if requested_id.is_empty() {
model.id().to_owned()
} else {
requested_id.to_owned()
},
messages,
turn: effective.turn,
engine: effective.engine,
idle_timeout: Duration::from_secs(preferences.idle_timeout_minutes.max(1) as u64 * 60),
stream: request.stream,
include_usage: request
.stream_options
.is_some_and(|options| options.include_usage),
has_tools: tools_enabled,
})
}
fn request_reasoning(
request: &ChatRequest,
model_id: &str,
) -> Result<ReasoningMode, (u16, String)> {
let explicit_thinking = request
.think
.or_else(|| request.thinking.as_ref().and_then(thinking_enabled));
let mut reasoning = match request.reasoning_effort.as_deref() {
Some("max") => ReasoningMode::Max,
Some("none") => ReasoningMode::Direct,
Some("xhigh" | "high" | "medium" | "low" | "minimal") | None => ReasoningMode::High,
Some(value) => return Err((400, format!("unsupported reasoning_effort: {value}"))),
};
if explicit_thinking == Some(false)
|| (explicit_thinking.is_none()
&& matches!(
model_id,
"deepseek-chat" | "glm-5.2-chat" | "glm-5.2-no-think" | "glm-5.2-nothink"
))
{
reasoning = ReasoningMode::Direct;
}
Ok(reasoning)
}
fn thinking_enabled(value: &Value) -> Option<bool> {
value.as_bool().or_else(|| {
value.as_str().map(|value| value != "disabled").or_else(|| {
value
.get("type")
.and_then(Value::as_str)
.map(|value| value != "disabled")
})
})
}

974
src/server/response.rs Normal file
View File

@@ -0,0 +1,974 @@
use super::*;
pub(super) fn final_response(
stream: &mut TcpStream,
state: &State,
request: ResponseOptions,
active: crate::runtime::ActiveGeneration,
id: &str,
) -> Result<(), (u16, String)> {
let output = wait_for_output(active)?;
let (content, calls) = parse_generated_tools(state, &output.message.content, request.protocol);
let finish = if calls.is_empty() {
output.finish_reason
} else {
"tool_calls"
};
let reasoning = output.message.reasoning.filter(|value| !value.is_empty());
let usage = usage_json(
output.prompt_tokens,
output.cached_tokens,
output.completion_tokens,
);
let body = match request.protocol {
Protocol::Chat => {
let mut message = json!({"role": "assistant", "content": content});
if let Some(reasoning) = reasoning {
message["reasoning_content"] = Value::String(reasoning);
}
if !calls.is_empty() {
message["tool_calls"] = tool_calls_json(&calls);
}
json!({
"id": id, "object": "chat.completion", "created": unix_time(),
"model": request.model_id,
"choices": [{"index": 0, "message": message, "finish_reason": finish}],
"usage": usage,
})
}
Protocol::Completion => json!({
"id": id, "object": "text_completion", "created": unix_time(),
"model": request.model_id,
"choices": [{"text": content, "index": 0, "finish_reason": finish}],
"usage": usage,
}),
Protocol::Anthropic => {
let mut blocks = Vec::new();
if let Some(reasoning) = reasoning {
blocks.push(json!({"type": "thinking", "thinking": reasoning, "signature": id}));
}
if !content.is_empty() {
blocks.push(json!({"type": "text", "text": content}));
}
for call in &calls {
blocks.push(json!({
"type": "tool_use", "id": call.id, "name": call.function.name,
"input": serde_json::from_str::<Value>(&call.function.arguments).unwrap_or_else(|_| json!({}))
}));
}
if blocks.is_empty() || (blocks.iter().all(|block| block["type"] == "thinking")) {
blocks.push(json!({"type": "text", "text": ""}));
}
let cached = output.cached_tokens.min(output.prompt_tokens);
let written = output.prompt_tokens - cached;
json!({
"id": id, "type": "message", "role": "assistant", "model": request.model_id,
"content": blocks,
"stop_reason": if finish == "tool_calls" { "tool_use" } else if finish == "length" { "max_tokens" } else { "end_turn" },
"stop_sequence": Value::Null,
"usage": {
"input_tokens": output.prompt_tokens - cached - written,
"output_tokens": output.completion_tokens,
"cache_read_input_tokens": cached,
"cache_creation_input_tokens": written
}
})
}
Protocol::Responses => {
let status = if finish == "length" {
"incomplete"
} else if finish == "error" {
"failed"
} else {
"completed"
};
let mut items = Vec::new();
if let Some(reasoning) = reasoning {
items.push(json!({
"id": random_id("rs_"), "type": "reasoning", "status": status,
"summary": [{"type": "summary_text", "text": reasoning}]
}));
}
if !content.is_empty() {
items.push(json!({
"id": random_id("msg_"), "type": "message", "status": status,
"role": "assistant", "content": [{"type": "output_text", "text": content, "annotations": []}]
}));
}
for call in &calls {
items.push(json!({
"id": random_id("fc_"), "type": "function_call", "status": status,
"name": call.function.name, "call_id": call.id, "arguments": call.function.arguments
}));
}
json!({
"id": id, "object": "response", "created_at": unix_time(), "status": status,
"model": request.model_id, "output": items,
"usage": {
"input_tokens": output.prompt_tokens,
"input_tokens_details": {"cached_tokens": output.cached_tokens.min(output.prompt_tokens), "cache_write_tokens": output.prompt_tokens - output.cached_tokens.min(output.prompt_tokens)},
"output_tokens": output.completion_tokens,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": output.prompt_tokens + output.completion_tokens
}
})
}
};
send_json(stream, 200, &body).map_err(|error| (500, error))
}
pub(super) fn stream_response(
stream: &mut TcpStream,
state: &State,
request: ResponseOptions,
active: crate::runtime::ActiveGeneration,
id: &str,
) -> Result<(), (u16, String)> {
if request.protocol != Protocol::Chat {
return structured_stream_response(stream, state, request, active, id);
}
stream_response_with_keepalive(
stream,
state,
request,
active,
id,
PREFILL_KEEPALIVE_INTERVAL,
)
}
fn structured_stream_response(
stream: &mut TcpStream,
state: &State,
request: ResponseOptions,
active: crate::runtime::ActiveGeneration,
id: &str,
) -> Result<(), (u16, String)> {
match request.protocol {
Protocol::Completion => completion_stream_response(stream, request, active, id),
Protocol::Anthropic => anthropic_stream_response(stream, state, request, active, id),
Protocol::Responses => responses_stream_response(stream, state, request, active, id),
Protocol::Chat => unreachable!(),
}
}
fn completion_stream_response(
stream: &mut TcpStream,
request: ResponseOptions,
active: crate::runtime::ActiveGeneration,
id: &str,
) -> Result<(), (u16, String)> {
send_sse_headers(stream).map_err(|error| (500, error))?;
while let Ok(event) = active.events.recv() {
match event {
GenerationEvent::Chunk {
reasoning: false,
content,
} if !content.is_empty() => send_sse(
stream,
&json!({
"id": id, "object": "text_completion", "created": unix_time(),
"model": request.model_id,
"choices": [{"text": content, "index": 0, "finish_reason": Value::Null}]
}),
)
.map_err(|error| (500, error))?,
GenerationEvent::Finished(Ok(output)) => {
send_sse(
stream,
&json!({
"id": id, "object": "text_completion", "created": unix_time(),
"model": request.model_id,
"choices": [{"text": "", "index": 0, "finish_reason": output.finish_reason}]
}),
)
.map_err(|error| (500, error))?;
if request.include_usage {
send_sse(
stream,
&json!({
"id": id, "object": "text_completion", "created": unix_time(),
"model": request.model_id, "choices": [],
"usage": usage_json(output.prompt_tokens, output.cached_tokens, output.completion_tokens)
}),
)
.map_err(|error| (500, error))?;
}
return stream
.write_all(b"data: [DONE]\n\n")
.map_err(|error| (500, error.to_string()));
}
GenerationEvent::Finished(Err(error)) => {
let _ = send_sse_error(stream, &error);
return Ok(());
}
_ => {}
}
}
Err((500, "The model runtime stopped unexpectedly.".into()))
}
fn anthropic_stream_response(
stream: &mut TcpStream,
state: &State,
request: ResponseOptions,
active: crate::runtime::ActiveGeneration,
id: &str,
) -> Result<(), (u16, String)> {
send_sse_headers(stream).map_err(|error| (500, error))?;
let mut prompt_tokens = 0;
let mut started = false;
let mut block = None::<(usize, bool)>;
let mut next_index = 0;
let mut projector = ToolProjector::new();
let mut tool_indices = Vec::new();
while let Ok(event) = active.events.recv() {
match event {
GenerationEvent::Context {
used,
tokens_per_second,
..
} => {
if tokens_per_second.is_none() {
prompt_tokens = used;
} else if !started {
anthropic_stream_start(stream, &request, id, prompt_tokens, 0)?;
started = true;
}
}
GenerationEvent::Chunk { reasoning, content } => {
if !started {
anthropic_stream_start(stream, &request, id, prompt_tokens, 0)?;
started = true;
}
if !reasoning && request.has_tools {
let events = projector.push(&content, false, "toolu_");
send_anthropic_projection_events(
stream,
events,
&mut block,
&mut next_index,
&mut tool_indices,
)?;
continue;
}
if block.is_some_and(|(_, current_reasoning)| current_reasoning != reasoning) {
let (index, _) = block.take().unwrap();
send_named_sse(
stream,
"content_block_stop",
&json!({"type": "content_block_stop", "index": index}),
)?;
}
let index = if let Some((index, _)) = block {
index
} else {
let index = next_index;
next_index += 1;
send_named_sse(
stream,
"content_block_start",
&json!({
"type": "content_block_start", "index": index,
"content_block": if reasoning { json!({"type": "thinking", "thinking": "", "signature": ""}) } else { json!({"type": "text", "text": ""}) }
}),
)?;
block = Some((index, reasoning));
index
};
let delta = if reasoning {
json!({"type": "thinking_delta", "thinking": content})
} else {
json!({"type": "text_delta", "text": content})
};
send_named_sse(
stream,
"content_block_delta",
&json!({"type": "content_block_delta", "index": index, "delta": delta}),
)?;
}
GenerationEvent::Finished(Ok(output)) => {
if !started {
anthropic_stream_start(
stream,
&request,
id,
output.prompt_tokens,
output.cached_tokens,
)?;
}
if request.has_tools {
let events = projector.push("", true, "toolu_");
send_anthropic_projection_events(
stream,
events,
&mut block,
&mut next_index,
&mut tool_indices,
)?;
}
if let Some((index, _)) = block.take() {
send_named_sse(
stream,
"content_block_stop",
&json!({"type": "content_block_stop", "index": index}),
)?;
}
let (_, calls) = parse_generated_tools_with_ids(
state,
&output.message.content,
Protocol::Anthropic,
&projector.ids,
);
if projector.ids.is_empty() {
for call in &calls {
send_named_sse(
stream,
"content_block_start",
&json!({"type": "content_block_start", "index": next_index, "content_block": {"type": "tool_use", "id": call.id, "name": call.function.name, "input": {}}}),
)?;
send_named_sse(
stream,
"content_block_delta",
&json!({"type": "content_block_delta", "index": next_index, "delta": {"type": "input_json_delta", "partial_json": call.function.arguments}}),
)?;
send_named_sse(
stream,
"content_block_stop",
&json!({"type": "content_block_stop", "index": next_index}),
)?;
next_index += 1;
}
}
let finish = if calls.is_empty() {
output.finish_reason
} else {
"tool_calls"
};
send_named_sse(
stream,
"message_delta",
&json!({"type": "message_delta", "delta": {"stop_reason": if finish == "tool_calls" { "tool_use" } else if finish == "length" { "max_tokens" } else { "end_turn" }, "stop_sequence": Value::Null}, "usage": {"output_tokens": output.completion_tokens}}),
)?;
return send_named_sse(stream, "message_stop", &json!({"type": "message_stop"}));
}
GenerationEvent::Finished(Err(error)) => {
return send_named_sse(
stream,
"error",
&json!({"type": "error", "error": {"type": "api_error", "message": error}}),
);
}
_ => {}
}
}
Err((500, "The model runtime stopped unexpectedly.".into()))
}
fn send_anthropic_projection_events(
stream: &mut impl Write,
events: Vec<ToolProjectionEvent>,
block: &mut Option<(usize, bool)>,
next_index: &mut usize,
tool_indices: &mut Vec<usize>,
) -> Result<(), (u16, String)> {
for event in events {
match event {
ToolProjectionEvent::Text(text) if !text.is_empty() => {
if block.is_some_and(|(_, reasoning)| reasoning) {
let (index, _) = block.take().unwrap();
send_named_sse(
stream,
"content_block_stop",
&json!({"type": "content_block_stop", "index": index}),
)?;
}
let index = if let Some((index, _)) = *block {
index
} else {
let index = *next_index;
*next_index += 1;
send_named_sse(
stream,
"content_block_start",
&json!({"type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}}),
)?;
*block = Some((index, false));
index
};
send_named_sse(
stream,
"content_block_delta",
&json!({"type": "content_block_delta", "index": index, "delta": {"type": "text_delta", "text": text}}),
)?;
}
ToolProjectionEvent::Start { index, id, name } => {
if let Some((open_index, _)) = block.take() {
send_named_sse(
stream,
"content_block_stop",
&json!({"type": "content_block_stop", "index": open_index}),
)?;
}
let content_index = *next_index;
*next_index += 1;
if tool_indices.len() == index {
tool_indices.push(content_index);
}
send_named_sse(
stream,
"content_block_start",
&json!({"type": "content_block_start", "index": content_index, "content_block": {"type": "tool_use", "id": id, "name": name, "input": {}}}),
)?;
}
ToolProjectionEvent::Arguments { index, fragment } => {
if let Some(content_index) = tool_indices.get(index) {
send_named_sse(
stream,
"content_block_delta",
&json!({"type": "content_block_delta", "index": content_index, "delta": {"type": "input_json_delta", "partial_json": fragment}}),
)?;
}
}
ToolProjectionEvent::End { index } => {
if let Some(content_index) = tool_indices.get(index) {
send_named_sse(
stream,
"content_block_stop",
&json!({"type": "content_block_stop", "index": content_index}),
)?;
}
}
ToolProjectionEvent::Text(_) => {}
}
}
Ok(())
}
fn anthropic_stream_start(
stream: &mut impl Write,
request: &ResponseOptions,
id: &str,
prompt_tokens: u32,
cached_tokens: u32,
) -> Result<(), (u16, String)> {
let cached = cached_tokens.min(prompt_tokens);
let written = prompt_tokens - cached;
send_named_sse(
stream,
"message_start",
&json!({"type": "message_start", "message": {"id": id, "type": "message", "role": "assistant", "model": request.model_id, "content": [], "stop_reason": Value::Null, "stop_sequence": Value::Null, "usage": {"input_tokens": prompt_tokens - cached - written, "output_tokens": 0, "cache_read_input_tokens": cached, "cache_creation_input_tokens": written}}}),
)
}
fn responses_stream_response(
stream: &mut TcpStream,
state: &State,
request: ResponseOptions,
active: crate::runtime::ActiveGeneration,
id: &str,
) -> Result<(), (u16, String)> {
send_sse_headers(stream).map_err(|error| (500, error))?;
let created = unix_time();
let message_id = random_id("msg_");
let reasoning_id = random_id("rs_");
let mut sequence = 0;
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.created", "response": {"id": id, "object": "response", "created_at": created, "status": "in_progress", "model": request.model_id, "output": []}}),
)?;
let mut reasoning_open = false;
let mut message_open = false;
let mut reasoning = String::new();
let mut content = String::new();
while let Ok(event) = active.events.recv() {
match event {
GenerationEvent::Chunk {
reasoning: true,
content: chunk,
} => {
if !request.reasoning_summary {
continue;
}
if !reasoning_open {
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_item.added", "output_index": 0, "item": {"id": reasoning_id, "type": "reasoning", "status": "in_progress", "summary": []}}),
)?;
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.reasoning_summary_part.added", "item_id": reasoning_id, "output_index": 0, "summary_index": 0, "part": {"type": "summary_text", "text": ""}}),
)?;
reasoning_open = true;
}
reasoning.push_str(&chunk);
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.reasoning_summary_text.delta", "item_id": reasoning_id, "output_index": 0, "summary_index": 0, "delta": chunk}),
)?;
}
GenerationEvent::Chunk {
reasoning: false,
content: chunk,
} => {
if request.has_tools {
content.push_str(&chunk);
continue;
}
if !message_open {
let output_index = usize::from(reasoning_open);
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_item.added", "output_index": output_index, "item": {"id": message_id, "type": "message", "status": "in_progress", "role": "assistant", "content": []}}),
)?;
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.content_part.added", "item_id": message_id, "output_index": output_index, "content_index": 0, "part": {"type": "output_text", "text": "", "annotations": []}}),
)?;
message_open = true;
}
content.push_str(&chunk);
let output_index = usize::from(reasoning_open);
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_text.delta", "item_id": message_id, "output_index": output_index, "content_index": 0, "delta": chunk}),
)?;
}
GenerationEvent::Finished(Ok(output)) => {
let (parsed_content, calls) =
parse_generated_tools(state, &output.message.content, Protocol::Responses);
if request.has_tools {
content = parsed_content;
}
let finish = if calls.is_empty() {
output.finish_reason
} else {
"tool_calls"
};
let status = if finish == "length" {
"incomplete"
} else if finish == "error" {
"failed"
} else {
"completed"
};
let mut terminal_items = Vec::new();
let mut output_index = 0;
if reasoning_open {
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.reasoning_summary_text.done", "item_id": reasoning_id, "output_index": output_index, "summary_index": 0, "text": reasoning}),
)?;
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.reasoning_summary_part.done", "item_id": reasoning_id, "output_index": output_index, "summary_index": 0, "part": {"type": "summary_text", "text": reasoning}}),
)?;
let item = json!({"id": reasoning_id, "type": "reasoning", "status": status, "summary": [{"type": "summary_text", "text": reasoning}]});
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_item.done", "output_index": output_index, "item": item}),
)?;
terminal_items.push(item);
output_index += 1;
}
if !content.is_empty() {
if !message_open {
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_item.added", "output_index": output_index, "item": {"id": message_id, "type": "message", "status": "in_progress", "role": "assistant", "content": []}}),
)?;
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.content_part.added", "item_id": message_id, "output_index": output_index, "content_index": 0, "part": {"type": "output_text", "text": "", "annotations": []}}),
)?;
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_text.delta", "item_id": message_id, "output_index": output_index, "content_index": 0, "delta": content}),
)?;
}
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_text.done", "item_id": message_id, "output_index": output_index, "content_index": 0, "text": content}),
)?;
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.content_part.done", "item_id": message_id, "output_index": output_index, "content_index": 0, "part": {"type": "output_text", "text": content, "annotations": []}}),
)?;
let item = json!({"id": message_id, "type": "message", "status": status, "role": "assistant", "content": [{"type": "output_text", "text": content, "annotations": []}]});
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_item.done", "output_index": output_index, "item": item}),
)?;
terminal_items.push(item);
output_index += 1;
}
for call in &calls {
let item_id = random_id("fc_");
let mut item = json!({"id": item_id, "type": "function_call", "status": status, "name": call.function.name, "call_id": call.id, "arguments": call.function.arguments});
let mut added = item.clone();
added["status"] = Value::String("in_progress".into());
added["arguments"] = Value::String(String::new());
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_item.added", "output_index": output_index, "item": added}),
)?;
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.function_call_arguments.delta", "item_id": item_id, "output_index": output_index, "delta": call.function.arguments}),
)?;
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.function_call_arguments.done", "item_id": item_id, "output_index": output_index, "name": call.function.name, "arguments": call.function.arguments}),
)?;
item["id"] = Value::String(item_id);
send_responses_sse(
stream,
&mut sequence,
json!({"type": "response.output_item.done", "output_index": output_index, "item": item}),
)?;
terminal_items.push(item);
output_index += 1;
}
let event_type = if finish == "length" {
"response.incomplete"
} else if finish == "error" {
"response.failed"
} else {
"response.completed"
};
let cached = output.cached_tokens.min(output.prompt_tokens);
return send_responses_sse(
stream,
&mut sequence,
json!({"type": event_type, "response": {"id": id, "object": "response", "created_at": created, "status": status, "model": request.model_id, "output": terminal_items, "usage": {"input_tokens": output.prompt_tokens, "input_tokens_details": {"cached_tokens": cached, "cache_write_tokens": output.prompt_tokens - cached}, "output_tokens": output.completion_tokens, "output_tokens_details": {"reasoning_tokens": 0}, "total_tokens": output.prompt_tokens + output.completion_tokens}}}),
);
}
GenerationEvent::Finished(Err(error)) => {
let _ = send_sse_error(stream, &error);
return Ok(());
}
_ => {}
}
}
Err((500, "The model runtime stopped unexpectedly.".into()))
}
#[allow(clippy::too_many_arguments)]
fn send_named_sse(
stream: &mut impl Write,
event: &str,
value: &Value,
) -> Result<(), (u16, String)> {
let body = serde_json::to_vec(value).map_err(|error| (500, error.to_string()))?;
stream
.write_all(b"event: ")
.and_then(|()| stream.write_all(event.as_bytes()))
.and_then(|()| stream.write_all(b"\ndata: "))
.and_then(|()| stream.write_all(&body))
.and_then(|()| stream.write_all(b"\n\n"))
.map_err(|error| (500, error.to_string()))
}
fn send_responses_sse(
stream: &mut impl Write,
sequence: &mut u32,
value: Value,
) -> Result<(), (u16, String)> {
let mut object = value
.as_object()
.cloned()
.ok_or_else(|| (500, "Responses event is not an object".to_owned()))?;
let event_type = object.shift_remove("type").unwrap_or(Value::Null);
let mut ordered = Map::new();
ordered.insert("type".into(), event_type);
ordered.insert("sequence_number".into(), Value::from(*sequence));
ordered.extend(object);
*sequence += 1;
send_sse(stream, &Value::Object(ordered)).map_err(|error| (500, error))
}
fn stream_response_with_keepalive(
stream: &mut TcpStream,
state: &State,
request: ResponseOptions,
active: crate::runtime::ActiveGeneration,
id: &str,
keepalive_interval: Duration,
) -> Result<(), (u16, String)> {
let mut projector = ToolProjector::new();
let mut output = None;
let mut prefilling = true;
let mut headers_sent = false;
let mut role_sent = false;
let mut last_keepalive = Instant::now();
loop {
let event = match receive_stream_event(
stream,
&active,
prefilling && headers_sent,
&mut last_keepalive,
keepalive_interval,
) {
Ok(Some(event)) => event,
Ok(None) => break,
Err(_) => return Ok(()),
};
match event {
GenerationEvent::Loading => {}
GenerationEvent::Context {
tokens_per_second, ..
} => {
prefilling = tokens_per_second.is_none();
if prefilling && !headers_sent {
send_sse_headers(stream).map_err(|error| (500, error))?;
headers_sent = true;
last_keepalive = Instant::now();
} else if !prefilling {
send_stream_start(stream, &request, id, &mut headers_sent, &mut role_sent)?;
}
}
GenerationEvent::Chunk { reasoning, content } => {
prefilling = false;
send_stream_start(stream, &request, id, &mut headers_sent, &mut role_sent)?;
if request.has_tools && !reasoning {
let events = projector.push(&content, false, "call_");
if send_chat_projection_events(stream, &request, id, events).is_err() {
active.cancel.store(true, Ordering::Relaxed);
return Ok(());
}
if TOOL_SYNTAXES
.iter()
.any(|syntax| projector.raw.contains(syntax.tool_end))
{
active.cancel.store(true, Ordering::Relaxed);
}
} else if !content.is_empty() {
let field = if reasoning {
"reasoning_content"
} else {
"content"
};
let chunk = chunk_json(id, &request.model_id, json!({field: content}), None);
if send_sse(stream, &chunk).is_err() {
active.cancel.store(true, Ordering::Relaxed);
return Ok(());
}
}
}
GenerationEvent::Finished(result) => {
match result {
Ok(result) => {
send_stream_start(stream, &request, id, &mut headers_sent, &mut role_sent)?;
output = Some(result);
}
Err(error) => {
if headers_sent {
let _ = send_sse_error(stream, &error);
return Ok(());
}
return Err((500, error));
}
}
break;
}
}
}
let output = match output {
Some(output) => output,
None if headers_sent => {
let _ = send_sse_error(stream, "The model runtime stopped unexpectedly.");
return Ok(());
}
None => return Err((500, "The model runtime stopped unexpectedly.".into())),
};
if request.has_tools {
let events = projector.push("", true, "call_");
send_chat_projection_events(stream, &request, id, events)?;
}
let (_, calls) = parse_generated_tools_with_ids(
state,
&output.message.content,
Protocol::Chat,
&projector.ids,
);
if request.has_tools && projector.ids.is_empty() && !calls.is_empty() {
send_sse(
stream,
&chunk_json(
id,
&request.model_id,
json!({"tool_calls": tool_calls_json(&calls)}),
None,
),
)
.map_err(|error| (500, error))?;
}
let finish = if calls.is_empty() {
output.finish_reason
} else {
"tool_calls"
};
send_sse(
stream,
&chunk_json(id, &request.model_id, json!({}), Some(finish)),
)
.map_err(|error| (500, error))?;
if request.include_usage {
let usage = json!({
"id": id,
"object": "chat.completion.chunk",
"created": unix_time(),
"model": request.model_id,
"choices": [],
"usage": usage_json(output.prompt_tokens, output.cached_tokens, output.completion_tokens),
});
send_sse(stream, &usage).map_err(|error| (500, error))?;
}
stream
.write_all(b"data: [DONE]\n\n")
.map_err(|error| (500, error.to_string()))
}
fn send_chat_projection_events(
stream: &mut impl Write,
request: &ResponseOptions,
response_id: &str,
events: Vec<ToolProjectionEvent>,
) -> Result<(), (u16, String)> {
for event in events {
let delta = match event {
ToolProjectionEvent::Text(content) if !content.is_empty() => {
json!({"content": content})
}
ToolProjectionEvent::Start { index, id, name } => json!({"tool_calls": [{
"index": index, "id": id, "type": "function",
"function": {"name": name, "arguments": ""}
}]}),
ToolProjectionEvent::Arguments { index, fragment } => json!({
"tool_calls": [{"index": index, "function": {"arguments": fragment}}]
}),
ToolProjectionEvent::End { .. } | ToolProjectionEvent::Text(_) => continue,
};
send_sse(
stream,
&chunk_json(response_id, &request.model_id, delta, None),
)
.map_err(|error| (500, error))?;
}
Ok(())
}
pub(super) fn send_stream_start(
stream: &mut impl Write,
request: &ResponseOptions,
id: &str,
headers_sent: &mut bool,
role_sent: &mut bool,
) -> Result<(), (u16, String)> {
if !*headers_sent {
send_sse_headers(stream).map_err(|error| (500, error))?;
*headers_sent = true;
}
if !*role_sent {
let role = chunk_json(id, &request.model_id, json!({"role": "assistant"}), None);
send_sse(stream, &role).map_err(|error| (500, error))?;
*role_sent = true;
}
Ok(())
}
pub(super) fn receive_stream_event(
stream: &mut impl Write,
active: &crate::runtime::ActiveGeneration,
prefilling: bool,
last_keepalive: &mut Instant,
keepalive_interval: Duration,
) -> Result<Option<GenerationEvent>, String> {
if !prefilling {
return Ok(active.events.recv().ok());
}
loop {
if last_keepalive.elapsed() >= keepalive_interval {
if let Err(error) = stream.write_all(b": prefill\n\n") {
active.cancel.store(true, Ordering::Relaxed);
return Err(error.to_string());
}
*last_keepalive = Instant::now();
}
let remaining = keepalive_interval.saturating_sub(last_keepalive.elapsed());
match active.events.recv_timeout(remaining) {
Ok(event) => return Ok(Some(event)),
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => return Ok(None),
}
}
}
fn wait_for_output(
active: crate::runtime::ActiveGeneration,
) -> Result<crate::engine::GenerationOutput, (u16, String)> {
let mut content = String::new();
while let Ok(event) = active.events.recv() {
match event {
GenerationEvent::Chunk {
reasoning: false,
content: chunk,
} => {
content.push_str(&chunk);
if TOOL_SYNTAXES
.iter()
.any(|syntax| content.contains(syntax.tool_end))
{
active.cancel.store(true, Ordering::Relaxed);
}
}
GenerationEvent::Finished(result) => {
return result.map_err(|error| (500, error));
}
_ => {}
}
}
Err((500, "The model runtime stopped unexpectedly.".into()))
}
pub(super) fn usage_json(prompt: u32, cached: u32, completion: u32) -> Value {
let cached = cached.min(prompt);
json!({
"prompt_tokens": prompt,
"completion_tokens": completion,
"total_tokens": prompt + completion,
"prompt_tokens_details": {
"cached_tokens": cached,
"cache_write_tokens": prompt - cached
}
})
}
pub(super) fn chunk_json(id: &str, model: &str, delta: Value, finish: Option<&str>) -> Value {
json!({
"id": id,
"object": "chat.completion.chunk",
"created": unix_time(),
"model": model,
"choices": [{"index": 0, "delta": delta, "finish_reason": finish}]
})
}

733
src/server/tools.rs Normal file
View File

@@ -0,0 +1,733 @@
use super::*;
enum ToolProjectionState {
Seeking,
Invokes,
Parameters,
Value,
Done,
Failed,
}
pub(super) enum ToolProjectionEvent {
Text(String),
Start {
index: usize,
id: String,
name: String,
},
Arguments {
index: usize,
fragment: String,
},
End {
index: usize,
},
}
pub(super) struct ToolProjector {
pub(super) raw: String,
position: usize,
text_emitted: usize,
state: ToolProjectionState,
index: usize,
pub(super) ids: Vec<String>,
first_parameter: bool,
string_parameter: bool,
syntax: Option<ToolSyntax>,
}
impl ToolProjector {
pub(super) fn new() -> Self {
Self {
raw: String::new(),
position: 0,
text_emitted: 0,
state: ToolProjectionState::Seeking,
index: 0,
ids: Vec::new(),
first_parameter: true,
string_parameter: false,
syntax: None,
}
}
pub(super) fn push(
&mut self,
chunk: &str,
final_chunk: bool,
prefix: &str,
) -> Vec<ToolProjectionEvent> {
self.raw.push_str(chunk);
let mut events = Vec::new();
loop {
match self.state {
ToolProjectionState::Seeking => {
if let Some((start, syntax)) = TOOL_SYNTAXES
.iter()
.filter_map(|syntax| {
self.raw
.find(syntax.tool_start)
.map(|start| (start, *syntax))
})
.min_by_key(|(start, _)| *start)
{
if start > self.text_emitted {
let text = self.raw[self.text_emitted..start].trim_end();
if !text.is_empty() {
events.push(ToolProjectionEvent::Text(text.to_owned()));
}
}
self.position = start + syntax.tool_start.len();
self.text_emitted = start;
self.syntax = Some(syntax);
self.state = ToolProjectionState::Invokes;
} else {
let limit = if final_chunk {
self.raw.len()
} else {
TOOL_SYNTAXES
.iter()
.map(|syntax| {
safe_before_partial_marker(&self.raw, syntax.tool_start)
})
.min()
.unwrap_or(self.raw.len())
};
if limit > self.text_emitted {
let text = &self.raw[self.text_emitted..limit];
if !text.trim().is_empty() {
events.push(ToolProjectionEvent::Text(text.to_owned()));
self.text_emitted = limit;
}
}
break;
}
}
ToolProjectionState::Invokes => {
let syntax = self.syntax.unwrap();
self.skip_whitespace();
if self.full_at(syntax.tool_end) {
self.position += syntax.tool_end.len();
self.state = ToolProjectionState::Done;
break;
}
if self.partial_at(syntax.tool_end) || self.partial_at(syntax.invoke_start) {
break;
}
if !self.full_at(syntax.invoke_start) {
self.state = ToolProjectionState::Failed;
break;
}
let Some(tag_end) = self.raw[self.position..].find('>') else {
break;
};
let tag_end = self.position + tag_end + 1;
let Some(name) = dsml_attribute(&self.raw[self.position..tag_end], "name")
else {
self.state = ToolProjectionState::Failed;
break;
};
let id = random_tool_id(prefix);
self.ids.push(id.clone());
events.push(ToolProjectionEvent::Start {
index: self.index,
id,
name,
});
events.push(ToolProjectionEvent::Arguments {
index: self.index,
fragment: "{".into(),
});
self.position = tag_end;
self.first_parameter = true;
self.state = ToolProjectionState::Parameters;
}
ToolProjectionState::Parameters => {
let syntax = self.syntax.unwrap();
self.skip_whitespace();
if self.full_at(syntax.invoke_end) {
events.push(ToolProjectionEvent::Arguments {
index: self.index,
fragment: "}".into(),
});
events.push(ToolProjectionEvent::End { index: self.index });
self.position += syntax.invoke_end.len();
self.index += 1;
self.state = ToolProjectionState::Invokes;
continue;
}
if self.partial_at(syntax.invoke_end) || self.partial_at(syntax.parameter_start)
{
break;
}
if !self.full_at(syntax.parameter_start) {
self.state = ToolProjectionState::Failed;
break;
}
let Some(tag_end) = self.raw[self.position..].find('>') else {
break;
};
let tag_end = self.position + tag_end + 1;
let tag = &self.raw[self.position..tag_end];
let Some(name) = dsml_attribute(tag, "name") else {
self.state = ToolProjectionState::Failed;
break;
};
self.string_parameter =
dsml_attribute(tag, "string").as_deref() != Some("false");
let mut fragment = if self.first_parameter {
String::new()
} else {
",".into()
};
self.first_parameter = false;
fragment
.push_str(&serde_json::to_string(&name).unwrap_or_else(|_| "\"\"".into()));
fragment.push(':');
if self.string_parameter {
fragment.push('"');
}
events.push(ToolProjectionEvent::Arguments {
index: self.index,
fragment,
});
self.position = tag_end;
self.state = ToolProjectionState::Value;
}
ToolProjectionState::Value => {
let syntax = self.syntax.unwrap();
if let Some(relative_end) = self.raw[self.position..].find(syntax.parameter_end)
{
let end = self.position + relative_end;
self.emit_value(end, &mut events);
if self.string_parameter {
events.push(ToolProjectionEvent::Arguments {
index: self.index,
fragment: "\"".into(),
});
}
self.position = end + syntax.parameter_end.len();
self.state = ToolProjectionState::Parameters;
continue;
}
let limit = safe_parameter_value_limit(
&self.raw,
self.position,
syntax.parameter_end,
self.string_parameter,
);
self.emit_value(limit, &mut events);
break;
}
ToolProjectionState::Done | ToolProjectionState::Failed => break,
}
}
events
}
fn emit_value(&mut self, end: usize, events: &mut Vec<ToolProjectionEvent>) {
if end <= self.position {
return;
}
let raw = &self.raw[self.position..end];
let fragment = if self.string_parameter {
let value = unescape_dsml(raw);
let encoded = serde_json::to_string(&value).unwrap_or_else(|_| "\"\"".into());
encoded[1..encoded.len() - 1].to_owned()
} else {
raw.to_owned()
};
events.push(ToolProjectionEvent::Arguments {
index: self.index,
fragment,
});
self.position = end;
}
fn skip_whitespace(&mut self) {
while self.raw[self.position..]
.chars()
.next()
.is_some_and(char::is_whitespace)
{
self.position += self.raw[self.position..].chars().next().unwrap().len_utf8();
}
}
fn full_at(&self, marker: &str) -> bool {
self.raw.as_bytes()[self.position..].starts_with(marker.as_bytes())
}
fn partial_at(&self, marker: &str) -> bool {
let tail = &self.raw.as_bytes()[self.position..];
tail.len() < marker.len() && marker.as_bytes().starts_with(tail)
}
}
fn dsml_attribute(tag: &str, name: &str) -> Option<String> {
let start = tag.find(&format!("{name}=\""))? + name.len() + 2;
let end = start + tag[start..].find('"')?;
Some(unescape_dsml(&tag[start..end]))
}
fn safe_before_partial_marker(text: &str, marker: &str) -> usize {
let mut limit = text.len().saturating_sub(marker.len().saturating_sub(1));
while !text.is_char_boundary(limit) {
limit -= 1;
}
limit
}
fn safe_parameter_value_limit(text: &str, start: usize, end_marker: &str, string: bool) -> usize {
let bytes = text.as_bytes();
let marker = end_marker.as_bytes();
let mut limit = bytes.len();
for length in (1..marker.len().min(bytes.len().saturating_sub(start) + 1)).rev() {
if bytes[start..].ends_with(&marker[..length]) {
limit -= length;
break;
}
}
if string {
for entity in ["&amp;", "&lt;", "&gt;", "&quot;", "&apos;"] {
let entity = entity.as_bytes();
for length in 1..entity.len() {
if bytes[start..limit].ends_with(&entity[..length]) {
limit -= length;
return limit;
}
}
}
}
limit
}
pub(super) fn render_messages(
state: &State,
messages: &[ApiMessage],
tools: &[Value],
tool_schemas: &[String],
tools_enabled: bool,
protocol: Protocol,
) -> Result<(String, Vec<ChatTurn>), (u16, String)> {
validate_tool_results(state, messages, protocol)?;
let preserve_reasoning = tools_enabled
|| messages.iter().any(|message| {
matches!(message.role.as_str(), "tool" | "function") || !message.tool_calls.is_empty()
});
let mut system = String::new();
if tools_enabled {
system.push_str(TOOLS_PROMPT);
if tool_schemas.is_empty() {
for tool in tools {
let schema = tool.get("function").unwrap_or(tool);
if !system.ends_with("\n\n") {
system.push('\n');
}
system.push_str(
&serde_json::to_string(schema)
.map_err(|error| (400, format!("invalid tool schema: {error}")))?,
);
system.push('\n');
}
} else {
for schema in tool_schemas {
if !system.ends_with("\n\n") {
system.push('\n');
}
system.push_str(schema);
system.push('\n');
}
}
system.push_str(
"\nYou MUST strictly follow the above defined tool name and parameter schemas to invoke tool calls. Use the exact parameter names from the schemas.",
);
}
let mut turns = Vec::<ChatTurn>::new();
for message in messages {
let content = content_text(&message.content);
match message.role.as_str() {
"system" | "developer" => {
if !system.is_empty() {
system.push_str("\n\n");
}
system.push_str(&content);
}
"user" => turns.push(ChatTurn {
user: true,
skip_previous_eos: false,
reasoning: None,
reasoning_complete: true,
content,
}),
"tool" | "function" => {
let wrapped = format!(
"<tool_result>{}</tool_result>",
escape_tool_result(&content)
);
if let Some(previous) = turns.last_mut()
&& previous.user
&& previous.content.starts_with("<tool_result>")
{
previous.content.push_str(&wrapped);
} else {
turns.push(ChatTurn {
user: true,
skip_previous_eos: protocol == Protocol::Responses,
reasoning: None,
reasoning_complete: true,
content: wrapped,
});
}
}
"assistant" => {
let mut content = content;
if !message.tool_calls.is_empty() {
content.push_str(&replayed_or_canonical_tools(state, &message.tool_calls));
}
let reasoning = content_text(&message.reasoning_content);
turns.push(ChatTurn {
user: false,
skip_previous_eos: false,
reasoning: (preserve_reasoning && !reasoning.is_empty()).then_some(reasoning),
reasoning_complete: true,
content,
});
}
role => return Err((400, format!("unsupported message role: {role}"))),
}
}
Ok((system, turns))
}
pub(super) fn validate_tool_results(
state: &State,
messages: &[ApiMessage],
protocol: Protocol,
) -> Result<(), (u16, String)> {
if !matches!(protocol, Protocol::Anthropic | Protocol::Responses) {
return Ok(());
}
let memory = state.tool_memory.lock().ok();
for (index, message) in messages.iter().enumerate() {
if !matches!(message.role.as_str(), "tool" | "function") || message.tool_call_id.is_empty()
{
continue;
}
let id = &message.tool_call_id;
let live = memory
.as_ref()
.is_some_and(|memory| memory.contains_key(id));
let replayed = messages[..index].iter().any(|message| {
message.role == "assistant" && message.tool_calls.iter().any(|call| call.id == *id)
});
if live || replayed {
continue;
}
let message = match protocol {
Protocol::Anthropic => format!(
"Anthropic continuation state is not available for tool_use_id {id}; retry by replaying the full messages history"
),
Protocol::Responses => format!(
"Responses continuation state is not available for call_id {id}; retry by replaying the full input history"
),
_ => unreachable!(),
};
return Err((400, message));
}
Ok(())
}
fn replayed_or_canonical_tools(state: &State, calls: &[ApiToolCall]) -> String {
if let Ok(memory) = state.tool_memory.lock()
&& let Some(raw) = calls.iter().find_map(|call| {
(!call.id.is_empty())
.then(|| memory.get(&call.id))
.flatten()
})
{
return raw.clone();
}
canonical_tools(calls)
}
pub(super) fn canonical_tools(calls: &[ApiToolCall]) -> String {
let mut output = String::from("\n\n<｜DSML｜tool_calls>\n");
for call in calls {
output.push_str("<｜DSML｜invoke name=\"");
output.push_str(&escape_attribute(&call.function.name));
output.push_str("\">\n");
match serde_json::from_str::<Value>(&call.function.arguments) {
Ok(Value::Object(arguments)) => {
for (name, value) in arguments {
output.push_str("<｜DSML｜parameter name=\"");
output.push_str(&escape_attribute(&name));
let string = value.as_str();
output.push_str(if string.is_some() {
"\" string=\"true\">"
} else {
"\" string=\"false\">"
});
if let Some(value) = string {
output.push_str(&escape_parameter(value));
} else {
output.push_str(&escape_json_parameter(&value.to_string()));
}
output.push_str("</｜DSML｜parameter>\n");
}
}
_ => {
output.push_str("<｜DSML｜parameter name=\"arguments\" string=\"true\">");
output.push_str(&escape_parameter(&call.function.arguments));
output.push_str("</｜DSML｜parameter>\n");
}
}
output.push_str("</｜DSML｜invoke>\n");
}
output.push_str("</｜DSML｜tool_calls>");
output
}
pub(super) fn parse_generated_tools(
state: &State,
text: &str,
protocol: Protocol,
) -> (String, Vec<ApiToolCall>) {
parse_generated_tools_with_ids(state, text, protocol, &[])
}
pub(super) fn parse_generated_tools_with_ids(
state: &State,
text: &str,
protocol: Protocol,
streamed_ids: &[String],
) -> (String, Vec<ApiToolCall>) {
let Some((start, syntax)) = TOOL_SYNTAXES
.iter()
.filter_map(|syntax| text.find(syntax.tool_start).map(|start| (start, *syntax)))
.min_by_key(|(start, _)| *start)
else {
return (text.to_owned(), Vec::new());
};
let Some(relative_end) = text[start..].find(syntax.tool_end) else {
return (text.to_owned(), Vec::new());
};
let end = start + relative_end + syntax.tool_end.len();
let content = text[..start].trim_end();
let raw = &text[content.len()..end];
let mut calls = Vec::new();
let mut cursor = raw.find(syntax.tool_start).unwrap() + syntax.tool_start.len();
loop {
skip_text_whitespace(raw, &mut cursor);
if raw[cursor..].starts_with(syntax.tool_end) {
break;
}
if !raw[cursor..].starts_with(syntax.invoke_start) {
return (text.to_owned(), Vec::new());
}
let Some(tag_end) = raw[cursor..].find('>').map(|end| cursor + end + 1) else {
return (text.to_owned(), Vec::new());
};
let Some(name) = dsml_attribute(&raw[cursor..tag_end], "name") else {
return (text.to_owned(), Vec::new());
};
cursor = tag_end;
let mut arguments = Map::new();
loop {
skip_text_whitespace(raw, &mut cursor);
if raw[cursor..].starts_with(syntax.invoke_end) {
cursor += syntax.invoke_end.len();
break;
}
let Some((name, value)) = parse_tool_parameter(raw, &mut cursor, syntax) else {
return (text.to_owned(), Vec::new());
};
arguments.insert(name, value);
}
calls.push(ApiToolCall {
id: String::new(),
function: ApiFunction {
name,
arguments: Value::Object(arguments).to_string(),
},
});
}
if calls.is_empty() {
return (text.to_owned(), Vec::new());
}
let prefix = if protocol == Protocol::Anthropic {
"toolu_"
} else {
"call_"
};
for (index, call) in calls.iter_mut().enumerate() {
call.id = streamed_ids
.get(index)
.cloned()
.unwrap_or_else(|| random_tool_id(prefix));
}
if let Ok(mut memory) = state.tool_memory.lock() {
// ponytail: one process-local replay table; add LRU eviction if 100k live tool ids is measured insufficient.
if memory.len() >= 100_000 {
memory.clear();
}
for call in &calls {
memory.insert(call.id.clone(), raw.to_owned());
}
}
(content.to_owned(), calls)
}
#[derive(Clone, Copy)]
pub(super) struct ToolSyntax {
pub(super) tool_start: &'static str,
pub(super) tool_end: &'static str,
pub(super) invoke_start: &'static str,
pub(super) invoke_end: &'static str,
pub(super) parameter_start: &'static str,
pub(super) parameter_end: &'static str,
}
pub(super) const TOOL_SYNTAXES: [ToolSyntax; 3] = [
ToolSyntax {
tool_start: "<｜DSML｜tool_calls>",
tool_end: "</｜DSML｜tool_calls>",
invoke_start: "<｜DSML｜invoke",
invoke_end: "</｜DSML｜invoke>",
parameter_start: "<｜DSML｜parameter",
parameter_end: "</｜DSML｜parameter>",
},
ToolSyntax {
tool_start: "<DSML｜tool_calls>",
tool_end: "</DSML｜tool_calls>",
invoke_start: "<DSML｜invoke",
invoke_end: "</DSML｜invoke>",
parameter_start: "<DSML｜parameter",
parameter_end: "</DSML｜parameter>",
},
ToolSyntax {
tool_start: "<tool_calls>",
tool_end: "</tool_calls>",
invoke_start: "<invoke",
invoke_end: "</invoke>",
parameter_start: "<parameter",
parameter_end: "</parameter>",
},
];
fn parse_tool_parameter(
text: &str,
cursor: &mut usize,
syntax: ToolSyntax,
) -> Option<(String, Value)> {
if !text[*cursor..].starts_with(syntax.parameter_start) {
return None;
}
let tag_end = text[*cursor..].find('>').map(|end| *cursor + end + 1)?;
let tag = &text[*cursor..tag_end];
let name = dsml_attribute(tag, "name")?;
let is_string = dsml_attribute(tag, "string");
*cursor = tag_end;
let mut nested_start = *cursor;
skip_text_whitespace(text, &mut nested_start);
if is_string.is_none() && text[nested_start..].starts_with(syntax.parameter_start) {
*cursor = nested_start;
let mut nested = Map::new();
loop {
skip_text_whitespace(text, cursor);
if !text[*cursor..].starts_with(syntax.parameter_start) {
break;
}
let (name, value) = parse_tool_parameter(text, cursor, syntax)?;
nested.insert(name, value);
}
skip_text_whitespace(text, cursor);
if !text[*cursor..].starts_with(syntax.parameter_end) {
return None;
}
*cursor += syntax.parameter_end.len();
return Some((name, Value::Object(nested)));
}
let value_end = text[*cursor..]
.find(syntax.parameter_end)
.map(|end| *cursor + end)?;
let raw = &text[*cursor..value_end];
*cursor = value_end + syntax.parameter_end.len();
let value = if is_string.as_deref().unwrap_or("true") == "true" {
Value::String(unescape_dsml(raw))
} else {
serde_json::from_str(raw).unwrap_or(Value::Null)
};
Some((name, value))
}
fn skip_text_whitespace(text: &str, cursor: &mut usize) {
while text[*cursor..]
.chars()
.next()
.is_some_and(char::is_whitespace)
{
*cursor += text[*cursor..].chars().next().unwrap().len_utf8();
}
}
pub(super) fn tool_calls_json(calls: &[ApiToolCall]) -> Value {
Value::Array(
calls
.iter()
.map(|call| {
json!({
"id": call.id,
"type": "function",
"function": {
"name": call.function.name,
"arguments": call.function.arguments,
}
})
})
.collect(),
)
}
pub(super) fn content_text(value: &Value) -> String {
match value {
Value::String(text) => text.clone(),
Value::Array(parts) => parts
.iter()
.filter_map(|part| match part {
Value::String(text) => Some(text.as_str()),
Value::Object(object) => object.get("text").and_then(Value::as_str),
_ => None,
})
.collect(),
_ => String::new(),
}
}
pub(super) fn escape_attribute(text: &str) -> String {
text.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;")
}
pub(super) fn escape_parameter(text: &str) -> String {
text.replace("</｜DSML｜parameter>", "&lt;/｜DSML｜parameter>")
}
pub(super) fn escape_json_parameter(text: &str) -> String {
text.replace("</｜DSML｜parameter>", "\\u003c/｜DSML｜parameter>")
}
pub(super) fn escape_tool_result(text: &str) -> String {
text.replace("</tool_result>", "&lt;/tool_result>")
}
pub(super) fn unescape_dsml(text: &str) -> String {
text.replace("&quot;", "\"")
.replace("&gt;", ">")
.replace("&lt;", "<")
.replace("&amp;", "&")
}