feat(api): add finite durable session history pages (#34097)

packages/core/src/event.ts

@@ -3,7 +3,7 @@ export * as EventV2 from "./event"
 import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
-import { and, asc, eq, gt } from "drizzle-orm"
+import { and, asc, eq, gt, inArray } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
 import { Location } from "./location"
@@ -47,6 +47,66 @@ export class InvalidDurableEventError extends Schema.TaggedErrorClass<InvalidDur
   },
 ) {}
 
+const decodeSerializedEvent = (event: SerializedEvent): Payload => {
+  const definition = Durable.get(event.type)
+  if (!definition?.durable) {
+    throw new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` })
+  }
+  return {
+    id: event.id,
+    type: definition.type,
+    durable: { aggregateID: event.aggregateID, seq: event.seq, version: definition.durable.version },
+    data: Schema.decodeUnknownSync(definition.data)(event.data),
+  }
+}
+
+export const readAggregate = Effect.fn("EventV2.readAggregate")(function* <A>(
+  db: Database.Interface["db"],
+  input: {
+    readonly aggregateID: string
+    readonly after?: number
+    readonly limit: number
+    readonly manifest: {
+      readonly definitions: ReadonlyMap<string, Definition>
+      readonly schema: Schema.Decoder<A, never>
+    }
+  },
+) {
+  const after = input.after ?? -1
+  const rows = yield* db
+    .select()
+    .from(EventTable)
+    .where(
+      and(
+        eq(EventTable.aggregate_id, input.aggregateID),
+        gt(EventTable.seq, after),
+        inArray(EventTable.type, Array.from(input.manifest.definitions.keys())),
+      ),
+    )
+    .orderBy(asc(EventTable.seq))
+    .limit(input.limit + 1)
+    .all()
+    .pipe(Effect.orDie)
+  const page = rows.slice(0, input.limit)
+  const decode = Schema.decodeUnknownSync(input.manifest.schema)
+  const events = page.map((event) =>
+    decode({
+      id: event.id,
+      type: input.manifest.definitions.get(event.type)?.type ?? event.type,
+      durable: {
+        aggregateID: event.aggregate_id,
+        seq: event.seq,
+        version: input.manifest.definitions.get(event.type)?.durable?.version,
+      },
+      data: event.data,
+    }),
+  )
+  return {
+    events,
+    hasMore: rows.length > input.limit,
+  }
+})
+
 export const define = Event.define
 export const versionedType = Event.versionedType
 
@@ -459,19 +519,6 @@ export const layerWith = (options?: LayerOptions) =>
 
       const streamAll = (): Stream.Stream<Payload> => Stream.fromPubSub(pubsub.all)
 
-      const decodeSerializedEvent = (event: SerializedEvent) => {
-        const definition = Durable.get(event.type)
-        if (!definition?.durable) {
-          throw new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` })
-        }
-        return {
-          id: event.id,
-          type: definition.type,
-          durable: { aggregateID: event.aggregateID, seq: event.seq, version: definition.durable.version },
-          data: Schema.decodeUnknownSync(definition.data)(event.data),
-        }
-      }
-
       const readAfter = (aggregateID: string, after: number) =>
         (options?.beforeAggregateRead?.(aggregateID) ?? Effect.void).pipe(
           Effect.andThen(

packages/core/src/session.ts

@@ -34,6 +34,7 @@ import { Snapshot } from "./snapshot"
 import { SessionRevert } from "./session/revert"
 import { Revert } from "@opencode-ai/schema/revert"
 import { FSUtil } from "./fs-util"
+import { SessionDurable } from "@opencode-ai/schema/durable-event-manifest"
 
 export const RevertState = Revert.State
 export type RevertState = Revert.State
@@ -131,6 +132,14 @@ export interface Interface {
     sessionID: SessionSchema.ID
     after?: number
   }) => Stream.Stream<SessionEvent.DurableEvent, NotFoundError>
+  readonly history: (input: {
+    sessionID: SessionSchema.ID
+    after?: number
+    limit: number
+  }) => Effect.Effect<
+    { events: ReadonlyArray<SessionEvent.DurableEvent>; hasMore: boolean },
+    NotFoundError
+  >
   readonly switchAgent: (input: { sessionID: SessionSchema.ID; agent: string }) => Effect.Effect<void, NotFoundError>
   readonly switchModel: (input: {
     sessionID: SessionSchema.ID
@@ -347,6 +356,14 @@ export const layer = Layer.unwrap(
                   .get(input.sessionID)
                   .pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))),
               ).pipe(Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event))),
