diff --git a/packages/app/src/context/global-sync/event-reducer.test.ts b/packages/app/src/context/global-sync/event-reducer.test.ts index b53fb691b3c..06d536618e0 100644 --- a/packages/app/src/context/global-sync/event-reducer.test.ts +++ b/packages/app/src/context/global-sync/event-reducer.test.ts @@ -15,12 +15,12 @@ const rootSession = (input: { id: string; parentID?: string; archived?: number } }, }) as Session -const userMessage = (id: string, sessionID: string) => +const userMessage = (id: string, sessionID: string, created = 1) => ({ id, sessionID, role: "user", - time: { created: 1 }, + time: { created }, agent: "assistant", model: { providerID: "openai", modelID: "gpt" }, }) as Message @@ -370,13 +370,13 @@ describe("applyDirectoryEvent", () => { const sessionID = "ses_1" const [store, setStore] = createStore( baseState({ - message: { [sessionID]: [userMessage("msg_1", sessionID), userMessage("msg_3", sessionID)] }, - part: { msg_2: [textPart("prt_1", sessionID, "msg_2")] }, + message: { [sessionID]: [userMessage("msg_z", sessionID, 1), userMessage("msg_b", sessionID, 3)] }, + part: { msg_a: [textPart("prt_1", sessionID, "msg_a")] }, }), ) applyDirectoryEvent({ - event: { type: "message.updated", properties: { info: userMessage("msg_2", sessionID) } }, + event: { type: "message.updated", properties: { info: userMessage("msg_a", sessionID, 2) } }, store, setStore, push() {}, @@ -384,14 +384,14 @@ describe("applyDirectoryEvent", () => { loadLsp() {}, }) - expect(store.message[sessionID]?.map((x) => x.id)).toEqual(["msg_1", "msg_2", "msg_3"]) + expect(store.message[sessionID]?.map((x) => x.id)).toEqual(["msg_z", "msg_a", "msg_b"]) applyDirectoryEvent({ event: { type: "message.updated", properties: { info: { - ...userMessage("msg_2", sessionID), + ...userMessage("msg_a", sessionID, 2), role: "assistant", } as Message, }, @@ -403,10 +403,10 @@ describe("applyDirectoryEvent", () => { loadLsp() {}, }) - expect(store.message[sessionID]?.find((x) => x.id === "msg_2")?.role).toBe("assistant") + expect(store.message[sessionID]?.find((x) => x.id === "msg_a")?.role).toBe("assistant") applyDirectoryEvent({ - event: { type: "message.removed", properties: { sessionID, messageID: "msg_2" } }, + event: { type: "message.removed", properties: { sessionID, messageID: "msg_a" } }, store, setStore, push() {}, @@ -414,8 +414,8 @@ describe("applyDirectoryEvent", () => { loadLsp() {}, }) - expect(store.message[sessionID]?.map((x) => x.id)).toEqual(["msg_1", "msg_3"]) - expect(store.part.msg_2).toBeUndefined() + expect(store.message[sessionID]?.map((x) => x.id)).toEqual(["msg_z", "msg_b"]) + expect(store.part.msg_a).toBeUndefined() }) test("upserts and prunes message parts", () => { diff --git a/packages/app/src/context/global-sync/event-reducer.ts b/packages/app/src/context/global-sync/event-reducer.ts index 8f203715a6d..b84d5201cf5 100644 --- a/packages/app/src/context/global-sync/event-reducer.ts +++ b/packages/app/src/context/global-sync/event-reducer.ts @@ -15,6 +15,7 @@ import type { State, VcsCache } from "./types" import { trimSessions } from "./session-trim" import { dropSessionCaches } from "./session-cache" import { diffs as list, message as clean } from "@/utils/diffs" +import { messageKey } from "@/utils/session-message" const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) const SESSION_CONTENT_EVENTS = new Set([ @@ -275,7 +276,7 @@ export function applyDirectoryEvent(input: { input.setStore("message", info.sessionID, [info]) break } - const result = Binary.search(messages, info.id, (m) => m.id) + const result = Binary.search(messages, messageKey(info), messageKey) if (result.found) { input.setStore("message", info.sessionID, result.index, reconcile(info)) break @@ -295,8 +296,8 @@ export function applyDirectoryEvent(input: { produce((draft) => { const messages = draft.message[props.sessionID] if (messages) { - const result = Binary.search(messages, props.messageID, (m) => m.id) - if (result.found) messages.splice(result.index, 1) + const index = messages.findIndex((message) => message.id === props.messageID) + if (index >= 0) messages.splice(index, 1) } const parts = draft.part[props.messageID] if (parts) { @@ -322,7 +323,7 @@ export function applyDirectoryEvent(input: { input.setStore("part", part.messageID, [part]) break } - const result = Binary.search(parts, part.id, (p) => p.id) + const result = Binary.search(parts, part.id, (item) => item.id) if (result.found) { input.setStore("part", part.messageID, result.index, reconcile(part)) break @@ -345,13 +346,13 @@ export function applyDirectoryEvent(input: { ) const parts = input.store.part[props.messageID] if (!parts) break - const result = Binary.search(parts, props.partID, (p) => p.id) + const result = Binary.search(parts, props.partID, (part) => part.id) if (result.found) { input.setStore( produce((draft) => { const list = draft.part[props.messageID] if (!list) return - const next = Binary.search(list, props.partID, (p) => p.id) + const next = Binary.search(list, props.partID, (part) => part.id) if (!next.found) return list.splice(next.index, 1) if (list.length === 0) delete draft.part[props.messageID] @@ -364,7 +365,7 @@ export function applyDirectoryEvent(input: { const props = event.properties as { messageID: string; partID: string; field: string; delta: string } const parts = input.store.part[props.messageID] if (!parts) break - const result = Binary.search(parts, props.partID, (p) => p.id) + const result = Binary.search(parts, props.partID, (part) => part.id) if (!result.found) break const field = props.field as keyof (typeof parts)[number] const current = parts[result.index]?.[field] diff --git a/packages/app/src/context/server-session.test.ts b/packages/app/src/context/server-session.test.ts index e9595e13760..2ebf5f88ef0 100644 --- a/packages/app/src/context/server-session.test.ts +++ b/packages/app/src/context/server-session.test.ts @@ -264,6 +264,7 @@ describe("server session", () => { expect(requests).toEqual([{ sessionID: "root", limit: 20, 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]) }) test("extends a current page to include the user for split assistant turns", async () => { @@ -1497,7 +1498,7 @@ describe("server session", () => { await store.sync("child", { force: true }) - expect(store.data.message.child).toEqual([boundary, older]) + expect(store.data.message.child).toEqual([older, boundary]) }) test("preserves a part update for a message being loaded from history", async () => { diff --git a/packages/app/src/context/server-session.ts b/packages/app/src/context/server-session.ts index 69bec61ba2f..c5d98682f03 100644 --- a/packages/app/src/context/server-session.ts +++ b/packages/app/src/context/server-session.ts @@ -18,7 +18,7 @@ import { message as cleanMessage } from "@/utils/diffs" import { sessionNotFoundError } from "@/utils/server-errors" import { rootSession } from "@/utils/session-route" import { normalizeSessionInfo } from "@/utils/session" -import { normalizeSessionMessages } from "@/utils/session-message" +import { compareMessages, messageKey, normalizeSessionMessages } from "@/utils/session-message" import { dropSessionCaches, pickSessionCacheEvictions, SESSION_CACHE_LIMIT } from "./global-sync/session-cache" import { createV2SessionReducer, type V2SessionReduction } from "./server-session-v2-reducer" import type { ServerApi } from "@/utils/server" @@ -26,7 +26,6 @@ import type { ServerApi } from "@/utils/server" type MessageApi = ServerApi["message"] const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0) -const cmpMessage = (a: Message, b: Message) => a.time.created - b.time.created || cmp(a.id, b.id) const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) const initialMessagePageSize = 20 const historyMessagePageSize = 200 @@ -64,7 +63,7 @@ type MessagePage = { function legacyMessageSource(items: { info: Message; parts: Part[] }[]): SessionMessageInfo[] { return items .slice() - .sort((a, b) => cmp(a.info.id, b.info.id)) + .sort((a, b) => compareMessages(a.info, b.info)) .map((item) => { if (item.info.role === "user") { return { @@ -111,17 +110,16 @@ function mergeOptimisticPage(page: MessagePage, items: OptimisticItem[]) { const part = new Map(page.part.map((item) => [item.id, item.part])) const observed: { messageID: string; parts: Part[] }[] = [] for (const item of items) { - const result = Binary.search(session, item.message.id, (message) => message.id) - if (!result.found) session.splice(result.index, 0, item.message) + const result = Binary.search(session, messageKey(item.message), messageKey) + const found = result.found + if (!found) session.splice(result.index, 0, item.message) const current = part.get(item.message.id) - const confirmed = result.found - ? item.parts.filter((part) => Binary.search(current ?? [], part.id, (value) => value.id).found) - : [] - if (result.found) observed.push({ messageID: item.message.id, parts: confirmed }) + const confirmed = found ? item.parts.filter((part) => current?.some((value) => value.id === part.id)) : [] + if (found) observed.push({ messageID: item.message.id, parts: confirmed }) part.set( item.message.id, merge( - result.found ? (current ?? []) : merge(item.confirmedParts ?? [], current ?? []), + found ? (current ?? []) : merge(item.confirmedParts ?? [], current ?? []), item.parts.filter((part) => !confirmed.includes(part)), ), ) @@ -158,6 +156,7 @@ function reconcileFetched( retained?: ReadonlySet removed?: ReadonlySet preserveUnfetched?: boolean | ((item: T) => boolean) + compare?: (a: T, b: T) => number } = {}, ) { const result = new Map(fetched.map((item) => [item.id, item])) @@ -180,7 +179,8 @@ function reconcileFetched( if (!item) result.delete(id) } for (const id of options.removed ?? emptyIDs) result.delete(id) - return [...result.values()].sort((a, b) => cmp(a.id, b.id)) + const items = [...result.values()] + return options.compare ? items.sort(options.compare) : items } type ServerSessionOptions = { retry?: typeof retry; protocol?: Promise<"v1" | "v2"> } @@ -413,8 +413,7 @@ export function createServerSession( if (!load) return // A part event keeps an existing parent when the fetched page omits it without overriding fetched metadata. const messages = data.message[sessionID] - if (messages && Binary.search(messages, messageID, (message) => message.id).found) - load.retainedMessages.add(messageID) + if (messages?.some((message) => message.id === messageID)) load.retainedMessages.add(messageID) const parts = load.touchedParts.get(messageID) if (parts) { parts.add(partID) @@ -437,16 +436,14 @@ export function createServerSession( load.touchedParts.set(messageID, new Set(parts)) load.carriedDeltaParts.set(messageID, new Set(parts)) const messages = data.message[sessionID] - if (messages && Binary.search(messages, messageID, (message) => message.id).found) - load.retainedMessages.add(messageID) + if (messages?.some((message) => message.id === messageID)) load.retainedMessages.add(messageID) } for (const [messageID, parts] of load.removedParts) { const touched = load.touchedParts.get(messageID) ?? new Set() parts.forEach((partID) => touched.add(partID)) load.touchedParts.set(messageID, touched) const messages = data.message[sessionID] - if (messages && Binary.search(messages, messageID, (message) => message.id).found) - load.retainedMessages.add(messageID) + if (messages?.some((message) => message.id === messageID)) load.retainedMessages.add(messageID) } for (const [messageID, parts] of load.optimisticParts) { load.removedMessages.delete(messageID) @@ -555,7 +552,7 @@ export function createServerSession( const source = pages.flatMap((page) => page.data).toReversed() const normalized = normalizeSessionMessages(sessionID, source) return { - session: normalized.messages.sort((a, b) => cmp(a.id, b.id)), + session: normalized.messages.sort(compareMessages), part: [...normalized.parts.entries()] .map(([id, part]) => ({ id, part: part.sort((a, b) => cmp(a.id, b.id)) })) .sort((a, b) => cmp(a.id, b.id)), @@ -572,7 +569,7 @@ export function createServerSession( }) const items = (response.data ?? []).filter((item) => !!item?.info?.id) return { - session: items.map((item) => cleanMessage(item.info)).sort((a, b) => cmp(a.id, b.id)), + session: items.map((item) => cleanMessage(item.info)).sort(compareMessages), part: items.map((item) => ({ id: item.info.id, part: item.parts.filter((part) => !!part?.id).sort((a, b) => cmp(a.id, b.id)), @@ -696,7 +693,7 @@ export function createServerSession( const normalized = normalizeSessionMessages(sessionID, source) return { ...page, - session: normalized.messages.sort((a, b) => cmp(a.id, b.id)), + session: normalized.messages.sort(compareMessages), part: [...normalized.parts.entries()] .map(([id, part]) => ({ id, part: part.sort((a, b) => cmp(a.id, b.id)) })) .sort((a, b) => cmp(a.id, b.id)), @@ -713,6 +710,7 @@ export function createServerSession( retained: load?.retainedMessages, removed: load?.removedMessages, preserveUnfetched, + compare: compareMessages, }) batch(() => { if (source) setData("session_message", sessionID, reconcile(source)) @@ -754,7 +752,7 @@ export function createServerSession( try { const page = await fetchMessages(sessionID, limit, before, () => resetMessageLoad(sessionID, load)) const first = page.session.reduce( - (oldest, message) => (!oldest || cmpMessage(message, oldest) < 0 ? message : oldest), + (oldest, message) => (!oldest || compareMessages(message, oldest) < 0 ? message : oldest), undefined, ) if (generations.get(sessionID) !== active) return @@ -804,14 +802,15 @@ export function createServerSession( session: merge( page.session, parents.map((parent) => parent.message), - ), + ).sort(compareMessages), 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))) + mode === "prepend" || + (!result.complete && (!first || ((message: Message) => compareMessages(message, first) < 0))) applyMessagePage( sessionID, result, @@ -928,7 +927,7 @@ export function createServerSession( .message({ sessionID, messageID }) .then((message) => { const current = data.session_message[sessionID] ?? [] - const messages = [...current.filter((item) => item.id !== message.id), message].sort((a, b) => cmp(a.id, b.id)) + const messages = [...current.filter((item) => item.id !== message.id), message].sort(compareMessages) projectV2({ sessionID, messages, touched: [message.id] }) }) .catch(() => {}) @@ -1051,7 +1050,7 @@ export function createServerSession( setData("message", info.sessionID, [info]) return } - const result = Binary.search(messages, info.id, (message) => message.id) + const result = Binary.search(messages, messageKey(info), messageKey) if (result.found) setData("message", info.sessionID, result.index, reconcile(info)) if (!result.found) setData("message", info.sessionID, (value = []) => { @@ -1084,8 +1083,8 @@ export function createServerSession( produce((draft) => { const messages = draft.message[props.sessionID] if (messages) { - const result = Binary.search(messages, props.messageID, (message) => message.id) - if (result.found) messages.splice(result.index, 1) + const index = messages.findIndex((message) => message.id === props.messageID) + if (index >= 0) messages.splice(index, 1) } deleteMessageParts(draft, props.messageID) }), @@ -1097,7 +1096,7 @@ export function createServerSession( if (SKIP_PARTS.has(part.type)) return const messages = data.message[part.sessionID] const load = messageLoads.get(part.sessionID) - const missing = !messages || !Binary.search(messages, part.messageID, (message) => message.id).found + const missing = !messages?.some((message) => message.id === part.messageID) // Outside a page load, accepting a part without its ordered parent event would create an unbounded orphan. if ( missing && @@ -1341,7 +1340,7 @@ export function createServerSession( if (items) items.set(input.message.id, { ...input, parts, confirmedParts: [] }) if (!items) optimistic.set(input.sessionID, new Map([[input.message.id, { ...input, parts, confirmedParts: [] }]])) - setData("message", input.sessionID, (messages = []) => merge(messages, [input.message])) + setData("message", input.sessionID, (messages = []) => merge(messages, [input.message]).sort(compareMessages)) setData( "part_text_accum_delta", produce((draft) => { diff --git a/packages/app/src/context/sync-optimistic.test.ts b/packages/app/src/context/sync-optimistic.test.ts index 94324f8a060..d7ac9fd9641 100644 --- a/packages/app/src/context/sync-optimistic.test.ts +++ b/packages/app/src/context/sync-optimistic.test.ts @@ -4,11 +4,11 @@ import { applyOptimisticAdd, applyOptimisticRemove, mergeOptimisticPage } from " type Text = Extract -const userMessage = (id: string, sessionID: string): Message => ({ +const userMessage = (id: string, sessionID: string, created = 1): Message => ({ id, sessionID, role: "user", - time: { created: 1 }, + time: { created }, agent: "assistant", model: { providerID: "openai", modelID: "gpt" }, }) @@ -22,21 +22,21 @@ const textPart = (id: string, sessionID: string, messageID: string): Text => ({ }) describe("sync optimistic reducers", () => { - test("applyOptimisticAdd inserts message in sorted order and stores parts", () => { + test("applyOptimisticAdd inserts by creation time", () => { const sessionID = "ses_1" const draft = { - message: { [sessionID]: [userMessage("msg_2", sessionID)] }, + message: { [sessionID]: [userMessage("msg_z", sessionID, 1)] }, part: {} as Record, } applyOptimisticAdd(draft, { sessionID, - message: userMessage("msg_1", sessionID), - parts: [textPart("prt_2", sessionID, "msg_1"), textPart("prt_1", sessionID, "msg_1")], + message: userMessage("msg_a", sessionID, 2), + parts: [textPart("prt_2", sessionID, "msg_a"), textPart("prt_1", sessionID, "msg_a")], }) - expect(draft.message[sessionID]?.map((x) => x.id)).toEqual(["msg_1", "msg_2"]) - expect(draft.part.msg_1?.map((x) => x.id)).toEqual(["prt_1", "prt_2"]) + expect(draft.message[sessionID]?.map((x) => x.id)).toEqual(["msg_z", "msg_a"]) + expect(draft.part.msg_a?.map((x) => x.id)).toEqual(["prt_1", "prt_2"]) }) test("applyOptimisticRemove removes message and part entries", () => { @@ -60,19 +60,33 @@ describe("sync optimistic reducers", () => { const sessionID = "ses_1" const page = mergeOptimisticPage( { - session: [userMessage("msg_1", sessionID)], - part: [{ id: "msg_1", part: [textPart("prt_1", sessionID, "msg_1")] }], + session: [userMessage("msg_z", sessionID, 1)], + part: [{ id: "msg_z", part: [textPart("prt_1", sessionID, "msg_z")] }], complete: true, }, - [{ message: userMessage("msg_2", sessionID), parts: [textPart("prt_2", sessionID, "msg_2")] }], + [{ message: userMessage("msg_a", sessionID, 2), parts: [textPart("prt_2", sessionID, "msg_a")] }], ) - expect(page.session.map((x) => x.id)).toEqual(["msg_1", "msg_2"]) - expect(page.part.find((x) => x.id === "msg_2")?.part.map((x) => x.id)).toEqual(["prt_2"]) + expect(page.session.map((x) => x.id)).toEqual(["msg_z", "msg_a"]) + expect(page.part.find((x) => x.id === "msg_a")?.part.map((x) => x.id)).toEqual(["prt_2"]) expect(page.confirmed).toEqual([]) expect(page.complete).toBe(true) }) + test("mergeOptimisticPage uses IDs only to break equal-time ties", () => { + const sessionID = "ses_1" + const page = mergeOptimisticPage( + { + session: [userMessage("msg_z", sessionID, 1)], + part: [], + complete: true, + }, + [{ message: userMessage("msg_a", sessionID, 1), parts: [] }], + ) + + expect(page.session.map((message) => message.id)).toEqual(["msg_a", "msg_z"]) + }) + test("mergeOptimisticPage keeps missing optimistic parts until the server has them", () => { const sessionID = "ses_1" const page = mergeOptimisticPage( diff --git a/packages/app/src/context/sync.tsx b/packages/app/src/context/sync.tsx index b90f49ffd49..7dbf66eb736 100644 --- a/packages/app/src/context/sync.tsx +++ b/packages/app/src/context/sync.tsx @@ -3,6 +3,7 @@ import { createMemo } from "solid-js" import { useServerSync } from "./server-sync" import { useSDK } from "./sdk" import type { Message, Part } from "@opencode-ai/sdk/v2/client" +import { messageKey } from "@/utils/session-message" const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) @@ -67,7 +68,7 @@ export function mergeOptimisticPage(page: MessagePage, items: OptimisticItem[]) const confirmed: string[] = [] for (const item of items) { - const result = Binary.search(session, item.message.id, (message) => message.id) + const result = Binary.search(session, messageKey(item.message), messageKey) const found = result.found if (!found) session.splice(result.index, 0, item.message) @@ -92,7 +93,7 @@ export function mergeOptimisticPage(page: MessagePage, items: OptimisticItem[]) export function applyOptimisticAdd(draft: OptimisticStore, input: OptimisticAddInput) { const messages = draft.message[input.sessionID] if (messages) { - const result = Binary.search(messages, input.message.id, (m) => m.id) + const result = Binary.search(messages, messageKey(input.message), messageKey) messages.splice(result.index, 0, input.message) } else { draft.message[input.sessionID] = [input.message] @@ -103,8 +104,8 @@ export function applyOptimisticAdd(draft: OptimisticStore, input: OptimisticAddI export function applyOptimisticRemove(draft: OptimisticStore, input: OptimisticRemoveInput) { const messages = draft.message[input.sessionID] if (messages) { - const result = Binary.search(messages, input.messageID, (m) => m.id) - if (result.found) messages.splice(result.index, 1) + const index = messages.findIndex((message) => message.id === input.messageID) + if (index >= 0) messages.splice(index, 1) } delete draft.part[input.messageID] } diff --git a/packages/app/src/utils/session-message.ts b/packages/app/src/utils/session-message.ts index 93d86a66bb2..eef9f4e6dfb 100644 --- a/packages/app/src/utils/session-message.ts +++ b/packages/app/src/utils/session-message.ts @@ -12,6 +12,14 @@ const emptyTokens = { input: 0, output: 0, reasoning: 0, cache: { read: 0, write const emptyModel: { id: string; providerID: string; variant?: string } = { id: "", providerID: "" } const decodeToolInput = Schema.decodeUnknownOption(Schema.UnknownFromJsonString) +export function compareMessages(a: Pick, b: Pick) { + const left = messageKey(a) + const right = messageKey(b) + return left < right ? -1 : left > right ? 1 : 0 +} + +export const messageKey = (message: Pick) => message.time.created + message.id + function record(value: unknown): value is Record { return !!value && typeof value === "object" && !Array.isArray(value) }