fix(core): gate durable event persistence

packages/core/src/event.ts

@@ -10,6 +10,7 @@ import { Location } from "./location"
 import { makeGlobalNode } from "./effect/app-node"
 import { isDeepStrictEqual } from "node:util"
 import { Durable } from "@opencode-ai/schema/durable-event-manifest"
+import { truthy } from "./flag/flag"
 
 export const ID = Event.ID
 export type ID = import("@opencode-ai/schema/event").ID
@@ -165,6 +166,7 @@ export const allBounded = (events: Interface, capacity: number) =>
 
 export interface LayerOptions {
   readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect<void>
+  readonly persistDurableEvents?: boolean
 }
 
 export const layerWith = (options?: LayerOptions) =>
@@ -180,6 +182,7 @@ export const layerWith = (options?: LayerOptions) =>
       // TODO: Bind durable projectors to exact type+version before supporting incompatible historical payloads.
       const listeners = new Array<Subscriber>()
       const { db } = yield* Database.Service
+      const persistDurableEvents = options?.persistDurableEvents ?? truthy("OPENCODE_EXPERIMENTAL_EVENT_PERSISTENCE")
 
       const getOrCreate = (definition: Definition) =>
         Effect.gen(function* () {
@@ -333,19 +336,20 @@ export const layerWith = (options?: LayerOptions) =>
                             })
                             .run()
                             .pipe(Effect.orDie)
-                          yield* db
-                            .insert(EventTable)
-                            .values([
-                              {
-                                id: event.id,
-                                aggregate_id: aggregateID,
-                                seq,
-                                type: versionedType(definition.type, durable.version),
-                                data: encoded,
-                              },
-                            ])
-                            .run()
-                            .pipe(Effect.orDie)
+                          if (persistDurableEvents)
+                            yield* db
+                              .insert(EventTable)
+                              .values([
+                                {
+                                  id: event.id,
+                                  aggregate_id: aggregateID,
+                                  seq,
+                                  type: versionedType(definition.type, durable.version),
+                                  data: encoded,
+                                },
+                              ])
+                              .run()
+                              .pipe(Effect.orDie)
                           return { aggregateID, seq }
                         }),
                       { behavior: "immediate" },

packages/core/test/event.test.ts

@@ -82,6 +82,9 @@ const it = testEffect(
   AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [[Location.node, locationLayer]]),
 )
 const itWithoutLocation = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node])))
+const itWithoutHistory = testEffect(
+  EventV2.layerWith({ persistDurableEvents: false }).pipe(Layer.provideMerge(AppNodeBuilder.build(Database.node))),
+)
 
 describe("EventV2", () => {
   it.effect("publishes events with the current location", () =>
@@ -400,6 +403,29 @@ describe("EventV2", () => {
     }),
   )
 
+  itWithoutHistory.effect("commits projections and sequences without retaining event history", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+      const projected = new Array<number>()
+      yield* events.project(SyncMessage, (event) =>
+        Effect.sync(() => {
+          projected.push(event.durable!.seq)
+        }),
+      )
+
+      yield* events.publish(SyncMessage, { id: aggregateID, text: "first" })
+      yield* events.publish(SyncMessage, { id: aggregateID, text: "second" })
+
+      expect(projected).toEqual([0, 1])
+      expect(yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all()).toEqual([])
+      expect(
+        yield* db.select().from(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).get(),
+      ).toMatchObject({ aggregate_id: aggregateID, seq: 1 })
+    }),
+  )
+
   it.effect("increments durable event seq per aggregate", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service

packages/core/test/preload.ts

@@ -3,3 +3,4 @@ import path from "path"
 process.env.OPENCODE_DB = ":memory:"
 process.env.OPENCODE_MODELS_PATH = path.join(import.meta.dir, "plugin", "fixtures", "models-dev.json")
 process.env.OPENCODE_DISABLE_MODELS_FETCH = "true"
+process.env.OPENCODE_EXPERIMENTAL_EVENT_PERSISTENCE = "true"

packages/opencode/test/preload.ts

@@ -37,6 +37,7 @@ process.env["XDG_CONFIG_HOME"] = path.join(dir, "config")
 process.env["XDG_STATE_HOME"] = path.join(dir, "state")
 process.env["OPENCODE_MODELS_PATH"] = path.join(import.meta.dir, "tool", "fixtures", "models-api.json")
 process.env["OPENCODE_EXPERIMENTAL_EVENT_SYSTEM"] = "true"
+process.env["OPENCODE_EXPERIMENTAL_EVENT_PERSISTENCE"] = "true"
 process.env["OPENCODE_EXPERIMENTAL_WORKSPACES"] = "true"
 
 // Set test home directory to isolate tests from user's actual home directory