refactor(server): share event stream encoding (#36484)

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),

packages/server/test/event-feed.test.ts

@@ -0,0 +1,155 @@
+import { describe, expect, test } from "bun:test"
+import { AgentV2 } from "@opencode-ai/core/agent"
+import { EventV2 } from "@opencode-ai/core/event"
+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"
+import { EventFeed } from "../src/event-feed"
+
+const Internal = EventV2.ephemeral({ type: "test.internal", schema: { value: Schema.String } })
+
+const event = (id: string): EventV2.Payload<typeof AgentV2.Event.Updated> => ({
+  id: EventV2.ID.make(`evt_${id}`),
+  created: DateTime.makeUnsafe(Date.now()),
+  type: AgentV2.Event.Updated.type,
+  data: {},
+})
+
+const internal = (value: string): EventV2.Payload<typeof Internal> => ({
+  id: EventV2.ID.create(),
+  created: DateTime.makeUnsafe(Date.now()),
+  type: Internal.type,
+  data: { value },
+})
+
+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)),
+  }
+}
+
+describe("EventFeed", () => {
+  test("preserves the public SSE frame encoding", () => {
+    const payload = event("wire")
+    expect(EventFeed.frame(payload)).toBe(
+      `data: ${JSON.stringify(Schema.encodeUnknownSync(OpenCodeEvent)(payload))}\n\n`,
+    )
+  })
+
+  it.effect("encodes once and delivers the same frame to every subscriber", () =>
+    Effect.gen(function* () {
+      let encodes = 0
+      const source = makeSource()
+      const feed = yield* EventFeed.make(source.observe, {
+        encode: (event) => {
+          encodes += 1
+          return event.type
+        },
+      })
+      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)
+
+      yield* source.publish(event("example"))
+
+      expect([Array.from(yield* Fiber.join(left)), Array.from(yield* Fiber.join(right))]).toEqual([
+        [AgentV2.Event.Updated.type],
+        [AgentV2.Event.Updated.type],
+      ])
+      expect(encodes).toBe(1)
+    }),
+  )
+
+  it.effect("fails only the subscriber that exceeds its lag capacity", () =>
+    Effect.gen(function* () {
+      const source = makeSource()
+      const feed = yield* EventFeed.make(source.observe, {
+        capacity: 1,
+        encode: (event) => event.id,
+      })
+      const slow = yield* feed.subscribe
+      const fast = yield* feed.subscribe
+      const first = yield* Deferred.make<void>()
+      const second = yield* Deferred.make<void>()
+      const received = new Array<string>()
+      const fastFiber = yield* fast.pipe(
+        Stream.take(3),
+        Stream.runForEach((frame) =>
+          Effect.sync(() => received.push(frame)).pipe(
+            Effect.andThen(
+              frame === "evt_one"
+                ? Deferred.succeed(first, undefined)
+                : frame === "evt_two"
+                  ? Deferred.succeed(second, undefined)
+                  : Effect.void,
+            ),
+          ),
+        ),
+        Effect.forkScoped,
+      )
+
+      yield* source.publish(event("one"))
+      yield* Deferred.await(first)
+      yield* source.publish(event("two"))
+      yield* Deferred.await(second)
+      yield* source.publish(event("three"))
+
+      yield* Fiber.join(fastFiber)
+
+      const result = yield* slow.pipe(Stream.runCollect, Effect.exit)
+      expect(received).toEqual(["evt_one", "evt_two", "evt_three"])
+      expect(Exit.isFailure(result)).toBeTrue()
+      if (Exit.isSuccess(result)) return
+      expect(Option.getOrUndefined(Exit.findErrorOption(result))).toBeInstanceOf(EventFeed.SubscriberOverflowError)
+    }),
+  )
+
+  it.effect("filters internal events before they consume subscriber capacity", () =>
+    Effect.gen(function* () {
+      const source = makeSource()
+      const feed = yield* EventFeed.make(source.observe, { capacity: 1, encode: (event) => event.type })
+      const stream = yield* feed.subscribe
+
+      yield* source.publish(internal("one"))
+      yield* source.publish(internal("two"))
+      yield* source.publish(event("public"))
+
+      expect(Array.from(yield* stream.pipe(Stream.take(1), Stream.runCollect))).toEqual([AgentV2.Event.Updated.type])
+    }),
+  )
+
+  it.effect("disconnects current subscribers after an encoding failure and continues for later subscribers", () =>
+    Effect.gen(function* () {
+      const source = makeSource()
+      const feed = yield* EventFeed.make(source.observe, {
+        encode: (event) => {
+          if (event.id === EventV2.ID.make("evt_bad")) throw new Error("invalid event")
+          return event.id
+        },
+      })
+      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 received = yield* next.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      yield* source.publish(event("good"))
+
+      expect(Exit.isFailure(exit)).toBeTrue()
+      if (Exit.isSuccess(exit)) return
+      expect(Option.getOrUndefined(Exit.findErrorOption(exit))).toBeInstanceOf(EventFeed.EncodingError)
+      expect(Array.from(yield* Fiber.join(received))).toEqual(["evt_good"])
+    }),
+  )
+})