Merge pull request #3537 from deepseek-harness/perf/03-client-stream-fold

perf: read embedded Assistant streams per compact record
2026-09-13 04:03:30 +00:00 · 2026-09-06 13:17:51 +08:00
parent 53c2ed9916 07245b9e88
commit 81dcacbb44
17 changed files with 632 additions and 61 deletions
@@ -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-09-06-embedded-stream-record-readers.md
+2026-09-06-embedded-stream-record-readers.md: 972fee634833cef5fd7b0a54f69780b9370f0cc3
+2026-09-06-embedded-stream-record-readers.zh.md: faba6e179887a9943926f2c73aa8e42a10cee300
@@ -0,0 +1,50 @@
+# Agent Note: Embedded Assistant stream consumers read compact records
+
+Status: implemented
+
+English | [中文](2026-09-06-embedded-stream-record-readers.zh.md)
+
+## Problem
+
+Session format v2 embeds each model attempt's compact stream (`AssistantStreamRecord[]`: packed `text-chunks`, `reasoning-chunks`, and `tool-call-chunks` runs plus timestamped raw `chunk` records) in `assistant/message` and `assistant/attempt`. Consumers that folded those settlements called `expandAssistantStream()` first; it materializes the complete per-member array, so a consumer that needs one fact (`find` on the first token, the last usage chunk, a joined text, one block-end) paid O(members) allocation and time: about two objects per member on top of the compact form.
+
+After v2 embedded streams settlement widened with the message content and Chat and Trajectory sections settled directly from it, the remaining expand consumers are the Host and client folds: Session Stats reads the first-token time per `assistant/attempt` and `assistant/message` (the projection phase of every Session open), the token meter rebuilds provider content and scans every stream for its last usage chunk (the projection unit still scans to the end), the subagent output fold joins plain text, and the Session Controller image lookup scans for block-end chunks.
+
+## Decision
+
+`@deepseek-ai/dsh-llm` answers consumer questions directly from compact records; every remaining consumer folds records once with early exit.
+
+`packages/llm/llm/src/assistant-stream.ts` exports record-level readers beside the accumulator and `expandAssistantStream`:
+
+- Chunk rules: `isTokenDelta` (non-empty text, reasoning, or Tool-call arguments fragment, or any name-bearing Tool-call delta), `isVisibleChunk` (non-whitespace text or reasoning, or a block start or end of any kind other than text, reasoning, or Tool call), and `chunkHasVisibleText` (non-whitespace text delta or completed text block).
+- Run readers: `runFirstTokenTime` and `runFirstVisibleTime` reconstruct the first qualifying member's time from `time0` and the `dt` gaps and stop scanning there; a name-bearing Tool-call run yields `time0` without reading a fragment.
+- Stream readers: `assistantStreamFirstTokenTime`, `assistantStreamHasVisibleContent`, `assistantStreamHasVisibleText`, `lastAssistantStreamChunk(stream, type)` (backward scan), `assistantStreamChunks(stream, type)`, `joinAssistantStreamText`, and `assembleAssistantStream`, which feeds a `BlockAssembler` one joined delta per run (assembly only concatenates, so blocks, usage, finish, and replay state equal the per-member result). `RawStreamChunkType` excludes the delta types, so a raw-chunk lookup can never silently skip packed members.
+
+Session Stats reads `assistantStreamFirstTokenTime`; the token meter reads `lastAssistantStreamChunk(stream, 'usage')` and assembles provider output through `assembleAssistantStream`; the subagent output fold appends `joinAssistantStreamText`; the Session Controller scans `assistantStreamChunks(stream, 'block-end')` for images.
+
+`expandAssistantStream` keeps its strict validation and its remaining callers, which need every member or validate the stream at a durable boundary: Session restore validation, the v1-to-v2 migration validator and publication Worker replay, the reconnect baseline, and test support.
+
+### Measurements
+
+The repo's synthetic first-open benchmark (200 turns, 127,400 released-v0 events, 500,000 streamed deltas in 1,600 compact records; five samples, median):
+
+| Phase | Before | After |
+|---|---|---|
+| first-open projection | 28.0 ms | 5.9 ms |
+| first-open total | 76.9 ms | 53.8 ms |
+| first-open peak RSS | 137.2 MB | 94.6 MB |
+| reopen projection | 17.8 ms | 6.5 ms |
+
+Open, read, and restore phases are unchanged; the reader keeps the same first-token time by construction (the first qualifying member is the first record's first qualifying fragment, and the deltas stay ordered).
+
+## Alternatives considered
+
+**Memoize `expandAssistantStream` per input array.** Expanding all streams once costs tens of milliseconds, but retaining the expansions costs about ten times the compact stream for the event's lifetime — a permanent version of the transient allocation the change removes. The readers remove the need for retained expansions entirely.
+
+**Keep the per-member fold.** Early-exit `.find` still materializes the whole array first, so the allocation and O(members) time remain.
+
+## Consequences
+
+Host and Client folds of an embedded settlement cost O(records) plus one join per run, and no consumer materializes members unless it validates at a durable boundary or needs every member. The token, visibility, and visible-text rules have one home in `dsh-llm`, so a record reader and the accumulator's packing rules cannot drift apart.
+
+Publication verification (`assertCurrentAssistantStreams`) still replays every settlement at publish time; because it must prove content-by-chunk agreement, converting it to run-aware assembly without member materialization remains open work.
@@ -0,0 +1,50 @@
+# Agent Note: 内嵌 Assistant 流的消费方直接读取紧凑记录
+
+Status: implemented
+
+[English](2026-09-06-embedded-stream-record-readers.md) | 中文
+
+## 问题
+
+Session 格式 v2 将每次模型尝试的紧凑流（`AssistantStreamRecord[]`：打包的 `text-chunks`、`reasoning-chunks`、`tool-call-chunks` run 加上带时间戳的原始 `chunk` 记录）嵌入 `assistant/message` 与 `assistant/attempt`。折叠这些 settlement 的消费方会先调用 `expandAssistantStream()`；它会物化完整的逐成员数组，因此只需一个事实的消费方（find 首个 token、最后一个 usage chunk、拼接文本、一个 block-end）也要付出 O(members) 的分配与时间：在紧凑形式之上每个成员约两个对象。
+
+在 v2 内嵌流 settlement 随消息内容扩展、Chat 与 Trajectory 区块直接由内容结算之后，剩余的 expand 消费方是 Host 与客户端折叠：Session Stats 读取每个 `assistant/attempt` 与 `assistant/message` 的首 token 时间（每次打开 Session 的 projection 阶段）、token 计量重建提供商内容并扫描每个流到最后一个 usage chunk（projection 单元仍扫描到末尾）、子代理输出折叠拼接纯文本、Session Controller 镜像查找扫描 block-end chunk。
+
+## 决策
+
+`@deepseek-ai/dsh-llm` 直接从紧凑记录回答消费方问题；剩余消费方对记录做一次带提前退出的折叠。
+
+`packages/llm/llm/src/assistant-stream.ts` 在累加器与 `expandAssistantStream` 之外导出记录级读取器：
+
+- Chunk 规则：`isTokenDelta`（非空文本、reasoning 或 Tool-call 参数片段，或任何带名称的 Tool-call delta）、`isVisibleChunk`（非空白文本或 reasoning，或 text/reasoning/Tool call 之外的任意块开始或结束）、`chunkHasVisibleText`（非空白文本 delta 或完成的文本块）。
+- Run 读取器：`runFirstTokenTime` 与 `runFirstVisibleTime` 从 `time0` 与 `dt` 间隔重建首个合格成员的时间并停止扫描；带名称的 Tool-call run 直接产出 `time0`，不读片段。
+- 流读取器：`assistantStreamFirstTokenTime`、`assistantStreamHasVisibleContent`、`assistantStreamHasVisibleText`、`lastAssistantStreamChunk(stream, type)`（逆向扫描）、`assistantStreamChunks(stream, type)`、`joinAssistantStreamText` 与 `assembleAssistantStream`（每个 run 向 `BlockAssembler` 喂入一个拼接后的 delta；组装只做拼接，因此 blocks、usage、finish 与 replay state 与逐成员结果一致）。`RawStreamChunkType` 排除 delta 类型，因此原始 chunk 查找不可能静默跳过打包成员。
+
+Session Stats 读取 `assistantStreamFirstTokenTime`；token 计量读取 `lastAssistantStreamChunk(stream, 'usage')` 并通过 `assembleAssistantStream` 组装提供商输出；子代理输出折叠追加 `joinAssistantStreamText`；Session Controller 用 `assistantStreamChunks(stream, 'block-end')` 扫描镜像。
+
+`expandAssistantStream` 保留其严格校验与其余调用方（需要每个成员或在持久边界校验流）：Session 恢复校验、v1-to-v2 迁移校验器与发布 Worker 重放、重连基线、测试支撑。
+
+### 测量
+
+仓库的合成 first-open 基准（200 循环、127,400 个 released-v0 事件、1,600 条紧凑记录中的 500,000 个流式 delta；五次采样取中位数）：
+
+| 阶段 | 之前 | 之后 |
+|---|---|---|
+| first-open projection | 28.0 ms | 5.9 ms |
+| first-open 总计 | 76.9 ms | 53.8 ms |
+| first-open 峰值 RSS | 137.2 MB | 94.6 MB |
+| reopen projection | 17.8 ms | 6.5 ms |
+
+Open、read、restore 阶段不变；读取器按构造保持相同的首 token 时间（首个合格成员即首条记录的首个合格片段，且 delta 保持有序）。
+
+## 备选方案
+
+**按输入数组记忆化 `expandAssistantStream`。** 展开全部流只需几十毫秒，但保留展开结果在事件生命周期内约花费紧凑流的十倍内存——这是本变更移除的瞬时分配的永久版本。读取器完全消除了对保留展开的需求。
+
+**保留逐成员折叠。** 提前退出的 `.find` 仍然先物化整个数组，因此分配与 O(members) 时间仍在。
+
+## 后果
+
+Host 与客户端折叠一次内嵌结算的代价为 O(records) 加每个 run 一次拼接，且除非在持久边界校验或需要每个成员，消费方不再物化成员。token、可见性与可见文本规则在 `dsh-llm` 中只有一处，因此记录读取器与累加器的打包规则不可能漂移。
+
+发布校验（`assertCurrentAssistantStreams`）仍在发布时重放每个 settlement；因为它必须按 chunk 证明内容一致，将其转为不入成员的 run 感知组装仍是未完成工作。
@@ -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 docs/subsystems/llm-streaming.md
-llm-streaming.md: b523df1c970a2bfae23450d7f7909e9e6e960e8c
-llm-streaming.zh.md: 745733ed639ff06bfd154592929bae4812868c30
+llm-streaming.md: 97062fb326a2718daf33b19b2f7f00175a2ec1fa
+llm-streaming.zh.md: 9f4dc7d32bee62f55e971afb44905141cabe4e80
@@ -225,7 +225,7 @@ type StreamChunk =

 `snapshot()` returns a detached immutable stream. `expandAssistantStream()` strictly checks record keys, member counts, indexes, timestamps, tool-call identity, and lossless JSON before recreating the exact timed chunk sequence. The Session log embeds this stream in `assistant/message` for a surface result or `assistant/attempt` for an attempt with no surface message.

-Process-local `agent/assistant-stream` frames carry live presentation. Durable replay, telemetry, token accounting, and historical UI assembly expand the embedded settlement instead of treating live frames as persisted facts.
+Process-local `agent/assistant-stream` frames carry live presentation. Durable replay and restore validation still expand the embedded settlement; telemetry, token accounting, and Host folds read the compact records directly. Record-level readers (`assistantStreamFirstTokenTime`, `assistantStreamHasVisibleContent`, `assistantStreamHasVisibleText`, `lastAssistantStreamChunk`, `assistantStreamChunks`, `joinAssistantStreamText`, `assembleAssistantStream`, and the per-run `runFirstTokenTime` and `runFirstVisibleTime`) answer consumer questions in one pass over the records with early exit, so a large history costs O(records) per settlement instead of O(members) expansion ([fold decision](../../.agents/notes/implemented/architecture/2026-09-06-embedded-stream-record-readers.md)). `expandAssistantStream()` remains the validating path for records read at a durable boundary and for consumers that need every member.

 ## `LlmFailure`

@@ -225,7 +225,7 @@ type StreamChunk =

 `snapshot()` 返回分离且不可变的 stream。`expandAssistantStream()` 会严格检查 record key、成员数、index、时间戳、tool-call identity 与无损 JSON，再重建精确的带时间 chunk 序列。Session 日志会把该 stream 嵌入作为 surface result 的 `assistant/message`，或嵌入没有 surface message 的 `assistant/attempt`。

-进程本地 `agent/assistant-stream` frame 承载实时呈现。持久回放、遥测、token 记账与历史 UI 组装会展开嵌入式 settlement，而不会把 live frame 当作持久事实。
+进程本地 `agent/assistant-stream` frame 承载实时呈现。持久回放与恢复校验仍会展开内嵌 settlement；遥测、token 记账与 Host 折叠直接读取紧凑记录。记录级读取器（`assistantStreamFirstTokenTime`、`assistantStreamHasVisibleContent`、`assistantStreamHasVisibleText`、`lastAssistantStreamChunk`、`assistantStreamChunks`、`joinAssistantStreamText`、`assembleAssistantStream` 以及按 run 的 `runFirstTokenTime` 与 `runFirstVisibleTime`）以提前退出在一次扫描内回答消费方问题，因此大历史每次结算的代价为 O(records) 而非 O(members) 展开（[折叠决策](../../.agents/notes/implemented/architecture/2026-09-06-embedded-stream-record-readers.zh.md)）。`expandAssistantStream()` 仍是持久边界读取记录与需要每个成员的消费方的校验路径。

 <a id="llmfailure"></a>

@@ -11,7 +11,7 @@ import type {
 import type { FileUploadReceiptId } from '@deepseek-ai/dsh-client-file-upload/types'
 import type {} from '@deepseek-ai/dsh-client-file-upload'
 import {
-  ReasoningEffortId, createUserMessage, expandAssistantStream, freezeMessage,
+  ReasoningEffortId, assistantStreamChunks, createUserMessage, freezeMessage,
 } from '@deepseek-ai/dsh-llm'
 import type { MessageSource } from '@deepseek-ai/dsh-llm'
 import { SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
@@ -593,8 +593,7 @@ function imageInEvent(
    if (found !== undefined) return found
  }
  if (event.type === 'assistant/message' || event.type === 'assistant/attempt') {
-    for (const { chunk } of expandAssistantStream(event.data.stream)) {
-      if (chunk.type !== 'block-end') continue
+    for (const chunk of assistantStreamChunks(event.data.stream, 'block-end')) {
      const found = imageBlockIn([chunk.block], match)
      if (found !== undefined) return found
    }
@@ -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/llm/llm/README.md
-README.md: 59b27eaa56769c2b7915e86c88140be59795e3ef
-README.zh.md: f517154fe335aca1054deb9b2c30694c3fab5d44
+README.md: 0f84af8418f916a77589907656ed310ae2bb73f7
+README.zh.md: 2c306940af912176a87d80a2552808cc2b644554
@@ -63,6 +63,7 @@ After a successful mount, `ctx.llm.listProviders()` reports the registered route
 - **Expose and activate providers through configuration** — adapters declare configurable-provider routes plus a settings namespace, so configuration surfaces can activate dormant providers and edit connection facts without a restart.
 - **Discover and resolve models** — list the models an adapter advertises, interrogate an endpoint for the models it serves, and resolve one exact model's context window, output default, reasoning efforts, and input modalities.
 - **Validate call config** — an explicit or configured reasoning effort is checked against the exact model before any provider I/O, and an adapter-configured output cap is materialized when the request omits one.
+- **Read an embedded Assistant stream without expanding it** — `assistantStreamFirstTokenTime` (first token), `assistantStreamHasVisibleContent` (any visible content), and `assistantStreamHasVisibleText` (any visible text) answer their questions from the compact records with early exit; `lastAssistantStreamChunk` scans backward to the last raw chunk of one type, `assistantStreamChunks` and `joinAssistantStreamText` scan the whole stream, and `assembleAssistantStream` feeds a `BlockAssembler` one joined delta per run with the same blocks, usage, and replay state as the per-member expansion. `runFirstTokenTime` and `runFirstVisibleTime` do the early-exit scan for one packed run, and `isTokenDelta`, `isVisibleChunk`, and `chunkHasVisibleText` define the token and visibility rules for a single chunk. `expandAssistantStream` remains the validating path for records read at a durable boundary; it is not memoized, because a retained expansion costs roughly ten times the compact stream for as long as the event lives.

 ### Failures and recovery

@@ -90,7 +91,7 @@ The service is built on one separation: **the logical contract is provider-neutr
 | [`src/types.ts`](src/types.ts) | The `StreamChunk` protocol, content-block map, finish reasons, and shared vocabulary |
 | [`src/message.ts`](src/message.ts) | Immutable message constructors shared by delivery, history, and requests |
 | [`src/assembler.ts`](src/assembler.ts) | `BlockAssembler`: incremental chunk-to-block assembly |
-| [`src/assistant-stream.ts`](src/assistant-stream.ts) | Compact timed Assistant stream accumulation, strict validation, and exact expansion |
+| [`src/assistant-stream.ts`](src/assistant-stream.ts) | Compact timed Assistant stream accumulation, strict validation, exact expansion, and record-level readers |
 | [`src/call-config.ts`](src/call-config.ts) | Call-config validation, adapter-default materialization, and request freezing |
 | [`src/retry-policy.ts`](src/retry-policy.ts) | Provider-owned retry policy resolution (normal and always modes) |
 | [`src/error.ts`](src/error.ts) | `HarnessError`/`LlmError` taxonomy and provider-neutral failure codes |
@@ -63,6 +63,7 @@ for await (const chunk of ctx.llm.stream({
 - **通过配置暴露并激活提供方**——适配器声明可配置提供方路由与 settings namespace，配置界面因此可以激活休眠提供方并编辑连接事实，无需重启。
 - **发现与解析模型**——列出适配器公布的模型、询问端点它提供哪些模型，并解析某个精确模型的上下文窗口、输出默认值、推理（reasoning）强度与输入模态。
 - **校验调用配置**——显式或配置的推理强度会在任何提供方 I/O 之前对照精确模型校验；请求省略输出上限时，会填入适配器配置的输出上限。
+- **不展开即读取内嵌 Assistant 流**——`assistantStreamFirstTokenTime`（首 token）、`assistantStreamHasVisibleContent`（任一可见内容）与 `assistantStreamHasVisibleText`（任一可见文本）以提前退出从紧凑记录回答各自的问题；`lastAssistantStreamChunk` 反向扫描到某一类型的最后一个原始 chunk，`assistantStreamChunks` 与 `joinAssistantStreamText` 扫描整个流，`assembleAssistantStream` 向 `BlockAssembler` 每个 run 喂一段拼接 delta，blocks／usage／replayState 与逐成员展开相同。`runFirstTokenTime` 与 `runFirstVisibleTime` 对单个打包 run 做提前退出扫描，`isTokenDelta`、`isVisibleChunk` 与 `chunkHasVisibleText` 定义单个 chunk 的 token 与可见性规则。`expandAssistantStream` 仍是持久边界读取记录的校验路径；它不被记忆化，因为保留的展开在事件生命周期内约花费紧凑流的十倍内存。

 ### 失败与恢复

@@ -90,7 +91,7 @@ for await (const chunk of ctx.llm.stream({
 | [`src/types.ts`](src/types.ts) | `StreamChunk` 协议、内容块映射、结束原因与共享词汇 |
 | [`src/message.ts`](src/message.ts) | 投递、历史与请求共享的不可变消息构造函数 |
 | [`src/assembler.ts`](src/assembler.ts) | `BlockAssembler`：分片到块的增量组装 |
-| [`src/assistant-stream.ts`](src/assistant-stream.ts) | 紧凑带时间 Assistant stream 的累积、严格校验与精确展开 |
+| [`src/assistant-stream.ts`](src/assistant-stream.ts) | 紧凑带时间 Assistant stream 的累积、严格校验、精确展开与记录级读取器 |
 | [`src/call-config.ts`](src/call-config.ts) | 调用配置校验、适配器默认值填入与请求冻结 |
 | [`src/retry-policy.ts`](src/retry-policy.ts) | 提供方自有重试策略解析（normal 与 always 模式） |
 | [`src/error.ts`](src/error.ts) | `HarnessError`/`LlmError` 分类体系与提供方无关失败 code |
@@ -1,8 +1,14 @@
-/** Lossless compact representation of one model-stream attempt. */
+/**
+ * Lossless compact representation of one model-stream attempt, plus record-level
+ * readers that answer common consumer questions without materializing members.
+ * Readers trust the static record type; expandAssistantStream is the validating
+ * path for records read at a durable boundary.
+ */

 import { assertNever, deepFreeze, snapshotJsonValue } from '@deepseek-ai/dsh-util-values'
+import { BlockAssembler } from './assembler.ts'
 import type { ToolCallId } from './brand.ts'
-import type { StreamChunk } from './types.ts'
+import type { ContentBlock, StreamChunk } from './types.ts'

 /** One model chunk paired with its original Session timestamp. */
 export interface TimedStreamChunk {
@@ -37,6 +43,15 @@ export type AssistantStreamRecord =
  }
  | { readonly type: 'chunk'; readonly time: number; readonly chunk: StreamChunk }

+/** One packed delta run: every compact record except a raw `chunk`. */
+export type AssistantStreamRun = Exclude<AssistantStreamRecord, { type: 'chunk' }>
+
+/**
+ * Chunk types the accumulator never packs into runs, so every occurrence is a raw
+ * `chunk` record. Delta types are excluded because their packed members are not raw chunks.
+ */
+export type RawStreamChunkType = Exclude<StreamChunk['type'], 'text-delta' | 'reasoning-delta' | 'tool-call-delta'>
+
 type MutableRecord =
  | {
    type: 'text-chunks' | 'reasoning-chunks'
@@ -216,6 +231,228 @@ export function expandAssistantStream(stream: readonly AssistantStreamRecord[]):
  return chunks
 }

+function hasNonWhitespace(text: string): boolean {
+  return /\S/.test(text)
+}
+
+function blockIsVisible(block: ContentBlock): boolean {
+  if (block.type === 'tool-call') return false
+  if (block.type === 'text' || block.type === 'reasoning') return hasNonWhitespace(block.text)
+  return true
+}
+
+/**
+ * Whether one chunk carries the model's first output token for latency measurement.
+ * @param chunk - any stream chunk.
+ * @returns true for a non-empty text, reasoning, or Tool-call arguments fragment and for
+ *   every name-bearing Tool-call delta; false for block, usage, and finish chunks.
+ */
+export function isTokenDelta(chunk: StreamChunk): boolean {
+  switch (chunk.type) {
+    case 'text-delta':
+    case 'reasoning-delta':
+      return chunk.text !== ''
+    case 'tool-call-delta':
+      return chunk.argumentsDelta !== '' || chunk.name !== undefined
+    default:
+      return false
+  }
+}
+
+/**
+ * Whether one chunk by itself contributes reader-visible transcript content.
+ * Text and reasoning count only with non-whitespace content, streamed as a delta or
+ * completed as a block; a block of any other kind counts at its start and its end,
+ * except a Tool call, which is protocol rather than content. Usage and finish never count.
+ * @param chunk - any stream chunk.
+ * @returns whether a transcript reader would see this chunk.
+ */
+export function isVisibleChunk(chunk: StreamChunk): boolean {
+  switch (chunk.type) {
+    case 'text-delta':
+    case 'reasoning-delta':
+      return hasNonWhitespace(chunk.text)
+    case 'block-start':
+      return chunk.blockType !== 'text' && chunk.blockType !== 'reasoning' && chunk.blockType !== 'tool-call'
+    case 'block-end':
+      return blockIsVisible(chunk.block)
+    default:
+      return false
+  }
+}
+
+/**
+ * Whether one chunk carries non-whitespace text, as a text delta or a completed text block.
+ * Reasoning, Tool calls, and other block kinds never count.
+ * @param chunk - any stream chunk.
+ * @returns whether the chunk contributes visible text.
+ */
+export function chunkHasVisibleText(chunk: StreamChunk): boolean {
+  if (chunk.type === 'text-delta') return hasNonWhitespace(chunk.text)
+  return chunk.type === 'block-end' && chunk.block.type === 'text' && hasNonWhitespace(chunk.block.text)
+}
+
+function firstRunMemberTime(run: AssistantStreamRun, predicate: (fragment: string) => boolean): number | undefined {
+  const fragments = run.type === 'tool-call-chunks' ? run.args : run.texts
+  let time = run.time0
+  for (let index = 0; index < fragments.length; index += 1) {
+    if (index > 0) time += run.dt[index - 1] as number
+    if (predicate(fragments[index] as string)) return time
+  }
+  return undefined
+}
+
+/**
+ * Time of the first member of one packed run that {@link isTokenDelta} accepts: a
+ * name-bearing Tool-call run starts at its first member, otherwise the first non-empty fragment.
+ * Stops scanning at that member.
+ * @param run - one packed delta run.
+ * @returns the member's reconstructed time, or undefined when no member qualifies.
+ */
+export function runFirstTokenTime(run: AssistantStreamRun): number | undefined {
+  if (run.type === 'tool-call-chunks' && run.name !== undefined) return run.time0
+  return firstRunMemberTime(run, fragment => fragment !== '')
+}
+
+/**
+ * Time of the first member of one packed run that {@link isVisibleChunk} accepts: the first
+ * non-whitespace text or reasoning fragment. A Tool-call run has none. Stops scanning at that member.
+ * @param run - one packed delta run.
+ * @returns the member's reconstructed time, or undefined when no member qualifies.
+ */
+export function runFirstVisibleTime(run: AssistantStreamRun): number | undefined {
+  return run.type === 'tool-call-chunks' ? undefined : firstRunMemberTime(run, hasNonWhitespace)
+}
+
+/**
+ * Time of the first token in one compact stream per {@link isTokenDelta}, read from the
+ * records themselves and stopping at the first qualifying member.
+ * @param stream - compact records from one durable Assistant settlement.
+ * @returns the first token's time, or undefined when the stream carries no token.
+ */
+export function assistantStreamFirstTokenTime(stream: readonly AssistantStreamRecord[]): number | undefined {
+  for (const record of stream) {
+    const time = record.type === 'chunk'
+      ? (isTokenDelta(record.chunk) ? record.time : undefined)
+      : runFirstTokenTime(record)
+    if (time !== undefined) return time
+  }
+  return undefined
+}
+
+/**
+ * Whether one compact stream carries any reader-visible content per {@link isVisibleChunk},
+ * stopping at the first qualifying member.
+ * @param stream - compact records from one durable Assistant settlement.
+ * @returns whether a transcript reader would see anything from this stream.
+ */
+export function assistantStreamHasVisibleContent(stream: readonly AssistantStreamRecord[]): boolean {
+  return stream.some(record => record.type === 'chunk'
+    ? isVisibleChunk(record.chunk)
+    : runFirstVisibleTime(record) !== undefined)
+}
+
+/**
+ * Whether one compact stream carries non-whitespace text per {@link chunkHasVisibleText},
+ * stopping at the first qualifying member.
+ * @param stream - compact records from one durable Assistant settlement.
+ * @returns whether the stream contributes visible text.
+ */
+export function assistantStreamHasVisibleText(stream: readonly AssistantStreamRecord[]): boolean {
+  return stream.some(record => record.type === 'text-chunks'
+    ? record.texts.some(hasNonWhitespace)
+    : record.type === 'chunk' && chunkHasVisibleText(record.chunk))
+}
+
+/**
+ * The last raw chunk of one never-packed type, scanning backwards and stopping at the first hit.
+ * @param stream - compact records from one durable Assistant settlement.
+ * @param type - chunk type that only appears as a raw record.
+ * @returns the stream's final chunk of that type, or undefined when it has none.
+ */
+export function lastAssistantStreamChunk<T extends RawStreamChunkType>(
+  stream: readonly AssistantStreamRecord[],
+  type: T,
+): Extract<StreamChunk, { type: T }> | undefined {
+  for (let index = stream.length - 1; index >= 0; index -= 1) {
+    const record = stream[index] as AssistantStreamRecord
+    if (record.type === 'chunk' && record.chunk.type === type) return record.chunk as Extract<StreamChunk, { type: T }>
+  }
+  return undefined
+}
+
+/**
+ * Every raw chunk of one never-packed type, in stream order.
+ * @param stream - compact records from one durable Assistant settlement.
+ * @param type - chunk type that only appears as a raw record.
+ * @returns the matching chunks; empty when the stream has none.
+ */
+export function assistantStreamChunks<T extends RawStreamChunkType>(
+  stream: readonly AssistantStreamRecord[],
+  type: T,
+): readonly Extract<StreamChunk, { type: T }>[] {
+  const chunks: Extract<StreamChunk, { type: T }>[] = []
+  for (const record of stream) {
+    if (record.type === 'chunk' && record.chunk.type === type) chunks.push(record.chunk as Extract<StreamChunk, { type: T }>)
+  }
+  return chunks
+}
+
+/**
+ * Every streamed text-delta fragment joined in stream order; reasoning and Tool-call fragments are excluded.
+ * @param stream - compact records from one durable Assistant settlement.
+ * @returns the joined text, empty when the stream carries no text delta.
+ */
+export function joinAssistantStreamText(stream: readonly AssistantStreamRecord[]): string {
+  const parts: string[] = []
+  for (const record of stream) {
+    if (record.type === 'text-chunks') parts.push(record.texts.join(''))
+    else if (record.type === 'chunk' && record.chunk.type === 'text-delta') parts.push(record.chunk.text)
+  }
+  return parts.join('')
+}
+
+/**
+ * Feed one compact stream into a {@link BlockAssembler} without materializing members.
+ * Each run contributes one delta carrying its joined fragments, which assembles the same
+ * blocks as the original per-member deltas because assembly only concatenates them;
+ * raw chunks are pushed as recorded. The records are trusted, not validated: validate a
+ * stream read at a durable boundary with {@link expandAssistantStream} first.
+ * @param stream - compact records from one durable Assistant settlement.
+ * @param assembler - assembler to feed; a fresh one by default.
+ * @returns the same assembler after every record was pushed.
+ */
+export function assembleAssistantStream(
+  stream: readonly AssistantStreamRecord[],
+  assembler = new BlockAssembler(),
+): BlockAssembler {
+  for (const record of stream) {
+    switch (record.type) {
+      case 'chunk':
+        assembler.push(record.chunk)
+        break
+      case 'text-chunks':
+        assembler.push({ type: 'text-delta', index: record.index, text: record.texts.join('') })
+        break
+      case 'reasoning-chunks':
+        assembler.push({ type: 'reasoning-delta', index: record.index, text: record.texts.join('') })
+        break
+      case 'tool-call-chunks':
+        assembler.push({
+          type: 'tool-call-delta',
+          index: record.index,
+          id: record.id,
+          ...record.name === undefined ? {} : { name: record.name },
+          argumentsDelta: record.args.join(''),
+        })
+        break
+      default:
+        assertNever(record, 'assembleAssistantStream')
+    }
+  }
+  return assembler
+}
+
 function validateRecord(value: unknown): AssistantStreamRecord {
  if (typeof value !== 'object' || value === null || Array.isArray(value)) {
    throw new TypeError('Assistant stream record must be an object')
@@ -1,10 +1,23 @@
 import { describe, expect, it } from 'vitest'
 import {
  AssistantStreamAccumulator,
+  BlockAssembler,
  ToolCallId,
+  assembleAssistantStream,
+  assistantStreamChunks,
+  assistantStreamFirstTokenTime,
+  assistantStreamHasVisibleContent,
+  assistantStreamHasVisibleText,
+  chunkHasVisibleText,
  expandAssistantStream,
+  isTokenDelta,
+  isVisibleChunk,
+  joinAssistantStreamText,
+  lastAssistantStreamChunk,
+  runFirstTokenTime,
+  runFirstVisibleTime,
 } from '@deepseek-ai/dsh-llm'
-import type { TimedStreamChunk } from '@deepseek-ai/dsh-llm'
+import type { AssistantStreamRecord, AssistantStreamRun, StreamChunk, TimedStreamChunk } from '@deepseek-ai/dsh-llm'

 describe('AssistantStreamAccumulator', () => {
  it('keeps delta boundaries and timestamps while compacting one attempt', () => {
@@ -223,3 +236,252 @@ describe('AssistantStreamAccumulator', () => {
    expect(() => expandAssistantStream([record] as never)).toThrow(message)
  })
 })
+
+/** Fragment array that counts index reads, so a scan's early exit is observable. */
+function countedFragments(values: readonly string[]): { readonly fragments: readonly string[]; reads(): number } {
+  let reads = 0
+  const fragments = new Proxy([...values], {
+    get(target, property, receiver): unknown {
+      if (typeof property === 'string' && /^\d+$/.test(property)) reads += 1
+      return Reflect.get(target, property, receiver)
+    },
+  })
+  return { fragments, reads: () => reads }
+}
+
+/** Record whose every property read throws, proving a stream scan never reached it. */
+const unreachableRecord = new Proxy({}, {
+  get() {
+    throw new Error('scan continued past the first qualifying record')
+  },
+}) as AssistantStreamRecord
+
+type RunOf<Type extends AssistantStreamRun['type']> = Extract<AssistantStreamRun, { type: Type }>
+
+function textRun(time0: number, dt: readonly number[], texts: readonly string[], index = 0): RunOf<'text-chunks'> {
+  return { type: 'text-chunks', time0, index, dt, texts }
+}
+
+function reasoningRun(
+  time0: number,
+  dt: readonly number[],
+  texts: readonly string[],
+  index = 0,
+): RunOf<'reasoning-chunks'> {
+  return { type: 'reasoning-chunks', time0, index, dt, texts }
+}
+
+function toolRun(
+  time0: number,
+  dt: readonly number[],
+  args: readonly string[],
+  name?: string,
+): RunOf<'tool-call-chunks'> {
+  return {
+    type: 'tool-call-chunks', time0, index: 0, dt, id: ToolCallId('call'),
+    ...name === undefined ? {} : { name },
+    args,
+  }
+}
+
+function raw(time: number, chunk: StreamChunk): AssistantStreamRecord {
+  return { type: 'chunk', time, chunk }
+}
+
+describe('stream chunk classification', () => {
+  it('recognizes the first token as a non-empty fragment or a name-bearing Tool-call delta', () => {
+    expect(isTokenDelta({ type: 'text-delta', index: 0, text: ' ' })).toBe(true)
+    expect(isTokenDelta({ type: 'text-delta', index: 0, text: '' })).toBe(false)
+    expect(isTokenDelta({ type: 'reasoning-delta', index: 0, text: 'r' })).toBe(true)
+    expect(isTokenDelta({ type: 'reasoning-delta', index: 0, text: '' })).toBe(false)
+    expect(isTokenDelta({ type: 'tool-call-delta', index: 0, id: ToolCallId('c'), argumentsDelta: '{' })).toBe(true)
+    expect(isTokenDelta({ type: 'tool-call-delta', index: 0, id: ToolCallId('c'), argumentsDelta: '' })).toBe(false)
+    expect(isTokenDelta({ type: 'tool-call-delta', index: 0, id: ToolCallId('c'), name: 'read', argumentsDelta: '' })).toBe(true)
+    expect(isTokenDelta({ type: 'tool-call-delta', index: 0, id: ToolCallId('c'), name: '', argumentsDelta: '' })).toBe(true)
+    expect(isTokenDelta({ type: 'block-start', index: 0, blockType: 'text' })).toBe(false)
+    expect(isTokenDelta({ type: 'block-end', index: 0, block: { type: 'text', text: 'x' } })).toBe(false)
+    expect(isTokenDelta({ type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } })).toBe(false)
+    expect(isTokenDelta({ type: 'finish', reason: { kind: 'stop' } })).toBe(false)
+  })
+
+  it('classifies reader-visible chunks by non-whitespace text and non-Tool-call block kinds', () => {
+    expect(isVisibleChunk({ type: 'text-delta', index: 0, text: ' \t\n' })).toBe(false)
+    expect(isVisibleChunk({ type: 'text-delta', index: 0, text: ' x' })).toBe(true)
+    expect(isVisibleChunk({ type: 'reasoning-delta', index: 0, text: '\u00a0' })).toBe(false)
+    expect(isVisibleChunk({ type: 'reasoning-delta', index: 0, text: 'r' })).toBe(true)
+    expect(isVisibleChunk({ type: 'block-start', index: 0, blockType: 'text' })).toBe(false)
+    expect(isVisibleChunk({ type: 'block-start', index: 0, blockType: 'reasoning' })).toBe(false)
+    expect(isVisibleChunk({ type: 'block-start', index: 0, blockType: 'tool-call' })).toBe(false)
+    expect(isVisibleChunk({ type: 'block-start', index: 0, blockType: 'image' })).toBe(true)
+    expect(isVisibleChunk({ type: 'block-end', index: 0, block: { type: 'text', text: '  ' } })).toBe(false)
+    expect(isVisibleChunk({ type: 'block-end', index: 0, block: { type: 'reasoning', text: 'why' } })).toBe(true)
+    expect(isVisibleChunk({
+      type: 'block-end', index: 0, block: { type: 'tool-call', id: ToolCallId('c'), name: 'read', arguments: '{}' },
+    })).toBe(false)
+    expect(isVisibleChunk({ type: 'block-end', index: 0, block: { type: 'image', attachment: {} as never } })).toBe(true)
+    expect(isVisibleChunk({ type: 'tool-call-delta', index: 0, id: ToolCallId('c'), name: 'read', argumentsDelta: '{}' })).toBe(false)
+    expect(isVisibleChunk({ type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } })).toBe(false)
+    expect(isVisibleChunk({ type: 'finish', reason: { kind: 'stop' } })).toBe(false)
+  })
+
+  it('counts visible text only from text deltas and completed text blocks', () => {
+    expect(chunkHasVisibleText({ type: 'text-delta', index: 0, text: 'a' })).toBe(true)
+    expect(chunkHasVisibleText({ type: 'text-delta', index: 0, text: '\r\n' })).toBe(false)
+    expect(chunkHasVisibleText({ type: 'reasoning-delta', index: 0, text: 'a' })).toBe(false)
+    expect(chunkHasVisibleText({ type: 'block-end', index: 0, block: { type: 'text', text: ' a ' } })).toBe(true)
+    expect(chunkHasVisibleText({ type: 'block-end', index: 0, block: { type: 'text', text: ' ' } })).toBe(false)
+    expect(chunkHasVisibleText({ type: 'block-end', index: 0, block: { type: 'reasoning', text: 'a' } })).toBe(false)
+    expect(chunkHasVisibleText({ type: 'block-start', index: 0, blockType: 'text' })).toBe(false)
+    expect(chunkHasVisibleText({ type: 'finish', reason: { kind: 'stop' } })).toBe(false)
+  })
+})
+
+describe('packed run boundaries', () => {
+  it('reconstructs the first token member time from time0 and the preceding gaps', () => {
+    expect(runFirstTokenTime(textRun(1_000, [5, -3, 10], ['', '', 'x', 'y']))).toBe(1_002)
+    expect(runFirstTokenTime(textRun(1_000, [5], ['a', 'b']))).toBe(1_000)
+    expect(runFirstTokenTime(reasoningRun(7, [1, 1], ['', '', '']))).toBeUndefined()
+    expect(runFirstTokenTime(toolRun(50, [2, 2], ['', '', '{']))).toBe(54)
+    expect(runFirstTokenTime(toolRun(50, [2], ['', '']))).toBeUndefined()
+    expect(runFirstTokenTime(toolRun(50, [2], ['', ''], 'read'))).toBe(50)
+  })
+
+  it('reconstructs the first visible member time from non-whitespace fragments only', () => {
+    expect(runFirstVisibleTime(textRun(1_000, [5, 1, 1], ['', '   ', '\t', 'answer']))).toBe(1_007)
+    expect(runFirstVisibleTime(reasoningRun(20, [3], [' ', 'think']))).toBe(23)
+    expect(runFirstVisibleTime(textRun(20, [3], [' ', '\n']))).toBeUndefined()
+    expect(runFirstVisibleTime(toolRun(20, [3], ['{"x":', '1}'], 'read'))).toBeUndefined()
+  })
+
+  it('stops reading fragments at the first qualifying member', () => {
+    const token = countedFragments(['', 'x', 'unread', 'unread'])
+    expect(runFirstTokenTime({ ...textRun(0, [1, 1, 1], []), texts: token.fragments })).toBe(1)
+    expect(token.reads()).toBe(2)
+
+    const visible = countedFragments([' ', ' ', 'v', 'unread'])
+    expect(runFirstVisibleTime({ ...reasoningRun(0, [1, 1, 1], []), texts: visible.fragments })).toBe(2)
+    expect(visible.reads()).toBe(3)
+
+    const named = countedFragments(['unread'])
+    expect(runFirstTokenTime({ ...toolRun(9, [], [], 'read'), args: named.fragments })).toBe(9)
+    expect(named.reads()).toBe(0)
+  })
+})
+
+describe('compact stream readers', () => {
+  const usage = { inputTokens: 10, outputTokens: 4 }
+  const laterUsage = { inputTokens: 10, outputTokens: 9 }
+  const stream: readonly AssistantStreamRecord[] = [
+    raw(100, { type: 'block-start', index: 0, blockType: 'reasoning' }),
+    reasoningRun(101, [2, 2], ['', ' ', 'think']),
+    raw(106, { type: 'block-end', index: 0, block: { type: 'reasoning', text: ' think' } }),
+    raw(107, { type: 'block-start', index: 1, blockType: 'text' }),
+    textRun(108, [1, 1], ['\n', 'ans', 'wer'], 1),
+    raw(111, { type: 'block-end', index: 1, block: { type: 'text', text: '\nanswer' } }),
+    raw(112, { type: 'usage', usage }),
+    raw(113, { type: 'usage', usage: laterUsage }),
+    raw(114, { type: 'finish', reason: { kind: 'stop' } }),
+  ]
+
+  it('answers first token, visibility, text, and usage questions from records', () => {
+    expect(assistantStreamFirstTokenTime(stream)).toBe(103)
+    expect(assistantStreamHasVisibleContent(stream)).toBe(true)
+    expect(assistantStreamHasVisibleText(stream)).toBe(true)
+    expect(lastAssistantStreamChunk(stream, 'usage')?.usage).toBe(laterUsage)
+    expect(lastAssistantStreamChunk(stream, 'finish')).toStrictEqual({ type: 'finish', reason: { kind: 'stop' } })
+    expect(lastAssistantStreamChunk(stream, 'block-start')).toStrictEqual({ type: 'block-start', index: 1, blockType: 'text' })
+    expect(assistantStreamChunks(stream, 'block-end').map(chunk => chunk.index)).toStrictEqual([0, 1])
+    expect(assistantStreamChunks(stream, 'usage').map(chunk => chunk.usage)).toStrictEqual([usage, laterUsage])
+    expect(joinAssistantStreamText(stream)).toBe('\nanswer')
+  })
+
+  it('reports absence on empty, whitespace-only, and Tool-call-only streams', () => {
+    const silent: readonly AssistantStreamRecord[] = [
+      raw(1, { type: 'block-start', index: 0, blockType: 'tool-call' }),
+      toolRun(2, [1], ['', ''], 'read'),
+      raw(4, { type: 'block-end', index: 0, block: { type: 'tool-call', id: ToolCallId('call'), name: 'read', arguments: '' } }),
+      textRun(5, [1], [' ', '\t'], 1),
+      reasoningRun(7, [], ['  '], 2),
+      raw(8, { type: 'block-end', index: 1, block: { type: 'text', text: ' \t' } }),
+    ]
+    expect(assistantStreamFirstTokenTime([])).toBeUndefined()
+    expect(assistantStreamFirstTokenTime(silent)).toBe(2)
+    expect(assistantStreamHasVisibleContent([])).toBe(false)
+    expect(assistantStreamHasVisibleContent(silent)).toBe(false)
+    expect(assistantStreamHasVisibleText([])).toBe(false)
+    expect(assistantStreamHasVisibleText(silent)).toBe(false)
+    expect(lastAssistantStreamChunk(silent, 'usage')).toBeUndefined()
+    expect(lastAssistantStreamChunk([], 'finish')).toBeUndefined()
+    expect(assistantStreamChunks(silent, 'usage')).toStrictEqual([])
+    expect(joinAssistantStreamText(silent)).toBe(' \t')
+    expect(joinAssistantStreamText([])).toBe('')
+  })
+
+  it('reads raw text deltas and empty-argument Tool-call deltas the accumulator kept as chunks', () => {
+    const degenerate: readonly AssistantStreamRecord[] = [
+      raw(1, { type: 'tool-call-delta', index: 0, id: ToolCallId(''), argumentsDelta: '' }),
+      raw(2, { type: 'tool-call-delta', index: 0, id: ToolCallId('call'), name: '', argumentsDelta: '' }),
+      raw(3, { type: 'text-delta', index: 1, text: ' ' }),
+      raw(4, { type: 'text-delta', index: 1, text: 'raw' }),
+    ]
+    expect(assistantStreamFirstTokenTime(degenerate)).toBe(2)
+    expect(assistantStreamHasVisibleContent(degenerate)).toBe(true)
+    expect(assistantStreamHasVisibleContent(degenerate.slice(0, 3))).toBe(false)
+    expect(assistantStreamHasVisibleText(degenerate)).toBe(true)
+    expect(assistantStreamHasVisibleText(degenerate.slice(0, 3))).toBe(false)
+    expect(joinAssistantStreamText(degenerate)).toBe(' raw')
+  })
+
+  it('stops at the first qualifying record', () => {
+    expect(assistantStreamFirstTokenTime([textRun(5, [], ['x']), unreachableRecord])).toBe(5)
+    expect(assistantStreamFirstTokenTime([
+      raw(6, { type: 'tool-call-delta', index: 0, id: ToolCallId('call'), name: '', argumentsDelta: '' }),
+      unreachableRecord,
+    ])).toBe(6)
+    expect(assistantStreamHasVisibleContent([raw(1, { type: 'block-start', index: 0, blockType: 'image' }), unreachableRecord])).toBe(true)
+    expect(assistantStreamHasVisibleContent([reasoningRun(1, [], ['r']), unreachableRecord])).toBe(true)
+    expect(assistantStreamHasVisibleText([textRun(1, [], ['t']), unreachableRecord])).toBe(true)
+    expect(assistantStreamHasVisibleText([
+      raw(1, { type: 'block-end', index: 0, block: { type: 'text', text: 't' } }),
+      unreachableRecord,
+    ])).toBe(true)
+    expect(lastAssistantStreamChunk([unreachableRecord, raw(9, { type: 'finish', reason: { kind: 'stop' } })], 'finish')?.type)
+      .toBe('finish')
+  })
+
+  it('assembles the same blocks, usage, finish, and replay state as the expanded members', () => {
+    const accumulator = new AssistantStreamAccumulator()
+    const chunks: readonly StreamChunk[] = [
+      { type: 'block-start', index: 0, blockType: 'reasoning' },
+      { type: 'reasoning-delta', index: 0, text: 'th' },
+      { type: 'reasoning-delta', index: 0, text: 'ink' },
+      { type: 'block-end', index: 0, block: { type: 'reasoning', text: 'think' } },
+      { type: 'text-delta', index: 1, text: 'an' },
+      { type: 'text-delta', index: 1, text: 'swer' },
+      { type: 'tool-call-delta', index: 2, id: ToolCallId('call-1'), name: 'read', argumentsDelta: '' },
+      { type: 'tool-call-delta', index: 2, id: ToolCallId('call-1'), argumentsDelta: '{"path":' },
+      { type: 'tool-call-delta', index: 2, id: ToolCallId('call-1'), argumentsDelta: '"a"}' },
+      { type: 'tool-call-delta', index: 3, id: ToolCallId(''), argumentsDelta: '{}' },
+      { type: 'usage', usage: { inputTokens: 3, outputTokens: 2 } },
+      { type: 'finish', reason: { kind: 'tool-calls' }, replayState: { response: { id: 'r' } } },
+    ]
+    for (const [index, chunk] of chunks.entries()) accumulator.push({ time: 1_000 + index, chunk })
+    const stream = accumulator.snapshot()
+    expect(stream.filter(record => record.type !== 'chunk')).toHaveLength(4)
+
+    const expanded = new BlockAssembler()
+    for (const member of expandAssistantStream(stream)) expanded.push(member.chunk)
+    const assembled = assembleAssistantStream(stream)
+
+    expect(assembled.blocks()).toStrictEqual(expanded.blocks())
+    expect(assembled.blocks().map(block => block.type)).toStrictEqual(['reasoning', 'text', 'tool-call', 'tool-call'])
+    expect(assembled.usage).toStrictEqual({ inputTokens: 3, outputTokens: 2 })
+    expect(assembled.finish).toStrictEqual({ kind: 'tool-calls' })
+    expect(assembled.replayState).toStrictEqual(expanded.replayState)
+
+    const reused = new BlockAssembler()
+    expect(assembleAssistantStream([], reused)).toBe(reused)
+    expect(reused.blocks()).toStrictEqual([])
+    expect(() => assembleAssistantStream([{ type: 'future' }] as never)).toThrow(/unreachable variant in assembleAssistantStream/)
+  })
+})
@@ -6,7 +6,7 @@

 import { Context, Service } from '@deepseek-ai/cordis'
 import z from '@deepseek-ai/schemastery'
-import { BlockAssembler, expandAssistantStream } from '@deepseek-ai/dsh-llm'
+import { assembleAssistantStream } from '@deepseek-ai/dsh-llm'
 import type { LlmImageRequestPricing, LlmRuntime, Message, TokenUsage } from '@deepseek-ai/dsh-llm'
 import { deepFreeze } from '@deepseek-ai/dsh-util-values'
 import type {
@@ -316,9 +316,7 @@ export class TokenMeter extends Service {
  private _estimateProviderAssistant(
    event: SessionEvent<'assistant/message'>,
  ): number {
-    const assembler = new BlockAssembler()
-    for (const member of expandAssistantStream(event.data.stream)) assembler.push(member.chunk)
-    const providerContent = assembler.blocks()
+    const providerContent = assembleAssistantStream(event.data.stream).blocks()
    return providerContent.length === 0 ? 0 : estimateContent(providerContent) + ROLE_OVERHEAD
  }
 }
@@ -1,4 +1,4 @@
-import { expandAssistantStream } from '@deepseek-ai/dsh-llm/assistant-stream'
+import { lastAssistantStreamChunk } from '@deepseek-ai/dsh-llm/assistant-stream'
 import type { AssistantMessage, TokenUsage } from '@deepseek-ai/dsh-llm/types'
 import type {} from '@deepseek-ai/dsh-llm-retry/types'
 import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
@@ -75,11 +75,7 @@ function messageRoute(message: AssistantMessage): TurnTokenUsageRoute | undefine
 }

 function streamUsage(stream: SessionEvent<'assistant/message'>['data']['stream']): TokenUsage | undefined {
-  let sample: TokenUsage | undefined
-  for (const member of expandAssistantStream(stream)) {
-    if (member.chunk.type === 'usage') sample = member.chunk.usage
-  }
-  return sample
+  return lastAssistantStreamChunk(stream, 'usage')?.usage
 }

 function normalizeUsage(usage: TokenUsage, route?: TurnTokenUsageRoute): NormalizedAttempt | undefined {
@@ -234,10 +230,7 @@ export function deriveTurnTokenUsage(events: readonly SessionEvent[]): TurnToken
        invalid = true
        continue
      }
-      let sample: TokenUsage | undefined = state.sample
-      for (const member of expandAssistantStream(event.data.stream)) {
-        if (member.chunk.type === 'usage') sample = member.chunk.usage
-      }
+      const sample: TokenUsage | undefined = streamUsage(event.data.stream) ?? state.sample
      state = { kind: 'open', turn, step: event.data.step, ...(sample === undefined ? {} : { sample }) }
      if (!closeOpen()) invalid = true
      else state = { kind: 'finishClosed', turn, step: event.data.step }
@@ -3,7 +3,7 @@
 */

 import { z } from 'zod'
-import { expandAssistantStream, type TokenUsage } from '@deepseek-ai/dsh-llm'
+import { lastAssistantStreamChunk, type TokenUsage } from '@deepseek-ai/dsh-llm'
 import type {} from '@deepseek-ai/dsh-llm-retry/types'
 import { SessionSeq } from '@deepseek-ai/dsh-session'
 import type { SessionEvent } from '@deepseek-ai/dsh-session'
@@ -82,10 +82,7 @@ const pressureFrom = (usage: TokenUsage): number =>
 function usageOf(event: SessionEvent): TokenUsage | undefined {
  if (event.type === 'assistant/message' && event.data.usage !== undefined) return event.data.usage
  if (event.type !== 'assistant/message' && event.type !== 'assistant/attempt') return undefined
-  for (const member of expandAssistantStream(event.data.stream).toReversed()) {
-    if (member.chunk.type === 'usage') return member.chunk.usage
-  }
-  return undefined
+  return lastAssistantStreamChunk(event.data.stream, 'usage')?.usage
 }

 declare module '@deepseek-ai/dsh-session-projection/types' {
@@ -24,30 +24,9 @@
 */

 import { z } from 'zod'
-import { expandAssistantStream, type AssistantStreamRecord, type StreamChunk } from '@deepseek-ai/dsh-llm'
+import { assistantStreamFirstTokenTime } from '@deepseek-ai/dsh-llm'
 import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'

-/* jscpd:ignore-start -- Session Stats owns its whole-log timing projection independently. */
-
-/** Whether a stream chunk carries a non-empty first-token delta. */
-function isTokenDelta(chunk: StreamChunk): boolean {
-  switch (chunk.type) {
-    case 'text-delta':
-    case 'reasoning-delta':
-      return chunk.text !== ''
-    case 'tool-call-delta':
-      return chunk.argumentsDelta !== '' || chunk.name !== undefined
-    default:
-      return false
-  }
-}
-
-/** First non-empty token timestamp in one durable Assistant stream. */
-function firstTokenTime(stream: readonly AssistantStreamRecord[]): number | null {
-  return expandAssistantStream(stream).find(member => isTokenDelta(member.chunk))?.time ?? null
-}
-
-/* jscpd:ignore-end */

 /** Accumulated whole-log figures (the view is exactly these totals). */
 interface SessionStatsTotals {
@@ -159,14 +138,14 @@ export const sessionStatsProjectionDefinition = {
      case 'assistant/attempt': {
        const open = state.openStep
        if (open === null || open.turn !== event.data.turn || open.step !== event.data.step) return state
-        const first = firstTokenTime(event.data.stream)
+        const first = assistantStreamFirstTokenTime(event.data.stream) ?? null
        if (open.firstTokenTime !== null || first === null) return state
        return { ...state, openStep: { ...open, firstTokenTime: first } }
      }
      case 'assistant/message': {
        const open = state.openStep
        if (open === null || open.turn !== event.data.turn || open.step !== event.data.step) return state
-        const firstToken = open.firstTokenTime ?? firstTokenTime(event.data.stream)
+        const firstToken = open.firstTokenTime ?? assistantStreamFirstTokenTime(event.data.stream) ?? null
        // One assembled message per step: closing the boundary means a
        // defensive duplicate cannot accrue twice.
        const next: SessionStatsState = {
@@ -10,7 +10,7 @@
 * @module @deepseek-ai/dsh-subagent/assistant-output
 */

-import { expandAssistantStream, type ContentBlock } from '@deepseek-ai/dsh-llm'
+import { joinAssistantStreamText, type ContentBlock } from '@deepseek-ai/dsh-llm'
 import type { SessionEvent } from '@deepseek-ai/dsh-session'

 /**
@@ -35,9 +35,7 @@ export class AssistantOutputFold {
      if (content.length > 0) this.message = content
    }
    if (event.type === 'assistant/message' || event.type === 'assistant/attempt') {
-      for (const { chunk } of expandAssistantStream(event.data.stream)) {
-        if (chunk.type === 'text-delta') this.pushText(chunk.text)
-      }
+      this.pushText(joinAssistantStreamText(event.data.stream))
    }
  }