feat(agent-loop): finalize a cancelled stream's delivered prefix

A turn cancelled mid-stream dropped everything the interrupted step had
streamed: chunks stayed in the log for replay, but no assistant/message
joined the surface, so the next request carried nothing the user had
watched stream. Follow-ups after cancel could not connect and forks
inherited the gap.

Keep the streaming attempt alive across the step's request loop and, when
an abort escapes with the attempt uncommitted, finalize its user-visible
prefix as the step's ordinary assistant/message citing the logged chunk
seqs. BlockAssembler.interruptedBlocks() owns the safe subset next to the
max-tokens rule: closed and open text/reasoning blocks with content, in
stream order; tool calls, empty blocks, and unknown open blocks drop.
Retry clears the attempt first, so an abort after llm/retry finalizes
nothing from the reset stream. Tool-phase cancellation and provider
failures keep their existing shapes.

.agents/notes/implemented/architecture/2026-08-10-cancelled-stream-prefix-finalize.i18n.yaml

@@ -0,0 +1,6 @@
+# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
+# side as of the last confirmed-consistent state. Both languages carry equal authority;
+# after editing either side, bring the other along and re-record with:
+#   pnpm run verify-translation-pairing --write .agents/notes/implemented/architecture/2026-08-10-cancelled-stream-prefix-finalize.md
+2026-08-10-cancelled-stream-prefix-finalize.md: a2201cd17fd89744590bd4024bff80f3e3835e39
+2026-08-10-cancelled-stream-prefix-finalize.zh.md: 08abb5c097bdf3c6c21e0a96873ab4a995e8f3e9

.agents/notes/implemented/architecture/2026-08-10-cancelled-stream-prefix-finalize.md

@@ -0,0 +1,37 @@
+# Agent Note: Cancelled streams finalize their delivered prefix
+
+Status: implemented
+
+English | [中文](2026-08-10-cancelled-stream-prefix-finalize.zh.md)
+
+## Problem
+
+A turn cancelled mid-stream used to drop everything the interrupted step had streamed: `assistant/chunk` events stayed in the log for replay, but no `assistant/message` joined the surface, so `deriveMessages()` carried nothing of the interrupted output into the next request. The user had watched the text stream and clients keep rendering it after the abort, yet from the model's perspective that text was never said. A follow-up like "expand on your second point" could not connect, and a fork taken at the cancelled turn inherited a surface missing content its owner had read. This divergence was never a weighed decision — the first agent-loop implementation checked the abort signal inside the chunk loop and threw before the finalize append, and the later surface whitelist froze that shape in.
+
+The governing principle this violated: what the user can see, the next model request contains.
+
+## Decision
+
+`Agent.step()` keeps the current streaming attempt (assembler, logged chunk seqs, provider route) alive across the request loop. When an abort escapes the step while an attempt is uncommitted, `appendInterruptedAssistant` finalizes the attempt's user-visible prefix as the step's ordinary `assistant/message` — `surfaceOp: 'append'`, `sourceEventSeqs` citing exactly the logged chunks — before the abort continues to the `step/end`/`turn/end` teardown. A retry decision clears the attempt first: `llm/retry` resets what clients render, so an abort after it finalizes nothing from the abandoned attempt.
+
+`BlockAssembler.interruptedBlocks()` owns what is safe to finalize, next to the existing max-tokens truncation rule: closed and open `text`/`reasoning` blocks with non-whitespace content, in stream order. Tool calls are dropped whole — interruption precedes dispatch, so a kept call would demand a fabricated result — as are empty blocks and open blocks of unknown type. When nothing survives, no event is appended and the turn keeps its previous shape: chunks, `step/end`, `turn/end` aborted.
+
+Cancellation during tool execution is untouched: the tool-call message was already finalized, started calls drain into real results, and undispatched calls keep their synthetic `ABORTED_BEFORE_DISPATCH` pairs. Provider failures (terminal error or aborted finishes) still commit nothing; only turn cancellation finalizes a prefix, because only there did the user watch content that would otherwise vanish from the model's history.
+
+## Alternatives considered
+
+**Keep dropping the prefix (status quo).** Safe and simple, but it makes cancel-then-redirect — a high-frequency flow — manufacture a user-visible/model-visible split on every use, and fork inherits the gap. Rejected: the split's cost recurs; the finalize cost is one-time.
+
+**Project the prefix at request time from the logged chunks.** No new surface event; `deriveMessages()` would assemble chunk prefixes for aborted steps. Rejected: it moves assembly policy into every surface consumer, breaks the "three message-producing event types" surface contract, and makes the derived history depend on non-surface events.
+
+**Finalize complete tool-call blocks too, with synthetic aborted results.** Preserves more of the model's intent. Rejected: the calls never dispatched and never rendered as tool cards, so parity does not ask for them, and fabricated result pairs add model-visible noise; the max-tokens rule already drops undispatchable calls.
+
+**Append an explicit interruption marker (`[interrupted by user]` user message).** What Claude Code does; tells the model its answer was cut off rather than complete. Deferred, not rejected: it is a separate model-visible vocabulary decision (source kind, UI rendering, locale strings) stacked on top of this parity fix, and the durable `turn/end aborted` already records the fact for a future projection to use.
+
+## Consequences
+
+The surface now contains what the user saw at the moment of cancellation, so post-cancel follow-ups and forks connect. The cancel and goal snapshot fixtures record the finalized prefix event, and the ACP bridge forwards it as a final `agent_message_chunk` update after the cancelled stop reason. An interrupted step's `assistant/message` can now carry a mid-sentence prefix; consumers reading the aborted `turn/end` can classify it. Terminal provider errors keep the old behavior — their streamed prefix still vanishes from the surface — an asymmetry deliberately left for a follow-up decision because error turns end without the user choosing to stop.
+
+## Testing
+
+`packages/core/agent-loop/tests/cancel.spec.ts` pins mid-stream finalize (content, cited seqs, event order, next-request parity), reasoning-only finalize, half-streamed tool-call dropping, and the nothing-to-finalize case. `packages/llm/llm/tests/assembler.spec.ts` pins `interruptedBlocks()`. The keyless `cancel` ACP snapshot and the goal-session snapshot carry the assembled-application transcript.

.agents/notes/implemented/architecture/2026-08-10-cancelled-stream-prefix-finalize.zh.md

