perf(ui): incrementally materialize appended transcript turns

packages/ui/src/__tests__/transcript-projection.test.ts

@@ -23,7 +23,7 @@ import type { ShellRunSnapshotResult, ShellRunUpdate } from '@maka/core/events';
 import type { ShellRunToolResult } from '@maka/core/shell-run-result';
 import type { StoredMessage } from '@maka/core/session';
 import { createTranscriptProjection, valuesEqual } from '../transcript-projection.js';
-import { foldShellRunToolActivities, timelineTools, type ToolActivityItem, type TurnViewModel } from '../materialize.js';
+import { foldShellRunToolActivities, materializeTurns, timelineTools, type ToolActivityItem, type TurnViewModel } from '../materialize.js';
 import type { LiveTurnProjection } from '../live-turn-projection.js';
 
 const REF = 'maka://runtime/background-tasks/pty-1';
@@ -67,6 +67,173 @@ function streamingTurn(text: string): LiveTurnProjection {
 }
 
 describe('incremental transcript projection', () => {
+  test('a durable append does not reread unchanged historical message contents', () => {
+    let historicalReads = 0;
+    const messages = history().map((message, index) => index < 4
+      ? new Proxy(message, {
+        get(target, property, receiver) {
+          historicalReads += 1;
+          return Reflect.get(target, property, receiver);
+        },
+      })
+      : message);
+    const projection = createTranscriptProjection();
+    const before = projection.project({ locale: 'en', sessionId: SESSION, messages });
+    assert.ok(historicalReads > 0, 'the initial projection must read the history');
+    historicalReads = 0;
+
+    const appended: StoredMessage[] = [...messages, {
+      type: 'assistant', id: 'a2-next', turnId: 'turn-2', ts: 7,
+      text: 'one more step', modelId: 'model-1',
+    }];
+    const after = projection.project({ locale: 'en', sessionId: SESSION, messages: appended });
+
+    assert.equal(historicalReads, 0, 'stable message references must skip historical materialization');
+    assert.strictEqual(after[0], before[0]);
+    assert.notStrictEqual(after[1], before[1]);
+    assert.equal(before[1]?.assistant?.text, 'done', 'previous projections are immutable snapshots');
+    assert.equal(after[1]?.assistant?.text, 'done\n\none more step');
+    assert.strictEqual(projection.project({ locale: 'en', sessionId: SESSION, messages: appended }), after);
+  });
+
+  test('a durable append keeps unchanged timeline items inside the affected turn', () => {
+    const projection = createTranscriptProjection();
+    const messages = history();
+    const before = projection.project({ locale: 'en', sessionId: SESSION, messages });
+    const appended: StoredMessage[] = [...messages, {
+      type: 'assistant', id: 'a1-next', turnId: 'turn-1', ts: 7,
+      text: 'still running', modelId: 'model-1',
+    }];
+    const after = projection.project({ locale: 'en', sessionId: SESSION, messages: appended });
+
+    assert.deepEqual(after, materializeTurns(appended, 'en'));
+    assert.notStrictEqual(after[0], before[0], 'the appended answer changes its turn');
+    assert.strictEqual(after[1], before[1], 'the unrelated turn keeps its identity');
+    const beforeTools = before[0]!.timeline.find((item) => item.kind === 'tools');
+    const afterTools = after[0]!.timeline.find((item) => item.kind === 'tools');
+    assert.ok(beforeTools);
+    assert.strictEqual(afterTools, beforeTools, 'the existing tool row keeps its identity');
+    assert.strictEqual(
+      after[0]!.timeline.find((item) => item.kind === 'text'),
+      before[0]!.timeline.find((item) => item.kind === 'text'),
+      'the earlier answer keeps its identity',
+    );
+    assert.deepEqual(after[0]!.timeline.at(-1), {
+      kind: 'text', text: 'still running', messageId: 'a1-next', ts: 7,
+    });
+  });
+
+  test('durable appends preserve cross-turn tool ownership and storage order', () => {
+    const projection = createTranscriptProjection();
+    let messages: StoredMessage[] = history();
+    let turns = projection.project({ locale: 'en', sessionId: SESSION, messages });
+    const unrelated = turns[1];
+    const steps: StoredMessage[][] = [
+      [toolCall('write-1', 'turn-3', 'WriteStdin', { ref: REF, input: 'go\n' }, 7)],
+      // The result's own Turn can differ from the call's owner.
+      [toolResult('write-1', 'turn-results', shellRun(4), 8)],
+      [{ type: 'assistant', id: 'a3', turnId: 'turn-3', ts: 9, text: 'sent input', modelId: 'model-1' }],
+      [toolCall('read-1', 'turn-4', 'Read', { path: REF }, 10), toolResult('read-1', 'turn-4', shellRun(5), 11)],
+      // Results are selected by the last storage position, not their timestamp.
+      [toolResult('write-1', 'turn-results', shellRun(6), 1)],
+      // A Turn can acquire another result for a different resource before its
+      // call arrives, and that call can appear in an older, interleaved Turn.
+      [toolResult('late-bash', 'turn-results', { ...shellRun(2), ref: `${REF}-other` }, 12)],
+      [toolCall('late-bash', 'turn-1', 'Bash', { command: 'other job', pty: true }, 13)],
+      [{ type: 'turn_state', id: 'ended', turnId: 'turn-3', ts: 14, status: 'aborted' }],
+    ];
+    for (const step of steps) {
+      const previousSnapshot = structuredClone(turns);
+      const previous = turns;
+      messages = [...messages, ...step];
+      turns = projection.project({ locale: 'en', sessionId: SESSION, messages });
+      assert.deepEqual(turns, materializeTurns(messages, 'en'));
+      assert.deepEqual(previous, previousSnapshot, 'appends must not mutate earlier projections');
+      assert.strictEqual(turns[1], unrelated, 'unrelated history keeps its identity');
+      assert.strictEqual(projection.project({ locale: 'en', sessionId: SESSION, messages }), turns);
+    }
+    assert.equal(revisionOf(turns[0]), 6);
+    assert.equal(turns.find((turn) => turn.turnId === 'turn-3')?.tools[0]?.result?.kind, 'shell_run');
+    assert.deepEqual(turns.find((turn) => turn.turnId === 'turn-4')?.tools, [], 'Read folds into its Bash');
+  });
+
+  test('appends after replacement, prepend, filtering and locale changes match full projection', () => {
+    const projection = createTranscriptProjection();
+    let messages = history();
+    let locale: 'en' | 'zh-CN' = 'en';
+    const project = () => {
+      const turns = projection.project({ sessionId: SESSION, locale, messages });
+      assert.deepEqual(turns, materializeTurns(messages, locale));
+      return turns;
+    };
+    const first = project();
+    messages = [...messages, { type: 'system_note', id: 'note', turnId: 'turn-1', ts: 7, kind: 'context_compacted' }];
+    project();
+    locale = 'zh-CN';
+    assert.equal(project()[0]?.notes[0]?.text, '已压缩较早的上下文。');
+
+    // Same IDs, timestamps, array endpoints and length, changed middle text.
+    messages = messages.map((message) => message.id === 'a1'
+      ? { ...message, text: 'replaced answer' } as StoredMessage
+      : message);
+    assert.equal(project()[0]?.assistant?.text, 'replaced answer');
+    messages = [...messages, { type: 'assistant', id: 'a3', turnId: 'turn-2', ts: 8, text: 'after edit', modelId: 'model-1' }];
+    project();
+
+    // Visible messages can temporarily omit an assistant while its live
+    // buffer drains, and put it back before rows already in the snapshot.
+    const unfiltered = messages;
+    messages = messages.filter((message) => message.id !== 'a1');
+    project();
+    messages = unfiltered;
+    project();
+    messages = [
+      { type: 'user', id: 'older', turnId: 'turn-0', ts: 0, text: 'earlier history' },
+      ...messages,
+    ];
+    project();
+    messages = messages.filter((message) => message.turnId !== 'turn-1');
+    project();
+    messages = [...messages, toolCall('read-new', 'turn-2', 'Read', { path: REF }, 9), toolResult('read-new', 'turn-2', shellRun(8), 10)];
+    assert.equal(project().at(-1)?.tools[0]?.toolName, 'Read', 'a removed Bash cannot hide a new Read');
+    assert.equal(first[0]?.assistant?.text, 'started');
+    assert.equal(first[0]?.notes.length, 0);
+  });
+
+  test('a ShellRun child appended before its Bash is folded when the owner arrives', () => {
+    const projection = createTranscriptProjection();
+    let messages: StoredMessage[] = [
+      toolCall('read-first', 'turn-child', 'Read', { path: REF }, 1),
+      toolResult('read-first', 'turn-child', shellRun(8), 2),
+      { type: 'assistant', id: 'child-answer', turnId: 'turn-child', ts: 3, text: 'read output', modelId: 'm' },
+    ];
+    const before = projection.project({ sessionId: SESSION, locale: 'en', messages });
+    messages = [...messages,
+      toolCall('bash-later', 'turn-owner', 'Bash', { command: 'job', pty: true }, 4),
+      toolResult('bash-later', 'turn-owner', shellRun(1), 5),
+    ];
+    const after = projection.project({ sessionId: SESSION, locale: 'en', messages });
+    assert.deepEqual(after, materializeTurns(messages, 'en'));
+    assert.deepEqual(after[0]?.tools, []);
+    assert.equal(revisionOf(after[1]), 8);
+    assert.equal(before[0]?.tools[0]?.toolName, 'Read');
+  });
+
+  test('snapshot reordering keeps surviving Turn identities at their new positions', () => {
+    const projection = createTranscriptProjection();
+    const messages = history();
+    const before = projection.project({ sessionId: SESSION, locale: 'en', messages });
+    const moved: StoredMessage[] = [
+      { type: 'user', id: 'new-first', turnId: 'turn-first', ts: 0, text: 'prepended' },
+      ...messages.slice(4),
+      ...messages.slice(0, 4),
+    ];
+    const after = projection.project({ sessionId: SESSION, locale: 'en', messages: moved });
+    assert.deepEqual(after.map((turn) => turn.turnId), ['turn-first', 'turn-2', 'turn-1']);
+    assert.strictEqual(after[1], before[1]);
+    assert.strictEqual(after[2], before[0]);
+  });
+
   test('a locale change rematerializes localized system notes', () => {
     const projection = createTranscriptProjection();
     const messages: StoredMessage[] = [{

packages/ui/src/incremental-turn-materializer.ts

@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { StoredMessage } from '@maka/core/session';
+import type { UiLocale } from '@maka/core/ui-locale';
+import { materializeTurns, type TurnViewModel } from './materialize.js';
+
+/**
+ * Index immutable message snapshots before deriving view data. An append only
+ * rebuilds the changed turns and their tool dependencies; replacement, prepend
+ * and locale changes use the complete materializer. The caller reconciles the
+ * returned candidates by value and supplies those settled turns on the next call.
+ *
+ * Tool results can belong to another turn than their call, and ShellRun child
+ * results update the Bash in an earlier turn. Materialize their whole connected
+ * group in storage order so the existing session-wide folding remains the sole
+ * authority, including children preceding their parent and repeated results.
+ */
+export function createIncrementalTurnMaterializer() {
+  let lastMessages: readonly StoredMessage[] | undefined;
+  let lastLocale: UiLocale | undefined;
+  const indicesByTurn = new Map<string, number[]>();
+  const turnsByToolUseId = new Map<string, Set<string>>();
+  const turnsByShellRun = new Map<string, Set<string>>();
+  const dependenciesByTurn = new Map<string, Set<Set<string>>>();
+
+  function link(index: Map<string, Set<string>>, key: string, turnId: string): void {
+    let peers = index.get(key);
+    if (!peers) {
+      peers = new Set();
+      index.set(key, peers);
+    }
+    peers.add(turnId);
+    let dependencies = dependenciesByTurn.get(turnId);
+    if (!dependencies) {
+      dependencies = new Set();
+      dependenciesByTurn.set(turnId, dependencies);
+    }
+    dependencies.add(peers);
+  }
+
+  function materialize(
+    messages: readonly StoredMessage[],
+    locale: UiLocale,
+    previous: readonly TurnViewModel[],
+  ): readonly TurnViewModel[] {
+    const appended = lastMessages !== undefined
+      && lastLocale === locale
+      && extendsSnapshot(lastMessages, messages);
+    const from = appended ? lastMessages!.length : 0;
+    // A failed materialization must not leave a partially advanced index for
+    // the next call to append to. A retry will rebuild from its input snapshot.
+    lastMessages = undefined;
+    if (!appended) {
+      indicesByTurn.clear();
+      turnsByToolUseId.clear();
+      turnsByShellRun.clear();
+      dependenciesByTurn.clear();
+    }
+
+    const affected = new Set<string>();
+    const added = new Set<string>();
+    const length = messages.length;
+    for (let index = from; index < length; index += 1) {
+      const message = messages[index]!;
+      const turnId = message.turnId ?? '__loose';
+      let indices = indicesByTurn.get(turnId);
+      if (!indices) {
+        indices = [];
+        indicesByTurn.set(turnId, indices);
+        added.add(turnId);
+      }
+      indices.push(index);
+      affected.add(turnId);
+      if (message.type === 'tool_call') {
+        link(turnsByToolUseId, message.id, turnId);
+      } else if (message.type === 'tool_result') {
+        link(turnsByToolUseId, message.toolUseId, turnId);
+        if (message.content.kind === 'shell_run') {
+          link(turnsByShellRun, message.content.ref, turnId);
+        }
+      }
+    }
+
+    let result: readonly TurnViewModel[];
+    if (!appended) {
+      result = materializeTurns(messages, locale);
+    } else if (affected.size === 0) {
+      result = previous;
+    } else {
+      // Visit each shared dependency once, even when many turns refer to the
+      // same background command. Set iteration also visits newly added turns.
+      const visited = new Set<Set<string>>();
+      for (const turnId of affected) {
+        for (const peers of dependenciesByTurn.get(turnId) ?? []) {
+          if (visited.has(peers)) continue;
+          visited.add(peers);
+          for (const peer of peers) affected.add(peer);
+        }
+      }
+      const indices: number[] = [];
+      for (const turnId of affected) {
+        for (const index of indicesByTurn.get(turnId)!) indices.push(index);
+      }
+      indices.sort((left, right) => left - right);
+      const changed = new Map(materializeTurns(
+        indices.map((index) => messages[index]!),
+        locale,
+      ).map((turn) => [turn.turnId, turn]));
+      const next = previous.map((turn) => changed.get(turn.turnId) ?? turn);
+      for (const turnId of added) next.push(changed.get(turnId)!);
+      result = next;
+    }
+    lastMessages = messages;
+    lastLocale = locale;
+    return result;
+  }
+
+  return { materialize };
+}
+
+function extendsSnapshot(
+  previous: readonly StoredMessage[],
+  next: readonly StoredMessage[],
+): boolean {
+  const length = previous.length;
+  if (next.length < length) return false;
+  for (let index = 0; index < length; index += 1) {
+    if (previous[index] !== next[index]) return false;
+  }
+  return true;
+}