diff --git a/backend/windmill-ai/src/providers/anthropic.rs b/backend/windmill-ai/src/providers/anthropic.rs index 038aef2c2e..0bb7316dcb 100644 --- a/backend/windmill-ai/src/providers/anthropic.rs +++ b/backend/windmill-ai/src/providers/anthropic.rs @@ -902,6 +902,7 @@ mod tests { attachments: None, has_websearch: false, prompt_cache_key: None, + reasoning_summary: false, }; AnthropicQueryBuilder::new(AIProvider::Anthropic, platform) diff --git a/backend/windmill-ai/src/providers/openai.rs b/backend/windmill-ai/src/providers/openai.rs index 8fa65dca21..14c9fdb4d3 100644 --- a/backend/windmill-ai/src/providers/openai.rs +++ b/backend/windmill-ai/src/providers/openai.rs @@ -1,6 +1,7 @@ use crate::{ ai_providers::AIProvider, ai_types::OpenAIToolCall, + credentials::ProviderCredentials, image_handler::{prepare_messages_for_api, s3_object_to_content_part}, proxy::{build_openai_compatible_proxy_request, ProxyBuildArgs, ProxyRequest}, query_builder::{BuildRequestArgs, ParsedResponse, QueryBuilder, StreamEventSink}, @@ -11,7 +12,14 @@ use crate::{ use async_trait::async_trait; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; -use windmill_common::{client::AuthedClient, error::Error}; +use std::{ + collections::{BTreeMap, HashMap}, + hash::{DefaultHasher, Hash, Hasher}, + time::{Duration, Instant}, +}; +use windmill_common::{cache::Cache, client::AuthedClient, error::Error}; + +use super::REASONING_OFF_SENTINEL; // Responses API structures #[derive(Deserialize)] @@ -192,13 +200,79 @@ pub struct ResponsesApiTextFormat { pub format: ResponsesApiTextFormatConfig, } -/// Reasoning config for the Responses API (`reasoning: { effort }`). -/// The summary is intentionally not requested, mirroring the copilot chat: OpenAI -/// gates reasoning summaries behind organization verification, so asking for one -/// would fail the request for unverified orgs. +/// Reasoning config for the Responses API (`reasoning: { effort, summary }`). #[derive(Serialize)] pub struct ResponsesApiReasoning { pub effort: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub summary: Option, +} + +lazy_static::lazy_static! { + /// Refused reasoning summaries, so later requests skip asking instead of paying a + /// rejected call each. A refusal belongs to the organization the request bills or to the + /// model, so the key holds the model and everything that authenticates (the API key and + /// the resource headers, which can carry it instead). Entries expire as an org gets verified. + static ref REASONING_SUMMARY_UNAVAILABLE: Cache<(String, u64), Instant> = Cache::new(500); +} + +const REASONING_SUMMARY_UNAVAILABLE_TTL: Duration = Duration::from_secs(3600); + +fn reasoning_summary_cache_key( + base_url: &str, + model: &str, + api_key: Option<&str>, + custom_headers: &HashMap, +) -> (String, u64) { + let mut hasher = DefaultHasher::new(); + model.hash(&mut hasher); + api_key.hash(&mut hasher); + // Sorted: two maps with the same entries can iterate them in different orders. + custom_headers + .iter() + .collect::>() + .hash(&mut hasher); + (base_url.to_string(), hasher.finish()) +} + +fn credentials_cache_key(credentials: &ProviderCredentials, model: &str) -> (String, u64) { + reasoning_summary_cache_key( + &credentials.base_url, + model, + credentials.api_key.as_deref(), + &credentials.custom_headers, + ) +} + +/// Whether this model is known to be refused reasoning summaries with these credentials. +pub fn is_reasoning_summary_unavailable(credentials: &ProviderCredentials, model: &str) -> bool { + REASONING_SUMMARY_UNAVAILABLE + .get(&credentials_cache_key(credentials, model)) + .is_some_and(|learned_at| learned_at.elapsed() < REASONING_SUMMARY_UNAVAILABLE_TTL) +} + +/// Record that this model was refused a reasoning summary with these credentials. +pub fn remember_reasoning_summary_unavailable(credentials: &ProviderCredentials, model: &str) { + REASONING_SUMMARY_UNAVAILABLE.insert(credentials_cache_key(credentials, model), Instant::now()); +} + +/// Whether a rejected request was refused over its reasoning summary, e.g. `Your +/// organization must be verified to generate reasoning summaries` (param +/// `reasoning.summary`). An OpenAI-kind resource can also point at a gateway that validates +/// the body strictly and names only the unknown `summary` property. +pub fn rejects_reasoning_summary(status: u16, body: &str) -> bool { + // 422 is how FastAPI-based gateways reject a body that fails validation. + if !matches!(status, 400 | 403 | 422) { + return false; + } + let body = body.to_lowercase(); + let unknown_field = body.contains("additional properties are not allowed") + || body.contains("unrecognized request argument") + || body.contains("extra inputs are not permitted"); + body.contains("reasoning.summary") + || body.contains("verified to generate reasoning summar") + || body.contains("verified to stream reasoning summar") + || (unknown_field && body.contains("summary")) } #[derive(Serialize)] @@ -435,9 +509,13 @@ impl OpenAIQueryBuilder { tools, stream: Some(true), temperature: args.temperature, - reasoning: args - .reasoning_effort - .map(|effort| ResponsesApiReasoning { effort: effort.to_string() }), + reasoning: args.reasoning_effort.map(|effort| ResponsesApiReasoning { + effort: effort.to_string(), + // A request that does not reason has nothing to summarize, yet asking still + // gets an unverified organization's request rejected. + summary: (args.reasoning_summary && effort != REASONING_OFF_SENTINEL) + .then(|| "auto".to_string()), + }), max_output_tokens: args.max_tokens, text, prompt_cache_key: args.prompt_cache_key, @@ -538,9 +616,8 @@ impl QueryBuilder for OpenAIQueryBuilder { } else { Some(parser.accumulated_content) }, - // The Responses stream has no reasoning-summary event in - // `OpenAIResponsesSSEEvent`, so nothing thinks out loud on this path yet. - reasoning: None, + reasoning: (!parser.accumulated_reasoning.is_empty()) + .then_some(parser.accumulated_reasoning), tool_calls: parser.accumulated_tool_calls.into_values().collect(), events_str: Some(parser.events_str), annotations: parser.annotations, @@ -640,8 +717,11 @@ mod tests { } } - async fn build_text_body(messages: &[OpenAIMessage], system_prompt: Option<&str>) -> String { - let args = BuildRequestArgs { + fn text_args<'a>( + messages: &'a [OpenAIMessage], + system_prompt: Option<&'a str>, + ) -> BuildRequestArgs<'a> { + BuildRequestArgs { messages, tools: None, model: "gpt-5", @@ -655,14 +735,98 @@ mod tests { attachments: None, has_websearch: false, prompt_cache_key: Some(PROMPT_CACHE_KEY), - }; + reasoning_summary: true, + } + } + async fn build_body(args: &BuildRequestArgs<'_>) -> String { OpenAIQueryBuilder::new(AIProvider::OpenAI) - .build_request(&args, &client(), "test-workspace") + .build_request(args, &client(), "test-workspace") .await .unwrap() } + async fn build_text_body(messages: &[OpenAIMessage], system_prompt: Option<&str>) -> String { + build_body(&text_args(messages, system_prompt)).await + } + + async fn reasoning_of(effort: Option<&str>, reasoning_summary: bool) -> serde_json::Value { + let messages = vec![message("user", "hi")]; + let args = BuildRequestArgs { + reasoning_effort: effort, + reasoning_summary, + ..text_args(&messages, None) + }; + let request: serde_json::Value = serde_json::from_str(&build_body(&args).await).unwrap(); + request["reasoning"].clone() + } + + #[tokio::test] + async fn requests_a_reasoning_summary_only_when_the_model_reasons() { + assert_eq!( + reasoning_of(Some("high"), true).await, + serde_json::json!({ "effort": "high", "summary": "auto" }) + ); + assert_eq!( + reasoning_of(Some("none"), true).await, + serde_json::json!({ "effort": "none" }) + ); + assert_eq!( + reasoning_of(Some("high"), false).await, + serde_json::json!({ "effort": "high" }) + ); + assert!(reasoning_of(None, true).await.is_null()); + } + + #[test] + fn recognizes_a_refused_reasoning_summary() { + let unverified = r#"{"error":{"message":"Your organization must be verified to generate reasoning summaries. Please go to: https://platform.openai.com/settings/organization/general and click on Verify Organization.","type":"invalid_request_error","param":"reasoning.summary","code":"unsupported_value"}}"#; + assert!(rejects_reasoning_summary(400, unverified)); + assert!(!rejects_reasoning_summary(500, unverified)); + assert!(rejects_reasoning_summary( + 400, + r#"{"detail":"Additional properties are not allowed ('summary' was unexpected)"}"# + )); + assert!(rejects_reasoning_summary( + 422, + r#"{"detail":[{"type":"extra_forbidden","loc":["body","reasoning","summary"],"msg":"Extra inputs are not permitted","input":"auto"}]}"# + )); + assert!(!rejects_reasoning_summary( + 400, + r#"{"error":{"message":"Invalid 'prompt_cache_key': string too long","param":"prompt_cache_key"}}"# + )); + } + + /// A resource can authenticate through `headers` with no API key: one organization's + /// refusal must not withhold summaries from another's. + #[test] + fn keys_a_refused_summary_by_the_header_credential() { + let headers = |pairs: &[(&str, &str)]| { + pairs + .iter() + .map(|(name, value)| (name.to_string(), value.to_string())) + .collect::>() + }; + let url = "https://api.openai.com/v1"; + let org_a = headers(&[("Authorization", "Bearer org-a"), ("X-Trace", "1")]); + let org_a_reordered = headers(&[("X-Trace", "1"), ("Authorization", "Bearer org-a")]); + let org_b = headers(&[("Authorization", "Bearer org-b"), ("X-Trace", "1")]); + + assert_ne!( + reasoning_summary_cache_key(url, "gpt-5", None, &org_a), + reasoning_summary_cache_key(url, "gpt-5", None, &org_b) + ); + assert_eq!( + reasoning_summary_cache_key(url, "gpt-5", None, &org_a), + reasoning_summary_cache_key(url, "gpt-5", None, &org_a_reordered) + ); + // A model can refuse summaries that another model on the same credentials streams. + assert_ne!( + reasoning_summary_cache_key(url, "gpt-5", None, &org_a), + reasoning_summary_cache_key(url, "gpt-5-mini", None, &org_a) + ); + } + /// The worker prepends the system prompt as a system message *and* passes it as /// `system_prompt`; the request must still carry it exactly once. #[tokio::test] diff --git a/backend/windmill-ai/src/query_builder.rs b/backend/windmill-ai/src/query_builder.rs index 60ece300a3..3281b51cd0 100644 --- a/backend/windmill-ai/src/query_builder.rs +++ b/backend/windmill-ai/src/query_builder.rs @@ -27,6 +27,9 @@ pub struct BuildRequestArgs<'a> { /// the prefix (the step), never from the request. `None` retries a key the /// endpoint rejected. pub prompt_cache_key: Option<&'a str>, + /// Ask for a summary of the model's reasoning where the provider streams one. + /// `false` once the provider refused summaries to these credentials. + pub reasoning_summary: bool, } /// Response from AI provider diff --git a/backend/windmill-ai/src/sse.rs b/backend/windmill-ai/src/sse.rs index 986642fd7e..054baeb5de 100644 --- a/backend/windmill-ai/src/sse.rs +++ b/backend/windmill-ai/src/sse.rs @@ -815,6 +815,14 @@ pub enum OpenAIResponsesSSEEvent { #[serde(rename = "response.output_text.annotation.added")] AnnotationAdded { annotation: OpenAIUrlCitationEvent }, + /// A new reasoning summary part starts (only sent when `reasoning.summary` was requested) + #[serde(rename = "response.reasoning_summary_part.added")] + ReasoningSummaryPartAdded {}, + + /// Reasoning summary text delta + #[serde(rename = "response.reasoning_summary_text.delta")] + ReasoningSummaryTextDelta { delta: String }, + /// Catch-all for unknown event types #[serde(other)] Other, @@ -823,6 +831,8 @@ pub enum OpenAIResponsesSSEEvent { /// OpenAI Responses API SSE Parser for streaming responses pub struct OpenAIResponsesSSEParser { pub accumulated_content: String, + /// The reasoning summary streamed before the answer, kept so it can be stored with it. + pub accumulated_reasoning: String, pub accumulated_tool_calls: HashMap, /// Maps item_id -> (name, call_id) for function calls tool_call_metadata: HashMap, @@ -836,12 +846,15 @@ pub struct OpenAIResponsesSSEParser { pub used_websearch: bool, /// Token usage from response.completed event pub usage: Option, + /// Reasoning summary parts seen so far, to separate them as paragraphs + reasoning_summary_parts: usize, } impl OpenAIResponsesSSEParser { pub fn new(stream_event_processor: Box) -> Self { Self { accumulated_content: String::new(), + accumulated_reasoning: String::new(), accumulated_tool_calls: HashMap::new(), tool_call_metadata: HashMap::new(), tool_call_arguments: HashMap::new(), @@ -850,6 +863,7 @@ impl OpenAIResponsesSSEParser { annotations: Vec::new(), used_websearch: false, usage: None, + reasoning_summary_parts: 0, } } } @@ -959,6 +973,28 @@ impl SSEParser for OpenAIResponsesSSEParser { } } + OpenAIResponsesSSEEvent::ReasoningSummaryPartAdded {} => { + self.reasoning_summary_parts += 1; + if self.reasoning_summary_parts > 1 { + self.accumulated_reasoning.push_str("\n\n"); + let event = + StreamingEvent::ReasoningTokenDelta { content: "\n\n".to_string() }; + self.stream_event_processor + .send(event, &mut self.events_str) + .await?; + } + } + + OpenAIResponsesSSEEvent::ReasoningSummaryTextDelta { delta } => { + if !delta.is_empty() { + self.accumulated_reasoning.push_str(&delta); + let event = StreamingEvent::ReasoningTokenDelta { content: delta }; + self.stream_event_processor + .send(event, &mut self.events_str) + .await?; + } + } + // Ignore other event types OpenAIResponsesSSEEvent::Done {} | OpenAIResponsesSSEEvent::Created {} @@ -1015,6 +1051,33 @@ mod tests { assert_eq!(token_usage.total_tokens, Some(4821)); } + struct ReasoningSink; + + #[async_trait::async_trait] + impl StreamEventSink for ReasoningSink { + async fn send(&self, event: StreamingEvent, events_str: &mut String) -> Result<(), Error> { + if let StreamingEvent::ReasoningTokenDelta { content } = event { + events_str.push_str(&content); + } + Ok(()) + } + } + + #[tokio::test] + async fn streams_openai_responses_reasoning_summary_parts_as_paragraphs() { + let mut parser = OpenAIResponsesSSEParser::new(Box::new(ReasoningSink)); + for data in [ + r#"{"type":"response.reasoning_summary_part.added","item_id":"rs_1","output_index":0,"summary_index":0,"part":{"type":"summary_text","text":""}}"#, + r#"{"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":0,"delta":"**Planning**"}"#, + r#"{"type":"response.reasoning_summary_part.added","item_id":"rs_1","output_index":0,"summary_index":1,"part":{"type":"summary_text","text":""}}"#, + r#"{"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":1,"delta":"Then answer"}"#, + ] { + parser.parse_event_data(data).await.unwrap(); + } + assert_eq!(parser.events_str, "**Planning**\n\nThen answer"); + assert_eq!(parser.accumulated_reasoning, "**Planning**\n\nThen answer"); + } + #[test] fn openai_delta_parses_reasoning_content() { // DeepSeek and similar stream reasoning under `reasoning_content`. diff --git a/backend/windmill-ai/src/types.rs b/backend/windmill-ai/src/types.rs index 8dfb9af1c6..ae85718e8a 100644 --- a/backend/windmill-ai/src/types.rs +++ b/backend/windmill-ai/src/types.rs @@ -377,8 +377,7 @@ pub struct AIAgentResult<'a> { /// The model's thinking across every iteration of the loop, in order, blank-line /// separated. Present whenever the provider's parser surfaced any, whether or not /// the step streams, so a downstream step never has to pick it out of `wm_stream`. - /// Absent when the model thought nothing, and on the OpenAI Responses path, whose - /// parser does not return reasoning yet. + /// Absent when the model thought nothing. #[serde(skip_serializing_if = "Option::is_none")] pub reasoning: Option, #[serde(skip_serializing_if = "Option::is_none")] diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index debfbeff39..156369e095 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -26,6 +26,10 @@ use windmill_ai::{ image_handler::upload_image_to_s3, providers::{ create_chat_completions_query_builder, create_query_builder, is_chat_completions_only, + openai::{ + is_reasoning_summary_unavailable, rejects_reasoning_summary, + remember_reasoning_summary_unavailable, + }, remember_chat_completions_only, }, proxy::{ @@ -1300,6 +1304,10 @@ pub async fn run_agent( attachments: args.user_attachments.as_deref(), has_websearch, prompt_cache_key: include_prompt_cache_key.then_some(prompt_cache_key.as_str()), + reasoning_summary: !is_reasoning_summary_unavailable( + &credentials, + args.provider.get_model(), + ), }; // A worker cannot run the client credentials exchange, so an OAuth resource @@ -1350,7 +1358,8 @@ pub async fn run_agent( // An endpoint can reject the request shape rather than the model: // `stream_options` and `prompt_cache_key`, which not every OpenAI-compatible - // gateway accepts, and the route itself, when an Azure resource is outside + // gateway accepts, a reasoning summary, which OpenAI refuses to unverified + // organizations, and the route itself, when an Azure resource is outside // the Responses API's model/region matrix. Each is retried once with that // part dropped. // Set where the route is found to be absent, and read once the fallback has @@ -1407,6 +1416,9 @@ pub async fn run_agent( && status.as_u16() == 400 && text.contains("prompt_cache_key"); + let summary_refused = build_args.reasoning_summary + && rejects_reasoning_summary(status.as_u16(), &text); + // Only the first call of the step may re-route: an endpoint that // does not serve this API rejects that one already, whereas a // rejection once the conversation is under way is about the @@ -1430,6 +1442,15 @@ pub async fn run_agent( ); include_prompt_cache_key = false; build_args.prompt_cache_key = None; + } else if summary_refused { + tracing::info!( + "Retrying request without the reasoning summary the endpoint refused" + ); + remember_reasoning_summary_unavailable( + &credentials, + args.provider.get_model(), + ); + build_args.reasoning_summary = false; } else if route_unserved { tracing::info!( "Endpoint rejected the request ({}), falling back to chat/completions",