From 50059c4147cf0053d04c73e9b043ac97b3532b9c Mon Sep 17 00:00:00 2001 From: lishuceo Date: Wed, 10 Jun 2026 23:14:51 +0800 Subject: [PATCH 1/2] fix(proxy): emit spec-complete message_start (content/stop_reason/stop_sequence) The Anthropic streaming spec requires the message_start snapshot to carry "content": [], "stop_reason": null, "stop_sequence": null. Omitting "content" crashes the official @anthropic-ai/sdk stream accumulator (snapshot.content.push -> 'Cannot read properties of undefined') on every translated streamed response, forcing clients into per-turn non-streaming fallback (double billing) and intermittently surfacing as a silent empty end-of-turn that terminates agent sessions mid-task. Fixed in all 6 message_start construction sites: openai_chat (streaming.rs), openai_responses (streaming_responses.rs x3, incl. the tool-call fallback path), gemini (streaming_gemini.rs x2). Adds a regression test per transform path asserting message_start carries the spec-required fields. Co-Authored-By: Claude Opus 4.8 (1M context) --- src-tauri/src/proxy/providers/streaming.rs | 30 +++++++++++++ .../src/proxy/providers/streaming_gemini.rs | 33 ++++++++++++++ .../proxy/providers/streaming_responses.rs | 43 +++++++++++++++++++ 3 files changed, 106 insertions(+) diff --git a/src-tauri/src/proxy/providers/streaming.rs b/src-tauri/src/proxy/providers/streaming.rs index aa1e8c1f..563bdc2f 100644 --- a/src-tauri/src/proxy/providers/streaming.rs +++ b/src-tauri/src/proxy/providers/streaming.rs @@ -166,6 +166,15 @@ pub fn create_anthropic_sse_stream( "id": message_id.clone().unwrap_or_default(), "type": "message", "role": "assistant", + // Anthropic SSE spec requires these fields in the + // message snapshot; the official SDK's stream + // accumulator does `snapshot.content.push(...)` and + // crashes ("Cannot read properties of undefined") + // when `content` is missing, forcing clients into + // non-streaming fallback (or silent stream loss). + "content": [], + "stop_reason": serde_json::Value::Null, + "stop_sequence": serde_json::Value::Null, "model": current_model.clone().unwrap_or_default(), "usage": start_usage } @@ -819,6 +828,27 @@ mod tests { ); } + #[tokio::test] + async fn streaming_message_start_is_spec_complete() { + // message_start must carry content/stop_reason/stop_sequence or the + // official Anthropic SDK stream accumulator crashes on snapshot.content. + let input = concat!( + "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n", + "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1}}\n\n", + "data: [DONE]\n\n" + ); + + let events = collect_events(input).await; + let message = &events + .iter() + .find(|event| event["type"] == "message_start") + .expect("message_start event")["message"]; + + assert_eq!(message["content"], serde_json::json!([])); + assert_eq!(message["stop_reason"], serde_json::Value::Null); + assert_eq!(message["stop_sequence"], serde_json::Value::Null); + } + #[tokio::test] async fn streaming_accepts_deepseek_reasoning_content_alias() { let input = concat!( diff --git a/src-tauri/src/proxy/providers/streaming_gemini.rs b/src-tauri/src/proxy/providers/streaming_gemini.rs index d51c29f3..fab54a2d 100644 --- a/src-tauri/src/proxy/providers/streaming_gemini.rs +++ b/src-tauri/src/proxy/providers/streaming_gemini.rs @@ -316,6 +316,11 @@ pub fn create_anthropic_sse_stream_from_gemini(Bytes::from_static( diff --git a/src-tauri/src/proxy/providers/streaming_responses.rs b/src-tauri/src/proxy/providers/streaming_responses.rs index bae4755d..e7f4c882 100644 --- a/src-tauri/src/proxy/providers/streaming_responses.rs +++ b/src-tauri/src/proxy/providers/streaming_responses.rs @@ -170,6 +170,11 @@ pub fn create_anthropic_sse_stream_from_responses( "id": message_id.clone().unwrap_or_default(), "type": "message", "role": "assistant", + // Spec-required fields; missing `content` crashes the + // official Anthropic SDK stream accumulator. + "content": [], + "stop_reason": serde_json::Value::Null, + "stop_sequence": serde_json::Value::Null, "model": current_model.clone().unwrap_or_default(), "usage": build_anthropic_usage_from_responses(response_obj.get("usage")) } @@ -185,6 +190,11 @@ pub fn create_anthropic_sse_stream_from_responses( "id": message_id.clone().unwrap_or_default(), "type": "message", "role": "assistant", + // Spec-required fields; missing `content` crashes the + // official Anthropic SDK stream accumulator. + "content": [], + "stop_reason": serde_json::Value::Null, + "stop_sequence": serde_json::Value::Null, "model": current_model.clone().unwrap_or_default(), "usage": { "input_tokens": 0, "output_tokens": 0 } } @@ -301,6 +311,11 @@ pub fn create_anthropic_sse_stream_from_responses( "id": message_id.clone().unwrap_or_default(), "type": "message", "role": "assistant", + // Spec-required fields; missing `content` crashes the + // official Anthropic SDK stream accumulator. + "content": [], + "stop_reason": serde_json::Value::Null, + "stop_sequence": serde_json::Value::Null, "model": current_model.clone().unwrap_or_default(), "usage": { "input_tokens": 0, "output_tokens": 0 } } @@ -748,6 +763,34 @@ data: {\"response\":{\"status\":\"completed\"}}\r\n\ assert!(merged.contains("\"stop_reason\":\"tool_use\"")); } + #[tokio::test] + async fn tool_first_message_start_is_spec_complete() { + // A response that opens with a function_call (no preceding text) emits + // message_start from the tool path; it must still carry the spec-required + // fields or the official Anthropic SDK stream accumulator crashes. + let input = concat!( + "event: response.created\n", + "data: {\"response\":{\"id\":\"resp_tool\",\"model\":\"gpt-5.4\"}}\n\n", + "event: response.output_item.added\n", + "data: {\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"get_weather\"}}\n\n", + "event: response.function_call_arguments.done\n", + "data: {\"item_id\":\"fc_1\",\"arguments\":\"{}\"}\n\n", + "event: response.completed\n", + "data: {\"response\":{\"status\":\"completed\"}}\n\n" + ); + + let (merged, _) = collect_stream(vec![Bytes::from(input)]).await; + let events = parse_anthropic_events(&merged); + let message = &events + .iter() + .find(|event| event["type"] == "message_start") + .expect("message_start event")["message"]; + + assert_eq!(message["content"], serde_json::json!([])); + assert_eq!(message["stop_reason"], Value::Null); + assert_eq!(message["stop_sequence"], Value::Null); + } + #[tokio::test] async fn non_read_tool_empty_delta_still_uses_done_arguments() { let input = concat!( From 5f3d6a64bfd4ebbaacece79d3d10ce56ae8b7b20 Mon Sep 17 00:00:00 2001 From: lishuceo Date: Wed, 1 Jul 2026 15:04:06 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(proxy):=20map=20streaming=20usage=20on?= =?UTF-8?q?=20Claude=E2=86=92OpenAI=20chat=20path?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Token/cost/cache accounting was zeroed for streaming requests routed through api_format="openai_chat" (e.g. gpt-5.5 via a LiteLLM gateway). Two gaps, both mirroring handling the codex chat bridge already had: - Request side (transform.rs): the Anthropic→OpenAI chat translation copied `stream` but never injected `stream_options.include_usage`, so OpenAI-compatible upstreams never emitted a usage chunk while streaming. - Response side (streaming.rs): the final usage arrives in a trailing `choices:[]` chunk *after* the finish_reason chunk, but the `choices.first()` guard dropped it, leaving message_start/message_delta at {input_tokens:0, output_tokens:0}. Cache usage from every chunk before the guard and flush a single deferred message_delta (carrying the final usage) at [DONE]/stream end. Adds unit tests for both sides. Co-Authored-By: Claude Opus 4.8 (1M context) --- src-tauri/src/proxy/providers/streaming.rs | 150 ++++++++++++++++----- src-tauri/src/proxy/providers/transform.rs | 50 +++++++ 2 files changed, 168 insertions(+), 32 deletions(-) diff --git a/src-tauri/src/proxy/providers/streaming.rs b/src-tauri/src/proxy/providers/streaming.rs index 563bdc2f..d062a9bf 100644 --- a/src-tauri/src/proxy/providers/streaming.rs +++ b/src-tauri/src/proxy/providers/streaming.rs @@ -53,7 +53,7 @@ struct DeltaFunction { arguments: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Clone, Deserialize)] struct Usage { #[serde(default)] prompt_tokens: u32, @@ -67,7 +67,7 @@ struct Usage { cache_creation_input_tokens: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Clone, Deserialize)] struct PromptTokensDetails { #[serde(default)] cached_tokens: u32, @@ -98,6 +98,11 @@ pub fn create_anthropic_sse_stream( let mut open_tool_block_indices: HashSet = HashSet::new(); let mut legacy_function_name: Option = None; let mut legacy_function_block_index: Option = None; + // Final usage and stop reason are deferred: OpenAI-compatible upstreams emit + // them in separate trailing chunks (the usage chunk has empty `choices`), so + // we cache them here and flush a single message_delta at [DONE]/stream end. + let mut latest_usage: Option = None; + let mut pending_stop_reason: Option = None; tokio::pin!(stream); @@ -120,6 +125,17 @@ pub fn create_anthropic_sse_stream( }; if data.trim() == "[DONE]" { + if has_sent_message_start && pending_stop_reason.is_some() { + let event = build_message_delta_event( + pending_stop_reason.as_deref(), + latest_usage.as_ref(), + ); + let sse_data = format!( + "event: message_delta\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default() + ); + yield Ok(Bytes::from(sse_data)); + } let event = json!({"type": "message_stop"}); let sse_data = format!( "event: message_stop\ndata: {}\n\n", @@ -141,6 +157,15 @@ pub fn create_anthropic_sse_stream( current_model = Some(chunk.model.clone()); } + // Cache usage from any chunk (incl. the trailing + // `choices: []` usage chunk that include_usage produces). + // Must happen before the choices.first() guard below, which + // otherwise drops that chunk and zeroes token/cost/cache + // accounting. Mirrors streaming_codex_chat's latest_usage. + if let Some(usage) = &chunk.usage { + latest_usage = Some(usage.clone()); + } + let Some(choice) = chunk.choices.first() else { continue; }; @@ -597,36 +622,13 @@ pub fn create_anthropic_sse_stream( open_tool_block_indices.clear(); } - let usage_json = if let Some(usage) = chunk.usage.as_ref() { - let mut usage_json = json!({ - "input_tokens": usage.prompt_tokens, - "output_tokens": usage.completion_tokens - }); - if let Some(cached) = extract_cache_read_tokens(usage) { - usage_json["cache_read_input_tokens"] = json!(cached); - } - if let Some(created) = usage.cache_creation_input_tokens { - usage_json["cache_creation_input_tokens"] = json!(created); - } - usage_json - } else { - json!({ - "output_tokens": 0 - }) - }; - let event = json!({ - "type": "message_delta", - "delta": { - "stop_reason": map_stop_reason(Some(finish_reason)), - "stop_sequence": null - }, - "usage": usage_json - }); - let sse_data = format!( - "event: message_delta\ndata: {}\n\n", - serde_json::to_string(&event).unwrap_or_default() - ); - yield Ok(Bytes::from(sse_data)); + // Defer the message_delta: the final usage often arrives + // in a *later* chunk than finish_reason (OpenAI-compatible + // streams send `choices:[{finish_reason}]` then a separate + // `choices:[]` usage chunk). Emitting now would lock in + // output_tokens=0; instead record the stop reason and flush + // one message_delta with the cached usage at stream end. + pending_stop_reason = Some(finish_reason.clone()); } } } @@ -650,10 +652,53 @@ pub fn create_anthropic_sse_stream( } } + // Flush the deferred message_delta if the upstream closed without a [DONE] + // marker, so the final usage still reaches the client. + if has_sent_message_start && pending_stop_reason.is_some() { + let event = build_message_delta_event( + pending_stop_reason.as_deref(), + latest_usage.as_ref(), + ); + let sse_data = format!( + "event: message_delta\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default() + ); + yield Ok(Bytes::from(sse_data)); + } + stream_completion.record_success(); } } +fn build_message_delta_event( + stop_reason: Option<&str>, + usage: Option<&Usage>, +) -> serde_json::Value { + let usage_json = if let Some(usage) = usage { + let mut usage_json = json!({ + "input_tokens": usage.prompt_tokens, + "output_tokens": usage.completion_tokens + }); + if let Some(cached) = extract_cache_read_tokens(usage) { + usage_json["cache_read_input_tokens"] = json!(cached); + } + if let Some(created) = usage.cache_creation_input_tokens { + usage_json["cache_creation_input_tokens"] = json!(created); + } + usage_json + } else { + json!({ "output_tokens": 0 }) + }; + json!({ + "type": "message_delta", + "delta": { + "stop_reason": map_stop_reason(stop_reason), + "stop_sequence": null + }, + "usage": usage_json + }) +} + fn map_stop_reason(finish_reason: Option<&str>) -> Option { finish_reason.map(|reason| { match reason { @@ -828,6 +873,47 @@ mod tests { ); } + #[tokio::test] + async fn streaming_usage_from_trailing_empty_choices_chunk_is_mapped() { + // OpenAI-compatible upstreams (LiteLLM, kimi, MiniMax, ...) send the final + // usage in a separate `choices: []` chunk *after* the finish_reason chunk + // when stream_options.include_usage=true. The choices.first() guard used to + // drop it, leaving message_delta usage at 0, zeroing token/cost accounting. + let input = concat!( + "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-5.5\",\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n", + "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-5.5\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n", + "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-5.5\",\"choices\":[],\"usage\":{\"prompt_tokens\":12,\"completion_tokens\":9,\"total_tokens\":21}}\n\n", + "data: [DONE]\n\n" + ); + + let events = collect_events(input).await; + let message_delta = events + .iter() + .find(|event| event["type"] == "message_delta") + .expect("message_delta event"); + + assert_eq!(message_delta["delta"]["stop_reason"], "end_turn"); + assert_eq!(message_delta["usage"]["input_tokens"], 12); + assert_eq!(message_delta["usage"]["output_tokens"], 9); + + // Exactly one message_delta is emitted, and it precedes message_stop. + let delta_count = events + .iter() + .filter(|event| event["type"] == "message_delta") + .count(); + assert_eq!(delta_count, 1); + let names: Vec<&str> = events + .iter() + .filter_map(|event| event["type"].as_str()) + .collect(); + let delta_pos = names.iter().position(|n| *n == "message_delta"); + let stop_pos = names.iter().position(|n| *n == "message_stop"); + assert!( + delta_pos < stop_pos, + "message_delta must precede message_stop" + ); + } + #[tokio::test] async fn streaming_message_start_is_spec_complete() { // message_start must carry content/stop_reason/stop_sequence or the diff --git a/src-tauri/src/proxy/providers/transform.rs b/src-tauri/src/proxy/providers/transform.rs index 3b06d158..b2894a51 100644 --- a/src-tauri/src/proxy/providers/transform.rs +++ b/src-tauri/src/proxy/providers/transform.rs @@ -162,6 +162,28 @@ pub fn anthropic_to_openai_with_reasoning_content( result["stream"] = v.clone(); } + // OpenAI-compatible upstreams do not emit usage in a streaming response + // unless stream_options.include_usage is set; without it the final usage + // chunk never arrives, so streaming.rs cannot populate message_start / + // message_delta and every input/output/cache token is reported as 0. + // Mirrors the injection already done on the Codex chat bridge path. + let is_stream = result + .get("stream") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + if is_stream { + match result.get_mut("stream_options") { + // Preserve any other stream_options the client passed through; + // only add include_usage. + Some(Value::Object(opts)) => { + opts.insert("include_usage".to_string(), json!(true)); + } + _ => { + result["stream_options"] = json!({ "include_usage": true }); + } + } + } + if supports_reasoning_effort(model) { if let Some(effort) = resolve_reasoning_effort(&body) { result["reasoning_effort"] = json!(effort); @@ -673,6 +695,34 @@ mod tests { assert_eq!(result["prompt_cache_key"], "provider-123"); } + #[test] + fn anthropic_to_openai_stream_injects_include_usage() { + let input = json!({ + "model": "gpt-5.5", + "max_tokens": 1024, + "stream": true, + "messages": [{"role": "user", "content": "Hello"}] + }); + + let result = anthropic_to_openai(input, None).unwrap(); + + assert_eq!(result["stream"], true); + assert_eq!(result["stream_options"]["include_usage"], true); + } + + #[test] + fn anthropic_to_openai_non_stream_omits_stream_options() { + let input = json!({ + "model": "gpt-5.5", + "max_tokens": 1024, + "messages": [{"role": "user", "content": "Hello"}] + }); + + let result = anthropic_to_openai(input, None).unwrap(); + + assert!(result.get("stream_options").is_none()); + } + #[test] fn anthropic_to_openai_preserves_system_cache_control() { let input = json!({