refactor(server): share event stream encoding

packages/core/src/event.ts

@@ -1,6 +1,6 @@
 export * as EventV2 from "./event"
 
-import { Cause, Context, DateTime, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect"
+import { Cause, Context, DateTime, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
 import type { EventLog } from "@opencode-ai/schema/event-log"
@@ -89,11 +89,6 @@ const decodeSerializedEvent = (event: SerializedEvent): Payload => {
   }
 }
 
-export class SubscriberOverflowError extends Schema.TaggedErrorClass<SubscriberOverflowError>()(
-  "EventV2.SubscriberOverflow",
-  { capacity: Schema.Int },
-) {}
-
 export const versionedType = Event.versionedType
 export const durable = Event.durable
 export const ephemeral = Event.ephemeral
@@ -167,27 +162,6 @@ export interface Interface {
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/Event") {}
 
-export const liveBounded = (
-  events: Interface,
-  options: { readonly capacity: number; readonly accept?: (event: Payload) => boolean },
-) =>
-  Effect.gen(function* () {
-    const queue = yield* Queue.dropping<Payload, SubscriberOverflowError>(options.capacity)
-    const unsubscribe = yield* events.listen((event) =>
-      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)
-  })
-
 export interface LayerOptions {
   readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect<void>
   /** Maximum durable rows read per page while replaying or tailing an aggregate log. */

packages/core/test/event.test.ts

@@ -2,8 +2,6 @@ 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"
@@ -363,51 +361,6 @@ describe("EventV2", () => {
     }),
   )
 
-  it.effect("ends only an overflowing bounded subscriber without blocking other listeners", () =>
-    Effect.gen(function* () {
-      const events = yield* EventV2.Service
-      const consuming = yield* Deferred.make<void>()
-      const release = yield* Deferred.make<void>()
-      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,
-      )
-      const fast = yield* fastStream.pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
-
-      yield* events.publish(Message, { text: "one" })
-      yield* Deferred.await(consuming)
-      yield* events.publish(Message, { text: "two" })
-      yield* events.publish(Message, { text: "overflow" })
-      const last = yield* events.publish(Message, { text: "still delivered" })
-      yield* Deferred.succeed(release, undefined)
-
-      const slowExit = yield* Fiber.await(slow)
-      expect(Exit.findErrorOption(slowExit).pipe(Option.getOrUndefined)).toBeInstanceOf(EventV2.SubscriberOverflowError)
-      expect(Array.from(yield* Fiber.join(fast))).toEqual([
-        expect.objectContaining({ data: { text: "one" } }),
-        expect.objectContaining({ data: { text: "two" } }),
-        expect.objectContaining({ data: { text: "overflow" } }),
-        last,
-      ])
-    }),
-  )
-
-  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

packages/server/package.json

@@ -9,6 +9,7 @@
     "./*": "./src/*.ts"
   },
   "scripts": {
+    "test": "bun test --only-failures",
     "typecheck": "tsgo --noEmit"
   },
   "dependencies": {

packages/server/src/event-feed.ts

@@ -0,0 +1,90 @@
+export * as EventFeed from "./event-feed"
+
+import { EventV2 } from "@opencode-ai/core/event"
+import { isOpenCodeEvent, OpenCodeEvent } from "@opencode-ai/protocol/groups/event"
+import { Cause, Context, Effect, Layer, Queue, Schema, Scope, Stream } from "effect"
+
+export const SubscriberCapacity = 4_096
+
+export class SubscriberOverflowError extends Schema.TaggedErrorClass<SubscriberOverflowError>()(
+  "EventFeed.SubscriberOverflow",
+  { capacity: Schema.Int },
+) {}
+
+export class EncodingError extends Schema.TaggedErrorClass<EncodingError>()("EventFeed.EncodingError", {
+  eventID: EventV2.ID,
+  eventType: Schema.String,
+  cause: Schema.Defect(),
+}) {}
+
+export type Error = SubscriberOverflowError | EncodingError
+
+export interface Interface {
+  readonly subscribe: Effect.Effect<Stream.Stream<string, Error>, never, Scope.Scope>
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/server/EventFeed") {}
+
+const encode = Schema.encodeUnknownSync(OpenCodeEvent)
+
+export function frame(event: OpenCodeEvent) {
+  return `data: ${JSON.stringify(encode(event))}\n\n`
+}
+
+export const make = Effect.fn("EventFeed.make")(function* (
+  observe: (subscriber: EventV2.Subscriber) => Effect.Effect<EventV2.Unsubscribe>,
+  options?: { readonly capacity?: number; readonly encode?: (event: OpenCodeEvent) => string },
+) {
+  const capacity = options?.capacity ?? SubscriberCapacity
+  const render = options?.encode ?? frame
+  const subscribers = new Set<Queue.Queue<string, Error>>()
+
+  const fail = (error: Error) =>
+    Effect.sync(() => {
+      const current = Array.from(subscribers)
+      subscribers.clear()
+      for (const subscriber of current) Queue.failCauseUnsafe(subscriber, Cause.fail(error))
+    })
+
+  const publish = Effect.fnUntraced(function* (event: EventV2.Payload) {
+    if (!isOpenCodeEvent(event)) return
+    if (subscribers.size === 0) return
+    const encoded = yield* Effect.try({
+      try: () => render(event),
+      catch: (cause) => new EncodingError({ eventID: event.id, eventType: event.type, cause }),
+    }).pipe(
+      Effect.catch((error) =>
+        Effect.logError("Failed to encode public event", {
+          eventID: error.eventID,
+          eventType: error.eventType,
+          cause: error.cause,
+        }).pipe(Effect.andThen(fail(error)), Effect.as(undefined)),
+      ),
+    )
+    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 })))
+    }
+  })
+
+  const unsubscribe = yield* observe(publish)
+  yield* Effect.addFinalizer(() => unsubscribe)
+
+  return Service.of({
+    subscribe: Effect.acquireRelease(
+      Queue.dropping<string, Error>(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)),
+  })
+})
+
+export const layer = Layer.effect(
+  Service,
+  Effect.gen(function* () {
+    const events = yield* EventV2.Service
+    return yield* make(events.listen)
+  }),
+)