@@ -0,0 +1,37 @@
+# Agent Note: 被取消的流定稿其已送达前缀
+
+Status: implemented
+
+[English](2026-08-10-cancelled-stream-prefix-finalize.md) | 中文
+
+## Problem
+
+轮次在流式输出中途被取消时，被打断的 step 已流出的内容过去会被整体丢弃：`assistant/chunk` 事件留在日志里供回放，但没有任何 `assistant/message` 进入 surface，`deriveMessages()` 不会把被打断的输出带进下一次请求。用户看着文字流出来，客户端在 abort 之后也继续显示它，但从模型的视角那段话从没说过。取消后追问「第二点展开讲讲」接不上，在被取消的轮次上 fork 出的分支继承的 surface 也缺少其主人读过的内容。这个分歧从来不是权衡后的决定：agent loop 的第一版实现就在分片循环里检查 abort 信号并在定稿 append 之前抛出，后来的 surface 白名单把这个形状固化了下来。
+
+它违反的主导原则是：用户能看到什么，下一次模型请求就包含什么。
+
+## Decision
+
+`Agent.step()` 让当前流式尝试（assembler、已记录的分片 seq、提供方路由）在请求循环之间保持存活。当 abort 在尝试未提交时逃出 step，`appendInterruptedAssistant` 会在 abort 继续走向 `step/end`/`turn/end` 收尾之前，把该尝试的用户可见前缀定稿为该 step 的普通 `assistant/message`，`surfaceOp: 'append'`，`sourceEventSeqs` 恰好引用已记录的分片。重试决定会先清空尝试：`llm/retry` 会重置客户端渲染的内容，因此其后的 abort 不会从被放弃的尝试中定稿任何东西。
+
+`BlockAssembler.interruptedBlocks()` 拥有「什么可以安全定稿」的规则，与既有的 max-tokens 截断规则放在一起：按流顺序保留内容非空白的已闭合与未闭合 `text`/`reasoning` 块。工具调用整块丢弃，因为打断先于分派，保留的调用会要求捏造一个结果；空块和未知类型的未闭合块同样丢弃。没有内容存活时不追加任何事件，轮次保持原有形状：分片、`step/end`、`turn/end` aborted。
+
+工具执行期间的取消不受影响：带工具调用的消息此前已定稿，已启动的调用排空为真实结果，未分派的调用保留合成的 `ABORTED_BEFORE_DISPATCH` 对。提供方故障（终局 error 或 aborted finish）仍然不提交任何内容；只有轮次取消定稿前缀，因为只有在那里用户看到过将从模型历史中消失的内容。
+
+## Alternatives considered
+
+**继续丢弃前缀（维持现状）。** 安全且简单，但它让「取消然后转向」这个高频操作每次都制造一个用户可见与模型可见的分裂，fork 也继承这个缺口。否决：分裂的成本反复发生，定稿的成本只付一次。
+
+**在请求时从已记录分片投影前缀。** 不加新 surface 事件，让 `deriveMessages()` 为 aborted step 装配分片前缀。否决：它把装配策略搬进每个 surface 消费者，破坏「三类产生消息的事件」的 surface 合同，并让派生历史依赖非 surface 事件。
+
+**连完整的工具调用块也定稿，配合成的 aborted 结果。** 保留更多模型意图。否决：这些调用从未分派也从未渲染成工具卡片，对等原则并不要求它们，捏造的结果对还会增加模型可见的噪音；max-tokens 规则本来就丢弃不可分派的调用。
+
+**追加显式打断标记（`[interrupted by user]` 用户消息）。** Claude Code 的做法，告诉模型回答是被切断的而不是完整的。搁置而非否决：它是叠加在本次对等修复之上的一个独立的模型可见词汇决定（source 种类、UI 渲染、locale 文案），而持久的 `turn/end aborted` 已经记录了这个事实，未来的投影可以使用。
+
+## Consequences
+
+surface 现在包含取消瞬间用户看到的内容，取消后的追问和 fork 都能接上。cancel 与 goal 两组快照 fixture 记录了定稿前缀事件，ACP 桥在 cancelled stop reason 之后把它作为最后一条 `agent_message_chunk` 更新转发。被打断 step 的 `assistant/message` 现在可能带着一个中途截断的前缀；消费者读到 aborted 的 `turn/end` 即可归类。终局提供方错误保持旧行为，其已流出前缀仍会从 surface 消失，这个不对称是有意留给后续决定的，因为 error 轮次的结束不是用户主动选择的停止。
+
+## Testing
+
+`packages/core/agent-loop/tests/cancel.spec.ts` 固定了流中取消的定稿（内容、引用的 seq、事件顺序、下一请求的对等）、仅 reasoning 的定稿、半流式工具调用的丢弃和无可定稿内容的情形。`packages/llm/llm/tests/assembler.spec.ts` 固定了 `interruptedBlocks()`。keyless 的 `cancel` ACP 快照和 goal-session 快照承载装配后应用的 transcript。

docs/persistence-catalog.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write docs/persistence-catalog.md
-persistence-catalog.md: a17cae015eaa107a900069de916dddb216b87ec7
-persistence-catalog.zh.md: 3aef073dedcff0b6addb99d7c287f4e5f372c402
+persistence-catalog.md: 54598207c763a1801cd1f6b15d9887bf7875107f
+persistence-catalog.zh.md: 7730f03ee220e43f21045ff9d0078241cd352845

docs/persistence-catalog.md

