diff --git a/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.i18n.yaml b/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.i18n.yaml index 96ad0e6d2a..917bb7de47 100644 --- a/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.i18n.yaml +++ b/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.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/bug-fix/2026-08-17-durable-web-queue-recovery.md -2026-08-17-durable-web-queue-recovery.md: ffc699c7bc793da10e1f261be9ff6f7cfe079c59 -2026-08-17-durable-web-queue-recovery.zh.md: 789131c3ddfa944e43fccd71ac2ad4c39a418882 +2026-08-17-durable-web-queue-recovery.md: 360be105a9ce702d97c9254e341eeb13cd228f61 +2026-08-17-durable-web-queue-recovery.zh.md: 221ae628f02ee63a99aeb608721b6a94beffc199 diff --git a/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.md b/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.md index ffc699c7bc..360be105a9 100644 --- a/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.md +++ b/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.md @@ -18,6 +18,8 @@ The projection registry owns both durable reconstruction and live state. Its exi The generic session-projection carrier is the only Web transport. It sends higher-seq `session/projection` values, includes the complete values block on history tail pages, folds detached cold logs, and uses the projection cache when valid. There is no Host-owned `queue` projection, placement vocabulary, handoff list, dedicated queue frame, or live-Agent reconnect enumeration. +Each Host connection reset discards every retained projection value and watermark before refreshing queries, including cold Sessions absent from the process-local control baseline. Observable faces retain their identities and subscriptions. A list request from an earlier generation cannot publish values or settle the current request; history and list values from the new generation may therefore establish a lower durable seq without losing to unpersisted state. Clearing at connection reset also preserves fresh list values when the control baseline arrives later. + The client Session binding retains `inbox` in its generic per-session projection store and does not copy it into `SessionSnapshot`. QueueDock reads `next-turn` directly. ChatView reads user-origin `next-step` messages directly and ignores injected context. Claiming removes a pending value through the durable splice; a later `user/message` is rendered through the ordinary conversation projection. `session.updateQueue` resolves an ordinary cold Session through the shared Agent resolver before mutating its Inbox. A restored pending row therefore remains editable, removable, or steerable after restart, while subagent ownership keeps the same fence as other Agent operations. @@ -28,7 +30,7 @@ No new session event or on-disk format is introduced. The existing splice stream Inbox tests prove that service creation restores both lists through the registered projection, a direct durable append is immediately visible through the same live cell, a fork projects pending input inherited in its seed, and Inbox mutations reject duplicate pending identities before append. Host projection coverage reads a detached persisted Session with a pending splice, returns `values.inbox` through `session.history`, and proves that no live Agent is required. A separate cold-operation test proves `session.updateQueue` resumes the Session and appends the durable removal splice. -Client coverage pins generic Inbox projection delivery, reconnect truncation, higher-seq retention before Session materialization, and the absence of queue state from `SessionSnapshot`. UI coverage pins direct `next-turn` QueueDock rendering and user-origin `next-step` ChatView rendering. The keyless Web fixture opens a cold persisted Session and observes its pending row after restart. +Client coverage pins generic Inbox projection delivery, reconnect invalidation for omitted cold Sessions, both baseline arrival orders, obsolete list request outcomes, higher-seq retention before Session materialization, and the absence of queue state from `SessionSnapshot`. UI coverage pins direct `next-turn` QueueDock rendering and user-origin `next-step` ChatView rendering. The keyless Web fixture opens a cold persisted Session and observes its pending row after restart. ## Alternatives considered diff --git a/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.zh.md b/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.zh.md index 789131c3dd..221ae628f0 100644 --- a/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.zh.md +++ b/.agents/notes/implemented/bug-fix/2026-08-17-durable-web-queue-recovery.zh.md @@ -18,6 +18,8 @@ Inbox 接受消息时会记录规范化的 `agent/inbox/spliced` 事件，但 We 通用会话投影传输层是唯一 Web 传输。它发送 seq 更高的 `session/projection` 值，在历史尾页中包含完整 values 块，折叠已分离的冷日志，并在缓存有效时使用投影缓存。系统不存在 Host 拥有的 `queue` 投影、placement 词汇、handoff 列表、专用 queue 帧或枚举 live Agent 的重连逻辑。 +每次 Host 连接重置都会先丢弃所有保留的投影值及其水位，再刷新查询，其中也包括进程本地 control baseline 中没有列出的冷 Session。Observable face 保留自身标识及订阅。较早 generation 的 list 请求不能发布值或使当前请求结束，因此新 generation 的历史与 list 值可以建立较低的持久 seq，而不会被尚未持久化的状态挡住。在连接重置时清理，也能在 control baseline 较晚到达时保留新 list 值。 + 客户端 Session binding 在通用逐会话投影存储中保留 `inbox`，不会把它复制进 `SessionSnapshot`。QueueDock 直接读取 `next-turn`。ChatView 直接读取用户来源的 `next-step` 消息，并忽略注入上下文。认领操作通过持久 splice 移除待处理值；后续 `user/message` 由普通会话投影渲染。 `session.updateQueue` 在修改 Inbox 前通过共享 Agent 解析器解析普通冷 Session。因此，恢复出的待处理行在重启后仍可编辑、移除或 steering，而 subagent ownership 保持与其他 Agent 操作相同的 fence。 @@ -28,7 +30,7 @@ Inbox 接受消息时会记录规范化的 `agent/inbox/spliced` 事件，但 We Inbox 测试证明服务创建会通过已注册投影恢复两份列表、直接追加的持久事件会立即通过同一 live cell 可见、fork 会投影其 seed 中继承的待处理输入，并且 Inbox 变更会在 append 前拒绝重复的待处理标识。Host 投影覆盖会读取包含待处理 splice 的已分离持久 Session，经 `session.history` 返回 `values.inbox`，并证明不需要 live Agent。另一项冷操作测试证明 `session.updateQueue` 会恢复 Session 并追加持久删除 splice。 -客户端覆盖固定通用 Inbox 投影投递、重连截断、Session 实例化前保留 seq 更高的值，以及 `SessionSnapshot` 不含 queue 状态。UI 覆盖固定 QueueDock 直接渲染 `next-turn`，以及 ChatView 渲染用户来源的 `next-step`。无密钥 Web fixture 会打开一份冷持久 Session，并在重启后观察其待处理行。 +客户端覆盖固定通用 Inbox 投影投递、重连时清理遗漏冷 Session 的旧值、基线的两种到达顺序、过期 list 请求的结果、Session 实例化前保留 seq 更高的值，以及 `SessionSnapshot` 不含 queue 状态。UI 覆盖固定 QueueDock 直接渲染 `next-turn`，以及 ChatView 渲染用户来源的 `next-step`。无密钥 Web fixture 会打开一份冷持久 Session，并在重启后观察其待处理行。 ## 考虑过的替代方案 diff --git a/packages/api/session-controller/README.i18n.yaml b/packages/api/session-controller/README.i18n.yaml index 398d37f80e..6125d4c3fa 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: dc1b2a4cf8c322e9cfd9981fd4adc29d1dc544d5 -README.zh.md: 118786581b2611f2dae19cdbb6b07dee91df48d0 +README.md: 35b00005d15086b5b9f2ba07d6af028ea6a248b7 +README.zh.md: 30013b27e03e6a65a58c5b51faf36ab7312084f4 diff --git a/packages/api/session-controller/README.md b/packages/api/session-controller/README.md index dc1b2a4cf8..35b00005d1 100644 --- a/packages/api/session-controller/README.md +++ b/packages/api/session-controller/README.md @@ -30,7 +30,7 @@ The Client journal validates exact V3 event envelopes before publishing follow s Each endpoint states its activation policy. List reads only stored headers and projection-cache rows: it never calls per-session stat or opens a cold Session body. A current-format cache identity may supply every list hint; a lifecycle-matching predecessor cache may supply only its version-compatible title as a stale display fact, never as an authoritative fold seed. 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. Cancellation requires live state; queue mutation, model, rename, prompt, and file-reference operations may resolve or resume an ordinary Session. Prompt rejects content with neither non-whitespace text nor an attachment before resolving the Agent or appending Session events; queue edits accept only non-empty text content. Prompt admission consumes opaque receipts from the injected [`fileUploads`](../../client/file-upload/README.md) Host service and resolves every same-Agent receipt before sending the complete ordered content list through `ctx.attachments`. Prompt retries whose `requestId` is already queued or logged return the original acceptance without inserting another message. Create and fork are the only operations that create a new Agent directly. The service applies one preset-aware resume policy and subagent ownership fence to its own methods and to the Typert Agent and Session lookups used by other Remote namespaces. Queue mutation has one narrow exception: a live child whose current projected identity is continuable and comes from its own non-seed suffix accepts the ordinary Edit, Remove, and QueueDock Steer actions across both inbox destinations. One-shot, missing, unknown, corrupt, seed-only, or cold children remain rejected without resume. The skill catalog 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`, `append`, and `settle-assistant` 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 `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 publishes one named settlement delta that retires the attempt's transient rows and adds the durable entry while earlier same-step retries remain visible. Revision, dense-index, or settlement gaps for a known attempt reopen follow, while a controller that missed the start ignores unknown-attempt frames and publishes their durable settlement normally. An abandoned end publishes a settlement delta without a durable entry so its transient rows retire immediately. 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 jobs and projection state instead of treating transient values as durable events. The durable `inbox` projection carries both pending lists through the same cold-read and reconnect path as other projections. Client Agent contexts provide the identity used by the independent [`fileUpload`](../../client/file-upload/README.md) service; Session objects expose lifecycle, prompt, queue, and history operations rather than file transfer. +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`, `append`, and `settle-assistant` 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 `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 publishes one named settlement delta that retires the attempt's transient rows and adds the durable entry while earlier same-step retries remain visible. Revision, dense-index, or settlement gaps for a known attempt reopen follow, while a controller that missed the start ignores unknown-attempt frames and publishes their durable settlement normally. An abandoned end publishes a settlement delta without a durable entry so its transient rows retire immediately. 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 jobs and projection state instead of treating transient values as durable events. At each Host connection reset, the Client clears retained projection values and watermarks before refreshing queries, including Sessions absent from the control baseline. Outstanding list responses from the previous generation cannot republish those values. The durable `inbox` projection carries both pending lists through the same cold-read and reconnect path as other projections. Client Agent contexts provide the identity used by the independent [`fileUpload`](../../client/file-upload/README.md) service; Session objects expose lifecycle, prompt, queue, and history operations rather than file transfer. 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. The echo stores ordered image previews and durable file references. Session derives its `transcript`, `queued`, or `steering` placement from the current running state and 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`, including pending messages in the `inbox` projection. An echo retires one animation frame after its durable event or queue occurrence is observed, immediately when its identified prompt fails or is abandoned, and as failed on disposal. Each retirement fires `onRetire` exactly once; an observed retirement includes the ordered durable attachment references so the composer can release successful cards while preserving failed drafts. 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 118786581b..30013b27e0 100644 --- a/packages/api/session-controller/README.zh.md +++ b/packages/api/session-controller/README.zh.md @@ -30,7 +30,7 @@ Client journal 在发布 follow snapshot、live entry 或历史页之前验证 每个 endpoint 都声明自己的激活策略。列表只读取持久化 header 与 projection cache row，绝不调用逐 Session stat 或打开冷 Session body。当前格式 cache identity 可以提供全部列表 hint；生命周期匹配的 predecessor cache 只能提供版本兼容的 title，作为可能过时的展示事实，绝不能作为权威 fold seed。搜索、附件、历史页、日志跟随、skill 发现和工作区路径打开可以在不激活 Agent 的情况下检查 persistence；`canOpenWorkspacePath()` 无需指定 Session 即可报告原生打开能力。取消要求 live 状态；queue 变更、模型、重命名、prompt 和文件引用操作可以解析或恢复普通 Session。prompt 会在解析 Agent 或追加 Session event 前，拒绝既没有非空白文本也没有附件的 content；queue edit 只接受非空文本 content。prompt 准入从注入的 [`fileUploads`](../../client/file-upload/README.zh.md) Host 服务取得不透明凭证，在把完整有序内容列表交给 `ctx.attachments` 前解析每个属于同一 Agent 的凭证。`requestId` 已进入 queue 或日志时，prompt 重试直接返回原来的接受结果，不会重复插入消息。只有 create 与 fork 会直接创建新 Agent。该服务把同一套感知 preset 的恢复策略和 subagent ownership fence 同时用于自身方法，以及其他 Remote namespace 使用的 Typert Agent 与 Session lookup。Queue 变更只有一个狭窄例外：当前 projection identity 为 continuable 且来自自身非 seed suffix 的在线 child，可以在两个 inbox 目标上使用普通 Edit、Remove 与 QueueDock Steer action。One-shot、缺失、未知、损坏、仅含 seed identity 或冷 child 继续被拒绝，且不会恢复。skill 目录优先使用已有 live Agent，否则使用所记录 preset 的常驻 scope，因此列表查询绝不会启动 Agent。 -Client adapter 提供 `SessionEventStream`，即绑定到一个普通 Session 或 direct subagent address 的 Gateway `RemoteJournalStream`。它在读取首个 page 前打开 follow，只发布连续的 `replace`、`prepend`、`append` 与 `settle-assistant` 变更，并通过 tail page 修复重连或 seq 缺口。向后分页有两个动词：`loadOlder()` 拉一页 50 条 message，而 `loadThrough(seq)`——轮次跳转加载器——按 200 条 message 一页循环拉取直到窗口覆盖目标 seq，重复调用会下调共享目标，遇到无进展的页即停止，忙碌状态复用同一个 `loadingOlder` 快照位。Web adapter 显式选择接收无 cursor 的 Assistant frame：每个 opening 携带活跃 attempt 的 `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 会发布一个具名 settlement delta，删除该 attempt 的瞬态 row、加入持久条目，并保留同一步骤中更早的 retry。已知 attempt 的 revision、密集 index 或 settlement 缺口会重新打开 follow；若 controller 错过 start，则忽略 unknown-attempt frame，并正常发布其持久 settlement。Abandoned end 会发布不含持久条目的 settlement delta，使瞬态 row 立即退出。持久缺口修复 page 不携带 Assistant baseline，因此 held notification 会重新打开 follow 一次，以取得配对的 page 与 baseline。每条历史 record 只覆盖自身的 event seq。业务、persistence 或无法恢复的连续性错误会终止 stream，只有物理载体断开才触发自动恢复。`SessionControlStream` 是 Gateway `RemoteSnapshotStream`；每代都以完整的进程本地 baseline 开始，因此重连会替换 jobs 和 projection 状态，而不会把瞬态值当作 durable event。持久 `inbox` 投影通过与其他投影相同的冷读取和重连路径传输两份待处理列表。Client Agent context 提供独立 [`fileUpload`](../../client/file-upload/README.zh.md) 服务使用的身份；Session 对象提供生命周期、prompt、queue 与历史操作，不提供文件传输。 +Client adapter 提供 `SessionEventStream`，即绑定到一个普通 Session 或 direct subagent address 的 Gateway `RemoteJournalStream`。它在读取首个 page 前打开 follow，只发布连续的 `replace`、`prepend`、`append` 与 `settle-assistant` 变更，并通过 tail page 修复重连或 seq 缺口。向后分页有两个动词：`loadOlder()` 拉一页 50 条 message，而 `loadThrough(seq)`——轮次跳转加载器——按 200 条 message 一页循环拉取直到窗口覆盖目标 seq，重复调用会下调共享目标，遇到无进展的页即停止，忙碌状态复用同一个 `loadingOlder` 快照位。Web adapter 显式选择接收无 cursor 的 Assistant frame：每个 opening 携带活跃 attempt 的 `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 会发布一个具名 settlement delta，删除该 attempt 的瞬态 row、加入持久条目，并保留同一步骤中更早的 retry。已知 attempt 的 revision、密集 index 或 settlement 缺口会重新打开 follow；若 controller 错过 start，则忽略 unknown-attempt frame，并正常发布其持久 settlement。Abandoned end 会发布不含持久条目的 settlement delta，使瞬态 row 立即退出。持久缺口修复 page 不携带 Assistant baseline，因此 held notification 会重新打开 follow 一次，以取得配对的 page 与 baseline。每条历史 record 只覆盖自身的 event seq。业务、persistence 或无法恢复的连续性错误会终止 stream，只有物理载体断开才触发自动恢复。`SessionControlStream` 是 Gateway `RemoteSnapshotStream`；每代都以完整的进程本地 baseline 开始，因此重连会替换 jobs 和 projection 状态，而不会把瞬态值当作 durable event。每次 Host 连接重置时，Client 会先清除保留的投影值及其水位，再刷新查询，其中也包括 control baseline 中没有列出的 Session。上一代尚未完成的 list 响应无法重新发布这些值。持久 `inbox` 投影通过与其他投影相同的冷读取和重连路径传输两份待处理列表。Client Agent context 提供独立 [`fileUpload`](../../client/file-upload/README.zh.md) 服务使用的身份；Session 对象提供生命周期、prompt、queue 与历史操作，不提供文件传输。 Session 对象还承载本地提交回显：`session.beginSubmission` 在调用方序列化与 prompt 之前，同步把一条回显写入 `SessionSnapshot.pendingSubmissions`，会话 UI 因此能在点击提交的当帧显示消息。回显按顺序存放图片预览与持久文件引用。Session 根据当前运行状态与请求的投递模式推导其 `transcript`、`queued` 或 `steering` 位置，并在序列化期间保留该位置。prompt 的 `requestId` 是关联标识：Host 把它回显为 durable user source 的 `rpcId`，`inbox` 投影中的待处理消息也保留同一 source。回显在观察到其 durable event 或 queue occurrence 后延迟一个动画帧退休，带标识的 prompt 失败或被放弃时立即退休，销毁时按 failed 退休。每次退休恰好触发一次 `onRetire`；observed 退休还会携带有序的持久附件引用，让 composer 释放成功卡片并保留失败草稿。回显只存在于 Client 内存；刷新与重连只从 durable event 重建会话。 diff --git a/packages/api/session-controller/src/client/sessions/manager.ts b/packages/api/session-controller/src/client/sessions/manager.ts index 1ef550fb08..e87c4b22c7 100644 --- a/packages/api/session-controller/src/client/sessions/manager.ts +++ b/packages/api/session-controller/src/client/sessions/manager.ts @@ -114,7 +114,7 @@ export class SessionManager { private listPhase: SessionListPhase = 'pending' private listError: RemoteFailure | null = null private listInflight: Promise | null = null - /** Mutations arriving after a list request starts are replayed over its response. */ + /** Active list request's mutation log; its identity also fences completion after reconnect. */ private listMutations: SessionListMutation[] | null = null private readonly addresses = new Map() private readonly catalogs = new Map() @@ -441,7 +441,7 @@ export class SessionManager { // ---- List API ---- - /** Full refresh via session.list (single-flight: an in-flight call is reused). */ + /** Full refresh via session.list (single-flight within one Host generation). */ refreshList(): Promise { if (this.listInflight !== null) return this.listInflight this.listState = 'loading' @@ -453,6 +453,7 @@ export class SessionManager { this.listInflight = (async () => { try { const result = await this.remote.session.list({}) + if (this.listMutations !== mutations) return if (result.ok) { const baseline: SessionSummary[] = this.listPhase === 'pending' ? [...result.value.items] @@ -502,12 +503,15 @@ export class SessionManager { } } catch (error) { if (!isRemoteFailure(error)) throw error + if (this.listMutations !== mutations) return this.listState = 'error' this.listError = error } finally { - this.listMutations = null - this.listInflight = null - this.notifier.markDirty() + if (this.listMutations === mutations) { + this.listMutations = null + this.listInflight = null + this.notifier.markDirty() + } } })() return this.listInflight @@ -766,9 +770,14 @@ export class SessionManager { /** * Repair one re-established Host-event generation with queryable baselines. + * Discard old projection cuts before new queries, including cold Sessions + * absent from the process-local control baseline. * Opened Session follow streams resume independently through API Gateway. */ handleConnected(): void { + for (const store of this.projectionStores.values()) store.clear() + this.listMutations = null + this.listInflight = null void this.refreshList() const selectedAddress = this.selected === undefined ? undefined : this.addresses.get(this.selected) if (selectedAddress !== undefined) void this.refreshSubagents(selectedAddress.parentSessionId) diff --git a/packages/api/session-controller/src/client/sessions/projection-store.ts b/packages/api/session-controller/src/client/sessions/projection-store.ts index 79ad1b0e62..1dec337f5d 100644 --- a/packages/api/session-controller/src/client/sessions/projection-store.ts +++ b/packages/api/session-controller/src/client/sessions/projection-store.ts @@ -67,9 +67,9 @@ interface Channel { /** * One session's projection values. Framework semantics, uniform across every * key: a baseline seeds rows at its cut, a push frame updates one row, and in - * both paths a lower-or-equal seq loses — a replayed frame cannot regress a - * value, a stale baseline cannot overwrite a newer frame. A key the store has - * never seen reads `undefined` (capability absent). Faces are identity-stable + * both paths a lower-or-equal seq within the Host generation loses. A replayed + * frame cannot regress a value; a stale baseline cannot overwrite a newer + * frame. A key the store has never seen reads `undefined` (capability absent). Faces are identity-stable * per key (create-on-demand, cached) so the React side binds each exactly * once; the store-level channel (`subscribeAny`) serves coarse consumers (the * manager's list projection reads the `title` key). @@ -174,6 +174,14 @@ export class ProjectionValueStore { } } + /** Discard one Host generation's values and watermarks while preserving subscribed faces. */ + clear(): void { + for (const key of this.rows.keys()) { + this.rows.delete(key) + this.changed(key) + } + } + private changed(key: string): void { this.valuesCache = undefined this.channels.get(key)?.notifier.markDirty() diff --git a/packages/api/session-controller/tests/inbox-projection.client.spec.ts b/packages/api/session-controller/tests/inbox-projection.client.spec.ts index 6e008f0f07..0d20a6278c 100644 --- a/packages/api/session-controller/tests/inbox-projection.client.spec.ts +++ b/packages/api/session-controller/tests/inbox-projection.client.spec.ts @@ -7,7 +7,7 @@ import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' import type { SessionControlFrame } from '@deepseek-ai/dsh-api-session-controller/types' import type { SessionId } from '@deepseek-ai/dsh-session/types' import { SessionManager } from '../src/client/sessions/manager.ts' -import { FakeApiClient, fakeRemote } from './fake-api.client.ts' +import { deferred, FakeApiClient, fakeRemote, ok } from './fake-api.client.ts' const SID = 'fk-q1' as SessionId const text = (value: string): ContentBlock[] => [{ type: 'text', text: value }] @@ -63,6 +63,50 @@ describe('Inbox projection intake', () => { expect(manager.get(SID).projections.faceOf('inbox').getSnapshot()).toBeUndefined() }) + it.each(['control-first', 'list-first'] as const)( + 'replaces cold Session Inbox values across Host generations (%s)', + async (order) => { + const api = new FakeApiClient() + const list = deferred>>() + api.onList = () => list.promise + const manager = new SessionManager(fakeRemote(api)) + const hiddenSessionId = 'cold-hidden-inbox' as SessionId + const ghost = message('ghost', 'acceptance was not persisted') + const pending = message('pending', 'claim was not persisted') + const empty = { 'next-turn': [], 'next-step': [] } + const restored = { 'next-turn': [pending], 'next-step': [] } + manager.handleControlFrame({ ...inboxFrame({ ...empty, 'next-turn': [ghost] }), seq: 20 }) + manager.handleControlFrame({ ...inboxFrame(empty), sessionId: hiddenSessionId, seq: 20 }) + const face = manager.get(SID).projections.faceOf('inbox') + const baseline = { type: 'baseline', value: { jobs: {}, projections: {} } } as const + const result = ok({ items: [ + { sessionId: SID, updatedAt: 1, running: false, blank: false, + projections: { asOfSeq: 1, values: { inbox: empty } } }, + { sessionId: hiddenSessionId, updatedAt: 1, running: false, blank: false, + projections: { asOfSeq: 1, values: { inbox: restored } } }, + ] }) as Awaited> + let refreshed: Promise | undefined + + try { + manager.handleConnected() + refreshed = manager.refreshList() + expect(face.getSnapshot()).toBeUndefined() + if (order === 'control-first') manager.handleControlFrame(baseline) + list.resolve(result) + await refreshed + if (order === 'list-first') manager.handleControlFrame(baseline) + + expect(manager.get(SID).projections.faceOf('inbox')).toBe(face) + expect(face.getSnapshot()).toEqual(empty) + expect(manager.get(hiddenSessionId).projections.faceOf('inbox').getSnapshot()).toEqual(restored) + } finally { + list.resolve(result) + await refreshed + await manager.dispose() + } + }, + ) + it('retains only the highest-seq value received before Session materialization', () => { const manager = new SessionManager(fakeRemote(new FakeApiClient())) manager.handleControlFrame(inboxFrame({ diff --git a/packages/api/session-controller/tests/manager.client.spec.ts b/packages/api/session-controller/tests/manager.client.spec.ts index 0435873278..3b3de8e4e6 100644 --- a/packages/api/session-controller/tests/manager.client.spec.ts +++ b/packages/api/session-controller/tests/manager.client.spec.ts @@ -751,6 +751,76 @@ describe('remaining branches', () => { }) describe('connected generation', () => { + it.each(['old-first', 'new-first'] as const)( + 'ignores a previous generation list response (%s)', + async (order) => { + const api = new FakeApiClient() + const oldList = deferred>>() + const newList = deferred>>() + let calls = 0 + api.onList = () => calls++ === 0 ? oldList.promise : newList.promise + const manager = new SessionManager(fakeRemote(api)) + const oldResult = ok({ items: [{ ...summary(S1), projections: { + asOfSeq: 20, values: { title: 'Unpersisted title' }, + } }] as never[] }) + const newResult = ok({ items: [{ ...summary(S1), projections: { + asOfSeq: 1, values: { title: 'Durable title' }, + } }] as never[] }) + const oldPull = manager.refreshList() + let newPull: Promise | undefined + try { + manager.handleConnected() + newPull = manager.refreshList() + expect(api.callsOf('session.list')).toHaveLength(2) + if (order === 'old-first') { + oldList.resolve(oldResult) + await oldPull + expect(manager.getListSnapshot().state).toBe('loading') + expect(manager.refreshList()).toBe(newPull) + } + newList.resolve(newResult) + await newPull + oldList.resolve(oldResult) + await oldPull + + expect(manager.getListSnapshot()).toMatchObject({ state: 'idle', error: null }) + expect(manager.getListSnapshot().items[0]?.title).toBe('Durable title') + } finally { + oldList.resolve(oldResult) + newList.resolve(newResult) + await Promise.all([oldPull, newPull]) + await manager.dispose() + } + }, + ) + + it('ignores a previous generation request failure while the new list is loading', async () => { + const api = new FakeApiClient() + const oldList = deferred>>() + const newList = deferred>>() + let calls = 0 + api.onList = () => calls++ === 0 ? oldList.promise : newList.promise + const manager = new SessionManager(fakeRemote(api)) + const oldPull = manager.refreshList() + let newPull: Promise | undefined + try { + manager.handleConnected() + newPull = manager.refreshList() + oldList.reject(new RemoteError('gateway/internal', 'old Host disconnected', {})) + await oldPull + expect(manager.getListSnapshot()).toMatchObject({ state: 'loading', error: null }) + expect(manager.refreshList()).toBe(newPull) + newList.resolve(ok({ items: [summary(S1)] as never[] })) + await newPull + expect(manager.getListSnapshot()).toMatchObject({ state: 'idle', error: null }) + } finally { + oldList.resolve(ok({ items: [] })) + newList.resolve(ok({ items: [] })) + await Promise.all([oldPull, newPull]) + await manager.dispose() + } + }) + it('refreshes query baselines without rebuilding independently resumed Session sources', async () => { const api = new FakeApiClient() api.onHistory = () => Promise.resolve(ok({ diff --git a/packages/api/session-controller/tests/projection-store.client.spec.ts b/packages/api/session-controller/tests/projection-store.client.spec.ts index 274f7b095c..4fe9d05fd3 100644 --- a/packages/api/session-controller/tests/projection-store.client.spec.ts +++ b/packages/api/session-controller/tests/projection-store.client.spec.ts @@ -68,6 +68,32 @@ describe('Session projection value semantics', () => { expect(store.get('other')).toBeUndefined() }) + it('clears all generation watermarks without replacing subscribed faces', async () => { + const store = new ProjectionValueStore() + const face = store.faceOf('test/marks') + const observed: unknown[] = [] + const unsubscribe = face.subscribe(() => { observed.push(face.getSnapshot()) }) + try { + store.apply('test/marks', { marks: ['lost-tail'] }, SessionSeq(20)) + store.apply('empty-session', 'old generation', -1) + const previous = store.values() + await Promise.resolve() + store.clear() + await Promise.resolve() + + expect(face.getSnapshot()).toBeUndefined() + expect(store.get('empty-session')).toBeUndefined() + expect(store.faceOf('test/marks')).toBe(face) + expect(store.values()).toEqual({}) + expect(store.values()).not.toBe(previous) + store.seed({ asOfSeq: SessionSeq(1), values: { 'test/marks': { marks: ['durable'] } } }) + await Promise.resolve() + expect(observed).toEqual([{ marks: ['lost-tail'] }, undefined, { marks: ['durable'] }]) + } finally { + unsubscribe() + } + }) + it('notifies the key face on change (batched) and not on dropped applications', async () => { const store = new ProjectionValueStore() let keyTicks = 0