From b325597d29d637d3f8e3bcbea5fc7977cfb51824 Mon Sep 17 00:00:00 2001 From: yyh <92089059+lyzno1@users.noreply.github.com> Date: Fri, 10 Jul 2026 12:26:04 +0800 Subject: [PATCH] test(e2e): consume agent service API SSE incrementally (#38634) --- .github/workflows/web-e2e.yml | 4 + e2e/features/agent-v2/support/access-point.ts | 92 ++++-------- .../agent-v2/support/service-api-sse.ts | 135 ++++++++++++++++++ .../access-point-service-api.steps.ts | 59 ++++---- .../step-definitions/agent-v2/tools.steps.ts | 39 ++--- e2e/package.json | 3 + e2e/tests/service-api-sse.test.ts | 97 +++++++++++++ pnpm-lock.yaml | 8 +- 8 files changed, 325 insertions(+), 112 deletions(-) create mode 100644 e2e/features/agent-v2/support/service-api-sse.ts create mode 100644 e2e/tests/service-api-sse.test.ts diff --git a/.github/workflows/web-e2e.yml b/.github/workflows/web-e2e.yml index 5dba2776af4..0f24fd9be64 100644 --- a/.github/workflows/web-e2e.yml +++ b/.github/workflows/web-e2e.yml @@ -44,6 +44,10 @@ jobs: - name: Install API dependencies run: uv sync --project api --dev + - name: Run E2E support unit tests + working-directory: ./e2e + run: vp run test:unit + - name: Install Playwright browser working-directory: ./e2e run: vp run e2e:install diff --git a/e2e/features/agent-v2/support/access-point.ts b/e2e/features/agent-v2/support/access-point.ts index 272364ba192..a53c176372f 100644 --- a/e2e/features/agent-v2/support/access-point.ts +++ b/e2e/features/agent-v2/support/access-point.ts @@ -6,10 +6,9 @@ import type { ChatRequestPayloadWithUser, PostChatMessagesResponse, } from '@dify/contracts/api/service/types.gen' -import type { APIResponse } from '@playwright/test' -import { request } from '@playwright/test' import { createApiContext, expectApiResponseOK, setAppSiteEnabled } from '../../../support/api' import { getTestAgent } from './agent' +import { consumeServiceApiSse, SERVICE_API_STREAM_TIMEOUT_MS } from './service-api-sse' export type AgentServiceApiChatResult = { body: PostChatMessagesResponse | unknown @@ -17,17 +16,13 @@ export type AgentServiceApiChatResult = { status: number } -type ServiceApiSseEvent = { - data: unknown - event?: string -} - -async function parseServiceApiChatResponse(response: APIResponse) { - const contentType = response.headers()['content-type'] ?? '' - const text = await response.text().catch(() => '') +async function parseServiceApiChatResponse(response: Response) { + const contentType = response.headers.get('content-type') ?? '' if (contentType.includes('text/event-stream')) - return parseServiceApiSseText(text) + return consumeServiceApiSse(response.body) + + const text = await response.text().catch(() => '') if (contentType.includes('application/json')) { try { @@ -46,55 +41,6 @@ async function parseServiceApiChatResponse(response: APIResponse) { } } -function parseServiceApiSseText(text: string) { - const events: ServiceApiSseEvent[] = [] - const answers: string[] = [] - - for (const block of text.split(/\r?\n\r?\n/)) { - const lines = block.split(/\r?\n/) - const eventName = lines - .find(line => line.startsWith('event:')) - ?.slice('event:'.length) - .trim() - const dataText = lines - .filter(line => line.startsWith('data:')) - .map(line => line.slice('data:'.length).trimStart()) - .join('\n') - - if (!dataText) - continue - - let data: unknown = dataText - try { - data = JSON.parse(dataText) as unknown - } - catch { - data = dataText - } - - events.push({ - data, - ...(eventName ? { event: eventName } : {}), - }) - - if ( - data - && typeof data === 'object' - && !Array.isArray(data) - && 'answer' in data - && typeof data.answer === 'string' - ) { - answers.push(data.answer) - } - } - - return { - answer: answers.join(''), - events, - raw: text, - } -} - export async function setAgentSiteAccessAndGetURL( agentId: string, enabled: boolean, @@ -152,30 +98,40 @@ export async function sendAgentServiceApiChatMessage({ query?: string serviceApiBaseURL: string }): Promise { - const ctx = await request.newContext() const body = { inputs: {}, query, response_mode: 'streaming', user: 'e2e-agent-access-point', } satisfies ChatRequestPayloadWithUser + const signal = AbortSignal.timeout(SERVICE_API_STREAM_TIMEOUT_MS) try { - const response = await ctx.post(`${serviceApiBaseURL.replace(/\/$/, '')}/chat-messages`, { - data: body, + const response = await fetch(`${serviceApiBaseURL.replace(/\/$/, '')}/chat-messages`, { + body: JSON.stringify(body), headers: { - Authorization: `Bearer ${apiKey}`, + 'Accept': 'text/event-stream', + 'Authorization': `Bearer ${apiKey}`, + 'Content-Type': 'application/json', }, + method: 'POST', + signal, }) const responseBody = await parseServiceApiChatResponse(response) return { body: responseBody as PostChatMessagesResponse | unknown, - ok: response.ok(), - status: response.status(), + ok: response.ok, + status: response.status, } } - finally { - await ctx.dispose() + catch (error) { + if (signal.aborted) { + throw new Error( + `Agent v2 Service API stream timed out after ${SERVICE_API_STREAM_TIMEOUT_MS}ms.`, + { cause: error }, + ) + } + throw error } } diff --git a/e2e/features/agent-v2/support/service-api-sse.ts b/e2e/features/agent-v2/support/service-api-sse.ts new file mode 100644 index 00000000000..0ed82663214 --- /dev/null +++ b/e2e/features/agent-v2/support/service-api-sse.ts @@ -0,0 +1,135 @@ +import type { EventSourceMessage } from 'eventsource-parser' +import { EventSourceParserStream } from 'eventsource-parser/stream' + +const SERVICE_API_SSE_MAX_BUFFER_SIZE = 1_048_576 +const SERVICE_API_SSE_SUMMARY_LIMIT = 2_000 +const SUCCESS_TERMINAL_EVENTS = new Set(['message_end', 'workflow_finished']) + +export const SERVICE_API_STREAM_TIMEOUT_MS = 120_000 +export const SERVICE_API_RUNTIME_STEP_TIMEOUT_MS = SERVICE_API_STREAM_TIMEOUT_MS + 10_000 + +export type ServiceApiSseEvent = { + data: unknown + event?: string + id?: string +} + +export type ServiceApiSseResult = { + answer: string + events: ServiceApiSseEvent[] +} + +const isRecord = (value: unknown): value is Record => ( + Boolean(value) && typeof value === 'object' && !Array.isArray(value) +) + +const truncate = (value: string, limit = SERVICE_API_SSE_SUMMARY_LIMIT) => ( + value.length > limit ? `${value.slice(0, limit)}...` : value +) + +const parseEventData = (event: EventSourceMessage) => { + try { + return JSON.parse(event.data) as unknown + } + catch (error) { + throw new Error( + `Agent v2 Service API SSE event contained invalid JSON: ${truncate(JSON.stringify(event.data))}`, + { cause: error }, + ) + } +} + +const getEventName = (event: EventSourceMessage, data: unknown) => { + if (event.event) + return event.event + if (isRecord(data) && typeof data.event === 'string') + return data.event + return undefined +} + +const summarizeEvents = (events: ServiceApiSseEvent[]) => truncate(JSON.stringify(events.map((event) => { + if (!isRecord(event.data)) + return { event: event.event, data: truncate(String(event.data), 200) } + + const data = event.data + return { + event: event.event, + ...(['code', 'message', 'conversation_id', 'message_id', 'task_id'] as const).reduce>( + (summary, key) => { + if (key in data) + summary[key] = data[key] + return summary + }, + {}, + ), + } +}))) + +const createBackendError = (event: ServiceApiSseEvent, events: ServiceApiSseEvent[]) => { + const data = event.data + const details = isRecord(data) + ? (['code', 'message', 'conversation_id', 'message_id', 'task_id'] as const) + .flatMap(key => key in data ? [`${key}=${JSON.stringify(data[key])}`] : []) + .join(' ') + : `data=${JSON.stringify(data)}` + + return new Error( + `Agent v2 Service API SSE error${details ? `: ${details}` : ''}. Received events: ${summarizeEvents(events)}`, + ) +} + +export async function consumeServiceApiSse( + body: ReadableStream | null, +): Promise { + if (!body) + throw new Error('Agent v2 Service API SSE response did not expose a readable body.') + + const events: ServiceApiSseEvent[] = [] + const answers: string[] = [] + const stream = body + .pipeThrough(new TextDecoderStream()) + .pipeThrough(new EventSourceParserStream({ + maxBufferSize: SERVICE_API_SSE_MAX_BUFFER_SIZE, + onError: 'terminate', + })) + + try { + for await (const message of stream) { + const data = parseEventData(message) + const eventName = getEventName(message, data) + const event: ServiceApiSseEvent = { + data, + ...(eventName ? { event: eventName } : {}), + ...(message.id ? { id: message.id } : {}), + } + events.push(event) + + if (isRecord(data) && typeof data.answer === 'string') + answers.push(data.answer) + + if (eventName === 'error') + throw createBackendError(event, events) + + if (eventName && SUCCESS_TERMINAL_EVENTS.has(eventName)) { + return { + answer: answers.join(''), + events, + } + } + } + } + catch (error) { + if (error instanceof Error && error.message.startsWith('Agent v2 Service API SSE')) + throw error + + const message = error instanceof Error ? error.message : String(error) + throw new Error( + `Agent v2 Service API SSE parsing failed: ${message}. Received events: ${summarizeEvents(events)}`, + { cause: error }, + ) + } + + throw new Error( + `Agent v2 Service API SSE stream closed before a terminal event. Received events: ${summarizeEvents(events)}`, + ) +} diff --git a/e2e/features/step-definitions/agent-v2/access-point-service-api.steps.ts b/e2e/features/step-definitions/agent-v2/access-point-service-api.steps.ts index a61301eb413..1f4513b9ff0 100644 --- a/e2e/features/step-definitions/agent-v2/access-point-service-api.steps.ts +++ b/e2e/features/step-definitions/agent-v2/access-point-service-api.steps.ts @@ -7,6 +7,7 @@ import { setAgentApiAccess, } from '../../agent-v2/support/access-point' import { agentBuilderExpectedTokens, agentBuilderFixedInputs } from '../../agent-v2/support/agent-builder-resources' +import { SERVICE_API_RUNTIME_STEP_TIMEOUT_MS } from '../../agent-v2/support/service-api-sse' import { getCurrentAgentId, getServiceApiCard } from './access-point-helpers' async function enableAgentApiAccessWithKey(world: DifyWorld) { @@ -172,34 +173,42 @@ Then('the Agent v2 API Reference should open in a new tab', async function (this this.agentBuilder.accessPoint.apiReferencePage = undefined }) -When('I send the Agent v2 Backend service API minimal request', async function (this: DifyWorld) { - const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL - const apiKey = this.agentBuilder.accessPoint.generatedApiKey - if (!serviceApiBaseURL) - throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.') - if (!apiKey) - throw new Error('No Agent v2 API key found. Create a Backend service API key first.') +When( + 'I send the Agent v2 Backend service API minimal request', + { timeout: SERVICE_API_RUNTIME_STEP_TIMEOUT_MS }, + async function (this: DifyWorld) { + const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL + const apiKey = this.agentBuilder.accessPoint.generatedApiKey + if (!serviceApiBaseURL) + throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.') + if (!apiKey) + throw new Error('No Agent v2 API key found. Create a Backend service API key first.') - this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({ - apiKey, - serviceApiBaseURL, - }) -}) + this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({ + apiKey, + serviceApiBaseURL, + }) + }, +) -When('I send the Agent v2 Backend service API knowledge request', async function (this: DifyWorld) { - const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL - const apiKey = this.agentBuilder.accessPoint.generatedApiKey - if (!serviceApiBaseURL) - throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.') - if (!apiKey) - throw new Error('No Agent v2 API key found. Create a Backend service API key first.') +When( + 'I send the Agent v2 Backend service API knowledge request', + { timeout: SERVICE_API_RUNTIME_STEP_TIMEOUT_MS }, + async function (this: DifyWorld) { + const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL + const apiKey = this.agentBuilder.accessPoint.generatedApiKey + if (!serviceApiBaseURL) + throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.') + if (!apiKey) + throw new Error('No Agent v2 API key found. Create a Backend service API key first.') - this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({ - apiKey, - query: agentBuilderFixedInputs.knowledgeRuntimeQuery, - serviceApiBaseURL, - }) -}) + this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({ + apiKey, + query: agentBuilderFixedInputs.knowledgeRuntimeQuery, + serviceApiBaseURL, + }) + }, +) const stringifyServiceApiBody = (body: unknown) => { try { diff --git a/e2e/features/step-definitions/agent-v2/tools.steps.ts b/e2e/features/step-definitions/agent-v2/tools.steps.ts index b8a78018c96..41718724fb8 100644 --- a/e2e/features/step-definitions/agent-v2/tools.steps.ts +++ b/e2e/features/step-definitions/agent-v2/tools.steps.ts @@ -9,6 +9,7 @@ import { createAgentSoulConfigWithDifyTool, createAgentSoulConfigWithModel, norm import { getPreseededOAuthToolConfig } from '../../agent-v2/support/preflight/agents' import { asArray, asRecord, asString } from '../../agent-v2/support/preflight/common' import { hasToolEntry } from '../../agent-v2/support/preflight/tools' +import { SERVICE_API_RUNTIME_STEP_TIMEOUT_MS } from '../../agent-v2/support/service-api-sse' import { getPreseededToolContract } from '../../agent-v2/support/tools' import { expectProviderToolActionVisible, getCurrentAgentId } from './configure-helpers' @@ -271,24 +272,28 @@ Then( }, ) -When('I send the Agent v2 Backend service API JSON Replace request', async function (this: DifyWorld) { - const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL - const apiKey = this.agentBuilder.accessPoint.generatedApiKey - if (!serviceApiBaseURL) - throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.') - if (!apiKey) - throw new Error('No Agent v2 API key found. Create a Backend service API key first.') +When( + 'I send the Agent v2 Backend service API JSON Replace request', + { timeout: SERVICE_API_RUNTIME_STEP_TIMEOUT_MS }, + async function (this: DifyWorld) { + const serviceApiBaseURL = this.agentBuilder.accessPoint.serviceApiBaseURL + const apiKey = this.agentBuilder.accessPoint.generatedApiKey + if (!serviceApiBaseURL) + throw new Error('No Agent v2 service API endpoint found. Enable Backend service API first.') + if (!apiKey) + throw new Error('No Agent v2 API key found. Create a Backend service API key first.') - this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({ - apiKey, - query: [ - 'Verify JSON Replace now.', - `Replace ${agentBuilderExpectedTokens.jsonToolBefore} with ${agentBuilderExpectedTokens.jsonToolAfter}.`, - `Return the JSON result and include ${agentBuilderExpectedTokens.jsonToolAfter}.`, - ].join(' '), - serviceApiBaseURL, - }) -}) + this.agentBuilder.accessPoint.serviceApiResponse = await sendAgentServiceApiChatMessage({ + apiKey, + query: [ + 'Verify JSON Replace now.', + `Replace ${agentBuilderExpectedTokens.jsonToolBefore} with ${agentBuilderExpectedTokens.jsonToolAfter}.`, + `Return the JSON result and include ${agentBuilderExpectedTokens.jsonToolAfter}.`, + ].join(' '), + serviceApiBaseURL, + }) + }, +) Then( 'the Agent v2 Backend service API response should include the JSON Replace E2E marker', diff --git a/e2e/package.json b/e2e/package.json index 61120b1ba83..d584a515a5d 100644 --- a/e2e/package.json +++ b/e2e/package.json @@ -14,6 +14,7 @@ "e2e:middleware:up": "tsx ./scripts/setup.ts middleware-up", "e2e:reset": "tsx ./scripts/setup.ts reset", "seed": "tsx ./scripts/seed.ts", + "test:unit": "vitest run", "type-check": "tsc" }, "devDependencies": { @@ -24,10 +25,12 @@ "@t3-oss/env-core": "catalog:", "@types/node": "catalog:", "@typescript/native": "catalog:", + "eventsource-parser": "catalog:", "tsx": "catalog:", "typescript": "catalog:", "vite": "catalog:", "vite-plus": "catalog:", + "vitest": "catalog:", "zod": "catalog:" } } diff --git a/e2e/tests/service-api-sse.test.ts b/e2e/tests/service-api-sse.test.ts new file mode 100644 index 00000000000..fb57d1b6d44 --- /dev/null +++ b/e2e/tests/service-api-sse.test.ts @@ -0,0 +1,97 @@ +import assert from 'node:assert/strict' +import { describe, it } from 'vitest' +import { consumeServiceApiSse } from '../features/agent-v2/support/service-api-sse' + +const encodeChunks = (text: string, splitPoints: number[]) => { + const bytes = new TextEncoder().encode(text) + const chunks: Uint8Array[] = [] + let start = 0 + + for (const end of splitPoints) { + chunks.push(bytes.slice(start, end)) + start = end + } + + chunks.push(bytes.slice(start)) + return chunks +} + +const streamFromChunks = (chunks: Uint8Array[]) => new ReadableStream>({ + start(controller) { + for (const chunk of chunks) + controller.enqueue(chunk) + controller.close() + }, +}) + +describe('consumeServiceApiSse', () => { + it('parses UTF-8 events split across byte chunks and multiline data fields', async () => { + const stream = [ + 'data: {"event":"message","answer":"测试"}\n\n', + 'event: message_end\n', + 'data: {\n', + 'data: "event": "message_end",\n', + 'data: "conversation_id": "conversation-1"\n', + 'data: }\n\n', + ].join('') + const result = await consumeServiceApiSse(streamFromChunks(encodeChunks(stream, [1, 7, 31, 43, 58]))) + + assert.equal(result.answer, '测试') + assert.equal(result.events.length, 2) + assert.deepEqual(result.events[1], { + data: { + conversation_id: 'conversation-1', + event: 'message_end', + }, + event: 'message_end', + }) + }) + + it('reports SSE error details immediately', async () => { + const stream = streamFromChunks(encodeChunks( + 'data: {"event":"error","code":"internal_server_error","message":"workspace failed","conversation_id":"conversation-1","message_id":"message-1"}\n\n', + [13, 37, 81], + )) + + await assert.rejects( + consumeServiceApiSse(stream), + (error: unknown) => { + assert.ok(error instanceof Error) + assert.match(error.message, /internal_server_error/) + assert.match(error.message, /workspace failed/) + assert.match(error.message, /conversation-1/) + assert.match(error.message, /message-1/) + return true + }, + ) + }) + + it('rejects invalid JSON event data', async () => { + const stream = streamFromChunks(encodeChunks('data: {not-json}\n\n', [])) + + await assert.rejects(consumeServiceApiSse(stream), /invalid JSON/i) + }) + + it('rejects a stream that closes before a terminal event', async () => { + const stream = streamFromChunks(encodeChunks( + 'data: {"event":"message","answer":"partial"}\n\n', + [], + )) + + await assert.rejects( + consumeServiceApiSse(stream), + /closed before a terminal event.*message/s, + ) + }) + + it('rejects a response without a readable body', async () => { + await assert.rejects(consumeServiceApiSse(null), /readable body/i) + }) + + it('bounds incomplete event buffering', async () => { + const oversizedEvent = `data: ${'x'.repeat(1_048_577)}` + const stream = streamFromChunks(encodeChunks(oversizedEvent, [512_000])) + + await assert.rejects(consumeServiceApiSse(stream), /buffer/i) + }) +}) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 91980e6dd20..e11645e0eaa 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -786,6 +786,9 @@ importers: '@typescript/native': specifier: 'catalog:' version: typescript@7.0.2 + eventsource-parser: + specifier: 'catalog:' + version: 3.1.0 tsx: specifier: 'catalog:' version: 4.23.0 @@ -798,6 +801,9 @@ importers: vite-plus: specifier: 'catalog:' version: 0.2.4(@types/node@25.9.5)(@typescript/typescript6@6.0.2)(@vitest/browser-playwright@4.1.10)(@vitest/coverage-v8@4.1.10)(@voidzero-dev/vite-plus-core@0.2.4(@types/node@25.9.5)(@typescript/typescript6@6.0.2)(esbuild@0.28.1)(jiti@2.7.0)(tsx@4.23.0)(yaml@2.9.0))(esbuild@0.28.1)(happy-dom@20.10.6)(jiti@2.7.0)(tsx@4.23.0)(yaml@2.9.0) + vitest: + specifier: 'catalog:' + version: 4.1.10(@types/node@25.9.5)(@vitest/browser-playwright@4.1.10)(@vitest/browser-preview@4.1.10)(@vitest/coverage-v8@4.1.10)(@voidzero-dev/vite-plus-core@0.2.4(@types/node@25.9.5)(@typescript/typescript6@6.0.2)(esbuild@0.28.1)(jiti@2.7.0)(tsx@4.23.0)(yaml@2.9.0))(happy-dom@20.10.6) zod: specifier: 'catalog:' version: 4.4.3 @@ -8252,7 +8258,6 @@ packages: os: linux libc: musl version: 22.23.1 - hasBin: true normalize-package-data@8.0.0: resolution: {integrity: sha512-RWk+PI433eESQ7ounYxIp67CYuVsS1uYSonX3kA6ps/3LWfjVQa/ptEg6Y3T6uAMq1mWpX9PQ+qx+QaHpsc7gQ==} @@ -19771,7 +19776,6 @@ time: knip@6.25.0: '2026-07-07T08:47:13.416Z' ky@2.0.2: '2026-04-21T08:58:46.923Z' lamejs@1.2.1: '2021-12-02T15:44:40.036Z' - lexical-code-no-prism@0.41.0: '2026-03-08T16:50:40.266Z' lexical@0.46.0: '2026-06-26T04:53:00.532Z' lockfile@1.0.4: '2018-04-17T00:36:12.565Z' loro-crdt@1.13.6: '2026-06-21T15:40:04.671Z'