fix(snapshot): own fixture listener lifecycle

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<void>
+}
+
+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<undefined>()
+  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')
+  })
+})

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) ?? '<missing>'
+      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 () => {

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')
 }

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')
 }