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>

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)
   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)
   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/windmill-worker/src/ai_executor.rs

@@ -1621,8 +1621,15 @@ pub async fn run_agent(
 
                 // An iteration that answered with tool calls has no message row to carry its
                 // thinking, and the next iteration's row holds only its own. Stored on a row
-                // of its own so a reader sees what led to the call.
+                // of its own so a reader sees what led to the call. A call of the
+                // structured-output tool is the answer itself, and its row below carries
+                // the thinking: a bare row before it would read as the turn's last word.
+                let calls_structured_output =
+                    structured_output_tool_name.as_ref().map_or(false, |name| {
+                        tool_calls.iter().any(|tc| tc.function.name == *name)
+                    });
                 if persist_output_to_conversation
+                    && !calls_structured_output
                     && response_content.as_deref().unwrap_or("").is_empty()
                 {
                     if let (Some(memory_id), Some(reasoning)) =
@@ -1727,6 +1734,45 @@ 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 with the thinking of the iteration that produced it.
+                if tool_used_structured_output && persist_output_to_conversation {
+                    if let (Some(memory_id), Some(OpenAIContent::Text(answer))) =
+                        (memory_id, tool_content.as_ref())
+                    {
+                        let agent_job_id = job.id;
+                        let db_clone = db.clone();
+                        let message_content = answer.clone();
+                        let step_name = step_name.clone();
+                        let extras = response_reasoning.clone().map(|reasoning| MessageExtras {
+                            reasoning: Some(reasoning),
+                            ..Default::default()
+                        });
+                        tokio::spawn(async move {
+                            if let Err(e) = add_message_to_conversation(
+                                &db_clone,
+                                &memory_id,
+                                Some(agent_job_id),
+                                &message_content,
+                                MessageType::Assistant,
+                                &step_name,
+                                true,
+                                extras.as_ref(),
+                            )
+                            .await
+                            {
+                                tracing::warn!(
+                                    "Failed to add structured answer to conversation {}: {}",
+                                    memory_id,
+                                    e
+                                );
+                            }
+                        });
+                    }
+                }
+
                 if let Some(tc) = tool_content {
                     content = Some(tc);
                 }

chat-sdk/src/chat.ts

@@ -507,6 +507,9 @@ class ChatImpl implements Chat {
    * job is one of the turn's, which leaves out an earlier turn whose job outlived
    * `stop()` (a token without `jobs:write` cannot cancel it); a tool row without one
    * (an MCP call runs inside the agent step) belongs to whatever turn is under way.
+   * An assistant row with no text carries only the thinking before a tool call and
+   * is not an answer: a structured answer the agent never wrote a row for would
+   * otherwise be dropped instead of read from the flow result.
    */
   #answered(turn: Turn): boolean {
     const messages = this.#state.messages
@@ -519,7 +522,7 @@ class ChatImpl implements Chat {
       if (m.seq === undefined || m.role === 'user' || !ownJob(m)) continue
       if (latest === undefined || m.seq > latest.seq!) latest = m
     }
-    return latest?.role === 'assistant'
+    return latest?.role === 'assistant' && latest.content !== ''
   }
 
   /**

chat-sdk/test/chat.test.ts

@@ -615,6 +615,28 @@ describe('createChat with server history', () => {
     expect(chat.getState().messages.map((m) => m.content)).toEqual(['hi', 'Let me check', 'Used search tool', 'Final answer'])
   })
 
+  test('a reasoning-only row is not the answer of a turn that streamed no text', async () => {
+    const { fetch } = fetchMock(
+      run,
+      (c) =>
+        c.url.pathname === streamPath
+          ? sse([{ type: 'update', completed: true, only_result: { output: '{"n":1}', messages: [] } }])
+          : undefined,
+      (c) =>
+        c.url.pathname.endsWith('/messages')
+          ? json([messageRow(71, 'user', 'hi'), messageRow(72, 'assistant', '', { 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])).toEqual([
+      ['user', 'hi', undefined],
+      ['assistant', '', 'hmm'],
+      ['assistant', '{"n":1}', 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