diff --git a/backend/.sqlx/query-b1a9a433e577133869c067b2ce383fc6ce4e9df307feb5fd3edc0d1276d61ff1.json b/backend/.sqlx/query-12329c3359a7944ab5fa3aa27ddca1b26f340ccf574b9fa07641fe88b2d2987c.json similarity index 61% rename from backend/.sqlx/query-b1a9a433e577133869c067b2ce383fc6ce4e9df307feb5fd3edc0d1276d61ff1.json rename to backend/.sqlx/query-12329c3359a7944ab5fa3aa27ddca1b26f340ccf574b9fa07641fe88b2d2987c.json index 1818efc0c0..4c120699aa 100644 --- a/backend/.sqlx/query-b1a9a433e577133869c067b2ce383fc6ce4e9df307feb5fd3edc0d1276d61ff1.json +++ b/backend/.sqlx/query-12329c3359a7944ab5fa3aa27ddca1b26f340ccf574b9fa07641fe88b2d2987c.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, attachments)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)", "describe": { "columns": [], "parameters": { @@ -21,10 +21,14 @@ "Text", "Uuid", "Varchar", - "Bool" + "Bool", + "Text", + "Text", + "Text", + "Jsonb" ] }, "nullable": [] }, - "hash": "b1a9a433e577133869c067b2ce383fc6ce4e9df307feb5fd3edc0d1276d61ff1" + "hash": "12329c3359a7944ab5fa3aa27ddca1b26f340ccf574b9fa07641fe88b2d2987c" } diff --git a/backend/.sqlx/query-e8802be9203c1e88a06e337260ccca029380139f89a01a89033e36a6ed9ac082.json b/backend/.sqlx/query-a4a823f70b3dbe6aaf4a61c98345e94c5042fd5e6351fea139a66ecb1fb812ab.json similarity index 59% rename from backend/.sqlx/query-e8802be9203c1e88a06e337260ccca029380139f89a01a89033e36a6ed9ac082.json rename to backend/.sqlx/query-a4a823f70b3dbe6aaf4a61c98345e94c5042fd5e6351fea139a66ecb1fb812ab.json index a3374d6cdf..9ec8c1e2d5 100644 --- a/backend/.sqlx/query-e8802be9203c1e88a06e337260ccca029380139f89a01a89033e36a6ed9ac082.json +++ b/backend/.sqlx/query-a4a823f70b3dbe6aaf4a61c98345e94c5042fd5e6351fea139a66ecb1fb812ab.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, attachments\n FROM (\n SELECT id, conversation_id, message_type, content, job_id, created_at, created_seq, step_name, success, tool_arguments, tool_result, reasoning, attachments\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,26 @@ "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" + }, + { + "ordinal": 12, + "name": "attachments", + "type_info": "Jsonb" } ], "parameters": { @@ -76,8 +96,12 @@ false, false, true, - false + false, + true, + true, + true, + true ] }, - "hash": "e8802be9203c1e88a06e337260ccca029380139f89a01a89033e36a6ed9ac082" + "hash": "a4a823f70b3dbe6aaf4a61c98345e94c5042fd5e6351fea139a66ecb1fb812ab" } diff --git a/backend/.sqlx/query-1c3473a0f9f6b6148b2c975f9f05bdefedf8a51c4e6ddf0eca367b9cc778d051.json b/backend/.sqlx/query-d6fa78c43b6c5f8040d7bccb29ad8627be1dac6fbe0097735a52f47c173f51c9.json similarity index 65% rename from backend/.sqlx/query-1c3473a0f9f6b6148b2c975f9f05bdefedf8a51c4e6ddf0eca367b9cc778d051.json rename to backend/.sqlx/query-d6fa78c43b6c5f8040d7bccb29ad8627be1dac6fbe0097735a52f47c173f51c9.json index fb27bd9446..baf0e51d4f 100644 --- a/backend/.sqlx/query-1c3473a0f9f6b6148b2c975f9f05bdefedf8a51c4e6ddf0eca367b9cc778d051.json +++ b/backend/.sqlx/query-d6fa78c43b6c5f8040d7bccb29ad8627be1dac6fbe0097735a52f47c173f51c9.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, attachments\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,26 @@ "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" + }, + { + "ordinal": 12, + "name": "attachments", + "type_info": "Jsonb" } ], "parameters": { @@ -76,8 +96,12 @@ false, false, true, - false + false, + true, + true, + true, + true ] }, - "hash": "1c3473a0f9f6b6148b2c975f9f05bdefedf8a51c4e6ddf0eca367b9cc778d051" + "hash": "d6fa78c43b6c5f8040d7bccb29ad8627be1dac6fbe0097735a52f47c173f51c9" } diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 67e37830fb..b2b9d65c97 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -14896,6 +14896,7 @@ dependencies = [ "eventsource-stream", "futures", "http 1.5.0", + "indexmap 2.14.2", "lazy_static", "mime_guess", "reqwest 0.13.5", diff --git a/backend/migrations/20260916144653_flow_conversation_message_tool_call.down.sql b/backend/migrations/20260916144653_flow_conversation_message_tool_call.down.sql new file mode 100644 index 0000000000..56f20ba722 --- /dev/null +++ b/backend/migrations/20260916144653_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/20260916144653_flow_conversation_message_tool_call.up.sql b/backend/migrations/20260916144653_flow_conversation_message_tool_call.up.sql new file mode 100644 index 0000000000..a6f7822246 --- /dev/null +++ b/backend/migrations/20260916144653_flow_conversation_message_tool_call.up.sql @@ -0,0 +1,12 @@ +-- A chat is rebuilt from its rows without reading jobs, so every tool row carries its call: +-- the arguments the model wrote and the text the model got back, or what the call failed +-- with. A script or flow tool's job holds the args its input transforms produced, not the +-- model's; an MCP tool runs inside the agent's job, whose result lists every call of the +-- turn with nothing tying one to a row. A provider-native web search carries only its +-- citations, the provider never returning the query. +ALTER TABLE flow_conversation_message ADD COLUMN tool_arguments TEXT; +ALTER TABLE flow_conversation_message ADD COLUMN tool_result TEXT; + +-- The thinking behind this row. The agent job keeps the turn's thinking as one string; +-- the rows keep it per iteration, next to the answer or tool call it led to. +ALTER TABLE flow_conversation_message ADD COLUMN reasoning TEXT; diff --git a/backend/migrations/20260917074040_flow_conversation_message_attachments.down.sql b/backend/migrations/20260917074040_flow_conversation_message_attachments.down.sql new file mode 100644 index 0000000000..01d81a9719 --- /dev/null +++ b/backend/migrations/20260917074040_flow_conversation_message_attachments.down.sql @@ -0,0 +1 @@ +ALTER TABLE flow_conversation_message DROP COLUMN attachments; diff --git a/backend/migrations/20260917074040_flow_conversation_message_attachments.up.sql b/backend/migrations/20260917074040_flow_conversation_message_attachments.up.sql new file mode 100644 index 0000000000..dec5365f89 --- /dev/null +++ b/backend/migrations/20260917074040_flow_conversation_message_attachments.up.sql @@ -0,0 +1,4 @@ +-- The files a user message carried, as object-storage references: `[{input, s3, storage?, +-- filename?}]`. Only references, never file bytes and never a presigned URL, so a +-- transcript can show a message's files without reading its run's args. +ALTER TABLE flow_conversation_message ADD COLUMN attachments JSONB; diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index cab1d7fe68..e27cb2893c 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -101,7 +101,7 @@ flow: workspace_id(char), path(char), summary(text), description(text), value(js FK: (workspace_id) -> workspace(id) flow_conversation: id(uuid), workspace_id(char), flow_path(char), title(char), created_at(ts), updated_at(ts), created_by(char), is_test(bool) FK: (workspace_id) -> workspace(id) -flow_conversation_message: id(uuid), conversation_id(uuid), message_type(message_type), content(text), job_id(uuid), created_at(ts), created_seq(int8), step_name(char), success(bool) +flow_conversation_message: id(uuid), conversation_id(uuid), message_type(message_type), content(text), job_id(uuid), created_at(ts), created_seq(int8), step_name(char), success(bool), tool_arguments(text), tool_result(text), reasoning(text), attachments(jsonb) FK: (conversation_id) -> flow_conversation(id) | (job_id) -> v2_job(id) flow_iterator_data: job_id(uuid), itered(jsonb) flow_node: id(bigint), workspace_id(char), hash(bigint), path(char), lock(text), code(text), flow(jsonb), hash_v2(char(64)) diff --git a/backend/tests/v2_job_delete_orphans.rs b/backend/tests/v2_job_delete_orphans.rs index b45fa4201f..2b11deab77 100644 --- a/backend/tests/v2_job_delete_orphans.rs +++ b/backend/tests/v2_job_delete_orphans.rs @@ -269,6 +269,7 @@ async fn test_new_turns_wait_for_conversation_cleanup_and_recreate( windmill_common::flow_conversations::MessageType::User, None, true, + None, ) .await?; tx.commit().await?; diff --git a/backend/windmill-ai/Cargo.toml b/backend/windmill-ai/Cargo.toml index 0276240246..98ea87a4e3 100644 --- a/backend/windmill-ai/Cargo.toml +++ b/backend/windmill-ai/Cargo.toml @@ -23,6 +23,7 @@ async-trait.workspace = true async-stream.workspace = true base64.workspace = true bytes.workspace = true +indexmap.workspace = true eventsource-stream.workspace = true futures.workspace = true http.workspace = true diff --git a/backend/windmill-ai/src/providers/bedrock.rs b/backend/windmill-ai/src/providers/bedrock.rs index adee6e5053..412e057396 100644 --- a/backend/windmill-ai/src/providers/bedrock.rs +++ b/backend/windmill-ai/src/providers/bedrock.rs @@ -1074,7 +1074,8 @@ impl BedrockQueryBuilder { let mut accumulated_text = String::new(); let mut events_str = String::new(); - let mut accumulated_tool_calls: HashMap = HashMap::new(); + let mut accumulated_tool_calls: indexmap::IndexMap = + indexmap::IndexMap::new(); let mut current_tool_use_id: Option = None; let mut usage: Option = None; // Claude reasoning block for the turn (only populated when thinking is on), @@ -1263,7 +1264,10 @@ mod tests { // recovers the uncached share by subtracting the details back out. assert_eq!(usage["usage"]["prompt_tokens"], 1010); assert_eq!(usage["usage"]["completion_tokens"], 7); - assert_eq!(usage["usage"]["prompt_tokens_details"]["cached_tokens"], 900); + assert_eq!( + usage["usage"]["prompt_tokens_details"]["cached_tokens"], + 900 + ); assert_eq!( usage["usage"]["prompt_tokens_details"]["cache_write_tokens"], 100 diff --git a/backend/windmill-ai/src/sse.rs b/backend/windmill-ai/src/sse.rs index 054baeb5de..57acd66738 100644 --- a/backend/windmill-ai/src/sse.rs +++ b/backend/windmill-ai/src/sse.rs @@ -1,6 +1,7 @@ use std::collections::HashMap; use eventsource_stream::Eventsource; +use indexmap::IndexMap; use reqwest::Response; use serde::Deserialize; use tokio_stream::StreamExt; @@ -137,7 +138,9 @@ 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, + // Insertion-ordered in every parser: tool calls run and are persisted in the order the + // stream showed them, and a chat attaches a round's thinking to its first call. + pub accumulated_tool_calls: IndexMap, pub events_str: String, pub stream_event_processor: Box, /// Token usage from final chunk (when stream_options.include_usage is true) @@ -149,7 +152,7 @@ impl OpenAISSEParser { Self { accumulated_content: String::new(), accumulated_reasoning: String::new(), - accumulated_tool_calls: HashMap::new(), + accumulated_tool_calls: IndexMap::new(), events_str: String::new(), stream_event_processor, usage: None, @@ -359,7 +362,7 @@ 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 accumulated_tool_calls: IndexMap, pub events_str: String, pub stream_event_processor: Box, /// Track content block types by index @@ -382,7 +385,7 @@ impl AnthropicSSEParser { Self { accumulated_content: String::new(), accumulated_reasoning: String::new(), - accumulated_tool_calls: HashMap::new(), + accumulated_tool_calls: IndexMap::new(), events_str: String::new(), stream_event_processor, content_blocks: HashMap::new(), @@ -601,7 +604,7 @@ 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 accumulated_tool_calls: IndexMap, pub events_str: String, pub stream_event_processor: Box, tool_call_index: i64, @@ -615,7 +618,7 @@ impl GeminiSSEParser { Self { accumulated_content: String::new(), accumulated_reasoning: String::new(), - accumulated_tool_calls: HashMap::new(), + accumulated_tool_calls: IndexMap::new(), events_str: String::new(), stream_event_processor, tool_call_index: 0, @@ -833,7 +836,7 @@ pub struct OpenAIResponsesSSEParser { pub accumulated_content: String, /// The reasoning summary streamed before the answer, kept so it can be stored with it. pub accumulated_reasoning: String, - pub accumulated_tool_calls: HashMap, + pub accumulated_tool_calls: IndexMap, /// Maps item_id -> (name, call_id) for function calls tool_call_metadata: HashMap, /// Maps item_id -> accumulated arguments @@ -855,7 +858,7 @@ impl OpenAIResponsesSSEParser { Self { accumulated_content: String::new(), accumulated_reasoning: String::new(), - accumulated_tool_calls: HashMap::new(), + accumulated_tool_calls: IndexMap::new(), tool_call_metadata: HashMap::new(), tool_call_arguments: HashMap::new(), events_str: String::new(), diff --git a/backend/windmill-api-flow-conversations/src/lib.rs b/backend/windmill-api-flow-conversations/src/lib.rs index 80eebfeda9..0ec295c23f 100644 --- a/backend/windmill-api-flow-conversations/src/lib.rs +++ b/backend/windmill-api-flow-conversations/src/lib.rs @@ -37,6 +37,19 @@ pub struct FlowConversationMessage { pub created_seq: i64, pub step_name: Option, pub success: bool, + /// On a tool row, the arguments the model wrote. For a Windmill tool these exclude the + /// inputs its step wires in. Null for a web search, whose query the provider does not + /// return. + pub tool_arguments: Option, + /// On a tool row, the text the model got back, or what the call failed with; a web + /// search's citations. + pub tool_result: Option, + /// On an answer, the thinking that produced it; on a tool row, the thinking that led to + /// the call. The agent job keeps the turn's thinking as one string. + pub reasoning: Option, + /// The files a user message carried, as object-storage references + /// (`[{input, s3, storage?, filename?}]`). + pub attachments: Option, } /// Which conversations a listing holds. A test chat was started from the editor's test @@ -247,7 +260,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, attachments FROM flow_conversation_message WHERE conversation_id = $1 AND created_seq > $2 @@ -264,9 +277,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, attachments 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, attachments 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 a6039ca2e4..7184f8ecec 100644 --- a/backend/windmill-api-jobs/src/execution.rs +++ b/backend/windmill-api-jobs/src/execution.rs @@ -26,7 +26,9 @@ use windmill_api_auth::{check_scopes, get_scope_tags, ApiAuthed}; use windmill_common::{ db::{UserDB, UserDbWithAuthed}, error::{self, Error}, - flow_conversations::{add_message_to_conversation_tx, MessageType}, + flow_conversations::{ + add_message_to_conversation_tx, message_attachments, MessageExtras, MessageType, + }, get_latest_flow_version_info_for_path, jobs::{ check_tag_available_for_workspace_internal, format_result, script_path_to_payload, @@ -672,6 +674,8 @@ pub async fn handle_chat_conversation_messages( user_message_raw: Option<&Box>, job_id: Uuid, is_test: bool, + // The run's args, for the files the message carried. + args: &HashMap>, ) -> error::Result<()> { // Names the query parameter rather than the field: it is not a flow argument, and // supplying it as one is the first thing tried on reading `memory_id is required`. @@ -708,8 +712,8 @@ pub async fn handle_chat_conversation_messages( ) .await?; - // The run this message started. Its args are the only record of what the message - // carried besides its text — attachments and every other flow input — and nothing + // The run this message started. The row keeps the files the message carried as + // references; its args are the only record of every other flow input, and nothing // written later points at them: an assistant row holds the AI agent step's job. add_message_to_conversation_tx( tx, @@ -719,6 +723,7 @@ pub async fn handle_chat_conversation_messages( MessageType::User, None, true, + Some(&MessageExtras { attachments: message_attachments(args), ..Default::default() }), ) .await?; @@ -841,6 +846,7 @@ pub async fn run_flow<'c>( args.args.get("user_message"), uuid, false, + &args.args, ) .await?; } diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 60b3c2ad25..8447666a94 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -28402,6 +28402,50 @@ components: success: type: boolean description: Whether the message is a success + tool_arguments: + type: string + nullable: true + description: >- + On a tool row, the arguments the model wrote for the call. For a script, flow or + AI agent tool these exclude the inputs its step wires in, which only the tool's + job holds. Null for a provider-native web search, whose query the provider does + not return. + tool_result: + type: string + nullable: true + description: >- + On a tool row, the text the model got back from the call, or what the call + failed with — the row's own text names the tool rather than the reason. For a + provider-native web search, its citations. + reasoning: + type: string + nullable: true + description: >- + On an answer, the thinking that produced it; on a tool row, the thinking that + led to the call. Each round's thinking is on one row. The agent job's result + keeps the turn's thinking as a single string. + attachments: + type: array + nullable: true + description: >- + The files a user message carried, as object-storage references: every flow + input other than user_message that held one or a list of them, at most 20. Never + file bytes or a presigned URL. + items: + type: object + required: [input, s3] + properties: + input: + type: string + description: The flow input that held the file + s3: + type: string + description: The file's key in object storage + storage: + type: string + description: The secondary storage holding the file, absent for the primary one + filename: + type: string EndpointTool: type: object diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 41a42d58d8..4149292200 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -9556,6 +9556,7 @@ async fn run_preview_flow_job( uuid, // Run from the editor's test panel: a trial, not a real conversation. true, + &flow_args, ) .await?; } diff --git a/backend/windmill-common/src/flow_conversations.rs b/backend/windmill-common/src/flow_conversations.rs index 1abe82e5a7..91bfa4a39f 100644 --- a/backend/windmill-common/src/flow_conversations.rs +++ b/backend/windmill-common/src/flow_conversations.rs @@ -1,7 +1,11 @@ +use std::collections::HashMap; + use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; +use serde_json::value::RawValue; use sqlx::{self, FromRow}; use uuid::Uuid; +use windmill_types::s3::S3Object; use crate::db::DB; use crate::error::Result; @@ -139,6 +143,65 @@ async fn lock_conversation( .await?) } +/// What a row carries beyond its text. A chat is rebuilt from its rows alone, without +/// reading jobs, so every tool row carries the model's call and what the model got back: a +/// Windmill tool's job holds the args its input transforms produced rather than the model's, +/// and an MCP tool's call sits among every call of the turn in the agent's job. That job's +/// `reasoning` is one string for the whole turn, where the rows keep it per iteration. +#[derive(Debug, Clone, Default)] +pub struct MessageExtras { + pub tool_arguments: Option, + pub tool_result: Option, + pub reasoning: Option, + /// The files a user message carried; see `message_attachments`. + pub attachments: Vec, +} + +/// The most files a user message keeps references to; the rest are dropped. +pub const MAX_MESSAGE_ATTACHMENTS: usize = 20; + +/// A file a user message carried, as the object-storage reference its run received. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MessageAttachment { + /// The flow input that held it. + pub input: String, + pub s3: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub storage: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub filename: Option, +} + +/// The files a run's args carry for its user message: every top-level input other than +/// `user_message` whose value is an object-storage reference or a list of them, in input +/// name order, capped at `MAX_MESSAGE_ATTACHMENTS`. Only the reference is kept: `presigned` +/// grants access to the file, and any other value may be the file's bytes. +pub fn message_attachments(args: &HashMap>) -> Vec { + let mut inputs: Vec<_> = args + .iter() + .filter(|(name, _)| name.as_str() != "user_message") + .collect(); + inputs.sort_by(|a, b| a.0.cmp(b.0)); + inputs + .into_iter() + .flat_map(|(name, value)| { + serde_json::from_str::(value.get()) + .map(|object| vec![object]) + .or_else(|_| serde_json::from_str::>(value.get())) + .unwrap_or_default() + .into_iter() + .filter(|object| !object.s3.is_empty()) + .map(move |object| MessageAttachment { + input: name.clone(), + s3: object.s3, + storage: object.storage, + filename: object.filename, + }) + }) + .take(MAX_MESSAGE_ATTACHMENTS) + .collect() +} + /// 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 @@ -150,6 +213,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!( @@ -170,14 +234,21 @@ 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, attachments) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)", 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()), + extras + .map(|e| &e.attachments) + .filter(|attachments| !attachments.is_empty()) + .map(sqlx::types::Json) as Option>> ) .execute(&mut **tx) .await?; @@ -213,6 +284,47 @@ pub async fn delete_conversation_memory( #[cfg(test)] mod tests { use super::*; + use serde_json::{json, value::to_raw_value}; + + fn args(values: serde_json::Value) -> HashMap> { + values + .as_object() + .unwrap() + .iter() + .map(|(name, value)| (name.clone(), to_raw_value(value).unwrap())) + .collect() + } + + #[test] + fn keeps_only_object_storage_references() { + let attachments = message_attachments(&args(json!({ + "user_message": { "s3": "not/an/attachment.png" }, + "avatar": { "s3": "u/a.png", "storage": "secondary", "presigned": "https://signed" }, + "files": [ + { "s3": "u/b.pdf", "filename": "b.pdf" }, + { "s3": "" } + ], + "photo": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==", + "count": 3 + }))); + assert_eq!( + serde_json::to_value(&attachments).unwrap(), + json!([ + { "input": "avatar", "s3": "u/a.png", "storage": "secondary" }, + { "input": "files", "s3": "u/b.pdf", "filename": "b.pdf" } + ]) + ); + } + + #[test] + fn caps_the_references_of_one_message() { + let files: Vec<_> = (0..25) + .map(|i| json!({ "s3": format!("u/{i}.png") })) + .collect(); + let attachments = message_attachments(&args(json!({ "files": files }))); + assert_eq!(attachments.len(), MAX_MESSAGE_ATTACHMENTS); + assert_eq!(attachments.last().unwrap().s3, "u/19.png"); + } /// A string names a memory only within its workspace and flow; a uuid is used as is. #[test] diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 9c15229ba6..7e775f34fe 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}, @@ -74,6 +74,9 @@ pub struct ToolExecutionContext<'a> { pub stream_event_processor: Option<&'a StreamEventProcessor>, pub flow_context: &'a mut FlowContext, pub omit_output_from_conversation: bool, + /// The thinking that led to this round's calls, stored on the first tool row written. + /// None when the round wrote text, whose row carries it. + pub reasoning: Option, pub previous_result: &'a Option>, pub id_context: &'a Option, @@ -235,9 +238,24 @@ 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 + // An MCP tool runs inside the agent's job, whose result holds every call of the + // turn and nothing tying one of them to this row: same job id for all of them, + // no call id on the row. Kept here so the card shows this call — and the row + // names that job, so retention sweeps it with every other row of the turn. let content = format!("Used {} tool", tool_call.function.name); - add_tool_message_to_chat(ctx, None, &content, true).await; + let agent_job_id = ctx.job.id; + add_tool_message_to_chat( + ctx, + Some(agent_job_id), + &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); @@ -271,8 +289,23 @@ async fn execute_mcp_tool_call( update_flow_status_module_with_actions_success(ctx.db, parent_job, false).await?; } - // Add tool message to conversation if chat_input_enabled - add_tool_message_to_chat(ctx, None, &error_msg, false).await; + // Add tool message to conversation if chat_input_enabled. The row is worded from + // the tool, like every other tool row, and the error it failed with is its result + // — the one field a call that produced nothing else still has something to put in. + let agent_job_id = ctx.job.id; + let content = format!("Error executing {}", tool_name); + add_tool_message_to_chat( + ctx, + Some(agent_job_id), + &content, + false, + Some(MessageExtras { + tool_arguments: Some(tool_call.function.arguments.clone()), + tool_result: Some(error_msg.clone()), + ..Default::default() + }), + ) + .await; } } @@ -680,8 +713,8 @@ async fn handle_tool_execution_error( update_flow_status_module_with_actions_success(ctx.db, parent_job, false).await?; } - // Add tool message to conversation if chat_input_enabled (error case) - add_tool_message_to_chat(ctx, Some(job_id), &error_message, false).await; + let (content, extras) = windmill_tool_row(tool_call, false, &error_message); + add_tool_message_to_chat(ctx, Some(job_id), &content, false, Some(extras)).await; Ok(()) } @@ -782,13 +815,17 @@ async fn handle_tool_execution_success( ..Default::default() }); - // Stream tool result (success case) + let (content, extras) = windmill_tool_row(tool_call, success, &tool_result); + + // The job ran; whether it ran successfully is `success`, and the row stored below is + // worded from it. The stream has to carry the same value, or the card the reader watches + // and the row that replaces it describe the same call differently. if let Some(stream_event_processor) = ctx.stream_event_processor { let tool_result_event = StreamingEvent::ToolResult { call_id: tool_call.id.clone(), function_name: tool_call.function.name.clone(), result: tool_result, - success: true, + success, }; stream_event_processor .send(tool_result_event, final_events_str) @@ -799,28 +836,56 @@ async fn handle_tool_execution_success( update_flow_status_module_with_actions_success(ctx.db, parent_job, success).await?; } - // Add tool message to conversation if chat_input_enabled + add_tool_message_to_chat(ctx, Some(job_id), &content, success, Some(extras)).await; + + Ok(()) +} + +/// A Windmill tool's conversation row: worded from the tool, carrying the model's call and +/// the exact text the model got back, the same text agent memory keeps for that tool +/// message, so a card needs no job fetch. The call is the model's arguments, not the job's +/// args: the step's input transforms add inputs the model never wrote. +fn windmill_tool_row( + tool_call: &OpenAIToolCall, + success: bool, + sent_to_model: &str, +) -> (String, MessageExtras) { let content = if success { format!("Used {} tool", tool_call.function.name) } else { format!("Error executing {}", tool_call.function.name) }; - - add_tool_message_to_chat(ctx, Some(job_id), &content, success).await; - - Ok(()) + let extras = MessageExtras { + tool_arguments: Some(tool_call.function.arguments.clone()), + tool_result: Some(sent_to_model.to_string()), + ..Default::default() + }; + (content, extras) } /// Add tool message to conversation if chat is enabled async fn add_tool_message_to_chat( ctx: &mut ToolExecutionContext<'_>, + // The job this row belongs to: the tool's own where it has one, else the agent's, which + // is the job it ran inside. Every row names one so that retention collects the whole + // turn — `delete_jobs` removes messages by `job_id = ANY(..)` (there is no FK on the + // column; `drop_v2_job_side_table_cascades` dropped it), and a row naming no job would + // survive every purge and leave a conversation that can never become empty. tool_job_id: Option, content: &str, success: bool, + // The model's call and what it got back; every tool row carries both. + extras: Option, ) { if ctx.omit_output_from_conversation { return; } + let extras = match ctx.reasoning.take() { + Some(reasoning) => { + Some(MessageExtras { reasoning: Some(reasoning), ..extras.unwrap_or_default() }) + } + None => extras, + }; let chat_enabled = ctx .flow_context @@ -835,41 +900,74 @@ async fn add_tool_message_to_chat( .as_ref() .and_then(|fs| fs.memory_id) { - let db_clone = ctx.db.clone(); let effective_step_id = ctx .flow_step_id_override .or(ctx.job.flow_step_id.as_deref()); let step_name = get_step_name_from_flow(ctx.summary.as_deref(), effective_step_id); - let content = content.to_string(); - // Spawn task because we do not need to wait for the result - tokio::spawn(async move { - if let Err(e) = add_message_to_conversation( - &db_clone, - &memory_id, - tool_job_id, - &content, - MessageType::Tool, - &step_name, - success, - ) - .await - { - tracing::warn!( - "Failed to add tool message to conversation {}: {}", - memory_id, - e - ); - } - }); + // Awaited, not spawned: `created_seq` is the transcript's order, so a round's rows + // must commit in the order of its calls. Calls run one after another; running them + // in parallel would need their rows written in call order all the same. + if let Err(e) = add_message_to_conversation( + ctx.db, + &memory_id, + tool_job_id, + content, + MessageType::Tool, + &step_name, + success, + extras.as_ref(), + ) + .await + { + tracing::warn!( + "Failed to add tool message to conversation {}: {}", + memory_id, + e + ); + } } } } #[cfg(test)] mod tests { - use super::extract_ai_agent_output; + use super::{extract_ai_agent_output, windmill_tool_row}; use serde_json::value::RawValue; + use windmill_ai::ai_types::{OpenAIFunction, OpenAIToolCall}; + + #[test] + fn a_windmill_tool_row_carries_the_models_call_and_what_it_got_back() { + let tool_call = OpenAIToolCall { + id: "call_1".to_string(), + function: OpenAIFunction { + name: "get_price".to_string(), + arguments: r#"{"item":"widget"}"#.to_string(), + }, + r#type: "function".to_string(), + extra_content: None, + }; + + let (content, extras) = windmill_tool_row(&tool_call, true, r#"{"price":42}"#); + assert_eq!(content, "Used get_price tool"); + assert_eq!( + extras.tool_arguments.as_deref(), + Some(r#"{"item":"widget"}"#) + ); + assert_eq!(extras.tool_result.as_deref(), Some(r#"{"price":42}"#)); + + let (content, extras) = + windmill_tool_row(&tool_call, false, "Error running tool: ExecutionErr: boom"); + assert_eq!(content, "Error executing get_price"); + assert_eq!( + extras.tool_arguments.as_deref(), + Some(r#"{"item":"widget"}"#) + ); + assert_eq!( + extras.tool_result.as_deref(), + Some("Error running tool: ExecutionErr: boom") + ); + } #[test] fn extracts_only_the_output_of_an_agent_result() { diff --git a/backend/windmill-worker/src/ai/utils.rs b/backend/windmill-worker/src/ai/utils.rs index e717f2172e..3efdc67a3d 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, MessageExtras, MessageType}, flow_status::AgentAction, flows::{InputTransform, Step}, jobs::JobKind, @@ -213,6 +213,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( @@ -223,6 +224,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 af3aa01b2e..efc7ffa1b3 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -44,7 +44,7 @@ use windmill_common::{ client::AuthedClient, db::DB, error::{self, Error}, - flow_conversations::{memory_key, MessageType}, + flow_conversations::{memory_key, MessageExtras, MessageType}, flow_status::AgentAction, flows::{AgentTool, FlowModule, FlowModuleValue, InputTransform, ToolValue}, get_latest_hash_for_path, @@ -1690,29 +1690,34 @@ pub async fn run_agent( }); if persist_output_to_conversation { if let Some(conversation_id) = conversation_id { - let agent_job_id = job.id; - let db_clone = db.clone(); - let message_content = "Used websearch tool successfully".to_string(); - let step_name = step_name.clone(); - tokio::spawn(async move { - if let Err(e) = add_message_to_conversation( - &db_clone, - &conversation_id, - Some(agent_job_id), - &message_content, - MessageType::Tool, - &step_name, - true, - ) - .await - { - tracing::warn!( - "Failed to add websearch tool message to conversation {}: {}", - conversation_id, - e - ); - } + // 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() }); + // Awaited like every row of the loop, so rows commit in turn order. + // Worded like every other tool row, so a reader recovers the tool + // name from the sentence. + if let Err(e) = add_message_to_conversation( + db, + &conversation_id, + Some(job.id), + "Used websearch tool", + MessageType::Tool, + &step_name, + true, + extras.as_ref(), + ) + .await + { + tracing::warn!( + "Failed to add websearch tool message to conversation {}: {}", + conversation_id, + e + ); + } } } } @@ -1742,31 +1747,29 @@ pub async fn run_agent( // Add assistant message to conversation if chat_input_enabled if persist_output_to_conversation && !response_content.is_empty() { if let Some(conversation_id) = conversation_id { - let agent_job_id = job.id; - let db_clone = db.clone(); - let message_content = response_content.clone(); - let step_name = step_name.clone(); - - // Spawn task because we do not need to wait for the result - tokio::spawn(async move { - if let Err(e) = add_message_to_conversation( - &db_clone, - &conversation_id, - Some(agent_job_id), - &message_content, - MessageType::Assistant, - &step_name, - true, - ) - .await - { - tracing::warn!( - "Failed to add assistant message to conversation {}: {}", - conversation_id, - e - ); - } + // This iteration's thinking goes on the answer's row; the job + // result only keeps the turn's thinking as one string. + let extras = response_reasoning.clone().map(|reasoning| { + MessageExtras { reasoning: Some(reasoning), ..Default::default() } }); + if let Err(e) = add_message_to_conversation( + db, + &conversation_id, + Some(job.id), + response_content, + MessageType::Assistant, + &step_name, + true, + extras.as_ref(), + ) + .await + { + tracing::warn!( + "Failed to add assistant message to conversation {}: {}", + conversation_id, + e + ); + } } } } @@ -1805,6 +1808,18 @@ pub async fn run_agent( ..Default::default() }); + // A round's thinking is stored on one row, the first the round writes, which is + // where the stream shows it: its text row when it wrote text, else the row of + // its first call — the answer row below when that call is the structured-output + // tool. Two rows carrying it would show it twice after a reload. + let call_reasoning = response_reasoning + .clone() + .filter(|_| response_content.as_deref().unwrap_or("").is_empty()); + let structured_output_first = structured_output_tool_name + .as_ref() + .zip(tool_calls.first()) + .map_or(false, |(name, tc)| tc.function.name == *name); + // Handle tool calls using extracted tools module let tool_execution_ctx = ToolExecutionContext { db, @@ -1823,6 +1838,11 @@ pub async fn run_agent( stream_event_processor: stream_event_processor.as_ref(), flow_context: &mut flow_context, omit_output_from_conversation, + reasoning: if structured_output_first { + None + } else { + call_reasoning.clone() + }, previous_result: &previous_result, id_context: &id_context, tool_abort_handles: tool_abort_handles.clone(), @@ -1841,6 +1861,41 @@ pub async fn run_agent( .await?; messages.extend(tool_messages); + + // A structured answer is the arguments of the structured-output tool call, + // on which the loop ends without a text iteration, so its row is written here. + if tool_used_structured_output && persist_output_to_conversation { + if let (Some(conversation_id), Some(OpenAIContent::Text(answer))) = + (conversation_id, tool_content.as_ref()) + { + let extras = call_reasoning + .clone() + .filter(|_| structured_output_first) + .map(|reasoning| MessageExtras { + reasoning: Some(reasoning), + ..Default::default() + }); + if let Err(e) = add_message_to_conversation( + db, + &conversation_id, + Some(job.id), + answer, + MessageType::Assistant, + &step_name, + true, + extras.as_ref(), + ) + .await + { + tracing::warn!( + "Failed to add structured answer to conversation {}: {}", + conversation_id, + e + ); + } + } + } + if let Some(tc) = tool_content { content = Some(tc); } @@ -1861,9 +1916,6 @@ pub async fn run_agent( // Add assistant message to conversation if chat_input_enabled if persist_output_to_conversation { if let Some(conversation_id) = conversation_id { - let agent_job_id = job.id; - let db_clone = db.clone(); - // Create extended version with type discriminator for conversation storage // This avoids conflicts with outputs that are of the same format as S3 objects let s3_with_type = S3ObjectWithType { @@ -1874,26 +1926,24 @@ pub async fn run_agent( let message_content = serde_json::to_string(&s3_with_type) .unwrap_or_else(|_| content.get().to_string()); - // Spawn task because we do not need to wait for the result - tokio::spawn(async move { - if let Err(e) = add_message_to_conversation( - &db_clone, - &conversation_id, - Some(agent_job_id), - &message_content, - MessageType::Assistant, - &step_name, - true, - ) - .await - { - tracing::warn!( - "Failed to add assistant message to conversation {}: {}", - conversation_id, - e - ); - } - }); + if let Err(e) = add_message_to_conversation( + db, + &conversation_id, + Some(job.id), + &message_content, + MessageType::Assistant, + &step_name, + true, + None, + ) + .await + { + tracing::warn!( + "Failed to add assistant message to conversation {}: {}", + conversation_id, + e + ); + } } } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index ed543bd517..6c0dcaf279 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -2236,6 +2236,7 @@ async fn add_tool_message_to_conversation( MessageType::Assistant, None, success, + None, ) .await?; tx.commit().await?; diff --git a/chat-sdk/README.md b/chat-sdk/README.md index ba76d1d507..0f1082de45 100644 --- a/chat-sdk/README.md +++ b/chat-sdk/README.md @@ -56,8 +56,10 @@ not a replacement of the previous answer. The transport also carries the history helpers: `transport.loadMessages(id)` returns `UIMessage`s for `useChat({ messages })` or `setMessages`, `transport.listConversations()` -and `transport.deleteConversation(id)`. Attachments are not supported: `sendMessage` with -`files` is refused with an explanatory error. +and `transport.deleteConversation(id)`. A loaded user message lists the files it carried +in `metadata.attachments`; `WindmillChatApi.attachmentUrl` gives each one's download URL. +Sending attachments is not supported: `sendMessage` with `files` is refused with an +explanatory error. ## assistant-ui diff --git a/chat-sdk/src/ai-sdk.ts b/chat-sdk/src/ai-sdk.ts index ba500446a3..d93bb99e2f 100644 --- a/chat-sdk/src/ai-sdk.ts +++ b/chat-sdk/src/ai-sdk.ts @@ -1,5 +1,10 @@ import type { ChatTransport, UIMessage, UIMessageChunk, UIMessagePart } from 'ai' -import { WindmillApiError, WindmillChatApi, type WindmillChatApiOptions } from './api' +import { + WindmillApiError, + WindmillChatApi, + type FlowConversationMessage, + type WindmillChatApiOptions +} from './api' import { followJob } from './follow' import type { AgentStreamEvent } from './stream' import type { ChatMessage, Conversation } from './types' @@ -101,7 +106,9 @@ export function createWindmillChatTransport { await this.#request(`jobs_u/queue/cancel/${encodeURIComponent(jobId)}`, { method: 'POST', diff --git a/chat-sdk/src/assistant-ui.ts b/chat-sdk/src/assistant-ui.ts index 36133c2272..b9a76db679 100644 --- a/chat-sdk/src/assistant-ui.ts +++ b/chat-sdk/src/assistant-ui.ts @@ -92,6 +92,7 @@ export function toThreadMessage(turn: WindmillTurn): ThreadMessageLike { } const content: ThreadContentPart[] = [] for (const m of turn.messages) { + if (m.reasoning) content.push({ type: 'reasoning', text: m.reasoning }) if (m.role === 'tool') { const args = parseJsonOr(m.tool?.arguments) content.push({ @@ -100,12 +101,12 @@ export function toThreadMessage(turn: WindmillTurn): ThreadMessageLike { toolName: m.tool?.name ?? 'tool', args: (isJsonObject(args) ? args : args === undefined ? {} : { input: args }) as ToolCallArgs, argsText: m.tool?.arguments ?? '', - result: m.tool?.status === 'running' ? undefined : (parseJsonOr(m.tool?.result) ?? m.content), + result: + m.tool?.status === 'running' ? undefined : m.tool?.result !== undefined ? parseJsonOr(m.tool.result) : m.content, isError: m.tool?.status === 'error' }) continue } - if (m.reasoning) content.push({ type: 'reasoning', text: m.reasoning }) if (m.content) content.push({ type: 'text', text: m.content }) } const last = turn.messages[turn.messages.length - 1] diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index 8055ea2234..b13fa89173 100644 --- a/chat-sdk/src/chat.ts +++ b/chat-sdk/src/chat.ts @@ -375,16 +375,21 @@ class ChatImpl implements Chat { tool: { ...existing.tool!, ...toolPatch } } } else { + // Thinking that produced no text led to this call, and is stored on its row. + const a = turn.assistantId ? messages.findIndex((m) => m.id === turn.assistantId) : -1 + const reasoning = a >= 0 && messages[a].content === '' ? messages.splice(a, 1)[0].reasoning : undefined messages.push({ id: `pending-${randomId()}`, role: 'tool', content: content ?? '', + reasoning, success: success ?? true, createdAt: now(), pending: true, tool: { callId, name, status: 'running', ...toolPatch } }) } + turn.assistantId = undefined } const appendAssistant = (text: string, reasoning: string) => { const i = turn.assistantId @@ -419,17 +424,14 @@ class ChatImpl implements Chat { case 'reasoning_token_delta': appendAssistant('', event.content) break + // A call completes the round's text: text after it is a new message. case 'tool_call': - // The round's text is complete; text after the tool result is a new message. - turn.assistantId = undefined upsertTool(event.call_id, event.function_name, { status: 'running' }) break case 'tool_call_arguments': - turn.assistantId = undefined upsertTool(event.call_id, event.function_name, { arguments: event.arguments }) break case 'tool_execution': - turn.assistantId = undefined upsertTool(event.call_id, event.function_name, { status: 'running' }) break case 'tool_result': @@ -643,22 +645,54 @@ class ChatImpl implements Chat { for (const row of rows.map(fromRow)) { if (known.has(row.id)) continue known.add(row.id) - const i = messages.findIndex( + let i = messages.findIndex( (m) => m.seq === undefined && m.role === row.role && (m.content === row.content || (row.tool !== undefined && m.tool?.name === row.tool.name)) ) + // A structured answer streams as the call of the structured-output tool, whose + // arguments are the answer's text: its row replaces that call. Only past the newest + // user message, where a stopped turn's identical call cannot be. + if (i < 0 && row.role === 'assistant') { + let j = messages.length - 1 + while (j >= 0 && messages[j].role !== 'user') { + const m = messages[j] + if (m.seq === undefined && m.role === 'tool' && m.tool?.arguments === row.content) i = j + j-- + } + } if (i >= 0) { const m = messages[i] messages[i] = { ...row, id: m.id, reasoning: m.reasoning ?? row.reasoning, - tool: m.tool ? { ...m.tool, status: row.tool?.status ?? m.tool.status } : row.tool + // The stream's call wins where it has a value; a stream cut short leaves gaps the row fills. + tool: + row.role === 'tool' && m.tool + ? { + ...m.tool, + arguments: m.tool.arguments ?? row.tool?.arguments, + result: m.tool.result ?? row.tool?.result, + status: row.tool?.status ?? m.tool.status + } + : row.tool } } else { - messages.push(row) + // A tool row nothing streamed, such as a provider-native web search, goes where a + // reload puts it: after the last message of a lower `seq`, before the streamed answer. + // Any other row closes the turn, a failure included, and stays last: above a streamed + // message that never got a row, it would hide the turn's failure. + let at = messages.length + for (let j = messages.length - 1; row.role === 'tool' && j >= 0; j--) { + const seq = messages[j].seq + if (seq !== undefined && seq < row.seq!) { + at = j + 1 + break + } + } + messages.splice(at, 0, row) } } this.#set({ messages }) @@ -756,7 +790,19 @@ function fromRow(row: FlowConversationMessage): ChatMessage { stepName: row.step_name ?? undefined, pending: false, seq: row.created_seq, - tool: toolName ? { name: toolName, status: success ? 'success' : 'error' } : undefined + reasoning: row.reasoning ?? undefined, + attachments: row.attachments ?? undefined, + // The call the row carries: the model's arguments and what the model got back. For a + // failed tool the result is what it failed with, and the row's text names the tool + // rather than the reason. + tool: toolName + ? { + name: toolName, + status: success ? 'success' : 'error', + arguments: row.tool_arguments ?? undefined, + result: row.tool_result ?? undefined + } + : undefined } } diff --git a/chat-sdk/src/index.ts b/chat-sdk/src/index.ts index 1c18a5924b..aedaa8b19f 100644 --- a/chat-sdk/src/index.ts +++ b/chat-sdk/src/index.ts @@ -16,6 +16,7 @@ export { followJob, type FollowEvent } from './follow' export { extractChatAnswer, conversationIdFor } from './utils' export type { Chat, + ChatAttachment, ChatMessage, ChatOptions, ChatRole, diff --git a/chat-sdk/src/types.ts b/chat-sdk/src/types.ts index 5e0d673bd9..87fd5a7b0a 100644 --- a/chat-sdk/src/types.ts +++ b/chat-sdk/src/types.ts @@ -26,19 +26,23 @@ export interface ToolInvocation { status: 'running' | 'success' | 'error' } +export interface ChatAttachment { input: string; s3: string; storage?: string; filename?: string } + export interface ChatMessage { id: string role: ChatRole content: string /** The model's reasoning summary, when the provider streams one. */ reasoning?: string - /** Set on `tool` messages that came from the live stream. */ + /** The call on a `tool` message, from the live stream or its stored row; `callId` is only known from the stream. */ tool?: ToolInvocation success: boolean createdAt: string jobId?: string /** The flow step that produced the message. */ stepName?: string + /** The files a user message carried, as object-storage references. */ + attachments?: ChatAttachment[] /** True while the message is optimistic or still streaming. */ pending: boolean /** Id of the persisted row once the server has it; `id` itself never changes, so list keys stay stable. */ diff --git a/chat-sdk/test/ai-sdk.test.ts b/chat-sdk/test/ai-sdk.test.ts index 7f30f26edc..70b7c40f33 100644 --- a/chat-sdk/test/ai-sdk.test.ts +++ b/chat-sdk/test/ai-sdk.test.ts @@ -2,7 +2,7 @@ import { describe, expect, test } from 'bun:test' import type { UIMessage, UIMessageChunk } from 'ai' import { createWindmillChatTransport, toUIMessages } from '../src/ai-sdk' import type { ChatMessage } from '../src/types' -import { fetchMock, json, ndjson, sse, text, type Route } from './support' +import { fetchMock, json, messageRow, ndjson, sse, text, type Route } from './support' const FLOW = 'f/chat/agent' const run: Route = (c) => @@ -141,6 +141,26 @@ describe('createWindmillChatTransport', () => { expect(second[1]).toMatchObject({ errorText: 'ExecutionErr: boom' }) }) + test('loads the attachments of a user row, the call an MCP tool row carries and the reasoning behind an answer', async () => { + const { fetch } = fetchMock((c) => + c.method === 'GET' && c.url.pathname.endsWith('/messages') + ? json([ + messageRow(1, 'user', 'hi', { attachments: [{ input: 'files', s3: 'chat/a.png', filename: 'a.png' }] }), + messageRow(2, 'tool', 'Used lookup tool', { job_id: 'agent-job', reasoning: 'why', tool_arguments: '{"q":1}', tool_result: '42' }), + messageRow(3, 'assistant', 'The answer is 42', { reasoning: 'hmm' }) + ]) + : undefined + ) + const transport = createWindmillChatTransport({ baseUrl: 'http://wm.test', workspace: 'ws', flowPath: FLOW, fetch }) + const ui = await transport.loadMessages('c') + expect(ui.map((m) => m.parts.map((p) => p.type))).toEqual([['text'], ['reasoning', 'dynamic-tool', 'reasoning', 'text']]) + expect(ui[0].metadata).toEqual({ attachments: [{ input: 'files', s3: 'chat/a.png', filename: 'a.png' }] }) + expect(ui[1].metadata).toBeUndefined() + expect(ui[1].parts[0]).toMatchObject({ type: 'reasoning', text: 'why' }) + expect(ui[1].parts[1]).toMatchObject({ toolName: 'lookup', state: 'output-available', input: { q: 1 }, output: 42 }) + expect(ui[1].parts[2]).toMatchObject({ type: 'reasoning', text: 'hmm' }) + }) + test('refuses attachments with a clear error', async () => { const transport = createWindmillChatTransport({ baseUrl: 'http://wm.test', workspace: 'ws', flowPath: FLOW, fetch: fetchMock().fetch }) await expect( @@ -175,4 +195,11 @@ describe('toUIMessages', () => { expect(ui[1].parts[0]).toMatchObject({ toolCallId: 'c1', toolName: 'lookup', state: 'output-available', input: { q: 1 }, output: 42 }) expect(ui[3].parts[0]).toMatchObject({ state: 'output-error', errorText: 'Error executing lookup' }) }) + + test('keeps a stored JSON null result rather than the row text', () => { + const ui = toUIMessages([ + { success: true, createdAt: '2026-01-01T00:00:00Z', pending: false, id: 't1', role: 'tool', content: 'Used notify tool', tool: { callId: 'c1', name: 'notify', status: 'success', arguments: '{}', result: 'null' } } + ]) + expect(ui[0].parts[0]).toMatchObject({ type: 'dynamic-tool', state: 'output-available', output: null }) + }) }) diff --git a/chat-sdk/test/api.test.ts b/chat-sdk/test/api.test.ts new file mode 100644 index 0000000000..0facf03e97 --- /dev/null +++ b/chat-sdk/test/api.test.ts @@ -0,0 +1,14 @@ +import { describe, expect, test } from 'bun:test' +import { WindmillChatApi } from '../src/api' + +describe('WindmillChatApi.attachmentUrl', () => { + test('points at the workspace download endpoint, with the storage only when there is one', () => { + const api = new WindmillChatApi({ baseUrl: 'https://wm.test/api/', workspace: 'my ws' }) + expect(api.attachmentUrl({ input: 'files', s3: 'chat/a b&c.png', storage: 'secondary' })).toBe( + 'https://wm.test/api/w/my%20ws/job_helpers/download_s3_file?file_key=chat%2Fa+b%26c.png&storage=secondary' + ) + expect(api.attachmentUrl({ input: 'files', s3: 'chat/a.png' })).toBe( + 'https://wm.test/api/w/my%20ws/job_helpers/download_s3_file?file_key=chat%2Fa.png' + ) + }) +}) diff --git a/chat-sdk/test/assistant-ui.test.ts b/chat-sdk/test/assistant-ui.test.ts index 5641dc2907..8601ad221a 100644 --- a/chat-sdk/test/assistant-ui.test.ts +++ b/chat-sdk/test/assistant-ui.test.ts @@ -31,6 +31,13 @@ describe('assistant-ui conversion', () => { expect(toThreadMessage(turns[0])).toMatchObject({ role: 'user', content: [{ type: 'text', text: 'hi' }] }) }) + test('keeps a stored JSON null result rather than the row text', () => { + const turn = groupTurns([ + { ...base, id: 't1', role: 'tool', content: 'Used notify tool', tool: { callId: 'c1', name: 'notify', status: 'success', arguments: '{}', result: 'null' } } + ])[0] + expect(toThreadMessage(turn).content).toMatchObject([{ type: 'tool-call', toolName: 'notify', result: null }]) + }) + test('marks a failed answer as incomplete', () => { const [turn] = groupTurns([{ ...base, id: 'a', role: 'assistant', content: 'boom', success: false }]) expect(toThreadMessage(turn).status).toEqual({ type: 'incomplete', reason: 'error', error: 'boom' }) diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index 89c4565942..a739abf3ad 100644 --- a/chat-sdk/test/chat.test.ts +++ b/chat-sdk/test/chat.test.ts @@ -327,6 +327,113 @@ describe('createChat with server history', () => { expect(messagesCall.headers.authorization).toBeUndefined() }) + test('a persisted row brings back its attachments, reasoning and the call an MCP tool row carries', async () => { + const { fetch } = fetchMock( + (c) => + c.method === 'GET' && c.url.pathname === '/api/w/ws/flow_conversations/conv-1/messages' + ? json([ + messageRow(1, 'user', 'hi', { + attachments: [{ input: 'files', s3: 'chat/a.png', storage: 'secondary', filename: 'a.png' }] + }), + messageRow(2, 'tool', 'Used lookup tool', { + job_id: 'agent-job', + tool_arguments: '{"q":1}', + tool_result: '42' + }), + messageRow(3, 'tool', 'Error executing lookup', { + job_id: 'agent-job', + success: false, + tool_arguments: '{"q":2}', + tool_result: 'MCP tool error: boom' + }), + messageRow(4, 'assistant', 'The answer is 42', { reasoning: 'hmm' }), + messageRow(5, 'assistant', 'Hello'), + messageRow(6, 'tool', 'Used get_price tool', { + job_id: 'script-tool-job', + tool_arguments: '{"item":"widget"}', + tool_result: '{"price":42}' + }) + ]) + : undefined + ) + const chat = createChat(options({ history: 'server' }, fetch)) + await chat.selectConversation('conv-1') + + const [user, used, failed, answer, plain, scriptTool] = chat.getState().messages + expect(scriptTool).toMatchObject({ jobId: 'script-tool-job', tool: { name: 'get_price', status: 'success', arguments: '{"item":"widget"}', result: '{"price":42}' } }) + expect(user.attachments).toEqual([{ input: 'files', s3: 'chat/a.png', storage: 'secondary', filename: 'a.png' }]) + expect(answer.attachments).toBeUndefined() + expect(used.tool).toEqual({ name: 'lookup', status: 'success', arguments: '{"q":1}', result: '42' }) + expect(failed.tool).toEqual({ name: 'lookup', status: 'error', arguments: '{"q":2}', result: 'MCP tool error: boom' }) + expect(answer.reasoning).toBe('hmm') + expect(plain.reasoning).toBeUndefined() + }) + + test('a row nothing streamed, like a web search, lands before the answer as on reload', async () => { + const { fetch } = fetchMock( + run, + (c) => + c.url.pathname === streamPath + ? sse([ + { + type: 'update', + new_result_stream: ndjson({ type: 'token_delta', content: 'Rust.' }), + stream_offset: 1, + completed: true, + only_result: { output: 'Rust.', messages: [] } + } + ]) + : undefined, + (c) => + c.url.pathname.endsWith('/messages') + ? json([ + messageRow(91, 'user', 'hi'), + messageRow(92, 'tool', 'Used websearch tool', { job_id: 'step-1', tool_result: '[{"url":"https://example.com"}]' }), + messageRow(93, 'assistant', 'Rust.', { job_id: 'step-1' }) + ]) + : undefined, + (c) => (c.url.pathname === '/api/w/ws/flow_conversations/list' ? json([]) : undefined) + ) + const chat = createChat(options({}, fetch)) + await chat.sendMessage('hi') + expect(chat.getState().messages.map((m) => [m.role, m.content, m.seq])).toEqual([ + ['user', 'hi', 91], + ['tool', 'Used websearch tool', 92], + ['assistant', 'Rust.', 93] + ]) + }) + + test('a failure row stays after streamed text that never got a row', async () => { + const { fetch } = fetchMock( + run, + (c) => + c.url.pathname === streamPath + ? sse([ + { + type: 'update', + new_result_stream: ndjson({ type: 'token_delta', content: 'Let me look' }), + stream_offset: 1, + completed: true, + only_result: { error: { name: 'ExecutionErr', message: 'boom' } } + } + ]) + : undefined, + (c) => + c.url.pathname.endsWith('/messages') + ? json([messageRow(91, 'user', 'hi'), messageRow(92, 'assistant', 'boom', { job_id: 'step-1', success: false })]) + : undefined, + (c) => (c.url.pathname === '/api/w/ws/flow_conversations/list' ? json([]) : undefined) + ) + const chat = createChat(options({}, fetch)) + await chat.sendMessage('hi') + const messages = chat.getState().messages + expect(messages.map((m) => [m.content, m.success])).toEqual([ + ['hi', true], + ['Let me look', true], + ['boom', false] + ]) + }) + test('keeps the streamed answer until its row lands, even when a tool row lands first', async () => { let messageFetches = 0 const { fetch } = fetchMock( @@ -689,6 +796,120 @@ describe('createChat with server history', () => { expect(chat.getState().messages.map((m) => m.content)).toEqual(['hi', 'Let me check', 'Used search tool', 'Final answer']) }) + test('thinking that led to a tool call rides on the call, live and once its row lands', async () => { + const { fetch } = fetchMock( + run, + (c) => + c.url.pathname === streamPath + ? sse([ + { + type: 'update', + // No `tool_call_arguments`: a stream cut short leaves the call without them. + new_result_stream: ndjson( + { type: 'reasoning_token_delta', content: 'r1' }, + { type: 'tool_call', call_id: 'c1', function_name: 'lookup' }, + { type: 'tool_result', call_id: 'c1', function_name: 'lookup', result: '1', success: true }, + { type: 'token_delta', content: 'Final' } + ), + stream_offset: 4, + completed: true, + only_result: { output: 'Final', messages: [] } + } + ]) + : undefined, + (c) => + c.url.pathname.endsWith('/messages') + ? json([ + messageRow(71, 'user', 'hi'), + messageRow(72, 'tool', 'Used lookup tool', { job_id: 'step-1', reasoning: 'r1', tool_arguments: '{"q":1}', tool_result: '1' }), + messageRow(73, 'assistant', 'Final', { job_id: 'step-1' }) + ]) + : undefined, + (c) => (c.url.pathname === '/api/w/ws/flow_conversations/list' ? json([]) : undefined) + ) + const chat = createChat(options({}, fetch)) + await chat.sendMessage('hi') + const messages = chat.getState().messages + expect(messages.map((m) => [m.role, m.content, m.reasoning, m.seq])).toEqual([ + ['user', 'hi', undefined, 71], + ['tool', 'Used lookup tool', 'r1', 72], + ['assistant', 'Final', undefined, 73] + ]) + expect(messages[1].tool).toMatchObject({ callId: 'c1', arguments: '{"q":1}', result: '1', status: 'success' }) + }) + + test('a structured answer row replaces the call it streamed as', async () => { + const { fetch } = fetchMock( + run, + (c) => + c.url.pathname === streamPath + ? sse([ + { + type: 'update', + new_result_stream: ndjson( + { type: 'reasoning_token_delta', content: 'hmm' }, + { type: 'tool_call', call_id: 'c9', function_name: 'structured_output' }, + { type: 'tool_call_arguments', call_id: 'c9', function_name: 'structured_output', arguments: '{"n": 1}' }, + { type: 'tool_execution', call_id: 'c9', function_name: 'structured_output' } + ), + stream_offset: 4, + completed: true, + only_result: { output: { n: 1 }, messages: [] } + } + ]) + : undefined, + (c) => + c.url.pathname.endsWith('/messages') + ? json([messageRow(75, 'user', 'hi'), messageRow(76, 'assistant', '{"n": 1}', { job_id: 'step-1', reasoning: 'hmm' })]) + : undefined, + (c) => (c.url.pathname === '/api/w/ws/flow_conversations/list' ? json([]) : undefined) + ) + const chat = createChat(options({}, fetch)) + await chat.sendMessage('hi') + expect(chat.getState().messages.map((m) => [m.role, m.content, m.reasoning, m.tool])).toEqual([ + ['user', 'hi', undefined, undefined], + ['assistant', '{"n": 1}', 'hmm', undefined] + ]) + }) + + test('a structured answer row leaves a stopped turn its identical call', async () => { + const call = (id: string) => + ndjson( + { type: 'tool_call', call_id: id, function_name: 'structured_output' }, + { type: 'tool_call_arguments', call_id: id, function_name: 'structured_output', arguments: '{"ok": true}' } + ) + let jobs = 0 + const { fetch } = fetchMock( + (c) => (c.method === 'POST' && c.url.pathname.includes('/jobs/run/f/') ? text(`job-${++jobs}`) : undefined), + (c) => + c.url.pathname.endsWith('/getupdate_sse/job-1') ? sse([{ type: 'update', new_result_stream: call('c1'), stream_offset: 2 }]) : undefined, + (c) => + c.url.pathname.endsWith('/getupdate_sse/job-2') + ? sse([{ type: 'update', new_result_stream: call('c2'), stream_offset: 2, completed: true, only_result: { output: { ok: true }, messages: [] } }]) + : undefined, + (c) => (c.url.pathname.includes('/queue/cancel/') ? text('ok') : undefined), + (c) => + c.url.pathname.endsWith('/messages') + ? json([messageRow(41, 'user', 'first'), messageRow(42, 'user', 'again'), messageRow(43, 'assistant', '{"ok": true}', { job_id: 'step-2' })]) + : undefined, + (c) => (c.url.pathname === '/api/w/ws/flow_conversations/list' ? json([]) : undefined) + ) + const chat = createChat(options({}, fetch)) + const first = chat.sendMessage('first') + await new Promise((r) => setTimeout(r, 50)) + const stopped = chat.stop() + await first + const second = chat.sendMessage('again') + await stopped + await second + expect(chat.getState().messages.map((m) => [m.role, m.content, m.tool?.callId])).toEqual([ + ['user', 'first', undefined], + ['tool', '', 'c1'], + ['user', 'again', undefined], + ['assistant', '{"ok": true}', undefined] + ]) + }) + test('the stream asks for a server poll interval only when one is set', async () => { const answer: Route = (c) => c.url.pathname === streamPath ? sse([{ type: 'update', completed: true, only_result: 'ok' }]) : undefined diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts index 78118546ac..9c9c865c1b 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts @@ -59,28 +59,77 @@ function lastTurnFailed(messages: readonly ChatMessage[]): boolean { return false } -export function toDisplayMessages(messages: readonly ChatMessage[]): DisplayMessage[] { +/** + * What a failed tool call returned, as the card's error: the card shows the error in place of + * the result. A job failure is stored as `{ message, name, stack }`; an MCP failure as plain text. + */ +function toolErrorText(result: unknown): string | undefined { + if (typeof result === 'string') return result || undefined + if (result && typeof result === 'object') { + const message = (result as { message?: unknown }).message + return typeof message === 'string' ? message : JSON.stringify(result, null, 2) + } + return undefined +} + +/** The name the worker gives the structured-output tool: suffixed when an agent tool already has it. */ +const STRUCTURED_OUTPUT_CALL = /^structured_output(_\d+)?$/ + +/** + * `busy`: the latest turn is still running, so a failed tool call is not yet its outcome. + * `stopped`: ids of the user messages of the turns the reader stopped, as message id or row id. + * A stopped turn never offers Retry, though Stop leaves its failed tool row or the cancelled + * flow's failure as its last message. + */ +export function toDisplayMessages( + messages: readonly ChatMessage[], + busy = false, + stopped: ReadonlySet = new Set() +): DisplayMessage[] { let userIndex = 0 - return messages.map((message, i): DisplayMessage => { + let latestUser = -1 + for (let i = messages.length - 1; i >= 0 && latestUser < 0; i--) { + if (messages[i].role === 'user') latestUser = i + } + return messages.flatMap((message, i): DisplayMessage[] => { switch (message.role) { - case 'user': - return { - role: 'user', - index: userIndex++, - content: message.content, - // Drives the shared Retry button. - error: turnFailed(messages, i) || undefined - } + case 'user': { + const index = userIndex++ + const settled = + !(busy && i === latestUser) && + !stopped.has(message.id) && + !(message.serverId && stopped.has(message.serverId)) + return [ + { + role: 'user', + index, + content: message.content, + // Drives the shared Retry button. + error: (settled && turnFailed(messages, i)) || undefined + } + ] + } case 'tool': { const parameters = parseToolPayload(message.tool?.arguments) const result = parseToolPayload(message.tool?.result) const failed = message.success === false - return { + // A call the turn finished without, stopped or lost before its row was written. The + // call a structured answer streams as never gets a result of its own, the answer + // being the turn's text, so it is not one. + const unfinished = + message.tool && + !message.pending && + !message.content && + !STRUCTURED_OUTPUT_CALL.test(message.tool.name) + ? `${message.tool.name} did not finish` + : undefined + const call: DisplayMessage = { role: 'tool', tool_call_id: message.id, // The card's header is the row's text, which the server only words once the // tool has returned; until then the row says what is running. - content: message.content || (message.tool ? `Running ${message.tool.name}` : ''), + content: + message.content || unfinished || (message.tool ? `Running ${message.tool.name}` : ''), // Withheld for the copilot's two plan-mode names: `toolName` is what makes // ToolExecutionDisplay render a plan card, and an agent tool that happened to // share one would silently become one. @@ -88,22 +137,39 @@ export function toDisplayMessages(messages: readonly ChatMessage[]): DisplayMess parameters, result, showDetails: parameters !== undefined || result !== undefined, - error: failed ? message.content : undefined, + error: failed ? (toolErrorText(result) ?? message.content) : unfinished, isLoading: message.pending && message.tool?.status === 'running' } + // The tool card has no thinking section: the thinking that led to the call reads + // as a card of its own, just before it. + return message.reasoning + ? [ + { + role: 'assistant', + content: '', + reasoning: message.reasoning, + stepName: message.stepName, + jobId: message.jobId, + createdAt: message.createdAt + }, + call + ] + : [call] } default: - return { - role: 'assistant', - content: message.content, - // Only the message a turn is still writing: a finalized reasoning-only - // message must not look in progress. - streaming: message.pending || undefined, - reasoning: message.reasoning, - stepName: message.stepName, - jobId: message.jobId, - createdAt: message.createdAt - } + return [ + { + role: 'assistant', + content: message.content, + // Only the message a turn is still writing: a finalized reasoning-only + // message must not look in progress. + streaming: message.pending || undefined, + reasoning: message.reasoning, + stepName: message.stepName, + jobId: message.jobId, + createdAt: message.createdAt + } + ] } }) } @@ -147,6 +213,12 @@ export class FlowChatViewHost implements ChatViewHost { #onState(state: ChatState) { const previous = this.#state this.#state = state + const landed = state.messages.filter( + (m) => m.serverId && this.#stoppedTurns.has(m.id) && !this.#stoppedTurns.has(m.serverId) + ) + if (landed.length > 0) { + this.#stoppedTurns = new Set([...this.#stoppedTurns, ...landed.map((m) => m.serverId!)]) + } if (previous.conversationId !== state.conversationId) { // A conversation opens at its end, whatever the reader was doing in the last one. this.#automaticScroll = true @@ -169,7 +241,14 @@ export class FlowChatViewHost implements ChatViewHost { } // Transcript - displayMessages = $derived.by(() => toDisplayMessages(this.#state.messages)) + displayMessages = $derived.by(() => + toDisplayMessages(this.#state.messages, isBusy(this.#state.status), this.#stoppedTurns) + ) + /** The user message of each turn stopped in this view, by message id and, once the chat has + * read its row, row id: a reopened conversation or an older page rebuilds messages from rows, + * so a turn left before its row was read is not recognised there. Only this session knows; + * a reload shows the stopped turn as its rows left it. */ + #stoppedTurns = $state.raw>(new Set()) get messages(): readonly unknown[] { return this.#state.messages } @@ -232,6 +311,13 @@ export class FlowChatViewHost implements ChatViewHost { // Stop means stop: what was typed during the run goes back to the composer rather // than waiting there to go out after some later turn settles. this.dequeueMessage() + const { messages, status } = this.#state + const turn = isBusy(status) ? [...messages].reverse().find((m) => m.role === 'user') : undefined + if (turn) { + this.#stoppedTurns = new Set( + [...this.#stoppedTurns, turn.id, turn.serverId].filter((id) => id !== undefined) + ) + } void this.#chat.stop() } // Typed off the interface: a Svelte component's own type resolves differently @@ -274,9 +360,10 @@ export class FlowChatViewHost implements ChatViewHost { // Per-message actions storedImages = () => undefined - /** Send the user message at this transcript position again. */ + /** Send the user message at this transcript position again. The position is in + * `displayMessages`, which holds more entries than the chat's messages. */ retryRequest = (messageIndex: number) => { - const message = this.#state.messages[messageIndex] + const message = this.displayMessages[messageIndex] if (!message || message.role !== 'user' || this.loading) return void this.sendRequest({ instructions: message.content }) } diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts index a7e953ecf5..eab3cc0df6 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts @@ -100,6 +100,59 @@ describe('toDisplayMessages', () => { }) }) + it('shows a call the turn finished without as an error, not as still running', () => { + const display = toDisplayMessages([ + message({ role: 'user', content: 'hi' }), + message({ role: 'tool', tool: { name: 'search', status: 'running' } }) + ]) + expect(display[1]).toMatchObject({ + content: 'search did not finish', + error: 'search did not finish', + isLoading: false + }) + + const structured = toDisplayMessages([ + message({ role: 'user', content: 'hi' }), + message({ + role: 'tool', + tool: { name: 'structured_output', status: 'running', arguments: '{"n":1}' } + }) + ]) + expect(structured[1]).toMatchObject({ content: 'Running structured_output', error: undefined }) + }) + + it('shows the thinking that led to a call as its own card, and retries by transcript position', async () => { + const rows = [ + message({ role: 'user', content: 'first' }), + message({ + role: 'tool', + content: 'Used search tool', + reasoning: 'why', + tool: { name: 'search', status: 'success' } + }), + message({ role: 'assistant', content: 'done' }), + message({ role: 'user', content: 'second' }) + ] + const display = toDisplayMessages(rows) + expect(display.map((m) => [m.role, m.content])).toEqual([ + ['user', 'first'], + ['assistant', ''], + ['tool', 'Used search tool'], + ['assistant', 'done'], + ['user', 'second'] + ]) + expect(display[1]).toMatchObject({ reasoning: 'why' }) + expect(display[1]).not.toHaveProperty('streaming') + + const { chat } = fakeChat(idleState({ messages: rows })) + const host = new FlowChatViewHost(chat) + host.retryRequest(4) + await vi.waitFor(() => + expect(chat.sendMessage).toHaveBeenCalledWith('second', expect.anything()) + ) + host.dispose() + }) + it('does not flag a turn whose tool failed but whose agent still answered', () => { const display = toDisplayMessages([ message({ role: 'user', content: 'try' }), @@ -114,6 +167,50 @@ describe('toDisplayMessages', () => { expect(display[0]).toMatchObject({ role: 'user', error: undefined }) }) + it('offers no retry while the turn whose tool failed is still running', () => { + const messages = [ + message({ role: 'user', content: 'first' }), + message({ role: 'assistant', content: 'boom', success: false }), + message({ role: 'user', content: 'try' }), + message({ + role: 'tool', + content: 'Error executing search', + success: false, + tool: { name: 'search', status: 'error' } + }) + ] + const running = toDisplayMessages(messages, true) + expect(running[0]).toMatchObject({ role: 'user', error: true }) + expect(running[2]).toMatchObject({ role: 'user', error: undefined }) + expect(toDisplayMessages(messages, false)[2]).toMatchObject({ role: 'user', error: true }) + }) + + it('shows what a failed tool returned as its error, not the row label', () => { + const display = toDisplayMessages([ + message({ + role: 'tool', + content: 'Error executing lookup_stock', + success: false, + tool: { + name: 'lookup_stock', + status: 'error', + result: '{"message":"stock service unavailable","name":"Error","stack":"at main"}' + } + }), + message({ + role: 'tool', + content: 'Error executing mcp_search', + success: false, + tool: { name: 'mcp_search', status: 'error', result: 'connection refused' } + }) + ]) + expect(display[0]).toMatchObject({ + content: 'Error executing lookup_stock', + error: 'stock service unavailable' + }) + expect(display[1]).toMatchObject({ error: 'connection refused' }) + }) + it('flags the streaming assistant message and a failed tool', () => { const display = toDisplayMessages([ message({ role: 'assistant', content: 'partial', pending: true }), @@ -257,6 +354,59 @@ describe('FlowChatViewHost', () => { host.dispose() }) + it('offers no retry on a turn the reader stopped', () => { + const failedTool = message({ + role: 'tool', + content: 'Error executing search', + success: false, + tool: { name: 'search', status: 'error' } + }) + const { chat, set } = fakeChat( + idleState({ + status: 'streaming', + messages: [message({ id: 'live', role: 'user', content: 'go' }), failedTool] + }) + ) + const host = new FlowChatViewHost(chat) + host.cancel() + expect(chat.stop).toHaveBeenCalled() + set({ status: 'idle' }) + expect(host.displayMessages[0]).toMatchObject({ role: 'user', error: undefined }) + // The chat re-reads the rows: the user message gets its row id, and the cancelled + // flow's failure lands as the answer. + const cancelled = message({ role: 'assistant', content: 'Job canceled', success: false }) + set({ + messages: [ + message({ id: 'live', serverId: 'row-2', role: 'user', content: 'go' }), + failedTool, + cancelled + ] + }) + expect(host.displayMessages[0]).toMatchObject({ role: 'user', error: undefined }) + // Reopened, with an older page that failed in front: messages are rebuilt from rows. + set({ + messages: [ + message({ id: 'row-0', serverId: 'row-0', role: 'user', content: 'before' }), + message({ role: 'assistant', content: 'boom', success: false }), + message({ id: 'row-2', serverId: 'row-2', role: 'user', content: 'go' }), + failedTool, + cancelled + ] + }) + expect(host.displayMessages[0]).toMatchObject({ content: 'before', error: true }) + expect(host.displayMessages[2]).toMatchObject({ content: 'go', error: undefined }) + // The next turn is not stopped and fails on its own. + set({ + messages: [ + ...chat.getState().messages, + message({ role: 'user', content: 'again' }), + message({ role: 'assistant', content: 'boom', success: false }) + ] + }) + expect(host.displayMessages.at(-2)).toMatchObject({ role: 'user', error: true }) + host.dispose() + }) + it('stops following the chat once disposed', () => { const { chat, set } = fakeChat() const host = new FlowChatViewHost(chat)