chore: merge dev into v2 (#34317)

Co-authored-by: Brendan Allan <14191578+Brendonovich@users.noreply.github.com>
Co-authored-by: opencode-agent[bot] <opencode-agent[bot]@users.noreply.github.com>
Co-authored-by: Affan Ali <93028901+affanali2k3@users.noreply.github.com>
Co-authored-by: affanali2k3 <affanalikhanxx@gmail.com>
Co-authored-by: Frank <frank@anoma.ly>
Co-authored-by: opencode-agent[bot] <219766164+opencode-agent[bot]@users.noreply.github.com>
Co-authored-by: James Long <longster@gmail.com>
Co-authored-by: Aiden Cline <63023139+rekram1-node@users.noreply.github.com>
Co-authored-by: Jay V <air@live.ca>
Co-authored-by: Dax Raad <d@ironbay.co>
Co-authored-by: Aarav Sareen <96787824+arvsrn@users.noreply.github.com>
Co-authored-by: OpeOginni <107570612+OpeOginni@users.noreply.github.com>
Co-authored-by: Luke Parker <10430890+Hona@users.noreply.github.com>
Co-authored-by: Ben Guthrie <benjee.012@gmail.com>
Co-authored-by: Dax <mail@thdxr.com>
Co-authored-by: Filip <34747899+neriousy@users.noreply.github.com>

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