diff --git a/packages/cli/src/mini/noninteractive.ts b/packages/cli/src/mini/noninteractive.ts index 0ea58394a8fe..4dd366a582fe 100644 --- a/packages/cli/src/mini/noninteractive.ts +++ b/packages/cli/src/mini/noninteractive.ts @@ -55,7 +55,7 @@ const GLOBAL_FORM_SESSION_ID = "global" export async function runNonInteractivePrompt(input: Input) { const controller = new AbortController() - const stream = input.client.event.subscribe({ signal: controller.signal })[Symbol.asyncIterator]() + const stream = input.client.event.subscribe(undefined, { signal: controller.signal })[Symbol.asyncIterator]() const connected = await stream.next() if (connected.done) throw new Error("Event stream disconnected before prompt admission") diff --git a/packages/cli/src/mini/stream-v2.transport.ts b/packages/cli/src/mini/stream-v2.transport.ts index 057801d2cd83..ea34a0cc791b 100644 --- a/packages/cli/src/mini/stream-v2.transport.ts +++ b/packages/cli/src/mini/stream-v2.transport.ts @@ -832,7 +832,7 @@ export async function createSessionTransport(input: StreamInput): Promise connection.abort() controller.signal.addEventListener("abort", abortConnection, { once: true }) - const stream = input.sdk.event.subscribe({ signal: connection.signal })[Symbol.asyncIterator]() + const stream = input.sdk.event.subscribe(undefined, { signal: connection.signal })[Symbol.asyncIterator]() try { const first = await stream.next() if (first.done || first.value.type !== "server.connected") throw new Error("Event stream disconnected") diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 27da77289432..f5129de8ee72 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -767,8 +767,13 @@ export interface SkillApi { readonly list: SkillListOperation } +type Endpoint19_0Request = Parameters[0] +export type Endpoint19_0Input = { + readonly location?: Endpoint19_0Request["query"]["location"] + readonly session?: Endpoint19_0Request["query"]["session"] +} export type Endpoint19_0Output = StreamValue>> -export type EventSubscribeOperation = () => Stream.Stream +export type EventSubscribeOperation = (input?: Endpoint19_0Input) => Stream.Stream export interface EventApi { readonly subscribe: EventSubscribeOperation @@ -992,8 +997,12 @@ export type Endpoint26_1Input = { readonly location?: Endpoint26_1Request["query export type Endpoint26_1Output = EffectValue> export type DebugLocationEvictOperation = (input?: Endpoint26_1Input) => Effect.Effect +export type Endpoint26_2Output = EffectValue> +export type DebugEventFeedOperation = () => Effect.Effect + export interface DebugApi { readonly location: { readonly list: DebugLocationListOperation; readonly evict: DebugLocationEvictOperation } + readonly "event-feed": DebugEventFeedOperation } export interface AppApi { diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index c7a2f860353b..ebc79d72823e 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -908,9 +908,14 @@ const Endpoint18_0 = (raw: RawClient["server.skill"]) => (input?: Endpoint18_0In const adaptGroup18 = (raw: RawClient["server.skill"]) => ({ list: Endpoint18_0(raw) }) -const Endpoint19_0 = (raw: RawClient["server.event"]) => () => +type Endpoint19_0Request = Parameters[0] +type Endpoint19_0Input = { + readonly location?: Endpoint19_0Request["query"]["location"] + readonly session?: Endpoint19_0Request["query"]["session"] +} +const Endpoint19_0 = (raw: RawClient["server.event"]) => (input?: Endpoint19_0Input) => Stream.unwrap( - raw["event.subscribe"]({}).pipe( + raw["event.subscribe"]({ query: { location: input?.["location"], session: input?.["session"] } }).pipe( Effect.mapError(mapClientError), Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))), ), @@ -1181,8 +1186,12 @@ type Endpoint26_1Input = { readonly location?: Endpoint26_1Request["query"]["loc const Endpoint26_1 = (raw: RawClient["server.debug"]) => (input?: Endpoint26_1Input) => raw["debug.location.evict"]({ query: { location: input?.["location"] } }).pipe(Effect.mapError(mapClientError)) +const Endpoint26_2 = (raw: RawClient["server.debug"]) => () => + raw["debug.event-feed"]({}).pipe(Effect.mapError(mapClientError)) + const adaptGroup26 = (raw: RawClient["server.debug"]) => ({ location: { list: Endpoint26_0(raw), evict: Endpoint26_1(raw) }, + "event-feed": Endpoint26_2(raw), }) const adaptClient = (raw: RawClient) => ({ diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 3783a5e5c011..2696350aa614 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -151,6 +151,7 @@ import type { CommandListOutput, SkillListInput, SkillListOutput, + EventSubscribeInput, EventSubscribeOutput, PtyListInput, PtyListOutput, @@ -197,6 +198,7 @@ import type { DebugLocationListOutput, DebugLocationEvictInput, DebugLocationEvictOutput, + DebugEventFeedOutput, } from "./types" import { ClientError } from "./client-error" @@ -292,6 +294,12 @@ export function make(options: ClientOptions) { const reader = response.body.getReader() const decoder = new TextDecoder() let buffer = "" + const signal = requestOptions?.signal + const onAbort = () => { + void reader.cancel(signal?.reason).catch(() => undefined) + } + if (signal?.aborted) onAbort() + else signal?.addEventListener("abort", onAbort, { once: true }) try { while (true) { let next @@ -327,10 +335,16 @@ export function make(options: ClientOptions) { if (next.done) return } } finally { + signal?.removeEventListener("abort", onAbort) try { await reader.cancel() } catch {} - reader.releaseLock() + try { + reader.releaseLock() + } catch {} + try { + await response.body?.cancel() + } catch {} } }, }) @@ -1350,9 +1364,16 @@ export function make(options: ClientOptions) { ), }, event: { - subscribe: (requestOptions?: RequestOptions): AsyncIterable => + subscribe: (input?: EventSubscribeInput, requestOptions?: RequestOptions): AsyncIterable => sse( - { method: "GET", path: `/api/event`, successStatus: 200, declaredStatuses: [401, 400], empty: false }, + { + method: "GET", + path: `/api/event`, + query: { location: input?.["location"], session: input?.["session"] }, + successStatus: 200, + declaredStatuses: [401, 400], + empty: false, + }, requestOptions, ), }, @@ -1663,6 +1684,17 @@ export function make(options: ClientOptions) { requestOptions, ), }, + "event-feed": (requestOptions?: RequestOptions) => + request( + { + method: "GET", + path: `/api/debug/event-feed`, + successStatus: 200, + declaredStatuses: [401, 400], + empty: false, + }, + requestOptions, + ), }, } } diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index a6f20cf2f985..3f3892f12126 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -4542,6 +4542,17 @@ export type SkillListOutput = { data: Array } +export type EventSubscribeInput = { + readonly location?: { + readonly location?: { readonly directory: string; readonly workspace?: string | undefined } | undefined + readonly session?: string | ReadonlyArray | ReadonlyArray | undefined + }["location"] + readonly session?: { + readonly location?: { readonly directory: string; readonly workspace?: string | undefined } | undefined + readonly session?: string | ReadonlyArray | ReadonlyArray | undefined + }["session"] +} + export type EventSubscribeOutput = V2Event export type PtyListInput = { @@ -4859,3 +4870,12 @@ export type DebugLocationEvictInput = { } export type DebugLocationEvictOutput = void + +export type DebugEventFeedOutput = { + active: number + opens: number + closes: number + serializedEvents: number + serializedBytes: number + overflows: number +} diff --git a/packages/httpapi-codegen/src/index.ts b/packages/httpapi-codegen/src/index.ts index 88204e771a31..494d214e9f54 100644 --- a/packages/httpapi-codegen/src/index.ts +++ b/packages/httpapi-codegen/src/index.ts @@ -722,7 +722,7 @@ function renderPromiseClient(groups: ReadonlyArray) { if (group.endpoints[0]?.topLevel) return fields return `${JSON.stringify(group.identifier)}: { ${fields} }` }) - return `import type { ${imports.join(", ")} } from "./types"\nimport { ClientError } from "./client-error"\n\nexport interface ClientOptions {\n readonly baseUrl: string\n readonly fetch?: typeof globalThis.fetch\n readonly headers?: RequestInit["headers"]\n}\n\nexport interface RequestOptions {\n readonly signal?: AbortSignal\n readonly headers?: RequestInit["headers"]\n}\n\ninterface RequestDescriptor {\n readonly method: string\n readonly path: string\n readonly query?: Record\n readonly headers?: Record\n readonly body?: unknown\n readonly successStatus: number\n readonly declaredStatuses: ReadonlyArray\n readonly empty: boolean\n}\n\nconst maxSseEventBytes = 16 * 1024 * 1024\n\nexport function make(options: ClientOptions) {\n const fetch = options.fetch ?? globalThis.fetch\n\n const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n const url = new URL(descriptor.path, options.baseUrl)\n for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)\n const headers = new Headers(options.headers)\n for (const [key, value] of Object.entries(descriptor.headers ?? {})) {\n if (value !== undefined && value !== null) headers.set(key, String(value))\n }\n for (const [key, value] of new Headers(requestOptions?.headers)) headers.set(key, value)\n if (descriptor.body !== undefined && !headers.has("content-type")) headers.set("content-type", "application/json")\n return {\n url,\n init: {\n method: descriptor.method,\n signal: requestOptions?.signal,\n headers,\n body: descriptor.body === undefined ? undefined : JSON.stringify(descriptor.body),\n } satisfies RequestInit,\n }\n }\n\n const execute = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n try {\n const prepared = prepare(descriptor, requestOptions)\n return await fetch(prepared.url, prepared.init)\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n }\n\n const responseError = async (response: Response, descriptor: RequestDescriptor): Promise => {\n if (descriptor.declaredStatuses.includes(response.status)) throw await json(response)\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnexpectedStatus", { cause: { status: response.status } })\n }\n\n const request = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions): Promise => {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) return responseError(response, descriptor)\n if (descriptor.empty) {\n try {\n await response.body?.cancel()\n } catch {}\n return undefined as A\n }\n return await json(response) as A\n }\n\n const sse = (descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable => ({\n async *[Symbol.asyncIterator]() {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) await responseError(response, descriptor)\n if (!isContentType(response, "text/event-stream")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n if (response.body === null) throw new ClientError("MalformedResponse")\n const reader = response.body.getReader()\n const decoder = new TextDecoder()\n let buffer = ""\n try {\n while (true) {\n let next: ReadableStreamReadResult\n try {\n next = await reader.read()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n buffer += decoder.decode(next.value, { stream: !next.done })\n if (buffer.length > maxSseEventBytes) throw new ClientError("SseEventTooLarge")\n const trailingCarriageReturn = !next.done && buffer.endsWith("\\r")\n if (trailingCarriageReturn) buffer = buffer.slice(0, -1)\n buffer = buffer.replaceAll("\\r\\n", "\\n").replaceAll("\\r", "\\n")\n if (trailingCarriageReturn) buffer += "\\r"\n if (next.done && buffer !== "") buffer += "\\n\\n"\n let boundary = buffer.indexOf("\\n\\n")\n while (boundary >= 0) {\n const block = buffer.slice(0, boundary)\n buffer = buffer.slice(boundary + 2)\n const data = block.split("\\n").flatMap((line) => line.startsWith("data:") ? [line.slice(5).trimStart()] : []).join("\\n")\n if (data !== "") {\n try {\n yield JSON.parse(data) as A\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n }\n boundary = buffer.indexOf("\\n\\n")\n }\n if (next.done) return\n }\n } finally {\n try {\n await reader.cancel()\n } catch {}\n reader.releaseLock()\n }\n },\n })\n\n return { ${fields.join(", ")} }\n}\n\nfunction appendQuery(params: URLSearchParams, key: string, value: unknown): void {\n if (value === undefined) return\n if (value === null) {\n params.append(key, "null")\n return\n }\n if (Array.isArray(value)) {\n for (const item of value) appendQuery(params, key, item)\n return\n }\n if (typeof value === "object") {\n for (const [child, item] of Object.entries(value)) appendQuery(params, \`\${key}[\${child}]\`, item)\n return\n }\n params.append(key, String(value))\n}\n\nasync function json(response: Response): Promise {\n if (!isContentType(response, "application/json") && !response.headers.get("content-type")?.includes("+json")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n let text: string\n try {\n text = await response.text()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n if (text === "") throw new ClientError("MalformedResponse")\n try {\n return JSON.parse(text)\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n}\n\nfunction isContentType(response: Response, expected: string) {\n return response.headers.get("content-type")?.split(";", 1)[0]?.trim().toLowerCase() === expected\n}\n` + return `import type { ${imports.join(", ")} } from "./types"\nimport { ClientError } from "./client-error"\n\nexport interface ClientOptions {\n readonly baseUrl: string\n readonly fetch?: typeof globalThis.fetch\n readonly headers?: RequestInit["headers"]\n}\n\nexport interface RequestOptions {\n readonly signal?: AbortSignal\n readonly headers?: RequestInit["headers"]\n}\n\ninterface RequestDescriptor {\n readonly method: string\n readonly path: string\n readonly query?: Record\n readonly headers?: Record\n readonly body?: unknown\n readonly successStatus: number\n readonly declaredStatuses: ReadonlyArray\n readonly empty: boolean\n}\n\nconst maxSseEventBytes = 16 * 1024 * 1024\n\nexport function make(options: ClientOptions) {\n const fetch = options.fetch ?? globalThis.fetch\n\n const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n const url = new URL(descriptor.path, options.baseUrl)\n for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)\n const headers = new Headers(options.headers)\n for (const [key, value] of Object.entries(descriptor.headers ?? {})) {\n if (value !== undefined && value !== null) headers.set(key, String(value))\n }\n for (const [key, value] of new Headers(requestOptions?.headers)) headers.set(key, value)\n if (descriptor.body !== undefined && !headers.has("content-type")) headers.set("content-type", "application/json")\n return {\n url,\n init: {\n method: descriptor.method,\n signal: requestOptions?.signal,\n headers,\n body: descriptor.body === undefined ? undefined : JSON.stringify(descriptor.body),\n } satisfies RequestInit,\n }\n }\n\n const execute = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n try {\n const prepared = prepare(descriptor, requestOptions)\n return await fetch(prepared.url, prepared.init)\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n }\n\n const responseError = async (response: Response, descriptor: RequestDescriptor): Promise => {\n if (descriptor.declaredStatuses.includes(response.status)) throw await json(response)\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnexpectedStatus", { cause: { status: response.status } })\n }\n\n const request = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions): Promise => {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) return responseError(response, descriptor)\n if (descriptor.empty) {\n try {\n await response.body?.cancel()\n } catch {}\n return undefined as A\n }\n return await json(response) as A\n }\n\n const sse = (descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable => ({\n async *[Symbol.asyncIterator]() {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) await responseError(response, descriptor)\n if (!isContentType(response, "text/event-stream")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n if (response.body === null) throw new ClientError("MalformedResponse")\n const reader = response.body.getReader()\n const decoder = new TextDecoder()\n let buffer = ""\n const signal = requestOptions?.signal\n const onAbort = () => {\n void reader.cancel(signal?.reason).catch(() => undefined)\n }\n if (signal?.aborted) onAbort()\n else signal?.addEventListener("abort", onAbort, { once: true })\n try {\n while (true) {\n let next: ReadableStreamReadResult\n try {\n next = await reader.read()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n buffer += decoder.decode(next.value, { stream: !next.done })\n if (buffer.length > maxSseEventBytes) throw new ClientError("SseEventTooLarge")\n const trailingCarriageReturn = !next.done && buffer.endsWith("\\r")\n if (trailingCarriageReturn) buffer = buffer.slice(0, -1)\n buffer = buffer.replaceAll("\\r\\n", "\\n").replaceAll("\\r", "\\n")\n if (trailingCarriageReturn) buffer += "\\r"\n if (next.done && buffer !== "") buffer += "\\n\\n"\n let boundary = buffer.indexOf("\\n\\n")\n while (boundary >= 0) {\n const block = buffer.slice(0, boundary)\n buffer = buffer.slice(boundary + 2)\n const data = block.split("\\n").flatMap((line) => line.startsWith("data:") ? [line.slice(5).trimStart()] : []).join("\\n")\n if (data !== "") {\n try {\n yield JSON.parse(data) as A\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n }\n boundary = buffer.indexOf("\\n\\n")\n }\n if (next.done) return\n }\n } finally {\n signal?.removeEventListener("abort", onAbort)\n try {\n await reader.cancel()\n } catch {}\n try {\n reader.releaseLock()\n } catch {}\n try {\n await response.body?.cancel()\n } catch {}\n }\n },\n })\n\n return { ${fields.join(", ")} }\n}\n\nfunction appendQuery(params: URLSearchParams, key: string, value: unknown): void {\n if (value === undefined) return\n if (value === null) {\n params.append(key, "null")\n return\n }\n if (Array.isArray(value)) {\n for (const item of value) appendQuery(params, key, item)\n return\n }\n if (typeof value === "object") {\n for (const [child, item] of Object.entries(value)) appendQuery(params, \`\${key}[\${child}]\`, item)\n return\n }\n params.append(key, String(value))\n}\n\nasync function json(response: Response): Promise {\n if (!isContentType(response, "application/json") && !response.headers.get("content-type")?.includes("+json")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n let text: string\n try {\n text = await response.text()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n if (text === "") throw new ClientError("MalformedResponse")\n try {\n return JSON.parse(text)\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n}\n\nfunction isContentType(response: Response, expected: string) {\n return response.headers.get("content-type")?.split(";", 1)[0]?.trim().toLowerCase() === expected\n}\n` } function promiseTypePrefix(group: string, path: ReadonlyArray) { diff --git a/packages/protocol/src/groups/debug.ts b/packages/protocol/src/groups/debug.ts index fd7cd9def513..118cb6c687ae 100644 --- a/packages/protocol/src/groups/debug.ts +++ b/packages/protocol/src/groups/debug.ts @@ -3,6 +3,15 @@ import { Schema } from "effect" import { HttpApiEndpoint, HttpApiGroup, HttpApiSchema, OpenApi } from "effect/unstable/httpapi" import { LocationQuery, locationQueryOpenApi } from "./location.js" +export const EventFeedDiagnostics = Schema.Struct({ + active: Schema.Int, + opens: Schema.Int, + closes: Schema.Int, + serializedEvents: Schema.Int, + serializedBytes: Schema.Int, + overflows: Schema.Int, +}) + export const DebugGroup = HttpApiGroup.make("server.debug") .add( HttpApiEndpoint.get("debug.location", "/api/debug/location", { @@ -29,4 +38,16 @@ export const DebugGroup = HttpApiGroup.make("server.debug") }), ), ) + .add( + HttpApiEndpoint.get("debug.event-feed", "/api/debug/event-feed", { + success: EventFeedDiagnostics, + }).annotateMerge( + OpenApi.annotations({ + identifier: "v2.debug.event-feed", + summary: "Event feed diagnostics", + description: + "Content-free counters for active SSE subscribers, opens/closes, serialized event volume, and overflows.", + }), + ), + ) .annotateMerge(OpenApi.annotations({ title: "debug" })) diff --git a/packages/protocol/src/groups/event.ts b/packages/protocol/src/groups/event.ts index 242aa745ef53..87fec115f8ce 100644 --- a/packages/protocol/src/groups/event.ts +++ b/packages/protocol/src/groups/event.ts @@ -4,6 +4,20 @@ import { Location } from "@opencode-ai/schema/location" import type { Definition } from "@opencode-ai/schema/event" import { Schema } from "effect" import { HttpApiEndpoint, HttpApiGroup, HttpApiSchema, OpenApi } from "effect/unstable/httpapi" +import { locationQueryOpenApi } from "./location.js" + +export const sessionIDOf = Event.sessionIDOf + +export const EventSubscribeQuery = Schema.Struct({ + location: Schema.optional( + Schema.Struct({ + /** Required when `location` is present; workspace alone is not a valid scope. */ + directory: Schema.String, + workspace: Schema.optional(Schema.String), + }), + ), + session: Schema.optional(Schema.Union([Schema.String, Schema.Array(Schema.String)])), +}).annotate({ identifier: "EventSubscribeQuery" }) const fields = { id: Event.ID, @@ -32,15 +46,18 @@ const make = >(definitions: group: HttpApiGroup.make("server.event") .add( HttpApiEndpoint.get("event.subscribe", "/api/event", { + query: EventSubscribeQuery, success: HttpApiSchema.StreamSse({ data: EventSchema }), - }).annotateMerge( - OpenApi.annotations({ - identifier: "v2.event.subscribe", - summary: "Subscribe to events", - description: - "Subscribe to native event payloads for the server. Volatile by contract: a slow consumer overflows and fails the stream, and events during disconnection are missed.", - }), - ), + }) + .annotateMerge(locationQueryOpenApi) + .annotateMerge( + OpenApi.annotations({ + identifier: "v2.event.subscribe", + summary: "Subscribe to events", + description: + "Subscribe to native event payloads for the server. Omit query params for the global public feed. Pass location to narrow by directory/workspace, and optional repeated session IDs to further narrow session-scoped events. Volatile by contract: a slow consumer overflows and fails the stream, and events during disconnection are missed.", + }), + ), ) .annotateMerge(OpenApi.annotations({ title: "event", description: "Experimental event stream routes." })), } diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index e4f2f9b8139f..0211e19d2210 100644 --- a/packages/schema/src/event.ts +++ b/packages/schema/src/event.ts @@ -1,6 +1,6 @@ export * as Event from "./event.js" -import { Schema, SchemaTransformation } from "effect" +import { Option, Schema, SchemaTransformation } from "effect" import { optional } from "./schema.js" import { ascending } from "./identifier.js" import { Location } from "./location.js" @@ -196,3 +196,15 @@ function readonlyMap(map: Map): ReadonlyMap }) return result } + +const SessionData = Schema.Struct({ sessionID: Schema.String }) +const decodeSessionData = Schema.decodeUnknownOption(SessionData) + +/** Session identity from a durable aggregate or an ephemeral `data.sessionID` field. */ +export function sessionIDOf(event: { + readonly durable?: { readonly aggregateID: string } + readonly data?: unknown +}): string | undefined { + if (event.durable?.aggregateID) return event.durable.aggregateID + return Option.getOrUndefined(decodeSessionData(event.data))?.sessionID +} diff --git a/packages/schema/test/event.test.ts b/packages/schema/test/event.test.ts index 404b49aa743c..cfddf15fb954 100644 --- a/packages/schema/test/event.test.ts +++ b/packages/schema/test/event.test.ts @@ -49,4 +49,14 @@ describe("public event schemas", () => { aggregateID: "ses_test", }) }) + + test("sessionIDOf reads durable aggregate or ephemeral data.sessionID", () => { + expect(Event.sessionIDOf({ durable: { aggregateID: "ses_durable" }, data: { sessionID: "ses_other" } })).toBe( + "ses_durable", + ) + expect(Event.sessionIDOf({ data: { sessionID: "ses_ephemeral" } })).toBe("ses_ephemeral") + expect(Event.sessionIDOf({ data: {} })).toBeUndefined() + expect(Event.sessionIDOf({ data: { sessionID: 1 } })).toBeUndefined() + expect(Event.sessionIDOf({ data: null })).toBeUndefined() + }) }) diff --git a/packages/server/src/event-feed.ts b/packages/server/src/event-feed.ts index 7b668df5b42a..5302c1820c82 100644 --- a/packages/server/src/event-feed.ts +++ b/packages/server/src/event-feed.ts @@ -1,11 +1,24 @@ export * as EventFeed from "./event-feed" import { EventV2 } from "@opencode-ai/core/event" -import { isOpenCodeEvent, OpenCodeEvent } from "@opencode-ai/protocol/groups/event" -import { Cause, Context, Effect, Layer, Queue, Schema, Scope, Stream } from "effect" +import { EventFeedDiagnostics } from "@opencode-ai/protocol/groups/debug" +import { + EventSubscribeQuery, + isOpenCodeEvent, + OpenCodeEvent, + sessionIDOf, +} from "@opencode-ai/protocol/groups/event" +import { Cause, Context, Effect, Layer, Option, Queue, Schema, Scope, Stream } from "effect" export const SubscriberCapacity = 4_096 +/** Core-published types that always pass interest filters (location and session). */ +export const GlobalEventTypes = new Set([ + "installation.updated", + "installation.update-available", + "global.disposed", +]) + export class SubscriberOverflowError extends Schema.TaggedErrorClass()( "EventFeed.SubscriberOverflow", { capacity: Schema.Int }, @@ -19,8 +32,23 @@ export class EncodingError extends Schema.TaggedErrorClass()("Eve export type Error = SubscriberOverflowError | EncodingError +export type LocationInterest = { + readonly directory: string + readonly workspace?: string +} + +/** Omit or leave empty for the global public feed; set location/sessions to narrow delivery. */ +export type Interest = { + readonly location?: LocationInterest + readonly sessions?: ReadonlyArray +} + +/** Content-free counters for leak detection — never includes event payloads. */ +export type Diagnostics = typeof EventFeedDiagnostics.Type + export interface Interface { - readonly subscribe: Effect.Effect, never, Scope.Scope> + readonly subscribe: (interest?: Interest) => Effect.Effect, never, Scope.Scope> + readonly diagnostics: () => Diagnostics } export class Service extends Context.Service()("@opencode/server/EventFeed") {} @@ -31,24 +59,121 @@ export function frame(event: OpenCodeEvent) { return `data: ${JSON.stringify(encode(event))}\n\n` } +export function matchesInterest(event: EventV2.Payload, interest?: Interest): boolean { + if (interest === undefined) return true + if (GlobalEventTypes.has(event.type)) return true + + const location = interest.location + if (location !== undefined) { + const ref = event.location + if (ref === undefined) return false + if (ref.directory !== location.directory) return false + if (location.workspace !== undefined && ref.workspaceID !== location.workspace) return false + } + + const sessions = interest.sessions + if (sessions !== undefined && sessions.length > 0) { + const sessionID = sessionIDOf(event) + if (sessionID !== undefined) return sessions.includes(sessionID) + } + + return true +} + +const decodeSubscribeQuery = Schema.decodeUnknownOption(EventSubscribeQuery) + +/** Map an already-decoded subscribe query into feed interest (HttpApi `ctx.query`). */ +export function interestFromSubscribeQuery(query: typeof EventSubscribeQuery.Type): Interest | undefined { + const sessions = normalizeSessions(query.session) + if (query.location === undefined && sessions.length === 0) return undefined + return { + ...(query.location + ? { + location: { + directory: query.location.directory, + ...(query.location.workspace !== undefined ? { workspace: query.location.workspace } : {}), + }, + } + : {}), + ...(sessions.length > 0 ? { sessions } : {}), + } +} + +/** Parse deepObject / repeated `session` search params through `EventSubscribeQuery`. */ +export function interestFromQuery(params: URLSearchParams): Interest | undefined { + const decoded = Option.getOrUndefined(decodeSubscribeQuery(subscribeQueryCandidate(params))) + if (decoded === undefined) return undefined + return interestFromSubscribeQuery(decoded) +} + +function subscribeQueryCandidate(params: URLSearchParams) { + const directory = params.get("location[directory]") ?? undefined + const workspace = params.get("location[workspace]") ?? undefined + const sessions = params + .getAll("session") + .flatMap((value) => value.split(",")) + .map((value) => value.trim()) + .filter((value) => value.length > 0) + return { + ...(directory !== undefined || workspace !== undefined + ? { + location: { + ...(directory !== undefined ? { directory } : {}), + ...(workspace !== undefined ? { workspace } : {}), + }, + } + : {}), + ...(sessions.length === 0 ? {} : { session: sessions.length === 1 ? sessions[0] : sessions }), + } +} + +function normalizeSessions(session: typeof EventSubscribeQuery.Type.session): string[] { + if (session === undefined) return [] + if (typeof session === "string") return session.trim() === "" ? [] : [session] + return session.map((value) => value.trim()).filter((value) => value.length > 0) +} + +type Subscriber = { + readonly queue: Queue.Queue + readonly interest?: Interest +} + export const make = Effect.fn("EventFeed.make")(function* ( observe: (subscriber: EventV2.Subscriber) => Effect.Effect, options?: { readonly capacity?: number; readonly encode?: (event: OpenCodeEvent) => string }, ) { const capacity = options?.capacity ?? SubscriberCapacity const render = options?.encode ?? frame - const subscribers = new Set>() + const subscribers = new Map, Subscriber>() + let opens = 0 + let closes = 0 + let serializedEvents = 0 + let serializedBytes = 0 + let overflows = 0 + + const diagnostics = (): Diagnostics => ({ + active: subscribers.size, + opens, + closes, + serializedEvents, + serializedBytes, + overflows, + }) const fail = (error: Error) => Effect.sync(() => { - const current = Array.from(subscribers) + const current = Array.from(subscribers.values()) subscribers.clear() - for (const subscriber of current) Queue.failCauseUnsafe(subscriber, Cause.fail(error)) + for (const subscriber of current) Queue.failCauseUnsafe(subscriber.queue, Cause.fail(error)) }) const publish = Effect.fnUntraced(function* (event: EventV2.Payload) { if (!isOpenCodeEvent(event)) return if (subscribers.size === 0) return + const targets = Array.from(subscribers.values()).filter((subscriber) => + matchesInterest(event, subscriber.interest), + ) + if (targets.length === 0) return const encoded = yield* Effect.try({ try: () => render(event), catch: (cause) => new EncodingError({ eventID: event.id, eventType: event.type, cause }), @@ -62,10 +187,13 @@ export const make = Effect.fn("EventFeed.make")(function* ( ), ) if (encoded === undefined) return - for (const subscriber of subscribers) { - if (Queue.offerUnsafe(subscriber, encoded)) continue - subscribers.delete(subscriber) - Queue.failCauseUnsafe(subscriber, Cause.fail(new SubscriberOverflowError({ capacity }))) + serializedEvents += 1 + serializedBytes += Buffer.byteLength(encoded, "utf-8") + for (const subscriber of targets) { + if (Queue.offerUnsafe(subscriber.queue, encoded)) continue + overflows += 1 + subscribers.delete(subscriber.queue) + Queue.failCauseUnsafe(subscriber.queue, Cause.fail(new SubscriberOverflowError({ capacity }))) } }) @@ -73,11 +201,22 @@ export const make = Effect.fn("EventFeed.make")(function* ( yield* Effect.addFinalizer(() => unsubscribe) return Service.of({ - subscribe: Effect.acquireRelease( - Queue.dropping(capacity).pipe(Effect.tap((queue) => Effect.sync(() => subscribers.add(queue)))), - (queue) => - Effect.sync(() => subscribers.delete(queue)).pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid), - ).pipe(Effect.map(Stream.fromQueue)), + diagnostics, + subscribe: (interest) => + Effect.acquireRelease( + Queue.dropping(capacity).pipe( + Effect.tap((queue) => + Effect.sync(() => { + opens += 1 + subscribers.set(queue, { queue, interest }) + }), + ), + ), + (queue) => + Effect.sync(() => { + if (subscribers.delete(queue)) closes += 1 + }).pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid), + ).pipe(Effect.map(Stream.fromQueue)), }) }) diff --git a/packages/server/src/handlers/debug.ts b/packages/server/src/handlers/debug.ts index ec797c003213..5d2905b2d517 100644 --- a/packages/server/src/handlers/debug.ts +++ b/packages/server/src/handlers/debug.ts @@ -2,6 +2,7 @@ import { LocationServiceMap } from "@opencode-ai/core/location-service-map" import { Effect, Option, RcMap } from "effect" import { HttpApiBuilder } from "effect/unstable/httpapi" import { Api } from "../api" +import { EventFeed } from "../event-feed" import { requestRef } from "../location" export const DebugHandler = HttpApiBuilder.group(Api, "server.debug", (handlers) => @@ -21,5 +22,12 @@ export const DebugHandler = HttpApiBuilder.group(Api, "server.debug", (handlers) // middleware cached the services under. yield* locations.invalidate(requestRef(ctx.request)) }), + ) + .handle( + "debug.event-feed", + Effect.fn(function* () { + const feed = Option.getOrThrow(yield* Effect.serviceOption(EventFeed.Service)) + return feed.diagnostics() + }), ), ) diff --git a/packages/server/src/handlers/event.ts b/packages/server/src/handlers/event.ts index feeba74288d6..2f88cb3d3e0f 100644 --- a/packages/server/src/handlers/event.ts +++ b/packages/server/src/handlers/event.ts @@ -1,36 +1,70 @@ +import { NodeHttpServerRequest } from "@effect/platform-node" import { EventV2 } from "@opencode-ai/core/event" +import { EventSubscribeQuery } from "@opencode-ai/protocol/groups/event" import { Effect, Stream } from "effect" -import { HttpServerResponse } from "effect/unstable/http" +import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http" import { HttpApiBuilder } from "effect/unstable/httpapi" import { Api } from "../api" import { EventFeed } from "../event-feed" +/** Bun may emit IncomingMessage aborted/close without ServerResponse.close — destroy so the fiber ends. */ +export function bridgeClientDisconnect(request: HttpServerRequest.HttpServerRequest) { + return Effect.sync(() => { + try { + const incoming = NodeHttpServerRequest.toIncomingMessage(request) + const response = NodeHttpServerRequest.toServerResponse(request) + const destroy = () => { + if (!response.writableEnded && !response.destroyed) response.destroy() + } + if (incoming.aborted || incoming.destroyed) { + destroy() + return + } + incoming.once("aborted", destroy) + incoming.once("close", destroy) + } catch { + // Non-Node sources (tests using fromWeb) have no ServerResponse to bridge. + } + }) +} + +/** Shared subscribe response for the production handler and ownership integration tests. */ +export const subscribeResponse = ( + feed: EventFeed.Interface, + request: HttpServerRequest.HttpServerRequest, + query: typeof EventSubscribeQuery.Type, +) => + Effect.gen(function* () { + yield* bridgeClientDisconnect(request) + // handleRaw still Schema-decodes query; map into feed interest without re-parsing the URL. + const interest = EventFeed.interestFromSubscribeQuery(query) + const connected = { + id: EventV2.ID.create(), + type: "server.connected", + data: {}, + } as const + const output = Stream.unwrap( + feed + .subscribe(interest) + .pipe(Effect.map((live) => Stream.make(EventFeed.frame(connected)).pipe(Stream.concat(live)))), + ) + const heartbeat = Stream.tick("15 seconds").pipe(Stream.map(() => ": heartbeat\n\n")) + return HttpServerResponse.stream( + output.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }), Stream.encodeText), + { + contentType: "text/event-stream", + headers: { + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + "X-Content-Type-Options": "nosniff", + }, + }, + ) + }) + export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) => Effect.gen(function* () { const feed = yield* EventFeed.Service - return handlers.handleRaw("event.subscribe", () => - Effect.gen(function* () { - const connected = { - id: EventV2.ID.create(), - type: "server.connected", - data: {}, - } as const - const output = Stream.unwrap( - feed.subscribe.pipe(Effect.map((live) => Stream.make(EventFeed.frame(connected)).pipe(Stream.concat(live)))), - ) - const heartbeat = Stream.tick("15 seconds").pipe(Stream.map(() => ": heartbeat\n\n")) - return HttpServerResponse.stream( - output.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }), Stream.encodeText), - { - contentType: "text/event-stream", - headers: { - "Cache-Control": "no-cache, no-transform", - "X-Accel-Buffering": "no", - "X-Content-Type-Options": "nosniff", - }, - }, - ) - }), - ) + return handlers.handleRaw("event.subscribe", (ctx) => subscribeResponse(feed, ctx.request, ctx.query)) }), ) diff --git a/packages/server/test/event-feed-fanout.bench.test.ts b/packages/server/test/event-feed-fanout.bench.test.ts new file mode 100644 index 000000000000..f722fe5c46e8 --- /dev/null +++ b/packages/server/test/event-feed-fanout.bench.test.ts @@ -0,0 +1,166 @@ +import { describe, expect } from "bun:test" +import { EventV2 } from "@opencode-ai/core/event" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { DateTime, Effect, Fiber, Stream } from "effect" +import { it } from "../../core/test/lib/effect" +import { EventFeed } from "../src/event-feed" + +const Clients = [1, 2, 4, 8, 11, 16] as const +const Sizes = { + small: 1 * 1024, + medium: 64 * 1024, + large: 256 * 1024, +} as const + +function makeSource() { + let subscriber: EventV2.Subscriber | undefined + return { + observe: (next: EventV2.Subscriber) => + Effect.sync(() => { + subscriber = next + return Effect.sync(() => { + if (subscriber === next) subscriber = undefined + }) + }), + publish: (event: EventV2.Payload) => Effect.suspend(() => (subscriber ? subscriber(event) : Effect.void)), + } +} + +function sessionEvent(id: string, sessionID: string, location: { directory: string }): EventV2.Payload { + return { + id: EventV2.ID.make(`evt_${id}`), + created: DateTime.makeUnsafe(Date.now()), + type: "session.renamed", + data: { sessionID, title: id }, + durable: { aggregateID: sessionID, seq: EventV2.Seq.make(1), version: EventV2.Version.make(1) }, + location: { + directory: AbsolutePath.make(location.directory), + }, + } +} + +function eventLoopDelayMs() { + const start = performance.now() + return new Promise((resolve) => { + setTimeout(() => resolve(performance.now() - start), 0) + }) +} + +type Row = { + clients: number + size: keyof typeof Sizes + payloadBytes: number + interested: number + serializedEvents: number + serializedBytes: number + peakActive: number + receivedInterested: number + receivedUninterested: number + publishMs: number + clientHandleMs: number + rssBefore: number + rssAfter: number + eventLoopDelayMs: number +} + +describe("EventFeed fan-out benchmark", () => { + it.live( + "accounting laws hold across client counts and payload sizes", + () => + Effect.gen(function* () { + const rows: Row[] = [] + + for (const clients of Clients) { + for (const size of Object.keys(Sizes) as Array) { + rows.push(yield* measureCell(clients, size)) + } + } + + // Same payload class → same serialized byte cost regardless of uninterested N. + for (const size of Object.keys(Sizes) as Array) { + const bytes = rows.filter((row) => row.size === size).map((row) => row.serializedBytes) + expect(new Set(bytes).size).toBe(1) + expect(bytes[0]).toBe(Sizes[size]) + } + + // Encode-once: every cell accepted exactly one hot publish. + expect(rows.every((row) => row.serializedEvents === 1)).toBe(true) + + console.log(JSON.stringify({ benchmark: "event-feed-fanout", rows }, null, 2)) + }), + 60_000, + ) +}) + +function measureCell(clients: number, size: keyof typeof Sizes) { + const payloadBytes = Sizes[size] + const interested = Math.max(1, Math.floor(clients / 2)) + const frame = "x".repeat(payloadBytes) + + return Effect.scoped( + Effect.gen(function* () { + let serializedEvents = 0 + let serializedBytes = 0 + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { + encode: () => { + serializedEvents += 1 + serializedBytes += Buffer.byteLength(frame, "utf-8") + return frame + }, + }) + + const received = Array.from({ length: clients }, () => 0) + const fibers: Array> = [] + + for (let index = 0; index < clients; index++) { + const sessions = index < interested ? ["ses_hot"] : [`ses_${index}`] + const stream = yield* feed.subscribe({ + location: { directory: "/tmp/project" }, + sessions, + }) + fibers.push( + yield* stream.pipe( + Stream.take(1), + Stream.runForEach(() => Effect.sync(() => (received[index] += 1))), + Effect.forkScoped, + ), + ) + } + + const rssBefore = process.memoryUsage().rss + const loopDelay = yield* Effect.promise(eventLoopDelayMs) + const publishStarted = performance.now() + yield* source.publish(sessionEvent("hot", "ses_hot", { directory: "/tmp/project" })) + const publishMs = performance.now() - publishStarted + + const handleStarted = performance.now() + for (const fiber of fibers.slice(0, interested)) yield* Fiber.join(fiber) + const clientHandleMs = performance.now() - handleStarted + + for (const fiber of fibers.slice(interested)) yield* Fiber.interrupt(fiber) + + expect(serializedEvents).toBe(1) + expect(serializedBytes).toBe(payloadBytes) + expect(received.slice(0, interested).every((count) => count === 1)).toBe(true) + expect(received.slice(interested).every((count) => count === 0)).toBe(true) + + return { + clients, + size, + payloadBytes, + interested, + serializedEvents, + serializedBytes, + peakActive: clients, + receivedInterested: received.slice(0, interested).reduce((sum, count) => sum + count, 0), + receivedUninterested: received.slice(interested).reduce((sum, count) => sum + count, 0), + publishMs, + clientHandleMs, + rssBefore, + rssAfter: process.memoryUsage().rss, + eventLoopDelayMs: loopDelay, + } satisfies Row + }), + ) +} diff --git a/packages/server/test/event-feed-ownership.test.ts b/packages/server/test/event-feed-ownership.test.ts new file mode 100644 index 000000000000..5a562f2bccf7 --- /dev/null +++ b/packages/server/test/event-feed-ownership.test.ts @@ -0,0 +1,114 @@ +import { NodeHttpServer } from "@effect/platform-node" +import { EventV2 } from "@opencode-ai/core/event" +import { describe, expect } from "bun:test" +import { Context, Effect, Fiber, Layer, Stream } from "effect" +import { HttpServer, HttpServerRequest } from "effect/unstable/http" +import { createServer } from "node:http" +import { it } from "../../core/test/lib/effect" +import { EventFeed } from "../src/event-feed" +import { subscribeResponse } from "../src/handlers/event" + +function makeSource() { + let subscriber: EventV2.Subscriber | undefined + return { + observe: (next: EventV2.Subscriber) => + Effect.sync(() => { + subscriber = next + return Effect.sync(() => { + if (subscriber === next) subscriber = undefined + }) + }), + publish: (event: EventV2.Payload) => Effect.suspend(() => (subscriber ? subscriber(event) : Effect.void)), + } +} + +async function wait(fn: () => boolean, timeout = 3000) { + const start = Date.now() + while (!fn()) { + if (Date.now() - start > timeout) throw new Error("timed out waiting for condition") + await Bun.sleep(10) + } +} + +describe("EventFeed ownership", () => { + it.effect("canceling a subscriber scope returns diagnostics.active to baseline", () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { encode: (value) => value.id }) + const baseline = feed.diagnostics().active + + const fiber = yield* Effect.gen(function* () { + const stream = yield* feed.subscribe() + expect(feed.diagnostics().active).toBe(baseline + 1) + yield* stream.pipe(Stream.runCollect) + }).pipe(Effect.scoped, Effect.forkChild) + + yield* Effect.yieldNow + yield* Fiber.interrupt(fiber) + yield* Fiber.await(fiber) + + expect(feed.diagnostics().active).toBe(baseline) + expect(feed.diagnostics().opens).toBeGreaterThanOrEqual(1) + expect(feed.diagnostics().closes).toBeGreaterThanOrEqual(1) + }), + ) + + it.live( + "aborting a real Node HTTP client returns EventFeed.active to baseline", + () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe) + const baseline = feed.diagnostics().active + + const context = yield* Layer.build(NodeHttpServer.layer(createServer, { host: "127.0.0.1", port: 0 })) + const server = Context.get(context, HttpServer.HttpServer) + yield* server + .serve(HttpServerRequest.HttpServerRequest.use((request) => subscribeResponse(feed, request, {}))) + .pipe(Effect.forkScoped) + + const base = HttpServer.formatAddress(server.address) + const controller = new AbortController() + const response = yield* Effect.promise(() => fetch(`${base}/api/event`, { signal: controller.signal })) + expect(response.ok).toBe(true) + expect(response.body).toBeDefined() + const reader = response.body?.getReader() + expect(reader).toBeDefined() + if (!reader) throw new Error("expected event stream body reader") + + yield* Effect.promise(() => reader.read()) + expect(feed.diagnostics().active).toBe(baseline + 1) + + controller.abort() + // Bun may never settle `reader.closed` after abort; cancel and watch feed release. + yield* Effect.promise(() => reader.cancel().catch(() => undefined)) + yield* Effect.promise(() => wait(() => feed.diagnostics().active === baseline)) + + expect(feed.diagnostics().active).toBe(baseline) + expect(feed.diagnostics().closes).toBeGreaterThanOrEqual(1) + const dump = JSON.stringify(feed.diagnostics()) + expect(dump.includes("payload")).toBe(false) + }), + 15_000, + ) + + it.effect("eleven sequential subscribe scopes never accumulate active subscribers", () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { encode: (value) => value.id }) + + for (let i = 0; i < 11; i++) { + yield* Effect.scoped( + Effect.gen(function* () { + yield* feed.subscribe() + expect(feed.diagnostics().active).toBe(1) + }), + ) + expect(feed.diagnostics().active).toBe(0) + } + + expect(feed.diagnostics().opens).toBe(11) + expect(feed.diagnostics().closes).toBe(11) + }), + ) +}) diff --git a/packages/server/test/event-feed.test.ts b/packages/server/test/event-feed.test.ts index 29129c735c2f..3fff4d3f50dd 100644 --- a/packages/server/test/event-feed.test.ts +++ b/packages/server/test/event-feed.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test" import { AgentV2 } from "@opencode-ai/core/agent" import { EventV2 } from "@opencode-ai/core/event" +import { AbsolutePath } from "@opencode-ai/core/schema" import { OpenCodeEvent } from "@opencode-ai/protocol/groups/event" import { DateTime, Deferred, Effect, Exit, Fiber, Option, Schema, Stream } from "effect" import { it } from "../../core/test/lib/effect" @@ -8,11 +9,21 @@ import { EventFeed } from "../src/event-feed" const Internal = EventV2.ephemeral({ type: "test.internal", schema: { value: Schema.String } }) -const event = (id: string): EventV2.Payload => ({ +const event = ( + id: string, + location?: { directory: string }, +): EventV2.Payload => ({ id: EventV2.ID.make(`evt_${id}`), created: DateTime.makeUnsafe(Date.now()), type: AgentV2.Event.Updated.type, data: {}, + ...(location + ? { + location: { + directory: AbsolutePath.make(location.directory), + }, + } + : {}), }) const internal = (value: string): EventV2.Payload => ({ @@ -54,8 +65,8 @@ describe("EventFeed", () => { return event.type }, }) - const first = yield* feed.subscribe - const second = yield* feed.subscribe + const first = yield* feed.subscribe() + const second = yield* feed.subscribe() const left = yield* first.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) const right = yield* second.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) @@ -76,8 +87,8 @@ describe("EventFeed", () => { capacity: 1, encode: (event) => event.id, }) - const slow = yield* feed.subscribe - const fast = yield* feed.subscribe + const slow = yield* feed.subscribe() + const fast = yield* feed.subscribe() const first = yield* Deferred.make() const second = yield* Deferred.make() const received = new Array() @@ -117,7 +128,7 @@ describe("EventFeed", () => { Effect.gen(function* () { const source = makeSource() const feed = yield* EventFeed.make(source.observe, { capacity: 1, encode: (event) => event.type }) - const stream = yield* feed.subscribe + const stream = yield* feed.subscribe() yield* source.publish(internal("one")) yield* source.publish(internal("two")) @@ -136,13 +147,13 @@ describe("EventFeed", () => { return event.id }, }) - const current = yield* feed.subscribe + const current = yield* feed.subscribe() const failed = yield* current.pipe(Stream.runCollect, Effect.exit, Effect.forkScoped) yield* source.publish(event("bad")) const exit = yield* Fiber.join(failed) - const next = yield* feed.subscribe + const next = yield* feed.subscribe() const received = yield* next.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) yield* source.publish(event("good")) @@ -152,4 +163,279 @@ describe("EventFeed", () => { expect(Array.from(yield* Fiber.join(received))).toEqual(["evt_good"]) }), ) + + it.effect("delivers location-scoped events only to matching subscribers", () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { encode: (event) => event.id }) + const alpha = yield* feed.subscribe({ location: { directory: "/tmp/alpha" } }) + const beta = yield* feed.subscribe({ location: { directory: "/tmp/beta" } }) + const left = yield* alpha.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + const right = yield* beta.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + + yield* source.publish(event("alpha", { directory: "/tmp/alpha" })) + yield* source.publish(event("beta", { directory: "/tmp/beta" })) + + expect(Array.from(yield* Fiber.join(left))).toEqual(["evt_alpha"]) + expect(Array.from(yield* Fiber.join(right))).toEqual(["evt_beta"]) + }), + ) + + it.effect("keeps the unscoped feed global while scoped subscribers filter", () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { encode: (event) => event.id }) + const global = yield* feed.subscribe() + const scoped = yield* feed.subscribe({ location: { directory: "/tmp/alpha" } }) + const all = yield* global.pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped) + const only = yield* scoped.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + + yield* source.publish(event("alpha", { directory: "/tmp/alpha" })) + yield* source.publish(event("beta", { directory: "/tmp/beta" })) + + expect(Array.from(yield* Fiber.join(all))).toEqual(["evt_alpha", "evt_beta"]) + expect(Array.from(yield* Fiber.join(only))).toEqual(["evt_alpha"]) + }), + ) + + it.effect("encodes once for multiple scoped subscribers that match the same event", () => + Effect.gen(function* () { + let encodes = 0 + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { + encode: (event) => { + encodes += 1 + return event.id + }, + }) + const first = yield* feed.subscribe({ location: { directory: "/tmp/alpha" } }) + const second = yield* feed.subscribe({ location: { directory: "/tmp/alpha" } }) + const left = yield* first.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + const right = yield* second.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + + yield* source.publish(event("shared", { directory: "/tmp/alpha" })) + + expect([Array.from(yield* Fiber.join(left)), Array.from(yield* Fiber.join(right))]).toEqual([ + ["evt_shared"], + ["evt_shared"], + ]) + expect(encodes).toBe(1) + }), + ) + + it.effect("does not encode when no subscriber interest matches", () => + Effect.gen(function* () { + let encodes = 0 + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { + encode: (event) => { + encodes += 1 + return event.id + }, + }) + yield* feed.subscribe({ location: { directory: "/tmp/alpha" } }) + yield* source.publish(event("beta", { directory: "/tmp/beta" })) + expect(encodes).toBe(0) + }), + ) + + test("parses location interest from deepObject query params", () => { + const params = new URLSearchParams() + params.set("location[directory]", "/tmp/project") + params.set("location[workspace]", "ws_1") + expect(EventFeed.interestFromQuery(params)).toEqual({ + location: { directory: "/tmp/project", workspace: "ws_1" }, + }) + expect(EventFeed.interestFromQuery(new URLSearchParams())).toBeUndefined() + }) + + test("parses repeated session interest from query params", () => { + const params = new URLSearchParams() + params.append("session", "ses_a") + params.append("session", "ses_b,ses_c") + expect(EventFeed.interestFromQuery(params)).toEqual({ + sessions: ["ses_a", "ses_b", "ses_c"], + }) + }) + + test("rejects workspace interest without a directory", () => { + const params = new URLSearchParams() + params.set("location[workspace]", "ws_1") + params.append("session", "ses_a") + expect(EventFeed.interestFromQuery(params)).toBeUndefined() + }) + + test("maps a decoded EventSubscribeQuery into feed interest", () => { + expect(EventFeed.interestFromSubscribeQuery({})).toBeUndefined() + expect( + EventFeed.interestFromSubscribeQuery({ + location: { directory: "/tmp/project", workspace: "ws_1" }, + session: ["ses_a", "ses_b"], + }), + ).toEqual({ + location: { directory: "/tmp/project", workspace: "ws_1" }, + sessions: ["ses_a", "ses_b"], + }) + expect(EventFeed.interestFromSubscribeQuery({ session: "ses_a" })).toEqual({ sessions: ["ses_a"] }) + }) + + test("matchesInterest delivers global types to location-scoped subscribers", () => { + expect( + EventFeed.matchesInterest( + { + id: EventV2.ID.make("evt_global"), + created: DateTime.makeUnsafe(0), + type: "global.disposed", + data: {}, + }, + { location: { directory: "/tmp/project" } }, + ), + ).toBe(true) + }) + + test("matchesInterest rejects workspace mismatches", () => { + expect( + EventFeed.matchesInterest(event("ws", { directory: "/tmp/project" }), { + location: { directory: "/tmp/project", workspace: "ws_other" }, + }), + ).toBe(false) + }) + + it.effect("delivers session-scoped events only to interested subscribers", () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { encode: (event) => event.id }) + const alpha = yield* feed.subscribe({ + location: { directory: "/tmp/project" }, + sessions: ["ses_a"], + }) + const beta = yield* feed.subscribe({ + location: { directory: "/tmp/project" }, + sessions: ["ses_b"], + }) + const left = yield* alpha.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + const right = yield* beta.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + + yield* source.publish(sessionEvent("a", "ses_a", { directory: "/tmp/project" })) + yield* source.publish(sessionEvent("b", "ses_b", { directory: "/tmp/project" })) + + expect(Array.from(yield* Fiber.join(left))).toEqual(["evt_a"]) + expect(Array.from(yield* Fiber.join(right))).toEqual(["evt_b"]) + }), + ) + + it.effect("keeps location-only events flowing when session interest is set", () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { encode: (event) => event.id }) + const stream = yield* feed.subscribe({ + location: { directory: "/tmp/project" }, + sessions: ["ses_a"], + }) + const received = yield* stream.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + yield* source.publish(event("catalog", { directory: "/tmp/project" })) + expect(Array.from(yield* Fiber.join(received))).toEqual(["evt_catalog"]) + }), + ) + + it.effect("fans out eleven concurrent clients by interest, encoding once per event", () => + Effect.gen(function* () { + let encodes = 0 + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { + encode: (event) => { + encodes += 1 + return event.id + }, + }) + const fibers: Array> = [] + const counts = Array.from({ length: 11 }, () => 0) + for (let i = 0; i < 11; i++) { + const index = i + // 0-4: disjoint sessions, 5-9: shared session, 10: location-only (overlapping) + const sessions = index < 5 ? [`ses_${index}`] : index < 10 ? ["ses_shared"] : undefined + const expected = index < 10 ? 1 : 2 + const stream = yield* feed.subscribe({ + location: { directory: "/tmp/project" }, + ...(sessions ? { sessions } : {}), + }) + fibers.push( + yield* stream.pipe( + Stream.take(expected), + Stream.runForEach(() => Effect.sync(() => (counts[index] += 1))), + Effect.forkScoped, + ), + ) + } + + for (let i = 0; i < 5; i++) { + yield* source.publish(sessionEvent(`s${i}`, `ses_${i}`, { directory: "/tmp/project" })) + } + yield* source.publish(sessionEvent("shared", "ses_shared", { directory: "/tmp/project" })) + + for (const fiber of fibers) yield* Fiber.join(fiber) + + expect(encodes).toBe(6) + expect(counts).toEqual([1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2]) + expect(feed.diagnostics().active).toBe(11) + expect(feed.diagnostics().serializedEvents).toBe(6) + }), + ) + + it.effect("returns active subscriber count to baseline when scopes close after eleven-client churn", () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { encode: (event) => event.id }) + expect(feed.diagnostics()).toMatchObject({ active: 0, opens: 0, closes: 0 }) + + yield* Effect.scoped( + Effect.gen(function* () { + for (let i = 0; i < 11; i++) yield* feed.subscribe() + expect(feed.diagnostics().active).toBe(11) + expect(feed.diagnostics().opens).toBe(11) + yield* source.publish(event("diag")) + expect(feed.diagnostics().serializedEvents).toBe(1) + expect(feed.diagnostics().serializedBytes).toBeGreaterThan(0) + }), + ) + + expect(feed.diagnostics().active).toBe(0) + expect(feed.diagnostics().closes).toBe(11) + const dump = JSON.stringify(feed.diagnostics()) + expect(dump.includes("evt_diag")).toBe(false) + expect(dump.includes("payload")).toBe(false) + }), + ) + + it.effect("counts overflows without retaining failed subscriber queues", () => + Effect.gen(function* () { + const source = makeSource() + const feed = yield* EventFeed.make(source.observe, { + capacity: 1, + encode: (event) => event.id, + }) + const slow = yield* feed.subscribe() + yield* source.publish(event("one")) + yield* source.publish(event("two")) + const exit = yield* slow.pipe(Stream.runCollect, Effect.exit) + expect(Exit.isFailure(exit)).toBeTrue() + expect(feed.diagnostics().overflows).toBe(1) + expect(feed.diagnostics().active).toBe(0) + }), + ) +}) + +const sessionEvent = ( + id: string, + sessionID: string, + location: { directory: string }, +): EventV2.Payload => ({ + id: EventV2.ID.make(`evt_${id}`), + created: DateTime.makeUnsafe(Date.now()), + type: "session.renamed", + data: { sessionID, title: id }, + durable: { aggregateID: sessionID, seq: EventV2.Seq.make(1), version: EventV2.Version.make(1) }, + location: { + directory: AbsolutePath.make(location.directory), + }, }) diff --git a/packages/tui/src/app.tsx b/packages/tui/src/app.tsx index a0b2e3fc0be0..e440600c34a4 100644 --- a/packages/tui/src/app.tsx +++ b/packages/tui/src/app.tsx @@ -334,7 +334,11 @@ export const run = Effect.fn("Tui.run")(function* (input: TuiInput) { } > - + diff --git a/packages/tui/src/context/client.tsx b/packages/tui/src/context/client.tsx index 83182f241e64..cb2a300af9c6 100644 --- a/packages/tui/src/context/client.tsx +++ b/packages/tui/src/context/client.tsx @@ -5,6 +5,10 @@ import { createStore } from "solid-js/store" import { errorMessage } from "../util/error" import { createSimpleContext } from "./helper" import { useLog } from "./log" +import { interestEqual, subscribeInput, type EventInterest } from "./event-interest" + +export type { EventInterest } from "./event-interest" +export { interestEqual, subscribeInput } from "./event-interest" export type ClientConnectionStatus = "connected" | "connecting" | "reconnecting" export type ClientConnectionEvent = { @@ -25,14 +29,17 @@ type ManagedService = { type ClientEventMap = { [Type in OpenCodeEvent["type"]]: Extract } const connectTimeout = 2_000 const connectionHistoryLimit = 50 +const reconnectBaseMs = 1_000 +const reconnectJitterMs = 500 export const { use: useClient, provider: ClientProvider } = createSimpleContext({ name: "Client", - init: (props: { api: OpenCodeClient; service?: ManagedService }) => { + init: (props: { api: OpenCodeClient; service?: ManagedService; interest?: EventInterest }) => { const log = useLog({ component: "client" }) const abort = new AbortController() const history: ClientConnectionEvent[] = [] let api = props.api + let interest = props.interest const events = createGlobalEmitter() const [connection, setConnection] = createStore<{ status: ClientConnectionStatus @@ -43,6 +50,8 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( attempt: 0, }) let stream: AbortController | undefined + let generation = 0 + let running: Promise | undefined function record(status: ClientConnectionEvent["data"]["status"], attempt: number, error?: string) { history.push({ type: "client.connection", created: Date.now(), data: { status, attempt, error } }) @@ -50,12 +59,16 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( } function start() { + const gen = ++generation stream?.abort() + const previous = running const controller = new AbortController() stream = controller - void (async () => { + running = (async () => { + if (previous) await previous.catch(() => undefined) + if (gen !== generation || abort.signal.aborted || controller.signal.aborted) return let attempt = 0 - while (!abort.signal.aborted && !controller.signal.aborted) { + while (!abort.signal.aborted && !controller.signal.aborted && gen === generation) { let connectedAt: number | undefined const request = new AbortController() const cancel = () => request.abort(controller.signal.reason) @@ -63,10 +76,12 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( controller.signal.addEventListener("abort", cancel, { once: true }) const error = await (async () => { record(attempt === 0 ? "connecting" : "reconnecting", attempt) - log.info("event stream connecting", { attempt }) - const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]() + log.info("event stream connecting", { attempt, generation: gen }) + const iterator = api.event + .subscribe(subscribeInput(interest), { signal: request.signal }) + [Symbol.asyncIterator]() const first = await iterator.next() - if (abort.signal.aborted || controller.signal.aborted) return undefined + if (abort.signal.aborted || controller.signal.aborted || gen !== generation) return undefined if (first.done) return request.signal.reason instanceof Error ? request.signal.reason @@ -76,12 +91,12 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( clearTimeout(timeout) record("connected", attempt) connectedAt = Date.now() - log.info("event stream connected") + log.info("event stream connected", { generation: gen }) events.emit(first.value.type, first.value) setConnection({ status: "connected", attempt: 0, error: undefined }) - while (!abort.signal.aborted && !controller.signal.aborted) { + while (!abort.signal.aborted && !controller.signal.aborted && gen === generation) { const event = await iterator.next() - if (abort.signal.aborted || controller.signal.aborted) return undefined + if (abort.signal.aborted || controller.signal.aborted || gen !== generation) return undefined if (event.done) return new Error("Event stream disconnected") if ("durable" in event.value) log.debug("event", { @@ -99,13 +114,14 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( clearTimeout(timeout) controller.signal.removeEventListener("abort", cancel) }) - if (abort.signal.aborted || controller.signal.aborted) return + if (abort.signal.aborted || controller.signal.aborted || gen !== generation) return if (connectedAt !== undefined && Date.now() - connectedAt >= 1_000) attempt = 0 attempt += 1 const message = errorMessage(error) record("disconnected", attempt, message) log.info("event stream disconnected", { attempt, + generation: gen, error: message, }) setConnection({ status: "reconnecting", attempt, error: message }) @@ -120,19 +136,20 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( error: errorMessage(error), }) }) - if (abort.signal.aborted || controller.signal.aborted) return + if (abort.signal.aborted || controller.signal.aborted || gen !== generation) return if (next) { api = next.api if (attempt === 1) continue } } - await wait(1_000, controller.signal) + await wait(reconnectBaseMs + Math.floor(Math.random() * reconnectJitterMs), controller.signal) } })() } onMount(start) onCleanup(() => { + generation += 1 abort.abort() stream?.abort() events.clear() @@ -145,6 +162,11 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( event: { on: events.on, listen: events.listen, + scope(next: EventInterest) { + if (interestEqual(interest, next)) return + interest = next + start() + }, }, connection: { status() { @@ -160,6 +182,9 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( history() { return history.slice() }, + generation() { + return generation + }, }, }, restart: props.service?.restart, diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 14990fb93373..fad43bb81c71 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -32,6 +32,7 @@ import type { Plugin } from "@opencode-ai/plugin/v2/tui" import { createStore, produce, reconcile } from "solid-js/store" import { createSimpleContext } from "./helper" import { useClient } from "./client" +import { useRoute } from "./route" import { createEffect, createSignal, onCleanup } from "solid-js" export type DataSessionStatus = "idle" | "running" @@ -142,6 +143,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }) const client = useClient() + const route = useRoute() const [defaultLocation, setDefaultLocation] = createSignal({ directory: process.cwd(), }) @@ -1272,6 +1274,22 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ sync.invalidate() }) + createEffect(() => { + const location = defaultLocation() + const current = route.data + const sessions = + current.type === "session" + ? (store.session.family[resolveRoot(current.sessionID)] ?? [current.sessionID]).slice().sort() + : undefined + client.event.scope({ + location: { + directory: location.directory, + ...(location.workspaceID ? { workspace: location.workspaceID } : {}), + }, + ...(sessions && sessions.length > 0 ? { sessions } : {}), + }) + }) + onCleanup( client.event.listen(({ details }) => { if (details.type === "server.connected") { diff --git a/packages/tui/src/context/event-interest.ts b/packages/tui/src/context/event-interest.ts new file mode 100644 index 000000000000..cd40ec7c97f2 --- /dev/null +++ b/packages/tui/src/context/event-interest.ts @@ -0,0 +1,30 @@ +export type EventInterest = { + readonly location?: { + readonly directory: string + readonly workspace?: string + } + readonly sessions?: ReadonlyArray +} + +export function interestEqual(left?: EventInterest, right?: EventInterest) { + if (left === right) return true + if (!left || !right) return false + if (left.location?.directory !== right.location?.directory) return false + if (left.location?.workspace !== right.location?.workspace) return false + const a = left.sessions ?? [] + const b = right.sessions ?? [] + if (a.length !== b.length) return false + for (let i = 0; i < a.length; i++) if (a[i] !== b[i]) return false + return true +} + +/** Map TUI interest into the generated `event.subscribe` query shape. */ +export function subscribeInput(interest?: EventInterest) { + if (!interest) return undefined + const location = interest.location + const session = interest.sessions && interest.sessions.length > 0 ? [...interest.sessions] : undefined + return { + ...(location ? { location } : {}), + ...(session ? { session } : {}), + } +} diff --git a/packages/tui/test/cli/tui/client-ownership.test.tsx b/packages/tui/test/cli/tui/client-ownership.test.tsx new file mode 100644 index 000000000000..30fab7f5cf08 --- /dev/null +++ b/packages/tui/test/cli/tui/client-ownership.test.tsx @@ -0,0 +1,177 @@ +/** @jsxImportSource @opentui/solid */ +import { describe, expect, test } from "bun:test" +import type { OpenCodeEvent } from "@opencode-ai/client" +import { testRender } from "@opentui/solid" +import { onMount } from "solid-js" +import { ClientProvider, useClient } from "../../../src/context/client" +import { createApi, createEventStream, createFetch, directory } from "../../fixture/tui-client" +import { TestTuiContexts } from "../../fixture/tui-environment" + +async function wait(fn: () => boolean, timeout = 3000) { + const start = Date.now() + while (!fn()) { + if (Date.now() - start > timeout) throw new Error("timed out waiting for condition") + await Bun.sleep(10) + } +} + +function trackStreams(events: ReturnType) { + let concurrent = 0 + let maxConcurrent = 0 + const calls = createFetch((url, request) => { + if (url.pathname !== "/api/event") return undefined + concurrent += 1 + maxConcurrent = Math.max(maxConcurrent, concurrent) + const signal = request.signal + const response = events.v2(signal) + const onAbort = () => { + concurrent = Math.max(0, concurrent - 1) + signal.removeEventListener("abort", onAbort) + } + if (signal.aborted) onAbort() + else signal.addEventListener("abort", onAbort, { once: true }) + return response + }, events) + return { + fetch: calls.fetch, + get concurrent() { + return concurrent + }, + get maxConcurrent() { + return maxConcurrent + }, + } +} + +async function mountClient(input: { + fetch: typeof globalThis.fetch + onReady?: (client: ReturnType) => void +}) { + let client!: ReturnType + let done!: () => void + const ready = new Promise((resolve) => { + done = resolve + }) + + const app = await testRender(() => ( + + + { + client = ctx + input.onReady?.(ctx) + done() + }} + /> + + + )) + + await ready + return { app, client } +} + +async function mount(input?: { onReady?: (client: ReturnType) => void }) { + const events = createEventStream() + const { app, client } = await mountClient({ + fetch: createFetch((url, request) => { + if (url.pathname !== "/api/event") return undefined + return events.v2(request.signal) + }, events).fetch, + onReady: input?.onReady, + }) + return { app, client, events } +} + +async function mountTracked(input?: { onReady?: (client: ReturnType) => void }) { + const events = createEventStream() + const streams = trackStreams(events) + const { app, client } = await mountClient({ + fetch: streams.fetch, + onReady: input?.onReady, + }) + return { app, client, events, streams } +} + +function connected(client: ReturnType) { + return () => client.connection.status() === "connected" +} + +function connectedAlone(client: ReturnType, streams: { concurrent: number }) { + return () => client.connection.status() === "connected" && streams.concurrent === 1 +} + +describe("ClientProvider stream ownership", () => { + test("rapid interest changes never exceed one active event stream", async () => { + const { app, client, streams } = await mountTracked() + + try { + await wait(connectedAlone(client, streams)) + + for (let i = 0; i < 8; i++) { + client.event.scope({ location: { directory }, sessions: [`ses_${i}`] }) + } + await wait(connectedAlone(client, streams)) + expect(streams.maxConcurrent).toBeLessThanOrEqual(2) + expect(streams.concurrent).toBe(1) + expect(client.connection.internal.generation()).toBeGreaterThanOrEqual(8) + } finally { + app.renderer.destroy() + } + await Bun.sleep(20) + expect(streams.concurrent).toBe(0) + }) + + test("event IDs from a prior generation are not delivered after interest reconnect", async () => { + const seen: Array<{ id: string; generation: number }> = [] + const { app, client, events } = await mount({ + onReady: (ctx) => { + ctx.event.on("vcs.branch.updated", (event) => { + seen.push({ id: event.id, generation: ctx.connection.internal.generation() }) + }) + }, + }) + + try { + await wait(connected(client)) + const generation = client.connection.internal.generation() + events.emit(vcs("before")) + await wait(() => seen.some((entry) => entry.id === "evt_vcs_before")) + + client.event.scope({ location: { directory }, sessions: ["ses_next"] }) + await wait(() => connected(client)() && client.connection.internal.generation() > generation) + const next = client.connection.internal.generation() + events.emit(vcs("after")) + await wait(() => seen.some((entry) => entry.id === "evt_vcs_after")) + + expect(seen.filter((entry) => entry.id === "evt_vcs_before")).toEqual([{ id: "evt_vcs_before", generation }]) + expect(seen.filter((entry) => entry.id === "evt_vcs_after")).toEqual([{ id: "evt_vcs_after", generation: next }]) + } finally { + app.renderer.destroy() + } + }) + + test("teardown leaves zero active event streams", async () => { + const { app, client, streams } = await mountTracked() + + await wait(connectedAlone(client, streams)) + app.renderer.destroy() + await wait(() => streams.concurrent === 0) + expect(streams.concurrent).toBe(0) + }) +}) + +function Probe(props: { onReady: (client: ReturnType) => void }) { + const client = useClient() + onMount(() => props.onReady(client)) + return +} + +function vcs(branch: string): OpenCodeEvent { + return { + id: `evt_vcs_${branch}`, + created: 0, + type: "vcs.branch.updated", + data: { branch }, + } +} diff --git a/packages/tui/test/cli/tui/client-scope.test.tsx b/packages/tui/test/cli/tui/client-scope.test.tsx new file mode 100644 index 000000000000..e47cfb4944f1 --- /dev/null +++ b/packages/tui/test/cli/tui/client-scope.test.tsx @@ -0,0 +1,97 @@ +/** @jsxImportSource @opentui/solid */ +import { describe, expect, test } from "bun:test" +import { testRender } from "@opentui/solid" +import { onMount } from "solid-js" +import { ClientProvider, useClient } from "../../../src/context/client" +import { createApi, createEventStream, createFetch, directory } from "../../fixture/tui-client" +import { TestTuiContexts } from "../../fixture/tui-environment" + +async function wait(fn: () => boolean, timeout = 2000) { + const start = Date.now() + while (!fn()) { + if (Date.now() - start > timeout) throw new Error("timed out waiting for condition") + await Bun.sleep(10) + } +} + +async function mount(interest: { location: { directory: string }; sessions?: string[] }) { + const events = createEventStream() + const eventUrls: URL[] = [] + const calls = createFetch((url, request) => { + if (url.pathname !== "/api/event") return undefined + eventUrls.push(url) + return events.v2(request.signal) + }, events) + + let client!: ReturnType + let done!: () => void + const ready = new Promise((resolve) => { + done = resolve + }) + + const app = await testRender(() => ( + + + { + client = ctx + done() + }} + /> + + + )) + + await ready + return { app, client, eventUrls } +} + +describe("ClientProvider event.scope", () => { + test("connects the event stream with location interest", async () => { + const { app, client, eventUrls } = await mount({ location: { directory } }) + try { + await wait(() => client.connection.status() === "connected" && eventUrls.length >= 1) + expect(eventUrls[0]?.searchParams.get("location[directory]")).toBe(directory) + expect(eventUrls[0]?.searchParams.getAll("session")).toEqual([]) + } finally { + app.renderer.destroy() + } + }) + + test("reconnects when session interest changes", async () => { + const { app, client, eventUrls } = await mount({ location: { directory } }) + try { + await wait(() => client.connection.status() === "connected" && eventUrls.length >= 1) + + client.event.scope({ location: { directory }, sessions: ["ses_a"] }) + await wait(() => eventUrls.length >= 2) + expect(eventUrls[1]?.searchParams.get("location[directory]")).toBe(directory) + expect(eventUrls[1]?.searchParams.getAll("session")).toEqual(["ses_a"]) + } finally { + app.renderer.destroy() + } + }) + + test("does not reconnect when interest is unchanged", async () => { + const { app, client, eventUrls } = await mount({ + location: { directory }, + sessions: ["ses_a"], + }) + try { + await wait(() => client.connection.status() === "connected" && eventUrls.length >= 1) + const before = eventUrls.length + + client.event.scope({ location: { directory }, sessions: ["ses_a"] }) + await Bun.sleep(50) + expect(eventUrls.length).toBe(before) + } finally { + app.renderer.destroy() + } + }) +}) + +function Probe(props: { onReady: (client: ReturnType) => void }) { + const client = useClient() + onMount(() => props.onReady(client)) + return +} diff --git a/packages/tui/test/context/client-interest.test.ts b/packages/tui/test/context/client-interest.test.ts new file mode 100644 index 000000000000..b34a4a53f762 --- /dev/null +++ b/packages/tui/test/context/client-interest.test.ts @@ -0,0 +1,35 @@ +import { expect, test } from "bun:test" +import { interestEqual, subscribeInput } from "../../src/context/event-interest" + +test("subscribeInput omits empty sessions and maps sessions to session", () => { + expect(subscribeInput(undefined)).toBeUndefined() + expect(subscribeInput({})).toEqual({}) + expect(subscribeInput({ location: { directory: "/tmp/project" }, sessions: [] })).toEqual({ + location: { directory: "/tmp/project" }, + }) + expect( + subscribeInput({ + location: { directory: "/tmp/project", workspace: "ws_1" }, + sessions: ["ses_a", "ses_b"], + }), + ).toEqual({ + location: { directory: "/tmp/project", workspace: "ws_1" }, + session: ["ses_a", "ses_b"], + }) +}) + +test("interestEqual compares location and session sets", () => { + const base = { location: { directory: "/tmp/project" }, sessions: ["ses_a"] } + expect(interestEqual(base, base)).toBe(true) + expect(interestEqual(base, { location: { directory: "/tmp/project" }, sessions: ["ses_a"] })).toBe(true) + expect(interestEqual(base, { location: { directory: "/tmp/other" }, sessions: ["ses_a"] })).toBe(false) + expect( + interestEqual(base, { location: { directory: "/tmp/project", workspace: "ws_1" }, sessions: ["ses_a"] }), + ).toBe(false) + expect(interestEqual(base, { location: { directory: "/tmp/project" }, sessions: ["ses_b"] })).toBe(false) + expect(interestEqual({ location: { directory: "/tmp/project" } }, { location: { directory: "/tmp/project" } })).toBe( + true, + ) + expect(interestEqual(undefined, undefined)).toBe(true) + expect(interestEqual(base, undefined)).toBe(false) +}) diff --git a/packages/tui/test/fixture/tui-client.ts b/packages/tui/test/fixture/tui-client.ts index e60f79e964fd..68d35c0d8c92 100644 --- a/packages/tui/test/fixture/tui-client.ts +++ b/packages/tui/test/fixture/tui-client.ts @@ -20,22 +20,31 @@ export function createEventStream() { controllers: Set>, queued: Uint8Array[], initial?: unknown, + signal?: AbortSignal, ) => { let current: ReadableStreamDefaultController | undefined - return new Response( - new ReadableStream({ - start(controller) { - current = controller - controllers.add(controller) - if (initial) controller.enqueue(encoder.encode(`data: ${JSON.stringify(initial)}\n\n`)) - for (const chunk of queued.splice(0)) controller.enqueue(chunk) - }, - cancel() { - if (current) controllers.delete(current) - }, - }), - { headers: { "content-type": "text/event-stream" } }, - ) + const stream = new ReadableStream({ + start(controller) { + current = controller + controllers.add(controller) + if (initial) controller.enqueue(encoder.encode(`data: ${JSON.stringify(initial)}\n\n`)) + for (const chunk of queued.splice(0)) controller.enqueue(chunk) + }, + cancel() { + if (current) controllers.delete(current) + }, + }) + const onAbort = () => { + try { + current?.error(signal?.reason ?? new DOMException("Aborted", "AbortError")) + } catch { + // already closed + } + if (current) controllers.delete(current) + } + if (signal?.aborted) onAbort() + else signal?.addEventListener("abort", onAbort, { once: true }) + return new Response(stream, { headers: { "content-type": "text/event-stream" } }) } const send = ( controllers: Set>, @@ -54,8 +63,8 @@ export function createEventStream() { emit(event: OpenCodeEvent) { send(v2, pending, event) }, - v2() { - return response(v2, pending, { id: "evt_connected", type: "server.connected", data: {} }) + v2(signal?: AbortSignal) { + return response(v2, pending, { id: "evt_connected", type: "server.connected", data: {} }, signal) }, disconnect() { for (const controller of v2) controller.close() @@ -74,7 +83,7 @@ export function createFetch(override?: FetchHandler, events?: ReturnType - {props.children} + + {props.children} + diff --git a/specs/v2/event-stream-architecture.md b/specs/v2/event-stream-architecture.md index 34eb8f137cc1..887998311619 100644 --- a/specs/v2/event-stream-architecture.md +++ b/specs/v2/event-stream-architecture.md @@ -42,7 +42,7 @@ With `N` connected TUIs, schema and wire encoding therefore ran `N` times for ev `liveBounded` was introduced before zero-argument `events.subscribe()` became the unified live interface. It stayed on deprecated `listen` because it provided a stronger contract than the shared unbounded Core PubSub: one slow subscriber could overflow and fail without blocking healthy subscribers. -The global cross-location event stream is intentional. The endpoint is outside `LocationMiddleware`, and the TUI uses event location metadata to update state for multiple locations. The feed must not add request-location filtering. +`GET /api/event` stays outside `LocationMiddleware` so omitting query params preserves the global public feed for unscoped clients. Callers may opt in to location interest via `location[directory]` / `location[workspace]` and optional repeated `session` query params. The Server feed filters matched subscribers before queue offer while still encoding each accepted public event once. Session interest narrows events that carry a session identity (`durable.aggregateID` or `data.sessionID`); location-scoped events without a session still deliver. Installation lifecycle events (`installation.updated`, `installation.update-available`) and `global.disposed` always pass interest filters. Changing interest requires reconnecting with a new query rather than overlapping streams. ## Delivery Law