From 3eac0eb397b4fe5a15552d09b5e9e24de57227f6 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 21:58:33 +1000 Subject: [PATCH 1/4] 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 7674f994e73854d3ebbbdca9250d5064c88cf314 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 22:33:03 +1000 Subject: [PATCH 2/4] 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 4b54724efca1b24cf986efa7de97ac9054992677 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Fri, 17 Jul 2026 23:05:17 +1000 Subject: [PATCH 3/4] 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 51ff5f83dac3db84a26300ba17d62524955d0641 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 08:05:55 +1000 Subject: [PATCH 4/4] 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 })))