@@ -79,7 +79,7 @@ export type SessionEvent<T extends SessionEventType = SessionEventType> = {
 }[T]
 ```
 
-Sources: [`packages/core/session/src/types.ts:308`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:315`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:344`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:376`](../packages/core/session/src/types.ts)
+Sources: [`packages/core/session/src/types.ts:311`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:318`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:347`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:379`](../packages/core/session/src/types.ts)
 
 ## Events
 
@@ -185,14 +185,17 @@ Source: [`packages/core/session/src/types.ts:238`](../packages/core/session/src/
  * Assembled assistant message for one step (derived history uses this).
  * Carries the step's `usage` when the adapter reported token accounting, so
  * the model output and its accounting travel together (there is no separate
- * usage record). `usage` is absent when the adapter reported none.
+ * usage record). `usage` is absent when the adapter reported none. A turn
+ * cancelled mid-stream finalizes its delivered text/reasoning prefix as this
+ * event (undispatched tool calls are absent); an aborted turn with no such
+ * event streamed no visible content.
  */
 'assistant/message': { turn: number; step: number; message: AssistantMessage; usage?: TokenUsage }
 ```
 
 Types: [TokenUsage](subsystems/llm-streaming.md)
 
-Source: [`packages/core/session/src/types.ts:245`](../packages/core/session/src/types.ts)
+Source: [`packages/core/session/src/types.ts:248`](../packages/core/session/src/types.ts)
 
 ### `command/*`
 
@@ -472,7 +475,7 @@ Source: [`packages/plan/plan-mode/src/index.ts:52`](../packages/plan/plan-mode/s
 'request/context': RequestContext
 ```
 
-Source: [`packages/core/session/src/types.ts:281`](../packages/core/session/src/types.ts)
+Source: [`packages/core/session/src/types.ts:284`](../packages/core/session/src/types.ts)
 
 #### `request/header` — log-only
 
@@ -484,7 +487,7 @@ Source: [`packages/core/session/src/types.ts:281`](../packages/core/session/src/
 'request/header': { header: EpochHeader; reason: RequestHeaderReason }
 ```
 
-Source: [`packages/core/session/src/types.ts:276`](../packages/core/session/src/types.ts)
+Source: [`packages/core/session/src/types.ts:279`](../packages/core/session/src/types.ts)
 
 ### `sandbox/*`
 
@@ -537,7 +540,7 @@ Source: [`packages/sandbox/sandbox-policy/src/session-mode.ts:33`](../packages/s
 'session/end-seed': Record<string, never>
 ```
 
-Source: [`packages/core/session/src/types.ts:304`](../packages/core/session/src/types.ts)
+Source: [`packages/core/session/src/types.ts:307`](../packages/core/session/src/types.ts)
 
 #### `session/title` — log-only
 
@@ -612,7 +615,7 @@ Source: [`packages/subagent/subagent/src/descriptor.ts:37`](../packages/subagent
 
 Types: [TodoItem](subsystems/session.md)
 
-Source: [`packages/core/session/src/types.ts:271`](../packages/core/session/src/types.ts)
+Source: [`packages/core/session/src/types.ts:274`](../packages/core/session/src/types.ts)
 
 ### `tool/*`
 
@@ -629,7 +632,7 @@ Source: [`packages/core/session/src/types.ts:271`](../packages/core/session/src/
 
 Types: [CallId](subsystems/core.md)
 
-Source: [`packages/core/session/src/types.ts:251`](../packages/core/session/src/types.ts)
+Source: [`packages/core/session/src/types.ts:254`](../packages/core/session/src/types.ts)
 
 #### `tool/code-dispatch` — log-only
 
@@ -698,7 +701,7 @@ Source: [`packages/core/tools/src/types.ts:40`](../packages/core/tools/src/types
 }
 ```
 
-Source: [`packages/core/session/src/types.ts:263`](../packages/core/session/src/types.ts)
+Source: [`packages/core/session/src/types.ts:266`](../packages/core/session/src/types.ts)
 
 ### `turn/*`
 

docs/persistence-catalog.zh.md

@@ -81,7 +81,7 @@ export type SessionEvent<T extends SessionEventType = SessionEventType> = {
 }[T]
 ```
 
-来源：[`packages/core/session/src/types.ts:308`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:315`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:344`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:376`](../packages/core/session/src/types.ts)
+来源：[`packages/core/session/src/types.ts:311`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:318`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:347`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:379`](../packages/core/session/src/types.ts)
 
 ## 事件
 
@@ -187,14 +187,17 @@ export type SessionEvent<T extends SessionEventType = SessionEventType> = {
  * Assembled assistant message for one step (derived history uses this).
  * Carries the step's `usage` when the adapter reported token accounting, so
  * the model output and its accounting travel together (there is no separate
- * usage record). `usage` is absent when the adapter reported none.
+ * usage record). `usage` is absent when the adapter reported none. A turn
+ * cancelled mid-stream finalizes its delivered text/reasoning prefix as this
+ * event (undispatched tool calls are absent); an aborted turn with no such
+ * event streamed no visible content.
  */
 'assistant/message': { turn: number; step: number; message: AssistantMessage; usage?: TokenUsage }
 ```
 
 类型：[TokenUsage](subsystems/llm-streaming.md)
 
-来源：[`packages/core/session/src/types.ts:245`](../packages/core/session/src/types.ts)
+来源：[`packages/core/session/src/types.ts:248`](../packages/core/session/src/types.ts)
 
 ### `command/*`
 

docs/subsystems/llm-streaming.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write docs/subsystems/llm-streaming.md
-llm-streaming.md: 4d450c19ec2bbfacabcefc83466e67c8a6c82bd6
-llm-streaming.zh.md: 777c44eaff2b1ee6e5939e04e580d6ad34b1ad1a
+llm-streaming.md: 93669b0781a16121765ffafb094ea14c9021e1ec
+llm-streaming.zh.md: bab98094f6a9ab3ca0aa9dc6178b46a612b205e8

docs/subsystems/llm-streaming.md

@@ -289,6 +289,15 @@ declare class BlockAssembler {
    *   its accumulated deltas (an unknown block type never closed by `block-end` throws).
    */
   blocks(): ContentBlock[];
+  /**
+   * Assemble the prefix an interrupted stream can safely finalize: closed and
+   * open text/reasoning blocks with any streamed content, in stream order.
+   * Tool calls are dropped whole — interruption precedes dispatch, so a kept
+   * call would demand a fabricated result — as are empty text/reasoning blocks
+   * and open blocks of unknown type (there is nothing assembled to keep).
+   * @returns the kept blocks; empty when nothing streamed before the interruption.
+   */
+  interruptedBlocks(): ContentBlock[];
   /** Usage from the `usage` chunk; undefined until one arrives. */
   get usage(): TokenUsage | undefined;
   /** Finish reason from the `finish` chunk; `{kind: 'stop'}` when the stream ended without one. */

docs/subsystems/llm-streaming.zh.md

@@ -297,6 +297,15 @@ declare class BlockAssembler {
    *   its accumulated deltas (an unknown block type never closed by `block-end` throws).
    */
   blocks(): ContentBlock[];
+  /**
+   * Assemble the prefix an interrupted stream can safely finalize: closed and
+   * open text/reasoning blocks with any streamed content, in stream order.
+   * Tool calls are dropped whole — interruption precedes dispatch, so a kept
+   * call would demand a fabricated result — as are empty text/reasoning blocks
+   * and open blocks of unknown type (there is nothing assembled to keep).
+   * @returns the kept blocks; empty when nothing streamed before the interruption.
+   */
+  interruptedBlocks(): ContentBlock[];
   /** Usage from the `usage` chunk; undefined until one arrives. */
   get usage(): TokenUsage | undefined;
   /** Finish reason from the `finish` chunk; `{kind: 'stop'}` when the stream ended without one. */

docs/subsystems/session.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write docs/subsystems/session.md
-session.md: f5b9e63e2320885cc41b30a09398dd341700152d
-session.zh.md: 985e0a448d1cf860ccbb0f2885d855ad6830af9f
+session.md: 521dab4e225b871d28ee870abfd3827c3fa999f3
+session.zh.md: 80dc4b4ec4dee11011480e8a28c8d4c0eae92adb

docs/subsystems/session.md

@@ -59,7 +59,10 @@ interface SessionEventMap {
    * Assembled assistant message for one step (derived history uses this).
    * Carries the step's `usage` when the adapter reported token accounting, so
    * the model output and its accounting travel together (there is no separate
-   * usage record). `usage` is absent when the adapter reported none.
+   * usage record). `usage` is absent when the adapter reported none. A turn
+   * cancelled mid-stream finalizes its delivered text/reasoning prefix as this
+   * event (undispatched tool calls are absent); an aborted turn with no such
+   * event streamed no visible content.
    */
   'assistant/message': { turn: number; step: number; message: AssistantMessage; usage?: TokenUsage }
   /**

docs/subsystems/session.zh.md

@@ -59,7 +59,10 @@ interface SessionEventMap {
    * Assembled assistant message for one step (derived history uses this).
    * Carries the step's `usage` when the adapter reported token accounting, so
    * the model output and its accounting travel together (there is no separate
-   * usage record). `usage` is absent when the adapter reported none.
+   * usage record). `usage` is absent when the adapter reported none. A turn
+   * cancelled mid-stream finalizes its delivered text/reasoning prefix as this
+   * event (undispatched tool calls are absent); an aborted turn with no such
+   * event streamed no visible content.
    */
   'assistant/message': { turn: number; step: number; message: AssistantMessage; usage?: TokenUsage }
   /**

examples/acp-agent/tests/goal-snapshots/goal-session/session.expected.jsonl

@@ -57,6 +57,7 @@
 {"type":"user/message","seq":55,"time":0,"data":{"content":[{"type":"text","text":"<goal_round>\nObjective: \"Finish the ACP goal-session snapshot proof\"\nRound: 2/2\n\nContinue working toward the objective in this same session. Treat the current workspace, tool results, and durable session state as authoritative; inspect them instead of assuming earlier narration is still current. Make concrete progress and verify the result. Before claiming completion, gather evidence that the whole objective is achieved, read the current goal, and mark it complete. If work remains, leave the goal active for the next round. Follow the configured goal-tool policy before reporting a blocker.\n</goal_round>"}],"source":{"kind":"goal","goalId":"goal-{{sessionId}}","revision":1,"round":2},"role":"user","id":"{{sessionId}}"},"surfaceOp":"append"}
 {"type":"assistant/chunk","seq":56,"time":0,"data":{"turn":3,"step":1,"chunk":{"type":"block-start","index":0,"blockType":"text"}}}
 {"type":"assistant/chunk","seq":57,"time":0,"data":{"turn":3,"step":1,"chunk":{"type":"text-delta","index":0,"text":"partial"}}}
-{"type":"step/end","seq":58,"time":0,"data":{"turn":3,"step":1}}
-{"type":"turn/end","seq":59,"time":0,"data":{"turn":3,"reason":{"kind":"aborted","reason":{"kind":"user"}}}}
-{"type":"goal/change","seq":60,"time":0,"data":{"kind":"goal/change","version":1,"operation":"pause","goal":{"id":"goal-{{sessionId}}","revision":2,"objective":"Finish the ACP goal-session snapshot proof","phase":"paused","maxGoalRounds":2},"roundsStarted":2,"createdAt":0,"updatedAt":0}}
+{"type":"assistant/message","seq":58,"time":0,"data":{"turn":3,"step":1,"message":{"role":"assistant","content":[{"type":"text","text":"partial"}],"source":{"kind":"model","provider":"deepseek-official","model":"deepseek-v4-flash"},"id":"{{sessionId}}"}},"sourceEventSeqs":[56,57],"surfaceOp":"append"}
+{"type":"step/end","seq":59,"time":0,"data":{"turn":3,"step":1}}
+{"type":"turn/end","seq":60,"time":0,"data":{"turn":3,"reason":{"kind":"aborted","reason":{"kind":"user"}}}}
+{"type":"goal/change","seq":61,"time":0,"data":{"kind":"goal/change","version":1,"operation":"pause","goal":{"id":"goal-{{sessionId}}","revision":2,"objective":"Finish the ACP goal-session snapshot proof","phase":"paused","maxGoalRounds":2},"roundsStarted":2,"createdAt":0,"updatedAt":0}}

examples/acp-agent/tests/goal-snapshots/goal-session/stdout.expected.jsonl

@@ -3,3 +3,4 @@
 {"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"GOAL READY"}}}}
 {"jsonrpc":"2.0","id":3,"result":{"stopReason":"end_turn"}}
 {"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"GOAL ROUND ONE"}}}}
+{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"partial"}}}}

examples/acp-agent/tests/snapshots/cancel/session.jsonl

@@ -10,5 +10,6 @@
 {"type":"request/context","seq":8,"time":1785730444532,"data":{"provider":"deepseek-official","model":"deepseek-v4-flash"}}
 {"type":"assistant/chunk","seq":9,"time":1785498791456,"data":{"turn":1,"step":1,"chunk":{"type":"block-start","index":0,"blockType":"text"}}}
 {"type":"assistant/chunk","seq":10,"time":1785730444541,"data":{"turn":1,"step":1,"chunk":{"type":"text-delta","index":0,"text":"partial"}}}
-{"type":"step/end","seq":11,"time":1785730444547,"data":{"turn":1,"step":1}}
-{"type":"turn/end","seq":12,"time":1785730444547,"data":{"turn":1,"reason":{"kind":"aborted","reason":{"kind":"user"}}}}
+{"type":"assistant/message","seq":11,"time":1786334791338,"data":{"turn":1,"step":1,"message":{"role":"assistant","content":[{"type":"text","text":"partial"}],"source":{"kind":"model","provider":"deepseek-official","model":"deepseek-v4-flash"},"id":"104e9294-f9b8-4248-b7df-0b7e2a069c0a"}},"sourceEventSeqs":[9,10],"surfaceOp":"append"}
+{"type":"step/end","seq":12,"time":1786334791338,"data":{"turn":1,"step":1}}
+{"type":"turn/end","seq":13,"time":1786334791338,"data":{"turn":1,"reason":{"kind":"aborted","reason":{"kind":"user"}}}}

examples/acp-agent/tests/snapshots/cancel/stdout.expected.jsonl

@@ -1,3 +1,4 @@
 {"jsonrpc":"2.0","id":1,"result":{"protocolVersion":1,"agentInfo":{"name":"deepseek-harness-acp","version":"0.0.1"},"agentCapabilities":{"promptCapabilities":{"image":false,"audio":false,"embeddedContext":false}},"authMethods":[]}}
 {"jsonrpc":"2.0","id":2,"result":{"sessionId":"{{sessionId}}"}}
 {"jsonrpc":"2.0","id":3,"result":{"stopReason":"cancelled"}}
+{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"partial"}}}}

packages/core/agent-loop/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/core/agent-loop/README.md
-README.md: 6092363fae2853d6c5d92aaf8cd01e41e18e0b52
-README.zh.md: b65b5334d735a1e0b51fa517ce41c0c953f87cf7
+README.md: fe9f58b8fa9a3851a518300b31c161b45780b80a
+README.zh.md: 62f4f63c092251e69d5788275475164fd83c8100

packages/core/agent-loop/README.md

@@ -63,7 +63,7 @@ Every inbox mutation publishes one normalized `agent/inbox/spliced` event before
 
 The driver owns one agent for its lifetime and runs inside `ctx.agents.withInitiator(agent, ...)`. Package-private orchestration entry points recover the exact Agent, derive `agent.session` once, and let operation-local helpers capture it instead of forwarding the concrete driver or per-operation `Session` through shallow interfaces. A helper keeps an explicit `Session` when that is its actual interface, while creation, persistence load, unpublished setup, services, workers, processes, persistence, and wire protocols retain their explicit identities. The [agent service](../agent/README.md#initiating-agent-scope) owns propagation, teardown, and detached-work rules.
 
-Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. The anchor records the assembled content as-is, lists the exact chunk seqs in `sourceEventSeqs` (`[]` for a stream with no chunks), and includes usage when available; empty content stays out of derived message history.
+Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. The anchor records the assembled content as-is, lists the exact chunk seqs in `sourceEventSeqs` (`[]` for a stream with no chunks), and includes usage when available; empty content stays out of derived message history. A turn cancellation that interrupts streaming also appends one anchor when visible content already streamed: the logged chunks reached the user, so the closed and open text/reasoning blocks they assemble join the surface, citing those chunk seqs — the next request contains what the user saw. Tool calls are dropped whole (interruption precedes dispatch), and nothing is appended when only empty or undispatchable content streamed; provider failures still commit nothing ([decision](../../../.agents/notes/implemented/architecture/2026-08-10-cancelled-stream-prefix-finalize.md)).
 
 After `agent/request` returns a provider/model call config, the loop asks `ctx.llm.prepareCall()` to validate adapter-owned fields and materialize configured reasoning-effort and output-token defaults under the active turn signal. The prepared call retains the exact adapter registration across this asynchronous resolution, `request/header` logging, and terminal dispatch, so HMR cannot mix one adapter's capability result with another adapter's request. The header records the effective config and which fields came from the adapter. Before the next waterfall, the loop removes those marked fields from the proposal so the current exact route rematerializes its own defaults; unmarked explicit settings persist across steps and route changes. A route with no registered adapter preserves the proposed config so an `llm/stream` listener can own and short-circuit it; unhandled terminal dispatch still fails with `NO_ADAPTER`. A new loop instance follows the same adapter-default marker rule when resuming.
 

packages/core/agent-loop/README.zh.md

@@ -63,7 +63,7 @@ interface Config {
 
 驱动器在其整个生命周期内拥有一个 agent，并在 `ctx.agents.withInitiator(agent, ...)` 内运行。包私有的编排入口点会恢复确切的 Agent，一次性派生 `agent.session`，并让操作局部的辅助函数捕获它，而不是通过浅层接口继续传递具体驱动器或每次操作的 `Session`。如果显式 `Session` 正是辅助函数的实际接口，该辅助函数会保留它；创建、持久化加载、未发布 setup、服务、worker、进程、持久化和 wire 协议则继续保留各自的显式身份。[agent 服务](../agent/README.md#initiating-agent-scope)规定传播、teardown 和分离工作规则。
 
-每次提供方调用成功结束时，都会恰好追加一个 `assistant/message` 完成锚点，包括无内容调用和以 `max-tokens` 结束的调用。该锚点原样记录组装后的内容，在 `sourceEventSeqs` 中列出确切的分片 seq（流没有分片时为 `[]`），并在用量可用时包含用量；空内容不会进入派生消息历史。
+每次提供方调用成功结束时，都会恰好追加一个 `assistant/message` 完成锚点，包括无内容调用和以 `max-tokens` 结束的调用。该锚点原样记录组装后的内容，在 `sourceEventSeqs` 中列出确切的分片 seq（流没有分片时为 `[]`），并在用量可用时包含用量；空内容不会进入派生消息历史。打断流式输出的轮次取消在已流出可见内容时同样追加一个锚点：已记录的分片已经送达用户，因此由它们组装出的已闭合与未闭合 text/reasoning 块进入 surface，并引用这些分片 seq，因此下一次请求包含用户看到的内容。工具调用整块丢弃（打断先于分派），只流出空内容或不可分派内容时不追加任何事件；提供方故障仍然不提交任何内容（[决策](../../../.agents/notes/implemented/architecture/2026-08-10-cancelled-stream-prefix-finalize.md)）。
 
 在 `agent/request` 返回提供方／模型调用配置后，循环会调用 `ctx.llm.prepareCall()`，在活跃轮次信号的控制下校验由适配器持有的字段，并填入配置的推理（reasoning）强度和输出 token 默认值。准备完成的调用会在这次异步解析、`request/header` 日志记录和最终分派期间保留同一项确切的适配器注册，因此 HMR（热模块替换）不会把某个适配器的能力解析结果与另一适配器的请求混用。请求 header 会记录生效配置以及哪些字段来自适配器。下一次 waterfall（瀑布式事件）前，循环会从提议中移除这些带标记字段，使当前精确路由重新填入自身默认值；未带标记的显式设置会跨步骤和路由变化保留。没有已注册适配器的路由会保留原定配置，使 `llm/stream` 监听器可以接管并短路该请求；最终分派仍会以 `NO_ADAPTER` 拒绝未得到处理的路由。新循环实例在恢复时会遵循同一套适配器默认值标记规则。
 

packages/core/agent-loop/src/agent.ts

@@ -51,6 +51,14 @@ type PreparedStep =
   | { kind: 'reject' }
   | { kind: 'enter'; messages: UserMessage[]; assembly: PromptAssembly }
 
+/** One live streaming attempt whose logged chunk prefix an abort can still finalize. */
+interface InterruptedAttempt {
+  readonly assembler: BlockAssembler
+  readonly chunkSeqs: number[]
+  readonly provider: string
+  readonly model: string
+}
+
 /** Remove adapter-derived values before plugins propose the next request config. */
 function requestProposal(header: EpochHeader): LlmCallConfig {
   if (header.adapterDefaults === undefined) return header.config
@@ -336,70 +344,112 @@ export class ReactLoopAgent implements Agent {
     signal.throwIfAborted()
     const system = renderPrompt(assembly)
 
-    while (true) {
-      const { request, preparedCall } = await this.buildRequest(
-        turn, step, assembly.tools, system, this.session.deriveMessages(), signal,
-      )
-      const assembler = new BlockAssembler()
-      const chunkSeqs: number[] = []
-      const stream = preparedCall?.stream(request) ?? this.loopCtx.llm.stream(request)
-      signal.throwIfAborted()
-      for await (const chunk of stream) {
+    // The streaming attempt an abort may still finalize: chunks already logged
+    // reached the user, so cancellation commits their assemblable prefix to the
+    // surface instead of dropping it (see appendInterruptedAssistant). Cleared
+    // once the attempt commits normally or a retry resets the visible stream.
+    let attempt: InterruptedAttempt | undefined
+    try {
+      while (true) {
+        const { request, preparedCall } = await this.buildRequest(
+          turn, step, assembly.tools, system, this.session.deriveMessages(), signal,
+        )
+        const assembler = new BlockAssembler()
+        const chunkSeqs: number[] = []
+        attempt = { assembler, chunkSeqs, provider: request.provider, model: request.model }
+        const stream = preparedCall?.stream(request) ?? this.loopCtx.llm.stream(request)
         signal.throwIfAborted()
-        chunkSeqs.push(this.session.append('assistant/chunk', { turn, step, chunk }).seq)
-        assembler.push(chunk)
-      }
-      signal.throwIfAborted()
-      const finish = assembler.finish
-      if (finish.kind === 'error' || finish.kind === 'aborted') {
-        const action = await this.dispatch.waterfall(
-          'agent/request-error', {
+        for await (const chunk of stream) {
+          signal.throwIfAborted()
+          chunkSeqs.push(this.session.append('assistant/chunk', { turn, step, chunk }).seq)
+          assembler.push(chunk)
+        }
+        signal.throwIfAborted()
+        const finish = assembler.finish
+        if (finish.kind === 'error' || finish.kind === 'aborted') {
+          const action = await this.dispatch.waterfall(
+            'agent/request-error', {
+              turn,
+              step,
+              provider: request.provider,
+              failure: finish.failure,
+              retryPolicy: preparedCall?.retryPolicy,
+              signal,
+            },
+            () => Promise.resolve<RequestErrorAction>(undefined),
+          )
+          signal.throwIfAborted()
+          if (action?.kind !== 'retry') {
+            throw new LlmError(finish.failure.message, finish.failure.code, finish.failure)
+          }
+          attempt = undefined
+          continue
+        }
+
+        const message = createAssistantMessage({
+          content: assembler.blocks(),
+          source: {
+            provider: request.provider,
+            model: request.model,
+            ...assembler.replayState !== undefined ? { replayState: assembler.replayState } : {},
+          },
+        })
+        attempt = undefined
+        this.session.append(
+          'assistant/message',
+          {
             turn,
             step,
-            provider: request.provider,
-            failure: finish.failure,
-            retryPolicy: preparedCall?.retryPolicy,
-            signal,
+            message,
+            ...assembler.usage === undefined ? {} : { usage: assembler.usage },
           },
-          () => Promise.resolve<RequestErrorAction>(undefined),
+          { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
         )
-        signal.throwIfAborted()
-        if (action?.kind !== 'retry') {
-          throw new LlmError(finish.failure.message, finish.failure.code, finish.failure)
-        }
-        continue
-      }
+        if (finish.kind === 'max-tokens') return { kind: 'max-tokens' }
 
-      const message = createAssistantMessage({
-        content: assembler.blocks(),
-        source: {
-          provider: request.provider,
-          model: request.model,
-          ...assembler.replayState !== undefined ? { replayState: assembler.replayState } : {},
-        },
-      })
-      this.session.append(
-        'assistant/message',
-        {
-          turn,
-          step,
-          message,
-          ...assembler.usage === undefined ? {} : { usage: assembler.usage },
-        },
-        { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
-      )
-      if (finish.kind === 'max-tokens') return { kind: 'max-tokens' }
-
-      const toolCalls = message.content.filter(block => block.type === 'tool-call')
-      if (toolCalls.length === 0) return { kind: 'completed' }
-      const { concluded } = await executeToolCalls(
-        this.loopCtx, turn, step, toolCalls, signal,
-        context => this.inbox.splice('next-step', this.inbox.nextStep.length, 0, [context]),
-      )
-      return concluded ? { kind: 'completed' } : null
+        const toolCalls = message.content.filter(block => block.type === 'tool-call')
+        if (toolCalls.length === 0) return { kind: 'completed' }
+        const { concluded } = await executeToolCalls(
+          this.loopCtx, turn, step, toolCalls, signal,
+          context => this.inbox.splice('next-step', this.inbox.nextStep.length, 0, [context]),
+        )
+        return concluded ? { kind: 'completed' } : null
+      }
+    } catch (error: unknown) {
+      if (signal.aborted && attempt !== undefined) {
+        this.appendInterruptedAssistant(turn, step, attempt)
+      }
+      throw error
     }
   }
 
+  /**
+   * Finalize a cancelled streaming attempt's user-visible prefix onto the
+   * surface: everything already logged as `assistant/chunk` events was
+   * delivered to the user, and the next request must contain what the user saw.
+   * Keeps the assembler's interrupted-safe blocks (text/reasoning; tool calls
+   * were never dispatched and are dropped); appends nothing when no visible
+   * content streamed before the interruption.
+   */
+  private appendInterruptedAssistant(turn: number, step: number, attempt: InterruptedAttempt): void {
+    const content = attempt.assembler.interruptedBlocks()
+    if (content.length === 0) return
+    const message = createAssistantMessage({
+      content,
+      source: { provider: attempt.provider, model: attempt.model },
+    })
+    this.session.append(
+      'assistant/message',
+      {
+        turn,
+        step,
+        message,
+        ...attempt.assembler.usage === undefined ? {} : { usage: attempt.assembler.usage },
+      },
+      { surfaceOp: 'append', sourceEventSeqs: attempt.chunkSeqs },
+    )
+  }
+
   /**
    * Compose one frozen request and bind it to the adapter registration that
    * resolved its exact-model defaults.

packages/core/agent-loop/tests/cancel.spec.ts

@@ -1,4 +1,4 @@
-import { createUserMessage } from '@deepseek-ai/dsh-llm'
+import { CallId, createUserMessage } from '@deepseek-ai/dsh-llm'
 /**
  * Tests for the queue-aware `Agent.cancel()` primitive. The default clears
  * queued and steering work, while `keepInbox` preserves pending input for a
@@ -479,6 +479,104 @@ describe('Agent.cancel()', () => {
     expect(reasons.length).toBe(2)
   })
 
+  it('cancel mid-stream finalizes the streamed prefix onto the surface', async () => {
+    const adapter = new MockAdapter(['hang', textResponse('after')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('partial-finalize'), { provider: 'mock', model: 'mock' })
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    agent.cancel({ kind: 'user' })
+    await waitForIdle(ctx, agent)
+
+    // The prefix the user watched stream is committed as the step's message,
+    // citing exactly the chunk events that delivered it.
+    const message = agent.session.events.find(e => e.type === 'assistant/message')
+    expect(message?.type === 'assistant/message' ? message.data.message.content : undefined)
+      .toEqual([{ type: 'text', text: 'partial' }])
+    const chunkSeqs = agent.session.events.filter(e => e.type === 'assistant/chunk').map(e => e.seq)
+    expect(message?.sourceEventSeqs).toEqual(chunkSeqs)
+    const types = agent.session.events.map(e => e.type)
+    expect(types.indexOf('assistant/message')).toBeLessThan(types.indexOf('step/end'))
+    expect(types.indexOf('step/end')).toBeLessThan(types.indexOf('turn/end'))
+
+    // The next request derives the finalized prefix: the model sees what the user saw.
+    send(agent, 'continue')
+    await waitForIdle(ctx, agent)
+    const replayed = adapter.requests[1]!.messages
+      .filter(m => m.role === 'assistant')
+      .flatMap(m => m.content)
+      .flatMap(b => b.type === 'text' ? [b.text] : [])
+    expect(replayed).toContain('partial')
+  })
+
+  it('cancel during reasoning-only streaming finalizes the reasoning prefix', async () => {
+    const adapter = new MockAdapter([{
+      hangAfter: [
+        { type: 'block-start', index: 0, blockType: 'reasoning' },
+        { type: 'reasoning-delta', index: 0, text: 'thinking about it' },
+        { type: 'usage', usage: { inputTokens: 7, outputTokens: 4 } },
+      ],
+    }])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('reasoning-finalize'), { provider: 'mock', model: 'mock' })
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    agent.cancel({ kind: 'user' })
+    await waitForIdle(ctx, agent)
+
+    const message = agent.session.events.find(e => e.type === 'assistant/message')
+    expect(message?.type === 'assistant/message' ? message.data.message.content : undefined)
+      .toEqual([{ type: 'reasoning', text: 'thinking about it' }])
+    // A usage chunk delivered before the cancel travels with the finalized prefix.
+    expect(message?.type === 'assistant/message' ? message.data.usage : undefined)
+      .toEqual({ inputTokens: 7, outputTokens: 4 })
+  })
+
+  it('cancel drops a half-streamed tool call and keeps the completed text before it', async () => {
+    const adapter = new MockAdapter([{
+      hangAfter: [
+        { type: 'block-start', index: 0, blockType: 'text' },
+        { type: 'text-delta', index: 0, text: 'reading the file' },
+        { type: 'block-end', index: 0, block: { type: 'text', text: 'reading the file' } },
+        { type: 'block-start', index: 1, blockType: 'tool-call' },
+        { type: 'tool-call-delta', index: 1, id: CallId('c1'), name: 'read', argumentsDelta: '{"pa' },
+      ],
+    }])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('tool-call-drop'), { provider: 'mock', model: 'mock' })
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    agent.cancel({ kind: 'user' })
+    await waitForIdle(ctx, agent)
+
+    // The undispatched call is dropped whole — no dangling tool_use to pair.
+    const message = agent.session.events.find(e => e.type === 'assistant/message')
+    expect(message?.type === 'assistant/message' ? message.data.message.content : undefined)
+      .toEqual([{ type: 'text', text: 'reading the file' }])
+    expect(agent.session.events.some(e => e.type === 'tool/call')).toBe(false)
+  })
+
+  it('cancel before any visible content finalizes nothing', async () => {
+    const adapter = new MockAdapter([{
+      hangAfter: [
+        { type: 'block-start', index: 0, blockType: 'tool-call' },
+        { type: 'tool-call-delta', index: 0, id: CallId('c1'), name: 'read', argumentsDelta: '{"pa' },
+      ],
+    }])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('nothing-to-finalize'), { provider: 'mock', model: 'mock' })
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    agent.cancel({ kind: 'user' })
+    await waitForIdle(ctx, agent)
+
+    expect(agent.session.events.some(e => e.type === 'assistant/message')).toBe(false)
+  })
+
   it('cancel from a synchronous step/start session-event listener drops the step (post-step-start window)', async () => {
     const adapter = new MockAdapter([textResponse('should not stream')])
     const ctx = await harness(adapter)

packages/core/agent-loop/tests/mock-adapter.ts

@@ -55,19 +55,25 @@ export function toolCallResponse(rawCallId: string, name: string, args: object,
   return chunks
 }
 
+/** Script entry that streams the given chunks, then hangs until aborted. */
+export interface HangAfter {
+  hangAfter: StreamChunk[]
+}
+
 /**
  * Mock adapter driven by a script: each model call consumes the next entry.
  * Records every request it receives for assertions. An entry may be a
  * function to compute chunks from the request, a 'hang' marker that
- * streams one chunk then waits until aborted, or 'hang-slow' which takes
+ * streams one chunk then waits until aborted, 'hang-slow' which takes
  * 50ms to notice the abort — a stand-in for slow real-world teardown
- * (LLM stream cancellation, tool unwinding).
+ * (LLM stream cancellation, tool unwinding) — or a {@link HangAfter}
+ * scripting the exact chunks delivered before the hang.
  */
 export class MockAdapter extends LlmAdapter {
   requests: GenerateOptions[] = []
 
   constructor(
-    private script: (StreamChunk[] | ((options: GenerateOptions) => StreamChunk[]) | 'hang' | 'hang-slow')[],
+    private script: (StreamChunk[] | ((options: GenerateOptions) => StreamChunk[]) | 'hang' | 'hang-slow' | HangAfter)[],
     private readonly reasoning?: LlmModelReasoningInfo,
     private readonly defaultMaxTokens?: number,
   ) {
@@ -100,6 +106,14 @@ export class MockAdapter extends LlmAdapter {
       })
       return
     }
+    if (typeof entry === 'object' && !Array.isArray(entry) && 'hangAfter' in entry) {
+      for (const chunk of entry.hangAfter) yield chunk
+      await new Promise<void>((_resolve, reject) => {
+        if (options.signal?.aborted) { reject(new Error('aborted')); return }
+        options.signal?.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
+      })
+      return
+    }
     if (entry === 'hang-slow') {
       yield { type: 'block-start', index: 0, blockType: 'text' }
       yield { type: 'text-delta', index: 0, text: 'partial' }

packages/core/session/src/types.ts

@@ -240,7 +240,10 @@ export interface SessionEventMap {
    * Assembled assistant message for one step (derived history uses this).
    * Carries the step's `usage` when the adapter reported token accounting, so
    * the model output and its accounting travel together (there is no separate
-   * usage record). `usage` is absent when the adapter reported none.
+   * usage record). `usage` is absent when the adapter reported none. A turn
+   * cancelled mid-stream finalizes its delivered text/reasoning prefix as this
+   * event (undispatched tool calls are absent); an aborted turn with no such
+   * event streamed no visible content.
    */
   'assistant/message': { turn: number; step: number; message: AssistantMessage; usage?: TokenUsage }
   /**

packages/llm/llm/src/assembler.ts

@@ -138,6 +138,26 @@ export class BlockAssembler {
       : blocks
   }
 
+  /**
+   * Assemble the prefix an interrupted stream can safely finalize: closed and
+   * open text/reasoning blocks with any streamed content, in stream order.
+   * Tool calls are dropped whole — interruption precedes dispatch, so a kept
+   * call would demand a fabricated result — as are empty text/reasoning blocks
+   * and open blocks of unknown type (there is nothing assembled to keep).
+   * @returns the kept blocks; empty when nothing streamed before the interruption.
+   */
+  interruptedBlocks(): ContentBlock[] {
+    return this.order
+      .map((index) => {
+        const partial = this.mustGet(index)
+        const type = partial.block?.type ?? partial.blockType
+        if (type !== 'text' && type !== 'reasoning') return undefined
+        return this.assemble(partial, index)
+      })
+      .filter((block): block is ContentBlock =>
+        (block?.type === 'text' || block?.type === 'reasoning') && block.text.trim() !== '')
+  }
+
   /** Usage from the `usage` chunk; undefined until one arrives. */
   get usage(): TokenUsage | undefined {
     return this._usage

packages/llm/llm/tests/assembler.spec.ts

@@ -145,3 +145,41 @@ describe('BlockAssembler duplicate-close contract', () => {
     expect(assembler.blocks()).toEqual([{ type: 'reasoning', text: 'first' }])
   })
 })
+
+describe('BlockAssembler.interruptedBlocks', () => {
+  it('keeps closed and open text/reasoning blocks with streamed content, in order', () => {
+    const assembler = new BlockAssembler()
+    assembler.push({ type: 'block-start', index: 0, blockType: 'reasoning' })
+    assembler.push({ type: 'reasoning-delta', index: 0, text: 'planning' })
+    assembler.push({ type: 'block-end', index: 0, block: { type: 'reasoning', text: 'planning' } })
+    assembler.push({ type: 'block-start', index: 1, blockType: 'text' })
+    assembler.push({ type: 'text-delta', index: 1, text: 'half an ans' })
+    expect(assembler.interruptedBlocks()).toEqual([
+      { type: 'reasoning', text: 'planning' },
+      { type: 'text', text: 'half an ans' },
+    ])
+  })
+
+  it('drops tool calls whether open or closed — interruption precedes dispatch', () => {
+    const assembler = new BlockAssembler()
+    assembler.push({ type: 'block-start', index: 0, blockType: 'text' })
+    assembler.push({ type: 'text-delta', index: 0, text: 'calling' })
+    assembler.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'calling' } })
+    assembler.push({ type: 'block-start', index: 1, blockType: 'tool-call' })
+    assembler.push({ type: 'tool-call-delta', index: 1, id: CallId('c1'), name: 'read', argumentsDelta: '{"a":1}' })
+    assembler.push({ type: 'block-end', index: 1, block: { type: 'tool-call', id: CallId('c1'), name: 'read', arguments: '{"a":1}' } })
+    assembler.push({ type: 'block-start', index: 2, blockType: 'tool-call' })
+    assembler.push({ type: 'tool-call-delta', index: 2, id: CallId('c2'), name: 'read', argumentsDelta: '{"pa' })
+    expect(assembler.interruptedBlocks()).toEqual([{ type: 'text', text: 'calling' }])
+  })
+
+  it('drops empty and whitespace-only text/reasoning blocks and unknown open block types', () => {
+    const assembler = new BlockAssembler()
+    assembler.push({ type: 'block-start', index: 0, blockType: 'text' })
+    assembler.push({ type: 'text-delta', index: 0, text: '  \n' })
+    // A merge-extended block kind this build does not know how to assemble.
+    assembler.push({ type: 'block-start', index: 1, blockType: 'mystery' } as unknown as StreamChunk)
+    assembler.push({ type: 'block-start', index: 2, blockType: 'reasoning' })
+    expect(assembler.interruptedBlocks()).toEqual([])
+  })
+})