perf(server): paginate session history (#17134)

packages/opencode/migration/20260312043431_session_message_cursor/migration.sql

@@ -0,0 +1,4 @@
+DROP INDEX IF EXISTS `message_session_idx`;--> statement-breakpoint
+DROP INDEX IF EXISTS `part_message_idx`;--> statement-breakpoint
+CREATE INDEX `message_session_time_created_id_idx` ON `message` (`session_id`,`time_created`,`id`);--> statement-breakpoint
+CREATE INDEX `part_message_id_id_idx` ON `part` (`message_id`,`id`);

packages/opencode/src/server/routes/session.ts

@@ -569,17 +569,65 @@ export const SessionRoutes = lazy(() =>
       ),
       validator(
         "query",
-        z.object({
-          limit: z.coerce.number().optional(),
-        }),
+        z
+          .object({
+            limit: z.coerce
+              .number()
+              .int()
+              .min(0)
+              .optional()
+              .meta({ description: "Maximum number of messages to return" }),
+            before: z
+              .string()
+              .optional()
+              .meta({ description: "Opaque cursor for loading older messages" })
+              .refine(
+                (value) => {
+                  if (!value) return true
+                  try {
+                    MessageV2.cursor.decode(value)
+                    return true
+                  } catch {
+                    return false
+                  }
+                },
+                { message: "Invalid cursor" },
+              ),
+          })
+          .refine((value) => !value.before || value.limit !== undefined, {
+            message: "before requires limit",
+            path: ["before"],
+          }),
       ),
       async (c) => {
         const query = c.req.valid("query")
-        const messages = await Session.messages({
-          sessionID: c.req.valid("param").sessionID,
+        const sessionID = c.req.valid("param").sessionID
+        if (query.limit === undefined) {
+          await Session.get(sessionID)
+          const messages = await Session.messages({ sessionID })
+          return c.json(messages)
+        }
+
+        if (query.limit === 0) {
+          await Session.get(sessionID)
+          const messages = await Session.messages({ sessionID })
+          return c.json(messages)
+        }
+
+        const page = await MessageV2.page({
+          sessionID,
           limit: query.limit,
+          before: query.before,
         })
-        return c.json(messages)
+        if (page.cursor) {
+          const url = new URL(c.req.url)
+          url.searchParams.set("limit", query.limit.toString())
+          url.searchParams.set("before", page.cursor)
+          c.header("Access-Control-Expose-Headers", "Link, X-Next-Cursor")
+          c.header("Link", `<${url.toString()}>; rel=\"next\"`)
+          c.header("X-Next-Cursor", page.cursor)
+        }
+        return c.json(page.items)
       },
     )
     .get(

packages/opencode/src/session/message-v2.ts

@@ -6,8 +6,8 @@ import { APICallError, convertToModelMessages, LoadAPIKeyError, type ModelMessag
 import { LSP } from "../lsp"
 import { Snapshot } from "@/snapshot"
 import { fn } from "@/util/fn"
-import { Database, eq, desc, inArray } from "@/storage/db"
-import { MessageTable, PartTable } from "./session.sql"
+import { Database, NotFoundError, and, desc, eq, inArray, lt, or } from "@/storage/db"
+import { MessageTable, PartTable, SessionTable } from "./session.sql"
 import { ProviderTransform } from "@/provider/transform"
 import { STATUS_CODES } from "http"
 import { Storage } from "@/storage/storage"
@@ -494,6 +494,68 @@ export namespace MessageV2 {
   })
   export type WithParts = z.infer<typeof WithParts>
 
+  const Cursor = z.object({
+    id: MessageID.zod,
+    time: z.number(),
+  })
+  type Cursor = z.infer<typeof Cursor>
+
+  export const cursor = {
+    encode(input: Cursor) {
+      return Buffer.from(JSON.stringify(input)).toString("base64url")
+    },
+    decode(input: string) {
+      return Cursor.parse(JSON.parse(Buffer.from(input, "base64url").toString("utf8")))
+    },
+  }
+
+  const info = (row: typeof MessageTable.$inferSelect) =>
+    ({
+      ...row.data,
+      id: row.id,
+      sessionID: row.session_id,
+    }) as MessageV2.Info
+
+  const part = (row: typeof PartTable.$inferSelect) =>
+    ({
+      ...row.data,
+      id: row.id,
+      sessionID: row.session_id,
+      messageID: row.message_id,
+    }) as MessageV2.Part
+
+  const older = (row: Cursor) =>
+    or(
+      lt(MessageTable.time_created, row.time),
+      and(eq(MessageTable.time_created, row.time), lt(MessageTable.id, row.id)),
+    )
+
+  async function hydrate(rows: (typeof MessageTable.$inferSelect)[]) {
+    const ids = rows.map((row) => row.id)
+    const partByMessage = new Map<string, MessageV2.Part[]>()
+    if (ids.length > 0) {
+      const partRows = Database.use((db) =>
+        db
+          .select()
+          .from(PartTable)
+          .where(inArray(PartTable.message_id, ids))
+          .orderBy(PartTable.message_id, PartTable.id)
+          .all(),
+      )
+      for (const row of partRows) {
+        const next = part(row)
+        const list = partByMessage.get(row.message_id)
+        if (list) list.push(next)
+        else partByMessage.set(row.message_id, [next])
+      }
+    }
+
+    return rows.map((row) => ({
+      info: info(row),
+      parts: partByMessage.get(row.id) ?? [],
+    }))
+  }
+
   export function toModelMessages(
     input: WithParts[],
     model: Provider.Model,
@@ -729,56 +791,61 @@ export namespace MessageV2 {
     )
   }
 
-  export const stream = fn(SessionID.zod, async function* (sessionID) {
-    const size = 50
-    let offset = 0
-    while (true) {
+  export const page = fn(
+    z.object({
+      sessionID: SessionID.zod,
+      limit: z.number().int().positive(),
+      before: z.string().optional(),
+    }),
+    async (input) => {
+      const before = input.before ? cursor.decode(input.before) : undefined
+      const where = before
+        ? and(eq(MessageTable.session_id, input.sessionID), older(before))
+        : eq(MessageTable.session_id, input.sessionID)
       const rows = Database.use((db) =>
         db
           .select()
           .from(MessageTable)
-          .where(eq(MessageTable.session_id, sessionID))
-          .orderBy(desc(MessageTable.time_created))
-          .limit(size)
-          .offset(offset)
+          .where(where)
+          .orderBy(desc(MessageTable.time_created), desc(MessageTable.id))
+          .limit(input.limit + 1)
           .all(),
       )
-      if (rows.length === 0) break
-
-      const ids = rows.map((row) => row.id)
-      const partsByMessage = new Map<string, MessageV2.Part[]>()
-      if (ids.length > 0) {
-        const partRows = Database.use((db) =>
-          db
-            .select()
-            .from(PartTable)
-            .where(inArray(PartTable.message_id, ids))
-            .orderBy(PartTable.message_id, PartTable.id)
-            .all(),
+      if (rows.length === 0) {
+        const row = Database.use((db) =>
+          db.select({ id: SessionTable.id }).from(SessionTable).where(eq(SessionTable.id, input.sessionID)).get(),
         )
-        for (const row of partRows) {
-          const part = {
-            ...row.data,
-            id: row.id,
-            sessionID: row.session_id,
-            messageID: row.message_id,
-          } as MessageV2.Part
-          const list = partsByMessage.get(row.message_id)
-          if (list) list.push(part)
-          else partsByMessage.set(row.message_id, [part])
+        if (!row) throw new NotFoundError({ message: `Session not found: ${input.sessionID}` })
+        return {
+          items: [] as MessageV2.WithParts[],
+          more: false,
         }
       }
 
-      for (const row of rows) {
-        const info = { ...row.data, id: row.id, sessionID: row.session_id } as MessageV2.Info
-        yield {
-          info,
-          parts: partsByMessage.get(row.id) ?? [],
-        }
+      const more = rows.length > input.limit
+      const page = more ? rows.slice(0, input.limit) : rows
+      const items = await hydrate(page)
+      items.reverse()
+      const tail = page.at(-1)
+      return {
+        items,
+        more,
+        cursor: more && tail ? cursor.encode({ id: tail.id, time: tail.time_created }) : undefined,
       }
+    },
+  )
 
-      offset += rows.length
-      if (rows.length < size) break
+  export const stream = fn(SessionID.zod, async function* (sessionID) {
+    const size = 50
+    let before: string | undefined
+    while (true) {
+      const next = await page({ sessionID, limit: size, before })
+      if (next.items.length === 0) break
+      for (let i = next.items.length - 1; i >= 0; i--) {
+        yield next.items[i]
+      }
+      if (!next.more || !next.cursor) break
+      before = next.cursor
     }
   })
 
@@ -797,11 +864,16 @@ export namespace MessageV2 {
       messageID: MessageID.zod,
     }),
     async (input): Promise<WithParts> => {
-      const row = Database.use((db) => db.select().from(MessageTable).where(eq(MessageTable.id, input.messageID)).get())
-      if (!row) throw new Error(`Message not found: ${input.messageID}`)
-      const info = { ...row.data, id: row.id, sessionID: row.session_id } as MessageV2.Info
+      const row = Database.use((db) =>
+        db
+          .select()
+          .from(MessageTable)
+          .where(and(eq(MessageTable.id, input.messageID), eq(MessageTable.session_id, input.sessionID)))
+          .get(),
+      )
+      if (!row) throw new NotFoundError({ message: `Message not found: ${input.messageID}` })
       return {
-        info,
+        info: info(row),
         parts: await parts(input.messageID),
       }
     },

packages/opencode/src/session/session.sql.ts

@@ -54,7 +54,7 @@ export const MessageTable = sqliteTable(
     ...Timestamps,
     data: text({ mode: "json" }).notNull().$type<InfoData>(),
   },
-  (table) => [index("message_session_idx").on(table.session_id)],
+  (table) => [index("message_session_time_created_id_idx").on(table.session_id, table.time_created, table.id)],
 )
 
 export const PartTable = sqliteTable(
@@ -69,7 +69,10 @@ export const PartTable = sqliteTable(
     ...Timestamps,
     data: text({ mode: "json" }).notNull().$type<PartData>(),
   },
-  (table) => [index("part_message_idx").on(table.message_id), index("part_session_idx").on(table.session_id)],
+  (table) => [
+    index("part_message_id_id_idx").on(table.message_id, table.id),
+    index("part_session_idx").on(table.session_id),
+  ],
 )
 
 export const TodoTable = sqliteTable(

packages/opencode/test/server/session-messages.test.ts

@@ -0,0 +1,119 @@
+import { describe, expect, test } from "bun:test"
+import path from "path"
+import { Instance } from "../../src/project/instance"
+import { Server } from "../../src/server/server"
+import { Session } from "../../src/session"
+import { MessageV2 } from "../../src/session/message-v2"
+import { MessageID, PartID, type SessionID } from "../../src/session/schema"
+import { Log } from "../../src/util/log"
+
+const root = path.join(__dirname, "../..")
+Log.init({ print: false })
+
+async function fill(sessionID: SessionID, count: number, time = (i: number) => Date.now() + i) {
+  const ids = [] as MessageID[]
+  for (let i = 0; i < count; i++) {
+    const id = MessageID.ascending()
+    ids.push(id)
+    await Session.updateMessage({
+      id,
+      sessionID,
+      role: "user",
+      time: { created: time(i) },
+      agent: "test",
+      model: { providerID: "test", modelID: "test" },
+      tools: {},
+      mode: "",
+    } as unknown as MessageV2.Info)
+    await Session.updatePart({
+      id: PartID.ascending(),
+      sessionID,
+      messageID: id,
+      type: "text",
+      text: `m${i}`,
+    })
+  }
+  return ids
+}
+
+describe("session messages endpoint", () => {
+  test("returns cursor headers for older pages", async () => {
+    await Instance.provide({
+      directory: root,
+      fn: async () => {
+        const session = await Session.create({})
+        const ids = await fill(session.id, 5)
+        const app = Server.Default()
+
+        const a = await app.request(`/session/${session.id}/message?limit=2`)
+        expect(a.status).toBe(200)
+        const aBody = (await a.json()) as MessageV2.WithParts[]
+        expect(aBody.map((item) => item.info.id)).toEqual(ids.slice(-2))
+        const cursor = a.headers.get("x-next-cursor")
+        expect(cursor).toBeTruthy()
+        expect(a.headers.get("link")).toContain('rel="next"')
+
+        const b = await app.request(`/session/${session.id}/message?limit=2&before=${encodeURIComponent(cursor!)}`)
+        expect(b.status).toBe(200)
+        const bBody = (await b.json()) as MessageV2.WithParts[]
+        expect(bBody.map((item) => item.info.id)).toEqual(ids.slice(-4, -2))
+
+        await Session.remove(session.id)
+      },
+    })
+  })
+
+  test("keeps full-history responses when limit is omitted", async () => {
+    await Instance.provide({
+      directory: root,
+      fn: async () => {
+        const session = await Session.create({})
+        const ids = await fill(session.id, 3)
+        const app = Server.Default()
+
+        const res = await app.request(`/session/${session.id}/message`)
+        expect(res.status).toBe(200)
+        const body = (await res.json()) as MessageV2.WithParts[]
+        expect(body.map((item) => item.info.id)).toEqual(ids)
+
+        await Session.remove(session.id)
+      },
+    })
+  })
+
+  test("rejects invalid cursors and missing sessions", async () => {
+    await Instance.provide({
+      directory: root,
+      fn: async () => {
+        const session = await Session.create({})
+        const app = Server.Default()
+
+        const bad = await app.request(`/session/${session.id}/message?limit=2&before=bad`)
+        expect(bad.status).toBe(400)
+
+        const miss = await app.request(`/session/ses_missing/message?limit=2`)
+        expect(miss.status).toBe(404)
+
+        await Session.remove(session.id)
+      },
+    })
+  })
+
+  test("does not truncate large legacy limit requests", async () => {
+    await Instance.provide({
+      directory: root,
+      fn: async () => {
+        const session = await Session.create({})
+        await fill(session.id, 520)
+        const app = Server.Default()
+
+        const res = await app.request(`/session/${session.id}/message?limit=510`)
+        expect(res.status).toBe(200)
+        const body = (await res.json()) as MessageV2.WithParts[]
+        expect(body).toHaveLength(510)
+
+        await Session.remove(session.id)
+      },
+    })
+  })
+})

packages/opencode/test/session/messages-pagination.test.ts

@@ -0,0 +1,115 @@
+import { describe, expect, test } from "bun:test"
+import path from "path"
+import { Instance } from "../../src/project/instance"
+import { Session } from "../../src/session"
+import { MessageV2 } from "../../src/session/message-v2"
+import { MessageID, PartID, type SessionID } from "../../src/session/schema"
+import { Log } from "../../src/util/log"
+
+const root = path.join(__dirname, "../..")
+Log.init({ print: false })
+
+async function fill(sessionID: SessionID, count: number, time = (i: number) => Date.now() + i) {
+  const ids = [] as MessageID[]
+  for (let i = 0; i < count; i++) {
+    const id = MessageID.ascending()
+    ids.push(id)
+    await Session.updateMessage({
+      id,
+      sessionID,
+      role: "user",
+      time: { created: time(i) },
+      agent: "test",
+      model: { providerID: "test", modelID: "test" },
+      tools: {},
+      mode: "",
+    } as unknown as MessageV2.Info)
+    await Session.updatePart({
+      id: PartID.ascending(),
+      sessionID,
+      messageID: id,
+      type: "text",
+      text: `m${i}`,
+    })
+  }
+  return ids
+}
+
+describe("session message pagination", () => {
+  test("pages backward with opaque cursors", async () => {
+    await Instance.provide({
+      directory: root,
+      fn: async () => {
+        const session = await Session.create({})
+        const ids = await fill(session.id, 6)
+
+        const a = await MessageV2.page({ sessionID: session.id, limit: 2 })
+        expect(a.items.map((item) => item.info.id)).toEqual(ids.slice(-2))
+        expect(a.items.every((item) => item.parts.length === 1)).toBe(true)
+        expect(a.more).toBe(true)
+        expect(a.cursor).toBeTruthy()
+
+        const b = await MessageV2.page({ sessionID: session.id, limit: 2, before: a.cursor! })
+        expect(b.items.map((item) => item.info.id)).toEqual(ids.slice(-4, -2))
+        expect(b.more).toBe(true)
+        expect(b.cursor).toBeTruthy()
+
+        const c = await MessageV2.page({ sessionID: session.id, limit: 2, before: b.cursor! })
+        expect(c.items.map((item) => item.info.id)).toEqual(ids.slice(0, 2))
+        expect(c.more).toBe(false)
+        expect(c.cursor).toBeUndefined()
+
+        await Session.remove(session.id)
+      },
+    })
+  })
+
+  test("keeps stream order newest first", async () => {
+    await Instance.provide({
+      directory: root,
+      fn: async () => {
+        const session = await Session.create({})
+        const ids = await fill(session.id, 5)
+
+        const items = await Array.fromAsync(MessageV2.stream(session.id))
+        expect(items.map((item) => item.info.id)).toEqual(ids.slice().reverse())
+
+        await Session.remove(session.id)
+      },
+    })
+  })
+
+  test("accepts cursors generated from fractional timestamps", async () => {
+    await Instance.provide({
+      directory: root,
+      fn: async () => {
+        const session = await Session.create({})
+        const ids = await fill(session.id, 4, (i) => 1000.5 + i)
+
+        const a = await MessageV2.page({ sessionID: session.id, limit: 2 })
+        const b = await MessageV2.page({ sessionID: session.id, limit: 2, before: a.cursor! })
+
+        expect(a.items.map((item) => item.info.id)).toEqual(ids.slice(-2))
+        expect(b.items.map((item) => item.info.id)).toEqual(ids.slice(0, 2))
+
+        await Session.remove(session.id)
+      },
+    })
+  })
+
+  test("scopes get by session id", async () => {
+    await Instance.provide({
+      directory: root,
+      fn: async () => {
+        const a = await Session.create({})
+        const b = await Session.create({})
+        const [id] = await fill(a.id, 1)
+
+        await expect(MessageV2.get({ sessionID: b.id, messageID: id })).rejects.toMatchObject({ name: "NotFoundError" })
+
+        await Session.remove(a.id)
+        await Session.remove(b.id)
+      },
+    })
+  })
+})

packages/sdk/js/src/v2/gen/sdk.gen.ts

@@ -1793,6 +1793,7 @@ export class Session2 extends HeyApiClient {
       directory?: string
       workspace?: string
       limit?: number
+      before?: string
     },
     options?: Options<never, ThrowOnError>,
   ) {
@@ -1805,6 +1806,7 @@ export class Session2 extends HeyApiClient {
             { in: "query", key: "directory" },
             { in: "query", key: "workspace" },
             { in: "query", key: "limit" },
+            { in: "query", key: "before" },
           ],
         },
       ],

packages/sdk/js/src/v2/gen/types.gen.ts

@@ -3232,7 +3232,11 @@ export type SessionMessagesData = {
   query?: {
     directory?: string
     workspace?: string
+    /**
+     * Maximum number of messages to return
+     */
     limit?: number
+    before?: string
   }
   url: "/session/{sessionID}/message"
 }