mirror of
https://github.com/farion1231/cc-switch.git
synced 2026-07-31 11:01:36 +08:00
Dex Miller
67e074c0a7
* style(frontend): reformat provider forms, constants and hooks
Apply prettier formatting across 5 frontend files. No logic changes.
Changed files:
- AddProviderDialog.tsx: reformat generic type annotation and callback
- ClaudeFormFields.tsx: consolidate multi-line useState and Collapsible props
- CodexConfigSections.tsx: expand single-line React imports to multi-line,
collapse removeCodexTopLevelField() call
- constants.ts: merge TemplateType into single line
- useSkills.ts: expand single-line TanStack Query imports to multi-line,
reformat uninstallSkill mutationFn chain
* deps(proxy): add hyper ecosystem crates and manual decompression libs
reqwest internally normalizes all header names to lowercase and does not
preserve insertion order, causing proxied requests to differ from the
original client requests. To achieve transparent header forwarding with
original casing and order, introduce lower-level hyper HTTP client libs.
New dependencies:
- hyper-util 0.1: TokioExecutor + legacy Client with
preserve_header_case support for HTTP/1.1
- hyper-rustls 0.27: rustls-based TLS connector for hyper
- http 1 / http-body 1 / http-body-util 0.1: HTTP type crates for
hyper 1.x request/response construction
- flate2 1: manual gzip/deflate decompression (replaces reqwest auto)
- brotli 7: manual brotli decompression
Changed dependencies:
- serde_json: enable preserve_order feature to keep JSON field order
- reqwest: drop gzip feature to prevent reqwest from overriding the
client's original accept-encoding header
* refactor(proxy): use hyper client for header-case preserving forwarding
Previously the proxy used reqwest for all upstream requests. reqwest
normalizes header names to lowercase and reorders them internally,
making proxied requests distinguishable from direct CLI requests.
Some upstream providers are sensitive to these differences.
This commit replaces reqwest with a hyper-based HTTP client on the
default (non-proxy) path, achieving wire-level header fidelity:
Server layer (server.rs):
- Replace axum::serve with a manual hyper HTTP/1.1 accept loop
- Enable preserve_header_case(true) so incoming header casing is
captured in a HeaderCaseMap extension on each request
- Bridge hyper requests to axum Router via tower::Service
New hyper client module (hyper_client.rs):
- Lazy-initialized hyper-util Client with preserve_header_case
- ProxyResponse enum wrapping both hyper::Response and reqwest::Response
behind a unified interface (status, headers, bytes, bytes_stream)
- send_request() builds requests with ordered HeaderMap + case map
Request handlers (handlers.rs):
- Switch from (HeaderMap, Json<Value>) extractors to raw
axum::extract::Request to preserve Extensions (containing the
HeaderCaseMap from the accept loop)
- Pass extensions through the forwarding chain
Forwarder (forwarder.rs):
- Remove HEADER_BLACKLIST array; replace with ordered header iteration
that preserves original header sequence and casing
- Build ordered_headers by iterating client headers, skipping only
auth/host/content-length, and inserting auth headers at the original
authorization position to maintain order
- Handle anthropic-beta (ensure claude-code-20250219 tag) and
anthropic-version (passthrough or default) inline during iteration
- Remove should_force_identity_encoding() — accept-encoding is now
transparently forwarded to upstream
- Use hyper client by default; fall back to reqwest only when an
HTTP/SOCKS5 proxy tunnel is configured
Provider adapters (adapter.rs, claude.rs, codex.rs, gemini.rs):
- Replace add_auth_headers(RequestBuilder) -> RequestBuilder with
get_auth_headers(AuthInfo) -> Vec<(HeaderName, HeaderValue)>
- Adapters now return header pairs instead of mutating a reqwest builder
- Claude adapter: merge Anthropic/ClaudeAuth/Bearer into single branch;
move Copilot fingerprint headers into get_auth_headers
Response processing (response_processor.rs):
- Add manual decompression (gzip/deflate/brotli via flate2 + brotli)
for non-streaming responses, since reqwest auto-decompression is now
disabled to allow accept-encoding passthrough
- Add compressed-SSE warning log for streaming responses
- Accept ProxyResponse instead of reqwest::Response
HTTP client (http_client.rs):
- Disable reqwest auto-decompression (.no_gzip/.no_brotli/.no_deflate)
on both global and per-provider clients
Streaming adapters (streaming.rs, streaming_responses.rs):
- Generalize stream error type from reqwest::Error to generic E: Error
Misc:
- log_codes.rs: add SRV-005 (ACCEPT_ERR) and SRV-006 (CONN_ERR)
- stream_check.rs: reformat copilot header lines
- transform.rs: fix trailing whitespace alignment
* fix(lint): resolve 35 clippy warnings across Rust codebase
Fix all clippy warnings reported by `cargo clippy --lib`:
- codex_config.rs: fix doc_overindented_list_items (3 spaces -> 2)
- commands/copilot.rs: inline format args in 2 log::error! calls
- commands/provider.rs: inline format args in 3 map_err closures
- proxy/hyper_client.rs: inline format arg in log::debug! call
- proxy/providers/copilot_auth.rs: inline format args in 16 locations
(log macros, format! in headers, error constructors)
- proxy/thinking_optimizer.rs: inline format args in 2 log::info! calls
- services/skill.rs: inline format args in log::debug! call
- services/webdav_sync.rs: inline format args in 6 format! calls
(version compat messages, download limit messages)
- services/webdav_sync/archive.rs: inline format args in 2 format! calls
- session_manager/providers/opencode.rs: inline format args in
source_path format!
All fixes use the clippy::uninlined_format_args suggestion pattern:
format!("msg: {}", var) -> format!("msg: {var}")
* deps(proxy): add raw HTTP write and native TLS cert dependencies
Add crates required for the raw TCP/TLS write path that bypasses
hyper's header encoder to preserve original header name casing:
- httparse: parse raw TCP peek bytes to capture header casings
- tokio-rustls + rustls: direct TLS connections for raw write path
- webpki-roots: Mozilla CA bundle baseline
- rustls-native-certs: load system keychain CAs (trusts proxy MITM
certificates from Clash, mitmproxy, etc.)
* fix(proxy): address code review feedback on response handling
Fixes from PR #1714 code review:
- Extract `read_decoded_body()` and `strip_entity_headers_for_rebuilt_body()`
in response_processor to properly clean content-encoding/content-length
headers after decompression
- Reuse `read_decoded_body()` in handlers.rs for Claude transform path,
ensuring compressed responses are decoded before format conversion
- Make `build_proxy_url_from_config()` public so forwarder can pass proxy
URL to the hyper raw write path
- Add `has_system_proxy_env()` utility with test coverage
- Add 50ms backoff after accept() failures in server.rs to prevent
tight-loop CPU spin on transient socket errors
* feat(proxy): implement raw TCP/TLS write with HTTP CONNECT tunnel
Rewrite hyper_client with a two-tier strategy for header case preservation:
Primary path (raw write):
- Peek raw TCP bytes in server.rs to capture OriginalHeaderCases before
hyper lowercases them
- Build raw HTTP/1.1 request bytes with exact original header name casing
- Write directly to TLS stream, then use WriteFilter to let hyper parse
the response while discarding its duplicate request writes
- Support HTTP CONNECT tunneling through upstream proxies, so header case
is preserved even when a proxy (Clash, V2Ray) is configured
Fallback path (hyper-util Client):
- Used when OriginalHeaderCases is empty or raw write fails
- Configured with title_case_headers(true) for best-effort casing
TLS improvements:
- Load native system certificates alongside webpki roots so proxy MITM
CAs (installed in system keychain) are trusted through CONNECT tunnels
Key types added:
- OriginalHeaderCases: maps lowercase name → original wire-casing bytes
- WriteFilter<S>: AsyncRead+AsyncWrite wrapper that discards writes
- connect_via_proxy(): HTTP CONNECT tunnel establishment
- ExtensionDebugMarker: diagnostic marker for extension chain debugging
* refactor(proxy): route requests through hyper with proxy-aware forwarding
Rework forwarder request dispatch to always prefer the hyper raw write
path (header case preservation) over reqwest:
Request routing:
- HTTP/HTTPS proxy: hyper raw write through CONNECT tunnel (case preserved)
- SOCKS5 proxy: reqwest fallback (CONNECT not supported for SOCKS5)
- No proxy: hyper raw write direct connection
Header handling improvements:
- Replace host header in-place at original position instead of
skip-and-append, preserving client's header ordering
- Preserve client's original accept-encoding for transparent passthrough;
only force identity encoding when transform path needs decompression
- Add should_force_identity_encoding() to centralize the decision
- Remove hardcoded 'br, gzip, deflate' override that masked client values
Proxy URL resolution (priority order):
1. Provider-specific proxy config (if enabled)
2. Global proxy URL configured in CC Switch
3. Direct connection (no proxy)
* chore(proxy): remove dead code, redundant tests and debug scaffolding
- Inline should_force_identity_encoding() (was just `needs_transform`)
and delete its 5 test cases
- Remove ExtensionDebugMarker diagnostic type
- Remove unused has_system_proxy_env() and its test
- Remove strip_entity_headers test
- Simplify hyper path: remove redundant is_socks_proxy ternary
- Update hyper_client module doc to reflect CONNECT tunnel support
* fix(proxy): block direct-connect fallback and complete CONNECT tunnel support
* feat(hooks): improve proxy requirement warnings with specific reasons
- Remove redundant OpenAI format hint toast messages
- Add detailed reason detection for proxy requirements (OpenAI Chat, OpenAI Responses, full URL mode)
- Update i18n files with new reason-specific keys
* style(*): format code with prettier
- Remove extra whitespace in http_client.rs
- Fix formatting issues in useProviderActions.ts
* fix(proxy): post-merge fixes for forward return type and clippy warnings
- Restore forward() return type to (ProxyResponse, Option<String>)
to pass claude_api_format through to callers
- Inline format args in log::warn! macro (clippy::uninlined_format_args)
- Suppress too_many_arguments for check_claude_stream
* refactor(proxy): preserve original header wire order and add non-streaming body timeout
- Rewrite build_raw_request to emit headers in original
client-sent sequence instead of hash-map order
- Remove unused OriginalHeaderCases::get_all method
- Add body_timeout to read_decoded_body to prevent
requests hanging when upstream stalls after headers
1031 lines
54 KiB
Rust
1031 lines
54 KiB
Rust
//! OpenAI Responses API 流式转换模块
|
|
//!
|
|
//! 实现 Responses API SSE → Anthropic SSE 格式转换。
|
|
//!
|
|
//! Responses API 使用命名事件 (named events) 的生命周期模型：
|
|
//! response.created → output_item.added → content_part.added →
|
|
//! output_text.delta → content_part.done → output_item.done → response.completed
|
|
//!
|
|
//! 与 Chat Completions 的 delta chunk 模型完全不同，需要独立的状态机处理。
|
|
|
|
use super::transform_responses::{build_anthropic_usage_from_responses, map_responses_stop_reason};
|
|
use crate::proxy::sse::strip_sse_field;
|
|
use bytes::Bytes;
|
|
use futures::stream::{Stream, StreamExt};
|
|
use serde_json::{json, Value};
|
|
use std::collections::{HashMap, HashSet};
|
|
|
|
#[inline]
|
|
fn response_object_from_event(data: &Value) -> &Value {
|
|
data.get("response").unwrap_or(data)
|
|
}
|
|
|
|
#[inline]
|
|
fn content_part_key(data: &Value) -> Option<String> {
|
|
if let (Some(item_id), Some(content_index)) = (
|
|
data.get("item_id").and_then(|v| v.as_str()),
|
|
data.get("content_index").and_then(|v| v.as_u64()),
|
|
) {
|
|
return Some(format!("part:{item_id}:{content_index}"));
|
|
}
|
|
if let (Some(output_index), Some(content_index)) = (
|
|
data.get("output_index").and_then(|v| v.as_u64()),
|
|
data.get("content_index").and_then(|v| v.as_u64()),
|
|
) {
|
|
return Some(format!("part:out:{output_index}:{content_index}"));
|
|
}
|
|
None
|
|
}
|
|
|
|
#[inline]
|
|
fn tool_item_key_from_added(data: &Value, item: &Value) -> Option<String> {
|
|
if let Some(item_id) = item.get("id").and_then(|v| v.as_str()) {
|
|
return Some(format!("tool:{item_id}"));
|
|
}
|
|
if let Some(item_id) = data.get("item_id").and_then(|v| v.as_str()) {
|
|
return Some(format!("tool:{item_id}"));
|
|
}
|
|
if let Some(output_index) = data.get("output_index").and_then(|v| v.as_u64()) {
|
|
return Some(format!("tool:out:{output_index}"));
|
|
}
|
|
None
|
|
}
|
|
|
|
#[inline]
|
|
fn tool_item_key_from_event(data: &Value) -> Option<String> {
|
|
if let Some(item_id) = data.get("item_id").and_then(|v| v.as_str()) {
|
|
return Some(format!("tool:{item_id}"));
|
|
}
|
|
if let Some(output_index) = data.get("output_index").and_then(|v| v.as_u64()) {
|
|
return Some(format!("tool:out:{output_index}"));
|
|
}
|
|
None
|
|
}
|
|
|
|
/// Resolve content index for a text/refusal content part event.
|
|
///
|
|
/// Uses `content_part_key` to look up or assign a stable index, falling back to
|
|
/// `fallback_open_index` when no key is available.
|
|
#[inline]
|
|
fn resolve_content_index(
|
|
data: &Value,
|
|
next_content_index: &mut u32,
|
|
index_by_key: &mut HashMap<String, u32>,
|
|
fallback_open_index: &mut Option<u32>,
|
|
) -> u32 {
|
|
if let Some(k) = content_part_key(data) {
|
|
if let Some(existing) = index_by_key.get(&k).copied() {
|
|
existing
|
|
} else {
|
|
let assigned = *next_content_index;
|
|
*next_content_index += 1;
|
|
index_by_key.insert(k, assigned);
|
|
assigned
|
|
}
|
|
} else if let Some(existing) = *fallback_open_index {
|
|
existing
|
|
} else {
|
|
let assigned = *next_content_index;
|
|
*next_content_index += 1;
|
|
*fallback_open_index = Some(assigned);
|
|
assigned
|
|
}
|
|
}
|
|
|
|
/// 创建从 Responses API SSE 到 Anthropic SSE 的转换流
|
|
///
|
|
/// 状态机跟踪: message_id, current_model, has_sent_message_start, item/content index map
|
|
/// SSE 解析支持 named events (event: + data: 行)
|
|
pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send + 'static>(
|
|
stream: impl Stream<Item = Result<Bytes, E>> + Send + 'static,
|
|
) -> impl Stream<Item = Result<Bytes, std::io::Error>> + Send {
|
|
async_stream::stream! {
|
|
let mut buffer = String::new();
|
|
let mut message_id: Option<String> = None;
|
|
let mut current_model: Option<String> = None;
|
|
let mut has_sent_message_start = false;
|
|
let mut has_tool_use = false;
|
|
let mut next_content_index: u32 = 0;
|
|
let mut index_by_key: HashMap<String, u32> = HashMap::new();
|
|
let mut open_indices: HashSet<u32> = HashSet::new();
|
|
let mut fallback_open_index: Option<u32> = None;
|
|
let mut current_text_index: Option<u32> = None;
|
|
let mut tool_index_by_item_id: HashMap<String, u32> = HashMap::new();
|
|
let mut last_tool_index: Option<u32> = None;
|
|
|
|
tokio::pin!(stream);
|
|
|
|
while let Some(chunk) = stream.next().await {
|
|
match chunk {
|
|
Ok(bytes) => {
|
|
let text = String::from_utf8_lossy(&bytes);
|
|
buffer.push_str(&text);
|
|
|
|
// SSE 事件由 \n\n 分隔
|
|
while let Some(pos) = buffer.find("\n\n") {
|
|
let block = buffer[..pos].to_string();
|
|
buffer = buffer[pos + 2..].to_string();
|
|
|
|
if block.trim().is_empty() {
|
|
continue;
|
|
}
|
|
|
|
// 解析 SSE 块：提取 event: 和 data: 行
|
|
let mut event_type: Option<String> = None;
|
|
let mut data_parts: Vec<String> = Vec::new();
|
|
|
|
for line in block.lines() {
|
|
if let Some(evt) = strip_sse_field(line, "event") {
|
|
event_type = Some(evt.trim().to_string());
|
|
} else if let Some(d) = strip_sse_field(line, "data") {
|
|
data_parts.push(d.to_string());
|
|
}
|
|
}
|
|
|
|
if data_parts.is_empty() {
|
|
continue;
|
|
}
|
|
|
|
let data_str = data_parts.join("\n");
|
|
let event_name = event_type.as_deref().unwrap_or("");
|
|
|
|
// 解析 JSON 数据
|
|
let data: Value = match serde_json::from_str(&data_str) {
|
|
Ok(v) => v,
|
|
Err(_) => continue,
|
|
};
|
|
|
|
log::debug!("[Claude/Responses] <<< SSE event: {event_name}");
|
|
|
|
match event_name {
|
|
// ================================================
|
|
// response.created → message_start
|
|
// ================================================
|
|
"response.created" => {
|
|
let response_obj = response_object_from_event(&data);
|
|
if let Some(id) = response_obj.get("id").and_then(|i| i.as_str()) {
|
|
message_id = Some(id.to_string());
|
|
}
|
|
if let Some(model) =
|
|
response_obj.get("model").and_then(|m| m.as_str())
|
|
{
|
|
current_model = Some(model.to_string());
|
|
}
|
|
|
|
has_sent_message_start = true;
|
|
// Build usage with cache tokens if available
|
|
let start_usage = build_anthropic_usage_from_responses(
|
|
response_obj.get("usage"),
|
|
);
|
|
|
|
let event = json!({
|
|
"type": "message_start",
|
|
"message": {
|
|
"id": message_id.clone().unwrap_or_default(),
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"model": current_model.clone().unwrap_or_default(),
|
|
"usage": start_usage
|
|
}
|
|
});
|
|
let sse = format!("event: message_start\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
log::debug!("[Claude/Responses] >>> Anthropic SSE: message_start");
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
|
|
// ================================================
|
|
// response.content_part.added → content_block_start (text)
|
|
// ================================================
|
|
"response.content_part.added" => {
|
|
// 确保 message_start 已发送
|
|
if !has_sent_message_start {
|
|
let start_event = json!({
|
|
"type": "message_start",
|
|
"message": {
|
|
"id": message_id.clone().unwrap_or_default(),
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"model": current_model.clone().unwrap_or_default(),
|
|
"usage": { "input_tokens": 0, "output_tokens": 0 }
|
|
}
|
|
});
|
|
let sse = format!("event: message_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
has_sent_message_start = true;
|
|
}
|
|
|
|
if let Some(part) = data.get("part") {
|
|
let part_type = part.get("type").and_then(|t| t.as_str());
|
|
if matches!(part_type, Some("output_text") | Some("refusal")) {
|
|
let index = if let Some(index) = current_text_index {
|
|
index
|
|
} else {
|
|
let index = resolve_content_index(
|
|
&data,
|
|
&mut next_content_index,
|
|
&mut index_by_key,
|
|
&mut fallback_open_index,
|
|
);
|
|
current_text_index = Some(index);
|
|
index
|
|
};
|
|
|
|
if open_indices.contains(&index) {
|
|
continue;
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "text",
|
|
"text": ""
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
open_indices.insert(index);
|
|
}
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.output_text.delta → content_block_delta (text_delta)
|
|
// ================================================
|
|
"response.output_text.delta" => {
|
|
if let Some(delta) = data.get("delta").and_then(|d| d.as_str()) {
|
|
let index = if let Some(index) = current_text_index {
|
|
index
|
|
} else {
|
|
let index = resolve_content_index(
|
|
&data,
|
|
&mut next_content_index,
|
|
&mut index_by_key,
|
|
&mut fallback_open_index,
|
|
);
|
|
current_text_index = Some(index);
|
|
index
|
|
};
|
|
|
|
if !open_indices.contains(&index) {
|
|
let start_event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "text",
|
|
"text": ""
|
|
}
|
|
});
|
|
let start_sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(start_sse));
|
|
open_indices.insert(index);
|
|
}
|
|
let event = json!({
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {
|
|
"type": "text_delta",
|
|
"text": delta
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.refusal.delta → content_block_delta (text_delta)
|
|
// ================================================
|
|
"response.refusal.delta" => {
|
|
if let Some(delta) = data.get("delta").and_then(|d| d.as_str()) {
|
|
let index = if let Some(index) = current_text_index {
|
|
index
|
|
} else {
|
|
let index = resolve_content_index(
|
|
&data,
|
|
&mut next_content_index,
|
|
&mut index_by_key,
|
|
&mut fallback_open_index,
|
|
);
|
|
current_text_index = Some(index);
|
|
index
|
|
};
|
|
|
|
if !open_indices.contains(&index) {
|
|
let start_event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "text",
|
|
"text": ""
|
|
}
|
|
});
|
|
let start_sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(start_sse));
|
|
open_indices.insert(index);
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {
|
|
"type": "text_delta",
|
|
"text": delta
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.content_part.done → content_block_stop
|
|
// ================================================
|
|
"response.content_part.done" => {}
|
|
|
|
// ================================================
|
|
// response.output_item.added (function_call) → content_block_start (tool_use)
|
|
// ================================================
|
|
"response.output_item.added" => {
|
|
if let Some(item) = data.get("item") {
|
|
let item_type = item.get("type").and_then(|t| t.as_str()).unwrap_or("");
|
|
if item_type == "function_call" {
|
|
has_tool_use = true;
|
|
if let Some(index) = current_text_index.take() {
|
|
if open_indices.remove(&index) {
|
|
let stop_event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let stop_sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(stop_sse));
|
|
}
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
// 确保 message_start 已发送
|
|
if !has_sent_message_start {
|
|
let start_event = json!({
|
|
"type": "message_start",
|
|
"message": {
|
|
"id": message_id.clone().unwrap_or_default(),
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"model": current_model.clone().unwrap_or_default(),
|
|
"usage": { "input_tokens": 0, "output_tokens": 0 }
|
|
}
|
|
});
|
|
let sse = format!("event: message_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
has_sent_message_start = true;
|
|
}
|
|
|
|
let call_id = item.get("call_id").and_then(|i| i.as_str()).unwrap_or("");
|
|
let name = item.get("name").and_then(|n| n.as_str()).unwrap_or("");
|
|
let index = if let Some(k) = tool_item_key_from_added(&data, item) {
|
|
if let Some(existing) = index_by_key.get(&k).copied() {
|
|
existing
|
|
} else {
|
|
let assigned = next_content_index;
|
|
next_content_index += 1;
|
|
index_by_key.insert(k, assigned);
|
|
assigned
|
|
}
|
|
} else {
|
|
let assigned = next_content_index;
|
|
next_content_index += 1;
|
|
assigned
|
|
};
|
|
if let Some(item_id) = item
|
|
.get("id")
|
|
.and_then(|v| v.as_str())
|
|
.or_else(|| data.get("item_id").and_then(|v| v.as_str()))
|
|
{
|
|
tool_index_by_item_id.insert(item_id.to_string(), index);
|
|
}
|
|
last_tool_index = Some(index);
|
|
|
|
if open_indices.contains(&index) {
|
|
continue;
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "tool_use",
|
|
"id": call_id,
|
|
"name": name
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
open_indices.insert(index);
|
|
}
|
|
// message type output_item.added is handled via content_part.added
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.function_call_arguments.delta → content_block_delta (input_json_delta)
|
|
// ================================================
|
|
"response.function_call_arguments.delta" => {
|
|
if let Some(delta) = data.get("delta").and_then(|d| d.as_str()) {
|
|
let item_id = data.get("item_id").and_then(|v| v.as_str());
|
|
let index = if let Some(id) = item_id {
|
|
tool_index_by_item_id.get(id).copied()
|
|
} else {
|
|
None
|
|
}
|
|
.or_else(|| {
|
|
tool_item_key_from_event(&data)
|
|
.and_then(|k| index_by_key.get(&k).copied())
|
|
})
|
|
.or(last_tool_index)
|
|
.unwrap_or_else(|| {
|
|
let assigned = next_content_index;
|
|
next_content_index += 1;
|
|
assigned
|
|
});
|
|
|
|
if !open_indices.contains(&index) {
|
|
let start_event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "tool_use",
|
|
"id": data
|
|
.get("call_id")
|
|
.and_then(|v| v.as_str())
|
|
.or(item_id)
|
|
.unwrap_or(""),
|
|
"name": data
|
|
.get("name")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("")
|
|
}
|
|
});
|
|
let start_sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(start_sse));
|
|
open_indices.insert(index);
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {
|
|
"type": "input_json_delta",
|
|
"partial_json": delta
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.function_call_arguments.done → content_block_stop
|
|
// ================================================
|
|
"response.function_call_arguments.done" => {
|
|
let item_id = data.get("item_id").and_then(|v| v.as_str());
|
|
let index = if let Some(id) = item_id {
|
|
tool_index_by_item_id.get(id).copied()
|
|
} else {
|
|
None
|
|
}
|
|
.or_else(|| {
|
|
tool_item_key_from_event(&data)
|
|
.and_then(|k| index_by_key.get(&k).copied())
|
|
})
|
|
.or(last_tool_index);
|
|
if let Some(index) = index {
|
|
if !open_indices.remove(&index) {
|
|
continue;
|
|
}
|
|
let event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
if let Some(item_id) = item_id {
|
|
tool_index_by_item_id.remove(item_id);
|
|
}
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.refusal.done → content_block_stop
|
|
// ================================================
|
|
"response.refusal.done" => {
|
|
let index = current_text_index.take().or_else(|| {
|
|
let key = content_part_key(&data);
|
|
if let Some(k) = key {
|
|
index_by_key.get(&k).copied()
|
|
} else {
|
|
fallback_open_index
|
|
}
|
|
});
|
|
if let Some(index) = index {
|
|
if !open_indices.remove(&index) {
|
|
continue;
|
|
}
|
|
let event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.reasoning.delta → content_block_delta (thinking_delta)
|
|
// ================================================
|
|
"response.reasoning.delta" => {
|
|
if let Some(delta) = data
|
|
.get("delta")
|
|
.or_else(|| data.get("text"))
|
|
.and_then(|d| d.as_str())
|
|
{
|
|
if let Some(index) = current_text_index.take() {
|
|
if open_indices.remove(&index) {
|
|
let stop_event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let stop_sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(stop_sse));
|
|
}
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
let index = resolve_content_index(
|
|
&data,
|
|
&mut next_content_index,
|
|
&mut index_by_key,
|
|
&mut fallback_open_index,
|
|
);
|
|
|
|
if !open_indices.contains(&index) {
|
|
let start_event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "thinking",
|
|
"thinking": ""
|
|
}
|
|
});
|
|
let start_sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(start_sse));
|
|
open_indices.insert(index);
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {
|
|
"type": "thinking_delta",
|
|
"thinking": delta
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.reasoning.done → content_block_stop
|
|
// ================================================
|
|
"response.reasoning.done" => {
|
|
let key = content_part_key(&data);
|
|
let index = if let Some(k) = key {
|
|
index_by_key.get(&k).copied()
|
|
} else {
|
|
fallback_open_index
|
|
};
|
|
if let Some(index) = index {
|
|
if !open_indices.remove(&index) {
|
|
continue;
|
|
}
|
|
let event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.completed → message_delta + message_stop
|
|
// ================================================
|
|
"response.completed" => {
|
|
let response_obj = response_object_from_event(&data);
|
|
let stop_reason = map_responses_stop_reason(
|
|
response_obj.get("status").and_then(|s| s.as_str()),
|
|
has_tool_use,
|
|
response_obj
|
|
.pointer("/incomplete_details/reason")
|
|
.and_then(|r| r.as_str()),
|
|
);
|
|
|
|
// Best effort: close any dangling blocks before message_delta/message_stop.
|
|
if !open_indices.is_empty() {
|
|
let mut remaining: Vec<u32> = open_indices.iter().copied().collect();
|
|
remaining.sort_unstable();
|
|
for index in remaining {
|
|
let stop_event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let stop_sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(stop_sse));
|
|
open_indices.remove(&index);
|
|
}
|
|
}
|
|
fallback_open_index = None;
|
|
|
|
let usage_json = response_obj.get("usage").map(|u| {
|
|
build_anthropic_usage_from_responses(Some(u))
|
|
});
|
|
|
|
// Emit message_delta (with usage + stop_reason)
|
|
let delta_event = json!({
|
|
"type": "message_delta",
|
|
"delta": {
|
|
"stop_reason": stop_reason,
|
|
"stop_sequence": null
|
|
},
|
|
"usage": usage_json
|
|
});
|
|
let sse = format!("event: message_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&delta_event).unwrap_or_default());
|
|
log::debug!("[Claude/Responses] >>> Anthropic SSE: message_delta");
|
|
yield Ok(Bytes::from(sse));
|
|
|
|
// Emit message_stop
|
|
let stop_event = json!({"type": "message_stop"});
|
|
let stop_sse = format!("event: message_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
log::debug!("[Claude/Responses] >>> Anthropic SSE: message_stop");
|
|
yield Ok(Bytes::from(stop_sse));
|
|
}
|
|
|
|
// Lifecycle events that don't need Anthropic counterparts.
|
|
// Listed explicitly so new events trigger a match-completeness review.
|
|
"response.output_text.done" => {
|
|
if let Some(index) = current_text_index.take() {
|
|
if open_indices.remove(&index) {
|
|
let stop_event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let stop_sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(stop_sse));
|
|
}
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
}
|
|
"response.output_item.done"
|
|
| "response.in_progress" => {}
|
|
|
|
// Any other unknown/future events — silently skip.
|
|
_ => {}
|
|
}
|
|
}
|
|
}
|
|
Err(e) => {
|
|
log::error!("Responses stream error: {e}");
|
|
let error_event = json!({
|
|
"type": "error",
|
|
"error": {
|
|
"type": "stream_error",
|
|
"message": format!("Stream error: {e}")
|
|
}
|
|
});
|
|
let sse = format!("event: error\ndata: {}\n\n",
|
|
serde_json::to_string(&error_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use futures::stream;
|
|
use futures::StreamExt;
|
|
use std::collections::HashMap;
|
|
|
|
#[test]
|
|
fn test_map_responses_stop_reason_tool_use() {
|
|
assert_eq!(
|
|
map_responses_stop_reason(Some("completed"), true, None),
|
|
Some("tool_use")
|
|
);
|
|
assert_eq!(
|
|
map_responses_stop_reason(Some("completed"), false, None),
|
|
Some("end_turn")
|
|
);
|
|
assert_eq!(
|
|
map_responses_stop_reason(Some("incomplete"), false, Some("max_output_tokens")),
|
|
Some("max_tokens")
|
|
);
|
|
assert_eq!(
|
|
map_responses_stop_reason(Some("incomplete"), false, Some("content_filter")),
|
|
Some("end_turn")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_response_object_from_event_with_wrapper() {
|
|
let data = json!({
|
|
"type": "response.created",
|
|
"response": {
|
|
"id": "resp_1",
|
|
"model": "gpt-4o"
|
|
}
|
|
});
|
|
let obj = response_object_from_event(&data);
|
|
assert_eq!(obj["id"], "resp_1");
|
|
assert_eq!(obj["model"], "gpt-4o");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_streaming_conversion_with_wrapped_response_events() {
|
|
let input = concat!(
|
|
"event: response.created\n",
|
|
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-4o\",\"usage\":{\"input_tokens\":12,\"output_tokens\":0}}}\n\n",
|
|
"event: response.output_item.added\n",
|
|
"data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"get_weather\"}}\n\n",
|
|
"event: response.function_call_arguments.delta\n",
|
|
"data: {\"type\":\"response.function_call_arguments.delta\",\"delta\":\"{\\\"city\\\":\\\"Tokyo\\\"}\"}\n\n",
|
|
"event: response.function_call_arguments.done\n",
|
|
"data: {\"type\":\"response.function_call_arguments.done\"}\n\n",
|
|
"event: response.completed\n",
|
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":12,\"output_tokens\":3}}}\n\n"
|
|
);
|
|
|
|
let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(
|
|
input.as_bytes().to_vec(),
|
|
))]);
|
|
let converted = create_anthropic_sse_stream_from_responses(upstream);
|
|
let chunks: Vec<_> = converted.collect().await;
|
|
|
|
let merged = chunks
|
|
.into_iter()
|
|
.map(|c| String::from_utf8_lossy(c.unwrap().as_ref()).to_string())
|
|
.collect::<String>();
|
|
|
|
assert!(merged.contains("\"type\":\"message_start\""));
|
|
assert!(merged.contains("\"id\":\"resp_1\""));
|
|
assert!(merged.contains("\"model\":\"gpt-4o\""));
|
|
assert!(merged.contains("\"type\":\"tool_use\""));
|
|
assert!(merged.contains("\"name\":\"get_weather\""));
|
|
assert!(merged.contains("\"type\":\"input_json_delta\""));
|
|
assert!(merged.contains("\"stop_reason\":\"tool_use\""));
|
|
assert!(merged.contains("\"input_tokens\":12"));
|
|
assert!(merged.contains("\"output_tokens\":3"));
|
|
assert!(merged.contains("\"type\":\"message_stop\""));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_streaming_conversion_interleaved_tool_deltas_by_item_id() {
|
|
let input = concat!(
|
|
"event: response.created\n",
|
|
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_2\",\"model\":\"gpt-4o\"}}\n\n",
|
|
"event: response.output_item.added\n",
|
|
"data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"first_tool\"}}\n\n",
|
|
"event: response.output_item.added\n",
|
|
"data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"fc_2\",\"type\":\"function_call\",\"call_id\":\"call_2\",\"name\":\"second_tool\"}}\n\n",
|
|
"event: response.function_call_arguments.delta\n",
|
|
"data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_2\",\"delta\":\"{\\\"b\\\":2}\"}\n\n",
|
|
"event: response.function_call_arguments.delta\n",
|
|
"data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\"{\\\"a\\\":1}\"}\n\n",
|
|
"event: response.function_call_arguments.done\n",
|
|
"data: {\"type\":\"response.function_call_arguments.done\",\"item_id\":\"fc_1\"}\n\n",
|
|
"event: response.function_call_arguments.done\n",
|
|
"data: {\"type\":\"response.function_call_arguments.done\",\"item_id\":\"fc_2\"}\n\n",
|
|
"event: response.completed\n",
|
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":8,\"output_tokens\":4}}}\n\n"
|
|
);
|
|
|
|
let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(
|
|
input.as_bytes().to_vec(),
|
|
))]);
|
|
let converted = create_anthropic_sse_stream_from_responses(upstream);
|
|
let chunks: Vec<_> = converted.collect().await;
|
|
let merged = chunks
|
|
.into_iter()
|
|
.map(|c| String::from_utf8_lossy(c.unwrap().as_ref()).to_string())
|
|
.collect::<String>();
|
|
|
|
let events: Vec<Value> = merged
|
|
.split("\n\n")
|
|
.filter_map(|block| {
|
|
let data = block
|
|
.lines()
|
|
.find_map(|line| strip_sse_field(line, "data"))?;
|
|
serde_json::from_str::<Value>(data).ok()
|
|
})
|
|
.collect();
|
|
|
|
let mut tool_index_by_call: HashMap<String, u64> = HashMap::new();
|
|
for event in &events {
|
|
if event.get("type").and_then(|v| v.as_str()) == Some("content_block_start") {
|
|
let cb = event.get("content_block");
|
|
if cb.and_then(|v| v.get("type")).and_then(|v| v.as_str()) == Some("tool_use") {
|
|
if let (Some(call_id), Some(index)) = (
|
|
cb.and_then(|v| v.get("id")).and_then(|v| v.as_str()),
|
|
event.get("index").and_then(|v| v.as_u64()),
|
|
) {
|
|
tool_index_by_call.insert(call_id.to_string(), index);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let delta_indices: Vec<u64> = events
|
|
.iter()
|
|
.filter(|event| {
|
|
event.get("type").and_then(|v| v.as_str()) == Some("content_block_delta")
|
|
&& event.pointer("/delta/type").and_then(|v| v.as_str())
|
|
== Some("input_json_delta")
|
|
})
|
|
.filter_map(|event| event.get("index").and_then(|v| v.as_u64()))
|
|
.collect();
|
|
|
|
assert_eq!(delta_indices.len(), 2);
|
|
assert_eq!(delta_indices[0], *tool_index_by_call.get("call_2").unwrap());
|
|
assert_eq!(delta_indices[1], *tool_index_by_call.get("call_1").unwrap());
|
|
assert_ne!(
|
|
tool_index_by_call.get("call_1"),
|
|
tool_index_by_call.get("call_2")
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_streaming_reasoning_delta_emits_thinking_blocks() {
|
|
let input = concat!(
|
|
"event: response.created\n",
|
|
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_r\",\"model\":\"o3\",\"usage\":{\"input_tokens\":5,\"output_tokens\":0}}}\n\n",
|
|
"event: response.reasoning.delta\n",
|
|
"data: {\"type\":\"response.reasoning.delta\",\"delta\":\"Let me think...\"}\n\n",
|
|
"event: response.reasoning.done\n",
|
|
"data: {\"type\":\"response.reasoning.done\"}\n\n",
|
|
"event: response.content_part.added\n",
|
|
"data: {\"type\":\"response.content_part.added\",\"part\":{\"type\":\"output_text\",\"text\":\"\"},\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.output_text.delta\n",
|
|
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"42\",\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.content_part.done\n",
|
|
"data: {\"type\":\"response.content_part.done\",\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.completed\n",
|
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":10}}}\n\n"
|
|
);
|
|
|
|
let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(
|
|
input.as_bytes().to_vec(),
|
|
))]);
|
|
let converted = create_anthropic_sse_stream_from_responses(upstream);
|
|
let chunks: Vec<_> = converted.collect().await;
|
|
let merged = chunks
|
|
.into_iter()
|
|
.map(|c| String::from_utf8_lossy(c.unwrap().as_ref()).to_string())
|
|
.collect::<String>();
|
|
|
|
// Should contain thinking block start, thinking delta, and text content
|
|
assert!(
|
|
merged.contains("\"type\":\"thinking\""),
|
|
"should emit thinking content_block_start"
|
|
);
|
|
assert!(
|
|
merged.contains("\"type\":\"thinking_delta\""),
|
|
"should emit thinking_delta"
|
|
);
|
|
assert!(
|
|
merged.contains("\"thinking\":\"Let me think...\""),
|
|
"should contain thinking text"
|
|
);
|
|
assert!(
|
|
merged.contains("\"type\":\"text_delta\""),
|
|
"should also emit text content"
|
|
);
|
|
assert!(
|
|
merged.contains("\"text\":\"42\""),
|
|
"should contain text delta"
|
|
);
|
|
assert!(merged.contains("\"stop_reason\":\"end_turn\""));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_streaming_text_parts_are_merged_into_one_text_block() {
|
|
let input = concat!(
|
|
"event: response.created\n",
|
|
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_merge\",\"model\":\"gpt-5.4\",\"usage\":{\"input_tokens\":5,\"output_tokens\":0}}}\n\n",
|
|
"event: response.content_part.added\n",
|
|
"data: {\"type\":\"response.content_part.added\",\"part\":{\"type\":\"output_text\",\"text\":\"\"},\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.output_text.delta\n",
|
|
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"你\",\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.content_part.done\n",
|
|
"data: {\"type\":\"response.content_part.done\",\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.content_part.added\n",
|
|
"data: {\"type\":\"response.content_part.added\",\"part\":{\"type\":\"output_text\",\"text\":\"\"},\"output_index\":0,\"content_index\":1}\n\n",
|
|
"event: response.output_text.delta\n",
|
|
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"好\",\"output_index\":0,\"content_index\":1}\n\n",
|
|
"event: response.content_part.done\n",
|
|
"data: {\"type\":\"response.content_part.done\",\"output_index\":0,\"content_index\":1}\n\n",
|
|
"event: response.output_text.done\n",
|
|
"data: {\"type\":\"response.output_text.done\",\"output_index\":0,\"content_index\":1}\n\n",
|
|
"event: response.completed\n",
|
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n"
|
|
);
|
|
|
|
let upstream = stream::iter(vec![Ok(Bytes::from(input.as_bytes().to_vec()))]);
|
|
let converted = create_anthropic_sse_stream_from_responses(upstream);
|
|
let chunks: Vec<_> = converted.collect().await;
|
|
let events: Vec<Value> = chunks
|
|
.into_iter()
|
|
.flat_map(|chunk| {
|
|
let bytes = chunk.unwrap();
|
|
let text = String::from_utf8_lossy(bytes.as_ref()).to_string();
|
|
text.split("\n\n")
|
|
.filter_map(|block| {
|
|
block.lines().find_map(|line| {
|
|
strip_sse_field(line, "data")
|
|
.and_then(|payload| serde_json::from_str::<Value>(payload).ok())
|
|
})
|
|
})
|
|
.collect::<Vec<_>>()
|
|
})
|
|
.collect();
|
|
|
|
let text_starts = events
|
|
.iter()
|
|
.filter(|event| {
|
|
event.get("type").and_then(|v| v.as_str()) == Some("content_block_start")
|
|
&& event
|
|
.pointer("/content_block/type")
|
|
.and_then(|v| v.as_str())
|
|
== Some("text")
|
|
})
|
|
.count();
|
|
let text_stops = events
|
|
.iter()
|
|
.filter(|event| {
|
|
event.get("type").and_then(|v| v.as_str()) == Some("content_block_stop")
|
|
})
|
|
.count();
|
|
let text_deltas: Vec<String> = events
|
|
.iter()
|
|
.filter(|event| {
|
|
event.get("type").and_then(|v| v.as_str()) == Some("content_block_delta")
|
|
&& event.pointer("/delta/type").and_then(|v| v.as_str()) == Some("text_delta")
|
|
})
|
|
.filter_map(|event| {
|
|
event
|
|
.pointer("/delta/text")
|
|
.and_then(|v| v.as_str())
|
|
.map(ToString::to_string)
|
|
})
|
|
.collect();
|
|
|
|
assert_eq!(text_starts, 1);
|
|
assert_eq!(text_stops, 1);
|
|
assert_eq!(text_deltas, vec!["你".to_string(), "好".to_string()]);
|
|
}
|
|
}
|