feat(providers): rewrite Anthropic and OpenAI providers

- Anthropic Messages API (2023-06-01) SSE streaming: text/thinking/tool_use
  blocks, cache usage, stop-reason mapping, errors encoded as events
- OpenAI Responses API + Chat Completions (switchable), tool_calls delta
  accumulation, include_usage, reasoning_content -> thinking
- ProviderConfig with base_url/api_key/context_window/timeout; known-model
  context-window table with three-layer resolution
- background current-thread runtime bridges sync agent loop to async I/O
- mock-transport replay tests (chunks straddle SSE boundaries) + full agent
  integration test

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

crates/focus-providers/src/common.rs

@@ -1,140 +1,94 @@
-//! provider 实现的共享辅助：请求配置、用于测试的 mock transport，
+//! provider 共享的流式驱动与辅助类型。
-//! 以及从配置组装 model-id 的辅助函数。
+//! Shared streaming driver and helper types for providers.
-//! Shared helpers for provider implementations: request config, a mock
+//!
-//! transport for tests, and helpers to assemble the model-id from config.
+//! `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_core::model::{now_ms, AssistantMessage, StopReason, Usage};
-use focus_transport::{HttpRequest, HttpResponse, Transport};
+use focus_core::provider::{StreamEvent, StreamIterator, StreamResult};
-use std::collections::HashMap;
+use focus_core::CoreError;
-use std::sync::{Arc, Mutex};
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::mpsc::{Receiver, SyncSender};
 
-/// provider 配置：API key、model id、base URL。
+/// 在后台线程上驱动一个异步闭包，把事件经同步 channel 送到主线程。
-/// Configuration for a provider: API key, model id, base URL.
+/// Drive an async closure on a background thread, delivering events to the
-#[derive(Debug, Clone)]
+/// main thread through a sync channel.
-pub struct ProviderConfig {
+///
-    /// API 密钥（或待解析的 `$VAR`/`${VAR}` 模板）。
+/// `drive` 负责：发请求 → 读流 → 翻译成 [`StreamEvent`] 并 `tx.send`，
-    /// API key (or a `$VAR`/`${VAR}` template to be resolved).
+/// 并且必须**始终**以终止性事件（`Done`/`Error`）收尾，否则 agent 循环会
-    pub api_key: String,
+/// 合成一条错误消息（见 `focus-core` 的 `stream_assistant_response`）。
-    /// 模型 id，例如 `claude-sonnet-4-5`、`glm-5.2`。
+/// `drive` must always end with a terminal event (`Done`/`Error`); otherwise
-    /// Model id, e.g. `claude-sonnet-4-5`, `glm-5.2`.
+/// the agent loop synthesizes an error message (see `focus-core`'s
-    pub model: String,
+/// `stream_assistant_response`).
-    /// 仅主机名（不含 scheme、不含 path），例如 `api.anthropic.com`。
+pub(crate) fn run_stream(
-    /// Host only (no scheme, no path), e.g. `api.anthropic.com`.
+    drive: impl FnOnce(SyncSender<StreamEvent>) -> Pin<Box<dyn Future<Output = ()> + Send>>
-    pub base_url: String,
+        + Send
-    /// 默认最大输出 token 数。
+        + 'static,
-    /// Default max output tokens.
+) -> StreamResult {
-    pub max_tokens: u64,
+    let (tx, rx) = std::sync::mpsc::sync_channel(256);
-    /// 请求超时（毫秒）。
+    std::thread::Builder::new()
-    /// Request timeout in milliseconds.
+        .name("focus-provider-stream".into())
-    pub timeout_ms: u64,
+        .spawn(move || {
-}
+            // 每个流使用独立的 current-thread 运行时，避免依赖调用方是否有
-
+            // 活动的 tokio 运行时（agent 循环是同步的）。
-impl ProviderConfig {
+            // Each stream gets its own current-thread runtime, so we don't
-    /// Anthropic 默认配置（`api.anthropic.com`，4096 tokens，120s 超时）。
+            // depend on the caller having an active tokio runtime (the agent
-    /// Default Anthropic config (`api.anthropic.com`, 4096 tokens, 120s timeout).
+            // loop is synchronous).
-    pub fn anthropic(api_key: impl Into<String>, model: impl Into<String>) -> Self {
+            let rt = match tokio::runtime::Builder::new_current_thread()
-        Self {
+                .enable_all()
-            api_key: api_key.into(),
+                .build()
-            model: model.into(),
+            {
-            base_url: "api.anthropic.com".into(),
+                Ok(rt) => rt,
-            max_tokens: 4096,
+                Err(e) => {
-            timeout_ms: 120_000,
+                    let _ = tx.send(StreamEvent::Error {
-        }
+                        error: error_message("unknown", &format!("failed to start runtime: {}", e)),
-    }
+                    });
-
+                    return;
-    /// OpenAI 默认配置（`api.openai.com`，4096 tokens，120s 超时）。
+                }
-    /// Default OpenAI config (`api.openai.com`, 4096 tokens, 120s timeout).
+            };
-    pub fn openai(api_key: impl Into<String>, model: impl Into<String>) -> Self {
+            rt.block_on(drive(tx));
-        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<String>,
-        model: impl Into<String>,
-        base_url: impl Into<String>,
-    ) -> 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<String>, model: impl Into<String>) -> Self {
-        Self::new(api_key, model, "open.bigmodel.cn")
-    }
-
-    /// Z.ai 配置（国际端点）。
-    /// Config for Z.ai (国际端点).
-    pub fn zai_intl(api_key: impl Into<String>, model: impl Into<String>) -> 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<Mutex<Option<HttpResponse>>>,
-}
-
-impl MockTransport {
-    /// 用给定状态码与 SSE 文本体构造一个预置响应。
-    /// Build a canned response with the given status code and SSE body.
-    pub fn new(status: u16, body: impl Into<String>) -> 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<Output = focus_transport::TransportResult<HttpResponse>>
-                + Send
-                + 'a,
-        >,
-    > {
-        Box::pin(async move {
-            self.response
-                .lock()
-                .unwrap()
-                .clone()
-                .ok_or_else(|| focus_transport::TransportError::Io("no canned response".into()))
         })
+        .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<StreamEvent>,
+}
+
+impl StreamIterator for ChannelIterator {
+    fn next_event(&mut self) -> Option<StreamEvent> {
+        // 所有发送者都断开（后台线程结束）时 recv 返回 Err → None（流结束）。
+        // When all senders drop (the background thread finished), recv returns
+        // Err → None (end of stream).
+        self.rx.recv().ok()
     }
 }
 
-/// 暂时不做任何剥离——这是为未来共享转换预留的挂载点。
+/// 构造一条错误 assistant 消息（stop_reason = Error）。
-/// Strip and ignore nothing for now — a place to hang shared transformations.
+/// Build an error assistant message (stop_reason = Error).
-pub fn _unused_keep_request_in_scope(_req: &ProviderRequest) {}
+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<StreamEvent>, event: StreamEvent) -> bool {
+    tx.send(event).is_ok()
+}

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<String>,
+    /// 模型上下文窗口大小（token 数）；`None` 时按已知模型表兜底。
+    /// Model context window size (tokens); `None` falls back to the known-model
+    /// table.
+    pub context_window: Option<u64>,
+    /// 单次网络操作的超时时间。
+    /// 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<String>) -> 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<u64> {
+    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<u64>, 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());
+    }
+}

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 格式翻译为统一的
+//! 目前包含 Anthropic（Messages API）与 OpenAI（Responses API + Chat
-//! [`StreamEvent`] 协议，从而实现 [`focus_core::StreamProvider`]。
+//! Completions 双协议）。所有 provider 都把 HTTP/SSE 翻译成
-//! Each provider implements [`focus_core::StreamProvider`] by translating its
+//! [`focus_core::provider::StreamEvent`] 的统一流，错误一律编码为
-//! vendor-specific HTTP/SSE format into the unified [`StreamEvent`] protocol.
+//! `StreamEvent::Error`，绝不通过返回 `Err` 表示 API 失败。
-//!
+//! Currently ships Anthropic (Messages API) and OpenAI (both the Responses
-//! 主要目标是 [`ZaiProvider`]（智谱 GLM 编程模型，国内端点），它构建在共享的
+//! API and Chat Completions). Every provider translates HTTP/SSE into the
-//! [`OpenAiCompatProvider`] 引擎之上。此外也提供 Anthropic 与原生 OpenAI provider。
+//! unified [`focus_core::provider::StreamEvent`] stream; failures are always
-//! The primary target is [`ZaiProvider`] (智谱 GLM coding models, domestic
+//! encoded as `StreamEvent::Error` rather than returned as `Err`.
-//! endpoint), built on the shared [`OpenAiCompatProvider`] engine.
-//! Anthropic and raw-OpenAI providers are also available.
 
 #![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;
