From 945006c984e98991f16b01acaa38a043811f9ab9 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Fri, 30 Oct 2026 14:14:03 +0800 Subject: [PATCH] perf(session): send one projection frame per committed event (#7587) * perf(session): send one projection frame per committed event * fix(session): deliver keys changed before a drive failure; address review * perf(session): carry projection frame changes as key and val entries --- ...ession-projection-mandatory-seam.i18n.yaml | 12 +- ...08-19-session-projection-mandatory-seam.md | 7 + ...19-session-projection-mandatory-seam.zh.md | 7 + ...nd-projection-owned-client-state.i18n.yaml | 4 +- ...tions-and-projection-owned-client-state.md | 2 +- ...ns-and-projection-owned-client-state.zh.md | 2 +- ...sion-history-stream-and-snapshot.i18n.yaml | 4 +- ...-21-session-history-stream-and-snapshot.md | 2 +- ...-session-history-stream-and-snapshot.zh.md | 2 +- apps/web/tests/idle-submission-handoff.e2e.ts | 6 +- apps/web/tests/steering.e2e.ts | 4 +- docs/event-producer-consumer.md | 10 +- docs/event-producer-consumer.zh.md | 10 +- docs/subsystems/session-projection.i18n.yaml | 4 +- docs/subsystems/session-projection.md | 29 +++- docs/subsystems/session-projection.zh.md | 29 +++- docs/subsystems/session.md | 4 +- docs/subsystems/session.zh.md | 4 +- .../projection-event-frames/guide.i18n.yaml | 16 ++ .../projection-event-frames/guide.md | 32 ++++ .../projection-event-frames/guide.zh.md | 32 ++++ .../src/client/sessions/manager.ts | 8 +- .../api/session-controller/src/control.ts | 8 +- .../src/history/controller.ts | 4 +- .../src/history/listener.ts | 15 +- packages/api/session-controller/src/index.ts | 4 +- packages/api/session-controller/src/types.ts | 19 ++- .../tests/control-queue.host.spec.ts | 26 ++-- .../tests/history-controller.host.spec.ts | 13 +- .../tests/inbox-projection.client.spec.ts | 3 +- .../tests/listeners.host.spec.ts | 17 ++- .../tests/manager.client.spec.ts | 90 ++++++++---- .../tests/projection-store.client.spec.ts | 18 ++- .../tests/remote-access.host.spec.ts | 2 +- .../tests/session-projections.host.spec.ts | 70 ++++++--- .../tests/sessions-service.client.spec.ts | 10 +- .../tests/subagent-catalog.host.spec.ts | 11 +- .../session-controller/tests/test-remote.ts | 3 + .../tests/transport.client.spec.ts | 2 +- .../liveshare/src/host/viewer-fork.ts | 2 +- .../liveshare/tests/viewer-authority.spec.ts | 4 +- .../liveshare/tests/viewer-fixture.client.ts | 102 ++++--------- .../extensions/tool-cordis/src/api-catalog.ts | 16 +- .../session-projection/README.i18n.yaml | 8 +- packages/session/session-projection/README.md | 4 +- .../session/session-projection/README.zh.md | 4 +- .../session/session-projection/src/index.ts | 137 ++++++++++++------ .../session-projection/tests/registry.spec.ts | 83 +++++++++++ scripts/gen-cordis-catalog.ts | 1 + scripts/no-unknown-casts.baseline.json | 3 +- scripts/type-equiv.manifest.json | 5 + 51 files changed, 631 insertions(+), 283 deletions(-) create mode 100644 docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.i18n.yaml create mode 100644 docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.md create mode 100644 docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.zh.md diff --git a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.i18n.yaml b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.i18n.yaml index 30244c0085..fb1ee4e295 100644 --- a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.i18n.yaml @@ -9,11 +9,11 @@ en: fca00eeb7a8d9abd zh: d7a9766f5322dff9 /agent-note-session-projections-as-a-required-reader-seam/decision: - en: 8d38ed9356085c7e - zh: e15ede51415192f3 + en: ed3ee779d31afdba + zh: 2ffad21003be59c5 /agent-note-session-projections-as-a-required-reader-seam/alternatives-considered: - en: aea3da939c90d240 - zh: 21e91c481cdd23b7 + en: 778e12e73a9a6900 + zh: 5d4f313e4ea87372 /agent-note-session-projections-as-a-required-reader-seam/consequences: - en: ab45eb87f7c22cdb - zh: e65a97991a45164f + en: b1935fb4cc8d9cc6 + zh: 0611d7033f5702a3 diff --git a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.md b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.md index 091e8b277a..3256c6dbb0 100644 --- a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.md +++ b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.md @@ -18,6 +18,8 @@ Owners acquire state through [awaited registration and retained handles](2026-10 `onChanged` publishes client-visible value changes only. Registration changes do not create a second Host event stream or client tombstone protocol. A later authoritative history or list baseline reflects the active key set. +`onEventChanged` groups the changes of one committed event. After every unit has been driven, it receives one `{ key, val }` entry per client-visible key whose view changed, in registration order, where `val` is the key's view. If a unit or an `onChanged` listener throws, it still receives the entries collected before the throw. Remote projection frames follow this grouping: `session/control` and `session/observeProjections` emit one frame per committed event with a `changes` array of the same entries, so a Client applies one event's changes and publishes its Session list once instead of once per key. Each entry names its key, so a key changed to `null` stays distinct from an unchanged key and no consumer pairs values with keys by position. `onChanged` remains for Host consumers that react to one key. + ## Alternatives considered - **Default missing projected state.** This preserves more partial compositions but makes missing host state indistinguishable from a valid empty value. Rejected because official profiles mount the registry and configuration errors must fail explicitly. @@ -25,9 +27,14 @@ Owners acquire state through [awaited registration and retained handles](2026-10 - **Use `snapshot()` for every read.** This keeps one method but computes unrelated wire views and encourages consumers to depend on batch transport data for host logic. Rejected in favor of typed single-key state reads. - **Send full host values to clients.** This avoids separate view types but exposes producer identities and policy knobs that no client consumes. Rejected in favor of explicit cropped views. - **Broadcast registry additions and removals across Host and mux streams.** The streams have no shared ordering, so clients need tombstones, buffered frames, and baseline retries to reconcile them. Rejected because plugin-key churn does not justify a second synchronization protocol. +- **Coalesce per-key frames in the controller or Client.** Microtask batching or grouping frames by sequence number needs buffering and timing assumptions, while the registry already knows where each event's drive ends. Rejected in favor of grouping at the drive. +- **Carry frame changes as a key-to-value object.** Consumers would infer the changed keys from object members and lose the registration order. Rejected in favor of an ordered array of `{ key, val }` entries. +- **Carry parallel `keys` and `values` arrays.** This avoids repeating the entry field names, but every consumer must keep the two arrays aligned by index, and the wire types cannot express their equal length. Rejected in favor of self-contained entries. +- **Replace `onChanged` with the event-level feed.** Host consumers such as forge react to one key and gain nothing from grouping. Rejected; both feeds remain. ## Consequences - Missing projection composition fails during activation; it never degrades to a default value. - Host consumers avoid repeated whole-registry snapshots and log scans. - Wire payloads exclude host-only fields and per-key watermark wrappers; ordinary baselines communicate the active key set. +- One committed event produces at most one Remote projection frame per Session, whose `seq` is that event's sequence number. diff --git a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.zh.md b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.zh.md index c30f7bd35d..26644a05d9 100644 --- a/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.zh.md +++ b/.agents/notes/implemented/architecture/2026-08-19-session-projection-mandatory-seam.zh.md @@ -18,6 +18,8 @@ Status: implemented `onChanged` 只发布客户端可见值的变化。注册变化不会创建第二条 Host 事件流或客户端 tombstone 协议。后续权威 history 或 list 基线会反映活跃 key 集。 +`onEventChanged` 把一个已提交事件的变化归为一组。所有单元驱动完成后，它按注册顺序收到视图发生变化的每个客户端可见 key 一项 `{ key, val }`，`val` 是该 key 的视图。某个单元或 `onChanged` 监听器抛错时，它仍会收到抛错前已收集的各项。Remote projection 帧沿用这一分组：`session/control` 与 `session/observeProjections` 每个已提交事件发出一帧，携带由同样各项组成的 `changes` 数组，因此 Client 一次应用一个事件的变化，Session 列表只发布一次，而不是每个 key 一次。每项都写明自己的 key，因此改为 `null` 的 key 与未变化的 key 可以区分，消费方也不必按位置把值和 key 配对。`onChanged` 保留给只关心单个 key 的 Host 消费方。 + ## 考虑过的替代方案 - **为缺失的投影状态提供默认值。** 这会保留更多不完整组合，但缺失 host 状态将无法与合法空值区分。正式 profile 已挂载注册表，配置错误必须显式失败，因此否决。 @@ -25,9 +27,14 @@ Status: implemented - **每次读取都使用 `snapshot()`。** 这只保留一个方法，但会计算无关 wire view，并鼓励消费方让 host 逻辑依赖批量传输数据。改用类型化单 key 状态读取。 - **向客户端发送完整 host 值。** 这避免单独的 view 类型，但会暴露客户端不消费的来源信息和策略旋钮。改用显式裁剪的 view。 - **跨 Host 和 mux stream 广播注册表新增和移除。** 两条 stream 没有共享顺序，客户端因此需要 tombstone、缓冲帧和基线重试来协调。插件 key 变化不值得引入第二套同步协议。 +- **在 controller 或 Client 中合并逐 key 帧。** 按 microtask 攒批或按序号分组都需要缓冲和时序假设，而注册表本来就知道每个事件的驱动在哪里结束。改为在驱动处分组。 +- **把帧内的变化做成 key 到值的对象。** 消费方要从对象成员推断哪些 key 变化，也会丢失注册顺序。改用有序的 `{ key, val }` 数组。 +- **使用平行的 `keys` 与 `values` 两个数组。** 这样不必重复每项的字段名，但每个消费方都要按下标保持两个数组对齐，线路类型也无法表达两者等长。改用自带 key 的各项。 +- **用事件级变更流取代 `onChanged`。** forge 等 Host 消费方只关心单个 key，分组对它们没有好处。两条变更流都保留。 ## 后果 - 缺少投影组合会在激活期间失败，绝不会降级为默认值。 - host 消费方避免重复的全注册表快照和日志扫描。 - 协议负载排除 host 内部字段和逐 key 水位包装；普通基线会传达活跃 key 集。 +- 一个已提交事件对每个 Session 至多产生一个 Remote projection 帧，其 `seq` 是该事件的序号。 diff --git a/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.i18n.yaml b/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.i18n.yaml index c67dff0722..0bb6fa05e0 100644 --- a/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.i18n.yaml @@ -30,8 +30,8 @@ en: efded09f0cd30a8e zh: 5e16580a39ff3359 /agent-note-session-observations-and-projection-owned-client-state/decision/client-merge-rules: - en: fb8740be674e4708 - zh: 7dc0c89558f51792 + en: 135a93a2acecaf2a + zh: 87b69116a64b80b5 /agent-note-session-observations-and-projection-owned-client-state/decision/domain-applications: en: 4dcae9e00148b11e zh: acfd1b6831cf50fe diff --git a/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.md b/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.md index 530aeb36e1..f10764daab 100644 --- a/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.md +++ b/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.md @@ -106,7 +106,7 @@ These distinctions prevent one overloaded `undefined` from representing cache mi |---|---|---|---| | Session list hints | Partial | Last durable checkpoint or bounded fallback cut | Unknown | | Follow opening baseline | Complete for the Host composition | Exact opening cursor | Capability absent | -| Projection frame | One whole key | Event sequence carried by the frame | Not applicable | +| Projection frame | Every key the frame lists, each a whole value | Event sequence carried by the frame | Unchanged | The Client stores one row per key. A row the connected Host computed carries its sequence number: a newer baseline or frame replaces it, an equal or older input is ignored, and reconnect can therefore replace the event window without rolling back a projection frame that was already accepted at a later sequence. A list hint viewed from the persisted cache carries no comparable sequence number and yields to every Host-sequenced write ([projection cache listing identity and cached rows](2026-09-19-projection-cache-listing-identity-and-cached-rows.md)). diff --git a/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.zh.md b/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.zh.md index 62ac9db852..96b4a70242 100644 --- a/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.zh.md +++ b/.agents/notes/implemented/architecture/2026-08-25-session-observations-and-projection-owned-client-state.zh.md @@ -106,7 +106,7 @@ Projection 的三种交付状态含义不同： |---|---|---|---| | Session list hints | 部分 | 上次持久 checkpoint 或有界 fallback cut | 未知 | | Follow opening baseline | 对当前 Host composition 完整 | 精确 opening cursor | Capability 不存在 | -| Projection frame | 单个完整 key | Frame 携带的 event sequence | 不适用 | +| Projection frame | 帧列出的每个 key，各为完整值 | Frame 携带的 event sequence | 未变化 | Client 为每个 key 保存一行。由建连 Host 算出的行带 sequence number：更新的 baseline 或 frame 会替换它，相同或更旧的输入被忽略，因此 reconnect 可以替换 event window，而不会回退已经在更晚 sequence 接受的 projection frame。从持久化缓存看出来的 list hint 不带可比的 sequence number，让位于所有 Host 排序过的写入（[Projection cache 只读面按 lifecycle 身份匹配，客户端 store 区分 cached 与 sequenced 行](2026-09-19-projection-cache-listing-identity-and-cached-rows.zh.md)）。 diff --git a/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.i18n.yaml b/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.i18n.yaml index 9ae0452069..575a8026c7 100644 --- a/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.i18n.yaml @@ -12,8 +12,8 @@ en: 45c63df5dac389eb zh: d72a5bcb94de22f1 /agent-note-session-snapshots-passive-subscriptions-and-web-aggregation/decision/reads-and-subscriptions: - en: 38df4f3c4fba2cc9 - zh: a34f3376faf26f0c + en: a2439f02cb9654f9 + zh: 9de786792cba6a1d /agent-note-session-snapshots-passive-subscriptions-and-web-aggregation/decision/follow-ordering-and-ownership: en: 352ff87a3e9cd110 zh: 4ac6366cecfc08a3 diff --git a/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.md b/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.md index b5e1112a18..98d632eda8 100644 --- a/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.md +++ b/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.md @@ -18,7 +18,7 @@ History reads return chronological records, `hasMore`, and an inclusive cursor. `usingObservation({ sessionId, projection }, signal, callback?)` acquires the existing disposable SessionObservation. A callback borrows it until the callback settles; without a callback the caller owns disposal. Ordinary history, state, and Wait output reads do not request projections. Follow, exact projection reads, and subagent history descriptor validation request them. No read activates an Agent implicitly. -`observeEvent` and `observeProjections` install listeners without acquiring an Observation. Their initial `ready` confirms subscription, not a snapshot cursor, existence, parent-child validity, or Agent readiness. History establishes continuity from its first received event; projection updates replace individual keys and may share a sequence number. `observeState` instead begins with a complete state read and coalesces later Session and Agent invalidations. +`observeEvent` and `observeProjections` install listeners without acquiring an Observation. Their initial `ready` confirms subscription, not a snapshot cursor, existence, parent-child validity, or Agent readiness. History establishes continuity from its first received event; each projection change frame covers one committed event and replaces only the keys it lists. `observeState` instead begins with a complete state read and coalesces later Session and Agent invalidations. History, history observation, and Follow accept Address requests; projection and state operations accept Session IDs. All internal readers and listeners use Session IDs. History and Follow validate Address against their opening Observation at the Controller layer, including the existing subagent descriptor rules. Pure history observation does not acquire a read for entity validation. Gateway resource authorization remains independent: the registered Session resolver extracts the target ID from the request before access is decided. diff --git a/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.zh.md b/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.zh.md index 773ab039b3..fa9d53c2d8 100644 --- a/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.zh.md +++ b/.agents/notes/implemented/architecture/2026-10-21-session-history-stream-and-snapshot.zh.md @@ -18,7 +18,7 @@ Status: implemented `usingObservation({ sessionId, projection }, signal, callback?)` 获取已有的可释放 SessionObservation。有回调时租约借给回调直到其结束；无回调时由调用方释放。普通历史、状态及 Wait 输出读取不请求投影。Follow、精确投影读取和子会话历史 descriptor 校验请求投影。读取不会隐式激活 Agent。 -`observeEvent` 和 `observeProjections` 安装监听而不获取 Observation。首个 `ready` 确认订阅生效，不代表快照游标、实体存在、父子关系有效或 Agent 就绪。历史从首个收到的事件建立连续性；投影更新替换单个键，多个更新可以共享序号。`observeState` 则从完整状态读取开始，并合并后续 Session 和 Agent 失效通知。 +`observeEvent` 和 `observeProjections` 安装监听而不获取 Observation。首个 `ready` 确认订阅生效，不代表快照游标、实体存在、父子关系有效或 Agent 就绪。历史从首个收到的事件建立连续性；每个投影变更帧对应一个已提交事件，只替换帧内列出的键。`observeState` 则从完整状态读取开始，并合并后续 Session 和 Agent 失效通知。 历史、历史观察和 Follow 接收 Address 请求；投影和状态操作接收 Session ID。所有内部读取器和监听器都使用 Session ID。History 和 Follow 在 Controller 层以开场 Observation 校验 Address，包括已有的子会话 descriptor 规则。纯历史观察不为实体校验获取读取。Gateway 资源授权与之独立：注册的 Session 解析器在访问判定前从请求提取目标 ID。 diff --git a/apps/web/tests/idle-submission-handoff.e2e.ts b/apps/web/tests/idle-submission-handoff.e2e.ts index 958c1f5e30..98f8b25f90 100644 --- a/apps/web/tests/idle-submission-handoff.e2e.ts +++ b/apps/web/tests/idle-submission-handoff.e2e.ts @@ -45,9 +45,9 @@ class HistoryDeliveryGate { return } const control = endpoint === 'session/control' && frame.type === 'item' - ? frame.value as { type: string; sessionId?: string; key?: string; value?: InboxState } : undefined - const inbox = control?.type === 'projection' && control.sessionId === SESSION_ID && control.key === 'inbox' - ? control.value : undefined + ? frame.value as { type: string; sessionId?: string; changes?: readonly { key: string; val: unknown }[] } : undefined + const inbox = control?.type === 'projection' && control.sessionId === SESSION_ID + ? control.changes?.find(change => change.key === 'inbox')?.val as InboxState | undefined : undefined const follow = endpoint === 'session/follow' && frame.type === 'item' ? frame.value as { type: string; event?: SessionEvent } : undefined this.buffered.push({ stream: frame.streamId, send: () => { socket.send(message) }, diff --git a/apps/web/tests/steering.e2e.ts b/apps/web/tests/steering.e2e.ts index 3b47c0a4b5..a687c45f57 100644 --- a/apps/web/tests/steering.e2e.ts +++ b/apps/web/tests/steering.e2e.ts @@ -371,8 +371,8 @@ describe.each(['reconnect', 'delayed-inbox'] as const)('web e2e: composer shortc server.onMessage((message) => { const frame = parseRemoteStreamServerMessage(String(message)) if (holdInbox && frame.type === 'item') { - const value = frame.value as { type?: string; key?: string } - if (value.type === 'projection' && value.key === 'inbox') heldStream = frame.streamId + const value = frame.value as { type?: string; changes?: readonly { key: string }[] } + if (value.type === 'projection' && value.changes?.some(change => change.key === 'inbox') === true) heldStream = frame.streamId } // Hold the remaining control stream in FIFO order, while history keeps flowing. if (holdInbox && frame.streamId === heldStream) heldControl.push(() => { route.send(message) }) diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index b708c88637..976971787d 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -23,11 +23,11 @@ Application listeners are outside this package matrix. Desktop Host subscribes t | `agent/request-error` | `waterfall` | [`packages/core/agent/src/runtime-types.ts:389`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`compaction-basic`](../packages/compaction/compaction-basic), [`compaction-image-offload`](../packages/compaction/compaction-image-offload), [`llm-retry`](../packages/llm/llm-retry) | | `agent/status` | `emit` | [`packages/core/agent/src/runtime-types.ts:309`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`acp`](../packages/acp/acp), `agent-team`, [`compaction-basic`](../packages/compaction/compaction-basic), [`goal-round-driver`](../packages/goal/goal-round-driver), `server`, `session-controller` | | `agent/turn-stopping` | `serial` | [`packages/core/agent/src/runtime-types.ts:417`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`hooks-claude-code`](../packages/experimental/hooks-claude-code), [`hooks-codex`](../packages/experimental/hooks-codex), [`workspace-changes`](../packages/deliverables/workspace-changes) | -| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:691`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | -| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:671`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | -| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:698`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | -| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:677`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | -| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:684`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:700`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:680`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:707`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:686`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:693`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | | `app-boot/config-reload` | `emit` | [`packages/boot/app-boot/src/index.ts:55`](../packages/boot/app-boot/src/index.ts) | [`app-boot`](../packages/boot/app-boot) (`emit`) | [`settings`](../packages/settings/settings) | | `approval/request` | `waterfall` | [`packages/interaction/user-approval/src/types.ts:87`](../packages/interaction/user-approval/src/types.ts) | [`user-approval`](../packages/interaction/user-approval) (`waterfall`) | [`acp`](../packages/acp/acp), `remotes`, `server` | | `authorization/settled` | `emit` | [`packages/credentials/authorization/src/index.ts:57`](../packages/credentials/authorization/src/index.ts) | [`authorization`](../packages/credentials/authorization) (`events.dispatch`) | - | diff --git a/docs/event-producer-consumer.zh.md b/docs/event-producer-consumer.zh.md index ab1c8e803d..8f2a0d34a9 100644 --- a/docs/event-producer-consumer.zh.md +++ b/docs/event-producer-consumer.zh.md @@ -25,11 +25,11 @@ | `agent/request-error` | `waterfall` | [`packages/core/agent/src/runtime-types.ts:389`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`compaction-basic`](../packages/compaction/compaction-basic), [`compaction-image-offload`](../packages/compaction/compaction-image-offload), [`llm-retry`](../packages/llm/llm-retry) | | `agent/status` | `emit` | [`packages/core/agent/src/runtime-types.ts:309`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`acp`](../packages/acp/acp), `agent-team`, [`compaction-basic`](../packages/compaction/compaction-basic), [`goal-round-driver`](../packages/goal/goal-round-driver), `server`, `session-controller` | | `agent/turn-stopping` | `serial` | [`packages/core/agent/src/runtime-types.ts:417`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`hooks-claude-code`](../packages/experimental/hooks-claude-code), [`hooks-codex`](../packages/experimental/hooks-codex), [`workspace-changes`](../packages/deliverables/workspace-changes) | -| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:691`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | -| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:671`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | -| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:698`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | -| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:677`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | -| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:684`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/activity` | `emit` | [`packages/api/session-controller/src/types.ts:700`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/added` | `emit` | [`packages/api/session-controller/src/types.ts:680`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/error` | `emit` | [`packages/api/session-controller/src/types.ts:707`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/removed` | `emit` | [`packages/api/session-controller/src/types.ts:686`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | +| `api-session/status` | `emit` | [`packages/api/session-controller/src/types.ts:693`](../packages/api/session-controller/src/types.ts) | `session-controller` (`emit`) | `remotes` | | `app-boot/config-reload` | `emit` | [`packages/boot/app-boot/src/index.ts:55`](../packages/boot/app-boot/src/index.ts) | [`app-boot`](../packages/boot/app-boot) (`emit`) | [`settings`](../packages/settings/settings) | | `approval/request` | `waterfall` | [`packages/interaction/user-approval/src/types.ts:87`](../packages/interaction/user-approval/src/types.ts) | [`user-approval`](../packages/interaction/user-approval) (`waterfall`) | [`acp`](../packages/acp/acp), `remotes`, `server` | | `authorization/settled` | `emit` | [`packages/credentials/authorization/src/index.ts:57`](../packages/credentials/authorization/src/index.ts) | [`authorization`](../packages/credentials/authorization) (`events.dispatch`) | - | diff --git a/docs/subsystems/session-projection.i18n.yaml b/docs/subsystems/session-projection.i18n.yaml index b38666ddb4..917f2152fb 100644 --- a/docs/subsystems/session-projection.i18n.yaml +++ b/docs/subsystems/session-projection.i18n.yaml @@ -9,8 +9,8 @@ en: d48814d3ab0e997f zh: d0337d8eba097a63 /session-projections/the-snapshot-and-the-change-feed: - en: 630c1041f41f89bf - zh: a3361bdead116cd7 + en: bc9d438b90fc2baa + zh: df7cbea2dad3786b /session-projections/summary-reads: en: d2bc9bcc775acfb7 zh: b1d2ba74496305b8 diff --git a/docs/subsystems/session-projection.md b/docs/subsystems/session-projection.md index cee8103ef0..7ea7c06716 100644 --- a/docs/subsystems/session-projection.md +++ b/docs/subsystems/session-projection.md @@ -135,7 +135,22 @@ type ProjectionChangeListener = ( ) => void ``` -`snapshot(session)` is fully synchronous: a carrier reads it in the same tick as its page slice, so `asOfSeq` covers both reads at one sequence number. It returns only client views, and every value passes its unit's `viewSchema` before return. `handle.stateOf(session)` reads one live host state without computing unrelated views; callers must not mutate the borrowed reference. A state-reference change computes one cached raw view, and the change feed fires only when that result changes by `Object.is`; an object-valued view must preserve its reference to suppress publication across internal-only state changes. +```ts type-equiv +/** + * Event-level change-feed listener: one committed event changed the raw `view` + * result of at least one client-visible unit by `Object.is` for one session. + * `changes` holds one entry per such unit in registration order, each with the + * unit's `key` and its schema-validated view as `val`; `seq` is the seq of the + * causing event. + */ +type ProjectionEventChangeListener = ( + session: Session, + changes: readonly { readonly key: Extract; readonly val: unknown }[], + seq: SessionSeq, +) => void +``` + +`snapshot(session)` is fully synchronous: a carrier reads it in the same tick as its page slice, so `asOfSeq` covers both reads at one sequence number. It returns only client views, and every value passes its unit's `viewSchema` before return. `handle.stateOf(session)` reads one live host state without computing unrelated views; callers must not mutate the borrowed reference. A state-reference change computes one cached raw view, and the change feed fires only when that result changes by `Object.is`; an object-valued view must preserve its reference to suppress publication across internal-only state changes. After every unit has been driven for one committed event, the event-level feed calls each `onEventChanged` listener once with every client-visible key whose view the event changed; an event that changes no client view makes no event-level call. ## Summary reads @@ -346,6 +361,18 @@ register( definition: ProjectionDefinition & { key: E */ onChanged(listener: ProjectionChangeListener): () => void +/** + * Subscribe to the event-level change feed. The registration is an effect on + * the calling context's fiber. The listener runs synchronously inside the + * committed event's `session/event` handling, once after every unit has been + * driven and after every {@link onChanged} call for that event. If a unit or + * an {@link onChanged} listener throws during that drive, the listener still + * receives the changes made before the throw. + * @param listener - called once per committed event that changed at least one client-visible unit's raw view by `Object.is`. + * @returns the exact disposer that unsubscribes. + */ +onEventChanged(listener: ProjectionEventChangeListener): () => void + /** * One consistent cut over every registered client-visible unit for one session, read from * the watermark cache (missing cells fold lazily over the in-memory log). diff --git a/docs/subsystems/session-projection.zh.md b/docs/subsystems/session-projection.zh.md index 6e076a18a3..b84720e804 100644 --- a/docs/subsystems/session-projection.zh.md +++ b/docs/subsystems/session-projection.zh.md @@ -135,7 +135,22 @@ type ProjectionChangeListener = ( ) => void ``` -`snapshot(session)` 完全同步：载体在切出页面切片的同一 tick 内读取它，因此 `asOfSeq` 使两次读取使用同一个序号。它只返回客户端视图，并在返回前通过各单元的 `viewSchema` 校验。`handle.stateOf(session)` 可在不计算无关视图的情况下读取一份实时 host 状态；调用方不得修改这一借用引用。state 引用变化时，注册表计算并缓存一次原始 view；只有该结果通过 `Object.is` 判定为变化时才触发变更流，对象 view 若要在仅内部 state 变化时抑制发布就必须保留引用。 +```ts type-equiv +/** + * Event-level change-feed listener: one committed event changed the raw `view` + * result of at least one client-visible unit by `Object.is` for one session. + * `changes` holds one entry per such unit in registration order, each with the + * unit's `key` and its schema-validated view as `val`; `seq` is the seq of the + * causing event. + */ +type ProjectionEventChangeListener = ( + session: Session, + changes: readonly { readonly key: Extract; readonly val: unknown }[], + seq: SessionSeq, +) => void +``` + +`snapshot(session)` 完全同步：载体在切出页面切片的同一 tick 内读取它，因此 `asOfSeq` 使两次读取使用同一个序号。它只返回客户端视图，并在返回前通过各单元的 `viewSchema` 校验。`handle.stateOf(session)` 可在不计算无关视图的情况下读取一份实时 host 状态；调用方不得修改这一借用引用。state 引用变化时，注册表计算并缓存一次原始 view；只有该结果通过 `Object.is` 判定为变化时才触发变更流，对象 view 若要在仅内部 state 变化时抑制发布就必须保留引用。一个已提交事件驱动完所有单元后，事件级变更流以该事件改变了视图的全部客户端可见 key 调用每个 `onEventChanged` 监听器一次；未改变任何客户端视图的事件不产生事件级调用。 ## 摘要读取 @@ -346,6 +361,18 @@ register( definition: ProjectionDefinition & { key: E */ onChanged(listener: ProjectionChangeListener): () => void +/** + * Subscribe to the event-level change feed. The registration is an effect on + * the calling context's fiber. The listener runs synchronously inside the + * committed event's `session/event` handling, once after every unit has been + * driven and after every {@link onChanged} call for that event. If a unit or + * an {@link onChanged} listener throws during that drive, the listener still + * receives the changes made before the throw. + * @param listener - called once per committed event that changed at least one client-visible unit's raw view by `Object.is`. + * @returns the exact disposer that unsubscribes. + */ +onEventChanged(listener: ProjectionEventChangeListener): () => void + /** * One consistent cut over every registered client-visible unit for one session, read from * the watermark cache (missing cells fold lazily over the in-memory log). diff --git a/docs/subsystems/session.md b/docs/subsystems/session.md index fb3e56ee87..7a0ad908d9 100644 --- a/docs/subsystems/session.md +++ b/docs/subsystems/session.md @@ -998,10 +998,10 @@ workspaceDesktop(): { name: string; available: boolean; fileManager: 'finder' | @Remote({ mode: 'stream' }) observeEvent(request: SessionObserveEventRequest, signal: AbortSignal): AsyncIterable /** - * Subscribe to sparse Session projection changes. + * Subscribe to Session projection changes. * @param request - target Session. * @param signal - subscription cancellation. - * @returns readiness and sparse projection changes. + * @returns readiness followed by one projection change frame per committed event. */ @Remote({ mode: 'stream' }) observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable diff --git a/docs/subsystems/session.zh.md b/docs/subsystems/session.zh.md index fb26a5105c..ec9fab642f 100644 --- a/docs/subsystems/session.zh.md +++ b/docs/subsystems/session.zh.md @@ -1002,10 +1002,10 @@ workspaceDesktop(): { name: string; available: boolean; fileManager: 'finder' | @Remote({ mode: 'stream' }) observeEvent(request: SessionObserveEventRequest, signal: AbortSignal): AsyncIterable /** - * Subscribe to sparse Session projection changes. + * Subscribe to Session projection changes. * @param request - target Session. * @param signal - subscription cancellation. - * @returns readiness and sparse projection changes. + * @returns readiness followed by one projection change frame per committed event. */ @Remote({ mode: 'stream' }) observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable diff --git a/docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.i18n.yaml b/docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.i18n.yaml new file mode 100644 index 0000000000..5adf2d9984 --- /dev/null +++ b/docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.i18n.yaml @@ -0,0 +1,16 @@ +# Bilingual-pair consistency record for guide.md (docs/i18n/README.md): per heading +# section, a hash of its English and Chinese blocks outside code blocks and generated regions. +# After editing either side, bring the other along and re-record with: +# pnpm run verify-translation-pairing --write docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.md +/: + en: e48b8561bd5e8c36 + zh: 250e816f38495c97 +/projection-frames-list-every-key-one-event-changed: + en: 3f22f3e46a3903db + zh: 7694cef52b9aa7f0 +/projection-frames-list-every-key-one-event-changed/change: + en: f0d17c8787e7da13 + zh: 3b193ae03232b6be +/projection-frames-list-every-key-one-event-changed/migration: + en: fa9b8d4e4c426754 + zh: 70f21eb8ec280fdd diff --git a/docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.md b/docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.md new file mode 100644 index 0000000000..54b292f15f --- /dev/null +++ b/docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.md @@ -0,0 +1,32 @@ +--- +kind: upgrade-guide +description: "Live Session projection frames list every key one committed event changed instead of carrying one key each." +--- + +# Projection frames list every key one event changed + +English | [中文](guide.zh.md) + +## Change + +Live `projection` frames on the `session/control` stream and change frames on the `session/observeProjections` stream carried one projection key each: `{ type: 'projection', sessionId, key, value, seq }`. An event that changed several keys produced several frames with the same `seq`. Each frame now carries every key one committed event changed: `{ type: 'projection', sessionId, seq, changes }` on `session/control`, and the same fields without `sessionId` on `session/observeProjections`. `changes` holds one `{ key, val }` entry per changed key, without duplicate keys, where `val` is the key's new value. A key absent from `changes` is unchanged. The exported types `SessionProjectionUpdate`, `SessionProjectionChange`, `SessionControlFrame`, and `SessionProjectionFrame` of `@deepseek-ai/dsh-api-session-controller` follow the new fields. + +Baseline frames, `session/projections` reads, and the Host listener `ctx.sessionProjections.onChanged` keep their behavior. The Host adds `ctx.sessionProjections.onEventChanged` for one call per committed event. + +This affects Remote clients that read either stream directly, such as test harnesses, proxies, and external dashboards. The bundled Web client and `@deepseek-ai/dsh-api-session-controller/client` read the new frames. + +## Migration + +1. Replace a call such as `apply(frame.key, frame.value, frame.seq)` with a loop over `frame.changes`: + + ```ts + import type { SessionControlFrame } from '@deepseek-ai/dsh-api-session-controller' + + declare const frame: Extract + declare function apply(key: string, value: unknown, seq: number): void + + for (const change of frame.changes) apply(change.key, change.val, frame.seq) + ``` + +2. Replace key tests such as `frame.key === 'inbox'` with `frame.changes.find(change => change.key === 'inbox')`, and read the value from that entry's `val`. +3. Confirm that one committed event yields at most one `projection` frame per Session, and that the values your client holds match a fresh `session/projections` read. diff --git a/docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.zh.md b/docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.zh.md new file mode 100644 index 0000000000..20b86923d2 --- /dev/null +++ b/docs/upgrade-guide/v0.2.2-alpha.2/projection-event-frames/guide.zh.md @@ -0,0 +1,32 @@ +--- +kind: upgrade-guide +description: "实时 Session projection 帧改为列出一个已提交事件改变的全部 key，不再每帧只带一个 key。" +--- + +# Projection 帧列出一个事件改变的全部 key + +[English](guide.md) | 中文 + +## 变更 + +`session/control` 流上的实时 `projection` 帧与 `session/observeProjections` 流上的变更帧，原先每帧只带一个 projection key：`{ type: 'projection', sessionId, key, value, seq }`。一个事件改变多个 key 时，会产生多帧 `seq` 相同的帧。现在每帧携带一个已提交事件改变的全部 key：`session/control` 上是 `{ type: 'projection', sessionId, seq, changes }`，`session/observeProjections` 上是去掉 `sessionId` 的同样字段。`changes` 中每个变化的 key 一项 `{ key, val }`，key 不重复，`val` 是该 key 的新值。未出现在 `changes` 中的 key 没有变化。`@deepseek-ai/dsh-api-session-controller` 导出的 `SessionProjectionUpdate`、`SessionProjectionChange`、`SessionControlFrame` 与 `SessionProjectionFrame` 类型随新字段更新。 + +baseline 帧、`session/projections` 读取以及 Host 监听器 `ctx.sessionProjections.onChanged` 的行为不变。Host 新增 `ctx.sessionProjections.onEventChanged`，按每个已提交事件调用一次。 + +受影响的是直接读取这两条流的 Remote 客户端，例如测试 harness、代理和外部看板。内置 Web 客户端与 `@deepseek-ai/dsh-api-session-controller/client` 已读取新帧。 + +## 迁移 + +1. 把 `apply(frame.key, frame.value, frame.seq)` 这类调用改为遍历 `frame.changes`： + + ```ts + import type { SessionControlFrame } from '@deepseek-ai/dsh-api-session-controller' + + declare const frame: Extract + declare function apply(key: string, value: unknown, seq: number): void + + for (const change of frame.changes) apply(change.key, change.val, frame.seq) + ``` + +2. 把 `frame.key === 'inbox'` 这类 key 判断改为 `frame.changes.find(change => change.key === 'inbox')`，并从该项的 `val` 读取值。 +3. 确认一个已提交事件对每个 Session 至多产生一个 `projection` 帧，且客户端持有的值与重新调用 `session/projections` 读到的一致。 diff --git a/packages/api/session-controller/src/client/sessions/manager.ts b/packages/api/session-controller/src/client/sessions/manager.ts index 105930d1ef..ecb6a0e127 100644 --- a/packages/api/session-controller/src/client/sessions/manager.ts +++ b/packages/api/session-controller/src/client/sessions/manager.ts @@ -642,15 +642,17 @@ export class SessionManager { // ---- Live control and Host-event sinks ---- /** - * Apply a complete control baseline or one later replacement frame. - * @param frame - baseline or live control replacement from Session Controller. + * Apply a complete control baseline or one later per-event projection frame. + * @param frame - baseline or live per-event projection change from Session Controller. */ handleControlFrame(frame: SessionControlFrame): void { if (frame.type === 'baseline') { this.replaceControlBaseline(frame.value) return } - this.projectionStore(frame.sessionId).apply(frame.key, frame.value, SessionSeq(frame.seq)) + const store = this.projectionStore(frame.sessionId) + const seq = SessionSeq(frame.seq) + for (const change of frame.changes) store.apply(change.key, change.val, seq) this.notifier.markDirty() } diff --git a/packages/api/session-controller/src/control.ts b/packages/api/session-controller/src/control.ts index 8d27396d29..7a9d04021e 100644 --- a/packages/api/session-controller/src/control.ts +++ b/packages/api/session-controller/src/control.ts @@ -5,11 +5,11 @@ import { Deque } from '@deepseek-ai/dsh-deque' import type { Session, SessionId, } from '@deepseek-ai/dsh-session' -import type { JsonValue } from '@deepseek-ai/dsh-util-values' import type { SessionControlBaseline, SessionControlFrame, SessionProjectionBaseline, + SessionProjectionUpdate, SessionProjectionValues, } from './types.ts' @@ -19,13 +19,13 @@ export class SessionControlController { /** @param ctx - Host context carrying live Agent and projection services. */ constructor(private readonly ctx: Context) { - ctx.sessionProjections.onChanged((session, key, value, seq) => { + ctx.sessionProjections.onEventChanged((session, changes, seq) => { this.broadcast({ type: 'projection', sessionId: session.id, - key, - value: value as JsonValue, seq, + // Every projection definition validates its value before change publication. + changes: changes as SessionProjectionUpdate['changes'], }) }) ctx.effect(() => () => { diff --git a/packages/api/session-controller/src/history/controller.ts b/packages/api/session-controller/src/history/controller.ts index d090f0a3f7..1899aff823 100644 --- a/packages/api/session-controller/src/history/controller.ts +++ b/packages/api/session-controller/src/history/controller.ts @@ -103,10 +103,10 @@ export class HistoryController { } /** - * Subscribe to sparse projection replacements without reading a baseline. + * Subscribe to projection changes without reading a baseline. * @param request - Session identity to subscribe without reading a baseline. * @param signal - stream cancellation. - * @returns readiness followed by sparse per-key replacement values. + * @returns readiness followed by one change frame per committed event, listing every key it changed. */ async *observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable { using listener = new ProjectionListener(this.ctx, request.sessionId, signal) diff --git a/packages/api/session-controller/src/history/listener.ts b/packages/api/session-controller/src/history/listener.ts index a272081620..cb41847dbf 100644 --- a/packages/api/session-controller/src/history/listener.ts +++ b/packages/api/session-controller/src/history/listener.ts @@ -5,7 +5,6 @@ import { Deque } from '@deepseek-ai/dsh-deque' import { SessionLogOffset } from '@deepseek-ai/dsh-session' import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session' import { RemoteError } from '@deepseek-ai/dsh-typert-protocol' -import type { JsonValue } from '@deepseek-ai/dsh-util-values' import { entryFor } from '../history.ts' import type { SessionProjectionChange, SessionEventEntry } from '../types.ts' @@ -128,7 +127,7 @@ export function eventItem(item: EventItem, next?: number): { return { events, next } } -/** Sparse projection changes retain every key, including keys sharing a seq. */ +/** Projection changes of one Session, one change per committed event that changed a projection. */ export class ProjectionListener implements Disposable { private readonly buffer: ListenerBuffer<{ session: Session; change: SessionProjectionChange }> @@ -142,12 +141,12 @@ export class ProjectionListener implements Disposable { const input = this.buffer.createInput<{ session: Session; change: SessionProjectionChange }>(item => [item]) const target = sessionId try { - this.buffer.own(ctx.sessionProjections.onChanged((session, key, value, seq) => { + this.buffer.own(ctx.sessionProjections.onEventChanged((session, changes, seq) => { if (session.id !== target) return this.buffer.capture(() => { - input.append({ - session, change: { type: 'projection', key, value: value as JsonValue, seq } satisfies SessionProjectionChange, - }) + // Every projection definition validates its value before change publication. + const change: SessionProjectionChange = { type: 'projection', seq, changes: changes as SessionProjectionChange['changes'] } + input.append({ session, change }) }) })) } catch (error: unknown) { @@ -156,8 +155,8 @@ export class ProjectionListener implements Disposable { } } - /** Consume sparse per-key projection replacements. - * @returns validated replacement frames; projection seqs need not be contiguous. */ + /** Consume per-event projection changes. + * @returns validated change frames, one per committed event; projection seqs need not be contiguous. */ async *changes(): AsyncIterable { try { for await (const { change } of this.buffer.items()) { diff --git a/packages/api/session-controller/src/index.ts b/packages/api/session-controller/src/index.ts index 94d08c77e3..8e81986f0b 100644 --- a/packages/api/session-controller/src/index.ts +++ b/packages/api/session-controller/src/index.ts @@ -553,10 +553,10 @@ export class SessionController extends TypertRemoteService { } /** - * Subscribe to sparse Session projection changes. + * Subscribe to Session projection changes. * @param request - target Session. * @param signal - subscription cancellation. - * @returns readiness and sparse projection changes. + * @returns readiness followed by one projection change frame per committed event. */ @Remote({ mode: 'stream' }) observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable { diff --git a/packages/api/session-controller/src/types.ts b/packages/api/session-controller/src/types.ts index 9e399d8f3b..6c3c737cb6 100644 --- a/packages/api/session-controller/src/types.ts +++ b/packages/api/session-controller/src/types.ts @@ -569,12 +569,21 @@ export interface SessionControlBaseline { readonly projections: Readonly> } -/** One finished projection value and its durable watermark. */ +/** + * Projection changes caused by one committed Session event; the Host emits one + * update per event that changed at least one client-visible projection view. + */ export interface SessionProjectionUpdate { readonly sessionId: SessionId - readonly key: string - readonly value: JsonValue + /** Seq of the committed event that caused every listed change. */ readonly seq: number + /** + * One entry per client-visible key whose view the event changed, in + * registration order without duplicate keys, with the key's new value as + * `val`. A key absent from `changes` is unchanged; unlike a baseline, absence + * does not mean an absent capability. + */ + readonly changes: readonly { readonly key: string; readonly val: SessionProjectionValue }[] } /** Host-wide live state stream. Each generation starts with exactly one baseline. */ @@ -652,10 +661,10 @@ export type SessionEventFrame = | { readonly type: 'ready' } | SessionEventEntry -/** One sparse per-key replacement; different keys can share a seq. */ +/** Projection changes caused by one committed event of the observed Session. */ export type SessionProjectionChange = { readonly type: 'projection' } & Omit -/** Subscription readiness without a baseline, followed by per-key changes. */ +/** Subscription readiness without a baseline, followed by one change frame per committed event. */ export type SessionProjectionFrame = | { readonly type: 'ready' } | SessionProjectionChange diff --git a/packages/api/session-controller/tests/control-queue.host.spec.ts b/packages/api/session-controller/tests/control-queue.host.spec.ts index ad24ad94fb..fee05cb585 100644 --- a/packages/api/session-controller/tests/control-queue.host.spec.ts +++ b/packages/api/session-controller/tests/control-queue.host.spec.ts @@ -52,10 +52,15 @@ describe('Session control Inbox projection', () => { for (;;) { const next = await iterator.next() if (next.done) throw new Error('stream ended before an Inbox value') - if (next.value.type === 'projection' && next.value.key === 'inbox') return next.value + if (next.value.type === 'projection' && next.value.changes.some(change => change.key === 'inbox')) return next.value } } + /** The Inbox value one projection frame carries. */ + function inboxValue(frame: Extract): unknown { + return frame.changes.find(change => change.key === 'inbox')?.val + } + it('projects both pending lists in baselines and live replacement frames', async () => { const { control, inbox } = await harness() const queued = message('queued') @@ -82,10 +87,10 @@ describe('Session control Inbox projection', () => { const replacement = message('replacement') inbox.append('next-turn', replacement) const replaced = await nextInboxFrame(iterator) - expect(replaced.value).toMatchObject({ 'next-turn': [queued, replacement] }) + expect(inboxValue(replaced)).toMatchObject({ 'next-turn': [queued, replacement] }) inbox.remove(steering.id) const removed = await nextInboxFrame(iterator) - expect(removed.value).toMatchObject({ 'next-step': [context] }) + expect(inboxValue(removed)).toMatchObject({ 'next-step': [context] }) abort.abort() await iterator.next() @@ -109,8 +114,7 @@ describe('Session control Inbox projection', () => { await expect(iterator.next()).resolves.toMatchObject({ value: { type: 'projection', - key: 'inbox', - value: { 'next-turn': [{ id: pending.id }], 'next-step': [] }, + changes: [{ key: 'inbox', val: { 'next-turn': [{ id: pending.id }], 'next-step': [] } }], }, }) @@ -167,9 +171,9 @@ describe('Session control Inbox projection', () => { target: 'next-turn', start: 0, inserted: [pending], }) expect(ctx.agents.get(session.id)).toBeUndefined() - await expect(nextInboxFrame(iterator)).resolves.toMatchObject({ - sessionId: session.id, value: { 'next-turn': [pending], 'next-step': [] }, - }) + const frame = await nextInboxFrame(iterator) + expect(frame.sessionId).toBe(session.id) + expect(inboxValue(frame)).toMatchObject({ 'next-turn': [pending], 'next-step': [] }) abort.abort() await iterator.next() }) @@ -197,14 +201,14 @@ describe('Session control Inbox projection', () => { inbox.append('next-turn', first) inbox.append('next-turn', second) - const values: InboxState[] = [] + const values: unknown[] = [] ownedContexts.delete(ctx) await ctx.fiber.dispose() for (;;) { const next = await iterator.next() if (next.done) break - if (next.value.type === 'projection' && next.value.key === 'inbox') values.push(next.value.value as unknown as InboxState) + if (next.value.type === 'projection' && next.value.changes.some(change => change.key === 'inbox')) values.push(inboxValue(next.value)) } - expect(values.map(value => value['next-turn'].map(item => item.id))).toEqual([[first.id], [first.id, second.id]]) + expect(values).toMatchObject([{ 'next-turn': [{ id: first.id }] }, { 'next-turn': [{ id: first.id }, { id: second.id }] }]) }) }) diff --git a/packages/api/session-controller/tests/history-controller.host.spec.ts b/packages/api/session-controller/tests/history-controller.host.spec.ts index e3592ddbad..1f71f48e5e 100644 --- a/packages/api/session-controller/tests/history-controller.host.spec.ts +++ b/packages/api/session-controller/tests/history-controller.host.spec.ts @@ -40,19 +40,22 @@ describe('HistoryController passive reads', () => { expect(await stream.next()).toMatchObject({ done: true }) }) - it('subscribes to sparse projection changes without an Observation', async () => { + it('subscribes to per-event projection changes without an Observation', async () => { const f = await fixture() const observe = vi.spyOn(f.ctx.sessionQuery, 'observeSession') - let receive: Parameters[0] | undefined + let receive: Parameters[0] | undefined const dispose = vi.fn() - vi.spyOn(f.ctx.sessionProjections, 'onChanged').mockImplementation((callback) => { + vi.spyOn(f.ctx.sessionProjections, 'onEventChanged').mockImplementation((callback) => { receive = callback return dispose }) const stream = f.history.observeProjections({ sessionId: f.session.id }, f.abort.signal)[Symbol.asyncIterator]() expect(await stream.next()).toMatchObject({ value: { type: 'ready' } }) - receive?.(f.session, 'title', 'new title', SessionSeq(0)) - expect(await stream.next()).toMatchObject({ value: { type: 'projection', key: 'title', value: 'new title', seq: 0 } }) + receive?.(f.session, [{ key: 'title', val: 'new title' }, { key: 'goal', val: null }], SessionSeq(0)) + expect(await stream.next()).toEqual({ + done: false, + value: { type: 'projection', seq: 0, changes: [{ key: 'title', val: 'new title' }, { key: 'goal', val: null }] }, + }) expect(observe).not.toHaveBeenCalled() f.abort.abort() expect(await stream.next()).toMatchObject({ done: true }) 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 0522c8db35..6e3c9b0bed 100644 --- a/packages/api/session-controller/tests/inbox-projection.client.spec.ts +++ b/packages/api/session-controller/tests/inbox-projection.client.spec.ts @@ -34,9 +34,8 @@ function inboxFrame(value: InboxState): Extract { }) describe('ProjectionListener', () => { - it('emits sparse changes only for the selected Session', async () => { + it('emits one change per event only for the selected Session', async () => { const { ctx, session, abort } = await fixture() - let receive!: Parameters[0] + let receive!: Parameters[0] const dispose = vi.fn() - vi.spyOn(ctx.sessionProjections, 'onChanged').mockImplementation((listener) => { receive = listener; return dispose }) + vi.spyOn(ctx.sessionProjections, 'onEventChanged').mockImplementation((listener) => { receive = listener; return dispose }) using listener = new ProjectionListener(ctx, session.id, abort.signal) const other = ctx.sessions.create(SessionId('other')) - receive(other, 'subagent', null, SessionSeq(0)) - receive(session, 'subagent', null, SessionSeq(1)) + receive(other, [{ key: 'subagent', val: null }], SessionSeq(0)) + receive(session, [{ key: 'subagent', val: null }, { key: 'title', val: 'renamed' }], SessionSeq(1)) const changes = listener.changes()[Symbol.asyncIterator]() - expect(await changes.next()).toMatchObject({ value: { type: 'projection', key: 'subagent', value: null, seq: 1 } }) + expect(await changes.next()).toEqual({ + done: false, + value: { type: 'projection', seq: 1, changes: [{ key: 'subagent', val: null }, { key: 'title', val: 'renamed' }] }, + }) abort.abort() expect(await changes.next()).toMatchObject({ done: true }) expect(dispose).toHaveBeenCalledOnce() @@ -197,7 +200,7 @@ describe('ProjectionListener', () => { it('releases its buffer when projection subscription fails', async () => { const { ctx, session, abort } = await fixture() const remove = vi.spyOn(abort.signal, 'removeEventListener') - vi.spyOn(ctx.sessionProjections, 'onChanged').mockImplementation(() => { throw new Error('projection subscribe failed') }) + vi.spyOn(ctx.sessionProjections, 'onEventChanged').mockImplementation(() => { throw new Error('projection subscribe failed') }) expect(() => new ProjectionListener(ctx, session.id, abort.signal)).toThrow('projection subscribe failed') expect(remove).toHaveBeenCalled() }) diff --git a/packages/api/session-controller/tests/manager.client.spec.ts b/packages/api/session-controller/tests/manager.client.spec.ts index 9661f66a01..1946dfdeca 100644 --- a/packages/api/session-controller/tests/manager.client.spec.ts +++ b/packages/api/session-controller/tests/manager.client.spec.ts @@ -378,9 +378,9 @@ describe('SessionManager instances', () => { const stopObserving = manager.subscribe(() => { published.resolve(undefined) }) try { manager.handleControlFrame({ type: 'projection', sessionId: S1, - key: 'sessionListMetadata', value: { blank: false, lastPromptAt: 1200 }, seq: 8 }) + changes: [{ key: 'sessionListMetadata', val: { blank: false, lastPromptAt: 1200 } }], seq: 8 }) manager.handleControlFrame({ type: 'projection', sessionId: S2, - key: 'sessionListMetadata', value: { blank: true, lastPromptAt: null }, seq: 2 }) + changes: [{ key: 'sessionListMetadata', val: { blank: true, lastPromptAt: null } }], seq: 2 }) // The initial notification must finish while neither Session is resident. await published.promise @@ -427,7 +427,7 @@ describe('SessionManager query lifetime', () => { const entries = [{ id: S2, createdAt: 1, mode: 'one-shot' as const }] try { if (warm) { - manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 0, value: entries }) + manager.handleControlFrame({ type: 'projection', sessionId: S1, changes: [{ key: 'subagentCatalog', val: entries }], seq: 0 }) } remote.session.projections.mockRejectedValueOnce(failure) await manager.refreshProjections(S1) @@ -446,9 +446,9 @@ describe('list lifecycle', () => { const manager = makeManager(mock, remote) try { manager.handleControlFrame({ type: 'projection', sessionId: S1, - key: 'sessionListMetadata', value: { blank: false, lastPromptAt: 1200 }, seq: 8 }) + changes: [{ key: 'sessionListMetadata', val: { blank: false, lastPromptAt: 1200 } }], seq: 8 }) manager.handleControlFrame({ type: 'projection', sessionId: S2, - key: 'sessionListMetadata', value: { blank: true, lastPromptAt: null }, seq: 2 }) + changes: [{ key: 'sessionListMetadata', val: { blank: true, lastPromptAt: null } }], seq: 2 }) remote.session.list.mockResolvedValue(ok({ items: [ summary(S1, { blank: true }), summary(S2), ] as never[] })) @@ -457,7 +457,7 @@ describe('list lifecycle', () => { expect(manager.get(S1).getSnapshot().blank).toBe(false) expect(manager.get(S2).getSnapshot().blank).toBe(false) manager.handleControlFrame({ type: 'projection', sessionId: S1, - key: 'sessionListMetadata', value: { blank: true, lastPromptAt: null }, seq: 7 }) + changes: [{ key: 'sessionListMetadata', val: { blank: true, lastPromptAt: null } }], seq: 7 }) expect(manager.getListSnapshot().items[0]).toMatchObject({ blank: false, updatedAt: 1200 }) } finally { await manager.dispose() @@ -614,7 +614,7 @@ describe('list lifecycle', () => { it('retains title projections before list arrival, keeps last-wins by seq, and clears them on removal', async ({ mock, remote }) => { const manager = makeManager(mock, remote) const titleFrame = (title: string, seq: number) => { - manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'title', value: title, seq }) + manager.handleControlFrame({ type: 'projection', sessionId: S1, changes: [{ key: 'title', val: title }], seq }) } titleFrame('Newest', 4) titleFrame('Stale', 3) @@ -634,11 +634,34 @@ describe('list lifecycle', () => { expect(manager.getListSnapshot().items.find(item => item.sessionId === S1)?.title).toBeUndefined() }) + it('applies every key of one per-event frame with one list invalidation and per-key seq gating', async ({ mock, remote }) => { + remote.session.list.mockResolvedValue(ok({ items: [summary(S1, { blank: true })] as never[] })) + const manager = makeManager(mock, remote) + onTestFinished(() => manager.dispose()) + await manager.refreshList() + manager.handleControlFrame({ type: 'projection', sessionId: S1, changes: [{ key: 'title', val: 'Newest' }], seq: 4 }) + await new Promise(resolve => setTimeout(resolve, 0)) + // Bracket access reaches the private list notifier; projection-store notifiers stay out of the count. + const markDirty = vi.spyOn(manager['notifier'], 'markDirty') + + manager.handleControlFrame({ + type: 'projection', sessionId: S1, seq: 3, changes: [ + { key: 'title', val: 'Stale' }, + { key: 'sessionListMetadata', val: { blank: false, lastPromptAt: 1200 } }, + ], + }) + expect(markDirty).toHaveBeenCalledOnce() + await new Promise(resolve => setTimeout(resolve, 0)) + + expect(manager.getListSnapshot().items[0]).toMatchObject({ sessionId: S1, title: 'Newest', blank: false }) + expect(manager.projectionValues(S1)).toMatchObject({ sessionListMetadata: { blank: false, lastPromptAt: 1200 } }) + }) + it('fills cold titles from the list rows\' projections block as cached values', async ({ mock, remote }) => { const manager = makeManager(mock, remote) // A push frame landed before the list: S2's title is a sequenced value. manager.handleControlFrame({ - type: 'projection', sessionId: S2, key: 'title', value: 'Pushed', seq: 9, + type: 'projection', sessionId: S2, changes: [{ key: 'title', val: 'Pushed' }], seq: 9, }) remote.session.list.mockResolvedValue(ok({ items: [ @@ -660,7 +683,7 @@ describe('list lifecycle', () => { const manager = makeManager(mock, remote) try { await manager.refreshList() - manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'title', value: 'Unflushed', seq: 4 }) + manager.handleControlFrame({ type: 'projection', sessionId: S1, changes: [{ key: 'title', val: 'Unflushed' }], seq: 4 }) manager.handleConnected() await manager.refreshList() @@ -738,7 +761,7 @@ describe('subagent catalogs', () => { const manager = makeManager(mock, remote) onTestFinished(() => manager.dispose()) const entries = [{ id: S2, createdAt: 1, mode: 'external' as const, label: 'external task' }] - manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 0, value: entries }) + manager.handleControlFrame({ type: 'projection', sessionId: S1, changes: [{ key: 'subagentCatalog', val: entries }], seq: 0 }) expect(manager.getListSnapshot().projectionsBySession[S1]?.values.subagentCatalog).toEqual(entries) expect(manager.subagentAddress(S2)).toBeUndefined() @@ -750,8 +773,9 @@ describe('subagent catalogs', () => { onTestFinished(() => manager.dispose()) manager.handleSessionAdded(summary(S2, { parentSessionId: S1, origin: 'subagent' })) manager.handleControlFrame({ - type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 0, - value: [{ id: S2, createdAt: 1, mode: 'continuable', label: 'worker' }], + type: 'projection', sessionId: S1, seq: 0, changes: [ + { key: 'subagentCatalog', val: [{ id: S2, createdAt: 1, mode: 'continuable', label: 'worker' }] }, + ], }) expect(manager.subagentAddress(S2)).toEqual({ @@ -906,14 +930,17 @@ describe('subagent catalogs', () => { const client = await start() const manager = new SessionManager(client.ctx.remote) const entries = [{ id: S2, createdAt: 1, mode: 'unknown' as const }] - manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 0, value: entries }) + manager.handleControlFrame({ type: 'projection', sessionId: S1, changes: [{ key: 'subagentCatalog', val: entries }], seq: 0 }) const address = manager.subagentAddress(S2)! expect(address).toEqual({ parentSessionId: S1, childSessionId: S2, mode: 'unknown' }) manager.resolveTarget(address) const child = manager.get(S2) expect(child.getSnapshot().subagent?.address.mode).toBe('unknown') - manager.handleControlFrame({ type: 'projection', sessionId: S2, key: 'subagent', seq: 0, - value: { seq: SessionSeq(0), mode: 'continuable', label: 'repaired child' } }) + manager.handleControlFrame({ + type: 'projection', sessionId: S2, seq: 0, changes: [ + { key: 'subagent', val: { seq: SessionSeq(0), mode: 'continuable', label: 'repaired child' } }, + ], + }) await child.open() expect(child.getSnapshot().subagent?.address).toEqual({ ...address, mode: 'continuable' }) expect(manager.getListSnapshot().projectionsBySession[S1]?.values.subagentCatalog).toEqual(entries) @@ -924,8 +951,9 @@ describe('subagent catalogs', () => { const manager = makeManager(mock, remote) manager.handleSessionAdded(summary(S2, { origin: 'subagent', parentSessionId: S1, running: true })) manager.handleControlFrame({ - type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 4, - value: [{ id: S2, createdAt: 1, mode: 'continuable', label: 'worker' }], + type: 'projection', sessionId: S1, seq: 4, changes: [ + { key: 'subagentCatalog', val: [{ id: S2, createdAt: 1, mode: 'continuable', label: 'worker' }] }, + ], }) expect(manager.getListSnapshot().projectionsBySession[S1]?.values.subagentCatalog).toEqual([ { id: S2, createdAt: 1, mode: 'continuable', label: 'worker' }, @@ -1008,8 +1036,9 @@ describe('subagent catalogs', () => { const manager = makeManager(mock, remote) const read = manager.refreshProjections(S1) manager.handleControlFrame({ - type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 5, - value: [{ id: S2, createdAt: 1, mode: 'continuable', label: 'pushed' }], + type: 'projection', sessionId: S1, seq: 5, changes: [ + { key: 'subagentCatalog', val: [{ id: S2, createdAt: 1, mode: 'continuable', label: 'pushed' }] }, + ], }) response.resolve(ok({ kind: 'sequenced', asOfSeq: 3, values: { subagentCatalog: [] } })) await read @@ -1067,8 +1096,11 @@ describe('subagent catalogs', () => { remote.session.projections.mockImplementation(() => response.promise) const manager = makeManager(mock, remote) const read = manager.refreshProjections(S1) - manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 1, - value: [{ id: S2, createdAt: 1, mode: 'one-shot' }] }) + manager.handleControlFrame({ + type: 'projection', sessionId: S1, seq: 1, changes: [ + { key: 'subagentCatalog', val: [{ id: S2, createdAt: 1, mode: 'one-shot' }] }, + ], + }) response.resolve(ok(null)) await read expect(manager.getListSnapshot().projectionsBySession[S1]?.values.subagentCatalog).toEqual([ @@ -1117,9 +1149,12 @@ describe('subagent catalogs', () => { it('keeps completed unselected child metadata with its durable catalog row', ({ mock, remote }) => { const manager = makeManager(mock, remote) manager.handleSessionAdded(summary(S2, { origin: 'subagent', parentSessionId: S1, running: true })) - manager.handleControlFrame({ type: 'projection', sessionId: S2, key: 'title', value: 'finished child', seq: 2 }) - manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 1, - value: [{ id: S2, createdAt: 1, mode: 'one-shot' }] }) + manager.handleControlFrame({ type: 'projection', sessionId: S2, changes: [{ key: 'title', val: 'finished child' }], seq: 2 }) + manager.handleControlFrame({ + type: 'projection', sessionId: S1, seq: 1, changes: [ + { key: 'subagentCatalog', val: [{ id: S2, createdAt: 1, mode: 'one-shot' }] }, + ], + }) manager.handleSessionRemoved(S2) expect(manager.getListSnapshot().items[0]).toMatchObject({ title: 'finished child', running: false }) expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)?.running).toBe(false) @@ -1134,8 +1169,11 @@ describe('subagent catalogs', () => { it('retries a failed initial read without discarding pushed entries', async ({ mock, remote }) => { remote.session.projections.mockImplementation(() => Promise.resolve(err(new RemoteError('gateway/internal', 'offline', {})))) const manager = makeManager(mock, remote) - manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 2, - value: [{ id: S2, createdAt: 1, mode: 'one-shot' }] }) + manager.handleControlFrame({ + type: 'projection', sessionId: S1, seq: 2, changes: [ + { key: 'subagentCatalog', val: [{ id: S2, createdAt: 1, mode: 'one-shot' }] }, + ], + }) await manager.refreshProjections(S1) expect(manager.getListSnapshot().projectionsBySession[S1]).toMatchObject({ state: 'error', values: { subagentCatalog: [{ id: S2 }] } }) remote.session.projections.mockImplementation(() => Promise.resolve(ok({ kind: 'sequenced', asOfSeq: 2, values: { subagentCatalog: [] } }))) 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 ab1e185744..3c0d196c80 100644 --- a/packages/api/session-controller/tests/projection-store.client.spec.ts +++ b/packages/api/session-controller/tests/projection-store.client.spec.ts @@ -228,13 +228,13 @@ describe('manager frame routing', () => { it('lands projection frames before instantiation and the Session adopts the same store', ({ mock, remote }) => { const manager = makeManager(mock, remote) manager.handleControlFrame({ - type: 'projection', sessionId: sid('s1'), key: 'test/marks', value: { marks: ['early'] }, seq: 7, + type: 'projection', sessionId: sid('s1'), changes: [{ key: 'test/marks', val: { marks: ['early'] } }], seq: 7, }) const session = manager.get(sid('s1')) expect(session.projections.get('test/marks')).toEqual({ marks: ['early'] }) // Frames after instantiation land in the same store. manager.handleControlFrame({ - type: 'projection', sessionId: sid('s1'), key: 'test/marks', value: { marks: ['later'] }, seq: 9, + type: 'projection', sessionId: sid('s1'), changes: [{ key: 'test/marks', val: { marks: ['later'] } }], seq: 9, }) expect(session.projections.get('test/marks')).toEqual({ marks: ['later'] }) }) @@ -246,7 +246,7 @@ describe('manager frame routing', () => { })) await manager.refreshList() manager.handleControlFrame({ - type: 'projection', sessionId: sid('s1'), key: 'title', value: 'Projected title', seq: 4, + type: 'projection', sessionId: sid('s1'), changes: [{ key: 'title', val: 'Projected title' }], seq: 4, }) await Promise.resolve() expect(manager.getListSnapshot().items[0]?.title).toBe('Projected title') @@ -318,7 +318,7 @@ describe('manager frame routing', () => { // A frame past the block's watermark still advances it. manager.handleControlFrame({ - type: 'projection', sessionId: sid('s1'), key: 'title', value: 'Frame title', seq: 41, + type: 'projection', sessionId: sid('s1'), changes: [{ key: 'title', val: 'Frame title' }], seq: 41, }) await Promise.resolve() expect(manager.getListSnapshot().items[0]?.title).toBe('Frame title') @@ -342,8 +342,8 @@ describe('manager frame routing', () => { expect(manager.getListSnapshot().items[0]?.projectionValues).toBe(baseline) manager.handleControlFrame({ - type: 'projection', sessionId: sid('s1'), key: 'test/marks', - value: { marks: ['live'] }, seq: 3, + type: 'projection', sessionId: sid('s1'), changes: [{ key: 'test/marks', val: { marks: ['live'] } }], + seq: 3, }) await Promise.resolve() expect(manager.getListSnapshot().items[0]?.projectionValues) @@ -358,10 +358,12 @@ describe('manager frame routing', () => { })) await manager.refreshList() manager.handleControlFrame({ - type: 'projection', sessionId: sid('s1'), key: 'title', value: 'Doomed', seq: 4, + type: 'projection', sessionId: sid('s1'), changes: [{ key: 'title', val: 'Doomed' }], seq: 4, }) if (catalog !== undefined) { - manager.handleControlFrame({ type: 'projection', sessionId: sid('s1'), key: 'subagentCatalog', value: catalog, seq: 4 }) + manager.handleControlFrame({ + type: 'projection', sessionId: sid('s1'), changes: [{ key: 'subagentCatalog', val: catalog }], seq: 4, + }) } manager.handleSessionRemoved(sid('s1')) expect(manager.getListSnapshot().projectionsBySession[sid('s1')]).toBeUndefined() diff --git a/packages/api/session-controller/tests/remote-access.host.spec.ts b/packages/api/session-controller/tests/remote-access.host.spec.ts index 8be0dac78e..a11b157b1b 100644 --- a/packages/api/session-controller/tests/remote-access.host.spec.ts +++ b/packages/api/session-controller/tests/remote-access.host.spec.ts @@ -239,7 +239,7 @@ describe('Session Controller access', () => { for (;;) { const next = await iterator.next() if (next.done === true) throw new Error('control stream ended') - if (next.value.type === 'projection' && next.value.key === 'inbox') return next.value.sessionId + if (next.value.type === 'projection' && next.value.changes.some(change => change.key === 'inbox')) return next.value.sessionId } } // The owner sees the private Session's frame first; the guest's first Inbox frame is the shared one. diff --git a/packages/api/session-controller/tests/session-projections.host.spec.ts b/packages/api/session-controller/tests/session-projections.host.spec.ts index db37e8bcb8..89dc6d6963 100644 --- a/packages/api/session-controller/tests/session-projections.host.spec.ts +++ b/packages/api/session-controller/tests/session-projections.host.spec.ts @@ -360,7 +360,7 @@ describe('session.history projections block', () => { await new Promise(resolve => setTimeout(resolve, 0)) await expect(next).resolves.toMatchObject({ done: false, - value: { type: 'projection', key: 'sessionListMetadata' }, + value: { type: 'projection', changes: [{ key: 'sessionListMetadata' }] }, }) const extra = iterator.next() const quiet = Symbol('quiet') @@ -418,7 +418,7 @@ describe('session.history projections block', () => { const changed = await iterator.next() expect(changed).toMatchObject({ done: false, - value: { type: 'projection', key: 'sessionListMetadata' }, + value: { type: 'projection', changes: [{ key: 'sessionListMetadata' }] }, }) abort.abort() await iterator.return?.() @@ -752,16 +752,16 @@ describe('Session control projection frames', () => { return frames } - it('broadcasts changed view references with the causing seq and skips same-reference applies', async () => { + it('broadcasts one frame per event listing changed views in registration order and skips same-reference applies', async () => { const { ctx, session } = await harness(true) await ctx.sessionProjections.register(lastUserUnit()) const proxy = await remote(ctx) - // The controller's onChanged subscription lives in an inject child whose + // The controller's onEventChanged subscription lives in an inject child whose // fiber activates asynchronously; yield until it lands before appending. await new Promise(resolve => setTimeout(resolve, 0)) const abort = new AbortController() const stream = proxy.control(abort.signal) - const collected = collect(stream, 5, abort) + const collected = collect(stream, 3, abort) const now = vi.spyOn(Date, 'now').mockReturnValue(100) seedMessages(session, 1) @@ -773,24 +773,56 @@ describe('Session control projection frames', () => { now.mockRestore() const frames = await collected - const pushes = frames.filter( - (f): f is Extract => - f.type === 'projection' && f.key === 'test/last-user', - ) + const pushes = frames.filter((f): f is Extract => f.type === 'projection') expect(pushes).toEqual([ - { type: 'projection', sessionId: session.id, key: 'test/last-user', value: { text: 'm0' }, seq: 0 }, - { type: 'projection', sessionId: session.id, key: 'test/last-user', value: { text: 'm0' }, seq: 2 }, - ]) - expect(frames.filter( - (f): f is Extract => - f.type === 'projection' && f.key === 'sessionListMetadata', - )).toEqual([ - { type: 'projection', sessionId: session.id, key: 'sessionListMetadata', value: { blank: true, lastPromptAt: 100 }, seq: 0 }, - { type: 'projection', sessionId: session.id, key: 'sessionListMetadata', value: { blank: false, lastPromptAt: 100 }, seq: 1 }, - { type: 'projection', sessionId: session.id, key: 'sessionListMetadata', value: { blank: false, lastPromptAt: 300 }, seq: 2 }, + { + type: 'projection', sessionId: session.id, seq: 0, changes: [ + { key: 'test/last-user', val: { text: 'm0' } }, + { key: 'sessionListMetadata', val: { blank: true, lastPromptAt: 100 } }, + ], + }, + { + type: 'projection', sessionId: session.id, seq: 1, changes: [ + { key: 'sessionListMetadata', val: { blank: false, lastPromptAt: 100 } }, + ], + }, + { + type: 'projection', sessionId: session.id, seq: 2, changes: [ + { key: 'test/last-user', val: { text: 'm0' } }, + { key: 'sessionListMetadata', val: { blank: false, lastPromptAt: 300 } }, + ], + }, ]) // Frame seq aligns with the tail block's asOfSeq vocabulary (higher-seq-wins compatible). const tail = await opening(proxy, session.id) expect(tail.projections.asOfSeq).toBe(pushes.at(-1)?.seq) }) + + it('streams one observeProjections change per event for the observed Session only', async () => { + const { ctx, session } = await harness(true) + await ctx.sessionProjections.register(lastUserUnit()) + const proxy = await remote(ctx) + await new Promise(resolve => setTimeout(resolve, 0)) + const other = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } }) + const abort = new AbortController() + const changes = proxy.observeProjections({ sessionId: session.id }, abort.signal)[Symbol.asyncIterator]() + expect(await changes.next()).toEqual({ done: false, value: { type: 'ready' } }) + + const now = vi.spyOn(Date, 'now').mockReturnValue(100) + seedMessages(other, 1) + seedMessages(session, 1) + now.mockRestore() + + expect(await changes.next()).toEqual({ + done: false, + value: { + type: 'projection', seq: 0, changes: [ + { key: 'test/last-user', val: { text: 'm0' } }, + { key: 'sessionListMetadata', val: { blank: true, lastPromptAt: 100 } }, + ], + }, + }) + abort.abort() + expect(await changes.next()).toEqual({ done: true, value: 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 212320fdf6..4301be69d6 100644 --- a/packages/api/session-controller/tests/sessions-service.client.spec.ts +++ b/packages/api/session-controller/tests/sessions-service.client.spec.ts @@ -107,7 +107,7 @@ describe('list store projection', () => { it('projects durable titles separately from cwd/id display fallbacks and parent links', async ({ bench }) => { const b = bench() b.svc.handleControlFrame({ - type: 'projection', sessionId: sid('s1'), key: 'title', value: 'Durable title', seq: 2, + type: 'projection', sessionId: sid('s1'), changes: [{ key: 'title', val: 'Durable title' }], seq: 2, }) await feedList(b, [ { id: 's1', cwd: '/home/u/proj-a/' }, @@ -128,7 +128,7 @@ describe('list store projection', () => { expect(b.svc.list.getSnapshot().byId[sid('s1')]?.projectionValues?.agentPreset).toBe('standard') b.svc.handleControlFrame({ - type: 'projection', sessionId: sid('s1'), key: 'agentPreset', value: 'minimal', seq: 1, + type: 'projection', sessionId: sid('s1'), changes: [{ key: 'agentPreset', val: 'minimal' }], seq: 1, }) await Promise.resolve() @@ -784,7 +784,7 @@ describe('catalog-addressed navigation', () => { it(`loads an unretained one-shot child with projected title ${String(title)}`, async ({ bench }) => { const b = bench() if (title !== undefined) b.svc.handleControlFrame({ - type: 'projection', sessionId: sid('one-shot'), key: 'title', value: title, seq: 2, + type: 'projection', sessionId: sid('one-shot'), changes: [{ key: 'title', val: title }], seq: 2, }) b.mock.remote.session.projections.mockResolvedValue(ok({ kind: 'sequenced', asOfSeq: 0, values: { subagentCatalog: [{ id: sid('one-shot'), createdAt: 1, mode: 'one-shot' }] }, @@ -984,7 +984,7 @@ describe('fork', () => { ] as const)('increments the durable title %j after the child is published', async ([sourceTitle, childTitle], { bench }) => { const b = bench() b.svc.handleControlFrame({ - type: 'projection', sessionId: sid('source'), key: 'title', value: sourceTitle, seq: 2, + type: 'projection', sessionId: sid('source'), changes: [{ key: 'title', val: sourceTitle }], seq: 2, }) await feedList(b, [{ id: 'source', cwd: '/work' }]) b.mock.remote.session.fork.mockResolvedValue(ok({ sessionId: sid('child') })) @@ -1035,7 +1035,7 @@ describe('fork', () => { it('rejects child rename failure while preserving its catalog row without retaining the child', async ({ bench }) => { const b = bench() b.svc.handleControlFrame({ - type: 'projection', sessionId: sid('source'), key: 'title', value: 'Roadmap', seq: 2, + type: 'projection', sessionId: sid('source'), changes: [{ key: 'title', val: 'Roadmap' }], seq: 2, }) await feedList(b, [{ id: 'source' }]) b.mock.remote.session.fork.mockResolvedValue(ok({ sessionId: sid('child') })) diff --git a/packages/api/session-controller/tests/subagent-catalog.host.spec.ts b/packages/api/session-controller/tests/subagent-catalog.host.spec.ts index 36216a1f1c..0f834506d9 100644 --- a/packages/api/session-controller/tests/subagent-catalog.host.spec.ts +++ b/packages/api/session-controller/tests/subagent-catalog.host.spec.ts @@ -77,12 +77,15 @@ describe('SessionController subagent catalog', () => { version: 0, childId: CHILD, childCreatedAt: 8, mode: 'continuable', label: 'worker', }) let frame = await stream.next() - while (!frame.done && !(frame.value.type === 'projection' && frame.value.key === 'subagentCatalog' - && Array.isArray(frame.value.value) && frame.value.value.length === 1)) { + let catalog = frame.done || frame.value.type !== 'projection' + ? undefined : frame.value.changes.find(change => change.key === 'subagentCatalog')?.val + while (!frame.done && !(Array.isArray(catalog) && catalog.length === 1)) { frame = await stream.next() + catalog = frame.done || frame.value.type !== 'projection' + ? undefined : frame.value.changes.find(change => change.key === 'subagentCatalog')?.val } - expect(frame.value).toMatchObject({ type: 'projection', sessionId: PARENT, key: 'subagentCatalog', - value: [{ id: CHILD, createdAt: 8, mode: 'continuable', label: 'worker' }] }) + expect(frame.value).toMatchObject({ type: 'projection', sessionId: PARENT }) + expect(catalog).toMatchObject([{ id: CHILD, createdAt: 8, mode: 'continuable', label: 'worker' }]) } finally { signal.abort() await stream.return?.() diff --git a/packages/api/session-controller/tests/test-remote.ts b/packages/api/session-controller/tests/test-remote.ts index 145b48f5c3..04ed6788cf 100644 --- a/packages/api/session-controller/tests/test-remote.ts +++ b/packages/api/session-controller/tests/test-remote.ts @@ -62,6 +62,7 @@ import type { SessionSelectModelRequest, SessionSelectModelValue, SessionProjectionsRequest, + SessionProjectionFrame, SessionUpdateQueueRequest, SessionUpdateQueueValue, } from '../src/types.ts' @@ -90,6 +91,7 @@ export interface TestSessionRemote { page(request: SessionPageRequest, signal?: AbortSignal): Promise> projections(request: SessionProjectionsRequest, signal?: AbortSignal): Promise> follow(request: SessionFollowRequest, signal?: AbortSignal): AsyncIterable + observeProjections(request: SessionProjectionsRequest, signal?: AbortSignal): AsyncIterable control(signal?: AbortSignal): AsyncIterable } @@ -397,6 +399,7 @@ export async function createSessionTestRemote( signal, ), follow: (request, signal = new AbortController().signal) => direct.follow(request, signal), + observeProjections: (request, signal = new AbortController().signal) => direct.observeProjections(request, signal), control: (signal = new AbortController().signal) => direct.control(signal), } } diff --git a/packages/api/session-controller/tests/transport.client.spec.ts b/packages/api/session-controller/tests/transport.client.spec.ts index 2c60701f80..2568f3f807 100644 --- a/packages/api/session-controller/tests/transport.client.spec.ts +++ b/packages/api/session-controller/tests/transport.client.spec.ts @@ -669,7 +669,7 @@ describe('Session Client stream adapters', () => { value: { projections: {} }, } const update: SessionControlFrame = { - type: 'projection', sessionId: 'session-1' as never, key: 'title', value: 'updated', seq: 1, + type: 'projection', sessionId: 'session-1' as never, changes: [{ key: 'title', val: 'updated' }], seq: 1, } const remote = new ScriptedSessionRemote([], [], [baseline, update]) const accept = vi.fn<(frame: SessionControlFrame) => void>() diff --git a/packages/experimental/liveshare/src/host/viewer-fork.ts b/packages/experimental/liveshare/src/host/viewer-fork.ts index 1606a816e8..d85ddbf5c3 100644 --- a/packages/experimental/liveshare/src/host/viewer-fork.ts +++ b/packages/experimental/liveshare/src/host/viewer-fork.ts @@ -563,7 +563,7 @@ export function createViewerFork(input: ViewerForkDeps): ViewerFork { continue } // `session/control` carries one baseline block per Session the controller let this reader see. A later - // frame replaces one projection key, and this one is never replaced by the owner's own registry. + // frame replaces only the projection keys it lists, and the owner's own registry never lists this one. const value = isRecord(frame.value) ? frame.value : undefined const projections = value !== undefined && isRecord(value.projections) ? value.projections : undefined if (frame.type !== 'baseline' || value === undefined || projections === undefined) { diff --git a/packages/experimental/liveshare/tests/viewer-authority.spec.ts b/packages/experimental/liveshare/tests/viewer-authority.spec.ts index cf9451af0f..0476b5ea1c 100644 --- a/packages/experimental/liveshare/tests/viewer-authority.spec.ts +++ b/packages/experimental/liveshare/tests/viewer-authority.spec.ts @@ -68,7 +68,7 @@ describe('viewer authority', () => { const block = { asOfSeq: 0, values: {} } const read = reader({ sharedSessionId: () => 'source', forks, gateway: gateway([ { type: 'baseline', value: { projections: { source: block, 'alice-fork': { asOfSeq: 1, values: { title: 'Mine' } } } } }, - { type: 'projection', sessionId: 'alice-fork', key: 'title', value: 'Renamed', seq: 2 }, + { type: 'projection', sessionId: 'alice-fork', seq: 2, changes: [{ key: 'title', val: 'Renamed' }] }, ]) }) // The fork's facts are not folded from its log, so the baseline that replaces the block the session // list published states them again; the shared Session and later updates are relayed unchanged. @@ -77,7 +77,7 @@ describe('viewer authority', () => { source: block, 'alice-fork': { asOfSeq: 1, values: { title: 'Mine', forkLineage: forkLineageValue('source') } }, } } }, - { type: 'projection', sessionId: 'alice-fork', key: 'title', value: 'Renamed', seq: 2 }, + { type: 'projection', sessionId: 'alice-fork', seq: 2, changes: [{ key: 'title', val: 'Renamed' }] }, ]) }) }) diff --git a/packages/experimental/liveshare/tests/viewer-fixture.client.ts b/packages/experimental/liveshare/tests/viewer-fixture.client.ts index cce21efaa4..27d4682a83 100644 --- a/packages/experimental/liveshare/tests/viewer-fixture.client.ts +++ b/packages/experimental/liveshare/tests/viewer-fixture.client.ts @@ -407,9 +407,8 @@ function parseFixtureRemoteEventResult(value: unknown): FixtureRemoteEventResult interface FixtureProjectionFrame { readonly type: 'projection' readonly sessionId: SessionId - readonly key: string - readonly value: unknown readonly seq: number + readonly changes: readonly { readonly key: string; readonly val: unknown }[] } interface FixtureQuestionItem { @@ -1608,107 +1607,62 @@ function sameModelSelection(left: ModelSelection | null, right: ModelSelection | && left.reasoningEffort === right.reasoningEffort) } -/** Host parallel: emit one Session control projection frame per key advanced by the event. */ +/** Host parallel: emit one Session control projection frame listing every key advanced by the event. */ function projectionFramesOf( id: SessionId, log: readonly SessionEvent[], event: SessionEvent, ): FixtureProjectionFrame[] { + const changes = projectionChangesOf(log, event) + const keys = Object.keys(changes) + if (keys.length === 0) return [] + return [{ type: 'projection', sessionId: id, seq: event.seq, changes: keys.map(key => ({ key, val: changes[key] })) }] +} + +/** + * The new value of each projection key the event advances, in the order this + * fixture lists them; the Host lists keys in registration order. + */ +function projectionChangesOf(log: readonly SessionEvent[], event: SessionEvent): Record { const type = event.type - const frames: FixtureProjectionFrame[] = [] - if (type === 'model/selection' || type === 'request/header') { - frames.push({ - type: 'projection', - sessionId: id, - key: 'modelSelection', - value: modelSelectionProjectionOf(log), - seq: event.seq, - }) - } + const changes: Record = {} + if (type === 'model/selection' || type === 'request/header') changes['modelSelection'] = modelSelectionProjectionOf(log) // One usage sample advances both token-meter units. if (usageSampleOf(event) !== undefined) { - frames.push( - { type: 'projection', sessionId: id, key: 'tokenUsage', value: tokenUsageOf(log), seq: event.seq }, - { type: 'projection', sessionId: id, key: 'contextPressure', value: contextPressureOf(log), seq: event.seq }, - ) - } - if (type === 'request/context') { - frames.push({ - type: 'projection', - sessionId: id, - key: 'contextPressure', - value: contextPressureOf(log), - seq: event.seq, - }) + changes['tokenUsage'] = tokenUsageOf(log) + changes['contextPressure'] = contextPressureOf(log) } + if (type === 'request/context') changes['contextPressure'] = contextPressureOf(log) if (type === 'request/header' || type === 'system/message' || type === 'user/message' || type === 'assistant/message' || type === 'tool/result') { - frames.push({ - type: 'projection', - sessionId: id, - key: 'contextBreakdown', - value: contextBreakdownOf(log), - seq: event.seq, - }) + changes['contextBreakdown'] = contextBreakdownOf(log) } // The stats fold's view advances on message assembly and tool settlement // (wall times) and on step close (counts). - if (type === 'assistant/message' || type === 'tool/result' || type === 'step/end') { - frames.push({ - type: 'projection', - sessionId: id, - key: 'sessionStats', - value: sessionStatsOf(log), - seq: event.seq, - }) - } - if (frames.length > 0) return frames + if (type === 'assistant/message' || type === 'tool/result' || type === 'step/end') changes['sessionStats'] = sessionStatsOf(log) + if (Object.keys(changes).length > 0) return changes if (type === 'session/title') { const values = projectionValuesOf(log) /* v8 ignore next -- the advancing title event is in the log, so the key is present. */ - if (!Object.hasOwn(values, 'title')) return [] - return [{ type: 'projection', sessionId: id, key: 'title', value: values['title'], seq: event.seq }] + if (!Object.hasOwn(values, 'title')) return {} + return { title: values['title'] } } // The goal domain's own durable change advances its projection. - if (type === 'goal/change') { - return [{ type: 'projection', sessionId: id, key: 'goal', value: backscanGoal(log), seq: event.seq }] - } + if (type === 'goal/change') return { goal: backscanGoal(log) } // Standing-plan fold: writes replace the list; turn/start clears it (null). - if (type === 'todo/write' || type === 'turn/start') { - return [{ - type: 'projection', - sessionId: id, - key: 'todos', - value: backscanTodos(log) ?? null, - seq: event.seq, - }] - } + if (type === 'todo/write' || type === 'turn/start') return { todos: backscanTodos(log) ?? null } // Knob fold: any of the three whole-value knob events advances the select. - if (type === 'permission/preset' || type === 'sandbox/mode' || type === 'approval/policy') { - return [{ - type: 'projection', - sessionId: id, - key: 'permissions', - value: permissionSelectOf(log), - seq: event.seq, - }] - } + if (type === 'permission/preset' || type === 'sandbox/mode' || type === 'approval/policy') return { permissions: permissionSelectOf(log) } // The plan unit advances on its two folded event kinds when the command // lifecycle contains the input that represents a plan selection. if (type === 'plan/mode' || (event.type === 'command/run' && event.data.name === 'plan' && typeof event.data.args === 'string')) { - return [{ - type: 'projection', - sessionId: id, - key: 'plan', - value: planViewOf(log), - seq: event.seq, - }] + return { plan: planViewOf(log) } } - return [] + return {} } /** diff --git a/packages/extensions/tool-cordis/src/api-catalog.ts b/packages/extensions/tool-cordis/src/api-catalog.ts index f7a464f606..0175783793 100644 --- a/packages/extensions/tool-cordis/src/api-catalog.ts +++ b/packages/extensions/tool-cordis/src/api-catalog.ts @@ -2786,9 +2786,9 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ }, { signature: '@Remote({ mode: \'stream\' }) observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable', - description: 'Subscribe to sparse Session projection changes.', + description: 'Subscribe to Session projection changes.', parameters: [{ name: 'request', description: 'target Session.' }, { name: 'signal', description: 'subscription cancellation.' }], - returns: 'readiness and sparse projection changes.', + returns: 'readiness followed by one projection change frame per committed event.', }, { signature: '@Remote state(request: SessionStateRequest, signal: AbortSignal): Promise', @@ -2972,6 +2972,12 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ parameters: [{ name: 'listener', description: 'called once per client-visible unit whose raw view changed by `Object.is`, per committed event.' }], returns: 'the exact disposer that unsubscribes.', }, + { + signature: 'onEventChanged(listener: ProjectionEventChangeListener): () => void', + description: 'Subscribe to the event-level change feed. The registration is an effect on the calling context\'s fiber. The listener runs synchronously inside the committed event\'s `session/event` handling, once after every unit has been driven and after every onChanged call for that event. If a unit or an onChanged listener throws during that drive, the listener still receives the changes made before the throw.', + parameters: [{ name: 'listener', description: 'called once per committed event that changed at least one client-visible unit\'s raw view by `Object.is`.' }], + returns: 'the exact disposer that unsubscribes.', + }, { signature: 'snapshot( session: Session, keys?: ProjectionKeySelection, ): ProjectionSnapshot', description: 'One consistent cut over every registered client-visible unit for one session, read from the watermark cache (missing cells fold lazily over the in-memory log). Fully synchronous — every value and `asOfSeq` reflect the same log position. Each value passes its unit\'s `viewSchema` before leaving.', @@ -7635,6 +7641,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [ name: 'ProjectionDefinition', declaration: 'export interface ProjectionDefinition {\n key: K;\n stateSchema: ZodType;\n init(header: SessionHeader, inheritedEventCount: SessionLogOffset): NoInfer;\n apply(state: NoInfer, event: SessionEvent): NoInfer;\n wire?: K extends keyof SessionProjectionMap ? {\n viewSchema: ZodType>;\n view(state: NoInfer): SessionProjectionViewValue;\n summary?: boolean;\n } : never;\n stateVersion: number;\n}', }, + { + name: 'ProjectionEventChangeListener', + declaration: 'export type ProjectionEventChangeListener = (session: Session, changes: readonly {\n readonly key: Extract;\n readonly val: unknown;\n}[], seq: SessionSeq) => void;', + }, { name: 'ProjectionHandle', declaration: 'export interface ProjectionHandle {\n stateOf(session: Session): S;\n readerFor(session: Session): () => S;\n dispose(): Promise;\n}', @@ -8585,7 +8595,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [ }, { name: 'SessionProjectionUpdate', - declaration: 'export interface SessionProjectionUpdate {\n readonly sessionId: SessionId;\n readonly key: string;\n readonly value: JsonValue;\n readonly seq: number;\n}', + declaration: 'export interface SessionProjectionUpdate {\n readonly sessionId: SessionId;\n readonly seq: number;\n readonly changes: readonly {\n readonly key: string;\n readonly val: SessionProjectionValue;\n }[];\n}', }, { name: 'SessionProjectionValue', diff --git a/packages/session/session-projection/README.i18n.yaml b/packages/session/session-projection/README.i18n.yaml index 7b4fd598e0..93f6b54a6f 100644 --- a/packages/session/session-projection/README.i18n.yaml +++ b/packages/session/session-projection/README.i18n.yaml @@ -24,8 +24,8 @@ en: 09758d1475fad86e zh: 473637cb34aebe72 /deepseek-ai-dsh-session-projection/use-this-package/register-and-read: - en: f64bc8038aecfcae - zh: fb5fb0999468fac0 + en: 84f39e64f252a4d3 + zh: e3f6dff8c5df9016 /deepseek-ai-dsh-session-projection/use-this-package/bind-several-states-to-one-session: en: dac4ea4d67ce057d zh: 1cf3e5c7af5a5203 @@ -48,8 +48,8 @@ en: c19f656b022506fa zh: fdcba0083f08a76b /deepseek-ai-dsh-session-projection/understand-the-implementation/drive-and-checkpoint-flow: - en: 00136880bb586403 - zh: bd35f645111c068c + en: 6d38875ac6f20b38 + zh: cc4fb064642f30d9 /deepseek-ai-dsh-session-projection/further-exploration: en: 304a03d08ea5c9f9 zh: 200a0505e8d11e0d diff --git a/packages/session/session-projection/README.md b/packages/session/session-projection/README.md index 6882d93515..a8dcdaebff 100644 --- a/packages/session/session-projection/README.md +++ b/packages/session/session-projection/README.md @@ -55,7 +55,7 @@ const definition = { ### Register and read -`await register(definition)` infers the host state type from `stateSchema`, installs the unit, and returns a handle with synchronous `stateOf(session)` reads and `dispose(): Promise`. Await `dispose()` to release the registration; repeated calls are safe. Registrants with the same key and `stateVersion` share cells; incompatible or invalid versions reject registration. Registration belongs to the calling fiber. The last unregister removes the unit from global event drives, snapshots, checkpoints, and change notifications. Existing handles remain readable: each retains its original unit and folds new Session events on demand, even if another version later registers the same key. Reading a handle does not initialize unrelated units or register it again. Carriers use `snapshot(session)` for a consistent client-view cut and `onChanged(listener)` for changes to registered units. +`await register(definition)` infers the host state type from `stateSchema`, installs the unit, and returns a handle with synchronous `stateOf(session)` reads and `dispose(): Promise`. Await `dispose()` to release the registration; repeated calls are safe. Registrants with the same key and `stateVersion` share cells; incompatible or invalid versions reject registration. Registration belongs to the calling fiber. The last unregister removes the unit from global event drives, snapshots, checkpoints, and change notifications. Existing handles remain readable: each retains its original unit and folds new Session events on demand, even if another version later registers the same key. Reading a handle does not initialize unrelated units or register it again. Carriers use `snapshot(session)` for a consistent client-view cut, `onChanged(listener)` for one call per changed unit, and `onEventChanged(listener)` for one call per committed event that lists every client-visible unit whose view it changed. ```text const todoProjectionHandle = await ctx.sessionProjections.register(definition) @@ -128,7 +128,7 @@ The package is the Service Definition and drive role of a capability seam: the f ### Drive and checkpoint flow -One committed event drives every registered unit in registration order; a client-visible unit whose raw view changes by `Object.is` notifies the change feed with its schema-validated view and the causing seq. The live drive retains its previous and current raw views; snapshots and cold reads remain complete independent reads. `checkpoint(session)` returns one detached `(key → {ver, seq, val})` row per unit for the persisted cache; `restoreFloor` anchors a tail read one event below the lowest usable watermark so a shrunk log is detected, and `restore` refolds persisted rows over a stored suffix, discarding any row whose `ver` does not match or that claims events past the stored end. +One committed event drives every registered unit in registration order; a client-visible unit whose raw view changes by `Object.is` notifies each `onChanged` listener with its schema-validated view and the causing seq. After every unit has been driven, each `onEventChanged` listener receives one call with one `{ key, val }` entry per changed client-visible unit in registration order, where `val` is the schema-validated view, and the causing seq; an event that changes no client-visible view makes no event-level call. If a unit or an `onChanged` listener throws during the drive, event listeners still receive the changes made before the throw, and the error then propagates to the Session's listener containment. The live drive retains its previous and current raw views; snapshots and cold reads remain complete independent reads. `checkpoint(session)` returns one detached `(key → {ver, seq, val})` row per unit for the persisted cache; `restoreFloor` anchors a tail read one event below the lowest usable watermark so a shrunk log is detected, and `restore` refolds persisted rows over a stored suffix, discarding any row whose `ver` does not match or that claims events past the stored end. diff --git a/packages/session/session-projection/README.zh.md b/packages/session/session-projection/README.zh.md index a3398965f6..744af01bed 100644 --- a/packages/session/session-projection/README.zh.md +++ b/packages/session/session-projection/README.zh.md @@ -55,7 +55,7 @@ const definition = { ### 注册与读取 -`await register(definition)` 从 `stateSchema` 推导宿主状态类型，安装单元，并返回提供同步 `stateOf(session)` 读取和 `dispose(): Promise` 的 handle。等待 `dispose()` 完成即可释放注册；重复调用是安全的。具有相同 key 和 `stateVersion` 的注册方共享 cell；版本不兼容或非法时注册会拒绝。注册归调用方 fiber 所有。最后一个注册方注销后，单元退出全局事件驱动、快照、检查点和变更通知。已有 handle 仍可读取：它保留原有单元，按需折叠新增的 Session 事件，即使同一 key 后来注册了其他版本也不受影响。读取 handle 不会初始化无关单元，也不会重新注册。载体用 `snapshot(session)` 读取一致的客户端视图切面，用 `onChanged(listener)` 订阅已注册单元的变更。 +`await register(definition)` 从 `stateSchema` 推导宿主状态类型，安装单元，并返回提供同步 `stateOf(session)` 读取和 `dispose(): Promise` 的 handle。等待 `dispose()` 完成即可释放注册；重复调用是安全的。具有相同 key 和 `stateVersion` 的注册方共享 cell；版本不兼容或非法时注册会拒绝。注册归调用方 fiber 所有。最后一个注册方注销后，单元退出全局事件驱动、快照、检查点和变更通知。已有 handle 仍可读取：它保留原有单元，按需折叠新增的 Session 事件，即使同一 key 后来注册了其他版本也不受影响。读取 handle 不会初始化无关单元，也不会重新注册。载体用 `snapshot(session)` 读取一致的客户端视图切面，用 `onChanged(listener)` 按每个变化的单元接收一次调用，用 `onEventChanged(listener)` 按每个已提交事件接收一次调用，列出该事件改变了视图的全部客户端可见单元。 ```text const todoProjectionHandle = await ctx.sessionProjections.register(definition) @@ -128,7 +128,7 @@ Session 摘要包含每个已注册客户端视图，但排除 `SessionProjectio ### 驱动与检查点流程 -一个已提交事件按注册顺序驱动每个已注册单元；原始 view 通过 `Object.is` 判定为变化的客户端可见单元会以经 schema 校验的视图与致因 seq 通知变更流。live drive 保留前后两个原始 view；snapshot 与冷读仍是彼此独立的完整读取。`checkpoint(session)` 为持久缓存返回每个单元一份独立的 `(key → {ver, seq, val})` 行；`restoreFloor` 把尾部读取锚定在最低可用水位之前一个事件处，使缩短的日志可被检出；`restore` 把持久行在存储后缀上重新折叠，丢弃任何 `ver` 不匹配或声称越过存储末尾的行。 +一个已提交事件按注册顺序驱动每个已注册单元；原始 view 通过 `Object.is` 判定为变化的客户端可见单元会以经 schema 校验的视图与致因 seq 通知每个 `onChanged` 监听器。所有单元驱动完成后，每个 `onEventChanged` 监听器收到一次调用，携带按注册顺序排列、每个变化的客户端可见单元一项的 `{ key, val }`（`val` 为经 schema 校验的视图），以及致因 seq；未改变任何客户端可见视图的事件不产生事件级调用。驱动中某个单元或 `onChanged` 监听器抛错时，事件级监听器仍会收到抛错前已产生的变化，随后错误继续抛给 Session 的监听器兜底处理。live drive 保留前后两个原始 view；snapshot 与冷读仍是彼此独立的完整读取。`checkpoint(session)` 为持久缓存返回每个单元一份独立的 `(key → {ver, seq, val})` 行；`restoreFloor` 把尾部读取锚定在最低可用水位之前一个事件处，使缩短的日志可被检出；`restore` 把持久行在存储后缀上重新折叠，丢弃任何 `ver` 不匹配或声称越过存储末尾的行。 diff --git a/packages/session/session-projection/src/index.ts b/packages/session/session-projection/src/index.ts index 63342e374b..4ce8408a90 100644 --- a/packages/session/session-projection/src/index.ts +++ b/packages/session/session-projection/src/index.ts @@ -142,6 +142,19 @@ export type ProjectionChangeListener = ( seq: SessionSeq, ) => void +/** + * Event-level change-feed listener: one committed event changed the raw `view` + * result of at least one client-visible unit by `Object.is` for one session. + * `changes` holds one entry per such unit in registration order, each with the + * unit's `key` and its schema-validated view as `val`; `seq` is the seq of the + * causing event. + */ +export type ProjectionEventChangeListener = ( + session: Session, + changes: readonly { readonly key: Extract; readonly val: unknown }[], + seq: SessionSeq, +) => void + /** * One read's client-visible key selection: an explicit key list, or a derived * set such as {@link SessionProjectionRegistry.summaryKeys}. A set is used as @@ -261,6 +274,7 @@ export class SessionProjectionRegistry extends Service { /** Stable summary key set; dropped whenever the key table changes. */ private summaryKeySet: ReadonlySet> | undefined private readonly listeners = new Set() + private readonly eventListeners = new Set() /** Keys whose deprecated registry-level read has already been reported. */ private readonly deprecatedStateOfKeys = new Set() @@ -456,6 +470,26 @@ export class SessionProjectionRegistry extends Service { return () => void dispose() } + /** + * Subscribe to the event-level change feed. The registration is an effect on + * the calling context's fiber. The listener runs synchronously inside the + * committed event's `session/event` handling, once after every unit has been + * driven and after every {@link onChanged} call for that event. If a unit or + * an {@link onChanged} listener throws during that drive, the listener still + * receives the changes made before the throw. + * @param listener - called once per committed event that changed at least one client-visible unit's raw view by `Object.is`. + * @returns the exact disposer that unsubscribes. + */ + onEventChanged(listener: ProjectionEventChangeListener): () => void { + const dispose = this.ctx.effect(() => { + this.eventListeners.add(listener) + return () => { + this.eventListeners.delete(listener) + } + }, 'sessionProjections.onEventChanged()') + return () => void dispose() + } + /** * One consistent cut over every registered client-visible unit for one session, read from * the watermark cache (missing cells fold lazily over the in-memory log). @@ -807,53 +841,70 @@ export class SessionProjectionRegistry extends Service { } } - /** Eager drive: pass one committed event through every unit; notify on changed raw view references. */ + /** + * Eager drive: pass one committed event through every unit; notify per-key + * listeners on each changed raw view reference, then event listeners once. + * When a unit or a per-key listener throws, event listeners still receive + * the changes collected before the throw, and the error then propagates. + */ private drive(session: Session, event: SessionEvent): void { - for (const registration of this.registrations.values()) { - let cell = registration.cells.get(session) - if (cell !== undefined && cell.observedSeq >= event.seq) continue - if (cell === undefined) { - // Late build mid-stream: fold history before this event (seq = log - // index, so the prefix slice is exact), then take the normal gate. - cell = this.buildCell( - registration.def, - session.header, - session.inheritedEventCount, - // oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred. - session.snapshotEvents(SessionLogOffset(0), SessionLogOffset(event.seq)), - ) - registration.cells.set(session, cell) - } else { - this.advanceCell( - registration.def, - cell, - session, - event.seq === 0 ? -1 : SessionSeq(event.seq - 1), - ) - } - const previousState = cell.state - const next = registration.def.apply(previousState, event) - const changed = !Object.is(next, previousState) - cell.state = next - cell.observedSeq = event.seq - const wire = registration.def.wire - if (changed && wire !== undefined) { - const views = cell.views - views[0] = views[1] - if (this.listeners.size > 0) { - views[1] = wire.view(next) - if (!Object.is(views[0], views[1])) { - const value = wire.viewSchema.parse(views[1]) - for (const listener of this.listeners) { - listener(session, registration.def.key as Extract, value, event.seq) - } - } + const observed = this.listeners.size > 0 || this.eventListeners.size > 0 + const changes: { key: Extract; val: unknown }[] = [] + try { + for (const registration of this.registrations.values()) { + let cell = registration.cells.get(session) + if (cell !== undefined && cell.observedSeq >= event.seq) continue + if (cell === undefined) { + // Late build mid-stream: fold history before this event (seq = log + // index, so the prefix slice is exact), then take the normal gate. + cell = this.buildCell( + registration.def, + session.header, + session.inheritedEventCount, + // oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred. + session.snapshotEvents(SessionLogOffset(0), SessionLogOffset(event.seq)), + ) + registration.cells.set(session, cell) } else { - views[1] = undefined + this.advanceCell( + registration.def, + cell, + session, + event.seq === 0 ? -1 : SessionSeq(event.seq - 1), + ) + } + const previousState = cell.state + const next = registration.def.apply(previousState, event) + const changed = !Object.is(next, previousState) + cell.state = next + cell.observedSeq = event.seq + const wire = registration.def.wire + if (changed && wire !== undefined) { + const views = cell.views + views[0] = views[1] + if (observed) { + views[1] = wire.view(next) + if (!Object.is(views[0], views[1])) { + const key = registration.def.key as Extract + const value = wire.viewSchema.parse(views[1]) + changes.push({ key, val: value }) + for (const listener of this.listeners) { + listener(session, key, value, event.seq) + } + } + } else { + views[1] = undefined + } + } + // An unchanged state keeps its current view as the valid comparison + // value for the next state change. + } + } finally { + if (changes.length > 0) { + for (const listener of this.eventListeners) { + listener(session, changes, event.seq) } } - // An unchanged state keeps its current view as the valid comparison - // value for the next state change. } } diff --git a/packages/session/session-projection/tests/registry.spec.ts b/packages/session/session-projection/tests/registry.spec.ts index df2d9c26e9..a761519e08 100644 --- a/packages/session/session-projection/tests/registry.spec.ts +++ b/packages/session/session-projection/tests/registry.spec.ts @@ -234,6 +234,89 @@ describe('SessionProjectionRegistry drive', () => { expect(seen).toEqual([{ key: 'test/marks', value: { marks: ['a'] }, seq: event.seq, sessionId: String(session.id) }]) }) + it('notifies onEventChanged once per event with every changed key in registration order', async () => { + const { ctx, session } = await harness() + await ctx.sessionProjections.register(marksUnit()) + await ctx.sessionProjections.register(countUnit()) + // A distinct but reference-stable view, so the two entries carry different values. + const upper = new WeakMap() + await ctx.sessionProjections.register(stableViewUnit((state) => { + const cached = upper.get(state.value) + if (cached !== undefined) return cached + const view = { marks: state.value.marks.map(item => item.toUpperCase()) } + upper.set(state.value, view) + return view + })) + const order: string[] = [] + const events: { changes: readonly { key: string; val: unknown }[]; seq: SessionSeq; sessionId: string }[] = [] + ctx.sessionProjections.onChanged((_session, key) => { order.push(`key:${key}`) }) + ctx.sessionProjections.onEventChanged((changedSession, changes, seq) => { + order.push('event') + events.push({ changes, seq, sessionId: String(changedSession.id) }) + }) + + const event = mark(session, ['a']) + // Host-only and same-reference view changes publish nothing. + session.append('turn/start', { turn: 1 }) + + expect(events).toEqual([{ + changes: [{ key: 'test/marks', val: { marks: ['a'] } }, { key: 'test/stable-view', val: { marks: ['A'] } }], + seq: event.seq, + sessionId: String(session.id), + }]) + expect(order).toEqual(['key:test/marks', 'key:test/stable-view', 'event']) + }) + + it('still delivers the changes made before a later unit throws', async () => { + const { ctx, session } = await harness() + await ctx.sessionProjections.register(marksUnit()) + // A fresh invalid view on every change: its schema parse throws after test/marks was collected. + await ctx.sessionProjections.register(stableViewUnit(() => ({ marks: 1 }) as never)) + const events: { changes: readonly { key: string; val: unknown }[]; seq: SessionSeq }[] = [] + ctx.sessionProjections.onEventChanged((_session, changes, seq) => { + events.push({ changes, seq }) + }) + + const event = mark(session, ['a']) + + expect(events).toEqual([{ changes: [{ key: 'test/marks', val: { marks: ['a'] } }], seq: event.seq }]) + }) + + it('computes views for an event listener alone and stops after its disposer runs', async () => { + const { ctx, session } = await harness() + const view = vi.fn((state: StableViewState) => state.value) + await ctx.sessionProjections.register(stableViewUnit(view)) + const seen: { keys: readonly string[]; seq: SessionSeq }[] = [] + const stop = ctx.sessionProjections.onEventChanged((_session, changes, seq) => { + seen.push({ keys: changes.map(change => change.key), seq }) + }) + + const event = session.append('turn/start', { turn: 1 }) + expect(view).toHaveBeenCalledOnce() + expect(seen).toEqual([{ keys: ['test/stable-view'], seq: event.seq }]) + + stop() + mark(session, ['after-stop']) + expect(seen).toHaveLength(1) + expect(view).toHaveBeenCalledOnce() + }) + + it('removes event-level change listeners when their owning fiber unloads (HMR safety)', async () => { + const { ctx, session } = await harness() + await ctx.sessionProjections.register(marksUnit()) + const notifications: (readonly string[])[] = [] + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + inner.sessionProjections.onEventChanged((_session, changes) => { + notifications.push(changes.map(change => change.key)) + }) + }, { inject: ['sessionProjections'] })) + mark(session, ['live']) + await fiber.dispose() + mark(session, ['after-dispose']) + expect(notifications).toEqual([['test/marks']]) + expect(ctx.sessionProjections.snapshot(session).values).toEqual({ 'test/marks': { marks: ['after-dispose'] } }) + }) + it('does not compute a view while no change listener exists', async () => { const { ctx, session } = await harness() const view = vi.fn((state: StableViewState) => state.value) diff --git a/scripts/gen-cordis-catalog.ts b/scripts/gen-cordis-catalog.ts index 67f9752d2c..24abebb2e5 100644 --- a/scripts/gen-cordis-catalog.ts +++ b/scripts/gen-cordis-catalog.ts @@ -939,6 +939,7 @@ export const LINK_MAP: Readonly> = { SessionProjectionMap: 'session-projection.md', ProjectionKeySelection: 'session-projection.md', ProjectionChangeListener: 'session-projection.md', + ProjectionEventChangeListener: 'session-projection.md', ProjectionSnapshot: 'session-projection.md', ProjectionCheckpoint: 'session-projection.md', DirectoryPickerCapability: 'workspace.md', diff --git a/scripts/no-unknown-casts.baseline.json b/scripts/no-unknown-casts.baseline.json index b963880da1..d23c51577c 100644 --- a/scripts/no-unknown-casts.baseline.json +++ b/scripts/no-unknown-casts.baseline.json @@ -217,8 +217,7 @@ "e9721557773cba96e50cea1b0c6737991a9fb2490bcd83e934311e46310d8b68": 1 }, "packages/api/session-controller/tests/control-queue.host.spec.ts": { - "9b2258e1e99500b30d66b0b93e0f0bc5077278877903fc8cc6712167411f6109": 1, - "f2fe8bd862649a2492b7f99f0c35e78ba051a248f339d248d0dcfab163c69752": 1 + "9b2258e1e99500b30d66b0b93e0f0bc5077278877903fc8cc6712167411f6109": 1 }, "packages/api/session-controller/tests/controller.host.spec.ts": { "40725eab4d86929b16d87c88d5cf83168035def78b4d901613dd0369876f0731": 3, diff --git a/scripts/type-equiv.manifest.json b/scripts/type-equiv.manifest.json index 2f2713a8ec..a248a29e29 100644 --- a/scripts/type-equiv.manifest.json +++ b/scripts/type-equiv.manifest.json @@ -2136,6 +2136,11 @@ "symbol": "ProjectionChangeListener", "source": "packages/session/session-projection/src/index.ts" }, + { + "doc": "docs/subsystems/session-projection.md", + "symbol": "ProjectionEventChangeListener", + "source": "packages/session/session-projection/src/index.ts" + }, { "doc": "docs/subsystems/core.md", "symbol": "AgentHandle",