From 3eac0eb397b4fe5a15552d09b5e9e24de57227f6 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 21:58:33 +1000 Subject: [PATCH 01/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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/15] 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 d7325fd6b108f8d36a189bb3f3b8341e830418d6 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 12:09:33 +1000 Subject: [PATCH 11/15] 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 12/15] 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 13/15] 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 196337372d2c1ea0b1667b32d61e59fd1239e593 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 13:14:49 +1000 Subject: [PATCH 14/15] 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 85f9a1f172ef99aaa4091da504ddb56e98c82743 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 14:01:17 +1000 Subject: [PATCH 15/15] 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 + }), + ) +}