Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 89 Packages Projects Releases Wiki Activity

fix(app): use fixed message page size (#42743)

Browse Source
This commit is contained in:
Luke Parker
2026-08-15 22:12:20 +10:00
committed by GitHub
parent 79fc74afbf
commit f4baba2824
5 changed files with 80 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/e2e/regression/session-timeline-history-root.spec.ts
+8 -15
View File
@@ -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([])
packages/app/e2e/smoke/session-timeline.fixture.ts
+1 -1
View File
@@ -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)]),
packages/app/e2e/smoke/session-timeline.spec.ts
+1 -1
View File
@@ -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) {
packages/app/src/context/server-session.test.ts
+50 -6
View File
@@ -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)
packages/app/src/context/server-session.ts
+20 -22
View File
@@ -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<string> = new Set()
@@ -242,7 +241,6 @@ export function createServerSession(
return created
}
const [meta, setMeta] = createStore({
limit: {} as Record<string, number | undefined>,
cursor: {} as Record<string, string | undefined>,
complete: {} as Record<string, boolean | undefined>,
loading: {} as Record<string, boolean | undefined>,
@@ -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<Message | undefined>(
(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) {