Session.fork publishes its copy in one transaction

packages/core/src/event.ts

@@ -123,12 +123,35 @@ export interface PublishOptions {
   readonly commit?: (seq: number) => Effect.Effect<void>
 }
 
+/** documentor: one durable event of a {@link Interface.publishMany} batch. */
+export interface Publication<D extends Definition = Definition> {
+  readonly definition: D
+  readonly data: Data<D>
+}
+
+/** documentor: pairs a definition with its data, typed, for {@link Interface.publishMany}. */
+export const publication = <D extends Definition>(definition: D, data: Data<D>): Publication =>
+  ({ definition, data }) as Publication
+
+/** documentor: events per INSERT in publishMany. Parts carry tool output, so a long chat's rows can be large. */
+const PUBLISH_MANY_CHUNK = 200
+
 export interface Interface {
   readonly publish: <D extends Definition>(
     definition: D,
     data: Data<D>,
     options?: PublishOptions,
   ) => Effect.Effect<Payload<D>>
+  /**
+   * documentor: publishes durable events of one aggregate in a single transaction, in order.
+   * Each event is projected and notified exactly as `publish` would; only the round trips
+   * of the per-event transaction (lock, sequence read, sequence and event writes) are shared.
+   * All or nothing: a failing projector commits none of them.
+   */
+  readonly publishMany: (
+    publications: readonly Publication[],
+    options?: { readonly location?: Location.Ref },
+  ) => Effect.Effect<Payload[]>
   readonly subscribe: <D extends Definition>(definition: D) => Stream.Stream<Payload<D>>
   readonly all: () => Stream.Stream<Payload>
   readonly durable: (input: { readonly aggregateID: string; readonly after?: number }) => Stream.Stream<Payload>
@@ -438,6 +461,104 @@ export const layerWith = (options?: LayerOptions) =>
         })
       }
 
+      // documentor: commitDurableEvent's non-replay path for a batch of one aggregate. A separate
+      // copy rather than a shared helper, so upstream changes to that function rebase cleanly.
+      function publishMany(publications: readonly Publication[], options?: { readonly location?: Location.Ref }) {
+        return Effect.gen(function* () {
+          if (publications.length === 0) return [] as Payload[]
+          const serviceLocation = Option.getOrUndefined(yield* Effect.serviceOption(Location.Service))
+          const location =
+            options?.location ??
+            (serviceLocation
+              ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID }
+              : undefined)
+          let aggregateID: string | undefined
+          const pending = new Array<{ definition: Definition; event: Payload }>()
+          for (const { definition, data } of publications) {
+            const durable = definition.durable
+            const aggregate = durable ? (data as Record<string, unknown>)[durable.aggregate] : undefined
+            if (!durable || typeof aggregate !== "string")
+              return yield* Effect.die(
+                new InvalidDurableEventError({
+                  type: definition.type,
+                  message: "publishMany takes durable events with a string aggregate field only",
+                }),
+              )
+            if (aggregateID !== undefined && aggregate !== aggregateID)
+              return yield* Effect.die(
+                new InvalidDurableEventError({
+                  type: definition.type,
+                  message: `Aggregate mismatch: expected ${aggregateID}, got ${aggregate}`,
+                }),
+              )
+            aggregateID = aggregate
+            pending.push({
+              definition,
+              event: { id: ID.create(), type: definition.type, ...(location ? { location } : {}), data } as Payload,
+            })
+          }
+          const aggregate = aggregateID!
+          const committed = yield* Effect.uninterruptible(
+            Effect.gen(function* () {
+              const events = yield* db
+                .transaction(
+                  () =>
+                    Effect.gen(function* () {
+                      const latest = yield* latestSequence(db, aggregate)
+                      const rows = new Array<typeof EventTable.$inferInsert>()
+                      const events = new Array<Payload>()
+                      for (const [index, { definition, event }] of pending.entries()) {
+                        const durable = definition.durable!
+                        const seq = latest + 1 + index
+                        const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record<string, unknown>
+                        const committed = {
+                          ...event,
+                          durable: { aggregateID: aggregate, seq, version: durable.version },
+                        } as Payload
+                        for (const projector of projectors.get(event.type) ?? []) {
+                          yield* projector(committed)
+                        }
+                        rows.push({
+                          id: event.id,
+                          aggregate_id: aggregate,
+                          seq,
+                          type: versionedType(definition.type, durable.version),
+                          data: encoded,
+                        })
+                        events.push(committed)
+                      }
+                      const last = latest + pending.length
+                      yield* db
+                        .insert(EventSequenceTable)
+                        .values([{ aggregate_id: aggregate, seq: last }])
+                        .onConflictDoUpdate({ target: EventSequenceTable.aggregate_id, set: { seq: last } })
+                        .run()
+                        .pipe(Effect.orDie)
+                      for (let start = 0; start < rows.length; start += PUBLISH_MANY_CHUNK) {
+                        yield* db
+                          .insert(EventTable)
+                          .values(rows.slice(start, start + PUBLISH_MANY_CHUNK))
+                          .run()
+                          .pipe(Effect.orDie)
+                      }
+                      return events
+                    }),
+                  { behavior: "immediate" },
+                )
+                .pipe(Effect.orDie)
+              yield* Effect.forEach(pubsub.durable.get(aggregate) ?? [], (wake) => PubSub.publish(wake, undefined), {
+                discard: true,
+              })
+              return events
+            }),
+          )
+          for (const event of committed) {
+            yield* notify(event, true)
+          }
+          return committed
+        })
+      }
+
       function replay(
         event: SerializedEvent,
         options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean },
