From 905123b9c0716b20c09a03991ff314dd57d08795 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 4 Jul 2026 21:12:45 -0400 Subject: [PATCH] fix(protocol): keep internal events off SSE (#35378) --- .../client/src/promise/generated/types.ts | 8 ------ packages/core/src/event.ts | 21 ++++++++++------ packages/core/src/plugin/host.ts | 4 +-- packages/core/test/event.test.ts | 20 +++++++++++++-- packages/protocol/src/groups/event.ts | 2 ++ packages/protocol/test/event.test.ts | 25 ++++--------------- packages/schema/src/event-manifest.ts | 6 ++++- packages/schema/test/event-manifest.test.ts | 3 ++- packages/server/src/handlers/event.ts | 7 ++++-- 9 files changed, 52 insertions(+), 44 deletions(-) diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index ba3c1a563b..151f0ff03f 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -5402,14 +5402,6 @@ export type EventSubscribeOutput = readonly location?: { readonly directory: string; readonly workspaceID?: string } readonly data: { readonly branch?: string } } - | { - readonly id: string - readonly created: number - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "mcp.tools.changed" - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { readonly server: string } - } | { readonly id: string readonly created: number diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index 587688d814..3e352c6e6b 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -160,15 +160,22 @@ export interface Interface { export class Service extends Context.Service()("@opencode/Event") {} -export const liveBounded = (events: Interface, capacity: number) => +export const liveBounded = ( + events: Interface, + options: { readonly capacity: number; readonly accept?: (event: Payload) => boolean }, +) => Effect.gen(function* () { - const queue = yield* Queue.dropping(capacity) + const queue = yield* Queue.dropping(options.capacity) const unsubscribe = yield* events.listen((event) => - Queue.offer(queue, event).pipe( - Effect.flatMap((accepted) => - accepted ? Effect.void : Queue.fail(queue, new SubscriberOverflowError({ capacity })).pipe(Effect.asVoid), - ), - ), + options.accept && !options.accept(event) + ? Effect.void + : Queue.offer(queue, event).pipe( + Effect.flatMap((accepted) => + accepted + ? Effect.void + : Queue.fail(queue, new SubscriberOverflowError({ capacity: options.capacity })).pipe(Effect.asVoid), + ), + ), ) yield* Effect.addFinalizer(() => unsubscribe.pipe(Effect.andThen(Queue.shutdown(queue)), Effect.asVoid)) return Stream.fromQueue(queue) diff --git a/packages/core/src/plugin/host.ts b/packages/core/src/plugin/host.ts index 585c254a78..4c7bb8538c 100644 --- a/packages/core/src/plugin/host.ts +++ b/packages/core/src/plugin/host.ts @@ -23,8 +23,6 @@ import { ToolHooks } from "../tool/hooks" import { WorkspaceV2 } from "../workspace" const mutable = (value: T) => value as DeepMutable -const isEvent = Schema.is(Schema.Union(EventManifest.ServerDefinitions)) - export const make = Effect.fn("PluginHost.make")(function* (plugin: PluginV2.Interface) { const agents = yield* AgentV2.Service const aisdk = yield* AISDK.Service @@ -160,7 +158,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: PluginV2.Int }), }, event: { - subscribe: () => events.live().pipe(Stream.filter(isEvent)), + subscribe: () => events.live().pipe(Stream.filter(EventManifest.isServer)), }, integration: { list: () => response(integration.list()), diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index 65c370f1eb..70a8083d16 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -2,6 +2,8 @@ import { describe, expect } from "bun:test" import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Ref, Schema, Stream } from "effect" import { EventV2 } from "@opencode-ai/core/event" import { Event } from "@opencode-ai/schema/event" +import { EventManifest } from "@opencode-ai/schema/event-manifest" +import { McpEvent } from "@opencode-ai/schema/mcp-event" import { Session } from "@opencode-ai/schema/session" import { SessionEvent } from "@opencode-ai/schema/session-event" import { SessionV1 } from "@opencode-ai/schema/session-v1" @@ -329,8 +331,8 @@ describe("EventV2", () => { const events = yield* EventV2.Service const consuming = yield* Deferred.make() const release = yield* Deferred.make() - const slowStream = yield* EventV2.liveBounded(events, 1) - const fastStream = yield* EventV2.liveBounded(events, 8) + const slowStream = yield* EventV2.liveBounded(events, { capacity: 1 }) + const fastStream = yield* EventV2.liveBounded(events, { capacity: 8 }) const slow = yield* slowStream.pipe( Stream.runForEach(() => Deferred.succeed(consuming, undefined).pipe(Effect.andThen(Deferred.await(release)))), Effect.forkScoped, @@ -355,6 +357,20 @@ describe("EventV2", () => { }), ) + it.effect("filters internal events before they enter a bounded server stream", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const stream = yield* EventV2.liveBounded(events, { capacity: 1, accept: EventManifest.isServer }) + const received = yield* stream.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + + yield* events.publish(McpEvent.ToolsChanged, { server: "one" }) + yield* events.publish(McpEvent.ToolsChanged, { server: "two" }) + const published = yield* events.publish(McpEvent.StatusChanged, { server: "example" }) + + expect(Array.from(yield* Fiber.join(received))).toEqual([published]) + }), + ) + it.effect("preserves observer interruption", () => Effect.gen(function* () { const events = yield* EventV2.Service diff --git a/packages/protocol/src/groups/event.ts b/packages/protocol/src/groups/event.ts index 9860aa0cad..94cedd1a57 100644 --- a/packages/protocol/src/groups/event.ts +++ b/packages/protocol/src/groups/event.ts @@ -67,3 +67,5 @@ export const EventGroup = event.group export const OpenCodeEvent = event.schema export type OpenCodeEvent = typeof OpenCodeEvent.Type export type OpenCodeEventEncoded = typeof OpenCodeEvent.Encoded +export const isOpenCodeEvent = (event: { readonly type: string }): event is OpenCodeEvent => + event.type === "server.connected" || EventManifest.isServer(event) diff --git a/packages/protocol/test/event.test.ts b/packages/protocol/test/event.test.ts index 52757ad30e..33ece777c4 100644 --- a/packages/protocol/test/event.test.ts +++ b/packages/protocol/test/event.test.ts @@ -1,23 +1,8 @@ import { expect, test } from "bun:test" -import { Event } from "@opencode-ai/schema/event" -import { AbsolutePath } from "@opencode-ai/schema/schema" -import { DateTime, Schema } from "effect" -import { OpenCodeEvent } from "../src/groups/event.js" +import { isOpenCodeEvent } from "../src/groups/event.js" -test("encodes MCP tool changes emitted by the server", () => { - expect( - Schema.encodeSync(OpenCodeEvent)({ - id: Event.ID.make("evt_test"), - created: DateTime.makeUnsafe(0), - type: "mcp.tools.changed", - location: { directory: AbsolutePath.make("/tmp") }, - data: { server: "example" }, - }), - ).toEqual({ - id: "evt_test", - created: 0, - type: "mcp.tools.changed", - location: { directory: "/tmp" }, - data: { server: "example" }, - }) +test("classifies public events by type", () => { + expect(isOpenCodeEvent({ type: "server.connected" })).toBe(true) + expect(isOpenCodeEvent({ type: "mcp.status.changed" })).toBe(true) + expect(isOpenCodeEvent({ type: "mcp.tools.changed" })).toBe(false) }) diff --git a/packages/schema/src/event-manifest.ts b/packages/schema/src/event-manifest.ts index 09a5db8da4..23ed3fd044 100644 --- a/packages/schema/src/event-manifest.ts +++ b/packages/schema/src/event-manifest.ts @@ -1,5 +1,6 @@ export * as EventManifest from "./event-manifest.js" +import { Schema } from "effect" import { Agent } from "./agent.js" import { Catalog } from "./catalog.js" import { Command } from "./command.js" @@ -78,7 +79,7 @@ export const ServerDefinitions = Event.inventory( ...TuiEvent.Definitions, ...InstallationEvent.Definitions, ...VcsEvent.Definitions, - ...McpEvent.Definitions, + McpEvent.StatusChanged, // Shared transitional: V1 contracts the current TUI still consumes during // the migration (permission.asked/replied, question.asked, session.error). // Remove when the TUI moves to the current permission/question surfaces. @@ -86,6 +87,9 @@ export const ServerDefinitions = Event.inventory( ...QuestionV1.Event.Definitions, SessionV1.Error, ) +export const Server = Event.latest(ServerDefinitions) +export type ServerEvent = Schema.Schema.Type<(typeof ServerDefinitions)[number]> +export const isServer = (event: { readonly type: string }): event is ServerEvent => Server.has(event.type) export const Definitions = Event.inventory( ...foundationDefinitions, diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index 22003313ba..0147c7469b 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -27,7 +27,6 @@ describe("public event manifest", () => { Agent.Event.Updated, ]) expect(EventManifest.Definitions).toContain(Agent.Event.Updated) - expect(EventManifest.ServerDefinitions).toContain(McpEvent.ToolsChanged) expect(EventManifest.Definitions.filter((definition) => definition.type === "agent.updated")).toEqual([ Agent.Event.Updated, ]) @@ -47,6 +46,8 @@ describe("public event manifest", () => { EventManifest.Definitions.map((definition) => definition.type), ) expect(EventManifest.Latest.get("agent.updated")).toBe(Agent.Event.Updated) + expect(EventManifest.Server.get("mcp.status.changed")).toBe(McpEvent.StatusChanged) + expect(EventManifest.Server.has("mcp.tools.changed")).toBe(false) expect(Agent.Event.Updated.durable).toBeUndefined() expect(EventManifest.Durable.has("agent.updated")).toBe(false) }) diff --git a/packages/server/src/handlers/event.ts b/packages/server/src/handlers/event.ts index ca9c02620d..1c8b94d124 100644 --- a/packages/server/src/handlers/event.ts +++ b/packages/server/src/handlers/event.ts @@ -1,5 +1,5 @@ import { EventV2 } from "@opencode-ai/core/event" -import { OpenCodeEvent } from "@opencode-ai/protocol/groups/event" +import { isOpenCodeEvent, OpenCodeEvent } from "@opencode-ai/protocol/groups/event" import { Effect, Schema, Stream } from "effect" import { Sse } from "effect/unstable/encoding" import { HttpServerResponse } from "effect/unstable/http" @@ -35,7 +35,10 @@ export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) const output = Stream.unwrap( Effect.gen(function* () { // Acquiring the bounded stream installs its listener before readiness is observable. - const live = yield* EventV2.liveBounded(events, subscriberCapacity) + const live = yield* EventV2.liveBounded(events, { + capacity: subscriberCapacity, + accept: isOpenCodeEvent, + }) return Stream.make(connected).pipe(Stream.concat(live)) }), ).pipe(Stream.map(eventData), Stream.pipeThroughChannel(Sse.encode()))