From 6ce5e7f89c35d9e2ed14a36d62e846d496bc1622 Mon Sep 17 00:00:00 2001 From: Guilhem Lemouel Date: Tue, 8 Sep 2026 18:03:43 +0200 Subject: [PATCH] feat(ai-chat): keep an MCP tool's call and the model's thinking on the conversation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A tool row's call is read back from the tool's own job, which an MCP tool and a provider-native one never have — they run inside the agent's job — so those rows could only ever be a summary line. Thinking had nowhere at all: it is streamed and never returned in a response body. `flow_conversation_message` now carries `tool_arguments`, `tool_result` and `reasoning`, written for exactly the parts of a turn no job holds: an MCP call, a web search's sources, and the thinking behind an answer or behind a tool call. `ParsedResponse::Text` gains the reasoning the parsers already had in hand; each SSE parser accumulates it beside the answer text. Flow chat renders both through the components the session chat uses, and paces them with the same `TypewriterReveal`: the worker's events arrive in bursts, so display is decoupled from arrival — measured over one 1000-character answer, 54 growth steps of ~18 characters where it was 15 of ~54. The row is built from the reveal rather than once per chunk, which is also what lets thinking survive arriving in the same chunk as the tool call that ends it. Enabling chat mode now turns streaming on, next to the memory it already sets: without it the chat has no stream to read and a turn shows nothing until it ends. The answer footer renders only under an actual answer, carries no margin of its own, and drops the day from today's timestamps. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01QN7VboDEm9HAB1t4sMxMdE --- ...12adaeeead8b39b007a32badadf67ec5ca17.json} | 24 +++- ...9df7ede61e050d3e634b795aaf7f60b3b05d.json} | 9 +- ...abd9633bbe5219bfd9683f336de19763cb58.json} | 24 +++- ...ow_conversation_message_tool_call.down.sql | 3 + ...flow_conversation_message_tool_call.up.sql | 9 ++ .../windmill-ai/src/providers/anthropic.rs | 2 + backend/windmill-ai/src/providers/bedrock.rs | 5 + .../windmill-ai/src/providers/google_ai.rs | 2 + backend/windmill-ai/src/providers/openai.rs | 2 + backend/windmill-ai/src/providers/other.rs | 2 + backend/windmill-ai/src/query_builder.rs | 2 + backend/windmill-ai/src/sse.rs | 16 +++ .../src/lib.rs | 13 +- backend/windmill-api-jobs/src/execution.rs | 1 + backend/windmill-api/openapi.yaml | 12 ++ .../windmill-common/src/flow_conversations.rs | 20 ++- backend/windmill-worker/src/ai/tools.rs | 37 ++++- backend/windmill-worker/src/ai/utils.rs | 4 +- backend/windmill-worker/src/ai_executor.rs | 59 +++++++- backend/windmill-worker/src/worker_flow.rs | 1 + .../copilot/chat/AIChatManager.svelte.ts | 7 +- .../copilot/chat/AssistantMessage.svelte | 22 ++- .../copilot/chat/typewriterReveal.ts | 7 + .../components/flows/content/FlowInput.svelte | 11 ++ .../conversations/FlowChatManager.svelte.ts | 133 +++++++++++++----- .../conversations/flowChatViewHost.svelte.ts | 6 +- 26 files changed, 362 insertions(+), 71 deletions(-) rename backend/.sqlx/{query-1c3473a0f9f6b6148b2c975f9f05bdefedf8a51c4e6ddf0eca367b9cc778d051.json => query-318ed7a45d8326e3ecbc9727ee3812adaeeead8b39b007a32badadf67ec5ca17.json} (69%) rename backend/.sqlx/{query-b1a9a433e577133869c067b2ce383fc6ce4e9df307feb5fd3edc0d1276d61ff1.json => query-55dbe12954489532644b91e67b2a9df7ede61e050d3e634b795aaf7f60b3b05d.json} (63%) rename backend/.sqlx/{query-e8802be9203c1e88a06e337260ccca029380139f89a01a89033e36a6ed9ac082.json => query-d6c65ce1d443d1e1783818d36c7eabd9633bbe5219bfd9683f336de19763cb58.json} (62%) create mode 100644 backend/migrations/20260908110914_flow_conversation_message_tool_call.down.sql create mode 100644 backend/migrations/20260908110914_flow_conversation_message_tool_call.up.sql 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)