perf(client): streamed row updates no longer rescan question history

packages/client-runtime/src/state/threadRequests.test.ts

@@ -2,6 +2,8 @@ import { presentThreadShell } from "./models.ts";
 import { threadKey } from "./entities.ts";
 import {
   EnvironmentId,
+  EventId,
+  MessageId,
   ThreadId,
   type ScopedThreadRef,
   NodeId,
@@ -13,6 +15,7 @@ import {
 import { describe, expect, it, vi } from "vite-plus/test";
 import { Atom, AtomRegistry, AsyncResult } from "effect/reactivity";
 import * as Cause from "effect/Cause";
+import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts";
 import { v2Now, v2Projection, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
 import {
   createQuestionHistoryProjector,
@@ -360,6 +363,210 @@ it("restores old text answers without mutating history or replacing unchanged ro
   expect(project({ ...answered, visibleTurnItems: [...rows] })[0]).toBe(result[0]);
 });
 
+type QuestionItem = Extract<
+  OrchestrationV2ThreadProjection["turnItems"][number],
+  { readonly type: "user_input_request" }
+>;
+
+describe("question history through exact row replacements", () => {
+  const baseItem = {
+    threadId: v2Projection.thread.id,
+    runId: null,
+    nodeId,
+    providerThreadId: null,
+    providerTurnId: null,
+    nativeItemRef: null,
+    parentItemId: null,
+    status: "running" as const,
+    title: null,
+    startedAt: v2Now,
+    completedAt: null,
+    updatedAt: v2Now,
+  };
+  const assistantItem = (ordinal: number, text = "Hel") => ({
+    ...baseItem,
+    id: TurnItemId.make(`assistant-${ordinal}`),
+    ordinal,
+    type: "assistant_message" as const,
+    messageId: MessageId.make(`message-${ordinal}`),
+    text,
+    streaming: true,
+  });
+  const localRow = (
+    item: OrchestrationV2ThreadProjection["turnItems"][number],
+    position: number,
+  ) => ({
+    item,
+    position,
+    sourceItemId: item.id,
+    sourceThreadId: item.threadId,
+    visibility: "local" as const,
+  });
+  const updated = (
+    current: OrchestrationV2ThreadProjection,
+    item: OrchestrationV2ThreadProjection["turnItems"][number],
+  ) =>
+    applyOrchestrationV2ProjectionEvent(current, {
+      id: EventId.make(`event-${item.id}`),
+      type: "turn-item.updated",
+      threadId: current.thread.id,
+      occurredAt: v2Now,
+      payload: item,
+    })!;
+
+  it("reads only the replaced row and leaves the request index alone", () => {
+    const size = 1_000;
+    const counts = { items: 0, requestIds: 0 };
+    const items = Array.from({ length: size }, (_, index) => assistantItem(index));
+    const countedRows = items.map((item, index) => {
+      const row = localRow(item, index);
+      return Object.defineProperty(row, "item", {
+        enumerable: true,
+        get: () => {
+          counts.items += 1;
+          return item;
+        },
+      });
+    });
+    const countedRequests = Array.from({ length: 400 }, (_, index) => {
+      const request = {
+        ...projection.runtimeRequests[0]!,
+        id: RuntimeRequestId.make(`request-${index}`),
+      };
+      return Object.defineProperty(request, "id", {
+        enumerable: true,
+        get: () => {
+          counts.requestIds += 1;
+          return RuntimeRequestId.make(`request-${index}`);
+        },
+      });
+    });
+    const initial: OrchestrationV2ThreadProjection = {
+      ...v2Projection,
+      turnItems: items,
+      visibleTurnItems: countedRows,
+      runtimeRequests: countedRequests,
+    };
+    const project = createQuestionHistoryProjector();
+    const scan = (current: OrchestrationV2ThreadProjection) => {
+      counts.items = 0;
+      counts.requestIds = 0;
+      const result = project(current);
+      return { result, ...counts };
+    };
+    expect(scan(initial)).toMatchObject({ items: size, requestIds: 400 });
+
+    const adjacent = updated(initial, assistantItem(size - 1, "Hello"));
+    const hinted = scan(adjacent);
+    expect(hinted.result).toBe(adjacent.visibleTurnItems);
+    expect(hinted).toMatchObject({ items: 0, requestIds: 0 });
+
+    // A skipped delta has no hint for this exact pair, so it scans again.
+    const skipped = updated(
+      updated(adjacent, assistantItem(size - 1, "Hello t")),
+      assistantItem(size - 1, "Hello there"),
+    );
+    const fallback = scan(skipped);
+    expect(fallback.result).toBe(skipped.visibleTurnItems);
+    expect(fallback).toMatchObject({ items: size - 1, requestIds: 0 });
+  });
+
+  it("matches a full projection through streamed rows, request changes and answers", () => {
+    const question = projection.turnItems[0] as QuestionItem;
+    const resolved = (answers: Record<string, string>) => ({
+      ...projection.runtimeRequests[0]!,
+      status: "resolved" as const,
+      answers,
+    });
+    const explicitRequestId = RuntimeRequestId.make("explicit-question");
+    const explicitAnswer = {
+      requestId: explicitRequestId,
+      answers: { next: "Explicit" },
+      attachmentsByQuestionId: {},
+      questionTextById: { next: "What should happen next?" },
+    };
+    const unanswered: QuestionItem = {
+      ...question,
+      id: TurnItemId.make("explicit-question-item"),
+      ordinal: 1,
+      requestId: explicitRequestId,
+    };
+    const explicit: QuestionItem = { ...unanswered, questionAnswer: explicitAnswer };
+    const explicitRequest = {
+      ...resolved({ next: "From request" }),
+      id: explicitRequestId,
+    };
+    const stream = assistantItem(2);
+    const inherited = {
+      ...localRow({ ...question, id: TurnItemId.make("inherited-question-item") }, 3),
+      visibility: "inherited" as const,
+      sourceThreadId: ThreadId.make("parent-thread"),
+    };
+    const turnItems = [question, explicit, stream];
+    let current: OrchestrationV2ThreadProjection = {
+      ...v2Projection,
+      turnItems,
+      visibleTurnItems: [...turnItems.map(localRow), inherited],
+      runtimeRequests: [projection.runtimeRequests[0]!, explicitRequest],
+    };
+    const project = createQuestionHistoryProjector();
+    const step = (next: OrchestrationV2ThreadProjection) => {
+      current = next;
+      const result = project(next);
+      expect(result).toEqual(createQuestionHistoryProjector()(next));
+      return result;
+    };
+    const answerOf = (rows: ReadonlyArray<{ readonly item: unknown }>, index: number) =>
+      (rows[index]!.item as { readonly questionAnswer?: { readonly answers: unknown } })
+        .questionAnswer?.answers;
+
+    expect(step(current)).toBe(current.visibleTurnItems);
+    expect(step(updated(current, assistantItem(2, "Hello")))).toBe(current.visibleTurnItems);
+
+    // The last duplicate request wins, and every row asking it gets the answer.
+    const answered = step({
+      ...current,
+      runtimeRequests: [
+        projection.runtimeRequests[0]!,
+        resolved({ next: "Continue" }),
+        explicitRequest,
+      ],
+    });
+    expect(answerOf(answered, 0)).toEqual({ next: "Continue" });
+    expect(answerOf(answered, 1)).toEqual({ next: "Explicit" });
+    expect(answerOf(answered, 3)).toEqual({ next: "Continue" });
+
+    const whileOverlaid = step(updated(current, assistantItem(2, "Hello there")));
+    expect(whileOverlaid[0]).toBe(answered[0]);
+    expect(whileOverlaid[3]).toBe(answered[3]);
+
+    const changedAnswer = step({
+      ...current,
+      runtimeRequests: [resolved({ next: "Stop" }), explicitRequest],
+    });
+    expect(answerOf(changedAnswer, 0)).toEqual({ next: "Stop" });
+
+    const reopened = step({
+      ...current,
+      runtimeRequests: [projection.runtimeRequests[0]!, explicitRequest],
+    });
+    expect(reopened).toBe(current.visibleTurnItems);
+
+    // Dropping an explicit answer falls back to its resolved request.
+    const restored = step(updated(current, unanswered));
+    expect(answerOf(restored, 1)).toEqual({ next: "From request" });
+
+    const retitled: QuestionItem = {
+      ...unanswered,
+      questions: [{ ...question.questions[0]!, question: "Now what?" }],
+    };
+    const retitledRows = step(updated(current, retitled));
+    expect(retitledRows[1]!.item).toMatchObject({
+      questionAnswer: { questionTextById: { next: "Now what?" } },
+    });
+  });
+});
+
 it("loads only child approvals and retains their destination across environments", () => {
   const local = EnvironmentId.make("local");
   const remote = EnvironmentId.make("remote");

packages/client-runtime/src/state/threadRequests.ts

@@ -13,6 +13,7 @@ import type {
 } from "@t3tools/contracts";
 import { Atom } from "effect/reactivity";
 import { parseThreadKey, threadKey } from "./entities.ts";
+import { getProjectionArrayReplacementIndex } from "./orchestrationV2Projection.ts";
 import type { EnvironmentThreadShell } from "./models.ts";
 import * as DateTime from "effect/DateTime";
 import type { AtomCommandResult } from "./runtime.ts";
@@ -292,54 +293,61 @@ export function createQuestionHistoryProjector<
   let previousRows: Rows | undefined;
   let previousRequests: Requests | undefined;
   let previousResult: Rows | undefined;
+  let byId = new Map<RuntimeRequestId, Requests[number]>();
   const cache = new WeakMap<Row, { request: Requests[number]; row: Row }>();
-  return (projection: {
-    readonly visibleTurnItems: Rows;
-    readonly runtimeRequests: Requests;
-  }): Rows => {
-    const { visibleTurnItems: rows, runtimeRequests: requests } = projection;
-    if (rows === previousRows && requests === previousRequests && previousResult)
-      return previousResult;
-    const byId = new Map(requests.map((request) => [request.id, request]));
+  const projectRow = (row: Row): Row => {
+    const item = row.item;
+    if (item.type !== "user_input_request" || item.questionAnswer) return row;
+    const request = byId.get(item.requestId);
+    if (request?.status !== "resolved" || request.answers === undefined) return row;
+    const cached = cache.get(row);
+    const next: Row =
+      cached?.request === request
+        ? cached.row
+        : {
+            ...row,
+            item: {
+              ...item,
+              questionAnswer: {
+                requestId: request.id,
+                answers: request.answers,
+                attachmentsByQuestionId: {},
+                questionTextById: Object.fromEntries(
+                  item.questions.map((question) => [question.id, question.question]),
+                ),
+              },
+            },
+          };
+    cache.set(row, { request, row: next });
+    return next;
+  };
+  const projectRows = (rows: Rows): Rows => {
     let result: Row[] | undefined;
     for (const [index, row] of rows.entries()) {
-      const item = row.item;
-      const request =
-        item.type === "user_input_request" && !item.questionAnswer
-          ? byId.get(item.requestId)
-          : undefined;
-      let next = row;
-      if (
-        item.type === "user_input_request" &&
-        request?.status === "resolved" &&
-        request.answers !== undefined
-      ) {
-        const cached = cache.get(row);
-        next =
-          cached?.request === request
-            ? cached.row
-            : {
-                ...row,
-                item: {
-                  ...item,
-                  questionAnswer: {
-                    requestId: request.id,
-                    answers: request.answers,
-                    attachmentsByQuestionId: {},
-                    questionTextById: Object.fromEntries(
-                      item.questions.map((question) => [question.id, question.question]),
-                    ),
-                  },
-                },
-              };
-        cache.set(row, { request, row: next });
-      }
+      const next = projectRow(row);
       if (next !== row) result ??= rows.slice(0, index);
       result?.push(next);
     }
+    return result ?? rows;
+  };
+  return (projection: {
+    readonly visibleTurnItems: Rows;
+    readonly runtimeRequests: Requests;
+  }): Rows => {
+    const { visibleTurnItems: rows, runtimeRequests: requests } = projection;
+    const sameRequests = requests === previousRequests;
+    if (rows === previousRows && sameRequests && previousResult) return previousResult;
+    if (!sameRequests) byId = new Map(requests.map((request) => [request.id, request]));
+    // Nothing was overlaid and requests are unchanged, so only an exactly replaced row can differ.
+    const replacedIndex =
+      sameRequests && previousRows !== undefined && previousResult === previousRows
+        ? getProjectionArrayReplacementIndex(previousRows, rows)
+        : undefined;
+    const unchanged =
+      replacedIndex !== undefined && projectRow(rows[replacedIndex]!) === rows[replacedIndex];
     previousRows = rows;
     previousRequests = requests;
-    previousResult = result ?? rows;
+    previousResult = unchanged ? rows : projectRows(rows);
     return previousResult;
   };
 }