From 29fbcd03adced2d404ceb65d988666c8d4ba7626 Mon Sep 17 00:00:00 2001 From: Riqwan Thamir Date: Fri, 9 Oct 2026 08:58:18 +0200 Subject: [PATCH] fix(core): Record first_response on tool calls and tighten metric tests (no-changelog) Co-authored-by: Cursor --- .../resumable-stream-executor.test.ts | 23 +++++++--- .../src/runtime/resumable-stream-executor.ts | 4 +- .../__tests__/build-agent.tool.test.ts | 45 ++++++++++++------- 3 files changed, 50 insertions(+), 22 deletions(-) diff --git a/packages/@n8n/instance-ai/src/runtime/__tests__/resumable-stream-executor.test.ts b/packages/@n8n/instance-ai/src/runtime/__tests__/resumable-stream-executor.test.ts index 08887fcc13f..a9e77b964c4 100644 --- a/packages/@n8n/instance-ai/src/runtime/__tests__/resumable-stream-executor.test.ts +++ b/packages/@n8n/instance-ai/src/runtime/__tests__/resumable-stream-executor.test.ts @@ -340,13 +340,14 @@ describe('executeResumableStream', () => { expect(onActivity).toHaveBeenCalledTimes(2); }); - it('calls onResponseChunk for text, reasoning and tool starts, before publishing', async () => { + it('calls onResponseChunk for text, reasoning and tool calls, before publishing', async () => { const eventBus = createEventBus(); + const order: string[] = []; + eventBus.publish.mockImplementation((_threadId: string, event: { type: string }) => { + order.push(event.type); + }); const onResponseChunk = vi.fn(() => { - expect(eventBus.publish).not.toHaveBeenCalledWith( - 'thread-1', - expect.objectContaining({ type: 'text-delta' }), - ); + order.push('callback'); }); await executeResumableStream({ @@ -357,6 +358,7 @@ describe('executeResumableStream', () => { { type: 'start-step' }, { type: 'reasoning-delta', delta: 'Thinking' }, { type: 'tool-input-start', toolCallId: 'call-1', toolName: 'list_workflows' }, + { type: 'tool-call', toolCallId: 'call-2', toolName: 'list_workflows', input: {} }, textChunk('Hi'), ]), }, @@ -372,7 +374,16 @@ describe('executeResumableStream', () => { control: { mode: 'manual' }, }); - expect(onResponseChunk).toHaveBeenCalledTimes(3); + expect(order).toEqual([ + 'callback', + 'reasoning-delta', + 'callback', + 'tool-input-start', + 'callback', + 'tool-call', + 'callback', + 'text-delta', + ]); }); it('assigns stable response IDs from native start-step chunks', async () => { diff --git a/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts b/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts index 351a836fe3c..0b4fe8abfb5 100644 --- a/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts +++ b/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts @@ -23,6 +23,8 @@ const RESPONSE_CHUNK_TYPES: ReadonlyArray = [ 'text-delta', 'reasoning-delta', 'tool-input-start', + // Some providers send a complete tool call without a `tool-input-start`. + 'tool-call', ]; export interface ResumableStreamSource { @@ -45,7 +47,7 @@ export interface ResumableStreamContext { signal: AbortSignal; logger: Logger; onActivity?: () => void; - /** Called for each response chunk (text, reasoning, tool start), just before its event is published. */ + /** Called for each response chunk (text, reasoning, tool call), just before its event is published. */ onResponseChunk?: () => void; /** Stop consuming after the current chunk has been mapped and published. */ stopSignal?: () => OrchestratorRunStopSignal | undefined; diff --git a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/build-agent.tool.test.ts b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/build-agent.tool.test.ts index 5b7a83b7f38..1d6a88bdaa8 100644 --- a/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/build-agent.tool.test.ts +++ b/packages/@n8n/instance-ai/src/tools/orchestration/__tests__/build-agent.tool.test.ts @@ -2116,23 +2116,38 @@ describe('build-agent tool', () => { }); it('ORs carried configUpdated with the resumed pass when finishing', async () => { - const { context, delegate } = makeContext(); - context.domainContext!.agentBuilderTarget = { agentId: 'agent-1', projectId: 'proj-1' }; - vi.mocked(delegate.findOpenSuspensions).mockResolvedValue([ - { runId: 'builder-run-1', toolCallId: 'builder-call-1' }, - ]); - vi.mocked(delegate.resumeBuild).mockResolvedValue(fakeStream([], 'Done.')); + vi.useFakeTimers({ now: 100_000 }); + try { + const { context, delegate } = makeContext(); + context.domainContext!.agentBuilderTarget = { agentId: 'agent-1', projectId: 'proj-1' }; + vi.mocked(delegate.findOpenSuspensions).mockResolvedValue([ + { runId: 'builder-run-1', toolCallId: 'builder-call-1' }, + ]); + vi.mocked(delegate.resumeBuild).mockResolvedValue(fakeStream([], 'Done.')); - const result = await runToolWithCtx( - context, - { message: 'Build it', name: 'New Agent' }, - { - resumeData: { approved: true }, - suspendPayload: suspendPayloadWithCheckpoint({ configUpdated: true }), - }, - ); + const result = await runToolWithCtx( + context, + { message: 'Build it', name: 'New Agent' }, + { + resumeData: { approved: true }, + suspendPayload: suspendPayloadWithCheckpoint({ + configUpdated: true, + suspendedAt: 70_000, + }), + }, + ); - expect(result.configUpdated).toBe(true); + expect(result.configUpdated).toBe(true); + expect(emitAgentBuildMetrics).toHaveBeenCalledWith( + expect.objectContaining({ + outcome: 'completed', + configUpdated: true, + userWaitMs: 30_000, + }), + ); + } finally { + vi.useRealTimers(); + } }); it('reports carried configUpdated when the resumed pass errors', async () => {