diff --git a/src/features/chat/thread-history.ts b/src/features/chat/thread-history.ts index 90167f34..fdcdfe84 100644 --- a/src/features/chat/thread-history.ts +++ b/src/features/chat/thread-history.ts @@ -12,8 +12,11 @@ export interface ThreadHistoryLoaderHost { setThreadTurnPresence: (hadTurns: boolean) => void; } +type ThreadHistoryLoadLifecycleState = { kind: "idle" } | { kind: "loading"; threadId: string; mode: "latest" | "older" }; +type ActiveThreadHistoryLoad = Extract; + export class ThreadHistoryLoader { - private generation = 0; + private lifecycle: ThreadHistoryLoadLifecycleState = { kind: "idle" }; constructor(private readonly host: ThreadHistoryLoaderHost) {} @@ -26,30 +29,25 @@ export class ThreadHistoryLoader { } invalidate(): void { - this.generation += 1; + this.lifecycle = { kind: "idle" }; this.dispatch({ type: "history/loading-set", loading: false }); } async loadLatest(threadId = this.state.activeThreadId): Promise { const client = this.host.currentClient(); if (!client || !threadId) return; - const generation = ++this.generation; - this.dispatch({ type: "history/loading-set", loading: true }); - this.host.render(); + const load = this.startLoading(threadId, "latest"); try { const response = await client.threadTurnsList(threadId, null, 20); - if (this.isStale(generation, threadId)) return; + if (this.isStale(load)) return; this.host.setThreadTurnPresence(response.data.length > 0); this.dispatch({ type: "display/items-replaced", items: displayItemsFromTurns(response.data), historyCursor: response.nextCursor }); this.host.forceMessagesToBottom(); } catch (error) { - if (this.isStale(generation, threadId)) return; + if (this.isStale(load)) return; this.host.addSystemMessage(error instanceof Error ? error.message : String(error)); } finally { - if (!this.isStale(generation, threadId)) { - this.dispatch({ type: "history/loading-set", loading: false }); - this.host.render(); - } + this.finishLoading(load); } } @@ -59,12 +57,10 @@ export class ThreadHistoryLoader { if (!client || !state.activeThreadId || !state.historyCursor || state.loadingHistory) return; const threadId = state.activeThreadId; const cursor = state.historyCursor; - const generation = ++this.generation; - this.dispatch({ type: "history/loading-set", loading: true }); - this.host.render(); + const load = this.startLoading(threadId, "older"); try { const response = await client.threadTurnsList(threadId, cursor, 20); - if (this.isStale(generation, threadId)) return; + if (this.isStale(load)) return; const current = this.state; const olderItems = displayItemsFromTurns(response.data); const existingIds = new Set(current.displayItems.map((item) => item.id)); @@ -76,17 +72,29 @@ export class ThreadHistoryLoader { messagesPinnedToBottom: false, }); } catch (error) { - if (this.isStale(generation, threadId)) return; + if (this.isStale(load)) return; this.host.addSystemMessage(error instanceof Error ? error.message : String(error)); } finally { - if (!this.isStale(generation, threadId)) { - this.dispatch({ type: "history/loading-set", loading: false }); - this.host.render(); - } + this.finishLoading(load); } } - private isStale(generation: number, threadId: string): boolean { - return generation !== this.generation || this.state.activeThreadId !== threadId; + private startLoading(threadId: string, mode: ActiveThreadHistoryLoad["mode"]): ActiveThreadHistoryLoad { + const load: ActiveThreadHistoryLoad = { kind: "loading", threadId, mode }; + this.lifecycle = load; + this.dispatch({ type: "history/loading-set", loading: true }); + this.host.render(); + return load; + } + + private finishLoading(load: ActiveThreadHistoryLoad): void { + if (this.isStale(load)) return; + this.lifecycle = { kind: "idle" }; + this.dispatch({ type: "history/loading-set", loading: false }); + this.host.render(); + } + + private isStale(load: ActiveThreadHistoryLoad): boolean { + return this.lifecycle !== load || this.state.activeThreadId !== load.threadId; } } diff --git a/tests/features/chat/thread-history.test.ts b/tests/features/chat/thread-history.test.ts new file mode 100644 index 00000000..76007746 --- /dev/null +++ b/tests/features/chat/thread-history.test.ts @@ -0,0 +1,100 @@ +import { describe, expect, it, vi } from "vitest"; + +import { createChatState, createChatStateStore } from "../../../src/features/chat/chat-state"; +import { ThreadHistoryLoader } from "../../../src/features/chat/thread-history"; +import type { AppServerClient } from "../../../src/app-server/client"; +import type { ThreadItem } from "../../../src/generated/app-server/v2/ThreadItem"; +import type { Turn } from "../../../src/generated/app-server/v2/Turn"; + +describe("ThreadHistoryLoader", () => { + it("keeps the latest history load when an older request resolves later", async () => { + const first = deferred(); + const second = deferred(); + const { loader, stateStore } = historyFixture({ + threadTurnsList: vi.fn().mockReturnValueOnce(first.promise).mockReturnValueOnce(second.promise), + }); + + const firstLoad = loader.loadLatest(); + const secondLoad = loader.loadLatest(); + + second.resolve(threadTurnsResponse([], "second-cursor")); + await secondLoad; + first.resolve(threadTurnsResponse([], "first-cursor")); + await firstLoad; + + expect(stateStore.getState().historyCursor).toBe("second-cursor"); + expect(stateStore.getState().loadingHistory).toBe(false); + }); + + it("ignores a history load that is invalidated while pending", async () => { + const pending = deferred(); + const { loader, stateStore, addSystemMessage } = historyFixture({ + threadTurnsList: vi.fn().mockReturnValue(pending.promise), + }); + + const loading = loader.loadLatest(); + expect(stateStore.getState().loadingHistory).toBe(true); + + loader.invalidate(); + pending.resolve(threadTurnsResponse([turnFixture([assistantMessage("assistant", "Stale")])], "stale-cursor")); + await loading; + + expect(stateStore.getState().displayItems).toEqual([]); + expect(stateStore.getState().historyCursor).toBeNull(); + expect(stateStore.getState().loadingHistory).toBe(false); + expect(addSystemMessage).not.toHaveBeenCalled(); + }); +}); + +type ThreadTurnsListResponse = Awaited>; + +function historyFixture(options: { threadTurnsList: ReturnType }) { + const state = createChatState(); + state.activeThreadId = "thread"; + const stateStore = createChatStateStore(state); + const addSystemMessage = vi.fn(); + const loader = new ThreadHistoryLoader({ + stateStore, + currentClient: () => + ({ + threadTurnsList: options.threadTurnsList, + }) as unknown as AppServerClient, + render: vi.fn(), + addSystemMessage, + forceMessagesToBottom: vi.fn(), + keepCurrentScrollPosition: vi.fn(), + setThreadTurnPresence: vi.fn(), + }); + return { loader, stateStore, addSystemMessage }; +} + +function threadTurnsResponse(data: Turn[], nextCursor: string | null): ThreadTurnsListResponse { + return { data, nextCursor, backwardsCursor: null }; +} + +function turnFixture(items: ThreadItem[]): Turn { + return { + id: "turn", + items, + itemsView: "full", + status: "completed", + error: null, + startedAt: 1, + completedAt: 2, + durationMs: 1000, + }; +} + +function assistantMessage(id: string, text: string): ThreadItem { + return { type: "agentMessage", id, text, phase: "final_answer", memoryCitation: null }; +} + +function deferred(): { promise: Promise; resolve: (value: T) => void; reject: (error: unknown) => void } { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((promiseResolve, promiseReject) => { + resolve = promiseResolve; + reject = promiseReject; + }); + return { promise, resolve, reject }; +}