+/// Config types and the known-model context-window table.
-pub mod zai;
+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::{resolve_context_window, ProviderConfig};
-pub use config_value::resolve as resolve_config_value;
+pub use openai::{OpenAiProtocol, OpenAiProvider};
-pub use openai_compat::OpenAiCompatProvider;
-pub use zai::{ZaiProvider, ZAI_CODING_PATH};

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<dyn Transport>,
+}
+
+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", &"<Arc<dyn 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<dyn Transport>,
+    ) -> 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::<Vec<_>>()
+        .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<usize>,
+    /// 输入 / 输出 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<StreamEvent>) -> Result<bool, String> {
+        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<StreamEvent>) -> Result<bool, String> {
+        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<StreamEvent>,
+    ) -> Result<bool, String> {
+        // 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<StreamEvent>,
+    ) -> Result<bool, String> {
+        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<focus_core::tool::ToolResult, CoreError> {
+            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);
+    }
+}

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<std::sync::Mutex<Vec<JsonValue>>>,
+}
+
+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<ToolResult, CoreError> {
+        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);
+}

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));
+}

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<Mutex<VecDeque<Vec<u8>>>>,
+    /// 待返回的传输层错误（每个消费一个，优先于 streams）。
+    /// Transport errors to return (one consumed per request; take precedence).
+    errors: Arc<Mutex<VecDeque<TransportError>>>,
+    /// 已发出的请求记录。
+    /// Requests sent so far.
+    requests: Arc<Mutex<Vec<HttpRequest>>>,
+}
+
+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<Vec<u8>>) {
+        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<Output = focus_transport::TransportResult<HttpResponse>>
+                + 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<Box<dyn ChunkStream + Send>>,
+                > + 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<dyn ChunkStream + Send>)
+        })
+    }
+}
+
+/// 按固定小尺寸逐块回放的 chunk 流。
+/// A chunk stream that replays its data in small fixed-size chunks.
+struct ReplayChunkStream {
+    data: Vec<u8>,
+    pos: usize,
+    chunk_size: usize,
+}
+
+impl ChunkStream for ReplayChunkStream {
+    fn next_chunk<'a>(
+        &'a mut self,
+    ) -> Pin<
+        Box<
+            dyn std::future::Future<Output = focus_transport::TransportResult<Option<Vec<u8>>>>
+                + 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<focus_core::provider::StreamEvent> {
+    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
+}

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<ToolResult, CoreError> {
+        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(
+        "<summary>earlier talk</summary><key-facts>- f1</key-facts>",
+    ));
+    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("<summary>"), "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(_)));
+}

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" }));
+}