feat: stream reasoning summaries in AI agent Responses API steps (#11124)

* feat: stream reasoning summaries in AI agent Responses API steps

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix: retry without a reasoning summary a strict gateway rejects

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix: request the reasoning summary whether or not the step streams

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* feat: return the OpenAI reasoning summary in the agent step's reasoning

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix: retry without a reasoning summary a gateway rejects with 422

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

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)

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<String>,
+}
+
+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, String>,
+) -> (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::<BTreeMap<_, _>>()
+        .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::<HashMap<_, _>>()
+        };
+        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]

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

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<String, OpenAIToolCall>,
     /// Maps item_id -> (name, call_id) for function calls
     tool_call_metadata: HashMap<String, (String, String)>,
@@ -836,12 +846,15 @@ pub struct OpenAIResponsesSSEParser {
     pub used_websearch: bool,
     /// Token usage from response.completed event
     pub usage: Option<OpenAIResponsesUsage>,
+    /// Reasoning summary parts seen so far, to separate them as paragraphs
+    reasoning_summary_parts: usize,
 }
 
 impl OpenAIResponsesSSEParser {
     pub fn new(stream_event_processor: Box<dyn StreamEventSink>) -> 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`.

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<String>,
     #[serde(skip_serializing_if = "Option::is_none")]

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",