diff --git a/crates/focus-providers/Cargo.toml b/crates/focus-providers/Cargo.toml index afb333d..45fd6e6 100644 --- a/crates/focus-providers/Cargo.toml +++ b/crates/focus-providers/Cargo.toml @@ -11,3 +11,6 @@ focus-core.workspace = true focus-transport.workspace = true focus-json.workspace = true tokio = { workspace = true, features = ["rt", "sync", "macros"] } + +[dev-dependencies] +focus-harness.workspace = true diff --git a/crates/focus-providers/src/anthropic.rs b/crates/focus-providers/src/anthropic.rs index 33cb9df..2112857 100644 --- a/crates/focus-providers/src/anthropic.rs +++ b/crates/focus-providers/src/anthropic.rs @@ -1,272 +1,333 @@ -//! Anthropic Messages API provider。 -//! Anthropic Messages API provider. +//! Anthropic Messages API provider（SSE 流式）。 +//! Anthropic Messages API provider (SSE streaming). //! -//! 将 [`ProviderRequest`] 翻译为 Anthropic 的 `/v1/messages` 流式格式， -//! 并把返回的 SSE 流解析为 [`StreamEvent`]。 -//! Translates [`ProviderRequest`] into Anthropic's `/v1/messages` streaming -//! format and parses the resulting SSE stream into [`StreamEvent`]s. +//! 实现 `POST /v1/messages` 的最新协议（`anthropic-version: 2023-06-01`）， +//! 覆盖 text / thinking（扩展思考，含 signature）/ tool_use 内容块、 +//! 缓存用量（cache_creation / cache_read）与全部流式事件。 +//! Implements the latest `POST /v1/messages` protocol +//! (`anthropic-version: 2023-06-01`) with text / thinking (extended thinking, +//! incl. signatures) / tool_use content blocks, cache usage +//! (cache_creation / cache_read), and the full set of stream events. -use crate::common::{MockTransport, ProviderConfig}; +use crate::common; +use crate::config::{split_base_url, ProviderConfig}; use focus_core::model::*; -use focus_core::provider::{ - ProviderRequest, StreamEvent, StreamIterator, StreamProvider, StreamResult, -}; +use focus_core::provider::{ProviderRequest, StreamEvent, StreamProvider, StreamResult}; use focus_core::CoreError; -use focus_json::{parse, JsonValue}; -use focus_transport::{HttpTransport, Transport}; +use focus_json::JsonValue; +use focus_transport::{HttpRequest, SseEvent, SseParser, Transport}; use std::fmt; +use std::sync::mpsc::SyncSender; use std::sync::Arc; -/// Anthropic provider。`transport` 通过 trait object 泛化，以便测试注入 -/// [`MockTransport`]。 -/// The Anthropic provider. `transport` is generic via a trait object so tests -/// can inject a [`MockTransport`]. +/// Anthropic 默认 API 基址。 +/// The default Anthropic API base. +const DEFAULT_BASE_URL: &str = "https://api.anthropic.com"; + +/// 实现 Anthropic Messages API 的 provider。 +/// A provider implementing the Anthropic Messages API. +#[derive(Clone)] pub struct AnthropicProvider { - /// provider 配置（key、model、base URL 等）。 - /// Provider configuration (key, model, base URL, …). config: ProviderConfig, - /// 实际执行 HTTP 请求的 transport（生产 `HttpTransport`，测试 `MockTransport`）。 - /// Transport performing the HTTP request (`HttpTransport` in prod, `MockTransport` in tests). transport: Arc, } -impl AnthropicProvider { - /// 使用真实 HTTP transport 构造。 - /// Build with a real HTTP transport. - pub fn new(config: ProviderConfig) -> Self { - Self { - config, - transport: Arc::new(HttpTransport::new()), - } - } - - /// 使用自定义 transport 构造（用于测试）。 - /// Construct with a custom transport (for tests). - pub fn with_transport(config: ProviderConfig, transport: Arc) -> Self { - Self { config, transport } - } - - /// 测试辅助：用预置 SSE 文本的 mock transport 构造。 - /// Test helper: build with a mock transport from canned SSE text. - pub fn with_mock(config: ProviderConfig, sse_body: impl Into) -> Self { - Self::with_transport(config, Arc::new(MockTransport::new(200, sse_body))) - } - - /// 将 [`ProviderRequest`] 组装为 Anthropic `/v1/messages` 的 JSON 请求体与 - /// HTTP 请求（system 提示作为顶层字段、工具透传）。 - /// Assemble the [`ProviderRequest`] into an Anthropic `/v1/messages` JSON body - /// and HTTP request (system prompt as a top-level field, tools passed through). - fn build_request(&self, request: &ProviderRequest) -> focus_transport::HttpRequest { - let mut body = JsonValue::obj(); - body.insert("model", self.config.model.clone().into()).ok(); - body.insert( - "max_tokens", - (request.max_tokens.unwrap_or(self.config.max_tokens) as f64).into(), - ) - .ok(); - body.insert("stream", true.into()).ok(); - if !request.system_prompt.is_empty() { - body.insert("system", request.system_prompt.clone().into()) - .ok(); - } - body.insert("messages", convert_messages(&request.messages)) - .ok(); - if !tools_is_empty(&request.tools) { - body.insert("tools", request.tools.clone()).ok(); - } - - focus_transport::HttpRequest::post(&self.config.base_url, "/v1/messages") - .header("x-api-key", &self.config.api_key) - .header("anthropic-version", "2023-06-01") - .header("Accept", "text/event-stream") - .json_body(&focus_json::to_string(&body)) - } -} - impl fmt::Debug for AnthropicProvider { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("AnthropicProvider") - .field("model", &self.config.model) + .field("config", &self.config) + .field("transport", &">") .finish() } } +impl AnthropicProvider { + /// 使用默认 HTTP 传输构造 provider。 + /// Build a provider with the default HTTP transport. + pub fn new(config: ProviderConfig) -> Self { + Self::with_transport(config, Arc::new(focus_transport::HttpTransport::new())) + } + + /// 注入自定义传输（测试用 mock）构造 provider。 + /// Build a provider with an injected transport (e.g. a mock in tests). + pub fn with_transport(config: ProviderConfig, transport: Arc) -> Self { + Self { config, transport } + } + + /// 依据配置解析目标 (host, port, path_prefix)。 + /// Resolve the target (host, port, path_prefix) from the config. + fn endpoint(&self) -> Result<(String, u16, String), String> { + let base = self + .config + .base_url + .clone() + .unwrap_or_else(|| DEFAULT_BASE_URL.to_string()); + let (host, port, prefix) = split_base_url(&base)?; + let path = format!("{}/v1/messages", prefix); + Ok((host, port, path)) + } + + /// 构造请求体 JSON。 + /// Build the request-body JSON. + fn build_body(&self, request: &ProviderRequest) -> Result { + let mut body = JsonValue::obj(); + body.insert("model", request.model.clone().into()).ok(); + if !request.system_prompt.is_empty() { + body.insert("system", request.system_prompt.clone().into()) + .ok(); + } + body.insert("messages", messages_to_anthropic(&request.messages)) + .ok(); + if has_tools(&request.tools) { + body.insert("tools", tools_to_anthropic(&request.tools)) + .ok(); + } + // Anthropic 强制要求 max_tokens；未配置时用默认值。 + // Anthropic requires max_tokens; fall back to a default when unset. + let max_tokens = request.max_tokens.unwrap_or(4096); + body.insert("max_tokens", (max_tokens as f64).into()).ok(); + if let Some(t) = request.temperature { + body.insert("temperature", t.into()).ok(); + } + body.insert("stream", true.into()).ok(); + Ok(focus_json::to_string(&body)) + } +} + impl StreamProvider for AnthropicProvider { fn stream(&self, request: &ProviderRequest) -> StreamResult { - let http_req = self.build_request(request); - // 我们在此同步执行请求（在一个小型 runtime 上阻塞），因为 core 的 - // StreamIterator 是同步拉取模型。 - // We perform the request synchronously here (blocking on a mini - // runtime) because the core's StreamIterator is a sync pull model. - let response = block_on_request(&self.transport, &http_req)?; - let body = response.body_string(); - Ok(Box::new(AnthropicStream::new(&body, &self.config.model))) + let (host, port, path) = self.endpoint().map_err(CoreError::Provider)?; + let body = self.build_body(request).map_err(CoreError::Provider)?; + let model = request.model.clone(); + let req = HttpRequest::post(&host, &path) + .port(port) + .header("x-api-key", &self.config.api_key) + .header("anthropic-version", "2023-06-01") + .header("accept", "text/event-stream") + .json_body(&body); + let transport = self.transport.clone(); + let timeout = self.config.timeout; + + common::run_stream(move |tx| { + Box::pin(async move { + let model = model.clone(); + let result: Result<(), String> = async { + let mut chunks = transport + .stream_request(&req, timeout) + .await + .map_err(|e| e.to_string())?; + let mut sse = SseParser::new(); + let mut turn = AnthropicTurn::new(&model); + loop { + let chunk = chunks.next_chunk().await.map_err(|e| e.to_string())?; + match chunk { + Some(bytes) => { + sse.feed(&bytes); + while let Some(ev) = sse.next_event() { + match turn.handle_sse(&ev, &tx) { + Ok(true) => {} + Ok(false) => return Ok(()), // 消费者已离开 / consumer gone + Err(msg) => { + let _ = tx.send(StreamEvent::Error { + error: common::error_message(&model, &msg), + }); + return Ok(()); + } + } + if turn.done { + return Ok(()); + } + } + } + None => break, // 流结束 / end of stream + } + } + if !turn.done { + return Err("stream ended without a message_stop event".to_string()); + } + Ok(()) + } + .await; + if let Err(msg) = result { + let _ = tx.send(StreamEvent::Error { + error: common::error_message(&model, &msg), + }); + } + }) + }) } } -/// 在一个临时单线程 tokio runtime 上同步阻塞地执行一次 HTTP 请求。 -/// Run one HTTP request synchronously by blocking on a throwaway single-thread tokio runtime. -fn block_on_request( - transport: &Arc, - req: &focus_transport::HttpRequest, -) -> Result { - let rt = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .map_err(|e| CoreError::Provider(format!("failed to build runtime: {}", e)))?; - rt.block_on(async { transport.request(req).await }) - .map_err(|e| CoreError::Provider(e.to_string())) -} - -// ---- 消息转换 ---- -// ---- message conversion --------------------------------------------------- - -/// 将统一消息列表转为 Anthropic 的 messages 数组（ToolResult 映射为 -/// 带 `tool_result` 块的 user 消息）。 -/// Convert the unified message list into Anthropic's messages array (a -/// ToolResult maps to a user message holding a `tool_result` block). -fn convert_messages(messages: &[Message]) -> JsonValue { - let mut arr = JsonValue::arr(); - for m in messages { - match m { +/// 把核心消息列表翻译成 Anthropic messages 数组。 +/// Translate core messages into the Anthropic messages array. +/// +/// 注意：多个连续的 tool_result 必须合并进**一条** user 消息（Anthropic +/// 要求 role 交替出现）。 +/// Note: consecutive tool_results must be merged into **one** user message +/// (Anthropic requires alternating roles). +fn messages_to_anthropic(messages: &[Message]) -> JsonValue { + let mut out = JsonValue::arr(); + let mut i = 0; + while i < messages.len() { + match &messages[i] { + Message::ToolResult(_) => { + // 合并连续的 tool_result。 + // Merge consecutive tool_results. + let mut blocks = JsonValue::arr(); + while i < messages.len() { + match &messages[i] { + Message::ToolResult(t) => { + let mut b = JsonValue::obj(); + b.insert("type", "tool_result".into()).ok(); + b.insert("tool_use_id", t.tool_call_id.clone().into()).ok(); + let content = content_to_anthropic(&t.content); + // 空内容以空字符串代替（API 不接受空数组）。 + // Empty content is sent as an empty string (the API + // rejects empty arrays here). + let content = match content { + JsonValue::Arr(ref a) if a.is_empty() => "".into(), + other => other, + }; + b.insert("content", content).ok(); + b.insert("is_error", t.is_error.into()).ok(); + blocks.push(b).ok(); + i += 1; + } + _ => break, + } + } + let mut o = JsonValue::obj(); + o.insert("role", "user".into()).ok(); + o.insert("content", blocks).ok(); + out.push(o).ok(); + } Message::User(u) => { let mut o = JsonValue::obj(); o.insert("role", "user".into()).ok(); - o.insert("content", convert_content_out(&u.content)).ok(); - arr.push(o).ok(); + o.insert("content", content_to_anthropic(&u.content)).ok(); + out.push(o).ok(); + i += 1; } Message::Assistant(a) => { let mut o = JsonValue::obj(); o.insert("role", "assistant".into()).ok(); - o.insert("content", convert_content_out(&a.content)).ok(); - arr.push(o).ok(); - } - Message::ToolResult(t) => { - let mut o = JsonValue::obj(); - o.insert("role", "user".into()).ok(); - let mut tr = JsonValue::obj(); - tr.insert("type", "tool_result".into()).ok(); - tr.insert("tool_use_id", t.tool_call_id.clone().into()).ok(); - tr.insert("content", convert_content_out(&t.content)).ok(); - tr.insert("is_error", t.is_error.into()).ok(); - let mut content_arr = JsonValue::arr(); - content_arr.push(tr).ok(); - o.insert("content", content_arr).ok(); - arr.push(o).ok(); + o.insert("content", content_to_anthropic(&a.content)).ok(); + out.push(o).ok(); + i += 1; } } } - arr + out } -/// 将内容块转为 Anthropic 的结构化 content 数组（text/thinking/image/tool_use）。 -/// Convert content blocks into Anthropic's structured content array -/// (text/thinking/image/tool_use). -fn convert_content_out(blocks: &[ContentBlock]) -> JsonValue { +/// 把核心内容块翻译成 Anthropic content 数组。 +/// Translate core content blocks into the Anthropic content array. +fn content_to_anthropic(blocks: &[ContentBlock]) -> JsonValue { let mut arr = JsonValue::arr(); for b in blocks { + let mut o = JsonValue::obj(); match b { ContentBlock::Text(t) => { - let mut o = JsonValue::obj(); o.insert("type", "text".into()).ok(); o.insert("text", t.text.clone().into()).ok(); - arr.push(o).ok(); } ContentBlock::Thinking(t) => { - let mut o = JsonValue::obj(); o.insert("type", "thinking".into()).ok(); o.insert("thinking", t.thinking.clone().into()).ok(); - arr.push(o).ok(); + if let Some(sig) = &t.signature { + o.insert("signature", sig.clone().into()).ok(); + } } ContentBlock::Image(i) => { - let mut o = JsonValue::obj(); o.insert("type", "image".into()).ok(); - let mut src = JsonValue::obj(); - src.insert("type", "base64".into()).ok(); - src.insert("media_type", i.mime_type.clone().into()).ok(); - src.insert("data", i.data.clone().into()).ok(); - o.insert("source", src).ok(); - arr.push(o).ok(); + let mut source = JsonValue::obj(); + source.insert("type", "base64".into()).ok(); + source.insert("media_type", i.mime_type.clone().into()).ok(); + source.insert("data", i.data.clone().into()).ok(); + o.insert("source", source).ok(); } ContentBlock::ToolCall(tc) => { - let mut o = JsonValue::obj(); o.insert("type", "tool_use".into()).ok(); o.insert("id", tc.id.clone().into()).ok(); o.insert("name", tc.name.clone().into()).ok(); o.insert("input", tc.arguments.clone()).ok(); - arr.push(o).ok(); } } + arr.push(o).ok(); } arr } -/// 工具数组是否为空（缺失或空数组均视为空）。 -/// Whether the tools value is empty (missing or empty array both count). -fn tools_is_empty(tools: &JsonValue) -> bool { - matches!(tools, JsonValue::Arr(a) if a.is_empty()) +/// 把核心工具定义翻译成 Anthropic tools 数组（`input_schema` 字段）。 +/// Translate core tool definitions into the Anthropic tools array +/// (the `input_schema` field). +fn tools_to_anthropic(tools: &JsonValue) -> JsonValue { + match tools { + JsonValue::Arr(items) => { + let mut arr = JsonValue::arr(); + for t in items { + let mut o = JsonValue::obj(); + if let Some(n) = t.get_str("name") { + o.insert("name", n.into()).ok(); + } + if let Some(d) = t.get_str("description") { + o.insert("description", d.into()).ok(); + } + if let Some(p) = t.get("parameters") { + o.insert("input_schema", p.clone()).ok(); + } + arr.push(o).ok(); + } + arr + } + _ => JsonValue::arr(), + } } -// ---- SSE 转 StreamEvent ---- -// ---- SSE -> StreamEvent --------------------------------------------------- +fn has_tools(tools: &JsonValue) -> bool { + matches!(tools, JsonValue::Arr(items) if !items.is_empty()) +} -/// 驱动 Anthropic 的 SSE 响应体并产出 [`StreamEvent`]。 -/// Drives an Anthropic SSE body and emits [`StreamEvent`]s. -pub struct AnthropicStream { - /// 增量 SSE 解析器。 - /// Incremental SSE parser. - parser: focus_transport::SseParser, - /// 当前模型名（用于构造 partial 消息）。 - /// Current model name (used when building partial messages). +/// 把 Anthropic 的 stop_reason 映射到核心类型。 +/// Map an Anthropic stop_reason onto the core type. +fn map_stop_reason(s: &str) -> StopReason { + match s { + "max_tokens" => StopReason::Length, + "tool_use" => StopReason::ToolUse, + _ => StopReason::Stop, // end_turn / stop_sequence / pause_turn + } +} + +/// 单个回合的累积状态：负责把 SSE 事件翻译成 [`StreamEvent`]。 +/// Per-turn accumulator: translates SSE events into [`StreamEvent`]s. +struct AnthropicTurn { model: String, - /// 是否已收到终止事件（message_stop / error）。 - /// Whether a terminal event (message_stop / error) has been seen. - done: bool, - // 累加状态。 - // Accumulator state. - /// 已累积的内容块（按 Anthropic 的 index 顺序）。 - /// Content blocks accumulated so far (in Anthropic's index order). content: Vec, - /// 每个 tool_use 块尚未解析的 partial_json 片段。 - /// Per-block (tool_use) not-yet-parsed partial_json fragments. - partial_json: Vec, - /// 当前用量统计。 - /// Current usage counters. + /// 与 `content` 对齐的工具调用参数缓冲。 + /// Tool-call argument buffer aligned with `content`. + args: Vec>, usage: Usage, - /// 当前停止原因。 - /// Current stop reason. stop_reason: StopReason, - /// 若收到 error 事件，记录其消息。 - /// Message captured from an error event, if any. error_message: Option, - /// 已产生但尚未被 `next_event` 取走的事件缓冲。 - /// Buffer of produced events not yet drained by `next_event`. - pending: std::collections::VecDeque, + done: bool, } -impl AnthropicStream { - /// 用完整响应体字符串与模型名构造（SSE 一次性喂入）。 - /// Build from the full response-body string and model name (whole SSE body fed at once). - fn new(body: &str, model: &str) -> Self { - let mut parser = focus_transport::SseParser::new(); - parser.feed_str(body); +impl AnthropicTurn { + fn new(model: &str) -> Self { Self { - parser, model: model.to_string(), - done: false, content: Vec::new(), - partial_json: Vec::new(), + args: Vec::new(), usage: Usage::default(), stop_reason: StopReason::Stop, error_message: None, - pending: std::collections::VecDeque::new(), + done: false, } } - /// 用当前累加状态快照构造一条 [`AssistantMessage`]（作为事件中的 partial）。 - /// Build an [`AssistantMessage`] snapshot from the current accumulator state - /// (used as the `partial` field on events). - fn current_partial(&self) -> AssistantMessage { + /// 当前累积状态的 partial 消息快照。 + /// Partial message snapshot of the current accumulated state. + fn partial(&self) -> AssistantMessage { AssistantMessage { content: self.content.clone(), model: self.model.clone(), @@ -277,293 +338,337 @@ impl AnthropicStream { } } - /// 处理一个 Anthropic SSE 事件（按 `event_type` 分派），返回本次产生的事件列表。 - /// Handle one Anthropic SSE event (dispatched by `event_type`); returns the - /// list of events it produced. - fn handle_event(&mut self, event_type: &str, data: &str) -> Vec { - let mut out = Vec::new(); - let value = match parse(data) { - Ok(v) => v, - Err(_) => return out, + /// 处理一个 SSE 事件。返回 `Ok(true)` 继续、`Ok(false)` 消费者已离开、 + /// `Err(msg)` 数据畸形（调用方应发出 Error 事件并中止）。 + /// Handle one SSE event. Returns `Ok(true)` to continue, `Ok(false)` when + /// the consumer is gone, and `Err(msg)` on malformed data (the caller + /// should emit an Error event and abort). + fn handle_sse(&mut self, ev: &SseEvent, tx: &SyncSender) -> Result { + let data = ev.data.trim(); + let json: JsonValue = if data.is_empty() { + JsonValue::obj() + } else { + focus_json::parse(data).map_err(|e| format!("invalid SSE JSON: {}", e))? }; - match event_type { + match ev.event.as_str() { "message_start" => { - // 如存在 message.usage，则从中初始化用量统计。 - // seed usage from message.usage if present - if let Some(u) = value.get("message").and_then(|m| m.get("usage")) { - self.usage.input_tokens = u.get_num("input_tokens").unwrap_or(0.0) as u64; - self.usage.output_tokens = u.get_num("output_tokens").unwrap_or(0.0) as u64; - self.usage.cache_read_tokens = - u.get_num("cache_read_input_tokens").unwrap_or(0.0) as u64; - self.usage.cache_write_tokens = - u.get_num("cache_creation_input_tokens").unwrap_or(0.0) as u64; + if let Some(message) = json.get("message") { + if let Some(u) = message.get("usage") { + self.usage.input_tokens = u.get_num("input_tokens").unwrap_or(0.0) as u64; + self.usage.cache_write_tokens = + u.get_num("cache_creation_input_tokens").unwrap_or(0.0) as u64; + self.usage.cache_read_tokens = + u.get_num("cache_read_input_tokens").unwrap_or(0.0) as u64; + } } - out.push(StreamEvent::Start { - partial: self.current_partial(), - }); + Ok(common::send( + tx, + StreamEvent::Start { + partial: self.partial(), + }, + )) } "content_block_start" => { - let idx = value.get_num("index").unwrap_or(0.0) as usize; - if let Some(block) = value.get("content_block") { - let ty = block.get_str("type").unwrap_or(""); - match ty { - "text" => { - self.content.push(ContentBlock::Text(TextContent { - text: String::new(), - signature: None, - })); - self.partial_json.push(String::new()); - out.push(StreamEvent::TextStart { - content_index: idx, - partial: self.current_partial(), - }); - } - "thinking" => { - self.content.push(ContentBlock::Thinking(ThinkingContent { - thinking: String::new(), - signature: None, - redacted: false, - })); - self.partial_json.push(String::new()); - out.push(StreamEvent::ThinkingStart { - content_index: idx, - partial: self.current_partial(), - }); - } - "tool_use" => { - self.content.push(ContentBlock::ToolCall(ToolCall { - id: block.get_str("id").unwrap_or("").to_string(), - name: block.get_str("name").unwrap_or("").to_string(), - arguments: JsonValue::obj(), - })); - self.partial_json.push(String::new()); - out.push(StreamEvent::ToolCallStart { - content_index: idx, - partial: self.current_partial(), - }); - } - _ => {} + let index = json.get_num("index").unwrap_or(0.0) as usize; + let block = json.get("content_block").cloned().unwrap_or_default(); + match block.get_str("type").unwrap_or("") { + "text" => { + self.content.push(ContentBlock::Text(TextContent { + text: String::new(), + signature: None, + })); + self.args.push(None); + Ok(common::send( + tx, + StreamEvent::TextStart { + content_index: index, + partial: self.partial(), + }, + )) } + "thinking" => { + self.content.push(ContentBlock::Thinking(ThinkingContent { + thinking: String::new(), + signature: block.get_str("signature").map(String::from), + redacted: false, + })); + self.args.push(None); + Ok(common::send( + tx, + StreamEvent::ThinkingStart { + content_index: index, + partial: self.partial(), + }, + )) + } + "tool_use" => { + let id = block.get_str("id").unwrap_or("").to_string(); + let name = block.get_str("name").unwrap_or("").to_string(); + self.content.push(ContentBlock::ToolCall(ToolCall { + id, + name, + arguments: JsonValue::obj(), + })); + self.args.push(Some(String::new())); + Ok(common::send( + tx, + StreamEvent::ToolCallStart { + content_index: index, + partial: self.partial(), + }, + )) + } + _ => Ok(true), } } "content_block_delta" => { - let idx = value.get_num("index").unwrap_or(0.0) as usize; - if let Some(delta) = value.get("delta") { - let ty = delta.get_str("type").unwrap_or(""); - match ty { - "text_delta" => { - let text = delta.get_str("text").unwrap_or("").to_string(); - if let Some(ContentBlock::Text(t)) = self.content.get_mut(idx) { - t.text.push_str(&text); - } - out.push(StreamEvent::TextDelta { - content_index: idx, + let index = json.get_num("index").unwrap_or(0.0) as usize; + let delta = json.get("delta").cloned().unwrap_or_default(); + match delta.get_str("type").unwrap_or("") { + "text_delta" => { + let text = delta.get_str("text").unwrap_or("").to_string(); + if let Some(ContentBlock::Text(t)) = self.content.get_mut(index) { + t.text.push_str(&text); + } + Ok(common::send( + tx, + StreamEvent::TextDelta { + content_index: index, delta: text, - partial: self.current_partial(), - }); - } - "thinking_delta" => { - let text = delta.get_str("thinking").unwrap_or("").to_string(); - if let Some(ContentBlock::Thinking(t)) = self.content.get_mut(idx) { - t.thinking.push_str(&text); - } - out.push(StreamEvent::ThinkingDelta { - content_index: idx, - delta: text, - partial: self.current_partial(), - }); - } - "input_json_delta" => { - let pj = delta.get_str("partial_json").unwrap_or("").to_string(); - if let Some(slot) = self.partial_json.get_mut(idx) { - slot.push_str(&pj); - } - out.push(StreamEvent::ToolCallDelta { - content_index: idx, - delta: pj, - partial: self.current_partial(), - }); - } - _ => {} + partial: self.partial(), + }, + )) } + "thinking_delta" => { + let thinking = delta.get_str("thinking").unwrap_or("").to_string(); + if let Some(ContentBlock::Thinking(t)) = self.content.get_mut(index) { + t.thinking.push_str(&thinking); + } + Ok(common::send( + tx, + StreamEvent::ThinkingDelta { + content_index: index, + delta: thinking, + partial: self.partial(), + }, + )) + } + "input_json_delta" => { + let partial = delta.get_str("partial_json").unwrap_or("").to_string(); + if let Some(slot) = self.args.get_mut(index).and_then(|s| s.as_mut()) { + slot.push_str(&partial); + } + Ok(common::send( + tx, + StreamEvent::ToolCallDelta { + content_index: index, + delta: partial, + partial: self.partial(), + }, + )) + } + _ => Ok(true), } } "content_block_stop" => { - let idx = value.get_num("index").unwrap_or(0.0) as usize; - match self.content.get_mut(idx) { - Some(ContentBlock::Text(_)) => out.push(StreamEvent::TextEnd { - content_index: idx, - partial: self.current_partial(), - }), - Some(ContentBlock::Thinking(_)) => out.push(StreamEvent::ThinkingEnd { - content_index: idx, - partial: self.current_partial(), - }), - Some(ContentBlock::ToolCall(tc)) => { - // 用累积的 partial json 最终确定工具参数。 - // finalize arguments from accumulated partial json - let pj = self.partial_json.get(idx).cloned().unwrap_or_default(); - tc.arguments = parse(&pj).unwrap_or(JsonValue::obj()); - let finalized_tc = tc.clone(); - out.push(StreamEvent::ToolCallEnd { - content_index: idx, - tool_call: finalized_tc, - partial: self.current_partial(), - }); + let index = json.get_num("index").unwrap_or(0.0) as usize; + match self.content.get(index) { + Some(ContentBlock::Text(_)) => Ok(common::send( + tx, + StreamEvent::TextEnd { + content_index: index, + partial: self.partial(), + }, + )), + Some(ContentBlock::Thinking(_)) => Ok(common::send( + tx, + StreamEvent::ThinkingEnd { + content_index: index, + partial: self.partial(), + }, + )), + Some(ContentBlock::ToolCall(_)) => { + // 工具调用结束时解析完整参数。 + // Parse the full arguments when the tool call ends. + let args = self + .args + .get(index) + .and_then(|s| s.as_ref()) + .cloned() + .unwrap_or_default(); + let parsed = focus_json::parse(&args).unwrap_or(JsonValue::obj()); + if let Some(ContentBlock::ToolCall(tc)) = self.content.get_mut(index) { + tc.arguments = parsed; + } + let tool_call = match self.content.get(index) { + Some(ContentBlock::ToolCall(tc)) => tc.clone(), + _ => ToolCall { + id: String::new(), + name: String::new(), + arguments: JsonValue::obj(), + }, + }; + Ok(common::send( + tx, + StreamEvent::ToolCallEnd { + content_index: index, + tool_call, + partial: self.partial(), + }, + )) } - _ => {} + _ => Ok(true), } } "message_delta" => { - if let Some(d) = value.get("delta") { - if let Some(sr) = d.get_str("stop_reason") { + if let Some(delta) = json.get("delta") { + if let Some(sr) = delta.get_str("stop_reason") { self.stop_reason = map_stop_reason(sr); } } - if let Some(u) = value.get("usage") { - if let Some(o) = u.get_num("output_tokens") { - self.usage.output_tokens = o as u64; - } + if let Some(u) = json.get("usage") { + self.usage.output_tokens = u.get_num("output_tokens").unwrap_or(0.0) as u64; } + Ok(true) } "message_stop" => { self.done = true; - out.push(StreamEvent::Done { - message: self.current_partial(), - }); + Ok(common::send( + tx, + StreamEvent::Done { + message: self.partial(), + }, + )) } "error" => { self.done = true; - let msg = value - .get("error") - .and_then(|e| e.get_str("message")) - .unwrap_or("unknown error") - .to_string(); - self.error_message = Some(msg.clone()); - self.stop_reason = StopReason::Error; - out.push(StreamEvent::Error { - error: self.current_partial(), - }); + let err = json.get("error").cloned().unwrap_or_default(); + let ty = err.get_str("type").unwrap_or("error"); + let msg = err.get_str("message").unwrap_or("unknown error"); + Ok(common::send( + tx, + StreamEvent::Error { + error: common::error_message(&self.model, &format!("{}: {}", ty, msg)), + }, + )) } - _ => {} + _ => Ok(true), // ping 等事件忽略 / ignore events like ping } - out - } -} - -impl StreamIterator for AnthropicStream { - /// 拉取下一个事件：先排空 `pending`；缓冲空则取下一个 SSE 事件并处理。 - /// Pull the next event: drain `pending` first; when empty, pull the next - /// SSE event and handle it. - fn next_event(&mut self) -> Option { - loop { - if let Some(ev) = self.pending.pop_front() { - return Some(ev); - } - if let Some(sse) = self.parser.next_event() { - let events = self.handle_event(&sse.event, &sse.data); - self.pending.extend(events); - if self.done && self.pending.is_empty() { - return None; - } - continue; - } - return None; - } - } -} - -/// 将 Anthropic 的 stop_reason 字符串映射为统一的 [`StopReason`]。 -/// Map an Anthropic stop_reason string to the unified [`StopReason`]. -fn map_stop_reason(s: &str) -> StopReason { - match s { - "end_turn" | "stop_sequence" | "pause_turn" => StopReason::Stop, - "max_tokens" => StopReason::Length, - "tool_use" => StopReason::ToolUse, - _ => StopReason::Stop, } } #[cfg(test)] mod tests { use super::*; + use focus_core::tool::{ToolEffects, ToolRegistry}; - fn make_provider(sse: &str) -> AnthropicProvider { - AnthropicProvider::with_mock(ProviderConfig::anthropic("test-key", "claude-test"), sse) + /// 无副作用 echo 工具（供请求体测试）。 + /// Side-effect-free echo tool (for request-body tests). + #[derive(Debug)] + struct EchoTool; + impl focus_core::Tool for EchoTool { + fn name(&self) -> &str { + "echo" + } + fn description(&self) -> &str { + "echoes text" + } + fn parameters(&self) -> JsonValue { + let mut o = JsonValue::obj(); + o.insert("type", "object".into()).ok(); + let mut props = JsonValue::obj(); + props.insert("text", JsonValue::Str("the text".into())).ok(); + o.insert("properties", props).ok(); + o + } + fn effects(&self) -> ToolEffects { + ToolEffects::READ + } + fn execute( + &self, + _id: &str, + args: &JsonValue, + _u: Option<&focus_core::tool::ToolUpdateSink>, + ) -> Result { + Ok(focus_core::tool::ToolResult::text( + args.get_str("text").unwrap_or("").to_string(), + )) + } } #[test] - fn parses_text_response() { - let sse = "\ -event: message_start\n\ -data: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":10,\"output_tokens\":1}}}\n\n\ -event: content_block_start\n\ -data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n\ -event: content_block_delta\n\ -data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"Hello\"}}\n\n\ -event: content_block_stop\n\ -data: {\"type\":\"content_block_stop\",\"index\":0}\n\n\ -event: message_delta\n\ -data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":5}}\n\n\ -event: message_stop\n\ -data: {\"type\":\"message_stop\"}\n\n"; - let provider = make_provider(sse); - let req = ProviderRequest { - model: "claude-test".into(), - system_prompt: "".into(), - messages: vec![Message::user_text("hi")], - tools: JsonValue::arr(), - max_tokens: Some(100), - temperature: None, + fn serializes_request_body() { + let provider = AnthropicProvider::new(ProviderConfig::new("sk-test")); + let request = ProviderRequest { + model: "claude-sonnet-4".into(), + system_prompt: "You are helpful.".into(), + messages: vec![ + Message::user_text("hi"), + Message::Assistant(AssistantMessage { + content: vec![ + ContentBlock::Thinking(ThinkingContent { + thinking: "hmm".into(), + signature: Some("sig".into()), + redacted: false, + }), + ContentBlock::ToolCall(ToolCall { + id: "call_1".into(), + name: "echo".into(), + arguments: focus_json::parse(r#"{"text":"x"}"#).unwrap(), + }), + ], + model: "claude-sonnet-4".into(), + usage: Usage::default(), + stop_reason: StopReason::ToolUse, + error_message: None, + timestamp: 0, + }), + Message::ToolResult(ToolResultMessage { + tool_call_id: "call_1".into(), + tool_name: "echo".into(), + content: vec![ContentBlock::text("ok")], + details: JsonValue::obj(), + is_error: false, + timestamp: 0, + }), + ], + tools: ToolRegistry::with(Box::new(EchoTool)).tool_definitions(), + max_tokens: Some(1024), + temperature: Some(0.5), }; - let mut iter = provider.stream(&req).unwrap(); - let mut text = String::new(); - let mut final_stop = StopReason::Stop; - while let Some(ev) = iter.next_event() { - match ev { - StreamEvent::TextDelta { delta, .. } => text.push_str(&delta), - StreamEvent::Done { message } => final_stop = message.stop_reason, - _ => {} - } - } - assert_eq!(text, "Hello"); - assert_eq!(final_stop, StopReason::Stop); + let body = provider.build_body(&request).unwrap(); + let json: JsonValue = focus_json::parse(&body).unwrap(); + assert_eq!(json.get_str("model"), Some("claude-sonnet-4")); + assert_eq!(json.get_str("system"), Some("You are helpful.")); + assert_eq!(json.get_bool("stream"), Some(true)); + // 工具被翻译成 input_schema。 + // Tools are translated with input_schema. + let tools = json.get_arr("tools").unwrap(); + assert_eq!(tools[0].get_str("name"), Some("echo")); + assert!(tools[0].get("input_schema").is_some()); + // 思考块带签名回传。 + // Thinking blocks are replayed with their signature. + let messages = json.get_arr("messages").unwrap(); + let assistant = &messages[1]; + let content = assistant.get_arr("content").unwrap(); + assert_eq!(content[0].get_str("type"), Some("thinking")); + assert_eq!(content[0].get_str("signature"), Some("sig")); + assert_eq!(content[1].get_str("type"), Some("tool_use")); + assert_eq!(content[1].get_str("id"), Some("call_1")); + // tool_result 落在一条 user 消息里。 + // tool_result lands inside one user message. + let tool_msg = &messages[2]; + assert_eq!(tool_msg.get_str("role"), Some("user")); + let blocks = tool_msg.get_arr("content").unwrap(); + assert_eq!(blocks[0].get_str("type"), Some("tool_result")); + assert_eq!(blocks[0].get_str("tool_use_id"), Some("call_1")); } #[test] - fn parses_tool_use_response() { - let sse = "\ -event: message_start\n\ -data: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":5,\"output_tokens\":1}}}\n\n\ -event: content_block_start\n\ -data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"call_1\",\"name\":\"read\"}}\n\n\ -event: content_block_delta\n\ -data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"path\\\":\\\"/x\\\"}\"}}\n\n\ -event: content_block_stop\n\ -data: {\"type\":\"content_block_stop\",\"index\":0}\n\n\ -event: message_delta\n\ -data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":10}}\n\n\ -event: message_stop\n\ -data: {\"type\":\"message_stop\"}\n\n"; - let provider = make_provider(sse); - let req = ProviderRequest { - model: "claude-test".into(), - system_prompt: "".into(), - messages: vec![Message::user_text("read")], - tools: JsonValue::arr(), - max_tokens: Some(100), - temperature: None, - }; - let mut iter = provider.stream(&req).unwrap(); - let mut tool_call: Option = None; - while let Some(ev) = iter.next_event() { - if let StreamEvent::ToolCallEnd { tool_call: tc, .. } = ev { - tool_call = Some(tc); - } - } - let tc = tool_call.expect("no tool call"); - assert_eq!(tc.id, "call_1"); - assert_eq!(tc.name, "read"); - assert_eq!(tc.arguments.get_str("path"), Some("/x")); + fn maps_stop_reasons() { + assert_eq!(map_stop_reason("end_turn"), StopReason::Stop); + assert_eq!(map_stop_reason("max_tokens"), StopReason::Length); + assert_eq!(map_stop_reason("tool_use"), StopReason::ToolUse); + assert_eq!(map_stop_reason("pause_turn"), StopReason::Stop); } } diff --git a/crates/focus-providers/src/common.rs b/crates/focus-providers/src/common.rs index 5cd293d..d135842 100644 --- a/crates/focus-providers/src/common.rs +++ b/crates/focus-providers/src/common.rs @@ -1,140 +1,94 @@ -//! provider 实现的共享辅助：请求配置、用于测试的 mock transport， -//! 以及从配置组装 model-id 的辅助函数。 -//! Shared helpers for provider implementations: request config, a mock -//! transport for tests, and helpers to assemble the model-id from config. +//! provider 共享的流式驱动与辅助类型。 +//! Shared streaming driver and helper types for providers. +//! +//! `focus-core` 的 agent 循环是同步拉取式（[`StreamIterator`]），而网络 I/O 是 +//! 异步的。本模块在后台线程上创建一个 current-thread tokio 运行时来驱动网络， +//! 通过同步 channel 把 [`StreamEvent`] 送达主线程的迭代器，从而把两种模型粘合起来。 +//! The core agent loop is a synchronous pull model ([`StreamIterator`]) while +//! network I/O is async. This module spins up a current-thread tokio runtime on +//! a background thread to drive the network, delivering [`StreamEvent`]s to the +//! main thread's iterator through a sync channel — gluing the two models. -use focus_core::ProviderRequest; -use focus_transport::{HttpRequest, HttpResponse, Transport}; -use std::collections::HashMap; -use std::sync::{Arc, Mutex}; +use focus_core::model::{now_ms, AssistantMessage, StopReason, Usage}; +use focus_core::provider::{StreamEvent, StreamIterator, StreamResult}; +use focus_core::CoreError; +use std::future::Future; +use std::pin::Pin; +use std::sync::mpsc::{Receiver, SyncSender}; -/// provider 配置：API key、model id、base URL。 -/// Configuration for a provider: API key, model id, base URL. -#[derive(Debug, Clone)] -pub struct ProviderConfig { - /// API 密钥（或待解析的 `$VAR`/`${VAR}` 模板）。 - /// API key (or a `$VAR`/`${VAR}` template to be resolved). - pub api_key: String, - /// 模型 id，例如 `claude-sonnet-4-5`、`glm-5.2`。 - /// Model id, e.g. `claude-sonnet-4-5`, `glm-5.2`. - pub model: String, - /// 仅主机名（不含 scheme、不含 path），例如 `api.anthropic.com`。 - /// Host only (no scheme, no path), e.g. `api.anthropic.com`. - pub base_url: String, - /// 默认最大输出 token 数。 - /// Default max output tokens. - pub max_tokens: u64, - /// 请求超时（毫秒）。 - /// Request timeout in milliseconds. - pub timeout_ms: u64, -} - -impl ProviderConfig { - /// Anthropic 默认配置（`api.anthropic.com`，4096 tokens，120s 超时）。 - /// Default Anthropic config (`api.anthropic.com`, 4096 tokens, 120s timeout). - pub fn anthropic(api_key: impl Into, model: impl Into) -> Self { - Self { - api_key: api_key.into(), - model: model.into(), - base_url: "api.anthropic.com".into(), - max_tokens: 4096, - timeout_ms: 120_000, - } - } - - /// OpenAI 默认配置（`api.openai.com`，4096 tokens，120s 超时）。 - /// Default OpenAI config (`api.openai.com`, 4096 tokens, 120s timeout). - pub fn openai(api_key: impl Into, model: impl Into) -> Self { - Self { - api_key: api_key.into(), - model: model.into(), - base_url: "api.openai.com".into(), - max_tokens: 4096, - timeout_ms: 120_000, - } - } - - /// 为任意 OpenAI 兼容厂商构造一份通用配置。`base_url` 仅为主机名 - /// （不含 scheme、不含 path）。 - /// Build a generic config for an OpenAI-compatible vendor. `base_url` is - /// the host only (no scheme, no path). - pub fn new( - api_key: impl Into, - model: impl Into, - base_url: impl Into, - ) -> Self { - Self { - api_key: api_key.into(), - model: model.into(), - base_url: base_url.into(), - max_tokens: 4096, - timeout_ms: 120_000, - } - } - - /// Z.ai / 智谱 BigModel 配置（国内端点）。 - /// Config for Z.ai / 智谱 BigModel (国内端点). - pub fn zai_cn(api_key: impl Into, model: impl Into) -> Self { - Self::new(api_key, model, "open.bigmodel.cn") - } - - /// Z.ai 配置（国际端点）。 - /// Config for Z.ai (国际端点). - pub fn zai_intl(api_key: impl Into, model: impl Into) -> Self { - Self::new(api_key, model, "api.z.ai") - } -} - -/// 一个返回预置 HTTP 响应的 mock transport。用于在不触网的情况下对 -/// provider 做单元测试。 -/// A mock transport that returns a canned HTTP response. Used to unit-test -/// providers without touching the network. -#[derive(Debug, Clone)] -pub struct MockTransport { - /// 预置响应；每次请求取一份克隆（取空后再请求会报错）。 - /// Canned response; each request clones it (requests after it's drained error). - response: Arc>>, -} - -impl MockTransport { - /// 用给定状态码与 SSE 文本体构造一个预置响应。 - /// Build a canned response with the given status code and SSE body. - pub fn new(status: u16, body: impl Into) -> Self { - let body = body.into(); - let mut headers = HashMap::new(); - headers.insert("content-type".into(), "text/event-stream".into()); - headers.insert("content-length".into(), body.len().to_string()); - Self { - response: Arc::new(Mutex::new(Some(HttpResponse { - status, - headers, - body: body.into_bytes(), - }))), - } - } -} - -impl Transport for MockTransport { - fn request<'a>( - &'a self, - _req: &'a HttpRequest, - ) -> std::pin::Pin< - Box< - dyn std::future::Future> - + Send - + 'a, - >, - > { - Box::pin(async move { - self.response - .lock() - .unwrap() - .clone() - .ok_or_else(|| focus_transport::TransportError::Io("no canned response".into())) +/// 在后台线程上驱动一个异步闭包，把事件经同步 channel 送到主线程。 +/// Drive an async closure on a background thread, delivering events to the +/// main thread through a sync channel. +/// +/// `drive` 负责：发请求 → 读流 → 翻译成 [`StreamEvent`] 并 `tx.send`， +/// 并且必须**始终**以终止性事件（`Done`/`Error`）收尾，否则 agent 循环会 +/// 合成一条错误消息（见 `focus-core` 的 `stream_assistant_response`）。 +/// `drive` must always end with a terminal event (`Done`/`Error`); otherwise +/// the agent loop synthesizes an error message (see `focus-core`'s +/// `stream_assistant_response`). +pub(crate) fn run_stream( + drive: impl FnOnce(SyncSender) -> Pin + Send>> + + Send + + 'static, +) -> StreamResult { + let (tx, rx) = std::sync::mpsc::sync_channel(256); + std::thread::Builder::new() + .name("focus-provider-stream".into()) + .spawn(move || { + // 每个流使用独立的 current-thread 运行时，避免依赖调用方是否有 + // 活动的 tokio 运行时（agent 循环是同步的）。 + // Each stream gets its own current-thread runtime, so we don't + // depend on the caller having an active tokio runtime (the agent + // loop is synchronous). + let rt = match tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + { + Ok(rt) => rt, + Err(e) => { + let _ = tx.send(StreamEvent::Error { + error: error_message("unknown", &format!("failed to start runtime: {}", e)), + }); + return; + } + }; + rt.block_on(drive(tx)); }) + .map_err(|e| CoreError::Provider(format!("failed to spawn stream thread: {}", e)))?; + Ok(Box::new(ChannelIterator { rx })) +} + +/// 从同步 channel 拉取事件的迭代器。 +/// An iterator that pulls events from a sync channel. +struct ChannelIterator { + rx: Receiver, +} + +impl StreamIterator for ChannelIterator { + fn next_event(&mut self) -> Option { + // 所有发送者都断开（后台线程结束）时 recv 返回 Err → None（流结束）。 + // When all senders drop (the background thread finished), recv returns + // Err → None (end of stream). + self.rx.recv().ok() } } -/// 暂时不做任何剥离——这是为未来共享转换预留的挂载点。 -/// Strip and ignore nothing for now — a place to hang shared transformations. -pub fn _unused_keep_request_in_scope(_req: &ProviderRequest) {} +/// 构造一条错误 assistant 消息（stop_reason = Error）。 +/// Build an error assistant message (stop_reason = Error). +pub(crate) fn error_message(model: &str, msg: &str) -> AssistantMessage { + AssistantMessage { + content: Vec::new(), + model: model.to_string(), + usage: Usage::default(), + stop_reason: StopReason::Error, + error_message: Some(msg.to_string()), + timestamp: now_ms(), + } +} + +/// 向 channel 发送事件；消费者已离开时返回 `false`（调用方应停止驱动）。 +/// Send an event to the channel; returns `false` when the consumer is gone +/// (callers should stop driving). +pub(crate) fn send(tx: &SyncSender, event: StreamEvent) -> bool { + tx.send(event).is_ok() +} diff --git a/crates/focus-providers/src/config.rs b/crates/focus-providers/src/config.rs new file mode 100644 index 0000000..cb2f74d --- /dev/null +++ b/crates/focus-providers/src/config.rs @@ -0,0 +1,171 @@ +//! provider 共享的配置类型与辅助函数。 +//! Shared provider configuration types and helpers. +//! +//! 配置（api_key / base_url / 上下文窗口等）在运行时由上层（TUI）构造， +//! 因此全部是普通的可 Clone 结构体，而不是环境变量。 +//! Configuration (api_key / base_url / context window etc.) is constructed at +//! runtime by the upper layer (TUI), so these are plain Clone-able structs +//! rather than environment variables. + +use std::time::Duration; + +/// 一个 LLM 后端的连接配置。 +/// Connection configuration for one LLM backend. +#[derive(Debug, Clone)] +pub struct ProviderConfig { + /// API 密钥。 + /// The API key. + pub api_key: String, + /// API 基础地址；`None` 时使用该 provider 的默认端点。 + /// API base URL; `None` uses the provider's default endpoint. + /// + /// 约定：base_url 是「API 基址」（如 `https://api.anthropic.com`）， + /// 具体端点路径（如 `/v1/messages`）由 provider 自行拼接。 + /// Convention: base_url is an "API base" (e.g. `https://api.anthropic.com`); + /// the concrete endpoint path (e.g. `/v1/messages`) is appended by the + /// provider itself. + pub base_url: Option, + /// 模型上下文窗口大小（token 数）；`None` 时按已知模型表兜底。 + /// Model context window size (tokens); `None` falls back to the known-model + /// table. + pub context_window: Option, + /// 单次网络操作的超时时间。 + /// Timeout for individual network operations. + pub timeout: Duration, +} + +impl Default for ProviderConfig { + fn default() -> Self { + Self { + api_key: String::new(), + base_url: None, + context_window: None, + timeout: Duration::from_secs(60), + } + } +} + +impl ProviderConfig { + /// 便捷构造：只给 api_key，其余用默认值。 + /// Convenience constructor: only the api_key, everything else defaulted. + pub fn new(api_key: impl Into) -> Self { + Self { + api_key: api_key.into(), + ..Default::default() + } + } +} + +/// 上下文窗口未知时的保守兜底值（token 数）。 +/// Conservative fallback context window (tokens) when the model is unknown. +pub const DEFAULT_CONTEXT_WINDOW: u64 = 128_000; + +/// 已知模型的上下文窗口查询（大小写不敏感、前缀匹配）。 +/// Known-model context window lookup (case-insensitive, prefix-matched). +pub fn known_context_window(model: &str) -> Option { + let m = model.to_ascii_lowercase(); + // Anthropic 全线模型目前均为 200k 上下文。 + // All current Anthropic models use a 200k context window. + if m.starts_with("claude") { + return Some(200_000); + } + // 注意顺序：更具体的模式必须先于更宽泛的（如 "gpt-4.1" 在 "gpt-4" 之前）。 + // Order matters: more specific patterns must precede broader ones + // (e.g. "gpt-4.1" before "gpt-4"). + const WINDOWS: &[(&str, u64)] = &[ + ("gpt-5", 400_000), + ("gpt-4.1", 1_000_000), + ("gpt-4o", 128_000), + ("gpt-4-turbo", 128_000), + ("gpt-4", 8_192), + ("gpt-3.5-turbo", 16_385), + ("o4-mini", 200_000), + ("o3", 200_000), + ("o1", 200_000), + ]; + WINDOWS + .iter() + .find(|(pattern, _)| m.starts_with(pattern)) + .map(|(_, window)| *window) +} + +/// 按「配置值 → 已知模型表 → 保守兜底」的顺序解析上下文窗口。 +/// Resolve the context window in the order: configured value → known-model +/// table → conservative default. +pub fn resolve_context_window(configured: Option, model: &str) -> u64 { + configured + .or_else(|| known_context_window(model)) + .unwrap_or(DEFAULT_CONTEXT_WINDOW) +} + +/// 把 base_url 拆成 (host, port, path_prefix)。仅支持 `https://`。 +/// Split a base_url into (host, port, path_prefix). Only `https://` is +/// supported. +pub fn split_base_url(base: &str) -> Result<(String, u16, String), String> { + let rest = base + .strip_prefix("https://") + .ok_or_else(|| "only https:// base URLs are supported".to_string())?; + let (authority, path) = match rest.find('/') { + Some(i) => (&rest[..i], &rest[i..]), + None => (rest, ""), + }; + let (host, port) = match authority.rsplit_once(':') { + Some((h, p)) => { + let port: u16 = p + .parse() + .map_err(|_| format!("invalid port in base_url: {}", p))?; + (h.to_string(), port) + } + None => (authority.to_string(), 443), + }; + if host.is_empty() { + return Err("empty host in base_url".to_string()); + } + Ok((host, port, path.trim_end_matches('/').to_string())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn known_windows_prefix_matching() { + assert_eq!( + known_context_window("claude-sonnet-4-20250514"), + Some(200_000) + ); + assert_eq!(known_context_window("gpt-4o"), Some(128_000)); + assert_eq!(known_context_window("gpt-4.1-mini"), Some(1_000_000)); + assert_eq!(known_context_window("unknown-model"), None); + } + + #[test] + fn resolve_fallback_order() { + // 配置优先。 + // Configured value wins. + assert_eq!(resolve_context_window(Some(42), "claude-sonnet-4"), 42); + // 已知表兜底。 + // Known table as fallback. + assert_eq!(resolve_context_window(None, "claude-sonnet-4"), 200_000); + // 保守默认。 + // Conservative default. + assert_eq!( + resolve_context_window(None, "totally-unknown"), + DEFAULT_CONTEXT_WINDOW + ); + } + + #[test] + fn splits_base_urls() { + assert_eq!( + split_base_url("https://api.anthropic.com").unwrap(), + ("api.anthropic.com".to_string(), 443, String::new()) + ); + assert_eq!( + split_base_url("https://localhost:8080/v1").unwrap(), + ("localhost".to_string(), 8080, "/v1".to_string()) + ); + assert!(split_base_url("http://insecure.example.com").is_err()); + assert!(split_base_url("https:///no-host").is_err()); + } +} diff --git a/crates/focus-providers/src/lib.rs b/crates/focus-providers/src/lib.rs index e752378..8f17eb4 100644 --- a/crates/focus-providers/src/lib.rs +++ b/crates/focus-providers/src/lib.rs @@ -1,27 +1,30 @@ -//! focus-providers：具体的 LLM provider 实现。 +//! focus-providers：具体 LLM provider 实现。 //! focus-providers: concrete LLM provider implementations. //! -//! 每个 provider 都通过将厂商特定的 HTTP/SSE 格式翻译为统一的 -//! [`StreamEvent`] 协议，从而实现 [`focus_core::StreamProvider`]。 -//! Each provider implements [`focus_core::StreamProvider`] by translating its -//! vendor-specific HTTP/SSE format into the unified [`StreamEvent`] protocol. -//! -//! 主要目标是 [`ZaiProvider`]（智谱 GLM 编程模型，国内端点），它构建在共享的 -//! [`OpenAiCompatProvider`] 引擎之上。此外也提供 Anthropic 与原生 OpenAI provider。 -//! The primary target is [`ZaiProvider`] (智谱 GLM coding models, domestic -//! endpoint), built on the shared [`OpenAiCompatProvider`] engine. -//! Anthropic and raw-OpenAI providers are also available. +//! 目前包含 Anthropic（Messages API）与 OpenAI（Responses API + Chat +//! Completions 双协议）。所有 provider 都把 HTTP/SSE 翻译成 +//! [`focus_core::provider::StreamEvent`] 的统一流，错误一律编码为 +//! `StreamEvent::Error`，绝不通过返回 `Err` 表示 API 失败。 +//! Currently ships Anthropic (Messages API) and OpenAI (both the Responses +//! API and Chat Completions). Every provider translates HTTP/SSE into the +//! unified [`focus_core::provider::StreamEvent`] stream; failures are always +//! encoded as `StreamEvent::Error` rather than returned as `Err`. #![forbid(unsafe_code)] +/// Anthropic Messages API provider。 +/// The Anthropic Messages API provider. pub mod anthropic; +/// provider 共享的配置类型与辅助函数。 +/// Shared provider config types and helpers. pub mod common; -pub mod config_value; -pub mod openai_compat; -pub mod zai; +/// 配置类型与已知模型上下文窗口表。 +/// Config types and the known-model context-window table. +pub mod config; +/// OpenAI provider（Responses + Chat Completions）。 +/// The OpenAI provider (Responses + Chat Completions). +pub mod openai; pub use anthropic::AnthropicProvider; -pub use common::{MockTransport, ProviderConfig}; -pub use config_value::resolve as resolve_config_value; -pub use openai_compat::OpenAiCompatProvider; -pub use zai::{ZaiProvider, ZAI_CODING_PATH}; +pub use config::{resolve_context_window, ProviderConfig}; +pub use openai::{OpenAiProtocol, OpenAiProvider}; diff --git a/crates/focus-providers/src/openai.rs b/crates/focus-providers/src/openai.rs new file mode 100644 index 0000000..418176b --- /dev/null +++ b/crates/focus-providers/src/openai.rs @@ -0,0 +1,863 @@ +//! OpenAI provider：Responses API 与 Chat Completions 双协议（SSE 流式）。 +//! OpenAI provider: both the Responses API and Chat Completions (SSE streaming). +//! +//! 两套协议共享同一套核心类型翻译（messages / tools / usage / stop_reason）， +//! 仅请求格式与流式事件不同；通过 [`OpenAiProtocol`] 切换。 +//! Both protocols share the same core-type translation (messages / tools / +//! usage / stop_reason); only the request format and stream events differ, +//! selected via [`OpenAiProtocol`]. + +use crate::common; +use crate::config::{split_base_url, ProviderConfig}; +use focus_core::model::*; +use focus_core::provider::{ + ProviderEventReducer, ProviderRequest, StreamEvent, StreamProvider, StreamResult, +}; +use focus_core::CoreError; +use focus_json::JsonValue; +use focus_transport::{HttpRequest, SseEvent, SseParser, Transport}; +use std::collections::HashSet; +use std::fmt; +use std::sync::mpsc::SyncSender; +use std::sync::Arc; + +/// OpenAI 默认 API 基址。 +/// The default OpenAI API base. +const DEFAULT_BASE_URL: &str = "https://api.openai.com"; + +/// 两套 OpenAI 协议。 +/// The two OpenAI protocols. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum OpenAiProtocol { + /// 最新的 Responses API（`/v1/responses`）。 + /// The latest Responses API (`/v1/responses`). + #[default] + Responses, + /// 传统 Chat Completions（`/v1/chat/completions`），兼容绝大多数第三方端点。 + /// Legacy Chat Completions (`/v1/chat/completions`), compatible with most + /// third-party endpoints. + ChatCompletions, +} + +impl OpenAiProtocol { + fn endpoint_path(&self, prefix: &str) -> String { + match self { + OpenAiProtocol::Responses => format!("{}/v1/responses", prefix), + OpenAiProtocol::ChatCompletions => format!("{}/v1/chat/completions", prefix), + } + } +} + +/// 实现 OpenAI API 的 provider。 +/// A provider implementing the OpenAI API. +#[derive(Clone)] +pub struct OpenAiProvider { + config: ProviderConfig, + protocol: OpenAiProtocol, + transport: Arc, +} + +impl fmt::Debug for OpenAiProvider { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("OpenAiProvider") + .field("config", &self.config) + .field("protocol", &self.protocol) + .field("transport", &">") + .finish() + } +} + +impl OpenAiProvider { + /// 使用默认 HTTP 传输构造 provider。 + /// Build a provider with the default HTTP transport. + pub fn new(config: ProviderConfig, protocol: OpenAiProtocol) -> Self { + Self::with_transport( + config, + protocol, + Arc::new(focus_transport::HttpTransport::new()), + ) + } + + /// 注入自定义传输（测试用 mock）构造 provider。 + /// Build a provider with an injected transport (e.g. a mock in tests). + pub fn with_transport( + config: ProviderConfig, + protocol: OpenAiProtocol, + transport: Arc, + ) -> Self { + Self { + config, + protocol, + transport, + } + } + + /// 依据配置解析目标 (host, port, path)。 + /// Resolve the target (host, port, path) from the config. + fn endpoint(&self) -> Result<(String, u16, String), String> { + let base = self + .config + .base_url + .clone() + .unwrap_or_else(|| DEFAULT_BASE_URL.to_string()); + let (host, port, prefix) = split_base_url(&base)?; + Ok((host, port, self.protocol.endpoint_path(&prefix))) + } +} + +impl StreamProvider for OpenAiProvider { + fn stream(&self, request: &ProviderRequest) -> StreamResult { + let (host, port, path) = self.endpoint().map_err(CoreError::Provider)?; + let body = match self.protocol { + OpenAiProtocol::Responses => build_responses_body(request), + OpenAiProtocol::ChatCompletions => build_chat_body(request), + }; + let body = focus_json::to_string(&body); + let model = request.model.clone(); + let req = HttpRequest::post(&host, &path) + .port(port) + .header("Authorization", &format!("Bearer {}", self.config.api_key)) + .header("accept", "text/event-stream") + .json_body(&body); + let transport = self.transport.clone(); + let timeout = self.config.timeout; + // 先取出协议（Copy 类型），避免闭包借用 self。 + // Copy the protocol out first so the closure doesn't borrow self. + let protocol = self.protocol; + + common::run_stream(move |tx| { + Box::pin(async move { + let model = model.clone(); + let result: Result<(), String> = async { + let mut chunks = transport + .stream_request(&req, timeout) + .await + .map_err(|e| e.to_string())?; + let mut sse = SseParser::new(); + let mut turn = match protocol { + OpenAiProtocol::Responses => OpenAiTurn::responses(&model), + OpenAiProtocol::ChatCompletions => OpenAiTurn::chat(&model), + }; + loop { + let chunk = chunks.next_chunk().await.map_err(|e| e.to_string())?; + match chunk { + Some(bytes) => { + sse.feed(&bytes); + while let Some(ev) = sse.next_event() { + match turn.handle_sse(&ev, &tx) { + Ok(true) => {} + Ok(false) => return Ok(()), // 消费者已离开 / consumer gone + Err(msg) => { + let _ = tx.send(StreamEvent::Error { + error: common::error_message(&model, &msg), + }); + return Ok(()); + } + } + if turn.done { + return Ok(()); + } + } + } + None => break, + } + } + if !turn.done { + return Err("stream ended without a terminal event".to_string()); + } + Ok(()) + } + .await; + if let Err(msg) = result { + let _ = tx.send(StreamEvent::Error { + error: common::error_message(&model, &msg), + }); + } + }) + }) + } +} + +// ---- 请求体构造 ---- +// ---- request-body construction ------------------------------------------- + +/// Chat Completions 请求体。 +/// Chat Completions request body. +fn build_chat_body(request: &ProviderRequest) -> JsonValue { + let mut body = JsonValue::obj(); + body.insert("model", request.model.clone().into()).ok(); + let mut messages = JsonValue::arr(); + // 系统提示作为第一条 system 消息。 + // The system prompt becomes the first system message. + if !request.system_prompt.is_empty() { + let mut sys = JsonValue::obj(); + sys.insert("role", "system".into()).ok(); + sys.insert("content", request.system_prompt.clone().into()) + .ok(); + messages.push(sys).ok(); + } + for m in &request.messages { + messages.push(chat_message(m)).ok(); + } + body.insert("messages", messages).ok(); + if has_tools(&request.tools) { + body.insert("tools", tools_to_chat(&request.tools)).ok(); + body.insert("tool_choice", "auto".into()).ok(); + } + if let Some(mt) = request.max_tokens { + // 新模型使用 max_completion_tokens（旧字段 max_tokens 已被弃用）。 + // Newer models use max_completion_tokens (max_tokens is deprecated). + body.insert("max_completion_tokens", (mt as f64).into()) + .ok(); + } + if let Some(t) = request.temperature { + body.insert("temperature", t.into()).ok(); + } + body.insert("stream", true.into()).ok(); + // 请求结束 chunk 携带用量。 + // Ask for usage in the final chunk. + let mut so = JsonValue::obj(); + so.insert("include_usage", true.into()).ok(); + body.insert("stream_options", so).ok(); + body +} + +/// Responses API 请求体。 +/// Responses API request body. +fn build_responses_body(request: &ProviderRequest) -> JsonValue { + let mut body = JsonValue::obj(); + body.insert("model", request.model.clone().into()).ok(); + if !request.system_prompt.is_empty() { + body.insert("instructions", request.system_prompt.clone().into()) + .ok(); + } + body.insert("input", messages_to_responses(&request.messages)) + .ok(); + if has_tools(&request.tools) { + body.insert("tools", tools_to_responses(&request.tools)) + .ok(); + } + if let Some(mt) = request.max_tokens { + body.insert("max_output_tokens", (mt as f64).into()).ok(); + } + if let Some(t) = request.temperature { + body.insert("temperature", t.into()).ok(); + } + body.insert("stream", true.into()).ok(); + body +} + +/// 一条核心消息 → Chat Completions 消息。 +/// One core message → a Chat Completions message. +fn chat_message(m: &Message) -> JsonValue { + match m { + Message::User(u) => { + let mut o = JsonValue::obj(); + o.insert("role", "user".into()).ok(); + let text = join_text(&u.content); + let images: Vec<&ImageContent> = u + .content + .iter() + .filter_map(|c| match c { + ContentBlock::Image(i) => Some(i), + _ => None, + }) + .collect(); + if images.is_empty() { + o.insert("content", text.into()).ok(); + } else { + let mut parts = JsonValue::arr(); + if !text.is_empty() { + let mut p = JsonValue::obj(); + p.insert("type", "text".into()).ok(); + p.insert("text", text.clone().into()).ok(); + parts.push(p).ok(); + } + for img in images { + let mut p = JsonValue::obj(); + p.insert("type", "image_url".into()).ok(); + let mut url = JsonValue::obj(); + url.insert( + "url", + format!("data:{};base64,{}", img.mime_type, img.data).into(), + ) + .ok(); + p.insert("image_url", url).ok(); + parts.push(p).ok(); + } + o.insert("content", parts).ok(); + } + o + } + Message::Assistant(a) => { + let mut o = JsonValue::obj(); + o.insert("role", "assistant".into()).ok(); + // 思考块不回传（OpenAI 不接受 reasoning 内容）。 + // Thinking blocks are not replayed (OpenAI rejects reasoning content). + o.insert("content", join_text(&a.content).into()).ok(); + let calls: Vec<&ToolCall> = a.content.iter().filter_map(|c| c.as_tool_call()).collect(); + if !calls.is_empty() { + let mut arr = JsonValue::arr(); + for tc in calls { + let mut c = JsonValue::obj(); + c.insert("id", tc.id.clone().into()).ok(); + c.insert("type", "function".into()).ok(); + let mut f = JsonValue::obj(); + f.insert("name", tc.name.clone().into()).ok(); + f.insert("arguments", focus_json::to_string(&tc.arguments).into()) + .ok(); + c.insert("function", f).ok(); + arr.push(c).ok(); + } + o.insert("tool_calls", arr).ok(); + } + o + } + Message::ToolResult(t) => { + let mut o = JsonValue::obj(); + o.insert("role", "tool".into()).ok(); + o.insert("tool_call_id", t.tool_call_id.clone().into()).ok(); + o.insert("content", join_text(&t.content).into()).ok(); + o + } + } +} + +/// 核心消息列表 → Responses API input items。 +/// Core messages → Responses API input items. +fn messages_to_responses(messages: &[Message]) -> JsonValue { + let mut items = JsonValue::arr(); + for m in messages { + match m { + Message::User(u) => { + let mut item = JsonValue::obj(); + item.insert("role", "user".into()).ok(); + let mut content = JsonValue::arr(); + let text = join_text(&u.content); + if !text.is_empty() { + let mut part = JsonValue::obj(); + part.insert("type", "input_text".into()).ok(); + part.insert("text", text.into()).ok(); + content.push(part).ok(); + } + item.insert("content", content).ok(); + items.push(item).ok(); + } + Message::Assistant(a) => { + let mut item = JsonValue::obj(); + item.insert("role", "assistant".into()).ok(); + let mut content = JsonValue::arr(); + let text = join_text(&a.content); + if !text.is_empty() { + let mut part = JsonValue::obj(); + part.insert("type", "output_text".into()).ok(); + part.insert("text", text.into()).ok(); + content.push(part).ok(); + } + item.insert("content", content).ok(); + items.push(item).ok(); + // 工具调用作为独立的 function_call items（function_call_output + // 的 call_id 必须对应它们）。 + // Tool calls become standalone function_call items (the + // function_call_output call_ids must match them). + for tc in a.content.iter().filter_map(|c| c.as_tool_call()) { + let mut fc = JsonValue::obj(); + fc.insert("type", "function_call".into()).ok(); + fc.insert("call_id", tc.id.clone().into()).ok(); + fc.insert("name", tc.name.clone().into()).ok(); + fc.insert("arguments", focus_json::to_string(&tc.arguments).into()) + .ok(); + items.push(fc).ok(); + } + } + Message::ToolResult(t) => { + let mut item = JsonValue::obj(); + item.insert("type", "function_call_output".into()).ok(); + item.insert("call_id", t.tool_call_id.clone().into()).ok(); + item.insert("output", join_text(&t.content).into()).ok(); + items.push(item).ok(); + } + } + } + items +} + +/// 核心工具定义 → Chat Completions tools（function 格式）。 +/// Core tool definitions → Chat Completions tools (function format). +fn tools_to_chat(tools: &JsonValue) -> JsonValue { + match tools { + JsonValue::Arr(items) => { + let mut arr = JsonValue::arr(); + for t in items { + let mut o = JsonValue::obj(); + o.insert("type", "function".into()).ok(); + let mut f = JsonValue::obj(); + if let Some(n) = t.get_str("name") { + f.insert("name", n.into()).ok(); + } + if let Some(d) = t.get_str("description") { + f.insert("description", d.into()).ok(); + } + if let Some(p) = t.get("parameters") { + f.insert("parameters", p.clone()).ok(); + } + o.insert("function", f).ok(); + arr.push(o).ok(); + } + arr + } + _ => JsonValue::arr(), + } +} + +/// 核心工具定义 → Responses API tools。 +/// Core tool definitions → Responses API tools. +fn tools_to_responses(tools: &JsonValue) -> JsonValue { + match tools { + JsonValue::Arr(items) => { + let mut arr = JsonValue::arr(); + for t in items { + let mut o = JsonValue::obj(); + o.insert("type", "function".into()).ok(); + if let Some(n) = t.get_str("name") { + o.insert("name", n.into()).ok(); + } + if let Some(d) = t.get_str("description") { + o.insert("description", d.into()).ok(); + } + if let Some(p) = t.get("parameters") { + o.insert("parameters", p.clone()).ok(); + } + arr.push(o).ok(); + } + arr + } + _ => JsonValue::arr(), + } +} + +fn has_tools(tools: &JsonValue) -> bool { + matches!(tools, JsonValue::Arr(items) if !items.is_empty()) +} + +/// 拼接文本块；思考块不计入（OpenAI 无法消费 reasoning 内容）。 +/// Join text blocks; thinking is excluded (OpenAI cannot consume reasoning). +fn join_text(blocks: &[ContentBlock]) -> String { + blocks + .iter() + .filter_map(|c| match c { + ContentBlock::Text(t) => Some(t.text.clone()), + _ => None, + }) + .collect::>() + .join("\n") +} + +/// 把 Chat Completions 的 finish_reason 映射到核心类型。 +/// Map a Chat Completions finish_reason onto the core type. +fn map_chat_finish(s: &str) -> StopReason { + match s { + "length" => StopReason::Length, + "tool_calls" | "function_call" => StopReason::ToolUse, + _ => StopReason::Stop, // stop / content_filter + } +} + +/// 两种协议共享的回合累积状态（基于 [`ProviderEventReducer`]）。 +/// Per-turn accumulator shared by both protocols (built on +/// [`ProviderEventReducer`]). +struct OpenAiTurn { + reducer: ProviderEventReducer, + model: String, + protocol: OpenAiProtocol, + done: bool, + /// 已打开的工具调用（provider 局部索引），用于去重。 + /// Tool calls already opened (provider-local index), for dedup. + seen_calls: HashSet, + /// 输入 / 输出 token（最后一块 usage 提供）。 + /// Input / output tokens (from the final usage chunk). + input_tokens: u64, + output_tokens: u64, + cached_tokens: u64, +} + +impl OpenAiTurn { + fn responses(model: &str) -> Self { + Self::new(model, OpenAiProtocol::Responses) + } + + fn chat(model: &str) -> Self { + Self::new(model, OpenAiProtocol::ChatCompletions) + } + + fn new(model: &str, protocol: OpenAiProtocol) -> Self { + Self { + reducer: ProviderEventReducer::new(model), + model: model.to_string(), + protocol, + done: false, + seen_calls: HashSet::new(), + input_tokens: 0, + output_tokens: 0, + cached_tokens: 0, + } + } + + /// 收尾：关闭所有未闭合块并发送 Done。 + /// Finalize: close all open blocks and send Done. + fn finalize(&mut self, tx: &SyncSender) -> Result { + for ev in self.reducer.finalize_events() { + if !common::send(tx, ev) { + return Ok(false); + } + } + let mut message = self + .reducer + .finish() + .ok_or_else(|| "reducer produced no final message".to_string())?; + message.usage = Usage { + input_tokens: self.input_tokens, + output_tokens: self.output_tokens, + cache_read_tokens: self.cached_tokens, + cache_write_tokens: 0, + }; + self.done = true; + Ok(common::send(tx, StreamEvent::Done { message })) + } + + /// 处理一个 SSE 事件。 + /// Handle one SSE event. + fn handle_sse(&mut self, ev: &SseEvent, tx: &SyncSender) -> Result { + let data = ev.data.trim(); + // Chat Completions 的流结束标记不是合法 JSON，必须特判。 + // Chat Completions' end-of-stream marker is not valid JSON; special-case it. + if data == "[DONE]" { + return self.finalize(tx); + } + let json: JsonValue = if data.is_empty() { + JsonValue::obj() + } else { + focus_json::parse(data).map_err(|e| format!("invalid SSE JSON: {}", e))? + }; + match self.protocol { + OpenAiProtocol::ChatCompletions => self.handle_chat_event(&json, tx), + OpenAiProtocol::Responses => self.handle_responses_event(ev, &json, tx), + } + } + + fn handle_chat_event( + &mut self, + json: &JsonValue, + tx: &SyncSender, + ) -> Result { + // API 错误以 `{"error": {...}}` chunk 出现。 + // API errors arrive as a `{"error": {...}}` chunk. + if let Some(err) = json.get("error") { + let ty = err.get_str("type").unwrap_or("api_error"); + let message = err.get_str("message").unwrap_or("unknown error"); + self.done = true; + return Ok(common::send( + tx, + StreamEvent::Error { + error: common::error_message(&self.model, &format!("{}: {}", ty, message)), + }, + )); + } + if let Some(choices) = json.get_arr("choices") { + for choice in choices { + if let Some(delta) = choice.get("delta") { + if let Some(content) = delta.get_str("content") { + if !content.is_empty() { + for ev in self.reducer.text_delta(content) { + if !common::send(tx, ev) { + return Ok(false); + } + } + } + } + // 推理内容（如 DeepSeek reasoner）→ 思考块。 + // Reasoning content (e.g. DeepSeek reasoner) → thinking block. + if let Some(reasoning) = delta.get_str("reasoning_content") { + if !reasoning.is_empty() { + for ev in self.reducer.thinking_delta(reasoning) { + if !common::send(tx, ev) { + return Ok(false); + } + } + } + } + if let Some(tcs) = delta.get_arr("tool_calls") { + for tc in tcs { + let idx = tc.get_num("index").unwrap_or(0.0) as usize; + if let Some(f) = tc.get("function") { + if let Some(name) = f.get_str("name") { + if !name.is_empty() && self.seen_calls.insert(idx) { + for ev in self.reducer.tool_call_start(idx, name) { + if !common::send(tx, ev) { + return Ok(false); + } + } + } + } + if let Some(args) = f.get_str("arguments") { + if !args.is_empty() { + for ev in self.reducer.tool_call_delta(idx, args) { + if !common::send(tx, ev) { + return Ok(false); + } + } + } + } + } + } + } + } + if let Some(fr) = choice.get_str("finish_reason") { + if !fr.is_empty() { + self.reducer.set_stop(map_chat_finish(fr)); + } + } + } + } + if let Some(usage) = json.get("usage") { + self.input_tokens = usage.get_num("prompt_tokens").unwrap_or(0.0) as u64; + self.output_tokens = usage.get_num("completion_tokens").unwrap_or(0.0) as u64; + if let Some(details) = usage.get("prompt_tokens_details") { + self.cached_tokens = details.get_num("cached_tokens").unwrap_or(0.0) as u64; + } + } + Ok(true) + } + + fn handle_responses_event( + &mut self, + ev: &SseEvent, + json: &JsonValue, + tx: &SyncSender, + ) -> Result { + match ev.event.as_str() { + "response.output_item.added" => { + if let Some(item) = json.get("item") { + match item.get_str("type").unwrap_or("") { + "message" => { + // 打开一个空的文本块，随后的 output_text.delta 填充。 + // Open an empty text block; output_text.delta fills it. + for e in self.reducer.text_delta("") { + if !common::send(tx, e) { + return Ok(false); + } + } + } + "function_call" => { + let idx = json.get_num("output_index").unwrap_or(0.0) as usize; + let name = item.get_str("name").unwrap_or("").to_string(); + if !name.is_empty() { + for e in self.reducer.tool_call_start(idx, &name) { + if !common::send(tx, e) { + return Ok(false); + } + } + } + } + _ => {} + } + } + Ok(true) + } + "response.output_text.delta" => { + if let Some(delta) = json.get_str("delta") { + if !delta.is_empty() { + for e in self.reducer.text_delta(delta) { + if !common::send(tx, e) { + return Ok(false); + } + } + } + } + Ok(true) + } + "response.function_call_arguments.delta" => { + if let Some(delta) = json.get_str("delta") { + if !delta.is_empty() { + let idx = json.get_num("output_index").unwrap_or(0.0) as usize; + for e in self.reducer.tool_call_delta(idx, delta) { + if !common::send(tx, e) { + return Ok(false); + } + } + } + } + Ok(true) + } + "response.completed" => { + if let Some(resp) = json.get("response") { + match resp.get_str("status").unwrap_or("completed") { + "completed" => self.reducer.set_stop(StopReason::Stop), + "incomplete" => { + let reason = resp + .get("incomplete_details") + .and_then(|d| d.get_str("reason")) + .unwrap_or("unknown"); + self.reducer.set_stop(if reason == "max_output_tokens" { + StopReason::Length + } else { + StopReason::Stop + }); + } + _ => {} + } + if let Some(usage) = resp.get("usage") { + self.input_tokens = usage.get_num("input_tokens").unwrap_or(0.0) as u64; + self.output_tokens = usage.get_num("output_tokens").unwrap_or(0.0) as u64; + if let Some(details) = usage.get("input_tokens_details") { + self.cached_tokens = + details.get_num("cached_tokens").unwrap_or(0.0) as u64; + } + } + } + self.finalize(tx) + } + "response.failed" | "error" => { + let code = json.get_str("code").unwrap_or("response_failed"); + let message = json.get_str("message").unwrap_or("unknown error"); + self.done = true; + Ok(common::send( + tx, + StreamEvent::Error { + error: common::error_message( + &self.model, + &format!("{}: {}", code, message), + ), + }, + )) + } + _ => Ok(true), // created / in_progress / output_item.done 等忽略 + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use focus_core::tool::{ToolEffects, ToolRegistry}; + + #[derive(Debug)] + struct EchoTool; + impl focus_core::Tool for EchoTool { + fn name(&self) -> &str { + "echo" + } + fn description(&self) -> &str { + "echoes text" + } + fn parameters(&self) -> JsonValue { + let mut o = JsonValue::obj(); + o.insert("type", "object".into()).ok(); + let mut props = JsonValue::obj(); + props.insert("text", JsonValue::Str("the text".into())).ok(); + o.insert("properties", props).ok(); + o + } + fn effects(&self) -> ToolEffects { + ToolEffects::READ + } + fn execute( + &self, + _id: &str, + args: &JsonValue, + _u: Option<&focus_core::tool::ToolUpdateSink>, + ) -> Result { + Ok(focus_core::tool::ToolResult::text( + args.get_str("text").unwrap_or("").to_string(), + )) + } + } + + fn sample_request() -> ProviderRequest { + ProviderRequest { + model: "gpt-4o".into(), + system_prompt: "Be concise.".into(), + messages: vec![ + Message::user_text("hi"), + Message::Assistant(AssistantMessage { + content: vec![ContentBlock::ToolCall(ToolCall { + id: "call_abc".into(), + name: "echo".into(), + arguments: focus_json::parse(r#"{"text":"x"}"#).unwrap(), + })], + model: "gpt-4o".into(), + usage: Usage::default(), + stop_reason: StopReason::ToolUse, + error_message: None, + timestamp: 0, + }), + Message::ToolResult(ToolResultMessage { + tool_call_id: "call_abc".into(), + tool_name: "echo".into(), + content: vec![ContentBlock::text("ok")], + details: JsonValue::obj(), + is_error: false, + timestamp: 0, + }), + ], + tools: ToolRegistry::with(Box::new(EchoTool)).tool_definitions(), + max_tokens: Some(512), + temperature: Some(0.2), + } + } + + #[test] + fn chat_body_translation() { + let body = build_chat_body(&sample_request()); + let messages = body.get_arr("messages").unwrap(); + assert_eq!(messages[0].get_str("role"), Some("system")); + assert_eq!(messages[0].get_str("content"), Some("Be concise.")); + // assistant 携带 tool_calls（arguments 为 JSON 字符串）。 + // Assistant carries tool_calls (arguments as a JSON string). + let assistant = &messages[2]; + let calls = assistant.get_arr("tool_calls").unwrap(); + assert_eq!(calls[0].get_str("id"), Some("call_abc")); + let func = calls[0].get("function").unwrap(); + assert_eq!(func.get_str("name"), Some("echo")); + assert_eq!(func.get_str("arguments"), Some(r#"{"text":"x"}"#)); + // tool 消息带 tool_call_id。 + // Tool message carries tool_call_id. + assert_eq!(messages[3].get_str("role"), Some("tool")); + assert_eq!(messages[3].get_str("tool_call_id"), Some("call_abc")); + // tools 被翻译成 function 格式。 + // Tools translated into the function format. + let tools = body.get_arr("tools").unwrap(); + assert_eq!(tools[0].get_str("type"), Some("function")); + let f = tools[0].get("function").unwrap(); + assert_eq!(f.get_str("name"), Some("echo")); + assert!(f.get("parameters").is_some()); + assert_eq!(body.get_bool("stream"), Some(true)); + assert_eq!(body.get_num("max_completion_tokens"), Some(512.0)); + } + + #[test] + fn responses_body_translation() { + let body = build_responses_body(&sample_request()); + assert_eq!(body.get_str("instructions"), Some("Be concise.")); + let input = body.get_arr("input").unwrap(); + // 顺序：user → assistant(role) → function_call → function_call_output。 + // Order: user → assistant(role) → function_call → function_call_output. + assert_eq!(input[0].get_str("role"), Some("user")); + assert_eq!(input[1].get_str("role"), Some("assistant")); + assert_eq!(input[2].get_str("type"), Some("function_call")); + assert_eq!(input[2].get_str("call_id"), Some("call_abc")); + assert_eq!(input[3].get_str("type"), Some("function_call_output")); + assert_eq!(input[3].get_str("call_id"), Some("call_abc")); + let tools = body.get_arr("tools").unwrap(); + assert_eq!(tools[0].get_str("name"), Some("echo")); + assert_eq!(body.get_num("max_output_tokens"), Some(512.0)); + } + + #[test] + fn maps_chat_finish_reasons() { + assert_eq!(map_chat_finish("stop"), StopReason::Stop); + assert_eq!(map_chat_finish("length"), StopReason::Length); + assert_eq!(map_chat_finish("tool_calls"), StopReason::ToolUse); + assert_eq!(map_chat_finish("content_filter"), StopReason::Stop); + } +} diff --git a/crates/focus-providers/tests/agent_integration.rs b/crates/focus-providers/tests/agent_integration.rs new file mode 100644 index 0000000..6a20599 --- /dev/null +++ b/crates/focus-providers/tests/agent_integration.rs @@ -0,0 +1,140 @@ +//! 端到端集成：Agent + AnthropicProvider + mock transport 跑完整工具回合。 +//! End-to-end integration: Agent + AnthropicProvider + mock transport running +//! a full tool-call turn. + +mod common; + +use focus_core::event::VecSink; +use focus_core::model::*; +use focus_core::tool::{Tool, ToolEffects, ToolRegistry, ToolResult, ToolUpdateSink}; +use focus_core::{Agent, AgentConfig, CoreError}; +use focus_json::JsonValue; +use focus_providers::anthropic::AnthropicProvider; +use focus_providers::config::ProviderConfig; +use std::sync::Arc; + +/// 记录调用参数的 echo 工具。 +/// An echo tool that records its arguments. +#[derive(Debug, Default)] +struct EchoTool { + calls: Arc>>, +} + +impl Clone for EchoTool { + fn clone(&self) -> Self { + Self { + calls: self.calls.clone(), + } + } +} + +impl Tool for EchoTool { + fn name(&self) -> &str { + "echo" + } + fn description(&self) -> &str { + "echoes the arguments back" + } + fn parameters(&self) -> JsonValue { + let mut o = JsonValue::obj(); + o.insert("type", "object".into()).ok(); + let mut props = JsonValue::obj(); + props.insert("text", JsonValue::Str("the text".into())).ok(); + o.insert("properties", props).ok(); + o + } + fn effects(&self) -> ToolEffects { + ToolEffects::READ + } + fn execute( + &self, + _id: &str, + args: &JsonValue, + _u: Option<&ToolUpdateSink>, + ) -> Result { + self.calls.lock().unwrap().push(args.clone()); + Ok(ToolResult::text(format!( + "echo: {}", + args.get_str("text").unwrap_or("") + ))) + } +} + +/// 回合 1：模型调用 echo 工具；回合 2：模型给出最终文本。 +/// Turn 1: the model calls the echo tool; turn 2: the model gives final text. +#[test] +fn agent_runs_full_tool_turn_against_anthropic() { + let mut mock = common::MockTransport::new(); + // 回合 1：tool_use → 回合 2：文本回答。 + // Turn 1: tool_use → turn 2: text answer. + mock.push_body( + "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":20}}}\n\n\ + event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_1\",\"name\":\"echo\",\"input\":{}}}\n\n\ + event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"text\\\":\\\"world\\\"}\"}}\n\n\ + event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n\ + event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":8}}\n\n\ + event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n", + ); + mock.push_body( + "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":30}}}\n\n\ + event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n\ + event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"all done\"}}\n\n\ + event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n\ + event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":4}}\n\n\ + event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n", + ); + + let provider = + AnthropicProvider::with_transport(ProviderConfig::new("sk-test"), Arc::new(mock.clone())); + let config = AgentConfig { + model: "claude-sonnet-4".into(), + system_prompt: "You are helpful.".into(), + max_tokens: Some(1024), + ..Default::default() + }; + let echo = EchoTool::default(); + let calls = echo.calls.clone(); + let tools = ToolRegistry::with(Box::new(echo)); + let mut agent = Agent::new(config, Box::new(provider), tools); + let mut sink = VecSink::new(); + + agent + .prompt("please echo world", &mut sink) + .expect("run failed"); + + // 两次回合、一次工具执行。 + // Two turns, one tool execution. + assert_eq!(sink.count("turn_end"), 2); + assert_eq!(sink.count("tool_execution_end"), 1); + + let msgs = agent.messages(); + assert_eq!(msgs.len(), 4); + assert!(matches!(msgs[0], Message::User(_))); + assert!(matches!( + &msgs[1], + Message::Assistant(a) if a.content.iter().any(|c| c.as_tool_call().is_some()) + )); + assert!(matches!( + &msgs[2], + Message::ToolResult(tr) if tr.tool_name == "echo" && !tr.is_error + )); + let final_text = match &msgs[3] { + Message::Assistant(a) => a + .content + .iter() + .find_map(|c| c.as_text()) + .map(|t| t.text.clone()), + _ => None, + }; + assert_eq!(final_text.as_deref(), Some("all done")); + + // 工具确实收到了模型传入的参数。 + // The tool actually received the model's arguments. + let recorded = calls.lock().unwrap(); + assert_eq!(recorded.len(), 1); + assert_eq!(recorded[0].get_str("text"), Some("world")); + + // 两个回合都发出了 HTTP 请求。 + // Both turns issued HTTP requests. + assert_eq!(mock.request_count(), 2); +} diff --git a/crates/focus-providers/tests/anthropic_tests.rs b/crates/focus-providers/tests/anthropic_tests.rs new file mode 100644 index 0000000..26c1061 --- /dev/null +++ b/crates/focus-providers/tests/anthropic_tests.rs @@ -0,0 +1,237 @@ +//! Anthropic provider 的录制/回放集成测试。 +//! Recorded/replayed integration tests for the Anthropic provider. + +mod common; + +use focus_core::model::*; +use focus_core::provider::{ProviderRequest, StreamEvent}; +use focus_providers::anthropic::AnthropicProvider; +use focus_providers::config::ProviderConfig; +use focus_transport::TransportError; +use std::sync::Arc; + +fn provider(mock: common::MockTransport) -> AnthropicProvider { + AnthropicProvider::with_transport(ProviderConfig::new("sk-test"), Arc::new(mock)) +} + +fn request() -> ProviderRequest { + ProviderRequest { + model: "claude-sonnet-4".into(), + system_prompt: "You are helpful.".into(), + messages: vec![Message::user_text("hello")], + tools: focus_json::JsonValue::arr(), + max_tokens: Some(1024), + temperature: None, + } +} + +/// 事件序列化简：只保留 (标签, 文本/名称) 供断言。 +/// Event-sequence digest: keep only (label, text/name) for assertions. +fn digest(events: &[StreamEvent]) -> Vec<(&'static str, String)> { + events + .iter() + .map(|e| match e { + StreamEvent::Start { .. } => ("start", String::new()), + StreamEvent::TextStart { .. } => ("text_start", String::new()), + StreamEvent::TextDelta { delta, .. } => ("text_delta", delta.clone()), + StreamEvent::TextEnd { .. } => ("text_end", String::new()), + StreamEvent::ThinkingStart { .. } => ("thinking_start", String::new()), + StreamEvent::ThinkingDelta { delta, .. } => ("thinking_delta", delta.clone()), + StreamEvent::ThinkingEnd { .. } => ("thinking_end", String::new()), + StreamEvent::ToolCallStart { .. } => ("tool_start", String::new()), + StreamEvent::ToolCallDelta { delta, .. } => ("tool_delta", delta.clone()), + StreamEvent::ToolCallEnd { tool_call, .. } => ("tool_end", tool_call.name.clone()), + StreamEvent::Done { message } => ("done", message.stop_reason.as_str().into()), + StreamEvent::Error { error } => { + ("error", error.error_message.clone().unwrap_or_default()) + } + }) + .collect() +} + +#[test] +fn streams_text_response() { + let mut mock = common::MockTransport::new(); + mock.push_body( + "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4\",\"stop_reason\":null,\"usage\":{\"input_tokens\":10,\"cache_creation_input_tokens\":4,\"cache_read_input_tokens\":2}}}\n\n\ + event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n\ + event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"Hello\"}}\n\n\ + event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\" world\"}}\n\n\ + event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n\ + event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",\"stop_sequence\":null},\"usage\":{\"output_tokens\":6}}\n\n\ + event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n", + ); + let provider = provider(mock); + let events = common::collect(&provider, &request()); + + let d = digest(&events); + assert_eq!(d[0].0, "start"); + assert_eq!(d[1].0, "text_start"); + assert_eq!(d[2], ("text_delta", "Hello".into())); + assert_eq!(d[3], ("text_delta", " world".into())); + assert_eq!(d[4].0, "text_end"); + assert_eq!(d[5].0, "done"); + assert_eq!(d[5].1, "stop"); + + // 最终消息内容与用量。 + // Final message content and usage. + let done = events + .iter() + .find_map(|e| match e { + StreamEvent::Done { message } => Some(message), + _ => None, + }) + .expect("done event"); + let text = done + .content + .iter() + .find_map(|c| c.as_text()) + .expect("text block"); + assert_eq!(text.text, "Hello world"); + assert_eq!(done.usage.input_tokens, 10); + assert_eq!(done.usage.output_tokens, 6); + assert_eq!(done.usage.cache_write_tokens, 4); + assert_eq!(done.usage.cache_read_tokens, 2); + assert_eq!(done.stop_reason, StopReason::Stop); +} + +#[test] +fn streams_thinking_then_text() { + let mut mock = common::MockTransport::new(); + mock.push_body( + "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":5}}}\n\n\ + event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"thinking\",\"thinking\":\"\",\"signature\":\"sig-1\"}}\n\n\ + event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"thinking_delta\",\"thinking\":\"Let me think\"}}\n\n\ + event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n\ + event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":1,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n\ + event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"text_delta\",\"text\":\"Answer: 42\"}}\n\n\ + event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":1}\n\n\ + event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n", + ); + let provider = provider(mock); + let events = common::collect(&provider, &request()); + + let d = digest(&events); + assert!(d.contains(&("thinking_start", String::new()))); + assert!(d.contains(&("thinking_delta", "Let me think".into()))); + assert!(d.contains(&("thinking_end", String::new()))); + + let done = events + .iter() + .find_map(|e| match e { + StreamEvent::Done { message } => Some(message), + _ => None, + }) + .unwrap(); + assert_eq!(done.content.len(), 2); + match &done.content[0] { + ContentBlock::Thinking(t) => { + assert_eq!(t.thinking, "Let me think"); + assert_eq!(t.signature.as_deref(), Some("sig-1")); + } + other => panic!("expected thinking block, got {:?}", other), + } + let text = done.content[1].as_text().expect("text block"); + assert_eq!(text.text, "Answer: 42"); +} + +#[test] +fn streams_tool_use_with_partial_json() { + let mut mock = common::MockTransport::new(); + mock.push_body( + "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":7}}}\n\n\ + event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_1\",\"name\":\"read\",\"input\":{}}}\n\n\ + event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"path\\\":\\\"a\"}}\n\n\ + event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"bc.txt\\\"}\"}}\n\n\ + event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n\ + event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":9}}\n\n\ + event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n", + ); + let provider = provider(mock); + let events = common::collect(&provider, &request()); + + let done = events + .iter() + .find_map(|e| match e { + StreamEvent::Done { message } => Some(message), + _ => None, + }) + .unwrap(); + assert_eq!(done.stop_reason, StopReason::ToolUse); + let tc = done.content[0].as_tool_call().expect("tool call"); + assert_eq!(tc.id, "toolu_1"); + assert_eq!(tc.name, "read"); + assert_eq!(tc.arguments.get_str("path"), Some("abc.txt")); + + // ToolCallEnd 携带最终的 tool call。 + // ToolCallEnd carries the final tool call. + let end = events + .iter() + .find_map(|e| match e { + StreamEvent::ToolCallEnd { tool_call, .. } => Some(tool_call), + _ => None, + }) + .unwrap(); + assert_eq!(end.arguments.get_str("path"), Some("abc.txt")); +} + +#[test] +fn encodes_api_error_event() { + let mut mock = common::MockTransport::new(); + mock.push_body( + "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"overloaded\"}}\n\n", + ); + let provider = provider(mock); + let events = common::collect(&provider, &request()); + + match events.last().unwrap() { + StreamEvent::Error { error } => { + assert_eq!(error.stop_reason, StopReason::Error); + let msg = error.error_message.clone().unwrap_or_default(); + assert!(msg.contains("overloaded"), "got: {}", msg); + } + other => panic!("expected error event, got {:?}", other), + } +} + +#[test] +fn encodes_http_error() { + let mut mock = common::MockTransport::new(); + mock.push_error(TransportError::Http { + status: 429, + body: "rate limited".into(), + }); + let provider = provider(mock); + let events = common::collect(&provider, &request()); + + match events.last().unwrap() { + StreamEvent::Error { error } => { + let msg = error.error_message.clone().unwrap_or_default(); + assert!(msg.contains("429"), "got: {}", msg); + } + other => panic!("expected error event, got {:?}", other), + } +} + +#[test] +fn request_is_well_formed() { + let mut mock = common::MockTransport::new(); + mock.push_body("event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n"); + let provider = provider(mock.clone()); + let _ = common::collect(&provider, &request()); + + let req = mock.last_request(); + assert_eq!(req.host, "api.anthropic.com"); + assert_eq!(req.path, "/v1/messages"); + let headers: Vec<(String, String)> = req.headers.clone(); + assert!(headers + .iter() + .any(|(k, v)| k == "x-api-key" && v == "sk-test")); + assert!(headers + .iter() + .any(|(k, v)| k == "anthropic-version" && v == "2023-06-01")); + let body: focus_json::JsonValue = + focus_json::parse(&String::from_utf8_lossy(&req.body)).unwrap(); + assert_eq!(body.get_str("model"), Some("claude-sonnet-4")); + assert_eq!(body.get_bool("stream"), Some(true)); +} diff --git a/crates/focus-providers/tests/common/mod.rs b/crates/focus-providers/tests/common/mod.rs new file mode 100644 index 0000000..1523ad8 --- /dev/null +++ b/crates/focus-providers/tests/common/mod.rs @@ -0,0 +1,170 @@ +//! 测试共享的 mock transport：录制请求并回放录制的 SSE 流。 +//! Shared test mock transport: records requests and replays recorded SSE +//! streams. +//! +//! 回放时把流按固定小尺寸切块，刻意让 SSE 事件跨 chunk 边界， +//! 以验证 provider 的增量解析。 +//! Replays are split into small fixed-size chunks to deliberately straddle +//! SSE event boundaries, exercising the providers' incremental parsing. + +use focus_transport::{ChunkStream, HttpRequest, HttpResponse, Transport, TransportError}; +use std::collections::{HashMap, VecDeque}; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +/// 录制请求并回放响应体的 mock transport。 +/// A mock transport that records requests and replays response bodies. +#[derive(Debug, Clone, Default)] +pub struct MockTransport { + /// 待回放的响应体（每个消费一个）；流式接口按小块回放。 + /// Bodies to replay (one consumed per request); the streaming interface + /// replays them in small chunks. + streams: Arc>>>, + /// 待返回的传输层错误（每个消费一个，优先于 streams）。 + /// Transport errors to return (one consumed per request; take precedence). + errors: Arc>>, + /// 已发出的请求记录。 + /// Requests sent so far. + requests: Arc>>, +} + +impl MockTransport { + /// 创建一个空的 mock transport。 + /// Create an empty mock transport. + pub fn new() -> Self { + Self::default() + } + + /// 加入一条待回放的响应体。 + /// Queue a response body to replay. + pub fn push_body(&mut self, body: impl Into>) { + self.streams.lock().unwrap().push_back(body.into()); + } + + /// 加入一个待返回的传输错误。 + /// Queue a transport error to return. + #[allow(dead_code)] + pub fn push_error(&mut self, err: TransportError) { + self.errors.lock().unwrap().push_back(err); + } + + /// 最近一次请求的克隆。 + /// Clone of the most recent request. + #[allow(dead_code)] + pub fn last_request(&self) -> HttpRequest { + self.requests + .lock() + .unwrap() + .last() + .cloned() + .expect("no request recorded") + } + + /// 已记录的请求数量。 + /// Number of recorded requests. + #[allow(dead_code)] + pub fn request_count(&self) -> usize { + self.requests.lock().unwrap().len() + } +} + +impl Transport for MockTransport { + fn request<'a>( + &'a self, + req: &'a HttpRequest, + ) -> Pin< + Box< + dyn std::future::Future> + + Send + + 'a, + >, + > { + Box::pin(async move { + self.requests.lock().unwrap().push(req.clone()); + if let Some(err) = self.errors.lock().unwrap().pop_front() { + return Err(err); + } + let body = self.streams.lock().unwrap().pop_front().unwrap_or_default(); + Ok(HttpResponse { + status: 200, + headers: HashMap::new(), + body, + }) + }) + } + + fn stream_request<'a>( + &'a self, + req: &'a HttpRequest, + _timeout: Duration, + ) -> Pin< + Box< + dyn std::future::Future< + Output = focus_transport::TransportResult>, + > + Send + + 'a, + >, + > { + Box::pin(async move { + self.requests.lock().unwrap().push(req.clone()); + if let Some(err) = self.errors.lock().unwrap().pop_front() { + return Err(err); + } + let body = self.streams.lock().unwrap().pop_front().unwrap_or_default(); + Ok(Box::new(ReplayChunkStream { + data: body, + pos: 0, + chunk_size: 7, + }) as Box) + }) + } +} + +/// 按固定小尺寸逐块回放的 chunk 流。 +/// A chunk stream that replays its data in small fixed-size chunks. +struct ReplayChunkStream { + data: Vec, + pos: usize, + chunk_size: usize, +} + +impl ChunkStream for ReplayChunkStream { + fn next_chunk<'a>( + &'a mut self, + ) -> Pin< + Box< + dyn std::future::Future>>> + + Send + + 'a, + >, + > { + Box::pin(async move { + if self.pos >= self.data.len() { + Ok(None) + } else { + let end = (self.pos + self.chunk_size).min(self.data.len()); + let out = self.data[self.pos..end].to_vec(); + self.pos = end; + Ok(Some(out)) + } + }) + } +} + +/// 把一个 provider 的全部流事件收集成 Vec。 +/// Collect all stream events from a provider into a Vec. +#[allow(dead_code)] +pub fn collect( + provider: &dyn focus_core::provider::StreamProvider, + request: &focus_core::provider::ProviderRequest, +) -> Vec { + let mut iter = provider + .stream(request) + .expect("provider.stream must not return Err for a well-formed request"); + let mut out = Vec::new(); + while let Some(ev) = iter.next_event() { + out.push(ev); + } + out +} diff --git a/crates/focus-providers/tests/compaction_integration.rs b/crates/focus-providers/tests/compaction_integration.rs new file mode 100644 index 0000000..9eae6d7 --- /dev/null +++ b/crates/focus-providers/tests/compaction_integration.rs @@ -0,0 +1,198 @@ +//! 全链路集成：agent 跑对话 → harness 压缩规划 → provider 生成摘要 → +//! 摘要写回 → agent 继续（对应未来 TUI 的自动压缩流程）。 +//! Full-chain integration: agent runs a conversation → harness plans +//! compaction → the provider summarizes → the summary is written back → the +//! agent continues (mirroring the TUI's future auto-compaction flow). + +mod common; + +use focus_core::event::VecSink; +use focus_core::model::*; +use focus_core::provider::{ProviderRequest, StreamEvent}; +use focus_core::tool::{ToolEffects, ToolRegistry, ToolResult}; +use focus_core::{Agent, AgentConfig, CoreError}; +use focus_harness::compaction::{ + apply_summary, plan_compaction, DEFAULT_KEEP_RATIO, DEFAULT_THRESHOLD_RATIO, +}; +use focus_harness::prompt::{default_context, ContextUsage, SystemPromptTemplate}; +use focus_json::JsonValue; +use focus_providers::anthropic::AnthropicProvider; +use focus_providers::config::{resolve_context_window, ProviderConfig}; +use std::sync::Arc; + +/// 简单无副作用工具。 +/// A trivial side-effect-free tool. +#[derive(Debug)] +struct NoopTool; +impl focus_core::Tool for NoopTool { + fn name(&self) -> &str { + "noop" + } + fn description(&self) -> &str { + "does nothing" + } + fn parameters(&self) -> JsonValue { + let mut o = JsonValue::obj(); + o.insert("type", "object".into()).ok(); + o + } + fn effects(&self) -> ToolEffects { + ToolEffects::NONE + } + fn execute( + &self, + _id: &str, + _args: &JsonValue, + _u: Option<&focus_core::tool::ToolUpdateSink>, + ) -> Result { + Ok(ToolResult::text("ok")) + } +} + +/// 构造一条「只输出文本」的 Anthropic SSE 回复。 +/// Build an Anthropic SSE reply that only outputs text. +fn text_sse(text: &str) -> String { + format!( + "event: message_start\ndata: {{\"type\":\"message_start\",\"message\":{{\"usage\":{{\"input_tokens\":10}}}}}}\n\n\ + event: content_block_start\ndata: {{\"type\":\"content_block_start\",\"index\":0,\"content_block\":{{\"type\":\"text\",\"text\":\"\"}}}}\n\n\ + event: content_block_delta\ndata: {{\"type\":\"content_block_delta\",\"index\":0,\"delta\":{{\"type\":\"text_delta\",\"text\":\"{}\"}}}}\n\n\ + event: content_block_stop\ndata: {{\"type\":\"content_block_stop\",\"index\":0}}\n\n\ + event: message_delta\ndata: {{\"type\":\"message_delta\",\"delta\":{{\"stop_reason\":\"end_turn\"}},\"usage\":{{\"output_tokens\":5}}}}\n\n\ + event: message_stop\ndata: {{\"type\":\"message_stop\"}}\n\n", + text + ) +} + +fn agent_with(mock: common::MockTransport) -> Agent { + let provider = + AnthropicProvider::with_transport(ProviderConfig::new("sk-test"), Arc::new(mock)); + let config = AgentConfig { + model: "claude-sonnet-4".into(), + system_prompt: "You are helpful.".into(), + max_tokens: Some(256), + ..Default::default() + }; + Agent::new( + config, + Box::new(provider), + ToolRegistry::with(Box::new(NoopTool)), + ) +} + +#[test] +fn full_conversation_compaction_and_continue() { + // ---- 阶段 1：跑几轮对话（mock 每次都回同样的话）。---- + // ---- Phase 1: run a few turns (the mock replies the same text). ---- + let mut mock = common::MockTransport::new(); + for _ in 0..6 { + mock.push_body(text_sse("some answer")); + } + let mut agent = agent_with(mock); + let mut sink = VecSink::new(); + for i in 0..6 { + agent + .prompt(format!("question {}", i), &mut sink) + .expect("run failed"); + } + let messages = agent.messages().to_vec(); + assert_eq!(messages.len(), 12); // 6 user + 6 assistant / 6 user + 6 assistant + + // ---- 阶段 2：上下文窗口未知时，按三层解析取窗口。---- + // ---- Phase 2: unknown window → resolve via the three-layer lookup. ---- + let window = resolve_context_window(None, "claude-sonnet-4"); + assert_eq!(window, 200_000); + + // ---- 阶段 3：估算并规划压缩。---- + // ---- Phase 3: estimate and plan compaction. ---- + let estimated = focus_harness::estimate_messages(&messages); + assert!(estimated > 0); + // 小窗口强制触发压缩。 + // Force a trigger with a tiny window. + let plan = plan_compaction(&messages, 64, DEFAULT_THRESHOLD_RATIO, DEFAULT_KEEP_RATIO) + .expect("compaction plan"); + assert!(!plan.summarize.is_empty()); + assert!(!plan.keep.is_empty()); + + // ---- 阶段 4：用 provider「生成」摘要（同样走 mock 流）。---- + // ---- Phase 4: "generate" the summary via the provider (mock stream). ---- + let mut mock2 = common::MockTransport::new(); + mock2.push_body(text_sse( + "earlier talk- f1", + )); + let provider = + AnthropicProvider::with_transport(ProviderConfig::new("sk-test"), Arc::new(mock2)); + let summary_request = ProviderRequest { + model: "claude-sonnet-4".into(), + system_prompt: plan.summary_instruction.clone(), + messages: plan.summarize.clone(), + tools: JsonValue::arr(), + max_tokens: Some(512), + temperature: None, + }; + let events = common::collect(&provider, &summary_request); + let summary = match events.last().unwrap() { + StreamEvent::Done { message } => message + .content + .iter() + .find_map(|c| c.as_text()) + .map(|t| t.text.clone()) + .unwrap_or_default(), + other => panic!("expected done, got {:?}", other), + }; + assert!(summary.contains(""), "got: {}", summary); + + // ---- 阶段 5：写回摘要并注入动态系统提示（方案 E）。---- + // ---- Phase 5: write the summary back and inject a dynamic system + // prompt (scheme E). ---- + let compacted = apply_summary(&summary, &plan); + agent.replace_messages(compacted); + + let mut ctx = default_context(); + ctx.context_usage = Some(ContextUsage { + estimated_tokens: focus_harness::estimate_messages(agent.messages()), + context_window: window, + window_known: true, + }); + let tpl = SystemPromptTemplate::default(); + agent.set_system_prompt(tpl.render(&ctx)); + assert!(agent.config().system_prompt.contains("Context usage")); + + // ---- 阶段 6：压缩后继续对话。---- + // ---- Phase 6: continue the conversation after compaction. ---- + let mut mock3 = common::MockTransport::new(); + mock3.push_body(text_sse("post-compaction answer")); + // 重新注入 provider（mock 是一次性的）。 + // Re-inject a provider (the mock is single-use). + let provider = + AnthropicProvider::with_transport(ProviderConfig::new("sk-test"), Arc::new(mock3)); + agent.replace_messages(agent.messages().to_vec()); + // 通过内部字段替换 provider：直接重建 agent 以保留消息。 + // Rebuild the agent keeping the transcript, since the provider is not + // swappable on an existing agent. + let mut agent2 = Agent::new( + AgentConfig { + model: "claude-sonnet-4".into(), + system_prompt: agent.config().system_prompt.clone(), + max_tokens: Some(256), + ..Default::default() + }, + Box::new(provider), + ToolRegistry::with(Box::new(NoopTool)), + ); + agent2.replace_messages(agent.messages().to_vec()); + + let mut sink2 = VecSink::new(); + agent2.prompt("what now?", &mut sink2).expect("run failed"); + let msgs = agent2.messages(); + // 摘要消息 + 保留消息 + 新 user + 新 assistant。 + // Summary message + kept messages + new user + new assistant. + assert_eq!( + msgs.len(), + plan.keep.len() + 3, + "transcript: summary + keep + user + assistant" + ); + let first = &msgs[0]; + assert!(matches!(first, Message::User(_))); + let last = msgs.last().unwrap(); + assert!(matches!(last, Message::Assistant(_))); +} diff --git a/crates/focus-providers/tests/openai_tests.rs b/crates/focus-providers/tests/openai_tests.rs new file mode 100644 index 0000000..2acb6de --- /dev/null +++ b/crates/focus-providers/tests/openai_tests.rs @@ -0,0 +1,193 @@ +//! OpenAI provider（Responses + Chat Completions）的录制/回放集成测试。 +//! Recorded/replayed integration tests for the OpenAI provider (both +//! Responses and Chat Completions). + +mod common; + +use focus_core::model::*; +use focus_core::provider::{ProviderRequest, StreamEvent}; +use focus_providers::config::ProviderConfig; +use focus_providers::openai::{OpenAiProtocol, OpenAiProvider}; +use std::sync::Arc; + +fn provider(mock: common::MockTransport, protocol: OpenAiProtocol) -> OpenAiProvider { + OpenAiProvider::with_transport(ProviderConfig::new("sk-test"), protocol, Arc::new(mock)) +} + +fn request() -> ProviderRequest { + ProviderRequest { + model: "gpt-4o".into(), + system_prompt: "Be concise.".into(), + messages: vec![Message::user_text("hi")], + tools: focus_json::JsonValue::arr(), + max_tokens: Some(512), + temperature: None, + } +} + +fn final_message(events: &[StreamEvent]) -> &AssistantMessage { + events + .iter() + .find_map(|e| match e { + StreamEvent::Done { message } => Some(message), + _ => None, + }) + .expect("done event") +} + +// ---- Chat Completions ---- + +#[test] +fn chat_streams_text_and_usage() { + let mut mock = common::MockTransport::new(); + mock.push_body( + "data: {\"id\":\"c1\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"\"},\"finish_reason\":null}]}\n\n\ + data: {\"id\":\"c1\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"Hello\"},\"finish_reason\":null}]}\n\n\ + data: {\"id\":\"c1\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\" world\"},\"finish_reason\":null}]}\n\n\ + data: {\"id\":\"c1\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n\ + data: {\"id\":\"c1\",\"choices\":[],\"usage\":{\"prompt_tokens\":9,\"completion_tokens\":5}}\n\n\ + data: [DONE]\n\n", + ); + let provider = provider(mock, OpenAiProtocol::ChatCompletions); + let events = common::collect(&provider, &request()); + + // 事件序列：start → text_start → 两个 text_delta → text_end → done。 + // Sequence: start → text_start → two text_deltas → text_end → done. + assert!(matches!(events[0], StreamEvent::Start { .. })); + assert!(matches!(events[1], StreamEvent::TextStart { .. })); + assert!(matches!(events[2], StreamEvent::TextDelta { ref delta, .. } if delta == "Hello")); + assert!(matches!(events[3], StreamEvent::TextDelta { ref delta, .. } if delta == " world")); + assert!(matches!(events[4], StreamEvent::TextEnd { .. })); + + let msg = final_message(&events); + let text = msg.content[0].as_text().expect("text block"); + assert_eq!(text.text, "Hello world"); + assert_eq!(msg.stop_reason, StopReason::Stop); + assert_eq!(msg.usage.input_tokens, 9); + assert_eq!(msg.usage.output_tokens, 5); +} + +#[test] +fn chat_streams_interleaved_tool_calls() { + let mut mock = common::MockTransport::new(); + mock.push_body( + // 第一个 chunk 同时打开两个调用；随后参数交错到达。 + // First chunk opens both calls; arguments then arrive interleaved. + "data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_a\",\"type\":\"function\",\"function\":{\"name\":\"read\",\"arguments\":\"\"}},{\"index\":1,\"id\":\"call_b\",\"type\":\"function\",\"function\":{\"name\":\"write\",\"arguments\":\"\"}}]},\"finish_reason\":null}]}\n\n\ + data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"path\\\":\\\"a\\\"}\"}}]},\"finish_reason\":null}]}\n\n\ + data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":1,\"function\":{\"arguments\":\"{\\\"path\\\":\\\"b\\\"}\"}}]},\"finish_reason\":null}]}\n\n\ + data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n\ + data: [DONE]\n\n", + ); + let provider = provider(mock, OpenAiProtocol::ChatCompletions); + let events = common::collect(&provider, &request()); + + let msg = final_message(&events); + assert_eq!(msg.stop_reason, StopReason::ToolUse); + assert_eq!(msg.content.len(), 2, "two tool calls"); + let tc0 = msg.content[0].as_tool_call().expect("call 0"); + let tc1 = msg.content[1].as_tool_call().expect("call 1"); + assert_eq!(tc0.name, "read"); + assert_eq!(tc0.arguments.get_str("path"), Some("a")); + assert_eq!(tc1.name, "write"); + assert_eq!(tc1.arguments.get_str("path"), Some("b")); + // 交错参数必须归位（回归：见 focus-core reducer 修复）。 + // Interleaved arguments must land in the right slots (regression: see the + // focus-core reducer fix). + assert!(tc0.arguments.get_str("path") == Some("a")); +} + +#[test] +fn chat_encodes_error_chunk() { + let mut mock = common::MockTransport::new(); + mock.push_body( + "data: {\"error\":{\"message\":\"invalid api key\",\"type\":\"authentication_error\"}}\n\n", + ); + let provider = provider(mock, OpenAiProtocol::ChatCompletions); + let events = common::collect(&provider, &request()); + + match events.last().unwrap() { + StreamEvent::Error { error } => { + let msg = error.error_message.clone().unwrap_or_default(); + assert!(msg.contains("invalid api key"), "got: {}", msg); + } + other => panic!("expected error event, got {:?}", other), + } +} + +// ---- Responses API ---- + +#[test] +fn responses_streams_text() { + let mut mock = common::MockTransport::new(); + mock.push_body( + "event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"r1\",\"status\":\"in_progress\"}}\n\n\ + event: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"it1\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"\",\"annotations\":[]}]}}\n\n\ + event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"item_id\":\"it1\",\"output_index\":0,\"delta\":\"Hi\"}\n\n\ + event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"item_id\":\"it1\",\"output_index\":0,\"delta\":\" there\"}\n\n\ + event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"r1\",\"status\":\"completed\",\"usage\":{\"input_tokens\":12,\"output_tokens\":4,\"total_tokens\":16,\"input_tokens_details\":{\"cached_tokens\":3}}}}\n\n", + ); + let provider = provider(mock, OpenAiProtocol::Responses); + let events = common::collect(&provider, &request()); + + let msg = final_message(&events); + let text = msg.content[0].as_text().expect("text block"); + assert_eq!(text.text, "Hi there"); + assert_eq!(msg.stop_reason, StopReason::Stop); + assert_eq!(msg.usage.input_tokens, 12); + assert_eq!(msg.usage.output_tokens, 4); + assert_eq!(msg.usage.cache_read_tokens, 3); +} + +#[test] +fn responses_streams_function_call() { + let mut mock = common::MockTransport::new(); + mock.push_body( + "event: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"fc1\",\"type\":\"function_call\",\"call_id\":\"call_x\",\"name\":\"read\",\"arguments\":\"\",\"status\":\"in_progress\"}}\n\n\ + event: response.function_call_arguments.delta\ndata: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc1\",\"output_index\":0,\"delta\":\"{\\\"path\\\":\\\"x\"}\n\n\ + event: response.function_call_arguments.delta\ndata: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc1\",\"output_index\":0,\"delta\":\"y.txt\\\"}\"}\n\n\ + event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":6}}}\n\n", + ); + let provider = provider(mock, OpenAiProtocol::Responses); + let events = common::collect(&provider, &request()); + + let msg = final_message(&events); + assert_eq!(msg.content.len(), 1); + let tc = msg.content[0].as_tool_call().expect("tool call"); + assert_eq!(tc.name, "read"); + assert_eq!(tc.arguments.get_str("path"), Some("xy.txt")); +} + +#[test] +fn responses_encodes_error() { + let mut mock = common::MockTransport::new(); + mock.push_body( + "event: error\ndata: {\"type\":\"error\",\"code\":\"invalid_request_error\",\"message\":\"bad model\"}\n\n", + ); + let provider = provider(mock, OpenAiProtocol::Responses); + let events = common::collect(&provider, &request()); + + match events.last().unwrap() { + StreamEvent::Error { error } => { + let msg = error.error_message.clone().unwrap_or_default(); + assert!(msg.contains("bad model"), "got: {}", msg); + } + other => panic!("expected error event, got {:?}", other), + } +} + +#[test] +fn request_headers_and_path() { + let mut mock = common::MockTransport::new(); + mock.push_body("data: [DONE]\n\n"); + let provider = provider(mock.clone(), OpenAiProtocol::ChatCompletions); + let _ = common::collect(&provider, &request()); + + let req = mock.last_request(); + assert_eq!(req.host, "api.openai.com"); + assert_eq!(req.path, "/v1/chat/completions"); + let headers: Vec<(String, String)> = req.headers.clone(); + assert!(headers + .iter() + .any(|(k, v)| { k.eq_ignore_ascii_case("authorization") && v == "Bearer sk-test" })); +}