feat(plugin): add v2 effect transform api

2026-06-20 22:27:11 -04:00
46 changed files with 806 additions and 654 deletions
@@ -99,8 +99,7 @@ jobs:
      - name: Setup Node
        uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
        with:
-          # Playwright 1.59 hangs while extracting Chromium with Node 24.16.
-          node-version: "24.15"
+          node-version: "24"

      - name: Setup Bun
        uses: ./.github/actions/setup-bun
@@ -1,8 +1,8 @@
 {
  "nodeModules": {
-    "x86_64-linux": "sha256-oWSGu+SP66Aquy/0Vaq7Bgp8404ZdOWbQX+O7h3jxHU=",
-    "aarch64-linux": "sha256-UsS0+c+GwtIukmWwQeFbY/3Oaz3t4Q7C6cFMGkmlyAY=",
-    "aarch64-darwin": "sha256-CArz92ewPmXO+ORFCBkCH8LzMpU/DjyaO4ic7QL0UpI=",
-    "x86_64-darwin": "sha256-rhnz9gmG6L06wIzfMhTaXDDEf6IbMD32CavqwXoqcUs="
+    "x86_64-linux": "sha256-g0tDvRf7MErZ1PEeUazEYi492ZHiRT8kYv3bPdkss/I=",
+    "aarch64-linux": "sha256-6sKgf3ftbIqlPxlFkoPzoWPsJp3IwXD+H3Y6g874xmk=",
+    "aarch64-darwin": "sha256-Se/Nls/KlkuK2ysDQ9DeAzSaX3NsL2iDdf/dsv2GIXc=",
+    "x86_64-darwin": "sha256-V9MCkqnvQ1nkD2PaaTfNFKkBZGymj6KxrSAK6+DTF8Y="
  }
 }
@@ -5,7 +5,7 @@ import { and, asc, eq, gt } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
 import { Location } from "./location"
-import { externalID, type ExternalID, withStatics } from "./schema"
+import { externalID, type ExternalID, NonNegativeInt, withStatics } from "./schema"
 import { Identifier } from "./util/identifier"
 import { LayerNode } from "./effect/layer-node"
 import { isDeepStrictEqual } from "node:util"
@@ -19,9 +19,16 @@ export const ID = Schema.String.check(Schema.isStartsWith("evt_")).pipe(
 )
 export type ID = typeof ID.Type

