From 0e022036fb3819b60349a93bd306ae5dcbd89996 Mon Sep 17 00:00:00 2001 From: Brendan Allan <14191578+Brendonovich@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:53:03 +0800 Subject: [PATCH] refactor(app): use shared client connection (#43016) --- packages/app/src/context/server-sdk.test.ts | 59 +--- packages/app/src/context/server-sdk.tsx | 295 ++------------------ 2 files changed, 22 insertions(+), 332 deletions(-) diff --git a/packages/app/src/context/server-sdk.test.ts b/packages/app/src/context/server-sdk.test.ts index 005e517b72..106cf91e2a 100644 --- a/packages/app/src/context/server-sdk.test.ts +++ b/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"]) - }) -}) diff --git a/packages/app/src/context/server-sdk.tsx b/packages/app/src/context/server-sdk.tsx index ecd2358a4d..0c533fee9a 100644 --- a/packages/app/src/context/server-sdk.tsx +++ b/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 } -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 } type ServerEventEmitter = ReturnType> type ServerLocationEventEmitter = ReturnType> -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() 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 | 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((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 | 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),