diff --git a/packages/app/src/context/notification.tsx b/packages/app/src/context/notification.tsx index 52fb6fc3292..a4f2fa21a53 100644 --- a/packages/app/src/context/notification.tsx +++ b/packages/app/src/context/notification.tsx @@ -287,7 +287,8 @@ export function createServerNotificationState(input: { sdk: ServerSDK; sync: Ser ) return - const directory = e.name + const directory = event.current?.location?.directory + if (!directory) return const time = Date.now() if (event.type === "session.execution.failed") { handleSessionError(directory, event, time) diff --git a/packages/app/src/context/permission.tsx b/packages/app/src/context/permission.tsx index a4f9f840b13..6905a16aec6 100644 --- a/packages/app/src/context/permission.tsx +++ b/packages/app/src/context/permission.tsx @@ -194,7 +194,7 @@ export function createServerPermissionState(input: { sdk: ServerSDK; sync: Serve const handlePermission = (e: PermissionEvent) => { const event = e.details if (event?.type !== "permission.asked") return - void respondPending(event.properties, e.name) + void respondPending(event.properties, event.current?.location?.directory) } const unsubscribe = input.sdk.event.listen((event) => { diff --git a/packages/app/src/context/server-sdk.test.ts b/packages/app/src/context/server-sdk.test.ts index 8fe6c99de0c..005e517b72d 100644 --- a/packages/app/src/context/server-sdk.test.ts +++ b/packages/app/src/context/server-sdk.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test } from "bun:test" import type { OpenCodeEvent } from "@opencode-ai/client/promise" -import { adaptServerEvent, coalesceServerEvents, enqueueServerEvent, resumeStreamAfterPageShow } from "./server-sdk" +import { adaptServerEvent, coalesceServerEvents, resumeStreamAfterPageShow } from "./server-sdk" describe("resumeStreamAfterPageShow", () => { test("restarts a stream only after a back-forward cache restore", () => { @@ -45,23 +45,21 @@ describe("adaptServerEvent", () => { }) describe("current event buffering", () => { - const delta = (id: string, value: string, ordinal = 0) => ({ - directory: "/repo", - payload: adaptServerEvent({ + 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), - }) + } 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]?.payload.current).toMatchObject({ id: "evt_2", data: { delta: "hello world" } }) - expect(result[0]?.payload.properties).toMatchObject({ delta: "hello world" }) + 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", () => { @@ -74,26 +72,19 @@ describe("current event buffering", () => { data: { sessionID: "ses", assistantMessageID: "msg", id, delta }, } as OpenCodeEvent) const result = coalesceServerEvents([ - { directory: "/repo", payload: current("evt_1", "call_1", "{") }, - { directory: "/repo", payload: current("evt_2", "call_1", "}") }, - { directory: "/repo", payload: current("evt_3", "call_2", "[]") }, + current("evt_1", "call_1", "{"), + current("evt_2", "call_1", "}"), + current("evt_3", "call_2", "[]"), ]) expect(result).toHaveLength(2) - expect(result[0]?.payload.current).toMatchObject({ id: "evt_2", data: { id: "call_1", delta: "{}" } }) - expect(result[1]?.payload.current).toMatchObject({ id: "evt_3", data: { id: "call_2", delta: "[]" } }) + 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.payload.current?.id)).toEqual(["evt_1", "evt_2", "evt_3"]) - }) - - test("preserves current event order when enqueuing", () => { - const events: Parameters[0] = [] - ;[delta("evt_1", "a"), delta("evt_2", "b", 1)].forEach((event) => enqueueServerEvent(events, event)) - - expect(events.map((event) => event.payload.current?.id)).toEqual(["evt_1", "evt_2"]) + 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 63942adaa99..ecd2358a4d5 100644 --- a/packages/app/src/context/server-sdk.tsx +++ b/packages/app/src/context/server-sdk.tsx @@ -12,7 +12,7 @@ import { ServerScope } from "@/utils/server-scope" import { useServer } from "./server" export type ServerEvent = Event & { id?: string; current?: OpenCodeEvent } -type QueuedServerEvent = { directory: string; payload: ServerEvent } +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" } @@ -22,22 +22,17 @@ export function adaptServerEvent(event: OpenCodeEvent): ServerEvent { return { id: event.id, type: event.type, properties: event.data, current: event } as ServerEvent } -export function enqueueServerEvent(queue: QueuedServerEvent[], event: QueuedServerEvent) { - queue.push(event) - return true -} - -export function coalesceServerEvents(events: QueuedServerEvent[]) { - const output: QueuedServerEvent[] = [] +export function coalesceServerEvents(events: ServerEvent[]) { + const output: ServerEvent[] = [] events.forEach((event) => { - const current = currentDelta(event.payload.current) + const current = currentDelta(event.current) if (current) { const previous = output[output.length - 1] - const prior = currentDelta(previous?.payload.current) + const prior = currentDelta(previous?.current) if ( previous && prior && - previous.directory === event.directory && + prior.location?.directory === current.location?.directory && currentDeltaKey(prior) === currentDeltaKey(current) ) { const fragment = currentDeltaFragment(prior) + currentDeltaFragment(current) @@ -46,13 +41,10 @@ export function coalesceServerEvents(events: QueuedServerEvent[]) { ? { ...current.data, text: fragment } : { ...current.data, delta: fragment } output[output.length - 1] = { - directory: event.directory, - payload: { - ...event.payload, - properties: data, - current: { ...current, data } as CurrentDelta, - } as ServerEvent, - } + ...event, + properties: data, + current: { ...current, data } as CurrentDelta, + } as ServerEvent return } output.push(event) @@ -89,7 +81,8 @@ export function resumeStreamAfterPageShow(event: PageTransitionEvent, start: () start() } -type ServerEventEmitter = ReturnType> +type ServerEventEmitter = ReturnType> +type ServerLocationEventEmitter = ReturnType> export type ServerConnectionStatus = "connecting" | "connected" | "reconnecting" type ServerSDKBase = { server: ServerConnection.Any @@ -104,6 +97,9 @@ type ServerSDKBase = { event: { on: ServerEventEmitter["on"] listen: ServerEventEmitter["listen"] + location: { + on: ServerLocationEventEmitter["on"] + } } } @@ -123,18 +119,16 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS })() const eventApi = createApiForServer({ server: server.http, fetch: eventFetch }) - const emitter = createGlobalEmitter<{ - [key: string]: ServerEvent - }>() + const emitter = createGlobalEmitter() + const locations = createGlobalEmitter<{ [directory: string]: ServerEvent }>() - type Queued = QueuedServerEvent const FLUSH_FRAME_MS = 16 const STREAM_YIELD_MS = 8 const CONNECT_TIMEOUT_MS = 2_000 const RECONNECT_DELAY_MS = 1_000 - let queue: Queued[] = [] - let buffer: Queued[] = [] + let queue: ServerEvent[] = [] + let buffer: ServerEvent[] = [] let timer: ReturnType | undefined let last = 0 @@ -152,7 +146,11 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS last = Date.now() const output = coalesceServerEvents(events) batch(() => { - output.forEach((event) => emitter.emit(event.directory, event.payload)) + output.forEach((event) => { + emitter.emit(event.type, event) + const directory = event.current?.location?.directory + if (directory) locations.emit(directory, event) + }) }) buffer.length = 0 @@ -165,8 +163,8 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS } function publish(event: OpenCodeEvent) { - const directory = event.location?.directory ?? "global" - if (enqueueServerEvent(queue, { directory, payload: adaptServerEvent(event) })) schedule() + queue.push(adaptServerEvent(event)) + schedule() } function wait(delay: number, signal: AbortSignal) { @@ -313,6 +311,7 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS queue = [] buffer = [] emitter.clear() + locations.clear() }) const api = createApiForServer({ server: server.http, fetch: platform.fetch }) @@ -330,6 +329,9 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS event: { on: emitter.on.bind(emitter), listen: emitter.listen.bind(emitter), + location: { + on: locations.on.bind(locations), + }, }, } } @@ -365,7 +367,7 @@ export type DirectorySDK = { function createDirSdkContext(directory: string, serverSDK: ServerSDKBase): DirectorySDK { const emitter = createGlobalEmitter() - const unsub = serverSDK.event.on(directory, (event) => { + const unsub = serverSDK.event.location.on(directory, (event) => { emitter.emit(event.type, event) }) onCleanup(unsub) diff --git a/packages/app/src/context/server-sync.test.ts b/packages/app/src/context/server-sync.test.ts index 31f3e9f6b53..df68c12ca87 100644 --- a/packages/app/src/context/server-sync.test.ts +++ b/packages/app/src/context/server-sync.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test" import type { McpListInput, McpResourceCatalogInput, + OpenCodeEvent, SessionApi, SessionInfo, SessionListInput, @@ -15,11 +16,13 @@ import { loadMcpResourcesQuery, reconcileActiveSessionStatuses, seedActiveSessionStatuses, + sessionListEventDirectories, shouldRefreshWorkspaceSessions, } from "./server-sync" import { ServerScope } from "@/utils/server-scope" import { createServerSession } from "./server-session" import type { ServerApi } from "@/utils/server" +import { adaptServerEvent } from "./server-sdk" type McpApi = ServerApi["mcp"] @@ -214,6 +217,23 @@ describe("workspace session inventory", () => { expect(shouldRefreshWorkspaceSessions(event("session.updated", "session.moved"))).toBe(true) expect(shouldRefreshWorkspaceSessions(event("message.updated"))).toBe(false) }) + + test("invalidates both locations when a session moves", () => { + const event = adaptServerEvent({ + id: "evt_moved", + created: 1, + type: "session.moved", + durable: { aggregateID: "ses_1", seq: 1, version: 1 }, + location: { directory: "/source" }, + data: { + sessionID: "ses_1", + location: { directory: "/destination" }, + projectID: "project_2", + }, + } satisfies Extract) + + expect(sessionListEventDirectories(event)).toEqual(["/source", "/destination"]) + }) }) describe("canDisposeDirectory", () => { diff --git a/packages/app/src/context/server-sync.tsx b/packages/app/src/context/server-sync.tsx index e4befa5a521..1471aa1c57c 100644 --- a/packages/app/src/context/server-sync.tsx +++ b/packages/app/src/context/server-sync.tsx @@ -88,6 +88,12 @@ const SESSION_LIST_EVENTS = new Set([ "session.usage.updated", ]) +export function sessionListEventDirectories(event: ServerEvent) { + if (!SESSION_LIST_EVENTS.has(event.current?.type ?? event.type)) return [] + const destination = event.current?.type === "session.moved" ? event.current.data.location.directory : undefined + return [...new Set([event.current?.location?.directory, destination].filter((item): item is string => !!item))] +} + type McpListApi = { readonly list: (input?: McpListInput) => Promise } @@ -554,14 +560,11 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { type: "session.updated", properties: { sessionID: info.id, info }, }) - const markSessionListChanged = (event: ServerEvent, directory: string, previousDirectory?: string) => { - if (SESSION_LIST_EVENTS.has(event.current?.type ?? event.type)) { + const markSessionListsChanged = (event: ServerEvent) => { + sessionListEventDirectories(event).forEach((directory) => { const key = directoryKey(directory) sessionRevision.set(key, (sessionRevision.get(key) ?? 0) + 1) - } - if (!previousDirectory || previousDirectory === directory) return - const key = directoryKey(previousDirectory) - sessionRevision.set(key, (sessionRevision.get(key) ?? 0) + 1) + }) } const toDirectoryEvent = (event: ServerEvent) => { if (event.current?.type === "session.created") return @@ -572,15 +575,10 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { } const unsub = serverSDK.event.listen((e) => { - const directory = e.name - const key = directoryKey(directory) const event = e.details + const directory = event.current?.location?.directory const eventType: string = event.type - const previousDirectory = - event.current?.type === "session.moved" - ? session.get(event.current.data.sessionID)?.location.directory - : undefined - markSessionListChanged(event, directory, previousDirectory) + markSessionListsChanged(event) if (event.current) session.applyV2(event.current) session.apply(event) if (event.current?.type === "session.moved") { @@ -632,9 +630,9 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { } homeSessions.refresh(event.type) catalog.handleEvent({ type: eventType, directory }) - connection.handleEvent({ type: eventType, directory }) + connection.handleEvent({ type: eventType }) - if (directory === "global") { + if (!directory) { applyGlobalEvent({ event, project: globalStore.project, @@ -647,6 +645,7 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) { return } + const key = directoryKey(directory) if (event.current?.type === "session.forked") void session .resolve(event.current.data.sessionID, { force: true }) diff --git a/packages/app/src/context/server-sync/catalog.test.ts b/packages/app/src/context/server-sync/catalog.test.ts index c653b79ceb8..82c497409ef 100644 --- a/packages/app/src/context/server-sync/catalog.test.ts +++ b/packages/app/src/context/server-sync/catalog.test.ts @@ -42,7 +42,7 @@ test("invalidates global and active catalogs after connection", async () => { load: async () => {}, }) - catalog.handleEvent({ type: "server.connected", directory: "global" }) + catalog.handleEvent({ type: "server.connected" }) await Bun.sleep(0) expect(queryClient.getQueryState(global)?.isInvalidated).toBe(true) diff --git a/packages/app/src/context/server-sync/catalog.ts b/packages/app/src/context/server-sync/catalog.ts index 02c6153e3d1..4dc4f86d44e 100644 --- a/packages/app/src/context/server-sync/catalog.ts +++ b/packages/app/src/context/server-sync/catalog.ts @@ -4,7 +4,7 @@ import { pathKey, type PathKey } from "@/utils/path-key" type CatalogEvent = { type: string - directory: string + directory?: string } export function createCatalogSync(input: { @@ -24,7 +24,7 @@ export function createCatalogSync(input: { event.type === "integration.updated" || event.type === "integration.connection.updated" ) { - void refresh(event.directory === "global" ? null : pathKey(event.directory)).catch(() => undefined) + void refresh(event.directory ? pathKey(event.directory) : null).catch(() => undefined) } } diff --git a/packages/app/src/context/server-sync/connection.test.ts b/packages/app/src/context/server-sync/connection.test.ts index cfa50e9291a..1a0b5c4bc4d 100644 --- a/packages/app/src/context/server-sync/connection.test.ts +++ b/packages/app/src/context/server-sync/connection.test.ts @@ -12,10 +12,8 @@ test("invalidates disconnected data and synchronizes after the handshake", () => connected: () => calls.push("connected"), }) - connection.handleEvent({ type: "server.connected", directory: "global" }) + connection.handleEvent({ type: "server.connected" }) expect(calls).toContain("connected") - connection.handleEvent({ type: "server.connected", directory: "/repo" }) - expect(calls.filter((call) => call === "connected")).toHaveLength(1) setStatus("connected") return dispose }) diff --git a/packages/app/src/context/server-sync/connection.ts b/packages/app/src/context/server-sync/connection.ts index b9c7b73cfa8..44190201e40 100644 --- a/packages/app/src/context/server-sync/connection.ts +++ b/packages/app/src/context/server-sync/connection.ts @@ -12,8 +12,8 @@ export function createConnectionSync(input: { }) let connectedOnce = false - function handleEvent(event: { type: string; directory: string }) { - if (event.directory !== "global" || event.type !== "server.connected") return + function handleEvent(event: { type: string }) { + if (event.type !== "server.connected") return input.connected({ reconnect: connectedOnce }) connectedOnce = true }