+/**
+ * Durable aggregate continuation position for embedded replay streams.
+ * TODO: Decide whether a future HTTP / SDK surface should expose an opaque cursor instead.
+ */
+export const Cursor = NonNegativeInt.pipe(Schema.brand("EventV2.Cursor"))
+export type Cursor = typeof Cursor.Type
+
 export type Definition<Type extends string = string, DataSchema extends Schema.Top = Schema.Top> = {
  readonly type: Type
-  readonly durable?: {
+  readonly sync?: {
    readonly version: number
    readonly aggregate: string
  }
@@ -34,16 +41,20 @@ export type Payload<D extends Definition = Definition> = {
  readonly id: ID
  readonly type: D["type"]
  readonly data: Data<D>
-  readonly durable?: {
-    readonly aggregateID: string
-    readonly seq: number
-    readonly version: number
-  }
+  /** Durable aggregate order, populated while synchronized events are projected. */
+  readonly seq?: number
+  readonly version?: number
  readonly location?: Location.Ref
  readonly metadata?: Record<string, unknown>
+  /** Internal replay marker for projectors that own non-replicated operational state. */
+  readonly replay?: boolean
 }

-export type Subscriber<D extends Definition = Definition> = (event: Payload<D>) => Effect.Effect<void>
+export type Projector<D extends Definition = Definition> = (event: Payload<D>) => Effect.Effect<void>
+type AnyProjector = (event: Payload) => Effect.Effect<void>
+export type CommitGuard = (event: Payload) => Effect.Effect<void>
+export type Listener = (event: Payload) => Effect.Effect<void>
+export type Sync = (event: Payload) => Effect.Effect<void>
 export type Unsubscribe = Effect.Effect<void>

 export type SerializedEvent = {
@@ -54,8 +65,13 @@ export type SerializedEvent = {
  readonly data: Record<string, unknown>
 }

-export class InvalidDurableEventError extends Schema.TaggedErrorClass<InvalidDurableEventError>()(
-  "EventV2.InvalidDurableEvent",
+export type CursorEvent<E extends Payload = Payload> = {
+  readonly cursor: Cursor
+  readonly event: E
+}
+
+export class InvalidSyncEventError extends Schema.TaggedErrorClass<InvalidSyncEventError>()(
+  "EventV2.InvalidSyncEvent",
  {
    type: Schema.String,
    message: Schema.String,
@@ -67,11 +83,19 @@ export function versionedType(type: string, version: number) {
 }

 export const registry = new Map<string, Definition>()
-const durableRegistry = new Map<string, Definition>()
+type SyncDefinition = Definition & {
+  readonly sync: NonNullable<Definition["sync"]>
+  readonly encode: (data: unknown) => unknown
+  readonly decode: (data: unknown) => unknown
+}
+const syncRegistry = new Map<string, SyncDefinition>()
+
+// Synchronized events cross a JSON boundary, so their data schemas must encode and decode without services.
+const syncCodec = (definition: Definition) => definition.data as Schema.Codec<unknown, unknown, never, never>

 export function define<const Type extends string, Fields extends Schema.Struct.Fields>(input: {
  readonly type: Type
-  readonly durable?: {
+  readonly sync?: {
    readonly version: number
    readonly aggregate: string
  }
@@ -82,25 +106,28 @@ export function define<const Type extends string, Fields extends Schema.Struct.F
    id: ID,
    metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
    type: Schema.Literal(input.type),
-    durable: Schema.optional(Schema.Struct({ aggregateID: Schema.String, seq: Schema.Number, version: Schema.Number })),
+    version: Schema.optional(Schema.Number),
    location: Schema.optional(Location.Ref),
    data: Data,
  }).annotate({ identifier: input.type })

  const definition = Object.assign(Payload, {
    type: input.type,
-    ...(input.durable === undefined ? {} : { durable: input.durable }),
+    ...(input.sync === undefined ? {} : { sync: input.sync }),
    data: Data,
  })
  const existing = registry.get(input.type)
-  if (
-    input.durable === undefined ||
-    existing?.durable === undefined ||
-    input.durable.version >= existing.durable.version
-  ) {
+  if (input.sync === undefined || existing?.sync === undefined || input.sync.version >= existing.sync.version) {
    registry.set(input.type, definition)
  }
-  if (input.durable) durableRegistry.set(versionedType(input.type, input.durable.version), definition)
+  if (input.sync)
+    syncRegistry.set(
+      versionedType(input.type, input.sync.version),
+      Object.assign(definition, {
+        encode: Schema.encodeUnknownSync(syncCodec(definition)),
+        decode: Schema.decodeUnknownSync(syncCodec(definition)),
+      }) as SyncDefinition,
+    )
  return definition as Schema.Schema<Payload<Definition<Type, Schema.Struct<Fields>>>> &
    Definition<Type, Schema.Struct<Fields>>
 }
@@ -113,7 +140,7 @@ export interface PublishOptions {
  readonly id?: ID
  readonly metadata?: Record<string, unknown>
  readonly location?: Location.Ref
-  /** Local operational projection committed atomically with a new durable event. Not replayed or serialized. */
+  /** Local operational projection committed atomically with a new synchronized event. Not replayed or serialized. */
  readonly commit?: (seq: number) => Effect.Effect<void>
 }

@@ -125,10 +152,14 @@ export interface Interface {
  ) => Effect.Effect<Payload<D>>
  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>
-  /** @deprecated Use `all()` and consume the returned stream. */
-  readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe>
-  readonly project: <D extends Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
+  readonly aggregateEvents: (input: {
+    readonly aggregateID: string
+    readonly after?: Cursor
+  }) => Stream.Stream<CursorEvent>
+  readonly sync: (handler: Sync) => Effect.Effect<Unsubscribe>
+  readonly listen: (listener: Listener) => Effect.Effect<Unsubscribe>
+  readonly beforeCommit: (guard: CommitGuard) => Effect.Effect<void>
+  readonly project: <D extends Definition>(definition: D, projector: Projector<D>) => Effect.Effect<void>
  readonly replay: (
    event: SerializedEvent,
    options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean },
@@ -151,37 +182,37 @@ export const layerWith = (options?: LayerOptions) =>
  Layer.effect(
    Service,
    Effect.gen(function* () {
-      const pubsub = {
-        all: yield* PubSub.unbounded<Payload>(),
-        durable: new Map<string, Set<PubSub.PubSub<void>>>(),
-        typed: new Map<string, PubSub.PubSub<Payload>>(),
-      }
-      const projectors = new Map<string, Subscriber[]>()
-      const listeners = new Array<Subscriber>()
+      const all = yield* PubSub.unbounded<Payload>()
+      const synchronized = new Map<string, Set<PubSub.PubSub<void>>>()
+      const typed = new Map<string, PubSub.PubSub<Payload>>()
+      const projectors = new Map<string, AnyProjector[]>()
+      const commitGuards = new Array<CommitGuard>()
+      const listeners = new Array<Listener>()
+      const syncHandlers = new Array<Sync>()
      const { db } = yield* Database.Service

      const getOrCreate = (definition: Definition) =>
        Effect.gen(function* () {
-          const existing = pubsub.typed.get(definition.type)
+          const existing = typed.get(definition.type)
          if (existing) return existing
-          const created = yield* PubSub.unbounded<Payload>()
-          pubsub.typed.set(definition.type, created)
-          return created
+          const pubsub = yield* PubSub.unbounded<Payload>()
+          typed.set(definition.type, pubsub)
+          return pubsub
        })

      yield* Effect.addFinalizer(() =>
        Effect.gen(function* () {
-          yield* PubSub.shutdown(pubsub.all)
+          yield* PubSub.shutdown(all)
          yield* Effect.forEach(
-            pubsub.durable.values(),
+            synchronized.values(),
            (pubsubs) => Effect.forEach(pubsubs, PubSub.shutdown, { discard: true }),
            { discard: true },
          )
-          yield* Effect.forEach(pubsub.typed.values(), PubSub.shutdown, { discard: true })
+          yield* Effect.forEach(typed.values(), PubSub.shutdown, { discard: true })
        }),
      )

-      function commitDurableEvent(
+      function commitSyncEvent(
        event: Payload,
        input?: {
          readonly seq: number
@@ -193,20 +224,28 @@ export const layerWith = (options?: LayerOptions) =>
      ) {
        return Effect.gen(function* () {
          const definition = registry.get(event.type)
-          const durable = definition?.durable
-          if (durable) {
-            const aggregateID = (event.data as Record<string, unknown>)[durable.aggregate]
+          const sync = definition?.sync
+          if (sync) {
+            if (event.version !== sync.version) {
+              yield* Effect.die(
+                new InvalidSyncEventError({
+                  type: event.type,
+                  message: `Expected event version ${sync.version}, got ${event.version}`,
+                }),
+              )
+            }
+            const aggregateID = (event.data as Record<string, unknown>)[sync.aggregate]
            if (typeof aggregateID !== "string") {
              yield* Effect.die(
-                new InvalidDurableEventError({
+                new InvalidSyncEventError({
                  type: event.type,
-                  message: `Expected string aggregate field ${durable.aggregate}`,
+                  message: `Expected string aggregate field ${sync.aggregate}`,
                }),
              )
            } else {
              if (input && input.aggregateID !== aggregateID) {
                yield* Effect.die(
-                  new InvalidDurableEventError({
+                  new InvalidSyncEventError({
                    type: event.type,
                    message: `Aggregate mismatch: expected ${input.aggregateID}, got ${aggregateID}`,
                  }),
@@ -226,12 +265,12 @@ export const layerWith = (options?: LayerOptions) =>
                            .get()
                            .pipe(Effect.orDie)
                          const latest = row?.seq ?? -1
-                          const encoded = Schema.encodeUnknownSync(
-                            definition.data as Schema.Codec<unknown, unknown, never, never>,
-                          )(event.data) as Record<string, unknown>
+                          const encoded = syncRegistry
+                            .get(versionedType(definition.type, sync.version))!
+                            .encode(event.data) as Record<string, unknown>
                          if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) {
                            yield* Effect.die(
-                              new InvalidDurableEventError({
+                              new InvalidSyncEventError({
                                type: event.type,
                                message: `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`,
                              }),
@@ -246,7 +285,7 @@ export const layerWith = (options?: LayerOptions) =>
                              .pipe(Effect.orDie)
                            if (
                              stored?.id === event.id &&
-                              stored.type === versionedType(definition.type, durable.version) &&
+                              stored.type === versionedType(definition.type, sync.version) &&
                              isDeepStrictEqual(stored.data, encoded)
                            ) {
                              if (input.ownerID && row?.ownerID == null) {
@@ -260,7 +299,7 @@ export const layerWith = (options?: LayerOptions) =>
                              return
                            }
                            yield* Effect.die(
-                              new InvalidDurableEventError({
+                              new InvalidSyncEventError({
                                type: event.type,
                                message: `Replay diverged at aggregate ${aggregateID} sequence ${input.seq}`,
                              }),
@@ -272,7 +311,7 @@ export const layerWith = (options?: LayerOptions) =>
                          const seq = input?.seq ?? latest + 1
                          if (input && seq !== latest + 1) {
                            yield* Effect.die(
-                              new InvalidDurableEventError({
+                              new InvalidSyncEventError({
                                type: event.type,
                                message: `Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}`,
                              }),
@@ -286,17 +325,16 @@ export const layerWith = (options?: LayerOptions) =>
                            .pipe(Effect.orDie)
                          if (stored)
                            yield* Effect.die(
-                              new InvalidDurableEventError({
+                              new InvalidSyncEventError({
                                type: event.type,
                                message: `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`,
                              }),
                            )
-                          const committed = {
-                            ...event,
-                            durable: { aggregateID, seq, version: durable.version },
-                          } as Payload
+                          for (const guard of commitGuards) {
+                            yield* guard(event)
+                          }
                          for (const projector of list) {
-                            yield* projector(committed)
+                            yield* projector({ ...event, seq } as Payload)
                          }
                          if (commit) yield* commit(seq)
                          yield* db
@@ -318,7 +356,7 @@ export const layerWith = (options?: LayerOptions) =>
                                id: event.id,
                                aggregate_id: aggregateID,
                                seq,
-                                type: versionedType(definition.type, durable.version),
+                                type: versionedType(definition.type, sync.version),
                                data: encoded,
                              },
                            ])
@@ -331,8 +369,8 @@ export const layerWith = (options?: LayerOptions) =>
                    .pipe(Effect.orDie)
                  if (committed) {
                    yield* Effect.forEach(
-                      pubsub.durable.get(committed.aggregateID) ?? [],
-                      (wake) => PubSub.publish(wake, undefined),
+                      synchronized.get(committed.aggregateID) ?? [],
+                      (pubsub) => PubSub.publish(pubsub, undefined),
                      { discard: true },
                    )
                  }
@@ -346,25 +384,19 @@ export const layerWith = (options?: LayerOptions) =>

      function publishEvent<D extends Definition>(event: Payload<D>, commit?: PublishOptions["commit"]) {
        return Effect.gen(function* () {
-          const definition = registry.get(event.type)
-          if (!definition?.durable && commit)
+          const durable = registry.get(event.type)?.sync !== undefined
+          if (!durable && commit)
            return yield* Effect.die(
-              new InvalidDurableEventError({
+              new InvalidSyncEventError({
                type: event.type,
-                message: "Local commit hooks require a durable event",
+                message: "Local commit hooks require a synchronized event",
              }),
            )
-          if (definition?.durable) {
-            const committed = yield* commitDurableEvent(event as Payload, undefined, commit)
+          if (durable) {
+            const committed = yield* commitSyncEvent(event as Payload, undefined, commit)
            if (committed) {
-              event = {
-                ...event,
-                durable: {
-                  aggregateID: committed.aggregateID,
-                  seq: committed.seq,
-                  version: definition.durable.version,
-                },
-              }
+              event = { ...event, seq: committed.seq }
+              yield* Effect.forEach(syncHandlers, (sync) => observe(event as Payload, "sync", sync), { discard: true })
              yield* notify(event as Payload, true)
              return event
            }
@@ -374,11 +406,12 @@ export const layerWith = (options?: LayerOptions) =>
        })
      }

-      const observe = (event: Payload, observer: (event: Payload) => Effect.Effect<void>) =>
+      const observe = (event: Payload, kind: "sync" | "listener", observer: (event: Payload) => Effect.Effect<void>) =>
        Effect.suspend(() => observer(event)).pipe(
          Effect.catchCauseIf(
            (cause) => !Cause.hasInterrupts(cause),
-            (cause) => Effect.logError("Event listener failed", { eventID: event.id, eventType: event.type, cause }),
+            (cause) =>
+              Effect.logError("Event observer failed", { eventID: event.id, eventType: event.type, kind, cause }),
          ),
        )

@@ -386,12 +419,12 @@ export const layerWith = (options?: LayerOptions) =>
        return Effect.gen(function* () {
          yield* Effect.forEach(
            listeners,
-            (listener) => (isolateListeners ? observe(event, listener) : listener(event)),
+            (listener) => (isolateListeners ? observe(event, "listener", listener) : listener(event)),
            { discard: true },
          )
-          const typed = pubsub.typed.get(event.type)
-          if (typed) yield* PubSub.publish(typed, event)
-          yield* PubSub.publish(pubsub.all, event)
+          const pubsub = typed.get(event.type)
+          if (pubsub) yield* PubSub.publish(pubsub, event)
+          yield* PubSub.publish(all, event)
        })
      }

@@ -408,6 +441,7 @@ export const layerWith = (options?: LayerOptions) =>
              id: options?.id ?? ID.create(),
              ...(options?.metadata ? { metadata: options.metadata } : {}),
              type: definition.type,
+              ...(definition.sync === undefined ? {} : { version: definition.sync.version }),
              ...(location ? { location } : {}),
              data,
            } as Payload<D>,
@@ -421,37 +455,27 @@ export const layerWith = (options?: LayerOptions) =>
        options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean },
      ) {
        return Effect.gen(function* () {
-          const definition = durableRegistry.get(event.type)
-          if (!definition?.durable) {
+          const definition = syncRegistry.get(event.type)
+          if (!definition) {
            yield* Effect.die(
-              new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` }),
+              new InvalidSyncEventError({ type: event.type, message: `Unknown sync event type ${event.type}` }),
            )
          } else {
            const payload = {
              id: event.id,
              type: definition.type,
-              data: Schema.decodeUnknownSync(definition.data as Schema.Codec<unknown, unknown, never, never>)(
-                event.data,
-              ),
+              version: definition.sync.version,
+              data: definition.decode(event.data),
+              replay: true,
            } as Payload
-            const committed = yield* commitDurableEvent(payload, {
+            const committed = yield* commitSyncEvent(payload, {
              seq: event.seq,
              aggregateID: event.aggregateID,
              ownerID: options?.ownerID,
              strictOwner: options?.strictOwner,
            })
            if (committed && options?.publish) {
-              yield* notify(
-                {
-                  ...payload,
-                  durable: {
-                    aggregateID: committed.aggregateID,
-                    seq: committed.seq,
-                    version: definition.durable.version,
-                  },
-                },
-                true,
-              )
+              yield* notify({ ...payload, seq: committed.seq }, true)
            }
          }
        })
@@ -466,7 +490,7 @@ export const layerWith = (options?: LayerOptions) =>
          if (!source) return undefined
          if (events.some((event) => event.aggregateID !== source)) {
            yield* Effect.die(
-              new InvalidDurableEventError({
+              new InvalidSyncEventError({
                type: events[0]?.type ?? "unknown",
                message: "Replay events must belong to the same aggregate",
              }),
@@ -477,7 +501,7 @@ export const layerWith = (options?: LayerOptions) =>
            const seq = start + index
            if (event.seq !== seq) {
              yield* Effect.die(
-                new InvalidDurableEventError({
+                new InvalidSyncEventError({
                  type: event.type,
                  message: `Replay sequence mismatch at index ${index}: expected ${seq}, got ${event.seq}`,
                }),
@@ -516,18 +540,22 @@ export const layerWith = (options?: LayerOptions) =>
          Stream.map((event) => event as Payload<D>),
        )

-      const streamAll = (): Stream.Stream<Payload> => Stream.fromPubSub(pubsub.all)
+      const streamAll = (): Stream.Stream<Payload> => Stream.fromPubSub(all)

-      const decodeSerializedEvent = (event: SerializedEvent): Payload => {
-        const definition = durableRegistry.get(event.type)
-        if (!definition?.durable) {
-          throw new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` })
+      const decodeSerializedEvent = (event: SerializedEvent): CursorEvent => {
+        const definition = syncRegistry.get(event.type)
+        if (!definition) {
+          throw new InvalidSyncEventError({ type: event.type, message: `Unknown sync event type ${event.type}` })
        }
        return {
-          id: event.id,
-          type: definition.type,
-          durable: { aggregateID: event.aggregateID, seq: event.seq, version: definition.durable.version },
-          data: Schema.decodeUnknownSync(definition.data as Schema.Codec<unknown, unknown, never, never>)(event.data),
+          cursor: Cursor.make(event.seq),
+          event: {
+            id: event.id,
+            type: definition.type,
+            version: definition.sync.version,
+            seq: event.seq,
+            data: definition.decode(event.data),
+          },
        }
      }

@@ -555,40 +583,43 @@ export const layerWith = (options?: LayerOptions) =>
          ),
        )

-      const subscribeDurable = (aggregateID: string) =>
+      const subscribeSynchronized = (aggregateID: string) =>
        Effect.gen(function* () {
-          const wake = yield* PubSub.sliding<void>(1)
-          const subscription = yield* PubSub.subscribe(wake)
+          const pubsub = yield* PubSub.sliding<void>(1)
+          const subscription = yield* PubSub.subscribe(pubsub)
          yield* Effect.acquireRelease(
            Effect.sync(() => {
-              const wakes = pubsub.durable.get(aggregateID) ?? new Set()
-              wakes.add(wake)
-              pubsub.durable.set(aggregateID, wakes)
+              const pubsubs = synchronized.get(aggregateID) ?? new Set()
+              pubsubs.add(pubsub)
+              synchronized.set(aggregateID, pubsubs)
            }),
            () =>
              Effect.sync(() => {
-                const wakes = pubsub.durable.get(aggregateID)
-                wakes?.delete(wake)
-                if (wakes?.size === 0) pubsub.durable.delete(aggregateID)
-              }).pipe(Effect.andThen(PubSub.shutdown(wake))),
+                const pubsubs = synchronized.get(aggregateID)
+                pubsubs?.delete(pubsub)
+                if (pubsubs?.size === 0) synchronized.delete(aggregateID)
+              }).pipe(Effect.andThen(PubSub.shutdown(pubsub))),
          )
          return subscription
        })

-      const durable = (input: { readonly aggregateID: string; readonly after?: number }): Stream.Stream<Payload> =>
+      const streamEvents = (input: {
+        readonly aggregateID: string
+        readonly after?: Cursor
+      }): Stream.Stream<CursorEvent> =>
        Stream.unwrap(
          Effect.gen(function* () {
-            const wakes = yield* subscribeDurable(input.aggregateID)
-            let sequence = input.after ?? -1
-            const read = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe(
+            const synchronized = yield* subscribeSynchronized(input.aggregateID)
+            let cursor = input.after ?? -1
+            const read = Effect.suspend(() => readAfter(input.aggregateID, cursor)).pipe(
              Effect.tap((events) =>
                Effect.sync(() => {
-                  sequence = events.at(-1)?.durable?.seq ?? sequence
+                  cursor = events.at(-1)?.cursor ?? cursor
                }),
              ),
            )
            const historical = yield* read
-            const live = Stream.fromSubscription(wakes).pipe(
+            const live = Stream.fromSubscription(synchronized).pipe(
              Stream.mapEffect(() => read),
              Stream.flattenIterable,
            )
@@ -596,7 +627,7 @@ export const layerWith = (options?: LayerOptions) =>
          }),
        )

-      const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> =>
+      const listen = (listener: Listener): Effect.Effect<Unsubscribe> =>
        Effect.sync(() => {
          listeners.push(listener)
          return Effect.sync(() => {
@@ -605,7 +636,21 @@ export const layerWith = (options?: LayerOptions) =>
          })
        })

-      const project = <D extends Definition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> =>
+      const sync = (handler: Sync): Effect.Effect<Unsubscribe> =>
+        Effect.sync(() => {
+          syncHandlers.push(handler)
+          return Effect.sync(() => {
+            const index = syncHandlers.indexOf(handler)
+            if (index >= 0) syncHandlers.splice(index, 1)
+          })
+        })
+
+      const beforeCommit = (guard: CommitGuard): Effect.Effect<void> =>
+        Effect.sync(() => {
+          commitGuards.push(guard)
+        })
+
+      const project = <D extends Definition>(definition: D, projector: Projector<D>): Effect.Effect<void> =>
        Effect.sync(() => {
          const list = projectors.get(definition.type) ?? []
          list.push((event) => projector(event as Payload<D>))
@@ -616,8 +661,10 @@ export const layerWith = (options?: LayerOptions) =>
        publish,
        subscribe,
        all: streamAll,
-        durable,
+        aggregateEvents: streamEvents,
+        sync,
        listen,
+        beforeCommit,
        project,
        replay,
        replayAll,
@@ -7,8 +7,8 @@ import { FSUtil } from "./fs-util"
 import { Location } from "./location"
 import { PositiveInt, RelativePath } from "./schema"
 import { FileSystemSearch } from "./filesystem/search"
-import { Entry, Match, PathError } from "./filesystem/schema"
-export { Entry, Match, PathError, Submatch } from "./filesystem/schema"
+import { Entry, Match } from "./filesystem/schema"
+export { Entry, Match, Submatch } from "./filesystem/schema"

 export const ReadInput = Schema.Struct({
  path: RelativePath,
@@ -58,10 +58,8 @@ export const Event = {
 }

 export interface Interface {
-  readonly read: (
-    input: ReadInput,
-  ) => Effect.Effect<{ readonly content: Uint8Array; readonly mime: string }, PathError | FSUtil.Error>
-  readonly list: (input?: ListInput) => Effect.Effect<Entry[], PathError | FSUtil.Error>
+  readonly read: (input: ReadInput) => Effect.Effect<{ readonly content: Uint8Array; readonly mime: string }>
+  readonly list: (input?: ListInput) => Effect.Effect<Entry[]>
  readonly find: (input: FindInput) => Effect.Effect<Entry[]>
  readonly glob: (input: GlobInput) => Effect.Effect<readonly Entry[]>
  readonly grep: (input: GrepInput) => Effect.Effect<readonly Match[]>
@@ -79,9 +77,9 @@ const baseLayer = Layer.effect(
    const resolve = Effect.fnUntraced(function* (input?: RelativePath) {
      const absolute = path.resolve(location.directory, input ?? ".")
      if (!FSUtil.contains(location.directory, absolute))
-        return yield* new PathError({ path: input ?? ".", reason: "lexical_escape" })
-      const real = yield* fs.realPath(absolute)
-      if (!FSUtil.contains(root, real)) return yield* new PathError({ path: input ?? ".", reason: "symlink_escape" })
+        return yield* Effect.die(new Error("Path escapes the location"))
+      const real = yield* fs.realPath(absolute).pipe(Effect.orDie)
+      if (!FSUtil.contains(root, real)) return yield* Effect.die(new Error("Path escapes the location"))
      return { absolute, real, directory: location.directory, root }
    })
    return Service.of({
@@ -90,18 +88,19 @@ const baseLayer = Layer.effect(
      grep: search.grep,
      read: Effect.fn("FileSystem.read")(function* (input) {
        const target = yield* resolve(input.path)
-        const info = yield* fs.stat(target.real)
-        if (info.type !== "File") return yield* new PathError({ path: input.path, reason: "not_file" })
+        const info = yield* fs.stat(target.real).pipe(Effect.orDie)
+        if (info.type !== "File") return yield* Effect.die(new Error("Path is not a file"))
        return {
-          content: yield* fs.readFile(target.real),
+          content: yield* fs.readFile(target.real).pipe(Effect.orDie),
          mime: FSUtil.mimeType(target.real),
        }
      }),
      list: Effect.fn("FileSystem.list")(function* (input = {}) {
        const target = yield* resolve(input.path)
-        const info = yield* fs.stat(target.real)
-        if (info.type !== "Directory") return yield* new PathError({ path: input.path ?? ".", reason: "not_directory" })
+        const info = yield* fs.stat(target.real).pipe(Effect.orDie)
+        if (info.type !== "Directory") return yield* Effect.die(new Error("Path is not a directory"))
        return yield* fs.readDirectoryEntries(target.real).pipe(
+          Effect.orDie,
          Effect.map((items) =>
            items
              .flatMap((item) => {
@@ -21,8 +21,3 @@ export class Match extends Schema.Class<Match>("FileSystem.Match")({
  text: Schema.String,
  submatches: Schema.Array(Submatch),
 }) {}
-
-export class PathError extends Schema.TaggedErrorClass<PathError>()("FileSystem.PathError", {
-  path: Schema.String,
-  reason: Schema.Literals(["lexical_escape", "symlink_escape", "not_file", "not_directory"]),
-}) {}
@@ -38,7 +38,8 @@ export const Native = Schema.Struct({
 export const Api = Schema.Union([AISDK, Native]).pipe(Schema.toTaggedUnion("type"))
 export type Api = typeof Api.Type
 export type MutableApi<T extends Api = Api> = T extends Api
-  ? Omit<Types.DeepMutable<T>, "settings"> & (undefined extends T["settings"] ? { settings?: any } : { settings: any })
+  ? Omit<Types.DeepMutable<T>, "settings"> &
+      (undefined extends T["settings"] ? { settings?: any } : { settings: any })
  : never

 export const Request = Schema.Struct({
@@ -1,6 +1,7 @@
 export * as Session from "./session"

 import { Effect, Schema, Stream } from "effect"
+import { EventV2 } from "../event"
 import { ModelV2 } from "../model"
 import { SessionV2 } from "../session"
 import { MessageDecodeError } from "../session/error"
@@ -33,7 +34,9 @@ export type Delivery = SessionInput.Delivery
 export const ListInput = SessionV2.ListInput
 export type ListInput = SessionV2.ListInput

-export type Event = SessionEvent.DurableEvent
+export const EventCursor = EventV2.Cursor
+export type EventCursor = EventV2.Cursor
+export type Event = EventV2.CursorEvent<SessionEvent.DurableEvent>

 export const NotFoundError = SessionV2.NotFoundError
 export type NotFoundError = SessionV2.NotFoundError
@@ -96,7 +99,7 @@ export interface MessageInput {

 export interface EventsInput {
  readonly sessionID: ID
-  readonly after?: number
+  readonly after?: EventCursor
 }

 export interface Interface {
@@ -124,8 +124,8 @@ export interface Interface {
  ) => Effect.Effect<SessionMessage.Message[], NotFoundError | MessageDecodeError>
  readonly events: (input: {
    sessionID: SessionSchema.ID
-    after?: number
-  }) => Stream.Stream<SessionEvent.DurableEvent, NotFoundError>
+    after?: EventV2.Cursor
+  }) => Stream.Stream<EventV2.CursorEvent<SessionEvent.DurableEvent>, NotFoundError>
  readonly switchAgent: (input: {
    sessionID: SessionSchema.ID
    agent: string
@@ -339,8 +339,12 @@ export const layer = Layer.effect(
        Stream.unwrap(
          result
            .get(input.sessionID)
-            .pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))),
-        ).pipe(Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event))),
+            .pipe(Effect.as(events.aggregateEvents({ aggregateID: input.sessionID, after: input.after }))),
+        ).pipe(
+          Stream.filter((event): event is EventV2.CursorEvent<SessionEvent.DurableEvent> =>
+            isDurableSessionEvent(event.event),
+          ),
+        ),
      prompt: Effect.fn("V2Session.prompt")((input) =>
        Effect.uninterruptible(
          Effect.gen(function* () {
@@ -409,9 +413,9 @@ export const layer = Layer.effect(
              sessionID,
              timestamp: yield* DateTime.now,
            })
-            if (event.durable === undefined)
+            if (event.seq === undefined)
              return yield* Effect.die("Interrupt request event is missing aggregate sequence")
-            yield* execution.interrupt(sessionID, event.durable.seq)
+            yield* execution.interrupt(sessionID, event.seq)
          }),
        ),
      ),
@@ -27,13 +27,13 @@ const Base = {
 }

 const options = {
-  durable: {
+  sync: {
    aggregate: "sessionID",
    version: 1,
  },
 } as const
 const stepSettlementOptions = {
-  durable: {
+  sync: {
    aggregate: "sessionID",
    version: 2,
  },
@@ -456,7 +456,7 @@ export namespace Compaction {

  export const Ended = EventV2.define({
    type: "session.next.compaction.ended",
-    durable: { aggregate: "sessionID", version: 2 },
+    sync: { aggregate: "sessionID", version: 2 },
    schema: {
      ...Base,
      messageID: SessionMessageID.ID,
@@ -74,11 +74,11 @@ export const admit = Effect.fn("SessionInput.admit")(function* (
    })
    .pipe(
      Effect.flatMap((event) =>
-        event.durable === undefined
+        event.seq === undefined
          ? Effect.die("Prompt admission event is missing aggregate sequence")
          : Effect.succeed(
              new Admitted({
-                admittedSeq: event.durable.seq,
+                admittedSeq: event.seq,
                id: input.id,
                sessionID: input.sessionID,
                prompt: input.prompt,
@@ -117,6 +117,13 @@ export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(functio
    readonly timeCreated: DateTime.Utc
  },
 ) {
+  const message = yield* db
+    .select({ id: SessionMessageTable.id })
+    .from(SessionMessageTable)
+    .where(eq(SessionMessageTable.id, input.id))
+    .get()
+    .pipe(Effect.orDie)
+  if (message) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
  const stored = yield* db
    .insert(SessionInputTable)
    .values({
@@ -201,6 +208,37 @@ const matchesPrompt = (input: Admitted, expected: { readonly sessionID: SessionS
  input.sessionID === expected.sessionID &&
  JSON.stringify(encodePrompt(input.prompt)) === JSON.stringify(encodePrompt(expected.prompt))

+export const guardReservedID = Effect.fn("SessionInput.guardReservedID")(function* (
+  db: DatabaseService,
+  event: EventV2.Payload,
+) {
+  if (
+    Schema.is(SessionEvent.PromptLifecycle.Admitted)(event) ||
+    Schema.is(SessionEvent.PromptLifecycle.Promoted)(event)
+  )
+    return
+  const id = reservedID(event)
+  if (id === undefined) return
+  const admitted = yield* db
+    .select({ id: SessionInputTable.id })
+    .from(SessionInputTable)
+    .where(eq(SessionInputTable.id, id))
+    .get()
+    .pipe(Effect.orDie)
+  if (admitted === undefined) return
+  return yield* Effect.die(new LifecycleConflict({ id }))
+})
+
+const reservedID = (event: EventV2.Payload) => {
+  if (Schema.is(SessionEvent.Step.Started)(event)) return event.data.assistantMessageID
+  if (Schema.is(SessionEvent.AgentSwitched)(event)) return event.data.messageID
+  if (Schema.is(SessionEvent.ModelSwitched)(event)) return event.data.messageID
+  if (Schema.is(SessionEvent.Prompted)(event)) return event.data.messageID
+  if (Schema.is(SessionEvent.Synthetic)(event)) return event.data.messageID
+  if (Schema.is(SessionEvent.Shell.Started)(event)) return event.data.messageID
+  if (Schema.is(SessionEvent.Compaction.Started)(event)) return event.data.messageID
+}
+
 export const projectLegacyPrompted = Effect.fn("SessionInput.projectLegacyPrompted")(function* (
  db: DatabaseService,
  input: {
@@ -115,7 +115,7 @@ function run(db: DatabaseService, event: SessionEvent.Event) {
    const decodeRow = (row: typeof SessionMessageTable.$inferSelect) =>
      decodeMessage({ ...row.data, id: row.id, type: row.type })
    const updateMessage = (message: SessionMessage.Message) => {
-      if (event.durable === undefined) return Effect.die("Durable Session event is missing aggregate sequence")
+      if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
      const encoded = encodeMessage(message)
      const { id, type, ...data } = encoded
      return db
@@ -192,7 +192,7 @@ function run(db: DatabaseService, event: SessionEvent.Event) {
 }

 function insertMessage(db: DatabaseService, event: SessionEvent.Event, message: SessionMessage.Message) {
-  if (event.durable === undefined) return Effect.die("Durable Session event is missing aggregate sequence")
+  if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
  const encoded = encodeMessage(message)
  const { id, type, ...data } = encoded
  return db
@@ -201,7 +201,7 @@ function insertMessage(db: DatabaseService, event: SessionEvent.Event, message:
      id: SessionMessage.ID.make(id),
      session_id: event.data.sessionID,
      type,
-      seq: event.durable.seq,
+      seq: event.seq,
      time_created: DateTime.toEpochMillis(message.time.created),
      data,
    })
@@ -213,6 +213,7 @@ export const layer = Layer.effectDiscard(
  Effect.gen(function* () {
    const events = yield* EventV2.Service
    const { db } = yield* Database.Service
+    yield* events.beforeCommit((event) => SessionInput.guardReservedID(db, event))
    yield* events.project(SessionV1.Event.Created, (event) =>
      Effect.gen(function* () {
        const stored = yield* db
@@ -330,7 +331,7 @@ export const layer = Layer.effectDiscard(
      }),
    )
    yield* events.project(SessionEvent.AgentSwitched, (event) => {
-      if (event.durable === undefined) return Effect.die("Durable Session event is missing aggregate sequence")
+      if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
      return db
        .update(SessionTable)
        .set({ agent: event.data.agent, time_updated: DateTime.toEpochMillis(event.data.timestamp) })
@@ -339,7 +340,7 @@ export const layer = Layer.effectDiscard(
        .pipe(
          Effect.orDie,
          Effect.andThen(run(db, event)),
-          Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.durable.seq)),
+          Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)),
        )
    })
    yield* events.project(SessionEvent.ModelSwitched, (event) =>
@@ -351,8 +352,9 @@ export const layer = Layer.effectDiscard(
          .run()
          .pipe(Effect.orDie)
        yield* run(db, event)
-        if (event.durable === undefined) return yield* Effect.die("Durable Session event is missing aggregate sequence")
-        yield* SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.durable.seq)
+        if (event.seq === undefined)
+          return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
+        yield* SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)
      }),
    )
    yield* events.project(SessionEvent.Prompted, (event) =>
@@ -366,22 +368,24 @@ export const layer = Layer.effectDiscard(
          .pipe(Effect.orDie)
        if (existing) return yield* Effect.die(new PromptAlreadyProjected())
        yield* run(db, event)
-        if (event.durable === undefined) return yield* Effect.die("Durable Session event is missing aggregate sequence")
+        if (event.seq === undefined)
+          return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
        yield* SessionInput.projectLegacyPrompted(db, {
          id: messageID,
          sessionID: event.data.sessionID,
          prompt: event.data.prompt,
          delivery: event.data.delivery,
          timeCreated: event.data.timestamp,
-          promotedSeq: event.durable.seq,
+          promotedSeq: event.seq,
        })
      }),
    )
    yield* events.project(SessionEvent.PromptLifecycle.Admitted, (event) =>
      Effect.gen(function* () {
-        if (event.durable === undefined) return yield* Effect.die("Durable Session event is missing aggregate sequence")
+        if (event.seq === undefined)
+          return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
        yield* SessionInput.projectAdmitted(db, {
-          admittedSeq: event.durable.seq,
+          admittedSeq: event.seq,
          id: event.data.messageID,
          sessionID: event.data.sessionID,
          prompt: event.data.prompt,
@@ -392,7 +396,8 @@ export const layer = Layer.effectDiscard(
    )
    yield* events.project(SessionEvent.PromptLifecycle.Promoted, (event) =>
      Effect.gen(function* () {
-        if (event.durable === undefined) return yield* Effect.die("Durable Session event is missing aggregate sequence")
+        if (event.seq === undefined)
+          return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
        yield* insertMessage(
          db,
          event,
@@ -401,14 +406,18 @@ export const layer = Layer.effectDiscard(
            sessionID: event.data.sessionID,
            prompt: event.data.prompt,
            timeCreated: event.data.timeCreated,
-            promotedSeq: event.durable.seq,
+            promotedSeq: event.seq,
          }),
        )
      }),
    )
    yield* events.project(SessionEvent.InterruptRequested, () => Effect.void)
-    // TODO: Reconstruct context epoch replacement state during replay without adding replay state to every EventV2 payload.
-    yield* events.project(SessionEvent.ContextUpdated, (event) => run(db, event))
+    yield* events.project(SessionEvent.ContextUpdated, (event) => {
+      if (!event.replay || event.seq === undefined) return run(db, event)
+      return run(db, event).pipe(
+        Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)),
+      )
+    })
    yield* events.project(SessionEvent.Synthetic, (event) => run(db, event))
    yield* events.project(SessionEvent.Shell.Started, (event) => run(db, event))
    yield* events.project(SessionEvent.Shell.Ended, (event) => run(db, event))
@@ -427,9 +436,9 @@ export const layer = Layer.effectDiscard(
    yield* events.project(SessionEvent.Reasoning.Ended, (event) => run(db, event))
    // yield* events.project(SessionEvent.Retried, (event) => run(db, event))
    yield* events.project(SessionEvent.Compaction.Ended, (event) => {
-      if (event.durable === undefined) return Effect.die("Durable Session event is missing aggregate sequence")
-      if (event.durable.version === 1) return Effect.void
-      const seq = event.durable.seq
+      if (event.version === 1) return Effect.void
+      const seq = event.seq
+      if (seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
      return Effect.gen(function* () {
        yield* run(db, event)
        yield* SessionContextEpoch.requestReplacement(db, event.data.sessionID, seq)
@@ -502,7 +502,7 @@ export type WithParts = {
 }

 const options = {
-  durable: {
+  sync: {
    aggregate: "sessionID",
    version: 1,
  },
@@ -61,7 +61,9 @@ describe("ConfigProviderPlugin.Plugin", () => {
        ...ConfigProviderPlugin.Plugin,
        effect: ConfigProviderPlugin.Plugin.effect(
          host({ catalog: catalogHost(catalog), integration: integrationHost(integrations) }),
-        ).pipe(Effect.provideService(Config.Service, config)),
+        ).pipe(
+          Effect.provideService(Config.Service, config),
+        ),
      })

      const model = required(yield* catalog.model.get(providerID, modelID))
@@ -120,7 +122,9 @@ describe("ConfigProviderPlugin.Plugin", () => {
        ...ConfigProviderPlugin.Plugin,
        effect: ConfigProviderPlugin.Plugin.effect(
          host({ catalog: catalogHost(catalog), integration: integrationHost(integrations) }),
-        ).pipe(Effect.provideService(Config.Service, config)),
+        ).pipe(
+          Effect.provideService(Config.Service, config),
+        ),
      })

      const model = required(yield* catalog.model.get(providerID, modelID))
@@ -221,7 +225,9 @@ describe("ConfigProviderPlugin.Plugin", () => {
          ...ConfigProviderPlugin.Plugin,
          effect: ConfigProviderPlugin.Plugin.effect(
            host({ catalog: catalogHost(catalog), integration: integrationHost(integrations) }),
-          ).pipe(Effect.provideService(Config.Service, config)),
+          ).pipe(
+            Effect.provideService(Config.Service, config),
+          ),
        })

        const provider = required(yield* catalog.provider.get(providerID))
@@ -30,7 +30,7 @@ const Message = EventV2.define({

 const SyncMessage = EventV2.define({
  type: "test.sync",
-  durable: {
+  sync: {
    version: 1,
    aggregate: "id",
  },
@@ -42,7 +42,7 @@ const SyncMessage = EventV2.define({

 const SyncSent = EventV2.define({
  type: "test.sent",
-  durable: {
+  sync: {
    version: 1,
    aggregate: "messageID",
  },
@@ -61,7 +61,7 @@ const GlobalMessage = EventV2.define({

 const VersionedMessage = EventV2.define({
  type: "test.versioned",
-  durable: {
+  sync: {
    version: 2,
    aggregate: "id",
  },
@@ -73,7 +73,7 @@ const VersionedMessage = EventV2.define({

 const SyncTimestamp = EventV2.define({
  type: "test.timestamp",
-  durable: {
+  sync: {
    version: 1,
    aggregate: "id",
  },
@@ -132,7 +132,7 @@ describe("EventV2", () => {
      const event = yield* events.publish(VersionedMessage, { id: "one", text: "hello" })

      expect(event.type).toBe("test.versioned")
-      expect(event.durable?.version).toBe(2)
+      expect(event.version).toBe(2)
    }),
  )

@@ -146,12 +146,12 @@ describe("EventV2", () => {
    Effect.sync(() => {
      const latest = EventV2.define({
        type: "test.out-of-order",
-        durable: { version: 2, aggregate: "id" },
+        sync: { version: 2, aggregate: "id" },
        schema: { id: Schema.String },
      })
      EventV2.define({
        type: "test.out-of-order",
-        durable: { version: 1, aggregate: "id" },
+        sync: { version: 1, aggregate: "id" },
        schema: { id: Schema.String },
      })

@@ -190,7 +190,7 @@ describe("EventV2", () => {
    }),
  )

-  it.effect("commits local operational state inside a new durable event transaction", () =>
+  it.effect("commits local operational state inside a new synchronized event transaction", () =>
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const received = new Array<string>()
@@ -207,7 +207,7 @@ describe("EventV2", () => {
    }),
  )

-  it.effect("rolls back the durable event and projector when the local commit fails", () =>
+  it.effect("rolls back the synchronized event and projector when the local commit fails", () =>
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const { db } = yield* Database.Service
@@ -236,7 +236,7 @@ describe("EventV2", () => {
      const events = yield* EventV2.Service
      const exit = yield* events.publish(Message, { text: "hello" }, { commit: () => Effect.void }).pipe(Effect.exit)

-      expect(String(exit)).toContain("Local commit hooks require a durable event")
+      expect(String(exit)).toContain("Local commit hooks require a synchronized event")
    }),
  )

@@ -290,6 +290,7 @@ describe("EventV2", () => {
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const received = new Array<string>()
+      yield* events.sync(() => Effect.die("sync defect"))
      yield* events.listen(() => {
        throw new Error("listener defect")
      })
@@ -302,7 +303,7 @@ describe("EventV2", () => {
      const event = yield* events.publish(SyncMessage, { id: "one", text: "hello" })

      expect(received).toEqual([SyncMessage.type])
-      expect(event.durable?.seq).toBeNumber()
+      expect(event.seq).toBeNumber()
    }),
  )

@@ -335,7 +336,49 @@ describe("EventV2", () => {
    }),
  )

-  it.effect("inserts durable event rows on publish", () =>
+  it.effect("does not synchronize live-only events", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const synchronized = new Array<string>()
+      const unsubscribe = yield* events.sync((event) =>
+        Effect.sync(() => {
+          synchronized.push(event.type)
+        }),
+      )
+      yield* Effect.addFinalizer(() => unsubscribe)
+
+      yield* events.publish(Message, { text: "live only" })
+      yield* events.publish(SyncMessage, { id: "one", text: "durable" })
+
+      expect(synchronized).toEqual([SyncMessage.type])
+    }),
+  )
+
+  it.effect("synchronizes only after the durable event commits", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const synchronized = new Array<boolean>()
+      yield* events.sync((event) =>
+        db
+          .select({ id: EventTable.id })
+          .from(EventTable)
+          .where(eq(EventTable.id, event.id))
+          .get()
+          .pipe(
+            Effect.orDie,
+            Effect.map((row) => synchronized.push(row !== undefined)),
+            Effect.asVoid,
+          ),
+      )
+
+      yield* events.publish(SyncMessage, { id: EventV2.ID.create(), text: "durable" })
+
+      expect(synchronized).toEqual([true])
+    }),
+  )
+
+  it.effect("inserts sync event rows on publish", () =>
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const { db } = yield* Database.Service
@@ -355,7 +398,7 @@ describe("EventV2", () => {
    }),
  )

-  it.effect("increments durable event seq per aggregate", () =>
+  it.effect("increments sync event seq per aggregate", () =>
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const { db } = yield* Database.Service
@@ -374,22 +417,22 @@ describe("EventV2", () => {
    }),
  )

-  it.effect("replays durable aggregate events after a sequence and tails new events", () =>
+  it.effect("replays durable aggregate events after a cursor and tails new events", () =>
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const aggregateID = EventV2.ID.create()
      yield* events.publish(SyncMessage, { id: aggregateID, text: "zero" })
      yield* events.publish(SyncMessage, { id: aggregateID, text: "one" })
      const fiber = yield* events
-        .durable({ aggregateID, after: 0 })
+        .aggregateEvents({ aggregateID, after: EventV2.Cursor.make(0) })
        .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
      yield* Effect.yieldNow

      yield* events.publish(SyncMessage, { id: aggregateID, text: "two" })

-      expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.data])).toEqual([
-        [1, { id: aggregateID, text: "one" }],
-        [2, { id: aggregateID, text: "two" }],
+      expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.cursor, event.event.data])).toEqual([
+        [EventV2.Cursor.make(1), { id: aggregateID, text: "one" }],
+        [EventV2.Cursor.make(2), { id: aggregateID, text: "two" }],
      ])
    }),
  )
@@ -399,18 +442,20 @@ describe("EventV2", () => {
      const events = yield* EventV2.Service
      const aggregateID = EventV2.ID.create()
      yield* events.publish(SyncMessage, { id: aggregateID, text: "zero" })
-      const fiber = yield* events.durable({ aggregateID }).pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
+      const fiber = yield* events
+        .aggregateEvents({ aggregateID })
+        .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)

      yield* events.publish(SyncMessage, { id: aggregateID, text: "one" })

      expect(
        Array.from(yield* Fiber.join(fiber)).map((event) => [
-          event.durable?.seq,
-          (event.data as { text: string }).text,
+          event.cursor,
+          (event.event.data as { text: string }).text,
        ]),
      ).toEqual([
-        [0, "zero"],
-        [1, "one"],
+        [EventV2.Cursor.make(0), "zero"],
+        [EventV2.Cursor.make(1), "one"],
      ])
    }),
  )
@@ -431,15 +476,17 @@ describe("EventV2", () => {
      yield* Effect.gen(function* () {
        const events = yield* EventV2.Service
        const aggregateID = EventV2.ID.create()
-        const fiber = yield* events.durable({ aggregateID }).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+        const fiber = yield* events
+          .aggregateEvents({ aggregateID })
+          .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
        yield* Deferred.await(readStarted)

        pause = false
        yield* events.publish(SyncMessage, { id: aggregateID, text: "during handoff" })
        yield* Deferred.succeed(continueRead, undefined)

-        expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.data])).toEqual([
-          [0, { id: aggregateID, text: "during handoff" }],
+        expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.cursor, event.event.data])).toEqual([
+          [EventV2.Cursor.make(0), { id: aggregateID, text: "during handoff" }],
        ])
      }).pipe(Effect.provide(Layer.mergeAll(database, eventLayer)))
    }),
@@ -451,7 +498,7 @@ describe("EventV2", () => {
      const aggregateID = EventV2.ID.create()
      const count = 64
      const fiber = yield* events
-        .durable({ aggregateID })
+        .aggregateEvents({ aggregateID })
        .pipe(Stream.take(count), Stream.runCollect, Effect.forkScoped)
      yield* Effect.yieldNow

@@ -459,8 +506,11 @@ describe("EventV2", () => {
        yield* events.publish(SyncMessage, { id: aggregateID, text: String(index) })
      }

-      expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.data])).toEqual(
-        Array.from({ length: count }, (_, index) => [index, { id: aggregateID, text: String(index) }]),
+      expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.cursor, event.event.data])).toEqual(
+        Array.from({ length: count }, (_, index) => [
+          EventV2.Cursor.make(index),
+          { id: aggregateID, text: String(index) },
+        ]),
      )
    }),
  )
@@ -469,13 +519,15 @@ describe("EventV2", () => {
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const aggregateID = EventV2.ID.create()
-      const fiber = yield* events.durable({ aggregateID }).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      const fiber = yield* events
+        .aggregateEvents({ aggregateID })
+        .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
      yield* Effect.yieldNow

      yield* events.publish(Message, { text: "live only" })
      yield* events.publish(SyncMessage, { id: aggregateID, text: "durable" })

-      expect(Array.from(yield* Fiber.join(fiber)).map((event) => event.type)).toEqual([SyncMessage.type])
+      expect(Array.from(yield* Fiber.join(fiber)).map((event) => event.event.type)).toEqual([SyncMessage.type])
    }),
  )

@@ -498,7 +550,7 @@ describe("EventV2", () => {
    }),
  )

-  it.effect("replays durable events through projectors", () =>
+  it.effect("replays sync events through projectors", () =>
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const received = new Array<EventV2.Payload>()
@@ -654,7 +706,7 @@ describe("EventV2", () => {
        })
        .pipe(Effect.exit)

-      expect(String(exit)).toContain("Unknown durable event type")
+      expect(String(exit)).toContain("Unknown sync event type")
    }),
  )

@@ -791,7 +843,7 @@ describe("EventV2", () => {
      const replayed = {
        id: published.id,
        type: EventV2.versionedType(SyncMessage.type, 1),
-        seq: published.durable!.seq,
+        seq: published.seq!,
        aggregateID,
        data: published.data,
      }
@@ -936,7 +988,7 @@ describe("EventV2", () => {
      yield* events.replay(replayed, { publish: true })
      yield* events.replay(replayed, { publish: true })

-      expect(received).toMatchObject([{ id: replayed.id, durable: { seq: 0, version: 1 }, data: replayed.data }])
+      expect(received).toMatchObject([{ id: replayed.id, seq: 0, data: replayed.data }])
    }),
  )

@@ -1058,7 +1110,7 @@ describe("EventV2", () => {
    }),
  )

-  it.effect("remove clears durable event sequence", () =>
+  it.effect("remove clears sync event sequence", () =>
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const received = new Array<EventV2.Payload>()
@@ -1,7 +1,7 @@
 import fs from "fs/promises"
 import path from "path"
 import { describe, expect } from "bun:test"
-import { Cause, Effect, Exit, Layer } from "effect"
+import { Effect, Exit, Layer } from "effect"
 import { FileSystem } from "@opencode-ai/core/filesystem"
 import { FSUtil } from "@opencode-ai/core/fs-util"
 import { Location } from "@opencode-ai/core/location"
@@ -31,13 +31,6 @@ const withTmp = <A, E, R>(f: (directory: string) => Effect.Effect<A, E, R>) =>
  ).pipe(Effect.flatMap((tmp) => f(tmp.path)))

 describe("FileSystem", () => {
-  const expectFail = (exit: Exit.Exit<unknown, unknown>) => {
-    expect(Exit.isFailure(exit)).toBe(true)
-    if (Exit.isSuccess(exit)) return
-    expect(Cause.hasFails(exit.cause)).toBe(true)
-    expect(Cause.hasDies(exit.cause)).toBe(false)
-  }
-
  it.live("reads text and binary files", () =>
    withTmp((directory) =>
      Effect.gen(function* () {
@@ -67,39 +60,13 @@ describe("FileSystem", () => {
    ),
  )

-  it.live("fails for missing paths", () =>
+  it.live("rejects lexical escapes", () =>
    withTmp((directory) =>
      Effect.gen(function* () {
-        const exit = yield* (yield* FileSystem.Service)
-          .read({ path: RelativePath.make("missing.txt") })
+        const result = yield* (yield* FileSystem.Service)
+          .read({ path: RelativePath.make("../outside.txt") })
          .pipe(Effect.exit)
-        expectFail(exit)
-      }).pipe(provide(directory)),
-    ),
-  )
-
-  it.live("fails for wrong path kinds", () =>
-    withTmp((directory) =>
-      Effect.gen(function* () {
-        yield* Effect.promise(() => fs.mkdir(path.join(directory, "src")))
-        yield* Effect.promise(() => fs.writeFile(path.join(directory, "README.md"), "# Test"))
-        const service = yield* FileSystem.Service
-        expectFail(yield* service.read({ path: RelativePath.make("src") }).pipe(Effect.exit))
-        expectFail(yield* service.list({ path: RelativePath.make("README.md") }).pipe(Effect.exit))
-      }).pipe(provide(directory)),
-    ),
-  )
-
-  it.live("fails for lexical and symlink escapes", () =>
-    withTmp((directory) =>
-      Effect.gen(function* () {
-        const outside = path.join(directory, "..", "outside.txt")
-        yield* Effect.promise(() => fs.writeFile(outside, "outside"))
-        yield* Effect.promise(() => fs.symlink(outside, path.join(directory, "linked.txt")))
-        const service = yield* FileSystem.Service
-        expectFail(yield* service.read({ path: RelativePath.make("../outside.txt") }).pipe(Effect.exit))
-        expectFail(yield* service.read({ path: RelativePath.make("linked.txt") }).pipe(Effect.exit))
-        yield* Effect.promise(() => fs.rm(outside))
+        expect(Exit.isFailure(result)).toBe(true)
      }).pipe(provide(directory)),
    ),
  )
@@ -255,9 +255,7 @@ function method(value: Integration.Method) {
  }
 }

-function internalMethod(
-  value: IntegrationOAuthMethod | IntegrationKeyMethod | IntegrationEnvMethod,
-): Integration.Method {
+function internalMethod(value: IntegrationOAuthMethod | IntegrationKeyMethod | IntegrationEnvMethod): Integration.Method {
  if (value.type === "env") return value
  if (value.type === "key") return value
  return {
@@ -35,9 +35,7 @@ describe("AnthropicPlugin", () => {
      const catalog = yield* Catalog.Service
      yield* addPlugin(plugin, AnthropicPlugin)
      yield* catalog.transform((catalog) => catalog.provider.update(provider("openai").id, () => {}))
-      expect(
-        required(yield* catalog.provider.get(ProviderV2.ID.openai)).request.headers["anthropic-beta"],
-      ).toBeUndefined()
+      expect(required(yield* catalog.provider.get(ProviderV2.ID.openai)).request.headers["anthropic-beta"]).toBeUndefined()
    }),
  )

@@ -157,8 +157,7 @@ describe("GithubCopilotPlugin", () => {
        catalog.model.update(ProviderV2.ID.make("github-copilot"), ModelV2.ID.make("gpt-5-chat-latest"), () => {})
      })
      expect(
-        required(yield* catalog.model.get(ProviderV2.ID.make("github-copilot"), ModelV2.ID.make("gpt-5-chat-latest")))
-          .enabled,
+        required(yield* catalog.model.get(ProviderV2.ID.make("github-copilot"), ModelV2.ID.make("gpt-5-chat-latest"))).enabled,
      ).toBe(false)
    }),
  )
@@ -173,8 +172,7 @@ describe("GithubCopilotPlugin", () => {
        catalog.model.update(ProviderV2.ID.make("custom-copilot"), ModelV2.ID.make("gpt-5-chat-latest"), () => {})
      })
      expect(
-        required(yield* catalog.model.get(ProviderV2.ID.make("custom-copilot"), ModelV2.ID.make("gpt-5-chat-latest")))
-          .enabled,
+        required(yield* catalog.model.get(ProviderV2.ID.make("custom-copilot"), ModelV2.ID.make("gpt-5-chat-latest"))).enabled,
      ).toBe(true)
    }),
  )
@@ -90,7 +90,7 @@ export function addPlugin(plugin: PluginV2.Interface, definition: Plugin<any>) {
    const npm = yield* Effect.serviceOption(Npm.Service)
    const effect =
      typeof definition.effect === "function"
-        ? definition.effect(
+          ? definition.effect(
            host({
              aisdk: aisdkHost(plugin),
              ...(Option.isSome(catalog) ? { catalog: catalogHost(catalog.value) } : {}),
@@ -118,9 +118,7 @@ describe("OpenAIPlugin", () => {
        catalog.model.update(item.id, ModelV2.ID.make("gpt-5-chat-latest"), () => {})
      })
      expect(required(yield* catalog.model.get(ProviderV2.ID.openai, ModelV2.ID.make("gpt-5"))).enabled).toBe(true)
-      expect(
-        required(yield* catalog.model.get(ProviderV2.ID.openai, ModelV2.ID.make("gpt-5-chat-latest"))).enabled,
-      ).toBe(false)
+      expect(required(yield* catalog.model.get(ProviderV2.ID.openai, ModelV2.ID.make("gpt-5-chat-latest"))).enabled).toBe(false)
    }),
  )

@@ -135,8 +133,7 @@ describe("OpenAIPlugin", () => {
        catalog.model.update(item.id, ModelV2.ID.make("gpt-5-chat-latest"), () => {})
      })
      expect(
-        required(yield* catalog.model.get(ProviderV2.ID.make("custom-openai"), ModelV2.ID.make("gpt-5-chat-latest")))
-          .enabled,
+        required(yield* catalog.model.get(ProviderV2.ID.make("custom-openai"), ModelV2.ID.make("gpt-5-chat-latest"))).enabled,
      ).toBe(true)
    }),
  )
@@ -22,7 +22,9 @@ const locationLayer = Layer.succeed(

 const pluginWithIntegrations = (catalog: Catalog.Interface, integrations: Integration.Interface) => ({
  ...OpencodePlugin,
-  effect: OpencodePlugin.effect(host({ catalog: catalogHost(catalog), integration: integrationHost(integrations) })),
+  effect: OpencodePlugin.effect(
+    host({ catalog: catalogHost(catalog), integration: integrationHost(integrations) }),
+  ),
 })

 describe("OpencodePlugin", () => {
@@ -81,9 +83,7 @@ describe("OpencodePlugin", () => {
          })
        })
        expect(required(yield* catalog.provider.get(ProviderV2.ID.opencode)).request.body.apiKey).toBe("public")
-        expect(required(yield* catalog.model.get(ProviderV2.ID.opencode, ModelV2.ID.make("output-only"))).enabled).toBe(
-          true,
-        )
+        expect(required(yield* catalog.model.get(ProviderV2.ID.opencode, ModelV2.ID.make("output-only"))).enabled).toBe(true)
      }),
    ),
  )
@@ -92,15 +92,12 @@ describe("OpenRouterPlugin", () => {
      })

      expect(
-        required(yield* catalog.model.get(ProviderV2.ID.make("openrouter"), ModelV2.ID.make("openai/gpt-5-chat")))
-          .enabled,
+        required(yield* catalog.model.get(ProviderV2.ID.make("openrouter"), ModelV2.ID.make("openai/gpt-5-chat"))).enabled,
      ).toBe(false)
      expect(
        required(yield* catalog.model.get(ProviderV2.ID.make("openrouter"), ModelV2.ID.make("openai/gpt-5"))).enabled,
      ).toBe(true)
-      expect(
-        required(yield* catalog.model.get(ProviderV2.ID.openai, ModelV2.ID.make("openai/gpt-5-chat"))).enabled,
-      ).toBe(true)
+      expect(required(yield* catalog.model.get(ProviderV2.ID.openai, ModelV2.ID.make("openai/gpt-5-chat"))).enabled).toBe(true)
    }),
  )

@@ -114,9 +111,8 @@ describe("OpenRouterPlugin", () => {
        catalog.model.update(ProviderV2.ID.make("custom-openrouter"), ModelV2.ID.make("gpt-5-chat-latest"), () => {})
      })
      expect(
-        required(
-          yield* catalog.model.get(ProviderV2.ID.make("custom-openrouter"), ModelV2.ID.make("gpt-5-chat-latest")),
-        ).enabled,
+        required(yield* catalog.model.get(ProviderV2.ID.make("custom-openrouter"), ModelV2.ID.make("gpt-5-chat-latest")))
+          .enabled,
      ).toBe(true)
    }),
  )
@@ -44,9 +44,7 @@ describe("VercelPlugin", () => {
      expect(required(yield* catalog.provider.get(ProviderV2.ID.make("vercel"))).request.headers).not.toHaveProperty(
        "HTTP-Referer",
      )
-      expect(required(yield* catalog.provider.get(ProviderV2.ID.make("vercel"))).request.headers).not.toHaveProperty(
-        "X-Title",
-      )
+      expect(required(yield* catalog.provider.get(ProviderV2.ID.make("vercel"))).request.headers).not.toHaveProperty("X-Title")
    }),
  )

@@ -220,8 +220,8 @@ describe("SessionV2.create", () => {
      expect(
        Array.from(yield* session.events({ sessionID: created.id }).pipe(Stream.take(2), Stream.runCollect)),
      ).toMatchObject([
-        { durable: { seq: 1 }, type: "session.next.prompt.admitted", data: { prompt: { text: "Hello" } } },
-        { durable: { seq: 2 }, type: "session.next.prompt.promoted" },
+        { cursor: 1, event: { type: "session.next.prompt.admitted", data: { prompt: { text: "Hello" } } } },
+        { cursor: 2, event: { type: "session.next.prompt.promoted" } },
      ])
    }),
  )
@@ -355,7 +355,7 @@ describe("SessionV2.create", () => {
      expect(yield* session.get(created.id)).toMatchObject({ model })
      expect(
        Array.from(yield* session.events({ sessionID: created.id }).pipe(Stream.take(1), Stream.runCollect)),
-      ).toMatchObject([{ type: "session.next.model.switched", data: { model } }])
+      ).toMatchObject([{ event: { type: "session.next.model.switched", data: { model } } }])
    }),
  )

@@ -162,7 +162,7 @@ describe("SessionProjector", () => {

      expect(
        yield* db.select().from(SessionInputTable).where(eq(SessionInputTable.id, id)).get().pipe(Effect.orDie),
-      ).toMatchObject({ promoted_seq: event.durable?.seq })
+      ).toMatchObject({ promoted_seq: event.seq })
    }),
  )

@@ -334,6 +334,134 @@ describe("SessionProjector", () => {
    }),
  )

+  it.effect("rejects a Prompted event that conflicts with an admitted inbox row", () =>
+    Effect.gen(function* () {
+      const { db } = yield* Database.Service
+      yield* db
+        .insert(ProjectTable)
+        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
+        .run()
+        .pipe(Effect.orDie)
+      yield* db
+        .insert(SessionTable)
+        .values({
+          id: sessionID,
+          project_id: Project.ID.global,
+          slug: "test",
+          directory: "/project",
+          title: "test",
+          version: "test",
+        })
+        .run()
+        .pipe(Effect.orDie)
+      const events = yield* EventV2.Service
+      const id = SessionMessage.ID.make("msg_conflict")
+      yield* SessionInput.admit(db, events, {
+        id,
+        sessionID,
+        prompt: new Prompt({ text: "admitted" }),
+        delivery: "steer",
+      })
+
+      const exit = yield* events
+        .publish(SessionEvent.Prompted, {
+          sessionID,
+          messageID: id,
+          timestamp: created,
+          prompt: new Prompt({ text: "different" }),
+          delivery: "steer",
+        })
+        .pipe(Effect.exit)
+
+      expect(String(exit)).toContain("SessionInput.LifecycleConflict")
+      expect(
+        yield* db.select().from(SessionInputTable).where(eq(SessionInputTable.id, id)).get().pipe(Effect.orDie),
+      ).toMatchObject({ promoted_seq: null })
+    }),
+  )
+
+  it.effect("rejects an assistant message ID that conflicts with an admitted inbox row", () =>
+    Effect.gen(function* () {
+      const { db } = yield* Database.Service
+      yield* db
+        .insert(ProjectTable)
+        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
+        .run()
+        .pipe(Effect.orDie)
+      yield* db
+        .insert(SessionTable)
+        .values({
+          id: sessionID,
+          project_id: Project.ID.global,
+          slug: "test",
+          directory: "/project",
+          title: "test",
+          version: "test",
+        })
+        .run()
+        .pipe(Effect.orDie)
+      const events = yield* EventV2.Service
+      const id = SessionMessage.ID.make("msg_conflict")
+      yield* SessionInput.admit(db, events, {
+        id,
+        sessionID,
+        prompt: new Prompt({ text: "admitted" }),
+        delivery: "steer",
+      })
+
+      const exit = yield* events
+        .publish(SessionEvent.Step.Started, {
+          sessionID,
+          timestamp: created,
+          assistantMessageID: id,
+          agent: "build",
+          model,
+        })
+        .pipe(Effect.exit)
+
+      expect(String(exit)).toContain("SessionInput.LifecycleConflict")
+      expect(
+        yield* db.select().from(SessionMessageTable).where(eq(SessionMessageTable.id, id)).get().pipe(Effect.orDie),
+      ).toBeUndefined()
+    }),
+  )
+
+  it.effect("rejects a Prompted delivery mode that conflicts with an admitted inbox row", () =>
+    Effect.gen(function* () {
+      const { db } = yield* Database.Service
+      yield* db
+        .insert(ProjectTable)
+        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
+        .run()
+        .pipe(Effect.orDie)
+      yield* db
+        .insert(SessionTable)
+        .values({
+          id: sessionID,
+          project_id: Project.ID.global,
+          slug: "test",
+          directory: "/project",
+          title: "test",
+          version: "test",
+        })
+        .run()
+        .pipe(Effect.orDie)
+      const events = yield* EventV2.Service
+      const id = SessionMessage.ID.make("msg_delivery_conflict")
+      const prompt = new Prompt({ text: "admitted" })
+      yield* SessionInput.admit(db, events, { id, sessionID, prompt, delivery: "queue" })
+
+      const exit = yield* events
+        .publish(SessionEvent.Prompted, { sessionID, messageID: id, timestamp: created, prompt, delivery: "steer" })
+        .pipe(Effect.exit)
+
+      expect(String(exit)).toContain("SessionInput.LifecycleConflict")
+      expect(
+        yield* db.select().from(SessionInputTable).where(eq(SessionInputTable.id, id)).get().pipe(Effect.orDie),
+      ).toMatchObject({ delivery: "queue", promoted_seq: null })
+    }),
+  )
+
  it.effect("does not revive a stale incomplete in-memory assistant projection", () =>
    Effect.gen(function* () {
      const stale = new SessionMessage.Assistant({
@@ -177,7 +177,7 @@ describe("SessionV2.prompt", () => {
    }),
  )

-  it.effect("streams durable Session events after an aggregate sequence", () =>
+  it.effect("streams durable Session events after an aggregate cursor", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* SessionV2.Service
@@ -191,19 +191,17 @@ describe("SessionV2.prompt", () => {
      yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER)
      const streamed = Array.from(yield* Fiber.join(fiber))

-      expect(streamed.map((event) => [event.durable?.seq, event.type])).toEqual([
-        [0, "session.next.prompt.admitted"],
-        [1, "session.next.prompt.admitted"],
-        [2, "session.next.prompt.promoted"],
-        [3, "session.next.prompt.promoted"],
+      expect(streamed.map((event) => [event.cursor, event.event.type])).toEqual([
+        [EventV2.Cursor.make(0), "session.next.prompt.admitted"],
+        [EventV2.Cursor.make(1), "session.next.prompt.admitted"],
+        [EventV2.Cursor.make(2), "session.next.prompt.promoted"],
+        [EventV2.Cursor.make(3), "session.next.prompt.promoted"],
      ])
      expect(
        Array.from(
-          yield* session
-            .events({ sessionID, after: streamed[0]!.durable?.seq })
-            .pipe(Stream.take(1), Stream.runCollect),
-        ).map((event) => [event.durable?.seq, event.type]),
-      ).toEqual([[1, "session.next.prompt.admitted"]])
+          yield* session.events({ sessionID, after: streamed[0]!.cursor }).pipe(Stream.take(1), Stream.runCollect),
+        ).map((event) => [event.cursor, event.event.type]),
+      ).toEqual([[EventV2.Cursor.make(1), "session.next.prompt.admitted"]])
    }),
  )

@@ -474,6 +472,58 @@ describe("SessionV2.prompt", () => {
    }),
  )

+  it.effect("rejects an input ID already used by a durable non-prompt event", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      const events = yield* EventV2.Service
+      yield* events.publish(SessionEvent.Synthetic, {
+        sessionID,
+        messageID,
+        timestamp: yield* DateTime.now,
+        text: "Collision",
+      })
+
+      const failure = yield* session
+        .prompt({ id: messageID, sessionID, prompt: new Prompt({ text: "Collision" }), resume: false })
+        .pipe(Effect.flip)
+
+      expect(failure._tag).toBe("Session.PromptConflictError")
+      expect(yield* admitted(messageID)).toBeUndefined()
+    }),
+  )
+
+  it.effect("rejects a durable event ID reserved by an admitted prompt without poisoning promotion", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const { db } = yield* Database.Service
+      const session = yield* SessionV2.Service
+      const events = yield* EventV2.Service
+      const prompt = new Prompt({ text: "Reserved prompt" })
+      yield* session.prompt({ id: messageID, sessionID, prompt, resume: false })
+
+      const failure = yield* events
+        .publish(SessionEvent.Synthetic, {
+          sessionID,
+          messageID,
+          timestamp: yield* DateTime.now,
+          text: "Conflicting synthetic",
+        })
+        .pipe(Effect.catchDefect(Effect.succeed))
+
+      expect(String(failure)).toContain("SessionInput.LifecycleConflict")
+      expect(yield* admitted(messageID)).not.toHaveProperty("promotedSeq")
+      expect(yield* session.messages({ sessionID })).toEqual([])
+
+      yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER)
+
+      expect(yield* admitted(messageID)).toMatchObject({ promotedSeq: 1 })
+      expect(yield* session.messages({ sessionID })).toMatchObject([
+        { id: messageID, type: "user", text: "Reserved prompt" },
+      ])
+    }),
+  )
+
  it.effect("rejects reuse of one globally unique message ID across sessions", () =>
    Effect.gen(function* () {
      yield* setup
@@ -19,17 +19,17 @@ const capture = () => {
      Effect.sync(() => {
        const event = { id: EventV2.ID.create(), type: definition.type, data } as EventV2.Payload<typeof definition>
        published.push({
-          type: definition.durable
-            ? EventV2.versionedType(definition.type, definition.durable.version)
-            : definition.type,
+          type: definition.sync ? EventV2.versionedType(definition.type, definition.sync.version) : definition.type,
          data,
        })
        return event
      }),
    subscribe: () => Stream.empty,
    all: () => Stream.empty,
-    durable: () => Stream.empty,
+    aggregateEvents: () => Stream.empty,
+    sync: () => Effect.succeed(Effect.void),
    listen: () => Effect.succeed(Effect.void),
+    beforeCommit: () => Effect.void,
    project: () => Effect.void,
    replay: () => Effect.void,
    replayAll: () => Effect.succeed(undefined),
@@ -1355,6 +1355,34 @@ describe("SessionRunnerLLM", () => {
    }),
  )

+  it.effect("replays retained context projections while replacement is pending", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      const events = yield* EventV2.Service
+      yield* session.prompt({ sessionID, prompt: new Prompt({ text: "First" }), resume: false })
+
+      requests.length = 0
+      response = []
+      yield* session.resume(sessionID)
+      systemBaseline = "Changed context"
+      yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Second" }), resume: false })
+      yield* session.resume(sessionID)
+      yield* events.publish(SessionEvent.ModelSwitched, {
+        sessionID,
+        messageID: SessionMessage.ID.create(),
+        timestamp: DateTime.makeUnsafe(1),
+        model: { id: ModelV2.ID.make("replacement"), providerID: ProviderV2.ID.make("fake") },
+      })
+
+      yield* replaySessionProjection(sessionID)
+      systemBaseline = "Replacement context"
+      yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Third" }), resume: false })
+      yield* session.resume(sessionID)
+      expect(requests.at(-1)?.system.map((part) => part.text)).toEqual(["Replacement context"])
+    }),
+  )
+
  it.effect("replaces the baseline lazily after completed compaction without reopening replacement on replay", () =>
    Effect.gen(function* () {
      yield* setup
@@ -1,11 +1,10 @@
 import { EOL } from "os"
 import { Effect } from "effect"
 import { FileSystem } from "@opencode-ai/core/filesystem"
-import { FSUtil } from "@opencode-ai/core/fs-util"
 import { LocationServiceMap } from "@opencode-ai/core/location-layer"
 import { Location } from "@opencode-ai/core/location"
 import { AbsolutePath, RelativePath } from "@opencode-ai/core/schema"
-import { CliError, effectCmd } from "../../effect-cmd"
+import { effectCmd } from "../../effect-cmd"
 import { cmd } from "../cmd"

 const filesystem = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
@@ -14,8 +13,6 @@ const filesystem = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
    Effect.provide(LocationServiceMap.layer),
  )

-const fileError = (error: FileSystem.PathError | FSUtil.Error) => new CliError({ message: error.message })
-
 const FileSearchCommand = effectCmd({
  command: "search <query>",
  describe: "search files by query",
@@ -41,9 +38,7 @@ const FileReadCommand = effectCmd({
      description: "File path to read",
    }),
  handler: Effect.fn("Cli.debug.file.read")(function* (args) {
-    const file = yield* filesystem(
-      FileSystem.Service.use((svc) => svc.read({ path: RelativePath.make(args.path) })),
-    ).pipe(Effect.mapError(fileError))
+    const file = yield* filesystem(FileSystem.Service.use((svc) => svc.read({ path: RelativePath.make(args.path) })))
    process.stdout.write(
      JSON.stringify(
        { content: Buffer.from(file.content).toString("base64"), encoding: "base64", mime: file.mime },
@@ -64,9 +59,7 @@ const FileListCommand = effectCmd({
      description: "File path to list",
    }),
  handler: Effect.fn("Cli.debug.file.list")(function* (args) {
-    const files = yield* filesystem(
-      FileSystem.Service.use((svc) => svc.list({ path: RelativePath.make(args.path) })),
-    ).pipe(Effect.mapError(fileError))
+    const files = yield* filesystem(FileSystem.Service.use((svc) => svc.list({ path: RelativePath.make(args.path) })))
    process.stdout.write(JSON.stringify(files, null, 2) + EOL)
  }),
 })
@@ -23,7 +23,10 @@ export const V2Command = effectCmd({
        small: Object.fromEntries(
          yield* Effect.all(
            all.map((provider) =>
-              Effect.map(catalog.model.small(provider.id), (model) => [provider.id, model?.id] as const),
+              Effect.map(
+                catalog.model.small(provider.id),
+                (model) => [provider.id, model?.id] as const,
+              ),
            ),
            { concurrency: "unbounded" },
          ),
@@ -45,9 +45,9 @@ export const layer = Layer.effect(
          workspace: workspaceID,
          payload: { id: event.id, type: event.type, properties: event.data },
        })
-        const durable = EventV2.registry.get(event.type)?.durable
-        if (durable === undefined || event.durable === undefined) return
-        const aggregateID = (event.data as Record<string, unknown>)[durable.aggregate]
+        const sync = EventV2.registry.get(event.type)?.sync
+        if (sync === undefined || event.seq === undefined || event.version === undefined) return
+        const aggregateID = (event.data as Record<string, unknown>)[sync.aggregate]
        if (typeof aggregateID !== "string") return
        GlobalBus.emit("event", {
          directory: event.location?.directory ?? ctx?.directory,
@@ -57,8 +57,8 @@ export const layer = Layer.effect(
            type: "sync",
            syncEvent: {
              id: event.id,
-              type: EventV2.versionedType(event.type, event.durable.version),
-              seq: event.durable.seq,
+              type: EventV2.versionedType(event.type, event.version),
+              seq: event.seq,
              aggregateID,
              data: event.data,
            },
@@ -16,13 +16,13 @@ const GlobalHealth = Schema.Struct({
 const SyncEventSchemas = EventV2.registry
  .values()
  .flatMap((definition) => {
-    if (!definition.durable) return []
+    if (!definition.sync) return []
    return [
      Schema.Struct({
        type: Schema.Literal("sync"),
        id: EventV2.ID,
        syncEvent: Schema.Struct({
-          type: Schema.Literal(EventV2.versionedType(definition.type, definition.durable.version)),
+          type: Schema.Literal(EventV2.versionedType(definition.type, definition.sync.version)),
          id: EventV2.ID,
          seq: Schema.Finite,
          aggregateID: Schema.String,
@@ -10,10 +10,6 @@ import ignore from "ignore"
 import path from "path"
 import { HttpApiBuilder } from "effect/unstable/httpapi"
 import { InstanceHttpApi } from "../api"
-import { InvalidRequestError } from "../errors"
-
-const invalidRequest = (error: FileSystem.PathError | FSUtil.Error) =>
-  new InvalidRequestError({ message: error.message, kind: error._tag })

 export const fileHandlers = HttpApiBuilder.group(InstanceHttpApi, "file", (handlers) =>
  Effect.gen(function* () {
@@ -83,9 +79,7 @@ export const fileHandlers = HttpApiBuilder.group(InstanceHttpApi, "file", (handl
            .readFileString(path.join(location.project.directory, ".ignore"))
            .pipe(Effect.catch(() => Effect.succeed("")))
          if (ignorefile) ignored.add(ignorefile)
-          return (yield* fs
-            .list({ path: RelativePath.make(ctx.query.path) })
-            .pipe(Effect.mapError(invalidRequest))).map((item) => ({
+          return (yield* fs.list({ path: RelativePath.make(ctx.query.path) })).map((item) => ({
            name: path.basename(item.path),
            path: item.path,
            absolute: path.resolve(location.directory, item.path),
@@ -107,7 +101,6 @@ export const fileHandlers = HttpApiBuilder.group(InstanceHttpApi, "file", (handl
      return yield* filesystem(
        FileSystem.Service.use((fs) => fs.read({ path: RelativePath.make(ctx.query.path) })),
      ).pipe(
-        Effect.mapError(invalidRequest),
        Effect.flatMap((item) =>
          Effect.gen(function* () {
            const text = item.content.includes(0)
@@ -1,25 +1,9 @@
 import type { FileSystemEntry } from "@opencode-ai/sdk/v2/types"
 import type { Effect } from "effect"
-import type { PlatformError } from "effect/PlatformError"
-
-export type FileSystemError =
-  | PlatformError
-  | {
-      readonly _tag: "FileSystemError"
-      readonly method: string
-      readonly cause?: unknown
-    }
-  | {
-      readonly _tag: "FileSystem.PathError"
-      readonly path: string
-      readonly reason: "lexical_escape" | "symlink_escape" | "not_file" | "not_directory"
-    }

 export interface FileSystem {
-  read(input: {
-    readonly path: string
-  }): Effect.Effect<{ readonly content: Uint8Array; readonly mime: string }, FileSystemError>
-  list(input?: { readonly path?: string }): Effect.Effect<FileSystemEntry[], FileSystemError>
+  read(input: { readonly path: string }): Effect.Effect<{ readonly content: Uint8Array; readonly mime: string }>
+  list(input?: { readonly path?: string }): Effect.Effect<FileSystemEntry[]>
  find(input: {
    readonly query: string
    readonly type?: "file" | "directory"
@@ -7,7 +7,7 @@ export type { AISDK, AISDKHooks } from "./aisdk.js"
 export type { Catalog, CatalogDraft, CatalogProviderRecord } from "./catalog.js"
 export type { Command, CommandDraft } from "./command.js"
 export type { Event, EventMap } from "./event.js"
-export type { FileSystem, FileSystemError } from "./filesystem.js"
+export type { FileSystem } from "./filesystem.js"
 export type { Integration, IntegrationDraft, IntegrationMethod, IntegrationMethodRegistration } from "./integration.js"
 export type { Location } from "./location.js"
 export type { Npm } from "./npm.js"
@@ -6,6 +6,7 @@ export type ClientOptions = {

 export type Event =
  | EventModelsDevRefreshed
+  | EventPluginAdded
  | EventIntegrationUpdated
  | EventCatalogUpdated
  | EventSessionCreated
@@ -52,7 +53,6 @@ export type Event =
  | EventInstallationUpdated
  | EventInstallationUpdateAvailable
  | EventFileEdited
-  | EventPluginAdded
  | EventPermissionV2Asked
  | EventPermissionV2Replied
  | EventReferenceUpdated
@@ -737,6 +737,13 @@ export type GlobalEvent = {
          [key: string]: unknown
        }
      }
+    | {
+        id: string
+        type: "plugin.added"
+        properties: {
+          id: string
+        }
+      }
    | {
        id: string
        type: "integration.updated"
@@ -1257,13 +1264,6 @@ export type GlobalEvent = {
          file: string
        }
      }
-    | {
-        id: string
-        type: "plugin.added"
-        properties: {
-          id: string
-        }
-      }
    | {
        id: string
        type: "permission.v2.asked"
@@ -4003,7 +4003,7 @@ export type ModelV2Info = {
    }
  }>
  time: {
-    released: number
+    released: number | "NaN" | "Infinity" | "-Infinity" | "Infinity" | "-Infinity" | "NaN"
  }
  cost: Array<{
    tier?: {
@@ -4228,6 +4228,14 @@ export type EventModelsDevRefreshed = {
  }
 }

+export type EventPluginAdded = {
+  id: string
+  type: "plugin.added"
+  properties: {
+    id: string
+  }
+}
+
 export type EventIntegrationUpdated = {
  id: string
  type: "integration.updated"
@@ -4794,14 +4802,6 @@ export type EventFileEdited = {
  }
 }

-export type EventPluginAdded = {
-  id: string
-  type: "plugin.added"
-  properties: {
-    id: string
-  }
-}
-
 export type EventPermissionV2Asked = {
  id: string
  type: "permission.v2.asked"
@@ -14481,6 +14481,9 @@
          {
            "$ref": "#/components/schemas/EventModels-devRefreshed"
          },
+          {
+            "$ref": "#/components/schemas/EventPluginAdded"
+          },
          {
            "$ref": "#/components/schemas/EventIntegrationUpdated"
          },
@@ -14619,9 +14622,6 @@
          {
            "$ref": "#/components/schemas/EventFileEdited"
          },
-          {
-            "$ref": "#/components/schemas/EventPluginAdded"
-          },
          {
            "$ref": "#/components/schemas/EventPermissionV2Asked"
          },
@@ -16645,6 +16645,31 @@
                "required": ["id", "type", "properties"],
                "additionalProperties": false
              },
+              {
+                "type": "object",
+                "properties": {
+                  "id": {
+                    "type": "string",
+                    "pattern": "^evt_"
+                  },
+                  "type": {
+                    "type": "string",
+                    "enum": ["plugin.added"]
+                  },
+                  "properties": {
+                    "type": "object",
+                    "properties": {
+                      "id": {
+                        "type": "string"
+                      }
+                    },
+                    "required": ["id"],
+                    "additionalProperties": false
+                  }
+                },
+                "required": ["id", "type", "properties"],
+                "additionalProperties": false
+              },
              {
                "type": "object",
                "properties": {
@@ -18428,31 +18453,6 @@
                "required": ["id", "type", "properties"],
                "additionalProperties": false
              },
-              {
-                "type": "object",
-                "properties": {
-                  "id": {
-                    "type": "string",
-                    "pattern": "^evt_"
-                  },
-                  "type": {
-                    "type": "string",
-                    "enum": ["plugin.added"]
-                  },
-                  "properties": {
-                    "type": "object",
-                    "properties": {
-                      "id": {
-                        "type": "string"
-                      }
-                    },
-                    "required": ["id"],
-                    "additionalProperties": false
-                  }
-                },
-                "required": ["id", "type", "properties"],
-                "additionalProperties": false
-              },
              {
                "type": "object",
                "properties": {
@@ -26900,7 +26900,27 @@
            "type": "object",
            "properties": {
              "released": {
-                "type": "number"
+                "anyOf": [
+                  {
+                    "type": "number"
+                  },
+                  {
+                    "type": "string",
+                    "enum": ["NaN"]
+                  },
+                  {
+                    "type": "string",
+                    "enum": ["Infinity"]
+                  },
+                  {
+                    "type": "string",
+                    "enum": ["-Infinity"]
+                  },
+                  {
+                    "type": "string",
+                    "enum": ["Infinity", "-Infinity", "NaN"]
+                  }
+                ]
              }
            },
            "required": ["released"],
@@ -27623,6 +27643,31 @@
        "required": ["id", "type", "properties"],
        "additionalProperties": false
      },
+      "EventPluginAdded": {
+        "type": "object",
+        "properties": {
+          "id": {
+            "type": "string",
+            "pattern": "^evt_"
+          },
+          "type": {
+            "type": "string",
+            "enum": ["plugin.added"]
+          },
+          "properties": {
+            "type": "object",
+            "properties": {
+              "id": {
+                "type": "string"
+              }
+            },
+            "required": ["id"],
+            "additionalProperties": false
+          }
+        },
+        "required": ["id", "type", "properties"],
+        "additionalProperties": false
+      },
      "EventIntegrationUpdated": {
        "type": "object",
        "properties": {
@@ -29398,31 +29443,6 @@
        "required": ["id", "type", "properties"],
        "additionalProperties": false
      },
-      "EventPluginAdded": {
-        "type": "object",
-        "properties": {
-          "id": {
-            "type": "string",
-            "pattern": "^evt_"
-          },
-          "type": {
-            "type": "string",
-            "enum": ["plugin.added"]
-          },
-          "properties": {
-            "type": "object",
-            "properties": {
-              "id": {
-                "type": "string"
-              }
-            },
-            "required": ["id"],
-            "additionalProperties": false
-          }
-        },
-        "required": ["id", "type", "properties"],
-        "additionalProperties": false
-      },
      "EventPermissionV2Asked": {
        "type": "object",
        "properties": {
@@ -1,28 +1,21 @@
 import { FileSystem } from "@opencode-ai/core/filesystem"
-import { FSUtil } from "@opencode-ai/core/fs-util"
 import { RelativePath } from "@opencode-ai/core/schema"
 import { Effect } from "effect"
 import { HttpServerResponse } from "effect/unstable/http"
 import { HttpApiBuilder } from "effect/unstable/httpapi"
 import { Api } from "../api"
-import { InvalidRequestError } from "../errors"
 import { response } from "../groups/location"

-const invalidRequest = (error: FileSystem.PathError | FSUtil.Error) =>
-  new InvalidRequestError({ message: error.message, kind: error._tag })
-
 export const FileSystemHandler = HttpApiBuilder.group(Api, "server.fs", (handlers) =>
  Effect.gen(function* () {
    return handlers
      .handleRaw("fs.read", (ctx) =>
        Effect.gen(function* () {
-          const file = yield* (yield* FileSystem.Service)
-            .read({
-              path: RelativePath.make(
-                decodeURIComponent(new URL(ctx.request.url, "http://localhost").pathname.slice(13)),
-              ),
-            })
-            .pipe(Effect.mapError(invalidRequest))
+          const file = yield* (yield* FileSystem.Service).read({
+            path: RelativePath.make(
+              decodeURIComponent(new URL(ctx.request.url, "http://localhost").pathname.slice(13)),
+            ),
+          })
          return HttpServerResponse.uint8Array(file.content, { contentType: file.mime })
        }),
      )
@@ -30,7 +23,7 @@ export const FileSystemHandler = HttpApiBuilder.group(Api, "server.fs", (handler
        response(
          Effect.gen(function* () {
            const fs = yield* FileSystem.Service
-            return yield* fs.list(ctx.query).pipe(Effect.mapError(invalidRequest))
+            return yield* fs.list(ctx.query)
          }),
        ),
      )
@@ -14,6 +14,7 @@ import {
  type StatsModelData,
  type UsageRange,
 } from "@opencode-ai/stats-core/domain/home"
+import { runtime } from "@opencode-ai/stats-core/runtime"
 import { createAsync, query, useParams } from "@solidjs/router"
 import { createMemo, createSignal, For, onMount, Show, type JSX } from "solid-js"
 import { getRequestEvent } from "solid-js/web"
@@ -26,7 +27,6 @@ import {
  type ModelCatalogCost,
  type ModelCatalogEntry,
 } from "../model-catalog"
-import { runStatsEffect } from "../../stats-runtime"
 import {
  applyThemePreference,
  Footer,
@@ -96,7 +96,7 @@ const worldBorderPath = worldPath(mesh(worldTopology, worldCountryGeometries, (a

 const getModelData = query(async (lab: string, model: string) => {
  "use server"
-  return runStatsEffect(getStatsModelData(model, lab))
+  return runtime.runPromise(getStatsModelData(model, lab))
 }, "getStatsModelData")

 export default function StatsModel() {
@@ -6,6 +6,7 @@ import {
  type ModelUsagePoint,
  type StatsLabData,
 } from "@opencode-ai/stats-core/domain/home"
+import { runtime } from "@opencode-ai/stats-core/runtime"
 import { createAsync, query, useParams } from "@solidjs/router"
 import { createMemo, createSignal, For, onMount, Show, type JSX } from "solid-js"
 import { getRequestEvent } from "solid-js/web"
@@ -16,7 +17,6 @@ import {
  type ModelCatalogEntry,
  type ModelCatalogLab,
 } from "../model-catalog"
-import { runStatsEffect } from "../../stats-runtime"
 import {
  applyThemePreference,
  Footer,
@@ -46,7 +46,7 @@ const labFooterLinks: readonly HeaderLink[] = [

 const getLabData = query(async (lab: string) => {
  "use server"
-  return runStatsEffect(getStatsLabData(lab))
+  return runtime.runPromise(getStatsLabData(lab))
 }, "getStatsLabData")

 export default function StatsLab() {
@@ -1,3 +1,19 @@
+import { AppConfig } from "@opencode-ai/stats-core/config"
+import { runtime } from "@opencode-ai/stats-core/runtime"
+import { Effect } from "effect"
+
 export async function GET() {
-  return Response.json({ ok: true, app: "stats" })
+  return Response.json(
+    await runtime.runPromise(
+      Effect.gen(function* () {
+        const config = yield* AppConfig
+        return {
+          ok: true,
+          app: "stats",
+          stage: config.stage,
+          publicUrl: config.publicUrl,
+        }
+      }),
+    ),
+  )
 }
@@ -21,12 +21,12 @@ import {
  type TokenCostEntry,
  type UsagePoint,
 } from "@opencode-ai/stats-core/domain/home"
+import { runtime } from "@opencode-ai/stats-core/runtime"
 import { createAsync, query } from "@solidjs/router"
 import { createEffect, createMemo, createSignal, For, onCleanup, onMount, Show, type JSX } from "solid-js"
 import { getRequestEvent } from "solid-js/web"
 import type { FeatureCollection, GeometryObject, GeoJsonProperties } from "geojson"
 import type { GeometryCollection, Topology } from "topojson-specification"
-import { runStatsEffect } from "../stats-runtime"
 import { findModelCatalogEntry, getModelCatalog, type ModelCatalog } from "./model-catalog"
 import {
  applyThemePreference,
@@ -109,7 +109,7 @@ const worldBorderPath = worldPath(mesh(worldTopology, worldCountryGeometries, (a

 const getData = query(async () => {
  "use server"
-  return runStatsEffect(getStatsHomeData())
+  return runtime.runPromise(getStatsHomeData())
 }, "getStatsHomeData")

 export default function StatsHome() {
@@ -1,5 +0,0 @@
-import { Effect } from "effect"
-
-export function runStatsEffect<A, E>(effect: Effect.Effect<A, E>) {
-  return Effect.runPromise(effect)
-}
@@ -17,8 +17,6 @@
  },
  "scripts": {
    "db:generate": "drizzle-kit generate --config=drizzle.config.ts",
-    "db:check-unique-users": "bun src/ensure-unique-users.ts --check",
-    "db:ensure-unique-users": "bun src/ensure-unique-users.ts",
    "db:migrate": "bun src/migrate.ts",
    "db:push": "drizzle-kit push --config=drizzle.config.ts",
    "db:studio": "drizzle-kit studio --config=drizzle.config.ts",
@@ -1,9 +1,8 @@
-import { Client } from "@planetscale/database"
 import { Effect } from "effect"
-import { Resource } from "sst/resource"
-import type { GeoStatMetric } from "./geo"
-import type { ModelStatMetric } from "./model"
-import type { ProviderStatMetric } from "./provider"
+import { DatabaseError } from "../database"
+import { GeoStatRepo, type GeoStatMetric } from "./geo"
+import { ModelStatRepo, type ModelStatMetric } from "./model"
+import { ProviderStatRepo, type ProviderStatMetric } from "./provider"

 export type UsageProduct = "All Users" | "Zen" | "Go" | "Enterprise"
 export type TokenProduct = "Zen" | "Go" | "Enterprise"
@@ -92,14 +91,6 @@ export type StatsHomeData = {
  country: Record<UsageRange, CountryEntry[]>
 }

-export class StatsDataError extends Error {
-  override name = "StatsDataError"
-
-  constructor(readonly cause: unknown) {
-    super("Failed to load stats data")
-  }
-}
-
 const DAY_MS = 86_400_000
 const TOKEN_SCALE = 1_000_000
 const DOLLARS_PER_MICROCENT = 1 / 100_000_000
@@ -139,136 +130,49 @@ type ModelAggregate = {
  totalCostMicrocents: number
 }

-type RawRow = Record<string, unknown>
+export const getStatsHomeData: () => Effect.Effect<
+  StatsHomeData,
+  DatabaseError,
+  ModelStatRepo | ProviderStatRepo | GeoStatRepo
+> = Effect.fn("StatsHome.getData")(function* () {
+  const modelStats = yield* ModelStatRepo
+  const providerStats = yield* ProviderStatRepo
+  const geoStats = yield* GeoStatRepo
+  const [modelRows, providerRows, geoRows] = yield* Effect.all(
+    [modelStats.listDaily(), providerStats.listDaily(), geoStats.listDaily()],
+    { concurrency: "unbounded" },
+  )
+  return buildStatsHomeData(modelRows, providerRows, geoRows)
+})

-export function getStatsHomeData(): Effect.Effect<StatsHomeData, StatsDataError> {
-  return Effect.tryPromise({
-    try: async () => {
-      const [modelRows, providerRows, geoRows] = await Promise.all([
-        listModelDaily(),
-        listProviderDaily(),
-        listGeoDaily(),
-      ])
-      return buildStatsHomeData(modelRows, providerRows, geoRows)
-    },
-    catch: (cause) => new StatsDataError(cause),
-  })
-}
-
-export function getStatsModelData(
+export const getStatsModelData: (
  model: string,
  provider?: string,
-): Effect.Effect<StatsModelData | null, StatsDataError> {
-  return Effect.tryPromise({
-    try: async () => {
-      const modelRows = await listModelDaily()
-      const normalized = modelRows.flatMap(normalizeStatRow)
-      const resolvedModel = resolveModelName(model, normalized, provider)
-      if (!resolvedModel) return null
-      return buildStatsModelData(
-        resolvedModel,
-        modelRows,
-        await listGeoDaily({
-          model: resolvedModel,
-          provider: resolveModelProvider(resolvedModel, normalized, provider),
-        }),
-        provider,
-      )
-    },
-    catch: (cause) => new StatsDataError(cause),
-  })
-}
-
-export function getStatsLabData(provider: string): Effect.Effect<StatsLabData | null, StatsDataError> {
-  return Effect.tryPromise({
-    try: async () => buildStatsLabData(provider, await listModelDaily()),
-    catch: (cause) => new StatsDataError(cause),
-  })
-}
-
-async function listModelDaily(): Promise<ModelStatMetric[]> {
-  return (
-    await queryRows(`select period_key, updated_at, tier, provider, model, sessions, unique_users, input_tokens,
-    output_tokens, reasoning_tokens, cache_read_tokens, total_tokens, input_cost_microcents, output_cost_microcents,
-    total_cost_microcents from model_stat where grain = 'day' and client = 'all' and source = 'all'
-    and tier in ('Go', 'go') order by period_key`)
-  ).map((row) => ({
-    periodKey: stringValue(row.period_key),
-    updatedAt: dateValue(row.updated_at),
-    tier: stringValue(row.tier),
-    provider: stringValue(row.provider),
-    model: stringValue(row.model),
-    sessions: numberValue(row.sessions),
-    uniqueUsers: numberValue(row.unique_users),
-    inputTokens: numberValue(row.input_tokens),
-    outputTokens: numberValue(row.output_tokens),
-    reasoningTokens: numberValue(row.reasoning_tokens),
-    cacheReadTokens: numberValue(row.cache_read_tokens),
-    totalTokens: numberValue(row.total_tokens),
-    inputCostMicrocents: numberValue(row.input_cost_microcents),
-    outputCostMicrocents: numberValue(row.output_cost_microcents),
-    totalCostMicrocents: numberValue(row.total_cost_microcents),
-  }))
-}
-
-async function listProviderDaily(): Promise<ProviderStatMetric[]> {
-  return (
-    await queryRows(`select period_key, updated_at, tier, provider, total_tokens from provider_stat
-    where grain = 'day' and client = 'all' and source = 'all' and tier in ('Go', 'go') order by period_key`)
-  ).map((row) => ({
-    periodKey: stringValue(row.period_key),
-    updatedAt: dateValue(row.updated_at),
-    tier: stringValue(row.tier),
-    provider: stringValue(row.provider),
-    totalTokens: numberValue(row.total_tokens),
-  }))
-}
-
-async function listGeoDaily(opts?: { provider?: string; model?: string }): Promise<GeoStatMetric[]> {
-  const scope =
-    opts?.model && opts.provider
-      ? "and provider = ? and model = ?"
-      : opts?.model
-        ? "and model = ?"
-        : "and provider = 'all' and model = 'all'"
-  const params = opts?.model && opts.provider ? [opts.provider, opts.model] : opts?.model ? [opts.model] : []
-  return (
-    await queryRows(
-      `select period_key, updated_at, tier, provider, model, country, continent, total_tokens from geo_stat
-    where grain = 'day' and client = 'all' and source = 'all' and tier in ('Go', 'go') ${scope} order by period_key`,
-      params,
+) => Effect.Effect<StatsModelData | null, DatabaseError, ModelStatRepo | GeoStatRepo> = Effect.fn("StatsModel.getData")(
+  function* (model, provider) {
+    const modelStats = yield* ModelStatRepo
+    const geoStats = yield* GeoStatRepo
+    const modelRows = yield* modelStats.listDaily()
+    const normalized = modelRows.flatMap(normalizeStatRow)
+    const resolvedModel = resolveModelName(model, normalized, provider)
+    if (!resolvedModel) return null
+    return buildStatsModelData(
+      resolvedModel,
+      modelRows,
+      yield* geoStats.listDaily({
+        model: resolvedModel,
+        provider: resolveModelProvider(resolvedModel, normalized, provider),
+      }),
+      provider,
    )
-  ).map((row) => ({
-    periodKey: stringValue(row.period_key),
-    updatedAt: dateValue(row.updated_at),
-    tier: stringValue(row.tier),
-    provider: stringValue(row.provider),
-    model: stringValue(row.model),
-    country: stringValue(row.country),
-    continent: stringValue(row.continent),
-    totalTokens: numberValue(row.total_tokens),
-  }))
-}
+  },
+)

-async function queryRows(query: string, params: string[] = []) {
-  return (await new Client({ url: databaseUrl() }).execute(query, params)).rows as RawRow[]
-}
-
-function databaseUrl() {
-  return process.env.DATABASE_URL ?? Resource.StatsDatabase.url
-}
-
-function stringValue(value: unknown) {
-  return value == null ? "" : String(value)
-}
-
-function numberValue(value: unknown) {
-  return Number(value ?? 0)
-}
-
-function dateValue(value: unknown) {
-  return value instanceof Date ? value : new Date(stringValue(value))
-}
+export const getStatsLabData: (provider: string) => Effect.Effect<StatsLabData | null, DatabaseError, ModelStatRepo> =
+  Effect.fn("StatsLab.getData")(function* (provider) {
+    const modelStats = yield* ModelStatRepo
+    return buildStatsLabData(provider, yield* modelStats.listDaily())
+  })

 function buildStatsHomeData(
  modelRows: ModelStatMetric[],
@@ -1,58 +0,0 @@
-import { Client } from "@planetscale/database"
-import { Resource } from "sst/resource"
-
-const tables = ["geo_stat", "model_stat", "provider_stat"] as const
-const checkOnly = process.argv.includes("--check")
-
-const client = new Client({ url: databaseUrl() })
-
-const missing = await tables.reduce<Promise<(typeof tables)[number][]>>(async (promise, table) => {
-  const result = await promise
-  if (await hasUniqueUsersColumn(table)) {
-    console.log(`unique_users column already exists on ${table}`)
-    return result
-  }
-  return [...result, table]
-}, Promise.resolve([]))
-
-if (missing.length === 0) {
-  console.log("unique_users columns complete")
-  process.exit(0)
-}
-
-if (checkOnly) {
-  console.log(`unique_users columns missing on ${missing.join(", ")}`)
-  process.exit(1)
-}
-
-await missing.reduce((promise, table) => promise.then(() => addUniqueUsersColumn(table)), Promise.resolve())
-
-function databaseUrl() {
-  if (
-    process.env.PLANETSCALE_HOST &&
-    process.env.PLANETSCALE_USERNAME &&
-    process.env.PLANETSCALE_PASSWORD &&
-    process.env.PLANETSCALE_DATABASE
-  )
-    return `mysql://${encodeURIComponent(process.env.PLANETSCALE_USERNAME)}:${encodeURIComponent(
-      process.env.PLANETSCALE_PASSWORD,
-    )}@${process.env.PLANETSCALE_HOST}/${process.env.PLANETSCALE_DATABASE}?ssl=${encodeURIComponent(
-      JSON.stringify({ rejectUnauthorized: true }),
-    )}`
-
-  return process.env.DATABASE_URL ?? Resource.StatsDatabase.url
-}
-
-async function hasUniqueUsersColumn(table: (typeof tables)[number]) {
-  const result = await client.execute<{ column_name: string }>(
-    "SELECT column_name FROM information_schema.columns WHERE table_schema = database() AND table_name = ? AND column_name = 'unique_users'",
-    [table],
-  )
-
-  return result.rows.length > 0
-}
-
-async function addUniqueUsersColumn(table: (typeof tables)[number]) {
-  await client.execute(`ALTER TABLE \`${table}\` ADD \`unique_users\` bigint NOT NULL DEFAULT 0`)
-  console.log(`added unique_users column to ${table}`)
-}