From a47e80678e34aae9f70e8d6826e909a972937b8a Mon Sep 17 00:00:00 2001 From: kingwl Date: Tue, 18 Aug 2026 15:35:43 +0800 Subject: [PATCH] fix(history): preserve per-delta replay --- ...packed-session-history-transport.i18n.yaml | 4 +- ...-08-15-packed-session-history-transport.md | 16 +-- ...-15-packed-session-history-transport.zh.md | 16 +-- .../src/client/sessions/history-records.ts | 91 +----------- .../tests/history-records.client.spec.ts | 135 ++++++------------ .../tests/history-transport.perf.client.ts | 55 ++++++- 6 files changed, 118 insertions(+), 199 deletions(-) diff --git a/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.i18n.yaml b/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.i18n.yaml index 6a5aa7ecfc..086cb661be 100644 --- a/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write .agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.md -2026-08-15-packed-session-history-transport.md: 3766c6454a65f2350832e873b59ee8afd6984c4a -2026-08-15-packed-session-history-transport.zh.md: 9ef5ea8caeb4f952c9b69473ddb21d6ee18f1c57 +2026-08-15-packed-session-history-transport.md: 9f4c49b8c106e65371b54d07817a58d2bf324fd5 +2026-08-15-packed-session-history-transport.zh.md: 883dd72346efdf5b50ca64b64fc709b257ba8ed7 diff --git a/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.md b/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.md index 3766c6454a..9f4c49b8c1 100644 --- a/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.md +++ b/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.md @@ -6,7 +6,7 @@ English | [中文](2026-08-15-packed-session-history-transport.zh.md) ## Problem -`session.history` and `subagent.history` serve a bounded logical Session-event interval to remote clients. Provider streams can place hundreds of thousands of token-sized `assistant/chunk` events in one incomplete tail. Expanding every persisted row and then serializing every logical event repeats the same envelope on the wire; expanding every record again in the browser recreates the same object fan-out before the conversation fold joins the text. +`session.history` and `subagent.history` serve a bounded logical Session-event interval to remote clients. Provider streams can place hundreds of thousands of token-sized `assistant/chunk` events in one incomplete tail. Expanding every persisted row and then serializing every logical event repeats the same envelope on the wire and makes browser parsing and validation process that repetition before conversation replay can begin. The transport must remain lossless. Session sequence numbers are pagination and reconnect evidence; exact token boundaries remain useful to diagnostics and non-UI API consumers; live streaming, durable export, replay, and model-history derivation continue to require the canonical event stream. A server-side transcript projection that discards completed-step chunks would make the API's evidence depend on one UI policy. @@ -14,9 +14,9 @@ The transport must remain lossless. Session sequence numbers are pagination and History methods return `records: HistoryRecord[]` plus inclusive `fromSeq` and exclusive `toSeq` watermarks. An ordinary record carries `{event, view?}`. Consecutive same-block Assistant delta events carry `{chunks: ChunkRow}` using the shared lossless codec from [the packed JSONL decision](2026-07-26-packed-chunk-rows-by-default.md). The page is selected from logical events before packing, so message-aligned pagination remains independent of physical persistence layout. -The wire schema validates every row, rejects unsafe reconstruction, and requires the records to cover `[fromSeq, toSeq)` exactly without gaps or overlaps. The watermarks, not the number or visible seq adjacency of browser fold inputs, own older-page stitching, reconnect repair, and live-event deduplication. `session.history` and `subagent.history` share the same response schema. +The wire schema validates every row, rejects unsafe reconstruction, and requires the records to cover `[fromSeq, toSeq)` exactly without gaps or overlaps. The watermarks, not the number of transport records, own older-page stitching, reconnect repair, and live-event deduplication. `session.history` and `subagent.history` share the same response schema. -The ordinary browser UI does not decode a packed row into one object per token. It coalesces a row into at most two `assistant/chunk` inputs while preserving accumulated content, the first non-empty token timestamp, and a later first non-whitespace visibility timestamp when those boundaries differ. Tool-call rows retain call identity, name presence, joined argument fragments, and first-token timing. Other API consumers may call `decodeStorageRecord()` when exact token boundaries are required. +The browser calls the shared `decodeStorageRecord()` codec before handing history to `ConversationNodeAssembler`. Every packed member becomes its exact original `assistant/chunk` event, including `seq`, timestamp, chunk type, block index, text or argument fragment, call identity, and optional-name presence. A registered `ConversationNodeDefinition` therefore receives one `match()` call per historical delta and folds accepted matches with the same start/update sequence it observes for live events. Packing changes transport encoding without changing the public Definition replay semantics. Live `session/event` frames remain individual events. Session persistence, raw export, replay, model-history derivation, and the canonical in-memory log are unchanged. @@ -32,15 +32,15 @@ A production-sized private session sample was measured without retaining or comm Packing reduced uncompressed JSON by 90.8% relative to raw logical events and by 83.4% relative to the lossy completed-step projection candidate. Brotli output was 73.2% smaller than raw and 44.8% smaller than that projection candidate. These figures describe this sample rather than a protocol guarantee; savings scale with the length and regularity of delta runs. -The opt-in `packages/client/runtime/tests/history-transport.perf.client.ts` benchmark constructs the same logical-event, ordinary-event, and delta-run cardinalities from synthetic content. `DSH_SNAPSHOT=replay pnpm exec vitest run --config vitest.web.perf.config.ts packages/client/runtime/tests/history-transport.perf.client.ts` reports wire sizes, Host/client timing, and sampled additional V8 heap peaks under `HISTORY_TRANSPORT_PERF_RESULT`. Heap measurements force garbage collection before three runs and report the median peak observed after each major Host construction/serialization or Client parse/validation/preparation/fold stage, relative to the same initialized benchmark state. They do not measure process RSS and can miss transients within a sampled stage. The manual performance inventory does not run in CI and carries no machine-dependent timing or memory assertions; structural assertions pin the fixture cardinalities, compact input count, and identical final state from its two-consumer Assistant fold fixture. +The opt-in `packages/client/runtime/tests/history-transport.perf.client.ts` benchmark constructs the same logical-event, ordinary-event, and delta-run cardinalities from synthetic content. `DSH_SNAPSHOT=replay pnpm exec vitest run --config vitest.web.perf.config.ts packages/client/runtime/tests/history-transport.perf.client.ts` reports wire sizes, Host/client timing, and sampled additional V8 heap peaks under `HISTORY_TRANSPORT_PERF_RESULT`; a second inventory reports the median of five exact decodes for 10,000-, 20,000-, and 40,000-member whitespace-prefix runs under `HISTORY_WHITESPACE_PREFIX_PERF_RESULT`. Heap measurements force garbage collection before three runs and report the median peak observed after each major Host construction/serialization or Client parse/validation/decoding/fold stage, relative to the same initialized benchmark state. They do not measure process RSS and can miss transients within a sampled stage. The manual performance inventory does not run in CI and carries no machine-dependent timing or memory assertions; structural assertions pin the fixture cardinalities, exact decoded event count, and identical final state—including delta count and last-delta sequence—from its two-consumer Assistant fold fixture. ## Alternatives considered **Discard completed-step chunks on the Host.** This lowers logical event count but makes transport semantics depend on the current transcript policy, removes exact evidence from all consumers, and still sends every retained incomplete-step token as a separate envelope. The measured packed response is smaller while remaining lossless. -**Send packed rows and expand every member in the browser.** This removes repeated JSON envelopes on the network but recreates hundreds of thousands of event objects, fold matches, and temporary arrays before producing the same accumulated UI state. +**Coalesce a packed run before registered Definitions see it.** This reduces browser event objects and fold calls, but an open `ConversationNodeDefinition` may count deltas, inspect their individual `seq` or timestamps, or derive state from fragment boundaries. Equal accumulated text does not make those state machines equivalent, so the transport cannot change their replay input cardinality. -**Rely on HTTP content encoding.** gzip and Brotli reduce bytes on the network but do not remove repeated JSON parsing, validation, allocation, and fold work. Packed rows remain substantially smaller after both encodings in the measured sample. +**Rely on HTTP content encoding.** gzip and Brotli reduce bytes on the network but do not remove repeated JSON parsing and validation. Packed rows remain substantially smaller after both encodings in the measured sample, while exact browser replay retains the required allocation and fold work. **Page directly over physical persistence rows.** This could also avoid logical expansion in a cold Host read, but page cuts depend on append-origin messages and replacement provenance rather than backend row boundaries. The current decision keeps the API independent of JSONL, SQLite, and future persistence layouts. @@ -48,8 +48,8 @@ The opt-in `packages/client/runtime/tests/history-transport.perf.client.ts` benc ## Consequences -History responses preserve every logical event while reducing wire bytes, client JSON objects, and ordinary conversation-fold work for long delta runs. Pagination and reconnect logic use explicit raw interval watermarks, so compact browser inputs do not create false gaps. Existing consumers must switch from `events` to the `HistoryRecord` union and choose compact UI folding or exact decoding explicitly. +History responses preserve every logical event while reducing wire bytes, Host response serialization and heap, and browser JSON parsing and validation for long delta runs. Pagination and reconnect logic use explicit raw interval watermarks, so packed transport records do not create false gaps. Existing direct consumers must switch from `events` to the `HistoryRecord` union and decode packed rows before event-level processing. Cold persisted history is still decoded into the complete logical `SessionEvent[]` before the Host selects and repacks a page. This decision therefore improves transport and browser work, not the Host's cold-read decode memory. Eliminating that expansion requires a persistence-neutral message-boundary index or a separate streaming page reader and remains a distinct optimization. -Historical replay no longer reproduces one UI update per original token. The browser already installs history in a batch rather than animating past tokens; content and timing boundaries used by the settled view remain preserved. Live streaming behavior is unchanged. +Browser history replay still allocates and folds one event per original token, so this decision does not reduce Definition match/update count or settled-history heap and may add a small decode-time peak while packed records and expanded events coexist. History installs as one batch rather than animating old tokens; live streaming behavior is unchanged. diff --git a/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.zh.md b/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.zh.md index 9ef5ea8cae..883dd72346 100644 --- a/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.zh.md +++ b/.agents/notes/implemented/architecture/2026-08-15-packed-session-history-transport.zh.md @@ -6,7 +6,7 @@ Status: implemented ## 问题 -`session.history` 与 `subagent.history` 会向远程客户端提供一段有界的逻辑会话事件区间。提供方流可能在一个未完成尾部中产生数十万个 token 大小的 `assistant/chunk` 事件。先展开每条持久化行，再序列化每个逻辑事件，会在协议中重复相同信封；浏览器再次展开每条记录，则会在 conversation 折叠拼接文本之前重建同样的对象扩散。 +`session.history` 与 `subagent.history` 会向远程客户端提供一段有界的逻辑会话事件区间。提供方流可能在一个未完成尾部中产生数十万个 token 大小的 `assistant/chunk` 事件。先展开每条持久化行，再序列化每个逻辑事件，会在协议中重复相同信封，并让浏览器在 conversation 回放开始前解析和校验这些重复内容。 传输必须保持无损。会话序号是分页与重连证据；精确 token 边界对诊断和非 UI API 消费方仍然有用；实时流式传输、持久导出、回放与模型历史派生仍然需要规范事件流。如果由服务端 transcript 投影丢弃已完成步骤的分片，API 证据就会取决于一项 UI 策略。 @@ -14,9 +14,9 @@ Status: implemented 历史方法返回 `records: HistoryRecord[]`，以及包含端 `fromSeq` 与不包含端 `toSeq` 水位。普通记录携带 `{event, view?}`。连续且属于同一块的 Assistant delta 事件使用[打包 JSONL 决策](2026-07-26-packed-chunk-rows-by-default.zh.md)中的共享无损编解码器，携带 `{chunks: ChunkRow}`。系统先从逻辑事件中选择页面，再执行打包，因此按消息对齐的分页不依赖物理持久化布局。 -协议 schema 校验每一行，拒绝不安全的重建，并要求记录无间隙、无重叠地精确覆盖 `[fromSeq, toSeq)`。更早页面拼接、重连修复与实时事件去重以水位为准，而不以浏览器折叠输入的数量或可见 seq 邻接关系为准。`session.history` 与 `subagent.history` 共用相同的响应 schema。 +协议 schema 校验每一行，拒绝不安全的重建，并要求记录无间隙、无重叠地精确覆盖 `[fromSeq, toSeq)`。更早页面拼接、重连修复与实时事件去重以水位为准，而不以传输记录数量为准。`session.history` 与 `subagent.history` 共用相同的响应 schema。 -普通浏览器 UI 不会把打包行解码成每个 token 一个对象。它会把一行合并成最多两个 `assistant/chunk` 输入，同时保留累计内容、首个非空 token 时间戳，以及这两个边界不同时较晚出现的首个非空白可见时间戳。工具调用行保留调用身份、名称存在性、拼接后的参数片段与首 token 时间。其他 API 消费方在需要精确 token 边界时可以调用 `decodeStorageRecord()`。 +浏览器会先调用共享的 `decodeStorageRecord()` 编解码器，再把历史交给 `ConversationNodeAssembler`。每个打包成员都会还原为完全一致的原始 `assistant/chunk` 事件，包括 `seq`、时间戳、chunk 类型、block 索引、文本或参数片段、调用身份，以及可选名称是否存在。因此，已注册的 `ConversationNodeDefinition` 会对每个历史 delta 收到一次 `match()` 调用，并按实时事件所具有的同一 start／update 顺序折叠已接受的 match。打包只改变传输编码，不改变公共 Definition 的回放语义。 实时 `session/event` 帧仍是单个事件。会话持久化、原始导出、回放、模型历史派生与规范内存日志均不改变。 @@ -32,15 +32,15 @@ Status: implemented 与原始逻辑事件相比，打包使未压缩 JSON 减少 90.8%；与有损的已完成步骤投影候选相比减少 83.4%。Brotli 输出相对原始形式减少 73.2%，相对该投影候选减少 44.8%。这些数字描述该样本，并非协议保证；收益随 delta run 的长度与规律性变化。 -可选运行的 `packages/client/runtime/tests/history-transport.perf.client.ts` benchmark 使用合成内容构造相同的逻辑事件数、普通事件数与 delta run 数。`DSH_SNAPSHOT=replay pnpm exec vitest run --config vitest.web.perf.config.ts packages/client/runtime/tests/history-transport.perf.client.ts` 会在 `HISTORY_TRANSPORT_PERF_RESULT` 下报告协议体积、Host／client 计时与采样的额外 V8 堆峰值。堆测量会在三次运行前强制执行垃圾回收，并相对于相同的已初始化 benchmark 状态，报告 Host 构造／序列化或 Client 解析／校验／准备／折叠各主要阶段之后所观察峰值的中位数。该指标不测量进程 RSS，也可能遗漏单个采样阶段内部的瞬态峰值。CI 不执行这组手动性能用例，其中也没有依赖机器性能的耗时或内存断言；结构断言固定 fixture 的事件规模、紧凑输入数，以及双消费方 Assistant 折叠 fixture 的一致最终状态。 +可选运行的 `packages/client/runtime/tests/history-transport.perf.client.ts` benchmark 使用合成内容构造相同的逻辑事件数、普通事件数与 delta run 数。`DSH_SNAPSHOT=replay pnpm exec vitest run --config vitest.web.perf.config.ts packages/client/runtime/tests/history-transport.perf.client.ts` 会在 `HISTORY_TRANSPORT_PERF_RESULT` 下报告协议体积、Host／client 计时与采样的额外 V8 堆峰值；第二组清单会在 `HISTORY_WHITESPACE_PREFIX_PERF_RESULT` 下报告 10,000、20,000 与 40,000 个成员的空白前缀 run 各五次精确解码的中位数。堆测量会在三次运行前强制执行垃圾回收，并相对于相同的已初始化 benchmark 状态，报告 Host 构造／序列化或 Client 解析／校验／解码／折叠各主要阶段之后所观察峰值的中位数。该指标不测量进程 RSS，也可能遗漏单个采样阶段内部的瞬态峰值。CI 不执行这组手动性能用例，其中也没有依赖机器性能的耗时或内存断言；结构断言固定 fixture 的事件规模、精确解码事件数，以及双消费方 Assistant 折叠 fixture 的一致最终状态，包括 delta 数量与末个 delta 序号。 ## 曾考虑的替代方案 **在 Host 丢弃已完成步骤的分片。** 这会减少逻辑事件数，但会让传输语义取决于当前 transcript 策略，从所有消费方移除精确证据，同时仍把保留的未完成步骤 token 逐个装入信封。实测打包响应在保持无损的同时更小。 -**发送打包行，再在浏览器展开每个成员。** 这会移除网络上的重复 JSON 信封，却会在生成相同累计 UI 状态之前，重建数十万个事件对象、折叠匹配与临时数组。 +**在已注册 Definition 看到打包 run 前先进行合并。** 这会减少浏览器事件对象与折叠调用，但开放的 `ConversationNodeDefinition` 可能统计 delta、检查各自的 `seq` 或时间戳，或者根据片段边界派生状态。累计文本相同不代表这些状态机等价，因此传输不能改变其回放输入数量。 -**只依赖 HTTP 内容编码。** gzip 与 Brotli 会减少网络字节，但不会移除重复的 JSON 解析、校验、分配与折叠工作。在实测样本中，打包行经过这两种编码后仍然显著更小。 +**只依赖 HTTP 内容编码。** gzip 与 Brotli 会减少网络字节，但不会移除重复的 JSON 解析与校验。在实测样本中，打包行经过这两种编码后仍然显著更小；精确浏览器回放则保留契约要求的分配与折叠工作。 **直接按物理持久化行分页。** 这还可以避免冷 Host 读取时的逻辑展开，但页面切分取决于追加来源消息与替换 provenance，而不是后端行边界。当前决策让 API 保持对 JSONL、SQLite 与未来持久化布局的独立性。 @@ -48,8 +48,8 @@ Status: implemented ## 后果 -历史响应保留每个逻辑事件，同时减少长 delta run 的协议字节、客户端 JSON 对象与普通 conversation 折叠工作。分页与重连逻辑使用显式原始区间水位，因此紧凑浏览器输入不会产生伪间隙。现有消费方必须从 `events` 切换到 `HistoryRecord` 联合，并明确选择紧凑 UI 折叠或精确解码。 +历史响应保留每个逻辑事件，同时减少长 delta run 的协议字节、Host 响应序列化与堆占用，以及浏览器 JSON 解析与校验工作。分页与重连逻辑使用显式原始区间水位，因此打包传输记录不会产生伪间隙。现有直接消费方必须从 `events` 切换到 `HistoryRecord` 联合，并在逐事件处理前解码打包行。 冷持久历史仍会先解码成完整的逻辑 `SessionEvent[]`，Host 再选择页面并重新打包。因此，本决策改善的是传输与浏览器工作，不是 Host 冷读取的解码内存。消除该展开需要提供方无关的消息边界索引或单独的流式页面读取器，属于另一项优化。 -历史回放不再为每个原始 token 重现一次 UI 更新。浏览器本就会批量安装历史，而不会为过去的 token 播放动画；settled view 使用的内容与计时边界仍会保留。实时流式行为不变。 +浏览器历史回放仍会为每个原始 token 分配和折叠一个事件，因此本决策不会减少 Definition 的 match／update 次数或 settled history 堆占用；打包记录与展开事件同时存在时，还可能增加少量解码期峰值。历史仍作为一个批次安装，而不会为旧 token 播放动画；实时流式行为不变。 diff --git a/packages/api/session-controller/src/client/sessions/history-records.ts b/packages/api/session-controller/src/client/sessions/history-records.ts index ff6b77eb00..2d8226f9f3 100644 --- a/packages/api/session-controller/src/client/sessions/history-records.ts +++ b/packages/api/session-controller/src/client/sessions/history-records.ts @@ -1,99 +1,18 @@ -/** Compact client folding for packed Assistant delta runs in history responses. */ +/** Lossless client decoding for packed Assistant delta runs in history responses. */ -import type { ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows' -import type { SessionEvent } from '@deepseek-ai/dsh-session/types' +import { decodeStorageRecord } from '@deepseek-ai/dsh-session/chunk-rows' import type { SessionEventEntry, SessionHistoryRecord, } from '../../types.ts' -/** Resolve one packed member's original timestamp from the row's delta gaps. */ -function memberTime(row: ChunkRow, index: number): number { - let time = row.time0 - for (let cursor = 0; cursor < index; cursor++) time += row.data.dt[cursor] as number - return time -} - -/** Build one coalesced text or reasoning event from a contiguous member slice. */ -function textEvent( - row: Extract, - start: number, - text: string, -): SessionEvent<'assistant/chunk'> { - return { - type: 'assistant/chunk', - seq: row.seq0 + start, - time: memberTime(row, start), - data: { - turn: row.data.turn, - step: row.data.step, - chunk: row.type === 'text-chunks' - ? { type: 'text-delta', index: row.data.index, text } - : { type: 'reasoning-delta', index: row.data.index, text }, - }, - } -} - /** - * Coalesce one packed run into the smallest event set that preserves the - * conversation fold's accumulated content, first-token time, and first - * non-whitespace visibility boundary. Exact token boundaries remain available - * in the wire row to consumers that explicitly decode it. - * @param row - one validated packed history record. - * @returns At most two Assistant chunk events for the ordinary UI fold. - */ -export function coalesceHistoryChunkRun(row: ChunkRow): SessionEvent<'assistant/chunk'>[] { - if (row.type === 'tool-call-chunks') { - const firstToken = row.data.name === undefined - ? row.data.args.findIndex(fragment => fragment !== '') - : 0 - const start = firstToken < 0 ? 0 : firstToken - return [{ - type: 'assistant/chunk', - seq: row.seq0 + start, - time: memberTime(row, start), - data: { - turn: row.data.turn, - step: row.data.step, - chunk: { - type: 'tool-call-delta', - index: row.data.index, - id: row.data.id, - ...row.data.name === undefined ? {} : { name: row.data.name }, - argumentsDelta: row.data.args.join(''), - }, - }, - }] - } - - const texts = row.data.texts - const firstToken = texts.findIndex(text => text !== '') - const tokenStart = firstToken < 0 ? 0 : firstToken - let visibleStart = -1 - let accumulated = '' - for (let index = 0; index < texts.length; index++) { - accumulated += texts[index] as string - if (accumulated.trim() !== '') { - visibleStart = index - break - } - } - if (visibleStart > tokenStart) { - return [ - textEvent(row, tokenStart, texts.slice(0, visibleStart).join('')), - textEvent(row, visibleStart, texts.slice(visibleStart).join('')), - ] - } - return [textEvent(row, tokenStart, texts.join(''))] -} - -/** - * Convert history wire records into compact event inputs for the ordinary UI. + * Convert history wire records into exact event inputs for Conversation Definitions. * @param records - validated lossless history transport records. - * @returns Ordinary entries unchanged and packed runs coalesced for folding. + * @returns Ordinary entries unchanged and packed runs expanded member-for-member. */ export function historyEntries(records: readonly SessionHistoryRecord[]): SessionEventEntry[] { return records.flatMap(record => 'event' in record ? [record] - : coalesceHistoryChunkRun(record.chunks).map(event => ({ event }))) + : decodeStorageRecord(record.chunks).map(event => ({ event }))) } diff --git a/packages/api/session-controller/tests/history-records.client.spec.ts b/packages/api/session-controller/tests/history-records.client.spec.ts index a6d792e0f6..5dd0160743 100644 --- a/packages/api/session-controller/tests/history-records.client.spec.ts +++ b/packages/api/session-controller/tests/history-records.client.spec.ts @@ -1,99 +1,15 @@ -/** Packed history record folding without token-by-token browser expansion. */ +/** Packed history records decode to the exact Session event stream. */ import { describe, expect, it } from 'vitest' import { CallId } from '@deepseek-ai/dsh-llm/brand' -import type { ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows' import type { SessionHistoryRecord } from '../src/types.ts' -import { coalesceHistoryChunkRun, historyEntries } from '../src/client/sessions/history-records.ts' - -describe('coalesceHistoryChunkRun', () => { - it('preserves first-token and first-visible boundaries with at most two text events', () => { - const row: ChunkRow = { - type: 'text-chunks', - seq0: 10, - time0: 100, - data: { - turn: 2, - step: 3, - index: 0, - dt: [1, 2, 3, 4], - texts: ['', ' ', '', 'hello', ' world'], - }, - } - - const events = coalesceHistoryChunkRun(row) - expect(events).toHaveLength(2) - expect(events.map(event => ({ seq: event.seq, time: event.time, text: event.data.chunk.type === 'text-delta' ? event.data.chunk.text : '' }))) - .toEqual([ - { seq: 11, time: 101, text: ' ' }, - { seq: 13, time: 106, text: 'hello world' }, - ]) - expect(events[0]?.time).toBe(101) - expect(events.find(event => event.data.chunk.type === 'text-delta' && event.data.chunk.text.trim() !== '')?.time) - .toBe(106) - }) - - it('joins visible reasoning members into one event at the first non-empty member', () => { - const row: ChunkRow = { - type: 'reasoning-chunks', - seq0: 4, - time0: 50, - data: { turn: 1, step: 1, index: 2, dt: [5, 7], texts: ['', 'a', 'b'] }, - } - const [event] = coalesceHistoryChunkRun(row) - expect(event).toMatchObject({ - seq: 5, - time: 55, - data: { chunk: { type: 'reasoning-delta', index: 2, text: 'ab' } }, - }) - }) - - it('joins tool arguments while retaining name presence and first-token time', () => { - const named: ChunkRow = { - type: 'tool-call-chunks', - seq0: 20, - time0: 200, - data: { - turn: 2, - step: 4, - index: 1, - id: CallId('call-1'), - name: 'write', - dt: [2, 3], - args: ['', '{"x":', '1}'], - }, - } - expect(coalesceHistoryChunkRun(named)).toMatchObject([{ - seq: 20, - time: 200, - data: { chunk: { type: 'tool-call-delta', name: 'write', argumentsDelta: '{"x":1}' } }, - }]) - - const unnamed: ChunkRow = { - type: 'tool-call-chunks', - seq0: 20, - time0: 200, - data: { - turn: 2, - step: 4, - index: 1, - id: CallId('call-1'), - dt: [2, 3], - args: ['', '', 'x'], - }, - } - const [event] = coalesceHistoryChunkRun(unnamed) - expect(event).toMatchObject({ seq: 22, time: 205, data: { chunk: { argumentsDelta: 'x' } } }) - expect(Object.hasOwn(event?.data.chunk ?? {}, 'name')).toBe(false) - }) -}) +import { historyEntries } from '../src/client/sessions/history-records.ts' describe('historyEntries', () => { - it('keeps ordinary entries and views while folding a packed run without expansion', () => { + it('keeps ordinary entries and expands every packed text member with its exact boundary', () => { const ordinary = { event: { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }, - view: { for: 'call', view: { card: 'generic' } }, - } as unknown as SessionHistoryRecord + } as SessionHistoryRecord const packed: SessionHistoryRecord = { chunks: { type: 'text-chunks', @@ -102,9 +18,48 @@ describe('historyEntries', () => { data: { turn: 1, step: 1, index: 0, dt: [1, 1, 1], texts: ['a', 'b', 'c', 'd'] }, }, } + const entries = historyEntries([ordinary, packed]) - expect(entries).toHaveLength(2) + + expect(entries).toHaveLength(5) expect(entries[0]).toBe(ordinary) - expect(entries[1]?.event).toMatchObject({ seq: 1, data: { chunk: { text: 'abcd' } } }) + expect(entries.slice(1).map(entry => ({ + seq: entry.event.seq, + time: entry.event.time, + chunk: entry.event.type === 'assistant/chunk' ? entry.event.data.chunk : undefined, + }))).toEqual([ + { seq: 1, time: 2, chunk: { type: 'text-delta', index: 0, text: 'a' } }, + { seq: 2, time: 3, chunk: { type: 'text-delta', index: 0, text: 'b' } }, + { seq: 3, time: 4, chunk: { type: 'text-delta', index: 0, text: 'c' } }, + { seq: 4, time: 5, chunk: { type: 'text-delta', index: 0, text: 'd' } }, + ]) + }) + + it('preserves every tool-call fragment and optional-name presence', () => { + const packed: SessionHistoryRecord = { + chunks: { + type: 'tool-call-chunks', + seq0: 20, + time0: 200, + data: { + turn: 2, + step: 4, + index: 1, + id: CallId('call-1'), + dt: [2, 3], + args: ['', '{"x":', '1}'], + }, + }, + } + + const events = historyEntries([packed]).map(entry => entry.event) + + expect(events).toMatchObject([ + { seq: 20, time: 200, data: { chunk: { argumentsDelta: '' } } }, + { seq: 21, time: 202, data: { chunk: { argumentsDelta: '{"x":' } } }, + { seq: 22, time: 205, data: { chunk: { argumentsDelta: '1}' } } }, + ]) + expect(events.every(event => event.type === 'assistant/chunk' + && !Object.hasOwn(event.data.chunk, 'name'))).toBe(true) }) }) diff --git a/packages/api/session-controller/tests/history-transport.perf.client.ts b/packages/api/session-controller/tests/history-transport.perf.client.ts index c6546f414f..6f7d34e103 100644 --- a/packages/api/session-controller/tests/history-transport.perf.client.ts +++ b/packages/api/session-controller/tests/history-transport.perf.client.ts @@ -1,4 +1,4 @@ -/** Opt-in synthetic benchmark for packed session-history transport and folding. */ +/** Opt-in synthetic benchmark for packed session-history transport and exact replay. */ import { createHash } from 'node:crypto' import { performance } from 'node:perf_hooks' @@ -40,6 +40,8 @@ interface HeapPeaks { interface FoldState { readonly blocks: readonly string[] + readonly deltaCount: number + readonly lastDeltaSeq?: number readonly firstTokenTime?: number readonly firstVisibleSeq?: number readonly firstVisibleTime?: number @@ -176,7 +178,7 @@ function foldDefinition(kind: string, target: string): ConversationNodeDefinitio } return null }, - start: () => ({ blocks: [] }), + start: () => ({ blocks: [], deltaCount: 0 }), update: (context, match) => { if (match.event.type !== 'assistant/chunk' || match.event.data.chunk.type !== 'reasoning-delta') { return context.state @@ -188,6 +190,8 @@ function foldDefinition(kind: string, target: string): ConversationNodeDefinitio return { ...context.state, blocks, + deltaCount: context.state.deltaCount + 1, + lastDeltaSeq: match.event.seq, ...context.state.firstTokenTime === undefined ? { firstTokenTime: match.event.time } : {}, ...visible && context.state.firstVisibleSeq === undefined ? { firstVisibleSeq: match.event.seq, firstVisibleTime: match.event.time } @@ -242,7 +246,7 @@ function digest(value: unknown): string { return createHash('sha256').update(JSON.stringify(value)).digest('hex') } -it('reports packed history transport and compact fold costs', () => { +it('reports packed history transport and exact replay costs', () => { const fixture = timed(buildEvents) assemble(conversationInputs(fixture.value.slice(0, 1_000).map(event => ({ event })))) @@ -332,7 +336,7 @@ it('reports packed history transport and compact fold costs', () => { expect(fixture.value.filter(event => event.type !== 'assistant/chunk')).toHaveLength(ORDINARY_EVENTS) expect(packedRows).toHaveLength(DELTA_RUNS) expect(packed.value).toHaveLength(696) - expect(packedPreparation.value).toHaveLength(696) + expect(packedPreparation.value).toHaveLength(LOGICAL_EVENTS) expect(digest(packedFold.value)).toBe(digest(rawFold.value)) expect(packedClientHeap.value).toBe(rawClientHeap.value) expect(rawHostHeap.value).toBe(rawBytes) @@ -351,7 +355,7 @@ it('reports packed history transport and compact fold costs', () => { deltaEvents: DELTA_EVENTS, deltaRuns: packedRows.length, packedRecords: packed.value.length, - compactFoldInputs: packedPreparation.value.length, + decodedEvents: packedPreparation.value.length, }, bytes: { rawJson: rawBytes, @@ -406,3 +410,44 @@ it('reports packed history transport and compact fold costs', () => { }, })}\n`) }, 600_000) + +it('reports exact decoding cost for long whitespace-prefix runs', () => { + historyEntries([{ + chunks: { + type: 'reasoning-chunks', + seq0: 0, + time0: TIME_ZERO, + data: { turn: 1, step: 1, index: 0, dt: [], texts: ['x'] }, + }, + }]) + const results = [10_000, 20_000, 40_000].map((members) => { + const record: HistoryRecord = { + chunks: { + type: 'reasoning-chunks', + seq0: 0, + time0: TIME_ZERO, + data: { + turn: 1, + step: 1, + index: 0, + dt: Array.from({ length: members - 1 }, () => 1), + texts: Array.from({ length: members }, (_, index) => index === members - 1 ? 'x' : ' '), + }, + }, + } + const decoded = historyEntries([record]) + const samplesMs = Array.from({ length: 5 }, () => timed(() => historyEntries([record])).ms) + expect(decoded).toHaveLength(members) + expect(decoded.at(-1)?.event).toMatchObject({ + seq: members - 1, + time: TIME_ZERO + members - 1, + data: { chunk: { type: 'reasoning-delta', text: 'x' } }, + }) + return { + members, + medianMs: rounded(median(samplesMs)), + samplesMs: samplesMs.map(rounded), + } + }) + process.stdout.write(`HISTORY_WHITESPACE_PREFIX_PERF_RESULT ${JSON.stringify(results)}\n`) +})