fix(cli): recover unresponsive service restarts

packages/client/src/effect/service.ts

@@ -199,6 +199,22 @@ export const stop = Effect.fn("service.stop")(function* (options: Options = {},
   if (existing !== undefined) yield* kill(existing, options, metadata.targetVersion)
 })
 
+// Explicit recovery path: unlike stop(), restart may terminate an unchanged
+// registered process that no longer answers authenticated health checks.
+export const restart = Effect.fn("service.restart")(function* (options: StartOptions = {}) {
+  const existing = yield* registered(options.file, true)
+  if (existing.service !== undefined) {
+    const result = yield* kill(existing.service, options, options.version)
+    if (result === "rejected") return yield* Effect.fail(new Error("Background service rejected restart"))
+    if (result === "changed") {
+      const current = yield* read(options.file)
+      if (current !== undefined && same(current, existing.info))
+        yield* terminate(existing.info, read(options.file), true)
+    }
+  } else if (existing.info !== undefined) yield* terminate(existing.info, read(options.file), true)
+  return yield* start(options)
+})
+
 function fallback() {
   const state = process.env["XDG_STATE_HOME"] ?? join(homedir(), ".local", "state")
   return join(state, "opencode", "service.json")
@@ -217,6 +233,7 @@ export const Info = Schema.Struct({
   password: Schema.optional(Schema.String),
 })
 export type Info = typeof Info.Type
+const same = Schema.toEquivalence(Info)
 
 const decode = Schema.decodeUnknownEffect(Schema.fromJsonString(Info))
 const decodeHealth = Schema.decodeUnknownOption(ServiceStatus.Health)
@@ -305,27 +322,34 @@ const stopped = Effect.fnUntraced(function* (pid: number) {
   return yield* Effect.fail(new Error(`Server process ${pid} is still running`))
 })
 
-function same(left: Info, right: Info) {
-  return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid
-}
+const terminate = Effect.fnUntraced(function* (
+  info: Info,
+  current: Effect.Effect<Info | undefined, never, FileSystem.FileSystem>,
+  graceful: boolean,
+) {
+  if (graceful) {
+    const owner = yield* current
+    if (owner === undefined || !same(owner, info)) return "changed" as const
+    yield* signal(info.pid, "SIGTERM")
+  }
+  const done = yield* stopped(info.pid).pipe(Effect.retry(poll), Effect.option)
+  if (Option.isSome(done)) return "stopped" as const
+
+  const latest = yield* current
+  if (latest === undefined || !same(latest, info)) return "changed" as const
+  yield* signal(info.pid, "SIGKILL")
+  yield* stopped(info.pid).pipe(Effect.retry(poll))
+  return "stopped" as const
+})
 
 const kill = Effect.fnUntraced(function* (service: LocalService, options: Options, targetVersion?: string) {
   const requested = yield* requestStop(service, targetVersion)
-  if (requested === "rejected") return
-  if (requested === "unsupported") {
-    // A stale registration may point at a reused PID. Authenticate again
-    // immediately before the legacy signal fallback.
-    const current = yield* find(options)
-    if (current === undefined || !same(current.info, service.info)) return
-    yield* signal(service.info.pid, "SIGTERM")
-  }
-  const done = yield* stopped(service.info.pid).pipe(Effect.retry(poll), Effect.option)
-  if (Option.isSome(done)) return
-
-  const latest = yield* find(options)
-  if (latest === undefined || !same(latest.info, service.info)) return
-  yield* signal(service.info.pid, "SIGKILL")
-  yield* stopped(service.info.pid).pipe(Effect.retry(poll))
+  if (requested === "rejected") return "rejected" as const
+  return yield* terminate(
+    service.info,
+    registered(options.file, true).pipe(Effect.map((result) => result.service?.info)),
+    requested === "unsupported",
+  )
 })
 
 const decodeStopResponse = Schema.decodeUnknownOption(ServiceStatus.StopResponse)

packages/client/test/fixture/service.ts

@@ -1,4 +1,4 @@
-import { appendFile, rename, writeFile } from "node:fs/promises"
+import { appendFile, rename, rm, writeFile } from "node:fs/promises"
 
 const [registration, mode, delay] = process.argv.slice(2)
 if (registration === undefined || mode === undefined) throw new Error("Missing service fixture arguments")
@@ -8,6 +8,35 @@ if (mode === "record-start") {
   process.exit(1)
 }
 if (mode === "signal") process.kill(process.pid, process.platform === "win32" ? "SIGTERM" : "SIGKILL")
+if (mode === "unresponsive" || mode === "unresponsive-stubborn" || mode === "unresponsive-slow") {
+  const stalled =
+    mode === "unresponsive-slow"
+      ? Bun.serve({
+          port: 0,
+          fetch: () => {
+            void writeFile(registration + ".health-request", "")
+            return new Promise<Response>(() => {})
+          },
+        })
+      : undefined
+  await writeFile(
+    registration,
+    JSON.stringify({
+      id: crypto.randomUUID(),
+      version: "test",
+      url: stalled?.url.toString() ?? "http://127.0.0.1:1",
+      pid: process.pid,
+      password: "secret",
+    }),
+    { mode: 0o600 },
+  )
+  process.on("SIGTERM", async () => {
+    if (mode === "unresponsive-stubborn") return
+    await rm(registration, { force: true })
+    process.exit()
+  })
+  await new Promise(() => {})
+}
 
 if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated") {
   await appendFile(registration + ".starts", process.pid + "\n")
@@ -22,6 +51,7 @@ if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated") {
 }
 
 let requests = 0
