opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-21 02:28:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

test(e2e): consume agent service API SSE incrementally (#38634)

Browse Source
This commit is contained in:
yyh 2026-07-10 12:26:04 +08:00 committed by GitHub
parent b7255e6f27
commit b325597d29
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 325 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.github/workflows/web-e2e.yml vendored
View File

@ -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

92
e2e/features/agent-v2/support/access-point.ts
View File

@ -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<AgentServiceApiChatResult> {
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
}
}

135
e2e/features/agent-v2/support/service-api-sse.ts Normal file
View File

@ -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<string, unknown> => (
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<Record<string, unknown>>(
(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<BufferSource> | null,
): Promise<ServiceApiSseResult> {
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)}`,
)
}

59
e2e/features/step-definitions/agent-v2/access-point-service-api.steps.ts
View File

@ -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 {

39
e2e/features/step-definitions/agent-v2/tools.steps.ts
View File

@ -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',

3
e2e/package.json
View File

@ -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:"
}
}

97
e2e/tests/service-api-sse.test.ts Normal file
View File

@ -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<ArrayBuffer>[] = []
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<ArrayBuffer>[]) => new ReadableStream<Uint8Array<ArrayBuffer>>({
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)
})
})

8
pnpm-lock.yaml generated
View File

@ -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'