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); + } + } + });