+            history: Effect.fn("V2Session.history")(function* (input) {
+              yield* result.get(input.sessionID)
+              return yield* EventV2.readAggregate(db, {
+                ...input,
+                aggregateID: input.sessionID,
+                manifest: SessionDurable,
+              })
+            }),
             prompt: Effect.fn("V2Session.prompt")((input) =>
               Effect.uninterruptible(
                 Effect.gen(function* () {

packages/core/test/session-history.test.ts

@@ -0,0 +1,174 @@
+import { describe, expect } from "bun:test"
+import { Effect, Layer, Schema } from "effect"
+import { Database } from "@opencode-ai/core/database/database"
+import { EventV2 } from "@opencode-ai/core/event"
+import { Location } from "@opencode-ai/core/location"
+import { LocationServiceMap } from "@opencode-ai/core/location-layer"
+import { ProjectV2 } from "@opencode-ai/core/project"
+import { ProjectTable } from "@opencode-ai/core/project/sql"
+import { AbsolutePath } from "@opencode-ai/core/schema"
+import { SessionV2 } from "@opencode-ai/core/session"
+import { SessionExecution } from "@opencode-ai/core/session/execution"
+import { SessionProjector } from "@opencode-ai/core/session/projector"
+import { SessionStore } from "@opencode-ai/core/session/store"
+import { SessionTable } from "@opencode-ai/core/session/sql"
+import { testEffect } from "./lib/effect"
+
+const projects = Layer.succeed(
+  ProjectV2.Service,
+  ProjectV2.Service.of({
+    resolve: (directory) => Effect.succeed({ id: ProjectV2.ID.global, directory }),
+    directories: () => Effect.succeed([]),
+    commit: () => Effect.void,
+  }),
+)
+const sessions = SessionV2.layer.pipe(
+  Layer.provide(LocationServiceMap.layer),
+  Layer.provide(EventV2.defaultLayer),
+  Layer.provide(Database.defaultLayer),
+  Layer.provide(SessionStore.defaultLayer),
+  Layer.provide(projects),
+  Layer.provide(SessionExecution.noopLayer),
+)
+const it = testEffect(
+  Layer.mergeAll(
+    Database.defaultLayer,
+    EventV2.defaultLayer,
+    projects,
+    SessionProjector.defaultLayer,
+    SessionStore.defaultLayer,
+    SessionExecution.noopLayer,
+    sessions,
+  ),
+)
+const location = Location.Ref.make({ directory: AbsolutePath.make("/project") })
+
+const GapEvent = EventV2.define({
+  type: "test.session.history.gap",
+  durable: { aggregate: "sessionID", version: 1 },
+  schema: { sessionID: SessionV2.ID, value: Schema.String },
+})
+
+describe("SessionV2.history", () => {
+  it.effect("returns an exhausted page for a migrated Session with no event sequence", () =>
+    Effect.gen(function* () {
+      const db = (yield* Database.Service).db
+      const session = yield* SessionV2.Service
+      const sessionID = SessionV2.ID.make("ses_empty_history")
+      yield* db
+        .insert(ProjectTable)
+        .values({ id: ProjectV2.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
+        .onConflictDoNothing()
+        .run()
+      yield* db
+        .insert(SessionTable)
+        .values({
+          id: sessionID,
+          project_id: ProjectV2.ID.global,
+          slug: "empty-history",
+          directory: "/project",
+          title: "Empty history",
+          version: "test",
+        })
+        .run()
+
+      const first = yield* session.history({ sessionID, limit: 10 })
+
+      expect(first).toEqual({ events: [], hasMore: false })
+    }),
+  )
+
+  it.effect("treats after as an exclusive aggregate sequence", () =>
+    Effect.gen(function* () {
+      const session = yield* SessionV2.Service
+      const created = yield* session.create({ location })
+      yield* session.switchAgent({ sessionID: created.id, agent: "one" })
+      yield* session.switchAgent({ sessionID: created.id, agent: "two" })
+
+      const page = yield* session.history({ sessionID: created.id, after: 1, limit: 10 })
+
+      expect(page.events.map((event) => event.durable?.seq)).toEqual([2])
+      expect(page.hasMore).toBe(false)
+    }),
+  )
+
+  it.effect("paginates public events in aggregate order across filtered gaps without duplicates", () =>
+    Effect.gen(function* () {
+      const session = yield* SessionV2.Service
+      const events = yield* EventV2.Service
+      const created = yield* session.create({ location })
+      yield* session.switchAgent({ sessionID: created.id, agent: "one" })
+      yield* events.publish(GapEvent, { sessionID: created.id, value: "filtered" })
+      yield* session.switchAgent({ sessionID: created.id, agent: "two" })
+      yield* session.switchAgent({ sessionID: created.id, agent: "three" })
+
+      const first = yield* session.history({ sessionID: created.id, limit: 2 })
+      const after = first.events.at(-1)?.durable?.seq
+      const second = yield* session.history({
+        sessionID: created.id,
+        after,
+        limit: 2,
+      })
+      const sequence = [...first.events, ...second.events].map((event) => event.durable?.seq)
+
+      expect(first.hasMore).toBe(true)
+      expect(second.hasMore).toBe(false)
+      expect(sequence).toEqual([1, 3, 4])
+      expect(new Set(sequence).size).toBe(sequence.length)
+    }),
+  )
+
+  it.effect("includes events committed between pages", () =>
+    Effect.gen(function* () {
+      const session = yield* SessionV2.Service
+      const created = yield* session.create({ location })
+      yield* session.switchAgent({ sessionID: created.id, agent: "one" })
+      yield* session.switchAgent({ sessionID: created.id, agent: "two" })
+
+      const first = yield* session.history({ sessionID: created.id, limit: 1 })
+      yield* session.switchAgent({ sessionID: created.id, agent: "later" })
+      const second = yield* session.history({
+        sessionID: created.id,
+        after: first.events.at(-1)?.durable?.seq,
+        limit: 10,
+      })
+
+      expect(first.hasMore).toBe(true)
+      expect([...first.events, ...second.events].map((event) => event.durable?.seq)).toEqual([1, 2, 3])
+      expect(second.hasMore).toBe(false)
+    }),
+  )
+
+  it.effect("reports exhaustion for exact-limit and limit-plus-one pages", () =>
+    Effect.gen(function* () {
+      const session = yield* SessionV2.Service
+      const created = yield* session.create({ location })
+      yield* session.switchAgent({ sessionID: created.id, agent: "one" })
+      yield* session.switchAgent({ sessionID: created.id, agent: "two" })
+
+      const exact = yield* session.history({ sessionID: created.id, limit: 2 })
+      const oneMore = yield* session.history({ sessionID: created.id, limit: 1 })
+      const exhausted = yield* session.history({
+        sessionID: created.id,
+        after: oneMore.events.at(-1)?.durable?.seq,
+        limit: 1,
+      })
+
+      expect(exact.events).toHaveLength(2)
+      expect(exact.hasMore).toBe(false)
+      expect(oneMore.events).toHaveLength(1)
+      expect(oneMore.hasMore).toBe(true)
+      expect(exhausted.events).toHaveLength(1)
+      expect(exhausted.hasMore).toBe(false)
+    }),
+  )
+
+  it.effect("fails with NotFoundError for a missing Session", () =>
+    Effect.gen(function* () {
+      const session = yield* SessionV2.Service
+      const error = yield* session.history({ sessionID: SessionV2.ID.make("ses_missing"), limit: 10 }).pipe(Effect.flip)
+
+      expect(error._tag).toBe("Session.NotFoundError")
+    }),
+  )
+})