feat(tui): batch event delivery

packages/cli/test/acp/service-lifecycle.test.ts

@@ -136,7 +136,7 @@ describe("acp service lifecycle", () => {
       method: "POST",
       path: "/api/session/ses_loaded/fork",
       query: {},
-      body: { boundary: { type: "through" } },
+      body: {},
     })
   })
 

packages/core/src/codemode/catalog.ts

@@ -6,7 +6,6 @@ export const Entry = Schema.Struct({
   path: Schema.String,
   description: Schema.String,
   signature: Schema.String,
-  pinned: Schema.optionalKey(Schema.Boolean),
 })
 export type Entry = typeof Entry.Type
 
@@ -57,39 +56,26 @@ export function summarize(entries: ReadonlyArray<Entry>, budget = INLINE_BUDGET)
           if (left.path > right.path) return 1
           return 0
         })
-      const ranked = rankListings(listings)
-      const pinned = new Set(
-        namespaceEntries
-          .filter((entry) => entry.pinned)
-          .map((entry) => listings.find((listing) => listing.path === entry.path))
-          .filter((listing) => listing !== undefined),
-      )
       return {
         name,
         listings,
-        selectionOrder: ranked.filter((candidate) => !pinned.has(candidate.listing)),
-        selectedListings: pinned,
-        selectionIndex: 0,
+        selectionOrder: rankListings(listings),
+        selectedListings: new Set<typeof Listing.Type>(),
       }
     })
 
   const active = new Set(namespaces)
-  let remaining =
-    budget -
-    namespaces
-      .flatMap((namespace) => namespace.listings.filter((listing) => namespace.selectedListings.has(listing)))
-      .reduce((total, listing) => total + Math.round(listing.line.length / CHARACTERS_PER_TOKEN), 0)
+  let remaining = budget
   while (active.size > 0) {
     for (const namespace of active) {
-      const candidate = namespace.selectionOrder[namespace.selectionIndex]
+      const candidate = namespace.selectionOrder[namespace.selectedListings.size]
       if (!candidate || candidate.cost > remaining) {
         active.delete(namespace)
         continue
       }
       namespace.selectedListings.add(candidate.listing)
-      namespace.selectionIndex += 1
       remaining -= candidate.cost
-      if (namespace.selectionIndex === namespace.selectionOrder.length) active.delete(namespace)
+      if (namespace.selectedListings.size === namespace.selectionOrder.length) active.delete(namespace)
     }
   }
 

packages/core/src/codemode/tool.ts

