From 0bb7bba015f5cc382d4997c2de93f0e22e154144 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Wed, 2 Sep 2026 03:49:17 +0800 Subject: [PATCH] fix(session): close live assistant framing races --- ...-31-live-assistant-stream-frames.i18n.yaml | 4 +- ...2026-08-31-live-assistant-stream-frames.md | 4 +- ...6-08-31-live-assistant-stream-frames.zh.md | 4 +- .../api/session-controller/README.i18n.yaml | 4 +- packages/api/session-controller/README.md | 2 +- packages/api/session-controller/README.zh.md | 2 +- .../src/client/sessions/assistant-stream.ts | 22 +----- .../src/client/sessions/session.ts | 3 + .../session-history-journal.host.spec.ts | 68 +++++++++++++++++++ .../tests/sessions-service.client.spec.ts | 61 +++++++++++++---- packages/core/agent-loop/src/agent.ts | 4 +- packages/core/agent-loop/tests/loop.spec.ts | 31 ++++++++- 12 files changed, 163 insertions(+), 46 deletions(-) diff --git a/.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.i18n.yaml b/.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.i18n.yaml index cb72a0e41a..0fb0ceca81 100644 --- a/.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.i18n.yaml +++ b/.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: 91022a8c857ca05841efedb0e468b1d2e9d148a0 -2026-08-31-live-assistant-stream-frames.zh.md: f3f4c3138f944054d73026d7a6592611ea18c5e0 +2026-08-31-live-assistant-stream-frames.md: 9e848cadbfa71f8281caf14207b36d05f1f95d99 +2026-08-31-live-assistant-stream-frames.zh.md: dc672d6d8af31f1b45694299125d3f91cb1874ee diff --git a/.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.md b/.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.md index 91022a8c85..9e848cadbf 100644 --- a/.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.md +++ b/.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. 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. When opening lands between a durable final message and its end frame, the Web Session exposes the active chunks, stages only the final message with identical ordered legacy seq provenance, and releases it after the matching `end.index`; 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 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. ## 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. The frame declaration remains agent-scoped, so a listener observes only its owning Agent unless it explicitly registers globally. +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. diff --git a/.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.zh.md b/.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.zh.md index f3f4c3138f..dc672d6d8a 100644 --- a/.agents/notes/implemented/architecture/2026-08-31-live-assistant-stream-frames.zh.md +++ b/.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 位置。循环在匹配的实时 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 位于最终持久 message 与对应 end 帧之间时，Web Session 会公开活跃 chunk，只暂存 ordered legacy seq 来源完全相同的最终 message，并在匹配的 `end.index` 到达后释放；同一 Turn 和 Step 中更早的 retry 仍保持可见。revision、连续 index 或来源缺口会重新打开 follow 并替换 baseline。TypeScript 和 Python SDK 协议不公开这些帧。持久 log 仍然是重放和模型历史的真源。 +`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 仍然是重放和模型历史的真源。 ## 曾考虑的替代方案 @@ -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。帧声明保持 agent 作用域，因此监听器只观察所属 Agent，除非它显式全局注册。 +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，除非它显式全局注册。 diff --git a/packages/api/session-controller/README.i18n.yaml b/packages/api/session-controller/README.i18n.yaml index bfe10777f1..c1a4319677 100644 --- a/packages/api/session-controller/README.i18n.yaml +++ b/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: 81baecb0f08bd469534ec980e7cfa040dd3800fa -README.zh.md: 5267c8cb20461ecbdb44b2b98634003d9c22904e +README.md: 601e1037792ff0965ce862324d1e85a714cb3c45 +README.zh.md: a0bc858ce92783b6e3e1c95a3664faa85f10a03a diff --git a/packages/api/session-controller/README.md b/packages/api/session-controller/README.md index 81baecb0f0..601e103779 100644 --- a/packages/api/session-controller/README.md +++ b/packages/api/session-controller/README.md @@ -27,7 +27,7 @@ History pages and follow opening snapshots carry a discriminated `SessionHistory Each endpoint states its activation policy. List, search, attachment, history pages, log following, skill discovery, and workspace-path opening can inspect persistence without activating an Agent; `canOpenWorkspacePath()` reports native-opening availability without addressing a Session. Queue mutation and cancellation require live state; model, rename, prompt, and file-reference operations may resolve or resume an ordinary Session. Create and fork are the only operations that create a new Agent directly. The skill catalog instead uses a live Agent when present or the recorded preset's standing scope when cold, so listing never starts an Agent. -The Client adapter exposes `SessionEventStream`, a Gateway `RemoteJournalStream` bound to one ordinary or direct-subagent address. It opens follow before the initial page, publishes only contiguous `replace`, `prepend`, and `append` changes, and repairs reconnect or sequence gaps through a tail page. Backwards paging has two verbs: `loadOlder()` pulls one 50-message page, and `loadThrough(seq)` — the turn-jump loader — loops 200-message pages until the window covers the target seq, lowering a shared target on repeated calls, stopping on a page that makes no progress, and reporting busy through the same `loadingOlder` snapshot bit. The Web adapter explicitly opts into cursorless assistant notifications: each opening carries active attempts with their `startedTime`, current chunks, and exact v1 seq provenance. The Host captures a follower-local arrival ordinal with that baseline and suppresses buffered frames at or before the cut; a replacement Agent may restart frame revision at one. A revision or dense-index gap reopens follow. If an opening observes a final durable message before its committed end frame, the Session object exposes the active chunks but stages only the message whose ordered source seqs match that attempt, then publishes it when `end.index` equals the next chunk position. Ordinary records cover `[event.seq, event.seq]`; packed rows cover `[event.seq, event.seq + memberCount - 1]`. A business, persistence, or unresolved continuity failure terminates the stream, while only physical carrier loss selects automatic resumption. `SessionControlStream` is a Gateway `RemoteSnapshotStream`; every generation opens with a complete process-local baseline, so reconnect replaces queue, jobs, and projection state instead of treating transient values as durable events. +The Client adapter exposes `SessionEventStream`, a Gateway `RemoteJournalStream` bound to one ordinary or direct-subagent address. It opens follow before the initial page, publishes only contiguous `replace`, `prepend`, and `append` changes, and repairs reconnect or sequence gaps through a tail page. Backwards paging has two verbs: `loadOlder()` pulls one 50-message page, and `loadThrough(seq)` — the turn-jump loader — loops 200-message pages until the window covers the target seq, lowering a shared target on repeated calls, stopping on a page that makes no progress, and reporting busy through the same `loadingOlder` snapshot bit. The Web adapter explicitly opts into cursorless assistant notifications: each opening carries active attempts with their `startedTime`, current chunks, and exact v1 seq provenance. The Host captures a follower-local arrival ordinal with that baseline and suppresses buffered frames at or before the cut; a replacement Agent may restart frame revision at one. If the Assistant baseline already contains a chunk whose durable event was captured after the opening page, the Client publishes that event on arrival from the exact baseline seq proof. A final durable message arriving after an active opening remains staged until its committed end frame carries the same ordered source seqs and `end.index` equals the next chunk position. A revision or dense-index gap reopens follow. A durable gap-repair page carries no Assistant baseline; the Client clears transient attempts, and a held notification reopens follow for a paired page and baseline. Ordinary records cover `[event.seq, event.seq]`; packed rows cover `[event.seq, event.seq + memberCount - 1]`. A business, persistence, or unresolved continuity failure terminates the stream, while only physical carrier loss selects automatic resumption. `SessionControlStream` is a Gateway `RemoteSnapshotStream`; every generation opens with a complete process-local baseline, so reconnect replaces queue, jobs, and projection state instead of treating transient values as durable events. The Session object also carries local submission echoes: `session.beginSubmission` inserts one into `SessionSnapshot.pendingSubmissions` synchronously, before the caller serializes and prompts, so a conversation UI can show the message on the submit click's own frame. Session derives each echo's `transcript`, `queued`, or `steering` placement from its current running state and the requested delivery mode, then retains that placement while serialization is in flight. The prompt's `requestId` is the correlation identity: the Host echoes it as the durable user source's `rpcId`, and queue occurrences project it as `SessionQueuedItem.rpcId`. An echo retires one animation frame after its durable event or queue occurrence is observed (the delay keeps it renderable until the replacement is ready), immediately when its identified prompt fails or is abandoned, and as failed on disposal; each retirement fires the registered `onRetire` callback exactly once. Echoes are Client memory only; reload and reconnect rebuild the conversation from durable events alone. diff --git a/packages/api/session-controller/README.zh.md b/packages/api/session-controller/README.zh.md index 5267c8cb20..a0bc858ce9 100644 --- a/packages/api/session-controller/README.zh.md +++ b/packages/api/session-controller/README.zh.md @@ -27,7 +27,7 @@ kind: "package-reference" 每个 endpoint 都声明自己的激活策略。列表、搜索、附件、历史页、日志跟随、skill 发现和工作区路径打开可以在不激活 Agent 的情况下检查 persistence；`canOpenWorkspacePath()` 无需指定 Session 即可报告原生打开能力。queue 变更与取消要求 live 状态；模型、重命名、prompt 和文件引用操作可以解析或恢复普通 Session。只有 create 与 fork 会直接创建新 Agent。skill 目录则优先使用已有 live Agent，否则使用所记录 preset 的常驻 scope，因此列表查询绝不会启动 Agent。 -Client adapter 提供 `SessionEventStream`，即绑定到一个普通 Session 或 direct subagent address 的 Gateway `RemoteJournalStream`。它在读取首个 page 前打开 follow，只发布连续的 `replace`、`prepend` 和 `append` 变更，并通过 tail page 修复重连或 seq 缺口。向后分页有两个动词：`loadOlder()` 拉一页 50 条 message，而 `loadThrough(seq)`——轮次跳转加载器——按 200 条 message 一页循环拉取直到窗口覆盖目标 seq，重复调用会下调共享目标，遇到无进展的页即停止，忙碌状态复用同一个 `loadingOlder` 快照位。Web adapter 显式选择接收无 cursor 的 assistant 通知：每个 opening 携带活跃尝试的 `startedTime`、当前 chunk 和精确 v1 seq 来源。Host 会随该 baseline 捕获 follower 本地到达序号，并抑制该 cut 及之前的 buffered frame；replacement Agent 可以从 revision 一重新开始。revision 或连续 index 缺口会重新打开 follow。如果 opening 已观察到最终持久 message，但尚未观察到对应的已提交 end 帧，Session 对象会公开活跃 chunk，但只暂存 ordered source seq 与该尝试匹配的 message；当 `end.index` 等于下一个 chunk 位置时，再发布该 message。普通 record 覆盖 `[event.seq, event.seq]`，packed row 覆盖 `[event.seq, event.seq + memberCount - 1]`。业务、persistence 或无法恢复的连续性错误会终止 stream，只有物理载体断开才触发自动恢复。`SessionControlStream` 是 Gateway `RemoteSnapshotStream`；每代都以完整的进程本地 baseline 开始，因此重连会替换 queue、jobs 和 projection 状态，而不会把瞬态值当作 durable event。 +Client adapter 提供 `SessionEventStream`，即绑定到一个普通 Session 或 direct subagent address 的 Gateway `RemoteJournalStream`。它在读取首个 page 前打开 follow，只发布连续的 `replace`、`prepend` 和 `append` 变更，并通过 tail page 修复重连或 seq 缺口。向后分页有两个动词：`loadOlder()` 拉一页 50 条 message，而 `loadThrough(seq)`——轮次跳转加载器——按 200 条 message 一页循环拉取直到窗口覆盖目标 seq，重复调用会下调共享目标，遇到无进展的页即停止，忙碌状态复用同一个 `loadingOlder` 快照位。Web adapter 显式选择接收无 cursor 的 assistant 通知：每个 opening 携带活跃尝试的 `startedTime`、当前 chunk 和精确 v1 seq 来源。Host 会随该 baseline 捕获 follower 本地到达序号，并抑制该 cut 及之前的 buffered frame；replacement Agent 可以从 revision 一重新开始。如果 Assistant baseline 已包含一个在 opening page 之后捕获持久事件的 chunk，Client 会依据 baseline 中精确的 seq 证明，在该事件到达时直接发布。活跃 opening 之后到达的最终持久 message 会保持暂存，直到 committed end 帧携带相同的有序 source seq，且 `end.index` 等于下一个 chunk 位置。revision 或连续 index 缺口会重新打开 follow。持久缺口修复 page 不携带 Assistant baseline；Client 会清空瞬态尝试，并由 held notification 重新打开 follow，以取得配对的 page 与 baseline。普通 record 覆盖 `[event.seq, event.seq]`，packed row 覆盖 `[event.seq, event.seq + memberCount - 1]`。业务、persistence 或无法恢复的连续性错误会终止 stream，只有物理载体断开才触发自动恢复。`SessionControlStream` 是 Gateway `RemoteSnapshotStream`；每代都以完整的进程本地 baseline 开始，因此重连会替换 queue、jobs 和 projection 状态，而不会把瞬态值当作 durable event。 Session 对象还承载本地提交回显：`session.beginSubmission` 在调用方序列化与 prompt 之前，同步把一条回显写入 `SessionSnapshot.pendingSubmissions`，会话 UI 因此能在点击提交的当帧显示消息。Session 根据当前运行状态与请求的投递模式推导每条回显的 `transcript`、`queued` 或 `steering` 位置，并在序列化期间保留该位置。prompt 的 `requestId` 是关联标识：Host 把它回显为 durable user source 的 `rpcId`，queue occurrence 也把它投影为 `SessionQueuedItem.rpcId`。回显在观察到其 durable event 或 queue occurrence 后延迟一个动画帧退休，该延迟保证替代内容就绪前回显仍可渲染；带标识的 prompt 失败或被放弃时立即退休，销毁时按 failed 退休；每次退休恰好触发一次注册的 `onRetire` 回调。回显只存在于 Client 内存；刷新与重连只从 durable event 重建会话。 diff --git a/packages/api/session-controller/src/client/sessions/assistant-stream.ts b/packages/api/session-controller/src/client/sessions/assistant-stream.ts index 6b33f86293..85973799cc 100644 --- a/packages/api/session-controller/src/client/sessions/assistant-stream.ts +++ b/packages/api/session-controller/src/client/sessions/assistant-stream.ts @@ -65,14 +65,7 @@ export class ClientAssistantStream { }) } } - const visible = entries.filter((entry) => { - if (entry.type !== 'event' || entry.event.type !== 'assistant/message' - || entry.event.surfaceOp !== 'append') return true - const attempt = this.attemptForSettlement(entry.event) - if (attempt === undefined) return true - this.pendingMessages.set(positionKey(attempt.turn, attempt.step), entry) - return false - }) + const visible = entries this.publishedSeqs = new Set(visible.map(entry => entry.event.seq)) return visible } @@ -87,6 +80,7 @@ export class ClientAssistantStream { if (event.type === 'assistant/chunk') { 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) this.pendingChunks.set(event.seq, entry) return undefined } @@ -157,18 +151,6 @@ export class ClientAssistantStream { )) } - private attemptForSettlement( - event: Extract, - ): ActiveAttempt | undefined { - const sourceEventSeqs = event.sourceEventSeqs - if (sourceEventSeqs === undefined) return undefined - return [...this.attempts.values()].find(attempt => ( - attempt.turn === event.data.turn - && attempt.step === event.data.step - && sameSeqs([...attempt.legacyChunkSeqs], sourceEventSeqs) - )) - } - private publish(entry: SessionLiveEventEntry): ClientAssistantStreamResult { this.publishedSeqs.add(entry.event.seq) return { type: 'publish', entry } diff --git a/packages/api/session-controller/src/client/sessions/session.ts b/packages/api/session-controller/src/client/sessions/session.ts index e6067923fd..d8af64b0c5 100644 --- a/packages/api/session-controller/src/client/sessions/session.ts +++ b/packages/api/session-controller/src/client/sessions/session.ts @@ -647,6 +647,9 @@ export class Session implements SessionFace { projections?: ProjectionsBaseline, assistantStream?: SessionAssistantStreamBaseline, ): void { + // A durable gap-repair page has no assistant baseline. Clearing transient + // attempts makes a held notification reopen follow once for an atomic + // page/baseline pair instead of applying it to an unrelated repair cut. const visible = this.assistantStream.replace(entries, assistantStream) this.baseSeq = SessionLogOffset(visible[0]?.event.seq ?? 0) this.hasMore = hasMore diff --git a/packages/api/session-controller/tests/session-history-journal.host.spec.ts b/packages/api/session-controller/tests/session-history-journal.host.spec.ts index 80ce626688..eb22bbdc1c 100644 --- a/packages/api/session-controller/tests/session-history-journal.host.spec.ts +++ b/packages/api/session-controller/tests/session-history-journal.host.spec.ts @@ -484,6 +484,74 @@ describe('Session history raw journal', () => { } }) + it('delivers a baseline-framed chunk whose durable event lands after the opening snapshot', async () => { + const { ctx } = await harness() + const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } }) + const agent = { id: session.id, session, status: 'running', ctx } as Agent + const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() }) + const attemptId = LlmAttemptId('opening-durable-cut-attempt') + ctx.emit('agent/assistant-stream', { + agent, + frame: { + type: 'start', attemptId, revision: 1, startedTime: 100, + turn: 1, step: 1, + }, + }) + const observationCaptured = Promise.withResolvers() + const releaseObservation = Promise.withResolvers() + const originalObserve = ctx.sessionQuery.observeSession.bind(ctx.sessionQuery) + const observe = vi.spyOn(ctx.sessionQuery, 'observeSession').mockImplementation(async (sessionId, options) => { + const observation = await originalObserve(sessionId, options) + observationCaptured.resolve(undefined) + await releaseObservation.promise + return observation + }) + const abort = new AbortController() + const iterator = history.follow({ + address: { kind: 'session', sessionId: session.id }, + assistantStream: true, + }, abort.signal)[Symbol.asyncIterator]() + + try { + const opening = iterator.next() + await observationCaptured.promise + const chunk = session.append('assistant/chunk', { + turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'cut-safe' }, + }) + ctx.emit('agent/assistant-stream', { + agent, + frame: { + type: 'chunk', attemptId, revision: 2, index: 0, + chunk: chunk.data.chunk, legacyChunkSeq: chunk.seq, + }, + }) + const after = session.append('turn/start', { turn: 2 }) + releaseObservation.resolve(undefined) + + await expect(opening).resolves.toMatchObject({ + done: false, + value: { + type: 'snapshot', + records: [], + assistantStream: { + revision: 2, + attempts: [{ attemptId, legacyChunkSeqs: [chunk.seq] }], + }, + }, + }) + await expect(iterator.next()).resolves.toEqual({ + done: false, value: { type: 'event', event: chunk }, + }) + await expect(iterator.next()).resolves.toEqual({ + done: false, value: { type: 'event', event: after }, + }) + } finally { + releaseObservation.resolve(undefined) + observe.mockRestore() + await disposeFollow(ctx, iterator, abort) + } + }) + it('does not release an old-lifecycle frame after the opening baseline resets to revision one', async () => { const { ctx } = await harness() const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } }) diff --git a/packages/api/session-controller/tests/sessions-service.client.spec.ts b/packages/api/session-controller/tests/sessions-service.client.spec.ts index 2a1977ef4e..7214b33361 100644 --- a/packages/api/session-controller/tests/sessions-service.client.spec.ts +++ b/packages/api/session-controller/tests/sessions-service.client.spec.ts @@ -271,6 +271,40 @@ describe('scope tree', () => { expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.seq)).toEqual([0, 1]) }) + it('publishes a baseline-framed chunk when its durable event follows the opening cut', async () => { + const b = bench() + const attemptId = LlmAttemptId('opening-cut-attempt') + const chunk = { + type: 'event' as const, + event: { + type: 'assistant/chunk', seq: 0, time: 1, + data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'late durable' } }, + }, + } + b.api.onHistory = () => Promise.resolve(ok({ records: [], hasMore: false })) + b.api.followCursor = -1 + b.api.assistantStreamBaseline = { + revision: 2, + attempts: [{ + attemptId, startedTime: 1, turn: 1, step: 1, + chunks: [chunk.event.data.chunk], legacyChunkSeqs: [0], + }], + } + await feedList(b, [{ id: 's1' }]) + b.svc.open(sid('s1')) + const binding = b.svc.binding(sid('s1')) + if (binding === undefined) throw new Error('expected Session binding') + await vi.waitFor(() => { + expect(binding.session.getSnapshot().openState).toBe('open') + }) + + await b.api.pushFollow(sid('s1'), chunk) + await vi.waitFor(() => { + expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.seq)).toEqual([0]) + }) + expect(b.api.followStarts.filter(id => id === sid('s1'))).toHaveLength(1) + }) + it('stages a reconnect-tail assistant settlement behind its exact active attempt', async () => { const b = bench() const attemptId = LlmAttemptId('reconnect-settlement-attempt') @@ -325,7 +359,7 @@ describe('scope tree', () => { }, } b.api.onHistory = () => Promise.resolve(ok({ - records: [priorChunk, priorMessage, currentChunk, currentMessage] as never[], + records: [priorChunk, priorMessage, currentChunk] as never[], hasMore: false, })) b.api.assistantStreamBaseline = { @@ -351,6 +385,10 @@ describe('scope tree', () => { expect(binding.eventSource.getSnapshot().entries.at(1)?.event).toBe(priorMessage.event) expect(binding.eventSource.getSnapshot().entries.at(-1)?.event).toBe(currentChunk.event) + await b.api.pushFollow(sid('s1'), currentMessage) + await Promise.resolve() + expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.seq)).toEqual([0, 1, 2]) + await b.api.pushFollow(sid('s1'), { type: 'assistant-stream', frame: { @@ -366,7 +404,7 @@ describe('scope tree', () => { }) }) - it('rebaselines a reconnect settlement whose end index skips the active tail', async () => { + it('replaces an invalid settlement with the authoritative post-end baseline', async () => { const b = bench() const attemptId = LlmAttemptId('reconnect-end-index-attempt') const chunk = { @@ -394,8 +432,9 @@ describe('scope tree', () => { surfaceOp: 'append' as const, }, } + let records = [chunk] as never[] b.api.onHistory = () => Promise.resolve(ok({ - records: [chunk, message] as never[], + records, hasMore: false, })) b.api.assistantStreamBaseline = { @@ -417,6 +456,11 @@ describe('scope tree', () => { expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.seq)).toEqual([0]) }) + await b.api.pushFollow(sid('s1'), message) + await Promise.resolve() + expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.seq)).toEqual([0]) + records = [chunk, message] as never[] + b.api.assistantStreamBaseline = { revision: 3, attempts: [] } await b.api.pushFollow(sid('s1'), { type: 'assistant-stream', frame: { @@ -426,17 +470,6 @@ describe('scope tree', () => { }) await vi.waitFor(() => { expect(b.api.followStarts.filter(id => id === sid('s1'))).toHaveLength(2) - }) - expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.seq)).toEqual([0]) - - await b.api.pushFollow(sid('s1'), { - type: 'assistant-stream', - frame: { - type: 'end', attemptId, revision: 3, index: 1, - outcome: 'committed', legacyChunkSeqs: [0], - }, - }) - await vi.waitFor(() => { expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.seq)).toEqual([0, 1]) }) }) diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index 9f3ac68301..4dca7a045e 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -372,10 +372,12 @@ export class ReactLoopAgent implements Agent { step, (frame) => { this.dispatch.emit('agent/assistant-stream', { frame }) }, ) + let liveStarted = false try { const stream = preparedCall?.stream(request) ?? this.loopCtx.llm.stream(request) signal.throwIfAborted() live.start() + liveStarted = true for await (const chunk of stream) { signal.throwIfAborted() const legacyChunkSeq = this.session.append('assistant/chunk', { turn, step, chunk }).seq @@ -400,7 +402,7 @@ export class ReactLoopAgent implements Agent { }, { surfaceOp: 'append', sourceEventSeqs: chunkSeqs }) } } - live.end('aborted') + if (liveStarted) live.end('aborted') throw error } const finish = assembler.finish diff --git a/packages/core/agent-loop/tests/loop.spec.ts b/packages/core/agent-loop/tests/loop.spec.ts index fc13b64cbe..5c691c7ddc 100644 --- a/packages/core/agent-loop/tests/loop.spec.ts +++ b/packages/core/agent-loop/tests/loop.spec.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { Context } from '@deepseek-ai/cordis' import LlmRuntime, { createUserMessage, ToolCallId, LlmError, ReasoningEffortId, StreamChunk } from '@deepseek-ai/dsh-llm' import SessionStore, { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session' @@ -97,6 +97,35 @@ describe('agent loop', () => { } }) + it('does not emit an end frame when prepared dispatch throws before start', async () => { + const ctx = await harness(new MockAdapter([])) + const agent = ctx.agentLoop.create(SessionId('assistant-dispatch-throw'), { + provider: 'mock', + model: 'mock', + }) + const frames: AssistantStreamFrame[] = [] + ctx.on('agent/assistant-stream', ({ agent: subject, frame }) => { + if (subject === agent) frames.push(frame) + }) + const prepareCall = ctx.llm.prepareCall.bind(ctx.llm) + vi.spyOn(ctx.llm, 'prepareCall').mockImplementation(async (config, signal) => { + const prepared = await prepareCall(config, signal) + return { + ...prepared, + stream: () => { throw new Error('dispatch failed before start') }, + } + }) + + send(agent, 'fail before streaming') + await waitForIdle(ctx, agent) + + expect(frames).toEqual([]) + expect(agent.session.snapshotEvents().at(-1)).toMatchObject({ + type: 'turn/end', + data: { reason: { kind: 'error' } }, + }) + }) + it.each([0, -1, 1.5, Number.NaN, Number.MAX_SAFE_INTEGER + 1])( 'rejects invalid AgentOptions.maxTokens %s before publication', async (maxTokens) => {