+let stopDropped = false
 const version = mode === "old" || mode === "reject-stop" ? "old" : "test"
 const id = crypto.randomUUID()
 const server = Bun.serve({
@@ -32,6 +62,10 @@ const server = Bun.serve({
       await writeFile(registration + ".stop-attempt", "")
       return Response.json({ accepted: false })
     }
+    if (pathname === "/api/service/stop" && mode === "drop-stop") {
+      stopDropped = true
+      return new Promise<Response>(() => {})
+    }
     if (pathname === "/api/service/stop" && mode === "graceful") {
       const body = await request.json()
       if (typeof body !== "object" || body === null || body.instanceID !== id) return Response.json({ accepted: false })
@@ -40,6 +74,7 @@ const server = Bun.serve({
       return Response.json({ accepted: true })
     }
     if (pathname !== "/api/health") return new Response(null, { status: 404 })
+    if (stopDropped) return new Promise<Response>(() => {})
     requests += 1
     if (mode === "modern" && requests === 1) {
       await writeFile(registration + ".first-request", "")
@@ -63,7 +98,7 @@ const server = Bun.serve({
         },
         { status: 503 },
       )
-    if (mode === "starting" || mode === "graceful" || mode === "reject-stop")
+    if (mode === "starting" || mode === "graceful" || mode === "reject-stop" || mode === "drop-stop")
       return Response.json({ healthy: true, version, pid: process.pid, instanceID: id, status: { type: "ready" } })
     return Response.json({ healthy: true, version, pid: process.pid })
   },
@@ -81,8 +116,9 @@ await writeFile(
 )
 await rename(registration + ".tmp", registration)
 
-function shutdown() {
+async function shutdown() {
   server.stop(true)
+  await rm(registration, { force: true })
   process.exit()
 }
 process.on("SIGTERM", shutdown)

packages/client/test/service.test.ts

@@ -11,8 +11,14 @@ const processes: Bun.Subprocess[] = []
 const directories: string[] = []
 
 afterEach(async () => {
-  processes.forEach((process) => process.kill("SIGTERM"))
-  await Promise.all(processes.splice(0).map((process) => process.exited))
+  await Promise.all(
+    processes.splice(0).map(async (process) => {
+      process.kill("SIGTERM")
+      const exited = await Promise.race([process.exited.then(() => true), Bun.sleep(1_000).then(() => false)])
+      if (!exited) process.kill("SIGKILL")
+      await process.exited
+    }),
+  )
   await Promise.all(directories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })))
 })
 
@@ -89,6 +95,124 @@ test("requests graceful replacement of the exact service instance", async () =>
   expect(await Bun.file(registration + ".stop").json()).toEqual({ instanceID: info.id, targetVersion: "next" })
 })
 
+test("explicit restart replaces an unresponsive registered process", async () => {
+  const directory = await temp()
+  const registration = join(directory, "service.json")
+  const existing = spawn(registration, "unresponsive")
+  await waitForFile(registration)
+
+  const endpoint = await run(
+    Service.restart({
+      file: registration,
+      version: "test",
+      command: [process.execPath, fixture, registration, "ready"],
+    }),
+  )
+  await existing.exited
+  const info = await Bun.file(registration).json()
+
+  try {
+    expect(endpoint.url).toBe(info.url)
+    expect(info.pid).not.toBe(existing.pid)
+    expect(await health(endpoint.url)).toMatchObject({ healthy: true, version: "test", pid: info.pid })
+  } finally {
+    await run(Service.stop({ file: registration }))
+  }
+}, 15_000)
+
+test("restart recovers when a healthy owner stops responding during shutdown", async () => {
+  const directory = await temp()
+  const registration = join(directory, "service.json")
+  const existing = spawn(registration, "drop-stop")
+  await waitForFile(registration)
+
+  const endpoint = await run(
+    Service.restart({
+      file: registration,
+      version: "test",
+      command: [process.execPath, fixture, registration, "ready"],
+    }),
+  )
+  await existing.exited
+  const info = await Bun.file(registration).json()
+
+  try {
+    expect(endpoint.url).toBe(info.url)
+    expect(info.pid).not.toBe(existing.pid)
+  } finally {
+    await run(Service.stop({ file: registration }))
+  }
+}, 15_000)
+
+test("restart fails when a responsive owner rejects shutdown", async () => {
+  const directory = await temp()
+  const registration = join(directory, "service.json")
+  const existing = spawn(registration, "reject-stop")
+  await waitForFile(registration)
+
+  await expect(run(Service.restart({ file: registration, version: "test" }))).rejects.toThrow(
+    "Background service rejected restart",
+  )
+  expect(existing.exitCode).toBe(null)
+})
+
+test("ordinary stop never signals an unresponsive registered process", async () => {
+  const directory = await temp()
+  const registration = join(directory, "service.json")
+  const existing = spawn(registration, "unresponsive")
+  await waitForFile(registration)
+
+  await run(Service.stop({ file: registration }))
+
+  expect(existing.exitCode).toBe(null)
+})
+
+test.skipIf(process.platform === "win32")(
+  "restart escalates when an unresponsive process ignores SIGTERM",
+  async () => {
+    const directory = await temp()
+    const registration = join(directory, "service.json")
+    const existing = spawn(registration, "unresponsive-stubborn")
+    await waitForFile(registration)
+
+    const endpoint = await run(
+      Service.restart({
+        file: registration,
+        version: "test",
+        command: [process.execPath, fixture, registration, "ready"],
+      }),
+    )
+    await existing.exited
+    const info = await Bun.file(registration).json()
+
+    try {
+      expect(endpoint.url).toBe(info.url)
+      expect(existing.signalCode).toBe("SIGKILL")
+    } finally {
+      await run(Service.stop({ file: registration }))
+    }
+  },
+  20_000,
+)
+
+test("restart does not signal a process after registration changes", async () => {
+  const directory = await temp()
+  const registration = join(directory, "service.json")
+  const existing = spawn(registration, "unresponsive-slow")
+  await waitForFile(registration)
+  const restarting = run(
+    Service.restart({ file: registration, version: "test", command: [] }),
+  )
+
+  await waitForFile(registration + ".health-request")
+  const replacement = spawn(registration, "ready")
+  await waitForRegistration(registration, replacement.pid)
+  const endpoint = await restarting
+
+  expect(existing.exitCode).toBe(null)
+  expect(endpoint.url).toBe((await Bun.file(registration).json()).url)
+})
+
 test("does not spawn contenders while an incompatible service rejects replacement", async () => {
   const directory = await temp()
   const registration = join(directory, "service.json")
@@ -241,6 +365,17 @@ async function waitForFile(file: string) {
   throw new Error(`Timed out waiting for ${file}`)
 }
 
+async function waitForRegistration(file: string, pid: number) {
+  for (let attempt = 0; attempt < 600; attempt++) {
+    const info = await Bun.file(file)
+      .json()
+      .catch(() => undefined)
+    if (info?.pid === pid) return
+    await Bun.sleep(5)
+  }
+  throw new Error(`Timed out waiting for registration from ${pid}`)
+}
+
 async function health(url: string) {
   return fetch(new URL("/api/health", url), { signal: AbortSignal.timeout(1_000) }).then((response) => response.json())
 }