diff --git a/packages/ui/src/__tests__/transcript-projection.test.ts b/packages/ui/src/__tests__/transcript-projection.test.ts index ebf92fe0ad..e84559d7ae 100644 --- a/packages/ui/src/__tests__/transcript-projection.test.ts +++ b/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[] = [{ diff --git a/packages/ui/src/incremental-turn-materializer.ts b/packages/ui/src/incremental-turn-materializer.ts new file mode 100644 index 0000000000..c8a49d77ee --- /dev/null +++ b/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(); + const turnsByToolUseId = new Map>(); + const turnsByShellRun = new Map>(); + const dependenciesByTurn = new Map>>(); + + function link(index: Map>, 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(); + const added = new Set(); + 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>(); + 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; +} diff --git a/packages/ui/src/transcript-projection.ts b/packages/ui/src/transcript-projection.ts index 267c6972a8..6f9dcb214a 100644 --- a/packages/ui/src/transcript-projection.ts +++ b/packages/ui/src/transcript-projection.ts @@ -21,10 +21,10 @@ import type { ShellRunUpdate } from '@maka/core/events'; import type { StoredMessage } from '@maka/core/session'; import type { UiLocale } from '@maka/core/ui-locale'; import type { LiveTurnProjection } from './live-turn-projection.js'; +import { createIncrementalTurnMaterializer } from './incremental-turn-materializer.js'; import { applyShellRunOverlayEntry, foldShellRunUpdates, - materializeTurns, overlayLiveTurn, projectTurnTools, timelineItemKey, @@ -45,9 +45,9 @@ import { * any session with background-command history. * * This layer owns the derived state instead of re-deriving it: it remembers the - * settled turns and hands the previous object back for any turn a message - * refresh did not actually change, so "what changed" is decided by value here - * rather than guessed downstream from reference equality. + * settled turns, indexes their immutable messages and only rematerializes the + * affected turns on append. Value reconciliation preserves identity even when + * a changed input or a replacement snapshot projects to the same view. * * Contract (mirrors `parseMarkdownIncremental`'s in `@astryxdesign/core`): * a turn keeps its object identity unless its projected value changed, and @@ -78,6 +78,7 @@ const NO_TURNS: readonly TurnViewModel[] = []; export function createTranscriptProjection(): TranscriptProjection { let sessionId: string | undefined; let hasProjected = false; + let materializer = createIncrementalTurnMaterializer(); // Stage inputs, remembered so a stage only reruns when its own input moved. let lastMessages: readonly StoredMessage[] | undefined; @@ -100,6 +101,7 @@ export function createTranscriptProjection(): TranscriptProjection { function reset(): void { hasProjected = false; + materializer = createIncrementalTurnMaterializer(); lastMessages = undefined; lastLocale = undefined; lastLiveTurn = undefined; @@ -143,7 +145,7 @@ export function createTranscriptProjection(): TranscriptProjection { if (input.messages !== lastMessages || input.locale !== lastLocale) { settledTurns = reconcileTurnIdentities( settledTurns, - materializeTurns(input.messages, input.locale), + materializer.materialize(input.messages, input.locale, settledTurns), ); lastMessages = input.messages; lastLocale = input.locale; @@ -206,27 +208,41 @@ export function createTranscriptProjection(): TranscriptProjection { /** * Keep the previous object for every turn whose projected value is unchanged. - * A message refresh rebuilds the whole snapshot from freshly deserialized IPC - * rows, so nothing upstream can carry identity — the equality check here is - * what narrows a refresh to the turns whose messages actually changed. + * Appends already preserve unrelated candidates by reference. Replacement + * snapshots and touched turns still need value comparison: changed messages + * need not change the view, and a reconnect can replace every message object. */ export function reconcileTurnIdentities( previous: readonly TurnViewModel[], next: readonly TurnViewModel[], ): readonly TurnViewModel[] { + if (previous === next) return previous; if (previous.length === 0) return next; - const previousById = new Map(previous.map((turn) => [turn.turnId, turn])); - const reconciled = next.map((turn) => { - const prior = previousById.get(turn.turnId); - if (!prior || valuesEqual(prior, turn)) return prior ?? turn; - // The turn moved, but usually only its tail did: hand the previous - // timeline entry back for every item whose value did not change, so the - // entry-level memo boundaries downstream see what actually moved. - return { ...turn, timeline: reconcileTimelineItems(prior.timeline, turn.timeline) }; + // Appends keep positions, so the usual pass needs no transcript-wide index. + // Build it lazily when deletion, insertion or reordering actually moves IDs. + let previousById: Map | undefined; + let changed = next.length !== previous.length; + const reconciled = next.map((turn, index) => { + const atIndex = previous[index]; + if (turn === atIndex) return turn; + let prior: TurnViewModel | undefined = atIndex; + if (prior?.turnId !== turn.turnId) { + if (index < previous.length) { + previousById ??= new Map(previous.map((item) => [item.turnId, item])); + } + prior = previousById?.get(turn.turnId); + } + let value = prior && valuesEqual(prior, turn) ? prior : turn; + if (prior && value !== prior) { + // The turn moved, but usually only its tail did: hand the previous + // timeline entry back for every item whose value did not change, so the + // entry-level memo boundaries downstream see what actually moved. + value = { ...turn, timeline: reconcileTimelineItems(prior.timeline, turn.timeline) }; + } + if (value !== atIndex) changed = true; + return value; }); - return reconciled.length === previous.length && reconciled.every((turn, index) => turn === previous[index]) - ? previous - : reconciled; + return changed ? reconciled : previous; } /**