State-machine thread history loading

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<ThreadHistoryLoadLifecycleState, { kind: "loading" }>;
+
 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<void> {
     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;
   }
 }

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<ThreadTurnsListResponse>();
+    const second = deferred<ThreadTurnsListResponse>();
+    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<ThreadTurnsListResponse>();
+    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<ReturnType<AppServerClient["threadTurnsList"]>>;
+
+function historyFixture(options: { threadTurnsList: ReturnType<typeof vi.fn> }) {
+  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<T>(): { promise: Promise<T>; resolve: (value: T) => void; reject: (error: unknown) => void } {
+  let resolve!: (value: T) => void;
+  let reject!: (error: unknown) => void;
+  const promise = new Promise<T>((promiseResolve, promiseReject) => {
+    resolve = promiseResolve;
+    reject = promiseReject;
+  });
+  return { promise, resolve, reject };
+}