fix(client): preserve slow-starting managed services

packages/cli/src/services/service-registration.ts

@@ -26,6 +26,7 @@ export const register = Effect.fnUntraced(function* (options: {
     version: OPENCODE_VERSION,
     url: HttpServer.formatAddress(options.address),
     pid: process.pid,
+    startedAt: Date.now(),
     password: options.password,
   }
   const encoded = yield* encodeInfo(info)
@@ -36,14 +37,12 @@ export const register = Effect.fnUntraced(function* (options: {
     found.url === info.url &&
     found.pid === info.pid &&
     found.password === info.password
-  yield* fs
-    .writeFileString(temp, encoded, { mode: 0o600 })
-    .pipe(
-      // Windows rename can fail with EPERM while a contender or test polls the registration file.
-      Effect.andThen(
-        fs.rename(temp, options.file).pipe(Effect.retry({ times: 20, schedule: Schedule.spaced("10 millis") })),
-      ),
-    )
+  yield* fs.writeFileString(temp, encoded, { mode: 0o600 }).pipe(
+    // Windows rename can fail with EPERM while a contender or test polls the registration file.
+    Effect.andThen(
+      fs.rename(temp, options.file).pipe(Effect.retry({ times: 20, schedule: Schedule.spaced("10 millis") })),
+    ),
+  )
   yield* current.pipe(
     Effect.catchCause((cause) =>
       Effect.logWarning("managed service registration check failed; shutting down", {

packages/cli/test/service.test.ts

@@ -743,6 +743,7 @@ test("service registration replaces a stale owner with the bound address", async
       version: OPENCODE_VERSION,
       url: "http://127.0.0.1:4321",
       pid: process.pid,
+      startedAt: expect.any(Number),
       password: "secret",
     })
     await Effect.runPromise(cleanup.pipe(Effect.provide(NodeFileSystem.layer)))

packages/client/src/effect/service.ts

@@ -4,7 +4,7 @@ import { contenderPool, spawnServiceContender } from "../service-contender.js"
 import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js"
 import { matchesVersion } from "../service-version.js"
 import { PtyHandoff } from "../pty-handoff.js"
-import { decide, fallback, headers, probeResult, same } from "../service-probe.js"
+import { decide, fallback, headers, probeResult, same, withinStartupWindow } from "../service-probe.js"
 
 export * from "../service.js"
 export { headers }
@@ -72,12 +72,13 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
     const registration = yield* registered(options.file, timing.requestTimeout)
     const info = registration.info
     const service = registration.service
+    const starting = info !== undefined && (pool.starting(info.pid) || withinStartupWindow(info, timing))
     if (registration.timedOut && info !== undefined) {
       timeouts = {
         info,
         count: timeouts !== undefined && same(timeouts.info, info) ? timeouts.count + 1 : 1,
       }
-      if (timeouts.count >= 3) {
+      if (timeouts.count >= 3 && !starting) {
         yield* announce("missing")
         yield* Effect.logWarning("Background service is unresponsive; recovery cannot preserve persistent terminals")
         yield* Effect.tryPromise(() => PtyHandoff.clear(options.file ?? fallback()))
@@ -107,7 +108,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
 
     const failed = pool.reap()
     if (failed !== undefined) return yield* Effect.fail(failed)
-    if (pool.shouldRecruit(info !== undefined)) {
+    if (!starting && pool.shouldRecruit(info !== undefined)) {
       yield* announce("missing")
       pool.add(yield* spawnContender)
     }
@@ -144,6 +145,7 @@ export const Info = Schema.Struct({
   version: Schema.optional(Schema.String),
   url: Schema.String,
   pid: Schema.Int.check(Schema.isGreaterThan(0)),
+  startedAt: Schema.optional(Schema.Int.check(Schema.isGreaterThan(0))),
   password: Schema.optional(Schema.String),
 })
 

packages/client/src/promise/service.ts

@@ -4,7 +4,7 @@ import { contenderPool, spawnServiceContender } from "../service-contender.js"
 import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js"
 import { matchesVersion } from "../service-version.js"
 import { PtyHandoff } from "../pty-handoff.js"
-import { decide, fallback, headers, probeResult, same } from "../service-probe.js"
+import { decide, fallback, headers, probeResult, same, withinStartupWindow } from "../service-probe.js"
 
 export * from "../service.js"
 export { fallback, headers }
@@ -52,12 +52,15 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
       if (Date.now() >= deadline)
         throw pool.failure() ?? new Error("Timed out waiting for the background service to start")
       const registration = await registered(options.file, timing.requestTimeout)
+      const starting =
+        registration.info !== undefined &&
+        (pool.starting(registration.info.pid) || withinStartupWindow(registration.info, timing))
       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) {
+        if (timeouts.count >= 3 && !starting) {
           announce("missing")
           console.warn("Background service is unresponsive; recovery cannot preserve persistent terminals")
           await PtyHandoff.clear(options.file ?? fallback())
@@ -87,7 +90,7 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
       } else {
         const failed = pool.reap()
         if (failed !== undefined) throw failed
-        if (pool.shouldRecruit(registration.info !== undefined)) {
+        if (!starting && pool.shouldRecruit(registration.info !== undefined)) {
           announce("missing")
           pool.add(await spawnContender())
         }

packages/client/src/service-contender.ts

@@ -64,6 +64,10 @@ export function contenderPool(timing: EnsureTiming) {
   let spawnDelay = timing.spawnDelay
   let lastSpawn = 0
   return {
+    /** A live attempt owned by this ensure call still has its startup deadline to answer. */
+    starting(pid: number) {
+      return [...contenders].some((item) => item.child.pid === pid && !contenderFinished(item))
+    },
     /** The first startup failure seen since the last eviction. */
     failure: () => failure,
     /** A registered service answered, so the next attempt waits the base delay again. */

packages/client/src/service-probe.ts

@@ -1,7 +1,7 @@
 import { homedir } from "node:os"
 import { join } from "node:path"
 import type { DiscoverOptions, Endpoint, Info } from "./service.js"
-import { defaultEnsureTiming } from "./service-timing.js"
+import { defaultEnsureTiming, type EnsureTiming } from "./service-timing.js"
 import { matchesVersion } from "./service-version.js"
 
 // Shared by the Effect and Promise clients. Keep this module free of Effect so the
@@ -45,6 +45,16 @@ export function same(left: Info, right: Info) {
   return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid
 }
 
+/** Keep a registered service's shared startup window open for every concurrent client. */
+export function withinStartupWindow(info: Info, timing: EnsureTiming) {
+  return (
+    info.startedAt !== undefined &&
+    Number.isSafeInteger(info.startedAt) &&
+    info.startedAt > 0 &&
+    Date.now() < info.startedAt + timing.promiseTimeout
+  )
+}
+
 /** Ask a registered owner for its health and classify the answer. */
 export async function probeResult(info: Info, timeout = defaultEnsureTiming.requestTimeout) {
   const endpoint = {

packages/client/src/service.ts

@@ -52,6 +52,8 @@ export type Info = {
   readonly url: string
   /** Operating system process identifier. */
   readonly pid: number
+  /** Epoch milliseconds when the managed service began starting. */
+  readonly startedAt?: number
   /** Private service password, when authentication is enabled. */
   readonly password?: string
 }

packages/client/test/fixture/service.ts

@@ -48,6 +48,7 @@ if (mode === "old" || mode === "handoff") version = "old"
 if (mode === "incompatible") version = "1.9.0"
 if (mode === "compatible" || mode === "delayed-compatible") version = "2.1.0-next.1"
 const id = crypto.randomUUID()
+const readyAt = Date.now() + Number(delay ?? 0)
 const handoff = {
   directory: registration + ".daemon",
   instanceID: crypto.randomUUID(),
@@ -79,6 +80,7 @@ const server = Bun.serve({
       await appendFile(registration + ".requests", process.pid + "\n")
       return new Promise<Response>(() => {})
     }
+    if (mode === "coordinated") await Bun.sleep(Math.max(0, readyAt - Date.now()))
     if (mode === "modern" && requests === 1) {
       await writeFile(registration + ".first-request", "")
       while (!(await Bun.file(registration + ".release").exists())) await Bun.sleep(5)
@@ -116,11 +118,20 @@ await writeFile(
     version,
     url: server.url.toString(),
     pid: process.pid,
+    ...(mode === "coordinated" ? { startedAt: Date.now() } : {}),
     password: "private",
   }),
   { mode: 0o600 },
 )
-await rename(registration + ".tmp", registration)
+for (let attempt = 0; attempt < 20; attempt++) {
+  try {
+    await rename(registration + ".tmp", registration)
+    break
+  } catch (cause) {
+    if (attempt === 19) throw cause
+    await Bun.sleep(10)
+  }
+}
 
 async function shutdown(signal?: NodeJS.Signals) {
   if (signal !== undefined) await writeFile(registration + ".signal", signal)

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

@@ -48,7 +48,7 @@ test("ensures a missing service with native promises", async () => {
   const endpoint = await ensure({
     file: registration,
     version: "test",
-    command: fixture.command("coordinated"),
+    command: fixture.command("coordinated", "600"),
     onStart: (reason) => starts.push(reason),
   })
   const info = await Bun.file(registration).json()
@@ -58,6 +58,25 @@ test("ensures a missing service with native promises", async () => {
   expect(starts).toEqual(["missing"])
 })
 
+test("does not evict a slow service started by another client", async () => {
+  await using fixture = await serviceFixture()
+  const first = ensure({ file: fixture.registration, version: "test", command: fixture.command("coordinated", "600") })
+  const second = ensure({
+    file: fixture.registration,
+    version: "test",
+    command: fixture.command("coordinated", "600"),
+  })
+
+  await fixture.waitForFile()
+  const owner = await Bun.file(fixture.registration).json()
+  const [firstEndpoint, secondEndpoint] = await Promise.all([first, second])
+
+  expect(firstEndpoint.url).toBe(owner.url)
+  expect(secondEndpoint.url).toBe(owner.url)
+  expect((await Bun.file(fixture.registration).json()).pid).toBe(owner.pid)
+  expect(() => process.kill(owner.pid, 0)).not.toThrow()
+})
+
 test("adds configured environment variables with native promises", async () => {
   await using fixture = await serviceFixture()
   const registration = fixture.registration
@@ -172,20 +191,22 @@ test("accepts a surviving contender after a failure without recruiting replaceme
   expect(await fixture.starts()).toHaveLength(2)
 })
 
-test("recovers when an unresponsive contender is evicted after a prior failure", async () => {
+test("reports the prior failure at the startup deadline without killing a live contender", async () => {
   await using fixture = await serviceFixture()
-  const pending = ensure({ file: fixture.registration, version: "test", command: fixture.command("controlled") })
+  const pending = ensure({ file: fixture.registration, version: "test", command: fixture.command("controlled") }).catch(
+    (error: unknown) => error,
+  )
   const [first, second] = await fixture.waitForStarts(2)
   await fixture.release(second, "hang")
   await fixture.waitForFile()
   await fixture.release(first, "fail")
-  const [, , third] = await fixture.waitForStarts(3)
-  await fixture.release(third, "ready")
-  const endpoint = await pending
-  fixture.track(third)
-
-  expect((await Service.discover({ file: fixture.registration, version: "test" }))?.url).toBe(endpoint.url)
-  expect((await Bun.file(fixture.registration).json()).pid).toBe(third)
+  const error = await pending
+  expect(error).toBeInstanceOf(Error)
+  if (!(error instanceof Error)) throw error
+  expect(error.message).toContain("Server process exited with code 23")
+  expect(await fixture.starts()).toHaveLength(2)
+  expect((await Bun.file(fixture.registration).json()).pid).toBe(second)
+  expect(() => process.kill(second, 0)).not.toThrow()
 })
 
 test("reports a bounded contender stderr tail with native promises", async () => {

packages/client/test/service.test.ts

@@ -303,7 +303,7 @@ test("waits for a slow winner while bounding lock probes", async () => {
     ensure({
       file: registration,
       version: "test",
-      command: fixture.command("coordinated"),
+      command: fixture.command("coordinated", "600"),
     }),
   )
   const info = await Bun.file(registration).json()
@@ -314,6 +314,25 @@ test("waits for a slow winner while bounding lock probes", async () => {
   expect((await Bun.file(registration + ".starts").text()).trim().split("\n")).toHaveLength(2)
 })
 
+test("does not evict a slow service started by another client", async () => {
+  await using fixture = await serviceFixture()
+  const first = run(
+    ensure({ file: fixture.registration, version: "test", command: fixture.command("coordinated", "600") }),
+  )
+  const second = run(
+    ensure({ file: fixture.registration, version: "test", command: fixture.command("coordinated", "600") }),
+  )
+
+  await fixture.waitForFile()
+  const owner = await Bun.file(fixture.registration).json()
+  const [firstEndpoint, secondEndpoint] = await Promise.all([first, second])
+
+  expect(firstEndpoint.url).toBe(owner.url)
+  expect(secondEndpoint.url).toBe(owner.url)
+  expect((await Bun.file(fixture.registration).json()).pid).toBe(owner.pid)
+  expect(() => process.kill(owner.pid, 0)).not.toThrow()
+})
+
 test("waits for a live contender when another contender fails", async () => {
   await using fixture = await serviceFixture()
   const registration = fixture.registration
@@ -398,20 +417,22 @@ test("accepts a surviving contender after a failure without recruiting replaceme
   expect(await fixture.starts()).toHaveLength(2)
 })
 
-test("recovers when an unresponsive contender is evicted after a prior failure", async () => {
+test("reports the prior failure at the startup deadline without killing a live contender", async () => {
   await using fixture = await serviceFixture()
-  const pending = run(ensure({ file: fixture.registration, version: "test", command: fixture.command("controlled") }))
+  const pending = run(
+    ensure({ file: fixture.registration, version: "test", command: fixture.command("controlled") }),
+  ).catch((error: unknown) => error)
   const [first, second] = await fixture.waitForStarts(2)
   await fixture.release(second, "hang")
   await fixture.waitForFile()
   await fixture.release(first, "fail")
-  const [, , third] = await fixture.waitForStarts(3)
-  await fixture.release(third, "ready")
-  const endpoint = await pending
-  fixture.track(third)
-
-  expect((await run(Service.discover({ file: fixture.registration, version: "test" })))?.url).toBe(endpoint.url)
-  expect((await Bun.file(fixture.registration).json()).pid).toBe(third)
+  const error = await pending
+  expect(error).toBeInstanceOf(Error)
+  if (!(error instanceof Error)) throw error
+  expect(error.message).toContain("Server process exited with code 23")
+  expect(await fixture.starts()).toHaveLength(2)
+  expect((await Bun.file(fixture.registration).json()).pid).toBe(second)
+  expect(() => process.kill(second, 0)).not.toThrow()
 })
 
 test("reports a bounded contender stderr tail", async () => {