@@ -138,14 +138,7 @@ export const create = (
 }
 
 export const catalog = (registrations: ReadonlyMap<string, Info>) => {
-  const pinned = new Set(
-    Array.from(registrations.values())
-      .filter((registration) => registration.options?.pinned === true)
-      .map(qualifiedName),
-  )
-  return runtime(registrations, () => Effect.fail(toolError("Execute context is unavailable")))
-    .catalog()
-    .map((entry) => ({ ...entry, pinned: pinned.has(entry.path) }))
+  return runtime(registrations, () => Effect.fail(toolError("Execute context is unavailable"))).catalog()
 }
 
 function runtime(
@@ -156,7 +149,11 @@ function runtime(
   const tools: Record<string, Tool.Tool<never>> = {}
   for (const [name, registration] of registrations) {
     const child = definition(registration)
-    const path = qualifiedName(registration)
+    const normalized = registration.name.replace(/[^a-zA-Z0-9_-]/g, "_")
+    const path =
+      registration.options?.namespace === undefined
+        ? normalized
+        : `${registration.options.namespace}.${normalized}`
     tools[path] = Tool.make({
       description: child.description,
       input: child.inputSchema,
@@ -167,12 +164,6 @@ function runtime(
   return CodeMode.make<typeof tools>({ tools, ...hooks })
 }
 
-function qualifiedName(registration: Info) {
-  const normalized = registration.name.replace(/[^a-zA-Z0-9_-]/g, "_")
-  if (registration.options?.namespace === undefined) return normalized
-  return `${registration.options.namespace}.${normalized}`
-}
-
 // Tool inputs arrive as parsed JSON, so the JSON value cast is a boundary fact.
 function displayInput(input: unknown): Record<string, typeof Schema.Json.Type> | undefined {
   if (input === null || input === undefined) return

packages/core/test/codemode.test.ts

@@ -16,7 +16,6 @@ describe("CodeMode", () => {
             description: "Echo text",
             input: Schema.Struct({ text: Schema.String }),
             output: Schema.String,
-            options: { pinned: true },
             execute: ({ text }) => Effect.succeed({ output: text }),
         }),
       )
@@ -28,7 +27,6 @@ describe("CodeMode", () => {
           path: "echo",
           description: "Echo text",
           signature: "tools.echo(input: {\n  text: string,\n}): Promise<string>",
-          pinned: true,
         },
       ])
     }).pipe(

packages/core/test/codemode/catalog.test.ts

@@ -2,11 +2,10 @@ import { describe, expect, test } from "bun:test"
 import { CodeModeCatalog } from "@opencode-ai/core/codemode/catalog"
 import { CodeModeInstructions } from "@opencode-ai/core/codemode/instructions"
 
-const entry = (path: string, description: string, signature?: string, pinned = false): CodeModeCatalog.Entry => ({
+const entry = (path: string, description: string, signature?: string): CodeModeCatalog.Entry => ({
   path,
   description,
   signature: signature ?? `tools.${path}(input: {\n  q: string,\n}): Promise<string>`,
-  pinned,
 })
 
 const lookup = entry(
@@ -47,30 +46,6 @@ describe("CodeModeCatalog.summarize", () => {
     expect(catalog.namespaces.every((namespace) => namespace.entries.length === 0)).toBe(true)
   })
 
-  test("always retains pinned tools beyond the inline budget", () => {
-    const pinned = [
-      entry("alpha.first", "First", undefined, true),
-      entry("beta.second", "Second", undefined, true),
-    ]
-    const catalog = CodeModeCatalog.summarize([...pinned, entry("alpha.unpinned", "Unpinned")], 0)
-
-    expect(catalog.shown).toBe(2)
-    expect(catalog.namespaces.flatMap((namespace) => namespace.entries.map((item) => item.path))).toEqual([
-      "alpha.first",
-      "beta.second",
-    ])
-  })
-
-  test("spends the budget remaining after pinned tools on unpinned tools", () => {
-    const pinned = entry("alpha.pinned", "Pinned", undefined, true)
-    const unpinned = entry("beta.unpinned", "Unpinned")
-    const pinCost = Math.round(`  - ${pinned.signature} // Pinned`.length / 4)
-    const unpinnedCost = Math.round(`  - ${unpinned.signature} // Unpinned`.length / 4)
-
-    expect(CodeModeCatalog.summarize([pinned, unpinned], pinCost + unpinnedCost).shown).toBe(2)
-    expect(CodeModeCatalog.summarize([pinned, unpinned], pinCost + unpinnedCost - 1).shown).toBe(1)
-  })
-
   test("retains only the rendered portion of inline descriptions", () => {
     const catalog = CodeModeCatalog.summarize([entry("alpha.one", `Summary\n${"detail".repeat(10_000)}`)])
     expect(catalog.namespaces[0]?.entries[0]?.line).toEndWith("// Summary")

packages/core/test/session-runner-tool-registry.test.ts

@@ -67,7 +67,7 @@ const transform = (
 ) =>
   service.transform((draft) =>
     Object.entries(tools).forEach(([name, tool]) =>
-      draft.add({ ...tool, name, options: options ?? tool.options }),
+      draft.add({ ...tool, name, options: { ...tool.options, ...options } }),
     ),
   )
 

packages/core/test/session-runner.test.ts

@@ -231,7 +231,7 @@ const permission = Layer.succeed(
 const transformTools = (registry: Tool.Interface, tools: Readonly<Record<string, Info>>, options?: Tool.Options) =>
   registry.transform((draft) =>
     Object.entries(tools).forEach(([name, tool]) =>
-      draft.add({ ...tool, name, options: options ?? tool.options }),
+      draft.add({ ...tool, name, options: { ...tool.options, ...options } }),
     ),
   )
 const echo = Layer.effectDiscard(

packages/schema/src/tool.ts

@@ -19,23 +19,12 @@ export interface Context {
   readonly progress: (update: Metadata) => Effect.Effect<void>
 }
 
-interface BaseOptions {
+export interface Options {
   readonly namespace?: string
+  readonly codemode?: boolean
   readonly permission?: string
 }
 
-export type Options = BaseOptions &
-  (
-    | {
-        readonly codemode?: true
-        readonly pinned?: boolean
-      }
-    | {
-        readonly codemode: boolean
-        readonly pinned?: never
-      }
-  )
-
 export type ValueSchema<A = unknown> =
   | Schema.Codec<A, any>
   | (StandardSchemaV1<any, A> & StandardJSONSchemaV1<any, A>)

packages/tui/src/context/client.tsx

@@ -1,6 +1,6 @@
 import type { OpenCodeClient, OpenCodeEvent } from "@opencode-ai/client"
 import { createGlobalEmitter } from "@solid-primitives/event-bus"
-import { onCleanup, onMount } from "solid-js"
+import { batch, onCleanup, onMount } from "solid-js"
 import { createStore } from "solid-js/store"
 import { errorMessage } from "../util/error"
 import { createSimpleContext } from "./helper"
@@ -25,6 +25,7 @@ type ManagedService = {
 type ClientEventMap = { [Type in OpenCodeEvent["type"]]: Extract<OpenCodeEvent, { type: Type }> }
 const connectTimeout = 2_000
 const connectionHistoryLimit = 50
+const eventFlushInterval = 10
 
 export const { use: useClient, provider: ClientProvider } = createSimpleContext({
   name: "Client",
@@ -34,6 +35,8 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
     const history: ClientConnectionEvent[] = []
     let api = props.api
     const events = createGlobalEmitter<ClientEventMap>()
+    const pending: OpenCodeEvent[] = []
+    let flushTimer: ReturnType<typeof setTimeout> | undefined
     const [connection, setConnection] = createStore<{
       status: ClientConnectionStatus
       attempt: number
@@ -49,6 +52,18 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
       if (history.length > connectionHistoryLimit) history.shift()
     }
 
+    function flushEvents() {
+      flushTimer = undefined
+      const queued = pending.splice(0)
+      batch(() => queued.forEach((event) => events.emit(event.type, event)))
+    }
+
+    function emit(event: OpenCodeEvent) {
+      pending.push(event)
+      if (flushTimer) return
+      flushTimer = setTimeout(flushEvents, eventFlushInterval)
+    }
+
     async function connect(signal: AbortSignal, attempt: number) {
       let connectedAt: number | undefined
 
@@ -80,7 +95,7 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
         record("connected", attempt)
         connectedAt = Date.now()
         log.info("event stream connected")
-        events.emit(first.value.type, first.value)
+        emit(first.value)
         setConnection({ status: "connected", attempt: 0, error: undefined })
 
         // Forward events until the stream closes or this connection is cancelled.
@@ -97,7 +112,7 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
               seq: event.value.durable.seq,
             })
 
-          events.emit(event.value.type, event.value)
+          emit(event.value)
         }
 
         return { error: undefined, connectedAt }
@@ -154,6 +169,8 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
     onCleanup(() => {
       abort.abort()
       stream?.abort()
+      if (flushTimer) clearTimeout(flushTimer)
+      pending.length = 0
       events.clear()
     })
 

packages/tui/test/cli/tui/client-connection-characterization.test.tsx

@@ -1,634 +0,0 @@
-/** @jsxImportSource @opentui/solid */
-import { afterAll, describe, expect, test } from "bun:test"
-import type { OpenCodeClient, OpenCodeEvent } from "@opencode-ai/client"
-import { testRender } from "@opentui/solid"
-import { onMount } from "solid-js"
-import type { LogLevel, LogSink } from "../../../src/context/log"
-import { createApi, createFetch } from "../../fixture/tui-client"
-
-const packageRoot = process.env.OPENCODE_TUI_ROOT
-const contextModule = packageRoot
-  ? await import(`${packageRoot}/src/context/client.tsx`)
-  : await import("../../../src/context/client")
-const environmentModule = packageRoot
-  ? await import(`${packageRoot}/test/fixture/tui-environment.tsx`)
-  : await import("../../fixture/tui-environment")
-const { ClientProvider, useClient } = contextModule as typeof import("../../../src/context/client")
-const { TestTuiContexts } = environmentModule as typeof import("../../fixture/tui-environment")
-
-type Client = ReturnType<typeof useClient>
-type Service = {
-  reconnect: (signal: AbortSignal) => Promise<{ api: OpenCodeClient }>
-  restart: () => Promise<void>
-}
-type Observation = {
-  scenario: string
-  value: unknown
-}
-
-const observations: Observation[] = []
-const connected = { id: "evt_connected", type: "server.connected", data: {} } as OpenCodeEvent
-
-afterAll(async () => {
-  const output = process.env.CLIENT_BEHAVIOR_OUTPUT
-  if (output) await Bun.write(output, `${JSON.stringify(observations, null, 2)}\n`)
-})
-
-function observe(scenario: string, value: unknown) {
-  observations.push({ scenario, value })
-}
-
-function normalizeError(error: unknown) {
-  if (error instanceof Error) return `${error.name}:${error.message}`
-  return String(error)
-}
-
-function history(client: Client) {
-  return client.connection.internal.history().map((event) => ({
-    status: event.data.status,
-    attempt: event.data.attempt,
-    error: event.data.error,
-  }))
-}
-
-async function waitFor(check: () => boolean, timeout = 3_000) {
-  const started = Date.now()
-  while (!check()) {
-    if (Date.now() - started > timeout) throw new Error("timed out waiting for condition")
-    await Bun.sleep(5)
-  }
-}
-
-function event(type: "vcs" | "update" | "rename", suffix: string): OpenCodeEvent {
-  if (type === "vcs") {
-    return {
-      id: `evt_vcs_${suffix}`,
-      created: 1,
-      type: "vcs.branch.updated",
-      location: { directory: "/tmp/project" },
-      data: { branch: suffix },
-    }
-  }
-  if (type === "update") {
-    return {
-      id: `evt_update_${suffix}`,
-      created: 2,
-      type: "installation.update-available",
-      data: { version: suffix },
-    }
-  }
-  return {
-    id: `evt_rename_${suffix}`,
-    created: 3,
-    type: "session.renamed",
-    durable: { aggregateID: "ses_test", seq: 1, version: 1 },
-    location: { directory: "/tmp/project" },
-    data: { sessionID: "ses_test", title: suffix },
-  }
-}
-
-function createStream(options?: { first?: OpenCodeEvent; closeBeforeHandshake?: boolean }) {
-  const encoder = new TextEncoder()
-  const controllers = new Set<ReadableStreamDefaultController<Uint8Array>>()
-  const requests: Request[] = []
-  const aborts: string[] = []
-  let cancellations = 0
-
-  function response(request: Request) {
-    requests.push(request)
-    request.signal.addEventListener("abort", () => aborts.push(normalizeError(request.signal.reason)), { once: true })
-
-    let current: ReadableStreamDefaultController<Uint8Array> | undefined
-    return new Response(
-      new ReadableStream<Uint8Array>({
-        start(controller) {
-          current = controller
-          controllers.add(controller)
-          if (options?.closeBeforeHandshake) {
-            controllers.delete(controller)
-            controller.close()
-            return
-          }
-          controller.enqueue(encoder.encode(`data: ${JSON.stringify(options?.first ?? connected)}\n\n`))
-        },
-        cancel() {
-          cancellations += 1
-          if (current) controllers.delete(current)
-        },
-      }),
-      { headers: { "content-type": "text/event-stream" } },
-    )
-  }
-
-  return {
-    response,
-    emit(value: OpenCodeEvent) {
-      const chunk = encoder.encode(`data: ${JSON.stringify(value)}\n\n`)
-      for (const controller of controllers) controller.enqueue(chunk)
-    },
-    raw(value: string) {
-      const chunk = encoder.encode(value)
-      for (const controller of controllers) controller.enqueue(chunk)
-    },
-    close() {
-      for (const controller of [...controllers]) {
-        controllers.delete(controller)
-        controller.close()
-      }
-    },
-    fail(message: string) {
-      for (const controller of [...controllers]) {
-        controllers.delete(controller)
-        controller.error(new Error(message))
-      }
-    },
-    snapshot() {
-      return {
-        requests: requests.length,
-        requestAborted: requests.map((request) => request.signal.aborted),
-        aborts,
-        cancellations,
-        active: controllers.size,
-      }
-    },
-  }
-}
-
-function apiFor(stream: ReturnType<typeof createStream>) {
-  return createApi(
-    createFetch((url, request) => {
-      if (url.pathname === "/api/event") return stream.response(request)
-    }).fetch,
-  )
-}
-
-async function mount(input: {
-  api: OpenCodeClient
-  service?: Service
-  throwOn?: OpenCodeEvent["type"]
-}) {
-  const seen: Array<{ type: string; status: string }> = []
-  const typed: string[] = []
-  const logs: Array<{ level: LogLevel; message: string; tags: Record<string, unknown> }> = []
-  let initialStatus = ""
-  let client!: Client
-  let ready!: () => void
-  const mounted = new Promise<void>((resolve) => {
-    ready = resolve
-  })
-  const log: LogSink = (level, message, tags) => {
-    logs.push({ level, message, tags: { ...tags } })
-  }
-
-  const app = await testRender(() => (
-    <TestTuiContexts log={log}>
-      <ClientProvider api={input.api} service={input.service}>
-        <Probe
-          onReady={(value) => {
-            client = value
-            initialStatus = value.connection.status()
-            ready()
-          }}
-          onEvent={(value) => {
-            seen.push({ type: value.type, status: client.connection.status() })
-            if (value.type === input.throwOn) throw new Error(`listener failed for ${value.type}`)
-          }}
-          onBranch={(branch) => typed.push(branch)}
-        />
-      </ClientProvider>
-    </TestTuiContexts>
-  ))
-  await mounted
-
-  return { app, client, initialStatus, seen, typed, logs }
-}
-
-function Probe(props: {
-  onReady: (client: Client) => void
-  onEvent: (event: OpenCodeEvent) => void
-  onBranch: (branch: string) => void
-}) {
-  const client = useClient()
-  onMount(() => {
-    client.event.listen(({ details }) => props.onEvent(details))
-    client.event.on("vcs.branch.updated", (value) => props.onBranch(value.data.branch ?? ""))
-    props.onReady(client)
-  })
-  return <box />
-}
-
-describe("ClientProvider connection characterization", () => {
-  test("records handshake ordering, event delivery, logging, and active-stream cleanup", async () => {
-    const stream = createStream()
-    const setup = await mount({ api: apiFor(stream) })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    stream.emit(event("vcs", "main"))
-    stream.emit(event("rename", "renamed"))
-    stream.emit(event("update", "2.0.0"))
-    await waitFor(() => setup.seen.length === 4)
-
-    observe("healthy.connected", {
-      initialStatus: setup.initialStatus,
-      finalStatus: setup.client.connection.status(),
-      seen: setup.seen,
-      typed: setup.typed,
-      logs: setup.logs,
-      history: history(setup.client),
-      stream: stream.snapshot(),
-    })
-
-    setup.app.renderer.destroy()
-    await waitFor(() => stream.snapshot().requestAborted[0] === true)
-    await Bun.sleep(20)
-
-    observe("healthy.cleanup", {
-      history: history(setup.client),
-      stream: stream.snapshot(),
-    })
-
-    expect(setup.seen.map((item) => item.type)).toEqual([
-      "server.connected",
-      "vcs.branch.updated",
-      "session.renamed",
-      "installation.update-available",
-    ])
-    expect(setup.seen.map((item) => item.status)).toEqual(["connecting", "connected", "connected", "connected"])
-    expect(setup.logs.filter((item) => item.message === "event")).toHaveLength(1)
-  })
-
-  test("records an invalid first event", async () => {
-    const stream = createStream({ first: event("vcs", "invalid-handshake") })
-    const setup = await mount({ api: apiFor(stream) })
-
-    await waitFor(() => setup.client.connection.status() === "reconnecting")
-    observe("handshake.invalid", {
-      status: setup.client.connection.status(),
-      error: setup.client.connection.error(),
-      seen: setup.seen,
-      history: history(setup.client),
-      stream: stream.snapshot(),
-    })
-
-    setup.app.renderer.destroy()
-    expect(setup.client.connection.error()).toBe("Event stream did not start with server.connected")
-  })
-
-  test("records EOF before the handshake", async () => {
-    const stream = createStream({ closeBeforeHandshake: true })
-    const setup = await mount({ api: apiFor(stream) })
-
-    await waitFor(() => setup.client.connection.status() === "reconnecting")
-    observe("handshake.eof", {
-      status: setup.client.connection.status(),
-      error: setup.client.connection.error(),
-      seen: setup.seen,
-      history: history(setup.client),
-      stream: stream.snapshot(),
-    })
-
-    setup.app.renderer.destroy()
-    expect(setup.client.connection.error()).toBe("Event stream disconnected")
-  })
-
-  test("records a fetch failure before the handshake", async () => {
-    const calls = createFetch((url) => {
-      if (url.pathname === "/api/event") throw new Error("network unavailable")
-      return undefined
-    })
-    const setup = await mount({ api: createApi(calls.fetch) })
-
-    await waitFor(() => setup.client.connection.status() === "reconnecting")
-    observe("handshake.fetch-error", {
-      status: setup.client.connection.status(),
-      error: setup.client.connection.error(),
-      seen: setup.seen,
-      history: history(setup.client),
-      logs: setup.logs,
-    })
-
-    setup.app.renderer.destroy()
-    expect(setup.client.connection.error()).toBe("Transport")
-  })
-
-  test("records the initial connection timeout and request cancellation", async () => {
-    const requests: Request[] = []
-    const calls = createFetch((url, request) => {
-      if (url.pathname !== "/api/event") return
-      requests.push(request)
-      return new Promise<Response>((_, reject) => {
-        request.signal.addEventListener("abort", () => reject(request.signal.reason), { once: true })
-      })
-    })
-    const setup = await mount({ api: createApi(calls.fetch) })
-
-    await waitFor(() => setup.client.connection.status() === "reconnecting", 3_000)
-    observe("handshake.timeout", {
-      status: setup.client.connection.status(),
-      error: setup.client.connection.error(),
-      requestCount: requests.length,
-      requestAborted: requests.map((request) => request.signal.aborted),
-      abortReasons: requests.map((request) => normalizeError(request.signal.reason)),
-      history: history(setup.client),
-    })
-
-    setup.app.renderer.destroy()
-    expect(setup.client.connection.error()).toBe("Transport")
-  })
-
-  test("records static transport reconnection after a connected stream closes", async () => {
-    const stream = createStream()
-    const setup = await mount({ api: apiFor(stream) })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    stream.close()
-    await waitFor(() => stream.snapshot().requests === 2, 2_000)
-    await waitFor(() => setup.client.connection.status() === "connected")
-
-    observe("reconnect.static", {
-      status: setup.client.connection.status(),
-      seen: setup.seen,
-      history: history(setup.client),
-      stream: stream.snapshot(),
-      logs: setup.logs.filter((item) => item.message !== "event"),
-    })
-
-    setup.app.renderer.destroy()
-    expect(setup.seen.map((item) => item.type)).toEqual(["server.connected", "server.connected"])
-  })
-
-  test("records immediate managed-service replacement", async () => {
-    const initial = createStream()
-    const replacement = createStream()
-    const replacementApi = apiFor(replacement)
-    const reconnectSignals: boolean[] = []
-    const service: Service = {
-      reconnect(signal) {
-        reconnectSignals.push(signal.aborted)
-        return Promise.resolve({ api: replacementApi })
-      },
-      restart: () => Promise.resolve(),
-    }
-    const setup = await mount({ api: apiFor(initial), service })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    initial.close()
-    await waitFor(() => replacement.snapshot().requests === 1)
-    await waitFor(() => setup.client.connection.status() === "connected")
-    replacement.emit(event("vcs", "replacement"))
-    await waitFor(() => setup.typed.includes("replacement"))
-
-    observe("reconnect.managed-replacement", {
-      status: setup.client.connection.status(),
-      apiReplaced: setup.client.api === replacementApi,
-      reconnectSignals,
-      seen: setup.seen,
-      typed: setup.typed,
-      history: history(setup.client),
-      initial: initial.snapshot(),
-      replacement: replacement.snapshot(),
-    })
-
-    setup.app.renderer.destroy()
-    expect(setup.client.api).toBe(replacementApi)
-  })
-
-  test("records managed-service resolution failure and delayed retry", async () => {
-    const stream = createStream()
-    let reconnects = 0
-    const service: Service = {
-      reconnect() {
-        reconnects += 1
-        return Promise.reject(new Error("service unavailable"))
-      },
-      restart: () => Promise.resolve(),
-    }
-    const setup = await mount({ api: apiFor(stream), service })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    stream.close()
-    await waitFor(() => stream.snapshot().requests === 2, 2_000)
-    await waitFor(() => setup.client.connection.status() === "connected")
-
-    observe("reconnect.managed-failure", {
-      reconnects,
-      status: setup.client.connection.status(),
-      seen: setup.seen,
-      history: history(setup.client),
-      stream: stream.snapshot(),
-      resolutionLogs: setup.logs.filter((item) => item.message === "server resolution failed"),
-    })
-
-    setup.app.renderer.destroy()
-    expect(reconnects).toBe(1)
-  })
-
-  test("records cleanup while the initial fetch is pending", async () => {
-    const requests: Request[] = []
-    const aborts: string[] = []
-    const calls = createFetch((url, request) => {
-      if (url.pathname !== "/api/event") return
-      requests.push(request)
-      return new Promise<Response>((_, reject) => {
-        request.signal.addEventListener(
-          "abort",
-          () => {
-            aborts.push(normalizeError(request.signal.reason))
-            reject(request.signal.reason)
-          },
-          { once: true },
-        )
-      })
-    })
-    const setup = await mount({ api: createApi(calls.fetch) })
-
-    await waitFor(() => requests.length === 1)
-    setup.app.renderer.destroy()
-    await waitFor(() => requests[0].signal.aborted)
-    await Bun.sleep(20)
-
-    observe("cleanup.pending-handshake", {
-      status: setup.client.connection.status(),
-      requestAborted: requests[0].signal.aborted,
-      aborts,
-      history: history(setup.client),
-      logs: setup.logs,
-    })
-
-    expect(history(setup.client).map((item) => item.status)).toEqual(["connecting"])
-  })
-
-  test("records an event listener failure as a connection failure", async () => {
-    const stream = createStream()
-    const setup = await mount({ api: apiFor(stream), throwOn: "vcs.branch.updated" })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    stream.emit(event("vcs", "throws"))
-    await waitFor(() => setup.client.connection.status() === "reconnecting")
-
-    observe("listener.failure", {
-      status: setup.client.connection.status(),
-      error: setup.client.connection.error(),
-      seen: setup.seen,
-      typed: setup.typed,
-      history: history(setup.client),
-      stream: stream.snapshot(),
-    })
-
-    setup.app.renderer.destroy()
-    expect(setup.client.connection.error()).toBe("listener failed for vcs.branch.updated")
-  })
-
-  test("records stream reader failure after connection", async () => {
-    const stream = createStream()
-    const setup = await mount({ api: apiFor(stream) })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    stream.fail("reader exploded")
-    await waitFor(() => setup.client.connection.status() === "reconnecting")
-
-    observe("stream.reader-failure", {
-      status: setup.client.connection.status(),
-      error: setup.client.connection.error(),
-      seen: setup.seen,
-      history: history(setup.client),
-      stream: stream.snapshot(),
-    })
-
-    setup.app.renderer.destroy()
-    expect(setup.client.connection.error()).toBe("Transport")
-  })
-
-  test("records malformed SSE data after connection", async () => {
-    const stream = createStream()
-    const setup = await mount({ api: apiFor(stream) })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    stream.raw("data: not-json\n\n")
-    await waitFor(() => setup.client.connection.status() === "reconnecting")
-
-    observe("stream.malformed-data", {
-      status: setup.client.connection.status(),
-      error: setup.client.connection.error(),
-      seen: setup.seen,
-      history: history(setup.client),
-      stream: stream.snapshot(),
-    })
-
-    setup.app.renderer.destroy()
-    expect(setup.client.connection.error()).toBe("MalformedResponse")
-  })
-
-  test("records a server.connected listener failure before connected state publication", async () => {
-    const stream = createStream()
-    const setup = await mount({ api: apiFor(stream), throwOn: "server.connected" })
-
-    await waitFor(() => setup.client.connection.status() === "reconnecting")
-    observe("listener.connected-failure", {
-      status: setup.client.connection.status(),
-      error: setup.client.connection.error(),
-      seen: setup.seen,
-      history: history(setup.client),
-      stream: stream.snapshot(),
-    })
-
-    setup.app.renderer.destroy()
-    expect(history(setup.client).map((item) => item.status)).toEqual(["connecting", "connected", "disconnected"])
-  })
-
-  test("records cleanup during static reconnect backoff", async () => {
-    const stream = createStream()
-    const setup = await mount({ api: apiFor(stream) })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    stream.close()
-    await waitFor(() => setup.client.connection.status() === "reconnecting")
-    setup.app.renderer.destroy()
-    await Bun.sleep(1_050)
-
-    observe("cleanup.reconnect-backoff", {
-      status: setup.client.connection.status(),
-      history: history(setup.client),
-      stream: stream.snapshot(),
-    })
-
-    expect(stream.snapshot().requests).toBe(1)
-  })
-
-  test("records cleanup during managed-service resolution", async () => {
-    const stream = createStream()
-    let resolutionStarted = false
-    let resolutionAborted = false
-    const service: Service = {
-      reconnect(signal) {
-        resolutionStarted = true
-        return new Promise((_, reject) => {
-          signal.addEventListener(
-            "abort",
-            () => {
-              resolutionAborted = true
-              reject(signal.reason)
-            },
-            { once: true },
-          )
-        })
-      },
-      restart: () => Promise.resolve(),
-    }
-    const setup = await mount({ api: apiFor(stream), service })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    stream.close()
-    await waitFor(() => resolutionStarted)
-    setup.app.renderer.destroy()
-    await waitFor(() => resolutionAborted)
-    await Bun.sleep(20)
-
-    observe("cleanup.service-resolution", {
-      resolutionStarted,
-      resolutionAborted,
-      status: setup.client.connection.status(),
-      history: history(setup.client),
-      stream: stream.snapshot(),
-      logs: setup.logs,
-    })
-
-    expect(resolutionAborted).toBe(true)
-  })
-
-  test("records attempt reset after a stable connection", async () => {
-    const streams = [createStream(), createStream(), createStream()]
-    const apis = streams.map(apiFor)
-    let reconnects = 0
-    const service: Service = {
-      reconnect() {
-        const api = apis[Math.min(reconnects + 1, apis.length - 1)]
-        reconnects += 1
-        return Promise.resolve({ api })
-      },
-      restart: () => Promise.resolve(),
-    }
-    const setup = await mount({ api: apis[0], service })
-
-    await waitFor(() => setup.client.connection.status() === "connected")
-    streams[0].close()
-    await waitFor(() => streams[1].snapshot().requests === 1)
-    streams[1].close()
-    await waitFor(() => streams[2].snapshot().requests === 1)
-    await Bun.sleep(1_050)
-    streams[2].close()
-    await waitFor(() => reconnects === 3)
-
-    observe("reconnect.stable-reset", {
-      reconnects,
-      status: setup.client.connection.status(),
-      history: history(setup.client),
-      streams: streams.map((stream) => stream.snapshot()),
-    })
-
-    setup.app.renderer.destroy()
-    expect(history(setup.client).filter((item) => item.status === "disconnected").map((item) => item.attempt)).toEqual([
-      1, 2, 1,
-    ])
-  })
-})