feat: return an ai agent step's thinking in its job result (#11140)

* feat: return an ai agent step's thinking in its job result

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* docs: say when an agent result carries no reasoning

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

backend/windmill-ai/src/providers/anthropic.rs

@@ -767,6 +767,7 @@ impl QueryBuilder for AnthropicQueryBuilder {
 
         let AnthropicSSEParser {
             accumulated_content,
+            accumulated_reasoning,
             accumulated_tool_calls,
             events_str,
             annotations,
@@ -790,6 +791,7 @@ impl QueryBuilder for AnthropicQueryBuilder {
             } else {
                 Some(accumulated_content)
             },
+            reasoning: (!accumulated_reasoning.is_empty()).then_some(accumulated_reasoning),
             tool_calls: accumulated_tool_calls.into_values().collect(),
             events_str: Some(events_str),
             annotations,

backend/windmill-ai/src/providers/bedrock.rs

@@ -1188,6 +1188,13 @@ impl BedrockQueryBuilder {
             Some(accumulated_text)
         };
 
+        // The block folded for replay is also what the reader sees as thinking. Read out
+        // before the block itself moves into the tool calls below.
+        let reasoning_text = reasoning
+            .as_ref()
+            .and_then(|r| r.reasoning_text.clone())
+            .filter(|t| !t.is_empty());
+
         let tool_calls = streaming_tool_calls_to_openai(
             accumulated_tool_calls.into_values().collect(),
             reasoning,
@@ -1195,6 +1202,7 @@ impl BedrockQueryBuilder {
 
         Ok(ParsedResponse::Text {
             content,
+            reasoning: reasoning_text,
             tool_calls,
             events_str: if events_str.is_empty() {
                 None

backend/windmill-ai/src/providers/google_ai.rs

@@ -666,6 +666,7 @@ impl QueryBuilder for GoogleAIQueryBuilder {
 
         let GeminiSSEParser {
             accumulated_content,
+            accumulated_reasoning,
             accumulated_tool_calls,
             mut events_str,
             stream_event_processor,
@@ -698,6 +699,7 @@ impl QueryBuilder for GoogleAIQueryBuilder {
             } else {
                 Some(accumulated_content)
             },
+            reasoning: (!accumulated_reasoning.is_empty()).then_some(accumulated_reasoning),
             tool_calls: accumulated_tool_calls.into_values().collect(),
             events_str: Some(events_str),
             annotations,

backend/windmill-ai/src/providers/openai.rs

@@ -538,6 +538,9 @@ 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,
             tool_calls: parser.accumulated_tool_calls.into_values().collect(),
             events_str: Some(parser.events_str),
             annotations: parser.annotations,

backend/windmill-ai/src/providers/other.rs

@@ -251,6 +251,7 @@ impl QueryBuilder for OtherQueryBuilder {
 
         let OpenAISSEParser {
             accumulated_content,
+            accumulated_reasoning,
             accumulated_tool_calls,
             mut events_str,
             stream_event_processor,
@@ -277,6 +278,7 @@ impl QueryBuilder for OtherQueryBuilder {
             } else {
                 Some(accumulated_content)
             },
+            reasoning: (!accumulated_reasoning.is_empty()).then_some(accumulated_reasoning),
             tool_calls: accumulated_tool_calls.into_values().collect(),
             events_str: Some(events_str),
             annotations: Vec::new(),

backend/windmill-ai/src/query_builder.rs

@@ -33,6 +33,8 @@ pub struct BuildRequestArgs<'a> {
 pub enum ParsedResponse {
     Text {
         content: Option<String>,
+        /// The thinking the model streamed before the answer, when it emitted any.
+        reasoning: Option<String>,
         tool_calls: Vec<OpenAIToolCall>,
         events_str: Option<String>,
         annotations: Vec<UrlCitation>,

backend/windmill-ai/src/sse.rs

@@ -135,6 +135,8 @@ pub trait SSEParser {
 
 pub struct OpenAISSEParser {
     pub accumulated_content: String,
+    /// The thinking streamed before the answer, kept so it can be stored with it.
+    pub accumulated_reasoning: String,
     pub accumulated_tool_calls: HashMap<i64, OpenAIToolCall>,
     pub events_str: String,
     pub stream_event_processor: Box<dyn StreamEventSink>,
@@ -146,6 +148,7 @@ impl OpenAISSEParser {
     pub fn new(stream_event_processor: Box<dyn StreamEventSink>) -> Self {
         Self {
             accumulated_content: String::new(),
+            accumulated_reasoning: String::new(),
             accumulated_tool_calls: HashMap::new(),
             events_str: String::new(),
             stream_event_processor,
@@ -175,6 +178,7 @@ impl SSEParser for OpenAISSEParser {
             if let Some(mut choices) = event.choices.filter(|s| !s.is_empty()) {
                 if let Some(delta) = choices.remove(0).delta {
                     if let Some(reasoning) = delta.reasoning_content.filter(|s| !s.is_empty()) {
+                        self.accumulated_reasoning.push_str(&reasoning);
                         let event = StreamingEvent::ReasoningTokenDelta { content: reasoning };
                         self.stream_event_processor
                             .send(event, &mut self.events_str)
@@ -353,6 +357,8 @@ enum ContentBlockState {
 /// Anthropic SSE Parser for streaming responses
 pub struct AnthropicSSEParser {
     pub accumulated_content: String,
+    /// The thinking streamed before the answer, kept so it can be stored with it.
+    pub accumulated_reasoning: String,
     pub accumulated_tool_calls: HashMap<i64, OpenAIToolCall>,
     pub events_str: String,
     pub stream_event_processor: Box<dyn StreamEventSink>,
@@ -375,6 +381,7 @@ impl AnthropicSSEParser {
     pub fn new(stream_event_processor: Box<dyn StreamEventSink>) -> Self {
         Self {
             accumulated_content: String::new(),
+            accumulated_reasoning: String::new(),
             accumulated_tool_calls: HashMap::new(),
             events_str: String::new(),
             stream_event_processor,
@@ -455,6 +462,7 @@ impl SSEParser for AnthropicSSEParser {
                                     .thinking
                                     .get_or_insert_with(String::new)
                                     .push_str(&thinking);
+                                self.accumulated_reasoning.push_str(&thinking);
                                 self.stream_event_processor
                                     .send(
                                         StreamingEvent::ReasoningTokenDelta { content: thinking },
@@ -523,6 +531,7 @@ impl SSEParser for AnthropicSSEParser {
                                     .thinking
                                     .get_or_insert_with(String::new)
                                     .push_str(&thinking);
+                                self.accumulated_reasoning.push_str(&thinking);
                                 self.stream_event_processor
                                     .send(
                                         StreamingEvent::ReasoningTokenDelta { content: thinking },
@@ -590,6 +599,8 @@ impl SSEParser for AnthropicSSEParser {
 /// `windmill_ai::ai_google` so the logic can be shared with the API proxy.
 pub struct GeminiSSEParser {
     pub accumulated_content: String,
+    /// The thinking streamed before the answer, kept so it can be stored with it.
+    pub accumulated_reasoning: String,
     pub accumulated_tool_calls: HashMap<i64, OpenAIToolCall>,
     pub events_str: String,
     pub stream_event_processor: Box<dyn StreamEventSink>,
@@ -603,6 +614,7 @@ impl GeminiSSEParser {
     pub fn new(stream_event_processor: Box<dyn StreamEventSink>) -> Self {
         Self {
             accumulated_content: String::new(),
+            accumulated_reasoning: String::new(),
             accumulated_tool_calls: HashMap::new(),
             events_str: String::new(),
             stream_event_processor,
@@ -621,6 +633,7 @@ impl SSEParser for GeminiSSEParser {
         };
 
         if let Some(reasoning) = parsed.reasoning.filter(|s| !s.is_empty()) {
+            self.accumulated_reasoning.push_str(&reasoning);
             self.stream_event_processor
                 .send(
                     StreamingEvent::ReasoningTokenDelta { content: reasoning },

backend/windmill-ai/src/types.rs

@@ -374,6 +374,13 @@ pub struct AIAgentResult<'a> {
     pub messages: Vec<Message<'a>>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub wm_stream: Option<String>,
+    /// 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.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub reasoning: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub usage: Option<TokenUsage>,
 }

backend/windmill-worker/src/ai_executor.rs

@@ -1211,6 +1211,7 @@ pub async fn run_agent(
     *has_stream = user_wants_streaming && is_text_output;
 
     let mut final_events_str = String::new();
+    let mut final_reasoning = String::new();
 
     // Always create a StreamEventProcessor for text output (use silent mode if user doesn't want streaming)
     let stream_event_processor = if is_text_output {
@@ -1463,6 +1464,7 @@ pub async fn run_agent(
         match parsed {
             ParsedResponse::Text {
                 content: response_content,
+                reasoning: response_reasoning,
                 tool_calls,
                 events_str,
                 annotations,
@@ -1479,6 +1481,7 @@ pub async fn run_agent(
                 if let Some(events_str) = events_str {
                     final_events_str.push_str(&events_str);
                 }
+                append_reasoning(&mut final_reasoning, response_reasoning.as_deref());
 
                 // Add websearch tool message if websearch was used
                 if used_websearch {
@@ -1801,6 +1804,7 @@ pub async fn run_agent(
         } else {
             None
         },
+        reasoning: (!final_reasoning.is_empty()).then_some(final_reasoning),
         usage: if final_usage.as_ref().map(|u| u.is_empty()).unwrap_or(true) {
             None
         } else {
@@ -1820,6 +1824,19 @@ fn streaming_requested(streaming: Option<bool>) -> bool {
     streaming.unwrap_or(true)
 }
 
+/// Add one iteration's thinking to the step's. Every iteration thinks, and a tool-call
+/// iteration's thinking is what led to the call, so the result keeps all of them in order,
+/// blank-line separated, rather than only the answering turn's.
+fn append_reasoning(accumulated: &mut String, reasoning: Option<&str>) {
+    let Some(reasoning) = reasoning.map(str::trim).filter(|r| !r.is_empty()) else {
+        return;
+    };
+    if !accumulated.is_empty() {
+        accumulated.push_str("\n\n");
+    }
+    accumulated.push_str(reasoning);
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1832,6 +1849,16 @@ mod tests {
         }
     }
 
+    #[test]
+    fn reasoning_keeps_every_iteration_in_order() {
+        let mut acc = String::new();
+        append_reasoning(&mut acc, Some("I need both cities.\n\n"));
+        append_reasoning(&mut acc, None);
+        append_reasoning(&mut acc, Some("   "));
+        append_reasoning(&mut acc, Some("Paris is closer."));
+        assert_eq!(acc, "I need both cities.\n\nParis is closer.");
+    }
+
     #[test]
     fn an_unwritten_streaming_field_streams() {
         assert!(streaming_requested(None));