mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-08-29 04:26:38 +00:00
perf(history): carry packed assistant chunks
This commit is contained in:
kingwl
+2
-2
@@ -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-07-26-packed-chunk-rows-by-default.md
|
||||
2026-07-26-packed-chunk-rows-by-default.md: 141c9a32a07b5cb4885a21b419df30d45dc1061b
|
||||
2026-07-26-packed-chunk-rows-by-default.zh.md: 30088b8640efa783380a3fe083ed82efdad9e5c8
|
||||
2026-07-26-packed-chunk-rows-by-default.md: 14da6b3cbe650e80118e7c960c96bf618acd1e48
|
||||
2026-07-26-packed-chunk-rows-by-default.zh.md: f62a8e52a67adc960ac3150552594b4f061e6205
|
||||
|
||||
@@ -18,7 +18,7 @@ Reading is unconditional and layout-blind. Packed, unpacked, and mixed files loa
|
||||
|
||||
### Logical events and physical rows
|
||||
|
||||
Packing stays at the `dsh-session` storage seam through `packChunkRuns()` and `decodeStorageRecord()`. The encoder recognizes exact delta-event shapes, preserves unrecognized events verbatim, and packs only runs of at least three. A packed row is storage vocabulary, not a `SessionEventMap` member: it never enters `Session.events` or fires `session/event`.
|
||||
The JSONL packing path stays at the `dsh-session` storage seam through `packChunkRuns()` and `decodeStorageRecord()`. The encoder recognizes exact delta-event shapes, preserves unrecognized events verbatim, and packs only runs of at least three. A packed row is encoding vocabulary, not a `SessionEventMap` member: it never enters `Session.events` or fires `session/event`. The [packed session-history transport decision](2026-08-15-packed-session-history-transport.md) reuses this vocabulary for a bounded lossless wire interval without changing those event semantics.
|
||||
|
||||
The JSONL backend packs each durable append batch. Raw `compression: 'none'` and default Zstandard framing carry the same logical storage records; selecting raw mode for reviewable fixtures does not disable packing. Repository replay readers and normalizers decode the shared row format instead of maintaining snapshot-specific codecs.
|
||||
|
||||
|
||||
+1
-1
@@ -18,7 +18,7 @@ JSONL 存储 seam 可以在不改变逻辑日志的情况下减少这部分封
|
||||
|
||||
### 逻辑事件与物理行
|
||||
|
||||
打包保留在 `dsh-session` 的存储 seam，并通过 `packChunkRuns()` 和 `decodeStorageRecord()` 实现。编码器识别精确的增量事件形态，原样保留无法识别的事件，并且只打包至少包含 3 个事件的连续段。打包行属于存储词汇，不是 `SessionEventMap` 成员：它绝不会进入 `Session.events`，也不会触发 `session/event`。
|
||||
JSONL 打包路径保留在 `dsh-session` 的存储 seam，并通过 `packChunkRuns()` 和 `decodeStorageRecord()` 实现。编码器识别精确的增量事件形态，原样保留无法识别的事件，并且只打包至少包含 3 个事件的连续段。打包行属于编码词汇，不是 `SessionEventMap` 成员：它绝不会进入 `Session.events`，也不会触发 `session/event`。[打包会话历史传输决策](2026-08-15-packed-session-history-transport.zh.md)会为有界的无损协议区间复用该词汇，而不改变这些事件语义。
|
||||
|
||||
JSONL 后端会打包每个持久追加批次。原始模式 `compression: 'none'` 与默认 Zstandard 帧承载相同的逻辑存储记录；为使 fixture 便于评审而选择原始模式，不会禁用打包。仓库中的回放读取器和规范化器会解码共享行格式，而不维护快照专用编解码器。
|
||||
|
||||
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# 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: f005bdc3e2815505d64996ad1e003d8e601e3ec8
|
||||
2026-08-15-packed-session-history-transport.zh.md: a8260c2a1d1c38512a2a4cad3d5c2bc67280923d
|
||||
@@ -0,0 +1,53 @@
|
||||
# Agent Note: Carry packed chunk rows through session history
|
||||
|
||||
Status: implemented
|
||||
|
||||
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.
|
||||
|
||||
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.
|
||||
|
||||
## Decision
|
||||
|
||||
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 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.
|
||||
|
||||
Live `session/event` frames remain individual events. Session persistence, raw export, replay, model-history derivation, and the canonical in-memory log are unchanged.
|
||||
|
||||
## Measured result
|
||||
|
||||
A production-sized private session sample was measured without retaining or committing its content. Its tail page contained 416,756 logical events. The lossless packed response used 696 top-level records, including 116 packed rows.
|
||||
|
||||
| Representation | Top-level records | JSON bytes | gzip bytes | Brotli bytes |
|
||||
| --- | ---: | ---: | ---: | ---: |
|
||||
| Raw logical events | 416,756 | 69,433,638 | 4,190,226 | 1,972,998 |
|
||||
| Completed-step projection candidate | 228,129 | 38,427,209 | 2,324,688 | 957,350 |
|
||||
| Lossless packed history | 696 | 6,362,724 | 1,154,206 | 528,145 |
|
||||
|
||||
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.
|
||||
|
||||
## 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.
|
||||
|
||||
**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.
|
||||
|
||||
**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.
|
||||
|
||||
**Return only assembled Assistant snapshots.** The [assembled-messages-only rejection](../../rejected/simplification/2026-06-20-assembled-assistant-messages-only.md) remains applicable: event families outside finalized messages carry user-visible and diagnostic state, and incomplete steps need their actual accumulated chunks.
|
||||
|
||||
## 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.
|
||||
|
||||
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.
|
||||
+53
@@ -0,0 +1,53 @@
|
||||
# Agent Note: 在会话历史中传输打包分片行
|
||||
|
||||
Status: implemented
|
||||
|
||||
[English](2026-08-15-packed-session-history-transport.md) | 中文
|
||||
|
||||
## 问题
|
||||
|
||||
`session.history` 与 `subagent.history` 会向远程客户端提供一段有界的逻辑会话事件区间。提供方流可能在一个未完成尾部中产生数十万个 token 大小的 `assistant/chunk` 事件。先展开每条持久化行，再序列化每个逻辑事件，会在协议中重复相同信封；浏览器再次展开每条记录，则会在 conversation 折叠拼接文本之前重建同样的对象扩散。
|
||||
|
||||
传输必须保持无损。会话序号是分页与重连证据；精确 token 边界对诊断和非 UI API 消费方仍然有用；实时流式传输、持久导出、回放与模型历史派生仍然需要规范事件流。如果由服务端 transcript 投影丢弃已完成步骤的分片，API 证据就会取决于一项 UI 策略。
|
||||
|
||||
## 决策
|
||||
|
||||
历史方法返回 `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。
|
||||
|
||||
普通浏览器 UI 不会把打包行解码成每个 token 一个对象。它会把一行合并成最多两个 `assistant/chunk` 输入，同时保留累计内容、首个非空 token 时间戳，以及这两个边界不同时较晚出现的首个非空白可见时间戳。工具调用行保留调用身份、名称存在性、拼接后的参数片段与首 token 时间。其他 API 消费方在需要精确 token 边界时可以调用 `decodeStorageRecord()`。
|
||||
|
||||
实时 `session/event` 帧仍是单个事件。会话持久化、原始导出、回放、模型历史派生与规范内存日志均不改变。
|
||||
|
||||
## 测量结果
|
||||
|
||||
测量使用了一份生产规模的私有会话样本，未保留或签入其内容。其尾页包含 416,756 个逻辑事件。无损打包响应使用 696 条顶层记录，其中包含 116 条打包行。
|
||||
|
||||
| 表示 | 顶层记录数 | JSON 字节 | gzip 字节 | Brotli 字节 |
|
||||
| --- | ---: | ---: | ---: | ---: |
|
||||
| 原始逻辑事件 | 416,756 | 69,433,638 | 4,190,226 | 1,972,998 |
|
||||
| 已完成步骤投影候选 | 228,129 | 38,427,209 | 2,324,688 | 957,350 |
|
||||
| 无损打包历史 | 696 | 6,362,724 | 1,154,206 | 528,145 |
|
||||
|
||||
与原始逻辑事件相比，打包使未压缩 JSON 减少 90.8%；与有损的已完成步骤投影候选相比减少 83.4%。Brotli 输出相对原始形式减少 73.2%，相对该投影候选减少 44.8%。这些数字描述该样本，并非协议保证；收益随 delta run 的长度与规律性变化。
|
||||
|
||||
## 曾考虑的替代方案
|
||||
|
||||
**在 Host 丢弃已完成步骤的分片。** 这会减少逻辑事件数，但会让传输语义取决于当前 transcript 策略，从所有消费方移除精确证据，同时仍把保留的未完成步骤 token 逐个装入信封。实测打包响应在保持无损的同时更小。
|
||||
|
||||
**发送打包行，再在浏览器展开每个成员。** 这会移除网络上的重复 JSON 信封，却会在生成相同累计 UI 状态之前，重建数十万个事件对象、折叠匹配与临时数组。
|
||||
|
||||
**只依赖 HTTP 内容编码。** gzip 与 Brotli 会减少网络字节，但不会移除重复的 JSON 解析、校验、分配与折叠工作。在实测样本中，打包行经过这两种编码后仍然显著更小。
|
||||
|
||||
**直接按物理持久化行分页。** 这还可以避免冷 Host 读取时的逻辑展开，但页面切分取决于追加来源消息与替换 provenance，而不是后端行边界。当前决策让 API 保持对 JSONL、SQLite 与未来持久化布局的独立性。
|
||||
|
||||
**只返回组装后的 Assistant 快照。** [仅保留组装消息的否决记录](../../rejected/simplification/2026-06-20-assembled-assistant-messages-only.zh.md)仍然适用：final message 之外的事件族承载用户可见状态与诊断状态，未完成步骤也需要其实际累计分片。
|
||||
|
||||
## 后果
|
||||
|
||||
历史响应保留每个逻辑事件，同时减少长 delta run 的协议字节、客户端 JSON 对象与普通 conversation 折叠工作。分页与重连逻辑使用显式原始区间水位，因此紧凑浏览器输入不会产生伪间隙。现有消费方必须从 `events` 切换到 `HistoryRecord` 联合，并明确选择紧凑 UI 折叠或精确解码。
|
||||
|
||||
冷持久历史仍会先解码成完整的逻辑 `SessionEvent[]`，Host 再选择页面并重新打包。因此，本决策改善的是传输与浏览器工作，不是 Host 冷读取的解码内存。消除该展开需要提供方无关的消息边界索引或单独的流式页面读取器，属于另一项优化。
|
||||
|
||||
历史回放不再为每个原始 token 重现一次 UI 更新。浏览器本就会批量安装历史，而不会为过去的 token 播放动画；settled view 使用的内容与计时边界仍会保留。实时流式行为不变。
|
||||
@@ -180,17 +180,23 @@ async function sessionCursor(baseUrl: string, sessionId: string): Promise<number
|
||||
}
|
||||
|
||||
interface HistoryPage {
|
||||
events: { event: { type: string; data: unknown } }[]
|
||||
records: ({ event: HistoryEvent } | { chunks: unknown })[]
|
||||
hasMore: boolean
|
||||
}
|
||||
|
||||
interface HistoryEvent {
|
||||
type: string
|
||||
data: unknown
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null
|
||||
}
|
||||
|
||||
function providerTitle(page: HistoryPage): string | undefined {
|
||||
for (let index = page.events.length - 1; index >= 0; index--) {
|
||||
const event = page.events[index]!.event
|
||||
const events = page.records.flatMap(record => 'event' in record ? [record.event] : [])
|
||||
for (let index = events.length - 1; index >= 0; index--) {
|
||||
const event = events[index] as HistoryEvent
|
||||
if (event.type !== 'session/title' || !isRecord(event.data)) continue
|
||||
const source = event.data.source
|
||||
if (typeof event.data.title === 'string' && isRecord(source) && source.kind === 'provider') {
|
||||
@@ -201,7 +207,9 @@ function providerTitle(page: HistoryPage): string | undefined {
|
||||
}
|
||||
|
||||
function hasAssistantMarker(page: HistoryPage, marker: string): boolean {
|
||||
return page.events.some(({ event }) => {
|
||||
return page.records.some((record) => {
|
||||
if (!('event' in record)) return false
|
||||
const { event } = record
|
||||
if (event.type !== 'assistant/message' || !isRecord(event.data) || !isRecord(event.data.message)) return false
|
||||
const content = event.data.message.content
|
||||
if (!Array.isArray(content)) return false
|
||||
@@ -556,16 +564,16 @@ describe('dsh web keyless CLI smoke', () => {
|
||||
return hasAssistantMarker(page, recoveredMarker)
|
||||
}, { timeout: 20_000 }).toBe(true)
|
||||
if (page === undefined) throw new Error('retry history was not observed')
|
||||
const retry = page.events.find(({ event }) => event.type === 'llm/retry')?.event
|
||||
const retry = page.records.find(record => 'event' in record && record.event.type === 'llm/retry')
|
||||
expect(mainAttempts).toBe(2)
|
||||
expect(retry?.data).toMatchObject({
|
||||
expect(retry !== undefined && 'event' in retry ? retry.event.data : undefined).toMatchObject({
|
||||
turn: 1,
|
||||
step: 1,
|
||||
retry: 1,
|
||||
maxRetries: 5,
|
||||
failure: { code: 'TRANSPORT' },
|
||||
})
|
||||
expect(JSON.stringify(page.events)).toContain('WEB_RETRY_DISCARDED')
|
||||
expect(JSON.stringify(page.records)).toContain('WEB_RETRY_DISCARDED')
|
||||
} finally {
|
||||
const closed = child.exitCode === null
|
||||
? new Promise<void>((resolveClose) => { child.once('close', () => { resolveClose() }) })
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
/** Compact client folding 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 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<ChunkRow, { type: 'text-chunks' | 'reasoning-chunks' }>,
|
||||
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.
|
||||
* @param records - validated lossless history transport records.
|
||||
* @returns Ordinary entries unchanged and packed runs coalesced for folding.
|
||||
*/
|
||||
export function historyEntries(records: readonly SessionHistoryRecord[]): SessionEventEntry[] {
|
||||
return records.flatMap(record => 'event' in record
|
||||
? [record]
|
||||
: coalesceHistoryChunkRun(record.chunks).map(event => ({ event })))
|
||||
}
|
||||
@@ -19,6 +19,7 @@ import type {
|
||||
SessionPageRequest,
|
||||
SessionProjectionBaseline,
|
||||
} from '../types.ts'
|
||||
import { historyEntries } from './sessions/history-records.ts'
|
||||
|
||||
export {
|
||||
SESSION_SEARCH_RESULT_LIMIT,
|
||||
@@ -117,7 +118,7 @@ export class SessionEventStream extends RemoteJournalStream<
|
||||
super(remote, {
|
||||
name: 'session event stream',
|
||||
emptyCursor: -1,
|
||||
entries: page => page.events,
|
||||
entries: page => historyEntries(page.records),
|
||||
hasMore: page => page.hasMore,
|
||||
cursor: entry => entry.event.seq,
|
||||
compare: (left, right) => left - right,
|
||||
@@ -144,7 +145,7 @@ export class SessionEventStream extends RemoteJournalStream<
|
||||
type: 'opened',
|
||||
cursor: frame.cursor,
|
||||
page: {
|
||||
events: frame.events,
|
||||
records: frame.records,
|
||||
hasMore: frame.hasMore,
|
||||
projections: frame.projections,
|
||||
},
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import type { Context } from '@deepseek-ai/cordis'
|
||||
import { isAppendSurfaceEvent } from '@deepseek-ai/dsh-session'
|
||||
import { isChunkRow, packChunkRuns } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
|
||||
import type {} from '@deepseek-ai/dsh-subagent'
|
||||
@@ -11,6 +12,7 @@ import type {
|
||||
SessionEventEntry,
|
||||
SessionFollowRequest,
|
||||
SessionFollowFrame,
|
||||
SessionHistoryRecord,
|
||||
SessionPage,
|
||||
SessionPageRequest,
|
||||
SessionProjectionBaseline,
|
||||
@@ -68,9 +70,9 @@ export class SessionHistoryController {
|
||||
request.maxMessages ?? DEFAULT_MAX_MESSAGES,
|
||||
request.throughSeq,
|
||||
)
|
||||
const entries = page.events.map(entryFor)
|
||||
const records = pageRecords(page.events)
|
||||
return {
|
||||
events: entries,
|
||||
records,
|
||||
hasMore: page.hasMore,
|
||||
}
|
||||
}
|
||||
@@ -128,7 +130,7 @@ export class SessionHistoryController {
|
||||
type: 'snapshot',
|
||||
header: source.header,
|
||||
cursor,
|
||||
events: page.events.map(entryFor),
|
||||
records: pageRecords(page.events),
|
||||
hasMore: page.hasMore,
|
||||
projections: source.projections === undefined
|
||||
? { asOfSeq: cursor, values: {} }
|
||||
@@ -317,3 +319,10 @@ function entryFor(event: SessionEvent): SessionEventEntry {
|
||||
event: event as unknown as SessionWireEvent,
|
||||
}
|
||||
}
|
||||
|
||||
/** Encode one bounded logical page without changing its pagination cut. */
|
||||
function pageRecords(events: readonly SessionEvent[]): SessionHistoryRecord[] {
|
||||
return packChunkRuns(events).map(record => isChunkRow(record)
|
||||
? { chunks: record }
|
||||
: entryFor(record))
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ import type {
|
||||
import type { Branded } from '@deepseek-ai/dsh-brand'
|
||||
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
|
||||
import type { ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import type { JsonValue, SessionHeader, SessionId, SurfaceOp } from '@deepseek-ai/dsh-session/types'
|
||||
import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
|
||||
import type { JobId } from '@deepseek-ai/dsh-jobs/brand'
|
||||
@@ -364,6 +365,14 @@ export interface SessionEventEntry {
|
||||
readonly event: SessionWireEvent
|
||||
}
|
||||
|
||||
/** One lossless run of consecutive Assistant delta events in a history page. */
|
||||
export interface SessionChunkRun {
|
||||
readonly chunks: ChunkRow
|
||||
}
|
||||
|
||||
/** One history-page record: a raw event or a packed Assistant delta run. */
|
||||
export type SessionHistoryRecord = SessionEventEntry | SessionChunkRun
|
||||
|
||||
/** Session event wire form; durable readers own recognition of merge-extensible event names. */
|
||||
export interface SessionWireEvent {
|
||||
readonly type: string
|
||||
@@ -392,7 +401,7 @@ export interface SessionFollowRequest {
|
||||
|
||||
/** One contiguous backwards page of a Session log. */
|
||||
export interface SessionPage {
|
||||
readonly events: readonly SessionEventEntry[]
|
||||
readonly records: readonly SessionHistoryRecord[]
|
||||
readonly hasMore: boolean
|
||||
}
|
||||
|
||||
@@ -402,7 +411,7 @@ export type SessionFollowFrame =
|
||||
readonly type: 'snapshot'
|
||||
readonly header: SessionHeader
|
||||
readonly cursor: number
|
||||
readonly events: readonly SessionEventEntry[]
|
||||
readonly records: readonly SessionHistoryRecord[]
|
||||
readonly hasMore: boolean
|
||||
readonly projections: SessionProjectionBaseline
|
||||
}
|
||||
|
||||
@@ -146,3 +146,14 @@ export function plainTurn(startSeq: number, turn: number, ask: string, answer: s
|
||||
export function entries(events: readonly SessionEvent[]): { event: SessionEvent }[] {
|
||||
return events.map(event => ({ event }))
|
||||
}
|
||||
|
||||
/** Build one view-less history response value. */
|
||||
export function historyValue(events: readonly SessionEvent[], hasMore = false): {
|
||||
records: { event: SessionEvent }[]
|
||||
hasMore: boolean
|
||||
} {
|
||||
return {
|
||||
records: entries(events),
|
||||
hasMore,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,6 +28,7 @@ import {
|
||||
} from '@deepseek-ai/dsh-api-gateway/client'
|
||||
import { RpcId } from '@deepseek-ai/dsh-client-connection/client'
|
||||
import type { SessionRemotes } from '../src/client/sessions/remotes.ts'
|
||||
import { historyEntries } from '../src/client/sessions/history-records.ts'
|
||||
|
||||
const AVAILABLE_STREAM_CONNECTION = {
|
||||
hostDescription: {
|
||||
@@ -137,7 +138,7 @@ export class FakeApiClient implements IApiClient {
|
||||
onFork: (payload: unknown) => Promise<RpcResponse<{ sessionId: SessionId }>> = () => Promise.resolve(ok({ sessionId: 'fk-fork' as SessionId }))
|
||||
onHistory: (payload: { sessionId: SessionId; throughSeq?: number; beforeSeq?: number; maxMessages?: number })
|
||||
=> Promise<RpcResponse<SessionPage & { readonly projections?: SessionProjectionBaseline }>> =
|
||||
() => Promise.resolve(ok({ events: [], hasMore: false }))
|
||||
() => Promise.resolve(ok({ records: [], hasMore: false }))
|
||||
|
||||
onPrompt: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
|
||||
onAttachment: (payload: unknown) => Promise<RpcResponse<{ attachment: { attachmentId: never; mediaType: 'image/png'; bytes: number; width: number; height: number }; data: string }>> =
|
||||
@@ -439,7 +440,8 @@ export class FakeApiClient implements IApiClient {
|
||||
ok: true,
|
||||
value: {
|
||||
...result.value,
|
||||
events: result.value.events.filter(entry => entry.event.seq <= request.throughSeq),
|
||||
records: historyEntries(result.value.records)
|
||||
.filter(entry => entry.event.seq <= request.throughSeq),
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -467,7 +469,8 @@ export class FakeApiClient implements IApiClient {
|
||||
)
|
||||
}
|
||||
const page = response.result.value
|
||||
const cursor = this.followCursor ?? page.events.at(-1)?.event.seq ?? -1
|
||||
const entries = historyEntries(page.records)
|
||||
const cursor = this.followCursor ?? entries.at(-1)?.event.seq ?? -1
|
||||
yield {
|
||||
type: 'snapshot',
|
||||
header: {
|
||||
@@ -479,7 +482,7 @@ export class FakeApiClient implements IApiClient {
|
||||
: {}),
|
||||
},
|
||||
cursor,
|
||||
events: page.events.filter(entry => entry.event.seq <= cursor),
|
||||
records: entries.filter(entry => entry.event.seq <= cursor),
|
||||
hasMore: page.hasMore,
|
||||
projections: page.projections ?? { asOfSeq: cursor, values: {} },
|
||||
}
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
/** Packed history record folding without token-by-token browser expansion. */
|
||||
|
||||
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)
|
||||
})
|
||||
})
|
||||
|
||||
describe('historyEntries', () => {
|
||||
it('keeps ordinary entries and views while folding a packed run without expansion', () => {
|
||||
const ordinary = {
|
||||
event: { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
|
||||
view: { for: 'call', view: { card: 'generic' } },
|
||||
} as unknown as SessionHistoryRecord
|
||||
const packed: SessionHistoryRecord = {
|
||||
chunks: {
|
||||
type: 'text-chunks',
|
||||
seq0: 1,
|
||||
time0: 2,
|
||||
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[0]).toBe(ordinary)
|
||||
expect(entries[1]?.event).toMatchObject({ seq: 1, data: { chunk: { text: 'abcd' } } })
|
||||
})
|
||||
})
|
||||
@@ -759,7 +759,7 @@ describe('connected generation', () => {
|
||||
it('refreshes query baselines without rebuilding independently resumed Session sources', async () => {
|
||||
const api = new FakeApiClient()
|
||||
api.onHistory = () => Promise.resolve(ok({
|
||||
events: entries(plainTurn(0, 0, 'a', 'b')) as never[],
|
||||
records: entries(plainTurn(0, 0, 'a', 'b')) as never[],
|
||||
hasMore: false,
|
||||
modelSelection: { provider: 'deepseek-official', model: 'deepseek-chat' },
|
||||
}))
|
||||
|
||||
@@ -105,7 +105,7 @@ describe('Session tail-page seeding', () => {
|
||||
const api = new FakeApiClient()
|
||||
const session = new Session(SID, api, fakeRemote(api))
|
||||
api.onHistory = () => Promise.resolve(ok({
|
||||
events: entries(plainTurn(0, 0, '问', '答')) as never[], hasMore: false,
|
||||
records: entries(plainTurn(0, 0, '问', '答')) as never[], hasMore: false,
|
||||
projections: { asOfSeq: 5, values: { 'test/marks': { marks: ['from-baseline'] } } },
|
||||
} as never))
|
||||
await session.open()
|
||||
@@ -116,7 +116,7 @@ describe('Session tail-page seeding', () => {
|
||||
const api = new FakeApiClient()
|
||||
const session = new Session(SID, api, fakeRemote(api))
|
||||
api.onHistory = () => Promise.resolve(ok({
|
||||
events: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
|
||||
records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false,
|
||||
projections: { asOfSeq: 5, values: { 'test/marks': { marks: ['baseline'] } } },
|
||||
} as never))
|
||||
await session.open()
|
||||
@@ -128,7 +128,7 @@ describe('Session tail-page seeding', () => {
|
||||
it('treats a blockless response as no reset: pushed values survive', async () => {
|
||||
const api = new FakeApiClient()
|
||||
const session = new Session(SID, api, fakeRemote(api))
|
||||
api.onHistory = () => Promise.resolve(ok({ events: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false }))
|
||||
api.onHistory = () => Promise.resolve(ok({ records: entries(plainTurn(0, 0, 'a', 'b')) as never[], hasMore: false }))
|
||||
await session.open()
|
||||
session.projections.apply('test/marks', { marks: ['pushed'] }, 9)
|
||||
await session.resync()
|
||||
|
||||
@@ -390,7 +390,7 @@ describe('cold history recovery view', () => {
|
||||
maxMessages: 10,
|
||||
})
|
||||
if (!history.ok) throw new Error('history failed')
|
||||
expect(history.value.events.map(entry => entry.event)).toMatchInlineSnapshot(`
|
||||
expect(history.value.records.map(record => 'event' in record ? record.event : record.chunks)).toMatchInlineSnapshot(`
|
||||
[
|
||||
{
|
||||
"data": {
|
||||
@@ -564,7 +564,8 @@ describe('subagent ownership fence', () => {
|
||||
},
|
||||
throughSeq: 3,
|
||||
}, new AbortController().signal)
|
||||
expect(history.events.map(entry => entry.event.type)).toEqual(events.map(event => event.type))
|
||||
expect(history.records.map(record => 'event' in record ? record.event.type : record.chunks.type))
|
||||
.toEqual(events.map(event => event.type))
|
||||
expect(ctx.agents.get(sessionId)).toBeUndefined()
|
||||
|
||||
const prompt = await remote.prompt(promptRequest({
|
||||
|
||||
@@ -4,10 +4,15 @@ import { describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||
import SessionStore from '@deepseek-ai/dsh-session'
|
||||
import { decodeStorageRecord } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import { CallId, createMessage, createToolResultMessage, createUserMessage } from '@deepseek-ai/dsh-llm'
|
||||
import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts'
|
||||
import type { SessionFollowFrame } from '@deepseek-ai/dsh-api-session-controller/types'
|
||||
import type {
|
||||
SessionFollowFrame,
|
||||
SessionPage,
|
||||
SessionWireEvent,
|
||||
} from '@deepseek-ai/dsh-api-session-controller/types'
|
||||
import { createSessionTestRemote, installSessionReadTestServices } from './test-remote.ts'
|
||||
|
||||
/** Append a production-shaped human prompt to the session surface. */
|
||||
@@ -77,6 +82,13 @@ async function openFollow(
|
||||
return { [Symbol.asyncIterator]: () => iterator }
|
||||
}
|
||||
|
||||
/** Expand packed page records for assertions over the logical journal. */
|
||||
function pageEvents(page: SessionPage): SessionWireEvent[] {
|
||||
return page.records.flatMap(record => 'event' in record
|
||||
? [record.event]
|
||||
: decodeStorageRecord(record.chunks))
|
||||
}
|
||||
|
||||
describe('Session history raw journal', () => {
|
||||
it('follows raw tool events and preserves result metadata without a Tools service', async () => {
|
||||
const { ctx } = await harness()
|
||||
@@ -169,7 +181,7 @@ describe('Session history raw journal', () => {
|
||||
})
|
||||
expect(response.ok).toBe(true)
|
||||
if (!response.ok) throw new Error('unreachable')
|
||||
expect(response.value.events).toEqual([
|
||||
expect(response.value.records).toEqual([
|
||||
{ event: start },
|
||||
{ event: call },
|
||||
{ event: result },
|
||||
@@ -210,7 +222,7 @@ describe('Session history raw journal', () => {
|
||||
maxMessages: 2,
|
||||
})
|
||||
if (!response.ok) throw new Error('unreachable')
|
||||
const page = response.value.events.map(entry => entry.event)
|
||||
const page = pageEvents(response.value)
|
||||
// Two append-origin messages fill the page even though a replacement copy of
|
||||
// the same event type sits in the window: the copy is model-only.
|
||||
const messages = page.filter(event => event.type === 'user/message' || event.type === 'assistant/message')
|
||||
@@ -257,7 +269,8 @@ describe('Session history raw journal', () => {
|
||||
maxMessages: 1,
|
||||
})
|
||||
if (!response.ok) throw new Error('unreachable')
|
||||
expect(response.value.events.map(entry => entry.event.seq)).toEqual([...sources, message.seq])
|
||||
expect(pageEvents(response.value).map(event => event.seq)).toEqual([...sources, message.seq])
|
||||
expect(response.value.records.filter(record => 'chunks' in record)).toHaveLength(1)
|
||||
expect(response.value.hasMore).toBe(true)
|
||||
} finally {
|
||||
min.mockRestore()
|
||||
|
||||
@@ -148,11 +148,12 @@ describe('session.history projections block', () => {
|
||||
ctx.sessionProjections.register(lastUserUnit())
|
||||
seedMessages(session, 3)
|
||||
const snapshot = await opening(remote(ctx), session.id)
|
||||
const { events, projections } = snapshot
|
||||
const { records, projections } = snapshot
|
||||
expect(projections.asOfSeq).toBe(session.seq - 1)
|
||||
expect(projections.values['test/last-user']).toEqual({ text: 'm2' })
|
||||
// asOfSeq IS the window tail: the last served event carries it.
|
||||
expect(events.at(-1)?.event.seq).toBe(projections.asOfSeq)
|
||||
const last = records.at(-1)
|
||||
expect(last !== undefined && 'event' in last ? last.event.seq : last?.chunks.seq0).toBe(projections.asOfSeq)
|
||||
})
|
||||
|
||||
it('returns a complete current replacement cut on each follow generation', async () => {
|
||||
@@ -162,7 +163,7 @@ describe('session.history projections block', () => {
|
||||
|
||||
const snapshot = await opening(remote(ctx), session.id)
|
||||
|
||||
expect(snapshot.events.map(entry => entry.event.seq)).toEqual([0, 1])
|
||||
expect(snapshot.records.map(record => 'event' in record ? record.event.seq : record.chunks.seq0)).toEqual([0, 1])
|
||||
expect(snapshot.projections.asOfSeq).toBe(1)
|
||||
expect(snapshot.projections.values).toEqual(
|
||||
expect.objectContaining({ 'test/last-user': { text: 'm1' } }),
|
||||
@@ -175,7 +176,7 @@ describe('session.history projections block', () => {
|
||||
|
||||
const snapshot = await opening(remote(ctx), session.id)
|
||||
|
||||
expect(snapshot.events).toEqual([])
|
||||
expect(snapshot.records).toEqual([])
|
||||
expect(snapshot.projections.asOfSeq).toBe(-1)
|
||||
expect(snapshot.projections.values).toEqual(
|
||||
expect.objectContaining({ 'test/last-user': null }),
|
||||
|
||||
@@ -6,7 +6,7 @@ import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
|
||||
import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client'
|
||||
import { Session, type SessionOptions } from '../src/client/sessions/session.ts'
|
||||
import { FakeApiClient, deferred, err, fakeRemote, ok } from './fake-api.client.ts'
|
||||
import { entries, ev, plainTurn } from './event-script.client.ts'
|
||||
import { entries, ev, historyValue, plainTurn } from './event-script.client.ts'
|
||||
|
||||
const SID = 'fk-s1' as SessionId
|
||||
const PARENT = 'fk-parent' as SessionId
|
||||
@@ -41,8 +41,7 @@ function eventSeqs(session: Session): number[] {
|
||||
}
|
||||
|
||||
function histResponse(events: SessionEvent[], hasMore = false) {
|
||||
// History returns raw journal envelopes around each event.
|
||||
return Promise.resolve(ok({ events: entries(events) as never[], hasMore }))
|
||||
return Promise.resolve(ok(historyValue(events, hasMore)))
|
||||
}
|
||||
|
||||
describe('Session open', () => {
|
||||
@@ -106,7 +105,7 @@ describe('Session open', () => {
|
||||
follow(api, ev.user(16, '插进来的')),
|
||||
]
|
||||
gate.resolve(ok({
|
||||
events: entries(page) as never[],
|
||||
records: entries(page) as never[],
|
||||
hasMore: false,
|
||||
modelSelection: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
|
||||
}))
|
||||
@@ -227,7 +226,7 @@ describe('paging', () => {
|
||||
const first = session.loadOlder()
|
||||
const second = session.loadOlder()
|
||||
gate.resolve(ok({
|
||||
events: entries(plainTurn(0, 0, 'a', 'b')) as never[],
|
||||
records: entries(plainTurn(0, 0, 'a', 'b')) as never[],
|
||||
hasMore: false,
|
||||
modelSelection: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
|
||||
}))
|
||||
@@ -571,7 +570,7 @@ describe('remaining branches', () => {
|
||||
api.onHistory = () => histResponse(plainTurn(6, 1, '新', '代'))
|
||||
const resynced = session.resync()
|
||||
stale.resolve(ok({
|
||||
events: entries(plainTurn(0, 0, '旧', '代')) as never[],
|
||||
records: entries(plainTurn(0, 0, '旧', '代')) as never[],
|
||||
hasMore: false,
|
||||
modelSelection: { provider: 'deepseek-official', model: 'stale' },
|
||||
})) // success, but its generation is gone
|
||||
@@ -590,7 +589,7 @@ describe('remaining branches', () => {
|
||||
api.onHistory = () => histResponse(plainTurn(6, 1, 'c', 'd'))
|
||||
const resynced = session.resync() // bumps the generation
|
||||
repairPull.resolve(ok({
|
||||
events: entries(plainTurn(0, 0, '旧', '页')) as never[],
|
||||
records: entries(plainTurn(0, 0, '旧', '页')) as never[],
|
||||
hasMore: false,
|
||||
modelSelection: { provider: 'deepseek-official', model: 'stale' },
|
||||
})) // repair result: stale, dropped
|
||||
@@ -617,7 +616,7 @@ describe('remaining branches', () => {
|
||||
const historyCall = ev.toolCall(6, 1, 'h1', 'bash', '{"cmd":"pwd"}')
|
||||
const historyResult = ev.toolResult(7, 1, 'h1', 'done')
|
||||
api.onHistory = () => Promise.resolve(ok({
|
||||
events: [
|
||||
records: [
|
||||
...entries(plainTurn(0, 0, 'a', 'b')),
|
||||
{ event: historyCall },
|
||||
{ event: historyResult },
|
||||
@@ -669,7 +668,7 @@ describe('resync', () => {
|
||||
])
|
||||
expect(session.eventSource.getSnapshot()).toBe(oldWindow)
|
||||
replacement.resolve(ok({
|
||||
events: entries(plainTurn(10, 2, '终', '页')) as never[],
|
||||
records: entries(plainTurn(10, 2, '终', '页')) as never[],
|
||||
hasMore: false,
|
||||
modelSelection: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
|
||||
}))
|
||||
|
||||
@@ -274,7 +274,7 @@ describe('Agent scope disposal lifecycle', () => {
|
||||
createdAt: 0,
|
||||
},
|
||||
cursor: -1,
|
||||
events: [],
|
||||
records: [],
|
||||
hasMore: false,
|
||||
projections: { asOfSeq: -1, values: {} },
|
||||
} as const,
|
||||
@@ -343,7 +343,7 @@ describe('Agent scope disposal lifecycle', () => {
|
||||
type: 'snapshot',
|
||||
header: { version: 0, id: sessionId, createdAt: 0 },
|
||||
cursor: -1,
|
||||
events: [],
|
||||
records: [],
|
||||
hasMore: false,
|
||||
projections: { asOfSeq: -1, values: {} },
|
||||
} as const,
|
||||
|
||||
@@ -40,7 +40,7 @@ function entry(seq: number): SessionEventEntry {
|
||||
}
|
||||
|
||||
function page(events: readonly SessionEventEntry[], hasMore = false): SessionPage {
|
||||
return { events, hasMore }
|
||||
return { records: events, hasMore }
|
||||
}
|
||||
|
||||
function snapshot(
|
||||
@@ -56,7 +56,7 @@ function snapshot(
|
||||
createdAt: 0,
|
||||
},
|
||||
cursor,
|
||||
events,
|
||||
records: events,
|
||||
hasMore,
|
||||
projections: { asOfSeq: cursor, values: {} },
|
||||
}
|
||||
|
||||
@@ -87,7 +87,7 @@ describe('SessionHistoryController', () => {
|
||||
{ address: { kind: 'session', sessionId: session.id }, throughSeq: 1 },
|
||||
new AbortController().signal,
|
||||
)
|
||||
expect(page.events.map(entry => entry.event.seq)).toEqual([0, 1])
|
||||
expect(page.records.map(entry => 'event' in entry ? entry.event.seq : entry.chunks.seq0)).toEqual([0, 1])
|
||||
|
||||
abort.abort()
|
||||
expect(await iterator.next()).toMatchObject({ done: true })
|
||||
@@ -137,7 +137,7 @@ describe('SessionHistoryController', () => {
|
||||
value: {
|
||||
type: 'snapshot',
|
||||
cursor: 2,
|
||||
events: [{ event: { seq: 0 } }, { event: { seq: 1 } }, { event: { seq: 2 } }],
|
||||
records: [{ event: { seq: 0 } }, { event: { seq: 1 } }, { event: { seq: 2 } }],
|
||||
},
|
||||
})
|
||||
session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
|
||||
@@ -200,7 +200,7 @@ describe('SessionHistoryController', () => {
|
||||
await expect(opening).resolves.toMatchObject({
|
||||
done: false,
|
||||
value: {
|
||||
type: 'snapshot', cursor: 1, events: [{ event: { seq: 0 } }, { event: { seq: 1 } }],
|
||||
type: 'snapshot', cursor: 1, records: [{ event: { seq: 0 } }, { event: { seq: 1 } }],
|
||||
},
|
||||
})
|
||||
expect(attached.id).toBe(sessionId)
|
||||
@@ -301,7 +301,7 @@ describe('SessionHistoryController', () => {
|
||||
await expect(iterator.next()).resolves.toMatchObject({ done: false, value: { type: 'snapshot', cursor: -1 } })
|
||||
await expect(transport.page({
|
||||
address: { kind: 'session', sessionId: session.id }, throughSeq: -1,
|
||||
}, signal())).resolves.toMatchObject({ events: [], hasMore: false })
|
||||
}, signal())).resolves.toMatchObject({ records: [], hasMore: false })
|
||||
abort.abort()
|
||||
await expect(iterator.next()).resolves.toMatchObject({ done: true })
|
||||
})
|
||||
@@ -377,7 +377,7 @@ describe('SessionHistoryController', () => {
|
||||
await expect(transport.page({
|
||||
address: { kind: 'subagent', parentSessionId, childSessionId, mode: 'continuable' },
|
||||
throughSeq: 0,
|
||||
}, signal)).resolves.toMatchObject({ events: [{ event: { type: 'subagent/descriptor' } }] })
|
||||
}, signal)).resolves.toMatchObject({ records: [{ event: { type: 'subagent/descriptor' } }] })
|
||||
await expect(transport.page({
|
||||
address: {
|
||||
kind: 'subagent',
|
||||
@@ -505,7 +505,7 @@ describe('SessionHistoryController', () => {
|
||||
await expect(ordinaryBench.transport.page({
|
||||
address: { kind: 'session', sessionId: ordinaryId },
|
||||
throughSeq: 0,
|
||||
}, signal())).resolves.toMatchObject({ events: [{ event: { seq: 0 } }] })
|
||||
}, signal())).resolves.toMatchObject({ records: [{ event: { seq: 0 } }] })
|
||||
|
||||
const parentSessionId = SessionId('cold-parent')
|
||||
const childSessionId = SessionId('cold-child')
|
||||
@@ -618,12 +618,13 @@ describe('SessionHistoryController', () => {
|
||||
const page = await transport.page({
|
||||
address: { kind: 'session', sessionId: session.id }, throughSeq: replacement.seq, maxMessages: 2,
|
||||
}, signal())
|
||||
expect(page.events.map(entry => entry.event.seq)).toEqual([3, 4, 5, replacement.seq])
|
||||
expect(page.records.map(entry => 'event' in entry ? entry.event.seq : entry.chunks.seq0))
|
||||
.toEqual([3, 4, 5, replacement.seq])
|
||||
expect(page.hasMore).toBe(true)
|
||||
const before = await transport.page({
|
||||
address: { kind: 'session', sessionId: session.id }, throughSeq: replacement.seq, beforeSeq: 3, maxMessages: 1,
|
||||
}, signal())
|
||||
expect(before.events.map(entry => entry.event.seq)).toEqual([2])
|
||||
expect(before.records.map(entry => 'event' in entry ? entry.event.seq : entry.chunks.seq0)).toEqual([2])
|
||||
})
|
||||
|
||||
it('keeps cited source events in the page that owns their appended message', async () => {
|
||||
@@ -637,7 +638,7 @@ describe('SessionHistoryController', () => {
|
||||
const page = await transport.page({
|
||||
address: { kind: 'session', sessionId: session.id }, throughSeq: 1, maxMessages: 1,
|
||||
}, signal())
|
||||
expect(page.events.map(entry => entry.event.seq)).toEqual([0, 1])
|
||||
expect(page.records.map(entry => 'event' in entry ? entry.event.seq : entry.chunks.seq0)).toEqual([0, 1])
|
||||
expect(page.hasMore).toBe(false)
|
||||
})
|
||||
|
||||
|
||||
@@ -22,6 +22,8 @@ import type {
|
||||
SessionHeader,
|
||||
SessionId,
|
||||
} from '@deepseek-ai/dsh-session/types'
|
||||
import { isChunkRow, packChunkRuns } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import type { ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import type { TodoItem } from '@deepseek-ai/dsh-tool-todo/client'
|
||||
// Type-only: the brand constructor is host-side; the fixture casts at its
|
||||
// wire-fabrication boundary (the schema layer's one-cast-point posture).
|
||||
@@ -76,6 +78,12 @@ interface FixtureHistoryEntry {
|
||||
readonly event: SessionEvent
|
||||
}
|
||||
|
||||
interface FixtureHistoryChunkRun {
|
||||
readonly chunks: ChunkRow
|
||||
}
|
||||
|
||||
type FixtureHistoryRecord = FixtureHistoryEntry | FixtureHistoryChunkRun
|
||||
|
||||
type FixtureSessionAddress =
|
||||
| { readonly kind: 'session'; readonly sessionId: SessionId }
|
||||
| {
|
||||
@@ -102,7 +110,7 @@ type FixtureFollowFrame =
|
||||
readonly type: 'snapshot'
|
||||
readonly header: SessionHeader
|
||||
readonly cursor: number
|
||||
readonly events: readonly FixtureHistoryEntry[]
|
||||
readonly records: readonly FixtureHistoryRecord[]
|
||||
readonly hasMore: boolean
|
||||
readonly projections: FixtureProjectionsBlock
|
||||
}
|
||||
@@ -1392,7 +1400,7 @@ function pageOf(
|
||||
log: readonly SessionEvent[],
|
||||
beforeSeq: number | undefined,
|
||||
maxMessages: number,
|
||||
): { events: FixtureHistoryEntry[]; hasMore: boolean } {
|
||||
): { records: FixtureHistoryRecord[]; hasMore: boolean } {
|
||||
const end = beforeSeq === undefined ? log.length : Math.max(0, Math.min(beforeSeq, log.length))
|
||||
let start = 0
|
||||
let messages = 0
|
||||
@@ -1406,8 +1414,10 @@ function pageOf(
|
||||
break
|
||||
}
|
||||
}
|
||||
const events = log.slice(start, end).map((event): FixtureHistoryEntry => ({ event }))
|
||||
return { events, hasMore: start > 0 }
|
||||
const records = packChunkRuns(log.slice(start, end)).map((record): FixtureHistoryRecord => (
|
||||
isChunkRow(record) ? { chunks: record } : { event: record }
|
||||
))
|
||||
return { records, hasMore: start > 0 }
|
||||
}
|
||||
|
||||
/** Fixture mirror of host session-scoped attachment authorization. */
|
||||
@@ -2968,7 +2978,7 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
|
||||
...(summary.agentPreset === undefined ? {} : { agentPreset: summary.agentPreset }),
|
||||
},
|
||||
cursor,
|
||||
events: initial.events,
|
||||
records: initial.records,
|
||||
hasMore: initial.hasMore,
|
||||
projections: { asOfSeq: cursor, values: projectionValuesOf(snapshot) },
|
||||
}
|
||||
|
||||
@@ -9,6 +9,8 @@ import type {
|
||||
SessionId,
|
||||
} from '../src/client/api.ts'
|
||||
import { RpcId } from '../src/client/api.ts'
|
||||
import { decodeStorageRecord } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import type { ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import {
|
||||
FixtureApiClient,
|
||||
createFixtureFaces,
|
||||
@@ -38,16 +40,28 @@ interface FixtureHistoryEntry {
|
||||
readonly event: SessionEvent
|
||||
}
|
||||
|
||||
interface FixtureHistoryChunkRun {
|
||||
readonly chunks: ChunkRow
|
||||
}
|
||||
|
||||
type FixtureHistoryRecord = FixtureHistoryEntry | FixtureHistoryChunkRun
|
||||
|
||||
interface FixturePage {
|
||||
readonly events: readonly FixtureHistoryEntry[]
|
||||
readonly records: readonly FixtureHistoryRecord[]
|
||||
readonly hasMore: boolean
|
||||
}
|
||||
|
||||
function historyEvents(records: readonly FixtureHistoryRecord[]): SessionEvent[] {
|
||||
return records.flatMap(record => 'event' in record
|
||||
? [record.event]
|
||||
: decodeStorageRecord(record.chunks))
|
||||
}
|
||||
|
||||
type FixtureFollowFrame =
|
||||
| {
|
||||
readonly type: 'snapshot'
|
||||
readonly cursor: number
|
||||
readonly events: readonly FixtureHistoryEntry[]
|
||||
readonly records: readonly FixtureHistoryRecord[]
|
||||
readonly hasMore: boolean
|
||||
readonly projections: {
|
||||
readonly asOfSeq: number
|
||||
@@ -580,21 +594,22 @@ describe('createFixtureApi', () => {
|
||||
if (!tail.result.ok) throw new Error('history failed')
|
||||
const tailPage = tail.result.value
|
||||
expect(tailPage.hasMore).toBe(true)
|
||||
expect(tailPage.events[0]?.event.type).toBe('turn/start') // cut lands on a turn boundary
|
||||
const boundary = tailPage.events[0]?.event.seq ?? 0
|
||||
const tailEvents = historyEvents(tailPage.records)
|
||||
expect(tailEvents[0]?.type).toBe('turn/start') // cut lands on a turn boundary
|
||||
const boundary = tailEvents[0]?.seq ?? 0
|
||||
expect(boundary).toBeGreaterThan(0)
|
||||
const older = await api.sessions.history(req({ sessionId: sid('fx-alpha'), beforeSeq: boundary, maxMessages: 10 }))
|
||||
if (!older.result.ok) throw new Error('older failed')
|
||||
const olderTail = older.result.value.events.at(-1)?.event
|
||||
const olderTail = historyEvents(older.result.value.records).at(-1)
|
||||
expect((olderTail?.seq ?? -1) + 1).toBe(boundary) // pages stitch with no hole/overlap
|
||||
// Out-of-range beforeSeq clamps instead of exploding.
|
||||
const clamped = await api.sessions.history(req({ sessionId: sid('fx-alpha'), beforeSeq: -5, maxMessages: 10 }))
|
||||
if (!clamped.result.ok) throw new Error('clamped failed')
|
||||
expect(clamped.result.value.events).toEqual([])
|
||||
expect(clamped.result.value.records).toEqual([])
|
||||
// Unknown session: empty page, not an error (history of a bare id).
|
||||
const empty = await api.sessions.history(req({ sessionId: sid('no-such'), maxMessages: 10 }))
|
||||
if (!empty.result.ok) throw new Error('empty failed')
|
||||
expect(empty.result.value).toEqual({ events: [], hasMore: false })
|
||||
expect(empty.result.value).toEqual({ records: [], hasMore: false })
|
||||
})
|
||||
|
||||
it('serves raw history entries with replayable tool-result metadata', async () => {
|
||||
@@ -602,10 +617,8 @@ describe('createFixtureApi', () => {
|
||||
const response = await api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 200 }))
|
||||
if (!response.result.ok) throw new Error('history failed')
|
||||
|
||||
const entries = response.result.value.events
|
||||
expect(entries.every(entry => !Object.hasOwn(entry, 'view'))).toBe(true)
|
||||
const results = entries
|
||||
.map(entry => entry.event)
|
||||
const records = response.result.value.records
|
||||
const results = historyEvents(records)
|
||||
.filter(event => event.type === 'tool/result')
|
||||
|
||||
expect(results.find(event => event.data.turn === 64)).toMatchObject({
|
||||
@@ -668,7 +681,7 @@ describe('createFixtureApi', () => {
|
||||
await new Promise(resolve => setTimeout(resolve, 600))
|
||||
const after = await api.sessions.history(req({ sessionId }))
|
||||
if (!after.result.ok) throw new Error('history failed')
|
||||
expect(JSON.stringify(after.result.value.events)).toContain('openai/gpt-5')
|
||||
expect(JSON.stringify(after.result.value.records)).toContain('openai/gpt-5')
|
||||
})
|
||||
|
||||
it('serves configured DeepSeek readiness and keeps credential values write-only', async () => {
|
||||
@@ -705,7 +718,7 @@ describe('createFixtureApi', () => {
|
||||
const api = createFixtureApi()
|
||||
const tail = await api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 10 }))
|
||||
if (!tail.result.ok) throw new Error('history failed')
|
||||
const events = tail.result.value.events.map(e => e.event)
|
||||
const events = historyEvents(tail.result.value.records)
|
||||
const todoAt = events.findIndex(e => e.type === 'todo/write')
|
||||
expect(todoAt).toBeGreaterThan(0)
|
||||
// Production ordering (the tool appends mid-execution): call → snapshot → result.
|
||||
@@ -1134,8 +1147,8 @@ describe('createFixtureApi', () => {
|
||||
// so the event is located by seq and its payload checked structurally).
|
||||
const history = await api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 100 }))
|
||||
if (!history.result.ok) throw new Error('history failed')
|
||||
const appended = history.result.value.events.find(entry => entry.event.seq === acceptedSeq)
|
||||
expect(appended?.event).toMatchObject({
|
||||
const appended = historyEvents(history.result.value.records).find(event => event.seq === acceptedSeq)
|
||||
expect(appended).toMatchObject({
|
||||
type: 'session/title',
|
||||
data: { title: '重命名', messageSeqs: [], source: { kind: 'user' } },
|
||||
})
|
||||
@@ -1433,8 +1446,9 @@ describe('createFixtureApi', () => {
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
await vi.waitFor(() => {
|
||||
const snapshot = followed.find(frame => frame.type === 'snapshot')
|
||||
expect(snapshot?.events.some(entry => JSON.stringify(entry.event.data).includes('静默丢帧'))).toBe(true)
|
||||
expect(snapshot?.events.some(entry => JSON.stringify(entry.event.data).includes('正常直播'))).toBe(true)
|
||||
const events = snapshot === undefined ? [] : historyEvents(snapshot.records)
|
||||
expect(events.some(event => JSON.stringify(event.data).includes('静默丢帧'))).toBe(true)
|
||||
expect(events.some(event => JSON.stringify(event.data).includes('正常直播'))).toBe(true)
|
||||
})
|
||||
hooks.appendTitle('fx-alpha', 'Fixture 修订标题')
|
||||
hooks.beginModelRetry('fx-alpha')
|
||||
@@ -1456,7 +1470,7 @@ describe('createFixtureApi', () => {
|
||||
// Paging and resumed follow agree on the recovered durable event.
|
||||
const repull = await api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 5 }))
|
||||
if (!repull.result.ok) throw new Error('repull failed')
|
||||
expect(JSON.stringify(repull.result.value.events)).toContain('静默丢帧')
|
||||
expect(JSON.stringify(repull.result.value.records)).toContain('静默丢帧')
|
||||
// breakStreams force-ends follow and control without client aborts.
|
||||
await new Promise(resolve => setTimeout(resolve, 10))
|
||||
hooks.breakStreams()
|
||||
@@ -1593,7 +1607,7 @@ describe('FixtureApiClient (protocol-level fake carrier)', () => {
|
||||
|
||||
const goalHistory = await sessions.history({ sessionId: id })
|
||||
if (!goalHistory.result.ok) throw new Error('goal history failed')
|
||||
const goalEvents = goalHistory.result.value.events.map(entry => entry.event as unknown as {
|
||||
const goalEvents = historyEvents(goalHistory.result.value.records).map(event => event as unknown as {
|
||||
type: string
|
||||
data: {
|
||||
operation?: string
|
||||
|
||||
@@ -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 packages/core/session/README.md
|
||||
README.md: 0e3cdcb1e0135cda4d1ac469a0cbc2c2f44c3d94
|
||||
README.zh.md: 383777227fa5903c0e7285d31e8d70ee9ebb1eab
|
||||
README.md: 49fd31d2fae2d80de5c427ba064a4751ec1983f3
|
||||
README.zh.md: bf819ba5f4d43a981ee673521833d3db391d00f9
|
||||
|
||||
@@ -52,7 +52,7 @@ Session-event import separates ownership from message validation. `snapshotSessi
|
||||
|
||||
### Chunk-row storage codec (`chunk-rows.ts`)
|
||||
|
||||
The shared [storage codec](src/chunk-rows.ts) losslessly converts event sequences to compact rows and back. It preserves unrecognized events verbatim and rejects malformed encoded rows; persistence backends decide whether to enable packed writes.
|
||||
The shared [row codec](src/chunk-rows.ts) losslessly converts event sequences to compact rows and back. It preserves unrecognized events verbatim and rejects malformed encoded rows. Persistence backends decide whether to enable packed writes; bounded history transports may use the same rows while retaining the complete logical event interval and exposing exact decoding to consumers that need token boundaries.
|
||||
|
||||
### Surface types
|
||||
|
||||
|
||||
@@ -52,7 +52,7 @@
|
||||
|
||||
### 分片行存储编解码器（`chunk-rows.ts`）
|
||||
|
||||
共享的[存储编解码器](src/chunk-rows.ts)在事件序列与紧凑行之间无损转换。它会逐字保留无法识别的事件，并拒绝形态错误的编码行；是否启用打包写入由持久化后端决定。
|
||||
共享的[行编解码器](src/chunk-rows.ts)在事件序列与紧凑行之间无损转换。它会逐字保留无法识别的事件，并拒绝形态错误的编码行。持久化后端决定是否启用打包写入；有界历史传输也可以使用同一种行，同时保留完整的逻辑事件区间，并向需要 token 边界的消费方提供精确解码能力。
|
||||
|
||||
### Surface 类型
|
||||
|
||||
|
||||
@@ -26,6 +26,10 @@
|
||||
"types": "./lib/types/types.d.ts",
|
||||
"default": "./lib/types/types.js"
|
||||
},
|
||||
"./chunk-rows": {
|
||||
"types": "./lib/types/chunk-rows.d.ts",
|
||||
"default": "./lib/types/chunk-rows.js"
|
||||
},
|
||||
"./src/*": "./src/*",
|
||||
"./package.json": "./package.json",
|
||||
"./surface": {
|
||||
|
||||
@@ -1,24 +1,25 @@
|
||||
/**
|
||||
* Lossless storage packing for `assistant/chunk` delta runs. Providers stream
|
||||
* Lossless row packing for `assistant/chunk` delta runs. Providers stream
|
||||
* token-sized deltas, so a log stores hundreds of near-identical event lines
|
||||
* whose JSON envelopes dwarf their payloads (~56× measured on a real DeepSeek
|
||||
* session). This module packs each run of consecutive same-block delta chunks
|
||||
* into ONE storage row — `text-chunks`, `reasoning-chunks`, or
|
||||
* `tool-call-chunks` — and expands rows back to the exact original events.
|
||||
*
|
||||
* Storage rows are a durable-encoding vocabulary, NOT session events: they
|
||||
* never enter `Session.events`, have no `SessionEventMap` entry, and use bare
|
||||
* (slash-less) type tags so a reader cannot confuse them with the event
|
||||
* taxonomy (precedent: the JSONL header line's `session` tag). The encoder
|
||||
* whitelists exact shapes — anything it does not fully recognize is stored
|
||||
* verbatim, so unknown fields or future chunk variants lose compression, never
|
||||
* data. The decoder validates before expanding and fails loud on a malformed
|
||||
* row-tagged value instead of silently dropping a whole run.
|
||||
* Packed rows are an encoding vocabulary, NOT session events: they never enter
|
||||
* `Session.events`, have no `SessionEventMap` entry, and use bare (slash-less)
|
||||
* type tags so a reader cannot confuse them with the event taxonomy
|
||||
* (precedent: the JSONL header line's `session` tag). Persistence and bounded
|
||||
* history transport both use the codec. The encoder whitelists exact shapes —
|
||||
* anything it does not fully recognize stays verbatim, so unknown fields or
|
||||
* future chunk variants lose compression, never data. The decoder validates
|
||||
* before expanding and fails loud on a malformed row-tagged value instead of
|
||||
* silently dropping a whole run.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-session/chunk-rows
|
||||
*/
|
||||
|
||||
import { CallId, assertNever } from '@deepseek-ai/dsh-llm'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm/brand'
|
||||
import type { StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import type { SessionEvent } from './types.ts'
|
||||
|
||||
@@ -69,6 +70,26 @@ export type ChunkRow =
|
||||
/** One durable log line's JSON value: a session event verbatim, or a packed chunk row. */
|
||||
export type StorageRecord = SessionEvent | ChunkRow
|
||||
|
||||
/**
|
||||
* Test whether an encoded record is a packed chunk row rather than a Session event.
|
||||
* @param record - one persistence or bounded-history encoding record.
|
||||
* @returns Whether the record is a packed chunk row.
|
||||
*/
|
||||
export function isChunkRow(record: StorageRecord): record is ChunkRow {
|
||||
return record.type === 'text-chunks'
|
||||
|| record.type === 'reasoning-chunks'
|
||||
|| record.type === 'tool-call-chunks'
|
||||
}
|
||||
|
||||
/**
|
||||
* Number of logical Session events represented by one packed row.
|
||||
* @param row - validated or encoder-produced packed row.
|
||||
* @returns Count of consecutive chunk events in the row.
|
||||
*/
|
||||
export function chunkRowLength(row: ChunkRow): number {
|
||||
return row.type === 'tool-call-chunks' ? row.data.args.length : row.data.texts.length
|
||||
}
|
||||
|
||||
/**
|
||||
* Minimum members before a run packs. Below it a row's envelope rivals the
|
||||
* event lines it replaces. A format constant, not a tunable: both layouts
|
||||
@@ -278,7 +299,7 @@ function validateRow(value: Record<string, unknown>, tag: ChunkRow['type']): Chu
|
||||
// outside any encoder's image: float arithmetic would round it to a
|
||||
// different number than exact arithmetic, a silent corruption. Within safe
|
||||
// range every step is exact, so the first departure is always caught.
|
||||
if (!Number.isSafeInteger((value.seq0 as number) + payload.length - 1)) {
|
||||
if (payload.length - 1 > Number.MAX_SAFE_INTEGER - (value.seq0 as number)) {
|
||||
malformed(tag, 'member seqs must stay safe integers')
|
||||
}
|
||||
let time = value.time0 as number
|
||||
@@ -313,9 +334,11 @@ function expandRow(row: ChunkRow): SessionEvent[] {
|
||||
argumentsDelta: members[k] as string,
|
||||
}
|
||||
break
|
||||
/* v8 ignore next 2 -- validateRow only returns the three row tags */
|
||||
default:
|
||||
return assertNever(row, 'chunk-rows expandRow')
|
||||
/* v8 ignore next 4 -- validateRow only returns the three row tags */
|
||||
default: {
|
||||
const unreachable: never = row
|
||||
throw new Error(`chunk-rows received unsupported row ${String(unreachable)}`)
|
||||
}
|
||||
}
|
||||
events.push({
|
||||
type: 'assistant/chunk',
|
||||
|
||||
@@ -9,6 +9,7 @@ import fc from 'fast-check'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import type { StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import { decodeStorageRecord, packChunkRuns } from '@deepseek-ai/dsh-session'
|
||||
import { chunkRowLength, isChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
|
||||
import type { ChunkRow, SessionEvent, StorageRecord } from '@deepseek-ai/dsh-session'
|
||||
|
||||
/** Build an `assistant/chunk` event with the exact live-append shape. */
|
||||
@@ -37,6 +38,9 @@ describe('packChunkRuns', () => {
|
||||
expect(row.seq0).toBe(0)
|
||||
expect(row.time0).toBe(1000)
|
||||
expect(row.data).toMatchObject({ turn: 1, step: 1, index: 0, dt: [10, 10, 10, 10], texts: ['t0', 't1', 't2', 't3', 't4'] })
|
||||
expect(isChunkRow(row)).toBe(true)
|
||||
expect(chunkRowLength(row)).toBe(5)
|
||||
expect(isChunkRow(events[0] as SessionEvent)).toBe(false)
|
||||
expect(decodeAll(packed)).toStrictEqual(events)
|
||||
})
|
||||
|
||||
@@ -48,6 +52,7 @@ describe('packChunkRuns', () => {
|
||||
expect(packed.map(r => (r as ChunkRow).type)).toStrictEqual(['reasoning-chunks', 'tool-call-chunks'])
|
||||
const row = packed[1] as ChunkRow & { type: 'tool-call-chunks' }
|
||||
expect(row.data).toMatchObject({ id: 'c1', name: 'write', args: ['a4', 'a5', 'a6'] })
|
||||
expect(chunkRowLength(row)).toBe(3)
|
||||
expect(decodeAll(packed)).toStrictEqual([...reasoning, ...toolCall])
|
||||
})
|
||||
|
||||
@@ -167,7 +172,7 @@ describe('decodeStorageRecord', () => {
|
||||
['a dt arity mismatch', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [1, 2], texts: ['a', 'b'] } }],
|
||||
['a non-finite dt gap', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [NaN], texts: ['a', 'b'] } }],
|
||||
['a fractional dt gap', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [0.5], texts: ['a', 'b'] } }],
|
||||
['a member seq leaving safe range', { type: 'text-chunks', seq0: Number.MAX_SAFE_INTEGER, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [0, 0], texts: ['a', 'b', 'c'] } }],
|
||||
['a member seq leaving safe range', { type: 'text-chunks', seq0: Number.MAX_SAFE_INTEGER, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [0], texts: ['a', 'b'] } }],
|
||||
['a member time leaving safe range', { type: 'text-chunks', seq0: 0, time0: Number.MAX_SAFE_INTEGER, data: { turn: 1, step: 1, index: 0, dt: [1], texts: ['a', 'b'] } }],
|
||||
['a non-numeric turn', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 'x', step: 1, index: 0, dt: [], texts: ['a'] } }],
|
||||
['a tool-call row without id', { type: 'tool-call-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], args: ['a'] } }],
|
||||
|
||||
@@ -64,6 +64,7 @@
|
||||
"@deepseek-ai/dsh-typert-generator": ["./packages/typert/generator/src/index.ts"],
|
||||
"@deepseek-ai/dsh-attachment/types": ["./packages/attachment/attachment/src/types.ts"],
|
||||
"@deepseek-ai/dsh-session/types": ["./packages/core/session/src/types.ts"],
|
||||
"@deepseek-ai/dsh-session/chunk-rows": ["./packages/core/session/src/chunk-rows.ts"],
|
||||
"@deepseek-ai/dsh-session/surface": ["./packages/core/session/src/surface.ts"],
|
||||
"@deepseek-ai/dsh-session-projection/types": ["./packages/session/session-projection/src/types.ts"],
|
||||
"@deepseek-ai/dsh-tool-todo/types": ["./packages/todo/tool-todo/src/types.ts"],
|
||||
|
||||
Reference in New Issue
Block a user