From de088fcec661a215df76e234c5375c952e2f5c58 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sat, 24 Oct 2026 22:20:51 +0800 Subject: [PATCH] fix(verifiers): join model fixture request finalization Keep each admitted HTTP handler owned through Evidence writes and error rendering, including after its client disconnects. Close joins graceful transport shutdown and handler work before publishing completion, and propagates exact late failures while preserving ordinary request diagnostics. Share lifecycle ownership across OpenAI, Anthropic, and Gemini fixtures while retaining protocol-specific parsing and responses. Exercise real disconnects, graceful concurrent and repeated close, ordinary 400s, and opaque late failures. --- .../clients/anthropic-model-fixture.mjs | 263 +++++----- .../verify/clients/gemini-model-fixture.mjs | 172 +++--- .../verify/clients/model-fixture-server.d.mts | 16 + .../verify/clients/model-fixture-server.mjs | 87 +++ scripts/verify/clients/model-fixture.mjs | 288 +++++----- .../process/modelFixtureShutdown.test.ts | 494 ++++++++++++++++++ 6 files changed, 961 insertions(+), 359 deletions(-) create mode 100644 scripts/verify/clients/model-fixture-server.d.mts create mode 100644 scripts/verify/clients/model-fixture-server.mjs create mode 100644 tests/boundary/process/modelFixtureShutdown.test.ts diff --git a/scripts/verify/clients/anthropic-model-fixture.mjs b/scripts/verify/clients/anthropic-model-fixture.mjs index 8f8e448c92..30e1f1ce41 100644 --- a/scripts/verify/clients/anthropic-model-fixture.mjs +++ b/scripts/verify/clients/anthropic-model-fixture.mjs @@ -1,158 +1,159 @@ import assert from "node:assert/strict"; -import { createServer } from "node:http"; import { writeFile } from "node:fs/promises"; import { join } from "node:path"; +import { createModelFixtureServer } from "./model-fixture-server.mjs"; /** Exercise native Anthropic adapters locally, without live-provider claims. */ export const createAnthropicModelFixture = async ({ directory, onRequest }) => { const requests = []; const probes = []; let incoming = 0; - let failure; - const server = createServer(async (request, response) => { - try { - const chunks = []; - let bytes = 0; - for await (const chunk of request) { - bytes += chunk.length; - assert( - bytes <= 10 * 1024 * 1024, - "Model fixture request exceeds 10 MiB budget", - ); - chunks.push(chunk); - } - const raw = Buffer.concat(chunks).toString("utf8"); - const body = raw.length === 0 ? null : JSON.parse(raw); - const path = new URL(request.url, "http://localhost").pathname; - // Retain producer bodies and URLs, without transport authentication headers. - await writeFile( - join(directory, `claude-incoming-${++incoming}.json`), - JSON.stringify({ method: request.method, path: request.url, body }), - ); - if (path === "/v1/messages/count_tokens") { - probes.push({ path, kind: "synthetic-token-count" }); - response.setHeader("Content-Type", "application/json"); - response.end(JSON.stringify({ input_tokens: 100 })); - return; - } - if (path !== "/v1/messages") { - probes.push({ path, status: 404 }); - response.statusCode = 404; - response.end( - JSON.stringify({ - error: { message: "Unsupported fixture endpoint" }, - }), - ); - return; - } - assert.equal(request.method, "POST"); - assert(body && typeof body === "object" && !Array.isArray(body)); - assert(Array.isArray(body.messages)); - const tools = body.tools ?? []; - requests.push({ - path: request.url, - stream: body.stream, - tools: tools.map((t) => t.name), - messageRoles: body.messages.map((m) => m.role), - }); - const content = await onRequest({ - body, - tools, - requestIndex: requests.length, - }); - const stopReason = content.some((c) => c.type === "tool_use") - ? "tool_use" - : "end_turn"; - const usage = { - input_tokens: 100, - output_tokens: 20, - cache_creation_input_tokens: 0, - cache_read_input_tokens: 0, - }; - const result = { - id: `msg_rea_${requests.length}`, - type: "message", - role: "assistant", - model: body.model, - content, - stop_reason: stopReason, - stop_sequence: null, - usage, - }; - if (body.stream) { - response.setHeader("Content-Type", "text/event-stream"); - const emit = (event, data) => - response.write( - `event: ${event}\ndata: ${JSON.stringify({ type: event, ...data })}\n\n`, + const server = await createModelFixtureServer( + async (request, response, recordFailure) => { + try { + const chunks = []; + let bytes = 0; + for await (const chunk of request) { + bytes += chunk.length; + assert( + bytes <= 10 * 1024 * 1024, + "Model fixture request exceeds 10 MiB budget", ); - emit("message_start", { - message: { - ...result, - content: [], - stop_reason: null, - usage: { ...usage, output_tokens: 0 }, - }, + chunks.push(chunk); + } + const raw = Buffer.concat(chunks).toString("utf8"); + const body = raw.length === 0 ? null : JSON.parse(raw); + const path = new URL(request.url, "http://localhost").pathname; + // Retain producer bodies and URLs, without transport authentication headers. + await writeFile( + join(directory, `claude-incoming-${++incoming}.json`), + JSON.stringify({ method: request.method, path: request.url, body }), + ); + if (path === "/v1/messages/count_tokens") { + probes.push({ path, kind: "synthetic-token-count" }); + response.setHeader("Content-Type", "application/json"); + response.end(JSON.stringify({ input_tokens: 100 })); + return; + } + if (path !== "/v1/messages") { + probes.push({ path, status: 404 }); + response.statusCode = 404; + response.end( + JSON.stringify({ + error: { message: "Unsupported fixture endpoint" }, + }), + ); + return; + } + assert.equal(request.method, "POST"); + assert(body && typeof body === "object" && !Array.isArray(body)); + assert(Array.isArray(body.messages)); + const tools = body.tools ?? []; + requests.push({ + path: request.url, + stream: body.stream, + tools: tools.map((t) => t.name), + messageRoles: body.messages.map((m) => m.role), }); - for (const [index, block] of content.entries()) { - emit("content_block_start", { - index, - content_block: - block.type === "tool_use" - ? { ...block, input: {} } - : { type: "text", text: "" }, + const content = await onRequest({ + body, + tools, + requestIndex: requests.length, + }); + const stopReason = content.some((c) => c.type === "tool_use") + ? "tool_use" + : "end_turn"; + const usage = { + input_tokens: 100, + output_tokens: 20, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }; + const result = { + id: `msg_rea_${requests.length}`, + type: "message", + role: "assistant", + model: body.model, + content, + stop_reason: stopReason, + stop_sequence: null, + usage, + }; + if (body.stream) { + response.setHeader("Content-Type", "text/event-stream"); + const emit = (event, data) => + response.write( + `event: ${event}\ndata: ${JSON.stringify({ type: event, ...data })}\n\n`, + ); + emit("message_start", { + message: { + ...result, + content: [], + stop_reason: null, + usage: { ...usage, output_tokens: 0 }, + }, }); - if (block.type === "tool_use") { - const json = JSON.stringify(block.input); - const split = Math.floor(json.length / 2); - for (const partialJson of [ - json.slice(0, split), - json.slice(split), - ]) { + for (const [index, block] of content.entries()) { + emit("content_block_start", { + index, + content_block: + block.type === "tool_use" + ? { ...block, input: {} } + : { type: "text", text: "" }, + }); + if (block.type === "tool_use") { + const json = JSON.stringify(block.input); + const split = Math.floor(json.length / 2); + for (const partialJson of [ + json.slice(0, split), + json.slice(split), + ]) { + emit("content_block_delta", { + index, + delta: { + type: "input_json_delta", + partial_json: partialJson, + }, + }); + } + } else { emit("content_block_delta", { index, - delta: { type: "input_json_delta", partial_json: partialJson }, + delta: { type: "text_delta", text: block.text }, }); } - } else { - emit("content_block_delta", { - index, - delta: { type: "text_delta", text: block.text }, - }); + emit("content_block_stop", { index }); } - emit("content_block_stop", { index }); + emit("message_delta", { + delta: { stop_reason: stopReason, stop_sequence: null }, + usage: { output_tokens: 20 }, + }); + emit("message_stop", {}); + response.end(); + } else { + response.setHeader("Content-Type", "application/json"); + response.end(JSON.stringify(result)); } - emit("message_delta", { - delta: { stop_reason: stopReason, stop_sequence: null }, - usage: { output_tokens: 20 }, - }); - emit("message_stop", {}); - response.end(); - } else { + } catch (cause) { + const failure = recordFailure(cause); + response.statusCode = 400; response.setHeader("Content-Type", "application/json"); - response.end(JSON.stringify(result)); + response.end( + JSON.stringify({ + type: "error", + error: { type: "invalid_request_error", message: failure }, + }), + ); } - } catch (cause) { - failure = String(cause); - response.statusCode = 400; - response.setHeader("Content-Type", "application/json"); - response.end( - JSON.stringify({ - type: "error", - error: { type: "invalid_request_error", message: failure }, - }), - ); - } - }); - await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); - const address = server.address(); - assert(address !== null && typeof address !== "string"); + }, + ); return { - baseUrl: `http://127.0.0.1:${address.port}`, + baseUrl: server.origin, requests, probes, get failure() { - return failure; + return server.failure; }, - close: () => new Promise((resolve) => server.close(resolve)), + close: () => server.close(), }; }; diff --git a/scripts/verify/clients/gemini-model-fixture.mjs b/scripts/verify/clients/gemini-model-fixture.mjs index 58b21302da..bbfc0617b0 100644 --- a/scripts/verify/clients/gemini-model-fixture.mjs +++ b/scripts/verify/clients/gemini-model-fixture.mjs @@ -1,104 +1,108 @@ import assert from "node:assert/strict"; -import { createServer } from "node:http"; import { writeFile } from "node:fs/promises"; import { join } from "node:path"; +import { createModelFixtureServer } from "./model-fixture-server.mjs"; /** Exercise the native Gemini API adapter against a local model, without cloud claims. */ export const createGeminiModelFixture = async ({ directory, onRequest }) => { const requests = []; const probes = []; let incoming = 0; - let failure; - const server = createServer(async (request, response) => { - try { - const chunks = []; - let bytes = 0; - for await (const chunk of request) { - bytes += chunk.length; - assert( - bytes <= 10 * 1024 * 1024, - "Model fixture request exceeds 10 MiB budget", + const server = await createModelFixtureServer( + async (request, response, recordFailure) => { + try { + const chunks = []; + let bytes = 0; + for await (const chunk of request) { + bytes += chunk.length; + assert( + bytes <= 10 * 1024 * 1024, + "Model fixture request exceeds 10 MiB budget", + ); + chunks.push(chunk); + } + const raw = Buffer.concat(chunks).toString("utf8"); + const body = raw.length === 0 ? null : JSON.parse(raw); + await writeFile( + join(directory, `gemini-incoming-${++incoming}.json`), + JSON.stringify({ method: request.method, path: request.url, body }), ); - chunks.push(chunk); - } - const raw = Buffer.concat(chunks).toString("utf8"); - const body = raw.length === 0 ? null : JSON.parse(raw); - await writeFile( - join(directory, `gemini-incoming-${++incoming}.json`), - JSON.stringify({ method: request.method, path: request.url, body }), - ); - const path = new URL(request.url, "http://localhost").pathname; - if (path.endsWith(":countTokens")) { - probes.push({ path, kind: "synthetic-token-count" }); - response.setHeader("Content-Type", "application/json"); - response.end(JSON.stringify({ totalTokens: 100 })); - return; - } - if ( - !/\/models\/[^/]+:(?:streamGenerateContent|generateContent)$/u.test( + const path = new URL(request.url, "http://localhost").pathname; + if (path.endsWith(":countTokens")) { + probes.push({ path, kind: "synthetic-token-count" }); + response.setHeader("Content-Type", "application/json"); + response.end(JSON.stringify({ totalTokens: 100 })); + return; + } + if ( + !/\/models\/[^/]+:(?:streamGenerateContent|generateContent)$/u.test( + path, + ) + ) { + probes.push({ path, status: 404 }); + response.statusCode = 404; + response.end( + JSON.stringify({ + error: { message: "Unsupported fixture endpoint" }, + }), + ); + return; + } + assert.equal(request.method, "POST"); + assert(body && typeof body === "object" && !Array.isArray(body)); + assert(Array.isArray(body.contents)); + const tools = (body.tools ?? []).flatMap( + (t) => t.functionDeclarations ?? [], + ); + requests.push({ path, - ) - ) { - probes.push({ path, status: 404 }); - response.statusCode = 404; - response.end( - JSON.stringify({ - error: { message: "Unsupported fixture endpoint" }, - }), - ); - return; - } - assert.equal(request.method, "POST"); - assert(body && typeof body === "object" && !Array.isArray(body)); - assert(Array.isArray(body.contents)); - const tools = (body.tools ?? []).flatMap( - (t) => t.functionDeclarations ?? [], - ); - requests.push({ - path, - tools: tools.map((t) => t.name), - contentRoles: body.contents.map((c) => c.role), - }); - const parts = await onRequest({ - body, - tools, - requestIndex: requests.length, - }); - const result = { - candidates: [ - { index: 0, content: { role: "model", parts }, finishReason: "STOP" }, - ], - usageMetadata: { - promptTokenCount: 100, - candidatesTokenCount: 20, - totalTokenCount: 120, - }, - modelVersion: "local-rea-fixture", - }; - if (path.endsWith(":streamGenerateContent")) { - response.setHeader("Content-Type", "text/event-stream"); - response.end(`data: ${JSON.stringify(result)}\n\n`); - } else { + tools: tools.map((t) => t.name), + contentRoles: body.contents.map((c) => c.role), + }); + const parts = await onRequest({ + body, + tools, + requestIndex: requests.length, + }); + const result = { + candidates: [ + { + index: 0, + content: { role: "model", parts }, + finishReason: "STOP", + }, + ], + usageMetadata: { + promptTokenCount: 100, + candidatesTokenCount: 20, + totalTokenCount: 120, + }, + modelVersion: "local-rea-fixture", + }; + if (path.endsWith(":streamGenerateContent")) { + response.setHeader("Content-Type", "text/event-stream"); + response.end(`data: ${JSON.stringify(result)}\n\n`); + } else { + response.setHeader("Content-Type", "application/json"); + response.end(JSON.stringify(result)); + } + } catch (cause) { + const failure = recordFailure(cause); + response.statusCode = 400; response.setHeader("Content-Type", "application/json"); - response.end(JSON.stringify(result)); + response.end( + JSON.stringify({ error: { message: failure, code: 400 } }), + ); } - } catch (cause) { - failure = String(cause); - response.statusCode = 400; - response.setHeader("Content-Type", "application/json"); - response.end(JSON.stringify({ error: { message: failure, code: 400 } })); - } - }); - await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); - const address = server.address(); - assert(address !== null && typeof address !== "string"); + }, + ); return { - baseUrl: `http://127.0.0.1:${address.port}`, + baseUrl: server.origin, requests, probes, get failure() { - return failure; + return server.failure; }, - close: () => new Promise((resolve) => server.close(resolve)), + close: () => server.close(), }; }; diff --git a/scripts/verify/clients/model-fixture-server.d.mts b/scripts/verify/clients/model-fixture-server.d.mts new file mode 100644 index 0000000000..c3f9d50215 --- /dev/null +++ b/scripts/verify/clients/model-fixture-server.d.mts @@ -0,0 +1,16 @@ +import type { IncomingMessage, ServerResponse } from "node:http"; + +export interface ModelFixtureServer { + readonly origin: string; + readonly failure: string | undefined; + close(): Promise; +} + +/** Own the HTTP transport and every admitted protocol handler through finalization. */ +export function createModelFixtureServer( + handleRequest: ( + request: IncomingMessage, + response: ServerResponse, + recordFailure: (cause: unknown) => string, + ) => Promise, +): Promise; diff --git a/scripts/verify/clients/model-fixture-server.mjs b/scripts/verify/clients/model-fixture-server.mjs new file mode 100644 index 0000000000..2b0dd7a3e9 --- /dev/null +++ b/scripts/verify/clients/model-fixture-server.mjs @@ -0,0 +1,87 @@ +import assert from "node:assert/strict"; +import { createServer } from "node:http"; + +/** Own the HTTP transport and every admitted protocol handler through finalization. */ +export async function createModelFixtureServer(handleRequest) { + const active = new Set(); + let firstFailure; + let firstLateFailure; + let closing = false; + let transportClosed = false; + let closeAttempt; + const recordFailure = (cause) => { + const observation = { cause, message: String(cause) }; + firstFailure ??= observation; + if (closing) firstLateFailure ??= observation; + return observation.message; + }; + const server = createServer((request, response) => { + if (closing) { + response.destroy(); + return; + } + const work = Promise.resolve() + .then(() => handleRequest(request, response, recordFailure)) + .catch((cause) => { + // Protocol handlers render their own errors. An escaping failure still + // belongs to this request, including failure while rendering that error. + recordFailure(cause); + response.destroy(); + }); + active.add(work); + void work.then(() => active.delete(work)); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { + server.off("error", reject); + resolve(); + }); + }); + const address = server.address(); + assert(address !== null && typeof address !== "string"); + return { + origin: `http://127.0.0.1:${address.port}`, + get failure() { + return firstFailure?.message; + }, + close() { + if (closeAttempt !== undefined) return closeAttempt; + closing = true; + const transport = transportClosed + ? Promise.resolve({ ok: true }) + : new Promise((resolve, reject) => { + server.close((cause) => { + if (cause !== undefined) { + reject(cause); + return; + } + transportClosed = true; + resolve(); + }); + }).then( + () => ({ ok: true }), + (cause) => ({ ok: false, cause }), + ); + const handlers = Promise.all(active); + closeAttempt = (async () => { + const result = await transport; + await handlers; + if (!result.ok) { + // Keep this exact server available for retry after a real transport + // close error; completed transport closure is never requested again. + closeAttempt = undefined; + if (firstLateFailure !== undefined) + throw new AggregateError( + [firstLateFailure.cause, result.cause], + "Model fixture handler and transport closure failed", + { cause: firstLateFailure.cause }, + ); + throw result.cause; + } + if (firstLateFailure !== undefined) throw firstLateFailure.cause; + })(); + return closeAttempt; + }, + }; +} diff --git a/scripts/verify/clients/model-fixture.mjs b/scripts/verify/clients/model-fixture.mjs index b218a0985e..dd9f971c7d 100644 --- a/scripts/verify/clients/model-fixture.mjs +++ b/scripts/verify/clients/model-fixture.mjs @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; -import { createServer } from "node:http"; import { writeFile } from "node:fs/promises"; import { join } from "node:path"; +import { createModelFixtureServer } from "./model-fixture-server.mjs"; /** A loopback model fixture exercises real clients without claiming live-provider compatibility. */ export const createOpenAiModelFixture = async ({ @@ -12,161 +12,161 @@ export const createOpenAiModelFixture = async ({ const requests = []; const probes = []; let incoming = 0; - let failure; - const server = createServer(async (request, response) => { - try { - const path = new URL(request.url ?? "/", "http://localhost").pathname; - if (request.method === "GET" && path.endsWith("/models")) { - response.setHeader("Content-Type", "application/json"); - response.end( - JSON.stringify({ - object: "list", - data: [ - { - id: "rea-client-fixture", - object: "model", - owned_by: "local-fixture", - }, - ], - }), - ); - return; - } - const chunks = []; - let bytes = 0; - for await (const chunk of request) { - bytes += chunk.length; - assert( - bytes <= 10 * 1024 * 1024, - "Local model fixture exceeded its 10 MiB request resource budget", - ); - chunks.push(chunk); - } - const rawBody = Buffer.concat(chunks).toString("utf8"); - const body = rawBody.length === 0 ? null : JSON.parse(rawBody); - await writeFile( - join(directory, `${prefix}-incoming-${++incoming}.json`), - JSON.stringify({ method: request.method, path: request.url, body }), - ); - if (path !== "/v1/chat/completions") { - // Hermes probes custom endpoints for Ollama/llama.cpp metadata. - // Unsupported HTTP endpoints return 404, as on an OpenAI-only server. - probes.push({ - method: request.method, - path: request.url, - body, - status: 404, - }); - response.statusCode = 404; - response.setHeader("Content-Type", "application/json"); - response.end( - JSON.stringify({ - error: - "This OpenAI fixture exposes models and chat completions only.", - }), - ); - return; - } - assert.equal(request.method, "POST"); - assert(body !== null && typeof body === "object" && !Array.isArray(body)); - assert(Array.isArray(body.messages)); - const tools = (body.tools ?? []) - .filter((t) => t.type === "function") - .map((t) => t.function ?? t); - requests.push({ - path: request.url, - tools: tools.map((t) => t.name), - stream: body.stream, - messageRoles: body.messages.map((m) => m.role), - }); - await writeFile( - join(directory, `${prefix}-request-${requests.length}.json`), - JSON.stringify(body), - ); - const message = await onRequest({ - body, - tools, - requestIndex: requests.length, - }); - const base = { - id: `chatcmpl-rea-${requests.length}`, - object: "chat.completion", - created: Math.floor(Date.now() / 1000), - model: body.model, - }; - const finish = message.tool_calls ? "tool_calls" : "stop"; - const usage = { - prompt_tokens: 100, - completion_tokens: 20, - total_tokens: 120, - }; - if (body.stream) { - response.setHeader("Content-Type", "text/event-stream"); - const emit = (delta, finishReason = null) => - response.write( - `data: ${JSON.stringify({ ...base, object: "chat.completion.chunk", choices: [{ index: 0, delta, finish_reason: finishReason }] })}\n\n`, - ); - if (message.tool_calls) { - const fragments = message.tool_calls.map((call, index) => { - const split = Math.floor(call.function.arguments.length / 2); - return { - first: { - index, - ...call, - function: { - name: call.function.name, - arguments: call.function.arguments.slice(0, split), + const server = await createModelFixtureServer( + async (request, response, recordFailure) => { + try { + const path = new URL(request.url ?? "/", "http://localhost").pathname; + if (request.method === "GET" && path.endsWith("/models")) { + response.setHeader("Content-Type", "application/json"); + response.end( + JSON.stringify({ + object: "list", + data: [ + { + id: "rea-client-fixture", + object: "model", + owned_by: "local-fixture", }, - }, - second: { - index, - function: { arguments: call.function.arguments.slice(split) }, - }, - }; - }); - emit({ - role: "assistant", - content: null, - tool_calls: fragments.map((f) => f.first), - }); - emit({ tool_calls: fragments.map((f) => f.second) }); - } else emit(message); - emit({}, finish); - response.write( - `data: ${JSON.stringify({ ...base, object: "chat.completion.chunk", choices: [], usage })}\n\n`, + ], + }), + ); + return; + } + const chunks = []; + let bytes = 0; + for await (const chunk of request) { + bytes += chunk.length; + assert( + bytes <= 10 * 1024 * 1024, + "Local model fixture exceeded its 10 MiB request resource budget", + ); + chunks.push(chunk); + } + const rawBody = Buffer.concat(chunks).toString("utf8"); + const body = rawBody.length === 0 ? null : JSON.parse(rawBody); + await writeFile( + join(directory, `${prefix}-incoming-${++incoming}.json`), + JSON.stringify({ method: request.method, path: request.url, body }), ); - response.end("data: [DONE]\n\n"); - } else { + if (path !== "/v1/chat/completions") { + // Hermes probes custom endpoints for Ollama/llama.cpp metadata. + // Unsupported HTTP endpoints return 404, as on an OpenAI-only server. + probes.push({ + method: request.method, + path: request.url, + body, + status: 404, + }); + response.statusCode = 404; + response.setHeader("Content-Type", "application/json"); + response.end( + JSON.stringify({ + error: + "This OpenAI fixture exposes models and chat completions only.", + }), + ); + return; + } + assert.equal(request.method, "POST"); + assert( + body !== null && typeof body === "object" && !Array.isArray(body), + ); + assert(Array.isArray(body.messages)); + const tools = (body.tools ?? []) + .filter((t) => t.type === "function") + .map((t) => t.function ?? t); + requests.push({ + path: request.url, + tools: tools.map((t) => t.name), + stream: body.stream, + messageRoles: body.messages.map((m) => m.role), + }); + await writeFile( + join(directory, `${prefix}-request-${requests.length}.json`), + JSON.stringify(body), + ); + const message = await onRequest({ + body, + tools, + requestIndex: requests.length, + }); + const base = { + id: `chatcmpl-rea-${requests.length}`, + object: "chat.completion", + created: Math.floor(Date.now() / 1000), + model: body.model, + }; + const finish = message.tool_calls ? "tool_calls" : "stop"; + const usage = { + prompt_tokens: 100, + completion_tokens: 20, + total_tokens: 120, + }; + if (body.stream) { + response.setHeader("Content-Type", "text/event-stream"); + const emit = (delta, finishReason = null) => + response.write( + `data: ${JSON.stringify({ ...base, object: "chat.completion.chunk", choices: [{ index: 0, delta, finish_reason: finishReason }] })}\n\n`, + ); + if (message.tool_calls) { + const fragments = message.tool_calls.map((call, index) => { + const split = Math.floor(call.function.arguments.length / 2); + return { + first: { + index, + ...call, + function: { + name: call.function.name, + arguments: call.function.arguments.slice(0, split), + }, + }, + second: { + index, + function: { arguments: call.function.arguments.slice(split) }, + }, + }; + }); + emit({ + role: "assistant", + content: null, + tool_calls: fragments.map((f) => f.first), + }); + emit({ tool_calls: fragments.map((f) => f.second) }); + } else emit(message); + emit({}, finish); + response.write( + `data: ${JSON.stringify({ ...base, object: "chat.completion.chunk", choices: [], usage })}\n\n`, + ); + response.end("data: [DONE]\n\n"); + } else { + response.setHeader("Content-Type", "application/json"); + response.end( + JSON.stringify({ + ...base, + choices: [{ index: 0, message, finish_reason: finish }], + usage, + }), + ); + } + } catch (cause) { + const failure = recordFailure(cause); + response.statusCode = 400; response.setHeader("Content-Type", "application/json"); response.end( JSON.stringify({ - ...base, - choices: [{ index: 0, message, finish_reason: finish }], - usage, + error: { message: failure, type: "invalid_request_error" }, }), ); } - } catch (cause) { - failure = String(cause); - response.statusCode = 400; - response.setHeader("Content-Type", "application/json"); - response.end( - JSON.stringify({ - error: { message: failure, type: "invalid_request_error" }, - }), - ); - } - }); - await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); - const address = server.address(); - assert(address !== null && typeof address !== "string"); + }, + ); return { - baseUrl: `http://127.0.0.1:${address.port}/v1`, + baseUrl: `${server.origin}/v1`, requests, probes, get failure() { - return failure; + return server.failure; }, - close: () => new Promise((resolve) => server.close(resolve)), + close: () => server.close(), }; }; diff --git a/tests/boundary/process/modelFixtureShutdown.test.ts b/tests/boundary/process/modelFixtureShutdown.test.ts new file mode 100644 index 0000000000..28ab529e6d --- /dev/null +++ b/tests/boundary/process/modelFixtureShutdown.test.ts @@ -0,0 +1,494 @@ +import { channel } from "node:diagnostics_channel"; +import { existsSync } from "node:fs"; +import { readFile, writeFile } from "node:fs/promises"; +import { + IncomingMessage, + request, + Server, + ServerResponse, + type ClientRequest, +} from "node:http"; +import { setImmediate } from "node:timers/promises"; +import { expect, it } from "vitest"; +import { z } from "zod"; + +import { createOpenAiModelFixture } from "../../../scripts/verify/clients/model-fixture.mjs"; +import { + createTestWorkspace, + removeTestWorkspace, +} from "../../support/workspace/workspaceFixture.js"; + +interface ModelFixture { + readonly baseUrl: string; + readonly requests: readonly unknown[]; + readonly failure: string | undefined; + close(): Promise; +} + +type FixtureFactory = (options: { + readonly directory: string; + readonly prefix: string; + readonly onRequest: () => Promise; +}) => Promise; + +const loadFixtureFactory = async (protocol: { + readonly module: string; + readonly exportName: string; +}): Promise => { + const exports = z + .record(z.string(), z.unknown()) + .parse( + await import( + new URL( + `../../../scripts/verify/clients/${protocol.module}`, + import.meta.url, + ).href + ), + ); + return z + .custom((value) => typeof value === "function") + .parse(exports[protocol.exportName]); +}; + +const protocols = [ + { + name: "OpenAI", + module: "model-fixture.mjs", + exportName: "createOpenAiModelFixture", + path: "/v1/chat/completions", + body: { + model: "rea-client-fixture", + messages: [{ role: "user", content: "authored shutdown request" }], + }, + response: { role: "assistant", content: "authored response" }, + }, + { + name: "Anthropic", + module: "anthropic-model-fixture.mjs", + exportName: "createAnthropicModelFixture", + path: "/v1/messages", + body: { + model: "rea-client-fixture", + messages: [{ role: "user", content: "authored shutdown request" }], + }, + response: [{ type: "text", text: "authored response" }], + }, + { + name: "Gemini", + module: "gemini-model-fixture.mjs", + exportName: "createGeminiModelFixture", + path: "/v1beta/models/rea-client-fixture:generateContent", + body: { + contents: [ + { role: "user", parts: [{ text: "authored shutdown request" }] }, + ], + }, + response: [{ text: "authored response" }], + }, +] as const; + +const deferred = () => { + let resolve!: (value: Value) => void; + const promise = new Promise((finish) => { + resolve = finish; + }); + + return { promise, resolve }; +}; + +const bounded = async ( + work: Promise, + boundary: string, +): Promise => { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + work, + new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new Error(`Fixture did not reach ${boundary}`)), + 5_000, + ); + }), + ]); + } finally { + clearTimeout(timer); + } +}; + +const serverRequestSchema = z.object({ + request: z.instanceof(IncomingMessage), + response: z.instanceof(ServerResponse), + server: z.instanceof(Server), +}); + +// This scoped Node diagnostic observes real HTTP resources; the producers keep +// their existing public API, request parsers, callbacks and filesystem writes. +for (const protocol of protocols) + for (const outcome of ["success", "late-error"] as const) + it(`${protocol.name} joins a disconnected request's ${outcome} finalization before fixture close`, async () => { + const workspace = await createTestWorkspace("rea-model-fixture-close-"); + const artifactPath = workspace.path("final request evidence.bin"); + const artifact = Buffer.from("accepted request evidence α\0\n"); + const entered = deferred(); + const release = deferred(); + const handlerFinished = deferred(); + const observedRequest = deferred>(); + const serverResponseClosed = deferred(); + const serverSocketClosed = deferred(); + const clientSocketClosed = deferred(); + const lateFailure = new Error("authored late model handler failure"); + const diagnostic = channel("http.server.request.start"); + let port: number | undefined; + let callbackEntered = false; + let callbackFinished = false; + let fixture: ModelFixture | undefined; + let client: ClientRequest | undefined; + const observeRequest = (message: unknown) => { + const parsed = serverRequestSchema.safeParse(message); + if ( + !parsed.success || + parsed.data.request.socket.localPort !== port || + parsed.data.request.url !== protocol.path + ) + return; + parsed.data.response.once("close", () => + serverResponseClosed.resolve(), + ); + parsed.data.request.socket.once("close", () => + serverSocketClosed.resolve(), + ); + observedRequest.resolve(parsed.data); + }; + diagnostic.subscribe(observeRequest); + try { + const createFixture = await loadFixtureFactory(protocol); + fixture = await createFixture({ + directory: workspace.root, + prefix: "shutdown", + onRequest: async () => { + callbackEntered = true; + entered.resolve(); + try { + await release.promise; + await writeFile(artifactPath, artifact); + if (outcome === "late-error") throw lateFailure; + return protocol.response; + } finally { + callbackFinished = true; + handlerFinished.resolve(); + } + }, + }); + port = Number(new URL(fixture.baseUrl).port); + client = request(new URL(protocol.path, fixture.baseUrl), { + method: "POST", + agent: false, + headers: { "content-type": "application/json" }, + }); + // A peer disconnect is the authored stimulus, so observe its error and + // close independently rather than letting ECONNRESET escape the test. + client.on("error", () => {}); + client.once("socket", (socket) => { + socket.once("close", () => clientSocketClosed.resolve()); + }); + client.end(JSON.stringify(protocol.body)); + await bounded(entered.promise, "request callback admission"); + const physical = await bounded( + observedRequest.promise, + "server request observation", + ); + client.destroy(); + await bounded( + Promise.all([ + clientSocketClosed.promise, + serverResponseClosed.promise, + serverSocketClosed.promise, + ]), + "both ends of the disconnected HTTP request", + ); + expect(fixture.requests).toHaveLength(1); + expect(physical.response.destroyed).toBe(true); + expect(physical.request.socket.destroyed).toBe(true); + const serverClosed = deferred(); + physical.server.once("close", () => serverClosed.resolve()); + const closing = fixture.close().then( + () => ({ kind: "fixture-closed" as const }), + (cause: unknown) => ({ + kind: "fixture-close-failed" as const, + cause, + }), + ); + // Node emits server close and invokes close callbacks together. The + // following check-phase barrier lets those callbacks and their promise + // reactions settle; no elapsed duration decides whether work was joined. + const transportClosed = serverClosed.promise.then(async () => { + await setImmediate(); + return { kind: "transport-closed" as const }; + }); + const first = await bounded( + Promise.race([closing, transportClosed]), + "physical server close with the handler still admitted", + ); + expect(physical.server.listening).toBe(false); + expect(physical.server.address()).toBeNull(); + expect(callbackFinished).toBe(false); + expect(existsSync(artifactPath)).toBe(false); + expect(first).toEqual({ kind: "transport-closed" }); + release.resolve(); + const closed = await bounded(closing, "joined fixture finalization"); + if (outcome === "late-error") { + expect(closed.kind).toBe("fixture-close-failed"); + if (closed.kind !== "fixture-close-failed") + throw new Error("Fixture close discarded the late handler error"); + expect(closed.cause).toBe(lateFailure); + } else expect(closed).toEqual({ kind: "fixture-closed" }); + expect(callbackFinished).toBe(true); + expect(await readFile(artifactPath)).toEqual(artifact); + expect(fixture.failure).toBe( + outcome === "late-error" ? String(lateFailure) : undefined, + ); + } finally { + release.resolve(); + client?.destroy(); + try { + if (callbackEntered) + await bounded(handlerFinished.promise, "authored callback cleanup"); + if (fixture !== undefined) { + const cleanup = await bounded( + fixture.close().then( + () => ({ kind: "closed" as const }), + (cause: unknown) => ({ kind: "failed" as const, cause }), + ), + "HTTP fixture cleanup", + ); + if ( + cleanup.kind === "failed" && + !(outcome === "late-error" && cleanup.cause === lateFailure) + ) + throw cleanup.cause; + } + // The baseline close does not join handlers. Let the released callback + // finish its producer catch before removing the test-owned directory. + await setImmediate(); + } finally { + diagnostic.unsubscribe(observeRequest); + await removeTestWorkspace(workspace.root); + } + } + }); + +for (const protocol of protocols) + it(`${protocol.name} preserves an ordinary 400 and gracefully joins admitted work across repeated closes`, async () => { + const workspace = await createTestWorkspace("rea-model-fixture-graceful-"); + const artifactPath = workspace.path("graceful request evidence.bin"); + const artifact = Buffer.from("graceful request evidence α\0\n"); + const entered = deferred(); + const release = deferred(); + const handlerFinished = deferred(); + const observedServer = deferred(); + const diagnostic = channel("http.server.request.start"); + const controller = new AbortController(); + let callbackEntered = false; + let callbackFinished = false; + let callbacks = 0; + let closedCalls = 0; + let port: number | undefined; + let fixture: ModelFixture | undefined; + let requestFinished: Promise | undefined; + const observeRequest = (message: unknown) => { + const parsed = serverRequestSchema.safeParse(message); + if ( + parsed.success && + parsed.data.request.socket.localPort === port && + parsed.data.request.url === protocol.path + ) + observedServer.resolve(parsed.data.server); + }; + diagnostic.subscribe(observeRequest); + try { + const createFixture = await loadFixtureFactory(protocol); + fixture = await createFixture({ + directory: workspace.root, + prefix: "graceful", + onRequest: async () => { + callbacks += 1; + callbackEntered = true; + entered.resolve(); + try { + await release.promise; + await writeFile(artifactPath, artifact); + return protocol.response; + } finally { + callbackFinished = true; + handlerFinished.resolve(); + } + }, + }); + port = Number(new URL(fixture.baseUrl).port); + const endpoint = new URL(protocol.path, fixture.baseUrl); + const invalid = await bounded( + fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json" }, + body: "{}", + signal: controller.signal, + }), + "ordinary invalid request response", + ); + await bounded(invalid.text(), "ordinary invalid response consumption"); + expect(invalid.status).toBe(400); + expect(callbacks).toBe(0); + expect(fixture.requests).toHaveLength(0); + const ordinaryFailure = fixture.failure; + expect(typeof ordinaryFailure).toBe("string"); + const responding = fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(protocol.body), + signal: controller.signal, + }) + .then(async (response) => ({ + kind: "response" as const, + status: response.status, + text: await response.text(), + })) + .catch((cause: unknown) => ({ + kind: "request-failed" as const, + cause, + })); + requestFinished = responding; + await bounded(entered.promise, "graceful request callback admission"); + const server = await bounded( + observedServer.promise, + "graceful HTTP server observation", + ); + expect(callbacks).toBe(1); + expect(fixture.requests).toHaveLength(1); + expect(server).toBeInstanceOf(Server); + const closing = [fixture.close(), fixture.close()].map((close) => + close.then( + () => { + closedCalls += 1; + return { kind: "fixture-closed" as const }; + }, + (cause: unknown) => { + closedCalls += 1; + return { kind: "fixture-close-failed" as const, cause }; + }, + ), + ); + // Admission is acknowledged and its live request still needs a response. + // Give both close calls a check-phase turn while that work remains gated. + await setImmediate(); + expect(closedCalls).toBe(0); + expect(callbackFinished).toBe(false); + expect(existsSync(artifactPath)).toBe(false); + release.resolve(); + const response = await bounded(responding, "graceful model response"); + expect(response.kind).toBe("response"); + if (response.kind !== "response") throw response.cause; + expect(response.status).toBe(200); + expect(response.text).toContain("authored response"); + expect( + await bounded(Promise.all(closing), "both admitted fixture closes"), + ).toEqual([{ kind: "fixture-closed" }, { kind: "fixture-closed" }]); + expect(callbackFinished).toBe(true); + expect(await readFile(artifactPath)).toEqual(artifact); + expect(fixture.failure).toBe(ordinaryFailure); + expect(server.listening).toBe(false); + expect(server.address()).toBeNull(); + await bounded(fixture.close(), "repeated completed fixture close"); + expect(fixture.failure).toBe(ordinaryFailure); + } finally { + release.resolve(); + controller.abort(new Error("authored fixture control cleanup")); + try { + if (requestFinished !== undefined) + await bounded(requestFinished, "graceful client cleanup"); + if (callbackEntered) + await bounded(handlerFinished.promise, "graceful callback cleanup"); + if (fixture !== undefined) + await bounded(fixture.close(), "graceful HTTP fixture cleanup"); + } finally { + diagnostic.unsubscribe(observeRequest); + await removeTestWorkspace(workspace.root); + } + } + }); + +for (const opaqueFailure of [undefined, null]) + it(`OpenAI retains a late ${String(opaqueFailure)} rejection through HTTP fixture close`, async () => { + const workspace = await createTestWorkspace("rea-model-fixture-opaque-"); + const entered = deferred(); + const release = deferred(); + const handlerFinished = deferred(); + const clientClosed = deferred(); + let callbackEntered = false; + let fixture: ModelFixture | undefined; + let client: ClientRequest | undefined; + try { + fixture = await createOpenAiModelFixture({ + directory: workspace.root, + prefix: "opaque", + onRequest: async () => { + callbackEntered = true; + entered.resolve(); + try { + await release.promise; + throw opaqueFailure; + } finally { + handlerFinished.resolve(); + } + }, + }); + client = request(new URL("/v1/chat/completions", fixture.baseUrl), { + method: "POST", + agent: false, + headers: { "content-type": "application/json" }, + }); + client.on("error", () => {}); + client.once("close", () => clientClosed.resolve()); + client.end( + JSON.stringify({ + model: "rea-client-fixture", + messages: [{ role: "user", content: "authored opaque failure" }], + }), + ); + await bounded(entered.promise, "opaque request callback admission"); + client.destroy(); + await bounded(clientClosed.promise, "opaque client disconnect"); + const closing = fixture.close().then( + () => ({ ok: true as const }), + (cause: unknown) => ({ ok: false as const, cause }), + ); + release.resolve(); + const closed = await bounded(closing, "opaque handler finalization"); + expect(closed.ok).toBe(false); + if (closed.ok) + throw new Error("Fixture close discarded an opaque handler rejection"); + expect(Object.hasOwn(closed, "cause")).toBe(true); + expect(closed.cause).toBe(opaqueFailure); + expect(fixture.failure).toBe(String(opaqueFailure)); + } finally { + release.resolve(); + client?.destroy(); + try { + if (callbackEntered) + await bounded(handlerFinished.promise, "opaque callback cleanup"); + if (fixture !== undefined) { + const cleanup = await bounded( + fixture.close().then( + () => ({ ok: true as const }), + (cause: unknown) => ({ ok: false as const, cause }), + ), + "opaque HTTP fixture cleanup", + ); + if (!cleanup.ok && !Object.is(cleanup.cause, opaqueFailure)) + throw cleanup.cause; + } + } finally { + await removeTestWorkspace(workspace.root); + } + } + });