feat(sdk): expose live event stream (#34098)

packages/core/src/event.ts

@@ -1,6 +1,6 @@
 export * as EventV2 from "./event"
 
-import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
+import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
 import { and, asc, eq, gt, inArray } from "drizzle-orm"
@@ -107,6 +107,11 @@ export const readAggregate = Effect.fn("EventV2.readAggregate")(function* <A>(
   }
 })
 
+export class SubscriberOverflowError extends Schema.TaggedErrorClass<SubscriberOverflowError>()(
+  "EventV2.SubscriberOverflow",
+  { capacity: Schema.Int },
+) {}
+
 export const define = Event.define
 export const versionedType = Event.versionedType
 
@@ -144,6 +149,20 @@ export interface Interface {
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/Event") {}
 
+export const allBounded = (events: Interface, capacity: number) =>
+  Effect.gen(function* () {
+    const queue = yield* Queue.dropping<Payload, SubscriberOverflowError>(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),
+        ),
+      ),
+    )
+    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>
 }

packages/core/test/event.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect"
+import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Schema, Stream } from "effect"
 import { EventV2 } from "@opencode-ai/core/event"
 import { Event } from "@opencode-ai/schema/event"
 import { Session } from "@opencode-ai/schema/session"
@@ -285,6 +285,69 @@ describe("EventV2", () => {
     }),
   )
 
+  it.effect("notifies global listeners only after a durable event is committed", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+      const observed = new Array<{ id: string; seq: number }>()
+      yield* events.listen((event) =>
+        event.type !== SyncMessage.type
+          ? Effect.void
+          : db
+              .select({ id: EventTable.id, seq: EventTable.seq })
+              .from(EventTable)
+              .where(eq(EventTable.id, event.id))
+              .get()
+              .pipe(
+                Effect.orDie,
+                Effect.tap((row) =>
+                  Effect.sync(() => {
+                    if (row) observed.push(row)
+                  }),
+                ),
+                Effect.asVoid,
+              ),
+      )
+
+      const event = yield* events.publish(SyncMessage, { id: aggregateID, text: "committed" })
+      if (!event.durable) throw new Error("Expected durable event metadata")
+
+      expect(observed).toEqual([{ id: event.id, seq: event.durable.seq }])
+    }),
+  )
+
+  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.allBounded(events, 1)
+      const fastStream = yield* EventV2.allBounded(events, 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("preserves observer interruption", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service