feat(openai): local-openai invoke streams and collects, so clipped unary replies arrive whole

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<reqwest::Client>,
@@ -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))
 }

crates/tinyinference-llm/src/providers/openai/convert.rs

@@ -624,6 +624,82 @@ fn usd_cost_nanos(cost: &serde_json::Value) -> Option<i64> {
     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, 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, model: &str) -> 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<Value> = response
+        .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": arguments }
+            })
+        })
+        .collect();
+    if !tool_calls.is_empty() {
+        message["tool_calls"] = json!(tool_calls);
+    }
+    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 {
+        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,
+                "cache_write_tokens": usage.cache_creation_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

crates/tinyinference-llm/src/providers/openai/mod_tests.rs

@@ -1211,6 +1211,88 @@ 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<u8>> = 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, "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<u8>> = 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

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,26 @@ 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 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, 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 {
+        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 +2159,21 @@ 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<Duration> {
+    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
+/// 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<u64>, streaming: bool) -> Option<Duration> {
     match timeout_ms {
         Some(ms) => Some(Duration::from_millis(ms)),
@@ -2282,13 +2321,56 @@ impl<State: Send + Sync> ChatModel<State> 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<ModelResponse> {
+    async fn invoke(&self, state: &State, mut request: ModelRequest) -> Result<ModelResponse> {
         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. 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 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),
+            }
+        }
         let response = self
             .post_chat_with_degrade(&request, false, "request")
             .await?;