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