feat: store mcp tool call, result and reasoning on flow conversation rows (#11176)

* feat: store mcp tool call, result and reasoning on flow conversation rows

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

* fix: keep the failure reason and web search citations on tool rows

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

* fix: persist a structured answer as its own row and keep reasoning-only rows from closing a turn

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

* fix: keep a failed Windmill tool's error on its conversation row

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

* fix: let a structured answer row claim its streamed thinking

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

* fix: scope a structured answer's claim to its own turn and document the row fields as stored

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

* fix: claim a textless assistant row only within the newest turn

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

* refactor: store the thinking that led to a tool call on the tool row

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: keep tool calls in stream order and scope a structured answer to its turn

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* feat: store the files a user message carried as object-storage references on its row

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* feat(chat-sdk): read a message's attachments and build their download URL

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* feat(chat-sdk): carry a loaded user message's attachments in AI SDK metadata

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* feat: store the model's call and what it got back on every tool row

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: pass no extras in the orphaned-conversation test's message insert

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(chat-sdk): keep a stored JSON null tool result instead of the row text

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: label a tool call the turn finished without as not finished

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: keep a structured answer's streamed call out of the did-not-finish label

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: write a turn's conversation rows in order and describe stored tool calls

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix: await the image answer row like the agent loop's other rows

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(chat-sdk): place a row nothing streamed by its sequence, not at the end

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(chat-sdk): place only tool rows by sequence, keep a closing row last

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(flow chat): show a failed tool's error, and no Retry on a running or stopped turn

A failed tool card showed the row label in place of the error the row now
stores. The live tool result now reports failures, so a failed call briefly
flagged a running turn as failed; a stopped turn, whose last row is the
failed tool or the cancelled flow's failure, offered Retry too.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

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

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

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

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

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

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;

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;

backend/migrations/20260917074040_flow_conversation_message_attachments.down.sql

@@ -0,0 +1 @@
+ALTER TABLE flow_conversation_message DROP COLUMN attachments;

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;

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))

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?;

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

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<String, StreamingToolCall> = HashMap::new();
+        let mut accumulated_tool_calls: indexmap::IndexMap<String, StreamingToolCall> =
+            indexmap::IndexMap::new();
         let mut current_tool_use_id: Option<String> = None;
         let mut usage: Option<TokenUsage> = 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

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<i64, OpenAIToolCall>,
+    // 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<i64, OpenAIToolCall>,
     pub events_str: String,
     pub stream_event_processor: Box<dyn StreamEventSink>,
     /// 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<i64, OpenAIToolCall>,
+    pub accumulated_tool_calls: IndexMap<i64, OpenAIToolCall>,
     pub events_str: String,
     pub stream_event_processor: Box<dyn StreamEventSink>,
     /// 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<i64, OpenAIToolCall>,
+    pub accumulated_tool_calls: IndexMap<i64, OpenAIToolCall>,
     pub events_str: String,
     pub stream_event_processor: Box<dyn StreamEventSink>,
     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<String, OpenAIToolCall>,
+    pub accumulated_tool_calls: IndexMap<String, OpenAIToolCall>,
     /// Maps item_id -> (name, call_id) for function calls
     tool_call_metadata: HashMap<String, (String, String)>,
     /// 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(),

backend/windmill-api-flow-conversations/src/lib.rs

@@ -37,6 +37,19 @@ pub struct FlowConversationMessage {
     pub created_seq: i64,
     pub step_name: Option<String>,
     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<String>,
+    /// 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<String>,
+    /// 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<String>,
+    /// The files a user message carried, as object-storage references
+    /// (`[{input, s3, storage?, filename?}]`).
+    pub attachments: Option<sqlx::types::JsonValue>,
 }
 
 /// 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

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<serde_json::value::RawValue>>,
     job_id: Uuid,
     is_test: bool,
+    // The run's args, for the files the message carried.
+    args: &HashMap<String, Box<serde_json::value::RawValue>>,
 ) -> 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?;
     }

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

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?;
     }

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<String>,
+    pub tool_result: Option<String>,
+    pub reasoning: Option<String>,
+    /// The files a user message carried; see `message_attachments`.
+    pub attachments: Vec<MessageAttachment>,
+}
+
+/// 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<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub filename: Option<String>,
+}
+
+/// 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<String, Box<RawValue>>) -> Vec<MessageAttachment> {
+    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::<S3Object>(value.get())
+                .map(|object| vec![object])
+                .or_else(|_| serde_json::from_str::<Vec<S3Object>>(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<sqlx::types::Json<&Vec<MessageAttachment>>>
     )
     .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<String, Box<RawValue>> {
+        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]

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<String>,
     pub previous_result: &'a Option<Box<RawValue>>,
     pub id_context: &'a Option<crate::js_eval::IdContext>,
 
@@ -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<Uuid>,
     content: &str,
     success: bool,
+    // The model's call and what it got back; every tool row carries both.
+    extras: Option<MessageExtras>,
 ) {
     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() {

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<String>,
     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?;

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
+                            );
+                        }
                     }
                 }
 

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?;

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
 

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<UI_MESSAGE extends UIMessage = UIMes
           stepName: row.step_name ?? undefined,
           pending: false,
           seq: row.created_seq,
-          tool: toolFromRowContent(row.message_type, row.content, row.success ?? true)
+          reasoning: row.reasoning ?? undefined,
+          attachments: row.attachments ?? undefined,
+          tool: toolFromRow(row)
         }))
       ) as UI_MESSAGE[]
     },
@@ -123,10 +130,20 @@ export function createWindmillChatTransport<UI_MESSAGE extends UIMessage = UIMes
   }
 }
 
-function toolFromRowContent(role: string, content: string, success: boolean): ChatMessage['tool'] {
-  if (role !== 'tool') return undefined
-  const name = /^Used (.+) tool$/.exec(content)?.[1] ?? /^Error executing (.+)$/.exec(content)?.[1]
-  return name ? { name, status: success ? 'success' : 'error' } : undefined
+/** The call a stored tool row carries: its tool, named by the sentence the worker words
+ *  every tool row from, and the model's arguments and what it got back — for a failed
+ *  tool, the result is what it failed with. */
+function toolFromRow(row: FlowConversationMessage): ChatMessage['tool'] {
+  if (row.message_type !== 'tool') return undefined
+  const name =
+    /^Used (.+) tool$/.exec(row.content)?.[1] ?? /^Error executing (.+)$/.exec(row.content)?.[1]
+  if (!name) return undefined
+  return {
+    name,
+    status: (row.success ?? true) ? 'success' : 'error',
+    arguments: row.tool_arguments ?? undefined,
+    result: row.tool_result ?? undefined
+  }
 }
 
 /** Streams a job's answer as AI SDK chunks; resumes from `entry.offset` when the job is already running. */
@@ -292,13 +309,21 @@ class PartWriter {
 
 /**
  * `ChatMessage`s (Windmill's role-per-row model) as `UIMessage`s: an assistant
- * turn becomes one message whose parts carry its text, reasoning and tool calls.
+ * turn becomes one message whose parts carry its text, reasoning and tool calls. A
+ * user message's attachments ride in `metadata.attachments`, as references for
+ * `WindmillChatApi.attachmentUrl`: a `file` part would need a URL the browser can
+ * load unauthenticated.
  */
 export function toUIMessages(messages: ChatMessage[]): UIMessage[] {
   const out: UIMessage[] = []
   for (const m of messages) {
     if (m.role === 'user' || m.role === 'system') {
-      out.push({ id: m.id, role: m.role, parts: [{ type: 'text', text: m.content }] })
+      out.push({
+        id: m.id,
+        role: m.role,
+        ...(m.attachments?.length ? { metadata: { attachments: m.attachments } } : {}),
+        parts: [{ type: 'text', text: m.content }]
+      })
       continue
     }
     let target = out[out.length - 1]
@@ -306,18 +331,18 @@ export function toUIMessages(messages: ChatMessage[]): UIMessage[] {
       target = { id: m.id, role: 'assistant', parts: [] }
       out.push(target)
     }
+    if (m.reasoning) target.parts.push({ type: 'reasoning', text: m.reasoning, state: 'done' })
     if (m.role === 'tool') {
       const toolCallId = m.tool?.callId ?? m.id
       const toolName = m.tool?.name ?? 'tool'
       const input = parseJsonOr(m.tool?.arguments)
       target.parts.push(
         m.success
-          ? { type: 'dynamic-tool', toolName, toolCallId, state: 'output-available', input, output: parseJsonOr(m.tool?.result) ?? m.content }
+          ? { type: 'dynamic-tool', toolName, toolCallId, state: 'output-available', input, output: m.tool?.result !== undefined ? parseJsonOr(m.tool.result) : m.content }
           : { type: 'dynamic-tool', toolName, toolCallId, state: 'output-error', input, errorText: m.tool?.result ?? m.content }
       )
       continue
     }
-    if (m.reasoning) target.parts.push({ type: 'reasoning', text: m.reasoning, state: 'done' })
     if (m.content) target.parts.push({ type: 'text', text: m.content, state: 'done' })
   }
   return out

chat-sdk/src/api.ts

@@ -1,4 +1,4 @@
-import type { FetchLike, TokenSource } from './types'
+import type { ChatAttachment, FetchLike, TokenSource } from './types'
 
 export interface WindmillChatApiOptions {
   baseUrl: string
@@ -48,6 +48,14 @@ export interface FlowConversationMessage {
   created_seq: number
   step_name?: string | null
   success?: boolean
+  /** On a tool row, the arguments the model wrote, without the inputs a step wires in; null for a web search. */
+  tool_arguments?: string | null
+  /** On a tool row, the text the model got back or what the call failed with; a web search's citations. */
+  tool_result?: string | null
+  /** On an answer, the thinking that produced it; on a tool row, the thinking that led to the call. */
+  reasoning?: string | null
+  /** The files a user message carried, as object-storage references. */
+  attachments?: ChatAttachment[] | null
 }
 
 export type JobUpdateEvent =
@@ -166,6 +174,17 @@ export class WindmillChatApi {
     return (await res.json()) as FlowJobStatus
   }
 
+  /**
+   * Where a message's attachment downloads from. The endpoint authenticates like every other
+   * request: a consumer holding a token must fetch it with that token, not put the URL in an
+   * `img src`, which would send only the Windmill session cookie.
+   */
+  attachmentUrl(attachment: ChatAttachment): string {
+    const query = new URLSearchParams({ file_key: attachment.s3 })
+    if (attachment.storage) query.set('storage', attachment.storage)
+    return `${this.#baseUrl}/api/w/${encodeURIComponent(this.#workspace)}/job_helpers/download_s3_file?${query}`
+  }
+
   async cancelJob(jobId: string, reason = 'Stopped from the chat'): Promise<void> {
     await this.#request(`jobs_u/queue/cancel/${encodeURIComponent(jobId)}`, {
       method: 'POST',

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]

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
   }
 }
 

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,

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. */

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 })
+  })
 })

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'
+    )
+  })
+})

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' })

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

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<string> = 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<ReadonlySet<string>>(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 })
 	}

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)