fix(session-projection): compare observed live views
This commit is contained in:
imccyu
parent
12bef3b577
commit
6f0daff1dd
14 changed files with 160 additions and 196 deletions
|
|
@ -2,5 +2,5 @@
|
|||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write .agents/notes/implemented/feature/2026-08-30-web-turn-rail-outline-jump.md
|
||||
2026-08-30-web-turn-rail-outline-jump.md: c1f190832ebfb83200d8f7cb5c01b9a45167ccab
|
||||
2026-08-30-web-turn-rail-outline-jump.zh.md: 66196430ce095e822a5e05863f591ea5cebce570
|
||||
2026-08-30-web-turn-rail-outline-jump.md: af2a464513d108429859c4f9ed8b4bfc58175e3c
|
||||
2026-08-30-web-turn-rail-outline-jump.zh.md: b63b1c1f656689997b8a8266ee332dfdb47d4035
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ Three cooperating pieces, each useful alone.
|
|||
|
||||
**Data: the `turnOutline` session projection.** `packages/session/session-turn-outline` registers one pure fold on `ctx.sessionProjections`: every `turn/start` appends an entry (skipping boundaries that do not advance the turn number, keeping the outline strictly increasing), the turn's first human `user/message` fills the prompt preview, and the newest text-bearing `assistant/message` buffers a response draft that `turn/end` commits (`turn/end` itself carries no text). Preview budgets mirror the rail card's clamps — one prompt line at 50 characters, up to three response lines at 120, an ellipsis marking a clip — and match the loaded-turn previews so a turn shows the same words before and after its events load. The wire value is the bare entry array so draft-only state changes keep its identity, and the feed's identity gate (below) then holds pushes to three per turn: boundary, prompt, settled response. The value rides the existing projection carriers — tail-page seed, `session/projection` control frames, projcache — and the web-app bundle mounts the plugin. `seq` is the `turn/start` event seq: the loop logs it before the turn's prompt and steps, so paging a window back through that seq loads the whole turn.
|
||||
|
||||
**Change-feed identity gate (session-projection).** The registry's change feed previously fired on every changed state reference of a client-visible unit; it now also compares the raw `view` output of the previous and next states and stays quiet when `Object.is`-identical; views are memoized by state object identity (review moved this off a stored last-delivered value, then off per-step recomputation), so each distinct state's view computes once and the memo — keyed by the state itself, not by what the feed delivered — cannot go stale across listener generations. This is what lets a unit buffer working fields (the response draft) in state behind an identity-stable projection instead of pushing its whole value per streamed assistant message; units whose views build fresh objects per call are unaffected. The alternative — value-equality dedup in the carrier by serialized comparison — was rejected as it pays a full serialization per quiet change.
|
||||
**Change-feed identity gate (session-projection).** Each live unit cell keeps `[previousView, currentView]` raw outputs. When a state reference changes, the drive shifts current to previous; while a change listener exists it computes `view(nextState)` once, stores current, and emits only when the two outputs differ by `Object.is`. Without listeners, current becomes `undefined` without evaluating `view`, so the first later computed value publishes conservatively; catch-up folding invalidates current in the same way. This lets a unit buffer working fields (the response draft) in state behind an identity-stable projection instead of pushing its whole value per streamed assistant message; units whose views build fresh objects per call are unaffected. The alternative — value-equality dedup in the carrier by serialized comparison — is rejected because every quiet change would pay for full serialization.
|
||||
|
||||
**Paging: `Session.loadThrough(seq)`.** The session-controller client gains a jump loader beside `loadOlder()`: it loops the existing prepend pager in 200-message pages (`JUMP_PAGE_MESSAGES`) until `baseSeq <= seq`, lowers a shared low-water target when called again mid-jump, stops on a page that leaves `baseSeq` unmoved (the no-progress guard against an empty page still claiming history), and reports busy through the existing `loadingOlder` snapshot bit. No wire change: seqs are dense, so the client computes everything from `beforeSeq` arithmetic.
|
||||
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ Web 聊天的轮次导航栏从已加载的事件窗口推导刻度，而窗口
|
|||
|
||||
**数据：`turnOutline` 会话投影。** `packages/session/session-turn-outline` 在 `ctx.sessionProjections` 上注册一个纯 fold：每个 `turn/start` 追加一个条目（跳过未推进轮次号的边界，保持大纲严格递增），该轮首条人类 `user/message` 填入提示词预览，最新一条带文本的 `assistant/message` 缓冲为回复草稿、由 `turn/end` 提交（`turn/end` 自身不带文本）。预览预算对齐导航卡片的截断——提示词一行 50 字符、回复至多三行 120 字符、被裁剪时补省略号——并与已加载轮次的预览一致，同一轮在事件载入前后显示相同的文字。wire 值是裸条目数组，纯草稿的状态变化因此保持其身份，配合下述变更流身份门把推送压到每轮三次：开轮、提示词、落定回复。值搭现有投影载体——尾页 seed、`session/projection` 控制帧、projcache——web-app bundle 挂载该插件。`seq` 是 `turn/start` 事件的 seq：loop 先记它再记该轮的提示词与步骤，窗口向后分页越过该 seq 即载入整轮。
|
||||
|
||||
**变更流身份门（session-projection）。** 注册表的变更流此前对客户端可见单元的每次状态引用变化都触发；现在还会把前后两个状态的原始 `view` 输出相互比较，`Object.is` 相同即保持安静；视图按状态对象身份做备忘（评审先从「存上一次交付值」改为现算、再改为按状态键缓存），每个不同状态只算一次，备忘的键是状态本身而非交付历史，监听器换代也无从拿到过期基线。正是它让单元可以把工作字段（回复草稿）缓冲在身份稳定投影之后的状态里，而不是每条流式助手消息都推送整值；view 每次新建对象的单元不受影响。备选——在载体侧按序列化比较去重——被否决，因为每次安静变化都要付一次完整序列化。
|
||||
**变更流身份门（session-projection）。** 每个实时单元 cell 保存 `[previousView, currentView]` 原始输出。state 引用变化时，drive 先把 current 移到 previous；存在变更 listener 时只计算一次 `view(nextState)` 并写入 current，两个输出通过 `Object.is` 判定为不同时才发出通知。没有 listener 时不计算 `view`，而是把 current 写成 `undefined`，因此之后首次计算出的值会保守地发布；补折叠也以相同方式使 current 失效。这让单元可以把工作字段（回复草稿）缓冲在身份稳定投影之后的 state 里，而不是每条流式助手消息都推送整值；view 每次新建对象的单元不受影响。备选——在载体侧按序列化比较去重——被否决，因为每次安静变化都要付一次完整序列化。
|
||||
|
||||
**分页：`Session.loadThrough(seq)`。** session-controller 客户端在 `loadOlder()` 旁新增跳转加载器：按 200 条 message 一页（`JUMP_PAGE_MESSAGES`）循环现有 prepend 分页器直到 `baseSeq <= seq`，跳转中再次调用会下调共享低水位目标，遇到 `baseSeq` 未动的页即停（对空页仍声称有历史的无进展守卫），忙碌状态复用现有 `loadingOlder` 快照位。零 wire 改动：seq 稠密，客户端仅凭 `beforeSeq` 算术即可。
|
||||
|
||||
|
|
|
|||
|
|
@ -2,5 +2,5 @@
|
|||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write docs/subsystems/session-projection.md
|
||||
session-projection.md: 2a08f1a7131094b6049ecd9f5d43df448bc6dbf3
|
||||
session-projection.zh.md: 73c1f02b897f5ac5408e3199096876d3ba438224
|
||||
session-projection.md: b5bacc4846a9a9709bb8c102aaa78810a2752110
|
||||
session-projection.zh.md: 36653721a125ceef13bff8b1a4562e7981a45696
|
||||
|
|
|
|||
|
|
@ -47,7 +47,10 @@ interface ProjectionDefinition<
|
|||
/** Validates the wire payload before it leaves the host. */
|
||||
viewSchema: ZodType<SessionProjectionMap[K]>
|
||||
/**
|
||||
* State → wire payload (the read-side projection).
|
||||
* State → wire payload (the read-side projection). The live drive keeps
|
||||
* the two latest raw results and compares them with `Object.is`; an
|
||||
* object-valued view must reuse its reference to suppress publication
|
||||
* across internal-only state changes.
|
||||
* @param state - the current state.
|
||||
* @returns the whole current value for this unit's key.
|
||||
*/
|
||||
|
|
@ -83,12 +86,9 @@ interface ProjectionSnapshot {
|
|||
|
||||
```ts type-equiv
|
||||
/**
|
||||
* Change-feed listener: one unit's served value changed for one session.
|
||||
* `value` is the schema-validated `view` output; `seq` is the unit's
|
||||
* watermark at emission (the seq of the event that caused the change). A
|
||||
* changed state whose raw `view` output is `Object.is`-identical to the
|
||||
* unit's previous projection does not fire, so a unit can buffer working
|
||||
* fields in state behind an identity-stable projection.
|
||||
* Change-feed listener: one unit's raw `view` result changed by `Object.is`
|
||||
* for one session. `value` is the schema-validated output; `seq` is the
|
||||
* unit's watermark at emission (the seq of the event that caused the change).
|
||||
*/
|
||||
type ProjectionChangeListener = (
|
||||
session: Session,
|
||||
|
|
@ -98,7 +98,7 @@ 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. `stateOf(session, key)` reads one live host state without computing unrelated views; callers must not mutate the borrowed reference. The change feed fires once per client-visible unit whose state *reference* changed for each committed event — unless the raw `view` output is `Object.is`-identical to the previous state's (views are memoized by state object identity — one computation per distinct state, and no last-delivered record exists to go stale), so a unit can buffer working fields in state behind an identity-stable projection and a later listener generation still sees every value transition; `apply` must return the same reference when its state did not change.
|
||||
`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. `stateOf(session, key)` 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.
|
||||
|
||||
## The registry: `ctx.sessionProjections`
|
||||
|
||||
|
|
@ -180,7 +180,7 @@ Source: [`packages/session/session-projection-cache/src/index.ts`](../../package
|
|||
|
||||
### `ctx.sessionProjections` — `SessionProjectionRegistry`
|
||||
|
||||
`ctx.sessionProjections`: the projection unit table and its drive. The service subscribes to `session/event` once; every committed event passes every registered unit's `apply` (eager drive), and a changed state reference in a client-visible unit notifies the change feed with the schema-validated view — unless the raw view output is `Object.is`-identical to the unit's previous projection (identity-stable projections stay quiet). Views are memoized by state object identity, so each distinct state's view computes once and no last-delivered record exists to go stale across listener generations. Cells build lazily — a unit registered after events flowed, or a session older than the registry, folds `init` over the in-memory log on first touch (event or read). Registration is an effect (disposer rides the calling fiber): an unloaded domain plugin's key disappears from snapshots and clients read it as capability absence. A host reader either declares `sessionProjections` in its plugin `inject` or fails explicitly when the registry or required key is absent. Contributors may preserve optional registration through `ctx.inject(['sessionProjections'], ...)`. Registrants sharing a key share one unit and are counted: the same tool package mounted in N agent presets registers N times, and the key survives until the last one unloads.
|
||||
`ctx.sessionProjections`: the projection unit table and its drive. The service subscribes to `session/event` once; every committed event passes every registered unit's `apply` (eager drive). A changed state reference computes the next client view; the change feed is notified only when its raw result changes by `Object.is`. Cells build lazily — a unit registered after events flowed, or a session older than the registry, folds `init` over the in-memory log on first touch (event or read). Registration is an effect (disposer rides the calling fiber): an unloaded domain plugin's key disappears from snapshots and clients read it as capability absence. A host reader either declares `sessionProjections` in its plugin `inject` or fails explicitly when the registry or required key is absent. Contributors may preserve optional registration through `ctx.inject(['sessionProjections'], ...)`. Registrants sharing a key share one unit and are counted: the same tool package mounted in N agent presets registers N times, and the key survives until the last one unloads.
|
||||
|
||||
```ts cordis-catalog
|
||||
/**
|
||||
|
|
@ -204,7 +204,7 @@ register< K extends Exclude<keyof SessionProjectionStateMap, keyof SessionProjec
|
|||
/**
|
||||
* Subscribe to the change feed. The registration is an effect on the
|
||||
* calling context's fiber.
|
||||
* @param listener - called once per client-visible unit whose state reference changed, per committed event.
|
||||
* @param listener - called once per client-visible unit whose raw view changed by `Object.is`, per committed event.
|
||||
* @returns the exact disposer that unsubscribes.
|
||||
*/
|
||||
onChanged(listener: ProjectionChangeListener): () => void
|
||||
|
|
|
|||
|
|
@ -47,7 +47,10 @@ interface ProjectionDefinition<
|
|||
/** Validates the wire payload before it leaves the host. */
|
||||
viewSchema: ZodType<SessionProjectionMap[K]>
|
||||
/**
|
||||
* State → wire payload (the read-side projection).
|
||||
* State → wire payload (the read-side projection). The live drive keeps
|
||||
* the two latest raw results and compares them with `Object.is`; an
|
||||
* object-valued view must reuse its reference to suppress publication
|
||||
* across internal-only state changes.
|
||||
* @param state - the current state.
|
||||
* @returns the whole current value for this unit's key.
|
||||
*/
|
||||
|
|
@ -83,12 +86,9 @@ interface ProjectionSnapshot {
|
|||
|
||||
```ts type-equiv
|
||||
/**
|
||||
* Change-feed listener: one unit's served value changed for one session.
|
||||
* `value` is the schema-validated `view` output; `seq` is the unit's
|
||||
* watermark at emission (the seq of the event that caused the change). A
|
||||
* changed state whose raw `view` output is `Object.is`-identical to the
|
||||
* unit's previous projection does not fire, so a unit can buffer working
|
||||
* fields in state behind an identity-stable projection.
|
||||
* Change-feed listener: one unit's raw `view` result changed by `Object.is`
|
||||
* for one session. `value` is the schema-validated output; `seq` is the
|
||||
* unit's watermark at emission (the seq of the event that caused the change).
|
||||
*/
|
||||
type ProjectionChangeListener = (
|
||||
session: Session,
|
||||
|
|
@ -98,7 +98,7 @@ type ProjectionChangeListener = (
|
|||
) => void
|
||||
```
|
||||
|
||||
`snapshot(session)` 完全同步：载体在切出页面切片的同一 tick 内读取它，因此 `asOfSeq` 使两次读取使用同一个序号。它只返回客户端视图，并在返回前通过各单元的 `viewSchema` 校验。`stateOf(session, key)` 可在不计算无关视图的情况下读取一份实时 host 状态；调用方不得修改这一借用引用。对于每个已提交事件，变更流会为每个状态*引用*已变化的客户端可见单元触发一次——除非原始 `view` 输出与上一个状态的投影 `Object.is` 相同（视图按状态对象身份做备忘——每个不同状态只算一次，且不存在会过期的「上次交付」记录），因此单元可以把工作字段缓冲在状态里，用身份稳定的投影保持安静，后来的监听者也不会错过任何值变化；状态未变时，`apply` 必须返回同一引用。
|
||||
`snapshot(session)` 完全同步：载体在切出页面切片的同一 tick 内读取它，因此 `asOfSeq` 使两次读取使用同一个序号。它只返回客户端视图，并在返回前通过各单元的 `viewSchema` 校验。`stateOf(session, key)` 可在不计算无关视图的情况下读取一份实时 host 状态；调用方不得修改这一借用引用。state 引用变化时，注册表计算并缓存一次原始 view；只有该结果通过 `Object.is` 判定为变化时才触发变更流，对象 view 若要在仅内部 state 变化时抑制发布就必须保留引用。
|
||||
|
||||
## 注册表：`ctx.sessionProjections`
|
||||
|
||||
|
|
@ -180,7 +180,7 @@ Source: [`packages/session/session-projection-cache/src/index.ts`](../../package
|
|||
|
||||
### `ctx.sessionProjections` — `SessionProjectionRegistry`
|
||||
|
||||
`ctx.sessionProjections`: the projection unit table and its drive. The service subscribes to `session/event` once; every committed event passes every registered unit's `apply` (eager drive), and a changed state reference in a client-visible unit notifies the change feed with the schema-validated view — unless the raw view output is `Object.is`-identical to the unit's previous projection (identity-stable projections stay quiet). Views are memoized by state object identity, so each distinct state's view computes once and no last-delivered record exists to go stale across listener generations. Cells build lazily — a unit registered after events flowed, or a session older than the registry, folds `init` over the in-memory log on first touch (event or read). Registration is an effect (disposer rides the calling fiber): an unloaded domain plugin's key disappears from snapshots and clients read it as capability absence. A host reader either declares `sessionProjections` in its plugin `inject` or fails explicitly when the registry or required key is absent. Contributors may preserve optional registration through `ctx.inject(['sessionProjections'], ...)`. Registrants sharing a key share one unit and are counted: the same tool package mounted in N agent presets registers N times, and the key survives until the last one unloads.
|
||||
`ctx.sessionProjections`: the projection unit table and its drive. The service subscribes to `session/event` once; every committed event passes every registered unit's `apply` (eager drive). A changed state reference computes the next client view; the change feed is notified only when its raw result changes by `Object.is`. Cells build lazily — a unit registered after events flowed, or a session older than the registry, folds `init` over the in-memory log on first touch (event or read). Registration is an effect (disposer rides the calling fiber): an unloaded domain plugin's key disappears from snapshots and clients read it as capability absence. A host reader either declares `sessionProjections` in its plugin `inject` or fails explicitly when the registry or required key is absent. Contributors may preserve optional registration through `ctx.inject(['sessionProjections'], ...)`. Registrants sharing a key share one unit and are counted: the same tool package mounted in N agent presets registers N times, and the key survives until the last one unloads.
|
||||
|
||||
```ts cordis-catalog
|
||||
/**
|
||||
|
|
@ -204,7 +204,7 @@ register< K extends Exclude<keyof SessionProjectionStateMap, keyof SessionProjec
|
|||
/**
|
||||
* Subscribe to the change feed. The registration is an effect on the
|
||||
* calling context's fiber.
|
||||
* @param listener - called once per client-visible unit whose state reference changed, per committed event.
|
||||
* @param listener - called once per client-visible unit whose raw view changed by `Object.is`, per committed event.
|
||||
* @returns the exact disposer that unsubscribes.
|
||||
*/
|
||||
onChanged(listener: ProjectionChangeListener): () => void
|
||||
|
|
|
|||
|
|
@ -538,7 +538,7 @@ describe('Session control projection frames', () => {
|
|||
return frames
|
||||
}
|
||||
|
||||
it('broadcasts a frame per changed unit with the causing seq, and none for same-reference applies', async () => {
|
||||
it('broadcasts changed view references with the causing seq and skips same-reference applies', async () => {
|
||||
const { ctx, session } = await harness(true)
|
||||
ctx.sessionProjections.register(lastUserUnit())
|
||||
const proxy = remote(ctx)
|
||||
|
|
@ -554,6 +554,7 @@ describe('Session control projection frames', () => {
|
|||
now.mockReturnValue(200)
|
||||
session.append('turn/start', { turn: 1 })
|
||||
now.mockReturnValue(300)
|
||||
// The equal payload is a new object, so Object.is still treats its view as changed.
|
||||
seedMessages(session, 1)
|
||||
now.mockRestore()
|
||||
|
||||
|
|
|
|||
|
|
@ -1574,7 +1574,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
|
|||
{
|
||||
key: 'sessionProjections',
|
||||
summary: '`ctx.sessionProjections`: the projection unit table and its drive.',
|
||||
description: '`ctx.sessionProjections`: the projection unit table and its drive. The service subscribes to `session/event` once; every committed event passes every registered unit\'s `apply` (eager drive), and a changed state reference in a client-visible unit notifies the change feed with the schema-validated view — unless the raw view output is `Object.is`-identical to the unit\'s previous projection (identity-stable projections stay quiet). Views are memoized by state object identity, so each distinct state\'s view computes once and no last-delivered record exists to go stale across listener generations. Cells build lazily — a unit registered after events flowed, or a session older than the registry, folds `init` over the in-memory log on first touch (event or read). Registration is an effect (disposer rides the calling fiber): an unloaded domain plugin\'s key disappears from snapshots and clients read it as capability absence. A host reader either declares `sessionProjections` in its plugin `inject` or fails explicitly when the registry or required key is absent. Contributors may preserve optional registration through `ctx.inject([\'sessionProjections\'], ...)`. Registrants sharing a key share one unit and are counted: the same tool package mounted in N agent presets registers N times, and the key survives until the last one unloads.',
|
||||
description: '`ctx.sessionProjections`: the projection unit table and its drive. The service subscribes to `session/event` once; every committed event passes every registered unit\'s `apply` (eager drive). A changed state reference computes the next client view; the change feed is notified only when its raw result changes by `Object.is`. Cells build lazily — a unit registered after events flowed, or a session older than the registry, folds `init` over the in-memory log on first touch (event or read). Registration is an effect (disposer rides the calling fiber): an unloaded domain plugin\'s key disappears from snapshots and clients read it as capability absence. A host reader either declares `sessionProjections` in its plugin `inject` or fails explicitly when the registry or required key is absent. Contributors may preserve optional registration through `ctx.inject([\'sessionProjections\'], ...)`. Registrants sharing a key share one unit and are counted: the same tool package mounted in N agent presets registers N times, and the key survives until the last one unloads.',
|
||||
methods: [
|
||||
{
|
||||
signature: 'register< K extends keyof SessionProjectionMap, S extends SessionProjectionStateMap[K], >( definition: Omit<ProjectionDefinition<K, S>, \'wire\'> & { wire: NonNullable<ProjectionDefinition<K, S>[\'wire\']> }, ): () => void',
|
||||
|
|
@ -1591,7 +1591,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
|
|||
{
|
||||
signature: 'onChanged(listener: ProjectionChangeListener): () => void',
|
||||
description: 'Subscribe to the change feed. The registration is an effect on the calling context\'s fiber.',
|
||||
parameters: [{ name: 'listener', description: 'called once per client-visible unit whose state reference changed, per committed event.' }],
|
||||
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.',
|
||||
},
|
||||
{
|
||||
|
|
|
|||
|
|
@ -2,5 +2,5 @@
|
|||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write packages/session/session-projection/README.md
|
||||
README.md: 4ea86cb187f1081978be97ec0094977ef5e6778e
|
||||
README.zh.md: ab051795bdcb0547e817063bd91374954d79cb0d
|
||||
README.md: 2ef81c00b283e76d0553c85f1ce5acd12ab863da
|
||||
README.zh.md: 996c51887689eed95a5cabee8cf89e6433575647
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ const definition = {
|
|||
}
|
||||
```
|
||||
|
||||
`apply` must be synchronous and must return the same state reference for events that do not concern the unit — an unchanged reference means zero downstream work. A state-carrying log event must carry the complete post-change state, never a bare delta.
|
||||
`apply` must be synchronous and must return the same state reference for events that do not concern the unit — an unchanged reference means zero downstream work. The registry compares consecutive raw `wire.view` results with `Object.is`; an object or array view must reuse its reference to suppress publication across internal-only state changes, while a structurally equal new object is still a change. A state-carrying log event must carry the complete post-change state, never a bare delta.
|
||||
|
||||
### Register and read
|
||||
|
||||
|
|
@ -78,7 +78,7 @@ This section explains the drive machinery and the unit contract; the observable
|
|||
|
||||
### Design concept
|
||||
|
||||
The package is the Service Definition and drive role of a capability seam: the framework drives, the domain computes. The registry subscribes to `session/event` once; every committed event passes every registered unit's `apply` eagerly (cells build lazily on first touch). The change feed is gated on `Object.is` twice — a unit that returns the same state reference costs one call and nothing downstream, and a changed state whose raw `view` output is identical to the previous state's stays quiet — views are memoized by state object identity, so each distinct state's view computes once and no last-delivered record exists to go stale across listener generations (a unit can buffer working fields behind an identity-stable projection). Carriers read `snapshot()` in the same tick as their page slice, which is what makes `asOfSeq` one consistent cut; an accidentally async view returns a Promise and fails `wire.viewSchema.parse`.
|
||||
The package is the Service Definition and drive role of a capability seam: the framework drives, the domain computes. The registry subscribes to `session/event` once; every committed event passes every registered unit's `apply` eagerly (cells build lazily on first touch). The first `Object.is` gate skips view work when the state reference is unchanged; a two-slot live-drive cache reuses the previous raw view and a second `Object.is` gate suppresses publication while the raw view reference is unchanged. Carriers read `snapshot()` in the same tick as their page slice, which is what makes `asOfSeq` one consistent cut; an accidentally async view returns a Promise and fails `wire.viewSchema.parse`.
|
||||
|
||||
### Source map
|
||||
|
||||
|
|
@ -90,7 +90,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 changed client-visible unit notifies the change feed with its schema-validated view and the causing seq. `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 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.
|
||||
|
||||
</details>
|
||||
|
||||
|
|
@ -127,7 +127,7 @@ These limits define where the projection registry needs care at scale. They are
|
|||
|
||||
- **Every tail page carries every client-visible key** — there is no per-key opt-out or lazy-key request shape yet; acceptable while values are UI-scale whole states, revisit if a domain's value grows large.
|
||||
- **The unit table is process-wide, so key presence is not a per-session capability signal** — a key registered by any agent preset appears in every session's snapshot; a client must read the value rather than treat an absent key as absence of the feature.
|
||||
- **Eager drive touches every unit per event** — cheap by construction (whole-value rule, same-reference gate), but a hot path would justify per-unit event-type prefilters.
|
||||
- **Eager drive touches every unit per event** — cheap by construction (whole-value rule and state/view reference gates), but a hot path would justify per-unit event-type prefilters.
|
||||
- **Registry cells live in memory only** — a restart rebuilds by folding the log on first touch; compositions that mount `dsh-session-projection-cache` seed that fold from persisted rows instead.
|
||||
- **Synchronous unit discipline is only partially mechanical** — `wire.viewSchema.parse` rejects a Promise-returning view, but an `apply` that blocks or reads torn non-session state is a review concern.
|
||||
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ const definition = {
|
|||
}
|
||||
```
|
||||
|
||||
`apply` 必须同步，且对与单元无关的事件必须返回同一个状态引用——引用不变意味着零下游工作。携带状态的日志事件必须携带变更后的完整状态，绝不携带裸增量。
|
||||
`apply` 必须同步，且对与单元无关的事件必须返回同一个状态引用——引用不变意味着零下游工作。注册表用 `Object.is` 比较相邻的 `wire.view` 原始结果；对象或数组 view 若要在仅内部 state 变化时抑制发布，就必须复用引用，结构相同的新对象仍算变化。携带状态的日志事件必须携带变更后的完整状态，绝不携带裸增量。
|
||||
|
||||
### 注册与读取
|
||||
|
||||
|
|
@ -78,7 +78,7 @@ const { asOfSeq, values } = ctx.sessionProjections.snapshot(session)
|
|||
|
||||
### 设计理念
|
||||
|
||||
本包是能力 seam 的 Service Definition 与驱动角色：框架负责驱动，领域负责计算。注册表只订阅一次 `session/event`；每个已提交事件都会主动经过每个已注册单元的 `apply`（cell 在首次触达时惰性构建）。变更流以 `Object.is` 把关两道——返回同一状态引用的单元只花一次调用、不产生任何下游工作；状态已变但原始 `view` 输出与上一个状态的投影相同的同样保持安静——视图按状态对象身份做备忘，每个不同状态的视图只计算一次，且不存在会在监听器换代期间过期的「上次交付」记录（单元因此可以把工作字段缓冲在身份稳定的投影之后）。载体在切出页面切片的同一 tick 内读取 `snapshot()`，`asOfSeq` 之所以是一个一致切面正系于此；误写成异步的 view 会返回 Promise，并被 `wire.viewSchema.parse` 拒绝。
|
||||
本包是能力 seam 的 Service Definition 与驱动角色：框架负责驱动，领域负责计算。注册表只订阅一次 `session/event`；每个已提交事件都会主动经过每个已注册单元的 `apply`（cell 在首次触达时惰性构建）。第一层 `Object.is` 闸门在 state 引用不变时跳过 view 工作；live drive 的双槽缓存复用前一个原始 view，第二层 `Object.is` 闸门在原始 view 引用不变时抑制发布。载体在切出页面切片的同一 tick 内读取 `snapshot()`，`asOfSeq` 之所以是一个一致切面正系于此；误写成异步的 view 会返回 Promise，并被 `wire.viewSchema.parse` 拒绝。
|
||||
|
||||
### 源码地图
|
||||
|
||||
|
|
@ -90,7 +90,7 @@ const { asOfSeq, values } = ctx.sessionProjections.snapshot(session)
|
|||
|
||||
### 驱动与检查点流程
|
||||
|
||||
一个已提交事件按注册顺序驱动每个已注册单元；状态引用变化的客户端可见单元会以经 schema 校验的视图与致因 seq 通知变更流。`checkpoint(session)` 为持久缓存返回每个单元一份独立的 `(key → {ver, seq, val})` 行；`restoreFloor` 把尾部读取锚定在最低可用水位之前一个事件处，使缩短的日志可被检出；`restore` 把持久行在存储后缀上重新折叠，丢弃任何 `ver` 不匹配或声称越过存储末尾的行。
|
||||
一个已提交事件按注册顺序驱动每个已注册单元；原始 view 通过 `Object.is` 判定为变化的客户端可见单元会以经 schema 校验的视图与致因 seq 通知变更流。live drive 保留前后两个原始 view；snapshot 与冷读仍是彼此独立的完整读取。`checkpoint(session)` 为持久缓存返回每个单元一份独立的 `(key → {ver, seq, val})` 行；`restoreFloor` 把尾部读取锚定在最低可用水位之前一个事件处，使缩短的日志可被检出；`restore` 把持久行在存储后缀上重新折叠，丢弃任何 `ver` 不匹配或声称越过存储末尾的行。
|
||||
|
||||
</details>
|
||||
|
||||
|
|
@ -127,7 +127,7 @@ const { asOfSeq, values } = ctx.sessionProjections.snapshot(session)
|
|||
|
||||
- **每个尾页携带每个 client-visible key**——尚无逐 key 的 opt-out 或惰性 key 请求形状；在值都是 UI 量级的全量状态时可以接受，若某领域的值变大再重议。
|
||||
- **单元表是进程级的，因此 key 是否存在不能当作逐会话的能力信号**——任何 agent preset 注册的 key 都会出现在每个会话的快照里；客户端必须读值，不能把 key 缺席当作功能缺席。
|
||||
- **主动驱动逐事件触达每个单元**——按构造开销很低（全量值规则、同引用闸门），但若出现热点路径，可加按单元的事件类型预过滤。
|
||||
- **主动驱动逐事件触达每个单元**——按构造开销很低（全量值规则与 state/view 引用闸门），但若出现热点路径，可加按单元的事件类型预过滤。
|
||||
- **注册表 cell 只活在内存里**——重启后首次触达时靠折叠日志重建；挂载了 `dsh-session-projection-cache` 的组合改由持久行播种该折叠。
|
||||
- **单元同步纪律只有部分可机械把关**——`wire.viewSchema.parse` 能拒绝返回 Promise 的 view，但阻塞的 `apply`、或读取撕裂的非会话状态的 `apply`，只能靠评审把关。
|
||||
|
||||
|
|
|
|||
|
|
@ -67,7 +67,10 @@ export interface ProjectionDefinition<
|
|||
/** Validates the wire payload before it leaves the host. */
|
||||
viewSchema: ZodType<SessionProjectionMap[K]>
|
||||
/**
|
||||
* State → wire payload (the read-side projection).
|
||||
* State → wire payload (the read-side projection). The live drive keeps
|
||||
* the two latest raw results and compares them with `Object.is`; an
|
||||
* object-valued view must reuse its reference to suppress publication
|
||||
* across internal-only state changes.
|
||||
* @param state - the current state.
|
||||
* @returns the whole current value for this unit's key.
|
||||
*/
|
||||
|
|
@ -83,12 +86,9 @@ export interface ProjectionDefinition<
|
|||
}
|
||||
|
||||
/**
|
||||
* Change-feed listener: one unit's served value changed for one session.
|
||||
* `value` is the schema-validated `view` output; `seq` is the unit's
|
||||
* watermark at emission (the seq of the event that caused the change). A
|
||||
* changed state whose raw `view` output is `Object.is`-identical to the
|
||||
* unit's previous projection does not fire, so a unit can buffer working
|
||||
* fields in state behind an identity-stable projection.
|
||||
* Change-feed listener: one unit's raw `view` result changed by `Object.is`
|
||||
* for one session. `value` is the schema-validated output; `seq` is the
|
||||
* unit's watermark at emission (the seq of the event that caused the change).
|
||||
*/
|
||||
export type ProjectionChangeListener = (
|
||||
session: Session,
|
||||
|
|
@ -139,11 +139,13 @@ interface ErasedDefinition {
|
|||
stateVersion: number
|
||||
}
|
||||
|
||||
/** Per-session per-unit watermark cache row. */
|
||||
/** Per-session per-unit watermark and fixed live-drive view buffer. */
|
||||
interface UnitCell {
|
||||
state: unknown
|
||||
/** Seq of the last event passed through `apply` (regardless of change). */
|
||||
observedSeq: number
|
||||
/** `[previousView, currentView]`; undefined slots mean no cached comparison. */
|
||||
readonly views: [unknown, unknown]
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -160,13 +162,6 @@ interface UnitCell {
|
|||
interface Registration {
|
||||
readonly def: ErasedDefinition
|
||||
readonly cells: WeakMap<Session, UnitCell>
|
||||
/**
|
||||
* Raw `view` output per state object (pure-view memo). An entry is the
|
||||
* view of that exact state — not a last-delivered record — so it cannot go
|
||||
* stale; a missing entry recomputes. Weak keys die with their states;
|
||||
* primitive states bypass the memo.
|
||||
*/
|
||||
readonly viewMemo: WeakMap<object, unknown>
|
||||
/** Live registrants sharing this unit; the last one out removes the key. */
|
||||
refs: number
|
||||
}
|
||||
|
|
@ -174,13 +169,9 @@ interface Registration {
|
|||
/**
|
||||
* `ctx.sessionProjections`: the projection unit table and its drive. The
|
||||
* service subscribes to `session/event` once; every committed event passes
|
||||
* every registered unit's `apply` (eager drive), and a changed state
|
||||
* reference in a client-visible unit notifies the change feed with the
|
||||
* schema-validated view — unless the raw view output is `Object.is`-identical
|
||||
* to the unit's previous projection (identity-stable projections stay quiet).
|
||||
* Views are memoized by state object identity, so each distinct state's view
|
||||
* computes once and no last-delivered record exists to go stale across
|
||||
* listener generations.
|
||||
* every registered unit's `apply` (eager drive). A changed state reference
|
||||
* computes the next client view; the change feed is notified only when its
|
||||
* raw result changes by `Object.is`.
|
||||
* Cells build lazily — a unit registered after events flowed, or a session
|
||||
* older than the registry, folds `init` over the in-memory log on first
|
||||
* touch (event or read). Registration is an effect (disposer rides the
|
||||
|
|
@ -210,6 +201,7 @@ export class SessionProjectionRegistry extends Service {
|
|||
registration.cells.set(session, {
|
||||
state: registration.def.init(session.header),
|
||||
observedSeq: -1,
|
||||
views: [undefined, undefined],
|
||||
})
|
||||
}
|
||||
})
|
||||
|
|
@ -270,7 +262,7 @@ export class SessionProjectionRegistry extends Service {
|
|||
const key = erased.key
|
||||
const existing = this.registrations.get(key)
|
||||
if (existing === undefined) {
|
||||
this.registrations.set(key, { def: erased, cells: new WeakMap(), viewMemo: new WeakMap(), refs: 1 })
|
||||
this.registrations.set(key, { def: erased, cells: new WeakMap(), refs: 1 })
|
||||
} else {
|
||||
if (existing.def.stateVersion !== erased.stateVersion) {
|
||||
throw new Error(`session projection key ${JSON.stringify(key)} is already registered at stateVersion ${String(existing.def.stateVersion)}; refusing to share it with stateVersion ${String(erased.stateVersion)}`)
|
||||
|
|
@ -291,7 +283,7 @@ export class SessionProjectionRegistry extends Service {
|
|||
/**
|
||||
* Subscribe to the change feed. The registration is an effect on the
|
||||
* calling context's fiber.
|
||||
* @param listener - called once per client-visible unit whose state reference changed, per committed event.
|
||||
* @param listener - called once per client-visible unit whose raw view changed by `Object.is`, per committed event.
|
||||
* @returns the exact disposer that unsubscribes.
|
||||
*/
|
||||
onChanged(listener: ProjectionChangeListener): () => void {
|
||||
|
|
@ -573,6 +565,7 @@ export class SessionProjectionRegistry extends Service {
|
|||
registration.cells.set(session, {
|
||||
state: row.val,
|
||||
observedSeq: row.seq,
|
||||
views: [undefined, undefined],
|
||||
})
|
||||
}
|
||||
return restored.snapshot
|
||||
|
|
@ -591,7 +584,7 @@ export class SessionProjectionRegistry extends Service {
|
|||
): UnitCell {
|
||||
let state = def.init(header)
|
||||
for (const event of events) state = def.apply(state, event)
|
||||
return { state, observedSeq: (events.at(-1)?.seq ?? -1) }
|
||||
return { state, observedSeq: (events.at(-1)?.seq ?? -1), views: [undefined, undefined] }
|
||||
}
|
||||
|
||||
/** Read (or lazily build, folding the full in-memory log) one unit's cell. */
|
||||
|
|
@ -620,12 +613,16 @@ export class SessionProjectionRegistry extends Service {
|
|||
throw new Error(`session projection ${JSON.stringify(def.key)} cannot advance across missing seq ${String(seq)}`)
|
||||
}
|
||||
const next = def.apply(cell.state, event)
|
||||
if (!Object.is(next, cell.state)) {
|
||||
cell.views[0] = cell.views[1]
|
||||
cell.views[1] = undefined
|
||||
}
|
||||
cell.state = next
|
||||
cell.observedSeq = seq
|
||||
}
|
||||
}
|
||||
|
||||
/** Eager drive: pass one committed event through every registered unit; notify on changed references. */
|
||||
/** Eager drive: pass one committed event through every unit; notify on changed raw view references. */
|
||||
private drive(session: Session, event: SessionEvent): void {
|
||||
for (const registration of this.registrations.values()) {
|
||||
let cell = registration.cells.get(session)
|
||||
|
|
@ -638,22 +635,25 @@ export class SessionProjectionRegistry extends Service {
|
|||
} else {
|
||||
this.advanceCell(registration.def, cell, session.events, event.seq - 1)
|
||||
}
|
||||
const previous = cell.state
|
||||
const next = registration.def.apply(previous, event)
|
||||
const changed = !Object.is(next, previous)
|
||||
const previousState = cell.state
|
||||
const next = registration.def.apply(previousState, event)
|
||||
const changed = !Object.is(next, previousState)
|
||||
cell.state = next
|
||||
cell.observedSeq = event.seq
|
||||
if (changed && registration.def.wire !== undefined && this.listeners.size > 0) {
|
||||
// Identity gate on the raw view, memoized by state identity: the
|
||||
// previous state's view was cached when that state was current, so
|
||||
// each distinct state's view computes once and the quiet path
|
||||
// allocates nothing. The memo cannot go stale — an entry is the view
|
||||
// of that exact state, not a record of what the feed last delivered.
|
||||
const raw = this.viewOf(registration.def.wire, registration.viewMemo, next)
|
||||
if (Object.is(this.viewOf(registration.def.wire, registration.viewMemo, previous), raw)) continue
|
||||
const value = registration.def.wire.viewSchema.parse(raw)
|
||||
for (const listener of this.listeners) {
|
||||
listener(session, registration.def.key as Extract<keyof SessionProjectionMap, string>, value, 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<keyof SessionProjectionMap, string>, value, event.seq)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
views[1] = undefined
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -663,22 +663,7 @@ export class SessionProjectionRegistry extends Service {
|
|||
private viewCell(registration: Registration, cell: UnitCell): unknown {
|
||||
const wire = registration.def.wire
|
||||
if (wire === undefined) throw new Error(`session projection ${JSON.stringify(registration.def.key)} has no wire view`)
|
||||
return wire.viewSchema.parse(this.viewOf(wire, registration.viewMemo, cell.state))
|
||||
}
|
||||
|
||||
/**
|
||||
* One unit's raw `view` output for one state, memoized by state object
|
||||
* identity (the pure-view contract makes the entry permanently correct).
|
||||
* Primitive states have no WeakMap key and compute directly.
|
||||
* @param wire - the unit's wire block.
|
||||
* @param memo - the unit's per-state view memo.
|
||||
* @param state - a state produced by the unit's `init`/`apply`.
|
||||
* @returns the raw (pre-validation) `view` output for that state.
|
||||
*/
|
||||
private viewOf(wire: NonNullable<ErasedDefinition['wire']>, memo: WeakMap<object, unknown>, state: unknown): unknown {
|
||||
if (typeof state !== 'object' || state === null) return wire.view(state)
|
||||
if (!memo.has(state)) memo.set(state, wire.view(state))
|
||||
return memo.get(state)
|
||||
return wire.viewSchema.parse(wire.view(cell.state))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -16,8 +16,8 @@ export const inject = ['invariants']
|
|||
|
||||
/**
|
||||
* No runtime invariant: the registry's own contracts (duplicate-key and
|
||||
* stateVersion rejection, effect-tied removal, the Object.is change gate) are
|
||||
* enforced synchronously inside the service and proven by its spec, the
|
||||
* stateVersion rejection, effect-tied removal, and the state/view `Object.is`
|
||||
* gates) are enforced synchronously inside the service and proven by its spec, the
|
||||
* drive relation (every committed `session/event` passes every unit) would
|
||||
* require re-running the drive to check — duplicating the implementation
|
||||
* rather than detecting drift — and the served-value relation (every served
|
||||
|
|
|
|||
|
|
@ -1,13 +1,13 @@
|
|||
/**
|
||||
* SessionProjectionRegistry unit drive: eager apply on committed events with
|
||||
* lazy cell build (registration after events, session after registration),
|
||||
* the Object.is no-change gate (same reference ⇒ zero change-feed work),
|
||||
* snapshot consistency (asOfSeq = last event seq; values from the watermark
|
||||
* cache), duplicate-key rejection, stateVersion validation, and effect-tied
|
||||
* removal of registrations and change listeners (HMR safety).
|
||||
* the Object.is no-change gates (same state or raw view reference ⇒ zero
|
||||
* change-feed work), snapshot consistency (asOfSeq = last event seq; values
|
||||
* from the watermark cache), duplicate-key rejection, stateVersion validation,
|
||||
* and effect-tied removal of registrations and change listeners (HMR safety).
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import { z } from 'zod'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
|
|
@ -19,14 +19,12 @@ declare module '@deepseek-ai/dsh-session-projection/types' {
|
|||
interface SessionProjectionStateMap {
|
||||
'test/marks': MarksState
|
||||
'test/count': number
|
||||
'test/buffered': { marks: string[]; draft: string }
|
||||
'test/label': string
|
||||
'test/stable-view': StableViewState
|
||||
}
|
||||
|
||||
interface SessionProjectionMap {
|
||||
'test/marks': { marks: string[] }
|
||||
'test/buffered': string[]
|
||||
'test/label': string
|
||||
'test/stable-view': { marks: string[] }
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -36,7 +34,15 @@ declare module '@deepseek-ai/dsh-session/types' {
|
|||
}
|
||||
}
|
||||
|
||||
type MarksState = { marks: string[] } | null
|
||||
interface MarksView {
|
||||
marks: string[]
|
||||
}
|
||||
type MarksState = MarksView | null
|
||||
interface StableViewState {
|
||||
revision: number
|
||||
value: MarksView
|
||||
}
|
||||
const marksViewSchema: z.ZodType<MarksView> = z.object({ marks: z.array(z.string()) })
|
||||
const RESTORE_HEADER: SessionHeader = {
|
||||
version: 0,
|
||||
id: SessionId('projection-restore'),
|
||||
|
|
@ -46,11 +52,11 @@ const RESTORE_HEADER: SessionHeader = {
|
|||
const marksUnit = (): Omit<ProjectionDefinition<'test/marks', MarksState>, 'wire'>
|
||||
& { wire: NonNullable<ProjectionDefinition<'test/marks', MarksState>['wire']> } => ({
|
||||
key: 'test/marks',
|
||||
stateSchema: z.object({ marks: z.array(z.string()) }).nullable(),
|
||||
stateSchema: marksViewSchema.nullable(),
|
||||
init: () => null,
|
||||
apply: (state, event) => (event.type === 'test/mark' ? (event).data : state),
|
||||
wire: {
|
||||
viewSchema: z.object({ marks: z.array(z.string()) }),
|
||||
viewSchema: marksViewSchema,
|
||||
view: state => state ?? { marks: [] },
|
||||
},
|
||||
stateVersion: 1,
|
||||
|
|
@ -65,6 +71,27 @@ const countUnit = (): ProjectionDefinition<'test/count', number> => ({
|
|||
stateVersion: 1,
|
||||
})
|
||||
|
||||
const stableViewUnit = (
|
||||
view: (state: StableViewState) => StableViewState['value'],
|
||||
) => ({
|
||||
key: 'test/stable-view',
|
||||
stateSchema: z.object({
|
||||
revision: z.number().int().nonnegative(),
|
||||
value: marksViewSchema,
|
||||
}),
|
||||
init: () => ({ revision: 0, value: { marks: [] } }),
|
||||
apply: (state, event) => {
|
||||
if (event.type === 'turn/start') return { ...state, revision: state.revision + 1 }
|
||||
if (event.type === 'test/mark') return { revision: state.revision + 1, value: event.data }
|
||||
return state
|
||||
},
|
||||
wire: {
|
||||
viewSchema: marksViewSchema,
|
||||
view,
|
||||
},
|
||||
stateVersion: 1,
|
||||
}) satisfies ProjectionDefinition<'test/stable-view', StableViewState>
|
||||
|
||||
async function harness(): Promise<{ ctx: Context; session: Session }> {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
|
|
@ -117,111 +144,62 @@ describe('SessionProjectionRegistry drive', () => {
|
|||
expect(seen).toEqual([{ key: 'test/marks', value: { marks: ['a'] }, seq: event.seq, sessionId: String(session.id) }])
|
||||
})
|
||||
|
||||
it('keeps the feed quiet while a changed state serves an identity-stable view (draft buffering)', async () => {
|
||||
it('does not compute a view while no change listener exists', async () => {
|
||||
const { ctx, session } = await harness()
|
||||
// Unit buffering a working field beside its wire array: the view projects
|
||||
// only `marks`, whose identity survives draft-only applies.
|
||||
ctx.sessionProjections.register({
|
||||
key: 'test/buffered',
|
||||
stateSchema: z.object({ marks: z.array(z.string()), draft: z.string() }),
|
||||
init: () => ({ marks: [], draft: '' }),
|
||||
apply: (state, event) => {
|
||||
if (event.type === 'test/mark') return { marks: event.data.marks, draft: '' }
|
||||
if (event.type === 'turn/start') return { marks: state.marks, draft: `draft-${String(event.seq)}` }
|
||||
return state
|
||||
},
|
||||
wire: { viewSchema: z.array(z.string()), view: state => state.marks },
|
||||
stateVersion: 1,
|
||||
})
|
||||
const seen: { value: unknown; seq: number }[] = []
|
||||
ctx.sessionProjections.onChanged((_session, key, value, seq) => {
|
||||
if (key === 'test/buffered') seen.push({ value, seq })
|
||||
})
|
||||
const first = mark(session, ['a'])
|
||||
// Draft-only applies change the state reference but not the served view.
|
||||
const view = vi.fn((state: StableViewState) => state.value)
|
||||
ctx.sessionProjections.register(stableViewUnit(view))
|
||||
|
||||
session.append('turn/start', { turn: 1 })
|
||||
session.append('turn/start', { turn: 2 })
|
||||
const second = mark(session, ['a', 'b'])
|
||||
expect(seen).toEqual([
|
||||
{ value: ['a'], seq: first.seq },
|
||||
{ value: ['a', 'b'], seq: second.seq },
|
||||
])
|
||||
// The quiet applies still advanced the state itself.
|
||||
expect(ctx.sessionProjections.stateOf(session, 'test/buffered')?.draft).toBe('')
|
||||
expect(ctx.sessionProjections.snapshot(session).values['test/buffered']).toEqual(['a', 'b'])
|
||||
|
||||
expect(ctx.sessionProjections.stateOf(session, 'test/stable-view')?.revision).toBe(2)
|
||||
expect(view).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it("computes each distinct state's view once: the memo serves previous states to the gate and snapshots", async () => {
|
||||
it('publishes the first observed view and suppresses later same-reference views', async () => {
|
||||
const { ctx, session } = await harness()
|
||||
let viewCalls = 0
|
||||
ctx.sessionProjections.register({
|
||||
key: 'test/buffered',
|
||||
stateSchema: z.object({ marks: z.array(z.string()), draft: z.string() }),
|
||||
init: () => ({ marks: [], draft: '' }),
|
||||
apply: (state, event) => {
|
||||
if (event.type === 'test/mark') return { marks: event.data.marks, draft: '' }
|
||||
if (event.type === 'turn/start') return { marks: state.marks, draft: `draft-${String(event.seq)}` }
|
||||
return state
|
||||
},
|
||||
wire: {
|
||||
viewSchema: z.array(z.string()),
|
||||
view: (state) => {
|
||||
viewCalls += 1
|
||||
return state.marks
|
||||
},
|
||||
},
|
||||
stateVersion: 1,
|
||||
})
|
||||
const view = vi.fn((state: StableViewState) => state.value)
|
||||
ctx.sessionProjections.register(stableViewUnit(view))
|
||||
|
||||
const seen: unknown[] = []
|
||||
ctx.sessionProjections.onChanged((_session, key, value) => {
|
||||
if (key === 'test/buffered') seen.push(value)
|
||||
if (key === 'test/stable-view') seen.push(value)
|
||||
})
|
||||
// First change touches two never-seen states (init and next): two calls.
|
||||
mark(session, ['a'])
|
||||
expect(viewCalls).toBe(2)
|
||||
// Draft-only change: the new state computes, the previous is a memo hit.
|
||||
|
||||
session.append('turn/start', { turn: 1 })
|
||||
expect(viewCalls).toBe(3)
|
||||
mark(session, ['a', 'b'])
|
||||
expect(viewCalls).toBe(4)
|
||||
expect(seen).toEqual([['a'], ['a', 'b']])
|
||||
// Snapshot reads reuse the same memo instead of recomputing the view.
|
||||
expect(ctx.sessionProjections.snapshot(session).values['test/buffered']).toEqual(['a', 'b'])
|
||||
expect(viewCalls).toBe(4)
|
||||
session.append('turn/start', { turn: 2 })
|
||||
|
||||
expect(seen).toEqual([{ marks: [] }])
|
||||
expect(view).toHaveBeenCalledTimes(2)
|
||||
|
||||
mark(session, ['changed'])
|
||||
expect(seen).toEqual([{ marks: [] }, { marks: ['changed'] }])
|
||||
expect(view).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('keeps dedup honest across listener generations: a return to an old value after an unobserved change still fires', async () => {
|
||||
it('publishes the first view after an unobserved state change', async () => {
|
||||
const { ctx, session } = await harness()
|
||||
// A primitive-valued view compares by value under Object.is (the title
|
||||
// unit's shape), which is exactly where remembering a delivered value —
|
||||
// instead of comparing the two states in hand — would silence a real
|
||||
// transition.
|
||||
ctx.sessionProjections.register({
|
||||
key: 'test/label',
|
||||
stateSchema: z.string(),
|
||||
init: () => '',
|
||||
apply: (state, event) => (event.type === 'test/mark' ? event.data.marks[0] ?? '' : state),
|
||||
wire: { viewSchema: z.string(), view: state => state },
|
||||
stateVersion: 1,
|
||||
})
|
||||
const first: string[] = []
|
||||
const view = vi.fn((state: StableViewState) => state.value)
|
||||
ctx.sessionProjections.register(stableViewUnit(view))
|
||||
const first: unknown[] = []
|
||||
const stop = ctx.sessionProjections.onChanged((_session, key, value) => {
|
||||
if (key === 'test/label') first.push(value as string)
|
||||
if (key === 'test/stable-view') first.push(value)
|
||||
})
|
||||
mark(session, ['A'])
|
||||
|
||||
session.append('turn/start', { turn: 1 })
|
||||
stop()
|
||||
// Unobserved transition away from 'A'…
|
||||
mark(session, ['B'])
|
||||
// …then a new listener generation subscribes and the value returns:
|
||||
// dedup memory frozen at the delivered 'A' would silence this delivery;
|
||||
// the per-step previous-state comparison sees 'B' → 'A' and fires.
|
||||
const second: string[] = []
|
||||
session.append('turn/start', { turn: 2 })
|
||||
expect(view).toHaveBeenCalledTimes(1)
|
||||
|
||||
const resumed: unknown[] = []
|
||||
ctx.sessionProjections.onChanged((_session, key, value) => {
|
||||
if (key === 'test/label') second.push(value as string)
|
||||
if (key === 'test/stable-view') resumed.push(value)
|
||||
})
|
||||
mark(session, ['A'])
|
||||
expect(first).toEqual(['A'])
|
||||
expect(second).toEqual(['A'])
|
||||
session.append('turn/start', { turn: 3 })
|
||||
|
||||
expect(first).toEqual([{ marks: [] }])
|
||||
expect(resumed).toEqual([{ marks: [] }])
|
||||
expect(view).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('drives independently per session (cells are per-session watermarks)', async () => {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue