fix(tui): desynchronize managed reconnect ensure herd

packages/cli/src/mini/noninteractive.ts

@@ -55,7 +55,7 @@ const GLOBAL_FORM_SESSION_ID = "global"
 
 export async function runNonInteractivePrompt(input: Input) {
   const controller = new AbortController()
-  const stream = input.client.event.subscribe({ signal: controller.signal })[Symbol.asyncIterator]()
+  const stream = input.client.event.subscribe(undefined, { signal: controller.signal })[Symbol.asyncIterator]()
   const connected = await stream.next()
   if (connected.done) throw new Error("Event stream disconnected before prompt admission")
 

packages/cli/src/mini/stream-v2.transport.ts

@@ -832,7 +832,7 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
         const connection = new AbortController()
         const abortConnection = () => connection.abort()
         controller.signal.addEventListener("abort", abortConnection, { once: true })
-        const stream = input.sdk.event.subscribe({ signal: connection.signal })[Symbol.asyncIterator]()
+        const stream = input.sdk.event.subscribe(undefined, { signal: connection.signal })[Symbol.asyncIterator]()
         try {
           const first = await stream.next()
           if (first.done || first.value.type !== "server.connected") throw new Error("Event stream disconnected")

packages/cli/src/services/server-connection.ts

@@ -49,10 +49,12 @@ export const resolve = Effect.fn("cli.server-connection.resolve")(function* (arg
   } satisfies Resolved
 })
 
+/** Passive reconnect: ensure-running with no version replacement authority. */
+export const managedReconnect = (options: EnsureOptions) => Service.ensure({ ...options, version: undefined })
+
 function managedService(options: EnsureOptions) {
-  const reconnectOptions = { ...options, version: undefined }
   return {
-    reconnect: () => Service.ensure(reconnectOptions),
+    reconnect: () => managedReconnect(options),
     restart: () =>
       Effect.gen(function* () {
         yield* Service.stop(options)

packages/cli/test/server-connection.test.ts

@@ -1,14 +1,102 @@
 import { NodeFileSystem } from "@effect/platform-node"
+import type { EnsureReason } from "@opencode-ai/client/effect/service"
 import { Global } from "@opencode-ai/core/global"
 import { InstallationVersion } from "@opencode-ai/core/installation/version"
 import { expect, test } from "bun:test"
-import { Effect, FileSystem, Scope } from "effect"
+import { Effect, Exit, FileSystem, Scope } from "effect"
 import fs from "node:fs/promises"
 import os from "node:os"
 import path from "node:path"
 import { ServerConnection } from "../src/services/server-connection"
 import { ServiceConfig } from "../src/services/service-config"
 
+const runReconnect = <A, E>(effect: Effect.Effect<A, E, FileSystem.FileSystem>) =>
+  Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer)))
+
+async function withTempRoot(prefix: string, body: (root: string) => Promise<void>) {
+  const root = await fs.mkdtemp(path.join(os.tmpdir(), prefix))
+  try {
+    await body(root)
+  } finally {
+    await fs.rm(root, { recursive: true, force: true })
+  }
+}
+
+async function withHealthyRegisteredService(
+  prefix: string,
+  body: (input: { root: string; file: string; url: string }) => Promise<void>,
+) {
+  await withTempRoot(prefix, async (root) => {
+    const server = Bun.serve({
+      port: 0,
+      fetch() {
+        return Response.json({ healthy: true, version: "older", pid: process.pid })
+      },
+    })
+    const file = path.join(root, "service.json")
+    const url = server.url.toString()
+    await Bun.write(file, JSON.stringify({ url, pid: process.pid, version: "older" }))
+    try {
+      await body({ root, file, url })
+    } finally {
+      await server.stop(true)
+    }
+  })
+}
+
+test("managed reconnect ensures the service on the first failure", async () => {
+  await withTempRoot("opencode-service-reconnect-", async (root) => {
+    const starts: EnsureReason[] = []
+    // Short-circuit ensure once it decides to spawn — we only need the reason.
+    const exit = await runReconnect(
+      ServerConnection.managedReconnect({
+        file: path.join(root, "service.json"),
+        onStart: (reason) => {
+          starts.push(reason)
+          throw new Error("service ensure invoked")
+        },
+      }).pipe(Effect.exit),
+    )
+    expect(Exit.isFailure(exit)).toBe(true)
+    expect(starts).toEqual(["missing"])
+  })
+})
+
+test("managed reconnect reuses a healthy service from another version", async () => {
+  await withHealthyRegisteredService("opencode-service-reconnect-version-", async ({ root, file, url }) => {
+    const endpoint = await runReconnect(
+      ServerConnection.managedReconnect({
+        file,
+        version: "newer",
+        command: [path.join(root, "must-not-start")],
+      }),
+    )
+    expect(endpoint.url).toBe(url)
+  })
+})
+
+test("concurrent managed reconnects reuse one healthy endpoint without spawning", async () => {
+  await withHealthyRegisteredService("opencode-service-reconnect-herd-", async ({ root, file, url }) => {
+    const mustNotStart = path.join(root, "must-not-start")
+    const endpoints = await Promise.all(
+      Array.from({ length: 7 }, () =>
+        runReconnect(
+          ServerConnection.managedReconnect({
+            file,
+            version: "newer",
+            command: [mustNotStart],
+          }),
+        ),
+      ),
+    )
+    expect(endpoints.map((endpoint) => endpoint.url)).toEqual(Array.from({ length: 7 }, () => url))
+    expect(await fs.stat(mustNotStart).then(
+      () => true,
+      () => false,
+    )).toBe(false)
+  })
+})
+
 test("resolution groups Effect-native lifecycle operations only for the managed service", async () => {
   const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-server-resolution-"))
   const id = "server-resolution-test"

packages/client/src/effect/api/api.ts

@@ -767,8 +767,13 @@ export interface SkillApi<E = never> {
   readonly list: SkillListOperation<E>
 }
 
+type Endpoint19_0Request = Parameters<RawClient["server.event"]["event.subscribe"]>[0]
+export type Endpoint19_0Input = {
+  readonly location?: Endpoint19_0Request["query"]["location"]
+  readonly session?: Endpoint19_0Request["query"]["session"]
+}
 export type Endpoint19_0Output = StreamValue<EffectValue<ReturnType<RawClient["server.event"]["event.subscribe"]>>>
-export type EventSubscribeOperation<E = never> = () => Stream.Stream<Endpoint19_0Output, E>
+export type EventSubscribeOperation<E = never> = (input?: Endpoint19_0Input) => Stream.Stream<Endpoint19_0Output, E>
 
 export interface EventApi<E = never> {
   readonly subscribe: EventSubscribeOperation<E>
@@ -992,8 +997,12 @@ export type Endpoint26_1Input = { readonly location?: Endpoint26_1Request["query
 export type Endpoint26_1Output = EffectValue<ReturnType<RawClient["server.debug"]["debug.location.evict"]>>
 export type DebugLocationEvictOperation<E = never> = (input?: Endpoint26_1Input) => Effect.Effect<Endpoint26_1Output, E>
 
+export type Endpoint26_2Output = EffectValue<ReturnType<RawClient["server.debug"]["debug.event-feed"]>>
+export type DebugEventFeedOperation<E = never> = () => Effect.Effect<Endpoint26_2Output, E>
+
 export interface DebugApi<E = never> {
   readonly location: { readonly list: DebugLocationListOperation<E>; readonly evict: DebugLocationEvictOperation<E> }
+  readonly "event-feed": DebugEventFeedOperation<E>
 }
 
 export interface AppApi<E = never> {

packages/client/src/effect/generated/client.ts

@@ -908,9 +908,14 @@ const Endpoint18_0 = (raw: RawClient["server.skill"]) => (input?: Endpoint18_0In
 
 const adaptGroup18 = (raw: RawClient["server.skill"]) => ({ list: Endpoint18_0(raw) })
 
-const Endpoint19_0 = (raw: RawClient["server.event"]) => () =>
+type Endpoint19_0Request = Parameters<RawClient["server.event"]["event.subscribe"]>[0]
+type Endpoint19_0Input = {
+  readonly location?: Endpoint19_0Request["query"]["location"]
+  readonly session?: Endpoint19_0Request["query"]["session"]
+}
+const Endpoint19_0 = (raw: RawClient["server.event"]) => (input?: Endpoint19_0Input) =>
   Stream.unwrap(
-    raw["event.subscribe"]({}).pipe(
+    raw["event.subscribe"]({ query: { location: input?.["location"], session: input?.["session"] } }).pipe(
       Effect.mapError(mapClientError),
       Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))),
     ),
@@ -1181,8 +1186,12 @@ type Endpoint26_1Input = { readonly location?: Endpoint26_1Request["query"]["loc
 const Endpoint26_1 = (raw: RawClient["server.debug"]) => (input?: Endpoint26_1Input) =>
   raw["debug.location.evict"]({ query: { location: input?.["location"] } }).pipe(Effect.mapError(mapClientError))
 
+const Endpoint26_2 = (raw: RawClient["server.debug"]) => () =>
+  raw["debug.event-feed"]({}).pipe(Effect.mapError(mapClientError))
+
 const adaptGroup26 = (raw: RawClient["server.debug"]) => ({
   location: { list: Endpoint26_0(raw), evict: Endpoint26_1(raw) },
+  "event-feed": Endpoint26_2(raw),
 })
 
 const adaptClient = (raw: RawClient) => ({

packages/client/src/promise/generated/client.ts

@@ -151,6 +151,7 @@ import type {
   CommandListOutput,
   SkillListInput,
   SkillListOutput,
+  EventSubscribeInput,
   EventSubscribeOutput,
   PtyListInput,
   PtyListOutput,
@@ -197,6 +198,7 @@ import type {
   DebugLocationListOutput,
   DebugLocationEvictInput,
   DebugLocationEvictOutput,
+  DebugEventFeedOutput,
 } from "./types"
 import { ClientError } from "./client-error"
 
@@ -292,6 +294,12 @@ export function make(options: ClientOptions) {
       const reader = response.body.getReader()
       const decoder = new TextDecoder()
       let buffer = ""
+      const signal = requestOptions?.signal
+      const onAbort = () => {
+        void reader.cancel(signal?.reason).catch(() => undefined)
+      }
+      if (signal?.aborted) onAbort()
+      else signal?.addEventListener("abort", onAbort, { once: true })
       try {
         while (true) {
           let next
@@ -327,10 +335,16 @@ export function make(options: ClientOptions) {
           if (next.done) return
         }
       } finally {
+        signal?.removeEventListener("abort", onAbort)
         try {
           await reader.cancel()
         } catch {}
-        reader.releaseLock()
+        try {
+          reader.releaseLock()
+        } catch {}
+        try {
+          await response.body?.cancel()
+        } catch {}
       }
     },
   })
@@ -1350,9 +1364,16 @@ export function make(options: ClientOptions) {
         ),
     },
     event: {
-      subscribe: (requestOptions?: RequestOptions): AsyncIterable<EventSubscribeOutput> =>
+      subscribe: (input?: EventSubscribeInput, requestOptions?: RequestOptions): AsyncIterable<EventSubscribeOutput> =>
         sse<EventSubscribeOutput>(
-          { method: "GET", path: `/api/event`, successStatus: 200, declaredStatuses: [401, 400], empty: false },
+          {
+            method: "GET",
+            path: `/api/event`,
+            query: { location: input?.["location"], session: input?.["session"] },
+            successStatus: 200,
+            declaredStatuses: [401, 400],
+            empty: false,
+          },
           requestOptions,
         ),
     },
@@ -1663,6 +1684,17 @@ export function make(options: ClientOptions) {
             requestOptions,
           ),
       },
+      "event-feed": (requestOptions?: RequestOptions) =>
+        request<DebugEventFeedOutput>(
+          {
+            method: "GET",
+            path: `/api/debug/event-feed`,
+            successStatus: 200,
+            declaredStatuses: [401, 400],
+            empty: false,
+          },
+          requestOptions,
+        ),
     },
   }
 }

packages/client/src/promise/generated/types.ts

@@ -4542,6 +4542,17 @@ export type SkillListOutput = {
   data: Array<SkillInfo>
 }
 
+export type EventSubscribeInput = {
+  readonly location?: {
+    readonly location?: { readonly directory: string; readonly workspace?: string | undefined } | undefined
+    readonly session?: string | ReadonlyArray<string> | ReadonlyArray<string> | undefined
+  }["location"]
+  readonly session?: {
+    readonly location?: { readonly directory: string; readonly workspace?: string | undefined } | undefined
+    readonly session?: string | ReadonlyArray<string> | ReadonlyArray<string> | undefined
+  }["session"]
+}
+
 export type EventSubscribeOutput = V2Event
 
 export type PtyListInput = {
@@ -4859,3 +4870,12 @@ export type DebugLocationEvictInput = {
 }
 
 export type DebugLocationEvictOutput = void
+
+export type DebugEventFeedOutput = {
+  active: number
+  opens: number
+  closes: number
+  serializedEvents: number
+  serializedBytes: number
+  overflows: number
+}