fix(app): route global events without directory sentinel (#42719)

This commit is contained in:
Luke Parker
2026-08-15 16:47:37 +10:00
committed by GitHub
parent 8610d90838
commit d24a24a2ca
10 changed files with 86 additions and 75 deletions
+2 -1
View File
@@ -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)
+1 -1
View File
@@ -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) => {
+12 -21
View File
@@ -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<typeof enqueueServerEvent>[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"])
})
})
+31 -29
View File
@@ -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<ServerEvent, { type: Type }> }
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<typeof createGlobalEmitter<{ [key: string]: ServerEvent }>>
type ServerEventEmitter = ReturnType<typeof createGlobalEmitter<ServerEventMap>>
type ServerLocationEventEmitter = ReturnType<typeof createGlobalEmitter<{ [directory: string]: ServerEvent }>>
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<ServerEventMap>()
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<typeof setTimeout> | 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<SDKEventMap>()
const unsub = serverSDK.event.on(directory, (event) => {
const unsub = serverSDK.event.location.on(directory, (event) => {
emitter.emit(event.type, event)
})
onCleanup(unsub)
@@ -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<OpenCodeEvent, { type: "session.moved" }>)
expect(sessionListEventDirectories(event)).toEqual(["/source", "/destination"])
})
})
describe("canDisposeDirectory", () => {
+14 -15
View File
@@ -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<McpListOutput>
}
@@ -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 })
@@ -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)
@@ -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)
}
}
@@ -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
})
@@ -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
}