fix(agent): terminate live assistant attempts

.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write .agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.md
-2026-08-31-live-assistant-stream-frames.md: 9e848cadbfa71f8281caf14207b36d05f1f95d99
-2026-08-31-live-assistant-stream-frames.zh.md: dc672d6d8af31f1b45694299125d3f91cb1874ee
+2026-08-31-live-assistant-stream-frames.md: c01a919b1c331f82e9c25cc01625f412414591fd
+2026-08-31-live-assistant-stream-frames.zh.md: 0c1a21a09dff2c3d092f707a8970617002e25282

.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.md

@@ -10,7 +10,7 @@ The session log keeps every `assistant/chunk` so replay, cold reads, telemetry,
 
 ## Decision
 
-`dsh-agent-loop` emits scoped `agent/assistant-stream` frames for each model attempt. `start`, `chunk`, and `end` carry a branded process-local `LlmAttemptId`; every emitted frame advances one Session-local revision. The `start` frame captures a safe-integer wall-clock `startedTime`, chunk indexes are dense from zero, and `end.index` equals the next chunk position. Stream acquisition and its final cancellation check occur before `start`; a failure there emits no frame, while every started attempt emits a terminal `end`. The loop appends every v1 `assistant/chunk` before its matching live chunk frame, records that exact `legacyChunkSeq`, and appends the final `assistant/message` before a committed end frame. The existing authenticated Session-follow accepts an explicit Web opt-in, opens with a cached active-attempt baseline, and carries durable events and cursorless frames in one FIFO. Each follower captures a local arrival ordinal with the opening baseline and drops buffered frames at or before that cut; frame revisions can restart at one with a replacement Agent, so they do not define the opening cut. If the durable opening snapshot precedes the Assistant baseline, a baseline may already acknowledge a chunk whose durable event remains buffered; the Web Session publishes that event when it arrives because the baseline's exact `legacyChunkSeq` proves its matching frame. A final message arriving after an active opening remains staged until the matching `end.index` and ordered provenance arrive; an earlier retry at the same Turn and Step remains visible. Revision, dense-index, or provenance gaps re-open follow and replace the baseline. The TypeScript and Python SDK protocols do not expose these frames. The durable log remains the source of replay and model history.
+`dsh-agent-loop` emits scoped `agent/assistant-stream` frames for each model attempt. `start`, `chunk`, and `end` carry a branded `LlmAttemptId` unique within one Agent lifecycle; every emitted frame advances one revision local to that lifecycle. The `start` frame captures a safe-integer wall-clock `startedTime`, chunk indexes are dense from zero, and `end.index` equals the next chunk position. Stream acquisition and its final cancellation check occur before `start`; a failure there emits no frame, while every started attempt emits a terminal `end`. The loop appends every v1 `assistant/chunk` before its matching live chunk frame, records that exact `legacyChunkSeq`, and appends the final `assistant/message` before a committed end frame. The existing authenticated Session-follow accepts an explicit Web opt-in, opens with a cached active-attempt baseline, and carries durable events and cursorless frames in one FIFO. Each follower captures a local arrival ordinal with the opening baseline and drops buffered frames at or before that cut; frame revisions can restart at one with a replacement Agent, so they do not define the opening cut. If the durable opening snapshot precedes the Assistant baseline, a baseline may already acknowledge a chunk whose durable event remains buffered; the Web Session publishes that event when it arrives because the baseline's exact `legacyChunkSeq` proves its matching frame. A final message arriving after an active opening remains staged until the matching `end.index` and ordered provenance arrive; an earlier retry at the same Turn and Step remains visible. Revision, dense-index, or provenance gaps for a known attempt re-open follow and replace the baseline; an unknown attempt uses the durable fallback. The TypeScript and Python SDK protocols do not expose these frames. The durable log remains the source of replay and model history.
 
 ## Alternatives considered
 
@@ -21,4 +21,4 @@ The session log keeps every `assistant/chunk` so replay, cold reads, telemetry,
 
 ## Consequences
 
-The Web client renders in-memory chunks before persistence flush while retaining one durable v1 history, without changing `SESSION_FORMAT_VERSION` or the chunk-row encoding. A process restart has no active assistant frames; reconnect and cold replay use durable records. Cursorless notifications never advance the journal cursor, and notifications observed during durable gap repair wait for the replacement page. That page has no Assistant baseline, so the Client clears transient attempts and lets the held notification reopen follow once for an atomically paired page and baseline. The frame declaration remains agent-scoped, so a listener observes only its owning Agent unless it explicitly registers globally.
+Assistant frames gate and settle publication of durable events without changing `SESSION_FORMAT_VERSION`, chunk-row encoding, or the user-visible rendering source. A process restart has no active assistant frames; reconnect and cold replay use durable records. A Client that joins after an attempt start cannot reconstruct that transient attempt, so unknown-attempt frames fall back to ordinary durable publication until the next known start. Cursorless notifications never advance the journal cursor, and notifications observed during durable gap repair wait for the replacement page. That page has no Assistant baseline, so the Client clears transient attempts and lets the held notification reopen follow once for an atomically paired page and baseline. The frame declaration remains agent-scoped, so a listener observes only its owning Agent unless it explicitly registers globally.

.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.zh.md

@@ -10,7 +10,7 @@ Session log 保留每个 `assistant/chunk`，因此重放、冷读、遥测和
 
 ## 决定
 
