refactor(app): use shared client connection (#43016)

packages/app/src/context/server-sdk.test.ts

@@ -1,18 +1,6 @@
 import { describe, expect, test } from "bun:test"
 import type { OpenCodeEvent } from "@opencode-ai/client/promise"
-import { adaptServerEvent, coalesceServerEvents, resumeStreamAfterPageShow } from "./server-sdk"
-
-describe("resumeStreamAfterPageShow", () => {
-  test("restarts a stream only after a back-forward cache restore", () => {
-    let starts = 0
-    const start = () => starts++
-
-    resumeStreamAfterPageShow({ persisted: false } as PageTransitionEvent, start)
-    resumeStreamAfterPageShow({ persisted: true } as PageTransitionEvent, start)
-
-    expect(starts).toBe(1)
-  })
-})
+import { adaptServerEvent } from "./server-sdk"
 
 describe("adaptServerEvent", () => {
   test("preserves current permission requests", () => {
@@ -43,48 +31,3 @@ describe("adaptServerEvent", () => {
     })
   })
 })
-
-describe("current event buffering", () => {
-  const delta = (id: string, value: string, ordinal = 0) =>
-    adaptServerEvent({
-      id,
-      created: 1,
-      type: "session.text.delta",
-      location: { directory: "/repo" },
-      data: { sessionID: "ses", assistantMessageID: "msg", ordinal, delta: value },
-    } as OpenCodeEvent)
-
-  test("merges adjacent text deltas for the same message and ordinal", () => {
-    const result = coalesceServerEvents([delta("evt_1", "hello "), delta("evt_2", "world")])
-
-    expect(result).toHaveLength(1)
-    expect(result[0]?.current).toMatchObject({ id: "evt_2", data: { delta: "hello world" } })
-    expect(result[0]?.properties).toMatchObject({ delta: "hello world" })
-  })
-
-  test("coalesces current tool input deltas by tool ID", () => {
-    const current = (eventID: string, id: string, delta: string) =>
-      adaptServerEvent({
-        id: eventID,
-        created: 1,
-        type: "session.tool.input.delta",
-        location: { directory: "/repo" },
-        data: { sessionID: "ses", assistantMessageID: "msg", id, delta },
-      } as OpenCodeEvent)
-    const result = coalesceServerEvents([
-      current("evt_1", "call_1", "{"),
-      current("evt_2", "call_1", "}"),
-      current("evt_3", "call_2", "[]"),
-    ])
-
-    expect(result).toHaveLength(2)
-    expect(result[0]?.current).toMatchObject({ id: "evt_2", data: { id: "call_1", delta: "{}" } })
-    expect(result[1]?.current).toMatchObject({ id: "evt_3", data: { id: "call_2", delta: "[]" } })
-  })
-
-  test("preserves boundaries between distinct delta streams", () => {
-    const events = [delta("evt_1", "a"), delta("evt_2", "b", 1), delta("evt_3", "c")]
-
-    expect(coalesceServerEvents(events).map((event) => event.current?.id)).toEqual(["evt_1", "evt_2", "evt_3"])
-  })
-})

packages/app/src/context/server-sdk.tsx

@@ -1,9 +1,8 @@
 import type { OpenCodeEvent } from "@opencode-ai/client/promise"
+import { createClientConnection, type ClientConnectionStatus } from "@opencode-ai/client/solid"
 import type { Event } from "@/types"
 import { createGlobalEmitter } from "@solid-primitives/event-bus"
-import { makeEventListener } from "@solid-primitives/event-listener"
-import { type Accessor, batch, onCleanup, onMount } from "solid-js"
-import { createStore } from "solid-js/store"
+import { type Accessor, onCleanup } from "solid-js"
 import { createApiForServer, type ServerApi } from "@/utils/server"
 import { usePlatform } from "./platform"
 import { ServerConnection } from "./servers"
@@ -12,78 +11,15 @@ import { ServerScope } from "@/utils/server-scope"
 import { useServer } from "./server"
 
 export type ServerEvent = Event & { id?: string; current?: OpenCodeEvent }
-type ServerEventMap = { [Type in ServerEvent["type"]]: Extract<ServerEvent, { type: Type }> }
-type CurrentDelta = Extract<
-  OpenCodeEvent,
-  { type: "session.text.delta" | "session.reasoning.delta" | "session.tool.input.delta" | "session.compaction.delta" }
->
 
 export function adaptServerEvent(event: OpenCodeEvent): ServerEvent {
   return { id: event.id, type: event.type, properties: event.data, current: event } as ServerEvent
 }
 
-export function coalesceServerEvents(events: ServerEvent[]) {
-  const output: ServerEvent[] = []
-  events.forEach((event) => {
-    const current = currentDelta(event.current)
-    if (current) {
-      const previous = output[output.length - 1]
-      const prior = currentDelta(previous?.current)
-      if (
-        previous &&
-        prior &&
-        prior.location?.directory === current.location?.directory &&
-        currentDeltaKey(prior) === currentDeltaKey(current)
-      ) {
-        const fragment = currentDeltaFragment(prior) + currentDeltaFragment(current)
-        const data =
-          current.type === "session.compaction.delta"
-            ? { ...current.data, text: fragment }
-            : { ...current.data, delta: fragment }
-        output[output.length - 1] = {
-          ...event,
-          properties: data,
-          current: { ...current, data } as CurrentDelta,
-        } as ServerEvent
-        return
-      }
-      output.push(event)
-      return
-    }
-    output.push(event)
-  })
-  return output
-}
-
-function currentDelta(event: OpenCodeEvent | undefined): CurrentDelta | undefined {
-  if (
-    event?.type === "session.text.delta" ||
-    event?.type === "session.reasoning.delta" ||
-    event?.type === "session.tool.input.delta" ||
-    event?.type === "session.compaction.delta"
-  )
-    return event
-}
-
-function currentDeltaKey(event: CurrentDelta) {
-  if (event.type === "session.tool.input.delta")
-    return `${event.type}:${event.data.sessionID}:${event.data.assistantMessageID}:${event.data.id}`
-  if (event.type === "session.compaction.delta") return `${event.type}:${event.data.sessionID}`
-  return `${event.type}:${event.data.sessionID}:${event.data.assistantMessageID}:${event.data.ordinal}`
-}
-
-function currentDeltaFragment(event: CurrentDelta) {
-  return event.type === "session.compaction.delta" ? event.data.text : event.data.delta
-}
-
-export function resumeStreamAfterPageShow(event: PageTransitionEvent, start: () => unknown) {
-  if (!event.persisted) return
-  start()
-}
-
+type ServerEventMap = { [Type in ServerEvent["type"]]: Extract<ServerEvent, { type: Type }> }
 type ServerEventEmitter = ReturnType<typeof createGlobalEmitter<ServerEventMap>>
 type ServerLocationEventEmitter = ReturnType<typeof createGlobalEmitter<{ [directory: string]: ServerEvent }>>
-export type ServerConnectionStatus = "connecting" | "connected" | "reconnecting"
+export type ServerConnectionStatus = ClientConnectionStatus
 type ServerSDKBase = {
   server: ServerConnection.Any
   scope: ServerScope
@@ -105,227 +41,38 @@ type ServerSDKBase = {
 
 function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerScope): ServerSDKBase {
   const platform = usePlatform()
-  const abort = new AbortController()
-
-  const eventFetch = (() => {
-    if (!platform.fetch || !server) return
-    try {
-      const url = new URL(server.http.url)
-      const loopback = url.hostname === "localhost" || url.hostname === "127.0.0.1" || url.hostname === "::1"
-      if (url.protocol === "http:" && !loopback) return platform.fetch
-    } catch {
-      return
-    }
-  })()
-
-  const eventApi = createApiForServer({ server: server.http, fetch: eventFetch })
+  const api = createApiForServer({ server: server.http, fetch: platform.fetch })
   const emitter = createGlobalEmitter<ServerEventMap>()
   const locations = createGlobalEmitter<{ [directory: string]: ServerEvent }>()
 
-  const FLUSH_FRAME_MS = 16
-  const STREAM_YIELD_MS = 8
-  const CONNECT_TIMEOUT_MS = 2_000
-  const RECONNECT_DELAY_MS = 1_000
-
-  let queue: ServerEvent[] = []
-  let buffer: ServerEvent[] = []
-  let timer: ReturnType<typeof setTimeout> | undefined
-  let last = 0
-
-  function flush() {
-    if (timer) clearTimeout(timer)
-    timer = undefined
-
-    if (queue.length === 0) return
-
-    const events = queue
-    queue = buffer
-    buffer = events
-    queue.length = 0
-
-    last = Date.now()
-    const output = coalesceServerEvents(events)
-    batch(() => {
-      output.forEach((event) => {
-        emitter.emit(event.type, event)
-        const directory = event.current?.location?.directory
-        if (directory) locations.emit(directory, event)
-      })
-    })
-
-    buffer.length = 0
-  }
-
-  function schedule() {
-    if (timer) return
-    const elapsed = Date.now() - last
-    timer = setTimeout(flush, Math.max(0, FLUSH_FRAME_MS - elapsed))
-  }
-
-  function publish(event: OpenCodeEvent) {
-    queue.push(adaptServerEvent(event))
-    schedule()
-  }
-
-  function wait(delay: number, signal: AbortSignal) {
-    return new Promise<void>((resolve) => {
-      const timer = setTimeout(done, delay)
-      signal.addEventListener("abort", done, { once: true })
-      function done() {
-        clearTimeout(timer)
-        signal.removeEventListener("abort", done)
-        resolve()
-      }
-    })
-  }
-  let attempt: AbortController | undefined
-  let run: Promise<void> | undefined
-  let started = false
-  let generation = 0
-  const [connection, setConnection] = createStore<{
-    status: ServerConnectionStatus
-    attempt: number
-    error?: string
-  }>({ status: "connecting", attempt: 0 })
-
-  async function connect(signal: AbortSignal): Promise<{ error: unknown; connectedAt: number | undefined }> {
-    let connectedAt: number | undefined
-
-    // Bound the initial handshake and tie this request to the stream lifetime.
-    const request = new AbortController()
-    const cancel = () => request.abort(signal.reason)
-    const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), CONNECT_TIMEOUT_MS)
-    signal.addEventListener("abort", cancel, { once: true })
-
-    try {
-      // Open the event stream and validate its initial handshake.
-      const iterator = eventApi.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]()
-      const first = await iterator.next()
-
-      if (signal.aborted) return { error: undefined, connectedAt }
-      if (first.done) {
-        const error =
-          request.signal.reason instanceof Error ? request.signal.reason : new Error("Event stream disconnected")
-        return { error, connectedAt }
-      }
-      if (first.value.type !== "server.connected")
-        return { error: new Error("Event stream did not start with server.connected"), connectedAt }
-
-      // Publish the connected state before forwarding live events.
-      clearTimeout(timeout)
-      publish(first.value)
-      connectedAt = Date.now()
-      setConnection({ status: "connected", attempt: 0, error: undefined })
-
-      // Forward events until the stream closes or this connection is cancelled.
-      let yielded = Date.now()
-      while (!signal.aborted) {
-        const event = await iterator.next()
-        if (signal.aborted) return { error: undefined, connectedAt }
-        if (event.done) return { error: new Error("Event stream disconnected"), connectedAt }
-        publish(event.value)
-        if (Date.now() - yielded < STREAM_YIELD_MS) continue
-        yielded = Date.now()
-        await wait(0, signal)
-      }
-      return { error: undefined, connectedAt }
-    } catch (error) {
-      return { error, connectedAt }
-    } finally {
-      request.abort()
-      clearTimeout(timeout)
-      signal.removeEventListener("abort", cancel)
-    }
-  }
-
-  async function runStream(active: number) {
-    let retries = 0
-    // oxlint-disable-next-line no-unmodified-loop-condition -- stop() changes the lifecycle flags and aborts the active request
-    while (!abort.signal.aborted && started && generation === active) {
-      setConnection({ status: retries === 0 ? "connecting" : "reconnecting", attempt: retries, error: undefined })
-      const controller = new AbortController()
-      attempt = controller
-      const onAbort = () => controller.abort()
-      abort.signal.addEventListener("abort", onAbort)
-      const result = await connect(controller.signal)
-      abort.signal.removeEventListener("abort", onAbort)
-
-      if (abort.signal.aborted || !started || generation !== active) {
-        if (attempt === controller) attempt = undefined
-        return
-      }
-      if (result.connectedAt !== undefined && Date.now() - result.connectedAt >= 1_000) retries = 0
-      retries += 1
-      const message =
-        result.error === undefined
-          ? undefined
-          : result.error instanceof Error
-            ? result.error.message
-            : String(result.error)
-      console.info("[global-sdk] event stream disconnected", {
-        url: server.http.url,
-        fetch: eventFetch ? "platform" : "webview",
-        attempt: retries,
-        error: message,
-      })
-      setConnection({ status: "reconnecting", attempt: retries, error: message })
-      await wait(RECONNECT_DELAY_MS, controller.signal)
-      if (attempt === controller) attempt = undefined
-    }
-  }
-
-  function start() {
-    if (started) return run
-    started = true
-    const active = ++generation
-    const previous = run
-    const current = (async () => {
-      if (previous) await previous
-      await runStream(active)
-    })().finally(() => {
-      if (run !== current) return
-      run = undefined
-      flush()
-    })
-    run = current
-    return run
-  }
-
-  function stop() {
-    started = false
-    generation++
-    attempt?.abort()
-  }
-
-  onMount(() => {
-    makeEventListener(window, "pagehide", stop)
-    makeEventListener(window, "pageshow", (event) => resumeStreamAfterPageShow(event, start))
-    void start()
+  const connection = createClientConnection(api, {
+    flushInterval: 16,
+    pageLifecycle: true,
+    onEvent(event) {
+      const adapted = adaptServerEvent(event)
+      emitter.emit(adapted.type, adapted)
+      const directory = event.location?.directory
+      if (directory) locations.emit(directory, adapted)
+    },
+    log: {
+      info(message, data) {
+        if (message !== "event stream disconnected") return
+        console.info("[global-sdk] event stream disconnected", { url: server.http.url, ...data })
+      },
+    },
   })
 
   onCleanup(() => {
-    stop()
-    abort.abort()
-    if (timer) clearTimeout(timer)
-    timer = undefined
-    queue = []
-    buffer = []
     emitter.clear()
     locations.clear()
   })
 
-  const api = createApiForServer({ server: server.http, fetch: platform.fetch })
-
   return {
     server,
     scope,
     url: server.http.url,
     api,
-    connection: {
-      status: () => connection.status,
-      attempt: () => connection.attempt,
-      error: () => connection.error,
-    },
+    connection,
     event: {
       on: emitter.on.bind(emitter),
       listen: emitter.listen.bind(emitter),