diff --git a/packages/app/e2e/regression/session-timeline-history-root.spec.ts b/packages/app/e2e/regression/session-timeline-history-root.spec.ts index d64a8cddf6..f892cb6f3f 100644 --- a/packages/app/e2e/regression/session-timeline-history-root.spec.ts +++ b/packages/app/e2e/regression/session-timeline-history-root.spec.ts @@ -16,9 +16,8 @@ import { mockOpenCodeServer } from "../utils/mock-server" import { installSseTransport } from "../utils/sse-transport" import { expectSessionTitle } from "../utils/waits" -const initialPageSize = 20 -const historyPageSize = 200 -const messages = Array.from({ length: initialPageSize + 1 }, (_, index) => { +const messagePageSize = 200 +const messages = Array.from({ length: messagePageSize / 2 + 1 }, (_, index) => { const id = `msg_${String(index + 1001).padStart(4, "0")}_history_root_user` return [ userMessage(undefined, { id, created: 1700000000000 + index * 2_000 }), @@ -26,7 +25,7 @@ const messages = Array.from({ length: initialPageSize + 1 }, (_, index) => { id: `msg_${String(index + 1001).padStart(4, "0")}_history_root_assistant`, parentID: id, created: 1700000001000 + index * 2_000, - completed: index < initialPageSize, + completed: index < messagePageSize / 2, }), ] }).flat() @@ -160,21 +159,18 @@ for (const scenario of scenarios) { await expect(page.locator(`[data-timeline-part-id="${userPartID}"]`)).toBeVisible() const viewport = page.locator(".scroll-view__viewport", { has: page.locator("[data-timeline-row]") }) await viewport.hover() - const deadline = Date.now() + 10_000 + const deadline = Date.now() + 30_000 while (requests.filter((request) => request.phase === "start").length < 2) { if (Date.now() >= deadline) throw new Error("Timed out scrolling to the history boundary") - await page.mouse.wheel(0, -240) + await page.mouse.wheel(0, -1_200) await page.waitForTimeout(20) } expect(requests.filter((request) => request.phase === "end")).toHaveLength(1) expect(sequence.slice(0, 3)).toEqual([ "messages:start:latest", "messages:end:latest", - `messages:start:${messages.at(-initialPageSize)!.info.id}`, + `messages:start:${messages.at(-messagePageSize)!.info.id}`, ]) - await expect(page.locator('[data-timeline-part-id*="_history_root_assistant:text:0"]')).toHaveCount( - initialPageSize / 2, - ) await page.evaluate(() => { ;( window as Window & { @@ -186,15 +182,12 @@ for (const scenario of scenarios) { expect(await visibleContentHidden(page)).toBe(false) const beforeHistory = await probeSamples(page) history.resolve() - await expect - .poll(() => page.locator('[data-timeline-part-id*="_history_root_assistant:text:0"]').count()) - .toBeGreaterThan(initialPageSize / 2) await expect.poll(() => requests.filter((request) => request.phase === "end").length).toBe(2) await expect(page.getByRole("button", { name: "Stop" })).toBeVisible() await waitForProbeSamples(page, beforeHistory) expect(pages).toEqual([ - { before: undefined, limit: initialPageSize }, - { before: messages.at(-initialPageSize)!.info.id, limit: historyPageSize }, + { before: undefined, limit: messagePageSize }, + { before: messages.at(-messagePageSize)!.info.id, limit: messagePageSize }, ]) expect(roots).toEqual([]) diff --git a/packages/app/e2e/smoke/session-timeline.fixture.ts b/packages/app/e2e/smoke/session-timeline.fixture.ts index 6da01458a3..2ad400576f 100644 --- a/packages/app/e2e/smoke/session-timeline.fixture.ts +++ b/packages/app/e2e/smoke/session-timeline.fixture.ts @@ -222,7 +222,7 @@ function turn(index: number): Message[] { return [user, assistantMessage(targetID, index, user.info.id, parts)] } -const targetMessages = Array.from({ length: 72 }, (_, index) => turn(index)).flat() +const targetMessages = Array.from({ length: 101 }, (_, index) => turn(index)).flat() const sourceMessages = Array.from({ length: 12 }, (_, index) => [ userMessage(sourceID, index + 1000, 120), assistantMessage(sourceID, index + 1000, id("msg_user", index + 1000), [textPart(index + 1000, 0, 240)]), diff --git a/packages/app/e2e/smoke/session-timeline.spec.ts b/packages/app/e2e/smoke/session-timeline.spec.ts index 7e4949a320..d1cf5d5bf9 100644 --- a/packages/app/e2e/smoke/session-timeline.spec.ts +++ b/packages/app/e2e/smoke/session-timeline.spec.ts @@ -727,7 +727,7 @@ function expectCompleteScroll( ).toEqual([]) expect(new Set(expectedPartIDs).size).toBe(expectedPartIDs.length) expect(new Set(expectedMessageIDs).size).toBe(expectedMessageIDs.length) - expect(expectedPartIDs.length).toBe(331) + expect(expectedPartIDs.length).toBe(465) } async function selectHomeProject(page: Page, projectName: string) { diff --git a/packages/app/src/context/server-session.test.ts b/packages/app/src/context/server-session.test.ts index c3883bba0b..263e2acb44 100644 --- a/packages/app/src/context/server-session.test.ts +++ b/packages/app/src/context/server-session.test.ts @@ -618,7 +618,7 @@ describe("server session", () => { await ctx.store.sync("root") expect(ctx.get).toEqual([{ sessionID: "root" }]) - expect(ctx.messages).toEqual([{ sessionID: "root", limit: 20, order: "desc" }]) + expect(ctx.messages).toEqual([{ sessionID: "root", limit: 200, order: "desc" }]) expect(ctx.store.data.message.root).toEqual([]) }) @@ -629,8 +629,32 @@ describe("server session", () => { ctx.store.invalidate() await ctx.store.sync("root") + expect(ctx.store.data.message.root).toEqual([]) expect(ctx.get).toHaveLength(2) - expect(ctx.messages).toHaveLength(2) + expect(ctx.messages).toEqual([ + { sessionID: "root", limit: 200, order: "desc" }, + { sessionID: "root", limit: 200, order: "desc" }, + ]) + }) + + test("keeps a fixed page size after the local message cache exceeds the API limit", async () => { + const client = messageClient(response(), response()) + const store = createServerSession(client) + await store.sync("child") + Array.from({ length: 428 }, (_, index) => + store.apply({ + type: "message.updated", + properties: { info: userMessage(`message-${index}`, { time: { created: index } }) }, + }), + ) + + expect(store.data.message.child).toHaveLength(428) + await store.sync("child", { force: true }) + + expect(client.requests).toEqual([ + { sessionID: "child", limit: 200, order: "desc" }, + { sessionID: "child", limit: 200, order: "desc" }, + ]) }) test("loads current session content through the current message API", async () => { @@ -655,7 +679,7 @@ describe("server session", () => { await store.sync("root") - expect(requests).toEqual([{ sessionID: "root", limit: 20, order: "desc" }]) + expect(requests).toEqual([{ sessionID: "root", limit: 200, order: "desc" }]) expect(store.data.session_message.root.map((message) => message.id)).toEqual([user.id, assistant.id]) expect(store.data.message.root.map((message) => message.id)).toEqual([user.id, assistant.id]) }) @@ -731,8 +755,8 @@ describe("server session", () => { await store.sync("root") expect(requests).toEqual([ - { sessionID: "root", limit: 20, order: "desc" }, - { sessionID: "root", limit: 20, cursor: "older" }, + { sessionID: "root", limit: 200, order: "desc" }, + { sessionID: "root", limit: 200, cursor: "older" }, ]) expect(store.data.message.root.map((message) => message.id)).toEqual([ user.id, @@ -741,6 +765,26 @@ describe("server session", () => { expect(assistants.map((item) => store.data.part[item.id]?.[0]?.type)).toEqual(["text", "text", "text"]) }) + test("loads older messages by cursor with the fixed page size", async () => { + const older = userMessage("message-1") + const latest = userMessage("message-2", { time: { created: 2 } }) + const client = messageClient( + response([{ info: latest, parts: [] }], "older"), + response([{ info: older, parts: [] }]), + ) + const store = createServerSession(client) + await store.sync("child") + + await store.history.loadMore("child") + + expect(client.requests).toEqual([ + { sessionID: "child", limit: 200, order: "desc" }, + { sessionID: "child", limit: 200, cursor: "older" }, + ]) + expect(store.data.message.child).toEqual([older, latest]) + expect(store.history.more("child")).toBe(false) + }) + // V2 messages are ordered projections and do not expose V1 assistant parent IDs. describe.skip("V1 assistant parent projections", () => { test("backfills an assistant-only initial page through its user root", async () => { @@ -754,7 +798,7 @@ describe("server session", () => { await store.sync("child") - expect(client.requests).toEqual([{ sessionID: "child", limit: 20, order: "desc" }]) + expect(client.requests).toEqual([{ sessionID: "child", limit: 200, order: "desc" }]) expect(client.rootRequests).toEqual([{ sessionID: "child", messageID: user.id }]) expect(store.data.message.child).toEqual([user, ...assistants]) expect(store.history.more("child")).toBe(false) diff --git a/packages/app/src/context/server-session.ts b/packages/app/src/context/server-session.ts index 089615fc9e..11eadf2201 100644 --- a/packages/app/src/context/server-session.ts +++ b/packages/app/src/context/server-session.ts @@ -30,8 +30,7 @@ type MessageApi = ServerApi["message"] const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0) const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) -const initialMessagePageSize = 20 -const historyMessagePageSize = 200 +const messagePageSize = 200 const sessionInfoLimit = 2_048 const emptyIDs: ReadonlySet = new Set() @@ -242,7 +241,6 @@ export function createServerSession( return created } const [meta, setMeta] = createStore({ - limit: {} as Record, cursor: {} as Record, complete: {} as Record, loading: {} as Record, @@ -423,7 +421,6 @@ export function createServerSession( setMeta( produce((draft) => { for (const sessionID of sessionIDs) { - delete draft.limit[sessionID] delete draft.cursor[sessionID] delete draft.complete[sessionID] delete draft.loading[sessionID] @@ -457,11 +454,15 @@ export function createServerSession( pickSessionCacheEvictions({ seen, keep: sessionID, limit: SESSION_CACHE_LIMIT, preserve: protectedSessions() }), ) - const fetchMessages = async (sessionID: string, limit: number, before?: string, onAttempt?: () => void) => { + const fetchMessages = async (sessionID: string, before?: string, onAttempt?: () => void) => { const request = (cursor?: string) => (options?.retry ?? retry)(() => { onAttempt?.() - return messageApi.list(cursor ? { sessionID, limit, cursor } : { sessionID, limit, order: "desc" }) + return messageApi.list( + cursor + ? { sessionID, limit: messagePageSize, cursor } + : { sessionID, limit: messagePageSize, order: "desc" }, + ) }) const first = await request(before) const pages = [first] @@ -632,14 +633,13 @@ export function createServerSession( } orphanParts.delete(sessionID) } - setMeta("limit", sessionID, messages.length) setMeta("cursor", sessionID, merged.cursor) setMeta("complete", sessionID, merged.complete) setMeta("at", sessionID, Date.now()) }) } - const loadMessages = async (sessionID: string, limit: number, before?: string, mode?: "replace" | "prepend") => { + const loadMessages = async (sessionID: string, before?: string, mode?: "replace" | "prepend") => { if (meta.loading[sessionID]) return const active = generation(sessionID) const load: MessageLoadState = { @@ -658,7 +658,7 @@ export function createServerSession( setMeta("loading", sessionID, true) let applied = false try { - const page = await fetchMessages(sessionID, limit, before, () => resetMessageLoad(sessionID, load)) + const page = await fetchMessages(sessionID, before, () => resetMessageLoad(sessionID, load)) const first = page.session.reduce( (oldest, message) => (!oldest || compareMessages(message, oldest) < 0 ? message : oldest), undefined, @@ -737,32 +737,30 @@ export function createServerSession( } } - const sync = (sessionID: string, options?: { force?: boolean; messageLimit?: number }) => { + const sync = (sessionID: string, options?: { force?: boolean }) => { touch(sessionID) return runInflight(inflight, sessionID, async () => { - const cached = data.message[sessionID] !== undefined && meta.limit[sessionID] !== undefined + const cached = data.message[sessionID] !== undefined && meta.complete[sessionID] !== undefined const invalid = invalidated.has(sessionID) const revision = invalidationRevision if (cached && data.info[sessionID] && !invalid && !options?.force) return await Promise.all([ resolve(sessionID, invalid ? { ...options, force: true } : options), - cached && !invalid && !options?.force - ? Promise.resolve() - : loadMessages(sessionID, options?.messageLimit ?? meta.limit[sessionID] ?? initialMessagePageSize), + cached && !invalid && !options?.force ? Promise.resolve() : loadMessages(sessionID), ]) if (invalid && invalidationRevision === revision) invalidated.delete(sessionID) }) } - const prefetch = async (sessionID: string, limit: number) => { + const prefetch = async (sessionID: string, messageCount: number) => { touch(sessionID) await inflight.get(sessionID) if ( Date.now() - (meta.at[sessionID] ?? 0) <= 15_000 && - (meta.complete[sessionID] || (data.message[sessionID]?.length ?? 0) >= limit) + (meta.complete[sessionID] || (data.message[sessionID]?.length ?? 0) >= messageCount) ) return - await runInflight(inflight, sessionID, () => loadMessages(sessionID, limit)) + await runInflight(inflight, sessionID, () => loadMessages(sessionID)) } const eventSessionID = (event: { type: string; properties?: unknown }) => { @@ -1354,11 +1352,11 @@ export function createServerSession( setMeta("at", {}) }, prefetch, - shouldPrefetch(sessionID: string, limit: number) { + shouldPrefetch(sessionID: string, messageCount: number) { if (data.message[sessionID] === undefined) return true if (Date.now() - (meta.at[sessionID] ?? 0) > 15_000) return true if (meta.complete[sessionID]) return false - return (meta.limit[sessionID] ?? 0) <= limit + return (data.message[sessionID]?.length ?? 0) <= messageCount }, fresh(sessionID: string, ttl: number) { return Date.now() - (meta.at[sessionID] ?? 0) <= ttl @@ -1440,14 +1438,14 @@ export function createServerSession( history: { more: (sessionID: string) => data.message[sessionID] !== undefined && - meta.limit[sessionID] !== undefined && + meta.complete[sessionID] !== undefined && !meta.complete[sessionID] && !!meta.cursor[sessionID], loading: (sessionID: string) => meta.loading[sessionID] ?? false, - async loadMore(sessionID: string, count = historyMessagePageSize) { + async loadMore(sessionID: string) { touch(sessionID) if (meta.loading[sessionID] || meta.complete[sessionID] || !meta.cursor[sessionID]) return - await loadMessages(sessionID, count, meta.cursor[sessionID], "prepend") + await loadMessages(sessionID, meta.cursor[sessionID], "prepend") }, }, evict(sessionID: string) {