//! Provider-facing DTOs: chat completion request, response, streaming types, //! and the SSE stream parser. //! //! # Flow //! //! 1. **Request** — [`ChatRequest`] is built with model, messages, tools, //! streaming options and sent to the LLM provider. //! 2. **Response** — Non-streaming responses arrive as [`ChatResponse`] with //! [`Choice`]s containing the full [`ChatMessage`](super::message::ChatMessage). //! 3. **Streaming** — SSE chunks are fed into [`SseParser::feed`] which yields //! [`StreamEvent`]s: token/text, reasoning, tool-call deltas, usage, done. //! //! # Components //! //! - `ChatRequest` / `StreamOptions` / `ToolDef` / `ToolFunctionDef` — outbound //! - `ChatResponse` / `Choice` / `Delta` / `TokenUsage` — non-streaming inbound //! - `StreamEvent` — one atomic streaming event (Token, Reasoning, //! ToolCallDelta, Usage, Done, Error) //! - `SseParser` — incremental SSE frame parser: `feed()` → `Vec` use serde::{Deserialize, Serialize}; use serde_json::Value; use tracing; // --------------------------------------------------------------------------- // Chat request / response // --------------------------------------------------------------------------- /// Outbound chat completion request body sent to an OpenAI/Anthropic-compatible /// provider. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChatRequest { /// Model identifier, e.g. `"anthropic/claude-opus-4-8"`. pub model: String, /// Full message history (system + user + assistant + tool turns). pub messages: Vec, /// Maximum number of output tokens. #[serde(skip_serializing_if = "Option::is_none")] pub max_tokens: Option, /// Sampling temperature (0.0 – 2.0). #[serde(skip_serializing_if = "Option::is_none")] pub temperature: Option, /// Tool definitions available to the model. #[serde(skip_serializing_if = "Option::is_none")] pub tools: Option>, /// Controls which (if any) function is called by the model. #[serde(skip_serializing_if = "Option::is_none")] pub tool_choice: Option, /// Whether to use SSE streaming (`true`) or a single response. #[serde(skip_serializing_if = "Option::is_none")] pub stream: Option, /// Nucleus sampling threshold. #[serde(skip_serializing_if = "Option::is_none")] pub top_p: Option, /// Sequences where the model should stop generation. #[serde(skip_serializing_if = "Option::is_none")] pub stop: Option>, /// Additional streaming options (e.g. `include_usage`). #[serde(skip_serializing_if = "Option::is_none")] pub stream_options: Option, } /// Streaming options for the request; `include_usage` asks the provider to /// emit a final usage chunk in the SSE stream. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StreamOptions { pub include_usage: bool, } /// Wire format for a single tool definition sent to the provider. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ToolDef { /// The tool type discriminator, e.g. `"function"`. #[serde(rename = "type")] pub type_: String, /// The function definition (name, description, JSON schema). pub function: ToolFunctionDef, } /// Name, description, and JSON schema parameters for a tool definition. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ToolFunctionDef { /// The function name the model may invoke. pub name: String, /// Human-readable description of what the function does. pub description: String, /// JSON Schema object describing the expected arguments. pub parameters: Value, } /// Non-streaming chat completion response returned by the provider. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChatResponse { /// Unique response identifier from the provider. pub id: String, /// Object type, e.g. `"chat.completion"`. pub object: Option, /// Model identifier that produced this response. pub model: String, /// One or more completion candidates. pub choices: Vec, /// Token usage statistics (prompt, completion, total). pub usage: Option, /// Unix-timestamp of response creation. pub created: Option, } /// One completion candidate within a `ChatResponse.choices` list. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Choice { /// Zero-based index of this choice in the candidate list. pub index: u32, /// Full message (non-streaming response). #[serde(skip_serializing_if = "Option::is_none")] pub message: Option, /// Incremental delta (streaming response). #[serde(skip_serializing_if = "Option::is_none")] pub delta: Option, /// Why the model stopped: `"stop"`, `"tool_calls"`, `"length"`, etc. #[serde(skip_serializing_if = "Option::is_none")] pub finish_reason: Option, } /// Incremental delta emitted in a streaming SSE chunk. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Delta { /// Role being set for the first streaming chunk. #[serde(skip_serializing_if = "Option::is_none")] pub role: Option, /// Incremental text content delta. #[serde(skip_serializing_if = "Option::is_none")] pub content: Option, /// Incremental tool-call delta (partial name/arguments). #[serde(skip_serializing_if = "Option::is_none")] pub tool_calls: Option>, } /// Token counts and optional cost breakdown for a single completion request. #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct TokenUsage { /// Tokens consumed by the prompt (input). pub prompt_tokens: u32, /// Tokens consumed by the completion (output). pub completion_tokens: u32, /// Sum of prompt + completion tokens. pub total_tokens: u32, /// Estimated cost for prompt tokens (provider-specific). #[serde(skip_serializing_if = "Option::is_none")] pub prompt_tokens_cost: Option, /// Estimated cost for completion tokens (provider-specific). #[serde(skip_serializing_if = "Option::is_none")] pub completion_tokens_cost: Option, } // --------------------------------------------------------------------------- // SSE streaming // --------------------------------------------------------------------------- /// One atomic event extracted from an LLM streaming response stream. #[derive(Debug, Clone, Serialize, Deserialize)] pub enum StreamEvent { /// An incremental text token. Token(String), /// An incremental reasoning token (Anthropic `reasoning_content`). Reasoning(String), /// An incremental tool-call delta (partial ID, name, or arguments). ToolCallDelta { /// Tool-call index (multiple calls in one response). index: usize, /// Optional tool-call ID (usually in the first delta for a call). id: Option, /// Optional function name (usually in the first delta for a call). name: Option, /// Partial JSON arguments delta for this tool call. arguments_delta: String, }, /// Final usage chunk with token counts. Usage { prompt_tokens: u64, completion_tokens: u64, total_tokens: u64, }, /// Stream complete (all tokens have been delivered). Done, /// A stream-level error occurred. Error(String), } /// Buffered SSE frame parser that accumulates raw `data:` lines and /// flushes a `StreamEvent` on each blank-line boundary. pub struct SseParser { /// Leftover bytes from the last chunk that did not end with `\n`. buffer: String, /// The current `event:` type (set by `event:` lines, cleared on flush). event_type: Option, /// Accumulated `data:` lines for the current event frame. data_lines: Vec, } impl SseParser { /// Create a new parser with an empty buffer. pub fn new() -> Self { SseParser { buffer: String::new(), event_type: None, data_lines: Vec::new(), } } /// Feed a raw SSE chunk and produce any completed events. /// /// Flow: append chunk to buffer → scan for '\n' → strip '\r' → on /// blank line, call `flush_event` to parse the accumulated data → /// on `event:` line, store the event type → on `data:` line, append /// to data accumulator → continue until buffer exhausted. /// /// Edge case: a chunk may split mid-line; the remainder stays in the /// buffer for the next `feed()` call. /// /// Return: all `StreamEvent`s completed by this chunk. pub fn feed(&mut self, chunk: &str) -> Vec { self.buffer.push_str(chunk); let mut events = Vec::new(); while let Some(line_end) = self.buffer.find('\n') { let line = self.buffer[..line_end].trim_end_matches('\r').to_string(); self.buffer = self.buffer[line_end + 1..].to_string(); if line.is_empty() { events.extend(self.flush_event()); } else if let Some(ty) = line.strip_prefix("event: ") { self.event_type = Some(ty.trim().to_string()); } else if let Some(data) = line.strip_prefix("data:") { let data = data.trim_start().to_string(); self.data_lines.push(data); } } events } /// Flush the current buffered `data:` lines as one or more `StreamEvent`s. /// /// Flow: join data lines → handle `[DONE]` sentinel → JSON-parse → /// emit `Usage` if a usage object is present → else match `event_type` /// ("message.stop", "message.delta", etc.) → extract content, /// reasoning, tool-call deltas, or finish-reason from the delta /// structure (supporting both Anthropic-style top-level delta and /// OpenAI-style `choices` array). /// /// Return: 0, 1, or more `StreamEvent`s from the flushed frame. fn flush_event(&mut self) -> Vec { let data = self.data_lines.join("\n"); self.data_lines.clear(); let event_type = self.event_type.take().unwrap_or_default(); if data.is_empty() || data == "[DONE]" { if data == "[DONE]" { return vec![StreamEvent::Done]; } return vec![]; } let value: Value = match serde_json::from_str(&data) { Ok(v) => v, Err(e) => { tracing::warn!("[stream] failed to parse chunk: {}", e); return vec![]; } }; let mut events = Vec::new(); if let Some(usage) = value.get("usage") { if !usage.is_null() { let prompt_tokens = usage .get("prompt_tokens") .and_then(Value::as_u64) .unwrap_or_else(|| { tracing::warn!("[stream] prompt_tokens missing in usage chunk"); 0 }); let completion_tokens = usage .get("completion_tokens") .and_then(Value::as_u64) .unwrap_or_else(|| { tracing::warn!("[stream] completion_tokens missing in usage chunk"); 0 }); let total_tokens = usage .get("total_tokens") .and_then(Value::as_u64) .unwrap_or_else(|| { tracing::warn!("[stream] total_tokens missing in usage chunk"); prompt_tokens + completion_tokens }); events.push(StreamEvent::Usage { prompt_tokens, completion_tokens, total_tokens, }); } } let mut other_events = match event_type.as_str() { "message.stop" => vec![StreamEvent::Done], "message.delta" | "" => { let mut d_events = Vec::new(); if let Some(delta) = value.get("delta").or_else(|| value.get("choices")) { if let Some(choices) = delta.as_array() { if let Some(choice) = choices.first() { if let Some(d) = choice.get("delta") { // Content token if let Some(content) = d.get("content").and_then(|c| c.as_str()) { d_events.push(StreamEvent::Token(content.to_string())); } // Reasoning token if let Some(reasoning) = d.get("reasoning_content").and_then(|r| r.as_str()) { d_events.push(StreamEvent::Reasoning(reasoning.to_string())); } // Tool calls — iterate ALL entries, not just first() if let Some(tool_calls) = d.get("tool_calls").and_then(|tc| tc.as_array()) { for tc in tool_calls { let index = tc.get("index").and_then(Value::as_u64).unwrap_or_else( || { tracing::warn!( "[stream] tool call delta missing index, \ defaulting to 0" ); 0 }, ); let index = usize::try_from(index).unwrap_or(0); let id = tc .get("id") .and_then(|i| i.as_str()) .map(std::string::ToString::to_string); let name = tc .get("function") .and_then(|f| f.get("name")) .and_then(|n| n.as_str()) .map(std::string::ToString::to_string); let args_delta = tc .get("function") .and_then(|f| f.get("arguments")) .and_then(|a| a.as_str()) .unwrap_or("") .to_string(); d_events.push(StreamEvent::ToolCallDelta { index, id, name, arguments_delta: args_delta, }); } } // Finish reason if let Some(reason) = choice.get("finish_reason").and_then(|r| r.as_str()) { if reason == "stop" || reason == "tool_calls" { d_events.push(StreamEvent::Done); } } } } } else if let Some(content) = delta.get("content").and_then(|c| c.as_str()) { d_events.push(StreamEvent::Token(content.to_string())); } } d_events } _ => vec![], }; events.append(&mut other_events); events } } impl Default for SseParser { fn default() -> Self { Self::new() } }