-`dsh-agent-loop` 为每次模型尝试发出作用域内的 `agent/assistant-stream` 帧。`start`、`chunk` 和 `end` 带有带品牌的进程本地 `LlmAttemptId`；每个已发出的帧都会推进一次 Session 本地 revision。`start` 帧会把壁钟时间捕获为安全整数 `startedTime`，chunk index 从零开始连续递增，`end.index` 等于下一个 chunk 位置。循环会先取得 stream 并执行最终取消检查，再发出 `start`；这些步骤失败时不发出任何帧，而每个已开始的尝试都会发出终态 `end`。循环在匹配的实时 chunk 帧之前追加每个 v1 `assistant/chunk`，记录精确的 `legacyChunkSeq`，并在已提交的 end 帧之前追加最终 `assistant/message`。现有的已认证 Session-follow 接受显式 Web opt-in，以缓存的活跃尝试 baseline 打开，并在一个 FIFO 中携带持久事件和无 cursor 的帧。每个 follower 会随 opening baseline 捕获本地到达序号，并丢弃该 cut 及之前的 buffered frame；replacement Agent 的 frame revision 可以从一重新开始，因此 revision 不定义 opening cut。如果持久 opening snapshot 早于 Assistant baseline，baseline 可能已经确认一个持久事件仍在 buffer 中的 chunk；该事件到达时，Web Session 会依据 baseline 中精确的 `legacyChunkSeq` 已证明其匹配帧而直接发布。活跃 opening 之后到达的最终 message 会保持暂存，直到匹配的 `end.index` 与有序来源到达；同一 Turn 和 Step 中更早的 retry 仍保持可见。revision、连续 index 或来源缺口会重新打开 follow 并替换 baseline。TypeScript 和 Python SDK 协议不公开这些帧。持久 log 仍然是重放和模型历史的真源。
+`dsh-agent-loop` 为每次模型尝试发出作用域内的 `agent/assistant-stream` 帧。`start`、`chunk` 和 `end` 带有在单个 Agent lifecycle 内唯一的 branded `LlmAttemptId`；每个已发出的帧都会推进一次该 lifecycle 本地 revision。`start` 帧会把壁钟时间捕获为安全整数 `startedTime`，chunk index 从零开始连续递增，`end.index` 等于下一个 chunk 位置。循环会先取得 stream 并执行最终取消检查，再发出 `start`；这些步骤失败时不发出任何帧，而每个已开始的尝试都会发出终态 `end`。循环在匹配的实时 chunk 帧之前追加每个 v1 `assistant/chunk`，记录精确的 `legacyChunkSeq`，并在已提交的 end 帧之前追加最终 `assistant/message`。现有的已认证 Session-follow 接受显式 Web opt-in，以缓存的活跃尝试 baseline 打开，并在一个 FIFO 中携带持久事件和无 cursor 的帧。每个 follower 会随 opening baseline 捕获本地到达序号，并丢弃该 cut 及之前的 buffered frame；replacement Agent 的 frame revision 可以从一重新开始，因此 revision 不定义 opening cut。如果持久 opening snapshot 早于 Assistant baseline，baseline 可能已经确认一个持久事件仍在 buffer 中的 chunk；该事件到达时，Web Session 会依据 baseline 中精确的 `legacyChunkSeq` 已证明其匹配帧而直接发布。活跃 opening 之后到达的最终 message 会保持暂存，直到匹配的 `end.index` 与有序来源到达；同一 Turn 和 Step 中更早的 retry 仍保持可见。已知 attempt 的 revision、连续 index 或来源缺口会重新打开 follow 并替换 baseline；unknown attempt 使用持久回退。TypeScript 和 Python SDK 协议不公开这些帧。持久 log 仍然是重放和模型历史的真源。
 
 ## 曾考虑的替代方案
 
@@ -21,4 +21,4 @@ Session log 保留每个 `assistant/chunk`，因此重放、冷读、遥测和
 
 ## 影响
 
-Web client 可以在 persistence flush 前渲染内存 chunk，同时保留一份持久 v1 历史，而不改变 `SESSION_FORMAT_VERSION` 或 chunk-row 编码。进程重启后没有活跃 assistant 帧；重连和冷重放使用持久记录。无 cursor 的通知绝不推进 journal cursor，在持久缺口修复期间观察到的通知会等待 replacement page。该 page 不携带 Assistant baseline，因此 Client 会清空瞬态尝试，并让 held notification 重新打开 follow 一次，以取得原子配对的 page 与 baseline。帧声明保持 agent 作用域，因此监听器只观察所属 Agent，除非它显式全局注册。
+Assistant 帧会门控并结算持久事件的发布，而不改变 `SESSION_FORMAT_VERSION`、chunk-row 编码或用户可见的渲染来源。进程重启后没有活跃 assistant 帧；重连和冷重放使用持久记录。在 attempt start 之后才加入的 Client 无法重建该瞬态 attempt，因此 unknown-attempt 帧会回退到普通持久发布，直到下一个已知 start。无 cursor 的通知绝不推进 journal cursor，在持久缺口修复期间观察到的通知会等待 replacement page。该 page 不携带 Assistant baseline，因此 Client 会清空瞬态尝试，并让 held notification 重新打开 follow 一次，以取得原子配对的 page 与 baseline。帧声明保持 agent 作用域，因此监听器只观察所属 Agent，除非它显式全局注册。

docs/subsystems/core.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/core.md
-core.md: 9abda9c2fbe036673d6cce0d138dc8741fc4a8c8
-core.zh.md: c06d712fa46b1bb70557ad01635bc5470b96a075
+core.md: 7965e90bfb8641c8403663041694501b5575e64e
+core.zh.md: b2a68328fcdbf19cf1c031d8c56b716f73ab12c5

docs/subsystems/core.md

