From d2f17c9321b2fb6a335c52d6b4ee244f04237e07 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 11 Nov 2026 22:27:36 +0000 Subject: [PATCH 1/8] test(openai): cover streaming invoke and unary fallback for local models Add wire tests that pin down how a local-openai model fulfils invoke: it streams and collects so a clipped non-streaming reply cannot lose the opening text or reasoning, and it falls back to a single unary request when a gateway refuses streaming. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/providers/openai/wire_tests.rs | 147 ++++++++++++++++++ 1 file changed, 147 insertions(+) diff --git a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs index 2aee13d5..1f0b16dd 100644 --- a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs @@ -676,6 +676,153 @@ async fn a_list_content_completion_is_read_as_reasoning_then_text() { ); } +/// The full answer as a thinking model streams it through Surplus +/// (`glm-5.3-flash`), captured live: list-shaped thinking, a transition chunk +/// carrying the first letters, plain string chunks, then usage. +const THINKING_SSE: &str = concat!( + r#"data: {"id":"c1","model":"glm-5.3-flash","choices":[{"index":0,"delta":{"role":"assistant","content":[{"type":"thinking","thinking":[{"type":"text","text":"Reply verbatim"}],"closed":false}]}}]}"#, + "\n\n", + r#"data: {"id":"c1","model":"glm-5.3-flash","choices":[{"index":0,"delta":{"content":[{"type":"thinking","thinking":[{"type":"text","text":", or formatting."}],"closed":true},{"type":"text","text":"AL"}]}}]}"#, + "\n\n", + r#"data: {"id":"c1","model":"glm-5.3-flash","choices":[{"index":0,"delta":{"content":"PHA BRAVO CHARLIE"}}]}"#, + "\n\n", + r#"data: {"id":"c1","model":"glm-5.3-flash","choices":[{"index":0,"delta":{"content":" DELTA ECHO"},"finish_reason":"stop"}]}"#, + "\n\n", + r#"data: {"id":"c1","choices":[],"usage":{"prompt_tokens":31,"completion_tokens":40,"total_tokens":71}}"#, + "\n\n", + "data: [DONE]\n\n", +); + +/// The same request answered without streaming, as the same sellers return it +/// (captured live): the opening of the answer and all reasoning are gone. +fn clipped_completion_body() -> String { + json!({ + "id": "c2", + "object": "chat.completion", + "model": "glm-5.3-flash", + "choices": [{ + "index": 0, + "message": { "role": "assistant", "content": "VO CHARLIE DELTA ECHO" }, + "finish_reason": "stop" + }], + "usage": { "prompt_tokens": 31, "completion_tokens": 40, "total_tokens": 71 } + }) + .to_string() +} + +/// Serves `connections` requests, answering each by its `stream` flag: +/// `sse` as `text/event-stream` to a streaming request, `unary` as JSON to any +/// other. Returns the raw requests received. +fn serve_by_stream_flag( + connections: usize, + sse: &'static str, + unary: String, +) -> (String, std::thread::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let base = format!("http://{}/v1", listener.local_addr().unwrap()); + let handle = std::thread::spawn(move || { + let mut received = Vec::new(); + for _ in 0..connections { + let (mut sock, _) = listener.accept().unwrap(); + let raw = String::from_utf8_lossy(&read_request(&mut sock)).into_owned(); + let (content_type, body) = if request_body(&raw)["stream"] == json!(true) { + ("text/event-stream", sse.to_string()) + } else { + ("application/json", unary.clone()) + }; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: {content_type}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + let _ = sock.write_all(response.as_bytes()); + received.push(raw); + } + received + }); + (base, handle) +} + +fn local_runtime_model(provider: &str, base: &str) -> Arc> { + build_local_runtime_chat_model( + provider, + base, + "k", + AuthStyle::Bearer, + "glm-5.3-flash", + &[], + None, + None, + ) +} + +/// Some OpenAI-compatible sellers clip a non-streaming reply (the opening of +/// the answer and all reasoning are lost) while the same reply streams whole. +/// A `local-openai` model therefore fulfils `invoke` by streaming and +/// collecting, and returns exactly what the stream carried: full text, +/// reasoning, usage, finish reason, and the chat-completions origin. +#[tokio::test] +async fn local_openai_invoke_streams_and_returns_the_whole_answer() { + let (base, server) = serve_by_stream_flag(1, THINKING_SSE, clipped_completion_body()); + let model = local_runtime_model("local-openai", &base); + + let response = model + .invoke(&(), ModelRequest::new(vec![Message::user("say it")])) + .await + .unwrap(); + let raw = server.join().unwrap(); + + assert_eq!(request_body(&raw[0])["stream"], json!(true)); + assert_eq!(response.text(), "ALPHA BRAVO CHARLIE DELTA ECHO"); + assert_eq!( + response.message.content.first().and_then(|block| block.as_thinking()), + Some(("Reply verbatim, or formatting.", None)) + ); + assert_eq!(response.finish_reason.as_deref(), Some("stop")); + assert_eq!(response.usage.map(|usage| usage.total_tokens), Some(71)); + let origin = response.message.origin.expect("origin stamped"); + assert_eq!(origin.api, "chat_completions"); + assert_eq!(origin.model, "glm-5.3-flash"); +} + +/// A gateway that refuses `stream: true` still answers: the invoke falls back +/// to one non-streaming request. +#[tokio::test] +async fn local_openai_invoke_falls_back_when_streaming_is_refused() { + let refused = json!({ + "error": { "message": "streaming is not supported", "type": "invalid_request_error" } + }) + .to_string(); + let (base, server) = serve_sequence(vec![(400, refused), (200, completion_body())]); + let model = local_runtime_model("local-openai", &base); + + let response = model + .invoke(&(), ModelRequest::new(vec![Message::user("say it")])) + .await + .unwrap(); + let raw = server.join().unwrap(); + + assert_eq!(response.text(), "Hello!"); + assert_eq!(raw.len(), 2, "one refused stream, one unary retry"); + assert_eq!(request_body(&raw[0])["stream"], json!(true)); + assert_ne!(request_body(&raw[1])["stream"], json!(true)); +} + +/// Other local runtimes keep the plain non-streaming invoke. +#[tokio::test] +async fn other_local_runtimes_invoke_without_streaming() { + let (base, server) = serve_sequence(vec![(200, completion_body())]); + let model = local_runtime_model("ollama", &base); + + let response = model + .invoke(&(), ModelRequest::new(vec![Message::user("say it")])) + .await + .unwrap(); + let raw = server.join().unwrap(); + + assert_eq!(response.text(), "Hello!"); + assert_ne!(request_body(&raw[0])["stream"], json!(true)); +} + #[tokio::test] async fn a_rejected_parameter_is_dropped_and_the_request_retried_once() { let (base, server) = serve_sequence(vec![ From f02712a41a821fd6ada66a468b7a2f1fbea4e241 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 11 Nov 2026 22:28:29 +0000 Subject: [PATCH 2/8] feat(openai): stream invoke for local-openai models Local OpenAI-compatible endpoints can clip non-streaming replies, losing the opening text and reasoning tokens, so invoke now streams and collects the response when the provider is local-openai. A gateway that rejects streaming is retried once without it, and the unary timeout still bounds the call. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/providers/openai/config.rs | 10 +++- .../src/providers/openai/transport.rs | 57 ++++++++++++++++++- 2 files changed, 65 insertions(+), 2 deletions(-) diff --git a/crates/tinyinference-llm/src/providers/openai/config.rs b/crates/tinyinference-llm/src/providers/openai/config.rs index 784d59c5..f2d45e09 100644 --- a/crates/tinyinference-llm/src/providers/openai/config.rs +++ b/crates/tinyinference-llm/src/providers/openai/config.rs @@ -135,10 +135,17 @@ pub(crate) fn build_openai_model_with( Arc::new(configure_openai_model(model, config, http)) } +/// Provider name of the generic OpenAI-compatible local runtime. +const LOCAL_OPENAI_PROVIDER: &str = "local-openai"; + /// Builds a local OpenAI-compatible model from a completed config: no /// Anthropic cache markers, and every system message sent as one leading /// `system` message, because local chat templates commonly reject a `system` /// turn anywhere but first (tinyhumansai/openhuman#7440). +/// +/// A `local-openai` model (any OpenAI-compatible endpoint, often a reseller or +/// marketplace) fulfils `invoke` by streaming; see +/// [`OpenAiModel::with_invoke_via_stream`]. pub(crate) fn build_local_openai_model_with( config: OpenAiConfig<'_>, http: Option, @@ -151,7 +158,8 @@ pub(crate) fn build_local_openai_model_with( ) .with_auth_style(config.auth_style.clone()) .with_anthropic_cache_control(false) - .with_single_system_message(); + .with_single_system_message() + .with_invoke_via_stream(config.provider_name == LOCAL_OPENAI_PROVIDER); Arc::new(configure_openai_model(model, config, http)) } diff --git a/crates/tinyinference-llm/src/providers/openai/transport.rs b/crates/tinyinference-llm/src/providers/openai/transport.rs index b17b4ebf..1d314cf4 100644 --- a/crates/tinyinference-llm/src/providers/openai/transport.rs +++ b/crates/tinyinference-llm/src/providers/openai/transport.rs @@ -94,6 +94,9 @@ pub struct OpenAiModel { /// When `true`, calls go to the OpenAI **Responses API** (`/v1/responses`) /// instead of Chat Completions. See [`Self::with_responses_api_primary`]. responses_api_primary: bool, + /// When `true`, a Chat Completions `invoke` is fulfilled by streaming and + /// collecting the reply. See [`Self::with_invoke_via_stream`]. + invoke_via_stream: bool, responses_document_input: bool, /// When `true` (Responses path only), omit `max_output_tokens` from the wire /// body — the OpenAI Codex OAuth backend rejects it. See @@ -439,6 +442,7 @@ impl OpenAiModel { explicit_cache_control: false, anthropic_cache_control: true, responses_api_primary: false, + invoke_via_stream: false, responses_document_input: false, responses_omit_max_output_tokens: false, extra_query_params: Vec::new(), @@ -481,6 +485,22 @@ impl OpenAiModel { self } + /// Fulfils [`ChatModel::invoke`] on Chat Completions by streaming the + /// reply and collecting it, instead of one non-streaming request. + /// + /// Some OpenAI-compatible sellers clip a non-streaming reply (the opening + /// of the answer and every reasoning token are lost) while the same reply + /// streams whole. With this on, `invoke` returns what the stream carried: + /// the same text, reasoning, tool calls, usage and finish reason a + /// streaming caller collects. The unary request timeout still bounds the + /// whole call. A gateway that refuses `stream: true` (a 400, 404, 405, + /// 406, 415, 422 or 501 that is not a context overflow) is retried once + /// without streaming. The Responses API path is unaffected. + pub fn with_invoke_via_stream(mut self, enabled: bool) -> Self { + self.invoke_via_stream = enabled; + self + } + /// Omits `max_output_tokens` from Responses requests (the Codex OAuth backend /// rejects it). No effect on the Chat Completions path. pub fn with_responses_omit_max_output_tokens(mut self) -> Self { @@ -2135,6 +2155,15 @@ pub(super) fn degrade_for_400( /// falls back to [`DEFAULT_REQUEST_TIMEOUT_SECS`], while a streaming call gets no /// overall cap (a total-request timeout would truncate a legitimately /// long-running stream mid-flight). +/// Whether a failed `stream: true` request reads as the gateway refusing to +/// stream (so a non-streaming retry may succeed), rather than a failure the +/// retry would only repeat: auth, rate limits, server errors, or a context +/// overflow. +fn streaming_refused(error: &ProviderError) -> bool { + matches!(error.status, Some(400 | 404 | 405 | 406 | 415 | 422 | 501)) + && error.code.as_deref() != Some(CONTEXT_OVERFLOW_CODE) +} + pub(super) fn request_timeout(timeout_ms: Option, streaming: bool) -> Option { match timeout_ms { Some(ms) => Some(Duration::from_millis(ms)), @@ -2282,13 +2311,39 @@ impl ChatModel for OpenAiModel { /// Returns [`Error::Model`] on transport failure or a non-2xx /// status (the message includes the status code and response body), and /// [`Error::Serialization`] when the response cannot be decoded. - async fn invoke(&self, _state: &State, request: ModelRequest) -> Result { + async fn invoke(&self, state: &State, request: ModelRequest) -> Result { if self.responses_api_primary { return self .invoke_responses(&request) .await .map(|response| response.inherit_correlation(request.correlation)); } + if self.invoke_via_stream { + // Keep the unary cap: an uncapped stream could otherwise hang a + // caller that asked for one bounded answer. + let mut streamed = request.clone(); + streamed.timeout_ms = Some( + request + .timeout_ms + .unwrap_or(DEFAULT_REQUEST_TIMEOUT_SECS * 1000), + ); + match self.stream(state, streamed).await { + Ok(stream) => { + let mut response = crate::model::collect_model_stream(stream).await?; + self.stamp_origin(&mut response, &request, CHAT_COMPLETIONS_API); + return Ok(response.inherit_correlation(request.correlation)); + } + Err(Error::Provider(error)) if streaming_refused(&error) => { + tracing::warn!( + provider = %self.provider, + model = %self.model, + status = ?error.status, + "openai invoke: streaming refused; retrying without streaming" + ); + } + Err(error) => return Err(error), + } + } let response = self .post_chat_with_degrade(&request, false, "request") .await?; From 32c65282731ddd58b2b32656eb13345be425119b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 11 Nov 2026 22:28:51 +0000 Subject: [PATCH 3/8] test(openai): cover streamed recovery of prompt-guided tool calls Adds a wire test that streams a prompt-guided tool call split across two chunks through the local-openai provider and checks the recovered calls and text match a non-streaming invoke of the same reply. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/providers/openai/wire_tests.rs | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs index 1f0b16dd..acd159aa 100644 --- a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs @@ -807,6 +807,58 @@ async fn local_openai_invoke_falls_back_when_streaming_is_refused() { assert_ne!(request_body(&raw[1])["stream"], json!(true)); } +/// A prompt-guided tool call split across streamed chunks. +const TOOL_CALL_SSE: &str = concat!( + r#"data: {"id":"c3","choices":[{"index":0,"delta":{"content":"looking {\"name\":\"lookup\","}}]}"#, + "\n\n", + r#"data: {"id":"c3","choices":[{"index":0,"delta":{"content":"\"arguments\":{\"q\":\"hi\"}}"},"finish_reason":"stop"}]}"#, + "\n\n", + "data: [DONE]\n\n", +); + +/// A `local-openai` invoke that streams recovers prompt-guided tool calls +/// exactly as a non-streaming invoke recovers them from the same reply. +#[tokio::test] +async fn local_openai_streamed_invoke_recovers_tool_calls_like_a_unary_invoke() { + let request = || { + ModelRequest::new(vec![Message::user("look it up")]).with_tools(vec![ + crate::tool::ToolSchema::new("lookup", "look something up", json!({"type": "object"})), + ]) + }; + let unary = json!({ + "id": "c3", + "choices": [{ + "index": 0, + "message": { + "role": "assistant", + "content": r#"looking {"name":"lookup","arguments":{"q":"hi"}}"# + }, + "finish_reason": "stop" + }] + }) + .to_string(); + + let (base, server) = serve_by_stream_flag(1, TOOL_CALL_SSE, unary.clone()); + let streamed = local_runtime_model("local-openai", &base) + .invoke(&(), request()) + .await + .unwrap(); + server.join().unwrap(); + let (base, server) = serve_by_stream_flag(1, TOOL_CALL_SSE, unary); + let plain = local_runtime_model("ollama", &base) + .invoke(&(), request()) + .await + .unwrap(); + server.join().unwrap(); + + let calls = streamed.tool_calls(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].name, "lookup"); + assert_eq!(calls[0].arguments, json!({ "q": "hi" })); + assert_eq!(calls, plain.tool_calls()); + assert_eq!(streamed.text(), plain.text()); +} + /// Other local runtimes keep the plain non-streaming invoke. #[tokio::test] async fn other_local_runtimes_invoke_without_streaming() { From bc74062e566c229f596e04e34ec666210e88d1ff Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 11 Nov 2026 22:29:20 +0000 Subject: [PATCH 4/8] test(openai): compare recovered tool calls without generated ids The streamed invoke test now checks that recovered tool call ids are minted with the expected prefix while comparing name, arguments and invalid flag separately, since ids come from a process-wide sequence and differ by design. It also asserts the streamed finish reason matches the unary invoke. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/providers/openai/wire_tests.rs | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs index acd159aa..817a6cdf 100644 --- a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs @@ -774,7 +774,11 @@ async fn local_openai_invoke_streams_and_returns_the_whole_answer() { assert_eq!(request_body(&raw[0])["stream"], json!(true)); assert_eq!(response.text(), "ALPHA BRAVO CHARLIE DELTA ECHO"); assert_eq!( - response.message.content.first().and_then(|block| block.as_thinking()), + response + .message + .content + .first() + .and_then(|block| block.as_thinking()), Some(("Reply verbatim, or formatting.", None)) ); assert_eq!(response.finish_reason.as_deref(), Some("stop")); @@ -855,8 +859,18 @@ async fn local_openai_streamed_invoke_recovers_tool_calls_like_a_unary_invoke() assert_eq!(calls.len(), 1); assert_eq!(calls[0].name, "lookup"); assert_eq!(calls[0].arguments, json!({ "q": "hi" })); - assert_eq!(calls, plain.tool_calls()); + // Recovered ids are minted from a process-wide sequence, so they differ + // by design; everything else must match. + let without_id = |calls: Vec| { + calls + .into_iter() + .map(|call| (call.name, call.arguments, call.invalid)) + .collect::>() + }; + assert!(calls[0].id.starts_with("text-"), "{}", calls[0].id); + assert_eq!(without_id(calls), without_id(plain.tool_calls())); assert_eq!(streamed.text(), plain.text()); + assert_eq!(streamed.finish_reason, plain.finish_reason); } /// Other local runtimes keep the plain non-streaming invoke. From 1445cf625d58c65d59f24cef103e4a5a31aee267 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 11 Nov 2026 22:29:38 +0000 Subject: [PATCH 5/8] test(openai): borrow tool calls in streamed invoke comparison The helper that strips recovered ids now takes a slice and clones the fields it compares, so the original calls stay available for the id assertion that follows. Auto-committed-on: cuttlefish Co-authored-by: Medulla --- .../src/providers/openai/wire_tests.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs index 817a6cdf..e3507561 100644 --- a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs @@ -861,10 +861,16 @@ async fn local_openai_streamed_invoke_recovers_tool_calls_like_a_unary_invoke() assert_eq!(calls[0].arguments, json!({ "q": "hi" })); // Recovered ids are minted from a process-wide sequence, so they differ // by design; everything else must match. - let without_id = |calls: Vec| { + let without_id = |calls: &[crate::tool::ToolCall]| { calls - .into_iter() - .map(|call| (call.name, call.arguments, call.invalid)) + .iter() + .map(|call| { + ( + call.name.clone(), + call.arguments.clone(), + call.invalid.clone(), + ) + }) .collect::>() }; assert!(calls[0].id.starts_with("text-"), "{}", calls[0].id); From 7c806a06d1ec7daf716ba583aaa1d0685636ffb7 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 11 Nov 2026 22:42:07 +0000 Subject: [PATCH 6/8] fix(openai): show the response observer an assembled body on a streamed invoke A streamed invoke has no raw body, so on_response was skipped. It now receives a Chat Completions body assembled from the stream, which parses back to the same answer. Review feedback on #151. Co-authored-by: Medulla --- .../src/providers/openai/convert.rs | 66 +++++++++++++++++++ .../src/providers/openai/transport.rs | 10 ++- .../src/providers/openai/wire_tests.rs | 48 ++++++++++++++ 3 files changed, 123 insertions(+), 1 deletion(-) diff --git a/crates/tinyinference-llm/src/providers/openai/convert.rs b/crates/tinyinference-llm/src/providers/openai/convert.rs index 8eeba275..f0c3853c 100644 --- a/crates/tinyinference-llm/src/providers/openai/convert.rs +++ b/crates/tinyinference-llm/src/providers/openai/convert.rs @@ -624,6 +624,72 @@ fn usd_cost_nanos(cost: &serde_json::Value) -> Option { Some((value * 1_000_000_000.0).round() as i64) } +/// A Chat Completions body for a reply collected from a stream, in the shape +/// [`parse_chat_response`] reads: visible text as `content`, thinking as +/// `reasoning_content`, tool calls with stringified arguments, the finish +/// reason, usage, and any gateway extension fields kept on `raw`. +/// +/// It lets a response observer see one body per successful call whether or +/// not the call streamed; parsing it back yields the same answer. +pub(super) fn assembled_completion_body(response: &ModelResponse) -> Value { + let reasoning: String = response + .message + .content + .iter() + .filter_map(|block| block.as_thinking().map(|(text, _)| text)) + .collect(); + let mut message = json!({ "role": "assistant", "content": response.text() }); + if !reasoning.is_empty() { + message["reasoning_content"] = json!(reasoning); + } + let tool_calls: Vec = response + .tool_calls() + .iter() + .map(|call| { + json!({ + "id": call.id, + "type": "function", + "function": { "name": call.name, "arguments": call.arguments.to_string() } + }) + }) + .collect(); + if !tool_calls.is_empty() { + message["tool_calls"] = json!(tool_calls); + } + let mut body = json!({ + "object": "chat.completion", + "choices": [{ "index": 0, "message": message, "finish_reason": response.finish_reason }] + }); + if let Some(id) = &response.message.id { + body["id"] = json!(id); + } + if let Some(usage) = response.usage { + body["usage"] = json!({ + "prompt_tokens": usage.input_tokens, + "completion_tokens": usage.output_tokens, + "total_tokens": usage.total_tokens, + "prompt_tokens_details": { "cached_tokens": usage.cache_read_tokens }, + "completion_tokens_details": { "reasoning_tokens": usage.reasoning_tokens } + }); + } + if let (Value::Object(body), Some(Value::Object(raw))) = (&mut body, &response.raw) { + for (key, value) in raw { + match (body.get_mut(key), value) { + (Some(Value::Object(known)), Value::Object(extra)) => { + for (field, item) in extra { + known.entry(field.clone()).or_insert_with(|| item.clone()); + } + } + (None, _) => { + body.insert(key.clone(), value.clone()); + } + _ => {} + } + } + } + body +} + /// Normalizes provider-specific reasoning/thinking payloads into text. /// /// OpenAI-compatible gateways do not agree on this field: some stream a plain diff --git a/crates/tinyinference-llm/src/providers/openai/transport.rs b/crates/tinyinference-llm/src/providers/openai/transport.rs index 1d314cf4..58f60cf8 100644 --- a/crates/tinyinference-llm/src/providers/openai/transport.rs +++ b/crates/tinyinference-llm/src/providers/openai/transport.rs @@ -493,7 +493,9 @@ impl OpenAiModel { /// streams whole. With this on, `invoke` returns what the stream carried: /// the same text, reasoning, tool calls, usage and finish reason a /// streaming caller collects. The unary request timeout still bounds the - /// whole call. A gateway that refuses `stream: true` (a 400, 404, 405, + /// whole call. A [`crate::providers::ProviderRequestOptions::on_response`] observer still + /// sees one body per successful call: a Chat Completions body assembled + /// from the stream. A gateway that refuses `stream: true` (a 400, 404, 405, /// 406, 415, 422 or 501 that is not a context overflow) is retried once /// without streaming. The Responses API path is unaffected. pub fn with_invoke_via_stream(mut self, enabled: bool) -> Self { @@ -2330,6 +2332,12 @@ impl ChatModel for OpenAiModel { match self.stream(state, streamed).await { Ok(stream) => { let mut response = crate::model::collect_model_stream(stream).await?; + // The unary path shows its raw body to `on_response`; a + // collected stream has none, so show the assembled one. + if self.request_options.on_response.is_some() { + self.request_options + .observe_response(&assembled_completion_body(&response)); + } self.stamp_origin(&mut response, &request, CHAT_COMPLETIONS_API); return Ok(response.inherit_correlation(request.correlation)); } diff --git a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs index e3507561..5d6ea0cd 100644 --- a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs @@ -788,6 +788,54 @@ async fn local_openai_invoke_streams_and_returns_the_whole_answer() { assert_eq!(origin.model, "glm-5.3-flash"); } +/// An invoke fulfilled by streaming still reports its success to the +/// `on_response` observer, with a Chat Completions body assembled from the +/// stream that parses back to the same answer. +#[tokio::test] +async fn streamed_invoke_reports_an_assembled_body_to_the_response_observer() { + let (base, server) = serve_by_stream_flag(1, THINKING_SSE, clipped_completion_body()); + let observed: Arc>> = Arc::default(); + let sink = Arc::clone(&observed); + let model = OpenAiModel::compatible_provider("local-openai", "k", &base, "glm-5.3-flash") + .with_invoke_via_stream(true) + .with_request_options(crate::providers::ProviderRequestOptions { + on_response: Some(Arc::new(move |body: &Value| { + sink.lock().unwrap().push(body.clone()); + })), + ..Default::default() + }); + + let response = ChatModel::<()>::invoke( + &model, + &(), + ModelRequest::new(vec![Message::user("say it")]), + ) + .await + .unwrap(); + server.join().unwrap(); + + let observed = observed.lock().unwrap(); + assert_eq!( + observed.len(), + 1, + "observer called once per successful invoke" + ); + let body = &observed[0]; + assert_eq!(body["object"], json!("chat.completion")); + assert_eq!( + body["choices"][0]["message"]["content"], + json!("ALPHA BRAVO CHARLIE DELTA ECHO") + ); + let reparsed = parse_chat_response(body.clone(), None).unwrap(); + assert_eq!(reparsed.text(), response.text()); + assert_eq!(reparsed.message.content, response.message.content); + assert_eq!(reparsed.finish_reason, response.finish_reason); + assert_eq!( + reparsed.usage.map(|usage| usage.total_tokens), + response.usage.map(|usage| usage.total_tokens) + ); +} + /// A gateway that refuses `stream: true` still answers: the invoke falls back /// to one non-streaming request. #[tokio::test] From f5a375c72ce2d074e28f7e8a676b3143411d8be6 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 11 Nov 2026 22:49:00 +0000 Subject: [PATCH 7/8] fix(openai): keep cache-write tokens in the assembled observer body; document on_response for streamed invokes The body on_response receives for a streamed invoke dropped prompt_tokens_details.cache_write_tokens, so parsing it back lost cache-write accounting. The on_response docs now say a streamed invoke passes an assembled Chat Completions body. Review feedback on #151 and tinyagents#610. Co-authored-by: Medulla --- .../src/providers/openai/convert.rs | 5 +++- .../src/providers/openai/mod_tests.rs | 26 +++++++++++++++++++ .../tinyinference-llm/src/providers/types.rs | 9 +++++-- 3 files changed, 37 insertions(+), 3 deletions(-) diff --git a/crates/tinyinference-llm/src/providers/openai/convert.rs b/crates/tinyinference-llm/src/providers/openai/convert.rs index f0c3853c..26723afc 100644 --- a/crates/tinyinference-llm/src/providers/openai/convert.rs +++ b/crates/tinyinference-llm/src/providers/openai/convert.rs @@ -668,7 +668,10 @@ pub(super) fn assembled_completion_body(response: &ModelResponse) -> Value { "prompt_tokens": usage.input_tokens, "completion_tokens": usage.output_tokens, "total_tokens": usage.total_tokens, - "prompt_tokens_details": { "cached_tokens": usage.cache_read_tokens }, + "prompt_tokens_details": { + "cached_tokens": usage.cache_read_tokens, + "cache_write_tokens": usage.cache_creation_tokens + }, "completion_tokens_details": { "reasoning_tokens": usage.reasoning_tokens } }); } diff --git a/crates/tinyinference-llm/src/providers/openai/mod_tests.rs b/crates/tinyinference-llm/src/providers/openai/mod_tests.rs index a3383226..f3d07137 100644 --- a/crates/tinyinference-llm/src/providers/openai/mod_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/mod_tests.rs @@ -1211,6 +1211,32 @@ fn parse_chat_response_reads_list_content_parts() { )); } +/// The body a response observer sees for a streamed invoke parses back to the +/// same usage the collected stream reports, every counter included. +#[tokio::test] +async fn an_assembled_completion_body_keeps_the_streamed_usage() { + let raw: Vec> = vec![ + b"data: {\"id\":\"c1\",\"choices\":[{\"delta\":{\"reasoning_content\":\"hm\"}}]}\n\n".to_vec(), + b"data: {\"id\":\"c1\",\"choices\":[{\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\n".to_vec(), + b"data: {\"choices\":[],\"usage\":{\"prompt_tokens\":40,\"completion_tokens\":9,\"total_tokens\":49,\"prompt_tokens_details\":{\"cached_tokens\":11,\"cache_write_tokens\":7},\"completion_tokens_details\":{\"reasoning_tokens\":3}}}\n\n".to_vec(), + b"data: [DONE]\n\n".to_vec(), + ]; + let response = collect_sse(raw) + .await + .into_iter() + .find_map(|item| match item { + ModelStreamItem::Completed(response) => Some(response), + _ => None, + }) + .expect("stream ends with Completed"); + let usage = response.usage.expect("streamed usage"); + assert_eq!(usage.cache_creation_tokens, 7); + + let reparsed = parse_chat_response(assembled_completion_body(&response), None).unwrap(); + assert_eq!(reparsed.usage, Some(usage)); + assert_eq!(reparsed.message.content, response.message.content); +} + #[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 diff --git a/crates/tinyinference-llm/src/providers/types.rs b/crates/tinyinference-llm/src/providers/types.rs index cbb8d6a2..89dc3bd4 100644 --- a/crates/tinyinference-llm/src/providers/types.rs +++ b/crates/tinyinference-llm/src/providers/types.rs @@ -25,7 +25,10 @@ use crate::model::ModelResponse; /// provider-specific fields (for example a beta header's JSON companion, or /// an organization id) without the harness knowing about them. `on_response` /// runs after a successful response is parsed into JSON, before the adapter -/// normalizes it, so a host can log or inspect the raw payload. `http` +/// normalizes it, so a host can log or inspect the raw payload. An OpenAI +/// `invoke` fulfilled by streaming (`OpenAiModel::with_invoke_via_stream`) +/// has no single raw payload; it passes a Chat Completions body assembled from +/// the collected stream instead, which parses back to the same answer. `http` /// overrides the adapter's own [`reqwest::Client`] (for a custom proxy, /// timeout, or TLS configuration) when set. /// @@ -36,7 +39,9 @@ use crate::model::ModelResponse; pub struct ProviderRequestOptions { /// Invoked with the mutable wire payload immediately before it is sent. pub on_payload: Option, - /// Invoked with the raw response payload after a successful call. + /// Invoked with the raw response payload after a successful call (for an + /// OpenAI invoke fulfilled by streaming, a Chat Completions body assembled + /// from the stream). pub on_response: Option, /// HTTP client to use in place of the adapter's own, when set. pub http: Option, From e8156d1d48a5fef0c787c6556adcdd8aff5ad266 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 11 Nov 2026 22:57:31 +0000 Subject: [PATCH 8/8] fix(openai): streamed invoke returns raw, keeps invalid tool args, names the model, and bounds the retry Review round 3 on #151: - response.raw carries the assembled completion body when the stream left it empty, as the unary path returns its body there; - an invalid tool call keeps its original argument text in that body, so it reads back as invalid; - the body names the request's effective model; - a non-streaming retry after a refused stream gets only what is left of the call's time budget, and the refusal is returned once it is spent. Co-authored-by: Medulla --- .../src/providers/openai/convert.rs | 13 ++++- .../src/providers/openai/mod_tests.rs | 58 ++++++++++++++++++- .../src/providers/openai/transport.rs | 41 +++++++++---- .../src/providers/openai/wire_tests.rs | 28 +++++++++ 4 files changed, 125 insertions(+), 15 deletions(-) diff --git a/crates/tinyinference-llm/src/providers/openai/convert.rs b/crates/tinyinference-llm/src/providers/openai/convert.rs index 26723afc..0cf36ed1 100644 --- a/crates/tinyinference-llm/src/providers/openai/convert.rs +++ b/crates/tinyinference-llm/src/providers/openai/convert.rs @@ -627,11 +627,13 @@ fn usd_cost_nanos(cost: &serde_json::Value) -> Option { /// A Chat Completions body for a reply collected from a stream, in the shape /// [`parse_chat_response`] reads: visible text as `content`, thinking as /// `reasoning_content`, tool calls with stringified arguments, the finish -/// reason, usage, and any gateway extension fields kept on `raw`. +/// reason, usage, the request's effective `model`, and any gateway extension +/// fields kept on `raw`. An invalid tool call keeps its original argument text, +/// so it reads back as invalid. /// /// It lets a response observer see one body per successful call whether or /// not the call streamed; parsing it back yields the same answer. -pub(super) fn assembled_completion_body(response: &ModelResponse) -> Value { +pub(super) fn assembled_completion_body(response: &ModelResponse, model: &str) -> Value { let reasoning: String = response .message .content @@ -646,10 +648,14 @@ pub(super) fn assembled_completion_body(response: &ModelResponse) -> Value { .tool_calls() .iter() .map(|call| { + let arguments = match (&call.invalid, &call.arguments) { + (Some(_), Value::String(original)) => original.clone(), + _ => call.arguments.to_string(), + }; json!({ "id": call.id, "type": "function", - "function": { "name": call.name, "arguments": call.arguments.to_string() } + "function": { "name": call.name, "arguments": arguments } }) }) .collect(); @@ -658,6 +664,7 @@ pub(super) fn assembled_completion_body(response: &ModelResponse) -> Value { } let mut body = json!({ "object": "chat.completion", + "model": model, "choices": [{ "index": 0, "message": message, "finish_reason": response.finish_reason }] }); if let Some(id) = &response.message.id { diff --git a/crates/tinyinference-llm/src/providers/openai/mod_tests.rs b/crates/tinyinference-llm/src/providers/openai/mod_tests.rs index f3d07137..9838f122 100644 --- a/crates/tinyinference-llm/src/providers/openai/mod_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/mod_tests.rs @@ -1232,11 +1232,67 @@ async fn an_assembled_completion_body_keeps_the_streamed_usage() { let usage = response.usage.expect("streamed usage"); assert_eq!(usage.cache_creation_tokens, 7); - let reparsed = parse_chat_response(assembled_completion_body(&response), None).unwrap(); + let reparsed = parse_chat_response(assembled_completion_body(&response, "m"), None).unwrap(); assert_eq!(reparsed.usage, Some(usage)); assert_eq!(reparsed.message.content, response.message.content); } +/// A streamed tool call whose arguments cannot be parsed stays invalid when +/// the assembled body is parsed back: its original text is not re-encoded +/// into a valid JSON string. +#[tokio::test] +async fn an_assembled_completion_body_keeps_invalid_tool_arguments_invalid() { + let raw: Vec> = vec![ + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-1\",\"function\":{\"name\":\"lookup\",\"arguments\":\"not json\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\n".to_vec(), + b"data: [DONE]\n\n".to_vec(), + ]; + let response = collect_sse(raw) + .await + .into_iter() + .find_map(|item| match item { + ModelStreamItem::Completed(response) => Some(response), + _ => None, + }) + .expect("stream ends with Completed"); + assert!(response.tool_calls()[0].invalid.is_some()); + + let reparsed = parse_chat_response(assembled_completion_body(&response, "m"), None).unwrap(); + let call = &reparsed.tool_calls()[0]; + assert!( + call.invalid.is_some(), + "re-parsed call lost its invalid marker: {call:?}" + ); + assert_eq!(call.arguments, response.tool_calls()[0].arguments); +} + +/// The assembled body names the model the request used. +#[test] +fn an_assembled_completion_body_names_the_model() { + let response = parse_chat_response( + json!({ "choices": [{ "message": { "role": "assistant", "content": "hi" }, "finish_reason": "stop" }] }), + None, + ) + .unwrap(); + assert_eq!( + assembled_completion_body(&response, "glm-5.3-flash")["model"], + json!("glm-5.3-flash") + ); +} + +/// A non-streaming retry after a refused stream gets only what is left of the +/// call's time budget, and none once it is spent. +#[test] +fn remaining_budget_subtracts_elapsed_time() { + use std::time::Duration; + let ms = Duration::from_millis; + assert_eq!( + super::transport::remaining_budget(ms(1000), ms(400)), + Some(ms(600)) + ); + assert_eq!(super::transport::remaining_budget(ms(1000), ms(1000)), None); + assert_eq!(super::transport::remaining_budget(ms(1000), ms(1500)), None); +} + #[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 diff --git a/crates/tinyinference-llm/src/providers/openai/transport.rs b/crates/tinyinference-llm/src/providers/openai/transport.rs index 58f60cf8..737c2954 100644 --- a/crates/tinyinference-llm/src/providers/openai/transport.rs +++ b/crates/tinyinference-llm/src/providers/openai/transport.rs @@ -493,9 +493,11 @@ impl OpenAiModel { /// streams whole. With this on, `invoke` returns what the stream carried: /// the same text, reasoning, tool calls, usage and finish reason a /// streaming caller collects. The unary request timeout still bounds the - /// whole call. A [`crate::providers::ProviderRequestOptions::on_response`] observer still + /// whole call, a non-streaming retry included. A + /// [`crate::providers::ProviderRequestOptions::on_response`] observer still /// sees one body per successful call: a Chat Completions body assembled - /// from the stream. A gateway that refuses `stream: true` (a 400, 404, 405, + /// from the stream, which is also returned on `ModelResponse::raw` when + /// the stream left it empty. A gateway that refuses `stream: true` (a 400, 404, 405, /// 406, 415, 422 or 501 that is not a context overflow) is retried once /// without streaming. The Responses API path is unaffected. pub fn with_invoke_via_stream(mut self, enabled: bool) -> Self { @@ -2157,6 +2159,12 @@ pub(super) fn degrade_for_400( /// falls back to [`DEFAULT_REQUEST_TIMEOUT_SECS`], while a streaming call gets no /// overall cap (a total-request timeout would truncate a legitimately /// long-running stream mid-flight). +/// What is left of a call's time `budget` after `elapsed`, or `None` when +/// nothing is. +pub(super) fn remaining_budget(budget: Duration, elapsed: Duration) -> Option { + budget.checked_sub(elapsed).filter(|rest| !rest.is_zero()) +} + /// Whether a failed `stream: true` request reads as the gateway refusing to /// stream (so a non-streaming retry may succeed), rather than a failure the /// retry would only repeat: auth, rate limits, server errors, or a context @@ -2313,7 +2321,7 @@ impl ChatModel for OpenAiModel { /// Returns [`Error::Model`] on transport failure or a non-2xx /// status (the message includes the status code and response body), and /// [`Error::Serialization`] when the response cannot be decoded. - async fn invoke(&self, state: &State, request: ModelRequest) -> Result { + async fn invoke(&self, state: &State, mut request: ModelRequest) -> Result { if self.responses_api_primary { return self .invoke_responses(&request) @@ -2322,32 +2330,43 @@ impl ChatModel for OpenAiModel { } if self.invoke_via_stream { // Keep the unary cap: an uncapped stream could otherwise hang a - // caller that asked for one bounded answer. - let mut streamed = request.clone(); - streamed.timeout_ms = Some( + // caller that asked for one bounded answer. The cap covers the + // whole call, a non-streaming retry included. + let budget = Duration::from_millis( request .timeout_ms .unwrap_or(DEFAULT_REQUEST_TIMEOUT_SECS * 1000), ); + let started = std::time::Instant::now(); + let mut streamed = request.clone(); + streamed.timeout_ms = Some(u64::try_from(budget.as_millis()).unwrap_or(u64::MAX)); match self.stream(state, streamed).await { Ok(stream) => { let mut response = crate::model::collect_model_stream(stream).await?; - // The unary path shows its raw body to `on_response`; a - // collected stream has none, so show the assembled one. - if self.request_options.on_response.is_some() { - self.request_options - .observe_response(&assembled_completion_body(&response)); + // The unary path returns its body on `raw` and shows it to + // `on_response`; a collected stream has none, so use one + // assembled from it. + let model = request.model.as_deref().unwrap_or(&self.model); + let body = assembled_completion_body(&response, model); + self.request_options.observe_response(&body); + if response.raw.is_none() { + response.raw = Some(body); } self.stamp_origin(&mut response, &request, CHAT_COMPLETIONS_API); return Ok(response.inherit_correlation(request.correlation)); } Err(Error::Provider(error)) if streaming_refused(&error) => { + let Some(rest) = remaining_budget(budget, started.elapsed()) else { + return Err(Error::Provider(error)); + }; tracing::warn!( provider = %self.provider, model = %self.model, status = ?error.status, + remaining_ms = rest.as_millis(), "openai invoke: streaming refused; retrying without streaming" ); + request.timeout_ms = Some(u64::try_from(rest.as_millis()).unwrap_or(u64::MAX)); } Err(error) => return Err(error), } diff --git a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs index 5d6ea0cd..7995dbe5 100644 --- a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs @@ -836,6 +836,34 @@ async fn streamed_invoke_reports_an_assembled_body_to_the_response_observer() { ); } +/// A streamed invoke returns the assembled completion body on `raw`, as the +/// unary path returns its body there, naming the request's effective model. +#[tokio::test] +async fn streamed_invoke_returns_the_assembled_body_on_raw() { + let (base, server) = serve_by_stream_flag(1, THINKING_SSE, clipped_completion_body()); + let model = local_runtime_model("local-openai", &base); + + let response = model + .invoke( + &(), + ModelRequest::new(vec![Message::user("say it")]).with_model("glm-override"), + ) + .await + .unwrap(); + server.join().unwrap(); + + let raw = response + .raw + .as_ref() + .expect("raw body on a streamed invoke"); + assert_eq!( + raw["choices"][0]["message"]["content"], + json!("ALPHA BRAVO CHARLIE DELTA ECHO") + ); + assert_eq!(raw["usage"]["total_tokens"], json!(71)); + assert_eq!(raw["model"], json!("glm-override")); +} + /// A gateway that refuses `stream: true` still answers: the invoke falls back /// to one non-streaming request. #[tokio::test]