fix(client): evict unresponsive service (#40896)

packages/client/src/effect/service.ts

@@ -53,6 +53,7 @@ const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) {
 /** Ensure a healthy, compatible local service is running. */
 export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOptions = {}) {
   const contenders = new Set<Contender>()
+  let timeouts: { readonly info: Info; readonly count: number } | undefined
   let announced = false
   let lastSpawn = 0
   let spawnDelay = 5_000
@@ -82,6 +83,18 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
     const registration = yield* registered(options.file, true)
     const info = registration.info
     const service = registration.service
+    if (registration.timedOut && info !== undefined) {
+      timeouts = {
+        info,
+        count: timeouts !== undefined && same(timeouts.info, info) ? timeouts.count + 1 : 1,
+      }
+      if (timeouts.count >= 3) {
+        yield* announce("missing")
+        yield* evict(info, options)
+        timeouts = undefined
+        lastSpawn = Date.now() - spawnDelay
+      }
+    } else timeouts = undefined
     if (service !== undefined) {
       spawnDelay = 5_000
       const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
@@ -182,6 +195,10 @@ type LocalService = {
 }
 
 const probe = Effect.fnUntraced(function* (info: Info, allowLegacy = false) {
+  return (yield* probeResult(info, allowLegacy)).service
+})
+
+const probeResult = Effect.fnUntraced(function* (info: Info, allowLegacy = false) {
   const endpoint = {
     url: info.url,
     auth:
@@ -189,39 +206,53 @@ const probe = Effect.fnUntraced(function* (info: Info, allowLegacy = false) {
         ? undefined
         : { type: "basic" as const, username: "opencode", password: info.password },
   } satisfies Endpoint
-  const response = yield* Effect.tryPromise(() =>
+  const signal = AbortSignal.timeout(2_000)
+  const result = yield* Effect.promise(() =>
     fetch(new URL("/api/health", info.url), {
       headers: headers(endpoint),
-      signal: AbortSignal.timeout(2_000),
-    }),
-  ).pipe(Effect.option, Effect.map(Option.getOrUndefined))
-  if (response === undefined) return undefined
-  const body = yield* Effect.tryPromise(() => response.json()).pipe(Effect.option, Effect.map(Option.getOrUndefined))
+      signal,
+    })
+      .then(async (response) => ({ response, body: (await response.json()) as unknown }))
+      .then(
+        (value) => ({ value }),
+        (cause: unknown) => ({ cause }),
+      ),
+  )
+  if ("cause" in result) return { service: undefined, timedOut: signal.aborted }
+  const response = result.value.response
+  const body = result.value.body
   const health = decodeHealth(body)
   if (Option.isSome(health)) {
-    if (health.value.pid !== info.pid) return undefined
-    if (info.version !== undefined && health.value.version !== info.version) return undefined
+    if (health.value.pid !== info.pid) return { service: undefined, timedOut: false }
+    if (info.version !== undefined && health.value.version !== info.version)
+      return { service: undefined, timedOut: false }
     return {
-      info,
-      endpoint,
-      version: health.value.version,
-      state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting",
-      legacy: false,
-    } satisfies LocalService
+      service: {
+        info,
+        endpoint,
+        version: health.value.version,
+        state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting",
+        legacy: false,
+      } satisfies LocalService,
+      timedOut: false,
+    }
   }
   if (
     !allowLegacy ||
     Option.isNone(decodeLegacyHealth(body)) ||
     (typeof body === "object" && body !== null && ("version" in body || "pid" in body))
   )
-    return undefined
-  return { info, endpoint, state: "ready", legacy: true } satisfies LocalService
+    return { service: undefined, timedOut: false }
+  return {
+    service: { info, endpoint, state: "ready", legacy: true } satisfies LocalService,
+    timedOut: false,
+  }
 })
 
 const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = false) {
   const info = yield* read(file)
-  if (info === undefined) return { info: undefined, service: undefined }
-  return { info, service: yield* probe(info, allowLegacy) }
+  if (info === undefined) return { info: undefined, service: undefined, timedOut: false }
+  return { info, ...(yield* probeResult(info, allowLegacy)) }
 })
 
 // Health-checked lookup without the version gate: lifecycle operations must be
@@ -249,6 +280,19 @@ function same(left: Info, right: Info) {
   return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid
 }
 
