perf(core): stop writing duplicate snapshot events in local mode

packages/core/src/event.ts

@@ -3,7 +3,7 @@ export * as EventV2 from "./event"
 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"
+import { and, asc, eq, gt, inArray, sql } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
 import { Location } from "./location"
@@ -121,6 +121,13 @@ export interface PublishOptions {
   readonly location?: Location.Ref
   /** Local operational projection committed atomically with a new durable event. Not replayed or serialized. */
   readonly commit?: (seq: number) => Effect.Effect<void>
+  /**
+   * When false, the durable event is projected locally but NOT persisted to the
+   * event table or sequence. The payload is still notified to in-process
+   * listeners (SSE, UI) but carries no `durable` envelope, so cross-instance
+   * sync does not observe it. Defaults to true (full event sourcing).
+   */
+  readonly persist?: boolean
 }
 
 export interface Interface {
@@ -212,6 +219,7 @@ export const layerWith = (options?: LayerOptions) =>
           readonly strictOwner?: boolean
         },
         commit?: (seq: number) => Effect.Effect<void>,
+        persist = true,
       ) {
         return Effect.gen(function* () {
           const durable = definition?.durable
@@ -234,6 +242,40 @@ export const layerWith = (options?: LayerOptions) =>
                 )
               }
               const list = projectors.get(event.type) ?? []
+              if (!persist) {
+                // Local-only publish: project the event into the operational
+                // tables (MessageTable/PartTable/SessionTable) atomically, but
+                // do not append to the durable event log or advance the
+                // aggregate sequence. Returning undefined signals the caller to
+                // notify listeners with no `durable` envelope, so cross-instance
+                // sync never observes this event. The `commit` hook is not
+                // invoked either: it is documented as requiring a committed seq,
+                // and no caller combines `commit` with `persist:false`. The
+                // projector receives `durable.seq = -1` as a placeholder; none
+                // of the current projectors read seq (they upsert by entity id),
+                // so the value is inert — kept only to satisfy the Payload type.
+                return yield* Effect.uninterruptible(
+                  Effect.gen(function* () {
+                    yield* db
+                      .transaction(
+                        () =>
+                          Effect.gen(function* () {
+                            const committed = {
+                              ...event,
+                              durable: { aggregateID, seq: -1, version: durable.version },
+                            } as Payload
+                            for (const projector of list) {
+                              yield* projector(committed)
+                            }
+                            return
+                          }),
+                        { behavior: "immediate" },
+                      )
+                      .pipe(Effect.orDie)
+                    return undefined
+                  }),
+                )
+              }
               return yield* Effect.uninterruptible(
                 Effect.gen(function* () {
                   const committed = yield* db
@@ -366,7 +408,12 @@ export const layerWith = (options?: LayerOptions) =>
         })
       }
 
-      function publishEvent<D extends Definition>(definition: D, event: Payload<D>, commit?: PublishOptions["commit"]) {
+      function publishEvent<D extends Definition>(
+        definition: D,
+        event: Payload<D>,
+        commit?: PublishOptions["commit"],
+        persist = true,
+      ) {
         return Effect.gen(function* () {
           if (!definition?.durable && commit)
             return yield* Effect.die(
@@ -376,7 +423,7 @@ export const layerWith = (options?: LayerOptions) =>
               }),
             )
           if (definition?.durable) {
-            const committed = yield* commitDurableEvent(definition, event as Payload, undefined, commit)
+            const committed = yield* commitDurableEvent(definition, event as Payload, undefined, commit, persist)
             if (committed) {
               event = {
                 ...event,
@@ -434,6 +481,7 @@ export const layerWith = (options?: LayerOptions) =>
               data,
             } as Payload<D>,
             options?.commit,
+            options?.persist ?? true,
           )
         })
       }
@@ -636,3 +684,84 @@ export const layerWith = (options?: LayerOptions) =>
 
 const layer = layerWith()
 export const node = makeGlobalNode({ service: Service, layer: layer, deps: [Database.node] })
+
+export const SNAPSHOT_TYPES = ["message.updated", "message.part.updated"] as const
+
+/**
+ * Compact snapshot-like durable events, keeping only the latest occurrence per
+ * (aggregate, type, entity) and deleting intermediate full-state copies.
+ *
+ * These events carry the complete message/part payload on every update, so a
+ * single message renders N rows whose payloads are total supersets of their
+ * predecessors. Replaying the retained latest row reproduces the identical
+ * final projection (the projector upserts by id), while intermediate rows are
+ * pure write amplification.
+ *
+ * Deleting rows leaves `seq` gaps; sequence stream readers (`readAfter`,
+ * `history`) use `seq > after` so gaps are transparent. Replay packets are
+ * re-cost in the sender and validated against their own emitted `seq` order,
+ * not against DB adjacency, so gaps are also safe for sync replay.
+ *
+ * Non-snapshot lifecycle rows (`session.created`, `message.removed`,
+ * `message.part.delta`, ...) are never touched, and `event_sequence` is left
+ * at its current high-water mark.
+ */
+export const compactSnapshotEvents = Effect.fn("EventV2.compactSnapshotEvents")(function* (
+  db: Database.Interface["db"],
+) {
+  const snapshotTypes = SNAPSHOT_TYPES.map((type) => versionedType(type, 1))
+  const stats = yield* db
+    .select({
+      rows: sql<number>`count(*)`,
+      bytes: sql<number>`sum(length(data))`,
+    })
+    .from(EventTable)
+    .where(inArray(EventTable.type, snapshotTypes))
+    .get()
+    .pipe(Effect.orDie)
+  yield* db
+    .run(
+      sql.raw(`
+        DELETE FROM "event"
+        WHERE "type" IN ('message.updated.1', 'message.part.updated.1')
+          AND "id" NOT IN (
+            SELECT "id" FROM (
+              SELECT
+                "id",
+                ROW_NUMBER() OVER (
+                  PARTITION BY "aggregate_id", "type", "entity"
+                  ORDER BY "seq" DESC
+                ) AS "rn"
+              FROM (
+                SELECT
+                  "id",
+                  "aggregate_id",
+                  "type",
+                  "seq",
+                  CASE "type"
+                    WHEN 'message.updated.1' THEN json_extract("data", '$.info.id')
+                    WHEN 'message.part.updated.1' THEN json_extract("data", '$.part.id')
+                  END AS "entity"
+                FROM "event"
+                WHERE "type" IN ('message.updated.1', 'message.part.updated.1')
+              )
+            )
+            WHERE "rn" = 1
+          )
+      `),
+    )
+    .pipe(Effect.orDie)
+  const remaining = yield* db
+    .select({
+      rows: sql<number>`count(*)`,
+      bytes: sql<number>`sum(length(data))`,
+    })
+    .from(EventTable)
+    .where(inArray(EventTable.type, snapshotTypes))
+    .get()
+    .pipe(Effect.orDie)
+  return {
+    removed: (stats?.rows ?? 0) - (remaining?.rows ?? 0),
+    bytes: (stats?.bytes ?? 0) - (remaining?.bytes ?? 0),
+  }
+})

packages/core/test/event-compact.test.ts

@@ -0,0 +1,92 @@
+import { describe, expect } from "bun:test"
+import { Effect, Layer } from "effect"
+import { EventV2 } from "@opencode-ai/core/event"
+import { Database } from "@opencode-ai/core/database/database"
+import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql"
+import { Location } from "@opencode-ai/core/location"
+import { AbsolutePath } from "@opencode-ai/core/schema"
+import { WorkspaceV2 } from "@opencode-ai/core/workspace"
+import { eq } from "drizzle-orm"
+import { location } from "./fixture/location"
+import { testEffect } from "./lib/effect"
+import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
+import { LayerNode } from "@opencode-ai/core/effect/layer-node"
+
+const locationLayer = Layer.succeed(
+  Location.Service,
+  Location.Service.of(
+    location({ directory: AbsolutePath.make("project"), workspaceID: WorkspaceV2.ID.make("wrk_test") }),
+  ),
+)
+
+const it = testEffect(
+  AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [[Location.node, locationLayer]]),
+)
+
+const insert = (db: Database.Interface["db"]) =>
+  (rows: { id: string; aggregateID: string; seq: number; type: string; data: Record<string, unknown> }[]) =>
+    db
+      .insert(EventTable)
+      .values(
+        rows.map((row) => ({
+          id: row.id,
+          aggregate_id: row.aggregateID,
+          seq: row.seq,
+          type: row.type,
+          data: row.data,
+        })) as never,
+      )
+      .run()
+      .pipe(Effect.orDie)
+
+describe("EventV2.compactSnapshotEvents", () => {
+  it.effect("keeps only the latest message.updated and part.updated per entity", () =>
+    Effect.gen(function* () {
+      const { db } = yield* Database.Service
+      yield* db.insert(EventSequenceTable).values([{ aggregate_id: "ses_a", seq: 10 }]).run().pipe(Effect.orDie)
+      yield* insert(db)([
+        { id: "e1", aggregateID: "ses_a", seq: 1, type: "message.updated.1", data: { info: { id: "msg_m1", text: "v1" } } },
+        { id: "e9", aggregateID: "ses_a", seq: 9, type: "session.created.1", data: { sessionID: "ses_a" } },
+        { id: "e10", aggregateID: "ses_a", seq: 10, type: "session.updated.1", data: { sessionID: "ses_a" } },
+        { id: "e2", aggregateID: "ses_a", seq: 2, type: "message.updated.1", data: { info: { id: "msg_m1", text: "v2" } } },
+        { id: "e3", aggregateID: "ses_a", seq: 3, type: "message.updated.1", data: { info: { id: "msg_m1", text: "v3" } } },
+        { id: "e4", aggregateID: "ses_a", seq: 4, type: "message.updated.1", data: { info: { id: "msg_m2", text: "x" } } },
+        { id: "e5", aggregateID: "ses_a", seq: 5, type: "message.part.updated.1", data: { part: { id: "prt_p1", text: "a" } } },
+        { id: "e6", aggregateID: "ses_a", seq: 6, type: "message.part.updated.1", data: { part: { id: "prt_p1", text: "ab" } } },
+        { id: "e7", aggregateID: "ses_a", seq: 7, type: "message.part.updated.1", data: { part: { id: "prt_p1", text: "abc" } } },
+        { id: "e8", aggregateID: "ses_a", seq: 8, type: "message.removed.1", data: { sessionID: "ses_a", messageID: "msg_m9" } },
+      ])
+
+      const result = yield* EventV2.compactSnapshotEvents(db)
+      expect(result.removed).toBe(4)
+
+      const rows = yield* db
+        .select()
+        .from(EventTable)
+        .where(eq(EventTable.aggregate_id, "ses_a"))
+        .all()
+        .pipe(Effect.orDie)
+      const updated = rows.filter((row) => row.type === "message.updated.1")
+      const parts = rows.filter((row) => row.type === "message.part.updated.1")
+      const removed = rows.filter((row) => row.type === "message.removed.1")
+      expect(updated).toHaveLength(2)
+      expect(parts).toHaveLength(1)
+      const texts = updated.map((row) => (row.data as { info?: { text?: string } }).info?.text)
+      expect(texts).toContain("x")
+      expect(texts).toContain("v3")
+      expect(removed).toHaveLength(1)
+    }),
+  )
+
+  it.effect("removes nothing when no duplicate snapshots exist", () =>
+    Effect.gen(function* () {
+      const { db } = yield* Database.Service
+      yield* db.insert(EventSequenceTable).values([{ aggregate_id: "ses_b", seq: 1 }]).run().pipe(Effect.orDie)
+      yield* insert(db)([
+        { id: "e1", aggregateID: "ses_b", seq: 1, type: "message.updated.1", data: { info: { id: "msg_m1", text: "only" } } },
+      ])
+      const result = yield* EventV2.compactSnapshotEvents(db)
+      expect(result.removed).toBe(0)
+    }),
+  )
+})