@@ -621,6 +742,7 @@ export const layerWith = (options?: LayerOptions) =>
 
       return Service.of({
         publish,
+        publishMany,
         subscribe,
         all: streamAll,
         durable,

packages/core/test/event-publish-many.test.ts

@@ -0,0 +1,204 @@
+import { describe, expect } from "bun:test"
+import { Effect, Exit, Fiber, Schema, Stream } from "effect"
+import { EventV2 } from "@opencode-ai/core/event"
+import { Database } from "@opencode-ai/core/database/database"
+import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
+import { LayerNode } from "@opencode-ai/core/effect/layer-node"
+import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql"
+import { asc, eq } from "drizzle-orm"
+import { SessionV1 } from "@opencode-ai/schema/session-v1"
+import { Session } from "@opencode-ai/schema/session"
+import { testEffect } from "./lib/effect"
+
+const Note = EventV2.define({
+  type: "test.batch.note",
+  durable: { version: 1, aggregate: "id" },
+  schema: { id: Schema.String, text: Schema.String },
+})
+
+const Tag = EventV2.define({
+  type: "test.batch.tag",
+  durable: { version: 2, aggregate: "id" },
+  schema: { id: Schema.String, label: Schema.String },
+})
+
+const Live = EventV2.define({
+  type: "test.batch.live",
+  schema: { id: Schema.String },
+})
+
+const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node])))
+
+const rowsOf = (aggregateID: string) =>
+  Effect.gen(function* () {
+    const { db } = yield* Database.Service
+    return yield* db
+      .select()
+      .from(EventTable)
+      .where(eq(EventTable.aggregate_id, aggregateID))
+      .orderBy(asc(EventTable.seq))
+      .all()
+      .pipe(Effect.orDie)
+  })
+
+const batch = (aggregateID: string, count: number) =>
+  Array.from({ length: count }, (_, index) =>
+    index % 3 === 2
+      ? EventV2.publication(Tag, { id: aggregateID, label: `tag ${index}` })
+      : EventV2.publication(Note, { id: aggregateID, text: `note ${index}` }),
+  )
+
+describe("documentor: EventV2.publishMany", () => {
+  it.effect("stores the same rows as publishing the events one by one", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const batched = EventV2.ID.create()
+      const sequential = EventV2.ID.create()
+
+      yield* events.publishMany(batch(batched, 7))
+      for (const { definition, data } of batch(sequential, 7)) {
+        yield* events.publish(definition, data as never)
+      }
+
+      const shape = (rows: (typeof EventTable.$inferSelect)[], aggregateID: string) =>
+        rows.map((row) => ({ seq: row.seq, type: row.type, data: { ...row.data, id: row.data.id === aggregateID } }))
+      expect(shape(yield* rowsOf(batched), batched)).toEqual(shape(yield* rowsOf(sequential), sequential))
+      expect((yield* rowsOf(batched)).map((row) => row.seq)).toEqual([0, 1, 2, 3, 4, 5, 6])
+    }),
+  )
+
+  it.effect("continues the aggregate's sequence and is followed by single publishes", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.publish(Note, { id: aggregateID, text: "before" })
+      const published = yield* events.publishMany(batch(aggregateID, 3))
+      yield* events.publish(Note, { id: aggregateID, text: "after" })
+
+      expect(published.map((event) => event.durable?.seq)).toEqual([1, 2, 3])
+      expect((yield* rowsOf(aggregateID)).map((row) => row.seq)).toEqual([0, 1, 2, 3, 4])
+      const sequence = yield* db
+        .select()
+        .from(EventSequenceTable)
+        .where(eq(EventSequenceTable.aggregate_id, aggregateID))
+        .get()
+        .pipe(Effect.orDie)
+      expect(sequence?.seq).toBe(4)
+    }),
+  )
+
+  it.effect("projects every event in order, then notifies each one after the commit", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const aggregateID = EventV2.ID.create()
+      const order = new Array<string>()
+      yield* events.project(Note, (event) => Effect.sync(() => order.push(`project ${event.data.text}`)))
+      yield* events.listen((event) =>
+        Effect.sync(() => {
+          if ((event.data as { id?: string }).id === aggregateID) order.push(`notify ${event.durable?.seq}`)
+        }),
+      )
+      const typed = yield* events.subscribe(Note).pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+
+      yield* events.publishMany([
+        EventV2.publication(Note, { id: aggregateID, text: "a" }),
+        EventV2.publication(Note, { id: aggregateID, text: "b" }),
+      ])
+
+      expect(order).toEqual(["project a", "project b", "notify 0", "notify 1"])
+      expect(Array.from(yield* Fiber.join(typed)).map((event) => event.data.text)).toEqual(["a", "b"])
+    }),
+  )
+
+  it.effect("commits nothing and notifies no one when a projector fails mid-batch", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+      const notified = new Array<string>()
+      yield* db.run("CREATE TABLE IF NOT EXISTS publish_many_probe (value text NOT NULL)")
+      yield* db.run("DELETE FROM publish_many_probe")
+      yield* events.project(Note, (event) =>
+        event.data.text === "boom"
+          ? Effect.die("projector failed")
+          : db
+              .run(`INSERT INTO publish_many_probe (value) VALUES ('${event.data.text}')`)
+              .pipe(Effect.orDie, Effect.asVoid),
+      )
+      yield* events.listen((event) => Effect.sync(() => notified.push(event.id)))
+
+      const exit = yield* events
+        .publishMany([
+          EventV2.publication(Note, { id: aggregateID, text: "kept?" }),
+          EventV2.publication(Note, { id: aggregateID, text: "boom" }),
+        ])
+        .pipe(Effect.exit)
+
+      expect(Exit.isFailure(exit)).toBeTrue()
+      expect(yield* db.all("SELECT value FROM publish_many_probe")).toEqual([])
+      expect(yield* rowsOf(aggregateID)).toEqual([])
+      expect(
+        yield* db.select().from(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).all(),
+      ).toEqual([])
+      expect(notified).toEqual([])
+    }),
+  )
+
+  it.effect("refuses a batch that spans aggregates or holds a live-only event", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const first = EventV2.ID.create()
+      const second = EventV2.ID.create()
+
+      const mixed = yield* events
+        .publishMany([
+          EventV2.publication(Note, { id: first, text: "a" }),
+          EventV2.publication(Note, { id: second, text: "b" }),
+        ])
+        .pipe(Effect.exit)
+      const live = yield* events.publishMany([EventV2.publication(Live, { id: first })]).pipe(Effect.exit)
+
+      expect(String(mixed)).toContain("Aggregate mismatch")
+      expect(String(live)).toContain("durable events")
+      expect(yield* rowsOf(first)).toEqual([])
+      expect(yield* events.publishMany([])).toEqual([])
+    }),
+  )
+
+  it.effect("wakes a durable aggregate stream that is tailing", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      // The stream decodes through the durable manifest, which knows real event types only.
+      const aggregateID = Session.ID.create()
+      const fiber = yield* events.durable({ aggregateID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+
+      yield* events.publishMany(
+        Array.from({ length: 4 }, (_, index) =>
+          EventV2.publication(SessionV1.Event.MessageRemoved, {
+            sessionID: aggregateID,
+            messageID: SessionV1.MessageID.ascending(`msg_${index}`),
+          }),
+        ),
+      )
+
+      expect(Array.from(yield* Fiber.join(fiber)).map((event) => event.durable?.seq)).toEqual([0, 1, 2, 3])
+    }),
+  )
+
+  it.effect("writes a batch larger than one insert chunk", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.publishMany(batch(aggregateID, 450))
+
+      const rows = yield* rowsOf(aggregateID)
+      expect(rows).toHaveLength(450)
+      expect(rows.map((row) => row.seq)).toEqual(Array.from({ length: 450 }, (_, index) => index))
+    }),
+  )
+})

