From 3eac0eb397b4fe5a15552d09b5e9e24de57227f6 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 21:58:33 +1000 Subject: [PATCH 01/21] feat(server): opt-in location interest for event subscriptions Co-authored-by: Cursor --- packages/cli/src/mini/noninteractive.ts | 2 +- packages/cli/src/mini/stream-v2.transport.ts | 2 +- packages/client/src/effect/api/api.ts | 4 +- .../client/src/effect/generated/client.ts | 6 +- .../client/src/promise/generated/client.ts | 12 +- .../client/src/promise/generated/types.ts | 6 + packages/protocol/src/groups/event.ts | 20 ++-- packages/server/src/event-feed.ts | 83 +++++++++++-- packages/server/src/handlers/event.ts | 7 +- packages/server/test/event-feed.test.ts | 113 ++++++++++++++++-- packages/tui/src/app.tsx | 6 +- packages/tui/src/context/client.tsx | 13 +- specs/v2/event-stream-architecture.md | 2 +- 13 files changed, 234 insertions(+), 42 deletions(-) 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..7ba85854948c 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -767,8 +767,10 @@ export interface SkillApi { readonly list: SkillListOperation } +type Endpoint19_0Request = Parameters[0] +export type Endpoint19_0Input = { readonly location?: Endpoint19_0Request["query"]["location"] } export type Endpoint19_0Output = StreamValue>> -export type EventSubscribeOperation = () => Stream.Stream +export type EventSubscribeOperation = (input?: Endpoint19_0Input) => Stream.Stream export interface EventApi { readonly subscribe: EventSubscribeOperation diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index c7a2f860353b..d0b361880fa7 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -908,9 +908,11 @@ 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"] } +const Endpoint19_0 = (raw: RawClient["server.event"]) => (input?: Endpoint19_0Input) => Stream.unwrap( - raw["event.subscribe"]({}).pipe( + raw["event.subscribe"]({ query: { location: input?.["location"] } }).pipe( Effect.mapError(mapClientError), Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))), ), diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 3783a5e5c011..3e93a98a1d13 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, @@ -1350,9 +1351,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"] }, + 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..e0688f44293f 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -4542,6 +4542,12 @@ export type SkillListOutput = { data: Array } +export type EventSubscribeInput = { + readonly location?: { + readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined + }["location"] +} + export type EventSubscribeOutput = V2Event export type PtyListInput = { diff --git a/packages/protocol/src/groups/event.ts b/packages/protocol/src/groups/event.ts index 242aa745ef53..c05819e77f1c 100644 --- a/packages/protocol/src/groups/event.ts +++ b/packages/protocol/src/groups/event.ts @@ -4,6 +4,7 @@ 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 { LocationQuery, locationQueryOpenApi } from "./location.js" const fields = { id: Event.ID, @@ -32,15 +33,18 @@ const make = >(definitions: group: HttpApiGroup.make("server.event") .add( HttpApiEndpoint.get("event.subscribe", "/api/event", { + query: LocationQuery, 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 location to receive the global public feed; pass location to narrow delivery to that directory (and optional workspace). 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/server/src/event-feed.ts b/packages/server/src/event-feed.ts index 7b668df5b42a..8c1bcfaf793a 100644 --- a/packages/server/src/event-feed.ts +++ b/packages/server/src/event-feed.ts @@ -6,6 +6,13 @@ import { Cause, Context, Effect, Layer, Queue, Schema, Scope, Stream } from "eff export const SubscriberCapacity = 4_096 +/** Core-published types that always pass location interest filters. */ +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 +26,18 @@ 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 to narrow delivery. */ +export type Interest = { + readonly location?: LocationInterest +} + export interface Interface { - readonly subscribe: Effect.Effect, never, Scope.Scope> + readonly subscribe: (interest?: Interest) => Effect.Effect, never, Scope.Scope> } export class Service extends Context.Service()("@opencode/server/EventFeed") {} @@ -31,24 +48,56 @@ export function frame(event: OpenCodeEvent) { return `data: ${JSON.stringify(encode(event))}\n\n` } +export function matchesInterest(event: EventV2.Payload, interest?: Interest): boolean { + const location = interest?.location + if (location === undefined) return true + if (GlobalEventTypes.has(event.type)) return true + 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 + return true +} + +export function interestFromQuery(query: URLSearchParams): Interest | undefined { + const directory = query.get("location[directory]") ?? undefined + const workspace = query.get("location[workspace]") ?? undefined + if (directory === undefined && workspace === undefined) return undefined + if (directory === undefined) return undefined + return { location: { directory, ...(workspace !== undefined ? { workspace } : {}) } } +} + +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>() 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 + let matched = false + for (const subscriber of subscribers.values()) { + if (matchesInterest(event, subscriber.interest)) { + matched = true + break + } + } + if (!matched) return const encoded = yield* Effect.try({ try: () => render(event), catch: (cause) => new EncodingError({ eventID: event.id, eventType: event.type, cause }), @@ -62,10 +111,11 @@ 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 }))) + for (const subscriber of Array.from(subscribers.values())) { + if (!matchesInterest(event, subscriber.interest)) continue + if (Queue.offerUnsafe(subscriber.queue, encoded)) continue + subscribers.delete(subscriber.queue) + Queue.failCauseUnsafe(subscriber.queue, Cause.fail(new SubscriberOverflowError({ capacity }))) } }) @@ -73,11 +123,18 @@ 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)), + subscribe: (interest) => + Effect.acquireRelease( + Queue.dropping(capacity).pipe( + Effect.tap((queue) => + Effect.sync(() => { + subscribers.set(queue, { queue, interest }) + }), + ), + ), + (queue) => + Effect.sync(() => subscribers.delete(queue)).pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid), + ).pipe(Effect.map(Stream.fromQueue)), }) }) diff --git a/packages/server/src/handlers/event.ts b/packages/server/src/handlers/event.ts index feeba74288d6..2bf69f43b102 100644 --- a/packages/server/src/handlers/event.ts +++ b/packages/server/src/handlers/event.ts @@ -8,15 +8,18 @@ import { EventFeed } from "../event-feed" export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) => Effect.gen(function* () { const feed = yield* EventFeed.Service - return handlers.handleRaw("event.subscribe", () => + return handlers.handleRaw("event.subscribe", (ctx) => Effect.gen(function* () { + const interest = EventFeed.interestFromQuery(new URL(ctx.request.url, "http://localhost").searchParams) 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)))), + 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( diff --git a/packages/server/test/event-feed.test.ts b/packages/server/test/event-feed.test.ts index 29129c735c2f..ffc63c251db0 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,22 @@ 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; workspaceID?: 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), + ...(location.workspaceID !== undefined ? { workspaceID: location.workspaceID as never } : {}), + }, + } + : {}), }) const internal = (value: string): EventV2.Payload => ({ @@ -54,8 +66,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 +88,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 +129,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 +148,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 +164,89 @@ 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() + }) }) diff --git a/packages/tui/src/app.tsx b/packages/tui/src/app.tsx index a0b2e3fc0be0..e98a5496d023 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..f95c77f09a6e 100644 --- a/packages/tui/src/context/client.tsx +++ b/packages/tui/src/context/client.tsx @@ -22,13 +22,20 @@ type ManagedService = { restart: () => Promise } +export type EventInterest = { + readonly location?: { + readonly directory: string + readonly workspace?: string + } +} + type ClientEventMap = { [Type in OpenCodeEvent["type"]]: Extract } const connectTimeout = 2_000 const connectionHistoryLimit = 50 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[] = [] @@ -64,7 +71,9 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( 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]() + const iterator = api.event + .subscribe(props.interest, { signal: request.signal }) + [Symbol.asyncIterator]() const first = await iterator.next() if (abort.signal.aborted || controller.signal.aborted) return undefined if (first.done) diff --git a/specs/v2/event-stream-architecture.md b/specs/v2/event-stream-architecture.md index 34eb8f137cc1..e1fa092abbf1 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]` query params. The Server feed filters matched subscribers before queue offer while still encoding each accepted public event once. Installation lifecycle events (`installation.updated`, `installation.update-available`) and `global.disposed` always pass location filters. Changing interest requires reconnecting with a new query rather than overlapping streams. ## Delivery Law From 8c92fcc3b2ee16cd14818c0d474af04cf037db3b Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 22:03:23 +1000 Subject: [PATCH 02/21] feat(server): narrow event subscriptions by session interest Co-authored-by: Cursor --- packages/client/src/effect/api/api.ts | 5 +- .../client/src/effect/generated/client.ts | 7 +- .../client/src/promise/generated/client.ts | 2 +- .../client/src/promise/generated/types.ts | 5 + packages/protocol/src/groups/event.ts | 18 ++- packages/schema/src/event.ts | 13 +++ packages/schema/test/event.test.ts | 8 ++ packages/server/src/event-feed.ts | 42 +++++-- packages/server/test/event-feed.test.ts | 104 ++++++++++++++++++ packages/tui/src/context/client.tsx | 29 ++++- packages/tui/src/context/data.tsx | 18 +++ packages/tui/test/fixture/tui-environment.tsx | 5 +- specs/v2/event-stream-architecture.md | 2 +- 13 files changed, 237 insertions(+), 21 deletions(-) diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 7ba85854948c..8c947bcd347f 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -768,7 +768,10 @@ export interface SkillApi { } type Endpoint19_0Request = Parameters[0] -export type Endpoint19_0Input = { readonly location?: Endpoint19_0Request["query"]["location"] } +export type Endpoint19_0Input = { + readonly location?: Endpoint19_0Request["query"]["location"] + readonly session?: Endpoint19_0Request["query"]["session"] +} export type Endpoint19_0Output = StreamValue>> export type EventSubscribeOperation = (input?: Endpoint19_0Input) => Stream.Stream diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index d0b361880fa7..89bccfc8e25d 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -909,10 +909,13 @@ const Endpoint18_0 = (raw: RawClient["server.skill"]) => (input?: Endpoint18_0In const adaptGroup18 = (raw: RawClient["server.skill"]) => ({ list: Endpoint18_0(raw) }) type Endpoint19_0Request = Parameters[0] -type Endpoint19_0Input = { readonly location?: Endpoint19_0Request["query"]["location"] } +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"]({ query: { location: input?.["location"] } }).pipe( + raw["event.subscribe"]({ query: { location: input?.["location"], session: input?.["session"] } }).pipe( Effect.mapError(mapClientError), Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))), ), diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 3e93a98a1d13..1bb9a83acff7 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -1356,7 +1356,7 @@ export function make(options: ClientOptions) { { method: "GET", path: `/api/event`, - query: { location: input?.["location"] }, + query: { location: input?.["location"], session: input?.["session"] }, successStatus: 200, declaredStatuses: [401, 400], empty: false, diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index e0688f44293f..01909486b873 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -4545,7 +4545,12 @@ export type SkillListOutput = { export type EventSubscribeInput = { readonly location?: { readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined + readonly session?: string | ReadonlyArray | ReadonlyArray | undefined }["location"] + readonly session?: { + readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined + readonly session?: string | ReadonlyArray | ReadonlyArray | undefined + }["session"] } export type EventSubscribeOutput = V2Event diff --git a/packages/protocol/src/groups/event.ts b/packages/protocol/src/groups/event.ts index c05819e77f1c..199eba8fd051 100644 --- a/packages/protocol/src/groups/event.ts +++ b/packages/protocol/src/groups/event.ts @@ -4,7 +4,19 @@ 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 { LocationQuery, locationQueryOpenApi } from "./location.js" +import { locationQueryOpenApi } from "./location.js" + +export const sessionIDOf = Event.sessionIDOf + +export const EventSubscribeQuery = Schema.Struct({ + location: Schema.optional( + Schema.Struct({ + directory: Schema.optional(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, @@ -33,7 +45,7 @@ const make = >(definitions: group: HttpApiGroup.make("server.event") .add( HttpApiEndpoint.get("event.subscribe", "/api/event", { - query: LocationQuery, + query: EventSubscribeQuery, success: HttpApiSchema.StreamSse({ data: EventSchema }), }) .annotateMerge(locationQueryOpenApi) @@ -42,7 +54,7 @@ const make = >(definitions: identifier: "v2.event.subscribe", summary: "Subscribe to events", description: - "Subscribe to native event payloads for the server. Omit location to receive the global public feed; pass location to narrow delivery to that directory (and optional workspace). Volatile by contract: a slow consumer overflows and fails the stream, and events during disconnection are missed.", + "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.", }), ), ) diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index e4f2f9b8139f..4a61c024d946 100644 --- a/packages/schema/src/event.ts +++ b/packages/schema/src/event.ts @@ -196,3 +196,16 @@ function readonlyMap(map: Map): ReadonlyMap }) return result } + +/** 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 + if (event.data !== undefined && event.data !== null && typeof event.data === "object" && "sessionID" in event.data) { + const sessionID = (event.data as { readonly sessionID?: unknown }).sessionID + if (typeof sessionID === "string") return sessionID + } + return undefined +} diff --git a/packages/schema/test/event.test.ts b/packages/schema/test/event.test.ts index 404b49aa743c..62ffbb9e797d 100644 --- a/packages/schema/test/event.test.ts +++ b/packages/schema/test/event.test.ts @@ -49,4 +49,12 @@ 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() + }) }) diff --git a/packages/server/src/event-feed.ts b/packages/server/src/event-feed.ts index 8c1bcfaf793a..acb6ed110e49 100644 --- a/packages/server/src/event-feed.ts +++ b/packages/server/src/event-feed.ts @@ -1,7 +1,7 @@ export * as EventFeed from "./event-feed" import { EventV2 } from "@opencode-ai/core/event" -import { isOpenCodeEvent, OpenCodeEvent } from "@opencode-ai/protocol/groups/event" +import { isOpenCodeEvent, OpenCodeEvent, sessionIDOf } from "@opencode-ai/protocol/groups/event" import { Cause, Context, Effect, Layer, Queue, Schema, Scope, Stream } from "effect" export const SubscriberCapacity = 4_096 @@ -31,9 +31,10 @@ export type LocationInterest = { readonly workspace?: string } -/** Omit or leave empty for the global public feed; set location to narrow delivery. */ +/** Omit or leave empty for the global public feed; set location/sessions to narrow delivery. */ export type Interest = { readonly location?: LocationInterest + readonly sessions?: ReadonlyArray } export interface Interface { @@ -49,22 +50,41 @@ export function frame(event: OpenCodeEvent) { } export function matchesInterest(event: EventV2.Payload, interest?: Interest): boolean { - const location = interest?.location - if (location === undefined) return true + if (interest === undefined) return true if (GlobalEventTypes.has(event.type)) return true - 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 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 } export function interestFromQuery(query: URLSearchParams): Interest | undefined { const directory = query.get("location[directory]") ?? undefined const workspace = query.get("location[workspace]") ?? undefined - if (directory === undefined && workspace === undefined) return undefined - if (directory === undefined) return undefined - return { location: { directory, ...(workspace !== undefined ? { workspace } : {}) } } + const sessions = query + .getAll("session") + .flatMap((value) => value.split(",")) + .map((value) => value.trim()) + .filter((value) => value.length > 0) + if (directory === undefined && workspace === undefined && sessions.length === 0) return undefined + return { + ...(directory !== undefined + ? { location: { directory, ...(workspace !== undefined ? { workspace } : {}) } } + : {}), + ...(sessions.length > 0 ? { sessions } : {}), + } } type Subscriber = { diff --git a/packages/server/test/event-feed.test.ts b/packages/server/test/event-feed.test.ts index ffc63c251db0..79509af39aa7 100644 --- a/packages/server/test/event-feed.test.ts +++ b/packages/server/test/event-feed.test.ts @@ -249,4 +249,108 @@ describe("EventFeed", () => { }) 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"], + }) + }) + + 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]) + }), + ) +}) + +const sessionEvent = ( + id: string, + sessionID: string, + location: { directory: string; workspaceID?: 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), + ...(location.workspaceID !== undefined ? { workspaceID: location.workspaceID as never } : {}), + }, }) diff --git a/packages/tui/src/context/client.tsx b/packages/tui/src/context/client.tsx index f95c77f09a6e..0004b926be92 100644 --- a/packages/tui/src/context/client.tsx +++ b/packages/tui/src/context/client.tsx @@ -27,12 +27,33 @@ export type EventInterest = { readonly directory: string readonly workspace?: string } + readonly sessions?: ReadonlyArray } type ClientEventMap = { [Type in OpenCodeEvent["type"]]: Extract } const connectTimeout = 2_000 const connectionHistoryLimit = 50 +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 +} + +function subscribeInput(interest?: EventInterest) { + if (!interest) return undefined + return { + ...(interest.location ? { location: interest.location } : {}), + ...(interest.sessions && interest.sessions.length > 0 ? { session: [...interest.sessions] } : {}), + } +} + export const { use: useClient, provider: ClientProvider } = createSimpleContext({ name: "Client", init: (props: { api: OpenCodeClient; service?: ManagedService; interest?: EventInterest }) => { @@ -40,6 +61,7 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( const abort = new AbortController() const history: ClientConnectionEvent[] = [] let api = props.api + let interest = props.interest const events = createGlobalEmitter() const [connection, setConnection] = createStore<{ status: ClientConnectionStatus @@ -72,7 +94,7 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( record(attempt === 0 ? "connecting" : "reconnecting", attempt) log.info("event stream connecting", { attempt }) const iterator = api.event - .subscribe(props.interest, { signal: request.signal }) + .subscribe(subscribeInput(interest), { signal: request.signal }) [Symbol.asyncIterator]() const first = await iterator.next() if (abort.signal.aborted || controller.signal.aborted) return undefined @@ -154,6 +176,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() { 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/test/fixture/tui-environment.tsx b/packages/tui/test/fixture/tui-environment.tsx index e2ef9ba77450..44ad19a2ad59 100644 --- a/packages/tui/test/fixture/tui-environment.tsx +++ b/packages/tui/test/fixture/tui-environment.tsx @@ -7,6 +7,7 @@ import { } from "../../src/context/runtime" import type { ParentProps } from "solid-js" import { LogProvider, type LogSink } from "../../src/context/log" +import { RouteProvider } from "../../src/context/route" export function TestTuiContexts( props: ParentProps<{ @@ -28,7 +29,9 @@ export function TestTuiContexts( }} > - {props.children} + + {props.children} + diff --git a/specs/v2/event-stream-architecture.md b/specs/v2/event-stream-architecture.md index e1fa092abbf1..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. -`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]` query params. The Server feed filters matched subscribers before queue offer while still encoding each accepted public event once. Installation lifecycle events (`installation.updated`, `installation.update-available`) and `global.disposed` always pass location filters. Changing interest requires reconnecting with a new query rather than overlapping streams. +`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 From 7bfcc92cd14e4838683e1258ccce1ae71cd1e278 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 22:33:03 +1000 Subject: [PATCH 03/21] style(tui): alphabetize ClientProvider props Co-authored-by: Cursor --- packages/tui/src/app.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/tui/src/app.tsx b/packages/tui/src/app.tsx index e98a5496d023..e440600c34a4 100644 --- a/packages/tui/src/app.tsx +++ b/packages/tui/src/app.tsx @@ -336,8 +336,8 @@ export const run = Effect.fn("Tui.run")(function* (input: TuiInput) { From 7674f994e73854d3ebbbdca9250d5064c88cf314 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 22:33:03 +1000 Subject: [PATCH 04/21] style(tui): alphabetize ClientProvider props Co-authored-by: Cursor --- packages/tui/src/app.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/tui/src/app.tsx b/packages/tui/src/app.tsx index e98a5496d023..e440600c34a4 100644 --- a/packages/tui/src/app.tsx +++ b/packages/tui/src/app.tsx @@ -336,8 +336,8 @@ export const run = Effect.fn("Tui.run")(function* (input: TuiInput) { From 162a75f99cc10925fe2922b86940b72a49dbb60b Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 22:59:19 +1000 Subject: [PATCH 05/21] refactor(schema): drop cast in sessionIDOf Narrow unknown event data with object/`in` checks so sessionID is read without an assertion. Co-authored-by: Cursor --- packages/schema/src/event.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index 4a61c024d946..989413a4e8a7 100644 --- a/packages/schema/src/event.ts +++ b/packages/schema/src/event.ts @@ -203,9 +203,10 @@ export function sessionIDOf(event: { readonly data?: unknown }): string | undefined { if (event.durable?.aggregateID) return event.durable.aggregateID - if (event.data !== undefined && event.data !== null && typeof event.data === "object" && "sessionID" in event.data) { - const sessionID = (event.data as { readonly sessionID?: unknown }).sessionID - if (typeof sessionID === "string") return sessionID - } + const data = event.data + if (data === undefined || data === null || typeof data !== "object") return undefined + if (!("sessionID" in data)) return undefined + const sessionID = data.sessionID + if (typeof sessionID === "string") return sessionID return undefined } From 6f87215d7a0bb350ad2f9d93d585e9967c00dbb1 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 23:03:46 +1000 Subject: [PATCH 06/21] test(server): drop unused workspaceID from event fixtures Helpers only need directory; no test supplies a workspace. Co-authored-by: Cursor --- packages/server/test/event-feed.test.ts | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/packages/server/test/event-feed.test.ts b/packages/server/test/event-feed.test.ts index 79509af39aa7..ad3042282cf4 100644 --- a/packages/server/test/event-feed.test.ts +++ b/packages/server/test/event-feed.test.ts @@ -11,7 +11,7 @@ const Internal = EventV2.ephemeral({ type: "test.internal", schema: { value: Sch const event = ( id: string, - location?: { directory: string; workspaceID?: string }, + location?: { directory: string }, ): EventV2.Payload => ({ id: EventV2.ID.make(`evt_${id}`), created: DateTime.makeUnsafe(Date.now()), @@ -21,7 +21,6 @@ const event = ( ? { location: { directory: AbsolutePath.make(location.directory), - ...(location.workspaceID !== undefined ? { workspaceID: location.workspaceID as never } : {}), }, } : {}), @@ -342,7 +341,7 @@ describe("EventFeed", () => { const sessionEvent = ( id: string, sessionID: string, - location: { directory: string; workspaceID?: string }, + location: { directory: string }, ): EventV2.Payload => ({ id: EventV2.ID.make(`evt_${id}`), created: DateTime.makeUnsafe(Date.now()), @@ -351,6 +350,5 @@ const sessionEvent = ( durable: { aggregateID: sessionID, seq: EventV2.Seq.make(1), version: EventV2.Version.make(1) }, location: { directory: AbsolutePath.make(location.directory), - ...(location.workspaceID !== undefined ? { workspaceID: location.workspaceID as never } : {}), }, }) From 4b54724efca1b24cf986efa7de97ac9054992677 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 23:05:17 +1000 Subject: [PATCH 07/21] test(server): drop unused workspaceID from event fixture No test supplies a workspace; avoid an unnecessary brand cast. Co-authored-by: Cursor --- packages/server/test/event-feed.test.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/server/test/event-feed.test.ts b/packages/server/test/event-feed.test.ts index ffc63c251db0..24a016c0ab8f 100644 --- a/packages/server/test/event-feed.test.ts +++ b/packages/server/test/event-feed.test.ts @@ -11,7 +11,7 @@ const Internal = EventV2.ephemeral({ type: "test.internal", schema: { value: Sch const event = ( id: string, - location?: { directory: string; workspaceID?: string }, + location?: { directory: string }, ): EventV2.Payload => ({ id: EventV2.ID.make(`evt_${id}`), created: DateTime.makeUnsafe(Date.now()), @@ -21,7 +21,6 @@ const event = ( ? { location: { directory: AbsolutePath.make(location.directory), - ...(location.workspaceID !== undefined ? { workspaceID: location.workspaceID as never } : {}), }, } : {}), From f637389160e16a6b9447ea1bc0d201cb523a78a2 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 08:05:03 +1000 Subject: [PATCH 08/21] refactor(server): tighten event subscription interest matching Filter interested subscribers once before encode, require directory for location scope (reject workspace-only queries), and cover those cases in tests. Co-authored-by: Cursor --- .../client/src/promise/generated/types.ts | 4 +-- packages/protocol/src/groups/event.ts | 3 +- packages/server/src/event-feed.ts | 21 ++++++-------- packages/server/test/event-feed.test.ts | 29 +++++++++++++++++++ 4 files changed, 42 insertions(+), 15 deletions(-) diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 01909486b873..a52983cb98d9 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -4544,11 +4544,11 @@ export type SkillListOutput = { export type EventSubscribeInput = { readonly location?: { - readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined + readonly location?: { readonly directory: string; readonly workspace?: string | undefined } | undefined readonly session?: string | ReadonlyArray | ReadonlyArray | undefined }["location"] readonly session?: { - readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined + readonly location?: { readonly directory: string; readonly workspace?: string | undefined } | undefined readonly session?: string | ReadonlyArray | ReadonlyArray | undefined }["session"] } diff --git a/packages/protocol/src/groups/event.ts b/packages/protocol/src/groups/event.ts index 199eba8fd051..87fec115f8ce 100644 --- a/packages/protocol/src/groups/event.ts +++ b/packages/protocol/src/groups/event.ts @@ -11,7 +11,8 @@ export const sessionIDOf = Event.sessionIDOf export const EventSubscribeQuery = Schema.Struct({ location: Schema.optional( Schema.Struct({ - directory: Schema.optional(Schema.String), + /** Required when `location` is present; workspace alone is not a valid scope. */ + directory: Schema.String, workspace: Schema.optional(Schema.String), }), ), diff --git a/packages/server/src/event-feed.ts b/packages/server/src/event-feed.ts index acb6ed110e49..71f483d7fdad 100644 --- a/packages/server/src/event-feed.ts +++ b/packages/server/src/event-feed.ts @@ -6,7 +6,7 @@ import { Cause, Context, Effect, Layer, Queue, Schema, Scope, Stream } from "eff export const SubscriberCapacity = 4_096 -/** Core-published types that always pass location interest filters. */ +/** Core-published types that always pass interest filters (location and session). */ export const GlobalEventTypes = new Set([ "installation.updated", "installation.update-available", @@ -78,7 +78,9 @@ export function interestFromQuery(query: URLSearchParams): Interest | undefined .flatMap((value) => value.split(",")) .map((value) => value.trim()) .filter((value) => value.length > 0) - if (directory === undefined && workspace === undefined && sessions.length === 0) return undefined + // Workspace interest requires a directory; alone it is an invalid location scope. + if (directory === undefined && workspace !== undefined) return undefined + if (directory === undefined && sessions.length === 0) return undefined return { ...(directory !== undefined ? { location: { directory, ...(workspace !== undefined ? { workspace } : {}) } } @@ -110,14 +112,10 @@ export const make = Effect.fn("EventFeed.make")(function* ( const publish = Effect.fnUntraced(function* (event: EventV2.Payload) { if (!isOpenCodeEvent(event)) return if (subscribers.size === 0) return - let matched = false - for (const subscriber of subscribers.values()) { - if (matchesInterest(event, subscriber.interest)) { - matched = true - break - } - } - if (!matched) 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 }), @@ -131,8 +129,7 @@ export const make = Effect.fn("EventFeed.make")(function* ( ), ) if (encoded === undefined) return - for (const subscriber of Array.from(subscribers.values())) { - if (!matchesInterest(event, subscriber.interest)) continue + for (const subscriber of targets) { if (Queue.offerUnsafe(subscriber.queue, encoded)) continue subscribers.delete(subscriber.queue) Queue.failCauseUnsafe(subscriber.queue, Cause.fail(new SubscriberOverflowError({ capacity }))) diff --git a/packages/server/test/event-feed.test.ts b/packages/server/test/event-feed.test.ts index ad3042282cf4..8ae63f418c9e 100644 --- a/packages/server/test/event-feed.test.ts +++ b/packages/server/test/event-feed.test.ts @@ -258,6 +258,35 @@ describe("EventFeed", () => { }) }) + 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("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() From 51ff5f83dac3db84a26300ba17d62524955d0641 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 08:05:55 +1000 Subject: [PATCH 09/21] refactor(server): filter event-feed interest once before encode Collect matching subscribers first so encode/offer share one pass. Co-authored-by: Cursor --- packages/server/src/event-feed.ts | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/packages/server/src/event-feed.ts b/packages/server/src/event-feed.ts index 8c1bcfaf793a..34a58ea5e930 100644 --- a/packages/server/src/event-feed.ts +++ b/packages/server/src/event-feed.ts @@ -90,14 +90,10 @@ export const make = Effect.fn("EventFeed.make")(function* ( const publish = Effect.fnUntraced(function* (event: EventV2.Payload) { if (!isOpenCodeEvent(event)) return if (subscribers.size === 0) return - let matched = false - for (const subscriber of subscribers.values()) { - if (matchesInterest(event, subscriber.interest)) { - matched = true - break - } - } - if (!matched) 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 }), @@ -111,8 +107,7 @@ export const make = Effect.fn("EventFeed.make")(function* ( ), ) if (encoded === undefined) return - for (const subscriber of Array.from(subscribers.values())) { - if (!matchesInterest(event, subscriber.interest)) continue + for (const subscriber of targets) { if (Queue.offerUnsafe(subscriber.queue, encoded)) continue subscribers.delete(subscriber.queue) Queue.failCauseUnsafe(subscriber.queue, Cause.fail(new SubscriberOverflowError({ capacity }))) From 7444cfabbed33135c5a316ccc1af78326f4f490a Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 08:16:22 +1000 Subject: [PATCH 10/21] test(tui): lock event interest query shape and scope reconnect Extract pure interest helpers and cover subscribeInput/equality plus ClientProvider.event.scope reconnecting only when interest changes. Co-authored-by: Cursor --- packages/tui/src/context/client.tsx | 32 +------- packages/tui/src/context/event-interest.ts | 28 +++++++ .../tui/test/cli/tui/client-scope.test.tsx | 74 +++++++++++++++++++ .../tui/test/context/client-interest.test.ts | 35 +++++++++ 4 files changed, 141 insertions(+), 28 deletions(-) create mode 100644 packages/tui/src/context/event-interest.ts create mode 100644 packages/tui/test/cli/tui/client-scope.test.tsx create mode 100644 packages/tui/test/context/client-interest.test.ts diff --git a/packages/tui/src/context/client.tsx b/packages/tui/src/context/client.tsx index 0004b926be92..e043811aae7e 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 = { @@ -22,38 +26,10 @@ type ManagedService = { restart: () => Promise } -export type EventInterest = { - readonly location?: { - readonly directory: string - readonly workspace?: string - } - readonly sessions?: ReadonlyArray -} - type ClientEventMap = { [Type in OpenCodeEvent["type"]]: Extract } const connectTimeout = 2_000 const connectionHistoryLimit = 50 -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 -} - -function subscribeInput(interest?: EventInterest) { - if (!interest) return undefined - return { - ...(interest.location ? { location: interest.location } : {}), - ...(interest.sessions && interest.sessions.length > 0 ? { session: [...interest.sessions] } : {}), - } -} - export const { use: useClient, provider: ClientProvider } = createSimpleContext({ name: "Client", init: (props: { api: OpenCodeClient; service?: ManagedService; interest?: EventInterest }) => { diff --git a/packages/tui/src/context/event-interest.ts b/packages/tui/src/context/event-interest.ts new file mode 100644 index 000000000000..8db39edd171e --- /dev/null +++ b/packages/tui/src/context/event-interest.ts @@ -0,0 +1,28 @@ +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 + return { + ...(interest.location ? { location: interest.location } : {}), + ...(interest.sessions && interest.sessions.length > 0 ? { session: [...interest.sessions] } : {}), + } +} 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..3c2de5384d1b --- /dev/null +++ b/packages/tui/test/cli/tui/client-scope.test.tsx @@ -0,0 +1,74 @@ +/** @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) + } +} + +describe("ClientProvider event.scope", () => { + test("reconnects the event stream when interest changes, not when it is unchanged", async () => { + const events = createEventStream() + const eventUrls: URL[] = [] + const calls = createFetch((url) => { + if (url.pathname !== "/api/event") return undefined + eventUrls.push(url) + return events.v2() + }, events) + + let client!: ReturnType + let done!: () => void + const ready = new Promise((resolve) => { + done = resolve + }) + + const app = await testRender(() => ( + + + { + client = ctx + done() + }} + /> + + + )) + + try { + await ready + 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([]) + + 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"]) + + 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) +}) From 0d49a51a044e70766b1ced4066a58c1f755d2612 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 11:48:58 +1000 Subject: [PATCH 11/21] fix(server): enforce event stream ownership and diagnostics Serialize TUI subscribe generations, cancel SSE readers on abort, bridge Bun disconnects so EventFeed queues release, and expose content-free feed counters for leak detection. Co-authored-by: Cursor --- packages/client/src/effect/api/api.ts | 4 + .../client/src/effect/generated/client.ts | 4 + .../client/src/promise/generated/client.ts | 26 +++- .../client/src/promise/generated/types.ts | 9 ++ packages/httpapi-codegen/src/index.ts | 2 +- packages/protocol/src/groups/debug.ts | 21 +++ packages/server/src/event-feed.ts | 34 ++++- packages/server/src/handlers/debug.ts | 8 ++ packages/server/src/handlers/event.ts | 25 +++- .../server/test/event-feed-ownership.test.ts | 109 ++++++++++++++++ packages/server/test/event-feed.test.ts | 44 +++++++ packages/tui/src/context/client.tsx | 33 +++-- .../test/cli/tui/client-ownership.test.tsx | 122 ++++++++++++++++++ .../tui/test/cli/tui/client-scope.test.tsx | 4 +- packages/tui/test/fixture/tui-client.ts | 43 +++--- 15 files changed, 455 insertions(+), 33 deletions(-) create mode 100644 packages/server/test/event-feed-ownership.test.ts create mode 100644 packages/tui/test/cli/tui/client-ownership.test.tsx diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 8c947bcd347f..f5129de8ee72 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -997,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 89bccfc8e25d..ebc79d72823e 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -1186,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 1bb9a83acff7..2696350aa614 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -198,6 +198,7 @@ import type { DebugLocationListOutput, DebugLocationEvictInput, DebugLocationEvictOutput, + DebugEventFeedOutput, } from "./types" import { ClientError } from "./client-error" @@ -293,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 @@ -328,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 {} } }, }) @@ -1671,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 a52983cb98d9..3f3892f12126 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -4870,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/server/src/event-feed.ts b/packages/server/src/event-feed.ts index 71f483d7fdad..90b18b5f449c 100644 --- a/packages/server/src/event-feed.ts +++ b/packages/server/src/event-feed.ts @@ -37,8 +37,19 @@ export type Interest = { readonly sessions?: ReadonlyArray } +/** Content-free counters for leak detection — never includes event payloads. */ +export type Diagnostics = { + readonly active: number + readonly opens: number + readonly closes: number + readonly serializedEvents: number + readonly serializedBytes: number + readonly overflows: number +} + export interface Interface { readonly subscribe: (interest?: Interest) => Effect.Effect, never, Scope.Scope> + readonly diagnostics: () => Diagnostics } export class Service extends Context.Service()("@opencode/server/EventFeed") {} @@ -101,6 +112,20 @@ export const make = Effect.fn("EventFeed.make")(function* ( const capacity = options?.capacity ?? SubscriberCapacity const render = options?.encode ?? frame 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(() => { @@ -129,8 +154,11 @@ export const make = Effect.fn("EventFeed.make")(function* ( ), ) if (encoded === undefined) return + 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 }))) } @@ -140,17 +168,21 @@ export const make = Effect.fn("EventFeed.make")(function* ( yield* Effect.addFinalizer(() => unsubscribe) return Service.of({ + diagnostics, subscribe: (interest) => Effect.acquireRelease( Queue.dropping(capacity).pipe( Effect.tap((queue) => Effect.sync(() => { + opens += 1 subscribers.set(queue, { queue, interest }) }), ), ), (queue) => - Effect.sync(() => subscribers.delete(queue)).pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid), + 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..0e9bfc6b486b 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 = yield* EventFeed.Service + return feed.diagnostics() + }), ), ) diff --git a/packages/server/src/handlers/event.ts b/packages/server/src/handlers/event.ts index 2bf69f43b102..43586e5d5e61 100644 --- a/packages/server/src/handlers/event.ts +++ b/packages/server/src/handlers/event.ts @@ -1,15 +1,38 @@ +import { NodeHttpServerRequest } from "@effect/platform-node" import { EventV2 } from "@opencode-ai/core/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. */ +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. + } + }) +} + export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) => Effect.gen(function* () { const feed = yield* EventFeed.Service return handlers.handleRaw("event.subscribe", (ctx) => Effect.gen(function* () { + yield* bridgeClientDisconnect(ctx.request) const interest = EventFeed.interestFromQuery(new URL(ctx.request.url, "http://localhost").searchParams) const connected = { id: EventV2.ID.create(), 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..1b100a4bcd11 --- /dev/null +++ b/packages/server/test/event-feed-ownership.test.ts @@ -0,0 +1,109 @@ +import { describe, expect, test } from "bun:test" +import { EventV2 } from "@opencode-ai/core/event" +import { Effect, Fiber, Stream } from "effect" +import { it } from "../../core/test/lib/effect" +import { EventFeed } from "../src/event-feed" + +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)), + } +} + +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) + }), + ) + + test("real network abort closes the response body stream", async () => { + let closed = false + const server = Bun.serve({ + port: 0, + fetch(request) { + let current: ReadableStreamDefaultController | undefined + const stream = new ReadableStream({ + start(controller) { + current = controller + controller.enqueue(new TextEncoder().encode('data: {"type":"server.connected","data":{}}\n\n')) + }, + cancel() { + closed = true + }, + }) + request.signal.addEventListener( + "abort", + () => { + closed = true + try { + current?.close() + } catch { + // already closed by the client disconnect + } + }, + { once: true }, + ) + return new Response(stream, { headers: { "content-type": "text/event-stream" } }) + }, + }) + + try { + const controller = new AbortController() + const response = await fetch(`http://127.0.0.1:${server.port}/api/event`, { signal: controller.signal }) + expect(response.ok).toBe(true) + const reader = response.body!.getReader() + await reader.read() + controller.abort() + await reader.closed.catch(() => undefined) + await Bun.sleep(50) + expect(closed).toBe(true) + } finally { + await server.stop(true) + } + }) + + 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 8ae63f418c9e..1201e6cd5483 100644 --- a/packages/server/test/event-feed.test.ts +++ b/packages/server/test/event-feed.test.ts @@ -363,6 +363,50 @@ describe("EventFeed", () => { 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) }), ) }) diff --git a/packages/tui/src/context/client.tsx b/packages/tui/src/context/client.tsx index e043811aae7e..cb2a300af9c6 100644 --- a/packages/tui/src/context/client.tsx +++ b/packages/tui/src/context/client.tsx @@ -29,6 +29,8 @@ 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", @@ -48,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 } }) @@ -55,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) @@ -68,12 +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 }) + 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 @@ -83,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", { @@ -106,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 }) @@ -127,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() @@ -172,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/test/cli/tui/client-ownership.test.tsx b/packages/tui/test/cli/tui/client-ownership.test.tsx new file mode 100644 index 000000000000..459f5a6e43b6 --- /dev/null +++ b/packages/tui/test/cli/tui/client-ownership.test.tsx @@ -0,0 +1,122 @@ +/** @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 = 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("ClientProvider stream ownership", () => { + test("rapid interest changes never exceed one active event stream", async () => { + const events = createEventStream() + 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) + + let client!: ReturnType + let done!: () => void + const ready = new Promise((resolve) => { + done = resolve + }) + + const app = await testRender(() => ( + + + { + client = ctx + done() + }} + /> + + + )) + + try { + await ready + await wait(() => client.connection.status() === "connected" && concurrent === 1) + + for (let i = 0; i < 8; i++) { + client.event.scope({ location: { directory }, sessions: [`ses_${i}`] }) + } + await wait(() => client.connection.status() === "connected" && concurrent === 1) + expect(maxConcurrent).toBeLessThanOrEqual(2) + expect(concurrent).toBe(1) + expect(client.connection.internal.generation()).toBeGreaterThanOrEqual(8) + } finally { + app.renderer.destroy() + await Bun.sleep(20) + expect(concurrent).toBe(0) + } + }) + + test("teardown leaves zero active event streams", async () => { + const events = createEventStream() + let concurrent = 0 + const calls = createFetch((url, request) => { + if (url.pathname !== "/api/event") return undefined + concurrent += 1 + 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) + + let client!: ReturnType + let done!: () => void + const ready = new Promise((resolve) => { + done = resolve + }) + + const app = await testRender(() => ( + + + { + client = ctx + done() + }} + /> + + + )) + + await ready + await wait(() => client.connection.status() === "connected" && concurrent === 1) + app.renderer.destroy() + await wait(() => concurrent === 0) + expect(concurrent).toBe(0) + }) +}) + +function Probe(props: { onReady: (client: ReturnType) => void }) { + const client = useClient() + onMount(() => props.onReady(client)) + return +} diff --git a/packages/tui/test/cli/tui/client-scope.test.tsx b/packages/tui/test/cli/tui/client-scope.test.tsx index 3c2de5384d1b..36443cb6f055 100644 --- a/packages/tui/test/cli/tui/client-scope.test.tsx +++ b/packages/tui/test/cli/tui/client-scope.test.tsx @@ -18,10 +18,10 @@ describe("ClientProvider event.scope", () => { test("reconnects the event stream when interest changes, not when it is unchanged", async () => { const events = createEventStream() const eventUrls: URL[] = [] - const calls = createFetch((url) => { + const calls = createFetch((url, request) => { if (url.pathname !== "/api/event") return undefined eventUrls.push(url) - return events.v2() + return events.v2(request.signal) }, events) let client!: ReturnType 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 Date: Sat, 18 Jul 2026 11:51:23 +1000 Subject: [PATCH 12/21] fix(server): read event-feed diagnostics via serviceOption Avoid a hard EventFeed.Service requirement on the debug handler so CLI serve Effect inference stays compatible with the runtime handler map. Co-authored-by: Cursor --- packages/server/src/handlers/debug.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/server/src/handlers/debug.ts b/packages/server/src/handlers/debug.ts index 0e9bfc6b486b..5d2905b2d517 100644 --- a/packages/server/src/handlers/debug.ts +++ b/packages/server/src/handlers/debug.ts @@ -26,7 +26,7 @@ export const DebugHandler = HttpApiBuilder.group(Api, "server.debug", (handlers) .handle( "debug.event-feed", Effect.fn(function* () { - const feed = yield* EventFeed.Service + const feed = Option.getOrThrow(yield* Effect.serviceOption(EventFeed.Service)) return feed.diagnostics() }), ), From 379be6f4fed4d155e69b9b8481942686cb3e0439 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 11:54:08 +1000 Subject: [PATCH 13/21] fix(tui): desynchronize managed reconnect ensure herd Spread ensure calls with pre-ensure jitter, keep attempt-1 ensure-on-first- failure after the spread, and lock concurrent reconnect reuse without spawn. Co-authored-by: Cursor --- .../cli/src/services/server-connection.ts | 6 +- packages/cli/test/server-connection.test.ts | 88 ++++++++++++++++++- packages/tui/src/context/client.tsx | 21 ++++- .../tui/test/cli/tui/reconnect-herd.test.ts | 18 ++++ packages/tui/test/cli/tui/use-event.test.tsx | 20 +++-- 5 files changed, 139 insertions(+), 14 deletions(-) create mode 100644 packages/tui/test/cli/tui/reconnect-herd.test.ts diff --git a/packages/cli/src/services/server-connection.ts b/packages/cli/src/services/server-connection.ts index 4eccc95bc94f..666f113034a7 100644 --- a/packages/cli/src/services/server-connection.ts +++ b/packages/cli/src/services/server-connection.ts @@ -49,10 +49,12 @@ export const resolve = Effect.fn("cli.server-connection.resolve")(function* (arg } satisfies Resolved }) +/** Passive reconnect: ensure-running with no version replacement authority. */ +export const managedReconnect = (options: EnsureOptions) => Service.ensure({ ...options, version: undefined }) + function managedService(options: EnsureOptions) { - const reconnectOptions = { ...options, version: undefined } return { - reconnect: () => Service.ensure(reconnectOptions), + reconnect: () => managedReconnect(options), restart: () => Effect.gen(function* () { yield* Service.stop(options) diff --git a/packages/cli/test/server-connection.test.ts b/packages/cli/test/server-connection.test.ts index 41a061f57c03..98c55bf10b2c 100644 --- a/packages/cli/test/server-connection.test.ts +++ b/packages/cli/test/server-connection.test.ts @@ -1,14 +1,100 @@ import { NodeFileSystem } from "@effect/platform-node" +import type { EnsureReason } from "@opencode-ai/client/effect/service" import { Global } from "@opencode-ai/core/global" import { InstallationVersion } from "@opencode-ai/core/installation/version" import { expect, test } from "bun:test" -import { Effect, FileSystem, Scope } from "effect" +import { Effect, Exit, FileSystem, Scope } from "effect" import fs from "node:fs/promises" import os from "node:os" import path from "node:path" import { ServerConnection } from "../src/services/server-connection" import { ServiceConfig } from "../src/services/service-config" +const runReconnect = (effect: Effect.Effect) => + Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer))) + +test("managed reconnect ensures the service on the first failure", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-reconnect-")) + const starts: EnsureReason[] = [] + try { + // Short-circuit ensure once it decides to spawn — we only need the reason. + const exit = await runReconnect( + ServerConnection.managedReconnect({ + file: path.join(root, "service.json"), + onStart: (reason) => { + starts.push(reason) + throw new Error("service ensure invoked") + }, + }).pipe(Effect.exit), + ) + expect(Exit.isFailure(exit)).toBe(true) + expect(starts).toEqual(["missing"]) + } finally { + await fs.rm(root, { recursive: true, force: true }) + } +}) + +test("managed reconnect reuses a healthy service from another version", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-reconnect-version-")) + const server = Bun.serve({ + port: 0, + fetch() { + return Response.json({ healthy: true, version: "older", pid: process.pid }) + }, + }) + const file = path.join(root, "service.json") + const url = server.url.toString() + await Bun.write(file, JSON.stringify({ url, pid: process.pid, version: "older" })) + try { + const endpoint = await runReconnect( + ServerConnection.managedReconnect({ + file, + version: "newer", + command: [path.join(root, "must-not-start")], + }), + ) + expect(endpoint.url).toBe(url) + } finally { + await server.stop(true) + await fs.rm(root, { recursive: true, force: true }) + } +}) + +test("concurrent managed reconnects reuse one healthy endpoint without spawning", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-reconnect-herd-")) + const server = Bun.serve({ + port: 0, + fetch() { + return Response.json({ healthy: true, version: "older", pid: process.pid }) + }, + }) + const file = path.join(root, "service.json") + const url = server.url.toString() + await Bun.write(file, JSON.stringify({ url, pid: process.pid, version: "older" })) + const mustNotStart = path.join(root, "must-not-start") + try { + const endpoints = await Promise.all( + Array.from({ length: 7 }, () => + runReconnect( + ServerConnection.managedReconnect({ + file, + version: "newer", + command: [mustNotStart], + }), + ), + ), + ) + expect(endpoints.map((endpoint) => endpoint.url)).toEqual(Array.from({ length: 7 }, () => url)) + expect(await fs.stat(mustNotStart).then( + () => true, + () => false, + )).toBe(false) + } finally { + await server.stop(true) + await fs.rm(root, { recursive: true, force: true }) + } +}) + test("resolution groups Effect-native lifecycle operations only for the managed service", async () => { const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-server-resolution-")) const id = "server-resolution-test" diff --git a/packages/tui/src/context/client.tsx b/packages/tui/src/context/client.tsx index cb2a300af9c6..bee0c6c7ec7e 100644 --- a/packages/tui/src/context/client.tsx +++ b/packages/tui/src/context/client.tsx @@ -32,6 +32,16 @@ const connectionHistoryLimit = 50 const reconnectBaseMs = 1_000 const reconnectJitterMs = 500 +/** Post-ensure SSE retry backoff: base + jitter so N clients do not phase-lock. */ +export function reconnectBackoffMs(random = Math.random) { + return reconnectBaseMs + Math.floor(random() * reconnectJitterMs) +} + +/** Pre-ensure spread so N reconnecting TUIs do not stampede Service.ensure together. */ +export function reconnectEnsureSpreadMs(random = Math.random) { + return Math.floor(random() * reconnectJitterMs) +} + export const { use: useClient, provider: ClientProvider } = createSimpleContext({ name: "Client", init: (props: { api: OpenCodeClient; service?: ManagedService; interest?: EventInterest }) => { @@ -125,9 +135,12 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( error: message, }) setConnection({ status: "reconnecting", attempt, error: message }) - // Re-resolve the transport before retrying: the server may have - // moved (service restarted on a new port) or need starting. Static - // transports (--server, standalone) resolve to the same address. + // Spread ensure calls across reconnecting clients, then re-resolve the + // transport. Static transports (--server, standalone) resolve to the + // same address. Attempt 1 still ensures immediately after the spread + // (ensure-on-first-failure) and skips the post-ensure backoff. + await wait(reconnectEnsureSpreadMs(), controller.signal) + if (abort.signal.aborted || controller.signal.aborted || gen !== generation) return if (props.service) { const next = await props.service.reconnect(controller.signal).catch((error) => { if (!controller.signal.aborted) @@ -142,7 +155,7 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( if (attempt === 1) continue } } - await wait(reconnectBaseMs + Math.floor(Math.random() * reconnectJitterMs), controller.signal) + await wait(reconnectBackoffMs(), controller.signal) } })() } diff --git a/packages/tui/test/cli/tui/reconnect-herd.test.ts b/packages/tui/test/cli/tui/reconnect-herd.test.ts new file mode 100644 index 000000000000..6a727fbf29cb --- /dev/null +++ b/packages/tui/test/cli/tui/reconnect-herd.test.ts @@ -0,0 +1,18 @@ +import { describe, expect, test } from "bun:test" +import { reconnectBackoffMs, reconnectEnsureSpreadMs } from "../../../src/context/client" + +describe("managed reconnect herd timing", () => { + test("ensure spread stays within jitter bounds and desynchronizes samples", () => { + const samples = Array.from({ length: 64 }, (_, i) => reconnectEnsureSpreadMs(() => i / 64)) + expect(Math.min(...samples)).toBe(0) + expect(Math.max(...samples)).toBeLessThan(500) + expect(new Set(samples).size).toBeGreaterThan(1) + }) + + test("post-ensure backoff stays in [1000, 1500)", () => { + const samples = Array.from({ length: 64 }, (_, i) => reconnectBackoffMs(() => i / 64)) + expect(Math.min(...samples)).toBe(1000) + expect(Math.max(...samples)).toBeLessThan(1500) + expect(new Set(samples).size).toBeGreaterThan(1) + }) +}) diff --git a/packages/tui/test/cli/tui/use-event.test.tsx b/packages/tui/test/cli/tui/use-event.test.tsx index c26d2c876b5a..c493a7f77a68 100644 --- a/packages/tui/test/cli/tui/use-event.test.tsx +++ b/packages/tui/test/cli/tui/use-event.test.tsx @@ -265,10 +265,12 @@ describe("useEvent", () => { try { await wait(() => client.connection.status() === "connected") events.disconnect() + // Ensure-on-first-failure still runs on attempt 1 after pre-ensure spread + // (0–500ms). Allow two ensure spreads before requiring the second call. await Promise.race([ - wait(() => calls === 2), - Bun.sleep(500).then(() => { - throw new Error("resolved event stream did not retry immediately") + wait(() => calls === 2, 2_500), + Bun.sleep(2_500).then(() => { + throw new Error("resolved event stream did not retry after ensure spread") }), ]) await Bun.sleep(200) @@ -279,10 +281,12 @@ describe("useEvent", () => { }) test("cancels pending endpoint resolution on cleanup", async () => { + let started = false let aborted = false const { app, events, client } = await mount( - (signal) => - new Promise((_, reject) => { + (signal) => { + started = true + return new Promise((_, reject) => { signal.addEventListener( "abort", () => { @@ -291,12 +295,14 @@ describe("useEvent", () => { }, { once: true }, ) - }), + }) + }, ) await wait(() => client.connection.status() === "connected") events.disconnect() - await wait(() => client.connection.status() === "reconnecting") + // Pre-ensure spread runs before reconnect; wait until ensure is in flight. + await wait(() => started, 2_500) app.renderer.destroy() await wait(() => aborted) }) From d7325fd6b108f8d36a189bb3f3b8341e830418d6 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 12:09:33 +1000 Subject: [PATCH 14/21] refactor(schema): decode sessionIDOf data via Schema Replace the hand-rolled unknown guards with decodeUnknownOption so ephemeral sessionID extraction matches host Schema patterns. Co-authored-by: Cursor --- packages/schema/src/event.ts | 12 +++++------- packages/schema/test/event.test.ts | 2 ++ 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index 989413a4e8a7..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" @@ -197,16 +197,14 @@ 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 - const data = event.data - if (data === undefined || data === null || typeof data !== "object") return undefined - if (!("sessionID" in data)) return undefined - const sessionID = data.sessionID - if (typeof sessionID === "string") return sessionID - return undefined + return Option.getOrUndefined(decodeSessionData(event.data))?.sessionID } diff --git a/packages/schema/test/event.test.ts b/packages/schema/test/event.test.ts index 62ffbb9e797d..cfddf15fb954 100644 --- a/packages/schema/test/event.test.ts +++ b/packages/schema/test/event.test.ts @@ -56,5 +56,7 @@ describe("public event schemas", () => { ) 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() }) }) From 723236088fc9fb1944a9380f6edacc767839bc66 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 12:14:05 +1000 Subject: [PATCH 15/21] refactor(server): decode event interest via EventSubscribeQuery Map HttpApi-decoded ctx.query through interestFromSubscribeQuery, and route deepObject URLSearchParams through the same Schema before building Interest. Co-authored-by: Cursor --- packages/server/src/event-feed.ts | 63 ++++++++++++++++++++----- packages/server/src/handlers/event.ts | 3 +- packages/server/test/event-feed.test.ts | 14 ++++++ 3 files changed, 67 insertions(+), 13 deletions(-) diff --git a/packages/server/src/event-feed.ts b/packages/server/src/event-feed.ts index 71f483d7fdad..3df3545eecbc 100644 --- a/packages/server/src/event-feed.ts +++ b/packages/server/src/event-feed.ts @@ -1,8 +1,13 @@ export * as EventFeed from "./event-feed" import { EventV2 } from "@opencode-ai/core/event" -import { isOpenCodeEvent, OpenCodeEvent, sessionIDOf } from "@opencode-ai/protocol/groups/event" -import { Cause, Context, Effect, Layer, Queue, Schema, Scope, Stream } from "effect" +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 @@ -70,25 +75,59 @@ export function matchesInterest(event: EventV2.Payload, interest?: Interest): bo return true } -export function interestFromQuery(query: URLSearchParams): Interest | undefined { - const directory = query.get("location[directory]") ?? undefined - const workspace = query.get("location[workspace]") ?? undefined - const sessions = query +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) - // Workspace interest requires a directory; alone it is an invalid location scope. - if (directory === undefined && workspace !== undefined) return undefined - if (directory === undefined && sessions.length === 0) return undefined return { - ...(directory !== undefined - ? { location: { directory, ...(workspace !== undefined ? { workspace } : {}) } } + ...(directory !== undefined || workspace !== undefined + ? { + location: { + ...(directory !== undefined ? { directory } : {}), + ...(workspace !== undefined ? { workspace } : {}), + }, + } : {}), - ...(sessions.length > 0 ? { sessions } : {}), + ...(sessions.length === 0 ? {} : { session: sessions.length === 1 ? sessions[0] : sessions }), } } +function normalizeSessions(session: typeof EventSubscribeQuery.Type.session) { + if (session === undefined) return [] as string[] + 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 diff --git a/packages/server/src/handlers/event.ts b/packages/server/src/handlers/event.ts index 2bf69f43b102..ab49777d1db8 100644 --- a/packages/server/src/handlers/event.ts +++ b/packages/server/src/handlers/event.ts @@ -10,7 +10,8 @@ export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) const feed = yield* EventFeed.Service return handlers.handleRaw("event.subscribe", (ctx) => Effect.gen(function* () { - const interest = EventFeed.interestFromQuery(new URL(ctx.request.url, "http://localhost").searchParams) + // handleRaw still Schema-decodes query; map into feed interest without re-parsing the URL. + const interest = EventFeed.interestFromSubscribeQuery(ctx.query) const connected = { id: EventV2.ID.create(), type: "server.connected", diff --git a/packages/server/test/event-feed.test.ts b/packages/server/test/event-feed.test.ts index 8ae63f418c9e..4e40fc34570f 100644 --- a/packages/server/test/event-feed.test.ts +++ b/packages/server/test/event-feed.test.ts @@ -265,6 +265,20 @@ describe("EventFeed", () => { 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( From 364c7a5d6414710a52b4b0f774a5229b109c2745 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 12:37:19 +1000 Subject: [PATCH 16/21] test(tui): split event.scope reconnect assertions One test was asserting connect, interest change, and no-op scope together. Split into three named cases with a shared mount helper. Co-authored-by: Cursor --- .../tui/test/cli/tui/client-scope.test.tsx | 83 ++++++++++++------- 1 file changed, 53 insertions(+), 30 deletions(-) diff --git a/packages/tui/test/cli/tui/client-scope.test.tsx b/packages/tui/test/cli/tui/client-scope.test.tsx index 3c2de5384d1b..d90d3f437286 100644 --- a/packages/tui/test/cli/tui/client-scope.test.tsx +++ b/packages/tui/test/cli/tui/client-scope.test.tsx @@ -14,50 +14,73 @@ async function wait(fn: () => boolean, timeout = 2000) { } } -describe("ClientProvider event.scope", () => { - test("reconnects the event stream when interest changes, not when it is unchanged", async () => { - const events = createEventStream() - const eventUrls: URL[] = [] - const calls = createFetch((url) => { - if (url.pathname !== "/api/event") return undefined - eventUrls.push(url) - return events.v2() - }, events) +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() + }, events) - let client!: ReturnType - let done!: () => void - const ready = new Promise((resolve) => { - done = resolve - }) + let client!: ReturnType + let done!: () => void + const ready = new Promise((resolve) => { + done = resolve + }) - const app = await testRender(() => ( - - - { - client = ctx - done() - }} - /> - - - )) + 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 ready 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) From 1a2e1a703ee857fc118025aabd0930703de2a612 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 12:55:48 +1000 Subject: [PATCH 17/21] test(tui): share mount and stream tracker in ownership suite Co-authored-by: Cursor --- .../test/cli/tui/client-ownership.test.tsx | 218 +++++++++--------- 1 file changed, 105 insertions(+), 113 deletions(-) diff --git a/packages/tui/test/cli/tui/client-ownership.test.tsx b/packages/tui/test/cli/tui/client-ownership.test.tsx index a7baa325128b..30fab7f5cf08 100644 --- a/packages/tui/test/cli/tui/client-ownership.test.tsx +++ b/packages/tui/test/cli/tui/client-ownership.test.tsx @@ -15,105 +15,131 @@ async function wait(fn: () => boolean, timeout = 3000) { } } -describe("ClientProvider stream ownership", () => { - test("rapid interest changes never exceed one active event stream", async () => { - const events = createEventStream() - let concurrent = 0 - let maxConcurrent = 0 - const calls = createFetch((url, request) => { +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 - 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) - - let client!: ReturnType - let done!: () => void - const ready = new Promise((resolve) => { - done = resolve - }) + return events.v2(request.signal) + }, events).fetch, + onReady: input?.onReady, + }) + return { app, client, events } +} - const app = await testRender(() => ( - - - { - client = ctx - done() - }} - /> - - - )) +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 ready - await wait(() => client.connection.status() === "connected" && concurrent === 1) + await wait(connectedAlone(client, streams)) for (let i = 0; i < 8; i++) { client.event.scope({ location: { directory }, sessions: [`ses_${i}`] }) } - await wait(() => client.connection.status() === "connected" && concurrent === 1) - expect(maxConcurrent).toBeLessThanOrEqual(2) - expect(concurrent).toBe(1) + 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(concurrent).toBe(0) } + await Bun.sleep(20) + expect(streams.concurrent).toBe(0) }) test("event IDs from a prior generation are not delivered after interest reconnect", async () => { - const events = createEventStream() - const calls = createFetch((url, request) => { - if (url.pathname !== "/api/event") return undefined - return events.v2(request.signal) - }, events) - const seen: Array<{ id: string; generation: number }> = [] - let client!: ReturnType - let done!: () => void - const ready = new Promise((resolve) => { - done = resolve + 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() }) + }) + }, }) - const app = await testRender(() => ( - - - { - client = ctx - client.event.on("vcs.branch.updated", (event) => { - seen.push({ id: event.id, generation: client.connection.internal.generation() }) - }) - done() - }} - /> - - - )) - try { - await ready - await wait(() => client.connection.status() === "connected") + 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( - () => - client.connection.status() === "connected" && client.connection.internal.generation() > generation, - ) + 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")) @@ -126,46 +152,12 @@ describe("ClientProvider stream ownership", () => { }) test("teardown leaves zero active event streams", async () => { - const events = createEventStream() - let concurrent = 0 - const calls = createFetch((url, request) => { - if (url.pathname !== "/api/event") return undefined - concurrent += 1 - 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) - - let client!: ReturnType - let done!: () => void - const ready = new Promise((resolve) => { - done = resolve - }) + const { app, client, streams } = await mountTracked() - const app = await testRender(() => ( - - - { - client = ctx - done() - }} - /> - - - )) - - await ready - await wait(() => client.connection.status() === "connected" && concurrent === 1) + await wait(connectedAlone(client, streams)) app.renderer.destroy() - await wait(() => concurrent === 0) - expect(concurrent).toBe(0) + await wait(() => streams.concurrent === 0) + expect(streams.concurrent).toBe(0) }) }) From a6022592321aa82ea23ee02ac877c563153c9fe8 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 12:56:24 +1000 Subject: [PATCH 18/21] test(cli): share temp root helpers in managed reconnect suite Co-authored-by: Cursor --- packages/cli/test/server-connection.test.ts | 76 +++++++++++---------- 1 file changed, 39 insertions(+), 37 deletions(-) diff --git a/packages/cli/test/server-connection.test.ts b/packages/cli/test/server-connection.test.ts index 98c55bf10b2c..9e2a4e9523be 100644 --- a/packages/cli/test/server-connection.test.ts +++ b/packages/cli/test/server-connection.test.ts @@ -13,10 +13,40 @@ import { ServiceConfig } from "../src/services/service-config" const runReconnect = (effect: Effect.Effect) => Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer))) -test("managed reconnect ensures the service on the first failure", async () => { - const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-reconnect-")) - const starts: EnsureReason[] = [] +async function withTempRoot(prefix: string, body: (root: string) => Promise) { + const root = await fs.mkdtemp(path.join(os.tmpdir(), prefix)) try { + await body(root) + } finally { + await fs.rm(root, { recursive: true, force: true }) + } +} + +async function withHealthyRegisteredService( + prefix: string, + body: (input: { root: string; file: string; url: string }) => Promise, +) { + await withTempRoot(prefix, async (root) => { + const server = Bun.serve({ + port: 0, + fetch() { + return Response.json({ healthy: true, version: "older", pid: process.pid }) + }, + }) + const file = path.join(root, "service.json") + const url = server.url.toString() + await Bun.write(file, JSON.stringify({ url, pid: process.pid, version: "older" })) + try { + await body({ root, file, url }) + } finally { + await server.stop(true) + } + }) +} + +test("managed reconnect ensures the service on the first failure", async () => { + await withTempRoot("opencode-service-reconnect-", async (root) => { + const starts: EnsureReason[] = [] // Short-circuit ensure once it decides to spawn — we only need the reason. const exit = await runReconnect( ServerConnection.managedReconnect({ @@ -29,23 +59,11 @@ test("managed reconnect ensures the service on the first failure", async () => { ) expect(Exit.isFailure(exit)).toBe(true) expect(starts).toEqual(["missing"]) - } finally { - await fs.rm(root, { recursive: true, force: true }) - } + }) }) test("managed reconnect reuses a healthy service from another version", async () => { - const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-reconnect-version-")) - const server = Bun.serve({ - port: 0, - fetch() { - return Response.json({ healthy: true, version: "older", pid: process.pid }) - }, - }) - const file = path.join(root, "service.json") - const url = server.url.toString() - await Bun.write(file, JSON.stringify({ url, pid: process.pid, version: "older" })) - try { + await withHealthyRegisteredService("opencode-service-reconnect-version-", async ({ root, file, url }) => { const endpoint = await runReconnect( ServerConnection.managedReconnect({ file, @@ -54,25 +72,12 @@ test("managed reconnect reuses a healthy service from another version", async () }), ) expect(endpoint.url).toBe(url) - } finally { - await server.stop(true) - await fs.rm(root, { recursive: true, force: true }) - } + }) }) test("concurrent managed reconnects reuse one healthy endpoint without spawning", async () => { - const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-reconnect-herd-")) - const server = Bun.serve({ - port: 0, - fetch() { - return Response.json({ healthy: true, version: "older", pid: process.pid }) - }, - }) - const file = path.join(root, "service.json") - const url = server.url.toString() - await Bun.write(file, JSON.stringify({ url, pid: process.pid, version: "older" })) - const mustNotStart = path.join(root, "must-not-start") - try { + await withHealthyRegisteredService("opencode-service-reconnect-herd-", async ({ root, file, url }) => { + const mustNotStart = path.join(root, "must-not-start") const endpoints = await Promise.all( Array.from({ length: 7 }, () => runReconnect( @@ -89,10 +94,7 @@ test("concurrent managed reconnects reuse one healthy endpoint without spawning" () => true, () => false, )).toBe(false) - } finally { - await server.stop(true) - await fs.rm(root, { recursive: true, force: true }) - } + }) }) test("resolution groups Effect-native lifecycle operations only for the managed service", async () => { From 196337372d2c1ea0b1667b32d61e59fd1239e593 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 13:14:49 +1000 Subject: [PATCH 19/21] refactor(tui): clarify subscribeInput locals and drop test bangs Co-authored-by: Cursor --- packages/tui/src/context/event-interest.ts | 6 ++++-- packages/tui/test/cli/tui/client-scope.test.tsx | 8 ++++---- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/packages/tui/src/context/event-interest.ts b/packages/tui/src/context/event-interest.ts index 8db39edd171e..cd40ec7c97f2 100644 --- a/packages/tui/src/context/event-interest.ts +++ b/packages/tui/src/context/event-interest.ts @@ -21,8 +21,10 @@ export function interestEqual(left?: EventInterest, right?: EventInterest) { /** 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 { - ...(interest.location ? { location: interest.location } : {}), - ...(interest.sessions && interest.sessions.length > 0 ? { session: [...interest.sessions] } : {}), + ...(location ? { location } : {}), + ...(session ? { session } : {}), } } diff --git a/packages/tui/test/cli/tui/client-scope.test.tsx b/packages/tui/test/cli/tui/client-scope.test.tsx index d90d3f437286..5742bc23389f 100644 --- a/packages/tui/test/cli/tui/client-scope.test.tsx +++ b/packages/tui/test/cli/tui/client-scope.test.tsx @@ -51,8 +51,8 @@ describe("ClientProvider event.scope", () => { 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([]) + expect(eventUrls[0]?.searchParams.get("location[directory]")).toBe(directory) + expect(eventUrls[0]?.searchParams.getAll("session")).toEqual([]) } finally { app.renderer.destroy() } @@ -65,8 +65,8 @@ describe("ClientProvider event.scope", () => { 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"]) + expect(eventUrls[1]?.searchParams.get("location[directory]")).toBe(directory) + expect(eventUrls[1]?.searchParams.getAll("session")).toEqual(["ses_a"]) } finally { app.renderer.destroy() } From b79e5433f1806703695d6ddb40fca0d36347dbb2 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 14:00:04 +1000 Subject: [PATCH 20/21] test(server): prove Node HTTP abort releases EventFeed subscribers Co-authored-by: Cursor --- packages/server/src/handlers/event.ts | 67 +++++++------ .../server/test/event-feed-ownership.test.ts | 97 ++++++++++--------- 2 files changed, 88 insertions(+), 76 deletions(-) diff --git a/packages/server/src/handlers/event.ts b/packages/server/src/handlers/event.ts index bf82384ed514..2f88cb3d3e0f 100644 --- a/packages/server/src/handlers/event.ts +++ b/packages/server/src/handlers/event.ts @@ -1,5 +1,6 @@ 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 { HttpServerRequest, HttpServerResponse } from "effect/unstable/http" import { HttpApiBuilder } from "effect/unstable/httpapi" @@ -7,7 +8,7 @@ import { Api } from "../api" import { EventFeed } from "../event-feed" /** Bun may emit IncomingMessage aborted/close without ServerResponse.close — destroy so the fiber ends. */ -function bridgeClientDisconnect(request: HttpServerRequest.HttpServerRequest) { +export function bridgeClientDisconnect(request: HttpServerRequest.HttpServerRequest) { return Effect.sync(() => { try { const incoming = NodeHttpServerRequest.toIncomingMessage(request) @@ -27,37 +28,43 @@ function bridgeClientDisconnect(request: HttpServerRequest.HttpServerRequest) { }) } +/** 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", (ctx) => - Effect.gen(function* () { - yield* bridgeClientDisconnect(ctx.request) - // handleRaw still Schema-decodes query; map into feed interest without re-parsing the URL. - const interest = EventFeed.interestFromSubscribeQuery(ctx.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", - }, - }, - ) - }), - ) + return handlers.handleRaw("event.subscribe", (ctx) => subscribeResponse(feed, ctx.request, ctx.query)) }), ) diff --git a/packages/server/test/event-feed-ownership.test.ts b/packages/server/test/event-feed-ownership.test.ts index 1b100a4bcd11..5a562f2bccf7 100644 --- a/packages/server/test/event-feed-ownership.test.ts +++ b/packages/server/test/event-feed-ownership.test.ts @@ -1,8 +1,12 @@ -import { describe, expect, test } from "bun:test" +import { NodeHttpServer } from "@effect/platform-node" import { EventV2 } from "@opencode-ai/core/event" -import { Effect, Fiber, Stream } from "effect" +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 @@ -18,6 +22,14 @@ function makeSource() { } } +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* () { @@ -41,51 +53,44 @@ describe("EventFeed ownership", () => { }), ) - test("real network abort closes the response body stream", async () => { - let closed = false - const server = Bun.serve({ - port: 0, - fetch(request) { - let current: ReadableStreamDefaultController | undefined - const stream = new ReadableStream({ - start(controller) { - current = controller - controller.enqueue(new TextEncoder().encode('data: {"type":"server.connected","data":{}}\n\n')) - }, - cancel() { - closed = true - }, - }) - request.signal.addEventListener( - "abort", - () => { - closed = true - try { - current?.close() - } catch { - // already closed by the client disconnect - } - }, - { once: true }, - ) - return new Response(stream, { headers: { "content-type": "text/event-stream" } }) - }, - }) + 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) - try { - const controller = new AbortController() - const response = await fetch(`http://127.0.0.1:${server.port}/api/event`, { signal: controller.signal }) - expect(response.ok).toBe(true) - const reader = response.body!.getReader() - await reader.read() - controller.abort() - await reader.closed.catch(() => undefined) - await Bun.sleep(50) - expect(closed).toBe(true) - } finally { - await server.stop(true) - } - }) + 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* () { From 85f9a1f172ef99aaa4091da504ddb56e98c82743 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 14:01:17 +1000 Subject: [PATCH 21/21] test(server): add non-gating EventFeed fan-out benchmark Co-authored-by: Cursor --- .../test/event-feed-fanout.bench.test.ts | 166 ++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 packages/server/test/event-feed-fanout.bench.test.ts 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 + }), + ) +}