diff --git a/crates/tinyinference-llm/src/providers/openai/convert.rs b/crates/tinyinference-llm/src/providers/openai/convert.rs index ad29e9d0..8eeba275 100644 --- a/crates/tinyinference-llm/src/providers/openai/convert.rs +++ b/crates/tinyinference-llm/src/providers/openai/convert.rs @@ -408,11 +408,16 @@ pub(super) fn parse_chat_response( } } + // Thinking parts of a list `content` follow the side-channel reasoning. + let (text, content_reasoning) = choice + .message + .content + .map(split_response_content) + .unwrap_or_default(); + reasoning.push_str(&content_reasoning); + // Inline `` extraction on the visible content, when enabled. - let visible = match ( - choice.message.content.filter(|t| !t.is_empty()), - reasoning_tags, - ) { + let visible = match (Some(text).filter(|t| !t.is_empty()), reasoning_tags) { (Some(text), Some(config)) => { let (visible, inline) = extract_reasoning(config, &text); if !inline.is_empty() { @@ -644,6 +649,43 @@ pub(super) fn reasoning_value_text(value: Value) -> Option { } } +/// Splits a response or delta `content` into `(visible text, reasoning)`. +/// +/// A string is all visible text. In a list of parts, `text` parts are visible +/// text, concatenated in order; `thinking` / `reasoning` parts are reasoning, +/// read from their `thinking` field (a string or a list of text parts) or else +/// their `text` field, and normalized like `reasoning_content`. Parts of any +/// other type, and anything that is not an object, are skipped. +pub(super) fn split_response_content(content: ResponseContentWire) -> (String, String) { + let parts = match content { + ResponseContentWire::Text(text) => return (text, String::new()), + ResponseContentWire::Parts(parts) => parts, + }; + let mut text = String::new(); + let mut reasoning = String::new(); + for part in &parts { + match part.get("type").and_then(Value::as_str) { + Some("text") => { + if let Some(fragment) = part.get("text").and_then(Value::as_str) { + text.push_str(fragment); + } + } + Some("thinking" | "reasoning") => { + if let Some(fragment) = part + .get("thinking") + .or_else(|| part.get("text")) + .cloned() + .and_then(reasoning_value_text) + { + reasoning.push_str(&fragment); + } + } + _ => {} + } + } + (text, reasoning) +} + /// Extracts the reasoning/thinking text from a streamed delta, accepting the /// common OpenAI-compatible aliases. pub(super) fn delta_reasoning_text(delta: &mut ChunkDeltaWire) -> String { diff --git a/crates/tinyinference-llm/src/providers/openai/mod_tests.rs b/crates/tinyinference-llm/src/providers/openai/mod_tests.rs index 8f90e7a1..a3383226 100644 --- a/crates/tinyinference-llm/src/providers/openai/mod_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/mod_tests.rs @@ -1180,6 +1180,37 @@ fn parse_chat_response_without_config_leaves_inline_tags_in_text() { assert_eq!(response_reasoning(&response), ""); } +/// A thinking model behind an OpenAI-compatible gateway returns the message +/// `content` as a list of typed parts. Text parts are the answer; thinking +/// parts are reasoning, after any side-channel reasoning, and never visible. +#[test] +fn parse_chat_response_reads_list_content_parts() { + let body = json!({ + "id": "chatcmpl-list", + "choices": [{ + "message": { + "role": "assistant", + "reasoning_content": "side ", + "content": [ + { "type": "thinking", "thinking": [{ "type": "text", "text": "plan" }], "closed": true }, + { "type": "text", "text": "{\"reason\":" }, + { "type": "unknown_part", "data": 1 }, + { "type": "text", "text": "\"ok\"}" } + ] + }, + "finish_reason": "stop" + }] + }); + + let response = parse_chat_response(body, None).unwrap(); + assert_eq!(response.text(), "{\"reason\":\"ok\"}"); + assert_eq!(response_reasoning(&response), "side plan"); + assert!(matches!( + response.message.content.as_slice(), + [ContentBlock::Thinking { .. }, ContentBlock::Text(_)] + )); +} + #[tokio::test] async fn sse_stream_reassembles_multibyte_char_split_across_chunks() { // A 4-byte emoji in the content payload, split down the middle across two @@ -3035,7 +3066,10 @@ fn a_null_tool_calls_delta_is_read_as_no_fragments() { })) .expect("a null tool_calls must not fail the chunk"); assert!(chunk.choices[0].delta.tool_calls.is_empty()); - assert_eq!(chunk.choices[0].delta.content.as_deref(), Some("Hi")); + assert_eq!( + chunk.choices[0].delta.content, + Some(ResponseContentWire::Text("Hi".into())) + ); } /// Unwraps a [`ModelStreamItem::BlockStart`], panicking with the actual item diff --git a/crates/tinyinference-llm/src/providers/openai/sse.rs b/crates/tinyinference-llm/src/providers/openai/sse.rs index d3ff5b3d..afc870f1 100644 --- a/crates/tinyinference-llm/src/providers/openai/sse.rs +++ b/crates/tinyinference-llm/src/providers/openai/sse.rs @@ -236,7 +236,17 @@ impl OpenAiStreamAcc { if let Some(reason) = choice.finish_reason { self.finish_reason = Some(reason); } - let reasoning = delta_reasoning_text(&mut choice.delta); + // A list `content` can carry thinking and text parts in one chunk; + // its reasoning streams before its text, after any side-channel + // reasoning, so the block order stays thinking -> text. + let (content, content_reasoning) = choice + .delta + .content + .take() + .map(split_response_content) + .unwrap_or_default(); + let mut reasoning = delta_reasoning_text(&mut choice.delta); + reasoning.push_str(&content_reasoning); if !reasoning.is_empty() { self.reasoning.push_str(&reasoning); self.push_block_delta(BlockRequest::Reasoning, &reasoning, pending); @@ -246,7 +256,7 @@ impl OpenAiStreamAcc { tool_call: None, })); } - if let Some(content) = choice.delta.content.filter(|c| !c.is_empty()) { + if !content.is_empty() { // Retain the raw content so the terminal response can recompute // an authoritative inline-tag split (see `into_response`). self.text.push_str(&content); @@ -710,7 +720,27 @@ impl SseState { self.acc.keep_usage_extensions(usage); } } - if let Ok(chunk) = serde_json::from_value::(value) { + let chunk = match serde_json::from_value::(value) { + Ok(chunk) => Some(chunk), + Err(error) => { + // Still skipped, but never silently: a dropped chunk loses + // whatever text it carried, and a silent drop hides that the + // answer arrived truncated. The payload can hold answer or + // reasoning text, so only DEBUG carries it, truncated. + tracing::warn!( + provider = %self.provider, + model = %self.model, + %error, + "openai sse: skipped a chunk that does not match the chunk schema" + ); + tracing::debug!( + payload = %log_preview(payload), + "openai sse: skipped chunk payload" + ); + None + } + }; + if let Some(chunk) = chunk { // `resolve_slot` grows the slot vector up to the wire index, so an // out-of-range index is rejected before it can size an allocation. if let Some(index) = chunk @@ -805,6 +835,18 @@ impl SseState { } } +/// Characters of an SSE payload kept when one is logged. +const LOG_PREVIEW_CHARS: usize = 512; + +/// The first [`LOG_PREVIEW_CHARS`] characters of `payload` (UTF-8 safe), +/// marked with `...` when cut, so a log line stays bounded. +fn log_preview(payload: &str) -> String { + match payload.char_indices().nth(LOG_PREVIEW_CHARS) { + Some((end, _)) => format!("{}...", &payload[..end]), + None => payload.to_string(), + } +} + /// Advances the SSE [`SseState`] by one item for [`futures::stream::unfold`]. pub(super) async fn sse_next(mut state: SseState) -> Option<(ModelStreamItem, SseState)> { loop { diff --git a/crates/tinyinference-llm/src/providers/openai/sse_tests.rs b/crates/tinyinference-llm/src/providers/openai/sse_tests.rs index 53f97fa9..beae8656 100644 --- a/crates/tinyinference-llm/src/providers/openai/sse_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/sse_tests.rs @@ -182,3 +182,227 @@ async fn huge_tool_call_index_is_rejected() { "an out-of-range tool-call index was accepted" ); } + +/// Concatenates the visible-text and reasoning fragments across every streamed +/// `MessageDelta`, as `(text, reasoning)`. +fn streamed_channels(items: &[ModelStreamItem]) -> (String, String) { + let mut text = String::new(); + let mut reasoning = String::new(); + for item in items { + if let ModelStreamItem::MessageDelta(delta) = item { + text.push_str(&delta.text); + reasoning.push_str(&delta.reasoning); + } + } + (text, reasoning) +} + +/// The text of every `Thinking` block on a response, concatenated. +fn thinking_text(response: &ModelResponse) -> String { + response + .message + .content + .iter() + .filter_map(|block| block.as_thinking().map(|(text, _)| text.to_string())) + .collect() +} + +/// A thinking model behind an OpenAI-compatible gateway (Surplus serving +/// `glm-5.3-flash`) streams `delta.content` as a list of typed parts while it +/// thinks, and the chunk that closes the thinking also carries the first +/// visible words. These are the captured frames. That transition chunk used +/// to fail to deserialize and was silently dropped, so the answer lost its +/// opening ("ALPHA BRAVO CHARLIE" arrived as " CHARLIE"). +#[tokio::test] +async fn sse_stream_keeps_the_answer_opening_carried_with_list_content_thinking() { + let items = stream_items(vec![ + br#"data: {"id":"c1","choices":[{"index":0,"delta":{"role":"assistant","content":[{"type":"thinking","thinking":[{"type":"text","text":" user wants"}],"closed":false}]}}]} + +"#, + br#"data: {"id":"c1","choices":[{"index":0,"delta":{"content":[{"type":"thinking","thinking":[{"type":"text","text":"."}],"closed":true},{"type":"text","text":"ALPHA BRAVO"}]}}]} + +"#, + br#"data: {"id":"c1","choices":[{"index":0,"delta":{"content":" CHARLIE"},"finish_reason":"stop"}]} + +"#, + b"data: [DONE]\n\n", + ]) + .await; + + let (text, reasoning) = streamed_channels(&items); + assert_eq!(text, "ALPHA BRAVO CHARLIE"); + assert_eq!(reasoning, " user wants."); + + // Within the transition chunk the closing thought streams before the + // first visible words, so the block order stays thinking -> text. + let position = |wanted: &dyn Fn(&MessageDelta) -> bool| { + items + .iter() + .position(|item| matches!(item, ModelStreamItem::MessageDelta(d) if wanted(d))) + .expect("delta present") + }; + assert!(position(&|d| d.reasoning == ".") < position(&|d| d.text == "ALPHA BRAVO")); + + let response = items + .into_iter() + .find_map(|item| match item { + ModelStreamItem::Completed(response) => Some(response), + _ => None, + }) + .expect("stream ends with Completed"); + assert_eq!(response.text(), "ALPHA BRAVO CHARLIE"); + assert_eq!(thinking_text(&response), " user wants."); + assert!( + matches!( + response.message.content.as_slice(), + [ContentBlock::Thinking { .. }, ContentBlock::Text(_)] + ), + "{:?}", + response.message.content + ); +} + +/// A chunk whose list content holds only thinking parts feeds the reasoning +/// channel and never the visible text. The `thinking` field may be a plain +/// string as well as a list of text parts. +#[tokio::test] +async fn sse_stream_routes_a_thinking_only_list_content_to_reasoning() { + let items = stream_items(vec![ + br#"data: {"choices":[{"delta":{"content":[{"type":"thinking","thinking":[{"type":"text","text":"weigh "}],"closed":false}]}}]} + +"#, + br#"data: {"choices":[{"delta":{"content":[{"type":"thinking","thinking":"it up","closed":true}]},"finish_reason":"stop"}]} + +"#, + b"data: [DONE]\n\n", + ]) + .await; + + let (text, reasoning) = streamed_channels(&items); + assert_eq!(text, ""); + assert_eq!(reasoning, "weigh it up"); + let response = items + .into_iter() + .find_map(|item| match item { + ModelStreamItem::Completed(response) => Some(response), + _ => None, + }) + .expect("stream ends with Completed"); + assert_eq!(response.text(), ""); + assert_eq!(thinking_text(&response), "weigh it up"); +} + +/// Plain string content, the shape every non-thinking reply uses, streams +/// exactly as before: all visible text, no reasoning. +#[tokio::test] +async fn sse_stream_plain_string_content_is_all_visible_text() { + let items = stream_items(vec![ + br#"data: {"choices":[{"delta":{"content":"ALPHA"}}]} + +"#, + br#"data: {"choices":[{"delta":{"content":" BRAVO"},"finish_reason":"stop"}]} + +"#, + b"data: [DONE]\n\n", + ]) + .await; + + assert_eq!( + streamed_channels(&items), + ("ALPHA BRAVO".to_string(), String::new()) + ); +} + +/// A list mixing text, `reasoning` parts (carrying `text`), and a part type +/// this parser does not know: the text parts concatenate in order, the +/// reasoning part goes to the reasoning channel, and the unknown part is +/// skipped without losing the rest of the chunk. +#[tokio::test] +async fn sse_stream_list_content_skips_unknown_parts_and_keeps_the_rest() { + let response = completed(vec![ + br#"data: {"choices":[{"delta":{"content":[{"type":"text","text":"one "},{"type":"image_url","image_url":{"url":"https://example.invalid/x.png"}},{"type":"reasoning","text":"aside"},{"type":"text","text":"two"}]},"finish_reason":"stop"}]} + +"#, + b"data: [DONE]\n\n", + ]) + .await; + + assert_eq!(response.text(), "one two"); + assert_eq!(thinking_text(&response), "aside"); +} + +/// Formatted log output, captured for assertions on what was warned. +#[derive(Clone, Default)] +struct LogBuffer(std::sync::Arc>>); + +impl LogBuffer { + fn contents(&self) -> String { + String::from_utf8_lossy(&self.0.lock().unwrap()).into_owned() + } +} + +impl std::io::Write for LogBuffer { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for LogBuffer { + type Writer = LogBuffer; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } +} + +/// A chunk that still does not fit the chunk schema is skipped, as before, +/// but no longer silently: it is warned about (provider, model, error), and the +/// rest of the stream still arrives. The payload can hold answer or reasoning +/// text, so it stays off the WARN line; a truncated copy is logged at DEBUG. +#[tokio::test] +async fn sse_stream_warns_about_a_chunk_it_cannot_read() { + let logs = LogBuffer::default(); + let subscriber = tracing_subscriber::fmt() + .with_writer(logs.clone()) + .with_ansi(false) + .with_max_level(tracing::Level::DEBUG) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + + let unreadable: &'static [u8] = format!( + "data: {{\"choices\":[{{\"index\":\"zero\",\"delta\":{{\"content\":\"{}\"}}}}]}}\n\n", + "x".repeat(4_000) + ) + .leak() + .as_bytes(); + let response = completed(vec![ + b"data: {\"choices\":[{\"delta\":{\"content\":\"kept\"}}]}\n\n", + unreadable, + b"data: {\"choices\":[{\"delta\":{\"content\":\" too\"},\"finish_reason\":\"stop\"}]}\n\n", + b"data: [DONE]\n\n", + ]) + .await; + + assert_eq!(response.text(), "kept too"); + let logs = logs.contents(); + let line = |level: &str| { + logs.lines() + .find(|line| line.contains(level) && line.contains("openai sse: skipped")) + .unwrap_or_else(|| panic!("no {level} line: {logs}")) + }; + let warn = line("WARN"); + assert!(warn.contains("model=gpt-4.1-mini"), "{warn}"); + assert!(warn.contains("error="), "{warn}"); + assert!(!warn.contains("xxxx"), "payload on the WARN line: {warn}"); + let debug = line("DEBUG"); + assert!(debug.contains(r#""index":"zero""#), "{debug}"); + assert!( + !debug.contains(&"x".repeat(4_000)), + "payload not truncated: {debug}" + ); +} diff --git a/crates/tinyinference-llm/src/providers/openai/types.rs b/crates/tinyinference-llm/src/providers/openai/types.rs index 36f0d3b2..1be42f88 100644 --- a/crates/tinyinference-llm/src/providers/openai/types.rs +++ b/crates/tinyinference-llm/src/providers/openai/types.rs @@ -106,9 +106,10 @@ pub struct ChunkChoiceWire { /// The incremental `delta` object inside a [`ChunkChoiceWire`]. #[derive(Clone, Debug, Default, Deserialize)] pub struct ChunkDeltaWire { - /// Incremental text fragment, when present. + /// Incremental content fragment, when present: a string, or a list of + /// typed parts from a thinking model (see [`ResponseContentWire`]). #[serde(default)] - pub content: Option, + pub content: Option, /// Incremental reasoning/thinking fragment used by some OpenAI-compatible /// providers (for example DeepSeek-compatible streams). #[serde(default)] @@ -388,9 +389,10 @@ pub struct ChoiceWire { /// The assistant message inside a [`ChoiceWire`]. #[derive(Clone, Debug, Deserialize)] pub struct ResponseMessageWire { - /// Textual content, when present. + /// Message content, when present: a string, or a list of typed parts + /// from a thinking model (see [`ResponseContentWire`]). #[serde(default)] - pub content: Option, + pub content: Option, /// Reasoning/thinking content on non-streaming responses, when a compatible /// provider exposes it. Kept off visible assistant text by the translator. #[serde(default)] @@ -403,6 +405,30 @@ pub struct ResponseMessageWire { pub tool_calls: Vec, } +/// The `content` of a [`ResponseMessageWire`] or a streamed [`ChunkDeltaWire`]. +/// +/// OpenAI sends a plain string. Some OpenAI-compatible gateways serving a +/// thinking model (Surplus with `glm-5.3-flash`, for one) send a list of typed +/// parts instead, and the chunk that closes the thinking also carries the +/// first visible text: +/// +/// ```json +/// [{"type": "thinking", "thinking": [{"type": "text", "text": "."}], "closed": true}, +/// {"type": "text", "text": "ALPHA BRAVO"}] +/// ``` +/// +/// `text` parts are visible text; `thinking` and `reasoning` parts are +/// reasoning and never visible. Parts stay raw JSON so a part of an unknown +/// type or shape is skipped on its own instead of failing the whole chunk. +#[derive(Clone, Debug, PartialEq, Deserialize)] +#[serde(untagged)] +pub enum ResponseContentWire { + /// Plain-string content: all of it visible text. + Text(String), + /// A list of typed content parts. + Parts(Vec), +} + /// Token usage reported by OpenAI. #[derive(Clone, Debug, Default, Deserialize)] pub struct UsageWire { diff --git a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs index 6c5aaf9e..2aee13d5 100644 --- a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs @@ -633,6 +633,49 @@ fn unsupported(parameter: &str) -> String { .to_string() } +/// Over the wire, a non-streaming completion whose `content` is a list of +/// thinking and text parts (as a thinking model behind Surplus returns it) +/// is read rather than failing the call: the answer is the text parts, the +/// thinking parts are reasoning. +#[tokio::test] +async fn a_list_content_completion_is_read_as_reasoning_then_text() { + let body = json!({ + "id": "chatcmpl-list", + "object": "chat.completion", + "model": "glm-5.3-flash", + "choices": [{ + "index": 0, + "message": { + "role": "assistant", + "content": [ + { "type": "thinking", "thinking": [{ "type": "text", "text": "." }], "closed": true }, + { "type": "text", "text": "ALPHA BRAVO CHARLIE" } + ] + }, + "finish_reason": "stop" + }] + }) + .to_string(); + let (base, server) = serve_sequence(vec![(200, body)]); + let model = OpenAiModel::new("k").with_base_url(&base); + + let response = model + .invoke(&(), request("list-content-model", 0.2)) + .await + .unwrap(); + server.join().unwrap(); + + assert_eq!(response.text(), "ALPHA BRAVO CHARLIE"); + assert_eq!( + response + .message + .content + .first() + .and_then(|block| block.as_thinking()), + Some((".", None)) + ); +} + #[tokio::test] async fn a_rejected_parameter_is_dropped_and_the_request_retried_once() { let (base, server) = serve_sequence(vec![