fix(cli): increment ports from default

fix(core): settle interrupted assistant steps (#33266 )
fix(core): handle read file failures (#33260 )
2026-06-21 19:19:56 -04:00 · 2026-06-21 20:22:09 +00:00 · 2026-06-21 20:12:42 +00:00 · 2026-06-21 14:53:46 -05:00 · 2026-06-21 15:37:25 -04:00 · 2026-06-21 13:50:14 -04:00
56 changed files with 1102 additions and 943 deletions
@@ -99,7 +99,8 @@ jobs:
      - name: Setup Node
        uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
        with:
-          node-version: "24"
+          # Playwright 1.59 hangs while extracting Chromium with Node 24.16.
+          node-version: "24.15"

      - name: Setup Bun
        uses: ./.github/actions/setup-bun
@@ -1,8 +1,8 @@
 {
  "nodeModules": {
-    "x86_64-linux": "sha256-g0tDvRf7MErZ1PEeUazEYi492ZHiRT8kYv3bPdkss/I=",
-    "aarch64-linux": "sha256-6sKgf3ftbIqlPxlFkoPzoWPsJp3IwXD+H3Y6g874xmk=",
-    "aarch64-darwin": "sha256-Se/Nls/KlkuK2ysDQ9DeAzSaX3NsL2iDdf/dsv2GIXc=",
-    "x86_64-darwin": "sha256-V9MCkqnvQ1nkD2PaaTfNFKkBZGymj6KxrSAK6+DTF8Y="
+    "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="
  }
 }
@@ -27,9 +27,11 @@ export default Runtime.handler(

 function listen(hostname: string, port: Option.Option<number>, password: string) {
  if (Option.isSome(port)) return bind(hostname, port.value, password)
-  // Preserve the familiar default when available, but let the OS choose a free
-  // port when another local server already owns 4096.
-  return bind(hostname, 4096, password).pipe(Effect.catch(() => bind(hostname, 0, password)))
+  const next = (port: number): ReturnType<typeof bind> =>
+    bind(hostname, port, password).pipe(
+      Effect.catch((error) => (port === 65_535 ? Effect.fail(error) : next(port + 1))),
+    )
+  return next(4096)
 }

 function bind(hostname: string, port: number, password: string) {
@@ -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, NonNegativeInt, withStatics } from "./schema"
+import { externalID, type ExternalID, withStatics } from "./schema"
 import { Identifier } from "./util/identifier"
 import { LayerNode } from "./effect/layer-node"
 import { isDeepStrictEqual } from "node:util"
@@ -19,16 +19,9 @@ 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 sync?: {
+  readonly durable?: {
    readonly version: number
    readonly aggregate: string
  }
@@ -41,20 +34,16 @@ export type Payload<D extends Definition = Definition> = {
  readonly id: ID
  readonly type: D["type"]
  readonly data: Data<D>
-  /** Durable aggregate order, populated while synchronized events are projected. */
-  readonly seq?: number
-  readonly version?: number
+  readonly durable?: {
+    readonly aggregateID: string
+    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 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 Subscriber<D extends Definition = Definition> = (event: Payload<D>) => Effect.Effect<void>
 export type Unsubscribe = Effect.Effect<void>

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

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

 export const registry = 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>
+const durableRegistry = new Map<string, Definition>()

 export function define<const Type extends string, Fields extends Schema.Struct.Fields>(input: {
  readonly type: Type
-  readonly sync?: {
+  readonly durable?: {
    readonly version: number
    readonly aggregate: string
  }
@@ -106,28 +82,25 @@ 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),
-    version: Schema.optional(Schema.Number),
+    durable: Schema.optional(Schema.Struct({ aggregateID: Schema.String, seq: Schema.Number, version: Schema.Number })),
    location: Schema.optional(Location.Ref),
    data: Data,
  }).annotate({ identifier: input.type })

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

@@ -152,14 +125,10 @@ export interface Interface {
  ) => Effect.Effect<Payload<D>>
  readonly subscribe: <D extends Definition>(definition: D) => Stream.Stream<Payload<D>>
  readonly all: () => Stream.Stream<Payload>
-  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 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 replay: (
    event: SerializedEvent,
    options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean },
@@ -182,37 +151,37 @@ export const layerWith = (options?: LayerOptions) =>
  Layer.effect(
    Service,
    Effect.gen(function* () {
-      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 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 { db } = yield* Database.Service

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

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

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

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

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

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

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

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

-      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}` })
+      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}` })
        }
        return {
-          cursor: Cursor.make(event.seq),
-          event: {
-            id: event.id,
-            type: definition.type,
-            version: definition.sync.version,
-            seq: event.seq,
-            data: definition.decode(event.data),
-          },
+          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),
        }
      }

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

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

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

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

-      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> =>
+      const project = <D extends Definition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> =>
        Effect.sync(() => {
          const list = projectors.get(definition.type) ?? []
          list.push((event) => projector(event as Payload<D>))
@@ -661,10 +616,8 @@ export const layerWith = (options?: LayerOptions) =>
        publish,
        subscribe,
        all: streamAll,
-        aggregateEvents: streamEvents,
-        sync,
+        durable,
        listen,
-        beforeCommit,
        project,
        replay,
        replayAll,
@@ -38,8 +38,7 @@ 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,7 +1,6 @@
 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"
@@ -34,9 +33,7 @@ export type Delivery = SessionInput.Delivery
 export const ListInput = SessionV2.ListInput
 export type ListInput = SessionV2.ListInput

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

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

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

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

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

  export const Ended = EventV2.define({
    type: "session.next.compaction.ended",
-    sync: { aggregate: "sessionID", version: 2 },
+    durable: { 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.seq === undefined
+        event.durable === undefined
          ? Effect.die("Prompt admission event is missing aggregate sequence")
          : Effect.succeed(
              new Admitted({
-                admittedSeq: event.seq,
+                admittedSeq: event.durable.seq,
                id: input.id,
                sessionID: input.sessionID,
                prompt: input.prompt,
@@ -117,13 +117,6 @@ 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({
@@ -208,37 +201,6 @@ 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.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
+      if (event.durable === undefined) return Effect.die("Durable 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.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
+  if (event.durable === undefined) return Effect.die("Durable 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.seq,
+      seq: event.durable.seq,
      time_created: DateTime.toEpochMillis(message.time.created),
      data,
    })
@@ -213,7 +213,6 @@ 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
@@ -331,7 +330,7 @@ export const layer = Layer.effectDiscard(
      }),
    )
    yield* events.project(SessionEvent.AgentSwitched, (event) => {
-      if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
+      if (event.durable === undefined) return Effect.die("Durable Session event is missing aggregate sequence")
      return db
        .update(SessionTable)
        .set({ agent: event.data.agent, time_updated: DateTime.toEpochMillis(event.data.timestamp) })
@@ -340,7 +339,7 @@ export const layer = Layer.effectDiscard(
        .pipe(
          Effect.orDie,
          Effect.andThen(run(db, event)),
-          Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)),
+          Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.durable.seq)),
        )
    })
    yield* events.project(SessionEvent.ModelSwitched, (event) =>
@@ -352,9 +351,8 @@ export const layer = Layer.effectDiscard(
          .run()
          .pipe(Effect.orDie)
        yield* run(db, event)
-        if (event.seq === undefined)
-          return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
-        yield* SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)
+        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)
      }),
    )
    yield* events.project(SessionEvent.Prompted, (event) =>
@@ -368,24 +366,22 @@ export const layer = Layer.effectDiscard(
          .pipe(Effect.orDie)
        if (existing) return yield* Effect.die(new PromptAlreadyProjected())
        yield* run(db, event)
-        if (event.seq === undefined)
-          return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
+        if (event.durable === undefined) return yield* Effect.die("Durable 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.seq,
+          promotedSeq: event.durable.seq,
        })
      }),
    )
    yield* events.project(SessionEvent.PromptLifecycle.Admitted, (event) =>
      Effect.gen(function* () {
-        if (event.seq === undefined)
-          return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
+        if (event.durable === undefined) return yield* Effect.die("Durable Session event is missing aggregate sequence")
        yield* SessionInput.projectAdmitted(db, {
-          admittedSeq: event.seq,
+          admittedSeq: event.durable.seq,
          id: event.data.messageID,
          sessionID: event.data.sessionID,
          prompt: event.data.prompt,
@@ -396,8 +392,7 @@ export const layer = Layer.effectDiscard(
    )
    yield* events.project(SessionEvent.PromptLifecycle.Promoted, (event) =>
      Effect.gen(function* () {
-        if (event.seq === undefined)
-          return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
+        if (event.durable === undefined) return yield* Effect.die("Durable Session event is missing aggregate sequence")
        yield* insertMessage(
          db,
          event,
@@ -406,18 +401,14 @@ export const layer = Layer.effectDiscard(
            sessionID: event.data.sessionID,
            prompt: event.data.prompt,
            timeCreated: event.data.timeCreated,
-            promotedSeq: event.seq,
+            promotedSeq: event.durable.seq,
          }),
        )
      }),
    )
    yield* events.project(SessionEvent.InterruptRequested, () => Effect.void)
-    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)),
-      )
-    })
+    // 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.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))
@@ -436,9 +427,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.version === 1) return Effect.void
-      const seq = event.seq
-      if (seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
+      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
      return Effect.gen(function* () {
        yield* run(db, event)
        yield* SessionContextEpoch.requestReplacement(db, event.data.sessionID, seq)
@@ -305,14 +305,7 @@ export const layer = Layer.effect(
          const llmFailure = failure instanceof LLMError ? failure : undefined
          if (llmFailure && !publisher.hasProviderError()) {
            yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true))
-            yield* withPublication(
-              events.publish(SessionEvent.Step.Failed, {
-                sessionID: session.id,
-                timestamp: yield* DateTime.now,
-                assistantMessageID: yield* publisher.startAssistant(),
-                error: { type: "unknown", message: llmFailure.reason.message },
-              }),
-            )
+            yield* withPublication(publisher.failAssistant(llmFailure.reason.message))
          }
          if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers)
          const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit)
@@ -327,6 +320,8 @@ export const layer = Layer.effect(
          ) {
            yield* FiberSet.clear(toolFibers)
            yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted"))
+            if (publisher.hasActiveAssistant())
+              yield* withPublication(publisher.failAssistant("Provider turn interrupted"))
          }
          if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) {
            const failure = Cause.squash(settled.cause)
@@ -65,11 +65,14 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
  >()
  const timestamp = DateTime.now
  let assistantMessageID: SessionMessage.ID | undefined
+  let assistantActive = false
+  let assistantFailed = false
  let providerFailed = false

  const startAssistant = Effect.fnUntraced(function* () {
    if (assistantMessageID !== undefined) return assistantMessageID
    assistantMessageID = SessionMessage.ID.create()
+    assistantActive = true
    yield* events.publish(SessionEvent.Step.Started, {
      ...input,
      assistantMessageID,
@@ -190,6 +193,20 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
    yield* flushFragments()
  })

+  const failAssistant = Effect.fnUntraced(function* (message: string) {
+    if (assistantFailed) return
+    yield* flush()
+    const assistantMessageID = yield* startAssistant()
+    assistantActive = false
+    assistantFailed = true
+    yield* events.publish(SessionEvent.Step.Failed, {
+      sessionID: input.sessionID,
+      timestamp: yield* timestamp,
+      assistantMessageID,
+      error: { type: "unknown", message },
+    })
+  })
+
  const failUnsettledTools = Effect.fn("SessionRunner.failUnsettledTools")(function* (
    message: string,
    hostedOnly = false,
@@ -375,6 +392,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
      }
      case "step-finish":
        yield* flush()
+        assistantActive = false
        yield* events.publish(SessionEvent.Step.Ended, {
          sessionID: input.sessionID,
          timestamp: yield* timestamp,
@@ -388,13 +406,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
        return
      case "provider-error":
        providerFailed = true
-        yield* flush()
-        yield* events.publish(SessionEvent.Step.Failed, {
-          sessionID: input.sessionID,
-          timestamp: yield* timestamp,
-          assistantMessageID: yield* startAssistant(),
-          error: { type: "unknown", message: event.message },
-        })
+        yield* failAssistant(event.message)
        return
    }
  })
@@ -402,10 +414,11 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
  return {
    publish,
    flush,
+    failAssistant,
    failUnsettledTools,
+    hasActiveAssistant: () => assistantActive,
    hasAssistantStarted: () => assistantMessageID !== undefined,
    hasProviderError: () => providerFailed,
-    startAssistant,
    assistantMessageID: assistantMessageIDForTool,
  }
 }
@@ -82,12 +82,21 @@ const assistant = (message: SessionMessage.Assistant, model: Model) => {
    const result = toolResult(item, sameModel ? (item.provider?.resultMetadata ?? item.provider?.metadata) : undefined)
    return item.provider?.executed === true && result ? [call, result] : [call]
  })
+  const meaningful = content.filter((part) => {
+    if (part.type === "text") return part.text !== ""
+    if (part.type !== "reasoning") return true
+    return part.text !== "" || (part.providerMetadata !== undefined && Object.keys(part.providerMetadata).length > 0)
+  })
  const results = message.content
    .filter((item): item is SessionMessage.AssistantTool => item.type === "tool" && item.provider?.executed !== true)
    .map((item) => toolResult(item, sameModel ? (item.provider?.resultMetadata ?? item.provider?.metadata) : undefined))
    .filter((message) => message !== undefined)
    .map(Message.tool)
-  return [Message.make({ id: message.id, role: "assistant", content, metadata: message.metadata }), ...results]
+  if (meaningful.length === 0) return results
+  return [
+    Message.make({ id: message.id, role: "assistant", content: meaningful, metadata: message.metadata }),
+    ...results,
+  ]
 }

 function toLLMMessage(message: SessionMessage.Message, model: Model): Message[] {
@@ -0,0 +1,30 @@
+import { Effect, Stream } from "effect"
+import { HttpClientResponse } from "effect/unstable/http"
+
+export const collectBoundedResponseBody = (
+  response: HttpClientResponse.HttpClientResponse,
+  maximumBytes: number,
+  tooLarge: () => Error,
+) =>
+  Effect.gen(function* () {
+    const contentLength = response.headers["content-length"]
+    const parsedSize = contentLength ? Number.parseInt(contentLength, 10) : undefined
+    const declaredSize =
+      parsedSize !== undefined && Number.isSafeInteger(parsedSize) && parsedSize >= 0 ? parsedSize : undefined
+    if (declaredSize !== undefined && declaredSize > maximumBytes) return yield* Effect.fail(tooLarge())
+    let body = Buffer.allocUnsafe(Math.min(maximumBytes, declaredSize || 64 * 1024))
+    let size = 0
+    yield* Stream.runForEach(response.stream, (chunk) => {
+      if (chunk.byteLength === 0) return Effect.void
+      if (size + chunk.byteLength > maximumBytes) return Effect.fail(tooLarge())
+      if (size + chunk.byteLength > body.byteLength) {
+        const grown = Buffer.allocUnsafe(Math.min(maximumBytes, Math.max(size + chunk.byteLength, body.byteLength * 2)))
+        body.copy(grown, 0, 0, size)
+        body = grown
+      }
+      body.set(chunk, size)
+      size += chunk.byteLength
+      return Effect.void
+    })
+    return body.subarray(0, size)
+  })
@@ -13,23 +13,61 @@ export const MAX_MEDIA_INGEST_BYTES = 20 * 1024 * 1024
 const MAX_LINE_LENGTH = 2_000
 const MAX_LINE_SUFFIX = `... (line truncated to ${MAX_LINE_LENGTH} chars)`

-export class BinaryFileError extends Error {
-  constructor(readonly resource: string) {
-    super(`Cannot read binary file: ${resource}`)
-    this.name = "BinaryFileError"
+export class BinaryFileError extends Schema.TaggedErrorClass<BinaryFileError>()("ReadTool.BinaryFileError", {
+  resource: Schema.String,
+}) {
+  override get message() {
+    return `Cannot read binary file: ${this.resource}`
  }
 }

-export class MediaIngestLimitError extends Error {
-  constructor(
-    readonly resource: string,
-    readonly maximumBytes: number,
-  ) {
-    super(`Media exceeds ${maximumBytes} byte ingestion limit: ${resource}`)
-    this.name = "MediaIngestLimitError"
+export class MediaIngestLimitError extends Schema.TaggedErrorClass<MediaIngestLimitError>()(
+  "ReadTool.MediaIngestLimitError",
+  {
+    resource: Schema.String,
+    maximumBytes: Schema.Number,
+  },
+) {
+  override get message() {
+    return `Media exceeds ${this.maximumBytes} byte ingestion limit: ${this.resource}`
  }
 }

+export class MalformedUtf8Error extends Schema.TaggedErrorClass<MalformedUtf8Error>()("ReadTool.MalformedUtf8Error", {
+  resource: Schema.String,
+}) {
+  override get message() {
+    return `File is not valid UTF-8: ${this.resource}`
+  }
+}
+
+export class OffsetOutOfRangeError extends Schema.TaggedErrorClass<OffsetOutOfRangeError>()(
+  "ReadTool.OffsetOutOfRangeError",
+  { offset: Schema.Number },
+) {
+  override get message() {
+    return `Offset ${this.offset} is out of range`
+  }
+}
+
+export class PathKindError extends Schema.TaggedErrorClass<PathKindError>()("ReadTool.PathKindError", {
+  resource: Schema.String,
+  expected: Schema.Literals(["a file", "a file or directory"]),
+}) {
+  override get message() {
+    return `Path is not ${this.expected}: ${this.resource}`
+  }
+}
+
+export type InspectError = FSUtil.Error | PathKindError
+export type ReadError =
+  | FSUtil.Error
+  | BinaryFileError
+  | MediaIngestLimitError
+  | MalformedUtf8Error
+  | OffsetOutOfRangeError
+  | PathKindError
+
 export const PageInput = Schema.Struct({
  offset: PositiveInt.pipe(Schema.optional),
  limit: PositiveInt.check(Schema.isLessThanOrEqualTo(MAX_READ_LINES)).pipe(Schema.optional),
@@ -52,13 +90,13 @@ export class ListPage extends Schema.Class<ListPage>("ReadTool.ListPage")({
 }) {}

 export interface Interface {
-  readonly inspect: (path: AbsolutePath) => Effect.Effect<"file" | "directory">
+  readonly inspect: (path: AbsolutePath) => Effect.Effect<"file" | "directory", InspectError>
  readonly read: (
    path: AbsolutePath,
    resource: string,
    page?: PageInput,
-  ) => Effect.Effect<FileSystem.Content | TextPage>
-  readonly list: (path: AbsolutePath, page?: PageInput) => Effect.Effect<ListPage>
+  ) => Effect.Effect<FileSystem.Content | TextPage, ReadError>
+  readonly list: (path: AbsolutePath, page?: PageInput) => Effect.Effect<ListPage, FSUtil.Error>
 }

 export class Service extends Context.Service<Service, Interface>()("@opencode/ReadToolFileSystem") {}
@@ -111,11 +149,21 @@ const binary = (resource: string, bytes: Uint8Array) => {
  }
  return nonPrintable / bytes.length > 0.3
 }
+const decodeUtf8 = (resource: string, decoder: TextDecoder, bytes?: Uint8Array) =>
+  Effect.try({
+    try: () => decoder.decode(bytes, { stream: bytes !== undefined }),
+    catch: (error) => {
+      if (error instanceof TypeError) return new MalformedUtf8Error({ resource })
+      throw error
+    },
+  })
+const decodeChunk = (resource: string, decoder: TextDecoder, bytes: Uint8Array) =>
+  bytes.includes(0) ? Effect.fail(new BinaryFileError({ resource })) : decodeUtf8(resource, decoder, bytes)

 export const inspect = Effect.fn("ReadTool.inspect")(function* (fs: FSUtil.Interface, input: string) {
-  const info = yield* fs.stat(input).pipe(Effect.orDie)
+  const info = yield* fs.stat(input)
  const type = info.type === "File" ? "file" : info.type === "Directory" ? "directory" : undefined
-  if (!type) return yield* Effect.die(new Error("Path is not a file or directory"))
+  if (!type) return yield* Effect.fail(new PathKindError({ resource: input, expected: "a file or directory" }))
  return type
 })

@@ -125,32 +173,30 @@ export const read = Effect.fn("ReadTool.read")(function* (
  resource: string,
  page: PageInput = {},
 ) {
-  const real = yield* fs.realPath(input).pipe(Effect.orDie)
+  const real = yield* fs.realPath(input)
  return yield* Effect.scoped(
    Effect.gen(function* () {
-      const file = yield* fs.open(real, { flag: "r" }).pipe(Effect.orDie)
-      const info = yield* file.stat.pipe(Effect.orDie)
-      if (info.type !== "File") return yield* Effect.die(new Error("Path is not a file"))
+      const file = yield* fs.open(real, { flag: "r" })
+      const info = yield* file.stat
+      if (info.type !== "File") return yield* Effect.fail(new PathKindError({ resource, expected: "a file" }))
      const first = Option.getOrElse(
-        yield* file.readAlloc(Math.min(64 * 1024, Number(info.size) || 4 * 1024)).pipe(Effect.orDie),
+        yield* file.readAlloc(Math.min(64 * 1024, Number(info.size) || 4 * 1024)),
        () => new Uint8Array(),
      )
      const mime = imageMime(first)
      if (mime) {
        if (info.size > MAX_MEDIA_INGEST_BYTES)
-          return yield* Effect.die(new MediaIngestLimitError(resource, MAX_MEDIA_INGEST_BYTES))
+          return yield* Effect.fail(new MediaIngestLimitError({ resource, maximumBytes: MAX_MEDIA_INGEST_BYTES }))
        const chunks = [first]
        let total = first.length
        while (total <= MAX_MEDIA_INGEST_BYTES) {
-          const chunk = yield* file
-            .readAlloc(Math.min(64 * 1024, MAX_MEDIA_INGEST_BYTES + 1 - total))
-            .pipe(Effect.orDie)
+          const chunk = yield* file.readAlloc(Math.min(64 * 1024, MAX_MEDIA_INGEST_BYTES + 1 - total))
          if (Option.isNone(chunk)) break
          chunks.push(chunk.value)
          total += chunk.value.length
        }
        if (total > MAX_MEDIA_INGEST_BYTES)
-          return yield* Effect.die(new MediaIngestLimitError(resource, MAX_MEDIA_INGEST_BYTES))
+          return yield* Effect.fail(new MediaIngestLimitError({ resource, maximumBytes: MAX_MEDIA_INGEST_BYTES }))
        return {
          uri: pathToFileURL(real).href,
          name: path.basename(real),
@@ -162,19 +208,19 @@ export const read = Effect.fn("ReadTool.read")(function* (
          mime,
        }
      }
-      if (startsWith(first, [0x25, 0x50, 0x44, 0x46]) || binary(resource, first))
-        return yield* Effect.die(new BinaryFileError(resource))
+      if (startsWith(first, [0x25, 0x50, 0x44, 0x46]) || extensions.has(path.extname(resource).toLowerCase()))
+        return yield* Effect.fail(new BinaryFileError({ resource }))
      const paged = info.size > MAX_READ_BYTES || page.offset !== undefined || page.limit !== undefined
      if (!paged) {
+        if (binary(resource, first)) return yield* Effect.fail(new BinaryFileError({ resource }))
        const decoder = new TextDecoder("utf-8", { fatal: true })
-        const text = [yield* Effect.sync(() => decoder.decode(first, { stream: true }))]
+        const text = [yield* decodeUtf8(resource, decoder, first)]
        while (true) {
-          const chunk = yield* file.readAlloc(64 * 1024).pipe(Effect.orDie)
+          const chunk = yield* file.readAlloc(64 * 1024)
          if (Option.isNone(chunk)) break
-          if (chunk.value.includes(0)) return yield* Effect.die(new BinaryFileError(resource))
-          text.push(yield* Effect.sync(() => decoder.decode(chunk.value, { stream: true })))
+          text.push(yield* decodeChunk(resource, decoder, chunk.value))
        }
-        text.push(yield* Effect.sync(() => decoder.decode()))
+        text.push(yield* decodeUtf8(resource, decoder))
        return {
          uri: pathToFileURL(real).href,
          name: path.basename(real),
@@ -191,34 +237,29 @@ export const read = Effect.fn("ReadTool.read")(function* (
      let discard = false
      let line = 1
      let bytes = 0
-      let found = false
-      let truncated = false
      let next: number | undefined
      const append = (input: string) => {
        if (line < offset) {
          line++
-          return
+          return true
        }
        if (lines.length >= limit || bytes >= MAX_READ_BYTES) {
-          truncated = true
-          next ??= line++
-          return
+          next = line
+          return false
        }
-        found = true
        const text = input.length > MAX_LINE_LENGTH ? input.slice(0, MAX_LINE_LENGTH) + MAX_LINE_SUFFIX : input
        const size = Buffer.byteLength(text, "utf-8") + (lines.length > 0 ? 1 : 0)
        if (bytes + size > MAX_READ_BYTES) {
-          truncated = true
-          next ??= line++
-          return
+          next = line
+          return false
        }
        lines.push(text)
        bytes += size
        line++
+        return true
      }
-      const consume = (chunk: Uint8Array) => {
-        if (chunk.includes(0)) throw new BinaryFileError(resource)
-        let text = decoder.decode(chunk, { stream: true })
+      const consume = (input: string) => {
+        let text = input
        while (true) {
          const index = text.indexOf("\n")
          if (index === -1) {
@@ -235,25 +276,44 @@ export const read = Effect.fn("ReadTool.read")(function* (
          pending = ""
          discard = false
          text = text.slice(index + 1)
-          append(current.endsWith("\r") ? current.slice(0, -1) : current)
+          if (!append(current.endsWith("\r") ? current.slice(0, -1) : current)) return false
        }
+        return true
      }
-      yield* Effect.sync(() => consume(first))
-      while (true) {
-        const chunk = yield* file.readAlloc(64 * 1024).pipe(Effect.orDie)
+      const consumeChunk = Effect.fnUntraced(function* (chunk: Uint8Array) {
+        let start = 0
+        while (start < chunk.length) {
+          if (lines.length >= limit || bytes >= MAX_READ_BYTES) {
+            next = line
+            return false
+          }
+          const newline = chunk.indexOf(10, start)
+          const end = newline === -1 ? chunk.length : newline + 1
+          const segment = chunk.subarray(start, end)
+          if (binary(resource, segment)) return yield* Effect.fail(new BinaryFileError({ resource }))
+          if (!consume(yield* decodeUtf8(resource, decoder, segment))) return false
+          start = end
+        }
+        return true
+      })
+      let done = !(yield* consumeChunk(first))
+      while (!done) {
+        const chunk = yield* file.readAlloc(64 * 1024)
        if (Option.isNone(chunk)) break
-        yield* Effect.sync(() => consume(chunk.value))
+        done = !(yield* consumeChunk(chunk.value))
      }
-      const tail = yield* Effect.sync(() => decoder.decode())
-      if (!discard) pending += tail
-      if (pending) append(pending.endsWith("\r") ? pending.slice(0, -1) : pending)
-      if (!found && offset !== 1) return yield* Effect.die(new Error(`Offset ${offset} is out of range`))
+      if (!done) {
+        const tail = yield* decodeUtf8(resource, decoder)
+        if (!discard) pending += tail
+        if (pending) append(pending.endsWith("\r") ? pending.slice(0, -1) : pending)
+      }
+      if (lines.length === 0 && offset !== 1) return yield* Effect.fail(new OffsetOutOfRangeError({ offset }))
      return new TextPage({
        type: "text-page",
        content: lines.join("\n"),
        mime: FSUtil.mimeType(real),
        offset,
-        truncated,
+        truncated: next !== undefined,
        ...(next === undefined ? {} : { next }),
      })
    }),
@@ -261,8 +321,8 @@ export const read = Effect.fn("ReadTool.read")(function* (
 })

 export const list = Effect.fn("ReadTool.list")(function* (fs: FSUtil.Interface, input: string, page: PageInput = {}) {
-  const real = yield* fs.realPath(input).pipe(Effect.orDie)
-  const items = yield* fs.readDirectoryEntries(real).pipe(Effect.orDie)
+  const real = yield* fs.realPath(input)
+  const items = yield* fs.readDirectoryEntries(real)
  const offset = page.offset ?? 1
  const limit = Math.min(page.limit ?? MAX_READ_LINES, MAX_READ_LINES)
  const entries = yield* Effect.forEach(
@@ -57,8 +57,8 @@ export const layer = Layer.effectDiscard(
              const selected = path.isAbsolute(input.path) ? path.dirname(absolute) : location.directory
              if (!path.isAbsolute(input.path) && !FSUtil.contains(location.directory, absolute))
                return yield* Effect.die(new Error("Path escapes the allowed read root"))
-              const real = yield* fs.realPath(absolute).pipe(Effect.orDie)
-              const root = yield* fs.realPath(selected).pipe(Effect.orDie)
+              const real = yield* fs.realPath(absolute)
+              const root = yield* fs.realPath(selected)
              if (!FSUtil.contains(root, real))
                return yield* Effect.die(new Error("Path escapes the allowed read root"))
              const resource = path.relative(root, real).replaceAll("\\", "/") || "."
@@ -83,7 +83,7 @@ export const layer = Layer.effectDiscard(
                  .pipe(Effect.catchTag("Image.ResizerUnavailableError", () => Effect.succeed(content)))
              }
              if ("encoding" in content && content.encoding === "base64")
-                return yield* Effect.fail(new ReadToolFileSystem.BinaryFileError(resource))
+                return yield* Effect.fail(new ReadToolFileSystem.BinaryFileError({ resource }))
              return content
            }).pipe(
              Effect.mapError((error) => {
@@ -1,11 +1,12 @@
 export * as WebFetchTool from "./webfetch"

 import { ToolFailure } from "@opencode-ai/llm"
-import { Duration, Effect, Layer, Schema, Stream } from "effect"
+import { Duration, Effect, Layer, Schema } from "effect"
 import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
 import { Parser } from "htmlparser2"
 import TurndownService from "turndown"
 import { PermissionV2 } from "../permission"
+import { collectBoundedResponseBody } from "./http-body"
 import { Tool } from "./tool"
 import { Tools } from "./tools"

@@ -86,24 +87,11 @@ const execute = (http: HttpClient.HttpClient, url: string, format: Format, userA
  http.execute(request(url, format, userAgent)).pipe(Effect.flatMap(HttpClientResponse.filterStatusOk))

 const collectBody = (response: HttpClientResponse.HttpClientResponse) =>
-  Effect.gen(function* () {
-    const contentLength = response.headers["content-length"]
-    if (contentLength && Number.parseInt(contentLength, 10) > MAX_RESPONSE_BYTES) {
-      return yield* Effect.fail(new Error(`Response too large (exceeds ${MAX_RESPONSE_BYTES} byte limit)`))
-    }
-    const chunks: Uint8Array[] = []
-    let size = 0
-    yield* Stream.runForEach(response.stream, (chunk) =>
-      Effect.gen(function* () {
-        size += chunk.byteLength
-        if (size > MAX_RESPONSE_BYTES)
-          return yield* Effect.fail(new Error(`Response too large (exceeds ${MAX_RESPONSE_BYTES} byte limit)`))
-        chunks.push(chunk)
-        return undefined
-      }),
-    )
-    return Buffer.concat(chunks, size)
-  })
+  collectBoundedResponseBody(
+    response,
+    MAX_RESPONSE_BYTES,
+    () => new Error(`Response too large (exceeds ${MAX_RESPONSE_BYTES} byte limit)`),
+  )

 const mimeFrom = (contentType: string) => contentType.split(";", 1)[0]?.trim().toLowerCase() ?? ""
 const isImageAttachment = (mime: string) =>
@@ -171,12 +159,16 @@ export const layer = Layer.effectDiscard(
                  orElse: () => Effect.fail(new Error("Request timed out")),
                }),
              )
-              const content = convert(new TextDecoder().decode(body), contentType, input.format)
+              const content = new TextDecoder().decode(body)
+              const output = yield* Effect.try({
+                try: () => convert(content, contentType, input.format),
+                catch: (error) => error,
+              })
              return {
                url: input.url,
                contentType,
                format: input.format,
-                output: content,
+                output,
              }
            }).pipe(Effect.mapError(() => new ToolFailure({ message: `Unable to fetch ${input.url}` }))),
        }),
@@ -9,6 +9,7 @@ import { PositiveInt } from "../schema"
 import { PermissionV2 } from "../permission"
 import { Tool } from "./tool"
 import { Tools } from "./tools"
+import { collectBoundedResponseBody } from "./http-body"
 import { checksum } from "../util/encode"

 export const name = "websearch"
@@ -164,10 +165,12 @@ const callMcp = <F extends Schema.Struct.Fields>(
    )
    return yield* Effect.gen(function* () {
      const response = yield* HttpClient.filterStatusOk(http).execute(request)
-      const body = yield* response.text
-      if (Buffer.byteLength(body, "utf8") > MAX_RESPONSE_BYTES)
-        return yield* Effect.fail(new Error(`${tool} response exceeded ${MAX_RESPONSE_BYTES} bytes`))
-      return yield* parseResponse(body)
+      const body = yield* collectBoundedResponseBody(
+        response,
+        MAX_RESPONSE_BYTES,
+        () => new Error(`${tool} response exceeded ${MAX_RESPONSE_BYTES} bytes`),
+      )
+      return yield* parseResponse(body.toString("utf8"))
    }).pipe(
      Effect.timeoutOrElse({
        duration: Duration.seconds(25),
@@ -502,7 +502,7 @@ export type WithParts = {
 }

 const options = {
-  sync: {
+  durable: {
    aggregate: "sessionID",
    version: 1,
  },
@@ -61,9 +61,7 @@ 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))
@@ -122,9 +120,7 @@ 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))
@@ -225,9 +221,7 @@ 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",
-  sync: {
+  durable: {
    version: 1,
    aggregate: "id",
  },
@@ -42,7 +42,7 @@ const SyncMessage = EventV2.define({

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

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

 const SyncTimestamp = EventV2.define({
  type: "test.timestamp",
-  sync: {
+  durable: {
    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.version).toBe(2)
+      expect(event.durable?.version).toBe(2)
    }),
  )

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

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

-  it.effect("commits local operational state inside a new synchronized event transaction", () =>
+  it.effect("commits local operational state inside a new durable 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 synchronized event and projector when the local commit fails", () =>
+  it.effect("rolls back the durable 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 synchronized event")
+      expect(String(exit)).toContain("Local commit hooks require a durable event")
    }),
  )

@@ -290,7 +290,6 @@ 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")
      })
@@ -303,7 +302,7 @@ describe("EventV2", () => {
      const event = yield* events.publish(SyncMessage, { id: "one", text: "hello" })

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

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

-  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", () =>
+  it.effect("inserts durable event rows on publish", () =>
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const { db } = yield* Database.Service
@@ -398,7 +355,7 @@ describe("EventV2", () => {
    }),
  )

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

-  it.effect("replays durable aggregate events after a cursor and tails new events", () =>
+  it.effect("replays durable aggregate events after a sequence 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
-        .aggregateEvents({ aggregateID, after: EventV2.Cursor.make(0) })
+        .durable({ aggregateID, after: 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.cursor, event.event.data])).toEqual([
-        [EventV2.Cursor.make(1), { id: aggregateID, text: "one" }],
-        [EventV2.Cursor.make(2), { 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" }],
      ])
    }),
  )
@@ -442,20 +399,18 @@ describe("EventV2", () => {
      const events = yield* EventV2.Service
      const aggregateID = EventV2.ID.create()
      yield* events.publish(SyncMessage, { id: aggregateID, text: "zero" })
-      const fiber = yield* events
-        .aggregateEvents({ aggregateID })
-        .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
+      const fiber = yield* events.durable({ 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.cursor,
-          (event.event.data as { text: string }).text,
+          event.durable?.seq,
+          (event.data as { text: string }).text,
        ]),
      ).toEqual([
-        [EventV2.Cursor.make(0), "zero"],
-        [EventV2.Cursor.make(1), "one"],
+        [0, "zero"],
+        [1, "one"],
      ])
    }),
  )
@@ -476,17 +431,15 @@ describe("EventV2", () => {
      yield* Effect.gen(function* () {
        const events = yield* EventV2.Service
        const aggregateID = EventV2.ID.create()
-        const fiber = yield* events
-          .aggregateEvents({ aggregateID })
-          .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+        const fiber = yield* events.durable({ 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.cursor, event.event.data])).toEqual([
-          [EventV2.Cursor.make(0), { id: aggregateID, text: "during handoff" }],
+        expect(Array.from(yield* Fiber.join(fiber)).map((event) => [event.durable?.seq, event.data])).toEqual([
+          [0, { id: aggregateID, text: "during handoff" }],
        ])
      }).pipe(Effect.provide(Layer.mergeAll(database, eventLayer)))
    }),
@@ -498,7 +451,7 @@ describe("EventV2", () => {
      const aggregateID = EventV2.ID.create()
      const count = 64
      const fiber = yield* events
-        .aggregateEvents({ aggregateID })
+        .durable({ aggregateID })
        .pipe(Stream.take(count), Stream.runCollect, Effect.forkScoped)
      yield* Effect.yieldNow

@@ -506,11 +459,8 @@ describe("EventV2", () => {
        yield* events.publish(SyncMessage, { 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) },
-        ]),
+      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) }]),
      )
    }),
  )
@@ -519,15 +469,13 @@ describe("EventV2", () => {
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const aggregateID = EventV2.ID.create()
-      const fiber = yield* events
-        .aggregateEvents({ aggregateID })
-        .pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      const fiber = yield* events.durable({ 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.event.type)).toEqual([SyncMessage.type])
+      expect(Array.from(yield* Fiber.join(fiber)).map((event) => event.type)).toEqual([SyncMessage.type])
    }),
  )

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

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

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

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

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

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

-  it.effect("remove clears sync event sequence", () =>
+  it.effect("remove clears durable event sequence", () =>
    Effect.gen(function* () {
      const events = yield* EventV2.Service
      const received = new Array<EventV2.Payload>()
@@ -255,7 +255,9 @@ 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,7 +35,9 @@ 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,7 +157,8 @@ 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)
    }),
  )
@@ -172,7 +173,8 @@ 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,7 +118,9 @@ 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)
    }),
  )

@@ -133,7 +135,8 @@ 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,9 +22,7 @@ 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", () => {
@@ -83,7 +81,9 @@ 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,12 +92,15 @@ 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)
    }),
  )

@@ -111,8 +114,9 @@ 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,7 +44,9 @@ 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([
-        { cursor: 1, event: { type: "session.next.prompt.admitted", data: { prompt: { text: "Hello" } } } },
-        { cursor: 2, event: { type: "session.next.prompt.promoted" } },
+        { durable: { seq: 1 }, type: "session.next.prompt.admitted", data: { prompt: { text: "Hello" } } },
+        { durable: { seq: 2 }, 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([{ event: { type: "session.next.model.switched", data: { model } } }])
+      ).toMatchObject([{ 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.seq })
+      ).toMatchObject({ promoted_seq: event.durable?.seq })
    }),
  )

@@ -334,134 +334,6 @@ 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 cursor", () =>
+  it.effect("streams durable Session events after an aggregate sequence", () =>
    Effect.gen(function* () {
      yield* setup
      const session = yield* SessionV2.Service
@@ -191,17 +191,19 @@ 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.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(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(
        Array.from(
-          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"]])
+          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"]])
    }),
  )

@@ -472,58 +474,6 @@ 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
@@ -14,6 +14,39 @@ const id = (value: string) => SessionMessage.ID.make(`msg_${value}`)
 const model = Model.make({ id: "model", provider: "provider", route: OpenAIChat.route })

 describe("toLLMMessages", () => {
+  test("omits empty assistant turns", () => {
+    const assistant = (value: string, content: SessionMessage.Assistant["content"]) =>
+      new SessionMessage.Assistant({
+        id: id(value),
+        type: "assistant",
+        agent: "build",
+        model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") },
+        content,
+        time: { created, completed: created },
+      })
+    const messages = toLLMMessages(
+      [
+        assistant("empty", []),
+        assistant("empty-text", [new SessionMessage.AssistantText({ type: "text", id: "empty", text: "" })]),
+        assistant("empty-reasoning", [
+          new SessionMessage.AssistantReasoning({ type: "reasoning", id: "empty-reasoning", text: "" }),
+        ]),
+        assistant("text", [new SessionMessage.AssistantText({ type: "text", id: "text", text: "Partial" })]),
+        assistant("reasoning", [
+          new SessionMessage.AssistantReasoning({
+            type: "reasoning",
+            id: "reasoning",
+            text: "",
+            providerMetadata: { anthropic: { signature: "sig_1" } },
+          }),
+        ]),
+      ],
+      model,
+    )
+
+    expect(messages.map((message) => message.id)).toEqual([id("text"), id("reasoning")])
+  })
+
  test("maps every top-level V2 Session message type", () => {
    const file = new FileAttachment({ uri: "data:image/png;base64,aGVsbG8=", mime: "image/png", name: "hello.png" })
    const messages = toLLMMessages(
@@ -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.sync ? EventV2.versionedType(definition.type, definition.sync.version) : definition.type,
+          type: definition.durable
+            ? EventV2.versionedType(definition.type, definition.durable.version)
+            : definition.type,
          data,
        })
        return event
      }),
    subscribe: () => Stream.empty,
    all: () => Stream.empty,
-    aggregateEvents: () => Stream.empty,
-    sync: () => Effect.succeed(Effect.void),
+    durable: () => Stream.empty,
    listen: () => Effect.succeed(Effect.void),
-    beforeCommit: () => Effect.void,
    project: () => Effect.void,
    replay: () => Effect.void,
    replayAll: () => Effect.succeed(undefined),
@@ -547,6 +547,8 @@ const verifyPartialFlushOnInterruption = (kind: FragmentKind) =>
      { type: "user", text: prompt },
      {
        type: "assistant",
+        finish: "error",
+        error: { type: "unknown", message: "Provider turn interrupted" },
        content: [
          kind === "tool input"
            ? { type: "tool", id: fragmentID(kind, "interrupted"), state: { status: "error" } }
@@ -1355,34 +1357,6 @@ 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
@@ -0,0 +1,117 @@
+import { describe, expect } from "bun:test"
+import { NodeFileSystem } from "@effect/platform-node"
+import path from "path"
+import { Effect, FileSystem, Layer } from "effect"
+import { FSUtil } from "@opencode-ai/core/fs-util"
+import { ReadToolFileSystem } from "@opencode-ai/core/tool/read-filesystem"
+import { testEffect } from "./lib/effect"
+
+const it = testEffect(FSUtil.layer.pipe(Layer.provideMerge(NodeFileSystem.layer)))
+const fixture = Effect.gen(function* () {
+  const fs = yield* FSUtil.Service
+  const files = yield* FileSystem.FileSystem
+  const directory = yield* files.makeTempDirectoryScoped()
+  return { fs, files, directory }
+})
+
+describe("ReadToolFileSystem", () => {
+  it.effect("fails with a typed filesystem error when a resolved file disappears", () =>
+    Effect.gen(function* () {
+      const { fs, directory } = yield* fixture
+      const file = path.join(directory, "missing.txt")
+
+      const error = yield* ReadToolFileSystem.read(fs, file, "missing.txt").pipe(Effect.flip)
+
+      expect(error).toMatchObject({ _tag: "PlatformError" })
+    }),
+  )
+
+  it.effect("fails when a file becomes the wrong path kind", () =>
+    Effect.gen(function* () {
+      const { fs, directory } = yield* fixture
+
+      const error = yield* ReadToolFileSystem.read(fs, directory, "folder").pipe(Effect.flip)
+
+      expect(error).toBeInstanceOf(ReadToolFileSystem.PathKindError)
+    }),
+  )
+
+  it.effect("fails with a typed filesystem error when directory listing fails", () =>
+    Effect.gen(function* () {
+      const { fs, files, directory } = yield* fixture
+      const file = path.join(directory, "file.txt")
+      yield* files.writeFileString(file, "hello")
+
+      const error = yield* ReadToolFileSystem.list(fs, file).pipe(Effect.flip)
+
+      expect(error).toBeInstanceOf(FSUtil.FileSystemError)
+      if (error instanceof FSUtil.FileSystemError) expect(error.method).toBe("readDirectoryEntries")
+    }),
+  )
+
+  it.effect("reports binary and malformed UTF-8 content as typed errors", () =>
+    Effect.gen(function* () {
+      const { fs, files, directory } = yield* fixture
+      const binary = path.join(directory, "archive.dat")
+      const malformed = path.join(directory, "malformed.txt")
+      yield* files.writeFile(binary, Uint8Array.of(0, 1, 2, 3))
+      const malformedContent = new Uint8Array(64 * 1024 + 1).fill(97)
+      malformedContent[64 * 1024] = 0x80
+      yield* files.writeFile(malformed, malformedContent)
+
+      const binaryError = yield* ReadToolFileSystem.read(fs, binary, "archive.dat").pipe(Effect.flip)
+      const malformedError = yield* ReadToolFileSystem.read(fs, malformed, "malformed.txt").pipe(Effect.flip)
+
+      expect(binaryError).toBeInstanceOf(ReadToolFileSystem.BinaryFileError)
+      expect(binaryError.message).toBe("Cannot read binary file: archive.dat")
+      expect(malformedError).toBeInstanceOf(ReadToolFileSystem.MalformedUtf8Error)
+    }),
+  )
+
+  it.effect("reports out-of-range pagination as a typed error", () =>
+    Effect.gen(function* () {
+      const { fs, files, directory } = yield* fixture
+      const file = path.join(directory, "short.txt")
+      yield* files.writeFileString(file, "one\n")
+
+      const error = yield* ReadToolFileSystem.read(fs, file, "short.txt", { offset: 2 }).pipe(Effect.flip)
+
+      expect(error).toBeInstanceOf(ReadToolFileSystem.OffsetOutOfRangeError)
+      expect(error.message).toBe("Offset 2 is out of range")
+    }),
+  )
+
+  it.effect("stops reading after the requested page is complete", () =>
+    Effect.gen(function* () {
+      const { fs, files, directory } = yield* fixture
+      const prefix = new TextEncoder().encode("one\n")
+      for (const [name, trailing] of [
+        ["malformed.txt", 0x80],
+        ["nul.txt", 0],
+      ] as const) {
+        const file = path.join(directory, name)
+        yield* files.writeFile(file, Uint8Array.from([...prefix, trailing]))
+
+        const result = yield* ReadToolFileSystem.read(fs, file, name, { limit: 1 })
+
+        expect(result).toMatchObject({ type: "text-page", content: "one", truncated: true, next: 2 })
+      }
+    }),
+  )
+
+  it.effect("preserves the media ingestion limit message", () =>
+    Effect.gen(function* () {
+      const { fs, files, directory } = yield* fixture
+      const file = path.join(directory, "oversized.png")
+      yield* files.writeFile(file, Uint8Array.of(0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a))
+      yield* files.truncate(file, ReadToolFileSystem.MAX_MEDIA_INGEST_BYTES + 1)
+
+      const error = yield* ReadToolFileSystem.read(fs, file, "oversized.png").pipe(Effect.flip)
+
+      expect(error).toBeInstanceOf(ReadToolFileSystem.MediaIngestLimitError)
+      expect(error.message).toBe(
+        `Media exceeds ${ReadToolFileSystem.MAX_MEDIA_INGEST_BYTES} byte ingestion limit: oversized.png`,
+      )
+    }),
+  )
+})
@@ -1,5 +1,5 @@
 import { beforeEach, describe, expect } from "bun:test"
-import { Effect, Exit, Layer } from "effect"
+import { Effect, Exit, Layer, PlatformError } from "effect"
 import { Config } from "@opencode-ai/core/config"
 import { ConfigAttachments } from "@opencode-ai/core/config/attachments"
 import { FileSystem } from "@opencode-ai/core/filesystem"
@@ -18,6 +18,8 @@ import { testEffect } from "./lib/effect"
 import { toolIdentity, executeTool, settleTool, toolDefinitions } from "./lib/tool"

 const assertions: PermissionV2.AssertInput[] = []
+const missingPath = "__missing_read_target__.txt"
+const missingAbsolutePath = `${process.cwd()}/${missingPath}`
 const readCalls: {
  input: AbsolutePath
  page: ReadToolFileSystem.PageInput
@@ -32,7 +34,7 @@ let readResult: FileSystem.Content | ReadToolFileSystem.TextPage = {
  encoding: "utf8",
  mime: "text/plain",
 }
-let readFailure: unknown
+let readFailure: ReadToolFileSystem.ReadError | undefined
 let configEntries: Config.Entry[] = []
 const reader = Layer.succeed(
  ReadToolFileSystem.Service,
@@ -40,7 +42,7 @@ const reader = Layer.succeed(
    inspect: () => (resolveFailure === undefined ? Effect.succeed(resolvedType) : Effect.die(resolveFailure)),
    read: (input, _resource, page = {}) => {
      readCalls.push({ input, page })
-      if (readFailure !== undefined) return Effect.die(readFailure)
+      if (readFailure !== undefined) return Effect.fail(readFailure)
      return Effect.succeed(readResult)
    },
    list: (_path, input = {}) =>
@@ -70,7 +72,24 @@ const config = Layer.succeed(Config.Service, Config.Service.of({ entries: () =>
 const image = Image.layer.pipe(Layer.provide(config))
 const testFileSystem = Layer.effect(
  FSUtil.Service,
-  FSUtil.Service.use((fs) => Effect.succeed(FSUtil.Service.of({ ...fs, realPath: (path) => Effect.succeed(path) }))),
+  FSUtil.Service.use((fs) =>
+    Effect.succeed(
+      FSUtil.Service.of({
+        ...fs,
+        realPath: (path) =>
+          path === missingAbsolutePath
+            ? Effect.fail(
+                PlatformError.systemError({
+                  _tag: "NotFound",
+                  module: "FileSystem",
+                  method: "realPath",
+                  pathOrDescriptor: path,
+                }),
+              )
+            : Effect.succeed(path),
+      }),
+    ),
+  ),
 ).pipe(Layer.provide(FSUtil.defaultLayer))
 const infrastructure = Layer.mergeAll(
  testFileSystem,
@@ -412,9 +431,32 @@ describe("ReadTool", () => {
    }),
  )

+  it.effect("returns expected filesystem failures to the model", () =>
+    Effect.gen(function* () {
+      readFailure = new ReadToolFileSystem.BinaryFileError({ resource: "archive.dat" })
+      const registry = yield* ToolRegistry.Service
+
+      expect(
+        yield* executeTool(registry, {
+          sessionID,
+          ...toolIdentity,
+          call: {
+            type: "tool-call",
+            id: "call-binary",
+            name: "read",
+            input: { path: "archive.dat", offset: 2, limit: 1 },
+          },
+        }),
+      ).toEqual({ type: "error", value: "Cannot read binary file: archive.dat" })
+      expect(readCalls).toEqual([
+        { input: AbsolutePath.make(`${process.cwd()}/archive.dat`), page: { offset: 2, limit: 1 } },
+      ])
+    }),
+  )
+
  it.effect("preserves unexpected filesystem defects", () =>
    Effect.gen(function* () {
-      readFailure = new ReadToolFileSystem.BinaryFileError("archive.dat")
+      resolveFailure = new Error("unexpected")
      const registry = yield* ToolRegistry.Service

      expect(
@@ -422,18 +464,10 @@ describe("ReadTool", () => {
          yield* executeTool(registry, {
            sessionID,
            ...toolIdentity,
-            call: {
-              type: "tool-call",
-              id: "call-binary",
-              name: "read",
-              input: { path: "archive.dat", offset: 2, limit: 1 },
-            },
+            call: { type: "tool-call", id: "call-defect", name: "read", input: { path: "README.md" } },
          }).pipe(Effect.exit),
        ),
      ).toBe(true)
-      expect(readCalls).toEqual([
-        { input: AbsolutePath.make(`${process.cwd()}/archive.dat`), page: { offset: 2, limit: 1 } },
-      ])
    }),
  )

@@ -453,6 +487,22 @@ describe("ReadTool", () => {
    }),
  )

+  it.effect("returns missing paths as model-visible tool failures", () =>
+    Effect.gen(function* () {
+      const registry = yield* ToolRegistry.Service
+
+      expect(
+        yield* executeTool(registry, {
+          sessionID,
+          ...toolIdentity,
+          call: { type: "tool-call", id: "call-missing-path", name: "read", input: { path: missingPath } },
+        }),
+      ).toEqual({ type: "error", value: `Unable to read ${missingPath}` })
+      expect(assertions).toEqual([])
+      expect(readCalls).toEqual([])
+    }),
+  )
+
  it.effect("lists a bounded directory page through read", () =>
    Effect.gen(function* () {
      resolvedType = "directory"
@@ -176,6 +176,25 @@ describe("WebFetchTool registration", () => {
    }),
  )

+  it.effect("returns an error result when HTML-to-Markdown conversion throws", () =>
+    Effect.gen(function* () {
+      reset()
+      respond = () =>
+        Effect.succeed(
+          new Response("<div>".repeat(10_000) + "content" + "</div>".repeat(10_000), {
+            headers: { "content-type": "text/html" },
+          }),
+        )
+      const registry = yield* ToolRegistry.Service
+      const url = "https://1.1.1.1/deep-html"
+
+      expect(yield* executeTool(registry, call({ url, format: "markdown" }))).toEqual({
+        type: "error",
+        value: `Unable to fetch ${url}`,
+      })
+    }),
+  )
+
  it.effect("rejects declared and streamed oversized bodies", () =>
    Effect.gen(function* () {
      reset()
@@ -1,4 +1,4 @@
-import { describe, expect, test } from "bun:test"
+import { beforeEach, describe, expect, test } from "bun:test"
 import { Effect, Layer, Schema } from "effect"
 import { HttpClient, HttpClientResponse } from "effect/unstable/http"
 import { PermissionV2 } from "@opencode-ai/core/permission"
@@ -66,8 +66,14 @@ interface Request {
 const requests: Request[] = []
 const assertions: PermissionV2.AssertInput[] = []
 let responseBody = payload("search results")
+let makeResponse = () => new Response(responseBody, { status: 200 })
 let config: WebSearchTool.Config = { enableExa: false, enableParallel: false }

+beforeEach(() => {
+  responseBody = payload("search results")
+  makeResponse = () => new Response(responseBody, { status: 200 })
+})
+
 const http = Layer.succeed(
  HttpClient.HttpClient,
  HttpClient.make((request) =>
@@ -78,7 +84,7 @@ const http = Layer.succeed(
        headers: request.headers,
        body: JSON.parse(new TextDecoder().decode(request.body.body)),
      })
-      return HttpClientResponse.fromWeb(request, new Response(responseBody, { status: 200 }))
+      return HttpClientResponse.fromWeb(request, makeResponse())
    }),
  ),
 )
@@ -270,7 +276,22 @@ describe("WebSearchTool registration", () => {
    Effect.gen(function* () {
      requests.length = 0
      assertions.length = 0
-      responseBody = "x".repeat(WebSearchTool.MAX_RESPONSE_BYTES + 1)
+      let chunksRead = 0
+      let cancelled = false
+      makeResponse = () =>
+        new Response(
+          new ReadableStream({
+            pull(controller) {
+              chunksRead++
+              if (chunksRead === 10) throw new Error("response was not stopped at the byte limit")
+              controller.enqueue(new Uint8Array(64 * 1024))
+            },
+            cancel() {
+              cancelled = true
+            },
+          }),
+          { status: 200 },
+        )
      config = { provider: "exa", enableExa: false, enableParallel: false }
      const registry = yield* ToolRegistry.Service

@@ -281,6 +302,8 @@ describe("WebSearchTool registration", () => {
          call: { type: "tool-call", id: "call-large-response", name: "websearch", input: { query: "too much" } },
        }),
      ).toEqual({ type: "error", value: "Unable to search the web for too much" })
+      expect(chunksRead).toBeLessThan(10)
+      expect(cancelled).toBe(true)
    }),
  )
 })
@@ -23,10 +23,7 @@ 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 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]
+        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]
        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.version),
-              seq: event.seq,
+              type: EventV2.versionedType(event.type, event.durable.version),
+              seq: event.durable.seq,
              aggregateID,
              data: event.data,
            },
@@ -16,13 +16,13 @@ const GlobalHealth = Schema.Struct({
 const SyncEventSchemas = EventV2.registry
  .values()
  .flatMap((definition) => {
-    if (!definition.sync) return []
+    if (!definition.durable) return []
    return [
      Schema.Struct({
        type: Schema.Literal("sync"),
        id: EventV2.ID,
        syncEvent: Schema.Struct({
-          type: Schema.Literal(EventV2.versionedType(definition.type, definition.sync.version)),
+          type: Schema.Literal(EventV2.versionedType(definition.type, definition.durable.version)),
          id: EventV2.ID,
          seq: Schema.Finite,
          aggregateID: Schema.String,
@@ -6,7 +6,6 @@ export type ClientOptions = {

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

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

+export type EventPluginAdded = {
+  id: string
+  type: "plugin.added"
+  properties: {
+    id: string
+  }
+}
+
 export type EventPermissionV2Asked = {
  id: string
  type: "permission.v2.asked"
@@ -14481,9 +14481,6 @@
          {
            "$ref": "#/components/schemas/EventModels-devRefreshed"
          },
-          {
-            "$ref": "#/components/schemas/EventPluginAdded"
-          },
          {
            "$ref": "#/components/schemas/EventIntegrationUpdated"
          },
@@ -14622,6 +14619,9 @@
          {
            "$ref": "#/components/schemas/EventFileEdited"
          },
+          {
+            "$ref": "#/components/schemas/EventPluginAdded"
+          },
          {
            "$ref": "#/components/schemas/EventPermissionV2Asked"
          },
@@ -16645,31 +16645,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": {
@@ -18453,6 +18428,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": {
@@ -26900,27 +26900,7 @@
            "type": "object",
            "properties": {
              "released": {
-                "anyOf": [
-                  {
-                    "type": "number"
-                  },
-                  {
-                    "type": "string",
-                    "enum": ["NaN"]
-                  },
-                  {
-                    "type": "string",
-                    "enum": ["Infinity"]
-                  },
-                  {
-                    "type": "string",
-                    "enum": ["-Infinity"]
-                  },
-                  {
-                    "type": "string",
-                    "enum": ["Infinity", "-Infinity", "NaN"]
-                  }
-                ]
+                "type": "number"
              }
            },
            "required": ["released"],
@@ -27643,31 +27623,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
-      },
      "EventIntegrationUpdated": {
        "type": "object",
        "properties": {
@@ -29443,6 +29398,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
+      },
      "EventPermissionV2Asked": {
        "type": "object",
        "properties": {
@@ -14,7 +14,6 @@ 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"
@@ -27,6 +26,7 @@ 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 runtime.runPromise(getStatsModelData(model, lab))
+  return runStatsEffect(getStatsModelData(model, lab))
 }, "getStatsModelData")

 export default function StatsModel() {
@@ -6,7 +6,6 @@ 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"
@@ -17,6 +16,7 @@ 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 runtime.runPromise(getStatsLabData(lab))
+  return runStatsEffect(getStatsLabData(lab))
 }, "getStatsLabData")

 export default function StatsLab() {
@@ -1,19 +1,3 @@
-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(
-    await runtime.runPromise(
-      Effect.gen(function* () {
-        const config = yield* AppConfig
-        return {
-          ok: true,
-          app: "stats",
-          stage: config.stage,
-          publicUrl: config.publicUrl,
-        }
-      }),
-    ),
-  )
+  return Response.json({ ok: true, app: "stats" })
 }
@@ -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 runtime.runPromise(getStatsHomeData())
+  return runStatsEffect(getStatsHomeData())
 }, "getStatsHomeData")

 export default function StatsHome() {
@@ -0,0 +1,5 @@
+import { Effect } from "effect"
+
+export function runStatsEffect<A, E>(effect: Effect.Effect<A, E>) {
+  return Effect.runPromise(effect)
+}
@@ -17,6 +17,8 @@
  },
  "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",
@@ -5,28 +5,36 @@ import {
  StartQueryExecutionCommand,
  type Row,
 } from "@aws-sdk/client-athena"
-import { Effect, Layer, Schema } from "effect"
+import { Effect, Layer } from "effect"
 import * as Context from "effect/Context"
 import { Resource } from "sst/resource"

-const ATHENA_MAX_POLL_ATTEMPTS = 60
+const ATHENA_MAX_POLL_ATTEMPTS = 300
 const ATHENA_PAGE_SIZE = 1000

 export type AthenaData = Record<string, string>

-export class AthenaQueryError extends Schema.TaggedErrorClass<AthenaQueryError>()("AthenaQueryError", {
-  message: Schema.String,
-  queryExecutionId: Schema.optional(Schema.String),
-  cause: Schema.optional(Schema.Defect()),
-}) {}
+export class AthenaQueryError extends Error {
+  readonly _tag = "AthenaQueryError"
+  readonly queryExecutionId?: string

-export class AthenaQueryTimeoutError extends Schema.TaggedErrorClass<AthenaQueryTimeoutError>()(
-  "AthenaQueryTimeoutError",
-  {
-    message: Schema.String,
-    queryExecutionId: Schema.String,
-  },
-) {}
+  constructor(input: { message: string; queryExecutionId?: string; cause?: unknown }) {
+    super(input.message, { cause: input.cause })
+    this.name = "AthenaQueryError"
+    this.queryExecutionId = input.queryExecutionId
+  }
+}
+
+export class AthenaQueryTimeoutError extends Error {
+  readonly _tag = "AthenaQueryTimeoutError"
+  readonly queryExecutionId: string
+
+  constructor(input: { message: string; queryExecutionId: string }) {
+    super(input.message)
+    this.name = "AthenaQueryTimeoutError"
+    this.queryExecutionId = input.queryExecutionId
+  }
+}

 export declare namespace Athena {
  export interface Service {
@@ -57,7 +65,7 @@ export class Athena extends Context.Service<Athena, Athena.Service>()("@opencode
        })
        const queryExecutionId = started.QueryExecutionId
        if (!queryExecutionId)
-          return yield* new AthenaQueryError({ message: "Athena did not return a query execution id" })
+          return yield* Effect.fail(new AthenaQueryError({ message: "Athena did not return a query execution id" }))

        yield* poll(client, queryExecutionId)
        return yield* results(client, queryExecutionId)
@@ -87,16 +95,20 @@ const poll: (

  if (status?.State === "SUCCEEDED") return
  if (status?.State === "FAILED" || status?.State === "CANCELLED")
-    return yield* new AthenaQueryError({
-      message: `Athena stats query ${status.State.toLowerCase()}: ${status.StateChangeReason ?? "unknown reason"}`,
-      queryExecutionId,
-    })
+    return yield* Effect.fail(
+      new AthenaQueryError({
+        message: `Athena stats query ${status.State.toLowerCase()}: ${status.StateChangeReason ?? "unknown reason"}`,
+        queryExecutionId,
+      }),
+    )

  if (attempt >= ATHENA_MAX_POLL_ATTEMPTS - 1)
-    return yield* new AthenaQueryTimeoutError({
-      message: `Athena stats query ${queryExecutionId} did not complete`,
-      queryExecutionId,
-    })
+    return yield* Effect.fail(
+      new AthenaQueryTimeoutError({
+        message: `Athena stats query ${queryExecutionId} did not complete`,
+        queryExecutionId,
+      }),
+    )

  return yield* poll(client, queryExecutionId, attempt + 1)
 })
@@ -44,16 +44,29 @@ export class DrizzleClient extends Context.Service<DrizzleClient, Drizzle>()("@o
  )
 }

-export class DatabaseError extends Schema.TaggedErrorClass<DatabaseError>()("DatabaseError", {
-  cause: Schema.Defect(),
-}) {}
+export class DatabaseError extends Error {
+  readonly _tag = "DatabaseError"
+
+  constructor(input: { cause: unknown }) {
+    super("Database operation failed", { cause: input.cause })
+    this.name = "DatabaseError"
+  }
+
+  static make(input: { cause: unknown }) {
+    return new DatabaseError(input)
+  }
+}

 export const catchDbError = Effect.mapError((cause) => DatabaseError.make({ cause }))

-export class MigrationError extends Schema.TaggedErrorClass<MigrationError>()("MigrationError", {
-  message: Schema.String,
-  cause: Schema.optional(Schema.Defect()),
-}) {}
+export class MigrationError extends Error {
+  readonly _tag = "MigrationError"
+
+  constructor(input: { message: string; cause?: unknown }) {
+    super(input.message, { cause: input.cause })
+    this.name = "MigrationError"
+  }
+}

 export const migrate = Effect.fn("Database.migrate")(function* () {
  const settings = yield* DatabaseConfig
@@ -68,9 +81,11 @@ export const migrate = Effect.fn("Database.migrate")(function* () {
    catch: (cause) => new MigrationError({ message: "Failed to apply database migrations", cause }),
  })
  if (result)
-    return yield* new MigrationError({
-      message: `Failed to initialize database migrations: ${result.exitCode}`,
-    })
+    return yield* Effect.fail(
+      new MigrationError({
+        message: `Failed to initialize database migrations: ${result.exitCode}`,
+      }),
+    )
  yield* Effect.logInfo("database migrations complete").pipe(
    Effect.annotateLogs({ migrationsDir: settings.migrationsDir }),
  )
@@ -1,8 +1,9 @@
+import { Client } from "@planetscale/database"
 import { Effect } from "effect"
-import { DatabaseError } from "../database"
-import { GeoStatRepo, type GeoStatMetric } from "./geo"
-import { ModelStatRepo, type ModelStatMetric } from "./model"
-import { ProviderStatRepo, type ProviderStatMetric } from "./provider"
+import { Resource } from "sst/resource"
+import type { GeoStatMetric } from "./geo"
+import type { ModelStatMetric } from "./model"
+import type { ProviderStatMetric } from "./provider"

 export type UsageProduct = "All Users" | "Zen" | "Go" | "Enterprise"
 export type TokenProduct = "Zen" | "Go" | "Enterprise"
@@ -91,6 +92,14 @@ 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
@@ -130,49 +139,136 @@ type ModelAggregate = {
  totalCostMicrocents: number
 }

-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)
-})
+type RawRow = Record<string, unknown>

-export const getStatsModelData: (
+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(
  model: string,
  provider?: string,
-) => 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,
-    )
-  },
-)
-
-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())
+): 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,
+    )
+  ).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))
+}

 function buildStatsHomeData(
  modelRows: ModelStatMetric[],
@@ -17,6 +17,8 @@ export type StatDimension = "model" | "provider" | "geo" | "geo_model"
 export function buildStatsQuery(periodStart: Date, periodEnd: Date, dimension: StatDimension) {
  const periodStartValue = sqlString(periodStart.toISOString())
  const periodEndValue = sqlString(periodEnd.toISOString())
+  const periodStartDateValue = sqlString(periodStart.toISOString().slice(0, 10))
+  const periodEndDateValue = sqlString(periodEnd.toISOString().slice(0, 10))
  const sourceTable = [Resource.InferenceEvent.catalog, Resource.InferenceEvent.database, Resource.InferenceEvent.table]
    .map(sqlIdentifier)
    .join(".")
@@ -95,6 +97,9 @@ WITH normalized AS (
  WHERE event_type = 'completions'
    AND model IS NOT NULL
    AND model <> ''
+    AND source = 'lite'
+    AND event_date >= ${periodStartDateValue}
+    AND event_date <= ${periodEndDateValue}
    AND event_timestamp >= ${periodStartValue}
    AND event_timestamp < ${periodEndValue}
 ), filtered AS (
@@ -0,0 +1,58 @@
+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}`)
+}
@@ -1,6 +1,6 @@
 import { Buffer } from "node:buffer"
 import { FirehoseClient, PutRecordBatchCommand } from "@aws-sdk/client-firehose"
-import { Effect, Layer, Schema } from "effect"
+import { Effect, Layer } from "effect"
 import * as Context from "effect/Context"
 import { Resource } from "sst/resource"

@@ -12,11 +12,16 @@ type IngestEvent = Record<string, unknown>
 type LakeRoute = { database: string; table: string }
 type FirehoseRecord = { Data: Uint8Array }

-export class IngestError extends Schema.TaggedErrorClass<IngestError>()("IngestError", {
-  message: Schema.String,
-  failed: Schema.Number,
-  cause: Schema.optional(Schema.Defect()),
-}) {}
+export class IngestError extends Error {
+  readonly _tag = "IngestError"
+  readonly failed: number
+
+  constructor(input: { message: string; failed: number; cause?: unknown }) {
+    super(input.message, { cause: input.cause })
+    this.name = "IngestError"
+    this.failed = input.failed
+  }
+}

 export declare namespace Ingest {
  export interface Service {
@@ -37,10 +42,12 @@ export class Ingest extends Context.Service<Ingest, Ingest.Service>()("@opencode
          yield* Effect.logWarning(
            `lake ingest rejected ${JSON.stringify({ records: counts.records, unsupported: counts.unsupported })}`,
          )
-          return yield* new IngestError({
-            message: "Unsupported lake event type",
-            failed: counts.unsupported,
-          })
+          return yield* Effect.fail(
+            new IngestError({
+              message: "Unsupported lake event type",
+              failed: counts.unsupported,
+            }),
+          )
        }
        if (counts.records === 0) return { records: 0 }

@@ -66,7 +73,7 @@ export class Ingest extends Context.Service<Ingest, Ingest.Service>()("@opencode

        if (failed > 0) {
          yield* Effect.logWarning(`lake ingest incomplete ${JSON.stringify({ records: counts.records, failed })}`)
-          return yield* new IngestError({ message: "Failed to ingest all lake records", failed })
+          return yield* Effect.fail(new IngestError({ message: "Failed to ingest all lake records", failed }))
        }

        yield* Effect.logInfo(`lake ingest complete ${JSON.stringify({ records: counts.records, batches })}`)
Author	SHA1	Message	Date
Dax Raad	34c4eb5df5	fix(cli): increment ports from default	2026-06-21 19:19:56 -04:00
Kit Langton	49593c1ec4	fix(core): settle interrupted assistant steps (#33266 )	2026-06-21 20:22:09 +00:00
Kit Langton	ff837fe949	fix(core): handle read file failures (#33260 )	2026-06-21 20:12:42 +00:00
Adam	4c6750d464	fix(stats): unblock stats sync	2026-06-21 14:53:46 -05:00
Kit Langton	69f1ec22e3	fix(core): bound web tool failures (#33259 )	2026-06-21 15:37:25 -04:00
Kit Langton	823d327401	fix(core): handle missing read paths (#33255 )	2026-06-21 13:50:14 -04:00
opencode-agent[bot]	82d9cab48d	chore: generate	2026-06-21 14:36:19 +00:00
Dax	fb43c15f88	refactor(core): simplify event model (#33238 )	2026-06-21 10:34:57 -04:00
opencode-agent[bot]	ca006a2d20	chore: update nix node_modules hashes	2026-06-21 12:21:25 +00:00
Dax Raad	8396395f17	fix(stats): update defect schemas	2026-06-21 08:08:31 -04:00
Dax Raad	1b8bab3e35	fix(ci): avoid Playwright Chromium install hang	2026-06-21 08:08:31 -04:00
opencode-agent[bot]	02687b6324	chore: generate	2026-06-21 12:07:16 +00:00
Dax	c780d7cee7	feat(plugin): add v2 effect host (#33111 )	2026-06-21 08:05:49 -04:00
opencode-agent[bot]	7a9337da8a	chore: generate	2026-06-21 11:49:50 +00:00
Adam	f96e6aa6ed	fix(stats): bypass worker runtime crash	2026-06-21 06:48:25 -05:00
Adam	c6f719e153	fix(stats): restore worker data exports	2026-06-21 06:39:49 -05:00
Adam	1a111be494	fix(stats): run worker effects directly	2026-06-21 06:30:15 -05:00
Adam	a0aee82be9	fix(stats): inline worker runtime import	2026-06-21 06:22:26 -05:00
Adam	c0dc6e50a7	fix(stats): restore worker runtime	2026-06-21 06:16:02 -05:00
Adam	418a9e4e66	fix(stats): restore model page rendering	2026-06-21 05:58:39 -05:00
Adam	bd8ce5e6a9	chore(stats): remove deploy migrations	2026-06-21 05:41:10 -05:00
Adam	a97c6de9db	fix(stats): support planetscale cli variants	2026-06-21 05:17:22 -05:00
opencode-agent[bot]	24ea4dd008	chore: generate	2026-06-21 10:03:16 +00:00
Adam	ffcb7542e1	fix(stats): run production migration safely	2026-06-21 05:01:43 -05:00
opencode-agent[bot]	d4d841bafd	chore: generate	2026-06-21 09:40:15 +00:00
Adam	6f0e934573	fix(stats): make unique users migration idempotent	2026-06-21 04:38:47 -05:00
Adam	233d065dd5	feat(stats): show model users metric	2026-06-21 04:30:25 -05:00