diff --git a/scripts/e2e/lib/openai-chat-tools/client.mjs b/scripts/e2e/lib/openai-chat-tools/client.mjs index 50229fd0e717..a97600ffbdd9 100644 --- a/scripts/e2e/lib/openai-chat-tools/client.mjs +++ b/scripts/e2e/lib/openai-chat-tools/client.mjs @@ -19,7 +19,35 @@ if (!Number.isFinite(maxBodyBytes) || maxBodyBytes <= 0) { throw new Error(`invalid OPENCLAW_OPENAI_CHAT_TOOLS_MAX_BODY_BYTES: ${maxBodyBytes}`); } -async function readBoundedResponseText(response, byteLimit) { +function cancelReaderSoon(reader) { + void Promise.resolve() + .then(() => reader.cancel()) + .catch(() => undefined); +} + +async function readResponseChunk(reader, timeoutPromise, markCanceled) { + const readPromise = reader.read(); + if (!timeoutPromise) { + return await readPromise; + } + + let waitingForRead = true; + const timeoutReadPromise = timeoutPromise.catch((error) => { + if (waitingForRead) { + markCanceled(); + cancelReaderSoon(reader); + } + throw error; + }); + + try { + return await Promise.race([readPromise, timeoutReadPromise]); + } finally { + waitingForRead = false; + } +} + +async function readBoundedResponseText(response, byteLimit, timeoutPromise) { const contentLength = response.headers?.get?.("content-length"); if (contentLength && /^\d+$/u.test(contentLength)) { const parsedContentLength = Number(contentLength); @@ -35,67 +63,88 @@ async function readBoundedResponseText(response, byteLimit) { } const chunks = []; let totalBytes = 0; - for (;;) { - const { done, value } = await reader.read(); - if (done) { - break; + let canceled = false; + try { + for (;;) { + const { done, value } = await readResponseChunk(reader, timeoutPromise, () => { + canceled = true; + }); + if (done) { + break; + } + totalBytes += value.byteLength; + if (totalBytes > byteLimit) { + canceled = true; + await reader.cancel(); + throw new Error(`chat completions response body exceeded ${byteLimit} bytes`); + } + chunks.push(Buffer.from(value)); } - totalBytes += value.byteLength; - if (totalBytes > byteLimit) { - await reader.cancel(); - throw new Error(`chat completions response body exceeded ${byteLimit} bytes`); + } finally { + if (!canceled) { + reader.releaseLock(); } - chunks.push(Buffer.from(value)); } return Buffer.concat(chunks, totalBytes).toString("utf8"); } const controller = new AbortController(); -const timeout = setTimeout(() => controller.abort(), timeoutSeconds * 1000); +const timeoutError = new Error(`chat completions request timed out after ${timeoutSeconds}s`); +let timeout; +const timeoutPromise = new Promise((_, reject) => { + timeout = setTimeout(() => { + controller.abort(timeoutError); + reject(timeoutError); + }, timeoutSeconds * 1000); + timeout.unref?.(); +}); const started = Date.now(); let response; let text; try { - response = await fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { - method: "POST", - headers: { - authorization: `Bearer ${token}`, - "content-type": "application/json", - "x-openclaw-model": backendModel, - }, - body: JSON.stringify({ - model: "openclaw", - stream: false, - messages: [ - { - role: "user", - content: - "Use the get_weather tool exactly once for Paris, France. Return the tool call only.", - }, - ], - tool_choice: "auto", - tools: [ - { - type: "function", - function: { - name: "get_weather", - description: "Return weather for a city.", - strict: true, - parameters: { - type: "object", - additionalProperties: false, - properties: { - city: { type: "string", description: "City and country." }, + response = await Promise.race([ + fetch(`http://127.0.0.1:${port}/v1/chat/completions`, { + method: "POST", + headers: { + authorization: `Bearer ${token}`, + "content-type": "application/json", + "x-openclaw-model": backendModel, + }, + body: JSON.stringify({ + model: "openclaw", + stream: false, + messages: [ + { + role: "user", + content: + "Use the get_weather tool exactly once for Paris, France. Return the tool call only.", + }, + ], + tool_choice: "auto", + tools: [ + { + type: "function", + function: { + name: "get_weather", + description: "Return weather for a city.", + strict: true, + parameters: { + type: "object", + additionalProperties: false, + properties: { + city: { type: "string", description: "City and country." }, + }, + required: ["city"], }, - required: ["city"], }, }, - }, - ], + ], + }), + signal: controller.signal, }), - signal: controller.signal, - }); - text = await readBoundedResponseText(response, maxBodyBytes); + timeoutPromise, + ]); + text = await readBoundedResponseText(response, maxBodyBytes, timeoutPromise); } finally { clearTimeout(timeout); } diff --git a/test/scripts/openai-chat-tools-client.test.ts b/test/scripts/openai-chat-tools-client.test.ts index 1f6b9020e756..8b40cf7cdc59 100644 --- a/test/scripts/openai-chat-tools-client.test.ts +++ b/test/scripts/openai-chat-tools-client.test.ts @@ -139,11 +139,16 @@ function toolCallResponse(messageOverrides: Record = {}) { describe("scripts/e2e/lib/openai-chat-tools/client.mjs", () => { let bodyReadTimeoutProbe: { elapsedMs: number; + responseClosed: boolean; result: ClientResult; }; beforeAll(async () => { + let responseClosed = false; const server = createServer((_request, response) => { + response.on("close", () => { + responseClosed = true; + }); response.writeHead(200, { "content-type": "application/json" }); response.write('{"choices":'); }); @@ -153,6 +158,7 @@ describe("scripts/e2e/lib/openai-chat-tools/client.mjs", () => { bodyReadTimeoutProbe = { result: await runClient(port, {}, 4_000), elapsedMs: Date.now() - startedAt, + responseClosed, }; } finally { server.close(); @@ -356,8 +362,9 @@ describe("scripts/e2e/lib/openai-chat-tools/client.mjs", () => { it("keeps the request timeout active while reading the response body", async () => { expect(bodyReadTimeoutProbe.result.error).toBeUndefined(); expect(bodyReadTimeoutProbe.result.status).not.toBe(0); - expect(bodyReadTimeoutProbe.result.stderr).toMatch(/aborted|AbortError/iu); + expect(bodyReadTimeoutProbe.result.stderr).toMatch(/timed out|aborted|AbortError/iu); expect(bodyReadTimeoutProbe.elapsedMs).toBeLessThan(3_500); + expect(bodyReadTimeoutProbe.responseClosed).toBe(true); }); it("caps chat completion response bodies before JSON parsing", async () => {