Merge pull request #5403 from deepseek-harness/perf/session-corpus-benchmarks

test(bench): measure Session list, search, and fork over large corpora

.agents/notes/implemented/testing/2026-09-28-session-corpus-performance.i18n.yaml

@@ -0,0 +1,31 @@
+# Bilingual-pair consistency record for 2026-09-28-session-corpus-performance.md (docs/i18n/README.md): per heading
+# section, a hash of its English and Chinese blocks outside code blocks and generated regions.
+# After editing either side, bring the other along and re-record with:
+#   pnpm run verify-translation-pairing --write .agents/notes/implemented/testing/2026-09-28-session-corpus-performance.md
+/agent-note-performance-baselines-for-large-session-corpora:
+  en: e84801a4e9d16b7a
+  zh: 2fb8bc57ed14ea0d
+/agent-note-performance-baselines-for-large-session-corpora/problem:
+  en: d2a5a514f5c0aec3
+  zh: becf99e4e8ae7b8f
+/agent-note-performance-baselines-for-large-session-corpora/decision:
+  en: f8d67782422a1f40
+  zh: 296d52c006d41661
+/agent-note-performance-baselines-for-large-session-corpora/decision/workload:
+  en: ae4682fdb06d3192
+  zh: 6a48d412f15bedfa
+/agent-note-performance-baselines-for-large-session-corpora/decision/corpus-construction:
+  en: 1310a89c351e5bfe
+  zh: e6386ae10b893c2d
+/agent-note-performance-baselines-for-large-session-corpora/decision/measured-endpoints:
+  en: b4d81575ac574f6f
+  zh: f4f291ab7ce83094
+/agent-note-performance-baselines-for-large-session-corpora/calibration-evidence:
+  en: f934c95748dcbfe0
+  zh: df6a68286c21510f
+/agent-note-performance-baselines-for-large-session-corpora/alternatives-considered:
+  en: 4eaa4666262a1aee
+  zh: 7e2eb44b94bad05d
+/agent-note-performance-baselines-for-large-session-corpora/consequences:
+  en: f68218a31c17e0e6
+  zh: 2ffd8943e41774bf

.agents/notes/implemented/testing/2026-09-28-session-corpus-performance.md

