fix: dedupe byte-identical durable event appends without consuming seq

Author: LeXwDeX

.specgit.yaml

@@ -1,8 +1,8 @@
 version: 1
-delivery: summary-diff-continue
+delivery: event-idempotency-gate
 context:
   kind: branch
-  branch: fix/525-summary-diff-continue
+  branch: fix/523-event-idempotency-gate
 issues:
-  - 525
-pr: 526
+  - 523
+pr: 527

packages/core/schema.json

@@ -1,9 +1,9 @@
 {
   "version": "7",
   "dialect": "sqlite",
-  "id": "abadf28b-1770-46c6-bbf6-b7800a9ca874",
+  "id": "7a2e2a70-584a-4604-bf73-4c6e116c20a3",
   "prevIds": [
-    "874d8e74-d354-4dcb-b98c-c893660c9371"
+    "abadf28b-1770-46c6-bbf6-b7800a9ca874"
   ],
   "ddl": [
     {
@@ -1052,6 +1052,16 @@
       "entityType": "columns",
       "table": "event"
     },
+    {
+      "type": "text",
+      "notNull": false,
+      "autoincrement": false,
+      "default": null,
+      "generated": null,
+      "name": "data_hash",
+      "entityType": "columns",
+      "table": "event"
+    },
     {
       "type": "text",
       "notNull": false,

packages/core/src/database/migration.gen.ts

@@ -0,0 +1,11 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260903062324_add_event_data_hash",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`ALTER TABLE \`event\` ADD \`data_hash\` text;`)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/migration/20260903062324_add_event_data_hash.ts

@@ -149,6 +149,7 @@ export default {
           \`seq\` integer NOT NULL,
           \`type\` text NOT NULL,
           \`data\` text NOT NULL,
+          \`data_hash\` text,
           CONSTRAINT \`fk_event_aggregate_id_event_sequence_aggregate_id_fk\` FOREIGN KEY (\`aggregate_id\`) REFERENCES \`event_sequence\`(\`aggregate_id\`) ON DELETE CASCADE
         );
       `)

packages/core/src/database/schema.gen.ts

@@ -3,13 +3,14 @@ export * as EventV2 from "./event"
 import { Cause, Context, Effect, FiberSet, Layer, Option, PubSub, 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 } from "drizzle-orm"
+import { and, asc, desc, eq, gt } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
 import { Location } from "./location"
 import { LayerNode } from "./effect/layer-node"
 import { isDeepStrictEqual } from "node:util"
 import { Durable } from "@opencode-ai/schema/durable-event-manifest"
+import { Hash } from "./util/hash"
 
 export const ID = Event.ID
 export type ID = import("@opencode-ai/schema/event").ID
@@ -227,6 +228,30 @@ export const layerWith = (options?: LayerOptions) =>
           if (input && row?.ownerID && row.ownerID !== input.ownerID) {
             return undefined
           }
+          const dataHash = Hash.sha256(JSON.stringify(encoded))
+          if (!input) {
+            // Idempotency gate (#523): a fresh append that byte-for-byte repeats
+            // this aggregate's latest same-type event carries zero information
+            // delta. Skip it entirely — no seq consumed, no projectors, no commit
+            // hook, no durable wake — so the persisted sequence stays dense and
+            // both the replayAll contiguity check and gt(seq, after) readers are
+            // unaffected. Replay appends (input) keep their exact-seq contract,
+            // and legacy rows carry a NULL hash so they never match.
+            const previous = yield* db
+              .select({ dataHash: EventTable.data_hash })
+              .from(EventTable)
+              .where(
+                and(
+                  eq(EventTable.aggregate_id, aggregateID),
+                  eq(EventTable.type, versionedType(definition.type, durable.version)),
+                ),
+              )
+              .orderBy(desc(EventTable.seq))
+              .limit(1)
+              .get()
+              .pipe(Effect.orDie)
+            if (previous && previous.dataHash === dataHash) return undefined
+          }
           const seq = input?.seq ?? latest + 1
           if (input && seq !== latest + 1) {
             yield* Effect.die(
@@ -278,6 +303,7 @@ export const layerWith = (options?: LayerOptions) =>
                 seq,
                 type: versionedType(definition.type, durable.version),
                 data: encoded,
+                data_hash: dataHash,
               },
             ])
             .run()
@@ -479,7 +505,9 @@ export const layerWith = (options?: LayerOptions) =>
                 .transaction(
                   () =>
                     Effect.gen(function* () {
-                      const results = new Array<{ aggregateID: string; seq: number }>()
+                      // Aligned with entries by index: a deduped entry yields
+                      // undefined so the payload pairing below stays positional.
+                      const results = new Array<{ aggregateID: string; seq: number } | undefined>()
                       for (const entry of entries) {
                         // No replay input: seq is allocated contiguously from the latest sequence inside the transaction.
                         const result = yield* commitDurableEventInner(
@@ -488,7 +516,7 @@ export const layerWith = (options?: LayerOptions) =>
                           undefined,
                           entry.commit,
                         )
-                        if (result) results.push(result)
+                        results.push(result)
                       }
                       return results
                     }),
@@ -499,19 +527,19 @@ export const layerWith = (options?: LayerOptions) =>
               return results
             }),
           )
-          const payloads = entries.flatMap((entry, index) => {
+          const payloads = entries.map((entry, index) => {
             const result = committed[index]
-            if (!result) return []
-            return [
-              {
-                ...entry.event,
-                durable: {
-                  aggregateID: result.aggregateID,
-                  seq: result.seq,
-                  version: entry.durable.version,
-                },
-              } as Payload,
-            ]
+            // A deduped entry is still notified (mirrors the single-publish
+            // path) but stays unstamped: it occupies no sequence position.
+            if (!result) return entry.event
+            return {
+              ...entry.event,
+              durable: {
+                aggregateID: result.aggregateID,
+                seq: result.seq,
+                version: entry.durable.version,
+              },
+            } as Payload
           })
           for (const payload of payloads) {
             yield* notify(payload)

packages/core/src/event.ts

@@ -17,6 +17,11 @@ export const EventTable = sqliteTable(
     seq: integer().notNull(),
     type: text().notNull(),
     data: text({ mode: "json" }).$type<Record<string, unknown>>().notNull(),
+    // sha256 of the serialized payload, written once at append time. The
+    // idempotency gate compares against the latest same-type row via
+    // event_aggregate_type_seq_idx instead of re-hashing MiB-scale payloads.
+    // Nullable: legacy rows predate the column and never match the gate.
+    data_hash: text(),
   },
   (table) => [
     uniqueIndex("event_aggregate_seq_idx").on(table.aggregate_id, table.seq),