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/convert.rs b/crates/tinyinference-llm/src/providers/openai/convert.rs index 8eeba275..0cf36ed1 100644 --- a/crates/tinyinference-llm/src/providers/openai/convert.rs +++ b/crates/tinyinference-llm/src/providers/openai/convert.rs @@ -624,6 +624,82 @@ 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, 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 = 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 diff --git a/crates/tinyinference-llm/src/providers/openai/mod_tests.rs b/crates/tinyinference-llm/src/providers/openai/mod_tests.rs index a3383226..9838f122 100644 --- a/crates/tinyinference-llm/src/providers/openai/mod_tests.rs +++ b/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![ + 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![ + 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 b17b4ebf..737c2954 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,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 { + 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, streaming: bool) -> Option { match timeout_ms { Some(ms) => Some(Duration::from_millis(ms)), @@ -2282,13 +2321,56 @@ 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) .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?; diff --git a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs index 2aee13d5..7995dbe5 100644 --- a/crates/tinyinference-llm/src/providers/openai/wire_tests.rs +++ b/crates/tinyinference-llm/src/providers/openai/wire_tests.rs @@ -676,6 +676,301 @@ 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"); +} + +/// 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 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] +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)); +} + +/// 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" })); + // Recovered ids are minted from a process-wide sequence, so they differ + // by design; everything else must match. + let without_id = |calls: &[crate::tool::ToolCall]| { + calls + .iter() + .map(|call| { + ( + call.name.clone(), + call.arguments.clone(), + call.invalid.clone(), + ) + }) + .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. +#[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![ 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,