+const evict = Effect.fnUntraced(function* (info: Info, options: { readonly file?: string }) {
+  const current = yield* read(options.file)
+  if (current === undefined || !same(current, info)) return
+  yield* signal(info.pid, "SIGTERM")
+  const done = yield* stopped(info.pid).pipe(Effect.retry(poll), Effect.option)
+  if (Option.isSome(done)) return
+
+  const latest = yield* read(options.file)
+  if (latest === undefined || !same(latest, info)) return
+  yield* signal(info.pid, "SIGKILL")
+  yield* stopped(info.pid).pipe(Effect.retry(poll))
+})
+
 const kill = Effect.fnUntraced(function* (service: LocalService, options: { readonly file?: string }) {
   const requested = yield* requestStop(service)
   if (requested === "rejected") return

packages/client/src/promise/service.ts

@@ -34,6 +34,7 @@ async function discoverLocal(options: DiscoverOptions) {
 export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
   const deadline = Date.now() + 120_000
   const contenders = new Set<Contender>()
+  let timeouts: { readonly info: Info; readonly count: number } | undefined
   let announced = false
   let lastSpawn = 0
   let spawnDelay = 5_000
@@ -62,6 +63,19 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
   while (true) {
     if (Date.now() >= deadline) throw new Error("Timed out waiting for the background service to start")
     const registration = await registered(options.file, true)
+    if (registration.timedOut && registration.info !== undefined) {
+      timeouts = {
+        info: registration.info,
+        count:
+          timeouts !== undefined && same(timeouts.info, registration.info) ? timeouts.count + 1 : 1,
+      }
+      if (timeouts.count >= 3) {
+        announce("missing")
+        await evict(registration.info, options)
+        timeouts = undefined
+        lastSpawn = Date.now() - spawnDelay
+      }
+    } else timeouts = undefined
 
     if (registration.service !== undefined) {
       spawnDelay = 5_000
@@ -145,6 +159,10 @@ type LocalService = {
 }
 
 async function probe(info: Info, allowLegacy = false): Promise<LocalService | undefined> {
+  return (await probeResult(info, allowLegacy)).service
+}
+
+async function probeResult(info: Info, allowLegacy = false) {
   const endpoint = {
     url: info.url,
     auth:
@@ -152,30 +170,48 @@ async function probe(info: Info, allowLegacy = false): Promise<LocalService | un
         ? undefined
         : { type: "basic" as const, username: "opencode", password: info.password },
   } satisfies Endpoint
-  const response = await fetch(new URL("/api/health", info.url), {
+  const signal = AbortSignal.timeout(2_000)
+  const result = await fetch(new URL("/api/health", info.url), {
     headers: headers(endpoint),
-    signal: AbortSignal.timeout(2_000),
-  }).catch(() => undefined)
-  const body = (await response?.json().catch(() => undefined)) as ServiceHealth | { readonly healthy: true } | undefined
+    signal,
+  })
+    .then(async (response) => ({
+      response,
+      body: (await response.json()) as ServiceHealth | { readonly healthy: true },
+    }))
+    .then(
+      (value) => ({ value }),
+      (cause: unknown) => ({ cause }),
+    )
+  if ("cause" in result) return { service: undefined, timedOut: signal.aborted }
+  const response = result.value.response
+  const body = result.value.body
   if (body !== undefined && "version" in body && "pid" in body) {
-    if (body.pid !== info.pid) return undefined
-    if (info.version !== undefined && body.version !== info.version) return undefined
+    if (body.pid !== info.pid) return { service: undefined, timedOut: false }
+    if (info.version !== undefined && body.version !== info.version)
+      return { service: undefined, timedOut: false }
     return {
-      info,
-      endpoint,
-      version: body.version,
-      state: response?.ok ? "ready" : response?.status === 500 ? "failed" : "waiting",
-      legacy: false,
+      service: {
+        info,
+        endpoint,
+        version: body.version,
+        state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting",
+        legacy: false,
+      } satisfies LocalService,
+      timedOut: false,
     }
   }
-  if (!allowLegacy || body?.healthy !== true) return undefined
-  return { info, endpoint, state: "ready", legacy: true }
+  if (!allowLegacy || body?.healthy !== true) return { service: undefined, timedOut: false }
+  return {
+    service: { info, endpoint, state: "ready", legacy: true } satisfies LocalService,
+    timedOut: false,
+  }
 }
 
 async function registered(file?: string, allowLegacy = false) {
   const info = await read(file)
-  if (info === undefined) return { info: undefined, service: undefined }
-  return { info, service: await probe(info, allowLegacy) }
+  if (info === undefined) return { info: undefined, service: undefined, timedOut: false }
+  return { info, ...(await probeResult(info, allowLegacy)) }
 }
 
 async function find(options: { readonly file?: string }) {
@@ -209,6 +245,18 @@ function same(left: Info, right: Info) {
   return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid
 }
 
+async function evict(info: Info, options: { readonly file?: string }) {
+  const current = await read(options.file)
+  if (current === undefined || !same(current, info)) return
+  signal(info.pid, "SIGTERM")
+  if (await waitUntilStopped(info.pid)) return
+
+  const latest = await read(options.file)
+  if (latest === undefined || !same(latest, info)) return
+  signal(info.pid, "SIGKILL")
+  if (!(await waitUntilStopped(info.pid))) throw new Error(`Server process ${info.pid} is still running`)
+}
+
 async function kill(service: LocalService, options: { readonly file?: string }) {
   const requested = await requestStop(service)
   if (requested === "rejected") return

packages/client/test/fixture/service.ts

@@ -47,6 +47,10 @@ const server = Bun.serve({
     }
     if (pathname !== "/api/health") return new Response(null, { status: 404 })
     requests += 1
+    if (mode === "hanging") {
+      await appendFile(registration + ".requests", process.pid + "\n")
+      return new Promise<Response>(() => {})
+    }
     if (mode === "modern" && requests === 1) {
       await writeFile(registration + ".first-request", "")
       while (!(await Bun.file(registration + ".release").exists())) await Bun.sleep(5)

packages/client/test/promise-service.test.ts

@@ -70,6 +70,32 @@ test("reports a failed registered service", async () => {
   )
 })
 
+test("evicts an unresponsive registered service before starting its replacement", async () => {
+  const directory = await temp()
+  const registration = join(directory, "service.json")
+  const existing = Bun.spawn([process.execPath, fixture, registration, "hanging"], {
+    stdout: "ignore",
+    stderr: "inherit",
+  })
+  processes.push(existing)
+  await waitForFile(registration)
+  const original = await Bun.file(registration).json()
+
+  const endpoint = await Service.ensure({
+    file: registration,
+    version: "test",
+    command: [process.execPath, fixture, registration, "delayed", "10"],
+  })
+  const replacement = await Bun.file(registration).json()
+
+  expect((await Bun.file(registration + ".requests").text()).trim().split("\n")).toHaveLength(3)
+  expect(await existing.exited).toBe(0)
+  expect(replacement.pid).not.toBe(original.pid)
+  expect(endpoint.url).toBe(replacement.url)
+  process.kill(replacement.pid, "SIGTERM")
+  await waitForExit(replacement.pid)
+}, 20_000)
+
 test("requests graceful stop of the exact service instance", async () => {
   const registration = await setup("graceful")
   const info = await Bun.file(registration).json()

packages/client/test/service.test.ts

@@ -70,6 +70,30 @@ test("reports a failed registered service without spawning", async () => {
   expect(process.exitCode).toBe(null)
 })
 
+test("evicts an unresponsive registered service before starting its replacement", async () => {
+  const directory = await temp()
+  const registration = join(directory, "service.json")
+  const existing = spawn(registration, "hanging")
+  await waitForFile(registration)
+  const original = await Bun.file(registration).json()
+
+  const endpoint = await run(
+    Service.ensure({
+      file: registration,
+      version: "test",
+      command: [process.execPath, fixture, registration, "delayed", "10"],
+    }),
+  )
+  const replacement = await Bun.file(registration).json()
+
+  expect((await Bun.file(registration + ".requests").text()).trim().split("\n")).toHaveLength(3)
+  expect(await existing.exited).toBe(0)
+  expect(replacement.pid).not.toBe(original.pid)
+  expect(endpoint.url).toBe(replacement.url)
+  expect(await health(endpoint.url)).toEqual({ healthy: true, version: "test", pid: replacement.pid })
+  process.kill(replacement.pid, "SIGTERM")
+}, 20_000)
+
 test("requests graceful stop of the exact service instance", async () => {
   const directory = await temp()
   const registration = join(directory, "service.json")