packages/core/test/session-runner-tool-events.test.ts

@@ -26,6 +26,7 @@ const capture = () => {
         })
         return event
       }),
+    publishMany: () => Effect.succeed([]),
     subscribe: () => Stream.empty,
     all: () => Stream.empty,
     durable: () => Stream.empty,

packages/opencode/src/event-v2-bridge.ts

@@ -32,6 +32,22 @@ const layer = Layer.effect(
         })
       })
 
+    // documentor: the same location routing as `publish`, for a batch.
+    const publishMany: EventV2.Interface["publishMany"] = (publications, options) =>
+      Effect.gen(function* () {
+        if (options?.location) return yield* events.publishMany(publications, options)
+        const ctx = yield* InstanceRef
+        if (!ctx) return yield* events.publishMany(publications, options)
+        const workspaceID = yield* WorkspaceRef
+        return yield* events.publishMany(publications, {
+          location: new Location.Info({
+            directory: AbsolutePath.make(ctx.directory),
+            ...(workspaceID ? { workspaceID } : {}),
+            project: { id: Project.ID.make(ctx.project.id), directory: AbsolutePath.make(ctx.worktree) },
+          }),
+        })
+      })
+
     const unsubscribe = yield* events.listen((event) =>
       Effect.gen(function* () {
         const ctx = yield* InstanceRef
@@ -62,7 +78,7 @@ const layer = Layer.effect(
     )
     yield* Effect.addFinalizer(() => unsubscribe)
 
-    return Service.of({ ...events, publish })
+    return Service.of({ ...events, publish, publishMany })
   }),
 )