diff --git a/integration-tests/cli/qwen-serve-streaming.test.ts b/integration-tests/cli/qwen-serve-streaming.test.ts index 87d09be530..4afea18a88 100644 --- a/integration-tests/cli/qwen-serve-streaming.test.ts +++ b/integration-tests/cli/qwen-serve-streaming.test.ts @@ -7,9 +7,10 @@ /** * `qwen serve` daemon — streaming / multi-client / recovery integration. * - * These tests need a working model credential because they fire real - * prompts and observe the resulting SSE stream. They cover three flows - * that unit tests can't fully exercise: + * These tests fire real daemon prompts and observe the resulting SSE stream, + * but the model side is backed by a local OpenAI-compatible fake server so + * the suite can run without API keys. They cover three flows that unit tests + * can't fully exercise: * * 1. Real `qwen --acp` child crash → daemon publishes `session_died`, * removes the dead entry from the maps, and a subsequent @@ -21,14 +22,20 @@ * `Last-Event-ID: N` resumes the stream from id N+1 via the bus's * replay ring. * - * Skip on CI / no-auth via `SKIP_LLM_TESTS=1`. */ import { spawn, execSync, type ChildProcess } from 'node:child_process'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; import * as path from 'node:path'; import { fileURLToPath } from 'node:url'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { DaemonClient, parseSseStream } from '@qwen-code/sdk'; import type { DaemonEvent, DaemonSessionSummary } from '@qwen-code/sdk'; +import { + fakeToolCall, + startFakeOpenAIServer, + type FakeOpenAIServer, +} from '../fake-openai-server.js'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); // Match the rest of the integration suite: prefer `TEST_CLI_PATH` @@ -41,25 +48,42 @@ const CLI_BIN = const TOKEN = 'streaming-integ-secret'; const REPO_ROOT = path.resolve(__dirname, '../..'); -// Skip when: -// - explicit `SKIP_LLM_TESTS=1` (CI envs without provider API keys), OR -// - Windows: this suite shells out to `pgrep` / `kill -KILL` to -// simulate child-process crashes for the SIGKILL → `session_died` -// test, and those binaries are POSIX-only. A Windows-equivalent -// (`taskkill`) would need different test scaffolding; deferred to -// a follow-up rather than smuggling shell-shape divergence into -// the existing assertions. -const SKIP = - process.env['SKIP_LLM_TESTS'] === '1' || process.platform === 'win32'; -const describeLLM = SKIP ? describe.skip : describe; +// Windows: this suite shells out to `pgrep` / `kill -KILL` to simulate +// child-process crashes for the SIGKILL → `session_died` test, and those +// binaries are POSIX-only. A Windows-equivalent (`taskkill`) would need +// different test scaffolding. +const SKIP = process.platform === 'win32'; +const describePOSIX = SKIP ? describe.skip : describe; let daemon: ChildProcess; let port = 0; let base = ''; let client: DaemonClient; +let fakeServer: FakeOpenAIServer; +let homeDir = ''; +let pendingWritePath = ''; beforeAll(async () => { if (SKIP) return; + fakeServer = await startFakeOpenAIServer(({ body }) => { + const messages = JSON.stringify(body['messages'] ?? []); + const hasToolResult = + messages.includes('"role":"tool"') || messages.includes('"tool_call_id"'); + + if (pendingWritePath && messages.includes('fan-out') && !hasToolResult) { + return { + toolCalls: [ + fakeToolCall('write_file', { + file_path: pendingWritePath, + content: 'fan-out', + }), + ], + }; + } + + return { content: 'fake response complete' }; + }); + homeDir = mkdtempSync(path.join(tmpdir(), 'qwen-serve-streaming-home-')); daemon = spawn( process.execPath, [ @@ -77,13 +101,23 @@ beforeAll(async () => { // the test runner's cwd (CI / IDE-launcher / direct vitest // invocations all differ) and every session create returns // 400 workspace_mismatch — the SSE / permission / Last-Event-ID - // tests below would all silently 404 once `SKIP_LLM_TESTS` is - // unset. Same fix the sibling routes test received earlier in - // this PR — missed in this file in the original §02 pass. + // tests below would all silently 404. Same fix the sibling routes test + // received earlier in this PR — missed in this file in the original §02 + // pass. '--workspace', REPO_ROOT, ], - { stdio: ['ignore', 'pipe', 'pipe'] }, + { + stdio: ['ignore', 'pipe', 'pipe'], + env: { + ...process.env, + HOME: homeDir, + OPENAI_API_KEY: 'fake-key', + OPENAI_BASE_URL: fakeServer.baseUrl, + OPENAI_MODEL: 'fake-model', + QWEN_MODEL: 'fake-model', + }, + }, ); port = await new Promise((resolve, reject) => { let buf = ''; @@ -115,9 +149,14 @@ beforeAll(async () => { }, 30_000); afterAll(async () => { - if (SKIP || !daemon || daemon.exitCode !== null) return; - daemon.kill('SIGTERM'); - await new Promise((r) => daemon.once('exit', r)); + if (!SKIP && daemon && daemon.exitCode === null) { + daemon.kill('SIGTERM'); + await new Promise((r) => daemon.once('exit', r)); + } + await fakeServer?.close(); + if (homeDir) { + rmSync(homeDir, { recursive: true, force: true }); + } }, 15_000); /** Open an authenticated SSE stream and yield parsed frames. */ @@ -145,7 +184,7 @@ async function* sseFrames( yield* parseSseStream(res.body!, opts.signal); } -describeLLM('qwen serve — child-crash recovery (real SIGKILL)', () => { +describePOSIX('qwen serve — child-crash recovery (real SIGKILL)', () => { it('publishes session_died after the qwen --acp child is SIGKILL-ed', async () => { const session = await client.createOrAttachSession({ workspaceCwd: REPO_ROOT, @@ -220,7 +259,7 @@ describeLLM('qwen serve — child-crash recovery (real SIGKILL)', () => { }, 60_000); }); -describeLLM('qwen serve — multi-client first-responder permission', () => { +describePOSIX('qwen serve — multi-client first-responder permission', () => { it('fans out permission_request to both subscribers; only one vote wins', async () => { const session = await client.createOrAttachSession({ workspaceCwd: REPO_ROOT, @@ -265,90 +304,94 @@ describeLLM('qwen serve — multi-client first-responder permission', () => { await new Promise((r) => setTimeout(r, 200)); const tmp = `/tmp/qwen-serve-mc-${Date.now()}.txt`; - const promptTask = client.prompt(session.sessionId, { - prompt: [ - { - type: 'text', - text: `Please create a file at ${tmp} with contents "fan-out". After the tool runs, stop.`, - }, - ], - }); - - // Wait for both subscribers to see permission_request. - const t0 = Date.now(); - let req1: DaemonEvent | undefined; - let req2: DaemonEvent | undefined; - while (Date.now() - t0 < 30_000 && (!req1 || !req2)) { - req1 = req1 ?? seen1.find((e) => e.type === 'permission_request'); - req2 = req2 ?? seen2.find((e) => e.type === 'permission_request'); - await new Promise((r) => setTimeout(r, 100)); - } - expect(req1).toBeDefined(); - expect(req2).toBeDefined(); - const data1 = req1!.data as { - requestId: string; - options: Array<{ optionId: string; kind: string }>; - }; - const data2 = req2!.data as { requestId: string }; - expect(data1.requestId).toBe(data2.requestId); - - const optionId = - data1.options.find((o) => o.kind === 'allow_once')?.optionId ?? - data1.options[0]?.optionId; - - // Race two concurrent votes — exactly one should win. - const [voteA, voteB] = await Promise.all([ - fetch(`${base}/permission/${data1.requestId}`, { - method: 'POST', - headers: { - Authorization: `Bearer ${TOKEN}`, - 'content-type': 'application/json', - }, - body: JSON.stringify({ outcome: { outcome: 'selected', optionId } }), - }), - fetch(`${base}/permission/${data1.requestId}`, { - method: 'POST', - headers: { - Authorization: `Bearer ${TOKEN}`, - 'content-type': 'application/json', - }, - body: JSON.stringify({ outcome: { outcome: 'selected', optionId } }), - }), - ]); - expect([voteA.status, voteB.status].sort()).toEqual([200, 404]); - - // Wait for the prompt to complete (either succeed or time out). - await Promise.race([ - promptTask.catch(() => undefined), - new Promise((r) => setTimeout(r, 30_000)), - ]); - // The race above tolerates the turn still running (slow model). - // But ABANDONING an in-flight turn wedges the shared session: if - // the model asks for a SECOND permission after the allow_once - // vote, nobody is left to answer it, the pending request blocks - // the turn forever, and the per-session prompt FIFO holds every - // later prompt behind it — the Last-Event-ID resume test below - // then times out waiting for a turn_complete that never comes - // (the exact 60s × 3-retry hang from the 2026-06-12 nightly). - // Cancel the active prompt so the session is clean for the next - // test; harmless when the turn already finished. - await client.cancel(session.sessionId).catch(() => undefined); - await Promise.race([ - promptTask.catch(() => undefined), - new Promise((r) => setTimeout(r, 5_000)), - ]); - ac1.abort(); - ac2.abort(); - await Promise.all([sub1, sub2]); + pendingWritePath = tmp; + let promptTask: Promise | undefined; try { - execSync(`rm -f ${tmp}`); - } catch { - /* file may not exist if the tool didn't run */ + promptTask = client.prompt(session.sessionId, { + prompt: [ + { + type: 'text', + text: `Please create a file at ${tmp} with contents "fan-out". After the tool runs, stop.`, + }, + ], + }); + + // Wait for both subscribers to see permission_request. + const t0 = Date.now(); + let req1: DaemonEvent | undefined; + let req2: DaemonEvent | undefined; + while (Date.now() - t0 < 30_000 && (!req1 || !req2)) { + req1 = req1 ?? seen1.find((e) => e.type === 'permission_request'); + req2 = req2 ?? seen2.find((e) => e.type === 'permission_request'); + await new Promise((r) => setTimeout(r, 100)); + } + expect(req1).toBeDefined(); + expect(req2).toBeDefined(); + const data1 = req1!.data as { + requestId: string; + options: Array<{ optionId: string; kind: string }>; + }; + const data2 = req2!.data as { requestId: string }; + expect(data1.requestId).toBe(data2.requestId); + + const optionId = + data1.options.find((o) => o.kind === 'allow_once')?.optionId ?? + data1.options[0]?.optionId; + + // Race two concurrent votes — exactly one should win. + const [voteA, voteB] = await Promise.all([ + fetch(`${base}/permission/${data1.requestId}`, { + method: 'POST', + headers: { + Authorization: `Bearer ${TOKEN}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ outcome: { outcome: 'selected', optionId } }), + }), + fetch(`${base}/permission/${data1.requestId}`, { + method: 'POST', + headers: { + Authorization: `Bearer ${TOKEN}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ outcome: { outcome: 'selected', optionId } }), + }), + ]); + expect([voteA.status, voteB.status].sort()).toEqual([200, 404]); + + // Wait for the prompt to complete (either succeed or time out). + await Promise.race([ + promptTask.catch(() => undefined), + new Promise((r) => setTimeout(r, 30_000)), + ]); + } finally { + // The race above tolerates the turn still running (slow model). + // But ABANDONING an in-flight turn wedges the shared session: if + // the model asks for a SECOND permission after the allow_once + // vote, nobody is left to answer it, the pending request blocks + // the turn forever, and the per-session prompt FIFO holds every + // later prompt behind it — the Last-Event-ID resume test below + // then times out waiting for a turn_complete that never comes + // (the exact 60s × 3-retry hang from the 2026-06-12 nightly). + // Cancel the active prompt so the session is clean for the next + // test; harmless when the turn already finished. + await client.cancel(session.sessionId).catch(() => undefined); + if (promptTask) { + await Promise.race([ + promptTask.catch(() => undefined), + new Promise((r) => setTimeout(r, 5_000)), + ]); + } + ac1.abort(); + ac2.abort(); + await Promise.all([sub1, sub2]); + rmSync(tmp, { force: true }); + pendingWritePath = ''; } }, 90_000); }); -describeLLM('qwen serve — Last-Event-ID resume', () => { +describePOSIX('qwen serve — Last-Event-ID resume', () => { it('reconnect with Last-Event-ID:N yields events with id > N', async () => { const session = await client.createOrAttachSession({ workspaceCwd: REPO_ROOT, diff --git a/integration-tests/fake-openai-server.test.ts b/integration-tests/fake-openai-server.test.ts new file mode 100644 index 0000000000..f5417a1e14 --- /dev/null +++ b/integration-tests/fake-openai-server.test.ts @@ -0,0 +1,136 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { afterEach, describe, expect, it } from 'vitest'; +import { + fakeToolCall, + startFakeOpenAIServer, + type FakeOpenAIServer, +} from './fake-openai-server.js'; + +let server: FakeOpenAIServer | undefined; + +afterEach(async () => { + await server?.close(); + server = undefined; +}); + +describe('fake OpenAI server', () => { + it('serves non-streaming and streaming chat completions', async () => { + server = await startFakeOpenAIServer(({ requestIndex }) => + requestIndex === 0 + ? { content: 'hello from fake model' } + : { + toolCalls: [ + fakeToolCall('write_file', { + file_path: '/tmp/fake.txt', + content: 'fake', + }), + ], + }, + ); + + const nonStreaming = await fetch(`${server.baseUrl}/chat/completions`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + model: 'fake-model', + messages: [{ role: 'user', content: 'hi' }], + }), + }); + expect(nonStreaming.status).toBe(200); + await expect(nonStreaming.json()).resolves.toMatchObject({ + choices: [ + { + message: { + role: 'assistant', + content: 'hello from fake model', + }, + finish_reason: 'stop', + }, + ], + }); + + const streaming = await fetch(`${server.baseUrl}/chat/completions`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + model: 'fake-model', + stream: true, + messages: [{ role: 'user', content: 'write' }], + }), + }); + expect(streaming.status).toBe(200); + const streamText = await streaming.text(); + expect(streamText).toContain('"tool_calls"'); + expect(streamText).toContain('"write_file"'); + expect(streamText).toContain('data: [DONE]'); + expect(server.requests).toHaveLength(2); + }); + + it('returns 404 for wrong methods or paths', async () => { + server = await startFakeOpenAIServer(() => ({ content: 'unused' })); + + const response = await fetch(`${server.baseUrl}/chat/completions`, { + method: 'GET', + }); + + expect(response.status).toBe(404); + }); + + it('returns 400 for malformed JSON bodies', async () => { + server = await startFakeOpenAIServer(() => ({ content: 'unused' })); + + const response = await fetch(`${server.baseUrl}/chat/completions`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: 'not json', + }); + + expect(response.status).toBe(400); + }); + + it('returns 500 without exposing handler error details', async () => { + server = await startFakeOpenAIServer(() => { + throw new Error('secret stack detail'); + }); + + const response = await fetch(`${server.baseUrl}/chat/completions`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + model: 'fake-model', + messages: [{ role: 'user', content: 'hi' }], + }), + }); + + expect(response.status).toBe(500); + await expect(response.json()).resolves.toMatchObject({ + error: { + message: 'fake OpenAI server handler failed', + type: 'server_error', + }, + }); + }); + + it('closes the response when streaming fails after headers are sent', async () => { + server = await startFakeOpenAIServer(() => ({ + content: 1n as unknown as string, + })); + + await expect( + fetch(`${server.baseUrl}/chat/completions`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + model: 'fake-model', + stream: true, + messages: [{ role: 'user', content: 'hi' }], + }), + }), + ).rejects.toThrow(); + }); +}); diff --git a/integration-tests/fake-openai-server.ts b/integration-tests/fake-openai-server.ts new file mode 100644 index 0000000000..f0590fc790 --- /dev/null +++ b/integration-tests/fake-openai-server.ts @@ -0,0 +1,272 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + createServer, + type IncomingMessage, + type Server, + type ServerResponse, +} from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { randomUUID } from 'node:crypto'; + +type JsonObject = Record; + +export type FakeOpenAIToolCall = { + id: string; + type: 'function'; + function: { + name: string; + arguments: string; + }; +}; + +export type FakeOpenAIResponse = { + content?: string; + toolCalls?: FakeOpenAIToolCall[]; + finishReason?: 'stop' | 'tool_calls' | 'length'; + usage?: { + prompt_tokens: number; + completion_tokens: number; + total_tokens: number; + }; +}; + +export type FakeOpenAIRequest = { + body: JsonObject; +}; + +export type FakeOpenAIServer = { + baseUrl: string; + requests: FakeOpenAIRequest[]; + close: () => Promise; +}; + +export type FakeOpenAIHandler = (ctx: { + body: JsonObject; + requestIndex: number; +}) => FakeOpenAIResponse | Promise; + +export function fakeToolCall( + name: string, + args: JsonObject, + id = `call_${randomUUID()}`, +): FakeOpenAIToolCall { + return { + id, + type: 'function', + function: { + name, + arguments: JSON.stringify(args), + }, + }; +} + +export async function startFakeOpenAIServer( + handler: FakeOpenAIHandler, +): Promise { + const requests: FakeOpenAIRequest[] = []; + const server = createServer(async (req, res) => { + if (req.method !== 'POST' || !req.url?.endsWith('/chat/completions')) { + res.writeHead(404); + res.end('not found'); + return; + } + + try { + const rawBody = await readRequestBody(req); + const body = parseJsonBody(rawBody); + if (!body) { + res.writeHead(400); + res.end('bad json'); + return; + } + + const requestIndex = requests.length; + requests.push({ body }); + + const response = await handler({ body, requestIndex }); + if (body['stream'] === true) { + writeStreamed(res, getModel(body), response); + } else { + writeNonStreamed(res, getModel(body), response); + } + } catch { + if (res.headersSent) { + if (!res.writableEnded) { + res.destroy(); + } + return; + } + + res.writeHead(500, { 'content-type': 'application/json' }); + res.end( + JSON.stringify({ + error: { + message: 'fake OpenAI server handler failed', + type: 'server_error', + }, + }), + ); + } + }); + + await new Promise((resolve, reject) => { + const onError = (error: Error) => { + server.off('listening', onListening); + reject(error); + }; + const onListening = () => { + server.off('error', onError); + resolve(); + }; + server.once('error', onError); + server.once('listening', onListening); + server.listen(0, '127.0.0.1'); + }); + + const address = server.address(); + if (!address || typeof address === 'string') { + throw new Error('failed to start fake OpenAI server'); + } + + return { + baseUrl: `http://127.0.0.1:${(address as AddressInfo).port}/v1`, + requests, + close: () => closeServer(server), + }; +} + +function readRequestBody(req: IncomingMessage): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + req.on('data', (chunk: Buffer) => chunks.push(chunk)); + req.on('end', () => resolve(Buffer.concat(chunks).toString('utf8'))); + req.on('error', reject); + }); +} + +function parseJsonBody(rawBody: string): JsonObject | null { + try { + const parsed: unknown = JSON.parse(rawBody); + return isJsonObject(parsed) ? parsed : null; + } catch { + return null; + } +} + +function isJsonObject(value: unknown): value is JsonObject { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function getModel(body: JsonObject): string { + return typeof body['model'] === 'string' ? body['model'] : 'fake-model'; +} + +function writeNonStreamed( + res: ServerResponse, + model: string, + message: FakeOpenAIResponse, +): void { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end( + JSON.stringify({ + id: chatCompletionId(), + object: 'chat.completion', + created: nowSeconds(), + model, + choices: [ + { + index: 0, + message: { + role: 'assistant', + content: message.content ?? '', + ...(message.toolCalls ? { tool_calls: message.toolCalls } : {}), + }, + finish_reason: finishReason(message), + }, + ], + usage: message.usage ?? DEFAULT_USAGE, + }), + ); +} + +function writeStreamed( + res: ServerResponse, + model: string, + message: FakeOpenAIResponse, +): void { + res.writeHead(200, { + 'cache-control': 'no-cache', + connection: 'keep-alive', + 'content-type': 'text/event-stream', + }); + + const id = chatCompletionId(); + const created = nowSeconds(); + const chunk = ( + delta: JsonObject, + finish_reason: string | null = null, + usage?: FakeOpenAIResponse['usage'], + ) => ({ + id, + object: 'chat.completion.chunk', + created, + model, + choices: [{ index: 0, delta, finish_reason }], + ...(usage ? { usage } : {}), + }); + const send = (payload: unknown) => { + res.write(`data: ${JSON.stringify(payload)}\n\n`); + }; + + send(chunk({ role: 'assistant' })); + if (message.content) { + send(chunk({ content: message.content })); + } + for (const [index, toolCall] of (message.toolCalls ?? []).entries()) { + send( + chunk({ + tool_calls: [ + { + index, + id: toolCall.id, + type: toolCall.type, + function: toolCall.function, + }, + ], + }), + ); + } + send(chunk({}, finishReason(message), message.usage ?? DEFAULT_USAGE)); + res.write('data: [DONE]\n\n'); + res.end(); +} + +function finishReason(message: FakeOpenAIResponse): string { + return message.finishReason ?? (message.toolCalls ? 'tool_calls' : 'stop'); +} + +const DEFAULT_USAGE: NonNullable = { + prompt_tokens: 0, + completion_tokens: 0, + total_tokens: 0, +}; + +function chatCompletionId(): string { + return `chatcmpl-${randomUUID()}`; +} + +function nowSeconds(): number { + return Math.floor(Date.now() / 1000); +} + +function closeServer(server: Server): Promise { + return new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + server.closeAllConnections(); + }); +}