fix(snapshot): keep fixture lifecycle tests source-clean

scripts/snapshot-http-fixtures.spec.ts

@@ -1,6 +1,5 @@
 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() }))
@@ -11,10 +10,9 @@ vi.mock('node:http', () => ({ createServer: httpMock.createServer }))
 // @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'
+import * as loopbackFixtureModule from '../snapshots/session/loopback-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
@@ -22,8 +20,18 @@ interface FixturePlugin {
   apply(ctx: Context): Promise<void>
 }
 
+interface LoopbackFixtureOptions {
+  readonly label: string
+  readonly onCleanup: () => void
+  readonly onListening: (address: { port: number }) => void
+  readonly requestListener: () => void
+}
+
 const searchFixture = searchFixtureModule as unknown as FixturePlugin
-const webFetchFixture = webFetchFixtureModule as unknown as FixturePlugin
+const typedLoopbackFixtureModule = loopbackFixtureModule as unknown as {
+  readonly applyLoopbackServerEffect: (ctx: Context, options: LoopbackFixtureOptions) => Promise<void>
+}
+const { applyLoopbackServerEffect } = typedLoopbackFixtureModule
 const nativeFetch = globalThis.fetch
 
 class FixtureServer extends EventEmitter {
@@ -80,6 +88,17 @@ function captureErrors(ctx: Context): unknown[] {
   return errors
 }
 
+async function disposeWhileStarting(fiber: { dispose(): Promise<unknown> }, server: FixtureServer): Promise<void> {
+  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
+}
+
 afterEach(() => {
   globalThis.fetch = nativeFetch
   httpMock.createServer.mockReset()
@@ -91,42 +110,33 @@ describe('snapshot HTTP fixture lifecycle', () => {
     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
+    await disposeWhileStarting(fiber, server)
 
     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 () => {
+  it('runs owner cleanup and closes the 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
+    const onCleanup = vi.fn()
+    const onListening = vi.fn()
+    const fiber = ctx.plugin({
+      name: 'loopback-fixture-lifecycle-test',
+      apply: testCtx => applyLoopbackServerEffect(testCtx, {
+        label: 'loopback-fixture-lifecycle-test',
+        onCleanup,
+        onListening,
+        requestListener: () => {},
+      }),
+    })
+    await disposeWhileStarting(fiber, server)
 
     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(onListening).toHaveBeenCalledWith(expect.objectContaining({ port: 54321 }))
+    expect(onCleanup).toHaveBeenCalledOnce()
     expect(errors).toEqual([])
   })
 

snapshots/session/loopback-fixture-server.mjs

@@ -0,0 +1,75 @@
+/** Shared lifecycle for snapshot HTTP fixtures that bind an ephemeral loopback port. */
+import { createServer } from 'node:http'
+
+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()
+  })
+}
+
+async function cleanup(server, onCleanup, label) {
+  const errors = []
+  try {
+    onCleanup()
+  } 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, `${label}: cleanup failed`)
+}
+
+/**
+ * Start a loopback server as a Cordis effect and join cleanup with its setup.
+ * @param ctx - Cordis context that owns the listener effect.
+ * @param options - Fixture callbacks and the effect label used in diagnostics.
+ */
+export async function applyLoopbackServerEffect(ctx, options) {
+  const { label, onCleanup, onListening, requestListener } = options
+  await ctx.effect(async () => {
+    const server = createServer(requestListener)
+    try {
+      await listen(server)
+      const address = server.address()
+      if (address === null || typeof address === 'string') {
+        throw new Error(`${label}: loopback listener has no TCP address`)
+      }
+      onListening(address)
+      // Snapshot fixtures must never hold the process open past protocol shutdown.
+      server.unref()
+      return () => cleanup(server, onCleanup, label)
+    } catch (cause) {
+      try {
+        await cleanup(server, onCleanup, label)
+      } catch (cleanupError) {
+        throw new AggregateError([cause, cleanupError], `${label}: setup and cleanup failed`)
+      }
+      throw cause
+    }
+  }, label)
+}

snapshots/session/web-fetch/web-fetch-fixture-server.mjs

@@ -5,8 +5,8 @@
  * and replay therefore exercise fetch and markdown rendering without external
  * network while retaining the recorded request URL.
  */
-import { createServer } from 'node:http'
 import { HttpFetchProvider } from '@deepseek-ai/dsh-web-fetch-http'
+import { applyLoopbackServerEffect } from '../loopback-fixture-server.mjs'
 
 /** Model-visible URL retained by the recorded session. */
 const RECORDED_URL = 'http://public.test:43117/menu.html'
@@ -36,33 +36,6 @@ 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.
@@ -89,41 +62,30 @@ export async function apply(ctx) {
       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 () => {
+  try {
+    await applyLoopbackServerEffect(ctx, {
+      label: 'web-fetch-fixture-server',
+      requestListener: (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')
+      },
+      onListening: (address) => {
+        transportUrl = new URL(RECORDED_URL)
+        transportUrl.port = String(address.port)
+        readiness.resolve(undefined)
+      },
+      onCleanup: () => {
         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')
+      },
+    })
+  } catch (cause) {
+    startupError = cause
+    readiness.resolve(undefined)
+    throw cause
+  }
 }

snapshots/session/web-search-endpoint-guidance/web-search-error-fixture.mjs

@@ -1,5 +1,5 @@
 /** Deterministic authentication failure for the search endpoint guidance snapshot. */
-import { createServer } from 'node:http'
+import { applyLoopbackServerEffect } from '../loopback-fixture-server.mjs'
 
 /** Model-visible endpoint retained by the recorded session. */
 const RECORDED_ENDPOINT = 'http://127.0.0.1:43118/anthropic/v1/messages'
@@ -8,33 +8,6 @@ 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
@@ -60,26 +33,12 @@ function transportInput(input, transportEndpoint) {
   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) {
-  await ctx.effect(async () => {
-    const server = createServer((request, response) => {
+  let restoreFetch = () => {}
+  await applyLoopbackServerEffect(ctx, {
+    label: 'web-search-error-fixture',
+    requestListener: (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' } }))
@@ -87,31 +46,19 @@ export async function apply(ctx) {
       }
       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')
-      }
+    },
+    onListening: (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, () => {
+      restoreFetch = () => {
         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')
+    },
+    onCleanup: () => restoreFetch(),
+  })
 }