diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts
index 4d6236740..1229d3e8f 100644
--- a/packages/core/test/event.test.ts
+++ b/packages/core/test/event.test.ts
@@ -1,16 +1,20 @@
 import { describe, expect } from "bun:test"
 import { Effect, Fiber, Layer, Schema, Stream } from "effect"
 import { EventV2 } from "@opencode-ai/core/event"
+import { Database } from "@opencode-ai/core/database/database"
+import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql"
 import { Location } from "@opencode-ai/core/location"
 import { AbsolutePath } from "@opencode-ai/core/schema"
+import { eq } from "drizzle-orm"
 import { testEffect } from "./lib/effect"
 
 const locationLayer = Layer.succeed(
   Location.Service,
   Location.Service.of({ directory: AbsolutePath.make("project"), workspaceID: "workspace" }),
 )
-const it = testEffect(EventV2.defaultLayer.pipe(Layer.provideMerge(locationLayer)))
-const itWithoutLocation = testEffect(EventV2.defaultLayer)
+const eventLayer = Layer.mergeAll(EventV2.defaultLayer, Database.defaultLayer)
+const it = testEffect(eventLayer.pipe(Layer.provideMerge(locationLayer)))
+const itWithoutLocation = testEffect(eventLayer)
 
 const Message = EventV2.define({
   type: "test.message",
@@ -31,6 +35,18 @@ const SyncMessage = EventV2.define({
   },
 })
 
+const SyncSent = EventV2.define({
+  type: "test.sent",
+  sync: {
+    version: 1,
+    aggregate: "messageID",
+  },
+  schema: {
+    messageID: Schema.String,
+    text: Schema.String,
+  },
+})
+
 const GlobalMessage = EventV2.define({
   type: "test.global",
   schema: {
@@ -164,6 +180,49 @@ describe("EventV2", () => {
     }),
   )
 
+  it.effect("inserts sync event rows on publish", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.publish(SyncMessage, { id: aggregateID, text: "first" })
+      const rows = yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all().pipe(Effect.orDie)
+
+      expect(rows).toHaveLength(1)
+      expect(rows[0]?.type).toBe(EventV2.versionedType(SyncMessage.type, 1))
+      expect(rows[0]?.aggregate_id).toBe(aggregateID)
+    }),
+  )
+
+  it.effect("increments sync event seq per aggregate", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.publish(SyncMessage, { id: aggregateID, text: "first" })
+      yield* events.publish(SyncMessage, { id: aggregateID, text: "second" })
+      const rows = yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all().pipe(Effect.orDie)
+
+      expect(rows.map((row) => row.seq)).toEqual([0, 1])
+    }),
+  )
+
+  it.effect("uses custom sync aggregate field", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.publish(SyncSent, { messageID: aggregateID, text: "sent" })
+      const rows = yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all().pipe(Effect.orDie)
+
+      expect(rows).toHaveLength(1)
+      expect(rows[0]?.aggregate_id).toBe(aggregateID)
+    }),
+  )
+
   it.effect("replays sync events through projectors", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service
@@ -188,6 +247,69 @@ describe("EventV2", () => {
     }),
   )
 
+  it.effect("replay inserts external event rows", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.replay({
+        id: EventV2.ID.create(),
+        type: EventV2.versionedType(SyncMessage.type, 1),
+        seq: 0,
+        aggregateID,
+        data: { id: aggregateID, text: "replayed" },
+      })
+      const rows = yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all().pipe(Effect.orDie)
+
+      expect(rows).toHaveLength(1)
+      expect(rows[0]?.aggregate_id).toBe(aggregateID)
+    }),
+  )
+
+  it.effect("replay defects on sequence mismatch", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.replay({
+        id: EventV2.ID.create(),
+        type: EventV2.versionedType(SyncMessage.type, 1),
+        seq: 0,
+        aggregateID,
+        data: { id: aggregateID, text: "first" },
+      })
+      const exit = yield* events
+        .replay({
+          id: EventV2.ID.create(),
+          type: EventV2.versionedType(SyncMessage.type, 1),
+          seq: 5,
+          aggregateID,
+          data: { id: aggregateID, text: "bad" },
+        })
+        .pipe(Effect.exit)
+
+      expect(String(exit)).toContain("Sequence mismatch")
+    }),
+  )
+
+  it.effect("replay defects on unknown event type", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const exit = yield* events
+        .replay({
+          id: EventV2.ID.create(),
+          type: "unknown.event.1",
+          seq: 0,
+          aggregateID: EventV2.ID.create(),
+          data: {},
+        })
+        .pipe(Effect.exit)
+
+      expect(String(exit)).toContain("Unknown sync event type")
+    }),
+  )
+
   it.effect("replayAll validates contiguous aggregate events", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service
@@ -213,6 +335,52 @@ describe("EventV2", () => {
     }),
   )
 