@@ -0,0 +1,80 @@
+# Agent Note: Performance baselines for large Session corpora
+
+Status: implemented
+
+English | [中文](2026-09-28-session-corpus-performance.zh.md)
+
+## Problem
+
+Existing gates measure one long Session: [opening it](2026-09-04-session-open-performance-gate.md) or [continuing it](2026-09-06-backend-continuation-performance.md). Listing, content search, and fork scale with the whole corpus or with the source Session's length, and no gate covered a user with many long Sessions. A corpus of uniform small Sessions would understate the work, because real corpora are heavy-tailed.
+
+## Decision
+
+The [session-corpus benchmark](../../../../benchmarks/session-corpus/session-corpus.bench.ts) measures three Web Host endpoints over synthetic current-generation Zstandard corpora. No product implementation changes.
+
+### Workload
+
+On 2026-09-28, one developer's local DSH home held 1,650 Sessions in 88 project directories. Aggregate counts were extracted from the newest generation of each Session; no content, identity, or path left that machine. [corpus-shape.ts](../../../../benchmarks/session-corpus/corpus-shape.ts) records 22 upper-quantile anchors for three dimensions: logical events (compact delta rows expanded), decompressed bytes, and turns.
+
+| Quantile | Events | Logical bytes | Turns |
+|---|---:|---:|---:|
+| p50 | 53 | 86,195 | 1 |
+| p90 | 709 | 1,849,503 | 3 |
+| p99 | 7,229 | 8,479,554 | 18 |
+| max | 84,467 | 48,139,877 | 2,181 |
+
+Rank `r` of an `n`-Session corpus takes the first anchor at or above quantile `(r + 1) / n`, so every quantile of the synthetic corpus is at least the measured one. All dimensions share the rank: a long Session is long in every dimension. At 1,000 Sessions this upper-step sampling adds 16% events and 14% bytes over the measured quantile function; coarser anchors added up to 40%.
+
+Other measured aggregates also set the fixture: 19 Sessions per project directory, one subagent child per six Sessions (measured 288 of 1,650), a 20,000-character system prompt (the measured median largest event is 21.9 KB), 3.9 characters per streamed delta, and 1.6 events per Zstandard frame. Turns contain tool steps (assistant message, tool call, tool result) until the event target is reached; a secant search over payload characters then reaches the byte target from below. Text comes from a Zipf-distributed synthetic vocabulary. Its stored compression ratio is 2.8, below the measured 3.4, so the fixture stores more compressed bytes than the source corpus.
+
+### Corpus construction
+
+Two preparation processes, one per CPU of a standard hosted runner, split the anchors into byte-balanced groups. Each authors one event body per anchor, compresses it once, and saves it with the anchor's projection record. A seed process then writes each Session as its own header frame followed by its anchor's body. Frame compression and file writes run eight at a time: each pending compression holds a native Zstandard context, and unbounded compression peaked at 7.7 GB RSS, more than a standard hosted runner provides. Sessions sharing an anchor differ only in header identity, creation time, project directory, and parent link. Creation time is a fixed permutation of rank. List order follows each body's last prompt time, which Sessions sharing an anchor share; it does not affect list cost. Events carry the fields that `Session.append` records, compact streams come from the production `AssistantStreamAccumulator`, and rows are encoded with the persistence package's `eventLines`. Before saving, each preparation process replays its authored bodies through `Session.create`, as `sessionQuery.readSession` does.
+
+The projection cache is part of the list endpoint: each cold row carries cached projection values. Preparation folds each anchor once through the public `sessionProjectionCache.coldSnapshot` write-back. The seed then writes per-record documents for every Session with `serializeRecord`, rebinding identity to each header. Writing through the domain API costs one fsync per record, which took 89 s for 10,000 Sessions. The list worker requires every returned row to carry projections, so a storage layout change fails the benchmark instead of silently removing that work.
+
+### Measured endpoints
+
+Workers are compiled plain-Node processes, each started fresh against a seeded root. They compose the production Session store, projection registry, JSONL persistence, SQLite Session query with `openAt: first-search` on an in-memory index, JSON storage and projection cache, the Web projection set, AgentLoop, and the real `SessionController`. Transport lookups, upload registration, default model selection, and the Workspace registry are inert providers, because no measured path depends on them.
+
+| Case | Corpus | Samples | Endpoint |
+|---|---|---:|---|
+| List | 3,000 | 3 | Host boot; first `session.list`; repeated `session.list` in the same process |
+| Content search | 1,000 | 1 | First `session.search`, including the index build; a second query with a different term |
+| Fork | 1,000, separate root | 3, then 1 | `session.fork` return for rank midpoints of ten equal length strata and rank 988; then rank 999 once |
+
+The file must finish within five minutes on standard hosted CI, including seeding, and its last case asserts that limit. Search over 1,000 Sessions and one fork of the longest Session take about 170 s of it, so each has one sample. The list corpus is the largest that fits the remainder. Budgets enforce medians of list boot, first, and repeat, and of the fork median over strata and rank 990; search and the longest fork enforce their single samples. Search and fork use separate 1,000-Session roots, so fork children never reach the search index. Rank 988 is the highest top-level rank at the measured p99 (7,229 events); forking a subagent child adds a lineage lookup, and rank 990 already falls in the 99.3% anchor. Peak RSS of the list, search, and fork processes has a budget with the shared 1.25 headroom and no time scale. Each child's deadline equals the file limit, and each test's deadline outlasts its sequential children, so a failed case reaps every child before the scratch root is removed.
+
+## Calibration evidence
+
+Budgets use standard two-CPU hosted CI expectations (Linux x64, Node 24.21, PR #5403), rounded above the higher recorded median across the AMD EPYC 7763 and EPYC 9V74 runners and multiplied by the shared 1.25 headroom. Samples varied by up to 15% within one run; the runner models differed by up to 25% per endpoint. The hosted-to-reference ratio is 2.4 to 3.4, above the shared reference scale of 2, so reference-machine scaling would reject ordinary hosted runs.
+
+| Endpoint (7763 / 9V74) | Hosted medians | Expectation | Budget |
+|---|---|---:|---:|
+| List 3,000: boot | 2,317.8 / 2,204.5 ms | 2,400 ms | 3,000 ms |
+| List 3,000: first | 3,228.7 / 2,633.6 ms | 3,300 ms | 4,125 ms |
+| List 3,000: repeat | 2,636.4 / 2,235.1 ms | 2,700 ms | 3,375 ms |
+| Search 1,000: first (one sample) | 144,359 / 135,065 ms | 145,000 ms | 181,250 ms |
+| Search 1,000: repeat (one sample) | 3,426.3 / 2,704 ms | 3,500 ms | 4,375 ms |
+| Fork: strata median | 95.5 / 73.5 ms | 100 ms | 125 ms |
+| Fork: p99, rank 988 (7763 only) | 1,066.2 ms | 1,100 ms | 1,375 ms |
+| Fork: longest (one sample) | 19,162.1 / 20,326.7 ms | 20,400 ms | 25,500 ms |
+| Peak RSS: list / search / fork | 742.7 / 1,197.8 / 745.1 MB (higher model) | 750 / 1,200 / 750 MB | 938 / 1,500 / 938 MB |
+
+With seeding in one process, the complete file took 294.5 s on the EPYC 7763 runner (85 s seeding, 26 s listing, 149 s searching, 34 s forking) and 266.6 s on the 9V74. With parallel preparation it took 276.8 s on the 7763: the two preparation processes took 59 s and 65 s, because they share two CPUs with Zstandard and garbage-collection threads, and writing the corpora took 4.6 s. The other cases varied by less than 3% between runs on one runner model. Each additional 1,000 listed Sessions costs about 9 s, so a larger list corpus would leave less than a tenth of the limit for runner variation.
+
+On Apple M5 Pro with Node 26.5, isolated runs measured list 1,000 at 311–358 ms first, list 5,000 at 1,364–1,603 ms first, search at 57 s first, and fork of the longest Session at 6.0–6.4 s. A 10,000-Session list took 3.2–5.4 s for the first call with 1.9 GB peak RSS.
+
+## Alternatives considered
+
+- **Sanitized copies of real Sessions.** Rejected: benchmarks never use recorded user material.
+- **Authoring every Session through `Session.append`.** Rejected: per-append validation and freezing, repeated per Session, made seeding take minutes.
+- **Omitting the projection cache.** Rejected: rows would lack the values the Web list shows, which understates list work.
+- **A separate CI job.** Rejected: benchmarks share one idle standard runner. This file is limited to five minutes, and the job timeout rises from 15 to 20 minutes.
+- **Listing 5,000 Sessions with five samples, and two list sizes.** Rejected: with the other cases the file took 6.6 minutes on standard hosted CI.
+
+## Consequences
+
+The gate covers corpus-scale list, search, and fork costs that single-Session gates cannot see. On standard hosted CI it already shows that content search spends 139 s building its index on first use, that the first list took 5.1 s at 5,000 Sessions in calibration, and that forking the longest Session takes 17.5 to 20.3 s. Seeding writes about 1.5 GB.
+
+Exclusions: the corpus is current-generation only, although most real Sessions are stored in older generations, which list revision hashing and migration-on-read would add. Only 22 distinct bodies exist, so search indexes repeated text. Fork runs without Agent presets or Workspace attachment. Calls are in-process, without Typert transport or a browser. Filesystem caches are not evicted, so cold means a fresh process, not cold storage.

.agents/notes/implemented/testing/2026-09-28-session-corpus-performance.zh.md

@@ -0,0 +1,80 @@
+# Agent Note: 大型 Session 语料的性能基线
+
+Status: implemented
+
+[English](2026-09-28-session-corpus-performance.md) | 中文
+
+## 问题
+
+现有门禁只测量单个长 Session：[打开](2026-09-04-session-open-performance-gate.zh.md)或[续聊](2026-09-06-backend-continuation-performance.zh.md)。列表、内容搜索和 fork 的成本随整个语料或源 Session 长度增长，此前没有门禁覆盖拥有大量长 Session 的用户。由均匀小 Session 组成的语料会低估工作量，因为真实语料呈重尾分布。
+
+## 决定
+
+[session-corpus 基准](../../../../benchmarks/session-corpus/session-corpus.bench.ts)在合成的当前代 Zstandard 语料上测量三个 Web Host 端点。产品实现不做任何改动。
+
+### 工作负载
+
+2026-09-28，一名开发者的本地 DSH home 在 88 个项目目录中保存了 1,650 个 Session。统计只从每个 Session 的最新代中提取聚合计数；内容、身份和路径都没有离开那台机器。[corpus-shape.ts](../../../../benchmarks/session-corpus/corpus-shape.ts) 为三个维度记录 22 个上分位锚点：逻辑事件数（展开紧凑 delta 行）、解压字节数和轮次数。
+
+| 分位 | 事件数 | 逻辑字节数 | 轮次数 |
+|---|---:|---:|---:|
+| p50 | 53 | 86,195 | 1 |
+| p90 | 709 | 1,849,503 | 3 |
+| p99 | 7,229 | 8,479,554 | 18 |
+| max | 84,467 | 48,139,877 | 2,181 |
+
+`n` 个 Session 的语料中，排名 `r` 取不低于分位 `(r + 1) / n` 的第一个锚点，因此合成语料的每个分位都不低于实测分位。所有维度共享同一排名：长 Session 在每个维度上都长。在 1,000 个 Session 时，这种上阶梯采样比实测分位函数多出 16% 的事件和 14% 的字节；更粗的锚点最多多出 40%。
+
+其他实测聚合值也用于确定夹具：每个项目目录 19 个 Session，每六个 Session 中一个 subagent 子会话（实测为 1,650 个中的 288 个），20,000 字符的 system prompt（实测最大事件的中位数为 21.9 KB），每个流式 delta 3.9 个字符，每个 Zstandard 帧 1.6 个事件。轮次中持续加入工具步骤（assistant 消息、工具调用、工具结果）直到达到事件目标；随后对负载字符数做割线搜索，从下方达到字节目标。文本来自 Zipf 分布的合成词表。其存储压缩比为 2.8，低于实测的 3.4，因此夹具存储的压缩字节多于源语料。
+
+### 语料构建
+
+两个准备进程（对应标准托管运行器的每个 CPU）把锚点分成字节均衡的两组。每个进程为每个锚点生成一份事件主体，只压缩一次，并与该锚点的投影记录一起保存。随后播种进程把每个 Session 写成独立的 header 帧加上其锚点的主体。帧压缩和文件写入每次并发八个：每个待完成的压缩都持有一个原生 Zstandard 上下文，不加限制的压缩峰值 RSS 达到 7.7 GB，超过标准托管运行器的内存。共享锚点的 Session 只在 header 身份、创建时间、项目目录和父链接上不同。创建时间是排名的固定置换。列表顺序跟随各主体的最后提示时间，共享锚点的 Session 也共享该时间；这不影响列表成本。事件携带 `Session.append` 记录的字段，紧凑流来自生产 `AssistantStreamAccumulator`，行通过持久化包的 `eventLines` 编码。保存之前，每个准备进程按 `sessionQuery.readSession` 的方式，通过 `Session.create` 回放其生成的主体。
+
+投影缓存属于列表端点：每个冷行都携带缓存的投影值。准备过程通过公开的 `sessionProjectionCache.coldSnapshot` 回写，为每个锚点折叠一次。之后播种进程用 `serializeRecord` 为每个 Session 写入逐记录文档，并把身份重新绑定到各自的 header。通过 domain API 写入时每条记录一次 fsync，10,000 个 Session 耗时 89 s。列表 worker 要求返回的每一行都携带投影，因此存储布局变化会使基准失败，而不是静默去掉这部分工作。
+
+### 测量端点
+
+worker 是编译后的纯 Node 进程，每次都针对已播种根目录全新启动。它们组合生产 Session store、投影注册表、JSONL 持久化、使用 `openAt: first-search` 和内存索引的 SQLite Session query、JSON 存储和投影缓存、Web 投影集合、AgentLoop 以及真实的 `SessionController`。传输查找、上传注册、默认模型选择和 Workspace 注册表使用惰性提供者，因为没有被测路径依赖它们。
+
+| 用例 | 语料 | 样本数 | 终点 |
+|---|---|---:|---|
+| 列表 | 3,000 | 3 | Host 启动；首次 `session.list`；同一进程中重复的 `session.list` |
+| 内容搜索 | 1,000 | 1 | 首次 `session.search`，包括索引构建；使用不同词项的第二次查询 |
+| Fork | 1,000，独立根目录 | 3，然后 1 | 十个等长度分层的中点排名和排名 988 的 `session.fork` 返回；然后排名 999 一次 |
+
+此文件（包括播种）必须在标准托管 CI 上五分钟内完成，其最后一个用例断言该上限。1,000 个 Session 的搜索和最长 Session 的一次 fork 约占其中 170 s，因此两者各取一个样本。列表语料是剩余时间内能容纳的最大规模。预算约束列表启动、首次和重复的中位数，以及 fork 分层中位数和排名 990 的中位数；搜索和最长 fork 约束各自的单个样本。搜索和 fork 使用各自独立的 1,000 个 Session 根目录，因此 fork 子会话不会进入搜索索引。排名 988 是实测 p99（7,229 个事件）上最高的顶层排名；fork subagent 子会话会增加一次谱系查找，而排名 990 已落入 99.3% 锚点。列表、搜索和 fork 进程的峰值 RSS 设有预算，使用共享的 1.25 余量，不做时间缩放。每个子进程的截止时间等于文件上限，每个测试的截止时间长于其顺序子进程之和，因此失败的用例会在删除临时根目录之前回收所有子进程。
+
+## 校准证据
+
+预算使用标准双 CPU 托管 CI 的期望值（Linux x64、Node 24.21、PR #5403），取高于 AMD EPYC 7763 与 EPYC 9V74 运行器中较高记录中位数的整数，再乘以共享的 1.25 余量。同一次运行内的样本差异最多 15%；两种运行器型号在单个端点上最多相差 25%。托管与参考机器的比值为 2.4 到 3.4，高于共享的参考缩放系数 2，因此按参考机器缩放会拒绝正常的托管运行。
+
+| 端点（7763 / 9V74） | 托管中位数 | 期望值 | 预算 |
+|---|---|---:|---:|
+| 列表 3,000：启动 | 2,317.8 / 2,204.5 ms | 2,400 ms | 3,000 ms |
+| 列表 3,000：首次 | 3,228.7 / 2,633.6 ms | 3,300 ms | 4,125 ms |
+| 列表 3,000：重复 | 2,636.4 / 2,235.1 ms | 2,700 ms | 3,375 ms |
+| 搜索 1,000：首次（一个样本） | 144,359 / 135,065 ms | 145,000 ms | 181,250 ms |
+| 搜索 1,000：重复（一个样本） | 3,426.3 / 2,704 ms | 3,500 ms | 4,375 ms |
+| Fork：分层中位数 | 95.5 / 73.5 ms | 100 ms | 125 ms |
+| Fork：p99，排名 988（仅 7763） | 1,066.2 ms | 1,100 ms | 1,375 ms |
+| Fork：最长（一个样本） | 19,162.1 / 20,326.7 ms | 20,400 ms | 25,500 ms |
+| 峰值 RSS：列表 / 搜索 / fork | 742.7 / 1,197.8 / 745.1 MB（较高型号） | 750 / 1,200 / 750 MB | 938 / 1,500 / 938 MB |
+
+单进程播种时，整个文件在 EPYC 7763 运行器上耗时 294.5 s（播种 85 s，列表 26 s，搜索 149 s，fork 34 s），在 9V74 上耗时 266.6 s。并行准备后，它在 7763 上耗时 276.8 s：两个准备进程分别耗时 59 s 和 65 s，因为它们与 Zstandard 和垃圾回收线程共享两个 CPU；写入语料耗时 4.6 s。其他用例在同一运行器型号的多次运行之间差异小于 3%。列表语料每增加 1,000 个 Session 约多花 9 s，因此更大的列表语料会让留给运行器差异的时间不足上限的十分之一。
+
+在 Apple M5 Pro 和 Node 26.5 上的隔离运行中，1,000 个 Session 的首次列表为 311–358 ms，5,000 个 Session 的首次列表为 1,364–1,603 ms，首次搜索为 57 s，fork 最长 Session 为 6.0–6.4 s。10,000 个 Session 的首次列表耗时 3.2–5.4 s，峰值 RSS 为 1.9 GB。
+
+## 考虑过的替代方案
+
+- **脱敏复制真实 Session。** 已否决：基准从不使用录制的用户材料。
+- **通过 `Session.append` 生成每个 Session。** 已否决：逐次追加的验证和冻结在每个 Session 上重复，使播种耗时数分钟。
+- **省略投影缓存。** 已否决：行会缺少 Web 列表显示的值，从而低估列表工作量。
+- **单独的 CI 任务。** 已否决：基准共享一台空闲的标准运行器。此文件限制在五分钟内，任务超时从 15 分钟提高到 20 分钟。
+- **以五个样本列出 5,000 个 Session，并测两种列表规模。** 已否决：加上其他用例后，此文件在标准托管 CI 上耗时 6.6 分钟。
+
+## 后果
+
+该门禁覆盖单 Session 门禁看不到的语料规模列表、搜索和 fork 成本。在标准托管 CI 上，它已经显示：内容搜索在首次使用时花 139 s 构建索引，校准时 5,000 个 Session 的首次列表耗时 5.1 s，fork 最长 Session 耗时 17.5 到 20.3 s。播种写入约 1.5 GB。
+
+排除项：语料只包含当前代，而大多数真实 Session 以旧代存储，这会增加列表的修订哈希和读取时迁移。只有 22 份不同主体，因此搜索索引的是重复文本。Fork 不挂载 Agent preset，也不附加 Workspace。调用在进程内完成，不经过 Typert 传输或浏览器。不驱逐文件系统缓存，因此“冷”指全新进程，而不是冷存储。

.github/workflows/ci.yml

@@ -196,7 +196,7 @@ jobs:
     name: node 24 / benchmarks
     # Wall-clock budgets need an otherwise idle runner, so this job runs the
     # benchmark lane alone instead of joining a concurrent gate aggregate.
-    timeout-minutes: 15
+    timeout-minutes: 20
     steps:
       - uses: actions/checkout@v6
         with:

benchmarks/package.json

@@ -23,13 +23,21 @@
     "@deepseek-ai/dsh-llm": "workspace:*",
     "@deepseek-ai/dsh-sdk-client": "workspace:*",
     "@deepseek-ai/dsh-session": "workspace:*",
+    "@deepseek-ai/dsh-session-format": "workspace:*",
+    "@deepseek-ai/dsh-session-format-catalog": "workspace:*",
+    "@deepseek-ai/dsh-session-format-v3-to-v4": "workspace:*",
     "@deepseek-ai/dsh-session-persistence": "workspace:*",
     "@deepseek-ai/dsh-session-persistence-jsonl": "workspace:*",
     "@deepseek-ai/dsh-session-projection": "workspace:*",
+    "@deepseek-ai/dsh-session-projection-cache": "workspace:*",
     "@deepseek-ai/dsh-session-query": "workspace:*",
+    "@deepseek-ai/dsh-session-query-sqlite": "workspace:*",
     "@deepseek-ai/dsh-session-stats": "workspace:*",
     "@deepseek-ai/dsh-session-title": "workspace:*",
     "@deepseek-ai/dsh-session-turn-outline": "workspace:*",
+    "@deepseek-ai/dsh-storage": "workspace:*",
+    "@deepseek-ai/dsh-storage-domain": "workspace:*",
+    "@deepseek-ai/dsh-storage-json": "workspace:*",
     "@deepseek-ai/dsh-subagent": "workspace:*",
     "@deepseek-ai/dsh-tools": "workspace:*",
     "@deepseek-ai/dsh-token-meter": "workspace:*",

benchmarks/session-corpus/README.i18n.yaml

@@ -0,0 +1,22 @@
+# Bilingual-pair consistency record for README.md (docs/i18n/README.md): per heading
+# section, a hash of its English and Chinese blocks outside code blocks and generated regions.
+# After editing either side, bring the other along and re-record with:
+#   pnpm run verify-translation-pairing --write benchmarks/session-corpus/README.md
+/session-corpus-benchmarks:
+  en: 28ebbfcfa3ea913a
+  zh: 3ebad56e2c97c7c8
+/session-corpus-benchmarks/summary:
+  en: 651a02571a0f6f42
+  zh: dcd33bbc5baa0977
+/session-corpus-benchmarks/table-of-contents:
+  en: 66a769ce60d49ff3
+  zh: 7e0bfa7c598157fa
+/session-corpus-benchmarks/run:
+  en: 2560e832041ae481
+  zh: 2fbbff71135e52ab
+/session-corpus-benchmarks/measurements:
+  en: c357c231c3bfb588
+  zh: 3d901a2caccf2587
+/session-corpus-benchmarks/dev-note:
+  en: b75c8458928b72b9
+  zh: 003ba4e17b4b5269

benchmarks/session-corpus/README.md

@@ -0,0 +1,39 @@
+# Session corpus benchmarks
+
+English | [中文](README.zh.md)
+
+## Summary
+
+Measure the Web Host's Session list, content search, and fork over synthetic corpora in which every Session is at least as long as the same quantile of a measured local DSH corpus. Search and fork run over separate 1,000-Session corpora; the list runs over 3,000, the largest count that keeps this file within five minutes on standard hosted CI. Content search models a deployment that opts in with `openAt: first-search`; shipped profiles disable it. No case uses network services, recorded Sessions, or a browser.
+
+## Table of Contents
+
+- [Run](#run)
+- [Measurements](#measurements)
+- [Dev Note](#dev-note)
+
+<a id="run"></a>
+
+## Run
+
+From the repository root, build the libraries and workers with `pnpm run build:bench`, then run `pnpm exec vitest run --config vitest.bench.config.ts benchmarks/session-corpus/session-corpus.bench.ts`. Do not overlap timing runs with builds or other benchmarks. Seeding writes about 1.5 GB of Zstandard logs under a private temporary root, which the test removes after success or failure.
+
+The test reports every fresh-process sample, the CPU model, available parallelism, platform, and Node version, and enforces median budgets. Its last case fails when the file, including seeding, exceeds five minutes. A failed worker reports its exit, signal, timeout, and stderr tail. The required benchmark lane discovers this file automatically.
+
+<a id="measurements"></a>
+
+## Measurements
+
+[corpus-shape.ts](corpus-shape.ts) holds the measured quantile anchors. [synthetic-corpus.ts](synthetic-corpus.ts) authors one event body per anchor and stores each Session as its own header frame followed by that body. [The Agent Note](../../.agents/notes/implemented/testing/2026-09-28-session-corpus-performance.md) owns the workload derivation, timing endpoints, calibration, and exclusions.
+
+| Case | Corpus | Timed endpoints |
+|---|---|---|
+| List | 3,000 Sessions | Host boot; first and repeated `session.list`, each returning every Session with cached projections |
+| Content search | 1,000 Sessions | First `session.search`, which builds the in-memory index; a second query over the built index |
+| Fork | 1,000 Sessions | `session.fork` return for ten length strata, the p99 Session, and the longest Session |
+
+<a id="dev-note"></a>
+
+## Dev Note
+
+None.

benchmarks/session-corpus/README.zh.md

@@ -0,0 +1,39 @@
+# Session 语料基准
+
+[English](README.md) | 中文
+
+## 概述
+
+在合成语料上测量 Web Host 的 Session 列表、内容搜索和 fork；语料中每个 Session 的长度都不短于一份实测本地 DSH 语料在同一分位上的长度。搜索和 fork 用例分别运行在独立的 1,000 个 Session 语料上；列表用例运行 3,000 个 Session，这是在标准托管 CI 上让此文件五分钟内完成的最大数量。内容搜索模拟通过 `openAt: first-search` 启用该功能的部署；已发布的 profile 默认禁用它。所有用例均不使用网络服务、录制的 Session 或浏览器。
+
+## 目录
+
+- [运行](#run)
+- [测量](#measurements)
+- [开发备注](#dev-note)
+
+<a id="run"></a>
+
+## 运行
+
+在仓库根目录使用 `pnpm run build:bench` 构建库和 worker，然后运行 `pnpm exec vitest run --config vitest.bench.config.ts benchmarks/session-corpus/session-corpus.bench.ts`。不要让计时运行与构建或其他基准重叠。播种过程会在私有临时根目录下写入约 1.5 GB 的 Zstandard 日志，测试成功或失败后都会删除该目录。
+
+测试报告每个新进程样本、CPU 型号、可用并行度、平台和 Node 版本，并约束中位数预算。当整个文件（包括播种）超过五分钟时，最后一个用例失败。worker 失败时报告退出状态、信号、超时和 stderr 末尾。必需基准通道自动发现此文件。
+
+<a id="measurements"></a>
+
+## 测量
+
+[corpus-shape.ts](corpus-shape.ts) 保存实测分位锚点。[synthetic-corpus.ts](synthetic-corpus.ts) 为每个锚点生成一份事件主体，并把每个 Session 存储为独立的 header 帧加上该主体。[Agent Note](../../.agents/notes/implemented/testing/2026-09-28-session-corpus-performance.zh.md) 负责说明工作负载推导、计时终点、校准和排除项。
+
+| 用例 | 语料 | 计时终点 |
+|---|---|---|
+| 列表 | 3,000 个 Session | Host 启动；首次和重复的 `session.list`，每次都返回全部 Session 及其缓存投影 |
+| 内容搜索 | 1,000 个 Session | 构建内存索引的首次 `session.search`；在已构建索引上的第二次查询 |
+| Fork | 1,000 个 Session | 十个长度分层、p99 Session 和最长 Session 的 `session.fork` 返回 |
+
+<a id="dev-note"></a>
+
+## 开发备注
+
+无。

benchmarks/session-corpus/corpus-shape.ts

@@ -0,0 +1,76 @@
+/** Per-Session length targets for the synthetic Session corpus. */
+
+/**
+ * Upper-quantile anchors, in percent, of the measured corpus distribution.
+ * Each anchor pairs with one entry of every dimension table below.
+ */
+const QUANTILES = [10, 20, 30, 40, 50, 60, 70, 75, 80, 85, 90, 92, 94, 96, 97, 98, 99, 99.3, 99.6, 99.8, 99.9, 100] as const
+
+/**
+ * Aggregate quantiles of 1,650 local DSH Sessions measured on 2026-09-28
+ * (newest generation per Session; top-level and subagent Sessions together).
+ * Only counts and byte totals were extracted; no content, identity, or path.
+ */
+const MEASURED = {
+  /** Logical events, expanding compact delta rows to their event count. */
+  events: [22, 22, 27, 32, 53, 63, 84, 154, 251, 387, 709, 975, 1398, 2398, 2652, 3492, 7229, 11620, 21995, 26580, 60699, 84467],
+  /** Decompressed JSONL bytes including the header line. */
+  logicalBytes: [
+    37_233, 46_642, 54_432, 67_432, 86_195, 95_561, 184_414, 572_320, 882_968, 1_250_060, 1_849_503,
+    2_341_019, 2_849_779, 3_902_295, 4_874_426, 6_691_370, 8_479_554, 10_570_582, 13_024_700, 17_646_397,
+    21_225_976, 48_139_877,
+  ],
+  /** Started turns. */
+  turns: [1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 3, 4, 5, 7, 9, 11, 18, 20, 23, 39, 118, 2181],
+} as const
+
+/** Minimum length of one synthetic Session. */
+export interface SessionShape {
+  /** Index of the measured anchor; Sessions with one anchor share one shape. */
+  readonly anchor: number
+  /** Logical events the Session must contain at least. */
+  readonly events: number
+  /** Decompressed log bytes the Session must contain at least. */
+  readonly logicalBytes: number
+  /** Completed turns. */
+  readonly turns: number
+}
+
+/** Number of measured anchors, and therefore of distinct Session shapes. */
+export const ANCHOR_COUNT = QUANTILES.length
+
+/**
+ * Read the length targets of one measured anchor.
+ * @param anchor - zero-based anchor index.
+ * @returns the anchor's minimum event, byte, and turn counts.
+ */
+export function anchorShape(anchor: number): SessionShape {
+  if (!Number.isInteger(anchor) || anchor < 0 || anchor >= ANCHOR_COUNT) {
+    throw new Error(`corpus anchor ${String(anchor)} does not exist`)
+  }
+  return {
+    anchor,
+    events: MEASURED.events[anchor] as number,
+    logicalBytes: MEASURED.logicalBytes[anchor] as number,
+    turns: MEASURED.turns[anchor] as number,
+  }
+}
+
+/**
+ * Select the length targets for one rank of an ascending corpus.
+ *
+ * Every dimension uses the next measured anchor at or above the rank's
+ * quantile, so each quantile of the generated corpus is at least the measured
+ * one. Dimensions share the rank, so a long Session is long in every dimension.
+ * @param rank - zero-based ascending length rank.
+ * @param count - corpus size.
+ * @returns the rank's minimum event, byte, and turn counts.
+ */
+export function sessionShape(rank: number, count: number): SessionShape {
+  const quantile = (100 * (rank + 1)) / count
+  const anchor = QUANTILES.findIndex(value => value >= quantile)
+  if (!Number.isInteger(rank) || rank < 0 || anchor === -1) {
+    throw new Error(`corpus rank ${String(rank)} is outside a ${String(count)}-Session corpus`)
+  }
+  return anchorShape(anchor)
+}

benchmarks/session-corpus/session-corpus.bench.ts

@@ -0,0 +1,281 @@
+/** Required budgets for listing, searching, and forking Sessions in corpora with the measured local length distribution. */
+
+import { mkdtemp, rm } from 'node:fs/promises'
+import { availableParallelism, cpus, tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { performance } from 'node:perf_hooks'
+import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import { runBuiltBenchmarkWorker } from '../support/built-worker.ts'
+import { PERFORMANCE_BUDGET_HEADROOM } from '../support/calibration.ts'
+import { anchorShape, ANCHOR_COUNT, sessionShape } from './corpus-shape.ts'
+import { subagentRank } from './synthetic-corpus.ts'
+import type {
+  AnchorsReport,
+  ForkReport,
+  ListReport,
+  SearchReport,
+  SeedReport,
+  SessionCorpusReport,
+} from './session-corpus.worker.ts'
+
+/**
+ * Sessions per corpus. Search and fork use separate roots, so fork children never reach the search index;
+ * the list corpus is the largest that keeps this file within its time limit.
+ */
+const CORPUS = { search: 1_000, fork: 1_000, list: 3_000 } as const
+/** Upper bound on this file's hosted CI wall time, from seeding through the last case. */
+const FILE_LIMIT_MS = 300_000
+/** Per-child deadline: an endpoint slower than its budget still reports a measurement instead of a kill. */
+const WORKER_TIMEOUT_MS = FILE_LIMIT_MS
+/** Fresh processes per list and fork-strata scenario; the median enforces each budget. */
+const ATTEMPTS = { list: 3, fork: 3 } as const
+/** Concurrent anchor-preparation processes; standard hosted runners have two CPUs. */
+const PREPARATION_PROCESSES = 2
+/** Midpoints of ten equal length strata of the fork corpus. */
+const FORK_STRATA = Array.from({ length: 10 }, (_, stratum) => Math.floor(((stratum + 0.5) * CORPUS.fork) / 10))
+/** The highest top-level rank at the measured p99; forking a subagent child adds a lineage lookup. */
+const FORK_P99_RANK = topLevelRank(Math.ceil(0.99 * CORPUS.fork) - 1)
+/** The longest Session; one sample, because one fork takes about 20 s on standard hosted CI. */
+const FORK_LONGEST_RANK = CORPUS.fork - 1
+const WORKER = join(import.meta.dirname, '..', '.dsh-build', 'session-corpus', 'session-corpus.worker.js')
+
+/** Higher recorded median of the standard EPYC 7763 and EPYC 9V74 hosted runners, per budgeted endpoint. */
+const RECORDED_CI_MS = {
+  listBoot: 2_317.8,
+  listFirst: 3_228.7,
+  listRepeat: 2_636.4,
+  searchFirst: 144_359,
+  searchRepeat: 3_426.3,
+  forkStratumMedian: 95.5,
+  forkP99: 1_066.2,
+  forkLongest: 20_326.7,
+} as const
+/** Hosted CI expectations, rounded above {@link RECORDED_CI_MS} before variance headroom. */
+const EXPECTED_CI_MS = {
+  listBoot: 2_400,
+  listFirst: 3_300,
+  listRepeat: 2_700,
+  searchFirst: 145_000,
+  searchRepeat: 3_500,
+  forkStratumMedian: 100,
+  forkP99: 1_100,
+  forkLongest: 20_400,
+} as const satisfies Record<keyof typeof RECORDED_CI_MS, number>
+/** Higher recorded peak RSS of the two runner models; memory takes headroom but no time scale. */
+const RECORDED_CI_PEAK_RSS_MB = { list: 742.7, search: 1_197.8, fork: 745.1 } as const
+const EXPECTED_CI_PEAK_RSS_MB = {
+  list: 750,
+  search: 1_200,
+  fork: 750,
+} as const satisfies Record<keyof typeof RECORDED_CI_PEAK_RSS_MB, number>
+
+type CorpusName = keyof typeof CORPUS
+
+function topLevelRank(rank: number): number {
+  return subagentRank(rank) ? rank - 1 : rank
+}
+
+function budget(expected: number): number {
+  return Math.ceil(expected * PERFORMANCE_BUDGET_HEADROOM)
+}
+
+/** Test deadline for sequential workers; it outlasts their deadlines, so every child is reaped before cleanup. */
+function serialDeadline(workers: number): number {
+  return workers * WORKER_TIMEOUT_MS + 30_000
+}
+
+async function run<Report extends SessionCorpusReport>(args: readonly string[]): Promise<Report> {
+  const outcome = await runBuiltBenchmarkWorker<Report>({ worker: WORKER, args, timeoutMs: WORKER_TIMEOUT_MS, exposeGc: true })
+  if (outcome.report === undefined) {
+    const stderr = outcome.stderr.trim().split('\n').slice(-20).join('\n')
+    throw new Error(`session-corpus worker ${args[1] ?? ''} failed: exit=${String(outcome.exitCode)}, `
+      + `signal=${String(outcome.signal)}, timedOut=${String(outcome.timedOut)}\n${stderr}`)
+  }
+  return outcome.report
+}
+
+/**
+ * Split anchors into byte-balanced groups, each in ascending order so its byte model starts on small bodies.
+ * @returns one anchor group per preparation process.
+ */
+function anchorGroups(): number[][] {
+  const groups = Array.from({ length: PREPARATION_PROCESSES }, () => ({ bytes: 0, anchors: [] as number[] }))
+  const bySize = Array.from({ length: ANCHOR_COUNT }, (_, anchor) => anchor)
+    .sort((left, right) => anchorShape(right).logicalBytes - anchorShape(left).logicalBytes)
+  for (const anchor of bySize) {
+    const group = groups.reduce((smallest, candidate) => candidate.bytes < smallest.bytes ? candidate : smallest)
+    group.bytes += anchorShape(anchor).logicalBytes
+    group.anchors.push(anchor)
+  }
+  return groups.map(group => group.anchors.sort((left, right) => left - right))
+}
+
+function median(values: readonly number[]): number {
+  return [...values].sort((left, right) => left - right)[Math.floor(values.length / 2)] as number
+}
+
+function rounded(values: readonly number[]): number[] {
+  return values.map(value => Math.round(value * 10) / 10)
+}
+
+function expectWithinBudget(value: number, limit: number): void {
+  expect(value).toBeLessThanOrEqual(limit)
+}
+
+function environment() {
+  return {
+    cpu: cpus()[0]?.model,
+    availableParallelism: availableParallelism(),
+    platform: `${process.platform}-${process.arch}`,
+    node: process.version,
+  }
+}
+
+describe('Session corpus workload', () => {
+  it('is at least the measured distribution at every rank of every corpus', () => {
+    for (const count of Object.values(CORPUS)) {
+      const shapes = Array.from({ length: count }, (_, rank) => sessionShape(rank, count))
+      expect(new Set(shapes.map(shape => shape.anchor)).size).toBe(ANCHOR_COUNT)
+      expect(shapes[Math.floor(count / 2) - 1]).toEqual(anchorShape(4))
+      expect(shapes.at(-1)).toEqual({ anchor: ANCHOR_COUNT - 1, events: 84_467, logicalBytes: 48_139_877, turns: 2_181 })
+      for (let rank = 1; rank < count; rank++) {
+        const [previous, current] = [shapes[rank - 1], shapes[rank]] as const
+        expect(current?.events).toBeGreaterThanOrEqual(previous?.events ?? 0)
+        expect(current?.logicalBytes).toBeGreaterThanOrEqual(previous?.logicalBytes ?? 0)
+      }
+    }
+    expect(() => sessionShape(CORPUS.fork, CORPUS.fork)).toThrow('outside')
+  })
+
+  it('forks top-level Sessions at the strata midpoints, the measured p99, and the maximum', () => {
+    const ranks = [...FORK_STRATA, FORK_P99_RANK, FORK_LONGEST_RANK]
+    expect(ranks.filter(subagentRank)).toEqual([])
+    expect(FORK_STRATA).toEqual([50, 150, 250, 350, 450, 550, 650, 750, 850, 950])
+    expect(sessionShape(FORK_P99_RANK, CORPUS.fork).events).toBe(7_229)
+    expect(sessionShape(FORK_LONGEST_RANK, CORPUS.fork).events).toBe(84_467)
+  })
+
+  it('accepts every recorded hosted median and rejects one a quarter above its expectation', () => {
+    const endpoints = [
+      ...(Object.keys(EXPECTED_CI_MS) as (keyof typeof EXPECTED_CI_MS)[])
+        .map(key => [RECORDED_CI_MS[key], EXPECTED_CI_MS[key]] as const),
+      ...(Object.keys(EXPECTED_CI_PEAK_RSS_MB) as (keyof typeof EXPECTED_CI_PEAK_RSS_MB)[])
+        .map(key => [RECORDED_CI_PEAK_RSS_MB[key], EXPECTED_CI_PEAK_RSS_MB[key]] as const),
+    ]
+    expect(endpoints).toHaveLength(11)
+    for (const [recorded, expected] of endpoints) {
+      expect(recorded).toBeLessThanOrEqual(expected)
+      expectWithinBudget(recorded, budget(expected))
+      expect(() => expectWithinBudget(Math.ceil(expected * 1.26), budget(expected))).toThrow()
+    }
+  })
+})
+
+describe('Session corpus operations', () => {
+  let scratch = ''
+  let started = 0
+  let seeded: readonly SeedReport[] = []
+  const root = (name: CorpusName): string => join(scratch, `corpus-${name}`)
+
+  beforeAll(async () => {
+    started = performance.now()
+    scratch = await mkdtemp(join(tmpdir(), 'dsh-session-corpus-bench-'))
+    const prepared = await Promise.all(anchorGroups().map(group => run<AnchorsReport>([scratch, 'anchors', ...group.map(String)])))
+    seeded = await Promise.all((Object.keys(CORPUS) as CorpusName[])
+      .map(name => run<SeedReport>([scratch, 'seed', name, String(CORPUS[name])])))
+    console.log(JSON.stringify({ benchmark: 'session-corpus/seed', prepared, seeded, environment: environment() }))
+  }, serialDeadline(2))
+
+  afterAll(async () => {
+    if (scratch !== '') await rm(scratch, { recursive: true, force: true })
+  })
+
+  it('writes corpora at least as long as the measured distribution', () => {
+    expect(seeded.map(report => report.name).sort()).toEqual(Object.keys(CORPUS).sort())
+    for (const { name, corpus } of seeded) {
+      const count = CORPUS[name as CorpusName]
+      const shapes = Array.from({ length: count }, (_, rank) => sessionShape(rank, count))
+      expect(corpus.sessions).toBe(count)
+      expect(corpus.distinctBodies).toBe(ANCHOR_COUNT)
+      expect(corpus.events).toBeGreaterThanOrEqual(shapes.reduce((sum, shape) => sum + shape.events, 0))
+      expect(corpus.logicalBytes).toBeGreaterThanOrEqual(shapes.reduce((sum, shape) => sum + shape.logicalBytes, 0))
+    }
+  })
+
+  it(`lists ${String(CORPUS.list)} Sessions after a cold Host boot`, async () => {
+    const budgets = {
+      bootMs: budget(EXPECTED_CI_MS.listBoot),
+      firstMs: budget(EXPECTED_CI_MS.listFirst),
+      repeatMs: budget(EXPECTED_CI_MS.listRepeat),
+      peakRssMb: budget(EXPECTED_CI_PEAK_RSS_MB.list),
+    }
+    const reports: ListReport[] = []
+    for (let attempt = 0; attempt < ATTEMPTS.list; attempt++) {
+      reports.push(await run<ListReport>([root('list'), 'list', String(CORPUS.list)]))
+    }
+    const samples = {
+      bootMs: rounded(reports.map(report => report.bootMs)),
+      firstMs: rounded(reports.map(report => report.firstMs)),
+      repeatMs: rounded(reports.map(report => report.repeatMs)),
+      peakRssMb: reports.map(report => report.memory.peakRssMb),
+      heapUsedMb: reports.map(report => report.memory.heapUsedMb),
+    }
+    console.log(JSON.stringify({ benchmark: `session-corpus/list-${String(CORPUS.list)}`, samples, budgets, environment: environment() }))
+    for (const report of reports) expect(report.itemsWithProjections).toBe(CORPUS.list)
+    expectWithinBudget(median(samples.bootMs), budgets.bootMs)
+    expectWithinBudget(median(samples.firstMs), budgets.firstMs)
+    expectWithinBudget(median(samples.repeatMs), budgets.repeatMs)
+    expectWithinBudget(median(samples.peakRssMb), budgets.peakRssMb)
+  }, serialDeadline(ATTEMPTS.list))
+
+  it(`searches ${String(CORPUS.search)} Sessions with a cold and then a built content index`, async () => {
+    const budgets = {
+      firstMs: budget(EXPECTED_CI_MS.searchFirst),
+      repeatMs: budget(EXPECTED_CI_MS.searchRepeat),
+      peakRssMb: budget(EXPECTED_CI_PEAK_RSS_MB.search),
+    }
+    // One sample: the cold index build dominates this file's time.
+    const report = await run<SearchReport>([root('search'), 'search'])
+    console.log(JSON.stringify({ benchmark: 'session-corpus/search', report, budgets, environment: environment() }))
+    expectWithinBudget(report.firstMs, budgets.firstMs)
+    expectWithinBudget(report.repeatMs, budgets.repeatMs)
+    expectWithinBudget(report.memory.peakRssMb, budgets.peakRssMb)
+  }, serialDeadline(1))
+
+  it(`forks Sessions across the length distribution of ${String(CORPUS.fork)} Sessions`, async () => {
+    const budgets = {
+      stratumMedianMs: budget(EXPECTED_CI_MS.forkStratumMedian),
+      p99Ms: budget(EXPECTED_CI_MS.forkP99),
+      longestMs: budget(EXPECTED_CI_MS.forkLongest),
+      peakRssMb: budget(EXPECTED_CI_PEAK_RSS_MB.fork),
+    }
+    const reports: ForkReport[] = []
+    for (let attempt = 0; attempt < ATTEMPTS.fork; attempt++) {
+      reports.push(await run<ForkReport>([root('fork'), 'fork', ...[...FORK_STRATA, FORK_P99_RANK].map(String)]))
+    }
+    const longest = await run<ForkReport>([root('fork'), 'fork', String(FORK_LONGEST_RANK)])
+    const forkMs = (report: ForkReport, rank: number): number => {
+      const fork = report.forks.find(entry => entry.rank === rank)
+      if (fork === undefined) throw new Error(`fork report omits rank ${String(rank)}`)
+      return fork.forkMs
+    }
+    const samples = {
+      stratumMedianMs: rounded(reports.map(report => median(FORK_STRATA.map(rank => forkMs(report, rank))))),
+      p99Ms: rounded(reports.map(report => forkMs(report, FORK_P99_RANK))),
+      longestMs: Math.round(forkMs(longest, FORK_LONGEST_RANK) * 10) / 10,
+      perRank: reports.map(report => report.forks.map(({ rank, sourceEvents, forkMs }) => ({ rank, sourceEvents, forkMs: Math.round(forkMs) }))),
+      peakRssMb: [...reports, longest].map(report => report.memory.peakRssMb),
+    }
+    console.log(JSON.stringify({ benchmark: 'session-corpus/fork', samples, budgets, environment: environment() }))
+    expectWithinBudget(median(samples.stratumMedianMs), budgets.stratumMedianMs)
+    expectWithinBudget(median(samples.p99Ms), budgets.p99Ms)
+    expectWithinBudget(samples.longestMs, budgets.longestMs)
+    expectWithinBudget(Math.max(...samples.peakRssMb), budgets.peakRssMb)
+  }, serialDeadline(ATTEMPTS.fork + 1))
+
+  it(`completes within ${String(FILE_LIMIT_MS / 60_000)} minutes`, () => {
+    const elapsedMs = performance.now() - started
+    console.log(JSON.stringify({ benchmark: 'session-corpus/total', elapsedMs: Math.round(elapsedMs), limitMs: FILE_LIMIT_MS }))
+    expectWithinBudget(elapsedMs, FILE_LIMIT_MS)
+  })
+})

benchmarks/session-corpus/session-corpus.worker.ts

@@ -0,0 +1,368 @@
+/** Isolated worker that seeds the synthetic corpus or measures one Session-controller operation over it. */
+
+import { mkdir, readFile, writeFile } from 'node:fs/promises'
+import { join } from 'node:path'
+import { performance } from 'node:perf_hooks'
+import { scheduler } from 'node:timers/promises'
+import { Context } from '@deepseek-ai/cordis'
+import AgentLoop from '@deepseek-ai/dsh-agent-loop'
+import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
+import { agentPresetProjectionDefinition } from '@deepseek-ai/dsh-agent-preset-registry'
+import SessionController from '@deepseek-ai/dsh-api-session-controller'
+import { currentSessionMessageProjections } from '@deepseek-ai/dsh-session-format-catalog/message-projections'
+import { Session, SessionLogOffset } from '@deepseek-ai/dsh-session'
+import type { SessionId } from '@deepseek-ai/dsh-session'
+import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
+import SessionProjectionCache, { checkpointRecord, projectionCacheDomainSpec } from '@deepseek-ai/dsh-session-projection-cache'
+import type { CheckpointRecord } from '@deepseek-ai/dsh-session-projection-cache'
+import SqliteSessionQueryEngine from '@deepseek-ai/dsh-session-query-sqlite'
+import * as SessionStatsPlugin from '@deepseek-ai/dsh-session-stats'
+import SessionTitleService from '@deepseek-ai/dsh-session-title'
+import * as SessionTurnOutlinePlugin from '@deepseek-ai/dsh-session-turn-outline'
+import Storage from '@deepseek-ai/dsh-storage'
+import * as StorageDomain from '@deepseek-ai/dsh-storage-domain'
+import type { KvTable } from '@deepseek-ai/dsh-storage-domain'
+import * as StorageJson from '@deepseek-ai/dsh-storage-json'
+import TokenMeter from '@deepseek-ai/dsh-token-meter'
+import { assertBuiltBenchmarkRuntime } from '../support/built-worker.ts'
+import { serializeRecord } from '../../packages/storage/storage-json/src/format.ts'
+import { ANCHOR_COUNT } from './corpus-shape.ts'
+import {
+  anchorHeader,
+  corpusSessionId,
+  forEachConcurrently,
+  SEARCH_QUERIES,
+  SyntheticCorpusWriter,
+  type AnchorBodyFacts,
+  type SyntheticCorpusFacts,
+  type WrittenSession,
+} from './synthetic-corpus.ts'
+
+/** Process memory after the measured endpoint, with its result still reachable. */
+export interface CorpusMemory {
+  readonly heapUsedMb: number
+  readonly rssMb: number
+  readonly peakRssMb: number
+}
+
+/** Anchor preparation report: authoring, replay validation, and projection folding of one anchor group. */
+export interface AnchorsReport {
+  readonly mode: 'anchors'
+  readonly anchors: readonly number[]
+  readonly prepareMs: number
+}
+
+/** Seeding report: facts of one named corpus written from prepared anchors, and the time spent writing it. */
+export interface SeedReport {
+  readonly mode: 'seed'
+  readonly name: string
+  readonly corpus: SyntheticCorpusFacts
+  readonly seedMs: number
+}
+
+/** List report: Host boot, then a cold and a repeated Session list. */
+export interface ListReport {
+  readonly mode: 'list'
+  readonly bootMs: number
+  readonly firstMs: number
+  readonly repeatMs: number
+  readonly items: number
+  readonly itemsWithProjections: number
+  readonly memory: CorpusMemory
+}
+
+/** Search report: the first search builds the in-memory index; the second reuses it. */
+export interface SearchReport {
+  readonly mode: 'search'
+  readonly bootMs: number
+  readonly firstMs: number
+  readonly repeatMs: number
+  readonly firstItems: number
+  readonly repeatItems: number
+  readonly memory: CorpusMemory
+}
+
+/** Fork report: one fork of each requested source, in request order. */
+export interface ForkReport {
+  readonly mode: 'fork'
+  readonly bootMs: number
+  readonly forks: readonly {
+    readonly rank: number
+    readonly sourceEvents: number
+    readonly forkMs: number
+  }[]
+  readonly memory: CorpusMemory
+}
+
+/** Any worker report. */
+export type SessionCorpusReport = AnchorsReport | SeedReport | ListReport | SearchReport | ForkReport
+
+const BENCH_MODEL = { provider: 'bench', model: 'bench' } as const
+/** Bound on waiting for fire-and-forget projection-cache write-backs after seeding. */
+const SEED_DRAIN_TIMEOUT_MS = 120_000
+
+function megabytes(bytes: number): number {
+  return Math.round(bytes / 104_857.6) / 10
+}
+
+async function memory(): Promise<CorpusMemory> {
+  const gc = (globalThis as typeof globalThis & { gc?: () => void }).gc
+  if (gc === undefined) throw new Error('Session corpus benchmark requires --expose-gc')
+  gc()
+  await scheduler.yield()
+  gc()
+  const usage = process.memoryUsage()
+  return {
+    heapUsedMb: megabytes(usage.heapUsed),
+    rssMb: megabytes(usage.rss),
+    peakRssMb: Math.round(process.resourceUsage().maxRSS / 102.4) / 10,
+  }
+}
+
+/** Mount the shipped Web Host's Session list, search, and fork dependencies. */
+async function mountHost(root: string): Promise<{ readonly ctx: Context; readonly controller: SessionController }> {
+  const ctx = new Context()
+  await mountAgentLoopTestDependencies(ctx)
+  await ctx.plugin(JsonlSessionPersistence, { root: join(root, 'sessions'), compression: 'zstd' })
+  await ctx.plugin(Storage)
+  await ctx.plugin(StorageJson, { root: join(root, 'storages') })
+  await ctx.plugin(StorageDomain, { backend: 'json' })
+  await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 200, writeIntervalMs: 5_000 })
+  // Content search is opt-in; enabled deployments defer the in-memory index to the first search.
+  await ctx.plugin(SqliteSessionQueryEngine, { path: ':memory:', openAt: 'first-search' })
+  ctx.sessionProjections.register(agentPresetProjectionDefinition)
+  await ctx.plugin(SessionTitleService, { fallbackMaxWords: 5, fallbackMaxBytes: 40, maxTitleBytes: 80 })
+  await ctx.plugin(SessionStatsPlugin)
+  await ctx.plugin(SessionTurnOutlinePlugin)
+  await ctx.plugin(TokenMeter)
+  await ctx.plugin(AgentLoop, { agents: [] })
+  // Transport, upload, model-default, and Workspace services are outside every measured path.
+  const dispose = (): void => {}
+  ctx.provide('typert', { lookups: { configure: () => dispose }, contexts: { configureHost: () => dispose } } as never)
+  ctx.provide('fileUploads', { registerAgentResolver: () => dispose } as never)
+  ctx.provide('agentDefaultModel', { currentSelection: () => BENCH_MODEL } as never)
+  ctx.provide('workspaceRegistry', { list: () => [], archivedSessionIds: [] } as never)
+  const controller = new SessionController(ctx, { nativeOpen: false }, { canOpenPath: () => false })
+  return { ctx, controller }
+}
+
+/** Run one operation against the projection-cache table stored under `root`. */
+async function withCacheTable<Value>(
+  root: string,
+  operation: (table: KvTable<SessionId, CheckpointRecord>) => Promise<Value>,
+): Promise<Value> {
+  const ctx = new Context()
+  try {
+    await ctx.plugin(Storage)
+    await ctx.plugin(StorageJson, { root: join(root, 'storages') })
+    await ctx.plugin(StorageDomain, { backend: 'json' })
+    const domain = await ctx.storageDomain.open(projectionCacheDomainSpec)
+    try {
+      return await operation(domain.table('sessions'))
+    } finally {
+      await domain.close()
+    }
+  } finally {
+    await ctx.fiber.dispose()
+  }
+}
+
+function anchorFactsPath(root: string, anchor: number): string {
+  return join(root, 'bodies', `anchor-${String(anchor).padStart(2, '0')}.json`)
+}
+
+/** Author, validate, and fold one group of anchors, saving bodies and projection records for `seed`. */
+async function prepareAnchors(root: string, anchors: readonly number[]): Promise<AnchorsReport> {
+  const started = performance.now()
+  const writer = new SyntheticCorpusWriter()
+  const foldRoot = join(root, `fold-${anchors.join('-')}`)
+  const facts = new Map<number, AnchorBodyFacts>()
+  const host = await mountHost(foldRoot)
+  try {
+    for (const anchor of anchors) {
+      const header = anchorHeader(anchor)
+      const log = await writer.author(anchor)
+      // Replay validation, as in `sessionQuery.readSession`, rejects an invalid authored body before any write.
+      Session.create(header.id, log, header, SessionLogOffset(0), currentSessionMessageProjections)
+      // The production cold-read write-back folds each shared body once.
+      host.ctx.sessionProjectionCache.coldSnapshot(header, SessionLogOffset(0), log)
+      facts.set(anchor, await writer.saveBody(join(root, 'bodies'), anchor))
+    }
+    // Write-backs are fire-and-forget; the cache serves a record only after it is durable.
+    const deadline = performance.now() + SEED_DRAIN_TIMEOUT_MS
+    while (anchors.some(anchor => host.ctx.sessionProjectionCache.cachedSnapshot(anchorHeader(anchor)) === undefined)) {
+      if (performance.now() > deadline) throw new Error('projection-cache write-back did not become durable')
+      await scheduler.wait(20)
+    }
+  } finally {
+    await host.ctx.fiber.dispose()
+  }
+  await withCacheTable(foldRoot, async (table) => {
+    for (const anchor of anchors) {
+      const record = table.get(anchorHeader(anchor).id)
+      await writeFile(anchorFactsPath(root, anchor), JSON.stringify({ ...facts.get(anchor), record }))
+    }
+  })
+  return { mode: 'anchors', anchors, prepareMs: performance.now() - started }
+}
+
+/** Parse one prepared anchor file written by a sibling `anchors` process. */
+async function readPreparedAnchor(root: string, anchor: number): Promise<AnchorBodyFacts & { readonly record: CheckpointRecord }> {
+  const value: unknown = JSON.parse(await readFile(anchorFactsPath(root, anchor), 'utf8'))
+  if (typeof value !== 'object' || value === null) throw new Error(`prepared anchor ${String(anchor)} is not an object`)
+  const { frameCount, events, logicalBytes, record } = value as Record<string, unknown>
+  for (const count of [frameCount, events, logicalBytes]) {
+    if (!Number.isSafeInteger(count)) throw new Error(`prepared anchor ${String(anchor)} has invalid body facts`)
+  }
+  return {
+    frameCount: frameCount as number,
+    events: events as number,
+    logicalBytes: logicalBytes as number,
+    record: checkpointRecord.parse(record),
+  }
+}
+
+/** Write one named corpus from prepared anchors, rebinding each projection record to its Session header. */
+async function seed(root: string, name: string, count: number): Promise<SeedReport> {
+  const started = performance.now()
+  const writer = new SyntheticCorpusWriter()
+  const records = new Map<number, CheckpointRecord>()
+  for (let anchor = 0; anchor < ANCHOR_COUNT; anchor++) {
+    const { record, ...facts } = await readPreparedAnchor(root, anchor)
+    await writer.loadBody(join(root, 'bodies'), anchor, facts)
+    records.set(anchor, record)
+  }
+  const corpusRoot = join(root, `corpus-${name}`)
+  const written = await writer.writeCorpus(join(corpusRoot, 'sessions'), count)
+  // Sessions of one anchor share events, so their records differ only in the header-bound identity.
+  // Per-record documents are written directly: the domain's one-fsync-per-put chain would dominate seeding,
+  // and the list endpoint rejects a corpus whose rows the cache does not serve.
+  const table = join(corpusRoot, 'storages', projectionCacheDomainSpec.name, 'sessions')
+  await mkdir(table, { recursive: true })
+  await forEachConcurrently(written.sessions.length, async (rank) => {
+    const { header, anchor } = written.sessions[rank] as WrittenSession
+    const source = records.get(anchor)
+    if (source === undefined) throw new Error(`corpus anchor ${String(anchor)} has no folded projection record`)
+    await writeFile(join(table, `${header.id}.json`), serializeRecord(projectionCacheDomainSpec.version, {
+      identity: { ...source.identity, createdAt: header.createdAt, cwd: header.cwd },
+      rows: source.rows,
+    }))
+  })
+  return { mode: 'seed', name, corpus: written.facts, seedMs: performance.now() - started }
+}
+
+async function measureList(root: string, expected: number): Promise<ListReport> {
+  const started = performance.now()
+  const { ctx, controller } = await mountHost(root)
+  try {
+    const booted = performance.now()
+    const signal = new AbortController().signal
+    const first = await controller.list({}, signal)
+    const firstDone = performance.now()
+    const repeated = await controller.list({}, signal)
+    const done = performance.now()
+    if (first.items.length !== expected || repeated.items.length !== expected) {
+      throw new Error(`Session list returned ${String(first.items.length)} of ${String(expected)} Sessions`)
+    }
+    return {
+      mode: 'list',
+      bootMs: booted - started,
+      firstMs: firstDone - booted,
+      repeatMs: done - firstDone,
+      items: first.items.length,
+      itemsWithProjections: first.items.filter(item => item.projections !== undefined).length,
+      memory: await memory(),
+    }
+  } finally {
+    await ctx.fiber.dispose()
+  }
+}
+
+async function measureSearch(root: string): Promise<SearchReport> {
+  const started = performance.now()
+  const { ctx, controller } = await mountHost(root)
+  try {
+    const booted = performance.now()
+    const signal = new AbortController().signal
+    const first = await controller.search({ query: SEARCH_QUERIES[0] }, signal)
+    const firstDone = performance.now()
+    const repeated = await controller.search({ query: SEARCH_QUERIES[1] }, signal)
+    const done = performance.now()
+    if (first.items.length === 0 || repeated.items.length === 0) {
+      throw new Error('Session search matched no synthetic Session')
+    }
+    return {
+      mode: 'search',
+      bootMs: booted - started,
+      firstMs: firstDone - booted,
+      repeatMs: done - firstDone,
+      firstItems: first.items.length,
+      repeatItems: repeated.items.length,
+      memory: await memory(),
+    }
+  } finally {
+    await ctx.fiber.dispose()
+  }
+}
+
+async function measureFork(root: string, ranks: readonly number[]): Promise<ForkReport> {
+  const started = performance.now()
+  const { ctx, controller } = await mountHost(root)
+  try {
+    const booted = performance.now()
+    const forks: ForkReport['forks'][number][] = []
+    for (const rank of ranks) {
+      const forkStarted = performance.now()
+      const { sessionId } = await controller.fork({ sessionId: corpusSessionId(rank) })
+      const forkMs = performance.now() - forkStarted
+      const child = ctx.sessions.get(sessionId)
+      if (child === undefined) throw new Error(`fork of rank ${String(rank)} did not publish its child`)
+      forks.push({ rank, sourceEvents: child.inheritedEventCount, forkMs })
+    }
+    return { mode: 'fork', bootMs: booted - started, forks, memory: await memory() }
+  } finally {
+    await ctx.fiber.dispose()
+  }
+}
+
+assertBuiltBenchmarkRuntime(import.meta.url, Object.fromEntries([
+  '@deepseek-ai/dsh-api-session-controller',
+  '@deepseek-ai/dsh-session-persistence-jsonl',
+  '@deepseek-ai/dsh-session-query-sqlite',
+  '@deepseek-ai/dsh-session-projection-cache',
+].map(name => [name, import.meta.resolve(name)])))
+
+const USAGE = 'usage: session-corpus.worker.js <root> <anchors anchors...|seed name count|list count|search|fork ranks...>'
+const [root, mode, ...rest] = process.argv.slice(2)
+if (root === undefined) throw new Error(USAGE)
+
+function counts(values: readonly string[]): number[] {
+  const parsed = values.map(Number)
+  if (parsed.some(value => !Number.isSafeInteger(value) || value < 0)) throw new Error(USAGE)
+  return parsed
+}
+
+let report: SessionCorpusReport
+switch (mode) {
+  case 'anchors':
+    report = await prepareAnchors(root, counts(rest))
+    break
+  case 'seed': {
+    const [name, count] = rest
+    if (name === undefined || !/^[a-z]+$/.test(name) || count === undefined) throw new Error(USAGE)
+    report = await seed(root, name, counts([count])[0] as number)
+    break
+  }
+  case 'list':
+    report = await measureList(root, counts(rest)[0] ?? 0)
+    break
+  case 'search':
+    report = await measureSearch(root)
+    break
+  case 'fork':
+    report = await measureFork(root, counts(rest))
+    break
+  default:
+    throw new Error(`unknown Session corpus mode ${String(mode)}`)
+}
+process.stdout.write(JSON.stringify(report) + '\n')

benchmarks/session-corpus/synthetic-corpus.ts

@@ -0,0 +1,484 @@
+/** Deterministic current-generation Session corpus whose length distribution follows `corpus-shape.ts`. */
+
+import { mkdir, readFile, writeFile } from 'node:fs/promises'
+import { dirname, join } from 'node:path'
+import { AssistantStreamAccumulator } from '@deepseek-ai/dsh-llm/assistant-stream'
+import { MessageId, ToolCallId } from '@deepseek-ai/dsh-llm'
+import type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm'
+import { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionEventMap, SessionEventType, SessionHeader } from '@deepseek-ai/dsh-session'
+import {
+  eventLines,
+  generationLogPath,
+  toHeaderLine,
+} from '../../packages/session/session-persistence-jsonl/src/format.ts'
+import { compressZstdFrame } from '../../packages/session/session-persistence-jsonl/src/zstd.ts'
+import { anchorShape, sessionShape, type SessionShape } from './corpus-shape.ts'
+
+/** Time of every body's first event; Sessions are created in the preceding minutes. */
+const TIME_ZERO = 1_700_000_000_000
+/** Prime stride that permutes creation order independently of length rank. */
+const CREATION_STRIDE = 7_919
+/** Concurrent frame compressions and file writes while materializing a corpus. */
+const IO_CONCURRENCY = 8
+/** Measured local Sessions per project directory: 1,650 Sessions in 88 directories. */
+const SESSIONS_PER_PROJECT = 19
+/** Every sixth Session is a subagent child; the measured share is 288 of 1,650. */
+const CHILD_EVERY = 6
+/** Turn-one system prompt size; the measured median largest event is 21.9 KB. */
+const SYSTEM_PROMPT_CHARS = 20_000
+/** Characters per streamed text or reasoning delta; the measured mean is 3.9. */
+const DELTA_CHARS = 4
+/** Share of remaining payload characters by message slot. */
+const PAYLOAD_SHARE = { user: 0.1, reasoning: 0.2, text: 0.25, toolResult: 0.45 } as const
+/** Distinct words in the synthetic vocabulary and total words in the shared text pool. */
+const VOCABULARY = { words: 8_192, poolWords: 2_000_000 } as const
+/** Word ranks used by the measured searches: one frequent and one uncommon token. */
+export const SEARCH_QUERIES = [vocabularyWord(40), vocabularyWord(1_500)] as const
+
+/** Physical and logical facts of one written corpus. */
+export interface SyntheticCorpusFacts {
+  /** Logical Sessions. */
+  readonly sessions: number
+  /** Distinct event bodies; Sessions of one anchor share one body. */
+  readonly distinctBodies: number
+  /** Project directories. */
+  readonly projects: number
+  /** Logical events across all Sessions. */
+  readonly events: number
+  /** Decompressed JSONL bytes across all Sessions. */
+  readonly logicalBytes: number
+  /** Stored Zstandard bytes across all Sessions. */
+  readonly compressedBytes: number
+  /** Zstandard frames across all Sessions. */
+  readonly frames: number
+}
+
+/** One written Session. */
+export interface WrittenSession {
+  /** Stored header. */
+  readonly header: SessionHeader
+  /** Measured anchor whose body the Session stores. */
+  readonly anchor: number
+}
+
+/**
+ * Identify one Session by length rank.
+ * @param rank - zero-based ascending length rank.
+ * @returns stable Session id.
+ */
+export function corpusSessionId(rank: number): SessionId {
+  return SessionId('bench-' + String(rank).padStart(5, '0'))
+}
+
+function vocabularyWord(rank: number): string {
+  let value = rank + VOCABULARY.words
+  let word = ''
+  const letters = 'etaoinsrhdlucmfywgpbvkxqjz'
+  while (value > 0) {
+    word += letters[value % letters.length]
+    value = Math.floor(value / letters.length)
+  }
+  return word
+}
+
+/** Mulberry32: a small deterministic generator for fixture construction only. */
+function random(seed: number): () => number {
+  let state = seed >>> 0
+  return () => {
+    state = (state + 0x6D2B79F5) >>> 0
+    let value = state
+    value = Math.imul(value ^ (value >>> 15), value | 1)
+    value ^= value + Math.imul(value ^ (value >>> 7), value | 61)
+    return ((value ^ (value >>> 14)) >>> 0) / 4_294_967_296
+  }
+}
+
+/** Zipf-distributed prose from which every message body takes a deterministic slice. */
+class TextPool {
+  private readonly text: string
+
+  constructor() {
+    const next = random(1)
+    const cumulative: number[] = []
+    let total = 0
+    for (let rank = 1; rank <= VOCABULARY.words; rank++) {
+      total += 1 / rank
+      cumulative.push(total)
+    }
+    const words = Array.from({ length: VOCABULARY.words }, (_, rank) => vocabularyWord(rank))
+    const parts: string[] = []
+    for (let index = 0; index < VOCABULARY.poolWords; index++) {
+      const target = next() * total
+      let low = 0
+      let high = cumulative.length - 1
+      while (low < high) {
+        const middle = (low + high) >>> 1
+        if ((cumulative[middle] as number) < target) low = middle + 1
+        else high = middle
+      }
+      parts.push(words[low] as string, index % 13 === 12 ? '.\n' : ' ')
+    }
+    this.text = parts.join('')
+  }
+
+  slice(next: () => number, length: number): string {
+    if (length <= 0) return ''
+    const span = Math.min(length, this.text.length)
+    const start = Math.floor(next() * (this.text.length - span))
+    return this.text.slice(start, start + span)
+  }
+}
+
+interface PayloadPlan {
+  readonly userChars: number
+  readonly reasoningChars: number
+  readonly textChars: number
+  readonly resultChars: number
+}
+
+function planPayload(shape: SessionShape, toolSteps: number, payloadChars: number): PayloadPlan {
+  const replies = toolSteps + shape.turns
+  const textShare = toolSteps === 0 ? PAYLOAD_SHARE.text + PAYLOAD_SHARE.toolResult : PAYLOAD_SHARE.text
+  return {
+    userChars: Math.ceil((payloadChars * PAYLOAD_SHARE.user) / shape.turns),
+    reasoningChars: Math.ceil((payloadChars * PAYLOAD_SHARE.reasoning) / replies),
+    textChars: Math.ceil((payloadChars * textShare) / replies),
+    resultChars: toolSteps === 0 ? 0 : Math.ceil((payloadChars * PAYLOAD_SHARE.toolResult) / toolSteps),
+  }
+}
+
+function streamChunks(content: readonly ContentBlock[], finish: 'stop' | 'tool-calls'): StreamChunk[] {
+  const chunks: StreamChunk[] = []
+  content.forEach((block, index) => {
+    chunks.push({ type: 'block-start', index, blockType: block.type })
+    if (block.type === 'text' || block.type === 'reasoning') {
+      const type = block.type === 'text' ? 'text-delta' : 'reasoning-delta'
+      for (let offset = 0; offset < block.text.length; offset += DELTA_CHARS) {
+        chunks.push({ type, index, text: block.text.slice(offset, offset + DELTA_CHARS) })
+      }
+    } else if (block.type === 'tool-call') {
+      chunks.push({ type: 'tool-call-delta', index, id: block.id, name: block.name, argumentsDelta: block.arguments })
+    }
+    chunks.push({ type: 'block-end', index, block })
+  })
+  chunks.push({ type: 'usage', usage: { inputTokens: 20_000, outputTokens: 500 } })
+  chunks.push({ type: 'finish', reason: { kind: finish } })
+  return chunks
+}
+
+/** Raw log builder emitting the fields `Session.append` records, without its per-append validation. */
+class RawLog {
+  readonly events: SessionEvent[] = []
+  readonly frameStarts = new Set<number>()
+
+  constructor(private readonly createdAt: number) {}
+
+  get seq(): number {
+    return this.events.length
+  }
+
+  /** Start a durable flush batch at the next event. */
+  flush(): void {
+    this.frameStarts.add(this.events.length)
+  }
+
+  append<T extends SessionEventType>(
+    type: T,
+    data: SessionEventMap[T],
+    surface: { readonly surfaceOp?: 'append'; readonly sourceEventSeqs?: readonly number[] } = {},
+  ): number {
+    const seq = this.events.length
+    this.events.push({ type, seq, time: this.createdAt + seq * 1_000, data, ...surface } as SessionEvent<T>)
+    return seq
+  }
+}
+
+/** Authors one Session with production stream compaction; replay validation belongs to the fixture test. */
+function authorEvents(
+  shape: SessionShape,
+  payloadChars: number,
+  pool: TextPool,
+  seed: number,
+): { readonly events: SessionEvent[]; readonly frameStarts: ReadonlySet<number> } {
+  const next = random(seed)
+  const toolSteps = Math.max(0, Math.ceil((shape.events - 6 * shape.turns - 1) / 5))
+  const plan = planPayload(shape, toolSteps, payloadChars)
+  const log = new RawLog(TIME_ZERO)
+  for (let turn = 1; turn <= shape.turns; turn++) {
+    const turnSteps = Math.floor(toolSteps / shape.turns) + (turn <= toolSteps % shape.turns ? 1 : 0)
+    log.flush()
+    log.append('turn/start', { turn })
+    log.append('step/start', { turn, step: 1 })
+    if (turn === 1) log.append('system/message', {
+      turn, step: 1,
+      message: {
+        id: MessageId('system-head'), role: 'system',
+        content: [{ type: 'text', text: pool.slice(next, SYSTEM_PROMPT_CHARS) }],
+        source: { kind: 'system-prompt' },
+      },
+    }, { surfaceOp: 'append' })
+    log.append('user/message', {
+      id: MessageId('prompt-' + String(turn)), role: 'user',
+      content: [{ type: 'text', text: pool.slice(next, plan.userChars) }], source: { kind: 'user' },
+    }, { surfaceOp: 'append' })
+    for (let step = 1; step <= turnSteps + 1; step++) {
+      if (step > 1) log.append('step/start', { turn, step })
+      const callId = step <= turnSteps ? ToolCallId(`call-${String(turn)}-${String(step)}`) : undefined
+      const content: ContentBlock[] = [
+        { type: 'reasoning', text: pool.slice(next, plan.reasoningChars) },
+        { type: 'text', text: pool.slice(next, plan.textChars) },
+        ...callId === undefined ? [] : [{
+          type: 'tool-call' as const, id: callId, name: 'read_file',
+          arguments: JSON.stringify({ path: `src/module-${String(turn)}-${String(step)}.ts` }),
+        }],
+      ]
+      const stream = new AssistantStreamAccumulator()
+      const time = TIME_ZERO + log.seq * 1_000
+      for (const [index, chunk] of streamChunks(content, callId === undefined ? 'stop' : 'tool-calls').entries()) {
+        stream.push({ time: time + index, chunk })
+      }
+      log.flush()
+      log.append('assistant/message', {
+        turn, step,
+        message: {
+          id: MessageId(`reply-${String(turn)}-${String(step)}`), role: 'assistant', content,
+          source: { kind: 'model', provider: 'bench', model: 'bench' },
+        },
+        stream: [...stream.snapshot()],
+      }, { surfaceOp: 'append' })
+      if (callId !== undefined) {
+        log.flush()
+        const callSeq = log.append('tool/call', {
+          turn, step, callId, name: 'read_file', arguments: JSON.stringify({ path: `src/module-${String(turn)}-${String(step)}.ts` }),
+        })
+        log.flush()
+        log.append('tool/result', {
+          turn, step,
+          message: {
+            id: MessageId('result-' + callId), role: 'tool', toolCallId: callId, isError: false,
+            source: { kind: 'tool', callId },
+            content: [{ type: 'text', text: pool.slice(next, plan.resultChars) }],
+          },
+        }, { surfaceOp: 'append', sourceEventSeqs: [callSeq] })
+      }
+      log.append('step/end', { turn, step })
+    }
+    log.append('turn/end', { turn, reason: { kind: 'completed' } })
+  }
+  return { events: log.events, frameStarts: log.frameStarts }
+}
+
+function frameTexts(events: readonly SessionEvent[], frameStarts: ReadonlySet<number>): string[] {
+  const frames: string[] = []
+  let start = 0
+  for (let index = 1; index <= events.length; index++) {
+    if (index === events.length || frameStarts.has(index)) {
+      frames.push(eventLines(events.slice(start, index)) + '\n')
+      start = index
+    }
+  }
+  return frames
+}
+
+function headerLine(header: SessionHeader): string {
+  return JSON.stringify(toHeaderLine(header)) + '\n'
+}
+
+/** Size facts of one anchor body, saved beside its compressed frames. */
+export interface AnchorBodyFacts {
+  readonly frameCount: number
+  readonly events: number
+  readonly logicalBytes: number
+}
+
+/** One compressed event body shared by every Session of one measured anchor. */
+interface AnchorBody extends AnchorBodyFacts {
+  readonly frames: Buffer
+}
+
+/** One corpus written from authored anchor bodies. */
+export interface WrittenCorpus {
+  /** Physical and logical totals. */
+  readonly facts: SyntheticCorpusFacts
+  /** Every stored Session in rank order. */
+  readonly sessions: readonly WrittenSession[]
+}
+
+/**
+ * Tell whether one corpus rank stores a subagent child.
+ * @param rank - zero-based ascending length rank.
+ * @returns whether the rank is a child of the preceding rank.
+ */
+export function subagentRank(rank: number): boolean {
+  return rank % CHILD_EVERY === CHILD_EVERY - 1
+}
+
+/**
+ * Build the stored header of one corpus rank.
+ * @param rank - zero-based ascending length rank.
+ * @param count - corpus size.
+ * @returns the rank's header; every sixth rank is a subagent child of the preceding rank.
+ */
+export function corpusHeader(rank: number, count: number): SessionHeader {
+  if (count % CREATION_STRIDE === 0) throw new Error('corpus size must be coprime with the creation stride')
+  const child = subagentRank(rank)
+  // Creation time is a fixed permutation of length rank. List order follows each body's last prompt time,
+  // which Sessions sharing an anchor also share.
+  const createdAt = TIME_ZERO - (((rank * CREATION_STRIDE) % count) + 1) * 60_000
+  return {
+    version: SESSION_FORMAT_VERSION,
+    id: corpusSessionId(rank),
+    createdAt,
+    cwd: `/bench/project-${String(rank % Math.ceil(count / SESSIONS_PER_PROJECT))}`,
+    isSeeded: false,
+    ...child
+      ? { origin: 'subagent' as const, parentSession: corpusSessionId(rank - 1), delegationDepth: 1 }
+      : { delegationDepth: 0 },
+  }
+}
+
+/**
+ * Build the header under which one anchor's projections are folded before identity rebinding.
+ * @param anchor - measured anchor index.
+ * @returns a top-level header outside every corpus.
+ */
+export function anchorHeader(anchor: number): SessionHeader {
+  return {
+    version: SESSION_FORMAT_VERSION,
+    id: SessionId(`bench-anchor-${String(anchor).padStart(2, '0')}`),
+    createdAt: TIME_ZERO - 60_000,
+    cwd: '/bench/anchors',
+    isSeeded: false,
+    delegationDepth: 0,
+  }
+}
+
+function anchorBodyPath(directory: string, anchor: number): string {
+  return join(directory, `anchor-${String(anchor).padStart(2, '0')}.zst`)
+}
+
+/**
+ * Run one asynchronous compression or file operation per index with bounded concurrency.
+ * @param count - number of indexes, starting at zero.
+ * @param operation - operation for one index.
+ */
+export async function forEachConcurrently(count: number, operation: (index: number) => Promise<void>): Promise<void> {
+  let next = 0
+  const worker = async (): Promise<void> => {
+    for (let index = next++; index < count; index = next++) await operation(index)
+  }
+  await Promise.all(Array.from({ length: IO_CONCURRENCY }, () => worker()))
+}
+
+/** Authors each anchor body once, exchanges bodies between seeding processes, and writes corpora that share them. */
+export class SyntheticCorpusWriter {
+  private pool: TextPool | undefined
+  private readonly bodies = new Map<number, AnchorBody>()
+  /** Stored-byte model from the previous anchor: fixed system prompt, bytes per event, and bytes per payload character. */
+  private model = { bytesPerEvent: 0, bytesPerChar: 4 }
+
+  /**
+   * Author one anchor's body; its body bytes alone reach the anchor's logical-byte target.
+   * @param anchor - measured anchor index.
+   * @returns the authored events, which the writer does not retain.
+   */
+  async author(anchor: number): Promise<readonly SessionEvent[]> {
+    const shape = anchorShape(anchor)
+    // Start from the previous anchor's model, then take secant steps from below.
+    const { bytesPerEvent, bytesPerChar } = this.model
+    let payloadChars = Math.max(0, Math.ceil(
+      (1.01 * (shape.logicalBytes - SYSTEM_PROMPT_CHARS - bytesPerEvent * shape.events)) / bytesPerChar,
+    ))
+    let previous: { readonly payloadChars: number; readonly bytes: number } | undefined
+    for (let attempt = 0; attempt < 8; attempt++) {
+      this.pool ??= new TextPool()
+      const authored = authorEvents(shape, payloadChars, this.pool, anchor + 1)
+      if (authored.events.length < shape.events) {
+        throw new Error(`corpus anchor ${String(anchor)} authored ${String(authored.events.length)} of ${String(shape.events)} events`)
+      }
+      const frames = frameTexts(authored.events, authored.frameStarts)
+      const bytes = frames.reduce((sum, text) => sum + Buffer.byteLength(text), 0)
+      if (previous !== undefined && payloadChars > previous.payloadChars) {
+        this.model.bytesPerChar = (bytes - previous.bytes) / (payloadChars - previous.payloadChars)
+      }
+      this.model.bytesPerEvent = (bytes - SYSTEM_PROMPT_CHARS - this.model.bytesPerChar * payloadChars) / authored.events.length
+      if (bytes < shape.logicalBytes) {
+        previous = { payloadChars, bytes }
+        // Every message rounds its share up, so each step adds at least one character per event.
+        payloadChars += Math.ceil((1.01 * (shape.logicalBytes - bytes)) / this.model.bytesPerChar) + shape.events
+        continue
+      }
+      // Each pending compression owns a native Zstandard context, so compression is bounded like writes.
+      const compressed: Buffer[] = []
+      await forEachConcurrently(frames.length, async (index) => {
+        compressed[index] = await compressZstdFrame(frames[index] as string)
+      })
+      this.bodies.set(anchor, {
+        frames: Buffer.concat(compressed), frameCount: frames.length, events: authored.events.length, logicalBytes: bytes,
+      })
+      return authored.events
+    }
+    throw new Error(`corpus anchor ${String(anchor)} did not reach ${String(shape.logicalBytes)} logical bytes`)
+  }
+
+  /**
+   * Save one authored body's compressed frames for another process.
+   * @param directory - body directory shared by the seeding processes.
+   * @param anchor - authored anchor index.
+   * @returns the body's size facts.
+   */
+  async saveBody(directory: string, anchor: number): Promise<AnchorBodyFacts> {
+    const body = this.bodies.get(anchor)
+    if (body === undefined) throw new Error(`corpus anchor ${String(anchor)} was not authored`)
+    await mkdir(directory, { recursive: true })
+    await writeFile(anchorBodyPath(directory, anchor), body.frames)
+    return { frameCount: body.frameCount, events: body.events, logicalBytes: body.logicalBytes }
+  }
+
+  /**
+   * Load one body saved by another process.
+   * @param directory - body directory shared by the seeding processes.
+   * @param anchor - anchor index.
+   * @param facts - size facts recorded with the body.
+   */
+  async loadBody(directory: string, anchor: number, facts: AnchorBodyFacts): Promise<void> {
+    this.bodies.set(anchor, { ...facts, frames: await readFile(anchorBodyPath(directory, anchor)) })
+  }
+
+  /**
+   * Write every rank of one corpus from authored bodies.
+   * @param root - JSONL persistence root.
+   * @param count - corpus size; the length distribution is sampled at this many ranks.
+   * @returns totals and the stored headers.
+   */
+  async writeCorpus(root: string, count: number): Promise<WrittenCorpus> {
+    const totals = { events: 0, logicalBytes: 0, compressedBytes: 0, frames: 0 }
+    const sessions: WrittenSession[] = []
+    await forEachConcurrently(count, async (rank) => {
+      const { anchor } = sessionShape(rank, count)
+      const body = this.bodies.get(anchor)
+      if (body === undefined) throw new Error(`corpus anchor ${String(anchor)} was not authored`)
+      const header = corpusHeader(rank, count)
+      const line = headerLine(header)
+      const physical = Buffer.concat([await compressZstdFrame(line), body.frames])
+      const path = generationLogPath(root, header.cwd, header.id, SESSION_FORMAT_VERSION, 'zstd')
+      await mkdir(dirname(path), { recursive: true })
+      await writeFile(path, physical)
+      sessions[rank] = { header, anchor }
+      totals.events += body.events
+      totals.logicalBytes += Buffer.byteLength(line) + body.logicalBytes
+      totals.compressedBytes += physical.byteLength
+      totals.frames += body.frameCount + 1
+    })
+    return {
+      facts: {
+        sessions: count,
+        projects: Math.ceil(count / SESSIONS_PER_PROJECT),
+        distinctBodies: new Set(sessions.map(session => session.anchor)).size,
+        ...totals,
+      },
+      sessions,
+    }
+  }
+}

benchmarks/tsdown.config.ts

@@ -47,6 +47,13 @@ export default defineConfig([
     clean: true,
     tsconfig: 'tsconfig.host.json',
   },
+  {
+    ...shared,
+    entry: { 'session-corpus.worker': 'session-corpus/session-corpus.worker.ts' },
+    outDir: '.dsh-build/session-corpus',
+    clean: true,
+    tsconfig: 'tsconfig.host.json',
+  },
   {
     ...shared,
     entry: {

pnpm-lock.yaml

@@ -904,6 +904,15 @@ importers:
       '@deepseek-ai/dsh-session':
         specifier: workspace:*
         version: link:../packages/core/session
+      '@deepseek-ai/dsh-session-format':
+        specifier: workspace:*
+        version: link:../packages/session/session-format
+      '@deepseek-ai/dsh-session-format-catalog':
+        specifier: workspace:*
+        version: link:../packages/session/session-format-catalog
+      '@deepseek-ai/dsh-session-format-v3-to-v4':
+        specifier: workspace:*
+        version: link:../packages/session/session-format-v3-to-v4
       '@deepseek-ai/dsh-session-persistence':
         specifier: workspace:*
         version: link:../packages/session/session-persistence
@@ -913,9 +922,15 @@ importers:
       '@deepseek-ai/dsh-session-projection':
         specifier: workspace:*
         version: link:../packages/session/session-projection
+      '@deepseek-ai/dsh-session-projection-cache':
+        specifier: workspace:*
+        version: link:../packages/session/session-projection-cache
       '@deepseek-ai/dsh-session-query':
         specifier: workspace:*
         version: link:../packages/session-query/session-query
+      '@deepseek-ai/dsh-session-query-sqlite':
+        specifier: workspace:*
+        version: link:../packages/session-query/session-query-sqlite
       '@deepseek-ai/dsh-session-stats':
         specifier: workspace:*
         version: link:../packages/session/session-stats
@@ -925,6 +940,15 @@ importers:
       '@deepseek-ai/dsh-session-turn-outline':
         specifier: workspace:*
         version: link:../packages/session/session-turn-outline
+      '@deepseek-ai/dsh-storage':
+        specifier: workspace:*
+        version: link:../packages/storage/storage
+      '@deepseek-ai/dsh-storage-domain':
+        specifier: workspace:*
+        version: link:../packages/storage/storage-domain
+      '@deepseek-ai/dsh-storage-json':
+        specifier: workspace:*
+        version: link:../packages/storage/storage-json
       '@deepseek-ai/dsh-subagent':
         specifier: workspace:*
         version: link:../packages/subagent/subagent

scripts/ci-workflow.spec.ts

@@ -482,10 +482,10 @@ describe('CI workflow', () => {
     })
   })
 
-  it('bounds the complete benchmark job to fifteen minutes', () => {
+  it('bounds the complete benchmark job to twenty minutes', () => {
     const benchmark = workflowJob(loadWorkflow('.github/workflows/ci.yml'), 'node-24-bench')
 
-    expect(benchmark['timeout-minutes']).toBe(15)
+    expect(benchmark['timeout-minutes']).toBe(20)
     expect(benchmark.steps).toContainEqual({
       name: 'Run performance benchmarks',
       env: { DSH_GATE_VERBOSE: '1' },