ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(protocol): keep internal events off SSE (#35378)

Browse source
This commit is contained in:
Kit Langton 2026-07-04 21:12:45 -04:00 • committed by GitHub
commit 905123b9c0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 52 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

21
packages/core/src/event.ts
View file

@ -160,15 +160,22 @@ export interface Interface {
export class Service extends Context.Service<Service, Interface>()("@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<Payload, SubscriberOverflowError>(capacity)
const queue = yield* Queue.dropping<Payload, SubscriberOverflowError>(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)

4
packages/core/src/plugin/host.ts
View file

@ -23,8 +23,6 @@ import { ToolHooks } from "../tool/hooks"
import { WorkspaceV2 } from "../workspace"
const mutable = <T>(value: T) => value as DeepMutable<T>
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()),

20
packages/core/test/event.test.ts
View file

@ -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<void>()
const release = yield* Deferred.make<void>()
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