+  it.effect("replayAll accepts later chunks after the first batch", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+
+      const one = yield* events.replayAll([
+        {
+          id: EventV2.ID.create(),
+          type: EventV2.versionedType(SyncMessage.type, 1),
+          seq: 0,
+          aggregateID,
+          data: { id: aggregateID, text: "one" },
+        },
+        {
+          id: EventV2.ID.create(),
+          type: EventV2.versionedType(SyncMessage.type, 1),
+          seq: 1,
+          aggregateID,
+          data: { id: aggregateID, text: "two" },
+        },
+      ])
+      const two = yield* events.replayAll([
+        {
+          id: EventV2.ID.create(),
+          type: EventV2.versionedType(SyncMessage.type, 1),
+          seq: 2,
+          aggregateID,
+          data: { id: aggregateID, text: "three" },
+        },
+        {
+          id: EventV2.ID.create(),
+          type: EventV2.versionedType(SyncMessage.type, 1),
+          seq: 3,
+          aggregateID,
+          data: { id: aggregateID, text: "four" },
+        },
+      ])
+      const rows = yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all().pipe(Effect.orDie)
+
+      expect(one).toBe(aggregateID)
+      expect(two).toBe(aggregateID)
+      expect(rows.map((row) => row.seq)).toEqual([0, 1, 2, 3])
+    }),
+  )
+
   it.effect("claim fences replay owners", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service
@@ -241,6 +409,92 @@ describe("EventV2", () => {
     }),
   )
 
+  it.effect("replay with owner claims an unowned sequence", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.replay(
+        {
+          id: EventV2.ID.create(),
+          type: EventV2.versionedType(SyncMessage.type, 1),
+          seq: 0,
+          aggregateID,
+          data: { id: aggregateID, text: "owned" },
+        },
+        { ownerID: "owner-1" },
+      )
+      const row = yield* db
+        .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
+        .from(EventSequenceTable)
+        .where(eq(EventSequenceTable.aggregate_id, aggregateID))
+        .get()
+        .pipe(Effect.orDie)
+
+      expect(row).toEqual({ seq: 0, ownerID: "owner-1" })
+    }),
+  )
+
+  it.effect("replay from a different owner leaves claimed sequence unchanged", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.replay(
+        {
+          id: EventV2.ID.create(),
+          type: EventV2.versionedType(SyncMessage.type, 1),
+          seq: 0,
+          aggregateID,
+          data: { id: aggregateID, text: "first" },
+        },
+        { ownerID: "owner-1" },
+      )
+      yield* events.replay(
+        {
+          id: EventV2.ID.create(),
+          type: EventV2.versionedType(SyncMessage.type, 1),
+          seq: 1,
+          aggregateID,
+          data: { id: aggregateID, text: "ignored" },
+        },
+        { ownerID: "owner-2" },
+      )
+      const rows = yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all().pipe(Effect.orDie)
+      const sequence = yield* db
+        .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
+        .from(EventSequenceTable)
+        .where(eq(EventSequenceTable.aggregate_id, aggregateID))
+        .get()
+        .pipe(Effect.orDie)
+
+      expect(rows).toHaveLength(1)
+      expect(sequence).toEqual({ seq: 0, ownerID: "owner-1" })
+    }),
+  )
+
+  it.effect("claim updates the event sequence owner", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+
+      yield* events.publish(SyncMessage, { id: aggregateID, text: "claimed" })
+      yield* events.claim(aggregateID, "owner-1")
+      yield* events.claim(aggregateID, "owner-2")
+      const row = yield* db
+        .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
+        .from(EventSequenceTable)
+        .where(eq(EventSequenceTable.aggregate_id, aggregateID))
+        .get()
+        .pipe(Effect.orDie)
+
+      expect(row).toEqual({ seq: 0, ownerID: "owner-2" })
+    }),
+  )
+
   it.effect("remove clears sync event sequence", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service