import { describe, expect, test } from "bun:test" import type { retry } from "@opencode-ai/core/util/retry" import type { FormInfo, OpenCodeEvent, SessionApi, SessionInboxInfo, SessionInfo, SessionMessageAssistant, SessionMessageAssistantTool, SessionMessageInfo, } from "@opencode-ai/client/promise" import type { Message, Part } from "@/types" import { createServerSession } from "./server-session" import type { ServerApi } from "@/utils/server" type MessageApi = ServerApi["message"] const session = (id: string, parentID?: string): SessionInfo => ({ id, projectID: "project", location: { directory: "/repo" }, title: id, cost: 0, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, parentID, time: { created: 1, updated: 1 }, }) type UserMessage = Extract type AssistantMessage = Extract type TextPart = Extract type CurrentToolObject = Extract["input"] type MessageResponse = { data: { info: Message; parts: Part[] }[] response: { headers: Headers } } type SingleMessageResponse = { data: MessageResponse["data"][number] } function sessionInfo(value: SessionInfo): SessionInfo { return value } function currentMessages(data: MessageResponse["data"]): SessionMessageInfo[] { return data.flatMap((item): SessionMessageInfo[] => { if (item.info.role === "user") { return [ { id: `${item.info.id}:agent`, type: "agent-switched", agent: item.info.agent, time: item.info.time, }, { id: `${item.info.id}:model`, type: "model-switched", model: { id: item.info.model.modelID, providerID: item.info.model.providerID, variant: item.info.model.variant, }, time: item.info.time, }, { id: item.info.id, type: "user", text: item.parts.flatMap((part) => (part.type === "text" ? [part.text] : [])).join("\n"), time: item.info.time, }, ] } return [ { id: item.info.id, type: "assistant", agent: item.info.agent, model: { id: item.info.modelID, providerID: item.info.providerID, variant: item.info.variant }, content: item.parts.flatMap((part): SessionMessageAssistant["content"] => { if (part.type === "text") return [{ type: "text", text: part.text }] if (part.type === "reasoning") return [ { type: "reasoning", text: part.text, time: part.time ? { created: part.time.start, completed: part.time.end } : undefined, }, ] if (part.type !== "tool") return [] const state: SessionMessageAssistantTool["state"] = (() => { if (part.state.status === "pending") return { status: "streaming" as const, input: JSON.stringify(part.state.input) } if (part.state.status === "running") return { status: "running" as const, input: part.state.input as CurrentToolObject, metadata: (part.state.metadata ?? {}) as CurrentToolObject, } if (part.state.status === "error") return { status: "error" as const, input: part.state.input as CurrentToolObject, error: { type: "tool_error", message: part.state.error }, metadata: part.state.metadata as CurrentToolObject | undefined, } return { status: "completed" as const, input: part.state.input as CurrentToolObject, content: [{ type: "text" as const, text: part.state.output }], metadata: part.state.metadata as CurrentToolObject, } })() return [ { id: part.id, type: "tool" as const, name: part.tool, state, time: { created: part.state.status === "pending" ? item.info.time.created : part.state.time.start, ran: part.state.status === "pending" ? undefined : part.state.time.start, completed: part.state.status === "completed" || part.state.status === "error" ? part.state.time.end : undefined, }, }, ] }), time: item.info.time, cost: item.info.cost, tokens: item.info.tokens, finish: item.info.finish as | "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | undefined, }, ] }) } function currentPage(value: MessageResponse) { return { data: currentMessages(value.data).toReversed(), cursor: { next: value.response.headers.get("x-next-cursor") ?? undefined }, } } const userMessage = (id: string, input: Partial = {}): UserMessage => ({ id, sessionID: "child", role: "user", time: { created: 1 }, agent: "build", model: { providerID: "provider", modelID: "model", variant: undefined }, ...input, }) const assistantMessage = (id: string, parentID: string, input: Partial = {}): AssistantMessage => ({ id, sessionID: "child", role: "assistant", time: { created: Number(id.at(-1)), completed: Number(id.at(-1)) }, parentID, modelID: "model", providerID: "provider", variant: undefined, mode: "build", agent: "build", path: { cwd: "", root: "" }, cost: 0, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, error: undefined, finish: undefined, ...input, }) const textPart = (messageID: string, input: Partial = {}): TextPart => ({ sessionID: "child", messageID, type: "text", text: "text", ...input, id: `${messageID}:text:${input.id === "pending" ? 1 : 0}`, }) const promptEcho = (messageID: string, text = "hello") => ({ sessionID: "child", messageID, text, displayText: text, agent: "build", model: { providerID: "provider", modelID: "model" }, comments: [], }) const response = (data: MessageResponse["data"] = [], cursor?: string): MessageResponse => ({ data, response: { headers: new Headers(cursor ? { "x-next-cursor": cursor } : undefined) }, }) const singleResponse = (info: Message, parts: Part[] = []): SingleMessageResponse => ({ data: { info, parts } }) const deferredResponse = () => Promise.withResolvers() function messageClient(...responses: Array>) { let index = 0 const pages = responses.map((value) => value instanceof Promise ? value.then(currentPage) : Promise.resolve(currentPage(value)), ) const requests: unknown[] = [] const waiting = new Map void>() const client = { session: { get: async () => sessionInfo(session("child", "root")), message: async () => { throw new Error("Unexpected single message request") }, }, message: { list: async (input: unknown) => { requests.push(input) waiting.get(requests.length)?.() waiting.delete(requests.length) return pages[index++]! }, }, } as unknown as { session: SessionApi; message: MessageApi } return Object.assign(client, { requests, requested(count: number) { if (requests.length >= count) return Promise.resolve() return new Promise((resolve) => waiting.set(count, resolve)) }, }) } function rootMessageClient( pages: Array>, roots: Array>, ) { let pageIndex = 0 let rootIndex = 0 const requests: unknown[] = [] const rootRequests: unknown[] = [] const rootWaiting = new Map void>() const client = { session: { get: async () => sessionInfo(session("child", "root")), message: async (input: unknown) => { rootRequests.push(input) rootWaiting.get(rootRequests.length)?.() rootWaiting.delete(rootRequests.length) const value = await roots[rootIndex++]! return currentMessages([value.data]).find((message) => message.id === value.data.info.id)! }, }, message: { list: async (input: unknown) => { requests.push(input) return currentPage(await pages[pageIndex++]!) }, }, } as unknown as { session: SessionApi; message: MessageApi } return Object.assign(client, { requests, rootRequests, rootRequested(count: number) { if (rootRequests.length >= count) return Promise.resolve() return new Promise((resolve) => rootWaiting.set(count, resolve)) }, }) } const retryImmediately: typeof retry = async (task, options = {}) => { const attempts = options.attempts ?? 3 for (let attempt = 0; ; attempt++) { try { return await task() } catch (error) { if (attempt === attempts - 1) throw error } } } function setup(sessions: Record) { const get: unknown[] = [] const messages: unknown[] = [] const client = { session: { get: async (input: unknown) => { get.push(input) const id = (input as { sessionID: string }).sessionID return sessionInfo(sessions[id]!) }, message: async () => { throw new Error("Unexpected single message request") }, }, message: { list: async (input: unknown) => { messages.push(input) return currentPage(response()) }, }, } as unknown as { session: SessionApi; message: MessageApi } return { get, messages, store: createServerSession(client) } } describe("server session", () => { test("hydrates session info after a native session.created event", async () => { const ctx = setup({ created: session("created") }) ctx.store.apply({ type: "session.created", properties: { sessionID: "created", projectID: "project", location: { directory: "/repo" }, slug: "created", version: "test", }, }) expect(ctx.store.get("created")).toBeUndefined() await ctx.store.resolve("created") expect(ctx.store.get("created")?.location.directory).toBe("/repo") expect(ctx.get).toEqual([{ sessionID: "created" }]) }) test("projects V2 session events into current and legacy message state", () => { const ctx = setup({ child: session("child") }) ctx.store.remember(session("child")) ctx.store.set("session_message", "child", [ { id: "msg_1_user", type: "user", text: "hello", time: { created: 1 }, }, ]) const apply = (input: object) => ctx.store.applyV2(input as OpenCodeEvent) apply({ id: "evt_step", created: 2, type: "session.step.started", durable: { aggregateID: "child", seq: 1, version: 1 }, location: { directory: "/repo" }, data: { sessionID: "child", assistantMessageID: "msg_2_assistant", agent: "build", model: { id: "model", providerID: "provider" }, }, }) apply({ id: "evt_text_start", created: 3, type: "session.text.started", durable: { aggregateID: "child", seq: 2, version: 1 }, location: { directory: "/repo" }, data: { sessionID: "child", assistantMessageID: "msg_2_assistant", ordinal: 0 }, }) apply({ id: "evt_text_delta", created: 4, type: "session.text.delta", location: { directory: "/repo" }, data: { sessionID: "child", assistantMessageID: "msg_2_assistant", ordinal: 0, delta: "world" }, }) apply({ id: "evt_tool_z", created: 5, type: "session.tool.input.started", durable: { aggregateID: "child", seq: 3, version: 1 }, location: { directory: "/repo" }, data: { sessionID: "child", assistantMessageID: "msg_2_assistant", id: "call_z", name: "shell" }, }) apply({ id: "evt_tool_a", created: 6, type: "session.tool.input.started", durable: { aggregateID: "child", seq: 4, version: 1 }, location: { directory: "/repo" }, data: { sessionID: "child", assistantMessageID: "msg_2_assistant", id: "call_a", name: "shell" }, }) expect(ctx.store.data.session_message.child?.at(-1)).toMatchObject({ id: "msg_2_assistant", type: "assistant", content: [ { type: "text", text: "world" }, { type: "tool", id: "call_z" }, { type: "tool", id: "call_a" }, ], }) expect(ctx.store.data.message.child?.map((message) => message.id)).toEqual(["msg_1_user", "msg_2_assistant"]) expect(ctx.store.data.part.msg_2_assistant?.map((part) => part.id)).toEqual([ "msg_2_assistant:text:0", "call_z", "call_a", ]) }) test("projects V2 pending inputs and forms", () => { const ctx = setup({ child: session("child") }) const apply = (input: object) => ctx.store.applyV2(input as OpenCodeEvent) apply({ id: "evt_admitted", created: 1, type: "session.inbox.enqueued", data: { sessionID: "child", inboxID: "msg_input", item: { type: "user", delivery: "steer", payload: { text: "hello" } }, }, }) apply({ id: "evt_form", created: 2, type: "form.created", data: { form: { id: "frm_1", sessionID: "child", title: "Choose", fields: [] } }, }) expect(ctx.store.data.pending.child).toMatchObject([{ id: "msg_input", delivery: "steer" }]) expect(ctx.store.data.input.child).toEqual(["msg_input"]) expect(ctx.store.data.form.child).toMatchObject([{ id: "frm_1", title: "Choose" }]) expect(ctx.store.data.message.child).toMatchObject([{ id: "msg_input" }]) apply({ id: "evt_cancelled", created: 3, type: "session.inbox.cancelled", data: { sessionID: "child", inboxID: "msg_input" }, }) apply({ id: "evt_form_done", created: 4, type: "form.cancelled", data: { sessionID: "child", id: "frm_1" }, }) expect(ctx.store.data.pending.child).toEqual([]) expect(ctx.store.data.input.child).toEqual([]) expect(ctx.store.data.form.child).toEqual([]) expect(ctx.store.data.message.child).toEqual([]) }) test("does not let transient hydration overwrite newer events", async () => { const ctx = setup({ child: session("child") }) const response = Promise.withResolvers<{ pending: SessionInboxInfo[]; forms: FormInfo[] }>() const form: FormInfo = { id: "frm_1", sessionID: "child", title: "Choose", fields: [{ key: "choice", type: "string" as const, title: "Choice" }], } let loads = 0 const hydration = ctx.store.hydrateTransient("child", () => { loads += 1 if (loads === 1) return response.promise return Promise.resolve({ pending: ctx.store.data.pending.child ?? [], forms: [form] }) }) ctx.store.applyV2({ id: "evt_admitted", created: 1, type: "session.inbox.enqueued", data: { sessionID: "child", inboxID: "msg_input", item: { type: "user", delivery: "queue", payload: { text: "new" } }, }, } as OpenCodeEvent) response.resolve({ pending: [], forms: [form], }) await hydration expect(ctx.store.data.pending.child).toMatchObject([{ id: "msg_input" }]) expect(ctx.store.data.form.child).toMatchObject([{ id: "frm_1" }]) expect(loads).toBe(2) }) test("removes only the compaction input named by the start event", () => { const ctx = setup({ child: session("child") }) const apply = (input: object) => ctx.store.applyV2(input as OpenCodeEvent) apply({ id: "evt_compaction_inbox", created: 1, type: "session.inbox.enqueued", data: { sessionID: "child", inboxID: "msg_compaction", item: { type: "compaction", delivery: "queue", payload: { reason: "manual" } }, }, }) expect(ctx.store.data.input.child).toBeUndefined() expect(ctx.store.data.pending.child).toHaveLength(1) apply({ id: "evt_compaction_inbox_2", created: 2, type: "session.inbox.enqueued", data: { sessionID: "child", inboxID: "msg_compaction_2", item: { type: "compaction", delivery: "queue", payload: { reason: "manual" } }, }, }) apply({ id: "evt_compaction_ended", created: 3, type: "session.compaction.ended", data: { sessionID: "child", reason: "manual", text: "summary", recent: "recent" }, }) expect(ctx.store.data.pending.child).toHaveLength(2) apply({ id: "evt_compaction_started", created: 4, type: "session.compaction.started", data: { sessionID: "child", inputID: "msg_compaction", reason: "manual" }, }) expect(ctx.store.data.pending.child?.map((item) => item.id)).toEqual(["msg_compaction_2"]) }) test("projects committed revert before server reconciliation", () => { const ctx = setup({ child: session("child") }) ctx.store.remember({ ...session("child"), revert: { messageID: "msg_2", partID: "prt_1" } }) ctx.store.set("input", "child", ["msg_1", "msg_2"]) ctx.store.set("session_message", "child", [ { id: "msg_1", type: "user", text: "keep", time: { created: 1 } }, { id: "msg_2", type: "user", text: "remove", time: { created: 2 } }, ]) ctx.store.applyV2({ id: "evt_revert", created: 3, type: "session.revert.committed", data: { sessionID: "child", to: "msg_2" }, } as OpenCodeEvent) expect(ctx.store.data.info.child?.revert).toBeUndefined() expect(ctx.store.data.input.child).toEqual(["msg_1"]) expect(ctx.store.data.session_message.child?.map((message) => message.id)).toEqual(["msg_1"]) }) test("does not restore a message hydrated before a committed revert", async () => { const response = Promise.withResolvers() const store = createServerSession({ session: { get: async () => session("child"), message: () => response.promise, } as unknown as SessionApi, message: { list: async () => currentPage({ data: [], response: { headers: new Headers() } }), } as unknown as MessageApi, }) store.remember(session("child")) store.applyV2({ id: "evt_delivered", created: 1, type: "session.inbox.delivered", data: { sessionID: "child", inboxID: "msg_2" }, } as OpenCodeEvent) store.applyV2({ id: "evt_revert", created: 2, type: "session.revert.committed", data: { sessionID: "child", to: "msg_2" }, } as OpenCodeEvent) response.resolve({ id: "msg_2", type: "user", text: "stale", time: { created: 1 } }) await response.promise await Bun.sleep(0) expect(store.data.session_message.child ?? []).toEqual([]) }) test("resolves lineage by session ID without directory", async () => { const ctx = setup({ child: session("child", "root"), root: session("root") }) const result = await ctx.store.lineage.resolve("child") expect(result.root.id).toBe("root") expect(ctx.get).toEqual([{ sessionID: "child" }, { sessionID: "root" }]) expect(ctx.store.lineage.peek("child")).toEqual(result) }) test("applies moved session locations without evicting cached state", () => { const current = { ...session("child"), location: { directory: "/repo/worktree" } } const ctx = setup({ child: current }) ctx.store.remember(current) ctx.store.applyV2({ id: "evt_moved", created: 2, type: "session.moved", durable: { aggregateID: "child", seq: 1, version: 1 }, location: current.location, data: { sessionID: "child", location: { directory: "/repo" }, projectID: "project", subpath: "packages/app" }, } satisfies Extract) expect(ctx.store.get("child")).toMatchObject({ location: { directory: "/repo" }, subpath: "packages/app" }) }) test("loads session content through the server client", async () => { const ctx = setup({ root: session("root") }) await ctx.store.sync("root") expect(ctx.get).toEqual([{ sessionID: "root" }]) expect(ctx.messages).toEqual([{ sessionID: "root", limit: 20, order: "desc" }]) expect(ctx.store.data.message.root).toEqual([]) }) test("reloads cached sessions after reconnect invalidation", async () => { const ctx = setup({ root: session("root") }) await ctx.store.sync("root") ctx.store.invalidate() await ctx.store.sync("root") expect(ctx.get).toHaveLength(2) expect(ctx.messages).toHaveLength(2) }) test("loads current session content through the current message API", async () => { const requests: unknown[] = [] const user = { id: "msg_z_user", type: "user", text: "hello", time: { created: 1 } } const assistant = { id: "msg_a_assistant", type: "assistant", agent: "build", model: { id: "model", providerID: "provider" }, content: [{ type: "text", text: "hi" }], time: { created: 2, completed: 3 }, } const messageApi = { list: async (input: unknown) => { requests.push(input) return { data: [assistant, user], cursor: { previous: null, next: null } } }, } as unknown as MessageApi const store = createServerSession({} as SessionApi, messageApi) store.remember(session("root")) await store.sync("root") 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("preserves assistant content order from message history", async () => { const source = [ { id: "msg_user", type: "user", text: "inspect it", time: { created: 1 } }, { id: "msg_assistant", type: "assistant", agent: "build", model: { id: "model", providerID: "provider" }, content: [ { type: "text", text: "I will inspect it." }, { type: "tool", id: "call_z", name: "shell", state: { status: "streaming", input: "" }, time: { created: 2 }, }, { type: "tool", id: "call_a", name: "shell", state: { status: "streaming", input: "" }, time: { created: 3 }, }, ], time: { created: 2 }, }, ] satisfies SessionMessageInfo[] const messageApi = { list: async () => ({ data: source.toReversed(), cursor: { previous: null, next: null } }), } as unknown as MessageApi const store = createServerSession({} as SessionApi, messageApi) store.remember(session("root")) await store.sync("root") expect(store.data.part.msg_assistant?.map((part) => part.id)).toEqual(["msg_assistant:text:0", "call_z", "call_a"]) }) test("extends a current page to include the user for split assistant turns", async () => { const user = { id: "msg_1_user", type: "user", text: "hello", time: { created: 1 } } as const const assistant = (id: string, created: number) => ({ id, type: "assistant" as const, agent: "build", model: { id: "model", providerID: "provider" }, content: [{ type: "text" as const, text: id }], time: { created, completed: created }, }) const assistants = [ assistant("msg_2_assistant", 2), assistant("msg_3_assistant", 3), assistant("msg_4_assistant", 4), ] const pages = [ { data: assistants.slice(1).toReversed(), cursor: { previous: null, next: "older" } }, { data: [assistants[0], user], cursor: { previous: null, next: null } }, ] const requests: unknown[] = [] const messageApi = { list: async (input: unknown) => { requests.push(input) return pages.shift()! }, } as unknown as MessageApi const store = createServerSession({} as SessionApi, messageApi) store.remember(session("root")) await store.sync("root") expect(requests).toEqual([ { sessionID: "root", limit: 20, order: "desc" }, { sessionID: "root", limit: 20, cursor: "older" }, ]) expect(store.data.message.root.map((message) => message.id)).toEqual([ user.id, ...assistants.map((item) => item.id), ]) expect(assistants.map((item) => store.data.part[item.id]?.[0]?.type)).toEqual(["text", "text", "text"]) }) // 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 () => { 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: [] })))], [singleResponse(user)], ) const store = createServerSession(client) await store.sync("child") expect(client.requests).toEqual([{ sessionID: "child", limit: 20, 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) }) test("keeps assistant history when its deleted parent cannot be backfilled", async () => { const missing = Promise.withResolvers() const assistant = assistantMessage("message-2", "message-missing") const client = rootMessageClient([response([{ info: assistant, parts: [] }])], [missing.promise]) const store = createServerSession(client) const loading = store.sync("child") await client.rootRequested(1) missing.reject(new Error("Message not found: message-missing", { cause: { status: 404 } })) await loading expect(client.rootRequests).toEqual([{ sessionID: "child", messageID: "message-missing" }]) expect(store.data.message.child).toEqual([assistant]) expect(store.history.more("child")).toBe(false) }) test("drops a cached parent when a forced refresh confirms it was deleted", async () => { const missing = Promise.withResolvers() const parent = userMessage("message-1") const part = textPart(parent.id) const assistant = assistantMessage("message-2", parent.id) const client = rootMessageClient( [ response([ { info: parent, parts: [part] }, { info: assistant, parts: [] }, ]), response([{ info: assistant, parts: [] }]), ], [missing.promise], ) const store = createServerSession(client) await store.sync("child") const loading = store.sync("child", { force: true }) await client.rootRequested(1) missing.reject(new Error(`Message not found: ${parent.id}`, { cause: { status: 404 } })) await loading expect(store.data.message.child).toEqual([assistant]) expect(store.data.part[parent.id]).toBeUndefined() }) test("does not let an admitted user suppress initial root backfill", 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: [] })))], [singleResponse(user)], ) const store = createServerSession(client) store.inbox.echo(promptEcho(user.id, "text")) await store.sync("child") 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: [] })))], [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: [] }])], [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: [] }])], [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("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: [] }])) 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: [] })))], [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: [] }])], [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: [] }])], [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] }])], [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") const live = userMessage("message-2", { time: { created: 2 } }) const livePart = textPart(live.id, { text: "live" }) const store = createServerSession(messageClient(pending.promise)) const loading = store.sync("child") store.apply({ type: "message.updated", properties: { info: live } }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part: livePart, time: 2 } }) pending.resolve(response([{ info: user, parts: [] }])) await loading expect(store.data.message.child).toEqual([user, live]) expect(store.data.part[live.id]).toEqual([livePart]) }) test("preserves same-ID live updates over the initial page", async () => { const pending = deferredResponse() const fetched = userMessage("message") const fetchedPart = textPart(fetched.id, { text: "fetched" }) const live = { ...fetched, time: { created: 2 } } const livePart = { ...fetchedPart, text: "live" } const store = createServerSession(messageClient(pending.promise)) const loading = store.sync("child") store.apply({ type: "message.updated", properties: { info: live } }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part: livePart, time: 2 } }) pending.resolve(response([{ info: fetched, parts: [fetchedPart] }])) await loading expect(store.data.message.child).toEqual([live]) expect(store.data.part[live.id]).toEqual([livePart]) }) test("preserves removals received during the initial load", async () => { const pending = deferredResponse() const removed = userMessage("message-1") const kept = { ...removed, id: "message-2" } const part = textPart(kept.id, { text: "removed" }) const store = createServerSession(messageClient(pending.promise)) const loading = store.sync("child") store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: removed.id } }) store.apply({ type: "message.part.removed", properties: { sessionID: "child", messageID: kept.id, partID: part.id }, }) pending.resolve( response([ { info: removed, parts: [] }, { info: kept, parts: [part] }, ]), ) await loading expect(store.data.message.child).toEqual([kept]) expect(store.data.part[kept.id]).toBeUndefined() }) test("keeps removal tracking isolated across load generations", async () => { const firstResponse = deferredResponse() const secondResponse = deferredResponse() const message = userMessage("message") const store = createServerSession(messageClient(firstResponse.promise, secondResponse.promise)) const first = store.sync("child") store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: message.id } }) store.apply({ type: "session.deleted", properties: { sessionID: "child", info: session("child", "root") }, }) const second = store.sync("child") firstResponse.resolve(response()) await first secondResponse.resolve(response([{ info: message, parts: [] }])) await second expect(store.data.message.child).toEqual([message]) }) test("tracks removals in a replacement load generation", async () => { const firstResponse = deferredResponse() const secondResponse = deferredResponse() const message = userMessage("message") const store = createServerSession(messageClient(firstResponse.promise, secondResponse.promise)) const first = store.sync("child") store.apply({ type: "session.deleted", properties: { sessionID: "child", info: session("child", "root") }, }) const second = store.sync("child") store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: message.id } }) firstResponse.resolve(response()) await first secondResponse.resolve(response([{ info: message, parts: [] }])) await second expect(store.data.message.child).toEqual([]) }) test("preserves remove then re-add when a refresh omits the message", async () => { const pending = deferredResponse() const message = userMessage("message") const store = createServerSession(messageClient(response([{ info: message, parts: [] }]), pending.promise)) await store.sync("child") const refreshing = store.sync("child", { force: true }) store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: message.id } }) store.apply({ type: "message.updated", properties: { info: message } }) pending.resolve(response()) await refreshing expect(store.data.message.child).toEqual([message]) }) test("preserves a re-added message without restoring removed parts", async () => { const pending = deferredResponse() const message = userMessage("message") const part = textPart(message.id, { text: "stale" }) const store = createServerSession(messageClient(response([{ info: message, parts: [] }]), pending.promise)) await store.sync("child") const refreshing = store.sync("child", { force: true }) store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: message.id } }) store.apply({ type: "message.updated", properties: { info: message } }) pending.resolve(response([{ info: message, parts: [part] }])) await refreshing expect(store.data.message.child).toEqual([message]) expect(store.data.part[message.id]).toBeUndefined() }) test("drops stale event content omitted by a complete initial page", async () => { const stale = userMessage("stale") const store = createServerSession(messageClient(response())) store.apply({ type: "message.updated", properties: { info: stale } }) await store.sync("child") expect(store.data.message.child).toEqual([]) }) test("preserves event content outside an incomplete initial page", async () => { const live = userMessage("message-1") const fetched = userMessage("message-2", { time: { created: 2 } }) const store = createServerSession(messageClient(response([{ info: fetched, parts: [] }], "older"))) store.apply({ type: "message.updated", properties: { info: live } }) await store.sync("child") expect(store.data.message.child).toEqual([live, fetched]) }) test("echoes a prompt without changing durable message order", () => { const store = setup({ child: session("child") }).store store.inbox.echo({ ...promptEcho("msg_prompt"), text: "hello\nThe user made the following comment regarding line 4 of src/foo.ts: check this", files: [{ uri: "file:///repo/src/foo.ts", mime: "text/plain", name: "foo.ts" }], agents: [{ name: "explore" }], comments: [ { path: "src/foo.ts", selection: { startLine: 4, startChar: 1, endLine: 4, endChar: 5 }, comment: "check this", preview: "const value = 1", origin: "review", }, ], }) expect(store.data.pending.child).toMatchObject([{ id: "msg_prompt", type: "user", delivery: "steer" }]) expect(store.data.input.child).toEqual(["msg_prompt"]) expect(store.data.session_message.child).toBeUndefined() expect(store.data.message.child?.map((message) => message.id)).toEqual(["msg_prompt"]) expect(store.data.part.msg_prompt).toMatchObject([ { id: "msg_prompt:text:0", type: "text", text: "hello" }, { id: "msg_prompt:file:0", type: "file", filename: "foo.ts" }, { id: "msg_prompt:agent:0", type: "agent", name: "explore" }, { id: "msg_prompt:comment:0", type: "text", synthetic: true, metadata: { opencodeComment: { path: "src/foo.ts", selection: { startLine: 4, startChar: 1, endLine: 4, endChar: 5 }, comment: "check this", preview: "const value = 1", origin: "review", }, }, }, ]) store.applyV2({ id: "evt_prompt", created: 2, type: "session.inbox.enqueued", durable: { aggregateID: "child", seq: 1, version: 1 }, data: { sessionID: "child", inboxID: "msg_prompt", item: { type: "user", delivery: "steer", payload: { text: "hello\nThe user made the following comment regarding line 4 of src/foo.ts: check this", }, }, }, } as OpenCodeEvent) expect(store.data.part.msg_prompt).toMatchObject([ { id: "msg_prompt:text:0", type: "text", text: "hello" }, { id: "msg_prompt:comment:0", type: "text", synthetic: true }, ]) }) test("preserves a local echo while message history omits pending input", async () => { const store = createServerSession(messageClient(response())) store.inbox.echo(promptEcho("msg_prompt")) store.inbox.confirm({ id: "msg_prompt", sessionID: "child", timeCreated: 1, type: "user", delivery: "steer", payload: { text: "hello" }, }) await store.sync("child") expect(store.data.message.child?.map((message) => message.id)).toEqual(["msg_prompt"]) expect(store.data.part.msg_prompt).toMatchObject([{ type: "text", text: "hello" }]) }) test("preserves local comment presentation through message refresh", async () => { const note = "The user made the following comment regarding line 4 of src/foo.ts: check this" const message = userMessage("msg_prompt") const store = createServerSession( messageClient(response([{ info: message, parts: [textPart(message.id, { text: note })] }])), ) store.inbox.echo({ ...promptEcho(message.id), text: `hello\n${note}`, comments: [ { path: "src/foo.ts", selection: { startLine: 4, startChar: 1, endLine: 4, endChar: 5 }, comment: "check this", origin: "review", }, ], }) await store.sync("child") expect(store.data.part.msg_prompt).toMatchObject([ { id: "msg_prompt:text:0", type: "text", text: "hello" }, { id: "msg_prompt:comment:0", type: "text", synthetic: true }, ]) }) test("retires an admitted echo absent from authoritative reconnect state", async () => { const store = createServerSession(messageClient(response())) store.inbox.echo(promptEcho("msg_prompt")) store.inbox.confirm({ id: "msg_prompt", sessionID: "child", timeCreated: 1, type: "user", delivery: "steer", payload: { text: "hello" }, }) await Promise.all([store.sync("child"), store.hydrateTransient("child", async () => ({ pending: [], forms: [] }))]) store.inbox.reconcile("child") expect(store.data.pending.child).toEqual([]) expect(store.data.message.child).toEqual([]) expect(store.data.part.msg_prompt).toBeUndefined() }) test("retires a stale enqueued message when inbox hydration finishes after history", async () => { const store = createServerSession(messageClient(response())) store.applyV2({ id: "evt_prompt", created: 1, type: "session.inbox.enqueued", durable: { aggregateID: "child", seq: 1, version: 1 }, data: { sessionID: "child", inboxID: "msg_prompt", item: { type: "user", delivery: "steer", payload: { text: "hello" } }, }, } as OpenCodeEvent) await store.sync("child") await store.hydrateTransient("child", async () => ({ pending: [], forms: [] })) store.inbox.reconcile("child") expect(store.data.pending.child).toEqual([]) expect(store.data.session_message.child).toEqual([]) expect(store.data.message.child).toEqual([]) expect(store.data.part.msg_prompt).toBeUndefined() }) test("deduplicates the durable admission event against its local echo", () => { const store = setup({ child: session("child") }).store store.inbox.echo(promptEcho("msg_prompt")) store.applyV2({ id: "evt_prompt", created: 2, type: "session.inbox.enqueued", durable: { aggregateID: "child", seq: 1, version: 1 }, data: { sessionID: "child", inboxID: "msg_prompt", item: { type: "user", delivery: "steer", payload: { text: "hello" } }, }, } as OpenCodeEvent) expect(store.data.pending.child).toHaveLength(1) expect(store.data.input.child).toEqual(["msg_prompt"]) expect(store.data.session_message.child?.filter((message) => message.id === "msg_prompt")).toHaveLength(1) expect(store.data.message.child?.filter((message) => message.id === "msg_prompt")).toHaveLength(1) expect(store.data.part.msg_prompt).toMatchObject([{ type: "text", text: "hello" }]) }) test("uses the prompt response when the admission event was missed", () => { const store = setup({ child: session("child") }).store store.inbox.echo(promptEcho("msg_prompt")) store.inbox.confirm({ id: "msg_prompt", sessionID: "child", timeCreated: 2, type: "user", delivery: "steer", payload: { text: "hello" }, }) store.applyV2({ id: "evt_delivered", created: Date.now() + 1, type: "session.inbox.delivered", durable: { aggregateID: "child", seq: 2, version: 1 }, data: { sessionID: "child", inboxID: "msg_prompt" }, } as OpenCodeEvent) expect(store.data.pending.child).toEqual([]) expect(store.data.input.child).toEqual([]) expect(store.data.session_message.child).toMatchObject([{ id: "msg_prompt", type: "user", text: "hello" }]) expect(store.data.message.child?.filter((message) => message.id === "msg_prompt")).toHaveLength(1) expect(store.data.part.msg_prompt).toMatchObject([{ type: "text", text: "hello" }]) }) test("keeps a durable admission when the HTTP request later fails", () => { const store = setup({ child: session("child") }).store store.inbox.echo(promptEcho("msg_prompt")) store.applyV2({ id: "evt_prompt", created: 2, type: "session.inbox.enqueued", durable: { aggregateID: "child", seq: 1, version: 1 }, data: { sessionID: "child", inboxID: "msg_prompt", item: { type: "user", delivery: "steer", payload: { text: "hello" } }, }, } as OpenCodeEvent) expect(store.inbox.clearEcho({ sessionID: "child", messageID: "msg_prompt" })).toBe(false) expect(store.data.pending.child).toHaveLength(1) expect(store.data.message.child?.map((message) => message.id)).toEqual(["msg_prompt"]) }) test("places durable admission after delayed selection events", () => { const store = setup({ child: session("child") }).store store.remember(session("child")) store.inbox.echo(promptEcho("msg_prompt")) store.applyV2({ id: "evt_agent", created: 1, type: "session.agent.selected", durable: { aggregateID: "child", seq: 1, version: 1 }, data: { sessionID: "child", agent: "review" }, } as OpenCodeEvent) store.applyV2({ id: "evt_model", created: 2, type: "session.model.selected", durable: { aggregateID: "child", seq: 2, version: 1 }, data: { sessionID: "child", model: { id: "new-model", providerID: "new-provider" } }, } as OpenCodeEvent) store.applyV2({ id: "evt_prompt", created: 3, type: "session.inbox.enqueued", durable: { aggregateID: "child", seq: 3, version: 1 }, data: { sessionID: "child", inboxID: "msg_prompt", item: { type: "user", delivery: "steer", payload: { text: "hello" } }, }, } as OpenCodeEvent) expect(store.data.session_message.child?.map((message) => message.type)).toEqual([ "agent-switched", "model-switched", "user", ]) expect(store.data.message.child?.find((message) => message.id === "msg_prompt")).toMatchObject({ agent: "review", model: { providerID: "new-provider", modelID: "new-model" }, }) }) test("removes an echoed prompt when submission fails", () => { const store = setup({ child: session("child") }).store store.inbox.echo(promptEcho("msg_prompt")) expect(store.inbox.clearEcho({ sessionID: "child", messageID: "msg_prompt" })).toBe(true) expect(store.data.pending.child).toEqual([]) expect(store.data.input.child).toEqual([]) expect(store.data.session_message.child).toBeUndefined() expect(store.data.message.child).toEqual([]) expect(store.data.part.msg_prompt).toBeUndefined() }) test("removes a response-confirmed echo when the server cancels it", () => { const store = setup({ child: session("child") }).store store.inbox.echo(promptEcho("msg_prompt")) store.inbox.confirm({ id: "msg_prompt", sessionID: "child", timeCreated: 1, type: "user", delivery: "steer", payload: { text: "hello" }, }) store.applyV2({ id: "evt_cancelled", created: 2, type: "session.inbox.cancelled", durable: { aggregateID: "child", seq: 2, version: 1 }, data: { sessionID: "child", inboxID: "msg_prompt" }, } as OpenCodeEvent) expect(store.data.pending.child).toEqual([]) expect(store.data.message.child).toEqual([]) expect(store.data.part.msg_prompt).toBeUndefined() }) test("clears stale parts when the initial page has none", async () => { const pending = deferredResponse() const message = userMessage("message") const part = textPart(message.id, { text: "stale" }) const store = createServerSession(messageClient(pending.promise)) store.apply({ type: "message.updated", properties: { info: message } }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part, time: 1 } }) const loading = store.sync("child") pending.resolve(response([{ info: message, parts: [] }])) await loading expect(store.data.part[message.id]).toBeUndefined() }) test("clears delta buffers for parts omitted by the initial page", async () => { const pending = deferredResponse() const message = userMessage("message") const kept = textPart(message.id, { id: "part-1", text: "kept" }) const removed: Part = { ...kept, id: "part-2", text: "removed" } const store = createServerSession(messageClient(pending.promise)) store.apply({ type: "message.updated", properties: { info: message } }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part: kept, time: 1 } }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part: removed, time: 1 } }) store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: message.id, partID: removed.id, field: "text", delta: " delta" }, }) const loading = store.sync("child") pending.resolve(response([{ info: message, parts: [kept] }])) await loading expect(store.data.part[message.id]).toEqual([kept]) expect(store.data.part_text_accum_delta[removed.id]).toBeUndefined() }) test("clears a stale delta buffer when a refresh replaces its part", async () => { const message = userMessage("message") const stale = textPart(message.id, { text: "stale" }) const fetched = { ...stale, text: "fetched" } const store = createServerSession( messageClient(response([{ info: message, parts: [stale] }]), response([{ info: message, parts: [fetched] }])), ) await store.sync("child") store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: message.id, partID: stale.id, field: "text", delta: " delta" }, }) await store.sync("child", { force: true }) expect(store.data.part[message.id]).toEqual([fetched]) expect(store.data.part_text_accum_delta[stale.id]).toBeUndefined() }) test("preserves a non-durable delta received before refresh", async () => { const message = userMessage("message") const part = textPart(message.id, { text: "stale" }) const store = createServerSession( messageClient(response([{ info: message, parts: [part] }]), response([{ info: message, parts: [{ ...part }] }])), ) await store.sync("child") store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: message.id, partID: part.id, field: "text", delta: " delta" }, }) await store.sync("child", { force: true }) expect(store.data.part[message.id]).toEqual([{ ...part, text: "stale delta" }]) expect(store.data.part_text_accum_delta[part.id]).toBe("stale delta") }) test("accepts fetched text that intentionally replaces an accumulated prefix", async () => { const message = userMessage("message") const part = textPart(message.id, { text: "abc" }) const fetched = { ...part, text: "ab" } const store = createServerSession( messageClient(response([{ info: message, parts: [part] }]), response([{ info: message, parts: [fetched] }])), ) await store.sync("child") store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: message.id, partID: part.id, field: "text", delta: "def" }, }) await store.sync("child", { force: true }) expect(store.data.part[message.id]).toEqual([fetched]) expect(store.data.part_text_accum_delta[part.id]).toBeUndefined() }) test("preserves an unpersisted delta suffix after partial server catch-up", async () => { const message = userMessage("message") const part = textPart(message.id, { text: "a" }) const fetched = { ...part, text: "ab" } const store = createServerSession( messageClient(response([{ info: message, parts: [part] }]), response([{ info: message, parts: [fetched] }])), ) await store.sync("child") store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: message.id, partID: part.id, field: "text", delta: "bc" }, }) await store.sync("child", { force: true }) expect(store.data.part[message.id]).toEqual([{ ...part, text: "abc" }]) expect(store.data.part_text_accum_delta[part.id]).toBe("abc") }) test("clears delta state after exact server catch-up", async () => { const message = userMessage("message") const part = textPart(message.id, { text: "a" }) const fetched = { ...part, text: "ab" } const store = createServerSession( messageClient(response([{ info: message, parts: [part] }]), response([{ info: message, parts: [fetched] }])), ) await store.sync("child") store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: message.id, partID: part.id, field: "text", delta: "b" }, }) await store.sync("child", { force: true }) expect(store.data.part[message.id]).toEqual([fetched]) expect(store.data.part_text_accum_delta[part.id]).toBeUndefined() }) test("uses the successful retry response over events from a failed attempt", async () => { const failed = Promise.withResolvers() const retried = Promise.withResolvers() const message = userMessage("message") const stale = textPart(message.id, { text: "stale" }) const intermediate = { ...stale, text: "intermediate" } const fetched = { ...stale, text: "fetched" } const client = messageClient(failed.promise, retried.promise) const store = createServerSession(client, { retry: retryImmediately }) store.apply({ type: "message.updated", properties: { info: message } }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part: stale, time: 1 } }) const loading = store.sync("child") store.apply({ type: "message.part.updated", properties: { sessionID: "child", part: intermediate, time: 2 } }) failed.reject(new Error("failed to fetch")) await client.requested(2) retried.resolve(response([{ info: message, parts: [fetched] }])) await loading expect(store.data.part[message.id]).toEqual([fetched]) }) test("preserves non-durable deltas across message retries", async () => { const failed = Promise.withResolvers() const retried = Promise.withResolvers() const message = userMessage("message") const part = textPart(message.id, { text: "stale" }) const client = messageClient(failed.promise, retried.promise) const store = createServerSession(client, { retry: retryImmediately }) store.apply({ type: "message.updated", properties: { info: message } }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part, time: 1 } }) const loading = store.sync("child") store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: message.id, partID: part.id, field: "text", delta: " delta" }, }) failed.reject(new Error("failed to fetch")) await client.requested(2) retried.resolve(response([{ info: message, parts: [part] }])) await loading expect(store.data.part[message.id]).toEqual([{ ...part, text: "stale delta" }]) }) test("preserves part removals across message retries", async () => { const failed = Promise.withResolvers() const retried = Promise.withResolvers() const message = userMessage("message") const part = textPart(message.id) const client = messageClient(response([{ info: message, parts: [part] }]), failed.promise, retried.promise) const store = createServerSession(client, { retry: retryImmediately }) await store.sync("child") const loading = store.sync("child", { force: true }) store.apply({ type: "message.part.removed", properties: { sessionID: "child", messageID: message.id, partID: part.id }, }) failed.reject(new Error("failed to fetch")) await client.requested(3) retried.resolve(response([{ info: message, parts: [part] }])) await loading expect(store.data.part[message.id]).toBeUndefined() }) test("preserves message removals across message retries", async () => { const failed = Promise.withResolvers() const retried = Promise.withResolvers() const message = userMessage("message") const part = textPart(message.id) const client = messageClient(response([{ info: message, parts: [part] }]), failed.promise, retried.promise) const store = createServerSession(client, { retry: retryImmediately }) await store.sync("child") const loading = store.sync("child", { force: true }) store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: message.id } }) failed.reject(new Error("failed to fetch")) await client.requested(3) retried.resolve(response([{ info: message, parts: [part] }])) await loading expect(store.data.message.child).toEqual([]) expect(store.data.part[message.id]).toBeUndefined() }) test("accepts part omission from a successful retry after an earlier delta", async () => { const failed = Promise.withResolvers() const retried = Promise.withResolvers() const message = userMessage("message") const part = textPart(message.id) const client = messageClient(response([{ info: message, parts: [part] }]), failed.promise, retried.promise) const store = createServerSession(client, { retry: retryImmediately }) await store.sync("child") const loading = store.sync("child", { force: true }) store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: message.id, partID: part.id, field: "text", delta: " delta" }, }) failed.reject(new Error("failed to fetch")) await client.requested(3) retried.resolve(response([{ info: message, parts: [] }])) await loading expect(store.data.part[message.id]).toBeUndefined() expect(store.data.part_text_accum_delta[part.id]).toBeUndefined() }) test("clears load-owned orphan parts when all retries fail", async () => { const first = Promise.withResolvers() const second = Promise.withResolvers() const third = Promise.withResolvers() const message = userMessage("message") const part = textPart(message.id) const client = messageClient(first.promise, second.promise, third.promise) const store = createServerSession(client, { retry: retryImmediately }) const loading = store.sync("child").catch((error) => error) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part, time: 2 } }) first.reject(new Error("failed to fetch")) await client.requested(2) second.reject(new Error("failed to fetch")) await client.requested(3) third.reject(new Error("failed to fetch")) await loading expect(store.data.part[message.id]).toBeUndefined() }) test("preserves live updates during a forced refresh", async () => { const pending = deferredResponse() const stale = userMessage("message") const stalePart = textPart(stale.id, { text: "stale" }) const store = createServerSession(messageClient(response([{ info: stale, parts: [stalePart] }]), pending.promise)) await store.sync("child") const refreshing = store.sync("child", { force: true }) const live = { ...stale, time: { created: 2 } } store.apply({ type: "message.updated", properties: { info: live } }) store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: stale.id, partID: stalePart.id, field: "text", delta: " live" }, }) pending.resolve(response([{ info: stale, parts: [stalePart] }])) await refreshing expect(store.data.message.child).toEqual([live]) expect(store.data.part[stale.id]).toEqual([{ ...stalePart, text: "stale live" }]) }) test("keeps fetched message metadata when only a part changes", async () => { const pending = deferredResponse() const stale = userMessage("message") const fetched = { ...stale, time: { created: 2 } } const part = textPart(stale.id, { text: "stale" }) const store = createServerSession(messageClient(response([{ info: stale, parts: [part] }]), pending.promise)) await store.sync("child") const refreshing = store.sync("child", { force: true }) store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: stale.id, partID: part.id, field: "text", delta: " live" }, }) pending.resolve(response([{ info: fetched, parts: [part] }])) await refreshing expect(store.data.message.child).toEqual([fetched]) expect(store.data.part[stale.id]).toEqual([{ ...part, text: "stale live" }]) }) test("preserves a part update when a forced refresh omits its message", async () => { const pending = deferredResponse() const message = userMessage("message") const stale = textPart(message.id, { text: "stale" }) const live = { ...stale, text: "live" } const store = createServerSession(messageClient(response([{ info: message, parts: [stale] }]), pending.promise)) await store.sync("child") const refreshing = store.sync("child", { force: true }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part: live, time: 2 } }) pending.resolve(response()) await refreshing expect(store.data.message.child).toEqual([message]) expect(store.data.part[message.id]).toEqual([live]) }) test("ignores a late part update after its message is removed", async () => { const pending = deferredResponse() const message = userMessage("message") const part = textPart(message.id) const store = createServerSession(messageClient(pending.promise)) const loading = store.sync("child") store.apply({ type: "message.updated", properties: { info: message } }) store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: message.id } }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part, time: 2 } }) pending.resolve(response([{ info: message, parts: [part] }])) await loading expect(store.data.message.child).toEqual([]) expect(store.data.part[message.id]).toBeUndefined() }) test("ignores a late part update after a completed message removal", () => { const message = userMessage("message") const part = textPart(message.id) const store = setup({ child: session("child") }).store store.apply({ type: "message.updated", properties: { info: message } }) store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: message.id } }) store.apply({ type: "message.part.updated", properties: { sessionID: "child", part, time: 2 } }) expect(store.data.part[message.id]).toBeUndefined() }) test("does not restore a completed message removal from a stale refresh", async () => { const message = userMessage("message") const part = textPart(message.id) const store = createServerSession( messageClient(response([{ info: message, parts: [part] }]), response([{ info: message, parts: [part] }])), ) await store.sync("child") store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: message.id } }) await store.sync("child", { force: true }) expect(store.data.message.child).toEqual([]) expect(store.data.part[message.id]).toBeUndefined() }) test("does not restore a completed part removal from a stale refresh", async () => { const message = userMessage("message") const part = textPart(message.id) const store = createServerSession( messageClient(response([{ info: message, parts: [part] }]), response([{ info: message, parts: [part] }])), ) await store.sync("child") store.apply({ type: "message.part.removed", properties: { sessionID: "child", messageID: message.id, partID: part.id }, }) await store.sync("child", { force: true }) expect(store.data.part[message.id]).toBeUndefined() }) test("preserves removals during history prepend", async () => { const pending = deferredResponse() const latest = userMessage("message-2", { time: { created: 2 } }) const older = { ...latest, id: "message-1", time: { created: 1 } } const store = createServerSession(messageClient(response([{ info: latest, parts: [] }], "older"), pending.promise)) await store.sync("child") const loading = store.history.loadMore("child") store.apply({ type: "message.removed", properties: { sessionID: "child", messageID: older.id } }) pending.resolve(response([{ info: older, parts: [] }])) await loading 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") guard.active = false 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 } }) const fresh = userMessage("message-3", { time: { created: 3 } }) const store = createServerSession( messageClient( response( [ { info: older, parts: [] }, { info: latest, parts: [] }, ], "older", ), response( [ { info: latest, parts: [] }, { info: fresh, parts: [] }, ], "older", ), ), ) await store.sync("child") await store.sync("child", { force: true }) expect(store.data.message.child).toEqual([older, latest, fresh]) }) test("drops stale recent messages omitted by an incomplete refresh", async () => { const third = userMessage("message-3", { time: { created: 3 } }) const fourth = userMessage("message-4", { time: { created: 4 } }) const stale = userMessage("message-5", { time: { created: 5 } }) const store = createServerSession( messageClient( response( [ { info: fourth, parts: [] }, { info: stale, parts: [] }, ], "older", ), response( [ { info: third, parts: [] }, { info: fourth, parts: [] }, ], "older", ), ), ) await store.sync("child") await store.sync("child", { force: true }) expect(store.data.message.child).toEqual([third, fourth]) }) test("uses message creation time for incomplete refresh boundaries", async () => { const older = userMessage("msg_z", { time: { created: 1 } }) const boundary = userMessage("msg_m", { time: { created: 2 } }) const stale = userMessage("msg_a", { time: { created: 3 } }) const store = createServerSession( messageClient( response( [ { info: older, parts: [] }, { info: stale, parts: [] }, ], "older", ), response([{ info: boundary, parts: [] }], "older"), ), ) await store.sync("child") await store.sync("child", { force: true }) expect(store.data.message.child).toEqual([older, boundary]) }) test("preserves a part update for a message being loaded from history", async () => { const pending = deferredResponse() const latest = userMessage("message-2", { time: { created: 2 } }) const older = userMessage("message-1") const stale = textPart(older.id, { text: "stale" }) const live = { ...stale, text: "live" } const store = createServerSession(messageClient(response([{ info: latest, parts: [] }], "older"), pending.promise)) await store.sync("child") const loading = store.history.loadMore("child") store.apply({ type: "message.part.updated", properties: { sessionID: "child", part: live, time: 2 } }) pending.resolve(response([{ info: older, parts: [stale] }])) await loading expect(store.data.part[older.id]).toEqual([live]) }) test("does not clear newer orphan parts after terminal history prepend", async () => { const pending = deferredResponse() const latest = userMessage("message-2", { time: { created: 2 } }) const older = userMessage("message-1") const newer = userMessage("message-3", { time: { created: 3 } }) const part = textPart(newer.id, { text: "live" }) const store = createServerSession(messageClient(response([{ info: latest, parts: [] }], "older"), pending.promise)) await store.sync("child") const loading = store.history.loadMore("child") store.apply({ type: "message.part.updated", properties: { sessionID: "child", part, time: 3 } }) pending.resolve(response([{ info: older, parts: [] }])) await loading store.apply({ type: "message.updated", properties: { sessionID: "child", info: newer } }) expect(store.data.part[newer.id]).toEqual([part]) }) test("accepts an authoritative history part after an earlier unknown-parent update", async () => { const pending = deferredResponse() const history = deferredResponse() const latest = userMessage("message-2", { time: { created: 2 } }) const older = userMessage("message-1") const part = textPart(older.id, { text: "live" }) const store = createServerSession(messageClient(pending.promise, history.promise)) const loading = store.sync("child") store.apply({ type: "message.part.updated", properties: { sessionID: "child", part, time: 2 } }) pending.resolve(response([{ info: latest, parts: [] }], "older")) await loading expect(store.data.part[older.id]).toEqual([part]) const loadingHistory = store.history.loadMore("child") history.resolve(response([{ info: older, parts: [{ ...part, text: "stale" }] }])) await loadingHistory expect(store.data.part[older.id]).toEqual([{ ...part, text: "stale" }]) }) test("preserves an unknown-parent part removal across pages", async () => { const initial = deferredResponse() const history = deferredResponse() const latest = userMessage("message-2", { time: { created: 2 } }) const older = userMessage("message-1") const part = textPart(older.id) const store = createServerSession(messageClient(initial.promise, history.promise)) const loading = store.sync("child") store.apply({ type: "message.part.removed", properties: { sessionID: "child", messageID: older.id, partID: part.id }, }) initial.resolve(response([{ info: latest, parts: [] }], "older")) await loading const loadingHistory = store.history.loadMore("child") history.resolve(response([{ info: older, parts: [part] }])) await loadingHistory expect(store.data.part[older.id]).toBeUndefined() }) test("clears orphaned parts when a refresh drops a message", async () => { const message = userMessage("message") const part = textPart(message.id, { text: "stale" }) const store = createServerSession(messageClient(response([{ info: message, parts: [part] }]), response())) await store.sync("child") store.apply({ type: "message.part.delta", properties: { sessionID: "child", messageID: message.id, partID: part.id, field: "text", delta: " delta" }, }) await store.sync("child", { force: true }) expect(store.data.message.child).toEqual([]) expect(store.data.part[message.id]).toBeUndefined() expect(store.data.part_text_accum_delta[part.id]).toBeUndefined() }) test("applies events without a directory store", () => { const ctx = setup({}) ctx.store.apply({ type: "session.created", properties: { sessionID: "root", info: session("root") } }) ctx.store.apply({ type: "session.status", properties: { sessionID: "root", status: { type: "busy" } } }) expect(ctx.store.get("root")?.location.directory).toBe("/repo") expect(ctx.store.data.session_working("root")).toBe(true) expect(ctx.get).toEqual([]) }) test("preserves pinned session content under server-wide cache pressure", () => { const ctx = setup({}) ctx.store.pin("active") ctx.store.inbox.echo({ ...promptEcho("message", "keep"), sessionID: "active" }) for (let index = 0; index < 50; index++) { ctx.store.remember(session(`session-${index}`)) ctx.store.apply({ type: "session.status", properties: { sessionID: `session-${index}`, status: { type: "idle" } }, }) } expect(ctx.store.data.message.active?.map((message) => message.id)).toEqual(["message"]) expect(ctx.store.data.session_status["session-0"]).toBeUndefined() }) })