From 019b1cf24d6f4d87ecff8140bf8936b4a6068874 Mon Sep 17 00:00:00 2001 From: Guilhem Lemouel Date: Wed, 16 Sep 2026 18:10:18 +0200 Subject: [PATCH] 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 --- backend/summarized_schema.txt | 2 +- backend/windmill-worker/src/ai_executor.rs | 48 +++++++++++++++++++++- chat-sdk/src/chat.ts | 5 ++- chat-sdk/test/chat.test.ts | 22 ++++++++++ 4 files changed, 74 insertions(+), 3 deletions(-) diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index 1e27ee1ce4..d0f666cf64 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -101,7 +101,7 @@ flow: workspace_id(char), path(char), summary(text), description(text), value(js FK: (workspace_id) -> workspace(id) flow_conversation: id(uuid), workspace_id(char), flow_path(char), title(char), created_at(ts), updated_at(ts), created_by(char) 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)) diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 9480bd0645..35b204f364 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/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); } diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index 3067c2d4f0..c62671d1df 100644 --- a/chat-sdk/src/chat.ts +++ b/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 !== '' } /** diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index dc518772f9..9ceec47ba8 100644 --- a/chat-sdk/test/chat.test.ts +++ b/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