diff --git a/backend/.sqlx/query-1c3473a0f9f6b6148b2c975f9f05bdefedf8a51c4e6ddf0eca367b9cc778d051.json b/backend/.sqlx/query-318ed7a45d8326e3ecbc9727ee3812adaeeead8b39b007a32badadf67ec5ca17.json similarity index 69% rename from backend/.sqlx/query-1c3473a0f9f6b6148b2c975f9f05bdefedf8a51c4e6ddf0eca367b9cc778d051.json rename to backend/.sqlx/query-318ed7a45d8326e3ecbc9727ee3812adaeeead8b39b007a32badadf67ec5ca17.json index fb27bd9446..ae6ffd4e3c 100644 --- a/backend/.sqlx/query-1c3473a0f9f6b6148b2c975f9f05bdefedf8a51c4e6ddf0eca367b9cc778d051.json +++ b/backend/.sqlx/query-318ed7a45d8326e3ecbc9727ee3812adaeeead8b39b007a32badadf67ec5ca17.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT id, conversation_id, message_type as \"message_type: MessageType\", content, job_id, created_at, created_seq, step_name, success\n FROM (\n SELECT id, conversation_id, message_type, content, job_id, created_at, created_seq, step_name, success\n FROM flow_conversation_message\n WHERE conversation_id = $1\n ORDER BY created_seq DESC\n LIMIT $2 OFFSET $3\n ) AS messages\n ORDER BY created_seq ASC\n ", + "query": "SELECT id, conversation_id, message_type as \"message_type: MessageType\", content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning\n FROM flow_conversation_message\n WHERE conversation_id = $1\n AND created_seq > $2\n ORDER BY created_seq ASC\n LIMIT $3\n ", "describe": { "columns": [ { @@ -58,6 +58,21 @@ "ordinal": 8, "name": "success", "type_info": "Bool" + }, + { + "ordinal": 9, + "name": "tool_arguments", + "type_info": "Text" + }, + { + "ordinal": 10, + "name": "tool_result", + "type_info": "Text" + }, + { + "ordinal": 11, + "name": "reasoning", + "type_info": "Text" } ], "parameters": { @@ -76,8 +91,11 @@ false, false, true, - false + false, + true, + true, + true ] }, - "hash": "1c3473a0f9f6b6148b2c975f9f05bdefedf8a51c4e6ddf0eca367b9cc778d051" + "hash": "318ed7a45d8326e3ecbc9727ee3812adaeeead8b39b007a32badadf67ec5ca17" } diff --git a/backend/.sqlx/query-b1a9a433e577133869c067b2ce383fc6ce4e9df307feb5fd3edc0d1276d61ff1.json b/backend/.sqlx/query-55dbe12954489532644b91e67b2a9df7ede61e050d3e634b795aaf7f60b3b05d.json similarity index 63% rename from backend/.sqlx/query-b1a9a433e577133869c067b2ce383fc6ce4e9df307feb5fd3edc0d1276d61ff1.json rename to backend/.sqlx/query-55dbe12954489532644b91e67b2a9df7ede61e050d3e634b795aaf7f60b3b05d.json index 1818efc0c0..2ae7033d0d 100644 --- a/backend/.sqlx/query-b1a9a433e577133869c067b2ce383fc6ce4e9df307feb5fd3edc0d1276d61ff1.json +++ b/backend/.sqlx/query-55dbe12954489532644b91e67b2a9df7ede61e050d3e634b795aaf7f60b3b05d.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id, step_name, success)\n VALUES ($1, $2, $3, $4, $5, $6)", + "query": "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id, step_name, success, tool_arguments, tool_result, reasoning)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)", "describe": { "columns": [], "parameters": { @@ -21,10 +21,13 @@ "Text", "Uuid", "Varchar", - "Bool" + "Bool", + "Text", + "Text", + "Text" ] }, "nullable": [] }, - "hash": "b1a9a433e577133869c067b2ce383fc6ce4e9df307feb5fd3edc0d1276d61ff1" + "hash": "55dbe12954489532644b91e67b2a9df7ede61e050d3e634b795aaf7f60b3b05d" } diff --git a/backend/.sqlx/query-e8802be9203c1e88a06e337260ccca029380139f89a01a89033e36a6ed9ac082.json b/backend/.sqlx/query-d6c65ce1d443d1e1783818d36c7eabd9633bbe5219bfd9683f336de19763cb58.json similarity index 62% rename from backend/.sqlx/query-e8802be9203c1e88a06e337260ccca029380139f89a01a89033e36a6ed9ac082.json rename to backend/.sqlx/query-d6c65ce1d443d1e1783818d36c7eabd9633bbe5219bfd9683f336de19763cb58.json index a3374d6cdf..d3d76191f8 100644 --- a/backend/.sqlx/query-e8802be9203c1e88a06e337260ccca029380139f89a01a89033e36a6ed9ac082.json +++ b/backend/.sqlx/query-d6c65ce1d443d1e1783818d36c7eabd9633bbe5219bfd9683f336de19763cb58.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT id, conversation_id, message_type as \"message_type: MessageType\", content, job_id, created_at, created_seq, step_name, success\n FROM flow_conversation_message\n WHERE conversation_id = $1\n AND created_seq > $2\n ORDER BY created_seq ASC\n LIMIT $3\n ", + "query": "SELECT id, conversation_id, message_type as \"message_type: MessageType\", content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning\n FROM (\n SELECT id, conversation_id, message_type, content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning\n FROM flow_conversation_message\n WHERE conversation_id = $1\n ORDER BY created_seq DESC\n LIMIT $2 OFFSET $3\n ) AS messages\n ORDER BY created_seq ASC\n ", "describe": { "columns": [ { @@ -58,6 +58,21 @@ "ordinal": 8, "name": "success", "type_info": "Bool" + }, + { + "ordinal": 9, + "name": "tool_arguments", + "type_info": "Text" + }, + { + "ordinal": 10, + "name": "tool_result", + "type_info": "Text" + }, + { + "ordinal": 11, + "name": "reasoning", + "type_info": "Text" } ], "parameters": { @@ -76,8 +91,11 @@ false, false, true, - false + false, + true, + true, + true ] }, - "hash": "e8802be9203c1e88a06e337260ccca029380139f89a01a89033e36a6ed9ac082" + "hash": "d6c65ce1d443d1e1783818d36c7eabd9633bbe5219bfd9683f336de19763cb58" } diff --git a/backend/migrations/20260908110914_flow_conversation_message_tool_call.down.sql b/backend/migrations/20260908110914_flow_conversation_message_tool_call.down.sql new file mode 100644 index 0000000000..56f20ba722 --- /dev/null +++ b/backend/migrations/20260908110914_flow_conversation_message_tool_call.down.sql @@ -0,0 +1,3 @@ +ALTER TABLE flow_conversation_message DROP COLUMN tool_arguments; +ALTER TABLE flow_conversation_message DROP COLUMN tool_result; +ALTER TABLE flow_conversation_message DROP COLUMN reasoning; diff --git a/backend/migrations/20260908110914_flow_conversation_message_tool_call.up.sql b/backend/migrations/20260908110914_flow_conversation_message_tool_call.up.sql new file mode 100644 index 0000000000..2ada36e14a --- /dev/null +++ b/backend/migrations/20260908110914_flow_conversation_message_tool_call.up.sql @@ -0,0 +1,9 @@ +-- A tool row's call and result are read back from the tool's own job, which an MCP tool +-- and a provider-native tool never have: they run inside the agent's job. For those the +-- row is the only record, so it carries the call itself. +ALTER TABLE flow_conversation_message ADD COLUMN tool_arguments TEXT; +ALTER TABLE flow_conversation_message ADD COLUMN tool_result TEXT; + +-- The thinking that produced an answer is streamed, never returned in the response body, +-- so it exists nowhere once the stream is over. +ALTER TABLE flow_conversation_message ADD COLUMN reasoning TEXT; diff --git a/backend/windmill-ai/src/providers/anthropic.rs b/backend/windmill-ai/src/providers/anthropic.rs index 486b08b739..038aef2c2e 100644 --- a/backend/windmill-ai/src/providers/anthropic.rs +++ b/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, diff --git a/backend/windmill-ai/src/providers/bedrock.rs b/backend/windmill-ai/src/providers/bedrock.rs index cf977f2484..265d491f5c 100644 --- a/backend/windmill-ai/src/providers/bedrock.rs +++ b/backend/windmill-ai/src/providers/bedrock.rs @@ -1195,6 +1195,11 @@ impl BedrockQueryBuilder { Ok(ParsedResponse::Text { content, + // The block folded for replay is also what the reader sees as thinking. + reasoning: reasoning + .as_ref() + .and_then(|r| r.reasoning_text.clone()) + .filter(|t| !t.is_empty()), tool_calls, events_str: if events_str.is_empty() { None diff --git a/backend/windmill-ai/src/providers/google_ai.rs b/backend/windmill-ai/src/providers/google_ai.rs index 21428903fa..8320ef5661 100644 --- a/backend/windmill-ai/src/providers/google_ai.rs +++ b/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, diff --git a/backend/windmill-ai/src/providers/openai.rs b/backend/windmill-ai/src/providers/openai.rs index 34185a8852..c97c52891b 100644 --- a/backend/windmill-ai/src/providers/openai.rs +++ b/backend/windmill-ai/src/providers/openai.rs @@ -538,6 +538,8 @@ impl QueryBuilder for OpenAIQueryBuilder { } else { Some(parser.accumulated_content) }, + 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, diff --git a/backend/windmill-ai/src/providers/other.rs b/backend/windmill-ai/src/providers/other.rs index bc2d8f2992..b6040d8d79 100644 --- a/backend/windmill-ai/src/providers/other.rs +++ b/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(), diff --git a/backend/windmill-ai/src/query_builder.rs b/backend/windmill-ai/src/query_builder.rs index c8f5fbb092..60ece300a3 100644 --- a/backend/windmill-ai/src/query_builder.rs +++ b/backend/windmill-ai/src/query_builder.rs @@ -33,6 +33,8 @@ pub struct BuildRequestArgs<'a> { pub enum ParsedResponse { Text { content: Option, + /// The thinking the model streamed before the answer, when it emitted any. + reasoning: Option, tool_calls: Vec, events_str: Option, annotations: Vec, diff --git a/backend/windmill-ai/src/sse.rs b/backend/windmill-ai/src/sse.rs index 96692d0173..085474a060 100644 --- a/backend/windmill-ai/src/sse.rs +++ b/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, pub events_str: String, pub stream_event_processor: Box, @@ -146,6 +148,7 @@ impl OpenAISSEParser { pub fn new(stream_event_processor: Box) -> 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, pub events_str: String, pub stream_event_processor: Box, @@ -375,6 +381,7 @@ impl AnthropicSSEParser { pub fn new(stream_event_processor: Box) -> 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, pub events_str: String, pub stream_event_processor: Box, @@ -603,6 +614,7 @@ impl GeminiSSEParser { pub fn new(stream_event_processor: Box) -> 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 }, @@ -810,6 +823,8 @@ pub enum OpenAIResponsesSSEEvent { /// OpenAI Responses API SSE Parser for streaming responses pub struct OpenAIResponsesSSEParser { 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, /// Maps item_id -> (name, call_id) for function calls tool_call_metadata: HashMap, @@ -829,6 +844,7 @@ 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(), diff --git a/backend/windmill-api-flow-conversations/src/lib.rs b/backend/windmill-api-flow-conversations/src/lib.rs index 7c0265a74d..bae25cf834 100644 --- a/backend/windmill-api-flow-conversations/src/lib.rs +++ b/backend/windmill-api-flow-conversations/src/lib.rs @@ -36,6 +36,13 @@ pub struct FlowConversationMessage { pub created_seq: i64, pub step_name: Option, pub success: bool, + /// The call behind a tool row whose tool has no job of its own — an MCP tool, or a + /// provider-native one. Read back from the job otherwise, and null here. + pub tool_arguments: Option, + pub tool_result: Option, + /// The thinking that produced an answer, streamed by the provider and stored here + /// because nothing else keeps it. + pub reasoning: Option, } /// Which conversations a listing holds. A test chat was started from the editor's test @@ -202,7 +209,7 @@ async fn list_messages( let messages = if let Some(after_seq) = query.after_seq { sqlx::query_as!( FlowConversationMessage, - r#"SELECT id, conversation_id, message_type as "message_type: MessageType", content, job_id, created_at, created_seq, step_name, success + r#"SELECT id, conversation_id, message_type as "message_type: MessageType", content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning FROM flow_conversation_message WHERE conversation_id = $1 AND created_seq > $2 @@ -219,9 +226,9 @@ async fn list_messages( // Fetch messages for this conversation, oldest first, but reverse the order of the messages for easy rendering on the frontend sqlx::query_as!( FlowConversationMessage, - r#"SELECT id, conversation_id, message_type as "message_type: MessageType", content, job_id, created_at, created_seq, step_name, success + r#"SELECT id, conversation_id, message_type as "message_type: MessageType", content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning FROM ( - SELECT id, conversation_id, message_type, content, job_id, created_at, created_seq, step_name, success + SELECT id, conversation_id, message_type, content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning FROM flow_conversation_message WHERE conversation_id = $1 ORDER BY created_seq DESC diff --git a/backend/windmill-api-jobs/src/execution.rs b/backend/windmill-api-jobs/src/execution.rs index 46da7c6246..9db302e865 100644 --- a/backend/windmill-api-jobs/src/execution.rs +++ b/backend/windmill-api-jobs/src/execution.rs @@ -676,6 +676,7 @@ pub async fn handle_chat_conversation_messages( MessageType::User, None, true, + None, ) .await?; diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 6130869edb..8be56ba15f 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -27021,6 +27021,18 @@ components: success: type: boolean description: Whether the message is a success + tool_arguments: + type: string + nullable: true + description: The arguments of a tool call whose tool has no job of its own + tool_result: + type: string + nullable: true + description: What that tool call returned + reasoning: + type: string + nullable: true + description: The thinking the model streamed before this answer EndpointTool: type: object diff --git a/backend/windmill-common/src/flow_conversations.rs b/backend/windmill-common/src/flow_conversations.rs index e3bbe23f7d..e8ba57aa0b 100644 --- a/backend/windmill-common/src/flow_conversations.rs +++ b/backend/windmill-common/src/flow_conversations.rs @@ -77,6 +77,16 @@ pub async fn get_or_create_conversation_with_id( Ok(conversation) } +/// What a row carries beyond its text, for the parts of a turn that no job holds: an +/// MCP or provider-native tool runs inside the agent's job, and thinking is streamed +/// and never returned in a response body. +#[derive(Debug, Clone, Default)] +pub struct MessageExtras { + pub tool_arguments: Option, + pub tool_result: Option, + pub reasoning: Option, +} + /// Add a message to a conversation using an existing transaction /// If the conversation doesn't exist, logs a warning and returns Ok (no error thrown) /// This allows memory_id to be used for agent memory without requiring a conversation @@ -88,6 +98,7 @@ pub async fn add_message_to_conversation_tx( message_type: MessageType, step_name: Option<&str>, success: bool, + extras: Option<&MessageExtras>, ) -> Result<()> { // Check if conversation exists first let conversation_exists = sqlx::query!( @@ -108,14 +119,17 @@ pub async fn add_message_to_conversation_tx( // Insert the message sqlx::query!( - "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id, step_name, success) - VALUES ($1, $2, $3, $4, $5, $6)", + "INSERT INTO flow_conversation_message (conversation_id, message_type, content, job_id, step_name, success, tool_arguments, tool_result, reasoning) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)", conversation_id, message_type as MessageType, content, job_id, step_name, - success + success, + extras.and_then(|e| e.tool_arguments.as_deref()), + extras.and_then(|e| e.tool_result.as_deref()), + extras.and_then(|e| e.reasoning.as_deref()) ) .execute(&mut **tx) .await?; diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 9c15229ba6..216a872ef7 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -33,7 +33,7 @@ use windmill_common::{ client::AuthedClient, db::DB, error::Error, - flow_conversations::MessageType, + flow_conversations::{MessageExtras, MessageType}, flow_status::AgentAction, flows::FlowModuleValue, worker::{to_raw_value, Connection}, @@ -235,9 +235,21 @@ async fn execute_mcp_tool_call( update_flow_status_module_with_actions_success(ctx.db, parent_job, true).await?; } - // Add tool message to conversation if chat_input_enabled + // Add tool message to conversation if chat_input_enabled. An MCP tool runs + // inside this job, so the row is the only place its call can be read back from. let content = format!("Used {} tool", tool_call.function.name); - add_tool_message_to_chat(ctx, None, &content, true).await; + add_tool_message_to_chat( + ctx, + None, + &content, + true, + Some(MessageExtras { + tool_arguments: Some(tool_call.function.arguments.clone()), + tool_result: Some(result_str), + ..Default::default() + }), + ) + .await; } Err(e) => { let error_msg = format!("MCP tool error: {}", e); @@ -272,7 +284,17 @@ async fn execute_mcp_tool_call( } // Add tool message to conversation if chat_input_enabled - add_tool_message_to_chat(ctx, None, &error_msg, false).await; + add_tool_message_to_chat( + ctx, + None, + &error_msg, + false, + Some(MessageExtras { + tool_arguments: Some(tool_call.function.arguments.clone()), + ..Default::default() + }), + ) + .await; } } @@ -681,7 +703,7 @@ async fn handle_tool_execution_error( } // Add tool message to conversation if chat_input_enabled (error case) - add_tool_message_to_chat(ctx, Some(job_id), &error_message, false).await; + add_tool_message_to_chat(ctx, Some(job_id), &error_message, false, None).await; Ok(()) } @@ -806,7 +828,7 @@ async fn handle_tool_execution_success( format!("Error executing {}", tool_call.function.name) }; - add_tool_message_to_chat(ctx, Some(job_id), &content, success).await; + add_tool_message_to_chat(ctx, Some(job_id), &content, success, None).await; Ok(()) } @@ -817,6 +839,8 @@ async fn add_tool_message_to_chat( tool_job_id: Option, content: &str, success: bool, + // Only for a tool with no job of its own; a Windmill tool's call is read from its job. + extras: Option, ) { if ctx.omit_output_from_conversation { return; @@ -852,6 +876,7 @@ async fn add_tool_message_to_chat( MessageType::Tool, &step_name, success, + extras.as_ref(), ) .await { diff --git a/backend/windmill-worker/src/ai/utils.rs b/backend/windmill-worker/src/ai/utils.rs index 6e80551650..878cfe1fd8 100644 --- a/backend/windmill-worker/src/ai/utils.rs +++ b/backend/windmill-worker/src/ai/utils.rs @@ -13,7 +13,7 @@ use windmill_common::flows::FlowModuleValue; use windmill_common::{ db::DB, error::Error, - flow_conversations::{add_message_to_conversation_tx, MessageType}, + flow_conversations::{add_message_to_conversation_tx, MessageType, MessageExtras}, flow_status::AgentAction, flows::{InputTransform, Step}, jobs::JobKind, @@ -209,6 +209,7 @@ pub async fn add_message_to_conversation( message_type: MessageType, step_name: &Option, success: bool, + extras: Option<&MessageExtras>, ) -> Result<(), Error> { let mut tx = db.begin().await?; add_message_to_conversation_tx( @@ -219,6 +220,7 @@ pub async fn add_message_to_conversation( message_type, step_name.as_deref(), success, + extras, ) .await?; tx.commit().await?; diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index d948fee088..ffeea92882 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -40,7 +40,7 @@ use windmill_common::{ client::AuthedClient, db::DB, error::{self, Error}, - flow_conversations::MessageType, + flow_conversations::{MessageExtras, MessageType}, flow_status::AgentAction, flows::{AgentTool, FlowModule, FlowModuleValue, InputTransform, ToolValue}, get_latest_hash_for_path, @@ -1355,6 +1355,7 @@ pub async fn run_agent( match parsed { ParsedResponse::Text { content: response_content, + reasoning: response_reasoning, tool_calls, events_str, annotations, @@ -1394,6 +1395,13 @@ pub async fn run_agent( let db_clone = db.clone(); let message_content = "Used websearch tool successfully".to_string(); let step_name = step_name.clone(); + // The search ran inside the provider's call, so this job's args + // describe the agent, not the search: its sources reach the row + // only if they are written here. + let extras = (!annotations.is_empty()).then(|| MessageExtras { + tool_result: serde_json::to_string(&annotations).ok(), + ..Default::default() + }); tokio::spawn(async move { if let Err(e) = add_message_to_conversation( &db_clone, @@ -1403,6 +1411,7 @@ pub async fn run_agent( MessageType::Tool, &step_name, true, + extras.as_ref(), ) .await { @@ -1446,6 +1455,14 @@ pub async fn run_agent( let db_clone = db.clone(); let message_content = response_content.clone(); let step_name = step_name.clone(); + // The thinking is streamed and never returned in a response + // body, so the answer's row is the only place it can be kept. + let extras = response_reasoning + .clone() + .map(|reasoning| MessageExtras { + reasoning: Some(reasoning), + ..Default::default() + }); // Spawn task because we do not need to wait for the result tokio::spawn(async move { @@ -1457,6 +1474,7 @@ pub async fn run_agent( MessageType::Assistant, &step_name, true, + extras.as_ref(), ) .await { @@ -1471,6 +1489,44 @@ pub async fn run_agent( } } + // An iteration that answered with tool calls has no message row to carry its + // thinking, and the next iteration's row holds only its own. Stored on a row + // of its own so a reader sees what led to the call. + if persist_output_to_conversation + && response_content.as_deref().unwrap_or("").is_empty() + { + if let (Some(memory_id), Some(reasoning)) = + (memory_id, response_reasoning.clone()) + { + let agent_job_id = job.id; + let db_clone = db.clone(); + let step_name = step_name.clone(); + tokio::spawn(async move { + if let Err(e) = add_message_to_conversation( + &db_clone, + &memory_id, + Some(agent_job_id), + "", + MessageType::Assistant, + &step_name, + true, + Some(&MessageExtras { + reasoning: Some(reasoning), + ..Default::default() + }), + ) + .await + { + tracing::warn!( + "Failed to add reasoning message to conversation {}: {}", + memory_id, + e + ); + } + }); + } + } + if tool_calls.is_empty() { break; } else if i == max_iterations - 1 { @@ -1584,6 +1640,7 @@ pub async fn run_agent( MessageType::Assistant, &step_name, true, + None, ) .await { diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 34f46307cf..7c940dfcaa 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -2254,6 +2254,7 @@ async fn add_tool_message_to_conversation( MessageType::Assistant, None, success, + None, ) .await?; tx.commit().await?; diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts index be0ced56e2..4d201c1f66 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts @@ -91,7 +91,7 @@ import { copilotInfo } from '$lib/aiStore' import { copilotWorkspaceRequested, loadCopilot } from '$lib/components/copilot/loadCopilot' import { askTools, prepareAskSystemMessage, prepareAskUserMessage } from './ask/core' import { readDocsPageTool, searchDocsTool } from './docs/core' -import { TypewriterReveal } from './typewriterReveal' +import { prefersInstantReveal, TypewriterReveal } from './typewriterReveal' import { chatState, DEFAULT_SIZE, triggerablesByAi } from './sharedChatState.svelte' import { createAppBackendRunnableContextElement, @@ -151,11 +151,6 @@ import { appendAttachedFilesRoster } from './files/fileTools' import { ENTER_PLAN_MODE_TOOL, EXIT_PLAN_MODE_TOOL } from './planMode' import { PlanModeController, type PlanModeHost } from './planModeController.svelte' -// SSR and users who prefer reduced motion get no typewriter pacing. -function prefersInstantReveal(): boolean { - return !BROWSER || (window.matchMedia?.('(prefers-reduced-motion: reduce)').matches ?? false) -} - // Compaction of the stored history: once the projected request size // (contextTokens — the provider's report when current, a fresh chars/4 // estimate otherwise — plus the new user message) reaches the trigger ratio of diff --git a/frontend/src/lib/components/copilot/chat/AssistantMessage.svelte b/frontend/src/lib/components/copilot/chat/AssistantMessage.svelte index 47020750b2..69d2505eaa 100644 --- a/frontend/src/lib/components/copilot/chat/AssistantMessage.svelte +++ b/frontend/src/lib/components/copilot/chat/AssistantMessage.svelte @@ -32,6 +32,15 @@ const runHref = $derived( jobId ? `${base}/run/${jobId}?workspace=${$workspaceStore}` : undefined ) + // Today's answers show the time alone; the day earns its place only on a conversation + // read back later. Resolved at render, so a chat left open across midnight keeps + // yesterday's format until it is reopened. + const timestamp = $derived.by(() => { + if (!createdAt) return undefined + const at = new Date(createdAt) + const today = new Date().toDateString() === at.toDateString() + return displayDate(at, false, !today) + }) const reasoning = $derived( message.role === 'assistant' ? message.reasoning?.trim() || undefined : undefined @@ -150,17 +159,20 @@ {/if} -{#if message.content || createdAt || runHref} +{#if message.content} + does not nudge the message below, and with no margin of its own so it sits in the + gap the transcript already leaves between messages. A row carrying only thinking + has no answer to copy or date, and the run behind it is the one the next row + already links. -->
{#if message.content} {/if} - {#if createdAt} - {displayDate(createdAt)} + {#if timestamp} + {timestamp} {/if} {#if runHref} void) => unknown type Cancel = (handle: unknown) => void diff --git a/frontend/src/lib/components/flows/content/FlowInput.svelte b/frontend/src/lib/components/flows/content/FlowInput.svelte index 489cb9001f..c89210e441 100644 --- a/frontend/src/lib/components/flows/content/FlowInput.svelte +++ b/frontend/src/lib/components/flows/content/FlowInput.svelte @@ -665,6 +665,17 @@ applied.push('context memory set to 10') } + // Without streaming the chat has no SSE to read, so a turn shows nothing — + // no thinking, no answer — until the run ends and its rows are written. + if ( + isUnconfigured(value.input_transforms['streaming']) || + (value.input_transforms['streaming']?.type === 'static' && + value.input_transforms['streaming'].value === false) + ) { + value.input_transforms['streaming'] = { type: 'static', value: true } + applied.push('streaming turned on') + } + sendUserToast( applied.length > 0 ? `Chat mode enabled. AI agent configured with ${applied.join(' and ')}.` diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts index 5b467ca1ed..2c81cda941 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts @@ -12,18 +12,21 @@ import { workspaceStore, userStore } from '$lib/stores' import { get } from 'svelte/store' import { parseStreamEvents, toolSummary } from '$lib/components/chat/utils' import { randomUUID } from '$lib/utils/uuid' +import { + prefersInstantReveal, + TypewriterReveal +} from '$lib/components/copilot/chat/typewriterReveal' export interface ChatMessage extends FlowConversationMessage { loading?: boolean streaming?: boolean /** - * The call behind a tool row, as the stream reports it. Local to a running turn: the - * server stores only the summary sentence, and once the run settles the same details - * are read back from the tool's own job instead (see toolCallContext). + * The tool a row's call belongs to, as the stream reports it. Local to a running turn: + * afterwards the name comes from the summary the server stored, and the call itself + * from the tool's own job (see toolCallContext) or from the row's own + * `tool_arguments` / `tool_result` when the tool had no job. */ tool_name?: string - tool_arguments?: string - tool_result?: string } export interface ConversationWithDraft extends FlowConversation { @@ -51,6 +54,71 @@ export class FlowChatManager { conversations = $state([]) deletingConversationId = $state(undefined) isSidebarExpanded = $state(false) + /** The thinking of the turn in flight, until it is attached to the answer it produced. */ + currentReasoning = $state('') + /** The model is reasoning: true from the first thinking token until the answer starts. */ + isReasoningActive = $state(false) + + // The row the stream is currently writing into, and the text revealed so far. Fields + // rather than locals of the stream handler: the typewriter reveals on animation + // frames, long after the chunk that delivered the text was applied. + #streamConversationId = '' + #streamAssistantId = '' + #streamContent = '' + + // The worker's events reach us in bursts — the provider batches tokens, and the SSE + // endpoint ships whatever accumulated — so display is paced separately from arrival, + // exactly as the session chat does it. Answer and thinking pace independently. + #replyReveal = new TypewriterReveal({ + onReveal: (chunk) => { + this.#streamContent += chunk + this.#upsertStreamedAssistantRow() + }, + instant: prefersInstantReveal() + }) + #reasoningReveal = new TypewriterReveal({ + onReveal: (chunk) => { + this.currentReasoning += chunk + this.#upsertStreamedAssistantRow() + }, + instant: prefersInstantReveal() + }) + + /** Create or update the row holding the turn's answer and the thinking before it. */ + #upsertStreamedAssistantRow() { + const reasoning = this.currentReasoning === '' ? undefined : this.currentReasoning + if (this.#streamContent === '' && reasoning === undefined) return + if (this.#streamAssistantId === '') { + this.#streamAssistantId = 'temp-' + randomUUID() + this.messages = [ + ...this.messages, + { + id: this.#streamAssistantId, + content: this.#streamContent, + created_at: new Date().toISOString(), + created_seq: 0, + message_type: 'assistant', + conversation_id: this.#streamConversationId, + job_id: '', + loading: false, + streaming: true, + reasoning + } + ] + } else { + this.messages = this.messages.map((msg) => + msg.id === this.#streamAssistantId + ? { ...msg, content: this.#streamContent, reasoning } + : msg + ) + } + } + + /** Reveal everything buffered now, so the row is whole before the turn moves on. */ + #flushReveals() { + this.#replyReveal.flush() + this.#reasoningReveal.flush() + } /** * Which conversations the list holds. The editor shows its own test chats, since * testing is what happens there; a deployed flow shows the chats its users started, @@ -102,6 +170,12 @@ export class FlowChatManager { } cleanup() { + this.#replyReveal.reset() + this.#reasoningReveal.reset() + this.#streamAssistantId = '' + this.#streamContent = '' + this.currentReasoning = '' + this.isReasoningActive = false if (this.currentEventSource) { this.currentEventSource.close() this.currentEventSource = undefined @@ -574,8 +648,11 @@ export class FlowChatManager { this.#toolMessageIds.clear() // Track stream state for this message - let accumulatedContent = '' - let assistantMessageId = '' + this.#streamConversationId = currentConversationId + this.#streamAssistantId = '' + this.#streamContent = '' + this.#replyReveal.reset() + this.#reasoningReveal.reset() let isCompleted = false try { @@ -654,10 +731,16 @@ export class FlowChatManager { // chunk holding a call and its result must produce both. for (const event of parseStreamEvents(data.new_result_stream)) { if (event.kind === 'tool_call' || event.kind === 'tool_execution') { + // Whatever is still buffered belongs to the row before the tool — + // thinking that led straight to the call included, which is why this + // runs before the reset below. + this.#flushReveals() + this.currentReasoning = '' + this.isReasoningActive = false // The assistant text so far is finished; the tool row follows it. this.#settleStreamingMessage() - assistantMessageId = '' - accumulatedContent = '' + this.#streamAssistantId = '' + this.#streamContent = '' this.#upsertToolMessage(currentConversationId, event.callId, { tool_name: event.name, content: `Running ${event.name}`, @@ -676,33 +759,11 @@ export class FlowChatManager { success: event.success, loading: false }) + } else if (event.kind === 'reasoning') { + this.isReasoningActive = true + this.#reasoningReveal.push(event.content) } else if (event.kind === 'token') { - accumulatedContent += event.content - } - } - // The assistant's own text is one growing message until a tool - // interrupts it, which is what resets the id above. - if (accumulatedContent.length > 0) { - if (assistantMessageId.length === 0) { - assistantMessageId = 'temp-' + randomUUID() - this.messages = [ - ...this.messages, - { - id: assistantMessageId, - content: accumulatedContent, - created_at: new Date().toISOString(), - created_seq: 0, - message_type: 'assistant', - conversation_id: currentConversationId, - job_id: '', - loading: false, - streaming: true - } - ] - } else { - this.messages = this.messages.map((msg) => - msg.id === assistantMessageId ? { ...msg, content: accumulatedContent } : msg - ) + this.#replyReveal.push(event.content) } } } @@ -710,6 +771,8 @@ export class FlowChatManager { // Handle completion if (data.completed) { isCompleted = true + // Anything still buffered would be dropped by the temp-row sweep below. + this.#flushReveals() // Do a final poll to get all messages from database if (this.selectedConversationId) { await this.pollConversationMessages(this.selectedConversationId, { diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts index 4541e85854..833cdd3fc8 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts @@ -114,6 +114,7 @@ function toDisplayMessage( role: 'assistant', content: message.content, streaming: message.streaming, + reasoning: message.reasoning ?? undefined, stepName: showStepNames ? (message.step_name ?? undefined) : undefined, // The run behind the answer, so a reader can open what produced it. Absent on // the temp message a stream builds, which has no job id until it settles. @@ -205,8 +206,9 @@ export class FlowChatViewHost implements ChatViewHost { loadingLabel = undefined compacting = false currentReply = '' - currentReasoning = '' - currentReasoningActive = false + // The turn's thinking while it streams; it moves onto the answer once that starts. + currentReasoning = $derived.by(() => this.#manager.currentReasoning) + currentReasoningActive = $derived.by(() => this.#manager.isReasoningActive) reasoningHiddenIndicatorLabel = undefined #automaticScroll = $state(true)