packages/core/test/event-persist-gate.test.ts

@@ -0,0 +1,105 @@
+import { describe, expect } from "bun:test"
+import { Effect, Layer } from "effect"
+import { EventV2 } from "@opencode-ai/core/event"
+import { SessionV1 } from "@opencode-ai/schema/session-v1"
+import { Database } from "@opencode-ai/core/database/database"
+import { Session } from "@opencode-ai/schema/session"
+import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql"
+import { Location } from "@opencode-ai/core/location"
+import { AbsolutePath } from "@opencode-ai/core/schema"
+import { WorkspaceV2 } from "@opencode-ai/core/workspace"
+import { eq } from "drizzle-orm"
+import { location } from "./fixture/location"
+import { testEffect } from "./lib/effect"
+import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
+import { LayerNode } from "@opencode-ai/core/effect/layer-node"
+
+const locationLayer = Layer.succeed(
+  Location.Service,
+  Location.Service.of(
+    location({ directory: AbsolutePath.make("project"), workspaceID: WorkspaceV2.ID.make("wrk_test") }),
+  ),
+)
+const it = testEffect(
+  AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [[Location.node, locationLayer]]),
+)
+
+const messageUpdated = (
+  sid: Session.ID,
+  mid: SessionV1.MessageID,
+): EventV2.Data<typeof SessionV1.Event.MessageUpdated> =>
+  ({
+    sessionID: sid,
+    info: {
+      role: "user",
+      sessionID: sid,
+      id: mid,
+      time: { created: 1 },
+      files: [],
+      agents: [],
+      text: "hello",
+      agent: "build",
+      model: { providerID: "openrouter", modelID: "test/model" },
+    },
+  }) as never
+
+describe("EventV2.publish persist gate", () => {
+  it.effect("persist:false skips the event log entirely", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const sid = Session.ID.create()
+      const mid = SessionV1.MessageID.ascending()
+
+      const notified = yield* events.publish(SessionV1.Event.MessageUpdated, messageUpdated(sid, mid), {
+        persist: false,
+      })
+
+      const eventRows = yield* db
+        .select()
+        .from(EventTable)
+        .where(eq(EventTable.aggregate_id, sid))
+        .all()
+        .pipe(Effect.orDie)
+      const seqRows = yield* db
+        .select()
+        .from(EventSequenceTable)
+        .where(eq(EventSequenceTable.aggregate_id, sid))
+        .all()
+        .pipe(Effect.orDie)
+
+      expect(eventRows).toHaveLength(0)
+      expect(seqRows).toHaveLength(0)
+      // Payload still delivered to the caller (and thus to PubSub/SSE).
+      expect(notified.type).toBe("message.updated")
+      expect(notified.durable).toBeUndefined()
+    }),
+  )
+
+  it.effect("persist:true (default) writes the event log and advances the sequence", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const sid = Session.ID.create()
+      const mid = SessionV1.MessageID.ascending()
+
+      yield* events.publish(SessionV1.Event.MessageUpdated, messageUpdated(sid, mid))
+
+      const eventRows = yield* db
+        .select()
+        .from(EventTable)
+        .where(eq(EventTable.aggregate_id, sid))
+        .all()
+        .pipe(Effect.orDie)
+      const seqRows = yield* db
+        .select()
+        .from(EventSequenceTable)
+        .where(eq(EventSequenceTable.aggregate_id, sid))
+        .all()
+        .pipe(Effect.orDie)
+      expect(eventRows).toHaveLength(1)
+      expect(seqRows).toHaveLength(1)
+      expect(seqRows[0]?.seq).toBe(0)
+    }),
+  )
+})

packages/opencode/src/cli/cancelled-error.ts

@@ -0,0 +1,3 @@
+import { Schema } from "effect"
+
+export class CancelledError extends Schema.TaggedErrorClass<CancelledError>()("UICancelledError", {}) {}