From 1f8d7b73af84bee0c88b43812648c32368135141 Mon Sep 17 00:00:00 2001 From: Yichen Jiang Date: Sat, 29 Aug 2026 11:29:46 +0800 Subject: [PATCH] fix(snapshot): own fixture listener lifecycle --- scripts/snapshot-http-fixtures.spec.ts | 193 ++++++++++++++++++ snapshots/session/headless.snapshot.ts | 74 ++++++- .../web-fetch/web-fetch-fixture-server.mjs | 100 ++++++--- .../web-search-error-fixture.mjs | 133 +++++++++--- 4 files changed, 430 insertions(+), 70 deletions(-) create mode 100644 scripts/snapshot-http-fixtures.spec.ts diff --git a/scripts/snapshot-http-fixtures.spec.ts b/scripts/snapshot-http-fixtures.spec.ts new file mode 100644 index 0000000000..d54a4a3aa4 --- /dev/null +++ b/scripts/snapshot-http-fixtures.spec.ts @@ -0,0 +1,193 @@ +import { EventEmitter } from 'node:events' +import { Context } from '@deepseek-ai/cordis' +import { WebRuntime } from '@deepseek-ai/dsh-web' +import { afterEach, describe, expect, it, vi } from 'vitest' + +const httpMock = vi.hoisted(() => ({ createServer: vi.fn() })) + +vi.mock('node:http', () => ({ createServer: httpMock.createServer })) + +// Snapshot plugins are plain runtime JavaScript loaded by cordis.yml. +// @ts-expect-error The fixture intentionally has no declaration artifact. +import * as searchFixtureModule from '../snapshots/session/web-search-endpoint-guidance/web-search-error-fixture.mjs' +// @ts-expect-error The fixture intentionally has no declaration artifact. +import * as webFetchFixtureModule from '../snapshots/session/web-fetch/web-fetch-fixture-server.mjs' + +const RECORDED_ENDPOINT = 'http://127.0.0.1:43118/anthropic/v1/messages' +const RECORDED_URL = 'http://public.test:43117/menu.html' + +interface FixturePlugin { + readonly name: string + readonly inject?: readonly string[] + apply(ctx: Context): Promise +} + +const searchFixture = searchFixtureModule as unknown as FixturePlugin +const webFetchFixture = webFetchFixtureModule as unknown as FixturePlugin +const nativeFetch = globalThis.fetch + +class FixtureServer extends EventEmitter { + readonly started = Promise.withResolvers() + listening = false + closed = false + connectionsClosed = false + unreferenced = false + private listenCallback: (() => void) | undefined + private port = 0 + + listen(_port: number, _host: string, callback: () => void): this { + this.listenCallback = callback + this.started.resolve(undefined) + return this + } + + finishListening(port = 54321): void { + this.port = port + this.listening = true + this.listenCallback?.() + } + + address(): { address: string; family: string; port: number } | null { + return this.listening ? { address: '127.0.0.1', family: 'IPv4', port: this.port } : null + } + + unref(): this { + this.unreferenced = true + return this + } + + close(callback: (error?: Error) => void): this { + this.listening = false + this.closed = true + callback() + return this + } + + closeAllConnections(): void { + this.connectionsClosed = true + } +} + +function nextServer(): FixtureServer { + const server = new FixtureServer() + httpMock.createServer.mockReturnValueOnce(server) + return server +} + +function captureErrors(ctx: Context): unknown[] { + const errors: unknown[] = [] + ctx.logger.error = ((error: unknown) => { errors.push(error) }) as typeof ctx.logger.error + return errors +} + +afterEach(() => { + globalThis.fetch = nativeFetch + httpMock.createServer.mockReset() +}) + +describe('snapshot HTTP fixture lifecycle', () => { + it('joins search listener setup and cleanup when disposal wins the startup race', async () => { + const server = nextServer() + const ctx = new Context() + const errors = captureErrors(ctx) + const fiber = ctx.plugin(searchFixture) + await server.started.promise + + const disposal = fiber.dispose() + const settled = vi.fn() + void disposal.then(settled) + await Promise.resolve() + expect(settled).not.toHaveBeenCalled() + + server.finishListening() + await disposal + + expect(server).toMatchObject({ closed: true, connectionsClosed: true, unreferenced: true }) + expect(globalThis.fetch).toBe(nativeFetch) + expect(errors).toEqual([]) + }) + + it('removes the fetch provider and closes its listener when disposal wins the startup race', async () => { + const server = nextServer() + const ctx = new Context() + await ctx.plugin(WebRuntime) + const errors = captureErrors(ctx) + const fiber = ctx.plugin(webFetchFixture) + await server.started.promise + + const disposal = fiber.dispose() + const settled = vi.fn() + void disposal.then(settled) + await Promise.resolve() + expect(settled).not.toHaveBeenCalled() + + server.finishListening() + await disposal + + expect(server).toMatchObject({ closed: true, connectionsClosed: true, unreferenced: true }) + await expect(ctx.web.fetch({ url: RECORDED_URL })) + .rejects.toThrow(expect.objectContaining({ code: 'WEB_PROVIDER_UNAVAILABLE' })) + expect(errors).toEqual([]) + }) + + it('maps every fetch input form and rejects another path on the recorded authority', async () => { + const server = nextServer() + const fetchMock = vi.fn(async (_input: string | URL | Request, _init?: RequestInit) => new Response('{}')) + globalThis.fetch = fetchMock + const ctx = new Context() + const fiber = ctx.plugin(searchFixture) + await server.started.promise + server.finishListening(54322) + await fiber + + try { + await globalThis.fetch(RECORDED_ENDPOINT) + expect(fetchMock.mock.calls.at(-1)?.[0]).toBe('http://127.0.0.1:54322/anthropic/v1/messages') + + await globalThis.fetch(new URL(RECORDED_ENDPOINT)) + expect(fetchMock.mock.calls.at(-1)?.[0]).toBe('http://127.0.0.1:54322/anthropic/v1/messages') + + const request = new Request(RECORDED_ENDPOINT, { method: 'POST', headers: { 'x-fixture': 'request' } }) + await globalThis.fetch(request) + const mappedRequest = fetchMock.mock.calls.at(-1)?.[0] + expect(mappedRequest).toBeInstanceOf(Request) + if (!(mappedRequest instanceof Request)) throw new TypeError('mapped fetch input must be a Request') + expect(mappedRequest.url).toBe('http://127.0.0.1:54322/anthropic/v1/messages') + expect(mappedRequest.method).toBe('POST') + expect(mappedRequest.headers.get('x-fixture')).toBe('request') + + const unrelated = new URL('https://example.test/') + await globalThis.fetch(unrelated) + expect(fetchMock.mock.calls.at(-1)?.[0]).toBe(unrelated) + + await expect(globalThis.fetch('http://127.0.0.1:43118/unexpected')) + .rejects.toThrow('web-search-error-fixture: unexpected URL for recorded authority') + } finally { + await fiber.dispose() + } + + expect(globalThis.fetch).toBe(fetchMock) + expect(server.closed).toBe(true) + }) + + it('preserves a later fetch wrapper while still closing the listener and reporting the ownership error', async () => { + const server = nextServer() + const fetchMock = vi.fn(async (_input: string | URL | Request, _init?: RequestInit) => new Response('{}')) + globalThis.fetch = fetchMock + const ctx = new Context() + const errors = captureErrors(ctx) + const fiber = ctx.plugin(searchFixture) + await server.started.promise + server.finishListening() + await fiber + + const fixtureFetch = globalThis.fetch + const laterFetch = vi.fn((input: string | URL | Request, init?: RequestInit) => fixtureFetch(input, init)) + globalThis.fetch = laterFetch + await fiber.dispose() + + expect(globalThis.fetch).toBe(laterFetch) + expect(server).toMatchObject({ closed: true, connectionsClosed: true }) + expect(errors.map(String).join('\n')).toContain('web-search-error-fixture: global fetch owner changed before cleanup') + }) +}) diff --git a/snapshots/session/headless.snapshot.ts b/snapshots/session/headless.snapshot.ts index 1579c5d04e..5da4cb02e9 100644 --- a/snapshots/session/headless.snapshot.ts +++ b/snapshots/session/headless.snapshot.ts @@ -7,6 +7,7 @@ import { homedir } from 'node:os' import { basename, delimiter, dirname, join } from 'node:path' import { fileURLToPath } from 'node:url' import { describe, expect, it } from 'vitest' +import ts from 'typescript' import { captureExpectedWorkspaceSnapshot, captureWorkspaceSnapshot, @@ -84,6 +85,49 @@ interface SessionLog { readonly header: JsonObject } +function propertyName(node: ts.PropertyName): string | undefined { + if (ts.isIdentifier(node) || ts.isStringLiteral(node) || ts.isNumericLiteral(node)) return node.text + return undefined +} + +function bindsOsAssignedPort(argument: ts.Expression | undefined): boolean { + if (argument === undefined) return false + if (ts.isNumericLiteral(argument)) return Number(argument.text) === 0 + if (!ts.isObjectLiteralExpression(argument)) return false + let portIsZero: boolean | undefined + for (const property of argument.properties) { + if (ts.isSpreadAssignment(property)) { + portIsZero = undefined + continue + } + if (propertyName(property.name) !== 'port') continue + portIsZero = ts.isPropertyAssignment(property) + && ts.isNumericLiteral(property.initializer) + && Number(property.initializer.text) === 0 + } + return portIsZero === true +} + +function listenerPortViolations(path: string, sourceText: string): string[] { + const source = ts.createSourceFile(path, sourceText, ts.ScriptTarget.Latest, true, ts.ScriptKind.JS) + const violations: string[] = [] + const visit = (node: ts.Node): void => { + if (ts.isCallExpression(node) + && ts.isPropertyAccessExpression(node.expression) + && node.expression.name.text === 'listen' + && !bindsOsAssignedPort(node.arguments[0])) { + const line = source.getLineAndCharacterOfPosition(node.getStart(source)).line + 1 + const received = node.arguments[0]?.getText(source) ?? '' + violations.push( + `${path}:${line}: listener port ${received} must use listen(0, ...) or listen({ port: 0, ... })`, + ) + } + ts.forEachChild(node, visit) + } + visit(source) + return violations +} + function harvested(log: SessionLog): HarvestedLog { return { id: String(log.header.id), @@ -507,17 +551,27 @@ describe('headless recorded-session snapshots', () => { } }) + it('recognizes the supported OS-assigned listener forms', () => { + expect(listenerPortViolations('accepted.mjs', [ + "server.listen(0, '127.0.0.1')", + "server.listen({ port: 0, host: '127.0.0.1' })", + 'server.listen({ ...options, port: 0 })', + ].join('\n'))).toEqual([]) + expect(listenerPortViolations('fixed.mjs', 'server.listen(43118)')).toEqual([ + 'fixed.mjs:1: listener port 43118 must use listen(0, ...) or listen({ port: 0, ... })', + ]) + expect(listenerPortViolations('dynamic.mjs', 'server.listen({ port, ...options })')).toEqual([ + 'dynamic.mjs:1: listener port { port, ...options } must use listen(0, ...) or listen({ port: 0, ... })', + ]) + }) + it('binds scenario HTTP fixtures only to OS-assigned ports', async () => { - for (const scenario of scenarios) { - const fixtureNames = (await readdir(scenario.dir)).filter(name => name.endsWith('.mjs')) - for (const fixtureName of fixtureNames) { - const source = await readFile(join(scenario.dir, fixtureName), 'utf8') - const boundPorts = [...source.matchAll(/\.listen\(\s*([^,\s)]+)/gu)].map(match => match[1]) - for (const port of boundPorts) { - expect(port, `${scenario.name}/${fixtureName}: listener port`).toBe('0') - } - } - } + const fixtureNames = (await readdir(snapshotsRoot, { recursive: true })).filter(name => name.endsWith('.mjs')) + const violations = (await Promise.all(fixtureNames.map(async (fixtureName) => listenerPortViolations( + fixtureName, + await readFile(join(snapshotsRoot, fixtureName), 'utf8'), + )))).flat() + expect(violations).toEqual([]) }) it('stores session-owned inputs with typed redaction and no ACP transcript', async () => { diff --git a/snapshots/session/web-fetch/web-fetch-fixture-server.mjs b/snapshots/session/web-fetch/web-fetch-fixture-server.mjs index f0b8db6ebb..2979917043 100644 --- a/snapshots/session/web-fetch/web-fetch-fixture-server.mjs +++ b/snapshots/session/web-fetch/web-fetch-fixture-server.mjs @@ -36,54 +36,94 @@ const LIMITS = { userAgent: 'deepseek-harness-snapshot/1.0', } +function listen(server) { + return new Promise((resolve, reject) => { + const onError = (error) => { + server.off('error', onError) + reject(error) + } + server.once('error', onError) + try { + server.listen(0, '127.0.0.1', () => { + server.off('error', onError) + resolve(undefined) + }) + } catch (error) { + server.off('error', onError) + reject(error) + } + }) +} + +async function close(server) { + if (!server.listening) return + await new Promise((resolve, reject) => { + server.close(error => error ? reject(error) : resolve(undefined)) + server.closeAllConnections() + }) +} + /** * Register the deterministic provider and start its loopback server. * @param ctx - Cordis context; the effect disposes the server with the fiber. */ export async function apply(ctx) { - const server = createServer((req, res) => { - if (req.url === '/menu.html') { - res.writeHead(200, { 'content-type': 'text/html; charset=utf-8' }) - res.end(PAGE) - return - } - res.writeHead(404, { 'content-type': 'text/plain; charset=utf-8' }) - res.end('not found') - }) - const listening = new Promise((resolve, reject) => { - server.once('error', reject) - server.listen(0, '127.0.0.1', () => resolve(undefined)) - }) - await listening - const address = server.address() - if (address === null || typeof address === 'string') { - throw new Error('web-fetch-fixture-server: loopback listener has no TCP address') - } - const transportUrl = new URL(RECORDED_URL) - transportUrl.port = String(address.port) - // The fixture must never hold the process open past protocol shutdown. - server.unref() + const readiness = Promise.withResolvers() + let transportUrl + let startupError const resolveAddresses = async (hostname) => { if (hostname !== 'public.test') throw new Error(`unexpected snapshot hostname: ${hostname}`) return [{ address: '127.0.0.1', family: 4 }] } - ctx.effect(() => async () => { - await new Promise((resolve, reject) => { - server.close(error => error ? reject(error) : resolve(undefined)) - // Stop accepting first so a connection cannot arrive after the forced close. - server.closeAllConnections() - }) - }, 'web-fetch-fixture-server') const provider = new HttpFetchProvider(LIMITS, resolveAddresses) - ctx.web.registerFetchProvider({ + const unregister = ctx.web.registerFetchProvider({ id: provider.id, available: () => provider.available(), fetch: async (request, signal) => { if (request.url !== RECORDED_URL) throw new Error(`unexpected snapshot URL: ${request.url}`) + await readiness.promise + if (startupError !== undefined) throw startupError const result = await provider.fetch({ url: transportUrl.toString() }, signal) return { ...result, url: RECORDED_URL } }, }) + await ctx.effect(async () => { + const server = createServer((req, res) => { + if (req.url === '/menu.html') { + res.writeHead(200, { 'content-type': 'text/html; charset=utf-8' }) + res.end(PAGE) + return + } + res.writeHead(404, { 'content-type': 'text/plain; charset=utf-8' }) + res.end('not found') + }) + try { + await listen(server) + const address = server.address() + if (address === null || typeof address === 'string') { + throw new Error('web-fetch-fixture-server: loopback listener has no TCP address') + } + transportUrl = new URL(RECORDED_URL) + transportUrl.port = String(address.port) + // The fixture must never hold the process open past protocol shutdown. + server.unref() + readiness.resolve(undefined) + return async () => { + unregister() + await close(server) + } + } catch (cause) { + startupError = cause + readiness.resolve(undefined) + unregister() + try { + await close(server) + } catch (cleanupError) { + throw new AggregateError([cause, cleanupError], 'web-fetch-fixture-server: setup and cleanup failed') + } + throw cause + } + }, 'web-fetch-fixture-server') } diff --git a/snapshots/session/web-search-endpoint-guidance/web-search-error-fixture.mjs b/snapshots/session/web-search-endpoint-guidance/web-search-error-fixture.mjs index 314f9b2dee..ce2087aff6 100644 --- a/snapshots/session/web-search-endpoint-guidance/web-search-error-fixture.mjs +++ b/snapshots/session/web-search-endpoint-guidance/web-search-error-fixture.mjs @@ -3,42 +3,115 @@ import { createServer } from 'node:http' /** Model-visible endpoint retained by the recorded session. */ const RECORDED_ENDPOINT = 'http://127.0.0.1:43118/anthropic/v1/messages' +const RECORDED_URL = new URL(RECORDED_ENDPOINT) /** Cordis plugin name. */ export const name = 'web-search-error-fixture' +function listen(server) { + return new Promise((resolve, reject) => { + const onError = (error) => { + server.off('error', onError) + reject(error) + } + server.once('error', onError) + try { + server.listen(0, '127.0.0.1', () => { + server.off('error', onError) + resolve(undefined) + }) + } catch (error) { + server.off('error', onError) + reject(error) + } + }) +} + +async function close(server) { + if (!server.listening) return + await new Promise((resolve, reject) => { + server.close(error => error ? reject(error) : resolve(undefined)) + server.closeAllConnections() + }) +} + +function requestUrl(input) { + if (typeof input === 'string') return input + if (input instanceof URL) return input.href + if (input instanceof Request) return input.url + return undefined +} + +function transportInput(input, transportEndpoint) { + const url = requestUrl(input) + if (url === RECORDED_ENDPOINT) { + return input instanceof Request ? new Request(transportEndpoint, input) : transportEndpoint + } + if (url === undefined) return input + let parsed + try { + parsed = new URL(url) + } catch { + return input + } + if (parsed.host === RECORDED_URL.host) { + throw new Error(`web-search-error-fixture: unexpected URL for recorded authority: ${url}`) + } + return input +} + +async function cleanup(server, restoreFetch) { + const errors = [] + try { + restoreFetch() + } catch (error) { + errors.push(error) + } + try { + await close(server) + } catch (error) { + errors.push(error) + } + if (errors.length === 1) throw errors[0] + if (errors.length > 1) throw new AggregateError(errors, 'web-search-error-fixture: cleanup failed') +} + /** Start the local Messages endpoint and stop it with the plugin fiber. */ export async function apply(ctx) { - const server = createServer((request, response) => { - if (request.method === 'POST' && request.url === '/anthropic/v1/messages') { - response.writeHead(401, { 'content-type': 'application/json' }) - response.end(JSON.stringify({ error: { message: 'invalid snapshot API key' } })) - return - } - response.writeHead(404, { 'content-type': 'text/plain; charset=utf-8' }) - response.end('not found') - }) - await new Promise((resolve, reject) => { - server.once('error', reject) - server.listen(0, '127.0.0.1', () => resolve(undefined)) - }) - const address = server.address() - if (address === null || typeof address === 'string') { - throw new Error('web-search-error-fixture: loopback listener has no TCP address') - } - const transportEndpoint = `http://127.0.0.1:${String(address.port)}/anthropic/v1/messages` - const originalFetch = globalThis.fetch - const fixtureFetch = (input, init) => originalFetch( - typeof input === 'string' && input === RECORDED_ENDPOINT ? transportEndpoint : input, - init, - ) - globalThis.fetch = fixtureFetch - server.unref() - ctx.effect(() => async () => { - globalThis.fetch = originalFetch - await new Promise((resolve, reject) => { - server.close(error => error ? reject(error) : resolve(undefined)) - server.closeAllConnections() + await ctx.effect(async () => { + const server = createServer((request, response) => { + if (request.method === 'POST' && request.url === '/anthropic/v1/messages') { + response.writeHead(401, { 'content-type': 'application/json' }) + response.end(JSON.stringify({ error: { message: 'invalid snapshot API key' } })) + return + } + response.writeHead(404, { 'content-type': 'text/plain; charset=utf-8' }) + response.end('not found') }) + try { + await listen(server) + const address = server.address() + if (address === null || typeof address === 'string') { + throw new Error('web-search-error-fixture: loopback listener has no TCP address') + } + const transportEndpoint = `http://127.0.0.1:${String(address.port)}/anthropic/v1/messages` + server.unref() + const originalFetch = globalThis.fetch + const fixtureFetch = async (input, init) => originalFetch(transportInput(input, transportEndpoint), init) + globalThis.fetch = fixtureFetch + return () => cleanup(server, () => { + if (globalThis.fetch !== fixtureFetch) { + throw new Error('web-search-error-fixture: global fetch owner changed before cleanup') + } + globalThis.fetch = originalFetch + }) + } catch (cause) { + try { + await close(server) + } catch (cleanupError) { + throw new AggregateError([cause, cleanupError], 'web-search-error-fixture: setup and cleanup failed') + } + throw cause + } }, 'web-search-error-fixture') }