progress

Author: thdxr

packages/opencode/src/server/routes/instance/httpapi/v2.ts

@@ -3,21 +3,80 @@ import { SessionMessage } from "@/v2/session-message"
 import { Prompt } from "@/v2/session-prompt"
 import { SessionV2 } from "@/v2/session"
 import { Effect, Layer, Schema } from "effect"
-import { HttpApi, HttpApiBuilder, HttpApiEndpoint, HttpApiGroup, HttpApiSchema, OpenApi } from "effect/unstable/httpapi"
+import * as DateTime from "effect/DateTime"
+import {
+  HttpApi,
+  HttpApiBuilder,
+  HttpApiEndpoint,
+  HttpApiError,
+  HttpApiGroup,
+  HttpApiSchema,
+  OpenApi,
+} from "effect/unstable/httpapi"
 import { Authorization } from "./auth"
 
+const DefaultMessagesLimit = 50
+
+const Cursor = Schema.Struct({
+  id: SessionMessage.ID,
+  time: Schema.Number,
+})
+
+const decodeCursor = Schema.decodeUnknownSync(Cursor)
+
+const cursor = {
+  encode(message: SessionMessage.Message) {
+    return Buffer.from(
+      JSON.stringify({ id: message.id, time: DateTime.toEpochMillis(message.time.created) }),
+    ).toString("base64url")
+  },
+  decode(input: string) {
+    return decodeCursor(JSON.parse(Buffer.from(input, "base64url").toString("utf8")))
+  },
+}
+
 export const V2Api = HttpApi.make("v2")
   .add(
     HttpApiGroup.make("v2")
       .add(
         HttpApiEndpoint.get("messages", "/api/session/:sessionID/message", {
           params: { sessionID: SessionID },
-          success: Schema.Array(SessionMessage.Message),
+          query: Schema.Struct({
+            limit: Schema.optional(
+              Schema.NumberFromString.check(
+                Schema.isInt(),
+                Schema.isGreaterThanOrEqualTo(1),
+                Schema.isLessThanOrEqualTo(200),
+              ),
+            ).annotate({
+              description:
+                "Maximum number of messages to return. When omitted, the endpoint returns its default page size. Use limit without a cursor to fetch the newest page for chat history.",
+            }),
+            before: Schema.optional(Schema.String).annotate({
+              description:
+                "Opaque pagination cursor for the item at the start of the current window. Returns messages older than this cursor. Mutually exclusive with after.",
+            }),
+            after: Schema.optional(Schema.String).annotate({
+              description:
+                "Opaque pagination cursor for the item at the end of the current window. Returns messages newer than this cursor. Mutually exclusive with before.",
+            }),
+            from: Schema.optional(Schema.Literal("start")).annotate({
+              description:
+                "Start from the beginning of session history instead of the newest messages. Mutually exclusive with before and after.",
+            }),
+          }).annotate({ identifier: "V2SessionMessagesQuery" }),
+          success: Schema.Struct({
+            items: Schema.Array(SessionMessage.Message),
+            before: Schema.String.pipe(Schema.optional),
+            after: Schema.String.pipe(Schema.optional),
+          }).annotate({ identifier: "V2SessionMessagesResponse" }),
+          error: HttpApiError.BadRequest,
         }).annotateMerge(
           OpenApi.annotations({
             identifier: "v2.session.messages",
             summary: "Get v2 session messages",
-            description: "Retrieve projected v2 messages for a session directly from the message database.",
+            description:
+              "Retrieve projected v2 messages for a session. For chat clients, request the latest page with limit, page backward through older history with before, and catch up with newer messages using after.",
           }),
         ),
       )
@@ -80,11 +139,37 @@ export const V2Api = HttpApi.make("v2")
 export const v2Handlers = HttpApiBuilder.group(V2Api, "v2", (handlers) =>
   Effect.gen(function* () {
     const session = yield* SessionV2.Service
+
     return handlers
       .handle(
         "messages",
         Effect.fn(function* (ctx) {
-          return yield* session.messages(ctx.params.sessionID)
+          if (ctx.query.before && ctx.query.after) return yield* new HttpApiError.BadRequest({})
+          if (ctx.query.from && (ctx.query.before || ctx.query.after)) return yield* new HttpApiError.BadRequest({})
+          const decoded = yield* Effect.try({
+            try: () => {
+              return {
+                before: ctx.query.before ? cursor.decode(ctx.query.before) : undefined,
+                after: ctx.query.after ? cursor.decode(ctx.query.after) : undefined,
+              }
+            },
+            catch: () => new HttpApiError.BadRequest({}),
+          })
+          const limit = ctx.query.limit ?? DefaultMessagesLimit
+          const messages = yield* session.messages({
+            sessionID: ctx.params.sessionID,
+            limit,
+            before: decoded.before,
+            after: decoded.after,
+            start: ctx.query.from === "start",
+          })
+          const oldest = messages[0]
+          const newest = messages.at(-1)
+          return {
+            items: messages,
+            before: oldest ? cursor.encode(oldest) : undefined,
+            after: newest ? cursor.encode(newest) : undefined,
+          }
         }),
       )
       .handle(

packages/opencode/src/server/routes/instance/v2.ts

@@ -35,7 +35,7 @@ export const V2Routes = lazy(() =>
       return jsonRequest("V2Routes.messages", c, function* () {
         return yield* Effect.gen(function* () {
           const session = yield* SessionV2.Service
-          return yield* session.messages(sessionID)
+          return yield* session.messages({ sessionID })
         }).pipe(Effect.provide(SessionV2.defaultLayer))
       })
     },

packages/opencode/src/v2/session.ts

@@ -1,6 +1,6 @@
 import { SessionMessageTable } from "@/session/session.sql"
 import type { SessionID } from "@/session/schema"
-import { asc, eq } from "@/storage/db"
+import { and, asc, desc, eq, gt, inArray, lt, or } from "@/storage/db"
 import * as Database from "@/storage/db"
 import { Context, Effect, Layer, Schema } from "effect"
 import { SessionMessage } from "./session-message"
@@ -16,8 +16,32 @@ export type Delivery = Schema.Schema.Type<typeof Delivery>
 
 export const DefaultDelivery = "immediate" satisfies Delivery
 
+export type MessagesCursor = {
+  id: SessionMessage.ID
+  time: number
+}
+
+export type MessagesInput = {
+  sessionID: SessionID
+  limit?: number
+  cursor?: MessagesCursor
+  direction?: "before" | "after"
+}
+
+const older = (item: MessagesCursor) =>
+  or(
+    lt(SessionMessageTable.time_created, item.time),
+    and(eq(SessionMessageTable.time_created, item.time), lt(SessionMessageTable.id, item.id)),
+  )
+
+const newer = (item: MessagesCursor) =>
+  or(
+    gt(SessionMessageTable.time_created, item.time),
+    and(eq(SessionMessageTable.time_created, item.time), gt(SessionMessageTable.id, item.id)),
+  )
+
 export interface Interface {
-  readonly messages: (sessionID: SessionID) => Effect.Effect<SessionMessage.Message[], never>
+  readonly messages: (input: MessagesInput) => Effect.Effect<SessionMessage.Message[], never>
   readonly prompt: (input: {
     id?: Event.ID
     sessionID: SessionID
@@ -39,16 +63,54 @@ export const layer = Layer.effect(
       decodeMessage({ ...row.data, id: row.id, type: row.type })
 
     const result: Interface = {
-      messages: Effect.fn("V2Session.messages")(function* (sessionID) {
-        return Database.use((db) =>
-          db
+      messages: Effect.fn("V2Session.messages")(function* (input) {
+        if (input.limit === undefined) {
+          const rows = Database.use((db) =>
+            db
+              .select()
+              .from(SessionMessageTable)
+              .where(eq(SessionMessageTable.session_id, input.sessionID))
+              .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id))
+              .all(),
+          )
+          return rows.map((row) => decode(row))
+        }
+
+        const limit = input.limit
+        const direction = input.direction ?? "before"
+        const where = input.cursor
+          ? and(
+              eq(SessionMessageTable.session_id, input.sessionID),
+              direction === "after" ? newer(input.cursor) : older(input.cursor),
+            )
+          : eq(SessionMessageTable.session_id, input.sessionID)
+        const rows = Database.use((db) => {
+          if (direction === "after") {
+            return db
+              .select()
+              .from(SessionMessageTable)
+              .where(where)
+              .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id))
+              .limit(limit)
+              .all()
+          }
+          const ids = db
+            .select({ id: SessionMessageTable.id })
+            .from(SessionMessageTable)
+            .where(where)
+            .orderBy(desc(SessionMessageTable.time_created), desc(SessionMessageTable.id))
+            .limit(limit)
+            .all()
+            .map((row) => row.id)
+          if (ids.length === 0) return []
+          return db
             .select()
             .from(SessionMessageTable)
-            .where(eq(SessionMessageTable.session_id, sessionID))
+            .where(inArray(SessionMessageTable.id, ids))
             .orderBy(asc(SessionMessageTable.time_created), asc(SessionMessageTable.id))
             .all()
-            .map((row) => decode(row)),
-        )
+        })
+        return rows.map((row) => decode(row))
       }),
       prompt: Effect.fn("V2Session.prompt")(function* (input) {
         const delivery = input.delivery ?? DefaultDelivery

packages/opencode/test/session/compaction.test.ts

@@ -597,7 +597,9 @@ describe("session.compaction.create", () => {
           overflow: true,
         })
 
-        const v2 = yield* SessionV2.Service.use((svc) => svc.messages(info.id)).pipe(Effect.provide(SessionV2.defaultLayer))
+        const v2 = yield* SessionV2.Service.use((svc) => svc.messages({ sessionID: info.id })).pipe(
+          Effect.provide(SessionV2.defaultLayer),
+        )
         expect(v2.at(-1)).toMatchObject({
           type: "compaction",
           reason: "auto",

packages/opencode/test/session/prompt.test.ts

@@ -396,7 +396,7 @@ it.live("prompt emits v2 prompted and synthetic events", () =>
         ],
       })
 
-      const messages = yield* SessionV2.Service.use((session) => session.messages(chat.id)).pipe(
+      const messages = yield* SessionV2.Service.use((session) => session.messages({ sessionID: chat.id })).pipe(
         Effect.provide(SessionV2.layer),
       )
       const row = Database.use((db) =>