@@ -152,6 +152,42 @@ interface Agent {
 type AgentStatus = 'idle' | 'running'
 ```
 
+```ts type-equiv
+/** One process-local live assistant streaming publication. */
+type AssistantStreamFrame =
+  | {
+    readonly type: 'start'
+    readonly attemptId: LlmAttemptId
+    /** Monotone within one attached Agent lifecycle; replacement restarts at 1. */
+    readonly revision: number
+    /** Safe-integer wall-clock time captured when this attempt started. */
+    readonly startedTime: number
+    readonly turn: number
+    readonly step: number
+  }
+  | {
+    readonly type: 'chunk'
+    readonly attemptId: LlmAttemptId
+    readonly revision: number
+    /** Dense zero-based position within the attempt. */
+    readonly index: number
+    readonly chunk: StreamChunk
+    /** Matching durable v1 `assistant/chunk` record for duplicate suppression. */
+    readonly legacyChunkSeq: SessionSeq
+  }
+  | {
+    readonly type: 'end'
+    readonly attemptId: LlmAttemptId
+    readonly revision: number
+    /** Number of chunk frames emitted by this attempt. */
+    readonly index: number
+    /** `committed` follows its durable message; `aborted` has no such requirement. */
+    readonly outcome: 'committed' | 'aborted'
+    /** Every durable v1 chunk represented by this attempt. */
+    readonly legacyChunkSeqs: readonly SessionSeq[]
+  }
+```
+
 `running` describes the driver-wide drain interval and may span consecutive queued turns; it does not prove a turn is still open. Disposal removes the agent from the registry and emits `agent/disposed`; it is not a terminal status value. `followup()` returns no handle: its `MessageId` identifies durable inbox insertion, claim, and discard facts, not a later assistant output or turn ending. `whenIdle()` observes the whole agent, so callers may call a receipt-to-idle interval a run only when they explicitly own that interval ([decision](../../.agents/notes/implemented/architecture/2026-07-30-followup-enqueue-and-owned-runs.md)).
 
 ```ts type-equiv

docs/subsystems/core.zh.md

@@ -156,6 +156,42 @@ interface Agent {
 type AgentStatus = 'idle' | 'running'
 ```
 
+```ts type-equiv
+/** One process-local live assistant streaming publication. */
+type AssistantStreamFrame =
+  | {
+    readonly type: 'start'
+    readonly attemptId: LlmAttemptId
+    /** Monotone within one attached Agent lifecycle; replacement restarts at 1. */
+    readonly revision: number
+    /** Safe-integer wall-clock time captured when this attempt started. */
+    readonly startedTime: number
+    readonly turn: number
+    readonly step: number
+  }
+  | {
+    readonly type: 'chunk'
+    readonly attemptId: LlmAttemptId
+    readonly revision: number
+    /** Dense zero-based position within the attempt. */
+    readonly index: number
+    readonly chunk: StreamChunk
+    /** Matching durable v1 `assistant/chunk` record for duplicate suppression. */
+    readonly legacyChunkSeq: SessionSeq
+  }
+  | {
+    readonly type: 'end'
+    readonly attemptId: LlmAttemptId
+    readonly revision: number
+    /** Number of chunk frames emitted by this attempt. */
+    readonly index: number
+    /** `committed` follows its durable message; `aborted` has no such requirement. */
+    readonly outcome: 'committed' | 'aborted'
+    /** Every durable v1 chunk represented by this attempt. */
+    readonly legacyChunkSeqs: readonly SessionSeq[]
+  }
+```
+
 `running` 描述整个驱动器的排空区间，可能跨越连续的排队轮次；它不能证明某个轮次仍然打开。dispose 会把 agent 从注册表移除并发出 `agent/disposed`；它不是一个终态 status 值。`followup()` 不返回句柄：其 `MessageId` 标识的是持久的 inbox 插入、认领与丢弃事实，而非之后的助手输出或轮次结束。`whenIdle()` 观察的是整个 agent，因此只有当调用方明确拥有从回执到空闲的这段区间时，才能把它称为一次 run（[决策](../../.agents/notes/implemented/architecture/2026-07-30-followup-enqueue-and-owned-runs.zh.md)）。
 
 ```ts type-equiv

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: a9fcf39e7bcbee594fa575738f4530fa68385a45
-session.zh.md: 333007e4f9efe5977d0e1aaa05c76d4bf946fb24
+session.md: 4bcf654280b9e7d08a7dc44399a27c7f46815a30
+session.zh.md: c7f8bef4eaef2838f8701589353ee44c35fa6895

docs/subsystems/session.md

@@ -791,7 +791,8 @@ inspect( sessionId: SessionId, signal?: AbortSignal, ): Promise<SessionInspectio
  * Follow one Session log from its opening or resume cursor.
  * @param request - durable address and last committed sequence already held by the caller.
  * @param signal - cancellation owned by the Remote stream carrier.
- * @returns a complete opening snapshot followed by gap-free event frames.
+ * @returns a complete opening snapshot followed by gap-free durable event
+ *   frames and optional cursorless assistant-stream frames.
  */
 @Remote({ mode: 'stream' }) follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame>
 

docs/subsystems/session.zh.md

@@ -795,7 +795,8 @@ inspect( sessionId: SessionId, signal?: AbortSignal, ): Promise<SessionInspectio
  * Follow one Session log from its opening or resume cursor.
  * @param request - durable address and last committed sequence already held by the caller.
  * @param signal - cancellation owned by the Remote stream carrier.
- * @returns a complete opening snapshot followed by gap-free event frames.
+ * @returns a complete opening snapshot followed by gap-free durable event
+ *   frames and optional cursorless assistant-stream frames.
  */
 @Remote({ mode: 'stream' }) follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame>
 

packages/api/gateway/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/api/gateway/README.md
-README.md: c9f0cc726373af450da434a774279ac22e4889f6
-README.zh.md: 47699071e6e5d46039271388677e77bccaae5975
+README.md: 019aedfa1b900fd1e52744e91789b22c59d994e9
+README.zh.md: 5cf91d3f6c4f86333201209aabac9b0f9a58d703

packages/api/gateway/README.md

@@ -47,7 +47,7 @@ Every unary call resolves to `RemoteResult<T>` — `{ ok: true, value }` or `{ o
 
 `ctx.remote.$host` reads the fixed Host facts as plain values: `home` (undefined until the first ready frame) and `isLoopback`. It is not a store — no subscription, no generation counter — so a consumer that must react to reconnection listens for `connection/reset` instead of polling it.
 
-`ctx.remote.$stream()` returns a single-consumer `RemoteStream` spanning physical carrier generations. It permits one immediate retry while the Host remains available, otherwise waits for the next connected Host generation, and annotates each item with its physical generation. The domain consumer validates and accepts each generation's opening value; business and protocol failures remain terminal. Every terminal failure leaves this face as a `RemoteError`, including exhausted carrier retries and a generation that ends before its opening value, so a stream consumer discriminates the same way a unary caller does. `RemoteStreamCarrierError` names a retryable physical loss and reaches a domain only as the `carrierFailed` callback argument, never as a terminal outcome. `RemoteSnapshotStream` adds one opening snapshot followed by deltas. `RemoteJournalStream` adds follow-before-page opening, pagination, reconnect catch-up, and gap repair over domain-defined inclusive entry ranges; it removes complete duplicates and rejects gaps, inverted ranges, and partial overlaps. A domain may also carry cursorless notifications: they never advance or repair the durable cursor, and notifications received during gap repair publish only after the replacement page commits. Disposing any stream cancels its requests and resolves after the active iterator is fully stopped.
+`ctx.remote.$stream()` returns a single-consumer `RemoteStream` spanning physical carrier generations. It permits one immediate retry while the Host remains available, otherwise waits for the next connected Host generation, and annotates each item with its physical generation. The domain consumer validates and accepts each generation's opening value; business and protocol failures remain terminal. Every terminal failure leaves this face as a `RemoteError`, including exhausted carrier retries and a generation that ends before its opening value, so a stream consumer discriminates the same way a unary caller does. `RemoteStreamCarrierError` names a retryable physical loss and reaches a domain only as the `carrierFailed` callback argument, never as a terminal outcome. `RemoteSnapshotStream` adds one opening snapshot followed by deltas. `RemoteJournalStream` adds follow-before-page opening, pagination, reconnect catch-up, and gap repair over domain-defined inclusive entry ranges; it removes complete duplicates and rejects gaps, inverted ranges, and partial overlaps. A domain may also carry cursorless notifications: they never advance or repair the durable cursor, and notifications received during gap repair publish only after the replacement page commits. If a newer generation supersedes that repair, held notifications from the superseded generation are discarded with its page. Disposing any stream cancels its requests and resolves after the active iterator is fully stopped.
 
 `ctx.remote.$on()` subscribes to one forwarded Host event. Its legal keys are exactly the Host assembly's forwarding selection, and the listener type is the owning package's own Cordis `Events` declaration, so no second signature can drift from it. Each subscription belongs to the calling fiber and disappears with it. The Client Remote service registers the `$events` pump as a Connection generation source when it activates, whether any `$on` listener exists. Browsers use Remote mux, while in-process compositions use `connection.rpc.open`; the opening `ready` item establishes a Connection generation and supplies its Host facts. Carrier failure, Remote stream failure, unexpected normal completion, a non-ready opening item, or a malformed event item ends that generation and lets Connection reopen it under bounded jittered exponential backoff. Ordinary notifications run in registration order and isolate listener failures. Agent-scoped waterfalls let a listener return a result, call `next()`, or reject; Gateway returns that outcome through the existing HTTP unary carrier.
 

packages/api/gateway/README.zh.md

@@ -47,7 +47,7 @@ Host 组合可通过 `registerRemoteEvents()` 注册唯一的应用事件 source
 
 `ctx.remote.$host` 以普通值读取固定的 Host 事实：`home`（首个 ready 帧之前为 undefined）与 `isLoopback`。它不是 store——没有订阅、没有代次计数——所以需要响应重连的消费方去监听 `connection/reset`，而不是轮询它。
 
-`ctx.remote.$stream()` 返回跨越多个物理载体代次的单消费方 `RemoteStream`。Host 仍在线时，它允许一次立即重试；Host 离线时，它等待下一代连接，并为每个流项标注物理代次。领域消费方校验并接受各代次的 opening value；业务与协议错误仍然终止流。一切终态失败离开本面时都是 `RemoteError`，包括重试耗尽和在 opening value 之前就结束的代次，因此流消费方与一元调用方用同一种方式判别。`RemoteStreamCarrierError` 命名的是可重试的物理丢失，它只作为 `carrierFailed` 回调参数到达领域，绝不作为终态结果。`RemoteSnapshotStream` 在此之上规定每代由一个 opening snapshot 和后续 delta 组成。`RemoteJournalStream` 基于领域提供的 entry 闭区间提供 follow-before-page、分页、重连追赶与缺口修复；它丢弃完整重复项，并拒绝缺口、倒置区间和部分重叠。领域还可以携带无 cursor 的通知：通知绝不推进或修复持久 cursor，在缺口修复期间收到的通知只会在 replacement page 提交后发布。dispose 任一种 stream 都会取消其请求，并在活动 iterator 完全停止后完成。
+`ctx.remote.$stream()` 返回跨越多个物理载体代次的单消费方 `RemoteStream`。Host 仍在线时，它允许一次立即重试；Host 离线时，它等待下一代连接，并为每个流项标注物理代次。领域消费方校验并接受各代次的 opening value；业务与协议错误仍然终止流。一切终态失败离开本面时都是 `RemoteError`，包括重试耗尽和在 opening value 之前就结束的代次，因此流消费方与一元调用方用同一种方式判别。`RemoteStreamCarrierError` 命名的是可重试的物理丢失，它只作为 `carrierFailed` 回调参数到达领域，绝不作为终态结果。`RemoteSnapshotStream` 在此之上规定每代由一个 opening snapshot 和后续 delta 组成。`RemoteJournalStream` 基于领域提供的 entry 闭区间提供 follow-before-page、分页、重连追赶与缺口修复；它丢弃完整重复项，并拒绝缺口、倒置区间和部分重叠。领域还可以携带无 cursor 的通知：通知绝不推进或修复持久 cursor，在缺口修复期间收到的通知只会在 replacement page 提交后发布。若更新代次取代该修复，旧代次 held notification 会与其 page 一同丢弃。dispose 任一种 stream 都会取消其请求，并在活动 iterator 完全停止后完成。
 
 `ctx.remote.$on()` 订阅一条被转发的 Host 事件。它的合法键恰好等于 Host 装配声明的转发选择，listener 类型就是事件所属包自己的 Cordis `Events` 声明，因此不存在会与之漂移的第二份签名。每个订阅归属调用方 fiber，并随该 fiber 一起消失。Client Remote 服务激活时就把 `$events` pump 注册为 Connection generation source，因此即使当前无 `$on` 订阅，它也会在 Connection 循环启动时打开。浏览器使用 Remote mux，进程内组合使用 `connection.rpc.open`；opening `ready` 项建立 Connection generation 并提供 Host 信息。物理 carrier 失败、Remote stream error、意外正常结束、非 ready 首项或畸形事件项都会终止该 generation，由 Connection 按有界且带抖动的指数退避重开。普通通知按注册顺序运行并隔离 listener 失败；Agent-scoped waterfall 允许 listener 返回结果、调用 `next()` 或拒绝，Gateway 再通过现有 HTTP 一元载体回送该结果。
 

packages/api/session-controller/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/api/session-controller/README.md
-README.md: 304cc2ed1c4e460667d1a9badff50687f61f8911
-README.zh.md: f4ea36354d0d956771a6f3d080ed22f0cc59780b
+README.md: 627b614bc89514cc26e8588e342a5ef076ab571b
+README.zh.md: ace5318a40b6f8673c5d0c671a7f6ad3fd5ac98d

packages/api/session-controller/src/client/sessions/assistant-stream.ts

@@ -17,6 +17,11 @@ interface ActiveAttempt {
   nextIndex: number
 }
 
+interface PendingAssistantMessage {
+  readonly entry: SessionLiveEventEntry
+  readonly sourceEventSeqs: readonly number[] | undefined
+}
+
 /** One Web publication decision from the assistant stream fold. */
 export type ClientAssistantStreamResult =
   | { readonly type: 'publish'; readonly entry: SessionLiveEventEntry }
@@ -38,7 +43,7 @@ function sameSeqs(left: readonly number[], right: readonly number[]): boolean {
 export class ClientAssistantStream {
   private readonly attempts = new Map<string, ActiveAttempt>()
   private readonly pendingChunks = new Map<number, SessionLiveEventEntry>()
-  private readonly pendingMessages = new Map<string, SessionLiveEventEntry>()
+  private readonly pendingMessages = new Map<string, PendingAssistantMessage>()
   private publishedSeqs = new Set<number>()
 
   /**
@@ -81,6 +86,7 @@ export class ClientAssistantStream {
       const attempt = this.attemptFor(event.data.turn, event.data.step)
       if (attempt === undefined) return this.publish(entry)
       if (attempt.legacyChunkSeqs.has(event.seq)) return this.publish(entry)
+      if (this.pendingChunks.has(event.seq)) return { type: 'rebaseline' }
       this.pendingChunks.set(event.seq, entry)
       return undefined
     }
@@ -88,7 +94,9 @@ export class ClientAssistantStream {
       if (event.surfaceOp !== 'append') return this.publish(entry)
       const attempt = this.attemptFor(event.data.turn, event.data.step)
       if (attempt === undefined) return this.publish(entry)
-      this.pendingMessages.set(positionKey(event.data.turn, event.data.step), entry)
+      const key = positionKey(event.data.turn, event.data.step)
+      if (this.pendingMessages.has(key)) return { type: 'rebaseline' }
+      this.pendingMessages.set(key, { entry, sourceEventSeqs: event.sourceEventSeqs })
       return undefined
     }
     return this.publish(entry)
@@ -112,7 +120,11 @@ export class ClientAssistantStream {
         return undefined
       case 'chunk': {
         const attempt = this.attempts.get(String(frame.attemptId))
-        if (attempt === undefined || frame.index !== attempt.nextIndex) return { type: 'rebaseline' }
+        // A controller mounted after the Host saw this attempt has no start
+        // frame to reconstruct. Its durable events already publish directly;
+        // ignore the transient suffix until the next known attempt.
+        if (attempt === undefined) return undefined
+        if (frame.index !== attempt.nextIndex) return { type: 'rebaseline' }
         attempt.nextIndex += 1
         attempt.legacyChunkSeqs.add(frame.legacyChunkSeq)
         if (this.publishedSeqs.has(frame.legacyChunkSeq)) return undefined
@@ -124,23 +136,22 @@ export class ClientAssistantStream {
       case 'end': {
         const attempt = this.attempts.get(String(frame.attemptId))
         this.attempts.delete(String(frame.attemptId))
-        if (attempt === undefined) return { type: 'rebaseline' }
+        if (attempt === undefined) return undefined
         if (frame.index !== attempt.nextIndex) return { type: 'rebaseline' }
         if (!sameSeqs([...attempt.legacyChunkSeqs], frame.legacyChunkSeqs)) {
           return { type: 'rebaseline' }
         }
         const key = positionKey(attempt.turn, attempt.step)
-        const entry = this.pendingMessages.get(key)
-        if (entry === undefined) {
+        const pending = this.pendingMessages.get(key)
+        if (pending === undefined) {
           return frame.outcome === 'aborted' ? undefined : { type: 'rebaseline' }
         }
-        if (entry.event.type !== 'assistant/message') return { type: 'rebaseline' }
-        const sourceEventSeqs = entry.event.sourceEventSeqs
+        const sourceEventSeqs = pending.sourceEventSeqs
         if (sourceEventSeqs === undefined || !sameSeqs(sourceEventSeqs, frame.legacyChunkSeqs)) {
           return { type: 'rebaseline' }
         }
         this.pendingMessages.delete(key)
-        return this.publish(entry)
+        return this.publish(pending.entry)
       }
     }
   }

packages/api/session-controller/src/index.ts

@@ -369,7 +369,8 @@ export class SessionController extends TypertRemoteService {
    * Follow one Session log from its opening or resume cursor.
    * @param request - durable address and last committed sequence already held by the caller.
    * @param signal - cancellation owned by the Remote stream carrier.
-   * @returns a complete opening snapshot followed by gap-free event frames.
+   * @returns a complete opening snapshot followed by gap-free durable event
+   *   frames and optional cursorless assistant-stream frames.
    */
   @Remote({ mode: 'stream' })
   follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame> {

packages/api/session-controller/tests/assistant-stream.client.spec.ts

@@ -0,0 +1,210 @@
+import { describe, expect, it } from 'vitest'
+import { LlmAttemptId, createAssistantMessage } from '@deepseek-ai/dsh-llm'
+import { SessionSeq, type SessionEvent } from '@deepseek-ai/dsh-session'
+import type {
+  SessionAssistantStreamBaseline,
+  SessionAssistantStreamFrame,
+} from '../src/types.ts'
+import { ClientAssistantStream } from '../src/client/sessions/assistant-stream.ts'
+import type { SessionLiveEventEntry } from '../src/client/contract/events.ts'
+
+const ATTEMPT = LlmAttemptId('session:1')
+
+function entry(event: SessionEvent): SessionLiveEventEntry {
+  return { type: 'event', event }
+}
+
+function ordinary(seq: number): SessionLiveEventEntry {
+  return entry({ type: 'turn/start', seq: SessionSeq(seq), time: seq, data: { turn: 1 } })
+}
+
+function chunk(seq: number, turn = 1, step = 1): SessionLiveEventEntry {
+  return entry({
+    type: 'assistant/chunk',
+    seq: SessionSeq(seq),
+    time: seq,
+    data: { turn, step, chunk: { type: 'text-delta', index: 0, text: `chunk-${seq}` } },
+  })
+}
+
+function message(
+  seq: number,
+  sourceEventSeqs: readonly number[] | undefined,
+  turn = 1,
+  step = 1,
+  surfaceOp: 'append' | { readonly op: 'replace'; readonly start: number; readonly end: number } = 'append',
+): SessionLiveEventEntry {
+  return entry({
+    type: 'assistant/message',
+    seq: SessionSeq(seq),
+    time: seq,
+    data: {
+      turn,
+      step,
+      message: createAssistantMessage({
+        content: [{ type: 'text', text: 'done' }],
+        source: { provider: 'mock', model: 'mock' },
+      }),
+    },
+    surfaceOp: surfaceOp === 'append'
+      ? surfaceOp
+      : { ...surfaceOp, start: SessionSeq(surfaceOp.start), end: SessionSeq(surfaceOp.end) },
+    ...(sourceEventSeqs === undefined ? {} : { sourceEventSeqs: sourceEventSeqs.map(SessionSeq) }),
+  })
+}
+
+function start(attemptId = ATTEMPT): SessionAssistantStreamFrame {
+  return { type: 'start', attemptId, revision: 1, startedTime: 1, turn: 1, step: 1 }
+}
+
+function chunkFrame(
+  index: number,
+  legacyChunkSeq: number,
+  attemptId = ATTEMPT,
+): SessionAssistantStreamFrame {
+  return {
+    type: 'chunk',
+    attemptId,
+    revision: index + 2,
+    index,
+    chunk: { type: 'text-delta', index: 0, text: `chunk-${legacyChunkSeq}` },
+    legacyChunkSeq,
+  }
+}
+
+function end(
+  outcome: 'committed' | 'aborted',
+  index: number,
+  legacyChunkSeqs: readonly number[],
+  attemptId = ATTEMPT,
+): SessionAssistantStreamFrame {
+  return { type: 'end', attemptId, revision: index + 2, index, outcome, legacyChunkSeqs }
+}
+
+function baseline(
+  chunks: SessionAssistantStreamBaseline['attempts'][number]['chunks'] = [],
+  legacyChunkSeqs: readonly number[] = [],
+): SessionAssistantStreamBaseline {
+  return {
+    revision: chunks.length + 1,
+    attempts: [{ attemptId: ATTEMPT, startedTime: 1, turn: 1, step: 1, chunks, legacyChunkSeqs }],
+  }
+}
+
+describe('ClientAssistantStream', () => {
+  it('passes through ordinary and unowned durable events and adopts a baseline', () => {
+    const stream = new ClientAssistantStream()
+    const opening = ordinary(0)
+    const window = [opening]
+    expect(stream.replace(window)).toBe(window)
+
+    stream.replace([], baseline([{ type: 'text-delta', index: 0, text: 'old' }], [5]))
+    const represented = chunk(5)
+    expect(stream.acceptDurable(represented)).toEqual({ type: 'publish', entry: represented })
+
+    const unknownChunk = chunk(6, 2, 1)
+    expect(stream.acceptDurable(unknownChunk)).toEqual({ type: 'publish', entry: unknownChunk })
+    const replacement = message(7, [5], 1, 1, { op: 'replace', start: 0, end: 0 })
+    expect(stream.acceptDurable(replacement)).toEqual({ type: 'publish', entry: replacement })
+    const unknownMessage = message(8, [6], 2, 1)
+    expect(stream.acceptDurable(unknownMessage)).toEqual({ type: 'publish', entry: unknownMessage })
+    const turn = ordinary(9)
+    expect(stream.acceptDurable(turn)).toEqual({ type: 'publish', entry: turn })
+  })
+
+  it('stages owned durable chunks and messages until their matching frames', () => {
+    const stream = new ClientAssistantStream()
+    stream.replace([])
+    expect(stream.acceptFrame(start())).toBeUndefined()
+
+    const durableChunk = chunk(2)
+    expect(stream.acceptDurable(durableChunk)).toBeUndefined()
+    expect(stream.acceptFrame(chunkFrame(0, 2))).toEqual({ type: 'publish', entry: durableChunk })
+
+    const durableMessage = message(3, [2])
+    expect(stream.acceptDurable(durableMessage)).toBeUndefined()
+    expect(stream.acceptFrame(end('committed', 1, [2]))).toEqual({ type: 'publish', entry: durableMessage })
+  })
+
+  it('requests a new baseline for duplicate staged durable identities', () => {
+    const chunks = new ClientAssistantStream()
+    chunks.acceptFrame(start())
+    expect(chunks.acceptDurable(chunk(2))).toBeUndefined()
+    expect(chunks.acceptDurable(chunk(2))).toEqual({ type: 'rebaseline' })
+
+    const messages = new ClientAssistantStream()
+    messages.acceptFrame(start())
+    expect(messages.acceptDurable(message(3, []))).toBeUndefined()
+    expect(messages.acceptDurable(message(4, []))).toEqual({ type: 'rebaseline' })
+  })
+
+  it('falls back to durable publication for frames from an unknown attempt', () => {
+    const stream = new ClientAssistantStream()
+    const unknown = LlmAttemptId('session:unknown')
+
+    expect(stream.acceptFrame(chunkFrame(0, 2, unknown))).toBeUndefined()
+    expect(stream.acceptFrame(end('committed', 0, [], unknown))).toBeUndefined()
+    const durableChunk = chunk(2)
+    expect(stream.acceptDurable(durableChunk)).toEqual({ type: 'publish', entry: durableChunk })
+  })
+
+  it('rebaselines known attempts on chunk index or publication-order mismatch', () => {
+    const wrongIndex = new ClientAssistantStream()
+    wrongIndex.acceptFrame(start())
+    expect(wrongIndex.acceptFrame(chunkFrame(1, 2))).toEqual({ type: 'rebaseline' })
+
+    const frameFirst = new ClientAssistantStream()
+    frameFirst.acceptFrame(start())
+    expect(frameFirst.acceptFrame(chunkFrame(0, 2))).toEqual({ type: 'rebaseline' })
+
+    const alreadyPublished = new ClientAssistantStream()
+    alreadyPublished.replace([chunk(2)], baseline())
+    expect(alreadyPublished.acceptFrame(chunkFrame(0, 2))).toBeUndefined()
+  })
+
+  it('rebaselines a known terminal frame with inconsistent index or chunk provenance', () => {
+    const wrongIndex = new ClientAssistantStream()
+    wrongIndex.acceptFrame(start())
+    expect(wrongIndex.acceptFrame(end('committed', 1, []))).toEqual({ type: 'rebaseline' })
+
+    const wrongLength = new ClientAssistantStream()
+    wrongLength.acceptFrame(start())
+    wrongLength.acceptDurable(chunk(2))
+    expect(wrongLength.acceptFrame(chunkFrame(0, 2))).toEqual({ type: 'publish', entry: chunk(2) })
+    expect(wrongLength.acceptFrame(end('committed', 1, []))).toEqual({ type: 'rebaseline' })
+
+    const wrongMember = new ClientAssistantStream()
+    wrongMember.acceptFrame(start())
+    const durable = chunk(2)
+    wrongMember.acceptDurable(durable)
+    wrongMember.acceptFrame(chunkFrame(0, 2))
+    expect(wrongMember.acceptFrame(end('committed', 1, [3]))).toEqual({ type: 'rebaseline' })
+  })
+
+  it('settles committed and aborted attempts against staged assistant messages', () => {
+    const missingCommitted = new ClientAssistantStream()
+    missingCommitted.acceptFrame(start())
+    expect(missingCommitted.acceptFrame(end('committed', 0, []))).toEqual({ type: 'rebaseline' })
+
+    const emptyAborted = new ClientAssistantStream()
+    emptyAborted.acceptFrame(start())
+    expect(emptyAborted.acceptFrame(end('aborted', 0, []))).toBeUndefined()
+
+    const stagedAborted = new ClientAssistantStream()
+    stagedAborted.acceptFrame(start())
+    const interrupted = message(3, [])
+    stagedAborted.acceptDurable(interrupted)
+    expect(stagedAborted.acceptFrame(end('aborted', 0, [])))
+      .toEqual({ type: 'publish', entry: interrupted })
+
+    const missingSources = new ClientAssistantStream()
+    missingSources.acceptFrame(start())
+    missingSources.acceptDurable(message(3, undefined))
+    expect(missingSources.acceptFrame(end('committed', 0, []))).toEqual({ type: 'rebaseline' })
+
+    const mismatchedSources = new ClientAssistantStream()
+    mismatchedSources.acceptFrame(start())
+    mismatchedSources.acceptDurable(message(3, [7]))
+    expect(mismatchedSources.acceptFrame(end('committed', 0, []))).toEqual({ type: 'rebaseline' })
+  })
+})

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: 5e4ab0a741f0b7dae821e52f1a9b0a1faa90ceed
-README.zh.md: ef47f897978c8254242075a559f56faa2dceca19
+README.md: c5985f585f92f38cb27ae1385b1052dbce1d9b9d
+README.zh.md: 615cdd5a6e3e71ae129d782403f09d0309913929

packages/core/agent-loop/README.md

@@ -111,7 +111,7 @@ The loop is the production acquisition point for session write handles. When `ct
 
 ### Turn and step flow
 
-The driver owns one agent for its lifetime and runs inside `ctx.agents.withInitiator(agent, ...)`. At a turn boundary it opens the durable turn, then atomically claims pending next-step input plus one queued prompt; between steps it claims only next-step input. `agent/pre-step` decides what enters the step. An entered decision appends its complete `user/message` batch before the driver can claim again, while a rejected decision appends none. Each successful model call appends one `assistant/message` anchor citing its chunk seqs, and a cancelled stream appends an `interrupted: true` anchor with the delivered prefix so the next request contains what the user saw. Within a step, exclusive calls form barriers and parallel-safe calls use the bounded rolling pool; policy, durable results, and result context remain model-ordered.
+The driver owns one agent for its lifetime and runs inside `ctx.agents.withInitiator(agent, ...)`. At a turn boundary it opens the durable turn, then atomically claims pending next-step input plus one queued prompt; between steps it claims only next-step input. `agent/pre-step` decides what enters the step. An entered decision appends its complete `user/message` batch before the driver can claim again, while a rejected decision appends none. Each model attempt emits one process-local `start`, emits every `chunk` only after the matching durable `assistant/chunk`, and emits exactly one terminal `end`; final assembly or message-append failure settles it as `aborted`, while `committed` follows the durable `assistant/message`. Each successful model call appends one message anchor citing its chunk seqs, and a cancelled stream appends an `interrupted: true` anchor with the delivered prefix so the next request contains what the user saw. Within a step, exclusive calls form barriers and parallel-safe calls use the bounded rolling pool; policy, durable results, and result context remain model-ordered.
 
 ### Failure and cancellation
 

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

@@ -111,7 +111,7 @@ const handle = await ctx.agents.create({
 
 ### 轮次与步骤流程
 
-驱动器在其整个生命周期内拥有一个 agent，并在 `ctx.agents.withInitiator(agent, ...)` 内运行。在轮次边界，它先打开持久轮次，再原子领取待处理的 next-step 输入与一条排队提示词；在步骤之间则只领取 next-step 输入。`agent/pre-step` 决定什么进入该步骤。进入步骤的决定会在驱动器再次领取消息前追加完整的 `user/message` 批次，被拒绝的决定则不追加任何消息。每次成功的模型调用都恰好追加一个引用其分片 seq 的 `assistant/message` 锚点，被取消的流则追加带 `interrupted: true` 的锚点并携带已交付前缀，使下一次请求包含用户看到的内容。在步骤内，独占调用形成屏障，并行安全调用使用有界滚动池；策略、持久结果与结果上下文保持模型顺序。
+驱动器在其整个生命周期内拥有一个 agent，并在 `ctx.agents.withInitiator(agent, ...)` 内运行。在轮次边界，它先打开持久轮次，再原子领取待处理的 next-step 输入与一条排队提示词；在步骤之间则只领取 next-step 输入。`agent/pre-step` 决定什么进入该步骤。进入步骤的决定会在驱动器再次领取消息前追加完整的 `user/message` 批次，被拒绝的决定则不追加任何消息。每次模型尝试会发出一个进程本地 `start`，仅在匹配的持久 `assistant/chunk` 之后发出各个 `chunk`，并恰好发出一个终态 `end`；最终组装或消息追加失败时以 `aborted` 结算，`committed` 则出现在持久 `assistant/message` 之后。每次成功的模型调用都恰好追加一个引用其分片 seq 的 message 锚点，被取消的流则追加带 `interrupted: true` 的锚点并携带已交付前缀，使下一次请求包含用户看到的内容。在步骤内，独占调用形成屏障，并行安全调用使用有界滚动池；策略、持久结果与结果上下文保持模型顺序。
 
 ### 失败与取消
 

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

@@ -380,80 +380,88 @@ export class ReactLoopAgent implements Agent {
         liveStarted = true
         for await (const chunk of stream) {
           signal.throwIfAborted()
-          const legacyChunkSeq = this.session.append('assistant/chunk', { turn, step, chunk }).seq
-          chunkSeqs.push(legacyChunkSeq)
-          assembler.push(chunk)
-          live.push(chunk, legacyChunkSeq)
+          const durable = this.session.append('assistant/chunk', { turn, step, chunk })
+          chunkSeqs.push(durable.seq)
+          assembler.push(durable.data.chunk)
+          live.push(durable.data.chunk, durable.seq)
         }
         signal.throwIfAborted()
       } catch (error: unknown) {
-        if (signal.aborted) {
-          const content = assembler.interruptedBlocks()
-          if (content.length > 0) {
-            this.session.append('assistant/message', {
-              turn,
-              step,
-              message: createAssistantMessage({
-                content,
-                source: { provider: request.provider, model: request.model },
-              }),
-              interrupted: true,
-              ...assembler.usage === undefined ? {} : { usage: assembler.usage },
-            }, { surfaceOp: 'append', sourceEventSeqs: chunkSeqs })
+        try {
+          if (signal.aborted) {
+            const content = assembler.interruptedBlocks()
+            if (content.length > 0) {
+              this.session.append('assistant/message', {
+                turn,
+                step,
+                message: createAssistantMessage({
+                  content,
+                  source: { provider: request.provider, model: request.model },
+                }),
+                interrupted: true,
+                ...assembler.usage === undefined ? {} : { usage: assembler.usage },
+              }, { surfaceOp: 'append', sourceEventSeqs: chunkSeqs })
+            }
           }
+        } finally {
+          if (liveStarted) live.end('aborted')
         }
-        if (liveStarted) live.end('aborted')
         throw error
       }
-      const finish = assembler.finish
-      if (finish.kind === 'error' || finish.kind === 'aborted') {
-        live.end('aborted')
-        const action = await this.dispatch.waterfall(
-          'agent/request-error', {
+      try {
+        const finish = assembler.finish
+        if (finish.kind === 'error' || finish.kind === 'aborted') {
+          live.end('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)
+          }
+          continue
+        }
+
+        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,
-            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
+        live.end('committed')
+        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
+      } catch (error: unknown) {
+        if (!live.ended) live.end('aborted')
+        throw error
       }
-
-      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 },
-      )
-      live.end('committed')
-      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
     }
   }
 

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

@@ -8,9 +8,13 @@ import type { SessionId, SessionSeq } from '@deepseek-ai/dsh-session'
 export class AssistantStreamAttempt {
   private readonly legacyChunkSeqs: SessionSeq[] = []
   private index = 0
+  private terminal = false
   /** Process-local attempt identity. */
   readonly attemptId: LlmAttemptId
 
+  /** Whether this started attempt has emitted its terminal frame. */
+  get ended(): boolean { return this.terminal }
+
   /**
    * @param sessionId - identity embedded only in the process-local attempt id.
    * @param attempt - attached-Session-local attempt counter.
@@ -55,8 +59,9 @@ export class AssistantStreamAttempt {
     })
   }
 
-  /** Publish terminal settlement after the matching durable assistant message commits. */
+  /** Publish terminal settlement; `committed` follows its durable assistant message. */
   end(outcome: 'committed' | 'aborted'): void {
+    this.terminal = true
     this.emit({
       type: 'end',
       attemptId: this.attemptId,

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

@@ -60,9 +60,14 @@ describe('agent loop', () => {
     })
     const frames: AssistantStreamFrame[] = []
     let committedAfterMessage = false
+    const chunkUsesDurableValue: boolean[] = []
     ctx.on('agent/assistant-stream', ({ agent: subject, frame }) => {
       if (subject !== agent) return
       frames.push(frame)
+      if (frame.type === 'chunk') {
+        const durable = agent.session.eventAt(frame.legacyChunkSeq)
+        chunkUsesDurableValue.push(durable?.type === 'assistant/chunk' && frame.chunk === durable.data.chunk)
+      }
       if (frame.type === 'end' && frame.outcome === 'committed') {
         committedAfterMessage = agent.session.snapshotEvents().at(-1)?.type === 'assistant/message'
       }
@@ -87,6 +92,7 @@ describe('agent loop', () => {
     expect(chunks.map(frame => frame.index)).toEqual(chunks.map((_frame, index) => index))
     const durableChunks = agent.session.snapshotEvents().filter(event => event.type === 'assistant/chunk')
     expect(chunks).toHaveLength(durableChunks.length)
+    expect(chunkUsesDurableValue).toEqual(chunks.map(() => true))
     const end = frames.at(-1)
     expect(end).toMatchObject({
       type: 'end', outcome: 'committed', index: chunks.length,
@@ -97,6 +103,58 @@ describe('agent loop', () => {
     }
   })
 
+  it('aborts a started live attempt when final block assembly fails', async () => {
+    const chunks: StreamChunk[] = [
+      { type: 'block-start', index: 0, blockType: 'external-block' } as unknown as StreamChunk,
+      { type: 'finish', reason: { kind: 'stop' } },
+    ]
+    const ctx = await harness(new MockAdapter([chunks]))
+    const agent = await ctx.agentLoop.create(SessionId('assistant-assembly-failure'), {
+      provider: 'mock',
+      model: 'mock',
+    })
+    const frames: AssistantStreamFrame[] = []
+    ctx.on('agent/assistant-stream', ({ agent: subject, frame }) => {
+      if (subject === agent) frames.push(frame)
+    })
+
+    send(agent, 'produce an incomplete external block')
+    await waitForIdle(ctx, agent)
+
+    expect(frames.map(frame => frame.revision)).toEqual([1, 2, 3, 4])
+    expect(frames.at(-1)).toMatchObject({ type: 'end', outcome: 'aborted', index: 2 })
+    expect(agent.session.snapshotEvents().some(event => event.type === 'assistant/message')).toBe(false)
+    expect(agent.session.snapshotEvents().at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: { reason: { kind: 'error' } },
+    })
+  })
+
+  it('ends a failed attempt before retrying with a new dense attempt', async () => {
+    const failed: StreamChunk[] = [
+      { type: 'usage', usage: { inputTokens: 1, outputTokens: 0 } },
+      { type: 'finish', reason: { kind: 'error', failure: { message: 'retry', code: 'SERVER' } } },
+    ]
+    const ctx = await harness(new MockAdapter([failed, textResponse('recovered')]))
+    const agent = await ctx.agentLoop.create(SessionId('assistant-retry-attempt'), {
+      provider: 'mock',
+      model: 'mock',
+    })
+    const frames: AssistantStreamFrame[] = []
+    ctx.on('agent/assistant-stream', ({ agent: subject, frame }) => {
+      if (subject === agent) frames.push(frame)
+    })
+    ctx.on('agent/request-error', async () => ({ kind: 'retry' as const }))
+
+    send(agent, 'retry once')
+    await waitForIdle(ctx, agent)
+
+    expect(frames.map(frame => frame.revision)).toEqual(frames.map((_frame, index) => index + 1))
+    expect(frames.filter(frame => frame.type === 'start')).toHaveLength(2)
+    expect(frames.filter(frame => frame.type === 'end').map(frame => frame.outcome))
+      .toEqual(['aborted', 'committed'])
+  })
+
   it('does not emit an end frame when prepared dispatch throws before start', async () => {
     const ctx = await harness(new MockAdapter([]))
     const agent = await ctx.agentLoop.create(SessionId('assistant-dispatch-throw'), {

packages/core/agent/src/runtime-types.ts

@@ -75,7 +75,7 @@ export type AssistantStreamFrame =
   | {
     readonly type: 'start'
     readonly attemptId: LlmAttemptId
-    /** Monotone while this Session remains attached to this process. */
+    /** Monotone within one attached Agent lifecycle; replacement restarts at 1. */
     readonly revision: number
     /** Safe-integer wall-clock time captured when this attempt started. */
     readonly startedTime: number
@@ -98,7 +98,7 @@ export type AssistantStreamFrame =
     readonly revision: number
     /** Number of chunk frames emitted by this attempt. */
     readonly index: number
-    /** The durable assistant message committed before this notification. */
+    /** `committed` follows its durable message; `aborted` has no such requirement. */
     readonly outcome: 'committed' | 'aborted'
     /** Every durable v1 chunk represented by this attempt. */
     readonly legacyChunkSeqs: readonly SessionSeq[]

packages/extensions/tool-cordis/src/api-catalog.ts

@@ -1435,7 +1435,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
         signature: '@Remote({ mode: \'stream\' }) follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame>',
         description: 'Follow one Session log from its opening or resume cursor.',
         parameters: [{ name: 'request', description: 'durable address and last committed sequence already held by the caller.' }, { name: 'signal', description: 'cancellation owned by the Remote stream carrier.' }],
-        returns: 'a complete opening snapshot followed by gap-free event frames.',
+        returns: 'a complete opening snapshot followed by gap-free durable event frames and optional cursorless assistant-stream frames.',
       },
       {
         signature: '@Remote({ mode: \'stream\' }) control(signal: AbortSignal): AsyncIterable<SessionControlFrame>',

packages/llm/llm/src/brand.ts

@@ -51,12 +51,12 @@ export function ProviderRequestId(id: string): ProviderRequestId {
   return brandString<ProviderRequestId>(id)
 }
 
-/** Process-local identity of one loop-owned model streaming attempt. */
+/** Identity of one model streaming attempt, unique within one Agent lifecycle. */
 export type LlmAttemptId = Branded<'LlmAttemptId'>
 
 /**
  * Brand one loop-owned streaming attempt identifier.
- * @param id - the opaque process-local identifier.
+ * @param id - the opaque Agent-lifecycle-local identifier.
  * @returns the same string with the attempt-id brand.
  */
 export function LlmAttemptId(id: string): LlmAttemptId {

scripts/type-equiv.manifest.json

@@ -1911,6 +1911,11 @@
       "symbol": "AgentStatus",
       "source": "packages/core/agent/src/runtime-types.ts"
     },
+    {
+      "doc": "docs/subsystems/core.md",
+      "symbol": "AssistantStreamFrame",
+      "source": "packages/core/agent/src/runtime-types.ts"
+    },
     {
       "doc": "docs/subsystems/core.md",
       "symbol": "AgentOptions",

vitest.config.ts

@@ -221,8 +221,9 @@ export default defineConfig({
         'packages/client/ui-workspace/src/client/rows/WorkspaceBrowser.tsx',
         'packages/client/ui-renderer/src/client/*',
         // Session object internals retain the runtime GUI debt exemption; the
-        // new Controller entry, transport, Agent scope, and adapters stay gated.
-        'packages/api/session-controller/src/client/sessions/*',
+        // assistant-stream reconciler, Controller entry, transport, Agent scope,
+        // and adapters stay gated.
+        'packages/api/session-controller/src/client/sessions/!(assistant-stream).ts',
         'packages/api/session-controller/src/client/ordered-baseline.ts',
         'packages/api/session-controller/src/client/time-zone.ts',
         // Keep the browser conversation tree under its existing GUI debt