From eb7c290b2210bb43576dee0486344a3aabd0d231 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Wed, 2 Sep 2026 04:17:44 +0800 Subject: [PATCH] fix(session): retain reachable v2 stream settlements --- ...-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 +-- apps/web/tests/scaffold-generation.spec.ts | 9 ------ .../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 | 11 ++----- .../tests/sessions-service.client.spec.ts | 30 ++++++++++++++----- 9 files changed, 35 insertions(+), 35 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 007f722536..51c64190c9 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: b91dde47259c2e455a2a2068c644a66aef136836 -2026-08-31-live-assistant-stream-frames.zh.md: a32124f24db94c04360f36baca7be29f5c062b7d +2026-08-31-live-assistant-stream-frames.md: 8b37f559b500256611302074352db76bb66888bf +2026-08-31-live-assistant-stream-frames.zh.md: e9d7d77e60b0d9c3a8db6e7b9bfcfa0c38f551a5 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 b91dde4725..8b37f559b5 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 v2 session log keeps one `assistant/message` or `assistant/attempt` settleme ## 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 frame advances one Session-local revision. The start frame captures a safe-integer wall-clock `startedTime`, chunk indexes are dense from zero, chunk timestamps are reused by the compact stream, and `end.index` equals the next chunk position. The loop appends the final `assistant/message` or `assistant/attempt` before a committed end frame names that event and seq; an abandoned end names no durable event. Authenticated Session-follow accepts an explicit Web opt-in, opens with a cached active-attempt compact 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. An opening between a durable settlement and its end frame reconstructs the active Client-only `assistant/live-chunk` updates, stages only the settlement owned by that attempt's `startedTime`, Turn, and Step, and releases it after the matching end index, type, and seq; an earlier retry at the same Turn and Step remains visible. Revision, dense-index, or settlement gaps reopen follow and replace the baseline. The TypeScript and Python SDK protocols do not expose these frames. Durable settlements remain the source of replay and model history; the [v2 stream decision](2026-09-01-v2-embedded-assistant-streams.md) owns their representation. +`dsh-agent-loop` emits scoped `agent/assistant-stream` frames for each model attempt. `start`, `chunk`, and `end` carry a branded process-local `LlmAttemptId`; every frame advances one Session-local revision. The start frame captures a safe-integer wall-clock `startedTime`, chunk indexes are dense from zero, chunk timestamps are reused by the compact stream, and `end.index` equals the next chunk position. Stream acquisition and its final cancellation check occur before `start`; a failure there emits no frame. The loop appends the final `assistant/message` or `assistant/attempt` before a committed end frame names that event and seq; an abandoned end names no durable event. Authenticated Session-follow accepts an explicit Web opt-in, opens with a cached active-attempt compact 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. A settlement arriving after an active opening is owned by that attempt only when its seq follows `startedAfterSeq` and its Turn and Step match; it remains staged until the matching end index, type, and seq arrive, while an earlier retry at the same Turn and Step remains visible. Revision, dense-index, or settlement gaps reopen follow and replace the baseline. The TypeScript and Python SDK protocols do not expose these frames. Durable settlements remain the source of replay and model history; the [v2 stream decision](2026-09-01-v2-embedded-assistant-streams.md) owns their representation. ## Alternatives considered @@ -21,4 +21,4 @@ The v2 session log keeps one `assistant/message` or `assistant/attempt` settleme ## Consequences -The Web client renders in-memory chunks before the attempt settles while retaining one durable v2 history. A process restart has no active Assistant frames; reconnect can restore only the baseline held by the current process, while cold replay expands durable settlements. 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 the attempt settles while retaining one durable v2 history. A process restart has no active Assistant frames; reconnect can restore only the baseline held by the current process, while cold replay expands durable settlements. 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 a 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 a32124f24d..e9d7d77e60 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 @@ v2 Session log 通过一个 `assistant/message` 或 `assistant/attempt` settleme ## 决定 -`dsh-agent-loop` 为每次模型 attempt 发出作用域内的 `agent/assistant-stream` frame。`start`、`chunk` 和 `end` 带有带品牌的进程本地 `LlmAttemptId`；每个 frame 都会推进一次 Session 本地 revision。start frame 会把壁钟时间捕获为安全整数 `startedTime`，chunk index 从零开始密集递增，chunk 时间戳会被紧凑 stream 复用，`end.index` 等于下一个 chunk 位置。loop 会在 committed end frame 命名事件与 seq 前追加最终 `assistant/message` 或 `assistant/attempt`；abandoned end 不命名持久事件。已认证 Session-follow 接受显式 Web opt-in，以缓存的活跃 attempt 紧凑 baseline 打开，并在一个 FIFO 中携带持久事件和无 cursor frame。每个 follower 会随 opening baseline 捕获本地到达序号，并丢弃该 cut 及之前的 buffered frame；replacement Agent 的 frame revision 可以从一重新开始，因此 revision 不定义 opening cut。opening 位于持久 settlement 与对应 end frame 之间时，会重建活跃的 Client-only `assistant/live-chunk` update，只暂存由该 attempt 的 `startedTime`、Turn 与 Step 所有的 settlement，并在匹配的 end index、type 与 seq 到达后释放；同一 Turn 和 Step 中更早的 retry 仍保持可见。revision、密集 index 或 settlement 缺口会重新打开 follow 并替换 baseline。TypeScript 和 Python SDK 协议不公开这些 frame。持久 settlement 仍是 replay 与模型历史的真源；其表示由 [v2 stream 决策](2026-09-01-v2-embedded-assistant-streams.zh.md)负责。 +`dsh-agent-loop` 为每次模型 attempt 发出作用域内的 `agent/assistant-stream` frame。`start`、`chunk` 和 `end` 带有带品牌的进程本地 `LlmAttemptId`；每个 frame 都会推进一次 Session 本地 revision。start frame 会把壁钟时间捕获为安全整数 `startedTime`，chunk index 从零开始密集递增，chunk 时间戳会被紧凑 stream 复用，`end.index` 等于下一个 chunk 位置。loop 会先取得 stream 并执行最终取消检查，再发出 `start`；这些步骤失败时不发出任何 frame。loop 会在 committed end frame 命名事件与 seq 前追加最终 `assistant/message` 或 `assistant/attempt`；abandoned end 不命名持久事件。已认证 Session-follow 接受显式 Web opt-in，以缓存的活跃 attempt 紧凑 baseline 打开，并在一个 FIFO 中携带持久事件和无 cursor frame。每个 follower 会随 opening baseline 捕获本地到达序号，并丢弃该 cut 及之前的 buffered frame；replacement Agent 的 frame revision 可以从一重新开始，因此 revision 不定义 opening cut。活跃 opening 之后到达的 settlement 只有在其 seq 晚于 `startedAfterSeq` 且 Turn 与 Step 匹配时才属于该 attempt；它会保持暂存，直到匹配的 end index、type 与 seq 到达，而同一 Turn 和 Step 中更早的 retry 仍保持可见。revision、密集 index 或 settlement 缺口会重新打开 follow 并替换 baseline。TypeScript 和 Python SDK 协议不公开这些 frame。持久 settlement 仍是 replay 与模型历史的真源；其表示由 [v2 stream 决策](2026-09-01-v2-embedded-assistant-streams.zh.md)负责。 ## 曾考虑的替代方案 @@ -21,4 +21,4 @@ v2 Session log 通过一个 `assistant/message` 或 `assistant/attempt` settleme ## 影响 -Web client 可以在 attempt settlement 前渲染内存 chunk，同时保留一份持久 v2 历史。进程重启后没有活跃 Assistant frame；重连只能恢复当前进程持有的 baseline，冷 replay 则展开持久 settlement。无 cursor 通知绝不推进 journal cursor，在持久缺口修复期间观察到的通知会等待 replacement page。frame 声明保持 agent 作用域，因此监听器只观察所属 Agent，除非它显式全局注册。 +Web client 可以在 attempt settlement 前渲染内存 chunk，同时保留一份持久 v2 历史。进程重启后没有活跃 Assistant frame；重连只能恢复当前进程持有的 baseline，冷 replay 则展开持久 settlement。无 cursor 通知绝不推进 journal cursor，在持久缺口修复期间观察到的通知会等待 replacement page。该 page 不携带 Assistant baseline，因此 Client 会清空瞬态 attempt，并让 held notification 重新打开 follow 一次，以取得配对的 page 与 baseline。frame 声明保持 agent 作用域，因此监听器只观察所属 Agent，除非它显式全局注册。 diff --git a/apps/web/tests/scaffold-generation.spec.ts b/apps/web/tests/scaffold-generation.spec.ts index 67a25f4734..0060464b54 100644 --- a/apps/web/tests/scaffold-generation.spec.ts +++ b/apps/web/tests/scaffold-generation.spec.ts @@ -200,15 +200,6 @@ describe('Web snapshot generation filenames', () => { expect(parseSeedFixture(source).header).toMatchObject({ version: 2, isSeeded: false }) }) - it('selects a committed sibling when the requested older generation is absent', async () => { - const root = await mkdtemp(join(tmpdir(), 'dsh-web-fixture-generations-')) - roots.push(root) - await writeFile(join(root, 'session.v1.jsonl'), '') - - await expect(selectedSessionFixture(join(root, 'session.jsonl'))) - .resolves.toBe(join(root, 'session.v1.jsonl')) - }) - it('records beside an older generation and preserves the parent or child role', () => { const fixtures = join('/', 'fixtures') expect(recordedSessionFixturePath(join(fixtures, 'session.jsonl'), 1)) diff --git a/packages/api/session-controller/README.i18n.yaml b/packages/api/session-controller/README.i18n.yaml index 4162cb813c..902a94026e 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: 6d79ab00cd997065a140fcc95213334987e1db13 -README.zh.md: 30ca5ff7ccbb93d2fe46943c65b5474d01663dd5 +README.md: 5771a0a8bcc7404f0cc4f5987eb97aac3c123e75 +README.zh.md: 1f67807af16433b148a69a384abae30be00e84db diff --git a/packages/api/session-controller/README.md b/packages/api/session-controller/README.md index 6d79ab00cd..5771a0a8bc 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 one `{ type: 'event', event: Se 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 frames: each opening carries the active attempt's `startedTime`, `nextIndex`, and compact stream, and every stream member becomes a Client-only `assistant/live-chunk` entry ordered between durable cursors. 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 an opening lands between a durable `assistant/message` or `assistant/attempt` and its end frame, the Session object keeps that settlement staged while leaving earlier same-step retries visible; the matching end type, seq, and index publish it. Revision, dense-index, or settlement gaps reopen follow, and an abandoned end publishes no durable settlement. Every history record covers exactly its event seq. 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 frames: each opening carries the active attempt's `startedTime`, `startedAfterSeq`, `nextIndex`, and compact stream, and every stream member becomes a Client-only `assistant/live-chunk` entry ordered between durable cursors. 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 durable `assistant/message` or `assistant/attempt` arriving after an active opening stays staged only when its seq follows `startedAfterSeq` and its Turn and Step match; the matching end type, seq, and index publish it while earlier same-step retries remain visible. Revision, dense-index, or settlement gaps reopen follow, and an abandoned end publishes no durable settlement. A durable gap-repair page has no Assistant baseline, so its held notification reopens follow once for a paired page and baseline. Every history record covers exactly its event seq. 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 30ca5ff7cc..1f67807af1 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 frame：每个 opening 携带活跃 attempt 的 `startedTime`、`nextIndex` 与紧凑 stream，每个 stream member 都成为排在持久 cursor 之间的 Client-only `assistant/live-chunk` 条目。Host 会随该 baseline 捕获 follower 本地到达序号，并抑制该 cut 及之前的 buffered frame；replacement Agent 可以从 revision 一重新开始。如果 opening 位于持久 `assistant/message` 或 `assistant/attempt` 与对应 end frame 之间，Session 对象会暂存该 settlement，同时让同一步骤中更早的 retry 保持可见；匹配的 end type、seq 与 index 到达后再发布。revision、密集 index 或 settlement 缺口会重新打开 follow，abandoned end 不发布持久 settlement。每条历史 record 只覆盖自身的 event seq。业务、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 frame：每个 opening 携带活跃 attempt 的 `startedTime`、`startedAfterSeq`、`nextIndex` 与紧凑 stream，每个 stream member 都成为排在持久 cursor 之间的 Client-only `assistant/live-chunk` 条目。Host 会随该 baseline 捕获 follower 本地到达序号，并抑制该 cut 及之前的 buffered frame；replacement Agent 可以从 revision 一重新开始。活跃 opening 之后到达的持久 `assistant/message` 或 `assistant/attempt` 只有在其 seq 晚于 `startedAfterSeq` 且 Turn 与 Step 匹配时才会保持暂存；匹配的 end type、seq 与 index 到达后再发布，而同一步骤中更早的 retry 保持可见。revision、密集 index 或 settlement 缺口会重新打开 follow，abandoned end 不发布持久 settlement。持久缺口修复 page 不携带 Assistant baseline，因此 held notification 会重新打开 follow 一次，以取得配对的 page 与 baseline。每条历史 record 只覆盖自身的 event seq。业务、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 7ff90cabaf..b94c40ecfe 100644 --- a/packages/api/session-controller/src/client/sessions/assistant-stream.ts +++ b/packages/api/session-controller/src/client/sessions/assistant-stream.ts @@ -44,7 +44,7 @@ export class ClientAssistantStream { * Replace the durable Web window and adopt an optional reconnect baseline. * @param entries - durable entries in the replacement window. * @param baseline - compact prefix for an Assistant attempt that is still live. - * @returns immediately visible durable and reconstructed transient entries, with an active settlement withheld. + * @returns immediately visible durable entries plus reconstructed transient chunks. */ replace( entries: readonly SessionEventLikeEntry[], @@ -63,14 +63,7 @@ export class ClientAssistantStream { nextIndex: opening.nextIndex, } } - const pending = opening === undefined - ? undefined - : entries.findLast(entry => entry.type === 'event' - && this.attemptForSettlement(entry.event) !== undefined) - if (pending?.type === 'event') this.pending.set(pending.event.seq, pending) - const visible: SessionEventLikeEntry[] = pending === undefined - ? [...entries] - : entries.filter(entry => entry !== pending) + const visible: SessionEventLikeEntry[] = [...entries] this.publishedSeqs = new Set(visible.map(entry => entry.event.seq)) this.durableCursor = visible.reduce((cursor, entry) => Math.max(cursor, entry.event.seq), -1) if (opening !== undefined) { 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 141f59524c..79acccd49e 100644 --- a/packages/api/session-controller/tests/sessions-service.client.spec.ts +++ b/packages/api/session-controller/tests/sessions-service.client.spec.ts @@ -251,7 +251,7 @@ describe('scope tree', () => { ))).toEqual(['a', 'b']) }) - it('stages a reconnect-tail assistant settlement behind its exact active attempt', async () => { + it('stages a post-opening assistant settlement behind its exact active attempt', async () => { const b = bench() const attemptId = LlmAttemptId('reconnect-settlement-attempt') const priorMessage = { @@ -291,7 +291,7 @@ describe('scope tree', () => { }, } b.api.onHistory = () => Promise.resolve(ok({ - records: [priorMessage, currentMessage] as never[], + records: [priorMessage] as never[], hasMore: false, })) b.api.assistantStreamBaseline = { @@ -318,6 +318,11 @@ describe('scope tree', () => { .toEqual(['assistant/message', 'assistant/live-chunk']) expect(binding.eventSource.getSnapshot().entries[0]?.event).toBe(priorMessage.event) + await b.api.pushFollow(sid('s1'), currentMessage) + await Promise.resolve() + expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type)) + .toEqual(['assistant/message', 'assistant/live-chunk']) + await b.api.pushFollow(sid('s1'), { type: 'assistant-stream', frame: { @@ -334,9 +339,13 @@ 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 prior = { + type: 'event' as const, + event: { type: 'turn/start', seq: 0, time: 19, data: { turn: 1 } }, + } const message = { type: 'event' as const, event: { @@ -355,8 +364,9 @@ describe('scope tree', () => { surfaceOp: 'append' as const, }, } + let records = [prior] as never[] b.api.onHistory = () => Promise.resolve(ok({ - records: [message] as never[], + records, hasMore: false, })) b.api.assistantStreamBaseline = { @@ -377,10 +387,16 @@ describe('scope tree', () => { if (binding === undefined) throw new Error('expected Session binding') await vi.waitFor(() => { expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type)) - .toEqual(['assistant/live-chunk']) + .toEqual(['turn/start', 'assistant/live-chunk']) }) const openingRevision = binding.eventSource.getSnapshot().revision + await b.api.pushFollow(sid('s1'), message) + await Promise.resolve() + expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type)) + .toEqual(['turn/start', 'assistant/live-chunk']) + records = [prior, message] as never[] + b.api.assistantStreamBaseline = { revision: 3 } await b.api.pushFollow(sid('s1'), { type: 'assistant-stream', frame: { @@ -392,9 +408,9 @@ describe('scope tree', () => { expect(b.api.followStarts.filter(id => id === sid('s1'))).toHaveLength(2) expect(b.api.activeFollows(sid('s1'))).toBe(1) expect(binding.eventSource.getSnapshot().revision).toBeGreaterThan(openingRevision) + expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type)) + .toEqual(['turn/start', 'assistant/message']) }) - expect(binding.eventSource.getSnapshot().entries.map(entry => entry.event.type)) - .toEqual(['assistant/live-chunk']) }) it('retains a Host-addressed scope until the first Session baseline owns pruning', async () => {