diff --git a/.agents/notes/implemented/bug-fix/2026-09-16-streaming-cost-scales-with-conversation-length.md b/.agents/notes/implemented/bug-fix/2026-09-16-streaming-cost-scales-with-conversation-length.md new file mode 100644 index 000000000..3e06c330c --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-09-16-streaming-cost-scales-with-conversation-length.md @@ -0,0 +1,153 @@ +# Streaming cost scaled with conversation length, not with what changed + +Status: implemented +Translation: current + +[中文](2026-09-16-streaming-cost-scales-with-conversation-length.zh.md) + +## Abstract + +After windowed reads shipped, users reported the renderer becoming unresponsive on +long conversations. The window was not the problem: a content notification reported +the span between the lowest and highest position it touched and named every turn id +inside it, so one synced batch carrying an early status write alongside the streaming +tail told the display cache that the whole conversation had changed — measured at 996 +directory rows and 997 turn bodies re-read, about 1.2 s of main-thread CPU, for a +single token delta on a 1,000-turn conversation. Notifications now carry the exact +touched positions, the cache invalidates only the ids the reader named, and four +whole-conversation costs that ran per frame or per open were made incremental or +deferred. The same delta now re-reads two rows and two bodies. Total work for the +shared fact table is unchanged — it still materializes every turn once per session — +so a first open on a very long conversation is improved but not solved; that needs +facts to move into write-time metadata. + +## The reported symptom and what it was not + +The report was "the renderer freezes easily on the latest version, 0.93.3 was fine". +0.93.3 and 0.94.0 differ by the windowed conversation read. Two plausible causes were +ruled out first: Electron is unchanged across the two releases (`^39.2.6`), and a +pure tail delta — a token arriving with nothing else in the batch — already cost 7-9 ms +and still does. The regression only appears when a batch carries more than the tail. + +## Root cause: a sparse change reported as a dense range + +`changeRangeOf` in the Loro session adapter reduced a batch to `[min(position), +max(position) + 1)` and `observe` then read every id in that range. A batch that +touched positions 4 and 999 named 996 turns. The display cache merged its refresh +targets the same way, so two independent notifications arriving before one flush +produced the same span. + +The cache then re-read the whole directory for that span and, for every hydrated row +in it, re-materialized the body. That second amplification had its own cause: +`rowChanged` compared `itemCount` and `planCount`, which a container-backed directory +row deliberately never carries, against the real counts a hydrated row holds. The +comparison was `42 !== undefined` on every refresh, so every hydrated row reported a +change, bumped its content epoch and was re-read. Of the 140 bodies re-read in the +smaller reproduction, 2 had changed. + +Bumping those epochs also cancelled in-flight reads for turns nothing had touched. +A 300-turn window lease taken while such batches arrived every 5 ms took 1,890 ms and +performed 3.5 body reads per turn; the 5 ms interval fired 4 times in that window, +which is the unresponsiveness the report describes. + +## What changed + +- A content batch reports its exact positions. Structural batches keep the + shifted-suffix range they need, because later positions really did move. +- The view tracks content targets as the reported ids, resolves their positions at + flush time, and refreshes them in contiguous runs. A content refresh that finds the + length changed escalates to a structural re-key rather than splicing rows from a + sparse read. +- Body invalidation follows the reported ids, never the directory diff: a directory + row cannot tell whether a body changed, since a grown text item moves no scalar. +- `rowChanged` compares counts only after the refresh has carried forward the ones it + cannot supply, and an unchanged row keeps its object — placeholder items and Virtua + rows key on that identity, and replacing it defeated their caches. + +Four costs that scale with turn count were addressed alongside it: + +- A user turn's send configuration projected eagerly on every directory row. The + projection is a schema parse per turn while `resolveSessionConversationConfig` reads + the newest source and walks older ones only until it finds an explicit Role, so it + is now deferred and memoized at each hop that used to force it. Full directory read: + 4,000 turns 728 ms to 264 ms. +- The shared fact table discarded its facts on the last consumer release, so reopening + a session tab re-materialized every body to rebuild them. It is now held rather than + disposed, keeps its view subscription, and resumes. Its background pass yields to + real idle time instead of back-to-back macrotasks. +- `use-session-diff-summary` serialized every turn's file diffs each frame to decide + whether anything changed (144 KiB per frame at 4,000 turns). Identical entries are + settled by reference; only a replaced entry is serialized. +- `normalizeTexMathDelimiters` and the Mermaid fence test re-scanned the whole + accumulated answer on every delta, which is quadratic in answer length. Both now + settle the common case with a substring test: a 120 KiB math-free answer went from + 2.49 ms to 0.06 ms per delta. + +The index-row list, the chat stream items, the conversation config sources and the +ordered fact list hand back their previous array when nothing changed, and +`ConversationView` gained `structureVersion` so the accepted-history projection stops +rebuilding a whole-conversation slot array at token rate. + +## Alternatives considered + +Reporting the dense range but letting the cache filter it was rejected: the cache +would still read the whole directory to discover that nothing else moved, which is the +larger of the two costs at 4,000 turns. + +Making the directory row carry `itemCount`/`planCount` would also have removed the +`rowChanged` false positive, at one extra container crossing per turn on every +directory read. Carrying the previous row's counts forward costs nothing and keeps the +adapter's existing "omit rather than guess" rule. + +Keeping the fact table alive after the last release trades memory for the reopen cost. +The table is keyed by the view, so the session store's existing eviction bounds it; a +per-consumer timeout was rejected as a second lifetime to reason about. + +## Two corrections found in review + +Both were introduced by this work and are fixed in the same branch, each pinned by +a test that fails on the intermediate implementation. + +Holding the fact table instead of disposing it stopped the background pass but left +the view subscription in place. That is correct when the table is keyed on the +conversation's own view, and wrong when it is keyed on a projection wrapper: the +wrapper is rebuilt whenever an optimistic entry appears or resolves, so the base +view's listener set accumulated one released wrapper — and one live fact table — per +message sent, each still deriving on every token. Tables are now keyed and +subscribed on `factSource`, the underlying view. That also collapses a duplicate +that predates this branch: the diff summary acquired on the base view while the +turn-fact readers acquired on the wrapper, so an unconfirmed entry meant two full +fact tables for one conversation. + +Keeping the index row object when `rowChanged` reported no change assumed that +`rowChanged` sees everything a row carries. It does not see `inputConfig`, which is +a deferred projection that cannot be diffed without forcing it — exactly the cost +this branch removed. A user turn whose send configuration changed while outside +every hydrated window therefore kept its old model, Role and MCP selection in the +index, which is what the sticky-configuration resolver reads. A reported turn now +always takes the fresh row; identity is preserved only for turns the notification +did not name, which is where the churn this optimization targets came from. + +## Verification and limits + +Regression coverage pins the invalidation contract: a batch carrying an early edit and +the streaming tail reads 2 directory rows and 2 bodies, and fails at 23 rows on the +previous implementation. The fact-table test pins that facts survive the last release +and that the background pass is held until someone re-acquires. + +Numbers above come from synthetic Loro fixtures in Node on one machine and are library +measurements, not device acceptance. They are useful as ratios, not as budgets. Real +turns carry far more content than the fixture's. + +Two known costs are unchanged. The shared fact table still materializes every turn once +per session, because deriving a goal, scheduled task, proposed plan or file diff needs +the body; removing that needs those facts written alongside history, which +[versioned history hashes and primitive metadata](../architecture/2026-09-14-versioned-history-hashes-and-primitive-metadata.md) +opens the way for. Snapshot import still decodes on the renderer thread, as +[windowed conversation reads](../architecture/2026-09-10-windowed-reader-integration.md) +already recorded; moving it would mean moving the document off the main thread. + +That note's evidence boundary listed streaming frame time as separate acceptance work +that was never done. This note is the correction: the work was necessary, and the cost +it would have found was a correctness-shaped defect in change scoping rather than a +tuning problem. diff --git a/.agents/notes/implemented/bug-fix/2026-09-16-streaming-cost-scales-with-conversation-length.zh.md b/.agents/notes/implemented/bug-fix/2026-09-16-streaming-cost-scales-with-conversation-length.zh.md new file mode 100644 index 000000000..0bc1cb3cd --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-09-16-streaming-cost-scales-with-conversation-length.zh.md @@ -0,0 +1,118 @@ +# 流式开销随对话长度增长,而非随实际变更量增长 + +Status: implemented +Translation: current + +[English](2026-09-16-streaming-cost-scales-with-conversation-length.md) + +## 摘要 + +窗口化读取上线后,用户反馈长对话会让渲染进程失去响应。问题不在窗口化:内容变更通知 +上报的是它触及的最低与最高位置之间的整个区间,并把区间内每一个 turn id 都列为已变更, +因此一个同步批次只要同时带上一条早期 turn 的状态写入和正在流式输出的尾部,就等于告诉 +显示缓存「整个对话都变了」——实测在 1000 turn 的对话上,单个 token delta 触发 996 行 +目录重读、997 个 turn body 重新物化,约 1.2 秒主线程 CPU。现在通知只携带真正触及的位置, +缓存只让 reader 点名的 id 失效,另有四处按帧或按打开执行的全对话开销改为增量或惰性。 +同一个 delta 现在只重读 2 行和 2 个 body。共享 fact table 的总工作量未变——每个会话仍会 +把每个 turn 物化一次——所以超长对话的首次打开有所改善但未解决,那需要把 facts 落到写入时 +的元数据里。 + +## 报告的现象,以及排除了什么 + +反馈是「最新版很容易卡死,0.93.3 没遇到过」。0.93.3 与 0.94.0 之间的差异是窗口化对话读取。 +先排除了两种可能:两个版本的 Electron 完全一致(`^39.2.6`);纯尾部 delta——批次里只有一个 +token 到达——此前就是 7–9 ms,现在仍是。回归只在批次里不止有尾部时出现。 + +## 根因:稀疏变更被上报为稠密区间 + +Loro session 适配器的 `changeRangeOf` 把一个批次归约成 `[min(position), max(position)+1)`, +`observe` 随后读出该区间内的每一个 id。一个只触及位置 4 和 999 的批次会点名 996 个 turn。 +显示缓存用同样的方式合并刷新目标,所以两条独立通知只要在同一次 flush 之前到达,也会产生 +同样的区间。 + +缓存随后重读整个区间的目录,并把区间内每一个已 hydrate 的行的 body 重新物化。第二重放大 +另有原因:`rowChanged` 会比较 `itemCount` 与 `planCount`,而容器形态的目录行刻意从不携带 +这两个字段,已 hydrate 的行却持有真实值。于是每次刷新的比较都是 `42 !== undefined`,每个 +已 hydrate 的行都被判定为已变更、content epoch 被 bump、body 被重读。在较小的复现里重读的 +140 个 body 中,真正变了的是 2 个。 + +bump 这些 epoch 还会作废那些与变更无关的 turn 的在途读取。在这类批次每 5 ms 到达一次的情况下 +获取一个 300 turn 的窗口 lease 耗时 1890 ms、每个 turn 平均读取 3.5 次 body;那个 5 ms 的 +定时器在此期间只触发了 4 次——这正是反馈里描述的无响应。 + +## 改动内容 + +- 内容批次上报确切位置。结构性批次保留它需要的「位移后缀」区间,因为后续位置确实移动了。 +- 视图以上报的 id 记录内容刷新目标,在 flush 时解析位置,并按连续段刷新。内容刷新若发现 + 长度变化,会升级为结构性重建索引,而不是从稀疏读取里拼接行。 +- body 失效只跟随上报的 id,不跟随目录 diff:目录行无法判断 body 是否变化,因为一条变长的 + 文本条目不会移动任何标量。 +- `rowChanged` 只在刷新已把它无法提供的计数继承过来之后才比较它们;未变更的行保留原对象—— + placeholder 条目与 Virtua 行以该身份为键,替换它会让这些缓存失效。 + +同时处理了四处随 turn 数增长的开销: + +- user turn 的发送配置在每个目录行上被立即投影。该投影是每个 turn 一次 schema 解析,而 + `resolveSessionConversationConfig` 只读最新的来源、并且只在找到显式 Role 之前向前少量回溯, + 因此改为在每个原本会强制求值的环节上惰性求值并记忆化。完整目录读取:4000 turn 从 728 ms + 降到 264 ms。 +- 共享 fact table 在最后一个消费者释放时丢弃 facts,导致重新打开会话标签时要把每个 body 重新 + 物化。现在改为保持而非销毁,保留对视图的订阅并可续跑。其后台 pass 让出到真正的空闲时间, + 而不是连续的宏任务。 +- `use-session-diff-summary` 每帧序列化每个 turn 的文件改动来判断是否有变化(4000 turn 时每帧 + 144 KiB)。现在相同条目按引用判定,只有被替换的条目才会被序列化。 +- `normalizeTexMathDelimiters` 与 Mermaid 围栏检测在每个 delta 上重扫整段累计回答,对回答长度 + 是平方级。两者现在都用子串检测先行了结常见情况:120 KiB 且不含公式的回答从每个 delta + 2.49 ms 降到 0.06 ms。 + +索引行列表、聊天流条目、对话配置来源与有序 fact 列表在无变化时都返回上一次的数组; +`ConversationView` 新增 `structureVersion`,使已接受历史的投影不再以 token 频率重建整对话的 +槽位数组。 + +## 考虑过的替代方案 + +「仍上报稠密区间、由缓存自行过滤」被否决:缓存仍需读完整个目录才能发现其余部分没有变化, +而在 4000 turn 时这正是两项开销中较大的一项。 + +让目录行携带 `itemCount`/`planCount` 同样能消除 `rowChanged` 的假阳性,代价是每次目录读取里 +每个 turn 多一次容器跨越。继承上一行的计数没有额外成本,也保留了适配器现有的「宁可省略, +不做猜测」规则。 + +在最后一次释放后保留 fact table 是用内存换重新打开的开销。该表以视图为键,因此受会话 store +现有的淘汰机制约束;按消费者设置超时被否决,因为那会引入第二套需要推理的生命周期。 + +## 评审中发现的两处修正 + +两处都是本轮工作引入的,已在同一分支修复,各自由一个在中间实现上失败的用例锁定。 + +把 fact table 改为保持而非销毁时,只停掉了后台扫描,却保留了对视图的订阅。当表以对话 +自身的视图为键时这是对的;以投影包装器为键时则是错的:包装器会在乐观条目出现与确认时 +被重建,于是每发送一条消息,底层视图的监听集合就多积累一个已释放的包装器和一张仍在 +工作的 fact table,每个 token 都会各推导一次。现在表以 `factSource`(底层视图)为键并 +在其上订阅。这同时消除了一个早于本分支的重复:diff summary 在底层视图上获取,而 turn +fact 读取方在包装器上获取,因此只要存在未确认条目,一个对话就会有两张完整的 fact table。 + +在 `rowChanged` 判定无变化时保留索引行对象,前提是 `rowChanged` 能看到行携带的一切。 +它看不到 `inputConfig`——那是一个惰性投影,不强制求值就无法比较,而强制求值正是本分支 +要消除的开销。于是一个处于所有已 hydrate 窗口之外的 user turn,其发送配置变更后,索引 +里仍保留旧的模型、Role 与 MCP 选择,而这正是粘性配置解析器读取的内容。现在被点名的 +turn 一律采用新行;只有通知未点名的 turn 才保留身份,而那正是该优化针对的抖动来源。 + +## 验证与局限 + +回归用例锁定了失效契约:同时携带早期编辑与流式尾部的批次只读 2 行目录、2 个 body,在改动前的 +实现上会在 23 行处失败。fact table 的用例锁定了 facts 在最后一次释放后存活,且后台 pass 在 +有人重新获取之前保持暂停。 + +以上数字来自单机 Node 环境下的合成 Loro fixture,属于库级测量,不是设备验收。它们作为比例有 +参考价值,不能当作预算。真实 turn 携带的内容远多于 fixture。 + +有两项已知开销未改变。共享 fact table 仍然每个会话把每个 turn 物化一次,因为推导 goal、定时 +任务、提案计划或文件改动都需要 body;要去掉它,需要把这些 facts 与历史一起写入,而 +[版本化历史哈希与原始元数据](../architecture/2026-09-14-versioned-history-hashes-and-primitive-metadata.md) +为此铺好了路。快照 import 仍在渲染线程解码,这一点 +[窗口化对话读取](../architecture/2026-09-10-windowed-reader-integration.md) +已有记录;要移走它意味着把文档移出主线程。 + +那篇笔记的证据边界把流式帧时间列为「另行验收的工作」,而这项工作从未进行。本笔记即是对它的 +更正:这项工作是必要的,且它本应发现的开销是变更范围界定上的正确性缺陷,而不是调优问题。 diff --git a/packages/components/src/components/ai-gui/markdown-renderer.tsx b/packages/components/src/components/ai-gui/markdown-renderer.tsx index 69ee1e9a6..3823b5552 100644 --- a/packages/components/src/components/ai-gui/markdown-renderer.tsx +++ b/packages/components/src/components/ai-gui/markdown-renderer.tsx @@ -1160,6 +1160,8 @@ export const MarkdownRenderer = memo(function MarkdownRenderer({ const resolvedTheme = useResolvedTheme(); const readonly = useContext(SessionReadonlyContext); const containerRef = useRef(null); + /** Whether this block currently holds search marks that need unwrapping. */ + const markedRef = useRef(false); const search = useSessionSearch(); const searchMatch = useSessionSearchBlock(searchBlockId ?? ''); const copyCodeLabel = t('common.copyCode', 'Copy code'); @@ -1167,7 +1169,13 @@ export const MarkdownRenderer = memo(function MarkdownRenderer({ const openAgentFileLabel = t('sessions.openAgentFile', 'Open agent file'); const canvasLabel = t('sessions.diagram.canvas', 'Zoom and pan diagram'); const openDiagramLabel = t('sessions.diagramViewer.open', 'Open diagram'); - const hasMermaidBlock = useMemo(() => MERMAID_FENCE_PATTERN.test(text), [text]); + // Both scans below re-run over the whole accumulated answer on every streamed + // delta. A substring test settles the common case before the line-anchored + // pattern runs. + const hasMermaidBlock = useMemo( + () => text.includes('mermaid') && MERMAID_FENCE_PATTERN.test(text), + [text] + ); const normalizedText = useMemo(() => normalizeTexMathDelimiters(text), [text]); const { blocks: mermaidBlocks, @@ -1223,6 +1231,11 @@ export const MarkdownRenderer = memo(function MarkdownRenderer({ .forEach((button) => button.setAttribute('aria-label', copyCodeLabel)); const clearSearchHighlights = () => { + // Nothing was ever marked in this block, so there is nothing to unwrap. + // This effect re-runs on every streamed delta, and the query below walks + // the rendered subtree. + if (!markedRef.current) return; + markedRef.current = false; const existingMarks = root.querySelectorAll('mark[data-session-search-mark="true"]'); existingMarks.forEach((mark) => { const parent = mark.parentNode; @@ -1336,6 +1349,7 @@ export const MarkdownRenderer = memo(function MarkdownRenderer({ if (!parent) { return; } + markedRef.current = true; parent.insertBefore(fragment, node); parent.removeChild(node); }); diff --git a/packages/components/src/components/sessions/session-detail.tsx b/packages/components/src/components/sessions/session-detail.tsx index b05424a2a..ca9176dac 100644 --- a/packages/components/src/components/sessions/session-detail.tsx +++ b/packages/components/src/components/sessions/session-detail.tsx @@ -470,10 +470,30 @@ const PR_SIDEBAR_MIN_WIDTH_PX = 500; const selectSessionDetailMeta = (meta: SessionMeta | undefined): SessionMeta | undefined => meta; +/** + * Serialized meta, memoized on the object. + * + * This equality gate runs on every session-meta emission, and the retained + * previous value was re-serialized each time even though only the incoming + * one is new. + */ +const sessionDetailMetaFingerprints = new WeakMap(); +const sessionDetailMetaFingerprint = (meta: SessionMeta): string => { + const cached = sessionDetailMetaFingerprints.get(meta); + if (cached !== undefined) return cached; + const computed = JSON.stringify(meta); + sessionDetailMetaFingerprints.set(meta, computed); + return computed; +}; + const sessionDetailMetaEqual = ( left: SessionMeta | undefined, right: SessionMeta | undefined -): boolean => left === right || JSON.stringify(left) === JSON.stringify(right); +): boolean => + left === right || + (left !== undefined && + right !== undefined && + sessionDetailMetaFingerprint(left) === sessionDetailMetaFingerprint(right)); function PendingWorktreeForkObserver({ targetSessionId, @@ -3232,12 +3252,10 @@ const SessionDetail = ({ const resolution = await resolveSessionFileProviderOpenPath( activeSessionFileProvider, target.filePath - ).catch( - (): SessionFileProviderOpenPathResolution => ({ - path: target.filePath, - redirected: false, - }) - ); + ).catch((): SessionFileProviderOpenPathResolution => ({ + path: target.filePath, + redirected: false, + })); const resolvedFilePath = resolution.path; const requestSeq = nextFocusRequestSeq(); @@ -3314,12 +3332,10 @@ const SessionDetail = ({ const resolution = await resolveSessionFileProviderOpenPath( activeSessionFileProvider, tab.filePath - ).catch( - (): SessionFileProviderOpenPathResolution => ({ - path: tab.filePath, - redirected: false, - }) - ); + ).catch((): SessionFileProviderOpenPathResolution => ({ + path: tab.filePath, + redirected: false, + })); if (!resolution.redirected || resolution.path === tab.filePath) { return { tab, next: tab }; } @@ -3630,22 +3646,18 @@ const SessionDetail = ({ }); return [ ...fixedTabs, - ...visibleSideSessions.map( - (sideSession): SessionSidePanelTabItem => ({ - id: getSideSessionPanelTabId(sideSession.id), - label: sideSession.title?.trim() || t('sessions.detailTabs.sideSession', 'Side Chat'), - kind: 'session', - closeable: true, - pending: closingSideSessionIds.has(sideSession.id), - }) - ), - ...viewerTabItems.map( - (tab): SessionSidePanelTabItem => ({ - ...tab, - kind: tab.type, - closeable: true, - }) - ), + ...visibleSideSessions.map((sideSession): SessionSidePanelTabItem => ({ + id: getSideSessionPanelTabId(sideSession.id), + label: sideSession.title?.trim() || t('sessions.detailTabs.sideSession', 'Side Chat'), + kind: 'session', + closeable: true, + pending: closingSideSessionIds.has(sideSession.id), + })), + ...viewerTabItems.map((tab): SessionSidePanelTabItem => ({ + ...tab, + kind: tab.type, + closeable: true, + })), ]; }, [ closingSideSessionIds, diff --git a/packages/components/src/components/sessions/session-turn-facts.ts b/packages/components/src/components/sessions/session-turn-facts.ts index c0ccda44d..e13413e34 100644 --- a/packages/components/src/components/sessions/session-turn-facts.ts +++ b/packages/components/src/components/sessions/session-turn-facts.ts @@ -1,4 +1,4 @@ -import { useMemo } from 'react'; +import { useMemo, useRef } from 'react'; import { isAskUserQuestionPermissionMeta, resolveLatestSessionGoalFromHistory, @@ -145,14 +145,27 @@ export function useSessionTurnFacts( ): SessionTurnFactsResult { const rows = useConversationIndexRows(view); const { facts, complete, version } = useConversationDerivation(view, deriveSessionTurnFacts); + const previousOrderedRef = useRef(EMPTY_ORDERED); const ordered = useMemo(() => { - if (facts.size === 0) return EMPTY_ORDERED; + if (facts.size === 0) { + previousOrderedRef.current = EMPTY_ORDERED; + return EMPTY_ORDERED; + } const list: SessionTurnFacts[] = []; for (const row of rows as readonly TurnIndexRow[]) { const fact = facts.get(row.id); if (fact) list.push(fact); } - return list; + // A fact is replaced only when its turn changed, so a table version that + // moved for an unrelated reason — a chunk of the background pass landing, + // a turn re-derived to the same value — must not hand every reader below + // a new array to scan. + const previous = previousOrderedRef.current; + const reusable = + previous.length === list.length && previous.every((fact, index) => fact === list[index]); + const result = reusable ? previous : list; + previousOrderedRef.current = result; + return result; // `version` is the change signal for the fact table. // eslint-disable-next-line react-hooks/exhaustive-deps }, [facts, rows, version]); diff --git a/packages/components/src/components/sessions/use-session-diff-summary.ts b/packages/components/src/components/sessions/use-session-diff-summary.ts index dcc43516e..22ef6d253 100644 --- a/packages/components/src/components/sessions/use-session-diff-summary.ts +++ b/packages/components/src/components/sessions/use-session-diff-summary.ts @@ -102,28 +102,75 @@ function collectDiffInputs( return entries; } +/** The diff-relevant identity of one turn. */ +const diffInputEntryShape = (entry: SessionDiffInputEntry): unknown => [ + entry?.id ?? '', + getSessionHistoryEntryRole(entry) ?? '', + normalizeHistoryEntryFileDiffs(entry).map((fileDiff) => [ + fileDiff.filePath, + fileDiff.add, + fileDiff.del, + fileDiff.cc === undefined + ? null + : [ + fileDiff.cc.v, + fileDiff.cc.fileId, + fileDiff.cc.baseOpId ?? '', + fileDiff.cc.opId ?? '', + fileDiff.cc.base ?? '', + fileDiff.cc.deleted === true, + ], + ]), +]; + export function computeSessionDiffInputsFingerprint(history: SessionHistoryInput): string { - return JSON.stringify( - (history ?? []).map((entry) => [ - entry?.id ?? '', - getSessionHistoryEntryRole(entry) ?? '', - normalizeHistoryEntryFileDiffs(entry).map((fileDiff) => [ - fileDiff.filePath, - fileDiff.add, - fileDiff.del, - fileDiff.cc === undefined - ? null - : [ - fileDiff.cc.v, - fileDiff.cc.fileId, - fileDiff.cc.baseOpId ?? '', - fileDiff.cc.opId ?? '', - fileDiff.cc.base ?? '', - fileDiff.cc.deleted === true, - ], - ]), - ]) - ); + return JSON.stringify((history ?? []).map((entry) => diffInputEntryShape(entry))); +} + +type SessionDiffInputEntry = NonNullable[number]; + +/** + * Per-entry serialization, memoized on the entry object. + * + * A fact is replaced only when its turn changed, so an unchanged entry keeps + * its identity across passes and is never re-serialized. + */ +const diffInputEntryFingerprints = new WeakMap(); +const diffInputEntryFingerprint = (entry: SessionDiffInputEntry): string => { + if (!entry || typeof entry !== 'object') return JSON.stringify(diffInputEntryShape(entry)); + const cached = diffInputEntryFingerprints.get(entry); + if (cached !== undefined) return cached; + const computed = JSON.stringify(diffInputEntryShape(entry)); + diffInputEntryFingerprints.set(entry, computed); + return computed; +}; + +/** + * Whether the diff-relevant content of the collected turns changed. + * + * Facts arrive at token rate while a turn streams and in chunks while the + * background pass fills the conversation, so this runs once per frame over + * every turn. Serializing the whole conversation to answer it cost a 144 KiB + * string per frame on a 4,000-turn session; identical entries are now settled + * by reference and only a replaced entry is serialized. + */ +export function sessionDiffInputsChanged( + previous: SessionHistoryInput | undefined, + next: SessionHistoryInput +): boolean { + if (previous === undefined) return true; + const before = previous ?? []; + const after = next ?? []; + if (before.length !== after.length) return true; + for (let index = 0; index < after.length; index += 1) { + const beforeEntry = before[index]; + const afterEntry = after[index]; + if (beforeEntry === afterEntry) continue; + if (diffInputEntryFingerprint(beforeEntry) !== diffInputEntryFingerprint(afterEntry)) { + return true; + } + } + return false; } export function selectProviderDiffTurnIds(history: SessionHistoryInput): string[] { @@ -199,7 +246,8 @@ export function useSessionDiffSummary( const [state, setState] = useState(INITIAL_STATE); const [diffInputsVersion, setDiffInputsVersion] = useState(0); const historyRef = useRef(undefined); - const diffInputsFingerprintRef = useRef(undefined); + /** The entries the last accepted pass produced, for the per-frame compare. */ + const diffInputsSeenRef = useRef(undefined); const fileProviderRef = useRef(fileProvider); const providerSummaryRetryAttemptsRef = useRef(0); const providerSummaryRetryTimeoutRef = useRef(null); @@ -254,7 +302,7 @@ export function useSessionDiffSummary( useEffect(() => { if (!enabled) { historyRef.current = undefined; - diffInputsFingerprintRef.current = undefined; + diffInputsSeenRef.current = undefined; setState(INITIAL_STATE); return undefined; } @@ -401,7 +449,7 @@ export function useSessionDiffSummary( let lease: ReturnType> | null = null; historyRef.current = undefined; - diffInputsFingerprintRef.current = undefined; + diffInputsSeenRef.current = undefined; setDiffInputsVersion((prev) => prev + 1); setState(INITIAL_STATE); @@ -426,7 +474,7 @@ export function useSessionDiffSummary( const derivation = lease.table; const initialHistory = collectDiffInputs(store.history, derivation.facts) as never; historyRef.current = initialHistory; - diffInputsFingerprintRef.current = computeSessionDiffInputsFingerprint(initialHistory); + diffInputsSeenRef.current = initialHistory; setDiffInputsVersion((prev) => prev + 1); if (shouldUpdateFallbackSummary()) { const initialSummary = buildSessionDiffSummary(initialHistory); @@ -456,11 +504,10 @@ export function useSessionDiffSummary( frame = null; if (cancelled) return; const nextHistory = collectDiffInputs(store.history, activeDerivation.facts) as never; - const nextFingerprint = computeSessionDiffInputsFingerprint(nextHistory); - if (nextFingerprint === diffInputsFingerprintRef.current) { + if (!sessionDiffInputsChanged(diffInputsSeenRef.current, nextHistory)) { return; } - diffInputsFingerprintRef.current = nextFingerprint; + diffInputsSeenRef.current = nextHistory; historyRef.current = nextHistory; setDiffInputsVersion((prev) => prev + 1); if (!shouldUpdateFallbackSummary()) { diff --git a/packages/components/src/hooks/use-conversation-stream-items.ts b/packages/components/src/hooks/use-conversation-stream-items.ts index a49648783..bc6b1cd49 100644 --- a/packages/components/src/hooks/use-conversation-stream-items.ts +++ b/packages/components/src/hooks/use-conversation-stream-items.ts @@ -128,12 +128,27 @@ export function useConversationStreamItems( if (cacheRef.current === undefined) { cacheRef.current = chatStreamItemsCacheBySessionId.get(sessionId); } - const result = useMemo( - () => buildChatStreamItems(view, sessionId, cacheRef.current), + const previousResultRef = useRef(null); + const result = useMemo(() => { + const next = buildChatStreamItems(view, sessionId, cacheRef.current); + // Virtual rows, the outline and its anchors all recompute on this array's + // identity, and the view's version bumps at token rate. Per-turn items are + // already memoized, so an unchanged conversation rebuilds an array of the + // same entries — hand back the previous array instead and the whole chain + // below it short-circuits. + const previous = previousResultRef.current; + const reusable = + previous !== null && + previous.lastAssistantMessageId === next.lastAssistantMessageId && + previous.lastCompletedAssistantMessageId === next.lastCompletedAssistantMessageId && + previous.items.length === next.items.length && + previous.items.every((item, index) => item === next.items[index]); + const settled = reusable ? previous : next; + previousResultRef.current = settled; + return settled; // `version` is the change signal for the view's contents. // eslint-disable-next-line react-hooks/exhaustive-deps - [view, version, sessionId] - ); + }, [view, version, sessionId]); cacheRef.current = result.cache; useEffect(() => { chatStreamItemsCacheBySessionId.set(sessionId, result.cache); diff --git a/packages/components/src/hooks/use-conversation-view.ts b/packages/components/src/hooks/use-conversation-view.ts index 5c0330079..9d8311680 100644 --- a/packages/components/src/hooks/use-conversation-view.ts +++ b/packages/components/src/hooks/use-conversation-view.ts @@ -119,14 +119,27 @@ export function useConversationIndexRows( view: ConversationView | null | undefined ): readonly TurnIndexRow[] { const version = useConversationVersion(view); + const previousRef = useRef(EMPTY_ROWS); return useMemo(() => { - if (!view) return EMPTY_ROWS; + if (!view) { + previousRef.current = EMPTY_ROWS; + return EMPTY_ROWS; + } const rows: TurnIndexRow[] = []; for (let i = 0; i < view.turnCount; i += 1) { const row = view.index(i); if (row) rows.push(row); } - return rows; + // The view hands back the same row object for a turn whose index facts did + // not change, so an array of identical rows is the previous array. Every + // consumer of this list recomputes on its identity, and the version bumps + // at token rate. + const previous = previousRef.current; + const reusable = + previous.length === rows.length && previous.every((row, index) => row === rows[index]); + const result = reusable ? previous : rows; + previousRef.current = result; + return result; // `version` is the change signal for the view's contents. // eslint-disable-next-line react-hooks/exhaustive-deps }, [view, version]); diff --git a/packages/components/src/lib/conversation-view/AGENTS.md b/packages/components/src/lib/conversation-view/AGENTS.md index 1b84c4a20..bdc2cd57b 100644 --- a/packages/components/src/lib/conversation-view/AGENTS.md +++ b/packages/components/src/lib/conversation-view/AGENTS.md @@ -16,14 +16,30 @@ turn ids). Every body edit includes its id even when evicted. Derivations drop those cached facts before recomputing; shallow equality cannot detect body changes. Empty `changed.ids` only announces summary/cache bookkeeping. +- Only the reported ids invalidate a body: a directory row cannot tell whether a + body changed, and merging sparse targets into one span re-read the whole + conversation. Refresh those rows in contiguous runs; escalate to a structural + re-key if the length moved. A reported turn always takes the fresh index row; + `rowChanged` compares only what it can, and a deferred send configuration + cannot be diffed without forcing it. A turn nothing reported keeps its row + object — placeholder and Virtua caches key on that identity — so carry forward + the counts a directory refresh does not supply before comparing them. +- A row's send configuration projects on first read and memoizes. Do not force + it while collecting sources; the resolver reads the newest turn or two. - Derivations retain small facts and weak identity hints, not evicted bodies. Structure updates prune deleted ids and restart incomplete coverage. Search refreshes membership/positions after structure changes. - Use the one shared HistoryWriter. A display projection is never a write baseline or export/hash input; `readAll` forwards the authoritative read. The array adapter serves static shared pages, not a runtime fallback. -- Goal, permission, scheduling and diff consumers acquire the same per-view - fact table. Only the final consumer release disposes its background scan. +- Goal, permission, scheduling and diff consumers acquire the same fact table, + keyed and subscribed on `factSource` — the conversation's own view, never a + projection wrapper. Wrappers are rebuilt as optimistic entries appear and + resolve; a table acquired on one stays pinned by the base view's listeners and + keeps deriving after release. The final consumer release HOLDS the background + scan and keeps the facts: deriving one needs the turn's body, so discarding + them re-materialized the conversation on the next open. The table is collected + with its view. - Control-plane Mirror ignores history and does not enumerate its containers. Queue identity must retain non-enumerable `$cid` through Immer, not a `structuredClone` that drops it. diff --git a/packages/components/src/lib/conversation-view/create-conversation-view-from-history.ts b/packages/components/src/lib/conversation-view/create-conversation-view-from-history.ts index 1b2f2aa46..fceade617 100644 --- a/packages/components/src/lib/conversation-view/create-conversation-view-from-history.ts +++ b/packages/components/src/lib/conversation-view/create-conversation-view-from-history.ts @@ -24,6 +24,7 @@ export function createConversationViewFromHistory( let history = options.getHistory(); let indexById = buildIndexById(history); let version = 0; + let structureVersion = 0; let disposed = false; const rowOf = (entry: SessionHistory): TurnIndexRow => { @@ -44,6 +45,7 @@ export function createConversationViewFromHistory( previous.length !== next.length || previous.some((entry, i) => entry?.id !== next[i]?.id); if (structural) { indexById = buildIndexById(next); + structureVersion += 1; } version += 1; if (structural) { @@ -63,6 +65,9 @@ export function createConversationViewFromHistory( get version() { return version; }, + get structureVersion() { + return structureVersion; + }, ready: Promise.resolve(), index: (i) => { const entry = history[i]; diff --git a/packages/components/src/lib/conversation-view/create-conversation-view-from-reader.ts b/packages/components/src/lib/conversation-view/create-conversation-view-from-reader.ts index 5a7c29d63..7807e341d 100644 --- a/packages/components/src/lib/conversation-view/create-conversation-view-from-reader.ts +++ b/packages/components/src/lib/conversation-view/create-conversation-view-from-reader.ts @@ -146,8 +146,17 @@ export function createConversationViewFromReader( // it, so the gap-free initial + the queued events stay ordered. let initialApplied = false; const pendingChanges: SessionDataChange[] = []; + /** Structural refresh range: every position after an insert/delete shifted. */ let dirtyFrom = Infinity; let dirtyTo = -1; + /** + * Content refresh targets, as the ids the reader named. A content + * notification carries exact ids, so the refresh reads exactly those rows. + * Merging them into one `[min, max)` span meant one early status write landing + * in the same batch as a streaming delta re-read the whole directory and + * re-materialized every hydrated body between the two. + */ + const dirtyIds = new Set(); let flushRunning = false; const tailStart = () => conversationTailStart(ids.length, tailKeep); @@ -183,8 +192,32 @@ export function createConversationViewFromReader( // Send-critical metadata comes from the directory row itself, before any // body hydration; the shared projection already kept explicit empty // selections intact. - if (row.role === 'user' && entry.inputConfig !== undefined) { - row.inputConfig = pickIndexInputConfig(entry.inputConfig); + // + // Projecting it costs a schema parse per user turn, and the only consumer + // resolves sticky configuration from the newest turn or two — so the parse + // is deferred to first read and memoized on the row. `'inputConfig' in + // entry` is used instead of a value test because the directory row defers + // its own projection the same way. Opening a 4,000-turn conversation read + // the whole directory eagerly, and those two parses were most of it. + if (row.role === 'user' && 'inputConfig' in entry) { + let projected: TurnIndexRow['inputConfig']; + let done = false; + Object.defineProperty(row, 'inputConfig', { + enumerable: true, + configurable: true, + get: () => { + if (!done) { + done = true; + const source = entry.inputConfig; + projected = source === undefined ? undefined : pickIndexInputConfig(source); + } + return projected; + }, + set: (value: TurnIndexRow['inputConfig']) => { + done = true; + projected = value; + }, + }); } if (entry.itemCount !== undefined) row.itemCount = entry.itemCount; if (entry.planCount !== undefined) row.planCount = entry.planCount; @@ -199,8 +232,10 @@ export function createConversationViewFromReader( planCount: Array.isArray(turn.plan) ? turn.plan.length : 0, summary: summarizeTurn(turn), }; - if (row.inputConfig !== undefined) next.inputConfig = row.inputConfig; + // A user turn's body carries the authoritative configuration, so the + // directory row's deferred projection is never forced here. if (next.role === 'user') next.inputConfig = pickIndexInputConfig(turn.inputConfig); + else if (row.inputConfig !== undefined) next.inputConfig = row.inputConfig; return next; }; @@ -215,7 +250,30 @@ export function createConversationViewFromReader( } }; - /** Whether a directory refresh actually changed the turn's index facts. */ + /** + * Carry the body-derived facts a directory refresh cannot supply. + * + * A container-backed directory row omits `itemCount`/`planCount` (the adapter + * reads them only when they are free), so a refresh of an already-hydrated + * row would otherwise drop the real counts. Placeholder heights and the + * empty-assistant test read them long after the body is evicted, and + * `rowChanged` compares them. + */ + const carryBodyFacts = (old: TurnIndexRow | undefined, next: TurnIndexRow): TurnIndexRow => { + if (!old) return next; + if (next.itemCount === undefined && old.itemCount !== undefined) next.itemCount = old.itemCount; + if (next.planCount === undefined && old.planCount !== undefined) next.planCount = old.planCount; + return next; + }; + + /** + * Whether a directory refresh actually changed the turn's index facts. + * + * Counts are compared only once `carryBodyFacts` has filled the ones the + * refresh did not carry: comparing a hydrated row's real count against the + * directory's `undefined` reported a change on every refresh, which bumped + * every hydrated turn's content epoch and re-read its body. + */ const rowChanged = (old: TurnIndexRow | undefined, next: TurnIndexRow): boolean => { if (!old) return true; return ( @@ -416,7 +474,15 @@ export function createConversationViewFromReader( const applyChange = async ( from: number, entries: readonly SessionDirectoryRow[], - authoritativeCount: number + authoritativeCount: number, + /** + * The turn ids the reader reported as changed, when the flush came from a + * content notification. A directory row cannot tell whether a body changed + * — a grown text item moves no scalar — so this is the only authority for + * invalidating a body. `undefined` means "assume every entry changed", + * which is what a structural refresh needs. + */ + reportedIds?: ReadonlySet ): Promise => { let structuralFrom = Infinity; for (const entry of entries) { @@ -440,30 +506,42 @@ export function createConversationViewFromReader( if (!structural) { const toReRead: string[] = []; const evictedChanges: number[] = []; - let lo = Infinity; - let hi = -1; + let touched = false; for (const entry of entries) { const pos = entry.position; - const row = rowFromDirectory(entry); + const row = carryBodyFacts(rows[pos], rowFromDirectory(entry)); const old = rows[pos]; - if (old && !hydrated.has(old.id)) { + // A turn the notification did not name kept its body: re-reading it + // would cost a full materialization and hand the renderer a new object + // for a turn nothing changed. + const bodyChanged = reportedIds === undefined || reportedIds.has(row.id); + const indexChanged = rowChanged(old, row); + if (!bodyChanged && !indexChanged) continue; + if (bodyChanged && old && !hydrated.has(old.id) && old.summary !== undefined) { // Drop stale previews; the next explicit read will recompute them. - if (old.summary !== undefined) { - row.summary = undefined; - } + row.summary = undefined; } - // Invalidate only the turn(s) whose facts actually changed, so an - // unrelated turn's in-flight body read is not cancelled. - if (rowChanged(old, row)) bumpTurn(row.id); - rows[pos] = row; - if (row.id !== old?.id) rebuildLookups(pos); - if (hydrated.has(row.id)) toReRead.push(row.id); - else evictedChanges.push(pos); - lo = Math.min(lo, pos); - hi = Math.max(hi, pos); + if (bodyChanged || indexChanged) { + // A reported turn always takes the fresh row: `rowChanged` compares + // only the facts it can compare, and a user turn's send configuration + // is a deferred projection that cannot be diffed without forcing it. + // A turn nothing reported keeps its object — placeholder items and + // Virtua rows are keyed by that identity. + rows[pos] = row; + if (row.id !== old?.id) rebuildLookups(pos); + } + if (bodyChanged) { + // Invalidate only the turn(s) the reader named, so an unrelated + // turn's in-flight body read is not cancelled. + bumpTurn(row.id); + if (hydrated.has(row.id)) toReRead.push(row.id); + else evictedChanges.push(pos); + } + touched = true; } + if (!touched) return; bump(); - if (hi >= 0) emit({ kind: 'changed', ids: [] }); + emit({ kind: 'changed', ids: [] }); // Index notifications also occur for summary maintenance. A storage // content edit must separately invalidate body-derived facts even when // this view no longer holds the body. Otherwise an old goal/file diff @@ -522,35 +600,108 @@ export function createConversationViewFromReader( dirtyTo = Math.max(dirtyTo, to); }; + /** + * Contiguous `[lo, hi)` runs covering `positions`, so scattered targets still + * read in as few directory calls as they have runs — and never read the rows + * between two distant runs. + */ + const runsOf = (positions: readonly number[]): [number, number][] => { + const sorted = [...new Set(positions)].sort((a, b) => a - b); + const runs: [number, number][] = []; + for (const position of sorted) { + const last = runs[runs.length - 1]; + if (last && position === last[1]) last[1] = position + 1; + else runs.push([position, position + 1]); + } + return runs; + }; + + const flushStructural = async (from: number, to: number): Promise => { + // A structural refresh re-reads and re-keys the whole range, which subsumes + // any content target inside it. + for (const id of [...dirtyIds]) { + const position = indexById.get(id); + if (position !== undefined && position >= from && position < to) dirtyIds.delete(id); + } + // Read the directory and the count as ONE observation: capture the + // membership epoch first, and if a structural change lands before the + // pair is ready, re-dirty the window so the next iteration re-reads a + // coherent pair instead of pairing old rows with a newer length. + const structureBefore = structureEpoch; + let entries: readonly SessionDirectoryRow[]; + let count: number; + try { + entries = await reader.readDirectory(from, to); + count = await reader.count(); + } catch { + return; + } + if (disposed) return; + if (structureEpoch !== structureBefore) { + mergeDirty(from, to); + return; + } + await applyChange(from, entries, count); + }; + + const flushContent = async (): Promise => { + const reported = new Set(dirtyIds); + dirtyIds.clear(); + const positions: number[] = []; + let lowest = ids.length; + for (const id of reported) { + const position = indexById.get(id); + if (position === undefined) continue; + positions.push(position); + if (position < lowest) lowest = position; + } + if (positions.length === 0) return; + const structureBefore = structureEpoch; + let count: number; + try { + count = await reader.count(); + } catch { + return; + } + if (disposed) return; + // A content notification must not move membership. If the length changed + // anyway, re-key structurally rather than splicing rows from a sparse read. + if (count !== ids.length) { + mergeDirty(lowest, Math.max(count, ids.length)); + return; + } + for (const [lo, hi] of runsOf(positions)) { + if (disposed) return; + let entries: readonly SessionDirectoryRow[]; + try { + entries = await reader.readDirectory(lo, hi); + } catch { + continue; + } + if (disposed) return; + if (structureEpoch !== structureBefore) { + mergeDirty(lo, ids.length); + return; + } + await applyChange(lo, entries, count, reported); + } + }; + const flushDirty = async () => { if (flushRunning) return; flushRunning = true; try { - while (dirtyFrom <= dirtyTo) { + while (dirtyFrom <= dirtyTo || dirtyIds.size > 0) { if (disposed) break; - const from = dirtyFrom; - const to = dirtyTo; - dirtyFrom = Infinity; - dirtyTo = -1; - // Read the directory and the count as ONE observation: capture the - // membership epoch first, and if a structural change lands before the - // pair is ready, re-dirty the window so the next iteration re-reads a - // coherent pair instead of pairing old rows with a newer length. - const structureBefore = structureEpoch; - let entries: readonly SessionDirectoryRow[]; - let count: number; - try { - entries = await reader.readDirectory(from, to); - count = await reader.count(); - } catch { - continue; - } - if (disposed) break; - if (structureEpoch !== structureBefore) { - mergeDirty(from, to); + if (dirtyFrom <= dirtyTo) { + const from = dirtyFrom; + const to = dirtyTo; + dirtyFrom = Infinity; + dirtyTo = -1; + await flushStructural(from, to); continue; } - await applyChange(from, entries, count); + await flushContent(); } } finally { flushRunning = false; @@ -568,10 +719,11 @@ export function createConversationViewFromReader( structureEpoch++; mergeDirty(change.from, change.to); } else { + // Positions are resolved at flush time, not here: an id's position can + // move between the notification and the refresh. for (const id of change.ids) { bumpTurn(id); - const position = indexById.get(id); - if (position !== undefined) mergeDirty(position, position + 1); + dirtyIds.add(id); } } void flushDirty(); @@ -631,6 +783,9 @@ export function createConversationViewFromReader( get version() { return version; }, + get structureVersion() { + return structureEpoch; + }, ready, index: (i) => rows[i], indexOf: (turnId) => indexById.get(turnId) ?? -1, diff --git a/packages/components/src/lib/conversation-view/derivation.ts b/packages/components/src/lib/conversation-view/derivation.ts index 9068b2e30..5cacf6a96 100644 --- a/packages/components/src/lib/conversation-view/derivation.ts +++ b/packages/components/src/lib/conversation-view/derivation.ts @@ -19,6 +19,11 @@ export type ConversationDerivation = { readonly complete: boolean; readonly version: number; subscribe(listener: () => void): () => void; + /** + * Run or hold the background pass. Facts and view subscription survive a + * hold, so a table that is re-acquired resumes instead of starting over. + */ + setActive(active: boolean): void; dispose(): void; }; @@ -31,7 +36,19 @@ export type CreateConversationDerivationOptions = { yieldToEventLoop?: () => Promise; }; -const defaultYield = (): Promise => new Promise((resolve) => setTimeout(resolve, 0)); +/** + * Background chunks yield to real idle time where the platform has it. + * + * Back-to-back macrotasks kept the pass at the head of the queue, so filling a + * long conversation's facts held the main thread for as long as it ran. The + * timeout keeps it progressing on a busy tab. + */ +const defaultYield = (): Promise => + typeof requestIdleCallback === 'function' + ? new Promise((resolve) => { + requestIdleCallback(() => resolve(), { timeout: 200 }); + }) + : new Promise((resolve) => setTimeout(resolve, 0)); export function createConversationDerivation( view: ConversationView, @@ -47,6 +64,7 @@ export function createConversationDerivation( let version = 0; let complete = false; let disposed = false; + let active = true; let passRunning = false; let passRequested = false; let activeRange: ReturnType | undefined; @@ -121,7 +139,7 @@ export function createConversationDerivation( const runBackgroundPass = async () => { let end = view.turnCount; while (end > 0) { - if (disposed) return; + if (disposed || !active) return; // Next chunk of turns (from the tail backwards) that still lack a fact. const pending: number[] = []; let cursor = end; @@ -162,10 +180,16 @@ export function createConversationDerivation( passRunning = true; try { while (passRequested) { - // `disposed` flips from `dispose()` while this loop is awaiting. - if (disposed) return; + // `disposed` flips from `dispose()` while this loop is awaiting, and + // `active` from the last consumer releasing. + if (disposed || !active) return; passRequested = false; await runBackgroundPass(); + // A pass that stopped because it was held has not covered everything. + if (!active) { + passRequested = true; + return; + } } if (disposed) return; complete = true; @@ -178,7 +202,7 @@ export function createConversationDerivation( function requestPass(): void { passRequested = true; complete = false; - if (!passRunning && !disposed) void runPasses(); + if (active && !passRunning && !disposed) void runPasses(); } // Facts for what is already hydrated come for free before the pass starts. @@ -201,6 +225,11 @@ export function createConversationDerivation( listeners.delete(listener); }; }, + setActive: (next: boolean) => { + if (active === next || disposed) return; + active = next; + if (active && passRequested && !passRunning) void runPasses(); + }, dispose: () => { disposed = true; activeRange?.release(); @@ -217,29 +246,42 @@ const sharedDerivations = new WeakMap< Map, { table: ConversationDerivation; users: number }> >(); -/** Borrow a fact table; release the background scan after the last consumer. */ +/** + * Borrow a fact table; hold its background scan after the last consumer. + * + * The table is NOT disposed on the last release. Deriving a fact needs the + * turn's body, so a table that discarded its facts re-materialized the whole + * conversation the next time the session was opened — which is what closing + * and reopening a tab does. It stays keyed by the view instead, so it is + * collected with the view when the session store evicts it, and a re-acquire + * resumes with the facts it already has. + */ export function acquireConversationDerivation( view: ConversationView, derive: DeriveTurnFact ) { - let tables = sharedDerivations.get(view); - if (!tables) sharedDerivations.set(view, (tables = new Map())); + // Key and subscribe on the conversation's own view. A projection wrapper is + // rebuilt whenever an optimistic entry appears or resolves; giving each one a + // table left every released wrapper subscribed through the base view, so a + // later token derived once per wrapper ever created and none of them could be + // collected. + const owner = view.factSource ?? view; + let tables = sharedDerivations.get(owner); + if (!tables) sharedDerivations.set(owner, (tables = new Map())); let entry = tables.get(derive); if (!entry) { - entry = { table: createConversationDerivation(view, derive), users: 0 }; + entry = { table: createConversationDerivation(owner, derive), users: 0 }; tables.set(derive, entry); } entry.users++; - let active = true; + entry.table.setActive(true); + let held = true; return { table: entry.table as ConversationDerivation, release() { - if (!active) return; - active = false; - if (--entry.users === 0) { - entry.table.dispose(); - tables.delete(derive); - } + if (!held) return; + held = false; + if (--entry.users === 0) entry.table.setActive(false); }, }; } diff --git a/packages/components/src/lib/conversation-view/index-queries.ts b/packages/components/src/lib/conversation-view/index-queries.ts index a0ff3e287..af2e2f72b 100644 --- a/packages/components/src/lib/conversation-view/index-queries.ts +++ b/packages/components/src/lib/conversation-view/index-queries.ts @@ -103,6 +103,12 @@ export function collectHydratedRange( return turns; } +/** Last result per view, so an unchanged conversation reuses its array. */ +const previousConfigSources = new WeakMap< + object, + { id: string; role: unknown; inputConfig?: unknown }[] +>(); + /** * Source rows for `resolveSessionConversationConfig` and the source fence: the * hydrated tail (full input config) followed by every older user turn as an @@ -116,11 +122,27 @@ export function collectConversationConfigSources( for (let i = 0; i < tailFrom; i += 1) { const row = view.index(i); if (!row || row.role !== 'user') continue; - sources.push({ id: row.id, role: row.role, inputConfig: row.inputConfig }); + // The index row already IS a source: same id and role, and an + // `inputConfig` that projects itself on first read. Wrapping it would + // allocate one object per historical user turn on every streamed delta, + // and reading its config here would parse the whole conversation to + // resolve a question about its tail. + sources.push(row); } for (let i = tailFrom; i < view.turnCount; i += 1) { const turn = view.turn(i) ?? view.index(i); if (turn) sources.push(turn as { id: string; role: unknown; inputConfig?: unknown }); } + // Callers memoize the resolved configuration on this array's identity, so an + // unchanged conversation must hand back the same array. + const previous = previousConfigSources.get(view); + if ( + previous && + previous.length === sources.length && + previous.every((source, index) => source === sources[index]) + ) { + return previous; + } + previousConfigSources.set(view, sources); return sources; } diff --git a/packages/components/src/lib/conversation-view/projected-conversation-view.ts b/packages/components/src/lib/conversation-view/projected-conversation-view.ts index 35676444d..4ff992b36 100644 --- a/packages/components/src/lib/conversation-view/projected-conversation-view.ts +++ b/packages/components/src/lib/conversation-view/projected-conversation-view.ts @@ -19,14 +19,20 @@ export function createProjectedConversationView( ): ConversationView { if (projections.length === 0) return base; - let slotsVersion = -1; + let slotsStructure = -1; + let slotsCount = -1; let slots: Slot[] = []; let baseToSlot: number[] = []; let slotById = new Map(); const rebuild = () => { - if (slotsVersion === base.version) return; - slotsVersion = base.version; + // Slots follow membership, never content: a streamed delta leaves every + // turn where it was. Keying this on `version` rebuilt an array of the whole + // conversation on every token for as long as an unconfirmed entry existed — + // which is exactly while a turn streams. + if (slotsStructure === base.structureVersion && slotsCount === base.turnCount) return; + slotsStructure = base.structureVersion; + slotsCount = base.turnCount; const list: Slot[] = Array.from({ length: base.turnCount }, (_, i) => ({ base: i })); const idOfSlot = (slot: Slot) => ('base' in slot ? base.index(slot.base)?.id : slot.entry.id); const seen = new Set(); @@ -91,6 +97,13 @@ export function createProjectedConversationView( get version() { return base.version; }, + get structureVersion() { + return base.structureVersion; + }, + // Fact tables belong to the conversation, not to this wrapper: wrappers are + // rebuilt as optimistic entries appear and resolve, and a table acquired on + // one would be pinned by the base view's listener set forever. + factSource: base.factSource ?? base, ready: base.ready, index: (i) => { rebuild(); diff --git a/packages/components/src/lib/conversation-view/types.ts b/packages/components/src/lib/conversation-view/types.ts index a4be28258..04a656415 100644 --- a/packages/components/src/lib/conversation-view/types.ts +++ b/packages/components/src/lib/conversation-view/types.ts @@ -110,6 +110,23 @@ export interface ConversationView { readonly turnCount: number; /** Bumps on any structural, index, or hydrated-content change. */ readonly version: number; + /** + * Bumps only when membership or order moves. A consumer whose work depends + * on the turn LIST rather than its contents keys on this: `version` bumps at + * token rate, and rebuilding a per-turn layout that often is pure waste. + */ + readonly structureVersion: number; + /** + * The view that owns this conversation's shared per-turn fact tables. A + * projection wrapper points at the view it wraps; everything else leaves it + * unset and owns its own. + * + * A wrapper is rebuilt whenever an optimistic entry appears or resolves. A + * table acquired on one would subscribe through it, so the underlying view's + * listener set would keep every released wrapper — and its table — alive and + * deriving. + */ + readonly factSource?: ConversationView; /** Resolves once the initial directory and retained tail are ready; offscreen summaries stay lazy. */ readonly ready: Promise; index(i: number): TurnIndexRow | undefined; diff --git a/packages/components/src/lib/markdown-single-dollar-math.ts b/packages/components/src/lib/markdown-single-dollar-math.ts index 4f8528e44..e7f756592 100644 --- a/packages/components/src/lib/markdown-single-dollar-math.ts +++ b/packages/components/src/lib/markdown-single-dollar-math.ts @@ -217,6 +217,11 @@ const slashRunLength = (value: string, start: number): number => { * stay valid. */ export const normalizeTexMathDelimiters = (value: string): string => { + // Only an opening `\[` can produce a replacement, so text without one is + // returned unchanged. The scanner below walks the string character by + // character and a streaming turn re-runs it over the whole accumulated + // answer on every delta, which is quadratic in the answer's length. + if (!value.includes('\\[')) return value; const replacements: number[] = []; let opening: TexMathDelimiter | null = null; let cursor = 0; diff --git a/packages/components/tests/conversation-derivation.test.ts b/packages/components/tests/conversation-derivation.test.ts index f58ee1f41..ac828bad1 100644 --- a/packages/components/tests/conversation-derivation.test.ts +++ b/packages/components/tests/conversation-derivation.test.ts @@ -13,8 +13,11 @@ import { createLoroSessionData, type LoroSessionData } from '@lody/shared/sessio import { createConversationDerivation, createConversationViewFromReader, + createProjectedConversationView, type ConversationView, } from '../src/lib/conversation-view'; +import type { AcceptedSessionHistoryProjection } from '../src/atoms/session-history-projection'; +import type { SessionHistory, SessionId, WorkspaceId } from '@lody/shared'; import { buildFixtureHistory, buildSessionDoc, @@ -65,7 +68,7 @@ const deriveDiffCount = (turn: { fileDiff?: unknown }) => ({ }); describe('createConversationDerivation', () => { - it('shares goal and diff facts until the last consumer releases the view', async () => { + it('shares goal and diff facts and holds the table after the last consumer releases', async () => { const { view, doc } = await openView(2, { tailKeep: 4, maxHydrated: 4 }); const goalReader = acquireConversationDerivation(view, deriveSessionTurnFacts); const diffReader = acquireConversationDerivation(view, deriveSessionTurnFacts); @@ -77,8 +80,83 @@ describe('createConversationDerivation', () => { await flushReaderChanges(); await drain(() => diffReader.table.facts.get('a-0')?.fileDiff?.length === 0); expect(diffReader.table.facts.get('a-0')?.fileDiff).toEqual([]); + const held = diffReader.table; diffReader.release(); - expect(diffReader.table.facts.size).toBe(0); + // Deriving a fact needs the turn's body, so discarding the table on the + // last release re-materialized the whole conversation the next time the + // session was opened. Facts survive the release instead. + expect(held.facts.get('a-0')?.fileDiff).toEqual([]); + expect(held.facts.size).toBeGreaterThan(0); + + // ...and the background pass is held while nothing is reading. Turns that + // land inside the retained tail are still derived for free; one that falls + // outside it needs the pass, and stays underived until someone re-acquires. + const peer = reimport(doc); + const appended = buildFixtureHistory(12).slice(4); + const peerWriter = createHistoryWriter(peer); + for (const entry of appended) peerWriter.append(entry); + doc.import(peer.export({ mode: 'update', from: doc.version() })); + await flushReaderChanges(); + const appendedId = appended[0]!.id; + expect(view.isHydrated(view.indexOf(appendedId))).toBe(false); + expect(held.facts.has(appendedId)).toBe(false); + + const reopened = acquireConversationDerivation(view, deriveSessionTurnFacts); + expect(reopened.table).toBe(held); + await drain(() => reopened.table.facts.has(appendedId)); + expect(reopened.table.facts.has(appendedId)).toBe(true); + reopened.release(); + data.dispose(); + view.dispose(); + }); + + it('shares one table across projection wrappers and stops deriving for released ones', async () => { + const { doc, view } = await openView(3, { tailKeep: 4, maxHydrated: 8 }); + const data = createLoroSessionData({ doc, sessionId: FIXTURE_SESSION_ID }); + // An optimistic entry appears and then resolves, so `useSessionDoc` builds a + // NEW projection wrapper each time. A wrapper must not own a fact table: + // the table would subscribe through it, the base view's listener set would + // keep the released wrapper alive, and every later token would derive once + // per wrapper ever created. + const projection = (id: string): AcceptedSessionHistoryProjection => ({ + workspaceId: 'workspace-fixture' as WorkspaceId, + sessionId: FIXTURE_SESSION_ID as SessionId, + entry: { + id, + role: 'user', + timestamp: '2026-01-01T00:01:00.000Z', + items: [{ type: 'text', text: id }], + fileDiff: [], + } as unknown as SessionHistory, + }); + + let derived = 0; + const countingDerive = (turn: { items?: unknown }) => { + derived += 1; + return { items: Array.isArray(turn.items) ? turn.items.length : 0 }; + }; + + const first = createProjectedConversationView(view, [projection('optimistic-1')]); + const second = createProjectedConversationView(view, [projection('optimistic-2')]); + expect(first).not.toBe(second); + + const firstLease = acquireConversationDerivation(first, countingDerive); + await drain(() => firstLease.table.complete); + firstLease.release(); + + const secondLease = acquireConversationDerivation(second, countingDerive); + await drain(() => secondLease.table.complete); + // One table for the conversation, not one per wrapper. + expect(secondLease.table).toBe(firstLease.table); + + const before = derived; + data.writer.setField('a-0', 'finished', false as never); + await flushReaderChanges(); + await drain(() => derived > before); + // Exactly one derivation of the changed turn, not one per released wrapper. + expect(derived - before).toBe(1); + + secondLease.release(); data.dispose(); view.dispose(); }); diff --git a/packages/components/tests/conversation-view-from-reader.test.ts b/packages/components/tests/conversation-view-from-reader.test.ts index 83c8dc358..05efd5c9e 100644 --- a/packages/components/tests/conversation-view-from-reader.test.ts +++ b/packages/components/tests/conversation-view-from-reader.test.ts @@ -483,6 +483,97 @@ describe.each(backends)('createConversationViewFromReader over $name', (backend) } }); + it('re-reads only the named turns when one synced batch carries an early edit and the tail', async () => { + // The desktop path: the daemon commits into its own document and the + // renderer applies the batch as one import, so both edits arrive in a + // single observation. + const history = buildFixtureHistory(12); + const source = buildSessionDoc(history); + const doc = new LoroDoc(); + doc.setPeerId(2); + doc.import(source.export({ mode: 'snapshot' })); + const data = createLoroSessionData({ sessionId: FIXTURE_SESSION_ID, doc }); + const probe = probeReader(data.history); + const view = createConversationViewFromReader(probe.reader, { + sessionId: FIXTURE_SESSION_ID, + tailKeep: 4, + maxHydrated: 32, + scheduleIdle: () => () => {}, + yieldToEventLoop: () => Promise.resolve(), + }); + try { + await waitTurns(view, history.length); + // A reader parked over the head keeps those turns hydrated; a batch that + // touches one of them must not drag the rest in with it. + const head = view.acquireRange(0, 8); + await head.ready; + const last = view.turnCount - 1; + const lastId = view.index(last)!.id; + const earlyId = view.index(1)!.id; + expect(view.isHydrated(1)).toBe(true); + expect(view.isHydrated(last)).toBe(true); + probe.directories.length = 0; + probe.turns.length = 0; + + const writer = createHistoryWriter(source); + writer.setField(earlyId, 'finished', false as never); + writer.updateEntry(lastId, (entry) => ({ + ...entry, + items: [{ type: 'text', text: 'streamed token' }], + })); + doc.import(source.export({ mode: 'update', from: doc.version() })); + await flush(); + + // Two rows, not the span between them, and two bodies, not every hydrated + // turn in that span. + const rowsRead = probe.directories.reduce((total, [lo, hi]) => total + (hi - lo), 0); + expect(rowsRead).toBe(2); + expect([...probe.turns].sort()).toEqual([earlyId, lastId].sort()); + expect(view.index(1)?.finished).toBe(false); + expect(view.turn(last)?.items).toEqual([{ type: 'text', text: 'streamed token' }]); + head.release(); + } finally { + view.dispose(); + doc.free(); + source.free(); + } + }); + + it("refreshes an unhydrated user turn's send configuration when it changes", async () => { + const harness = openView(backend, 12, { tailKeep: 2, maxHydrated: 4 }); + const { idle, data } = harness; + try { + await settle(idle, harness.view); + const position = harness.view.indexOf('u-1'); + expect(position).toBeGreaterThanOrEqual(0); + // Outside every window: the index row is all a caller can read, and the + // sticky send-config resolver reads exactly this. + expect(harness.view.isHydrated(position)).toBe(false); + expect(harness.view.index(position)?.inputConfig?.modelId).toBe('sonnet'); + + data.writer.updateEntry('u-1', (entry) => ({ + ...entry, + inputConfig: { + ...(entry.inputConfig as Record), + modelId: 'opus', + agentRoleId: 'role-reassigned', + agentRoleRevision: 42, + mcpServerIds: [], + }, + })); + await flush(); + + const row = harness.view.index(position); + expect(row?.inputConfig?.modelId).toBe('opus'); + expect(row?.inputConfig?.agentRoleId).toBe('role-reassigned'); + expect(row?.inputConfig?.agentRoleRevision).toBe(42); + expect(row?.inputConfig?.mcpServerIds).toEqual([]); + } finally { + harness.view.dispose(); + harness.teardown(); + } + }); + it('patches a streamed tail update from its ranged event without a whole-history read', async () => { const harness = openView(backend, 12, { tailKeep: 4 }); const { idle, data } = harness; diff --git a/packages/shared/src/session-data/AGENTS.md b/packages/shared/src/session-data/AGENTS.md index 5aa01dbc3..27132da34 100644 --- a/packages/shared/src/session-data/AGENTS.md +++ b/packages/shared/src/session-data/AGENTS.md @@ -6,13 +6,18 @@ diff, legacy representation, stored-copy and conditional rollback rules. Shared business rules live in the planners, not in UI or CLI copies. - Directory reads contain identities, scalars and input configuration, never - bodies. Targeted reads materialize only the selected turn. Status queries use + bodies. Input configuration projects on first read and memoizes: the + projection is a schema parse per user turn, and a directory read covers the + whole conversation to answer a question about its tail. Targeted reads materialize only the selected turn. Status queries use the directory; explicit export/replay uses one consistent `readAll` observation. `readTurnOutput` reads the selected assistant and relevant failure notices in one observation instead of serializing the transcript on every token. - Repeated identity lookups reuse an unchanged pending transaction. Reads must not commit writes; uncommitted structural/ID edits and checkout still invalidate lookup. -- Subscribe and capture the initial directory without a gap. Notifications carry structural ranges or changed turn ids. The display +- Subscribe and capture the initial directory without a gap. Notifications carry + structural ranges or changed turn ids. A content batch reports the EXACT + positions it touched, never the span between the lowest and highest: one batch + routinely carries an early status write plus the streaming tail. The display cache rejects stale async reads. Turn ID edits (including deletion) are structural: notify by position so readers can remove the old identity before re-keying. CLI reads in-process synchronously; auto-seen scans directory scalars only and permission checks have no await gap. diff --git a/packages/shared/src/session-data/loro.ts b/packages/shared/src/session-data/loro.ts index dbcdd15af..a3cf94df9 100644 --- a/packages/shared/src/session-data/loro.ts +++ b/packages/shared/src/session-data/loro.ts @@ -111,10 +111,13 @@ export function createLoroSessionData(options: LoroSessionDataOptions) { const list = doc.getList(HISTORY_ROOT_KEY); const issuesOf = (error: HistoryWriteError) => error.issues; - /** Shallow send config for a user turn: small collections only, never the body. */ - const shallowInputConfig = (map: LoroMap): unknown => { + /** + * Shallow send config source for a user turn: small collections only, never + * the body. The container crossings here are cheap; the projection is not. + */ + const shallowInputConfigSource = (map: LoroMap): unknown => { const config = map.get('inputConfig'); - if (!isContainer(config)) return pickDirectoryInputConfig(config); + if (!isContainer(config)) return config; if (config.kind() !== 'Map') return undefined; const configMap = config as LoroMap; const value = { ...configMap.getShallowValue() } as Record; @@ -123,7 +126,32 @@ export function createLoroSessionData(options: LoroSessionDataOptions) { const field = configMap.get(key); value[key] = isContainer(field) ? (field as LoroList).toJSON() : field; } - return pickDirectoryInputConfig(value); + return value; + }; + + /** + * Attach the row's send configuration as a deferred, memoized projection. + * + * `pickDirectoryInputConfig` runs a schema parse. A directory read covers + * every user turn in the conversation, while its consumers resolve sticky + * configuration from the newest turn or two — so opening a long session paid + * thousands of parses to answer a question about its tail. + */ + const withDeferredInputConfig = (row: T, source: unknown): T => { + let projected: unknown; + let done = false; + Object.defineProperty(row, 'inputConfig', { + enumerable: true, + configurable: true, + get: () => { + if (!done) { + done = true; + projected = pickDirectoryInputConfig(source); + } + return projected; + }, + }); + return row; }; const readDirectoryRow = (position: number): SessionDirectoryRow => { @@ -133,16 +161,13 @@ export function createLoroSessionData(options: LoroSessionDataOptions) { const map = value as LoroMap; const scalars = pickDirectoryScalars(map.getShallowValue()); if (!scalars) return { position, state: 'invalid' }; - return { - position, - state: 'ready', - turnId: scalars.id, - scalars, - // Send config is eager for user turns; counts are deliberately omitted - // here (one container crossing each) and arrive with a summary or a - // hydration read. - ...(scalars.role === 'user' ? { inputConfig: shallowInputConfig(map) } : {}), - }; + const row: SessionDirectoryRow = { position, state: 'ready', turnId: scalars.id, scalars }; + // Send config is present for user turns but projected on first read; + // counts are deliberately omitted here (one container crossing each) and + // arrive with a summary or a hydration read. + return scalars.role === 'user' + ? withDeferredInputConfig(row, shallowInputConfigSource(map)) + : row; } if (value && typeof value === 'object' && !Array.isArray(value)) { const record = value as Record; @@ -150,17 +175,15 @@ export function createLoroSessionData(options: LoroSessionDataOptions) { if (!scalars) return { position, state: 'invalid' }; const itemCount = Array.isArray(record.items) ? record.items.length : undefined; const planCount = Array.isArray(record.plan) ? record.plan.length : undefined; - return { + const row: SessionDirectoryRow = { position, state: 'ready', turnId: scalars.id, scalars, - ...(scalars.role === 'user' - ? { inputConfig: pickDirectoryInputConfig(record.inputConfig) } - : {}), ...(itemCount !== undefined ? { itemCount } : {}), ...(planCount !== undefined ? { planCount } : {}), }; + return scalars.role === 'user' ? withDeferredInputConfig(row, record.inputConfig) : row; } return { position, state: 'invalid' }; }; @@ -182,17 +205,27 @@ export function createLoroSessionData(options: LoroSessionDataOptions) { Object.hasOwn(event.diff.updated, 'id'); /** - * The raw positions a batch touched, for a consumer that re-reads only the - * affected window. Structural list edits report `[structuralFrom, length)` - * because later positions shifted. In-place turn ID edits also need positions: - * a consumer still keyed by the old ID cannot resolve the new (or missing) ID. - * Other child edits report their own turn. + * What a batch touched, for a consumer that re-reads only the affected turns. + * + * Structural list edits report `[structuralFrom, length)` because later + * positions shifted. In-place turn ID edits are structural for the same + * reason: a consumer still keyed by the old ID cannot resolve the new (or + * missing) one. + * + * A content batch reports the EXACT positions it touched, never the span + * between the lowest and highest of them. One synced batch routinely carries + * an early turn's status write alongside the streaming tail; reporting the + * span made the display cache re-read every row and re-materialize every + * hydrated body between the two. */ - const changeRangeOf = ( + const changeScopeOf = ( batch: LoroEventBatch - ): { from: number; to: number; structural: boolean } | undefined => { - let from = Number.POSITIVE_INFINITY; - let to = -1; + ): + | { structural: true; from: number; to: number } + | { structural: false; positions: readonly number[] } + | undefined => { + const positions = new Set(); + let wholeDirectory = false; let structuralFrom = Number.POSITIVE_INFINITY; let structural = false; for (const event of batch.events) { @@ -218,29 +251,36 @@ export function createLoroSessionData(options: LoroSessionDataOptions) { structural = true; structuralFrom = Math.min(structuralFrom, index); } - from = Math.min(from, index); - to = Math.max(to, index + 1); + positions.add(index); } else { - from = 0; - to = list.length; + // An edit that does not resolve to a slot: the whole directory is the + // only safe answer. + wholeDirectory = true; } } if (structural) { // A mixed batch can carry an earlier child edit (a content change) plus a // list insert/delete. Keep the earlier content position too, so the // consumer re-reads every affected identity, not just the shifted suffix. - const contentFrom = Number.isFinite(from) ? from : structuralFrom; + let contentFrom = wholeDirectory ? 0 : Number.POSITIVE_INFINITY; + for (const position of positions) contentFrom = Math.min(contentFrom, position); + if (!Number.isFinite(contentFrom)) contentFrom = structuralFrom; const lo = Number.isFinite(structuralFrom) ? Math.min(structuralFrom, contentFrom) : contentFrom; - return { from: Math.max(0, Math.min(lo, list.length)), to: list.length, structural: true }; + return { structural: true, from: Math.max(0, Math.min(lo, list.length)), to: list.length }; + } + if (wholeDirectory) { + return { + structural: false, + positions: Array.from({ length: list.length }, (_, index) => index), + }; } - if (to < 0) return undefined; - return { - from: Math.max(0, Math.min(from, list.length)), - to: Math.max(0, Math.min(to, list.length)), - structural: false, - }; + if (positions.size === 0) return undefined; + const inRange = [...positions] + .filter((position) => position >= 0 && position < list.length) + .sort((left, right) => left - right); + return inRange.length > 0 ? { structural: false, positions: inRange } : undefined; }; // One shallow identity scan per structural/identity change, never per body. @@ -325,16 +365,16 @@ export function createLoroSessionData(options: LoroSessionDataOptions) { // Subscribe first, then snapshot in the same synchronous block: a change // can neither be missed between the two nor delivered before `initial`. const unsubscribeDoc = doc.subscribe((batch) => { - const range = changeRangeOf(batch); + const scope = changeScopeOf(batch); // A batch that does not touch `history` (e.g. a control root) is not a // history change; unrelated roots never invalidate the display cache. - if (!range) return; - if (range.structural) { - listener({ kind: 'structure', from: range.from, to: range.to }); + if (!scope) return; + if (scope.structural) { + listener({ kind: 'structure', from: scope.from, to: scope.to }); } else { const ids: string[] = []; - for (let i = range.from; i < range.to; i++) { - const id = readIdentity(list.get(i))?.turnId; + for (const position of scope.positions) { + const id = readIdentity(list.get(position))?.turnId; if (id !== undefined) ids.push(id); } listener({ kind: 'changed', ids }); diff --git a/packages/shared/src/session-input.ts b/packages/shared/src/session-input.ts index e98d9abc5..d48749065 100644 --- a/packages/shared/src/session-input.ts +++ b/packages/shared/src/session-input.ts @@ -101,7 +101,14 @@ const collectSessionConversationSources = ( const entry = history[index]; if (entry?.role !== 'user') continue; sources.push({ - value: entry.inputConfig, + // Read on demand: a windowed index row resolves its send configuration + // lazily (a schema parse per turn), and `resolveSessionConversationConfig` + // inspects the newest source plus however few older ones it takes to find + // an explicit Role. Reading every entry here would parse the whole + // conversation to answer a question about its tail. + get value() { + return entry.inputConfig; + }, configKey: `history:${entry.id}`, turnKey: `turn:${entry.id}`, });