ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(app): hydrate timeline message parents (#35269)

Browse source
This commit is contained in:
Luke Parker 2026-07-05 10:49:46 +10:00 • committed by GitHub
commit a12d50e15a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 984 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

308
packages/app/src/context/server-session.test.ts
View file

@ -15,11 +15,13 @@ const session = (id: string, parentID?: string): Session => ({
})
type UserMessage = Extract<Message, { role: "user" }>
type AssistantMessage = Extract<Message, { role: "assistant" }>
type TextPart = Extract<Part, { type: "text" }>
type MessageResponse = {
data: { info: Message; parts: Part[] }[]
response: { headers: Headers }
}
type SingleMessageResponse = { data: MessageResponse["data"][number] }
const userMessage = (id: string, input: Partial<UserMessage> = {}): UserMessage => ({
id,
@ -31,6 +33,22 @@ const userMessage = (id: string, input: Partial<UserMessage> = {}): UserMessage
...input,
})
const assistantMessage = (id: string, parentID: string, input: Partial<AssistantMessage> = {}): AssistantMessage => ({
id,
sessionID: "child",
role: "assistant",
time: { created: Number(id.at(-1)), completed: Number(id.at(-1)) },
parentID,
modelID: "model",
providerID: "provider",
mode: "build",
agent: "build",
path: { cwd: "/repo", root: "/repo" },
cost: 0,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
...input,
})
const textPart = (messageID: string, input: Partial<TextPart> = {}): TextPart => ({
id: "part",
sessionID: "child",
@ -45,6 +63,8 @@ const response = (data: MessageResponse["data"] = [], cursor?: string): MessageR
response: { headers: new Headers(cursor ? { "x-next-cursor": cursor } : undefined) },
})
const singleResponse = (info: Message, parts: Part[] = []): SingleMessageResponse => ({ data: { info, parts } })
const deferredResponse = () => Promise.withResolvers<MessageResponse>()
function messageClient(...responses: Array<MessageResponse | Promise<MessageResponse>>) {
@ -71,6 +91,40 @@ function messageClient(...responses: Array<MessageResponse | Promise<MessageResp
})
}
function rootMessageClient(
pages: Array<MessageResponse | Promise<MessageResponse>>,
roots: Array<SingleMessageResponse | Promise<SingleMessageResponse>>,
) {
let pageIndex = 0
let rootIndex = 0
const requests: unknown[] = []
const rootRequests: unknown[] = []
const rootWaiting = new Map<number, () => void>()
const client = {
session: {
get: async () => ({ data: session("child", "root") }),
messages: (input: unknown) => {
requests.push(input)
return pages[pageIndex++]
},
message: (input: unknown) => {
rootRequests.push(input)
rootWaiting.get(rootRequests.length)?.()
rootWaiting.delete(rootRequests.length)
return roots[rootIndex++]
},
},
} as unknown as OpencodeClient
return Object.assign(client, {
requests,
rootRequests,
rootRequested(count: number) {
if (rootRequests.length >= count) return Promise.resolve()
return new Promise<void>((resolve) => rootWaiting.set(count, resolve))
},
})
}
const retryImmediately: typeof retry = async (task, options = {}) => {
const attempts = options.attempts ?? 3
for (let attempt = 0; ; attempt++) {
@ -124,6 +178,240 @@ describe("server session", () => {
expect(ctx.store.data.message.root).toEqual([])
})
test("backfills an assistant-only initial page through its user root", async () => {
const user = userMessage("message-1")
const assistants = [assistantMessage("message-2", user.id), assistantMessage("message-3", user.id)]
const client = rootMessageClient(
[
response(
assistants.map((info) => ({ info, parts: [] })),
"older",
),
],
[singleResponse(user)],
)
const store = createServerSession(client)
await store.sync("child")
expect(client.requests).toEqual([
{ sessionID: "child", limit: 2, before: undefined },
])
expect(client.rootRequests).toEqual([{ sessionID: "child", messageID: user.id }])
expect(store.data.message.child).toEqual([user, ...assistants])
expect(store.history.more("child")).toBe(true)
})
test("does not let an optimistic user suppress initial root backfill", async () => {
const user = userMessage("message-1")
const part = textPart(user.id)
const assistants = [assistantMessage("message-2", user.id), assistantMessage("message-3", user.id)]
const client = rootMessageClient(
[
response(
assistants.map((info) => ({ info, parts: [] })),
"older",
),
],
[singleResponse(user)],
)
const store = createServerSession(client)
store.optimistic.add({ sessionID: "child", message: user, parts: [part] })
await store.sync("child")
store.optimistic.remove({ sessionID: "child", messageID: user.id })
expect(client.requests).toHaveLength(1)
expect(client.rootRequests).toHaveLength(1)
expect(store.data.message.child).toEqual([user, ...assistants])
})
test("backfills the parent of fetched assistants when another user is cached", async () => {
const unrelated = userMessage("message-0", { time: { created: 0 } })
const user = userMessage("message-1")
const assistants = [assistantMessage("message-2", user.id), assistantMessage("message-3", user.id)]
const client = rootMessageClient(
[
response([{ info: unrelated, parts: [] }]),
response(
assistants.map((info) => ({ info, parts: [] })),
"older",
),
],
[singleResponse(user)],
)
const store = createServerSession(client)
await store.sync("child")
await store.sync("child", { force: true })
expect(client.requests).toHaveLength(2)
expect(client.rootRequests).toHaveLength(1)
expect(store.data.message.child).toEqual([unrelated, user, ...assistants])
})
test("preserves cached history between an injected parent and the page boundary", async () => {
const user = userMessage("message-1")
const cached = userMessage("message-3", { time: { created: 3 } })
const assistant = assistantMessage("message-4", user.id)
const client = rootMessageClient(
[response([{ info: cached, parts: [] }]), response([{ info: assistant, parts: [] }], "older")],
[singleResponse(user)],
)
const store = createServerSession(client)
await store.sync("child")
await store.sync("child", { force: true })
expect(store.data.message.child).toEqual([user, cached, assistant])
})
test("refreshes a cached parent omitted by an assistant-only replacement page", async () => {
const stale = userMessage("message-1", { summary: { title: "stale", diffs: [] } })
const fresh = { ...stale, summary: { title: "fresh", diffs: [] } }
const stalePart = textPart(stale.id, { text: "stale" })
const freshPart = { ...stalePart, text: "fresh" }
const assistant = assistantMessage("message-2", stale.id)
const client = rootMessageClient(
[response([{ info: stale, parts: [stalePart] }]), response([{ info: assistant, parts: [] }], "older")],
[singleResponse(fresh, [freshPart])],
)
const store = createServerSession(client)
await store.sync("child")
await store.sync("child", { force: true })
expect(client.rootRequests).toEqual([{ sessionID: "child", messageID: stale.id }])
expect(store.data.message.child).toEqual([fresh, assistant])
expect(store.data.part[stale.id]).toEqual([freshPart])
})
test("refreshes a confirmed optimistic parent while preserving pending parts", async () => {
const stale = userMessage("message-1", { summary: { title: "stale", diffs: [] } })
const fresh = { ...stale, summary: { title: "fresh", diffs: [] } }
const confirmed = textPart(stale.id, { id: "confirmed", text: "stale" })
const refreshed = { ...confirmed, text: "fresh" }
const pending = textPart(stale.id, { id: "pending", text: "pending" })
const assistant = assistantMessage("message-2", stale.id)
const client = rootMessageClient(
[response([{ info: stale, parts: [confirmed] }]), response([{ info: assistant, parts: [] }], "older")],
[singleResponse(fresh, [refreshed])],
)
const store = createServerSession(client)
store.optimistic.add({ sessionID: "child", message: stale, parts: [confirmed, pending] })
await store.sync("child")
await store.sync("child", { force: true })
expect(client.rootRequests).toEqual([{ sessionID: "child", messageID: stale.id }])
expect(store.data.message.child).toEqual([fresh, assistant])
expect(store.data.part[stale.id]).toEqual([refreshed, pending])
})
test("uses a parent received by SSE during the replacement load", async () => {
const pending = deferredResponse()
const user = userMessage("message-1")
const assistant = assistantMessage("message-2", user.id)
const client = rootMessageClient([pending.promise], [])
const store = createServerSession(client)
const loading = store.sync("child")
store.apply({ type: "message.updated", properties: { info: user } })
pending.resolve(response([{ info: assistant, parts: [] }], "older"))
await loading
expect(client.rootRequests).toEqual([])
expect(store.data.message.child).toEqual([user, assistant])
})
test("uses a successful retry over events received by a failed backfill attempt", async () => {
const failed = deferredResponse()
const user = userMessage("message-1")
const live = { ...user, agent: "stale" }
const assistants = [assistantMessage("message-2", user.id), assistantMessage("message-3", user.id)]
const client = rootMessageClient(
[
response(
assistants.map((info) => ({ info, parts: [] })),
"older",
),
],
[failed.promise.then((result) => ({ data: result.data[0]! })), singleResponse(user)],
)
const store = createServerSession(client, { retry: retryImmediately })
const loading = store.sync("child")
await client.rootRequested(1)
store.apply({ type: "message.updated", properties: { info: live } })
failed.reject(new Error("retry"))
await loading
expect(client.requests).toHaveLength(1)
expect(client.rootRequests).toHaveLength(2)
expect(store.data.message.child).toEqual([user, ...assistants])
})
test("preserves newer-page events across a failed parent retry", async () => {
const failed = deferredResponse()
const user = userMessage("message-1")
const assistant = assistantMessage("message-2", user.id)
const live = { ...assistant, cost: 1 }
const client = rootMessageClient(
[response([{ info: assistant, parts: [] }], "older")],
[failed.promise.then((result) => ({ data: result.data[0]! })), singleResponse(user)],
)
const store = createServerSession(client, { retry: retryImmediately })
const loading = store.sync("child")
await client.rootRequested(1)
store.apply({ type: "message.updated", properties: { info: live } })
failed.reject(new Error("retry"))
await loading
expect(store.data.message.child).toEqual([user, live])
})
test("preserves unrelated message events across a failed parent retry", async () => {
const failed = deferredResponse()
const user = userMessage("message-1")
const assistant = assistantMessage("message-2", user.id)
const live = userMessage("message-4", { time: { created: 4 } })
const client = rootMessageClient(
[response([{ info: assistant, parts: [] }], "older")],
[failed.promise.then((result) => ({ data: result.data[0]! })), singleResponse(user)],
)
const store = createServerSession(client, { retry: retryImmediately })
const loading = store.sync("child")
await client.rootRequested(1)
store.apply({ type: "message.updated", properties: { info: live } })
failed.reject(new Error("retry"))
await loading
expect(store.data.message.child).toEqual([user, assistant, live])
})
test("preserves newer-page part events across a failed parent retry", async () => {
const failed = deferredResponse()
const user = userMessage("message-1")
const assistant = assistantMessage("message-2", user.id)
const stale = textPart(assistant.id, { text: "stale" })
const live = { ...stale, text: "live" }
const client = rootMessageClient(
[response([{ info: assistant, parts: [stale] }], "older")],
[failed.promise.then((result) => ({ data: result.data[0]! })), singleResponse(user)],
)
const store = createServerSession(client, { retry: retryImmediately })
const loading = store.sync("child")
await client.rootRequested(1)
store.apply({ type: "message.part.updated", properties: { sessionID: "child", part: live, time: 2 } })
failed.reject(new Error("retry"))
await loading
expect(store.data.part[assistant.id]).toEqual([live])
})
test("merges live events into the initial page", async () => {
const pending = deferredResponse()
const user = userMessage("message-1")
@ -905,6 +1193,26 @@ describe("server session", () => {
expect(store.data.message.child).toEqual([latest])
})
test("does not scan cached messages for user roots during history prepend", async () => {
const guard = { active: false }
const latest = new Proxy(userMessage("message-2", { time: { created: 2 } }), {
get(target, property, receiver) {
if (guard.active && property === "role") throw new Error("cached role accessed")
return Reflect.get(target, property, receiver)
},
})
const older = userMessage("message-1")
const store = createServerSession(
messageClient(response([{ info: latest, parts: [] }], "older"), response([{ info: older, parts: [] }])),
)
await store.sync("child")
guard.active = true
await store.history.loadMore("child")
expect(store.data.message.child).toEqual([older, latest])
})
test("preserves loaded history during an incomplete refresh", async () => {
const older = userMessage("message-1")
const latest = userMessage("message-2", { time: { created: 2 } })

139
packages/app/src/context/server-session.ts
View file

@ -54,6 +54,11 @@ type MessageLoadState = {
clearedMessageParts: Set<string>
}
type MessageLoadBaseline = Pick<
MessageLoadState,
"touchedMessages" | "retainedMessages" | "touchedParts" | "clearedMessageParts"
>
function mergeOptimisticPage(page: MessagePage, items: OptimisticItem[]) {
if (items.length === 0) return { ...page, observed: [] as { messageID: string; parts: Part[] }[] }
const session = [...page.session]
@ -347,7 +352,7 @@ export function createServerSession(client: OpencodeClient, options?: { retry?:
load.touchedParts.set(messageID, new Set([partID]))
}
const resetMessageLoad = (sessionID: string, load: MessageLoadState) => {
const resetMessageLoad = (sessionID: string, load: MessageLoadState, baseline?: MessageLoadBaseline) => {
load.touchedMessages.clear()
load.retainedMessages.clear()
load.touchedParts.clear()
@ -380,8 +385,27 @@ export function createServerSession(client: OpencodeClient, options?: { retry?:
parts.forEach((partID) => touched.add(partID))
load.touchedParts.set(messageID, touched)
}
baseline?.touchedMessages.forEach((messageID) => load.touchedMessages.add(messageID))
baseline?.retainedMessages.forEach((messageID) => load.retainedMessages.add(messageID))
baseline?.clearedMessageParts.forEach((messageID) => load.clearedMessageParts.add(messageID))
baseline?.touchedParts.forEach((parts, messageID) => {
const touched = load.touchedParts.get(messageID) ?? new Set<string>()
parts.forEach((partID) => touched.add(partID))
load.touchedParts.set(messageID, touched)
})
}
const messageLoadBaseline = (load: MessageLoadState, exclude: string): MessageLoadBaseline => ({
touchedMessages: new Set([...load.touchedMessages].filter((messageID) => messageID !== exclude)),
retainedMessages: new Set([...load.retainedMessages].filter((messageID) => messageID !== exclude)),
touchedParts: new Map(
[...load.touchedParts]
.filter(([messageID]) => messageID !== exclude)
.map(([messageID, parts]) => [messageID, new Set(parts)]),
),
clearedMessageParts: new Set([...load.clearedMessageParts].filter((messageID) => messageID !== exclude)),
})
const evict = (sessionIDs: string[]) => {
if (sessionIDs.length === 0) return
const evicted = new Set(sessionIDs)
@ -460,6 +484,18 @@ export function createServerSession(client: OpencodeClient, options?: { retry?:
}
}
const fetchMessage = async (sessionID: string, messageID: string, onAttempt?: () => void) => {
const response = await (options?.retry ?? retry)(() => {
onAttempt?.()
return client.session.message({ sessionID, messageID })
})
if (!response.data?.info?.id) throw new Error(`Message not found: ${messageID}`)
return {
message: cleanMessage(response.data.info),
parts: response.data.parts.filter((part) => !!part?.id).sort((a, b) => cmp(a.id, b.id)),
}
}
const replaceMessages = (sessionID: string, messages: Message[]) => {
const messageIDs = new Set(messages.map((message) => message.id))
const dropped = (data.message[sessionID] ?? []).filter((message) => !messageIDs.has(message.id))
@ -578,36 +614,79 @@ export function createServerSession(client: OpencodeClient, options?: { retry?:
messageLoads.set(sessionID, load)
setMeta("loading", sessionID, true)
let applied = false
await fetchMessages(sessionID, limit, before, () => resetMessageLoad(sessionID, load))
.then((page) => {
if (generations.get(sessionID) !== active) return
const first = page.session.reduce<Message | undefined>(
(oldest, message) => (!oldest || cmpMessage(message, oldest) < 0 ? message : oldest),
undefined,
)
const preserveUnfetched =
mode === "prepend" || (!page.complete && (!first || ((message: Message) => cmpMessage(message, first) < 0)))
applyMessagePage(
sessionID,
page,
messageLoads.get(sessionID) === load ? load : undefined,
preserveUnfetched,
mode !== "prepend",
)
applied = true
})
.finally(() => {
if (!applied && generations.get(sessionID) === active && messageLoads.get(sessionID) === load) {
for (const messageID of load.orphanParents) {
if (!orphanParts.get(sessionID)?.has(messageID)) continue
setData(produce((draft) => deleteMessageParts(draft, messageID)))
orphanParts.get(sessionID)?.delete(messageID)
}
if (orphanParts.get(sessionID)?.size === 0) orphanParts.delete(sessionID)
try {
const page = await fetchMessages(sessionID, limit, before, () => resetMessageLoad(sessionID, load))
const first = page.session.reduce<Message | undefined>(
(oldest, message) => (!oldest || cmpMessage(message, oldest) < 0 ? message : oldest),
undefined,
)
if (generations.get(sessionID) !== active) return
const parents = [] as Awaited<ReturnType<typeof fetchMessage>>[]
if (mode !== "prepend") {
const users = new Set([
...page.session.filter((message) => message.role === "user").map((message) => message.id),
...(data.message[sessionID] ?? [])
.filter((message) => {
if (message.role !== "user") return false
const item = optimistic.get(sessionID)?.get(message.id)
return load.touchedMessages.has(message.id) && (!item || item.confirmedMessage === true)
})
.map((message) => message.id),
])
const parentIDs = [
...new Set(
page.session.flatMap((message) =>
message.role === "assistant" && !users.has(message.parentID) ? [message.parentID] : [],
),
),
]
for (const parentID of parentIDs) {
if (generations.get(sessionID) !== active) break
const parent = await fetchMessage(sessionID, parentID, () =>
resetMessageLoad(sessionID, load, messageLoadBaseline(load, parentID)),
)
if (parent.message.role !== "user") throw new Error(`Assistant parent is not a user message: ${parentID}`)
parents.push(parent)
}
if (messageLoads.get(sessionID) === load) messageLoads.delete(sessionID)
if (generations.get(sessionID) === active) setMeta("loading", sessionID, false)
})
}
if (generations.get(sessionID) !== active) return
const result =
mode === "prepend"
? page
: {
...page,
session: merge(
page.session,
parents.map((parent) => parent.message),
),
part: merge(
page.part,
parents.map((parent) => ({ id: parent.message.id, part: parent.parts })),
),
}
const preserveUnfetched =
mode === "prepend" || (!result.complete && (!first || ((message: Message) => cmpMessage(message, first) < 0)))
applyMessagePage(
sessionID,
result,
messageLoads.get(sessionID) === load ? load : undefined,
preserveUnfetched,
mode !== "prepend",
)
applied = true
} finally {
if (!applied && generations.get(sessionID) === active && messageLoads.get(sessionID) === load) {
for (const messageID of load.orphanParents) {
if (!orphanParts.get(sessionID)?.has(messageID)) continue
setData(produce((draft) => deleteMessageParts(draft, messageID)))
orphanParts.get(sessionID)?.delete(messageID)
}
if (orphanParts.get(sessionID)?.size === 0) orphanParts.delete(sessionID)
}
if (messageLoads.get(sessionID) === load) messageLoads.delete(sessionID)
if (generations.get(sessionID) === active) setMeta("loading", sessionID, false)
}
}
const sync = (sessionID: string, options?: { force?: boolean; messageLimit?: number }) => {

8
packages/app/src/pages/session/timeline/model.test.ts
View file

@ -1,6 +1,6 @@
import { describe, expect, test } from "bun:test"
import type { AssistantMessage, Message, UserMessage } from "@opencode-ai/sdk/v2"
import { loadOlderTimeline, selectUserMessages, selectVisibleUserMessages } from "./model"
import { isTimelineReady, loadOlderTimeline, selectUserMessages, selectVisibleUserMessages } from "./model"
const user = (id: string) => ({ id, role: "user" }) as UserMessage
const assistant = (id: string) => ({ id, role: "assistant" }) as AssistantMessage
@ -15,6 +15,12 @@ describe("timeline model", () => {
expect(selectVisibleUserMessages(users)).toBe(users)
})
test("waits for an assistant-only load to hydrate its user root", () => {
expect(isTimelineReady([assistant("msg_2")], true)).toBe(false)
expect(isTimelineReady([user("msg_1"), assistant("msg_2")], true)).toBe(true)
expect(isTimelineReady([], false)).toBe(true)
})
test("loads exactly one opaque cursor page", async () => {
let calls = 0
const anchors: Array<string | boolean> = []

6
packages/app/src/pages/session/timeline/model.ts
View file

@ -45,7 +45,7 @@ export function createTimelineModel(input: {
})
const ready = createMemo(() => {
const id = input.sessionID()
return !id || sync().data.message[id] !== undefined
return !id || isTimelineReady(sync().data.message[id], serverSync().session.history.loading(id))
})
const userMessages = createMemo(() => selectUserMessages(messages()), emptyUserMessages, { equals: same })
const visibleUserMessages = createMemo(
@ -98,6 +98,10 @@ export function selectUserMessages(messages: Message[]) {
return messages.filter((message): message is UserMessage => message.role === "user")
}
export function isTimelineReady(messages: Message[] | undefined, loading: boolean) {
return messages !== undefined && (messages.some((message) => message.role === "user") || !loading)
}
export function selectVisibleUserMessages(messages: UserMessage[], revertMessageID?: string) {
if (!revertMessageID) return messages
return messages.filter((message) => message.id < revertMessageID)