packages/server/src/handlers.ts

@@ -26,6 +26,7 @@ import { CredentialHandler } from "./handlers/credential"
 import { ProjectHandler } from "./handlers/project"
 import { ProjectCopyHandler } from "./handlers/project-copy"
 import { VcsHandler } from "./handlers/vcs"
+import { EventFeed } from "./event-feed"
 
 export const handlers = Layer.mergeAll(
   HealthHandler,
@@ -48,7 +49,7 @@ export const handlers = Layer.mergeAll(
   FileSystemHandler,
   CommandHandler,
   SkillHandler,
-  EventHandler,
+  EventHandler.pipe(Layer.provide(EventFeed.layer)),
   PtyHandler,
   ShellHandler,
   QuestionHandler,

packages/server/src/handlers/event.ts

@@ -1,44 +1,23 @@
 import { EventV2 } from "@opencode-ai/core/event"
-import { isOpenCodeEvent, OpenCodeEvent } from "@opencode-ai/protocol/groups/event"
-import { Effect, Schema, Stream } from "effect"
-import { Sse } from "effect/unstable/encoding"
+import { Effect, Stream } from "effect"
 import { HttpServerResponse } from "effect/unstable/http"
 import { HttpApiBuilder } from "effect/unstable/httpapi"
 import { Api } from "../api"
-
-// Session execution emits dense event bursts; allow healthy SSE clients enough
-// time to absorb one without weakening the bounded slow-subscriber failure.
-const subscriberCapacity = 4_096
-
-function eventData(data: unknown): Sse.Event {
-  return {
-    _tag: "Event",
-    event: "message",
-    id: undefined,
-    data: JSON.stringify(Schema.encodeUnknownSync(OpenCodeEvent)(data)),
-  }
-}
+import { EventFeed } from "../event-feed"
 
 export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) =>
   Effect.gen(function* () {
-    const events = yield* EventV2.Service
+    const feed = yield* EventFeed.Service
     return handlers.handleRaw("event.subscribe", () =>
       Effect.gen(function* () {
         const connected = {
           id: EventV2.ID.create(),
           type: "server.connected",
           data: {},
-        }
+        } as const
         const output = Stream.unwrap(
-          Effect.gen(function* () {
-            // Acquiring the bounded stream installs its listener before readiness is observable.
-            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()))
+          feed.subscribe.pipe(Effect.map((live) => Stream.make(EventFeed.frame(connected)).pipe(Stream.concat(live)))),
+        )
         const heartbeat = Stream.tick("15 seconds").pipe(Stream.map(() => ": heartbeat\n\n"))
         return HttpServerResponse.stream(
           output.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }), Stream.encodeText),