From 43924892c65bf36a318d4b73acff12fc7e3a1d54 Mon Sep 17 00:00:00 2001 From: CasualDeveloper <10153929+CasualDeveloper@users.noreply.github.com> Date: Sat, 17 Jan 2026 13:22:18 +0800 Subject: [PATCH] feat(session): add bi-directional cursor-based pagination with Home/End navigation Implements cursor-based pagination for message loading to handle long sessions without memory explosion with absolute navigation via Home/End keys. API changes: - Add 'before' cursor param: fetch messages older than cursor (newest first) - Add 'after' cursor param: fetch messages newer than cursor (oldest first) - Add 'oldest' param: start from oldest messages (for jumpToOldest) - Link headers with rel="prev"/"next" for cursor discovery (RFC 5005) TUI changes: - loadOlder/loadNewer actions with sliding window eviction (500 msg limit) - jumpToOldest (Home): fetches oldest page via ?oldest=true - jumpToLatest (End): fetches newest page, preserves revert marker - Detached mode: ignores SSE when viewing history to prevent gaps Implementation: - Binary.lowerBound for efficient cursor lookup - parseLinkHeader utility for RFC 5988 parsing - Message.stream() reverse option for ascending order - Smart parts cleanup: only deletes parts for evicted messages Tests: - Unit tests for pagination logic and cursor handling - API tests for before/after/oldest params and Link headers Resolves: #6548 --- .../opencode/src/cli/cmd/tui/context/sync.tsx | 290 ++++++++++++++++- .../src/cli/cmd/tui/routes/session/index.tsx | 91 ++++++ .../opencode/src/server/routes/session.ts | 83 ++++- packages/opencode/src/session/index.ts | 125 ++++++- packages/opencode/src/session/message-v2.ts | 36 +- packages/opencode/src/util/link-header.ts | 27 ++ .../test/server/session-messages.test.ts | 184 +++++++++++ .../test/session/messages-pagination.test.ts | 308 ++++++++++++++++++ .../test/util/parse-link-header.test.ts | 59 ++++ packages/sdk/js/src/v2/gen/sdk.gen.ts | 6 + packages/sdk/js/src/v2/gen/types.gen.ts | 3 + packages/sdk/openapi.json | 27 +- packages/util/src/binary.ts | 15 + 13 files changed, 1229 insertions(+), 25 deletions(-) create mode 100644 packages/opencode/src/util/link-header.ts create mode 100644 packages/opencode/test/server/session-messages.test.ts create mode 100644 packages/opencode/test/session/messages-pagination.test.ts create mode 100644 packages/opencode/test/util/parse-link-header.test.ts diff --git a/packages/opencode/src/cli/cmd/tui/context/sync.tsx b/packages/opencode/src/cli/cmd/tui/context/sync.tsx index 392cfb7f121..2b9f2980842 100644 --- a/packages/opencode/src/cli/cmd/tui/context/sync.tsx +++ b/packages/opencode/src/cli/cmd/tui/context/sync.tsx @@ -28,6 +28,12 @@ import { useArgs } from "./args" import { batch, onMount } from "solid-js" import { Log } from "@/util/log" import type { Path } from "@opencode-ai/sdk" +import { parseLinkHeader } from "@/util/link-header" + +/** Maximum messages kept in memory per session */ +const MAX_LOADED_MESSAGES = 500 +/** Chunk size for eviction when limit exceeded */ +const EVICTION_CHUNK_SIZE = 50 export const { use: useSync, provider: SyncProvider } = createSimpleContext({ name: "Sync", @@ -48,6 +54,15 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({ } config: Config session: Session[] + message_page: { + [sessionID: string]: { + hasOlder: boolean + hasNewer: boolean + loading: boolean + loadingDirection?: "older" | "newer" + error?: string + } + } session_status: { [sessionID: string]: SessionStatus } @@ -89,6 +104,7 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({ provider: [], provider_default: {}, session: [], + message_page: {}, session_status: {}, session_diff: {}, todo: {}, @@ -226,19 +242,24 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({ } case "message.updated": { - const messages = store.message[event.properties.info.sessionID] + const sessionID = event.properties.info.sessionID + const page = store.message_page[sessionID] + const messages = store.message[sessionID] if (!messages) { - setStore("message", event.properties.info.sessionID, [event.properties.info]) + setStore("message", sessionID, [event.properties.info]) break } const result = Binary.search(messages, event.properties.info.id, (m) => m.id) if (result.found) { - setStore("message", event.properties.info.sessionID, result.index, reconcile(event.properties.info)) + setStore("message", sessionID, result.index, reconcile(event.properties.info)) + break + } + if (page?.hasNewer) { break } setStore( "message", - event.properties.info.sessionID, + sessionID, produce((draft) => { draft.splice(result.index, 0, event.properties.info) }), @@ -279,6 +300,13 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({ break } case "message.part.updated": { + const sessionID = event.properties.part.sessionID + const page = store.message_page[sessionID] + const messages = store.message[sessionID] + const messageExists = messages?.some((m) => m.id === event.properties.part.messageID) + if (page?.hasNewer && !messageExists) { + break + } const parts = store.part[event.properties.part.messageID] if (!parts) { setStore("part", event.properties.part.messageID, [event.properties.part]) @@ -389,6 +417,7 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({ }) const fullSyncedSessions = new Set() + const loadingGuard = new Set() const result = { data: store, set: setStore, @@ -422,6 +451,8 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({ sdk.client.session.todo({ sessionID }), sdk.client.session.diff({ sessionID }), ]) + const link = messages.response.headers.get("link") ?? "" + const hasOlder = parseLinkHeader(link).prev !== undefined setStore( produce((draft) => { const match = Binary.search(draft.session, sessionID, (s) => s.id) @@ -433,10 +464,261 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({ draft.part[message.info.id] = message.parts } draft.session_diff[sessionID] = diff.data ?? [] + draft.message_page[sessionID] = { hasOlder, hasNewer: false, loading: false, error: undefined } }), ) fullSyncedSessions.add(sessionID) }, + async loadOlder(sessionID: string) { + const page = store.message_page[sessionID] + if (page?.loading || !page?.hasOlder) return + const messages = store.message[sessionID] ?? [] + const oldest = messages.at(0) + if (!oldest) return + if (loadingGuard.has(sessionID)) return + loadingGuard.add(sessionID) + try { + setStore("message_page", sessionID, { ...page, loading: true, loadingDirection: "older", error: undefined }) + + const res = await sdk.client.session.messages( + { sessionID, before: oldest.id, limit: 100 }, + { throwOnError: true }, + ) + const link = res.response.headers.get("link") ?? "" + const hasOlder = parseLinkHeader(link).prev !== undefined + setStore( + produce((draft) => { + const existing = draft.message[sessionID] ?? [] + for (const msg of res.data ?? []) { + const match = Binary.search(existing, msg.info.id, (m) => m.id) + if (!match.found) { + existing.splice(match.index, 0, msg.info) + draft.part[msg.info.id] = msg.parts + } + } + if (existing.length > MAX_LOADED_MESSAGES + EVICTION_CHUNK_SIZE) { + const evicted = existing.splice(-(existing.length - MAX_LOADED_MESSAGES)) + for (const msg of evicted) delete draft.part[msg.id] + draft.message_page[sessionID] = { hasOlder, hasNewer: true, loading: false, error: undefined } + } else { + draft.message_page[sessionID] = { + hasOlder, + hasNewer: draft.message_page[sessionID]?.hasNewer ?? false, + loading: false, + error: undefined, + } + } + }), + ) + } catch (e) { + const page = store.message_page[sessionID] + setStore("message_page", sessionID, { + hasOlder: page?.hasOlder ?? false, + hasNewer: page?.hasNewer ?? false, + loading: false, + error: e instanceof Error ? e.message : String(e), + }) + } finally { + loadingGuard.delete(sessionID) + } + }, + async loadNewer(sessionID: string) { + const page = store.message_page[sessionID] + if (page?.loading || !page?.hasNewer) return + const messages = store.message[sessionID] ?? [] + const newest = messages.at(-1) + if (!newest) return + if (loadingGuard.has(sessionID)) return + loadingGuard.add(sessionID) + try { + setStore("message_page", sessionID, { ...page, loading: true, loadingDirection: "newer", error: undefined }) + + const res = await sdk.client.session.messages( + { sessionID, after: newest.id, limit: 100 }, + { throwOnError: true }, + ) + const link = res.response.headers.get("link") ?? "" + const hasNewer = parseLinkHeader(link).next !== undefined + setStore( + produce((draft) => { + const existing = draft.message[sessionID] ?? [] + for (const msg of res.data ?? []) { + const match = Binary.search(existing, msg.info.id, (m) => m.id) + if (!match.found) { + existing.splice(match.index, 0, msg.info) + draft.part[msg.info.id] = msg.parts + } + } + if (existing.length > MAX_LOADED_MESSAGES + EVICTION_CHUNK_SIZE) { + const evicted = existing.splice(0, existing.length - MAX_LOADED_MESSAGES) + for (const msg of evicted) delete draft.part[msg.id] + draft.message_page[sessionID] = { hasOlder: true, hasNewer, loading: false, error: undefined } + } else { + draft.message_page[sessionID] = { + hasOlder: draft.message_page[sessionID]?.hasOlder ?? false, + hasNewer, + loading: false, + error: undefined, + } + } + }), + ) + } catch (e) { + const page = store.message_page[sessionID] + setStore("message_page", sessionID, { + hasOlder: page?.hasOlder ?? false, + hasNewer: page?.hasNewer ?? false, + loading: false, + error: e instanceof Error ? e.message : String(e), + }) + } finally { + loadingGuard.delete(sessionID) + } + }, + async jumpToLatest(sessionID: string) { + const page = store.message_page[sessionID] + if (page?.loading || !page?.hasNewer) return + if (loadingGuard.has(sessionID)) return + loadingGuard.add(sessionID) + + try { + // Check for revert state + const session = store.session.find((s) => s.id === sessionID) + const revertMessageID = session?.revert?.messageID + + setStore("message_page", sessionID, { + ...page, + loading: true, + loadingDirection: "newer", + error: undefined, + }) + + // Fetch newest page (no cursor = newest) + const res = await sdk.client.session.messages({ sessionID, limit: 100 }, { throwOnError: true }) + + let messages = res.data ?? [] + const link = res.response.headers.get("link") ?? "" + const hasOlder = parseLinkHeader(link).prev !== undefined + + // Revert-aware: If in revert state and marker not in results, fetch it + if (revertMessageID && !messages.some((m) => m.info.id === revertMessageID)) { + try { + const revertResult = await sdk.client.session.message( + { sessionID, messageID: revertMessageID }, + { throwOnError: true }, + ) + if (revertResult.data) { + // Prepend revert message (it's older than newest page) + messages = [revertResult.data, ...messages] + } + } catch (e) { + // Revert message may have been deleted, continue without it + Log.Default.info("Revert marker fetch failed (may be deleted)", { + messageID: revertMessageID, + error: e, + }) + } + } + + setStore( + produce((draft) => { + // Clean up parts only for messages not in new results + const oldMessages = draft.message[sessionID] ?? [] + const newIds = new Set(messages.map((m) => m.info.id)) + for (const msg of oldMessages) { + if (!newIds.has(msg.id)) { + delete draft.part[msg.id] + } + } + + // Store new messages + draft.message[sessionID] = messages.map((m) => m.info) + for (const msg of messages) { + draft.part[msg.info.id] = msg.parts + } + draft.message_page[sessionID] = { + hasOlder, + hasNewer: false, + loading: false, + error: undefined, + } + }), + ) + } catch (e) { + setStore( + produce((draft) => { + const p = draft.message_page[sessionID] + if (p) { + p.loading = false + p.error = e instanceof Error ? e.message : String(e) + } + }), + ) + } finally { + loadingGuard.delete(sessionID) + } + }, + async jumpToOldest(sessionID: string) { + const page = store.message_page[sessionID] + if (page?.loading || !page?.hasOlder) return + if (loadingGuard.has(sessionID)) return + loadingGuard.add(sessionID) + + try { + setStore("message_page", sessionID, { + ...page, + loading: true, + loadingDirection: "older", + error: undefined, + }) + + const res = await sdk.client.session.messages( + { sessionID, oldest: true, limit: 100 }, + { throwOnError: true }, + ) + + const messages = res.data ?? [] + const link = res.response.headers.get("link") ?? "" + const hasNewer = parseLinkHeader(link).next !== undefined + + setStore( + produce((draft) => { + // Clean up parts only for messages not in new results + const oldMessages = draft.message[sessionID] ?? [] + const newIds = new Set(messages.map((m) => m.info.id)) + for (const msg of oldMessages) { + if (!newIds.has(msg.id)) { + delete draft.part[msg.id] + } + } + + // Store new messages + draft.message[sessionID] = messages.map((m) => m.info) + for (const msg of messages) { + draft.part[msg.info.id] = msg.parts + } + draft.message_page[sessionID] = { + hasOlder: false, + hasNewer, + loading: false, + error: undefined, + } + }), + ) + } catch (e) { + setStore( + produce((draft) => { + const p = draft.message_page[sessionID] + if (p) { + p.loading = false + p.error = e instanceof Error ? e.message : String(e) + } + }), + ) + } finally { + loadingGuard.delete(sessionID) + } + }, }, bootstrap, } diff --git a/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx b/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx index 1294ab849e9..d6ea9a10a76 100644 --- a/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx +++ b/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx @@ -119,6 +119,7 @@ export function Session() { .toSorted((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)) }) const messages = createMemo(() => sync.data.message[route.sessionID] ?? []) + const paging = createMemo(() => sync.data.message_page[route.sessionID]) const permissions = createMemo(() => { if (session()?.parentID) return [] return children().flatMap((x) => sync.data.permission[x.id] ?? []) @@ -128,6 +129,34 @@ export function Session() { return children().flatMap((x) => sync.data.question[x.id] ?? []) }) + const LOAD_MORE_THRESHOLD = 5 + + const loadOlder = () => { + const page = paging() + if (!page?.hasOlder || page.loading || !scroll) return + if (scroll.y > LOAD_MORE_THRESHOLD) return + + const height = scroll.scrollHeight + const y = scroll.y + sync.session.loadOlder(route.sessionID).then(() => { + queueMicrotask(() => { + requestAnimationFrame(() => { + const delta = scroll.scrollHeight - height + if (delta > 0) scroll.scrollTo(y + delta) + }) + }) + }) + } + + const loadNewer = () => { + const page = paging() + if (!page?.hasNewer || page.loading || !scroll) return + const bottomDistance = scroll.scrollHeight - scroll.y - scroll.height + if (bottomDistance > LOAD_MORE_THRESHOLD) return + + sync.session.loadNewer(route.sessionID) + } + const pending = createMemo(() => { return messages().findLast((x) => x.role === "assistant" && !x.time.completed)?.id }) @@ -963,8 +992,60 @@ export function Session() {
+ + + Loading older messages... + + + + + (scroll up for more) + + + + + Failed to load: {paging()?.error} + (scroll to retry) + + (scroll = r)} + onMouseScroll={() => { + loadOlder() + loadNewer() + }} + onKeyDown={(e) => { + // End: Jump to latest (only when detached from live) + if (e.name === "end" && paging()?.hasNewer && !paging()?.loading) { + e.preventDefault() + sync.session.jumpToLatest(route.sessionID).then(() => { + requestAnimationFrame(() => { + if (scroll) scroll.scrollTo(scroll.scrollHeight) + }) + }) + return + } + + // Home: Jump to oldest (only when more history exists) + if (e.name === "home" && paging()?.hasOlder && !paging()?.loading) { + e.preventDefault() + sync.session.jumpToOldest(route.sessionID).then(() => { + requestAnimationFrame(() => { + if (scroll) scroll.scrollTo(0) + }) + }) + return + } + + // Standard scroll triggers incremental load + if (["up", "pageup", "home"].includes(e.name)) { + setTimeout(loadOlder, 0) + } + if (["down", "pagedown", "end"].includes(e.name)) { + setTimeout(loadNewer, 0) + } + }} + viewportCulling={true} viewportOptions={{ paddingRight: showScrollbar() ? 1 : 0, }} @@ -1077,6 +1158,16 @@ export function Session() { )} + + + Loading newer messages... + + + + + (scroll down for more) + + 0}> diff --git a/packages/opencode/src/server/routes/session.ts b/packages/opencode/src/server/routes/session.ts index a98624dfae2..aee1d4e420c 100644 --- a/packages/opencode/src/server/routes/session.ts +++ b/packages/opencode/src/server/routes/session.ts @@ -4,6 +4,7 @@ import { describeRoute, validator, resolver } from "hono-openapi" import z from "zod" import { Session } from "../../session" import { MessageV2 } from "../../session/message-v2" +import { Identifier } from "../../id/id" import { SessionPrompt } from "../../session/prompt" import { SessionCompaction } from "../../session/compaction" import { SessionRevert } from "../../session/revert" @@ -566,16 +567,88 @@ export const SessionRoutes = lazy(() => validator( "query", z.object({ - limit: z.coerce.number().optional(), + limit: z.coerce.number().int().min(1).max(100).optional(), + before: Identifier.schema("message").optional(), + after: Identifier.schema("message").optional(), + oldest: z.coerce.boolean().optional(), }), ), async (c) => { const query = c.req.valid("query") - const messages = await Session.messages({ - sessionID: c.req.valid("param").sessionID, - limit: query.limit, + if (query.before && query.after) { + return c.json({ error: "Cannot specify both 'before' and 'after'" }, 400) + } + if (query.oldest && (query.before || query.after)) { + return c.json({ error: "Cannot use 'oldest' with 'before' or 'after'" }, 400) + } + + const limit = query.limit ?? 100 + const sessionID = c.req.valid("param").sessionID + + if (query.oldest) { + const page = await Session.messages({ + sessionID, + limit: limit + 1, + oldest: true, + }) + + if (page.length > limit) { + const messages = page.slice(0, -1) + const last = messages.at(-1) + if (last) { + const url = new URL(c.req.url) + url.searchParams.delete("oldest") + url.searchParams.set("limit", limit.toString()) + url.searchParams.set("after", last.info.id) + c.header("Link", `<${url.toString()}>; rel=\"next\"`) + } + return c.json(messages) + } + + return c.json(page) + } + + if (query.after) { + const page = await Session.messages({ + sessionID, + limit: limit + 1, + after: query.after, + }) + + if (page.length > limit) { + const messages = page.slice(0, -1) + const last = messages.at(-1) + if (last) { + const url = new URL(c.req.url) + url.searchParams.set("limit", limit.toString()) + url.searchParams.set("after", last.info.id) + c.header("Link", `<${url.toString()}>; rel=\"next\"`) + } + return c.json(messages) + } + + return c.json(page) + } + + const page = await Session.messages({ + sessionID, + limit: limit + 1, + before: query.before, }) - return c.json(messages) + + if (page.length > limit) { + const messages = page.slice(1) + const first = messages.at(0) + if (first) { + const url = new URL(c.req.url) + url.searchParams.set("limit", limit.toString()) + url.searchParams.set("before", first.info.id) + c.header("Link", `<${url.toString()}>; rel=\"prev\"`) + } + return c.json(messages) + } + + return c.json(page) }, ) .get( diff --git a/packages/opencode/src/session/index.ts b/packages/opencode/src/session/index.ts index 3fcdab5238c..01e93cb6bb9 100644 --- a/packages/opencode/src/session/index.ts +++ b/packages/opencode/src/session/index.ts @@ -1,4 +1,5 @@ import { Slug } from "@opencode-ai/util/slug" +import { Binary } from "@opencode-ai/util/binary" import path from "path" import { BusEvent } from "@/bus/bus-event" import { Bus } from "@/bus" @@ -9,7 +10,6 @@ import { Config } from "../config/config" import { Flag } from "../flag/flag" import { Identifier } from "../id/id" import { Installation } from "../installation" - import { Storage } from "../storage/storage" import { Log } from "../util/log" import { MessageV2 } from "./message-v2" @@ -18,7 +18,6 @@ import { SessionPrompt } from "./prompt" import { fn } from "@/util/fn" import { Command } from "../command" import { Snapshot } from "@/snapshot" - import type { Provider } from "@/provider/provider" import { PermissionNext } from "@/permission/next" import { Global } from "@/global" @@ -293,13 +292,127 @@ export namespace Session { z.object({ sessionID: Identifier.schema("session"), limit: z.number().optional(), + before: Identifier.schema("message").optional(), + after: Identifier.schema("message").optional(), + oldest: z.boolean().optional(), }), async (input) => { - const result = [] as MessageV2.WithParts[] - for await (const msg of MessageV2.stream(input.sessionID)) { - if (input.limit && result.length >= input.limit) break - result.push(msg) + // Mutual exclusion validation (fail-fast before I/O) + if (input.before && input.after) { + throw new Error("Cannot specify both 'before' and 'after' cursors") + } + if (input.oldest && (input.before || input.after)) { + throw new Error("Cannot use 'oldest' with 'before' or 'after' cursors") + } + + const list = await Storage.list(["message", input.sessionID]) + const ids = list.map((x) => x[2]).filter((x): x is string => typeof x === "string") + + // Handle oldest=true: iterate from oldest to newest + if (input.oldest) { + const targetIds = [] as string[] + for (let i = 0; i < ids.length; i++) { + if (input.limit !== undefined && targetIds.length >= input.limit) break + targetIds.push(ids[i]) + } + + const results = await Promise.allSettled( + targetIds.map((id) => MessageV2.get({ sessionID: input.sessionID, messageID: id })), + ) + const result = results + .filter((r): r is PromiseFulfilledResult => r.status === "fulfilled") + .map((r) => r.value) + + const failed = results.filter((r) => r.status === "rejected") + if (failed.length > 0) { + log.warn("pagination: skipped failed message fetches", { + count: failed.length, + sessionID: input.sessionID, + reasons: failed.map((r) => (r as PromiseRejectedResult).reason?.message ?? "unknown"), + }) + } + + if (result.length === 0 && targetIds.length > 0) { + throw new Error( + `Failed to fetch ${targetIds.length} messages for session ${input.sessionID} (${result.length} succeeded)`, + ) + } + + return result + } + + if (input.after) { + const cursorIndex = Binary.lowerBound(ids, input.after) + if (ids[cursorIndex] !== input.after) { + log.warn("pagination: cursor not found in session", { cursor: input.after, sessionID: input.sessionID }) + } + const start = ids[cursorIndex] === input.after ? cursorIndex + 1 : cursorIndex + + const targetIds = [] as string[] + for (let i = start; i < ids.length; i++) { + if (input.limit !== undefined && targetIds.length >= input.limit) break + targetIds.push(ids[i]) + } + + const results = await Promise.allSettled( + targetIds.map((id) => MessageV2.get({ sessionID: input.sessionID, messageID: id })), + ) + const result = results + .filter((r): r is PromiseFulfilledResult => r.status === "fulfilled") + .map((r) => r.value) + + const failed = results.filter((r) => r.status === "rejected") + if (failed.length > 0) { + log.warn("pagination: skipped failed message fetches", { + count: failed.length, + sessionID: input.sessionID, + reasons: failed.map((r) => (r as PromiseRejectedResult).reason?.message ?? "unknown"), + }) + } + + if (result.length === 0 && targetIds.length > 0) { + throw new Error( + `Failed to fetch ${targetIds.length} messages for session ${input.sessionID} (${result.length} succeeded)`, + ) + } + + return result + } + + const cursorIndex = input.before ? Binary.lowerBound(ids, input.before) : -1 + if (input.before && ids[cursorIndex] !== input.before) { + log.warn("pagination: cursor not found in session", { cursor: input.before, sessionID: input.sessionID }) } + const start = input.before ? cursorIndex - 1 : ids.length - 1 + + const targetIds = [] as string[] + for (let i = start; i >= 0; i--) { + if (input.limit !== undefined && targetIds.length >= input.limit) break + targetIds.push(ids[i]) + } + + const results = await Promise.allSettled( + targetIds.map((id) => MessageV2.get({ sessionID: input.sessionID, messageID: id })), + ) + const result = results + .filter((r): r is PromiseFulfilledResult => r.status === "fulfilled") + .map((r) => r.value) + + const failed = results.filter((r) => r.status === "rejected") + if (failed.length > 0) { + log.warn("pagination: skipped failed message fetches", { + count: failed.length, + sessionID: input.sessionID, + reasons: failed.map((r) => (r as PromiseRejectedResult).reason?.message ?? "unknown"), + }) + } + + if (result.length === 0 && targetIds.length > 0) { + throw new Error( + `Failed to fetch ${targetIds.length} messages for session ${input.sessionID} (${result.length} succeeded)`, + ) + } + result.reverse() return result }, diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index d326976f1ae..cb7e3f336d4 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -568,15 +568,33 @@ export namespace MessageV2 { return convertToModelMessages(result.filter((msg) => msg.parts.some((part) => part.type !== "step-start"))) } - export const stream = fn(Identifier.schema("session"), async function* (sessionID) { - const list = await Array.fromAsync(await Storage.list(["message", sessionID])) - for (let i = list.length - 1; i >= 0; i--) { - yield await get({ - sessionID, - messageID: list[i][2], - }) - } - }) + export const stream = fn( + z.union([ + Identifier.schema("session"), + z.object({ + sessionID: Identifier.schema("session"), + ascending: z.boolean().optional(), + }), + ]), + async function* (input) { + const sessionID = typeof input === "string" ? input : input.sessionID + const ascending = typeof input === "object" && input.ascending + + const list = await Array.fromAsync(await Storage.list(["message", sessionID])) + + if (ascending) { + // Oldest-first (chronological order) + for (let i = 0; i < list.length; i++) { + yield await get({ sessionID, messageID: list[i][2] }) + } + } else { + // Newest-first (default, for pagination loading older messages) + for (let i = list.length - 1; i >= 0; i--) { + yield await get({ sessionID, messageID: list[i][2] }) + } + } + }, + ) export const parts = fn(Identifier.schema("message"), async (messageID) => { const result = [] as MessageV2.Part[] diff --git a/packages/opencode/src/util/link-header.ts b/packages/opencode/src/util/link-header.ts new file mode 100644 index 00000000000..7ff5c0fb34d --- /dev/null +++ b/packages/opencode/src/util/link-header.ts @@ -0,0 +1,27 @@ +/** + * Parse RFC 8288 Link header into a map of rel -> URL + * @see https://www.rfc-editor.org/rfc/rfc8288 + */ +export function parseLinkHeader(header: string): Record { + if (!header) return {} + try { + const links: Record = {} + const parts = header.split(",") + for (const part of parts) { + const section = part.split(";") + if (section.length < 2) continue + const url = section[0].replace(/<(.*?)>/, "$1").trim() + + for (const attr of section.slice(1)) { + const match = attr.match(/rel=["']?(?[^"']+)["']?/) + if (match?.groups?.rel) { + links[match.groups.rel.trim()] = url + break + } + } + } + return links + } catch { + return {} + } +} diff --git a/packages/opencode/test/server/session-messages.test.ts b/packages/opencode/test/server/session-messages.test.ts new file mode 100644 index 00000000000..28e10cfea74 --- /dev/null +++ b/packages/opencode/test/server/session-messages.test.ts @@ -0,0 +1,184 @@ +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 { Identifier } from "../../src/id/id" +import { Log } from "../../src/util/log" + +const projectRoot = path.join(__dirname, "../..") +Log.init({ print: false }) + +const TEST_TIMEOUT_MS = 30_000 + +describe("session.messages API", () => { + test( + "returns 400 when both before and after specified", + async () => { + await Instance.provide({ + directory: projectRoot, + fn: async () => { + const app = Server.App() + const session = await Session.create({}) + + const response = await app.request(`/session/${session.id}/message?before=msg_01ABC&after=msg_01XYZ`) + + expect(response.status).toBe(400) + const body = (await response.json()) as { error: string } + expect(body.error).toContain("Cannot specify both") + }, + }) + }, + TEST_TIMEOUT_MS, + ) + + test("includes Link header with rel=prev when more pages exist (before cursor)", async () => { + await Instance.provide({ + directory: projectRoot, + fn: async () => { + const app = Server.App() + const session = await Session.create({}) + + // Create 5 messages + for (let i = 0; i < 5; i++) { + await Session.updateMessage({ + id: Identifier.ascending("message"), + role: "user", + sessionID: session.id, + agent: "default", + model: { providerID: "test", modelID: "test" }, + time: { created: Date.now() }, + }) + } + + // Request with limit=2 (should have more) + const response = await app.request(`/session/${session.id}/message?limit=2`) + + expect(response.status).toBe(200) + const link = response.headers.get("Link") + expect(link).toContain('rel="prev"') + expect(link).toContain("before=") + }, + }) + }) + + test("includes Link header with rel=next when using after cursor with more pages", async () => { + await Instance.provide({ + directory: projectRoot, + fn: async () => { + const app = Server.App() + const session = await Session.create({}) + + // Create 5 messages + const ids: string[] = [] + for (let i = 0; i < 5; i++) { + const msg = await Session.updateMessage({ + id: Identifier.ascending("message"), + role: "user", + sessionID: session.id, + agent: "default", + model: { providerID: "test", modelID: "test" }, + time: { created: Date.now() }, + }) + ids.push(msg.id) + } + + // Request after first message with limit=2 + const response = await app.request(`/session/${session.id}/message?after=${ids[0]}&limit=2`) + + expect(response.status).toBe(200) + const link = response.headers.get("Link") + expect(link).toContain('rel="next"') + expect(link).toContain("after=") + }, + }) + }) + + test("omits Link header when no more pages", async () => { + await Instance.provide({ + directory: projectRoot, + fn: async () => { + const app = Server.App() + const session = await Session.create({}) + + // Create 2 messages + for (let i = 0; i < 2; i++) { + await Session.updateMessage({ + id: Identifier.ascending("message"), + role: "user", + sessionID: session.id, + agent: "default", + model: { providerID: "test", modelID: "test" }, + time: { created: Date.now() }, + }) + } + + // Request with limit=10 (more than available) + const response = await app.request(`/session/${session.id}/message?limit=10`) + + expect(response.status).toBe(200) + const link = response.headers.get("Link") + expect(link).toBeNull() + }, + }) + }) + + test("returns 400 when oldest used with before or after", async () => { + await Instance.provide({ + directory: projectRoot, + fn: async () => { + const app = Server.App() + const session = await Session.create({}) + + const response1 = await app.request(`/session/${session.id}/message?oldest=true&before=msg_01ABC`) + expect(response1.status).toBe(400) + const body1 = (await response1.json()) as { error: string } + expect(body1.error).toContain("Cannot use 'oldest' with") + + const response2 = await app.request(`/session/${session.id}/message?oldest=true&after=msg_01XYZ`) + expect(response2.status).toBe(400) + const body2 = (await response2.json()) as { error: string } + expect(body2.error).toContain("Cannot use 'oldest' with") + }, + }) + }) + + test("oldest=true returns messages in ascending order with rel=next Link", async () => { + await Instance.provide({ + directory: projectRoot, + fn: async () => { + const app = Server.App() + const session = await Session.create({}) + + // Create 5 messages with small delay to ensure ordering + const ids: string[] = [] + for (let i = 0; i < 5; i++) { + const msg = await Session.updateMessage({ + id: Identifier.ascending("message"), + role: "user", + sessionID: session.id, + agent: "default", + model: { providerID: "test", modelID: "test" }, + time: { created: Date.now() }, + }) + ids.push(msg.id) + } + + // Request oldest with limit=2 (should have more pages) + const response = await app.request(`/session/${session.id}/message?oldest=true&limit=2`) + + expect(response.status).toBe(200) + const messages = (await response.json()) as Array<{ info: { id: string } }> + expect(messages.length).toBe(2) + // Oldest messages should be first (ascending order) + expect(messages[0].info.id).toBe(ids[0]) + expect(messages[1].info.id).toBe(ids[1]) + + const link = response.headers.get("Link") + expect(link).toContain('rel="next"') + expect(link).toContain("after=") + expect(link).not.toContain("oldest=") // oldest param stripped on subsequent pages + }, + }) + }) +}) diff --git a/packages/opencode/test/session/messages-pagination.test.ts b/packages/opencode/test/session/messages-pagination.test.ts new file mode 100644 index 00000000000..368d22409cf --- /dev/null +++ b/packages/opencode/test/session/messages-pagination.test.ts @@ -0,0 +1,308 @@ +import { describe, expect, test } from "bun:test" +import { Session } from "../../src/session" +import { Log } from "../../src/util/log" +import { Instance } from "../../src/project/instance" +import { Identifier } from "../../src/id/id" +import { tmpdir } from "../fixture/fixture" + +Log.init({ print: false }) + +const TEST_TIMEOUT_MS = 30_000 + +describe("session messages pagination", () => { + test( + "should paginate messages correctly", + async () => { + await using tmp = await tmpdir({ git: true }) + await Instance.provide({ + directory: tmp.path, + fn: async () => { + const session = await Session.create({}) + const sessionID = session.id + const messageCount = 10 + const messageIds: string[] = [] + + // Create 10 messages + for (let i = 0; i < messageCount; i++) { + const msg = await Session.updateMessage({ + id: Identifier.ascending("message"), + role: "user", + sessionID, + agent: "default", + model: { providerID: "openai", modelID: "gpt-4" }, + // time is optional/handled by default, ULID handles ordering + time: { created: Date.now() }, + }) + messageIds.push(msg.id) + await Session.updatePart({ + id: Identifier.ascending("part"), + messageID: msg.id, + sessionID, + type: "text", + text: `Message ${i}`, + }) + } + + // 1. Initial load (limit 3) -> should get last 3 (7, 8, 9) + const page1 = await Session.messages({ + sessionID, + limit: 3, + }) + expect(page1.length).toBe(3) + expect(page1[0].info.id).toBe(messageIds[7]) + expect(page1[2].info.id).toBe(messageIds[9]) + + // 2. Load before page1[0] (limit 3) -> should get 4, 5, 6 + const page2 = await Session.messages({ + sessionID, + limit: 3, + before: page1[0].info.id, + }) + expect(page2.length).toBe(3) + expect(page2[0].info.id).toBe(messageIds[4]) + expect(page2[2].info.id).toBe(messageIds[6]) + + // 3. Load before page2[0] (limit 3) -> should get 1, 2, 3 + const page3 = await Session.messages({ + sessionID, + limit: 3, + before: page2[0].info.id, + }) + expect(page3.length).toBe(3) + expect(page3[0].info.id).toBe(messageIds[1]) + expect(page3[2].info.id).toBe(messageIds[3]) + + // 4. Load before page3[0] (limit 3) -> should get 0 (and only 1 message) + const page4 = await Session.messages({ + sessionID, + limit: 3, + before: page3[0].info.id, + }) + expect(page4.length).toBe(1) + expect(page4[0].info.id).toBe(messageIds[0]) + + // 5. Load before page4[0] -> should be empty + const page5 = await Session.messages({ + sessionID, + limit: 3, + before: page4[0].info.id, + }) + expect(page5.length).toBe(0) + + // 6. Test boundary: exact match (before message 9, should get 0..8) + // Wait, 'before' filters out the cursor itself. + // If IDs are [0..9]. before=ids[9]. + // Should get ids[0..8]. Length 9. + const exact = await Session.messages({ + sessionID, + limit: 10, + before: messageIds[9], + }) + expect(exact.length).toBe(9) + expect(exact[8].info.id).toBe(messageIds[8]) + + // 7. Test boundary: unknown cursor (lexicographically larger) + const unknownFuture = "msg" + "z".repeat(26) + const pageFuture = await Session.messages({ + sessionID, + limit: 3, + before: unknownFuture, + }) + expect(pageFuture.length).toBe(3) + expect(pageFuture[2].info.id).toBe(messageIds[9]) + + // 8. Test boundary: unknown cursor (lexicographically smaller) + const unknownPast = "msg" + "0".repeat(26) + const pagePast = await Session.messages({ + sessionID, + limit: 3, + before: unknownPast, + }) + expect(pagePast.length).toBe(0) + + // 9. Test concurrent load + const [res1, res2] = await Promise.all([ + Session.messages({ sessionID, limit: 3, before: page1[0].info.id }), + Session.messages({ sessionID, limit: 3, before: page1[0].info.id }), + ]) + + expect(res1[0].info.id).toBe(res2[0].info.id) + }, + }) + }, + TEST_TIMEOUT_MS, + ) + + test( + "handles deleted message during pagination", + async () => { + await using tmp = await tmpdir({ git: true }) + await Instance.provide({ + directory: tmp.path, + fn: async () => { + const session = await Session.create({}) + const sessionID = session.id + const messageIds: string[] = [] + + // Create 10 messages + for (let i = 0; i < 10; i++) { + const msg = await Session.updateMessage({ + id: Identifier.ascending("message"), + role: "user", + sessionID, + agent: "default", + model: { providerID: "openai", modelID: "gpt-4" }, + time: { created: Date.now() }, + }) + messageIds.push(msg.id) + } + + // Get first page (messages 6-10) + const page1 = await Session.messages({ sessionID, limit: 5 }) + expect(page1.length).toBe(5) + expect(page1[4].info.id).toBe(messageIds[9]) // Last message is most recent + + // Delete message 3 (which would be in the next page, index 2) + await Session.removeMessage({ sessionID, messageID: messageIds[2] }) + + // Request next page with cursor from page1 + const page2 = await Session.messages({ sessionID, limit: 5, before: page1[0].info.id }) + + // Verify remaining messages are returned without error + // Should get 0, 1, 3, 4 (since 2 was deleted) = 4 messages + // OR 0, 1, 3, 4 + one more if available? No, limit applies to ID list which is stale? + // Storage.list is re-run, so index 2 is gone. + // IDs: [0, 1, 3, 4, 5, 6, 7, 8, 9] + // Cursor: before 5 (index 4 in new list) + // binaryLowerBound(5) -> index 4 + // start = 3 + // loop: 3, 2, 1, 0 -> IDs[3]=4, IDs[2]=3, IDs[1]=1, IDs[0]=0 + expect(page2.length).toBe(4) + expect(page2.map((m) => m.info.id)).toEqual([messageIds[0], messageIds[1], messageIds[3], messageIds[4]]) + }, + }) + }, + TEST_TIMEOUT_MS, + ) + + test( + "message IDs are lexicographically sorted (ULID invariant)", + async () => { + await using tmp = await tmpdir({ git: true }) + await Instance.provide({ + directory: tmp.path, + fn: async () => { + const session = await Session.create({}) + const messageIds: string[] = [] + + for (let i = 0; i < 5; i++) { + const msg = await Session.updateMessage({ + id: Identifier.ascending("message"), + role: "user", + sessionID: session.id, + agent: "default", + model: { providerID: "openai", modelID: "gpt-4" }, + time: { created: Date.now() }, + }) + messageIds.push(msg.id) + } + + // Verify IDs are lexicographically sorted (ULID invariant for binary search) + for (let i = 1; i < messageIds.length; i++) { + expect(messageIds[i] > messageIds[i - 1]).toBe(true) + } + }, + }) + }, + TEST_TIMEOUT_MS, + ) + + test( + "after cursor returns messages after cursor (ascending)", + async () => { + await using tmp = await tmpdir({ git: true }) + await Instance.provide({ + directory: tmp.path, + fn: async () => { + const session = await Session.create({}) + const sessionID = session.id + const messageIds: string[] = [] + + for (let i = 0; i < 10; i++) { + const msg = await Session.updateMessage({ + id: Identifier.ascending("message"), + role: "user", + sessionID, + agent: "default", + model: { providerID: "openai", modelID: "gpt-4" }, + time: { created: Date.now() }, + }) + messageIds.push(msg.id) + } + + // after=msg[2] should return msg[3], msg[4], msg[5] (limit 3) + const page1 = await Session.messages({ + sessionID, + limit: 3, + after: messageIds[2], + }) + expect(page1.length).toBe(3) + expect(page1[0].info.id).toBe(messageIds[3]) + expect(page1[1].info.id).toBe(messageIds[4]) + expect(page1[2].info.id).toBe(messageIds[5]) + + // after=msg[8] should return msg[9] only + const page2 = await Session.messages({ + sessionID, + limit: 3, + after: messageIds[8], + }) + expect(page2.length).toBe(1) + expect(page2[0].info.id).toBe(messageIds[9]) + + // after=msg[9] (last) should return empty + const page3 = await Session.messages({ + sessionID, + limit: 3, + after: messageIds[9], + }) + expect(page3.length).toBe(0) + }, + }) + }, + TEST_TIMEOUT_MS, + ) + + test( + "cannot specify both before and after cursors", + async () => { + await using tmp = await tmpdir({ git: true }) + await Instance.provide({ + directory: tmp.path, + fn: async () => { + const session = await Session.create({}) + const sessionID = session.id + + const msg = await Session.updateMessage({ + id: Identifier.ascending("message"), + role: "user", + sessionID, + agent: "default", + model: { providerID: "openai", modelID: "gpt-4" }, + time: { created: Date.now() }, + }) + + await expect( + Session.messages({ + sessionID, + limit: 3, + before: msg.id, + after: msg.id, + }), + ).rejects.toThrow("Cannot specify both") + }, + }) + }, + TEST_TIMEOUT_MS, + ) +}) diff --git a/packages/opencode/test/util/parse-link-header.test.ts b/packages/opencode/test/util/parse-link-header.test.ts new file mode 100644 index 00000000000..b1e316f2119 --- /dev/null +++ b/packages/opencode/test/util/parse-link-header.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, test } from "bun:test" +import { parseLinkHeader } from "../../src/util/link-header" + +describe("util.parseLinkHeader", () => { + test("returns empty object for empty string", () => { + expect(parseLinkHeader("")).toEqual({}) + }) + + test("returns empty object for undefined-ish input", () => { + expect(parseLinkHeader(undefined as unknown as string)).toEqual({}) + }) + + test("parses single link with rel", () => { + const header = '; rel="next"' + expect(parseLinkHeader(header)).toEqual({ + next: "https://api.example.com/items?page=2", + }) + }) + + test("parses multiple links", () => { + const header = + '; rel="prev", ; rel="next"' + expect(parseLinkHeader(header)).toEqual({ + prev: "https://api.example.com/items?page=1", + next: "https://api.example.com/items?page=3", + }) + }) + + test("handles unquoted rel values", () => { + const header = "; rel=next" + expect(parseLinkHeader(header)).toEqual({ + next: "https://example.com", + }) + }) + + test("handles single-quoted rel values", () => { + const header = "; rel='next'" + expect(parseLinkHeader(header)).toEqual({ + next: "https://example.com", + }) + }) + + test("ignores links without rel attribute", () => { + const header = "; type=text/html" + expect(parseLinkHeader(header)).toEqual({}) + }) + + test("handles malformed input gracefully", () => { + expect(parseLinkHeader("not a valid header")).toEqual({}) + expect(parseLinkHeader("<<>>;;")).toEqual({}) + }) + + test("handles extra whitespace", () => { + const header = ' ; rel="next" ' + expect(parseLinkHeader(header)).toEqual({ + next: "https://example.com", + }) + }) +}) diff --git a/packages/sdk/js/src/v2/gen/sdk.gen.ts b/packages/sdk/js/src/v2/gen/sdk.gen.ts index 706d0f9c227..b3e39ba8787 100644 --- a/packages/sdk/js/src/v2/gen/sdk.gen.ts +++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts @@ -1321,6 +1321,9 @@ export class Session extends HeyApiClient { sessionID: string directory?: string limit?: number + before?: string + after?: string + oldest?: boolean }, options?: Options, ) { @@ -1332,6 +1335,9 @@ export class Session extends HeyApiClient { { in: "path", key: "sessionID" }, { in: "query", key: "directory" }, { in: "query", key: "limit" }, + { in: "query", key: "before" }, + { in: "query", key: "after" }, + { in: "query", key: "oldest" }, ], }, ], diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index b7e72fbad8f..35b14555d7a 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -3187,6 +3187,9 @@ export type SessionMessagesData = { query?: { directory?: string limit?: number + before?: string + after?: string + oldest?: boolean } url: "/session/{sessionID}/message" } diff --git a/packages/sdk/openapi.json b/packages/sdk/openapi.json index c1be820f262..8bd295da646 100644 --- a/packages/sdk/openapi.json +++ b/packages/sdk/openapi.json @@ -2083,7 +2083,32 @@ "in": "query", "name": "limit", "schema": { - "type": "number" + "type": "integer", + "minimum": 1, + "maximum": 100 + } + }, + { + "in": "query", + "name": "before", + "schema": { + "type": "string", + "pattern": "^msg.*" + } + }, + { + "in": "query", + "name": "after", + "schema": { + "type": "string", + "pattern": "^msg.*" + } + }, + { + "in": "query", + "name": "oldest", + "schema": { + "type": "boolean" } } ], diff --git a/packages/util/src/binary.ts b/packages/util/src/binary.ts index 3d8f61851ae..048b6bc371b 100644 --- a/packages/util/src/binary.ts +++ b/packages/util/src/binary.ts @@ -38,4 +38,19 @@ export namespace Binary { array.splice(left, 0, item) return array } + + /** + * Find the first index where array[index] >= target (lower bound). + * For string arrays ordered lexicographically (e.g., ULIDs). + */ + export function lowerBound(array: string[], target: string): number { + let left = 0 + let right = array.length + while (left < right) { + const mid = (left + right) >>> 1 + if (array[mid] < target) left = mid + 1 + else right = mid + } + return left + } }