From 701e8decce54aa2e7d463986c4cba0cf3436627d Mon Sep 17 00:00:00 2001 From: hugocasa Date: Fri, 11 Sep 2026 16:56:55 +0200 Subject: [PATCH] fix(ai-agent): reset the streamed answer on providers that skip tool_call Co-Authored-By: Claude Opus 5 (1M context) --- backend/windmill-worker/src/ai_executor.rs | 15 ++++----------- .../src/lib/components/aiAgentResult.test.ts | 18 ++++++++++++++++++ frontend/src/lib/components/aiAgentResult.ts | 14 +++++++++++--- 3 files changed, 33 insertions(+), 14 deletions(-) diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index febb253253..925b8e9272 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -75,13 +75,10 @@ lazy_static::lazy_static! { const DEFAULT_MAX_AGENT_ITERATIONS: usize = 10; const HARD_MAX_AGENT_ITERATIONS: usize = 1000; -/// What a run stopped by `max_iterations` reports back: the conversation it got -/// through before giving up. -/// -/// `Message` rather than `OpenAIMessage` is load-bearing. `agent_action` is -/// `skip_serializing` on `OpenAIMessage` and only reaches JSON through this -/// wrapper, so serializing these raw drops every tool name and job id and leaves -/// the partial run unreadable — which is the one thing worth having on this path. +/// What a run stopped by `max_iterations` reports back. `Message` rather than +/// `OpenAIMessage` is load-bearing: `agent_action` is `skip_serializing` on the +/// latter and reaches JSON only through this wrapper, so serializing these raw +/// drops every tool name and job id and leaves the partial run unreadable. #[derive(serde::Serialize)] struct MaxIterPartialResult<'a> { messages: Vec>, @@ -1801,10 +1798,6 @@ mod tests { assert!(!streaming_requested(Some(false))); } - /// The frontend reads a capped run's partial actions out of this payload and - /// keys entirely off `agent_action`. That field is `skip_serializing` on - /// `OpenAIMessage`, so serializing these messages directly rather than through - /// `Message` silently empties the trace of the one run worth reading. #[test] fn max_iterations_partial_result_keeps_the_action_tags() { let messages = vec![OpenAIMessage { diff --git a/frontend/src/lib/components/aiAgentResult.test.ts b/frontend/src/lib/components/aiAgentResult.test.ts index 99ce2c5a56..75875e8c30 100644 --- a/frontend/src/lib/components/aiAgentResult.test.ts +++ b/frontend/src/lib/components/aiAgentResult.test.ts @@ -151,4 +151,22 @@ describe('a stream that narrates before calling a tool', () => { const poll3 = advanceAgentStream(afterCall + '{"type":"token_delta","content":"Done."}\n', poll2) expect(poll3.stream.answer).toBe('Done.') }) + + // Bedrock has its own streaming implementation rather than the shared SSE + // parsers, and never announces `tool_call` — only the arguments, then the + // worker's `tool_execution`. Keying the reset on `tool_call` alone leaves the + // narration in place for that provider. + it('resets on a provider that never announces the call itself', () => { + const raw = [ + '{"type":"token_delta","content":"Let me check the metrics."}', + '{"type":"tool_call_arguments","call_id":"c1","function_name":"q","arguments":"{}"}', + '{"type":"tool_execution","call_id":"c1","function_name":"q"}', + '{"type":"tool_result","call_id":"c1","function_name":"q","result":"{}","success":true}', + '{"type":"token_delta","content":"eu-central-1 is down."}', + '' + ].join('\n') + expect(advanceAgentStream(raw, emptyAgentStreamProgress()).stream.answer).toBe( + 'eu-central-1 is down.' + ) + }) }) diff --git a/frontend/src/lib/components/aiAgentResult.ts b/frontend/src/lib/components/aiAgentResult.ts index 2710e5493b..b7ec8d695e 100644 --- a/frontend/src/lib/components/aiAgentResult.ts +++ b/frontend/src/lib/components/aiAgentResult.ts @@ -156,6 +156,14 @@ export function emptyAgentStreamProgress(): AgentStreamProgress { return { consumed: 0, stream: { answer: '', reasoning: '' } } } +/** + * Any of these means a tool call is beginning, and which one arrives depends on + * the provider: the SSE parsers announce `tool_call`, while Bedrock streams only + * the arguments and the worker follows with `tool_execution`. Resetting on all + * three is idempotent and keeps the rule provider-independent. + */ +const TOOL_TURN_STARTED = ['tool_call', 'tool_call_arguments', 'tool_execution'] + const STREAM_EVENT_TYPES = [ 'token_delta', 'reasoning_token_delta', @@ -232,11 +240,11 @@ export function advanceAgentStream( } else if (event.type === 'reasoning_token_delta' && typeof event.content === 'string') { stream.reasoning += event.content } else if (typeof event.function_name === 'string') { - if (event.type === 'tool_call') { + if (TOOL_TURN_STARTED.includes(event.type)) { // A model can narrate and request a tool in the same turn, and the loop // then runs again. That narration is not part of the answer — the - // finished result keeps only the last turn's text — so a new call - // starts the answer over rather than appending to what came before. + // finished result keeps the text of the last turn that produced any — so + // a starting call resets rather than appending to what came before. stream.answer = '' stream.reasoning = '' }